This is an automated email from the ASF dual-hosted git repository.
gortiz pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/pinot.git
The following commit(s) were added to refs/heads/master by this push:
new 91416079849 Close op chain when scheduling fails to release operator
resources (#18928)
91416079849 is described below
commit 91416079849599668de91c200eb16aebc1288f46
Author: Yash Mayya <[email protected]>
AuthorDate: Tue Jul 7 01:33:30 2026 -0700
Close op chain when scheduling fails to release operator resources (#18928)
---
.../java/org/apache/pinot/query/runtime/QueryRunner.java | 14 +++++++++++++-
.../query/runtime/executor/OpChainSchedulerService.java | 13 +++++++++----
2 files changed, 22 insertions(+), 5 deletions(-)
diff --git
a/pinot-query-runtime/src/main/java/org/apache/pinot/query/runtime/QueryRunner.java
b/pinot-query-runtime/src/main/java/org/apache/pinot/query/runtime/QueryRunner.java
index afc05d45187..910f53480e4 100644
---
a/pinot-query-runtime/src/main/java/org/apache/pinot/query/runtime/QueryRunner.java
+++
b/pinot-query-runtime/src/main/java/org/apache/pinot/query/runtime/QueryRunner.java
@@ -324,8 +324,8 @@ public class QueryRunner {
OpChainExecutionContext executionContext =
OpChainExecutionContext.fromQueryContext(_mailboxService,
opChainMetadata, stageMetadata, workerMetadata,
pipelineBreakerResult, sendStats,
_keepPipelineBreakerStats.getAsBoolean());
+ OpChain opChain = null;
try {
- OpChain opChain;
if (workerMetadata.isLeafStageWorker()) {
Map<String, String> rlsFilters =
RlsUtils.extractRlsFilters(requestMetadata);
opChain =
@@ -337,6 +337,18 @@ public class QueryRunner {
// This can fail if the executor rejects the task.
_opChainScheduler.register(opChain);
} catch (RuntimeException e) {
+ // register() can fail after the op chain was built — the query was
cancelled/terminated before scheduling
+ // (checkTermination / cancelled-query cache), or the executor rejected
the task (e.g. under load shedding). In
+ // those cases the op chain is never scheduled, so the scheduler's
FutureCallback, which normally close()s the
+ // chain on completion, never fires — leaking any operator resources
that are released only in close(). Close the
+ // built chain here (idempotent; guarded so a close failure cannot mask
the original error).
+ if (opChain != null) {
+ try {
+ opChain.close();
+ } catch (RuntimeException closeError) {
+ LOGGER.warn("Failed to close op chain after scheduling failure",
closeError);
+ }
+ }
ErrorMseBlock errorBlock = ErrorMseBlock.fromException(e);
tryPropagateErrorViaOpChainConverter(workerMetadata, stagePlan,
opChainMetadata, pipelineBreakerResult,
errorBlock);
diff --git
a/pinot-query-runtime/src/main/java/org/apache/pinot/query/runtime/executor/OpChainSchedulerService.java
b/pinot-query-runtime/src/main/java/org/apache/pinot/query/runtime/executor/OpChainSchedulerService.java
index e4472a41629..75de3ad8eda 100644
---
a/pinot-query-runtime/src/main/java/org/apache/pinot/query/runtime/executor/OpChainSchedulerService.java
+++
b/pinot-query-runtime/src/main/java/org/apache/pinot/query/runtime/executor/OpChainSchedulerService.java
@@ -254,10 +254,15 @@ public class OpChainSchedulerService {
_executorService.submit(listenableFutureTask);
} catch (RuntimeException e) {
// The MSE executor is wrapped in HardLimitExecutor + heap throttling,
so submit() can throw
- // RejectedExecutionException. When it does, the task never runs and the
directExecutor FutureCallback above
- // (which decrements the active-opchain counter and removes the
per-request context entry) never fires. Back
- // out that bookkeeping here so the entry — and the
QueryExecutionContext it pins — does not leak until a later
- // cancel. Then rethrow so the caller propagates the failure as a stage
error.
+ // RejectedExecutionException. When it does, the task never runs, so
neither the directExecutor FutureCallback
+ // above (which decrements the active-opchain counter and removes the
per-request context entry) nor runJob's
+ // _opChainCache.invalidate() ever fires. Back out both here: invalidate
the cache entry this method put()
+ // above so it does not pin the rootOperator tree and
QueryExecutionContext until TTL/weight eviction (the same
+ // prompt release cancel() performs for cancelled entries), and back out
the active-opchain bookkeeping so the
+ // per-request context entry does not leak until a later cancel. Then
rethrow so the caller propagates the
+ // failure as a stage error. The caller,
QueryRunner#processQueryBlocking, close()s the op chain on this
+ // rethrow, releasing operator resources that the never-fired
FutureCallback would otherwise have closed.
+ _opChainCache.invalidate(opChainId);
decrementActiveOpChains(requestId);
throw e;
}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
