fix: match Spark's null short-circuiting in array_join and enable it natively

[Visorgood]

Which issue does this PR close?

Closes #3178.

Rationale for this change

array_join was flagged Incompatible with the note "Null handling may differ from Spark", but the specific difference was never pinned down. Investigating it turned up two divergences on the native path, neither of them the one #3178 guessed at.

1. A null nullReplacement does not nullify the row. Spark's ArrayJoin returns null whenever nullReplacement evaluates to null, even when the array holds no nulls to replace. DataFusion's array_to_string reads a null null_string as "omit null elements", i.e. the same as not passing a third argument:

arr	nullrep	Spark	Comet (before)
["a", null, "c"]	NULL	NULL	a,c
["a", "b", "c"]	NULL	NULL	a,b,c

2. Arguments are evaluated eagerly. Spark short-circuits to null on the array, then the delimiter, then the replacement, and never evaluates the later arguments. DataFusion's ScalarFunctionExpr evaluates all of them first, so an argument that can throw fails on rows where Spark simply returns null. array_join(arr, element_at(delims, idx)) with arr = NULL and idx = 0 throws in Comet instead of returning null. Thanks to @sunchao for spotting this one.

The two cases #3178 actually asks about – null elements skipped without a replacement, and substituted with one – were already correct; the tests here lock that in. The issue's "Current Comet Implementation" snippet quoted only the two-argument branch of convert; the three-argument branch has existed since #1490.

What changes are included in this PR?

Null guards. convert nests the array_to_string call inside IsNull guards, ordered the way Spark evaluates its arguments. This restores the short-circuit because Comet's IfExpr delegates to DataFusion's CaseExpr, which evaluates each branch against a filtered batch, so a guarded argument is never evaluated on a short-circuited row.

Guards are emitted only where they protect something: the array and delimiter are guarded only when a later argument is non-foldable, and the replacement whenever it is nullable. Non-nullable arguments are never guarded, so array_join(arr, ',') and array_join(arr, ',', 'X') keep exactly the plan they had before.

Non-deterministic arguments. A guarded argument is serialized twice, once under IsNull and once as a function argument, and CaseExpr evaluates the false branch on a filtered batch. A non-deterministic argument would therefore advance its state differently in the two copies. There is no way to bind a value once in a serialized expression tree, so getSupportLevel reports Incompatible for that case and it runs through the codegen dispatcher, where Spark's single evaluation is preserved.

Support level. With null handling matching Spark, non-collated input reports Compatible() and array_join runs natively without allowIncompatible. Non-default string collations remain Incompatible under #2190. This mirrors CometReverse, which CometArrayJoin already named as the precedent for its collation handling.

Docs. The array_join audit entry and the expressions.md note describe the guards and the non-deterministic carve-out.

No native or protobuf changes were required.

How are these changes tested?

Comet SQL Tests under spark/src/test/resources/sql-tests/expressions/array/:

• array_join.sql — expanded from 2 queries to 15: null elements leading / trailing / only / all, empty-string elements versus nulls, null delimiters, empty and multi-character delimiters, non-string element types, an empty-string replacement, and every combination of column and literal arguments. This file previously ran without allowIncompatible, so it exercised the codegen dispatcher rather than the native path and could not have caught either bug.
• array_join_null_replacement.sql — regression coverage for the first divergence, including the array-with-no-nulls case.
• array_join_null_array_guard.sql — regression coverage for the second, using element_at(delims, 0) as an argument that throws when evaluated.

CometArrayExpressionSuite:

• the existing array_join test no longer forces allowIncompatible, so the default path is exercised over the 10000-row dictionaryEnabled Parquet matrix, with an added query covering the guarded shape.
• array_join support level pins the native path and array_join emits a null guard only where it is needed assert the verdict and the guard placement directly. Because CometArrayJoin is a CodegenDispatchFallback, an Incompatible verdict runs Spark's own doGenCode and produces identical results, so every result assertion would keep passing if the support level silently regressed. Verified by reverting Compatible() locally: all .sql fixtures stayed green and only these tests failed.

Each new test was confirmed to fail when the behaviour it pins is removed.

Verified locally with CometArrayExpressionSuite in full alongside the sql-tests, scalastyle and spotless enabled:

Spark	Scala	JDK	Result
4.1.3	2.13	21	52 passed
4.0.4	2.13	17	52 passed
3.5.9	2.12	21	52 passed
3.4.3	2.12	17	50 passed, 2 cancelled (isSpark35Plus gates)

apache-rat and scalafix (syntactic and semantic) also run clean.

[andygrove]

