yashmayya commented on code in PR #19854:
URL: https://github.com/apache/pinot/pull/19854#discussion_r4286015551
##########
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:
Thanks! The branch is reachable. `cancel()` also runs from the scheduler's
`FutureCallback.onFailure(Throwable)`, which gets `Error`s too, for example an
`OutOfMemoryError` while the pipeline breaker collects its results. I added a
comment, and two tests in `PipelineBreakerExecutorTest`:
- `shouldReturnErrorBlockWhenPBFailsWithError`: a mailbox read throws an
`OutOfMemoryError` (this branch).
- `shouldReturnTerminationErrorWhenQueryIsTerminatedBeforePBStarts`: the
query is terminated before the pipeline breaker starts.
Both fail without the `cancel()` override.
##########
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:
Done. The comment now says that `_thread` is null at this point, so
`interruptTask()` can no longer interrupt this thread, and that this call
clears an interrupt it sent earlier.
--
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]
