apache · CalvinKirs · Sep 18, 2026
@@ -59,6 +59,15 @@ public interface HMSCachedClient {
 
     List<Partition> getPartitions(String dbName, String tblName, List<String> partitionNames);
 
+    /**
+     * Gets the subset of the requested partitions that exist. Unlike {@link #getPartitions}, a name the
+     * metastore no longer has is a normal answer and is simply absent from the result. The compatibility
+     * default preserves implementations whose {@code getPartitions} already returns only what exists.
+     */
+    default List<Partition> getExistingPartitions(String dbName, String tblName, List<String> partitionNames) {
+        return getPartitions(dbName, tblName, partitionNames);
+    }
+
     Table getTable(String dbName, String tblName);
 
     List<FieldSchema> getSchema(String dbName, String tblName);

@@ -20,6 +20,8 @@
 import org.apache.doris.common.util.Util;
 
 public class HMSClientException extends RuntimeException {
+    private HmsPartitionBatchStats partitionBatchStats;
+
     public HMSClientException(String format, Throwable cause, Object... msg) {
         super(String.format(format, msg) + (cause == null ? "" : ". reason: " + Util.getRootCauseMessage(cause)),
                 cause);
@@ -28,4 +30,13 @@ public HMSClientException(String format, Throwable cause, Object... msg) {
     public HMSClientException(String format, Object... msg) {
         super(String.format(format, msg));
     }
+
+    public HmsPartitionBatchStats getPartitionBatchStats() {
+        return partitionBatchStats;
+    }
+
+    HMSClientException withPartitionBatchStats(HmsPartitionBatchStats stats) {
+        this.partitionBatchStats = stats;
+        return this;
+    }
 }
@@ -1058,6 +1058,16 @@ public MTMVSnapshotIf getPartitionSnapshot(String partitionName, MTMVRefreshCont
         return dlaTable.getPartitionSnapshot(partitionName, context, snapshot);
     }
 
+    @Override
+    public Map<String, MTMVSnapshotIf> getPartitionSnapshots(Set<String> partitionNames,
+            MTMVRefreshContext context, Optional<MvccSnapshot> snapshot) throws AnalysisException {
+        makeSureInitialized();
+        if (dlaTable instanceof HiveDlaTable) {
+            return ((HiveDlaTable) dlaTable).getPartitionSnapshots(partitionNames, snapshot);
+        }
+        return MTMVRelatedTableIf.super.getPartitionSnapshots(partitionNames, context, snapshot);
+    }
+
     @Override
     public MTMVSnapshotIf getTableSnapshot(MTMVRefreshContext context, Optional<MvccSnapshot> snapshot)
             throws AnalysisException {

@@ -32,7 +32,9 @@
 import com.google.common.collect.Lists;
 import org.apache.commons.collections4.CollectionUtils;
 
+import java.util.ArrayList;
 import java.util.Collections;
+import java.util.LinkedHashMap;
 import java.util.List;
 import java.util.Map;
 import java.util.Optional;
@@ -80,6 +82,57 @@ public MTMVSnapshotIf getPartitionSnapshot(String partitionName, MTMVRefreshCont
         return new MTMVTimestampSnapshot(hivePartition.getLastModifiedTime());
     }
 
+    /**
+     * Bulk form of {@link #getPartitionSnapshot}: resolves every requested name against one partition-value
+     * listing and loads all cache misses through one batched HMS request instead of one RPC per partition.
+     * A name absent from the listing is omitted (the refresh context reports it per partition); any other
+     * failure is normalized to the checked AnalysisException this MTMV boundary declares, so the
+     * transparent-rewrite path degrades per-MV instead of a raw runtime exception disabling every MV
+     * candidate at the planner hook.
+     */
+    Map<String, MTMVSnapshotIf> getPartitionSnapshots(Set<String> partitionNames,
+            Optional<MvccSnapshot> snapshot) throws AnalysisException {
+        try {
+            HiveExternalMetaCache.HivePartitionValues hivePartitionValues =
+                    hmsTable.getHivePartitionValues(snapshot);
+            HiveExternalMetaCache cache = Env.getCurrentEnv().getExtMetaCacheMgr()
+                    .hive(hmsTable.getCatalog().getId());
+            List<String> resolvedNames = new ArrayList<>(partitionNames.size());
+            List<List<String>> resolvedValues = new ArrayList<>(partitionNames.size());
+            for (String partitionName : partitionNames) {
+                Long partitionId = hivePartitionValues.getPartitionNameToIdMap().get(partitionName);
+                if (partitionId == null) {
+                    continue;
+                }
+                List<String> partitionValues = hivePartitionValues.getPartitionValuesMap().get(partitionId);
+                if (CollectionUtils.isEmpty(partitionValues)) {
+                    continue;
+                }
+                resolvedNames.add(partitionName);
+                resolvedValues.add(partitionValues);
+            }
+            Map<String, MTMVSnapshotIf> result = new LinkedHashMap<>();
+            if (resolvedNames.isEmpty()) {
+                return result;
+            }
+            List<HivePartition> partitions = cache.getAllPartitionsWithCache(hmsTable, resolvedValues);
+            if (partitions.size() != resolvedNames.size()) {
+                throw new AnalysisException("Invalid HMS partition result: requested="
+                        + resolvedNames.size() + ", returned=" + partitions.size());
+            }
+            for (int i = 0; i < resolvedNames.size(); i++) {
+                result.put(resolvedNames.get(i),
+                        new MTMVTimestampSnapshot(partitions.get(i).getLastModifiedTime()));
+            }
+            return result;
+        } catch (AnalysisException e) {
+            throw e;
+        } catch (RuntimeException e) {
+            throw new AnalysisException("failed to load partition snapshots for "
+                    + hmsTable.getName() + ": " + e.getMessage(), e);
+        }
+    }
+
     @Override
     public MTMVSnapshotIf getTableSnapshot(MTMVRefreshContext context, Optional<MvccSnapshot> snapshot)
             throws AnalysisException {

@@ -482,7 +482,10 @@ private Map<PartitionCacheKey, HivePartition> loadPartitions(Iterable<? extends
             return sb.toString();
         }).collect(Collectors.toList());
 
-        List<Partition> partitions = catalog.getClient().getPartitions(
+        // Lenient existence semantics: a partition dropped remotely since the name list was captured is
+        // simply absent from the result (the historical getPartitionsByNames behavior); the client batches
+        // and validates the physical RPCs internally.
+        List<Partition> partitions = catalog.getClient().getExistingPartitions(
                 nameMapping.getRemoteDbName(), nameMapping.getRemoteTblName(), partitionNames);
         for (Partition partition : partitions) {
             StorageDescriptor sd = partition.getSd();
@@ -657,7 +660,30 @@ private List<HivePartition> getAllPartitions(ExternalTable dorisTable,
         List<HivePartition> partitions;
         if (withCache) {
             MetaCacheEntry<PartitionCacheKey, HivePartition> partitionEntry = this.partitionEntry.get(catalogId);
-            partitions = keys.stream().map(partitionEntry::get).collect(Collectors.toList());
+            // Serve hits from the cache and aggregate every miss into ONE bulk load (batched inside the
+            // client) instead of one single-partition RPC per missed key.
+            Map<PartitionCacheKey, HivePartition> resolved = new HashMap<>();
+            List<PartitionCacheKey> misses = new ArrayList<>();
+            for (PartitionCacheKey key : keys) {
+                HivePartition hit = partitionEntry.getIfPresent(key);
+                if (hit != null) {
+                    resolved.put(key, hit);
+                } else {
+                    misses.add(key);
+                }
+            }
+            if (!misses.isEmpty()) {
+                Map<PartitionCacheKey, HivePartition> loaded = loadPartitions(misses);
+                loaded.forEach(partitionEntry::put);
+                resolved.putAll(loaded);
+            }
+            partitions = new ArrayList<>(keys.size());
+            for (PartitionCacheKey key : keys) {
+                HivePartition partition = resolved.get(key);
+                // A key the bulk load did not return (partition dropped remotely) falls back to the
+                // per-key loader, preserving the original single-load not-found error for that partition.
+                partitions.add(partition != null ? partition : partitionEntry.get(key));
+            }
         } else {
             partitions = new ArrayList<>(loadPartitions(keys).values());
         }