[test](case) add insert into with tablet sink shuffle case (#44215)

### What problem does this PR solve?
Problem Summary:
add some test case

regression-test/data/insert_p0/test_insert_tablet_sink.out

@@ -0,0 +1,21 @@
+-- This file is automatically generated. You should know what you did if you want to edit this
+-- !select --
+1	170141183460469231731687303715884105727
+2	-170141183460469231731687303715884105728
+3	333
+4	444
+5	555
+6	666
+
+-- !select --
+1	170141183460469231731687303715884105727
+2	-170141183460469231731687303715884105728
+3	333
+3	333
+4	444
+4	444
+5	555
+5	555
+6	666
+6	666
+

regression-test/suites/insert_p0/test_insert_tablet_sink.groovy

@@ -0,0 +1,70 @@
+// 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.
+
+suite("test_insert_tablet_sink") {
+    sql """drop table if exists `table_largeint`;"""
+    sql """drop table if exists `tmp_varchar`;"""
+
+    sql """
+            CREATE TABLE `tmp_varchar` (
+            `k1` bigint(20) not NULL,
+            `c_varchar` varchar(65533) not NULL
+        ) ENGINE=OLAP
+        UNIQUE KEY(`k1`, c_varchar)
+        COMMENT 'OLAP'
+        AUTO PARTITION BY list (c_varchar) ()
+        DISTRIBUTED BY HASH(`k1`) BUCKETS 10
+        PROPERTIES("replication_num" = "1");
+    """
+
+    sql """
+    CREATE TABLE `table_largeint` (
+                `k1` bigint(20) not NULL,
+                `c_largeint` largeint not NULL,
+                str string null
+            ) ENGINE=OLAP
+            UNIQUE KEY(`k1`, c_largeint)
+            COMMENT 'OLAP'
+            AUTO PARTITION BY list (c_largeint) ()
+            DISTRIBUTED BY HASH(`k1`) BUCKETS 10
+            PROPERTIES("replication_num" = "1");
+
+    """
+
+    sql """insert into tmp_varchar values(1, "170141183460469231731687303715884105727")"""
+    sql """insert into tmp_varchar values(2, "-170141183460469231731687303715884105728")"""
+    sql """insert into tmp_varchar values(3,'333');"""
+    sql """insert into tmp_varchar values(4,'444');"""
+    sql """insert into tmp_varchar values(5,'555');"""
+    sql """insert into tmp_varchar values(6,'666');"""
+
+    qt_select """select * from tmp_varchar order by 1;"""
+
+    sql """ set skip_delete_bitmap = true; """
+    sql """ set enable_memtable_on_sink_node = true; """
+    sql """ set parallel_pipeline_task_num = 2; """
+
+
+    sql """ insert into table_largeint select k1,c_varchar,cast(rand() * 50000000 as bigint) from tmp_varchar where k1>=3; """
+    explain {
+        sql "insert into table_largeint select k1,c_varchar,cast(rand() * 50000000 as bigint) from tmp_varchar;"
+        contains "TABLET_SINK_SHUFFLE_PARTITIONED"
+    }
+
+    sql """ insert into table_largeint select k1,c_varchar,cast(rand() * 50000000 as bigint) from tmp_varchar; """
+    qt_select """select k1,c_largeint from table_largeint order by 1;"""
+}

samples/doris-demo/java-udf-demo/src/main/java/org/apache/doris/udf/SimpleDemo.java

@@ -0,0 +1,88 @@
+// 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.
+
+package org.apache.doris.udf;
+
+import java.io.DataInputStream;
+import java.io.DataOutputStream;
+import java.io.IOException;
+import java.util.logging.Logger;
+
+public class SimpleDemo  {
+
+    Logger log = Logger.getLogger("SimpleDemo");
+
+    //Need an inner class to store data
+    /*required*/
+    public static class State {
+        /*some variables if you need */
+        public int sum = 0;
+    }
+
+    /*required*/
+    public State create() {
+        /* here could do some init work if needed */
+        return new State();
+    }
+
+    /*required*/
+    public void destroy(State state) {
+        /* here could do some destroy work if needed */
+    }
+
+    /*Not Required*/
+    public void reset(State state) {
+        /*if you want this udaf function can work with window function.*/
+        /*Must impl this, it will be reset to init state after calculate every window frame*/
+        state.sum = 0;
+    }
+
+    /*required*/
+    //first argument is State, then other types your input
+    public void add(State state, Integer val) throws Exception {
+        /* here doing update work when input data*/
+        if (val != null) {
+            state.sum += val;
+        }
+    }
+
+    /*required*/
+    public void serialize(State state, DataOutputStream out) throws IOException  {
+        /* serialize some data into buffer */
+        out.writeInt(state.sum);
+    }
+
+    /*required*/
+    public void deserialize(State state, DataInputStream in) throws IOException  {
+        /* deserialize get data from buffer before you put */
+        int val = in.readInt();
+        state.sum = val;
+    }
+
+    /*required*/
+    public void merge(State state, State rhs) throws Exception {
+        /* merge data from state */
+        state.sum += rhs.sum;
+    }
+
+    /*required*/
+    //return Type you defined
+    public Integer getValue(State state) throws Exception {
+        /* return finally result */
+        return state.sum;
+    }
+}

samples/doris-demo/java-udf-demo/src/main/java/org/apache/doris/udf/UDTFStringTest.java

@@ -0,0 +1,31 @@
+// 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.
+
+package org.apache.doris.udf;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+
+public class UDTFStringTest {
+    public ArrayList<String> evaluate(String value, String separator) {
+        if (value == null || separator == null) {
+            return null;
+        } else {
+            return new ArrayList<>(Arrays.asList(value.split(separator)));
+        }
+    }
+}