Add latency tracking for MSE adaptive routing

[timothy-e]

#18553 added in-flight request tracking for MSE in the adaptive routing stats. This PR adds per-server latency. It looks very large, but much of it is tests, and some new files were introduced, which comes with a certain amount of boilerplate in Java.

1. The planner marks stages as needing timing. It assigns FragmentTypes and then the dispatcher uses those to turn those roles into timing collection in AdaptiveRoutingStageClassification.java. it enables collection only on trusted stages that either directly receive from pure leaves or are singleton-leaf stages themselves, while excluding stages fed by non-leaf senders so their timings do not include upstream cascade delay.

2. We increment server's inflight requests before submitting. In the inflight request PR, we did it after because we didn't want to extract the servers from submit before calling it. We thought submit would be near instant, so it wouldn't matter. In practice, we saw a case (reproduced by issuing a SIGSTOP) where the submit times out and MSE adaptive routing couldn't track it.

3. Stages track their elapsed time and return it as UPSTREAM_SERVER_RESPONSE_TIMES_MS stat (BaseMailboxReceiveOperator, BlockingMultiStreamConsumer, AdaptiveRoutingUpstreamTimings)

Each BaseMailboxReceiveOperator tracks per-sender wall-clock elapsed time (from when its BlockingMultiStreamConsumer.OfMseBlock is constructed until each sender's EOS arrives). The start time is captured at consumer construction time, not at query start, to avoid pipeline-breaker inflation: if a pipeline breaker on this stage blocks for 1000ms waiting on a slow sender, fast senders' EOS blocks sit unconsumed in their queues during that time, and measuring from query start would report ~1000ms for all of them.

These per-sender timings are encoded as "hostname|port=elapsedMs;..." in a new UPSTREAM_SERVER_RESPONSE_TIMES_MS stat key on BaseMailboxReceiveOperator.StatKey and propagated up through the query stats. A serialized string is used because it is more straightforward than adding a Map type to the StatKey system.

4. Broker reads leaf-server timings and resolves hostname|mailboxPort sender keys to full instance IDs and accumulates the maximum observed latency per instance across all trusted stages. A decrementedServers set prevents double-recording.

5. Query timeout: on query cancellation (e.g. reaching the timeout), we try to cancel with stats. We preserve completed timings and inject elapsed time for pending senders.

Testing

1. Performance impact: We ran some performance tests and saw no impact.

2. Manually with tc

Choose a server that we want to induce latency on. ssh to that server and:

sudo tc qdisc add dev ens5 root handle 1: prio
sudo tc qdisc add dev ens5 parent 1:3 handle 30: netem delay 1200ms

sudo tc filter add dev ens5 protocol ip parent 1:0 prio 3 u32 \
  match ip dport 8442 0xffff flowid 1:3

We induced latency only for MSE to validate that adaptive routing can be influenced by only MSE queries.

To reset the latency

sudo tc qdisc del dev ens5 root # cleanup

We saw the score spike for the targetted servers.
3. Chaos Agent
We kicked off a chaos agent run that does

1. 90s of 600ms MSE latency
2. 30s with no faults
3. 90s of 600ms SSE latency
4. 90s with no faults
5. 90s of 1200ms MSE latency
6. 30s with no faults
7. 90s of 1200ms SSE latency

(query timeout is 1s).

We did a similar experiment with SIGSTOP issued to a single server's Pinot process.

[codecov-commenter]

Codecov Report

❌ Patch coverage is 83.23864% with 59 lines in your changes missing coverage. Please review.
✅ Project coverage is 64.81%. Comparing base (25e378b) to head (c7682df).
⚠️ Report is 42 commits behind head on master.

Files with missing lines	Patch %	Lines
.../pinot/query/service/dispatch/QueryDispatcher.java	78.66%	22 Missing and 10 partials ⚠️
...e/dispatch/AdaptiveRoutingStageClassification.java	73.46%	7 Missing and 6 partials ⚠️
...rvice/dispatch/AdaptiveRoutingUpstreamTimings.java	83.78%	3 Missing and 3 partials ⚠️
...ava/org/apache/pinot/common/datatable/StatMap.java	20.00%	0 Missing and 4 partials ⚠️
...y/runtime/operator/BaseMailboxReceiveOperator.java	91.89%	1 Missing and 2 partials ⚠️
...uery/planner/physical/DispatchablePlanContext.java	83.33%	0 Missing and 1 partial ⚠️

Additional details and impacted files

@@             Coverage Diff              @@
##             master   #18791      +/-   ##
============================================
+ Coverage     64.79%   64.81%   +0.01%     
  Complexity     1322     1322              
============================================
  Files          3393     3396       +3     
  Lines        211235   211550     +315     
  Branches      33206    33274      +68     
============================================
+ Hits         136878   137122     +244     
- Misses        63303    63350      +47     
- Partials      11054    11078      +24     

Flag	Coverage Δ
custom-integration1	100.00% <ø> (ø)
integration	100.00% <ø> (ø)
integration1	100.00% <ø> (ø)
integration2	0.00% <ø> (ø)
java-21	64.81% <83.23%> (+0.01%)	⬆️
temurin	64.81% <83.23%> (+0.01%)	⬆️
unittests	64.81% <83.23%> (+0.01%)	⬆️
unittests1	57.06% <82.95%> (+0.04%)	⬆️
unittests2	37.10% <9.37%> (-0.05%)	⬇️

Flags with carried forward coverage won't be shown. Click here to find out more.

☔ View full report in Codecov by Harness.
📢 Have feedback on the report? Share it here.

🚀 New features to boost your workflow:

• 📦 JS Bundle Analysis: Save yourself from yourself by tracking and limiting bundle sizes in JS merges.

[timothy-e]

Hey @yashmayya, I re-opened this one to replace #18646 because I force-pushed while that one was closed.

It's 99% the same code, but with two bug fixes based on what we've seen with it deployed to prod - 402e089 and 5434659.

[xiangfu0]

This adds the trusted-stage/timing setup for the unary path, but submitAndReduceWithStream() still never does the equivalent classification / COLLECT_UPSTREAM_TIMING_KEY wiring and still decrements every server with -1. So a query that enables STREAM_STATS silently loses the new MSE latency signal entirely, including submit-timeout cases. Can we plumb the same flow into stream mode, or explicitly gate adaptive-routing latency off there?

[timothy-e]

Integrating this with the stream mode is best left for a follow up PR, because it will expand the scope and require a lot more testing than what I've already done.

Explicitly turning it off would require reverting some of the changes added in the streaming PR, which also feels weird to do in this PR.

Since the current state is "streaming stats doesn't track latency for MSE adaptive routing," I think it makes sense to leave it as is and create a github issue to support it.

[xiangfu0]

Unconditionally trusting stage 0 looks unsafe. When the broker receives from an intermediate stage, this mailbox timing already includes upstream cascade delay, and extractMaxTimingsPerInstance() later takes Math::max across all trusted stages. If the same server also ran a leaf stage, the stage-0 value can overwrite that server's real leaf latency and bias the adaptive-routing EMA upward. Can we only trust stage 0 when its direct senders are leaf stages, or otherwise ignore stage-0 timings once a trusted leaf stage already attributed that instance?

[timothy-e]

Nice catch. In my original iterations, I found this to be helpful. But in the iterations since then, it seems it was no longer necessary. I removed these lines and retested and found that we still clearly identify the problematic server.

[xiangfu0]

Found one high-signal correctness issue; see inline comment.

[xiangfu0]

This classification only looks at SINGLETON inputs. A stage that scans a table and also has a HASH/BROADCAST mailbox receive from a non-leaf stage still returns LEAF here, and a stage with both a leaf SINGLETON input and any non-leaf input still returns SINGLETON_LEAF. AdaptiveRoutingStageClassification then trusts/tracks those stages as if they were pure leaves, so the elapsed times from their receive operators include upstream cascade delay and can poison the adaptive-routing EMA for healthy servers. Please downgrade any fragment that has any mailbox receive from a non-leaf sender to INTERMEDIATE (or otherwise exclude it from trusted/tracked), and add a regression test for the mixed-input topology.

[timothy-e]

This is an awesome suggestion, thanks Xiang! I implemented it, retested, and the results are much better!

Repeating the testing from 2 days ago, we see 25% lower error counts during induced server latency with this latest change.