Add outer join tests

datafusion/physical-plan/src/joins/sort_merge_join.rs

@@ -1172,8 +1172,11 @@ impl SMJStream {
                         .collect::<Vec<_>>()
                 };
 
-            let filter_columns =
-                get_filter_column(&self.filter, &streamed_columns, &buffered_columns);
+            let filter_columns = if matches!(self.join_type, JoinType::Right) {
+                get_filter_column(&self.filter, &buffered_columns, &streamed_columns)
+            } else {
+                get_filter_column(&self.filter, &streamed_columns, &buffered_columns)
+            };
 
             let columns = if matches!(self.join_type, JoinType::Right) {
                 buffered_columns.extend(streamed_columns);

datafusion/sqllogictest/test_files/sort_merge_join.slt

@@ -23,13 +23,12 @@ statement ok
 set datafusion.optimizer.prefer_hash_join = false;
 
 statement ok
-CREATE TABLE t1(a text, b int) AS VALUES ('Alice', 50), ('Alice', 100);
+CREATE TABLE t1(a text, b int) AS VALUES ('Alice', 50), ('Alice', 100), ('Bob', 1);
 
 statement ok
 CREATE TABLE t2(a text, b int) AS VALUES ('Alice', 2), ('Alice', 1);
 
-# equijoin and join filter (sort merge join)
-
+# inner join query plan with join filter
 query TT
 EXPLAIN SELECT t1.a, t1.b, t2.a, t2.b FROM t1 JOIN t2 ON t1.a = t2.a AND t2.b * 50 <= t1.b
 ----
@@ -48,6 +47,7 @@ SortMergeJoin: join_type=Inner, on=[(a@0, a@0)], filter=CAST(b@1 AS Int64) * 50
 ------RepartitionExec: partitioning=Hash([a@0], 4), input_partitions=1
 --------MemoryExec: partitions=1, partition_sizes=[1]
 
+# inner join with join filter
 query TITI rowsort
 SELECT t1.a, t1.b, t2.a, t2.b FROM t1 JOIN t2 ON t1.a = t2.a AND t2.b * 50 <= t1.b
 ----
@@ -67,6 +67,79 @@ query TITI rowsort
 SELECT t1.a, t1.b, t2.a, t2.b FROM t1 JOIN t2 ON t1.a = t2.a AND t2.b > t1.b
 ----
 
+# left join without join filter
+query TITI rowsort
+SELECT * FROM t1 LEFT JOIN t2 ON t1.a = t2.a
+----
+Alice 100 Alice 1
+Alice 100 Alice 2
+Alice 50 Alice 1
+Alice 50 Alice 2
+Bob 1 NULL NULL
+
+# left join with join filter
+query TITI rowsort
+SELECT * FROM t1 LEFT JOIN t2 ON t1.a = t2.a AND t2.b * 50 <= t1.b
+----
+Alice 100 Alice 1
+Alice 100 Alice 2
+Alice 50 Alice 1
+
+query TITI rowsort
+SELECT * FROM t1 LEFT JOIN t2 ON t1.a = t2.a AND t2.b < t1.b
+----
+Alice 100 Alice 1
+Alice 100 Alice 2
+Alice 50 Alice 1
+Alice 50 Alice 2
+
+# right join without join filter
+query TITI rowsort
+SELECT * FROM t1 RIGHT JOIN t2 ON t1.a = t2.a
+----
+Alice 100 Alice 1
+Alice 100 Alice 2
+Alice 50 Alice 1
+Alice 50 Alice 2
+
+# right join with join filter
+query TITI rowsort
+SELECT * FROM t1 RIGHT JOIN t2 ON t1.a = t2.a AND t2.b * 50 <= t1.b
+----
+Alice 100 Alice 1
+Alice 100 Alice 2
+Alice 50 Alice 1
+
+query TITI rowsort
+SELECT * FROM t1 RIGHT JOIN t2 ON t1.a = t2.a AND t1.b > t2.b
+----
+Alice 100 Alice 1
+Alice 100 Alice 2
+Alice 50 Alice 1
+Alice 50 Alice 2
+
+# full join without join filter
+query TITI rowsort
+SELECT * FROM t1 FULL JOIN t2 ON t1.a = t2.a
+----
+Alice 100 Alice 1
+Alice 100 Alice 2
+Alice 50 Alice 1
+Alice 50 Alice 2
+Bob 1 NULL NULL
+
+# full join with join filter
+query TITI rowsort
+SELECT * FROM t1 FULL JOIN t2 ON t1.a = t2.a AND t2.b * 50 > t1.b
+----
+Alice 50 Alice 2
+
+query TITI rowsort
+SELECT * FROM t1 FULL JOIN t2 ON t1.a = t2.a AND t1.b > t2.b + 50
+----
+Alice 100 Alice 1
+Alice 100 Alice 2
+
 statement ok
 set datafusion.optimizer.prefer_hash_join = true;