This is an automated email from the ASF dual-hosted git repository.
raghavyadav01 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/pinot.git
The following commit(s) were added to refs/heads/master by this push:
new 657d7805e7a Support range index on ingestion-aggregated no-dictionary
columns (#19111)
657d7805e7a is described below
commit 657d7805e7a3e9920475ef927440ba6721545bf7
Author: RAGHVENDRA KUMAR YADAV <[email protected]>
AuthorDate: Wed Oct 7 11:33:47 2026 -0700
Support range index on ingestion-aggregated no-dictionary columns (#19111)
* Support BitSliced range index on ingestion-aggregated no-dict columns
Ingestion-time metrics aggregation forces its aggregated metric columns
to be no-dictionary and skips min/max tracking during consumption (the
values mutate in place). The BitSliced range index (version 2) creator
subtracts the column min for INT/LONG columns, so with a null value
domain segment commit failed with an NPE.
Instead of rejecting the combination, recover the value domain:
- MutableNoDictColumnStatistics computes min/max with a single scan of
the sealed forward index when the mutable segment reports null,
scoped to single-value INT/LONG (the only types whose BitSliced index
reads min/max; FLOAT/DOUBLE use the full floating-point ordinal
domain). The recovered values also flow into the committed segment
metadata, improving pruning.
- RangeIndexHandler recomputes min/max from the forward index on the
index-reload path for segments committed before this change, via a
new min/max override on IndexCreationContext.Builder.
Adds unit coverage for min/max recovery and an end-to-end seal-path
test that builds and verifies a range index on an aggregated
no-dictionary column, and flips the validation tests to assert the
combination is now accepted.
* Recover min/max in the same pass that already checks sortedness
Recovering an untracked value domain was a second full scan of the column:
computeSorted() calls getMinValue(), which now scans, and then scanned again
itself. That regressed the single-pass guarantee two existing tests assert
(testIsSortedScansOnce, testUntrackedMinMaxFallsBackToScan), and charged the
extra pass to every ingestion-aggregated INT/LONG column at commit, whether
or
not a range index was configured -- BaseSegmentCreator reads min/max for
every
column when writing segment metadata.
Fold the two together: the recovery scan walks docs in the order
computeSorted()
uses (honouring sortedDocIds) and records non-decreasing order alongside
min and
max; computeSorted() reuses that answer instead of rescanning. A segment
commit
reads such a column exactly once, as before.
* Cover the RangeIndexHandler null min/max path with a real segment
The handler-side recovery had no test: the existing RangeIndexHandlerTest is
mock-based and never builds an index, leaving that path at 22% coverage.
Build a real segment over unsorted, partly negative INT and LONG
no-dictionary
columns, strip min/max from its metadata -- which is what a segment
committed
with untracked min/max looks like on disk -- then run the handler and query
the
resulting BitSliced index. Reverting RangeIndexHandler to its pre-change
form
fails both tests with the NPE this PR fixes.
The second test pins the open-ended case: the reader still takes the column
max
from metadata and so falls back to Long.MAX_VALUE, but the RangeBitmap
domain is
self-contained, so matches stay correct.
* Correct the note on what fills column metadata after the handler scan
The note claimed metadata min/max stay null until the segment is rebuilt.
Under
the default columnMinMaxValueGeneratorMode they do not:
ColumnMinMaxValueGenerator
runs later in the same SegmentPreProcessor pass and populates them. The
weaker
max-pruning caveat applies only when that mode is NONE. Also record that the
default path consequently reads the column twice per preprocess.
* Compute min/max bounds in the MutableNoDictColumnStatistics constructor
Keep plain _minValue/_maxValue fields and the existing _sorted cache:
recover
missing bounds for single-value INT/LONG columns once at construction and
record sortedness from the same scan, so the getters are field reads.
* Clarify when range-index column metadata min/max may stay absent
* Skip min/max recovery scan in CompactedNoDictColumnStatistics
The compacted subclass computes its own bounds over valid docs only, and its
sortedDocIds covers just those docs, so the base-class constructor scan
would
index past the array. Let subclasses opt out via a protected constructor.
---
.../stats/CompactedNoDictColumnStatistics.java | 2 +-
.../stats/MutableNoDictColumnStatistics.java | 69 +++++-
.../loader/invertedindex/RangeIndexHandler.java | 69 +++++-
.../converter/RealtimeSegmentConverterTest.java | 84 +++++++
.../stats/MutableNoDictColumnStatisticsTest.java | 147 +++++++++++++
.../RangeIndexHandlerNullMinMaxTest.java | 243 +++++++++++++++++++++
.../utils/IndexCombinationValidationTest.java | 117 ++++++++++
.../segment/spi/creator/IndexCreationContext.java | 31 ++-
8 files changed, 756 insertions(+), 6 deletions(-)
diff --git
a/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/realtime/converter/stats/CompactedNoDictColumnStatistics.java
b/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/realtime/converter/stats/CompactedNoDictColumnStatistics.java
index 6d01827bfae..ff96b76f58b 100644
---
a/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/realtime/converter/stats/CompactedNoDictColumnStatistics.java
+++
b/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/realtime/converter/stats/CompactedNoDictColumnStatistics.java
@@ -47,7 +47,7 @@ public class CompactedNoDictColumnStatistics extends
MutableNoDictColumnStatisti
public CompactedNoDictColumnStatistics(DataSource dataSource, @Nullable
int[] sortedDocIds, boolean isSortedColumn,
RoaringBitmap validDocIds) {
- super(dataSource, sortedDocIds, isSortedColumn);
+ super(dataSource, sortedDocIds, isSortedColumn, false);
Preconditions.checkState(!validDocIds.isEmpty(), "Use
EmptyColumnStatistics for empty column: %s",
_fieldSpec.getName());
_totalDocs = validDocIds.getCardinality();
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 7896c13f96d..8516d230806 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,11 @@ public class MutableNoDictColumnStatistics implements
ColumnStatistics, CLPStats
protected final boolean _isSortedColumn;
protected final MutableForwardIndex _forwardIndex;
+ @Nullable
+ private final Comparable<?> _minValue;
+ @Nullable
+ private final Comparable<?> _maxValue;
+
// Lazily computed because it may require a full scan of the forward index,
and it is queried multiple times per
// column during segment creation. Left unsynchronized: an instance
describes a single column and is reached only
// through the per-column stats map, so it is confined to whichever thread
creates that column. Even if that ever
@@ -53,6 +58,13 @@ public class MutableNoDictColumnStatistics implements
ColumnStatistics, CLPStats
private Boolean _sorted;
public MutableNoDictColumnStatistics(DataSource dataSource, @Nullable int[]
sortedDocIds, boolean isSortedColumn) {
+ this(dataSource, sortedDocIds, isSortedColumn, true);
+ }
+
+ /// @param recoverMinMax whether to scan for min/max missing from the
metadata. Subclasses that compute their own
+ /// bounds over a subset of docs pass false to skip the
scan.
+ protected MutableNoDictColumnStatistics(DataSource dataSource, @Nullable
int[] sortedDocIds, boolean isSortedColumn,
+ boolean recoverMinMax) {
_dataSourceMetadata = dataSource.getDataSourceMetadata();
_fieldSpec = _dataSourceMetadata.getFieldSpec();
Preconditions.checkState(_dataSourceMetadata.getNumDocs() > 0,
@@ -62,6 +74,55 @@ public class MutableNoDictColumnStatistics implements
ColumnStatistics, CLPStats
_forwardIndex = (MutableForwardIndex) dataSource.getForwardIndex();
Preconditions.checkState(_forwardIndex != null, "Failed to find forward
index for column: %s",
_fieldSpec.getName());
+
+ // Ingestion-aggregated metric columns mutate in place during consumption,
so the mutable segment does not track
+ // min/max for them. Recover the bounds for single-value INT/LONG columns,
the only types whose BitSliced range
+ // index needs them, and record sortedness in the same scan. Other types
keep null bounds (unchanged behavior).
+ Comparable<?> minValue = (Comparable<?>) _dataSourceMetadata.getMinValue();
+ Comparable<?> maxValue = (Comparable<?>) _dataSourceMetadata.getMaxValue();
+ if (recoverMinMax && (minValue == null || maxValue == null) &&
isSingleValue()) {
+ int numDocs = _dataSourceMetadata.getNumDocs();
+ switch (getStoredType()) {
+ case INT: {
+ int min = _forwardIndex.getInt(docId(0));
+ int max = min;
+ int prev = min;
+ boolean sorted = true;
+ for (int i = 1; i < numDocs; i++) {
+ int curr = _forwardIndex.getInt(docId(i));
+ min = Math.min(min, curr);
+ max = Math.max(max, curr);
+ sorted &= curr >= prev;
+ prev = curr;
+ }
+ minValue = min;
+ maxValue = max;
+ _sorted = _isSortedColumn || sorted;
+ break;
+ }
+ case LONG: {
+ long min = _forwardIndex.getLong(docId(0));
+ long max = min;
+ long prev = min;
+ boolean sorted = true;
+ for (int i = 1; i < numDocs; i++) {
+ long curr = _forwardIndex.getLong(docId(i));
+ min = Math.min(min, curr);
+ max = Math.max(max, curr);
+ sorted &= curr >= prev;
+ prev = curr;
+ }
+ minValue = min;
+ maxValue = max;
+ _sorted = _isSortedColumn || sorted;
+ break;
+ }
+ default:
+ break;
+ }
+ }
+ _minValue = minValue;
+ _maxValue = maxValue;
}
@Override
@@ -76,12 +137,16 @@ public class MutableNoDictColumnStatistics implements
ColumnStatistics, CLPStats
@Override
public Comparable<?> getMinValue() {
- return (Comparable<?>) _dataSourceMetadata.getMinValue();
+ return _minValue;
}
@Override
public Comparable<?> getMaxValue() {
- return (Comparable<?>) _dataSourceMetadata.getMaxValue();
+ return _maxValue;
+ }
+
+ private int docId(int index) {
+ return _sortedDocIds != null ? _sortedDocIds[index] : index;
}
@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 239317ef835..a4c5f042553 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.HashSet;
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.index.reader.ForwardIndexReaderContext;
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 @@ public class RangeIndexHandler extends BaseIndexHandler {
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,68 @@ public class RangeIndexHandler extends BaseIndexHandler {
.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 goes into the range index header, not the
column metadata. The metadata may stay
+ /// absent when the configured {@code columnMinMaxValueGeneratorMode}
excludes the column; the reader's conservative
+ /// fallback remains correct in that case.
+ ///
+ /// Segments sealed with the recovery in {@code
MutableNoDictColumnStatistics} carry proper metadata min/max and do
+ /// not reach the scan here.
+ 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 da96d42c904..56f5f374a0d 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.SegmentPartitionConfig;
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 @@ public class RealtimeSegmentConverterTest implements
PinotBuffersAfterMethodChec
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<AggregationConfig> 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<GenericRow> 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<GenericRow> generateTestData() {
LinkedList<GenericRow> 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 7131513310b..22ae84fb6b3 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
@@ -722,6 +722,153 @@ public class MutableNoDictColumnStatisticsTest {
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());
+ }
+
+ @Test
+ public void testComputedMinMaxAndIsSortedShareOneScan() {
+ // Recovering an untracked value domain must not cost a second pass: the
recovery scan walks the same docs in the
+ // same order computeSorted() would, so it answers both. A segment commit
still reads such a column exactly once.
+ int numDocs = 3;
+ FieldSpec fieldSpec = new DimensionFieldSpec("col", DataType.INT, true);
+ Comparable[] values = fixedWidthValues(DataType.INT);
+
+ DataSourceMetadata metadata = mockMetadata(fieldSpec, numDocs);
+
+ MutableForwardIndex forwardIndex = mock(MutableForwardIndex.class);
+ when(forwardIndex.isSingleValue()).thenReturn(true);
+ stubForwardIndexReads(forwardIndex, DataType.INT, values);
+
+ MutableNoDictColumnStatistics stats =
+ new MutableNoDictColumnStatistics(mockNoDictDataSource(metadata,
forwardIndex), null, false);
+
+ assertEquals(stats.getMinValue(), 10);
+ assertEquals(stats.getMaxValue(), 30);
+ assertTrue(stats.isSorted());
+ verify(forwardIndex, times(numDocs)).getInt(anyInt());
+ }
+
+ @Test
+ public void testComputedMinMaxReportsUnsortedInOneScan() {
+ // Same single pass, but the values are not non-decreasing: the recovered
domain is still correct and sortedness
+ // comes back false without a second scan.
+ Comparable[] values = new Comparable[]{20, 10, 30};
+ int numDocs = values.length;
+ FieldSpec fieldSpec = new DimensionFieldSpec("col", DataType.INT, true);
+
+ DataSourceMetadata metadata = mockMetadata(fieldSpec, numDocs);
+
+ MutableForwardIndex forwardIndex = mock(MutableForwardIndex.class);
+ when(forwardIndex.isSingleValue()).thenReturn(true);
+ stubForwardIndexReads(forwardIndex, DataType.INT, values);
+
+ MutableNoDictColumnStatistics stats =
+ new MutableNoDictColumnStatistics(mockNoDictDataSource(metadata,
forwardIndex), null, false);
+
+ assertFalse(stats.isSorted());
+ assertEquals(stats.getMinValue(), 10);
+ assertEquals(stats.getMaxValue(), 30);
+ verify(forwardIndex, times(numDocs)).getInt(anyInt());
+ }
+
+ @Test
+ public void testComputedMinMaxHonoursSortedDocIdOrder() {
+ // With an explicit sort order the recovery scan must walk docs in that
order, so sortedness is judged on the
+ // sorted view while min/max stay order-independent.
+ Comparable[] values = new Comparable[]{30, 10, 20};
+ int numDocs = values.length;
+ FieldSpec fieldSpec = new DimensionFieldSpec("col", DataType.INT, true);
+
+ DataSourceMetadata metadata = mockMetadata(fieldSpec, numDocs);
+
+ MutableForwardIndex forwardIndex = mock(MutableForwardIndex.class);
+ when(forwardIndex.isSingleValue()).thenReturn(true);
+ stubForwardIndexReads(forwardIndex, DataType.INT, values);
+
+ // Read in the order 10, 20, 30 -> sorted, even though the raw docId order
is not.
+ MutableNoDictColumnStatistics stats =
+ new MutableNoDictColumnStatistics(mockNoDictDataSource(metadata,
forwardIndex), new int[]{1, 2, 0}, false);
+
+ assertEquals(stats.getMinValue(), 10);
+ assertEquals(stats.getMaxValue(), 30);
+ assertTrue(stats.isSorted());
+ verify(forwardIndex, times(numDocs)).getInt(anyInt());
+ }
+
// ======== Helpers ========
private static DataSourceMetadata mockMetadata(FieldSpec fieldSpec, int
numDocs) {
diff --git
a/pinot-segment-local/src/test/java/org/apache/pinot/segment/local/segment/index/loader/invertedindex/RangeIndexHandlerNullMinMaxTest.java
b/pinot-segment-local/src/test/java/org/apache/pinot/segment/local/segment/index/loader/invertedindex/RangeIndexHandlerNullMinMaxTest.java
new file mode 100644
index 00000000000..4f64f530f16
--- /dev/null
+++
b/pinot-segment-local/src/test/java/org/apache/pinot/segment/local/segment/index/loader/invertedindex/RangeIndexHandlerNullMinMaxTest.java
@@ -0,0 +1,243 @@
+/**
+ * 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.index.loader.invertedindex;
+
+import java.io.File;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Map;
+import org.apache.commons.configuration2.PropertiesConfiguration;
+import org.apache.commons.io.FileUtils;
+import
org.apache.pinot.segment.local.segment.creator.impl.SegmentIndexCreationDriverImpl;
+import
org.apache.pinot.segment.local.segment.index.readers.BitSlicedRangeIndexReader;
+import org.apache.pinot.segment.local.segment.readers.GenericRowRecordReader;
+import org.apache.pinot.segment.local.segment.store.SegmentLocalFSDirectory;
+import org.apache.pinot.segment.spi.ColumnMetadata;
+import org.apache.pinot.segment.spi.V1Constants;
+import org.apache.pinot.segment.spi.creator.SegmentGeneratorConfig;
+import org.apache.pinot.segment.spi.index.FieldIndexConfigs;
+import org.apache.pinot.segment.spi.index.RangeIndexConfig;
+import org.apache.pinot.segment.spi.index.StandardIndexes;
+import org.apache.pinot.segment.spi.index.metadata.SegmentMetadataImpl;
+import org.apache.pinot.segment.spi.memory.PinotDataBuffer;
+import org.apache.pinot.segment.spi.store.SegmentDirectory;
+import org.apache.pinot.segment.spi.utils.SegmentMetadataUtils;
+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.data.readers.GenericRow;
+import org.apache.pinot.spi.utils.ReadMode;
+import org.apache.pinot.spi.utils.builder.TableConfigBuilder;
+import org.roaringbitmap.buffer.ImmutableRoaringBitmap;
+import org.testng.annotations.AfterClass;
+import org.testng.annotations.Test;
+
+import static org.testng.Assert.assertEquals;
+import static org.testng.Assert.assertNotNull;
+import static org.testng.Assert.assertNull;
+import static org.testng.Assert.assertTrue;
+
+
+/// Tests [RangeIndexHandler] building a BitSliced (v2) range index on a
no-dictionary column whose metadata carries
+/// no min/max value.
+///
+/// Ingestion-aggregated metric columns are forced no-dictionary and skip
min/max tracking while consuming, so
+/// segments committed before that domain is recovered at seal time report
null min/max. The BitSliced creator
+/// subtracts the column min for INT/LONG, so without recovery it dereferences
a null and the load fails. The handler
+/// recovers the domain by scanning the forward index; these tests pin that
the index is built and answers correctly.
+///
+/// The segment is built normally and then stripped of its min/max metadata,
which is what such a segment looks like
+/// on disk.
+public class RangeIndexHandlerNullMinMaxTest {
+ private static final File TEMP_DIR =
+ new File(FileUtils.getTempDirectory(),
RangeIndexHandlerNullMinMaxTest.class.getSimpleName());
+ private static final String SEGMENT_NAME = "testSegment";
+ private static final String INT_COLUMN = "intCol";
+ private static final String LONG_COLUMN = "longCol";
+
+ // Deliberately unsorted, and negative so the subtract-min in the creator is
actually exercised.
+ private static final int[] INT_VALUES = {40, -10, 25, -10, 100};
+ private static final long[] LONG_VALUES = {4000L, -1000L, 2500L, -1000L,
10000L};
+
+ @Test
+ public void testBuildsRangeIndexWhenMetadataMinMaxMissing()
+ throws Exception {
+ File indexDir = buildSegmentWithoutMinMax();
+
+ updateRangeIndices(indexDir);
+
+ SegmentMetadataImpl reloaded = new SegmentMetadataImpl(indexDir);
+ // The recovery happens inside the handler and is not written back to
metadata, so the column still reports null.
+ assertNull(reloaded.getColumnMetadataFor(INT_COLUMN).getMinValue());
+ assertNull(reloaded.getColumnMetadataFor(INT_COLUMN).getMaxValue());
+
+ try (SegmentDirectory segmentDirectory = new
SegmentLocalFSDirectory(indexDir, reloaded, ReadMode.mmap);
+ SegmentDirectory.Reader reader = segmentDirectory.createReader()) {
+ assertNotNull(reader);
+
+ PinotDataBuffer intBuffer = reader.getIndexFor(INT_COLUMN,
StandardIndexes.range());
+ assertNotNull(intBuffer, "Range index expected to be built for the INT
column");
+ ColumnMetadata intMetadata = reloaded.getColumnMetadataFor(INT_COLUMN);
+ BitSlicedRangeIndexReader intReader = new
BitSlicedRangeIndexReader(intBuffer, intMetadata);
+
+ // Range fully inside the recovered domain.
+ assertDocIds(intReader.getMatchingDocIds(-10, 25), 1, 2, 3);
+ // Single point at the recovered minimum.
+ assertDocIds(intReader.getMatchingDocIds(-10, -10), 1, 3);
+ // Range covering the whole column.
+ assertDocIds(intReader.getMatchingDocIds(-10, 100), 0, 1, 2, 3, 4);
+ // Range entirely below the recovered minimum matches nothing.
+ assertDocIds(intReader.getMatchingDocIds(-100, -50));
+
+ PinotDataBuffer longBuffer = reader.getIndexFor(LONG_COLUMN,
StandardIndexes.range());
+ assertNotNull(longBuffer, "Range index expected to be built for the LONG
column");
+ BitSlicedRangeIndexReader longReader =
+ new BitSlicedRangeIndexReader(longBuffer,
reloaded.getColumnMetadataFor(LONG_COLUMN));
+
+ assertDocIds(longReader.getMatchingDocIds(-1000L, 2500L), 1, 2, 3);
+ assertDocIds(longReader.getMatchingDocIds(2500L, 10000L), 0, 2, 4);
+ assertDocIds(longReader.getMatchingDocIds(-5000L, -2000L));
+ }
+ }
+
+ @Test
+ public void testOpenEndedQueryIsCorrectWithoutMetadataMax()
+ throws Exception {
+ // The reader takes the column max from metadata, which is still null
here, so it falls back to Long.MAX_VALUE.
+ // The RangeBitmap domain is self-contained, so results must stay correct
even though max pruning is weaker.
+ File indexDir = buildSegmentWithoutMinMax();
+
+ updateRangeIndices(indexDir);
+
+ SegmentMetadataImpl reloaded = new SegmentMetadataImpl(indexDir);
+ try (SegmentDirectory segmentDirectory = new
SegmentLocalFSDirectory(indexDir, reloaded, ReadMode.mmap);
+ SegmentDirectory.Reader reader = segmentDirectory.createReader()) {
+ BitSlicedRangeIndexReader intReader =
+ new BitSlicedRangeIndexReader(reader.getIndexFor(INT_COLUMN,
StandardIndexes.range()),
+ reloaded.getColumnMetadataFor(INT_COLUMN));
+
+ // Upper bound far beyond the real max: every doc at or above the lower
bound must still match.
+ assertDocIds(intReader.getMatchingDocIds(25, Integer.MAX_VALUE), 0, 2,
4);
+ assertDocIds(intReader.getMatchingDocIds(Integer.MIN_VALUE,
Integer.MAX_VALUE), 0, 1, 2, 3, 4);
+ }
+ }
+
+ private static void assertDocIds(ImmutableRoaringBitmap actual, int...
expected) {
+ assertNotNull(actual);
+ int[] actualDocIds = actual.toArray();
+ assertEquals(actualDocIds.length, expected.length,
+ "Unexpected match count, got " + Arrays.toString(actualDocIds));
+ for (int i = 0; i < expected.length; i++) {
+ assertEquals(actualDocIds[i], expected[i]);
+ }
+ }
+
+ /// Runs the handler over both columns with a v2 range index configured,
which is the path that needs the value
+ /// domain up front.
+ private static void updateRangeIndices(File indexDir)
+ throws Exception {
+ SegmentMetadataImpl segmentMetadata = new SegmentMetadataImpl(indexDir);
+ FieldIndexConfigs rangeV2 =
+ new FieldIndexConfigs.Builder().add(StandardIndexes.range(), new
RangeIndexConfig(2)).build();
+ Map<String, FieldIndexConfigs> configs = Map.of(INT_COLUMN, rangeV2,
LONG_COLUMN, rangeV2);
+ // A directory hands out either a reader or a writer, not both, so check
and build in separate opens.
+ try (SegmentDirectory segmentDirectory = new
SegmentLocalFSDirectory(indexDir, segmentMetadata, ReadMode.mmap);
+ SegmentDirectory.Reader reader = segmentDirectory.createReader()) {
+ RangeIndexHandler handler =
+ new RangeIndexHandler(segmentDirectory, configs,
createTableConfig(), createSchema());
+ assertTrue(handler.needUpdateIndices(reader),
+ "Range index is absent on disk, so the handler is expected to build
it");
+ }
+ try (SegmentDirectory segmentDirectory = new
SegmentLocalFSDirectory(indexDir, segmentMetadata, ReadMode.mmap);
+ SegmentDirectory.Writer writer = segmentDirectory.createWriter()) {
+ RangeIndexHandler handler =
+ new RangeIndexHandler(segmentDirectory, configs,
createTableConfig(), createSchema());
+ handler.updateIndices(writer);
+ }
+ }
+
+ private static Schema createSchema() {
+ return new Schema.SchemaBuilder().setSchemaName("testSchema")
+ .addSingleValueDimension(INT_COLUMN, DataType.INT)
+ .addSingleValueDimension(LONG_COLUMN, DataType.LONG)
+ .build();
+ }
+
+ private static TableConfig createTableConfig() {
+ return new TableConfigBuilder(TableType.OFFLINE).setTableName("testTable")
+ .setNoDictionaryColumns(List.of(INT_COLUMN, LONG_COLUMN))
+ .build();
+ }
+
+ private static File buildSegmentWithoutMinMax()
+ throws Exception {
+ FileUtils.deleteQuietly(TEMP_DIR);
+ SegmentGeneratorConfig config = new
SegmentGeneratorConfig(createTableConfig(), createSchema());
+ config.setOutDir(TEMP_DIR.getAbsolutePath());
+ config.setSegmentName(SEGMENT_NAME);
+
+ SegmentIndexCreationDriverImpl driver = new
SegmentIndexCreationDriverImpl();
+ driver.init(config, new GenericRowRecordReader(createRows()));
+ driver.build();
+
+ File indexDir = new File(TEMP_DIR, SEGMENT_NAME);
+ removeMinMaxValuesFromMetadataFile(indexDir);
+ return indexDir;
+ }
+
+ private static List<GenericRow> createRows() {
+ List<GenericRow> rows = new ArrayList<>();
+ for (int i = 0; i < INT_VALUES.length; i++) {
+ GenericRow row = new GenericRow();
+ row.putValue(INT_COLUMN, INT_VALUES[i]);
+ row.putValue(LONG_COLUMN, LONG_VALUES[i]);
+ rows.add(row);
+ }
+ return rows;
+ }
+
+ /// Strips min/max from the segment metadata, reproducing what a segment
committed with untracked min/max looks
+ /// like on disk.
+ private static void removeMinMaxValuesFromMetadataFile(File indexDir)
+ throws Exception {
+ PropertiesConfiguration configuration =
SegmentMetadataUtils.getPropertiesConfiguration(indexDir);
+ Iterator<String> keys = configuration.getKeys();
+ List<String> keysToClear = new ArrayList<>();
+ while (keys.hasNext()) {
+ String key = keys.next();
+ if (key.endsWith(V1Constants.MetadataKeys.Column.MIN_VALUE)
+ || key.endsWith(V1Constants.MetadataKeys.Column.MAX_VALUE)
+ ||
key.endsWith(V1Constants.MetadataKeys.Column.MIN_MAX_VALUE_INVALID)) {
+ keysToClear.add(key);
+ }
+ }
+ for (String key : keysToClear) {
+ configuration.clearProperty(key);
+ }
+ SegmentMetadataUtils.savePropertiesConfiguration(configuration, indexDir);
+ }
+
+ @AfterClass
+ public void tearDown() {
+ FileUtils.deleteQuietly(TEMP_DIR);
+ }
+}
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 b2fb2c5b5cf..80d18b6d615 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 @@ package org.apache.pinot.segment.local.utils;
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 class IndexCombinationValidationTest {
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<String, String> streamConfigs() {
+ Map<String, String> 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 17ca2edbf1a..763772a5b6d 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 @@ public interface IndexCreationContext extends ColumnShape {
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 interface IndexCreationContext extends ColumnShape {
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(@Nullable Comparable<?> minValue) {
+ _minValue = minValue;
+ return this;
+ }
+
+ /// Overrides the max value derived from the source [ColumnShape]. See
[#withMinValue].
+ public Builder withMaxValue(@Nullable Comparable<?> maxValue) {
+ _maxValue = maxValue;
+ return this;
+ }
+
// Build-time toggle setters.
public Builder withOnHeap(boolean onHeap) {
@@ -299,6 +319,10 @@ public interface IndexCreationContext extends ColumnShape {
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 @@ public interface IndexCreationContext extends ColumnShape {
_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 interface IndexCreationContext extends ColumnShape
{
@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
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
