Add sorted index path to BenchmarkInvertedIndexDistinct

Add sortedIndexPath() benchmark method that simulates the sorted column
execution path: merge-iterates filter bitmap against contiguous doc
ranges using PeekableIntIterator.advanceIfNeeded(). This benchmarks
all three DISTINCT operator paths side-by-side:

- sortedIndexPath: O(cardinality + filteredDocs) merge iteration
- invertedIndexPath: O(cardinality) bitmap intersects() checks
- scanPath: O(filteredDocs) forward index lookups with dedup

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

Author: xiangfu0

pinot-core/src/main/java/org/apache/pinot/core/operator/query/InvertedIndexDistinctOperator.java

@@ -70,11 +70,13 @@
 /**
  * Inverted-index-based operator for single-column distinct queries on a single segment.
  *
- * <p>Supports two execution paths, chosen at runtime based on a cost heuristic:
+ * <p>Supports three execution paths, chosen at runtime:
  * <ul>
- *   <li><b>Inverted index path</b>: Iterates dictionary entries and uses inverted index bitmap intersections
- *       to check filter membership. Avoids the projection pipeline entirely. Preferred when dictionary cardinality
- *       is much smaller than the filtered doc count.</li>
+ *   <li><b>Sorted index path</b>: For sorted columns, merge-iterates filter bitmap against contiguous doc ranges.
+ *       Cost ~ O(cardinality + filteredDocs). Always chosen when the column has a sorted forward index.</li>
+ *   <li><b>Bitmap inverted index path</b>: Iterates dictionary entries and uses inverted index bitmap intersections
+ *       to check filter membership. Avoids the projection pipeline entirely. Chosen by cost heuristic when dictionary
+ *       cardinality is much smaller than the filtered doc count.</li>
  *   <li><b>Scan path (fallback)</b>: Uses ProjectOperator + DistinctExecutor to scan filtered docs.
  *       Used when the cost heuristic determines scanning is cheaper.</li>
  * </ul>
@@ -311,11 +313,9 @@ private DistinctResultsBlock executeInvertedIndexPath() {
     for (dictId = 0; dictId < dictLength; dictId++) {
       QueryThreadContext.checkTerminationAndSampleUsagePeriodically(dictId, EXPLAIN_NAME);
 
-      Object docIdsObj = _invertedIndexReader.getDocIds(dictId);
-      if (!(docIdsObj instanceof ImmutableRoaringBitmap)) {
-        continue;
-      }
-      ImmutableRoaringBitmap docIds = (ImmutableRoaringBitmap) docIdsObj;
+      // SortedIndexReader is handled separately in getNextBlock(), so this path only sees bitmap inverted indexes
+      // whose getDocIds() returns ImmutableRoaringBitmap.
+      ImmutableRoaringBitmap docIds = (ImmutableRoaringBitmap) _invertedIndexReader.getDocIds(dictId);
       if (docIds.isEmpty()) {
         continue;
       }
@@ -534,10 +534,15 @@ private static boolean addToTable(BytesDistinctTable table, ByteArray value,
 
   @Override
   public List<? extends Operator> getChildOperators() {
-    if (_usedInvertedIndexPath || _projectOperator == null) {
-      return Collections.emptyList();
+    if (_usedInvertedIndexPath) {
+      // For inverted/sorted index paths, the filter operator is the logical child
+      // (it provides the filtered docId set that we intersect with dictionary bitmaps).
+      return Collections.singletonList(_filterOperator);
+    }
+    if (_projectOperator != null) {
+      return Collections.singletonList(_projectOperator);
     }
-    return Collections.singletonList(_projectOperator);
+    return Collections.emptyList();
   }
 
   @Override
@@ -549,9 +554,10 @@ public IndexSegment getIndexSegment() {
   public ExecutionStatistics getExecutionStatistics() {
     int numTotalDocs = _indexSegment.getSegmentMetadata().getTotalDocs();
     if (_usedInvertedIndexPath || _projectOperator == null) {
-      // For the inverted index path, report dictionary entries examined as numDocsScanned.
-      // This isn't strictly "docs scanned" but represents the work done by this operator.
-      return new ExecutionStatistics(_numDocsScanned > 0 ? _numDocsScanned : _numEntriesExamined, 0, 0, numTotalDocs);
+      // For inverted/sorted index paths: numDocsScanned=0 (no forward index lookups),
+      // numEntriesScannedInFilter=0, numEntriesScannedPostFilter=numEntriesExamined
+      // (dictionary entries examined via bitmap intersection or sorted range checks).
+      return new ExecutionStatistics(0, 0, _numEntriesExamined, numTotalDocs);
     }
     long numEntriesScannedInFilter = _projectOperator.getExecutionStatistics().getNumEntriesScannedInFilter();
     long numEntriesScannedPostFilter = (long) _numDocsScanned * _projectOperator.getNumColumnsProjected();

pinot-perf/src/main/java/org/apache/pinot/perf/BenchmarkBitmapIntersectionVsAnd.java

@@ -35,20 +35,24 @@
 import org.openjdk.jmh.infra.Blackhole;
 import org.openjdk.jmh.runner.Runner;
 import org.openjdk.jmh.runner.options.OptionsBuilder;
+import org.roaringbitmap.PeekableIntIterator;
 import org.roaringbitmap.RoaringBitmap;
 
 
 /**
- * Benchmark to compare inverted-index-based distinct vs scan-based distinct execution paths.
+ * Benchmark comparing three execution paths for single-column DISTINCT queries:
  *
- * <p>Inverted index path: iterates all dictionary entries, does bitmap intersection with filter for each.
- * Cost ~ O(dictionaryCardinality * bitmapIntersectionCost).
+ * <ol>
+ *   <li><b>Sorted index path</b>: merge-iterates filter bitmap against contiguous doc ranges per dictId.
+ *       Cost ~ O(dictionaryCardinality + filterCardinality). Only applicable when the column is sorted.</li>
+ *   <li><b>Bitmap inverted index path</b>: iterates all dictionary entries, uses {@code intersects()} to check
+ *       filter membership per entry. Cost ~ O(dictionaryCardinality * bitmapIntersectionCost).</li>
+ *   <li><b>Scan path</b>: iterates all filtered docIds, looks up dictId from forward index, deduplicates.
+ *       Cost ~ O(filterCardinality * forwardIndexLookupCost).</li>
+ * </ol>
  *
- * <p>Scan path: iterates all filtered docIds, looks up dictId from forward index (int[]), deduplicates.
- * Cost ~ O(filterCardinality * forwardIndexLookupCost).
- *
- * <p>The goal is to determine the crossover ratio (dictionaryCardinality / filterCardinality) at which
- * the inverted index path becomes faster than the scan path.
+ * <p>The sorted index path is always the fastest when applicable (column is sorted), as it avoids both
+ * bitmap intersection and per-doc forward index lookups.
  *
  * <p>Usage: {@code java -jar pinot-perf/target/benchmarks.jar BenchmarkInvertedIndexDistinct}
  */
