fix: LimitPushdown rule uncorrect remove some GlobalLimitExec #14245

zhuqi-lucas · 2025-01-23T05:24:01Z

Which issue does this PR close?

Closes #14204

What changes are included in this PR?

Fix LimitPushdown rule uncorrect remove some GlobalLimitExec.

Are these changes tested?

Yes, added unit test and slt testing.

zhuqi-lucas · 2025-01-23T09:20:34Z

datafusion/sqllogictest/test_files/joins.slt

@@ -4225,7 +4225,7 @@ query IIIIB
 SELECT * FROM t0 FULL JOIN t1 ON t0.c2 >= t1.c2 LIMIT 2;
 ----
 2 2 2 2 true
-3 3 2 2 true
+2 2 2 2 false


Why this change?

According to this PR, the limit push down to full join both side:
#12963

zhuqi-lucas · 2025-01-23T09:21:17Z

datafusion/sqllogictest/test_files/joins.slt

@@ -4247,8 +4247,10 @@ logical_plan
 physical_plan
 01)CoalesceBatchesExec: target_batch_size=3, fetch=2
 02)--HashJoinExec: mode=CollectLeft, join_type=Full, on=[(c1@0, c1@0)]
-03)----MemoryExec: partitions=1, partition_sizes=[1]
-04)----MemoryExec: partitions=1, partition_sizes=[1]
+03)----GlobalLimitExec: skip=0, fetch=2


Why this change?

According to this PR, the limit push down to full join both side:
#12963

So the limit will apply both side already.

I didn't think it is correct to push a limit below a Full join -- a FullJoin will create null values to match any misisng rows 🤔 So even if you limited both sides you'll still get rows out of there that shouldn't be ...

I didn't think it is correct to push a limit below a Full join -- a FullJoin will create null values to match any misisng rows 🤔 So even if you limited both sides you'll still get rows out of there that shouldn't be ...

Thank you @alamb for review, this is a good point, and we don't do anything for the full join limit in current PR.

The logical plan already push down full join limit since:
#12963

For example the following code:

## Test !join.on.is_empty() && join.filter.is_none() query TT EXPLAIN SELECT * FROM t0 FULL JOIN t1 ON t0.c1 = t1.c1 LIMIT 2; ---- logical_plan 01)Limit: skip=0, fetch=2 02)--Full Join: t0.c1 = t1.c1 03)----Limit: skip=0, fetch=2 04)------TableScan: t0 projection=[c1, c2], fetch=2 05)----Limit: skip=0, fetch=2 06)------TableScan: t1 projection=[c1, c2, c3], fetch=2

And the physical plan will apply the limit pushdown, but without current PR, the child limit will be overridden by parent limit, so it does not show the limit in physical plan before.

I suggest we can create a following issue to discuss do we need to fallback or change the full join push down limit case:
#12963

What's your opinion? Thanks a lot!

If the current datafusion code pushes limits down through FULL OUTER JOINs I agree we should file a bug and fix it.

Hi @alamb :
I looked into the code about the push down for join:

/// Adds a limit to the inputs of a join, if possible fn push_down_join(mut join: Join, limit: usize) -> Transformed<Join> { use JoinType::*; fn is_no_join_condition(join: &Join) -> bool { join.on.is_empty() && join.filter.is_none() } let (left_limit, right_limit) = if is_no_join_condition(&join) { match join.join_type { Left | Right | Full | Inner => (Some(limit), Some(limit)), LeftAnti | LeftSemi | LeftMark => (Some(limit), None), RightAnti | RightSemi => (None, Some(limit)), } } else { match join.join_type { Left => (Some(limit), None), Right => (None, Some(limit)), Full => (Some(limit), Some(limit)), _ => (None, None), } }; if left_limit.is_none() && right_limit.is_none() { return Transformed::no(join); } if let Some(limit) = left_limit { join.left = make_arc_limit(0, limit, join.left); } if let Some(limit) = right_limit { join.right = make_arc_limit(0, limit, join.right); } Transformed::yes(join) }

I think it's safe we just want to limit any result, we don't care about which line to return for the optimization.

If we want to get the accurate limit same with not limit, i think we need to remove those push down join optimization?

