jayzhan211 commented on issue #26481:
URL: https://github.com/apache/datafusion/issues/26481#issuecomment-6405070779
@alamb thanks for going through it. Yes, it can be delivered incrementally.
Each step lands on its own and is useful by itself, so stopping after any of
them still leaves the code better than today. The proposed order at the end is
the next few steps.
I also have a new finding. I moved each rule that is tied to its position
around `EnsureRequirements` (enforcement) to the other side of it, and ran
every sqllogictest file:
- **Correctness depends on two positions today.** `join_selection` and the
`OutputRequirements` marker must run before enforcement. Every other rule gives
the same results on either side; moving it only changes the plan.
- **`WindowTopN` and `join_selection` would get slower if moved after
enforcement.** Both run before it today. Moved after it, they replace an
operator that enforcement has already placed repartitions or sorts for, and
those stay behind. I think we should fix these two before moving rules into the
optimizer phase or running the optimizer rules in a loop (#25572).
- **`EliminateConstantJoin` should stay where it is**, before enforcement
(re zhuqi-lucas's comment above, which suggested moving it to the optimizer).
Moved after enforcement, it loses the scan predicate in all 3 of its plans,
because `FilterPushdown`, which pushes its new filter into the scan, runs
before enforcement.
The two diagrams below are the main read; the evidence for each rule is in
collapsed sections, and the method is at the end.
### Today
```text
why it is here moved
across enforcement
planner joins start as PartitionMode::Auto
PhysicalAnalyzer
OutputRequirements (add) holds the query's ORDER BY INSERT
ORDER BY loses its order
aggregate_statistics EliminateConstantJoin needs it 3 plans
keep a join
EliminateConstantJoin FilterPushdown needs it 3 plans
lose a scan predicate
join_selection enforcement needs the join mode 767
SanityCheckPlan errors
LimitedDistinctAggregation matches only adjacent aggregates 6 plans
lose lim=[N]
FilterPushdown enforcement reads its row estimates 58 plans
change
WindowTopN replaces the window, not its sort 25 plans
keep a full sort
EnsureRequirements
════════════════════════════ the plan is valid from here
════════════════════════════
PhysicalOptimizer
CombinePartialFinalAggregate, OptimizeAggregateOrder, ProjectionPushdown
OutputRequirements (remove) removes the marker 5 plans
change
LimitAggregation ... PushdownSort
EnsureCooperative nothing found no change
FilterPushdown(Post) later rules rebuild the TopK 27
dynamic filters stay empty
SanityCheckPlan a check written as a rule
```
Only the `OutputRequirements (add)` and `join_selection` rows affect
correctness. The others only change plans.
### After the fixes
This is what I have found so far, and it may change as I find more.
```diff
planner
- joins start as PartitionMode::Auto
+ creates each join in a mode that can execute
+ records the query's output requirement (ORDER BY, one output partition)
PhysicalAnalyzer
- OutputRequirements (add) a marker node at the root holds the output
requirement
aggregate_statistics
EliminateConstantJoin
- join_selection resolves Auto, then chooses by statistics
+ join_selection chooses by statistics only
- LimitedDistinctAggregation matches only adjacent aggregates
+ LimitedDistinctAggregation also matches through exchanges
- FilterPushdown enforcement depends on where the filter
ends up
+ FilterPushdown same estimates either way, leftover filters
stay up
- WindowTopN leaves the window's sort behind
+ WindowTopN also drops the window's sort
- EnsureRequirements reads the output requirement from the marker
+ EnsureRequirements gets it from the planner, also inserts
CooperativeExec
═══════════════════════ the plan is valid from here
═══════════════════════
PhysicalOptimizer
CombinePartialFinalAggregate, OptimizeAggregateOrder, ProjectionPushdown
- OutputRequirements (remove) removes the marker
LimitAggregation ... PushdownSort
- EnsureCooperative
- FilterPushdown(Post) dynamic-filter links break if the plan
changes later
+ FilterPushdown(Post) nodes holding a dynamic filter guard its
links
- SanityCheckPlan
+ planner final check all requirements and the output requirement
```
`EnsureCooperative`, the final check (#26461) and the dynamic-filter guard
come from my earlier proposal. The moves only show that nothing ties them to
their current place.
### Why `OutputRequirements` goes
Today the query's output requirement (its `ORDER BY`, and one output
partition) lives in an `OutputRequirementExec` node: one rule adds it at the
root, another removes it. The node:
- must be added before enforcement. Added after it, `INSERT ... ORDER BY`
lost its order in 3 plans, and a file scan was no longer split into byte ranges
(`repartition_scan.slt:337`).
- changes what other rules do. When it is removed earlier, 5 window plans
place projections differently.
- leaks into the final plan when a custom optimizer list leaves out the
remove pass, and proto serialization then fails.
- is removed with 9 optimizer rules still to run, so no check sees whether
they keep the top-level sort.
The replacement is not another node. The planner keeps the requirement,
gives it to `EnsureRequirements` as the requirement on the root, and checks the
final plan against it. It has to keep today's scan splitting.
### `WindowTopN`
`WindowTopN` is opt-in today (`datafusion.optimizer.enable_window_topn`,
default `false`).
**Problem.** Moved after enforcement, the rule still replaces the window
with `PartitionedTopKExec`. But the full sort that enforcement placed for the
window stays underneath, in all 25 changed plans. `PartitionedTopKExec` only
needs its input partitioned on the partition keys, so the sort is wasted, and
avoiding it is the reason the rule exists.
**Fix.** When the rule replaces the window, it also drops the sort directly
below it, if that sort has no fetch and keeps its input's partitioning. With
the fix, no plan changes where the rule is today. After enforcement, 3 plans
change, all in where a repartition goes. In `window_topn.slt:1516` the top-K
runs on one partition instead of five, because enforcement placed the
repartition for the window.
For the top 3 rows per key over 20M rows, the query takes 0.034 s today,
0.435 s with the rule after enforcement, and 0.035 s after enforcement with the
fix.
<details>
<summary>Timing setup</summary>
Top 3 rows per key, 20M in-memory rows, 100 keys:
| `WindowTopN` | Time |
|---|---|
| before enforcement (today) | 0.034 s |
| after enforcement | 0.435 s |
| after enforcement, with the fix | 0.035 s |
</details>
### `join_selection`
It makes four decisions, and they need different fixes:
| Decision | Kind | Proposal |
|---|---|---|
| 1. Use `SymmetricHashJoinExec` when both inputs are unbounded | runs or
not | decide separately, with 2010YOUY01's suggestion to move it out of the
default pipeline (#24265) |
| 2. Swap so an unbounded input is never the build side | runs or not | same
|
| 3. Give each hash join an executable mode (resolve `Auto`) | correctness:
`Auto` can't execute | the planner creates joins in a mode that can execute |
| 4. Choose `CollectLeft` or `Partitioned`, and the build side, from
statistics | performance | stays in the rule |
Moved after enforcement as it is, the rule gives 767 `SanityCheckPlan`
errors:
- 604 hash joins whose mode was chosen after enforcement saw `Auto`, which
declares no requirement.
- 152 nested-loop and cross joins whose sides were swapped after enforcement
had placed a single-partition requirement on the other side.
To test decision 4 alone, I resolved `Auto` to `Partitioned` before
enforcement, made the statistics-based choice after it, and then repaired the
plan. Every repair keeps the plans valid, with no wrong rows. But inserting
what is missing, or also dropping the repartitions a `CollectLeft` join no
longer needs, changes 53 to 66 plans and makes a probe-heavy query up to 1.46x
slower. Only running `EnsureRequirements` again keeps today's plans, and it is
the most expensive physical rule.
<details>
<summary>Repair variants and timings</summary>
| Repair after the choice | Errors / wrong rows | Plans changed |
|---|---|---|
| insert what is missing (`satisfy_requirements`) | 0 / 0 | 66 |
| drop the hash repartitions a `CollectLeft` join no longer needs, then
insert | 0 / 0 | 53 |
| run `EnsureRequirements` again | 0 / 0 | 13, mostly from running it twice
(#25585); in 2 the build side flips |
- **Insert:** a join that becomes `CollectLeft` keeps the hash repartitions
placed for `Partitioned`, plus a coalesce on top.
- **Drop, then insert:** the probe side loses the round-robin repartition
that today's plan adds. All 43 that disappear had a single input partition, so
this only happens when the probe input arrives in one partition.
- **Run again:** keeps today's join plans apart from the 2 build-side flips,
but `EnsureRequirements` is the most expensive physical rule (about 28 of 75 ms
of physical optimization for all TPC-DS queries).
A 100K-row build side against a 20M-row Parquet file read as one partition
(`datafusion.optimizer.repartition_file_scans = false`), then a `GROUP BY`:
| | Time |
|---|---|
| today | 0.196 s |
| insert what is missing | 0.252 s |
| drop unneeded hash repartitions, then insert | 0.287 s |
</details>
### Proposed order
1. `WindowTopN`: drop the window's sort when it replaces the window.
2. Joins: the planner creates each join in a mode that can execute
(`Partitioned`, or `CollectLeft` where it is required), and `HashJoinExec`
rejects `Auto` as not executable. `join_selection` keeps only decision 4.
3. The planner holds the query's output requirement and `OutputRequirements`
goes (step 4 of my earlier order).
4. Open question: when `join_selection` runs after enforcement, for example
with statistics known only at runtime, how does the probe side get its
round-robin repartition back without a second enforcement pass?
These come before moving any rule into the optimizer phase.
Does this order make sense? If so, I'll open issues for 1 and 2. 4 needs a
design first.
<details>
<summary>How I tested</summary>
- Branch: #25688 rebased onto main, plus the fixes from my review of it and
`satisfy_requirements`, the insert-only part of `EnsureRequirements` (step 3 of
my earlier comment).
- Moves: analyzer rules to the start of the optimizer phase; late optimizer
rules to right after enforcement. Every sqllogictest file, including the TPC-H
plans, with the runner's 10-errors-per-file limit raised.
- Counts: records whose expected output changed, leaving out the 4 records
in explain.slt that only show rule order. Almost all are `EXPLAIN` plans. TPC-H
plans changed only when `join_selection` moved as it is.
- No query returned wrong rows. The only result differences: two
`WindowTopN` queries numbered two tied `NULL` rows the other way round; one
query without `ORDER BY` returned its rows in another order; and with the
`OutputRequirements` marker added after enforcement, one CSV query that should
fail succeeded, because its file was no longer split into byte ranges.
- Timings: `datafusion-cli` release builds, 12 partitions, median of 7 runs
on my laptop.
- I can share the patches I used for the moves.
</details>
--
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]
