From 5862cf33857e4ab1ae6250a9822469dac2df6f11 Mon Sep 17 00:00:00 2001 From: RAGHVENDRA KUMAR YADAV Date: Tue, 28 Jul 2026 19:01:00 -0700 Subject: [PATCH 1/2] Reject range index on ingestion-aggregated no-dictionary columns Ingestion-time metrics aggregation (aggregationConfigs / aggregateMetrics) forces its aggregated metric columns to be no-dictionary, and their min/max values are never tracked when the consuming segment is committed. The BitSliced range index (version 2) creator reads min/max for a single-value no-dictionary column, so this combination throws a NullPointerException for INT/LONG columns (and silently builds a degenerate index for FLOAT/DOUBLE), leaving segments unable to commit. Add validation in RangeIndexType.validate() to reject a version-2 range index on such columns upfront, and add a defense-in-depth guard in BitSlicedRangeIndexCreator so the raw-column path fails with a clear, actionable message instead of a NullPointerException on direct/programmatic creation and segment reload. --- .../impl/inv/BitSlicedRangeIndexCreator.java | 17 ++- .../segment/index/range/RangeIndexType.java | 35 +++++ .../creator/BitSlicedIndexCreatorTest.java | 9 +- .../utils/IndexCombinationValidationTest.java | 124 ++++++++++++++++++ 4 files changed, 183 insertions(+), 2 deletions(-) diff --git a/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/segment/creator/impl/inv/BitSlicedRangeIndexCreator.java b/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/segment/creator/impl/inv/BitSlicedRangeIndexCreator.java index b9b008fe309f..f00fc222494a 100644 --- a/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/segment/creator/impl/inv/BitSlicedRangeIndexCreator.java +++ b/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/segment/creator/impl/inv/BitSlicedRangeIndexCreator.java @@ -71,10 +71,25 @@ public BitSlicedRangeIndexCreator(File indexDir, FieldSpec fieldSpec, int cardin */ public BitSlicedRangeIndexCreator(File indexDir, FieldSpec fieldSpec, Comparable minValue, Comparable maxValue) { - this(indexDir, fieldSpec, minValue(fieldSpec, minValue), maxValue(fieldSpec, minValue, maxValue), + // Arguments are evaluated left-to-right, so checkedMinValue() validates min/max before maxValue() is reached. + this(indexDir, fieldSpec, checkedMinValue(fieldSpec, minValue, maxValue), maxValue(fieldSpec, minValue, maxValue), fieldSpec.getDataType().getStoredType()); } + /** + * Validates that a raw (no-dictionary) column has computed min/max values, then returns the min encoded as a long. + * Ingestion-aggregated metric columns never track min/max, which would otherwise throw a + * {@link NullPointerException} for INT/LONG or silently build a degenerate index for FLOAT/DOUBLE. Fail loudly with + * an actionable message instead. Such configurations are rejected upfront in {@code RangeIndexType.validate}; this + * guard also covers direct/programmatic creation and segment reload paths. + */ + private static long checkedMinValue(FieldSpec fieldSpec, Comparable minValue, Comparable maxValue) { + Preconditions.checkState(minValue != null && maxValue != null, + "Cannot build BitSliced range index on raw column: %s without min/max values; use range index version 1 or " + + "remove the range index on this column", fieldSpec.getName()); + return minValue(fieldSpec, minValue); + } + @Override public DataType getValueType() { return _valueType; diff --git a/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/segment/index/range/RangeIndexType.java b/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/segment/index/range/RangeIndexType.java index 03c77cd19990..8552d0ee8803 100644 --- a/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/segment/index/range/RangeIndexType.java +++ b/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/segment/index/range/RangeIndexType.java @@ -49,6 +49,9 @@ import org.apache.pinot.segment.spi.store.SegmentDirectory; import org.apache.pinot.spi.config.table.FieldConfig; 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; @@ -96,9 +99,41 @@ public void validate(FieldIndexConfigs indexConfigs, FieldSpec fieldSpec, TableC "Cannot create range index version %s on column: %s with RAW forward index and dictionary; " + "use BitSliced range index version %s instead", RangeIndexCreator.VERSION, column, BitSlicedRangeIndexCreator.VERSION); + // Ingestion-time metrics aggregation (aggregationConfigs / aggregateMetrics) only takes effect on REALTIME + // consuming segments. Its aggregated metric columns are forced to be no-dictionary and always have null + // min/max at segment commit time. The BitSliced range index (version 2) creator reads min/max for a + // single-value no-dictionary column, so this combination either throws a NullPointerException (INT/LONG) or + // silently builds a degenerate index over a wrong value domain (FLOAT/DOUBLE). Reject it upfront. + boolean bitSlicedSingleValueRangeIndex = + rangeIndexConfig.getVersion() == BitSlicedRangeIndexCreator.VERSION && fieldSpec.isSingleValueField(); + boolean realtimeAggregatedNoDictColumn = tableConfig.getTableType() == TableType.REALTIME && !dictionaryEnabled + && isAggregatedColumn(tableConfig, fieldSpec); + Preconditions.checkState(!(bitSlicedSingleValueRangeIndex && realtimeAggregatedNoDictColumn), + "Cannot create BitSliced range index (version %s) on aggregated no-dictionary column: %s; aggregated " + + "metric columns have no min/max at segment build time. Use range index version 1 or remove the range " + + "index on this column", + BitSlicedRangeIndexCreator.VERSION, column); } } + /** + * Returns {@code true} if the given column is produced by ingestion-time metrics aggregation, either via an + * explicit {@link AggregationConfig} or via the legacy {@code aggregateMetrics} flag (which aggregates all metric + * columns). + */ + private static boolean isAggregatedColumn(TableConfig tableConfig, FieldSpec fieldSpec) { + IngestionConfig ingestionConfig = tableConfig.getIngestionConfig(); + if (ingestionConfig != null && ingestionConfig.getAggregationConfigs() != null) { + for (AggregationConfig aggregationConfig : ingestionConfig.getAggregationConfigs()) { + if (fieldSpec.getName().equals(aggregationConfig.getColumnName())) { + return true; + } + } + } + return tableConfig.getIndexingConfig().isAggregateMetrics() + && fieldSpec.getFieldType() == FieldSpec.FieldType.METRIC; + } + @Override public String getPrettyName() { return INDEX_DISPLAY_NAME; diff --git a/pinot-segment-local/src/test/java/org/apache/pinot/segment/local/segment/index/creator/BitSlicedIndexCreatorTest.java b/pinot-segment-local/src/test/java/org/apache/pinot/segment/local/segment/index/creator/BitSlicedIndexCreatorTest.java index 7e4571a29045..dc24eb895381 100644 --- a/pinot-segment-local/src/test/java/org/apache/pinot/segment/local/segment/index/creator/BitSlicedIndexCreatorTest.java +++ b/pinot-segment-local/src/test/java/org/apache/pinot/segment/local/segment/index/creator/BitSlicedIndexCreatorTest.java @@ -65,7 +65,8 @@ public void tearDown() @Test(expectedExceptions = IllegalArgumentException.class) public void testFailToCreateRawString() { - new BitSlicedRangeIndexCreator(INDEX_DIR, new DimensionFieldSpec("foo", STRING, true), null, null); + // Non-null min/max so the unsupported-raw-type check is exercised (not the null min/max guard below). + new BitSlicedRangeIndexCreator(INDEX_DIR, new DimensionFieldSpec("foo", STRING, true), "a", "z"); } @Test(expectedExceptions = IllegalArgumentException.class) @@ -73,6 +74,12 @@ public void testFailToCreateMV() { new BitSlicedRangeIndexCreator(INDEX_DIR, new DimensionFieldSpec("foo", INT, false), 0, 10); } + @Test(expectedExceptions = IllegalStateException.class) + public void testFailToCreateRawWithoutMinMax() { + // Aggregated no-dictionary metric columns never track min/max; the creator must fail loudly, not NPE. + new BitSlicedRangeIndexCreator(INDEX_DIR, new DimensionFieldSpec("foo", LONG, true), null, null); + } + @Test public void testCreateAndQueryInt() throws IOException { 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 a61c2d813162..8ffa8bc1536a 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; @@ -662,4 +666,124 @@ 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 only takes effect on REALTIME consuming segments. Its aggregated + // metric columns are forced to be no-dictionary and have null min/max at segment build time, which makes + // the BitSliced range index creator throw an NPE. The combination must be rejected for REALTIME tables. + // ============================================================ + + 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 {@code SUM(s1)} into the no-dictionary metric column {@code 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 assertInvalid(TableConfig tableConfig, Schema schema, String errorFragment) { + try { + TableConfigUtils.validate(tableConfig, schema); + fail("Expected validation to fail with: " + errorFragment); + } catch (Exception e) { + assertTrue(e.getMessage() != null && e.getMessage().contains(errorFragment), + "Expected '" + errorFragment + "' in error but got: " + e.getMessage()); + } + } + + 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 testRangeIndexOnIngestionAggregatedColumnFails() { + // aggregationConfigs (SUM into a no-dictionary metric column) + range index on that metric column + TableConfig tc = realtimeBuilder() + .setIngestionConfig(sumAggregationIngestionConfig()) + .setNoDictionaryColumns(List.of(AGG_METRIC_COL)) + .setRangeIndexColumns(List.of(AGG_METRIC_COL)) + .build(); + assertInvalid(tc, aggregationSchema(), "Cannot create BitSliced range index"); + } + + @Test + public void testRangeIndexOnAggregateMetricsColumnFails() { + // Legacy aggregateMetrics flag (implicitly SUMs every metric) + range index on the no-dictionary metric column + TableConfig tc = realtimeBuilder() + .setAggregateMetrics(true) + .setNoDictionaryColumns(List.of(AGG_METRIC_COL)) + .setRangeIndexColumns(List.of(AGG_METRIC_COL)) + .build(); + assertInvalid(tc, aggregationSchema(), "Cannot create BitSliced range index"); + } + + @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()); + } } From 2ae1fdf5fd6a79dc70634859ecb7d5fee6aeb00b Mon Sep 17 00:00:00 2001 From: RAGHVENDRA KUMAR YADAV Date: Wed, 29 Jul 2026 10:00:28 -0700 Subject: [PATCH 2/2] Guard against null tableConfig in RangeIndexType.validate The ingestion-aggregation range-index check dereferenced tableConfig, but direct/programmatic index-creation paths (e.g. OpenStructColumnSplitter) call validate() with a null tableConfig, causing a NullPointerException. Skip the check when no table config is available. --- .../segment/local/segment/index/range/RangeIndexType.java | 5 ++++- 1 file changed, 4 insertions(+), 1 deletion(-) diff --git a/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/segment/index/range/RangeIndexType.java b/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/segment/index/range/RangeIndexType.java index 8552d0ee8803..c4a6c207a061 100644 --- a/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/segment/index/range/RangeIndexType.java +++ b/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/segment/index/range/RangeIndexType.java @@ -106,7 +106,10 @@ public void validate(FieldIndexConfigs indexConfigs, FieldSpec fieldSpec, TableC // silently builds a degenerate index over a wrong value domain (FLOAT/DOUBLE). Reject it upfront. boolean bitSlicedSingleValueRangeIndex = rangeIndexConfig.getVersion() == BitSlicedRangeIndexCreator.VERSION && fieldSpec.isSingleValueField(); - boolean realtimeAggregatedNoDictColumn = tableConfig.getTableType() == TableType.REALTIME && !dictionaryEnabled + // tableConfig may be null on direct/programmatic index-creation paths (e.g. OpenStructColumnSplitter); the + // ingestion-aggregation check only applies when a REALTIME table config is available. + boolean realtimeAggregatedNoDictColumn = tableConfig != null + && tableConfig.getTableType() == TableType.REALTIME && !dictionaryEnabled && isAggregatedColumn(tableConfig, fieldSpec); Preconditions.checkState(!(bitSlicedSingleValueRangeIndex && realtimeAggregatedNoDictColumn), "Cannot create BitSliced range index (version %s) on aggregated no-dictionary column: %s; aggregated "