Skip to content

Commit 68d13d2

Browse files
committed
[Enhancement] add support for restore to ccr
1 parent 86c3c76 commit 68d13d2

6 files changed

Lines changed: 110 additions & 5 deletions

File tree

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: 13 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -215,6 +215,9 @@ public enum RestoreJobState {
215215
@SerializedName("prop")
216216
private Map<String, String> properties = Maps.newHashMap();
217217

218+
@SerializedName("resbinlog")
219+
private RestoreBinlogInfo restoreBinlogInfo = null; // only set in FinishedCase.
220+
218221
private MarkedCountDownLatch<Long, Long> createReplicaTasksLatch = null;
219222

220223
public RestoreJob() {
@@ -401,6 +404,10 @@ public boolean isFinished() {
401404
return state == RestoreJobState.FINISHED;
402405
}
403406

407+
public RestoreBinlogInfo getRestoreBinlogInfo() {
408+
return restoreBinlogInfo;
409+
}
410+
404411
@Override
405412
public synchronized Status updateRepo(Repository repo) {
406413
this.repo = repo;
@@ -2036,6 +2043,7 @@ private Status allTabletCommitted(boolean isReplay) {
20362043
return new Status(ErrCode.NOT_FOUND, "database " + dbId + " does not exist");
20372044
}
20382045

2046+
restoreBinlogInfo = new RestoreBinlogInfo(dbId, db.getName());
20392047
// replace the origin tables in atomic.
20402048
if (isAtomicRestore) {
20412049
Status st = atomicReplaceOlapTables(db, isReplay);
@@ -2052,6 +2060,9 @@ private Status allTabletCommitted(boolean isReplay) {
20522060
if (tbl == null) {
20532061
continue;
20542062
}
2063+
//just restore existing table, then version will change.
2064+
// so we need to write bin log for those tables also.
2065+
restoreBinlogInfo.addTableInfo(tbl.getId(), tbl.getName());
20552066
OlapTable olapTbl = (OlapTable) tbl;
20562067
if (!tbl.writeLockIfExist()) {
20572068
continue;
@@ -2097,6 +2108,7 @@ private Status allTabletCommitted(boolean isReplay) {
20972108
}
20982109

20992110
if (!isReplay) {
2111+
restoredTbls.stream().forEach(tbl -> restoreBinlogInfo.addTableInfo(tbl.getId(), tbl.getName()));
21002112
restoredPartitions.clear();
21012113
restoredTbls.clear();
21022114
restoredResources.clear();
@@ -2112,6 +2124,7 @@ private Status allTabletCommitted(boolean isReplay) {
21122124
state = RestoreJobState.FINISHED;
21132125

21142126
env.getEditLog().logRestoreJob(this);
2127+
restoreBinlogInfo = null;
21152128
}
21162129

21172130
LOG.info("job is finished. is replay: {}. {}", 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 & 1 deletion
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;
@@ -1694,7 +1695,13 @@ public void logAlterRepository(Repository repo) {
16941695
}
16951696

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

17001707
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)