This is an automated email from the ASF dual-hosted git repository.
yashmayya 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 2f89dab09e7 Don't use the streaming group-by leaf when the server must
return final results (#19417)
2f89dab09e7 is described below
commit 2f89dab09e7096a5a42cdc4d1d74f035c5f51b5a
Author: Yash Mayya <[email protected]>
AuthorDate: Mon Aug 31 19:46:03 2026 -0400
Don't use the streaming group-by leaf when the server must return final
results (#19417)
---
.../streaming/StreamingGroupByCombineOperator.java | 8 ++++
.../apache/pinot/core/plan/CombinePlanNode.java | 10 +++--
.../StreamingGroupByCombineOperatorTest.java | 43 ++++++++++++++++++++++
3 files changed, 58 insertions(+), 3 deletions(-)
diff --git
a/pinot-core/src/main/java/org/apache/pinot/core/operator/streaming/StreamingGroupByCombineOperator.java
b/pinot-core/src/main/java/org/apache/pinot/core/operator/streaming/StreamingGroupByCombineOperator.java
index 0ad10b1325d..63aeeeb38aa 100644
---
a/pinot-core/src/main/java/org/apache/pinot/core/operator/streaming/StreamingGroupByCombineOperator.java
+++
b/pinot-core/src/main/java/org/apache/pinot/core/operator/streaming/StreamingGroupByCombineOperator.java
@@ -18,6 +18,7 @@
*/
package org.apache.pinot.core.operator.streaming;
+import com.google.common.base.Preconditions;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Iterator;
@@ -55,6 +56,10 @@ import org.apache.pinot.spi.query.QueryThreadContext;
///
/// - Hash exchange routes the same group key to the same FINAL worker
/// - AggregationFunction.merge() is associative
+///
+/// That FINAL stage is a precondition, not an implementation detail: a
flushed block carries only a partial
+/// aggregate, and one group key can span several flush windows. Leaves that
must return final results are rejected
+/// below; finalizing each flush instead would not make them correct, only
silent.
@SuppressWarnings({"rawtypes", "unchecked"})
public class StreamingGroupByCombineOperator extends
BaseStreamingCombineOperator<GroupByResultsBlock> {
private static final String EXPLAIN_NAME = "STREAMING_COMBINE_GROUP_BY";
@@ -74,6 +79,9 @@ public class StreamingGroupByCombineOperator extends
BaseStreamingCombineOperato
public StreamingGroupByCombineOperator(List<Operator> operators,
QueryContext queryContext,
ExecutorService executorService, int flushThreshold) {
super(null, operators, overrideMaxExecutionThreads(queryContext,
operators.size()), executorService);
+ Preconditions.checkState(
+ !queryContext.isServerReturnFinalResult() &&
!queryContext.isServerReturnFinalResultKeyUnpartitioned(),
+ "Streaming group-by combine requires a leaf that emits INTERMEDIATE
results");
_flushThreshold = flushThreshold;
AggregationFunction[] aggregationFunctions =
_queryContext.getAggregationFunctions();
diff --git
a/pinot-core/src/main/java/org/apache/pinot/core/plan/CombinePlanNode.java
b/pinot-core/src/main/java/org/apache/pinot/core/plan/CombinePlanNode.java
index 4ad61fe3a0d..dd3a720412b 100644
--- a/pinot-core/src/main/java/org/apache/pinot/core/plan/CombinePlanNode.java
+++ b/pinot-core/src/main/java/org/apache/pinot/core/plan/CombinePlanNode.java
@@ -130,10 +130,14 @@ public class CombinePlanNode implements PlanNode {
// Use streaming operator only for non-empty selection-only query
return new StreamingSelectionOnlyCombineOperator(operators,
_queryContext, _executorService);
}
+ // Streaming flushes partial aggregates, so it needs an aggregation
above to merge them back together.
+ // Leaves that must return final results are excluded, see
StreamingGroupByCombineOperator.
int flushThreshold = _queryContext.getStreamingGroupByFlushThreshold();
- if (flushThreshold > 0 &&
QueryContextUtils.isAggregationQuery(_queryContext)
- && _queryContext.getGroupByExpressions() != null) {
- // Use streaming group-by operator for MSE leaf stages with flush
threshold
+ if (flushThreshold > 0
+ && QueryContextUtils.isAggregationQuery(_queryContext)
+ && _queryContext.getGroupByExpressions() != null
+ && !_queryContext.isServerReturnFinalResult()
+ && !_queryContext.isServerReturnFinalResultKeyUnpartitioned()) {
return new StreamingGroupByCombineOperator(operators, _queryContext,
_executorService, flushThreshold);
}
}
diff --git
a/pinot-core/src/test/java/org/apache/pinot/core/operator/streaming/StreamingGroupByCombineOperatorTest.java
b/pinot-core/src/test/java/org/apache/pinot/core/operator/streaming/StreamingGroupByCombineOperatorTest.java
index 7ae970f95cb..abb6bcf8e72 100644
---
a/pinot-core/src/test/java/org/apache/pinot/core/operator/streaming/StreamingGroupByCombineOperatorTest.java
+++
b/pinot-core/src/test/java/org/apache/pinot/core/operator/streaming/StreamingGroupByCombineOperatorTest.java
@@ -33,6 +33,8 @@ import org.apache.pinot.core.common.Operator;
import org.apache.pinot.core.operator.blocks.results.BaseResultsBlock;
import org.apache.pinot.core.operator.blocks.results.GroupByResultsBlock;
import org.apache.pinot.core.operator.blocks.results.MetadataResultsBlock;
+import org.apache.pinot.core.operator.combine.BaseCombineOperator;
+import org.apache.pinot.core.plan.CombinePlanNode;
import org.apache.pinot.core.plan.PlanNode;
import org.apache.pinot.core.plan.maker.InstancePlanMakerImplV2;
import org.apache.pinot.core.plan.maker.PlanMaker;
@@ -72,6 +74,8 @@ public class StreamingGroupByCombineOperatorTest {
private static final int NUM_RECORDS_PER_SEGMENT = 100;
private static final int NUM_DISTINCT_GROUPS = 50;
+ private static final String GROUP_BY_SUM = "SELECT groupColumn,
SUM(intColumn) FROM testTable GROUP BY groupColumn";
+
private static final String GROUP_COLUMN = "groupColumn";
private static final String INT_COLUMN = "intColumn";
private static final TableConfig TABLE_CONFIG =
@@ -434,6 +438,45 @@ public class StreamingGroupByCombineOperatorTest {
assertEquals(groupSums.get(null), NUM_SEGMENTS * 2550.0, 0.001, "Incorrect
rollup total");
}
+ /// A leaf asked to return FINAL results must not stream: a flushed block
holds only a partial aggregate, and one
+ /// group key can span several flush windows. `serverReturnFinalResult`
comes from an `AggType.DIRECT` leaf
+ /// (`is_partitioned_by_group_by_keys`), which has no aggregation above it
to merge the pieces back together.
+ @Test
+ public void testServerReturnFinalResultDoesNotUseStreamingCombine() {
+ assertFallsBackToNonStreamingCombine("SET serverReturnFinalResult=true; "
+ GROUP_BY_SUM);
+ }
+
+ /// Same for `serverReturnFinalResultKeyUnpartitioned`
(`is_leaf_return_final_result`): a FINAL stage does sit
+ /// above, but it merges FINAL results, which double-counts across flushes
for functions whose mergeFinalResult
+ /// accumulates (e.g. DISTINCTCOUNT sums its inputs).
+ @Test
+ public void
testServerReturnFinalResultKeyUnpartitionedDoesNotUseStreamingCombine() {
+ assertFallsBackToNonStreamingCombine("SET
serverReturnFinalResultKeyUnpartitioned=true; " + GROUP_BY_SUM);
+ }
+
+ @Test
+ public void testPlainLeafUsesStreamingCombine() {
+ assertEquals(route(GROUP_BY_SUM, 10).getClass(),
StreamingGroupByCombineOperator.class);
+ }
+
+ /// Asserts that a flush threshold picks exactly the operator the same query
gets without one.
+ private void assertFallsBackToNonStreamingCombine(String query) {
+ assertEquals(route(query, 10).getClass(), route(query, 0).getClass(),
+ "A leaf asked to return FINAL results must be combined as if no flush
threshold were set");
+ }
+
+ /// Runs [CombinePlanNode] with a streamer attached, as an MSE leaf stage
does.
+ private BaseCombineOperator<?> route(String query, int flushThreshold) {
+ QueryContext queryContext =
QueryContextConverterUtils.getQueryContext(query);
+ queryContext.setEndTimeMs(System.currentTimeMillis() +
Server.DEFAULT_QUERY_EXECUTOR_TIMEOUT_MS);
+ queryContext.setStreamingGroupByFlushThreshold(flushThreshold);
+ List<PlanNode> planNodes = new ArrayList<>(NUM_SEGMENTS);
+ for (IndexSegment indexSegment : _indexSegments) {
+ planNodes.add(PLAN_MAKER.makeSegmentPlanNode(new
SegmentContext(indexSegment), queryContext));
+ }
+ return new CombinePlanNode(planNodes, queryContext, EXECUTOR, block -> {
}).run();
+ }
+
private List<Operator> buildOperators(QueryContext queryContext) {
List<Operator> operators = new ArrayList<>(NUM_SEGMENTS);
for (IndexSegment indexSegment : _indexSegments) {
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
