yiguolei commented on code in PR #68610:
URL: https://github.com/apache/doris/pull/68610#discussion_r4279668911
##########
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:
不要在这里处理eos 的逻辑。 eos 的含义表示的是底层读取返回的eos，跟limit 不一样。 limit 应该是scan operator
那层来处理的
--
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]
