Skip to content

Commit 5e011f0

Browse files
VallishpVallishp
committed
[Enhancement] add support for restore to ccr
1 parent 279c5b7 commit 5e011f0

7 files changed

Lines changed: 107 additions & 11 deletions

File tree

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

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -570,7 +570,7 @@ private void restore(Repository repository, Database db, RestoreStmt stmt) throw
570570
env, repository.getId());
571571
}
572572

573-
env.getEditLog().logRestoreJob(restoreJob);
573+
env.getEditLog().logRestoreJob(restoreJob, null);
574574

575575
// must put to dbIdToBackupOrRestoreJob after edit log, otherwise the state of job may be changed.
576576
addBackupOrRestoreJob(db.getId(), restoreJob);

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

Lines changed: 68 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,68 @@
1+
// Licensed to the Apache Software Foundation (ASF) under one
2+
// or more contributor license agreements. See the NOTICE file
3+
// distributed with this work for additional information
4+
// regarding copyright ownership. The ASF licenses this file
5+
// to you under the Apache License, Version 2.0 (the
6+
// "License"); you may not use this file except in compliance
7+
// with the License. You may obtain a copy of the License at
8+
//
9+
// http://www.apache.org/licenses/LICENSE-2.0
10+
//
11+
// Unless required by applicable law or agreed to in writing,
12+
// software distributed under the License is distributed on an
13+
// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
14+
// KIND, either express or implied. See the License for the
15+
// specific language governing permissions and limitations
16+
// under the License.
17+
18+
package org.apache.doris.backup;
19+
20+
import org.apache.doris.persist.gson.GsonUtils;
21+
22+
import com.google.common.collect.Maps;
23+
import com.google.gson.annotations.SerializedName;
24+
25+
import java.util.List;
26+
import java.util.Map;
27+
import java.util.stream.Collectors;
28+
29+
public class RestoreBinLogInfo {
30+
// currently we are sending only DB and table info.
31+
// partitions level restore not possible since, there can be
32+
// race condition when two partition recover and ccr-syncer try to sync it.
33+
@SerializedName(value = "dbId")
34+
private long dbId;
35+
@SerializedName(value = "dbName")
36+
private String dbName;
37+
@SerializedName(value = "tableInfo")
38+
// map of tableId and TableName.
39+
private Map<Long, String> tableInfo = Maps.newHashMap();
40+
41+
/*
42+
* constuctor
43+
*/
44+
public RestoreBinLogInfo(long dbId, String dbName) {
45+
this.dbId = dbId;
46+
this.dbName = dbName;
47+
}
48+
49+
public void addTableInfo(long tableId, String tableName) {
50+
tableInfo.put(tableId, tableName);
51+
}
52+
53+
public long getDbId() {
54+
return dbId;
55+
}
56+
57+
public List<Long> getTableIdList() {
58+
return tableInfo.entrySet().stream().map(Map.Entry::getKey).collect(Collectors.toList());
59+
}
60+
61+
public String toJson() {
62+
return GsonUtils.GSON.toJson(this);
63+
}
64+
65+
public static RestoreBinLogInfo fromJson(String json) {
66+
return GsonUtils.GSON.fromJson(json, RestoreBinLogInfo.class);
67+
}
68+
}

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

Lines changed: 9 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -1638,7 +1638,7 @@ private void waitingAllSnapshotsFinished() {
16381638
snapshotFinishedTime = System.currentTimeMillis();
16391639
state = RestoreJobState.DOWNLOAD;
16401640

1641-
env.getEditLog().logRestoreJob(this);
1641+
env.getEditLog().logRestoreJob(this, null);
16421642
LOG.info("finished making snapshots. {}", this);
16431643
return;
16441644
}
@@ -1981,7 +1981,7 @@ private void waitingAllDownloadFinished() {
19811981
// backupMeta is useless now
19821982
backupMeta = null;
19831983

1984-
env.getEditLog().logRestoreJob(this);
1984+
env.getEditLog().logRestoreJob(this, null);
19851985
LOG.info("finished to download. {}", this);
19861986
}
19871987

@@ -2036,6 +2036,7 @@ private Status allTabletCommitted(boolean isReplay) {
20362036
return new Status(ErrCode.NOT_FOUND, "database " + dbId + " does not exist");
20372037
}
20382038

