[Analytics Engine] Carry array-typed cells through RowResponseCodec without JSON-stringifying

The row-oriented fragment-execution wire format (`FragmentExecutionResponse`,
used when arrow-flight streaming is disabled — every single-node test cluster
today) shipped each cell through OpenSearch's `writeGenericValue` /
`readGenericValue`, which preserves `List` values as `ArrayList<Object>`. On
the coordinator side, `RowResponseCodec.decode` then re-materialized the rows
into a `VectorSchemaRoot` for `Iterable<VectorSchemaRoot>`-style consumers.

Two bugs in that re-materialization were eating array values:

1. `inferArrowType` walked rows for the first non-null cell and matched
   against {Long, Integer, …, CharSequence, byte[], Number}. {@code List}
   wasn't in the chain, so it fell through to {@code break} and the
   fallback {@link ArrowType.Utf8} — every array column became a VARCHAR
   column.
2. `setVectorValue` for {@link VarCharVector} called {@code value.toString()}.
   For a {@code JsonStringArrayList} that returns the JSON form
   {@code "[2,3,4]"}, which then got serialized as a JSON string in the
   final response. Tests like {@code testMvindexRangePositive} saw their
   array result come back as a string `"[2,3,4]"` instead of an array
   `[2, 3, 4]`.

Fix:

* Replace {@code inferArrowType} with {@code inferField} that returns a
  full {@link Field}. For {@code List} cells, build a list field with the
  inner element type inferred from the first non-null element (with a
  fallback that scans later rows in case the first list is empty/all-null).
* Add a {@code ListVector} arm to {@code setVectorValue} that delegates to
  a new {@code writeListValue}. The writer bypasses {@link UnionListWriter}
  entirely — it writes directly to the list's offset / validity buffers and
  to the inner data vector via the inner vector's typed `setSafe`. The
  writer-based API requires per-element `ArrowBuf` allocations for varchar
  elements that are easy to leak or use-after-free; the direct path is
  simpler and avoids both classes of bug.

Plus a separate Arrow gotcha that surfaced once arrays started flowing
through correctly:

* {@code ListVector.getObject} for a {@code VarCharVector} child returns a
  {@code JsonStringArrayList} whose elements are Arrow's {@link Text} class,
  not Java {@link String}. {@code ExprValueUtils.fromObjectValue} doesn't
  recognize {@code Text} and threw "unsupported object class
  org.apache.arrow.vector.util.Text". {@code ArrowValues.toJavaValue} now
  mirrors its top-level VarChar branch for list cells: when a list value
  comes back from a {@code ListVector}, normalize each {@code Text} element
  to a {@link String} before handing the list upward.

  * Before: 12/60 (mvindex range tests still showed expected-vs-actual
    diff because `[2,3,4]` came back as a JSON string, not an array).
  * After:  26/60.

  Newly passing:
    testMvindexRangePositive, testMvindexRangeNegative, testMvindexRangeMixed,
    testMvindexRangeFirstThree, testMvindexRangeLastThree,
    testMvindexRangeSingleElement,
    testMvdedupWithDuplicates, testMvdedupWithAllDuplicates,
    testMvdedupWithNoDuplicates, testMvdedupWithStrings,
    testArrayWithString,
    testSplitWithSemicolonDelimiter, testSplitWithMultiCharDelimiter,
    testSplitWithEmptyDelimiter.

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

Author: ahkcs

sandbox/plugins/analytics-engine/src/main/java/org/opensearch/analytics/exec/ArrowValues.java

@@ -10,9 +10,12 @@
 
 import org.apache.arrow.vector.FieldVector;
 import org.apache.arrow.vector.VarCharVector;
+import org.apache.arrow.vector.complex.ListVector;
 import org.apache.arrow.vector.util.Text;
 
 import java.nio.charset.StandardCharsets;
