ESQL: Keep ordinals in conversion functions

Make the conversion functions that process `BytesRef`s into `BytesRefs`
keep the `OrdinalBytesRefVector`s when processing. Let's use `TO_LOWER`
as an example. First, the performance numbers:
```
  (operation)  Mode   Score   Error ->  Score    Error Units
     to_lower  30.662 ± 6.163 -> 30.048 ±  0.479 ns/op
to_lower_ords  30.773 ± 0.370 ->  0.025 ±  0.001 ns/op
     to_upper  33.552 ± 0.529 -> 35.775 ±  1.799 ns/op
to_upper_ords  35.791 ± 0.658 ->  0.027 ±  0.001 ns/op
```
The test has a 8192 positions containing alternating `foo` and `bar`.
Running `TO_LOWER` via ordinals is super duper faster. No longer
`O(positions)` and now `O(unique_values)`.

Let's paint some pictures! `OrdinalBytesRefVector` is a lookup table.
Like this:
```
+-------+----------+
| bytes | ordinals |
| ----- | -------- |
|  FOO  | 0        |
|  BAR  | 1        |
|  BAZ  | 2        |
+-------+ 1        |
        | 1        |
        | 0        |
        +----------+
```

That lookup table is one block. When you read it you look up the
`ordinal` and match it to the `bytes`. Previously `TO_LOWER` would
process each value one at a time and make:
```
bytes
-----
 foo
 bar
 baz
 bar
 bar
 foo
```

So it'd run `TO_LOWER` once per `ordinal` and it'd make an ordinal
non-lookup table. With this change `TO_LOWER` will now make:
```
+-------+----------+
| bytes | ordinals |
| ----- | -------- |
|  foo  | 0        |
|  bar  | 1        |
|  baz  | 2        |
+-------+ 1        |
        | 1        |
        | 0        |
        +----------+
```
We don't even have to copy the `ordinals` - we can reuse those from the
input and just bump the reference count. That's why this goes from
`O(positions)` to `O(unique_values)`.

Author: nik9000

benchmarks/src/main/java/org/elasticsearch/benchmark/compute/operator/EvalBenchmark.java

@@ -24,6 +24,7 @@
 import org.elasticsearch.compute.data.DoubleVector;
 import org.elasticsearch.compute.data.LongBlock;
 import org.elasticsearch.compute.data.LongVector;
+import org.elasticsearch.compute.data.OrdinalBytesRefVector;
 import org.elasticsearch.compute.data.Page;
 import org.elasticsearch.compute.operator.DriverContext;
 import org.elasticsearch.compute.operator.EvalOperator;
@@ -127,7 +128,9 @@ static void selfTest() {
             "mv_min_ascending",
             "rlike",
             "to_lower",
-            "to_upper" }
+            "to_lower_ords",
+            "to_upper",
+            "to_upper_ords" }
     )
     public String operation;
 
@@ -235,12 +238,12 @@ private static EvalOperator.ExpressionEvaluator evaluator(String operation) {
                 RLike rlike = new RLike(Source.EMPTY, keywordField, new RLikePattern(".ar"));
                 yield EvalMapper.toEvaluator(FOLD_CONTEXT, rlike, layout(keywordField)).get(driverContext);
             }
-            case "to_lower" -> {
+            case "to_lower", "to_lower_ords" -> {
                 FieldAttribute keywordField = keywordField();
                 ToLower toLower = new ToLower(Source.EMPTY, keywordField, configuration());
                 yield EvalMapper.toEvaluator(FOLD_CONTEXT, toLower, layout(keywordField)).get(driverContext);
             }
-            case "to_upper" -> {
+            case "to_upper", "to_upper_ords" -> {
                 FieldAttribute keywordField = keywordField();
                 ToUpper toUpper = new ToUpper(Source.EMPTY, keywordField, configuration());
                 yield EvalMapper.toEvaluator(FOLD_CONTEXT, toUpper, layout(keywordField)).get(driverContext);
@@ -414,13 +417,15 @@ private static void checkExpected(String operation, Page actual) {
                     }
                 }
             }
-            case "to_lower" -> checkBytes(operation, actual, new BytesRef[] { new BytesRef("foo"), new BytesRef("bar") });
-            case "to_upper" -> checkBytes(operation, actual, new BytesRef[] { new BytesRef("FOO"), new BytesRef("BAR") });
+            case "to_lower" -> checkBytes(operation, actual, false, new BytesRef[] { new BytesRef("foo"), new BytesRef("bar") });
+            case "to_lower_ords" -> checkBytes(operation, actual, true, new BytesRef[] { new BytesRef("foo"), new BytesRef("bar") });
+            case "to_upper" -> checkBytes(operation, actual, false, new BytesRef[] { new BytesRef("FOO"), new BytesRef("BAR") });
+            case "to_upper_ords" -> checkBytes(operation, actual, true, new BytesRef[] { new BytesRef("FOO"), new BytesRef("BAR") });
             default -> throw new UnsupportedOperationException(operation);
         }
     }
 