Thanks a lot!

I think this line looks suspicious.

Full => (Some(limit), Some(limit)),

I will work on trying to file a bug tomorrow. BTW I don't think this is introduced in your PR

I have also marked this PR as must be part of the 45 release

Thanks a lot @alamb!

I am pretty sure this a bug (not introduced by this PR):

FULL OUTER JOIN and LIMIT produces wrong results #14335

I am now reviewing the rest of this PR more carefully

Thank you for your patience @zhuqi-lucas

zhuqi-lucas · 2025-01-23T09:28:02Z

datafusion/core/tests/dataframe/mod.rs

@@ -2217,11 +2217,6 @@ async fn write_parquet_with_order() -> Result<()> {
    let df = ctx.sql("SELECT * FROM data").await?;
    let results = df.collect().await?;

-    let df_explain = ctx.sql("explain SELECT a FROM data").await?;


Nit: Remove print info

alamb

Thank you @zhuqi-lucas -- I looked at this PR carefully and it looks good to me -- both code and tests ❤️ 🦾

Thank you for the fix

alamb · 2025-01-28T12:34:57Z

datafusion/sqllogictest/test_files/joins.slt

@@ -4264,8 +4266,10 @@ logical_plan
 physical_plan
 01)GlobalLimitExec: skip=0, fetch=2
 02)--NestedLoopJoinExec: join_type=Full, filter=c2@0 >= c2@1
-03)----MemoryExec: partitions=1, partition_sizes=[1]
-04)----MemoryExec: partitions=1, partition_sizes=[1]
+03)----GlobalLimitExec: skip=0, fetch=2


This plans is incorrect due to

FULL OUTER JOIN and LIMIT produces wrong results #14335

(not this PR)

alamb · 2025-01-28T12:37:48Z

datafusion/sqllogictest/test_files/limit.slt

+03)----SubqueryAlias: t1
+04)------Limit: skip=0, fetch=10
+05)--------TableScan: testsubquerylimit projection=[a, b], fetch=10
+06)----Limit: skip=0, fetch=1


this plan looks good to me -- the Limit 1 is still here

alamb · 2025-01-28T13:13:16Z

datafusion/physical-optimizer/src/limit_pushdown.rs

