public CompiledCondition compileExpression(Expression expression, Within within, Expression per,
                                               List<Variable> queryGroupByList,
                                               MatchingMetaInfoHolder matchingMetaInfoHolder,
                                               List<VariableExpressionExecutor> variableExpressionExecutors,
                                               Map<String, Table> tableMap, SiddhiQueryContext siddhiQueryContext) {

        String aggregationName = aggregationDefinition.getId();
        boolean isOptimisedTableLookup = isOptimisedLookup;
        Map<TimePeriod.Duration, CompiledCondition> withinTableCompiledConditions = new HashMap<>();
        CompiledCondition withinInMemoryCompileCondition;
        CompiledCondition onCompiledCondition;
        List<Attribute> additionalAttributes = new ArrayList<>();

        // Define additional attribute list
        additionalAttributes.add(new Attribute("_START", Attribute.Type.LONG));
        additionalAttributes.add(new Attribute("_END", Attribute.Type.LONG));

        int lowerGranularitySize = this.activeIncrementalDurations.size() - 1;
        List<String> lowerGranularityAttributes = new ArrayList<>();
        if (isDistributed) {
            //Add additional attributes to get base aggregation timestamps based on current timestamps
            // for values calculated in in-memory in the shards
            for (int i = 0; i < lowerGranularitySize; i++) {
                String attributeName = "_AGG_TIMESTAMP_FILTER_" + i;
                additionalAttributes.add(new Attribute(attributeName, Attribute.Type.LONG));
                lowerGranularityAttributes.add(attributeName);
            }
        }

        // Get table definition. Table definitions for all the tables used to persist aggregates are similar.
        // Therefore it's enough to get the definition from one table.
        AbstractDefinition tableDefinition = aggregationTables.get(activeIncrementalDurations.get(0)).
                getTableDefinition();

        boolean isOnDemandQuery = matchingMetaInfoHolder.getMetaStateEvent().getMetaStreamEvents().length == 1;

        // Alter existing meta stream event or create new one if a meta stream doesn't exist
        // After calling this method the original MatchingMetaInfoHolder's meta stream event would be altered
        // Alter meta info holder to contain stream event and aggregate both when it's a on-demand query
        MetaStreamEvent metaStreamEventForTableLookups;
        if (isOnDemandQuery) {
            metaStreamEventForTableLookups = alterMetaStreamEvent(true, new MetaStreamEvent(), additionalAttributes);
            matchingMetaInfoHolder = alterMetaInfoHolderForOnDemandQuery(metaStreamEventForTableLookups,
                    matchingMetaInfoHolder);
        } else {
            metaStreamEventForTableLookups = alterMetaStreamEvent(false,
                    matchingMetaInfoHolder.getMetaStateEvent().getMetaStreamEvent(0), additionalAttributes);
        }


        // Create new MatchingMetaInfoHolder containing newMetaStreamEventWithStartEnd and table meta event
        String aggReferenceId = matchingMetaInfoHolder.getMetaStateEvent().getMetaStreamEvent(1).getInputReferenceId();
        String referenceName = aggReferenceId == null ? aggregationName : aggReferenceId;

        MetaStreamEvent metaStoreEventForTableLookups = createMetaStoreEvent(tableDefinition, referenceName);

        // Create new MatchingMetaInfoHolder containing metaStreamEventForTableLookups and table meta event
        MatchingMetaInfoHolder metaInfoHolderForTableLookups = createNewStreamTableMetaInfoHolder(
                metaStreamEventForTableLookups, metaStoreEventForTableLookups);

        // Create per expression executor
        ExpressionExecutor perExpressionExecutor;
        if (per != null) {
            perExpressionExecutor = ExpressionParser.parseExpression(per, matchingMetaInfoHolder.getMetaStateEvent(),
                    matchingMetaInfoHolder.getCurrentState(), tableMap, variableExpressionExecutors,
                    false, 0, ProcessingMode.BATCH, false, siddhiQueryContext);
            if (perExpressionExecutor.getReturnType() != Attribute.Type.STRING) {
                throw new SiddhiAppCreationException(
                        "Query " + siddhiQueryContext.getName() + "'s per value expected a string but found "
                                + perExpressionExecutor.getReturnType(),
                        per.getQueryContextStartIndex(), per.getQueryContextEndIndex());
            }
            // Additional Per time function verification at compile time if it is a constant
            if (perExpressionExecutor instanceof ConstantExpressionExecutor) {
                String perValue = ((ConstantExpressionExecutor) perExpressionExecutor).getValue().toString();
                try {
                    normalizeDuration(perValue);
                } catch (SiddhiAppValidationException e) {
                    throw new SiddhiAppValidationException(
                            "Aggregation Query's per value is expected to be of a valid time function of the " +
                                    "following " + TimePeriod.Duration.SECONDS + ", " + TimePeriod.Duration.MINUTES
                                    + ", " + TimePeriod.Duration.HOURS + ", " + TimePeriod.Duration.DAYS + ", "
                                    + TimePeriod.Duration.MONTHS + ", " + TimePeriod.Duration.YEARS + ".");
                }
            }
        } else {
            throw new SiddhiAppCreationException("Syntax Error: Aggregation join query must contain a `per` " +
                    "definition for granularity");
        }

        // Create start and end time expression
        Expression startEndTimeExpression;
        ExpressionExecutor startTimeEndTimeExpressionExecutor;
        if (within != null) {
            if (within.getTimeRange().size() == 1) {
                startEndTimeExpression = new AttributeFunction("incrementalAggregator",
                        "startTimeEndTime", within.getTimeRange().get(0));
            } else { // within.getTimeRange().size() == 2
                startEndTimeExpression = new AttributeFunction("incrementalAggregator",
                        "startTimeEndTime", within.getTimeRange().get(0), within.getTimeRange().get(1));
            }
            startTimeEndTimeExpressionExecutor = ExpressionParser.parseExpression(startEndTimeExpression,
                    matchingMetaInfoHolder.getMetaStateEvent(), matchingMetaInfoHolder.getCurrentState(), tableMap,
                    variableExpressionExecutors, false, 0,
                    ProcessingMode.BATCH, false, siddhiQueryContext);
        } else {
            throw new SiddhiAppCreationException("Syntax Error : Aggregation read query must contain a `within` " +
                    "definition for filtering of aggregation data.");
        }

        // Create within expression
        Expression timeFilterExpression;
        if (isProcessingOnExternalTime) {
            timeFilterExpression = Expression.variable(AGG_EXTERNAL_TIMESTAMP_COL);
        } else {
            timeFilterExpression = Expression.variable(AGG_START_TIMESTAMP_COL);
        }
        Expression withinExpression;
        Expression start = Expression.variable(additionalAttributes.get(0).getName());
        Expression end = Expression.variable(additionalAttributes.get(1).getName());
        Expression compareWithStartTime = Compare.compare(start, Compare.Operator.LESS_THAN_EQUAL,
                timeFilterExpression);
        Expression compareWithEndTime = Compare.compare(timeFilterExpression, Compare.Operator.LESS_THAN, end);
        withinExpression = Expression.and(compareWithStartTime, compareWithEndTime);

        List<ExpressionExecutor> timestampFilterExecutors = new ArrayList<>();
        if (isDistributed) {
            for (int i = 0; i < lowerGranularitySize; i++) {
                Expression[] expressionArray = new Expression[]{
                        new AttributeFunction("", "currentTimeMillis", null),
                        Expression.value(this.activeIncrementalDurations.get(i + 1).toString())};
                Expression filterExpression = new AttributeFunction("incrementalAggregator",
                        "getAggregationStartTime", expressionArray);
                timestampFilterExecutors.add(ExpressionParser.parseExpression(filterExpression,
                        matchingMetaInfoHolder.getMetaStateEvent(), matchingMetaInfoHolder.getCurrentState(), tableMap,
                        variableExpressionExecutors, false, 0,
                        ProcessingMode.BATCH, false, siddhiQueryContext));
            }
        }

        // Create compile condition per each table used to persist aggregates.
        // These compile conditions are used to check whether the aggregates in tables are within the given duration.
        // Combine with and on condition for table query
        boolean shouldApplyReducedCondition = false;
        Expression reducedExpression = null;

        //Check if there is no on conditions
        if (!(expression instanceof BoolConstant)) {
            // For abstract queryable table
            AggregationExpressionBuilder aggregationExpressionBuilder = new AggregationExpressionBuilder(expression);
            AggregationExpressionVisitor expressionVisitor = new AggregationExpressionVisitor(
                    metaStreamEventForTableLookups.getInputReferenceId(),
                    metaStreamEventForTableLookups.getLastInputDefinition().getAttributeList(),
                    this.tableAttributesNameList
            );
            aggregationExpressionBuilder.build(expressionVisitor);
            shouldApplyReducedCondition = expressionVisitor.applyReducedExpression();
            reducedExpression = expressionVisitor.getReducedExpression();
        }

        Expression withinExpressionTable;
        if (shouldApplyReducedCondition) {
            withinExpressionTable = Expression.and(withinExpression, reducedExpression);
        } else {
            withinExpressionTable = withinExpression;
        }

        List<Variable> queryGroupByListCopy = new ArrayList<>(queryGroupByList);

        Variable timestampVariable = new Variable(AGG_START_TIMESTAMP_COL);
        List<String> queryGroupByNamesList = queryGroupByListCopy.stream()
                .map(Variable::getAttributeName)
                .collect(Collectors.toList());
        boolean queryGroupByContainsTimestamp = queryGroupByNamesList.remove(AGG_START_TIMESTAMP_COL);

        boolean isQueryGroupBySameAsAggGroupBy =
                queryGroupByListCopy.isEmpty() ||
                        (queryGroupByListCopy.contains(timestampVariable) &&
                                queryGroupByNamesList.equals(groupByVariablesList));

        List<VariableExpressionExecutor> variableExpExecutorsForTableLookups = new ArrayList<>();

        Map<TimePeriod.Duration, CompiledSelection> withinTableCompiledSelection = new HashMap<>();
        if (isOptimisedTableLookup) {
            Selector selector = Selector.selector();

            List<Variable> groupByList = new ArrayList<>();
            if (!isQueryGroupBySameAsAggGroupBy) {
                if (queryGroupByContainsTimestamp) {
                    if (isProcessingOnExternalTime) {
                        groupByList.add(new Variable(AGG_EXTERNAL_TIMESTAMP_COL));
                    } else {
                        groupByList.add(new Variable(AGG_START_TIMESTAMP_COL));
                    }
                    //Remove timestamp to process the rest
                    queryGroupByListCopy.remove(timestampVariable);
                }
                for (Variable queryGroupBy : queryGroupByListCopy) {
                    String referenceId = queryGroupBy.getStreamId();
                    if (referenceId == null) {
                        if (tableAttributesNameList.contains(queryGroupBy.getAttributeName())) {
                            groupByList.add(queryGroupBy);
                        }
                    } else if (referenceId.equalsIgnoreCase(referenceName)) {
                        groupByList.add(queryGroupBy);
                    }
                }
                // If query group bys are based on joining stream
                if (groupByList.isEmpty()) {
                    isQueryGroupBySameAsAggGroupBy = true;
                }
            }

            groupByList.forEach((groupBy) -> groupBy.setStreamId(referenceName));

            selector.addGroupByList(groupByList);

            List<OutputAttribute> selectorList;
            if (!isQueryGroupBySameAsAggGroupBy) {
                selectorList = constructSelectorList(isProcessingOnExternalTime, isDistributed, isLatestEventColAdded,
                        baseAggregatorBeginIndex, groupByVariablesList.size(), finalBaseExpressionsList,
                        tableDefinition, groupByList);
            } else {
                selectorList = defaultSelectorList;
            }

            for (OutputAttribute outputAttribute : selectorList) {
                if (outputAttribute.getExpression() instanceof Variable) {
                    ((Variable) outputAttribute.getExpression()).setStreamId(referenceName);
                } else {
                    for (Expression parameter :
                            ((AttributeFunction) outputAttribute.getExpression()).getParameters()) {
                        ((Variable) parameter).setStreamId(referenceName);
                    }
                }
            }
            selector.addSelectionList(selectorList);

            try {
                aggregationTables.entrySet().forEach(
                        (durationTableEntry -> {
                            CompiledSelection compiledSelection = ((QueryableProcessor) durationTableEntry.getValue())
                                    .compileSelection(
                                            selector, tableDefinition.getAttributeList(), metaInfoHolderForTableLookups,
                                            variableExpExecutorsForTableLookups, tableMap, siddhiQueryContext
                                    );
                            withinTableCompiledSelection.put(durationTableEntry.getKey(), compiledSelection);
                        })
                );
            } catch (SiddhiAppCreationException | SiddhiAppValidationException | QueryableRecordTableException e) {
                if (LOG.isDebugEnabled()) {
                    LOG.debug("Aggregation Query optimization failed for aggregation: '" + aggregationName + "'. " +
                            "Creating table lookup query in normal mode. Reason for failure: " + e.getMessage(), e);
                }
                isOptimisedTableLookup = false;
            }

        }

        for (Map.Entry<TimePeriod.Duration, Table> entry : aggregationTables.entrySet()) {
            CompiledCondition withinTableCompileCondition = entry.getValue().compileCondition(withinExpressionTable,
                    metaInfoHolderForTableLookups, variableExpExecutorsForTableLookups, tableMap,
                    siddhiQueryContext);
            withinTableCompiledConditions.put(entry.getKey(), withinTableCompileCondition);
        }

        // Create compile condition for in-memory data.
        // This compile condition is used to check whether the running aggregates (in-memory data)
        // are within given duration
        withinInMemoryCompileCondition = OperatorParser.constructOperator(new ComplexEventChunk<>(),
                withinExpression, metaInfoHolderForTableLookups, variableExpExecutorsForTableLookups, tableMap,
                siddhiQueryContext);

        // Create compile condition for in-memory data, in case of distributed
        // Look at the lower level granularities
        Map<TimePeriod.Duration, CompiledCondition> withinTableLowerGranularityCompileCondition = new HashMap<>();
        Expression lowerGranularity;
        if (isDistributed) {
            for (int i = 0; i < lowerGranularitySize; i++) {
                if (isProcessingOnExternalTime) {
                    lowerGranularity = Expression.and(
                            Expression.compare(
                                    Expression.variable("AGG_TIMESTAMP"),
                                    Compare.Operator.GREATER_THAN_EQUAL,
                                    Expression.variable(lowerGranularityAttributes.get(i))),
                            withinExpressionTable
                    );
                } else {
                    if (shouldApplyReducedCondition) {
                        lowerGranularity = Expression.and(
                                Expression.compare(
                                        Expression.variable("AGG_TIMESTAMP"),
                                        Compare.Operator.GREATER_THAN_EQUAL,
                                        Expression.variable(lowerGranularityAttributes.get(i))),
                                reducedExpression
                        );
                    } else {
                        lowerGranularity =
                                Expression.compare(
                                        Expression.variable("AGG_TIMESTAMP"),
                                        Compare.Operator.GREATER_THAN_EQUAL,
                                        Expression.variable(lowerGranularityAttributes.get(i)));
                    }
                }
                TimePeriod.Duration duration = this.activeIncrementalDurations.get(i);
                String tableName = aggregationName + "_" + duration.toString();
                CompiledCondition compiledCondition = tableMap.get(tableName).compileCondition(lowerGranularity,
                        metaInfoHolderForTableLookups, variableExpExecutorsForTableLookups, tableMap,
                        siddhiQueryContext);
                withinTableLowerGranularityCompileCondition.put(duration, compiledCondition);
            }
        }

        QueryParserHelper.reduceMetaComplexEvent(metaInfoHolderForTableLookups.getMetaStateEvent());

        // On compile condition.
        // After finding all the aggregates belonging to within duration, the final on condition (given as
        // "on stream1.name == aggregator.nickName ..." in the join query) must be executed on that data.
        // This condition is used for that purpose.
        onCompiledCondition = OperatorParser.constructOperator(new ComplexEventChunk<>(), expression,
                matchingMetaInfoHolder, variableExpressionExecutors, tableMap, siddhiQueryContext);

        return new IncrementalAggregateCompileCondition(isOnDemandQuery, aggregationName, isProcessingOnExternalTime,
                isDistributed, activeIncrementalDurations, aggregationTables, outputExpressionExecutors,
                isOptimisedTableLookup, withinTableCompiledSelection, withinTableCompiledConditions,
                withinInMemoryCompileCondition, withinTableLowerGranularityCompileCondition, onCompiledCondition,
                additionalAttributes, perExpressionExecutor, startTimeEndTimeExpressionExecutor,
                timestampFilterExecutors, aggregateMetaSteamEvent, matchingMetaInfoHolder,
                metaInfoHolderForTableLookups, variableExpExecutorsForTableLookups);

    }
Aggregation runtime managing aggregation operations for aggregation definition.
modules/siddhi-core/src/main/java/io/siddhi/core/aggregation/AggregationRuntime.java