public void init(AggregationDefinition aggregationDefinition, StreamEventFactory streamEventFactory,
                     Map<TimePeriod.Duration, Table> aggregationTables, Boolean isProcessingOnExternalTime,
                     SiddhiQueryContext siddhiQueryContext, List<TimePeriod.Duration> activeIncrementalDurations,
                     String timeZone, Map<String, Window> windowMap, Map<String, AggregationRuntime>
                             aggregationMap) {
        this.siddhiQueryContext = siddhiQueryContext;
        this.aggregationDefinition = aggregationDefinition;
        List<Annotation> annotations = aggregationDefinition.getAnnotations();
        this.streamEventFactory = streamEventFactory;
        this.aggregationTables = aggregationTables;
        this.activeIncrementalDurations = activeIncrementalDurations;
        this.windowMap = windowMap;
        this.aggregationMap = aggregationMap;
        if (isProcessingOnExternalTime) {
            purgingTimestampField = AGG_EXTERNAL_TIMESTAMP_COL;
        } else {
            purgingTimestampField = AGG_START_TIMESTAMP_COL;
        }
        aggregatedTimestampAttribute = new Attribute(purgingTimestampField, Attribute.Type.LONG);

        VariableExpressionExecutor variableExpressionExecutor = new VariableExpressionExecutor(
                aggregatedTimestampAttribute, 0, 1);
        variableExpressionExecutorList.add(variableExpressionExecutor);
        for (Map.Entry<TimePeriod.Duration, Table> entry : aggregationTables.entrySet()) {
            this.tableMap.put(entry.getValue().getTableDefinition().getId(), entry.getValue());
            switch (entry.getKey()) {
                case SECONDS:
                    retentionPeriods.put(entry.getKey(), Expression.Time.sec(120).value());
                    minimumDurationMap.put(entry.getKey(), Expression.Time.sec(120).value());
                    break;
                case MINUTES:
                    retentionPeriods.put(entry.getKey(), Expression.Time.hour(24).value());
                    minimumDurationMap.put(entry.getKey(), Expression.Time.minute(120).value());
                    break;
                case HOURS:
                    retentionPeriods.put(entry.getKey(), Expression.Time.day(30).value());
                    minimumDurationMap.put(entry.getKey(), Expression.Time.hour(25).value());
                    break;
                case DAYS:
                    retentionPeriods.put(entry.getKey(), Expression.Time.year(1).value());
                    minimumDurationMap.put(entry.getKey(), Expression.Time.day(32).value());
                    break;
                case MONTHS:
                    retentionPeriods.put(entry.getKey(), RETAIN_ALL);
                    minimumDurationMap.put(entry.getKey(), Expression.Time.month(13).value());
                    break;
                case YEARS:
                    retentionPeriods.put(entry.getKey(), RETAIN_ALL);
                    minimumDurationMap.put(entry.getKey(), 0L);
            }
        }
        this.timeZone = timeZone;
        Map<String, Annotation> annotationTypes = new HashMap<>();
        for (Annotation annotation : annotations) {
            annotationTypes.put(annotation.getName().toLowerCase(), annotation);
        }
        Annotation purge = annotationTypes.get(SiddhiConstants.NAMESPACE_PURGE);
        if (purge != null) {
            if (purge.getElement(SiddhiConstants.ANNOTATION_ELEMENT_ENABLE) != null) {
                String purgeEnable = purge.getElement(SiddhiConstants.ANNOTATION_ELEMENT_ENABLE);
                if (!("true".equalsIgnoreCase(purgeEnable) || "false".equalsIgnoreCase(purgeEnable))) {
                    throw new SiddhiAppCreationException("Invalid value for enable: " + purgeEnable + "." +
                            " Please use true or false");
                } else {
                    purgingEnabled = Boolean.parseBoolean(purgeEnable);
                }
            }
            if (purgingEnabled) {
                // If interval is defined, default value of 15 min will be replaced by user input value
                if (purge.getElement(SiddhiConstants.ANNOTATION_ELEMENT_INTERVAL) != null) {
                    String interval = purge.getElement(SiddhiConstants.ANNOTATION_ELEMENT_INTERVAL);
                    purgeExecutionInterval = timeToLong(interval);
                }
                List<Annotation> retentions = purge.getAnnotations(SiddhiConstants.NAMESPACE_RETENTION_PERIOD);
                if (retentions != null && !retentions.isEmpty()) {
                    Annotation retention = retentions.get(0);
                    List<Element> elements = retention.getElements();
                    for (Element element : elements) {
                        TimePeriod.Duration duration = normalizeDuration(element.getKey());
                        if (!activeIncrementalDurations.contains(duration)) {
                            throw new SiddhiAppCreationException(duration + " granularity cannot be purged since " +
                                    "aggregation has not performed in " + duration + " granularity");
                        }
                        if (element.getValue().equalsIgnoreCase(RETAIN_ALL_VALUES)) {
                            retentionPeriods.put(duration, RETAIN_ALL);
                        } else {
                            if (timeToLong(element.getValue()) >= minimumDurationMap.get(duration)) {
                                retentionPeriods.put(duration, timeToLong(element.getValue()));
                            } else {
                                throw new SiddhiAppCreationException(duration + " granularity cannot be purge" +
                                        " with a retention of '" + element.getValue() + "', minimum retention" +
                                        " should be greater  than " + TimeUnit.MILLISECONDS.toMinutes
                                        (minimumDurationMap.get(duration)) + " minutes");
                            }
                        }
                    }
                }
            }
        }
        compiledConditionsHolder = createCompileConditions(aggregationTables, tableMap);
    }
This class implements the logic which is needed to purge data which are related to incremental
modules/siddhi-core/src/main/java/io/siddhi/core/aggregation/IncrementalDataPurger.java