andygrove opened a new pull request, #2801:
URL: https://github.com/apache/datafusion-ballista/pull/2801
# Which issue does this PR close?
No issue filed. This follows up @comphead's review question on #2765: when
debugging AQE, it would help to see which rules applied and how they changed
the plan and, as LLVM's optimization remarks do, which optimizations applied or
were skipped and why. Related: #1359 (AQE tracking).
# Rationale for this change
AQE runs its rules over a job's plan many times: once when the job starts,
after every stage finishes, and on each stage as it launches. The only way to
follow that today is DataFusion's own `debug` log, which prints the whole plan
after every rule of every pass, whether or not the rule changed it. It has no
job id, and it misses the coalesce and skew join rules, which the planner runs
on each stage directly. It also can't say why a rule did nothing, and several
AQE rules skip without a word: the skew join rule logs only one of the reasons
it can leave a join alone, and exchange reuse never says why an exchange got no
key.
# What changes are included in this PR?
- **Plan change log** (`state/aqe/plan_change_log.rs`). With `debug` logging
on for that module, each pass of rules logs every rule that changed the job's
plan, with a diff, and then the plan the pass ended with. Each line names the
job and the pass: the initial optimization, the replan after stage N finished,
the replan after reusing a finished stage, or stage N as it launches. A rule
that leaves the plan as it was logs nothing, as with LLVM's `-print-changed` or
Spark's plan change log.
- Plans are compared by their rendering, since the coalesce and skew join
rules change an exchange in place. The rendering is only taken while the log is
enabled, so with `debug` off a pass costs one `log_enabled!` check.
- Lines are matched ignoring indentation, so a node a rule puts above a
subtree shows as one added line rather than the whole subtree re-indented.
- A change that a later rule of the pass undoes is dropped. DataFusion's
`OutputRequirements` adds a node and removes it again in every pass, so without
this every replan would log two diffs and a full plan even when nothing changed.
- A pass that fails still logs what its rules changed before it failed.
- This replaces the planner's "Distributed physical plan (before/after
optimization)" dumps, which had no job id. The replan timing line now names the
job and the pass.
- **Why rules skip:**
- `OptimizeSkewedJoinRule` logs why it left a join alone: the join type
can't be split, an input isn't read from a shuffle, an input is coalesced,
already split, sorted, broadcast, range-partitioned or not hash-partitioned,
the stage passes its partitions through, the inputs have different partition
counts, or no partition was split. The last one gives each side's largest
partition and skew threshold.
- Exchange reuse logs why an exchange can't share a stage. `reuse_key`
returns an `Unshareable` reason instead of `None`: range partitioning, an
operator with a volatile expression, an operator that needs a single consumer,
or a subtree the codec can't encode. The line for a shared stage now includes
the job id.
- `CoalescePartitionsRule` logs the one bail it was silent about, leaves
with different partition counts.
- Join selection already logs each decision with its sizes and thresholds,
so it is unchanged.
- **Docs:** a "Debugging AQE decisions" section in the tuning guide.
Tests:
- `plan_change_log` unit tests cover the diff (context lines, lines left
out, a subtree moved under a new node, a node moved to another parent, a diff
too large to match) and which changes are recorded (changes made in place,
changes a later rule undoes, and a disabled log).
- `test::reuse_exchange` checks that the exchanges of a volatile subplan are
unshareable because of their volatile expression.
- I ran `examples/standalone-broadcast-join.rs` with
`RUST_LOG=ballista_scheduler::state::aqe=debug` on a standalone cluster to
check the output end to end. The sample below and the one in the tuning guide
come from that run, with the temporary file paths shortened. The replan after
stage 0 logs only its timing, since the only rule that changed the plan there
was `OutputRequirements`, adding its node and removing it again.
<details>
<summary>Sample output</summary>
```text
DEBUG ballista_scheduler::state::aqe::execution_plan::dynamic_join: AQE join
decision plan_id=0 action=CollectLeft(broadcast) partition_mode=CollectLeft
build_side=right under_threshold=true ...
DEBUG ballista_scheduler::state::aqe::plan_change_log: Job 038DQQWXQR000,
initial optimization: SelectJoinRule changed the plan:
...
AggregateExec: mode=Final, gby=[], aggr=[count(Int64(1))]
AggregateExec: mode=Partial, gby=[], aggr=[count(Int64(1))]
ProjectionExec: expr=[]
- DynamicJoinSelectionExec: plan_id=0, join_type=Inner, on=[(k@0,
k@0)] repartitioned=false
- DataSourceExec: file_groups={1 group: [[big.parquet]]},
projection=[k], file_type=parquet
+ ProjectionExec: expr=[k@1 as k, k@0 as k]
+ HashJoinExec: mode=CollectLeft, join_type=Inner, on=[(k@0, k@0)]
DataSourceExec: file_groups={1 group: [[small.parquet]]},
projection=[k], file_type=parquet
+ DataSourceExec: file_groups={1 group: [[big.parquet]]},
projection=[k], file_type=parquet
DEBUG ballista_scheduler::state::aqe::plan_change_log: Job 038DQQWXQR000,
initial optimization: OutputRequirements changed the plan:
+OutputRequirementExec: order_by=[], dist_by=Unspecified
ProjectionExec: expr=[count(Int64(1))@0 as count(*)]
AggregateExec: mode=Final, gby=[], aggr=[count(Int64(1))]
AggregateExec: mode=Partial, gby=[], aggr=[count(Int64(1))]
...
DEBUG ballista_scheduler::state::aqe::plan_change_log: Job 038DQQWXQR000,
initial optimization: CombinePartialFinalAggregate changed the plan:
OutputRequirementExec: order_by=[], dist_by=Unspecified
ProjectionExec: expr=[count(Int64(1))@0 as count(*)]
- AggregateExec: mode=Final, gby=[], aggr=[count(Int64(1))]
- AggregateExec: mode=Partial, gby=[], aggr=[count(Int64(1))]
+ AggregateExec: mode=Single, gby=[], aggr=[count(Int64(1))]
ProjectionExec: expr=[]
ProjectionExec: expr=[k@1 as k, k@0 as k]
HashJoinExec: mode=CollectLeft, join_type=Inner, on=[(k@0, k@0)]
...
DEBUG ballista_scheduler::state::aqe::plan_change_log: Job 038DQQWXQR000,
initial optimization: ProjectionPushdown changed the plan:
OutputRequirementExec: order_by=[], dist_by=Unspecified
ProjectionExec: expr=[count(Int64(1))@0 as count(*)]
AggregateExec: mode=Single, gby=[], aggr=[count(Int64(1))]
- ProjectionExec: expr=[]
- ProjectionExec: expr=[k@1 as k, k@0 as k]
- HashJoinExec: mode=CollectLeft, join_type=Inner, on=[(k@0, k@0)]
+ HashJoinExec: mode=CollectLeft, join_type=Inner, on=[(k@0, k@0)],
projection=[]
DataSourceExec: file_groups={1 group: [[small.parquet]]},
projection=[k], file_type=parquet
DataSourceExec: file_groups={1 group: [[big.parquet]]},
projection=[k], file_type=parquet
DEBUG ballista_scheduler::state::aqe::plan_change_log: Job 038DQQWXQR000,
initial optimization: OutputRequirements changed the plan:
-OutputRequirementExec: order_by=[], dist_by=Unspecified
ProjectionExec: expr=[count(Int64(1))@0 as count(*)]
AggregateExec: mode=Single, gby=[], aggr=[count(Int64(1))]
HashJoinExec: mode=CollectLeft, join_type=Inner, on=[(k@0, k@0)],
projection=[]
...
DEBUG ballista_scheduler::state::aqe::plan_change_log: Job 038DQQWXQR000,
initial optimization: DistributedExchangeRule changed the plan:
+AdaptiveDatafusionExec: is_final=false, plan_id=1, stage_id=pending,
stage_resolved=false
ProjectionExec: expr=[count(Int64(1))@0 as count(*)]
AggregateExec: mode=Single, gby=[], aggr=[count(Int64(1))]
HashJoinExec: mode=CollectLeft, join_type=Inner, on=[(k@0, k@0)],
projection=[]
...
DEBUG ballista_scheduler::state::aqe::plan_change_log: Job 038DQQWXQR000,
initial optimization: plan changed by SelectJoinRule, OutputRequirements,
CombinePartialFinalAggregate, ProjectionPushdown, OutputRequirements,
DistributedExchangeRule:
AdaptiveDatafusionExec: is_final=false, plan_id=1, stage_id=pending,
stage_resolved=false
ProjectionExec: expr=[count(Int64(1))@0 as count(*)]
AggregateExec: mode=Single, gby=[], aggr=[count(Int64(1))]
HashJoinExec: mode=CollectLeft, join_type=Inner, on=[(k@0, k@0)],
projection=[]
DataSourceExec: file_groups={1 group: [[small.parquet]]},
projection=[k], file_type=parquet
DataSourceExec: file_groups={1 group: [[big.parquet]]},
projection=[k], file_type=parquet
DEBUG ballista_scheduler::state::aqe::planner: Job 038DQQWXQR000: initial
optimization took 2.851ms
DEBUG ballista_scheduler::state::aqe::planner: Job 038DQQWXQR000: replan
after stage 0 finished took 1.645875ms
```
</details>
# Are there any user-facing changes?
Only in `debug` logs. The AQE planner logs plan changes under
`ballista_scheduler::state::aqe::plan_change_log`, and the skew join rule and
exchange reuse log why they skipped. The planner's "Distributed physical plan
(before/after optimization)" debug lines are replaced by the plan change log.
There are no API or configuration 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]
