Repository navigation
Fail op chains of terminated MSE queries instead of dropping them - #19854
Conversation
Codecov Report✅ All modified and coverable lines are covered by tests. Additional details and impacted files@@ Coverage Diff @@
## master #19854 +/- ##
============================================
+ Coverage 69.26% 69.39% +0.12%
Complexity 1491 1491
============================================
Files 3543 3544 +1
Lines 233237 233458 +221
Branches 37277 37316 +39
============================================
+ Hits 161560 162007 +447
+ Misses 59184 58942 -242
- Partials 12493 12509 +16
Flags with carried forward coverage won't be shown. Click here to find out more. ☔ View full report in Codecov by Harness. 🚀 New features to boost your workflow:
|
xiangfu0
left a comment
There was a problem hiding this comment.
Reviewed the full diff against the op-chain termination/lifecycle flow. This is a clean, well-reasoned fix with excellent docs and tests, and I couldn't find any blocking issues.
What I checked:
- The
register()terminated path now cancels + closes the op chain on the calling thread and fires the completion listener. It doesn't go throughregisterInternal, so it never touches_activeOpChainsByRequest/_executionContextByRequest— self-consistent, no counter leak, and the listener still fires exactly once per op chain. execute()vssubmit(): confirmedDecoratorExecutorService.execute()still decorates (sets upQueryThreadContext) but skipsonSubmit/addTask, so the op chain is no longer auto-cancelled-and-dropped. The manualaddTask(opChainTask)afterexecute()restores interrupt-based cancellation without the drop.OpChainTaskinterrupt handling: the "null the thread →Thread.interrupted()→set(null)" ordering correctly closes the interrupt-leak window, andinterruptTask()is a no-op once_threadis cleared. The dedicated test (shouldNotLeakInterruptAfterOpChainCompletes) exercises exactly this.- Rejected-execution path still decrements the active counter once and rethrows; no double-decrement with the FutureCallback.
sendEarlyErrorandPipelineBreakerOperator.cancelnow preserve the real error code (TerminationException extends QueryException, sofromExceptionkeepsSERVER_RESOURCE_LIMIT_EXCEEDED).
A couple of minor/nit comments inline — none blocking.
Severity counts: CRITICAL: 0, MAJOR: 0, MINOR: 2.
xiangfu0
left a comment
There was a problem hiding this comment.
Reviewed the full diff against the op-chain termination/lifecycle flow. This is a clean, well-reasoned fix with excellent docs and tests, and I couldn't find any blocking issues.
What I checked:
- The
register()terminated path now cancels + closes the op chain on the calling thread and fires the completion listener. It doesn't go throughregisterInternal, so it never touches_activeOpChainsByRequest/_executionContextByRequest— self-consistent, no counter leak, and the listener still fires exactly once per op chain. execute()vssubmit(): confirmedDecoratorExecutorService.execute()still decorates (sets upQueryThreadContext) but skipsonSubmit/addTask, so the op chain is no longer auto-cancelled-and-dropped. The manualaddTask(opChainTask)afterexecute()restores interrupt-based cancellation without the drop.OpChainTaskinterrupt handling: the "null the thread →Thread.interrupted()→set(null)" ordering correctly closes the interrupt-leak window, andinterruptTask()is a no-op once_threadis cleared. The dedicated test (shouldNotLeakInterruptAfterOpChainCompletes) exercises exactly this.- Rejected-execution path still decrements the active counter once and rethrows; no double-decrement with the FutureCallback.
sendEarlyErrorandPipelineBreakerOperator.cancelnow preserve the real error code (TerminationException extends QueryException, sofromExceptionkeepsSERVER_RESOURCE_LIMIT_EXCEEDED).
A couple of minor/nit comments inline — none blocking.
Severity counts: CRITICAL: 0, MAJOR: 0, MINOR: 2.
xiangfu0
left a comment
There was a problem hiding this comment.
Re-reviewed on the new head (7ef306d) after the push. The production logic is unchanged from my previous approval, and the new revision addresses both of my earlier nits:
PipelineBreakerOperator.cancelnow documents why the cancel reason may be a non-Exception(Error/OutOfMemoryError), clarifying theinstanceof Exceptionbranch.- The
OpChainTask.runinterrupt-clear comment now explains the happens-before withinterruptTask().
The new PipelineBreakerExecutorTest cases close the coverage gap Codecov flagged:
shouldReturnTerminationErrorWhenQueryIsTerminatedBeforePBStarts— PB cancelled before it runs keeps the termination error instead of looking like zero rows.shouldReturnErrorBlockWhenPBFailsWithError— theOutOfMemoryError(non-Exception) cancel path is now exercised.
I verified the new tests reference real APIs (PipelineBreakerResult.getErrorBlock(), executePipelineBreakers, the _mailbox1/MAILBOX_ID_1 fixtures) and follow the existing test patterns. No CRITICAL/MAJOR issues; re-approving.
Severity counts: CRITICAL: 0, MAJOR: 0, MINOR: 0 (prior nits resolved).
xiangfu0
left a comment
There was a problem hiding this comment.
Re-reviewed on the new head (34a71a3) after the push. The production logic is unchanged from my prior approvals; the only substantive change is a necessary test alignment plus a rebase onto newer master:
HardLimitExecutorOpChainRegressionTestno longer asserts the queued task is aFutureTaskand no longer unwraps it viaFuture.get(). Since the scheduler now hands op chains to the executor withexecute()(notsubmit()), the queued task is theOpChainTaskrunnable directly. Running it on the test thread means an executor exception propagates straight to the test — a stronger check than the old swallowed-then-rethrown path. The droppedjava.util.concurrent.Futureimport is consistent.QueryThreadContextshows only a shifted hunk context header (rebase onto master that addedmergeRemoteAccessStatsupstream); the actualcontextAwareExecutorServiceJavadoc change is identical to what I reviewed.
No CRITICAL/MAJOR issues; re-approving.
Severity counts: CRITICAL: 0, MAJOR: 0, MINOR: 0.
PR flow
Terminated MSE queries now propagate errors to receivers instead of being dropped, avoiding timeout reports.
AI-generated · Green: added · Yellow: modified · Red: removed · Gray: existing
Diff evidence
Problem
When a server terminates a multi-stage query, a stage on that server that has not started yet sends nothing to its receivers. For example, the resource accountant terminates a query to protect the heap. The broker then waits until the query timeout and reports a timeout instead of the kill.
250"Timed out on stage 0 waiting for data from child stage 1", after the full timeout245with the kill message, in less than 1 s250, after the full timeout245with the kill message, in less than 1 sTwo paths drop the op chain:
OpChainSchedulerServicescheduled the op chain withExecutorService.submit(). The context-aware executor registers the returnedFuturewith theQueryExecutionContext, andterminate()cancels it. AFutureTaskcancelled before it starts never runs its task, so theFutureCallbackthat cancels and closes the op chain never runs. The server logs nothing.register()rejected the op chain of a terminated query.QueryRunnerthen closed the op chain, butGrpcSendingMailbox.close()sends nothing when no stream is open yet. The error op chain fromOpChainConverter.sendEarlyErroralso went throughregister(), and the same check rejected it ("Failed to propagate stage error via OpChainConverter").On the same path, a pipeline breaker cancelled before it ran looked like it collected no rows. Also,
sendEarlyErrorreplaced the error code withQUERY_EXECUTION, so a forwarded kill lost itsSERVER_RESOURCE_LIMIT_EXCEEDEDcode.Change
register()no longer rejects the op chain of a terminated query, including a query cancelled through the cancelled-query cache. It cancels the op chain with the termination error on the calling thread, without the executor. This sends the error code and message to the receivers.OpChainTaskwithexecute()and registers it with the query context. Cancelling it interrupts the op chain only while it runs. A queued op chain still runs, fails before it does any work, and the existing failure callback sends the termination error.PipelineBreakerOperator.cancel()keeps the cancel reason as its error block.OpChainConverter.sendEarlyErrorkeeps the error code of the error block.QueryThreadContext.contextAwareExecutorServicenow says that onlysubmit()registers tasks for cancellation.There is no change to configuration, SPI or wire format.
Tests
QueryTerminatedBeforeStageStartsTest(new) runs two servers and a broker reducer. One server terminates the query:GatedExecutorServicePlugin, a test executor, holds the op chains until the test terminates the query.The broker must report
SERVER_RESOURCE_LIMIT_EXCEEDEDwith the kill message within 10 s. The query timeout is 30 s.OpChainSchedulerServiceTest: an op chain that waits in the executor queue when the query is terminated, op chains registered after a termination and after a cancel, and no interrupt that leaks to the next task of the thread.OpChainConverterDispatcherTest:sendEarlyErrorkeeps the error code.HardLimitExecutorOpChainRegressionTest(from Check HardLimitExecutor limit only once, on submit #19864) cast the queued task toFuture, which assumed that the scheduler usessubmit(). The test now runs the task directly and checks that the op chain ran and closed, so it holds withsubmit()and withexecute().All the new and changed tests fail on master. The
pinot-query-runtimetests,QueryKillingIntegrationTest,WindowResourceAccountingTestandMultiStageEngineIntegrationTestpass.Not in this PR: a task can still time out in the heap throttling queue and never run. (#19864 removed the run-time limit check of
HardLimitExecutor.)