diff --git a/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/realtime/converter/stats/MutableNoDictColumnStatistics.java b/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/realtime/converter/stats/MutableNoDictColumnStatistics.java index 85c0e2423298..aeacde865ebd 100644 --- a/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/realtime/converter/stats/MutableNoDictColumnStatistics.java +++ b/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/realtime/converter/stats/MutableNoDictColumnStatistics.java @@ -45,6 +45,14 @@ public class MutableNoDictColumnStatistics implements ColumnStatistics, CLPStats protected final boolean _isSortedColumn; protected final MutableForwardIndex _forwardIndex; + // Lazily-computed min/max for columns whose mutable segment does not track them (ingestion-aggregated metric + // columns). Populated on first access by scanning the sealed forward index; see computeMinMaxIfNeeded(). + private boolean _minMaxComputed; + @Nullable + private Comparable _computedMinValue; + @Nullable + private Comparable _computedMaxValue; + public MutableNoDictColumnStatistics(DataSource dataSource, @Nullable int[] sortedDocIds, boolean isSortedColumn) { _dataSourceMetadata = dataSource.getDataSourceMetadata(); _fieldSpec = _dataSourceMetadata.getFieldSpec(); @@ -69,12 +77,75 @@ public int getTotalDocs() { @Override public Comparable getMinValue() { - return (Comparable) _dataSourceMetadata.getMinValue(); + Comparable minValue = (Comparable) _dataSourceMetadata.getMinValue(); + if (minValue != null) { + return minValue; + } + computeMinMaxIfNeeded(); + return _computedMinValue; } @Override public Comparable getMaxValue() { - return (Comparable) _dataSourceMetadata.getMaxValue(); + Comparable maxValue = (Comparable) _dataSourceMetadata.getMaxValue(); + if (maxValue != null) { + return maxValue; + } + computeMinMaxIfNeeded(); + return _computedMaxValue; + } + + /// Computes min/max by scanning the sealed forward index once, caching the result. Only invoked when the mutable + /// segment reports null min/max, which happens for ingestion-aggregated metric columns: their values mutate in + /// place during consumption, so `MutableSegmentImpl` deliberately skips min/max tracking for them. Without a value + /// domain the BitSliced range index creator cannot subtract the min for INT/LONG columns, so we recover it here + /// (the scan mirrors the one {@link #isSorted()} already performs at seal time). + /// + /// Scoped to single-value INT/LONG columns because those are the only types whose BitSliced range index reads + /// min/max: FLOAT/DOUBLE use the full floating-point ordinal domain, and other stored types do not support the + /// BitSliced range index. For every other case min/max remain null (unchanged behavior), so aggregated + /// FLOAT/DOUBLE and sketch columns are never scanned here. + /// + /// Not thread-safe: like the rest of this class it is only exercised on the single-threaded segment-seal path. + private void computeMinMaxIfNeeded() { + if (_minMaxComputed) { + return; + } + int numDocs = _dataSourceMetadata.getNumDocs(); + if (isSingleValue() && numDocs > 0) { + switch (getStoredType()) { + case INT: { + int min = _forwardIndex.getInt(0); + int max = min; + for (int i = 1; i < numDocs; i++) { + int curr = _forwardIndex.getInt(i); + min = Math.min(min, curr); + max = Math.max(max, curr); + } + _computedMinValue = min; + _computedMaxValue = max; + break; + } + case LONG: { + long min = _forwardIndex.getLong(0); + long max = min; + for (int i = 1; i < numDocs; i++) { + long curr = _forwardIndex.getLong(i); + min = Math.min(min, curr); + max = Math.max(max, curr); + } + _computedMinValue = min; + _computedMaxValue = max; + break; + } + default: + // Other stored types either do not need min/max for their range index (FLOAT/DOUBLE) or do not support a + // BitSliced range index at all: leave min/max null (unchanged behavior). + break; + } + } + // Set last so a re-entrant call cannot observe the flag as computed while the values are still being populated. + _minMaxComputed = true; } @Nullable diff --git a/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/segment/index/loader/invertedindex/RangeIndexHandler.java b/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/segment/index/loader/invertedindex/RangeIndexHandler.java index 239317ef835a..b9979483cf5a 100644 --- a/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/segment/index/loader/invertedindex/RangeIndexHandler.java +++ b/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/segment/index/loader/invertedindex/RangeIndexHandler.java @@ -24,6 +24,7 @@ import java.util.Map; import java.util.Set; import org.apache.commons.io.FileUtils; +import org.apache.pinot.segment.local.segment.creator.impl.inv.BitSlicedRangeIndexCreator; import org.apache.pinot.segment.local.segment.index.dictionary.DictionaryIndexType; import org.apache.pinot.segment.local.segment.index.loader.BaseIndexHandler; import org.apache.pinot.segment.local.segment.index.loader.IndexLoadingConfig; @@ -44,6 +45,7 @@ import org.apache.pinot.segment.spi.memory.PinotDataBuffer; import org.apache.pinot.segment.spi.store.SegmentDirectory; import org.apache.pinot.spi.config.table.TableConfig; +import org.apache.pinot.spi.data.FieldSpec.DataType; import org.apache.pinot.spi.data.Schema; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -226,7 +228,8 @@ private void handleNonDictionaryBasedColumn(SegmentDirectory.Writer segmentWrite try (ForwardIndexReader forwardIndexReader = readerFactory.createIndexReader(segmentWriter, _fieldIndexConfigs.get(columnMetadata.getColumnName()), columnMetadata); ForwardIndexReaderContext readerContext = forwardIndexReader.createContext(); - CombinedInvertedIndexCreator rangeIndexCreator = newRangeIndexCreator(columnMetadata)) { + CombinedInvertedIndexCreator rangeIndexCreator = + newRangeIndexCreator(columnMetadata, forwardIndexReader, readerContext, numDocs)) { if (columnMetadata.isSingleValue()) { // Single-value column. switch (columnMetadata.getDataType().getStoredType()) { @@ -301,4 +304,67 @@ private CombinedInvertedIndexCreator newRangeIndexCreator(ColumnMetadata columnM .getConfig(StandardIndexes.range()); return StandardIndexes.range().createIndexCreator(context, config); } + + /// Variant for the non-dictionary path. The BitSliced (v2) range index subtracts the column min for INT/LONG + /// columns, so it needs the value domain up front. Ingestion-aggregated no-dictionary columns committed before + /// min/max recovery report null min/max in their metadata, so recompute it here with a single scan of the forward + /// index before building the creator. This is a deliberate extra pass over the column: {@code RangeBitmap.appender} + /// requires the max at construction, so it cannot be folded into the add-loop that follows. + /// + /// Note: the recovered domain is written into the range index header (which the reader uses for the subtract-min), + /// but it is not written back into the column metadata. For such legacy segments the reader therefore still reads a + /// null metadata max and falls back to {@code Long.MAX_VALUE}; results stay correct (the RangeBitmap domain is + /// self-contained) but segment-level max pruning is weaker until the segment is rebuilt. Segments sealed with the + /// recovery in {@code MutableNoDictColumnStatistics} carry proper metadata min/max and do not hit this path. + private CombinedInvertedIndexCreator newRangeIndexCreator(ColumnMetadata columnMetadata, + ForwardIndexReader forwardIndexReader, ForwardIndexReaderContext readerContext, int numDocs) + throws Exception { + File indexDir = _segmentDirectory.getSegmentMetadata().getIndexDir(); + IndexCreationContext.Builder builder = new IndexCreationContext.Builder(indexDir, _tableConfig, columnMetadata); + RangeIndexConfig config = _fieldIndexConfigs.get(columnMetadata.getColumnName()) + .getConfig(StandardIndexes.range()); + if (config.getVersion() == BitSlicedRangeIndexCreator.VERSION && columnMetadata.isSingleValue() + && (columnMetadata.getMinValue() == null || columnMetadata.getMaxValue() == null)) { + Comparable[] minMax = computeRawMinMax(forwardIndexReader, readerContext, + columnMetadata.getDataType().getStoredType(), numDocs); + if (minMax != null) { + builder.withMinValue(minMax[0]).withMaxValue(minMax[1]); + } + } + return StandardIndexes.range().createIndexCreator(builder.build(), config); + } + + /// Computes {@code [min, max]} for a single-value INT/LONG no-dictionary column by scanning the forward index. + /// Returns {@code null} for stored types whose BitSliced range index does not read min/max (FLOAT/DOUBLE use the + /// full floating-point ordinal domain) or that do not support it, and for empty columns. + private static Comparable[] computeRawMinMax(ForwardIndexReader forwardIndexReader, + ForwardIndexReaderContext readerContext, DataType storedType, int numDocs) { + if (numDocs == 0) { + return null; + } + switch (storedType) { + case INT: { + int min = forwardIndexReader.getInt(0, readerContext); + int max = min; + for (int i = 1; i < numDocs; i++) { + int curr = forwardIndexReader.getInt(i, readerContext); + min = Math.min(min, curr); + max = Math.max(max, curr); + } + return new Comparable[]{min, max}; + } + case LONG: { + long min = forwardIndexReader.getLong(0, readerContext); + long max = min; + for (int i = 1; i < numDocs; i++) { + long curr = forwardIndexReader.getLong(i, readerContext); + min = Math.min(min, curr); + max = Math.max(max, curr); + } + return new Comparable[]{min, max}; + } + default: + return null; + } + } } diff --git a/pinot-segment-local/src/test/java/org/apache/pinot/segment/local/realtime/converter/RealtimeSegmentConverterTest.java b/pinot-segment-local/src/test/java/org/apache/pinot/segment/local/realtime/converter/RealtimeSegmentConverterTest.java index ca466549b47b..a22836e99de4 100644 --- a/pinot-segment-local/src/test/java/org/apache/pinot/segment/local/realtime/converter/RealtimeSegmentConverterTest.java +++ b/pinot-segment-local/src/test/java/org/apache/pinot/segment/local/realtime/converter/RealtimeSegmentConverterTest.java @@ -66,6 +66,8 @@ import org.apache.pinot.spi.config.table.SegmentZKPropsConfig; import org.apache.pinot.spi.config.table.TableConfig; import org.apache.pinot.spi.config.table.TableType; +import org.apache.pinot.spi.config.table.ingestion.AggregationConfig; +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; @@ -1557,6 +1559,88 @@ private static int compareValues(DataType storedType, Object v1, Object v2) { return storedType.compare(v1, v2); } + /// Ingestion-aggregated metric columns are forced to be no-dictionary and skip min/max tracking during consumption + /// (their values mutate in place). A BitSliced (version 2) range index on such a column previously failed to build + /// because the creator had no value domain. This test verifies the full seal path: after conversion, the aggregated + /// no-dictionary metric column has recovered min/max in its column metadata and carries a range index. + @Test + public void testRangeIndexOnIngestionAggregatedNoDictColumn() + throws Exception { + File tmpDir = new File(TMP_DIR, "tmp_" + System.currentTimeMillis()); + IngestionConfig ingestionConfig = new IngestionConfig(); + List aggregationConfigs = + List.of(new AggregationConfig(LONG_COLUMN4, "SUM(" + LONG_COLUMN4 + ")")); + ingestionConfig.setAggregationConfigs(aggregationConfigs); + TableConfig tableConfig = new TableConfigBuilder(TableType.REALTIME) + .setTableName("testTable") + .setTimeColumnName(DATE_TIME_COLUMN) + .setIngestionConfig(ingestionConfig) + .setNoDictionaryColumns(List.of(LONG_COLUMN4)) + .setRangeIndexColumns(List.of(LONG_COLUMN4)) + .setColumnMajorSegmentBuilderEnabled(false) + .build(); + Schema schema = new Schema.SchemaBuilder() + .setSchemaName("testTable") + .addSingleValueDimension(STRING_COLUMN1, DataType.STRING) + .addMetric(LONG_COLUMN4, DataType.LONG) + .addDateTime(DATE_TIME_COLUMN, DataType.LONG, "1:MILLISECONDS:EPOCH", "1:MILLISECONDS") + .build(); + + String tableNameWithType = tableConfig.getTableName(); + String segmentName = "testTable__0__0__123456"; + + RealtimeSegmentConfig realtimeSegmentConfig = new RealtimeSegmentConfig.Builder() + .setTableNameWithType(tableNameWithType) + .setSegmentName(segmentName) + .setStreamName(tableNameWithType) + .setSchema(schema) + .setTimeColumnName(DATE_TIME_COLUMN) + .setCapacity(1000) + .setIngestionAggregationConfigs(aggregationConfigs) + .setIndex(Set.of(LONG_COLUMN4), StandardIndexes.dictionary(), DictionaryIndexConfig.DISABLED) + .setIndex(Set.of(LONG_COLUMN4), StandardIndexes.forward(), + ForwardIndexConfig.getDefault(FieldConfig.EncodingType.RAW)) + .setSegmentZKMetadata(getSegmentZKMetadata(segmentName)) + .setOffHeap(true) + .setMemoryManager(new DirectMemoryManager(segmentName)) + .setStatsHistory(RealtimeSegmentStatsHistory.deserializeFrom(new File(tmpDir, "stats"))) + .setConsumerDir(new File(tmpDir, "consumerDir").getAbsolutePath()) + .build(); + + MutableSegmentImpl mutableSegmentImpl = new MutableSegmentImpl(realtimeSegmentConfig, null); + try { + // Each row carries a distinct dimension/time value, so no rows collapse and the aggregated metric retains its + // per-row values 64..73 (SUM over a single-row group). The mutable segment reports null min/max for it. + List rows = generateTestData(); + for (GenericRow row : rows) { + mutableSegmentImpl.index(row, null); + } + + File outputDir = new File(tmpDir, "outputDir"); + RealtimeSegmentConverter converter = + new RealtimeSegmentConverter(mutableSegmentImpl, new SegmentZKPropsConfig(), outputDir.getAbsolutePath(), + schema, tableNameWithType, tableConfig, segmentName, false); + converter.build(SegmentVersion.v3); + + File indexDir = new File(outputDir, segmentName); + SegmentMetadataImpl segmentMetadata = new SegmentMetadataImpl(indexDir); + ColumnMetadata metricMetadata = segmentMetadata.getColumnMetadataFor(LONG_COLUMN4); + assertFalse(metricMetadata.hasDictionary(), "Aggregated metric column must stay no-dictionary"); + // Min/max are recovered by scanning the sealed forward index despite the mutable segment not tracking them. + assertEquals(metricMetadata.getMinValue(), 64L); + assertEquals(metricMetadata.getMaxValue(), 73L); + + // The BitSliced range index must be present on the aggregated no-dictionary column. + try (SegmentLocalFSDirectory segmentDir = new SegmentLocalFSDirectory(indexDir, segmentMetadata, ReadMode.mmap); + SegmentDirectory.Reader segmentReader = segmentDir.createReader()) { + assertTrue(segmentReader.hasIndexFor(LONG_COLUMN4, StandardIndexes.range()), + "Range index should be built on the aggregated no-dictionary column"); + } + } finally { + mutableSegmentImpl.destroy(); + } + } + private List generateTestData() { LinkedList rows = new LinkedList<>(); diff --git a/pinot-segment-local/src/test/java/org/apache/pinot/segment/local/realtime/converter/stats/MutableNoDictColumnStatisticsTest.java b/pinot-segment-local/src/test/java/org/apache/pinot/segment/local/realtime/converter/stats/MutableNoDictColumnStatisticsTest.java index 276e5ec203f3..322ed6ae8ce1 100644 --- a/pinot-segment-local/src/test/java/org/apache/pinot/segment/local/realtime/converter/stats/MutableNoDictColumnStatisticsTest.java +++ b/pinot-segment-local/src/test/java/org/apache/pinot/segment/local/realtime/converter/stats/MutableNoDictColumnStatisticsTest.java @@ -661,6 +661,83 @@ public void testBytesMv() { assertEquals(stats.getMaxRowLengthInBytes(), 5); } + // ======== Computed min/max when the mutable segment does not track it (ingestion-aggregated columns) ======== + + @DataProvider(name = "computedMinMaxTypes") + public Object[][] computedMinMaxTypes() { + // {type, unsorted forward-index values, expectedMin, expectedMax}. Only INT/LONG are recovered — those are the + // only types whose BitSliced range index reads the value domain. + return new Object[][]{ + {DataType.INT, new Comparable[]{20, 10, 30}, 10, 30}, + {DataType.LONG, new Comparable[]{200L, 100L, 300L}, 100L, 300L} + }; + } + + @Test(dataProvider = "computedMinMaxTypes") + public void testComputesMinMaxWhenMetadataNull(DataType type, Comparable[] values, Comparable expectedMin, + Comparable expectedMax) { + // Ingestion-aggregated no-dictionary metric columns skip min/max tracking, so the mutable metadata reports null. + // The statistics must recover the value domain by scanning the sealed forward index. + int numDocs = values.length; + FieldSpec fieldSpec = new DimensionFieldSpec("col", type, true); + + DataSourceMetadata metadata = mockMetadata(fieldSpec, numDocs); + when(metadata.getMinValue()).thenReturn(null); + when(metadata.getMaxValue()).thenReturn(null); + + MutableForwardIndex forwardIndex = mock(MutableForwardIndex.class); + when(forwardIndex.isSingleValue()).thenReturn(true); + stubForwardIndexReads(forwardIndex, type, values); + + MutableNoDictColumnStatistics stats = + new MutableNoDictColumnStatistics(mockNoDictDataSource(metadata, forwardIndex), null, false); + + assertEquals(stats.getMinValue(), expectedMin); + assertEquals(stats.getMaxValue(), expectedMax); + } + + @DataProvider(name = "unrecoveredMinMaxTypes") + public Object[][] unrecoveredMinMaxTypes() { + return new Object[][]{{DataType.FLOAT}, {DataType.DOUBLE}}; + } + + @Test(dataProvider = "unrecoveredMinMaxTypes") + public void testMinMaxNotRecoveredForFloatingPointWhenMetadataNull(DataType type) { + // FLOAT/DOUBLE range indexes use the full floating-point ordinal domain and never read min/max, so the scan is + // intentionally skipped for them: min/max stay null (unchanged behavior) and no forward-index read is expected. + FieldSpec fieldSpec = new DimensionFieldSpec("col", type, true); + DataSourceMetadata metadata = mockMetadata(fieldSpec, 3); + when(metadata.getMinValue()).thenReturn(null); + when(metadata.getMaxValue()).thenReturn(null); + + MutableForwardIndex forwardIndex = mock(MutableForwardIndex.class); + when(forwardIndex.isSingleValue()).thenReturn(true); + + MutableNoDictColumnStatistics stats = + new MutableNoDictColumnStatistics(mockNoDictDataSource(metadata, forwardIndex), null, false); + + assertNull(stats.getMinValue()); + assertNull(stats.getMaxValue()); + } + + @Test + public void testMinMaxNullForMultiValueWhenMetadataNull() { + // Multi-value columns have no scalar min/max; when metadata reports null there is nothing to recover. + FieldSpec fieldSpec = new DimensionFieldSpec("col", DataType.INT, false); + DataSourceMetadata metadata = mockMetadata(fieldSpec, 3); + when(metadata.getMinValue()).thenReturn(null); + when(metadata.getMaxValue()).thenReturn(null); + + MutableForwardIndex forwardIndex = mock(MutableForwardIndex.class); + when(forwardIndex.isSingleValue()).thenReturn(false); + + MutableNoDictColumnStatistics stats = + new MutableNoDictColumnStatistics(mockNoDictDataSource(metadata, forwardIndex), null, false); + + assertNull(stats.getMinValue()); + assertNull(stats.getMaxValue()); + } + // ======== Helpers ======== private static DataSourceMetadata mockMetadata(FieldSpec fieldSpec, int numDocs) { diff --git a/pinot-segment-local/src/test/java/org/apache/pinot/segment/local/utils/IndexCombinationValidationTest.java b/pinot-segment-local/src/test/java/org/apache/pinot/segment/local/utils/IndexCombinationValidationTest.java index b2fb2c5b5cfc..80d18b6d6157 100644 --- a/pinot-segment-local/src/test/java/org/apache/pinot/segment/local/utils/IndexCombinationValidationTest.java +++ b/pinot-segment-local/src/test/java/org/apache/pinot/segment/local/utils/IndexCombinationValidationTest.java @@ -20,13 +20,17 @@ import com.fasterxml.jackson.databind.node.ObjectNode; import java.util.Arrays; +import java.util.HashMap; import java.util.List; +import java.util.Map; import org.apache.pinot.spi.config.table.FieldConfig; import org.apache.pinot.spi.config.table.FieldConfig.CompressionCodec; import org.apache.pinot.spi.config.table.FieldConfig.EncodingType; import org.apache.pinot.spi.config.table.FieldConfig.IndexType; import org.apache.pinot.spi.config.table.TableConfig; import org.apache.pinot.spi.config.table.TableType; +import org.apache.pinot.spi.config.table.ingestion.AggregationConfig; +import org.apache.pinot.spi.config.table.ingestion.IngestionConfig; import org.apache.pinot.spi.data.FieldSpec.DataType; import org.apache.pinot.spi.data.Schema; import org.apache.pinot.spi.utils.JsonUtils; @@ -660,4 +664,117 @@ public void testErrorMessageNamesFstIndex() { assertTrue(msg.contains("without dictionary"), "Error should explain the problem"); } } + + // ============================================================ + // 13. BitSliced range index (version 2) on ingestion-aggregated columns + // Ingestion-time metrics aggregation forces its aggregated metric columns to be no-dictionary and does not + // track their min/max during consumption. The value domain is recovered by scanning the sealed forward index + // at segment build time (see MutableNoDictColumnStatistics), so the combination is supported and must pass + // table-config validation. + // ============================================================ + + private static final String AGG_METRIC_COL = "m1"; // LONG metric, aggregation destination + private static final String AGG_SOURCE_COL = "s1"; // INT dimension, aggregation source + private static final String TIME_COL = "ts"; // realtime tables require a time column + + /// Schema with a metric (aggregation destination), a dimension source column, and a time column. + private static Schema aggregationSchema() { + return new Schema.SchemaBuilder() + .setSchemaName(TABLE_NAME) + .addSingleValueDimension(AGG_SOURCE_COL, DataType.INT) + .addMetric(AGG_METRIC_COL, DataType.LONG) + .addDateTime(TIME_COL, DataType.TIMESTAMP, "1:MILLISECONDS:EPOCH", "1:MILLISECONDS") + .build(); + } + + private static Map streamConfigs() { + Map streamConfigs = new HashMap<>(); + streamConfigs.put("streamType", "kafka"); + streamConfigs.put("stream.kafka.topic.name", "test"); + streamConfigs.put("stream.kafka.decoder.class.name", + "org.apache.pinot.plugin.stream.kafka.KafkaJSONMessageDecoder"); + return streamConfigs; + } + + /// REALTIME table builder wired with a stream config and time column so full validation can run. + private static TableConfigBuilder realtimeBuilder() { + return new TableConfigBuilder(TableType.REALTIME).setTableName(TABLE_NAME) + .setTimeColumnName(TIME_COL) + .setStreamConfigs(streamConfigs()); + } + + /// IngestionConfig that aggregates `SUM(s1)` into the no-dictionary metric column `m1`. + private static IngestionConfig sumAggregationIngestionConfig() { + IngestionConfig ingestionConfig = new IngestionConfig(); + ingestionConfig.setAggregationConfigs( + List.of(new AggregationConfig(AGG_METRIC_COL, "SUM(" + AGG_SOURCE_COL + ")"))); + return ingestionConfig; + } + + private static void assertValid(TableConfig tableConfig, Schema schema) { + try { + TableConfigUtils.validate(tableConfig, schema); + } catch (Exception e) { + fail("Expected validation to pass but got: " + e.getMessage(), e); + } + } + + @Test + public void testRangeIndexOnIngestionAggregatedColumnPasses() { + // aggregationConfigs (SUM into a no-dictionary metric column) + BitSliced range index on that metric column. + // Min/max are recovered at segment build time, so this combination is accepted. + TableConfig tc = realtimeBuilder() + .setIngestionConfig(sumAggregationIngestionConfig()) + .setNoDictionaryColumns(List.of(AGG_METRIC_COL)) + .setRangeIndexColumns(List.of(AGG_METRIC_COL)) + .build(); + assertValid(tc, aggregationSchema()); + } + + @Test + public void testRangeIndexOnAggregateMetricsColumnPasses() { + // Legacy aggregateMetrics flag (implicitly SUMs every metric) + BitSliced range index on the no-dictionary + // metric column. Also accepted for the same reason. + TableConfig tc = realtimeBuilder() + .setAggregateMetrics(true) + .setNoDictionaryColumns(List.of(AGG_METRIC_COL)) + .setRangeIndexColumns(List.of(AGG_METRIC_COL)) + .build(); + assertValid(tc, aggregationSchema()); + } + + @Test + public void testRangeIndexV1OnIngestionAggregatedColumnPasses() { + // Version 1 (legacy) range index does not read min/max, so it is safe on an aggregated no-dictionary column. + TableConfig tc = realtimeBuilder() + .setIngestionConfig(sumAggregationIngestionConfig()) + .setNoDictionaryColumns(List.of(AGG_METRIC_COL)) + .setRangeIndexColumns(List.of(AGG_METRIC_COL)) + .build(); + tc.getIndexingConfig().setRangeIndexVersion(1); + assertValid(tc, aggregationSchema()); + } + + @Test + public void testRangeIndexOnNonAggregatedNoDictNumericColumnPasses() { + // A plain no-dictionary numeric column (not an aggregation destination) still supports the range index. + TableConfig tc = realtimeBuilder() + .setIngestionConfig(sumAggregationIngestionConfig()) + .setNoDictionaryColumns(List.of(AGG_METRIC_COL, AGG_SOURCE_COL)) + .setRangeIndexColumns(List.of(AGG_SOURCE_COL)) + .build(); + assertValid(tc, aggregationSchema()); + } + + @Test + public void testRangeIndexOnAggregatedColumnOfflineTablePasses() { + // Ingestion aggregation is inert for OFFLINE tables (min/max are computed normally), so the range index is + // allowed there - the validation is scoped to REALTIME to avoid over-rejecting inert offline configs. + TableConfig tc = new TableConfigBuilder(TableType.OFFLINE).setTableName(TABLE_NAME) + .setIngestionConfig(sumAggregationIngestionConfig()) + .setNoDictionaryColumns(List.of(AGG_METRIC_COL)) + .setRangeIndexColumns(List.of(AGG_METRIC_COL)) + .build(); + assertValid(tc, aggregationSchema()); + } } diff --git a/pinot-segment-spi/src/main/java/org/apache/pinot/segment/spi/creator/IndexCreationContext.java b/pinot-segment-spi/src/main/java/org/apache/pinot/segment/spi/creator/IndexCreationContext.java index 17ca2edbf1a4..5059bde68413 100644 --- a/pinot-segment-spi/src/main/java/org/apache/pinot/segment/spi/creator/IndexCreationContext.java +++ b/pinot-segment-spi/src/main/java/org/apache/pinot/segment/spi/creator/IndexCreationContext.java @@ -127,6 +127,12 @@ final class Builder { private int _maxNumberOfMultiValues; private int _maxRowLengthInBytes; private boolean _hasDictionary; + // Overrides for min/max; null means "delegate to the source ColumnShape". Set only when the source reports a + // null value domain that the caller has recomputed (see withMinValue / withMaxValue). + @Nullable + private Comparable _minValue; + @Nullable + private Comparable _maxValue; // Build-time toggles. private boolean _onHeap; @@ -226,6 +232,20 @@ public Builder withDictionary(boolean hasDictionary) { return this; } + /// Overrides the min value that would otherwise be sourced from the [ColumnShape]. Used when the source reports a + /// null min/max (e.g. ingestion-aggregated no-dictionary columns on the index-handler path) and the caller has + /// recomputed it. A null argument leaves delegation to the source unchanged. + public Builder withMinValue(Comparable minValue) { + _minValue = minValue; + return this; + } + + /// Overrides the max value derived from the source [ColumnShape]. See [#withMinValue]. + public Builder withMaxValue(Comparable maxValue) { + _maxValue = maxValue; + return this; + } + // Build-time toggle setters. public Builder withOnHeap(boolean onHeap) { @@ -299,6 +319,10 @@ final class Common implements IndexCreationContext { private final int _maxNumberOfMultiValues; private final int _maxRowLengthInBytes; private final boolean _hasDictionary; + @Nullable + private final Comparable _minValue; + @Nullable + private final Comparable _maxValue; // Build-time toggles. private final boolean _onHeap; @@ -328,6 +352,8 @@ private Common(Builder builder) { _maxNumberOfMultiValues = builder._maxNumberOfMultiValues; _maxRowLengthInBytes = builder._maxRowLengthInBytes; _hasDictionary = builder._hasDictionary; + _minValue = builder._minValue; + _maxValue = builder._maxValue; _onHeap = builder._onHeap; _optimizeDictionary = builder._optimizedDictionary; _textCommitOnClose = builder._textCommitOnClose; @@ -388,13 +414,14 @@ public boolean isSorted() { @Override @Nullable public Comparable getMinValue() { - return _columnShape.getMinValue(); + // Delegate to the source ColumnShape unless a caller supplied an explicit override. + return _minValue != null ? _minValue : _columnShape.getMinValue(); } @Override @Nullable public Comparable getMaxValue() { - return _columnShape.getMaxValue(); + return _maxValue != null ? _maxValue : _columnShape.getMaxValue(); } @Override