-    private static void checkBytes(String operation, Page actual, BytesRef[] expectedVals) {
+    private static void checkBytes(String operation, Page actual, boolean expectOrds, BytesRef[] expectedVals) {
         BytesRef scratch = new BytesRef();
         BytesRefVector v = actual.<BytesRefBlock>getBlock(1).asVector();
         for (int i = 0; i < BLOCK_LENGTH; i++) {
@@ -430,6 +435,15 @@ private static void checkBytes(String operation, Page actual, BytesRef[] expecte
                 throw new AssertionError("[" + operation + "] expected [" + expected + "] but was [" + b + "]");
             }
         }
+        if (expectOrds) {
+            if (v.asOrdinals() == null) {
+                throw new IllegalArgumentException("expected ords but got " + v);
+            }
+        } else {
+            if (v.asOrdinals() != null) {
+                throw new IllegalArgumentException("expected non-ords but got " + v);
+            }
+        }
     }
 
     private static Page page(String operation) {
@@ -510,6 +524,16 @@ private static Page page(String operation) {
                 }
                 yield new Page(builder.build().asBlock());
             }
+            case "to_lower_ords", "to_upper_ords" -> {
+                var bytes = blockFactory.newBytesRefVectorBuilder(BLOCK_LENGTH);
+                bytes.appendBytesRef(new BytesRef("foo"));
+                bytes.appendBytesRef(new BytesRef("bar"));
+                var ordinals = blockFactory.newIntVectorFixedBuilder(BLOCK_LENGTH);
+                for (int i = 0; i < BLOCK_LENGTH; i++) {
+                    ordinals.appendInt(i % 2);
+                }
+                yield new Page(new OrdinalBytesRefVector(ordinals.build(), bytes.build()).asBlock());
+            }
             default -> throw new UnsupportedOperationException();
         };
     }

x-pack/plugin/esql/compute/gen/src/main/java/org/elasticsearch/compute/gen/ConvertEvaluatorImplementer.java

@@ -28,9 +28,12 @@
 import static org.elasticsearch.compute.gen.Types.ABSTRACT_CONVERT_FUNCTION_EVALUATOR;
 import static org.elasticsearch.compute.gen.Types.BLOCK;
 import static org.elasticsearch.compute.gen.Types.BYTES_REF;
+import static org.elasticsearch.compute.gen.Types.BYTES_REF_VECTOR_BUILDER;
 import static org.elasticsearch.compute.gen.Types.DRIVER_CONTEXT;
 import static org.elasticsearch.compute.gen.Types.EXPRESSION_EVALUATOR;
 import static org.elasticsearch.compute.gen.Types.EXPRESSION_EVALUATOR_FACTORY;
+import static org.elasticsearch.compute.gen.Types.INT_VECTOR;
+import static org.elasticsearch.compute.gen.Types.ORDINALS_BYTES_REF_VECTOR;
 import static org.elasticsearch.compute.gen.Types.SOURCE;
 import static org.elasticsearch.compute.gen.Types.VECTOR;
 import static org.elasticsearch.compute.gen.Types.blockType;
