[refine](exchange)remove template in ExchangeSinkOperatorX::channel_a…

…dd_rows (#42425)
apache · Oct 25, 2024 · fb23f41 · fb23f41
1 parent e9a14d7
commit fb23f41
Show file tree

Hide file tree

Showing 2 changed files with 15 additions and 15 deletions.
diff --git a/be/src/pipeline/exec/exchange_sink_operator.cpp b/be/src/pipeline/exec/exchange_sink_operator.cpp
@@ -659,11 +659,10 @@ void ExchangeSinkLocalState::register_channels(pipeline::ExchangeSinkBuffer* buf
     }
 }
 
-template <typename Channels, typename HashValueType>
-Status ExchangeSinkOperatorX::channel_add_rows(RuntimeState* state, Channels& channels,
-                                               int num_channels,
-                                               const HashValueType* __restrict channel_ids,
-                                               int rows, vectorized::Block* block, bool eos) {
+Status ExchangeSinkOperatorX::channel_add_rows(
+        RuntimeState* state, std::vector<std::shared_ptr<vectorized::Channel>>& channels,
+        int num_channels, const uint32_t* __restrict channel_ids, int rows,
+        vectorized::Block* block, bool eos) {
     std::vector<std::vector<uint32_t>> channel2rows;
     channel2rows.resize(num_channels);
     for (uint32_t i = 0; i < rows; i++) {
@@ -675,10 +674,10 @@ Status ExchangeSinkOperatorX::channel_add_rows(RuntimeState* state, Channels& ch
     return Status::OK();
 }
 
-template <typename Channels>
 Status ExchangeSinkOperatorX::channel_add_rows_with_idx(
-        RuntimeState* state, Channels& channels, int num_channels,
-        std::vector<std::vector<uint32_t>>& channel2rows, vectorized::Block* block, bool eos) {
+        RuntimeState* state, std::vector<std::shared_ptr<vectorized::Channel>>& channels,
+        int num_channels, std::vector<std::vector<uint32_t>>& channel2rows,
+        vectorized::Block* block, bool eos) {
     Status status = Status::OK();
     for (int i = 0; i < num_channels; ++i) {
         if (!channels[i]->is_receiver_eof() && !channel2rows[i].empty()) {

diff --git a/be/src/pipeline/exec/exchange_sink_operator.h b/be/src/pipeline/exec/exchange_sink_operator.h
@@ -232,13 +232,14 @@ class ExchangeSinkOperatorX final : public DataSinkOperatorX<ExchangeSinkLocalSt
     template <typename ChannelPtrType>
     void _handle_eof_channel(RuntimeState* state, ChannelPtrType channel, Status st);
 
-    template <typename Channels, typename HashValueType>
-    Status channel_add_rows(RuntimeState* state, Channels& channels, int num_channels,
-                            const HashValueType* channel_ids, int rows, vectorized::Block* block,
-                            bool eos);
-
-    template <typename Channels>
-    Status channel_add_rows_with_idx(RuntimeState* state, Channels& channels, int num_channels,
+    Status channel_add_rows(RuntimeState* state,
+                            std::vector<std::shared_ptr<vectorized::Channel>>& channels,
+                            int num_channels, const uint32_t* channel_ids, int rows,
+                            vectorized::Block* block, bool eos);
+
+    Status channel_add_rows_with_idx(RuntimeState* state,
+                                     std::vector<std::shared_ptr<vectorized::Channel>>& channels,
+                                     int num_channels,
                                      std::vector<std::vector<uint32_t>>& channel2rows,
                                      vectorized::Block* block, bool eos);
     RuntimeState* _state = nullptr;