Reject range index on ingestion-aggregated no-dictionary columns

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;

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,44 @@ 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();
+      // 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 "
+              + "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;

pinot-segment-local/src/test/java/org/apache/pinot/segment/local/segment/index/creator/BitSlicedIndexCreatorTest.java

@@ -65,14 +65,21 @@ 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)
   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 {

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<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 {@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());
+  }
 }