jayzhan211 opened a new pull request, #26757: URL: https://github.com/apache/datafusion/pull/26757
## Which issue does this PR close? - Part of #26481. - Supersedes #25688. @zhuqi-lucas agreed there that I take it over. Their work is the first commit, rebased onto main. ## Rationale for this change Physical optimizer rules run in one hand-ordered list, and the plan is only valid after `EnsureRequirements`. A custom rule list that drops or moves `EnsureRequirements` gets a plan that cannot run, and a rule that breaks a requirement has no shared way to repair it. This PR splits physical planning at enforcement. A `PhysicalAnalyzer` phase runs the rules whose output enforcement has to see, then `EnsureRequirements`, so every `PhysicalOptimizerRule` receives a valid plan and must return one. Unlike the logical analyzer, the physical analyzer may hold rules that only make the plan faster: a rule runs there when it does not need to see the repartitions and sorts that enforcement adds. Spark and Trino draw the same line, with rules that must run before or after `EnsureRequirements` / `AddExchanges`; here the line is explicit and checked. The default lists keep today's rule order, so plans and planning time do not change, and no rule moves across enforcement. The `WindowTopN` and `join_selection` fixes discussed in [#26481](https://github.com/apache/datafusion/issues/26481#issuecomment-6405070779) come before any such move, and don't depend on this PR. ```text PhysicalAnalyzer (rewrite, then enforce) PhysicalOptimizer (valid in, valid out) OutputRequirements (add) CombinePartialFinalAggregate aggregate_statistics ... EliminateConstantJoin OutputRequirements (remove) join_selection ... LimitedDistinctAggregation SanityCheckPlan FilterPushdown WindowTopN EnsureRequirements ``` It also adds `satisfy_requirements`, which inserts only the repartitions and sorts that are missing, and returns a plan that already meets its requirements unchanged. Debug builds use it to check, where the analyzer phase ends, that the phase really left a valid plan. A rule whose rewrite breaks a requirement can call it to repair the plan. ## What changes are included in this PR? One commit per step: 1. @zhuqi-lucas: the `PhysicalAnalyzerRule` trait and the `PhysicalAnalyzer` rule list, wired through `Session`, `SessionState` and `SessionStateBuilder`. The planner runs the analyzer rules before the optimizer rules. `EliminateConstantJoin`, which was added to main after the PR was opened, keeps its position before `join_selection`. Moved after enforcement, it would run after the analyzer's `FilterPushdown`, and the three plans in `eliminate_constant_join.slt` would lose their parquet `predicate=` and `pruning_predicate=`. 2. `RunInAnalyzer` runs a `PhysicalOptimizerRule` in the analyzer. It replaces eight copied `PhysicalAnalyzerRule` impls. Invariant-check errors now say which phase the failing rule ran in. 3. `ForeignSession` forwards the analyzer rules through FFI. Before this, a join planned through it kept `PartitionMode::Auto` and failed at execution. 4. The planner removes any `OutputRequirementExec` left after the optimizer rules. With a custom optimizer list that had no remove pass, the marker stayed in the final plan and could not be serialized to proto. 5. `satisfy_requirements`, a public function that inserts `RepartitionExec`, `CoalescePartitionsExec`, `SortPreservingMergeExec` and `SortExec` only where a declared requirement is unmet. It uses the same definition of "satisfied" as `SanityCheckPlan`. 6. `EnsureRequirements` first removes the enforcers that `satisfy_requirements` inserts, so running `satisfy_requirements` before it does not change its result. 7. Fix: `EquivalenceProperties` could not tell that a plan sorted on `sqrt(a)` satisfies a requirement on `sqrt(a)`; the same goes for `ln`, `log2`, `log10`, `acos`, `asin`, `acosh` and `atanh`. These functions fail to derive an ordering when the input may be outside their domain, and that failure also discarded the ordering the expression already had. No plan in the test suite changes. 8. Debug builds check where the analyzer phase ends that the plan is executable and that `satisfy_requirements` has nothing to insert. A session with no analyzer rules is not checked. 9. Docs: the optimizer contract now starts after the analyzer phase, plus upgrade notes. 10. Docs: the analyzer is defined as the rules that run before enforcement, with the contract its rules follow. 11. The observer passed to `DefaultPhysicalPlanner::optimize_physical_plan` keeps its signature. It sees each analyzer rule as a `PhysicalOptimizerRule` with the same name, so existing callers compile unchanged. 12. Docs: the upgrade guide lists who is affected and what to do (see "Downstream impact" below). ## What is the testing strategy for this PR? - Full sqllogictest suite with no golden changes, and in debug builds every query passes the new analyzer-phase check. Without commit 7, the check fails on `ORDER BY ACOSH(SIN(v1))` in `scalar.slt`. - In `core/tests/physical_optimizer/ensure_requirements.rs`, every test plan is also checked with `satisfy_requirements`: on the input it makes the plan pass `SanityCheckPlan` and then does nothing more, and on `EnsureRequirements`' output it does nothing. - `multi_smj_joins` runs `EnsureRequirements` three times. With commit 6, the second and third runs no longer move its sorts, so its snapshots change from a single-partition sort under a repartition to the one-run shape, a sort per partition above it. - New unit tests: `test_known_ordering_when_expr_properties_fail` for commit 7, and `analyzer_that_leaves_a_requirement_unmet_fails_in_debug` for commit 8. Both fail without their commit. - FFI: `test_foreign_session_runs_physical_analyzers` fails without commit 3 (`HashJoinExec: mode=Auto`), and `test_ffi_session` checks that both sessions list the same analyzer rules. - Proto's `analyze_roundtrip_unoptimized` and the FFI query planner tests pass again without clearing the analyzer list. - `output_requirements::scalar_subquery_root_preserves_global_ordering_end_to_end` now runs the analyzer rules before the optimizer rules, as the planner does. - Planning time, `sql_planner`, 3 interleaved rounds against main: | bench | this PR / main, median of 3 rounds | range over rounds | |---|---|---| | physical_plan_tpch_all | 1.011 | 0.934 - 1.014 | | physical_plan_tpcds_all | 0.996 | 0.940 - 1.000 | | physical_plan_clickbench_all | 0.989 | 0.981 - 0.989 | | physical_many_self_joins | 1.002 | 0.988 - 1.023 | | physical_select_all_from_1000 | 0.990 | 0.985 - 0.999 | | physical_join_consider_sort | 0.990 | 0.983 - 1.002 | | physical_plan_struct_join_agg_sort | 1.003 | 0.983 - 1.042 | No change beyond noise. The analyzer-phase check only exists in debug builds. Measured before the last rebase; the commits since then do not change the default pipeline's work. ## Are there any user-facing changes? Yes, documented in the 56.0.0 upgrade guide. Default plans do not change, and no public signature changes. New public API: `PhysicalAnalyzerRule`, `PhysicalAnalyzer`, `RunInAnalyzer`, `satisfy_requirements`, `Session::physical_analyzers`, `SessionState::physical_analyzers` and `add_physical_analyzer_rule`, and `SessionStateBuilder::with_physical_analyzer_rule(s)`. `EXPLAIN VERBOSE` shows the analyzer rules. ### Downstream impact I searched public code on GitHub for the APIs this PR touches. Private code can have the same patterns. | What changes | Who is affected | What to do | |---|---|---| | Rules placed in front of the default optimizer list now run after enforcement, and nothing enforces the plan again after them | InfluxDB (`iox_query` puts its rules before DataFusion's), LaminarDB (puts a streaming validator first) | Move rules that must run before enforcement into the analyzer list, wrapped in `RunInAnalyzer` | | `with_physical_optimizer_rules(vec![])` no longer gives a plan without enforcement | Arroyo (`arroyo-planner` clears the list) | Also call `with_physical_analyzer_rules(vec![])` | | `SessionState::physical_optimizers()` no longer lists the analyzer rules | datafusion-tracing (instruments each rule), Sail (builds `EXPLAIN` output from each rule), datafusion-dft (lists rule names) | Also handle `physical_analyzers()` | | A custom optimizer list that still contains analyzer rules runs them twice | lists copied from the old default | Drop those rules from the list | | A custom `Session` that overrides `physical_optimizers` gets no enforcement unless it also overrides `physical_analyzers`; planning or execution then fails | a few small projects | Return `SessionState::physical_analyzers()` | | The observer of `DefaultPhysicalPlanner::optimize_physical_plan` | Ballista, delta-rs, Sail, Spice and optd call it with an observer that ignores its arguments | Nothing: the signature is unchanged | | In debug builds, an analyzer list that leaves a requirement unmet fails at planning | new API, no existing users | End the list with `EnsureRequirements`, or repair with `satisfy_requirements` | -- 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]
