[Enhancement] add support for restore to ccr

Author: Vallishp

fe/fe-core/src/main/java/org/apache/doris/backup/BackupHandler.java

@@ -570,7 +570,7 @@ private void restore(Repository repository, Database db, RestoreStmt stmt) throw
                 env, repository.getId());
         }
 
-        env.getEditLog().logRestoreJob(restoreJob);
+        env.getEditLog().logRestoreJob(restoreJob, null);
 
         // must put to dbIdToBackupOrRestoreJob after edit log, otherwise the state of job may be changed.
         addBackupOrRestoreJob(db.getId(), restoreJob);

fe/fe-core/src/main/java/org/apache/doris/backup/RestoreBinLogInfo.java

@@ -0,0 +1,68 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package org.apache.doris.backup;
+
+import org.apache.doris.persist.gson.GsonUtils;
+
+import com.google.common.collect.Maps;
+import com.google.gson.annotations.SerializedName;
+
+import java.util.List;
+import java.util.Map;
+import java.util.stream.Collectors;
+
+public class RestoreBinLogInfo {
+    // currently we are sending only DB and table info.
+    // partitions level restore not possible since, there can be
+    // race condition when two partition recover and ccr-syncer try to sync it.
+    @SerializedName(value = "dbId")
+    private long dbId;
+    @SerializedName(value = "dbName")
+    private String dbName;
+    @SerializedName(value = "tableInfo")
+    // map of tableId and TableName.
+    private Map<Long, String> tableInfo = Maps.newHashMap();
+
+    /*
+     * constuctor
+     */
+    public RestoreBinLogInfo(long dbId, String dbName) {
+        this.dbId = dbId;
+        this.dbName = dbName;
+    }
+
+    public void addTableInfo(long tableId, String tableName) {
+        tableInfo.put(tableId, tableName);
+    }
+
+    public long getDbId() {
+        return dbId;
+    }
+
+    public List<Long> getTableIdList() {
+        return  tableInfo.entrySet().stream().map(Map.Entry::getKey).collect(Collectors.toList());
+    }
+
+    public String toJson() {
+        return GsonUtils.GSON.toJson(this);
+    }
+
+    public static RestoreBinLogInfo fromJson(String json) {
+        return GsonUtils.GSON.fromJson(json, RestoreBinLogInfo.class);
+    }
+}

fe/fe-core/src/main/java/org/apache/doris/backup/RestoreJob.java

@@ -1638,7 +1638,7 @@ private void waitingAllSnapshotsFinished() {
             snapshotFinishedTime = System.currentTimeMillis();
             state = RestoreJobState.DOWNLOAD;
 
-            env.getEditLog().logRestoreJob(this);
+            env.getEditLog().logRestoreJob(this, null);
             LOG.info("finished making snapshots. {}", this);
             return;
         }
@@ -1981,7 +1981,7 @@ private void waitingAllDownloadFinished() {
             // backupMeta is useless now
             backupMeta = null;
 
-            env.getEditLog().logRestoreJob(this);
+            env.getEditLog().logRestoreJob(this, null);
             LOG.info("finished to download. {}", this);
         }
 
@@ -2036,6 +2036,7 @@ private Status allTabletCommitted(boolean isReplay) {
             return new Status(ErrCode.NOT_FOUND, "database " + dbId + " does not exist");
         }
 
