Skip to content

[Query] Fix scheduler stop races and query attempt context leaks - #19224

Merged
JackieTien97 merged 5 commits into
masterfrom
ty/query-attempt-cleanup
Nov 13, 2026
Merged

JackieTien97 merged 5 commits into
masterfrom
ty/query-attempt-cleanup

Conversation

@JackieTien97

@JackieTien97 JackieTien97 commented Nov 11, 2026 •

Copy link
Copy Markdown
Contributor

Description

Fix two cleanup failures after partial query fragment dispatch: a stop request could be lost before scheduler publication, and a DataNode query context could wait indefinitely for an FI that was never dispatched.

Example

Query Q plans FI1, FI2 and FI3 on DN1. FI1 starts, FI2 fails with status 719, and FI3 is never dispatched. Closing (Q, 0) now stops further dispatch, cancels the actual registered FIs and releases their shared context when cleanup finishes. A retry uses (Q, 1), so delayed cancellation and release callbacks from the first attempt cannot remove the retry's context.

Scheduler and query lifecycle

  • Create schedulers without starting them, publish them through a per-attempt handoff and retain stops that arrive before publication.
  • Keep scheduler start/stop idempotent and abort topology dispatch waits. Use immediate termination for a failed or incomplete dispatch, preserving the normal completion grace period.
  • Terminal QueryExecution.stopAndCleanup completes the query state and wakes status waiters, including when dispatch is still blocked. Retry cleanup remains separate and preserves the query's reusable UDF scope. An existing terminal failure is preserved.
  • Release LOAD constructor-owned memory when stop wins before start. Preserve transferred UDF scopes through FI creation and release them when cancellation prevents an execution listener from being installed.

DataNode state and receiver-local admission

QueryAttemptRegistry owns shared state by (queryId, attemptId). Normal completion retains the planned-FI accounting needed between sequential arrivals; explicit closure releases shared state after the actual users finish, without waiting for unsent FIs.

Admission uses a token issued by the receiving DataNode through fetchQueryAdmissionToken. The token contains a process identifier, a deadline on that receiver's monotonic clock and a validity duration. Execution timeouts are also relative durations measured locally. No coordinator wall-clock timestamp or process-wide historical maximum is used for admission or expiry.

Token issuance is stateless, so a delayed issuance RPC cannot create or reopen an attempt. The coordinator caches tokens per destination within an attempt, normally adding one small RPC per destination. Transport retries reuse their FI's original token. Aborting dispatch freezes the latest published token for each destination; a token response received afterwards cannot publish another FI.

The requested token lifetime budgets both the initial RPC and the reconnect retry, capped by the FI's remaining relative query timeout. The socket budget is twice the configured internal DataNode timeout, or 120 seconds for an unlimited socket timeout; the receiver also caps validity at one day. For the default 60-second query timeout, a fast query therefore does not retain a fence for the full 120-second socket budget. Once shared query data is released, only a lightweight closure record remains until that receiver-issued cutoff. Expired tokens remain rejected after the record is removed, and process identifiers reject tokens issued before a receiver restart.

Admission expiry or a stale issuer returns retryable DISPATCH_ERROR, allowing a fresh query attempt when execution budget remains. Actual execution expiry fails FIs with QueryTimeoutException (720). Expiring an already closed record does not add a timeout cause to a previously cancelled or completed FI. Admission-expiry messages include the receiver's deadline and current monotonic reading.

Closed-record retention cost

Once the last actual user releases an attempt, the live entry is replaced atomically with a compact closed marker. Closed markers are kept outside the execution-timeout scan and indexed by an expiry queue. Cleanup processes due markers instead of scanning every retained tombstone. FragmentInstanceManager schedules attempt expiry on its existing instanceManagementExecutor, alongside the other maintenance tasks.

Duplicate closure does not add another expiry event unless it extends the cutoff. An old expiry event removes only its own marker, preserving a newer fence. The cutoff is never shortened after token publication.

A synthetic comparison against 2dec2e6b57 retained 1,200,000 closed attempts on one registry (Corretto 17, Serial GC, 768 MiB heap):

Measurement Previous implementation This change
Live heap delta, measured with GC.class_histogram 245.1 MiB 224.2 MiB
Marker shallow size 56 bytes 32 bytes
Cleanup with none due, median of 7 calls 66.30 ms <0.01 ms
Expire the first 1% 105.10 ms 12.55 ms
Drain all remaining records 68.28 ms 422.16 ms

