fix: preserve array_join evaluation semantics across Spark versions - #5851
Draft
sunchao wants to merge 3 commits into
Draft
fix: preserve array_join evaluation semantics across Spark versions#5851sunchao wants to merge 3 commits into
sunchao wants to merge 3 commits into
Conversation
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Which issue does this PR close?
Follow-up to #5558 and #3178.
Rationale for this change
An
array_joinquery can produce different null results, suppress an error, or evaluate a stateful replacement more than once when Comet evaluates the expression separately from its surrounding Spark query. Matching the joined string for ordinary rows does not cover these differences.For example, with
spark.sql.codegen.factoryMode=NO_CODEGEN, Spark evaluates the array before the replacement. Given a non-nullnestedarray and a nullnr,array_join(element_at(nested, 0), ',', nr)must raiseINVALID_INDEX_OF_ZERO. Checking the replacement first can instead return NULL without evaluating the array.Before Spark 4.2, a nullable replacement column also interacts with nullability inferred by the enclosing query:
With nullable Parquet columns and generated execution, a row containing
['a', NULL, 'b'],',', and'X'can return NULL on older Spark after these filters, while the isolated native join returnsa,X,b. The difference comes from the non-null information the filter supplies to Spark's projection. This is the Spark behavior fixed by SPARK-57200; Comet must preserve it on the older versions it supports.Evaluation count matters too. A nullable replacement such as
concat(cast(monotonically_increasing_id() AS string), nr)carries state. Computing it once to check for NULL and again to produce the joined string can change the result. This also needs protection when the join sits inside another expression, such ascoalesce.What changes are included in this PR?
The enclosing Spark projection or filter now evaluates joins that need this context. Sending only the join through Spark's generated-expression evaluator cannot reproduce the surrounding nullability information, and generated evaluation cannot substitute for interpreted argument order. The same rule applies when a parent such as
regexp_replacewould otherwise take the entire expression into Comet.In the default configuration, nullable non-literal replacements stay on Spark before 4.2; simple nullable column replacements can use native execution on Spark 4.2 and later.
NO_CODEGENjoins and compound nullable replacements stay on Spark across all supported versions.The existing
spark.comet.expression.ArrayJoin.allowIncompatible=truesetting still permits simple nullable column replacements on older Spark versions, but it cannot override the interpreted-evaluation or repeated-evaluation safeguards. Ordinary joins with literal replacements continue to use native execution when otherwise supported.How are these changes tested?
A focused native-backed run passed all 12 tests on both Spark 4.1 and Spark 4.2, with no skips or failures. It covers decimal promotion and the array_join routing/execution regressions, including the two decimal-overflow cases that failed in the original CI run. The tests use the official CI-built native library from
bdd4b90b, matching this branch's native source. Full reactor production/test compilation, Scalastyle, Spotless, and diff checks passed.The decimal tests now distinguish expressions that should remain native from a compound nullable replacement that must stay on Spark. They retain native decimal coverage in the array, delimiter, and a non-nullable replacement, while explicitly checking Spark fallback and result/error behavior for the original protected shape. The two new serializer-routing regressions were also verified to fail against the pre-fix implementation and pass with this change.
This branch includes upstream #5849, which corrects the separate shared codegen coverage assertion. See fresh CI for revision
12e3d85cfor the complete platform and Spark-version matrix.