fix: LimitPushdown rule uncorrect remove some GlobalLimitExec

[zhuqi-lucas]

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]

Why this change?

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

[zhuqi-lucas]

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.

[alamb]

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 ...

[zhuqi-lucas]

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!

[alamb]

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

[zhuqi-lucas]

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!

[alamb]

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

[zhuqi-lucas]

Thanks a lot @alamb!

[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]

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]

This plans is incorrect due to

• FULL OUTER JOIN and LIMIT produces wrong results #14335

(not this PR)

[alamb]

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

[alamb]

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.

[zhuqi-lucas]

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

[xudong963]

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

[zhuqi-lucas]

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]

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;

[zhuqi-lucas]

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]

Here is the original logic for setting back to true.

[alamb]

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