chore: prepare for DataFusion 56

Cargo.toml

@@ -33,12 +33,12 @@ version = "0.12.0"
 [workspace.dependencies]
 anyhow = "1.0.72"
 async-trait = "0.1.89"
-datafusion = "55.0.0"
-datafusion-cli = "55.0.0"
+datafusion = { git = "https://github.com/apache/datafusion.git", rev = "1635ed94742c1c060852bc34dc6ebb2302241c7b" }
+datafusion-cli = { git = "https://github.com/apache/datafusion.git", rev = "1635ed94742c1c060852bc34dc6ebb2302241c7b" }
 iceberg-datafusion = { path = "crates/datafusion" }
-datafusion-sqllogictest = "55.0.0"
-iceberg = { git = "https://github.com/apache/iceberg-rust", rev = "5b82ab99aeaf50bf4f5b3b776ac9cd6380a5f3e3" }
-iceberg-catalog-rest = { git = "https://github.com/apache/iceberg-rust", rev = "5b82ab99aeaf50bf4f5b3b776ac9cd6380a5f3e3" }
+datafusion-sqllogictest = { git = "https://github.com/apache/datafusion.git", rev = "1635ed94742c1c060852bc34dc6ebb2302241c7b" }
+iceberg = { git = "https://github.com/apache/iceberg-rust.git", rev = "73e772c7dbc18f4a3f9f92187b6c56cafbc2f145" }
+iceberg-catalog-rest = { git = "https://github.com/apache/iceberg-rust.git", rev = "73e772c7dbc18f4a3f9f92187b6c56cafbc2f145" }
 tokio = { version = "1.47", default-features = false, features = ["sync", "rt-multi-thread"] }
 toml = "1.1"
 tracing = "0.1.41"

crates/datafusion/Cargo.toml

@@ -35,15 +35,15 @@ dashmap = "6"
 datafusion = { workspace = true }
 futures = "0.3"
 iceberg = { workspace = true }
-parquet = "59.2"
+parquet = "60"
 tokio = { workspace = true }
 uuid = { version = "1.18", features = ["v7"] }
 serde = { version = "1", features = ["derive"] }
 typetag = "0.2"
 
 [dev-dependencies]
 insta = "1"
-parquet = "59.2"
+parquet = "60"
 tempfile = "3.18"
 
 [lints]

crates/datafusion/src/metadata/scan.rs

