yashmayya opened a new pull request, #19417: URL: https://github.com/apache/pinot/pull/19417
### Problem `streamingGroupByFlushThreshold` (added in #18035) routes every MSE group-by leaf to `StreamingGroupByCombineOperator`, with no check on whether that leaf is allowed to emit partial results. The streaming leaf flushes a partial aggregate whenever its table reaches the threshold, so **one group key can be emitted from several flush windows**. That is sound for the plain LEAF + EXCHANGE + FINAL shape it was written for — the FINAL stage merges the pieces back together. It is not sound when the leaf itself has to produce final values: - `is_partitioned_by_group_by_keys` makes the leaf aggregate `AggType.DIRECT`, which sets `serverReturnFinalResult` and leaves **no aggregation above it at all**. - `is_leaf_return_final_result` sets `serverReturnFinalResultKeyUnpartitioned`. A FINAL stage does sit above, but it merges *final* results. Symptoms, all reproduced by the tests in this PR (3 identical segments, `SET streamingGroupByFlushThreshold=1`): | Query | Result | |---|---| | `/*+ aggOptions(is_partitioned_by_group_by_keys='true') */ col1, COUNT(*)` | **Silently wrong** — 15 rows of `1` instead of 5 rows of `3` | | `/*+ aggOptions(is_partitioned_by_group_by_keys='true') */ col1, AVG(col3), COUNT(DISTINCT col3)` | `ClassCastException: IntOpenHashSet cannot be cast to Number` | | `/*+ aggOptions(is_leaf_return_final_result='true') */ col1, COUNT(DISTINCT col3)` | Same `ClassCastException` | The `ClassCastException` is the raw intermediate escaping the leaf: the flushed block's schema types stay `OBJECT` where the multi-stage plan expects `LONG`/`DOUBLE`. ### Why not just finalize each flush `flushTable()` calls `_indexedTable.finish(false)` unconditionally, where `GroupByCombineOperator.mergeResults()` branches to `finish(true, true)` / `finish(false, true)` for the same two flags. Mirroring that branching looks like the fix, and it does make both `ClassCastException`s go away — but it replaces them with silent wrong answers: - `is_leaf_return_final_result` + `COUNT(DISTINCT)` then returns `3` instead of `1`, because `DistinctCountAggregationFunction.mergeFinalResult` sums its inputs, so each flush window's partial count is double-counted. - `is_partitioned_by_group_by_keys` still returns duplicate rows, since nothing above merges them. A final result computed per flush window is not the group's final result. The precondition is structural, so the fix is to not stream at all in these cases. ### Fix Exclude both flags in `CombinePlanNode`, so these leaves fall back to `GroupByCombineOperator` exactly as they would without a flush threshold. The `serverReturnFinalResult` check also covers single-server SSE queries, where the broker sets the flag automatically and the gRPC path supplies a streamer. `StreamingGroupByCombineOperator`'s constructor now asserts the precondition it documents, so a future second construction site fails loudly instead of returning wrong results. ### Tests - `StreamingGroupByCombineOperatorTest` — three routing tests: both flags fall back to the non-streaming operator, plain leaves still stream. - `StreamingGroupByLeafFinalResultTest` (new) — end-to-end through the multi-stage engine, asserting the exact expected rows with and without the flush threshold, for both hints plus a plain-group-by control. A fifth test pins that the option actually reaches the leaf and makes it flush more than once, so the other four cannot silently degrade into comparing two identical non-streaming runs. Both sets fail without the fix, with the symptoms in the table above. `QueryRunnerTest`, `CombinePlanNodeTest` and `SortedGroupByCombineOperatorsTest` pass unchanged. No wire, config, or serialization change. The fix is server-side and plan-time only, so servers still on an older build keep returning wrong answers for these queries during a rolling upgrade. -- 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]