+import java.util.ArrayList;
+import java.util.List;
 
 /**
  * Helpers for reading Arrow vector cells as plain Java values at the
@@ -35,10 +38,22 @@ public static Object toJavaValue(FieldVector vector, int index) {
         if (vector instanceof VarCharVector v) {
             return new String(v.get(index), StandardCharsets.UTF_8);
         }
-        Object obj = vector.getObject(index);
-        if (obj instanceof Text t) {
+        Object value = vector.getObject(index);
+        if (vector instanceof ListVector && value instanceof List<?> raw) {
+            // ListVector.getObject returns a JsonStringArrayList whose elements are the
+            // child vector's typed values. For VarCharVector children that's Arrow's
+            // Text, which downstream consumers (e.g. {@code ExprValueUtils.fromObjectValue})
+            // don't recognize and reject as "unsupported object class". Mirror the
+            // top-level VarCharVector branch above and substitute Java strings.
+            List<Object> normalized = new ArrayList<>(raw.size());
+            for (Object element : raw) {
+                normalized.add(element instanceof Text t ? t.toString() : element);
+            }
+            return normalized;
+        }
+        if (value instanceof Text t) {
             return t.toString();
         }
-        return obj;
+        return value;
     }
 }

sandbox/plugins/analytics-engine/src/main/java/org/opensearch/analytics/exec/stage/RowResponseCodec.java

@@ -20,6 +20,8 @@
 import org.apache.arrow.vector.VarBinaryVector;
 import org.apache.arrow.vector.VarCharVector;
 import org.apache.arrow.vector.VectorSchemaRoot;
+import org.apache.arrow.vector.complex.ListVector;
+import org.apache.arrow.vector.complex.impl.UnionListWriter;
 import org.apache.arrow.vector.types.pojo.ArrowType;
 import org.apache.arrow.vector.types.pojo.Field;
 import org.apache.arrow.vector.types.pojo.FieldType;
@@ -28,6 +30,7 @@
 
 import java.nio.charset.StandardCharsets;
 import java.util.ArrayList;
+import java.util.Collections;
 import java.util.List;
 
 /**
@@ -60,8 +63,7 @@ public VectorSchemaRoot decode(FragmentExecutionResponse response, BufferAllocat
         // Infer Arrow type per column from the first non-null value
         List<Field> fields = new ArrayList<>();
         for (int col = 0; col < fieldNames.size(); col++) {
-            ArrowType arrowType = inferArrowType(rows, col);
-            fields.add(new Field(fieldNames.get(col), FieldType.nullable(arrowType), null));
+            fields.add(inferField(fieldNames.get(col), rows, col));
         }
         Schema schema = new Schema(fields);
 
@@ -86,29 +88,67 @@ public VectorSchemaRoot decode(FragmentExecutionResponse response, BufferAllocat
     }
 
     /**
-     * Infers the Arrow type for a column by scanning rows for the first
-     * non-null value. Falls back to {@code Utf8} (VarChar) if all values
-     * are null or the Java type is unrecognized.
+     * Infers the Arrow {@link Field} for a column by scanning rows for the first
+     * non-null value. Falls back to a nullable {@code Utf8} (VarChar) field if all
+     * values are null or the Java type is unrecognized.
+     *
+     * <p>For {@link List} cells (produced by analytics-engine routes that emit
+     * array-typed values — PPL {@code array(...)}, {@code array_slice}, …), this
+     * returns a {@code List<inner>} field where the inner element type is inferred
+     * from the first non-null element. Without this branch, list values fall
+     * through to the {@code Utf8} fallback and {@link #setVectorValue} produces
+     * {@code value.toString()} (e.g. {@code "[2,3,4]"} as a JSON-like string)
+     * instead of a typed array.
      */
