[Enhancement] Add regression case for group commit

regression-test/data/insert_p0/test_group_commit_partial_update.out

@@ -0,0 +1,7 @@
+-- This file is automatically generated. You should know what you did if you want to edit this
+-- !select --
+5	\N	\N
+6	\N	\N
+70	7	\N
+80	8	\N
+

regression-test/data/insert_p0/test_stream_load1.csv

@@ -0,0 +1,2 @@
+5,e
+6,f

regression-test/data/insert_p0/test_stream_load2.csv

@@ -0,0 +1,2 @@
+70|7|e
+80|8|f

regression-test/suites/insert_p0/test_group_commit_partial_update.groovy

@@ -0,0 +1,124 @@
+// 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.
+import org.awaitility.Awaitility
+
+suite("test_group_commit_partial_update") {
+    def tableName = "test_group_commit_partial_update"
+
+   def getRowCount = { expectedRowCount ->
+    Awaitility.await().untilAsserted(() -> {
+        def rowCount = sql "select count(*) from ${tableName}"
+        logger.info("rowCount: " + rowCount[0][0])
+        assert rowCount[0][0] >= expectedRowCount : "Expected row count not reached, current count: " + rowCount[0][0]
+    })
+}
+
+    def checkStreamLoadGroupCommitResult = { exception, result, total_rows, loaded_rows, filtered_rows, unselected_rows ->
+        if (exception != null) {
+            throw exception
+        }
+        log.info("Stream load result: ${result}".toString())
+        def json = parseJson(result)
+        assertEquals("success", json.Status.toLowerCase())
+        assertTrue(json.GroupCommit)
+        assertTrue(json.Label.startsWith("group_commit_"))
+        assertEquals(total_rows, json.NumberTotalRows)
+        assertEquals(loaded_rows, json.NumberLoadedRows)
+        assertEquals(filtered_rows, json.NumberFilteredRows)
+        assertEquals(unselected_rows, json.NumberUnselectedRows)
+        if (filtered_rows > 0) {
+            assertFalse(json.ErrorURL.isEmpty())
+        } else {
+            assertTrue(json.ErrorURL == null || json.ErrorURL.isEmpty())
+        }
+    }
+
+    def checkStreamLoadResult = { exception, result, total_rows, loaded_rows, filtered_rows, unselected_rows ->
+        if (exception != null) {
+            throw exception
+        }
+        log.info("Stream load result: ${result}".toString())
+        def json = parseJson(result)
+        assertEquals("success", json.Status.toLowerCase())
+        assertEquals(total_rows, json.NumberTotalRows)
+        assertEquals(loaded_rows, json.NumberLoadedRows)
+        assertEquals(filtered_rows, json.NumberFilteredRows)
+        assertEquals(unselected_rows, json.NumberUnselectedRows)
+        if (filtered_rows > 0) {
+            assertFalse(json.ErrorURL.isEmpty())
+        } else {
+            assertTrue(json.ErrorURL == null || json.ErrorURL.isEmpty())
+        }
+    }
+
+    try {
+        sql """ drop table if exists ${tableName}; """
+
+        sql """
+        CREATE TABLE `${tableName}` (
+            `id` int(11) NOT NULL,
+            `name` varchar(1100) NULL,
+            `score` int(11) NULL default "-1"
+        ) ENGINE=OLAP
+        UNIQUE KEY(`id`)
+        DISTRIBUTED BY HASH(`id`) BUCKETS 1
+        PROPERTIES (
+            "replication_num" = "1",
+            "enable_unique_key_merge_on_write" = "true"
+        );
+        """
+
+        streamLoad {
+            table "${tableName}"
+
+            set 'column_separator', '|'
+            set 'group_commit', 'async_mode'
+            set 'columns', 'id, name, score'
+            file "test_stream_load2.csv"
+            unset 'label'
+
+            time 10000 // limit inflight 10s
+
+            check { result, exception, startTime, endTime ->
+                checkStreamLoadGroupCommitResult(exception, result, 2, 2, 0, 0)
+            }
+        }
+
+        streamLoad {
+            table "${tableName}"
+
+            set 'column_separator', ','
+            set 'group_commit', 'async_mode'
+            set 'partial_columns', 'true'
+            set 'columns', 'id, score'
+            file "test_stream_load1.csv"
+            unset 'label'
+
+            time 10000 // limit inflight 10s
+
+            check { result, exception, startTime, endTime ->
+                checkStreamLoadResult(exception, result, 2, 2, 0, 0)
+            }
+        }
+
+        sql "sync"
+        qt_select  "select * from ${tableName} order by id;"
+
+    } finally {
+        sql """ DROP TABLE IF EXISTS ${tableName} """
+    }
+}