fix: FULL OUTER JOIN and LIMIT produces wrong results (apache#14338)

* fix: FULL OUTER JOIN and LIMIT produces wrong results

* Fix minor slt testing

* fix test

Author: zhuqi-lucas

datafusion/optimizer/src/push_down_limit.rs

@@ -255,7 +255,6 @@ fn push_down_join(mut join: Join, limit: usize) -> Transformed<Join> {
         match join.join_type {
             Left => (Some(limit), None),
             Right => (None, Some(limit)),
-            Full => (Some(limit), Some(limit)),
             _ => (None, None),
         }
     };

datafusion/sqllogictest/test_files/joins.slt

@@ -4234,10 +4234,8 @@ EXPLAIN SELECT * FROM t0 FULL JOIN t1 ON t0.c1 = t1.c1 LIMIT 2;
 logical_plan
 01)Limit: skip=0, fetch=2
 02)--Full Join: t0.c1 = t1.c1
-03)----Limit: skip=0, fetch=2
-04)------TableScan: t0 projection=[c1, c2], fetch=2
-05)----Limit: skip=0, fetch=2
-06)------TableScan: t1 projection=[c1, c2, c3], fetch=2
+03)----TableScan: t0 projection=[c1, c2]
+04)----TableScan: t1 projection=[c1, c2, c3]
 physical_plan
 01)CoalesceBatchesExec: target_batch_size=3, fetch=2
 02)--HashJoinExec: mode=CollectLeft, join_type=Full, on=[(c1@0, c1@0)]
@@ -4251,10 +4249,8 @@ EXPLAIN SELECT * FROM t0 FULL JOIN t1 ON t0.c2 >= t1.c2 LIMIT 2;
 logical_plan
 01)Limit: skip=0, fetch=2
 02)--Full Join:  Filter: t0.c2 >= t1.c2
-03)----Limit: skip=0, fetch=2
-04)------TableScan: t0 projection=[c1, c2], fetch=2
-05)----Limit: skip=0, fetch=2
-06)------TableScan: t1 projection=[c1, c2, c3], fetch=2
+03)----TableScan: t0 projection=[c1, c2]
+04)----TableScan: t1 projection=[c1, c2, c3]
 physical_plan
 01)GlobalLimitExec: skip=0, fetch=2
 02)--NestedLoopJoinExec: join_type=Full, filter=c2@0 >= c2@1
@@ -4268,12 +4264,268 @@ EXPLAIN SELECT * FROM t0 FULL JOIN t1 ON t0.c1 = t1.c1 AND t0.c2 >= t1.c2 LIMIT
 logical_plan
 01)Limit: skip=0, fetch=2
 02)--Full Join: t0.c1 = t1.c1 Filter: t0.c2 >= t1.c2
-03)----Limit: skip=0, fetch=2
-04)------TableScan: t0 projection=[c1, c2], fetch=2
-05)----Limit: skip=0, fetch=2
-06)------TableScan: t1 projection=[c1, c2, c3], fetch=2
+03)----TableScan: t0 projection=[c1, c2]
+04)----TableScan: t1 projection=[c1, c2, c3]
 physical_plan
 01)CoalesceBatchesExec: target_batch_size=3, fetch=2
 02)--HashJoinExec: mode=CollectLeft, join_type=Full, on=[(c1@0, c1@0)], filter=c2@0 >= c2@1
 03)----MemoryExec: partitions=1, partition_sizes=[1]
 04)----MemoryExec: partitions=1, partition_sizes=[1]
