local Date = require "LuaOrm.Date"
local summary_helper = require "reporter.summary_helper"
local manage         = require "reporter.manage"
local slices_helper  = require "reporter.slices_helper"
local scheduler = require "reporter.scheduler"

module(..., package.seeall)

-- If the slice of two rows is separated by less than rowFillMinutes, the rows between them are created.
local rowFillMinutes = 60 * 4
local m_utcTimezone = {offset = 0} -- Default to UTC

function QuerySummary (summarizerKS, query, utils, _T, idsToNames)
	--LogDebug("[REPORTER.SUMMARY_QUERY:QuerySummary] query '%s'", json.encode(query))
	idsToNames = idsToNames or {}

	local handle = {}
	local function AddWarning(w)
		handle.warnings = handle.warnings or {}
		table.insert(handle.warnings, w)
	end

	-- Makes sure t[k] is a table. If it's something different, it's set as an empty table.
	-- If it's something different and not nil, a warning is issued.
	local function CheckIsTable(t, k, path)
		local typeT = type(t[k])
		if typeT == "table" then
			return
		end
		if typeT ~= "nil" then
			AddWarning(path.." should be a table. Ignored.")
		end
		t[k] = {}
	end

	assert(summarizerKS, "Cassandra not connected")
	assert(type(query) == "table", "Table query not defined")
	assert(type(query.sliceSize) == "string", "Slice size not specified in query")
	assert(type(query.cut) == "string", "Cut ID not specified in query")
	assert(type(query.vcc) == "string", "VCC not specified in query")
	assert(type(query.filter) == "table", "Filter table not specified in query")
	assert(type(query.filter.DateFrom) == "string", "Filter 'DateFrom' not specified in query")
	assert(type(query.filter.DateTo) == "string", "Filter 'DateTo' not specified in query")

	local scheduleId       = query.queryFields and query.queryFields.scheduleId or nil
	local isSentToDatabase = utils and scheduleId and scheduler.is_send_type(utils, scheduleId, "database")
	local isUTC            = query.filter.isUTC
	local rawData          = isSentToDatabase
