Support range index on ingestion-aggregated no-dictionary columns

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

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;
+    }
+  }
 }

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<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<>();
 

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) {