-    static ArrowType inferArrowType(List<Object[]> rows, int col) {
+    static Field inferField(String name, List<Object[]> rows, int col) {
         for (Object[] row : rows) {
             Object value = row[col];
             if (value == null) continue;
-            if (value instanceof Long) return new ArrowType.Int(64, true);
-            if (value instanceof Integer) return new ArrowType.Int(32, true);
-            if (value instanceof Short) return new ArrowType.Int(16, true);
-            if (value instanceof Byte) return new ArrowType.Int(8, true);
-            if (value instanceof Double) return new ArrowType.FloatingPoint(org.apache.arrow.vector.types.FloatingPointPrecision.DOUBLE);
-            if (value instanceof Float) return new ArrowType.FloatingPoint(org.apache.arrow.vector.types.FloatingPointPrecision.SINGLE);
-            if (value instanceof Boolean) return ArrowType.Bool.INSTANCE;
-            if (value instanceof CharSequence) return ArrowType.Utf8.INSTANCE;
-            if (value instanceof byte[]) return ArrowType.Binary.INSTANCE;
-            if (value instanceof Number) return new ArrowType.Int(64, true);
-            break;
+            if (value instanceof List<?> list) {
+                ArrowType elementType = inferElementArrowType(list, rows, col);
+                Field elementField = new Field("$data$", FieldType.nullable(elementType), null);
+                return new Field(name, FieldType.nullable(ArrowType.List.INSTANCE), Collections.singletonList(elementField));
+            }
+            return new Field(name, FieldType.nullable(scalarArrowType(value)), null);
+        }
+        return new Field(name, FieldType.nullable(ArrowType.Utf8.INSTANCE), null);
+    }
+
+    /**
+     * Best-effort inference for a list element type. Looks at the first non-null
+     * element of the given list, then falls back to scanning later rows of the
+     * same column if this list is empty or all-null. Defaults to {@code Utf8}.
+     */
+    private static ArrowType inferElementArrowType(List<?> list, List<Object[]> rows, int col) {
+        for (Object element : list) {
+            if (element != null) return scalarArrowType(element);
+        }
+        for (Object[] row : rows) {
+            Object value = row[col];
+            if (value instanceof List<?> other) {
+                for (Object element : other) {
+                    if (element != null) return scalarArrowType(element);
+                }
+            }
         }
         return ArrowType.Utf8.INSTANCE;
     }
 
+    /** Maps a Java scalar value to the corresponding Arrow scalar type. */
+    private static ArrowType scalarArrowType(Object value) {
+        if (value instanceof Long) return new ArrowType.Int(64, true);
+        if (value instanceof Integer) return new ArrowType.Int(32, true);
+        if (value instanceof Short) return new ArrowType.Int(16, true);
+        if (value instanceof Byte) return new ArrowType.Int(8, true);
+        if (value instanceof Double) return new ArrowType.FloatingPoint(org.apache.arrow.vector.types.FloatingPointPrecision.DOUBLE);
+        if (value instanceof Float) return new ArrowType.FloatingPoint(org.apache.arrow.vector.types.FloatingPointPrecision.SINGLE);
+        if (value instanceof Boolean) return ArrowType.Bool.INSTANCE;
+        if (value instanceof CharSequence) return ArrowType.Utf8.INSTANCE;
+        if (value instanceof byte[]) return ArrowType.Binary.INSTANCE;
+        if (value instanceof Number) return new ArrowType.Int(64, true);
+        return ArrowType.Utf8.INSTANCE;
+    }
+
     /**
      * Sets a value on the appropriate Arrow vector type. Handles null by
      * calling {@code setNull}. For typed vectors, casts the Java value to
@@ -137,8 +177,64 @@ static void setVectorValue(FieldVector vector, int index, Object value) {
             ((VarCharVector) vector).setSafe(index, value.toString().getBytes(StandardCharsets.UTF_8));
         } else if (vector instanceof VarBinaryVector) {
             ((VarBinaryVector) vector).setSafe(index, (byte[]) value);
+        } else if (vector instanceof ListVector listVector) {
+            writeListValue(listVector, index, (List<?>) value);
         } else {
             throw new IllegalArgumentException("Unsupported Arrow vector type: " + vector.getClass().getSimpleName());
         }
     }
+
+    /**
+     * Writes a Java {@link List} into an Arrow {@link ListVector} at the given row
+     * index. Bypasses {@link UnionListWriter} entirely — writes directly to the
+     * list's offset / validity buffers and to the inner data vector via the inner
+     * vector's own typed setter. The writer-based API requires an
+     * {@link org.apache.arrow.memory.ArrowBuf} per varchar element which Arrow
+     * couples to a release lifecycle that's tricky to get right (early close →
+     * use-after-free, no close → leak); the direct path avoids that altogether.
+     *
+     * <p>The inner data vector's type was decided in {@link #inferField} from the
+     * first non-null list element, so the {@code instanceof} dispatch here simply
+     * needs to match.
+     */
+    private static void writeListValue(ListVector listVector, int index, List<?> list) {
+        FieldVector dataVector = listVector.getDataVector();
+        int startOffset = listVector.getOffsetBuffer().getInt((long) index * ListVector.OFFSET_WIDTH);
+        int writePos = startOffset;
+        for (Object element : list) {
+            // Grow the data vector if needed before writing. setSafe-style helpers handle this
+            // automatically per type, but we still need to pre-position the cursor.
+            if (element == null) {
+                dataVector.setNull(writePos);
+            } else if (dataVector instanceof BigIntVector v) {
+                v.setSafe(writePos, ((Number) element).longValue());
+            } else if (dataVector instanceof IntVector v) {
+                v.setSafe(writePos, ((Number) element).intValue());
+            } else if (dataVector instanceof SmallIntVector v) {
+                v.setSafe(writePos, ((Number) element).shortValue());
+            } else if (dataVector instanceof TinyIntVector v) {
+                v.setSafe(writePos, ((Number) element).byteValue());
+            } else if (dataVector instanceof Float8Vector v) {
+                v.setSafe(writePos, ((Number) element).doubleValue());
+            } else if (dataVector instanceof Float4Vector v) {
+                v.setSafe(writePos, ((Number) element).floatValue());
+            } else if (dataVector instanceof BitVector v) {
+                v.setSafe(writePos, ((Boolean) element) ? 1 : 0);
+            } else if (dataVector instanceof VarCharVector v) {
+                v.setSafe(writePos, element.toString().getBytes(StandardCharsets.UTF_8));
+            } else if (dataVector instanceof VarBinaryVector v) {
+                v.setSafe(writePos, (byte[]) element);
+            } else {
+                throw new IllegalArgumentException("Unsupported list element vector type: " + dataVector.getClass().getSimpleName());
+            }
+            writePos++;
+        }
+        // Mark this row's list as non-null and update its end offset.
+        listVector.setNotNull(index);
+        listVector.getOffsetBuffer().setInt((long) (index + 1) * ListVector.OFFSET_WIDTH, writePos);
+        // Keep the data vector's value count in sync so subsequent reads see the new tail.
+        if (writePos > dataVector.getValueCount()) {
+            dataVector.setValueCount(writePos);
+        }
+    }
 }