feat: arrow backed log replay and table state

Cargo.toml

@@ -24,6 +24,8 @@ arrow-arith = { version = "49" }
 arrow-array = { version = "49" }
 arrow-buffer = { version = "49" }
 arrow-cast = { version = "49" }
+arrow-ipc = { version = "49" }
+arrow-json = { version = "49" }
 arrow-ord = { version = "49" }
 arrow-row = { version = "49" }
 arrow-schema = { version = "49" }

crates/benchmarks/src/bin/merge.rs

@@ -193,10 +193,10 @@ async fn benchmark_merge_tpcds(
     merge: fn(DataFrame, DeltaTable) -> Result<MergeBuilder, DeltaTableError>,
 ) -> Result<(core::time::Duration, MergeMetrics), DataFusionError> {
     let table = DeltaTableBuilder::from_uri(path).load().await?;
-    let file_count = table.state.files().len();
+    let file_count = table.snapshot()?.files_count();
 
     let provider = DeltaTableProvider::try_new(
-        table.state.clone(),
+        table.snapshot()?.clone(),
         table.log_store(),
         DeltaScanConfig {
             file_column_name: Some("file_path".to_string()),

crates/deltalake-aws/Cargo.toml

@@ -26,6 +26,7 @@ url = { workspace = true }
 backoff = { version = "0.4", features = [ "tokio" ] }
 
 [dev-dependencies]
+deltalake-core = { path = "../deltalake-core", features = ["datafusion"] }
 chrono = { workspace = true }
 serial_test = "3"
 deltalake-test = { path = "../deltalake-test" }

crates/deltalake-aws/tests/integration_s3_dynamodb.rs

@@ -267,12 +267,11 @@ fn add_action(name: &str) -> Action {
     let ts = (SystemTime::now() - Duration::from_secs(1800))
         .duration_since(UNIX_EPOCH)
         .unwrap()
-        .as_secs();
-    Action::Add(Add {
+        .as_millis();
+    Add {
         path: format!("{}.parquet", name),
         size: 396,
         partition_values: HashMap::new(),
-        partition_values_parsed: None,
         modification_time: ts as i64,
         data_change: true,
         stats: None,
@@ -282,7 +281,8 @@ fn add_action(name: &str) -> Action {
         base_row_id: None,
         default_row_commit_version: None,
         clustering_provider: None,
-    })
+    }
+    .into()
 }
 
 async fn prepare_table(context: &IntegrationContext, table_name: &str) -> TestResult<DeltaTable> {
@@ -322,7 +322,7 @@ async fn append_to_table(
         table.log_store().as_ref(),
         &actions,
         operation,
-        &table.state,
+        Some(table.snapshot()?),
         metadata,
     )
     .await

crates/deltalake-core/Cargo.toml

@@ -19,19 +19,22 @@ features = ["azure", "datafusion", "gcs", "hdfs", "json", "python", "s3", "unity
 
 [dependencies]
 # arrow
-arrow = { workspace = true, optional = true }
-arrow-arith = { workspace = true, optional = true }
-arrow-array = { workspace = true, optional = true }
-arrow-buffer = { workspace = true, optional = true }
-arrow-cast = { workspace = true, optional = true }
-arrow-ord = { workspace = true, optional = true }
-arrow-row = { workspace = true, optional = true }
-arrow-schema = { workspace = true, optional = true, features = ["serde"] }
-arrow-select = { workspace = true, optional = true }
+arrow = { workspace = true }
+arrow-arith = { workspace = true }
+arrow-array = { workspace = true }
+arrow-buffer = { workspace = true }
+arrow-cast = { workspace = true }
+arrow-ipc = { workspace = true }
+arrow-json = { workspace = true }
+arrow-ord = { workspace = true }
+arrow-row = { workspace = true }
+arrow-schema = { workspace = true, features = ["serde"] }
+arrow-select = { workspace = true }
 parquet = { workspace = true, features = [
     "async",
     "object_store",
-], optional = true }
+] }
+pin-project-lite = "^0.2.7"
 
 # datafusion
 datafusion = { workspace = true, optional = true }
@@ -48,6 +51,7 @@ serde_json = { workspace = true }
 # "stdlib"
 bytes = { workspace = true }
 chrono = { workspace = true, default-features = false, features = ["clock"] }
+hashbrown = "*"
 regex = { workspace = true }
 thiserror = { workspace = true }
 uuid = { workspace = true, features = ["serde", "v4"] }
@@ -111,18 +115,7 @@ tokio = { version = "1", features = ["macros", "rt-multi-thread"] }
 utime = "0.3"
 
 [features]
-arrow = [
-    "dep:arrow",
-    "arrow-arith",
-    "arrow-array",
-    "arrow-cast",
-    "arrow-ord",
-    "arrow-row",
-    "arrow-schema",
-    "arrow-select",
-    "arrow-buffer",
-]
-default = ["arrow", "parquet"]
+default = []
 datafusion = [
     "dep:datafusion",
     "datafusion-expr",
@@ -131,14 +124,8 @@ datafusion = [
     "datafusion-physical-expr",
     "datafusion-sql",
     "sqlparser",
-    "arrow",
-    "parquet",
 ]
 datafusion-ext = ["datafusion"]
 json = ["parquet/json"]
 python = ["arrow/pyarrow"]
 unity-experimental = ["reqwest", "hyper"]
-
-[[bench]]
-name = "read_checkpoint"
-harness = false

crates/deltalake-core/src/delta_datafusion/expr.rs

@@ -575,7 +575,8 @@ mod test {
                     .cast_to::<DFSchema>(
                         &arrow_schema::DataType::Utf8,
                         &table
-                            .state
+                            .snapshot()
+                            .unwrap()
                             .input_schema()
                             .unwrap()
                             .as_ref()
@@ -612,7 +613,8 @@ mod test {
             assert_eq!(test.expected, actual);
 
             let actual_expr = table
-                .state
+                .snapshot()
+                .unwrap()
                 .parse_predicate_expression(actual, &session.state())
                 .unwrap();