jayzhan211 opened a new issue, #26697: URL: https://github.com/apache/datafusion/issues/26697
### Is your feature request related to a problem or challenge? DataFusion is built to be extended with your own `ExecutionPlan` nodes: tracing wrappers, remote and distributed exchanges, materialized-view nodes, custom joins and limits. But most physical optimizer rules decide what is safe around a node by checking its concrete type. A custom node falls into a generic fallback, and that fallback reads the node's default answers as permission instead of as "unknown". So custom nodes can get **wrong results**, not just slower plans. Examples, all on current `main`. In each case the equivalent built-in node gives the correct answer: | Custom node | What happens | Where | |---|---|---| | projection-like node (`-a AS a`) that keeps row order | `ORDER BY a` is pushed below it, so the output is sorted by the old value (#26656) | `enforce_sorting/sort_pushdown.rs` (`handle_custom_pushdown`) | | node with default methods above the query's `ORDER BY` | the sort is removed; rows come back unsorted | `output_requirements.rs` (only walks through nodes whose `maintains_input_order()` is true) | | limit node: `Sort(a)` over custom `Limit(5)` over `Sort(b)` | the inner sort is removed, so the limit keeps arbitrary rows | `enforce_sorting/mod.rs` (only built-in limits stop the walk) | | OFFSET node (skip 2, fetch 5) | 3 rows instead of 5 | `sort_pushdown.rs` (`input_fetch` reads `skip` from `GlobalLimitExec` only), `limit_pushdown.rs` | | partition merge (like `CoalescePartitionsExec`) under `LIMIT 10` | 40 rows | `limit_pushdown.rs` (`combines_input_partitions` checks types) | | partitioned join declared through `required_input_distribution` | 59 of 200 matches | `enforce_distribution.rs` (only `HashJoinExec` / `SortMergeJoinExec` turn off subset satisfaction) | | two-child node that keeps row order on both children | the right child is sorted by the left child's column | `sort_pushdown.rs` (`handle_custom_pushdown` copies one child's requirement to all) | None of these is caught at the end of optimization, because the query's output requirement is removed (`OutputRequirements::new_remove_mode`) before `SanityCheckPlan` runs. A node that declares everything it can still misses some optimizations that built-in nodes get: limited-distinct aggregation, window TopN, and a redundant extra merge from `PushdownSort`. A node that keeps the defaults also loses parquet pruning, dynamic filters, limit pushdown into the scan, and statistics-based shortcuts. The `downcast_delegate` hook (#22559) lets a wrapper look like the node it wraps, but rules that rebuild a node from its children drop the wrapper. In a 28-query test the wrapper disappeared from 10 final plans and its `execute()` never ran, so a tracing wrapper silently records nothing there. How big it is: `physical-optimizer/src` has about 130 type tests on plan nodes. Most (about 97) only rewrite the built-in node itself and are fine. About 20 decide "can I move this past that node?" by type, which costs custom nodes an optimization. About 11, plus 4 generic fallbacks, can give wrong results. Projects that have run into this: InfluxDB IOx (#8120, #11675), datafusion-tracing (#22557, #26656), datafusion-materialized-views (#23276), datafusion-distributed (#24068), Ballista (#22415), and FFI nodes (#25155, #22329). This is not only about nodes added after optimization: Ballista's adaptive planner runs DataFusion's whole physical optimizer again after each stage finishes, over plans that already contain its own exchange and merge nodes. #### Why one EPIC instead of fixing each bug - **They have one cause.** A rule recognizes built-in nodes by type. For any other node it falls back to a path that treats the node's default answers as permission. Fixing each case separately leaves those fallbacks in place, and any new rule can add another one. - **Fixing them one hook at a time hasn't worked.** That has been the approach since #8096 (2023). Each hook serves one rule, each one also needs FFI plumbing (#22329), and the unsafe fallbacks are still there. - **Several open PRs and issues are answering the same question separately:** #26660 (limit semantics), #26118 (sort pushdown), #26461 and #26481 (the final plan check), #26351 (the optimizer contract), and #26378 (property tests). They should follow one rule, not several. - **One shared test lets people work in parallel.** A conformance test that starts with the known failures marked as expected lets several contributors take separate items, and shows progress as each failure is removed. ### Describe the solution you'd like One rule for every optimizer rule: > A rule decides what it may move past or remove around a node only from what that node declares. Checking a concrete type is fine for rewriting that operator itself, not for deciding what is safe around it. When a node doesn't declare something, the rule assumes the answer that keeps results unchanged: keep the sort, keep the limit where it is, don't treat partitions as merged. And one clarification from #8120: `maintains_input_order` promises row order only. A rule must not infer from it that column values or positions are unchanged. This follows what has been said before: node semantics should carry the constraints (#25688), type checks are a stepping stone (#8073), and custom pipelines are guaranteed correct results even if not every optimization (#26351). Most of this needs no new API. Looking at every place where a decision depends on a node's type, five properties cover nearly all of the wrong-result cases, and most already exist or are in flight: - **limit semantics** (`skip`, `fetch`, merges with a fetch): `limit_behavior()` in #26660 - **co-partitioned children**: `input_distribution_requirements().is_co_partitioned()` - **merges partitions**: derivable from input and output partition counts - **row count unchanged**: `cardinality_effect()` - **which input columns an output column comes from**: start without new API by validating instead of inferring, as #26118 does. Rebuild the node over the candidate child and accept a pushdown only if the node's own equivalence properties still satisfy the requirement. A declared column mapping is only worth adding if something is still missing after that. Proposed work, roughly in order (each item will become its own issue with a runnable test): - [ ] A test for custom nodes: run a set of queries with nodes wrapped in (a) a pass-through node that declares everything and (b) a node that keeps the defaults, and check that results match the unwrapped plan, and that (a)'s plan matches apart from the wrapper. Known failures start as expected failures and are removed as the items below land. Related: #26378. - [ ] Keep the query's output requirement until the final check, so a lost `ORDER BY` becomes an error instead of a wrong result. This overlaps #26481 and #26461. - [ ] #26656: sort pushdown through custom nodes (#26118) - [ ] Treat any node with a `fetch()` as a boundary that sorts don't move across, and read `skip` from `limit_behavior()` instead of from `GlobalLimitExec` - [ ] Limit pushdown under a custom partition merge: decide "merges partitions" without a type check - [ ] Decide co-partitioned joins from the declaration instead of the type, and pick a safe default for nodes that require key partitioning on two children - [ ] Sort pushdown into multi-child custom nodes: don't copy one child's requirement to the others - [ ] Rules that rebuild nodes keep `downcast_delegate` wrappers, and a wrapped window isn't computed twice - [ ] Smaller missed optimizations for nodes that declare everything (good first issues): limited-distinct aggregation and limit pushdown through `cardinality_effect()`, window TopN and TopK repartition looking through nodes by declared properties, no extra merge from `PushdownSort` - [ ] Only if still needed: a declared output-to-input column mapping ### Describe alternatives you've considered - **Add a hook per rule as users ask** (for example #23276). This is local and incremental, but the hooks end up specific to one rule, every new hook also needs FFI plumbing (#22329), and it doesn't fix the unsafe fallbacks, which is where the wrong results come from. - **Make `downcast_delegate` the main mechanism.** It only helps wrappers, not new operators. Wrappers are dropped when a rule rebuilds a node, and the hook doesn't work across FFI (#25155). - **Ask users to rewrite their plans into shapes the built-in rules recognize** (#26351). This helps with missed optimizations but not with wrong results. ### Additional context This supersedes #8096, whose projection-pushdown part is done (`try_swapping_with_projection`, #14235). Related: #9146, #23276, #25155, #8120, #26481. I'm looking for a supervising maintainer. I can help with design and review, and I'd like contributors to pick up the items. I'll file each item as its own issue with a self-contained test. -- 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]
