Eliminate duplicated filter within (filter(TableScan)) plan (#51)

* Eliminate duplicated filter within (filter(TableScan)) plan

* Updates

* fix

* add test

* fix

Author: Sevenannn

datafusion/sql/src/unparser/expr.rs

@@ -18,7 +18,8 @@
 use datafusion_expr::expr::Unnest;
 use sqlparser::ast::Value::SingleQuotedString;
 use sqlparser::ast::{
-    self, BinaryOperator, Expr as AstExpr, Function, Ident, Interval, ObjectName, TimezoneInfo, UnaryOperator
+    self, BinaryOperator, Expr as AstExpr, Function, Ident, Interval, ObjectName,
+    TimezoneInfo, UnaryOperator,
 };
 use std::sync::Arc;
 use std::vec;
@@ -217,7 +218,7 @@ impl Unparser<'_> {
                     }
                 };
 
-                let order_by =order_by
+                let order_by = order_by
                     .iter()
                     .map(|sort_expr| self.sort_to_sql(sort_expr))
                     .collect::<Result<Vec<_>>>()?;
@@ -1881,7 +1882,7 @@ mod tests {
                         name: "array_col".to_string(),
                     })),
                 }),
-                r#"UNNEST("schema"."table".array_col)"#
+                r#"UNNEST("schema"."table".array_col)"#,
             ),
         ];
 

datafusion/sql/src/unparser/utils.rs

@@ -15,7 +15,7 @@
 // specific language governing permissions and limitations
 // under the License.
 
-use std::{cmp::Ordering, sync::Arc, vec};
+use std::{cmp::Ordering, collections::HashSet, sync::Arc, vec};
 
 use datafusion_common::{
     internal_err,
@@ -102,7 +102,7 @@ pub(crate) fn find_unnest_node_within_select(
 ///       TableScan: j1
 /// And filters: [ta.j1_id < 5, ta.j1_id > 10]
 pub (crate) fn try_transform_to_simple_table_scan_with_filters(plan: &LogicalPlan) -> Option<(LogicalPlan, Vec<Expr>)> { 
-    let mut filters: Vec<Expr> = vec![];
+    let mut filters: HashSet<Expr> = HashSet::new();
     let mut plan_stack = vec![plan];
     let mut table_alias = None;
 
@@ -113,7 +113,7 @@ pub (crate) fn try_transform_to_simple_table_scan_with_filters(plan: &LogicalPla
                 plan_stack.push(alias.input.as_ref());
             }
             LogicalPlan::Filter(filter) => {
-                filters.push(filter.predicate.clone());
+                filters.insert(filter.predicate.clone());
                 plan_stack.push(filter.input.as_ref());
             }
             LogicalPlan::TableScan(table_scan) => {
@@ -138,7 +138,9 @@ pub (crate) fn try_transform_to_simple_table_scan_with_filters(plan: &LogicalPla
                         }
                     }).collect::<Result<Vec<_>, DataFusionError>>().ok()?;
 
-                filters.extend(table_scan_filters);
+                for table_scan_filter in table_scan_filters {
+                    filters.insert(table_scan_filter);
+                }
 
                 let mut builder = LogicalPlanBuilder::scan(
                     table_scan.table_name.clone(),
@@ -151,6 +153,7 @@ pub (crate) fn try_transform_to_simple_table_scan_with_filters(plan: &LogicalPla
                 } 
 
                 let plan = builder.build().ok()?;
+                let filters: Vec<Expr> = filters.into_iter().collect();
 
                 return Some((plan, filters));
 

datafusion/sql/tests/cases/plan_to_sql.rs

@@ -1035,10 +1035,9 @@ fn test_join_with_table_scan_filters() -> Result<()> {
 
     assert_eq!(sql.to_string(), expected_sql);
 
-
-    let join_plan_no_filter = LogicalPlanBuilder::from(left_plan)
+    let join_plan_no_filter = LogicalPlanBuilder::from(left_plan.clone())
         .join(
-            right_plan, 
+            right_plan.clone(), 
             datafusion_expr::JoinType::Inner, 
             (vec!["left.id"], vec!["right_table.id"]), 
             None
@@ -1051,6 +1050,25 @@ fn test_join_with_table_scan_filters() -> Result<()> {
 
     assert_eq!(sql.to_string(), expected_sql);
 
+    let right_plan_with_filter_schema = table_scan_with_filters(Some("right_table"), &schema_right, None, vec![col("right_table.age").gt(lit(10))])?
+    .build()?;
+    let right_plan_with_duplicated_filter = LogicalPlanBuilder::from(right_plan_with_filter_schema.clone()).filter(col("right_table.age").gt(lit(10)))?.build()?;
+
+    let join_plan_duplicated_filter = LogicalPlanBuilder::from(left_plan)
+    .join(
+        right_plan_with_duplicated_filter,
+        datafusion_expr::JoinType::Inner, 
+        (vec!["left.id"], vec!["right_table.id"]), 
+        Some(col("left.id").gt(lit(5))),
+    )?
+    .build()?;
+
+    let sql = plan_to_sql(&join_plan_duplicated_filter)?;
+
+    let expected_sql = r#"SELECT * FROM left_table AS "left" JOIN right_table ON "left".id = right_table.id AND (("left".id > 5) AND ("left"."name" LIKE 'some_name' AND (right_table.age > 10)))"#;
+
+    assert_eq!(sql.to_string(), expected_sql);
+
     Ok(())
 }