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 bf956b3c204 Collect remote access stats per execution and add direct
remote access stats (#19859)
bf956b3c204 is described below
commit bf956b3c204680f98982fb7e70733d19594411a5
Author: Xiaotian (Jackie) Jiang <[email protected]>
AuthorDate: Wed Oct 14 17:59:00 2026 -0700
Collect remote access stats per execution and add direct remote access
stats (#19859)
---
.../apache/pinot/common/datatable/DataTable.java | 8 ++-
.../pinot/common/response/BrokerResponse.java | 8 +++
.../response/broker/BrokerResponseNative.java | 29 +++++++-
.../response/broker/BrokerResponseNativeV2.java | 24 +++++--
.../response/broker/CursorResponseNative.java | 8 ++-
.../response/mapper/TimeSeriesResponseMapper.java | 4 ++
.../response/broker/BrokerResponseNativeTest.java | 80 ++++++++++++++-------
.../broker/BrokerResponseNativeV2Test.java | 30 ++++++++
.../core/operator/InstanceResponseOperator.java | 16 ++---
.../query/executor/ServerQueryExecutorV1Impl.java | 39 ++++++++++
.../query/pruner/BloomFilterSegmentPruner.java | 4 +-
.../query/reduce/ExecutionStatsAggregator.java | 10 +++
.../operator/InstanceResponseOperatorTest.java | 55 +++++++-------
.../core/query/executor/QueryExecutorTest.java | 83 ++++++++++++++++++++++
.../query/pruner/BloomFilterSegmentPrunerTest.java | 13 +++-
.../query/reduce/ExecutionStatsAggregatorTest.java | 8 +++
.../pinot/query/runtime/operator/LeafOperator.java | 12 +++-
.../TimeSeriesExchangeReceiveOperator.java | 4 +-
.../query/runtime/operator/LeafOperatorTest.java | 8 +++
.../org/apache/pinot/segment/spi/FetchContext.java | 28 ++------
.../apache/pinot/spi/query/QueryThreadContext.java | 62 ++++++++++++++--
.../apache/pinot/spi/query/RemoteAccessStats.java | 75 +++++++++++++++++++
22 files changed, 504 insertions(+), 104 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 1187448c81b..989b148405f 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
@@ -164,11 +164,15 @@ public interface DataTable {
// 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);
+ REMOTE_ACCESS_BYTES(46, "remoteAccessBytes", MetadataValueType.LONG),
+ // The subset of the remote accesses made directly, i.e. outside the
buffers fetched for the query, and the number
+ // of bytes they read.
+ NUM_DIRECT_REMOTE_ACCESSES(47, "numDirectRemoteAccesses",
MetadataValueType.LONG),
+ DIRECT_REMOTE_ACCESS_BYTES(48, "directRemoteAccessBytes",
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 = REMOTE_ACCESS_BYTES.getId();
+ private static final int MAX_ID = DIRECT_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 cd9bc76f160..9fb3a9a79f1 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
@@ -229,6 +229,14 @@ public interface BrokerResponse {
/// Returns the number of bytes read by the remote accesses (see
[#getNumRemoteAccesses()]).
long getRemoteAccessBytes();
+ /// Returns the number of remote accesses (see [#getNumRemoteAccesses()])
made directly, i.e. outside the buffers
+ /// fetched for the query. Each of them is a round trip to the remote
storage tier for a single read, so a non-zero
+ /// value points at reads the fetch plan of the query did not cover.
+ long getNumDirectRemoteAccesses();
+
+ /// Returns the number of bytes read by the direct remote accesses (see
[#getNumDirectRemoteAccesses()]).
+ long getDirectRemoteAccessBytes();
+
/// 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 81749d6002f..7fd2bd5fdf8 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
@@ -49,8 +49,9 @@ import org.apache.pinot.spi.utils.JsonUtils;
"numConsumingSegmentsQueried", "numConsumingSegmentsProcessed",
"numConsumingSegmentsMatched",
"minConsumingFreshnessTimeMs", "numSegmentsPrunedByBroker",
"numSegmentsPrunedByServer",
"numSegmentsPrunedInvalid", "numSegmentsPrunedByLimit",
"numSegmentsPrunedByValue", "numRemoteAccesses",
- "remoteAccessBytes", "brokerReduceTimeMs", "offlineThreadCpuTimeNs",
"realtimeThreadCpuTimeNs",
- "offlineSystemActivitiesCpuTimeNs", "realtimeSystemActivitiesCpuTimeNs",
"offlineResponseSerializationCpuTimeNs",
+ "remoteAccessBytes", "numDirectRemoteAccesses", "directRemoteAccessBytes",
"brokerReduceTimeMs",
+ "offlineThreadCpuTimeNs", "realtimeThreadCpuTimeNs",
"offlineSystemActivitiesCpuTimeNs",
+ "realtimeSystemActivitiesCpuTimeNs",
"offlineResponseSerializationCpuTimeNs",
"realtimeResponseSerializationCpuTimeNs", "offlineTotalCpuTimeNs",
"realtimeTotalCpuTimeNs",
"explainPlanNumEmptyFilterSegments",
"explainPlanNumMatchAllFilterSegments", "traceInfo", "tablesQueried",
"offlineThreadMemAllocatedBytes", "realtimeThreadMemAllocatedBytes",
"offlineResponseSerMemAllocatedBytes",
@@ -101,6 +102,8 @@ public class BrokerResponseNative implements BrokerResponse
{
private long _numSegmentsPrunedByValue = 0L;
private long _numRemoteAccesses = 0L;
private long _remoteAccessBytes = 0L;
+ private long _numDirectRemoteAccesses = 0L;
+ private long _directRemoteAccessBytes = 0L;
private long _brokerReduceTimeMs = 0L;
private long _offlineThreadCpuTimeNs = 0L;
private long _realtimeThreadCpuTimeNs = 0L;
@@ -480,6 +483,7 @@ public class BrokerResponseNative implements BrokerResponse
{
_numSegmentsPrunedByValue = numSegmentsPrunedByValue;
}
+ @JsonInclude(JsonInclude.Include.NON_DEFAULT)
@Override
public long getNumRemoteAccesses() {
return _numRemoteAccesses;
@@ -489,6 +493,7 @@ public class BrokerResponseNative implements BrokerResponse
{
_numRemoteAccesses = numRemoteAccesses;
}
+ @JsonInclude(JsonInclude.Include.NON_DEFAULT)
@Override
public long getRemoteAccessBytes() {
return _remoteAccessBytes;
@@ -498,6 +503,26 @@ public class BrokerResponseNative implements
BrokerResponse {
_remoteAccessBytes = remoteAccessBytes;
}
+ @JsonInclude(JsonInclude.Include.NON_DEFAULT)
+ @Override
+ public long getNumDirectRemoteAccesses() {
+ return _numDirectRemoteAccesses;
+ }
+
+ public void setNumDirectRemoteAccesses(long numDirectRemoteAccesses) {
+ _numDirectRemoteAccesses = numDirectRemoteAccesses;
+ }
+
+ @JsonInclude(JsonInclude.Include.NON_DEFAULT)
+ @Override
+ public long getDirectRemoteAccessBytes() {
+ return _directRemoteAccessBytes;
+ }
+
+ public void setDirectRemoteAccessBytes(long directRemoteAccessBytes) {
+ _directRemoteAccessBytes = directRemoteAccessBytes;
+ }
+
@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 c9657f39ea6..d4b2152d4be 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
@@ -47,9 +47,9 @@ import org.apache.pinot.common.response.ProcessingException;
"numSegmentsQueried", "numSegmentsProcessed", "numSegmentsMatched",
"numConsumingSegmentsQueried",
"numConsumingSegmentsProcessed", "numConsumingSegmentsMatched",
"minConsumingFreshnessTimeMs",
"numSegmentsPrunedByBroker", "numSegmentsPrunedByServer",
"numSegmentsPrunedInvalid", "numSegmentsPrunedByLimit",
- "numSegmentsPrunedByValue", "numRemoteAccesses", "remoteAccessBytes",
"brokerReduceTimeMs",
- "offlineThreadCpuTimeNs", "realtimeThreadCpuTimeNs",
"offlineSystemActivitiesCpuTimeNs",
- "realtimeSystemActivitiesCpuTimeNs",
"offlineResponseSerializationCpuTimeNs",
+ "numSegmentsPrunedByValue", "numRemoteAccesses", "remoteAccessBytes",
"numDirectRemoteAccesses",
+ "directRemoteAccessBytes", "brokerReduceTimeMs", "offlineThreadCpuTimeNs",
"realtimeThreadCpuTimeNs",
+ "offlineSystemActivitiesCpuTimeNs", "realtimeSystemActivitiesCpuTimeNs",
"offlineResponseSerializationCpuTimeNs",
"realtimeResponseSerializationCpuTimeNs", "offlineTotalCpuTimeNs",
"realtimeTotalCpuTimeNs",
"explainPlanNumEmptyFilterSegments",
"explainPlanNumMatchAllFilterSegments", "traceInfo", "tablesQueried",
"offlineThreadMemAllocatedBytes", "realtimeThreadMemAllocatedBytes",
"offlineResponseSerMemAllocatedBytes",
@@ -409,16 +409,30 @@ public class BrokerResponseNativeV2 implements
BrokerResponse {
return _brokerStats.getLong(StatKey.NUM_SEGMENTS_PRUNED_BY_VALUE);
}
+ @JsonInclude(JsonInclude.Include.NON_DEFAULT)
@Override
public long getNumRemoteAccesses() {
return _brokerStats.getLong(StatKey.NUM_REMOTE_ACCESSES);
}
+ @JsonInclude(JsonInclude.Include.NON_DEFAULT)
@Override
public long getRemoteAccessBytes() {
return _brokerStats.getLong(StatKey.REMOTE_ACCESS_BYTES);
}
+ @JsonInclude(JsonInclude.Include.NON_DEFAULT)
+ @Override
+ public long getNumDirectRemoteAccesses() {
+ return _brokerStats.getLong(StatKey.NUM_DIRECT_REMOTE_ACCESSES);
+ }
+
+ @JsonInclude(JsonInclude.Include.NON_DEFAULT)
+ @Override
+ public long getDirectRemoteAccessBytes() {
+ return _brokerStats.getLong(StatKey.DIRECT_REMOTE_ACCESS_BYTES);
+ }
+
@Override
public long getBrokerReduceTimeMs() {
return _brokerReduceTimeMs;
@@ -582,7 +596,9 @@ public class BrokerResponseNativeV2 implements
BrokerResponse {
EARLY_TERMINATION_REASONS(StatMap.Type.STRING_SET),
LITE_MODE_LEAF_STAGE_LIMIT_REACHED(StatMap.Type.BOOLEAN),
NUM_REMOTE_ACCESSES(StatMap.Type.LONG),
- REMOTE_ACCESS_BYTES(StatMap.Type.LONG);
+ REMOTE_ACCESS_BYTES(StatMap.Type.LONG),
+ NUM_DIRECT_REMOTE_ACCESSES(StatMap.Type.LONG),
+ DIRECT_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 d88d30f92e0..a2f04533ab4 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
@@ -30,9 +30,9 @@ import org.apache.pinot.common.response.CursorResponse;
"numSegmentsQueried", "numSegmentsProcessed", "numSegmentsMatched",
"numConsumingSegmentsQueried",
"numConsumingSegmentsProcessed", "numConsumingSegmentsMatched",
"minConsumingFreshnessTimeMs",
"numSegmentsPrunedByBroker", "numSegmentsPrunedByServer",
"numSegmentsPrunedInvalid", "numSegmentsPrunedByLimit",
- "numSegmentsPrunedByValue", "numRemoteAccesses", "remoteAccessBytes",
"brokerReduceTimeMs",
- "offlineThreadCpuTimeNs", "realtimeThreadCpuTimeNs",
"offlineSystemActivitiesCpuTimeNs",
- "realtimeSystemActivitiesCpuTimeNs",
"offlineResponseSerializationCpuTimeNs",
+ "numSegmentsPrunedByValue", "numRemoteAccesses", "remoteAccessBytes",
"numDirectRemoteAccesses",
+ "directRemoteAccessBytes", "brokerReduceTimeMs", "offlineThreadCpuTimeNs",
"realtimeThreadCpuTimeNs",
+ "offlineSystemActivitiesCpuTimeNs", "realtimeSystemActivitiesCpuTimeNs",
"offlineResponseSerializationCpuTimeNs",
"realtimeResponseSerializationCpuTimeNs", "offlineTotalCpuTimeNs",
"realtimeTotalCpuTimeNs",
"explainPlanNumEmptyFilterSegments",
"explainPlanNumMatchAllFilterSegments", "traceInfo", "tableQueries",
"groupsTrimmed", "approximateFunctionApplied",
@@ -85,6 +85,8 @@ public class CursorResponseNative extends
BrokerResponseNative implements Cursor
setNumSegmentsPrunedByValue(response.getNumSegmentsPrunedByValue());
setNumRemoteAccesses(response.getNumRemoteAccesses());
setRemoteAccessBytes(response.getRemoteAccessBytes());
+ setNumDirectRemoteAccesses(response.getNumDirectRemoteAccesses());
+ setDirectRemoteAccessBytes(response.getDirectRemoteAccessBytes());
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 1d71932777b..7e3a81d18ad 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
@@ -170,6 +170,10 @@ public class TimeSeriesResponseMapper {
getLongMetadataValue(metadata,
DataTable.MetadataKey.NUM_REMOTE_ACCESSES));
map.merge(BrokerResponseNativeV2.StatKey.REMOTE_ACCESS_BYTES,
getLongMetadataValue(metadata,
DataTable.MetadataKey.REMOTE_ACCESS_BYTES));
+ map.merge(BrokerResponseNativeV2.StatKey.NUM_DIRECT_REMOTE_ACCESSES,
+ getLongMetadataValue(metadata,
DataTable.MetadataKey.NUM_DIRECT_REMOTE_ACCESSES));
+ map.merge(BrokerResponseNativeV2.StatKey.DIRECT_REMOTE_ACCESS_BYTES,
+ getLongMetadataValue(metadata,
DataTable.MetadataKey.DIRECT_REMOTE_ACCESS_BYTES));
brokerResponse.addBrokerStats(map);
brokerResponse.setTimeUsedMs(getLongMetadataValue(metadata,
DataTable.MetadataKey.TIME_USED_MS));
}
diff --git
a/pinot-common/src/test/java/org/apache/pinot/common/response/broker/BrokerResponseNativeTest.java
b/pinot-common/src/test/java/org/apache/pinot/common/response/broker/BrokerResponseNativeTest.java
index 7ede9c2932b..c0b37673202 100644
---
a/pinot-common/src/test/java/org/apache/pinot/common/response/broker/BrokerResponseNativeTest.java
+++
b/pinot-common/src/test/java/org/apache/pinot/common/response/broker/BrokerResponseNativeTest.java
@@ -19,12 +19,16 @@
package org.apache.pinot.common.response.broker;
import com.fasterxml.jackson.databind.JsonNode;
-import com.fasterxml.jackson.databind.ObjectMapper;
import java.io.IOException;
import org.apache.pinot.spi.exception.QueryErrorCode;
-import org.testng.Assert;
+import org.apache.pinot.spi.utils.JsonUtils;
import org.testng.annotations.Test;
+import static org.testng.Assert.assertEquals;
+import static org.testng.Assert.assertFalse;
+import static org.testng.Assert.assertNull;
+import static org.testng.Assert.assertTrue;
+
public class BrokerResponseNativeTest {
@Test
@@ -33,9 +37,39 @@ public class BrokerResponseNativeTest {
BrokerResponseNative expected = BrokerResponseNative.EMPTY_RESULT;
String brokerString = expected.toJsonString();
BrokerResponseNative actual =
BrokerResponseNative.fromJsonString(brokerString);
- Assert.assertEquals(actual.getNumDocsScanned(),
expected.getNumDocsScanned());
- Assert.assertEquals(actual.getTimeUsedMs(), expected.getTimeUsedMs());
- Assert.assertEquals(actual.getResultTable(), expected.getResultTable());
+ assertEquals(actual.getNumDocsScanned(), expected.getNumDocsScanned());
+ assertEquals(actual.getTimeUsedMs(), expected.getTimeUsedMs());
+ assertEquals(actual.getResultTable(), expected.getResultTable());
+ }
+
+ /// The remote access stats are serialized only when a remote access
happened, and the direct ones only when a direct
+ /// remote access happened.
+ @Test
+ public void testRemoteAccessStatsSerializedOnlyWhenNonZero()
+ throws IOException {
+ BrokerResponseNative response = new BrokerResponseNative();
+ JsonNode json = JsonUtils.stringToJsonNode(response.toJsonString());
+ assertFalse(json.has("numRemoteAccesses"));
+ assertFalse(json.has("remoteAccessBytes"));
+ assertFalse(json.has("numDirectRemoteAccesses"));
+ assertFalse(json.has("directRemoteAccessBytes"));
+
+ response.setNumRemoteAccesses(7L);
+ response.setRemoteAccessBytes(700L);
+ json = JsonUtils.stringToJsonNode(response.toJsonString());
+ assertEquals(json.get("numRemoteAccesses").asLong(), 7L);
+ assertEquals(json.get("remoteAccessBytes").asLong(), 700L);
+ assertFalse(json.has("numDirectRemoteAccesses"));
+ assertFalse(json.has("directRemoteAccessBytes"));
+
+ response.setNumDirectRemoteAccesses(2L);
+ response.setDirectRemoteAccessBytes(20L);
+ json = JsonUtils.stringToJsonNode(response.toJsonString());
+ assertEquals(json.get("numDirectRemoteAccesses").asLong(), 2L);
+ assertEquals(json.get("directRemoteAccessBytes").asLong(), 20L);
+ BrokerResponseNative deserialized =
BrokerResponseNative.fromJsonString(response.toJsonString());
+ assertEquals(deserialized.getNumDirectRemoteAccesses(), 2L);
+ assertEquals(deserialized.getDirectRemoteAccessBytes(), 20L);
}
@Test
@@ -44,9 +78,8 @@ public class BrokerResponseNativeTest {
BrokerResponseNative expected = BrokerResponseNative.NO_TABLE_RESULT;
String brokerString = expected.toJsonString();
BrokerResponseNative actual =
BrokerResponseNative.fromJsonString(brokerString);
- Assert.assertEquals(actual.getExceptions().get(0).getErrorCode(),
- QueryErrorCode.BROKER_RESOURCE_MISSING.getId());
- Assert.assertEquals(actual.getExceptions().get(0).getMessage(),
+ assertEquals(actual.getExceptions().get(0).getErrorCode(),
QueryErrorCode.BROKER_RESOURCE_MISSING.getId());
+ assertEquals(actual.getExceptions().get(0).getMessage(),
QueryErrorCode.BROKER_RESOURCE_MISSING.getDefaultMessage());
}
@@ -58,12 +91,12 @@ public class BrokerResponseNativeTest {
expected.addException(new QueryProcessingException(400, errorMsgStr));
String brokerString = expected.toJsonString();
BrokerResponseNative newBrokerResponse =
BrokerResponseNative.fromJsonString(brokerString);
-
Assert.assertEquals(newBrokerResponse.getExceptions().get(0).getErrorCode(),
+ assertEquals(newBrokerResponse.getExceptions().get(0).getErrorCode(),
QueryErrorCode.BROKER_RESOURCE_MISSING.getId());
- Assert.assertEquals(newBrokerResponse.getExceptions().get(0).getMessage(),
+ assertEquals(newBrokerResponse.getExceptions().get(0).getMessage(),
QueryErrorCode.BROKER_RESOURCE_MISSING.getDefaultMessage());
-
Assert.assertEquals(newBrokerResponse.getExceptions().get(1).getErrorCode(),
400);
- Assert.assertEquals(newBrokerResponse.getExceptions().get(1).getMessage(),
errorMsgStr);
+ assertEquals(newBrokerResponse.getExceptions().get(1).getErrorCode(), 400);
+ assertEquals(newBrokerResponse.getExceptions().get(1).getMessage(),
errorMsgStr);
}
@Test
@@ -73,10 +106,10 @@ public class BrokerResponseNativeTest {
expected.setMaterializedViewQueried("orders_by_day_OFFLINE");
String brokerString = expected.toJsonString();
- Assert.assertTrue(brokerString.contains("\"materializedViewQueried\""));
+ assertTrue(brokerString.contains("\"materializedViewQueried\""));
BrokerResponseNative actual =
BrokerResponseNative.fromJsonString(brokerString);
- Assert.assertEquals(actual.getMaterializedViewQueried(),
expected.getMaterializedViewQueried());
+ assertEquals(actual.getMaterializedViewQueried(),
expected.getMaterializedViewQueried());
}
@Test
@@ -84,7 +117,7 @@ public class BrokerResponseNativeTest {
throws IOException {
BrokerResponseNative response = new BrokerResponseNative();
String json = response.toJsonString();
- Assert.assertFalse(json.contains("materializedViewQueried"),
+ assertFalse(json.contains("materializedViewQueried"),
"Null materializedViewQueried should be suppressed by
@JsonInclude(NON_NULL)");
}
@@ -92,26 +125,25 @@ public class BrokerResponseNativeTest {
public void testServerStatsRoundTrip()
throws IOException {
BrokerResponseNative expected = new BrokerResponseNative();
- String stats =
-
"Server=SubmitDelayMs,ResponseDelayMs,ResponseSize,DeserializationTimeMs,RequestSentDelayMs;"
- + "pinot-server-0_O=0,1,7571,0,0;pinot-server-1_O=0,1,7574,0,0";
+ String stats =
"Server=SubmitDelayMs,ResponseDelayMs,ResponseSize,DeserializationTimeMs,RequestSentDelayMs;"
+ + "pinot-server-0_O=0,1,7571,0,0;pinot-server-1_O=0,1,7574,0,0";
expected.setServerStats(stats);
BrokerResponseNative actual =
BrokerResponseNative.fromJsonString(expected.toJsonString());
- Assert.assertEquals(actual.getServerStats(), stats);
+ assertEquals(actual.getServerStats(), stats);
}
@Test
public void testServerStatsAbsentWhenNull()
throws IOException {
BrokerResponseNative response = new BrokerResponseNative();
- JsonNode tree = new ObjectMapper().readTree(response.toJsonString());
- Assert.assertFalse(tree.has("serverStats"));
+ JsonNode tree = JsonUtils.stringToJsonNode(response.toJsonString());
+ assertFalse(tree.has("serverStats"));
}
@Test
public void testServerStatsDefaultsToNull() {
- Assert.assertNull(new BrokerResponseNative().getServerStats());
+ assertNull(new BrokerResponseNative().getServerStats());
}
/// Regression test for backward-compatible deserialization:
[BrokerResponseNative] does not
@@ -127,7 +159,7 @@ public class BrokerResponseNativeTest {
+ "\"details\":{\"count\":2}},\"numDocsScanned\":5}";
BrokerResponseNative actual = BrokerResponseNative.fromJsonString(json);
// The unknown field is ignored on the legacy impl; other fields still
deserialize correctly.
- Assert.assertEquals(actual.getNumDocsScanned(), 5);
- Assert.assertTrue(actual.getResponseMetadata().isEmpty());
+ assertEquals(actual.getNumDocsScanned(), 5);
+ assertTrue(actual.getResponseMetadata().isEmpty());
}
}
diff --git
a/pinot-common/src/test/java/org/apache/pinot/common/response/broker/BrokerResponseNativeV2Test.java
b/pinot-common/src/test/java/org/apache/pinot/common/response/broker/BrokerResponseNativeV2Test.java
index 22cf653d6e5..139fa96aa76 100644
---
a/pinot-common/src/test/java/org/apache/pinot/common/response/broker/BrokerResponseNativeV2Test.java
+++
b/pinot-common/src/test/java/org/apache/pinot/common/response/broker/BrokerResponseNativeV2Test.java
@@ -52,6 +52,36 @@ public class BrokerResponseNativeV2Test {
assertTrue(brokerResponse.isPartialResult());
}
+ /// The remote access stats are serialized only when a remote access
happened, and the direct ones only when a direct
+ /// remote access happened.
+ @Test
+ public void testRemoteAccessStatsSerializedOnlyWhenNonZero() {
+ BrokerResponseNativeV2 response = new BrokerResponseNativeV2();
+ JsonNode json = JsonUtils.objectToJsonNode(response);
+ assertFalse(json.has("numRemoteAccesses"));
+ assertFalse(json.has("remoteAccessBytes"));
+ assertFalse(json.has("numDirectRemoteAccesses"));
+ assertFalse(json.has("directRemoteAccessBytes"));
+
+ StatMap<BrokerResponseNativeV2.StatKey> stats = new
StatMap<>(BrokerResponseNativeV2.StatKey.class);
+ stats.merge(BrokerResponseNativeV2.StatKey.NUM_REMOTE_ACCESSES, 7L);
+ stats.merge(BrokerResponseNativeV2.StatKey.REMOTE_ACCESS_BYTES, 700L);
+ response.addBrokerStats(stats);
+ json = JsonUtils.objectToJsonNode(response);
+ assertEquals(json.get("numRemoteAccesses").asLong(), 7L);
+ assertEquals(json.get("remoteAccessBytes").asLong(), 700L);
+ assertFalse(json.has("numDirectRemoteAccesses"));
+ assertFalse(json.has("directRemoteAccessBytes"));
+
+ stats = new StatMap<>(BrokerResponseNativeV2.StatKey.class);
+ stats.merge(BrokerResponseNativeV2.StatKey.NUM_DIRECT_REMOTE_ACCESSES, 2L);
+ stats.merge(BrokerResponseNativeV2.StatKey.DIRECT_REMOTE_ACCESS_BYTES,
20L);
+ response.addBrokerStats(stats);
+ json = JsonUtils.objectToJsonNode(response);
+ assertEquals(json.get("numDirectRemoteAccesses").asLong(), 2L);
+ assertEquals(json.get("directRemoteAccessBytes").asLong(), 20L);
+ }
+
@Test
public void testResponseMetadataSerialization()
throws Exception {
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 5d63ac36cdd..28da024c3e4 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
@@ -29,6 +29,7 @@ import
org.apache.pinot.core.query.request.context.QueryContext;
import org.apache.pinot.segment.spi.FetchContext;
import org.apache.pinot.segment.spi.SegmentContext;
import org.apache.pinot.spi.accounting.ThreadResourceSnapshot;
+import org.apache.pinot.spi.query.QueryThreadContext;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -99,16 +100,6 @@ 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
@@ -162,9 +153,12 @@ public class InstanceResponseOperator extends
BaseOperator<InstanceResponseBlock
}
}
+ /// Releases the fetched buffers of all the segments, and reports their
remote accesses to the query execution.
public void releaseAll() {
for (int i = 0; i < _fetchContextSize; i++) {
- _segmentContexts.get(i).getIndexSegment().release(_fetchContexts.get(i));
+ FetchContext fetchContext = _fetchContexts.get(i);
+ _segmentContexts.get(i).getIndexSegment().release(fetchContext);
+
QueryThreadContext.mergeRemoteAccessStats(fetchContext.getRemoteAccessStats());
}
}
diff --git
a/pinot-core/src/main/java/org/apache/pinot/core/query/executor/ServerQueryExecutorV1Impl.java
b/pinot-core/src/main/java/org/apache/pinot/core/query/executor/ServerQueryExecutorV1Impl.java
index a948f1575b1..e0625491dd9 100644
---
a/pinot-core/src/main/java/org/apache/pinot/core/query/executor/ServerQueryExecutorV1Impl.java
+++
b/pinot-core/src/main/java/org/apache/pinot/core/query/executor/ServerQueryExecutorV1Impl.java
@@ -85,6 +85,7 @@ import org.apache.pinot.spi.plugin.PluginManager;
import org.apache.pinot.spi.query.QueryExecutionContext;
import org.apache.pinot.spi.query.QueryScanCostContext;
import org.apache.pinot.spi.query.QueryThreadContext;
+import org.apache.pinot.spi.query.RemoteAccessStats;
import org.apache.pinot.spi.trace.Tracer;
import org.apache.pinot.spi.trace.Tracing;
import org.apache.pinot.spi.utils.CommonConstants.Accounting.ScanKillingMode;
@@ -161,6 +162,44 @@ public class ServerQueryExecutorV1Impl implements
QueryExecutor {
@Override
public InstanceResponseBlock execute(ServerQueryRequest queryRequest,
ExecutorService executorService,
@Nullable ResultsBlockStreamer streamer, @Nullable PlanMaker
planMakerOverride) {
+ QueryThreadContext threadContext = QueryThreadContext.getIfAvailable();
+ // NOTE: In production code, threadContext should never be null. It might
be null in tests when QueryThreadContext
+ // is not set up.
+ if (threadContext == null) {
+ return executeWithTrace(queryRequest, executorService, streamer,
planMakerOverride);
+ }
+ // Collect the remote accesses of this execution from this thread and the
ones it spawns, and report them in the
+ // response metadata. The stats are replaced per execution because the
ones of the thread context are shared by
+ // every execution spawned from it, e.g. the two of a hybrid table.
+ RemoteAccessStats remoteAccessStats = new RemoteAccessStats();
+ threadContext.setRemoteAccessStats(remoteAccessStats);
+ InstanceResponseBlock instanceResponse =
+ executeWithTrace(queryRequest, executorService, streamer,
planMakerOverride);
+ addRemoteAccessStats(instanceResponse, remoteAccessStats);
+ return instanceResponse;
+ }
+
+ /// Adds the remote access stats to the response metadata, each pair of keys
only when its count is not zero.
+ private static void addRemoteAccessStats(InstanceResponseBlock
instanceResponse,
+ RemoteAccessStats remoteAccessStats) {
+ long numRemoteAccesses = remoteAccessStats.getNumRemoteAccesses();
+ if (numRemoteAccesses == 0) {
+ return;
+ }
+ instanceResponse.addMetadata(MetadataKey.NUM_REMOTE_ACCESSES.getName(),
Long.toString(numRemoteAccesses));
+ instanceResponse.addMetadata(MetadataKey.REMOTE_ACCESS_BYTES.getName(),
+ Long.toString(remoteAccessStats.getRemoteAccessBytes()));
+ long numDirectRemoteAccesses =
remoteAccessStats.getNumDirectRemoteAccesses();
+ if (numDirectRemoteAccesses > 0) {
+
instanceResponse.addMetadata(MetadataKey.NUM_DIRECT_REMOTE_ACCESSES.getName(),
+ Long.toString(numDirectRemoteAccesses));
+
instanceResponse.addMetadata(MetadataKey.DIRECT_REMOTE_ACCESS_BYTES.getName(),
+ Long.toString(remoteAccessStats.getDirectRemoteAccessBytes()));
+ }
+ }
+
+ private InstanceResponseBlock executeWithTrace(ServerQueryRequest
queryRequest, ExecutorService executorService,
+ @Nullable ResultsBlockStreamer streamer, @Nullable PlanMaker
planMakerOverride) {
if (!queryRequest.isEnableTrace()) {
return executeInternal(queryRequest, executorService, streamer,
planMakerOverride);
}
diff --git
a/pinot-core/src/main/java/org/apache/pinot/core/query/pruner/BloomFilterSegmentPruner.java
b/pinot-core/src/main/java/org/apache/pinot/core/query/pruner/BloomFilterSegmentPruner.java
index 623ad153baf..d3b6d00b363 100644
---
a/pinot-core/src/main/java/org/apache/pinot/core/query/pruner/BloomFilterSegmentPruner.java
+++
b/pinot-core/src/main/java/org/apache/pinot/core/query/pruner/BloomFilterSegmentPruner.java
@@ -37,6 +37,7 @@ import org.apache.pinot.segment.spi.datasource.DataSource;
import org.apache.pinot.segment.spi.datasource.DataSourceMetadata;
import org.apache.pinot.segment.spi.index.reader.BloomFilterReader;
import org.apache.pinot.spi.env.PinotConfiguration;
+import org.apache.pinot.spi.query.QueryThreadContext;
/// Prunes segments using bloom filters for EQ and IN predicates, with
optional prefetch.
@@ -99,11 +100,12 @@ public class BloomFilterSegmentPruner extends
ValueBasedSegmentPruner {
}
return pruneFunc.apply(fetchContexts);
} finally {
- // Release the prefetched bloom filters
+ // Release the prefetched bloom filters, and report their remote
accesses to the query execution
for (int i = 0; i < numSegments; i++) {
FetchContext fetchContext = fetchContexts[i];
if (fetchContext != null) {
segments.get(i).release(fetchContext);
+
QueryThreadContext.mergeRemoteAccessStats(fetchContext.getRemoteAccessStats());
}
}
}
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 15cc51ef7b9..af7b892d21e 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
@@ -73,6 +73,8 @@ public class ExecutionStatsAggregator {
private long _numSegmentsPrunedByValue = 0L;
private long _numRemoteAccesses = 0L;
private long _remoteAccessBytes = 0L;
+ private long _numDirectRemoteAccesses = 0L;
+ private long _directRemoteAccessBytes = 0L;
private long _explainPlanNumEmptyFilterSegments = 0L;
private long _explainPlanNumMatchAllFilterSegments = 0L;
private boolean _groupsTrimmed = false;
@@ -221,6 +223,10 @@ public class ExecutionStatsAggregator {
l -> _numSegmentsPrunedByValue += l);
withNotNullLongMetadata(metadata,
DataTable.MetadataKey.NUM_REMOTE_ACCESSES, l -> _numRemoteAccesses += l);
withNotNullLongMetadata(metadata,
DataTable.MetadataKey.REMOTE_ACCESS_BYTES, l -> _remoteAccessBytes += l);
+ withNotNullLongMetadata(metadata,
DataTable.MetadataKey.NUM_DIRECT_REMOTE_ACCESSES,
+ l -> _numDirectRemoteAccesses += l);
+ withNotNullLongMetadata(metadata,
DataTable.MetadataKey.DIRECT_REMOTE_ACCESS_BYTES,
+ l -> _directRemoteAccessBytes += l);
String explainPlanNumEmptyFilterSegments =
metadata.get(DataTable.MetadataKey.EXPLAIN_PLAN_NUM_EMPTY_FILTER_SEGMENTS.getName());
@@ -309,6 +315,8 @@ public class ExecutionStatsAggregator {
brokerResponseNative.setNumSegmentsPrunedByValue(_numSegmentsPrunedByValue);
brokerResponseNative.setNumRemoteAccesses(_numRemoteAccesses);
brokerResponseNative.setRemoteAccessBytes(_remoteAccessBytes);
+ brokerResponseNative.setNumDirectRemoteAccesses(_numDirectRemoteAccesses);
+ brokerResponseNative.setDirectRemoteAccessBytes(_directRemoteAccessBytes);
brokerResponseNative.setExplainPlanNumEmptyFilterSegments(_explainPlanNumEmptyFilterSegments);
brokerResponseNative.setExplainPlanNumMatchAllFilterSegments(_explainPlanNumMatchAllFilterSegments);
if (_numConsumingSegmentsQueried > 0) {
@@ -408,6 +416,8 @@ public class ExecutionStatsAggregator {
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.NUM_DIRECT_REMOTE_ACCESSES,
_numDirectRemoteAccesses);
+ putLong(metadata, DataTable.MetadataKey.DIRECT_REMOTE_ACCESS_BYTES,
_directRemoteAccessBytes);
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 ac700099b3b..e5c484df863 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
@@ -29,9 +29,14 @@ 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.segment.spi.IndexSegment;
+import org.apache.pinot.segment.spi.SegmentContext;
+import org.apache.pinot.spi.query.QueryThreadContext;
+import org.apache.pinot.spi.query.RemoteAccessStats;
import
org.apache.pinot.spi.utils.CommonConstants.Broker.Request.QueryOptionKey;
import org.testng.annotations.Test;
+import static org.mockito.Mockito.doAnswer;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.when;
import static org.testng.Assert.assertEquals;
@@ -128,36 +133,30 @@ public class InstanceResponseOperatorTest {
.containsKey(MetadataKey.LITE_MODE_LEAF_STAGE_LIMIT_REACHED.getName()));
}
+ /// Releasing the fetched buffers reports the remote accesses recorded on
the fetch contexts to the query execution.
@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<>());
+ public void testReleaseReportsRemoteAccessesOfFetchContexts() {
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()));
+ IndexSegment segment1 = mock(IndexSegment.class);
+ IndexSegment segment2 = mock(IndexSegment.class);
+ doAnswer(_ -> {
+ fetchContext1.getRemoteAccessStats().addRemoteAccesses(3, 300);
+ return null;
+ }).when(segment1).release(fetchContext1);
+ doAnswer(_ -> {
+ fetchContext2.getRemoteAccessStats().addRemoteAccesses(4, 400);
+ return null;
+ }).when(segment2).release(fetchContext2);
+ InstanceResponseOperator operator = new
InstanceResponseOperator(mock(BaseCombineOperator.class),
+ List.of(new SegmentContext(segment1), new SegmentContext(segment2)),
List.of(fetchContext1, fetchContext2),
+ mock(QueryContext.class));
+
+ try (QueryThreadContext threadContext =
QueryThreadContext.openForSseTest()) {
+ operator.releaseAll();
+ RemoteAccessStats remoteAccessStats =
threadContext.getRemoteAccessStats();
+ assertEquals(remoteAccessStats.getNumRemoteAccesses(), 7);
+ assertEquals(remoteAccessStats.getRemoteAccessBytes(), 700);
+ }
}
}
diff --git
a/pinot-core/src/test/java/org/apache/pinot/core/query/executor/QueryExecutorTest.java
b/pinot-core/src/test/java/org/apache/pinot/core/query/executor/QueryExecutorTest.java
index ecd4a223d88..61965252a8a 100644
---
a/pinot-core/src/test/java/org/apache/pinot/core/query/executor/QueryExecutorTest.java
+++
b/pinot-core/src/test/java/org/apache/pinot/core/query/executor/QueryExecutorTest.java
@@ -32,11 +32,13 @@ import java.util.Optional;
import java.util.UUID;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;
import org.apache.commons.configuration2.PropertiesConfiguration;
import org.apache.commons.io.FileUtils;
import org.apache.helix.HelixManager;
+import org.apache.pinot.common.datatable.DataTable.MetadataKey;
import org.apache.pinot.common.metrics.ServerMetrics;
import org.apache.pinot.common.request.InstanceRequest;
import org.apache.pinot.common.request.context.ExpressionContext;
@@ -48,6 +50,8 @@ import
org.apache.pinot.core.operator.blocks.InstanceResponseBlock;
import org.apache.pinot.core.operator.blocks.results.AggregationResultsBlock;
import org.apache.pinot.core.operator.blocks.results.GroupByResultsBlock;
import org.apache.pinot.core.operator.timeseries.TimeSeriesOperatorUtils;
+import org.apache.pinot.core.plan.Plan;
+import org.apache.pinot.core.plan.maker.PlanMaker;
import
org.apache.pinot.core.query.aggregation.function.TimeSeriesAggregationFunction;
import org.apache.pinot.core.query.request.ServerQueryRequest;
import org.apache.pinot.core.query.request.context.QueryContext;
@@ -85,6 +89,7 @@ import org.testng.annotations.AfterClass;
import org.testng.annotations.BeforeClass;
import org.testng.annotations.Test;
+import static org.mockito.Mockito.any;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.when;
import static org.testng.Assert.assertEquals;
@@ -226,6 +231,84 @@ public class QueryExecutorTest {
assertEquals(((AggregationResultsBlock)
instanceResponse.getResultsBlock()).getResults().get(0), 0.0);
}
+ @Test
+ public void testNoRemoteAccessStatsWithoutRemoteAccesses() {
+ String query = "SELECT COUNT(*) FROM " + OFFLINE_TABLE_NAME;
+ InstanceRequest instanceRequest = new InstanceRequest(0L,
CalciteSqlCompiler.compileToBrokerRequest(query));
+ instanceRequest.setSearchSegments(_segmentNames);
+ Map<String, String> responseMetadata =
execute(instanceRequest).getResponseMetadata();
+
assertFalse(responseMetadata.containsKey(MetadataKey.NUM_REMOTE_ACCESSES.getName()));
+
assertFalse(responseMetadata.containsKey(MetadataKey.REMOTE_ACCESS_BYTES.getName()));
+
assertFalse(responseMetadata.containsKey(MetadataKey.NUM_DIRECT_REMOTE_ACCESSES.getName()));
+
assertFalse(responseMetadata.containsKey(MetadataKey.DIRECT_REMOTE_ACCESS_BYTES.getName()));
+ }
+
+ /// The remote accesses reported while the plan executes, from the query
thread and the ones it spawns, are reported
+ /// in the response metadata, with the direct ones counted in the totals and
reported on their own.
+ @Test
+ public void testRemoteAccessStatsReportedForTheExecution()
+ throws Exception {
+ String query = "SELECT COUNT(*) FROM " + OFFLINE_TABLE_NAME;
+ InstanceRequest instanceRequest = new InstanceRequest(0L,
CalciteSqlCompiler.compileToBrokerRequest(query));
+ instanceRequest.setSearchSegments(_segmentNames);
+ ServerQueryRequest queryRequest =
+ new ServerQueryRequest(instanceRequest, ServerMetrics.get(),
System.currentTimeMillis());
+ Plan plan = mock(Plan.class);
+ when(plan.execute()).thenAnswer(_ -> {
+ QueryThreadContext.addRemoteAccesses(1, 10);
+ QUERY_RUNNERS.submit(() -> QueryThreadContext.addDirectRemoteAccesses(2,
20)).get();
+ return new InstanceResponseBlock();
+ });
+ PlanMaker planMaker = mock(PlanMaker.class);
+ when(planMaker.makeInstancePlan(any(), any(), any())).thenReturn(plan);
+ try (QueryThreadContext ignore = QueryThreadContext.openForSseTest()) {
+ Map<String, String> responseMetadata =
+ _queryExecutor.execute(queryRequest, QUERY_RUNNERS, null,
planMaker).getResponseMetadata();
+
assertEquals(responseMetadata.get(MetadataKey.NUM_REMOTE_ACCESSES.getName()),
"3");
+
assertEquals(responseMetadata.get(MetadataKey.REMOTE_ACCESS_BYTES.getName()),
"30");
+
assertEquals(responseMetadata.get(MetadataKey.NUM_DIRECT_REMOTE_ACCESSES.getName()),
"2");
+
assertEquals(responseMetadata.get(MetadataKey.DIRECT_REMOTE_ACCESS_BYTES.getName()),
"20");
+ }
+ }
+
+ /// Executions spawned from one thread context, like the two of a hybrid
table, each report their own remote
+ /// accesses.
+ @Test
+ public void
testExecutionsSpawnedFromOneThreadContextReportTheirOwnRemoteAccesses()
+ throws Exception {
+ String query = "SELECT COUNT(*) FROM " + OFFLINE_TABLE_NAME;
+ InstanceRequest instanceRequest = new InstanceRequest(0L,
CalciteSqlCompiler.compileToBrokerRequest(query));
+ instanceRequest.setSearchSegments(_segmentNames);
+ Plan plan = mock(Plan.class);
+ when(plan.execute()).thenAnswer(_ -> {
+ QueryThreadContext.addRemoteAccesses(1, 10);
+ return new InstanceResponseBlock();
+ });
+ PlanMaker planMaker = mock(PlanMaker.class);
+ when(planMaker.makeInstancePlan(any(), any(), any())).thenReturn(plan);
+ ExecutorService executor =
QueryThreadContext.contextAwareExecutorService(Executors.newFixedThreadPool(2));
+ try (QueryThreadContext ignore = QueryThreadContext.openForSseTest()) {
+ List<Future<Map<String, String>>> futures = new ArrayList<>();
+ for (int i = 0; i < 2; i++) {
+ futures.add(executor.submit(() -> {
+ ServerQueryRequest queryRequest =
+ new ServerQueryRequest(instanceRequest, ServerMetrics.get(),
System.currentTimeMillis());
+ return _queryExecutor.execute(queryRequest, QUERY_RUNNERS, null,
planMaker).getResponseMetadata();
+ }));
+ }
+ for (Future<Map<String, String>> future : futures) {
+ Map<String, String> responseMetadata = future.get();
+
assertEquals(responseMetadata.get(MetadataKey.NUM_REMOTE_ACCESSES.getName()),
"1");
+
assertEquals(responseMetadata.get(MetadataKey.REMOTE_ACCESS_BYTES.getName()),
"10");
+ // Reported only when a direct remote access happened
+
assertFalse(responseMetadata.containsKey(MetadataKey.NUM_DIRECT_REMOTE_ACCESSES.getName()));
+
assertFalse(responseMetadata.containsKey(MetadataKey.DIRECT_REMOTE_ACCESS_BYTES.getName()));
+ }
+ } finally {
+ executor.shutdownNow();
+ }
+ }
+
@Test
public void testTimeSeriesSumQuery() {
TimeBuckets timeBuckets =
TimeBuckets.ofSeconds(TIME_SERIES_TEST_START_TIME, Duration.ofHours(2), 2);
diff --git
a/pinot-core/src/test/java/org/apache/pinot/core/query/pruner/BloomFilterSegmentPrunerTest.java
b/pinot-core/src/test/java/org/apache/pinot/core/query/pruner/BloomFilterSegmentPrunerTest.java
index 2865581173c..d7b31670c16 100644
---
a/pinot-core/src/test/java/org/apache/pinot/core/query/pruner/BloomFilterSegmentPrunerTest.java
+++
b/pinot-core/src/test/java/org/apache/pinot/core/query/pruner/BloomFilterSegmentPrunerTest.java
@@ -61,6 +61,8 @@ import org.apache.pinot.spi.data.Schema;
import org.apache.pinot.spi.data.readers.GenericRow;
import org.apache.pinot.spi.env.PinotConfiguration;
import org.apache.pinot.spi.exception.QueryCancelledException;
+import org.apache.pinot.spi.query.QueryThreadContext;
+import org.apache.pinot.spi.query.RemoteAccessStats;
import org.apache.pinot.spi.utils.ReadMode;
import org.apache.pinot.spi.utils.builder.TableConfigBuilder;
import org.mockito.ArgumentCaptor;
@@ -285,7 +287,10 @@ public class BloomFilterSegmentPrunerTest {
return null;
}).when(segment).acquire(any(FetchContext.class));
doAnswer(invocation -> {
- acquired.set(false);
+ // Record a remote access at the release of the acquired bloom filter,
like a segment directory does
+ if (acquired.getAndSet(false)) {
+
invocation.<FetchContext>getArgument(0).getRemoteAccessStats().addRemoteAccesses(1,
10);
+ }
return null;
}).when(segment).release(any(FetchContext.class));
DataSource dataSource = segment.getDataSourceNullable("column");
@@ -299,7 +304,7 @@ public class BloomFilterSegmentPrunerTest {
query.setEnablePrefetch(enablePrefetch);
query.setMaxExecutionThreads(4);
ExecutorService executor = Executors.newFixedThreadPool(4);
- try {
+ try (QueryThreadContext threadContext =
QueryThreadContext.openForSseTest()) {
List<IndexSegment> selected = useExecutor ? PRUNER.prune(segments,
query, executor)
: PRUNER.prune(segments, query);
assertEquals(selected.size(), expected.size());
@@ -313,6 +318,10 @@ public class BloomFilterSegmentPrunerTest {
verify(segment, never()).release(any(FetchContext.class));
}
}
+ // The remote accesses of the released bloom filters are reported to the
query execution
+ RemoteAccessStats remoteAccessStats =
threadContext.getRemoteAccessStats();
+ assertEquals(remoteAccessStats.getNumRemoteAccesses(), enablePrefetch ?
numSegments : 0);
+ assertEquals(remoteAccessStats.getRemoteAccessBytes(), enablePrefetch ?
10L * numSegments : 0);
} finally {
executor.shutdownNow();
assertTrue(executor.awaitTermination(5, TimeUnit.SECONDS));
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 e351288ac43..bb26e922099 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
@@ -172,9 +172,13 @@ public class ExecutionStatsAggregatorTest {
DataTable table1 = emptyServerTable();
table1.getMetadata().put(MetadataKey.NUM_REMOTE_ACCESSES.getName(), "3");
table1.getMetadata().put(MetadataKey.REMOTE_ACCESS_BYTES.getName(), "300");
+ table1.getMetadata().put(MetadataKey.NUM_DIRECT_REMOTE_ACCESSES.getName(),
"1");
+ table1.getMetadata().put(MetadataKey.DIRECT_REMOTE_ACCESS_BYTES.getName(),
"10");
DataTable table2 = emptyServerTable();
table2.getMetadata().put(MetadataKey.NUM_REMOTE_ACCESSES.getName(), "4");
table2.getMetadata().put(MetadataKey.REMOTE_ACCESS_BYTES.getName(), "400");
+ table2.getMetadata().put(MetadataKey.NUM_DIRECT_REMOTE_ACCESSES.getName(),
"2");
+ table2.getMetadata().put(MetadataKey.DIRECT_REMOTE_ACCESS_BYTES.getName(),
"20");
ExecutionStatsAggregator aggregator = new ExecutionStatsAggregator(false);
aggregator.aggregate(SERVER, table1);
aggregator.aggregate(SERVER, table2);
@@ -183,10 +187,14 @@ public class ExecutionStatsAggregatorTest {
aggregator.setStats("testTable", response, mock(BrokerMetrics.class));
assertEquals(response.getNumRemoteAccesses(), 7L);
assertEquals(response.getRemoteAccessBytes(), 700L);
+ assertEquals(response.getNumDirectRemoteAccesses(), 3L);
+ assertEquals(response.getDirectRemoteAccessBytes(), 30L);
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");
+
assertEquals(merged.getMetadata().get(MetadataKey.NUM_DIRECT_REMOTE_ACCESSES.getName()),
"3");
+
assertEquals(merged.getMetadata().get(MetadataKey.DIRECT_REMOTE_ACCESS_BYTES.getName()),
"30");
}
}
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 30290091d27..18761c37c28 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
@@ -442,6 +442,12 @@ public class LeafOperator extends MultiStageOperator
implements ExplainableOpera
case REMOTE_ACCESS_BYTES:
_statMap.merge(StatKey.REMOTE_ACCESS_BYTES,
Long.parseLong(entry.getValue()));
break;
+ case NUM_DIRECT_REMOTE_ACCESSES:
+ _statMap.merge(StatKey.NUM_DIRECT_REMOTE_ACCESSES,
Long.parseLong(entry.getValue()));
+ break;
+ case DIRECT_REMOTE_ACCESS_BYTES:
+ _statMap.merge(StatKey.DIRECT_REMOTE_ACCESS_BYTES,
Long.parseLong(entry.getValue()));
+ break;
case EXPLAIN_PLAN_NUM_EMPTY_FILTER_SEGMENTS:
LOGGER.debug("Skipping empty filter segments: {}", entry.getValue());
break;
@@ -808,7 +814,11 @@ public class LeafOperator extends MultiStageOperator
implements ExplainableOpera
/// 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);
+ REMOTE_ACCESS_BYTES(StatMap.Type.LONG),
+ /// Number of remote accesses made directly, i.e. outside the buffers
fetched for the query.
+ NUM_DIRECT_REMOTE_ACCESSES(StatMap.Type.LONG),
+ /// Number of bytes read by the direct remote accesses.
+ DIRECT_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 bfede821faf..9c76f071058 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
@@ -70,7 +70,9 @@ public class TimeSeriesExchangeReceiveOperator extends
BaseTimeSeriesOperator {
DataTable.MetadataKey.NUM_SEGMENTS_PRUNED_BY_LIMIT,
DataTable.MetadataKey.NUM_SEGMENTS_PRUNED_BY_VALUE,
DataTable.MetadataKey.NUM_REMOTE_ACCESSES,
- DataTable.MetadataKey.REMOTE_ACCESS_BYTES
+ DataTable.MetadataKey.REMOTE_ACCESS_BYTES,
+ DataTable.MetadataKey.NUM_DIRECT_REMOTE_ACCESSES,
+ DataTable.MetadataKey.DIRECT_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 69d5baff59a..a403e5154bd 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
@@ -554,6 +554,8 @@ public class LeafOperatorTest {
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");
+
metadataBlock.getResponseMetadata().put(DataTable.MetadataKey.NUM_DIRECT_REMOTE_ACCESSES.getName(),
"2");
+
metadataBlock.getResponseMetadata().put(DataTable.MetadataKey.DIRECT_REMOTE_ACCESS_BYTES.getName(),
"20");
QueryExecutor queryExecutor = mockQueryExecutor(List.of(), metadataBlock);
LeafOperator operator =
new LeafOperator(OperatorTestUtil.getTracingContext(),
mockQueryRequests(1), schema, queryExecutor,
@@ -567,14 +569,20 @@ public class LeafOperatorTest {
StatMap<LeafOperator.StatKey> leafStats = operator.copyStatMaps();
assertEquals(leafStats.getLong(LeafOperator.StatKey.NUM_REMOTE_ACCESSES),
7L);
assertEquals(leafStats.getLong(LeafOperator.StatKey.REMOTE_ACCESS_BYTES),
700L);
+
assertEquals(leafStats.getLong(LeafOperator.StatKey.NUM_DIRECT_REMOTE_ACCESSES),
2L);
+
assertEquals(leafStats.getLong(LeafOperator.StatKey.DIRECT_REMOTE_ACCESS_BYTES),
20L);
BrokerResponseNativeV2 brokerResponse = new BrokerResponseNativeV2();
MultiStageOperator.Type.LEAF.mergeInto(brokerResponse, leafStats);
assertEquals(brokerResponse.getNumRemoteAccesses(), 7L);
assertEquals(brokerResponse.getRemoteAccessBytes(), 700L);
+ assertEquals(brokerResponse.getNumDirectRemoteAccesses(), 2L);
+ assertEquals(brokerResponse.getDirectRemoteAccessBytes(), 20L);
JsonNode responseJson = JsonUtils.objectToJsonNode(brokerResponse);
assertEquals(responseJson.path("numRemoteAccesses").asLong(), 7L);
assertEquals(responseJson.path("remoteAccessBytes").asLong(), 700L);
+ assertEquals(responseJson.path("numDirectRemoteAccesses").asLong(), 2L);
+ assertEquals(responseJson.path("directRemoteAccessBytes").asLong(), 20L);
operator.close();
}
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 9949aa44636..9f0cf7688d7 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,21 +23,16 @@ 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;
+import org.apache.pinot.spi.query.RemoteAccessStats;
/// 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();
+ private final RemoteAccessStats _remoteAccessStats = new RemoteAccessStats();
/// Create a new FetchRequest for this segment, to fetch all buffers of the
given columns
/// @param fetchId unique fetch id
@@ -83,19 +78,10 @@ public class FetchContext {
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();
+ /// Returns the remote accesses (local cache misses) performed to serve this
fetch and the bytes they read. A segment
+ /// directory backed by a remote storage tier records them, typically when
the fetched buffers are released, and the
+ /// owner of the fetch reports them to the query execution afterwards, so
that they end up in the query stats.
+ public RemoteAccessStats getRemoteAccessStats() {
+ return _remoteAccessStats;
}
}
diff --git
a/pinot-spi/src/main/java/org/apache/pinot/spi/query/QueryThreadContext.java
b/pinot-spi/src/main/java/org/apache/pinot/spi/query/QueryThreadContext.java
index 52fca64440f..ce07866de43 100644
--- a/pinot-spi/src/main/java/org/apache/pinot/spi/query/QueryThreadContext.java
+++ b/pinot-spi/src/main/java/org/apache/pinot/spi/query/QueryThreadContext.java
@@ -61,12 +61,15 @@ public class QueryThreadContext implements AutoCloseable {
@Nullable
private final MseWorkerInfo _mseWorkerInfo;
private final ThreadAccountant _accountant;
+ // The stats of the query execution served by this thread, replaced by the
query executor per execution
+ private RemoteAccessStats _remoteAccessStats;
private QueryThreadContext(QueryExecutionContext executionContext, @Nullable
MseWorkerInfo mseWorkerInfo,
- ThreadAccountant accountant) {
+ ThreadAccountant accountant, RemoteAccessStats remoteAccessStats) {
_executionContext = executionContext;
_mseWorkerInfo = mseWorkerInfo;
_accountant = accountant;
+ _remoteAccessStats = remoteAccessStats;
LoggerConstants.REQUEST_ID_KEY.registerInMdc(Long.toString(executionContext.getRequestId()));
LoggerConstants.CORRELATION_ID_KEY.registerInMdc(executionContext.getCid());
String queryHash = executionContext.getQueryHash();
@@ -103,6 +106,18 @@ public class QueryThreadContext implements AutoCloseable {
return _accountant;
}
+ /// Returns the [RemoteAccessStats] of the query execution served by the
current thread.
+ public RemoteAccessStats getRemoteAccessStats() {
+ return _remoteAccessStats;
+ }
+
+ /// Sets the [RemoteAccessStats] of the query execution served by the
current thread. Tasks submitted afterwards to a
+ /// [#contextAwareExecutorService] share the same instance, so a query
executor sets fresh stats when it starts an
+ /// execution and every thread serving it reports into them through
[#addRemoteAccesses].
+ public void setRemoteAccessStats(RemoteAccessStats remoteAccessStats) {
+ _remoteAccessStats = remoteAccessStats;
+ }
+
private void sampleUsageInternal() {
_accountant.sampleUsage();
}
@@ -198,7 +213,13 @@ public class QueryThreadContext implements AutoCloseable {
/// Opens a new [QueryThreadContext] for the current thread and add it to
the thread-local storage.
public static QueryThreadContext open(QueryExecutionContext
executionContext, @Nullable MseWorkerInfo mseWorkerInfo,
ThreadAccountant accountant) {
- QueryThreadContext threadContext = new
QueryThreadContext(executionContext, mseWorkerInfo, accountant);
+ return open(executionContext, mseWorkerInfo, accountant, new
RemoteAccessStats());
+ }
+
+ private static QueryThreadContext open(QueryExecutionContext
executionContext, @Nullable MseWorkerInfo mseWorkerInfo,
+ ThreadAccountant accountant, RemoteAccessStats remoteAccessStats) {
+ QueryThreadContext threadContext =
+ new QueryThreadContext(executionContext, mseWorkerInfo, accountant,
remoteAccessStats);
THREAD_LOCAL.set(threadContext);
accountant.setupTask(threadContext);
return threadContext;
@@ -258,6 +279,35 @@ public class QueryThreadContext implements AutoCloseable {
}
}
+ /// Records remote accesses (local cache misses) and the bytes they read on
the [RemoteAccessStats] of the query
+ /// execution served by the current thread. No-op when the current thread
has no [QueryThreadContext], e.g. when the
+ /// remote data is read to load a segment.
+ public static void addRemoteAccesses(long numRemoteAccesses, long
remoteAccessBytes) {
+ QueryThreadContext threadContext = THREAD_LOCAL.get();
+ if (threadContext != null) {
+ threadContext._remoteAccessStats.addRemoteAccesses(numRemoteAccesses,
remoteAccessBytes);
+ }
+ }
+
+ /// Records direct remote accesses, i.e. reads that went to the remote
storage tier directly instead of through the
+ /// buffers fetched for the query, and the bytes they read on the
[RemoteAccessStats] of the query execution served
+ /// by the current thread. No-op when the current thread has no
[QueryThreadContext].
+ public static void addDirectRemoteAccesses(long numDirectRemoteAccesses,
long directRemoteAccessBytes) {
+ QueryThreadContext threadContext = THREAD_LOCAL.get();
+ if (threadContext != null) {
+
threadContext._remoteAccessStats.addDirectRemoteAccesses(numDirectRemoteAccesses,
directRemoteAccessBytes);
+ }
+ }
+
+ /// Merges the given stats, e.g. the ones of a released fetch, into the
[RemoteAccessStats] of the query execution
+ /// served by the current thread. No-op when the current thread has no
[QueryThreadContext].
+ public static void mergeRemoteAccessStats(RemoteAccessStats
remoteAccessStats) {
+ QueryThreadContext threadContext = THREAD_LOCAL.get();
+ if (threadContext != null) {
+ threadContext._remoteAccessStats.merge(remoteAccessStats);
+ }
+ }
+
/// Returns a new [ExecutorService] whose tasks will be executed with the
[QueryThreadContext] initialized with the
/// state of the thread submitting the tasks. Tasks are registered for
cancellation when the query is terminated.
public static ExecutorService contextAwareExecutorService(ExecutorService
executorService) {
@@ -281,9 +331,11 @@ public class QueryThreadContext implements AutoCloseable {
// the thread context might not be set up.
QueryThreadContext parentThreadContext = registerTaskForCancellation ?
get() : getIfAvailable();
if (parentThreadContext != null) {
+ // Captured at submission, so that the task reports into the
execution that submitted it
+ RemoteAccessStats remoteAccessStats =
parentThreadContext._remoteAccessStats;
return () -> {
try (QueryThreadContext ignore =
open(parentThreadContext._executionContext,
- parentThreadContext._mseWorkerInfo,
parentThreadContext._accountant)) {
+ parentThreadContext._mseWorkerInfo,
parentThreadContext._accountant, remoteAccessStats)) {
return task.call();
}
};
@@ -298,9 +350,11 @@ public class QueryThreadContext implements AutoCloseable {
// the thread context might not be set up.
QueryThreadContext parentThreadContext = registerTaskForCancellation ?
get() : getIfAvailable();
if (parentThreadContext != null) {
+ // Captured at submission, so that the task reports into the
execution that submitted it
+ RemoteAccessStats remoteAccessStats =
parentThreadContext._remoteAccessStats;
return () -> {
try (QueryThreadContext ignore =
open(parentThreadContext._executionContext,
- parentThreadContext._mseWorkerInfo,
parentThreadContext._accountant)) {
+ parentThreadContext._mseWorkerInfo,
parentThreadContext._accountant, remoteAccessStats)) {
task.run();
}
};
diff --git
a/pinot-spi/src/main/java/org/apache/pinot/spi/query/RemoteAccessStats.java
b/pinot-spi/src/main/java/org/apache/pinot/spi/query/RemoteAccessStats.java
new file mode 100644
index 00000000000..34acdc8559c
--- /dev/null
+++ b/pinot-spi/src/main/java/org/apache/pinot/spi/query/RemoteAccessStats.java
@@ -0,0 +1,75 @@
+/**
+ * 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.spi.query;
+
+import java.util.concurrent.atomic.LongAdder;
+
+
+/// Thread-safe accumulator for the remote accesses (local cache misses)
performed to read segment data from a remote
+/// storage tier, and the bytes they read. The direct remote accesses, i.e.
the reads that went to the remote storage
+/// tier directly instead of through the buffers fetched for the query, are
counted both in the totals and on their
+/// own.
+///
+/// The counts are read independently, so a read concurrent with a record can
observe one without the others.
+public final class RemoteAccessStats {
+ private final LongAdder _numRemoteAccesses = new LongAdder();
+ private final LongAdder _remoteAccessBytes = new LongAdder();
+ private final LongAdder _numDirectRemoteAccesses = new LongAdder();
+ private final LongAdder _directRemoteAccessBytes = new LongAdder();
+
+ /// Records remote accesses and the bytes they read.
+ public void addRemoteAccesses(long numRemoteAccesses, long
remoteAccessBytes) {
+ _numRemoteAccesses.add(numRemoteAccesses);
+ _remoteAccessBytes.add(remoteAccessBytes);
+ }
+
+ /// Records direct remote accesses and the bytes they read, in the totals
and on their own.
+ public void addDirectRemoteAccesses(long numDirectRemoteAccesses, long
directRemoteAccessBytes) {
+ addRemoteAccesses(numDirectRemoteAccesses, directRemoteAccessBytes);
+ _numDirectRemoteAccesses.add(numDirectRemoteAccesses);
+ _directRemoteAccessBytes.add(directRemoteAccessBytes);
+ }
+
+ /// Merges all the counts of the given stats into these.
+ public void merge(RemoteAccessStats remoteAccessStats) {
+ addRemoteAccesses(remoteAccessStats.getNumRemoteAccesses(),
remoteAccessStats.getRemoteAccessBytes());
+
_numDirectRemoteAccesses.add(remoteAccessStats.getNumDirectRemoteAccesses());
+
_directRemoteAccessBytes.add(remoteAccessStats.getDirectRemoteAccessBytes());
+ }
+
+ /// Returns the number of remote accesses recorded, the direct ones included.
+ public long getNumRemoteAccesses() {
+ return _numRemoteAccesses.sum();
+ }
+
+ /// Returns the number of bytes read by the remote accesses recorded, the
direct ones included.
+ public long getRemoteAccessBytes() {
+ return _remoteAccessBytes.sum();
+ }
+
+ /// Returns the number of direct remote accesses recorded.
+ public long getNumDirectRemoteAccesses() {
+ return _numDirectRemoteAccesses.sum();
+ }
+
+ /// Returns the number of bytes read by the direct remote accesses recorded.
+ public long getDirectRemoteAccessBytes() {
+ return _directRemoteAccessBytes.sum();
+ }
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
