@Override
    public void execute(ComplexEventChunk streamEventChunk) {
        if (log.isDebugEnabled()) {
            log.debug("Event Chunk received by the Aggregation " + aggregatorName + " for duration " + this.duration +
                    " will be dropped since persisted aggregation has been scheduled ");
        }
        streamEventChunk.reset();
        while (streamEventChunk.hasNext()) {
            StreamEvent streamEvent = (StreamEvent) streamEventChunk.next();
            streamEventChunk.remove();
            ExecutorState executorState = stateHolder.getState();
            try {
                long timestamp = getTimestamp(streamEvent);
                if (timestamp >= executorState.nextEmitTime) {
                    log.debug("Next EmitTime: " + executorState.nextEmitTime + ", Current Time: " + timestamp);
                    long emittedTime = executorState.nextEmitTime;
                    long startedTime = executorState.startTimeOfAggregates;
                    executorState.startTimeOfAggregates = IncrementalTimeConverterUtil.getStartTimeOfAggregates(
                            timestamp, duration, timeZone);
                    executorState.nextEmitTime = IncrementalTimeConverterUtil.getNextEmitTime(
                            timestamp, duration, timeZone);
                    dispatchAggregateEvents(startedTime, emittedTime, timeZone);
                    sendTimerEvent(executorState);
                }
            } finally {
                stateHolder.returnState(executorState);
            }
        }
    }
Incremental Executor implementation class for Persisted Aggregation
modules/siddhi-core/src/main/java/io/siddhi/core/aggregation/persistedaggregation/PersistedIncrementalExecutor.java