private static Map<TimePeriod.Duration, Processor> initAggregateQueryExecutor(
            List<TimePeriod.Duration> aggregationDurations,
            Map<TimePeriod.Duration, List<ExpressionExecutor>> processExpressionExecutorsMap,
            StreamDefinition incomingOutputStreamDefinition, boolean isDistributed,
            String shardID, boolean isProcessingOnExternalTime, SiddhiQueryContext siddhiQueryContext,
            AggregationDefinition aggregationDefinition, ConfigManager configManager,
            Map<TimePeriod.Duration, Table> aggregationTables, List<Variable> groupByVariableList) {

        Map<TimePeriod.Duration, Processor> cudProcessors = new LinkedHashMap<>();
        String datasourceName = AnnotationHelper.getAnnotationElement(SiddhiConstants.NAMESPACE_STORE,
                "datasource", aggregationDefinition.getAnnotations()).getValue();
        if (datasourceName == null || datasourceName.isEmpty()) {
            throw new SiddhiAppCreationException("Datasource configuration must be provided inorder to use persisted " +
                    "aggregation mode");
        }
        Database databaseType = getDatabaseType(configManager, datasourceName);
        if (log.isDebugEnabled()) {
            log.debug("Database type " + databaseType);
        }
        SiddhiAppContext cudSiddhiAppContext = new SiddhiAppContext();
        SiddhiContext context = new SiddhiContext();
        context.setConfigManager(configManager);
        cudSiddhiAppContext.setSiddhiContext(context);
        StringConstant datasource = new StringConstant(datasourceName);
        ConstantExpressionExecutor datasourceExecutor = new ConstantExpressionExecutor(datasource.getValue(),
                Attribute.Type.STRING);
        Expression[] streamHandler;
        ExpressionExecutor[] cudStreamProcessorInputVariables;
        if (isProcessingOnExternalTime) {
            streamHandler = new Expression[7];
        } else {
            streamHandler = new Expression[5];
        }
        try {
            DBAggregationQueryConfigurationEntry dbAggregationQueryConfigurationEntry = DBAggregationQueryUtil.
                    lookupCurrentQueryConfigurationEntry(databaseType);
            if (log.isDebugEnabled()) {
                log.debug("CUD queries for aggregation " + aggregationDefinition.getId());
            }
            for (int i = aggregationDurations.size() - 1; i > 0; i--) {
                if (aggregationDurations.get(i).ordinal() >= 3) {
                    if (log.isDebugEnabled()) {
                        log.debug(" Initializing cudProcessors for duration " + aggregationDurations.get(i));
                    }
                    String databaseSelectQuery = generateDatabaseQuery(processExpressionExecutorsMap.
                                    get(aggregationDurations.get(i)), dbAggregationQueryConfigurationEntry,
                            incomingOutputStreamDefinition, isDistributed, shardID, isProcessingOnExternalTime,
                            aggregationTables.get(aggregationDurations.get(i)),
                            aggregationTables.get(aggregationDurations.get(i - 1)), groupByVariableList,
                            aggregationDurations.get(i));
                    StringConstant selectQuery = new StringConstant(databaseSelectQuery);
                    if (log.isDebugEnabled()) {
                        log.debug(selectQuery);
                    }
                    ConstantExpressionExecutor selectExecutor = new ConstantExpressionExecutor(selectQuery.getValue(),
                            Attribute.Type.STRING);
                    Map<Attribute, int[]> cudInputStreamAttributesMap =
                            generateCUDInputStreamAttributes(isProcessingOnExternalTime);
                    if (isProcessingOnExternalTime) {
                        cudStreamProcessorInputVariables = new ExpressionExecutor[7];
                    } else {
                        cudStreamProcessorInputVariables = new ExpressionExecutor[5];
                    }
                    cudStreamProcessorInputVariables[0] = datasourceExecutor;
                    cudStreamProcessorInputVariables[1] = selectExecutor;
                    streamHandler[0] = datasource;
                    streamHandler[1] = selectQuery;
                    MetaStreamEvent metaStreamEvent = generateCUDMetaStreamEvent(isProcessingOnExternalTime);
                    StreamDefinition outputStream = new StreamDefinition();
                    VariableExpressionExecutor variableExpressionExecutor;
                    int j = 0;
                    for (Map.Entry<Attribute, int[]> entry : cudInputStreamAttributesMap.entrySet()) {
                        Attribute attribute = entry.getKey();
                        Variable timestampVariable = new Variable(attribute.getName());
                        for (int position : entry.getValue()) {
                            streamHandler[position + 2] = timestampVariable;
                            variableExpressionExecutor = new VariableExpressionExecutor(attribute, 0, 0);
                            variableExpressionExecutor.setPosition(new int[]{2, j});
                            cudStreamProcessorInputVariables[position + 2] = variableExpressionExecutor;
                        }
                        outputStream.attribute(attribute.getName(), attribute.getType());
                        j++;
                    }

                    StreamFunction cudStreamFunction = new StreamFunction(NAMESPACE_RDBMS, FUNCTION_NAME_CUD,
                            streamHandler);

                    cudProcessors.put(aggregationDurations.get(i), getCudProcessor(cudStreamFunction, siddhiQueryContext,
                            metaStreamEvent, cudStreamProcessorInputVariables, aggregationDurations.get(i)));
                }
            }
            return cudProcessors;
        } catch (CannotLoadConfigurationException e) {
            throw new SiddhiAppCreationException("Error occurred while initializing the persisted incremental " +
                    "aggregation. Could not load the db quires for database type " + databaseType);
        }
    }
This is the parser class of incremental aggregation definition.
modules/siddhi-core/src/main/java/io/siddhi/core/util/parser/AggregationParser.java