HappenLee commented on code in PR #67140:
URL: https://github.com/apache/doris/pull/67140#discussion_r4024803999
##########
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:
Please address this distinct recovery path before merging. At the current
head, `7d172fc7fcca0976466ef044672014b92b271d1b`, a temporary budget breach
still permanently sets `_stop_emplace_flag`, so restoring the live budget
cannot restore local deduplication.
The workload-group low-memory switch does not cover every case. With the
DYNAMIC slot-memory policy and hard limits enabled, additional active query
slots can reduce this query's budget while the workload group remains below its
low watermark. When those queries finish, the budget can recover without
low-memory mode ever being set.
For example, with 8 parallel tasks and a retained hash set using 64 MiB per
task:
| Query memory limit | Per-task pre-aggregation budget | Current behavior |
| --- | --- | --- |
| 8 GiB | About 204.8 MiB | Deduplicate normally |
| Temporarily reduced to 1 GiB | 32 MiB | Set the permanent stop flag and
pass through |
| Restored to 8 GiB | About 204.8 MiB | Still pass through, although the
retained state is below budget |
For a long, duplicate-heavy query, this turns a temporary reduction into
additional exchange traffic and downstream work for the entire remaining
execution. Raising the budget or its floor only reduces the chance of
triggering the issue; it does not fix recovery.
Please separate the temporary, per-block budget decision from the existing
permanent stops for low reduction and low-memory mode. Preserve the retained
hash set and resume deduplication when the budget permits, provided no
permanent stop was triggered. Do not simply clear `_stop_emplace_flag` on
recovery, since it also represents those other stop reasons.
Please also carry the effective pass-through state consistently through
output/buffering and pushed-down LIMIT handling. In particular, raw rows
emitted during the temporary pass-through must not later consume the distinct
LIMIT allowance when deduplication resumes: `pull()` currently increments
`_num_rows_returned` for every output row, and `push()` consults that counter
when enforcing the limit.
Add lower-then-restore tests, both with and without a pushed-down LIMIT,
that verify:
- The low-budget block passes through without growing the hash set.
- After restoration, previously retained keys are filtered again and new
keys can be inserted.
- Earlier pass-through duplicates do not cause premature LIMIT/EOS after
recovery.
- Stops caused by low reduction or low-memory mode retain their existing
semantics.
This remains a performance/recovery issue in the current code. The LIMIT
points above are requirements for implementing recovery without reintroducing
the earlier correctness bug.
--
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]
