feat: execute concat_ws with array arguments natively

[peterxcli]

Which issue does this PR close?

Closes #5687.

Rationale for this change

concat_ws with string-array arguments currently uses Spark codegen dispatch, or falls back to Spark when dispatch is disabled. DataFusion Spark 55, already used by Comet, provides the required native implementation.

What changes are included in this PR?

Route arrays and separator-only calls through spark_concat_ws, with an explicit return type and a small adapter around upstream SparkConcatWs. The adapter evaluates all-scalar arguments once and returns a scalar for DataFusion to broadcast. This handles non-foldable scalar subqueries, which otherwise panic when the upstream kernel indexes a one-row array using the enclosing batch length.

Keep ordinary string-only calls on the existing DataFusion kernel. Retain the all-foldable fallback and NULL-literal separator shortcut. Add native and Spark regressions covering ordered mixed arguments, multiple arrays, nulls, empty values, Unicode, column separators, and runtime scalar subqueries. Update expression support and audit documentation.

Add CometConcatWsBenchmark for string-only and mixed-array inputs, including plan assertions with codegen dispatch disabled. The implement-comet-expression skill was used to scaffold the implementation workflow.

How are these changes tested?

• Reproduced the runtime-scalar panic before the fix; both Rust concat_ws regressions pass after it.
• Spark 4.0.4: all 3 focused tests pass, including the SQL-file case with 16 queries, native array coverage, and multi-row scalar-subquery coverage. Codegen dispatch is disabled in the Scala regressions.
• Debug and release native builds pass; workspace Clippy with -D warnings passes.
• Spark 4.0 semantic Scalafix, Spotless, compilation, and test compilation pass. The ArrayType import is now used by the routing check.

Matched benchmark results

Apple M4, 24 GiB RAM, macOS 26.6.2, Zulu JDK 21.0.6, Spark 4.0.4. Native libraries use the same optimized Cargo release profile without target-cpu=native. Each run uses a fresh JVM with 4 GiB heap, local[1], one Comet worker thread, AQE disabled, and 8,192-row batches for both readers. Standard Spark Benchmark warmup and minimum measurement duration apply.

Inputs are 65,536 deterministic Parquet rows in one file. String widths are 8 and 128; arrays vary from 1 to N elements for N=2/8/32. Every 13th string and 11th array is NULL, every fifth array element is NULL, and the column separator alternates |/-- with NULL every 17th row. Mixed expression: concat_ws(separator, a1, c1, a2, 'tail'); string expression: concat_ws(separator, c1, c1) (the previously supported benchmark shape).

All measurements verified Spark WholeStageCodegenExec with concat_ws present and fully native CometProject over CometNativeScan, with Scala UDF codegen dispatch disabled. Timing includes Parquet scanning, projection, and the noop sink; these are query timings, not isolated kernel speedups. This is a development laptop with background applications, so the variance below matters and small differences should not be interpreted as improvements.

Performance tradeoff: native support is not a universal speedup. Short-string cases are faster in both runs. With 128-character strings, arrays up to 32 elements and a literal separator, Comet takes 153 vs 126 ms in run 1 and 126 vs 114 ms in run 2 (about 21% and 11% slower). Column separators are near parity for that shape. The first run has substantial outliers, so both runs are included. This leaves a measurable long-string array optimization opportunity in the new path while preserving the existing string kernel.

Array/mixed cases, updated library, two fresh-JVM runs. Times are mean ± standard deviation in milliseconds; ratio is Spark mean / Comet mean.

