Skip to content

Commit

Permalink
[Enhancement] Support backup restore for logical view (backport #52077)…
Browse files Browse the repository at this point in the history 
… (#52237)

Signed-off-by: srlch <linzichao@starrocks.com>
  • Loading branch information
@srlch
srlch authored Oct 23, 2024
1 parent 400f097 commit 13d33f1
Show file tree
Hide file tree
Showing 10 changed files with 401 additions and 27 deletions.
17 changes: 13 additions & 4 deletions fe/fe-core/src/main/java/com/starrocks/backup/BackupHandler.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -323,8 +323,15 @@ private void backup(Repository repository, Database db, BackupStmt stmt) throws
ErrorReport.reportDdlException(ErrorCode.ERR_BAD_TABLE_ERROR, tblName);
return;
}
if (!tbl.isOlapTableOrMaterializedView()) {
ErrorReport.reportDdlException(ErrorCode.ERR_NOT_OLAP_TABLE, tblName);
if (!tbl.isSupportBackupRestore()) {
ErrorReport.reportDdlException(ErrorCode.ERR_COMMON_ERROR,
"Table: " + tblName + " can not support backup restore, type: " +
tbl.getType());
}

if (tbl.isOlapView()) {
backupTbls.add(tbl);
continue;
}

OlapTable olapTbl = (OlapTable) tbl;
Expand Down Expand Up @@ -441,8 +448,10 @@ private void restore(Repository repository, Database db, RestoreStmt stmt) throw
if (backupMeta != null) {
for (BackupTableInfo tblInfo : jobInfo.tables.values()) {
Table remoteTbl = backupMeta.getTable(tblInfo.name);
if (remoteTbl.isCloudNativeTable()) {
ErrorReport.reportDdlException(ErrorCode.ERR_NOT_OLAP_TABLE, remoteTbl.getName());
if (!remoteTbl.isSupportBackupRestore()) {
ErrorReport.reportDdlException(ErrorCode.ERR_COMMON_ERROR,
"Table: " + remoteTbl.getName() +
" can not support backup restore, type: " + remoteTbl.getType());
}
mvRestoreContext.addIntoMvBaseTableBackupInfoIfNeeded(db.getOriginName(), remoteTbl, jobInfo, tblInfo);
}
Expand Down
44 changes: 31 additions & 13 deletions fe/fe-core/src/main/java/com/starrocks/backup/BackupJob.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -58,9 +58,11 @@
import com.starrocks.catalog.Replica;
import com.starrocks.catalog.Table;
import com.starrocks.catalog.Tablet;
import com.starrocks.catalog.View;
import com.starrocks.common.AnalysisException;
import com.starrocks.common.Config;
import com.starrocks.common.UserException;
import com.starrocks.common.io.DeepCopy;
import com.starrocks.common.io.Text;
import com.starrocks.common.util.TimeUtils;
import com.starrocks.common.util.UUIDUtil;
Expand Down Expand Up @@ -392,12 +394,16 @@ protected void checkBackupTables(Database db) {
status = new Status(ErrCode.NOT_FOUND, "table " + tblName + " does not exist");
return;
}
if (!tbl.isOlapTableOrMaterializedView()) {
status = new Status(ErrCode.COMMON_ERROR, "table " + tblName
+ " is not OLAP table");
if (!tbl.isSupportBackupRestore()) {
status = new Status(ErrCode.UNSUPPORTED,
"Table: " + tblName + " can not support backup restore, type: " + tbl.getType());
return;
}

if (tbl.isOlapView()) {
continue;
}

OlapTable olapTbl = (OlapTable) tbl;
if (tableRef.getPartitionNames() != null) {
for (String partName : tableRef.getPartitionNames().getPartitionNames()) {
Expand Down Expand Up @@ -466,13 +472,17 @@ private void prepareAndSendSnapshotTask() {
// create snapshot tasks
for (TableRef tblRef : tableRefs) {
String tblName = tblRef.getName().getTbl();
OlapTable tbl = (OlapTable) db.getTable(tblName);
Table tbl = db.getTable(tblName);
if (tbl.isOlapView()) {
continue;
}
OlapTable olapTbl = (OlapTable) tbl;
List<Partition> partitions = Lists.newArrayList();
if (tblRef.getPartitionNames() == null) {
partitions.addAll(tbl.getPartitions());
partitions.addAll(olapTbl.getPartitions());
} else {
for (String partName : tblRef.getPartitionNames().getPartitionNames()) {
Partition partition = tbl.getPartition(partName);
Partition partition = olapTbl.getPartition(partName);
partitions.add(partition);
}
}
Expand All @@ -483,9 +493,9 @@ private void prepareAndSendSnapshotTask() {
long visibleVersion = physicalPartition.getVisibleVersion();
List<MaterializedIndex> indexes = physicalPartition.getMaterializedIndices(IndexExtState.VISIBLE);
for (MaterializedIndex index : indexes) {
int schemaHash = tbl.getSchemaHashByIndexId(index.getId());
int schemaHash = olapTbl.getSchemaHashByIndexId(index.getId());
for (Tablet tablet : index.getTablets()) {
prepareSnapshotTask(physicalPartition, tbl, tablet, index, visibleVersion, schemaHash);
prepareSnapshotTask(physicalPartition, olapTbl, tablet, index, visibleVersion, schemaHash);
if (status != Status.OK) {
return;
}
Expand All @@ -501,11 +511,17 @@ private void prepareAndSendSnapshotTask() {
List<Table> copiedTables = Lists.newArrayList();
for (TableRef tableRef : tableRefs) {
String tblName = tableRef.getName().getTbl();
OlapTable tbl = (OlapTable) db.getTable(tblName);
Table tbl = db.getTable(tblName);
if (tbl.isOlapView()) {
View view = (View) tbl;
copiedTables.add((Table) DeepCopy.copyWithGson(view, View.class));
continue;
}
OlapTable olapTbl = (OlapTable) tbl;
// only copy visible indexes
List<String> reservedPartitions = tableRef.getPartitionNames() == null ? null
: tableRef.getPartitionNames().getPartitionNames();
OlapTable copiedTbl = tbl.selectiveCopy(reservedPartitions, true, IndexExtState.VISIBLE);
OlapTable copiedTbl = olapTbl.selectiveCopy(reservedPartitions, true, IndexExtState.VISIBLE);
if (copiedTbl == null) {
status = new Status(ErrCode.COMMON_ERROR, "faild to copy table: " + tblName);
return;
Expand Down Expand Up @@ -684,9 +700,11 @@ private void saveMetaInfo() {
localMetaInfoFilePath = metaInfoFile.getAbsolutePath();

// 3. save job info file
jobInfo = BackupJobInfo.fromCatalog(createTime, label, dbName, dbId, backupMeta.getTables().values(),
snapshotInfos);
LOG.debug("job info: {}. {}", jobInfo, this);
// save table info into BackupJobInfo only for OlapTable or MV
List<Table> olapTbls = backupMeta.getTables().values().stream()
.filter(Table::isOlapTableOrMaterializedView).collect(Collectors.toList());
jobInfo = BackupJobInfo.fromCatalog(createTime, label, dbName, dbId, olapTbls, snapshotInfos);
LOG.warn("job info: {}. {}", jobInfo, this);
File jobInfoFile = new File(jobDir, Repository.PREFIX_JOB_INFO + createTimeStr);
if (!jobInfoFile.createNewFile()) {
status = new Status(ErrCode.COMMON_ERROR, "Failed to create job info file: " + jobInfoFile.toString());
Expand Down
73 changes: 65 additions & 8 deletions fe/fe-core/src/main/java/com/starrocks/backup/RestoreJob.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -46,6 +46,7 @@
import com.google.common.collect.Table.Cell;
import com.google.gson.annotations.SerializedName;
import com.starrocks.analysis.BrokerDesc;
import com.starrocks.analysis.TableName;
import com.starrocks.backup.BackupJobInfo.BackupIndexInfo;
import com.starrocks.backup.BackupJobInfo.BackupPartitionInfo;
import com.starrocks.backup.BackupJobInfo.BackupPhysicalPartitionInfo;
Expand Down Expand Up @@ -76,8 +77,10 @@
import com.starrocks.catalog.Table;
import com.starrocks.catalog.Tablet;
import com.starrocks.catalog.TabletMeta;
import com.starrocks.catalog.View;
import com.starrocks.common.AnalysisException;
import com.starrocks.common.Config;
import com.starrocks.common.DdlException;
import com.starrocks.common.MarkedCountDownLatch;
import com.starrocks.common.Pair;
import com.starrocks.common.UserException;
Expand All @@ -87,7 +90,10 @@
import com.starrocks.fs.HdfsUtil;
import com.starrocks.metric.MetricRepo;
import com.starrocks.persist.ColocatePersistInfo;
import com.starrocks.qe.ConnectContext;
import com.starrocks.server.GlobalStateMgr;
import com.starrocks.sql.ast.CreateViewStmt;
import com.starrocks.sql.parser.NodePosition;
import com.starrocks.task.AgentBatchTask;
import com.starrocks.task.AgentTask;
import com.starrocks.task.AgentTaskExecutor;
Expand Down Expand Up @@ -191,7 +197,7 @@ public enum RestoreJobState {
private AgentBatchTask batchTask;

boolean enableColocateRestore = Config.enable_colocate_restore;

public RestoreJob() {
super(JobType.RESTORE);
}
Expand Down Expand Up @@ -484,11 +490,17 @@ private void checkAndPrepareMeta() {
continue;
}

if (!tbl.isNativeTableOrMaterializedView()) {
status = new Status(ErrCode.COMMON_ERROR, "Only support restore OLAP table: " + tbl.getName());
if (!tbl.isSupportBackupRestore()) {
status = new Status(ErrCode.UNSUPPORTED,
"Table: " + tbl.getName() +
" can not support backup restore, type: {}" + tbl.getType());
return;
}

if (tbl.isOlapView()) {
continue;
}

OlapTable olapTbl = (OlapTable) tbl;
if (olapTbl.getState() != OlapTableState.NORMAL) {
status = new Status(ErrCode.COMMON_ERROR,
Expand Down Expand Up @@ -535,9 +547,10 @@ private void checkAndPrepareMeta() {

tblInfo.checkAndRecoverAutoIncrementId(localTbl);
// table already exist, check schema
if (!localTbl.isNativeTableOrMaterializedView()) {
status = new Status(ErrCode.COMMON_ERROR,
"Only support retore olap table: " + localTbl.getName());
if (!localTbl.isSupportBackupRestore()) {
status = new Status(ErrCode.UNSUPPORTED,
"Table: " + localTbl.getName() +
" can not support backup restore, type: {}" + localTbl.getType());
return;
}
OlapTable localOlapTbl = (OlapTable) localTbl;
Expand Down Expand Up @@ -762,6 +775,14 @@ private void checkAndPrepareMeta() {
return;
}

// add all restored olap view into globalStateMgr
List<View> restoredOlapViews = backupMeta.getTables().values().stream().filter(Table::isOlapView)
.map(x -> (View) x).collect(Collectors.toList());
addRestoreOlapView(restoredOlapViews);
if (!status.ok()) {
return;
}

LOG.info("finished to prepare meta. begin to make snapshot. {}", this);

// begin to make snapshots for all replicas
Expand Down Expand Up @@ -819,6 +840,38 @@ protected void sendCreateReplicaTasks() {
}
}

protected void addRestoreOlapView(List<View> restoredOlapViews) {
Database db = globalStateMgr.getLocalMetastore().getDb(dbId);

ConnectContext context = new ConnectContext();
context.setDatabase(db.getFullName());
context.setGlobalStateMgr(globalStateMgr);
context.setStartTime();
context.setThreadLocalInfo();

for (View restoredOlapView : restoredOlapViews) {
Table localTbl = db.getTable(restoredOlapView.getId());
if (localTbl != null && !localTbl.isOlapView()) {
status = new Status(ErrCode.BAD_REPLACE,
"Table: " + localTbl.getName() + " has existed and it is not a View");
return;
}

CreateViewStmt stmt = new CreateViewStmt(false, true, new TableName(db.getFullName(), restoredOlapView.getName()),
Lists.newArrayList(), restoredOlapView.getComment(), restoredOlapView.getQueryStatement(), NodePosition.ZERO);
stmt.setColumns(restoredOlapView.getColumns());
stmt.setInlineViewDef(restoredOlapView.getInlineViewDef());
context.getSessionVariable().setSqlMode(restoredOlapView.getSqlMode());
try {
GlobalStateMgr.getCurrentState().createView(stmt);
} catch (DdlException e) {
status = new Status(ErrCode.COMMON_ERROR,
"Failed to create view for restore. err message: " + e.getMessage());
return;
}
}
}

protected void addRestorePartitionsAndTables(Database db) {
db.writeLock();
try {
Expand Down Expand Up @@ -1100,6 +1153,10 @@ private void replayCheckAndPrepareMeta() {
db.writeUnlock();
}

List<View> restoredOlapViews = backupMeta.getTables().values().stream().filter(Table::isOlapView)
.map(x -> (View) x).collect(Collectors.toList());
addRestoreOlapView(restoredOlapViews);

LOG.info("replay check and prepare meta. {}", this);
}

Expand Down Expand Up @@ -1461,7 +1518,7 @@ private Status allTabletCommitted(boolean isReplay) {
tblInfo.name);
continue;
}
if (!tbl.isNativeTableOrMaterializedView()) {
if (!tbl.isSupportBackupRestore()) {
continue;
}
LOG.info("do post actions for table : {}", tbl.getName());
Expand Down Expand Up @@ -1692,7 +1749,7 @@ private void setTableStateToNormal(Database db) {
continue;
}

if (!tbl.isNativeTableOrMaterializedView()) {
if (!tbl.isSupportBackupRestore() || tbl.isOlapView()) {
continue;
}

Expand Down
3 changes: 2 additions & 1 deletion fe/fe-core/src/main/java/com/starrocks/backup/Status.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -31,7 +31,8 @@ public enum ErrCode {
BAD_CONNECTION,
COMMON_ERROR,
OLAP_VERSION_ALREADY_MERGED,
UNSUPPORTED
UNSUPPORTED,
BAD_REPLACE
}

@SerializedName("ec")
Expand Down
4 changes: 4 additions & 0 deletions fe/fe-core/src/main/java/com/starrocks/catalog/Table.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -825,4 +825,8 @@ public boolean isTable() {
!type.equals(TableType.VIEW) &&
!type.equals(TableType.HIVE_VIEW);
}

public boolean isSupportBackupRestore() {
return isOlapTableOrMaterializedView() || isOlapView();
}
}
4 changes: 4 additions & 0 deletions fe/fe-core/src/main/java/com/starrocks/catalog/View.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -136,6 +136,10 @@ public String getInlineViewDef() {
return inlineViewDef;
}

public long getSqlMode() {
return sqlMode;
}

/**
* Initializes the originalViewDef, inlineViewDef, and queryStmt members
* by parsing the expanded view definition SQL-string.
Expand Down
Loading

0 comments on commit 13d33f1

Please sign in to comment.