shahwazreza opened a new pull request, #27118: URL: https://github.com/apache/datafusion/pull/27118
## Which issue does this PR close? - Closes #26029. ## Rationale for this change With `datafusion.execution.parquet.pushdown_filters = true`, an `ORDER BY ... LIMIT` returns the wrong rows when one sort key contains another sort key, for example `ORDER BY CASE WHEN b % 2 = 0 THEN p ELSE q END, b`. With other column types (e.g. `b` as `UInt64`) the same query fails with an `Invalid comparison operation` error instead. Turning off `enable_topk_dynamic_filter_pushdown` gives the right answer, so the problem is in the TopK dynamic filter that gets pushed into the Parquet scan. ## What changes are included in this PR? When the filter is pushed below a projection, `DynamicFilterPhysicalExpr::remap_children` rewrites each original sort key into the new input's columns. It used `transform_up`, so in the example above the inner `b` was rewritten first. After that the outer `CASE` no longer equalled the original sort key, so it was never rewritten, and its `p@0` reached the scan, where index 0 is `b`. - `remap_children` now uses `transform_down`, so each sort key is matched as a whole before anything inside it changes. - After a match it returns `TreeNodeRecursion::Jump`, so the replacement is not searched again. The replacement is already written against the new input; without `Jump`, a column inside it can be rewritten a second time. This is the approach that was approved in review on #26031. That PR was closed without being merged, so I've picked the issue up here. The change is in the shared remapping code, so it applies to every dynamic filter, not only TopK. ## What is the testing strategy for this PR? Two sqllogictest cases at the end of `dynamic_filter_pushdown_config.slt`, each with an `EXPLAIN` that checks TopK's filter reaches the Parquet scan, followed by a check of the query results: - A smaller version of the query from the issue. Each row is its own row group, so TopK tightens its filter before the answer rows (the last three in the file) are read. On `main` it returns `9, 10, 11` instead of `0, 1, 2`. - A projection that swaps column names, `SELECT b + 1 AS a, a AS b ... ORDER BY a, b LIMIT 3`. In the scan, the sort key `a` becomes `b@1 + 1`, and that `b@1` is equal to the other sort key `b`. This passes on `main`, but fails if `Jump` is removed, because the `b@1` gets rewritten again and the wrong rows come back. The `EXPLAIN` checks follow the review feedback on #26031, so the tests can't silently stop testing the bug if the filter no longer reaches the scan. In the first case, the middle projection's column order changes between runs, so those two plan lines use `<slt:ignore>`. Both cases can be reached from SQL, so I didn't add Rust unit tests. ## Are there any user-facing changes? Queries like the ones above now return the correct rows, or no longer fail with a comparison error. No API changes. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected] --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
