fix: unparsing left/ right semi/mark join (apache#15212)

* fix: unparse semi/mark join

* recursive

* fix use

* update

* stackoverflow

* update stack size

* update test

* fix test

* format

* refine

* refine ci based on goldmedal's suggestion

Author: chenkovsky

.github/workflows/extended.yml

@@ -81,7 +81,7 @@ jobs:
       - name: Run tests (excluding doctests)
         env:
           RUST_BACKTRACE: 1
-        run: cargo test --profile ci --exclude datafusion-examples --exclude datafusion-benchmarks --workspace --lib --tests --bins --features avro,json,backtrace,extended_tests
+        run: cargo test --profile ci --exclude datafusion-examples --exclude datafusion-benchmarks --workspace --lib --tests --bins --features avro,json,backtrace,extended_tests,recursive_protection
       - name: Verify Working Directory Clean
         run: git diff --exit-code
       - name: Cleanup

datafusion/sql/src/unparser/ast.rs

@@ -16,9 +16,10 @@
 // under the License.
 
 use core::fmt;
+use std::ops::ControlFlow;
 
-use sqlparser::ast;
 use sqlparser::ast::helpers::attached_token::AttachedToken;
+use sqlparser::ast::{self, visit_expressions_mut};
 
 #[derive(Clone)]
 pub struct QueryBuilder {
@@ -176,6 +177,37 @@ impl SelectBuilder {
         self.lateral_views = value;
         self
     }
+
+    /// Replaces the selection with a new value.
+    ///
+    /// This function is used to replace a specific expression within the selection.
+    /// Unlike the `selection` method which combines existing and new selections with AND,
+    /// this method searches for and replaces occurrences of a specific expression.
+    ///
+    /// This method is primarily used to modify LEFT MARK JOIN expressions.
+    /// When processing a LEFT MARK JOIN, we need to replace the placeholder expression
+    /// with the actual join condition in the selection clause.
+    ///
+    /// # Arguments
+    ///
+    /// * `existing_expr` - The expression to replace
+    /// * `value` - The new expression to set as the selection
+    pub fn replace_mark(
+        &mut self,
+        existing_expr: &ast::Expr,
+        value: &ast::Expr,
+    ) -> &mut Self {
+        if let Some(selection) = &mut self.selection {
+            visit_expressions_mut(selection, |expr| {
+                if expr == existing_expr {
+                    *expr = value.clone();
+                }
+                ControlFlow::<()>::Continue(())
+            });
+        }
+        self
+    }
+
     pub fn selection(&mut self, value: Option<ast::Expr>) -> &mut Self {
         // With filter pushdown optimization, the LogicalPlan can have filters defined as part of `TableScan` and `Filter` nodes.
         // To avoid overwriting one of the filters, we combine the existing filter with the additional filter.

datafusion/sql/src/unparser/expr.rs

@@ -94,6 +94,7 @@ impl Unparser<'_> {
         Ok(root_expr)
     }
 
+    #[cfg_attr(feature = "recursive_protection", recursive::recursive)]
     fn expr_to_sql_inner(&self, expr: &Expr) -> Result<ast::Expr> {
         match expr {
             Expr::InList(InList {
@@ -674,7 +675,7 @@ impl Unparser<'_> {
         }
     }
 
-    fn col_to_sql(&self, col: &Column) -> Result<ast::Expr> {
+    pub fn col_to_sql(&self, col: &Column) -> Result<ast::Expr> {
         if let Some(table_ref) = &col.relation {
             let mut id = if self.dialect.full_qualified_col() {
                 table_ref.to_vec()

datafusion/sql/src/unparser/plan.rs

@@ -322,6 +322,7 @@ impl Unparser<'_> {
         }
     }
 
+    #[cfg_attr(feature = "recursive_protection", recursive::recursive)]
     fn select_to_sql_recursively(
         &self,
         plan: &LogicalPlan,
@@ -566,14 +567,20 @@ impl Unparser<'_> {
             }
             LogicalPlan::Join(join) => {
                 let mut table_scan_filters = vec![];
+                let (left_plan, right_plan) = match join.join_type {
+                    JoinType::RightSemi | JoinType::RightAnti => {
+                        (&join.right, &join.left)
+                    }
+                    _ => (&join.left, &join.right),
+                };
 
                 let left_plan =
-                    match try_transform_to_simple_table_scan_with_filters(&join.left)? {
+                    match try_transform_to_simple_table_scan_with_filters(left_plan)? {
                         Some((plan, filters)) => {
                             table_scan_filters.extend(filters);
                             Arc::new(plan)
                         }
-                        None => Arc::clone(&join.left),
+                        None => Arc::clone(left_plan),
                     };
 
                 self.select_to_sql_recursively(
@@ -584,12 +591,12 @@ impl Unparser<'_> {
                 )?;
 
                 let right_plan =
-                    match try_transform_to_simple_table_scan_with_filters(&join.right)? {
+                    match try_transform_to_simple_table_scan_with_filters(right_plan)? {
                         Some((plan, filters)) => {
                             table_scan_filters.extend(filters);
                             Arc::new(plan)
                         }
-                        None => Arc::clone(&join.right),
+                        None => Arc::clone(right_plan),
                     };
 
                 let mut right_relation = RelationBuilder::default();
@@ -641,19 +648,70 @@ impl Unparser<'_> {
                     &mut right_relation,
                 )?;
 
-                let Ok(Some(relation)) = right_relation.build() else {
-                    return internal_err!("Failed to build right relation");
-                };
-
-                let ast_join = ast::Join {
-                    relation,
-                    global: false,
-                    join_operator: self
-                        .join_operator_to_sql(join.join_type, join_constraint)?,
+                match join.join_type {
+                    JoinType::LeftSemi
+                    | JoinType::LeftAnti
+                    | JoinType::LeftMark
+                    | JoinType::RightSemi
+                    | JoinType::RightAnti => {
+                        let mut query_builder = QueryBuilder::default();
+                        let mut from = TableWithJoinsBuilder::default();
+                        let mut exists_select: SelectBuilder = SelectBuilder::default();
+                        from.relation(right_relation);
+                        exists_select.push_from(from);
+                        if let Some(filter) = &join.filter {
+                            exists_select.selection(Some(self.expr_to_sql(filter)?));
+                        }
+                        for (left, right) in &join.on {
+                            exists_select.selection(Some(
+                                self.expr_to_sql(&left.clone().eq(right.clone()))?,
+                            ));
+                        }
+                        exists_select.projection(vec![ast::SelectItem::UnnamedExpr(
+                            ast::Expr::Value(ast::Value::Number("1".to_string(), false)),
+                        )]);
+                        query_builder.body(Box::new(SetExpr::Select(Box::new(
+                            exists_select.build()?,
+                        ))));
+
+                        let negated = match join.join_type {
+                            JoinType::LeftSemi
+                            | JoinType::RightSemi
+                            | JoinType::LeftMark => false,
+                            JoinType::LeftAnti | JoinType::RightAnti => true,
+                            _ => unreachable!(),
+                        };
+                        let exists_expr = ast::Expr::Exists {
+                            subquery: Box::new(query_builder.build()?),
+                            negated,
+                        };
+                        if join.join_type == JoinType::LeftMark {
+                            let (table_ref, _) = right_plan.schema().qualified_field(0);
+                            let column = self
+                                .col_to_sql(&Column::new(table_ref.cloned(), "mark"))?;
+                            select.replace_mark(&column, &exists_expr);
+                        } else {
+                            select.selection(Some(exists_expr));
+                        }
+                    }
+                    JoinType::Inner
+                    | JoinType::Left
+                    | JoinType::Right
+                    | JoinType::Full => {
+                        let Ok(Some(relation)) = right_relation.build() else {
+                            return internal_err!("Failed to build right relation");
+                        };
+                        let ast_join = ast::Join {
+                            relation,
+                            global: false,
+                            join_operator: self
+                                .join_operator_to_sql(join.join_type, join_constraint)?,
+                        };
+                        let mut from = select.pop_from().unwrap();
+                        from.push_join(ast_join);
+                        select.push_from(from);
+                    }
                 };
-                let mut from = select.pop_from().unwrap();
-                from.push_join(ast_join);
-                select.push_from(from);
 
                 Ok(())
             }