branch-4.0: [Feature] doris cross-cluster query #57898

fe/fe-core/src/main/java/org/apache/doris/catalog/OlapTable.java

@@ -95,6 +95,7 @@
 import com.google.common.base.Joiner;
 import com.google.common.base.Preconditions;
 import com.google.common.base.Strings;
+import com.google.common.collect.ImmutableMap;
 import com.google.common.collect.Lists;
 import com.google.common.collect.Maps;
 import com.google.common.collect.Range;
@@ -182,10 +183,10 @@ public enum OlapTableState {
     private PartitionInfo partitionInfo;
     @SerializedName(value = "itp", alternate = {"idToPartition"})
     @Getter
-    private ConcurrentHashMap<Long, Partition> idToPartition = new ConcurrentHashMap<>();
+    protected ConcurrentHashMap<Long, Partition> idToPartition = new ConcurrentHashMap<>();
     // handled in postgsonprocess
     @Getter
-    private Map<String, Partition> nameToPartition = Maps.newTreeMap();
+    protected Map<String, Partition> nameToPartition = Maps.newTreeMap();
 
     @SerializedName(value = "di", alternate = {"distributionInfo"})
     private DistributionInfo defaultDistributionInfo;
@@ -3698,4 +3699,48 @@ public Index getInvertedIndex(Column column, List<String> subPath) {
                                         .filter(Index::isAnalyzedInvertedIndex).findFirst().orElse(null);
         }
     }
+
+    /**
+     * caller should acquire the read lock and should not modify any field of the return obj
+     */
+    public OlapTable copyTableMeta() {
+        OlapTable table = new OlapTable();
+        // metaobj
+        table.signature = signature;
+        table.lastCheckTime =  lastCheckTime;
+        // abstract table
+        table.id = id;
+        table.name = name;
+        table.qualifiedDbName = qualifiedDbName;
+        table.type = type;
+        table.createTime = createTime;
+        table.fullSchema = fullSchema;
+        table.comment = comment;
+        table.tableAttributes = tableAttributes;
+        // olap table
+        // NOTE: currently do not need temp partitions, colocateGroup, autoIncrementGenerator
+        table.idToPartition = new ConcurrentHashMap<>();
+        table.tempPartitions = new TempPartitions();
+
+        table.state = state;
+        table.indexIdToMeta = ImmutableMap.copyOf(indexIdToMeta);
+        table.indexNameToId = ImmutableMap.copyOf(indexNameToId);
+        table.keysType = keysType;
+        table.partitionInfo = partitionInfo;
+        table.defaultDistributionInfo = defaultDistributionInfo;
+        table.bfColumns = bfColumns;
+        table.bfFpp = bfFpp;
+        table.indexes = indexes;
+        table.baseIndexId = baseIndexId;
+        table.tableProperty = tableProperty;
+        return table;
+    }
+
+    public long getCatalogId() {
+        return Env.getCurrentInternalCatalog().getId();
+    }
+
+    public ImmutableMap<Long, Backend> getAllBackendsByAllCluster() throws AnalysisException {
+        return Env.getCurrentSystemInfo().getAllBackendsByAllCluster();
+    }
 }

fe/fe-core/src/main/java/org/apache/doris/catalog/TableIf.java

@@ -437,6 +437,8 @@ public String toEngineName() {
                     return "iceberg";
                 case DICTIONARY:
                     return "dictionary";
+                case DORIS_EXTERNAL_TABLE:
+                    return "External_Doris";
                 default:
                     return null;
             }

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

@@ -22,6 +22,7 @@
 import org.apache.doris.common.Config;
 import org.apache.doris.common.Pair;
 import org.apache.doris.common.ThreadPoolManager;
+import org.apache.doris.datasource.doris.DorisExternalMetaCacheMgr;
 import org.apache.doris.datasource.hive.HMSExternalCatalog;
 import org.apache.doris.datasource.hive.HMSExternalTable;
 import org.apache.doris.datasource.hive.HiveMetaStoreCache;
