[improvement](be) Eliminate redundant MultiCast block copies (#63580)

Related PR: #60386

Problem Summary: MultiCastDataStreamer already shares pulled blocks
through the column copy-on-write contract. The previous pull path still
cloned every column for each consumer through _copy_block(), which added
unnecessary per-consumer allocation and copy work. The copy-on-write
assertion changes and regression coverage from the original attempt are
already present on current master, so this change keeps the pull
completion accounting in _finish_pull() and returns the shared block
directly.

test performance with:
```sql
SET enable_profile = true;
  SET profile_level = 2;
  SET inline_cte_referenced_threshold = 0;
  SET parallel_pipeline_task_num = 16;

  WITH base AS (
    SELECT id, k1, ..., k8, v01, ..., v24, s1, s2, s3, s4
    FROM bench_cte_multicast_wide
    WHERE id >= 0
  )
  SELECT 0 AS branch, COUNT(*),
         SUM(id + k1 + ... + k8 + v01 + ... + v24),
         SUM(LENGTH(s1) + LENGTH(s2) + LENGTH(s3) + LENGTH(s4))
  FROM base WHERE k1 % 16 = 0
  UNION ALL
  ...
  UNION ALL
  SELECT 15 AS branch, COUNT(*),
         SUM(id + k1 + ... + k8 + v01 + ... + v24),
         SUM(LENGTH(s1) + LENGTH(s2) + LENGTH(s3) + LENGTH(s4))
  FROM base WHERE k1 % 16 = 15
  ORDER BY branch;
```

result:
| version | min | median | avg | max |
|---|---:|---:|---:|---:|
| before | 1.450s | 1.470s | 1.472s | 1.500s |
| after | 0.720s | 0.735s | 0.737s | 0.760s |

Author: zclllyybb

be/src/exec/operator/multi_cast_data_streamer.cpp

@@ -151,16 +151,10 @@ Status MultiCastDataStreamer::pull(RuntimeState* state, int sender_idx, Block* b
         }
     }
 
-    return _copy_block(state, sender_idx, block, *multi_cast_block);
+    return _finish_pull(state, *multi_cast_block);
 }
 
-Status MultiCastDataStreamer::_copy_block(RuntimeState* state, int32_t sender_idx, Block* block,
-                                          MultiCastBlock& multi_cast_block) {
-    const auto rows = block->rows();
-    for (int i = 0; i < block->columns(); ++i) {
-        block->get_by_position(i).column = block->get_by_position(i).column->clone_resized(rows);
-    }
-
+Status MultiCastDataStreamer::_finish_pull(RuntimeState* state, MultiCastBlock& multi_cast_block) {
     LockGuard l(_mutex);
     multi_cast_block._un_finish_copy--;
     auto copying_count = _copying_count.fetch_sub(1) - 1;
@@ -402,4 +396,4 @@ std::string MultiCastDataStreamer::debug_string() {
     return fmt::to_string(debug_string_buffer);
 }
 
-} // namespace doris
+} // namespace doris

be/src/exec/operator/multi_cast_data_streamer.h

@@ -97,8 +97,7 @@ class MultiCastDataStreamer {
     void _set_ready_for_read(int sender_idx);
     void _block_reading(int sender_idx);
 
-    Status _copy_block(RuntimeState* state, int32_t sender_idx, Block* block,
-                       MultiCastBlock& multi_cast_block);
+    Status _finish_pull(RuntimeState* state, MultiCastBlock& multi_cast_block);
 
     Status _start_spill_task(RuntimeState* state, SpillFileSPtr spill_file) REQUIRES(_mutex);
 
@@ -128,4 +127,4 @@ class MultiCastDataStreamer {
     // operator_profile of each source operator
     std::vector<RuntimeProfile*> _source_operator_profiles;
 };
-} // namespace doris
+} // namespace doris