Add RowIdSupplier to ColumnSelectorFactory.

processing/src/main/java/org/apache/druid/segment/ColumnSelectorFactory.java

@@ -53,4 +53,15 @@ public interface ColumnSelectorFactory extends ColumnInspector
   @Override
   @Nullable
   ColumnCapabilities getColumnCapabilities(String column);
+
+  /**
+   * Returns a {@link RowIdSupplier} that allows callers to detect whether the selectors returned by this
+   * factory have moved or not. Useful for selectors that wrap other selectors, such as virtual columns,
+   * as it allows them to cache their outputs.
+   */
+  @Nullable
+  default RowIdSupplier getRowIdSupplier()
+  {
+    return null;
+  }
 }

processing/src/main/java/org/apache/druid/segment/QueryableIndexColumnSelectorFactory.java

@@ -39,7 +39,7 @@
  * It's counterpart for incremental index is {@link
  * org.apache.druid.segment.incremental.IncrementalIndexColumnSelectorFactory}.
  */
-public class QueryableIndexColumnSelectorFactory implements ColumnSelectorFactory
+public class QueryableIndexColumnSelectorFactory implements ColumnSelectorFactory, RowIdSupplier
 {
   private final QueryableIndex index;
   private final VirtualColumns virtualColumns;
@@ -193,6 +193,19 @@ private <T extends BaseColumn> T getCachedColumn(final String columnName, final
     }
   }
 
+  @Nullable
+  @Override
+  public RowIdSupplier getRowIdSupplier()
+  {
+    return this;
+  }
+
+  @Override
+  public long getRowId()
+  {
+    return offset.getOffset();
+  }
+
   @Override
   @Nullable
   public ColumnCapabilities getColumnCapabilities(String columnName)

processing/src/main/java/org/apache/druid/segment/RowBasedColumnSelectorFactory.java

@@ -40,7 +40,6 @@
 import java.util.List;
 import java.util.Objects;
 import java.util.function.Function;
-import java.util.function.LongSupplier;
 import java.util.function.Supplier;
 import java.util.function.ToLongFunction;
 
