fix: widen local table scan child nullability to match native kernels#4843
Open
andygrove wants to merge 1 commit into
Open
fix: widen local table scan child nullability to match native kernels#4843andygrove wants to merge 1 commit into
andygrove wants to merge 1 commit into
Conversation
An in-memory Seq/Map column read through CometLocalTableScanExec encodes its array element and map value as non-null (containsNull=false / valueContainsNull=false), but Comet expression serdes promise nullable children. Feeding a non-null child into a native array/map kernel fails the planned-vs-actual type assertion (e.g. spark_array_slice returning List(non-null Int32) while List(Int32) was promised) or panics building a ListArray for map_entries. Widen nested child-field nullability on both sides so they agree: the declared scan schema (serialized from op.output in CometSink) and the Arrow batch produced by the local scan. Without widening the declared schema too, the native ScanExec casts the batch back to the non-null type in build_record_batch. The widening is scoped to the local table scan via an overridable scanFieldType hook, leaving other sinks untouched. Map keys stay non-null, matching Arrow. This mirrors the Parquet scan path, which already reads children as nullable. Closes apache#4789
095a1cc to
3d0ba56
Compare
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?
Closes #4789.
Rationale for this change
Several Comet array/map expressions crash natively when the input comes through
CometLocalTableScanExecwith a non-nullable child field. An in-memorySeq[Int]/Map[Int, Int]column encodes its array element / map value as non-null (containsNull=false/valueContainsNull=false), but Comet expression serdes promise nullable children. Feeding a non-null child into a native kernel disagrees with the planned output type:slice:Assertion failed: result_data_type == *expected_type: Function 'spark_array_slice' returned value of type 'List(non-null Int32)' while ... expected: 'List(Int32)'map_entries:ListArray expected data type Struct("key": non-null Int32, "value": Int32) got Struct("key": non-null Int32, "value": non-null Int32)array_insert/ SPARK-41233 array prepend: analogous item-vs-array type mismatchIt does not reproduce over a native Parquet scan (which normalizes children to nullable) or with a SQL
array(...)/map(...)literal (literal children are nullable), so it is specific to the local scan path.What changes are included in this PR?
The non-null child is carried in two places, so both must be reconciled:
CometLocalTableScanExec.mapToReaders(built fromoriginalPlan.schema).ScanExec's declared schema, serialized fromop.outputinCometSink.convert.scan.rs:build_record_batchcasts the imported batch back to this declared schema, so widening only the Arrow data is reverted by the cast.Changes:
QueryPlanSerde.widenChildNullability, which recursively widens array elements, map values, and struct fields to nullable while leaving top-level types and map keys unchanged (map keys stay non-null, matching Arrow, and their nullability is never serialized).scanFieldTypehook inCometSink(default identity) so the widening is scoped to the local table scan and does not affect Union/Coalesce/shuffle sinks, whose children come from already-normalized native execution.CometLocalTableScanExecoverrides the hook and applies the same widening to its Arrow schema, so the declared schema and the batch agree exactly and no per-batch cast is needed.This mirrors how the Parquet scan path already normalizes children to nullable. No native/Rust changes are required.
How are these changes tested?
Added regression tests that run through the local table scan with
ConvertToLocalRelationdisabled (so the expression executes natively rather than being folded at plan time):CometArrayExpressionSuite:sliceandarray_inserton non-null-element arrays.CometMapExpressionSuite:map_entrieson a non-null-value map.All new tests pass, along with the existing
CometArrayExpressionSuite,CometMapExpressionSuite, and theLocalTableScantests inCometExecSuite(including nestedNullTypein struct/array/map).