[Bug](tablet-shuffle) tablet shuffle sink data should use input block rather than convert_block

be/src/pipeline/exec/exchange_sink_operator.cpp

@@ -499,8 +499,10 @@ Status ExchangeSinkOperatorX::sink(RuntimeState* state, vectorized::Block* block
             local_state._row_distribution._deal_batched = true;
             RETURN_IF_ERROR(local_state._send_new_partition_batch());
         }
+        // the convert_block maybe different with block after execute exprs
+        // when send data we still use block
         RETURN_IF_ERROR(channel_add_rows_with_idx(state, local_state.channels, num_channels,
-                                                  channel2rows, convert_block.get(), eos));
+                                                  channel2rows, block, eos));
     } else if (_part_type == TPartitionType::TABLE_SINK_HASH_PARTITIONED) {
         {
             SCOPED_TIMER(local_state._split_block_hash_compute_timer);

regression-test/data/nereids_p0/insert_into_table/random.out

@@ -138,3 +138,9 @@
 -- !sql_select --
 1	11	11
 
+-- !sql_select2 --
+1
+
+-- !sql_select3 --
+601022201389484209	2024-04-09T20:58:49	卖卖	{"is_poi_first_order":0}
+

regression-test/suites/nereids_p0/insert_into_table/random.groovy

@@ -54,4 +54,80 @@ suite('nereids_insert_random') {
     sql """INSERT INTO tbl_4 SELECT k1, k2, k2 FROM tbl_1;"""
     sql 'sync'
     qt_sql_select """ select * from tbl_4; """;
+
+
+    sql 'drop table if exists tbl_5'
+    sql 'drop table if exists tbl_6'
+    sql 'drop table if exists tbl_7'
+
+    sql """    
+        CREATE TABLE `tbl_5` (
+        `orderId` varchar(96) NOT NULL,
+        `updated_at` datetime NOT NULL,
+        `userLabel` varchar(255) NULL,
+        `userTag` variant NULL
+        ) ENGINE=OLAP
+        duplicate KEY(`orderId`, `updated_at`)
+        DISTRIBUTED BY HASH(`orderId`) BUCKETS 1
+        PROPERTIES (
+        "replication_allocation" = "tag.location.default: 1"
+        );
+     """  
+     sql """   
+        CREATE TABLE tbl_6 
+        (
+        order_id VARCHAR(96) NOT NULL,
+        updated_at DATETIMEV2  NOT NULL
+        ) ENGINE=OLAP
+        duplicate KEY(`order_id`)
+        DISTRIBUTED BY HASH(`order_id`) BUCKETS 1
+        PROPERTIES (
+        "replication_allocation" = "tag.location.default: 1"
+        );
+        """ 
+
+        sql """ INSERT INTO `tbl_6` values('601022201389484209', '2024-04-09 20:58:49');""" 
+
+        sql """
+        CREATE TABLE tbl_7 
+        (
+        orderId VARCHAR(96) NOT NULL,
+        userLabel VARIANT NULL
+        )ENGINE=OLAP
+        UNIQUE KEY(`orderId`)
+        DISTRIBUTED BY HASH(orderId) BUCKETS AUTO
+        PROPERTIES (
+        "replication_num" = "1"
+        );
+        """ 
+        sql """INSERT INTO `tbl_7` values('601022201389484209','{\"is_poi_first_order\":0}');""" 
+
+        sql 'sync'
+        qt_sql_select2 """ INSERT INTO
+                        tbl_5
+                        SELECT
+                        A.order_id as orderId,
+                        A.updated_at,
+                        CASE
+                                WHEN LOCATE('下单1次', CAST(B.userLabel AS STRING)) > 0
+                                OR LOCATE('买买', CAST(B.userLabel AS STRING)) > 0 then '买买'
+                                when B.userLabel ["is_poi_first_order"] = 1 then '买买'
+                                else '卖卖'
+                        end as userLabel,
+                        B.userLabel AS `userTag`
+                        FROM
+                        (
+                                select
+                                order_id,updated_at
+                                from
+                                tbl_6
+                        ) AS A
+                        LEFT JOIN (
+                                select
+                                orderId,userLabel
+                                from
+                                tbl_7
+                        ) AS B ON A.order_id = B.orderId; """;
+        qt_sql_select3 """ select * from tbl_5; """;
+
 }