channels

apache · Nov 18, 2024 · 12e5391 · 12e5391
1 parent 30c4a11
commit 12e5391
Show file tree

Hide file tree

Showing 2 changed files with 12 additions and 6 deletions.
diff --git a/be/src/pipeline/exec/exchange_sink_operator.cpp b/be/src/pipeline/exec/exchange_sink_operator.cpp
@@ -650,6 +650,15 @@ void ExchangeSinkLocalState::register_channels(pipeline::ExchangeSinkBuffer* buf
     for (auto& channel : channels) {
         channel->register_exchange_buffer(buffer);
     }
+
+    std::set<InstanceLoId> ins_id_set;
+    for (auto& channel : channels) {
+        auto ins_id = channel->ins_id();
+        if (!channel->is_local() && !ins_id_set.contains(ins_id)) {
+            buffer->register_sink(ins_id);
+            ins_id_set.insert(ins_id);
+        }
+    }
 }
 
 Status ExchangeSinkOperatorX::channel_add_rows(

diff --git a/be/src/vec/sink/vdata_stream_sender.h b/be/src/vec/sink/vdata_stream_sender.h
@@ -166,12 +166,9 @@ class Channel {
         return Status::OK();
     }
 
-    void register_exchange_buffer(pipeline::ExchangeSinkBuffer* buffer) {
-        _buffer = buffer;
-        if (!_is_local) {
-            _buffer->register_sink(_fragment_instance_id.lo);
-        }
-    }
+    void register_exchange_buffer(pipeline::ExchangeSinkBuffer* buffer) { _buffer = buffer; }
+
+    InstanceLoId ins_id() const { return _fragment_instance_id.lo; }
 
     std::shared_ptr<pipeline::ExchangeSendCallback<PTransmitDataResult>> get_send_callback(
             InstanceLoId id, bool eos) {