private void dispatchEvent(long startTimeOfNewAggregates, BaseIncrementalValueStore aBaseIncrementalValueStore) {
        AtomicBoolean isProcessFinished = new AtomicBoolean(false);
        if (aBaseIncrementalValueStore.isProcessed()) {
            Map<String, StreamEvent> streamEventMap = aBaseIncrementalValueStore.getGroupedByEvents();
            ComplexEventChunk<StreamEvent> eventChunk = new ComplexEventChunk<>();
            for (StreamEvent event : streamEventMap.values()) {
                eventChunk.add(event);
            }
            Map<String, StreamEvent> tableStreamEventMap = aBaseIncrementalValueStore.getGroupedByEvents();
            ComplexEventChunk<StreamEvent> tableEventChunk = new ComplexEventChunk<>();
            for (StreamEvent event : tableStreamEventMap.values()) {
                tableEventChunk.add(event);
            }
            if (LOG.isDebugEnabled()) {
                LOG.debug("Event dispatched by aggregation " + aggregatorName + " for duration " + this.duration);
            }
            if (isProcessingExecutor) {
                executorService.execute(() -> {
                    try {
                        table.addEvents(tableEventChunk, streamEventMap.size());
                    } catch (Throwable t) {
                        LOG.error("Exception occurred at siddhi app '" + this.siddhiAppName +
                                "' when performing table writes of aggregation '" + this.aggregatorName +
                                "' for duration '" + this.duration + "'. Unless state persistence has been enabled, " +
                                "this should be investigated as this can cause accuracy loss.", t);
                        if (LOG.isDebugEnabled()) {
                            LOG.debug("Dropping Event chunk - \"" + eventChunk.toString() + "\"");
                        }
                    } finally {
                        isProcessFinished.set(true);
                    }
                });

            }
            if (waitUntillprocessFinish) {
                try {
                    while (!isProcessFinished.get()) {
                        Thread.sleep(1000);
                    }
                } catch (InterruptedException e) {
                    LOG.error("Error occurred while waiting until table update task finishes for duration " +
                            duration + "in aggregation " + aggregatorName, e);
                }
            }
            if (getNextExecutor() != null) {
                next.execute(eventChunk);
            }
        }
        cleanBaseIncrementalValueStore(startTimeOfNewAggregates, aBaseIncrementalValueStore);
    }
Incremental executor class which is responsible for performing incremental aggregation.
modules/siddhi-core/src/main/java/io/siddhi/core/aggregation/IncrementalExecutor.java