perf: send scalar UDF arguments as length-1 vectors

Literal-folded children no longer get expanded to batch-row count before
crossing JNI; ColumnarValue::Scalar is materialized at length 1, avoiding
an O(rows) copy of values that never vary across the batch. Document the
contract on CometUDF: scalar inputs arrive as length-1 vectors, vector
inputs at the batch row count, and the result must match the longest
input.

Author: andygrove

common/src/main/scala/org/apache/comet/udf/CometUDF.scala

@@ -23,9 +23,14 @@ import org.apache.arrow.vector.ValueVector
 
 /**
  * Scalar UDF invoked from native execution via JNI. Receives Arrow vectors as input and returns
- * an Arrow vector. The returned vector must have the same length as the input vectors.
+ * an Arrow vector.
  *
- * Implementations must have a public no-arg constructor.
+ *   - Vector arguments arrive at the row count of the current batch.
+ *   - Scalar (literal-folded) arguments arrive as length-1 vectors and must be read at index 0.
+ *   - The returned vector's length must match the longest input.
+ *
+ * Implementations must have a public no-arg constructor and should be stateless: instances are
+ * cached per executor thread for the lifetime of the JVM.
  */
 trait CometUDF {
   def evaluate(inputs: Array[ValueVector]): ValueVector

native/spark-expr/src/jvm_udf/mod.rs

@@ -105,12 +105,17 @@ impl PhysicalExpr for JvmScalarUdfExpr {
     }
 
     fn evaluate(&self, batch: &RecordBatch) -> DFResult<ColumnarValue> {
-        // Step 1: evaluate child expressions to get Arrow arrays.
-        let n = batch.num_rows();
+        // Step 1: evaluate child expressions to get Arrow arrays. Scalar children
+        // (e.g. literal patterns) are sent as length-1 vectors rather than expanded
+        // to batch-row count, so the JVM bridge does not pay an O(rows) copy for
+        // values that never vary across the batch.
         let arrays: Vec<ArrayRef> = self
             .args
             .iter()
-            .map(|e| e.evaluate(batch).and_then(|cv| cv.into_array(n)))
+            .map(|e| match e.evaluate(batch)? {
+                ColumnarValue::Array(a) => Ok(a),
+                ColumnarValue::Scalar(s) => s.to_array_of_size(1),
+            })
             .collect::<DFResult<_>>()?;
 
         // Step 2: allocate FFI structs on the Rust heap and collect their raw pointers.