[ES|QL] Support for First/Last (#137408) - Code Gen & ALL_LAST (#138066)

- Minor refactoring in "build.gradle" to allow the generation of the two variants of state classes (e.g.: LongLongState and AllLongLongState)
- Generate the new two state files using "All-X-2State.java.st", for all types.
- Generate the new timestamp aggregator files using "All-X-ValueByTimestampAggregator.java.st", for all types.
- Register ALL_LAST as snapshot function
- Augment existing csv tests to exercise ALL_LAST as well, and to check for all supported types.

Author: mouhc1ne

x-pack/plugin/esql/compute/build.gradle

@@ -496,22 +496,26 @@ tasks.named('stringTemplates').configure {
    * Generates pairwise states. We generate the ones that we need at the moment,
    * but add more if you need more.
    */
-  File twoStateInputFile = file("src/main/java/org/elasticsearch/compute/aggregation/X-2State.java.st")
-  [longProperties].forEach { v1 ->
-    [intProperties, longProperties, floatProperties, doubleProperties, bytesRefProperties].forEach { v2 ->
-      {
-        var properties = [:]
-        v1.forEach { k, v -> properties["v1_" + k] = v}
-        v2.forEach { k, v -> properties["v2_" + k] = v}
+  def generateTwoStateFiles = { inputFile, prefix = "" ->
+    def v1Props = [longProperties]
+    def v2Props = [intProperties, longProperties, floatProperties, doubleProperties, bytesRefProperties]
+    v1Props.forEach { v1 ->
+      v2Props.forEach { v2 ->
+        def properties = [:]
+        v1.forEach { k, v -> properties["v1_" + k] = v }
+        v2.forEach { k, v -> properties["v2_" + k] = v }
         template {
           it.properties = properties
-          it.inputFile = twoStateInputFile
-          it.outputFile = "org/elasticsearch/compute/aggregation/${v1.Type}${v2.Type}State.java"
+          it.inputFile = inputFile
+          it.outputFile = "org/elasticsearch/compute/aggregation/${prefix}${v1.Type}${v2.Type}State.java"
         }
       }
     }
   }
 
+  generateTwoStateFiles(file("src/main/java/org/elasticsearch/compute/aggregation/X-2State.java.st"))
+  generateTwoStateFiles(file("src/main/java/org/elasticsearch/compute/aggregation/X-All2State.java.st"), "All")
+
   File irateAggregatorInputFile = file("src/main/java/org/elasticsearch/compute/aggregation/X-IrateAggregator.java.st")
   template {
     it.properties = intProperties
@@ -958,37 +962,42 @@ tasks.named('stringTemplates').configure {
   }
 
   // TODO: add {value}_over_time for other types: boolean, bytes_refs
-  File valueByTimestampAggregatorInputFile = file("src/main/java/org/elasticsearch/compute/aggregation/X-ValueByTimestampAggregator.java.st")
-  ["First", "Last"].forEach { Occurrence ->
-    {
-      template {
-        it.properties = addOccurrence(intProperties, Occurrence)
-        it.inputFile = valueByTimestampAggregatorInputFile
-        it.outputFile = "org/elasticsearch/compute/aggregation/${Occurrence}IntByTimestampAggregator.java"
-      }
-      template {
-        it.properties = addOccurrence(longProperties, Occurrence)
-        it.inputFile = valueByTimestampAggregatorInputFile
-        it.outputFile = "org/elasticsearch/compute/aggregation/${Occurrence}LongByTimestampAggregator.java"
-      }
-      template {
-        it.properties = addOccurrence(floatProperties, Occurrence)
-        it.inputFile = valueByTimestampAggregatorInputFile
-        it.outputFile = "org/elasticsearch/compute/aggregation/${Occurrence}FloatByTimestampAggregator.java"
-      }
-      template {
-        it.properties = addOccurrence(doubleProperties, Occurrence)
-        it.inputFile = valueByTimestampAggregatorInputFile
-        it.outputFile = "org/elasticsearch/compute/aggregation/${Occurrence}DoubleByTimestampAggregator.java"
-      }
-      template {
-        it.properties = addOccurrence(bytesRefProperties, Occurrence)
-        it.inputFile = valueByTimestampAggregatorInputFile
-        it.outputFile = "org/elasticsearch/compute/aggregation/${Occurrence}BytesRefByTimestampAggregator.java"
+  def generateTimestampAggregatorClasses = { inputFilename, prefix = "" ->
+    def inputFile = file(inputFilename)
+    ["First", "Last"].forEach { Occurrence ->
+      {
+        template {
+          it.properties = addOccurrence(intProperties, Occurrence) + [Prefix: prefix]
+          it.inputFile = inputFile
+          it.outputFile = "org/elasticsearch/compute/aggregation/${prefix}${Occurrence}IntByTimestampAggregator.java"
+        }
+        template {
+          it.properties = addOccurrence(longProperties, Occurrence) + [Prefix: prefix]
+          it.inputFile = inputFile
+          it.outputFile = "org/elasticsearch/compute/aggregation/${prefix}${Occurrence}LongByTimestampAggregator.java"
+        }
+        template {
+          it.properties = addOccurrence(floatProperties, Occurrence) + [Prefix: prefix]
+          it.inputFile = inputFile
+          it.outputFile = "org/elasticsearch/compute/aggregation/${prefix}${Occurrence}FloatByTimestampAggregator.java"
+        }
+        template {
+          it.properties = addOccurrence(doubleProperties, Occurrence) + [Prefix: prefix]
+          it.inputFile = inputFile
+          it.outputFile = "org/elasticsearch/compute/aggregation/${prefix}${Occurrence}DoubleByTimestampAggregator.java"
+        }
+        template {
+          it.properties = addOccurrence(bytesRefProperties, Occurrence) + [Prefix: prefix]
+          it.inputFile = inputFile
+          it.outputFile = "org/elasticsearch/compute/aggregation/${prefix}${Occurrence}BytesRefByTimestampAggregator.java"
+        }
       }
     }
   }
 
+  generateTimestampAggregatorClasses("src/main/java/org/elasticsearch/compute/aggregation/X-ValueByTimestampAggregator.java.st", "")
+  generateTimestampAggregatorClasses("src/main/java/org/elasticsearch/compute/aggregation/X-AllValueByTimestampAggregator.java.st", "All")
+
   File rateAggregatorInputFile = file("src/main/java/org/elasticsearch/compute/aggregation/X-RateGroupingAggregatorFunction.java.st")
   template {
     it.properties = intProperties

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

@@ -21,6 +21,7 @@
 import static org.elasticsearch.compute.gen.Types.DOUBLE_BLOCK;
 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.FLOAT_BLOCK;
 import static org.elasticsearch.compute.gen.Types.INT_BLOCK;
 import static org.elasticsearch.compute.gen.Types.LONG_BLOCK;
 import static org.elasticsearch.compute.gen.Types.blockType;
@@ -120,6 +121,7 @@ static boolean isBlockType(TypeName type) {
         return type.equals(INT_BLOCK)
             || type.equals(LONG_BLOCK)
             || type.equals(DOUBLE_BLOCK)
+            || type.equals(FLOAT_BLOCK)
             || type.equals(BOOLEAN_BLOCK)
             || type.equals(BYTES_REF_BLOCK);
     }