[
https://issues.apache.org/jira/browse/SPARK-59434?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Cheng Pan updated SPARK-59434:
------------------------------
Description:
{{PushdownPredicatesAndPruneColumnsForCTEDef}} collects the predicates at each
CTE reference and pushes their OR-merged combination into the shared CTE
definition. Each reference keeps its own predicates, so a non-deterministic
predicate pushed into the definition as well is evaluated a second time, which
can produce wrong results.
{code:sql}
create or replace temp view t as select * from values (0), (1), (2) as t(c1);
with v as (select c1, rand(1) r from t)
select c1 from v where rand(2) < 0.5
union all
select c1 from v where rand(3) < 0.5;
{code}
The definition is non-deterministic and referenced twice, so it survives
{{InlineCTE}}. {{(rand(2) < 0.5) OR (rand(3) < 0.5)}} lands in the definition
while both references keep their own filter, so the optimized plan holds four
{{rand}} filters instead of two, and rows are dropped twice.
No {{MATERIALIZED}} option is needed to reach this: any definition that
survives {{InlineCTE}} has the same shape.
The rule has behaved this way since it was added in SPARK-37670 (3.4.0). Its
scaladoc claimed determinism was taken care of by {{ScanOperation}}, but that
was never true -- {{ScanOperation}} ran with {{legacyMode = false}}, whose
guard admits the first filter whatever it is. {{PhysicalOperation}}, swapped in
later by SPARK-39764 (3.4.0), behaves the same way here: it returns a single
filter even when it is non-deterministic.
Fix: only push deterministic predicates into the CTE definition. When that
leaves a reference with nothing pushable, the combined predicate becomes TRUE
and the definition gets no push-down at all, including for its other references.
was:
{{PushdownPredicatesAndPruneColumnsForCTEDef}} collects the predicates at each
CTE reference and pushes their OR-merged combination into the shared CTE
definition. Each reference keeps its own predicates, so a non-deterministic
predicate pushed into the definition as well is evaluated twice, which can
produce wrong results.
{code:sql}
with v as (select c1, rand(1) r from t)
select c1 from v where rand(2) < 0.5
union all
select c1 from v where rand(3) < 0.5
{code}
The definition is non-deterministic and referenced twice, so it survives
{{InlineCTE}}. {{(rand(2) < 0.5) OR (rand(3) < 0.5)}} lands in the definition
while both references keep their own filter, so the optimized plan holds four
{{rand}} filters instead of two, and rows are dropped twice.
No {{MATERIALIZED}} option is needed to reach this: any definition that
survives {{InlineCTE}} has the same shape. The rule's scaladoc claims
determinism is taken care of by {{ScanOperation}}, but SPARK-39764 (3.4.0)
replaced that with {{PhysicalOperation}}, which hands back a single filter even
when it is non-deterministic.
Fix: only push deterministic predicates into the CTE definition.
> Non-deterministic predicates should not be pushed down to CTE definitions
> -------------------------------------------------------------------------
>
> Key: SPARK-59434
> URL: https://issues.apache.org/jira/browse/SPARK-59434
> Project: Spark
> Issue Type: Bug
> Components: SQL
> Affects Versions: 3.4.0
> Reporter: Cheng Pan
> Priority: Major
> Labels: pull-request-available
>
> {{PushdownPredicatesAndPruneColumnsForCTEDef}} collects the predicates at
> each CTE reference and pushes their OR-merged combination into the shared CTE
> definition. Each reference keeps its own predicates, so a non-deterministic
> predicate pushed into the definition as well is evaluated a second time,
> which can produce wrong results.
> {code:sql}
> create or replace temp view t as select * from values (0), (1), (2) as t(c1);
> with v as (select c1, rand(1) r from t)
> select c1 from v where rand(2) < 0.5
> union all
> select c1 from v where rand(3) < 0.5;
> {code}
> The definition is non-deterministic and referenced twice, so it survives
> {{InlineCTE}}. {{(rand(2) < 0.5) OR (rand(3) < 0.5)}} lands in the definition
> while both references keep their own filter, so the optimized plan holds four
> {{rand}} filters instead of two, and rows are dropped twice.
> No {{MATERIALIZED}} option is needed to reach this: any definition that
> survives {{InlineCTE}} has the same shape.
> The rule has behaved this way since it was added in SPARK-37670 (3.4.0). Its
> scaladoc claimed determinism was taken care of by {{ScanOperation}}, but that
> was never true -- {{ScanOperation}} ran with {{legacyMode = false}}, whose
> guard admits the first filter whatever it is. {{PhysicalOperation}}, swapped
> in later by SPARK-39764 (3.4.0), behaves the same way here: it returns a
> single filter even when it is non-deterministic.
> Fix: only push deterministic predicates into the CTE definition. When that
> leaves a reference with nothing pushable, the combined predicate becomes TRUE
> and the definition gets no push-down at all, including for its other
> references.
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
