yashmayya commented on code in PR #19816:
URL: https://github.com/apache/pinot/pull/19816#discussion_r4264834096
##########
pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/MultiNodesOfflineClusterIntegrationTest.java:
##########
@@ -326,6 +326,35 @@ public void testConstantExpressionQuery()
assertEquals(result.get("numServersQueried").intValue(), 1);
}
+ @Test
+ public void testMultiStageTotalMemAllocatedBytes()
+ throws Exception {
+ setUseMultiStageQueryEngine(true);
+
+ // The semi-join query uses a pipeline breaker
+ for (String query : List.of("SELECT Carrier, COUNT(*) FROM mytable GROUP
BY Carrier",
+ "SELECT Carrier FROM mytable WHERE Origin IN (SELECT Origin FROM
mytable WHERE ArrDelay > 0) LIMIT 10")) {
+ JsonNode result = postQuery(query);
+ long totalMemAllocatedBytes =
result.path("totalMemAllocatedBytes").asLong();
+ assertTrue(totalMemAllocatedBytes > 0);
+ assertEquals(totalMemAllocatedBytes,
sumAllocatedBytes(result.get("stageStats"), true));
+ }
+ }
+
+ /// Sums the allocation of each op chain root (the top node, each
MAILBOX_SEND and each pipeline breaker, which is
+ /// rendered as the child of its LEAF) and of the single-stage threads of
each LEAF.
+ private static long sumAllocatedBytes(JsonNode node, boolean isOpChainRoot) {
+ long bytes = node.path("threadMemAllocatedBytes").asLong();
+ if (isOpChainRoot) {
+ bytes += node.path("allocatedMemoryBytes").asLong();
+ }
+ boolean isLeaf = node.path("type").asText().equals("LEAF");
+ for (JsonNode child : node.path("children")) {
+ bytes += sumAllocatedBytes(child, isLeaf ||
child.path("type").asText().equals("MAILBOX_SEND"));
Review Comment:
Thanks for checking this closely. I verified it locally. In the default
(legacy) stats mode, the node under `LEAF` carries the breaker's own stats.
Only its label is `MAILBOX_RECEIVE`.
- `InStageStatsTreeBuilder.extractPipelineBreakerResult()` moves `_index` to
the `PIPELINE_BREAKER` entry. Then `visitMailboxReceive()` calls
`selfNode(MAILBOX_RECEIVE, context)`, which copies
`_stageStats.getOperatorStats(_index)`, that is, the breaker's stat map. The
breaker's receive operator is not rendered.
- I ran this test with temporary logging of the raw leaf-stage stats. The
allocated bytes were `MAILBOX_RECEIVE=14256`, `PIPELINE_BREAKER=19944`,
`LEAF=2232` and `MAILBOX_SEND=136616`. The JSON node under `LEAF` had
`allocatedMemoryBytes=19944`, with only the breaker's keys (`executionTimeMs`,
`emittedRows`, `allocatedMemoryBytes`) and no `fanIn` or `upstreamWaitMs`. The
exact equality held.
- When I removed the `PIPELINE_BREAKER` addition in `mergeInto`, the test
failed by exactly the bytes of that node.
Your point is correct for stream mode:
`MultiStageStatsTreeBuilder.jsonFromStatsTreeNode()` collapses the breaker and
shows the stats of the receive. So the test now sets `streamStats = false`, and
the helper's Javadoc states this.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
