Support JOIN pipe operator

datafusion/sql/src/query.rs

@@ -194,6 +194,9 @@ impl<S: ContextProvider> SqlToRel<'_, S> {
                 group_by_expr,
                 planner_context,
             ),
+            PipeOperator::Join(join) => {
+                self.parse_relation_join(plan, join, planner_context)
+            }
 
             x => not_impl_err!("`{x}` pipe operator is not supported yet"),
         }

datafusion/sql/src/relation/join.rs

@@ -43,7 +43,7 @@ impl<S: ContextProvider> SqlToRel<'_, S> {
         Ok(left)
     }
 
-    fn parse_relation_join(
+    pub(crate) fn parse_relation_join(
         &self,
         left: LogicalPlan,
         join: Join,

datafusion/sqllogictest/test_files/pipe_operator.slt

@@ -177,3 +177,21 @@ query TII rowsort
 |> WHERE num_items > 1;
 ----
 apples 2 9
+
+# JOIN pipe
+query TII
+(
+  SELECT 'apples' AS item, 2 AS sales
+  UNION ALL
+  SELECT 'bananas' AS item, 5 AS sales
+)
+|> AS produce_sales
+|> LEFT JOIN
+     (
+       SELECT "apples" AS item, 123 AS id
+     ) AS produce_data
+   ON produce_sales.item = produce_data.item
+|> SELECT produce_sales.item, sales, id;
+----
+apples 2 123
+bananas 5 NULL

docs/source/user-guide/sql/select.md

@@ -350,6 +350,7 @@ DataFusion currently supports the following pipe operators:
 - [INTERSECT](#pipe_intersect)
 - [EXCEPT](#pipe_except)
 - [AGGREGATE](#pipe_aggregate)
+- [JOIN](#pipe_join)
 
 (pipe_where)=
 
@@ -514,3 +515,28 @@ select * from range(0,3)
 | 3     |
 +-------+
 ```
+
+(pipe_join)=
+
+### JOIN
+
+```sql
+> (
+  SELECT 'apples' AS item, 2 AS sales
+  UNION ALL
+  SELECT 'bananas' AS item, 5 AS sales
+)
+|> AS produce_sales
+|> LEFT JOIN
+     (
+       SELECT 'apples' AS item, 123 AS id
+     ) AS produce_data
+   ON produce_sales.item = produce_data.item
+|> SELECT produce_sales.item, sales, id;
++--------+-------+------+
+| item   | sales | id   |
++--------+-------+------+
+| apples | 2     | 123  |
+| bananas| 5     | NULL |
++--------+-------+------+
+```