[refactor](be) Add operator IO wrappers (#64139)

### What problem does this PR solve?

Issue Number: N/A

Problem Summary:

Pipeline operator source and sink paths need a common place to validate
output and input blocks. Before this change, `sink` and `get_block` were
the virtual override points, so common validation either had to stay in
call sites or be duplicated across operator implementations.

Root cause: the public operator data-flow entry points were also the
polymorphic implementation hooks, which left no wrapper layer for shared
checks.

This change makes `DataSinkOperatorXBase::sink` and
`OperatorXBase::get_block` non-virtual wrappers. The wrappers run
`Block::check_type_and_column()` at the source/sink boundary and then
dispatch to the new virtual `sink_impl` and `get_block_impl` methods.
All pipeline operator implementations, exchange operators, scan
operators, and related BE test mocks are migrated to the new impl
methods. The scan projection path is updated to call the base
`get_block` wrapper so the shared checks still apply.

Author: Mryange

be/src/exec/exchange/local_exchange_sink_operator.cpp

@@ -142,7 +142,7 @@ std::string LocalExchangeSinkLocalState::debug_string(int indentation_level) con
     return fmt::to_string(debug_string_buffer);
 }
 
-Status LocalExchangeSinkOperatorX::sink(RuntimeState* state, Block* in_block, bool eos) {
+Status LocalExchangeSinkOperatorX::sink_impl(RuntimeState* state, Block* in_block, bool eos) {
     auto& local_state = get_local_state(state);
     SCOPED_TIMER(local_state.exec_time_counter());
     COUNTER_UPDATE(local_state.rows_input_counter(), (int64_t)in_block->rows());

be/src/exec/exchange/local_exchange_sink_operator.h

@@ -103,7 +103,7 @@ class LocalExchangeSinkOperatorX final : public DataSinkOperatorX<LocalExchangeS
 
     Status prepare(RuntimeState* state) override;
 
-    Status sink(RuntimeState* state, Block* in_block, bool eos) override;
+    Status sink_impl(RuntimeState* state, Block* in_block, bool eos) override;
 
     void set_low_memory_mode(RuntimeState* state) override {
         auto& local_state = get_local_state(state);

be/src/exec/exchange/local_exchange_source_operator.cpp

@@ -89,7 +89,7 @@ std::string LocalExchangeSourceLocalState::debug_string(int indentation_level) c
     return fmt::to_string(debug_string_buffer);
 }
 
-Status LocalExchangeSourceOperatorX::get_block(RuntimeState* state, Block* block, bool* eos) {
+Status LocalExchangeSourceOperatorX::get_block_impl(RuntimeState* state, Block* block, bool* eos) {
     auto& local_state = get_local_state(state);
     SCOPED_TIMER(local_state.exec_time_counter());
     RETURN_IF_ERROR(local_state._exchanger->get_block(

be/src/exec/exchange/local_exchange_source_operator.h

@@ -78,7 +78,7 @@ class LocalExchangeSourceOperatorX final : public OperatorX<LocalExchangeSourceL
     RowDescriptor& row_descriptor() override { return _child->row_descriptor(); }
     const RowDescriptor& row_desc() const override { return _child->row_desc(); }
 
-    Status get_block(RuntimeState* state, Block* block, bool* eos) override;
+    Status get_block_impl(RuntimeState* state, Block* block, bool* eos) override;
 
     bool is_source() const override { return true; }
 

be/src/exec/operator/aggregation_sink_operator.cpp

@@ -988,7 +988,7 @@ Status AggSinkOperatorX::_check_agg_fn_output() {
     return Status::OK();
 }
 
-Status AggSinkOperatorX::sink(doris::RuntimeState* state, Block* in_block, bool eos) {
+Status AggSinkOperatorX::sink_impl(doris::RuntimeState* state, Block* in_block, bool eos) {
     auto& local_state = get_local_state(state);
     SCOPED_TIMER(local_state.exec_time_counter());
     COUNTER_UPDATE(local_state.rows_input_counter(), (int64_t)in_block->rows());

be/src/exec/operator/aggregation_sink_operator.h

@@ -154,7 +154,7 @@ class AggSinkOperatorX MOCK_REMOVE(final) : public DataSinkOperatorX<AggSinkLoca
 
     Status prepare(RuntimeState* state) override;
 
-    Status sink(RuntimeState* state, Block* in_block, bool eos) override;
+    Status sink_impl(RuntimeState* state, Block* in_block, bool eos) override;
 
     DataDistribution required_data_distribution(RuntimeState* state) const override {
         if (_partition_exprs.empty()) {

be/src/exec/operator/aggregation_source_operator.cpp

@@ -562,7 +562,7 @@ AggSourceOperatorX::AggSourceOperatorX(ObjectPool* pool, const TPlanNode& tnode,
           _needs_finalize(tnode.agg_node.need_finalize),
           _without_key(tnode.agg_node.grouping_exprs.empty()) {}
 
-Status AggSourceOperatorX::get_block(RuntimeState* state, Block* block, bool* eos) {
+Status AggSourceOperatorX::get_block_impl(RuntimeState* state, Block* block, bool* eos) {
     auto& local_state = get_local_state(state);
     SCOPED_TIMER(local_state.exec_time_counter());
     SCOPED_PEAK_MEM(&local_state._estimate_memory_usage);

be/src/exec/operator/aggregation_source_operator.h

@@ -102,7 +102,7 @@ class AggSourceOperatorX : public OperatorX<AggLocalState> {
     AggSourceOperatorX() = default;
 #endif
 
-    Status get_block(RuntimeState* state, Block* block, bool* eos) override;
+    Status get_block_impl(RuntimeState* state, Block* block, bool* eos) override;
 
     bool is_source() const override { return true; }
 

be/src/exec/operator/analytic_sink_operator.cpp

@@ -744,7 +744,7 @@ Status AnalyticSinkOperatorX::prepare(RuntimeState* state) {
     return Status::OK();
 }
 
-Status AnalyticSinkOperatorX::sink(doris::RuntimeState* state, Block* input_block, bool eos) {
+Status AnalyticSinkOperatorX::sink_impl(doris::RuntimeState* state, Block* input_block, bool eos) {
     auto& local_state = get_local_state(state);
     SCOPED_TIMER(local_state.exec_time_counter());
     COUNTER_UPDATE(local_state.rows_input_counter(), (int64_t)input_block->rows());

be/src/exec/operator/analytic_sink_operator.h

@@ -209,7 +209,7 @@ class AnalyticSinkOperatorX final : public DataSinkOperatorX<AnalyticSinkLocalSt
 
     Status prepare(RuntimeState* state) override;
 
-    Status sink(RuntimeState* state, Block* in_block, bool eos) override;
+    Status sink_impl(RuntimeState* state, Block* in_block, bool eos) override;
     DataDistribution required_data_distribution(RuntimeState* /*state*/) const override {
         if (_partition_by_eq_expr_ctxs.empty()) {
             return {ExchangeType::PASSTHROUGH};