[Analytics Backend / DataFusion] Reshape mvappend operands as uniform lists; add Decimal128 element support

Two follow-ons to the initial mvappend onboarding (40b2161), both surfaced
once the SQL companion #5424 (`MVAppendFunctionImpl.leastRestrictive`) let
homogeneous-type calls reach substrait conversion.

# Uniform-list operand reshape

Substrait's variadic-`any1` argument shape requires every operand at the same
variadic position to share a type. PPL's `mvappend(arg, …)` accepts a mix of
bare scalars and arrays, which substrait's signature matcher rejected with
`Unable to convert call mvappend(list<i32?>, i32?, i32?)`.

`MvappendAdapter` now wraps each scalar operand in a singleton
`make_array(scalar)` call (using the locally-declared `MakeArrayAdapter.LOCAL_MAKE_ARRAY_OP`)
so by the time the substrait converter sees the operands they're uniformly
`list<componentType>`. The yaml impl was correspondingly tightened from
`args: [{ value: any1 }] variadic` to `args: [{ value: list<any1?> }] variadic`.

Rust UDF (`udf::mvappend`) keeps its scalar-handling branch intact as a
defensive fallback, but in practice every operand it sees is a list now.

# Decimal128 element type

Calcite's leastRestrictive widening on INT + DECIMAL produces DECIMAL(p, s)
which substrait converts to Decimal128(p, s); the Java adapter casts every
operand's element type to that. The Rust UDF needed an explicit
`DataType::Decimal128(p, s)` branch — Decimal128Builder requires
`.with_precision_and_scale(p, s)` configuration before use, and Decimal128Array
elements are read via the `i128`-valued `value(i)` accessor (not via the
generic `build!` macro).

# Pass-rate (CalciteMVAppendFunctionIT, force-routed, with companion #5424 applied)

  * Before this commit: 6/15 (initial mvappend onboarding).
  * After this commit:  10/15.

  Newly passing:
    testMvappendWithMixedArrayAndScalar (uniform-list reshape),
    testMvappendWithComplexExpression (uniform-list reshape),
    testMvappendWithIntAndDouble (Decimal128 element),
    testMvappendWithNumericArrays (Decimal128 element).

  Remaining 5 failures:
    * testMvappendWithMixedTypes / WithFieldsAndLiterals / WithEmptyArray /
      WithNull — call legitimately widens to ARRAY<ANY> because operands
      contain pairs of types with no common widened type (INT + VARCHAR).
      The Calcite engine handles ANY via Object generic dispatch; substrait
      can't encode it. Out of scope without changing PPL UDF semantics.
    * testMvappendInWhereClause — uses `where array_length(combined) = 2`
      which the analytics-engine planner rejects with "No backend can
      evaluate filter predicate [EQUALS] on fields [combined:ARRAY]".
      Filter-side capability gap unrelated to mvappend.

Signed-off-by: Kai Huang <ahkcs@amazon.com>

Author: ahkcs

sandbox/plugins/analytics-backend-datafusion/rust/src/udf/mvappend.rs

@@ -36,11 +36,12 @@ use std::any::Any;
 use std::sync::Arc;
 
 use datafusion::arrow::array::{
-    Array, ArrayRef, AsArray, BooleanArray, BooleanBuilder, Float32Array, Float32Builder,
-    Float64Array, Float64Builder, GenericListArray, Int16Array, Int16Builder, Int32Array,
-    Int32Builder, Int64Array, Int64Builder, Int8Array, Int8Builder, ListArray, ListBuilder,
-    StringArray, StringBuilder, StringViewArray, StringViewBuilder, UInt16Array, UInt16Builder,
-    UInt32Array, UInt32Builder, UInt64Array, UInt64Builder, UInt8Array, UInt8Builder,
+    Array, ArrayRef, AsArray, BooleanArray, BooleanBuilder, Decimal128Array, Decimal128Builder,
+    Float32Array, Float32Builder, Float64Array, Float64Builder, GenericListArray, Int16Array,
+    Int16Builder, Int32Array, Int32Builder, Int64Array, Int64Builder, Int8Array, Int8Builder,
+    ListArray, ListBuilder, StringArray, StringBuilder, StringViewArray, StringViewBuilder,
+    UInt16Array, UInt16Builder, UInt32Array, UInt32Builder, UInt64Array, UInt64Builder,
+    UInt8Array, UInt8Builder,
 };
 use datafusion::arrow::datatypes::{DataType, Field};
 use datafusion::common::plan_err;
