[feature](replica version) Add admin set replica version statement (#23706)

docs/en/docs/sql-manual/sql-reference/Database-Administration-Statements/ADMIN-SET-REPLICA-VERSION.md

@@ -0,0 +1,83 @@
+---
+{
+    "title": "ADMIN-SET-REPLICA-VERSION",
+    "language": "en"
+}
+---
+
+<!--
+Licensed to the Apache Software Foundation (ASF) under one
+or more contributor license agreements.  See the NOTICE file
+distributed with this work for additional information
+regarding copyright ownership.  The ASF licenses this file
+to you under the Apache License, Version 2.0 (the
+"License"); you may not use this file except in compliance
+with the License.  You may obtain a copy of the License at
+
+  http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing,
+software distributed under the License is distributed on an
+"AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+KIND, either express or implied.  See the License for the
+specific language governing permissions and limitations
+under the License.
+-->
+
+## ADMIN-SET-REPLICA-VERSION
+
+### Name
+
+ADMIN SET REPLICA VERSION
+
+### Description
+
+This statement is used to set the version, maximum success version, and maximum failure version of the specified replica.
+
+This command is currently only used to manually repair the replica version when the program is abnormal, so that the replica can recover from the abnormal state.
+
+grammar:
+
+```sql
+ADMIN SET REPLICA VERSION
+        PROPERTIES ("key" = "value", ...);
+```
+
+The following properties are currently supported:
+
+1. `tablet_id`: Required. Specify a Tablet Id.
+2. `backend_id`: Required. Specify Backend Id.
+3. `version`: Optional. Set the replica version.
+4. `last_success_version`: Optional. Set the replica max success version.
+5. `last_failed_version`: Optional. Set the replica max failed version.
+
+If the specified replica does not exist, it will be ignored.
+
+> Note:
+>
+> Modifying these values ​​may cause subsequent data reading and writing failures, resulting in data inconsistency. Please operate with caution!
+>
+> Record the original value before modifying it. After the modification is completed, verify the read and write of the table. If the read and write fail, please restore the original value! But recovery may fail!
+>
+> It is strictly prohibited to operate the tablet that is writing data!
+
+### Example
+
+ 1. Clear the replica failed version of tablet 10003 on BE 10001.
+
+       ```sql
+ADMIN SET REPLICA VERSION PROPERTIES("tablet_id" = "10003", "backend_id" = "10001", "last_failed_version" = "-1");       
+       ```
+
+2. Set the replica status of tablet 10003 on BE 10001 to ok.
+
+```sql
+ADMIN SET REPLICA VERSION PROPERTIES("tablet_id" = "10003", "backend_id" = "10001", "version" = "1004");
+```
+
+### Keywords
+
+    ADMIN, SET, REPLICA, VERSION
+
+### Best Practice
+

docs/sidebars.json

@@ -822,6 +822,7 @@
                                 "sql-manual/sql-reference/Database-Administration-Statements/INSTALL-PLUGIN",
                                 "sql-manual/sql-reference/Database-Administration-Statements/UNINSTALL-PLUGIN",
                                 "sql-manual/sql-reference/Database-Administration-Statements/ADMIN-SET-REPLICA-STATUS",
+                                "sql-manual/sql-reference/Database-Administration-Statements/ADMIN-SET-REPLICA-VERSION",
                                 "sql-manual/sql-reference/Database-Administration-Statements/ADMIN-SET-PARTITION-VERSION",
                                 "sql-manual/sql-reference/Database-Administration-Statements/ADMIN-SET-TABLE-STATUS",
                                 "sql-manual/sql-reference/Database-Administration-Statements/ADMIN-SHOW-REPLICA-DISTRIBUTION",

docs/zh-CN/docs/sql-manual/sql-reference/Database-Administration-Statements/ADMIN-SET-REPLICA-VERSION.md

@@ -0,0 +1,85 @@
+---
+{
+    "title": "ADMIN-SET-REPLICA-VERSION",
+    "language": "zh-CN"
+}
+---
+
+<!--
+Licensed to the Apache Software Foundation (ASF) under one
+or more contributor license agreements.  See the NOTICE file
+distributed with this work for additional information
+regarding copyright ownership.  The ASF licenses this file
+to you under the Apache License, Version 2.0 (the
+"License"); you may not use this file except in compliance
+with the License.  You may obtain a copy of the License at
+
+  http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing,
+software distributed under the License is distributed on an
+"AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+KIND, either express or implied.  See the License for the
+specific language governing permissions and limitations
+under the License.
+-->
+
+## ADMIN-SET-REPLICA-VERSION
+
+### Name
+
+ADMIN SET REPLICA VERSION
+
+### Description
+
+该语句用于设置指定副本的版本、最大成功版本、最大失败版本。
+
+该命令目前仅用于在程序异常情况下，手动修复副本的版本，从而使得副本从异常状态恢复过来。
+
+语法：
+
+```sql
+ADMIN SET REPLICA VERSION
+        PROPERTIES ("key" = "value", ...);
+```
+
+ 目前支持如下属性：
+
+1. `tablet_id`：必需。指定一个 Tablet Id.
+2. `backend_id`：必需。指定 Backend Id.
+3. `version`：可选。设置副本的版本.
+4. `last_success_version`：可选。设置副本的最大成功版本.
+5. `last_failed_version`：可选。设置副本的最大失败版本。
+
+
+如果指定的副本不存在，则会被忽略。
+
+> 注意：
+>
+>  修改这几个数值，可能会导致后面数据读写失败，造成数据不一致，请谨慎操作！
+> 
+>   修改之前先记录原来的值。修改完毕之后，对表进行读写验证，如果读写失败，请恢复原来的值！但可能会恢复失败！
+> 
+>   严禁对正在写入数据的tablet进行操作 ！
+
+
+### Example
+
+ 1. 清除 tablet 10003 在 BE 10001 上的副本状态失败标志。
+
+```sql
+ADMIN SET REPLICA VERSION PROPERTIES("tablet_id" = "10003", "backend_id" = "10001", "last_failed_version" = "-1");
+```
+
+2. 设置 tablet 10003 在 BE 10001 上的副本版本号为 1004。
+
+```sql
+ADMIN SET REPLICA VERSION PROPERTIES("tablet_id" = "10003", "backend_id" = "10001", "version" = "1004");
+```
+
+### Keywords
+
+    ADMIN, SET, REPLICA, VERSION
+
+### Best Practice
+

fe/fe-core/src/main/cup/sql_parser.cup

@@ -7236,6 +7236,10 @@ admin_stmt ::=
     {:
         RESULT = new AdminSetReplicaStatusStmt(prop);
     :}
+    | KW_ADMIN KW_SET KW_REPLICA KW_VERSION KW_PROPERTIES LPAREN key_value_map:prop RPAREN
+    {:
+        RESULT = new AdminSetReplicaVersionStmt(prop);
+    :}
     | KW_ADMIN KW_REPAIR KW_TABLE base_table_ref:table_ref
     {:
         RESULT = new AdminRepairTableStmt(table_ref);

fe/fe-core/src/main/java/org/apache/doris/analysis/AdminSetReplicaVersionStmt.java

@@ -0,0 +1,151 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package org.apache.doris.analysis;
+
+import org.apache.doris.catalog.Env;
+import org.apache.doris.common.AnalysisException;
+import org.apache.doris.common.ErrorCode;
+import org.apache.doris.common.ErrorReport;
+import org.apache.doris.common.UserException;
+import org.apache.doris.mysql.privilege.PrivPredicate;
+import org.apache.doris.qe.ConnectContext;
+
+import java.util.Map;
+
+/*
+ *  admin set replicas status properties ("key" = "val", ..);
+ *  Required:
+ *      "tablet_id" = "10010",
+ *      "backend_id" = "10001",
+ *  Optional:
+ *      "version" = "100",
+ *      "last_success_version" = "100",
+ *      "last_failed_version" = "-1",
+ */
+public class AdminSetReplicaVersionStmt extends DdlStmt {
+
+    public static final String TABLET_ID = "tablet_id";
+    public static final String BACKEND_ID = "backend_id";
+    public static final String VERSION = "version";
+    public static final String LAST_SUCCESS_VERSION = "last_success_version";
+    public static final String LAST_FAILED_VERSION = "last_failed_version";
+
+    private Map<String, String> properties;
+    private long tabletId = -1;
+    private long backendId = -1;
+    private Long version = null;
+    private Long lastSuccessVersion = null;
+    private Long lastFailedVersion = null;
+
+    public AdminSetReplicaVersionStmt(Map<String, String> properties) {
+        this.properties = properties;
+    }
+
+    @Override
+    public void analyze(Analyzer analyzer) throws AnalysisException, UserException {
+        super.analyze(analyzer);
+
+        // check auth
+        if (!Env.getCurrentEnv().getAccessManager().checkGlobalPriv(ConnectContext.get(), PrivPredicate.ADMIN)) {
+            ErrorReport.reportAnalysisException(ErrorCode.ERR_SPECIFIC_ACCESS_DENIED_ERROR, "ADMIN");
+        }
+
+        checkProperties();
+    }
+
+    private void checkProperties() throws AnalysisException {
+        for (Map.Entry<String, String> entry : properties.entrySet()) {
+            String key = entry.getKey();
+            String val = entry.getValue();
+
+            if (key.equalsIgnoreCase(TABLET_ID)) {
+                try {
+                    tabletId = Long.valueOf(val);
+                } catch (NumberFormatException e) {
+                    throw new AnalysisException("Invalid tablet id format: " + val);
+                }
+            } else if (key.equalsIgnoreCase(BACKEND_ID)) {
+                try {
+                    backendId = Long.valueOf(val);
+                } catch (NumberFormatException e) {
+                    throw new AnalysisException("Invalid backend id format: " + val);
+                }
+            } else if (key.equalsIgnoreCase(VERSION)) {
+                try {
+                    version = Long.valueOf(val);
+                } catch (NumberFormatException e) {
+                    throw new AnalysisException("Invalid version format: " + val);
+                }
+                if (version <= 0) {
+                    throw new AnalysisException("Required version > 0");
+                }
+            } else if (key.equalsIgnoreCase(LAST_SUCCESS_VERSION)) {
+                try {
+                    lastSuccessVersion = Long.valueOf(val);
+                } catch (NumberFormatException e) {
+                    throw new AnalysisException("Invalid last success version format: " + val);
+                }
+                if (lastSuccessVersion <= 0) {
+                    throw new AnalysisException("Required last success version > 0");
+                }
+            } else if (key.equalsIgnoreCase(LAST_FAILED_VERSION)) {
+                try {
+                    lastFailedVersion = Long.valueOf(val);
+                } catch (NumberFormatException e) {
+                    throw new AnalysisException("Invalid last failed version format: " + val);
+                }
+                if (lastFailedVersion <= 0 && lastFailedVersion != -1) {
+                    throw new AnalysisException("Required last failed version > 0 or == -1");
+                }
+            } else {
+                throw new AnalysisException("Unknown property: " + key);
+            }
+        }
+
+        if (tabletId == -1 || backendId == -1
+                || (version == null && lastSuccessVersion == null && lastFailedVersion == null)) {
+            throw new AnalysisException("Should add following properties: TABLET_ID, BACKEND_ID, "
+                    + "VERSION, LAST_SUCCESS_VERSION, LAST_FAILED_VERSION");
+        }
+    }
+
+    public long getTabletId() {
+        return tabletId;
+    }
+
+    public long getBackendId() {
+        return backendId;
+    }
+
+    public Long getVersion() {
+        return version;
+    }
+
+    public Long getLastSuccessVersion() {
+        return lastSuccessVersion;
+    }
+
+    public Long getLastFailedVersion() {
+        return lastFailedVersion;
+    }
+
+    @Override
+    public RedirectStatus getRedirectStatus() {
+        return RedirectStatus.FORWARD_WITH_SYNC;
+    }
+}

fe/fe-core/src/main/java/org/apache/doris/catalog/Env.java

@@ -32,6 +32,7 @@ import org.apache.doris.analysis.AdminCompactTableStmt;
 import org.apache.doris.analysis.AdminSetConfigStmt;
 import org.apache.doris.analysis.AdminSetPartitionVersionStmt;
 import org.apache.doris.analysis.AdminSetReplicaStatusStmt;
+import org.apache.doris.analysis.AdminSetReplicaVersionStmt;
 import org.apache.doris.analysis.AdminSetTableStatusStmt;
 import org.apache.doris.analysis.AlterDatabasePropertyStmt;
 import org.apache.doris.analysis.AlterDatabaseQuotaStmt;
@@ -195,6 +196,7 @@ import org.apache.doris.persist.ReplacePartitionOperationLog;
 import org.apache.doris.persist.ReplicaPersistInfo;
 import org.apache.doris.persist.SetPartitionVersionOperationLog;
 import org.apache.doris.persist.SetReplicaStatusOperationLog;
+import org.apache.doris.persist.SetReplicaVersionOperationLog;
 import org.apache.doris.persist.SetTableStatusOperationLog;
 import org.apache.doris.persist.Storage;
 import org.apache.doris.persist.StorageInfo;
@@ -5417,6 +5419,56 @@ public class Env {
         }
     }
 
+    // Set specified replica's version. If replica does not exist, just ignore it.
+    public void setReplicaVersion(AdminSetReplicaVersionStmt stmt) throws MetaNotFoundException {
+        long tabletId = stmt.getTabletId();
+        long backendId = stmt.getBackendId();
+        Long version = stmt.getVersion();
+        Long lastSuccessVersion = stmt.getLastSuccessVersion();
+        Long lastFailedVersion = stmt.getLastFailedVersion();
+        long updateTime = System.currentTimeMillis();
+        setReplicaVersionInternal(tabletId, backendId, version, lastSuccessVersion, lastFailedVersion,
+                updateTime, false);
+    }
+
+    public void replaySetReplicaVersion(SetReplicaVersionOperationLog log) throws MetaNotFoundException {
+        setReplicaVersionInternal(log.getTabletId(), log.getBackendId(), log.getVersion(),
+                log.getLastSuccessVersion(), log.getLastFailedVersion(), log.getUpdateTime(), true);
+    }
+
+    private void setReplicaVersionInternal(long tabletId, long backendId, Long version, Long lastSuccessVersion,
+            Long lastFailedVersion, long updateTime, boolean isReplay)
+            throws MetaNotFoundException {
+        try {
+            TabletMeta meta = tabletInvertedIndex.getTabletMeta(tabletId);
+            if (meta == null) {
+                throw new MetaNotFoundException("tablet does not exist");
+            }
+            Database db = getInternalCatalog().getDbOrMetaException(meta.getDbId());
+            Table table = db.getTableOrMetaException(meta.getTableId());
+            table.writeLockOrMetaException();
+            try {
+                Replica replica = tabletInvertedIndex.getReplica(tabletId, backendId);
+                if (replica == null) {
+                    throw new MetaNotFoundException("replica does not exist on backend, beId=" + backendId);
+                }
+                replica.adminUpdateVersionInfo(version, lastFailedVersion, lastSuccessVersion, updateTime);
+                if (!isReplay) {
+                    SetReplicaVersionOperationLog log = new SetReplicaVersionOperationLog(backendId, tabletId,
+                            version, lastSuccessVersion, lastFailedVersion, updateTime);
+                    getEditLog().logSetReplicaVersion(log);
+                }
+                LOG.info("set replica {} of tablet {} on backend {} as version {}, last success version {} ,"
+                        + ", last failed version {}, update time {}. is replay: {}", replica.getId(), tabletId,
+                        backendId, version, lastSuccessVersion, lastFailedVersion, updateTime, isReplay);
+            } finally {
+                table.writeUnlock();
+            }
+        } catch (MetaNotFoundException e) {
+            throw new MetaNotFoundException("set replica version failed, tabletId=" + tabletId, e);
+        }
+    }
+
     public void eraseDatabase(long dbId, boolean needEditLog) {
         // remove jobs
         Env.getCurrentEnv().getLoadInstance().removeDbLoadJob(dbId);

fe/fe-core/src/main/java/org/apache/doris/catalog/Replica.java

@@ -298,6 +298,34 @@ public class Replica implements Writable {
         updateReplicaInfo(newVersion, lastFailedVersion, lastSuccessVersion, dataSize, remoteDataSize, rowCount);
     }
 
+    public synchronized void adminUpdateVersionInfo(Long version, Long lastFailedVersion, Long lastSuccessVersion,
+            long updateTime) {
+        if (version != null) {
+            this.version = version;
+        }
+        if (lastSuccessVersion != null) {
+            this.lastSuccessVersion = lastSuccessVersion;
+        }
+        if (lastFailedVersion != null) {
+            if (this.lastFailedVersion < lastFailedVersion) {
+                this.lastFailedTimestamp = updateTime;
+            }
+            this.lastFailedVersion = lastFailedVersion;
+        }
+        if (this.lastFailedVersion < this.version) {
+            this.lastFailedVersion = -1;
+            this.lastFailedTimestamp  = -1;
+            this.lastFailedVersionHash = 0;
+        }
+        if (this.lastFailedVersion > 0
+                && this.lastSuccessVersion > this.lastFailedVersion) {
+            this.lastSuccessVersion = this.version;
+        }
+        if (this.lastSuccessVersion < this.version) {
+            this.lastSuccessVersion = this.version;
+        }
+    }
+
     /* last failed version:  LFV
      * last success version: LSV
      * version:              V

fe/fe-core/src/main/java/org/apache/doris/journal/JournalEntity.java

@@ -109,6 +109,7 @@ import org.apache.doris.persist.ReplicaPersistInfo;
 import org.apache.doris.persist.RoutineLoadOperation;
 import org.apache.doris.persist.SetPartitionVersionOperationLog;
 import org.apache.doris.persist.SetReplicaStatusOperationLog;
+import org.apache.doris.persist.SetReplicaVersionOperationLog;
 import org.apache.doris.persist.SetTableStatusOperationLog;
 import org.apache.doris.persist.TableAddOrDropColumnsInfo;
 import org.apache.doris.persist.TableAddOrDropInvertedIndicesInfo;
@@ -630,6 +631,11 @@ public class JournalEntity implements Writable {
                 isRead = true;
                 break;
             }
+            case OperationType.OP_SET_REPLICA_VERSION: {
+                data = SetReplicaVersionOperationLog.read(in);
+                isRead = true;
+                break;
+            }
             case OperationType.OP_SET_PARTITION_VERSION: {
                 data = SetPartitionVersionOperationLog.read(in);
                 isRead = true;

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

@@ -821,6 +821,11 @@ public class EditLog {
                     env.replaySetReplicaStatus(log);
                     break;
                 }
+                case OperationType.OP_SET_REPLICA_VERSION: {
+                    SetReplicaVersionOperationLog log = (SetReplicaVersionOperationLog) journal.getData();
+                    env.replaySetReplicaVersion(log);
+                    break;
+                }
                 case OperationType.OP_REMOVE_ALTER_JOB_V2: {
                     RemoveAlterJobV2OperationLog log = (RemoveAlterJobV2OperationLog) journal.getData();
                     switch (log.getType()) {
@@ -1752,6 +1757,10 @@ public class EditLog {
         logEdit(OperationType.OP_SET_REPLICA_STATUS, log);
     }
 
+    public void logSetReplicaVersion(SetReplicaVersionOperationLog log) {
+        logEdit(OperationType.OP_SET_REPLICA_VERSION, log);
+    }
+
     public void logRemoveExpiredAlterJobV2(RemoveAlterJobV2OperationLog log) {
         logEdit(OperationType.OP_REMOVE_ALTER_JOB_V2, log);
     }

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

@@ -187,6 +187,9 @@ public class OperationType {
     public static final short OP_ADD_GLOBAL_FUNCTION = 132;
     public static final short OP_DROP_GLOBAL_FUNCTION = 133;
 
+    // modify database/table/tablet/replica meta
+    public static final short OP_SET_REPLICA_VERSION = 141;
+
     // routine load 200
     public static final short OP_CREATE_ROUTINE_LOAD_JOB = 200;
     public static final short OP_CHANGE_ROUTINE_LOAD_JOB = 201;

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

@@ -0,0 +1,124 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package org.apache.doris.persist;
+
+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;
+import java.io.IOException;
+
+public class SetReplicaVersionOperationLog implements Writable {
+
+    @SerializedName(value = "backendId")
+    private long backendId;
+    @SerializedName(value = "tabletId")
+    private long tabletId;
+    @SerializedName(value = "version")
+    private Long version = null;
+    @SerializedName(value = "lastSuccessVersion")
+    private Long lastSuccessVersion = null;
+    @SerializedName(value = "lastFailedVersion")
+    private Long lastFailedVersion = null;
+    @SerializedName(value = "updateTime")
+    private long updateTime;
+
+    public SetReplicaVersionOperationLog(long backendId, long tabletId, Long version,
+            Long lastSuccessVersion, Long lastFailedVersion, long updateTime) {
+        this.backendId = backendId;
+        this.tabletId = tabletId;
+        this.version = version;
+        this.lastSuccessVersion = lastSuccessVersion;
+        this.lastFailedVersion = lastFailedVersion;
+        this.updateTime = updateTime;
+    }
+
+    public long getTabletId() {
+        return tabletId;
+    }
+
+    public long getBackendId() {
+        return backendId;
+    }
+
+    public Long getVersion() {
+        return version;
+    }
+
+    public Long getLastSuccessVersion() {
+        return lastSuccessVersion;
+    }
+
+    public Long getLastFailedVersion() {
+        return lastFailedVersion;
+    }
+
+    public long getUpdateTime() {
+        return updateTime;
+    }
+
+    public static SetReplicaVersionOperationLog read(DataInput in) throws IOException {
+        String json = Text.readString(in);
+        return GsonUtils.GSON.fromJson(json, SetReplicaVersionOperationLog.class);
+    }
+
+    @Override
+    public boolean equals(Object obj) {
+        if (!(obj instanceof SetReplicaVersionOperationLog)) {
+            return false;
+        }
+
+        SetReplicaVersionOperationLog other = (SetReplicaVersionOperationLog) obj;
+        if (version == null) {
+            if (other.version != null) {
+                return false;
+            }
+        } else if (!version.equals(other.version)) {
+            return false;
+        }
+
+        if (lastSuccessVersion == null) {
+            if (other.lastSuccessVersion != null) {
+                return false;
+            }
+        } else if (!lastSuccessVersion.equals(other.lastSuccessVersion)) {
+            return false;
+        }
+
+        if (lastFailedVersion == null) {
+            if (other.lastFailedVersion != null) {
+                return false;
+            }
+        } else if (!lastFailedVersion.equals(other.lastFailedVersion)) {
+            return false;
+        }
+
+        return backendId == other.backendId && tabletId == other.tabletId
+                && updateTime == other.updateTime;
+    }
+
+    @Override
+    public void write(DataOutput out) throws IOException {
+        String json = GsonUtils.GSON.toJson(this);
+        Text.writeString(out, json);
+    }
+}

fe/fe-core/src/main/java/org/apache/doris/qe/DdlExecutor.java

@@ -27,6 +27,7 @@ import org.apache.doris.analysis.AdminRepairTableStmt;
 import org.apache.doris.analysis.AdminSetConfigStmt;
 import org.apache.doris.analysis.AdminSetPartitionVersionStmt;
 import org.apache.doris.analysis.AdminSetReplicaStatusStmt;
+import org.apache.doris.analysis.AdminSetReplicaVersionStmt;
 import org.apache.doris.analysis.AdminSetTableStatusStmt;
 import org.apache.doris.analysis.AlterCatalogNameStmt;
 import org.apache.doris.analysis.AlterCatalogPropertyStmt;
@@ -271,6 +272,8 @@ public class DdlExecutor {
             env.checkTablets((AdminCheckTabletsStmt) ddlStmt);
         } else if (ddlStmt instanceof AdminSetReplicaStatusStmt) {
             env.setReplicaStatus((AdminSetReplicaStatusStmt) ddlStmt);
+        } else if (ddlStmt instanceof AdminSetReplicaVersionStmt) {
+            env.setReplicaVersion((AdminSetReplicaVersionStmt) ddlStmt);
         } else if (ddlStmt instanceof AdminSetPartitionVersionStmt) {
             env.setPartitionVersion((AdminSetPartitionVersionStmt) ddlStmt);
         } else if (ddlStmt instanceof CreateResourceStmt) {

fe/fe-core/src/test/java/org/apache/doris/catalog/AdminStmtTest.java

@@ -18,12 +18,14 @@
 package org.apache.doris.catalog;
 
 import org.apache.doris.analysis.AdminSetReplicaStatusStmt;
+import org.apache.doris.analysis.AdminSetReplicaVersionStmt;
 import org.apache.doris.catalog.MaterializedIndex.IndexExtState;
 import org.apache.doris.catalog.Replica.ReplicaStatus;
 import org.apache.doris.common.AnalysisException;
 import org.apache.doris.common.Pair;
 import org.apache.doris.persist.SetPartitionVersionOperationLog;
 import org.apache.doris.persist.SetReplicaStatusOperationLog;
+import org.apache.doris.persist.SetReplicaVersionOperationLog;
 import org.apache.doris.utframe.TestWithFeService;
 
 import com.google.common.collect.Lists;
@@ -60,6 +62,16 @@ public class AdminStmtTest extends TestWithFeService {
                 + "PROPERTIES (\n"
                 + " \"replication_num\" = \"1\"\n"
                 + ");");
+        // for test set replica version
+        createTable("CREATE TABLE test.tbl3 (\n"
+                + "  `id` int(11) NULL COMMENT \"\",\n"
+                + "  `name` varchar(20) NULL\n"
+                + ") ENGINE=OLAP\n"
+                + "DUPLICATE KEY(`id`, `name`)\n"
+                + "DISTRIBUTED BY HASH(`id`) BUCKETS 3\n"
+                + "PROPERTIES (\n"
+                + " \"replication_num\" = \"1\"\n"
+                + ");");
     }
 
     @Test
@@ -102,6 +114,111 @@ public class AdminStmtTest extends TestWithFeService {
         Assertions.assertFalse(replica.isBad());
     }
 
+    @Test
+    public void testAdminSetReplicaVersion() throws Exception {
+        Database db = Env.getCurrentInternalCatalog().getDbNullable("default_cluster:test");
+        Assertions.assertNotNull(db);
+        OlapTable tbl = (OlapTable) db.getTableNullable("tbl3");
+        Assertions.assertNotNull(tbl);
+        // tablet id, backend id
+        List<Pair<Long, Long>> tabletToBackendList = Lists.newArrayList();
+        for (Partition partition : tbl.getPartitions()) {
+            for (MaterializedIndex index : partition.getMaterializedIndices(IndexExtState.VISIBLE)) {
+                for (Tablet tablet : index.getTablets()) {
+                    for (Replica replica : tablet.getReplicas()) {
+                        tabletToBackendList.add(Pair.of(tablet.getId(), replica.getBackendId()));
+                    }
+                }
+            }
+        }
+        Assertions.assertEquals(3, tabletToBackendList.size());
+        long tabletId = tabletToBackendList.get(0).first;
+        long backendId = tabletToBackendList.get(0).second;
+        Replica replica = Env.getCurrentInvertedIndex().getReplica(tabletId, backendId);
+
+        String adminStmt = "admin set replica version properties ('tablet_id' = '" + tabletId + "', 'backend_id' = '"
+                + backendId + "', 'version' = '10', 'last_failed_version' = '100');";
+        AdminSetReplicaVersionStmt stmt = (AdminSetReplicaVersionStmt) parseAndAnalyzeStmt(adminStmt);
+        Env.getCurrentEnv().setReplicaVersion(stmt);
+        Assertions.assertEquals(10L, replica.getVersion());
+        Assertions.assertEquals(10L, replica.getLastSuccessVersion());
+        Assertions.assertEquals(100L, replica.getLastFailedVersion());
+
+        adminStmt = "admin set replica version properties ('tablet_id' = '" + tabletId + "', 'backend_id' = '"
+                + backendId + "', 'version' = '50');";
+        stmt = (AdminSetReplicaVersionStmt) parseAndAnalyzeStmt(adminStmt);
+        Env.getCurrentEnv().setReplicaVersion(stmt);
+        Assertions.assertEquals(50L, replica.getVersion());
+        Assertions.assertEquals(50L, replica.getLastSuccessVersion());
+        Assertions.assertEquals(100L, replica.getLastFailedVersion());
+
+        adminStmt = "admin set replica version properties ('tablet_id' = '" + tabletId + "', 'backend_id' = '"
+                + backendId + "', 'version' = '200');";
+        stmt = (AdminSetReplicaVersionStmt) parseAndAnalyzeStmt(adminStmt);
+        Env.getCurrentEnv().setReplicaVersion(stmt);
+        Assertions.assertEquals(200L, replica.getVersion());
+        Assertions.assertEquals(200L, replica.getLastSuccessVersion());
+        Assertions.assertEquals(-1L, replica.getLastFailedVersion());
+
+        adminStmt = "admin set replica version properties ('tablet_id' = '" + tabletId + "', 'backend_id' = '"
+                + backendId + "', 'last_failed_version' = '300');";
+        stmt = (AdminSetReplicaVersionStmt) parseAndAnalyzeStmt(adminStmt);
+        Env.getCurrentEnv().setReplicaVersion(stmt);
+        Assertions.assertEquals(300L, replica.getLastFailedVersion());
+
+        adminStmt = "admin set replica version properties ('tablet_id' = '" + tabletId + "', 'backend_id' = '"
+                + backendId + "', 'last_failed_version' = '-1');";
+        stmt = (AdminSetReplicaVersionStmt) parseAndAnalyzeStmt(adminStmt);
+        Env.getCurrentEnv().setReplicaVersion(stmt);
+        Assertions.assertEquals(-1L, replica.getLastFailedVersion());
+    }
+
+    @Test
+    public void testSetReplicaVersionOperationLog() throws IOException, AnalysisException {
+        String fileName = "./SetReplicaVersionOperationLog";
+        Path path = Paths.get(fileName);
+        List<Long> versions = Lists.newArrayList(null, 10L, 1000L);
+        for (int i = 0; i < versions.size(); i++) {
+            for (int j = 0; j < versions.size(); j++) {
+                for (int k = 0; k < versions.size(); k++) {
+                    try {
+                        // 1. Write objects to file
+                        Files.createFile(path);
+                        DataOutputStream out = new DataOutputStream(Files.newOutputStream(path));
+
+                        Long version = versions.get(i);
+                        Long lastSuccessVersion = versions.get(j);
+                        Long lastFailedVersion = versions.get(k);
+                        if (version != null) {
+                            version = version + 1;
+                        }
+                        if (lastSuccessVersion != null) {
+                            lastSuccessVersion = lastSuccessVersion + 2;
+                        }
+                        if (lastFailedVersion != null) {
+                            lastFailedVersion = lastFailedVersion + 3;
+                        }
+                        SetReplicaVersionOperationLog log = new SetReplicaVersionOperationLog(123L, 567L, version,
+                                lastSuccessVersion, lastFailedVersion, 101112L);
+                        log.write(out);
+                        out.flush();
+                        out.close();
+
+                        // 2. Read objects from file
+                        DataInputStream in = new DataInputStream(Files.newInputStream(path));
+
+                        SetReplicaVersionOperationLog readLog = SetReplicaVersionOperationLog.read(in);
+                        Assertions.assertEquals(log, readLog);
+
+                        in.close();
+                    } finally {
+                        Files.deleteIfExists(path);
+                    }
+                }
+            }
+        }
+    }
+
     @Test
     public void testSetReplicaStatusOperationLog() throws IOException, AnalysisException {
         String fileName = "./SetReplicaStatusOperationLog";

regression-test/suites/version_p0/test_set_replica_version.groovy

@@ -0,0 +1,83 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+suite("test_set_replica_version") {
+    def tableName1 = "test_set_replica_version"
+    sql """ DROP TABLE IF EXISTS ${tableName1} """
+    sql """
+    CREATE TABLE ${tableName1} (
+       `id` int NOT NULL,
+       `version` int NOT NULL COMMENT '插入次数'
+    ) ENGINE=OLAP
+    DUPLICATE KEY(`id`)
+    COMMENT 'OLAP'
+    DISTRIBUTED BY HASH(`id`) BUCKETS 1
+    PROPERTIES 
+    (
+        "replication_allocation" = "tag.location.default: 1",
+        "disable_auto_compaction" = "true"
+    );
+    """
+
+    def res = sql """ show tablets from ${tableName1}; """
+    def tabletId = res[0][0].toString()
+    def backendId = res[0][2].toString()
+
+    def version = 10
+    def lastFailedVersion = 100
+    sql """ ADMIN SET REPLICA VERSION PROPERTIES (
+        "tablet_id" = "${tabletId}", "backend_id" = "${backendId}",
+        "version" = "${version}", "last_failed_version" = "${lastFailedVersion}"
+        ); """
+    res = sql """ show tablets from ${tableName1}; """
+    assertEquals(res[0][4].toString(), "10")
+    assertEquals(res[0][5].toString(), "10")
+    assertEquals(res[0][6].toString(), "100")
+
+    lastFailedVersion = -1
+    sql """ ADMIN SET REPLICA VERSION PROPERTIES (
+        "tablet_id" = "${tabletId}", "backend_id" = "${backendId}",
+        "last_failed_version" = "${lastFailedVersion}"
+        ); """
+    res = sql """ show tablets from ${tableName1}; """
+    assertEquals(res[0][4].toString(), "10")
+    assertEquals(res[0][5].toString(), "10")
+    assertEquals(res[0][6].toString(), "-1")
+
+    version = 20
+    lastFailedVersion = 100
+    sql """ ADMIN SET REPLICA VERSION PROPERTIES (
+        "tablet_id" = "${tabletId}", "backend_id" = "${backendId}",
+        "version" = "${version}", "last_failed_version" = "${lastFailedVersion}"
+        ); """
+    res = sql """ show tablets from ${tableName1}; """
+    assertEquals(res[0][4].toString(), "20")
+    assertEquals(res[0][5].toString(), "20")
+    assertEquals(res[0][6].toString(), "100")
+
+    version = 200
+    sql """ ADMIN SET REPLICA VERSION PROPERTIES (
+        "tablet_id" = "${tabletId}", "backend_id" = "${backendId}",
+        "version" = "${version}"
+        ); """
+    res = sql """ show tablets from ${tableName1}; """
+    assertEquals(res[0][4].toString(), "200")
+    assertEquals(res[0][5].toString(), "200")
+    assertEquals(res[0][6].toString(), "-1")
+
+    sql """ DROP TABLE IF EXISTS ${tableName1} """
+}