2039+
RestoreBinLogInfo restoreBinLogInfo = new RestoreBinLogInfo(dbId, db.getName());
20392040
// replace the origin tables in atomic.
20402041
if (isAtomicRestore) {
20412042
Status st = atomicReplaceOlapTables(db, isReplay);
@@ -2052,6 +2053,9 @@ private Status allTabletCommitted(boolean isReplay) {
20522053
if (tbl == null) {
20532054
continue;
20542055
}
2056+
//just restore existing table, then version will change.
2057+
// so we need to write bin log for those tables also.
2058+
restoreBinLogInfo.addTableInfo(tbl.getId(), tbl.getName());
20552059
OlapTable olapTbl = (OlapTable) tbl;
20562060
if (!tbl.writeLockIfExist()) {
20572061
continue;
@@ -2097,6 +2101,7 @@ private Status allTabletCommitted(boolean isReplay) {
20972101
}
20982102

20992103
if (!isReplay) {
2104+
restoredTbls.stream().forEach(tbl -> restoreBinLogInfo.addTableInfo(tbl.getId(), tbl.getName()));
21002105
restoredPartitions.clear();
21012106
restoredTbls.clear();
21022107
restoredResources.clear();
@@ -2111,7 +2116,7 @@ private Status allTabletCommitted(boolean isReplay) {
21112116
finishedTime = System.currentTimeMillis();
21122117
state = RestoreJobState.FINISHED;
21132118

2114-
env.getEditLog().logRestoreJob(this);
2119+
env.getEditLog().logRestoreJob(this, restoreBinLogInfo);
21152120
}
21162121

21172122
LOG.info("job is finished. is replay: {}. {}", isReplay, this);
@@ -2383,7 +2388,7 @@ private void cancelInternal(boolean isReplay) {
23832388
finishedTime = System.currentTimeMillis();
23842389
state = RestoreJobState.CANCELLED;
23852390
// log
2386-
env.getEditLog().logRestoreJob(this);
2391+
env.getEditLog().logRestoreJob(this, null);
23872392

23882393
LOG.info("finished to cancel restore job. current state: {}. is replay: {}. {}",
23892394
curState.name(), isReplay, this);

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

Lines changed: 12 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -19,6 +19,7 @@
1919

2020
import org.apache.doris.alter.AlterJobV2;
2121
import org.apache.doris.alter.IndexChangeJob;
22+
import org.apache.doris.backup.RestoreBinLogInfo;
2223
import org.apache.doris.catalog.Database;
2324
import org.apache.doris.catalog.Env;
2425
import org.apache.doris.common.Config;
@@ -329,6 +330,17 @@ public void addTruncateTable(TruncateTableInfo info, long commitSeq) {
329330
addBinlog(dbId, tableIds, commitSeq, timestamp, type, data, false, record);
330331
}
331332

333+
public void addRestoreInfo(RestoreBinLogInfo info, long commitSeq) {
334+
long dbId = info.getDbId();
335+
List<Long> tableIds = info.getTableIdList();
336+
long timestamp = -1;
337+
TBinlogType type = TBinlogType.RESTORE_INFO;
338+
String data = info.toJson();
339+
340+
addBinlog(dbId, tableIds, commitSeq, timestamp, type, data, false, info);
341+
}
342+
343+
332344
public void addTableRename(TableInfo info, long commitSeq) {
333345
long dbId = info.getDbId();
334346
List<Long> tableIds = Lists.newArrayList();

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

Lines changed: 8 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -23,6 +23,7 @@
2323
import org.apache.doris.analysis.UserIdentity;
2424
import org.apache.doris.backup.BackupJob;
2525
import org.apache.doris.backup.Repository;
26+
import org.apache.doris.backup.RestoreBinLogInfo;
2627
import org.apache.doris.backup.RestoreJob;
2728
import org.apache.doris.binlog.AddPartitionRecord;
2829
import org.apache.doris.binlog.CreateTableRecord;
@@ -1693,8 +1694,13 @@ public void logAlterRepository(Repository repo) {
16931694
logEdit(OperationType.OP_ALTER_REPOSITORY, repo);
16941695
}
16951696

1696-
public void logRestoreJob(RestoreJob job) {
1697-
logEdit(OperationType.OP_RESTORE_JOB, job);
1697+
public void logRestoreJob(RestoreJob job, RestoreBinLogInfo binInfo) {
1698+
long logId = logEdit(OperationType.OP_RESTORE_JOB, job);
1699+
// write bin log only if restore job the finished.
1700+
if ((job.isFinished()) && (binInfo != null)) {
1701+
LOG.info("log restore info, logId:{}, infos: {}", logId, binInfo.toJson());
1702+
Env.getCurrentEnv().getBinlogManager().addRestoreInfo(binInfo, logId);
1703+
}
16981704
}
16991705

17001706
public void logUpdateUserProperty(UserPropertyInfo propertyInfo) {

‎gensrc/thrift/FrontendService.thrift‎

Lines changed: 2 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -1200,7 +1200,7 @@ enum TBinlogType {
12001200
RENAME_PARTITION = 22,
12011201
DROP_ROLLUP = 23,
12021202
RECOVER_INFO = 24,
1203-
1203+
RESTORE_INFO = 25,
12041204
// Keep some IDs for allocation so that when new binlog types are added in the
12051205
// future, the changes can be picked back to the old versions without breaking
12061206
// compatibility.
@@ -1216,8 +1216,7 @@ enum TBinlogType {
12161216
// MODIFY_XXX = 17,
12171217
// MIN_UNKNOWN = 18,
12181218
// UNKNOWN_3 = 19,
1219-
MIN_UNKNOWN = 25,
1220-
UNKNOWN_10 = 26,
1219+
MIN_UNKNOWN = 26,
12211220
UNKNOWN_11 = 27,
12221221
UNKNOWN_12 = 28,
12231222
UNKNOWN_13 = 29,

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

Lines changed: 7 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -947,4 +947,10 @@ class Syncer {
947947
)
948948
"""
949949
}
950-
}
950+
951+
void disableDbBinlog() {
952+
suite.sql """
953+
ALTER DATABASE ${context.dbName} SET properties ("binlog.enable" = "false")
954+
"""
955+
}
956+
}

0 commit comments

Comments
 (0)