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 6e92426ef75 Report idle workers and the slowest worker's clock time in
MSE stage stats (#19364)
6e92426ef75 is described below
commit 6e92426ef75d9648ae4bc2dce58f542e61f0baec
Author: Gonzalo Ortiz Jaureguizar <[email protected]>
AuthorDate: Fri Aug 28 15:53:37 2026 +0200
Report idle workers and the slowest worker's clock time in MSE stage stats
(#19364)
---
.../core/query/request/ServerQueryRequest.java | 23 ++++
.../core/query/request/ServerQueryRequestTest.java | 91 +++++++++++++
.../pinot/query/runtime/SendStatsPredicate.java | 9 +-
.../operator/BaseMailboxReceiveOperator.java | 27 +++-
.../pinot/query/runtime/operator/LeafOperator.java | 32 ++++-
.../runtime/operator/MailboxSendOperator.java | 62 ++++++++-
.../query/runtime/operator/MultiStageOperator.java | 6 +
.../query/runtime/operator/LeafOperatorTest.java | 45 +++++++
.../operator/MailboxReceiveOperatorTest.java | 35 +++++
.../runtime/operator/MailboxSendOperatorTest.java | 146 ++++++++++++++++++++-
.../query/runtime/queries/QueryRunnerTest.java | 129 ++++++++++++++++++
11 files changed, 593 insertions(+), 12 deletions(-)
diff --git
a/pinot-core/src/main/java/org/apache/pinot/core/query/request/ServerQueryRequest.java
b/pinot-core/src/main/java/org/apache/pinot/core/query/request/ServerQueryRequest.java
index d9c0041f10a..7f3666624ac 100644
---
a/pinot-core/src/main/java/org/apache/pinot/core/query/request/ServerQueryRequest.java
+++
b/pinot-core/src/main/java/org/apache/pinot/core/query/request/ServerQueryRequest.java
@@ -22,6 +22,7 @@ import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import javax.annotation.Nullable;
+import org.apache.commons.collections4.CollectionUtils;
import org.apache.pinot.common.metrics.ServerMetrics;
import org.apache.pinot.common.proto.Server;
import org.apache.pinot.common.request.BrokerRequest;
@@ -205,10 +206,32 @@ public class ServerQueryRequest {
return _queryContext.getTableName();
}
+ /// The segments assigned to this worker, or null when the request carries
them per referenced table in
+ /// [#getTableSegmentsContexts()] instead, as it does for a logical table.
Callers must check that one first, the
+ /// way `ServerQueryExecutorV1Impl` does, or use [#hasSegmentsToQuery()]
when all they need is whether this
+ /// request has any segment at all.
+ @Nullable
public List<String> getSegmentsToQuery() {
return _segmentsToQuery;
}
+ /// Whether this request has at least one segment to read, whichever of the
two representations carries them.
+ ///
+ /// A request holds its segments either flat in [#getSegmentsToQuery()], for
a plain table, or grouped per
+ /// referenced table in [#getTableSegmentsContexts()], for a logical table;
the other one is null. Resolving that
+ /// here keeps callers that only need the question answered from having to
know which representation applies.
+ public boolean hasSegmentsToQuery() {
+ if (_tableSegmentsContexts != null) {
+ for (TableSegmentsContext tableSegmentsContext : _tableSegmentsContexts)
{
+ if (CollectionUtils.isNotEmpty(tableSegmentsContext.getSegments())) {
+ return true;
+ }
+ }
+ return false;
+ }
+ return CollectionUtils.isNotEmpty(_segmentsToQuery);
+ }
+
public List<String> getOptionalSegments() {
return _optionalSegments;
}
diff --git
a/pinot-core/src/test/java/org/apache/pinot/core/query/request/ServerQueryRequestTest.java
b/pinot-core/src/test/java/org/apache/pinot/core/query/request/ServerQueryRequestTest.java
new file mode 100644
index 00000000000..19a5c471907
--- /dev/null
+++
b/pinot-core/src/test/java/org/apache/pinot/core/query/request/ServerQueryRequestTest.java
@@ -0,0 +1,91 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pinot.core.query.request;
+
+import java.util.List;
+import javax.annotation.Nullable;
+import org.apache.pinot.common.metrics.ServerMetrics;
+import org.apache.pinot.common.request.InstanceRequest;
+import org.apache.pinot.common.request.TableSegmentsInfo;
+import org.apache.pinot.sql.parsers.CalciteSqlCompiler;
+import org.testng.annotations.Test;
+
+import static org.testng.Assert.assertFalse;
+import static org.testng.Assert.assertNull;
+import static org.testng.Assert.assertTrue;
+
+
+/// Tests how a request answers whether it has segments to read, which depends
on which of its two mutually
+/// exclusive segment representations is populated.
+public class ServerQueryRequestTest {
+
+ @Test
+ public void shouldHaveSegmentsWhenTheFlatListIsPopulated() {
+ assertTrue(request(List.of("segment1"), null).hasSegmentsToQuery());
+ }
+
+ @Test
+ public void shouldNotHaveSegmentsWhenTheFlatListIsEmpty() {
+ assertFalse(request(List.of(), null).hasSegmentsToQuery());
+ }
+
+ /// A logical table sets the per-table lists instead and leaves the flat one
unset, which Thrift reports as null.
+ /// Reading it without checking would throw, so this pins both that it stays
null and that the question is still
+ /// answered from the other representation.
+ @Test
+ public void shouldHaveSegmentsWhenOnlyThePerTableListsArePopulated() {
+ ServerQueryRequest request =
+ request(null, List.of(new TableSegmentsInfo("tbl_OFFLINE",
List.of("segment1"))));
+
+ assertNull(request.getSegmentsToQuery(), "expected the flat list to be
null on the logical table path");
+ assertTrue(request.hasSegmentsToQuery());
+ }
+
+ @Test
+ public void shouldNotHaveSegmentsWhenThePerTableListsAreAllEmpty() {
+ ServerQueryRequest request = request(null,
+ List.of(new TableSegmentsInfo("tbl1_OFFLINE", List.of()), new
TableSegmentsInfo("tbl2_OFFLINE", List.of())));
+
+ assertFalse(request.hasSegmentsToQuery());
+ }
+
+ @Test
+ public void shouldHaveSegmentsWhenOnlyOneOfThePerTableListsIsPopulated() {
+ assertTrue(request(null,
+ List.of(new TableSegmentsInfo("tbl1_OFFLINE", List.of()),
+ new TableSegmentsInfo("tbl2_OFFLINE",
List.of("segment1")))).hasSegmentsToQuery());
+ }
+
+ /// Exactly one of the two arguments is set on a real request, mirroring
+ /// `ServerPlanRequestUtils.compileInstanceRequest`.
+ private static ServerQueryRequest request(@Nullable List<String>
searchSegments,
+ @Nullable List<TableSegmentsInfo> tableSegmentsInfoList) {
+ InstanceRequest instanceRequest = new InstanceRequest();
+ instanceRequest.setRequestId(1);
+ instanceRequest.setBrokerId("broker");
+ instanceRequest.setQuery(CalciteSqlCompiler.compileToBrokerRequest("SELECT
* FROM tbl"));
+ if (searchSegments != null) {
+ instanceRequest.setSearchSegments(searchSegments);
+ }
+ if (tableSegmentsInfoList != null) {
+ instanceRequest.setTableSegmentsInfoList(tableSegmentsInfoList);
+ }
+ return new ServerQueryRequest(instanceRequest, ServerMetrics.get(),
System.currentTimeMillis());
+ }
+}
diff --git
a/pinot-query-runtime/src/main/java/org/apache/pinot/query/runtime/SendStatsPredicate.java
b/pinot-query-runtime/src/main/java/org/apache/pinot/query/runtime/SendStatsPredicate.java
index da2a6cb5bc1..a6d2b4b5c6c 100644
---
a/pinot-query-runtime/src/main/java/org/apache/pinot/query/runtime/SendStatsPredicate.java
+++
b/pinot-query-runtime/src/main/java/org/apache/pinot/query/runtime/SendStatsPredicate.java
@@ -49,10 +49,11 @@ import org.slf4j.LoggerFactory;
/// Therefore the cleanest and safer solution is to not send stats when we
know a problematic version is in the cluster.
///
/// We support three modes:
-/// - SAFE: This is the default mode. In this mode, we will send stats unless
we detect a problematic version in the
-/// cluster. This doesn't require human intervention and is the recommended
mode.
-/// - ALWAYS: In this mode, we will always send stats, regardless of the
version of the cluster. This mimics the
-/// behavior in 1.3.0 and lower versions.
+/// - ALWAYS: This is the default mode. In this mode, we will always send
stats, regardless of the version of the
+/// cluster.
+/// - SAFE: In this mode, we will send stats unless we detect a problematic
version in the cluster, which means any
+/// instance reporting a version other than this one. This doesn't require
human intervention, and is the mode to
+/// use for a cluster that may still run versions older than 1.4.
/// - NEVER: In this mode, we will never send stats, regardless of the version
of the cluster. This is useful for
/// testing purposes or if for whatever reason you want to disable stats.
public abstract class SendStatsPredicate implements
InstanceConfigChangeListener {
diff --git
a/pinot-query-runtime/src/main/java/org/apache/pinot/query/runtime/operator/BaseMailboxReceiveOperator.java
b/pinot-query-runtime/src/main/java/org/apache/pinot/query/runtime/operator/BaseMailboxReceiveOperator.java
index 0fcc70341e1..4e19b8a3a47 100644
---
a/pinot-query-runtime/src/main/java/org/apache/pinot/query/runtime/operator/BaseMailboxReceiveOperator.java
+++
b/pinot-query-runtime/src/main/java/org/apache/pinot/query/runtime/operator/BaseMailboxReceiveOperator.java
@@ -120,9 +120,16 @@ public abstract class BaseMailboxReceiveOperator extends
MultiStageOperator {
return _multiConsumer.calculateStats();
}
+ /// Returns a copy of this operator's stats, extended with
[StatKey#NON_ACTIVE_WORKERS] for this single worker.
@Override
public StatMap<StatKey> copyStatMaps() {
- return new StatMap<>(_statMap);
+ StatMap<StatKey> statMap = new StatMap<>(_statMap);
+ // This operator hands downstream what it reads from its mailboxes, so
having emitted no row means having
+ // received none.
+ if (statMap.getLong(StatKey.EMITTED_ROWS) == 0) {
+ statMap.merge(StatKey.NON_ACTIVE_WORKERS, 1);
+ }
+ return statMap;
}
protected void onEos() {
@@ -203,6 +210,10 @@ public abstract class BaseMailboxReceiveOperator extends
MultiStageOperator {
}
}
+ /// The stats reported by this operator.
+ ///
+ /// New keys must be appended at the end of this enum: [StatMap] identifies
keys by their ordinal on the wire, so
+ /// inserting, reordering or removing a constant breaks the compatibility
with other versions.
public enum StatKey implements StatMap.Key {
EXECUTION_TIME_MS(StatMap.Type.LONG) {
@Override
@@ -246,7 +257,19 @@ public abstract class BaseMailboxReceiveOperator extends
MultiStageOperator {
/// Allocated memory in bytes for this operator or its children in the
same stage.
ALLOCATED_MEMORY_BYTES(StatMap.Type.LONG),
/// Time spent on GC while this operator or its children in the same stage
were running.
- GC_TIME_MS(StatMap.Type.LONG);
+ GC_TIME_MS(StatMap.Type.LONG),
+ /// How many workers of this stage emitted no row out of this mailbox
receive.
+ ///
+ /// Reported as the count of idle workers rather than active ones so that
it is absent when every worker
+ /// received something, which is the common case. This is what tells apart
a worker that was handed no data at
+ /// all from one that was handed data and filtered it away downstream:
compare it against the `parallelism` the
+ /// stats tree renders on this node, and against what the operators above
and below report.
+ ///
+ /// This operator hands downstream what it reads, so emitting no row means
having received none. The converse
+ /// can fail in two corner cases, where rows are read but never emitted:
after the downstream operator has
+ /// early terminated, and when a sorted receive buffers its rows and then
ends in error. Such a worker is
+ /// reported as idle despite having been given data.
+ NON_ACTIVE_WORKERS(StatMap.Type.INT);
private final StatMap.Type _type;
diff --git
a/pinot-query-runtime/src/main/java/org/apache/pinot/query/runtime/operator/LeafOperator.java
b/pinot-query-runtime/src/main/java/org/apache/pinot/query/runtime/operator/LeafOperator.java
index 6b5ae73499b..d80adc418c8 100644
---
a/pinot-query-runtime/src/main/java/org/apache/pinot/query/runtime/operator/LeafOperator.java
+++
b/pinot-query-runtime/src/main/java/org/apache/pinot/query/runtime/operator/LeafOperator.java
@@ -290,9 +290,28 @@ public class LeafOperator extends MultiStageOperator {
}
}
+ /// Returns a copy of this operator's stats, extended with
[StatKey#NON_ACTIVE_WORKERS] for this single worker.
@Override
public StatMap<StatKey> copyStatMaps() {
- return new StatMap<>(_statMap);
+ StatMap<StatKey> statMap = new StatMap<>(_statMap);
+ if (!hasSegmentsAssigned()) {
+ statMap.merge(StatKey.NON_ACTIVE_WORKERS, 1);
+ }
+ return statMap;
+ }
+
+ /// Whether this worker was given at least one segment to read.
+ ///
+ /// Decided from the request rather than from anything the query produces,
so that a worker whose segments are all
+ /// pruned, or all of whose rows are filtered out, still counts as having
been given work. A hybrid table produces
+ /// one request per table type, and segments on either of them are enough.
+ private boolean hasSegmentsAssigned() {
+ for (ServerQueryRequest request : _requests) {
+ if (request.hasSegmentsToQuery()) {
+ return true;
+ }
+ }
+ return false;
}
@Override
@@ -768,11 +787,20 @@ public class LeafOperator extends MultiStageOperator {
GC_TIME_MS(StatMap.Type.LONG, null),
/// Time spent in single-stage execution engine for this leaf stage.
SSE_EXECUTION_TIME_MS(StatMap.Type.LONG, null),
- EARLY_TERMINATION_REASONS(StatMap.Type.STRING_SET);
+ EARLY_TERMINATION_REASONS(StatMap.Type.STRING_SET),
+ /// How many workers of this stage had no segment assigned to them.
+ ///
+ /// Reported as the count of idle workers rather than active ones so that
it is absent when every worker was
+ /// given something to read, which is the common case. A worker with no
segment was given no work at all, which
+ /// is different from a worker that read segments and produced nothing:
compare this against the `parallelism`
+ /// the stats tree renders on this node, and against what the operators
above it report.
+ NON_ACTIVE_WORKERS(StatMap.Type.INT, null);
// IMPORTANT: When adding new StatKeys, make sure to either create the
same key in BrokerResponseNativeV2.StatKey or
// call the constructor that accepts a String as last argument and set it
to null.
// Otherwise the constructor will fail with an IllegalArgumentException
which will not be caught and will
// propagate to the caller, causing the query to timeout.
+ // New keys must also be appended at the end: StatMap identifies keys by
their ordinal on the wire, so
+ // inserting, reordering or removing a constant breaks the compatibility
with other versions.
private final StatMap.Type _type;
@Nullable
diff --git
a/pinot-query-runtime/src/main/java/org/apache/pinot/query/runtime/operator/MailboxSendOperator.java
b/pinot-query-runtime/src/main/java/org/apache/pinot/query/runtime/operator/MailboxSendOperator.java
index 3e748deaa97..d0d19b3ac81 100644
---
a/pinot-query-runtime/src/main/java/org/apache/pinot/query/runtime/operator/MailboxSendOperator.java
+++
b/pinot-query-runtime/src/main/java/org/apache/pinot/query/runtime/operator/MailboxSendOperator.java
@@ -270,9 +270,27 @@ public class MailboxSendOperator extends
MultiStageOperator {
}
}
+ /// Returns a copy of this operator's stats, extended with the stats
describing this single worker.
+ ///
+ /// These cannot be accumulated in [#registerExecution] like the others,
because they are not per-block
+ /// quantities: merging [StatKey#MAX_EMITTED_ROWS] once per block would
report the largest block rather than the
+ /// busiest worker, and merging [StatKey#NON_ACTIVE_WORKERS] once per block
would count the blocks. They are
+ /// computed once, here, from the totals the operator ends up with.
+ ///
+ /// They describe a single worker, but every stat is merged across all the
workers of the stage before being
+ /// reported, which is what turns them into a description of how the work
was spread.
@Override
public StatMap<StatKey> copyStatMaps() {
- return new StatMap<>(_statMap);
+ StatMap<StatKey> statMap = new StatMap<>(_statMap);
+ long emittedRows = statMap.getLong(StatKey.EMITTED_ROWS);
+ if (emittedRows > 0) {
+ statMap.merge(StatKey.MAX_EMITTED_ROWS, emittedRows);
+ } else {
+ statMap.merge(StatKey.NON_ACTIVE_WORKERS, 1);
+ }
+ // Reported by every worker, idle ones included: a worker that sent
nothing still spent time deciding that.
+ statMap.merge(StatKey.MAX_CLOCK_TIME_MS,
statMap.getLong(StatKey.EXECUTION_TIME_MS));
+ return statMap;
}
private void sendMseBlock(MseBlock.Data block) {
@@ -319,6 +337,13 @@ public class MailboxSendOperator extends
MultiStageOperator {
}
}
+ /// The stats reported by this operator.
+ ///
+ /// As the root operator of its stage, this operator is also where the
stage-wide stats live, like [#PARALLELISM]
+ /// and [#NON_ACTIVE_WORKERS].
+ ///
+ /// New keys must be appended at the end of this enum: [StatMap] identifies
keys by their ordinal on the wire, so
+ /// inserting, reordering or removing a constant breaks the compatibility
with other versions.
public enum StatKey implements StatMap.Key {
EXECUTION_TIME_MS(StatMap.Type.LONG) {
@Override
@@ -381,7 +406,40 @@ public class MailboxSendOperator extends
MultiStageOperator {
/// Allocated memory in bytes for this operator or its children in the
same stage.
ALLOCATED_MEMORY_BYTES(StatMap.Type.LONG),
/// Time spent on GC while this operator or its children in the same stage
were running.
- GC_TIME_MS(StatMap.Type.LONG);
+ GC_TIME_MS(StatMap.Type.LONG),
+ /// How many workers of this stage sent no row at all.
+ ///
+ /// Reported as the count of idle workers rather than active ones so that
it is absent from the stats of a
+ /// stage where every worker produced something, which is the common case.
+ ///
+ /// Each operator reports which of the stage's workers it was idle on,
applying its own notion of activity:
+ /// this one sent no row, a mailbox receive operator received none, a leaf
operator had no segment assigned to
+ /// it. Comparing this against `parallelism` on the same node detects
distribution bias, and comparing it
+ /// against the operators below shows where the stage narrowed: a leaf
idle on no worker under a send idle on
+ /// nine means the work was spread but the output was not.
+ NON_ACTIVE_WORKERS(StatMap.Type.INT),
+ /// The highest number of rows sent by a single worker of this stage.
+ ///
+ /// [#EMITTED_ROWS] is the sum across all workers, so `maxEmittedRows`
greatly exceeding the average number of
+ /// rows per worker means the rows were not evenly distributed.
+ MAX_EMITTED_ROWS(StatMap.Type.LONG) {
+ @Override
+ public long merge(long value1, long value2) {
+ return Math.max(value1, value2);
+ }
+ },
+ /// How long the slowest worker of this stage took.
+ ///
+ /// The `clockTimeMs` reported for a stage is its [#EXECUTION_TIME_MS]
divided by its [#PARALLELISM], which
+ /// assumes the work was spread evenly across the workers. This is the
same measure taken on the worker that
+ /// took longest, so `maxClockTimeMs` greatly exceeding `clockTimeMs`
means that assumption does not hold and
+ /// the average understates how long the stage actually took.
+ MAX_CLOCK_TIME_MS(StatMap.Type.LONG) {
+ @Override
+ public long merge(long value1, long value2) {
+ return Math.max(value1, value2);
+ }
+ };
private final StatMap.Type _type;
diff --git
a/pinot-query-runtime/src/main/java/org/apache/pinot/query/runtime/operator/MultiStageOperator.java
b/pinot-query-runtime/src/main/java/org/apache/pinot/query/runtime/operator/MultiStageOperator.java
index 479f0fdeed5..0bead828f0f 100644
---
a/pinot-query-runtime/src/main/java/org/apache/pinot/query/runtime/operator/MultiStageOperator.java
+++
b/pinot-query-runtime/src/main/java/org/apache/pinot/query/runtime/operator/MultiStageOperator.java
@@ -230,6 +230,12 @@ public abstract class MultiStageOperator implements
Operator<MseBlock>, AutoClos
.orElse(MultiStageQueryStats.emptyStats(_context.getStageId()));
}
+ /// Returns the stats to report for this operator, as a copy that the caller
is free to merge into.
+ ///
+ /// Implementations may derive extra stats here instead of only copying the
ones they accumulated while running,
+ /// but this method is called several times per opchain and every call must
return the same values: deriving a
+ /// stat whose merge function is not idempotent (a sum, for instance) must
be done on the returned copy, never on
+ /// the stat map the operator keeps.
public abstract StatMap<?> copyStatMaps();
// TODO: Ideally close() call should finish within request deadline.
diff --git
a/pinot-query-runtime/src/test/java/org/apache/pinot/query/runtime/operator/LeafOperatorTest.java
b/pinot-query-runtime/src/test/java/org/apache/pinot/query/runtime/operator/LeafOperatorTest.java
index 21bea3dbbac..237d7033ff0 100644
---
a/pinot-query-runtime/src/test/java/org/apache/pinot/query/runtime/operator/LeafOperatorTest.java
+++
b/pinot-query-runtime/src/test/java/org/apache/pinot/query/runtime/operator/LeafOperatorTest.java
@@ -124,6 +124,51 @@ public class LeafOperatorTest {
return queryRequests;
}
+ private ServerQueryRequest mockQueryRequest(boolean hasSegments) {
+ ServerQueryRequest queryRequest = mock(ServerQueryRequest.class);
+ when(queryRequest.getQueryContext()).thenReturn(mock(QueryContext.class));
+ when(queryRequest.hasSegmentsToQuery()).thenReturn(hasSegments);
+ return queryRequest;
+ }
+
+ private int idleWorkers(List<ServerQueryRequest> requests) {
+ OpChainExecutionContext context = OperatorTestUtil.getTracingContext();
+ DataSchema schema = new DataSchema(new String[]{"intCol"},
+ new DataSchema.ColumnDataType[]{DataSchema.ColumnDataType.INT});
+ try (LeafOperator operator = new LeafOperator(context, requests, schema,
mock(QueryExecutor.class),
+ _executorService)) {
+ // Collected more than once, as the runtime does, so a derivation moved
onto the shared stat map would show up
+ // here as a doubled count.
+ operator.copyStatMaps();
+ return
operator.copyStatMaps().getInt(LeafOperator.StatKey.NON_ACTIVE_WORKERS);
+ }
+ }
+
+ /// A leaf operator is idle on the workers that were given no segment to
read. This is decided from the request
+ /// rather than from anything the query produces, so a worker whose segments
are all pruned, or all of whose rows
+ /// are filtered out, is not idle.
+ ///
+ /// Which of the request's two segment representations carries them, and
that one of them is null, is
+ /// [ServerQueryRequest#hasSegmentsToQuery()]'s business and is stubbed out
here.
+ @Test
+ public void shouldNotReportIdleWorkerWhenSegmentsAreAssigned() {
+ assertEquals(idleWorkers(List.of(mockQueryRequest(true))), 0,
+ "expected a worker with a segment assigned not to count as idle");
+ }
+
+ @Test
+ public void shouldReportIdleWorkerWhenNoSegmentIsAssigned() {
+ assertEquals(idleWorkers(List.of(mockQueryRequest(false))), 1,
+ "expected a worker with no segment assigned to be counted as idle");
+ }
+
+ /// A hybrid table produces one request per table type; having segments in
either of them is enough.
+ @Test
+ public void shouldNotReportIdleWorkerWhenOnlyTheSecondRequestHasSegments() {
+ assertEquals(idleWorkers(List.of(mockQueryRequest(false),
mockQueryRequest(true))), 0,
+ "expected segments on any of the requests to keep the worker out of
the idle count");
+ }
+
@Test
public void shouldReturnDataBlockThenMetadataBlock() {
// Given:
diff --git
a/pinot-query-runtime/src/test/java/org/apache/pinot/query/runtime/operator/MailboxReceiveOperatorTest.java
b/pinot-query-runtime/src/test/java/org/apache/pinot/query/runtime/operator/MailboxReceiveOperatorTest.java
index 6c57ef6ba01..0af5a807b3d 100644
---
a/pinot-query-runtime/src/test/java/org/apache/pinot/query/runtime/operator/MailboxReceiveOperatorTest.java
+++
b/pinot-query-runtime/src/test/java/org/apache/pinot/query/runtime/operator/MailboxReceiveOperatorTest.java
@@ -143,6 +143,41 @@ public class MailboxReceiveOperatorTest {
}
}
+ /// A mailbox receive operator is idle on the workers where it received
nothing, which is what tells a worker
+ /// that was handed no data apart from one that was handed data and filtered
it away downstream.
+ @Test
+ public void shouldNotReportIdleWorkerWhenRowsAreReceived() {
+
when(_mailboxService.getReceivingMailbox(eq(MAILBOX_ID_1))).thenReturn(_mailbox1);
+
when(_mailbox1.poll()).thenReturn(OperatorTestUtil.blockWithStats(DATA_SCHEMA,
new Object[]{1, 1}),
+ OperatorTestUtil.eosWithEmptyStats());
+ try (MailboxReceiveOperator operator = getOperator(_stageMetadata1,
RelDistribution.Type.SINGLETON)) {
+ drain(operator);
+
assertEquals(operator.copyStatMaps().getInt(BaseMailboxReceiveOperator.StatKey.NON_ACTIVE_WORKERS),
0,
+ "expected receiving a row to keep this worker out of the idle
count");
+ }
+ }
+
+ @Test
+ public void shouldReportIdleWorkerWhenNoRowIsReceived() {
+
when(_mailboxService.getReceivingMailbox(eq(MAILBOX_ID_1))).thenReturn(_mailbox1);
+ when(_mailbox1.poll()).thenReturn(OperatorTestUtil.eosWithEmptyStats());
+ try (MailboxReceiveOperator operator = getOperator(_stageMetadata1,
RelDistribution.Type.SINGLETON)) {
+ drain(operator);
+ // Collected more than once, as the runtime does, so a derivation moved
onto the shared stat map would show up
+ // here as a doubled count.
+ operator.copyStatMaps();
+
assertEquals(operator.copyStatMaps().getInt(BaseMailboxReceiveOperator.StatKey.NON_ACTIVE_WORKERS),
1,
+ "expected a worker that received no row to be counted as idle");
+ }
+ }
+
+ private static void drain(MailboxReceiveOperator operator) {
+ MseBlock block = operator.nextBlock();
+ while (block.isData()) {
+ block = operator.nextBlock();
+ }
+ }
+
@Test
public void shouldReceiveSingletonErrorMailbox() {
when(_mailboxService.getReceivingMailbox(eq(MAILBOX_ID_1))).thenReturn(_mailbox1);
diff --git
a/pinot-query-runtime/src/test/java/org/apache/pinot/query/runtime/operator/MailboxSendOperatorTest.java
b/pinot-query-runtime/src/test/java/org/apache/pinot/query/runtime/operator/MailboxSendOperatorTest.java
index 88d07ad2694..67c1b20eb63 100644
---
a/pinot-query-runtime/src/test/java/org/apache/pinot/query/runtime/operator/MailboxSendOperatorTest.java
+++
b/pinot-query-runtime/src/test/java/org/apache/pinot/query/runtime/operator/MailboxSendOperatorTest.java
@@ -18,8 +18,14 @@
*/
package org.apache.pinot.query.runtime.operator;
+import java.io.ByteArrayInputStream;
+import java.io.ByteArrayOutputStream;
+import java.io.DataInputStream;
+import java.io.DataOutputStream;
+import java.util.Arrays;
import java.util.List;
import java.util.Map;
+import org.apache.pinot.common.datatable.StatMap;
import org.apache.pinot.common.utils.DataSchema;
import org.apache.pinot.common.utils.DataSchema.ColumnDataType;
import org.apache.pinot.query.mailbox.MailboxService;
@@ -42,6 +48,7 @@ import static org.mockito.ArgumentMatchers.any;
import static org.mockito.ArgumentMatchers.eq;
import static org.mockito.Mockito.*;
import static org.mockito.MockitoAnnotations.openMocks;
+import static org.testng.Assert.assertEquals;
import static org.testng.Assert.assertSame;
import static org.testng.Assert.assertTrue;
@@ -211,17 +218,152 @@ public class MailboxSendOperatorTest {
"total " + total + " exceeds the " + wallTimeMs + "ms the call
actually took, so it was counted twice");
}
+ @Test
+ public void shouldReportPerWorkerStats()
+ throws Exception {
+ // Given: a worker that sends two single-row blocks
+ when(_input.nextBlock()).thenReturn(getDummyDataBlock(),
getDummyDataBlock(), SuccessMseBlock.INSTANCE);
+
+ // When:
+ MailboxSendOperator operator = getOperator();
+ drain(operator);
+
+ // Then: the per-worker view is this single worker, so the max is its own
emitted rows and it is not idle
+ StatMap<MailboxSendOperator.StatKey> statMap = operator.copyStatMaps();
+ assertEquals(statMap.getLong(MailboxSendOperator.StatKey.EMITTED_ROWS),
2L, "expected 2 emitted rows");
+
assertEquals(statMap.getLong(MailboxSendOperator.StatKey.MAX_EMITTED_ROWS), 2L,
"expected max to be 2");
+
assertEquals(statMap.getInt(MailboxSendOperator.StatKey.NON_ACTIVE_WORKERS), 0,
"expected no idle worker");
+ }
+
+ /// A send operator is idle exactly when it sent no row, so a worker that
sent nothing is counted however much
+ /// work the stage was given.
+ @Test
+ public void shouldReportIdleWorkerWhenNoRowIsSent()
+ throws Exception {
+ // Given: a worker that sends no data block at all
+ when(_input.nextBlock()).thenReturn(SuccessMseBlock.INSTANCE);
+
+ // When:
+ MailboxSendOperator operator = getOperator();
+ drain(operator);
+
+ // Then:
+ StatMap<MailboxSendOperator.StatKey> statMap = operator.copyStatMaps();
+
assertEquals(statMap.getInt(MailboxSendOperator.StatKey.NON_ACTIVE_WORKERS), 1,
"expected one idle worker");
+
assertEquals(statMap.getLong(MailboxSendOperator.StatKey.MAX_EMITTED_ROWS), 0L,
"expected no max");
+ }
+
+ /// [MultiStageOperator#calculateStats()] runs more than once per opchain,
so the derived stats must not
+ /// accumulate on the operator's own stat map.
+ @Test
+ public void shouldNotDoubleCountIdleWorkersOnRepeatedStatCollection()
+ throws Exception {
+ // Given: a worker that sends nothing, so it is the one counted as idle
+ when(_input.nextBlock()).thenReturn(SuccessMseBlock.INSTANCE);
+
+ // When: stats are collected several times, as the runtime does
+ MailboxSendOperator operator = getOperator();
+ drain(operator);
+ operator.copyStatMaps();
+ operator.copyStatMaps();
+ StatMap<MailboxSendOperator.StatKey> statMap = operator.copyStatMaps();
+
+ // Then: the worker is still counted exactly once
+
assertEquals(statMap.getInt(MailboxSendOperator.StatKey.NON_ACTIVE_WORKERS), 1,
"expected counted once");
+ }
+
+ /// Every stat is merged across the workers of the stage before being
reported, which is where these stats stop
+ /// describing one worker and start describing the distribution.
+ @Test
+ public void shouldMergePerWorkerStatsAcrossWorkers() {
+ // Given: three workers of the same stage, one of which sent nothing
+ StatMap<MailboxSendOperator.StatKey> stage = workerStats(5);
+ stage.merge(workerStats(100));
+ stage.merge(workerStats(0));
+
+ // Then:
+ assertEquals(stage.getLong(MailboxSendOperator.StatKey.EMITTED_ROWS),
105L, "expected rows to be summed");
+ assertEquals(stage.getInt(MailboxSendOperator.StatKey.NON_ACTIVE_WORKERS),
1, "expected 1 of 3 workers idle");
+ assertEquals(stage.getLong(MailboxSendOperator.StatKey.MAX_EMITTED_ROWS),
100L, "expected max across workers");
+ }
+
+ /// The stage's clock time is derived by dividing the summed execution time
by the parallelism, which assumes the
+ /// work was spread evenly. This is the same measure on the worker that took
longest, so it must survive the
+ /// merge as a maximum rather than a sum.
+ @Test
+ public void shouldReportTheSlowestWorkersClockTime() {
+ StatMap<MailboxSendOperator.StatKey> slow = new
StatMap<>(MailboxSendOperator.StatKey.class);
+ slow.merge(MailboxSendOperator.StatKey.MAX_CLOCK_TIME_MS, 500L);
+ StatMap<MailboxSendOperator.StatKey> fast = new
StatMap<>(MailboxSendOperator.StatKey.class);
+ fast.merge(MailboxSendOperator.StatKey.MAX_CLOCK_TIME_MS, 10L);
+
+ slow.merge(fast);
+
+ assertEquals(slow.getLong(MailboxSendOperator.StatKey.MAX_CLOCK_TIME_MS),
500L,
+ "expected the slowest worker to win the merge, not the sum of the
two");
+ }
+
+ @Test
+ public void shouldPreservePerWorkerStatsAcrossSerialization()
+ throws Exception {
+ // Given: two workers whose stats are merged through the serialized form,
as they are across servers
+ StatMap<MailboxSendOperator.StatKey> stage = workerStats(0);
+ ByteArrayOutputStream bytes = new ByteArrayOutputStream();
+ workerStats(100).serialize(new DataOutputStream(bytes));
+
+ // When:
+ stage.merge(new DataInputStream(new
ByteArrayInputStream(bytes.toByteArray())));
+
+ // Then:
+ assertEquals(stage.getInt(MailboxSendOperator.StatKey.NON_ACTIVE_WORKERS),
1, "expected one idle worker");
+ assertEquals(stage.getLong(MailboxSendOperator.StatKey.MAX_EMITTED_ROWS),
100L, "expected max across workers");
+ }
+
+ /// Builds the stats a single worker sending `emittedRows` rows would report.
+ ///
+ /// Each worker gets its own input mock so that draining one does not
exhaust the stubbing of the next.
+ private StatMap<MailboxSendOperator.StatKey> workerStats(int emittedRows) {
+ MultiStageOperator input = mock(MultiStageOperator.class);
+
when(input.calculateStats()).thenReturn(MultiStageQueryStats.emptyStats(SENDER_STAGE_ID));
+ if (emittedRows == 0) {
+ when(input.nextBlock()).thenReturn(SuccessMseBlock.INSTANCE);
+ } else {
+ when(input.nextBlock()).thenReturn(getDummyDataBlock(emittedRows),
SuccessMseBlock.INSTANCE);
+ }
+ MailboxSendOperator operator = getOperator(input);
+ drain(operator);
+ return operator.copyStatMaps();
+ }
+
+ private static void drain(MailboxSendOperator operator) {
+ MseBlock block = operator.nextBlock();
+ while (block.isData()) {
+ block = operator.nextBlock();
+ }
+ }
+
private MailboxSendOperator getOperator() {
+ return getOperator(_input);
+ }
+
+ private MailboxSendOperator getOperator(MultiStageOperator input) {
WorkerMetadata workerMetadata = new WorkerMetadata(0, Map.of(), Map.of());
StageMetadata stageMetadata = new StageMetadata(SENDER_STAGE_ID,
List.of(workerMetadata), Map.of());
OpChainExecutionContext context =
OpChainExecutionContext.fromQueryContext(_mailboxService, Map.of(),
stageMetadata, workerMetadata, null, true,
true, QueryExecutionContext.forMseTest());
- return new MailboxSendOperator(context, _input, statMap -> _exchange);
+ return new MailboxSendOperator(context, input, statMap -> _exchange);
}
private static MseBlock.Data getDummyDataBlock() {
+ return getDummyDataBlock(1);
+ }
+
+ /// Returns a single data block holding `numRows` rows, which must be at
least one.
+ private static MseBlock.Data getDummyDataBlock(int numRows) {
+ Object[][] rows = new Object[numRows][];
+ Arrays.setAll(rows, i -> new Object[]{i});
return OperatorTestUtil.block(new DataSchema(new String[]{"intCol"}, new
ColumnDataType[]{ColumnDataType.INT}),
- new Object[]{1});
+ rows);
}
}
diff --git
a/pinot-query-runtime/src/test/java/org/apache/pinot/query/runtime/queries/QueryRunnerTest.java
b/pinot-query-runtime/src/test/java/org/apache/pinot/query/runtime/queries/QueryRunnerTest.java
index b0d0eb3c404..951d7ae15ed 100644
---
a/pinot-query-runtime/src/test/java/org/apache/pinot/query/runtime/queries/QueryRunnerTest.java
+++
b/pinot-query-runtime/src/test/java/org/apache/pinot/query/runtime/queries/QueryRunnerTest.java
@@ -27,6 +27,7 @@ import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.concurrent.TimeUnit;
+import javax.annotation.Nullable;
import org.apache.pinot.common.response.broker.ResultTable;
import org.apache.pinot.query.QueryEnvironmentTestBase;
import org.apache.pinot.query.QueryServerEnclosure;
@@ -203,6 +204,134 @@ public class QueryRunnerTest extends QueryRunnerTestBase {
return checked;
}
+ /// Runs a shuffling query over the two-server setup and checks the
per-worker stats reported by every stage.
+ /// This is the only place these stats are exercised end to end, through
real multi-worker stages and the
+ /// cross-server merge of their stat maps.
+ @Test
+ public void testPerWorkerStats() {
+ ObjectNode statsTree = statsTreeOf("SELECT col1, COUNT(*) FROM a GROUP BY
col1");
+
+ // Idle workers are reported rather than active ones, so a query where
every worker did something must not
+ // report any: their absence is the healthy signal.
+ Assert.assertNull(findFieldOwner(statsTree, "nonActiveWorkers"),
+ "expected no idle worker in a query where every worker contributes,
got: " + statsTree);
+
+ // Assertions comparing a single worker's stats against the stage totals
collapse to identities on a
+ // single-worker stage, so require a stage that actually ran on several
workers, otherwise this test would keep
+ // passing if the cross-worker merge broke.
+ int multiWorkerSends = assertSendStats(statsTree);
+ Assert.assertTrue(multiWorkerSends > 0,
+ "expected a multi-worker send reporting maxEmittedRows and
maxClockTimeMs, got: " + statsTree);
+ }
+
+ /// Each operator decides for itself which workers it was idle on, so a
query whose filter matches nothing
+ /// separates them: the leaf operators were handed segments and are not
idle, while everything above them sent
+ /// and received nothing and is idle on every worker.
+ @Test
+ public void testPerWorkerStatsWhenNothingMatches() {
+ ObjectNode statsTree = statsTreeOf("SELECT col1, COUNT(*) FROM a WHERE
col1 = 'no-such-value' GROUP BY col1");
+
+ // StatMap drops zero-valued keys, so an absent field means zero
throughout.
+ ObjectNode leaf = findNodeOfType(statsTree, "LEAF");
+ Assert.assertNotNull(leaf, "expected a LEAF node in " + statsTree);
+ Assert.assertNull(leaf.get("nonActiveWorkers"),
+ "expected workers with segments assigned not to be idle even though
they emitted nothing: " + leaf);
+
+ ObjectNode leafStageSend = findSendAboveLeaf(statsTree);
+ Assert.assertNotNull(leafStageSend, "expected a leaf stage in " +
statsTree);
+ Assert.assertEquals(leafStageSend.path("emittedRows").asLong(0), 0L,
"expected the leaf stage to send nothing");
+ Assert.assertEquals(leafStageSend.path("nonActiveWorkers").asLong(0),
+ leafStageSend.path("parallelism").asLong(0),
+ "expected every worker of a send that sent nothing to be idle: " +
leafStageSend);
+
+ ObjectNode receive = findNodeOfType(statsTree, "MAILBOX_RECEIVE");
+ Assert.assertNotNull(receive, "expected a MAILBOX_RECEIVE node in " +
statsTree);
+ Assert.assertEquals(receive.path("nonActiveWorkers").asLong(0),
receive.path("parallelism").asLong(0),
+ "expected every worker of a receive that got no row to be idle: " +
receive);
+ }
+
+ private ObjectNode statsTreeOf(@Language("sql") String sql) {
+ QueryDispatcher.QueryResult queryResult = queryRunner(sql, true);
+ Map<Integer, DispatchablePlanFragment> planNodes =
planQuery(sql).getQueryPlan().getQueryStageMap();
+ return new MultiStageStatsTreeBuilder(planNodes,
queryResult.getQueryStats()).jsonStatsByStage(1);
+ }
+
+ @Nullable
+ private static ObjectNode findNodeOfType(JsonNode node, String type) {
+ if (type.equals(node.path("type").asText())) {
+ return (ObjectNode) node;
+ }
+ for (JsonNode child : node.path("children")) {
+ ObjectNode found = findNodeOfType(child, type);
+ if (found != null) {
+ return found;
+ }
+ }
+ return null;
+ }
+
+ /// Returns the first node in the tree carrying `field`, or null if none
does.
+ @Nullable
+ private static ObjectNode findFieldOwner(JsonNode node, String field) {
+ if (node.get(field) != null) {
+ return (ObjectNode) node;
+ }
+ for (JsonNode child : node.path("children")) {
+ ObjectNode found = findFieldOwner(child, field);
+ if (found != null) {
+ return found;
+ }
+ }
+ return null;
+ }
+
+ /// Returns the MAILBOX_SEND node of the leaf stage, that is, the one
holding the LEAF operator.
+ @Nullable
+ private static ObjectNode findSendAboveLeaf(JsonNode node) {
+ for (JsonNode child : node.path("children")) {
+ if ("MAILBOX_SEND".equals(node.path("type").asText()) &&
"LEAF".equals(child.path("type").asText())) {
+ return (ObjectNode) node;
+ }
+ ObjectNode found = findSendAboveLeaf(child);
+ if (found != null) {
+ return found;
+ }
+ }
+ return null;
+ }
+
+ /// Asserts the per-worker invariants on every send node reporting them, and
returns how many of those ran on
+ /// more than one worker.
+ private static int assertSendStats(JsonNode node) {
+ int multiWorker = 0;
+ JsonNode maxEmittedRows = node.get("maxEmittedRows");
+ if (maxEmittedRows != null) {
+ long max = maxEmittedRows.asLong();
+ long emitted = node.path("emittedRows").asLong(0);
+ long parallelism = node.path("parallelism").asLong(0);
+ String ctx = " for node " + node;
+
+ // Both are counts over one worker while emittedRows is the sum over all
of them, so neither can exceed it.
+ // This is what catches a merge function summing where it should take an
extremum.
+ Assert.assertTrue(max <= emitted, "maxEmittedRows " + max + " exceeds
emittedRows " + emitted + ctx);
+
+ long maxClockTimeMs = node.path("maxClockTimeMs").asLong(0);
+ long executionTimeMs = node.path("executionTimeMs").asLong(0);
+ Assert.assertTrue(maxClockTimeMs <= executionTimeMs,
+ "maxClockTimeMs " + maxClockTimeMs + " exceeds the summed
executionTimeMs " + executionTimeMs + ctx);
+ Assert.assertTrue(maxClockTimeMs >= node.path("clockTimeMs").asLong(0),
+ "maxClockTimeMs " + maxClockTimeMs + " is below the average
clockTimeMs" + ctx);
+
+ if (parallelism > 1) {
+ multiWorker++;
+ }
+ }
+ for (JsonNode child : node.path("children")) {
+ multiWorker += assertSendStats(child);
+ }
+ return multiWorker;
+ }
+
/// Test compares with expected row count only.
@Test(dataProvider = "testDataWithSqlToFinalRowCount")
public void testSqlWithFinalRowCountChecker(String sql, int expectedRows) {
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
