diff --git a/fe/fe-common/src/main/java/org/apache/doris/common/FeMetaVersion.java b/fe/fe-common/src/main/java/org/apache/doris/common/FeMetaVersion.java
index 9f5474fa01..00b40b9fc9 100644
--- a/fe/fe-common/src/main/java/org/apache/doris/common/FeMetaVersion.java
+++ b/fe/fe-common/src/main/java/org/apache/doris/common/FeMetaVersion.java
@@ -76,9 +76,11 @@ public final class FeMetaVersion {
     public static final int VERSION_127 = 127;
     // For statistics. Update rows, new partition loaded, AnalysisJobInfo and AnalysisTaskInfo
     public static final int VERSION_128 = 128;
+    // For table version
+    public static final int VERSION_129 = 129;
 
     // note: when increment meta version, should assign the latest version to VERSION_CURRENT
-    public static final int VERSION_CURRENT = VERSION_128;
+    public static final int VERSION_CURRENT = VERSION_129;
 
     // all logs meta version should >= the minimum version, so that we could remove many if clause, for example
     // if (FE_METAVERSION < VERSION_94) ...
diff --git a/fe/fe-core/src/main/java/org/apache/doris/catalog/Env.java b/fe/fe-core/src/main/java/org/apache/doris/catalog/Env.java
index 38b8546cfb..8c9c6ad052 100755
--- a/fe/fe-core/src/main/java/org/apache/doris/catalog/Env.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/catalog/Env.java
@@ -5428,11 +5428,13 @@ public class Env {
             }
         }
         olapTable.replaceTempPartitions(partitionNames, tempPartitionNames, isStrictRange, useTempPartitionName);
-
+        long version = olapTable.getNextVersion();
+        long versionTime = System.currentTimeMillis();
+        olapTable.updateVisibleVersionAndTime(version, versionTime);
         // write log
         ReplacePartitionOperationLog info =
                 new ReplacePartitionOperationLog(db.getId(), db.getFullName(), olapTable.getId(), olapTable.getName(),
-                        partitionNames, tempPartitionNames, isStrictRange, useTempPartitionName);
+                        partitionNames, tempPartitionNames, isStrictRange, useTempPartitionName, version, versionTime);
         editLog.logReplaceTempPartition(info);
         LOG.info("finished to replace partitions {} with temp partitions {} from table: {}", clause.getPartitionNames(),
                 clause.getTempPartitionNames(), olapTable.getName());
