[improvement](hive) Batch Hive metastore partition access

fe/fe-connector/fe-connector-cache/src/main/java/org/apache/doris/connector/cache/MetaCache.java

@@ -95,7 +95,8 @@ public void invalidateKey(K key) {
 
     /**
      * Starts a bulk remote load whose results all belong below {@code parentScope}. Publishing through the
-     * returned handle is fenced against concurrent scope and exact-key invalidation.
+     * returned handle is fenced against concurrent scope and exact-key invalidation. A disabled cache still
+     * returns a currentness fence, but {@link BulkLoad#publish(Object, Object)} retains no values.
      */
     public BulkLoad<K, V> beginBulkLoad(ScopePath parentScope) {
         return new BulkLoad<>(this, delegate.beginBulkLoad(parentScope));
@@ -144,6 +145,12 @@ public boolean publish(K key, V value) {
             return owner.delegate.publish(delegate, nonNullKey, owner.definition.scope(nonNullKey), value);
         }
 
+        /** Returns whether this load is still current for {@code key}, without publishing a value. */
+        public boolean isCurrent(K key) {
+            K nonNullKey = Objects.requireNonNull(key, "key can not be null");
+            return owner.delegate.isBulkLoadCurrent(delegate, nonNullKey);
+        }
+
         @Override
         public void close() {
             delegate.close();

fe/fe-connector/fe-connector-cache/src/main/java/org/apache/doris/connector/cache/ScopedMetaCache.java

@@ -377,9 +377,6 @@ void invalidateKey(
     public BulkLoadHandle beginBulkLoad(ScopePath parentScope) {
         Objects.requireNonNull(parentScope, "parentScope can not be null");
         checkOpen();
-        if (!effectiveEnabled) {
-            return BulkLoadHandle.disabled(this, parentScope);
-        }
         ScopeLease scopeLease = registry.acquire(parentScope);
         BigInteger exactSequence = bulkInvalidationGate.write(() -> {
             if (closed.get()) {
@@ -420,6 +417,15 @@ public boolean publish(
         }
     }
 
+    boolean isBulkLoadCurrent(BulkLoadHandle handle, K key) {
+        Objects.requireNonNull(handle, "handle can not be null");
+        Objects.requireNonNull(key, "key can not be null");
+        checkOpen();
+        handle.checkOwner(this);
+        return bulkInvalidationGate.readBoolean(() -> isBulkKeyCurrent(handle, key)
+                && handle.scopeLease.commitIfPublicationCurrent(handle.scopePublicationState, () -> true));
+    }
+
     public CacheMetrics metrics() {
         return bulkInvalidationGate.read(() -> new CacheMetrics(
                     data.estimatedSize(),
@@ -704,20 +710,17 @@ private void closeBulkHandle(BulkLoadHandle handle) {
             if (!handle.closed.compareAndSet(false, true)) {
                 return null;
             }
-            if (handle.scopeLease != null) {
-                Integer count = activeBulkStarts.get(handle.exactInvalidationSequence);
-                if (count == null) {
-                    throw new IllegalStateException("Bulk-load handle start sequence is not registered");
-                }
-                if (count == 1) {
-                    activeBulkStarts.remove(handle.exactInvalidationSequence);
-                } else {
-                    activeBulkStarts.put(handle.exactInvalidationSequence, count - 1);
-                }
-                pruneExactInvalidations();
-                return handle.scopeLease;
+            Integer count = activeBulkStarts.get(handle.exactInvalidationSequence);
+            if (count == null) {
+                throw new IllegalStateException("Bulk-load handle start sequence is not registered");
             }
-            return null;
+            if (count == 1) {
+                activeBulkStarts.remove(handle.exactInvalidationSequence);
+            } else {
+                activeBulkStarts.put(handle.exactInvalidationSequence, count - 1);
+            }
+            pruneExactInvalidations();
+            return handle.scopeLease;
         });
         if (leaseToClose != null) {
             leaseToClose.close();
@@ -913,11 +916,6 @@ private BulkLoadHandle(
             this.exactInvalidationSequence = exactInvalidationSequence;
         }
 
-        private static BulkLoadHandle disabled(
-                ScopedMetaCache<?, ?> owner, ScopePath parentScope) {
-            return new BulkLoadHandle(owner, parentScope, null, null, BigInteger.ZERO);
-        }
-
         private void checkOwner(ScopedMetaCache<?, ?> expectedOwner) {
             if (owner != expectedOwner) {
                 throw new IllegalArgumentException("Bulk-load handle belongs to another cache");

fe/fe-connector/fe-connector-cache/src/test/java/org/apache/doris/connector/cache/ScopedMetaCacheHierarchyTest.java

@@ -168,7 +168,7 @@ public void samePhysicalKeyCanMoveToANewScopeWithoutOldScopeOwningIt() {
     }
 
     @Test
-    public void disabledCacheNeverPublishesOrAllocatesIndexes() {
+    public void disabledCacheNeverPublishesOrRetainsIndexes() {
         List<CacheSpec> disabledSpecs = Arrays.asList(
                 CacheSpec.of(false, CacheSpec.CACHE_NO_TTL, 100L),
                 CacheSpec.of(true, CacheSpec.CACHE_TTL_DISABLE_CACHE, 100L),
@@ -182,6 +182,9 @@ public void disabledCacheNeverPublishesOrAllocatesIndexes() {
                 Assertions.assertEquals(2, cache.get("key", PARTITION, key -> loads.incrementAndGet()));
                 cache.put("key", PARTITION, 3);
                 try (BulkLoadHandle handle = cache.beginBulkLoad(TABLE)) {
+                    Assertions.assertTrue(cache.isBulkLoadCurrent(handle, "bulk"));
+                    registry.invalidate(TABLE);
+                    Assertions.assertFalse(cache.isBulkLoadCurrent(handle, "bulk"));
                     Assertions.assertFalse(cache.publish(handle, "bulk", PARTITION, 4));
                 }
 

fe/fe-connector/fe-connector-hive/src/main/java/org/apache/doris/connector/hive/HiveCatalogProperties.java

@@ -116,6 +116,10 @@ public final class HiveCatalogProperties {
             description = "size of the metastore client pool")
     private int hmsClientPoolSize = DEFAULT_HMS_CLIENT_POOL_SIZE;
 
+    @ConnectorProperty(names = {HmsClientConfig.PARTITION_BATCH_SIZE_KEY}, required = false,
+            description = "maximum partition names sent in one Hive Metastore RPC")
+    private int hmsPartitionsBatchSizePerRpc = HmsClientConfig.DEFAULT_PARTITION_BATCH_SIZE;
+
     @ConnectorProperty(names = {ENABLE_HMS_EVENTS_INCREMENTAL_SYNC}, required = false,
             description = "poll HMS notification events for incremental metadata refresh")
     private boolean enableHmsEventsIncrementalSync;
@@ -166,6 +170,7 @@ public static HiveCatalogProperties of(Map<String, String> properties) {
                 .require(p.metastoreUri, "HMS URI ('" + HIVE_METASTORE_URIS + "') is required")
                 .validate();
         p.hmsClientProperties = withCanonicalMetastoreUri(p.raw, p.metastoreUri);
+        new HmsClientConfig(p.hmsClientProperties, p.hmsClientPoolSize);
         return p;
     }
 

fe/fe-connector/fe-connector-hive/src/main/java/org/apache/doris/connector/hive/HiveConnector.java

@@ -259,7 +259,11 @@ public ConnectorScanPlanProvider getScanPlanProvider() {
     @Override
     public ConnectorScanPlanProvider getScanPlanProvider(ConnectorTableHandle handle) {
         if (handle instanceof HiveTableHandle) {
-            return getScanPlanProvider();
+            ConnectorScanPlanProvider provider = getScanPlanProvider();
+            if (provider instanceof HiveScanPlanProvider) {
+                ((HiveScanPlanProvider) provider).recordPruningProfile((HiveTableHandle) handle);
+            }
+            return provider;
         }
         return resolveSiblingOwner(handle).getScanPlanProvider(handle);
     }

fe/fe-connector/fe-connector-hive/src/main/java/org/apache/doris/connector/hive/HiveConnectorMetadata.java

@@ -25,6 +25,7 @@
 import org.apache.doris.connector.hms.HmsColumnStatistics;
 import org.apache.doris.connector.hms.HmsCreateDatabaseRequest;
 import org.apache.doris.connector.hms.HmsCreateTableRequest;
+import org.apache.doris.connector.hms.HmsPartitionBatchResult;
 import org.apache.doris.connector.hms.HmsPartitionInfo;
 import org.apache.doris.connector.hms.HmsTableInfo;
 import org.apache.doris.connector.hms.HmsTypeMapping;
@@ -1103,7 +1104,7 @@ private List<PartitionRef> resolvePartitionRefs(HiveTableHandle handle) {
         if (partNames.isEmpty()) {
             return Collections.emptyList();
         }
-        List<HmsPartitionInfo> partitions = hmsClient.getPartitions(
+        List<HmsPartitionInfo> partitions = hmsClient.getExistingPartitions(
                 handle.getDbName(), handle.getTableName(), partNames);
         List<PartitionRef> refs = new ArrayList<>(partitions.size());
         for (HmsPartitionInfo partition : partitions) {
@@ -1182,17 +1183,28 @@ public Optional<FilterApplicationResult<ConnectorTableHandle>> applyFilter(
             return Optional.empty();
         }
 
-        List<HmsPartitionInfo> prunedPartitions = matchedPartNames.isEmpty()
-                ? Collections.emptyList()
-                : hmsClient.getPartitions(hiveHandle.getDbName(),
-                        hiveHandle.getTableName(), matchedPartNames);
+        HmsPartitionBatchResult pruningResult;
+        try {
+            pruningResult = matchedPartNames.isEmpty()
+                    ? null : hmsClient.getExistingPartitionsWithStats(
+                            hiveHandle.getDbName(), hiveHandle.getTableName(), matchedPartNames);
+        } catch (HmsClientException e) {
+            if (e.getPartitionBatchStats() != null) {
+                HiveScanPlanProvider.recordPruningFailure(
+                        session, hiveHandle.getDbName(), hiveHandle.getTableName(), e.getPartitionBatchStats());
+            }
+            throw e;
+        }
+        List<HmsPartitionInfo> prunedPartitions = pruningResult == null
+                ? Collections.emptyList() : pruningResult.getPartitions();
 
         LOG.info("Partition pruning: {}.{} all={} pruned={}",
                 hiveHandle.getDbName(), hiveHandle.getTableName(),
                 allPartNames.size(), prunedPartitions.size());
 
         HiveTableHandle newHandle = hiveHandle.toBuilder()
                 .prunedPartitions(prunedPartitions)
+                .pruningBatchStats(pruningResult == null ? null : pruningResult.getStats())
                 .build();
         return Optional.of(new FilterApplicationResult<>(
                 newHandle, constraint.getExpression(), false));
@@ -1393,8 +1405,8 @@ public Optional<ConnectorTableFreshness> getTableFreshness(ConnectorSession sess
             // Parity: an empty partition list yields MTMVMaxTimestampSnapshot(tableName, 0).
             return Optional.of(new ConnectorTableFreshness(hiveHandle.getTableName(), 0L));
         }
-        List<HmsPartitionInfo> partitions =
-                hmsClient.getPartitions(hiveHandle.getDbName(), hiveHandle.getTableName(), partitionNames);
+        List<HmsPartitionInfo> partitions = hmsClient.getExistingPartitions(
+                hiveHandle.getDbName(), hiveHandle.getTableName(), partitionNames);
         String maxName = hiveHandle.getTableName();
         long maxMillis = 0L;
         for (HmsPartitionInfo partition : partitions) {
@@ -1423,14 +1435,38 @@ public OptionalLong getPartitionFreshnessMillis(ConnectorSession session, Connec
             return siblingMetadata(session, handle).getPartitionFreshnessMillis(session, handle, partitionName);
         }
         HiveTableHandle hiveHandle = (HiveTableHandle) handle;
-        List<HmsPartitionInfo> partitions = hmsClient.getPartitions(hiveHandle.getDbName(),
+        List<HmsPartitionInfo> partitions = hmsClient.getExistingPartitions(hiveHandle.getDbName(),
                 hiveHandle.getTableName(), Collections.singletonList(partitionName));
         if (partitions.isEmpty()) {
             return OptionalLong.empty();
         }
         return OptionalLong.of(lastDdlMillis(partitions.get(0).getParameters()));
     }
 
+    @Override
+    public Map<String, Long> getPartitionsFreshnessMillis(ConnectorSession session, ConnectorTableHandle handle,
+            List<String> partitionNames) {
+        if (!(handle instanceof HiveTableHandle)) {
+            return siblingMetadata(session, handle)
+                    .getPartitionsFreshnessMillis(session, handle, partitionNames);
+        }
+        if (partitionNames.isEmpty()) {
+            return Collections.emptyMap();
+        }
+        HiveTableHandle hiveHandle = (HiveTableHandle) handle;
+        List<HmsPartitionInfo> partitions = hmsClient.getExistingPartitions(
+                hiveHandle.getDbName(), hiveHandle.getTableName(), partitionNames);
+        Map<List<String>, String> namesByValues = new HashMap<>();
+        for (String partitionName : partitionNames) {
+            namesByValues.put(HiveWriteUtils.toPartitionValues(partitionName), partitionName);
+        }
+        Map<String, Long> freshness = new LinkedHashMap<>();
+        for (HmsPartitionInfo partition : partitions) {
+            freshness.put(namesByValues.get(partition.getValues()), lastDdlMillis(partition.getParameters()));
+        }
+        return freshness;
+    }
+
     /**
      * The last-DDL time in MILLIS from an HMS parameter map, byte-parity with legacy
      * {@code HivePartition.getLastModifiedTime} / {@code HMSExternalTable.getLastDdlTime}: the