[refactor](nereids)unify outputTupleDesc and projection

be/src/exec/exec_node.cpp

@@ -514,6 +514,24 @@ std::string ExecNode::get_name() {
 Status ExecNode::do_projections(vectorized::Block* origin_block, vectorized::Block* output_block) {
     SCOPED_TIMER(_exec_timer);
     SCOPED_TIMER(_projection_timer);
+    auto insert_column_datas = [&](auto& to, vectorized::ColumnPtr& from, size_t rows) {
+        if (to->is_nullable() && !from->is_nullable()) {
+            if (_keep_origin || !from->is_exclusive()) {
+                auto& null_column = reinterpret_cast<vectorized::ColumnNullable&>(*to);
+                null_column.get_nested_column().insert_range_from(*from, 0, rows);
+                null_column.get_null_map_column().get_data().resize_fill(rows, 0);
+            } else {
+                to = make_nullable(from, false)->assume_mutable();
+            }
+        } else {
+            if (_keep_origin || !from->is_exclusive()) {
+                to->insert_range_from(*from, 0, rows);
+            } else {
+                to = from->assume_mutable();
+            }
+        }
+    };
+
     using namespace vectorized;
     MutableBlock mutable_block =
             VectorizedUtils::build_mutable_mem_reuse_block(output_block, *_output_row_descriptor);
@@ -535,13 +553,7 @@ Status ExecNode::do_projections(vectorized::Block* origin_block, vectorized::Blo
             auto column_ptr = origin_block->get_by_position(result_column_id)
                                       .column->convert_to_full_column_if_const();
             //TODO: this is a quick fix, we need a new function like "change_to_nullable" to do it
-            if (mutable_columns[i]->is_nullable() xor column_ptr->is_nullable()) {
-                DCHECK(mutable_columns[i]->is_nullable() && !column_ptr->is_nullable());
-                reinterpret_cast<ColumnNullable*>(mutable_columns[i].get())
-                        ->insert_range_from_not_nullable(*column_ptr, 0, rows);
-            } else {
-                mutable_columns[i]->insert_range_from(*column_ptr, 0, rows);
-            }
+            insert_column_datas(mutable_columns[i], column_ptr, rows);
         }
         DCHECK(mutable_block.rows() == rows);
         output_block->set_columns(std::move(mutable_columns));

be/src/exec/exec_node.h

@@ -325,6 +325,8 @@ class ExecNode {
 
     std::shared_ptr<QueryStatistics> _query_statistics = nullptr;
 
+    bool _keep_origin = false;
+
 private:
     static Status create_tree_helper(RuntimeState* state, ObjectPool* pool,
                                      const std::vector<TPlanNode>& tnodes,

be/src/pipeline/exec/join_probe_operator.cpp

@@ -17,6 +17,11 @@
 
 #include "join_probe_operator.h"
 
+#include <memory>
+
+#include "common/exception.h"
+#include "common/logging.h"
+#include "common/status.h"
 #include "pipeline/exec/hashjoin_probe_operator.h"
 #include "pipeline/exec/nested_loop_join_probe_operator.h"
 #include "pipeline/exec/operator.h"
@@ -82,6 +87,14 @@ template <typename SharedStateArg, typename Derived>
 Status JoinProbeLocalState<SharedStateArg, Derived>::_build_output_block(
         vectorized::Block* origin_block, vectorized::Block* output_block, bool keep_origin) {
     auto& p = Base::_parent->template cast<typename Derived::Parent>();
+    if (!Base::_projections.empty()) {
+        *output_block = *origin_block;
+        if (p._is_outer_join) {
+            DCHECK(output_block->columns() >= 2);
+            output_block->erase_tail(output_block->columns() - 2);
+        }
+        return Status::OK();
+    }
     SCOPED_TIMER(_build_output_block_timer);
     auto is_mem_reuse = output_block->mem_reuse();
     vectorized::MutableBlock mutable_block =
@@ -93,10 +106,10 @@ Status JoinProbeLocalState<SharedStateArg, Derived>::_build_output_block(
     // TODO: After FE plan support same nullable of output expr and origin block and mutable column
     // we should replace `insert_column_datas` by `insert_range_from`
 
-    auto insert_column_datas = [keep_origin](auto& to, vectorized::ColumnPtr& from, size_t rows) {
+    auto insert_column_datas = [&](auto& to, vectorized::ColumnPtr& from, size_t rows) {
         if (to->is_nullable() && !from->is_nullable()) {
             if (keep_origin || !from->is_exclusive()) {
-                auto& null_column = reinterpret_cast<vectorized::ColumnNullable&>(*to);
+                auto& null_column = assert_cast<vectorized::ColumnNullable&>(*to);
                 null_column.get_nested_column().insert_range_from(*from, 0, rows);
                 null_column.get_null_map_column().get_data().resize_fill(rows, 0);
             } else {
@@ -109,7 +122,48 @@ Status JoinProbeLocalState<SharedStateArg, Derived>::_build_output_block(
                 to = from->assume_mutable();
             }
         }
+
+        if (to->is_nullable()) {
+            auto& null_column = assert_cast<vectorized::ColumnNullable&>(*to);
+            bool all_null = true;
+            for (int i = 0; i < null_column.size(); i++) {
+                if (!null_column.is_null_at(i)) {
+                    all_null = false;
+                }
+            }
+            LOG_WARNING("yxc test null").tag("null node id", p._node_id).tag("all null", all_null);
+        }
+    };
+
+    LOG_WARNING("yxc test empty copy");
+
+    auto print = [](vectorized::VExprContextSPtrs& ctxs) {
+        std::string s = "[    ";
+        for (auto& ctx : ctxs) {
+            s += ctx->root()->debug_string() + "\t\t\t";
+        }
+        return s + "    ]";
     };
+
+    auto print_desc = [](std::unique_ptr<RowDescriptor>& desc) {
+        std::string ret;
+        if (!desc) {
+            ret = "is null";
+        } else {
+            ret = desc->debug_string();
+        }
+        return ret;
+    };
+
+    LOG_WARNING("yxc  test  ")
+            .tag("node id", p._node_id)
+            .tag("Base::_projections", print(Base::_projections))
+            .tag("_output_expr_ctxs", print(_output_expr_ctxs))
+            .tag("output_block", mutable_block.dump_names() + "   " + mutable_block.dump_types())
+            .tag("orgin block ", origin_block->dump_structure())
+            .tag("_intermediate_row_desc", print_desc(p._intermediate_row_desc))
+            .tag("plan node", print_desc(p._output_row_descriptor))
+            .tag("join node", print_desc(p._output_row_desc));
     if (rows != 0) {
         auto& mutable_columns = mutable_block.mutable_columns();
         if (_output_expr_ctxs.empty()) {
@@ -197,18 +251,32 @@ JoinProbeOperatorX<LocalStateType>::JoinProbeOperatorX(ObjectPool* pool, const T
         _intermediate_row_desc.reset(new RowDescriptor(
                 descs, tnode.hash_join_node.vintermediate_tuple_id_list,
                 std::vector<bool>(tnode.hash_join_node.vintermediate_tuple_id_list.size())));
-        _output_row_desc.reset(
-                new RowDescriptor(descs, {tnode.hash_join_node.voutput_tuple_id}, {false}));
+        if (!Base::_output_row_descriptor) {
+            _output_row_desc.reset(
+                    new RowDescriptor(descs, {tnode.hash_join_node.voutput_tuple_id}, {false}));
+        } else {
+            if (tnode.hash_join_node.__isset.voutput_tuple_id) {
+                throw Exception {ErrorCode::INTERNAL_ERROR,
+                                 "yxc test error .__isset.voutput_tuple_id"};
+            }
+        }
+
     } else if (tnode.__isset.nested_loop_join_node) {
         _intermediate_row_desc.reset(new RowDescriptor(
                 descs, tnode.nested_loop_join_node.vintermediate_tuple_id_list,
                 std::vector<bool>(tnode.nested_loop_join_node.vintermediate_tuple_id_list.size())));
-        _output_row_desc.reset(
-                new RowDescriptor(descs, {tnode.nested_loop_join_node.voutput_tuple_id}, {false}));
+        if (!Base::_output_row_descriptor) {
+            _output_row_desc.reset(new RowDescriptor(
+                    descs, {tnode.nested_loop_join_node.voutput_tuple_id}, {false}));
+        }
     } else {
         // Iff BE has been upgraded and FE has not yet, we should keep origin logics for CROSS JOIN.
         DCHECK_EQ(_join_op, TJoinOp::CROSS_JOIN);
     }
+
+    if (Base::_node_id == 13) {
+        LOG_WARNING("yxc test");
+    }
 }
 
 template <typename LocalStateType>

be/src/pipeline/exec/join_probe_operator.h

@@ -70,7 +70,12 @@ class JoinProbeOperatorX : public StatefulOperatorX<LocalStateType> {
     Status init(const TPlanNode& tnode, RuntimeState* state) override;
 
     Status open(doris::RuntimeState* state) override;
-    [[nodiscard]] const RowDescriptor& row_desc() const override { return *_output_row_desc; }
+    [[nodiscard]] const RowDescriptor& row_desc() const override {
+        if (Base::_output_row_descriptor) {
+            return *Base::_output_row_descriptor;
+        }
+        return *_output_row_desc;
+    }
 
     [[nodiscard]] const RowDescriptor& intermediate_row_desc() const override {
         return *_intermediate_row_desc;

be/src/pipeline/exec/nested_loop_join_probe_operator.cpp

@@ -436,7 +436,9 @@ NestedLoopJoinProbeOperatorX::NestedLoopJoinProbeOperatorX(ObjectPool* pool, con
         : JoinProbeOperatorX<NestedLoopJoinProbeLocalState>(pool, tnode, operator_id, descs),
           _is_output_left_side_only(tnode.nested_loop_join_node.__isset.is_output_left_side_only &&
                                     tnode.nested_loop_join_node.is_output_left_side_only),
-          _old_version_flag(!tnode.__isset.nested_loop_join_node) {}
+          _old_version_flag(!tnode.__isset.nested_loop_join_node) {
+    _keep_origin = _is_output_left_side_only;
+}
 
 Status NestedLoopJoinProbeOperatorX::init(const TPlanNode& tnode, RuntimeState* state) {
     RETURN_IF_ERROR(JoinProbeOperatorX<NestedLoopJoinProbeLocalState>::init(tnode, state));

be/src/pipeline/exec/nested_loop_join_probe_operator.h

@@ -228,7 +228,7 @@ class NestedLoopJoinProbeOperatorX final
     const RowDescriptor& row_desc() const override {
         return _old_version_flag
                        ? (_output_row_descriptor ? *_output_row_descriptor : _row_descriptor)
-                       : *_output_row_desc;
+                       : (_output_row_descriptor ? *_output_row_descriptor : *_output_row_desc);
     }
 
     bool need_more_input_data(RuntimeState* state) const override;

be/src/pipeline/pipeline_x/operator.cpp

@@ -172,9 +172,28 @@ void PipelineXLocalStateBase::clear_origin_block() {
 
 Status OperatorXBase::do_projections(RuntimeState* state, vectorized::Block* origin_block,
                                      vectorized::Block* output_block) const {
-    auto local_state = state->get_local_state(operator_id());
+    auto* local_state = state->get_local_state(operator_id());
     SCOPED_TIMER(local_state->exec_time_counter());
     SCOPED_TIMER(local_state->_projection_timer);
+
+    auto insert_column_datas = [&](auto& to, vectorized::ColumnPtr& from, size_t rows) {
+        if (to->is_nullable() && !from->is_nullable()) {
+            if (_keep_origin || !from->is_exclusive()) {
+                auto& null_column = reinterpret_cast<vectorized::ColumnNullable&>(*to);
+                null_column.get_nested_column().insert_range_from(*from, 0, rows);
+                null_column.get_null_map_column().get_data().resize_fill(rows, 0);
+            } else {
+                to = make_nullable(from, false)->assume_mutable();
+            }
+        } else {
+            if (_keep_origin || !from->is_exclusive()) {
+                to->insert_range_from(*from, 0, rows);
+            } else {
+                to = from->assume_mutable();
+            }
+        }
+    };
+
     using namespace vectorized;
     vectorized::MutableBlock mutable_block =
             vectorized::VectorizedUtils::build_mutable_mem_reuse_block(output_block,
@@ -189,14 +208,7 @@ Status OperatorXBase::do_projections(RuntimeState* state, vectorized::Block* ori
             RETURN_IF_ERROR(local_state->_projections[i]->execute(origin_block, &result_column_id));
             auto column_ptr = origin_block->get_by_position(result_column_id)
                                       .column->convert_to_full_column_if_const();
-            //TODO: this is a quick fix, we need a new function like "change_to_nullable" to do it
-            if (mutable_columns[i]->is_nullable() xor column_ptr->is_nullable()) {
-                DCHECK(mutable_columns[i]->is_nullable() && !column_ptr->is_nullable());
-                reinterpret_cast<ColumnNullable*>(mutable_columns[i].get())
-                        ->insert_range_from_not_nullable(*column_ptr, 0, rows);
-            } else {
-                mutable_columns[i]->insert_range_from(*column_ptr, 0, rows);
-            }
+            insert_column_datas(mutable_columns[i], column_ptr, rows);
         }
         DCHECK(mutable_block.rows() == rows);
         output_block->set_columns(std::move(mutable_columns));

be/src/pipeline/pipeline_x/operator.h

@@ -326,6 +326,8 @@ class OperatorXBase : public OperatorBase {
     std::string _op_name;
     bool _ignore_data_distribution = false;
     int _parallel_tasks = 0;
+
+    bool _keep_origin = false;
 };
 
 template <typename LocalStateType>

be/src/vec/exec/join/vjoin_node_base.cpp

@@ -95,14 +95,18 @@ VJoinNodeBase::VJoinNodeBase(ObjectPool* pool, const TPlanNode& tnode, const Des
     }
 
     if (tnode.__isset.hash_join_node) {
-        _output_row_desc.reset(
-                new RowDescriptor(descs, {tnode.hash_join_node.voutput_tuple_id}, {false}));
+        if (!_output_row_descriptor) {
+            _output_row_desc.reset(
+                    new RowDescriptor(descs, {tnode.hash_join_node.voutput_tuple_id}, {false}));
+        }
         _intermediate_row_desc.reset(new RowDescriptor(
                 descs, tnode.hash_join_node.vintermediate_tuple_id_list,
                 std::vector<bool>(tnode.hash_join_node.vintermediate_tuple_id_list.size())));
     } else if (tnode.__isset.nested_loop_join_node) {
-        _output_row_desc.reset(
-                new RowDescriptor(descs, {tnode.nested_loop_join_node.voutput_tuple_id}, {false}));
+        if (!_output_row_descriptor) {
+            _output_row_desc.reset(new RowDescriptor(
+                    descs, {tnode.nested_loop_join_node.voutput_tuple_id}, {false}));
+        }
         _intermediate_row_desc.reset(new RowDescriptor(
                 descs, tnode.nested_loop_join_node.vintermediate_tuple_id_list,
                 std::vector<bool>(tnode.nested_loop_join_node.vintermediate_tuple_id_list.size())));
@@ -166,6 +170,14 @@ void VJoinNodeBase::_construct_mutable_join_block() {
 Status VJoinNodeBase::_build_output_block(Block* origin_block, Block* output_block,
                                           bool keep_origin) {
     SCOPED_TIMER(_build_output_block_timer);
+    if (!_projections.empty()) {
+        *output_block = *origin_block;
+        if (_is_outer_join) {
+            DCHECK(output_block->columns() >= 2);
+            output_block->erase_tail(output_block->columns() - 2);
+        }
+        return Status::OK();
+    }
     auto is_mem_reuse = output_block->mem_reuse();
     MutableBlock mutable_block =
             is_mem_reuse

be/src/vec/exec/join/vjoin_node_base.h

@@ -63,7 +63,12 @@ class VJoinNodeBase : public ExecNode {
 
     Status open(RuntimeState* state) override;
 
-    const RowDescriptor& row_desc() const override { return *_output_row_desc; }
+    const RowDescriptor& row_desc() const override {
+        if (_output_row_descriptor) {
+            return *_output_row_descriptor;
+        }
+        return *_output_row_desc;
+    }
 
     const RowDescriptor& intermediate_row_desc() const override { return *_intermediate_row_desc; }
 

be/src/vec/exec/join/vnested_loop_join_node.cpp

@@ -102,6 +102,7 @@ Status VNestedLoopJoinNode::init(const TPlanNode& tnode, RuntimeState* state) {
 
     if (tnode.nested_loop_join_node.__isset.is_output_left_side_only) {
         _is_output_left_side_only = tnode.nested_loop_join_node.is_output_left_side_only;
+        _keep_origin = _is_output_left_side_only;
     }
 
     if (tnode.nested_loop_join_node.__isset.join_conjuncts &&

be/src/vec/exec/join/vnested_loop_join_node.h

@@ -92,7 +92,7 @@ class VNestedLoopJoinNode final : public VJoinNodeBase {
     const RowDescriptor& row_desc() const override {
         return _old_version_flag
                        ? (_output_row_descriptor ? *_output_row_descriptor : _row_descriptor)
-                       : *_output_row_desc;
+                       : (_output_row_descriptor ? *_output_row_descriptor : *_output_row_desc);
     }
 
     std::shared_ptr<Block> get_left_block() { return _left_block; }

fe/fe-core/src/main/java/org/apache/doris/job/extensions/mtmv/MTMVTask.java

@@ -222,6 +222,7 @@ private void exec(ConnectContext ctx, Set<Long> refreshPartitionIds,
         executor = new StmtExecutor(ctx, new LogicalPlanAdapter(command, ctx.getStatementContext()));
         ctx.setExecutor(executor);
         ctx.setQueryId(queryId);
+        ctx.getState().setNereids(true);
         command.run(ctx, executor);
         if (ctx.getState().getStateType() != MysqlStateType.OK) {
             throw new JobException(ctx.getState().getErrorMessage());

fe/fe-core/src/main/java/org/apache/doris/nereids/glue/translator/PhysicalPlanTranslator.java

@@ -1539,8 +1539,8 @@ public PlanFragment visitPhysicalHashJoin(
             TupleDescriptor outputDescriptor = context.generateTupleDesc();
             outputSlotReferences.forEach(s -> context.createSlotDesc(outputDescriptor, s));
 
-            hashJoinNode.setvOutputTupleDesc(outputDescriptor);
-            hashJoinNode.setvSrcToOutputSMap(srcToOutput);
+            hashJoinNode.setOutputTupleDesc(outputDescriptor);
+            hashJoinNode.setProjectList(srcToOutput);
         }
         if (hashJoin.getStats() != null) {
             hashJoinNode.setCardinality((long) hashJoin.getStats().getRowCount());
@@ -1720,8 +1720,8 @@ public PlanFragment visitPhysicalNestedLoopJoin(
                 TupleDescriptor outputDescriptor = context.generateTupleDesc();
                 outputSlotReferences.forEach(s -> context.createSlotDesc(outputDescriptor, s));
 
-                nestedLoopJoinNode.setvOutputTupleDesc(outputDescriptor);
-                nestedLoopJoinNode.setvSrcToOutputSMap(srcToOutput);
+                nestedLoopJoinNode.setOutputTupleDesc(outputDescriptor);
+                nestedLoopJoinNode.setProjectList(srcToOutput);
             }
             if (nestedLoopJoin.getStats() != null) {
                 nestedLoopJoinNode.setCardinality((long) nestedLoopJoin.getStats().getRowCount());
@@ -1847,8 +1847,8 @@ public PlanFragment visitPhysicalProject(PhysicalProject<? extends Plan> project
         if (inputPlanNode instanceof JoinNodeBase) {
             TupleDescriptor tupleDescriptor = generateTupleDesc(slots, null, context);
             JoinNodeBase joinNode = (JoinNodeBase) inputPlanNode;
-            joinNode.setvOutputTupleDesc(tupleDescriptor);
-            joinNode.setvSrcToOutputSMap(projectionExprs);
+            joinNode.setOutputTupleDesc(tupleDescriptor);
+            joinNode.setProjectList(projectionExprs);
             // prune the hashOutputSlotIds
             if (joinNode instanceof HashJoinNode) {
                 ((HashJoinNode) joinNode).getHashOutputSlotIds().clear();