@@ -5450,6 +5452,8 @@ public class Env {
             olapTable.replaceTempPartitions(replaceTempPartitionLog.getPartitions(),
                     replaceTempPartitionLog.getTempPartitions(), replaceTempPartitionLog.isStrictRange(),
                     replaceTempPartitionLog.useTempPartitionName());
+            olapTable.updateVisibleVersionAndTime(replaceTempPartitionLog.getVersion(),
+                    replaceTempPartitionLog.getVersionTime());
         } catch (DdlException e) {
             throw new MetaNotFoundException(e);
         } finally {
diff --git a/fe/fe-core/src/main/java/org/apache/doris/catalog/OlapTable.java b/fe/fe-core/src/main/java/org/apache/doris/catalog/OlapTable.java
index 7df7690cf1..e9e4bbeab7 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/catalog/OlapTable.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/catalog/OlapTable.java
@@ -2502,6 +2502,28 @@ public class OlapTable extends Table implements MTMVRelatedTableIf {
         return tablets;
     }
 
+    // During `getNextVersion` and `updateVisibleVersionAndTime` period,
+    // the write lock on the table should be held continuously
+    public void updateVisibleVersionAndTime(long visibleVersion, long visibleVersionTime) {
+        LOG.info("updateVisibleVersionAndTime, tableName: {}, visibleVersion, {}, visibleVersionTime: {}", name,
+                visibleVersion, visibleVersionTime);
+        tableAttributes.updateVisibleVersionAndTime(visibleVersion, visibleVersionTime);
+    }
+
+    // During `getNextVersion` and `updateVisibleVersionAndTime` period,
+    // the write lock on the table should be held continuously
+    public long getNextVersion() {
+        return tableAttributes.getNextVersion();
+    }
+
+    public long getVisibleVersion() {
+        return tableAttributes.getVisibleVersion();
+    }
+
+    public long getVisibleVersionTime() {
+        return tableAttributes.getVisibleVersionTime();
+    }
+
     @Override
     public PartitionType getPartitionType() {
         return partitionInfo.getType();
diff --git a/fe/fe-core/src/main/java/org/apache/doris/catalog/Table.java b/fe/fe-core/src/main/java/org/apache/doris/catalog/Table.java
index 783016e7d8..7294577fc5 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/catalog/Table.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/catalog/Table.java
@@ -118,7 +118,7 @@ public abstract class Table extends MetaObject implements Writable, TableIf {
     protected String comment = "";
 
     @SerializedName(value = "ta")
-    private TableAttributes tableAttributes = new TableAttributes();
+    protected TableAttributes tableAttributes = new TableAttributes();
 
     // check read lock leaky
     private Map<Long, String> readLockThreads = null;
diff --git a/fe/fe-core/src/main/java/org/apache/doris/catalog/TableAttributes.java b/fe/fe-core/src/main/java/org/apache/doris/catalog/TableAttributes.java
index 2ac86fada5..6670df839c 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/catalog/TableAttributes.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/catalog/TableAttributes.java
@@ -34,20 +34,57 @@ import java.util.Map;
  * TableAttributes contains additional information about all table
  */
 public class TableAttributes implements Writable {
+    public static final long TABLE_INIT_VERSION = 1L;
+
     @SerializedName(value = "constraints")
     private final Map<String, Constraint> constraintsMap = new HashMap<>();
+    @SerializedName(value = "visibleVersion")
+    private long visibleVersion;
+    @SerializedName(value = "visibleVersionTime")
+    private long visibleVersionTime;
 
+    public TableAttributes() {
+        this.visibleVersion = TABLE_INIT_VERSION;
+        this.visibleVersionTime = System.currentTimeMillis();
+    }
+
+    public Map<String, Constraint> getConstraintsMap() {
+        return constraintsMap;
+    }
+
+    public long getVisibleVersion() {
+        return visibleVersion;
+    }
+
+    public long getVisibleVersionTime() {
+        return visibleVersionTime;
+    }
+
+    public void updateVisibleVersionAndTime(long visibleVersion, long visibleVersionTime) {
+        // To be compatible with previous versions
+        if (visibleVersion <= TABLE_INIT_VERSION) {
+            return;
+        }
+        this.visibleVersion = visibleVersion;
+        this.visibleVersionTime = visibleVersionTime;
+    }
+
+    public long getNextVersion() {
+        return visibleVersion + 1;
+    }
 
     @Override
     public void write(DataOutput out) throws IOException {
         Text.writeString(out, GsonUtils.GSON.toJson(this));
     }
 
-    public TableAttributes read(DataInput in)  throws IOException {
-        return GsonUtils.GSON.fromJson(Text.readString(in), TableAttributes.class);
-    }
-
-    public Map<String, Constraint> getConstraintsMap() {
-        return constraintsMap;
+    public TableAttributes read(DataInput in) throws IOException {
+        TableAttributes tableAttributes = GsonUtils.GSON.fromJson(Text.readString(in), TableAttributes.class);
+        // To be compatible with previous versions
+        if (tableAttributes.getVisibleVersion() < TABLE_INIT_VERSION) {
+            tableAttributes.visibleVersion = TABLE_INIT_VERSION;
+            tableAttributes.visibleVersionTime = System.currentTimeMillis();
+        }
+        return tableAttributes;
     }
 }
diff --git a/fe/fe-core/src/main/java/org/apache/doris/common/proc/TablesProcDir.java b/fe/fe-core/src/main/java/org/apache/doris/common/proc/TablesProcDir.java
index 1b11a9a91b..b3ce9be35c 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/common/proc/TablesProcDir.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/common/proc/TablesProcDir.java
@@ -45,7 +45,7 @@ public class TablesProcDir implements ProcDirInterface {
     public static final ImmutableList<String> TITLE_NAMES = new ImmutableList.Builder<String>()
             .add("TableId").add("TableName").add("IndexNum").add("PartitionColumnName")
             .add("PartitionNum").add("State").add("Type").add("LastConsistencyCheckTime").add("ReplicaCount")
-            .add("LastUpdateTime")
+            .add("VisibleVersion").add("VisibleVersionTime").add("LastUpdateTime")
             .build();
 
     private DatabaseIf db;
@@ -117,6 +117,8 @@ public class TablesProcDir implements ProcDirInterface {
                     // last check time
                     tableInfo.add(TimeUtils.longToTimeString(olapTable.getLastCheckTime()));
                     tableInfo.add(replicaCount);
+                    tableInfo.add(olapTable.getVisibleVersion());
+                    tableInfo.add(olapTable.getVisibleVersionTime());
                 } else {
                     tableInfo.add(table.getId());
                     tableInfo.add(table.getName());
@@ -128,6 +130,8 @@ public class TablesProcDir implements ProcDirInterface {
                     // last check time
                     tableInfo.add(FeConstants.null_string);
                     tableInfo.add(replicaCount);
+                    tableInfo.add(FeConstants.null_string);
+                    tableInfo.add(FeConstants.null_string);
                 }
                 tableInfo.add(TimeUtils.longToTimeString(table.getUpdateTime()));
                 tableInfos.add(tableInfo);
diff --git a/fe/fe-core/src/main/java/org/apache/doris/datasource/InternalCatalog.java b/fe/fe-core/src/main/java/org/apache/doris/datasource/InternalCatalog.java
index f5b703c810..e9cdd7e436 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/datasource/InternalCatalog.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/datasource/InternalCatalog.java
@@ -1771,10 +1771,12 @@ public class InternalCatalog implements CatalogIf<Database> {
                 recycleTime = Env.getCurrentRecycleBin().getRecycleTimeById(partition.getId());
             }
         }
-
+        long version = olapTable.getNextVersion();
+        long versionTime = System.currentTimeMillis();
+        olapTable.updateVisibleVersionAndTime(version, versionTime);
         // log
         DropPartitionInfo info = new DropPartitionInfo(db.getId(), olapTable.getId(), partitionName, isTempPartition,
-                clause.isForceDrop(), recycleTime);
+                clause.isForceDrop(), recycleTime, version, versionTime);
         Env.getCurrentEnv().getEditLog().logDropPartition(info);
 
         LOG.info("succeed in dropping partition[{}], table : [{}-{}], is temp : {}, is force : {}",
@@ -1795,6 +1797,7 @@ public class InternalCatalog implements CatalogIf<Database> {
                     Env.getCurrentRecycleBin().setRecycleTimeByIdForReplay(partition.getId(), info.getRecycleTime());
                 }
             }
+            olapTable.updateVisibleVersionAndTime(info.getVersion(), info.getVersionTime());
         } finally {
             olapTable.writeUnlock();
         }
diff --git a/fe/fe-core/src/main/java/org/apache/doris/persist/DropPartitionInfo.java b/fe/fe-core/src/main/java/org/apache/doris/persist/DropPartitionInfo.java
index f4fac91526..be5a6cf75a 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/persist/DropPartitionInfo.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/persist/DropPartitionInfo.java
@@ -42,12 +42,16 @@ public class DropPartitionInfo implements Writable {
     private long recycleTime = 0;
     @SerializedName(value = "sql")
     private String sql;
+    @SerializedName(value = "version")
+    private long version = 0L;
+    @SerializedName(value = "versionTime")
+    private long versionTime = 0L;
 
     private DropPartitionInfo() {
     }
 
     public DropPartitionInfo(Long dbId, Long tableId, String partitionName,
-            boolean isTempPartition, boolean forceDrop, long recycleTime) {
+            boolean isTempPartition, boolean forceDrop, long recycleTime, long version, long versionTime) {
         this.dbId = dbId;
         this.tableId = tableId;
         this.partitionName = partitionName;
@@ -65,6 +69,8 @@ public class DropPartitionInfo implements Writable {
             sb.append(" FORCE");
         }
         this.sql = sb.toString();
+        this.version = version;
+        this.versionTime = versionTime;
     }
 
     public Long getDbId() {
@@ -88,7 +94,15 @@ public class DropPartitionInfo implements Writable {
     }
 
     public Long getRecycleTime() {
-        return  recycleTime;
+        return recycleTime;
+    }
+
+    public long getVersion() {
+        return version;
+    }
+
+    public long getVersionTime() {
+        return versionTime;
     }
 
     @Deprecated
diff --git a/fe/fe-core/src/main/java/org/apache/doris/persist/ReplacePartitionOperationLog.java b/fe/fe-core/src/main/java/org/apache/doris/persist/ReplacePartitionOperationLog.java
index 358b45d5bf..b41e64f5b9 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/persist/ReplacePartitionOperationLog.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/persist/ReplacePartitionOperationLog.java
@@ -49,11 +49,15 @@ public class ReplacePartitionOperationLog implements Writable {
     private boolean strictRange;
     @SerializedName(value = "useTempPartitionName")
     private boolean useTempPartitionName;
+    @SerializedName(value = "version")
+    private long version = 0L;
+    @SerializedName(value = "versionTime")
+    private long versionTime = 0L;
 
     public ReplacePartitionOperationLog(long dbId, String dbName, long tblId, String tblName,
                                         List<String> partitionNames,
                                         List<String> tempPartitonNames, boolean strictRange,
-                                        boolean useTempPartitionName) {
+                                        boolean useTempPartitionName, long version, long versionTime) {
         this.dbId = dbId;
         this.dbName = dbName;
         this.tblId = tblId;
@@ -62,6 +66,8 @@ public class ReplacePartitionOperationLog implements Writable {
         this.tempPartitions = tempPartitonNames;
         this.strictRange = strictRange;
         this.useTempPartitionName = useTempPartitionName;
+        this.version = version;
+        this.versionTime = versionTime;
     }
 
     public long getDbId() {
@@ -88,6 +94,14 @@ public class ReplacePartitionOperationLog implements Writable {
         return useTempPartitionName;
     }
 
+    public long getVersion() {
+        return version;
+    }
+
+    public long getVersionTime() {
+        return versionTime;
+    }
+
     public static ReplacePartitionOperationLog read(DataInput in) throws IOException {
         String json = Text.readString(in);
         return GsonUtils.GSON.fromJson(json, ReplacePartitionOperationLog.class);
diff --git a/fe/fe-core/src/main/java/org/apache/doris/transaction/DatabaseTransactionMgr.java b/fe/fe-core/src/main/java/org/apache/doris/transaction/DatabaseTransactionMgr.java
index 565df045cc..4323300293 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/transaction/DatabaseTransactionMgr.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/transaction/DatabaseTransactionMgr.java
@@ -1269,8 +1269,9 @@ public class DatabaseTransactionMgr {
         transactionState.setTransactionStatus(TransactionStatus.PRECOMMITTED);
         transactionState.setErrorReplicas(errorReplicaIds);
         for (long tableId : tableToPartition.keySet()) {
-            TableCommitInfo tableCommitInfo = new TableCommitInfo(tableId);
             OlapTable table = (OlapTable) db.getTableNullable(tableId);
+            TableCommitInfo tableCommitInfo = new TableCommitInfo(tableId, table.getNextVersion(),
+                    System.currentTimeMillis());
             PartitionInfo tblPartitionInfo = table.getPartitionInfo();
             for (long partitionId : tableToPartition.get(tableId)) {
                 String partitionRange = "";
@@ -1309,8 +1310,9 @@ public class DatabaseTransactionMgr {
         transactionState.setTransactionStatus(TransactionStatus.COMMITTED);
         transactionState.setErrorReplicas(errorReplicaIds);
         for (long tableId : tableToPartition.keySet()) {
-            TableCommitInfo tableCommitInfo = new TableCommitInfo(tableId);
             OlapTable table = (OlapTable) db.getTableNullable(tableId);
+            TableCommitInfo tableCommitInfo = new TableCommitInfo(tableId, table.getNextVersion(),
+                    System.currentTimeMillis());
             PartitionInfo tblPartitionInfo = table.getPartitionInfo();
             for (long partitionId : tableToPartition.get(tableId)) {
                 Partition partition = table.getPartition(partitionId);
@@ -1944,6 +1946,9 @@ public class DatabaseTransactionMgr {
                             transactionState, partition.getId(), version);
                 }
             }
+            long tableVersion = tableCommitInfo.getVersion();
+            long tableVersionTime = tableCommitInfo.getVersionTime();
+            table.updateVisibleVersionAndTime(tableVersion, tableVersionTime);
         }
         Map<Long, Long> tableIdToTotalNumDeltaRows = transactionState.getTableIdToTotalNumDeltaRows();
         Map<Long, Long> tableIdToNumDeltaRows = Maps.newHashMap();
diff --git a/fe/fe-core/src/main/java/org/apache/doris/transaction/TableCommitInfo.java b/fe/fe-core/src/main/java/org/apache/doris/transaction/TableCommitInfo.java
index 2423489474..b8d968f3ca 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/transaction/TableCommitInfo.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/transaction/TableCommitInfo.java
@@ -17,7 +17,11 @@
 
 package org.apache.doris.transaction;
 
+import org.apache.doris.catalog.Env;
+import org.apache.doris.common.FeMetaVersion;
+import org.apache.doris.common.io.Text;
 import org.apache.doris.common.io.Writable;
+import org.apache.doris.persist.gson.GsonUtils;
 
 import com.google.common.collect.Maps;
 import com.google.gson.annotations.SerializedName;
@@ -33,30 +37,40 @@ public class TableCommitInfo implements Writable {
     private long tableId;
     @SerializedName(value = "idToPartitionCommitInfo")
     private Map<Long, PartitionCommitInfo> idToPartitionCommitInfo;
+    @SerializedName(value = "version")
+    private long version;
+    @SerializedName(value = "versionTime")
+    private long versionTime;
 
     public TableCommitInfo() {
 
     }
 
-    public TableCommitInfo(long tableId) {
+    public TableCommitInfo(long tableId, long version, long visibleTime) {
         this.tableId = tableId;
         idToPartitionCommitInfo = Maps.newHashMap();
+        this.version = version;
+        this.versionTime = visibleTime;
     }
 
     @Override
     public void write(DataOutput out) throws IOException {
-        out.writeLong(tableId);
-        if (idToPartitionCommitInfo == null) {
-            out.writeBoolean(false);
+        String json = GsonUtils.GSON.toJson(this);
+        Text.writeString(out, json);
+    }
+
+    public static TableCommitInfo read(DataInput in) throws IOException {
+        if (Env.getCurrentEnvJournalVersion() < FeMetaVersion.VERSION_129) {
+            TableCommitInfo info = new TableCommitInfo();
+            info.readFields(in);
+            return info;
         } else {
-            out.writeBoolean(true);
-            out.writeInt(idToPartitionCommitInfo.size());
-            for (PartitionCommitInfo partitionCommitInfo : idToPartitionCommitInfo.values()) {
-                partitionCommitInfo.write(out);
-            }
+            String json = Text.readString(in);
+            return GsonUtils.GSON.fromJson(json, TableCommitInfo.class);
         }
     }
 
+    @Deprecated
     public void readFields(DataInput in) throws IOException {
         tableId = in.readLong();
         boolean hasPartitionInfo = in.readBoolean();
@@ -89,4 +103,20 @@ public class TableCommitInfo implements Writable {
     public PartitionCommitInfo getPartitionCommitInfo(long partitionId) {
         return this.idToPartitionCommitInfo.get(partitionId);
     }
+
+    public long getVersion() {
+        return version;
+    }
+
+    public void setVersion(long version) {
+        this.version = version;
+    }
+
+    public long getVersionTime() {
+        return versionTime;
+    }
+
+    public void setVersionTime(long versionTime) {
+        this.versionTime = versionTime;
+    }
 }
diff --git a/fe/fe-core/src/main/java/org/apache/doris/transaction/TransactionState.java b/fe/fe-core/src/main/java/org/apache/doris/transaction/TransactionState.java
index f9a094eceb..dbe63a0c79 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/transaction/TransactionState.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/transaction/TransactionState.java
@@ -717,8 +717,7 @@ public class TransactionState implements Writable {
         dbId = in.readLong();
         int size = in.readInt();
         for (int i = 0; i < size; i++) {
-            TableCommitInfo info = new TableCommitInfo();
-            info.readFields(in);
+            TableCommitInfo info = TableCommitInfo.read(in);
             idToTableCommitInfos.put(info.getTableId(), info);
         }
         txnCoordinator = new TxnCoordinator(TxnSourceType.valueOf(in.readInt()), Text.readString(in));
diff --git a/fe/fe-core/src/test/java/org/apache/doris/persist/DropPartitionInfoTest.java b/fe/fe-core/src/test/java/org/apache/doris/persist/DropPartitionInfoTest.java
index 9adb0ecb60..2570415bf5 100644
--- a/fe/fe-core/src/test/java/org/apache/doris/persist/DropPartitionInfoTest.java
+++ b/fe/fe-core/src/test/java/org/apache/doris/persist/DropPartitionInfoTest.java
@@ -41,7 +41,7 @@ public class DropPartitionInfoTest {
         file.createNewFile();
         DataOutputStream dos = new DataOutputStream(new FileOutputStream(file));
 
-        DropPartitionInfo info1 = new DropPartitionInfo(1L, 2L, "test_partition", false, true, 0);
+        DropPartitionInfo info1 = new DropPartitionInfo(1L, 2L, "test_partition", false, true, 0, 0L, 0L);
         info1.write(dos);
 
         dos.flush();
@@ -60,12 +60,12 @@ public class DropPartitionInfoTest {
 
         Assert.assertEquals(rInfo1, info1);
         Assert.assertNotEquals(rInfo1, this);
-        Assert.assertNotEquals(info1, new DropPartitionInfo(-1L, 2L, "test_partition", false, true, 0));
-        Assert.assertNotEquals(info1, new DropPartitionInfo(1L, -2L, "test_partition", false, true, 0));
-        Assert.assertNotEquals(info1, new DropPartitionInfo(1L, 2L, "test_partition1", false, true, 0));
-        Assert.assertNotEquals(info1, new DropPartitionInfo(1L, 2L, "test_partition", true, true, 0));
-        Assert.assertNotEquals(info1, new DropPartitionInfo(1L, 2L, "test_partition", false, false, 0));
-        Assert.assertEquals(info1, new DropPartitionInfo(1L, 2L, "test_partition", false, true, 0));
+        Assert.assertNotEquals(info1, new DropPartitionInfo(-1L, 2L, "test_partition", false, true, 0, 0L, 0L));
+        Assert.assertNotEquals(info1, new DropPartitionInfo(1L, -2L, "test_partition", false, true, 0, 0L, 0L));
+        Assert.assertNotEquals(info1, new DropPartitionInfo(1L, 2L, "test_partition1", false, true, 0, 0L, 0L));
+        Assert.assertNotEquals(info1, new DropPartitionInfo(1L, 2L, "test_partition", true, true, 0, 0L, 0L));
+        Assert.assertNotEquals(info1, new DropPartitionInfo(1L, 2L, "test_partition", false, false, 0, 0L, 0L));
+        Assert.assertEquals(info1, new DropPartitionInfo(1L, 2L, "test_partition", false, true, 0, 0L, 0L));
 
         // 3. delete files
         dis.close();
diff --git a/regression-test/framework/src/main/groovy/org/apache/doris/regression/suite/Suite.groovy b/regression-test/framework/src/main/groovy/org/apache/doris/regression/suite/Suite.groovy
index 9f38ab5ed0..9e238570ac 100644
--- a/regression-test/framework/src/main/groovy/org/apache/doris/regression/suite/Suite.groovy
+++ b/regression-test/framework/src/main/groovy/org/apache/doris/regression/suite/Suite.groovy
@@ -400,6 +400,26 @@ class Suite implements GroovyInterceptable {
         }
     }
 
+    long getDbId() {
+        def dbInfo = sql "show proc '/dbs'"
+        for(List<Object> row : dbInfo) {
+            if (row[1].equals(context.dbName)) {
+                println(row[0])
+                return row[0].toLong()
+            }
+        }
+    }
+
+    long getTableVersion(long dbId, String tableName) {
+       def result = sql_return_maparray """show proc '/dbs/${dbId}'"""
+        for (def res : result) {
+            if(res.TableName.equals(tableName)) {
+                log.info(res.toString())
+                return res.VisibleVersion.toLong()
+            }
+        }
+    }
+
     List<List<Object>> order_sql(String sqlStr) {
         return sql(sqlStr,  true)
     }
diff --git a/regression-test/suites/table_p0/test_table_version.groovy b/regression-test/suites/table_p0/test_table_version.groovy
new file mode 100644
index 0000000000..250dcf96cb
--- /dev/null
+++ b/regression-test/suites/table_p0/test_table_version.groovy
@@ -0,0 +1,67 @@
+// 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.
+
+import org.junit.Assert;
+
+suite("test_table_version") {
+    def tableNameNum = "t_test_version_user_num"
+    def dbName = "regression_test_table_p0"
+    sql """drop table if exists `${tableNameNum}`"""
+
+    sql """
+        CREATE TABLE `${tableNameNum}` (
+          `user_id` LARGEINT NOT NULL COMMENT '\"用户id\"',
+          `date` DATE NOT NULL COMMENT '\"数据灌入日期时间\"',
+          `num` SMALLINT NOT NULL COMMENT '\"数量\"'
+        ) ENGINE=OLAP
+        DUPLICATE KEY(`user_id`, `date`, `num`)
+        COMMENT 'OLAP'
+        PARTITION BY RANGE(`date`)
+        (PARTITION p201701_1000 VALUES [('0000-01-01'), ('2017-02-01')),
+        PARTITION p201702_2000 VALUES [('2017-02-01'), ('2017-03-01')),
+        PARTITION p201703_all VALUES [('2017-03-01'), ('2017-04-01')))
+        DISTRIBUTED BY HASH(`user_id`) BUCKETS 2
+        PROPERTIES ('replication_num' = '1') ;
+        """
+
+    def dbId = getDbId();
+    def visibleVersion = getTableVersion(dbId,tableNameNum);
+    assertEquals(1, visibleVersion);
+
+    sql """
+        insert into ${tableNameNum} values(1,"2017-01-15",1);
+        """
+    visibleVersion = getTableVersion(dbId,tableNameNum);
+    assertEquals(2, visibleVersion);
+
+    sql """
+        alter table ${tableNameNum} drop partition p201703_all;
+        """
+    visibleVersion = getTableVersion(dbId,tableNameNum);
+    assertEquals(3, visibleVersion);
+
+    sql """
+        ALTER TABLE ${tableNameNum} ADD TEMPORARY PARTITION p201702_2000_1 VALUES [('2017-02-01'), ('2017-03-01'));
+    """
+    sql """
+        ALTER TABLE ${tableNameNum} REPLACE PARTITION (p201702_2000) WITH TEMPORARY PARTITION (p201702_2000_1);
+    """
+    visibleVersion = getTableVersion(dbId,tableNameNum);
+    assertEquals(4, visibleVersion);
+
+    sql """drop table if exists `${tableNameNum}`"""
+}