@@ -49,12 +48,10 @@
  */
 public class RowBasedColumnSelectorFactory<T> implements ColumnSelectorFactory
 {
-  private static final long NO_ID = -1;
-
   private final Supplier<T> rowSupplier;
 
   @Nullable
-  private final LongSupplier rowIdSupplier;
+  private final RowIdSupplier rowIdSupplier;
   private final RowAdapter<T> adapter;
   private final ColumnInspector columnInspector;
   private final boolean throwParseExceptions;
@@ -65,7 +62,7 @@ public class RowBasedColumnSelectorFactory<T> implements ColumnSelectorFactory
    */
   RowBasedColumnSelectorFactory(
       final Supplier<T> rowSupplier,
-      @Nullable final LongSupplier rowIdSupplier,
+      @Nullable final RowIdSupplier rowIdSupplier,
       final RowAdapter<T> adapter,
       final ColumnInspector columnInspector,
       final boolean throwParseExceptions
@@ -167,7 +164,7 @@ private DimensionSelector makeDimensionSelectorUndecorated(DimensionSpec dimensi
 
       return new BaseSingleValueDimensionSelector()
       {
-        private long currentId = NO_ID;
+        private long currentId = RowIdSupplier.INIT;
         private String currentValue;
 
         @Override
@@ -186,11 +183,11 @@ public void inspectRuntimeShape(RuntimeShapeInspector inspector)
 
         private void updateCurrentValue()
         {
-          if (rowIdSupplier == null || rowIdSupplier.getAsLong() != currentId) {
+          if (rowIdSupplier == null || rowIdSupplier.getRowId() != currentId) {
             currentValue = extractionFn.apply(timestampFunction.applyAsLong(rowSupplier.get()));
 
             if (rowIdSupplier != null) {
-              currentId = rowIdSupplier.getAsLong();
+              currentId = rowIdSupplier.getRowId();
             }
           }
         }
@@ -200,7 +197,7 @@ private void updateCurrentValue()
 
       return new DimensionSelector()
       {
-        private long currentId = NO_ID;
+        private long currentId = RowIdSupplier.INIT;
         private List<String> dimensionValues;
 
         private final RangeIndexedInts indexedInts = new RangeIndexedInts();
@@ -331,7 +328,7 @@ public void inspectRuntimeShape(RuntimeShapeInspector inspector)
 
         private void updateCurrentValues()
         {
-          if (rowIdSupplier == null || rowIdSupplier.getAsLong() != currentId) {
+          if (rowIdSupplier == null || rowIdSupplier.getRowId() != currentId) {
             try {
               final Object rawValue = dimFunction.apply(rowSupplier.get());
 
@@ -377,12 +374,12 @@ private void updateCurrentValues()
               }
             }
             catch (Throwable e) {
-              currentId = NO_ID;
+              currentId = RowIdSupplier.INIT;
               throw e;
             }
 
             if (rowIdSupplier != null) {
-              currentId = rowIdSupplier.getAsLong();
+              currentId = rowIdSupplier.getRowId();
             }
           }
         }
@@ -491,6 +488,13 @@ private Number getCurrentValueAsNumber()
     }
   }
 
+  @Nullable
+  @Override
+  public RowIdSupplier getRowIdSupplier()
+  {
+    return rowIdSupplier;
+  }
+
   @Nullable
   @Override
   public ColumnCapabilities getColumnCapabilities(String columnName)

processing/src/main/java/org/apache/druid/segment/RowIdSupplier.java

@@ -0,0 +1,43 @@
+/*
+ * 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.druid.segment;
+
+/**
+ * Returned by {@link ColumnSelectorFactory#getRowIdSupplier()}. Allows users of {@link ColumnSelectorFactory}
+ * to cache objects returned by their selectors.
+ */
+public interface RowIdSupplier
+{
+  /**
+   * A number that will never be returned from {@link #getRowId()}. Useful for initialization.
+   */
+  long INIT = -1;
+
+  /**
+   * Returns a number that uniquely identifies the current position of some underlying cursor. This is useful for
+   * caching: it is safe to assume nothing has changed in the selector as long as the row ID stays the same.
+   *
+   * Row IDs do not need to be contiguous or monotonic. They need not have any meaning. In particular: they may not
+   * be row *numbers* (row number 0 may have any arbitrary row ID).
+   *
+   * Valid row IDs are always nonnegative.
+   */
+  long getRowId();
+}

processing/src/main/java/org/apache/druid/segment/incremental/IncrementalIndexColumnSelectorFactory.java

@@ -25,6 +25,7 @@
 import org.apache.druid.segment.ColumnValueSelector;
 import org.apache.druid.segment.DimensionIndexer;
 import org.apache.druid.segment.DimensionSelector;
+import org.apache.druid.segment.RowIdSupplier;
 import org.apache.druid.segment.SingleScanTimeDimensionSelector;
 import org.apache.druid.segment.VirtualColumns;
 import org.apache.druid.segment.column.ColumnCapabilities;
@@ -37,7 +38,7 @@
  * The basic implementation of {@link ColumnSelectorFactory} over an {@link IncrementalIndex}. It's counterpart for
  * historical segments is {@link org.apache.druid.segment.QueryableIndexColumnSelectorFactory}.
  */
-class IncrementalIndexColumnSelectorFactory implements ColumnSelectorFactory
+class IncrementalIndexColumnSelectorFactory implements ColumnSelectorFactory, RowIdSupplier
 {
   private final IncrementalIndexStorageAdapter adapter;
   private final IncrementalIndex index;
@@ -133,4 +134,17 @@ public ColumnCapabilities getColumnCapabilities(String columnName)
     // Use adapter.getColumnCapabilities instead of index.getCapabilities (see note in IncrementalIndexStorageAdapater)
     return adapter.getColumnCapabilities(columnName);
   }
+
+  @Nullable
+  @Override
+  public RowIdSupplier getRowIdSupplier()
+  {
+    return this;
+  }
+
+  @Override
+  public long getRowId()
+  {
+    return rowHolder.get().getRowIndex();
+  }
 }

processing/src/main/java/org/apache/druid/segment/join/HashJoinEngine.java

@@ -26,6 +26,7 @@
 import org.apache.druid.segment.ColumnValueSelector;
 import org.apache.druid.segment.Cursor;
 import org.apache.druid.segment.DimensionSelector;
+import org.apache.druid.segment.RowIdSupplier;
 import org.apache.druid.segment.column.ColumnCapabilities;
 import org.joda.time.DateTime;
 
@@ -69,8 +70,10 @@ public static Cursor makeJoinCursor(
                                                       closer
                                                   );
 
-    class JoinColumnSelectorFactory implements ColumnSelectorFactory
+    class JoinColumnSelectorFactory implements ColumnSelectorFactory, RowIdSupplier
     {
+      private long rowId = 0;
+
       @Override
       public DimensionSelector makeDimensionSelector(DimensionSpec dimensionSpec)
       {
@@ -116,6 +119,29 @@ public ColumnCapabilities getColumnCapabilities(String column)
           return leftColumnSelectorFactory.getColumnCapabilities(column);
         }
       }
+
+      @Nullable
+      @Override
+      public RowIdSupplier getRowIdSupplier()
+      {
+        return this;
+      }
+
+      @Override
+      public long getRowId()
+      {
+        return rowId;
+      }
+
+      void advanceRowId()
+      {
+        rowId++;
+      }
+
+      void resetRowId()
+      {
+        rowId = 0;
+      }
     }
 
     final JoinColumnSelectorFactory joinColumnSelectorFactory = new JoinColumnSelectorFactory();
@@ -171,6 +197,8 @@ private void matchCurrentPosition()
       @Override
       public void advanceUninterruptibly()
       {
+        joinColumnSelectorFactory.advanceRowId();
+
         if (joinMatcher.hasMatch()) {
           joinMatcher.nextMatch();
 
@@ -218,6 +246,7 @@ public void reset()
       {
         leftCursor.reset();
         joinMatcher.reset();
+        joinColumnSelectorFactory.resetRowId();
       }
     }