mrhhsg commented on code in PR #68610:
URL: https://github.com/apache/doris/pull/68610#discussion_r4280620992
##########
be/src/exec/scan/scanner.cpp:
##########
@@ -87,8 +89,17 @@ Status Scanner::init(RuntimeState* state, const
VExprContextSPtrs& conjuncts) {
return Status::OK();
}
+// Keep the single-use padding loop inline with its projection and LIMIT
accounting.
+// NOLINTNEXTLINE(readability-function-cognitive-complexity)
Status Scanner::get_block_after_projects(RuntimeState* state, Block* block,
bool* eos) {
SCOPED_CONCURRENCY_COUNT(ConcurrencyStatsManager::instance().vscanner_get_block);
+ // Peer scanners may have exhausted the shared LIMIT while this scanner
waited to be
+ // scheduled. Admission does not look at the LIMIT, so the scanner reports
EOS itself with an
+ // empty block instead of reading on; the operator already has enough rows
and cuts the rest.
+ if (_shared_scan_limit &&
_shared_scan_limit->load(std::memory_order_acquire) <= 0) {
+ *eos = true;
Review Comment:
Done in e275f858f1d: the entry check now returns an empty block and leaves
`eos` untouched. The query still ends through the operator side only:
`ScannerContext::get_block_from_queue()` finishes the context once the counter
is exhausted with nothing in flight, and `ScanOperatorX::get_block_impl()`
stops the scanners after `reached_limit()`.
`ScannerProjectionTest.shared_limit_exhausted_before_call_returns_empty_block_without_eos`
(and the other `shared_limit_*` cases) now assert `eos == false`.
##########
be/src/exec/scan/scanner.cpp:
##########
@@ -99,44 +110,61 @@ Status Scanner::get_block_after_projects(RuntimeState*
state, Block* block, bool
// before the next split is read. Padding those blocks first would
make correctness
// depend on whether two file tails happen to share one output
batch.
RETURN_IF_ERROR(get_block(state, &_origin_block, eos));
- return _do_projections(&_origin_block, block);
- }
- const auto min_batch_size = std::max(state->batch_size() / 2, 1);
- const auto block_max_bytes = state->preferred_block_size_bytes();
- while (_padding_block.rows() < min_batch_size &&
_padding_block.bytes() < block_max_bytes &&
- !*eos) {
- RETURN_IF_ERROR(get_block(state, &_origin_block, eos));
- if (*eos) {
- // For the final block, merge any padding directly and return
eos in this call.
- // The merged tail can be larger than the target batch, but
each source block is
- // already bounded by the lower scanner.
- RETURN_IF_ERROR(_merge_padding_block());
- _origin_block.clear_column_data(row_descriptor.num_slots());
- break;
- }
- if (_origin_block.rows() >= min_batch_size) {
- break;
+ } else {
+ auto min_batch_size = std::max(state->batch_size() / 2, 1);
+ if (_shared_scan_limit) {
+ // Pending padding is not charged until it is returned. Bound
this call's batch
+ // target by the shared remaining rows to avoid reading on
just to fill a batch.
+ min_batch_size = cast_set<int>(std::clamp<int64_t>(
+ _shared_scan_limit->load(std::memory_order_acquire),
0, min_batch_size));
}
+ const auto block_max_bytes = state->preferred_block_size_bytes();
+ while (_padding_block.rows() < min_batch_size &&
+ _padding_block.bytes() < block_max_bytes && !*eos) {
+ RETURN_IF_ERROR(get_block(state, &_origin_block, eos));
+ if (*eos) {
+ // For the final block, merge any padding directly and
return eos in this call.
+ // The merged tail can be larger than the target batch,
but each source block
+ // is already bounded by the lower scanner.
+ RETURN_IF_ERROR(_merge_padding_block());
+
_origin_block.clear_column_data(row_descriptor.num_slots());
+ break;
+ }
+ if (_origin_block.rows() >= min_batch_size) {
+ break;
+ }
- if (_origin_block.rows() + _padding_block.rows() <=
state->batch_size() &&
- _origin_block.bytes() + _padding_block.bytes() <=
block_max_bytes) {
- RETURN_IF_ERROR(_merge_padding_block());
- _origin_block.clear_column_data(row_descriptor.num_slots());
- } else {
- if (_origin_block.rows() < _padding_block.rows()) {
- _padding_block.swap(_origin_block);
+ if (_origin_block.rows() + _padding_block.rows() <=
state->batch_size() &&
+ _origin_block.bytes() + _padding_block.bytes() <=
block_max_bytes) {
+ RETURN_IF_ERROR(_merge_padding_block());
+
_origin_block.clear_column_data(row_descriptor.num_slots());
+ } else {
+ if (_origin_block.rows() < _padding_block.rows()) {
+ _padding_block.swap(_origin_block);
+ }
+ break;
}
- break;
}
- }
- if (_origin_block.empty() && !_padding_block.empty()) {
- _padding_block.swap(_origin_block);
+ if (_origin_block.empty() && !_padding_block.empty()) {
+ _padding_block.swap(_origin_block);
+ }
}
- return _do_projections(&_origin_block, block);
+ RETURN_IF_ERROR(_do_projections(&_origin_block, block));
} else {
- return get_block(state, block, eos);
+ RETURN_IF_ERROR(get_block(state, block, eos));
}
+
+ // Only rows actually produced by this call consume the shared LIMIT. In
particular, padding
+ // kept for a later call must not make the context finish before that call
can return it.
+ if (_shared_scan_limit) {
+ if (block->rows() > 0) {
+ _shared_scan_limit->fetch_sub(cast_set<int64_t>(block->rows()),
+ std::memory_order_acq_rel);
+ }
+ *eos = *eos || _shared_scan_limit->load(std::memory_order_acquire) <=
0;
Review Comment:
Done in e275f858f1d: the tail only charges the produced rows and no longer
derives `*eos` from the counter. For consistency, the two pre-existing checks
in `Scanner::get_block()` that set `eof` from the shared counter are removed as
well, so the shared LIMIT is handled only in `get_block_after_projects()`:
return an empty block on entry when it is exhausted, stop the padding loop
(returning the rows already held) when peers exhaust it mid-read, and charge
the produced rows at the end. EOS now comes only from the underlying read (or
the per-scanner `_limit` / stop flag, as on master). Added
`shared_limit_exhausted_by_own_rows_stops_reading_without_eos` and
`shared_limit_exhausted_by_peer_mid_read_returns_held_rows_without_eos`; the
comment in `_pull_next_scan_task()` about pending scanners "reporting EOS" is
updated accordingly.
##########
be/src/exec/scan/scanner.cpp:
##########
@@ -101,6 +108,16 @@ Status Scanner::get_block_after_projects(RuntimeState*
state, Block* block, bool
RETURN_IF_ERROR(get_block(state, &_origin_block, eos));
return _do_projections(&_origin_block, block);
}
+ // get_block() charges the shared LIMIT as soon as rows pass the
filters, but rows that stay
+ // in _padding_block when this call returns are not returned by it.
Peer scanners may
+ // exhaust the LIMIT before this scanner runs again, and the context
then finishes without
+ // running it, so those rows would be charged but never returned. Keep
the counter equal
+ // to "LIMIT minus returned rows" between calls: charge the held rows
again on entry, so
+ // every row this scanner holds during the call is charged and the
read loop below stops
+ // as soon as the LIMIT is reached, and refund whatever is still held
on exit.
+ if (_shared_scan_limit && !_padding_block.empty()) {
Review Comment:
This line is gone since `[fix](be) Count scan LIMIT after producing output
rows`: the shared counter is now decremented only by the rows a call actually
returns (at the end of `get_block_after_projects()`), so `_padding_block` rows
are never charged on entry nor refunded on exit. `_padding_block` can
legitimately be non-empty between calls (a small block held for coalescing); it
simply does not touch the counter until it is returned. Resolving as addressed
at e275f858f1d.
--
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]
