[Enhancement] add spilling capability for multi_cast_local_exchange operator (StarRocks#47982)

Signed-off-by: silverbullet233 <[email protected]>

Author: silverbullet233

be/src/common/config.h

@@ -1067,6 +1067,7 @@ CONF_Int64(spill_max_log_block_container_bytes, "10737418240"); // 10GB
 CONF_mDouble(spill_max_dir_bytes_ratio, "0.8"); // 80%
 // min bytes size of spill read buffer. if the buffer size is less than this value, we will disable buffer read
 CONF_Int64(spill_read_buffer_min_bytes, "1048576");
+CONF_mInt64(mem_limited_chunk_queue_block_size, "8388608");
 
 CONF_Int32(internal_service_query_rpc_thread_num, "-1");
 

be/src/exec/CMakeLists.txt

@@ -176,6 +176,8 @@ set(EXEC_FILES
     pipeline/exchange/local_exchange_sink_operator.cpp
     pipeline/exchange/local_exchange_source_operator.cpp
     pipeline/exchange/multi_cast_local_exchange.cpp
+    pipeline/exchange/mem_limited_chunk_queue.cpp
+    pipeline/exchange/spillable_multi_cast_local_exchange.cpp
     pipeline/exchange/multi_cast_local_exchange_sink_operator.cpp
     pipeline/exchange/multi_cast_local_exchange_source_operator.cpp
     pipeline/exchange/sink_buffer.cpp

be/src/exec/data_sink.cpp

@@ -315,7 +315,13 @@ Status DataSink::decompose_data_sink_to_pipeline(pipeline::PipelineBuilderContex
         auto* upstream_source = context->source_operator(prev_operators);
         size_t upstream_plan_node_id = upstream->plan_node_id();
         // === create exchange ===
-        auto mcast_local_exchanger = std::make_shared<InMemoryMultiCastLocalExchanger>(runtime_state, sinks.size());
+        std::shared_ptr<MultiCastLocalExchanger> mcast_local_exchanger;
+        if (runtime_state->enable_spill() && runtime_state->enable_multi_cast_local_exchange_spill()) {
+            mcast_local_exchanger = std::make_shared<SpillableMultiCastLocalExchanger>(runtime_state, sinks.size(),
+                                                                                       upstream_plan_node_id);
+        } else {
+            mcast_local_exchanger = std::make_shared<InMemoryMultiCastLocalExchanger>(runtime_state, sinks.size());
+        }
 
         // === create sink op ====
         OpFactoryPtr sink_op = std::make_shared<MultiCastLocalExchangeSinkOperatorFactory>(