This is an uncontended data-structure experiment, not a production throughput/RSS measurement. The priority queue makes a full burst drain slower; expiry runs serially with the other tasks on the shared management executor. Both versions released all entries; the updated empty containers retained about 14.4 MiB of capacity versus 7.9 MiB previously. Shorter per-query token validity reduces the number of retained records separately from this fixed-count comparison.

Compatibility

Attempt/token fields are optional in the dispatch and cancellation envelopes, and the binary FI body remains unchanged. An old receiver's UNKNOWN_METHOD response selects the original protocol. The fallback recognizes the wrapped Thrift exception and returns the invalidated pooled client before borrowing a replacement, without consuming the FI's reconnect retry. Full attempt isolation requires metadata-aware endpoints.

Validation

  • JDK 17 targeted reactor clean test on the PR branch based on 54bff3d9a4: 142 tests passed across 21 classes, with zero failures, errors or skips (138 DataNode tests and 4 UDF common-module tests).
  • IoTDBInternalLocalReporterTest: 25/25 passed, including both previously failing CI cases.
  • New regression coverage includes serialized admission/cancellation RPCs, no scanning of unexpired closed markers, duplicate and extended fences, and relative timeout caps.
  • Regression coverage includes terminal cleanup during blocked dispatch, existing failure preservation, full socket-timeout recovery, retryable admission expiry, independent clock origins, monotonic counter wraparound, receiver restart, delayed token issuance, legacy client replacement, bounded closure records and UDF cleanup.
  • mvn test-compile -DskipTests: passed for the full reactor.
  • mvn test-compile -P with-zh-locale -DskipTests: passed for the full reactor.
  • Formatting, Checkstyle and English/Chinese message key/placeholder parity were checked.

Multi-node cluster and rolling-upgrade integration tests have not been run.

This PR has:

  • been self-reviewed.
  • added comments explaining lifecycle ownership and concurrency decisions.
  • added or updated unit tests for the changed code paths.
Key changed/added classes

QueryExecution, ClusterScheduler, FragmentInstanceDispatcherImpl, SimpleQueryTerminator, QueryAttemptId, QueryAttemptRegistry, FragmentInstanceManager, and FragmentInstanceContext.

Publish schedulers before starting them and preserve stop requests across creation. Isolate DataNode contexts by query attempt, close partially dispatched attempts, and bound late-dispatch fences with admission deadlines.

Add regression coverage for concurrent lifecycle transitions, retry isolation, partial dispatch cleanup, RPC compatibility, and admission expiry.

@JackieTien97 JackieTien97 left a comment

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Four correctness issues remain in the scheduler stop and admission paths; please address the inline comments before merging.

Validation on 0030ac2: a targeted JDK 17 clean reactor test run passed all 38 tests across 9 classes. Four additional focused unit probes reproduced the behaviors described below. Multi-node cluster and rolling-upgrade integration tests were not run.