@@ -44,7 +44,7 @@ impl IcebergMetadataDataSource {
     /// Creates a source with the requested output columns.
     pub fn try_new(
         provider: IcebergMetadataTableProvider,
-        projection: Option<&Vec<usize>>,
+        projection: Option<&[usize]>,
     ) -> Result<Self> {
         let schema = match projection {
             Some(indices) => Arc::new(provider.schema().project(indices)?),
@@ -53,7 +53,7 @@ impl IcebergMetadataDataSource {
         Ok(Self {
             provider,
             schema,
-            projection: projection.cloned(),
+            projection: projection.map(<[usize]>::to_vec),
         })
     }
 

crates/datafusion/src/metadata/table.rs

@@ -62,7 +62,7 @@ impl TableProvider for IcebergMetadataTableProvider {
     async fn scan(
         &self,
         _state: &dyn Session,
-        projection: Option<&Vec<usize>>,
+        projection: Option<&[usize]>,
         _filters: &[Expr],
         _limit: Option<usize>,
     ) -> Result<Arc<dyn ExecutionPlan>> {

crates/datafusion/src/read/scan.rs

@@ -115,8 +115,8 @@ impl IcebergDataSourceBuilder {
     /// Returns only the columns of the schema at these indices, or all of them
     /// if `None` (the default). Indices are into the schema passed to
     /// [`IcebergDataSource::builder`].
-    pub fn with_projection(mut self, projection: Option<&Vec<usize>>) -> Self {
-        self.projection = projection.cloned();
+    pub fn with_projection(mut self, projection: Option<&[usize]>) -> Self {
+        self.projection = projection.map(<[usize]>::to_vec);
         self
     }
 

crates/datafusion/src/table_provider/catalog_table_provider.rs

@@ -127,7 +127,7 @@ impl TableProvider for IcebergCatalogTableProvider {
     async fn scan(
         &self,
         _state: &dyn Session,
-        projection: Option<&Vec<usize>>,
+        projection: Option<&[usize]>,
         filters: &[Expr],
         limit: Option<usize>,
     ) -> Result<Arc<dyn ExecutionPlan>> {
@@ -456,7 +456,7 @@ mod tests {
 
         // Columns come back in the projection's order, not the table's.
         let source = IcebergDataSource::builder(table.clone(), schema.clone())
-            .with_projection(Some(&vec![1, 0]))
+            .with_projection(Some(&[1, 0]))
             .build()
             .unwrap();
         insta::assert_debug_snapshot!(source.projection(), @r#"
@@ -467,7 +467,7 @@ mod tests {
         "#);
 
         let err = IcebergDataSource::builder(table, schema)
-            .with_projection(Some(&vec![0, out_of_range]))
+            .with_projection(Some(&[0, out_of_range]))
             .build()
             .unwrap_err();
         assert_eq!(

crates/datafusion/src/table_provider/static_table_provider.rs

@@ -121,7 +121,7 @@ impl TableProvider for IcebergStaticTableProvider {
     async fn scan(
         &self,
         _state: &dyn Session,
-        projection: Option<&Vec<usize>>,
+        projection: Option<&[usize]>,
         filters: &[Expr],
         limit: Option<usize>,
     ) -> Result<Arc<dyn ExecutionPlan>> {

crates/datafusion/tests/filter_pushdown.rs

@@ -33,14 +33,14 @@ mod tests {
             )
             .await?;
 
-        insta::assert_snapshot!(plan, @r"
+        insta::assert_snapshot!(plan, @"
         ProjectionExec: expr=[count(Int64(1))@0 as trips]
           AggregateExec: mode=Final, gby=[], aggr=[count(Int64(1))]
             CoalescePartitionsExec
               AggregateExec: mode=Partial, gby=[], aggr=[count(Int64(1))]
-                FilterExec: pickup_date@2 = 2024-01-10 AND (payment_type@1 = 1 OR payment_type@1 = 2) AND trip_distance@0 >= 2, projection=[]
+                FilterExec: pickup_date@2 = 2024-01-10 AND payment_type@1 IN (SET) ([1, 2]) AND trip_distance@0 >= 2, projection=[]
                   RepartitionExec: partitioning=RoundRobinBatch(16), input_partitions=1
-                    DataSourceExec: format=iceberg, projection=[trip_distance, payment_type, pickup_date], predicate=((pickup_date = 2024-01-10) AND ((payment_type = 1) OR (payment_type = 2))) AND (trip_distance >= 2)
+                    DataSourceExec: format=iceberg, projection=[trip_distance, payment_type, pickup_date], predicate=((pickup_date = 2024-01-10) AND (payment_type IN (1, 2))) AND (trip_distance >= 2)
         ");
         insta::assert_snapshot!(batches, @r"
     +-------+

crates/datafusion/tests/integration_datafusion_test.rs

@@ -735,27 +735,23 @@ async fn test_insert_into_partitioned() -> Result<(), Box<dyn Error>> {
     // Identity-partition on category and both floating-point values
     let partition_spec = UnboundPartitionSpec::builder()
         .with_spec_id(0)
-        .add_partition_field(
+        .add_partition_fields([
             UnboundPartitionField::builder()
                 .source_ids(vec![2])
                 .name("category")
                 .transform(Transform::Identity)
                 .build()?,
-        )?
-        .add_partition_field(
             UnboundPartitionField::builder()
                 .source_ids(vec![4])
                 .name("float_partition")
                 .transform(Transform::Identity)
                 .build()?,
-        )?
-        .add_partition_field(
             UnboundPartitionField::builder()
                 .source_ids(vec![5])
                 .name("double_partition")
                 .transform(Transform::Identity)
                 .build()?,
-        )?
+        ])?
         .build();
 
     // Create the partitioned table

crates/sqllogictest/testdata/slts/df_test/binary_predicate_pushdown.slt

@@ -23,8 +23,8 @@ query TT
 EXPLAIN SELECT * FROM default.default.test_binary_table WHERE data = X'0102'
 ----
 logical_plan
-01)Filter: default.default.test_binary_table.data = LargeBinary("1,2")
-02)--TableScan: default.default.test_binary_table projection=[id, data], partial_filters=[default.default.test_binary_table.data = LargeBinary("1,2")]
+01)Filter: default.default.test_binary_table.data = Binary("1,2")
+02)--TableScan: default.default.test_binary_table projection=[id, data], partial_filters=[default.default.test_binary_table.data = Binary("1,2")]
 physical_plan
 01)FilterExec: data@1 = 0102
 02)--RepartitionExec: partitioning=RoundRobinBatch(4), input_partitions=1

crates/sqllogictest/testdata/slts/df_test/scan_all_types.slt

@@ -51,7 +51,7 @@ query IIRRTTT rowsort
 SELECT * FROM default.default.all_primitive_types
 ----
 -1 -100 -1.5 -2.5 -123.45 negative false
-0 0 0 0 0 zero NULL
+0 0 0 0 0.00 zero NULL
 1 100 1.5 2.5 123.45 hello true
 2 200 2.5 3.5 234.56 world false
 3 300 3.5 4.5 345.67 NULL true
@@ -73,7 +73,7 @@ SELECT col_int, col_decimal FROM default.default.all_primitive_types WHERE col_d
 query IIRRTTT rowsort
 SELECT * FROM default.default.all_primitive_types WHERE col_int >= 0 ORDER BY col_int
 ----
-0 0 0 0 0 zero NULL
+0 0 0 0 0.00 zero NULL
 1 100 1.5 2.5 123.45 hello true
 2 200 2.5 3.5 234.56 world false
 3 300 3.5 4.5 345.67 NULL true

crates/sqllogictest/testdata/slts/df_test/timestamp_predicate_pushdown.slt

@@ -253,12 +253,12 @@ EXPLAIN SELECT id FROM default.default.test_timestamptz WHERE ts_naive > '2023-0
 ----
 logical_plan
 01)Projection: default.default.test_timestamptz.id
-02)--Filter: default.default.test_timestamptz.ts_naive > TimestampNanosecond(1672876800000000000, None)
-03)----TableScan: default.default.test_timestamptz projection=[id, ts_naive], partial_filters=[default.default.test_timestamptz.ts_naive > TimestampNanosecond(1672876800000000000, None)]
+02)--Filter: CAST(default.default.test_timestamptz.ts_naive AS Timestamp(ns, "UTC")) > TimestampNanosecond(1672876800000000000, Some("UTC"))
+03)----TableScan: default.default.test_timestamptz projection=[id, ts_naive], partial_filters=[CAST(default.default.test_timestamptz.ts_naive AS Timestamp(ns, "UTC")) > TimestampNanosecond(1672876800000000000, Some("UTC"))]
 physical_plan
-01)FilterExec: ts_naive@1 > 1672876800000000000, projection=[id@0]
+01)FilterExec: CAST(ts_naive@1 AS Timestamp(ns, "UTC")) > 1672876800000000000, projection=[id@0]
 02)--RepartitionExec: partitioning=RoundRobinBatch(4), input_partitions=1
-03)----DataSourceExec: format=iceberg, projection=[id, ts_naive], predicate=ts_naive > 2023-01-05 00:00:00
+03)----DataSourceExec: format=iceberg, projection=[id, ts_naive]
 
 query I
 SELECT id FROM default.default.test_timestamptz WHERE ts_naive > '2023-01-05T00:00:00+00:00'::TIMESTAMPTZ