Optimization for expressions that hit a single long column.

benchmarks/src/main/java/org/apache/druid/benchmark/ExpressionSelectorBenchmark.java

@@ -80,7 +80,16 @@ public void setup()
   {
     final BenchmarkSchemaInfo schemaInfo = new BenchmarkSchemaInfo(
         ImmutableList.of(
-            BenchmarkColumnSchema.makeNormal("n", ValueType.LONG, false, 1, 0d, 0d, 10000d, false),
+            BenchmarkColumnSchema.makeZipf(
+                "n",
+                ValueType.LONG,
+                false,
+                1,
+                0d,
+                1000,
+                10000,
+                3d
+            ),
             BenchmarkColumnSchema.makeZipf(
                 "s",
                 ValueType.STRING,
@@ -146,10 +155,7 @@ public void timeFloorUsingExpression(Blackhole blackhole)
     final List<?> results = cursors
         .map(cursor -> {
           final ColumnValueSelector selector = cursor.getColumnSelectorFactory().makeColumnValueSelector("v");
-          while (!cursor.isDone()) {
-            blackhole.consume(selector.getLong());
-            cursor.advance();
-          }
+          consumeLong(cursor, selector, blackhole);
           return null;
         })
         .toList();
@@ -219,6 +225,71 @@ public void timeFloorUsingCursor(Blackhole blackhole)
     blackhole.consume(count);
   }
 
+  @Benchmark
+  public void timeFormatUsingExpression(Blackhole blackhole)
+  {
+    final Sequence<Cursor> cursors = new QueryableIndexStorageAdapter(index).makeCursors(
+        null,
+        index.getDataInterval(),
+        VirtualColumns.create(
+            ImmutableList.of(
+                new ExpressionVirtualColumn(
+                    "v",
+                    "timestamp_format(__time, 'yyyy-MM-dd')",
+                    ValueType.STRING,
+                    TestExprMacroTable.INSTANCE
+                )
+            )
+        ),
+        Granularities.ALL,
+        false,
+        null
+    );
+
+    final List<?> results = cursors
+        .map(cursor -> {
+          final DimensionSelector selector = cursor.getColumnSelectorFactory().makeDimensionSelector(
+              DefaultDimensionSpec.of("v")
+          );
+          consumeDimension(cursor, selector, blackhole);
+          return null;
+        })
+        .toList();
+
+    blackhole.consume(results);
+  }
+
+  @Benchmark
+  public void timeFormatUsingExtractionFn(Blackhole blackhole)
+  {
+    final Sequence<Cursor> cursors = new QueryableIndexStorageAdapter(index).makeCursors(
+        null,
+        index.getDataInterval(),
+        VirtualColumns.EMPTY,
+        Granularities.ALL,
+        false,
+        null
+    );
+
+    final List<?> results = cursors
+        .map(cursor -> {
+          final DimensionSelector selector = cursor
+              .getColumnSelectorFactory()
+              .makeDimensionSelector(
+                  new ExtractionDimensionSpec(
+                      ColumnHolder.TIME_COLUMN_NAME,
+                      "v",
+                      new TimeFormatExtractionFn("yyyy-MM-dd", null, null, null, false)
+                  )
+              );
+          consumeDimension(cursor, selector, blackhole);
+          return null;
+        })
+        .toList();
+
+    blackhole.consume(results);
+  }
+
   @Benchmark
   public void strlenUsingExpressionAsLong(Blackhole blackhole)
   {
@@ -312,6 +383,70 @@ public void strlenUsingExtractionFn(Blackhole blackhole)
     blackhole.consume(results);
   }
 
+  @Benchmark
+  public void arithmeticOnLong(Blackhole blackhole)
+  {
+    final Sequence<Cursor> cursors = new QueryableIndexStorageAdapter(index).makeCursors(
+        null,
+        index.getDataInterval(),
+        VirtualColumns.create(
+            ImmutableList.of(
+                new ExpressionVirtualColumn(
+                    "v",
+                    "n + 1",
+                    ValueType.LONG,
+                    TestExprMacroTable.INSTANCE
+                )
+            )
+        ),
+        Granularities.ALL,
+        false,
+        null
+    );
+
+    final List<?> results = cursors
+        .map(cursor -> {
+          final ColumnValueSelector selector = cursor.getColumnSelectorFactory().makeColumnValueSelector("v");
+          consumeLong(cursor, selector, blackhole);
+          return null;
+        })
+        .toList();
+
+    blackhole.consume(results);
+  }
+
+  @Benchmark
+  public void stringConcatAndCompareOnLong(Blackhole blackhole)
+  {
+    final Sequence<Cursor> cursors = new QueryableIndexStorageAdapter(index).makeCursors(
+        null,
+        index.getDataInterval(),
+        VirtualColumns.create(
+            ImmutableList.of(
+                new ExpressionVirtualColumn(
+                    "v",
+                    "concat(n, ' is my favorite number') == '3 is my favorite number'",
+                    ValueType.LONG,
+                    TestExprMacroTable.INSTANCE
+                )
+            )
+        ),
+        Granularities.ALL,
+        false,
+        null
+    );
+
+    final List<?> results = cursors
+        .map(cursor -> {
+          final ColumnValueSelector selector = cursor.getColumnSelectorFactory().makeColumnValueSelector("v");
+          consumeLong(cursor, selector, blackhole);
+          return null;
+        })
+        .toList();
+
+    blackhole.consume(results);
+  }
+
   private void consumeDimension(final Cursor cursor, final DimensionSelector selector, final Blackhole blackhole)
   {
     if (selector.getValueCardinality() >= 0) {

benchmarks/src/main/java/org/apache/druid/benchmark/datagen/SegmentGenerator.java

@@ -27,6 +27,7 @@
 import org.apache.druid.data.input.InputRow;
 import org.apache.druid.data.input.impl.DimensionSchema;
 import org.apache.druid.data.input.impl.DimensionsSpec;
+import org.apache.druid.data.input.impl.DoubleDimensionSchema;
 import org.apache.druid.data.input.impl.FloatDimensionSchema;
 import org.apache.druid.data.input.impl.LongDimensionSchema;
 import org.apache.druid.data.input.impl.StringDimensionSchema;
@@ -99,6 +100,9 @@ public QueryableIndex generate(
           case LONG:
             dimensions.add(new LongDimensionSchema(columnSchema.getName()));
             break;
+          case DOUBLE:
+            dimensions.add(new DoubleDimensionSchema(columnSchema.getName()));
+            break;
           case FLOAT:
             dimensions.add(new FloatDimensionSchema(columnSchema.getName()));
             break;

core/src/main/java/org/apache/druid/math/expr/ExprEval.java

@@ -30,6 +30,10 @@
  */
 public abstract class ExprEval<T>
 {
+  // Cached String values. Protected so they can be used by subclasses.
+  private boolean stringValueValid = false;
+  private String stringValue;
+
   public static ExprEval ofLong(@Nullable Number longValue)
   {
     return new LongExprEval(longValue);
@@ -89,7 +93,7 @@ public static ExprEval bestEffortOf(@Nullable Object val)
   @Nullable
   final T value;
 
-  private ExprEval(T value)
+  private ExprEval(@Nullable T value)
   {
     this.value = value;
   }
@@ -115,7 +119,17 @@ public Object value()
   @Nullable
   public String asString()
   {
-    return value == null ? null : String.valueOf(value);
+    if (!stringValueValid) {
+      if (value == null) {
+        stringValue = null;
+      } else {
+        stringValue = String.valueOf(value);
+      }
+
+      stringValueValid = true;
+    }
+
+    return stringValue;
   }
 
   public abstract boolean asBoolean();
@@ -126,7 +140,6 @@ public String asString()
 
   private abstract static class NumericExprEval extends ExprEval<Number>
   {
-
     private NumericExprEval(@Nullable Number value)
     {
       super(value);
@@ -247,6 +260,16 @@ public Expr toExpr()
 
   private static class StringExprEval extends ExprEval<String>
   {
+    // Cached primitive values.
+    private boolean intValueValid = false;
+    private boolean longValueValid = false;
+    private boolean doubleValueValid = false;
+    private boolean booleanValueValid = false;
+    private int intValue;
+    private long longValue;
+    private double doubleValue;
+    private boolean booleanValue;
+
     private static final StringExprEval OF_NULL = new StringExprEval(null);
 
     private Number numericVal;
@@ -263,31 +286,68 @@ public final ExprType type()
     }
 
     @Override
-    public final int asInt()
+    public int asInt()
+    {
+      if (!intValueValid) {
+        intValue = computeInt();
+        intValueValid = true;
+      }
+
+      return intValue;
+    }
+
+    @Override
+    public long asLong()
+    {
+      if (!longValueValid) {
+        longValue = computeLong();
+        longValueValid = true;
+      }
+
+      return longValue;
+    }
+
+    @Override
+    public double asDouble()
+    {
+      if (!doubleValueValid) {
+        doubleValue = computeDouble();
+        doubleValueValid = true;
+      }
+
+      return doubleValue;
+    }
+
+    @Nullable
+    @Override
+    public String asString()
     {
-      Number number = asNumber();
+      return value;
+    }
+
+    private int computeInt()
+    {
+      Number number = computeNumber();
       if (number == null) {
         assert NullHandling.replaceWithDefault();
         return 0;
       }
       return number.intValue();
     }
 
-    @Override
-    public final long asLong()
+    private long computeLong()
     {
-      Number number = asNumber();
+      Number number = computeNumber();
       if (number == null) {
         assert NullHandling.replaceWithDefault();
         return 0L;
       }
       return number.longValue();
     }
 
-    @Override
-    public final double asDouble()
+    private double computeDouble()
     {
-      Number number = asNumber();
+      Number number = computeNumber();
       if (number == null) {
         assert NullHandling.replaceWithDefault();
         return 0.0d;
@@ -296,7 +356,7 @@ public final double asDouble()
     }
 
     @Nullable
-    private Number asNumber()
+    private Number computeNumber()
     {
       if (value == null) {
         return null;
@@ -321,23 +381,28 @@ private Number asNumber()
     @Override
     public boolean isNumericNull()
     {
-      return asNumber() == null;
+      return computeNumber() == null;
     }
 
     @Override
     public final boolean asBoolean()
     {
-      return Evals.asBoolean(value);
+      if (!booleanValueValid) {
+        booleanValue = Evals.asBoolean(value);
+        booleanValueValid = true;
+      }
+
+      return booleanValue;
     }
 
     @Override
     public final ExprEval castTo(ExprType castTo)
     {
       switch (castTo) {
         case DOUBLE:
-          return ExprEval.ofDouble(asNumber());
+          return ExprEval.ofDouble(computeNumber());
         case LONG:
-          return ExprEval.ofLong(asNumber());
+          return ExprEval.ofLong(computeNumber());
         case STRING:
           return this;
       }

processing/src/main/java/org/apache/druid/segment/virtual/ExpressionSelectors.java

@@ -137,12 +137,12 @@ public static ColumnValueSelector<ExprEval> makeExprEvalSelector(
       final String column = Iterables.getOnlyElement(columns);
       final ColumnCapabilities capabilities = columnSelectorFactory.getColumnCapabilities(column);
 
-      if (column.equals(ColumnHolder.TIME_COLUMN_NAME)) {
-        // Optimization for expressions that hit the __time column and nothing else.
-        // May be worth applying this optimization to all long columns?
+      if (capabilities != null && capabilities.getType() == ValueType.LONG) {
+        // Optimization for expressions that hit one long column and nothing else.
         return new SingleLongInputCachingExpressionColumnValueSelector(
-            columnSelectorFactory.makeColumnValueSelector(ColumnHolder.TIME_COLUMN_NAME),
-            expression
+            columnSelectorFactory.makeColumnValueSelector(column),
+            expression,
+            !ColumnHolder.TIME_COLUMN_NAME.equals(column) // __time doesn't need an LRU cache since it is sorted.
         );
       } else if (capabilities != null
                  && capabilities.getType() == ValueType.STRING