Skip to content

Commit e332116

Browse files
srlchsrlch
authored
[Feature] Support Cluster Snapshot Backup: checkpoint and image backup (part3) (StarRocks#54695)
Signed-off-by: srlch <linzichao@starrocks.com>
1 parent 651ea25 commit e332116

22 files changed

Lines changed: 915 additions & 55 deletions

‎be/src/exec/schema_scanner/schema_cluster_snapshot_jobs_scanner.cpp‎

Lines changed: 12 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -25,8 +25,8 @@ namespace starrocks {
2525
SchemaScanner::ColumnDesc SchemaClusterSnapshotJobsScanner::_s_columns[] = {
2626
{"SNAPSHOT_NAME", TypeDescriptor::create_varchar_type(sizeof(StringValue)), sizeof(StringValue), true},
2727
{"JOB_ID", TypeDescriptor::from_logical_type(TYPE_BIGINT), sizeof(long), true},
28-
{"CREATED_TIME", TypeDescriptor::from_logical_type(TYPE_BIGINT), sizeof(long), true},
29-
{"FINISHED_TIME", TypeDescriptor::from_logical_type(TYPE_BIGINT), sizeof(long), true},
28+
{"CREATED_TIME", TypeDescriptor::from_logical_type(TYPE_DATETIME), sizeof(DateTimeValue), true},
29+
{"FINISHED_TIME", TypeDescriptor::from_logical_type(TYPE_DATETIME), sizeof(DateTimeValue), true},
3030
{"STATE", TypeDescriptor::create_varchar_type(sizeof(StringValue)), sizeof(StringValue), true},
3131
{"DETAIL_INFO", TypeDescriptor::create_varchar_type(sizeof(StringValue)), sizeof(StringValue), true},
3232
{"ERROR_MESSAGE", TypeDescriptor::create_varchar_type(sizeof(StringValue)), sizeof(StringValue), true},
@@ -52,8 +52,16 @@ Status SchemaClusterSnapshotJobsScanner::_fill_chunk(ChunkPtr* chunk) {
5252
auto& slot_id_map = (*chunk)->get_slot_id_to_index_map();
5353
const TClusterSnapshotJobsItem& info = _result.items[_index];
5454
DatumArray datum_array{
55-
Slice(info.snapshot_name), info.job_id, info.created_time,
56-
info.finished_time, Slice(info.state), Slice(info.detail_info),
55+
Slice(info.snapshot_name),
56+
info.job_id,
57+
58+
TimestampValue::create_from_unixtime(info.created_time, _runtime_state->timezone_obj()),
59+
60+
TimestampValue::create_from_unixtime(info.finished_time, _runtime_state->timezone_obj()),
61+
62+
Slice(info.state),
63+
Slice(info.detail_info),
64+
5765
Slice(info.error_message),
5866
};
5967
for (const auto& [slot_id, index] : slot_id_map) {

‎be/src/exec/schema_scanner/schema_cluster_snapshots_scanner.cpp‎

Lines changed: 12 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -25,8 +25,8 @@ namespace starrocks {
2525
SchemaScanner::ColumnDesc SchemaClusterSnapshotsScanner::_s_columns[] = {
2626
{"SNAPSHOT_NAME", TypeDescriptor::create_varchar_type(sizeof(StringValue)), sizeof(StringValue), true},
2727
{"SNAPSHOT_TYPE", TypeDescriptor::create_varchar_type(sizeof(StringValue)), sizeof(StringValue), true},
28-
{"CREATED_TIME", TypeDescriptor::from_logical_type(TYPE_BIGINT), sizeof(long), true},
29-
{"FINISHED_TIME", TypeDescriptor::from_logical_type(TYPE_BIGINT), sizeof(long), true},
28+
{"CREATED_TIME", TypeDescriptor::from_logical_type(TYPE_DATETIME), sizeof(DateTimeValue), true},
29+
{"FINISHED_TIME", TypeDescriptor::from_logical_type(TYPE_DATETIME), sizeof(DateTimeValue), true},
3030
{"FE_JOURNAL_ID", TypeDescriptor::from_logical_type(TYPE_BIGINT), sizeof(long), true},
3131
{"STARMGR_JOURNAL_ID", TypeDescriptor::from_logical_type(TYPE_BIGINT), sizeof(long), true},
3232
{"PROPERTIES", TypeDescriptor::create_varchar_type(sizeof(StringValue)), sizeof(StringValue), true},
@@ -54,9 +54,17 @@ Status SchemaClusterSnapshotsScanner::_fill_chunk(ChunkPtr* chunk) {
5454
auto& slot_id_map = (*chunk)->get_slot_id_to_index_map();
5555
const TClusterSnapshotsItem& info = _result.items[_index];
5656
DatumArray datum_array{
57-
Slice(info.snapshot_name), Slice(info.snapshot_type), info.created_time, info.finished_time,
57+
Slice(info.snapshot_name),
58+
Slice(info.snapshot_type),
5859

59-
info.fe_jouranl_id, info.starmgr_jouranl_id, Slice(info.properties), Slice(info.storage_volume),
60+
TimestampValue::create_from_unixtime(info.created_time, _runtime_state->timezone_obj()),
61+
62+
TimestampValue::create_from_unixtime(info.finished_time, _runtime_state->timezone_obj()),
63+
64+
info.fe_jouranl_id,
65+
info.starmgr_jouranl_id,
66+
Slice(info.properties),
67+
Slice(info.storage_volume),
6068

6169
Slice(info.storage_path),
6270
};

‎fe/fe-core/src/main/java/com/starrocks/catalog/system/information/ClusterSnapshotJobsTable.java‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -33,8 +33,8 @@ public static SystemTable create() {
3333
builder()
3434
.column("SNAPSHOT_NAME", ScalarType.createVarchar(NAME_CHAR_LEN))
3535
.column("JOB_ID", ScalarType.createType(PrimitiveType.BIGINT))
36-
.column("CREATED_TIME", ScalarType.createType(PrimitiveType.BIGINT))
37-
.column("FINISHED_TIME", ScalarType.createType(PrimitiveType.BIGINT))
36+
.column("CREATED_TIME", ScalarType.createType(PrimitiveType.DATETIME))
37+
.column("FINISHED_TIME", ScalarType.createType(PrimitiveType.DATETIME))
3838
.column("STATE", ScalarType.createVarchar(NAME_CHAR_LEN))
3939
.column("DETAIL_INFO", ScalarType.createVarchar(NAME_CHAR_LEN))
4040
.column("ERROR_MESSAGE", ScalarType.createVarchar(NAME_CHAR_LEN))

‎fe/fe-core/src/main/java/com/starrocks/catalog/system/information/ClusterSnapshotsTable.java‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -33,8 +33,8 @@ public static SystemTable create() {
3333
builder()
3434
.column("SNAPSHOT_NAME", ScalarType.createVarchar(NAME_CHAR_LEN))
3535
.column("SNAPSHOT_TYPE", ScalarType.createVarchar(NAME_CHAR_LEN))
36-
.column("CREATED_TIME", ScalarType.createType(PrimitiveType.BIGINT))
37-
.column("FINISHED_TIME", ScalarType.createType(PrimitiveType.BIGINT))
36+
.column("CREATED_TIME", ScalarType.createType(PrimitiveType.DATETIME))
37+
.column("FINISHED_TIME", ScalarType.createType(PrimitiveType.DATETIME))
3838
.column("FE_JOURNAL_ID", ScalarType.createType(PrimitiveType.BIGINT))
3939
.column("STARMGR_JOURNAL_ID", ScalarType.createType(PrimitiveType.BIGINT))
4040
.column("PROPERTIES", ScalarType.createVarchar(NAME_CHAR_LEN))

‎fe/fe-core/src/main/java/com/starrocks/common/Config.java‎

Lines changed: 6 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -3393,4 +3393,10 @@ public class Config extends ConfigBase {
33933393

33943394
@ConfField(mutable = false)
33953395
public static int query_deploy_threadpool_size = max(50, getRuntime().availableProcessors() * 10);
3396+
3397+
@ConfField(mutable = true)
3398+
public static long automated_cluster_snapshot_interval_seconds = 1800;
3399+
3400+
@ConfField(mutable = false)
3401+
public static int max_historical_automated_cluster_snapshot_jobs = 100;
33963402
}

‎fe/fe-core/src/main/java/com/starrocks/fs/HdfsUtil.java‎

Lines changed: 5 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -71,6 +71,11 @@ public static void copyToLocal(String srcPath, String destPath, Map<String, Stri
7171
hdfsService.copyToLocal(srcPath, destPath, properties);
7272
}
7373

74+
public static void copyFromLocal(String srcPath, String destPath, Map<String, String> properties)
75+
throws StarRocksException {
76+
hdfsService.copyFromLocal(srcPath, destPath, properties);
77+
}
78+
7479
/**
7580
* Parse file status in path with broker, except directory
7681
*

‎fe/fe-core/src/main/java/com/starrocks/fs/hdfs/HdfsFsManager.java‎

Lines changed: 18 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -39,6 +39,7 @@
3939
import org.apache.hadoop.fs.FSDataOutputStream;
4040
import org.apache.hadoop.fs.FileStatus;
4141
import org.apache.hadoop.fs.FileSystem;
42+
import org.apache.hadoop.fs.FileUtil;
4243
import org.apache.hadoop.fs.Path;
4344
import org.apache.hadoop.hdfs.HdfsConfiguration;
4445
import org.apache.hadoop.security.UserGroupInformation;
@@ -47,6 +48,7 @@
4748
import software.amazon.awssdk.awscore.util.AwsHostNameUtils;
4849
import software.amazon.awssdk.regions.Region;
4950

51+
import java.io.File;
5052
import java.io.FileNotFoundException;
5153
import java.io.IOException;
5254
import java.io.InterruptedIOException;
@@ -1212,6 +1214,22 @@ public void copyToLocal(String srcPath, String destPath, Map<String, String> pro
12121214
}
12131215
}
12141216

1217+
public void copyFromLocal(String srcPath, String destPath, Map<String, String> properties) throws StarRocksException {
1218+
HdfsFs fileSystem = getFileSystem(destPath, properties, null);
1219+
try {
1220+
WildcardURI destPathUri = new WildcardURI(destPath);
1221+
File srcFile = new File(srcPath);
1222+
FileUtil.copy(srcFile, fileSystem.getDFSFileSystem(), new Path(destPathUri.getPath()), false, new Configuration());
1223+
} catch (InterruptedIOException e) {
1224+
Thread.interrupted(); // clear interrupted flag
1225+
LOG.error("Interrupted while copy local {} to {} ", srcPath, destPath, e);
1226+
throw new StarRocksException("Failed to copy local " + srcPath + " to " + destPath, e);
1227+
} catch (Exception e) {
1228+
LOG.error("Exception while copy local {} to {} ", srcPath, destPath, e);
1229+
throw new StarRocksException("Failed to copy local " + srcPath + " to " + destPath, e);
1230+
}
1231+
}
1232+
12151233
public List<FileStatus> listFileMeta(String path, Map<String, String> properties) throws StarRocksException {
12161234
WildcardURI pathUri = new WildcardURI(path);
12171235
HdfsFs fileSystem = getFileSystem(path, properties, null);

‎fe/fe-core/src/main/java/com/starrocks/fs/hdfs/HdfsService.java‎

Lines changed: 5 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -59,6 +59,11 @@ public void copyToLocal(String srcPath, String destPath, Map<String, String> pro
5959
LOG.info("Copied {} to local {}", srcPath, destPath);
6060
}
6161

62+
public void copyFromLocal(String srcPath, String destPath, Map<String, String> properties) throws StarRocksException {
63+
fileSystemManager.copyFromLocal(srcPath, destPath, properties);
64+
LOG.info("Copied local {} to {}", srcPath, destPath);
65+
}
66+
6267
public void listPath(TBrokerListPathRequest request, List<TBrokerFileStatus> fileStatuses, boolean skipDir,
6368
boolean fileNameOnly) throws StarRocksException {
6469
LOG.info("receive a list path request, path: {}", request.path);

‎fe/fe-core/src/main/java/com/starrocks/journal/CheckpointWorker.java‎

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -80,9 +80,11 @@ protected void runAfterCatalogReady() {
8080
if (nextPoint == null) {
8181
return;
8282
}
83+
8384
if (nextPoint.journalId <= getImageJournalId()) {
8485
return;
8586
}
87+
8688
if (nextPoint.epoch != servingGlobalState.getEpoch()) {
8789
return;
8890
}

‎fe/fe-core/src/main/java/com/starrocks/lake/StarOSAgent.java‎

Lines changed: 4 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -140,6 +140,10 @@ public void getServiceId() {
140140
LOG.info("get serviceId {} from starMgr", serviceId);
141141
}
142142

143+
public String getRawServiceId() {
144+
return serviceId;
145+
}
146+
143147
public String addFileStore(FileStoreInfo fsInfo) throws DdlException {
144148
try {
145149
return client.addFileStore(fsInfo, serviceId);

0 commit comments

Comments
 (0)