database/doris
2
0
Fork 0
Code Issues Pull Requests Actions 3 Packages Projects Releases Wiki Activity

[pick](branch-2.1) pick #41676 #41740 #41857 (#41904)

Browse Source 
pick #41676 #41740 #41857
This commit is contained in:
Gabriel
2024-10-15 22:41:17 +08:00
committed by GitHub
parent a4b7d93ded
commit b185dfcbf6
7 changed files with 56 additions and 27 deletions
Download Patch File Download Diff File Expand all files Collapse all files

14
be/src/vec/sink/vdata_stream_sender.cpp
View File

@ -231,7 +231,7 @@ Status Channel<Parent>::send_local_block(Status exec_status, bool eos) {
}
template <typename Parent>
Status Channel<Parent>::send_local_block(Block* block) {
Status Channel<Parent>::send_local_block(Block* block, bool can_be_moved) {
SCOPED_TIMER(_parent->local_send_timer());
if (_recvr_is_valid()) {
if constexpr (!std::is_same_v<pipeline::ResultFileSinkLocalState, Parent>) {
@ -239,7 +239,7 @@ Status Channel<Parent>::send_local_block(Block* block) {
COUNTER_UPDATE(_parent->local_sent_rows(), block->rows());
COUNTER_UPDATE(_parent->blocks_sent_counter(), 1);
}
_local_recvr->add_block(block, _parent->sender_id(), false);
_local_recvr->add_block(block, _parent->sender_id(), can_be_moved);
return Status::OK();
} else {
return _receiver_status;
@ -646,7 +646,7 @@ Status VDataStreamSender::send(RuntimeState* state, Block* block, bool eos) {
Status status;
for (auto channel : _channels) {
if (!channel->is_receiver_eof()) {
status = channel->send_local_block(block);
status = channel->send_local_block(block, false);
HANDLE_CHANNEL_STATUS(state, channel, status);
}
}
@ -671,7 +671,7 @@ Status VDataStreamSender::send(RuntimeState* state, Block* block, bool eos) {
for (auto channel : _channels) {
if (!channel->is_receiver_eof()) {
if (channel->is_local()) {
status = channel->send_local_block(&cur_block);
status = channel->send_local_block(&cur_block, false);
} else {
SCOPED_CONSUME_MEM_TRACKER(_mem_tracker.get());
status = channel->send_broadcast_block(block_holder, eos);
@ -698,7 +698,7 @@ Status VDataStreamSender::send(RuntimeState* state, Block* block, bool eos) {
for (auto channel : _channels) {
if (!channel->is_receiver_eof()) {
if (channel->is_local()) {
status = channel->send_local_block(&cur_block);
status = channel->send_local_block(&cur_block, false);
} else {
SCOPED_CONSUME_MEM_TRACKER(_mem_tracker.get());
status = channel->send_remote_block(_cur_pb_block, false);
@ -717,7 +717,7 @@ Status VDataStreamSender::send(RuntimeState* state, Block* block, bool eos) {
if (!current_channel->is_receiver_eof()) {
// 2. serialize, send and rollover block
if (current_channel->is_local()) {
auto status = current_channel->send_local_block(block);
auto status = current_channel->send_local_block(block, false);
HANDLE_CHANNEL_STATUS(state, current_channel, status);
} else {
SCOPED_CONSUME_MEM_TRACKER(_mem_tracker.get());
@ -829,7 +829,7 @@ Status VDataStreamSender::close(RuntimeState* state, Status exec_status) {
for (auto channel : _channels) {
if (!channel->is_receiver_eof()) {
if (channel->is_local()) {
status = channel->send_local_block(&block);
status = channel->send_local_block(&block, false);
} else {
SCOPED_CONSUME_MEM_TRACKER(_mem_tracker.get());
status = channel->send_remote_block(_cur_pb_block, false);

2
be/src/vec/sink/vdata_stream_sender.h
View File

@ -307,7 +307,7 @@ public:
Status send_local_block(Status exec_status, bool eos = false);
Status send_local_block(Block* block);
Status send_local_block(Block* block, bool can_be_moved);
// Flush buffered rows and close channel. This function don't wait the response
// of close operation, client should call close_wait() to finish channel's close.
// We split one close operation into two phases in order to make multiple channels