--rawData = true
	--LogDebug("[REPORTER.QuerySummary] scheduleId: %s, isSentToDatabase %s, rawData %s, query.filter %s", scheduleId, isSentToDatabase, rawData, json.encode(query.filter))

	if isSentToDatabase then
		-- Si es un RUN de tipo database, tiene que consultar datos infinitos, por esto agregamos un numero
		query.resultSize = 99999999
	end

	CheckIsTable(query, "columnDetails", "Query parameter 'columnDetails'")
	CheckIsTable(query, "indicators", "Query parameter 'indicators'")

	-- Build the handle
	handle.queryCut = query.cut
	handle.querySliceSize = query.sliceSize
	handle.queryIndicators = query.indicators
	handle.queryReverse = query.reverse
	handle.queryTimezone = query.timezone or m_utcTimezone

	local fromTimezone, endTimezone = slices_helper.GetTimezonesFromAndTo(query.timezone, isUTC, utils)
	if rawData then
		fromTimezone = m_utcTimezone
	end

	-- Get the columns and filters
	local cut, errMessage = manage.getCut(summarizerKS, query.vcc, query.cut)
	assert(cut, errMessage)

	--LogDebug("[REPORTER.SUMMARY_QUERY:QuerySummary] query.cut: '%s', cut.columns %s", query.cut, json.encode(cut.columns))

	handle.cutColumns = cut.columns
	handle.cutVCC = cut.vcc
	local cutFilter = cut.filter
	-- Warn about wrong filters

	local objectStartDateUTC = slices_helper.ConvertTimezoneDateToUTCDate(query.filter.DateFrom, fromTimezone, nil, utils)
	local objectEndDateUTC   = slices_helper.ConvertTimezoneDateToUTCDate(query.filter.DateTo  , endTimezone, nil, utils)

	handle.fieldsMap = {
		-- Range is converted to UTC
		DateFrom = objectStartDateUTC.objectDate,
		DateTo   = objectEndDateUTC.objectDate
	}

	--LogDebug("[REPORTER.SUMMARY_QUERY:QuerySummary] '%s', fromTimezone %s, endTimezone %s",
	--	"from " .. handle.fieldsMap.DateFrom:toString().." to "..handle.fieldsMap.DateTo:toString(),
	--	json.encode(fromTimezone),
	--	json.encode(endTimezone)
	--)

	local hasSlice -- Specifies if the columns include Slice (which is mandatory)
	for index, field in ipairs(handle.cutColumns) do
		if field == "Slice" then
			handle.sliceColumn = index
			hasSlice = true
		elseif not hasSlice then
			-- Only Pre Slice columns may be used as filters, unless specified explicitly on filters.
			handle.fieldsMap[field] = ""
		end
	end
	assert(hasSlice, "Cut cannot be used, column definition lacks 'Slice'")
	for _, field in ipairs(cutFilter) do
		if field ~= "Slice" then
			handle.fieldsMap[field] = ""
		end
	end
	for filter, value in pairs(query.filter) do
		if not handle.fieldsMap[filter] then
			AddWarning("Cut "..query.cut.." cannot be filtered by "..filter..", filter field ignored.")
		elseif handle.fieldsMap[filter] == "" then
			handle.fieldsMap[filter] = value
		end
	end

	-- Check the queries that should be done
	local domainFilter = {}     -- Array of Pre Slice Columns and Back Filters, in the form of tables { field, value }.
	                            -- Value equals "" if not filtering by that field.
	                            -- Elements are sorted alphabetically by field.
	                            -- domainFilter is only filled up if there are Pre Slice Columns.
	local preSliceColumns = {}  -- Array of Pre Slice Columns, sorted in order of appearance on cutColumns
	local backFiltersMap = {}   -- Set of filter fields that are not Pre Slice columns
	local backFilters = {}      -- Elements of backFiltersMap, sorted alphabetically
	local allFields = {}        -- Table where the keys are the fields in columns and filters (except slice),
	                            -- and the values are if there's a column with that field.

	-- First set all filter fields in backFiltersMap. They will be removed if necessary when checking the columns.
	for _, field in ipairs(cutFilter) do
		if field ~= "Slice" then
			backFiltersMap[field] = true
			allFields[field] = false
		end
	end

	-- All Pre Slice columns should be inserted in domainFilter, and removed from backFiltersMap.
	local sliceReached
	for _, field in ipairs(handle.cutColumns) do
		if field == "Slice" then
			sliceReached = true
		else
			allFields[field] = true
		end
		if not sliceReached then
			-- This is a Pre Slice Column
			table.insert(preSliceColumns, field)
			table.insert(domainFilter, { field = field, value = handle.fieldsMap[field] })
			backFiltersMap[field] = nil
		end
	end

	--LogDebug("a --> sliceReached: '%s' domainFilter: %s", sliceReached, json.encode(domainFilter))

	if #domainFilter > 0 then
		-- There are Pre Slice Columns. Add the Back Filters to domainFilter and sort it.
		for field in pairs(backFiltersMap) do
			table.insert(domainFilter, { field = field, value = handle.fieldsMap[field] })
		end
		table.sort(domainFilter, function(lhs, rhs)
			return lhs.field < rhs.field
		end)
	end
	--LogDebug("b --> domainFilter: %s", json.encode(domainFilter))

	-- Create the keySuffix, which are the field values from the cut filters that are not Pre Slice columns,
	-- sorted alphabetically and separated by newline. If any, they start with a newline so they can be appended
	-- to the prefix of a key easily.

	--LogDebug("b.1 --> backFiltersMap: %s", json.encode(backFiltersMap))

	for fieldName in pairs(backFiltersMap) do
		table.insert(backFilters, fieldName)
	end
	table.sort(backFilters)
	--LogDebug("b.2 --> backFilters: %s", json.encode(backFilters))

	-- El keySuffix queda en una tabla, para contemplar los filtros que llegan en la tabla (campaign)
	handle.keySuffix = {}

	for _, fieldName in ipairs(backFilters) do
		local fieldContent = handle.fieldsMap[fieldName]

		--LogDebug("-- %s = '%s'", fieldName, type(fieldContent) == "table" and json.encode(fieldContent) or fieldContent)

		if type(fieldContent) == "table" then
			if #handle.keySuffix == 0 then
			-- Si es una tabla inserto
				for n, v in pairs(fieldContent) do
					local key = "" .. "\n" .. v
					table.insert(handle.keySuffix, key)
				end
			else
				-- Si No es una tabla y YA hay info en el keySuffix actualizo lo que ya existe.
				local temporalKeySuffix = {}
				for i, keySuffix in ipairs(handle.keySuffix) do
					for n, key in pairs(fieldContent) do
						table.insert(temporalKeySuffix, keySuffix .. "\n" .. key)
					end
				end
				handle.keySuffix = temporalKeySuffix
			end
		else
			if #handle.keySuffix == 0 then
			-- Si No es una tabla y aún no hay info en el keySuffix inserto
				local key = "" .. "\n" .. fieldContent
				table.insert(handle.keySuffix, key)
			else
			-- Si No es una tabla y YA hay info en el keySuffix actualizo lo que ya existe.
				for i, keySuffix in ipairs(handle.keySuffix) do
					handle.keySuffix[i] = keySuffix .. "\n" .. fieldContent
				end
			end
		end
	end
	--LogDebug("c --> handle.keySuffix: %s", json.encode(handle.keySuffix))

	-- Create the domainSuffix, which are the Pre Slice column names separated by newline
	handle.domainSuffix = table.concat(preSliceColumns, "\n")
	--LogDebug("d --> handle.domainSuffix: '%s'", handle.domainSuffix)

	-- Calculate the months
	handle.months = {}
	if query.reverse then
		local month = handle.fieldsMap.DateTo:monthDate()
		local fromMonth = handle.fieldsMap.DateFrom:monthDate()
		while month:time() >= fromMonth:time() do
			table.insert(handle.months, month:toString("%Y-%m"))
			month = month:add(0, -1, 0)
		end
	else
		local month = handle.fieldsMap.DateFrom:monthDate()
		while month:time() < handle.fieldsMap.DateTo:time() do
			table.insert(handle.months, month:toString("%Y-%m"))
			month = month:add(0, 1, 0)
		end
	end
	assert(#handle.months > 0, "Invalid period from "..handle.fieldsMap.DateFrom:toString().." to "..handle.fieldsMap.DateTo:toString())

	-- Create indicator keys to merge
	handle.indicatorKeys = {}
	handle.sortedIndicatorKeys = {}
	if #domainFilter > 0 then
		--LogDebug("[REPORTER.QuerySummary] CON %d domainFilter", #domainFilter)
		handle.baseDomainKey = {}
		handle.filterWithSomeCampaigns = false

		local domainFilterWVal = {{}}
		for _, f in ipairs(domainFilter) do
			-- Si el valor del campo de filtro es una tabla (caso de las campanas o los agentes que vienen en un array)
			if type(f.value) == "table" then
				if (f.field or ""):lower() == "campaign" then
					handle.filterWithSomeCampaigns = true
				end
				local temporalDomainFilterWVal = {}
				for _, filterWVal in ipairs(domainFilterWVal) do
					for n, v in ipairs(f.value) do
						local copy = utils.DeepCopy(filterWVal)
						table.insert(copy, f.field ..":".. v)
						table.insert(temporalDomainFilterWVal, copy)
					end
				end
				domainFilterWVal = temporalDomainFilterWVal
			else
			-- Si el valor del campo del filtro es un string
				for _, filterWVal in ipairs(domainFilterWVal) do
					table.insert(filterWVal, f.field ..":" .. f.value)
				end
			end
		end

		for _, filterWVal in ipairs(domainFilterWVal) do
			local stringDomainFilter = table.concat(filterWVal, "\n")
			--LogDebug("%d * -----> stringDomainFilter %s", _, stringDomainFilter)
			table.insert(handle.baseDomainKey, stringDomainFilter)
		end
		--LogDebug("[REPORTER.QuerySummary] 3--> handle.nextDomainKey: %s", handle.nextDomainKey)
		handle.nextDomainKey = ""
	else
		--LogDebug("[REPORTER.QuerySummary] SIN domainFilter")
		for _, month in ipairs(handle.months) do
			for _, suffix in ipairs(handle.keySuffix) do
				local indicatorKey = month.."\n15Min"..suffix
				handle.indicatorKeys[indicatorKey] = {
					rowPrefix = ""
				}
				table.insert(handle.sortedIndicatorKeys, indicatorKey)
			end
		end
	end
	--LogDebug("e1 --> handle.baseDomainKey: %s", json.encode(handle.baseDomainKey or {}))
	--LogDebug("e1.1 --> handle.sortedIndicatorKeys: %s", json.encode(handle.sortedIndicatorKeys))
	--LogDebug("e2 --> handle.indicatorKeys: %s", json.encode(handle.indicatorKeys))

	handle.currentIndicatorKey = 1

	-- Validate the column details
	handle.columnDetails = {}
	handle.indicators = {}

	-- creates a closure bound to the a getter, the connection and an indicatorID
	-- local doGet = makeGetter(manage.getSummaryIndicator, "IndicatorOne")
	-- local indicator = doGet("system")
	local function wrapGetter(getter, indicatorID)
		return (function(vcc)
			return getter(summarizerKS, vcc, indicatorID)
		end)
	end

	local function ProcessColumn(column, isIndicator)
		if handle.columnDetails[column] then
			return -- Already processed
		end

		CheckIsTable(query.columnDetails, column, "Column detail of '"..column.."'")
		local details = query.columnDetails[column]

		local columnDetail = {
			columnName		= details.columnName,
			description		= details.description,
			format			= details.format,
			min				= details.min
		}

		local function addColumn()
			if not columnDetail.columnName then
				columnDetail.columnName = column
			end
			columnDetail.columnLabel = _T(columnDetail.columnName)
			if columnDetail.description then
				columnDetail.description = _T(columnDetail.description)
			end
			handle.columnDetails[column] = columnDetail
		end

		if not isIndicator then
			addColumn()
			return -- Following values are only valid for indicators
		end

		-- Checks if the indicator is a composite
		local function checkValidComposite()
			if details.composite then
				if type(details.composite) ~= "table" then
					AddWarning("Composite specification of indicator '".. column .."' is not a table. Indicator is treated as a regular one.")
					return false
				end
				if type(details.composite.indicators) ~= "table" then
					AddWarning("Composite specification of indicator '".. column .."' lacks an 'indicators' table. Indicator is treated as a regular one.")
					return false
				end
				if type(details.composite.formula) ~= "string" then
					AddWarning("Composite specification of indicator '".. column .."' lacks a 'formula' string. Indicator is treated as a regular one.")
					return false
				end
				return true -- valid inline composite (builtin reports)
			end

			-- not "inline" composite (may be custom)
			local tryGet = wrapGetter(manage.getCompositeIndicator, column)
			local indicator = tryGet(query.vcc) or (query.vcc ~= "system" and tryGet("system"))
			if indicator then
				details.composite = indicator
				return true
			end
		end

		local function FindBaseIndicator(indicatorID)
			indicatorID = manage.cleanupIndicatorId(indicatorID)

			if handle.indicators[indicatorID] then
				return true
			end

			-- try to find as base indicator
			local tryGet = wrapGetter(manage.getSummaryIndicator, indicatorID)
			local indicator = tryGet(query.vcc) or (query.vcc ~= "system" and tryGet("system"))
			if indicator then
				handle.indicators[indicatorID] = {
					VCC			= indicator.vcc, 
					minMax		= indicator.extras and indicator.extras.MinMaxIndicator,
					format		= indicator.format,
					name		= indicator.name,
					description	= indicator.description
				}
				return true
			end

			-- indicator not found as base, try to find it as composite
			tryGet = wrapGetter(manage.getCompositeIndicator, indicatorID)
			indicator = tryGet(query.vcc) or (query.vcc ~= "system" and tryGet("system"))
			if indicator then
				return indicator
			end

			AddWarning(string.format("Indicator '%s' not found among the base indicators for VCC %s%s.",
					indicatorID, query.vcc, (query.vcc ~= "system" and " nor system" or "")
				))
			return false
		end

		local compositesBeingProcessed = { }
		local function processComposite(composite, indicatorID)
			-- check for cycles
			if compositesBeingProcessed[indicatorID] then
				AddWarning("Circular dependency detected for indicator '".. indicatorID .."'.")
				return {
					formula = [[return "Circular dependency"]],
					indicators = {},
				}
			end
			-- mark as in process
			compositesBeingProcessed[indicatorID] = true
			-- setup composites indicators
			local indicatorsInUse = { }
			for _, indicator in ipairs(composite.indicators) do
				local res = FindBaseIndicator(indicator)
				if res then
					table.insert(indicatorsInUse, {
						id        = indicator,
						composite = type(res) == "table" and processComposite(res, indicator) or nil
					})
				end
			end
			-- release this composite
			compositesBeingProcessed[indicatorID] = nil
			-- and return the processed value
			return {
				--id         = indicatorID,
				format		= composite.format,
				formula		= composite.formula,
				indicators	= indicatorsInUse,
				name		= composite.name,
				description = composite.description
			}
		end -- processComposite()

		if checkValidComposite() then
		-- The indicator is a valid composite. Save it as that.
			columnDetail.composite = processComposite(details.composite, column)
			if columnDetail.composite then
				columnDetail.columnName = columnDetail.columnName or columnDetail.composite.name
				columnDetail.description = columnDetail.description or columnDetail.composite.description
				columnDetail.format = columnDetail.format or columnDetail.composite.format
			end
		else
			-- This may be a base indicator
			FindBaseIndicator(column)
			if handle.indicators[column] then
				columnDetail.columnName = columnDetail.columnName or handle.indicators[column].name
				columnDetail.description = columnDetail.description or handle.indicators[column].description
				columnDetail.format = columnDetail.format or handle.indicators[column].format
			end
		end

		addColumn()
	end

	-- Process the columns, making the headers in the process
	handle.headers = {}
	handle.columnsMetadata = {}
	for _, column in ipairs(handle.cutColumns) do
		ProcessColumn(column)
		local data = handle.columnDetails[column]
		table.insert(handle.headers, data.columnLabel)
		table.insert(handle.columnsMetadata, {
			name		= column,
			label		= data.columnName or column,
			description = data.description,
			type		= "String",
		})
	end
	for _, indicator in ipairs(query.indicators) do
		ProcessColumn(indicator, true)
		local data = handle.columnDetails[indicator]
		table.insert(handle.headers, data.columnLabel)
		table.insert(handle.columnsMetadata, {
			name		= indicator,
			label		= data.columnName or indicator,
			description = data.description,
			type		= data.format and data.format.type,
		})
	end

	-- Convert the date range to string
	handle.fieldsMap.DateFrom = handle.fieldsMap.DateFrom:toString("%Y-%m-%d %H:%M")
	handle.fieldsMap.DateTo_d = handle.fieldsMap.DateTo
	handle.fieldsMap.DateTo = handle.fieldsMap.DateTo:toString("%Y-%m-%d %H:%M")

	-- Gather all non-slice fields
	handle.nonSliceFields = {}
	for field, isColumn in pairs(allFields) do
		table.insert(handle.nonSliceFields, { field = field, isColumn = isColumn })
	end
	table.sort(handle.nonSliceFields, function(lhs, rhs)
		return lhs.field < rhs.field
	end)

	--LogDebug("[REPORTER.SUMMARY_QUERY:QuerySummary] handle %s", json.encode(handle or {}))

	--LogDebug("[REPORTER.SUMMARY_QUERY:QuerySummary] handle.fieldsMap %s", json.encode(handle.fieldsMap or {}))
	--LogDebug("[REPORTER.SUMMARY_QUERY:QuerySummary] handle.indicators %s", json.encode(handle.indicators or {}))
	--LogDebug("[REPORTER.SUMMARY_QUERY:QuerySummary] handle.indicatorKeys %s", json.encode(handle.indicatorKeys or {}))
	--LogDebug("[REPORTER.SUMMARY_QUERY:QuerySummary] handle.keySuffix %s", json.encode(handle.keySuffix))
	--LogDebug("[REPORTER.SUMMARY_QUERY:QuerySummary] handle.sortedIndicatorKeys %s", json.encode(handle.sortedIndicatorKeys or {}))

	return handle
end

function IsComplete(handle)
	return handle.complete
end

function Reset(handle)
	handle.complete = nil
	handle.nextCol = nil
	handle.currentIndicatorKey = 1
	if handle.baseDomainKey then
		handle.indicatorKeys = {}
		handle.sortedIndicatorKeys = {}
		handle.nextDomainKey = ""
	end
end

function NextPage (summarizerKS, handle, rows, idsToNames, query, utils)
	--LogDebug("[REPORTER:NextPage] query '%s'", json.encode(query))
	--LogDebug("[REPORTER:NextPage] handle '%s'", json.encode(handle or {}))
	idsToNames = idsToNames or {}

	local scheduleId       = query.queryFields and query.queryFields.scheduleId or nil
	local isSentToDatabase = utils and scheduleId and scheduler.is_send_type(utils, scheduleId, "database")
	local rawData          = isSentToDatabase
--rawData = true
	LogDebug("[REPORTER.NextPage] scheduleId: %s, isSentToDatabase %s, rawData %s", scheduleId, isSentToDatabase, rawData)
	-- LogDebug("[REPORTER.NextPage] 1 baseDomainKey %s", json.encode(handle.baseDomainKey or {}))
	-- LogDebug("[REPORTER.NextPage] 2 nextDomainKey %s", json.encode(handle.nextDomainKey or {}))
	-- LogDebug("[REPORTER.NextPage] 3 filterWithSomeCampaigns %s", handle.filterWithSomeCampaigns)

	rows = rows or 100
	local result = {
		columnsMetadata = handle.columnsMetadata,
		headers = handle.headers,
		rows = {},
		length = 0
	}

	if handle.complete then
		return result, handle
	end

	local compiledComposites = {}

	local function CompileCompositeFunctions()

		local function compileComposite(indicatorID, composite)
			if composite.formula then
				composite.fn = loadstring(composite.formula)
				compiledComposites[indicatorID] = composite
			end

			for index, subindicator in ipairs(composite.indicators) do
				if subindicator.composite and not subindicator.composite.fn then
					if compiledComposites[subindicator.id] then
						subindicator.composite = compiledComposites[subindicator.id]
					else
						compileComposite(subindicator.id, subindicator.composite)
					end
				end
			end
		end

		for column, detail in pairs(handle.columnDetails or {}) do
			if detail.composite then
				if compiledComposites[column] then
					detail.composite = compiledComposites[column]
				else
					compileComposite(column, detail.composite)
				end
			end
		end
	end

	local function RemoveCompositeFunctions()
		for _,composite in pairs(compiledComposites) do
			composite.fn = nil
		end
	end

	-- Compile all composite functions
	CompileCompositeFunctions()


	--[[

	To group 15 minute rows into rows of the query slice size, we use 3 concepts:

	- Megarow: contains all 15 minute rows that start with the same pre-slice columns, and all the 15 minute slices fall in the same query size slice.
	- Row: contains all 15 minute rows that share all non-slice columns, and all the 15 minute slices fall in the same query size slice.
	- Minirow: 15 minute rows.

	Example (supposing the query size is Hour):

	This are the original 15 minute rows:
	Campaign | Slice                   | Agent     | Type
	---------|-------------------------|-----------|------
	sales    | 2013-07-03 17:15:00.000 | fernickk  | CALL
	sales    | 2013-07-03 17:15:00.000 | fernickk  | MAIL
	sales    | 2013-07-03 17:15:00.000 | juan      | MAIL
	sales    | 2013-07-03 17:30:00.000 | fernickk  | CALL
	sales    | 2013-07-03 17:45:00.000 | alexis    | CALL
	sales    | 2013-07-03 18:00:00.000 | fernickk  | MAIL
	support  | 2013-07-03 14:30:00.000 | fernickk  | IM

	Megarow: sales | 2013-07-03 17:00 - 18:00
		Row: sales | 2013-07-03 17:00 - 18:00 | alexis | CALL
			Minirow: sales | 2013-07-03 17:45:00.000 | alexis | CALL
		Row: sales | 2013-07-03 17:00 - 18:00 | fernickk | CALL
			Minirow: sales | 2013-07-03 17:15:00.000 | fernickk | CALL
			Minirow: sales | 2013-07-03 17:30:00.000 | fernickk | CALL
		Row: sales | 2013-07-03 17:00 - 18:00 | fernickk | MAIL
			Minirow: sales | 2013-07-03 17:15:00.000 | fernickk | MAIL
		Row: sales | 2013-07-03 17:00 - 18:00 | juan | MAIL
			Minirow: sales | 2013-07-03 17:15:00.000 | juan | MAIL

	Megarow: sales | 2013-07-03 18:00 - 19:00
		Row: sales | 2013-07-03 18:00 - 19:00 | fernickk | MAIL
			Minirow: sales | 2013-07-03 18:00:00.000 | fernickk | MAIL

	Megarow: support | 2013-07-03 14:00 - 15:00
		Row: support | 2013-07-03 14:00 - 15:00 | fernickk | IM
			Minirow: support | 2013-07-03 14:30:00.000 | fernickk | IM

	Minirows should be retrieved while one of this conditions apply:

	- The minirow falls inside an already retrieved megarow
	- There are less rows than the required by the query

	When a minirow appears that doesn't fall on the already retrieved megarows, and there are enough rows to satisfy the query already,
	then that minirow is used as the next to query on the next page, the rows are sorted inside the megarow, and the rows are returned.
	More than the requested rows may be returned, and the client should manage the extra rows, either displaying them or storing them
	locally for the next page.

	]]

	--[[

	Empty row fillup:

	In order to avoid "holes" on the report when there's no data for a specific slice, a fillup mechanism takes place,
	creating rows with zero data to complete the missing rows. This creation only takes place if the previous and next row happen
	with a difference equal or less than rowFillMinutes: if there's a gap between hours 14 and 17, hours 15 and 16 are created,
	but if the gap is between 22 and 8 of the next day, nothing is generated. This is done to avoid filling time ranges when the VCC is not operative.

	When there are post-slice columns, they are copied from the last megarow.

	Of course this only takes place when the slice size is smaller than a day.

	Example:

	If the original report is:

	  Campaign | Slice                    | Agent     | Type | Inbound.Total | Inbound.Abandoned
	  ---------|--------------------------|-----------|------|---------------|------------------
	  sales    | 2013-07-03 17:00 - 18:00 | alexis    | CALL |            34 |                 7
	  sales    | 2013-07-03 17:00 - 18:00 | fernickk  | CALL |            17 |                 1
	  sales    | 2013-07-03 17:00 - 18:00 | fernickk  | MAIL |             4 |                 0
	  sales    | 2013-07-03 17:00 - 18:00 | juan      | MAIL |             5 |                 0
	  sales    | 2013-07-03 19:00 - 20:00 | fernickk  | MAIL |            23 |                 0
	  sales    | 2013-07-03 22:00 - 23:00 | juan      | CALL |             9 |                 2
	  sales    | 2013-07-04 07:00 - 08:00 | alexis    | CALL |             3 |                 0
	  support  | 2013-07-03 14:00 - 15:00 | fernickk  | IM   |             3 |                 2
	  support  | 2013-07-03 17:00 - 18:00 | juan      | IM   |            56 |                24

	The filled up report becomes (* marks new rows):

	  Campaign | Slice                    | Agent     | Type | Inbound.Total | Inbound.Abandoned
	  ---------|--------------------------|-----------|------|---------------|------------------
	  sales    | 2013-07-03 17:00 - 18:00 | alexis    | CALL |            34 |                 7
	  sales    | 2013-07-03 17:00 - 18:00 | fernickk  | CALL |            17 |                 1
	  sales    | 2013-07-03 17:00 - 18:00 | fernickk  | MAIL |             4 |                 0
	  sales    | 2013-07-03 17:00 - 18:00 | juan      | MAIL |             5 |                 0
	* sales    | 2013-07-03 18:00 - 19:00 | alexis    | CALL |             0 |                 0
	* sales    | 2013-07-03 18:00 - 19:00 | fernickk  | CALL |             0 |                 0
	* sales    | 2013-07-03 18:00 - 19:00 | fernickk  | MAIL |             0 |                 0
	* sales    | 2013-07-03 18:00 - 19:00 | juan      | MAIL |             0 |                 0
	  sales    | 2013-07-03 19:00 - 20:00 | fernickk  | MAIL |            23 |                 0
	* sales    | 2013-07-03 20:00 - 21:00 | fernickk  | MAIL |             0 |                 0
	* sales    | 2013-07-03 21:00 - 22:00 | fernickk  | MAIL |             0 |                 0
	  sales    | 2013-07-03 22:00 - 23:00 | juan      | CALL |             9 |                 2
	  sales    | 2013-07-04 07:00 - 08:00 | alexis    | CALL |             3 |                 0
	  support  | 2013-07-03 14:00 - 15:00 | fernickk  | IM   |             3 |                 2
	* support  | 2013-07-03 15:00 - 16:00 | fernickk  | IM   |             0 |                 0
	* support  | 2013-07-03 16:00 - 17:00 | fernickk  | IM   |             0 |                 0
	  support  | 2013-07-03 17:00 - 18:00 | juan      | IM   |            56 |                24

	]]

	--[[

	Implementation details:

	The megarows table is an array, taking advantage than a minirow can only fit on the last megarow.
	Each element of the array is { key (string), keyTable (array), sliceStart (LuaOrm.Date), rows (table), rowKeys (array) }.
	Each "rows" is a table, where the keys are the row keys (only the post-slice part), and the values are a table { keyTable (array), columnValues (table), minirowIndicators (table) }.
	Each minirowIndicators is a table, where the keys are indicator names, and the values are the value for that indicator on the minirows. The individual minirows are discarded.

	Example:

	megarows = {
		{
			key = "sales | 2013-07-03 17:00 - 18:00",
			keyTable = { "sales", "2013-07-03 17:00 - 18:00" },
			sliceStart = LuaOrm.Date("2013-07-03 17:00:00"),
			rows = {
				["alexis | CALL"] = {
					keyTable = { "alexis", "CALL" },
					columnValues = { Campaign = "sales", Slice = "2013-07-03 17:00 - 18:00", Agent = "alexis", Type = "CALL" },
					minirowIndicators = { ["Inbound.Total"] = 34, ["Inbound.Abandoned"] = 7 }, -- These are the indicators for "sales | 2013-07-03 17:45:00.000 | alexis | CALL"
				},
				["fernickk | CALL"] = {
					keyTable = { "fernickk", "CALL" },
					columnValues = { Campaign = "sales", Slice = "2013-07-03 17:00 - 18:00", Agent = "fernickk", Type = "CALL" },
					minirowIndicators = { ["Inbound.Total"] = 17, ["Inbound.Abandoned"] = 1 }, -- These are the indicators for "sales | 2013-07-03 17:15:00.000 | fernickk | CALL"
					                                                                           -- and "sales | 2013-07-03 17:30:00.000 | fernickk | CALL" combined
				},
				["fernickk | MAIL"] = {
					keyTable = { "fernickk", "MAIL" },
					columnValues = { Campaign = "sales", Slice = "2013-07-03 17:00 - 18:00", Agent = "fernickk", Type = "MAIL" },
					minirowIndicators = { ["Inbound.Total"] = 4, ["Inbound.Unanswered"] = 1 }, -- These are the indicators for "sales | 2013-07-03 17:15:00.000 | fernickk | MAIL"
				},
				["juan | MAIL"] = {
					keyTable = { "juan", "MAIL" },
					columnValues = { Campaign = "sales", Slice = "2013-07-03 17:00 - 18:00", Agent = "juan", Type = "MAIL" },
					minirowIndicators = { ["Inbound.Total"] = 5, ["Outbound.Total"] = 3 },     -- These are the indicators for "sales | 2013-07-03 17:15:00.000 | juan | MAIL"
				},
			},
			rowKeys = { "alexis | CALL", "fernickk | CALL", "fernickk | MAIL", "juan | CALL" } -- Not necessarily ordered until the end
		},
		{
			key = "sales | 2013-07-03 18:00 - 19:00",
			keyTable = { "sales", "2013-07-03 18:00 - 19:00" },
			sliceStart = LuaOrm.Date("2013-07-03 18:00:00"),
			rows = {
				["fernickk | MAIL"] = {
					keyTable = { "fernickk", "MAIL" },
					columnValues = { Campaign = "sales", Slice = "2013-07-03 18:00 - 19:00", Agent = "fernickk", Type = "MAIL" },
					minirowIndicators = { ["Inbound.Total"] = 10, ["Outbound.Total"] = 3 },    -- These are the indicators for "sales | 2013-07-03 18:00:00.000 | fernickk | MAIL"
				},
			},
			rowKeys = { "fernickk | MAIL" }
		},
		{
			key = "support | 2013-07-03 14:00 - 15:00",
			keyTable = { "support", "2013-07-03 14:00 - 15:00" },
			sliceStart = LuaOrm.Date("2013-07-03 14:00:00"),
			rows = {
				["fernickk | IM"] = {
					keyTable = { "fernickk", "IM" },
					columnValues = { Campaign = "support", Slice = "2013-07-03 14:00 - 15:00", Agent = "fernickk", Type = "IM" },
					minirowIndicators = { ["Inbound.Total"] = 1, ["Outbound.Total"] = 37 },    -- These are the indicators for "support | 2013-07-03 14:30:00.000 | fernickk | IM"
				},
			},
			rowKeys = { "fernickk | IM" }
		},
	}

	]]

	local megarows = {}
	local lastMegarow

	-- This function tries to add a minirow to the megarows, checking if it's needed. Returns if it's added.
	-- If it's not added, it means the rows are ready to be returned (they have to be sorted first).
	-- Indicators are passed as a function, so they're only calculated when they're going to be used. This function receives the minirow date (as LuaOrm.Date).
	local function AddMiniRow(minirowName, minirowIndicators)
		-- Extract the pre-slice, slice and post-slice columns
		local preSlice = {}
		local minirowDateUTC -- This is a LuaOrm.Date with the minirow date (15Min slice) on UTC
		local minirowDate    -- This is a LuaOrm.Date with the minirow date (15Min slice) on the desired TZ
		local sliceStart     -- This is a LuaOrm.Date with the row date (slice as per query) on the desired TZ
		local slice          -- This is a string, representing the slice that starts at sliceStart
		local sliceDay       -- Same as slice, but for "Day" size, used for in-day row filling
		local postSlice = {}
		local columnIndex = 0

		local columnValues = {}
		for field in summary_helper.gsplit(minirowName, "\n") do
			columnIndex = columnIndex + 1
			if columnIndex == handle.sliceColumn then
				-- This is the slice
				minirowDateUTC = LuaOrm.Date(field)

				local utcDate = minirowDateUTC:toString()
				-- Convierte la fecha al TimeZone del Usuario.
				minirowDate = slices_helper.ConvertUTCDateToTimezoneDate(utcDate, handle.queryTimezone, nil, utils).objectDate

				sliceStart = summary_helper.GetSliceStart(minirowDate, handle.querySliceSize)
				slice = summary_helper.FormatDateForSlice(sliceStart, handle.querySliceSize)
				sliceDay = summary_helper.FormatDateForSlice(sliceStart, "Day")

				--LogDebug("[REPORTER.NextPage] ------> field: %s, minirowDateUTC: %s, minirowDate: %s, sliceStart: %s, slice: %s, sliceDay: %s", field, minirowDateUTC, minirowDate, sliceStart, slice, sliceDay)
				columnValues["Slice"] = slice
			else
				if handle.cutColumns[columnIndex] then
					columnValues[handle.cutColumns[columnIndex]] = field
				end
				-- This is a pre-slice or post-slice, depending on slice existing
				table.insert(slice and postSlice or preSlice, field)
			end
		end

		table.insert(preSlice, sliceDay)
		local megarowDayKey = table.concat(preSlice, "\n")
		preSlice[#preSlice] = slice -- Replacing sliceDay
		local megarowKey = table.concat(preSlice, "\n")
		local rowKey = table.concat(postSlice, "\n")

		-- This function is called when the minirow is going to be added, and returns the indicators to add.
		local function CalculateIndicators()
			local indicators = {}
			-- Get MinMax values for all MinMaxIndicators
			for indicator, indicatorInfo in pairs(minirowIndicators) do
				--LogDebug(" CalculateIndicators  indicator: %s, indicatorInfo: '%s'", indicator, json.encode(indicatorInfo))
				if handle.indicators[indicator].minMax then
					indicators[indicator] = {}
					if indicatorInfo.value > 0 then
						-- TODO: This could be made with a MultiGetSlice
						local key = indicator.."@"..handle.indicators[indicator].VCC.."\n15Min\n"..minirowDateUTC:toString("%Y-%m-%d %H:%M:%S")

						for _, fieldData in ipairs(handle.nonSliceFields) do
							local fieldName = fieldData.field or ""
							local fieldValue

							if fieldData.isColumn and columnValues[fieldName] then
								fieldValue = columnValues[fieldName]
							else
								fieldValue = handle.fieldsMap[fieldName]
							end
							fieldValue = fieldValue or ""

							-- El primer %s es el valor anterior de "key"
							-- El segundo %s es el nombre del campo del filtro
							-- El tercer %s es el valor del campo del filtro
							key = string.format("%s\n%s\n%s", key, fieldName, fieldValue)
						end

						--LogDebug("[REPORTER.NextPage]-a---> CF: SummaryMinMax, key %s", key)

						local maxColumns = summarizerKS:GetSliceRange(key, "SummaryMinMax", "", {
							slice_range = { start = "", finish = "", reversed = true, count = 1 }
						})
						for _, column in ipairs(maxColumns) do
							local comp = column.name_as_composite
							indicators[indicator].max = comp:GetNumber()
							-- Min value is only checked if there's a max value
							local minColumns = summarizerKS:GetSliceRange(key, "SummaryMinMax", "", {
								slice_range = { start = "", finish = "", reversed = false, count = 1 }
							})
							for _, column in ipairs(minColumns) do
								local comp = column.name_as_composite
								indicators[indicator].min = comp:GetNumber()
								break
							end
							break
						end
					end
				else
					indicators[indicator] = indicatorInfo.value
				end
			end
			--LogDebug("  CalculateIndicators - > indicators: %s:", json.encode(indicators))
			return indicators
		end

		-- Adds the minirow info to an existing row
		local function AddMinirowToRow(row, minirowIndicators)
			for indicator, value in pairs(minirowIndicators) do
				if not row.minirowIndicators[indicator] then
					row.minirowIndicators[indicator] = value
				elseif type(row.minirowIndicators[indicator]) == "table" then
					-- The indicator is a min-max value
					if row.minirowIndicators[indicator].min and value.min then
						row.minirowIndicators[indicator] = {
							min = math.min(value.min, row.minirowIndicators[indicator].min),
							max = math.max(value.max, row.minirowIndicators[indicator].max),
						}
					elseif value.min then
						row.minirowIndicators[indicator] = value
					end
				else
					row.minirowIndicators[indicator] = row.minirowIndicators[indicator] + value
				end
			end
		end

		-- Adds a new row to an existing megarow, containing the info of the minirow
		local function AddRowToMegarow(megarow, keyTable, columnValues, minirowIndicators)
			--LogDebug(" a - keyTable: %s", keyTable)
			megarow.rows[rowKey] = {
				keyTable = keyTable,
				columnValues = columnValues,
				minirowIndicators = minirowIndicators
			}
			--LogDebug("1--> add megarows. megarow.rowKeys: '%s'", json.encode(megarow.rowKeys or {}))
			table.insert(megarow.rowKeys, rowKey)
			result.length = result.length + 1
		end

		-- Check if the minirow fits on an existing row
		if lastMegarow and lastMegarow.key == megarowKey then
			if lastMegarow.rows[rowKey] then
				AddMinirowToRow(lastMegarow.rows[rowKey], CalculateIndicators())
				return true
			end
			-- It's a new row inside the megarow.
			AddRowToMegarow(lastMegarow, postSlice, columnValues, CalculateIndicators())
			return true
		end

		-- It's a new megarow. Fill up the megarows in between, if needed.
		-- This is done even if the next megarow won't fit in the query, so the sum of the pages returns the same data.
		if lastMegarow and math.abs(sliceStart:time() - lastMegarow.sliceStart:time()) <= rowFillMinutes then
			local sliceStartsToFill = summary_helper.GetSlicesInBetween(lastMegarow.sliceStart, sliceStart, handle.querySliceSize, handle.queryReverse)
			for _, sliceStartToFill in ipairs(sliceStartsToFill) do
				local sliceToFill = summary_helper.FormatDateForSlice(sliceStartToFill, handle.querySliceSize)
				local megarowKeyTable = summary_helper.ShallowCopy(preSlice)
				megarowKeyTable[#megarowKeyTable] = sliceToFill -- Replacing the slice of the next megarow
				--LogDebug(" b - keyTable: %s", megarowKeyTable)
				local fillMegarow = {
					key = table.concat(megarowKeyTable, "\n"),
					keyTable = megarowKeyTable,
					keyDay = megarowDayKey,
					sliceStart = sliceStartToFill,
					rows = {},
					rowKeys = {}
				}
				for rowKey, rowValues in pairs(lastMegarow.rows) do
					AddRowToMegarow(fillMegarow, rowValues.keyTable, rowValues.columnValues, {})
				end
				--LogDebug("2--> add megarows. fillMegarow: '%s'", json.encode(fillMegarow or {}))
				table.insert(megarows, fillMegarow)
			end
		end

		-- Check there aren't enough rows already.
		if result.length >= rows then
			return false
		end

		-- Create a new megarow
		--LogDebug(" c - keyTable: %s", preSlice)
		lastMegarow = {
			key = megarowKey,
			keyTable = preSlice,
			keyDay = megarowDayKey,
			sliceStart = sliceStart,
			rows = {},
			rowKeys = {}
		}
		AddRowToMegarow(lastMegarow, postSlice, columnValues, CalculateIndicators())
		--LogDebug("3--> add megarows. lastMegarow: '%s'", json.encode(lastMegarow or {}))
		table.insert(megarows, lastMegarow)
		return true
	end

	local function FormatIndicator (format, value)
		--LogDebug("  FormatIndicator format: '%s', value: '%s', rawData: '%s'", format , value , rawData)
		if not format or format.type == "default" then
			return value
		end
		if format.type == "Decimal" then
			value = tonumber(value) or 0
			if rawData then
				return value
			end
			return value * (format.multiplier or 1)
		end
		if format.type == "Time" then
			value = tonumber(value) or 0
			if rawData then
				return value
			end
			return slices_helper.formatTime(value * (format.multiplier or 1))
		end
		if format.type == "Percentage" then
			value = tonumber(value) or 0
			if rawData then
				return value
			end
			return ("%.2f%%"):format(value * (format.multiplier or 1))
		end
	end

	-- Once all rows are retrieved, this function processes all minirows, sorting them, calculating composite indicators and adding them to the output.
	local function ProcessRows()
		for _, megarow in ipairs(megarows) do
			-- Example megarow = {
			--    key = "sales | 2013-07-03 17:00 - 18:00",
			--    keyTable = { "sales", "2013-07-03 17:00 - 18:00" },
			--    keyDay = "sales | 2013-07-03",
			--    sliceStart = LuaOrm.Date("2013-07-03 17:00:00"),
			--    rows = {}, -- Table with rows
			--    rowKeys = {}, -- Array with row keys
			-- }

			if handle.queryReverse then
				table.sort(megarow.rowKeys, function(l, r) return l > r end)
			else
				table.sort(megarow.rowKeys)
			end
			for _, rowKey in ipairs(megarow.rowKeys) do
				local row = megarow.rows[rowKey]
				local rowOutput = {}

				-- Example row = {
				--    keyTable = { "fernickk", "IM" },
				--    columnValues = { Campaign = "support", Slice = "2013-07-03 14:00 - 15:00", Agent = "fernickk", Type = "IM" },
				--    minirowIndicators = { ["Inbound.Total"] = 1, ["Outbound.Total"] = 37 },
				-- }

				-- Add the cut columns to the row
				-- First the pre-slice and slice columns, which are stored at the megarow
				rowOutput = summary_helper.ShallowCopy(megarow.keyTable)
				--LogDebug("  a- rowOutput: %s", json.encode(rowOutput))
				-- Then the post-slice columns, which are stored on the row
				for _, elem in ipairs(row.keyTable) do
					table.insert(rowOutput, elem)
				end

				for index, value in pairs(rowOutput) do
					if handle.columnsMetadata[index] and handle.columnsMetadata[index].name then
						local indicatorName = handle.columnsMetadata[index].name
						if idsToNames[indicatorName] and idsToNames[indicatorName][value] then
							rowOutput[index] = idsToNames[indicatorName][value]
						end
					end
				end
				--LogDebug("  b- rowOutput: %s", json.encode(rowOutput))

				-- Calculate composite indicators
				local indicatorValues = {}

				local function calculateComposite(indicator, composite)
					-- Generate a sandbox environment to run the indicator script
					local env = {
						base = {},
						sliceSize = handle.querySliceSize,
						sliceSizeInMinutes = summary_helper.SliceSizeInMinutes(megarow.sliceStart, handle.querySliceSize, handle.fieldsMap.DateTo_d)
					}
					setmetatable(env, { __index = _G})
					-- Add the values of all base indicators to the environment. If an indicator has no value,
					-- add a 0 (or a { min = 0, max = 0 } if it's a min-max indicator).
					for _, subindicator in ipairs(composite.indicators) do

						if subindicator.composite then
						-- composite indicator, must check if it's already computed
							if not indicatorValues[subindicator.id] then
								calculateComposite(subindicator.id, subindicator.composite)
							end
							env.base[subindicator.id] = indicatorValues[subindicator.id].value
						else
						-- base indicator, just fetch the value from the minirow
							env.base[subindicator.id] =
								row.minirowIndicators[subindicator.id] or
								(handle.indicators[subindicator.id] and handle.indicators[subindicator.id].minMax and
								{ min = 0, max = 0 } or 0)
						end
					end
					-- Add the values of all the columns to the environment.
					for column, columnValue in pairs(row.columnValues) do
						env.base[column] = columnValue
					end
					setfenv(composite.fn, env)
					-- Execute the composite function
					local ok, result = pcall(composite.fn)

					indicatorValues[indicator] = ok and {
						format = composite.format,
						value = summary_helper.CheckNaN(result)
					} or {
						value = "ERROR: ".. result
					}
				end -- calculateComposite()

				for _, indicator in ipairs(handle.queryIndicators) do
					local indicatorId = manage.cleanupIndicatorId(indicator)

					local composite = handle.columnDetails[indicator].composite
					if composite and not indicatorValues[indicator] then
						calculateComposite(indicator, composite)
					else
						-- Set the value as the indicator to 0 if it has no value
						local indicatorValue = row.minirowIndicators[indicatorId] or 0
						if type(indicatorValue) == "table" then
							indicatorValue = handle.columnDetails[indicator].min and indicatorValue.min or indicatorValue.max
						end
						indicatorValues[indicator] = {
							value  = indicatorValue,
							format = handle.indicators[indicatorId] and handle.indicators[indicatorId].format
						}
					end
				end

				--	LogDebug(" 0.5---> indicatorValues: %s", json.encode(indicatorValues))
				--LogDebug("  1 antes ->rowOutput: %s", json.encode(rowOutput))

				-- Add the indicator columns to the row
				for _, indicator in ipairs(handle.queryIndicators) do
					local indicatorValue  = indicatorValues[indicator]
					local valueWithFormat = FormatIndicator(indicatorValue.format, indicatorValue.value) or 0
					--LogDebug("   -> rawValue: '%s' - valueWithFormat: '%s'", indicatorValue.value, valueWithFormat)
					table.insert(rowOutput, valueWithFormat)
				end
				-- Add the row to the result
				--LogDebug("  2 despues ->rowOutput: %s", json.encode(rowOutput))
				table.insert(result.rows, rowOutput)
			end
		end
	end

	-- Outer loop continues while there are domain keys to process
	while true do
		-- Query the columns for each key, indicatorKey by indicatorKey
		for i = handle.currentIndicatorKey, #handle.sortedIndicatorKeys do
			local indicatorKey = handle.sortedIndicatorKeys[i]
			local indicatorData = handle.indicatorKeys[indicatorKey]

			local function GetMoreMiniRows(keyResults, key, initialColumn, indicator)
				keyResults.cursor = 1
				keyResults.rows = {}

				local firstCol = initialColumn or (indicatorData.rowPrefix .. (handle.queryReverse and handle.fieldsMap.DateTo or handle.fieldsMap.DateFrom))
				local lastCol = indicatorData.rowPrefix .. (handle.queryReverse and handle.fieldsMap.DateFrom or handle.fieldsMap.DateTo)

				--LogDebug("[REPORTER.NextPage]-b---> %d - CF: Summaries, key %s, firstCol %s, lastCol %s", i, key, firstCol, lastCol)

				summary_helper.GetColumns(summarizerKS,
					key,
					"Summaries",
					firstCol,
					lastCol,
					rows + 1,
					handle.queryReverse,
					function(column)
						-- Dejo comentadas las lineas que loguean la key, column y valor que utiliza el reporte.
						-- Descomentando esto es posible obtener y luego modificar los valores del reporte.
						--local keyToLog = key:gsub("\n", "\\n")
						--local columnName = column and column.name or ""
						--local columnNameToLog = columnName and columnName:gsub("\n", "\\n") or ""
						--LogDebug("LOG_RESULT_REPORT => key: %s, columnName: %s, value: %s",keyToLog, columnNameToLog, column.value_as_number)
						table.insert(keyResults.rows, { indicator = indicator, name = column.name, value = column.value_as_number })
					end)
			end

			-- Get enough minirows of all indicators for this indicator key
			local allResults = {}
			for indicator, indicatorAttributes in pairs(handle.indicators) do
				local indicadorKey = string.format("%s@%s", indicator, indicatorAttributes.VCC)
				local cut = string.format("%s@%s", handle.queryCut, handle.cutVCC)

				key = string.format("%s\n%s\n%s", indicadorKey, cut, indicatorKey)

				--LogDebug("a   ---> key: %s", key)
				-- Example key = Abandoned@inconcert|SliceCampaign@system|2013-07|15Min|support|inconcert
				local keyResults = {}
				GetMoreMiniRows(keyResults, key, handle.nextCol, indicator)
				if #keyResults.rows > 0 then
					allResults[key] = keyResults
				end
			end

			-- Get complete minirows from all the indicators, until all required minirows are retrieved, or no more exist for this indicatorKey
			while true do
				local bestMiniRow
				local bestIndicators = {}
				local function BetterMinirowName(candidate)
					if not bestMiniRow then return true end
					if handle.queryReverse then
						return candidate > bestMiniRow
					end
					return candidate < bestMiniRow
				end

				-- From each indicator, get which of them share the smallest name (stored in bestMiniRow), and collect their values on bestIndicators
				for key, keyResults in pairs(allResults) do
					if #keyResults.rows >= keyResults.cursor and keyResults.cursor > rows then
						-- The cursor is beyond the existing rows, but there are more rows that can be retrieved from Cassandra.
						local pivot = keyResults.rows[rows+1]
						--LogDebug("b-> key: %s", key)
						GetMoreMiniRows(keyResults, key, pivot.name, pivot.indicator)
					end
					if #keyResults.rows >= keyResults.cursor then
						-- The cursor is pointing to an existing row
						local row = keyResults.rows[keyResults.cursor]
						if BetterMinirowName(row.name) then
							bestMiniRow = row.name
							bestIndicators = {}
							bestIndicators[row.indicator] = { key = key, value = row.value }
						elseif row.name == bestMiniRow then
							bestIndicators[row.indicator] = { key = key, value = row.value }
						end
					end
				end

				if not bestMiniRow then
					-- No more minirows for the queried indicators on the current indicator key
					break
				end

				-- Try to add the minirow to the megarows
				local added = AddMiniRow(bestMiniRow, bestIndicators)

				if not added then
					-- All rows needed were retrieved, and there are more rows for the current indicator key
					handle.nextCol = bestMiniRow
					ProcessRows()
					RemoveCompositeFunctions()
					return result, handle
				end

				-- Increment the cursor in all indicators
				for _, bestIndicatorData in pairs(bestIndicators) do
					allResults[bestIndicatorData.key].cursor = allResults[bestIndicatorData.key].cursor + 1
				end
			end

			-- No more rows on all the indicators for this indicatorKey, go with the next one
			handle.currentIndicatorKey = handle.currentIndicatorKey + 1
			handle.nextCol = nil
		end

		-- Check if there are more indicator keys
		if not handle.baseDomainKey then
			break
		end
		if not handle.nextDomainKey then
			break
		end

		handle.sortedIndicatorKeys = {}
		handle.indicatorKeys = {}
		handle.currentIndicatorKey = 1
		local domainMonths = {}
		local sortedDomains = {}
		local domainSliceSize = handle.filterWithSomeCampaigns == false and 1000 or 10

		for _, month in ipairs(handle.months) do
			for i, baseDomainKey in ipairs(handle.baseDomainKey or {}) do
				local key = baseDomainKey.."\n"..month.."\n"..handle.domainSuffix
				local queryDomainSliceSize = domainSliceSize + 1
				--LogDebug("[REPORTER.NextPage]-c---> %d - CF: SummaryDomains, key %s, nextDomainKey: '%s', queryDomainSliceSize: %d", _, key, nextDomainKey, queryDomainSliceSize)

				summary_helper.GetColumns(summarizerKS, key, "SummaryDomains", handle.nextDomainKey, "", queryDomainSliceSize, handle.queryReverse, function(column)
					if not domainMonths[column.name] then
						domainMonths[column.name] = {}
						table.insert(sortedDomains, column.name)
					end
					table.insert(domainMonths[column.name], month)
				end)
			end
		end

		if handle.queryReverse then
			table.sort(sortedDomains, function(l, r) return l > r end)
		else
			table.sort(sortedDomains)
		end

		handle.nextDomainKey = nil
		--LogDebug("[REPORTER.NextPage] 1--> handle.nextDomainKey: %s", handle.nextDomainKey)
		for idxD, d in ipairs(sortedDomains) do
			if idxD > domainSliceSize then
				handle.nextDomainKey = d
				--LogDebug("[REPORTER.NextPage] 2--> handle.nextDomainKey: %s", handle.nextDomainKey)
				break
			end

			for _, m in ipairs(domainMonths[d]) do
				for _, suffix in ipairs(handle.keySuffix) do
					local indicatorKey = d.."\n"..m.."\n15Min"..suffix
					handle.indicatorKeys[indicatorKey] = {
						rowPrefix = d.."\n"
					}
					table.insert(handle.sortedIndicatorKeys, indicatorKey)
				end
			end
		end
	end

	-- Nothing else to query
	handle.complete = true
--	LogDebug("[REPORTER.SUMMARY_QUERY:NextPage] 1 result %s", json.encode(result or {}))
	ProcessRows()
--	LogDebug("[REPORTER.SUMMARY_QUERY:NextPage] 2 result %s", json.encode(result or {}))
	RemoveCompositeFunctions()
--	LogDebug("[REPORTER.SUMMARY_QUERY:NextPage] 3 result %s", json.encode(result or {}))

	return result, handle
end

