abstractdog opened a new issue, #2827:
URL: https://github.com/apache/datafusion-ballista/issues/2827
**Describe the bug**
With adaptive query execution on, TPC-DS q23b (SF10, partitioned Parquet)
never finishes: all stages but one complete, the last one's tasks all succeed
on the executors, and the job stays running until it is cancelled. With
`ballista.planner.adaptive.enabled=false` it completes.
The scheduler drops those tasks' status updates.
`RunningStage::update_task_metrics` returns an error when a task reports a
different number of operator metric sets than the stage's earlier tasks:
```
Error updating task metrics to stage 30, task metrics array size 8 does not
equal with the stage metrics array size 7 for task 15
```
`AdaptiveExecutionGraph::update_task_status` (and the same code in
`StaticExecutionGraph` and `PipelinedExecutionGraph`) propagates it with `?`,
which abandons the rest of the update: the task is never marked successful, so
the stage never completes.
Tasks of one stage report different numbers of metric sets because of the
per-task plan restriction in `task_builder.rs`. The stage's plan is:
```
ShuffleWriterExec (240 partitions)
SortExec TopK(fetch=100)
UnionExec
RepartitionExec RoundRobinBatch(120), input_partitions=1
EmptyExec <- AQE's replacement of an empty
branch
ProjectionExec
AggregateExec FinalPartitioned
ShuffleReaderExec (stage 29, Hash 120)
```
Restricting a task's plan to its partition slice splits the slice across the
`UnionExec` children. For a child whose sub-slice is empty, the leaf
`EmptyExec` is replaced by a 0-partition `DataSourceExec`
(`MemorySourceConfig`). `EmptyExec` reports no metrics (`metrics()` is `None`)
and `DataSourceExec` does, and `collect_plan_metrics` skips `None`, so those
tasks report one more metric set than the others.
**To Reproduce**
Run TPC-DS q23 with AQE on (the default) against partitioned Parquet, where
AQE replaces one `UnionExec` branch with an `EmptyExec`. The scheduler log
shows the error above, and the job hangs.
A unit test that completes an AQE job whose tasks report different numbers
of metric sets fails with:
```
Internal("Error updating task metrics to stage 1, task metrics array size 2
does not equal with the stage metrics array size 1 for task 1")
```
**Expected behavior**
The job completes. Metrics only feed the plan display and the REST API, so a
task whose metrics cannot be merged should still count as succeeded.
**Additional context**
The display code (`print_stage_metrics`, the REST DTOs) already tolerates
mismatched metrics.
The `MemorySourceConfig` stand-in exists because a 0-partition `EmptyExec`
or `PlaceholderRowExec` does not survive a protobuf round-trip:
apache/datafusion#23643 (in 55.0.0) encodes their partition count, but decodes
it with `partitions.max(1)`, so 0 comes back as 1. If DataFusion could tell an
absent count from 0, the restriction could use `with_partitions(0)` instead,
and every task of a stage would report the same metrics. That would fix the
mismatch itself; the hang should be fixed regardless, since whether a job
completes should not depend on its tasks' metrics.
--
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]
