Add force_replace flag for restore

pkg/ccr/job.go

@@ -66,6 +66,7 @@ var (
 	featureSkipRollupBinlogs            bool
 	featureTxnInsert                    bool
 	featureFilterStorageMedium          bool
+	featureRestoreReplaceDiffSchema     bool
 
 	ErrMaterializedViewTable = xerror.NewWithoutStack(xerror.Meta, "Not support table type: materialized view")
 )
@@ -95,6 +96,8 @@ func init() {
 		"enable txn insert support")
 	flag.BoolVar(&featureFilterStorageMedium, "feature_filter_storage_medium", true,
 		"enable filter storage medium property")
+	flag.BoolVar(&featureRestoreReplaceDiffSchema, "feature_restore_replace_diff_schema", true,
+		"replace the table with different schema during restore")
 }
 
 type SyncType int
@@ -1057,6 +1060,9 @@ func (j *Job) fullSync() error {
 		if featureAtomicRestore {
 			restoreReq.AtomicRestore = true
 		}
+		if featureRestoreReplaceDiffSchema {
+			restoreReq.ForceReplace = true
+		}
 		restoreResp, err := destRpc.RestoreSnapshot(dest, &restoreReq)
 		if err != nil {
 			return err

pkg/rpc/fe.go

@@ -95,6 +95,7 @@ type RestoreSnapshotRequest struct {
 	CleanPartitions bool
 	CleanTables     bool
 	Compress        bool
+	ForceReplace    bool
 }
 
 type IFeRpc interface {
@@ -800,14 +801,15 @@ func (rpc *singleFeClient) RestoreSnapshot(spec *base.Spec, restoreReq *RestoreS
 		CleanPartitions: &restoreReq.CleanPartitions,
 		AtomicRestore:   &restoreReq.AtomicRestore,
 		Compressed:      utils.ThriftValueWrapper(restoreReq.Compress),
+		ForceReplace:    &restoreReq.ForceReplace,
 	}
 	setAuthInfo(req, spec)
 
 	// NOTE: ignore meta, because it's too large
-	log.Debugf("RestoreSnapshotRequest user %s, db %s, table %s, label name %s, properties %v, clean tables: %t, clean partitions: %t, atomic restore: %t, compressed: %t",
+	log.Debugf("RestoreSnapshotRequest user %s, db %s, table %s, label name %s, properties %v, clean tables: %t, clean partitions: %t, atomic restore: %t, compressed: %t, forceReplace: %t",
 		req.GetUser(), req.GetDb(), req.GetTable(), req.GetLabelName(), properties,
 		restoreReq.CleanTables, restoreReq.CleanPartitions, restoreReq.AtomicRestore,
-		req.GetCompressed())
+		req.GetCompressed(), restoreReq.ForceReplace)
 
 	if resp, err := client.RestoreSnapshot(context.Background(), req); err != nil {
 		return nil, xerror.Wrapf(err, xerror.RPC, "RestoreSnapshot failed")

pkg/rpc/thrift/FrontendService.thrift

@@ -1388,6 +1388,7 @@ struct TRestoreSnapshotRequest {
     14: optional bool clean_partitions
     15: optional bool atomic_restore
     16: optional bool compressed;
+    17: optional bool force_replace
 }
 
 struct TRestoreSnapshotResult {

regression-test/suites/db_sync/table/diff_schema/test_ds_tbl_diff_schema.groovy

@@ -0,0 +1,78 @@
+// 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_ds_tbl_diff_schema") {
+    def helper = new GroovyShell(new Binding(['suite': delegate]))
+            .evaluate(new File("${context.config.suitePath}/../common", "helper.groovy"))
+
+    def dbName = context.dbName
+    def tableName = "tbl_" + helper.randomSuffix()
+
+    def exist = { res -> Boolean
+        return res.size() != 0
+    }
+
+    def notExist = { res -> Boolean
+        return res.size() == 0
+    }
+
+    sql "DROP TABLE IF EXISTS ${tableName}"
+    target_sql "DROP TABLE IF EXISTS ${tableName}"
+
+    sql """
+        CREATE TABLE if NOT EXISTS ${tableName}
+        (
+            `test` INT,
+            `id` INT
+        )
+        ENGINE=OLAP
+        PARTITION BY RANGE(`test`, `id`)
+        (
+        )
+        DISTRIBUTED BY HASH(id) BUCKETS 1
+        PROPERTIES (
+            "replication_allocation" = "tag.location.default: 1",
+            "binlog.enable" = "true"
+        )
+    """
+
+    target_sql """
+        CREATE TABLE if NOT EXISTS ${tableName}
+        (
+            `test` INT,
+            `id` INT
+        )
+        ENGINE=OLAP
+        PARTITION BY LIST(`test`, `id`)
+        (
+        )
+        DISTRIBUTED BY HASH(id) BUCKETS 1
+        PROPERTIES (
+            "replication_allocation" = "tag.location.default: 1",
+            "binlog.enable" = "true"
+        )
+    """
+
+    helper.ccrJobDelete()
+    helper.ccrJobCreate()
+
+    assertTrue(helper.checkRestoreFinishTimesOf("${tableName}", 30))
+
+    assertTrue(helper.checkShowTimesOf("SHOW TABLES LIKE \"${tableName}\"", exist, 60, "sql"))
+
+    assertTrue(helper.checkShowTimesOf("SHOW TABLES LIKE \"${tableName}\"", exist, 60, "target"))
+}