[fix](catalog) Support TopN lazy materialization on remote doris catalog

be/src/exec/operator/materialization_opertor.cpp

@@ -494,6 +494,7 @@ Status MaterializationSharedState::init_multi_requests(
     // Initialize the base struct of PMultiGetRequestV2
     multi_get_request.set_be_exec_version(state->be_exec_version());
     multi_get_request.set_wg_id(state->get_query_ctx()->workload_group()->id());
+    multi_get_request.set_cluster_id(ExecEnv::GetInstance()->cluster_info()->cluster_id);
     multi_get_request.set_file_cache_remote_only_on_miss(
             config::is_cloud_mode() &&
             state->query_options().enable_topn_lazy_mat_phase2_no_write_file_cache);

be/src/service/internal_service.cpp

@@ -1999,14 +1999,31 @@ void PInternalService::multiget_data_v2(google::protobuf::RpcController* control
                                         PMultiGetResponseV2* response,
                                         google::protobuf::Closure* done) {
     std::vector<uint64_t> id_set;
-    id_set.push_back(request->wg_id());
+    // Cross-cluster request (e.g. topn lazy materialization over a remote doris catalog):
+    // the wg_id belongs to the sender cluster's own id space and must not be resolved
+    // locally, so the request is always charged to the default(normal) workload group
+    // instead of failing. Id 1 is the common normal group id; if it does not exist under
+    // the new compute-group model, WorkloadGroupMgr::get_group falls back to the group
+    // named "normal" by name, so it always resolves.
+    constexpr uint64_t DEFAULT_WORKLOAD_GROUP_ID = 1;
+    int32_t local_cluster_id = ExecEnv::GetInstance()->cluster_info()->cluster_id;
+    bool cross_cluster = request->has_cluster_id() && request->cluster_id() != 0 &&
+                         local_cluster_id != 0 && request->cluster_id() != local_cluster_id;
+    if (cross_cluster) {
+        id_set.push_back(DEFAULT_WORKLOAD_GROUP_ID);
+        LOG_EVERY_N(INFO, 20) << "receive cross-cluster multiget_data_v2, remote cluster id: "
+                              << request->cluster_id() << ", local cluster id: " << local_cluster_id
+                              << ", fallback to default workload group";
+    } else {
+        id_set.push_back(request->wg_id());
+    }
     auto wg = ExecEnv::GetInstance()->workload_group_mgr()->get_group(id_set);
     Status st = Status::OK();
 
     if (!wg) [[unlikely]] {
         brpc::ClosureGuard closure_guard(done);
         st = Status::Error<TStatusCode::CANCELLED>("fail to find wg: wg id:" +
-                                                   std::to_string(request->wg_id()));
+                                                   std::to_string(id_set[0]));
         st.to_protobuf(response->mutable_status());
         return;
     }

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

@@ -17,6 +17,8 @@
 
 package org.apache.doris.datasource.doris;
 
+import org.apache.doris.catalog.Env;
+import org.apache.doris.catalog.TableIf;
 import org.apache.doris.common.Config;
 import org.apache.doris.common.DdlException;
 import org.apache.doris.datasource.CatalogProperty;
@@ -32,8 +34,11 @@
 
 import java.util.ArrayList;
 import java.util.Arrays;
+import java.util.Collection;
+import java.util.HashSet;
 import java.util.List;
 import java.util.Map;
+import java.util.Set;
 
 public class RemoteDorisExternalCatalog extends ExternalCatalog {
     private static final Logger LOG = LogManager.getLogger(RemoteDorisExternalCatalog.class);
@@ -168,6 +173,42 @@ public boolean useArrowFlight() {
                 "true"));
     }
 
