924060929 commented on code in PR #67177:
URL: https://github.com/apache/doris/pull/67177#discussion_r3946171874
##########
fe/fe-core/src/main/java/org/apache/doris/nereids/NereidsPlanner.java:
##########
@@ -813,8 +816,13 @@ protected void doDistribute(boolean
canUseNereidsDistributePlanner, ExplainLevel
}
boolean useLoadBackendSelection =
physicalPlan.anyMatch(PhysicalOlapTableSink.class::isInstance);
+ SessionVariable sessionVariable =
statementContext.getConnectContext().getSessionVariable();
+ localShufflePlanned = sessionVariable.isEnableLocalShufflePlanner()
Review Comment:
Fixed in 370534071e2. The BE execution version and effective FE
local-shuffle-planner mode are now frozen immediately after SET_VAR
preprocessing and before Cascades optimization. RequestPropertyDeriver and
PhysicalPlanTranslator consume that query-level decision, so version 13 cannot
derive the incompatible bucket-shuffle set-operation shape. The unit test
covers both version-13 fallback and the valid enable_local_shuffle=false
bucket-shuffle path; NereidsCoordinatorTest passes 5/5.
##########
regression-test/suites/nereids_p0/local_shuffle/test_serial_aggregation_over_parallel_join.groovy:
##########
@@ -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.
+
+suite("test_serial_aggregation_over_parallel_join") {
+ ["serial_agg_join_probe", "serial_agg_join_left",
"serial_agg_join_right"].each { table ->
+ sql "DROP TABLE IF EXISTS ${table}"
+ }
+
+ sql """CREATE TABLE serial_agg_join_probe (
+ col_bigint BIGINT, col_v10 VARCHAR(10), col_v64 VARCHAR(64),
pk INT
+ ) ENGINE=OLAP DISTRIBUTED BY HASH(pk) BUCKETS 10
+ PROPERTIES ("replication_num"="1")"""
+ sql """CREATE TABLE serial_agg_join_left (
+ pk INT, col_bigint BIGINT, col_v10 VARCHAR(10), col_v64
VARCHAR(64)
+ ) ENGINE=OLAP DUPLICATE KEY(pk, col_bigint, col_v10)
+ DISTRIBUTED BY HASH(pk) BUCKETS 10 PROPERTIES
("replication_num"="1")"""
+ sql """CREATE TABLE serial_agg_join_right (
+ pk INT, col_v10 VARCHAR(10), col_bigint BIGINT, col_v64
VARCHAR(64)
+ ) ENGINE=OLAP DUPLICATE KEY(pk, col_v10)
+ DISTRIBUTED BY HASH(pk) BUCKETS 10 PROPERTIES
("replication_num"="1")"""
+
+ sql """INSERT INTO serial_agg_join_probe VALUES
+
(-94,'had','y',0),(672609,'k','h',1),(-3766684,'a','p',2),(5070261,'on','x',3),
+
(NULL,'u','at',4),(-86,'v','c',5),(21910,'how','m',6),(-63,'that''s','go',7),
+ (-8276281,'s','a',8),(-101,'w','y',9)"""
+ sql """INSERT INTO serial_agg_join_left VALUES
+
(0,NULL,'g','i'),(1,-6138328,'z','do'),(2,-23217,'g','about'),(3,104,'you''re','z'),
+
(4,NULL,'oh','i'),(5,-54,'want','to'),(6,NULL,'x','c'),(7,NULL,'you''re','come'),
+ (8,3447,'really','from'),(9,-5459,'i','will')"""
+ sql """INSERT INTO serial_agg_join_right VALUES
+
(0,'right',NULL,'g'),(1,'on',-486256,'on'),(2,'I''ll',-1,'at'),(3,'h',29263,'don''t'),
+
(4,'a',5453,'s'),(5,'j',-119,'can''t'),(6,'one',89,'n'),(7,'s',-7227,'u'),
+ (8,'time',94,'b'),(9,'yes',1816630,'yes')"""
+
+ def variables =
"enable_local_shuffle_planner=true,enable_local_shuffle=true," +
+
"enable_bucket_shuffle_join=true,ignore_storage_data_distribution=true," +
+ "bucket_shuffle_downgrade_ratio=0.8,use_serial_exchange=false," +
+ "parallel_pipeline_task_num=3,enable_sql_cache=false," +
+ "enable_share_hash_table_for_broadcast_join=false"
+
+ order_qt_count_distinct_left_join """SELECT /*+SET_VAR(${variables})*/
COUNT(DISTINCT t1.pk)
+ FROM serial_agg_join_left t1 LEFT JOIN serial_agg_join_probe t2 ON
t2.pk=t1.pk
+ WHERE (t1.col_v64>'FVjnKolDTt' AND t1.col_v64<='z') OR t1.col_v64
IS NULL
+ OR (t1.col_v10>'me' AND t1.col_v10<='zzzz' AND t1.col_bigint
BETWEEN 3 AND 7)"""
+
+ order_qt_sum_and_count_distinct_left_join """SELECT
/*+SET_VAR(${variables})*/
+ SUM(DISTINCT t1.pk), COUNT(DISTINCT t1.pk)
+ FROM serial_agg_join_right t1 LEFT JOIN serial_agg_join_probe t2
ON t2.pk=t1.pk
+ WHERE t1.pk IN (2,9) OR t1.col_bigint IN (1,8)
+ OR (t1.col_v64>='MijtyYyxeA' AND t1.col_v64<'z'
+ AND t1.col_v64>='on' AND t1.col_v64<'zzzz')"""
+
+ order_qt_sum_distinct_broad_predicate """SELECT /*+SET_VAR(${variables})*/
SUM(DISTINCT t1.pk)
+ FROM serial_agg_join_right t1 LEFT JOIN serial_agg_join_probe t2
ON t2.pk=t1.pk
+ WHERE (t1.col_v64>='QXQpaZhWfj' AND t1.col_v64<'z')
+ OR (t1.col_v64>='fvPsFBZelL' AND t1.col_v64<='well')
+ OR (t1.pk BETWEEN 0 AND 15 AND t1.col_v10 LIKE 'a%')
+ OR (t1.pk>=3 AND t1.pk<4) OR t1.pk BETWEEN 0 AND 100 OR
(t1.pk>7 AND t1.pk<=9)"""
+
+ order_qt_sum_distinct_reversed_join """SELECT /*+SET_VAR(${variables})*/
SUM(DISTINCT t1.pk)
+ FROM serial_agg_join_probe t1 LEFT JOIN serial_agg_join_right t2
ON t1.pk=t2.pk
+ WHERE (t1.pk IS NOT NULL AND t1.pk IN (3,8,2,2)
+ AND t1.col_v64 IN
('didn''t','when','a','come','AgpEFIOTAN'))
+ OR (t1.col_v64>'HoatMBMEwP' AND t1.col_v64<='zzzz') OR t1.pk
BETWEEN 6 AND 11
+ OR (t1.pk IS NULL AND t1.pk IN (5)) OR (t1.pk<=t1.col_bigint
AND t1.pk IN (8))"""
+
+ order_qt_sum_distinct_multi_outer_join """SELECT
/*+SET_VAR(${variables})*/ SUM(DISTINCT t1.pk)
+ FROM serial_agg_join_right t1 RIGHT OUTER JOIN
serial_agg_join_probe t2 ON t2.pk=t2.pk
+ LEFT JOIN serial_agg_join_left t3 ON t3.pk=t1.pk
+ WHERE (t1.col_v10>'jHKKlhlHDn' AND t1.col_v10<'z'
+ AND t1.col_v10 NOT IN ('him','you''re'))
+ OR (t1.col_v64>='j' AND t1.col_v64<='y')
+ OR (t1.col_v10 NOT BETWEEN 'rxpMJWfBRX' AND 'z' AND
t1.col_bigint IN (1000)
+ AND t1.col_bigint IS NULL AND t1.col_bigint BETWEEN 6 AND
15)"""
+
+ def nativeVariables =
"enable_local_shuffle_planner=false,enable_local_shuffle=true," +
+ "parallel_pipeline_task_num=3,enable_sql_cache=false," +
+ "enable_share_hash_table_for_broadcast_join=false"
+
+ order_qt_native_private_broadcast_build """SELECT
/*+SET_VAR(${nativeVariables})*/ SUM(DISTINCT t1.pk)
Review Comment:
Fixed in 370534071e2. The regression oracle now uses INNER JOIN plus
COUNT(t2.pk). If private broadcast incorrectly degenerates to PASS_TO_ONE,
probe tasks without the build side lose matches and the result becomes smaller
than 10, so the test is sensitive to the broken native-BE path. The focused
suite and all 8 nereids_p0/local_shuffle suites pass on the four-BE cluster.
##########
be/test/exec/pipeline/local_exchanger_test.cpp:
##########
@@ -532,9 +535,55 @@ TEST_F(LocalExchangerTest, PassthroughExchanger) {
}
}
-TEST_F(LocalExchangerTest, PassToOneExchanger) {
+TEST_F(LocalExchangerTest, FePlannedPassToOneUsesOneDownstreamSource) {
+ constexpr int num_instances = 4;
+ TPipelineFragmentParams params;
+ auto context = std::make_shared<PipelineFragmentContext>(
+ _query_id, params, _query_ctx, ExecEnv::GetInstance(),
[](RuntimeState*, Status*) {});
+ context->_num_instances = num_instances;
+ context->_total_instances = num_instances;
+ context->_runtime_state = RuntimeState::create_unique(_query_id,
_fragment_id, _query_options,
+
_query_ctx->query_globals,
+
ExecEnv::GetInstance(), _query_ctx.get());
+
+ auto downstream_pipe = context->add_pipeline();
+ downstream_pipe->set_num_tasks(1);
+ auto upstream_pipe = downstream_pipe;
+
+ TLocalExchangeNode local_exchange_node;
+ local_exchange_node.__set_partition_type(TLocalPartitionType::PASS_TO_ONE);
+ TPlanNode tnode;
+ tnode.__set_node_type(TPlanNodeType::LOCAL_EXCHANGE_NODE);
+ tnode.__set_node_id(0);
+ tnode.__set_num_children(1);
+ tnode.__set_local_exchange_node(local_exchange_node);
+
+ ObjectPool pool;
+ DescriptorTbl descs;
+ OperatorPtr op;
+ OperatorPtr cache_op;
+ ASSERT_TRUE(context->_create_operator(&pool, tnode, descs, op,
upstream_pipe,
+ /*parent_idx=*/-1, /*child_idx=*/0,
+
/*followed_by_shuffled_operator=*/false,
+
/*require_bucket_distribution=*/false, cache_op)
+ .ok());
+ ASSERT_EQ(context->_deferred_exchangers.size(), 1);
+ EXPECT_EQ(downstream_pipe->num_tasks(), 1);
+ EXPECT_EQ(upstream_pipe->num_tasks(), num_instances);
+
+ const auto& deferred = context->_deferred_exchangers.front();
+ EXPECT_EQ(deferred.shared_state->source_deps.size(), 1);
+ EXPECT_EQ(deferred.shared_state->mem_counters.size(), 1);
+ ASSERT_TRUE(context->_create_deferred_local_exchangers().ok());
+ ASSERT_NE(deferred.shared_state->exchanger, nullptr);
Review Comment:
Fixed in 370534071e2. The test now copies the shared_ptr before
_create_deferred_local_exchangers() clears the deferred vector, so all later
assertions use a live owner rather than a dangling reference. It also
constructs a valid descriptor table and row_tuples, addressing the BE-UT setup
failure observed in CI.
--
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]
