protected Query getAggregationQuery(AggregationEventProcessorParameters parameters, long searchWithinMs, long executeEveryMs) {
        final Pivot.Builder pivotBuilder = Pivot.builder()
                .id(PIVOT_ID)
                .rollup(true);

        final ImmutableList<SeriesSpec> series = config.series()
                .stream()
                .map(s -> s.withId(metricName(s)))
                .collect(ImmutableList.toImmutableList());

        if (!series.isEmpty()) {
            pivotBuilder.series(series);
        }

        // Wrap every aggregation with date range buckets of the searchWithin time range.
        // If the aggregation is configured to be using a sliding window (searchWithin > executeEveryMs)
        // the time ranges will overlap.
        // This allows us to run aggregations over larger time ranges than the searchWithin time.
        // The results will be received in time buckets of the searchWithin time size.
        final DateRangeBucket dateRangeBucket = buildDateRangeBuckets(parameters.timerange(), searchWithinMs, executeEveryMs);
        final List<BucketSpec> groupBy = new ArrayList<>();

        // The first bucket must be the date range!
        groupBy.add(dateRangeBucket);

         if (!config.groupBy().isEmpty()) {
             final Values values = Values.builder().fields(config.groupBy())
                     .limit(Integer.MAX_VALUE)
                     .build();
             groupBy.add(values);

             // The pivot search type (as of Graylog 3.1.0) is using the "terms" aggregation under
             // the hood. The "terms" aggregation is meant to return the "top" terms and does not allow
             // and efficient retrieval and pagination over all terms.
             // Using Integer.MAX_VALUE as a limit can be very expensive with high cardinality grouping.
             // The ES documentation recommends to use the "Composite" aggregation instead.
             //
             // See the ES documentation for more details:
             //   https://www.elastic.co/guide/en/elasticsearch/reference/current/search-aggregations-bucket-terms-aggregation.html#search-aggregations-bucket-terms-aggregation-size
             //   https://www.elastic.co/guide/en/elasticsearch/reference/current/search-aggregations-bucket-composite-aggregation.html
             //
             // The "Composite" aggregation is only available since ES version 6.1, unfortunately.
             //
             // TODO: Either find a way to use the composite aggregation when the ES version in use is
             //       recent enough, and/or use a more conservative limit here and make it configurable
             //       by the user.

         }

        // We always have row groups because of the date range buckets
        pivotBuilder.rowGroups(groupBy);

        final Set<SearchType> searchTypes = Sets.newHashSet(pivotBuilder.build());
        searchTypes.addAll(additionalSearchTypes);

        final Query.Builder queryBuilder = Query.builder()
                .id(QUERY_ID)
                .searchTypes(searchTypes)
                .query(decorateQuery(config))
                .timerange(parameters.timerange());

        final Set<String> streams = getStreams(parameters);
        if (!streams.isEmpty()) {
            queryBuilder.filter(filteringForStreamIds(streams));
        }

        return queryBuilder.build();
    }
@Test
    public void testQueryParameterSubstitution() {
        final long WINDOW_LENGTH = 30000;
        final AbsoluteRange timerange = AbsoluteRange.create(DateTime.now(DateTimeZone.UTC).minusSeconds(3600), DateTime.now(DateTimeZone.UTC));
        var seriesCount = Count.builder().field("source").build();
        var seriesCard = Cardinality.builder().field("source").build();
        final AggregationEventProcessorConfig config = AggregationEventProcessorConfig.builder()
                .query("source:$secret$")
                .queryParameters(ImmutableSet.of(ValueParameter.builder().dataType("any").name("secret").build()))
                .streams(Collections.emptySet())
                .groupBy(Collections.emptyList())
                .series(ImmutableList.of(seriesCount, seriesCard))
                .conditions(null)
                .searchWithinMs(WINDOW_LENGTH)
                .executeEveryMs(WINDOW_LENGTH)
                .build();
        final AggregationEventProcessorParameters parameters = AggregationEventProcessorParameters.builder()
                .streams(Collections.emptySet())
                .timerange(timerange)
                .batchSize(500)
                .build();

        final PivotAggregationSearch pivotAggregationSearch = new PivotAggregationSearch(
                config,
                parameters,
                new AggregationSearch.User("test", DateTimeZone.UTC),
                eventDefinition,
                Collections.emptyList(),
                searchJobService,
                queryEngine,
                EventsConfigurationTestProvider.create(),
                moreSearch,
                new PermittedStreams(() -> Stream.of("00001")),
                notificationService,
                new QueryStringDecorators(Optional.of((queryString, parameterProvider, query) -> {
                    if (queryString.equals("source:$secret$") && parameterProvider.getParameter("secret").isPresent()) {
                        return PositionTrackingQuery.of("source:example.org");
                    } else {
                        throw new IllegalArgumentException("Unexpected query decoration request!");
                    }
                }))
        );
        final Query query = pivotAggregationSearch.getAggregationQuery(parameters, WINDOW_LENGTH, WINDOW_LENGTH);
        Assertions.assertThat(query.query().queryString()).isEqualTo("source:example.org");
    }