fix: clear handled OFFSET before child recursion in LimitPushdown

[kumarUjjawal]

Which issue does this PR close?

• Closes DataFusion drops grouped row after inner ORDER BY/LIMIT and outer ORDER BY/OFFSET #22489.

Rationale for this change

LimitPushdown carries GlobalRequirements while walking the physical plan.
In the bad plan shape from #22489, an outer OFFSET was already handled above
a sort barrier, but its skip still remained in the state when recursion
continued into the child subtree. That stale skip then merged with an inner
LIMIT and reduced its fetch incorrectly, which caused a grouped row to be
dropped.

The fix is to clear skip once the limit requirement has already been handled,
while keeping fetch so valid limit pushdown into child sorts still happens.

What changes are included in this PR?

• Clear skip before recursing into children when the limit requirement is already handled.
• Keep fetch unchanged so valid TopK-style pushdown still works.
• Add a physical optimizer regression test for the exact outer OFFSET / sort barrier / inner LIMIT shape.
• Add an end-to-end sqllogictest for the SQL reproducer from DataFusion drops grouped row after inner ORDER BY/LIMIT and outer ORDER BY/OFFSET #22489.

Are these changes tested?

Yes

Are there any user-facing changes?

No API Change

[alamb]

I don't think this guards against the issue:

I reverted the code change locally

diff --git a/datafusion/physical-optimizer/src/limit_pushdown.rs b/datafusion/physical-optimizer/src/limit_pushdown.rs
index 63c4f21bd9..6164d86e53 100644
--- a/datafusion/physical-optimizer/src/limit_pushdown.rs
+++ b/datafusion/physical-optimizer/src/limit_pushdown.rs
@@ -375,14 +375,6 @@ pub(crate) fn pushdown_limits(
         (new_node, global_state) = pushdown_limit_helper(new_node.data, global_state)?;
     }

-    // Once a limit has been materialized above the current node, child
-    // subtrees should not inherit its `skip`. Keep `fetch`, but clear
-    // `skip` before recursing so child-local limits are not merged with
-    // an `OFFSET` that has already been applied.
-    if global_state.satisfied {
-        global_state.skip = 0;
-    }
-
     // Apply pushdown limits in children
     let children = new_node.data.children();
     let mut changed = false;

And then I ran the tests:

cargo test --profile=ci --test sqllogictests
...
Running with 16 test threads (available parallelism: 16)
Completed 472 test files in 9 seconds

[alamb]

Here is a proposed fix:

• Update regression test for apache/datafusion#22525 kumarUjjawal/datafusion#1

[kumarUjjawal]

Thanks you for this

[alamb]

Thank you @kumarUjjawal -- I don't fully understand the fix, but it does seem to fix the bug and I think the test coverage looks good

[alamb]

maybe worth calling out that this is a different sort order

[alamb]

I think we should also add a test (or confirm one already exists) where the two sorts DO have the same sort and ensure the limit is still pushed

[alamb]

I am surprised this doesn't need to check for the actual sort keys too 🤔