Thanks for digging into this. The diagnosis is solid. I read through Spark's ArrayJoin.eval and DataFusion's generate_string_array and both match your description, including the part where a null replacement nullifies the result even when the array holds no nulls to replace.

A few things I checked while reviewing, in case they save anyone else the trip. The IfExpr guard shape is consistent with how CometArraysZip already handles its own null propagation in the same file, so the construction is idiomatic. Dictionary-encoded columns are unpacked in ScanExec::get_next, so the extra null_strings argument cannot hit array_to_string's "unsupported type for third argument" arm. The synthetic IsNull node is not part of op.expressions, so rollUpInfoMessages will not pollute the extended-explain coverage counts with an IsNull the user never wrote. And the Current status: audit bullet matches the format array_intersect already uses right above it.

Four things I would like to see addressed before this merges.

CI has not run. gh pr checks reports nothing at all and the status rollup is empty, so a maintainer will need to approve the workflows. Since this moves array_join onto the native path by default on every supported Spark version, I would want the 3.4 and 3.5 jobs green and not just the local runs in the description.

The Scala test still forces the opt-in. CometArrayExpressionSuite.scala:522 wraps the whole array_join test in withSQLConf(CometConf.getExprAllowIncompatConfigKey(classOf[ArrayJoin]) -> "true"). That was correct before this change and is now stale. It matters more than it looks, because that test is the only array_join coverage that runs over real Parquet with the dictionaryEnabled matrix and 10000 rows. With the wrapper still in place, the default path never gets exercised on that input shape. Could it be dropped?

The new tests cannot distinguish native from the dispatcher. CometArrayJoin is a CodegenDispatchFallback, so an Incompatible verdict routes through Spark's own doGenCode and produces Spark-identical results. checkSparkAnswerAndOperator checks operator coverage rather than which path an individual expression took. If getSupportLevel ever went back to Incompatible, every query in array_join_null_replacement.sql would still pass, silently. That is the same vacuous-pass shape the suite already guards against for expect_error in requireSentinelForCodegenExpectError, and it is the mirror image of the point you make in the description about the old fixture. Would it be worth adding something that pins the native path explicitly, perhaps in CometArrayExpressionSuite?

Two test cases worth adding to array_join.sql. Spark's inputTypes accepts any array that implicitly casts to array<string>, so array_join(array(1, 2, 3), ',') and array_join(array(1, NULL, 3), ',', 'X') are both valid and fairly common in practice. Neither file covers a non-string element type today. It might also be worth adding an empty-string replacement such as array_join(array('a', NULL, 'b'), ',', ''). The file covers empty-string elements well, but '' versus NULL for the replacement is exactly the distinction the new guard draws, so it seems worth pinning.

[sunchao]

[P2] Keep delimiter evaluation behind the null-array guard

Making this Compatible changes the default behavior of array_join(arr, element_at(delims, idx)) over supported Parquet input columns. For a row with arr = NULL, delims = [','], and idx = 0 (arr/delims are ARRAY<STRING> and idx is INT), Spark's generated code returns NULL without evaluating the delimiter. The native ScalarFunctionExpr evaluates every argument first, so ListExtract throws for index zero before array_to_string can inspect the null array, even with ANSI disabled. The previous default JVM dispatcher preserved Spark's guard, and the native error propagates rather than falling back. Please preserve that guard on this newly enabled path. This is source reasoning for the inspected Spark 3.5/4.0 paths; I have not run it.

[sunchao]

[P2] Follow-up on the remaining null-array case

Thanks, the original column-based delimiter example is guarded now. Could the guard selection also distinguish literal values from surviving foldable expressions? With Parquet columns flag = true and arr = NULL, consider SELECT IF(flag, array_join(arr, element_at(array(','), 0)), NULL) FROM t. Spark's ConstantFolding deliberately retains a foldable expression that throws inside a conditional branch, so this does not require disabling the optimizer. The delimiter remains foldable, which makes the new afterArray.exists(!_.foldable) test omit the array guard. Spark's two-argument generated path returns NULL without evaluating it, while the native scalar-function path evaluates ListExtract and raises INVALID_INDEX_OF_ZERO.

Could the guard also protect a surviving foldable delimiter, with a regression using normal constant folding? This is the remaining part of this null-array issue, based on the maintained Spark 3.5/4.0 sources and the current native path. I have not executed the query.

[sunchao]

[P2] Evaluate a nullable replacement once per input row