@@ -182,6 +183,54 @@ impl ScalarUDFImpl for MvappendUdf {
             DataType::Float32 => build!(Float32Builder, Float32Array, ListArray),
             DataType::Float64 => build!(Float64Builder, Float64Array, ListArray),
             DataType::Boolean => build!(BooleanBuilder, BooleanArray, ListArray),
+            // Decimal128 element type — needs a builder configured with the same precision
+            // and scale as the input. Calcite's leastRestrictive widening for INT + DECIMAL
+            // produces DECIMAL(p, s) which substrait converts to Decimal128(p, s); the Java
+            // adapter's CAST aligns every operand's element type to that.
+            DataType::Decimal128(precision, scale) => {
+                let inner = Decimal128Builder::new()
+                    .with_precision_and_scale(*precision, *scale)
+                    .map_err(|e| DataFusionError::Plan(format!("mvappend: decimal builder: {e}")))?;
+                let mut builder = ListBuilder::new(inner);
+                for row in 0..n {
+                    let mut any_value = false;
+                    for arr in &operand_arrays {
+                        if arr.is_null(row) {
+                            continue;
+                        }
+                        if let Some(list_arr) = arr.as_any().downcast_ref::<GenericListArray<i32>>() {
+                            let row_list = list_arr.value(row);
+                            let typed = row_list
+                                .as_any()
+                                .downcast_ref::<Decimal128Array>()
+                                .ok_or_else(|| DataFusionError::Internal(format!(
+                                    "mvappend: list element vector type mismatch ({:?})",
+                                    row_list.data_type()
+                                )))?;
+                            for i in 0..typed.len() {
+                                if !typed.is_null(i) {
+                                    builder.values().append_value(typed.value(i));
+                                    any_value = true;
+                                }
+                            }
+                        } else if let Some(typed) = arr.as_any().downcast_ref::<Decimal128Array>() {
+                            builder.values().append_value(typed.value(row));
+                            any_value = true;
+                        } else {
+                            return plan_err!(
+                                "mvappend: unexpected operand vector type {:?}",
+                                arr.data_type()
+                            );
+                        }
+                    }
+                    if any_value {
+                        builder.append(true);
+                    } else {
+                        builder.append_null();
+                    }
+                }
+                Arc::new(builder.finish()) as ArrayRef
+            }
             // String element types — handled specially because list children may be any of
             // {Utf8, LargeUtf8, Utf8View} depending on whether the operand is a string literal,
             // a field read (DataFusion's substrait consumer uses Utf8View for column reads in

sandbox/plugins/analytics-backend-datafusion/src/main/java/org/opensearch/be/datafusion/MvappendAdapter.java

@@ -65,29 +65,31 @@ public RexNode adapt(RexCall original, List<FieldStorageInfo> fieldStorage, RelO
         RelDataType arrayType = original.getType();
         RelDataType componentType = arrayType.getComponentType();
         if (componentType == null) {
-            // Defensive — Calcite always assigns ARRAY<X> as mvappend's return type. Pass
-            // operands through untouched and let substrait surface the error if anything's
-            // off.
             return rexBuilder.makeCall(arrayType, LOCAL_MVAPPEND_OP, original.getOperands());
         }
+        // Substrait's variadic {@code any1} parameter requires every operand at the same
+        // variadic position to share a type. PPL's {@code mvappend(arg, …)} accepts a mix
+        // of bare scalars and arrays, which substrait's signature matcher rejects with
+        // {@code Unable to convert call mvappend(list<…>, scalar, …)}. Normalize every
+        // operand to {@code ARRAY<componentType>} — array operands cast their element
+        // type if it differs; scalar operands wrap in a {@code make_array(…)} singleton
+        // call. The Rust UDF then sees a uniform {@code list<any1>} variadic.
+        RelDataType targetArrayType = cluster.getTypeFactory().createArrayType(componentType, -1);
         List<RexNode> coerced = new ArrayList<>(original.getOperands().size());
         for (RexNode operand : original.getOperands()) {
             RelDataType operandType = operand.getType();
             if (operandType.getComponentType() != null) {
                 // Array operand — cast to ARRAY<componentType> if its element type differs.
-                RelDataType targetArray = cluster.getTypeFactory().createArrayType(componentType, -1);
-                if (operandType.equals(targetArray)) {
+                if (operandType.equals(targetArrayType)) {
                     coerced.add(operand);
                 } else {
-                    coerced.add(rexBuilder.makeCast(targetArray, operand, true, false));
+                    coerced.add(rexBuilder.makeCast(targetArrayType, operand, true, false));
                 }
             } else {
-                // Scalar operand — cast to componentType if its type differs.
-                if (operandType.equals(componentType)) {
-                    coerced.add(operand);
-                } else {
-                    coerced.add(rexBuilder.makeCast(componentType, operand, true, false));
-                }
+                // Scalar operand — first cast to componentType (so the singleton array's
+                // element type matches), then wrap in make_array so substrait sees a list.
+                RexNode casted = operandType.equals(componentType) ? operand : rexBuilder.makeCast(componentType, operand, true, false);
+                coerced.add(rexBuilder.makeCall(targetArrayType, MakeArrayAdapter.LOCAL_MAKE_ARRAY_OP, List.of(casted)));
             }
         }
         return rexBuilder.makeCall(arrayType, LOCAL_MVAPPEND_OP, coerced);

sandbox/plugins/analytics-backend-datafusion/src/main/resources/opensearch_array_functions.yaml

@@ -85,15 +85,17 @@ scalar_functions:
 
   - name: mvappend
     description: >-
-      Flatten a mixed list of array and scalar arguments into one array,
-      dropping null arguments and null elements within array arguments. Returns
-      NULL if no non-null elements were collected. PPL surface is
-      {@code mvappend(arg1, arg2, …)}; backed by a custom Rust UDF on the
-      analytics-backend-datafusion plugin (DataFusion's array_concat only
-      accepts arrays and preserves nulls — different semantics).
+      Flatten a list of arrays into one array, dropping null arrays and null
+      elements within array arguments. Returns NULL if no non-null elements
+      were collected. PPL surface is {@code mvappend(arg1, arg2, …)} which
+      accepts mixed scalar+array operands; the Java adapter wraps each
+      scalar in a singleton {@code make_array(…)} call so by the time the
+      Rust UDF sees the operands they're uniformly arrays. Backed by a custom
+      Rust UDF on the analytics-backend-datafusion plugin (DataFusion's
+      array_concat preserves nulls — different semantics).
     impls:
       - args:
-          - value: "any1"
+          - value: "list<any1?>"
             name: arg
         variadic:
           min: 1