feat(datafusion): upgrade to DataFusion 53 and use VERSION AS OF (#236)

Upgrade DataFusion from 52.3 to 53.0 (arrow/parquet 57→58, sqlparser 0.59→0.61,
orc-rust 0.7→0.8, pyo3 0.26→0.28) and replace the old `FOR SYSTEM_TIME AS OF`
time travel syntax with the new `VERSION AS OF` and `TIMESTAMP AS OF` syntax
supported by sqlparser 0.61.

Introduce `scan.version` option to unify snapshot id and tag name resolution:
at scan time, the version value is resolved by first checking if a tag with
that name exists, then trying to parse it as a snapshot id, otherwise returning
an error. Remove the now-redundant `scan.snapshot-id` and `scan.tag-name` options.

Author: JingsongLi

Cargo.toml

@@ -28,15 +28,15 @@ license = "Apache-2.0"
 rust-version = "1.86.0"
 
 [workspace.dependencies]
-arrow = "57.0"
-arrow-array = { version = "57.0", features = ["ffi"] }
-arrow-buffer = "57.0"
-arrow-schema = "57.0"
-arrow-cast = "57.0"
-arrow-ord = "57.0"
-arrow-select = "57.0"
-datafusion = "52.3.0"
-datafusion-ffi = "52.3.0"
-parquet = "57.0"
+arrow = "58.0"
+arrow-array = { version = "58.0", features = ["ffi"] }
+arrow-buffer = "58.0"
+arrow-schema = "58.0"
+arrow-cast = "58.0"
+arrow-ord = "58.0"
+arrow-select = "58.0"
+datafusion = "53.0.0"
+datafusion-ffi = "53.0.0"
+parquet = "58.0"
 tokio = "1.39.2"
 tokio-util = "0.7"

bindings/python/Cargo.toml

@@ -32,5 +32,5 @@ datafusion = { workspace = true }
 datafusion-ffi = { workspace = true }
 paimon = { path = "../../crates/paimon", features = ["storage-all"] }
 paimon-datafusion = { path = "../../crates/integrations/datafusion" }
-pyo3 = { version = "0.26", features = ["abi3-py310"] }
+pyo3 = { version = "0.28", features = ["abi3-py310"] }
 tokio = { workspace = true }

bindings/python/pyproject.toml

@@ -53,5 +53,5 @@ dev = [
     "maturin>=1.9.4,<2.0",
     "pytest>=8.0",
     "pyarrow>=17.0",
-    "datafusion==52.3.0",
+    "datafusion==53.0.0",
 ]

bindings/python/src/context.rs

@@ -16,15 +16,13 @@
 // under the License.
 
 use std::collections::HashMap;
-use std::ptr::NonNull;
 use std::sync::Arc;
 
 use datafusion::catalog::CatalogProvider;
 use datafusion_ffi::catalog_provider::FFI_CatalogProvider;
 use datafusion_ffi::proto::logical_extension_codec::FFI_LogicalExtensionCodec;
 use paimon::{CatalogFactory, Options};
 use paimon_datafusion::PaimonCatalogProvider;
-use pyo3::exceptions::PyValueError;
 use pyo3::prelude::*;
 use pyo3::types::PyCapsule;
 
@@ -52,23 +50,8 @@ fn ffi_logical_codec_from_pycapsule(obj: Bound<'_, PyAny>) -> PyResult<FFI_Logic
 
     let capsule = capsule.cast::<PyCapsule>()?;
     let expected_name = c"datafusion_logical_extension_codec";
-    match capsule.name()? {
-        Some(name) if name == expected_name => {}
-        Some(name) => {
-            return Err(PyValueError::new_err(format!(
-                "Expected capsule named {expected_name:?}, got {name:?}"
-            )));
-        }
-        None => {
-            return Err(PyValueError::new_err(format!(
-                "Expected capsule named {expected_name:?}, got unnamed capsule"
-            )));
-        }
-    }
-
-    let data = NonNull::new(capsule.pointer().cast::<FFI_LogicalExtensionCodec>())
-        .ok_or_else(|| PyValueError::new_err("Null logical extension codec capsule pointer"))?;
-    let codec = unsafe { data.as_ref() };
+    let ptr = capsule.pointer_checked(Some(expected_name))?;
+    let codec = unsafe { ptr.cast::<FFI_LogicalExtensionCodec>().as_ref() };
 
     Ok(codec.clone())
 }

crates/integration_tests/tests/read_tables.rs

@@ -1699,7 +1699,7 @@ async fn test_time_travel_by_snapshot_id() {
 
     // Snapshot 1: (1, 'alice'), (2, 'bob')
     let table_snap1 = table.copy_with_options(HashMap::from([(
-        "scan.snapshot-id".to_string(),
+        "scan.version".to_string(),
         "1".to_string(),
     )]));
     let rb = table_snap1.new_read_builder();
@@ -1720,7 +1720,7 @@ async fn test_time_travel_by_snapshot_id() {
 
     // Snapshot 2: (1, 'alice'), (2, 'bob'), (3, 'carol'), (4, 'dave')
     let table_snap2 = table.copy_with_options(HashMap::from([(
-        "scan.snapshot-id".to_string(),
+        "scan.version".to_string(),
         "2".to_string(),
     )]));
     let rb2 = table_snap2.new_read_builder();
@@ -1753,7 +1753,7 @@ async fn test_time_travel_by_tag_name() {
 
     // Tag 'snapshot1' -> snapshot 1: (1, 'alice'), (2, 'bob')
     let table_tag1 = table.copy_with_options(HashMap::from([(
-        "scan.tag-name".to_string(),
+        "scan.version".to_string(),
         "snapshot1".to_string(),
     )]));
     let rb = table_tag1.new_read_builder();
@@ -1774,7 +1774,7 @@ async fn test_time_travel_by_tag_name() {
 
     // Tag 'snapshot2' -> snapshot 2: all 4 rows
     let table_tag2 = table.copy_with_options(HashMap::from([(
-        "scan.tag-name".to_string(),
+        "scan.version".to_string(),
         "snapshot2".to_string(),
     )]));
     let rb2 = table_tag2.new_read_builder();
@@ -1805,8 +1805,8 @@ async fn test_time_travel_conflicting_selectors_fail() {
     let table = get_table_from_catalog(&catalog, "time_travel_table").await;
 
     let conflicted = table.copy_with_options(HashMap::from([
-        ("scan.tag-name".to_string(), "snapshot1".to_string()),
-        ("scan.snapshot-id".to_string(), "2".to_string()),
+        ("scan.version".to_string(), "snapshot1".to_string()),
+        ("scan.timestamp-millis".to_string(), "1234".to_string()),
     ]));
 
     let plan_err = conflicted
@@ -1823,40 +1823,40 @@ async fn test_time_travel_conflicting_selectors_fail() {
                 "unexpected conflict error: {message}"
             );
             assert!(
-                message.contains("scan.snapshot-id"),
-                "conflict error should mention scan.snapshot-id: {message}"
+                message.contains("scan.version"),
+                "conflict error should mention scan.version: {message}"
             );
             assert!(
-                message.contains("scan.tag-name"),
-                "conflict error should mention scan.tag-name: {message}"
+                message.contains("scan.timestamp-millis"),
+                "conflict error should mention scan.timestamp-millis: {message}"
             );
         }
         other => panic!("unexpected error: {other:?}"),
     }
 }
 
 #[tokio::test]
-async fn test_time_travel_invalid_numeric_selector_fails() {
+async fn test_time_travel_invalid_version_fails() {
     let catalog = create_file_system_catalog();
     let table = get_table_from_catalog(&catalog, "time_travel_table").await;
 
     let invalid = table.copy_with_options(HashMap::from([(
-        "scan.snapshot-id".to_string(),
-        "not-a-number".to_string(),
+        "scan.version".to_string(),
+        "nonexistent-tag".to_string(),
     )]));
 
     let plan_err = invalid
         .new_read_builder()
         .new_scan()
         .plan()
         .await
-        .expect_err("invalid numeric time-travel selector should fail");
+        .expect_err("invalid version should fail");
 
     match plan_err {
         Error::DataInvalid { message, .. } => {
             assert!(
-                message.contains("Invalid value for scan.snapshot-id"),
-                "unexpected invalid selector error: {message}"
+                message.contains("is not a valid tag name or snapshot id"),
+                "unexpected invalid version error: {message}"
             );
         }
         other => panic!("unexpected error: {other:?}"),

crates/integrations/datafusion/src/ddl.rs

@@ -99,12 +99,14 @@ impl PaimonDdlHandler {
 
         match &statements[0] {
             Statement::CreateTable(create_table) => self.handle_create_table(create_table).await,
-            Statement::AlterTable {
-                name,
-                operations,
-                if_exists,
-                ..
-            } => self.handle_alter_table(name, operations, *if_exists).await,
+            Statement::AlterTable(alter_table) => {
+                self.handle_alter_table(
+                    &alter_table.name,
+                    &alter_table.operations,
+                    alter_table.if_exists,
+                )
+                .await
+            }
             _ => self.ctx.sql(sql).await,
         }
     }
@@ -146,12 +148,12 @@ impl PaimonDdlHandler {
 
         // Primary key from constraints: PRIMARY KEY (col, ...)
         for constraint in &ct.constraints {
-            if let datafusion::sql::sqlparser::ast::TableConstraint::PrimaryKey {
-                columns, ..
-            } = constraint
-            {
-                let pk_cols: Vec<String> =
-                    columns.iter().map(|c| c.column.expr.to_string()).collect();
+            if let datafusion::sql::sqlparser::ast::TableConstraint::PrimaryKey(pk) = constraint {
+                let pk_cols: Vec<String> = pk
+                    .columns
+                    .iter()
+                    .map(|c| c.column.expr.to_string())
+                    .collect();
                 builder = builder.primary_key(pk_cols);
             }
         }

crates/integrations/datafusion/src/physical_plan/scan.rs

@@ -51,7 +51,7 @@ pub struct PaimonTableScan {
     /// Paimon splits that DataFusion partition `i` will read.
     /// Wrapped in `Arc` to avoid deep-cloning `DataSplit` metadata in `execute()`.
     planned_partitions: Vec<Arc<[DataSplit]>>,
-    plan_properties: PlanProperties,
+    plan_properties: Arc<PlanProperties>,
     /// Optional limit on the number of rows to return.
     limit: Option<usize>,
 }
@@ -65,12 +65,12 @@ impl PaimonTableScan {
         planned_partitions: Vec<Arc<[DataSplit]>>,
         limit: Option<usize>,
     ) -> Self {
-        let plan_properties = PlanProperties::new(
+        let plan_properties = Arc::new(PlanProperties::new(
             EquivalenceProperties::new(schema.clone()),
             Partitioning::UnknownPartitioning(planned_partitions.len()),
             EmissionType::Incremental,
             Boundedness::Bounded,
-        );
+        ));
         Self {
             table,
             projected_columns,
@@ -109,7 +109,7 @@ impl ExecutionPlan for PaimonTableScan {
         self
     }
 
-    fn properties(&self) -> &PlanProperties {
+    fn properties(&self) -> &Arc<PlanProperties> {
         &self.plan_properties
     }
 
@@ -168,10 +168,6 @@ impl ExecutionPlan for PaimonTableScan {
         )))
     }
 
-    fn statistics(&self) -> DFResult<Statistics> {
-        self.partition_statistics(None)
-    }
-
     fn partition_statistics(&self, partition: Option<usize>) -> DFResult<Statistics> {
         let partitions: &[Arc<[DataSplit]>] = match partition {
             Some(idx) => std::slice::from_ref(&self.planned_partitions[idx]),