jayzhan211 commented on code in PR #27256: URL: https://github.com/apache/datafusion/pull/27256#discussion_r4513875527
########## datafusion/sqllogictest/test_files/subquery.slt: ########## @@ -4741,5 +4741,58 @@ NULL NULL NULL statement ok DROP TABLE p26572; +# Regression for https://github.com/apache/datafusion/issues/27216: converting a +# null-aware NOT IN join to a symmetric hash join loses its NULL semantics. +query I +COPY (VALUES (1, 'x'), (2, 'x'), (CAST(NULL AS INT), 'x')) +TO 'test_files/scratch/subquery/not_in_left.csv' +STORED AS CSV OPTIONS ('format.has_header' 'true'); +---- +3 + +query I +COPY (VALUES (2, 'y'), (CAST(NULL AS INT), 'y')) +TO 'test_files/scratch/subquery/not_in_right.csv' +STORED AS CSV OPTIONS ('format.has_header' 'true'); +---- +2 + +statement ok +CREATE UNBOUNDED EXTERNAL TABLE unbounded_not_in_left(a INT, b VARCHAR) +STORED AS CSV LOCATION 'test_files/scratch/subquery/not_in_left.csv' +OPTIONS ('format.has_header' 'true'); + +statement ok +CREATE UNBOUNDED EXTERNAL TABLE unbounded_not_in_right(a INT, b VARCHAR) +STORED AS CSV LOCATION 'test_files/scratch/subquery/not_in_right.csv' +OPTIONS ('format.has_header' 'true'); + +# These null-aware joins cannot be executed as streaming symmetric hash joins. +# Reject them during planning instead of returning rows with incorrect NULL handling. +query error Cannot execute a null-aware join with two unbounded inputs: SymmetricHashJoinExec does not support null-aware semantics +SELECT a FROM unbounded_not_in_left +WHERE a NOT IN (SELECT a FROM unbounded_not_in_right); + +# NOT IN inside OR requires a null-aware LeftMark join. +query error Cannot execute a null-aware join with two unbounded inputs: SymmetricHashJoinExec does not support null-aware semantics +SELECT a FROM unbounded_not_in_left +WHERE a = 1 OR a NOT IN (SELECT a FROM unbounded_not_in_right); + +# Nullable outer values still require null-aware semantics when the subquery excludes NULLs. +query error Cannot execute a null-aware join with two unbounded inputs: SymmetricHashJoinExec does not support null-aware semantics +SELECT a FROM unbounded_not_in_left +WHERE a NOT IN (SELECT a FROM unbounded_not_in_right WHERE a IS NOT NULL); + +# IN in a projection also requires a null-aware LeftMark join to preserve UNKNOWN. +query error Cannot execute a null-aware join with two unbounded inputs: SymmetricHashJoinExec does not support null-aware semantics +SELECT a, a IN (SELECT a FROM unbounded_not_in_right) AS is_present +FROM unbounded_not_in_left; + +statement ok +DROP TABLE unbounded_not_in_left; + +statement ok +DROP TABLE unbounded_not_in_right; Review Comment: Optional: a test with `NOT NULL` keys would show that a `NOT IN` that isn't null-aware still runs over two unbounded inputs. Without it, a check that rejected every anti join over unbounded inputs would pass all the current tests. This passes on the current head: <details> <summary>slt block to add after the two DROP TABLE statements</summary> ``` # With non-nullable keys the join is not null-aware, so it still runs as a # symmetric hash join. statement ok CREATE UNBOUNDED EXTERNAL TABLE unbounded_not_in_left_nn(a INT NOT NULL, b VARCHAR) STORED AS CSV LOCATION 'test_files/scratch/subquery/not_in_left_nn.csv' OPTIONS ('format.has_header' 'true'); statement ok CREATE UNBOUNDED EXTERNAL TABLE unbounded_not_in_right_nn(a INT NOT NULL, b VARCHAR) STORED AS CSV LOCATION 'test_files/scratch/subquery/not_in_right_nn.csv' OPTIONS ('format.has_header' 'true'); query I COPY (VALUES (1, 'x'), (2, 'x')) TO 'test_files/scratch/subquery/not_in_left_nn.csv' STORED AS CSV OPTIONS ('format.has_header' 'true'); ---- 2 query I COPY (VALUES (2, 'y')) TO 'test_files/scratch/subquery/not_in_right_nn.csv' STORED AS CSV OPTIONS ('format.has_header' 'true'); ---- 1 query I SELECT a FROM unbounded_not_in_left_nn WHERE a NOT IN (SELECT a FROM unbounded_not_in_right_nn); ---- 1 statement ok DROP TABLE unbounded_not_in_left_nn; statement ok DROP TABLE unbounded_not_in_right_nn; ``` </details> ########## datafusion/physical-optimizer/src/join_selection.rs: ########## @@ -383,6 +388,12 @@ fn hash_join_convert_symmetric_subrule( hash_join.right.pipeline_behavior(), EmissionType::Incremental | EmissionType::Both ); + if left_unbounded && right_unbounded && hash_join.null_aware { + return plan_err!( + "Cannot execute a null-aware join with two unbounded inputs: \ + SymmetricHashJoinExec does not support null-aware semantics." + ); + } // Process only if both left and right sides are unbounded and incrementally emit. if left_unbounded && right_unbounded & left_incremental & right_incremental { Review Comment: @Phoenix500526 thanks for the fix! Optional cleanup: this check also fires when an unbounded input only emits at the end, e.g. `a NOT IN (SELECT max(a) FROM r GROUP BY b)` or `(SELECT a FROM r ORDER BY a LIMIT 5)`. No symmetric hash join would be chosen there, and on `main` those queries get SanityCheckPlan's "Cannot execute pipeline breaking queries" error, which names the blocking operator. Moving the check into the conversion branch keeps that message; your four tests still pass with it: ```suggestion // Process only if both left and right sides are unbounded and incrementally emit. if left_unbounded && right_unbounded & left_incremental & right_incremental { if hash_join.null_aware { return plan_err!( "Cannot execute a null-aware join with two unbounded inputs: \ SymmetricHashJoinExec does not support null-aware semantics." ); } ``` -- 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]