The replacement is serialized both as the function argument and again under IsNull. Consider IF(monotonically_increasing_id() % 2L = 0L, CAST(NULL AS STRING), 'X') with a sole input array column containing four rows of ['a', NULL, 'b'] in one native batch. The two native replacement trees have separate counters: the guard sees all four rows, while the function's copy sees only the two rows that passed the guard. That second copy produces one NULL replacement, which array_to_string treats as omission, yielding 'a,b'. Spark's single evaluation can produce only NULL or 'a,X,b'. The mismatch survives either starting parity, including counters advanced during planning. Please compute the replacement once and reuse its value for both the guard and join. This is source reasoning for the inspected Spark 3.5/4.0 paths, not an executed reproduction.

[Visorgood]

Thanks @andygrove and @sunchao! These were good catches. Pushed a fix that addresses them.

eager argument evaluation
Confirmed and fixed. I reproduced it before changing anything: with no guard, SELECT array_join(arr, element_at(delims, idx)) FROM t WHERE arr IS NULL throws inside Comet rather than returning null, exactly as was described.

convert now nests the array_to_string call inside IsNull guards in Spark's own evaluation order (array, then delimiter, then replacement). This works because Comet's IfExpr delegates to DataFusion's CaseExpr, which evaluates each branch against a filtered remainder_batch – so a guarded argument is never evaluated on the short-circuited rows. New fixture array_join_null_array_guard.sql covers it; removing the guard makes it fail.

Guards are emitted only where they protect something: earlier arguments are guarded only when a later one is non-foldable, and non-nullable arguments never are. So array_join(arr, ',') and array_join(arr, ',', 'X') keep the exact plan they had before this PR.

double evaluation of the replacement
Indeed there is no way to bind a value once in a serialized expression tree. Rather than paper over it, getSupportLevel now reports Incompatible when any argument that needs a guard is non-deterministic, so those route through the codegen dispatcher where Spark's single evaluation is preserved. The monotonically_increasing_id() % 2 example takes that path.

stale withSQLConf
Dropped. That test now exercises the default path over the 10000-row dictionaryEnabled Parquet matrix, plus one added query with a nullable non-foldable delimiter and replacement to cover the guarded shape.

tests cannot distinguish native from the dispatcher
Agreed. I checked, and reverting Compatible() to Incompatible left every .sql fixture green. Added array_join support level pins the native path, which asserts the verdict directly, and array_join emits a null guard only where it is needed, which asserts the guard is emitted only in the cases above. I verified both fail when the behavior they pin is removed.

extra test cases
Added non-string element types (array(1, 2, 3), array(1, NULL, 3) with and without a replacement, plus decimal and boolean) and the empty-string replacement, since '' versus NULL is exactly the distinction the guard draws.

Docs: the audit entry and the expressions.md note now describe the guards and the non-deterministic carve-out.

Verified locally on Spark 3.4.3, 3.5.9, 4.0.4 and 4.1.3 with CometArrayExpressionSuite in full alongside the sql-tests, scalastyle and spotless enabled: 52 passing on 4.0/3.5/4.1, 50 plus 2 isSpark35Plus cancellations on 3.4.

On CI: the earlier macos-14/Spark 4.0 [scans] failure looks unrelated to this change. Could someone re-approve the workflows so this gets a clean run?

[sunchao]

Re-reviewed b5f6d3effd2dfb22ce018c10058b8829557c219b. The original column-delimiter case and nullable nondeterministic replacement case are addressed. I found a new P2 in generated-code evaluation order and a remaining foldable-delimiter case for the existing null-array discussion.

These are source-derived findings. I did not execute Spark/Comet queries or tests. CI currently includes a failing Spark 3.5 Build Native + JVM Test Classes job, whose cause I have not classified.

[sunchao]

[P2] Preserve the generated code's replacement-first evaluation

Could these guards preserve ArrayJoin.doGenCode's evaluation order? In the maintained Spark 3.5/4.0 implementations, the replacement is evaluated and null-checked before the array and delimiter. With nullable Parquet columns arr = ['a','b'], delims = [','], idx = 0, and nullrep = NULL, array_join(arr, element_at(delims, idx), nullrep) therefore returns NULL in Spark's generated path. Here the deterministic arguments remain Compatible, and the delimiter guard runs before the replacement guard, so ListExtract raises INVALID_INDEX_OF_ZERO, even with ANSI disabled. The previous revision's outer replacement guard skipped that delimiter.

This also matters for a non-nullable replacement such as CAST(monotonically_increasing_id() AS STRING). It is excluded from guardedArgs, so the nondeterminism check permits it, but the new array guard advances its counter only for retained rows. Spark evaluates it before checking the array on every row. Moving only the nullable replacement guard would leave this case.

Could the lowering preserve replacement-first, single evaluation, or use the dispatcher for cases it cannot represent, with regression coverage for these shapes? These are source-derived cases, not executed queries.