Dandandan opened a new pull request, #2435:
URL: https://github.com/apache/datafusion-ballista/pull/2435
# Which issue does this PR close?
No issue filed for this one; it addresses the "Metric definitions I couldn't
pin down from the event log" section of #2419 (stage `output_rows` looking like
`sum(op.output_rows)`, and source-stage `input_rows` being inconsistent between
plan shapes).
# Rationale for this change
`QueryStageSummary.input_rows` / `output_rows` (served by `GET
/api/job/{id}/stages`, shown in the TUI, and embedded in the event log's
`JobEnd`), the per-task `TaskSummary` counts, and the `TaskEnd` event metrics
were all computed by summing every metric named `input_rows` / `output_rows`
across **every operator** of the stage plan.
Only some operators emit `input_rows` (the shuffle writers, `HashJoinExec`'s
probe side, `RangeFilterExec`), while every operator emits `output_rows`, so
the result depended on the plan shape rather than on what the stage read or
wrote. Q8 stage 2 from #2419 (partitioned `part ⋈ lineitem`) shows it:
```
ShuffleWriterExec: partitioning: Hash([l_partkey@2], 256)
HashJoinExec: mode=Partitioned, join_type=Inner, on=[(p_partkey@0,
l_partkey@1)]
ShuffleReaderExec: upstream_stage: 0 -- 1,332,379 rows
ShuffleReaderExec: upstream_stage: 1 -- 5,999,989,709 rows
```
was reported as `input_rows: 6,039,946,559, output_rows: 6,081,235,788`:
- `input_rows` = join probe input (5,999,989,709) + writer input (39,956,850)
- `output_rows` = reader 0 (1,332,379) + reader 1 (5,999,989,709) + join
output (39,956,850) + writer output (39,956,850)
The stage actually read ~6.0B rows and wrote 39,956,850. In the same way a
`SortShuffleWriterExec <- DataSourceExec(lineitem)` stage showed `output_rows`
equal to twice the scan.
# What changes are included in this PR?
- `dto_build::StageMetricSlots` walks the scheduler's copy of the stage plan
the same way the executor's `collect_plan_metrics` flattens metrics (pre-order,
skipping operators that report no metrics) to find the slots of the root (the
shuffle writer) and of the leaves (shuffle readers, data sources). Stage and
task `input_rows` are now the `output_rows` of the leaves, and `output_rows` is
the `output_rows` of the root, optionally restricted to the task's global
partitions. If the metrics vector does not line up with the plan (nothing
reported yet, or a different shape) the counts are zero rather than
misattributed.
- `task_row_counts` (the `TaskEnd` event) uses the same slot map.
`task_end_events` receives the job's stages through a read lock on the live
graph in `QueryStageScheduler` (no graph clone per status batch); if the graph
is already gone the events are still written, with zero rows.
- Unit tests for the slot mapping, stage / task / partition-filtered counts,
unpartitioned metrics, mismatched vectors, and the raw-metrics path.
- Doc comments on the DTO fields spell out the definitions.
Not changed: `elapsed_compute` on the stage summary. It is (and was) the
wall time between the first task start and the last task end of the stage,
which is why it equals the max task duration in #2419. Only the field name is
misleading, and renaming it would be a wire-format change.
# Are there any user-facing changes?
The values of `input_rows` / `output_rows` in the stages REST endpoint, the
TUI, and the event log's `JobEnd` / `TaskEnd` records change to the definitions
above. No schema changes.
🤖 Generated with [Claude Code](https://claude.com/claude-code)
--
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]
