You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pinot.apache.org by go...@apache.org on 2026/04/01 14:36:20 UTC
(pinot) branch master updated: Introduce error propagation via `OpChainConverter` to streamline stage failure handling. (#18026)
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 736ac51942c Introduce error propagation via `OpChainConverter` to streamline stage failure handling. (#18026)
736ac51942c is described below
commit 736ac51942c25f9f156eb949ac079f7eb5b323f8
Author: Gonzalo Ortiz Jaureguizar <go...@users.noreply.github.com>
AuthorDate: Wed Apr 1 16:36:13 2026 +0200
Introduce error propagation via `OpChainConverter` to streamline stage failure handling. (#18026)
---
.../apache/pinot/query/runtime/QueryRunner.java | 81 +++++++-------------
.../pinot/query/runtime/plan/OpChainConverter.java | 40 ++++++++++
.../runtime/plan/OpChainConverterDispatcher.java | 7 ++
.../plan/OpChainConverterDispatcherTest.java | 89 ++++++++++++++++++++++
4 files changed, 162 insertions(+), 55 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 6fe52583736..53398023be7 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
@@ -37,7 +37,6 @@ import javax.annotation.Nullable;
import org.apache.commons.lang3.StringUtils;
import org.apache.commons.lang3.tuple.Pair;
import org.apache.pinot.common.config.TlsConfig;
-import org.apache.pinot.common.datatable.StatMap;
import org.apache.pinot.common.metrics.ServerMeter;
import org.apache.pinot.common.metrics.ServerMetrics;
import org.apache.pinot.common.proto.Worker;
@@ -47,20 +46,14 @@ import org.apache.pinot.core.executor.ThrottleOnCriticalHeapUsageExecutor;
import org.apache.pinot.core.query.executor.QueryExecutor;
import org.apache.pinot.core.query.executor.ServerQueryExecutorV1Impl;
import org.apache.pinot.query.mailbox.MailboxService;
-import org.apache.pinot.query.mailbox.SendingMailbox;
-import org.apache.pinot.query.planner.physical.MailboxIdUtils;
import org.apache.pinot.query.planner.plannode.ExplainedNode;
-import org.apache.pinot.query.planner.plannode.MailboxSendNode;
import org.apache.pinot.query.planner.plannode.PlanNode;
-import org.apache.pinot.query.routing.MailboxInfo;
-import org.apache.pinot.query.routing.RoutingInfo;
import org.apache.pinot.query.routing.StageMetadata;
import org.apache.pinot.query.routing.StagePlan;
import org.apache.pinot.query.routing.WorkerMetadata;
import org.apache.pinot.query.runtime.blocks.ErrorMseBlock;
import org.apache.pinot.query.runtime.executor.OpChainSchedulerService;
import org.apache.pinot.query.runtime.operator.LeafOperator;
-import org.apache.pinot.query.runtime.operator.MailboxSendOperator;
import org.apache.pinot.query.runtime.operator.MultiStageOperator;
import org.apache.pinot.query.runtime.operator.OpChain;
import org.apache.pinot.query.runtime.plan.MultiStageQueryStats;
@@ -74,8 +67,6 @@ import org.apache.pinot.query.runtime.timeseries.TimeSeriesExecutionContext;
import org.apache.pinot.query.runtime.timeseries.serde.TimeSeriesBlockSerde;
import org.apache.pinot.spi.config.instance.InstanceType;
import org.apache.pinot.spi.env.PinotConfiguration;
-import org.apache.pinot.spi.exception.QueryErrorCode;
-import org.apache.pinot.spi.exception.QueryException;
import org.apache.pinot.spi.executor.ExecutorServiceUtils;
import org.apache.pinot.spi.executor.HardLimitExecutor;
import org.apache.pinot.spi.executor.MetricsExecutor;
@@ -298,7 +289,8 @@ public class QueryRunner {
/// Executes a {@link StagePlan} pseudo-synchronously.
///
/// First, the pipeline breaker is executed on the current thread. This is the blocking part of the method.
- /// If the pipeline breaker execution fails, the current thread will send the error block to the receivers mailboxes.
+ /// If the pipeline breaker execution fails, the query runner delegates downstream error propagation to the active
+ /// OpChainConverter.
///
/// If the pipeline breaker success, the rest of the stage is asynchronously executed on the [#_opChainScheduler].
private void processQueryBlocking(WorkerMetadata workerMetadata, StagePlan stagePlan,
@@ -314,7 +306,8 @@ public class QueryRunner {
// Send error block to all the receivers if pipeline breaker fails
if (pipelineBreakerResult != null && pipelineBreakerResult.getErrorBlock() != null) {
ErrorMseBlock errorBlock = pipelineBreakerResult.getErrorBlock();
- notifyErrorAfterSubmission(stageMetadata.getStageId(), errorBlock, workerMetadata, stagePlan);
+ tryPropagateErrorViaOpChainConverter(workerMetadata, stagePlan, opChainMetadata, pipelineBreakerResult,
+ errorBlock);
return;
}
@@ -336,53 +329,31 @@ public class QueryRunner {
_opChainScheduler.register(opChain);
} catch (RuntimeException e) {
ErrorMseBlock errorBlock = ErrorMseBlock.fromException(e);
- notifyErrorAfterSubmission(stageMetadata.getStageId(), errorBlock, workerMetadata, stagePlan);
+ tryPropagateErrorViaOpChainConverter(workerMetadata, stagePlan, opChainMetadata, pipelineBreakerResult,
+ errorBlock);
}
}
- private void notifyErrorAfterSubmission(int stageId, ErrorMseBlock errorBlock, WorkerMetadata workerMetadata,
- StagePlan stagePlan) {
- QueryExecutionContext executionContext = QueryThreadContext.get().getExecutionContext();
- long requestId = executionContext.getRequestId();
- LOGGER.error("Error executing pipeline breaker for request: {}, stage: {}, sending error block: {}", requestId,
- stageId, errorBlock);
- MailboxSendNode rootNode = (MailboxSendNode) stagePlan.getRootNode();
- List<RoutingInfo> routingInfos = new ArrayList<>();
- for (Integer receiverStageId : rootNode.getReceiverStageIds()) {
- List<MailboxInfo> receiverMailboxInfos =
- workerMetadata.getMailboxInfosMap().get(receiverStageId).getMailboxInfos();
- List<RoutingInfo> stageRoutingInfos =
- MailboxIdUtils.toRoutingInfos(requestId, stageId, workerMetadata.getWorkerId(), receiverStageId,
- receiverMailboxInfos);
- routingInfos.addAll(stageRoutingInfos);
- }
- long deadlineMs = executionContext.getPassiveDeadlineMs();
- for (RoutingInfo routingInfo : routingInfos) {
- String mailboxId = routingInfo.getMailboxId();
- StatMap<MailboxSendOperator.StatKey> statMap = new StatMap<>(MailboxSendOperator.StatKey.class);
- try (SendingMailbox sendingMailbox = _mailboxService.getSendingMailbox(routingInfo.getHostname(),
- routingInfo.getPort(), mailboxId, deadlineMs, statMap)) {
- // TODO: Here we are breaking the stats invariants, sending errors without including the stats of the
- // current stage. We will need to fix this in future, but for now, we are sending the error block without
- // the stats.
- sendingMailbox.send(errorBlock, Collections.emptyList());
- } catch (QueryException e) {
- QueryErrorCode errorCode = e.getErrorCode();
- switch (errorCode) {
- case EXECUTION_TIMEOUT:
- LOGGER.warn("Timed out sending error block to mailbox: {}", mailboxId, e);
- break;
- case QUERY_CANCELLATION:
- LOGGER.info("Query cancelled while offering blocks to mailbox: {}", mailboxId);
- break;
- default:
- LOGGER.error("{} exception while exception sending error block to mailbox: {}", errorCode, mailboxId, e);
- break;
- }
- } catch (Exception e) {
- LOGGER.error("Caught exception sending error block to mailbox: {} for request: {}, stage: {}",
- mailboxId, requestId, stageId, e);
- }
+ /**
+ * Attempts to propagate stage failures through the active {@link OpChainConverter}.
+ */
+ private void tryPropagateErrorViaOpChainConverter(WorkerMetadata workerMetadata, StagePlan stagePlan,
+ Map<String, String> opChainMetadata, @Nullable PipelineBreakerResult pipelineBreakerResult,
+ ErrorMseBlock errorBlock) {
+ StageMetadata stageMetadata = stagePlan.getStageMetadata();
+ QueryExecutionContext queryContext = QueryThreadContext.get().getExecutionContext();
+ long requestId = queryContext.getRequestId();
+ LOGGER.error("Error executing stage for request: {}, stage: {}, sending error block: {}", requestId,
+ stageMetadata.getStageId(), errorBlock);
+ try {
+ OpChainExecutionContext executionContext =
+ OpChainExecutionContext.fromQueryContext(_mailboxService, opChainMetadata, stageMetadata, workerMetadata,
+ pipelineBreakerResult, _sendStats.getAsBoolean(), _keepPipelineBreakerStats.getAsBoolean());
+ OpChain errorOpChain = OpChainConverterDispatcher.sendEarlyError(executionContext, stagePlan, errorBlock);
+ _opChainScheduler.register(errorOpChain);
+ } catch (RuntimeException e) {
+ LOGGER.warn("Failed to propagate stage error via OpChainConverter for request: {}, stage: {}",
+ requestId, stageMetadata.getStageId(), e);
}
}
diff --git a/pinot-query-runtime/src/main/java/org/apache/pinot/query/runtime/plan/OpChainConverter.java b/pinot-query-runtime/src/main/java/org/apache/pinot/query/runtime/plan/OpChainConverter.java
index 4014a841ad3..0777830ad31 100644
--- a/pinot-query-runtime/src/main/java/org/apache/pinot/query/runtime/plan/OpChainConverter.java
+++ b/pinot-query-runtime/src/main/java/org/apache/pinot/query/runtime/plan/OpChainConverter.java
@@ -20,9 +20,16 @@ package org.apache.pinot.query.runtime.plan;
import com.google.auto.service.AutoService;
import java.util.function.BiConsumer;
+import org.apache.commons.lang3.StringUtils;
+import org.apache.pinot.query.planner.plannode.MailboxSendNode;
import org.apache.pinot.query.planner.plannode.PlanNode;
+import org.apache.pinot.query.routing.StagePlan;
+import org.apache.pinot.query.runtime.blocks.ErrorMseBlock;
+import org.apache.pinot.query.runtime.operator.ErrorOperator;
+import org.apache.pinot.query.runtime.operator.MailboxSendOperator;
import org.apache.pinot.query.runtime.operator.MultiStageOperator;
import org.apache.pinot.query.runtime.operator.OpChain;
+import org.apache.pinot.spi.exception.QueryErrorCode;
/**
@@ -58,6 +65,39 @@ public interface OpChainConverter {
*/
OpChain convert(PlanNode node, OpChainExecutionContext context, BiConsumer<PlanNode, MultiStageOperator> tracker);
+ /**
+ * Builds a minimal op-chain that sends an already-detected stage error to downstream receivers.
+ *
+ * This is used, for example, when a pipeline breaker detects an error and needs to propagate it to downstream stages.
+ *
+ * <p>The default implementation uses a Java {@link ErrorOperator} + {@link MailboxSendOperator} pipeline.
+ */
+ default OpChain sendEarlyError(OpChainExecutionContext context, StagePlan stagePlan,
+ ErrorMseBlock errorBlock) {
+ if (!(stagePlan.getRootNode() instanceof MailboxSendNode)) {
+ throw new IllegalStateException("Expected MailboxSendNode root for error propagation, got: "
+ + stagePlan.getRootNode().getClass().getSimpleName());
+ }
+ MailboxSendNode rootNode = (MailboxSendNode) stagePlan.getRootNode();
+ String message = extractErrorMessage(errorBlock);
+ MultiStageOperator errorOp = new ErrorOperator(context, QueryErrorCode.QUERY_EXECUTION, message);
+ MultiStageOperator sendOp = new MailboxSendOperator(context, errorOp, rootNode);
+ return new OpChain(context, sendOp);
+ }
+
+ private static String extractErrorMessage(ErrorMseBlock errorBlock) {
+ String message = errorBlock.getErrorMessages().get(errorBlock.getMainErrorCode());
+ if (StringUtils.isNotBlank(message)) {
+ return message;
+ }
+ for (String value : errorBlock.getErrorMessages().values()) {
+ if (StringUtils.isNotBlank(value)) {
+ return value;
+ }
+ }
+ return "Query execution failed";
+ }
+
@AutoService(OpChainConverter.class)
class DefaultOpChainConverter implements OpChainConverter {
public static final String CONVERTER_ID = "DEFAULT";
diff --git a/pinot-query-runtime/src/main/java/org/apache/pinot/query/runtime/plan/OpChainConverterDispatcher.java b/pinot-query-runtime/src/main/java/org/apache/pinot/query/runtime/plan/OpChainConverterDispatcher.java
index 92556d288b8..0bf2c80ca14 100644
--- a/pinot-query-runtime/src/main/java/org/apache/pinot/query/runtime/plan/OpChainConverterDispatcher.java
+++ b/pinot-query-runtime/src/main/java/org/apache/pinot/query/runtime/plan/OpChainConverterDispatcher.java
@@ -26,6 +26,8 @@ import java.util.function.BiConsumer;
import javax.annotation.Nullable;
import org.apache.commons.lang3.StringUtils;
import org.apache.pinot.query.planner.plannode.PlanNode;
+import org.apache.pinot.query.routing.StagePlan;
+import org.apache.pinot.query.runtime.blocks.ErrorMseBlock;
import org.apache.pinot.query.runtime.operator.MultiStageOperator;
import org.apache.pinot.query.runtime.operator.OpChain;
import org.slf4j.Logger;
@@ -75,6 +77,11 @@ public final class OpChainConverterDispatcher {
return _active.convert(node, context, tracker);
}
+ public static OpChain sendEarlyError(OpChainExecutionContext context, StagePlan stagePlan,
+ ErrorMseBlock errorBlock) {
+ return _active.sendEarlyError(context, stagePlan, errorBlock);
+ }
+
private static OpChainConverter resolveActiveConverter(@Nullable String overrideConverterId) {
Map<String, OpChainConverter> byId = loadConvertersById();
String selectedId;
diff --git a/pinot-query-runtime/src/test/java/org/apache/pinot/query/runtime/plan/OpChainConverterDispatcherTest.java b/pinot-query-runtime/src/test/java/org/apache/pinot/query/runtime/plan/OpChainConverterDispatcherTest.java
index dbfd51cb573..f2f23446966 100644
--- a/pinot-query-runtime/src/test/java/org/apache/pinot/query/runtime/plan/OpChainConverterDispatcherTest.java
+++ b/pinot-query-runtime/src/test/java/org/apache/pinot/query/runtime/plan/OpChainConverterDispatcherTest.java
@@ -18,9 +18,37 @@
*/
package org.apache.pinot.query.runtime.plan;
+import java.util.List;
+import java.util.Map;
+import org.apache.calcite.rel.RelDistribution;
+import org.apache.pinot.calcite.rel.logical.PinotRelExchangeType;
+import org.apache.pinot.common.utils.DataSchema;
+import org.apache.pinot.query.mailbox.MailboxService;
+import org.apache.pinot.query.mailbox.SendingMailbox;
+import org.apache.pinot.query.planner.plannode.MailboxSendNode;
+import org.apache.pinot.query.planner.plannode.PlanNode;
+import org.apache.pinot.query.planner.plannode.ValueNode;
+import org.apache.pinot.query.routing.MailboxInfo;
+import org.apache.pinot.query.routing.MailboxInfos;
+import org.apache.pinot.query.routing.StageMetadata;
+import org.apache.pinot.query.routing.StagePlan;
+import org.apache.pinot.query.routing.WorkerMetadata;
+import org.apache.pinot.query.runtime.blocks.ErrorMseBlock;
+import org.apache.pinot.query.runtime.blocks.MseBlock;
+import org.apache.pinot.query.runtime.operator.MailboxSendOperator;
+import org.apache.pinot.query.runtime.operator.OpChain;
+import org.apache.pinot.spi.exception.QueryErrorCode;
+import org.testng.Assert;
import org.testng.annotations.AfterMethod;
import org.testng.annotations.Test;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.anyInt;
+import static org.mockito.ArgumentMatchers.anyLong;
+import static org.mockito.ArgumentMatchers.anyString;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
public class OpChainConverterDispatcherTest {
@@ -33,4 +61,65 @@ public class OpChainConverterDispatcherTest {
public void testUnknownOverrideRejected() {
OpChainConverterDispatcher.setActiveConverterIdOverride("no-such-converter-id");
}
+
+ @Test
+ public void testSendEarlyErrorReturnsMailboxSendOpChain() {
+ OpChainConverterDispatcher.setActiveConverterIdOverride("default");
+ DataSchema schema = new DataSchema(
+ new String[]{"c1"},
+ new DataSchema.ColumnDataType[]{DataSchema.ColumnDataType.INT});
+ int stageId = 1;
+ int receiverStageId = 0;
+ int workerId = 0;
+
+ MailboxService mailboxService = mock(MailboxService.class);
+ when(mailboxService.getHostname()).thenReturn("localhost");
+ when(mailboxService.getPort()).thenReturn(8000);
+ when(mailboxService.getSendingMailbox(anyString(), anyInt(), anyString(), anyLong(), any()))
+ .thenReturn(mock(SendingMailbox.class));
+
+ MailboxInfo receiverMailboxInfo = new MailboxInfo("localhost", 18080, List.of(0));
+ WorkerMetadata workerMetadata =
+ new WorkerMetadata(workerId, Map.of(receiverStageId, new MailboxInfos(receiverMailboxInfo)));
+ StageMetadata stageMetadata = new StageMetadata(stageId, List.of(workerMetadata), Map.of());
+ OpChainExecutionContext context =
+ new OpChainExecutionContext(mailboxService, 123L, "cid-123", System.currentTimeMillis() + 60_000,
+ System.currentTimeMillis() + 60_000, "broker", Map.of(), stageMetadata, workerMetadata, null, false, false);
+ StagePlan stagePlan = new StagePlan(createMailboxSendNode(schema, stageId, receiverStageId), stageMetadata);
+ ErrorMseBlock errorBlock = ErrorMseBlock.fromError(QueryErrorCode.QUERY_EXECUTION, "simulated pipeline failure");
+
+ OpChain opChain = OpChainConverterDispatcher.sendEarlyError(context, stagePlan, errorBlock);
+ try {
+ Assert.assertNotNull(opChain);
+ Assert.assertTrue(opChain.getRoot() instanceof MailboxSendOperator,
+ "Default converter should build a mailbox-send opchain for early errors");
+ MseBlock block = opChain.getRoot().nextBlock();
+ Assert.assertTrue(block.isError(), "Early-error opchain should immediately emit an error block");
+ Assert.assertTrue(block instanceof ErrorMseBlock);
+ ErrorMseBlock emittedError = (ErrorMseBlock) block;
+ QueryErrorCode mainErrorCode = emittedError.getMainErrorCode();
+ Assert.assertEquals(mainErrorCode, QueryErrorCode.QUERY_EXECUTION);
+ String message = emittedError.getErrorMessages().get(mainErrorCode);
+ Assert.assertTrue(message.contains("simulated pipeline failure"),
+ "Unexpected error message from early-error opchain: " + message);
+ } finally {
+ opChain.close();
+ }
+ }
+
+ private static MailboxSendNode createMailboxSendNode(DataSchema schema, int stageId, int receiverStageId) {
+ PlanNode inputNode = new ValueNode(stageId, schema, PlanNode.NodeHint.EMPTY, List.of(), List.of());
+ return new MailboxSendNode(
+ stageId,
+ schema,
+ List.of(inputNode),
+ receiverStageId,
+ PinotRelExchangeType.STREAMING,
+ RelDistribution.Type.RANDOM_DISTRIBUTED,
+ List.of(),
+ false,
+ List.of(),
+ false,
+ "MURMUR3");
+ }
}
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org