Run	Width	Maximum array length	Separator	Spark (ms)	Comet (ms)	Ratio
1	8	2	literal	35 ± 5	25 ± 4	1.40x
1	8	2	column	48 ± 12	30 ± 8	1.60x
1	8	8	literal	63 ± 42	47 ± 66	1.34x
1	8	8	column	82 ± 76	39 ± 8	2.10x
1	8	32	literal	113 ± 37	71 ± 12	1.59x
1	8	32	column	106 ± 11	59 ± 6	1.80x
1	128	2	literal	33 ± 2	25 ± 2	1.32x
1	128	2	column	34 ± 4	25 ± 2	1.36x
1	128	8	literal	51 ± 5	61 ± 17	0.84x
1	128	8	column	61 ± 11	61 ± 33	1.00x
1	128	32	literal	126 ± 7	153 ± 11	0.82x
1	128	32	column	137 ± 12	143 ± 7	0.96x
2	8	2	literal	28 ± 2	23 ± 5	1.22x
2	8	2	column	26 ± 1	22 ± 3	1.18x
2	8	8	literal	39 ± 6	25 ± 1	1.56x
2	8	8	column	35 ± 2	25 ± 2	1.40x
2	8	32	literal	75 ± 5	49 ± 2	1.53x
2	8	32	column	72 ± 2	50 ± 2	1.44x
2	128	2	literal	28 ± 1	22 ± 3	1.27x
2	128	2	column	27 ± 2	22 ± 4	1.23x
2	128	8	literal	47 ± 4	44 ± 2	1.07x
2	128	8	column	47 ± 4	46 ± 7	1.02x
2	128	32	literal	114 ± 6	126 ± 5	0.90x
2	128	32	column	114 ± 5	116 ± 4	0.98x

Existing string-only path, before and after. Before uses the native factory from parent 75fdddc9285ec61c0cd326977c61dd41fca39a8b; after uses this PR's final routing and adapter. The JVM benchmark and inputs are identical. Run order: before-1, after-1, before-2, after-2, updated matrix-1, updated matrix-2. No string regression was observed in either pair; the final implementation retains the original string kernel.

Run	Width	Separator	Spark before (ms)	Comet before (ms)	Spark after (ms)	Comet after (ms)
1	8	literal	31 ± 22	15 ± 2	22 ± 3	14 ± 2
1	8	column	19 ± 2	13 ± 2	19 ± 3	12 ± 1
1	128	literal	25 ± 13	13 ± 2	22 ± 18	12 ± 1
1	128	column	18 ± 1	12 ± 2	18 ± 1	11 ± 1
2	8	literal	22 ± 2	13 ± 1	21 ± 2	13 ± 2
2	8	column	26 ± 7	14 ± 2	18 ± 2	12 ± 1
2	128	literal	28 ± 7	19 ± 4	18 ± 2	19 ± 13
2	128	column	26 ± 10	14 ± 3	21 ± 4	12 ± 1

Library SHA-256: before ffe3b864c56bbfa2620250d1591380965d06733b7fc0b9581f8c64f1c9d346dd; after 3dad10e0a2d7d51ef67bec0b01ede096e771f91c79be6b1fc441e4fedc0f7d67.

The benchmark can be run through make benchmark-org.apache.spark.sql.benchmark.CometConcatWsBenchmark PROFILES=-Pspark-4.0 BENCH_HEAP=4g; pass strings for only the existing string cases. For the measurements above, the same class was launched directly with the reactor test classpath and an explicit java.library.path to select each release library, using JVM options from make print-benchmark-args. The normal make target additionally enables target-cpu=native, so use identical Cargo flags for both libraries when repeating the before/after comparison.

[sunchao]

Correctness

Reviewed 309dd7dc against c348f775. This replaces the array-argument JVM dispatch/fallback with DataFusion Spark 55's SparkConcatWs, and also replaces the existing native implementation for string-only calls. The explicit return type avoids DataFusion's string-only argument signature while preserving Spark's analyzed children. The literal-NULL separator shortcut and all-foldable fallback remain.

