github-actions[bot] commented on code in PR #67518:
URL: https://github.com/apache/doris/pull/67518#discussion_r4035872548


##########
fe/fe-core/src/main/java/org/apache/doris/nereids/processor/post/materialize/LazyMaterializeTopN.java:
##########
@@ -159,6 +166,61 @@ 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 backends directly (see 
MaterializationNode
+        // nodes info and the cross cluster multiget rpc). Compute the whole 
fetch address
+        // book here, once, and carry it on the PhysicalLazyMaterialize node, 
so the conflict
+        // guard and the address book can never drift apart. Tables of the 
same remote
+        // catalog share one backend map, so record the catalog only: a self 
join or two
+        // tables of one catalog is not an id conflict.
+        Set<RemoteDorisExternalCatalog> remoteCatalogs = new HashSet<>();
+        boolean involvesLocalCluster = false;
+        for (Relation relation : relationToLazySlotMap.keySet()) {
+            if (relation instanceof PhysicalTVFRelation) {
+                // TVF rowids are fetched from local backends.
+                involvesLocalCluster = true;
+                continue;
+            }
+            if (!(relation instanceof CatalogRelation)) {
+                continue;
+            }
+            TableIf relationTable = ((CatalogRelation) relation).getTable();
+            // A remote doris table reaching here is always a RemoteOlapTable 
bound in the
+            // virtual cluster mode; the arrow flight mode binds a 
RemoteDorisExternalTable
+            // which is rejected by MaterializeProbeVisitor beforehand.
+            if (relationTable instanceof RemoteOlapTable) {
+                remoteCatalogs.add(((RemoteOlapTable) 
relationTable).getCatalog());
+            } else {
+                involvesLocalCluster = true;
+            }
+        }
+        // Pass the raw compute group backends; the availability filtering 
(and the local
+        // blacklist semantics) is applied inside buildFetchBackends. 
Resolution failure
+        // degrades to normal execution here instead of failing the query at 
translate time.
+        List<Backend> localBackends = ImmutableList.of();
+        if (involvesLocalCluster) {
+            try {
+                ConnectContext context = ConnectContext.get();
+                if (context == null) {
+                    context = new ConnectContext();
+                }
+                ComputeGroup computeGroup = context.getComputeGroupSafely();
+                localBackends = computeGroup.getBackendList();
+            } catch (Exception e) {
+                LOG.warn("Skip TopN lazy materialization: failed to resolve 
local fetch backends", e);
+                return topN;
+            }
+        }
+        List<Backend> remoteBackends = new ArrayList<>();
+        for (RemoteDorisExternalCatalog remoteCatalog : remoteCatalogs) {
+            remoteBackends.addAll(remoteCatalog.getAllBackends().values());
+        }

Review Comment:
   [P1] Gate this rewrite on a `be_exec_version` supported by every remote BE. 
`CoordinatorContext` always sets the query version from this FE's 
`Config.be_exec_version`, but a one-BE virtual-cluster `OlapScan` is assigned 
first and the gather fragment can reuse that same remote worker. With local 
exchange enabled, phase-1 rows stay in memory, and the MySQL result sink uses 
Thrift rather than PBlock, so a release-mode older BE need not validate the 
newer version before phase 2. Older BEs already implement `PMultiGetRequestV2` 
and the materialization operator; that operator forwards the query version, and 
`RowIdStorageReader` calls `Block::serialize` with it, which rejects a value 
above the receiver's maximum. Expose/negotiate the remote capability or skip 
lazy materialization when it is unknown, and add a mixed-version one-BE 
remote-catalog test.



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: [email protected]

For queries about this service, please contact Infrastructure at:
[email protected]


---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]

Reply via email to