github-actions[bot] commented on code in PR #67518:
URL: https://github.com/apache/doris/pull/67518#discussion_r4043664242
##########
fe/fe-core/src/main/java/org/apache/doris/nereids/processor/post/materialize/LazyMaterializeTopN.java:
##########
@@ -247,6 +309,42 @@ static List<Slot> collectRequiredOutputSlots(Map<Slot,
MaterializeSource> materi
return requiredOutputSlots;
}
+ /**
+ * Builds the phase-2 fetch address book and rejects backend id collisions
in one pass.
+ * Local backends go through the full selection policy (alive, query
available, and the
+ * SimpleScheduler blacklist keyed by the numeric backend id); ids are
unique within one
+ * cluster, so the filtered result directly forms the base of the address
book. Remote
+ * doris catalog backends are numbered by an independent id space and must
skip that
+ * blacklist: a dead local backend would otherwise evict a healthy remote
backend sharing
+ * the same id. They are only filtered by query availability, which on a
+ * Backend.fromThrift object degrades to the alive flag, and each of them
must claim an
+ * id not used by the base or by an earlier remote backend.
+ *
+ * @return null when any backend id appears in more than one cluster's
contribution; the
+ * collision would silently route the fetch to a wrong backend, so
the caller
+ * must skip the lazy materialization rewrite.
+ */
+ @VisibleForTesting
+ static List<Backend> buildFetchBackends(List<Backend> localBackends,
List<Backend> remoteBackends) {
+ BeSelectionPolicy policy = new BeSelectionPolicy.Builder()
+ .needQueryAvailable()
+ .setRequireAliveBe()
+ .build();
+ List<Backend> fetchBackends = new
ArrayList<>(policy.getCandidateBackends(localBackends));
+ for (Backend backend : remoteBackends) {
+ if (!backend.isQueryAvailable()) {
+ continue;
+ }
+ boolean idTaken = fetchBackends.stream()
+ .anyMatch(existing -> existing.getId() == backend.getId());
Review Comment:
[P2] Avoid making this per-query planner path quadratic in the catalog-wide
backend count. For every query-available remote backend, this rescans all
selected local and previously accepted remote entries, yielding `L*R +
R*(R-1)/2` ID comparisons (plus a new stream pipeline each time), and multiple
large remote catalogs feed their full topology here. Please seed a
`HashSet<Long>` with the selected local IDs and use
`seenIds.add(backend.getId())` while appending remotes; that preserves the
collision check in expected `O(L+R)` time.
##########
be/src/service/internal_service.cpp:
##########
@@ -1994,14 +1995,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 &&
Review Comment:
[P1] Synchronize access to the cluster ID before using it in this RPC path.
`ClusterInfo::cluster_id` is a plain `int32_t`; BRPC is started before
`HeartbeatServer::init_cluster_id()` writes it, and the first heartbeat can
later replace `-1` under a mutex that neither this new receiver read nor the
sender read in `materialization_opertor.cpp` takes. There is therefore no
happens-before edge between the heartbeat writers and these BRPC workers, so
the new reads introduce a C++ data race/undefined behavior and can consume the
startup or stale value when choosing workload-group locality. Please
publish/read this field through an atomic or locked accessor and initialize it
before exposing BRPC where possible; the post-start heartbeat transition must
be synchronized too.
##########
regression-test/suites/external_table_p0/remote_doris/test_remote_doris_topn_lazy_materialization.groovy:
##########
@@ -0,0 +1,159 @@
+// 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"
+// (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 the virtual cluster catalog
(use_arrow_flight=false, the
+// only mode whose tables pass MaterializeProbeVisitor) and compares the
results with querying
+// the local table directly.
+//
+// Coverage note: for a query whose lazy-materialized relations are all remote
tables, the
+// address book contains only the remote cluster's backends (the local cluster
is not even a
+// participant), so the backend-id conflict guard does not fire and the
optimized two-phase
+// path runs end to end: phase-1 rowids, cross-cluster nodes_info, cluster_id
tagging and the
+// phase-2 multiget fetch are all exercised. Queries mixing a local table with
a remote table
+// fall back to normal execution when the id spaces collide (e.g. on a
self-referencing
+// docker environment where both sides are the same cluster).
+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";
+ def remote_doris_arrow_port = showres[0][6]
+ def remote_doris_http_port = showres[0][3]
+ def remote_doris_thrift_port = showres[0][5]
+ log.info("show frontends log = ${showres}, arrow:
${remote_doris_arrow_port}, "
+ + "http: ${remote_doris_http_port}, thrift:
${remote_doris_thrift_port}")
+
+ def db_name = "test_remote_doris_topn_lazy_materialization_db"
+ def table_name = "remote_topn_t"
+ def arrow_catalog = "test_remote_doris_topn_arrow_catalog"
+ def olap_catalog = "test_remote_doris_topn_olap_catalog"
+
+ sql """DROP CATALOG IF EXISTS `${arrow_catalog}`"""
+ sql """DROP CATALOG IF EXISTS `${olap_catalog}`"""
+ sql """DROP DATABASE IF EXISTS ${db_name}"""
+ sql """CREATE DATABASE IF NOT EXISTS ${db_name}"""
+
+ sql """
+ CREATE TABLE `${db_name}`.`${table_name}` (
+ `id` INT NOT NULL,
+ `k1` INT NOT NULL,
+ `v1` VARCHAR(64) NULL,
+ `v2` DOUBLE NULL
+ ) ENGINE=OLAP
+ DUPLICATE KEY(`id`)
+ DISTRIBUTED BY HASH(`id`) BUCKETS 1
+ PROPERTIES (
+ "replication_allocation" = "tag.location.default: 1"
+ );
+ """
+
+ // k1 is reverse of id, so ORDER BY k1 returns rows in descending id order.
+ StringBuilder values = new StringBuilder()
+ for (int i = 1; i <= 20; i++) {
+ if (i > 1) {
+ values.append(",")
+ }
+ values.append("(${i}, ${21 - i}, 'str_${i}', ${i * 1.5})")
+ }
+ sql """INSERT INTO `${db_name}`.`${table_name}` VALUES
${values.toString()}"""
+
+ // arrow flight mode: not supported by this fix (rejected by
MaterializeProbeVisitor).
+ // The catalog is still created to pin that boundary in the explain
assertion below.
+ sql """
+ CREATE CATALOG `${arrow_catalog}` PROPERTIES (
+ 'type' = 'doris',
+ 'fe_http_hosts' =
'http://${remote_doris_host}:${remote_doris_http_port}',
+ 'fe_arrow_hosts' =
'${remote_doris_host}:${remote_doris_arrow_port}',
+ 'fe_thrift_hosts' =
'${remote_doris_host}:${remote_doris_thrift_port}',
+ 'user' = '${remote_doris_user}',
+ 'password' = '${remote_doris_psw}',
+ 'use_arrow_flight' = 'true'
+ );
+ """
+
+ // virtual cluster mode: the remote table is bound as a RemoteOlapTable
(OlapScan)
+ sql """
+ CREATE CATALOG `${olap_catalog}` PROPERTIES (
+ 'type' = 'doris',
+ 'fe_http_hosts' =
'http://${remote_doris_host}:${remote_doris_http_port}',
+ 'fe_arrow_hosts' =
'${remote_doris_host}:${remote_doris_arrow_port}',
+ 'fe_thrift_hosts' =
'${remote_doris_host}:${remote_doris_thrift_port}',
+ 'user' = '${remote_doris_user}',
+ 'password' = '${remote_doris_psw}',
+ 'use_arrow_flight' = 'false'
+ );
+ """
+
+ String localRef = "`${db_name}`.`${table_name}`"
+ // topn lazy materialization is triggered when limit < 1024 (default
thresholds)
+ def topnTemplates = [
+ "SELECT * FROM %s ORDER BY k1 LIMIT 5",
+ // the shape reported in issue 63526: predicate + order by + limit
+ "SELECT id, k1, v1, v2 FROM %s WHERE id > 3 ORDER BY k1 LIMIT 5",
+ // projection variant
+ "SELECT v1, v2 FROM %s WHERE id > 5 ORDER BY k1 LIMIT 8",
+ "SELECT * FROM %s ORDER BY k1 LIMIT 1"
+ ]
+
+ def withDistributeHint = { String query ->
+ return query.replaceFirst("(?i)^SELECT ",
+ "SELECT /*+ SET_VAR(enable_nereids_distribute_planner=true) */
")
+ }
+
+ def compareTopn = { String catalogName, String tableRef ->
+ for (String template : topnTemplates) {
+ String localQuery = withDistributeHint(String.format(template,
localRef))
+ String remoteQuery = withDistributeHint(String.format(template,
tableRef))
+ def localRes = sql localQuery
+ def remoteRes = sql remoteQuery
+ log.info("topn query on ${catalogName}: ${remoteQuery}")
+ assertEquals(localRes, remoteRes,
+ "topn result mismatch on ${catalogName}: ${remoteQuery}")
+ }
+ }
+
+ compareTopn("virtual_cluster_catalog",
"`${olap_catalog}`.`${db_name}`.`${table_name}`")
+
+ // Plan shape oracles. Both the local table and the virtual cluster
catalog must produce
+ // the MaterializationNode: the local one proves the optimization is
enabled in this
+ // environment at all, and the remote one proves the remote-only rewrite
runs (its address
+ // book contains only the remote cluster's backends, so the local ids
never collide with
+ // it and the two-phase path executes). The arrow flight catalog is
rejected by
+ // MaterializeProbeVisitor and must stay on normal execution.
+ explain {
+ sql(""" verbose SELECT * FROM ${localRef} ORDER BY k1 LIMIT 5 """)
+ contains("VMaterializeNode")
+ }
+ explain {
+ sql(""" verbose SELECT /*+
SET_VAR(enable_nereids_distribute_planner=true) */ *
+ FROM `${olap_catalog}`.`${db_name}`.`${table_name}` ORDER BY
k1 LIMIT 5 """)
+ contains("VMaterializeNode")
+ }
+ explain {
+ sql(""" verbose SELECT * FROM
`${arrow_catalog}`.`${db_name}`.`${table_name}` ORDER BY k1 LIMIT 5 """)
+ notContains("VMaterializeNode")
+ }
+
+ sql """ DROP DATABASE IF EXISTS ${db_name} """
+ sql """ DROP CATALOG IF EXISTS `${arrow_catalog}` """
Review Comment:
[P2] Keep this regression's final state available for debugging. The suite
already drops both catalogs and the database before creation at lines 50-52, so
rerun isolation is covered; dropping the database here also deletes the test
table, and the following catalog drops discard the remote objects needed to
inspect or reproduce the final plan/result state. This conflicts with the
repository's mandatory drop-before/not-after regression-test rule. Please
remove these three trailing cleanup statements.
##########
regression-test/suites/external_table_p0/remote_doris/test_remote_doris_topn_lazy_materialization.groovy:
##########
@@ -0,0 +1,159 @@
+// 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"
+// (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 the virtual cluster catalog
(use_arrow_flight=false, the
+// only mode whose tables pass MaterializeProbeVisitor) and compares the
results with querying
+// the local table directly.
+//
+// Coverage note: for a query whose lazy-materialized relations are all remote
tables, the
+// address book contains only the remote cluster's backends (the local cluster
is not even a
+// participant), so the backend-id conflict guard does not fire and the
optimized two-phase
+// path runs end to end: phase-1 rowids, cross-cluster nodes_info, cluster_id
tagging and the
+// phase-2 multiget fetch are all exercised. Queries mixing a local table with
a remote table
+// fall back to normal execution when the id spaces collide (e.g. on a
self-referencing
+// docker environment where both sides are the same cluster).
+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";
+ def remote_doris_arrow_port = showres[0][6]
+ def remote_doris_http_port = showres[0][3]
+ def remote_doris_thrift_port = showres[0][5]
+ log.info("show frontends log = ${showres}, arrow:
${remote_doris_arrow_port}, "
+ + "http: ${remote_doris_http_port}, thrift:
${remote_doris_thrift_port}")
+
+ def db_name = "test_remote_doris_topn_lazy_materialization_db"
+ def table_name = "remote_topn_t"
+ def arrow_catalog = "test_remote_doris_topn_arrow_catalog"
+ def olap_catalog = "test_remote_doris_topn_olap_catalog"
+
+ sql """DROP CATALOG IF EXISTS `${arrow_catalog}`"""
+ sql """DROP CATALOG IF EXISTS `${olap_catalog}`"""
+ sql """DROP DATABASE IF EXISTS ${db_name}"""
+ sql """CREATE DATABASE IF NOT EXISTS ${db_name}"""
+
+ sql """
+ CREATE TABLE `${db_name}`.`${table_name}` (
+ `id` INT NOT NULL,
+ `k1` INT NOT NULL,
+ `v1` VARCHAR(64) NULL,
+ `v2` DOUBLE NULL
+ ) ENGINE=OLAP
+ DUPLICATE KEY(`id`)
+ DISTRIBUTED BY HASH(`id`) BUCKETS 1
+ PROPERTIES (
+ "replication_allocation" = "tag.location.default: 1"
+ );
+ """
+
+ // k1 is reverse of id, so ORDER BY k1 returns rows in descending id order.
+ StringBuilder values = new StringBuilder()
+ for (int i = 1; i <= 20; i++) {
+ if (i > 1) {
+ values.append(",")
+ }
+ values.append("(${i}, ${21 - i}, 'str_${i}', ${i * 1.5})")
+ }
+ sql """INSERT INTO `${db_name}`.`${table_name}` VALUES
${values.toString()}"""
+
+ // arrow flight mode: not supported by this fix (rejected by
MaterializeProbeVisitor).
+ // The catalog is still created to pin that boundary in the explain
assertion below.
+ sql """
+ CREATE CATALOG `${arrow_catalog}` PROPERTIES (
+ 'type' = 'doris',
+ 'fe_http_hosts' =
'http://${remote_doris_host}:${remote_doris_http_port}',
+ 'fe_arrow_hosts' =
'${remote_doris_host}:${remote_doris_arrow_port}',
+ 'fe_thrift_hosts' =
'${remote_doris_host}:${remote_doris_thrift_port}',
+ 'user' = '${remote_doris_user}',
+ 'password' = '${remote_doris_psw}',
+ 'use_arrow_flight' = 'true'
+ );
+ """
+
+ // virtual cluster mode: the remote table is bound as a RemoteOlapTable
(OlapScan)
+ sql """
+ CREATE CATALOG `${olap_catalog}` PROPERTIES (
+ 'type' = 'doris',
+ 'fe_http_hosts' =
'http://${remote_doris_host}:${remote_doris_http_port}',
+ 'fe_arrow_hosts' =
'${remote_doris_host}:${remote_doris_arrow_port}',
+ 'fe_thrift_hosts' =
'${remote_doris_host}:${remote_doris_thrift_port}',
+ 'user' = '${remote_doris_user}',
+ 'password' = '${remote_doris_psw}',
+ 'use_arrow_flight' = 'false'
+ );
+ """
+
+ String localRef = "`${db_name}`.`${table_name}`"
+ // topn lazy materialization is triggered when limit < 1024 (default
thresholds)
+ def topnTemplates = [
+ "SELECT * FROM %s ORDER BY k1 LIMIT 5",
+ // the shape reported in issue 63526: predicate + order by + limit
+ "SELECT id, k1, v1, v2 FROM %s WHERE id > 3 ORDER BY k1 LIMIT 5",
+ // projection variant
+ "SELECT v1, v2 FROM %s WHERE id > 5 ORDER BY k1 LIMIT 8",
+ "SELECT * FROM %s ORDER BY k1 LIMIT 1"
+ ]
+
+ def withDistributeHint = { String query ->
+ return query.replaceFirst("(?i)^SELECT ",
+ "SELECT /*+ SET_VAR(enable_nereids_distribute_planner=true) */
")
+ }
+
+ def compareTopn = { String catalogName, String tableRef ->
+ for (String template : topnTemplates) {
+ String localQuery = withDistributeHint(String.format(template,
localRef))
+ String remoteQuery = withDistributeHint(String.format(template,
tableRef))
+ def localRes = sql localQuery
+ def remoteRes = sql remoteQuery
+ log.info("topn query on ${catalogName}: ${remoteQuery}")
+ assertEquals(localRes, remoteRes,
+ "topn result mismatch on ${catalogName}: ${remoteQuery}")
Review Comment:
[P2] Pin the expected rows instead of making equality between two executions
the only result oracle. The seeded 20-row data and unique `ORDER BY k1` make
these results deterministic, but the EXPLAIN checks deliberately require both
the local and remote queries to use `VMaterializeNode`; a shared row-id decode,
merge, projection, or ordering defect can therefore return the same wrong rows
and pass this assertion. Please add separately named `qt`/`order_qt` cases for
the remote queries and commit the generated `.out` results (the local
comparison can remain as an additional check).
--
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]
