fix: Remove duplicate filter from CrossJoin unparsing (#17382)

* fix: Remove duplicate filter from `CrossJoin` unparsing

* move test

* add

Author: jonathanc-n

datafusion/core/tests/sql/joins.rs

@@ -18,8 +18,10 @@
 use insta::assert_snapshot;
 
 use datafusion::assert_batches_eq;
+use datafusion::catalog::MemTable;
 use datafusion::datasource::stream::{FileStreamProvider, StreamConfig, StreamTable};
 use datafusion::test_util::register_unbounded_file_with_ordering;
+use datafusion_sql::unparser::plan_to_sql;
 
 use super::*;
 
@@ -264,3 +266,45 @@ async fn join_using_uppercase_column() -> Result<()> {
 
     Ok(())
 }
+
+// Issue #17359: https://github.com/apache/datafusion/issues/17359
+#[tokio::test]
+async fn unparse_cross_join() -> Result<()> {
+    let ctx = SessionContext::new();
+
+    let j1_schema = Arc::new(Schema::new(vec![
+        Field::new("j1_id", DataType::Int32, true),
+        Field::new("j1_string", DataType::Utf8, true),
+    ]));
+    let j2_schema = Arc::new(Schema::new(vec![
+        Field::new("j2_id", DataType::Int32, true),
+        Field::new("j2_string", DataType::Utf8, true),
+    ]));
+
+    ctx.register_table("j1", Arc::new(MemTable::try_new(j1_schema, vec![vec![]])?))?;
+    ctx.register_table("j2", Arc::new(MemTable::try_new(j2_schema, vec![vec![]])?))?;
+
+    let df = ctx
+        .sql(
+            r#"
+            select j1.j1_id, j2.j2_string
+            from j1, j2
+            where j2.j2_id = 0
+            "#,
+        )
+        .await?;
+
+    let unopt_sql = plan_to_sql(df.logical_plan())?;
+    assert_snapshot!(unopt_sql, @r#"
+        SELECT j1.j1_id, j2.j2_string FROM j1 CROSS JOIN j2 WHERE (j2.j2_id = 0)
+    "#);
+
+    let optimized_plan = df.into_optimized_plan()?;
+
+    let opt_sql = plan_to_sql(&optimized_plan)?;
+    assert_snapshot!(opt_sql, @r#"
+        SELECT j1.j1_id, j2.j2_string FROM j1 CROSS JOIN j2 WHERE (j2.j2_id = 0)
+    "#);
+
+    Ok(())
+}

datafusion/sql/src/unparser/plan.rs

@@ -696,13 +696,6 @@ impl Unparser<'_> {
                     join_filters.as_ref(),
                 )?;
 
-                self.select_to_sql_recursively(
-                    right_plan.as_ref(),
-                    query,
-                    select,
-                    &mut right_relation,
-                )?;
-
                 let right_projection: Option<Vec<ast::SelectItem>> = if !already_projected
                 {
                     Some(select.pop_projections())

datafusion/sql/tests/cases/plan_to_sql.rs

@@ -16,6 +16,7 @@
 // under the License.
 
 use arrow::datatypes::{DataType, Field, Schema};
+
 use datafusion_common::{
     assert_contains, Column, DFSchema, DFSchemaRef, DataFusionError, Result,
     TableReference,