ESQL: Add key names to description of hash lookup (elastic#108012)

When I was debugging further work on our new hash stuff I found myself
really wanting the key names on the `description` and `toString` of the
hash lookup operator. This adds it.

x-pack/plugin/esql/compute/src/main/java/org/elasticsearch/compute/operator/HashLookupOperator.java

@@ -23,50 +23,50 @@
 import java.util.List;
 
 public class HashLookupOperator extends AbstractPageMappingToIteratorOperator {
+    public record Key(String name, Block block) {
+        @Override
+        public String toString() {
+            return "{name="
+                + name
+                + ", type="
+                + block.elementType()
+                + ", positions="
+                + block.getPositionCount()
+                + ", size="
+                + ByteSizeValue.ofBytes(block.ramBytesUsed())
+                + "}";
+        }
+    }
+
     /**
      * Factory for {@link HashLookupOperator}. It's received {@link Block}s
      * are never closed, so we need to build them from a non-tracking factory.
      */
-    public static class Factory implements Operator.OperatorFactory {
-        private final Block[] keys;
-        private final int[] blockMapping;
-
-        public Factory(Block[] keys, int[] blockMapping) {
-            this.keys = keys;
-            this.blockMapping = blockMapping;
-        }
-
+    public record Factory(Key[] keys, int[] blockMapping) implements Operator.OperatorFactory {
         @Override
         public Operator get(DriverContext driverContext) {
             return new HashLookupOperator(driverContext.blockFactory(), keys, blockMapping);
         }
 
         @Override
         public String describe() {
-            StringBuilder b = new StringBuilder();
-            b.append("HashLookup[keys=[");
-            for (int k = 0; k < keys.length; k++) {
-                Block key = keys[k];
-                if (k != 0) {
-                    b.append(", ");
-                }
-                b.append("{type=").append(key.elementType());
-                b.append(", positions=").append(key.getPositionCount());
-                b.append(", size=").append(ByteSizeValue.ofBytes(key.ramBytesUsed())).append("}");
-            }
-            b.append("], mapping=").append(Arrays.toString(blockMapping)).append("]");
-            return b.toString();
+            return "HashLookup[keys=" + Arrays.toString(keys) + ", mapping=" + Arrays.toString(blockMapping) + "]";
         }
     }
 
+    private final List<String> keys;
     private final BlockHash hash;
     private final int[] blockMapping;
 
-    public HashLookupOperator(BlockFactory blockFactory, Block[] keys, int[] blockMapping) {
+    public HashLookupOperator(BlockFactory blockFactory, Key[] keys, int[] blockMapping) {
         this.blockMapping = blockMapping;
+        this.keys = new ArrayList<>(keys.length);
+        Block[] blocks = new Block[keys.length];
         List<BlockHash.GroupSpec> groups = new ArrayList<>(keys.length);
         for (int k = 0; k < keys.length; k++) {
-            groups.add(new BlockHash.GroupSpec(k, keys[k].elementType()));
+            this.keys.add(keys[k].name);
+            blocks[k] = keys[k].block;
+            groups.add(new BlockHash.GroupSpec(k, keys[k].block.elementType()));
         }
         /*
          * Force PackedValuesBlockHash because it assigned ordinals in order
@@ -83,7 +83,7 @@ public HashLookupOperator(BlockFactory blockFactory, Block[] keys, int[] blockMa
         boolean success = false;
         try {
             final int[] lastOrd = new int[] { -1 };
-            hash.add(new Page(keys), new GroupingAggregatorFunction.AddInput() {
+            hash.add(new Page(blocks), new GroupingAggregatorFunction.AddInput() {
                 @Override
                 public void add(int positionOffset, IntBlock groupIds) {
                     // TODO support multiple rows with the same keys
@@ -128,7 +128,7 @@ protected ReleasableIterator<Page> receive(Page page) {
 
     @Override
     public String toString() {
-        return "HashLookup[hash=" + hash + ", mapping=" + Arrays.toString(blockMapping) + "]";
+        return "HashLookup[keys=" + keys + ", hash=" + hash + ", mapping=" + Arrays.toString(blockMapping) + "]";
     }
 
     @Override

x-pack/plugin/esql/compute/src/test/java/org/elasticsearch/compute/OperatorTests.java

@@ -370,7 +370,13 @@ public void testHashLookup() {
                 var driver = new Driver(
                     driverContext,
                     new SequenceLongBlockSourceOperator(driverContext.blockFactory(), values, 100),
-                    List.of(new HashLookupOperator(driverContext.blockFactory(), new Block[] { primesBlock }, new int[] { 0 })),
+                    List.of(
+                        new HashLookupOperator(
+                            driverContext.blockFactory(),
+                            new HashLookupOperator.Key[] { new HashLookupOperator.Key("primes", primesBlock) },
+                            new int[] { 0 }
+                        )
+                    ),
                     new PageConsumerOperator(page -> {
                         try {
                             BlockTestUtils.readInto(actualValues, page.getBlock(0));

x-pack/plugin/esql/compute/src/test/java/org/elasticsearch/compute/operator/HashLookupOperatorTests.java

@@ -7,7 +7,6 @@
 
 package org.elasticsearch.compute.operator;
 
-import org.elasticsearch.compute.data.Block;
 import org.elasticsearch.compute.data.BlockFactory;
 import org.elasticsearch.compute.data.Page;
 import org.elasticsearch.compute.data.TestBlockFactory;
@@ -31,18 +30,22 @@ protected void assertSimpleOutput(List<Page> input, List<Page> results) {
     @Override
     protected Operator.OperatorFactory simple() {
         return new HashLookupOperator.Factory(
-            new Block[] { TestBlockFactory.getNonBreakingInstance().newLongArrayVector(new long[] { 7, 14, 20 }, 3).asBlock() },
+            new HashLookupOperator.Key[] {
+                new HashLookupOperator.Key(
+                    "foo",
+                    TestBlockFactory.getNonBreakingInstance().newLongArrayVector(new long[] { 7, 14, 20 }, 3).asBlock()
+                ) },
             new int[] { 0 }
         );
     }
 
     @Override
     protected String expectedDescriptionOfSimple() {
-        return "HashLookup[keys=[{type=LONG, positions=3, size=96b}], mapping=[0]]";
+        return "HashLookup[keys=[{name=foo, type=LONG, positions=3, size=96b}], mapping=[0]]";
     }
 
     @Override
     protected String expectedToStringOfSimple() {
-        return "HashLookup[hash=PackedValuesBlockHash{groups=[0:LONG], entries=3, size=536b}, mapping=[0]]";
+        return "HashLookup[keys=[foo], hash=PackedValuesBlockHash{groups=[0:LONG], entries=3, size=536b}, mapping=[0]]";
     }
 }