fix: dispatch map lookups with normalized keys and nondeterministic null-guarded children - #5867
Conversation
ff812cc to
689c563
Compare
…ull-guarded children Map lookups with float, collated or complex keys were declined by the serde without a dispatch fallback, so the whole projection fell back to Spark. Four array and map serdes reproduce NULL propagation with a guard that serializes the child twice, so a stateful child was evaluated twice natively and returned wrong answers. Mix the codegen dispatch fallback into the map lookups, and decline any nondeterministic child in size, array_append, arrays_zip and map_from_arrays through one shared gate so Spark's generated code evaluates it once. Closes apache#5580 Closes apache#5781
689c563 to
dfbf000
Compare
andygrove
left a comment
There was a problem hiding this comment.
The comment above getSupportLevel says only the array operand sits under the NULL guard, but the item is inside the THEN branch too, and DataFusion's CaseExpr filters the batch before it evaluates THEN. So the item is evaluated only on the rows the guard selects.
That matters for array_append specifically, because it is the one expression in this group whose Spark codegen does not short-circuit. On 3.4 and 3.5 ArrayAppend.doGenCode emits leftGen.code + rightGen.code + ctx.nullSafeExec(left.nullable, leftGen.isNull) { ... }, so rightGen.code runs on every row even when the array is NULL. ElementAt and MapFromArrays go through nullSafeCodeGen and Size has one child, which is why the guard matches Spark for those three.
On your own 16 row table, SELECT _1, array_append(IF(_1 % 2 = 0, array(1), CAST(NULL AS ARRAY<INT>)), monotonically_increasing_id()) FROM test_array_append_nondet gives Spark [1,0], [1,2], [1,4] and so on for the even rows, because the counter advances on all 16. Comet evaluates the native monotonically_increasing_id only over the 8 filtered rows and gives [1,0], [1,1], [1,2] up to [1,7]. The same filtering also swallows an error the item would have raised. Under ANSI, array_append(IF(_1 % 2 = 0, array(1), CAST(NULL AS ARRAY<INT>)), 1 / (_1 - 1)) raises DIVIDE_BY_ZERO in Spark at _1 = 1 and returns cleanly in Comet.
The expect_native case below uses array(1), which is non-nullable, so the guard mask is all true and CaseExpr takes the branch that skips the filter entirely. That case passes whatever the serde does, which is why the fixture does not catch this. Would it make sense for NullGuardSupport to cover both children of ArrayAppend, and to change that fixture to a nullable operand with a stateful item so it fails without the change?
On sequencing, I have commented on #5875 pointing it here, since this PR carries the null-guard correctness work as well as the map lookup dispatch and that makes it the better base. Two things to settle before it lands though. #5875 measured this exact routing change and found lookup-only projections running 8.7 to 11.4 percent slower with the dispatcher on at 65536 rows, with no reproducible benefit at 1024 rows. Do you have numbers for the routing here? And #5854 teaches the native map builders LAST_WIN and null key rejection, which would remove the Incompatible branch this PR routes through the dispatcher for map_from_arrays, so that one needs an order agreed with @peterxcli.
Last thing is coverage for the routes this enables. map_from_arrays_dedup_policy.sql sets spark.comet.exec.scalaUDF.codegen.enabled=false and keeps asserting expect_fallback(mapKeyDedupPolicy), so the new LAST_WIN dispatch route is never exercised. Could those queries assert expect_dispatch(map_from_arrays) instead, with a separate dispatcher-off file if the fallback is still worth pinning? The four new nondeterministic fixtures also build every array inline over an int column, so no dispatched kernel in them reads an Arrow ListVector from a column. #5844 hit a real Spark 3.4.3 ColumnarArray.copy problem in exactly that shape, so an array<int> column in at least the size and map_from_arrays fixtures would be worth having. While you are in expressions.md, cardinality still says Native two rows above size, and Spark registers both names against Size.
Which issue does this PR close?
Closes #5580, closes #5781.
Rationale for this change
Two serde gaps in the array and map expressions, both resolved by the codegen dispatcher rather than by native changes.
map_col[key]andelement_at(map, key)decline float, collated and complex map keys because the native lookup compares raw Arrow values where Spark normalizes-0.0, treats NaN as equal to itself, compares strings by collation and compares complex keys with interpreted ordering. The declines are right, but neither serde mixed inCodegenDispatchFallback, so the whole projection fell back to Spark instead of running Spark's own generated code inside the Comet pipeline.size,array_append,arrays_zipandmap_from_arraysreproduce Spark's NULL propagation with aCASE WHEN child IS NOT NULLguard that serializes the child twice. A stateful child advances each copy independently, so the guard and the operation see different rows and the answer is silently wrong: on a 16-row table withIF(monotonically_increasing_id() % 2 = 0, array(1), NULL)as the operand,sizereturned -1 on five rows where Spark returns 1,arrays_zipreturned[null, 2]for[1, 2],array_appendreturned[2]for[1, 2]andmap_from_arraysreturned NULL for{1 -> 2}.element_athad the same shape and was fixed in #5766 for its ANSI arm; these four are not ANSI-gated, so the wrong answers were reachable in every configuration.What changes are included in this PR?
CometMapExtractandCometElementAtmix inCodegenDispatchFallback, so the declined key types run through the dispatcher.CometElementAt's ANSI arm for a nondeterministic operand dispatches the same way.NullGuardSupportgate declines any nondeterministic child inCometSize,CometArrayAppend(the array operand only; the item is not under the guard),CometArraysZipandCometMapFromArrays, and all four mix inCodegenDispatchFallback, so Spark's generated code evaluates the child once. Nullability is not consulted: a non-nullable stateful child only stays correct today because DataFusion skips the filter when the guard matches every row, which is not a contract to rely on.getUnsupportedReasonslists updated for the doc generator, and the six affected rows in the expressions guide move from Native to Hybrid.How are these changes tested?
SQL-file fixtures over parquet tables, all asserting Spark's answer and the execution path:
*_nondeterministic_child.sqlfixtures, one per serde, with the stateful operand (dispatched), a non-nullable stateful operand (dispatched), and a deterministic nullable operand (native).array_appendalso pins that a stateful item stays native, and its fixture is capped at Spark 3.5 because 4.x rewritesarray_appendtoarray_insert. Before the change, the stateful cases failed as result mismatches with the values above.element_at_map.sqlandget_map_value.sqlflip their fallback cases to dispatch and add NaN lookups and a struct-keyed map column with a per-row key, a NULL inside the key and a NULL key.element_at_map_collation.sqlflips the same way and passes on the Spark 4.0 profile.element_at_ansi.sqlpins the dispatched nondeterministic arm.map_from_arrays_dedup_policy.sqldisables the dispatcher so its LAST_WIN fallback assertion keeps meaning what it says, matching the existingmap_from_entriesfixture.CometMapExpressionSuiterenames five fallback tests to dispatch tests through the helper that checks the dispatch tag.578 of 578 across
CometSqlFileTestSuite,CometArrayExpressionSuiteandCometMapExpressionSuiteon Spark 3.5, the map fixtures on the Spark 4.0 profile, andtest-compileon 4.0.