feat(amber): supporting consistent operator stats retrieval

[shengquan-ni]

This PR introduces consistent operator retrieval in QueryStats by adopting a reverse-topological strategy.

Problem

Previously, we retrieved operator statistics by sending direct control messages to each worker independently. This approach could result in inconsistent snapshots: for example, a downstream operator might appear as completed while its upstream is still running, which is logically impossible in a pipelined execution.

Consistent Retrieval via Reverse-Topological Order

We now compute a layered topological order of operators, where operators in the same layer have the same rank, and retrieve stats layer by layer in reverse topological order. Only after all relevant stats are retrieved do we update the execution state and send the update to the frontend. This guarantees that downstream stats are not visible before their upstream stats are available, maintaining a consistent global view.

This approach increases the time complexity to O(# of layers), compared to the previous O(1) method. Therefore, we should avoid issuing stats queries too frequently, as doing so may flood the control message queue and delay other control operations.

SubDAG Query and Race Condition Mitigation

In some scenarios—such as querying the stats of a completed worker—we now query the entire subDAG rooted at that operator to ensure upstream context is also retrieved. This avoids inconsistencies in localized queries.

However, this introduces the possibility of a race condition:

1. A global query stats request is fired.
2. Operator A's stats are retrieved at timestamp T₀.
3. A subDAG stats query (including operator A) is fired.
4. Operator A's stats are retrieved again at timestamp T₁.
5. The subDAG query finishes and updates the execution state with timestamp T₁.
6. The earlier global query finishes and overwrites A's stats with older data from T₀.

To prevent this, we now attach a nanosecond-level timestamp to each execution state update, and only allow updates with newer timestamps to overwrite the existing state.

[aglinxinyuan]

Need more testing. Will do it offline.