[SPARK-59317][SQL] Unwrap the cast of a DSv2 scalar subquery runtime filter before pushing it down - #58912
Open
Vivek1106-04 wants to merge 1 commit into
Open
[SPARK-59317][SQL] Unwrap the cast of a DSv2 scalar subquery runtime filter before pushing it down#58912Vivek1106-04 wants to merge 1 commit into
Vivek1106-04 wants to merge 1 commit into
Conversation
…filter before pushing it down Type coercion of the compared sides may wrap the filtered column in a cast, e.g. `cast(part_col as bigint) = <scalar subquery>` when an INT partition column is compared with a BIGINT scalar subquery. `DataSourceV2Strategy.translateScalarSubqueryFilterV2` literalizes the subquery once its result is known and translates the comparison, so the cast is pushed but never unwrapped, and a source that only prunes on column references ignores the filter. `UnwrapCastInBinaryComparison` gains `unwrapCastInExpression`, which unwraps the cast of every binary comparison, `In` and `InSet` of an expression. The rule's own `apply` is rewritten on top of it, so the optimizer and the runtime path share one function. `translateScalarSubqueryFilterV2` calls it on the literalized filter. For a value out of the column's range the unwrapping produces a null-returning form that has no V2 predicate of its own, so the two are rewritten into what they filter by before translation: `and(isnull(col), null)` keeps no row and is pushed as `false`, `or(isnotnull(col), null)` keeps the rows where the column is not null and is pushed as `isnotnull(col)`. Only the whole filter is rewritten, as neither is interchangeable with its counterpart under a `Not`. A filter whose cast can't be unwrapped is pushed as it was. `InMemoryTableWithV2Filter`, which the tests prune with, matched its `=` branch on any predicate and cast the first child to `FieldReference`, so a pushed cast failed it with a `ClassCastException`. It now takes that branch only for a reference, as a source that prunes on column references does.
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.
What changes were proposed in this pull request?
Type coercion of the compared sides may wrap the filtered column in a cast, e.g.
cast(part_col as bigint) = <scalar subquery>when an INT partition column is compared with aBIGINT scalar subquery.
DataSourceV2Strategy.translateScalarSubqueryFilterV2literalizes thescalar subquery once its result is known and translates the comparison with
translateFilterV2,which pushes
CAST(part_col AS BIGINT) = <value>throughV2ExpressionBuilder. The cast ispushed but never unwrapped, so a source that only prunes on column references ignores the filter.
The optimizer cannot unwrap it either, because the other side is not a literal until runtime.
This PR adds
UnwrapCastInBinaryComparison.unwrapCastInExpression, which unwraps the cast ofevery binary comparison,
InandInSetof an expression. The rule's ownapplyis rewritten ontop of it, so the optimizer and the runtime path share one function.
translateScalarSubqueryFilterV2calls it on the literalized filter:part_col = 3;produces a null-returning form that has no V2 predicate of its own, so it is rewritten into what
it filters by before translation:
and(isnull(col), null)keeps no row and is pushed asfalse(
AlwaysFalse),or(isnotnull(col), null)keeps the rows where the column is not null and ispushed as
isnotnull(col). Only the whole filter is rewritten, as neither is interchangeablewith its counterpart under a
Not;This is the sibling of SPARK-59301 (#58580), which did the same for the dynamic partition pruning
IN filter. Unlike the IN case, the comparison rule rewrites out-of-range or rounded values into
different comparisons or constant results, hence the handling above.
InMemoryTableWithV2Filter, the test table the DSv2 pruning tests use, matched its=branch onany predicate and cast the first child to
FieldReference, so a pushed cast failed it with aClassCastException. It now takes that branch only for a reference, as a source that prunes oncolumn references does.
Why are the changes needed?
A DSv2 scan whose partition column is compared with a wider-typed scalar subquery received
CAST(part_col AS BIGINT) = <value>, which a source pruning on column references cannot use, soit read every partition. Reported in
#58580 (comment).
Reproduction, with the in-memory V2 table taking runtime filters as V2 predicates: a 10-partition
table with an INT partition column and a BIGINT dimension column.
Before this PR all 10 partitions are read, after it 1 is.
Does this PR introduce any user-facing change?
Yes. DSv2 sources now receive scalar subquery runtime filters whose column is wrapped in a
lossless numeric cast, so the scan prunes partitions instead of reading all of them. Query results
do not change.
How was this patch tested?
translateScalarSubqueryFilterV2inDataSourceV2StrategySuite: bare andnested columns, cast unwrapping for
=and>, a value rounded by the conversion, a valueabove the column's range for
=and for<, and a lossy cast that must not be unwrapped.DataSourceV2SQLSuiteV2Filter, which asserts the answer and that 1 of 10partitions survives pruning.
CAST(cint AS long) = 1instead ofcint = 1, and the end-to-end test reads all 10 partitions.UnwrapCastInBinaryComparisonSuite,DataSourceV2StrategySuite,DataSourceV2SQLSuiteV2Filter,DataSourceV2CatalystRuntimeFilterSuite,DynamicPartitionPruningV1SuiteandDynamicPartitionPruningV2Suitepass.Was this patch authored or co-authored using generative AI tooling?
Generated-by: Claude Code (Claude Opus 5)