HappenLee commented on code in PR #67140:
URL: https://github.com/apache/doris/pull/67140#discussion_r4023559414
##########
be/src/exec/operator/streaming_aggregation_operator.cpp:
##########
@@ -342,12 +350,19 @@ Status
StreamingAggLocalState::_pre_agg_with_serialized_key(doris::Block* in_blo
_places.resize(rows);
if (_should_not_do_pre_agg(rows)) {
+ // Serializing a row as a single-row state may allocate from the arena
(collect, map,
+ // foreach, ...). Those bytes are dead once the state is serialized,
so they must not
+ // accumulate in the operator-lifetime `_agg_arena_pool` for every
pass-through block.
+ Arena pass_through_arena;
if (limit > 0) {
DCHECK(do_sort_limit);
if (need_do_sort_limit == -1) {
+ // Only latch once the hash table can seed the heap. A smaller
table keeps the
+ // state undecided so that aggregation resumed after a
transient pass-through
Review Comment:
[P1] Preserve distinct TopN candidates across pass-through and recovery
The memory-triggered pass-through/recovery path can make the TopN heap count
the same group twice and then discard a group that belongs in the result. For a
plan shaped like `TopN(k ASC, 2) -> global aggregation -> exchange -> streaming
aggregation`, consider these input blocks:
| Step | Aggregation hash table | TopN candidates | Boundary |
| --- | --- | --- | --- |
| Aggregate `[10, 20]` | `{10, 20}` | `[10, 20]` | `20` |
| Lower the budget; pass through `[5, 30]` | `{10, 20}` | `[5, 10]` | `10` |
| Restore the budget; aggregate `[5]` | `{5, 10, 20}` | `[5, 5]` | `5` |
| Read `[7]` | Unchanged: `{5, 10, 20}` | `[5, 5]` | `5`; key `7` is
discarded |
The mixed block `[5, 30]` matters: it enters the filtering branch, where
[`_add_limit_heap_top()`](https://github.com/apache/doris/blob/6ef7d2447ef526d679e931595d01e4b01c0c7649/be/src/exec/operator/streaming_aggregation_operator.cpp#L710)
inserts `5` into the candidate heap without inserting it into the aggregation
hash table. On recovery, `5` is new to the hash table, so the [new-group
callback](https://github.com/apache/doris/blob/6ef7d2447ef526d679e931595d01e4b01c0c7649/be/src/exec/operator/streaming_aggregation_operator.cpp#L761)
invokes `_refresh_limit_heap()` and inserts a second `5` into the heap.
Neither heap update checks candidate uniqueness. The boundary now represents
two rows of one group, rather than two distinct groups.
For `SELECT k, SUM(v) FROM t GROUP BY k ORDER BY k LIMIT 2`, the expected
keys in this example are `5, 7`; the downstream stage cannot recover key `7`
once the local stage has discarded it.
I reproduced the operator-level failure with a temporary BE unit test using
the production `_should_not_do_pre_agg()` decision, a pushed TopN of 2, and
real `sum` serialization. The test changes the mock query limit from `5000 MiB`
to `10 bytes` and back with 5 tasks to force the transition deterministically.
It observes:
```text
After recovery: expected heap boundary 10, actual 5.
After reading 7: expected hash-table size 4, actual 3.
15 filtered tests ran: the PR's existing 14 passed; the added reproducer
failed.
```
Command: `./run-be-ut.sh -j 48 --run
--filter='StreamingAggMemoryLimitTest.*:StreamingAggOperatorTest.*:DistinctStreamingAggOperatorTest.*'`.
This validates the local filtering failure; I have not run an end-to-end SQL
reproducer.
The non-deduplicating heap code predates this PR, but this PR makes
memory-triggered pass-through and recovery reachable for additional non-spill
queries. Please cover this interaction before merging. A simple fix to consider
is to keep using the existing conservative TopN boundary during pass-through,
without tightening it from raw, non-deduplicated rows; normal hash-table
insertion can update the boundary after aggregation resumes. This avoids adding
another deduplication structure.
##########
be/src/exec/operator/distinct_streaming_aggregation_operator.cpp:
##########
@@ -176,8 +195,13 @@ Status
DistinctStreamingAggLocalState::_distinct_pre_agg_with_serialized_key(
const uint32_t rows = (uint32_t)in_block->rows();
_distinct_row.clear();
- if (_parent->cast<DistinctStreamingAggOperatorX>()._is_streaming_preagg &&
low_memory_mode()) {
- _stop_emplace_flag = true;
+ auto& parent = _parent->cast<DistinctStreamingAggOperatorX>();
+ if (parent._is_streaming_preagg) {
+ const auto memory_limit = parent._memory_limit(state());
+ COUNTER_SET(_memory_use_limit, static_cast<int64_t>(memory_limit));
+ if (low_memory_mode() || (memory_limit > 0 && _memory_usage() >
memory_limit)) {
+ _stop_emplace_flag = true;
Review Comment:
[P2] A temporary budget reduction can permanently disable distinct
pre-aggregation
Following up on this thread against the current head,
`6ef7d2447ef526d679e931595d01e4b01c0c7649`: the memory condition still
permanently sets `_stop_emplace_flag`, and subsequent blocks never resume
deduplication even if the live budget is restored above the retained usage.
I understand the simplicity argument for retaining the existing latch.
However, a budget reduction does not always coincide with entering low-memory
mode. For example, with the DYNAMIC slot-memory policy and hard limits enabled,
an increase in active query slots can reduce this query's weighted limit while
the workload group remains below its low watermark. When the other queries
finish, the limit can recover without this query ever having entered low-memory
mode. The current latch still makes the remainder of the query pass through
every raw key.
For a long, duplicate-heavy input, a short period of slot contention can
therefore cause substantially more exchange traffic and downstream aggregation
work for the rest of the query, while the ordinary streaming aggregate can
resume after the same budget recovery. This is a performance/state-transition
concern, not a wrong-result claim.
Please consider keeping a temporary memory-budget stop separate from the
permanent low-reduction/low-memory stops. If the permanent memory latch is
intentional, please document the recovery asymmetry explicitly and add a
lower-then-restore test plus a representative duplicate-heavy benchmark so the
tradeoff is measurable. The current `refresh_memory_limit` test stops after the
budget is lowered and does not check restoration.
--
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]
