From 29287d13754adb0257a09c6d4a88ec6f52aaafbc Mon Sep 17 00:00:00 2001 From: Martijn van Groningen Date: Mon, 21 Nov 2022 16:26:25 +0100 Subject: [PATCH 1/4] Add a time series aggregation implementation to make use of the fact that docids are emitted in tsid and parent bucket ordinal. This is true when the parent aggregation is data histogram (which is typical), due to the fact that TimeSeriesIndexSearcher emits docs in tsid and timestamp order. Relates to #74660 --- .../TimeSeriesAggregationBuilder.java | 4 +- .../TimeSeriesAggregationFactory.java | 12 +- .../TimeSeriesInOrderAggregator.java | 153 ++++++++++++++++++ 3 files changed, 165 insertions(+), 4 deletions(-) create mode 100644 modules/aggregations/src/main/java/org/elasticsearch/aggregations/bucket/timeseries/TimeSeriesInOrderAggregator.java diff --git a/modules/aggregations/src/main/java/org/elasticsearch/aggregations/bucket/timeseries/TimeSeriesAggregationBuilder.java b/modules/aggregations/src/main/java/org/elasticsearch/aggregations/bucket/timeseries/TimeSeriesAggregationBuilder.java index 5815b0e499672..5032a59777ff7 100644 --- a/modules/aggregations/src/main/java/org/elasticsearch/aggregations/bucket/timeseries/TimeSeriesAggregationBuilder.java +++ b/modules/aggregations/src/main/java/org/elasticsearch/aggregations/bucket/timeseries/TimeSeriesAggregationBuilder.java @@ -15,6 +15,7 @@ import org.elasticsearch.search.aggregations.AggregationBuilder; import org.elasticsearch.search.aggregations.AggregatorFactories; import org.elasticsearch.search.aggregations.AggregatorFactory; +import org.elasticsearch.search.aggregations.bucket.histogram.DateHistogramAggregatorFactory; import org.elasticsearch.search.aggregations.support.AggregationContext; import org.elasticsearch.xcontent.InstantiatingObjectParser; import org.elasticsearch.xcontent.ParseField; @@ -79,7 +80,8 @@ protected AggregatorFactory doBuild( AggregatorFactory parent, AggregatorFactories.Builder subFactoriesBuilder ) throws IOException { - return new TimeSeriesAggregationFactory(name, keyed, context, parent, subFactoriesBuilder, metadata); + boolean expectTsidBucketInOrder = parent instanceof DateHistogramAggregatorFactory; + return new TimeSeriesAggregationFactory(name, keyed, context, parent, subFactoriesBuilder, metadata, expectTsidBucketInOrder); } @Override diff --git a/modules/aggregations/src/main/java/org/elasticsearch/aggregations/bucket/timeseries/TimeSeriesAggregationFactory.java b/modules/aggregations/src/main/java/org/elasticsearch/aggregations/bucket/timeseries/TimeSeriesAggregationFactory.java index 881d23cd8c4fb..27b752f310821 100644 --- a/modules/aggregations/src/main/java/org/elasticsearch/aggregations/bucket/timeseries/TimeSeriesAggregationFactory.java +++ b/modules/aggregations/src/main/java/org/elasticsearch/aggregations/bucket/timeseries/TimeSeriesAggregationFactory.java @@ -20,6 +20,7 @@ public class TimeSeriesAggregationFactory extends AggregatorFactory { private final boolean keyed; + private final boolean expectTsidBucketInOrder; public TimeSeriesAggregationFactory( String name, @@ -27,15 +28,20 @@ public TimeSeriesAggregationFactory( AggregationContext context, AggregatorFactory parent, AggregatorFactories.Builder subFactoriesBuilder, - Map metadata - ) throws IOException { + Map metadata, + boolean expectTsidBucketInOrder) throws IOException { super(name, context, parent, subFactoriesBuilder, metadata); this.keyed = keyed; + this.expectTsidBucketInOrder = expectTsidBucketInOrder; } @Override protected Aggregator createInternal(Aggregator parent, CardinalityUpperBound cardinality, Map metadata) throws IOException { - return new TimeSeriesAggregator(name, factories, keyed, context, parent, cardinality, metadata); + if (expectTsidBucketInOrder) { + return new TimeSeriesInOrderAggregator(name, factories, keyed, context, parent, cardinality, metadata); + } else { + return new TimeSeriesAggregator(name, factories, keyed, context, parent, cardinality, metadata); + } } } diff --git a/modules/aggregations/src/main/java/org/elasticsearch/aggregations/bucket/timeseries/TimeSeriesInOrderAggregator.java b/modules/aggregations/src/main/java/org/elasticsearch/aggregations/bucket/timeseries/TimeSeriesInOrderAggregator.java new file mode 100644 index 0000000000000..5fe3847ec8482 --- /dev/null +++ b/modules/aggregations/src/main/java/org/elasticsearch/aggregations/bucket/timeseries/TimeSeriesInOrderAggregator.java @@ -0,0 +1,153 @@ +/* + * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one + * or more contributor license agreements. Licensed under the Elastic License + * 2.0 and the Server Side Public License, v 1; you may not use this file except + * in compliance with, at your election, the Elastic License 2.0 or the Server + * Side Public License, v 1. + */ + +package org.elasticsearch.aggregations.bucket.timeseries; + +import org.apache.lucene.util.BytesRef; +import org.elasticsearch.common.util.BytesRefArray; +import org.elasticsearch.common.util.LongObjectPagedHashMap; +import org.elasticsearch.core.Releasables; +import org.elasticsearch.search.aggregations.AggregationExecutionContext; +import org.elasticsearch.search.aggregations.Aggregator; +import org.elasticsearch.search.aggregations.AggregatorFactories; +import org.elasticsearch.search.aggregations.CardinalityUpperBound; +import org.elasticsearch.search.aggregations.InternalAggregation; +import org.elasticsearch.search.aggregations.InternalAggregations; +import org.elasticsearch.search.aggregations.LeafBucketCollector; +import org.elasticsearch.search.aggregations.LeafBucketCollectorBase; +import org.elasticsearch.search.aggregations.bucket.BucketsAggregator; +import org.elasticsearch.search.aggregations.support.AggregationContext; + +import java.io.IOException; +import java.util.ArrayList; +import java.util.List; +import java.util.Map; + +public class TimeSeriesInOrderAggregator extends BucketsAggregator { + + // reuse tsids between owning bucket ordinals: + private final BytesRefArray collectedTsids; + private final LongObjectPagedHashMap> results; + private final boolean keyed; + + public TimeSeriesInOrderAggregator( + String name, + AggregatorFactories factories, + boolean keyed, + AggregationContext context, + Aggregator parent, + CardinalityUpperBound cardinality, + Map metadata + ) throws IOException { + super(name, factories, context, parent, cardinality, metadata); + this.keyed = keyed; + this.results = new LongObjectPagedHashMap<>(1, context.bigArrays()); + this.collectedTsids = new BytesRefArray(1, context.bigArrays()); + } + + @Override + public InternalAggregation[] buildAggregations(long[] owningBucketOrds) throws IOException { + InternalAggregation[] result = new InternalAggregation[owningBucketOrds.length]; + for (int ordIdx = 0; ordIdx < owningBucketOrds.length; ordIdx++) { + long owningOrdinal = owningBucketOrds[ordIdx]; + List internalBuckets = results.get(owningOrdinal); + if (internalBuckets != null) { + BytesRef spare = new BytesRef(); + List buckets = new ArrayList<>(internalBuckets.size()); + for (InternalBucket internalBucket : internalBuckets) { + BytesRef key = collectedTsids.get(internalBucket.tsidOffset, spare); + InternalAggregations internalAggregations = buildSubAggsForBuckets(new long[] { internalBucket.bucketOrd })[0]; + buckets.add( + new InternalTimeSeries.InternalBucket( + BytesRef.deepCopyOf(key), + internalBucket.docCount, + internalAggregations, + keyed + ) + ); + } + result[ordIdx] = new InternalTimeSeries(name, buckets, keyed, metadata()); + } else { + result[ordIdx] = buildEmptyAggregation(); + } + } + return result; + } + + @Override + public InternalAggregation buildEmptyAggregation() { + return new InternalTimeSeries(name, new ArrayList<>(), false, metadata()); + } + + private BytesRef currentTsid; + private int currentTsidOrd = -1; + private long currentParentBucket = -1; + private long docCount; + // TODO use 0L as bucket ordinal and clear sub aggregations after bucket/parent bucket ordinal combination changes + // Ideally use a constant ordinal (0) here and tsid or parent bucket change reset sub and + // reuse the same ordinal. This is possible because a tsid / parent bucket ordinal are unique and + // don't reappear when either one changes. + private long bucketOrdinalGenerator; + + @Override + protected LeafBucketCollector getLeafCollector(AggregationExecutionContext aggCtx, LeafBucketCollector sub) { + return new LeafBucketCollectorBase(sub, null) { + + @Override + public void collect(int doc, long bucket) throws IOException { + if (currentTsidOrd == aggCtx.getTsidOrd() && currentParentBucket == bucket) { + docCount++; + sub.collect(doc, bucketOrdinalGenerator); + return; + } + if (currentTsid != null) { + completeBucket(); + bucketOrdinalGenerator++; + } + if (currentTsidOrd != aggCtx.getTsidOrd()) { + currentTsidOrd = aggCtx.getTsidOrd(); + currentTsid = aggCtx.getTsid(); + + collectedTsids.append(currentTsid); + } + if (currentParentBucket != bucket) { + currentParentBucket = bucket; + } + + docCount = 1; + sub.collect(doc, bucketOrdinalGenerator); + } + }; + } + + @Override + protected void doPostCollection() { + if (currentTsid != null) { + completeBucket(); + } + } + + private void completeBucket() { + InternalBucket bucket = new InternalBucket(collectedTsids.size() - 1, bucketOrdinalGenerator, docCount); + // TODO: instead of collecting all buckets, perform pipeline aggregations here: + // (Then we don't need to keep all these buckets in memory) + List result = results.get(currentParentBucket); + if (result == null) { + result = new ArrayList<>(); + results.put(currentParentBucket, result); + } + result.add(bucket); + } + + @Override + protected void doClose() { + Releasables.close(results, collectedTsids); + } + + record InternalBucket(long tsidOffset, long bucketOrd, long docCount) {} +} From 8c7d39916c62074275e68939c240b31e3248e954 Mon Sep 17 00:00:00 2001 From: Martijn van Groningen Date: Mon, 21 Nov 2022 16:46:19 +0100 Subject: [PATCH 2/4] spotless --- .../bucket/timeseries/TimeSeriesAggregationFactory.java | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/modules/aggregations/src/main/java/org/elasticsearch/aggregations/bucket/timeseries/TimeSeriesAggregationFactory.java b/modules/aggregations/src/main/java/org/elasticsearch/aggregations/bucket/timeseries/TimeSeriesAggregationFactory.java index 27b752f310821..638245e5bf092 100644 --- a/modules/aggregations/src/main/java/org/elasticsearch/aggregations/bucket/timeseries/TimeSeriesAggregationFactory.java +++ b/modules/aggregations/src/main/java/org/elasticsearch/aggregations/bucket/timeseries/TimeSeriesAggregationFactory.java @@ -29,7 +29,8 @@ public TimeSeriesAggregationFactory( AggregatorFactory parent, AggregatorFactories.Builder subFactoriesBuilder, Map metadata, - boolean expectTsidBucketInOrder) throws IOException { + boolean expectTsidBucketInOrder + ) throws IOException { super(name, context, parent, subFactoriesBuilder, metadata); this.keyed = keyed; this.expectTsidBucketInOrder = expectTsidBucketInOrder; From d7a7e3eca2b22bcbebfe8cafe5a6f8be5bbcc3cf Mon Sep 17 00:00:00 2001 From: Martijn van Groningen Date: Mon, 21 Nov 2022 22:44:17 +0100 Subject: [PATCH 3/4] iter --- .../bucket/timeseries/TimeSeriesAggregationBuilder.java | 2 +- .../bucket/timeseries/TimeSeriesAggregatorTests.java | 5 +++++ 2 files changed, 6 insertions(+), 1 deletion(-) diff --git a/modules/aggregations/src/main/java/org/elasticsearch/aggregations/bucket/timeseries/TimeSeriesAggregationBuilder.java b/modules/aggregations/src/main/java/org/elasticsearch/aggregations/bucket/timeseries/TimeSeriesAggregationBuilder.java index 5032a59777ff7..36cabbf227c47 100644 --- a/modules/aggregations/src/main/java/org/elasticsearch/aggregations/bucket/timeseries/TimeSeriesAggregationBuilder.java +++ b/modules/aggregations/src/main/java/org/elasticsearch/aggregations/bucket/timeseries/TimeSeriesAggregationBuilder.java @@ -80,7 +80,7 @@ protected AggregatorFactory doBuild( AggregatorFactory parent, AggregatorFactories.Builder subFactoriesBuilder ) throws IOException { - boolean expectTsidBucketInOrder = parent instanceof DateHistogramAggregatorFactory; + boolean expectTsidBucketInOrder = parent == null || parent instanceof DateHistogramAggregatorFactory; return new TimeSeriesAggregationFactory(name, keyed, context, parent, subFactoriesBuilder, metadata, expectTsidBucketInOrder); } diff --git a/modules/aggregations/src/test/java/org/elasticsearch/aggregations/bucket/timeseries/TimeSeriesAggregatorTests.java b/modules/aggregations/src/test/java/org/elasticsearch/aggregations/bucket/timeseries/TimeSeriesAggregatorTests.java index b98e464daefb3..fc438e8546fef 100644 --- a/modules/aggregations/src/test/java/org/elasticsearch/aggregations/bucket/timeseries/TimeSeriesAggregatorTests.java +++ b/modules/aggregations/src/test/java/org/elasticsearch/aggregations/bucket/timeseries/TimeSeriesAggregatorTests.java @@ -112,6 +112,11 @@ private void timeSeriesTestCase( System.arraycopy(fieldTypes, 0, newFieldTypes, 2, fieldTypes.length); testCase(buildIndex, verify, new AggTestConfig(builder, newFieldTypes).withQuery(query)); + // TODO: ensure implementation class differently + debugTestCase(builder, query, buildIndex, (internalAggregation, aClass, stringMapMap) -> { + assertThat(aClass, equalTo(TimeSeriesInOrderAggregator.class)); + verify.accept((InternalTimeSeries) internalAggregation); + }, fieldTypes); } } From 1c5008f26ee3a8d07e1ce11408eb0a1e3ca28c84 Mon Sep 17 00:00:00 2001 From: Martijn van Groningen Date: Wed, 23 Nov 2022 14:50:21 +0100 Subject: [PATCH 4/4] iter --- .../bucket/timeseries/TimeSeriesAggregationFactory.java | 2 +- .../bucket/timeseries/TimeSeriesInOrderAggregator.java | 3 +-- 2 files changed, 2 insertions(+), 3 deletions(-) diff --git a/modules/aggregations/src/main/java/org/elasticsearch/aggregations/bucket/timeseries/TimeSeriesAggregationFactory.java b/modules/aggregations/src/main/java/org/elasticsearch/aggregations/bucket/timeseries/TimeSeriesAggregationFactory.java index 638245e5bf092..0acb7c222f9b7 100644 --- a/modules/aggregations/src/main/java/org/elasticsearch/aggregations/bucket/timeseries/TimeSeriesAggregationFactory.java +++ b/modules/aggregations/src/main/java/org/elasticsearch/aggregations/bucket/timeseries/TimeSeriesAggregationFactory.java @@ -40,7 +40,7 @@ public TimeSeriesAggregationFactory( protected Aggregator createInternal(Aggregator parent, CardinalityUpperBound cardinality, Map metadata) throws IOException { if (expectTsidBucketInOrder) { - return new TimeSeriesInOrderAggregator(name, factories, keyed, context, parent, cardinality, metadata); + return new TimeSeriesInOrderAggregator(name, factories, keyed, context, parent, metadata); } else { return new TimeSeriesAggregator(name, factories, keyed, context, parent, cardinality, metadata); } diff --git a/modules/aggregations/src/main/java/org/elasticsearch/aggregations/bucket/timeseries/TimeSeriesInOrderAggregator.java b/modules/aggregations/src/main/java/org/elasticsearch/aggregations/bucket/timeseries/TimeSeriesInOrderAggregator.java index 5fe3847ec8482..738489327e1e9 100644 --- a/modules/aggregations/src/main/java/org/elasticsearch/aggregations/bucket/timeseries/TimeSeriesInOrderAggregator.java +++ b/modules/aggregations/src/main/java/org/elasticsearch/aggregations/bucket/timeseries/TimeSeriesInOrderAggregator.java @@ -41,10 +41,9 @@ public TimeSeriesInOrderAggregator( boolean keyed, AggregationContext context, Aggregator parent, - CardinalityUpperBound cardinality, Map metadata ) throws IOException { - super(name, factories, context, parent, cardinality, metadata); + super(name, factories, context, parent, CardinalityUpperBound.MANY, metadata); this.keyed = keyed; this.results = new LongObjectPagedHashMap<>(1, context.bigArrays()); this.collectedTsids = new BytesRefArray(1, context.bigArrays());