Add enhancement for MV_APPEND supports 2-n number of arguments

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

@@ -210,6 +210,17 @@ private MethodSpec realEval(boolean blockStyle) {
                                     builder.addStatement("allBlocksAreNulls = false");
                                 }
                                 builder.endControlFlow();
+                            } else if (a instanceof ArrayProcessFunctionArg aa) {
+                                builder.beginControlFlow("for (int i = 0; i < $L.length; i++)", aa.paramName(true));
+                                {
+                                    builder.beginControlFlow("if (!$L[i].isNull(p))", aa.paramName(true));
+                                    {
+                                        builder.addStatement("allBlocksAreNulls = false");
+                                    }
+                                    builder.endControlFlow();
+                                }
+                                builder.endControlFlow();
+
                             }
                         });
 
@@ -671,6 +682,12 @@ public void resolveVectors(MethodSpec.Builder builder, String invokeBlockEval) {
 
         @Override
         public void createScratch(MethodSpec.Builder builder) {
+            if (componentType.equals(INT_BLOCK) || componentType.equals(LONG_BLOCK) || componentType.equals(DOUBLE_BLOCK)
+                || componentType.equals(BOOLEAN_BLOCK) || componentType.equals(BYTES_REF_BLOCK)) {
+                // if componentType is a block, it is a multi-value field, we just assign the block
+                builder.addStatement("$T[] $LValues = $L", componentType, name, paramName(true));
+                return;
+            }
             builder.addStatement("$T[] $LValues = new $T[$L.length]", componentType, name, componentType, name);
             if (componentType.equals(BYTES_REF)) {
                 builder.addStatement("$T[] $LScratch = new $T[$L.length]", componentType, name, componentType, name);
@@ -689,6 +706,12 @@ public void skipNull(MethodSpec.Builder builder) {
 
         @Override
         public void unpackValues(MethodSpec.Builder builder, boolean blockStyle) {
+            if (componentType.equals(INT_BLOCK) || componentType.equals(LONG_BLOCK) || componentType.equals(DOUBLE_BLOCK)
+                || componentType.equals(BOOLEAN_BLOCK) || componentType.equals(BYTES_REF_BLOCK)) {
+                // if componentType is a block, we don't need to unpack it, it is a multi-value field
+                // nothing to do
+                return;
+            }
             builder.addComment("unpack $L into $LValues", paramName(blockStyle), name);
             builder.beginControlFlow("for (int i = 0; i < $L.length; i++)", paramName(blockStyle));
             String lookupVar;

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

@@ -120,7 +120,6 @@ public class Types {
     static final ClassName SOURCE = ClassName.get("org.elasticsearch.xpack.esql.core.tree", "Source");
 
     static final ClassName BYTES_REF = ClassName.get("org.apache.lucene.util", "BytesRef");
-
     static final ClassName RELEASABLE = ClassName.get("org.elasticsearch.core", "Releasable");
     static final ClassName RELEASABLES = ClassName.get("org.elasticsearch.core", "Releasables");
 
@@ -137,8 +136,14 @@ public static TypeDef of(TypeName type, String alias, String block, String vecto
         TypeDef.of(TypeName.LONG, "LONG", "LongBlock", "LongVector"),
         TypeDef.of(TypeName.FLOAT, "FLOAT", "FloatBlock", "FloatVector"),
         TypeDef.of(TypeName.DOUBLE, "DOUBLE", "DoubleBlock", "DoubleVector"),
-        TypeDef.of(BYTES_REF, "BYTES_REF", "BytesRefBlock", "BytesRefVector")
-    )
+        TypeDef.of(BYTES_REF, "BYTES_REF", "BytesRefBlock", "BytesRefVector"),
+        TypeDef.of(BOOLEAN_BLOCK, "BOOLEAN_BLOCK", "BooleanBlock", "BooleanVector"),
+        TypeDef.of(INT_BLOCK, "INT_BLOCK", "IntBlock", "IntVector"),
+        TypeDef.of(LONG_BLOCK, "LONG_BLOCK", "LongBlock", "LongVector"),
+        TypeDef.of(FLOAT_BLOCK, "FLOAT_BLOCK", "FloatBlock", "FloatVector"),
+        TypeDef.of(DOUBLE_BLOCK, "DOUBLE_BLOCK", "DoubleBlock", "DoubleVector"),
+        TypeDef.of(BYTES_REF_BLOCK, "BYTES_REF_BLOCK", "BytesRefBlock", "BytesRefVector")
+        )
         .flatMap(def -> Stream.of(def.type.toString(), def.type + "[]", def.alias).map(alias -> Map.entry(alias, def)))
         .collect(toUnmodifiableMap(Map.Entry::getKey, Map.Entry::getValue));
 

x-pack/plugin/esql/qa/testFixtures/src/main/resources/floats.csv-spec

@@ -536,12 +536,12 @@ FROM employees
        i = mv_append(salary_change.int, salary_change.int),
        i2 = mv_append(emp_no, salary_change.int),
        i3 = mv_append(emp_no, emp_no),  
-       s = mv_append(salary_change.keyword, salary_change.keyword)
+       s = mv_append(salary_change.keyword, salary_change.keyword, salary_change.keyword)
 | KEEP emp_no, salary_change, d, i, i2, i3, s
 | SORT emp_no;
 
 emp_no:integer | salary_change:double    | d:double                                       | i:integer              | i2:integer         | i3:integer     | s:keyword
-10008          | [-2.92,0.75,3.54,12.68] | [-2.92,0.75,3.54,12.68,-2.92,0.75,3.54,12.68]  | [-2,0,3,12,-2,0,3,12]  | [10008,-2,0,3,12]  | [10008, 10008] | [-2.92,0.75,12.68,3.54,-2.92,0.75,12.68,3.54]
+10008          | [-2.92,0.75,3.54,12.68] | [-2.92,0.75,3.54,12.68,-2.92,0.75,3.54,12.68]  | [-2,0,3,12,-2,0,3,12]  | [10008,-2,0,3,12]  | [10008, 10008] | [-2.92,0.75,12.68,3.54,-2.92,0.75,12.68,3.54,-2.92,0.75,12.68,3.54]
 10021          | null                    | null                                           | null                   | null               | [10021, 10021] | null
 ;
 

x-pack/plugin/esql/qa/testFixtures/src/main/resources/string.csv-spec

@@ -1725,7 +1725,7 @@ mvAppend
 required_capability: fn_mv_append
 
 ROW a = "a", b = ["b", "c"], n = null
-| EVAL aa = mv_append(a, a), bb = mv_append(b, b), ab = mv_append(a, b), abb = mv_append(mv_append(a, b), b), na = mv_append(n, a), an = mv_append(a, n)
+| EVAL aa = mv_append(a, a), bb = mv_append(b, b), ab = mv_append(a, b), abb = mv_append(a, b, b), na = mv_append(n, a), an = mv_append(a, n)
 ;
 
 a:keyword | b:keyword | n:null | aa:keyword | bb:keyword   | ab:keyword | abb:keyword    | na:keyword | an:keyword

x-pack/plugin/esql/qa/testFixtures/src/main/resources/union_types.csv-spec

@@ -1578,7 +1578,7 @@ FROM sample_data, sample_data_ts_long
 multiIndexIndirectUseOfUnionTypesInMvExpand
 required_capability: union_types
 FROM sample_data, sample_data_ts_long
-| EVAL foo = MV_APPEND(message, message)
+| EVAL foo = MV_APPEND(message, message, message)
 | SORT client_ip ASC
 | LIMIT 1
 | MV_EXPAND foo
@@ -1587,6 +1587,7 @@ FROM sample_data, sample_data_ts_long
 @timestamp:unsupported | client_ip:ip | event_duration:long | message:keyword | foo:keyword
            null        | 172.21.0.5   |             1232382 | Disconnected    | Disconnected
            null        | 172.21.0.5   |             1232382 | Disconnected    | Disconnected
+           null        | 172.21.0.5   |             1232382 | Disconnected    | Disconnected
 ;
 
 shortIntegerWidening

x-pack/plugin/esql/qa/testFixtures/src/main/resources/version.csv-spec

@@ -377,7 +377,7 @@ required_capability: fn_mv_append
 
 ROW a = to_version("1.2.0"), x1 = to_version("0.0.1"), x2 = to_version("1.0.0")
 | EVAL b = mv_append(x1, x2)
-| EVAL aa = mv_append(a, a), bb = mv_append(b, b), ab = mv_append(a, b), abb = mv_append(mv_append(a, b), b)
+| EVAL aa = mv_append(a, a), bb = mv_append(b, b), ab = mv_append(a, b), abb = mv_append(a, b, b)
 | KEEP a, b, aa, bb, ab, abb
 ;