+
+## Add more test cases for join limit pushdown
+statement ok
+drop table t1
+
+## Test limit pushdown through OUTER JOIN including left/right and full outer join cases
+statement ok
+set datafusion.execution.target_partitions = 1;
+
+### Limit pushdown through join
+
+# Note we use csv as MemoryExec does not support limit push down (so doesn't manifest
+# bugs if limits are improperly pushed down)
+query I
+COPY (values (1), (2), (3), (4), (5))  TO 'test_files/scratch/limit/t1.csv'
+STORED AS CSV
+----
+5
+
+# store t2 in different order so the top N rows are not the same as the top N rows of t1
+query I
+COPY (values (5), (4), (3), (2), (1))  TO 'test_files/scratch/limit/t2.csv'
+STORED AS CSV
+----
+5
+
+statement ok
+create external table t1(a int) stored as CSV location 'test_files/scratch/limit/t1.csv';
+
+statement ok
+create external table t2(b int) stored as CSV location 'test_files/scratch/limit/t2.csv';
+
+######
+## LEFT JOIN w/ LIMIT
+######
+query II
+select * from t1 LEFT JOIN t2 ON t1.a = t2.b LIMIT 2;
+----
+2 2
+1 1
+
+# the output of this query should be two rows from the previous query
+# there should be no nulls
+query II
+select * from t1 LEFT JOIN t2 ON t1.a = t2.b LIMIT 2;
+----
+2 2
+1 1
+
+# can only push down to t1 (preserved side)
+query TT
+explain select * from t1 LEFT JOIN t2 ON t1.a = t2.b LIMIT 2;
+----
+logical_plan
+01)Limit: skip=0, fetch=2
+02)--Left Join: t1.a = t2.b
+03)----Limit: skip=0, fetch=2
+04)------TableScan: t1 projection=[a], fetch=2
+05)----TableScan: t2 projection=[b]
+physical_plan
+01)CoalesceBatchesExec: target_batch_size=3, fetch=2
+02)--HashJoinExec: mode=CollectLeft, join_type=Left, on=[(a@0, b@0)]
+03)----CsvExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/limit/t1.csv]]}, projection=[a], limit=2, has_header=true
+04)----CsvExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/limit/t2.csv]]}, projection=[b], has_header=true
+
+######
+## RIGHT JOIN w/ LIMIT
+######
+
+query II
+select * from t1 RIGHT JOIN t2 ON t1.a = t2.b LIMIT 2;
+----
+5 5
+4 4
+
+# the output of this query should be two rows from the previous query
+# there should be no nulls
+query II
+select * from t1 RIGHT JOIN t2 ON t1.a = t2.b LIMIT 2;
+----
+5 5
+4 4
+
+# can only push down to t2 (preserved side)
+query TT
+explain select * from t1 RIGHT JOIN t2 ON t1.a = t2.b LIMIT 2;
+----
+logical_plan
+01)Limit: skip=0, fetch=2
+02)--Right Join: t1.a = t2.b
+03)----TableScan: t1 projection=[a]
+04)----Limit: skip=0, fetch=2
+05)------TableScan: t2 projection=[b], fetch=2
+physical_plan
+01)CoalesceBatchesExec: target_batch_size=3, fetch=2
+02)--HashJoinExec: mode=CollectLeft, join_type=Right, on=[(a@0, b@0)]
+03)----CsvExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/limit/t1.csv]]}, projection=[a], has_header=true
+04)----CsvExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/limit/t2.csv]]}, projection=[b], limit=2, has_header=true
+
+######
+## FULL JOIN w/ LIMIT
+######
+query II rowsort
+select * from t1 FULL JOIN t2 ON t1.a = t2.b;
+----
+1 1
+2 2
+3 3
+4 4
+5 5
+
+# the output of this query should be two rows from the previous query
+# there should be no nulls
+# Reproducer for https://github.com/apache/datafusion/issues/14335
+query II
+select * from t1 FULL JOIN t2 ON t1.a = t2.b LIMIT 2;
+----
+5 5
+4 4
+
+
+# can't push limit for full outer join
+query TT
+explain select * from t1 FULL JOIN t2 ON t1.a = t2.b LIMIT 2;
+----
+logical_plan
+01)Limit: skip=0, fetch=2
+02)--Full Join: t1.a = t2.b
+03)----TableScan: t1 projection=[a]
+04)----TableScan: t2 projection=[b]
+physical_plan
+01)CoalesceBatchesExec: target_batch_size=3, fetch=2
+02)--HashJoinExec: mode=CollectLeft, join_type=Full, on=[(a@0, b@0)]
+03)----CsvExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/limit/t1.csv]]}, projection=[a], has_header=true
+04)----CsvExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/limit/t2.csv]]}, projection=[b], has_header=true
+
+statement ok
+drop table t1;
+
+statement ok
+drop table t2;
+
+# Test Utf8View as Join Key
+# Issue: https://github.com/apache/datafusion/issues/12468
+statement ok
+CREATE TABLE table1(v1 STRING) AS VALUES ('foo'), (NULL);
+
+statement ok
+CREATE TABLE table1_stringview AS SELECT arrow_cast(v1, 'Utf8View') AS v1 FROM table1;
+
+query T
+select * from table1 as t1 natural join table1_stringview as t2;
+----
+foo
+
+query TT
+EXPLAIN SELECT count(*)
+FROM my_catalog.my_schema.table_with_many_types AS l
+JOIN my_catalog.my_schema.table_with_many_types AS r ON l.binary_col = r.binary_col
+----
+logical_plan
+01)Aggregate: groupBy=[[]], aggr=[[count(Int64(1)) AS count(*)]]
+02)--Projection: 
+03)----Inner Join: l.binary_col = r.binary_col
+04)------SubqueryAlias: l
+05)--------TableScan: my_catalog.my_schema.table_with_many_types projection=[binary_col]
+06)------SubqueryAlias: r
+07)--------TableScan: my_catalog.my_schema.table_with_many_types projection=[binary_col]
+physical_plan
+01)AggregateExec: mode=Single, gby=[], aggr=[count(*)]
+02)--ProjectionExec: expr=[]
+03)----CoalesceBatchesExec: target_batch_size=3
+04)------HashJoinExec: mode=CollectLeft, join_type=Inner, on=[(binary_col@0, binary_col@0)]
+05)--------MemoryExec: partitions=1, partition_sizes=[1]
+06)--------MemoryExec: partitions=1, partition_sizes=[1]
+
+# Test hash join sort push down
+# Issue: https://github.com/apache/datafusion/issues/13559
+statement ok
+CREATE TABLE test(a INT, b INT, c INT)
+
+statement ok
+insert into test values (1,2,3), (4,5,6), (null, 7, 8), (8, null, 9), (9, 10, null)
+
+statement ok
+set datafusion.execution.target_partitions = 2;
+
+query TT
+explain select * from test where a in (select a from test where b > 3) order by c desc nulls first;
+----
+logical_plan
+01)Sort: test.c DESC NULLS FIRST
+02)--LeftSemi Join: test.a = __correlated_sq_1.a
+03)----TableScan: test projection=[a, b, c]
+04)----SubqueryAlias: __correlated_sq_1
+05)------Projection: test.a
+06)--------Filter: test.b > Int32(3)
+07)----------TableScan: test projection=[a, b]
+physical_plan
+01)SortPreservingMergeExec: [c@2 DESC]
+02)--CoalesceBatchesExec: target_batch_size=3
+03)----HashJoinExec: mode=Partitioned, join_type=RightSemi, on=[(a@0, a@0)]
+04)------CoalesceBatchesExec: target_batch_size=3
+05)--------RepartitionExec: partitioning=Hash([a@0], 2), input_partitions=2
+06)----------CoalesceBatchesExec: target_batch_size=3
+07)------------FilterExec: b@1 > 3, projection=[a@0]
+08)--------------RepartitionExec: partitioning=RoundRobinBatch(2), input_partitions=1
+09)----------------MemoryExec: partitions=1, partition_sizes=[1]
+10)------SortExec: expr=[c@2 DESC], preserve_partitioning=[true]
+11)--------CoalesceBatchesExec: target_batch_size=3
+12)----------RepartitionExec: partitioning=Hash([a@0], 2), input_partitions=2
+13)------------RepartitionExec: partitioning=RoundRobinBatch(2), input_partitions=1
+14)--------------MemoryExec: partitions=1, partition_sizes=[1]
+
+query TT
+explain select * from test where a in (select a from test where b > 3) order by c desc nulls last;
+----
+logical_plan
+01)Sort: test.c DESC NULLS LAST
+02)--LeftSemi Join: test.a = __correlated_sq_1.a
+03)----TableScan: test projection=[a, b, c]
+04)----SubqueryAlias: __correlated_sq_1
+05)------Projection: test.a
+06)--------Filter: test.b > Int32(3)
+07)----------TableScan: test projection=[a, b]
+physical_plan
+01)SortPreservingMergeExec: [c@2 DESC NULLS LAST]
+02)--CoalesceBatchesExec: target_batch_size=3
+03)----HashJoinExec: mode=Partitioned, join_type=RightSemi, on=[(a@0, a@0)]
+04)------CoalesceBatchesExec: target_batch_size=3
+05)--------RepartitionExec: partitioning=Hash([a@0], 2), input_partitions=2
+06)----------CoalesceBatchesExec: target_batch_size=3
+07)------------FilterExec: b@1 > 3, projection=[a@0]
+08)--------------RepartitionExec: partitioning=RoundRobinBatch(2), input_partitions=1
+09)----------------MemoryExec: partitions=1, partition_sizes=[1]
+10)------SortExec: expr=[c@2 DESC NULLS LAST], preserve_partitioning=[true]
+11)--------CoalesceBatchesExec: target_batch_size=3
+12)----------RepartitionExec: partitioning=Hash([a@0], 2), input_partitions=2
+13)------------RepartitionExec: partitioning=RoundRobinBatch(2), input_partitions=1
+14)--------------MemoryExec: partitions=1, partition_sizes=[1]
+
+query III
+select * from test where a in (select a from test where b > 3) order by c desc nulls first;
+----
+9 10 NULL
+4 5 6
+
+query III
+select * from test where a in (select a from test where b > 3) order by c desc nulls last;
+----
+4 5 6
+9 10 NULL
+
+statement ok
+DROP TABLE test
+
+statement ok
+set datafusion.execution.target_partitions = 1;