@@ -70,12 +74,17 @@ public class BenchmarkInvertedIndexDistinct {
   @Param({"0.001", "0.01", "0.1", "0.5", "1.0"})
   double _filterSelectivity;
 
-  // -- Inverted index: dictId -> docIds bitmap
+  // -- Bitmap inverted index: dictId -> docIds bitmap (non-sorted, random distribution)
   private RoaringBitmap[] _invertedIndex;
 
-  // -- Forward index: docId -> dictId (simulates column forward index)
+  // -- Forward index: docId -> dictId (simulates column forward index for scan path)
   private int[] _forwardIndex;
 
+  // -- Sorted index ranges: dictId -> [startDocId, endDocId] (inclusive)
+  // Simulates SortedIndexReader.getDocIds(dictId) which returns contiguous doc ranges
+  private int[] _sortedRangeStarts;
+  private int[] _sortedRangeEnds;
+
   // -- Filter bitmap: which docIds pass the filter
   private RoaringBitmap _filterBitmap;
 
@@ -85,6 +94,8 @@ public class BenchmarkInvertedIndexDistinct {
   public void setup() {
     Random random = new Random(42);
 
+    // ---- Non-sorted data (for inverted index and scan paths) ----
+
     // Build forward index: assign each doc a random dictId
     _forwardIndex = new int[_numDocs];
     for (int docId = 0; docId < _numDocs; docId++) {
@@ -100,7 +111,22 @@ public void setup() {
       _invertedIndex[_forwardIndex[docId]].add(docId);
     }
 
-    // Build filter bitmap: randomly select filterSelectivity fraction of docs
+    // ---- Sorted data (for sorted index path) ----
+
+    // Each dictId maps to a contiguous doc range, simulating a sorted column.
+    // dictId=0: docs [0, docsPerValue-1], dictId=1: docs [docsPerValue, 2*docsPerValue-1], etc.
+    int docsPerValue = _numDocs / _dictionaryCardinality;
+    _sortedRangeStarts = new int[_dictionaryCardinality];
+    _sortedRangeEnds = new int[_dictionaryCardinality];
+    for (int dictId = 0; dictId < _dictionaryCardinality; dictId++) {
+      _sortedRangeStarts[dictId] = dictId * docsPerValue;
+      _sortedRangeEnds[dictId] = (dictId + 1) * docsPerValue - 1;
+    }
+    // Last range absorbs any remainder
+    _sortedRangeEnds[_dictionaryCardinality - 1] = _numDocs - 1;
+
+    // ---- Filter bitmap (shared across all paths) ----
+
     _filterCardinality = Math.max(1, (int) (_numDocs * _filterSelectivity));
     _filterBitmap = new RoaringBitmap();
     if (_filterSelectivity >= 1.0) {
@@ -131,8 +157,40 @@ public void setup() {
   }
 
   /**
-   * Inverted index path: iterate all dictionary entries, intersect each with filter bitmap.
-   * For each matching value, add to a dedup set (simulated by a RoaringBitmap of dictIds).
+   * Sorted index path: merge-iterate filter bitmap against contiguous doc ranges.
+   * Uses PeekableIntIterator.advanceIfNeeded() to skip filter docs between ranges.
+   * Cost ~ O(dictionaryCardinality + filterCardinality).
+   */
+  @Benchmark
+  public int sortedIndexPath(Blackhole bh) {
+    RoaringBitmap seenDictIds = new RoaringBitmap();
+    int valuesProcessed = 0;
+
+    PeekableIntIterator filterIter = _filterBitmap.getIntIterator();
+    for (int dictId = 0; dictId < _dictionaryCardinality && filterIter.hasNext(); dictId++) {
+      int startDocId = _sortedRangeStarts[dictId];
+      int endDocId = _sortedRangeEnds[dictId];
+
+      // Skip filter docs before this range
+      filterIter.advanceIfNeeded(startDocId);
+
+      // Check if any filter doc falls within this range
+      if (filterIter.hasNext() && filterIter.peekNext() <= endDocId) {
+        seenDictIds.add(dictId);
+        valuesProcessed++;
+        // Advance past the current range for next dictId
+        filterIter.advanceIfNeeded(endDocId + 1);
+      }
+    }
+
+    bh.consume(seenDictIds);
+    return valuesProcessed;
+  }
+
+  /**
+   * Bitmap inverted index path: iterate all dictionary entries, intersect each with filter bitmap.
+   * Uses intersects() for early termination instead of computing full intersection.
+   * Cost ~ O(dictionaryCardinality * bitmapIntersectionCost).
    */
   @Benchmark
   public int invertedIndexPath(Blackhole bh) {
@@ -157,6 +215,7 @@ public int invertedIndexPath(Blackhole bh) {
 
   /**
    * Scan path: iterate all filtered docIds, look up dictId from forward index, dedup.
+   * Cost ~ O(filterCardinality * forwardIndexLookupCost).
    */
   @Benchmark
   public int scanPath(Blackhole bh) {