Unify hudi/maxcompute meta caches under catalog scope

Author: suxiaogang223

fe/fe-core/src/main/java/org/apache/doris/datasource/hudi/source/HudiCatalogCache.java

@@ -0,0 +1,85 @@
+// 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.hudi.source;
+
+import org.apache.doris.common.Config;
+import org.apache.doris.datasource.ExternalCatalog;
+import org.apache.doris.datasource.hive.HMSExternalCatalog;
+import org.apache.doris.datasource.metacache.CacheSpec;
+
+import java.util.Objects;
+import java.util.concurrent.ExecutorService;
+
+/**
+ * Per-catalog cache container for Hudi metadata processors.
+ */
+public final class HudiCatalogCache {
+    private static final String PARTITION_MODULE_NAME = "partition";
+    private static final String FS_VIEW_MODULE_NAME = "fs-view";
+    private static final String META_CLIENT_MODULE_NAME = "meta-client";
+
+    private final HudiCachedPartitionProcessor partitionProcessor;
+    private final HudiCachedFsViewProcessor fsViewProcessor;
+    private final HudiCachedMetaClientProcessor metaClientProcessor;
+
+    private HudiCatalogCache(HudiCachedPartitionProcessor partitionProcessor,
+            HudiCachedFsViewProcessor fsViewProcessor,
+            HudiCachedMetaClientProcessor metaClientProcessor) {
+        this.partitionProcessor = Objects.requireNonNull(partitionProcessor, "partitionProcessor cannot be null");
+        this.fsViewProcessor = Objects.requireNonNull(fsViewProcessor, "fsViewProcessor cannot be null");
+        this.metaClientProcessor = Objects.requireNonNull(metaClientProcessor, "metaClientProcessor cannot be null");
+    }
+
+    public static HudiCatalogCache fromCatalog(ExternalCatalog catalog, ExecutorService executor) {
+        Objects.requireNonNull(catalog, "catalog cannot be null");
+        Objects.requireNonNull(executor, "executor cannot be null");
+        if (!(catalog instanceof HMSExternalCatalog)) {
+            throw new IllegalArgumentException("Hudi only supports hive(or compatible) catalog now");
+        }
+        HudiCachedPartitionProcessor partitionProcessor = new HudiCachedPartitionProcessor(catalog.getId(), executor,
+                resolveCacheSpec(catalog, PARTITION_MODULE_NAME));
+        HudiCachedFsViewProcessor fsViewProcessor = new HudiCachedFsViewProcessor(executor,
+                resolveCacheSpec(catalog, FS_VIEW_MODULE_NAME));
+        HudiCachedMetaClientProcessor metaClientProcessor = new HudiCachedMetaClientProcessor(executor,
+                resolveCacheSpec(catalog, META_CLIENT_MODULE_NAME));
+        return new HudiCatalogCache(partitionProcessor, fsViewProcessor, metaClientProcessor);
+    }
+
+    public HudiCachedPartitionProcessor getPartitionProcessor() {
+        return partitionProcessor;
+    }
+
+    public HudiCachedFsViewProcessor getFsViewProcessor() {
+        return fsViewProcessor;
+    }
+
+    public HudiCachedMetaClientProcessor getMetaClientProcessor() {
+        return metaClientProcessor;
+    }
+
+    public void cleanUp() {
+        partitionProcessor.cleanUp();
+        fsViewProcessor.cleanUp();
+        metaClientProcessor.cleanUp();
+    }
+
+    private static CacheSpec resolveCacheSpec(ExternalCatalog catalog, String moduleName) {
+        return CacheSpec.fromUnifiedProperties(catalog.getProperties(), HudiEngineCache.ENGINE_TYPE, moduleName, true,
+                Config.external_cache_expire_time_seconds_after_access, Config.max_external_table_cache_num);
+    }
+}

fe/fe-core/src/main/java/org/apache/doris/datasource/hudi/source/HudiMetadataCacheMgr.java

@@ -18,6 +18,7 @@
 package org.apache.doris.datasource.maxcompute;
 
 import org.apache.doris.common.Config;
+import org.apache.doris.datasource.ExternalCatalog;
 import org.apache.doris.datasource.TablePartitionValues;
 import org.apache.doris.datasource.metacache.CacheSpec;
 
@@ -26,11 +27,13 @@
 import com.github.benmanes.caffeine.cache.stats.CacheStats;
 
 import java.util.List;
+import java.util.Objects;
 import java.util.concurrent.TimeUnit;
 import java.util.function.Function;
 import java.util.stream.Collectors;
 
 public class MaxComputeMetadataCache {
+    private static final String PARTITION_VALUES_MODULE_NAME = "partition-values";
     private final Cache<MaxComputeCacheKey, TablePartitionValues> partitionValuesCache;
 
     public MaxComputeMetadataCache(CacheSpec cacheSpec) {
@@ -46,10 +49,18 @@ public MaxComputeMetadataCache(CacheSpec cacheSpec) {
 
     public MaxComputeMetadataCache() {
         this(CacheSpec.fromUnifiedProperties(java.util.Collections.emptyMap(),
-                MaxComputeEngineCache.ENGINE_TYPE, "partition-values", true,
+                MaxComputeEngineCache.ENGINE_TYPE, PARTITION_VALUES_MODULE_NAME, true,
                 Config.external_cache_expire_time_seconds_after_access, Config.max_hive_partition_cache_num));
     }
 
+    public static MaxComputeMetadataCache fromCatalog(ExternalCatalog catalog) {
+        Objects.requireNonNull(catalog, "catalog cannot be null");
+        CacheSpec cacheSpec = CacheSpec.fromUnifiedProperties(catalog.getProperties(),
+                MaxComputeEngineCache.ENGINE_TYPE, PARTITION_VALUES_MODULE_NAME, true,
+                Config.external_cache_expire_time_seconds_after_access, Config.max_hive_partition_cache_num);
+        return new MaxComputeMetadataCache(cacheSpec);
+    }
+
     public TablePartitionValues getCachedPartitionValues(MaxComputeCacheKey tablePartitionKey,
             Function<? super MaxComputeCacheKey, ? extends TablePartitionValues> loader) {
         return partitionValuesCache.get(tablePartitionKey, loader);