private static MetaStreamEvent alterMetaStreamEvent(boolean isOnDemandQuery,
                                                        MetaStreamEvent originalMetaStreamEvent,
                                                        List<Attribute> additionalAttributes) {

        StreamDefinition alteredStreamDef = new StreamDefinition();
        String inputReferenceId = originalMetaStreamEvent.getInputReferenceId();

        if (!isOnDemandQuery) {
            for (Attribute attribute : originalMetaStreamEvent.getLastInputDefinition().getAttributeList()) {
                alteredStreamDef.attribute(attribute.getName(), attribute.getType());
            }
            if (inputReferenceId == null) {
                alteredStreamDef.setId(originalMetaStreamEvent.getLastInputDefinition().getId());
            }
        } else {
            // If it is on-demand query, no original join stream
            alteredStreamDef.setId("OnDemandQueryStream");
        }

        additionalAttributes.forEach(attribute -> alteredStreamDef.attribute(attribute.getName(), attribute.getType()));

        initMetaStreamEvent(originalMetaStreamEvent, alteredStreamDef, inputReferenceId);
        return originalMetaStreamEvent;
    }
Aggregation runtime managing aggregation operations for aggregation definition.
modules/siddhi-core/src/main/java/io/siddhi/core/aggregation/AggregationRuntime.java