github-actions[bot] commented on code in PR #67518:
URL: https://github.com/apache/doris/pull/67518#discussion_r3963656513
##########
fe/fe-core/src/main/java/org/apache/doris/nereids/processor/post/materialize/LazyMaterializeTopN.java:
##########
@@ -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);
Review Comment:
[P1] This Arrow Flight branch is unreachable. `BindRelation` represents
`use_arrow_flight=true` as `PhysicalFileScan(RemoteDorisExternalTable)`, but
`MaterializeProbeVisitor.checkRelationTableSupportedType` admits only the exact
`OlapTable.class` or a capable `PluginDrivenExternalTable`; this wrapper is
neither. Thus every candidate slot stays eager and `doComputeTopN` returns
before `relationToLazySlotMap` reaches this code. Virtual mode works through
the separate `PhysicalOlapScan` visitor, but Arrow mode is still never lazily
materialized. Please admit the wrapper by resolving its `RemoteOlapTable`
during probing and apply the same AGG_KEYS, light-schema-change, and
sequence-map safety gates, with an EXPLAIN shape assertion.
##########
fe/fe-core/src/main/java/org/apache/doris/datasource/doris/RemoteDorisExternalCatalog.java:
##########
@@ -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)) {
Review Comment:
[P2] This treats ids as conflicting even when the address map for this plan
would be unambiguous. A self-join or two tables from one remote catalog
contributes the same catalog-wide map twice, so the second relation fails here
despite mapping every id to the identical endpoint (and the translator
explicitly merges that case). The local set is also overbroad:
`getAllBackendIds()` includes dead BEs and other compute/resource groups, while
`MaterializationNode` includes only alive, query-available BEs from the
selected compute group. Please compare the effective id-to-endpoint mappings
used by `nodes_info`, accepting exact duplicates and rejecting only genuinely
ambiguous endpoints; cover both same-catalog joins and irrelevant-local-BE
collisions.
##########
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;
Review Comment:
[P1] This field does not reliably prove that the request is local. An old
sender omits it and an old receiver ignores it, so either mixed-version
direction still resolves the sender's numeric `wg_id`; even two new clusters
can have equal `cluster_id` values because the value is user-configurable or a
random 31-bit integer, not a globally coordinated identity. In every case
`WorkloadGroupMgr::get_group` prefers an exact local id before falling back to
`normal`, so a same-numbered unrelated group can reject or throttle the fetch.
Please mark remoteness at the plan/request boundary or use a non-colliding
identity/capability scheme, with both mixed-version directions and equal-id
all-new clusters tested.
##########
regression-test/suites/external_table_p0/remote_doris/test_remote_doris_topn_lazy_materialization.groovy:
##########
@@ -0,0 +1,131 @@
+// 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.
+
+// Regression test for issue apache/doris#63526: TopN (ORDER BY ... LIMIT)
over a remote
+// doris catalog used to fail with "MaterializationSinkOperatorX failed to
find rpc_struct"
+// (arrow flight mode) or "Miss matched return row loc count" (virtual cluster
mode), because
+// the rowids of remote tables encode the remote cluster's backend ids while
the second phase
+// fetch address book only contained local backends.
+// This test runs TopN queries over both catalog modes and compares the
results with querying
+// the local table directly.
+suite("test_remote_doris_topn_lazy_materialization",
"p0,external,doris,external_docker,external_docker_doris") {
+ String remote_doris_host =
context.config.otherConfigs.get("extArrowFlightSqlHost")
+ String remote_doris_user =
context.config.otherConfigs.get("extArrowFlightSqlUser")
+ String remote_doris_psw =
context.config.otherConfigs.get("extArrowFlightSqlPassword")
+
+ def showres = sql "show frontends";
Review Comment:
[P1] This case points both catalogs back at the suite's own FE (the table is
created on the current connection and every port comes from its `SHOW
FRONTENDS`). `getBackendMeta` therefore returns the same backend ids as
`Env.getCurrentSystemInfo()`, so the new conflict check returns the original
TopN before `PhysicalLazyMaterialize` is built. The row comparison can pass
without exercising remote `nodes_info`, `cluster_id`, or phase-2 fetch. Please
run this against a genuinely separate cluster with non-colliding ids and assert
the shape contains `PhysicalLazyMaterialize` for both modes; keep a separate
assertion for the collision fallback if desired.
##########
fe/fe-core/src/main/java/org/apache/doris/planner/MaterializationNode.java:
##########
@@ -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) {
Review Comment:
[P1] Please avoid adding/contacting every remote backend unconditionally.
`getBackendMeta` includes dead BEs, this loop bypasses the availability policy
used above, and the BE eagerly resolves every advertised host and sends each
batch to every map entry. An unresolvable hostname fails `init_multi_requests`
even when that BE owns no surviving row; a transport failure for a zero-row
request is ignored only after `counter.wait()` with the full execution timeout.
This is especially harmful when a fragment in one cluster sees private
endpoints from both clusters. Filter the remote list and/or create/send RPCs
only for backend ids present in the row-id batch.
--
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]
