[fix](Hudi-mtmv) Support asynchronous materialized view partition refresh feature for Hudi external tables.

fe/fe-core/src/main/java/org/apache/doris/datasource/ExternalMetaCacheMgr.java

@@ -34,6 +34,7 @@
 import org.apache.doris.datasource.maxcompute.MaxComputeMetadataCache;
 import org.apache.doris.datasource.maxcompute.MaxComputeMetadataCacheMgr;
 import org.apache.doris.datasource.metacache.MetaCache;
+import org.apache.doris.datasource.mvcc.MvccUtil;
 import org.apache.doris.datasource.paimon.PaimonMetadataCache;
 import org.apache.doris.datasource.paimon.PaimonMetadataCacheMgr;
 import org.apache.doris.fs.FileSystemCache;
@@ -292,7 +293,7 @@ public void addPartitionsCache(long catalogId, HMSExternalTable table, List<Stri
         if (metaCache != null) {
             List<Type> partitionColumnTypes;
             try {
-                partitionColumnTypes = table.getPartitionColumnTypes();
+                partitionColumnTypes = table.getPartitionColumnTypes(MvccUtil.getSnapshotFromContext(table));
             } catch (NotSupportedException e) {
                 LOG.warn("Ignore not supported hms table, message: {} ", e.getMessage());
                 return;

fe/fe-core/src/main/java/org/apache/doris/datasource/ExternalTable.java

@@ -391,7 +391,7 @@ public List<Long> getChunkSizes() {
         throw new NotImplementedException("getChunkSized not implemented");
     }
 
-    protected Optional<SchemaCacheValue> getSchemaCacheValue() {
+    public Optional<SchemaCacheValue> getSchemaCacheValue() {
         ExternalSchemaCache cache = Env.getCurrentEnv().getExtMetaCacheMgr().getSchemaCache(catalog);
         return cache.getSchemaValue(dbName, name);
     }

fe/fe-core/src/main/java/org/apache/doris/datasource/TablePartitionValues.java

@@ -51,6 +51,7 @@ public class TablePartitionValues {
     private long nextPartitionId;
     private final Map<Long, PartitionItem> idToPartitionItem;
     private final Map<String, Long> partitionNameToIdMap;
+    private Map<String, Long> partitionNameToLastModifiedMap;
     private final Map<Long, String> partitionIdToNameMap;
 
     private Map<Long, List<UniqueId>> idToUniqueIdsMap;
@@ -68,15 +69,12 @@ public TablePartitionValues() {
         nextPartitionId = 0;
         idToPartitionItem = new HashMap<>();
         partitionNameToIdMap = new HashMap<>();
+        partitionNameToLastModifiedMap = new HashMap<>();
         partitionIdToNameMap = new HashMap<>();
     }
 
-    public TablePartitionValues(List<String> partitionNames, List<List<String>> partitionValues, List<Type> types) {
-        this();
-        addPartitions(partitionNames, partitionValues, types);
-    }
-
-    public void addPartitions(List<String> partitionNames, List<List<String>> partitionValues, List<Type> types) {
+    public void addPartitions(List<String> partitionNames, List<List<String>> partitionValues, List<Type> types,
+            List<Long> partitionLastUpdateTimestamp) {
         Preconditions.checkState(partitionNames.size() == partitionValues.size());
         List<String> addPartitionNames = new ArrayList<>();
         List<PartitionItem> addPartitionItems = new ArrayList<>();
@@ -90,6 +88,7 @@ public void addPartitions(List<String> partitionNames, List<List<String>> partit
                 addPartitionNames.add(partitionNames.get(i));
                 addPartitionItems.add(toListPartitionItem(partitionValues.get(i), types));
             }
+            partitionNameToLastModifiedMap.put(partitionNames.get(i), partitionLastUpdateTimestamp.get(i));
         }
         cleanPartitions();
 
@@ -123,23 +122,6 @@ private void addPartitionItems(List<String> partitionNames, List<PartitionItem>
         partitionValuesMap = ListPartitionPrunerV2.getPartitionValuesMap(idToPartitionItem);
     }
 
-    public void dropPartitions(List<String> partitionNames, List<Type> types) {
-        partitionNames.forEach(p -> {
-            Long removedPartition = partitionNameToIdMap.get(p);
-            if (removedPartition != null) {
-                idToPartitionItem.remove(removedPartition);
-            }
-        });
-        List<String> remainingPartitionNames = new ArrayList<>();
-        List<PartitionItem> remainingPartitionItems = new ArrayList<>();
-        partitionNameToIdMap.forEach((partitionName, partitionId) -> {
-            remainingPartitionNames.add(partitionName);
-            remainingPartitionItems.add(idToPartitionItem.get(partitionId));
-        });
-        cleanPartitions();
-        addPartitionItems(remainingPartitionNames, remainingPartitionItems, types);
-    }
-
     public long getLastUpdateTimestamp() {
         return lastUpdateTimestamp;
     }
@@ -148,6 +130,10 @@ public void setLastUpdateTimestamp(long lastUpdateTimestamp) {
         this.lastUpdateTimestamp = lastUpdateTimestamp;
     }
 
+    public Map<String, Long> getPartitionNameToLastModifiedMap() {
+        return partitionNameToLastModifiedMap;
+    }
+
     public Lock readLock() {
         return readWriteLock.readLock();
     }

fe/fe-core/src/main/java/org/apache/doris/datasource/hive/HMSDlaTable.java

@@ -0,0 +1,79 @@
+// 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.datasource.hive;
+
+import org.apache.doris.catalog.Column;
+import org.apache.doris.catalog.Env;
+import org.apache.doris.catalog.MTMV;
+import org.apache.doris.catalog.PartitionItem;
+import org.apache.doris.catalog.PartitionType;
+import org.apache.doris.common.AnalysisException;
+import org.apache.doris.common.DdlException;
+import org.apache.doris.datasource.mvcc.MvccSnapshot;
+import org.apache.doris.mtmv.MTMVBaseTableIf;
+import org.apache.doris.mtmv.MTMVRefreshContext;
+import org.apache.doris.mtmv.MTMVSnapshotIf;
+
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.Set;
+
+/**
+ * This abstract class represents a Hive Metastore (HMS) Dla Table and provides a blueprint for
+ * various operations related to metastore tables in Doris.
+ *
+ * Purpose:
+ * - To encapsulate common functionalities that HMS Dla tables should have for implementing other interfaces
+ *
+ * Why needed:
+ * - To provide a unified way to manage and interact with different kinds of Dla Table
+ * - To facilitate the implementation of multi-table materialized views (MTMV) by providing necessary
+ *   methods for snapshot and partition management.
+ * - To abstract out the specific details of HMS table operations, making the code more modular and maintainable.
+ */
+public abstract class HMSDlaTable implements MTMVBaseTableIf {
+    protected HMSExternalTable hmsTable;
+
+    public HMSDlaTable(HMSExternalTable table) {
+        this.hmsTable = table;
+    }
+
+    abstract Map<String, PartitionItem> getAndCopyPartitionItems(Optional<MvccSnapshot> snapshot)
+            throws AnalysisException;
+
+    abstract PartitionType getPartitionType(Optional<MvccSnapshot> snapshot);
+
+    abstract Set<String> getPartitionColumnNames(Optional<MvccSnapshot> snapshot);
+
+    abstract List<Column> getPartitionColumns(Optional<MvccSnapshot> snapshot);
+
+    abstract MTMVSnapshotIf getPartitionSnapshot(String partitionName, MTMVRefreshContext context,
+            Optional<MvccSnapshot> snapshot) throws AnalysisException;
+
+    abstract MTMVSnapshotIf getTableSnapshot(MTMVRefreshContext context, Optional<MvccSnapshot> snapshot)
+            throws AnalysisException;
+
+    abstract boolean isPartitionColumnAllowNull();
+
+    @Override
+    public void beforeMTMVRefresh(MTMV mtmv) throws DdlException {
+        Env.getCurrentEnv().getRefreshManager()
+                .refreshTable(hmsTable.getCatalog().getName(), hmsTable.getDbName(), hmsTable.getName(), true);
+    }
+}