diff --git a/pinot-core/src/main/java/org/apache/pinot/core/operator/query/NonScanBasedAggregationOperator.java b/pinot-core/src/main/java/org/apache/pinot/core/operator/query/NonScanBasedAggregationOperator.java index 04f388cd46..152b333f58 100644 --- a/pinot-core/src/main/java/org/apache/pinot/core/operator/query/NonScanBasedAggregationOperator.java +++ b/pinot-core/src/main/java/org/apache/pinot/core/operator/query/NonScanBasedAggregationOperator.java @@ -144,6 +144,10 @@ protected AggregationResultsBlock getNextBlock() { result = getDistinctCountSmartHLLResult(Objects.requireNonNull(dataSource.getDictionary()), (DistinctCountSmartHLLAggregationFunction) aggregationFunction); break; + case SUM: + // dataSource.getDataSourceMetadata().getSumValue() is always present for sum aggregation function. + result = dataSource.getDataSourceMetadata().getSumValue().get(); + break; default: throw new IllegalStateException( "Non-scan based aggregation operator does not support function type: " + aggregationFunction.getType()); diff --git a/pinot-core/src/main/java/org/apache/pinot/core/plan/AggregationPlanNode.java b/pinot-core/src/main/java/org/apache/pinot/core/plan/AggregationPlanNode.java index 5295d39f89..7998b8db2d 100644 --- a/pinot-core/src/main/java/org/apache/pinot/core/plan/AggregationPlanNode.java +++ b/pinot-core/src/main/java/org/apache/pinot/core/plan/AggregationPlanNode.java @@ -37,6 +37,7 @@ import org.apache.pinot.segment.spi.IndexSegment; import org.apache.pinot.segment.spi.SegmentContext; import org.apache.pinot.segment.spi.datasource.DataSource; +import org.apache.pinot.segment.spi.datasource.DataSourceMetadata; import org.apache.pinot.segment.spi.index.reader.NullValueVectorReader; import static org.apache.pinot.segment.spi.AggregationFunctionType.COUNT; @@ -63,6 +64,7 @@ import static org.apache.pinot.segment.spi.AggregationFunctionType.MINMAXRANGEMV; import static org.apache.pinot.segment.spi.AggregationFunctionType.MINMV; import static org.apache.pinot.segment.spi.AggregationFunctionType.SEGMENTPARTITIONEDDISTINCTCOUNT; +import static org.apache.pinot.segment.spi.AggregationFunctionType.SUM; /** @@ -79,7 +81,7 @@ public class AggregationPlanNode implements PlanNode { // DISTINCTCOUNT excluded because consuming segment metadata contains unknown cardinality when there is no dictionary private static final EnumSet METADATA_BASED_FUNCTIONS = - EnumSet.of(COUNT, MIN, MINMV, MAX, MAXMV, MINMAXRANGE, MINMAXRANGEMV); + EnumSet.of(COUNT, MIN, MINMV, MAX, MAXMV, MINMAXRANGE, MINMAXRANGEMV, SUM); private final IndexSegment _indexSegment; private final SegmentContext _segmentContext; @@ -213,8 +215,12 @@ private boolean isFitForNonScanBasedPlan() { } } if (METADATA_BASED_FUNCTIONS.contains(aggregationFunction.getType())) { - if (dataSource.getDataSourceMetadata().getMaxValue() != null - && dataSource.getDataSourceMetadata().getMinValue() != null) { + DataSourceMetadata metadata = dataSource.getDataSourceMetadata(); + if (aggregationFunction.getType() == SUM) { + if (metadata.getSumValue().isPresent()) { + continue; + } + } else if (metadata.getMaxValue() != null && metadata.getMinValue() != null) { continue; } } diff --git a/pinot-core/src/test/java/org/apache/pinot/core/plan/maker/MetadataAndDictionaryAggregationPlanMakerTest.java b/pinot-core/src/test/java/org/apache/pinot/core/plan/maker/MetadataAndDictionaryAggregationPlanMakerTest.java index 3b103cdd66..d87aef98df 100644 --- a/pinot-core/src/test/java/org/apache/pinot/core/plan/maker/MetadataAndDictionaryAggregationPlanMakerTest.java +++ b/pinot-core/src/test/java/org/apache/pinot/core/plan/maker/MetadataAndDictionaryAggregationPlanMakerTest.java @@ -116,6 +116,7 @@ public void buildSegment() segmentGeneratorConfig.setTableName("testTable"); segmentGeneratorConfig.setSegmentName(SEGMENT_NAME); segmentGeneratorConfig.setOutDir(INDEX_DIR.getAbsolutePath()); + segmentGeneratorConfig.setEnableColumnSumValue(true); // Build the index segment. SegmentIndexCreationDriver driver = new SegmentIndexCreationDriverImpl(); @@ -158,15 +159,15 @@ public void deleteSegment() { @Test(dataProvider = "testPlanMakerDataProvider") public void testPlanMaker(String query, Class> operatorClass, - Class> upsertOperatorClass) { + Class> upsertOperatorClass, String name) { QueryContext queryContext = QueryContextConverterUtils.getQueryContext(query); Operator operator = PLAN_MAKER.makeSegmentPlanNode(new SegmentContext(_indexSegment), queryContext).run(); - assertTrue(operatorClass.isInstance(operator)); + assertTrue(operatorClass.isInstance(operator), name); SegmentContext segmentContext = new SegmentContext(_upsertIndexSegment); segmentContext.setQueryableDocIdsSnapshot(UpsertUtils.getQueryableDocIdsSnapshotFromSegment(_upsertIndexSegment)); Operator upsertOperator = PLAN_MAKER.makeSegmentPlanNode(segmentContext, queryContext).run(); - assertTrue(upsertOperatorClass.isInstance(upsertOperator)); + assertTrue(upsertOperatorClass.isInstance(upsertOperator), name); } @DataProvider(name = "testPlanMakerDataProvider") @@ -174,66 +175,73 @@ public Object[][] testPlanMakerDataProvider() { List entries = new ArrayList<>(); // Selection entries.add(new Object[]{ - "select * from testTable", SelectionOnlyOperator.class, SelectionOnlyOperator.class + "select * from testTable", SelectionOnlyOperator.class, SelectionOnlyOperator.class, "select-all" }); // Selection entries.add(new Object[]{ - "select column1,column5 from testTable", SelectionOnlyOperator.class, SelectionOnlyOperator.class + "select column1,column5 from testTable", SelectionOnlyOperator.class, SelectionOnlyOperator.class, + "select-columns" }); // Selection with filter entries.add(new Object[]{ - "select * from testTable where daysSinceEpoch > 100", SelectionOnlyOperator.class, SelectionOnlyOperator.class + "select * from testTable where daysSinceEpoch > 100", SelectionOnlyOperator.class, SelectionOnlyOperator.class, + "select-all-with-filter" }); // COUNT from metadata (via non-scan-based aggregation because it is an all-match) entries.add(new Object[]{ - "select count(*) from testTable", NonScanBasedAggregationOperator.class, FastFilteredCountOperator.class + "select count(*) from testTable", NonScanBasedAggregationOperator.class, FastFilteredCountOperator.class, + "count-all" }); // COUNT from metadata with a non-match-all filter, fast filtered count is to be used entries.add(new Object[]{ "select count(*) from testTable where column1 <= 10", FastFilteredCountOperator.class, - FastFilteredCountOperator.class + FastFilteredCountOperator.class, "count-all-with-non-match-all-filter" }); // COUNT from metadata with match all filter entries.add(new Object[]{ "select count(*) from testTable where column1 > 10", NonScanBasedAggregationOperator.class, - FastFilteredCountOperator.class + FastFilteredCountOperator.class, "count-all-with-match-all-filter" }); // MIN/MAX from dictionary entries.add(new Object[]{ "select max(daysSinceEpoch),min(daysSinceEpoch) from testTable", NonScanBasedAggregationOperator.class, - AggregationOperator.class + AggregationOperator.class, "min-max-from-dictionary" }); // MIN/MAX from dictionary with match all filter entries.add(new Object[]{ "select max(daysSinceEpoch),min(daysSinceEpoch) from testTable where column1 > 10", - NonScanBasedAggregationOperator.class, AggregationOperator.class + NonScanBasedAggregationOperator.class, AggregationOperator.class, + "min-max-from-dictionary-with-match-all-filter" }); // MINMAXRANGE from dictionary entries.add(new Object[]{ "select minmaxrange(daysSinceEpoch) from testTable", NonScanBasedAggregationOperator.class, - AggregationOperator.class + AggregationOperator.class, "minmaxrange-from-dictionary" }); // MINMAXRANGE from dictionary with match all filter entries.add(new Object[]{ "select minmaxrange(daysSinceEpoch) from testTable where column1 > 10", NonScanBasedAggregationOperator.class, - AggregationOperator.class + AggregationOperator.class, "minmaxrange-from-dictionary-with-match-all-filter" }); - // Aggregation + // SUM from metadata (default segment version is v3, which stores sum metadata for numeric SV columns) entries.add(new Object[]{ - "select sum(column1) from testTable", AggregationOperator.class, AggregationOperator.class + "select sum(column1) from testTable", NonScanBasedAggregationOperator.class, + AggregationOperator.class, "sum-from-metadata" }); // Aggregation group-by entries.add(new Object[]{ - "select sum(column1) from testTable group by daysSinceEpoch", GroupByOperator.class, GroupByOperator.class + "select sum(column1) from testTable group by daysSinceEpoch", GroupByOperator.class, + GroupByOperator.class, "sum-from-metadata-group-by" }); // COUNT from metadata, MIN from dictionary entries.add(new Object[]{ - "select count(*),min(column17) from testTable", NonScanBasedAggregationOperator.class, AggregationOperator.class + "select count(*),min(column17) from testTable", NonScanBasedAggregationOperator.class, + AggregationOperator.class, "count-from-metadata-min-from-dictionary" }); // Aggregation group-by entries.add(new Object[]{ "select count(*),min(daysSinceEpoch) from testTable group by daysSinceEpoch", GroupByOperator.class, - GroupByOperator.class + GroupByOperator.class, "count-from-metadata-min-from-dictionary-group-by" }); return entries.toArray(new Object[entries.size()][]); diff --git a/pinot-core/src/test/java/org/apache/pinot/core/plan/maker/NonScanBasedSumAggregationPlanMakerTest.java b/pinot-core/src/test/java/org/apache/pinot/core/plan/maker/NonScanBasedSumAggregationPlanMakerTest.java new file mode 100644 index 0000000000..dc254ffb27 --- /dev/null +++ b/pinot-core/src/test/java/org/apache/pinot/core/plan/maker/NonScanBasedSumAggregationPlanMakerTest.java @@ -0,0 +1,181 @@ +/** + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ +package org.apache.pinot.core.plan.maker; + +import java.io.File; +import java.net.URL; +import java.util.concurrent.TimeUnit; +import org.apache.commons.io.FileUtils; +import org.apache.pinot.common.metrics.ServerMetrics; +import org.apache.pinot.core.common.Operator; +import org.apache.pinot.core.operator.query.AggregationOperator; +import org.apache.pinot.core.operator.query.NonScanBasedAggregationOperator; +import org.apache.pinot.core.query.request.context.QueryContext; +import org.apache.pinot.core.query.request.context.utils.QueryContextConverterUtils; +import org.apache.pinot.segment.local.indexsegment.immutable.ImmutableSegmentLoader; +import org.apache.pinot.segment.local.segment.creator.impl.SegmentIndexCreationDriverImpl; +import org.apache.pinot.segment.spi.IndexSegment; +import org.apache.pinot.segment.spi.SegmentContext; +import org.apache.pinot.segment.spi.creator.SegmentGeneratorConfig; +import org.apache.pinot.segment.spi.creator.SegmentIndexCreationDriver; +import org.apache.pinot.segment.spi.creator.SegmentVersion; +import org.apache.pinot.spi.config.table.TableConfig; +import org.apache.pinot.spi.config.table.TableType; +import org.apache.pinot.spi.config.table.ingestion.IngestionConfig; +import org.apache.pinot.spi.data.FieldSpec; +import org.apache.pinot.spi.data.FieldSpec.DataType; +import org.apache.pinot.spi.data.Schema; +import org.apache.pinot.spi.data.TimeGranularitySpec; +import org.apache.pinot.spi.utils.ReadMode; +import org.apache.pinot.spi.utils.builder.TableConfigBuilder; +import org.testng.annotations.AfterClass; +import org.testng.annotations.AfterTest; +import org.testng.annotations.BeforeClass; +import org.testng.annotations.BeforeTest; +import org.testng.annotations.Test; + +import static org.mockito.Mockito.mock; +import static org.testng.Assert.assertNotNull; +import static org.testng.Assert.assertTrue; + + +/** + * Tests that SUM aggregation uses the non-scan-based path when segment metadata contains sum values (v3 segments), + * and falls back to scan-based aggregation when sum metadata is absent (non-v3 segments). + */ +public class NonScanBasedSumAggregationPlanMakerTest { + private static final String AVRO_DATA = "data" + File.separator + "test_data-sv.avro"; + private static final File INDEX_DIR = + new File(FileUtils.getTempDirectory(), "NonScanBasedSumAggregationPlanMakerTest"); + private static final String V3_SEGMENT_NAME = "testTable_v3_segment"; + private static final String V1_SEGMENT_NAME = "testTable_v1_segment"; + private static final InstancePlanMakerImplV2 PLAN_MAKER = new InstancePlanMakerImplV2(); + + private IndexSegment _v3Segment; + private IndexSegment _v1Segment; + + @BeforeTest + public void buildSegments() + throws Exception { + FileUtils.deleteQuietly(INDEX_DIR); + + URL resource = getClass().getClassLoader().getResource(AVRO_DATA); + assertNotNull(resource); + String filePath = resource.getFile(); + + Schema schema = new Schema.SchemaBuilder().setSchemaName("testTable") + .addMetric("column1", FieldSpec.DataType.INT) + .addMetric("column3", FieldSpec.DataType.INT) + .addSingleValueDimension("column5", FieldSpec.DataType.STRING) + .addSingleValueDimension("column6", FieldSpec.DataType.INT) + .addSingleValueDimension("column7", FieldSpec.DataType.INT) + .addSingleValueDimension("column9", FieldSpec.DataType.INT) + .addSingleValueDimension("column11", FieldSpec.DataType.STRING) + .addSingleValueDimension("column12", FieldSpec.DataType.STRING) + .addMetric("column17", FieldSpec.DataType.INT) + .addMetric("column18", FieldSpec.DataType.INT) + .addTime(new TimeGranularitySpec(DataType.INT, TimeUnit.DAYS, "daysSinceEpoch"), null).build(); + + IngestionConfig ingestionConfig = new IngestionConfig(); + ingestionConfig.setSegmentTimeValueCheck(false); + ingestionConfig.setRowTimeValueCheck(false); + TableConfig tableConfig = + new TableConfigBuilder(TableType.OFFLINE).setTableName("testTable").setTimeColumnName("daysSinceEpoch") + .setIngestionConfig(ingestionConfig).build(); + + // Build v3 segment (has sum metadata) + SegmentGeneratorConfig v3Config = new SegmentGeneratorConfig(tableConfig, schema); + v3Config.setInputFilePath(filePath); + v3Config.setTableName("testTable"); + v3Config.setSegmentName(V3_SEGMENT_NAME); + v3Config.setOutDir(INDEX_DIR.getAbsolutePath()); + v3Config.setSegmentVersion(SegmentVersion.v3); + v3Config.setEnableColumnSumValue(true); + SegmentIndexCreationDriver v3Driver = new SegmentIndexCreationDriverImpl(); + v3Driver.init(v3Config); + v3Driver.build(); + + // Build v1 segment (no sum metadata) + SegmentGeneratorConfig v1Config = new SegmentGeneratorConfig(tableConfig, schema); + v1Config.setInputFilePath(filePath); + v1Config.setTableName("testTable"); + v1Config.setSegmentName(V1_SEGMENT_NAME); + v1Config.setOutDir(INDEX_DIR.getAbsolutePath()); + v1Config.setSegmentVersion(SegmentVersion.v1); + SegmentIndexCreationDriver v1Driver = new SegmentIndexCreationDriverImpl(); + v1Driver.init(v1Config); + v1Driver.build(); + } + + @BeforeClass + public void loadSegments() + throws Exception { + ServerMetrics.register(mock(ServerMetrics.class)); + _v3Segment = ImmutableSegmentLoader.load(new File(INDEX_DIR, V3_SEGMENT_NAME), ReadMode.heap); + _v1Segment = ImmutableSegmentLoader.load(new File(INDEX_DIR, V1_SEGMENT_NAME), ReadMode.heap); + } + + @AfterClass + public void destroySegments() { + _v3Segment.destroy(); + _v1Segment.destroy(); + } + + @AfterTest + public void deleteSegments() { + FileUtils.deleteQuietly(INDEX_DIR); + } + + @Test + public void testSumUsesNonScanPathForV3Segment() { + String query = "select sum(column1) from testTable"; + QueryContext queryContext = QueryContextConverterUtils.getQueryContext(query); + Operator operator = PLAN_MAKER.makeSegmentPlanNode(new SegmentContext(_v3Segment), queryContext).run(); + assertTrue(operator instanceof NonScanBasedAggregationOperator, + "SUM on v3 segment should use NonScanBasedAggregationOperator"); + } + + @Test + public void testSumFallsBackToScanForV1Segment() { + String query = "select sum(column1) from testTable"; + QueryContext queryContext = QueryContextConverterUtils.getQueryContext(query); + Operator operator = PLAN_MAKER.makeSegmentPlanNode(new SegmentContext(_v1Segment), queryContext).run(); + assertTrue(operator instanceof AggregationOperator, + "SUM on v1 segment should fall back to AggregationOperator"); + } + + @Test + public void testSumWithFilterFallsBackToScan() { + // SUM with non-match-all filter cannot use non-scan path + String query = "select sum(column1) from testTable where column1 <= 10"; + QueryContext queryContext = QueryContextConverterUtils.getQueryContext(query); + Operator operator = PLAN_MAKER.makeSegmentPlanNode(new SegmentContext(_v3Segment), queryContext).run(); + assertTrue(operator instanceof AggregationOperator, + "SUM with filter should fall back to AggregationOperator"); + } + + @Test + public void testCountAndSumCombinedUsesNonScanForV3() { + String query = "select count(*), sum(column1) from testTable"; + QueryContext queryContext = QueryContextConverterUtils.getQueryContext(query); + Operator operator = PLAN_MAKER.makeSegmentPlanNode(new SegmentContext(_v3Segment), queryContext).run(); + assertTrue(operator instanceof NonScanBasedAggregationOperator, + "COUNT(*) + SUM on v3 segment should use NonScanBasedAggregationOperator"); + } +} diff --git a/pinot-core/src/test/java/org/apache/pinot/queries/BaseSingleValueQueriesTest.java b/pinot-core/src/test/java/org/apache/pinot/queries/BaseSingleValueQueriesTest.java index 92966880fd..a9b3ecf7fd 100644 --- a/pinot-core/src/test/java/org/apache/pinot/queries/BaseSingleValueQueriesTest.java +++ b/pinot-core/src/test/java/org/apache/pinot/queries/BaseSingleValueQueriesTest.java @@ -125,6 +125,7 @@ public void buildSegment() segmentGeneratorConfig.setInputFilePath(filePath); segmentGeneratorConfig.setTableName(RAW_TABLE_NAME); segmentGeneratorConfig.setOutDir(INDEX_DIR.getAbsolutePath()); + segmentGeneratorConfig.setEnableColumnSumValue(true); // Build the index segment. SegmentIndexCreationDriver driver = new SegmentIndexCreationDriverImpl(); diff --git a/pinot-core/src/test/java/org/apache/pinot/queries/CastQueriesTest.java b/pinot-core/src/test/java/org/apache/pinot/queries/CastQueriesTest.java index e654542289..aac9410165 100644 --- a/pinot-core/src/test/java/org/apache/pinot/queries/CastQueriesTest.java +++ b/pinot-core/src/test/java/org/apache/pinot/queries/CastQueriesTest.java @@ -26,6 +26,7 @@ import java.util.List; import java.util.Random; import org.apache.commons.io.FileUtils; +import org.apache.pinot.core.operator.BaseOperator; import org.apache.pinot.core.operator.blocks.results.AggregationResultsBlock; import org.apache.pinot.core.operator.blocks.results.SelectionResultsBlock; import org.apache.pinot.core.operator.query.AggregationOperator; @@ -141,8 +142,9 @@ public void testCastSum() { String query = "select cast(sum(" + X_COL + ") as int), " + "cast(sum(" + Y_COL + ") as int) " + "from " + RAW_TABLE_NAME; - AggregationOperator aggregationOperator = getOperator(query); - List aggregationResult = aggregationOperator.nextBlock().getResults(); + // SUM on v3 numeric SV columns uses NonScanBasedAggregationOperator (metadata-based) + BaseOperator operator = getOperator(query); + List aggregationResult = operator.nextBlock().getResults(); assertNotNull(aggregationResult); assertEquals(aggregationResult.size(), 2); assertEquals(((Number) aggregationResult.get(0)).intValue(), NUM_RECORDS / 2); diff --git a/pinot-core/src/test/java/org/apache/pinot/queries/InterSegmentAggregationSingleValueQueriesTest.java b/pinot-core/src/test/java/org/apache/pinot/queries/InterSegmentAggregationSingleValueQueriesTest.java index 6fcb909374..da073356b7 100644 --- a/pinot-core/src/test/java/org/apache/pinot/queries/InterSegmentAggregationSingleValueQueriesTest.java +++ b/pinot-core/src/test/java/org/apache/pinot/queries/InterSegmentAggregationSingleValueQueriesTest.java @@ -151,12 +151,14 @@ public void testMin() { public void testSum() { String query = "SELECT SUM(column1) AS v1, SUM(column3) AS v2 FROM testTable"; + // Without filter, query should be answered by NonScanBasedAggregationOperator (numEntriesScannedPostFilter = 0) + // using sum metadata stored in the v3 segment. BrokerResponseNative brokerResponse = getBrokerResponse(query); DataSchema expectedDataSchema = new DataSchema(new String[]{"v1", "v2"}, new ColumnDataType[]{ColumnDataType.DOUBLE, ColumnDataType.DOUBLE}); ResultTable expectedResultTable = new ResultTable(expectedDataSchema, Collections.singletonList(new Object[]{129268741751388.0, 129156636756600.0})); - QueriesTestUtils.testInterSegmentsResult(brokerResponse, 120000L, 0L, 240000L, 120000L, expectedResultTable); + QueriesTestUtils.testInterSegmentsResult(brokerResponse, 120000L, 0L, 0L, 120000L, expectedResultTable); brokerResponse = getBrokerResponse(query + FILTER); expectedResultTable = new ResultTable(expectedDataSchema, diff --git a/pinot-core/src/test/java/org/apache/pinot/queries/MetadataBasedSumAggregationQueriesTest.java b/pinot-core/src/test/java/org/apache/pinot/queries/MetadataBasedSumAggregationQueriesTest.java new file mode 100644 index 0000000000..0c42e33c53 --- /dev/null +++ b/pinot-core/src/test/java/org/apache/pinot/queries/MetadataBasedSumAggregationQueriesTest.java @@ -0,0 +1,353 @@ +/** + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ +package org.apache.pinot.queries; + +import java.io.File; +import java.io.IOException; +import java.util.Arrays; +import java.util.Collections; +import java.util.List; +import org.apache.avro.file.DataFileWriter; +import org.apache.avro.generic.GenericData; +import org.apache.avro.generic.GenericDatumWriter; +import org.apache.avro.generic.GenericRecord; +import org.apache.commons.io.FileUtils; +import org.apache.pinot.common.response.broker.BrokerResponseNative; +import org.apache.pinot.common.response.broker.ResultTable; +import org.apache.pinot.common.utils.DataSchema; +import org.apache.pinot.common.utils.DataSchema.ColumnDataType; +import org.apache.pinot.segment.local.indexsegment.immutable.ImmutableSegmentLoader; +import org.apache.pinot.segment.local.segment.creator.impl.SegmentIndexCreationDriverImpl; +import org.apache.pinot.segment.local.segment.index.loader.IndexLoadingConfig; +import org.apache.pinot.segment.spi.ImmutableSegment; +import org.apache.pinot.segment.spi.IndexSegment; +import org.apache.pinot.segment.spi.creator.SegmentGeneratorConfig; +import org.apache.pinot.segment.spi.creator.SegmentIndexCreationDriver; +import org.apache.pinot.segment.spi.creator.SegmentVersion; +import org.apache.pinot.spi.config.table.TableConfig; +import org.apache.pinot.spi.config.table.TableType; +import org.apache.pinot.spi.config.table.ingestion.IngestionConfig; +import org.apache.pinot.spi.data.FieldSpec; +import org.apache.pinot.spi.data.Schema; +import org.apache.pinot.spi.utils.builder.TableConfigBuilder; +import org.testng.annotations.AfterClass; +import org.testng.annotations.BeforeClass; +import org.testng.annotations.Test; + + +/** + * Tests that SUM aggregation uses the metadata-based (non-scan) path on v3 segments, verifying that zero entries are + * scanned and the result is correct. Builds a v1 segment side by side to confirm that the scan-based fallback works + * as expected, making it easy to compare execution statistics between the two. + * + *

Data: 4 records with known metric values: + *

    + *
  • intMetric: 10, 20, 30, 40 → SUM = 100
    • + *
  • longMetric: 100, 200, 300, 400 → SUM = 1000
    • + *
  • floatMetric: 1.5, 2.5, 3.5, 4.5 → SUM = 12.0
    • + *
  • doubleMetric: 10.1, 20.2, 30.3, 40.4 → SUM = 101.0
    • + *
+ * + *

The test extends {@link BaseQueriesTest} to use getBrokerResponse (inter-segment path with 2 identical segments, + * duplicated for OFFLINE + REALTIME instances), so totals are 4x: e.g. numTotalDocs = 16, SUM(intMetric) = 400. + */ +public class MetadataBasedSumAggregationQueriesTest extends BaseQueriesTest { + private static final File INDEX_DIR = + new File(FileUtils.getTempDirectory(), "MetadataBasedSumAggregationQueriesTest"); + private static final String RAW_TABLE_NAME = "testTable"; + private static final String V3_SEGMENT_NAME = "testTable_v3_20432_20433"; + private static final String V1_SEGMENT_NAME = "testTable_v1_20432_20433"; + + private static final int NUM_RECORDS = 4; + private static final int[] INT_VALUES = {10, 20, 30, 40}; + private static final long[] LONG_VALUES = {100L, 200L, 300L, 400L}; + private static final float[] FLOAT_VALUES = {1.5f, 2.5f, 3.5f, 4.5f}; + private static final double[] DOUBLE_VALUES = {10.1, 20.2, 30.3, 40.4}; + + private static final Schema SCHEMA = new Schema.SchemaBuilder() + .setSchemaName(RAW_TABLE_NAME) + .addMetric("intMetric", FieldSpec.DataType.INT) + .addMetric("longMetric", FieldSpec.DataType.LONG) + .addMetric("floatMetric", FieldSpec.DataType.FLOAT) + .addMetric("doubleMetric", FieldSpec.DataType.DOUBLE) + .addSingleValueDimension("dimColumn", FieldSpec.DataType.STRING) + .addDateTime("daysSinceEpoch", FieldSpec.DataType.INT, "EPOCH|DAYS", "1:DAYS") + .build(); + + private static final TableConfig TABLE_CONFIG; + + static { + IngestionConfig ingestionConfig = new IngestionConfig(); + ingestionConfig.setSegmentTimeValueCheck(false); + ingestionConfig.setRowTimeValueCheck(false); + TABLE_CONFIG = new TableConfigBuilder(TableType.OFFLINE) + .setTableName(RAW_TABLE_NAME) + .setTimeColumnName("daysSinceEpoch") + .setIngestionConfig(ingestionConfig) + .build(); + } + + // Active segment and segment list, swapped between v1 and v3 by helper methods + private IndexSegment _activeSegment; + private List _activeSegments; + + private ImmutableSegment _v3Segment; + private ImmutableSegment _v1Segment; + + @Override + protected String getFilter() { + return ""; + } + + @Override + protected IndexSegment getIndexSegment() { + return _activeSegment; + } + + @Override + protected List getIndexSegments() { + return _activeSegments; + } + + @BeforeClass + public void setUp() + throws Exception { + FileUtils.deleteQuietly(INDEX_DIR); + INDEX_DIR.mkdirs(); + + File avroFile = createAvroFile(); + + buildSegment(avroFile, V3_SEGMENT_NAME, SegmentVersion.v3); + buildSegment(avroFile, V1_SEGMENT_NAME, SegmentVersion.v1); + FileUtils.deleteQuietly(avroFile); + + IndexLoadingConfig loadingConfig = new IndexLoadingConfig(TABLE_CONFIG, SCHEMA); + _v3Segment = ImmutableSegmentLoader.load(new File(INDEX_DIR, V3_SEGMENT_NAME), loadingConfig); + _v1Segment = ImmutableSegmentLoader.load(new File(INDEX_DIR, V1_SEGMENT_NAME), loadingConfig); + } + + @AfterClass + public void tearDown() { + _v3Segment.destroy(); + _v1Segment.destroy(); + FileUtils.deleteQuietly(INDEX_DIR); + } + + private File createAvroFile() + throws IOException { + String avroSchemaStr = "{\n" + + " \"type\": \"record\",\n" + + " \"name\": \"TestRecord\",\n" + + " \"namespace\": \"com.example.pinot\",\n" + + " \"fields\": [\n" + + " { \"name\": \"intMetric\", \"type\": \"int\" },\n" + + " { \"name\": \"longMetric\", \"type\": \"long\" },\n" + + " { \"name\": \"floatMetric\", \"type\": \"float\" },\n" + + " { \"name\": \"doubleMetric\", \"type\": \"double\" },\n" + + " { \"name\": \"dimColumn\", \"type\": \"string\" },\n" + + " { \"name\": \"daysSinceEpoch\", \"type\": \"int\" }\n" + + " ]\n" + + "}"; + + org.apache.avro.Schema avroSchema = new org.apache.avro.Schema.Parser().parse(avroSchemaStr); + File avroFile = new File(INDEX_DIR, "test_data.avro"); + GenericDatumWriter datumWriter = new GenericDatumWriter<>(avroSchema); + try (DataFileWriter writer = new DataFileWriter<>(datumWriter)) { + writer.create(avroSchema, avroFile); + for (int i = 0; i < NUM_RECORDS; i++) { + GenericRecord record = new GenericData.Record(avroSchema); + record.put("intMetric", INT_VALUES[i]); + record.put("longMetric", LONG_VALUES[i]); + record.put("floatMetric", FLOAT_VALUES[i]); + record.put("doubleMetric", DOUBLE_VALUES[i]); + record.put("dimColumn", "dim" + i); + record.put("daysSinceEpoch", 20432 + (i / 2)); + writer.append(record); + } + } + return avroFile; + } + + private void buildSegment(File avroFile, String segmentName, SegmentVersion version) + throws Exception { + SegmentGeneratorConfig config = new SegmentGeneratorConfig(TABLE_CONFIG, SCHEMA); + config.setInputFilePath(avroFile.getAbsolutePath()); + config.setTableName(RAW_TABLE_NAME); + config.setSegmentName(segmentName); + config.setOutDir(INDEX_DIR.getAbsolutePath()); + config.setSegmentVersion(version); + config.setEnableColumnSumValue(true); + + SegmentIndexCreationDriver driver = new SegmentIndexCreationDriverImpl(); + driver.init(config); + driver.build(); + } + + private void useV3Segment() { + _activeSegment = _v3Segment; + _activeSegments = Arrays.asList(_v3Segment, _v3Segment); + } + + private void useV1Segment() { + _activeSegment = _v1Segment; + _activeSegments = Arrays.asList(_v1Segment, _v1Segment); + } + + // ========================= v3 segment tests (metadata-based, no scan) ========================= + // getBrokerResponse runs 2 segments on server, then duplicates the response for OFFLINE + REALTIME instances. + // With 4 records per segment: numTotalDocs = 4 * 2 segments * 2 instances = 16. + // NonScanBasedAggregationOperator reports: numDocsScanned = numTotalDocs, entries scanned = 0. + + @Test + public void testV3SumIntSkipsScan() { + useV3Segment(); + String query = "SELECT SUM(intMetric) FROM testTable"; + BrokerResponseNative response = getBrokerResponse(query); + + DataSchema expectedSchema = + new DataSchema(new String[]{"sum(intMetric)"}, new ColumnDataType[]{ColumnDataType.DOUBLE}); + // SUM(intMetric) = (10+20+30+40) * 2 segments * 2 instances = 400.0 + QueriesTestUtils.testInterSegmentsResult(response, 16L, 0L, 0L, 16L, + new ResultTable(expectedSchema, Collections.singletonList(new Object[]{400.0}))); + } + + @Test + public void testV3SumLongSkipsScan() { + useV3Segment(); + String query = "SELECT SUM(longMetric) FROM testTable"; + BrokerResponseNative response = getBrokerResponse(query); + + DataSchema expectedSchema = + new DataSchema(new String[]{"sum(longMetric)"}, new ColumnDataType[]{ColumnDataType.DOUBLE}); + // SUM(longMetric) = (100+200+300+400) * 4 = 4000.0 + QueriesTestUtils.testInterSegmentsResult(response, 16L, 0L, 0L, 16L, + new ResultTable(expectedSchema, Collections.singletonList(new Object[]{4000.0}))); + } + + @Test + public void testV3SumFloatSkipsScan() { + useV3Segment(); + String query = "SELECT SUM(floatMetric) FROM testTable"; + BrokerResponseNative response = getBrokerResponse(query); + + DataSchema expectedSchema = + new DataSchema(new String[]{"sum(floatMetric)"}, new ColumnDataType[]{ColumnDataType.DOUBLE}); + // SUM(floatMetric) = (1.5+2.5+3.5+4.5) * 4 = 48.0 + QueriesTestUtils.testInterSegmentsResult(response, 16L, 0L, 0L, 16L, + new ResultTable(expectedSchema, Collections.singletonList(new Object[]{48.0}))); + } + + @Test + public void testV3SumDoubleSkipsScan() { + useV3Segment(); + String query = "SELECT SUM(doubleMetric) FROM testTable"; + BrokerResponseNative response = getBrokerResponse(query); + + DataSchema expectedSchema = + new DataSchema(new String[]{"sum(doubleMetric)"}, new ColumnDataType[]{ColumnDataType.DOUBLE}); + // SUM(doubleMetric) = (10.1+20.2+30.3+40.4) * 4 = 404.0 + QueriesTestUtils.testInterSegmentsResult(response, 16L, 0L, 0L, 16L, + new ResultTable(expectedSchema, Collections.singletonList(new Object[]{404.0}))); + } + + @Test + public void testV3SumCombinedWithCountSkipsScan() { + useV3Segment(); + String query = "SELECT COUNT(*), SUM(intMetric), SUM(longMetric) FROM testTable"; + BrokerResponseNative response = getBrokerResponse(query); + + DataSchema expectedSchema = new DataSchema( + new String[]{"count(*)", "sum(intMetric)", "sum(longMetric)"}, + new ColumnDataType[]{ColumnDataType.LONG, ColumnDataType.DOUBLE, ColumnDataType.DOUBLE}); + QueriesTestUtils.testInterSegmentsResult(response, 16L, 0L, 0L, 16L, + new ResultTable(expectedSchema, Collections.singletonList(new Object[]{16L, 400.0, 4000.0}))); + } + + @Test + public void testV3SumCombinedWithMinMaxSkipsScan() { + useV3Segment(); + String query = "SELECT MIN(intMetric), MAX(intMetric), SUM(intMetric) FROM testTable"; + BrokerResponseNative response = getBrokerResponse(query); + + DataSchema expectedSchema = new DataSchema( + new String[]{"min(intMetric)", "max(intMetric)", "sum(intMetric)"}, + new ColumnDataType[]{ColumnDataType.DOUBLE, ColumnDataType.DOUBLE, ColumnDataType.DOUBLE}); + QueriesTestUtils.testInterSegmentsResult(response, 16L, 0L, 0L, 16L, + new ResultTable(expectedSchema, Collections.singletonList(new Object[]{10.0, 40.0, 400.0}))); + } + + // ========================= v1 segment tests (scan-based, no sum metadata) ========================= + // SUM must scan docs. numEntriesScannedPostFilter = docs * columns * 2 segments * 2 instances. + + @Test + public void testV1SumIntRequiresScan() { + useV1Segment(); + String query = "SELECT SUM(intMetric) FROM testTable"; + BrokerResponseNative response = getBrokerResponse(query); + + DataSchema expectedSchema = + new DataSchema(new String[]{"sum(intMetric)"}, new ColumnDataType[]{ColumnDataType.DOUBLE}); + // Same result, but numEntriesScannedPostFilter = 4 * 1 col * 2 segments * 2 instances = 16 + QueriesTestUtils.testInterSegmentsResult(response, 16L, 0L, 16L, 16L, + new ResultTable(expectedSchema, Collections.singletonList(new Object[]{400.0}))); + } + + @Test + public void testV1SumLongRequiresScan() { + useV1Segment(); + String query = "SELECT SUM(longMetric) FROM testTable"; + BrokerResponseNative response = getBrokerResponse(query); + + DataSchema expectedSchema = + new DataSchema(new String[]{"sum(longMetric)"}, new ColumnDataType[]{ColumnDataType.DOUBLE}); + QueriesTestUtils.testInterSegmentsResult(response, 16L, 0L, 16L, 16L, + new ResultTable(expectedSchema, Collections.singletonList(new Object[]{4000.0}))); + } + + @Test + public void testV1SumCombinedWithCountRequiresScan() { + useV1Segment(); + String query = "SELECT COUNT(*), SUM(intMetric), SUM(longMetric) FROM testTable"; + BrokerResponseNative response = getBrokerResponse(query); + + DataSchema expectedSchema = new DataSchema( + new String[]{"count(*)", "sum(intMetric)", "sum(longMetric)"}, + new ColumnDataType[]{ColumnDataType.LONG, ColumnDataType.DOUBLE, ColumnDataType.DOUBLE}); + // COUNT uses no-scan path, but SUM scans: 4 docs * 2 columns * 2 segments * 2 instances = 32 + QueriesTestUtils.testInterSegmentsResult(response, 16L, 0L, 32L, 16L, + new ResultTable(expectedSchema, Collections.singletonList(new Object[]{16L, 400.0, 4000.0}))); + } + + // ========================= v3 with filter (should still scan) ========================= + + @Test + public void testV3SumWithFilterRequiresScan() { + useV3Segment(); + String query = "SELECT SUM(intMetric) FROM testTable WHERE intMetric > 10"; + BrokerResponseNative response = getBrokerResponse(query); + + DataSchema expectedSchema = + new DataSchema(new String[]{"sum(intMetric)"}, new ColumnDataType[]{ColumnDataType.DOUBLE}); + // 3 docs match per segment (20, 30, 40), SUM = 90 * 2 segments * 2 instances = 360.0 + // numDocsScanned = 3 * 2 * 2 = 12, numEntriesScannedInFilter = 0 (dictionary-based filter), + // numEntriesScannedPostFilter = 3 * 1 col * 2 * 2 = 12 + QueriesTestUtils.testInterSegmentsResult(response, 12L, 0L, 12L, 16L, + new ResultTable(expectedSchema, Collections.singletonList(new Object[]{360.0}))); + } +} diff --git a/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/segment/creator/impl/SegmentColumnarIndexCreator.java b/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/segment/creator/impl/SegmentColumnarIndexCreator.java index a55ef41dfd..c264ac6bf3 100644 --- a/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/segment/creator/impl/SegmentColumnarIndexCreator.java +++ b/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/segment/creator/impl/SegmentColumnarIndexCreator.java @@ -28,6 +28,7 @@ import java.util.HashMap; import java.util.List; import java.util.Map; +import java.util.Optional; import java.util.TreeMap; import java.util.concurrent.TimeUnit; import java.util.stream.Collectors; @@ -48,6 +49,7 @@ import org.apache.pinot.segment.spi.creator.IndexCreationContext; import org.apache.pinot.segment.spi.creator.SegmentCreator; import org.apache.pinot.segment.spi.creator.SegmentGeneratorConfig; +import org.apache.pinot.segment.spi.creator.SegmentVersion; import org.apache.pinot.segment.spi.index.DictionaryIndexConfig; import org.apache.pinot.segment.spi.index.FieldIndexConfigs; import org.apache.pinot.segment.spi.index.ForwardIndexConfig; @@ -102,6 +104,7 @@ import static org.apache.pinot.segment.spi.V1Constants.MetadataKeys.Column.PARTITION_VALUES; import static org.apache.pinot.segment.spi.V1Constants.MetadataKeys.Column.SCHEMA_MAX_LENGTH; import static org.apache.pinot.segment.spi.V1Constants.MetadataKeys.Column.SCHEMA_MAX_LENGTH_EXCEED_STRATEGY; +import static org.apache.pinot.segment.spi.V1Constants.MetadataKeys.Column.SUM_VALUE; import static org.apache.pinot.segment.spi.V1Constants.MetadataKeys.Column.TOTAL_DOCS; import static org.apache.pinot.segment.spi.V1Constants.MetadataKeys.Column.TOTAL_NUMBER_OF_ENTRIES; import static org.apache.pinot.segment.spi.V1Constants.MetadataKeys.Column.getKeyFor; @@ -605,7 +608,8 @@ private void writeMetadata() SegmentDictionaryCreator dictionaryCreator = _dictionaryCreatorMap.get(column); int dictionaryElementSize = (dictionaryCreator != null) ? dictionaryCreator.getNumBytesPerEntry() : 0; addColumnMetadataInfo(properties, column, columnIndexCreationInfo, _totalDocs, _schema.getFieldSpecFor(column), - dictionaryCreator != null, dictionaryElementSize); + dictionaryCreator != null, dictionaryElementSize, _config.getSegmentVersion(), + _config.isEnableColumnSumValue()); } SegmentZKPropsConfig segmentZKPropsConfig = _config.getSegmentZKPropsConfig(); @@ -617,9 +621,16 @@ private void writeMetadata() CommonsConfigurationUtils.saveToFile(properties, metadataFile); } + public static void addColumnMetadataInfo(PropertiesConfiguration properties, String column, + ColumnIndexCreationInfo columnIndexCreationInfo, int totalDocs, FieldSpec fieldSpec, + boolean hasDictionary, int dictionaryElementSize) { + addColumnMetadataInfo(properties, column, columnIndexCreationInfo, totalDocs, fieldSpec, + hasDictionary, dictionaryElementSize, null, false); + } + public static void addColumnMetadataInfo(PropertiesConfiguration properties, String column, ColumnIndexCreationInfo columnIndexCreationInfo, int totalDocs, FieldSpec fieldSpec, boolean hasDictionary, - int dictionaryElementSize) { + int dictionaryElementSize, @Nullable SegmentVersion segmentVersion, boolean enableColumnSumValue) { int cardinality = columnIndexCreationInfo.getDistinctValueCount(); properties.setProperty(getKeyFor(column, CARDINALITY), String.valueOf(cardinality)); properties.setProperty(getKeyFor(column, TOTAL_DOCS), String.valueOf(totalDocs)); @@ -685,6 +696,12 @@ public static void addColumnMetadataInfo(PropertiesConfiguration properties, Str } } + if (segmentVersion == SegmentVersion.v3 && enableColumnSumValue + && fieldSpec.isSingleValueField() && dataType.isNumeric() && totalDocs > 0) { + Optional sum = columnIndexCreationInfo.getSum(); + sum.ifPresent(sumValue -> properties.setProperty(getKeyFor(column, SUM_VALUE), String.valueOf(sumValue))); + } + String defaultNullValue = columnIndexCreationInfo.getDefaultNullValue().toString(); if (dataType.getStoredType() == DataType.STRING) { // NOTE: Do not limit length of default null value because we need exact value to determine whether the default diff --git a/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/segment/creator/impl/stats/BigDecimalColumnPreIndexStatsCollector.java b/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/segment/creator/impl/stats/BigDecimalColumnPreIndexStatsCollector.java index bd4506e2cf..b66b165994 100644 --- a/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/segment/creator/impl/stats/BigDecimalColumnPreIndexStatsCollector.java +++ b/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/segment/creator/impl/stats/BigDecimalColumnPreIndexStatsCollector.java @@ -21,6 +21,7 @@ import it.unimi.dsi.fastutil.objects.ObjectOpenHashSet; import java.math.BigDecimal; import java.util.Arrays; +import java.util.Optional; import org.apache.pinot.segment.spi.creator.StatsCollectorConfig; import org.apache.pinot.spi.utils.BigDecimalUtils; @@ -35,6 +36,7 @@ public class BigDecimalColumnPreIndexStatsCollector extends AbstractColumnStatis private int _maxRowLength = 0; private BigDecimal[] _sortedValues; private boolean _sealed = false; + private BigDecimal _sum = BigDecimal.ZERO; public BigDecimalColumnPreIndexStatsCollector(String column, StatsCollectorConfig statsCollectorConfig) { super(column, statsCollectorConfig); @@ -48,6 +50,7 @@ public void collect(Object entry) { throw new UnsupportedOperationException(); } else { BigDecimal value = (BigDecimal) entry; + _sum = _sum.add(value); int length = BigDecimalUtils.byteSize(value); addressSorted(value); if (_values.add(value)) { @@ -115,4 +118,9 @@ public void seal() { _sealed = true; } } + + @Override + public Optional getSumValue() { + return _fieldSpec.isSingleValueField() ? Optional.of(_sum.doubleValue()) : Optional.empty(); + } } diff --git a/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/segment/creator/impl/stats/DoubleColumnPreIndexStatsCollector.java b/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/segment/creator/impl/stats/DoubleColumnPreIndexStatsCollector.java index 9edd89b027..fef300f0d1 100644 --- a/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/segment/creator/impl/stats/DoubleColumnPreIndexStatsCollector.java +++ b/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/segment/creator/impl/stats/DoubleColumnPreIndexStatsCollector.java @@ -21,6 +21,7 @@ import it.unimi.dsi.fastutil.doubles.DoubleOpenHashSet; import it.unimi.dsi.fastutil.doubles.DoubleSet; import java.util.Arrays; +import java.util.Optional; import org.apache.pinot.segment.spi.creator.StatsCollectorConfig; @@ -29,6 +30,7 @@ public class DoubleColumnPreIndexStatsCollector extends AbstractColumnStatistics private double[] _sortedValues; private boolean _sealed = false; private double _prevValue = Double.NEGATIVE_INFINITY; + private double _sum = 0; public DoubleColumnPreIndexStatsCollector(String column, StatsCollectorConfig statsCollectorConfig) { super(column, statsCollectorConfig); @@ -57,6 +59,7 @@ public void collect(Object entry) { updateTotalNumberOfEntries(values.length); } else { double value = (double) entry; + _sum += value; addressSorted(value); if (_values.add(value)) { if (isPartitionEnabled()) { @@ -113,4 +116,9 @@ public void seal() { _sealed = true; } } + + @Override + public Optional getSumValue() { + return _fieldSpec.isSingleValueField() ? Optional.of(_sum) : Optional.empty(); + } } diff --git a/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/segment/creator/impl/stats/FloatColumnPreIndexStatsCollector.java b/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/segment/creator/impl/stats/FloatColumnPreIndexStatsCollector.java index 4be248124e..618ce10316 100644 --- a/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/segment/creator/impl/stats/FloatColumnPreIndexStatsCollector.java +++ b/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/segment/creator/impl/stats/FloatColumnPreIndexStatsCollector.java @@ -21,6 +21,7 @@ import it.unimi.dsi.fastutil.floats.FloatOpenHashSet; import it.unimi.dsi.fastutil.floats.FloatSet; import java.util.Arrays; +import java.util.Optional; import org.apache.pinot.segment.spi.creator.StatsCollectorConfig; @@ -29,6 +30,7 @@ public class FloatColumnPreIndexStatsCollector extends AbstractColumnStatisticsC private float[] _sortedValues; private boolean _sealed = false; private float _prevValue = Float.NEGATIVE_INFINITY; + private double _sum = 0; public FloatColumnPreIndexStatsCollector(String column, StatsCollectorConfig statsCollectorConfig) { super(column, statsCollectorConfig); @@ -57,6 +59,7 @@ public void collect(Object entry) { updateTotalNumberOfEntries(values.length); } else { float value = (float) entry; + _sum += value; addressSorted(value); if (_values.add(value)) { if (isPartitionEnabled()) { @@ -113,4 +116,9 @@ public void seal() { _sealed = true; } } + + @Override + public Optional getSumValue() { + return _fieldSpec.isSingleValueField() ? Optional.of(_sum) : Optional.empty(); + } } diff --git a/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/segment/creator/impl/stats/IntColumnPreIndexStatsCollector.java b/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/segment/creator/impl/stats/IntColumnPreIndexStatsCollector.java index 29678f6805..23b42c6013 100644 --- a/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/segment/creator/impl/stats/IntColumnPreIndexStatsCollector.java +++ b/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/segment/creator/impl/stats/IntColumnPreIndexStatsCollector.java @@ -21,6 +21,7 @@ import it.unimi.dsi.fastutil.ints.IntOpenHashSet; import it.unimi.dsi.fastutil.ints.IntSet; import java.util.Arrays; +import java.util.Optional; import org.apache.pinot.segment.spi.creator.StatsCollectorConfig; @@ -29,6 +30,7 @@ public class IntColumnPreIndexStatsCollector extends AbstractColumnStatisticsCol private int[] _sortedValues; private boolean _sealed = false; private int _prevValue = Integer.MIN_VALUE; + private double _sum = 0; public IntColumnPreIndexStatsCollector(String column, StatsCollectorConfig statsCollectorConfig) { super(column, statsCollectorConfig); @@ -57,6 +59,7 @@ public void collect(Object entry) { updateTotalNumberOfEntries(values.length); } else { int value = (int) entry; + _sum += value; addressSorted(value); if (_values.add(value)) { if (isPartitionEnabled()) { @@ -113,4 +116,9 @@ public void seal() { _sealed = true; } } + + @Override + public Optional getSumValue() { + return _fieldSpec.isSingleValueField() ? Optional.of(_sum) : Optional.empty(); + } } diff --git a/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/segment/creator/impl/stats/LongColumnPreIndexStatsCollector.java b/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/segment/creator/impl/stats/LongColumnPreIndexStatsCollector.java index 3d4c63d03b..24c276918b 100644 --- a/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/segment/creator/impl/stats/LongColumnPreIndexStatsCollector.java +++ b/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/segment/creator/impl/stats/LongColumnPreIndexStatsCollector.java @@ -21,6 +21,7 @@ import it.unimi.dsi.fastutil.longs.LongOpenHashSet; import it.unimi.dsi.fastutil.longs.LongSet; import java.util.Arrays; +import java.util.Optional; import org.apache.pinot.segment.spi.creator.StatsCollectorConfig; @@ -29,6 +30,9 @@ public class LongColumnPreIndexStatsCollector extends AbstractColumnStatisticsCo private long[] _sortedValues; private boolean _sealed = false; private long _prevValue = Long.MIN_VALUE; + // Accumulated as double for consistency with Pinot's SUM aggregation return type. + // Note: double has ~15-17 significant digits vs long's 19, so precision loss is possible for very large sums. + private double _sum = 0; public LongColumnPreIndexStatsCollector(String column, StatsCollectorConfig statsCollectorConfig) { super(column, statsCollectorConfig); @@ -57,6 +61,7 @@ public void collect(Object entry) { updateTotalNumberOfEntries(values.length); } else { long value = (long) entry; + _sum += value; addressSorted(value); if (_values.add(value)) { if (isPartitionEnabled()) { @@ -113,4 +118,9 @@ public void seal() { _sealed = true; } } + + @Override + public Optional getSumValue() { + return _fieldSpec.isSingleValueField() ? Optional.of(_sum) : Optional.empty(); + } } diff --git a/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/segment/index/datasource/ImmutableDataSource.java b/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/segment/index/datasource/ImmutableDataSource.java index f3098957ff..44fcebcaa2 100644 --- a/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/segment/index/datasource/ImmutableDataSource.java +++ b/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/segment/index/datasource/ImmutableDataSource.java @@ -18,6 +18,7 @@ */ package org.apache.pinot.segment.local.segment.index.datasource; +import java.util.Optional; import java.util.Set; import javax.annotation.Nullable; import org.apache.pinot.segment.spi.ColumnMetadata; @@ -47,6 +48,7 @@ private static class ImmutableDataSourceMetadata implements DataSourceMetadata { final Comparable _maxValue; final PartitionFunction _partitionFunction; final Set _partitions; + final Optional _sumValue; ImmutableDataSourceMetadata(ColumnMetadata columnMetadata) { _fieldSpec = columnMetadata.getFieldSpec(); @@ -63,6 +65,7 @@ private static class ImmutableDataSourceMetadata implements DataSourceMetadata { _partitionFunction = columnMetadata.getPartitionFunction(); _partitions = columnMetadata.getPartitions(); _cardinality = columnMetadata.getCardinality(); + _sumValue = columnMetadata.getSumValue(); } @Override @@ -118,5 +121,10 @@ public Set getPartitions() { public int getCardinality() { return _cardinality; } + + @Override + public Optional getSumValue() { + return _sumValue; + } } } diff --git a/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/segment/index/loader/defaultcolumn/BaseDefaultColumnHandler.java b/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/segment/index/loader/defaultcolumn/BaseDefaultColumnHandler.java index 46e3165be1..6c51767faf 100644 --- a/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/segment/index/loader/defaultcolumn/BaseDefaultColumnHandler.java +++ b/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/segment/index/loader/defaultcolumn/BaseDefaultColumnHandler.java @@ -567,7 +567,8 @@ private void createDefaultValueColumnV1Indices(String column) // Add the column metadata information to the metadata properties. SegmentColumnarIndexCreator.addColumnMetadataInfo(_segmentProperties, column, columnIndexCreationInfo, totalDocs, - fieldSpec, true/*hasDictionary*/, dictionaryElementSize); + fieldSpec, true/*hasDictionary*/, dictionaryElementSize, _segmentMetadata.getVersion(), + isEnableColumnSumValue()); } private boolean isNullable(FieldSpec fieldSpec) { @@ -580,6 +581,12 @@ private boolean isNullable(FieldSpec fieldSpec) { } } + private boolean isEnableColumnSumValue() { + return _indexLoadingConfig.getTableConfig() != null + && _indexLoadingConfig.getTableConfig().getIndexingConfig() != null + && _indexLoadingConfig.getTableConfig().getIndexingConfig().isEnableColumnSumValue(); + } + /** * Helper method to create the V1 indices (dictionary and forward index) for a column with derived values. * TODO: @@ -1107,7 +1114,8 @@ private void createDerivedColumnForwardIndexWithDictionary(String column, FieldS } // Add the column metadata SegmentColumnarIndexCreator.addColumnMetadataInfo(_segmentProperties, column, indexCreationInfo, numDocs, - fieldSpec, true, dictionaryCreator.getNumBytesPerEntry()); + fieldSpec, true, dictionaryCreator.getNumBytesPerEntry(), _segmentMetadata.getVersion(), + isEnableColumnSumValue()); } } } @@ -1185,7 +1193,7 @@ private void createDerivedColumnForwardIndexWithoutDictionary(String column, Fie // Add the column metadata SegmentColumnarIndexCreator.addColumnMetadataInfo(_segmentProperties, column, indexCreationInfo, numDocs, - fieldSpec, false, 0); + fieldSpec, false, 0, _segmentMetadata.getVersion(), isEnableColumnSumValue()); } private ForwardIndexCreator getForwardIndexCreator(FieldSpec fieldSpec, ColumnIndexCreationInfo indexCreationInfo, diff --git a/pinot-segment-local/src/test/java/org/apache/pinot/segment/local/segment/creator/impl/SegmentColumnarIndexCreatorTest.java b/pinot-segment-local/src/test/java/org/apache/pinot/segment/local/segment/creator/impl/SegmentColumnarIndexCreatorTest.java index 71bd9b2621..ace3fd671c 100644 --- a/pinot-segment-local/src/test/java/org/apache/pinot/segment/local/segment/creator/impl/SegmentColumnarIndexCreatorTest.java +++ b/pinot-segment-local/src/test/java/org/apache/pinot/segment/local/segment/creator/impl/SegmentColumnarIndexCreatorTest.java @@ -26,13 +26,19 @@ import org.apache.commons.io.FileUtils; import org.apache.pinot.segment.local.PinotBuffersAfterClassCheckRule; import org.apache.pinot.segment.local.indexsegment.immutable.ImmutableSegmentLoader; +import org.apache.pinot.segment.local.segment.creator.impl.stats.IntColumnPreIndexStatsCollector; import org.apache.pinot.segment.local.segment.readers.GenericRowRecordReader; import org.apache.pinot.segment.spi.IndexSegment; import org.apache.pinot.segment.spi.SegmentMetadata; +import org.apache.pinot.segment.spi.V1Constants; import org.apache.pinot.segment.spi.V1Constants.MetadataKeys.Column; +import org.apache.pinot.segment.spi.creator.ColumnIndexCreationInfo; import org.apache.pinot.segment.spi.creator.SegmentGeneratorConfig; +import org.apache.pinot.segment.spi.creator.SegmentVersion; +import org.apache.pinot.segment.spi.creator.StatsCollectorConfig; import org.apache.pinot.spi.config.table.TableConfig; import org.apache.pinot.spi.config.table.TableType; +import org.apache.pinot.spi.data.FieldSpec; import org.apache.pinot.spi.data.FieldSpec.DataType; import org.apache.pinot.spi.data.Schema; import org.apache.pinot.spi.data.readers.GenericRow; @@ -43,8 +49,10 @@ import org.testng.annotations.BeforeClass; import org.testng.annotations.Test; +import static org.apache.pinot.segment.spi.V1Constants.MetadataKeys.Segment.SEGMENT_PADDING_CHARACTER; import static org.testng.Assert.assertEquals; import static org.testng.Assert.assertFalse; +import static org.testng.Assert.assertNull; import static org.testng.Assert.assertTrue; @@ -133,6 +141,108 @@ private static long getStartTimeInSegmentMetadata(String testDateTimeFormat, Str } } + @Test + public void testSumValueWrittenForV3SingleValueNumericColumn() { + String column = "intCol"; + Schema schema = new Schema.SchemaBuilder().setSchemaName("testTable") + .addSingleValueDimension(column, DataType.INT).build(); + TableConfig tableConfig = new TableConfigBuilder(TableType.OFFLINE).setTableName("testTable").build(); + StatsCollectorConfig statsConfig = new StatsCollectorConfig(tableConfig, schema, null); + + IntColumnPreIndexStatsCollector collector = new IntColumnPreIndexStatsCollector(column, statsConfig); + collector.collect(100); + collector.collect(200); + collector.collect(250); + collector.seal(); + + ColumnIndexCreationInfo info = new ColumnIndexCreationInfo(collector, true, false, false, 0); + FieldSpec fieldSpec = schema.getFieldSpecFor(column); + PropertiesConfiguration properties = new PropertiesConfiguration(); + properties.setProperty(SEGMENT_PADDING_CHARACTER, String.valueOf(V1Constants.Str.DEFAULT_STRING_PAD_CHAR)); + + SegmentColumnarIndexCreator.addColumnMetadataInfo(properties, column, info, 3, fieldSpec, true, 4, + SegmentVersion.v3, true); + + assertEquals(properties.getString(Column.getKeyFor(column, Column.SUM_VALUE)), "550.0"); + } + + @Test + public void testSumValueNotWrittenForV1() { + String column = "intCol"; + Schema schema = new Schema.SchemaBuilder().setSchemaName("testTable") + .addSingleValueDimension(column, DataType.INT).build(); + TableConfig tableConfig = new TableConfigBuilder(TableType.OFFLINE).setTableName("testTable").build(); + StatsCollectorConfig statsConfig = new StatsCollectorConfig(tableConfig, schema, null); + + IntColumnPreIndexStatsCollector collector = new IntColumnPreIndexStatsCollector(column, statsConfig); + collector.collect(100); + collector.collect(200); + collector.collect(250); + collector.seal(); + + ColumnIndexCreationInfo info = new ColumnIndexCreationInfo(collector, true, false, false, 0); + FieldSpec fieldSpec = schema.getFieldSpecFor(column); + PropertiesConfiguration properties = new PropertiesConfiguration(); + properties.setProperty(SEGMENT_PADDING_CHARACTER, String.valueOf(V1Constants.Str.DEFAULT_STRING_PAD_CHAR)); + + SegmentColumnarIndexCreator.addColumnMetadataInfo(properties, column, info, 3, fieldSpec, true, 4, + SegmentVersion.v1, true); + + assertNull(properties.getString(Column.getKeyFor(column, Column.SUM_VALUE), null)); + } + + @Test + public void testSumValueNotWrittenForMultiValueColumn() { + String column = "intMvCol"; + Schema schema = new Schema.SchemaBuilder().setSchemaName("testTable") + .addMultiValueDimension(column, DataType.INT).build(); + TableConfig tableConfig = new TableConfigBuilder(TableType.OFFLINE).setTableName("testTable").build(); + StatsCollectorConfig statsConfig = new StatsCollectorConfig(tableConfig, schema, null); + + IntColumnPreIndexStatsCollector collector = new IntColumnPreIndexStatsCollector(column, statsConfig); + collector.collect(new int[]{100, 200}); + collector.collect(new int[]{250}); + collector.seal(); + + // MV collector returns empty Optional for getSumValue() + assertEquals(collector.getSumValue(), java.util.Optional.empty()); + + ColumnIndexCreationInfo info = new ColumnIndexCreationInfo(collector, true, false, false, 0); + FieldSpec fieldSpec = schema.getFieldSpecFor(column); + PropertiesConfiguration properties = new PropertiesConfiguration(); + properties.setProperty(SEGMENT_PADDING_CHARACTER, String.valueOf(V1Constants.Str.DEFAULT_STRING_PAD_CHAR)); + + SegmentColumnarIndexCreator.addColumnMetadataInfo(properties, column, info, 2, fieldSpec, true, 4, + SegmentVersion.v3, true); + + assertNull(properties.getString(Column.getKeyFor(column, Column.SUM_VALUE), null)); + } + + @Test + public void testSumValueNotWrittenWithoutSegmentVersion() { + String column = "intCol"; + Schema schema = new Schema.SchemaBuilder().setSchemaName("testTable") + .addSingleValueDimension(column, DataType.INT).build(); + TableConfig tableConfig = new TableConfigBuilder(TableType.OFFLINE).setTableName("testTable").build(); + StatsCollectorConfig statsConfig = new StatsCollectorConfig(tableConfig, schema, null); + + IntColumnPreIndexStatsCollector collector = new IntColumnPreIndexStatsCollector(column, statsConfig); + collector.collect(100); + collector.collect(200); + collector.collect(250); + collector.seal(); + + ColumnIndexCreationInfo info = new ColumnIndexCreationInfo(collector, true, false, false, 0); + FieldSpec fieldSpec = schema.getFieldSpecFor(column); + PropertiesConfiguration properties = new PropertiesConfiguration(); + properties.setProperty(SEGMENT_PADDING_CHARACTER, String.valueOf(V1Constants.Str.DEFAULT_STRING_PAD_CHAR)); + + // Overload without segmentVersion + SegmentColumnarIndexCreator.addColumnMetadataInfo(properties, column, info, 3, fieldSpec, true, 4); + + assertNull(properties.getString(Column.getKeyFor(column, Column.SUM_VALUE), null)); + } + @AfterClass public void tearDown() throws IOException { diff --git a/pinot-segment-local/src/test/java/org/apache/pinot/segment/local/segment/creator/impl/stats/PreIndexStatsCollectorSumTest.java b/pinot-segment-local/src/test/java/org/apache/pinot/segment/local/segment/creator/impl/stats/PreIndexStatsCollectorSumTest.java new file mode 100644 index 0000000000..e939ae6c96 --- /dev/null +++ b/pinot-segment-local/src/test/java/org/apache/pinot/segment/local/segment/creator/impl/stats/PreIndexStatsCollectorSumTest.java @@ -0,0 +1,150 @@ +/** + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ +package org.apache.pinot.segment.local.segment.creator.impl.stats; + +import java.math.BigDecimal; +import java.util.Optional; +import org.apache.pinot.segment.spi.creator.StatsCollectorConfig; +import org.apache.pinot.spi.config.table.TableConfig; +import org.apache.pinot.spi.config.table.TableType; +import org.apache.pinot.spi.data.FieldSpec.DataType; +import org.apache.pinot.spi.data.Schema; +import org.apache.pinot.spi.utils.builder.TableConfigBuilder; +import org.testng.annotations.Test; + +import static org.testng.Assert.assertEquals; + + +public class PreIndexStatsCollectorSumTest { + private static final String TABLE_NAME = "testTable"; + private static final String COLUMN_NAME = "testColumn"; + private static final TableConfig TABLE_CONFIG = + new TableConfigBuilder(TableType.OFFLINE).setTableName(TABLE_NAME).build(); + + private StatsCollectorConfig createSVConfig(DataType dataType) { + Schema schema = new Schema.SchemaBuilder().setSchemaName(TABLE_NAME) + .addSingleValueDimension(COLUMN_NAME, dataType).build(); + return new StatsCollectorConfig(TABLE_CONFIG, schema, null); + } + + private StatsCollectorConfig createMVConfig(DataType dataType) { + Schema schema = new Schema.SchemaBuilder().setSchemaName(TABLE_NAME) + .addMultiValueDimension(COLUMN_NAME, dataType).build(); + return new StatsCollectorConfig(TABLE_CONFIG, schema, null); + } + + @Test + public void testIntSVSum() { + IntColumnPreIndexStatsCollector collector = + new IntColumnPreIndexStatsCollector(COLUMN_NAME, createSVConfig(DataType.INT)); + collector.collect(10); + collector.collect(20); + collector.collect(-5); + collector.seal(); + assertEquals(collector.getSumValue(), Optional.of(25.0)); + } + + @Test + public void testIntMVSumReturnsEmpty() { + IntColumnPreIndexStatsCollector collector = + new IntColumnPreIndexStatsCollector(COLUMN_NAME, createMVConfig(DataType.INT)); + collector.collect(new int[]{10, 20}); + collector.seal(); + assertEquals(collector.getSumValue(), Optional.empty()); + } + + @Test + public void testLongSVSum() { + LongColumnPreIndexStatsCollector collector = + new LongColumnPreIndexStatsCollector(COLUMN_NAME, createSVConfig(DataType.LONG)); + collector.collect(100L); + collector.collect(200L); + collector.collect(-50L); + collector.seal(); + assertEquals(collector.getSumValue(), Optional.of(250.0)); + } + + @Test + public void testLongMVSumReturnsEmpty() { + LongColumnPreIndexStatsCollector collector = + new LongColumnPreIndexStatsCollector(COLUMN_NAME, createMVConfig(DataType.LONG)); + collector.collect(new long[]{100L, 200L}); + collector.seal(); + assertEquals(collector.getSumValue(), Optional.empty()); + } + + @Test + public void testFloatSVSum() { + FloatColumnPreIndexStatsCollector collector = + new FloatColumnPreIndexStatsCollector(COLUMN_NAME, createSVConfig(DataType.FLOAT)); + collector.collect(1.0f); + collector.collect(2.0f); + collector.collect(3.0f); + collector.seal(); + assertEquals(collector.getSumValue(), Optional.of(6.0)); + } + + @Test + public void testFloatMVSumReturnsEmpty() { + FloatColumnPreIndexStatsCollector collector = + new FloatColumnPreIndexStatsCollector(COLUMN_NAME, createMVConfig(DataType.FLOAT)); + collector.collect(new float[]{1.0f, 2.0f}); + collector.seal(); + assertEquals(collector.getSumValue(), Optional.empty()); + } + + @Test + public void testDoubleSVSum() { + DoubleColumnPreIndexStatsCollector collector = + new DoubleColumnPreIndexStatsCollector(COLUMN_NAME, createSVConfig(DataType.DOUBLE)); + collector.collect(1.0); + collector.collect(2.0); + collector.collect(3.0); + collector.seal(); + assertEquals(collector.getSumValue(), Optional.of(6.0)); + } + + @Test + public void testDoubleMVSumReturnsEmpty() { + DoubleColumnPreIndexStatsCollector collector = + new DoubleColumnPreIndexStatsCollector(COLUMN_NAME, createMVConfig(DataType.DOUBLE)); + collector.collect(new double[]{1.0, 2.0}); + collector.seal(); + assertEquals(collector.getSumValue(), Optional.empty()); + } + + @Test + public void testBigDecimalSVSum() { + BigDecimalColumnPreIndexStatsCollector collector = + new BigDecimalColumnPreIndexStatsCollector(COLUMN_NAME, createSVConfig(DataType.BIG_DECIMAL)); + collector.collect(new BigDecimal("10")); + collector.collect(new BigDecimal("20")); + collector.collect(new BigDecimal("30")); + collector.seal(); + assertEquals(collector.getSumValue(), Optional.of(60.0)); + } + + @Test + public void testSVSumNoEntries() { + IntColumnPreIndexStatsCollector collector = + new IntColumnPreIndexStatsCollector(COLUMN_NAME, createSVConfig(DataType.INT)); + collector.seal(); + assertEquals(collector.getSumValue(), Optional.of(0.0)); + } +} diff --git a/pinot-segment-local/src/test/java/org/apache/pinot/segment/local/segment/index/ColumnMetadataTest.java b/pinot-segment-local/src/test/java/org/apache/pinot/segment/local/segment/index/ColumnMetadataTest.java index b9168e49bd..83597b41fc 100644 --- a/pinot-segment-local/src/test/java/org/apache/pinot/segment/local/segment/index/ColumnMetadataTest.java +++ b/pinot-segment-local/src/test/java/org/apache/pinot/segment/local/segment/index/ColumnMetadataTest.java @@ -23,6 +23,7 @@ import java.util.Collections; import java.util.HashMap; import java.util.Map; +import java.util.Optional; import java.util.concurrent.TimeUnit; import java.util.stream.Collectors; import java.util.stream.Stream; @@ -228,6 +229,42 @@ public void testMetadataWithEscapedValue() "\r\n\r\n utils em::C:\\dir\\utils\r\nPSParentPath : Mi"); } + @Test + public void testSumValueRoundTrip() { + // Verify sumValue is deserialized from properties when present + String column = "intCol"; + DimensionFieldSpec fieldSpec = new DimensionFieldSpec(column, DataType.INT, true); + ColumnIndexCreationInfo columnIndexCreationInfo = + new ColumnIndexCreationInfo(new DefaultColumnStatistics(0, 100, new int[]{0, 100}, true, 10, 0), + true, false, false, 0); + PropertiesConfiguration config = new PropertiesConfiguration(); + config.setProperty(SEGMENT_PADDING_CHARACTER, String.valueOf(V1Constants.Str.DEFAULT_STRING_PAD_CHAR)); + SegmentColumnarIndexCreator.addColumnMetadataInfo(config, column, columnIndexCreationInfo, 10, fieldSpec, + true, 4); + // Manually set sumValue to simulate what a real stats collector would produce + config.setProperty( + V1Constants.MetadataKeys.Column.getKeyFor(column, V1Constants.MetadataKeys.Column.SUM_VALUE), "550.0"); + ColumnMetadataImpl metadata = ColumnMetadataImpl.fromPropertiesConfiguration(column, config); + Assert.assertNotNull(metadata.getSumValue()); + Assert.assertEquals(metadata.getSumValue(), Optional.of((Double) 550.0)); + } + + @Test + public void testSumValueAbsent() { + // Column metadata without sumValue property should return null + String column = "intCol"; + DimensionFieldSpec fieldSpec = new DimensionFieldSpec(column, DataType.INT, true); + ColumnIndexCreationInfo columnIndexCreationInfo = + new ColumnIndexCreationInfo(new DefaultColumnStatistics(0, 100, new int[]{0, 100}, true, 10, 0), + true, false, false, 0); + PropertiesConfiguration config = new PropertiesConfiguration(); + config.setProperty(SEGMENT_PADDING_CHARACTER, String.valueOf(V1Constants.Str.DEFAULT_STRING_PAD_CHAR)); + SegmentColumnarIndexCreator.addColumnMetadataInfo(config, column, columnIndexCreationInfo, 10, fieldSpec, + true, 4); + ColumnMetadataImpl metadata = ColumnMetadataImpl.fromPropertiesConfiguration(column, config); + Assert.assertEquals(metadata.getSumValue(), Optional.empty()); + } + @Test public void testComplexFieldSpec() { ComplexFieldSpec intMapFieldSpec = new ComplexFieldSpec("intMap", DataType.MAP, true, Map.of( diff --git a/pinot-segment-spi/src/main/java/org/apache/pinot/segment/spi/ColumnMetadata.java b/pinot-segment-spi/src/main/java/org/apache/pinot/segment/spi/ColumnMetadata.java index 2f6177ce5b..93650613f1 100644 --- a/pinot-segment-spi/src/main/java/org/apache/pinot/segment/spi/ColumnMetadata.java +++ b/pinot-segment-spi/src/main/java/org/apache/pinot/segment/spi/ColumnMetadata.java @@ -20,6 +20,7 @@ import com.fasterxml.jackson.annotation.JsonProperty; import java.util.Map; +import java.util.Optional; import java.util.Set; import javax.annotation.Nullable; import org.apache.pinot.segment.spi.index.IndexType; @@ -93,4 +94,8 @@ default boolean isMinMaxValueInvalid() { Map, Long> getIndexSizeMap(); boolean isAutoGenerated(); + + default Optional getSumValue() { + return Optional.empty(); + } } diff --git a/pinot-segment-spi/src/main/java/org/apache/pinot/segment/spi/V1Constants.java b/pinot-segment-spi/src/main/java/org/apache/pinot/segment/spi/V1Constants.java index 35b319dd60..126e0e4361 100644 --- a/pinot-segment-spi/src/main/java/org/apache/pinot/segment/spi/V1Constants.java +++ b/pinot-segment-spi/src/main/java/org/apache/pinot/segment/spi/V1Constants.java @@ -117,6 +117,7 @@ public static class Column { public static final String DEFAULT_NULL_VALUE = "defaultNullValue"; public static final String MIN_VALUE = "minValue"; public static final String MAX_VALUE = "maxValue"; + public static final String SUM_VALUE = "sumValue"; public static final String MIN_MAX_VALUE_INVALID = "minMaxValueInvalid"; public static final String PARTITION_FUNCTION = "partitionFunction"; public static final String PARTITION_FUNCTION_CONFIG = "partitionFunctionConfig"; diff --git a/pinot-segment-spi/src/main/java/org/apache/pinot/segment/spi/creator/ColumnIndexCreationInfo.java b/pinot-segment-spi/src/main/java/org/apache/pinot/segment/spi/creator/ColumnIndexCreationInfo.java index 9e47502098..2171a4660e 100644 --- a/pinot-segment-spi/src/main/java/org/apache/pinot/segment/spi/creator/ColumnIndexCreationInfo.java +++ b/pinot-segment-spi/src/main/java/org/apache/pinot/segment/spi/creator/ColumnIndexCreationInfo.java @@ -20,6 +20,7 @@ import java.io.Serializable; import java.util.Map; +import java.util.Optional; import java.util.Set; import javax.annotation.Nullable; import org.apache.commons.lang3.ArrayUtils; @@ -63,6 +64,10 @@ public Object getMax() { return _columnStatistics.getMaxValue(); } + public Optional getSum() { + return _columnStatistics.getSumValue(); + } + public Object getSortedUniqueElementsArray() { return _columnStatistics.getUniqueValuesSet(); } diff --git a/pinot-segment-spi/src/main/java/org/apache/pinot/segment/spi/creator/ColumnStatistics.java b/pinot-segment-spi/src/main/java/org/apache/pinot/segment/spi/creator/ColumnStatistics.java index e904fd0a9a..1704871672 100644 --- a/pinot-segment-spi/src/main/java/org/apache/pinot/segment/spi/creator/ColumnStatistics.java +++ b/pinot-segment-spi/src/main/java/org/apache/pinot/segment/spi/creator/ColumnStatistics.java @@ -20,6 +20,7 @@ import java.io.Serializable; import java.util.Map; +import java.util.Optional; import java.util.Set; import org.apache.pinot.segment.spi.partition.PartitionFunction; @@ -97,4 +98,11 @@ default int getMaxRowLengthInBytes() { Map getPartitionFunctionConfig(); Set getPartitions(); + + /** + * @return Sum value of the column + */ + default Optional getSumValue() { + return Optional.empty(); + } } diff --git a/pinot-segment-spi/src/main/java/org/apache/pinot/segment/spi/creator/SegmentGeneratorConfig.java b/pinot-segment-spi/src/main/java/org/apache/pinot/segment/spi/creator/SegmentGeneratorConfig.java index fff8782634..257b56590d 100644 --- a/pinot-segment-spi/src/main/java/org/apache/pinot/segment/spi/creator/SegmentGeneratorConfig.java +++ b/pinot-segment-spi/src/main/java/org/apache/pinot/segment/spi/creator/SegmentGeneratorConfig.java @@ -115,6 +115,7 @@ public enum TimeColumnType { private boolean _optimizeDictionaryType = false; private double _noDictionarySizeRatioThreshold = IndexingConfig.DEFAULT_NO_DICTIONARY_SIZE_RATIO_THRESHOLD; private Double _noDictionaryCardinalityRatioThreshold; + private boolean _enableColumnSumValue = false; private boolean _realtimeConversion = false; // consumerDir contains data from the consuming segment, and is used during _realtimeConversion optimization private File _consumerDir; @@ -172,6 +173,7 @@ public SegmentGeneratorConfig(TableConfig tableConfig, Schema schema, boolean cr _optimizeDictionaryType = indexingConfig.isOptimizeDictionaryType(); _noDictionarySizeRatioThreshold = indexingConfig.getNoDictionarySizeRatioThreshold(); _noDictionaryCardinalityRatioThreshold = indexingConfig.getNoDictionaryCardinalityRatioThreshold(); + _enableColumnSumValue = indexingConfig.isEnableColumnSumValue(); // Star-tree configs setStarTreeIndexConfigs(indexingConfig.getStarTreeIndexConfigs()); @@ -669,6 +671,14 @@ public double getNoDictionarySizeRatioThreshold() { return _noDictionarySizeRatioThreshold; } + public boolean isEnableColumnSumValue() { + return _enableColumnSumValue; + } + + public void setEnableColumnSumValue(boolean enableColumnSumValue) { + _enableColumnSumValue = enableColumnSumValue; + } + public boolean isRealtimeConversion() { return _realtimeConversion; } diff --git a/pinot-segment-spi/src/main/java/org/apache/pinot/segment/spi/datasource/DataSourceMetadata.java b/pinot-segment-spi/src/main/java/org/apache/pinot/segment/spi/datasource/DataSourceMetadata.java index 3f14601d32..02dcb38313 100644 --- a/pinot-segment-spi/src/main/java/org/apache/pinot/segment/spi/datasource/DataSourceMetadata.java +++ b/pinot-segment-spi/src/main/java/org/apache/pinot/segment/spi/datasource/DataSourceMetadata.java @@ -18,6 +18,7 @@ */ package org.apache.pinot.segment.spi.datasource; +import java.util.Optional; import java.util.Set; import javax.annotation.Nullable; import org.apache.pinot.segment.spi.partition.PartitionFunction; @@ -108,4 +109,11 @@ default boolean isSingleValue() { default int getMaxRowLengthInBytes() { return -1; } + + /** + * Returns the sum value of the column, or empty if it is not available. + */ + default Optional getSumValue() { + return Optional.empty(); + } } diff --git a/pinot-segment-spi/src/main/java/org/apache/pinot/segment/spi/index/metadata/ColumnMetadataImpl.java b/pinot-segment-spi/src/main/java/org/apache/pinot/segment/spi/index/metadata/ColumnMetadataImpl.java index 91eed38295..febcad79a9 100644 --- a/pinot-segment-spi/src/main/java/org/apache/pinot/segment/spi/index/metadata/ColumnMetadataImpl.java +++ b/pinot-segment-spi/src/main/java/org/apache/pinot/segment/spi/index/metadata/ColumnMetadataImpl.java @@ -18,6 +18,7 @@ */ package org.apache.pinot.segment.spi.index.metadata; +import com.fasterxml.jackson.annotation.JsonInclude; import com.google.common.base.Preconditions; import java.math.BigDecimal; import java.util.HashMap; @@ -25,6 +26,7 @@ import java.util.List; import java.util.Map; import java.util.Objects; +import java.util.Optional; import java.util.Set; import java.util.concurrent.TimeUnit; import javax.annotation.Nullable; @@ -69,12 +71,13 @@ public class ColumnMetadataImpl implements ColumnMetadata { private final Set _partitions; private final Map, Long> _indexSizeMap; private final boolean _autoGenerated; + private final Double _sumValue; private ColumnMetadataImpl(FieldSpec fieldSpec, int totalDocs, int cardinality, boolean sorted, Comparable minValue, Comparable maxValue, boolean minMaxValueInvalid, boolean hasDictionary, int columnMaxLength, int bitsPerElement, int maxNumberOfMultiValues, int totalNumberOfEntries, @Nullable PartitionFunction partitionFunction, @Nullable Set partitions, - Map, Long> indexSizeMap, boolean autoGenerated) { + Map, Long> indexSizeMap, boolean autoGenerated, Double sumValue) { _fieldSpec = fieldSpec; _totalDocs = totalDocs; _cardinality = cardinality; @@ -91,6 +94,7 @@ private ColumnMetadataImpl(FieldSpec fieldSpec, int totalDocs, int cardinality, _partitions = partitions; _indexSizeMap = indexSizeMap; _autoGenerated = autoGenerated; + _sumValue = sumValue; } @Override @@ -175,6 +179,12 @@ public boolean isAutoGenerated() { return _autoGenerated; } + @JsonInclude(JsonInclude.Include.NON_ABSENT) + @Override + public Optional getSumValue() { + return Optional.ofNullable(_sumValue); + } + @Override public boolean equals(Object o) { if (this == o) { @@ -189,15 +199,16 @@ public boolean equals(Object o) { && _bitsPerElement == that._bitsPerElement && _maxNumberOfMultiValues == that._maxNumberOfMultiValues && _totalNumberOfEntries == that._totalNumberOfEntries && _autoGenerated == that._autoGenerated && Objects.equals(_fieldSpec, that._fieldSpec) && Objects.equals(_minValue, that._minValue) && Objects.equals( - _maxValue, that._maxValue) && Objects.equals(_partitionFunction, that._partitionFunction) && Objects.equals( - _partitions, that._partitions); + _maxValue, that._maxValue) && Objects.equals(_partitionFunction, that._partitionFunction) + && Objects.equals(_partitions, that._partitions) + && Objects.equals(_sumValue, that._sumValue); } @Override public int hashCode() { return Objects.hash(_fieldSpec, _totalDocs, _cardinality, _sorted, _minValue, _maxValue, _hasDictionary, _columnMaxLength, _bitsPerElement, _maxNumberOfMultiValues, _totalNumberOfEntries, _partitionFunction, - _partitions, _autoGenerated); + _partitions, _autoGenerated, _sumValue); } @Override @@ -207,7 +218,7 @@ public String toString() { + ", _hasDictionary=" + _hasDictionary + ", _columnMaxLength=" + _columnMaxLength + ", _bitsPerElement=" + _bitsPerElement + ", _maxNumberOfMultiValues=" + _maxNumberOfMultiValues + ", _totalNumberOfEntries=" + _totalNumberOfEntries + ", _partitionFunction=" + _partitionFunction + ", _partitions=" + _partitions - + ", _autoGenerated=" + _autoGenerated + '}'; + + ", _autoGenerated=" + _autoGenerated + ", _sumValue=" + _sumValue + '}'; } public static ColumnMetadataImpl fromPropertiesConfiguration(String column, PropertiesConfiguration config) { @@ -237,6 +248,7 @@ public static ColumnMetadataImpl fromPropertiesConfiguration(String column, Prop // TODO: Use getProperty() for other properties as well to avoid the overhead of variable substitution String minString = (String) config.getProperty(Column.getKeyFor(column, Column.MIN_VALUE)); String maxString = (String) config.getProperty(Column.getKeyFor(column, Column.MAX_VALUE)); + // Set min/max value if available if (minString != null) { builder.setMinValue(builder.parseValue(storedType, column, minString)); @@ -246,6 +258,11 @@ public static ColumnMetadataImpl fromPropertiesConfiguration(String column, Prop builder.setMaxValue(builder.parseValue(storedType, column, maxString)); } builder.setMinMaxValueInvalid(config.getBoolean(Column.getKeyFor(column, Column.MIN_MAX_VALUE_INVALID), false)); + + String sumString = (String) config.getProperty(Column.getKeyFor(column, Column.SUM_VALUE)); + if (sumString != null) { + builder.setSumValue(Double.valueOf(sumString)); + } } // Only support zero padding String padding = config.getString(Segment.SEGMENT_PADDING_CHARACTER, null); @@ -356,6 +373,7 @@ public static class Builder { private Set _partitions; private boolean _autoGenerated; private Map, Long> _indexSizeMap = new HashMap<>(); + private Double _sumValue; public Builder setFieldSpec(FieldSpec fieldSpec) { _fieldSpec = fieldSpec; @@ -387,6 +405,11 @@ public Builder setMaxValue(Comparable maxValue) { return this; } + public Builder setSumValue(Double sumValue) { + _sumValue = sumValue; + return this; + } + public Builder setMinMaxValueInvalid(boolean minMaxValueInvalid) { _minMaxValueInvalid = minMaxValueInvalid; return this; @@ -439,7 +462,7 @@ public Builder setAutoGenerated(boolean autoGenerated) { public ColumnMetadataImpl build() { return new ColumnMetadataImpl(_fieldSpec, _totalDocs, _cardinality, _sorted, _minValue, _maxValue, _minMaxValueInvalid, _hasDictionary, _columnMaxLength, _bitsPerElement, _maxNumberOfMultiValues, - _totalNumberOfEntries, _partitionFunction, _partitions, _indexSizeMap, _autoGenerated); + _totalNumberOfEntries, _partitionFunction, _partitions, _indexSizeMap, _autoGenerated, _sumValue); } private Comparable parseValue(DataType storedType, String column, String valueString) { diff --git a/pinot-spi/pom.xml b/pinot-spi/pom.xml index 91927379c0..25aa67217d 100644 --- a/pinot-spi/pom.xml +++ b/pinot-spi/pom.xml @@ -191,6 +191,10 @@ com.fasterxml.jackson.core jackson-databind + + com.fasterxml.jackson.datatype + jackson-datatype-jdk8 + org.reflections reflections diff --git a/pinot-spi/src/main/java/org/apache/pinot/spi/config/table/IndexingConfig.java b/pinot-spi/src/main/java/org/apache/pinot/spi/config/table/IndexingConfig.java index b777bbc294..16f17fec23 100644 --- a/pinot-spi/src/main/java/org/apache/pinot/spi/config/table/IndexingConfig.java +++ b/pinot-spi/src/main/java/org/apache/pinot/spi/config/table/IndexingConfig.java @@ -62,6 +62,7 @@ public class IndexingConfig extends BaseJsonConfig { private boolean _enableDynamicStarTreeCreation; private SegmentPartitionConfig _segmentPartitionConfig; private boolean _aggregateMetrics; + private boolean _enableColumnSumValue; private boolean _nullHandlingEnabled; private boolean _columnMajorSegmentBuilderEnabled = true; private boolean _skipSegmentPreprocess; @@ -322,6 +323,14 @@ public void setAggregateMetrics(boolean value) { _aggregateMetrics = value; } + public boolean isEnableColumnSumValue() { + return _enableColumnSumValue; + } + + public void setEnableColumnSumValue(boolean enableColumnSumValue) { + _enableColumnSumValue = enableColumnSumValue; + } + /** * Returns {@code true} if null handling is enabled at table config level. * diff --git a/pinot-spi/src/main/java/org/apache/pinot/spi/utils/JsonUtils.java b/pinot-spi/src/main/java/org/apache/pinot/spi/utils/JsonUtils.java index 1f80cc9fc0..fed2d8fb37 100644 --- a/pinot-spi/src/main/java/org/apache/pinot/spi/utils/JsonUtils.java +++ b/pinot-spi/src/main/java/org/apache/pinot/spi/utils/JsonUtils.java @@ -30,6 +30,7 @@ import com.fasterxml.jackson.databind.node.ArrayNode; import com.fasterxml.jackson.databind.node.JsonNodeFactory; import com.fasterxml.jackson.databind.node.ObjectNode; +import com.fasterxml.jackson.datatype.jdk8.Jdk8Module; import com.google.common.base.Preconditions; import com.google.common.collect.MapDifference; import com.google.common.collect.Maps; @@ -88,7 +89,8 @@ private JsonUtils() { // NOTE: Do not expose the ObjectMapper to prevent configuration change - private static final ObjectMapper DEFAULT_MAPPER = new ObjectMapper(); + private static final ObjectMapper DEFAULT_MAPPER = new ObjectMapper() + .registerModule(new Jdk8Module()); public static final ObjectReader DEFAULT_READER = DEFAULT_MAPPER.reader(); public static final ObjectWriter DEFAULT_WRITER = DEFAULT_MAPPER.writer(); public static final ObjectWriter DEFAULT_PRETTY_WRITER = DEFAULT_MAPPER.writerWithDefaultPrettyPrinter();