I compared the implementation with the maintained Spark 3.5 (5947fd6e) and 4.0 (03f28fc4) sources. Their ConcatWs flattens arrays in argument order; NULL separators return NULL; NULL strings, arrays and elements contribute nothing; empty strings still count as elements; zero retained elements and separator-only calls return an empty string. The new kernel follows those rules for column inputs, including varying separators and UTF-8 strings. Spark's analyzer still owns implicit casts and invalid argument rejection; the native planner preserves the already-analyzed child types rather than invoking the kernel's broader coercion rules. Concatenation itself adds no ANSI arithmetic behavior. Spark 4.0 carries collation metadata but concatenates the original UTF-8 bytes. The maintained 3.4 and 4.1 branches were unavailable, so this is not a source-based compatibility qualification for those versions.

[P2] The all-foldable guard does not protect runtime scalar inputs. A string scalar subquery is non-foldable but evaluates to ColumnarValue::Scalar in Comet. With a literal separator and no column-valued argument, the new kernel expands inputs to one row and then indexes them using the surrounding batch's row count. It reaches an out-of-bounds string access on the second row. The previous DataFusion string implementation returns a scalar for this shape. The inline finding describes the required runtime handling and regression coverage; this conclusion is established from the exact sources, not a local execution claim.

Validation and CI

The new Rust registration test passed in job 101381376783. Its checkout log identifies merge eb17a9fd; I verified raw parents [c348f775, 309dd7dc] and matching feature/dependency/planner blobs. The four decimal-cast file differences between the current base and the PR head are branch divergence, not changes introduced by this PR.

The final current-head CI refresh at 2026-09-05 22:49:38 UTC reported 46 successful, four failed, 15 in progress and seven skipped checks. The four Java-lint failures report Scalafix errors; the Spark 3.5 lint diagnostic specifically requests removal of the now-unused ArrayType import in strings.scala. That must be cleared. The Scala regression explicitly disables codegen dispatch, but its presence is not a local test result. The PR body reports Spark 4.1 SQL/Scala passes; I did not independently run Spark/JVM tests or a local kernel reproduction, and the wider expression CI remained pending at the snapshot cutoff.

Performance

[P2] Add representative microbenchmark results before enabling this implementation. No results are provided in the PR body or discussion, and the benchmark CI job was skipped. The existing CometStringExpressionBenchmark only benchmarks concat_ws(' ', c1, c1); the PR adds no array case. Compare the new native array path against Spark codegen, and measure the existing string-only case before and after this change with the same data, configuration and verified execution paths.

The kernel writes directly into an Arrow builder, avoiding an intermediate concatenated string per row. However, values_to_arrays expands scalar separators and literal arguments across the batch, list rows create slices, and the output starts with a fixed 16-byte-per-row reservation. The former string kernel keeps scalar references and computes an input-derived capacity. These are material differences for a string hot path. Cover column and literal separators, mixed arguments, realistic array lengths, nulls, and short/long strings; correctness comparisons alone cannot establish the performance benefit or absence of regression.

Design

Reusing the already-locked Spark-specific kernel is a small and appropriate change: it avoids maintaining a second array-flattening implementation in Comet. The name-based factory arm takes precedence over the default registry, and the explicit return type prevents an unrelated string-only signature from rejecting Spark-valid arrays. Spark analysis remains responsible for admissible inputs, so bypassing DataFusion coercion is deliberate here.

The remaining correctness issue is at the boundary between Spark's foldability and DataFusion's runtime scalar/array representation. A runtime adapter that handles all-scalar batches, or a corrected upstream kernel, addresses that contract directly. Extending a compile-time literal check alone does not establish the required batch cardinality.

Abstraction & complexity

The PR introduces no new framework or broad abstraction: one factory arm and a smaller serde replace the array fallback branch. The SQL tests and dispatch-disabled Scala checks use existing test infrastructure. The separation between Spark admission, serialization and the reused kernel is understandable. Any adapter needed for the scalar case should stay local to this UDF and preserve existing fallback behavior; no further abstraction change is warranted by the reviewed scope.