This is an automated email from the ASF dual-hosted git repository.
Jackie-Jiang 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 274601b020e Add numRemoteAccesses and remoteAccessBytes query stats
(#19756)
274601b020e is described below
commit 274601b020eda36e28c917a3b94728ca37733bce
Author: Xiaotian (Jackie) Jiang <[email protected]>
AuthorDate: Tue Oct 6 17:10:44 2026 -0700
Add numRemoteAccesses and remoteAccessBytes query stats (#19756)
---
.../apache/pinot/common/datatable/DataTable.java | 8 +++-
.../pinot/common/response/BrokerResponse.java | 8 ++++
.../response/broker/BrokerResponseNative.java | 45 +++++++++++++++-------
.../response/broker/BrokerResponseNativeV2.java | 31 ++++++++++-----
.../response/broker/CursorResponseNative.java | 10 +++--
.../response/mapper/TimeSeriesResponseMapper.java | 4 ++
.../core/operator/InstanceResponseOperator.java | 10 +++++
.../query/reduce/ExecutionStatsAggregator.java | 8 ++++
.../operator/InstanceResponseOperatorTest.java | 36 +++++++++++++++++
.../query/reduce/ExecutionStatsAggregatorTest.java | 24 ++++++++++++
.../pinot/query/runtime/operator/LeafOperator.java | 12 +++++-
.../TimeSeriesExchangeReceiveOperator.java | 4 +-
.../query/runtime/operator/LeafOperatorTest.java | 33 ++++++++++++++++
.../org/apache/pinot/segment/spi/FetchContext.java | 23 +++++++++++
14 files changed, 224 insertions(+), 32 deletions(-)
diff --git
a/pinot-common/src/main/java/org/apache/pinot/common/datatable/DataTable.java
b/pinot-common/src/main/java/org/apache/pinot/common/datatable/DataTable.java
index d3b3a79dbaa..1187448c81b 100644
---
a/pinot-common/src/main/java/org/apache/pinot/common/datatable/DataTable.java
+++
b/pinot-common/src/main/java/org/apache/pinot/common/datatable/DataTable.java
@@ -160,11 +160,15 @@ public interface DataTable {
// inputs. How a downstream consumer reacts (skip, retry, accept with
annotation) is the
// consumer's policy.
INCOMPLETE_MERGE(43, "incompleteMerge", MetadataValueType.STRING),
- LITE_MODE_LEAF_STAGE_LIMIT_REACHED(44, "liteModeLeafStageLimitReached",
MetadataValueType.STRING);
+ LITE_MODE_LEAF_STAGE_LIMIT_REACHED(44, "liteModeLeafStageLimitReached",
MetadataValueType.STRING),
+ // Number of remote accesses (local cache misses) made to read segment
data from a remote storage tier, and the
+ // number of bytes they read.
+ NUM_REMOTE_ACCESSES(45, "numRemoteAccesses", MetadataValueType.LONG),
+ REMOTE_ACCESS_BYTES(46, "remoteAccessBytes", MetadataValueType.LONG);
// We keep this constant to track the max id added so far for backward
compatibility.
// Increase it when adding new keys, but NEVER DECREASE IT!!!
- private static final int MAX_ID =
LITE_MODE_LEAF_STAGE_LIMIT_REACHED.getId();
+ private static final int MAX_ID = REMOTE_ACCESS_BYTES.getId();
private static final MetadataKey[] ID_TO_ENUM_KEY_MAP = new
MetadataKey[MAX_ID + 1];
private static final Map<String, MetadataKey> NAME_TO_ENUM_KEY_MAP = new
HashMap<>();
diff --git
a/pinot-common/src/main/java/org/apache/pinot/common/response/BrokerResponse.java
b/pinot-common/src/main/java/org/apache/pinot/common/response/BrokerResponse.java
index 98362b1b4e2..cd9bc76f160 100644
---
a/pinot-common/src/main/java/org/apache/pinot/common/response/BrokerResponse.java
+++
b/pinot-common/src/main/java/org/apache/pinot/common/response/BrokerResponse.java
@@ -221,6 +221,14 @@ public interface BrokerResponse {
/// This value is always lower or equal than
[#getNumSegmentsPrunedByServer()]
long getNumSegmentsPrunedByValue();
+ /// Returns the number of remote accesses (local cache misses) made to read
segment data from a remote storage tier.
+ ///
+ /// This value is always `0` when the segments are not backed by a remote
storage tier.
+ long getNumRemoteAccesses();
+
+ /// Returns the number of bytes read by the remote accesses (see
[#getNumRemoteAccesses()]).
+ long getRemoteAccessBytes();
+
/// Returns the time used to reduce the server responses into the final
response in milliseconds.
long getBrokerReduceTimeMs();
diff --git
a/pinot-common/src/main/java/org/apache/pinot/common/response/broker/BrokerResponseNative.java
b/pinot-common/src/main/java/org/apache/pinot/common/response/broker/BrokerResponseNative.java
index b38bae56605..81749d6002f 100644
---
a/pinot-common/src/main/java/org/apache/pinot/common/response/broker/BrokerResponseNative.java
+++
b/pinot-common/src/main/java/org/apache/pinot/common/response/broker/BrokerResponseNative.java
@@ -42,23 +42,20 @@ import org.apache.pinot.spi.utils.JsonUtils;
/// This class can be used to serialize/deserialize the broker response.
@JsonPropertyOrder({
"resultTable", "numRowsResultSet", "partialResult", "exceptions",
"numGroupsLimitReached",
- "numGroupsWarningLimitReached", "maxRowsInDistinctReached",
- "maxRowsWithoutChangeInDistinctReached",
- "maxExecutionTimeInDistinctReached", "timeUsedMs",
- "requestId", "clientRequestId", "brokerId", "numDocsScanned", "totalDocs",
- "numEntriesScannedInFilter",
- "numEntriesScannedPostFilter", "numServersQueried", "numServersResponded",
"numSegmentsQueried",
- "numSegmentsProcessed", "numSegmentsMatched",
"numConsumingSegmentsQueried", "numConsumingSegmentsProcessed",
- "numConsumingSegmentsMatched", "minConsumingFreshnessTimeMs",
"numSegmentsPrunedByBroker",
- "numSegmentsPrunedByServer", "numSegmentsPrunedInvalid",
"numSegmentsPrunedByLimit", "numSegmentsPrunedByValue",
- "brokerReduceTimeMs", "offlineThreadCpuTimeNs", "realtimeThreadCpuTimeNs",
"offlineSystemActivitiesCpuTimeNs",
- "realtimeSystemActivitiesCpuTimeNs",
"offlineResponseSerializationCpuTimeNs",
+ "numGroupsWarningLimitReached", "maxRowsInDistinctReached",
"maxRowsWithoutChangeInDistinctReached",
+ "maxExecutionTimeInDistinctReached", "timeUsedMs", "requestId",
"clientRequestId", "brokerId", "numDocsScanned",
+ "totalDocs", "numEntriesScannedInFilter", "numEntriesScannedPostFilter",
"numServersQueried",
+ "numServersResponded", "numSegmentsQueried", "numSegmentsProcessed",
"numSegmentsMatched",
+ "numConsumingSegmentsQueried", "numConsumingSegmentsProcessed",
"numConsumingSegmentsMatched",
+ "minConsumingFreshnessTimeMs", "numSegmentsPrunedByBroker",
"numSegmentsPrunedByServer",
+ "numSegmentsPrunedInvalid", "numSegmentsPrunedByLimit",
"numSegmentsPrunedByValue", "numRemoteAccesses",
+ "remoteAccessBytes", "brokerReduceTimeMs", "offlineThreadCpuTimeNs",
"realtimeThreadCpuTimeNs",
+ "offlineSystemActivitiesCpuTimeNs", "realtimeSystemActivitiesCpuTimeNs",
"offlineResponseSerializationCpuTimeNs",
"realtimeResponseSerializationCpuTimeNs", "offlineTotalCpuTimeNs",
"realtimeTotalCpuTimeNs",
"explainPlanNumEmptyFilterSegments",
"explainPlanNumMatchAllFilterSegments", "traceInfo", "tablesQueried",
"offlineThreadMemAllocatedBytes", "realtimeThreadMemAllocatedBytes",
"offlineResponseSerMemAllocatedBytes",
- "realtimeResponseSerMemAllocatedBytes", "offlineTotalMemAllocatedBytes",
"realtimeTotalMemAllocatedBytes",
- "pools", "rlsFiltersApplied", "approximateFunctionApplied",
"groupsTrimmed", "materializedViewQueried",
- "serverStats"
+ "realtimeResponseSerMemAllocatedBytes", "offlineTotalMemAllocatedBytes",
"realtimeTotalMemAllocatedBytes", "pools",
+ "rlsFiltersApplied", "approximateFunctionApplied", "groupsTrimmed",
"materializedViewQueried", "serverStats"
})
@JsonIgnoreProperties(ignoreUnknown = true)
public class BrokerResponseNative implements BrokerResponse {
@@ -102,6 +99,8 @@ public class BrokerResponseNative implements BrokerResponse {
private long _numSegmentsPrunedInvalid = 0L;
private long _numSegmentsPrunedByLimit = 0L;
private long _numSegmentsPrunedByValue = 0L;
+ private long _numRemoteAccesses = 0L;
+ private long _remoteAccessBytes = 0L;
private long _brokerReduceTimeMs = 0L;
private long _offlineThreadCpuTimeNs = 0L;
private long _realtimeThreadCpuTimeNs = 0L;
@@ -481,6 +480,24 @@ public class BrokerResponseNative implements
BrokerResponse {
_numSegmentsPrunedByValue = numSegmentsPrunedByValue;
}
+ @Override
+ public long getNumRemoteAccesses() {
+ return _numRemoteAccesses;
+ }
+
+ public void setNumRemoteAccesses(long numRemoteAccesses) {
+ _numRemoteAccesses = numRemoteAccesses;
+ }
+
+ @Override
+ public long getRemoteAccessBytes() {
+ return _remoteAccessBytes;
+ }
+
+ public void setRemoteAccessBytes(long remoteAccessBytes) {
+ _remoteAccessBytes = remoteAccessBytes;
+ }
+
@Override
public long getBrokerReduceTimeMs() {
return _brokerReduceTimeMs;
diff --git
a/pinot-common/src/main/java/org/apache/pinot/common/response/broker/BrokerResponseNativeV2.java
b/pinot-common/src/main/java/org/apache/pinot/common/response/broker/BrokerResponseNativeV2.java
index 9919057b565..c9657f39ea6 100644
---
a/pinot-common/src/main/java/org/apache/pinot/common/response/broker/BrokerResponseNativeV2.java
+++
b/pinot-common/src/main/java/org/apache/pinot/common/response/broker/BrokerResponseNativeV2.java
@@ -40,23 +40,22 @@ import org.apache.pinot.common.response.ProcessingException;
/// TODO: Currently this class cannot be used to deserialize the JSON response.
@JsonPropertyOrder({
"resultTable", "numRowsResultSet", "partialResult", "exceptions",
"numGroupsLimitReached",
- "numGroupsWarningLimitReached", "numGroups", "earlyTerminationReasons",
- "maxRowsInJoinReached",
- "maxRowsInJoin", "maxRowsInWindowReached", "maxRowsInWindow",
"timeUsedMs", "stageStats", "streamStatsCoverage",
+ "numGroupsWarningLimitReached", "numGroups", "earlyTerminationReasons",
"maxRowsInJoinReached", "maxRowsInJoin",
+ "maxRowsInWindowReached", "maxRowsInWindow", "timeUsedMs", "stageStats",
"streamStatsCoverage",
"maxRowsInOperator", "requestId", "clientRequestId", "brokerId",
"numDocsScanned", "totalDocs",
"numEntriesScannedInFilter", "numEntriesScannedPostFilter",
"numServersQueried", "numServersResponded",
"numSegmentsQueried", "numSegmentsProcessed", "numSegmentsMatched",
"numConsumingSegmentsQueried",
"numConsumingSegmentsProcessed", "numConsumingSegmentsMatched",
"minConsumingFreshnessTimeMs",
"numSegmentsPrunedByBroker", "numSegmentsPrunedByServer",
"numSegmentsPrunedInvalid", "numSegmentsPrunedByLimit",
- "numSegmentsPrunedByValue", "brokerReduceTimeMs",
"offlineThreadCpuTimeNs", "realtimeThreadCpuTimeNs",
- "offlineSystemActivitiesCpuTimeNs", "realtimeSystemActivitiesCpuTimeNs",
"offlineResponseSerializationCpuTimeNs",
+ "numSegmentsPrunedByValue", "numRemoteAccesses", "remoteAccessBytes",
"brokerReduceTimeMs",
+ "offlineThreadCpuTimeNs", "realtimeThreadCpuTimeNs",
"offlineSystemActivitiesCpuTimeNs",
+ "realtimeSystemActivitiesCpuTimeNs",
"offlineResponseSerializationCpuTimeNs",
"realtimeResponseSerializationCpuTimeNs", "offlineTotalCpuTimeNs",
"realtimeTotalCpuTimeNs",
"explainPlanNumEmptyFilterSegments",
"explainPlanNumMatchAllFilterSegments", "traceInfo", "tablesQueried",
"offlineThreadMemAllocatedBytes", "realtimeThreadMemAllocatedBytes",
"offlineResponseSerMemAllocatedBytes",
- "realtimeResponseSerMemAllocatedBytes", "offlineTotalMemAllocatedBytes",
"realtimeTotalMemAllocatedBytes",
- "pools", "rlsFiltersApplied", "approximateFunctionApplied",
"groupsTrimmed",
- "mseLiteLeafStageLimitReached", "mseLiteLeafStageEffectiveLimit",
"mseLiteFanOutAdjustedLimitApplied",
- "responseMetadata"
+ "realtimeResponseSerMemAllocatedBytes", "offlineTotalMemAllocatedBytes",
"realtimeTotalMemAllocatedBytes", "pools",
+ "rlsFiltersApplied", "approximateFunctionApplied", "groupsTrimmed",
"mseLiteLeafStageLimitReached",
+ "mseLiteLeafStageEffectiveLimit", "mseLiteFanOutAdjustedLimitApplied",
"responseMetadata"
})
public class BrokerResponseNativeV2 implements BrokerResponse {
private final StatMap<StatKey> _brokerStats = new StatMap<>(StatKey.class);
@@ -410,6 +409,16 @@ public class BrokerResponseNativeV2 implements
BrokerResponse {
return _brokerStats.getLong(StatKey.NUM_SEGMENTS_PRUNED_BY_VALUE);
}
+ @Override
+ public long getNumRemoteAccesses() {
+ return _brokerStats.getLong(StatKey.NUM_REMOTE_ACCESSES);
+ }
+
+ @Override
+ public long getRemoteAccessBytes() {
+ return _brokerStats.getLong(StatKey.REMOTE_ACCESS_BYTES);
+ }
+
@Override
public long getBrokerReduceTimeMs() {
return _brokerReduceTimeMs;
@@ -571,7 +580,9 @@ public class BrokerResponseNativeV2 implements
BrokerResponse {
}
},
EARLY_TERMINATION_REASONS(StatMap.Type.STRING_SET),
- LITE_MODE_LEAF_STAGE_LIMIT_REACHED(StatMap.Type.BOOLEAN);
+ LITE_MODE_LEAF_STAGE_LIMIT_REACHED(StatMap.Type.BOOLEAN),
+ NUM_REMOTE_ACCESSES(StatMap.Type.LONG),
+ REMOTE_ACCESS_BYTES(StatMap.Type.LONG);
private final StatMap.Type _type;
diff --git
a/pinot-common/src/main/java/org/apache/pinot/common/response/broker/CursorResponseNative.java
b/pinot-common/src/main/java/org/apache/pinot/common/response/broker/CursorResponseNative.java
index 96565b68cad..d88d30f92e0 100644
---
a/pinot-common/src/main/java/org/apache/pinot/common/response/broker/CursorResponseNative.java
+++
b/pinot-common/src/main/java/org/apache/pinot/common/response/broker/CursorResponseNative.java
@@ -25,14 +25,14 @@ import org.apache.pinot.common.response.CursorResponse;
@JsonPropertyOrder({
"resultTable", "numRowsResultSet", "partialResult", "exceptions",
"numGroupsLimitReached",
- "numGroupsWarningLimitReached", "timeUsedMs", "requestId", "brokerId",
- "numDocsScanned", "totalDocs",
+ "numGroupsWarningLimitReached", "timeUsedMs", "requestId", "brokerId",
"numDocsScanned", "totalDocs",
"numEntriesScannedInFilter", "numEntriesScannedPostFilter",
"numServersQueried", "numServersResponded",
"numSegmentsQueried", "numSegmentsProcessed", "numSegmentsMatched",
"numConsumingSegmentsQueried",
"numConsumingSegmentsProcessed", "numConsumingSegmentsMatched",
"minConsumingFreshnessTimeMs",
"numSegmentsPrunedByBroker", "numSegmentsPrunedByServer",
"numSegmentsPrunedInvalid", "numSegmentsPrunedByLimit",
- "numSegmentsPrunedByValue", "brokerReduceTimeMs",
"offlineThreadCpuTimeNs", "realtimeThreadCpuTimeNs",
- "offlineSystemActivitiesCpuTimeNs", "realtimeSystemActivitiesCpuTimeNs",
"offlineResponseSerializationCpuTimeNs",
+ "numSegmentsPrunedByValue", "numRemoteAccesses", "remoteAccessBytes",
"brokerReduceTimeMs",
+ "offlineThreadCpuTimeNs", "realtimeThreadCpuTimeNs",
"offlineSystemActivitiesCpuTimeNs",
+ "realtimeSystemActivitiesCpuTimeNs",
"offlineResponseSerializationCpuTimeNs",
"realtimeResponseSerializationCpuTimeNs", "offlineTotalCpuTimeNs",
"realtimeTotalCpuTimeNs",
"explainPlanNumEmptyFilterSegments",
"explainPlanNumMatchAllFilterSegments", "traceInfo", "tableQueries",
"groupsTrimmed", "approximateFunctionApplied",
@@ -83,6 +83,8 @@ public class CursorResponseNative extends
BrokerResponseNative implements Cursor
setNumSegmentsPrunedInvalid(response.getNumSegmentsPrunedInvalid());
setNumSegmentsPrunedByLimit(response.getNumSegmentsPrunedByLimit());
setNumSegmentsPrunedByValue(response.getNumSegmentsPrunedByValue());
+ setNumRemoteAccesses(response.getNumRemoteAccesses());
+ setRemoteAccessBytes(response.getRemoteAccessBytes());
setBrokerReduceTimeMs(response.getBrokerReduceTimeMs());
setOfflineThreadCpuTimeNs(response.getOfflineThreadCpuTimeNs());
setRealtimeThreadCpuTimeNs(response.getRealtimeThreadCpuTimeNs());
diff --git
a/pinot-common/src/main/java/org/apache/pinot/common/response/mapper/TimeSeriesResponseMapper.java
b/pinot-common/src/main/java/org/apache/pinot/common/response/mapper/TimeSeriesResponseMapper.java
index 375e1979319..1d71932777b 100644
---
a/pinot-common/src/main/java/org/apache/pinot/common/response/mapper/TimeSeriesResponseMapper.java
+++
b/pinot-common/src/main/java/org/apache/pinot/common/response/mapper/TimeSeriesResponseMapper.java
@@ -166,6 +166,10 @@ public class TimeSeriesResponseMapper {
getIntMetadataValue(metadata,
DataTable.MetadataKey.NUM_SEGMENTS_PRUNED_BY_LIMIT));
map.merge(BrokerResponseNativeV2.StatKey.NUM_SEGMENTS_PRUNED_BY_VALUE,
getIntMetadataValue(metadata,
DataTable.MetadataKey.NUM_SEGMENTS_PRUNED_BY_VALUE));
+ map.merge(BrokerResponseNativeV2.StatKey.NUM_REMOTE_ACCESSES,
+ getLongMetadataValue(metadata,
DataTable.MetadataKey.NUM_REMOTE_ACCESSES));
+ map.merge(BrokerResponseNativeV2.StatKey.REMOTE_ACCESS_BYTES,
+ getLongMetadataValue(metadata,
DataTable.MetadataKey.REMOTE_ACCESS_BYTES));
brokerResponse.addBrokerStats(map);
brokerResponse.setTimeUsedMs(getLongMetadataValue(metadata,
DataTable.MetadataKey.TIME_USED_MS));
}
diff --git
a/pinot-core/src/main/java/org/apache/pinot/core/operator/InstanceResponseOperator.java
b/pinot-core/src/main/java/org/apache/pinot/core/operator/InstanceResponseOperator.java
index 031b5214496..5d63ac36cdd 100644
---
a/pinot-core/src/main/java/org/apache/pinot/core/operator/InstanceResponseOperator.java
+++
b/pinot-core/src/main/java/org/apache/pinot/core/operator/InstanceResponseOperator.java
@@ -99,6 +99,16 @@ public class InstanceResponseOperator extends
BaseOperator<InstanceResponseBlock
String.valueOf(_threadMemAllocatedBytes));
instanceResponseBlock.addMetadata(MetadataKey.SYSTEM_ACTIVITIES_CPU_TIME_NS.getName(),
String.valueOf(_systemActivitiesCpuTimeNs));
+ long numRemoteAccesses = 0;
+ long remoteAccessBytes = 0;
+ for (FetchContext fetchContext : _fetchContexts) {
+ numRemoteAccesses += fetchContext.getNumRemoteAccesses();
+ remoteAccessBytes += fetchContext.getRemoteAccessBytes();
+ }
+ if (numRemoteAccesses > 0) {
+
instanceResponseBlock.addMetadata(MetadataKey.NUM_REMOTE_ACCESSES.getName(),
Long.toString(numRemoteAccesses));
+
instanceResponseBlock.addMetadata(MetadataKey.REMOTE_ACCESS_BYTES.getName(),
Long.toString(remoteAccessBytes));
+ }
Integer implicitLimit =
QueryOptionsUtils.getLiteModeImplicitLeafStageLimit(
_queryContext.getQueryOptions());
// false-positive when table has exactly implicitLimit rows
diff --git
a/pinot-core/src/main/java/org/apache/pinot/core/query/reduce/ExecutionStatsAggregator.java
b/pinot-core/src/main/java/org/apache/pinot/core/query/reduce/ExecutionStatsAggregator.java
index 643b9fe606c..15cc51ef7b9 100644
---
a/pinot-core/src/main/java/org/apache/pinot/core/query/reduce/ExecutionStatsAggregator.java
+++
b/pinot-core/src/main/java/org/apache/pinot/core/query/reduce/ExecutionStatsAggregator.java
@@ -71,6 +71,8 @@ public class ExecutionStatsAggregator {
private long _numSegmentsPrunedInvalid = 0L;
private long _numSegmentsPrunedByLimit = 0L;
private long _numSegmentsPrunedByValue = 0L;
+ private long _numRemoteAccesses = 0L;
+ private long _remoteAccessBytes = 0L;
private long _explainPlanNumEmptyFilterSegments = 0L;
private long _explainPlanNumMatchAllFilterSegments = 0L;
private boolean _groupsTrimmed = false;
@@ -217,6 +219,8 @@ public class ExecutionStatsAggregator {
l -> _numSegmentsPrunedByLimit += l);
withNotNullLongMetadata(metadata,
DataTable.MetadataKey.NUM_SEGMENTS_PRUNED_BY_VALUE,
l -> _numSegmentsPrunedByValue += l);
+ withNotNullLongMetadata(metadata,
DataTable.MetadataKey.NUM_REMOTE_ACCESSES, l -> _numRemoteAccesses += l);
+ withNotNullLongMetadata(metadata,
DataTable.MetadataKey.REMOTE_ACCESS_BYTES, l -> _remoteAccessBytes += l);
String explainPlanNumEmptyFilterSegments =
metadata.get(DataTable.MetadataKey.EXPLAIN_PLAN_NUM_EMPTY_FILTER_SEGMENTS.getName());
@@ -303,6 +307,8 @@ public class ExecutionStatsAggregator {
brokerResponseNative.setNumSegmentsPrunedInvalid(_numSegmentsPrunedInvalid);
brokerResponseNative.setNumSegmentsPrunedByLimit(_numSegmentsPrunedByLimit);
brokerResponseNative.setNumSegmentsPrunedByValue(_numSegmentsPrunedByValue);
+ brokerResponseNative.setNumRemoteAccesses(_numRemoteAccesses);
+ brokerResponseNative.setRemoteAccessBytes(_remoteAccessBytes);
brokerResponseNative.setExplainPlanNumEmptyFilterSegments(_explainPlanNumEmptyFilterSegments);
brokerResponseNative.setExplainPlanNumMatchAllFilterSegments(_explainPlanNumMatchAllFilterSegments);
if (_numConsumingSegmentsQueried > 0) {
@@ -400,6 +406,8 @@ public class ExecutionStatsAggregator {
putLong(metadata, DataTable.MetadataKey.NUM_SEGMENTS_PRUNED_INVALID,
_numSegmentsPrunedInvalid);
putLong(metadata, DataTable.MetadataKey.NUM_SEGMENTS_PRUNED_BY_LIMIT,
_numSegmentsPrunedByLimit);
putLong(metadata, DataTable.MetadataKey.NUM_SEGMENTS_PRUNED_BY_VALUE,
_numSegmentsPrunedByValue);
+ putLong(metadata, DataTable.MetadataKey.NUM_REMOTE_ACCESSES,
_numRemoteAccesses);
+ putLong(metadata, DataTable.MetadataKey.REMOTE_ACCESS_BYTES,
_remoteAccessBytes);
putLong(metadata,
DataTable.MetadataKey.EXPLAIN_PLAN_NUM_EMPTY_FILTER_SEGMENTS,
_explainPlanNumEmptyFilterSegments);
putLong(metadata,
DataTable.MetadataKey.EXPLAIN_PLAN_NUM_MATCH_ALL_FILTER_SEGMENTS,
diff --git
a/pinot-core/src/test/java/org/apache/pinot/core/operator/InstanceResponseOperatorTest.java
b/pinot-core/src/test/java/org/apache/pinot/core/operator/InstanceResponseOperatorTest.java
index 92a2e1194da..ac700099b3b 100644
---
a/pinot-core/src/test/java/org/apache/pinot/core/operator/InstanceResponseOperatorTest.java
+++
b/pinot-core/src/test/java/org/apache/pinot/core/operator/InstanceResponseOperatorTest.java
@@ -21,11 +21,14 @@ package org.apache.pinot.core.operator;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
+import java.util.Set;
+import java.util.UUID;
import org.apache.pinot.common.datatable.DataTable.MetadataKey;
import org.apache.pinot.core.operator.blocks.InstanceResponseBlock;
import org.apache.pinot.core.operator.blocks.results.SelectionResultsBlock;
import org.apache.pinot.core.operator.combine.BaseCombineOperator;
import org.apache.pinot.core.query.request.context.QueryContext;
+import org.apache.pinot.segment.spi.FetchContext;
import
org.apache.pinot.spi.utils.CommonConstants.Broker.Request.QueryOptionKey;
import org.testng.annotations.Test;
@@ -124,4 +127,37 @@ public class InstanceResponseOperatorTest {
assertTrue(responseBlock.getResponseMetadata()
.containsKey(MetadataKey.LITE_MODE_LEAF_STAGE_LIMIT_REACHED.getName()));
}
+
+ @Test
+ public void testRemoteAccessesSummedAcrossFetchContexts() {
+ SelectionResultsBlock resultsBlock = mock(SelectionResultsBlock.class);
+ when(resultsBlock.getResultsMetadata()).thenReturn(new HashMap<>());
+ QueryContext queryContext = mock(QueryContext.class);
+ when(queryContext.getQueryOptions()).thenReturn(new HashMap<>());
+ FetchContext fetchContext1 = new FetchContext(UUID.randomUUID(),
"segment1", Set.of("col"));
+ fetchContext1.addRemoteAccesses(3, 300);
+ FetchContext fetchContext2 = new FetchContext(UUID.randomUUID(),
"segment2", Set.of("col"));
+ fetchContext2.addRemoteAccesses(4, 400);
+ InstanceResponseOperator operator = new
InstanceResponseOperator(mock(BaseCombineOperator.class), List.of(),
+ List.of(fetchContext1, fetchContext2), queryContext);
+ InstanceResponseBlock responseBlock =
operator.buildInstanceResponseBlock(resultsBlock);
+ Map<String, String> metadata = responseBlock.getResponseMetadata();
+ assertEquals(metadata.get(MetadataKey.NUM_REMOTE_ACCESSES.getName()), "7");
+ assertEquals(metadata.get(MetadataKey.REMOTE_ACCESS_BYTES.getName()),
"700");
+ }
+
+ @Test
+ public void testNoRemoteAccessMetadataWithoutRemoteAccesses() {
+ SelectionResultsBlock resultsBlock = mock(SelectionResultsBlock.class);
+ when(resultsBlock.getResultsMetadata()).thenReturn(new HashMap<>());
+ QueryContext queryContext = mock(QueryContext.class);
+ when(queryContext.getQueryOptions()).thenReturn(new HashMap<>());
+ FetchContext fetchContext = new FetchContext(UUID.randomUUID(), "segment",
Set.of("col"));
+ InstanceResponseOperator operator = new
InstanceResponseOperator(mock(BaseCombineOperator.class), List.of(),
+ List.of(fetchContext), queryContext);
+ InstanceResponseBlock responseBlock =
operator.buildInstanceResponseBlock(resultsBlock);
+ Map<String, String> metadata = responseBlock.getResponseMetadata();
+
assertFalse(metadata.containsKey(MetadataKey.NUM_REMOTE_ACCESSES.getName()));
+
assertFalse(metadata.containsKey(MetadataKey.REMOTE_ACCESS_BYTES.getName()));
+ }
}
diff --git
a/pinot-core/src/test/java/org/apache/pinot/core/query/reduce/ExecutionStatsAggregatorTest.java
b/pinot-core/src/test/java/org/apache/pinot/core/query/reduce/ExecutionStatsAggregatorTest.java
index 0759fc13d75..e351288ac43 100644
---
a/pinot-core/src/test/java/org/apache/pinot/core/query/reduce/ExecutionStatsAggregatorTest.java
+++
b/pinot-core/src/test/java/org/apache/pinot/core/query/reduce/ExecutionStatsAggregatorTest.java
@@ -29,6 +29,7 @@ import org.apache.pinot.spi.config.table.TableType;
import org.testng.annotations.Test;
import static org.mockito.Mockito.mock;
+import static org.testng.Assert.assertEquals;
import static org.testng.Assert.assertFalse;
import static org.testng.Assert.assertNull;
import static org.testng.Assert.assertTrue;
@@ -165,4 +166,27 @@ public class ExecutionStatsAggregatorTest {
assertFalse(flags[1]);
assertFalse(flags[2]);
}
+
+ @Test
+ public void testRemoteAccessesSumAcrossServersAndRoundTrip() {
+ DataTable table1 = emptyServerTable();
+ table1.getMetadata().put(MetadataKey.NUM_REMOTE_ACCESSES.getName(), "3");
+ table1.getMetadata().put(MetadataKey.REMOTE_ACCESS_BYTES.getName(), "300");
+ DataTable table2 = emptyServerTable();
+ table2.getMetadata().put(MetadataKey.NUM_REMOTE_ACCESSES.getName(), "4");
+ table2.getMetadata().put(MetadataKey.REMOTE_ACCESS_BYTES.getName(), "400");
+ ExecutionStatsAggregator aggregator = new ExecutionStatsAggregator(false);
+ aggregator.aggregate(SERVER, table1);
+ aggregator.aggregate(SERVER, table2);
+ aggregator.aggregate(SERVER, emptyServerTable());
+ BrokerResponseNative response = new BrokerResponseNative();
+ aggregator.setStats("testTable", response, mock(BrokerMetrics.class));
+ assertEquals(response.getNumRemoteAccesses(), 7L);
+ assertEquals(response.getRemoteAccessBytes(), 700L);
+
+ DataTable merged = emptyServerTable();
+ aggregator.setStatsOnMergedDataTable(merged);
+
assertEquals(merged.getMetadata().get(MetadataKey.NUM_REMOTE_ACCESSES.getName()),
"7");
+
assertEquals(merged.getMetadata().get(MetadataKey.REMOTE_ACCESS_BYTES.getName()),
"700");
+ }
}
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 dff4f49b05d..30290091d27 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
@@ -436,6 +436,12 @@ public class LeafOperator extends MultiStageOperator
implements ExplainableOpera
case NUM_SEGMENTS_PRUNED_BY_VALUE:
_statMap.merge(StatKey.NUM_SEGMENTS_PRUNED_BY_VALUE,
Integer.parseInt(entry.getValue()));
break;
+ case NUM_REMOTE_ACCESSES:
+ _statMap.merge(StatKey.NUM_REMOTE_ACCESSES,
Long.parseLong(entry.getValue()));
+ break;
+ case REMOTE_ACCESS_BYTES:
+ _statMap.merge(StatKey.REMOTE_ACCESS_BYTES,
Long.parseLong(entry.getValue()));
+ break;
case EXPLAIN_PLAN_NUM_EMPTY_FILTER_SEGMENTS:
LOGGER.debug("Skipping empty filter segments: {}", entry.getValue());
break;
@@ -798,7 +804,11 @@ public class LeafOperator extends MultiStageOperator
implements ExplainableOpera
/// 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);
+ NON_ACTIVE_WORKERS(StatMap.Type.INT, null),
+ /// Number of remote accesses (local cache misses) made to read segment
data from a remote storage tier.
+ NUM_REMOTE_ACCESSES(StatMap.Type.LONG),
+ /// Number of bytes read by the remote accesses.
+ REMOTE_ACCESS_BYTES(StatMap.Type.LONG);
// 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
diff --git
a/pinot-query-runtime/src/main/java/org/apache/pinot/query/runtime/timeseries/TimeSeriesExchangeReceiveOperator.java
b/pinot-query-runtime/src/main/java/org/apache/pinot/query/runtime/timeseries/TimeSeriesExchangeReceiveOperator.java
index 649cc607dc2..bfede821faf 100644
---
a/pinot-query-runtime/src/main/java/org/apache/pinot/query/runtime/timeseries/TimeSeriesExchangeReceiveOperator.java
+++
b/pinot-query-runtime/src/main/java/org/apache/pinot/query/runtime/timeseries/TimeSeriesExchangeReceiveOperator.java
@@ -68,7 +68,9 @@ public class TimeSeriesExchangeReceiveOperator extends
BaseTimeSeriesOperator {
DataTable.MetadataKey.NUM_SEGMENTS_PRUNED_BY_SERVER,
DataTable.MetadataKey.NUM_SEGMENTS_PRUNED_INVALID,
DataTable.MetadataKey.NUM_SEGMENTS_PRUNED_BY_LIMIT,
- DataTable.MetadataKey.NUM_SEGMENTS_PRUNED_BY_VALUE
+ DataTable.MetadataKey.NUM_SEGMENTS_PRUNED_BY_VALUE,
+ DataTable.MetadataKey.NUM_REMOTE_ACCESSES,
+ DataTable.MetadataKey.REMOTE_ACCESS_BYTES
);
private static final List<DataTable.MetadataKey> MIN_STATS_KEYS = List.of(
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 237d7033ff0..69d5baff59a 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
@@ -546,6 +546,39 @@ public class LeafOperatorTest {
operator.close();
}
+ @Test
+ public void shouldPropagateRemoteAccessStats() {
+ // Given:
+ DataSchema schema = new DataSchema(new String[]{"intCol"},
+ new DataSchema.ColumnDataType[]{DataSchema.ColumnDataType.INT});
+ InstanceResponseBlock metadataBlock = new InstanceResponseBlock(new
MetadataResultsBlock());
+
metadataBlock.getResponseMetadata().put(DataTable.MetadataKey.NUM_REMOTE_ACCESSES.getName(),
"7");
+
metadataBlock.getResponseMetadata().put(DataTable.MetadataKey.REMOTE_ACCESS_BYTES.getName(),
"700");
+ QueryExecutor queryExecutor = mockQueryExecutor(List.of(), metadataBlock);
+ LeafOperator operator =
+ new LeafOperator(OperatorTestUtil.getTracingContext(),
mockQueryRequests(1), schema, queryExecutor,
+ _executorService);
+ _operatorRef.set(operator);
+
+ // When:
+ assertTrue(operator.nextBlock().isEos(), "Expected EOS after reading the
metadata block");
+
+ // Then:
+ StatMap<LeafOperator.StatKey> leafStats = operator.copyStatMaps();
+ assertEquals(leafStats.getLong(LeafOperator.StatKey.NUM_REMOTE_ACCESSES),
7L);
+ assertEquals(leafStats.getLong(LeafOperator.StatKey.REMOTE_ACCESS_BYTES),
700L);
+
+ BrokerResponseNativeV2 brokerResponse = new BrokerResponseNativeV2();
+ MultiStageOperator.Type.LEAF.mergeInto(brokerResponse, leafStats);
+ assertEquals(brokerResponse.getNumRemoteAccesses(), 7L);
+ assertEquals(brokerResponse.getRemoteAccessBytes(), 700L);
+ JsonNode responseJson = JsonUtils.objectToJsonNode(brokerResponse);
+ assertEquals(responseJson.path("numRemoteAccesses").asLong(), 7L);
+ assertEquals(responseJson.path("remoteAccessBytes").asLong(), 700L);
+
+ operator.close();
+ }
+
@Test
public void shouldSkipNoneEarlyTerminationReason() {
assertSkippedEarlyTerminationReason(EarlyTerminationReason.NONE.name());
diff --git
a/pinot-segment-spi/src/main/java/org/apache/pinot/segment/spi/FetchContext.java
b/pinot-segment-spi/src/main/java/org/apache/pinot/segment/spi/FetchContext.java
index c2bd5b10177..9949aa44636 100644
---
a/pinot-segment-spi/src/main/java/org/apache/pinot/segment/spi/FetchContext.java
+++
b/pinot-segment-spi/src/main/java/org/apache/pinot/segment/spi/FetchContext.java
@@ -23,14 +23,21 @@ import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.UUID;
+import java.util.concurrent.atomic.AtomicLong;
import org.apache.pinot.segment.spi.index.IndexType;
/// The context for fetching buffers of a segment during query.
+///
+/// It also accumulates the remote accesses (local cache misses) performed to
serve the fetch and the bytes they read.
+/// A segment directory backed by a remote storage tier records them here,
typically when the fetched buffers are
+/// released, so that they can be reported in the query stats.
public class FetchContext {
private final UUID _fetchId;
private final String _segmentName;
private final Map<String, List<IndexType<?, ?, ?>>> _columnToIndexList;
+ private final AtomicLong _numRemoteAccesses = new AtomicLong();
+ private final AtomicLong _remoteAccessBytes = new AtomicLong();
/// Create a new FetchRequest for this segment, to fetch all buffers of the
given columns
/// @param fetchId unique fetch id
@@ -75,4 +82,20 @@ public class FetchContext {
public boolean isEmpty() {
return _columnToIndexList.isEmpty();
}
+
+ /// Records remote accesses (local cache misses) performed to serve this
fetch and the bytes they read.
+ public void addRemoteAccesses(long numRemoteAccesses, long
remoteAccessBytes) {
+ _numRemoteAccesses.addAndGet(numRemoteAccesses);
+ _remoteAccessBytes.addAndGet(remoteAccessBytes);
+ }
+
+ /// Returns the number of remote accesses (local cache misses) recorded for
this fetch.
+ public long getNumRemoteAccesses() {
+ return _numRemoteAccesses.get();
+ }
+
+ /// Returns the number of bytes read by the remote accesses recorded for
this fetch.
+ public long getRemoteAccessBytes() {
+ return _remoteAccessBytes.get();
+ }
}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
