xiangfu0 commented on code in PR #19854:
URL: https://github.com/apache/pinot/pull/19854#discussion_r4278285625
##########
pinot-query-runtime/src/main/java/org/apache/pinot/query/runtime/executor/OpChainSchedulerService.java:
##########
@@ -335,6 +342,55 @@ private ReadWriteLock getQueryLock(long requestId) {
return _queryLocks[(int) (requestId & QUERY_LOCK_MASK)];
}
+ /// Runs an op chain. It is registered with the [QueryExecutionContext] of
the query, so that terminating the query
+ /// interrupts the op chain.
+ ///
+ /// [QueryExecutionContext#terminate] cancels the registered tasks. A
`Future` from [ExecutorService#submit] that is
+ /// cancelled before its task starts never runs the task. An op chain
dropped this way never sends an EOS or an error
+ /// to its receivers, so they wait for data until the query deadline.
Cancelling this task only interrupts the op
+ /// chain while it runs. An op chain that has not started yet still runs,
and fails on its first check: the context
+ /// sets the termination error before it cancels the tasks.
+ ///
+ /// `_lock` makes sure that an interrupt reaches the thread only while it
runs the op chain, and never the next task
+ /// of that thread.
+ private static final class OpChainTask extends AbstractFuture<Void>
implements Runnable {
+ private final Runnable _opChainRunner;
+ private final Object _lock = new Object();
+ @GuardedBy("_lock")
+ @Nullable
+ private Thread _thread;
+
+ OpChainTask(Runnable opChainRunner) {
+ _opChainRunner = opChainRunner;
+ }
+
+ @Override
+ public void run() {
+ synchronized (_lock) {
+ _thread = Thread.currentThread();
+ }
+ try {
+ _opChainRunner.run();
+ } finally {
+ synchronized (_lock) {
+ _thread = null;
+ }
+ // Clear an interrupt that came after the op chain stopped checking
for it
+ Thread.interrupted();
Review Comment:
Subtle but correct: clearing the interrupt *after* nulling `_thread` (and
before `set(null)`) is what prevents a late `terminate()` interrupt from
leaking into the next task on a pooled thread. Worth the inline note you
already added. Nit: a one-liner cross-referencing `interruptTask()` here would
make the happens-before argument easier to follow.
##########
pinot-query-runtime/src/main/java/org/apache/pinot/query/runtime/plan/OpChainConverter.java:
##########
@@ -71,7 +71,7 @@ default OpChain sendEarlyError(OpChainExecutionContext
context, StagePlan stageP
}
MailboxSendNode rootNode = (MailboxSendNode) stagePlan.getRootNode();
String message = extractErrorMessage(errorBlock);
- MultiStageOperator errorOp = new ErrorOperator(context,
QueryErrorCode.QUERY_EXECUTION, message);
+ MultiStageOperator errorOp = new ErrorOperator(context,
errorBlock.getMainErrorCode(), message);
Review Comment:
Good catch preserving the original error code instead of flattening to
`QUERY_EXECUTION` — this is what lets a forwarded kill keep its
`SERVER_RESOURCE_LIMIT_EXCEEDED` code end to end. The updated
`OpChainConverterDispatcherTest` covers it well.
##########
pinot-query-runtime/src/main/java/org/apache/pinot/query/runtime/plan/pipeline/PipelineBreakerOperator.java:
##########
@@ -101,6 +101,16 @@ public ErrorMseBlock getErrorBlock() {
return _errorBlock;
}
+ /// Keeps the cancellation reason as the error, so that a pipeline breaker
cancelled before it collected its results
+ /// (e.g. the query was terminated before it started) does not look like it
collected no rows.
+ @Override
+ public void cancel(Throwable e) {
+ if (_errorBlock == null) {
+ _errorBlock = ErrorMseBlock.fromException(e instanceof Exception ?
(Exception) e : new RuntimeException(e));
Review Comment:
Nice fix — keeping the cancel reason as `_errorBlock` is what stops a
not-yet-run pipeline breaker from looking like it collected zero rows. One nit:
Codecov flags this ternary's non-`Exception` branch (`new RuntimeException(e)`)
as uncovered. In practice the cause here is always a `TerminationException` (an
`Exception`), so that branch is effectively dead — could you confirm whether
any caller can pass a raw `Throwable`? If not, a short comment noting the
branch is just defensive would help future readers.
##########
pinot-query-runtime/src/main/java/org/apache/pinot/query/runtime/plan/pipeline/PipelineBreakerOperator.java:
##########
@@ -101,6 +101,16 @@ public ErrorMseBlock getErrorBlock() {
return _errorBlock;
}
+ /// Keeps the cancellation reason as the error, so that a pipeline breaker
cancelled before it collected its results
+ /// (e.g. the query was terminated before it started) does not look like it
collected no rows.
+ @Override
+ public void cancel(Throwable e) {
+ if (_errorBlock == null) {
+ _errorBlock = ErrorMseBlock.fromException(e instanceof Exception ?
(Exception) e : new RuntimeException(e));
Review Comment:
Nice fix — keeping the cancel reason as `_errorBlock` is what stops a
not-yet-run pipeline breaker from looking like it collected zero rows. One nit:
Codecov flags this ternary's non-`Exception` branch (`new RuntimeException(e)`)
as uncovered. In practice the cause here is always a `TerminationException` (an
`Exception`), so that branch is effectively dead — could you confirm whether
any caller can pass a raw `Throwable`? If not, a short comment noting the
branch is just defensive would help future readers.
##########
pinot-query-runtime/src/main/java/org/apache/pinot/query/runtime/executor/OpChainSchedulerService.java:
##########
@@ -335,6 +342,55 @@ private ReadWriteLock getQueryLock(long requestId) {
return _queryLocks[(int) (requestId & QUERY_LOCK_MASK)];
}
+ /// Runs an op chain. It is registered with the [QueryExecutionContext] of
the query, so that terminating the query
+ /// interrupts the op chain.
+ ///
+ /// [QueryExecutionContext#terminate] cancels the registered tasks. A
`Future` from [ExecutorService#submit] that is
+ /// cancelled before its task starts never runs the task. An op chain
dropped this way never sends an EOS or an error
+ /// to its receivers, so they wait for data until the query deadline.
Cancelling this task only interrupts the op
+ /// chain while it runs. An op chain that has not started yet still runs,
and fails on its first check: the context
+ /// sets the termination error before it cancels the tasks.
+ ///
+ /// `_lock` makes sure that an interrupt reaches the thread only while it
runs the op chain, and never the next task
+ /// of that thread.
+ private static final class OpChainTask extends AbstractFuture<Void>
implements Runnable {
+ private final Runnable _opChainRunner;
+ private final Object _lock = new Object();
+ @GuardedBy("_lock")
+ @Nullable
+ private Thread _thread;
+
+ OpChainTask(Runnable opChainRunner) {
+ _opChainRunner = opChainRunner;
+ }
+
+ @Override
+ public void run() {
+ synchronized (_lock) {
+ _thread = Thread.currentThread();
+ }
+ try {
+ _opChainRunner.run();
+ } finally {
+ synchronized (_lock) {
+ _thread = null;
+ }
+ // Clear an interrupt that came after the op chain stopped checking
for it
+ Thread.interrupted();
Review Comment:
Subtle but correct: clearing the interrupt *after* nulling `_thread` (and
before `set(null)`) is what prevents a late `terminate()` interrupt from
leaking into the next task on a pooled thread. Worth the inline note you
already added. Nit: a one-liner cross-referencing `interruptTask()` here would
make the happens-before argument easier to follow.
##########
pinot-query-runtime/src/main/java/org/apache/pinot/query/runtime/plan/OpChainConverter.java:
##########
@@ -71,7 +71,7 @@ default OpChain sendEarlyError(OpChainExecutionContext
context, StagePlan stageP
}
MailboxSendNode rootNode = (MailboxSendNode) stagePlan.getRootNode();
String message = extractErrorMessage(errorBlock);
- MultiStageOperator errorOp = new ErrorOperator(context,
QueryErrorCode.QUERY_EXECUTION, message);
+ MultiStageOperator errorOp = new ErrorOperator(context,
errorBlock.getMainErrorCode(), message);
Review Comment:
Good catch preserving the original error code instead of flattening to
`QUERY_EXECUTION` — this is what lets a forwarded kill keep its
`SERVER_RESOURCE_LIMIT_EXCEEDED` code end to end. The updated
`OpChainConverterDispatcherTest` covers it well.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
