[Feature](fe) Add admin set partition version statement

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

@@ -0,0 +1,67 @@
+---
+{
+    "title": "ADMIN-SET-PARTITION-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-PARTITION-VERSION
+
+### Name
+
+ADMIN SET PARTITION VERSION
+
+### Description
+
+This statement is used to set the version of the specified partition.
+
+In certain cases, the version of the partition in the metadata may not be consistent with the version of the actual replica. This command can manually set the version of the partition in the metadata.
+
+grammar:
+
+```sql
+ADMIN SET TABLE table_name PARTITION VERSION
+        PROPERTIES ("key" = "value", ...);
+```
+
+The following properties are currently supported:
+
+1. "partition_id": Required. Specify a Partition Id.
+2. "visible_version": Required. Specify Version.
+
+> Note:
+>
+> It is necessary to first confirm the version of the actual replica on the Be before set the version of the partition. This command is generally only used for emergency troubleshooting, please proceed with caution.
+
+### Example
+
+1. Set the version of partition 1769152 to 100.
+
+```sql
+ADMIN SET TABLE tbl1 PARTITION VERSION PROPERTIES("partition_id" = "1769152", "visible_version" = "100");
+```
+
+### Keywords
+
+    ADMIN, SET, PARTITION, VERSION
+
+### Best Practice

docs/sidebars.json

@@ -812,6 +812,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-PARTITION-VERSION",
                                 "sql-manual/sql-reference/Database-Administration-Statements/ADMIN-SET-TABLE-STATUS",
                                 "sql-manual/sql-reference/Database-Administration-Statements/ADMIN-SHOW-REPLICA-DISTRIBUTION",
                                 "sql-manual/sql-reference/Database-Administration-Statements/ADMIN-SHOW-REPLICA-STATUS",

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

@@ -0,0 +1,67 @@
+---
+{
+    "title": "ADMIN-SET-PARTITION-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-PARTITION-VERSION
+
+### Name
+
+ADMIN SET PARTITION VERSION
+
+### Description
+
+该语句用于手动改变指定分区的可见版本。
+
+在某些特殊情况下，元数据中分区的版本有可能和实际副本的版本不一致，该命令可手动改变元数据中分区的版本。
+
+语法：
+
+```sql
+ADMIN SET TABLE table_name PARTITION VERSION
+        PROPERTIES ("key" = "value", ...);
+```
+
+目前支持如下属性：
+
+1. "partition_id"：必需。指定一个 Partition Id.
+2. "visible_version"：必需。指定 Version.
+
+> 注意：
+>
+>  设置分区的版本需要先确认Be机器上实际副本的版本，此命令一般只用于紧急故障修复，请谨慎操作。
+
+### Example
+
+1. 设置 partition 1769152 在 FE 元数据上的版本为 100。
+
+```sql
+ADMIN SET TABLE tbl1 PARTITION VERSION PROPERTIES("partition_id" = "1769152", "visible_version" = "100");
+```
+
+### Keywords
+
+    ADMIN, SET, PARTITION, VERSION
+
+### Best Practice

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

@@ -7221,6 +7221,10 @@ admin_stmt ::=
     {:
         RESULT = new AdminCleanTrashStmt(null);
     :}
+    | KW_ADMIN KW_SET KW_TABLE table_name:name KW_PARTITION KW_VERSION opt_properties:properties
+    {:
+        RESULT = new AdminSetPartitionVersionStmt(name, properties);
+    :}
     | KW_ADMIN KW_DIAGNOSE KW_TABLET INTEGER_LITERAL:tabletId
     {:
         RESULT = new AdminDiagnoseTabletStmt(tabletId);

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

@@ -0,0 +1,93 @@
+// 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.common.util.PropertyAnalyzer;
+import org.apache.doris.common.util.Util;
+import org.apache.doris.mysql.privilege.PrivPredicate;
+import org.apache.doris.qe.ConnectContext;
+
+import java.util.Map;
+
+// Modify version of specified partition. Only used in emergency.
+/*
+ *  admin set table db.tbl partition version properties ("key" = "val", ..);
+ *      "partition_id" = "20010",
+ *      "visible_version" = "101"
+ */
+public class AdminSetPartitionVersionStmt extends DdlStmt {
+    private long partitionId = -1;
+    private long visibleVersion = -1;
+    private final TableName tableName;
+    private final Map<String, String> properties;
+
+    public AdminSetPartitionVersionStmt(TableName tableName, Map<String, String> properties) {
+        this.tableName = tableName;
+        this.properties = properties;
+    }
+
+    public String getDatabase() {
+        return tableName.getDb();
+    }
+
+    public String getTable() {
+        return tableName.getTbl();
+    }
+
+    public Long getPartitionId() {
+        return partitionId;
+    }
+
+    public Long getVisibleVersion() {
+        return visibleVersion;
+    }
+
+    @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");
+        }
+
+        tableName.analyze(analyzer);
+        Util.prohibitExternalCatalog(tableName.getCtl(), this.getClass().getSimpleName());
+
+        checkProperties();
+    }
+
+    private void checkProperties() throws AnalysisException {
+        partitionId = PropertyAnalyzer.analyzePartitionId(properties);
+        if (partitionId == -1) {
+            throw new AnalysisException("Should specify 'partition_id' property.");
+        }
+        visibleVersion = PropertyAnalyzer.analyzeVisibleVersion(properties);
+        if (visibleVersion == -1) {
+            throw new AnalysisException("Should specify 'visible_version' property.");
+        }
+        if (properties != null && !properties.isEmpty()) {
+            throw new AnalysisException("Unknown properties: " + properties.keySet());
+        }
+    }
+}

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

@@ -206,6 +206,15 @@ default T getTableOrDdlException(long tableId) throws DdlException {
         return getTableOrException(tableId, t -> new DdlException(ErrorCode.ERR_BAD_TABLE_ERROR.formatErrorMsg(t)));
     }
 
+    default T getTableOrDdlException(long tableId, TableIf.TableType tableType) throws DdlException {
+        T table = getTableOrDdlException(tableId);
+        if (table.getType() != tableType) {
+            throw new DdlException(
+                    "table type is not " + tableType + ", tableId=" + tableId + ", type=" + table.getType());
+        }
+        return table;
+    }
+
     default T getTableOrAnalysisException(String tableName) throws AnalysisException {
         return getTableOrException(tableName,
                 t -> new AnalysisException(ErrorCode.ERR_BAD_TABLE_ERROR.formatErrorMsg(t)));

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

@@ -30,6 +30,7 @@
 import org.apache.doris.analysis.AdminCleanTrashStmt;
 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.AdminSetTableStatusStmt;
 import org.apache.doris.analysis.AlterDatabasePropertyStmt;
@@ -192,6 +193,7 @@
 import org.apache.doris.persist.RefreshExternalTableInfo;
 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.SetTableStatusOperationLog;
 import org.apache.doris.persist.Storage;
@@ -5423,6 +5425,55 @@ public void cleanTrash(AdminCleanTrashStmt stmt) {
         }
     }
 
+    public void setPartitionVersion(AdminSetPartitionVersionStmt stmt) throws DdlException {
+        String database = stmt.getDatabase();
+        String table = stmt.getTable();
+        long partitionId = stmt.getPartitionId();
+        long visibleVersion = stmt.getVisibleVersion();
+        int setSuccess = setPartitionVersionInternal(database, table, partitionId, visibleVersion, false);
+        if (setSuccess == -1) {
+            throw new DdlException("Failed to set partition visible version to " + visibleVersion + ". " + "Partition "
+                    + partitionId + " not exists. Database " + database + ", Table " + table + ".");
+        }
+    }
+
+    public void replaySetPartitionVersion(SetPartitionVersionOperationLog log) throws DdlException {
+        int setSuccess = setPartitionVersionInternal(log.getDatabase(), log.getTable(),
+                log.getPartitionId(), log.getVisibleVersion(), true);
+        if (setSuccess == -1) {
+            LOG.warn("Failed to set partition visible version to {}. "
+                    + "Database {}, Table {}, Partition {} not exists.", log.getDatabase(), log.getTable(),
+                    log.getVisibleVersion(), log.getPartitionId());
+        }
+    }
+
+    public int setPartitionVersionInternal(String database, String table, long partitionId,
+                                           long visibleVersion, boolean isReplay) throws DdlException {
+        int result = -1;
+        Database db = getInternalCatalog().getDbOrDdlException(database);
+        OlapTable olapTable = db.getOlapTableOrDdlException(table);
+        olapTable.writeLockOrDdlException();
+        try {
+            Partition partition = olapTable.getPartition(partitionId);
+            if (partition != null) {
+                Long oldVersion = partition.getVisibleVersion();
+                partition.updateVisibleVersion(visibleVersion);
+                partition.setNextVersion(visibleVersion + 1);
+                result = 0;
+                if (!isReplay) {
+                    SetPartitionVersionOperationLog log = new SetPartitionVersionOperationLog(
+                            database, table, partitionId, visibleVersion);
+                    getEditLog().logSetPartitionVersion(log);
+                }
+                LOG.info("set partition {} visible version from {} to {}. Database {}, Table {}, is replay:"
+                        + " {}.", partitionId, oldVersion, visibleVersion, database, table, isReplay);
+            }
+        } finally {
+            olapTable.writeUnlock();
+        }
+        return result;
+    }
+
     public static boolean isStoredTableNamesLowerCase() {
         return GlobalVariable.lowerCaseTableNames == 1;
     }

fe/fe-core/src/main/java/org/apache/doris/common/util/PropertyAnalyzer.java

@@ -70,6 +70,8 @@ public class PropertyAnalyzer {
     public static final String PROPERTIES_VERSION_INFO = "version_info";
     // for restore
     public static final String PROPERTIES_SCHEMA_VERSION = "schema_version";
+    public static final String PROPERTIES_PARTITION_ID = "partition_id";
+    public static final String PROPERTIES_VISIBLE_VERSION = "visible_version";
 
     public static final String PROPERTIES_BF_COLUMNS = "bloom_filter_columns";
     public static final String PROPERTIES_BF_FPP = "bloom_filter_fpp";
@@ -427,6 +429,30 @@ public static int analyzeSchemaVersion(Map<String, String> properties) throws An
         return schemaVersion;
     }
 
+    private static Long getPropertyLong(Map<String, String> properties, String propertyId) throws AnalysisException {
+        long id = -1;
+        if (properties != null && properties.containsKey(propertyId)) {
+            String propertyIdStr = properties.get(propertyId);
+            try {
+                id = Long.parseLong(propertyIdStr);
+            } catch (Exception e) {
+                throw new AnalysisException("Invalid property long id: " + propertyIdStr);
+            }
+
+            properties.remove(propertyId);
+        }
+
+        return id;
+    }
+
+    public static Long analyzePartitionId(Map<String, String> properties) throws AnalysisException {
+        return getPropertyLong(properties, PROPERTIES_PARTITION_ID);
+    }
+
+    public static Long analyzeVisibleVersion(Map<String, String> properties) throws AnalysisException {
+        return getPropertyLong(properties, PROPERTIES_VISIBLE_VERSION);
+    }
+
     public static Set<String> analyzeBloomFilterColumns(Map<String, String> properties, List<Column> columns,
             KeysType keysType) throws AnalysisException {
         Set<String> bfColumns = null;

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

@@ -107,6 +107,7 @@
 import org.apache.doris.persist.ReplaceTableOperationLog;
 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.SetTableStatusOperationLog;
 import org.apache.doris.persist.TableAddOrDropColumnsInfo;
@@ -620,6 +621,11 @@ public void readFields(DataInput in) throws IOException {
                 isRead = true;
                 break;
             }
+            case OperationType.OP_SET_PARTITION_VERSION: {
+                data = SetPartitionVersionOperationLog.read(in);
+                isRead = true;
+                break;
+            }
             case OperationType.OP_DYNAMIC_PARTITION:
             case OperationType.OP_MODIFY_IN_MEMORY:
             case OperationType.OP_MODIFY_REPLICATION_NUM: