github-actions[bot] commented on code in PR #67177: URL: https://github.com/apache/doris/pull/67177#discussion_r3923989684
########## 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: [P2] Make this regression observe the private build side This LEFT JOIN returns 45 even under the native-planner bug the case is intended to catch. With correct broadcast, every `t1.pk` from 0 through 9 matches; if PASS_TO_ONE starves the private hash tables on other tasks, those same left rows are merely null-extended. `SUM(DISTINCT t1.pk)` therefore sees the identical keys in both executions. Please make the oracle build-sensitive, for example by using an INNER JOIN or aggregating a non-null build-side value/count, so a starved private build changes the golden result. ########## 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: [P2] Do not dereference the deferred entry after this call `deferred` is a reference to `_deferred_exchangers.front()`, but `_create_deferred_local_exchangers()` unconditionally clears that vector before returning. The assertions below therefore access a destroyed element (they may appear to pass only because the vector capacity still contains stale bytes). Copy `deferred.shared_state` into a local `shared_ptr` before this call and assert through that live owner instead. ########## 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: [P1] Apply the v13 fallback before set-operation property derivation This effective-mode decision happens after optimization and translation. With a v13 BE version and both local-shuffle session flags enabled, `RequestPropertyDeriver.setOperationBucketShuffleAllowed()` can still choose a bucket-shuffle `INTERSECT`/`EXCEPT`, and `PhysicalPlanTranslator` marks `SetOperationNode` as `BUCKET_SHUFFLE`. This assignment then disables `AddLocalExchange`, so the required `BUCKET_HASH_SHUFFLE` boundaries are absent and the legacy BE planner executes a topology that the comments in those two classes explicitly say can compute wrong results. In reduced form the actual plan is `SetOp(BUCKET_SHUFFLE) -> bucketed scan + bucket-shuffle Exchange`, while the required FE-planned form also aligns every child through bucket-hash local exchanges. Please freeze the effective v13/v14 planner mode before property derivation and use it in both upstream gates as well as here. -- 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]
