public ComplexEventChunk<StreamEvent> aggregateInMemoryData(
            Map<TimePeriod.Duration, Executor> incrementalExecutorMap) {
        int startIndex = incrementalDurations.indexOf(durationToAggregate);
        Set<String> groupByKeys = new HashSet<>();
        for (int k = startIndex; k >= 0; k--) {
            TimePeriod.Duration duration = incrementalDurations.get(k);
            Executor incrementalExecutor = incrementalExecutorMap.get(duration);
            if (incrementalExecutor instanceof IncrementalExecutor) {
                BaseIncrementalValueStore aBaseIncrementalValueStore = ((IncrementalExecutor) incrementalExecutor)
                        .getBaseIncrementalValueStore();
                Map<String, StreamEvent> groupedByEvents = aBaseIncrementalValueStore.getGroupedByEvents();
                for (Map.Entry<String, StreamEvent> eventEntry : groupedByEvents.entrySet()) {
                    long startTimeOfAggregates = IncrementalTimeConverterUtil.getStartTimeOfAggregates(
                            eventEntry.getValue().getTimestamp(), durationToAggregate, timeZone);
                    String groupByKey = eventEntry.getKey() + "-" + startTimeOfAggregates;
                    synchronized (this) {
                        groupByKeys.add(groupByKey);
                        SiddhiAppContext.startGroupByFlow(groupByKey);
                        ValueState state = (ValueState) valueStateHolder.getState();
                        try {
                            boolean shouldUpdate = true;
                            if (shouldUpdateTimestamp != null) {
                                shouldUpdate = shouldUpdate(shouldUpdateTimestamp.execute(eventEntry.getValue()),
                                        state);
                            } else {
                                state.lastTimestamp = oldestEventTimestamp;
                            }
                            // keeping timestamp value location as null
                            for (int i = 0; i < baseExecutorsForFind.size(); i++) {
                                ExpressionExecutor expressionExecutor = baseExecutorsForFind.get(i);
                                if (shouldUpdate) {
                                    state.setValue(expressionExecutor.execute(eventEntry.getValue()), i + 1);
                                } else if (!(expressionExecutor instanceof VariableExpressionExecutor)) {
                                    state.setValue(expressionExecutor.execute(eventEntry.getValue()), i + 1);
                                }
                            }
                        } finally {
                            valueStateHolder.returnState(state);
                            SiddhiAppContext.stopGroupByFlow();
                        }
                    }
                }
            }
        }
        //clean all executors
        for (String groupByKey : groupByKeys) {
            SiddhiAppContext.startGroupByFlow(groupByKey);
            try {
                for (ExpressionExecutor expressionExecutor : baseExecutorsForFind) {
                    expressionExecutor.execute(resetEvent);
                }
            } finally {
                SiddhiAppContext.stopGroupByFlow();
            }
        }
        return getProcessedEventChunk();
    }
This class implements the logic to aggregate data that is in-memory or in tables, in incremental data processing.
<p>
In-memory data is required to be aggregated when retrieving values from an aggregate.
Table data (tables used to persist incremental aggregates) needs to be aggregated when querying aggregate data from
a different server (apart from the server, which was used to define the aggregation).
modules/siddhi-core/src/main/java/io/siddhi/core/aggregation/IncrementalDataAggregator.java