Statistics collection can wedge permanently: no timeout on layered query chains and a one-way in-flight flag

[aglinxinyuan]

What happened?

Coordinator statistics collection has no timeout and a one-way in-flight flag, so a single lost queryStatistics reply can permanently wedge region advancement and permanently degrade stats — with no error ever surfaced.

Three compounding defects in QueryWorkerStatisticsHandler.scala:

1. No timeout on layered query chains. The handler traverses the sender's upstream cone layer by layer, and each layer awaits all replies before the next is emitted: Future.collect(futures).flatMap(_ => processLayers(rest)) (:171) — no timeout, no fallback. PortCompletedHandler chains all port-completion bookkeeping (including region completion / scheduling the next region) on that future, so one dropped reply-expecting queryStatistics message wedges region advancement forever.
2. globalQueryStatsOngoing is set optimistically and cleared only on success. It is set to true at :101 when a full-graph query starts and cleared only inside the success .map (:183-185). If the chain never resolves, every later full-graph query short-circuits to the stale-cache branch (:86-91) for the life of the execution — statistics silently freeze.
3. Dead workers keep getting queried. ExecutionUtils.aggregateStates returns COMPLETED only when all worker states are COMPLETED or all are TERMINATED (:95-96); a mixed COMPLETED/TERMINATED set falls through to UNKNOWN (:98-107), so the operator is not skipped by the completed-operator check and its already-stopped workers are queried again — feeding defect 1, since those replies may never come. Worker states are additionally set to TERMINATED one at a time from Pekko dispatcher threads during teardown (RegionExecutionManager.scala:214-219), concurrent with these coordinator-thread reads.

lost queryStatistics reply
  └─> layer's Future.collect never resolves           (defect 1)
        ├─> portCompleted continuation never runs -> region advancement wedged
        └─> globalQueryStatsOngoing stuck true        (defect 2)
              └─> all later stats polls serve stale cache

Suggested direction: a .within(...) per layer with a logged fallback (treat missing metrics as stale rather than blocking), clear globalQueryStatsOngoing in a transform/ensure rather than map, and make the mixed COMPLETED/TERMINATED aggregate count as completed for the skip check.

How to reproduce?

Any path that loses a single worker→coordinator queryStatistics reply (e.g. a worker stopped between EndWorker acknowledgment and a late-layer stats query during teardown — reachable, see the analysis in #6916). The execution then hangs with no ERROR log, and stats stop updating.

Version/Branch

main (observed at 429be11; discovered during the investigation for #6916).