Upgrade DataFusion 48.0.0 (#61)

* Upgrade to DF48

* fix bug

* update

* update more

Author: xudong963

Cargo.toml

@@ -28,19 +28,19 @@ keywords = ["arrow", "arrow-rs", "datafusion"]
 rust-version = "1.80"
 
 [dependencies]
-arrow = "55"
-arrow-schema = "55"
+arrow = "55.1.0"
+arrow-schema = "55.1.0"
 async-trait = "0.1"
 dashmap = "6"
-datafusion = "47"
-datafusion-common = "47"
-datafusion-expr = "47"
-datafusion-functions = "47"
-datafusion-functions-aggregate = "47"
-datafusion-optimizer = "47"
-datafusion-physical-expr = "47"
-datafusion-physical-plan = "47"
-datafusion-sql = "47"
+datafusion = "48"
+datafusion-common = "48"
+datafusion-expr = "48"
+datafusion-functions = "48"
+datafusion-functions-aggregate = "48"
+datafusion-optimizer = "48"
+datafusion-physical-expr = "48"
+datafusion-physical-plan = "48"
+datafusion-sql = "48"
 futures = "0.3"
 itertools = "0.14"
 log = "0.4"

src/materialized/dependencies.rs

@@ -231,7 +231,7 @@ impl TableFunctionImpl for StaleFilesUdtf {
 /// Extract table name from args passed to TableFunctionImpl::call()
 fn get_table_name(args: &[Expr]) -> Result<&String> {
     match &args[0] {
-        Expr::Literal(ScalarValue::Utf8(Some(table_name))) => Ok(table_name),
+        Expr::Literal(ScalarValue::Utf8(Some(table_name)), _) => Ok(table_name),
         _ => Err(DataFusionError::Plan(
             "expected a single string literal argument to mv_dependencies".to_string(),
         )),

src/rewrite/exploitation.rs

@@ -427,7 +427,14 @@ impl ExecutionPlan for OneOfExec {
     }
 
     fn statistics(&self) -> Result<datafusion_common::Statistics> {
-        self.candidates[self.best].statistics()
+        self.candidates[self.best].partition_statistics(None)
+    }
+
+    fn partition_statistics(
+        &self,
+        partition: Option<usize>,
+    ) -> Result<datafusion_common::Statistics> {
+        self.candidates[self.best].partition_statistics(partition)
     }
 }
 

src/rewrite/normal_form.rs

@@ -455,6 +455,10 @@ impl Predicate {
                 self.eq_classes[idx].columns.insert(c2.clone());
             }
             (Some(&i), Some(&j)) => {
+                if i == j {
+                    // The two columns are already in the same equivalence class.
+                    return Ok(());
+                }
                 // We need to merge two existing column eq classes.
 
                 // Delete the eq class with a larger index,
@@ -593,15 +597,15 @@ impl Predicate {
     /// Add a binary expression to our collection of filters.
     fn insert_binary_expr(&mut self, left: &Expr, op: Operator, right: &Expr) -> Result<()> {
         match (left, op, right) {
-            (Expr::Column(c), op, Expr::Literal(v)) => {
+            (Expr::Column(c), op, Expr::Literal(v, _)) => {
                 if let Err(e) = self.add_range(c, &op, v) {
                     // Add a range can fail in some cases, so just fallthrough
                     log::debug!("failed to add range filter: {e}");
                 } else {
                     return Ok(());
                 }
             }
-            (Expr::Literal(_), op, Expr::Column(_)) => {
+            (Expr::Literal(_, _), op, Expr::Column(_)) => {
                 if let Some(swapped) = op.swap() {
                     return self.insert_binary_expr(right, swapped, left);
                 }
@@ -714,22 +718,22 @@ impl Predicate {
                     extra_range_filters.push(Expr::BinaryExpr(BinaryExpr {
                         left: Box::new(Expr::Column(other_column.clone())),
                         op: Operator::Eq,
-                        right: Box::new(Expr::Literal(range.lower().clone())),
+                        right: Box::new(Expr::Literal(range.lower().clone(), None)),
                     }))
                 } else {
                     if !range.lower().is_null() {
                         extra_range_filters.push(Expr::BinaryExpr(BinaryExpr {
                             left: Box::new(Expr::Column(other_column.clone())),
                             op: Operator::GtEq,
-                            right: Box::new(Expr::Literal(range.lower().clone())),
+                            right: Box::new(Expr::Literal(range.lower().clone(), None)),
                         }))
                     }
 
                     if !range.upper().is_null() {
                         extra_range_filters.push(Expr::BinaryExpr(BinaryExpr {
                             left: Box::new(Expr::Column(other_column.clone())),
                             op: Operator::LtEq,
-                            right: Box::new(Expr::Literal(range.upper().clone())),
+                            right: Box::new(Expr::Literal(range.upper().clone(), None)),
                         }))
                     }
                 }
@@ -984,7 +988,8 @@ mod test {
         let ctx = SessionContext::new_with_config(
             SessionConfig::new()
                 .set_bool("datafusion.execution.parquet.pushdown_filters", true)
-                .set_bool("datafusion.explain.logical_plan_only", true),
+                .set_bool("datafusion.explain.logical_plan_only", true)
+                .set_bool("datafusion.sql_parser.map_varchar_to_utf8view", false),
         );
 
         let t1_path = tempdir()?;
@@ -996,11 +1001,11 @@ mod test {
         ctx.sql(&format!(
             "
                 CREATE EXTERNAL TABLE t1 (
-                    column1 VARCHAR, 
-                    column2 BIGINT, 
+                    column1 VARCHAR,
+                    column2 BIGINT,
                     column3 CHAR
                 )
-                STORED AS PARQUET 
+                STORED AS PARQUET
                 LOCATION '{}'",
             t1_path.path().to_string_lossy()
         ))