public synchronized void initialiseExecutors() {
        if (this.isInitialised || isReadOnly) {
            // Only cleared when executors change from reading to processing state in one node deployment
            return;
        }
        log.info("Starting initialization of executors for aggregation: " + aggregationDefinition.getId());
        Event[] events;
        Long lastData = null;

        // Get max(AGG_TIMESTAMP) from table corresponding to max duration
        Table tableForMaxDuration = aggregationTables.get(incrementalDurations.get(incrementalDurations.size() - 1));
        OnDemandQuery onDemandQuery = getOnDemandQuery(tableForMaxDuration, true, endOFLatestEventTimestamp, false,
                OrderByAttribute.Order.DESC);
        onDemandQuery.setType(OnDemandQuery.OnDemandQueryType.FIND);
        OnDemandQueryRuntime onDemandQueryRuntime = OnDemandQueryParser.parse(onDemandQuery, null,
                siddhiAppContext, tableMap, windowMap, aggregationMap);

        // Get latest event timestamp in tableForMaxDuration and get the end time of the aggregation record
        events = onDemandQueryRuntime.execute();
        if (events != null) {
            lastData = (Long) events[events.length - 1].getData(0);
            endOFLatestEventTimestamp = IncrementalTimeConverterUtil
                    .getNextEmitTime(lastData, incrementalDurations.get(incrementalDurations.size() - 1), timeZone);
        }

        if (isPersistedAggregation) {
            for (int i = incrementalDurations.size() - 1; i > 0; i--) {
                if (lastData != null && !(IncrementalTimeConverterUtil.isAggregationDataCompleteAgainstTime(lastData,
                        incrementalDurations.get(i), timeZone) && isAggregationDataComplete(incrementalDurations.get(i),
                        incrementalDurations.get(i - 1), endOFLatestEventTimestamp, lastData))) {
                    recreateState(lastData, incrementalDurations.get(i),
                            aggregationTables.get(incrementalDurations.get(i - 1)), i == 1);
                } else if (lastData == null && !isAggregationDataComplete(incrementalDurations.get(i),
                        incrementalDurations.get(i - 1), endOFLatestEventTimestamp, null)) {
                    recreateState(null, incrementalDurations.get(i),
                            aggregationTables.get(incrementalDurations.get(i - 1)), i == 1);
                }
                if (i > 1) {
                    onDemandQuery = getOnDemandQuery(aggregationTables.get(incrementalDurations.get(i - 1)), true,
                            endOFLatestEventTimestamp, true, OrderByAttribute.Order.DESC);
                    onDemandQuery.setType(OnDemandQuery.OnDemandQueryType.FIND);
                    onDemandQueryRuntime = OnDemandQueryParser.parse(onDemandQuery, null,
                            siddhiAppContext, tableMap, windowMap, aggregationMap);
                    events = onDemandQueryRuntime.execute();
                    if (events != null) {
                        lastData = (Long) events[events.length - 1].getData(0);
                    } else {
                        lastData = null;
                    }
                }
            }
        } else {
            for (int i = incrementalDurations.size() - 1; i > 0; i--) {

                // Get the table previous to the duration for which we need to recreate (e.g. if we want to recreate
                // for minute duration, take the second table [provided that aggregation is done for seconds])
                // This lookup is filtered by endOFLatestEventTimestamp
                Table recreateFromTable = aggregationTables.get(incrementalDurations.get(i - 1));

                onDemandQuery = getOnDemandQuery(recreateFromTable, false, endOFLatestEventTimestamp, false,
                        OrderByAttribute.Order.DESC);
                onDemandQuery.setType(OnDemandQuery.OnDemandQueryType.FIND);
                onDemandQueryRuntime = OnDemandQueryParser.parse(onDemandQuery, null, siddhiAppContext,
                        tableMap, windowMap,
                        aggregationMap);
                events = onDemandQueryRuntime.execute();

                if (events != null) {
                    long referenceToNextLatestEvent = (Long) events[events.length - 1].getData(0);
                    endOFLatestEventTimestamp = IncrementalTimeConverterUtil
                            .getNextEmitTime(referenceToNextLatestEvent, incrementalDurations.get(i - 1), timeZone);

                    TimePeriod.Duration recreateForDuration = incrementalDurations.get(i);
                    if (!isStatePresentForAggregationDuration(recreateForDuration)) {
                        ComplexEventChunk<StreamEvent> complexEventChunk = new ComplexEventChunk<>();
                        for (Event event : events) {
                            StreamEvent streamEvent = streamEventFactory.newInstance();
                            streamEvent.setOutputData(event.getData());
                            complexEventChunk.add(streamEvent);
                        }
                        Executor incrementalExecutor = incrementalExecutorMap.get(recreateForDuration);
                        incrementalExecutor.execute(complexEventChunk);
                    }

                    if (i == 1) {
                        TimePeriod.Duration rootDuration = incrementalDurations.get(0);
                        Executor rootIncrementalExecutor = incrementalExecutorMap.get(rootDuration);
                        long emitTimeOfLatestEventInTable = IncrementalTimeConverterUtil.getNextEmitTime(
                                referenceToNextLatestEvent, rootDuration, timeZone);

                        rootIncrementalExecutor.setEmitTime(emitTimeOfLatestEventInTable);

                    }
                }
            }
        }
        this.isInitialised = true;
        log.info("Finished initialization of executors for aggregation: " + aggregationDefinition.getId());
    }
This class is used to recreate in-memory data from the tables (Such as RDBMS) in incremental aggregation.
This ensures that the aggregation calculations are done correctly in case of server restart.
modules/siddhi-core/src/main/java/io/siddhi/core/aggregation/IncrementalExecutorsInitialiser.java