+        RestoreBinLogInfo restoreBinLogInfo = new RestoreBinLogInfo(dbId, db.getName());
         // replace the origin tables in atomic.
         if (isAtomicRestore) {
             Status st = atomicReplaceOlapTables(db, isReplay);
@@ -2097,6 +2098,7 @@ private Status allTabletCommitted(boolean isReplay) {
         }
 
         if (!isReplay) {
+            restoredTbls.stream().forEach(tbl -> restoreBinLogInfo.addTableInfo(tbl.getId(), tbl.getName()));
             restoredPartitions.clear();
             restoredTbls.clear();
             restoredResources.clear();
@@ -2111,7 +2113,7 @@ private Status allTabletCommitted(boolean isReplay) {
             finishedTime = System.currentTimeMillis();
             state = RestoreJobState.FINISHED;
 
-            env.getEditLog().logRestoreJob(this);
+            env.getEditLog().logRestoreJob(this, restoreBinLogInfo);
         }
 
         LOG.info("job is finished. is replay: {}. {}", isReplay, this);
@@ -2383,7 +2385,7 @@ private void cancelInternal(boolean isReplay) {
             finishedTime = System.currentTimeMillis();
             state = RestoreJobState.CANCELLED;
             // log
-            env.getEditLog().logRestoreJob(this);
+            env.getEditLog().logRestoreJob(this, null);
 
             LOG.info("finished to cancel restore job. current state: {}. is replay: {}. {}",
                      curState.name(), isReplay, this);

fe/fe-core/src/main/java/org/apache/doris/binlog/BinlogManager.java

@@ -19,6 +19,7 @@
 
 import org.apache.doris.alter.AlterJobV2;
 import org.apache.doris.alter.IndexChangeJob;
+import org.apache.doris.backup.RestoreBinLogInfo;
 import org.apache.doris.catalog.Database;
 import org.apache.doris.catalog.Env;
 import org.apache.doris.common.Config;
@@ -328,6 +329,17 @@ public void addTruncateTable(TruncateTableInfo info, long commitSeq) {
         addBinlog(dbId, tableIds, commitSeq, timestamp, type, data, false, record);
     }
 
+    public void addRestoreInfo(RestoreBinLogInfo info, long commitSeq) {
+        long dbId = info.getDbId();
+        List<Long> tableIds = info.getTableIdList();
+        long timestamp = -1;
+        TBinlogType type = TBinlogType.RESTORE_INFO;
+        String data = info.toJson();
+
+        addBinlog(dbId, tableIds, commitSeq, timestamp, type, data, false, info);
+    }
+
+
     public void addTableRename(TableInfo info, long commitSeq) {
         long dbId = info.getDbId();
         List<Long> tableIds = Lists.newArrayList();

fe/fe-core/src/main/java/org/apache/doris/persist/EditLog.java

@@ -23,6 +23,7 @@
 import org.apache.doris.analysis.UserIdentity;
 import org.apache.doris.backup.BackupJob;
 import org.apache.doris.backup.Repository;
+import org.apache.doris.backup.RestoreBinLogInfo;
 import org.apache.doris.backup.RestoreJob;
 import org.apache.doris.binlog.AddPartitionRecord;
 import org.apache.doris.binlog.CreateTableRecord;
@@ -1689,8 +1690,12 @@ public void logAlterRepository(Repository repo) {
         logEdit(OperationType.OP_ALTER_REPOSITORY, repo);
     }
 
-    public void logRestoreJob(RestoreJob job) {
-        logEdit(OperationType.OP_RESTORE_JOB, job);
+    public void logRestoreJob(RestoreJob job, RestoreBinLogInfo binInfo) {
+        long logId = logEdit(OperationType.OP_RESTORE_JOB, job);
+        if (binInfo != null) {
+            LOG.info("log restore info, logId:{}, infos: {}", logId, binInfo.toJson());
+            Env.getCurrentEnv().getBinlogManager().addRestoreInfo(binInfo, logId);
+        }
     }
 
     public void logUpdateUserProperty(UserPropertyInfo propertyInfo) {

gensrc/thrift/FrontendService.thrift

@@ -1199,7 +1199,8 @@ enum TBinlogType {
   RENAME_ROLLUP = 21,
   RENAME_PARTITION = 22,
   DROP_ROLLUP = 23,
-
+  RECOVER_INFO = 24,
+  RESTORE_INFO = 25,
   // Keep some IDs for allocation so that when new binlog types are added in the
   // future, the changes can be picked back to the old versions without breaking
   // compatibility.
@@ -1215,9 +1216,7 @@ enum TBinlogType {
   //    MODIFY_XXX = 17,
   //    MIN_UNKNOWN = 18,
   //    UNKNOWN_3 = 19,
-  MIN_UNKNOWN = 24,
-  UNKNOWN_9 = 25,
-  UNKNOWN_10 = 26,
+  MIN_UNKNOWN = 26,
   UNKNOWN_11 = 27,
   UNKNOWN_12 = 28,
   UNKNOWN_13 = 29,

regression-test/conf/regression-conf.groovy

@@ -259,3 +259,20 @@ lakesoulMinioEndpoint="*******"
 metaServiceToken = "greedisgood9999"
 instanceId = "default_instance_id"
 multiClusterInstance = "default_instance_id"
+
+// JDBC configuration
+jdbcUrl = "jdbc:mysql://127.0.0.1:9030/?"
+targetJdbcUrl = "jdbc:mysql://127.0.0.1:9190/?"
+jdbcUser = "root"
+jdbcPassword = ""
+feSourceThriftAddress = "127.0.0.1:9020"
+feTargetThriftAddress = "127.0.0.1:9020"
+syncerAddress = "127.0.0.1:9190"
+feSyncerUser = "root"
+feSyncerPassword = ""
+feHttpAddress = "127.0.0.1:8330"
+// CCR configuration
+ccrDownstreamUrl = "jdbc:mysql://127.0.0.1:9030/?"
+ccrDownstreamUser = "root"
+ccrDownstreamPassword = ""
+ccrDownstreamFeThriftAddress = "127.0.0.1:9020"

regression-test/framework/src/main/groovy/org/apache/doris/regression/suite/Syncer.groovy

@@ -947,4 +947,35 @@ class Syncer {
         )
         """
     }
-}
+
+    void createMinoIoRepository(String name, boolean readOnly = false) {
+        String ak = "xuYlrTbIGdz8oA8HeHSa"
+        String sk = "6n7S3PFkDOBDSJiwQLURNywVwT20ptqOAY3wrOSF"
+        String endpoint = "http://127.0.0.1:9000"
+        String region = suite.getS3Region()
+        String bucket = "test"
+
+
+        suite.try_sql "DROP REPOSITORY `${name}`"
+        suite.sql """
+        CREATE ${readOnly ? "READ ONLY" : ""} REPOSITORY `${name}`
+        WITH S3
+        ON LOCATION "s3://${bucket}/${name}"
+        PROPERTIES
+        (
+            "s3.endpoint" = "http://127.0.0.1:9000",
+            "s3.access_key" = "${ak}",
+            "s3.secret_key" = "${sk}",
+            "s3.root"= "${name}",
+            "s3.use_path_style" = "true",
+            "s3.region" = "${region}"
+        )
+            """
+    }
+
+    void disableDbBinlog() {
+        suite.sql """
+            ALTER DATABASE ${context.dbName} SET properties ("binlog.enable" = "false")
+            """
+    }
+}