@@ -41,7 +44,7 @@ public class ConvertEvaluatorImplementer {
 
     private final TypeElement declarationType;
     private final EvaluatorImplementer.ProcessFunction processFunction;
-    private final String extraName;
+    private final boolean canProcessOrdinals;
     private final ClassName implementation;
     private final TypeName argumentType;
     private final List<TypeMirror> warnExceptions;
@@ -55,6 +58,10 @@ public ConvertEvaluatorImplementer(
     ) {
         this.declarationType = (TypeElement) processFunction.getEnclosingElement();
         this.processFunction = new EvaluatorImplementer.ProcessFunction(types, processFunction, warnExceptions);
+        this.canProcessOrdinals = warnExceptions.isEmpty()
+            && this.processFunction.returnType().equals(BYTES_REF)
+            && this.processFunction.args.getFirst() instanceof EvaluatorImplementer.StandardProcessFunctionArg s
+            && s.type().equals(BYTES_REF);
 
         if (this.processFunction.args.getFirst() instanceof EvaluatorImplementer.StandardProcessFunctionArg == false) {
             throw new IllegalArgumentException("first argument must be the field to process");
@@ -66,7 +73,6 @@ public ConvertEvaluatorImplementer(
             }
         }
 
-        this.extraName = extraName;
         this.argumentType = TypeName.get(processFunction.getParameters().get(0).asType());
         this.warnExceptions = warnExceptions;
 
@@ -102,6 +108,9 @@ private TypeSpec type() {
         builder.addMethod(evalValue(true));
         builder.addMethod(evalBlock());
         builder.addMethod(evalValue(false));
+        if (canProcessOrdinals) {
+            builder.addMethod(evalOrdinals());
+        }
         builder.addMethod(processFunction.toStringMethod(implementation));
         builder.addMethod(processFunction.close());
         builder.addType(factory());
@@ -132,6 +141,15 @@ private MethodSpec evalVector() {
 
         TypeName vectorType = vectorType(argumentType);
         builder.addStatement("$T vector = ($T) v", vectorType, vectorType);
+        if (canProcessOrdinals) {
+            builder.addStatement("$T ordinals = vector.asOrdinals()", ORDINALS_BYTES_REF_VECTOR);
+            builder.beginControlFlow("if (ordinals != null)");
+            {
+                builder.addStatement("return evalOrdinals(ordinals)");
+            }
+            builder.endControlFlow();
+        }
+
         builder.addStatement("int positionCount = v.getPositionCount()");
 
         String scratchPadName = argumentType.equals(BYTES_REF) ? "scratchPad" : null;
@@ -299,6 +317,31 @@ private MethodSpec evalValue(boolean forVector) {
         return builder.build();
     }
 
+    private MethodSpec evalOrdinals() {
+        MethodSpec.Builder builder = MethodSpec.methodBuilder("evalOrdinals").addModifiers(Modifier.PRIVATE);
+        builder.addParameter(ORDINALS_BYTES_REF_VECTOR, "v").returns(BLOCK);
+
+        builder.addStatement("int positionCount = v.getDictionaryVector().getPositionCount()");
+        builder.addStatement("BytesRef scratchPad = new BytesRef()");
+        builder.beginControlFlow(
+            "try ($T builder = driverContext.blockFactory().newBytesRefVectorBuilder(positionCount))",
+            BYTES_REF_VECTOR_BUILDER
+        );
+        {
+            builder.beginControlFlow("for (int p = 0; p < positionCount; p++)");
+            {
+                builder.addStatement("builder.appendBytesRef($N)", evalValueCall("v.getDictionaryVector()", "p", "scratchPad"));
+            }
+            builder.endControlFlow();
+            builder.addStatement("$T ordinals = v.getOrdinalsVector()", INT_VECTOR);
+            builder.addStatement("ordinals.incRef()");
+            builder.addStatement("return new $T(ordinals, builder.build()).asBlock()", ORDINALS_BYTES_REF_VECTOR);
+        }
+        builder.endControlFlow();
+
+        return builder.build();
+    }
+
     private TypeSpec factory() {
         TypeSpec.Builder builder = TypeSpec.classBuilder("Factory");
         builder.addSuperinterface(EXPRESSION_EVALUATOR_FACTORY);

x-pack/plugin/esql/compute/gen/src/main/java/org/elasticsearch/compute/gen/Types.java

@@ -61,6 +61,7 @@ public class Types {
 
     static final ClassName BOOLEAN_VECTOR = ClassName.get(DATA_PACKAGE, "BooleanVector");
     static final ClassName BYTES_REF_VECTOR = ClassName.get(DATA_PACKAGE, "BytesRefVector");
+    static final ClassName ORDINALS_BYTES_REF_VECTOR = ClassName.get(DATA_PACKAGE, "OrdinalBytesRefVector");
     static final ClassName INT_VECTOR = ClassName.get(DATA_PACKAGE, "IntVector");
     static final ClassName LONG_VECTOR = ClassName.get(DATA_PACKAGE, "LongVector");
     static final ClassName DOUBLE_VECTOR = ClassName.get(DATA_PACKAGE, "DoubleVector");