add sync for stream load test (#13185)

regression-test/suites/csv_header_p0/test_csv_with_header.groovy

@@ -63,6 +63,7 @@ suite("test_csv_with_header") {
         while(max_try_secs--) {
             result = sql "show load where label = '${checklabel}'"
             if(result[0][2] == "FINISHED") {
+                sql "sync"
                 result_count = sql "select count(*) from ${testTable4}"
                 assertEquals(result_count[0][0], expected_rows)
                 break
@@ -110,6 +111,7 @@ suite("test_csv_with_header") {
     label = UUID.randomUUID().toString()
     test_stream_load.call(testTable, label, format_csv_with_names_and_types, format_csv_with_names_and_types_file, expect_rows)
 
+    sql "sync"
     // check total rows
     def result_count = sql "select count(*) from ${testTable}"
     assertEquals(result_count[0][0], expect_rows*3)
@@ -176,6 +178,7 @@ suite("test_csv_with_header") {
             assertEquals(expectedTotalRows,totalLines)
         }
 
+        sql "sync"
         resultCount = sql "select count(*) from ${testTable}"
         currentTotalRows = resultCount[0][0]
 
@@ -203,6 +206,7 @@ suite("test_csv_with_header") {
         
         // select out file to hdfs 
         select_out_file = {outTable, outHdfsPath, outFormat, outHdfsFs, outBroker, outHdfsUser, outPasswd->
+            sql "sync"
             sql """
                 SELECT * FROM ${outTable}
                 INTO OUTFILE "${outHdfsPath}"

regression-test/suites/load_p0/stream_load/load_json_null_to_nullable.groovy

@@ -84,6 +84,7 @@ suite("test_load_json_null_to_nullable", "p0") {
 
         load_array_data.call(testTable, 'true', '', 'json', '', '', '', '', '', '', 'test_char.json')
         
+        sql "sync"
         // select the table and check whether the data is correct
         qt_select "select * from ${testTable} order by k1"
 
@@ -99,6 +100,7 @@ suite("test_load_json_null_to_nullable", "p0") {
 
         load_array_data.call(testTable, 'true', '', 'json', '', '', '', '', '', '', 'test_char.json')
         
+        sql "sync"
         // select the table and check whether the data is correct
         qt_select "select * from ${testTable} order by k1"
 

regression-test/suites/load_p0/stream_load/load_nullable_to_not_nullable.groovy

@@ -57,6 +57,7 @@ suite("load_nullable_to_not_nullable") {
     String out = process.getText()
     logger.info("Run command: command=" + command + ",code=" + code + ", out=" + out + ", err=" + err)
     assertEquals(code, 0)
+    sql "sync"
     qt_sql " SELECT * FROM ${tableName} "
     sql "DROP TABLE ${tableName} "
 }

regression-test/suites/load_p0/stream_load/test_stream_load.groovy

@@ -63,6 +63,7 @@ suite("test_stream_load", "p0") {
         time 10000 // limit inflight 10s
     }
 
+    sql "sync"
     qt_sql "select * from ${tableName} order by k1, k2"
 
     // test strict_mode fail
@@ -128,6 +129,7 @@ suite("test_stream_load", "p0") {
         }
     }
 
+    sql "sync"
     rowCount = sql "select count(1) from ${tableName}"
     assertEquals(3, rowCount[0][0])
 }