+    /**
+     * Returns the remote olap table behind the given table, or null if the table does not
+     * belong to a remote doris cluster. Covers both access modes: the virtual cluster mode
+     * binds a RemoteOlapTable directly, and the arrow flight mode binds a
+     * RemoteDorisExternalTable wrapping one.
+     */
+    public static RemoteOlapTable getRemoteOlapTable(TableIf table) {
+        if (table instanceof RemoteOlapTable) {
+            return (RemoteOlapTable) table;
+        }
+        if (table instanceof RemoteDorisExternalTable) {
+            return (RemoteOlapTable) ((RemoteDorisExternalTable) table).getOlapTable();
+        }
+        return null;
+    }
+
+    /**
+     * Whether any remote backend id collides with a local backend id, or two remote tables
+     * (e.g. from different remote catalogs) collide with each other. Backend ids of clusters
+     * are independently allocated; on collision the second phase fetch cannot distinguish the
+     * id spaces and would route rows to a wrong backend, so topn lazy materialization must
+     * be skipped.
+     */
+    public static boolean hasRemoteBackendIdConflict(Collection<RemoteOlapTable> remoteTables) {
+        Set<Long> localBackendIds = new HashSet<>(Env.getCurrentSystemInfo().getAllBackendIds());
+        Set<Long> seenRemoteBackendIds = new HashSet<>();
+        for (RemoteOlapTable remoteTable : remoteTables) {
+            for (Long backendId : remoteTable.getAllBackendsByAllCluster().keySet()) {
+                if (localBackendIds.contains(backendId) || !seenRemoteBackendIds.add(backendId)) {
+                    return true;
+                }
+            }
+        }
+        return false;
+    }
+
     @Override
     protected void initLocalObjectsImpl() {
         if (isCompatible()) {

fe/fe-core/src/main/java/org/apache/doris/nereids/glue/translator/PhysicalPlanTranslator.java

@@ -53,6 +53,7 @@
 import org.apache.doris.connector.spi.write.ConnectorWriteSortColumn;
 import org.apache.doris.datasource.ExternalTable;
 import org.apache.doris.datasource.connector.converter.ConnectorColumnConverter;
+import org.apache.doris.datasource.doris.RemoteDorisExternalCatalog;
 import org.apache.doris.datasource.doris.RemoteDorisExternalTable;
 import org.apache.doris.datasource.doris.RemoteOlapTable;
 import org.apache.doris.datasource.doris.source.RemoteDorisScanNode;
@@ -105,6 +106,7 @@
 import org.apache.doris.nereids.trees.plans.Plan;
 import org.apache.doris.nereids.trees.plans.PreAggStatus;
 import org.apache.doris.nereids.trees.plans.algebra.Aggregate;
+import org.apache.doris.nereids.trees.plans.algebra.CatalogRelation;
 import org.apache.doris.nereids.trees.plans.algebra.Relation;
 import org.apache.doris.nereids.trees.plans.commands.merge.MergeOperation;
 import org.apache.doris.nereids.trees.plans.physical.AbstractPhysicalJoin;
@@ -214,6 +216,7 @@
 import org.apache.doris.qe.ConnectContext;
 import org.apache.doris.qe.SessionVariable;
 import org.apache.doris.statistics.StatisticConstants;
+import org.apache.doris.system.Backend;
 import org.apache.doris.tablefunction.TableValuedFunctionIf;
 import org.apache.doris.thrift.TBinlogScanType;
 import org.apache.doris.thrift.TPartitionType;
@@ -2755,14 +2758,34 @@ public PlanFragment visitPhysicalWindow(PhysicalWindow<? extends Plan> physicalW
         return inputPlanFragment;
     }
 
+    /**
+     * Collects backends of the remote doris clusters referenced by the lazy materialized
+     * relations. Rowids of remote tables are generated by the remote cluster's BEs, so the
+     * second phase fetch must be able to reach them. Duplicate relations may resolve to the
+     * same remote table (e.g. self join), so merge by backend id.
+     */
+    private List<Backend> collectRemoteBackends(PhysicalLazyMaterialize<? extends Plan> materialize) {
+        Map<Long, Backend> mergedBackends = Maps.newHashMap();
+        for (Relation relation : materialize.getRelations()) {
+            if (relation instanceof CatalogRelation) {
+                RemoteOlapTable remoteTable = RemoteDorisExternalCatalog.getRemoteOlapTable(
+                        ((CatalogRelation) relation).getTable());
+                if (remoteTable != null) {
+                    mergedBackends.putAll(remoteTable.getAllBackendsByAllCluster());
+                }
+            }
+        }
+        return ImmutableList.copyOf(mergedBackends.values());
+    }
+
     @Override
     public PlanFragment visitPhysicalLazyMaterialize(PhysicalLazyMaterialize<? extends Plan> materialize,
             PlanTranslatorContext context) {
         PlanFragment inputPlanFragment = materialize.child(0).accept(this, context);
         TupleDescriptor materializeTupleDesc = generateTupleDesc(materialize.getOutput(), null, context);
 
         MaterializationNode materializeNode = new MaterializationNode(context.nextPlanNodeId(), materializeTupleDesc,
-                inputPlanFragment.getPlanRoot());
+                inputPlanFragment.getPlanRoot(), collectRemoteBackends(materialize));
 
         List<Expr> rowIds = materialize.getRowIds().stream()
                 .map(e -> ExpressionTranslator.translate(e, context))

fe/fe-core/src/main/java/org/apache/doris/nereids/processor/post/materialize/LazyMaterializeTopN.java

@@ -21,7 +21,10 @@
 import org.apache.doris.catalog.Column;
 import org.apache.doris.catalog.KeysType;
 import org.apache.doris.catalog.OlapTable;
+import org.apache.doris.catalog.TableIf;
 import org.apache.doris.catalog.Type;
+import org.apache.doris.datasource.doris.RemoteDorisExternalCatalog;
+import org.apache.doris.datasource.doris.RemoteOlapTable;
 import org.apache.doris.nereids.CascadesContext;
 import org.apache.doris.nereids.StatementContext;
 import org.apache.doris.nereids.processor.post.PlanPostProcessor;
@@ -159,6 +162,36 @@ private Plan doComputeTopN(PhysicalTopN<? extends Plan> topN, CascadesContext ct
         BiMap<Relation, SlotReference> relationToRowId = HashBiMap.create(relationToLazySlotMap.size());
         HashSet<SlotReference> rowIdSet = new HashSet<>();
         StatementContext threadStatementContext = StatementScopeIdGenerator.getStatementContext();
+        // Rowids of remote doris tables are generated by the remote cluster's backends, and
+        // the second phase fetch goes to those remote backends directly (see
+        // MaterializationNode nodes info and the cross cluster multiget rpc). Backend ids of
+        // clusters are independently allocated; any id collision (remote vs local, or remote
+        // vs remote across catalogs) would silently route the fetch to a wrong backend, so
+        // skip the rewrite and fall back to normal execution. Resolving the remote table may
+        // issue metadata rpcs (arrow flight mode), degrade on failure too.
+        List<RemoteOlapTable> remoteTables = new ArrayList<>();
+        for (Relation relation : relationToLazySlotMap.keySet()) {
+            if (!(relation instanceof CatalogRelation)) {
+                continue;
+            }
+            TableIf relationTable = ((CatalogRelation) relation).getTable();
+            try {
+                RemoteOlapTable remoteTable = RemoteDorisExternalCatalog.getRemoteOlapTable(relationTable);
+                if (remoteTable != null) {
+                    remoteTables.add(remoteTable);
+                }
+            } catch (Exception e) {
+                LOG.warn("Skip TopN lazy materialization: failed to resolve remote doris table {}",
+                        relationTable.getName(), e);
+                return topN;
+            }
+        }
+        if (!remoteTables.isEmpty()
+                && RemoteDorisExternalCatalog.hasRemoteBackendIdConflict(remoteTables)) {
+            LOG.warn("Skip TopN lazy materialization: backend id collides between the local"
+                    + " cluster and remote doris cluster(s), tables={}", remoteTables);
+            return topN;
+        }
         for (Relation relation : relationToLazySlotMap.keySet()) {
             // TopN lazy materialization relies on BE adding a GLOBAL_ROWID_COL to the
             // tablet schema. When light_schema_change=false, the table columns have

fe/fe-core/src/main/java/org/apache/doris/planner/MaterializationNode.java

@@ -38,6 +38,7 @@
 import org.apache.doris.thrift.TPlanNode;
 import org.apache.doris.thrift.TPlanNodeType;
 
+import com.google.common.collect.ImmutableList;
 import com.google.common.collect.Lists;
 
 import java.util.ArrayList;
@@ -137,13 +138,24 @@ public class MaterializationNode extends PlanNode {
     private boolean isTopMaterializeNode;
 
     public MaterializationNode(PlanNodeId id, TupleDescriptor desc, PlanNode child) {
+        this(id, desc, child, ImmutableList.of());
+    }
+
+    /**
+     * @param remoteBackends backends of remote doris clusters (remote doris catalog). The rowids of
+     *        remote tables are generated by the remote cluster's BEs, so the second phase fetch
+     *        must be able to reach them; they are merged into nodes_info besides the local
+     *        compute group backends.
+     */
+    public MaterializationNode(PlanNodeId id, TupleDescriptor desc, PlanNode child,
+            List<Backend> remoteBackends) {
         super(id, desc.getId().asList(), "MaterializeNode");
         this.materializeTupleDescriptor = desc;
-        initNodeInfo();
+        initNodeInfo(remoteBackends);
         this.children.add(child);
     }
 
-    public void initNodeInfo() {
+    public void initNodeInfo(List<Backend> remoteBackends) {
         BeSelectionPolicy policy = new BeSelectionPolicy.Builder()
                 .needQueryAvailable()
                 .setRequireAliveBe()
@@ -157,6 +169,11 @@ public void initNodeInfo() {
         for (Backend backend : policy.getCandidateBackends(computeGroup.getBackendList())) {
             nodesInfo.addToNodes(new TNodeInfo(backend.getId(), 0, backend.getHost(), backend.getBrpcPort()));
         }
+        // remote doris catalog backends; id conflicts are rejected before the plan rewrite
+        // (LazyMaterializeTopN), so no check here.
+        for (Backend backend : remoteBackends) {
+            nodesInfo.addToNodes(new TNodeInfo(backend.getId(), 0, backend.getHost(), backend.getBrpcPort()));
+        }
     }
 
     @Override

fe/fe-core/src/test/java/org/apache/doris/datasource/doris/RemoteDorisExternalCatalogTest.java

@@ -0,0 +1,97 @@
+// 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.system.Backend;
+
+import com.google.common.collect.ImmutableMap;
+import org.junit.After;
+import org.junit.Assert;
+import org.junit.Before;
+import org.junit.Test;
+
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+
+public class RemoteDorisExternalCatalogTest {
+
+    private Backend localBackend1;
+    private Backend localBackend2;
+
+    @Before
+    public void setUp() {
+        localBackend1 = new Backend(1L, "192.168.1.1", 9050);
+        localBackend2 = new Backend(2L, "192.168.1.2", 9050);
+        Env.getCurrentSystemInfo().addBackend(localBackend1);
+        Env.getCurrentSystemInfo().addBackend(localBackend2);
+    }
+
+    @After
+    public void tearDown() throws Exception {
+        Env.getCurrentSystemInfo().dropBackend(localBackend1.getId());
+        Env.getCurrentSystemInfo().dropBackend(localBackend2.getId());
+    }
+
+    private RemoteOlapTable remoteTableWithBackends(Map<Long, Backend> backends) {
+        return new RemoteOlapTable() {
+            @Override
+            public ImmutableMap<Long, Backend> getAllBackendsByAllCluster() {
+                return ImmutableMap.copyOf(backends);
+            }
+        };
+    }
+
+    @Test
+    public void testNoConflict() {
+        RemoteOlapTable remoteTable = remoteTableWithBackends(ImmutableMap.of(
+                100L, new Backend(100L, "10.1.1.1", 9050),
+                101L, new Backend(101L, "10.1.1.2", 9050)));
+        Assert.assertFalse(RemoteDorisExternalCatalog
+                .hasRemoteBackendIdConflict(Collections.singletonList(remoteTable)));
+    }
+
+    @Test
+    public void testConflictWithLocalBackend() {
+        // remote backend id 1 collides with local backend id 1
+        RemoteOlapTable remoteTable = remoteTableWithBackends(ImmutableMap.of(
+                100L, new Backend(100L, "10.1.1.1", 9050),
+                1L, new Backend(1L, "10.1.1.2", 9050)));
+        Assert.assertTrue(RemoteDorisExternalCatalog
+                .hasRemoteBackendIdConflict(Collections.singletonList(remoteTable)));
+    }
+
+    @Test
+    public void testConflictBetweenRemoteTables() {
+        // two remote catalogs independently allocate backend id 200
+        RemoteOlapTable remoteTableA = remoteTableWithBackends(ImmutableMap.of(
+                200L, new Backend(200L, "10.1.1.1", 9050)));
+        RemoteOlapTable remoteTableB = remoteTableWithBackends(ImmutableMap.of(
+                200L, new Backend(200L, "10.2.1.1", 9050)));
+        List<RemoteOlapTable> remoteTables = Arrays.asList(remoteTableA, remoteTableB);
+        Assert.assertTrue(RemoteDorisExternalCatalog.hasRemoteBackendIdConflict(remoteTables));
+    }
+
+    @Test
+    public void testNoRemoteTable() {
+        Assert.assertFalse(RemoteDorisExternalCatalog
+                .hasRemoteBackendIdConflict(Collections.emptyList()));
+    }
+}

gensrc/proto/internal_service.proto

@@ -854,6 +854,10 @@ message PMultiGetRequestV2 {
     optional bool gc_id_map = 4;
     optional uint64 wg_id = 5;
     optional bool file_cache_remote_only_on_miss = 6;
+    // cluster id of the sender cluster. Used by the receiver to detect cross-cluster
+    // requests (e.g. remote doris catalog topn lazy materialization) and fall back to
+    // a local workload group instead of failing.
+    optional int32 cluster_id = 7;
 };
 
 message PMultiGetBlockV2 {