Comment on lines +152 to +153
if (stopped) {
return;

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

[P1] Complete the query state when terminal cleanup stops dispatch

stopped does not imply that the query state machine is terminal: Coordinator.cleanUpStaleQueries() calls cleanupQueryExecution() -> stopAndCleanup(timeout) directly, without calling cancel() or transitioning the query to FAILED. If that cleanup runs while dispatch is blocked, this branch discards the dispatch result when it eventually returns, leaving the query in DISPATCHING. The state tracker has also been aborted, so the subsequent QueryExecution.getStatus() waits indefinitely and retains the RPC thread. A latch-controlled test with a READ query reproduced this: stopAndCleanup during dispatch, let dispatch return, then getStatus never completes. Terminal cleanup needs to complete the query state or otherwise wake its status waiters, while preserving the separate stop semantics used for retries.

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Three additions from a second pass.

This is new with the change. On base, a stop that arrived during dispatch was lost because the scheduler had not been published yet, so start() went on to RUNNING and getStatus() returned. Delivering that stop is the point of this PR; what is missing is completing the query state afterwards.

Why the stale-query sweep reaches a query that is still dispatching. cleanUpStaleQueries() skips executions for which isActive() is true, and isActive() returns startTimeOfCurrentRpc == -1, which is true when no RPC is in flight. So the sweep leaves idle executions alone and acts on the ones whose RPC is still running, including the initial executeStatement call that is inside start(). That looks inverted relative to the Javadoc of cleanUpStaleQueries(). It dates from #17358 and is outside this PR, so it probably deserves its own issue. The trigger here still needs dispatch to outlast timeout + 60 s.

A candidate fix, checked. Remember the cause in stop(). In this branch, if the state is not terminal yet, call transitionToFailed(cause), or transitionToAborted() when the cause is null, before returning. The probe uses a real QueryExecution and a real ClusterScheduler with a mocked dispatcher, and calls stopAndCleanup(cause) while dispatch() is blocked:

state after start() returns getStatus()
control, no stop RUNNING returns 200
this commit DISPATCHING still blocked after 5 s
fix, stop with the sweep's QueryTimeoutRuntimeException FAILED returns 720
fix, stop with a null cause ABORTED returns 301

ClusterSchedulerLifecycleTest, QueryExecutionLifecycleTest and QueryExecutionRetryTest still pass with it. Retries are not affected, because retry() only stops a scheduler whose start() has already returned.

The variant I ran had no explicit isDone() check. Add one: when the stop comes from the terminal-state listener, transitionToFailed(Throwable) would otherwise fill an empty failureException on a query that failed with a status only.

Comment on lines +812 to +816
if (deadline == 0) {
deadline =
Math.min(
queryDeadline, LongMath.saturatedAdd(System.currentTimeMillis(), admissionWindowMs));
instance.setAdmissionDeadline(deadline);

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

[P2] Preserve recovery after the first RPC exhausts its socket timeout

The admission window equals getDnConnectionTimeoutInMS(), but this branch only initializes the deadline once. When the first sendFragmentInstance call fails after waiting for its socket timeout, dispatchRemote() reconnects and resends the FI with an already expired admission deadline. The receiver then returns QUERY_TIMEOUT (720), which ClusterScheduler.needRetry() does not retry, even if the query still has ample execution budget and the first request never reached the receiver. This was reproduced with a 20 ms socket timeout and a one-hour query timeout. Preserve the immutable cutoff for an old request, but distinguish admission expiry from execution timeout and allow a fresh query attempt, or otherwise budget admission for the supported transport retry.

Comment on lines +90 to +94
long currentTime = now();
if (admissionDeadline <= currentTime
|| (existing != null
&& (existing.closed || existing.executionDeadline <= currentTime))) {
throw rejected(key, admissionDeadline <= currentTime);

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

[P2] Avoid comparing coordinator deadlines directly with receiver wall time

queryDeadline and admissionDeadline are produced using the coordinator's wall clock, whereas now() reads the receiving DataNode's clock. With the default 60-second admission window, a receiver whose clock is 61 seconds ahead rejects a freshly issued FI immediately, even when its query timeout is one hour and there is no network delay. A clock-injected test reproduced this; the same request using the legacy relative-timeout path is admitted. This introduces a cluster clock-synchronization requirement that the repository's clock assumptions explicitly do not provide (THREAT_MODEL.md, section 5). Deadline comparisons need a common time basis or an explicit clock-offset-safe admission protocol.

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

One more consequence of the same comparison, for skew smaller than the admission window.

The FI is admitted, but removeExpired() expires the entry when this node's clock passes executionDeadline, which is the coordinator's startTime + timeOut. With the receiver ahead by δ, running FIs are cancelled δ early, so the effective query timeout becomes timeout − δ. On base the FI deadline is a relative duration applied to the local clock (DriverTask.computeDeadlineTimeInMs(instance.getTimeOut())), so it did not depend on the two clocks agreeing. A clock-injected probe confirms that the cut-off follows the receiver's clock only: the sweep leaves the FI RUNNING 1 ms before executionDeadline and cancels it at executionDeadline.

Two changes that would narrow this without a new protocol:

  • Take executionDeadline from the local relative timeout, as the legacy branch just above already does (now() + instance.getTimeOut()), and keep the coordinator's absolute timestamp for the admission decision only. That removes the early cancellation. The rejection described above remains.
  • When a request is rejected as expired, put the request's deadline and the local clock reading into the message. Today both the DataNode log and the client only show Query attempt … is closed or expired (720), which is hard to tell apart from a real timeout in the field.

}

private long now() {
return observedTime.accumulateAndGet(clock.getAsLong(), Math::max);

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

[P2] Do not apply the historical maximum clock to unrelated new queries

observedTime is shared by every attempt and never decreases. After a wall-clock rollback, even a coordinator and receiver in the same JVM generate fresh request deadlines from the adjusted clock while registration continues comparing against the old maximum. For example, after a 120-second rollback, a new query's default 60-second admission deadline is already below observedTime, so new queries are rejected for approximately a minute despite having a one-hour execution budget. This was reproduced by calling removeExpired(), moving the injected clock backwards, and registering a completely new queryId. Keep the protection against admitting old requests scoped to the relevant attempt so that it does not reject fresh query IDs after clock correction.

@JackieTien97 JackieTien97 left a comment

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Second pass on 0030ac2, complementing the review above. Two points are new; the others add to existing threads or are small.

New

  1. CI is red because of this change. unit-test (17, ubuntu-latest, datanode), unit-test (17, windows-latest, datanode) and codecov fail on two IoTDBInternalLocalReporterTest cases; the unit-test jobs are green on the base commit 509c395. Inline on ClusterScheduler.start(), with a one-line suggestion.
  2. The expiry sweep ends FIs with cancel() and no cause, so a query that times out can surface as a generic failure instead of 720. Inline on QueryAttemptRegistry.removeExpired().

Added to existing threads

  • Stop during dispatch: a candidate fix checked with a probe and the lifecycle tests, plus why the stale-query sweep reaches a query whose RPC is still running.
  • Receiver wall clock: the same comparison also shortens the timeout of FIs that were admitted, when the skew is below the admission window.

Minor

  • Two inline nits with suggestions: the comment in stop(), and offer vs add.
  • The 7 new files have no blank line between the license header and package. SonarCloud reports it (checkstyle EmptyLineSeparator) on QueryAttemptId.java and QueryAttemptRegistry.java.

Checked, no problem found

  • Attempt metadata: the FI binary body is unchanged, and every consensus read path (Ratis, IoTConsensus, IoTConsensusV2, Simple) hands the same FragmentInstance object to the state machine, so the metadata survives local and consensus reads.
  • QueryAttemptRegistry bookkeeping: every Entry access is inside compute/computeIfPresent, and terminate() runs outside it. Every exit of execDataQueryFragmentInstance and execSchemaQueryFragmentInstance either releases the registration or hands it to the execution listener.
  • remaining matches the per-host dataNodeFINum that both planners compute, and READ FIs are not moved to another host at dispatch.
  • LoadTsFileScheduler: the started CAS gives teardown() exactly one owner.
  • Earlier fixes in this area still hold: #19085 (immediate termination after a failed dispatch), #17794 (repeated dispatch rejected before planning), #18439 (failure info captured before cleanup), #12762 (release when no FragmentInstanceExecution is created).

Not verified

  • The two new points end to end on a cluster; the probes are unit-level.
  • Rolling upgrade with old and new nodes mixed.

@Override
public void start() {
synchronized (this) {
if (started || stopped || stateMachine.getState().isDone()) {

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This clause is what turns CI red. unit-test (17, ubuntu-latest, datanode), unit-test (17, windows-latest, datanode) and codecov all fail on the same two cases, and the unit-test jobs are green on the base commit 509c395:

  • IoTDBInternalLocalReporterTest.testRemoteInterruptionDoesNotReplaceExistingFailure
  • IoTDBInternalLocalReporterTest.testRemoteInterruptionDoesNotHideExistingException

Both move the query state machine to FAILED, then call start() on a real ClusterScheduler and wait for the write batch to reach sendBatchPlanNode (dispatched.await(5, SECONDS) at IoTDBInternalLocalReporterTest.java:489). With this clause start() returns before dispatching, so the latch times out.

Local runs on this commit:

  • as is: IoTDBInternalLocalReporterTest runs 25 cases, 2 fail
  • with only this clause removed: IoTDBInternalLocalReporterTest 25/25 and ClusterSchedulerLifecycleTest 4/4. These are the only two test classes that construct a ClusterScheduler directly.

Two ways out:

  1. Drop the clause (suggestion below). QueryExecution.schedule() already turns "terminal before the handoff" into stop-before-start. The clause only adds the short window in which the state is terminal but the listener's stop() has not reached the scheduler yet. In that window a read dispatch is aborted and cancelled by the stop() that follows (!dispatchComplete selects terminateImmediately), so cleanup is still complete. A write batch would still be sent, but I found no production path that makes a write terminal before schedule(): only queries are registered in queryExecutionMap, which is what KILL QUERY and the stale-query sweep work from.
  2. Keep the clause, since LoadTsFileScheduler.start() has the same check, and change the two cases to record the failure after dispatch has started, for example from the sendBatchPlanNode stub. I have not tried this one.
Suggested change
if (started || stopped || stateMachine.getState().isDone()) {
if (started || stopped) {

return retireIfUnused(key, entry, currentTime);
});
}
terminate(toCancel, true);

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Expiry ends the FIs with cancel() and no cause, so a timeout may not surface as 720.

executionDeadline is the coordinator's startTime + timeOut. The FI's own deadline is set by DriverTask as arrival time plus the remaining timeout, which is later by the time between planning on the coordinator and submission on this node. This sweep runs every 2 s, so it can fire inside that gap, before DriverTaskTimeoutSentinelThread does. With synchronized clocks the gap is small and this is rare. Once this node's clock is ahead of the coordinator's by more than the sweep interval, it happens for every query that times out (see the wall-clock thread on register).

What the FI reports in the two cases, from a unit probe on FragmentInstanceContext.getInstanceInfo():

path state message error code
this sweep CANCELLED empty none
existing driver timeout, failed(new QueryTimeoutException()) FAILED Query execution is time out 720

CANCELLED counts as failed. For a failed FI with neither an error code nor failure info, FixedRateFragInsStateTracker.updateQueryState builds a plain RuntimeException("FragmentInstance[…] is failed. …"), so the client would get a generic failure instead of QUERY_TIMEOUT. That last step is read from the code, not run end to end.

Suggestion: when a non-closed entry expires on executionDeadline, fail its FIs with a QueryTimeoutException instead of cancelling them. Closed entries whose fence expires should stay as they are: their FIs were already terminated by close(), and failed() would add a timeout cause to them.

Comment on lines 208 to 210
if (terminateImmediately) {
// The FragmentInstances dispatched before the failure can't finish on their own and hold
// operator memory until they are cancelled, so don't wait for the grace period.

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Nit: this comment still describes only the failed-dispatch case. terminateImmediately is now also true when stop() arrives before start() or while the dispatch is in flight. The Javadoc of IQueryTerminator.terminateImmediately ("Used when the dispatch failed: …") has the same gap.

Suggested change
if (terminateImmediately) {
// The FragmentInstances dispatched before the failure can't finish on their own and hold
// operator memory until they are cancelled, so don't wait for the grace period.
if (terminateImmediately) {
// Either the dispatch failed or the stop arrived before the dispatch completed. The
// FragmentInstances dispatched so far can't finish on their own and hold operator memory
// until they are cancelled, so don't wait for the grace period.

aborted = true;
LinkedBlockingQueue<Pair<SubPlan, Boolean>> queue = dispatchQueue;
if (queue != null) {
queue.offer(new Pair<>(null, false));

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Nit: the queue is unbounded now, so offer cannot fail here, but SonarCloud reports the ignored return value (java:S899) as one of the new reliability items behind the failed quality gate. add matches the other call sites in this class.

Suggested change
queue.offer(new Pair<>(null, false));
queue.add(new Pair<>(null, false));

Comment on lines +152 to +153
if (stopped) {
return;

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Three additions from a second pass.

This is new with the change. On base, a stop that arrived during dispatch was lost because the scheduler had not been published yet, so start() went on to RUNNING and getStatus() returned. Delivering that stop is the point of this PR; what is missing is completing the query state afterwards.

Why the stale-query sweep reaches a query that is still dispatching. cleanUpStaleQueries() skips executions for which isActive() is true, and isActive() returns startTimeOfCurrentRpc == -1, which is true when no RPC is in flight. So the sweep leaves idle executions alone and acts on the ones whose RPC is still running, including the initial executeStatement call that is inside start(). That looks inverted relative to the Javadoc of cleanUpStaleQueries(). It dates from #17358 and is outside this PR, so it probably deserves its own issue. The trigger here still needs dispatch to outlast timeout + 60 s.

A candidate fix, checked. Remember the cause in stop(). In this branch, if the state is not terminal yet, call transitionToFailed(cause), or transitionToAborted() when the cause is null, before returning. The probe uses a real QueryExecution and a real ClusterScheduler with a mocked dispatcher, and calls stopAndCleanup(cause) while dispatch() is blocked:

state after start() returns getStatus()
control, no stop RUNNING returns 200
this commit DISPATCHING still blocked after 5 s
fix, stop with the sweep's QueryTimeoutRuntimeException FAILED returns 720
fix, stop with a null cause ABORTED returns 301

ClusterSchedulerLifecycleTest, QueryExecutionLifecycleTest and QueryExecutionRetryTest still pass with it. Retries are not affected, because retry() only stops a scheduler whose start() has already returned.

The variant I ran had no explicit isDone() check. Add one: when the stop comes from the terminal-state listener, transitionToFailed(Throwable) would otherwise fill an empty failureException on a query that failed with a status only.

Comment on lines +90 to +94
long currentTime = now();
if (admissionDeadline <= currentTime
|| (existing != null
&& (existing.closed || existing.executionDeadline <= currentTime))) {
throw rejected(key, admissionDeadline <= currentTime);

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

One more consequence of the same comparison, for skew smaller than the admission window.

The FI is admitted, but removeExpired() expires the entry when this node's clock passes executionDeadline, which is the coordinator's startTime + timeOut. With the receiver ahead by δ, running FIs are cancelled δ early, so the effective query timeout becomes timeout − δ. On base the FI deadline is a relative duration applied to the local clock (DriverTask.computeDeadlineTimeInMs(instance.getTimeOut())), so it did not depend on the two clocks agreeing. A clock-injected probe confirms that the cut-off follows the receiver's clock only: the sweep leaves the FI RUNNING 1 ms before executionDeadline and cancels it at executionDeadline.

Two changes that would narrow this without a new protocol:

  • Take executionDeadline from the local relative timeout, as the legacy branch just above already does (now() + instance.getTimeOut()), and keep the coordinator's absolute timestamp for the admission decision only. That removes the early cancellation. The rejection described above remains.
  • When a request is rejected as expired, put the request's deadline and the local clock reading into the message. Today both the DataNode log and the client only show Query attempt … is closed or expired (720), which is hard to tell apart from a real timeout in the field.

Wake query status waiters on terminal cleanup while retaining retry-only cleanup. Replace coordinator wall-clock deadlines with receiver-issued admission tokens and local monotonic execution timeouts. Preserve reconnect retries, timeout status 720, and legacy peer fallback.

Cover the review regressions, including the existing reporter CI cases; 126 targeted tests and both full-reactor locale compiles passed.
Retain receiver admission metadata through scoped FI deserialization. Release transferred UDF scopes when cancellation prevents execution creation, and keep retry cleanup from finalizing the coordinator scope.

Validated the merged tree with 135 targeted tests across DataNode and UDF common modules, plus full-reactor test compilation for both locales.
@codecov

codecov Bot commented Nov 12, 2026 •

Copy link
Copy Markdown

Codecov Report

❌ Patch coverage is 92.52747% with 34 lines in your changes missing coverage. Please review.
✅ Project coverage is 49.13%. Comparing base (54bff3d) to head (2608feb).
⚠️ Report is 11 commits behind head on master.

Files with missing lines Patch % Lines
...plan/scheduler/FragmentInstanceDispatcherImpl.java 88.75% 9 Missing ⚠️
...ne/execution/fragment/FragmentInstanceManager.java 87.50% 7 Missing ⚠️
...ngine/execution/fragment/QueryAttemptRegistry.java 95.93% 7 Missing ⚠️
...ryengine/plan/scheduler/SimpleQueryTerminator.java 78.57% 3 Missing ⚠️
...ngine/plan/scheduler/load/LoadTsFileScheduler.java 70.00% 3 Missing ⚠️
...he/iotdb/db/queryengine/common/QueryAttemptId.java 84.61% 2 Missing ⚠️
...b/queryengine/plan/scheduler/ClusterScheduler.java 93.93% 2 Missing ⚠️
.../db/queryengine/plan/execution/QueryExecution.java 97.29% 1 Missing ⚠️
Additional details and impacted files
@@             Coverage Diff              @@
##             master   #19224      +/-   ##
============================================
+ Coverage     48.94%   49.13%   +0.19%     
  Complexity      689      689              
============================================
  Files          5589     5592       +3     
  Lines        413857   414360     +503     
  Branches      54594    54697     +103     
============================================
+ Hits         202544   203610    +1066     
+ Misses       211313   210750     -563     

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

🚀 New features to boost your workflow:
  • ❄️ Test Analytics: Detect flaky tests, report on failures, and find test suite problems.

@JackieTien97 JackieTien97 left a comment

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Additional findings from background review

Two additional observations from the git history analysis and timeout path investigation:

1. Tombstone retention overhead at high QPS (low-medium confidence)

Location: QueryAttemptRegistry.java:178-190 (retireIfUnused), :143-161 (removeExpired)

Each closed attempt retains a tombstone entry on every DataNode it touched until fenceDeadline (default 60s). The removeExpired sweep runs every 2 seconds on the single FRAGMENT_INSTANCE_MANAGEMENT thread.

Impact estimation (10K QPS/DataNode, 1C3D topology, avg 2 DataNodes per query):

  • Steady-state tombstones: ~1.2M entries
  • Memory overhead: ~240 MB per DataNode
  • Sweep time: ~1.2 seconds per cycle (60% of the 2s period), potentially blocking other management tasks

Historical context: Prior to this PR, per-query state was freed immediately when the last FI finished (5a50f0df22, 3d436b9380). The new mechanism trades memory/CPU for stronger admission fencing.

Question: Is the 60s fence window acceptable for production workloads at this scale? Should the sweep use a more efficient structure (e.g., priority queue by deadline) or a dedicated thread?


2. Inconsistent timeout error reporting (medium confidence)

Location: QueryAttemptRegistry.java:101, :143-160

When removeExpired beats an FI's own timeout deadline, it calls terminate(toCancel, true) which triggers stateMachine.cancel(). The resulting error is a generic RuntimeException("FragmentInstance[..] is failed ..") without error code 720.

Race window:

  • Registry deadline: startTime + timeOut
  • FI deadline: arrival + (timeOut - (planTime - startTime))
  • Gap: dispatch delay (grows with fan-out)

Current behavior:

  • FI times out naturally → QueryTimeoutException(720)
  • Registry sweep fires first → generic failure, no 720

Suggestion: Pass an explicit QueryTimeoutException(720) as the cause when calling terminate from removeExpired:

// At QueryAttemptRegistry.java:~155
QueryTimeoutException timeoutCause = new QueryTimeoutException(
    entry.getStartTime(), 
    nowNanos, 
    entry.getExecutionDeadline() - entry.getStartTime()
);
terminate(toCancel, timeoutCause);

This ensures consistent error codes regardless of which timeout path fires first, improving observability and troubleshooting.


Priority: F2 (timeout consistency) affects observability; F1 (tombstone overhead) only significant at very high QPS (>5K/DN). Both are follow-up improvements rather than blockers for this PR.

Separate compact closed markers from live attempt state and expire them by deadline on a dedicated executor. Cap admission validity by the remaining query budget, keep extended fences safe from stale expiry events, and avoid duplicate queue entries.

Use an explicit legacy-admission signal for connection fallback and cover serialized RPC admission and cancellation. Validated 142 targeted tests and full-reactor compilation in both locales; measured retained heap and expiry costs with a 1.2-million-record synthetic workload.
Schedule query attempt expiration on instanceManagementExecutor and remove the unused dedicated executor and thread name.

Validated 142 targeted tests and full-reactor test compilation for both locales.
@sonarqubecloud

Copy link
Copy Markdown

Quality Gate Failed Quality Gate failed

Failed conditions
C Reliability Rating on New Code (required ≥ A)

See analysis details on SonarQube Cloud

Catch issues before they fail your Quality Gate with our IDE extension SonarQube for IDE

@JackieTien97
JackieTien97 merged commit 95d05dd into master Nov 13, 2026
41 of 42 checks passed
@JackieTien97
JackieTien97 deleted the ty/query-attempt-cleanup branch November 13, 2026 08:09
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Labels

None yet

Projects

None yet

Development

Successfully merging this pull request may close these issues.

1 participant