You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@doris.apache.org by yi...@apache.org on 2025/10/16 06:47:12 UTC
(doris) branch branch-4.0 updated: branch-4.0: [test](cloud) Add more multi cluster case for cloud #56840 (#57028)
This is an automated email from the ASF dual-hosted git repository.
yiguolei pushed a commit to branch branch-4.0
in repository https://gitbox.apache.org/repos/asf/doris.git
The following commit(s) were added to refs/heads/branch-4.0 by this push:
new e1d45cf7fa0 branch-4.0: [test](cloud) Add more multi cluster case for cloud #56840 (#57028)
e1d45cf7fa0 is described below
commit e1d45cf7fa0aeaee6f5b212369d8131201d18058
Author: github-actions[bot] <41...@users.noreply.github.com>
AuthorDate: Thu Oct 16 14:47:05 2025 +0800
branch-4.0: [test](cloud) Add more multi cluster case for cloud #56840 (#57028)
Cherry-picked from #56840
Co-authored-by: deardeng <de...@selectdb.com>
---
.../routine_load/test_routine_load.out | 33 +
.../test_schema_change_with_readd_cluster.out | 6 +
.../multi_cluster/stream_load/all_types.csv | 20 +
.../multi_cluster/stream_load/stream_load.out | 13 +
.../multi_cluster/stream_load/stream_load_2pc.out | 25 +
.../multi_cluster/stream_load/stream_load_lb.out | 25 +
.../multi_cluster/test_apsaradb_internal_stage.out | 7 +
.../cloud_p0/conf/regression-conf-custom.groovy | 1 +
.../nonConcurrent/conf/regression-conf.groovy | 1 +
.../async_copy_into/async_load.groovy | 275 +++++++
.../multi_cluster/async_copy_into/ddl/customer.sql | 13 +
.../async_copy_into/ddl/customer_delete.sql | 1 +
.../multi_cluster/async_copy_into/ddl/lineitem.sql | 21 +
.../async_copy_into/ddl/lineitem_delete.sql | 1 +
.../multi_cluster/async_copy_into/ddl/nation.sql | 9 +
.../async_copy_into/ddl/nation_delete.sql | 1 +
.../multi_cluster/async_copy_into/ddl/orders.sql | 14 +
.../async_copy_into/ddl/orders_delete.sql | 1 +
.../multi_cluster/async_copy_into/ddl/part.sql | 14 +
.../async_copy_into/ddl/part_delete.sql | 1 +
.../multi_cluster/async_copy_into/ddl/partsupp.sql | 10 +
.../async_copy_into/ddl/partsupp_delete.sql | 1 +
.../multi_cluster/async_copy_into/ddl/region.sql | 8 +
.../async_copy_into/ddl/region_delete.sql | 1 +
.../multi_cluster/async_copy_into/ddl/revenue1.sql | 11 +
.../async_copy_into/ddl/revenue1_delete.sql | 1 +
.../multi_cluster/async_copy_into/ddl/supplier.sql | 11 +
.../async_copy_into/ddl/supplier_delete.sql | 1 +
.../multi_cluster/copy_into/ddl/customer.sql | 13 +
.../copy_into/ddl/customer_delete.sql | 1 +
.../multi_cluster/copy_into/ddl/lineitem.sql | 21 +
.../copy_into/ddl/lineitem_delete.sql | 1 +
.../multi_cluster/copy_into/ddl/nation.sql | 9 +
.../multi_cluster/copy_into/ddl/nation_delete.sql | 1 +
.../multi_cluster/copy_into/ddl/orders.sql | 14 +
.../multi_cluster/copy_into/ddl/orders_delete.sql | 1 +
.../cloud_p0/multi_cluster/copy_into/ddl/part.sql | 14 +
.../multi_cluster/copy_into/ddl/part_delete.sql | 1 +
.../multi_cluster/copy_into/ddl/partsupp.sql | 10 +
.../copy_into/ddl/partsupp_delete.sql | 1 +
.../multi_cluster/copy_into/ddl/region.sql | 8 +
.../multi_cluster/copy_into/ddl/region_delete.sql | 1 +
.../multi_cluster/copy_into/ddl/revenue1.sql | 11 +
.../copy_into/ddl/revenue1_delete.sql | 1 +
.../multi_cluster/copy_into/ddl/supplier.sql | 11 +
.../copy_into/ddl/supplier_delete.sql | 1 +
.../multi_cluster/copy_into/sync_laod.groovy | 846 +++++++++++++++++++++
.../cloud_p0/multi_cluster/default_cluster.groovy | 172 +++++
.../multi_cluster_s3_load/ddl/customer.sql | 13 +
.../multi_cluster_s3_load/ddl/customer_delete.sql | 1 +
.../multi_cluster_s3_load/ddl/customer_load.sql | 6 +
.../multi_cluster_s3_load/ddl/lineitem.sql | 21 +
.../multi_cluster_s3_load/ddl/lineitem_delete.sql | 1 +
.../multi_cluster_s3_load/ddl/lineitem_load.sql | 6 +
.../multi_cluster_s3_load/ddl/nation.sql | 8 +
.../multi_cluster_s3_load/ddl/nation_delete.sql | 1 +
.../multi_cluster_s3_load/ddl/nation_load.sql | 6 +
.../multi_cluster_s3_load/ddl/orders.sql | 15 +
.../multi_cluster_s3_load/ddl/orders_delete.sql | 1 +
.../multi_cluster_s3_load/ddl/orders_load.sql | 6 +
.../multi_cluster_s3_load/ddl/part.sql | 15 +
.../multi_cluster_s3_load/ddl/part_delete.sql | 1 +
.../multi_cluster_s3_load/ddl/part_load.sql | 6 +
.../multi_cluster_s3_load/ddl/partsupp.sql | 11 +
.../multi_cluster_s3_load/ddl/partsupp_delete.sql | 1 +
.../multi_cluster_s3_load/ddl/partsupp_load.sql | 6 +
.../multi_cluster_s3_load/ddl/region.sql | 9 +
.../multi_cluster_s3_load/ddl/region_delete.sql | 1 +
.../multi_cluster_s3_load/ddl/region_load.sql | 6 +
.../multi_cluster_s3_load/ddl/supplier.sql | 12 +
.../multi_cluster_s3_load/ddl/supplier_delete.sql | 1 +
.../multi_cluster_s3_load/ddl/supplier_load.sql | 6 +
.../multi_cluster_s3_load/load.groovy | 166 ++++
.../op/test_drop_and_add_cluster.groovy | 221 ++++++
.../cloud_p0/multi_cluster/rename_cluster.groovy | 115 +++
.../routine_load/test_routine_load.groovy | 189 +++++
.../test_bitmap_index_with_drop_cluster.groovy | 143 ++++
.../test_bitmap_index_with_readd_cluster.groovy | 154 ++++
...test_materialized_view_with_drop_cluster.groovy | 123 +++
...est_materialized_view_with_readd_cluster.groovy | 155 ++++
.../test_rollup_with_drop_cluster.groovy | 122 +++
.../test_rollup_with_readd_cluster.groovy | 156 ++++
.../test_schema_change_with_drop_cluster.groovy | 120 +++
.../test_schema_change_with_readd_cluster.groovy | 169 ++++
.../multi_cluster/smooth_upgrade/ddl/customer.sql | 15 +
.../smooth_upgrade/ddl/customer_delete.sql | 1 +
.../smooth_upgrade/ddl/customer_load.sql | 6 +
.../smooth_upgrade/ddl/customer_load_sequence.sql | 7 +
.../smooth_upgrade/ddl/customer_part_delete.sql | 1 +
.../smooth_upgrade/ddl/customer_sequence.sql | 17 +
.../multi_cluster/smooth_upgrade/ddl/lineitem.sql | 23 +
.../smooth_upgrade/ddl/lineitem_delete.sql | 1 +
.../smooth_upgrade/ddl/lineitem_load.sql | 6 +
.../smooth_upgrade/ddl/lineitem_load_sequence.sql | 7 +
.../smooth_upgrade/ddl/lineitem_part_delete.sql | 1 +
.../smooth_upgrade/ddl/lineitem_sequence.sql | 25 +
.../multi_cluster/smooth_upgrade/ddl/nation.sql | 11 +
.../smooth_upgrade/ddl/nation_delete.sql | 1 +
.../smooth_upgrade/ddl/nation_load.sql | 6 +
.../smooth_upgrade/ddl/nation_load_sequence.sql | 7 +
.../smooth_upgrade/ddl/nation_part_delete.sql | 1 +
.../smooth_upgrade/ddl/nation_sequence.sql | 13 +
.../multi_cluster/smooth_upgrade/ddl/orders.sql | 15 +
.../smooth_upgrade/ddl/orders_delete.sql | 1 +
.../smooth_upgrade/ddl/orders_load.sql | 6 +
.../smooth_upgrade/ddl/orders_load_sequence.sql | 7 +
.../smooth_upgrade/ddl/orders_part_delete.sql | 1 +
.../smooth_upgrade/ddl/orders_sequence.sql | 17 +
.../multi_cluster/smooth_upgrade/ddl/part.sql | 16 +
.../smooth_upgrade/ddl/part_delete.sql | 1 +
.../multi_cluster/smooth_upgrade/ddl/part_load.sql | 6 +
.../smooth_upgrade/ddl/part_load_sequence.sql | 7 +
.../smooth_upgrade/ddl/part_part_delete.sql | 1 +
.../smooth_upgrade/ddl/part_sequence.sql | 18 +
.../multi_cluster/smooth_upgrade/ddl/partsupp.sql | 12 +
.../smooth_upgrade/ddl/partsupp_delete.sql | 1 +
.../smooth_upgrade/ddl/partsupp_load.sql | 6 +
.../smooth_upgrade/ddl/partsupp_load_sequence.sql | 7 +
.../smooth_upgrade/ddl/partsupp_part_delete.sql | 1 +
.../smooth_upgrade/ddl/partsupp_sequence.sql | 14 +
.../multi_cluster/smooth_upgrade/ddl/region.sql | 10 +
.../smooth_upgrade/ddl/region_delete.sql | 1 +
.../smooth_upgrade/ddl/region_load.sql | 6 +
.../smooth_upgrade/ddl/region_load_sequence.sql | 7 +
.../smooth_upgrade/ddl/region_part_delete.sql | 1 +
.../smooth_upgrade/ddl/region_sequence.sql | 12 +
.../multi_cluster/smooth_upgrade/ddl/supplier.sql | 13 +
.../smooth_upgrade/ddl/supplier_delete.sql | 1 +
.../smooth_upgrade/ddl/supplier_load.sql | 6 +
.../smooth_upgrade/ddl/supplier_load_sequence.sql | 7 +
.../smooth_upgrade/ddl/supplier_part_delete.sql | 1 +
.../smooth_upgrade/ddl/supplier_sequence.sql | 15 +
.../smooth_upgrade/smooth_upgrade.groovy | 165 ++++
.../multi_cluster/stream_load/stream_load.groovy | 322 ++++++++
.../stream_load/stream_load_2pc.groovy | 396 ++++++++++
.../stream_load/stream_load_lb.groovy | 227 ++++++
.../cloud_p0/multi_cluster/table_rebalance.groovy | 175 +++++
.../test_apsaradb_internal_stage.groovy | 199 +++++
.../multi_cluster/test_cloud_cluster.groovy | 202 +++++
.../multi_cluster/test_drop_cluster.groovy | 96 +++
.../cloud_p0/multi_cluster/test_drop_userid.groovy | 121 +++
.../test_group_commit_multi_cluster.groovy | 129 ++++
.../multi_cluster/test_overdue_instance.groovy | 243 ++++++
143 files changed, 6298 insertions(+)
diff --git a/regression-test/data/cloud_p0/multi_cluster/routine_load/test_routine_load.out b/regression-test/data/cloud_p0/multi_cluster/routine_load/test_routine_load.out
new file mode 100644
index 00000000000..8360519bb74
--- /dev/null
+++ b/regression-test/data/cloud_p0/multi_cluster/routine_load/test_routine_load.out
@@ -0,0 +1,33 @@
+-- This file is automatically generated. You should know what you did if you want to edit this
+-- !q1 --
+0 abc 0
+1 abc 2
+10 abc 20
+11 abc 22
+12 abc 24
+13 abc 26
+14 abc 28
+15 abc 30
+16 abc 32
+17 abc 34
+18 abc 36
+19 abc 38
+2 abc 4
+20 abc 40
+21 abc 42
+22 abc 44
+23 abc 46
+24 abc 48
+25 abc 50
+26 abc 52
+27 abc 54
+28 abc 56
+29 abc 58
+3 abc 6
+4 abc 8
+5 abc 10
+6 abc 12
+7 abc 14
+8 abc 16
+9 abc 18
+
diff --git a/regression-test/data/cloud_p0/multi_cluster/schema_change/test_schema_change_with_readd_cluster.out b/regression-test/data/cloud_p0/multi_cluster/schema_change/test_schema_change_with_readd_cluster.out
new file mode 100644
index 00000000000..b6ac74ebc17
--- /dev/null
+++ b/regression-test/data/cloud_p0/multi_cluster/schema_change/test_schema_change_with_readd_cluster.out
@@ -0,0 +1,6 @@
+-- This file is automatically generated. You should know what you did if you want to edit this
+-- !order --
+1 1 100 100
+2 1 100 100
+3 1 100 100
+
diff --git a/regression-test/data/cloud_p0/multi_cluster/stream_load/all_types.csv b/regression-test/data/cloud_p0/multi_cluster/stream_load/all_types.csv
new file mode 100644
index 00000000000..64d38b1db8f
--- /dev/null
+++ b/regression-test/data/cloud_p0/multi_cluster/stream_load/all_types.csv
@@ -0,0 +1,20 @@
+0,2,3,4,5,6.6,7.7,8.8,abc,def,ghiaaaaaa,2020-10-10,2020-10-10 11:12:59
+1,2,3,4,5,6.6,7.7,8.8,abc,def,ghiaaaaaa,2020-10-10,2020-10-10 11:12:59
+2,2,3,4,5,6.6,7.7,8.8,abc,def,ghiaaaaaa,2020-10-10,2020-10-10 11:12:59
+3,2,3,4,5,6.6,7.7,8.8,abc,def,ghiaaaaaa,2020-10-10,2020-10-10 11:12:59
+4,2,3,4,5,6.6,7.7,8.8,abc,def,ghiaaaaaa,2020-10-10,2020-10-10 11:12:59
+5,2,3,4,5,6.6,7.7,8.8,abc,def,ghiaaaaaa,2020-10-10,2020-10-10 11:12:59
+6,2,3,4,5,6.6,7.7,8.8,abc,def,ghiaaaaaa,2020-10-10,2020-10-10 11:12:59
+7,2,3,4,5,6.6,7.7,8.8,abc,def,ghiaaaaaa,2020-10-10,2020-10-10 11:12:59
+8,2,3,4,5,6.6,7.7,8.8,abc,def,ghiaaaaaa,2020-10-10,2020-10-10 11:12:59
+9,2,3,4,5,6.6,7.7,8.8,abc,def,ghiaaaaaa,2020-10-10,2020-10-10 11:12:59
+10,2,3,4,5,6.6,7.7,8.8,abc,def,ghiaaaaaa,2020-10-10,2020-10-10 11:12:59
+11,2,3,4,5,6.6,7.7,8.8,abc,def,ghiaaaaaa,2020-10-10,2020-10-10 11:12:59
+12,2,3,4,5,6.6,7.7,8.8,abc,def,ghiaaaaaa,2020-10-10,2020-10-10 11:12:59
+13,2,3,4,5,6.6,7.7,8.8,abc,def,ghiaaaaaa,2020-10-10,2020-10-10 11:12:59
+14,2,3,4,5,6.6,7.7,8.8,abc,def,ghiaaaaaa,2020-10-10,2020-10-10 11:12:59
+15,2,3,4,5,6.6,7.7,8.8,abc,def,ghiaaaaaa,2020-10-10,2020-10-10 11:12:59
+16,2,3,4,5,6.6,7.7,8.8,abc,def,ghiaaaaaa,2020-10-10,2020-10-10 11:12:59
+17,2,3,4,5,6.6,7.7,8.8,abc,def,ghiaaaaaa,2020-10-10,2020-10-10 11:12:59
+18,2,3,4,5,6.6,7.7,8.8,abc,def,ghiaaaaaa,2020-10-10,2020-10-10 11:12:59
+19,2,3,4,5,6.6,7.7,8.8,abc,def,ghiaaaaaa,2020-10-10,2020-10-10 11:12:59
diff --git a/regression-test/data/cloud_p0/multi_cluster/stream_load/stream_load.out b/regression-test/data/cloud_p0/multi_cluster/stream_load/stream_load.out
new file mode 100644
index 00000000000..adc5cddaaad
--- /dev/null
+++ b/regression-test/data/cloud_p0/multi_cluster/stream_load/stream_load.out
@@ -0,0 +1,13 @@
+-- This file is automatically generated. You should know what you did if you want to edit this
+-- !all11 --
+20
+
+-- !all12 --
+11
+
+-- !all11 --
+20
+
+-- !all12 --
+11
+
diff --git a/regression-test/data/cloud_p0/multi_cluster/stream_load/stream_load_2pc.out b/regression-test/data/cloud_p0/multi_cluster/stream_load/stream_load_2pc.out
new file mode 100644
index 00000000000..4f4362cda12
--- /dev/null
+++ b/regression-test/data/cloud_p0/multi_cluster/stream_load/stream_load_2pc.out
@@ -0,0 +1,25 @@
+-- This file is automatically generated. You should know what you did if you want to edit this
+-- !all11 --
+0
+
+-- !all12 --
+0
+
+-- !all11 --
+20
+
+-- !all12 --
+11
+
+-- !all11 --
+0
+
+-- !all12 --
+0
+
+-- !all11 --
+20
+
+-- !all12 --
+11
+
diff --git a/regression-test/data/cloud_p0/multi_cluster/stream_load/stream_load_lb.out b/regression-test/data/cloud_p0/multi_cluster/stream_load/stream_load_lb.out
new file mode 100644
index 00000000000..22ba015bf6c
--- /dev/null
+++ b/regression-test/data/cloud_p0/multi_cluster/stream_load/stream_load_lb.out
@@ -0,0 +1,25 @@
+-- This file is automatically generated. You should know what you did if you want to edit this
+-- !q1 --
+20
+
+-- !q2 --
+11
+
+-- !q3 --
+40
+
+-- !q4 --
+22
+
+-- !q7 --
+60
+
+-- !q8 --
+33
+
+-- !q9 --
+80
+
+-- !q10 --
+44
+
diff --git a/regression-test/data/cloud_p0/multi_cluster/test_apsaradb_internal_stage.out b/regression-test/data/cloud_p0/multi_cluster/test_apsaradb_internal_stage.out
new file mode 100644
index 00000000000..b37cf8928b2
--- /dev/null
+++ b/regression-test/data/cloud_p0/multi_cluster/test_apsaradb_internal_stage.out
@@ -0,0 +1,7 @@
+-- This file is automatically generated. You should know what you did if you want to edit this
+-- !sql --
+10
+
+-- !sql --
+10
+
diff --git a/regression-test/pipeline/cloud_p0/conf/regression-conf-custom.groovy b/regression-test/pipeline/cloud_p0/conf/regression-conf-custom.groovy
index 8ede4da7a33..d47e204324a 100644
--- a/regression-test/pipeline/cloud_p0/conf/regression-conf-custom.groovy
+++ b/regression-test/pipeline/cloud_p0/conf/regression-conf-custom.groovy
@@ -72,6 +72,7 @@ excludeSuites = "000_the_start_sentinel_do_not_touch," + // keep this line as th
excludeDirectories = "000_the_start_sentinel_do_not_touch," + // keep this line as the first line
"external_table_p0," + // run on external pipeline
"cloud/multi_cluster," + // run in specific regression pipeline
+ "cloud_p0/multi_cluster," + // run in specific regression pipeline
"cloud_p0/cache," +
"workload_manager_p1," +
"nereids_rules_p0/subquery," +
diff --git a/regression-test/pipeline/nonConcurrent/conf/regression-conf.groovy b/regression-test/pipeline/nonConcurrent/conf/regression-conf.groovy
index 699e6536b99..1dfcad207d5 100644
--- a/regression-test/pipeline/nonConcurrent/conf/regression-conf.groovy
+++ b/regression-test/pipeline/nonConcurrent/conf/regression-conf.groovy
@@ -71,6 +71,7 @@ excludeDirectories = "000_the_start_sentinel_do_not_touch," + // keep this line
"variant_github_events_new_p2," +
"hdfs_vault_p2," +
"nereids_p0/hbo," +
+ "cloud_p0/multi_cluster," +
"zzz_the_end_sentinel_do_not_touch"// keep this line as the last line
// for test csv with header
diff --git a/regression-test/suites/cloud_p0/multi_cluster/async_copy_into/async_load.groovy b/regression-test/suites/cloud_p0/multi_cluster/async_copy_into/async_load.groovy
new file mode 100644
index 00000000000..0e346d0389a
--- /dev/null
+++ b/regression-test/suites/cloud_p0/multi_cluster/async_copy_into/async_load.groovy
@@ -0,0 +1,275 @@
+// 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 groovy.json.JsonOutput
+
+suite("async_load") {
+ def externalStageName = "regression_test_tpch"
+ def prefix = "tpch/sf1"
+
+ List<String> ipList = new ArrayList<>()
+ List<String> hbPortList = new ArrayList<>()
+ List<String> httpPortList = new ArrayList<>()
+ List<String> beUniqueIdList = new ArrayList<>()
+
+ String[] bes = context.config.multiClusterBes.split(',');
+ println("the value is " + context.config.multiClusterBes);
+ for(String values : bes) {
+ println("the value is " + values);
+ String[] beInfo = values.split(':');
+ ipList.add(beInfo[0]);
+ hbPortList.add(beInfo[1]);
+ httpPortList.add(beInfo[2]);
+ beUniqueIdList.add(beInfo[3]);
+ }
+
+ println("the ip is " + ipList);
+ println("the heartbeat port is " + hbPortList);
+ println("the http port is " + httpPortList);
+ println("the be unique id is " + beUniqueIdList);
+
+ sleep(1000)
+ for (unique_id : beUniqueIdList) {
+ resp = get_cluster.call(unique_id);
+ for (cluster : resp) {
+ if (cluster.type == "COMPUTE") {
+ drop_cluster.call(cluster.cluster_name, cluster.cluster_id);
+ }
+ }
+ }
+ wait_cluster_change()
+
+ List<List<Object>> result = sql "show clusters"
+ assertTrue(result.size() == 0);
+
+ add_cluster.call(beUniqueIdList[0], ipList[0], hbPortList[0],
+ "regression_cluster_name0", "regression_cluster_id0");
+ add_cluster.call(beUniqueIdList[1], ipList[1], hbPortList[1],
+ "regression_cluster_name1", "regression_cluster_id1");
+ wait_cluster_change()
+
+ result = sql "show clusters"
+ assertEquals(result.size(), 2);
+
+ sql "use @regression_cluster_name1"
+ result = sql "show clusters"
+
+ def before_cluster0_load_rows = get_be_metric(ipList[0], httpPortList[0], "load_rows");
+ log.info("before_cluster0_load_rows : ${before_cluster0_load_rows}".toString())
+ def before_cluster0_flush = get_be_metric(ipList[0], httpPortList[0], "memtable_flush_total");
+ log.info("before_cluster0_flush : ${before_cluster0_flush}".toString())
+
+ def before_cluster1_load_rows = get_be_metric(ipList[1], httpPortList[1], "load_rows");
+ log.info("before_cluster1_load_rows : ${before_cluster1_load_rows}".toString())
+ def before_cluster1_flush = get_be_metric(ipList[1], httpPortList[1], "memtable_flush_total");
+ log.info("before_cluster1_flush : ${before_cluster1_flush}".toString())
+
+ // tpch_sf1_p1 is writted to test unique key table merge correctly.
+ // It creates unique key table and sets bucket num to 1 in order to make sure that
+ // many rowsets will be created during loading and then the merge process will be triggered.
+
+ def tables = ["customer", "lineitem", "nation", "orders", "part", "partsupp", "region", "supplier"]
+
+ for (String table in tables) {
+ sql """ DROP TABLE IF EXISTS ${table}; """
+ sql new File("""${context.file.parent}/ddl/${table}.sql""").text
+ }
+
+ sql """
+ create stage if not exists ${externalStageName} properties(
+ 'endpoint' = '${getS3Endpoint()}' ,
+ 'region' = '${getS3Region()}' ,
+ 'bucket' = '${getS3BucketName()}' ,
+ 'prefix' = 'regression' ,
+ 'ak' = '${getS3AK()}' ,
+ 'sk' = '${getS3SK()}' ,
+ 'provider' = '${getProvider()}',
+ 'access_type' = 'aksk',
+ 'default.file.column_separator' = "|"
+ );
+ """
+
+ for (String tableName in tables) {
+ result = sql " copy into ${tableName} from @${externalStageName}('${prefix}/${tableName}.csv.split00.gz') properties ('file.compression' = 'gz', 'copy.async' = 'true'); "
+ logger.info("copy result: " + result)
+ assertTrue(result.size() == 1)
+ assertTrue(result[0].size() == 8)
+ //assertTrue(result[0][1].equals("FINISHED"))
+ id = result[0][0]
+ while(true) {
+ result = sql "show copy where id = \'${id}\'"
+ logger.info("copy result: " + result)
+ if (result[0][3] == "FINISHED" || result[0][3] == "CANCELLED") {
+ break;
+ }
+ sleep(3000);
+ }
+
+ result = sql " copy into ${tableName} from @${externalStageName}('${prefix}/${tableName}.csv.split01.gz') properties ('file.compression' = 'gz', 'copy.async' = 'true'); "
+ logger.info("copy result: " + result)
+ assertTrue(result.size() == 1)
+ assertTrue(result[0].size() == 8)
+ //assertTrue(result[0][1].equals("FINISHED"))
+ id = result[0][0]
+ while(true) {
+ result = sql "show copy where id = \'${id}\'"
+ logger.info("show copy result: " + result)
+ if (result[0][3] == "FINISHED" || result[0][3] == "CANCELLED") {
+ break;
+ }
+ sleep(3000);
+ }
+ }
+
+ def table = "revenue1"
+ sql new File("""${context.file.parent}/ddl/${table}_delete.sql""").text
+ sql new File("""${context.file.parent}/ddl/${table}.sql""").text
+
+ def after_cluster0_load_rows = get_be_metric(ipList[0], httpPortList[0], "load_rows");
+ log.info("after_cluster0_load_rows : ${after_cluster0_load_rows}".toString())
+ def after_cluster0_flush = get_be_metric(ipList[0], httpPortList[0], "memtable_flush_total");
+ log.info("after_cluster0_flush : ${after_cluster0_flush}".toString())
+
+ def after_cluster1_load_rows = get_be_metric(ipList[1], httpPortList[1], "load_rows");
+ log.info("after_cluster1_load_rows : ${after_cluster1_load_rows}".toString())
+ def after_cluster1_flush = get_be_metric(ipList[1], httpPortList[1], "memtable_flush_total");
+ log.info("after_cluster1_flush : ${after_cluster1_flush}".toString())
+
+ assertTrue(before_cluster0_load_rows == after_cluster0_load_rows)
+ assertTrue(before_cluster0_flush == after_cluster0_flush)
+
+ assertTrue(before_cluster1_load_rows < after_cluster1_load_rows)
+ assertTrue(before_cluster1_flush < after_cluster1_flush)
+
+ sql "use @regression_cluster_name0"
+ result = sql "show clusters"
+ sql """ set enable_profile = true; """
+
+ def before_cluster0_query_scan_rows = get_be_metric(ipList[0], httpPortList[0], "query_scan_rows");
+ log.info("before_cluster0_query_scan_rows : ${before_cluster0_query_scan_rows}".toString())
+ def before_cluster1_query_scan_rows = get_be_metric(ipList[1], httpPortList[1], "query_scan_rows");
+ log.info("before_cluster1_query_scan_rows : ${before_cluster1_query_scan_rows}".toString())
+
+ sql """
+ SELECT
+ s_acctbal,
+ s_name,
+ n_name,
+ p_partkey,
+ p_mfgr,
+ s_address,
+ s_phone,
+ s_comment
+ FROM
+ part,
+ supplier,
+ partsupp,
+ nation,
+ region
+ WHERE
+ p_partkey = ps_partkey
+ AND s_suppkey = ps_suppkey
+ AND p_size = 15
+ AND p_type LIKE '%BRASS'
+ AND s_nationkey = n_nationkey
+ AND n_regionkey = r_regionkey
+ AND r_name = 'EUROPE'
+ AND ps_supplycost = (
+ SELECT min(ps_supplycost)
+ FROM
+ partsupp, supplier,
+ nation, region
+ WHERE
+ p_partkey = ps_partkey
+ AND s_suppkey = ps_suppkey
+ AND s_nationkey = n_nationkey
+ AND n_regionkey = r_regionkey
+ AND r_name = 'EUROPE'
+ )
+ ORDER BY
+ s_acctbal DESC,
+ n_name,
+ s_name,
+ p_partkey
+ LIMIT 100
+ """
+
+ def after_cluster0_query_scan_rows = get_be_metric(ipList[0], httpPortList[0], "query_scan_rows");
+ log.info("after_cluster0_query_scan_rows : ${after_cluster0_query_scan_rows}".toString())
+ def after_cluster1_query_scan_rows = get_be_metric(ipList[1], httpPortList[1], "query_scan_rows");
+ log.info("after_cluster1_query_scan_rows : ${after_cluster1_query_scan_rows}".toString())
+
+ assertTrue(before_cluster0_query_scan_rows < after_cluster0_query_scan_rows)
+ assertTrue(before_cluster1_query_scan_rows == after_cluster1_query_scan_rows)
+
+ before_cluster0_query_scan_rows = get_be_metric(ipList[0], httpPortList[0], "query_scan_rows");
+ log.info("before_cluster0_query_scan_rows : ${before_cluster0_query_scan_rows}".toString())
+ before_cluster1_query_scan_rows = get_be_metric(ipList[1], httpPortList[1], "query_scan_rows");
+ log.info("before_cluster1_query_scan_rows : ${before_cluster1_query_scan_rows}".toString())
+
+ sql """
+ SELECT /*+ SET_VAR(cloud_cluster = 'regression_cluster_name1') */
+ s_acctbal,
+ s_name,
+ n_name,
+ p_partkey,
+ p_mfgr,
+ s_address,
+ s_phone,
+ s_comment
+ FROM
+ part,
+ supplier,
+ partsupp,
+ nation,
+ region
+ WHERE
+ p_partkey = ps_partkey
+ AND s_suppkey = ps_suppkey
+ AND p_size = 15
+ AND p_type LIKE '%BRASS'
+ AND s_nationkey = n_nationkey
+ AND n_regionkey = r_regionkey
+ AND r_name = 'EUROPE'
+ AND ps_supplycost = (
+ SELECT min(ps_supplycost)
+ FROM
+ partsupp, supplier,
+ nation, region
+ WHERE
+ p_partkey = ps_partkey
+ AND s_suppkey = ps_suppkey
+ AND s_nationkey = n_nationkey
+ AND n_regionkey = r_regionkey
+ AND r_name = 'EUROPE'
+ )
+ ORDER BY
+ s_acctbal DESC,
+ n_name,
+ s_name,
+ p_partkey
+ LIMIT 100
+ """
+
+ after_cluster0_query_scan_rows = get_be_metric(ipList[0], httpPortList[0], "query_scan_rows");
+ log.info("after_cluster0_query_scan_rows : ${after_cluster0_query_scan_rows}".toString())
+ after_cluster1_query_scan_rows = get_be_metric(ipList[1], httpPortList[1], "query_scan_rows");
+ log.info("after_cluster1_query_scan_rows : ${after_cluster1_query_scan_rows}".toString())
+
+ assertTrue(before_cluster0_query_scan_rows == after_cluster0_query_scan_rows)
+ assertTrue(before_cluster1_query_scan_rows < after_cluster1_query_scan_rows)
+}
diff --git a/regression-test/suites/cloud_p0/multi_cluster/async_copy_into/ddl/customer.sql b/regression-test/suites/cloud_p0/multi_cluster/async_copy_into/ddl/customer.sql
new file mode 100644
index 00000000000..aebc1380721
--- /dev/null
+++ b/regression-test/suites/cloud_p0/multi_cluster/async_copy_into/ddl/customer.sql
@@ -0,0 +1,13 @@
+CREATE TABLE IF NOT EXISTS customer (
+ C_CUSTKEY INTEGER NOT NULL,
+ C_NAME VARCHAR(25) NOT NULL,
+ C_ADDRESS VARCHAR(40) NOT NULL,
+ C_NATIONKEY INTEGER NOT NULL,
+ C_PHONE CHAR(15) NOT NULL,
+ C_ACCTBAL DECIMAL(15,2) NOT NULL,
+ C_MKTSEGMENT CHAR(10) NOT NULL,
+ C_COMMENT VARCHAR(117) NOT NULL
+)
+UNIQUE KEY(C_CUSTKEY)
+DISTRIBUTED BY HASH(C_CUSTKEY) BUCKETS 1
+
diff --git a/regression-test/suites/cloud_p0/multi_cluster/async_copy_into/ddl/customer_delete.sql b/regression-test/suites/cloud_p0/multi_cluster/async_copy_into/ddl/customer_delete.sql
new file mode 100644
index 00000000000..a214fe920c9
--- /dev/null
+++ b/regression-test/suites/cloud_p0/multi_cluster/async_copy_into/ddl/customer_delete.sql
@@ -0,0 +1 @@
+DELETE FROM customer where C_CUSTKEY >= 0;
diff --git a/regression-test/suites/cloud_p0/multi_cluster/async_copy_into/ddl/lineitem.sql b/regression-test/suites/cloud_p0/multi_cluster/async_copy_into/ddl/lineitem.sql
new file mode 100644
index 00000000000..257c9cca29e
--- /dev/null
+++ b/regression-test/suites/cloud_p0/multi_cluster/async_copy_into/ddl/lineitem.sql
@@ -0,0 +1,21 @@
+CREATE TABLE IF NOT EXISTS lineitem (
+ L_ORDERKEY INTEGER NOT NULL,
+ L_PARTKEY INTEGER NOT NULL,
+ L_SUPPKEY INTEGER NOT NULL,
+ L_LINENUMBER INTEGER NOT NULL,
+ L_QUANTITY DECIMAL(15,2) NOT NULL,
+ L_EXTENDEDPRICE DECIMAL(15,2) NOT NULL,
+ L_DISCOUNT DECIMAL(15,2) NOT NULL,
+ L_TAX DECIMAL(15,2) NOT NULL,
+ L_RETURNFLAG CHAR(1) NOT NULL,
+ L_LINESTATUS CHAR(1) NOT NULL,
+ L_SHIPDATE DATE NOT NULL,
+ L_COMMITDATE DATE NOT NULL,
+ L_RECEIPTDATE DATE NOT NULL,
+ L_SHIPINSTRUCT CHAR(25) NOT NULL,
+ L_SHIPMODE CHAR(10) NOT NULL,
+ L_COMMENT VARCHAR(44) NOT NULL
+)
+UNIQUE KEY(L_ORDERKEY, L_PARTKEY, L_SUPPKEY, L_LINENUMBER)
+DISTRIBUTED BY HASH(L_ORDERKEY) BUCKETS 1
+
diff --git a/regression-test/suites/cloud_p0/multi_cluster/async_copy_into/ddl/lineitem_delete.sql b/regression-test/suites/cloud_p0/multi_cluster/async_copy_into/ddl/lineitem_delete.sql
new file mode 100644
index 00000000000..6dbbe49ed0b
--- /dev/null
+++ b/regression-test/suites/cloud_p0/multi_cluster/async_copy_into/ddl/lineitem_delete.sql
@@ -0,0 +1 @@
+DELETE from lineitem where L_ORDERKEY >= 0;
diff --git a/regression-test/suites/cloud_p0/multi_cluster/async_copy_into/ddl/nation.sql b/regression-test/suites/cloud_p0/multi_cluster/async_copy_into/ddl/nation.sql
new file mode 100644
index 00000000000..574565eab67
--- /dev/null
+++ b/regression-test/suites/cloud_p0/multi_cluster/async_copy_into/ddl/nation.sql
@@ -0,0 +1,9 @@
+CREATE TABLE IF NOT EXISTS nation (
+ N_NATIONKEY INTEGER NOT NULL,
+ N_NAME CHAR(25) NOT NULL,
+ N_REGIONKEY INTEGER NOT NULL,
+ N_COMMENT VARCHAR(152)
+)
+UNIQUE KEY(N_NATIONKEY)
+DISTRIBUTED BY HASH(N_NATIONKEY) BUCKETS 1
+
diff --git a/regression-test/suites/cloud_p0/multi_cluster/async_copy_into/ddl/nation_delete.sql b/regression-test/suites/cloud_p0/multi_cluster/async_copy_into/ddl/nation_delete.sql
new file mode 100644
index 00000000000..9fb7c085a6f
--- /dev/null
+++ b/regression-test/suites/cloud_p0/multi_cluster/async_copy_into/ddl/nation_delete.sql
@@ -0,0 +1 @@
+DELETE from nation where N_NATIONKEY >= 0;
diff --git a/regression-test/suites/cloud_p0/multi_cluster/async_copy_into/ddl/orders.sql b/regression-test/suites/cloud_p0/multi_cluster/async_copy_into/ddl/orders.sql
new file mode 100644
index 00000000000..71ae0815474
--- /dev/null
+++ b/regression-test/suites/cloud_p0/multi_cluster/async_copy_into/ddl/orders.sql
@@ -0,0 +1,14 @@
+CREATE TABLE IF NOT EXISTS orders (
+ O_ORDERKEY INTEGER NOT NULL,
+ O_CUSTKEY INTEGER NOT NULL,
+ O_ORDERSTATUS CHAR(1) NOT NULL,
+ O_TOTALPRICE DECIMAL(15,2) NOT NULL,
+ O_ORDERDATE DATE NOT NULL,
+ O_ORDERPRIORITY CHAR(15) NOT NULL,
+ O_CLERK CHAR(15) NOT NULL,
+ O_SHIPPRIORITY INTEGER NOT NULL,
+ O_COMMENT VARCHAR(79) NOT NULL
+)
+UNIQUE KEY(O_ORDERKEY)
+DISTRIBUTED BY HASH(O_ORDERKEY) BUCKETS 1
+
diff --git a/regression-test/suites/cloud_p0/multi_cluster/async_copy_into/ddl/orders_delete.sql b/regression-test/suites/cloud_p0/multi_cluster/async_copy_into/ddl/orders_delete.sql
new file mode 100644
index 00000000000..5199d8f296d
--- /dev/null
+++ b/regression-test/suites/cloud_p0/multi_cluster/async_copy_into/ddl/orders_delete.sql
@@ -0,0 +1 @@
+delete from orders where O_ORDERKEY >= 0;
diff --git a/regression-test/suites/cloud_p0/multi_cluster/async_copy_into/ddl/part.sql b/regression-test/suites/cloud_p0/multi_cluster/async_copy_into/ddl/part.sql
new file mode 100644
index 00000000000..b1edbcebd12
--- /dev/null
+++ b/regression-test/suites/cloud_p0/multi_cluster/async_copy_into/ddl/part.sql
@@ -0,0 +1,14 @@
+CREATE TABLE IF NOT EXISTS part (
+ P_PARTKEY INTEGER NOT NULL,
+ P_NAME VARCHAR(55) NOT NULL,
+ P_MFGR CHAR(25) NOT NULL,
+ P_BRAND CHAR(10) NOT NULL,
+ P_TYPE VARCHAR(25) NOT NULL,
+ P_SIZE INTEGER NOT NULL,
+ P_CONTAINER CHAR(10) NOT NULL,
+ P_RETAILPRICE DECIMAL(15,2) NOT NULL,
+ P_COMMENT VARCHAR(23) NOT NULL
+)
+UNIQUE KEY(P_PARTKEY)
+DISTRIBUTED BY HASH(P_PARTKEY) BUCKETS 1
+
diff --git a/regression-test/suites/cloud_p0/multi_cluster/async_copy_into/ddl/part_delete.sql b/regression-test/suites/cloud_p0/multi_cluster/async_copy_into/ddl/part_delete.sql
new file mode 100644
index 00000000000..ee1a2e51045
--- /dev/null
+++ b/regression-test/suites/cloud_p0/multi_cluster/async_copy_into/ddl/part_delete.sql
@@ -0,0 +1 @@
+delete from part where P_PARTKEY>=0;
diff --git a/regression-test/suites/cloud_p0/multi_cluster/async_copy_into/ddl/partsupp.sql b/regression-test/suites/cloud_p0/multi_cluster/async_copy_into/ddl/partsupp.sql
new file mode 100644
index 00000000000..3fa6d79fab2
--- /dev/null
+++ b/regression-test/suites/cloud_p0/multi_cluster/async_copy_into/ddl/partsupp.sql
@@ -0,0 +1,10 @@
+CREATE TABLE IF NOT EXISTS partsupp (
+ PS_PARTKEY INTEGER NOT NULL,
+ PS_SUPPKEY INTEGER NOT NULL,
+ PS_AVAILQTY INTEGER NOT NULL,
+ PS_SUPPLYCOST DECIMAL(15,2) NOT NULL,
+ PS_COMMENT VARCHAR(199) NOT NULL
+)
+UNIQUE KEY(PS_PARTKEY, PS_SUPPKEY)
+DISTRIBUTED BY HASH(PS_PARTKEY) BUCKETS 1
+
diff --git a/regression-test/suites/cloud_p0/multi_cluster/async_copy_into/ddl/partsupp_delete.sql b/regression-test/suites/cloud_p0/multi_cluster/async_copy_into/ddl/partsupp_delete.sql
new file mode 100644
index 00000000000..067cff3a6f0
--- /dev/null
+++ b/regression-test/suites/cloud_p0/multi_cluster/async_copy_into/ddl/partsupp_delete.sql
@@ -0,0 +1 @@
+delete from partsupp where PS_PARTKEY>=0;
diff --git a/regression-test/suites/cloud_p0/multi_cluster/async_copy_into/ddl/region.sql b/regression-test/suites/cloud_p0/multi_cluster/async_copy_into/ddl/region.sql
new file mode 100644
index 00000000000..adfedcdbbd4
--- /dev/null
+++ b/regression-test/suites/cloud_p0/multi_cluster/async_copy_into/ddl/region.sql
@@ -0,0 +1,8 @@
+CREATE TABLE IF NOT EXISTS region (
+ R_REGIONKEY INTEGER NOT NULL,
+ R_NAME CHAR(25) NOT NULL,
+ R_COMMENT VARCHAR(152)
+)
+UNIQUE KEY(R_REGIONKEY)
+DISTRIBUTED BY HASH(R_REGIONKEY) BUCKETS 1
+
diff --git a/regression-test/suites/cloud_p0/multi_cluster/async_copy_into/ddl/region_delete.sql b/regression-test/suites/cloud_p0/multi_cluster/async_copy_into/ddl/region_delete.sql
new file mode 100644
index 00000000000..4945d5105ab
--- /dev/null
+++ b/regression-test/suites/cloud_p0/multi_cluster/async_copy_into/ddl/region_delete.sql
@@ -0,0 +1 @@
+delete from region where R_REGIONKEY>=0;
diff --git a/regression-test/suites/cloud_p0/multi_cluster/async_copy_into/ddl/revenue1.sql b/regression-test/suites/cloud_p0/multi_cluster/async_copy_into/ddl/revenue1.sql
new file mode 100644
index 00000000000..a12a60a2848
--- /dev/null
+++ b/regression-test/suites/cloud_p0/multi_cluster/async_copy_into/ddl/revenue1.sql
@@ -0,0 +1,11 @@
+CREATE VIEW IF NOT EXISTS revenue1 AS
+ SELECT
+ l_suppkey AS supplier_no,
+ sum(l_extendedprice * (1 - l_discount)) AS total_revenue
+ FROM
+ lineitem
+ WHERE
+ l_shipdate >= DATE '1996-01-01'
+ AND l_shipdate < DATE '1996-01-01' + INTERVAL '3' MONTH
+GROUP BY
+ l_suppkey;
diff --git a/regression-test/suites/cloud_p0/multi_cluster/async_copy_into/ddl/revenue1_delete.sql b/regression-test/suites/cloud_p0/multi_cluster/async_copy_into/ddl/revenue1_delete.sql
new file mode 100644
index 00000000000..c2ee91a38a9
--- /dev/null
+++ b/regression-test/suites/cloud_p0/multi_cluster/async_copy_into/ddl/revenue1_delete.sql
@@ -0,0 +1 @@
+drop view IF EXISTS revenue1;
\ No newline at end of file
diff --git a/regression-test/suites/cloud_p0/multi_cluster/async_copy_into/ddl/supplier.sql b/regression-test/suites/cloud_p0/multi_cluster/async_copy_into/ddl/supplier.sql
new file mode 100644
index 00000000000..4eac18d07bc
--- /dev/null
+++ b/regression-test/suites/cloud_p0/multi_cluster/async_copy_into/ddl/supplier.sql
@@ -0,0 +1,11 @@
+CREATE TABLE IF NOT EXISTS supplier (
+ S_SUPPKEY INTEGER NOT NULL,
+ S_NAME CHAR(25) NOT NULL,
+ S_ADDRESS VARCHAR(40) NOT NULL,
+ S_NATIONKEY INTEGER NOT NULL,
+ S_PHONE CHAR(15) NOT NULL,
+ S_ACCTBAL DECIMAL(15,2) NOT NULL,
+ S_COMMENT VARCHAR(101) NOT NULL
+)
+UNIQUE KEY(S_SUPPKEY)
+DISTRIBUTED BY HASH(S_SUPPKEY) BUCKETS 1
diff --git a/regression-test/suites/cloud_p0/multi_cluster/async_copy_into/ddl/supplier_delete.sql b/regression-test/suites/cloud_p0/multi_cluster/async_copy_into/ddl/supplier_delete.sql
new file mode 100644
index 00000000000..d8327a8f856
--- /dev/null
+++ b/regression-test/suites/cloud_p0/multi_cluster/async_copy_into/ddl/supplier_delete.sql
@@ -0,0 +1 @@
+delete from supplier where S_SUPPKEY>=0;
diff --git a/regression-test/suites/cloud_p0/multi_cluster/copy_into/ddl/customer.sql b/regression-test/suites/cloud_p0/multi_cluster/copy_into/ddl/customer.sql
new file mode 100644
index 00000000000..aebc1380721
--- /dev/null
+++ b/regression-test/suites/cloud_p0/multi_cluster/copy_into/ddl/customer.sql
@@ -0,0 +1,13 @@
+CREATE TABLE IF NOT EXISTS customer (
+ C_CUSTKEY INTEGER NOT NULL,
+ C_NAME VARCHAR(25) NOT NULL,
+ C_ADDRESS VARCHAR(40) NOT NULL,
+ C_NATIONKEY INTEGER NOT NULL,
+ C_PHONE CHAR(15) NOT NULL,
+ C_ACCTBAL DECIMAL(15,2) NOT NULL,
+ C_MKTSEGMENT CHAR(10) NOT NULL,
+ C_COMMENT VARCHAR(117) NOT NULL
+)
+UNIQUE KEY(C_CUSTKEY)
+DISTRIBUTED BY HASH(C_CUSTKEY) BUCKETS 1
+
diff --git a/regression-test/suites/cloud_p0/multi_cluster/copy_into/ddl/customer_delete.sql b/regression-test/suites/cloud_p0/multi_cluster/copy_into/ddl/customer_delete.sql
new file mode 100644
index 00000000000..a214fe920c9
--- /dev/null
+++ b/regression-test/suites/cloud_p0/multi_cluster/copy_into/ddl/customer_delete.sql
@@ -0,0 +1 @@
+DELETE FROM customer where C_CUSTKEY >= 0;
diff --git a/regression-test/suites/cloud_p0/multi_cluster/copy_into/ddl/lineitem.sql b/regression-test/suites/cloud_p0/multi_cluster/copy_into/ddl/lineitem.sql
new file mode 100644
index 00000000000..257c9cca29e
--- /dev/null
+++ b/regression-test/suites/cloud_p0/multi_cluster/copy_into/ddl/lineitem.sql
@@ -0,0 +1,21 @@
+CREATE TABLE IF NOT EXISTS lineitem (
+ L_ORDERKEY INTEGER NOT NULL,
+ L_PARTKEY INTEGER NOT NULL,
+ L_SUPPKEY INTEGER NOT NULL,
+ L_LINENUMBER INTEGER NOT NULL,
+ L_QUANTITY DECIMAL(15,2) NOT NULL,
+ L_EXTENDEDPRICE DECIMAL(15,2) NOT NULL,
+ L_DISCOUNT DECIMAL(15,2) NOT NULL,
+ L_TAX DECIMAL(15,2) NOT NULL,
+ L_RETURNFLAG CHAR(1) NOT NULL,
+ L_LINESTATUS CHAR(1) NOT NULL,
+ L_SHIPDATE DATE NOT NULL,
+ L_COMMITDATE DATE NOT NULL,
+ L_RECEIPTDATE DATE NOT NULL,
+ L_SHIPINSTRUCT CHAR(25) NOT NULL,
+ L_SHIPMODE CHAR(10) NOT NULL,
+ L_COMMENT VARCHAR(44) NOT NULL
+)
+UNIQUE KEY(L_ORDERKEY, L_PARTKEY, L_SUPPKEY, L_LINENUMBER)
+DISTRIBUTED BY HASH(L_ORDERKEY) BUCKETS 1
+
diff --git a/regression-test/suites/cloud_p0/multi_cluster/copy_into/ddl/lineitem_delete.sql b/regression-test/suites/cloud_p0/multi_cluster/copy_into/ddl/lineitem_delete.sql
new file mode 100644
index 00000000000..6dbbe49ed0b
--- /dev/null
+++ b/regression-test/suites/cloud_p0/multi_cluster/copy_into/ddl/lineitem_delete.sql
@@ -0,0 +1 @@
+DELETE from lineitem where L_ORDERKEY >= 0;
diff --git a/regression-test/suites/cloud_p0/multi_cluster/copy_into/ddl/nation.sql b/regression-test/suites/cloud_p0/multi_cluster/copy_into/ddl/nation.sql
new file mode 100644
index 00000000000..574565eab67
--- /dev/null
+++ b/regression-test/suites/cloud_p0/multi_cluster/copy_into/ddl/nation.sql
@@ -0,0 +1,9 @@
+CREATE TABLE IF NOT EXISTS nation (
+ N_NATIONKEY INTEGER NOT NULL,
+ N_NAME CHAR(25) NOT NULL,
+ N_REGIONKEY INTEGER NOT NULL,
+ N_COMMENT VARCHAR(152)
+)
+UNIQUE KEY(N_NATIONKEY)
+DISTRIBUTED BY HASH(N_NATIONKEY) BUCKETS 1
+
diff --git a/regression-test/suites/cloud_p0/multi_cluster/copy_into/ddl/nation_delete.sql b/regression-test/suites/cloud_p0/multi_cluster/copy_into/ddl/nation_delete.sql
new file mode 100644
index 00000000000..9fb7c085a6f
--- /dev/null
+++ b/regression-test/suites/cloud_p0/multi_cluster/copy_into/ddl/nation_delete.sql
@@ -0,0 +1 @@
+DELETE from nation where N_NATIONKEY >= 0;
diff --git a/regression-test/suites/cloud_p0/multi_cluster/copy_into/ddl/orders.sql b/regression-test/suites/cloud_p0/multi_cluster/copy_into/ddl/orders.sql
new file mode 100644
index 00000000000..71ae0815474
--- /dev/null
+++ b/regression-test/suites/cloud_p0/multi_cluster/copy_into/ddl/orders.sql
@@ -0,0 +1,14 @@
+CREATE TABLE IF NOT EXISTS orders (
+ O_ORDERKEY INTEGER NOT NULL,
+ O_CUSTKEY INTEGER NOT NULL,
+ O_ORDERSTATUS CHAR(1) NOT NULL,
+ O_TOTALPRICE DECIMAL(15,2) NOT NULL,
+ O_ORDERDATE DATE NOT NULL,
+ O_ORDERPRIORITY CHAR(15) NOT NULL,
+ O_CLERK CHAR(15) NOT NULL,
+ O_SHIPPRIORITY INTEGER NOT NULL,
+ O_COMMENT VARCHAR(79) NOT NULL
+)
+UNIQUE KEY(O_ORDERKEY)
+DISTRIBUTED BY HASH(O_ORDERKEY) BUCKETS 1
+
diff --git a/regression-test/suites/cloud_p0/multi_cluster/copy_into/ddl/orders_delete.sql b/regression-test/suites/cloud_p0/multi_cluster/copy_into/ddl/orders_delete.sql
new file mode 100644
index 00000000000..5199d8f296d
--- /dev/null
+++ b/regression-test/suites/cloud_p0/multi_cluster/copy_into/ddl/orders_delete.sql
@@ -0,0 +1 @@
+delete from orders where O_ORDERKEY >= 0;
diff --git a/regression-test/suites/cloud_p0/multi_cluster/copy_into/ddl/part.sql b/regression-test/suites/cloud_p0/multi_cluster/copy_into/ddl/part.sql
new file mode 100644
index 00000000000..b1edbcebd12
--- /dev/null
+++ b/regression-test/suites/cloud_p0/multi_cluster/copy_into/ddl/part.sql
@@ -0,0 +1,14 @@
+CREATE TABLE IF NOT EXISTS part (
+ P_PARTKEY INTEGER NOT NULL,
+ P_NAME VARCHAR(55) NOT NULL,
+ P_MFGR CHAR(25) NOT NULL,
+ P_BRAND CHAR(10) NOT NULL,
+ P_TYPE VARCHAR(25) NOT NULL,
+ P_SIZE INTEGER NOT NULL,
+ P_CONTAINER CHAR(10) NOT NULL,
+ P_RETAILPRICE DECIMAL(15,2) NOT NULL,
+ P_COMMENT VARCHAR(23) NOT NULL
+)
+UNIQUE KEY(P_PARTKEY)
+DISTRIBUTED BY HASH(P_PARTKEY) BUCKETS 1
+
diff --git a/regression-test/suites/cloud_p0/multi_cluster/copy_into/ddl/part_delete.sql b/regression-test/suites/cloud_p0/multi_cluster/copy_into/ddl/part_delete.sql
new file mode 100644
index 00000000000..ee1a2e51045
--- /dev/null
+++ b/regression-test/suites/cloud_p0/multi_cluster/copy_into/ddl/part_delete.sql
@@ -0,0 +1 @@
+delete from part where P_PARTKEY>=0;
diff --git a/regression-test/suites/cloud_p0/multi_cluster/copy_into/ddl/partsupp.sql b/regression-test/suites/cloud_p0/multi_cluster/copy_into/ddl/partsupp.sql
new file mode 100644
index 00000000000..3fa6d79fab2
--- /dev/null
+++ b/regression-test/suites/cloud_p0/multi_cluster/copy_into/ddl/partsupp.sql
@@ -0,0 +1,10 @@
+CREATE TABLE IF NOT EXISTS partsupp (
+ PS_PARTKEY INTEGER NOT NULL,
+ PS_SUPPKEY INTEGER NOT NULL,
+ PS_AVAILQTY INTEGER NOT NULL,
+ PS_SUPPLYCOST DECIMAL(15,2) NOT NULL,
+ PS_COMMENT VARCHAR(199) NOT NULL
+)
+UNIQUE KEY(PS_PARTKEY, PS_SUPPKEY)
+DISTRIBUTED BY HASH(PS_PARTKEY) BUCKETS 1
+
diff --git a/regression-test/suites/cloud_p0/multi_cluster/copy_into/ddl/partsupp_delete.sql b/regression-test/suites/cloud_p0/multi_cluster/copy_into/ddl/partsupp_delete.sql
new file mode 100644
index 00000000000..067cff3a6f0
--- /dev/null
+++ b/regression-test/suites/cloud_p0/multi_cluster/copy_into/ddl/partsupp_delete.sql
@@ -0,0 +1 @@
+delete from partsupp where PS_PARTKEY>=0;
diff --git a/regression-test/suites/cloud_p0/multi_cluster/copy_into/ddl/region.sql b/regression-test/suites/cloud_p0/multi_cluster/copy_into/ddl/region.sql
new file mode 100644
index 00000000000..adfedcdbbd4
--- /dev/null
+++ b/regression-test/suites/cloud_p0/multi_cluster/copy_into/ddl/region.sql
@@ -0,0 +1,8 @@
+CREATE TABLE IF NOT EXISTS region (
+ R_REGIONKEY INTEGER NOT NULL,
+ R_NAME CHAR(25) NOT NULL,
+ R_COMMENT VARCHAR(152)
+)
+UNIQUE KEY(R_REGIONKEY)
+DISTRIBUTED BY HASH(R_REGIONKEY) BUCKETS 1
+
diff --git a/regression-test/suites/cloud_p0/multi_cluster/copy_into/ddl/region_delete.sql b/regression-test/suites/cloud_p0/multi_cluster/copy_into/ddl/region_delete.sql
new file mode 100644
index 00000000000..4945d5105ab
--- /dev/null
+++ b/regression-test/suites/cloud_p0/multi_cluster/copy_into/ddl/region_delete.sql
@@ -0,0 +1 @@
+delete from region where R_REGIONKEY>=0;
diff --git a/regression-test/suites/cloud_p0/multi_cluster/copy_into/ddl/revenue1.sql b/regression-test/suites/cloud_p0/multi_cluster/copy_into/ddl/revenue1.sql
new file mode 100644
index 00000000000..a12a60a2848
--- /dev/null
+++ b/regression-test/suites/cloud_p0/multi_cluster/copy_into/ddl/revenue1.sql
@@ -0,0 +1,11 @@
+CREATE VIEW IF NOT EXISTS revenue1 AS
+ SELECT
+ l_suppkey AS supplier_no,
+ sum(l_extendedprice * (1 - l_discount)) AS total_revenue
+ FROM
+ lineitem
+ WHERE
+ l_shipdate >= DATE '1996-01-01'
+ AND l_shipdate < DATE '1996-01-01' + INTERVAL '3' MONTH
+GROUP BY
+ l_suppkey;
diff --git a/regression-test/suites/cloud_p0/multi_cluster/copy_into/ddl/revenue1_delete.sql b/regression-test/suites/cloud_p0/multi_cluster/copy_into/ddl/revenue1_delete.sql
new file mode 100644
index 00000000000..c2ee91a38a9
--- /dev/null
+++ b/regression-test/suites/cloud_p0/multi_cluster/copy_into/ddl/revenue1_delete.sql
@@ -0,0 +1 @@
+drop view IF EXISTS revenue1;
\ No newline at end of file
diff --git a/regression-test/suites/cloud_p0/multi_cluster/copy_into/ddl/supplier.sql b/regression-test/suites/cloud_p0/multi_cluster/copy_into/ddl/supplier.sql
new file mode 100644
index 00000000000..4eac18d07bc
--- /dev/null
+++ b/regression-test/suites/cloud_p0/multi_cluster/copy_into/ddl/supplier.sql
@@ -0,0 +1,11 @@
+CREATE TABLE IF NOT EXISTS supplier (
+ S_SUPPKEY INTEGER NOT NULL,
+ S_NAME CHAR(25) NOT NULL,
+ S_ADDRESS VARCHAR(40) NOT NULL,
+ S_NATIONKEY INTEGER NOT NULL,
+ S_PHONE CHAR(15) NOT NULL,
+ S_ACCTBAL DECIMAL(15,2) NOT NULL,
+ S_COMMENT VARCHAR(101) NOT NULL
+)
+UNIQUE KEY(S_SUPPKEY)
+DISTRIBUTED BY HASH(S_SUPPKEY) BUCKETS 1
diff --git a/regression-test/suites/cloud_p0/multi_cluster/copy_into/ddl/supplier_delete.sql b/regression-test/suites/cloud_p0/multi_cluster/copy_into/ddl/supplier_delete.sql
new file mode 100644
index 00000000000..d8327a8f856
--- /dev/null
+++ b/regression-test/suites/cloud_p0/multi_cluster/copy_into/ddl/supplier_delete.sql
@@ -0,0 +1 @@
+delete from supplier where S_SUPPKEY>=0;
diff --git a/regression-test/suites/cloud_p0/multi_cluster/copy_into/sync_laod.groovy b/regression-test/suites/cloud_p0/multi_cluster/copy_into/sync_laod.groovy
new file mode 100644
index 00000000000..72c84017185
--- /dev/null
+++ b/regression-test/suites/cloud_p0/multi_cluster/copy_into/sync_laod.groovy
@@ -0,0 +1,846 @@
+// 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 groovy.json.JsonOutput
+
+suite("sync_load") {
+ def externalStageName = "regression_test_tpch"
+ def prefix = "tpch/sf1"
+
+ List<String> ipList = new ArrayList<>()
+ List<String> hbPortList = new ArrayList<>()
+ List<String> httpPortList = new ArrayList<>()
+ List<String> beUniqueIdList = new ArrayList<>()
+ List<String> bePortList = new ArrayList<>()
+
+ String[] bes = context.config.multiClusterBes.split(',');
+ println("the value is " + context.config.multiClusterBes);
+ for(String values : bes) {
+ println("the value is " + values);
+ String[] beInfo = values.split(':');
+ ipList.add(beInfo[0]);
+ hbPortList.add(beInfo[1]);
+ httpPortList.add(beInfo[2]);
+ beUniqueIdList.add(beInfo[3]);
+ }
+
+ println("the ip is " + ipList);
+ println("the heartbeat port is " + hbPortList);
+ println("the http port is " + httpPortList);
+ println("the be unique id is " + beUniqueIdList);
+
+ for (unique_id : beUniqueIdList) {
+ resp = get_cluster.call(unique_id);
+ for (cluster : resp) {
+ if (cluster.type == "COMPUTE") {
+ drop_cluster.call(cluster.cluster_name, cluster.cluster_id);
+ }
+ }
+ }
+ wait_cluster_change()
+
+ List<List<Object>> result = sql "show clusters"
+ assertTrue(result.size() == 0);
+
+ add_cluster.call(beUniqueIdList[0], ipList[0], hbPortList[0],
+ "regression_cluster_name0", "regression_cluster_id0");
+ add_cluster.call(beUniqueIdList[1], ipList[1], hbPortList[1],
+ "regression_cluster_name1", "regression_cluster_id1");
+ wait_cluster_change()
+
+ result = sql "show clusters"
+ assertEquals(result.size(), 2);
+
+ sql "use @regression_cluster_name1"
+ result = sql "show clusters"
+ sql """ set enable_profile = true; """
+
+ def before_cluster0_load_rows = get_be_metric(ipList[0], httpPortList[0], "load_rows");
+ log.info("before_cluster0_load_rows : ${before_cluster0_load_rows}".toString())
+ def before_cluster0_flush = get_be_metric(ipList[0], httpPortList[0], "memtable_flush_total");
+ log.info("before_cluster0_flush : ${before_cluster0_flush}".toString())
+
+ def before_cluster1_load_rows = get_be_metric(ipList[1], httpPortList[1], "load_rows");
+ log.info("before_cluster1_load_rows : ${before_cluster1_load_rows}".toString())
+ def before_cluster1_flush = get_be_metric(ipList[1], httpPortList[1], "memtable_flush_total");
+ log.info("before_cluster1_flush : ${before_cluster1_flush}".toString())
+
+ // tpch_sf1_p1 is writted to test unique key table merge correctly.
+ // It creates unique key table and sets bucket num to 1 in order to make sure that
+ // many rowsets will be created during loading and then the merge process will be triggered.
+
+ def tables = ["customer", "lineitem", "nation", "orders", "part", "partsupp", "region", "supplier"]
+
+ for (String table in tables) {
+ sql """ DROP TABLE IF EXISTS ${table}; """
+ sql new File("""${context.file.parent}/ddl/${table}.sql""").text
+ }
+
+ sql """
+ create stage if not exists ${externalStageName} properties(
+ 'endpoint' = '${getS3Endpoint()}' ,
+ 'region' = '${getS3Region()}' ,
+ 'bucket' = '${getS3BucketName()}' ,
+ 'prefix' = 'regression' ,
+ 'ak' = '${getS3AK()}' ,
+ 'sk' = '${getS3SK()}' ,
+ 'provider' = '${getProvider()}',
+ 'access_type' = 'aksk',
+ 'default.file.column_separator' = "|"
+ );
+ """
+
+ for (String tableName in tables) {
+ result = sql " copy into ${tableName} from @${externalStageName}('${prefix}/${tableName}.csv.split00.gz') properties ('file.compression' = 'gz', 'copy.async' = 'false'); "
+ logger.info("copy result: " + result)
+ assertTrue(result.size() == 1)
+ assertTrue(result[0].size() == 8)
+ assertTrue(result[0][1].equals("FINISHED"))
+
+ result = sql " copy into ${tableName} from @${externalStageName}('${prefix}/${tableName}.csv.split01.gz') properties ('file.compression' = 'gz', 'copy.async' = 'false'); "
+ logger.info("copy result: " + result)
+ assertTrue(result.size() == 1)
+ assertTrue(result[0].size() == 8)
+ assertTrue(result[0][1].equals("FINISHED"))
+ }
+
+ def table = "revenue1"
+ sql new File("""${context.file.parent}/ddl/${table}_delete.sql""").text
+ sql new File("""${context.file.parent}/ddl/${table}.sql""").text
+
+ def after_cluster0_load_rows = get_be_metric(ipList[0], httpPortList[0], "load_rows");
+ log.info("after_cluster0_load_rows : ${after_cluster0_load_rows}".toString())
+ def after_cluster0_flush = get_be_metric(ipList[0], httpPortList[0], "memtable_flush_total");
+ log.info("after_cluster0_flush : ${after_cluster0_flush}".toString())
+
+ def after_cluster1_load_rows = get_be_metric(ipList[1], httpPortList[1], "load_rows");
+ log.info("after_cluster1_load_rows : ${after_cluster1_load_rows}".toString())
+ def after_cluster1_flush = get_be_metric(ipList[1], httpPortList[1], "memtable_flush_total");
+ log.info("after_cluster1_flush : ${after_cluster1_flush}".toString())
+
+ assertTrue(before_cluster0_load_rows == after_cluster0_load_rows)
+ assertTrue(before_cluster0_flush == after_cluster0_flush)
+
+ assertTrue(before_cluster1_load_rows != after_cluster1_load_rows)
+ assertTrue(before_cluster1_flush != after_cluster1_flush)
+
+ sql "use @regression_cluster_name0"
+ result = sql "show clusters"
+
+ def before_cluster0_query_scan_rows = get_be_metric(ipList[0], httpPortList[0], "query_scan_rows");
+ log.info("before_cluster0_query_scan_rows : ${before_cluster0_query_scan_rows}".toString())
+ def before_cluster1_query_scan_rows = get_be_metric(ipList[1], httpPortList[1], "query_scan_rows");
+ log.info("before_cluster1_query_scan_rows : ${before_cluster1_query_scan_rows}".toString())
+
+ sql """
+ SELECT
+ s_acctbal,
+ s_name,
+ n_name,
+ p_partkey,
+ p_mfgr,
+ s_address,
+ s_phone,
+ s_comment
+ FROM
+ part,
+ supplier,
+ partsupp,
+ nation,
+ region
+ WHERE
+ p_partkey = ps_partkey
+ AND s_suppkey = ps_suppkey
+ AND p_size = 15
+ AND p_type LIKE '%BRASS'
+ AND s_nationkey = n_nationkey
+ AND n_regionkey = r_regionkey
+ AND r_name = 'EUROPE'
+ AND ps_supplycost = (
+ SELECT min(ps_supplycost)
+ FROM
+ partsupp, supplier,
+ nation, region
+ WHERE
+ p_partkey = ps_partkey
+ AND s_suppkey = ps_suppkey
+ AND s_nationkey = n_nationkey
+ AND n_regionkey = r_regionkey
+ AND r_name = 'EUROPE'
+ )
+ ORDER BY
+ s_acctbal DESC,
+ n_name,
+ s_name,
+ p_partkey
+ LIMIT 100
+ """
+
+ def after_cluster0_query_scan_rows = get_be_metric(ipList[0], httpPortList[0], "query_scan_rows");
+ log.info("after_cluster0_query_scan_rows : ${after_cluster0_query_scan_rows}".toString())
+ def after_cluster1_query_scan_rows = get_be_metric(ipList[1], httpPortList[1], "query_scan_rows");
+ log.info("after_cluster1_query_scan_rows : ${after_cluster1_query_scan_rows}".toString())
+
+ assertTrue(before_cluster0_query_scan_rows != after_cluster0_query_scan_rows)
+ assertTrue(before_cluster1_query_scan_rows == after_cluster1_query_scan_rows)
+
+ //def sqls = ["q01", "q02"]
+
+ //for (String sql in sqls) {
+ // //sql """ DROP TABLE IF EXISTS ${table}; """
+ // sql new File("""${context.file.parent}/ddl/${sql}.sql""").text
+ //}
+
+ // fill bePortList
+ for (int i = 0; i < ipList.size(); ++i) {
+ result = sql """show backends"""
+ for (row : result) {
+ println row
+ println row[2]
+ if (ipList[i] == row[1] && hbPortList[i] == row[2]) {
+ bePortList.add(row[3]);
+ }
+ }
+ }
+ assertEquals(bePortList.size(), 2);
+
+ // q01
+ sql """
+ SELECT
+ l_returnflag,
+ l_linestatus,
+ sum(l_quantity) AS sum_qty,
+ sum(l_extendedprice) AS sum_base_price,
+ sum(l_extendedprice * (1 - l_discount)) AS sum_disc_price,
+ sum(l_extendedprice * (1 - l_discount) * (1 + l_tax)) AS sum_charge,
+ avg(l_quantity) AS avg_qty,
+ avg(l_extendedprice) AS avg_price,
+ avg(l_discount) AS avg_disc,
+ count(*) AS count_order
+ FROM
+ lineitem
+ WHERE
+ l_shipdate <= DATE '1998-12-01' - INTERVAL '90' DAY
+ GROUP BY
+ l_returnflag,
+ l_linestatus
+ ORDER BY
+ l_returnflag,
+ l_linestatus
+ """
+
+ def set = [ipList[0] + ":" +bePortList[0]] as Set
+ checkProfile.call(set)
+
+ // q02
+ sql """
+ SELECT
+ s_acctbal,
+ s_name,
+ n_name,
+ p_partkey,
+ p_mfgr,
+ s_address,
+ s_phone,
+ s_comment
+ FROM
+ part,
+ supplier,
+ partsupp,
+ nation,
+ region
+ WHERE
+ p_partkey = ps_partkey
+ AND s_suppkey = ps_suppkey
+ AND p_size = 15
+ AND p_type LIKE '%BRASS'
+ AND s_nationkey = n_nationkey
+ AND n_regionkey = r_regionkey
+ AND r_name = 'EUROPE'
+ AND ps_supplycost = (
+ SELECT min(ps_supplycost)
+ FROM
+ partsupp, supplier,
+ nation, region
+ WHERE
+ p_partkey = ps_partkey
+ AND s_suppkey = ps_suppkey
+ AND s_nationkey = n_nationkey
+ AND n_regionkey = r_regionkey
+ AND r_name = 'EUROPE'
+ )
+ ORDER BY
+ s_acctbal DESC,
+ n_name,
+ s_name,
+ p_partkey
+ LIMIT 100
+ """
+ checkProfile.call(set)
+
+ // q03
+ sql """
+ SELECT
+ l_orderkey,
+ sum(l_extendedprice * (1 - l_discount)) AS revenue,
+ o_orderdate,
+ o_shippriority
+ FROM
+ customer,
+ orders,
+ lineitem
+ WHERE
+ c_mktsegment = 'BUILDING'
+ AND c_custkey = o_custkey
+ AND l_orderkey = o_orderkey
+ AND o_orderdate < DATE '1995-03-15'
+ AND l_shipdate > DATE '1995-03-15'
+ GROUP BY
+ l_orderkey,
+ o_orderdate,
+ o_shippriority
+ ORDER BY
+ revenue DESC,
+ o_orderdate
+ LIMIT 10
+ """
+ checkProfile.call(set)
+
+ // q04
+ sql """
+ SELECT
+ o_orderpriority,
+ count(*) AS order_count
+ FROM orders
+ WHERE
+ o_orderdate >= DATE '1993-07-01'
+ AND o_orderdate < DATE '1993-07-01' + INTERVAL '3' MONTH
+ AND EXISTS (
+ SELECT *
+ FROM lineitem
+ WHERE
+ l_orderkey = o_orderkey
+ AND l_commitdate < l_receiptdate
+ )
+ GROUP BY
+ o_orderpriority
+ ORDER BY
+ o_orderpriority
+ """
+ checkProfile.call(set)
+
+ // q05
+ sql """
+ SELECT
+ n_name,
+ sum(l_extendedprice * (1 - l_discount)) AS revenue
+ FROM
+ customer,
+ orders,
+ lineitem,
+ supplier,
+ nation,
+ region
+ WHERE
+ c_custkey = o_custkey
+ AND l_orderkey = o_orderkey
+ AND l_suppkey = s_suppkey
+ AND c_nationkey = s_nationkey
+ AND s_nationkey = n_nationkey
+ AND n_regionkey = r_regionkey
+ AND r_name = 'ASIA'
+ AND o_orderdate >= DATE '1994-01-01'
+ AND o_orderdate < DATE '1994-01-01' + INTERVAL '1' YEAR
+ GROUP BY
+ n_name
+ ORDER BY
+ revenue DESC
+ """
+ checkProfile.call(set)
+
+ // q06
+ sql """
+ SELECT sum(l_extendedprice * l_discount) AS revenue
+ FROM
+ lineitem
+ WHERE
+ l_shipdate >= DATE '1994-01-01'
+ AND l_shipdate < DATE '1994-01-01' + INTERVAL '1' YEAR
+ AND l_discount BETWEEN 0.06 - 0.01 AND .06 + 0.01
+ AND l_quantity < 24
+ """
+ checkProfile.call(set)
+
+ // q07
+ sql """
+ SELECT
+ supp_nation,
+ cust_nation,
+ l_year,
+ sum(volume) AS revenue
+ FROM (
+ SELECT
+ n1.n_name AS supp_nation,
+ n2.n_name AS cust_nation,
+ extract(YEAR FROM l_shipdate) AS l_year,
+ l_extendedprice * (1 - l_discount) AS volume
+ FROM
+ supplier,
+ lineitem,
+ orders,
+ customer,
+ nation n1,
+ nation n2
+ WHERE
+ s_suppkey = l_suppkey
+ AND o_orderkey = l_orderkey
+ AND c_custkey = o_custkey
+ AND s_nationkey = n1.n_nationkey
+ AND c_nationkey = n2.n_nationkey
+ AND (
+ (n1.n_name = 'FRANCE' AND n2.n_name = 'GERMANY')
+ OR (n1.n_name = 'GERMANY' AND n2.n_name = 'FRANCE')
+ )
+ AND l_shipdate BETWEEN DATE '1995-01-01' AND DATE '1996-12-31'
+ ) AS shipping
+ GROUP BY
+ supp_nation,
+ cust_nation,
+ l_year
+ ORDER BY
+ supp_nation,
+ cust_nation,
+ l_year
+ """
+ checkProfile.call(set)
+
+ sql "use @regression_cluster_name1"
+ set = [ipList[1] + ":" +bePortList[1]] as Set
+
+ // q08
+ sql """
+ SELECT
+ o_year,
+ sum(CASE
+ WHEN nation = 'BRAZIL'
+ THEN volume
+ ELSE 0
+ END) / sum(volume) AS mkt_share
+ FROM (
+ SELECT
+ extract(YEAR FROM o_orderdate) AS o_year,
+ l_extendedprice * (1 - l_discount) AS volume,
+ n2.n_name AS nation
+ FROM
+ part,
+ supplier,
+ lineitem,
+ orders,
+ customer,
+ nation n1,
+ nation n2,
+ region
+ WHERE
+ p_partkey = l_partkey
+ AND s_suppkey = l_suppkey
+ AND l_orderkey = o_orderkey
+ AND o_custkey = c_custkey
+ AND c_nationkey = n1.n_nationkey
+ AND n1.n_regionkey = r_regionkey
+ AND r_name = 'AMERICA'
+ AND s_nationkey = n2.n_nationkey
+ AND o_orderdate BETWEEN DATE '1995-01-01' AND DATE '1996-12-31'
+ AND p_type = 'ECONOMY ANODIZED STEEL'
+ ) AS all_nations
+ GROUP BY
+ o_year
+ ORDER BY
+ o_year
+ """
+ checkProfile.call(set)
+
+ // q09
+ sql """
+ SELECT
+ nation,
+ o_year,
+ sum(amount) AS sum_profit
+ FROM (
+ SELECT
+ n_name AS nation,
+ extract(YEAR FROM o_orderdate) AS o_year,
+ l_extendedprice * (1 - l_discount) - ps_supplycost * l_quantity AS amount
+ FROM
+ part,
+ supplier,
+ lineitem,
+ partsupp,
+ orders,
+ nation
+ WHERE
+ s_suppkey = l_suppkey
+ AND ps_suppkey = l_suppkey
+ AND ps_partkey = l_partkey
+ AND p_partkey = l_partkey
+ AND o_orderkey = l_orderkey
+ AND s_nationkey = n_nationkey
+ AND p_name LIKE '%green%'
+ ) AS profit
+ GROUP BY
+ nation,
+ o_year
+ ORDER BY
+ nation,
+ o_year DESC
+ """
+ checkProfile.call(set)
+
+ // q10
+ sql """
+ SELECT
+ c_custkey,
+ c_name,
+ sum(l_extendedprice * (1 - l_discount)) AS revenue,
+ c_acctbal,
+ n_name,
+ c_address,
+ c_phone,
+ c_comment
+ FROM
+ customer,
+ orders,
+ lineitem,
+ nation
+ WHERE
+ c_custkey = o_custkey
+ AND l_orderkey = o_orderkey
+ AND o_orderdate >= DATE '1993-10-01'
+ AND o_orderdate < DATE '1993-10-01' + INTERVAL '3' MONTH
+ AND l_returnflag = 'R'
+ AND c_nationkey = n_nationkey
+ GROUP BY
+ c_custkey,
+ c_name,
+ c_acctbal,
+ c_phone,
+ n_name,
+ c_address,
+ c_comment
+ ORDER BY
+ revenue DESC
+ LIMIT 20
+ """
+ checkProfile.call(set)
+
+ // q11
+ sql """
+ SELECT
+ ps_partkey,
+ sum(ps_supplycost * ps_availqty) AS value
+ FROM
+ partsupp,
+ supplier,
+ nation
+ WHERE
+ ps_suppkey = s_suppkey
+ AND s_nationkey = n_nationkey
+ AND n_name = 'GERMANY'
+ GROUP BY
+ ps_partkey
+ HAVING
+ sum(ps_supplycost * ps_availqty) > (
+ SELECT sum(ps_supplycost * ps_availqty) * 0.0001
+ FROM
+ partsupp,
+ supplier,
+ nation
+ WHERE
+ ps_suppkey = s_suppkey
+ AND s_nationkey = n_nationkey
+ AND n_name = 'GERMANY'
+ )
+ ORDER BY
+ value DESC
+ """
+ checkProfile.call(set)
+
+ // q12
+ sql """
+ SELECT
+ l_shipmode,
+ sum(CASE
+ WHEN o_orderpriority = '1-URGENT'
+ OR o_orderpriority = '2-HIGH'
+ THEN 1
+ ELSE 0
+ END) AS high_line_count,
+ sum(CASE
+ WHEN o_orderpriority <> '1-URGENT'
+ AND o_orderpriority <> '2-HIGH'
+ THEN 1
+ ELSE 0
+ END) AS low_line_count
+ FROM
+ orders,
+ lineitem
+ WHERE
+ o_orderkey = l_orderkey
+ AND l_shipmode IN ('MAIL', 'SHIP')
+ AND l_commitdate < l_receiptdate
+ AND l_shipdate < l_commitdate
+ AND l_receiptdate >= DATE '1994-01-01'
+ AND l_receiptdate < DATE '1994-01-01' + INTERVAL '1' YEAR
+ GROUP BY
+ l_shipmode
+ ORDER BY
+ l_shipmode
+ """
+ checkProfile.call(set)
+
+ // q13
+ sql """
+ SELECT
+ c_count,
+ count(*) AS custdist
+ FROM (
+ SELECT
+ c_custkey,
+ count(o_orderkey) AS c_count
+ FROM
+ customer
+ LEFT OUTER JOIN orders ON
+ c_custkey = o_custkey
+ AND o_comment NOT LIKE '%special%requests%'
+ GROUP BY
+ c_custkey
+ ) AS c_orders
+ GROUP BY
+ c_count
+ ORDER BY
+ custdist DESC,
+ c_count DESC
+ """
+ checkProfile.call(set)
+
+ // q14
+ sql """
+ SELECT 100.00 * sum(CASE
+ WHEN p_type LIKE 'PROMO%'
+ THEN l_extendedprice * (1 - l_discount)
+ ELSE 0
+ END) / sum(l_extendedprice * (1 - l_discount)) AS promo_revenue
+ FROM
+ lineitem,
+ part
+ WHERE
+ l_partkey = p_partkey
+ AND l_shipdate >= DATE '1995-09-01'
+ AND l_shipdate < DATE '1995-09-01' + INTERVAL '1' MONTH
+ """
+ checkProfile.call(set)
+
+ // q15
+ sql """
+ SELECT
+ s_suppkey,
+ s_name,
+ s_address,
+ s_phone,
+ total_revenue
+ FROM
+ supplier,
+ revenue1
+ WHERE
+ s_suppkey = supplier_no
+ AND total_revenue = (
+ SELECT max(total_revenue)
+ FROM
+ revenue1
+ )
+ ORDER BY
+ s_suppkey;
+ """
+ checkProfile.call(set)
+
+ // q16
+ sql """
+ SELECT
+ p_brand,
+ p_type,
+ p_size,
+ count(DISTINCT ps_suppkey) AS supplier_cnt
+ FROM
+ partsupp,
+ part
+ WHERE
+ p_partkey = ps_partkey
+ AND p_brand <> 'Brand#45'
+ AND p_type NOT LIKE 'MEDIUM POLISHED%'
+ AND p_size IN (49, 14, 23, 45, 19, 3, 36, 9)
+ AND ps_suppkey NOT IN (
+ SELECT s_suppkey
+ FROM
+ supplier
+ WHERE
+ s_comment LIKE '%Customer%Complaints%'
+ )
+ GROUP BY
+ p_brand,
+ p_type,
+ p_size
+ ORDER BY
+ supplier_cnt DESC,
+ p_brand,
+ p_type,
+ p_size
+ """
+ checkProfile.call(set)
+
+ // q17
+ sql """
+ SELECT sum(l_extendedprice) / 7.0 AS avg_yearly
+ FROM
+ lineitem,
+ part
+ WHERE
+ p_partkey = l_partkey
+ AND p_brand = 'Brand#23'
+ AND p_container = 'MED BOX'
+ AND l_quantity < (
+ SELECT 0.2 * avg(l_quantity)
+ FROM
+ lineitem
+ WHERE
+ l_partkey = p_partkey
+ )
+ """
+ checkProfile.call(set)
+
+ // q18
+ sql """
+ SELECT
+ c_name,
+ c_custkey,
+ o_orderkey,
+ o_orderdate,
+ o_totalprice,
+ sum(l_quantity)
+ FROM
+ customer,
+ orders,
+ lineitem
+ WHERE
+ o_orderkey IN (
+ SELECT l_orderkey
+ FROM
+ lineitem
+ GROUP BY
+ l_orderkey
+ HAVING
+ sum(l_quantity) > 300
+ )
+ AND c_custkey = o_custkey
+ AND o_orderkey = l_orderkey
+ GROUP BY
+ c_name,
+ c_custkey,
+ o_orderkey,
+ o_orderdate,
+ o_totalprice
+ ORDER BY
+ o_totalprice DESC,
+ o_orderdate
+ LIMIT 100
+ """
+ checkProfile.call(set)
+
+ // q19
+ sql """
+ SELECT sum(l_extendedprice * (1 - l_discount)) AS revenue
+ FROM
+ lineitem,
+ part
+ WHERE
+ (
+ p_partkey = l_partkey
+ AND p_brand = 'Brand#12'
+ AND p_container IN ('SM CASE', 'SM BOX', 'SM PACK', 'SM PKG')
+ AND l_quantity >= 1 AND l_quantity <= 1 + 10
+ AND p_size BETWEEN 1 AND 5
+ AND l_shipmode IN ('AIR', 'AIR REG')
+ AND l_shipinstruct = 'DELIVER IN PERSON'
+ )
+ OR
+ (
+ p_partkey = l_partkey
+ AND p_brand = 'Brand#23'
+ AND p_container IN ('MED BAG', 'MED BOX', 'MED PKG', 'MED PACK')
+ AND l_quantity >= 10 AND l_quantity <= 10 + 10
+ AND p_size BETWEEN 1 AND 10
+ AND l_shipmode IN ('AIR', 'AIR REG')
+ AND l_shipinstruct = 'DELIVER IN PERSON'
+ )
+ OR
+ (
+ p_partkey = l_partkey
+ AND p_brand = 'Brand#34'
+ AND p_container IN ('LG CASE', 'LG BOX', 'LG PACK', 'LG PKG')
+ AND l_quantity >= 20 AND l_quantity <= 20 + 10
+ AND p_size BETWEEN 1 AND 15
+ AND l_shipmode IN ('AIR', 'AIR REG')
+ AND l_shipinstruct = 'DELIVER IN PERSON'
+ )
+ """
+ checkProfile.call(set)
+
+ // q20
+ sql """
+ SELECT
+ s_name,
+ s_address
+ FROM
+ supplier, nation
+ WHERE
+ s_suppkey IN (
+ SELECT ps_suppkey
+ FROM
+ partsupp
+ WHERE
+ ps_partkey IN (
+ SELECT p_partkey
+ FROM
+ part
+ WHERE
+ p_name LIKE 'forest%'
+ )
+ AND ps_availqty > (
+ SELECT 0.5 * sum(l_quantity)
+ FROM
+ lineitem
+ WHERE
+ l_partkey = ps_partkey
+ AND l_suppkey = ps_suppkey
+ AND l_shipdate >= date('1994-01-01')
+ AND l_shipdate < date('1994-01-01') + interval '1' YEAR
+ )
+ )
+ AND s_nationkey = n_nationkey
+ AND n_name = 'CANADA'
+ ORDER BY s_name
+ """
+ checkProfile.call(set)
+}
diff --git a/regression-test/suites/cloud_p0/multi_cluster/default_cluster.groovy b/regression-test/suites/cloud_p0/multi_cluster/default_cluster.groovy
new file mode 100644
index 00000000000..7580ca13a8b
--- /dev/null
+++ b/regression-test/suites/cloud_p0/multi_cluster/default_cluster.groovy
@@ -0,0 +1,172 @@
+// 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 groovy.json.JsonOutput
+
+suite("default_cluster") {
+ def token = context.config.metaServiceToken
+ def instance_id = context.config.multiClusterInstance
+
+ List<String> ipList = new ArrayList<>()
+ List<String> hbPortList = new ArrayList<>()
+ List<String> httpPortList = new ArrayList<>()
+ List<String> beUniqueIdList = new ArrayList<>()
+
+ String[] bes = context.config.multiClusterBes.split(',');
+ println("the value is " + context.config.multiClusterBes);
+ for(String values : bes) {
+ println("the value is " + values);
+ String[] beInfo = values.split(':');
+ ipList.add(beInfo[0]);
+ hbPortList.add(beInfo[1]);
+ httpPortList.add(beInfo[2]);
+ beUniqueIdList.add(beInfo[3]);
+ }
+
+ println("the ip is " + ipList);
+ println("the heartbeat port is " + hbPortList);
+ println("the http port is " + httpPortList);
+ println("the be unique id is " + beUniqueIdList);
+
+ for (unique_id : beUniqueIdList) {
+ resp = get_cluster.call(unique_id);
+ for (cluster : resp) {
+ if (cluster.type == "COMPUTE") {
+ drop_cluster.call(cluster.cluster_name, cluster.cluster_id);
+ }
+ }
+ }
+ wait_cluster_change()
+
+ List<List<Object>> result = sql "show clusters"
+ assertTrue(result.size() == 0);
+
+ add_cluster.call(beUniqueIdList[0], ipList[0], hbPortList[0],
+ "regression_cluster_name2", "regression_cluster_id2");
+ add_cluster.call(beUniqueIdList[1], ipList[1], hbPortList[1],
+ "regression_cluster_name3", "regression_cluster_id3");
+ wait_cluster_change()
+
+ result = sql "show clusters"
+ assertTrue(result.size() == 2);
+ for (row : result) {
+ log.info("show cluster row: ${row}".toString())
+ }
+
+ sql "SET PROPERTY 'default_cloud_cluster' = 'regression_cluster_name3'"
+
+ context.reconnectFe()
+ def before_cluster0_flush = get_be_metric(ipList[0], httpPortList[0], "memtable_flush_total");
+ log.info("before_cluster0_flush : ${before_cluster0_flush}".toString())
+
+ def before_cluster1_flush = get_be_metric(ipList[1], httpPortList[1], "memtable_flush_total");
+ log.info("before_cluster1_flush : ${before_cluster1_flush}".toString())
+
+
+ sql """
+ CREATE TABLE if not exists test_table (
+ class INT,
+ id INT,
+ score INT SUM
+ )
+ AGGREGATE KEY(class, id)
+ DISTRIBUTED BY HASH(class) BUCKETS 16
+ """
+
+ sql """
+ insert into test_table values (1, 2, 3);
+ """
+
+ def after_cluster0_flush = get_be_metric(ipList[0], httpPortList[0], "memtable_flush_total");
+ log.info("after_cluster0_flush : ${after_cluster0_flush}".toString())
+
+ def after_cluster1_flush = get_be_metric(ipList[1], httpPortList[1], "memtable_flush_total");
+ log.info("after_cluster1_flush : ${after_cluster1_flush}".toString())
+
+ assertTrue(before_cluster0_flush == after_cluster0_flush)
+ assertTrue(before_cluster1_flush < after_cluster1_flush)
+
+ result = sql "show clusters"
+ for (row : result) {
+ println row
+ if(row[0] == "regression_cluster_name2") {
+ assertTrue(row[1].toString().toLowerCase() == "false")
+ }
+
+ if(row[0] == "regression_cluster_name3") {
+ assertTrue(row[1].toString().toLowerCase() == "true")
+ }
+ }
+
+ sql "use @regression_cluster_name2"
+ result = sql "show clusters"
+ for (row : result) {
+ println row
+ if(row[0] == "regression_cluster_name2") {
+ assertTrue(row[1].toString().toLowerCase() == "true")
+ }
+ if(row[0] == "regression_cluster_name3") {
+ assertTrue(row[1].toString().toLowerCase() == "false")
+ }
+ }
+
+ sql "SET PROPERTY 'default_cloud_cluster' = 'regression_cluster_name2'"
+ context.reconnectFe()
+ result = sql "show clusters"
+ for (row : result) {
+ println row
+ if(row[0] == "regression_cluster_name2") {
+ assertTrue(row[1].toString().toLowerCase() == "true")
+ }
+ if(row[0] == "regression_cluster_name3") {
+ assertTrue(row[1].toString().toLowerCase() == "false")
+ }
+ }
+
+ sql "use @regression_cluster_name3"
+ result = sql "show clusters"
+ for (row : result) {
+ println row
+ if(row[0] == "regression_cluster_name2") {
+ assertTrue(row[1].toString().toLowerCase() == "false")
+ }
+ if(row[0] == "regression_cluster_name3") {
+ assertTrue(row[1].toString().toLowerCase() == "true")
+ }
+ }
+
+ sql "SET PROPERTY 'default_cloud_cluster' = 'regression_cluster_name3'"
+ test {
+ sql "SET PROPERTY 'default_cloud_cluster' = 'xxxxxx'"
+ // check exception message contains
+ exception "errCode = 2,"
+ }
+
+ sql "SET PROPERTY 'default_cloud_cluster' = ''"
+
+ result = sql "show property"
+ for (row : result) {
+ println row
+ if(row[0] == "default_cloud_cluster") {
+ assertTrue(row[1].toString().isEmpty())
+ }
+ }
+
+ sql """
+ drop table if exists test_table
+ """
+}
diff --git a/regression-test/suites/cloud_p0/multi_cluster/multi_cluster_s3_load/ddl/customer.sql b/regression-test/suites/cloud_p0/multi_cluster/multi_cluster_s3_load/ddl/customer.sql
new file mode 100644
index 00000000000..eb33fd55dfb
--- /dev/null
+++ b/regression-test/suites/cloud_p0/multi_cluster/multi_cluster_s3_load/ddl/customer.sql
@@ -0,0 +1,13 @@
+CREATE TABLE IF NOT EXISTS customer (
+ C_CUSTKEY INTEGER NOT NULL,
+ C_NAME VARCHAR(25) NOT NULL,
+ C_ADDRESS VARCHAR(40) NOT NULL,
+ C_NATIONKEY INTEGER NOT NULL,
+ C_PHONE CHAR(15) NOT NULL,
+ C_ACCTBAL DECIMAL(15,2) NOT NULL,
+ C_MKTSEGMENT CHAR(10) NOT NULL,
+ C_COMMENT VARCHAR(117) NOT NULL
+)
+DUPLICATE KEY(C_CUSTKEY, C_NAME)
+DISTRIBUTED BY HASH(C_CUSTKEY) BUCKETS 32
+
diff --git a/regression-test/suites/cloud_p0/multi_cluster/multi_cluster_s3_load/ddl/customer_delete.sql b/regression-test/suites/cloud_p0/multi_cluster/multi_cluster_s3_load/ddl/customer_delete.sql
new file mode 100644
index 00000000000..a214fe920c9
--- /dev/null
+++ b/regression-test/suites/cloud_p0/multi_cluster/multi_cluster_s3_load/ddl/customer_delete.sql
@@ -0,0 +1 @@
+DELETE FROM customer where C_CUSTKEY >= 0;
diff --git a/regression-test/suites/cloud_p0/multi_cluster/multi_cluster_s3_load/ddl/customer_load.sql b/regression-test/suites/cloud_p0/multi_cluster/multi_cluster_s3_load/ddl/customer_load.sql
new file mode 100644
index 00000000000..4118e750b1f
--- /dev/null
+++ b/regression-test/suites/cloud_p0/multi_cluster/multi_cluster_s3_load/ddl/customer_load.sql
@@ -0,0 +1,6 @@
+LOAD LABEL ${loadLabel} (
+ DATA INFILE("s3://${s3BucketName}/regression/tpch/sf100/customer.tbl")
+ INTO TABLE customer
+ COLUMNS TERMINATED BY "|"
+ (c_custkey, c_name, c_address, c_nationkey, c_phone, c_acctbal, c_mktsegment, c_comment, temp)
+)
diff --git a/regression-test/suites/cloud_p0/multi_cluster/multi_cluster_s3_load/ddl/lineitem.sql b/regression-test/suites/cloud_p0/multi_cluster/multi_cluster_s3_load/ddl/lineitem.sql
new file mode 100644
index 00000000000..5d0cad877c5
--- /dev/null
+++ b/regression-test/suites/cloud_p0/multi_cluster/multi_cluster_s3_load/ddl/lineitem.sql
@@ -0,0 +1,21 @@
+CREATE TABLE IF NOT EXISTS lineitem (
+ L_ORDERKEY INTEGER NOT NULL,
+ L_PARTKEY INTEGER NOT NULL,
+ L_SUPPKEY INTEGER NOT NULL,
+ L_LINENUMBER INTEGER NOT NULL,
+ L_QUANTITY DECIMAL(15,2) NOT NULL,
+ L_EXTENDEDPRICE DECIMAL(15,2) NOT NULL,
+ L_DISCOUNT DECIMAL(15,2) NOT NULL,
+ L_TAX DECIMAL(15,2) NOT NULL,
+ L_RETURNFLAG CHAR(1) NOT NULL,
+ L_LINESTATUS CHAR(1) NOT NULL,
+ L_SHIPDATE DATE NOT NULL,
+ L_COMMITDATE DATE NOT NULL,
+ L_RECEIPTDATE DATE NOT NULL,
+ L_SHIPINSTRUCT CHAR(25) NOT NULL,
+ L_SHIPMODE CHAR(10) NOT NULL,
+ L_COMMENT VARCHAR(44) NOT NULL
+)
+DUPLICATE KEY(L_ORDERKEY, L_PARTKEY, L_SUPPKEY, L_LINENUMBER)
+DISTRIBUTED BY HASH(L_ORDERKEY) BUCKETS 32
+
diff --git a/regression-test/suites/cloud_p0/multi_cluster/multi_cluster_s3_load/ddl/lineitem_delete.sql b/regression-test/suites/cloud_p0/multi_cluster/multi_cluster_s3_load/ddl/lineitem_delete.sql
new file mode 100644
index 00000000000..6dbbe49ed0b
--- /dev/null
+++ b/regression-test/suites/cloud_p0/multi_cluster/multi_cluster_s3_load/ddl/lineitem_delete.sql
@@ -0,0 +1 @@
+DELETE from lineitem where L_ORDERKEY >= 0;
diff --git a/regression-test/suites/cloud_p0/multi_cluster/multi_cluster_s3_load/ddl/lineitem_load.sql b/regression-test/suites/cloud_p0/multi_cluster/multi_cluster_s3_load/ddl/lineitem_load.sql
new file mode 100644
index 00000000000..11ccc204fe5
--- /dev/null
+++ b/regression-test/suites/cloud_p0/multi_cluster/multi_cluster_s3_load/ddl/lineitem_load.sql
@@ -0,0 +1,6 @@
+LOAD LABEL ${loadLabel} (
+ DATA INFILE("s3://${s3BucketName}/regression/tpch/sf100/lineitem.tbl.*")
+ INTO TABLE lineitem
+ COLUMNS TERMINATED BY "|"
+ (l_orderkey, l_partkey, l_suppkey, l_linenumber, l_quantity, l_extendedprice, l_discount, l_tax, l_returnflag, l_linestatus, l_shipdate, l_commitdate, l_receiptdate, l_shipinstruct, l_shipmode, l_comment, temp)
+)
diff --git a/regression-test/suites/cloud_p0/multi_cluster/multi_cluster_s3_load/ddl/nation.sql b/regression-test/suites/cloud_p0/multi_cluster/multi_cluster_s3_load/ddl/nation.sql
new file mode 100644
index 00000000000..269e4064dc8
--- /dev/null
+++ b/regression-test/suites/cloud_p0/multi_cluster/multi_cluster_s3_load/ddl/nation.sql
@@ -0,0 +1,8 @@
+CREATE TABLE IF NOT EXISTS nation (
+ N_NATIONKEY INTEGER NOT NULL,
+ N_NAME CHAR(25) NOT NULL,
+ N_REGIONKEY INTEGER NOT NULL,
+ N_COMMENT VARCHAR(152)
+)
+DUPLICATE KEY(N_NATIONKEY, N_NAME)
+DISTRIBUTED BY HASH(N_NATIONKEY) BUCKETS 1
diff --git a/regression-test/suites/cloud_p0/multi_cluster/multi_cluster_s3_load/ddl/nation_delete.sql b/regression-test/suites/cloud_p0/multi_cluster/multi_cluster_s3_load/ddl/nation_delete.sql
new file mode 100644
index 00000000000..9fb7c085a6f
--- /dev/null
+++ b/regression-test/suites/cloud_p0/multi_cluster/multi_cluster_s3_load/ddl/nation_delete.sql
@@ -0,0 +1 @@
+DELETE from nation where N_NATIONKEY >= 0;
diff --git a/regression-test/suites/cloud_p0/multi_cluster/multi_cluster_s3_load/ddl/nation_load.sql b/regression-test/suites/cloud_p0/multi_cluster/multi_cluster_s3_load/ddl/nation_load.sql
new file mode 100644
index 00000000000..6e6f1363341
--- /dev/null
+++ b/regression-test/suites/cloud_p0/multi_cluster/multi_cluster_s3_load/ddl/nation_load.sql
@@ -0,0 +1,6 @@
+LOAD LABEL ${loadLabel} (
+ DATA INFILE("s3://${s3BucketName}/regression/tpch/sf100/nation.tbl")
+ INTO TABLE nation
+ COLUMNS TERMINATED BY "|"
+ (n_nationkey, n_name, n_regionkey, n_comment, temp)
+)
diff --git a/regression-test/suites/cloud_p0/multi_cluster/multi_cluster_s3_load/ddl/orders.sql b/regression-test/suites/cloud_p0/multi_cluster/multi_cluster_s3_load/ddl/orders.sql
new file mode 100644
index 00000000000..1bfa55a9bc4
--- /dev/null
+++ b/regression-test/suites/cloud_p0/multi_cluster/multi_cluster_s3_load/ddl/orders.sql
@@ -0,0 +1,15 @@
+CREATE TABLE IF NOT EXISTS orders (
+ O_ORDERKEY INTEGER NOT NULL,
+ O_CUSTKEY INTEGER NOT NULL,
+ O_ORDERSTATUS CHAR(1) NOT NULL,
+ O_TOTALPRICE DECIMAL(15,2) NOT NULL,
+ O_ORDERDATE DATE NOT NULL,
+ O_ORDERPRIORITY CHAR(15) NOT NULL,
+ O_CLERK CHAR(15) NOT NULL,
+ O_SHIPPRIORITY INTEGER NOT NULL,
+ O_COMMENT VARCHAR(79) NOT NULL
+)
+DUPLICATE KEY(O_ORDERKEY, O_CUSTKEY)
+DISTRIBUTED BY HASH(O_ORDERKEY) BUCKETS 32
+
+
diff --git a/regression-test/suites/cloud_p0/multi_cluster/multi_cluster_s3_load/ddl/orders_delete.sql b/regression-test/suites/cloud_p0/multi_cluster/multi_cluster_s3_load/ddl/orders_delete.sql
new file mode 100644
index 00000000000..5199d8f296d
--- /dev/null
+++ b/regression-test/suites/cloud_p0/multi_cluster/multi_cluster_s3_load/ddl/orders_delete.sql
@@ -0,0 +1 @@
+delete from orders where O_ORDERKEY >= 0;
diff --git a/regression-test/suites/cloud_p0/multi_cluster/multi_cluster_s3_load/ddl/orders_load.sql b/regression-test/suites/cloud_p0/multi_cluster/multi_cluster_s3_load/ddl/orders_load.sql
new file mode 100644
index 00000000000..695338016dc
--- /dev/null
+++ b/regression-test/suites/cloud_p0/multi_cluster/multi_cluster_s3_load/ddl/orders_load.sql
@@ -0,0 +1,6 @@
+LOAD LABEL ${loadLabel} (
+ DATA INFILE("s3://${s3BucketName}/regression/tpch/sf100/orders.tbl*")
+ INTO TABLE orders
+ COLUMNS TERMINATED BY "|"
+ (o_orderkey, o_custkey, o_orderstatus, o_totalprice, o_orderdate, o_orderpriority, o_clerk, o_shippriority, o_comment, temp)
+)
diff --git a/regression-test/suites/cloud_p0/multi_cluster/multi_cluster_s3_load/ddl/part.sql b/regression-test/suites/cloud_p0/multi_cluster/multi_cluster_s3_load/ddl/part.sql
new file mode 100644
index 00000000000..7bbe68b86d0
--- /dev/null
+++ b/regression-test/suites/cloud_p0/multi_cluster/multi_cluster_s3_load/ddl/part.sql
@@ -0,0 +1,15 @@
+CREATE TABLE IF NOT EXISTS part (
+ P_PARTKEY INTEGER NOT NULL,
+ P_NAME VARCHAR(55) NOT NULL,
+ P_MFGR CHAR(25) NOT NULL,
+ P_BRAND CHAR(10) NOT NULL,
+ P_TYPE VARCHAR(25) NOT NULL,
+ P_SIZE INTEGER NOT NULL,
+ P_CONTAINER CHAR(10) NOT NULL,
+ P_RETAILPRICE DECIMAL(15,2) NOT NULL,
+ P_COMMENT VARCHAR(23) NOT NULL
+)
+DUPLICATE KEY(P_PARTKEY, P_NAME)
+DISTRIBUTED BY HASH(P_PARTKEY) BUCKETS 32
+
+
diff --git a/regression-test/suites/cloud_p0/multi_cluster/multi_cluster_s3_load/ddl/part_delete.sql b/regression-test/suites/cloud_p0/multi_cluster/multi_cluster_s3_load/ddl/part_delete.sql
new file mode 100644
index 00000000000..ee1a2e51045
--- /dev/null
+++ b/regression-test/suites/cloud_p0/multi_cluster/multi_cluster_s3_load/ddl/part_delete.sql
@@ -0,0 +1 @@
+delete from part where P_PARTKEY>=0;
diff --git a/regression-test/suites/cloud_p0/multi_cluster/multi_cluster_s3_load/ddl/part_load.sql b/regression-test/suites/cloud_p0/multi_cluster/multi_cluster_s3_load/ddl/part_load.sql
new file mode 100644
index 00000000000..af2f7f712da
--- /dev/null
+++ b/regression-test/suites/cloud_p0/multi_cluster/multi_cluster_s3_load/ddl/part_load.sql
@@ -0,0 +1,6 @@
+LOAD LABEL ${loadLabel} (
+ DATA INFILE("s3://${s3BucketName}/regression/tpch/sf100/part.tbl")
+ INTO TABLE part
+ COLUMNS TERMINATED BY "|"
+ (p_partkey, p_name, p_mfgr, p_brand, p_type, p_size, p_container, p_retailprice, p_comment, temp)
+)
diff --git a/regression-test/suites/cloud_p0/multi_cluster/multi_cluster_s3_load/ddl/partsupp.sql b/regression-test/suites/cloud_p0/multi_cluster/multi_cluster_s3_load/ddl/partsupp.sql
new file mode 100644
index 00000000000..13825bf8016
--- /dev/null
+++ b/regression-test/suites/cloud_p0/multi_cluster/multi_cluster_s3_load/ddl/partsupp.sql
@@ -0,0 +1,11 @@
+CREATE TABLE IF NOT EXISTS partsupp (
+ PS_PARTKEY INTEGER NOT NULL,
+ PS_SUPPKEY INTEGER NOT NULL,
+ PS_AVAILQTY INTEGER NOT NULL,
+ PS_SUPPLYCOST DECIMAL(15,2) NOT NULL,
+ PS_COMMENT VARCHAR(199) NOT NULL
+)
+DUPLICATE KEY(PS_PARTKEY, PS_SUPPKEY)
+DISTRIBUTED BY HASH(PS_PARTKEY) BUCKETS 32
+
+
diff --git a/regression-test/suites/cloud_p0/multi_cluster/multi_cluster_s3_load/ddl/partsupp_delete.sql b/regression-test/suites/cloud_p0/multi_cluster/multi_cluster_s3_load/ddl/partsupp_delete.sql
new file mode 100644
index 00000000000..067cff3a6f0
--- /dev/null
+++ b/regression-test/suites/cloud_p0/multi_cluster/multi_cluster_s3_load/ddl/partsupp_delete.sql
@@ -0,0 +1 @@
+delete from partsupp where PS_PARTKEY>=0;
diff --git a/regression-test/suites/cloud_p0/multi_cluster/multi_cluster_s3_load/ddl/partsupp_load.sql b/regression-test/suites/cloud_p0/multi_cluster/multi_cluster_s3_load/ddl/partsupp_load.sql
new file mode 100644
index 00000000000..a38c0f9da0d
--- /dev/null
+++ b/regression-test/suites/cloud_p0/multi_cluster/multi_cluster_s3_load/ddl/partsupp_load.sql
@@ -0,0 +1,6 @@
+LOAD LABEL ${loadLabel} (
+ DATA INFILE("s3://${s3BucketName}/regression/tpch/sf100/partsupp.tbl*")
+ INTO TABLE partsupp
+ COLUMNS TERMINATED BY "|"
+ (ps_partkey, ps_suppkey, ps_availqty, ps_supplycost, ps_comment, temp)
+)
diff --git a/regression-test/suites/cloud_p0/multi_cluster/multi_cluster_s3_load/ddl/region.sql b/regression-test/suites/cloud_p0/multi_cluster/multi_cluster_s3_load/ddl/region.sql
new file mode 100644
index 00000000000..36b8d81508c
--- /dev/null
+++ b/regression-test/suites/cloud_p0/multi_cluster/multi_cluster_s3_load/ddl/region.sql
@@ -0,0 +1,9 @@
+CREATE TABLE IF NOT EXISTS region (
+ R_REGIONKEY INTEGER NOT NULL,
+ R_NAME CHAR(25) NOT NULL,
+ R_COMMENT VARCHAR(152)
+)
+DUPLICATE KEY(R_REGIONKEY, R_NAME)
+DISTRIBUTED BY HASH(R_REGIONKEY) BUCKETS 1
+
+
diff --git a/regression-test/suites/cloud_p0/multi_cluster/multi_cluster_s3_load/ddl/region_delete.sql b/regression-test/suites/cloud_p0/multi_cluster/multi_cluster_s3_load/ddl/region_delete.sql
new file mode 100644
index 00000000000..4945d5105ab
--- /dev/null
+++ b/regression-test/suites/cloud_p0/multi_cluster/multi_cluster_s3_load/ddl/region_delete.sql
@@ -0,0 +1 @@
+delete from region where R_REGIONKEY>=0;
diff --git a/regression-test/suites/cloud_p0/multi_cluster/multi_cluster_s3_load/ddl/region_load.sql b/regression-test/suites/cloud_p0/multi_cluster/multi_cluster_s3_load/ddl/region_load.sql
new file mode 100644
index 00000000000..e07175cbebf
--- /dev/null
+++ b/regression-test/suites/cloud_p0/multi_cluster/multi_cluster_s3_load/ddl/region_load.sql
@@ -0,0 +1,6 @@
+LOAD LABEL ${loadLabel} (
+ DATA INFILE("s3://${s3BucketName}/regression/tpch/sf100/region.tbl")
+ INTO TABLE region
+ COLUMNS TERMINATED BY "|"
+ (r_regionkey, r_name, r_comment, temp)
+)
diff --git a/regression-test/suites/cloud_p0/multi_cluster/multi_cluster_s3_load/ddl/supplier.sql b/regression-test/suites/cloud_p0/multi_cluster/multi_cluster_s3_load/ddl/supplier.sql
new file mode 100644
index 00000000000..cddfcce585c
--- /dev/null
+++ b/regression-test/suites/cloud_p0/multi_cluster/multi_cluster_s3_load/ddl/supplier.sql
@@ -0,0 +1,12 @@
+CREATE TABLE IF NOT EXISTS supplier (
+ S_SUPPKEY INTEGER NOT NULL,
+ S_NAME CHAR(25) NOT NULL,
+ S_ADDRESS VARCHAR(40) NOT NULL,
+ S_NATIONKEY INTEGER NOT NULL,
+ S_PHONE CHAR(15) NOT NULL,
+ S_ACCTBAL DECIMAL(15,2) NOT NULL,
+ S_COMMENT VARCHAR(101) NOT NULL
+)
+DUPLICATE KEY(S_SUPPKEY, S_NAME)
+DISTRIBUTED BY HASH(S_SUPPKEY) BUCKETS 32
+
diff --git a/regression-test/suites/cloud_p0/multi_cluster/multi_cluster_s3_load/ddl/supplier_delete.sql b/regression-test/suites/cloud_p0/multi_cluster/multi_cluster_s3_load/ddl/supplier_delete.sql
new file mode 100644
index 00000000000..d8327a8f856
--- /dev/null
+++ b/regression-test/suites/cloud_p0/multi_cluster/multi_cluster_s3_load/ddl/supplier_delete.sql
@@ -0,0 +1 @@
+delete from supplier where S_SUPPKEY>=0;
diff --git a/regression-test/suites/cloud_p0/multi_cluster/multi_cluster_s3_load/ddl/supplier_load.sql b/regression-test/suites/cloud_p0/multi_cluster/multi_cluster_s3_load/ddl/supplier_load.sql
new file mode 100644
index 00000000000..86e0e12c33d
--- /dev/null
+++ b/regression-test/suites/cloud_p0/multi_cluster/multi_cluster_s3_load/ddl/supplier_load.sql
@@ -0,0 +1,6 @@
+LOAD LABEL ${loadLabel} (
+ DATA INFILE("s3://${s3BucketName}/regression/tpch/sf100/supplier.tbl")
+ INTO TABLE supplier
+ COLUMNS TERMINATED BY "|"
+ (s_suppkey, s_name, s_address, s_nationkey, s_phone, s_acctbal, s_comment, temp)
+)
diff --git a/regression-test/suites/cloud_p0/multi_cluster/multi_cluster_s3_load/load.groovy b/regression-test/suites/cloud_p0/multi_cluster/multi_cluster_s3_load/load.groovy
new file mode 100644
index 00000000000..20108615679
--- /dev/null
+++ b/regression-test/suites/cloud_p0/multi_cluster/multi_cluster_s3_load/load.groovy
@@ -0,0 +1,166 @@
+// 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.
+
+// Most of the cases are copied from https://github.com/trinodb/trino/tree/master
+// /testing/trino-product-tests/src/main/resources/sql-tests/testcases
+// and modified by Doris.
+
+// syntax error:
+// q06 q13 q15
+// Test 23 suites, failed 3 suites
+
+// Note: To filter out tables from sql files, use the following one-liner comamnd
+// sed -nr 's/.*tables: (.*)$/\1/gp' /path/to/*.sql | sed -nr 's/,/\n/gp' | sort | uniq
+
+import groovy.json.JsonOutput
+
+suite("load") {
+ List<String> ipList = new ArrayList<>()
+ List<String> hbPortList = new ArrayList<>()
+ List<String> httpPortList = new ArrayList<>()
+ List<String> beUniqueIdList = new ArrayList<>()
+
+ String[] bes = context.config.multiClusterBes.split(',');
+ println("the value is " + context.config.multiClusterBes);
+ for(String values : bes) {
+ println("the value is " + values);
+ String[] beInfo = values.split(':');
+ ipList.add(beInfo[0]);
+ hbPortList.add(beInfo[1]);
+ httpPortList.add(beInfo[2]);
+ beUniqueIdList.add(beInfo[3]);
+ }
+
+ println("the ip is " + ipList);
+ println("the heartbeat port is " + hbPortList);
+ println("the http port is " + httpPortList);
+ println("the be unique id is " + beUniqueIdList);
+
+ for (unique_id : beUniqueIdList) {
+ resp = get_cluster.call(unique_id);
+ for (cluster : resp) {
+ if (cluster.type == "COMPUTE") {
+ drop_cluster.call(cluster.cluster_name, cluster.cluster_id);
+ }
+ }
+ }
+ wait_cluster_change()
+
+ List<List<Object>> result = sql "show clusters"
+ assertTrue(result.size() == 0);
+
+ add_cluster.call(beUniqueIdList[0], ipList[0], hbPortList[0],
+ "regression_cluster_name0", "regression_cluster_id0");
+ add_cluster.call(beUniqueIdList[1], ipList[1], hbPortList[1],
+ "regression_cluster_name1", "regression_cluster_id1");
+ wait_cluster_change()
+
+ result = sql "show clusters"
+ assertEquals(result.size(), 2);
+
+ sql "use @regression_cluster_name1"
+ result = sql "show clusters"
+
+ sql """ set enable_profile = true """
+
+ def before_cluster0_load_rows = get_be_metric(ipList[0], httpPortList[0], "load_rows");
+ log.info("before_cluster0_load_rows : ${before_cluster0_load_rows}".toString())
+ def before_cluster0_flush = get_be_metric(ipList[0], httpPortList[0], "memtable_flush_total");
+ log.info("before_cluster0_flush : ${before_cluster0_flush}".toString())
+
+ def before_cluster1_load_rows = get_be_metric(ipList[1], httpPortList[1], "load_rows");
+ log.info("before_cluster1_load_rows : ${before_cluster1_load_rows}".toString())
+ def before_cluster1_flush = get_be_metric(ipList[1], httpPortList[1], "memtable_flush_total");
+ log.info("before_cluster1_flush : ${before_cluster1_flush}".toString())
+
+ // Map[tableName, rowCount]
+ def tables = [supplier: 1000000]
+ def s3BucketName = getS3BucketName()
+ def s3WithProperties = """WITH S3 (
+ |"AWS_ACCESS_KEY" = "${getS3AK()}",
+ |"AWS_SECRET_KEY" = "${getS3SK()}",
+ |"AWS_ENDPOINT" = "${getS3Endpoint()}",
+ |"AWS_REGION" = "${getS3Region()}")
+ |PROPERTIES(
+ |"exec_mem_limit" = "8589934592",
+ |"load_parallelism" = "3")""".stripMargin()
+
+
+
+ def uniqueID = Math.abs(UUID.randomUUID().hashCode()).toString()
+ tables.each { table, rows ->
+ // create table if not exists
+ sql new File("""${context.file.parent}/ddl/${table}.sql""").text
+ // check row count
+ def rowCount = sql "select count(*) from ${table}"
+
+ def loadLabel = table + "_" + uniqueID
+ sql new File("""${context.file.parent}/ddl/${table}_delete.sql""").text
+ // load data from cos
+ def loadSql = new File("""${context.file.parent}/ddl/${table}_load.sql""").text.replaceAll("\\\$\\{s3BucketName\\}", s3BucketName)
+ loadSql = loadSql.replaceAll("\\\$\\{loadLabel\\}", loadLabel) + s3WithProperties
+ sql loadSql
+
+ // check load state
+ while (true) {
+ def stateResult = sql "show load where Label = '${loadLabel}'"
+ def loadState = stateResult[stateResult.size() - 1][2].toString()
+ if ("CANCELLED".equalsIgnoreCase(loadState)) {
+ throw new IllegalStateException("load ${loadLabel} failed.")
+ } else if ("FINISHED".equalsIgnoreCase(loadState)) {
+ break
+ }
+ sleep(5000)
+ }
+ }
+
+ def after_cluster0_load_rows = get_be_metric(ipList[0], httpPortList[0], "load_rows");
+ log.info("after_cluster0_load_rows : ${after_cluster0_load_rows}".toString())
+ def after_cluster0_flush = get_be_metric(ipList[0], httpPortList[0], "memtable_flush_total");
+ log.info("after_cluster0_flush : ${after_cluster0_flush}".toString())
+
+ def after_cluster1_load_rows = get_be_metric(ipList[1], httpPortList[1], "load_rows");
+ log.info("after_cluster1_load_rows : ${after_cluster1_load_rows}".toString())
+ def after_cluster1_flush = get_be_metric(ipList[1], httpPortList[1], "memtable_flush_total");
+ log.info("after_cluster1_flush : ${after_cluster1_flush}".toString())
+
+ assertTrue(before_cluster0_load_rows == after_cluster0_load_rows)
+ assertTrue(before_cluster0_flush == after_cluster0_flush)
+
+ assertTrue(before_cluster1_load_rows != after_cluster1_load_rows)
+ assertTrue(before_cluster1_flush != after_cluster1_flush)
+
+ sql "use @regression_cluster_name0"
+ result = sql "show clusters"
+
+ def before_cluster0_query_scan_rows = get_be_metric(ipList[0], httpPortList[0], "query_scan_rows");
+ log.info("before_cluster0_query_scan_rows : ${before_cluster0_query_scan_rows}".toString())
+ def before_cluster1_query_scan_rows = get_be_metric(ipList[1], httpPortList[1], "query_scan_rows");
+ log.info("before_cluster1_query_scan_rows : ${before_cluster1_query_scan_rows}".toString())
+
+ sql """
+ select count(*) from supplier where S_NATIONKEY = 11
+ """
+
+ def after_cluster0_query_scan_rows = get_be_metric(ipList[0], httpPortList[0], "query_scan_rows");
+ log.info("after_cluster0_query_scan_rows : ${after_cluster0_query_scan_rows}".toString())
+ def after_cluster1_query_scan_rows = get_be_metric(ipList[1], httpPortList[1], "query_scan_rows");
+ log.info("after_cluster1_query_scan_rows : ${after_cluster1_query_scan_rows}".toString())
+
+ assertTrue(before_cluster0_query_scan_rows != after_cluster0_query_scan_rows)
+ assertTrue(before_cluster1_query_scan_rows == after_cluster1_query_scan_rows)
+}
diff --git a/regression-test/suites/cloud_p0/multi_cluster/op/test_drop_and_add_cluster.groovy b/regression-test/suites/cloud_p0/multi_cluster/op/test_drop_and_add_cluster.groovy
new file mode 100644
index 00000000000..16c083261d6
--- /dev/null
+++ b/regression-test/suites/cloud_p0/multi_cluster/op/test_drop_and_add_cluster.groovy
@@ -0,0 +1,221 @@
+// 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_drop_and_add_node_and_cluster", "op, nonConcurrent") {
+
+ def clusters = [:]
+ def nodes = [:]
+
+ //decommission node
+ def OpTimeout = 3600
+ def DeltaTime = 30
+ def UseTime = 0
+ String user = 'root'
+
+ def wait_for_decommision_finish = { ip ->
+ def finish_tag = false
+ for(int t = DeltaTime; t <= OpTimeout; t += DeltaTime){
+ def conn = connect(user, '', context.config.jdbcUrl) {
+ def backends = sql_return_maparray "show backends;"
+ for (def be : backends) {
+ def nowTmpNodeStatus = be.Status
+ Map nowNodeStatus = new groovy.json.JsonSlurper().parseText(nowTmpNodeStatus)
+ if (be.Host == ip) {
+ if ( !nowNodeStatus.isActive.toBoolean() && be.SystemDecommissioned.toBoolean() ) {
+ log.info("${ip} finish decommission!")
+ finish_tag = true
+ break
+ }
+ log.info("${ip} not finish decommission, node info: " + be)
+ }
+ }
+ }
+ if (finish_tag) {
+ break
+ }
+ UseTime = t
+ sleep(DeltaTime*1000)
+ }
+ assertTrue(UseTime <= OpTimeout, "decommission ${ip} timeout")
+ }
+
+ def is_add_to_backends = { ip ->
+ def is_add = false
+ def conn = connect(user, '', context.config.jdbcUrl) {
+ def backends = sql_return_maparray "show backends;"
+ for (def be : backends) {
+ def nowTmpNodeStatus = be.Status
+ Map nowNodeStatus = new groovy.json.JsonSlurper().parseText(nowTmpNodeStatus)
+ log.info("be info: " + be)
+ log.info("nowNodeStatus: " + nowNodeStatus)
+ log.info("ip: ${ip}, be.Host: ${be.Host}, Alive status: ${be.Alive}, nowNodeStatus.isActive: ${nowNodeStatus.isActive}, is_add: ${is_add.toString()}")
+ if ( be.Host == ip && be.Alive.toBoolean() && nowNodeStatus.isActive.toBoolean() ) {
+ is_add = true
+ break
+ }
+ }
+ }
+ return is_add
+ }
+
+ def is_add_to_clusters = { cluster_name ->
+ def is_add = false
+ def conn = connect(user, '', context.config.jdbcUrl) {
+ def res = sql_return_maparray "show clusters;"
+ for (def cluster : res) {
+ if (cluster.cluster == cluster_name) {
+ is_add = true
+ break
+ }
+ }
+ }
+ return is_add
+ }
+
+ //do cluster operate in compacibility test at below code
+ // setp1. init nodes and clusters
+
+ // op case only run in root, get cluster basic info
+ def connRes = connect(user, '', context.config.jdbcUrl) {
+ // get cluster info, record
+ def res = sql_return_maparray "show clusters;"
+ // cluster_res = [[cluster:CloudCluster0, is_current:TRUE, users:root], [cluster:CloudCluster1, is_current:FALSE, users:admin]]
+ // if single cluster just not run
+ if ( res.size() < 2 ) {
+ log.info("not multi-cluster, not run. res: ${res}")
+ System.exit(0)
+ }
+ if ( context.config.otherConfigs.get("multiClusterInstance") == null && context.config.otherConfigs.get("metaServiceToken") == null ) {
+ log.info("multiClusterInstance and metaServiceToken not configed in regression-test.groovy, plz config!")
+ System.exit(1)
+ }
+ for (def cluster : res) {
+ clusters[cluster.cluster] = new ArrayList<>()
+ log.info("clusters :" + clusters)
+ }
+ // get node info, record
+ def backends = sql_return_maparray "show backends;"
+ for (def be : backends) {
+ log.info("be :" + be)
+ def TmpCloudInfo = be.Tag
+ Map CloudTag = new groovy.json.JsonSlurper().parseText(TmpCloudInfo)
+ node_cluster = CloudTag.cloud_cluster_name
+ log.info("clusters :" + clusters)
+ clusters[node_cluster].add(be.Host)
+
+ nodes[be.Host] = [:]
+ nodes[be.Host].heartbeat_port = be.HeartbeatPort
+ nodes[be.Host].http_port = be.HttpPort
+ nodes[be.Host].cloud_unique_id = CloudTag.cloud_unique_id
+ nodes[be.Host].cloud_cluster_name = CloudTag.cloud_cluster_name
+ nodes[be.Host].cloud_cluster_id = CloudTag.cloud_cluster_id
+ }
+
+ log.info("nodes: " + nodes)
+ log.info("clusterList: " + clusters);
+ }
+ /*
+ demo
+ clusters = ["CloudCluster0":["172.20.48.164", "172.20.48.165"], "CloudCluster1":["172.20.48.167", "172.20.48.166"]]
+ nodes = [
+ "172.20.48.164":["heartbeat_port":"9056", "http_port":"8046", "cloud_unique_id":"1:CloudCluster0:CLOUD_COMPUTE_ID", "cloud_cluster_name":"CloudCluster0", "cloud_cluster_id":"CLOUD_COMPUTE_ID_CLUSTER0"],
+ "172.20.48.165":["heartbeat_port":"9056", "http_port":"8046", "cloud_unique_id":"1:CloudCluster0:CLOUD_COMPUTE_ID", "cloud_cluster_name":"CloudCluster0", "cloud_cluster_id":"CLOUD_COMPUTE_ID_CLUSTER0"],
+ "172.20.48.166":["heartbeat_port":"9167", "http_port":"8157", "cloud_unique_id":"1:CloudCluster1:CLOUD_COMPUTE_ID", "cloud_cluster_name":"CloudCluster1", "cloud_cluster_id":"CLOUD_COMPUTE_ID_CLUSTER1"],
+ "172.20.48.167":["heartbeat_port":"9167", "http_port":"8157", "cloud_unique_id":"1:CloudCluster1:CLOUD_COMPUTE_ID", "cloud_cluster_name":"CloudCluster1", "cloud_cluster_id":"CLOUD_COMPUTE_ID_CLUSTER1"]
+ ]
+ */
+
+ def pick_to_op_node_ip = ''
+ def pick_to_op_cluster_name = ''
+ clusters.forEach { key, value ->
+ pick_to_op_cluster_name = key
+ for ( be in value ) {
+ pick_to_op_node_ip = be
+ break
+ }
+ }
+ log.info("pick_to_op_node_ip: " + pick_to_op_node_ip)
+ log.info("pick_to_op_cluster_name: " + pick_to_op_cluster_name)
+
+ pick_to_op_node_ip = "172.20.48.166"
+ pick_to_op_cluster_name = "CloudCluster1"
+ //decommsion node
+ d_node.call(
+ nodes[pick_to_op_node_ip].cloud_unique_id,
+ pick_to_op_node_ip,
+ nodes[pick_to_op_node_ip].heartbeat_port,
+ nodes[pick_to_op_node_ip].cloud_cluster_name,
+ nodes[pick_to_op_node_ip].cloud_cluster_id
+ )
+ wait_for_decommision_finish(pick_to_op_node_ip)
+ //drop node
+ drop_node.call(
+ nodes[pick_to_op_node_ip].cloud_unique_id,
+ pick_to_op_node_ip,
+ nodes[pick_to_op_node_ip].heartbeat_port,
+ nodes[pick_to_op_node_ip].cloud_cluster_name,
+ nodes[pick_to_op_node_ip].cloud_cluster_id
+ )
+ // wait 1min
+ sleep(60000)
+ assertTrue( !is_add_to_backends(pick_to_op_node_ip), "${pick_to_op_node_ip} drop fail")
+ //add node
+ add_node.call(
+ nodes[pick_to_op_node_ip].cloud_unique_id,
+ pick_to_op_node_ip,
+ nodes[pick_to_op_node_ip].heartbeat_port,
+ nodes[pick_to_op_node_ip].cloud_cluster_name,
+ nodes[pick_to_op_node_ip].cloud_cluster_id
+ )
+ // wait 1min
+ sleep(60000)
+ assertTrue( is_add_to_backends(pick_to_op_node_ip), "${pick_to_op_node_ip} add fail")
+
+ //drop_cluster
+ drop_cluster.call(pick_to_op_cluster_name, nodes[pick_to_op_node_ip].cloud_cluster_id)
+ sleep(60000)
+ assertTrue( !is_add_to_clusters(pick_to_op_cluster_name), "${pick_to_op_cluster_name} drop fail")
+
+ //add_cluster
+ nodes.forEach { node_ip, node_info ->
+ if (node_info.cloud_cluster_name == pick_to_op_cluster_name){
+ // if cluster not add, so add_cluster
+ if ( !is_add_to_clusters(pick_to_op_cluster_name) ) {
+
+ add_cluster.call(
+ node_info.cloud_unique_id,
+ node_ip,
+ node_info.heartbeat_port,
+ node_info.cloud_cluster_name,
+ node_info.cloud_cluster_id);
+ } else {
+ // cluster already add, but node not add, add_node
+ log.info("node ${node_ip} not add to cluster ${pick_to_op_cluster_name}, add now!")
+ if ( !is_add_to_backends(node_ip)) {
+ add_node.call(
+ node_info.cloud_unique_id,
+ node_ip,
+ node_info.heartbeat_port,
+ node_info.cloud_cluster_name,
+ node_info.cloud_cluster_id)
+ }
+ }
+ }
+ }
+ assertTrue( is_add_to_clusters(pick_to_op_cluster_name), "${pick_to_op_cluster_name} add fail")
+
+}
\ No newline at end of file
diff --git a/regression-test/suites/cloud_p0/multi_cluster/rename_cluster.groovy b/regression-test/suites/cloud_p0/multi_cluster/rename_cluster.groovy
new file mode 100644
index 00000000000..95ad112733f
--- /dev/null
+++ b/regression-test/suites/cloud_p0/multi_cluster/rename_cluster.groovy
@@ -0,0 +1,115 @@
+// 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 groovy.json.JsonOutput
+
+suite("test_rename_cluster") {
+ def token = context.config.metaServiceToken
+ def instance_id = context.config.multiClusterInstance
+
+ List<String> ipList = new ArrayList<>()
+ List<String> hbPortList = new ArrayList<>()
+ List<String> httpPortList = new ArrayList<>()
+ List<String> beUniqueIdList = new ArrayList<>()
+
+ String[] bes = context.config.multiClusterBes.split(',');
+ println("the value is " + context.config.multiClusterBes);
+ for(String values : bes) {
+ println("the value is " + values);
+ String[] beInfo = values.split(':');
+ ipList.add(beInfo[0]);
+ hbPortList.add(beInfo[1]);
+ httpPortList.add(beInfo[2]);
+ beUniqueIdList.add(beInfo[3]);
+ }
+
+ println("the ip is " + ipList);
+ println("the heartbeat port is " + hbPortList);
+ println("the http port is " + httpPortList);
+ println("the be unique id is " + beUniqueIdList);
+
+ for (unique_id : beUniqueIdList) {
+ resp = get_cluster.call(unique_id);
+ for (cluster : resp) {
+ if (cluster.type == "COMPUTE") {
+ drop_cluster.call(cluster.cluster_name, cluster.cluster_id);
+ }
+ }
+ }
+ wait_cluster_change()
+
+ List<List<Object>> result = sql "show clusters"
+ assertTrue(result.size() == 0);
+
+ add_cluster.call(beUniqueIdList[0], ipList[0], hbPortList[0],
+ "regression_cluster_name0", "regression_cluster_id0");
+ wait_cluster_change()
+
+ result = sql "show clusters"
+ assertTrue(result.size() == 1);
+
+ sql "use @regression_cluster_name0"
+ sql """ drop table IF EXISTS table_p2 """
+ sql """
+ CREATE TABLE table_p2 ( k1 int(11) NOT NULL, k2 varchar(20) NOT NULL, k3 int sum NOT NULL )
+ AGGREGATE KEY(k1, k2)
+ PARTITION BY RANGE(k1) (
+ PARTITION p1992 VALUES [("-2147483648"), ("19930101")),
+ PARTITION p1993 VALUES [("19930101"), ("19940101")),
+ PARTITION p1994 VALUES [("19940101"), ("19950101")),
+ PARTITION p1995 VALUES [("19950101"), ("19960101")),
+ PARTITION p1996 VALUES [("19960101"), ("19970101")),
+ PARTITION p1997 VALUES [("19970101"), ("19980101")),
+ PARTITION p1998 VALUES [("19980101"), ("19990101")))
+ DISTRIBUTED BY HASH(k1) BUCKETS 3
+ """
+
+ sql """
+ insert into table_p2 values(1, 'abc', 10);
+ """
+
+ sql """
+ select * from table_p2;
+ """
+
+ rename_cloud_cluster.call("regression_cluster_name1", "regression_cluster_id0");
+ wait_cluster_change()
+
+ // create same name cluster
+ add_cluster.call(beUniqueIdList[1], ipList[1], hbPortList[1],
+ "regression_cluster_name0", "regression_cluster_id2");
+ wait_cluster_change()
+
+ result = sql "show clusters"
+ log.info("clusters: " + result)
+ log.info("backends:")
+ for (def be : sql("show backends")) {
+ log.info("be: " + be)
+ }
+ assertEquals(2, result.size())
+
+ sql "use @regression_cluster_name0"
+ sql """
+ select * from table_p2;
+ """
+
+ sql "use @regression_cluster_name1"
+ sql """
+ select * from table_p2;
+ """
+ sql """ drop table IF EXISTS table_p2 """
+}
diff --git a/regression-test/suites/cloud_p0/multi_cluster/routine_load/test_routine_load.groovy b/regression-test/suites/cloud_p0/multi_cluster/routine_load/test_routine_load.groovy
new file mode 100644
index 00000000000..08d51883ccb
--- /dev/null
+++ b/regression-test/suites/cloud_p0/multi_cluster/routine_load/test_routine_load.groovy
@@ -0,0 +1,189 @@
+// 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.apache.kafka.clients.producer.*;
+import org.apache.kafka.clients.producer.KafkaProducer;
+import org.apache.kafka.clients.consumer.ConsumerRecords
+import org.apache.kafka.clients.consumer.KafkaConsumer
+import org.apache.kafka.clients.admin.AdminClient;
+import org.apache.kafka.clients.admin.CreateTopicsResult;
+import org.apache.kafka.clients.admin.NewTopic;
+import org.apache.kafka.clients.admin.ListTopicsOptions
+
+import java.util.concurrent.ExecutorService
+import java.util.concurrent.Executors
+
+suite("test_routine_load") {
+ def topic = "test-topic"
+ List<String> ipList = new ArrayList<>()
+ List<String> hbPortList = new ArrayList<>()
+ List<String> httpPortList = new ArrayList<>()
+ List<String> beUniqueIdList = new ArrayList<>()
+
+ String[] bes = context.config.multiClusterBes.split(',');
+ println("the value is " + context.config.multiClusterBes);
+ for(String values : bes) {
+ println("the value is " + values);
+ String[] beInfo = values.split(':');
+ ipList.add(beInfo[0]);
+ hbPortList.add(beInfo[1]);
+ httpPortList.add(beInfo[2]);
+ beUniqueIdList.add(beInfo[3]);
+ }
+
+ println("the ip is " + ipList);
+ println("the heartbeat port is " + hbPortList);
+ println("the http port is " + httpPortList);
+ println("the be unique id is " + beUniqueIdList);
+
+ for (unique_id : beUniqueIdList) {
+ resp = get_cluster.call(unique_id);
+ for (cluster : resp) {
+ if (cluster.type == "COMPUTE") {
+ drop_cluster.call(cluster.cluster_name, cluster.cluster_id);
+ }
+ }
+ }
+ wait_cluster_change()
+
+ List<List<Object>> result = sql "show clusters"
+ assertTrue(result.size() == 0);
+
+ add_cluster.call(beUniqueIdList[0], ipList[0], hbPortList[0],
+ "regression_cluster_name0", "regression_cluster_id0");
+ add_cluster.call(beUniqueIdList[1], ipList[1], hbPortList[1],
+ "regression_cluster_name1", "regression_cluster_id1");
+ wait_cluster_change()
+
+ result = sql "show clusters"
+ assertEquals(result.size(), 2);
+
+ sql "use @regression_cluster_name1"
+ result = sql "show clusters"
+
+ sql """ set enable_profile = true """
+
+ def before_cluster0_load_rows = get_be_metric(ipList[0], httpPortList[0], "load_rows");
+ log.info("before_cluster0_load_rows : ${before_cluster0_load_rows}".toString())
+ def before_cluster0_flush = get_be_metric(ipList[0], httpPortList[0], "memtable_flush_total");
+ log.info("before_cluster0_flush : ${before_cluster0_flush}".toString())
+
+ def before_cluster1_load_rows = get_be_metric(ipList[1], httpPortList[1], "load_rows");
+ log.info("before_cluster1_load_rows : ${before_cluster1_load_rows}".toString())
+ def before_cluster1_flush = get_be_metric(ipList[1], httpPortList[1], "memtable_flush_total");
+ log.info("before_cluster1_flush : ${before_cluster1_flush}".toString())
+
+ ExecutorService pool;
+ String kafka_broker_list = context.config.otherConfigs.get("externalEnvIp") + ":" + context.config.otherConfigs.get("kafka_port")
+ pool = Executors.newFixedThreadPool(1)
+ pool.execute{
+ def props = new Properties()
+ props.setProperty(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, kafka_broker_list)
+ props.setProperty(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG,
+ 'org.apache.kafka.common.serialization.StringSerializer')
+ props.setProperty(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,
+ 'org.apache.kafka.common.serialization.StringSerializer')
+
+ AdminClient adminClient = AdminClient.create(props);
+ def delResult = adminClient.deleteTopics([topic] as List<String>)
+ println("the result is " + delResult);
+ def cnt = 0
+ while (!delResult.all().isDone()) {
+ sleep(1000)
+ if (cnt++ > 100) {
+ log.info("failed to wait for delResult")
+ break
+ }
+ }
+
+ NewTopic newTopic = new NewTopic(topic, 10, (short)1); //new NewTopic(topicName, numPartitions, replicationFactor)
+ List<NewTopic> newTopics = new ArrayList<NewTopic>();
+ newTopics.add(newTopic);
+ def createResult = adminClient.createTopics(newTopics);
+ println("the result is " + createResult);
+
+ adminClient.close();
+
+ def producer = new KafkaProducer<String, String>(props)
+ for (int i = 0; i < 30; i++) {
+ String msg_key = i.toString();
+ String msg_value = i.toString() + "|" + "abc" + "|" + (i * 2).toString();
+ def message = new ProducerRecord<String, String>(topic, msg_key, msg_value)
+ producer.send(message)
+ sleep(1000)
+ }
+
+ producer.close()
+ }
+
+ pool.shutdown() //all tasks submitted
+
+ def tableName = "test_routine_load"
+ sql """
+ CREATE TABLE IF NOT EXISTS ${tableName}
+ (
+ id INT,
+ name CHAR(10),
+ score INT
+ )
+ DUPLICATE KEY(id)
+ DISTRIBUTED BY HASH(id) BUCKETS 1;
+ """
+
+ sleep(1000);
+
+ long timestamp = System.currentTimeMillis()
+ String job_name = "routine_load_test_" + String.valueOf(timestamp);
+ sql """
+ CREATE ROUTINE LOAD ${job_name} ON
+ ${tableName} COLUMNS TERMINATED BY "|",
+ COLUMNS(id, name, score)
+ PROPERTIES(
+ "desired_concurrent_number"="2",
+ "max_batch_interval"="6",
+ "max_batch_rows"="200000",
+ "max_batch_size"="104857600")
+ FROM KAFKA(
+ "kafka_broker_list"="${kafka_broker_list}",
+ "kafka_topic"="${topic}",
+ "property.group.id"="gid6",
+ "property.clinet.id"="cid6",
+ "property.kafka_default_offsets"="OFFSET_BEGINNING");
+ """
+
+ while (!pool.isTerminated()){}
+ sleep(30000);
+ order_qt_q1 "select * from ${tableName}"
+
+ def after_cluster0_load_rows = get_be_metric(ipList[0], httpPortList[0], "load_rows");
+ log.info("after_cluster0_load_rows : ${after_cluster0_load_rows}".toString())
+ def after_cluster0_flush = get_be_metric(ipList[0], httpPortList[0], "memtable_flush_total");
+ log.info("after_cluster0_flush : ${after_cluster0_flush}".toString())
+
+ def after_cluster1_load_rows = get_be_metric(ipList[1], httpPortList[1], "load_rows");
+ log.info("after_cluster1_load_rows : ${after_cluster1_load_rows}".toString())
+ def after_cluster1_flush = get_be_metric(ipList[1], httpPortList[1], "memtable_flush_total");
+ log.info("after_cluster1_flush : ${after_cluster1_flush}".toString())
+
+ assertTrue(before_cluster0_load_rows == after_cluster0_load_rows)
+ assertTrue(before_cluster0_flush == after_cluster0_flush)
+
+ assertTrue(before_cluster1_load_rows != after_cluster1_load_rows)
+ assertTrue(before_cluster1_flush != after_cluster1_flush)
+ sql """ DROP TABLE IF EXISTS ${tableName}; """
+}
+
diff --git a/regression-test/suites/cloud_p0/multi_cluster/schema_change/test_bitmap_index_with_drop_cluster.groovy b/regression-test/suites/cloud_p0/multi_cluster/schema_change/test_bitmap_index_with_drop_cluster.groovy
new file mode 100644
index 00000000000..105d50379bc
--- /dev/null
+++ b/regression-test/suites/cloud_p0/multi_cluster/schema_change/test_bitmap_index_with_drop_cluster.groovy
@@ -0,0 +1,143 @@
+// 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 groovy.json.JsonOutput
+
+suite("test_bitmap_index_with_drop_cluster") {
+ def token = context.config.metaServiceToken
+ def instance_id = context.config.multiClusterInstance
+
+ List<String> ipList = new ArrayList<>()
+ List<String> hbPortList = new ArrayList<>()
+ List<String> httpPortList = new ArrayList<>()
+ List<String> beUniqueIdList = new ArrayList<>()
+
+ String[] bes = context.config.multiClusterBes.split(',');
+ println("the value is " + context.config.multiClusterBes);
+ for(String values : bes) {
+ println("the value is " + values);
+ String[] beInfo = values.split(':');
+ ipList.add(beInfo[0]);
+ hbPortList.add(beInfo[1]);
+ httpPortList.add(beInfo[2]);
+ beUniqueIdList.add(beInfo[3]);
+ }
+
+ println("the ip is " + ipList);
+ println("the heartbeat port is " + hbPortList);
+ println("the http port is " + httpPortList);
+ println("the be unique id is " + beUniqueIdList);
+
+ for (unique_id : beUniqueIdList) {
+ resp = get_cluster.call(unique_id);
+ for (cluster : resp) {
+ if (cluster.type == "COMPUTE") {
+ drop_cluster.call(cluster.cluster_name, cluster.cluster_id);
+ }
+ }
+ }
+ wait_cluster_change()
+
+ List<List<Object>> result = sql "show clusters"
+ assertTrue(result.size() == 0);
+
+ // add cluster regression_cluster_name0
+ add_cluster.call(beUniqueIdList[0], ipList[0], hbPortList[0],
+ "regression_cluster_name0", "regression_cluster_id0");
+ wait_cluster_change()
+ result = sql "show clusters"
+ assertTrue(result.size() == 1);
+
+ sql "use @regression_cluster_name0"
+ result = sql "show clusters"
+ for (row : result) {
+ println row
+ if(row[0] == "regression_cluster_name0") {
+ assertTrue(row[1].toString().toLowerCase() == "true")
+ }
+ }
+
+ // create table and add bitmap index
+ def tbName1 = "test_bitmap_index_with_drop_cluster"
+ def getJobState = { tableName ->
+ def jobStateResult = sql """ SHOW ALTER TABLE COLUMN WHERE TableName='${tableName}' ORDER BY createtime DESC LIMIT 1 """
+ return jobStateResult[0][9]
+ }
+ sql "DROP TABLE IF EXISTS ${tbName1} FORCE"
+ sql """
+ CREATE TABLE IF NOT EXISTS ${tbName1} (
+ k1 TINYINT,
+ k2 SMALLINT,
+ k3 INT,
+ k4 BIGINT,
+ k5 CHAR,
+ k6 VARCHAR,
+ k7 DATE,
+ k8 DATETIME,
+ k9 LARGEINT,
+ k10 DECIMAL,
+ k11 BOOLEAN,
+ k12 DATEV2,
+ k13 DATETIMEV2,
+ k14 DATETIMEV2(3),
+ k15 DATETIMEV2(6)
+ )
+ DISTRIBUTED BY HASH(k1) BUCKETS 5 ;
+ """
+
+ sql "insert into ${tbName1} values(1,1,1,1,'1','1','2022-05-31','2022-05-31 10:00:00',1,1.0,1,'2022-05-31','2022-05-31 10:00:00.111111','2022-05-31 10:00:00.111111','2022-05-31 10:00:00.111111');"
+
+ sql """
+ ALTER TABLE ${tbName1}
+ ADD INDEX index1 (k1) USING BITMAP,
+ ADD INDEX index2 (k2) USING BITMAP,
+ ADD INDEX index3 (k3) USING BITMAP,
+ ADD INDEX index4 (k4) USING BITMAP,
+ ADD INDEX index5 (k5) USING BITMAP,
+ ADD INDEX index6 (k6) USING BITMAP,
+ ADD INDEX index7 (k7) USING BITMAP,
+ ADD INDEX index8 (k8) USING BITMAP,
+ ADD INDEX index9 (k9) USING BITMAP,
+ ADD INDEX index10 (k10) USING BITMAP,
+ ADD INDEX index11 (k11) USING BITMAP,
+ ADD INDEX index12 (k12) USING BITMAP,
+ ADD INDEX index13 (k13) USING BITMAP,
+ ADD INDEX index14 (k14) USING BITMAP,
+ ADD INDEX index15 (k15) USING BITMAP;
+ """
+
+ // drop cluster
+ drop_cluster.call("regression_cluster_name0", "regression_cluster_id0");
+ wait_cluster_change()
+
+ // get schema change job state, should cancel
+ int max_try_secs = 60
+ while (max_try_secs--) {
+ String res = getJobState(tbName1)
+ if (res == "CANCELLED") {
+ logger.info(tbName1 + " alter job CANCELLED")
+ break
+ } else {
+ Thread.sleep(1000)
+ if (max_try_secs < 1) {
+ println "test timeout," + "state:" + res
+ assertEquals("FINISHED", res)
+ }
+ }
+ }
+ sql "DROP TABLE IF EXISTS ${tbName1} FORCE"
+}
diff --git a/regression-test/suites/cloud_p0/multi_cluster/schema_change/test_bitmap_index_with_readd_cluster.groovy b/regression-test/suites/cloud_p0/multi_cluster/schema_change/test_bitmap_index_with_readd_cluster.groovy
new file mode 100644
index 00000000000..c55603d3154
--- /dev/null
+++ b/regression-test/suites/cloud_p0/multi_cluster/schema_change/test_bitmap_index_with_readd_cluster.groovy
@@ -0,0 +1,154 @@
+// 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 groovy.json.JsonOutput
+
+suite("test_bitmap_index_with_readd_cluster") {
+ def token = context.config.metaServiceToken
+ def instance_id = context.config.multiClusterInstance
+
+ List<String> ipList = new ArrayList<>()
+ List<String> hbPortList = new ArrayList<>()
+ List<String> httpPortList = new ArrayList<>()
+ List<String> beUniqueIdList = new ArrayList<>()
+
+ String[] bes = context.config.multiClusterBes.split(',');
+ println("the value is " + context.config.multiClusterBes);
+ for(String values : bes) {
+ println("the value is " + values);
+ String[] beInfo = values.split(':');
+ ipList.add(beInfo[0]);
+ hbPortList.add(beInfo[1]);
+ httpPortList.add(beInfo[2]);
+ beUniqueIdList.add(beInfo[3]);
+ }
+
+ println("the ip is " + ipList);
+ println("the heartbeat port is " + hbPortList);
+ println("the http port is " + httpPortList);
+ println("the be unique id is " + beUniqueIdList);
+
+ for (unique_id : beUniqueIdList) {
+ resp = get_cluster.call(unique_id);
+ for (cluster : resp) {
+ if (cluster.type == "COMPUTE") {
+ drop_cluster.call(cluster.cluster_name, cluster.cluster_id);
+ }
+ }
+ }
+ wait_cluster_change()
+
+ List<List<Object>> result = sql "show clusters"
+ assertTrue(result.size() == 0);
+
+ // add cluster regression_cluster_name0
+ add_cluster.call(beUniqueIdList[0], ipList[0], hbPortList[0],
+ "regression_cluster_name0", "regression_cluster_id0");
+ wait_cluster_change()
+ result = sql "show clusters"
+ assertTrue(result.size() == 1);
+
+ sql "use @regression_cluster_name0"
+ result = sql "show clusters"
+ for (row : result) {
+ logger.info("row:${row}");
+ if(row[0] == "regression_cluster_name0") {
+ assertTrue(row[1].toString().toLowerCase() == "true")
+ }
+ }
+
+ // create table and add bitmap index
+ def tbName1 = "test_bitmap_index_with_readd_cluster"
+ def getJobState = { tableName ->
+ def jobStateResult = sql """ SHOW ALTER TABLE COLUMN WHERE TableName='${tableName}' ORDER BY createtime DESC LIMIT 1 """
+ return jobStateResult[0][9]
+ }
+ sql "DROP TABLE IF EXISTS ${tbName1} FORCE"
+ sql """
+ CREATE TABLE IF NOT EXISTS ${tbName1} (
+ k1 TINYINT,
+ k2 SMALLINT,
+ k3 INT,
+ k4 BIGINT,
+ k5 CHAR,
+ k6 VARCHAR,
+ k7 DATE,
+ k8 DATETIME,
+ k9 LARGEINT,
+ k10 DECIMAL,
+ k11 BOOLEAN,
+ k12 DATEV2,
+ k13 DATETIMEV2,
+ k14 DATETIMEV2(3),
+ k15 DATETIMEV2(6)
+ )
+ DISTRIBUTED BY HASH(k1) BUCKETS 5 ;
+ """
+
+ sql "insert into ${tbName1} values(1,1,1,1,'1','1','2022-05-31','2022-05-31 10:00:00',1,1.0,1,'2022-05-31','2022-05-31 10:00:00.111111','2022-05-31 10:00:00.111111','2022-05-31 10:00:00.111111');"
+
+ sql """
+ ALTER TABLE ${tbName1}
+ ADD INDEX index1 (k1) USING BITMAP,
+ ADD INDEX index2 (k2) USING BITMAP,
+ ADD INDEX index3 (k3) USING BITMAP,
+ ADD INDEX index4 (k4) USING BITMAP,
+ ADD INDEX index5 (k5) USING BITMAP,
+ ADD INDEX index6 (k6) USING BITMAP,
+ ADD INDEX index7 (k7) USING BITMAP,
+ ADD INDEX index8 (k8) USING BITMAP,
+ ADD INDEX index9 (k9) USING BITMAP,
+ ADD INDEX index10 (k10) USING BITMAP,
+ ADD INDEX index11 (k11) USING BITMAP,
+ ADD INDEX index12 (k12) USING BITMAP,
+ ADD INDEX index13 (k13) USING BITMAP,
+ ADD INDEX index14 (k14) USING BITMAP,
+ ADD INDEX index15 (k15) USING BITMAP;
+ """
+
+ // drop cluster
+ drop_cluster.call("regression_cluster_name0", "regression_cluster_id0");
+ // add another cluster regression_cluster_name1
+ add_cluster.call(beUniqueIdList[1], ipList[1], hbPortList[1],
+ "regression_cluster_name1", "regression_cluster_id1");
+ wait_cluster_change()
+ result = sql "show clusters"
+ assertTrue(result.size() == 1);
+ for (row : result) {
+ logger.info("row:${row}");
+ if(row[0] == "regression_cluster_name1") {
+ assertTrue(row[1].toString().toLowerCase() == "false")
+ }
+ }
+
+ // get schema change job state, should cancel
+ int max_try_secs = 60
+ while (max_try_secs--) {
+ String res = getJobState(tbName1)
+ if (res == "CANCELLED") {
+ logger.info(tbName1 + " alter job CANCELLED")
+ break
+ } else {
+ Thread.sleep(1000)
+ if (max_try_secs < 1) {
+ println "test timeout," + "state:" + res
+ assertEquals("FINISHED", res)
+ }
+ }
+ }
+ sql "DROP TABLE IF EXISTS ${tbName1} FORCE"
+}
diff --git a/regression-test/suites/cloud_p0/multi_cluster/schema_change/test_materialized_view_with_drop_cluster.groovy b/regression-test/suites/cloud_p0/multi_cluster/schema_change/test_materialized_view_with_drop_cluster.groovy
new file mode 100644
index 00000000000..378c22e7f50
--- /dev/null
+++ b/regression-test/suites/cloud_p0/multi_cluster/schema_change/test_materialized_view_with_drop_cluster.groovy
@@ -0,0 +1,123 @@
+// 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 groovy.json.JsonOutput
+
+suite("test_materialized_with_drop_cluster") {
+ def token = context.config.metaServiceToken
+ def instance_id = context.config.multiClusterInstance
+
+ List<String> ipList = new ArrayList<>()
+ List<String> hbPortList = new ArrayList<>()
+ List<String> httpPortList = new ArrayList<>()
+ List<String> beUniqueIdList = new ArrayList<>()
+
+ String[] bes = context.config.multiClusterBes.split(',');
+
+ for(String values : bes) {
+ String[] beInfo = values.split(':');
+ if (beUniqueIdList.contains(beInfo[3])) {
+ continue
+ }
+ ipList.add(beInfo[0]);
+ hbPortList.add(beInfo[1]);
+ httpPortList.add(beInfo[2]);
+ beUniqueIdList.add(beInfo[3]);
+ }
+
+ logger.info("ipList:{}", ipList)
+ logger.info("hbPortList:{}", hbPortList)
+ logger.info("httpPortList:{}", httpPortList);
+ logger.info("beUniqueIdList:{}", beUniqueIdList);
+
+ for (unique_id : beUniqueIdList) {
+ resp = get_cluster.call(unique_id);
+ for (cluster : resp) {
+ if (cluster.type == "COMPUTE") {
+ drop_cluster.call(cluster.cluster_name, cluster.cluster_id);
+ }
+ }
+ }
+ wait_cluster_change()
+
+ List<List<Object>> result = sql "show clusters"
+ assertTrue(result.size() == 0);
+
+ // add cluster regression_cluster_name0
+ add_cluster.call(beUniqueIdList[0], ipList[0], hbPortList[0],
+ "regression_cluster_name0", "regression_cluster_id0");
+ wait_cluster_change()
+ result = sql "show clusters"
+ assertTrue(result.size() == 1);
+
+ sql "use @regression_cluster_name0"
+ result = sql "show clusters"
+ for (row : result) {
+ logger.info("row:${row}");
+ if(row[0] == "regression_cluster_name0") {
+ assertTrue(row[1].toString().toLowerCase() == "true")
+ }
+ }
+
+ def tbName1 = "test_materialized_with_drop_cluster"
+ def mvName1 = "mv1"
+
+ def getJobMaterializedState = { tableName ->
+ def jobStateResult = sql """ SHOW ALTER TABLE MATERIALIZED VIEW WHERE TableName='${tableName}' ORDER BY CreateTime DESC LIMIT 1; """
+ if (jobStateResult[0][8].equals("CANCELLED") || jobStateResult[0][8].equals("FINISHED")) {
+ logger.info("jobStateResult:{}", jobStateResult)
+ }
+ return jobStateResult[0][8]
+ }
+ sql "DROP TABLE IF EXISTS ${tbName1} FORCE"
+ sql """
+ CREATE TABLE IF NOT EXISTS ${tbName1}(
+ siteid INT(11) NOT NULL,
+ citycode SMALLINT(6) NOT NULL,
+ username VARCHAR(32) NOT NULL,
+ pv BIGINT(20) SUM NOT NULL DEFAULT '0',
+ uv BIGINT(20) SUM NOT NULL DEFAULT '0'
+ )
+ DISTRIBUTED BY HASH(siteid) BUCKETS 5 ;
+ """
+ sql "insert into ${tbName1} values(1, 1, 'test1', 100,100);"
+ sql "insert into ${tbName1} values(2, 1, 'test1', 100,100);"
+ sql "insert into ${tbName1} values(3, 1, 'test1', 100,100);"
+
+
+ sql """create materialized view ${mvName1} as select siteid from ${tbName1} group by siteid;""";
+
+ // drop cluster
+ drop_cluster.call("regression_cluster_name0", "regression_cluster_id0");
+ wait_cluster_change()
+
+ int max_try_secs = 60
+ while (max_try_secs--) {
+ String res = getJobMaterializedState(tbName1)
+ if (res.equals("CANCELLED") || res.equals("FINISHED")) {
+ logger.info("{}", res)
+ break
+ } else {
+ Thread.sleep(5000)
+ if (max_try_secs < 1) {
+ logger.info("test timeout res:{}", res)
+ assertEquals("FINISHED", res)
+ }
+ }
+ }
+ sql "DROP TABLE IF EXISTS ${tbName1} FORCE"
+}
diff --git a/regression-test/suites/cloud_p0/multi_cluster/schema_change/test_materialized_view_with_readd_cluster.groovy b/regression-test/suites/cloud_p0/multi_cluster/schema_change/test_materialized_view_with_readd_cluster.groovy
new file mode 100644
index 00000000000..76ed1c328a4
--- /dev/null
+++ b/regression-test/suites/cloud_p0/multi_cluster/schema_change/test_materialized_view_with_readd_cluster.groovy
@@ -0,0 +1,155 @@
+// 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 groovy.json.JsonOutput
+
+suite("test_materialized_view_with_readd_cluster") {
+ def token = context.config.metaServiceToken
+ def instance_id = context.config.multiClusterInstance
+
+ List<String> ipList = new ArrayList<>()
+ List<String> hbPortList = new ArrayList<>()
+ List<String> httpPortList = new ArrayList<>()
+ List<String> beUniqueIdList = new ArrayList<>()
+
+ String[] bes = context.config.multiClusterBes.split(',');
+ for(String values : bes) {
+ String[] beInfo = values.split(':');
+ if (beUniqueIdList.contains(beInfo[3])) {
+ continue
+ }
+ ipList.add(beInfo[0]);
+ hbPortList.add(beInfo[1]);
+ httpPortList.add(beInfo[2]);
+ beUniqueIdList.add(beInfo[3]);
+ }
+
+ logger.info("ipList:{}", ipList)
+ logger.info("hbPortList:{}", hbPortList)
+ logger.info("httpPortList:{}", httpPortList);
+ logger.info("beUniqueIdList:{}", beUniqueIdList);
+
+ for (unique_id : beUniqueIdList) {
+ resp = get_cluster.call(unique_id);
+ for (cluster : resp) {
+ if (cluster.type == "COMPUTE") {
+ drop_cluster.call(cluster.cluster_name, cluster.cluster_id);
+ }
+ }
+ }
+ wait_cluster_change()
+
+ List<List<Object>> result = sql "show clusters"
+ assertTrue(result.size() == 0);
+
+ // add cluster regression_cluster_name0
+ add_cluster.call(beUniqueIdList[0], ipList[0], hbPortList[0],
+ "regression_cluster_name0", "regression_cluster_id0");
+ wait_cluster_change()
+ result = sql "show clusters"
+ assertTrue(result.size() == 1);
+
+ sql "use @regression_cluster_name0"
+ result = sql "show clusters"
+ for (row : result) {
+ logger.info("row:${row}");
+ if(row[0] == "regression_cluster_name0") {
+ assertTrue(row[1].toString().toLowerCase() == "true")
+ }
+ }
+
+ def tbName1 = "test_materialized_view_with_readd_cluster"
+ def mvName1 = "mv1"
+ def mvName2 = "mv2"
+ def getJobMaterializedState = { tableName ->
+ def jobStateResult = sql """ SHOW ALTER TABLE MATERIALIZED VIEW WHERE TableName='${tableName}' ORDER BY CreateTime DESC LIMIT 1; """
+ if (jobStateResult[0][8].equals("CANCELLED") || jobStateResult[0][8].equals("FINISHED")) {
+ logger.info("jobStateResult:{}", jobStateResult)
+ }
+ return jobStateResult[0][8]
+ }
+ sql "DROP TABLE IF EXISTS ${tbName1} FORCE"
+ sql """
+ CREATE TABLE IF NOT EXISTS ${tbName1}(
+ siteid INT(11) NOT NULL,
+ citycode SMALLINT(6) NOT NULL,
+ username VARCHAR(32) NOT NULL,
+ pv BIGINT(20) SUM NOT NULL DEFAULT '0',
+ uv BIGINT(20) SUM NOT NULL DEFAULT '0'
+ )
+ DISTRIBUTED BY HASH(siteid) BUCKETS 5 ;
+ """
+ sql "insert into ${tbName1} values(1, 1, 'test1', 100,100);"
+ sql "insert into ${tbName1} values(2, 1, 'test1', 100,100);"
+ sql "insert into ${tbName1} values(3, 1, 'test1', 100,100);"
+
+ sql """create materialized view ${mvName1} as select siteid from ${tbName1} group by siteid;""";
+
+ // drop cluster
+ drop_cluster.call("regression_cluster_name0", "regression_cluster_id0");
+ // add another cluster regression_cluster_name1
+ add_cluster.call(beUniqueIdList[1], ipList[1], hbPortList[1],
+ "regression_cluster_name1", "regression_cluster_id1");
+ wait_cluster_change()
+ result = sql "show clusters"
+ assertTrue(result.size() == 1);
+ for (row : result) {
+ logger.info("row:${row}");
+ if(row[0] == "regression_cluster_name1") {
+ assertTrue(row[1].toString().toLowerCase() == "false")
+ }
+ }
+
+ int max_try_secs = 60
+ while (max_try_secs--) {
+ String res = getJobMaterializedState(tbName1)
+ if (res.equals("CANCELLED") || res.equals("FINISHED")) {
+ break
+ } else {
+ Thread.sleep(5000)
+ if (max_try_secs < 1) {
+ println "test timeout," + "state:" + res
+ assertEquals("FINISHED", res)
+ }
+ }
+ }
+
+ sql "use @regression_cluster_name1"
+ result = sql "show clusters"
+ for (row : result) {
+ logger.info("row:${row}");
+ if(row[0] == "regression_cluster_name1") {
+ assertTrue(row[1].toString().toLowerCase() == "true")
+ }
+ }
+
+ sql """create materialized view ${mvName2} as select siteid from ${tbName1} group by siteid;""";
+ max_try_secs = 60
+ while (max_try_secs--) {
+ String res = getJobMaterializedState(tbName1)
+ if (res.equals("CANCELLED") || res.equals("FINISHED")) {
+ break
+ } else {
+ Thread.sleep(5000)
+ if (max_try_secs < 1) {
+ logger.info("test timeout res:{}", res)
+ assertEquals("FINISHED", res)
+ }
+ }
+ }
+ sql "DROP TABLE IF EXISTS ${tbName1}"
+}
diff --git a/regression-test/suites/cloud_p0/multi_cluster/schema_change/test_rollup_with_drop_cluster.groovy b/regression-test/suites/cloud_p0/multi_cluster/schema_change/test_rollup_with_drop_cluster.groovy
new file mode 100644
index 00000000000..1633b8b57c1
--- /dev/null
+++ b/regression-test/suites/cloud_p0/multi_cluster/schema_change/test_rollup_with_drop_cluster.groovy
@@ -0,0 +1,122 @@
+// 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 groovy.json.JsonOutput
+
+suite("test_rollup_with_drop_cluster") {
+ def token = context.config.metaServiceToken
+ def instance_id = context.config.multiClusterInstance
+
+ List<String> ipList = new ArrayList<>()
+ List<String> hbPortList = new ArrayList<>()
+ List<String> httpPortList = new ArrayList<>()
+ List<String> beUniqueIdList = new ArrayList<>()
+
+ String[] bes = context.config.multiClusterBes.split(',');
+ for(String values : bes) {
+ String[] beInfo = values.split(':');
+ if (beUniqueIdList.contains(beInfo[3])) {
+ continue
+ }
+ ipList.add(beInfo[0]);
+ hbPortList.add(beInfo[1]);
+ httpPortList.add(beInfo[2]);
+ beUniqueIdList.add(beInfo[3]);
+ }
+
+ logger.info("ipList:{}", ipList)
+ logger.info("hbPortList:{}", hbPortList)
+ logger.info("httpPortList:{}", httpPortList);
+ logger.info("beUniqueIdList:{}", beUniqueIdList);
+
+ for (unique_id : beUniqueIdList) {
+ resp = get_cluster.call(unique_id);
+ for (cluster : resp) {
+ if (cluster.type == "COMPUTE") {
+ drop_cluster.call(cluster.cluster_name, cluster.cluster_id);
+ }
+ }
+ }
+ wait_cluster_change()
+
+ List<List<Object>> result = sql "show clusters"
+ assertTrue(result.size() == 0);
+
+ // add cluster regression_cluster_name0
+ add_cluster.call(beUniqueIdList[0], ipList[0], hbPortList[0],
+ "regression_cluster_name0", "regression_cluster_id0");
+ wait_cluster_change()
+ result = sql "show clusters"
+ assertTrue(result.size() == 1);
+
+ sql "use @regression_cluster_name0"
+ result = sql "show clusters"
+ for (row : result) {
+ logger.info("row:${row}");
+ if(row[0] == "regression_cluster_name0") {
+ assertTrue(row[1].toString().toLowerCase() == "true")
+ }
+ }
+
+ def tbName1 = "test_rollup_with_drop_cluster"
+ def rollupName1 = "rollupName1"
+
+ def getJobRollupState = { tableName ->
+ def jobStateResult = sql """ SHOW ALTER TABLE ROLLUP WHERE TableName='${tableName}' ORDER BY CreateTime DESC LIMIT 1; """
+ if (jobStateResult[0][8].equals("CANCELLED") || obStateResult[0][8].equals("FINISHED")) {
+ logger.info("jobStateResult:{}", jobStateResult)
+ }
+ return jobStateResult[0][8]
+ }
+ sql "DROP TABLE IF EXISTS ${tbName1} FORCE"
+ sql """
+ CREATE TABLE IF NOT EXISTS ${tbName1}(
+ siteid INT(11) NOT NULL,
+ citycode SMALLINT(6) NOT NULL,
+ username VARCHAR(32) NOT NULL,
+ pv BIGINT(20) SUM NOT NULL DEFAULT '0',
+ uv BIGINT(20) SUM NOT NULL DEFAULT '0'
+ )
+ AGGREGATE KEY (siteid,citycode,username)
+ DISTRIBUTED BY HASH(siteid) BUCKETS 5 ;
+ """
+ sql "insert into ${tbName1} values(1, 1, 'test1', 100,100);"
+ sql "insert into ${tbName1} values(2, 1, 'test1', 100,100);"
+ sql "insert into ${tbName1} values(3, 1, 'test1', 100,100);"
+
+ sql """ALTER TABLE ${tbName1} ADD ROLLUP ${rollupName1}(citycode, pv);"""
+
+ // drop cluster
+ drop_cluster.call("regression_cluster_name0", "regression_cluster_id0");
+ wait_cluster_change()
+
+ // get schema change job state, should cancel
+ int max_try_secs = 60
+ while (max_try_secs--) {
+ String res = getJobRollupState(tbName1)
+ if (res.equals("CANCELLED") || res.equals("FINISHED")) {
+ break
+ } else {
+ Thread.sleep(5000)
+ if (max_try_secs < 1) {
+ logger.info("test timeout res:{}", res)
+ assertEquals("FINISHED", res)
+ }
+ }
+ }
+ sql "DROP TABLE IF EXISTS ${tbName1}"
+}
diff --git a/regression-test/suites/cloud_p0/multi_cluster/schema_change/test_rollup_with_readd_cluster.groovy b/regression-test/suites/cloud_p0/multi_cluster/schema_change/test_rollup_with_readd_cluster.groovy
new file mode 100644
index 00000000000..a8eb4d82e6d
--- /dev/null
+++ b/regression-test/suites/cloud_p0/multi_cluster/schema_change/test_rollup_with_readd_cluster.groovy
@@ -0,0 +1,156 @@
+// 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 groovy.json.JsonOutput
+
+suite("test_rollup_with_readd_cluster") {
+ def token = context.config.metaServiceToken
+ def instance_id = context.config.multiClusterInstance
+
+ List<String> ipList = new ArrayList<>()
+ List<String> hbPortList = new ArrayList<>()
+ List<String> httpPortList = new ArrayList<>()
+ List<String> beUniqueIdList = new ArrayList<>()
+
+ String[] bes = context.config.multiClusterBes.split(',');
+ for(String values : bes) {
+ String[] beInfo = values.split(':');
+ if (beUniqueIdList.contains(beInfo[3])) {
+ continue
+ }
+ ipList.add(beInfo[0]);
+ hbPortList.add(beInfo[1]);
+ httpPortList.add(beInfo[2]);
+ beUniqueIdList.add(beInfo[3]);
+ }
+
+ logger.info("ipList:{}", ipList)
+ logger.info("hbPortList:{}", hbPortList)
+ logger.info("httpPortList:{}", httpPortList);
+ logger.info("beUniqueIdList:{}", beUniqueIdList);
+
+ for (unique_id : beUniqueIdList) {
+ resp = get_cluster.call(unique_id);
+ for (cluster : resp) {
+ if (cluster.type == "COMPUTE") {
+ drop_cluster.call(cluster.cluster_name, cluster.cluster_id);
+ }
+ }
+ }
+ wait_cluster_change()
+
+ List<List<Object>> result = sql "show clusters"
+ assertTrue(result.size() == 0);
+
+ // add cluster regression_cluster_name0
+ add_cluster.call(beUniqueIdList[0], ipList[0], hbPortList[0],
+ "regression_cluster_name0", "regression_cluster_id0");
+ wait_cluster_change()
+ result = sql "show clusters"
+ assertTrue(result.size() == 1);
+
+ sql "use @regression_cluster_name0"
+ result = sql "show clusters"
+ for (row : result) {
+ logger.info("row:${row}");
+ if(row[0] == "regression_cluster_name0") {
+ assertTrue(row[1].toString().toLowerCase() == "true")
+ }
+ }
+
+ def tbName1 = "test_rollup_with_readd_cluster"
+ def rollupName1 = "rollupName1"
+ def rollupName2 = "rollupName2"
+ def getJobRollupState = { tableName ->
+ def jobStateResult = sql """SHOW ALTER TABLE ROLLUP WHERE TableName='${tableName}' ORDER BY CreateTime DESC LIMIT 1; """
+ if (jobStateResult[0][8].equals("CANCELLED") || jobStateResult[0][8].equals("FINISHED")) {
+ logger.info("jobStateResult:{}", jobStateResult)
+ }
+ return jobStateResult[0][8]
+ }
+ sql "DROP TABLE IF EXISTS ${tbName1} FORCE"
+ sql """
+ CREATE TABLE IF NOT EXISTS ${tbName1}(
+ siteid INT(11) NOT NULL,
+ citycode SMALLINT(6) NOT NULL,
+ username VARCHAR(32) NOT NULL,
+ pv BIGINT(20) SUM NOT NULL DEFAULT '0',
+ uv BIGINT(20) SUM NOT NULL DEFAULT '0'
+ )
+ AGGREGATE KEY (siteid,citycode,username)
+ DISTRIBUTED BY HASH(siteid) BUCKETS 5 ;
+ """
+ sql "insert into ${tbName1} values(1, 1, 'test1', 100,100);"
+ sql "insert into ${tbName1} values(2, 1, 'test1', 100,100);"
+ sql "insert into ${tbName1} values(3, 1, 'test1', 100,100);"
+
+ sql """ALTER TABLE ${tbName1} ADD ROLLUP ${rollupName1}(citycode, pv);"""
+
+ // drop cluster
+ drop_cluster.call("regression_cluster_name0", "regression_cluster_id0");
+ // add another cluster regression_cluster_name1
+ add_cluster.call(beUniqueIdList[1], ipList[1], hbPortList[1],
+ "regression_cluster_name1", "regression_cluster_id1");
+ wait_cluster_change()
+ result = sql "show clusters"
+ assertTrue(result.size() == 1);
+ for (row : result) {
+ logger.info("row:${row}");
+ if(row[0] == "regression_cluster_name1") {
+ assertTrue(row[1].toString().toLowerCase() == "false")
+ }
+ }
+
+ int max_try_secs = 60
+ while (max_try_secs--) {
+ String res = getJobRollupState(tbName1)
+ if (res.equals("CANCELLED") || res.equals("FINISHED")) {
+ break
+ } else {
+ Thread.sleep(5000)
+ if (max_try_secs < 1) {
+ logger.info("test timeout res:{}", res)
+ assertEquals("FINISHED", res)
+ }
+ }
+ }
+
+ sql "use @regression_cluster_name1"
+ result = sql "show clusters"
+ for (row : result) {
+ logger.info("row:${row}");
+ if(row[0] == "regression_cluster_name1") {
+ assertTrue(row[1].toString().toLowerCase() == "true")
+ }
+ }
+
+ sql """ALTER TABLE ${tbName1} ADD ROLLUP ${rollupName2}(citycode, pv);"""
+ max_try_secs = 60
+ while (max_try_secs--) {
+ String res = getJobRollupState(tbName1)
+ if (res.equals("CANCELLED") || res.equals("FINISHED")) {
+ break
+ } else {
+ Thread.sleep(5000)
+ if (max_try_secs < 1) {
+ logger.info("test timeout res:{}", res)
+ assertEquals("FINISHED", res)
+ }
+ }
+ }
+ sql "DROP TABLE IF EXISTS ${tbName1}"
+}
diff --git a/regression-test/suites/cloud_p0/multi_cluster/schema_change/test_schema_change_with_drop_cluster.groovy b/regression-test/suites/cloud_p0/multi_cluster/schema_change/test_schema_change_with_drop_cluster.groovy
new file mode 100644
index 00000000000..fe156bbe03e
--- /dev/null
+++ b/regression-test/suites/cloud_p0/multi_cluster/schema_change/test_schema_change_with_drop_cluster.groovy
@@ -0,0 +1,120 @@
+// 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 groovy.json.JsonOutput
+
+suite("test_schema_change_with_drop_cluster") {
+ def token = context.config.metaServiceToken
+ def instance_id = context.config.multiClusterInstance
+
+ List<String> ipList = new ArrayList<>()
+ List<String> hbPortList = new ArrayList<>()
+ List<String> httpPortList = new ArrayList<>()
+ List<String> beUniqueIdList = new ArrayList<>()
+
+ String[] bes = context.config.multiClusterBes.split(',');
+ for(String values : bes) {
+ String[] beInfo = values.split(':');
+ if (beUniqueIdList.contains(beInfo[3])) {
+ continue
+ }
+ ipList.add(beInfo[0]);
+ hbPortList.add(beInfo[1]);
+ httpPortList.add(beInfo[2]);
+ beUniqueIdList.add(beInfo[3]);
+ }
+
+ logger.info("ipList:{}", ipList)
+ logger.info("hbPortList:{}", hbPortList)
+ logger.info("httpPortList:{}", httpPortList);
+ logger.info("beUniqueIdList:{}", beUniqueIdList);
+
+ for (unique_id : beUniqueIdList) {
+ resp = get_cluster.call(unique_id);
+ for (cluster : resp) {
+ if (cluster.type == "COMPUTE") {
+ drop_cluster.call(cluster.cluster_name, cluster.cluster_id);
+ }
+ }
+ }
+ wait_cluster_change()
+
+ List<List<Object>> result = sql "show clusters"
+ assertTrue(result.size() == 0);
+
+ // add cluster regression_cluster_name0
+ add_cluster.call(beUniqueIdList[0], ipList[0], hbPortList[0],
+ "regression_cluster_name0", "regression_cluster_id0");
+ wait_cluster_change()
+ result = sql "show clusters"
+ assertTrue(result.size() == 1);
+
+ sql "use @regression_cluster_name0"
+ result = sql "show clusters"
+ for (row : result) {
+ logger.info("row:${row}");
+ if(row[0] == "regression_cluster_name0") {
+ assertTrue(row[1].toString().toLowerCase() == "true")
+ }
+ }
+
+ def tbName1 = "test_schema_change_with_drop_cluster"
+ def getJobState = { tableName ->
+ def jobStateResult = sql """SHOW ALTER TABLE COLUMN WHERE IndexName='${tableName}' ORDER BY createtime DESC LIMIT 1"""
+ if (jobStateResult[0][9].equals("CANCELLED") || obStateResult[0][9].equals("FINISHED")) {
+ logger.info("jobStateResult:{}", jobStateResult)
+ }
+ return jobStateResult[0][9]
+ }
+ sql "DROP TABLE IF EXISTS ${tbName1} FORCE"
+ sql """
+ CREATE TABLE IF NOT EXISTS ${tbName1}(
+ siteid INT(11) NOT NULL,
+ citycode SMALLINT(6) NOT NULL,
+ username VARCHAR(32) NOT NULL,
+ pv BIGINT(20) SUM NOT NULL DEFAULT '0',
+ uv BIGINT(20) SUM NOT NULL DEFAULT '0'
+ )
+ AGGREGATE KEY (siteid,citycode,username)
+ DISTRIBUTED BY HASH(siteid) BUCKETS 5 ;
+ """
+ sql "insert into ${tbName1} values(1, 1, 'test1', 100,100);"
+ sql "insert into ${tbName1} values(2, 1, 'test1', 100,100);"
+ sql "insert into ${tbName1} values(3, 1, 'test1', 100,100);"
+
+ sql """ALTER TABLE ${tbName1} DROP COLUMN username;"""
+
+ // drop cluster
+ drop_cluster.call("regression_cluster_name0", "regression_cluster_id0");
+ wait_cluster_change()
+
+ // get schema change job state, should cancel
+ int max_try_secs = 60
+ while (max_try_secs--) {
+ String res = getJobState(tbName1)
+ if (res.equals("CANCELLED") || res.equals("FINISHED")) {
+ break
+ } else {
+ Thread.sleep(1000)
+ if (max_try_secs < 1) {
+ logger.info("test timeout res:{}", res)
+ assertEquals("FINISHED", res)
+ }
+ }
+ }
+ sql "DROP TABLE IF EXISTS ${tbName1}"
+}
diff --git a/regression-test/suites/cloud_p0/multi_cluster/schema_change/test_schema_change_with_readd_cluster.groovy b/regression-test/suites/cloud_p0/multi_cluster/schema_change/test_schema_change_with_readd_cluster.groovy
new file mode 100644
index 00000000000..924c3717a55
--- /dev/null
+++ b/regression-test/suites/cloud_p0/multi_cluster/schema_change/test_schema_change_with_readd_cluster.groovy
@@ -0,0 +1,169 @@
+// 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 groovy.json.JsonOutput
+
+suite("test_schema_change_with_readd_cluster") {
+ def token = context.config.metaServiceToken
+ def instance_id = context.config.multiClusterInstance
+
+ List<String> ipList = new ArrayList<>()
+ List<String> hbPortList = new ArrayList<>()
+ List<String> httpPortList = new ArrayList<>()
+ List<String> beUniqueIdList = new ArrayList<>()
+
+ String[] bes = context.config.multiClusterBes.split(',');
+ for(String values : bes) {
+ String[] beInfo = values.split(':');
+ if (beUniqueIdList.contains(beInfo[3])) {
+ continue
+ }
+ ipList.add(beInfo[0]);
+ hbPortList.add(beInfo[1]);
+ httpPortList.add(beInfo[2]);
+ beUniqueIdList.add(beInfo[3]);
+ }
+
+ logger.info("ipList:{}", ipList)
+ logger.info("hbPortList:{}", hbPortList)
+ logger.info("httpPortList:{}", httpPortList);
+ logger.info("beUniqueIdList:{}", beUniqueIdList);
+
+ for (unique_id : beUniqueIdList) {
+ resp = get_cluster.call(unique_id);
+ for (cluster : resp) {
+ if (cluster.type == "COMPUTE") {
+ drop_cluster.call(cluster.cluster_name, cluster.cluster_id);
+ }
+ }
+ }
+ wait_cluster_change()
+
+ List<List<Object>> result = sql "show clusters"
+ assertTrue(result.size() == 0);
+
+ // add cluster regression_cluster_name0
+ add_cluster.call(beUniqueIdList[0], ipList[0], hbPortList[0],
+ "regression_cluster_name0", "regression_cluster_id0");
+ wait_cluster_change()
+ result = sql "show clusters"
+ assertTrue(result.size() == 1);
+
+ sql "use @regression_cluster_name0"
+ result = sql "show clusters"
+ for (row : result) {
+ logger.info("row:${row}")
+ if(row[0] == "regression_cluster_name0") {
+ assertTrue(row[1].toString().toLowerCase() == "true")
+ }
+ }
+
+ def tbName1 = "test_schema_change_with_readd_cluster"
+ def getJobState = { tableName ->
+ def jobStateResult = sql """ SHOW ALTER TABLE COLUMN WHERE IndexName='${tableName}' ORDER BY createtime DESC LIMIT 1 """
+ if (jobStateResult[0][9].equals("CANCELLED") || jobStateResult[0][9].equals("FINISHED")) {
+ logger.info("jobStateResult:{}", jobStateResult)
+ }
+ return jobStateResult[0][9]
+ }
+
+ sql "DROP TABLE IF EXISTS ${tbName1} FORCE"
+ sql """
+ CREATE TABLE IF NOT EXISTS ${tbName1}(
+ siteid INT(11) NOT NULL,
+ citycode SMALLINT(6) NOT NULL,
+ username VARCHAR(32) NOT NULL,
+ pv BIGINT(20) SUM NOT NULL DEFAULT '0',
+ uv BIGINT(20) SUM NOT NULL DEFAULT '0'
+ )
+ AGGREGATE KEY (siteid,citycode,username)
+ DISTRIBUTED BY HASH(siteid) BUCKETS 5 ;
+ """
+ sql "insert into ${tbName1} values(1, 1, 'test1', 100,100);"
+ sql "insert into ${tbName1} values(2, 1, 'test1', 100,100);"
+ sql "insert into ${tbName1} values(3, 1, 'test1', 100,100);"
+
+ sql """ALTER TABLE ${tbName1} DROP COLUMN username;"""
+
+ // drop cluster
+ drop_cluster.call("regression_cluster_name0", "regression_cluster_id0");
+ // add another cluster regression_cluster_name1
+ add_cluster.call(beUniqueIdList[1], ipList[1], hbPortList[1],
+ "regression_cluster_name1", "regression_cluster_id1");
+ wait_cluster_change()
+ result = sql "show clusters"
+ assertTrue(result.size() == 1);
+ for (row : result) {
+ logger.info("row:${row}")
+ if(row[0] == "regression_cluster_name1") {
+ assertTrue(row[1].toString().toLowerCase() == "false")
+ }
+ }
+
+ boolean success = false;
+ // get schema change job state, should cancel
+ int max_try_secs = 60
+ while (max_try_secs--) {
+ String res = getJobState(tbName1)
+ if (res.equals("CANCELLED")) {
+ break
+ } else if (res.equals("FINISHED")) {
+ success = true;
+ break;
+ } else {
+ Thread.sleep(5000)
+ if (max_try_secs < 1) {
+ logger.info("test timeout res:{}", res)
+ assertEquals("FINISHED", res)
+ success = true;
+ }
+ }
+ }
+
+ sql "use @regression_cluster_name1"
+ result = sql "show clusters"
+ for (row : result) {
+ logger.info("row:${row}");
+ if(row[0] == "regression_cluster_name1") {
+ assertTrue(row[1].toString().toLowerCase() == "true")
+ }
+ }
+
+ if (success) {
+ qt_order """ select * from ${tbName1} order by siteid;"""
+ return;
+ }
+
+ sql """ALTER TABLE ${tbName1} DROP COLUMN username;"""
+
+ max_try_secs = 60
+ String res;
+ while (max_try_secs--) {
+ res = getJobState(tbName1)
+ if (res.equals("CANCELLED") || res.equals("FINISHED")) {
+ break
+ } else {
+ Thread.sleep(5000)
+ if (max_try_secs < 1) {
+ logger.info("test timeout res:{}", res)
+ assertEquals("FINISHED", res)
+ }
+ }
+ }
+ qt_order """ select * from ${tbName1} order by siteid;"""
+ sql "DROP TABLE IF EXISTS ${tbName1}"
+}
diff --git a/regression-test/suites/cloud_p0/multi_cluster/smooth_upgrade/ddl/customer.sql b/regression-test/suites/cloud_p0/multi_cluster/smooth_upgrade/ddl/customer.sql
new file mode 100644
index 00000000000..9cc8e6a4a97
--- /dev/null
+++ b/regression-test/suites/cloud_p0/multi_cluster/smooth_upgrade/ddl/customer.sql
@@ -0,0 +1,15 @@
+CREATE TABLE IF NOT EXISTS customer (
+ c_custkey int NOT NULL,
+ c_name VARCHAR(25) NOT NULL,
+ c_address VARCHAR(40) NOT NULL,
+ c_nationkey int NOT NULL,
+ c_phone VARCHAR(15) NOT NULL,
+ c_acctbal decimal(15, 2) NOT NULL,
+ c_mktsegment VARCHAR(10) NOT NULL,
+ c_comment VARCHAR(117) NOT NULL
+)ENGINE=OLAP
+UNIQUE KEY(`c_custkey`)
+COMMENT "OLAP"
+DISTRIBUTED BY HASH(`c_custkey`) BUCKETS 24
+
+
diff --git a/regression-test/suites/cloud_p0/multi_cluster/smooth_upgrade/ddl/customer_delete.sql b/regression-test/suites/cloud_p0/multi_cluster/smooth_upgrade/ddl/customer_delete.sql
new file mode 100644
index 00000000000..a214fe920c9
--- /dev/null
+++ b/regression-test/suites/cloud_p0/multi_cluster/smooth_upgrade/ddl/customer_delete.sql
@@ -0,0 +1 @@
+DELETE FROM customer where C_CUSTKEY >= 0;
diff --git a/regression-test/suites/cloud_p0/multi_cluster/smooth_upgrade/ddl/customer_load.sql b/regression-test/suites/cloud_p0/multi_cluster/smooth_upgrade/ddl/customer_load.sql
new file mode 100644
index 00000000000..4118e750b1f
--- /dev/null
+++ b/regression-test/suites/cloud_p0/multi_cluster/smooth_upgrade/ddl/customer_load.sql
@@ -0,0 +1,6 @@
+LOAD LABEL ${loadLabel} (
+ DATA INFILE("s3://${s3BucketName}/regression/tpch/sf100/customer.tbl")
+ INTO TABLE customer
+ COLUMNS TERMINATED BY "|"
+ (c_custkey, c_name, c_address, c_nationkey, c_phone, c_acctbal, c_mktsegment, c_comment, temp)
+)
diff --git a/regression-test/suites/cloud_p0/multi_cluster/smooth_upgrade/ddl/customer_load_sequence.sql b/regression-test/suites/cloud_p0/multi_cluster/smooth_upgrade/ddl/customer_load_sequence.sql
new file mode 100644
index 00000000000..6b6aa9ca9a8
--- /dev/null
+++ b/regression-test/suites/cloud_p0/multi_cluster/smooth_upgrade/ddl/customer_load_sequence.sql
@@ -0,0 +1,7 @@
+LOAD LABEL ${loadLabel} (
+ DATA INFILE("s3://${s3BucketName}/regression/tpch/sf100/customer.tbl")
+ INTO TABLE customer
+ COLUMNS TERMINATED BY "|"
+ (c_custkey, c_name, c_address, c_nationkey, c_phone, c_acctbal, c_mktsegment, c_comment, temp)
+ ORDER BY c_custkey
+)
diff --git a/regression-test/suites/cloud_p0/multi_cluster/smooth_upgrade/ddl/customer_part_delete.sql b/regression-test/suites/cloud_p0/multi_cluster/smooth_upgrade/ddl/customer_part_delete.sql
new file mode 100644
index 00000000000..8051a6c82cd
--- /dev/null
+++ b/regression-test/suites/cloud_p0/multi_cluster/smooth_upgrade/ddl/customer_part_delete.sql
@@ -0,0 +1 @@
+DELETE FROM customer where C_CUSTKEY > 7500000;
diff --git a/regression-test/suites/cloud_p0/multi_cluster/smooth_upgrade/ddl/customer_sequence.sql b/regression-test/suites/cloud_p0/multi_cluster/smooth_upgrade/ddl/customer_sequence.sql
new file mode 100644
index 00000000000..6da31369d31
--- /dev/null
+++ b/regression-test/suites/cloud_p0/multi_cluster/smooth_upgrade/ddl/customer_sequence.sql
@@ -0,0 +1,17 @@
+CREATE TABLE IF NOT EXISTS customer (
+ c_custkey int NOT NULL,
+ c_name VARCHAR(25) NOT NULL,
+ c_address VARCHAR(40) NOT NULL,
+ c_nationkey int NOT NULL,
+ c_phone VARCHAR(15) NOT NULL,
+ c_acctbal decimal(15, 2) NOT NULL,
+ c_mktsegment VARCHAR(10) NOT NULL,
+ c_comment VARCHAR(117) NOT NULL
+)ENGINE=OLAP
+UNIQUE KEY(`c_custkey`)
+COMMENT "OLAP"
+DISTRIBUTED BY HASH(`c_custkey`) BUCKETS 24
+PROPERTIES (
+ "function_column.sequence_type" = 'int'
+)
+
diff --git a/regression-test/suites/cloud_p0/multi_cluster/smooth_upgrade/ddl/lineitem.sql b/regression-test/suites/cloud_p0/multi_cluster/smooth_upgrade/ddl/lineitem.sql
new file mode 100644
index 00000000000..44810ca16ad
--- /dev/null
+++ b/regression-test/suites/cloud_p0/multi_cluster/smooth_upgrade/ddl/lineitem.sql
@@ -0,0 +1,23 @@
+CREATE TABLE IF NOT EXISTS lineitem (
+ l_shipdate DATE NOT NULL,
+ l_orderkey bigint NOT NULL,
+ l_linenumber int not null,
+ l_partkey int NOT NULL,
+ l_suppkey int not null,
+ l_quantity decimal(15, 2) NOT NULL,
+ l_extendedprice decimal(15, 2) NOT NULL,
+ l_discount decimal(15, 2) NOT NULL,
+ l_tax decimal(15, 2) NOT NULL,
+ l_returnflag VARCHAR(1) NOT NULL,
+ l_linestatus VARCHAR(1) NOT NULL,
+ l_commitdate DATE NOT NULL,
+ l_receiptdate DATE NOT NULL,
+ l_shipinstruct VARCHAR(25) NOT NULL,
+ l_shipmode VARCHAR(10) NOT NULL,
+ l_comment VARCHAR(44) NOT NULL
+)ENGINE=OLAP
+UNIQUE KEY(`l_shipdate`, `l_orderkey`,`l_linenumber`,`l_partkey`,`l_suppkey`)
+COMMENT "OLAP"
+DISTRIBUTED BY HASH(`l_orderkey`) BUCKETS 96
+
+
diff --git a/regression-test/suites/cloud_p0/multi_cluster/smooth_upgrade/ddl/lineitem_delete.sql b/regression-test/suites/cloud_p0/multi_cluster/smooth_upgrade/ddl/lineitem_delete.sql
new file mode 100644
index 00000000000..f141f10e009
--- /dev/null
+++ b/regression-test/suites/cloud_p0/multi_cluster/smooth_upgrade/ddl/lineitem_delete.sql
@@ -0,0 +1 @@
+DELETE FROM lineitem where l_linenumber >= 0;
\ No newline at end of file
diff --git a/regression-test/suites/cloud_p0/multi_cluster/smooth_upgrade/ddl/lineitem_load.sql b/regression-test/suites/cloud_p0/multi_cluster/smooth_upgrade/ddl/lineitem_load.sql
new file mode 100644
index 00000000000..7f1050c5dc3
--- /dev/null
+++ b/regression-test/suites/cloud_p0/multi_cluster/smooth_upgrade/ddl/lineitem_load.sql
@@ -0,0 +1,6 @@
+LOAD LABEL ${loadLabel} (
+ DATA INFILE("s3://${s3BucketName}/regression/tpch/sf100/lineitem.tbl.*")
+ INTO TABLE lineitem
+ COLUMNS TERMINATED BY "|"
+ (l_orderkey, l_partkey, l_suppkey, l_linenumber, l_quantity, l_extendedprice, l_discount, l_tax, l_returnflag,l_linestatus, l_shipdate,l_commitdate,l_receiptdate,l_shipinstruct,l_shipmode,l_comment,temp)
+)
\ No newline at end of file
diff --git a/regression-test/suites/cloud_p0/multi_cluster/smooth_upgrade/ddl/lineitem_load_sequence.sql b/regression-test/suites/cloud_p0/multi_cluster/smooth_upgrade/ddl/lineitem_load_sequence.sql
new file mode 100644
index 00000000000..4c0c304a34a
--- /dev/null
+++ b/regression-test/suites/cloud_p0/multi_cluster/smooth_upgrade/ddl/lineitem_load_sequence.sql
@@ -0,0 +1,7 @@
+LOAD LABEL ${loadLabel} (
+ DATA INFILE("s3://${s3BucketName}/regression/tpch/sf100/lineitem.tbl.*")
+ INTO TABLE lineitem
+ COLUMNS TERMINATED BY "|"
+ (l_orderkey, l_partkey, l_suppkey, l_linenumber, l_quantity, l_extendedprice, l_discount, l_tax, l_returnflag,l_linestatus, l_shipdate,l_commitdate,l_receiptdate,l_shipinstruct,l_shipmode,l_comment,temp)
+ ORDER BY l_orderkey
+)
\ No newline at end of file
diff --git a/regression-test/suites/cloud_p0/multi_cluster/smooth_upgrade/ddl/lineitem_part_delete.sql b/regression-test/suites/cloud_p0/multi_cluster/smooth_upgrade/ddl/lineitem_part_delete.sql
new file mode 100644
index 00000000000..6103c1dee7c
--- /dev/null
+++ b/regression-test/suites/cloud_p0/multi_cluster/smooth_upgrade/ddl/lineitem_part_delete.sql
@@ -0,0 +1 @@
+DELETE FROM lineitem where l_orderkey >= 240000001 and l_orderkey <= 360000000;
\ No newline at end of file
diff --git a/regression-test/suites/cloud_p0/multi_cluster/smooth_upgrade/ddl/lineitem_sequence.sql b/regression-test/suites/cloud_p0/multi_cluster/smooth_upgrade/ddl/lineitem_sequence.sql
new file mode 100644
index 00000000000..cfc2910a6de
--- /dev/null
+++ b/regression-test/suites/cloud_p0/multi_cluster/smooth_upgrade/ddl/lineitem_sequence.sql
@@ -0,0 +1,25 @@
+CREATE TABLE IF NOT EXISTS lineitem (
+ l_shipdate DATE NOT NULL,
+ l_orderkey bigint NOT NULL,
+ l_linenumber int not null,
+ l_partkey int NOT NULL,
+ l_suppkey int not null,
+ l_quantity decimal(15, 2) NOT NULL,
+ l_extendedprice decimal(15, 2) NOT NULL,
+ l_discount decimal(15, 2) NOT NULL,
+ l_tax decimal(15, 2) NOT NULL,
+ l_returnflag VARCHAR(1) NOT NULL,
+ l_linestatus VARCHAR(1) NOT NULL,
+ l_commitdate DATE NOT NULL,
+ l_receiptdate DATE NOT NULL,
+ l_shipinstruct VARCHAR(25) NOT NULL,
+ l_shipmode VARCHAR(10) NOT NULL,
+ l_comment VARCHAR(44) NOT NULL
+)ENGINE=OLAP
+UNIQUE KEY(`l_shipdate`, `l_orderkey`,`l_linenumber`,`l_partkey`,`l_suppkey`)
+COMMENT "OLAP"
+DISTRIBUTED BY HASH(`l_orderkey`) BUCKETS 96
+PROPERTIES (
+ "function_column.sequence_type" = 'DATE'
+)
+
diff --git a/regression-test/suites/cloud_p0/multi_cluster/smooth_upgrade/ddl/nation.sql b/regression-test/suites/cloud_p0/multi_cluster/smooth_upgrade/ddl/nation.sql
new file mode 100644
index 00000000000..288850dd827
--- /dev/null
+++ b/regression-test/suites/cloud_p0/multi_cluster/smooth_upgrade/ddl/nation.sql
@@ -0,0 +1,11 @@
+CREATE TABLE IF NOT EXISTS nation (
+ `n_nationkey` int(11) NOT NULL,
+ `n_name` varchar(25) NOT NULL,
+ `n_regionkey` int(11) NOT NULL,
+ `n_comment` varchar(152) NULL
+) ENGINE=OLAP
+UNIQUE KEY(`N_NATIONKEY`)
+COMMENT "OLAP"
+DISTRIBUTED BY HASH(`N_NATIONKEY`) BUCKETS 1
+;
+
diff --git a/regression-test/suites/cloud_p0/multi_cluster/smooth_upgrade/ddl/nation_delete.sql b/regression-test/suites/cloud_p0/multi_cluster/smooth_upgrade/ddl/nation_delete.sql
new file mode 100644
index 00000000000..b14f6c9e6e9
--- /dev/null
+++ b/regression-test/suites/cloud_p0/multi_cluster/smooth_upgrade/ddl/nation_delete.sql
@@ -0,0 +1 @@
+DELETE FROM nation where n_nationkey >= 0;
\ No newline at end of file
diff --git a/regression-test/suites/cloud_p0/multi_cluster/smooth_upgrade/ddl/nation_load.sql b/regression-test/suites/cloud_p0/multi_cluster/smooth_upgrade/ddl/nation_load.sql
new file mode 100644
index 00000000000..4cc158142eb
--- /dev/null
+++ b/regression-test/suites/cloud_p0/multi_cluster/smooth_upgrade/ddl/nation_load.sql
@@ -0,0 +1,6 @@
+LOAD LABEL ${loadLabel} (
+ DATA INFILE("s3://${s3BucketName}/regression/tpch/sf100/nation.tbl")
+ INTO TABLE nation
+ COLUMNS TERMINATED BY "|"
+ (n_nationkey, n_name, n_regionkey, n_comment, temp)
+)
\ No newline at end of file
diff --git a/regression-test/suites/cloud_p0/multi_cluster/smooth_upgrade/ddl/nation_load_sequence.sql b/regression-test/suites/cloud_p0/multi_cluster/smooth_upgrade/ddl/nation_load_sequence.sql
new file mode 100644
index 00000000000..b85cbb100e2
--- /dev/null
+++ b/regression-test/suites/cloud_p0/multi_cluster/smooth_upgrade/ddl/nation_load_sequence.sql
@@ -0,0 +1,7 @@
+LOAD LABEL ${loadLabel} (
+ DATA INFILE("s3://${s3BucketName}/regression/tpch/sf100/nation.tbl")
+ INTO TABLE nation
+ COLUMNS TERMINATED BY "|"
+ (n_nationkey, n_name, n_regionkey, n_comment, temp)
+ ORDER BY n_nationkey
+)
\ No newline at end of file
diff --git a/regression-test/suites/cloud_p0/multi_cluster/smooth_upgrade/ddl/nation_part_delete.sql b/regression-test/suites/cloud_p0/multi_cluster/smooth_upgrade/ddl/nation_part_delete.sql
new file mode 100644
index 00000000000..6722176fa9b
--- /dev/null
+++ b/regression-test/suites/cloud_p0/multi_cluster/smooth_upgrade/ddl/nation_part_delete.sql
@@ -0,0 +1 @@
+DELETE FROM nation where n_nationkey < 10;
\ No newline at end of file
diff --git a/regression-test/suites/cloud_p0/multi_cluster/smooth_upgrade/ddl/nation_sequence.sql b/regression-test/suites/cloud_p0/multi_cluster/smooth_upgrade/ddl/nation_sequence.sql
new file mode 100644
index 00000000000..eb1afb75e13
--- /dev/null
+++ b/regression-test/suites/cloud_p0/multi_cluster/smooth_upgrade/ddl/nation_sequence.sql
@@ -0,0 +1,13 @@
+CREATE TABLE IF NOT EXISTS nation (
+ `n_nationkey` int(11) NOT NULL,
+ `n_name` varchar(25) NOT NULL,
+ `n_regionkey` int(11) NOT NULL,
+ `n_comment` varchar(152) NULL
+) ENGINE=OLAP
+UNIQUE KEY(`N_NATIONKEY`)
+COMMENT "OLAP"
+DISTRIBUTED BY HASH(`N_NATIONKEY`) BUCKETS 1
+PROPERTIES (
+ "function_column.sequence_type" = 'int'
+);
+
diff --git a/regression-test/suites/cloud_p0/multi_cluster/smooth_upgrade/ddl/orders.sql b/regression-test/suites/cloud_p0/multi_cluster/smooth_upgrade/ddl/orders.sql
new file mode 100644
index 00000000000..06bcfc14edb
--- /dev/null
+++ b/regression-test/suites/cloud_p0/multi_cluster/smooth_upgrade/ddl/orders.sql
@@ -0,0 +1,15 @@
+CREATE TABLE IF NOT EXISTS orders (
+ o_orderkey bigint NOT NULL,
+ o_orderdate DATE NOT NULL,
+ o_custkey int NOT NULL,
+ o_orderstatus VARCHAR(1) NOT NULL,
+ o_totalprice decimal(15, 2) NOT NULL,
+ o_orderpriority VARCHAR(15) NOT NULL,
+ o_clerk VARCHAR(15) NOT NULL,
+ o_shippriority int NOT NULL,
+ o_comment VARCHAR(79) NOT NULL
+)ENGINE=OLAP
+UNIQUE KEY(`o_orderkey`, `o_orderdate`)
+COMMENT "OLAP"
+DISTRIBUTED BY HASH(`o_orderkey`) BUCKETS 96
+
diff --git a/regression-test/suites/cloud_p0/multi_cluster/smooth_upgrade/ddl/orders_delete.sql b/regression-test/suites/cloud_p0/multi_cluster/smooth_upgrade/ddl/orders_delete.sql
new file mode 100644
index 00000000000..7fcb2be1586
--- /dev/null
+++ b/regression-test/suites/cloud_p0/multi_cluster/smooth_upgrade/ddl/orders_delete.sql
@@ -0,0 +1 @@
+DELETE FROM orders where o_orderkey >= 0;
\ No newline at end of file
diff --git a/regression-test/suites/cloud_p0/multi_cluster/smooth_upgrade/ddl/orders_load.sql b/regression-test/suites/cloud_p0/multi_cluster/smooth_upgrade/ddl/orders_load.sql
new file mode 100644
index 00000000000..5c56232f140
--- /dev/null
+++ b/regression-test/suites/cloud_p0/multi_cluster/smooth_upgrade/ddl/orders_load.sql
@@ -0,0 +1,6 @@
+LOAD LABEL ${loadLabel} (
+ DATA INFILE("s3://${s3BucketName}/regression/tpch/sf100/orders.tbl.*")
+ INTO TABLE orders
+ COLUMNS TERMINATED BY "|"
+ (o_orderkey, o_custkey, o_orderstatus, o_totalprice, o_orderdate, o_orderpriority, o_clerk, o_shippriority, o_comment, temp)
+)
\ No newline at end of file
diff --git a/regression-test/suites/cloud_p0/multi_cluster/smooth_upgrade/ddl/orders_load_sequence.sql b/regression-test/suites/cloud_p0/multi_cluster/smooth_upgrade/ddl/orders_load_sequence.sql
new file mode 100644
index 00000000000..f4cb08085e7
--- /dev/null
+++ b/regression-test/suites/cloud_p0/multi_cluster/smooth_upgrade/ddl/orders_load_sequence.sql
@@ -0,0 +1,7 @@
+LOAD LABEL ${loadLabel} (
+ DATA INFILE("s3://${s3BucketName}/regression/tpch/sf100/orders.tbl.*")
+ INTO TABLE orders
+ COLUMNS TERMINATED BY "|"
+ (o_orderkey, o_custkey, o_orderstatus, o_totalprice, o_orderdate, o_orderpriority, o_clerk, o_shippriority, o_comment, temp)
+ ORDER BY o_orderkey
+)
\ No newline at end of file
diff --git a/regression-test/suites/cloud_p0/multi_cluster/smooth_upgrade/ddl/orders_part_delete.sql b/regression-test/suites/cloud_p0/multi_cluster/smooth_upgrade/ddl/orders_part_delete.sql
new file mode 100644
index 00000000000..3787764bb4c
--- /dev/null
+++ b/regression-test/suites/cloud_p0/multi_cluster/smooth_upgrade/ddl/orders_part_delete.sql
@@ -0,0 +1 @@
+DELETE FROM orders where o_orderkey >= 240000001 and o_orderkey <= 360000000;
\ No newline at end of file
diff --git a/regression-test/suites/cloud_p0/multi_cluster/smooth_upgrade/ddl/orders_sequence.sql b/regression-test/suites/cloud_p0/multi_cluster/smooth_upgrade/ddl/orders_sequence.sql
new file mode 100644
index 00000000000..2e1b825d8de
--- /dev/null
+++ b/regression-test/suites/cloud_p0/multi_cluster/smooth_upgrade/ddl/orders_sequence.sql
@@ -0,0 +1,17 @@
+CREATE TABLE IF NOT EXISTS orders (
+ o_orderkey bigint NOT NULL,
+ o_orderdate DATE NOT NULL,
+ o_custkey int NOT NULL,
+ o_orderstatus VARCHAR(1) NOT NULL,
+ o_totalprice decimal(15, 2) NOT NULL,
+ o_orderpriority VARCHAR(15) NOT NULL,
+ o_clerk VARCHAR(15) NOT NULL,
+ o_shippriority int NOT NULL,
+ o_comment VARCHAR(79) NOT NULL
+)ENGINE=OLAP
+UNIQUE KEY(`o_orderkey`, `o_orderdate`)
+COMMENT "OLAP"
+DISTRIBUTED BY HASH(`o_orderkey`) BUCKETS 96
+PROPERTIES (
+ "function_column.sequence_type" = 'bigint'
+)
diff --git a/regression-test/suites/cloud_p0/multi_cluster/smooth_upgrade/ddl/part.sql b/regression-test/suites/cloud_p0/multi_cluster/smooth_upgrade/ddl/part.sql
new file mode 100644
index 00000000000..0141b94b33e
--- /dev/null
+++ b/regression-test/suites/cloud_p0/multi_cluster/smooth_upgrade/ddl/part.sql
@@ -0,0 +1,16 @@
+CREATE TABLE IF NOT EXISTS part (
+ p_partkey int NOT NULL,
+ p_name VARCHAR(55) NOT NULL,
+ p_mfgr VARCHAR(25) NOT NULL,
+ p_brand VARCHAR(10) NOT NULL,
+ p_type VARCHAR(25) NOT NULL,
+ p_size int NOT NULL,
+ p_container VARCHAR(10) NOT NULL,
+ p_retailprice decimal(15, 2) NOT NULL,
+ p_comment VARCHAR(23) NOT NULL
+)ENGINE=OLAP
+UNIQUE KEY(`p_partkey`)
+COMMENT "OLAP"
+DISTRIBUTED BY HASH(`p_partkey`) BUCKETS 24
+
+
diff --git a/regression-test/suites/cloud_p0/multi_cluster/smooth_upgrade/ddl/part_delete.sql b/regression-test/suites/cloud_p0/multi_cluster/smooth_upgrade/ddl/part_delete.sql
new file mode 100644
index 00000000000..62107ab477e
--- /dev/null
+++ b/regression-test/suites/cloud_p0/multi_cluster/smooth_upgrade/ddl/part_delete.sql
@@ -0,0 +1 @@
+DELETE FROM part where p_partkey >= 0;
\ No newline at end of file
diff --git a/regression-test/suites/cloud_p0/multi_cluster/smooth_upgrade/ddl/part_load.sql b/regression-test/suites/cloud_p0/multi_cluster/smooth_upgrade/ddl/part_load.sql
new file mode 100644
index 00000000000..74b3230313b
--- /dev/null
+++ b/regression-test/suites/cloud_p0/multi_cluster/smooth_upgrade/ddl/part_load.sql
@@ -0,0 +1,6 @@
+LOAD LABEL ${loadLabel} (
+ DATA INFILE("s3://${s3BucketName}/regression/tpch/sf100/part.tbl")
+ INTO TABLE part
+ COLUMNS TERMINATED BY "|"
+ (p_partkey, p_name, p_mfgr, p_brand, p_type, p_size, p_container, p_retailprice, p_comment, temp)
+)
\ No newline at end of file
diff --git a/regression-test/suites/cloud_p0/multi_cluster/smooth_upgrade/ddl/part_load_sequence.sql b/regression-test/suites/cloud_p0/multi_cluster/smooth_upgrade/ddl/part_load_sequence.sql
new file mode 100644
index 00000000000..c408150fe88
--- /dev/null
+++ b/regression-test/suites/cloud_p0/multi_cluster/smooth_upgrade/ddl/part_load_sequence.sql
@@ -0,0 +1,7 @@
+LOAD LABEL ${loadLabel} (
+ DATA INFILE("s3://${s3BucketName}/regression/tpch/sf100/part.tbl")
+ INTO TABLE part
+ COLUMNS TERMINATED BY "|"
+ (p_partkey, p_name, p_mfgr, p_brand, p_type, p_size, p_container, p_retailprice, p_comment, temp)
+ ORDER BY p_partkey
+)
\ No newline at end of file
diff --git a/regression-test/suites/cloud_p0/multi_cluster/smooth_upgrade/ddl/part_part_delete.sql b/regression-test/suites/cloud_p0/multi_cluster/smooth_upgrade/ddl/part_part_delete.sql
new file mode 100644
index 00000000000..3061618c5e4
--- /dev/null
+++ b/regression-test/suites/cloud_p0/multi_cluster/smooth_upgrade/ddl/part_part_delete.sql
@@ -0,0 +1 @@
+DELETE FROM part where p_partkey > 10000000;
\ No newline at end of file
diff --git a/regression-test/suites/cloud_p0/multi_cluster/smooth_upgrade/ddl/part_sequence.sql b/regression-test/suites/cloud_p0/multi_cluster/smooth_upgrade/ddl/part_sequence.sql
new file mode 100644
index 00000000000..62137b58277
--- /dev/null
+++ b/regression-test/suites/cloud_p0/multi_cluster/smooth_upgrade/ddl/part_sequence.sql
@@ -0,0 +1,18 @@
+CREATE TABLE IF NOT EXISTS part (
+ p_partkey int NOT NULL,
+ p_name VARCHAR(55) NOT NULL,
+ p_mfgr VARCHAR(25) NOT NULL,
+ p_brand VARCHAR(10) NOT NULL,
+ p_type VARCHAR(25) NOT NULL,
+ p_size int NOT NULL,
+ p_container VARCHAR(10) NOT NULL,
+ p_retailprice decimal(15, 2) NOT NULL,
+ p_comment VARCHAR(23) NOT NULL
+)ENGINE=OLAP
+UNIQUE KEY(`p_partkey`)
+COMMENT "OLAP"
+DISTRIBUTED BY HASH(`p_partkey`) BUCKETS 24
+PROPERTIES (
+ "function_column.sequence_type" = 'int'
+)
+
diff --git a/regression-test/suites/cloud_p0/multi_cluster/smooth_upgrade/ddl/partsupp.sql b/regression-test/suites/cloud_p0/multi_cluster/smooth_upgrade/ddl/partsupp.sql
new file mode 100644
index 00000000000..c800cf76c7d
--- /dev/null
+++ b/regression-test/suites/cloud_p0/multi_cluster/smooth_upgrade/ddl/partsupp.sql
@@ -0,0 +1,12 @@
+CREATE TABLE IF NOT EXISTS partsupp (
+ ps_partkey int NOT NULL,
+ ps_suppkey int NOT NULL,
+ ps_availqty int NOT NULL,
+ ps_supplycost decimal(15, 2) NOT NULL,
+ ps_comment VARCHAR(199) NOT NULL
+)ENGINE=OLAP
+UNIQUE KEY(`ps_partkey`,`ps_suppkey`)
+COMMENT "OLAP"
+DISTRIBUTED BY HASH(`ps_partkey`) BUCKETS 24
+
+
diff --git a/regression-test/suites/cloud_p0/multi_cluster/smooth_upgrade/ddl/partsupp_delete.sql b/regression-test/suites/cloud_p0/multi_cluster/smooth_upgrade/ddl/partsupp_delete.sql
new file mode 100644
index 00000000000..423fb9cec36
--- /dev/null
+++ b/regression-test/suites/cloud_p0/multi_cluster/smooth_upgrade/ddl/partsupp_delete.sql
@@ -0,0 +1 @@
+DELETE FROM partsupp where ps_partkey >= 0;
\ No newline at end of file
diff --git a/regression-test/suites/cloud_p0/multi_cluster/smooth_upgrade/ddl/partsupp_load.sql b/regression-test/suites/cloud_p0/multi_cluster/smooth_upgrade/ddl/partsupp_load.sql
new file mode 100644
index 00000000000..73817fc57a8
--- /dev/null
+++ b/regression-test/suites/cloud_p0/multi_cluster/smooth_upgrade/ddl/partsupp_load.sql
@@ -0,0 +1,6 @@
+LOAD LABEL ${loadLabel} (
+ DATA INFILE("s3://${s3BucketName}/regression/tpch/sf100/partsupp.tbl.*")
+ INTO TABLE partsupp
+ COLUMNS TERMINATED BY "|"
+ (ps_partkey, ps_suppkey, ps_availqty, ps_supplycost, ps_comment, temp)
+)
\ No newline at end of file
diff --git a/regression-test/suites/cloud_p0/multi_cluster/smooth_upgrade/ddl/partsupp_load_sequence.sql b/regression-test/suites/cloud_p0/multi_cluster/smooth_upgrade/ddl/partsupp_load_sequence.sql
new file mode 100644
index 00000000000..2a22675cbdf
--- /dev/null
+++ b/regression-test/suites/cloud_p0/multi_cluster/smooth_upgrade/ddl/partsupp_load_sequence.sql
@@ -0,0 +1,7 @@
+LOAD LABEL ${loadLabel} (
+ DATA INFILE("s3://${s3BucketName}/regression/tpch/sf100/partsupp.tbl.*")
+ INTO TABLE partsupp
+ COLUMNS TERMINATED BY "|"
+ (ps_partkey, ps_suppkey, ps_availqty, ps_supplycost, ps_comment, temp)
+ ORDER BY ps_partkey
+)
\ No newline at end of file
diff --git a/regression-test/suites/cloud_p0/multi_cluster/smooth_upgrade/ddl/partsupp_part_delete.sql b/regression-test/suites/cloud_p0/multi_cluster/smooth_upgrade/ddl/partsupp_part_delete.sql
new file mode 100644
index 00000000000..e3eadade52b
--- /dev/null
+++ b/regression-test/suites/cloud_p0/multi_cluster/smooth_upgrade/ddl/partsupp_part_delete.sql
@@ -0,0 +1 @@
+DELETE FROM partsupp where ps_partkey >= 8000001 and ps_partkey <= 12000000;
\ No newline at end of file
diff --git a/regression-test/suites/cloud_p0/multi_cluster/smooth_upgrade/ddl/partsupp_sequence.sql b/regression-test/suites/cloud_p0/multi_cluster/smooth_upgrade/ddl/partsupp_sequence.sql
new file mode 100644
index 00000000000..a5ed4832454
--- /dev/null
+++ b/regression-test/suites/cloud_p0/multi_cluster/smooth_upgrade/ddl/partsupp_sequence.sql
@@ -0,0 +1,14 @@
+CREATE TABLE IF NOT EXISTS partsupp (
+ ps_partkey int NOT NULL,
+ ps_suppkey int NOT NULL,
+ ps_availqty int NOT NULL,
+ ps_supplycost decimal(15, 2) NOT NULL,
+ ps_comment VARCHAR(199) NOT NULL
+)ENGINE=OLAP
+UNIQUE KEY(`ps_partkey`,`ps_suppkey`)
+COMMENT "OLAP"
+DISTRIBUTED BY HASH(`ps_partkey`) BUCKETS 24
+PROPERTIES (
+ "function_column.sequence_type" = 'int'
+)
+
diff --git a/regression-test/suites/cloud_p0/multi_cluster/smooth_upgrade/ddl/region.sql b/regression-test/suites/cloud_p0/multi_cluster/smooth_upgrade/ddl/region.sql
new file mode 100644
index 00000000000..3444cf1b921
--- /dev/null
+++ b/regression-test/suites/cloud_p0/multi_cluster/smooth_upgrade/ddl/region.sql
@@ -0,0 +1,10 @@
+CREATE TABLE IF NOT EXISTS region (
+ r_regionkey int NOT NULL,
+ r_name VARCHAR(25) NOT NULL,
+ r_comment VARCHAR(152)
+)ENGINE=OLAP
+UNIQUE KEY(`r_regionkey`)
+COMMENT "OLAP"
+DISTRIBUTED BY HASH(`r_regionkey`) BUCKETS 1
+
+
diff --git a/regression-test/suites/cloud_p0/multi_cluster/smooth_upgrade/ddl/region_delete.sql b/regression-test/suites/cloud_p0/multi_cluster/smooth_upgrade/ddl/region_delete.sql
new file mode 100644
index 00000000000..e9829c406dd
--- /dev/null
+++ b/regression-test/suites/cloud_p0/multi_cluster/smooth_upgrade/ddl/region_delete.sql
@@ -0,0 +1 @@
+DELETE FROM region where r_regionkey >= 0;
\ No newline at end of file
diff --git a/regression-test/suites/cloud_p0/multi_cluster/smooth_upgrade/ddl/region_load.sql b/regression-test/suites/cloud_p0/multi_cluster/smooth_upgrade/ddl/region_load.sql
new file mode 100644
index 00000000000..ad051637c2b
--- /dev/null
+++ b/regression-test/suites/cloud_p0/multi_cluster/smooth_upgrade/ddl/region_load.sql
@@ -0,0 +1,6 @@
+LOAD LABEL ${loadLabel} (
+ DATA INFILE("s3://${s3BucketName}/regression/tpch/sf100/region.tbl")
+ INTO TABLE region
+ COLUMNS TERMINATED BY "|"
+ (r_regionkey, r_name, r_comment, temp)
+)
\ No newline at end of file
diff --git a/regression-test/suites/cloud_p0/multi_cluster/smooth_upgrade/ddl/region_load_sequence.sql b/regression-test/suites/cloud_p0/multi_cluster/smooth_upgrade/ddl/region_load_sequence.sql
new file mode 100644
index 00000000000..5d942d5bef6
--- /dev/null
+++ b/regression-test/suites/cloud_p0/multi_cluster/smooth_upgrade/ddl/region_load_sequence.sql
@@ -0,0 +1,7 @@
+LOAD LABEL ${loadLabel} (
+ DATA INFILE("s3://${s3BucketName}/regression/tpch/sf100/region.tbl")
+ INTO TABLE region
+ COLUMNS TERMINATED BY "|"
+ (r_regionkey, r_name, r_comment, temp)
+ ORDER BY r_regionkey
+)
\ No newline at end of file
diff --git a/regression-test/suites/cloud_p0/multi_cluster/smooth_upgrade/ddl/region_part_delete.sql b/regression-test/suites/cloud_p0/multi_cluster/smooth_upgrade/ddl/region_part_delete.sql
new file mode 100644
index 00000000000..f137eb89535
--- /dev/null
+++ b/regression-test/suites/cloud_p0/multi_cluster/smooth_upgrade/ddl/region_part_delete.sql
@@ -0,0 +1 @@
+DELETE FROM region where r_regionkey >= 3;
\ No newline at end of file
diff --git a/regression-test/suites/cloud_p0/multi_cluster/smooth_upgrade/ddl/region_sequence.sql b/regression-test/suites/cloud_p0/multi_cluster/smooth_upgrade/ddl/region_sequence.sql
new file mode 100644
index 00000000000..25e15355419
--- /dev/null
+++ b/regression-test/suites/cloud_p0/multi_cluster/smooth_upgrade/ddl/region_sequence.sql
@@ -0,0 +1,12 @@
+CREATE TABLE IF NOT EXISTS region (
+ r_regionkey int NOT NULL,
+ r_name VARCHAR(25) NOT NULL,
+ r_comment VARCHAR(152)
+)ENGINE=OLAP
+UNIQUE KEY(`r_regionkey`)
+COMMENT "OLAP"
+DISTRIBUTED BY HASH(`r_regionkey`) BUCKETS 1
+PROPERTIES (
+ "function_column.sequence_type" = 'int'
+)
+
diff --git a/regression-test/suites/cloud_p0/multi_cluster/smooth_upgrade/ddl/supplier.sql b/regression-test/suites/cloud_p0/multi_cluster/smooth_upgrade/ddl/supplier.sql
new file mode 100644
index 00000000000..ae4a41465ed
--- /dev/null
+++ b/regression-test/suites/cloud_p0/multi_cluster/smooth_upgrade/ddl/supplier.sql
@@ -0,0 +1,13 @@
+CREATE TABLE IF NOT EXISTS supplier (
+ s_suppkey int NOT NULL,
+ s_name VARCHAR(25) NOT NULL,
+ s_address VARCHAR(40) NOT NULL,
+ s_nationkey int NOT NULL,
+ s_phone VARCHAR(15) NOT NULL,
+ s_acctbal decimal(15, 2) NOT NULL,
+ s_comment VARCHAR(101) NOT NULL
+)ENGINE=OLAP
+UNIQUE KEY(`s_suppkey`)
+COMMENT "OLAP"
+DISTRIBUTED BY HASH(`s_suppkey`) BUCKETS 12
+
diff --git a/regression-test/suites/cloud_p0/multi_cluster/smooth_upgrade/ddl/supplier_delete.sql b/regression-test/suites/cloud_p0/multi_cluster/smooth_upgrade/ddl/supplier_delete.sql
new file mode 100644
index 00000000000..58b35549e42
--- /dev/null
+++ b/regression-test/suites/cloud_p0/multi_cluster/smooth_upgrade/ddl/supplier_delete.sql
@@ -0,0 +1 @@
+DELETE FROM supplier where s_suppkey >= 0;
\ No newline at end of file
diff --git a/regression-test/suites/cloud_p0/multi_cluster/smooth_upgrade/ddl/supplier_load.sql b/regression-test/suites/cloud_p0/multi_cluster/smooth_upgrade/ddl/supplier_load.sql
new file mode 100644
index 00000000000..86e0e12c33d
--- /dev/null
+++ b/regression-test/suites/cloud_p0/multi_cluster/smooth_upgrade/ddl/supplier_load.sql
@@ -0,0 +1,6 @@
+LOAD LABEL ${loadLabel} (
+ DATA INFILE("s3://${s3BucketName}/regression/tpch/sf100/supplier.tbl")
+ INTO TABLE supplier
+ COLUMNS TERMINATED BY "|"
+ (s_suppkey, s_name, s_address, s_nationkey, s_phone, s_acctbal, s_comment, temp)
+)
diff --git a/regression-test/suites/cloud_p0/multi_cluster/smooth_upgrade/ddl/supplier_load_sequence.sql b/regression-test/suites/cloud_p0/multi_cluster/smooth_upgrade/ddl/supplier_load_sequence.sql
new file mode 100644
index 00000000000..86553ffbfab
--- /dev/null
+++ b/regression-test/suites/cloud_p0/multi_cluster/smooth_upgrade/ddl/supplier_load_sequence.sql
@@ -0,0 +1,7 @@
+LOAD LABEL ${loadLabel} (
+ DATA INFILE("s3://${s3BucketName}/regression/tpch/sf100/supplier.tbl")
+ INTO TABLE supplier
+ COLUMNS TERMINATED BY "|"
+ (s_suppkey, s_name, s_address, s_nationkey, s_phone, s_acctbal, s_comment, temp)
+ ORDER BY s_suppkey
+)
diff --git a/regression-test/suites/cloud_p0/multi_cluster/smooth_upgrade/ddl/supplier_part_delete.sql b/regression-test/suites/cloud_p0/multi_cluster/smooth_upgrade/ddl/supplier_part_delete.sql
new file mode 100644
index 00000000000..56e10964937
--- /dev/null
+++ b/regression-test/suites/cloud_p0/multi_cluster/smooth_upgrade/ddl/supplier_part_delete.sql
@@ -0,0 +1 @@
+DELETE FROM supplier where s_suppkey > 500000;
\ No newline at end of file
diff --git a/regression-test/suites/cloud_p0/multi_cluster/smooth_upgrade/ddl/supplier_sequence.sql b/regression-test/suites/cloud_p0/multi_cluster/smooth_upgrade/ddl/supplier_sequence.sql
new file mode 100644
index 00000000000..61213229808
--- /dev/null
+++ b/regression-test/suites/cloud_p0/multi_cluster/smooth_upgrade/ddl/supplier_sequence.sql
@@ -0,0 +1,15 @@
+CREATE TABLE IF NOT EXISTS supplier (
+ s_suppkey int NOT NULL,
+ s_name VARCHAR(25) NOT NULL,
+ s_address VARCHAR(40) NOT NULL,
+ s_nationkey int NOT NULL,
+ s_phone VARCHAR(15) NOT NULL,
+ s_acctbal decimal(15, 2) NOT NULL,
+ s_comment VARCHAR(101) NOT NULL
+)ENGINE=OLAP
+UNIQUE KEY(`s_suppkey`)
+COMMENT "OLAP"
+DISTRIBUTED BY HASH(`s_suppkey`) BUCKETS 12
+PROPERTIES (
+ "function_column.sequence_type" = 'int'
+)
diff --git a/regression-test/suites/cloud_p0/multi_cluster/smooth_upgrade/smooth_upgrade.groovy b/regression-test/suites/cloud_p0/multi_cluster/smooth_upgrade/smooth_upgrade.groovy
new file mode 100644
index 00000000000..2111f573f1e
--- /dev/null
+++ b/regression-test/suites/cloud_p0/multi_cluster/smooth_upgrade/smooth_upgrade.groovy
@@ -0,0 +1,165 @@
+// 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.
+
+/* smooth-upgrade regression
+ * Objective:
+ * 1. old BE finishes inflight load jobs and quit eventually
+ * 2. all load jobs finishes successfully
+ *
+ * Limitations:
+ * 1. only support single BE test
+ * 2. cluster must be created before test
+ * 3. new BE process must be created (no need to add) before test
+ */
+
+import groovy.json.JsonOutput
+
+suite("smooth_upgrade") {
+ // def tables = [nation: 25, customer: 15000000, lineitem: 600037902, orders: 150000000, part: 20000000, partsupp: 80000000, region: 5, supplier: 1000000]
+ def tables = [nation: 25, customer: 15000000, region: 5, supplier: 1000000]
+
+ // new be info (TODO: get from config):
+ def ip = context.config.upgradeNewBeIp
+ def hbPort = context.config.upgradeNewBeHbPort
+ def httpPort = context.config.upgradeNewBeHttpPort
+ def beUniqueId = context.config.upgradeNewBeUniqueId
+
+ def uniqueID = Math.abs(UUID.randomUUID().hashCode()).toString()
+
+ def create_table_and_start_load = { ->
+ def s3BucketName = getS3BucketName()
+ def s3WithProperties = """WITH S3 (
+ |"AWS_ACCESS_KEY" = "${getS3AK()}",
+ |"AWS_SECRET_KEY" = "${getS3SK()}",
+ |"AWS_ENDPOINT" = "${getS3Endpoint()}",
+ |"AWS_REGION" = "${getS3Region()}")
+ |PROPERTIES(
+ |"exec_mem_limit" = "8589934592",
+ |"load_parallelism" = "3")""".stripMargin()
+
+
+
+ tables.each { table, rows ->
+ sql """ DROP TABLE IF EXISTS $table """
+ // create table if not exists
+ sql new File("""${context.file.parentFile}/ddl/${table}.sql""").text
+ def loadLabel = table + "_" + uniqueID
+ // load data from cos
+ def loadSql = new File("""${context.file.parentFile}/ddl/${table}_load.sql""").text.replaceAll("\\\$\\{s3BucketName\\}", s3BucketName)
+ loadSql = loadSql.replaceAll("\\\$\\{loadLabel\\}", loadLabel) + s3WithProperties
+ sql loadSql
+ }
+ }
+
+ def add_new_be = { ->
+ println("the ip is " + ip)
+ println("the heartbeat port is " + hbPort)
+ println("the http port is " + httpPort)
+ println("the be unique id is " + beUniqueId)
+
+ addNodeForSmoothUpgrade(beUniqueId, ip, hbPort, "selectdb_cloud_dev_smooth_upgrade_asan_name_cluster0", "selectdb_cloud_dev_smooth_upgrade_asan_id_cluster0")
+
+ sleep(20000)
+
+ result = sql "show clusters"
+ assertTrue(result.size() == 1)
+ }
+
+ def del_new_be = { ->
+ println("delete be unique id is " + beUniqueId)
+
+ d_node.call(beUniqueId, ip, hbPort, "selectdb_cloud_dev_smooth_upgrade_asan_name_cluster0", "selectdb_cloud_dev_smooth_upgrade_asan_id_cluster0")
+
+ sleep(20000)
+
+ result = sql "show clusters"
+ for (row : result) {
+ println row
+ }
+ assertTrue(result.size() == 1)
+ }
+
+ def wait_and_check_load = { ->
+ tables.each { table, rows ->
+ while (true) {
+ def loadLabel = table + "_" + uniqueID
+ def stateResult = sql "show load where Label = '${loadLabel}'"
+ def loadState = stateResult[stateResult.size() - 1][2].toString()
+ if ("CANCELLED".equalsIgnoreCase(loadState)) {
+ throw new IllegalStateException("load ${loadLabel} failed.")
+ } else if ("FINISHED".equalsIgnoreCase(loadState)) {
+ sql 'sync'
+ for (int i = 1; i <= 5; i++) {
+ def loadRowCount = sql "select count(1) from ${table}"
+ logger.info("select ${table} numbers: ${loadRowCount[0][0]}".toString())
+ assertTrue(loadRowCount[0][0] == rows)
+ }
+ sql new File("""${context.file.parentFile}/ddl/${table}_delete.sql""").text
+ for (int i = 1; i <= 5; i++) {
+ def loadRowCount = sql "select count(1) from ${table}"
+ logger.info("select ${table} numbers: ${loadRowCount[0][0]}".toString())
+ assertTrue(loadRowCount[0][0] == 0)
+ }
+ break
+ }
+ sleep(5000)
+ }
+ }
+ }
+
+ def wait_old_be_inactive = { ->
+ def sub = '"isActive":false'
+ def max_try_milli_secs = 1800000
+ def is_active = true
+ while (max_try_milli_secs > 0) {
+ String[][] result = sql """ show backends; """
+ for (row : result) {
+ println row
+ if (row[22].contains(sub)) {
+ is_active = false
+ break
+ }
+ }
+
+ if (is_active == false) {
+ break
+ }
+
+ Thread.sleep(5000)
+ max_try_milli_secs -= 5000
+ if(max_try_milli_secs <= 0) {
+ assertTrue(1 == 2, "wait old BE timeout")
+ }
+ }
+ }
+
+ sleep(10000)
+ try {
+ create_table_and_start_load.call()
+
+ sleep(10000)
+
+ add_new_be.call()
+
+ wait_old_be_inactive.call()
+
+ wait_and_check_load.call()
+ } finally {
+ del_new_be.call()
+ }
+}
+
diff --git a/regression-test/suites/cloud_p0/multi_cluster/stream_load/stream_load.groovy b/regression-test/suites/cloud_p0/multi_cluster/stream_load/stream_load.groovy
new file mode 100644
index 00000000000..6e702b96c29
--- /dev/null
+++ b/regression-test/suites/cloud_p0/multi_cluster/stream_load/stream_load.groovy
@@ -0,0 +1,322 @@
+// 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 groovy.json.JsonOutput
+
+suite("stream_load") {
+ // case1 specific cluster
+ def tableName3 = "test_all"
+ def tableName4 = "test_all_default_cluster"
+ try {
+ List<String> ipList = new ArrayList<>()
+ List<String> hbPortList = new ArrayList<>()
+ List<String> httpPortList = new ArrayList<>()
+ List<String> beUniqueIdList = new ArrayList<>()
+
+ String[] bes = context.config.multiClusterBes.split(',');
+ println("the value is " + context.config.multiClusterBes);
+ for(String values : bes) {
+ println("the value is " + values);
+ String[] beInfo = values.split(':');
+ ipList.add(beInfo[0]);
+ hbPortList.add(beInfo[1]);
+ httpPortList.add(beInfo[2]);
+ beUniqueIdList.add(beInfo[3]);
+ }
+
+ println("the ip is " + ipList);
+ println("the heartbeat port is " + hbPortList);
+ println("the http port is " + httpPortList);
+ println("the be unique id is " + beUniqueIdList);
+
+ for (unique_id : beUniqueIdList) {
+ resp = get_cluster.call(unique_id);
+ for (cluster : resp) {
+ if (cluster.type == "COMPUTE") {
+ drop_cluster.call(cluster.cluster_name, cluster.cluster_id);
+ }
+ }
+ }
+ sleep(20000)
+
+ List<List<Object>> result = sql "show clusters"
+ assertTrue(result.size() == 0);
+
+ add_cluster.call(beUniqueIdList[0], ipList[0], hbPortList[0],
+ "stream_load_cluster_name0", "stream_load_cluster_id0");
+
+ add_cluster.call(beUniqueIdList[1], ipList[1], hbPortList[1],
+ "stream_load_cluster_name1", "stream_load_cluster_id1");
+ sleep(20000)
+
+ result = sql "show clusters"
+ assertTrue(result.size() == 2);
+
+ for (row : result) {
+ println row
+ }
+
+ sql "SET PROPERTY 'default_cloud_cluster' = ''"
+ sql """ use @stream_load_cluster_name0 """
+
+
+
+
+ sql """
+ CREATE TABLE IF NOT EXISTS ${tableName3} (
+ `k1` int(11) NULL,
+ `k2` tinyint(4) NULL,
+ `k3` smallint(6) NULL,
+ `k4` bigint(20) NULL,
+ `k5` largeint(40) NULL,
+ `k6` float NULL,
+ `k7` double NULL,
+ `k8` decimal(9, 0) NULL,
+ `k9` char(10) NULL,
+ `k10` varchar(1024) NULL,
+ `k11` text NULL,
+ `k12` date NULL,
+ `k13` datetime NULL
+ ) ENGINE=OLAP
+ DISTRIBUTED BY HASH(`k1`) BUCKETS 3
+ ;
+ """
+
+ sql """ set enable_profile = true """
+
+ def before_cluster0_load_rows = get_be_metric(ipList[0], httpPortList[0], "load_rows");
+ log.info("before_cluster0_load_rows : ${before_cluster0_load_rows}".toString())
+ def before_cluster0_flush = get_be_metric(ipList[0], httpPortList[0], "memtable_flush_total");
+ log.info("before_cluster0_flush : ${before_cluster0_flush}".toString())
+
+ def before_cluster1_load_rows = get_be_metric(ipList[1], httpPortList[1], "load_rows");
+ log.info("before_cluster1_load_rows : ${before_cluster1_load_rows}".toString())
+ def before_cluster1_flush = get_be_metric(ipList[1], httpPortList[1], "memtable_flush_total");
+ log.info("before_cluster1_flush : ${before_cluster1_flush}".toString())
+
+ long txnId = -1;
+ streamLoad {
+ table "${tableName3}"
+
+ set 'column_separator', ','
+ set 'cloud_cluster', 'stream_load_cluster_name1'
+
+ file 'all_types.csv'
+ time 10000 // limit inflight 10s
+
+ check { loadResult, exception, startTime, endTime ->
+ if (exception != null) {
+ throw exception
+ }
+ log.info("Stream load result: ${loadResult}".toString())
+ def json = parseJson(loadResult)
+ assertEquals("success", json.Status.toLowerCase())
+ assertEquals(20, json.NumberTotalRows)
+ assertEquals(0, json.NumberFilteredRows)
+ txnId = json.TxnId
+ }
+ }
+ sql "sync"
+ order_qt_all11 "SELECT count(*) FROM ${tableName3}" // 20
+ order_qt_all12 "SELECT count(*) FROM ${tableName3} where k1 <= 10" // 11
+
+ def after_cluster0_load_rows = get_be_metric(ipList[0], httpPortList[0], "load_rows");
+ log.info("after_cluster0_load_rows : ${after_cluster0_load_rows}".toString())
+ def after_cluster0_flush = get_be_metric(ipList[0], httpPortList[0], "memtable_flush_total");
+ log.info("after_cluster0_flush : ${after_cluster0_flush}".toString())
+
+ def after_cluster1_load_rows = get_be_metric(ipList[1], httpPortList[1], "load_rows");
+ log.info("after_cluster1_load_rows : ${after_cluster1_load_rows}".toString())
+ def after_cluster1_flush = get_be_metric(ipList[1], httpPortList[1], "memtable_flush_total");
+ log.info("after_cluster1_flush : ${after_cluster1_flush}".toString())
+
+ assertTrue(before_cluster0_load_rows == after_cluster0_load_rows)
+ assertTrue(before_cluster0_flush == after_cluster0_flush)
+
+ assertTrue(before_cluster1_load_rows < after_cluster1_load_rows)
+ assertTrue(before_cluster1_flush < after_cluster1_flush)
+
+ /*
+ // case2 not specific cluster
+ tableName3 = "test_all_no_specific_cluster"
+ sql """ drop table if exists ${tableName3} """
+
+ sql """
+ CREATE TABLE IF NOT EXISTS ${tableName3} (
+ `k1` int(11) NULL,
+ `k2` tinyint(4) NULL,
+ `k3` smallint(6) NULL,
+ `k4` bigint(20) NULL,
+ `k5` largeint(40) NULL,
+ `k6` float NULL,
+ `k7` double NULL,
+ `k8` decimal(9, 0) NULL,
+ `k9` char(10) NULL,
+ `k10` varchar(1024) NULL,
+ `k11` text NULL,
+ `k12` date NULL,
+ `k13` datetime NULL
+ ) ENGINE=OLAP
+ DISTRIBUTED BY HASH(`k1`) BUCKETS 3
+ ;
+ """
+
+ sql """ set enable_profile = true """
+
+ before_cluster0_load_rows = get_be_metric(ipList[0], httpPortList[0], "load_rows");
+ log.info("before_cluster0_load_rows : ${before_cluster0_load_rows}".toString())
+ before_cluster0_flush = get_be_metric(ipList[0], httpPortList[0], "memtable_flush_total");
+ log.info("before_cluster0_flush : ${before_cluster0_flush}".toString())
+
+ before_cluster1_load_rows = get_be_metric(ipList[1], httpPortList[1], "load_rows");
+ log.info("before_cluster1_load_rows : ${before_cluster1_load_rows}".toString())
+ before_cluster1_flush = get_be_metric(ipList[1], httpPortList[1], "memtable_flush_total");
+ log.info("before_cluster1_flush : ${before_cluster1_flush}".toString())
+
+ txnId = -1;
+ streamLoad {
+ table "${tableName3}"
+
+ set 'column_separator', ','
+
+ file 'all_types.csv'
+ time 10000 // limit inflight 10s
+
+ check { loadResult, exception, startTime, endTime ->
+ if (exception != null) {
+ throw exception
+ }
+ log.info("Stream load result: ${loadResult}".toString())
+ def json = parseJson(loadResult)
+ assertEquals("success", json.Status.toLowerCase())
+ assertEquals(20, json.NumberTotalRows)
+ assertEquals(0, json.NumberFilteredRows)
+ txnId = json.TxnId
+ }
+ }
+ order_qt_all11 "SELECT count(*) FROM ${tableName3}" // 20
+ order_qt_all12 "SELECT count(*) FROM ${tableName3} where k1 <= 10" // 11
+
+ after_cluster0_load_rows = get_be_metric(ipList[0], httpPortList[0], "load_rows");
+ log.info("after_cluster0_load_rows : ${after_cluster0_load_rows}".toString())
+ after_cluster0_flush = get_be_metric(ipList[0], httpPortList[0], "memtable_flush_total");
+ log.info("after_cluster0_flush : ${after_cluster0_flush}".toString())
+
+ after_cluster1_load_rows = get_be_metric(ipList[1], httpPortList[1], "load_rows");
+ log.info("after_cluster1_load_rows : ${after_cluster1_load_rows}".toString())
+ after_cluster1_flush = get_be_metric(ipList[1], httpPortList[1], "memtable_flush_total");
+ log.info("after_cluster1_flush : ${after_cluster1_flush}".toString())
+
+ if (before_cluster0_load_rows == after_cluster0_load_rows) {
+ assertTrue(before_cluster0_flush == after_cluster0_flush)
+ }
+
+ if (before_cluster0_load_rows < after_cluster0_load_rows) {
+ assertTrue(before_cluster0_flush < after_cluster0_flush)
+ }
+
+ if (before_cluster1_load_rows == after_cluster1_load_rows) {
+ assertTrue(before_cluster1_flush == after_cluster1_flush)
+ }
+
+ if (before_cluster1_load_rows < after_cluster1_load_rows) {
+ assertTrue(before_cluster1_flush < after_cluster1_flush)
+ }
+ */
+
+ // case3 default cluster
+ sql "SET PROPERTY 'default_cloud_cluster' = 'stream_load_cluster_name0'"
+ sql """ drop table if exists ${tableName4} """
+
+ sql """
+ CREATE TABLE IF NOT EXISTS ${tableName4} (
+ `k1` int(11) NULL,
+ `k2` tinyint(4) NULL,
+ `k3` smallint(6) NULL,
+ `k4` bigint(20) NULL,
+ `k5` largeint(40) NULL,
+ `k6` float NULL,
+ `k7` double NULL,
+ `k8` decimal(9, 0) NULL,
+ `k9` char(10) NULL,
+ `k10` varchar(1024) NULL,
+ `k11` text NULL,
+ `k12` date NULL,
+ `k13` datetime NULL
+ ) ENGINE=OLAP
+ DISTRIBUTED BY HASH(`k1`) BUCKETS 3
+ ;
+ """
+
+ sql """ set enable_profile = true """
+
+ before_cluster0_load_rows = get_be_metric(ipList[0], httpPortList[0], "load_rows");
+ log.info("before_cluster0_load_rows : ${before_cluster0_load_rows}".toString())
+ before_cluster0_flush = get_be_metric(ipList[0], httpPortList[0], "memtable_flush_total");
+ log.info("before_cluster0_flush : ${before_cluster0_flush}".toString())
+
+ before_cluster1_load_rows = get_be_metric(ipList[1], httpPortList[1], "load_rows");
+ log.info("before_cluster1_load_rows : ${before_cluster1_load_rows}".toString())
+ before_cluster1_flush = get_be_metric(ipList[1], httpPortList[1], "memtable_flush_total");
+ log.info("before_cluster1_flush : ${before_cluster1_flush}".toString())
+
+ txnId = -1;
+ streamLoad {
+ table "${tableName4}"
+
+ set 'column_separator', ','
+
+ file 'all_types.csv'
+ time 10000 // limit inflight 10s
+
+ check { loadResult, exception, startTime, endTime ->
+ if (exception != null) {
+ throw exception
+ }
+ log.info("Stream load result: ${loadResult}".toString())
+ def json = parseJson(loadResult)
+ assertEquals("success", json.Status.toLowerCase())
+ assertEquals(20, json.NumberTotalRows)
+ assertEquals(0, json.NumberFilteredRows)
+ txnId = json.TxnId
+ }
+ }
+ sql "sync"
+ order_qt_all11 "SELECT count(*) FROM ${tableName4}" // 20
+ order_qt_all12 "SELECT count(*) FROM ${tableName4} where k1 <= 10" // 11
+
+ after_cluster0_load_rows = get_be_metric(ipList[0], httpPortList[0], "load_rows");
+ log.info("after_cluster0_load_rows : ${after_cluster0_load_rows}".toString())
+ after_cluster0_flush = get_be_metric(ipList[0], httpPortList[0], "memtable_flush_total");
+ log.info("after_cluster0_flush : ${after_cluster0_flush}".toString())
+
+ after_cluster1_load_rows = get_be_metric(ipList[1], httpPortList[1], "load_rows");
+ log.info("after_cluster1_load_rows : ${after_cluster1_load_rows}".toString())
+ after_cluster1_flush = get_be_metric(ipList[1], httpPortList[1], "memtable_flush_total");
+ log.info("after_cluster1_flush : ${after_cluster1_flush}".toString())
+
+ assertTrue(before_cluster0_load_rows < after_cluster0_load_rows)
+ assertTrue(before_cluster0_flush < after_cluster0_flush)
+
+ assertTrue(before_cluster1_load_rows == after_cluster1_load_rows)
+ assertTrue(before_cluster1_flush == after_cluster1_flush)
+ } finally {
+ sql """ drop table if exists ${tableName3} """
+ sql """ drop table if exists ${tableName4} """
+ }
+}
+
diff --git a/regression-test/suites/cloud_p0/multi_cluster/stream_load/stream_load_2pc.groovy b/regression-test/suites/cloud_p0/multi_cluster/stream_load/stream_load_2pc.groovy
new file mode 100644
index 00000000000..6bda5ac46c5
--- /dev/null
+++ b/regression-test/suites/cloud_p0/multi_cluster/stream_load/stream_load_2pc.groovy
@@ -0,0 +1,396 @@
+// 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 groovy.json.JsonOutput
+
+suite("stream_load_2pc") {
+ List<String> ipList = new ArrayList<>()
+ List<String> hbPortList = new ArrayList<>()
+ List<String> httpPortList = new ArrayList<>()
+ List<String> beUniqueIdList = new ArrayList<>()
+
+ String[] bes = context.config.multiClusterBes.split(',');
+ println("the value is " + context.config.multiClusterBes);
+ for(String values : bes) {
+ println("the value is " + values);
+ String[] beInfo = values.split(':');
+ ipList.add(beInfo[0]);
+ hbPortList.add(beInfo[1]);
+ httpPortList.add(beInfo[2]);
+ beUniqueIdList.add(beInfo[3]);
+ }
+
+ println("the ip is " + ipList);
+ println("the heartbeat port is " + hbPortList);
+ println("the http port is " + httpPortList);
+ println("the be unique id is " + beUniqueIdList);
+
+ for (unique_id : beUniqueIdList) {
+ resp = get_cluster.call(unique_id);
+ for (cluster : resp) {
+ if (cluster.type == "COMPUTE") {
+ drop_cluster.call(cluster.cluster_name, cluster.cluster_id);
+ }
+ }
+ }
+ sleep(20000)
+
+ List<List<Object>> result = sql "show clusters"
+ assertTrue(result.size() == 0);
+
+ add_cluster.call(beUniqueIdList[0], ipList[0], hbPortList[0],
+ "stream_load_2pc_cluster_name0", "stream_load_2pc_cluster_id0");
+
+ add_cluster.call(beUniqueIdList[1], ipList[1], hbPortList[1],
+ "stream_load_2pc_cluster_name1", "stream_load_2pc_cluster_id1");
+ sleep(20000)
+
+ result = sql "show clusters"
+ assertTrue(result.size() == 2);
+
+ for (row : result) {
+ println row
+ }
+
+ sql "SET PROPERTY 'default_cloud_cluster' = ''"
+ sql """ use @stream_load_2pc_cluster_name0 """
+
+ // case1 specific cluster
+ def tableName3 = "test_all_2pc"
+
+
+ sql """
+ CREATE TABLE IF NOT EXISTS ${tableName3} (
+ `k1` int(11) NULL,
+ `k2` tinyint(4) NULL,
+ `k3` smallint(6) NULL,
+ `k4` bigint(20) NULL,
+ `k5` largeint(40) NULL,
+ `k6` float NULL,
+ `k7` double NULL,
+ `k8` decimal(9, 0) NULL,
+ `k9` char(10) NULL,
+ `k10` varchar(1024) NULL,
+ `k11` text NULL,
+ `k12` date NULL,
+ `k13` datetime NULL
+ ) ENGINE=OLAP
+ DISTRIBUTED BY HASH(`k1`) BUCKETS 3
+ ;
+ """
+
+ sql """ set enable_profile = true """
+
+ def before_cluster0_load_rows = get_be_metric(ipList[0], httpPortList[0], "load_rows");
+ log.info("before_cluster0_load_rows : ${before_cluster0_load_rows}".toString())
+ def before_cluster0_flush = get_be_metric(ipList[0], httpPortList[0], "memtable_flush_total");
+ log.info("before_cluster0_flush : ${before_cluster0_flush}".toString())
+
+ def before_cluster1_load_rows = get_be_metric(ipList[1], httpPortList[1], "load_rows");
+ log.info("before_cluster1_load_rows : ${before_cluster1_load_rows}".toString())
+ def before_cluster1_flush = get_be_metric(ipList[1], httpPortList[1], "memtable_flush_total");
+ log.info("before_cluster1_flush : ${before_cluster1_flush}".toString())
+
+ long txnId = -1;
+ // 2pc precommit
+ streamLoad {
+ table "${tableName3}"
+
+ set 'column_separator', ','
+ set 'two_phase_commit', 'true'
+ set 'cloud_cluster', 'stream_load_2pc_cluster_name1'
+
+ file 'all_types.csv'
+ time 10000 // limit inflight 10s
+
+ check { loadResult, exception, startTime, endTime ->
+ if (exception != null) {
+ throw exception
+ }
+ log.info("Stream load result: ${loadResult}".toString())
+ def json = parseJson(loadResult)
+ assertEquals("success", json.Status.toLowerCase())
+ assertEquals(20, json.NumberTotalRows)
+ assertEquals(0, json.NumberFilteredRows)
+ txnId = json.TxnId
+ }
+ }
+ sql "sync"
+ order_qt_all11 "SELECT count(*) FROM ${tableName3}" // 0
+ order_qt_all12 "SELECT count(*) FROM ${tableName3} where k1 <= 10" // 0
+
+ // 2pc commit
+ streamLoad {
+ table "${tableName3}"
+
+ set 'column_separator', ','
+ set 'txn_id', txnId.toString()
+ set 'txn_operation', 'commit'
+
+ file 'all_types.csv'
+ twoPhaseCommit true
+
+ check { loadResult, exception, startTime, endTime ->
+ if (exception != null) {
+ throw exception
+ }
+ log.info("Stream load 2pc commit result: ${loadResult}".toString())
+ def json = parseJson(loadResult)
+ assertEquals("success", json.status.toLowerCase())
+ }
+ }
+
+ order_qt_all11 "SELECT count(*) FROM ${tableName3}" // 20
+ order_qt_all12 "SELECT count(*) FROM ${tableName3} where k1 <= 10" // 11
+
+ def after_cluster0_load_rows = get_be_metric(ipList[0], httpPortList[0], "load_rows");
+ log.info("after_cluster0_load_rows : ${after_cluster0_load_rows}".toString())
+ def after_cluster0_flush = get_be_metric(ipList[0], httpPortList[0], "memtable_flush_total");
+ log.info("after_cluster0_flush : ${after_cluster0_flush}".toString())
+
+ def after_cluster1_load_rows = get_be_metric(ipList[1], httpPortList[1], "load_rows");
+ log.info("after_cluster1_load_rows : ${after_cluster1_load_rows}".toString())
+ def after_cluster1_flush = get_be_metric(ipList[1], httpPortList[1], "memtable_flush_total");
+ log.info("after_cluster1_flush : ${after_cluster1_flush}".toString())
+
+ assertTrue(before_cluster0_load_rows == after_cluster0_load_rows)
+ assertTrue(before_cluster0_flush == after_cluster0_flush)
+
+ assertTrue(before_cluster1_load_rows < after_cluster1_load_rows)
+ assertTrue(before_cluster1_flush < after_cluster1_flush)
+
+ /*
+ // case2 not specific cluster
+ tableName3 = "test_all_2pc_no_specific_cluster"
+ sql """ drop table if exists ${tableName3} """
+
+ sql """
+ CREATE TABLE IF NOT EXISTS ${tableName3} (
+ `k1` int(11) NULL,
+ `k2` tinyint(4) NULL,
+ `k3` smallint(6) NULL,
+ `k4` bigint(20) NULL,
+ `k5` largeint(40) NULL,
+ `k6` float NULL,
+ `k7` double NULL,
+ `k8` decimal(9, 0) NULL,
+ `k9` char(10) NULL,
+ `k10` varchar(1024) NULL,
+ `k11` text NULL,
+ `k12` date NULL,
+ `k13` datetime NULL
+ ) ENGINE=OLAP
+ DISTRIBUTED BY HASH(`k1`) BUCKETS 3
+ ;
+ """
+
+ sql """ set enable_profile = true """
+
+ before_cluster0_load_rows = get_be_metric(ipList[0], httpPortList[0], "load_rows");
+ log.info("before_cluster0_load_rows : ${before_cluster0_load_rows}".toString())
+ before_cluster0_flush = get_be_metric(ipList[0], httpPortList[0], "memtable_flush_total");
+ log.info("before_cluster0_flush : ${before_cluster0_flush}".toString())
+
+ before_cluster1_load_rows = get_be_metric(ipList[1], httpPortList[1], "load_rows");
+ log.info("before_cluster1_load_rows : ${before_cluster1_load_rows}".toString())
+ before_cluster1_flush = get_be_metric(ipList[1], httpPortList[1], "memtable_flush_total");
+ log.info("before_cluster1_flush : ${before_cluster1_flush}".toString())
+
+ txnId = -1;
+ // 2pc precommit
+ streamLoad {
+ table "${tableName3}"
+
+ set 'column_separator', ','
+ set 'two_phase_commit', 'true'
+
+ file 'all_types.csv'
+ time 10000 // limit inflight 10s
+
+ check { loadResult, exception, startTime, endTime ->
+ if (exception != null) {
+ throw exception
+ }
+ log.info("Stream load result: ${loadResult}".toString())
+ def json = parseJson(loadResult)
+ assertEquals("success", json.Status.toLowerCase())
+ assertEquals(20, json.NumberTotalRows)
+ assertEquals(0, json.NumberFilteredRows)
+ txnId = json.TxnId
+ }
+ }
+ sql "sync"
+ order_qt_all11 "SELECT count(*) FROM ${tableName3}" // 0
+ order_qt_all12 "SELECT count(*) FROM ${tableName3} where k1 <= 10" // 0
+
+ // 2pc commit
+ streamLoad {
+ table "${tableName3}"
+
+ set 'column_separator', ','
+ set 'txn_id', txnId.toString()
+ set 'txn_operation', 'commit'
+
+ file 'all_types.csv'
+ twoPhaseCommit true
+
+ check { loadResult, exception, startTime, endTime ->
+ if (exception != null) {
+ throw exception
+ }
+ log.info("Stream load 2pc commit result: ${loadResult}".toString())
+ def json = parseJson(loadResult)
+ assertEquals("success", json.status.toLowerCase())
+ }
+ }
+
+ order_qt_all11 "SELECT count(*) FROM ${tableName3}" // 20
+ order_qt_all12 "SELECT count(*) FROM ${tableName3} where k1 <= 10" // 11
+
+ after_cluster0_load_rows = get_be_metric(ipList[0], httpPortList[0], "load_rows");
+ log.info("after_cluster0_load_rows : ${after_cluster0_load_rows}".toString())
+ after_cluster0_flush = get_be_metric(ipList[0], httpPortList[0], "memtable_flush_total");
+ log.info("after_cluster0_flush : ${after_cluster0_flush}".toString())
+
+ after_cluster1_load_rows = get_be_metric(ipList[1], httpPortList[1], "load_rows");
+ log.info("after_cluster1_load_rows : ${after_cluster1_load_rows}".toString())
+ after_cluster1_flush = get_be_metric(ipList[1], httpPortList[1], "memtable_flush_total");
+ log.info("after_cluster1_flush : ${after_cluster1_flush}".toString())
+
+ if (before_cluster0_load_rows == after_cluster0_load_rows) {
+ assertTrue(before_cluster0_flush == after_cluster0_flush)
+ }
+
+ if (before_cluster0_load_rows < after_cluster0_load_rows) {
+ assertTrue(before_cluster0_flush < after_cluster0_flush)
+ }
+
+ if (before_cluster1_load_rows == after_cluster1_load_rows) {
+ assertTrue(before_cluster1_flush == after_cluster1_flush)
+ }
+
+ if (before_cluster1_load_rows < after_cluster1_load_rows) {
+ assertTrue(before_cluster1_flush < after_cluster1_flush)
+ }
+ */
+
+ // case3 default cluster
+ sql "SET PROPERTY 'default_cloud_cluster' = 'stream_load_2pc_cluster_name0'"
+ tableName3 = "test_all_2pc_default_cluster"
+ sql """ drop table if exists ${tableName3} """
+
+ sql """
+ CREATE TABLE IF NOT EXISTS ${tableName3} (
+ `k1` int(11) NULL,
+ `k2` tinyint(4) NULL,
+ `k3` smallint(6) NULL,
+ `k4` bigint(20) NULL,
+ `k5` largeint(40) NULL,
+ `k6` float NULL,
+ `k7` double NULL,
+ `k8` decimal(9, 0) NULL,
+ `k9` char(10) NULL,
+ `k10` varchar(1024) NULL,
+ `k11` text NULL,
+ `k12` date NULL,
+ `k13` datetime NULL
+ ) ENGINE=OLAP
+ DISTRIBUTED BY HASH(`k1`) BUCKETS 3
+ ;
+ """
+
+ sql """ set enable_profile = true """
+
+ before_cluster0_load_rows = get_be_metric(ipList[0], httpPortList[0], "load_rows");
+ log.info("before_cluster0_load_rows : ${before_cluster0_load_rows}".toString())
+ before_cluster0_flush = get_be_metric(ipList[0], httpPortList[0], "memtable_flush_total");
+ log.info("before_cluster0_flush : ${before_cluster0_flush}".toString())
+
+ before_cluster1_load_rows = get_be_metric(ipList[1], httpPortList[1], "load_rows");
+ log.info("before_cluster1_load_rows : ${before_cluster1_load_rows}".toString())
+ before_cluster1_flush = get_be_metric(ipList[1], httpPortList[1], "memtable_flush_total");
+ log.info("before_cluster1_flush : ${before_cluster1_flush}".toString())
+
+ txnId = -1;
+ // 2pc precommit
+ streamLoad {
+ table "${tableName3}"
+
+ set 'column_separator', ','
+ set 'two_phase_commit', 'true'
+
+ file 'all_types.csv'
+ time 10000 // limit inflight 10s
+
+ check { loadResult, exception, startTime, endTime ->
+ if (exception != null) {
+ throw exception
+ }
+ log.info("Stream load result: ${loadResult}".toString())
+ def json = parseJson(loadResult)
+ assertEquals("success", json.Status.toLowerCase())
+ assertEquals(20, json.NumberTotalRows)
+ assertEquals(0, json.NumberFilteredRows)
+ txnId = json.TxnId
+ }
+ }
+ sql "sync"
+ order_qt_all11 "SELECT count(*) FROM ${tableName3}" // 0
+ order_qt_all12 "SELECT count(*) FROM ${tableName3} where k1 <= 10" // 0
+
+ // 2pc commit
+ streamLoad {
+ table "${tableName3}"
+
+ set 'column_separator', ','
+ set 'txn_id', txnId.toString()
+ set 'txn_operation', 'commit'
+
+ file 'all_types.csv'
+ twoPhaseCommit true
+
+ check { loadResult, exception, startTime, endTime ->
+ if (exception != null) {
+ throw exception
+ }
+ log.info("Stream load 2pc commit result: ${loadResult}".toString())
+ def json = parseJson(loadResult)
+ assertEquals("success", json.status.toLowerCase())
+ }
+ }
+
+ order_qt_all11 "SELECT count(*) FROM ${tableName3}" // 20
+ order_qt_all12 "SELECT count(*) FROM ${tableName3} where k1 <= 10" // 11
+
+ after_cluster0_load_rows = get_be_metric(ipList[0], httpPortList[0], "load_rows");
+ log.info("after_cluster0_load_rows : ${after_cluster0_load_rows}".toString())
+ after_cluster0_flush = get_be_metric(ipList[0], httpPortList[0], "memtable_flush_total");
+ log.info("after_cluster0_flush : ${after_cluster0_flush}".toString())
+
+ after_cluster1_load_rows = get_be_metric(ipList[1], httpPortList[1], "load_rows");
+ log.info("after_cluster1_load_rows : ${after_cluster1_load_rows}".toString())
+ after_cluster1_flush = get_be_metric(ipList[1], httpPortList[1], "memtable_flush_total");
+ log.info("after_cluster1_flush : ${after_cluster1_flush}".toString())
+
+ assertTrue(before_cluster0_load_rows < after_cluster0_load_rows)
+ assertTrue(before_cluster0_flush < after_cluster0_flush)
+
+ assertTrue(before_cluster1_load_rows == after_cluster1_load_rows)
+ assertTrue(before_cluster1_flush == after_cluster1_flush)
+ sql """ drop table if exists ${tableName3} """
+}
+
diff --git a/regression-test/suites/cloud_p0/multi_cluster/stream_load/stream_load_lb.groovy b/regression-test/suites/cloud_p0/multi_cluster/stream_load/stream_load_lb.groovy
new file mode 100644
index 00000000000..4fe90c76264
--- /dev/null
+++ b/regression-test/suites/cloud_p0/multi_cluster/stream_load/stream_load_lb.groovy
@@ -0,0 +1,227 @@
+// 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 groovy.json.JsonOutput
+
+suite("stream_load_lb") {
+ def tableName3 = "test_all_lb"
+
+ try {
+ List<String> ipList = new ArrayList<>()
+ List<String> hbPortList = new ArrayList<>()
+ List<String> httpPortList = new ArrayList<>()
+ List<String> beUniqueIdList = new ArrayList<>()
+
+ String[] bes = context.config.multiClusterBes.split(',');
+ println("the value is " + context.config.multiClusterBes);
+ for(String values : bes) {
+ println("the value is " + values);
+ String[] beInfo = values.split(':');
+ ipList.add(beInfo[0]);
+ hbPortList.add(beInfo[1]);
+ httpPortList.add(beInfo[2]);
+ beUniqueIdList.add(beInfo[3]);
+ }
+
+ println("the ip is " + ipList);
+ println("the heartbeat port is " + hbPortList);
+ println("the http port is " + httpPortList);
+ println("the be unique id is " + beUniqueIdList);
+
+ for (unique_id : beUniqueIdList) {
+ resp = get_cluster.call(unique_id);
+ for (cluster : resp) {
+ if (cluster.type == "COMPUTE") {
+ drop_cluster.call(cluster.cluster_name, cluster.cluster_id);
+ }
+ }
+ }
+ sleep(20000)
+
+ List<List<Object>> result = sql "show clusters"
+ assertTrue(result.size() == 0);
+
+ addClusterLb(beUniqueIdList[0], ipList[0], hbPortList[0],
+ "stream_load_cluster_name0", "stream_load_cluster_id0",
+ ipList[0] + ":" + httpPortList[0], ":" + httpPortList[1]);
+
+ addClusterLb(beUniqueIdList[1], ipList[1], hbPortList[1],
+ "stream_load_cluster_name1", "stream_load_cluster_id1",
+ ipList[0] + ":" + httpPortList[0], ":" + httpPortList[1]);
+ sleep(20000)
+
+ result = sql "show clusters"
+ assertTrue(result.size() == 2);
+
+ for (row : result) {
+ println row
+ }
+
+ sql "SET PROPERTY 'default_cloud_cluster' = ''"
+ sql """ use @stream_load_cluster_name0 """
+
+ sql """ drop table if exists ${tableName3} """
+
+ sql """
+ CREATE TABLE IF NOT EXISTS ${tableName3} (
+ `k1` int(11) NULL,
+ `k2` tinyint(4) NULL,
+ `k3` smallint(6) NULL,
+ `k4` bigint(20) NULL,
+ `k5` largeint(40) NULL,
+ `k6` float NULL,
+ `k7` double NULL,
+ `k8` decimal(9, 0) NULL,
+ `k9` char(10) NULL,
+ `k10` varchar(1024) NULL,
+ `k11` text NULL,
+ `k12` date NULL,
+ `k13` datetime NULL
+ ) ENGINE=OLAP
+ DISTRIBUTED BY HASH(`k1`) BUCKETS 3
+ ;
+ """
+
+ sql """ set enable_profile = true """
+
+ // case1 cloud public endpoint
+ long txnId = -1;
+ streamLoad {
+ table "${tableName3}"
+
+ set 'column_separator', ','
+ set 'cloud_cluster', 'stream_load_cluster_name1'
+ set 'Host', "${ipList[0]}"
+
+ file 'all_types.csv'
+ time 10000 // limit inflight 10s
+
+ check { loadResult, exception, startTime, endTime ->
+ if (exception != null) {
+ throw exception
+ }
+ log.info("Stream load result: ${loadResult}".toString())
+ def json = parseJson(loadResult)
+ assertEquals("success", json.Status.toLowerCase())
+ assertEquals(20, json.NumberTotalRows)
+ assertEquals(0, json.NumberFilteredRows)
+ txnId = json.TxnId
+ }
+ }
+
+ order_qt_q1 "SELECT count(*) FROM ${tableName3}" // 20
+ order_qt_q2 "SELECT count(*) FROM ${tableName3} where k1 <= 10" // 11
+
+ // case2 cloud private endpoint
+ streamLoad {
+ table "${tableName3}"
+
+ set 'column_separator', ','
+ set 'cloud_cluster', 'stream_load_cluster_name1'
+ set 'Host', "${ipList[1]}"
+
+ file 'all_types.csv'
+ time 10000 // limit inflight 10s
+
+ check { loadResult, exception, startTime, endTime ->
+ if (exception != null) {
+ throw exception
+ assertEquals
+ }
+ log.info("Stream load result: ${loadResult}".toString())
+ def json = parseJson(loadResult)
+ assertEquals("success", json.Status.toLowerCase())
+ assertEquals(20, json.NumberTotalRows)
+ assertEquals(0, json.NumberFilteredRows)
+ txnId = json.TxnId
+ }
+ }
+
+ order_qt_q3 "SELECT count(*) FROM ${tableName3}" // 20
+ order_qt_q4 "SELECT count(*) FROM ${tableName3} where k1 <= 10" // 11
+
+ updateClusterEndpoint("stream_load_cluster_id0", "xx:xx", "xx:xx")
+
+ updateClusterEndpoint("stream_load_cluster_id1", ipList[1] + ":" + httpPortList[1],
+ ipList[0] + ":" + httpPortList[0])
+
+ sleep(30000)
+
+ try {
+ setFeConfig('apsaradb_env_enabled', true)
+
+ // case4 apsaradb public endpoint
+ streamLoad {
+ table "${tableName3}"
+
+ set 'column_separator', ','
+ set 'cloud_cluster', 'stream_load_cluster_name1'
+ set 'Host', 'xxx-public-xxx'
+
+ file 'all_types.csv'
+ time 10000 // limit inflight 10s
+
+ check { loadResult, exception, startTime, endTime ->
+ if (exception != null) {
+ throw exception
+ }
+ log.info("Stream load result: ${loadResult}".toString())
+ def json = parseJson(loadResult)
+ assertEquals("success", json.Status.toLowerCase())
+ assertEquals(20, json.NumberTotalRows)
+ assertEquals(0, json.NumberFilteredRows)
+ txnId = json.TxnId
+ }
+ }
+
+ order_qt_q7 "SELECT count(*) FROM ${tableName3}" // 20
+ order_qt_q8 "SELECT count(*) FROM ${tableName3} where k1 <= 10" // 11
+
+ // case5 apsaradb private endpoint
+ streamLoad {
+ table "${tableName3}"
+
+ set 'column_separator', ','
+ set 'cloud_cluster', 'stream_load_cluster_name1'
+ set 'Host', 'xxxx-xxxxx'
+
+ file 'all_types.csv'
+ time 10000 // limit inflight 10s
+
+ check { loadResult, exception, startTime, endTime ->
+ if (exception != null) {
+ throw exception
+ }
+ log.info("Stream load result: ${loadResult}".toString())
+ def json = parseJson(loadResult)
+ assertEquals("success", json.Status.toLowerCase())
+ assertEquals(20, json.NumberTotalRows)
+ assertEquals(0, json.NumberFilteredRows)
+ txnId = json.TxnId
+ }
+ }
+
+ order_qt_q9 "SELECT count(*) FROM ${tableName3}" // 20
+ order_qt_q10 "SELECT count(*) FROM ${tableName3} where k1 <= 10" // 11
+ } finally {
+ setFeConfig('apsaradb_env_enabled', false)
+ }
+ } finally {
+ sql """ drop table if exists ${tableName3} """
+ }
+}
+
diff --git a/regression-test/suites/cloud_p0/multi_cluster/table_rebalance.groovy b/regression-test/suites/cloud_p0/multi_cluster/table_rebalance.groovy
new file mode 100644
index 00000000000..667f9235319
--- /dev/null
+++ b/regression-test/suites/cloud_p0/multi_cluster/table_rebalance.groovy
@@ -0,0 +1,175 @@
+// 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 groovy.json.JsonOutput
+
+suite("table_rebalance") {
+ def token = context.config.metaServiceToken
+ def instance_id = context.config.multiClusterInstance
+
+ List<String> ipList = new ArrayList<>()
+ List<String> hbPortList = new ArrayList<>()
+ List<String> httpPortList = new ArrayList<>()
+ List<String> beUniqueIdList = new ArrayList<>()
+
+ String[] bes = context.config.multiClusterBes.split(',');
+ println("the value is " + context.config.multiClusterBes);
+ for(String values : bes) {
+ println("the value is " + values);
+ String[] beInfo = values.split(':');
+ ipList.add(beInfo[0]);
+ hbPortList.add(beInfo[1]);
+ httpPortList.add(beInfo[2]);
+ beUniqueIdList.add(beInfo[3]);
+ }
+
+ println("the ip is " + ipList);
+ println("the heartbeat port is " + hbPortList);
+ println("the http port is " + httpPortList);
+ println("the be unique id is " + beUniqueIdList);
+
+ def testFunc = { ->
+ for (unique_id : beUniqueIdList) {
+ resp = get_cluster.call(unique_id);
+ for (cluster : resp) {
+ if (cluster.type == "COMPUTE") {
+ drop_cluster.call(cluster.cluster_name, cluster.cluster_id);
+ }
+ }
+ }
+ sleep(20000)
+
+ List<List<Object>> result = sql "show clusters"
+ assertTrue(result.size() == 0);
+
+ add_cluster.call(beUniqueIdList[0], ipList[0], hbPortList[0],
+ "regression_cluster_name0", "regression_cluster_id0");
+
+ add_node.call(beUniqueIdList[1], ipList[1], hbPortList[1],
+ "regression_cluster_name0", "regression_cluster_id0");
+ sleep(20000)
+
+ result = sql "show clusters"
+ assertTrue(result.size() == 1);
+
+ for (row : result) {
+ println row
+ }
+ sql """ drop table IF EXISTS table100 """
+ sql """ drop table IF EXISTS table_p2 """
+ sql """ use @regression_cluster_name0"""
+ sql """
+ CREATE TABLE table100 (
+ class INT,
+ id INT,
+ score INT SUM
+ )
+ AGGREGATE KEY(class, id)
+ DISTRIBUTED BY HASH(class) BUCKETS 48
+
+ """
+
+ sql """
+ CREATE TABLE table_p2 ( k1 int(11) NOT NULL, k2 varchar(20) NOT NULL, k3 int sum NOT NULL )
+ AGGREGATE KEY(k1, k2)
+ PARTITION BY RANGE(k1) (
+ PARTITION p1992 VALUES [("-2147483648"), ("19930101")),
+ PARTITION p1993 VALUES [("19930101"), ("19940101")),
+ PARTITION p1994 VALUES [("19940101"), ("19950101")),
+ PARTITION p1995 VALUES [("19950101"), ("19960101")),
+ PARTITION p1996 VALUES [("19960101"), ("19970101")),
+ PARTITION p1997 VALUES [("19970101"), ("19980101")),
+ PARTITION p1998 VALUES [("19980101"), ("19990101")),
+ PARTITION p1999 VALUES [("19990101"), ("20000101")),
+ PARTITION p2000 VALUES [("20000101"), ("20010101")),
+ PARTITION p2001 VALUES [("20010101"), ("20020101")),
+ PARTITION p2002 VALUES [("20020101"), ("20030101")),
+ PARTITION p2003 VALUES [("20030101"), ("20040101")),
+ PARTITION p2004 VALUES [("20040101"), ("20050101")),
+ PARTITION p2005 VALUES [("20050101"), ("20060101")),
+ PARTITION p2006 VALUES [("20060101"), ("20070101")),
+ PARTITION p2007 VALUES [("20070101"), ("20080101")),
+ PARTITION p2008 VALUES [("20080101"), ("20090101")),
+ PARTITION p2009 VALUES [("20090101"), ("20100101")),
+ PARTITION p2010 VALUES [("20100101"), ("20110101")),
+ PARTITION p2011 VALUES [("20110101"), ("20120101")),
+ PARTITION p2012 VALUES [("20120101"), ("20130101")),
+ PARTITION p2013 VALUES [("20130101"), ("30000000")))
+ DISTRIBUTED BY HASH(k1) BUCKETS 2
+ """
+
+ for (int i = 0; i < 100; ++i) {
+ int num = 19930201 + i * 3000
+ sql """
+ insert into table_p2 values ($num, 'a', 1);
+ """
+ }
+
+ sleep(20000)
+
+ add_node.call(beUniqueIdList[2], ipList[2], hbPortList[2],
+ "regression_cluster_name0", "regression_cluster_id0");
+ sleep(240000)
+
+ sql """ use @regression_cluster_name0 """
+
+ result = sql """ ADMIN SHOW REPLICA DISTRIBUTION FROM table100; """
+ assertEquals(result.size(), 3);
+
+ for (row : result) {
+ log.info("replica distribution: ${row} ".toString())
+ assertTrue(Integer.valueOf((String) row[1]) <= 17 && Integer.valueOf((String) row[1]) >= 15)
+ }
+
+ for (int i = 1992; i < 2014; ++i) {
+ result = sql """ ADMIN SHOW REPLICA DISTRIBUTION FROM table_p2 PARTITION(p$i) """
+ assertEquals(result.size(), 3);
+ for (row : result) {
+ log.info("replica distribution: ${row} ".toString())
+ assertTrue(Integer.valueOf((String) row[1]) <= 1 && Integer.valueOf((String) row[1]) >= 0)
+ }
+ }
+
+ result = sql """ ADMIN SHOW REPLICA DISTRIBUTION FROM table_p2 """
+ assertEquals(result.size(), 3);
+ for (row : result) {
+ log.info("replica distribution: ${row} ".toString())
+ assertTrue(Integer.valueOf((String) row[1]) <= 16 && Integer.valueOf((String) row[1]) >= 14)
+ }
+
+ }
+
+ def balance_tablet_percent_per_run = 0.5
+ for (def enable_global_balance in [false, true]) {
+ for (def preheating_enabled in [false, true]) {
+ log.info("test table rebalance with balance_tablet_percent_per_run=${balance_tablet_percent_per_run}, "
+ + "enable_global_balance=${enable_global_balance}, preheating_enabled=${preheating_enabled}")
+ try {
+ setFeConfig('balance_tablet_percent_per_run', balance_tablet_percent_per_run)
+ setFeConfig('enable_global_balance', enable_global_balance)
+ setFeConfig('preheating_enabled', preheating_enabled)
+ testFunc.call()
+ } finally {
+ setFeConfig('balance_tablet_percent_per_run', 0.05)
+ setFeConfig('enable_global_balance', true)
+ setFeConfig('preheating_enabled', true)
+ }
+ }
+ }
+ sql """ drop table IF EXISTS table100 """
+ sql """ drop table IF EXISTS table_p2 """
+}
diff --git a/regression-test/suites/cloud_p0/multi_cluster/test_apsaradb_internal_stage.groovy b/regression-test/suites/cloud_p0/multi_cluster/test_apsaradb_internal_stage.groovy
new file mode 100644
index 00000000000..e5ed1a752a9
--- /dev/null
+++ b/regression-test/suites/cloud_p0/multi_cluster/test_apsaradb_internal_stage.groovy
@@ -0,0 +1,199 @@
+// 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.codehaus.groovy.runtime.IOGroovyMethods
+
+suite("test_apsarad_internal_stage_copy_into") {
+ // Internal and external stage cross use
+ def tableNamExternal = "customer_internal_stage"
+ //def token = "greedisgood9999"
+ def instanceId = context.config.instanceId
+ def cloudUniqueId = context.config.cloudUniqueId
+ def token = context.config.metaServiceToken
+ def instance_id = context.config.multiClusterInstance
+
+ List<String> ipList = new ArrayList<>()
+ List<String> hbPortList = new ArrayList<>()
+ List<String> httpPortList = new ArrayList<>()
+ List<String> beUniqueIdList = new ArrayList<>()
+
+ String[] bes = context.config.multiClusterBes.split(',');
+ println("the value is " + context.config.multiClusterBes);
+ for(String values : bes) {
+ println("the value is " + values);
+ String[] beInfo = values.split(':');
+ ipList.add(beInfo[0]);
+ hbPortList.add(beInfo[1]);
+ httpPortList.add(beInfo[2]);
+ beUniqueIdList.add(beInfo[3]);
+ }
+
+ println("the ip is " + ipList);
+ println("the heartbeat port is " + hbPortList);
+ println("the http port is " + httpPortList);
+ println("the be unique id is " + beUniqueIdList);
+
+ for (unique_id : beUniqueIdList) {
+ resp = get_cluster.call(unique_id);
+ for (cluster : resp) {
+ if (cluster.type == "COMPUTE") {
+ drop_cluster.call(cluster.cluster_name, cluster.cluster_id);
+ }
+ }
+ }
+ sleep(20000)
+
+ List<List<Object>> showResult = sql "show clusters"
+ assertTrue(showResult.size() == 0);
+
+ add_cluster.call(beUniqueIdList[0], ipList[0], hbPortList[0],
+ "regression_cluster_name0", "regression_cluster_id0");
+ sleep(20000)
+
+ showResult = sql "show clusters"
+ assertTrue(showResult.size() == 1);
+
+ for (row : showResult) {
+ println row
+ }
+
+ try {
+ sql """
+ CREATE TABLE IF NOT EXISTS ${tableNamExternal} (
+ C_CUSTKEY INTEGER NOT NULL,
+ C_NAME VARCHAR(25) NOT NULL,
+ C_ADDRESS VARCHAR(40) NOT NULL,
+ C_NATIONKEY INTEGER NOT NULL,
+ C_PHONE CHAR(15) NOT NULL,
+ C_ACCTBAL DECIMAL(15,2) NOT NULL,
+ C_MKTSEGMENT CHAR(10) NOT NULL,
+ C_COMMENT VARCHAR(117) NOT NULL
+ )
+ UNIQUE KEY(C_CUSTKEY)
+ DISTRIBUTED BY HASH(C_CUSTKEY) BUCKETS 1
+ """
+ sql """ DROP TABLE IF EXISTS ${tableNamExternal}; """
+ } finally {
+ //try_sql("DROP TABLE IF EXISTS ${tableNamExternal}")
+ }
+
+ def tableName = "customer_apsaradb_internal_stage"
+
+ def uploadFile = { remoteFilePath, localFilePath ->
+ StringBuilder strBuilder = new StringBuilder()
+ strBuilder.append("""curl -u """ + context.config.feCloudHttpUser + ":" + context.config.feCloudHttpPassword)
+ strBuilder.append(""" -H fileName:""" + remoteFilePath)
+ strBuilder.append(""" -H host:""" + "private")
+ strBuilder.append(""" -T """ + localFilePath)
+ def feHttpAddress = context.config.isDorisEnv ? context.config.feHttpAddress : context.config.feCloudHttpAddress
+ strBuilder.append(""" -L http://""" + feHttpAddress + """/copy/upload""")
+
+ String command = strBuilder.toString()
+ logger.info("upload command=" + command)
+ def process = command.toString().execute()
+ def code = process.waitFor()
+ def err = IOGroovyMethods.getText(new BufferedReader(new InputStreamReader(process.getErrorStream())));
+ def out = process.getText()
+ logger.info("Request FE Config: code=" + code + ", out=" + out + ", err=" + err)
+ assertEquals(code, 0)
+ }
+
+ def createTable = {
+ sql """ DROP TABLE IF EXISTS ${tableName}; """
+ sql """
+ CREATE TABLE IF NOT EXISTS ${tableName} (
+ C_CUSTKEY INTEGER NOT NULL,
+ C_NAME VARCHAR(25) NOT NULL,
+ C_ADDRESS VARCHAR(40) NOT NULL,
+ C_NATIONKEY INTEGER NOT NULL,
+ C_PHONE CHAR(15) NOT NULL,
+ C_ACCTBAL DECIMAL(15,2) NOT NULL,
+ C_MKTSEGMENT CHAR(10) NOT NULL,
+ C_COMMENT VARCHAR(117) NOT NULL
+ )
+ DUPLICATE KEY(C_CUSTKEY)
+ DISTRIBUTED BY HASH(C_CUSTKEY) BUCKETS 1
+ """
+ }
+
+ def waitInternalStageFilesDeleted = { fileName ->
+ def retry = 10
+ do {
+ Thread.sleep(2000)
+ if (checkRecycleInternalStage(token, instanceId, cloudUniqueId, fileName)) {
+ Thread.sleep(2000) // wait for copy job kv is deleted
+ return
+ }
+ } while (retry--)
+ assertTrue(false, "Internal stage file is not deleted")
+ }
+
+ def getCloudConf = {
+ configResult = sql """ ADMIN SHOW FRONTEND CONFIG """
+ for (def r : configResult) {
+ assertTrue(r.size() > 2)
+ if (r[0] == "cloud_delete_loaded_internal_stage_files") {
+ return (r[1] == "true")
+ }
+ }
+ return false
+ }
+
+ boolean cloud_delete_loaded_internal_stage_files = getCloudConf()
+ logger.info("cloud_delete_loaded_internal_stage_files=" + cloud_delete_loaded_internal_stage_files)
+
+ try {
+ setFeConfig('apsaradb_env_enabled', true)
+ def fileName = "internal_customer.csv"
+ def filePath = "${context.config.dataPath}/cloud/copy_into/" + fileName
+ def remoteFileName = fileName + "test_apsaradb_internal_stage"
+ uploadFile(remoteFileName, filePath)
+
+ createTable()
+ def result = sql " copy into ${tableName} from @~('${remoteFileName}') properties ('file.type' = 'csv', 'file.column_separator' = '|', 'copy.async' = 'false'); "
+ logger.info("copy result: " + result)
+ assertTrue(result.size() == 1)
+ assertTrue(result[0].size() == 8)
+ assertTrue(result[0][1].equals("FINISHED"), "Finish copy into, state=" + result[0][1] + ", expected state=FINISHED")
+ qt_sql " SELECT COUNT(*) FROM ${tableName}; "
+
+ result = sql " copy into ${tableName} from @~('${remoteFileName}') properties ('file.type' = 'csv', 'file.column_separator' = '|', 'copy.async' = 'false'); "
+ logger.info("copy result: " + result)
+ assertTrue(result.size() == 1)
+ assertTrue(result[0].size() == 8)
+ assertTrue(result[0][1].equals("CANCELLED"), "Finish copy into, state=" + result[0][1] + ", expected state=CANCELLED")
+ qt_sql " SELECT COUNT(*) FROM ${tableName}; "
+
+ if (cloud_delete_loaded_internal_stage_files) {
+ // check file is deleted
+ waitInternalStageFilesDeleted(remoteFileName)
+ // check copy job and file keys are deleted
+ uploadFile(remoteFileName, filePath)
+ result = sql " copy into ${tableName} from @~('${remoteFileName}') properties ('file.type' = 'csv', 'file.column_separator' = '|', 'copy.async' = 'false'); "
+ logger.info("copy result: " + result)
+ assertTrue(result.size() == 1)
+ assertTrue(result[0].size() == 8)
+ assertTrue(result[0][1].equals("FINISHED"), "Finish copy into, state=" + result[0][1] + ", expected state=FINISHED")
+ // check file is deleted
+ waitInternalStageFilesDeleted(remoteFileName)
+ }
+
+ } finally {
+ setFeConfig('apsaradb_env_enabled', false)
+ sql """ DROP TABLE IF EXISTS ${tableName}; """
+ }
+}
diff --git a/regression-test/suites/cloud_p0/multi_cluster/test_cloud_cluster.groovy b/regression-test/suites/cloud_p0/multi_cluster/test_cloud_cluster.groovy
new file mode 100644
index 00000000000..43f70aed242
--- /dev/null
+++ b/regression-test/suites/cloud_p0/multi_cluster/test_cloud_cluster.groovy
@@ -0,0 +1,202 @@
+// 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 groovy.json.JsonOutput
+import java.sql.*;
+
+suite("test_cloud_cluster") {
+ def token = context.config.metaServiceToken
+ def instance_id = context.config.multiClusterInstance
+
+ List<String> ipList = new ArrayList<>()
+ List<String> hbPortList = new ArrayList<>()
+ List<String> httpPortList = new ArrayList<>()
+ List<String> beUniqueIdList = new ArrayList<>()
+
+ String[] bes = context.config.multiClusterBes.split(',');
+ println("the value is " + context.config.multiClusterBes);
+ for(String values : bes) {
+ println("the value is " + values);
+ String[] beInfo = values.split(':');
+ ipList.add(beInfo[0]);
+ hbPortList.add(beInfo[1]);
+ httpPortList.add(beInfo[2]);
+ beUniqueIdList.add(beInfo[3]);
+ }
+
+ println("the ip is " + ipList);
+ println("the heartbeat port is " + hbPortList);
+ println("the http port is " + httpPortList);
+ println("the be unique id is " + beUniqueIdList);
+
+ for (unique_id : beUniqueIdList) {
+ resp = get_cluster.call(unique_id);
+ for (cluster : resp) {
+ if (cluster.type == "COMPUTE") {
+ drop_cluster.call(cluster.cluster_name, cluster.cluster_id);
+ }
+ }
+ }
+ sleep(20000)
+
+ List<List<Object>> result = sql "show clusters"
+ assertTrue(result.size() == 0);
+
+ add_cluster.call(beUniqueIdList[0], ipList[0], hbPortList[0],
+ "regression_cluster_name0", "regression_cluster_id0");
+ add_cluster.call(beUniqueIdList[1], ipList[1], hbPortList[1],
+ "regression_cluster_name1", "regression_cluster_id1");
+ sleep(20000)
+
+ result = sql "show clusters"
+ assertTrue(result.size() == 2);
+
+ sql "use @regression_cluster_name0"
+ result = sql "show clusters"
+ for (row : result) {
+ println row
+ if(row[0] == "regression_cluster_name0") {
+ assertTrue(row[1].toString().toLowerCase() == "true")
+ }
+ }
+
+ sql "use @regression_cluster_name1"
+ result = sql "show clusters"
+ for (row : result) {
+ println row
+ if(row[0] == "regression_cluster_name1") {
+ assertTrue(row[1].toString().toLowerCase() == "true")
+ }
+ }
+
+ sql "use @regression_cluster_name1"
+ sql "create database if not exists test_jdbc_url_db"
+ sql "use test_jdbc_url_db"
+ sql """
+ CREATE TABLE test_table (
+ class INT,
+ id INT,
+ score INT SUM
+ )
+ AGGREGATE KEY(class, id)
+ DISTRIBUTED BY HASH(class) BUCKETS 16
+ """
+
+ sql """
+ insert into test_table values (666888, 2333333, 10010);
+ """
+
+ String jdbcUrl = context.config.jdbcUrl
+ String[] urlElems = jdbcUrl.split("/");
+ String feEndPoint = urlElems[2]
+ String newUrl = "jdbc:mysql://" + feEndPoint + "/" + "test_jdbc_url_db@regression_cluster_name0";
+ println("newurl " + newUrl);
+
+ String user = "root";
+ String password = "";
+
+ try {
+ Connection myCon = DriverManager.getConnection(newUrl, user, password);
+ Statement stmt = myCon.createStatement();
+ ResultSet rs = stmt.executeQuery("select * from test_table");
+ while (rs.next()) {
+ assertEquals(666888, rs.getInt(1));
+ assertEquals(2333333, rs.getInt(2));
+ assertEquals(10010, rs.getInt(3));
+ }
+ } catch (SQLException e) {
+ System.out.println(e);
+ e.printStackTrace()
+ assertTrue(false);
+ }
+
+ newUrl = "jdbc:mysql://" + feEndPoint + "/" + "@regression_cluster_name0";
+ println("newurl " + newUrl);
+
+ try {
+ Connection myCon = DriverManager.getConnection(newUrl, user, password);
+ Statement stmt = myCon.createStatement();
+ ResultSet rs = stmt.executeQuery("show clusters");
+ while (rs.next()) {
+ println rs.getString(1)
+ if (rs.getString(1) == "regression_cluster_name0") {
+ assertEquals(rs.getString(2).toLowerCase(), "true");
+ }
+ }
+ } catch (SQLException e) {
+ System.out.println(e);
+ e.printStackTrace()
+ assertTrue(false);
+ }
+
+ newUrl = "jdbc:mysql://" + feEndPoint + "/" + "test_jdbc_url_db";
+ println("newurl " + newUrl);
+
+ try {
+ Connection myCon = DriverManager.getConnection(newUrl, user, password);
+ Statement stmt = myCon.createStatement();
+ stmt.execute("use @regression_cluster_name1");
+ ResultSet rs = stmt.executeQuery("select * from test_table");
+ while (rs.next()) {
+ assertEquals(666888, rs.getInt(1));
+ assertEquals(2333333, rs.getInt(2));
+ assertEquals(10010, rs.getInt(3));
+ }
+ } catch (SQLException e) {
+ System.out.println(e);
+ e.printStackTrace()
+ assertTrue(false);
+ }
+
+ def executeMySQLCommand = { String command ->
+ try {
+ String line;
+ StringBuilder errMsg = new StringBuilder();
+ StringBuilder msg = new StringBuilder();
+ Process p = Runtime.getRuntime().exec(new String[]{"/bin/bash", "-c", command});
+
+ BufferedReader errInput = new BufferedReader(new InputStreamReader(p.getErrorStream()));
+ while ((line = errInput.readLine()) != null) {
+ errMsg.append(line);
+ }
+ assert errMsg.length() == 0: "error occurred!" + errMsg.toString();
+ errInput.close();
+
+ BufferedReader input = new BufferedReader(new InputStreamReader(p.getInputStream()));
+ while ((line = input.readLine()) != null) {
+ msg.append(line);
+ }
+ println "msg: " + msg.toString()
+ assert msg.toString().contains("666888"): "error occurred!" + errMsg.toString();
+ input.close();
+ } catch (IOException e) {
+ e.printStackTrace();
+ }
+ }
+
+ String jdbcUrlConfig = context.config.jdbcUrl;
+ String tempString = jdbcUrlConfig.substring(jdbcUrlConfig.indexOf("jdbc:mysql://") + 13);
+ String mysqlHost = tempString.substring(0, tempString.indexOf(":"));
+ String mysqlPort = tempString.substring(tempString.indexOf(":") + 1, tempString.indexOf("/"));
+ String cmd1 = "mysql -uroot -h" + mysqlHost + " -P" + mysqlPort + " -Dtest_jdbc_url_db@regression_cluster_name0 " + " -e \" select * from test_table \"";
+ executeMySQLCommand(cmd1);
+ String cmd2 = "mysql -uroot -h" + mysqlHost + " -P" + mysqlPort + " -D@regression_cluster_name0 " + " -e \" use test_jdbc_url_db; select * from test_table \"";
+ executeMySQLCommand(cmd2);
+ sql """
+ drop table if exists test_table
+ """
+}
diff --git a/regression-test/suites/cloud_p0/multi_cluster/test_drop_cluster.groovy b/regression-test/suites/cloud_p0/multi_cluster/test_drop_cluster.groovy
new file mode 100644
index 00000000000..f2310888154
--- /dev/null
+++ b/regression-test/suites/cloud_p0/multi_cluster/test_drop_cluster.groovy
@@ -0,0 +1,96 @@
+// 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 groovy.json.JsonOutput
+
+suite("test_drop_cluster") {
+ def token = context.config.metaServiceToken
+ def instance_id = context.config.multiClusterInstance
+
+ List<String> ipList = new ArrayList<>()
+ List<String> hbPortList = new ArrayList<>()
+ List<String> httpPortList = new ArrayList<>()
+ List<String> beUniqueIdList = new ArrayList<>()
+
+ String[] bes = context.config.multiClusterBes.split(',');
+ println("the value is " + context.config.multiClusterBes);
+ for(String values : bes) {
+ println("the value is " + values);
+ String[] beInfo = values.split(':');
+ ipList.add(beInfo[0]);
+ hbPortList.add(beInfo[1]);
+ httpPortList.add(beInfo[2]);
+ beUniqueIdList.add(beInfo[3]);
+ }
+
+ println("the ip is " + ipList);
+ println("the heartbeat port is " + hbPortList);
+ println("the http port is " + httpPortList);
+ println("the be unique id is " + beUniqueIdList);
+
+ for (unique_id : beUniqueIdList) {
+ resp = get_cluster.call(unique_id);
+ for (cluster : resp) {
+ if (cluster.type == "COMPUTE") {
+ drop_cluster.call(cluster.cluster_name, cluster.cluster_id);
+ }
+ }
+ }
+ wait_cluster_change()
+
+ List<List<Object>> result = sql "show clusters"
+ assertTrue(result.size() == 0);
+
+ add_cluster.call(beUniqueIdList[0], ipList[0], hbPortList[0],
+ "regression_cluster_name0", "regression_cluster_id0");
+ add_cluster.call(beUniqueIdList[1], ipList[1], hbPortList[1],
+ "regression_cluster_name1", "regression_cluster_id1");
+ wait_cluster_change()
+
+ result = sql "show clusters"
+ assertTrue(result.size() == 2);
+
+ for (i in 1..10) {
+ drop_cluster.call("regression_cluster_name0", "regression_cluster_id0");
+ drop_cluster.call("regression_cluster_name1", "regression_cluster_id1");
+ add_cluster.call(beUniqueIdList[0], ipList[0], hbPortList[0],
+ "regression_cluster_name0", "regression_cluster_id0");
+ add_cluster.call(beUniqueIdList[1], ipList[1], hbPortList[1],
+ "regression_cluster_name1", "regression_cluster_id1");
+ }
+ wait_cluster_change()
+ result = sql "show clusters"
+ assertTrue(result.size() == 2);
+
+ sql "use @regression_cluster_name0"
+ result = sql "show clusters"
+ for (row : result) {
+ println row
+ if(row[0] == "regression_cluster_name0") {
+ assertTrue(row[1].toString().toLowerCase() == "true")
+ }
+ }
+
+ sql "use @regression_cluster_name1"
+ result = sql "show clusters"
+ for (row : result) {
+ println row
+ if(row[0] == "regression_cluster_name1") {
+ assertTrue(row[1].toString().toLowerCase() == "true")
+ }
+ }
+}
diff --git a/regression-test/suites/cloud_p0/multi_cluster/test_drop_userid.groovy b/regression-test/suites/cloud_p0/multi_cluster/test_drop_userid.groovy
new file mode 100644
index 00000000000..33327c59ad3
--- /dev/null
+++ b/regression-test/suites/cloud_p0/multi_cluster/test_drop_userid.groovy
@@ -0,0 +1,121 @@
+// 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 groovy.json.JsonOutput
+import org.codehaus.groovy.runtime.IOGroovyMethods
+
+suite("test_drop_userid", "cloud_auth") {
+ def user1 = "selectdb_cloud_test_test_drop_userid"
+ sql """drop user if exists ${user1}"""
+ try {
+ // create user
+ sql """create user ${user1} IDENTIFIED BY 'A12345678_';"""
+ // copy into upload, ms record userid in instancePb
+ StringBuilder strBuilder = new StringBuilder()
+ strBuilder.append("""curl -XPUT -u ${user1}:A12345678_""")
+ strBuilder.append(""" -H fileName:1.dat""")
+ strBuilder.append(""" -d '1,1,2,"yyy"' """)
+ def feHttpAddress = context.config.isDorisEnv ? context.config.feHttpAddress : context.config.feCloudHttpAddress
+ strBuilder.append(""" -L http://""" + feHttpAddress + """/copy/upload""")
+ String command = strBuilder.toString()
+ logger.info("upload command=" + command)
+ def process = command.toString().execute()
+ def code = process.waitFor()
+ def err = IOGroovyMethods.getText(new BufferedReader(new InputStreamReader(process.getErrorStream())));
+ def out = process.getText()
+ logger.info("Request FE Config: code=" + code + ", out=" + out + ", err=" + err)
+ assertEquals(code, 0)
+
+ Thread.sleep(2000)
+ // get instance from ms can't get userid
+ def instance_id = context.config.multiClusterInstance
+ def ms_token = context.config.metaServiceToken
+ def get_instance = { requestBody, checkFunc ->
+ httpTest {
+ endpoint context.config.metaServiceHttpAddress
+ uri "/MetaService/http/get_instance?token=${ms_token}&instance_id=${instance_id}"
+ body requestBody
+ check checkFunc
+ }
+ }
+ def jsonOutput = new JsonOutput()
+ def requestBody = jsonOutput.toJson([])
+ def instance_info = null
+ get_instance.call(requestBody) {
+ respCode, body ->
+ log.info("http cli result: ${body} ${respCode}".toString())
+ instance_info = parseJson(body)
+ assertTrue(instance_info.code.equalsIgnoreCase("OK"))
+ }
+ def stages = instance_info.result.stages
+ if(stages != null) {
+ boolean found = false
+ for (def stage : stages) {
+ log.info("stage info ${stage}".toString())
+ // external stage doesn't contain mysql_user_name
+ if ("EXTERNAL".equals(stage.type)) {
+ log.info("external stage skip".toString())
+ continue;
+ }
+ String name = stage.mysql_user_name[0]
+ if(name.equals("${user1}".toString())) {
+ found = true
+ def userid = stage.mysql_user_id
+ assertTrue(userid != null, "before drop can find ${user1} stage in ms")
+ break
+ }
+ }
+ assertTrue(found, "before drop can find ${user1} stage in ms")
+ }
+
+ // drop user
+ sql """drop user ${user1}"""
+
+ Thread.sleep(2000)
+ // get instance from ms can't get userid
+ jsonOutput = new JsonOutput()
+ requestBody = jsonOutput.toJson([])
+ instance_info = null
+ get_instance.call(requestBody) {
+ respCode, body ->
+ log.info("http cli result: ${body} ${respCode}".toString())
+ instance_info = parseJson(body)
+ assertTrue(instance_info.code.equalsIgnoreCase("OK"))
+ }
+ stages = instance_info.result.stages
+ if(stages != null) {
+ boolean found = false
+ for (stage : stages) {
+ log.info("stage info ${stage}".toString())
+ // external stage doesn't contain mysql_user_name
+ if ("EXTERNAL".equals(stage.type)) {
+ log.info("external stage skip".toString())
+ continue;
+ }
+ String name = stage.mysql_user_name
+ if(name.equals("${user1}".toString())) {
+ found = true
+ break
+ }
+ }
+ assertFalse(found, "after drop cant find ${user1} stage in ms")
+ }
+ } finally {
+ sql """drop user if exists ${user1}"""
+ }
+}
+
diff --git a/regression-test/suites/cloud_p0/multi_cluster/test_group_commit_multi_cluster.groovy b/regression-test/suites/cloud_p0/multi_cluster/test_group_commit_multi_cluster.groovy
new file mode 100644
index 00000000000..1d8b936d70a
--- /dev/null
+++ b/regression-test/suites/cloud_p0/multi_cluster/test_group_commit_multi_cluster.groovy
@@ -0,0 +1,129 @@
+// 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 groovy.json.JsonOutput
+
+suite("test_group_commit_multi_cluster") {
+ def token = context.config.metaServiceToken
+ def instance_id = context.config.multiClusterInstance
+
+ List<String> ipList = new ArrayList<>()
+ List<String> hbPortList = new ArrayList<>()
+ List<String> httpPortList = new ArrayList<>()
+ List<String> beUniqueIdList = new ArrayList<>()
+
+ String[] bes = context.config.multiClusterBes.split(',');
+ println("the value is " + context.config.multiClusterBes);
+ for(String values : bes) {
+ println("the value is " + values);
+ String[] beInfo = values.split(':');
+ ipList.add(beInfo[0]);
+ hbPortList.add(beInfo[1]);
+ httpPortList.add(beInfo[2]);
+ beUniqueIdList.add(beInfo[3]);
+ }
+
+ println("the ip is " + ipList);
+ println("the heartbeat port is " + hbPortList);
+ println("the http port is " + httpPortList);
+ println("the be unique id is " + beUniqueIdList);
+
+ for (unique_id : beUniqueIdList) {
+ resp = get_cluster.call(unique_id);
+ for (cluster : resp) {
+ if (cluster.type == "COMPUTE") {
+ drop_cluster.call(cluster.cluster_name, cluster.cluster_id);
+ }
+ }
+ }
+ wait_cluster_change()
+
+ List<List<Object>> result = sql "show clusters"
+ assertTrue(result.size() == 0);
+
+ add_cluster.call(beUniqueIdList[0], ipList[0], hbPortList[0],
+ "regression_cluster_name0", "regression_cluster_id0");
+ wait_cluster_change()
+
+ result = sql "show clusters"
+ assertTrue(result.size() == 1);
+
+ sql "use @regression_cluster_name0"
+ sql """ drop table IF EXISTS table_p2 """
+ sql """
+ CREATE TABLE table_p2 ( k1 int(11) NOT NULL, k2 varchar(20) NOT NULL, k3 int sum NOT NULL )
+ UNIQUE KEY(k1, k2)
+ DISTRIBUTED BY HASH(k1) BUCKETS 3
+ """
+
+ sql """
+ set group_commit=sync_mode;
+ """
+
+ sql """
+ insert into table_p2 values(1, 'abc', 10);
+ """
+
+ sql """
+ select * from table_p2;
+ """
+
+ rename_cloud_cluster.call("regression_cluster_name1", "regression_cluster_id0");
+ wait_cluster_change()
+
+ // create same name cluster
+ add_cluster.call(beUniqueIdList[1], ipList[1], hbPortList[1],
+ "regression_cluster_name0", "regression_cluster_id2");
+ wait_cluster_change()
+
+ result = sql "show clusters"
+ log.info("clusters: " + result)
+ log.info("backends:")
+ for (def be : sql("show backends")) {
+ log.info("be: " + be)
+ }
+ assertEquals(2, result.size())
+
+ sql "use @regression_cluster_name0"
+
+ sql """
+ set group_commit=sync_mode;
+ """
+
+ sql """
+ insert into table_p2 values(2, 'def', 20);
+ """
+
+ sql """
+ select * from table_p2;
+ """
+
+ sql "use @regression_cluster_name1"
+
+ sql """
+ set group_commit=sync_mode;
+ """
+
+ sql """
+ insert into table_p2 values(3, 'gh', 30);
+ """
+
+ sql """
+ select * from table_p2;
+ """
+ sql """ drop table IF EXISTS table_p2 """
+}
diff --git a/regression-test/suites/cloud_p0/multi_cluster/test_overdue_instance.groovy b/regression-test/suites/cloud_p0/multi_cluster/test_overdue_instance.groovy
new file mode 100644
index 00000000000..3a50afe0509
--- /dev/null
+++ b/regression-test/suites/cloud_p0/multi_cluster/test_overdue_instance.groovy
@@ -0,0 +1,243 @@
+// 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 groovy.json.JsonOutput
+import groovy.json.JsonSlurper
+
+/**
+* @Params url is "/xxx", data is request body
+* @Return response body
+*/
+def http_post(addr, url, data = null, user = null, password = null) {
+ // def dst = "http://"+ context.config.feHttpAddress
+ def dst = "http://"+ addr
+ def conn = new URL(dst + url).openConnection()
+ conn.setRequestMethod("POST")
+ conn.setRequestProperty("Content-Type", "application/json")
+ logger.info("go into http_post")
+ if (user) {
+ def authString = user + ":" + password
+ def authEncoded = Base64.getEncoder().encodeToString(authString.getBytes())
+ conn.setRequestProperty('Authorization', "Basic ${authEncoded}")
+ } else {
+ conn.setRequestProperty("Authorization", "Basic cm9vdDo=")
+ }
+ if (data) {
+ conn.doOutput = true
+ def writer = new OutputStreamWriter(conn.outputStream)
+ writer.write(data)
+ writer.flush()
+ writer.close()
+ }
+ return conn.content.text
+}
+
+class Stmt {
+ String stmt
+}
+
+class AlterRequest {
+ String instance_id;
+ String op;
+}
+
+class AlterRequestWithoutOp {
+ String instance_id
+}
+
+suite('test_overdue') {
+ def token = context.config.metaServiceToken
+ def instance_id = context.config.multiClusterInstance
+
+ List<String> ipList = new ArrayList<>()
+ List<String> hbPortList = new ArrayList<>()
+ List<String> httpPortList = new ArrayList<>()
+ List<String> beUniqueIdList = new ArrayList<>()
+
+ String[] bes = context.config.multiClusterBes.split(',');
+ println("the value is " + context.config.multiClusterBes);
+ for(String values : bes) {
+ println("the value is " + values);
+ String[] beInfo = values.split(':');
+ ipList.add(beInfo[0]);
+ hbPortList.add(beInfo[1]);
+ httpPortList.add(beInfo[2]);
+ beUniqueIdList.add(beInfo[3]);
+ }
+
+ println("the ip is " + ipList);
+ println("the heartbeat port is " + hbPortList);
+ println("the http port is " + httpPortList);
+ println("the be unique id is " + beUniqueIdList);
+
+ for (unique_id : beUniqueIdList) {
+ resp = get_cluster.call(unique_id);
+ for (cluster : resp) {
+ if (cluster.type == "COMPUTE") {
+ drop_cluster.call(cluster.cluster_name, cluster.cluster_id);
+ }
+ }
+ }
+ sleep(20000)
+
+ List<List<Object>> result0 = sql "show clusters"
+ assertTrue(result0.size() == 0);
+
+ add_cluster.call(beUniqueIdList[0], ipList[0], hbPortList[0],
+ "regression_cluster_name0", "regression_cluster_id0");
+ sleep(20000)
+
+ result0 = sql "show clusters"
+ assertTrue(result0.size() == 1);
+
+ for (row : result0) {
+ println row
+ }
+
+ // add_node.call(beUniqueIdList[1], ipList[1], hbPortList[1],
+ // "regression_cluster_name0", "regression_cluster_id0");
+
+ // sleep(20000)
+ sql """ use @regression_cluster_name0 """
+
+ result0 = sql "show clusters"
+ assertTrue(result0.size() == 1);
+
+ for (row : result0) {
+ println row
+ }
+
+ def user = 'test_overdue_instance_user'
+ def result = sql """ SELECT DATABASE(); """
+ def url0 = '/api/query/default_cluster/' + result[0][0]
+ def stmt0 = """ show databases """
+ def stmt_json = JsonOutput.toJson(new Stmt(stmt: stmt0))
+
+
+ sql """ drop user if exists ${user} """
+ sql """ create user ${user} """
+ sql """ GRANT SELECT_PRIV ON *.*.* TO ${user}@'%'; """
+
+ // set wrarehouse to normal
+ try {
+ try {
+ alter_request = JsonOutput.toJson(new AlterRequest(instance_id: instance_id, op: "SET_NORMAL"))
+ result = http_post(context.config.metaServiceHttpAddress, "/MetaService/http/set_instance_status?token=greedisgood9999", alter_request)
+ obj = new JsonSlurper().parseText(result)
+ logger.info("try to set warehouse normal, the result is {}", obj)
+ sleep(40000)
+ }
+ catch (Exception e) {
+ logger.info("the error msg is {}", e.getMessage())
+ }
+
+ // when warehouse is noraml
+ result = sql """ show databases """
+ logger.info("when warehouse is normal the result of sql from root is {}", result)
+
+ result = connect(user = "${user}", password = '', url = context.config.jdbcUrl) {
+ sql """ show databases """
+ }
+ logger.info("when warehouse is normal the result of sql from user is {}", result)
+
+ result = http_post(context.config.feHttpAddress, url0, stmt_json)
+ def obj = new JsonSlurper().parseText(result)
+ logger.info("when warehouse is normal the result of http from root is {}", obj)
+ assertEquals(obj.code, 0)
+
+ result = http_post(context.config.feHttpAddress, url0, stmt_json, "${user}", "")
+ obj = new JsonSlurper().parseText(result)
+ logger.info("when warehouse is normal the result of http from user is {}", obj)
+ assertEquals(obj.code, 0)
+
+ // when warehouse is overdue
+ try {
+ def alter_request = JsonOutput.toJson(new AlterRequest(instance_id: instance_id, op: "SET_OVERDUE"))
+ result = http_post(context.config.metaServiceHttpAddress, "/MetaService/http/set_instance_status?token=greedisgood9999", alter_request)
+ obj = new JsonSlurper().parseText(result)
+ logger.info("try to set warehouse overdue, the result is {}", obj)
+ sleep(40000)
+
+ result = sql """ show databases """
+ logger.info("when warehouse is overdue the result of sql from root is {}", result)
+
+ try {
+ result = connect(user = "${user}", password = '', url = context.config.jdbcUrl) {
+ sql """ show databases """
+ }
+ logger.info("when warehouse is overdue the result of sql from user is {}", result)
+ assertTrue(false, "since warehouse is overdue, the sql from normal user should not success")
+ } catch (Exception e) {
+ logger.info("when warehouse is overdue the exception of sql from user is {}", e.getMessage())
+ }
+
+ result = http_post(context.config.feHttpAddress, url0, stmt_json)
+ obj = new JsonSlurper().parseText(result)
+ logger.info("when warehouse is overdue the result of http from root is {}", obj)
+ assertEquals(obj.code, 0)
+
+ result = http_post(context.config.feHttpAddress, url0, stmt_json, "${user}", "")
+ obj = new JsonSlurper().parseText(result)
+ logger.info("when warehouse is overdue the result of http from user is {}", obj)
+ assertEquals(obj.code, 1)
+ } catch (Exception e) {
+ alter_request = JsonOutput.toJson(new AlterRequest(instance_id: instance_id, op: "SET_NORMAL"))
+ result = http_post(context.config.metaServiceHttpAddress, "/MetaService/http/set_instance_status?token=greedisgood9999", alter_request)
+ obj = new JsonSlurper().parseText(result)
+ logger.info("try to set warehouse normal, the result is {}", obj)
+ throw e
+ }
+
+ // when warehouse transforms to normal
+ alter_request = JsonOutput.toJson(new AlterRequest(instance_id: instance_id, op: "SET_NORMAL"))
+ result = http_post(context.config.metaServiceHttpAddress, "/MetaService/http/set_instance_status?token=greedisgood9999", alter_request)
+ obj = new JsonSlurper().parseText(result)
+ logger.info("try to set warehouse normal, the result is {}", obj)
+ sleep(40000)
+
+ result = sql """ show databases """
+ logger.info("when warehouse transforms to normal the result of sql from root is {}", result)
+
+ result = connect(user = "${user}", password = '', url = context.config.jdbcUrl) {
+ sql """ show databases """
+ }
+ logger.info("when warehouse transforms to normal the result of sql from user is {}", result)
+
+ result = http_post(context.config.feHttpAddress, url0, stmt_json)
+ obj = new JsonSlurper().parseText(result)
+ logger.info("when warehouse transforms to normal the result of http from root is {}", obj)
+ assertEquals(obj.code, 0)
+
+ result = http_post(context.config.feHttpAddress, url0, stmt_json, "${user}", "")
+ obj = new JsonSlurper().parseText(result)
+ logger.info("when warehouse transforms to normal the result of http from user is {}", obj)
+ assertEquals(obj.code, 0)
+
+ // test the case when op parametr is not set
+ try {
+ alter_request = JsonOutput.toJson(new AlterRequestWithoutOp(instance_id: instance_id))
+ result = http_post(context.config.metaServiceHttpAddress, "/MetaService/http/set_instance_status?token=greedisgood9999", alter_request)
+ obj = new JsonSlurper().parseText(result)
+ logger.info("the result is {}", obj)
+ assertTrue(false, "it's not a vlid request")
+ } catch (Exception e) {
+ assertTrue(e.getMessage().contains("400"))
+ }
+ } finally {
+ try_sql("DROP USER ${user}")
+ }
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org