This is an automated email from the ASF dual-hosted git repository.
FrankChen021 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/druid.git
The following commit(s) were added to refs/heads/master by this push:
new e16bbf8bdc8 fix: implement makeAggregateCombiner for momentSketch
aggregators (#20413)
e16bbf8bdc8 is described below
commit e16bbf8bdc88d98219902f63d1e76b56d7d7690e
Author: kdelay <[email protected]>
AuthorDate: Sat Sep 26 13:12:55 2026 +0900
fix: implement makeAggregateCombiner for momentSketch aggregators (#20413)
Rollup ingestion that persists more than one intermediate index merges
them through RowCombiningTimeAndDimsIterator, which needs an
AggregateCombiner. MomentSketchAggregatorFactory did not provide one,
so the merge failed with UOE.
Fixes #13220
---
.../aggregator/MomentSketchAggregatorFactory.java | 47 ++++++++++++++++
.../MomentSketchAggregatorFactoryTest.java | 37 ++++++++++++
.../aggregator/MomentsSketchAggregatorTest.java | 65 ++++++++++++++++++++++
3 files changed, 149 insertions(+)
diff --git
a/extensions-contrib/momentsketch/src/main/java/org/apache/druid/query/aggregation/momentsketch/aggregator/MomentSketchAggregatorFactory.java
b/extensions-contrib/momentsketch/src/main/java/org/apache/druid/query/aggregation/momentsketch/aggregator/MomentSketchAggregatorFactory.java
index f873a023adc..b240510ede1 100644
---
a/extensions-contrib/momentsketch/src/main/java/org/apache/druid/query/aggregation/momentsketch/aggregator/MomentSketchAggregatorFactory.java
+++
b/extensions-contrib/momentsketch/src/main/java/org/apache/druid/query/aggregation/momentsketch/aggregator/MomentSketchAggregatorFactory.java
@@ -23,14 +23,17 @@ import com.fasterxml.jackson.annotation.JsonCreator;
import com.fasterxml.jackson.annotation.JsonProperty;
import org.apache.druid.java.util.common.ISE;
import org.apache.druid.java.util.common.StringUtils;
+import org.apache.druid.query.aggregation.AggregateCombiner;
import org.apache.druid.query.aggregation.Aggregator;
import org.apache.druid.query.aggregation.AggregatorFactory;
import
org.apache.druid.query.aggregation.AggregatorFactoryNotMergeableException;
import org.apache.druid.query.aggregation.AggregatorUtil;
import org.apache.druid.query.aggregation.BufferAggregator;
+import org.apache.druid.query.aggregation.ObjectAggregateCombiner;
import org.apache.druid.query.aggregation.momentsketch.MomentSketchWrapper;
import org.apache.druid.query.cache.CacheKeyBuilder;
import org.apache.druid.segment.ColumnSelectorFactory;
+import org.apache.druid.segment.ColumnValueSelector;
import org.apache.druid.segment.column.ColumnCapabilities;
import org.apache.druid.segment.column.ColumnType;
@@ -151,6 +154,50 @@ public class MomentSketchAggregatorFactory extends
AggregatorFactory
return union;
}
+ @Override
+ public AggregateCombiner<MomentSketchWrapper> makeAggregateCombiner()
+ {
+ return new ObjectAggregateCombiner<>()
+ {
+ @Nullable
+ private MomentSketchWrapper combined;
+
+ @Override
+ public void reset(final ColumnValueSelector selector)
+ {
+ combined = null;
+ fold(selector);
+ }
+
+ @Override
+ public void fold(final ColumnValueSelector selector)
+ {
+ final MomentSketchWrapper other = (MomentSketchWrapper)
selector.getObject();
+ if (other == null) {
+ return;
+ }
+ if (combined == null) {
+ combined = new MomentSketchWrapper(k);
+ combined.setCompressed(compress);
+ }
+ combined.merge(other);
+ }
+
+ @Nullable
+ @Override
+ public MomentSketchWrapper getObject()
+ {
+ return combined;
+ }
+
+ @Override
+ public Class<MomentSketchWrapper> classOfObject()
+ {
+ return MomentSketchWrapper.class;
+ }
+ };
+ }
+
@Override
public AggregatorFactory getCombiningFactory()
{
diff --git
a/extensions-contrib/momentsketch/src/test/java/org/apache/druid/query/aggregation/momentsketch/aggregator/MomentSketchAggregatorFactoryTest.java
b/extensions-contrib/momentsketch/src/test/java/org/apache/druid/query/aggregation/momentsketch/aggregator/MomentSketchAggregatorFactoryTest.java
index 55818bde47a..53da435261e 100644
---
a/extensions-contrib/momentsketch/src/test/java/org/apache/druid/query/aggregation/momentsketch/aggregator/MomentSketchAggregatorFactoryTest.java
+++
b/extensions-contrib/momentsketch/src/test/java/org/apache/druid/query/aggregation/momentsketch/aggregator/MomentSketchAggregatorFactoryTest.java
@@ -23,7 +23,10 @@ import com.fasterxml.jackson.databind.ObjectMapper;
import org.apache.druid.jackson.DefaultObjectMapper;
import org.apache.druid.java.util.common.granularity.Granularities;
import org.apache.druid.query.Druids;
+import org.apache.druid.query.aggregation.AggregateCombiner;
import org.apache.druid.query.aggregation.CountAggregatorFactory;
+import org.apache.druid.query.aggregation.TestObjectColumnSelector;
+import org.apache.druid.query.aggregation.momentsketch.MomentSketchWrapper;
import org.apache.druid.query.aggregation.post.FieldAccessPostAggregator;
import
org.apache.druid.query.aggregation.post.FinalizingFieldAccessPostAggregator;
import org.apache.druid.query.timeseries.TimeseriesQuery;
@@ -51,6 +54,40 @@ public class MomentSketchAggregatorFactoryTest
Assertions.assertEquals(factory, other);
}
+ @Test
+ public void testMakeAggregateCombiner()
+ {
+ final MomentSketchAggregatorFactory factory = new
MomentSketchAggregatorFactory("name", "fieldName", 10, true);
+
+ final MomentSketchWrapper first = new MomentSketchWrapper(10);
+ first.setCompressed(true);
+ first.add(1);
+ first.add(2);
+ final MomentSketchWrapper second = new MomentSketchWrapper(10);
+ second.setCompressed(true);
+ second.add(3);
+
+ final TestObjectColumnSelector<MomentSketchWrapper> selector =
+ new TestObjectColumnSelector<>(new MomentSketchWrapper[]{first,
second, null});
+ final AggregateCombiner<MomentSketchWrapper> combiner =
factory.makeAggregateCombiner();
+
+ combiner.reset(selector);
+ selector.increment();
+ combiner.fold(selector);
+ selector.increment();
+ combiner.fold(selector);
+
+ final MomentSketchWrapper combined = combiner.getObject();
+ Assertions.assertNotNull(combined);
+ Assertions.assertEquals(3.0, combined.getPowerSums()[0], 1e-10);
+ Assertions.assertEquals(1.0, combined.getMin(), 1e-10);
+ Assertions.assertEquals(3.0, combined.getMax(), 1e-10);
+ Assertions.assertTrue(combined.getCompressed());
+ // the inputs come from segment columns and must not be modified
+ Assertions.assertEquals(2.0, first.getPowerSums()[0], 1e-10);
+ Assertions.assertEquals(1.0, second.getPowerSums()[0], 1e-10);
+ }
+
@Test
public void testResultArraySignature()
{
diff --git
a/extensions-contrib/momentsketch/src/test/java/org/apache/druid/query/aggregation/momentsketch/aggregator/MomentsSketchAggregatorTest.java
b/extensions-contrib/momentsketch/src/test/java/org/apache/druid/query/aggregation/momentsketch/aggregator/MomentsSketchAggregatorTest.java
index a72939a3c02..147fa8a8b44 100644
---
a/extensions-contrib/momentsketch/src/test/java/org/apache/druid/query/aggregation/momentsketch/aggregator/MomentsSketchAggregatorTest.java
+++
b/extensions-contrib/momentsketch/src/test/java/org/apache/druid/query/aggregation/momentsketch/aggregator/MomentsSketchAggregatorTest.java
@@ -194,6 +194,71 @@ public class MomentsSketchAggregatorTest extends
InitializedNullHandlingTest
}
+ @Test
+ public void buildingSketchesAtIngestionTimeWithRollupAcrossPersists() throws
Exception
+ {
+ // The data has 9 distinct (timestamp, product) rows after rollup. A
maxRowCount of 3 forces several
+ // persisted indexes that share rows, so merging them has to combine
sketches with makeAggregateCombiner().
+ Sequence<ResultRow> seq = helper.createIndexAndRunQueryOnSegment(
+ new
File(this.getClass().getClassLoader().getResource("doubles_build_data.tsv").getFile()),
+ new InputRowSchema(
+ new TimestampSpec("timestamp", "yyyyMMddHH", null),
+ DimensionsSpec.builder()
+ .setDefaultSchemaDimensions(List.of("product"))
+ .setDimensionExclusions(List.of("sequenceNumber"))
+ .build(),
+ ColumnsFilter.all()
+ ),
+ DelimitedInputFormat.forColumns(
+ List.of("timestamp", "sequenceNumber", "product", "value",
"valueWithNulls")
+ ),
+ List.of(
+ new MomentSketchAggregatorFactory("sketch", "value", 10, true),
+ new MomentSketchAggregatorFactory("sketchWithNulls",
"valueWithNulls", 10, true)
+ ),
+ 0,
+ // minTimestamp
+ Granularities.NONE,
+ 3,
+ // maxRowCount
+ GroupByQuery.builder()
+ .setDataSource("test_datasource")
+ .setGranularity(Granularities.ALL)
+ .setDimensions(Collections.emptyList())
+ .setAggregatorSpecs(
+ new MomentSketchMergeAggregatorFactory("sketch", 10,
true),
+ new
MomentSketchMergeAggregatorFactory("sketchWithNulls", 10, true)
+ )
+ .setPostAggregatorSpecs(
+ new MomentSketchMinPostAggregator(
+ "min",
+ new FieldAccessPostAggregator("sketch", "sketch")
+ ),
+ new MomentSketchMaxPostAggregator(
+ "max",
+ new FieldAccessPostAggregator("sketch", "sketch")
+ )
+ )
+
.setInterval(Intervals.of("2016-01-01T00:00:00.000Z/2016-01-31T00:00:00.000Z"))
+ .build()
+ );
+ List<ResultRow> results = seq.toList();
+ Assertions.assertEquals(1, results.size());
+ ResultRow row = results.get(0);
+
+ MomentSketchWrapper sketchObject = (MomentSketchWrapper) row.get(0); //
"sketch"
+ Assertions.assertEquals(400.0, sketchObject.getPowerSums()[0], 1e-10);
+
+ MomentSketchWrapper sketchObjectWithNulls = (MomentSketchWrapper)
row.get(1); // "sketchWithNulls"
+ Assertions.assertEquals(377.0, sketchObjectWithNulls.getPowerSums()[0],
1e-10);
+
+ Double minValue = (Double) row.get(2); // "min"
+ Assertions.assertEquals(0.0011, minValue, 0.0001);
+
+ Double maxValue = (Double) row.get(3); // "max"
+ Assertions.assertEquals(0.9969, maxValue, 0.0001);
+ }
+
@Test
public void buildingSketchesAtQueryTime() throws Exception
{
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