@@ -100,6 +101,7 @@ public class ExternalMetaCacheMgr {
     private final IcebergMetadataCacheMgr icebergMetadataCacheMgr;
     private final MaxComputeMetadataCacheMgr maxComputeMetadataCacheMgr;
     private final PaimonMetadataCacheMgr paimonMetadataCacheMgr;
+    private final DorisExternalMetaCacheMgr dorisExternalMetaCacheMgr;
 
     public ExternalMetaCacheMgr(boolean isCheckpointCatalog) {
         rowCountRefreshExecutor = newThreadPool(isCheckpointCatalog,
@@ -131,6 +133,7 @@ public ExternalMetaCacheMgr(boolean isCheckpointCatalog) {
         icebergMetadataCacheMgr = new IcebergMetadataCacheMgr(commonRefreshExecutor);
         maxComputeMetadataCacheMgr = new MaxComputeMetadataCacheMgr();
         paimonMetadataCacheMgr = new PaimonMetadataCacheMgr(commonRefreshExecutor);
+        dorisExternalMetaCacheMgr = new DorisExternalMetaCacheMgr(commonRefreshExecutor);
     }
 
     private ExecutorService newThreadPool(boolean isCheckpointCatalog, int numThread, int queueSize,
@@ -219,6 +222,10 @@ public ExternalRowCountCache getRowCountCache() {
         return rowCountCache;
     }
 
+    public DorisExternalMetaCacheMgr getDorisExternalMetaCacheMgr() {
+        return dorisExternalMetaCacheMgr;
+    }
+
     public void removeCache(long catalogId) {
         if (cacheMap.remove(catalogId) != null) {
             LOG.info("remove hive metastore cache for catalog {}", catalogId);
@@ -232,6 +239,7 @@ public void removeCache(long catalogId) {
         icebergMetadataCacheMgr.removeCache(catalogId);
         maxComputeMetadataCacheMgr.removeCache(catalogId);
         paimonMetadataCacheMgr.removeCache(catalogId);
+        dorisExternalMetaCacheMgr.removeCache(catalogId);
     }
 
     public void invalidateTableCache(ExternalTable dorisTable) {
@@ -288,6 +296,7 @@ public void invalidateCatalogCache(long catalogId) {
         icebergMetadataCacheMgr.invalidateCatalogCache(catalogId);
         maxComputeMetadataCacheMgr.invalidateCatalogCache(catalogId);
         paimonMetadataCacheMgr.invalidateCatalogCache(catalogId);
+        dorisExternalMetaCacheMgr.invalidateCatalogCache(catalogId);
         if (LOG.isDebugEnabled()) {
             LOG.debug("invalid catalog cache for {}", catalogId);
         }

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

@@ -0,0 +1,83 @@
+// 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.doris;
+
+import org.apache.doris.catalog.Env;
+import org.apache.doris.common.CacheFactory;
+import org.apache.doris.common.Config;
+import org.apache.doris.system.Backend;
+
+import com.github.benmanes.caffeine.cache.LoadingCache;
+import com.google.common.collect.ImmutableMap;
+import com.google.common.collect.Maps;
+import org.apache.logging.log4j.LogManager;
+import org.apache.logging.log4j.Logger;
+
+import java.util.List;
+import java.util.Map;
+import java.util.OptionalLong;
+import java.util.concurrent.ExecutorService;
+import java.util.stream.Collectors;
+
+public class DorisExternalMetaCacheMgr {
+    private static final Logger LOG = LogManager.getLogger(DorisExternalMetaCacheMgr.class);
+    private final LoadingCache<Long, ImmutableMap<Long, Backend>> backendsCache;
+
+    public DorisExternalMetaCacheMgr(ExecutorService executor) {
+        CacheFactory cacheFactory = new CacheFactory(
+                OptionalLong.of(Config.external_cache_expire_time_seconds_after_access),
+                OptionalLong.of(Config.external_cache_refresh_time_minutes * 60),
+                20,
+                true,
+                null);
+        backendsCache = cacheFactory.buildCache(key -> loadBackends(key), executor);
+    }
+
+    private ImmutableMap<Long, Backend> loadBackends(Long catalogId) {
+        RemoteDorisExternalCatalog catalog = (RemoteDorisExternalCatalog) Env.getCurrentEnv().getCatalogMgr()
+                .getCatalog(catalogId);
+        List<Backend> backends = catalog.getFeServiceClient().listBackends();
+        if (LOG.isDebugEnabled()) {
+            List<String> names = backends.stream().map(b -> b.getAddress()).collect(Collectors.toList());
+            LOG.debug("load backends:{} from:{}", String.join(",", names), catalog.getName());
+        }
+        Map<Long, Backend> backendMap = Maps.newHashMap();
+        backends.forEach(backend -> backendMap.put(backend.getId(), backend));
+        return ImmutableMap.copyOf(backendMap);
+    }
+
+    public void removeCache(long catalogId) {
+        backendsCache.invalidate(catalogId);
+    }
+
+    public void invalidateBackendCache(long catalogId) {
+        backendsCache.invalidate(catalogId);
+    }
+
+    public void invalidateCatalogCache(long catalogId) {
+        invalidateBackendCache(catalogId);
+    }
+
+    public ImmutableMap<Long, Backend> getBackends(long catalogId) {
+        ImmutableMap<Long, Backend> backends = backendsCache.get(catalogId);
+        if (backends == null) {
+            return ImmutableMap.of();
+        }
+        return backends;
+    }
+}