[fix](pipeline) fix use error row desc when origin block clear

be/src/exec/exec_node.h

@@ -127,7 +127,7 @@ class ExecNode {
     bool has_output_row_descriptor() const { return _output_row_descriptor != nullptr; }
     // If use projection, we should clear `_origin_block`.
     void clear_origin_block() {
-        _origin_block.clear_column_data(_row_descriptor.num_materialized_slots());
+        _origin_block.clear_column_data(intermediate_row_desc().num_materialized_slots());
     }
 
     // Emit data, both need impl with method: sink
@@ -326,8 +326,8 @@ class ExecNode {
     std::shared_ptr<QueryStatistics> _query_statistics = nullptr;
 
     //_keep_origin is used to avoid copying during projection,
-    // currently set to true only in the nestloop join.
-    bool _keep_origin = false;
+    // currently set to false only in the nestloop join.
+    bool _keep_origin = true;
 
 private:
     static Status create_tree_helper(RuntimeState* state, ObjectPool* pool,

be/src/pipeline/exec/join_probe_operator.cpp

@@ -87,9 +87,7 @@ Status JoinProbeLocalState<SharedStateArg, Derived>::_build_output_block(
         // and you could see a 'todo' in the Thrift definition.
         //  Here, we have refactored it, but considering upgrade compatibility, we still need to retain the old code.
         if (!output_block->mem_reuse()) {
-            vectorized::MutableBlock tmp(
-                    vectorized::VectorizedUtils::create_columns_with_type_and_name(p.row_desc()));
-            output_block->swap(tmp.to_block());
+            output_block->swap(origin_block->clone_empty());
         }
         output_block->swap(*origin_block);
         return Status::OK();

be/src/pipeline/pipeline_x/operator.cpp

@@ -167,7 +167,7 @@ Status OperatorXBase::close(RuntimeState* state) {
 }
 
 void PipelineXLocalStateBase::clear_origin_block() {
-    _origin_block.clear_column_data(_parent->_row_descriptor.num_materialized_slots());
+    _origin_block.clear_column_data(_parent->intermediate_row_desc().num_materialized_slots());
 }
 
 Status OperatorXBase::do_projections(RuntimeState* state, vectorized::Block* origin_block,

be/src/pipeline/pipeline_x/operator.h

@@ -328,8 +328,8 @@ class OperatorXBase : public OperatorBase {
     int _parallel_tasks = 0;
 
     //_keep_origin is used to avoid copying during projection,
-    // currently set to true only in the nestloop join.
-    bool _keep_origin = false;
+    // currently set to false only in the nestloop join.
+    bool _keep_origin = true;
 };
 
 template <typename LocalStateType>

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

@@ -185,8 +185,7 @@ Status VJoinNodeBase::_build_output_block(Block* origin_block, Block* output_blo
         // and you could see a 'todo' in the Thrift definition.
         //  Here, we have refactored it, but considering upgrade compatibility, we still need to retain the old code.
         if (!output_block->mem_reuse()) {
-            MutableBlock tmp(VectorizedUtils::create_columns_with_type_and_name(row_desc()));
-            output_block->swap(tmp.to_block());
+            output_block->swap(origin_block->clone_empty());
         }
         output_block->swap(*origin_block);
         return Status::OK();

regression-test/data/correctness_p0/test_probe_clean.out

@@ -0,0 +1,10 @@
+-- This file is automatically generated. You should know what you did if you want to edit this
+-- !select_pipelineX --
+2020	-5.2
+
+-- !select_pipeline --
+2020	-5.2
+
+-- !select_non_pipeline --
+2020	-5.2
+

regression-test/suites/correctness_p0/test_probe_clean.groovy

@@ -0,0 +1,95 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+// The cases is copied from https://github.com/trinodb/trino/tree/master
+// /testing/trino-product-tests/src/main/resources/sql-tests/testcases/aggregate
+// and modified by Doris.
+
+suite("test_probe_clean") {
+
+sql """ drop table IF EXISTS clearblocktable1; """ 
+sql """ 
+ CREATE TABLE IF NOT EXISTS clearblocktable1 (
+            `col_int_undef_signed` INT NULL COMMENT "",
+            `col_int_undef_signed_not_null` INT NOT NULL COMMENT "",
+            `col_date_undef_signed_not_null` date(11)  NOT NULL COMMENT "",
+
+            ) ENGINE=OLAP
+            DUPLICATE KEY(`col_int_undef_signed`)
+            DISTRIBUTED BY HASH(`col_int_undef_signed`) BUCKETS 1
+            PROPERTIES (
+            'replication_num' = '1'
+);
+"""
+
+
+sql """ 
+insert into clearblocktable1 values(1,1,'2020-01-01');
+"""
+sql """ 
+drop table IF EXISTS clearblocktable2;
+""" 
+sql """
+CREATE TABLE IF NOT EXISTS clearblocktable2 (
+            `col_int_undef_signed` INT NULL COMMENT "",
+            `col_int_undef_signed_not_null` INT NOT NULL COMMENT "",
+            `col_date_undef_signed_not_null` date(11)  NOT NULL COMMENT "",
+
+            ) ENGINE=OLAP
+            DUPLICATE KEY(`col_int_undef_signed`)
+            DISTRIBUTED BY HASH(`col_int_undef_signed`) BUCKETS 1
+            PROPERTIES (
+            'replication_num' = '1'
+);
+"""
+
+sql """
+insert into clearblocktable2 values(1,1,'2020-01-01');
+"""
+
+sql """
+set enable_pipeline_x_engine=true, enable_pipeline_engine=true;
+"""
+qt_select_pipelineX """ 
+
+SELECT YEAR(ifnull(clearblocktable1.`col_date_undef_signed_not_null`, clearblocktable1.`col_date_undef_signed_not_null`)) AS field1 , 
+CASE  WHEN clearblocktable1.`col_int_undef_signed` != clearblocktable1.`col_int_undef_signed` * (8 + 1) THEN -5.2 ELSE clearblocktable1.`col_int_undef_signed` END AS field2 
+FROM clearblocktable1 INNER JOIN clearblocktable2 ON clearblocktable2.`col_int_undef_signed` = clearblocktable1.`col_int_undef_signed` WHERE clearblocktable1.`col_int_undef_signed_not_null` <> 7;
+
+"""
+
+sql """
+set enable_pipeline_x_engine=false,enable_pipeline_engine=true;
+"""
+qt_select_pipeline """ 
+
+SELECT YEAR(ifnull(clearblocktable1.`col_date_undef_signed_not_null`, clearblocktable1.`col_date_undef_signed_not_null`)) AS field1 , 
+CASE  WHEN clearblocktable1.`col_int_undef_signed` != clearblocktable1.`col_int_undef_signed` * (8 + 1) THEN -5.2 ELSE clearblocktable1.`col_int_undef_signed` END AS field2 
+FROM clearblocktable1 INNER JOIN clearblocktable2 ON clearblocktable2.`col_int_undef_signed` = clearblocktable1.`col_int_undef_signed` WHERE clearblocktable1.`col_int_undef_signed_not_null` <> 7;
+
+"""
+
+sql """
+set enable_pipeline_x_engine=false, enable_pipeline_engine=false;
+"""
+qt_select_non_pipeline """ 
+
+SELECT YEAR(ifnull(clearblocktable1.`col_date_undef_signed_not_null`, clearblocktable1.`col_date_undef_signed_not_null`)) AS field1 , 
+CASE  WHEN clearblocktable1.`col_int_undef_signed` != clearblocktable1.`col_int_undef_signed` * (8 + 1) THEN -5.2 ELSE clearblocktable1.`col_int_undef_signed` END AS field2 
+FROM clearblocktable1 INNER JOIN clearblocktable2 ON clearblocktable2.`col_int_undef_signed` = clearblocktable1.`col_int_undef_signed` WHERE clearblocktable1.`col_int_undef_signed_not_null` <> 7;
+"""
+}