diff --git a/be/src/exec/operator/materialization_opertor.cpp b/be/src/exec/operator/materialization_opertor.cpp index f8871c533c5c2d..a0189b2ff49af4 100644 --- a/be/src/exec/operator/materialization_opertor.cpp +++ b/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); diff --git a/be/src/service/internal_service.cpp b/be/src/service/internal_service.cpp index 9776461a61ae82..58964171f9a54b 100644 --- a/be/src/service/internal_service.cpp +++ b/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 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("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; } diff --git a/fe/fe-core/src/main/java/org/apache/doris/datasource/doris/RemoteDorisExternalCatalog.java b/fe/fe-core/src/main/java/org/apache/doris/datasource/doris/RemoteDorisExternalCatalog.java index e48c07303728e2..5b1d60901c6c8e 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/datasource/doris/RemoteDorisExternalCatalog.java +++ b/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 remoteTables) { + Set localBackendIds = new HashSet<>(Env.getCurrentSystemInfo().getAllBackendIds()); + Set 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()) { diff --git a/fe/fe-core/src/main/java/org/apache/doris/nereids/glue/translator/PhysicalPlanTranslator.java b/fe/fe-core/src/main/java/org/apache/doris/nereids/glue/translator/PhysicalPlanTranslator.java index 1d6923a409f913..ff69a53b9179bb 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/nereids/glue/translator/PhysicalPlanTranslator.java +++ b/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,6 +2758,26 @@ public PlanFragment visitPhysicalWindow(PhysicalWindow 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 collectRemoteBackends(PhysicalLazyMaterialize materialize) { + Map 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 materialize, PlanTranslatorContext context) { @@ -2762,7 +2785,7 @@ public PlanFragment visitPhysicalLazyMaterialize(PhysicalLazyMaterialize rowIds = materialize.getRowIds().stream() .map(e -> ExpressionTranslator.translate(e, context)) diff --git a/fe/fe-core/src/main/java/org/apache/doris/nereids/processor/post/materialize/LazyMaterializeTopN.java b/fe/fe-core/src/main/java/org/apache/doris/nereids/processor/post/materialize/LazyMaterializeTopN.java index 9acec57125a4b2..0b0d8482cbe4b6 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/nereids/processor/post/materialize/LazyMaterializeTopN.java +++ b/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 topN, CascadesContext ct BiMap relationToRowId = HashBiMap.create(relationToLazySlotMap.size()); HashSet 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 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 diff --git a/fe/fe-core/src/main/java/org/apache/doris/planner/MaterializationNode.java b/fe/fe-core/src/main/java/org/apache/doris/planner/MaterializationNode.java index 245f44b8c6a985..732dcc77f57be6 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/planner/MaterializationNode.java +++ b/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 remoteBackends) { super(id, desc.getId().asList(), "MaterializeNode"); this.materializeTupleDescriptor = desc; - initNodeInfo(); + initNodeInfo(remoteBackends); this.children.add(child); } - public void initNodeInfo() { + public void initNodeInfo(List 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 diff --git a/fe/fe-core/src/test/java/org/apache/doris/datasource/doris/RemoteDorisExternalCatalogTest.java b/fe/fe-core/src/test/java/org/apache/doris/datasource/doris/RemoteDorisExternalCatalogTest.java new file mode 100644 index 00000000000000..913a69368149a4 --- /dev/null +++ b/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 backends) { + return new RemoteOlapTable() { + @Override + public ImmutableMap 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 remoteTables = Arrays.asList(remoteTableA, remoteTableB); + Assert.assertTrue(RemoteDorisExternalCatalog.hasRemoteBackendIdConflict(remoteTables)); + } + + @Test + public void testNoRemoteTable() { + Assert.assertFalse(RemoteDorisExternalCatalog + .hasRemoteBackendIdConflict(Collections.emptyList())); + } +} diff --git a/gensrc/proto/internal_service.proto b/gensrc/proto/internal_service.proto index 00cd171dcc03d5..2eaf289ca03cf7 100644 --- a/gensrc/proto/internal_service.proto +++ b/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 { diff --git a/regression-test/suites/external_table_p0/remote_doris/test_remote_doris_topn_lazy_materialization.groovy b/regression-test/suites/external_table_p0/remote_doris/test_remote_doris_topn_lazy_materialization.groovy new file mode 100644 index 00000000000000..cb728a170d4253 --- /dev/null +++ b/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"; + 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: the remote table is scanned via RemoteDorisScanNode (FileScan) + 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("topn result mismatch on ${catalogName}: ${remoteQuery}", + localRes, remoteRes) + } + } + + compareTopn("arrow_flight_catalog", "`${arrow_catalog}`.`${db_name}`.`${table_name}`") + compareTopn("virtual_cluster_catalog", "`${olap_catalog}`.`${db_name}`.`${table_name}`") + + sql """ DROP DATABASE IF EXISTS ${db_name} """ + sql """ DROP CATALOG IF EXISTS `${arrow_catalog}` """ + sql """ DROP CATALOG IF EXISTS `${olap_catalog}` """ +}