@Override
    public synchronized void execute(ComplexEventChunk streamEventChunk) {
        if (LOG.isDebugEnabled()) {
            LOG.debug("Event Chunk received by " + this.duration + " incremental executor: " +
                    streamEventChunk.toString());
        }
        streamEventChunk.reset();
        while (streamEventChunk.hasNext()) {
            StreamEvent streamEvent = (StreamEvent) streamEventChunk.next();
            streamEventChunk.remove();
            ExecutorState executorState = stateHolder.getState();
            try {
                long timestamp = getTimestamp(streamEvent, executorState);
                long startTime = executorState.startTimeOfAggregates;
                executorState.startTimeOfAggregates = IncrementalTimeConverterUtil.getStartTimeOfAggregates(
                        timestamp, duration, timeZone);
                if (timestamp >= executorState.nextEmitTime) {
                    executorState.nextEmitTime = IncrementalTimeConverterUtil.getNextEmitTime(
                            timestamp, duration, timeZone);
                    dispatchAggregateEvents(executorState.startTimeOfAggregates);
                    sendTimerEvent(executorState);
                }
                if (streamEvent.getType() == ComplexEvent.Type.CURRENT) {
                    processAggregates(streamEvent, executorState);
                }
            } finally {
                stateHolder.returnState(executorState);
            }
        }
    }
Incremental executor class which is responsible for performing incremental aggregation.
modules/siddhi-core/src/main/java/io/siddhi/core/aggregation/IncrementalExecutor.java