DataFusion 52 migration

Author: comphead

native/Cargo.lock

@@ -41,6 +41,7 @@ parquet = { version = "57.0.0", default-features = false, features = ["experimen
 datafusion = { version = "52.0.0", default-features = false, features = ["unicode_expressions", "crypto_expressions", "nested_expressions", "parquet"] }
 datafusion-datasource = { version = "52.0.0" }
 datafusion-spark = { version = "52.0.0" }
+datafusion-physical-expr-adapter = { version = "52.0.0" }
 datafusion-comet-spark-expr = { path = "spark-expr" }
 datafusion-comet-proto = { path = "proto" }
 chrono = { version = "0.4", default-features = false, features = ["clock"] }

native/Cargo.toml

@@ -60,6 +60,7 @@ tempfile = "3.24.0"
 itertools = "0.14.0"
 paste = "1.0.14"
 datafusion = { workspace = true, features = ["parquet_encryption", "sql"] }
+datafusion-physical-expr-adapter = { workspace = true }
 datafusion-datasource = { workspace = true }
 datafusion-spark = { workspace = true }
 once_cell = "1.18.0"

native/core/Cargo.toml

@@ -42,9 +42,9 @@ use iceberg::io::FileIO;
 
 use crate::execution::operators::ExecutionError;
 use crate::parquet::parquet_support::SparkParquetOptions;
+use crate::parquet::schema_adapter::adapt_batch_with_expressions;
 use datafusion_comet_spark_expr::EvalMode;
 use datafusion_datasource::file_stream::FileStreamMetrics;
-use crate::parquet::schema_adapter::adapt_batch_with_expressions;
 
 /// Iceberg table scan operator that uses iceberg-rust to read Iceberg tables.
 ///

native/core/src/execution/operators/iceberg_scan.rs

@@ -1049,8 +1049,8 @@ impl PhysicalPlanner {
                                 .as_any()
                                 .downcast_ref::<DataFusionLiteral>()
                                 .ok_or_else(|| {
-                                    GeneralError("Expected literal of default value.".to_string())
-                                })?;
+                                GeneralError("Expected literal of default value.".to_string())
+                            })?;
                             Ok(df_literal.value().clone())
                         })
                         .collect();

native/core/src/execution/planner.rs

@@ -15,10 +15,10 @@
 // specific language governing permissions and limitations
 // under the License.
 
-use std::collections::HashMap;
 use crate::execution::operators::ExecutionError;
 use crate::parquet::encryption_support::{CometEncryptionConfig, ENCRYPTION_FACTORY_ID};
 use crate::parquet::parquet_support::SparkParquetOptions;
+use crate::parquet::schema_adapter::SparkPhysicalExprAdapterFactory;
 use arrow::datatypes::{Field, SchemaRef};
 use datafusion::config::TableParquetOptions;
 use datafusion::datasource::listing::PartitionedFile;
@@ -29,13 +29,13 @@ use datafusion::datasource::source::DataSourceExec;
 use datafusion::execution::object_store::ObjectStoreUrl;
 use datafusion::physical_expr::expressions::BinaryExpr;
 use datafusion::physical_expr::PhysicalExpr;
+use datafusion::physical_expr_adapter::PhysicalExprAdapterFactory;
 use datafusion::prelude::SessionContext;
+use datafusion::scalar::ScalarValue;
 use datafusion_comet_spark_expr::EvalMode;
 use datafusion_datasource::TableSchema;
+use std::collections::HashMap;
 use std::sync::Arc;
-use datafusion::physical_expr_adapter::PhysicalExprAdapterFactory;
-use datafusion::scalar::ScalarValue;
-use crate::parquet::schema_adapter::SparkPhysicalExprAdapterFactory;
 
 /// Initializes a DataSourceExec plan with a ParquetSource. This may be used by either the
 /// `native_datafusion` scan or the `native_iceberg_compat` scan.
@@ -138,11 +138,12 @@ pub(crate) fn init_datasource_exec(
         FileScanConfigBuilder::new(object_store_url, file_source).with_file_groups(file_groups);
 
     if let Some(projection_vector) = projection_vector {
-        file_scan_config_builder =
-            file_scan_config_builder.with_projection_indices(Some(projection_vector))?;
+        file_scan_config_builder = file_scan_config_builder
+            .with_projection_indices(Some(projection_vector))?
+            .with_expr_adapter(Some(expr_adapter_factory));
     }
 
-    let file_scan_config = file_scan_config_builder.with_expr_adapter(Some(expr_adapter_factory)).build();
+    let file_scan_config = file_scan_config_builder.build();
 
     Ok(Arc::new(DataSourceExec::new(Arc::new(file_scan_config))))
 }