From 2062c48761298db7d8a28d8cf3ca75f8904928f3 Mon Sep 17 00:00:00 2001 From: walter Date: Fri, 1 Nov 2024 16:18:49 +0800 Subject: [PATCH] [feature](binlog) Wrap rename table/column binlog in BarrierLog (#43033) This PR is part of #37031, #39782 --- .../apache/doris/binlog/BinlogManager.java | 20 +++++++++++++++++++ .../org/apache/doris/persist/EditLog.java | 10 ++++++++-- .../org/apache/doris/persist/TableInfo.java | 18 +++++++++++++++++ .../doris/persist/TableRenameColumnInfo.java | 4 ++++ gensrc/thrift/FrontendService.thrift | 2 ++ 5 files changed, 52 insertions(+), 2 deletions(-) diff --git a/fe/fe-core/src/main/java/org/apache/doris/binlog/BinlogManager.java b/fe/fe-core/src/main/java/org/apache/doris/binlog/BinlogManager.java index 350a6c70917fb7..47b0bb3c7670f5 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/binlog/BinlogManager.java +++ b/fe/fe-core/src/main/java/org/apache/doris/binlog/BinlogManager.java @@ -32,6 +32,8 @@ import org.apache.doris.persist.ModifyTablePropertyOperationLog; import org.apache.doris.persist.ReplacePartitionOperationLog; import org.apache.doris.persist.TableAddOrDropColumnsInfo; +import org.apache.doris.persist.TableInfo; +import org.apache.doris.persist.TableRenameColumnInfo; import org.apache.doris.persist.TruncateTableInfo; import org.apache.doris.thrift.TBinlog; import org.apache.doris.thrift.TBinlogType; @@ -355,6 +357,24 @@ public Pair getBinlogLag(long dbId, long tableId, long prevCommit } } + public void addTableRename(TableInfo info, long commitSeq) { + long dbId = info.getDbId(); + long tableId = info.getTableId(); + TBinlogType type = TBinlogType.RENAME_TABLE; + String data = info.toJson(); + BarrierLog log = new BarrierLog(dbId, tableId, type, data); + addBarrierLog(log, commitSeq); + } + + public void addColumnRename(TableRenameColumnInfo info, long commitSeq) { + long dbId = info.getDbId(); + long tableId = info.getTableId(); + TBinlogType type = TBinlogType.RENAME_COLUMN; + String data = info.toJson(); + BarrierLog log = new BarrierLog(dbId, tableId, type, data); + addBarrierLog(log, commitSeq); + } + // get the dropped partitions of the db. public List getDroppedPartitions(long dbId) { lock.readLock().lock(); diff --git a/fe/fe-core/src/main/java/org/apache/doris/persist/EditLog.java b/fe/fe-core/src/main/java/org/apache/doris/persist/EditLog.java index 186d0ff9536eb4..1f29e8a08371c8 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/persist/EditLog.java +++ b/fe/fe-core/src/main/java/org/apache/doris/persist/EditLog.java @@ -302,6 +302,7 @@ public static void loadJournal(Env env, Long logId, JournalEntity journal) { case OperationType.OP_RENAME_TABLE: { TableInfo info = (TableInfo) journal.getData(); env.replayRenameTable(info); + env.getBinlogManager().addTableRename(info, logId); break; } case OperationType.OP_MODIFY_VIEW_DEF: { @@ -317,6 +318,7 @@ public static void loadJournal(Env env, Long logId, JournalEntity journal) { case OperationType.OP_RENAME_COLUMN: { TableRenameColumnInfo info = (TableRenameColumnInfo) journal.getData(); env.replayRenameColumn(info); + Env.getCurrentEnv().getBinlogManager().addColumnRename(info, logId); break; } case OperationType.OP_BACKUP_JOB: { @@ -1521,7 +1523,9 @@ public void logDatabaseRename(DatabaseInfo databaseInfo) { } public void logTableRename(TableInfo tableInfo) { - logEdit(OperationType.OP_RENAME_TABLE, tableInfo); + long logId = logEdit(OperationType.OP_RENAME_TABLE, tableInfo); + LOG.info("log table rename, logId : {}, infos: {}", logId, tableInfo); + Env.getCurrentEnv().getBinlogManager().addTableRename(tableInfo, logId); } public void logModifyViewDef(AlterViewInfo alterViewInfo) { @@ -1537,7 +1541,9 @@ public void logPartitionRename(TableInfo tableInfo) { } public void logColumnRename(TableRenameColumnInfo info) { - logEdit(OperationType.OP_RENAME_COLUMN, info); + long logId = logEdit(OperationType.OP_RENAME_COLUMN, info); + LOG.info("log column rename, logId : {}, infos: {}", logId, info); + Env.getCurrentEnv().getBinlogManager().addColumnRename(info, logId); } public void logAddBroker(BrokerMgr.ModifyBrokerInfo info) { diff --git a/fe/fe-core/src/main/java/org/apache/doris/persist/TableInfo.java b/fe/fe-core/src/main/java/org/apache/doris/persist/TableInfo.java index 2d08a8a43fa9df..69360a0e08b67a 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/persist/TableInfo.java +++ b/fe/fe-core/src/main/java/org/apache/doris/persist/TableInfo.java @@ -19,6 +19,9 @@ import org.apache.doris.common.io.Text; import org.apache.doris.common.io.Writable; +import org.apache.doris.persist.gson.GsonUtils; + +import com.google.gson.annotations.SerializedName; import java.io.DataInput; import java.io.DataOutput; @@ -26,13 +29,20 @@ public class TableInfo implements Writable { + @SerializedName("db") private long dbId; + @SerializedName("tb") private long tableId; + @SerializedName("ind") private long indexId; + @SerializedName("p") private long partitionId; + @SerializedName("nT") private String newTableName; + @SerializedName("nR") private String newRollupName; + @SerializedName("nP") private String newPartitionName; public TableInfo() { @@ -124,4 +134,12 @@ public static TableInfo read(DataInput in) throws IOException { info.readFields(in); return info; } + + public String toJson() { + return GsonUtils.GSON.toJson(this); + } + + public static TableInfo fromJson(String json) { + return GsonUtils.GSON.fromJson(json, TableInfo.class); + } } diff --git a/fe/fe-core/src/main/java/org/apache/doris/persist/TableRenameColumnInfo.java b/fe/fe-core/src/main/java/org/apache/doris/persist/TableRenameColumnInfo.java index eafdb943e1164b..9434f7302c0179 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/persist/TableRenameColumnInfo.java +++ b/fe/fe-core/src/main/java/org/apache/doris/persist/TableRenameColumnInfo.java @@ -72,6 +72,10 @@ public Map getIndexIdToSchemaVersion() { return indexIdToSchemaVersion; } + public String toJson() { + return GsonUtils.GSON.toJson(this); + } + @Override public void write(DataOutput out) throws IOException { Text.writeString(out, GsonUtils.GSON.toJson(this)); diff --git a/gensrc/thrift/FrontendService.thrift b/gensrc/thrift/FrontendService.thrift index 2258fff8fdf177..99b5a3a420de77 100644 --- a/gensrc/thrift/FrontendService.thrift +++ b/gensrc/thrift/FrontendService.thrift @@ -1137,6 +1137,8 @@ enum TBinlogType { MODIFY_PARTITIONS = 11, REPLACE_PARTITIONS = 12, TRUNCATE_TABLE = 13, + RENAME_TABLE = 14, + RENAME_COLUMN = 15, } struct TBinlog {