github-actions[bot] commented on code in PR #65324:
URL: https://github.com/apache/doris/pull/65324#discussion_r3585615561
##########
be/src/exec/operator/data_queue.cpp:
##########
@@ -154,70 +157,70 @@ void DataQueue::clear_free_blocks() {
void DataQueue::terminate() {
for (int i = 0; i < _child_count; ++i) {
- set_finish(i);
+ mark_finish(i);
_sub_queues[i]->clear_blocks();
}
+ _cur_blocks_total_nums = 0;
clear_free_blocks();
+ set_source_ready();
}
-//check which queue have data, and save the idx in _flag_queue_idx,
-//so next loop, will check the record idx + 1 first
-//maybe it's useful with many queue, others maybe always 0
-bool DataQueue::remaining_has_data() {
- int count = _child_count;
- while (--count >= 0) {
- _flag_queue_idx++;
- if (_flag_queue_idx == _child_count) {
- _flag_queue_idx = 0;
- }
- if (_sub_queues[_flag_queue_idx]->blocks_in_queue.load() > 0) {
- return true;
+Result<DataQueueBlock> DataQueue::get_block_from_queue() {
+ DataQueueBlock result;
+ const int start_idx = (_flag_queue_idx + 1) % _child_count;
+ for (int offset = 0; offset < _child_count; ++offset) {
+ const int idx = (start_idx + offset) % _child_count;
+ if (_sub_queues[idx]->blocks_in_queue.load() == 0) {
+ continue;
}
- }
- return false;
-}
-//the _flag_queue_idx indicate which queue has data, and in check can_read
-//will be set idx in remaining_has_data function
-Status DataQueue::get_block_from_queue(std::unique_ptr<Block>* output_block,
int* child_idx) {
- const int idx = _flag_queue_idx;
- auto& sub = *_sub_queues[idx];
-
- sub.try_pop(output_block);
- if (*output_block) {
- if (child_idx) {
- *child_idx = idx;
+ auto& sub = *_sub_queues[idx];
+ sub.try_pop(&result.block);
+ if (!result.block) {
+ continue;
}
+ result.child_idx = idx;
+ _flag_queue_idx = idx;
auto old_total = _cur_blocks_total_nums.fetch_sub(1);
if (old_total == 1) {
set_source_block();
}
+ break;
}
- return Status::OK();
+
Review Comment:
This EOS check can race with a final producer. `&&` evaluates left to right,
so a source that was woken by one child finishing can scan all subqueues as
empty and read `has_more_data()` before another child executes
`push_block(block, child, true)`. That child then enqueues the block, marks all
children finished, and this line can read `is_all_finish()` as true, returning
`eos=true` with `block == nullptr` while the final block is queued.
`UnionSourceOperatorX` trusts that flag and returns immediately on the null
block, so the pipeline can close/terminate the queue without ever emitting
those rows. The old Union-side logic did a data recheck after observing
`is_all_finish()`. Please preserve that ordering here, for example by checking
`is_all_finish()` first and then rechecking `has_more_data()` before reporting
EOS, or otherwise serializing finish observation with enqueue visibility.
--
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]