-            global_state.satisfied = true;
+            // If the plan's children have limit, we shouldn't change the global state to true,
+            // because the children limit will be overridden if the global state is changed.
+            if pushdown_plan.children().iter().any(|child| {


It looks to me like most of the rest of this rule is implemented in terms of LimitExec (defined in this file) to abstract away directly looking for GlobalLimitExec and LocalLimitExec

I think you could do something like this instead to be more consistent with the rest of the codebase

if pushdown_plan.children().iter().any(|&child| { extract_limit(child).is_some() }) { global_state.satisfied = false; }

However, this works too and fixes the bug so I think it is fine as is.

I think it's a good suggestion and it makes code clear, addressed in latest PR. Thanks @alamb !

datafusion/core/tests/dataframe/mod.rs

xudong963 · 2025-01-28T15:57:46Z

datafusion/physical-optimizer/src/limit_pushdown.rs

@@ -247,7 +246,15 @@ pub fn pushdown_limit_helper(
            }
        } else {
            // Add fetch or a `LimitExec`:
-            global_state.satisfied = true;
+            // If the plan's children have limit, we shouldn't change the global state to true,


If the children's limit is >= the globe limit, can we push down the limit?

Good question @xudong963 , added the slt testing for children's limit is >= the globe limit, the limit should also support push down consistent with current behaviour, thanks!

xudong963 · 2025-01-29T13:38:36Z

datafusion/physical-optimizer/src/limit_pushdown.rs

+            if pushdown_plan
+                .children()
+                .iter()
+                .any(|&child| extract_limit(child).is_some())
+            {
+                global_state.satisfied = false;
+            }


I think the logic is strange, if comes to the else branch(248 lines), it means global_state.satisfied == false;, then here(257 lines) sets global_state.satisfied = false;

Got it @xudong963 , the previous logic will always setting true in the global_state.satisfied == false. This logic is keep the false for some cases. I change the logic to more clear that, we only setting to true to exclude the above case.
Thanks!

zhuqi-lucas · 2025-01-29T15:22:44Z

datafusion/physical-optimizer/src/limit_pushdown.rs

@@ -248,7 +247,15 @@ pub fn pushdown_limit_helper(
            }
        } else {
            // Add fetch or a `LimitExec`:
-            global_state.satisfied = true;


Here is the original logic for setting back to true.

alamb · 2025-01-29T20:48:20Z

Thanks again @zhuqi-lucas and @xudong963 -- this PR took a while but I think things are good in the end

fix: LimitPushdown rule uncorrect remove some GlobalLimitExec

9781cc1

github-actions bot added optimizer Optimizer rules core Core DataFusion crate sqllogictest SQL Logic Tests (.slt) labels Jan 23, 2025

xudong963 self-requested a review January 23, 2025 05:47

Fix some logic for maybe fetch

d3969a2

github-actions bot added the physical-expr Physical Expressions label Jan 23, 2025

Fix test

dbb20ee

github-actions bot removed the physical-expr Physical Expressions label Jan 23, 2025

zhuqi-lucas commented Jan 23, 2025

View reviewed changes

alamb mentioned this pull request Jan 28, 2025

FULL OUTER JOIN and LIMIT produces wrong results #14335

Closed

alamb approved these changes Jan 28, 2025

View reviewed changes

Address comments

87dba0b

xudong963 reviewed Jan 28, 2025

View reviewed changes

datafusion/core/tests/dataframe/mod.rs Outdated Show resolved Hide resolved

xudong963 reviewed Jan 28, 2025

View reviewed changes

zhuqi-lucas added 3 commits January 29, 2025 16:50

Merge remote-tracking branch 'upstream/main' into issue_14204

2421f91

Address comments

8263922

Add comments

ee2bda7

zhuqi-lucas requested a review from xudong963 January 29, 2025 09:14

xudong963 reviewed Jan 29, 2025

View reviewed changes

Address comments

412b104

zhuqi-lucas requested a review from xudong963 January 29, 2025 15:21

zhuqi-lucas commented Jan 29, 2025

View reviewed changes

alamb merged commit 2510e34 into apache:main Jan 29, 2025
25 checks passed

alamb mentioned this pull request Feb 4, 2025

Feb 4, 2025: This week(s) in DataFusion #14491

Open

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

fix: LimitPushdown rule uncorrect remove some GlobalLimitExec #14245

fix: LimitPushdown rule uncorrect remove some GlobalLimitExec #14245

zhuqi-lucas commented Jan 23, 2025 •

edited

Loading

zhuqi-lucas Jan 23, 2025

zhuqi-lucas Jan 23, 2025

alamb Jan 24, 2025

zhuqi-lucas Jan 25, 2025 •

edited

Loading

alamb Jan 25, 2025

zhuqi-lucas Jan 26, 2025 •

edited

Loading

alamb Jan 27, 2025

zhuqi-lucas Jan 28, 2025

alamb Jan 28, 2025

zhuqi-lucas Jan 23, 2025

alamb left a comment

alamb Jan 28, 2025

alamb Jan 28, 2025

alamb Jan 28, 2025

zhuqi-lucas Jan 28, 2025

xudong963 Jan 28, 2025

zhuqi-lucas Jan 29, 2025

xudong963 Jan 29, 2025

zhuqi-lucas Jan 29, 2025

zhuqi-lucas Jan 29, 2025

alamb commented Jan 29, 2025

fix: LimitPushdown rule uncorrect remove some GlobalLimitExec #14245

fix: LimitPushdown rule uncorrect remove some GlobalLimitExec #14245

Conversation

zhuqi-lucas commented Jan 23, 2025 • edited Loading

Which issue does this PR close?

What changes are included in this PR?

Are these changes tested?

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

zhuqi-lucas Jan 25, 2025 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

zhuqi-lucas Jan 26, 2025 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

alamb left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

alamb commented Jan 29, 2025

zhuqi-lucas commented Jan 23, 2025 •

edited

Loading

zhuqi-lucas Jan 25, 2025 •

edited

Loading

zhuqi-lucas Jan 26, 2025 •

edited

Loading