kosiew commented on code in PR #27345: URL: https://github.com/apache/datafusion/pull/27345#discussion_r4548115857
########## datafusion/sqllogictest/test_files/join_dynamic_filter_canceled_partition.slt: ########## @@ -0,0 +1,141 @@ +# SPDX-License-Identifier: Apache-2.0 +# SPDX-FileCopyrightText: See the NOTICE file distributed with this work for additional information regarding copyright ownership +# SPDX-FileContributor: Licensed to the Apache Software Foundation (ASF) under one or more contributor license agreements + +# Reproducer for https://github.com/apache/datafusion/issues/21625 +# (fixed by https://github.com/apache/datafusion/pull/21666). +# +# A partitioned hash join's dynamic filter waits until every build partition +# reports to a shared coordinator. A parent join can drop a child partition +# before that child reports, and the coordinator must not wait for it. +# +# The plan is a partitioned parent join with a partitioned child join on its +# probe side. Both joins use the same key, so no repartition sits between them. +# `s` holds only k = 1, so 3 of the 4 parent build partitions are empty: +# +# parent P0/P2/P3: empty build, Inner join -> Completed, probe never polled +# -> child P0/P2/P3 dropped before they report +# parent P1: non-empty build -> polls child P1 +# -> child P1 waits for the reports of P0, P2 and P3 +# +# Without the fix, child P1 waits forever and the query hangs. With the fix, +# a dropped child partition reports itself as canceled and the query returns. +# The hang does not depend on timing: hash routing of k = 1 is fixed, and a +# child partition reports only when its parent partition polls it. + +# The test runner also uses 4. The explicit setting keeps the test independent +# of that default, because the partition count decides which partitions are empty. +statement ok +set datafusion.execution.target_partitions = 4; + +# Force both joins into Partitioned mode. +statement ok +set datafusion.optimizer.hash_join_single_partition_threshold = 0; + +statement ok +set datafusion.optimizer.hash_join_single_partition_threshold_rows = 0; + +# Keep `s` on the parent build side and the child join on its probe side. +statement ok +set datafusion.optimizer.join_reordering = false; + +# Two files per table give two scan partitions, so the planner hash-repartitions +# each join input from 2 to 4 partitions. +query I +COPY (SELECT 1 AS k) TO 'test_files/scratch/join_dynamic_filter_canceled_partition/s/0.parquet' STORED AS PARQUET; +---- +1 + +query I +COPY (SELECT 1 AS k) TO 'test_files/scratch/join_dynamic_filter_canceled_partition/s/1.parquet' STORED AS PARQUET; +---- +1 + +query I +COPY (SELECT column1 AS k, column1 AS v FROM (VALUES (1), (2), (3), (4))) TO 'test_files/scratch/join_dynamic_filter_canceled_partition/a/0.parquet' STORED AS PARQUET; +---- +4 + +query I +COPY (SELECT column1 AS k, column1 AS v FROM (VALUES (5), (6), (7), (8))) TO 'test_files/scratch/join_dynamic_filter_canceled_partition/a/1.parquet' STORED AS PARQUET; +---- +4 + +query I +COPY (SELECT column1 AS k FROM (VALUES (1), (2), (3), (4))) TO 'test_files/scratch/join_dynamic_filter_canceled_partition/b/0.parquet' STORED AS PARQUET; +---- +4 + +query I +COPY (SELECT column1 AS k FROM (VALUES (5), (6), (7), (8))) TO 'test_files/scratch/join_dynamic_filter_canceled_partition/b/1.parquet' STORED AS PARQUET; +---- +4 + +statement ok +CREATE EXTERNAL TABLE s(k INT) +STORED AS PARQUET +LOCATION 'test_files/scratch/join_dynamic_filter_canceled_partition/s/'; + +statement ok +CREATE EXTERNAL TABLE a(k INT, v INT) +STORED AS PARQUET +LOCATION 'test_files/scratch/join_dynamic_filter_canceled_partition/a/'; + +statement ok +CREATE EXTERNAL TABLE b(k INT) +STORED AS PARQUET +LOCATION 'test_files/scratch/join_dynamic_filter_canceled_partition/b/'; + +# The reproducer needs this exact shape: both joins Partitioned, `s` on the +# parent build side, and the child join directly on the parent probe side. +query TT +EXPLAIN SELECT s.k, ab.v Review Comment: Consider setting `datafusion.explain.physical_plan_only = true` around this `EXPLAIN` and resetting it afterward. The physical plan already captures all cancellation prerequisites, while the logical-plan snapshot adds maintenance overhead from unrelated formatting changes. ########## datafusion/sqllogictest/test_files/join_dynamic_filter_canceled_partition.slt: ########## @@ -0,0 +1,141 @@ +# SPDX-License-Identifier: Apache-2.0 +# SPDX-FileCopyrightText: See the NOTICE file distributed with this work for additional information regarding copyright ownership +# SPDX-FileContributor: Licensed to the Apache Software Foundation (ASF) under one or more contributor license agreements + +# Reproducer for https://github.com/apache/datafusion/issues/21625 +# (fixed by https://github.com/apache/datafusion/pull/21666). +# +# A partitioned hash join's dynamic filter waits until every build partition +# reports to a shared coordinator. A parent join can drop a child partition +# before that child reports, and the coordinator must not wait for it. +# +# The plan is a partitioned parent join with a partitioned child join on its +# probe side. Both joins use the same key, so no repartition sits between them. +# `s` holds only k = 1, so 3 of the 4 parent build partitions are empty: +# +# parent P0/P2/P3: empty build, Inner join -> Completed, probe never polled +# -> child P0/P2/P3 dropped before they report +# parent P1: non-empty build -> polls child P1 +# -> child P1 waits for the reports of P0, P2 and P3 +# +# Without the fix, child P1 waits forever and the query hangs. With the fix, +# a dropped child partition reports itself as canceled and the query returns. +# The hang does not depend on timing: hash routing of k = 1 is fixed, and a +# child partition reports only when its parent partition polls it. + +# The test runner also uses 4. The explicit setting keeps the test independent +# of that default, because the partition count decides which partitions are empty. +statement ok +set datafusion.execution.target_partitions = 4; + +# Force both joins into Partitioned mode. Review Comment: Consider explicitly setting `datafusion.optimizer.enable_dynamic_filter_pushdown = true` and resetting it during cleanup to document the coordinator prerequisite. This also enables join dynamic-filter pushdown, so a separate setting is unnecessary; both currently default to true, and the existing `EXPLAIN` already guards against missing filters. -- 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]
