github-actions[bot] commented on code in PR #67140:
URL: https://github.com/apache/doris/pull/67140#discussion_r3861792961
##########
be/src/exec/operator/distinct_streaming_aggregation_operator.cpp:
##########
@@ -176,8 +195,17 @@ 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));
+ // A LIMIT pushed down to this stage bounds the hash set by itself and
must keep
+ // deduplicating: pass-through rows would otherwise consume the limit
with duplicates
+ // (see push()).
+ if (low_memory_mode() ||
+ (parent.limit() == -1 && memory_limit > 0 && _memory_usage() >
memory_limit)) {
Review Comment:
[P1] Keep the byte cap active when a LIMIT is pushed down. The LIMIT is
enforced only after `_emplace_into_hash_table_to_distinct()` has inserted the
entire block, so it does not bound the retained hash set or arena: with `LIMIT
1` and one block of many unique large strings, `lazy_emplace_batch_void()`
copies every key before `push()` truncates the output, and this non-spillable
state can exhaust the query tracker despite `MemoryUseLimit`. Let the existing
pass-through/no-local-truncation logic handle a cap breach (or stop emplacement
at the remaining limit/budget), and cover variable-length keys.
##########
be/src/exec/operator/streaming_aggregation_operator.cpp:
##########
@@ -1106,14 +1130,18 @@ Status StreamingAggOperatorX::pull(RuntimeState* state,
Block* block, bool* eos)
auto& local_state = get_local_state(state);
SCOPED_PEAK_MEM(&local_state._estimate_memory_usage);
if (!local_state._pre_aggregated_block->empty()) {
+ // Pass-through rows are neither aggregated nor deduplicated, so a
limit pushed down to
+ // this local stage must not count them: duplicates would consume the
allowance and stop
+ // the child before enough distinct keys reached the global stage,
which applies the
+ // limit again on the final result.
local_state._pre_aggregated_block->swap(*block);
} else {
RETURN_IF_ERROR(local_state._get_results_with_serialized_key(state,
block, eos));
local_state.make_nullable_output_key(block);
// dispose the having clause, should not be execute in prestreaming agg
RETURN_IF_ERROR(local_state.filter_block(local_state._conjuncts,
block));
+ local_state.reached_limit(block, eos);
Review Comment:
[P2] Preserve process-row accounting for the pass-through branch.
`reached_limit()` also increments the query `IOContext::process_rows`; after
moving it only here, every raw block returned above for a cap or low-reduction
fallback disappears from `TQueryStatistics.process_rows`. Split that accounting
from local LIMIT truncation, or explicitly update `process_rows` for
pass-through rows, and extend the recovery test to assert it while keeping `eos
== false`.
##########
be/src/exec/operator/distinct_streaming_aggregation_operator.cpp:
##########
@@ -67,9 +72,23 @@ Status DistinctStreamingAggLocalState::open(RuntimeState*
state) {
RETURN_IF_ERROR(p._probe_expr_ctxs[i]->clone(state,
_probe_expr_ctxs[i]));
}
RETURN_IF_ERROR(_init_hash_method(_probe_expr_ctxs));
+ COUNTER_SET(_memory_use_limit,
static_cast<int64_t>(p._memory_limit(state)));
return Status::OK();
}
+size_t DistinctStreamingAggLocalState::_memory_usage() const {
+ size_t usage = _arena.size();
Review Comment:
[P1] Count the hash method's retained serialization scratch here.
Composite-key probing fills `MethodSerialized::Base::arena`, `stored_keys`, and
`hash_values` for the whole block before deduplication, and `Arena::clear()`
retains its largest head chunk. A duplicate-heavy 4096-row block with roughly 2
KiB serialized keys can therefore leave about 8 MiB retained while the
persistent distinct set contains one key; this helper still reports below a 4
MiB `MemoryUseLimit`, so later blocks keep aggregating. Include the method's
retained buffers (or release oversized scratch) and cover duplicate composite
keys.
--
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]
