feat: Upgrade to arrow/parquet 15 and datafusion 9 (#652)

Co-authored-by: Tim Van Wassenhove <[email protected]>

rust/Cargo.toml

@@ -56,8 +56,8 @@ async-stream = { version = "0.3.2", default-features = true, optional = true }
 # High-level writer
 parquet-format = "~4.0.0"
 
-arrow = "13"
-parquet = "13"
+arrow = "15"
+parquet = "15"
 
 crossbeam = { version = "0", optional = true }
 
@@ -69,7 +69,7 @@ async-trait = "0.1"
 # rust-dataframe = {version = "0.*", optional = true }
 
 [dependencies.datafusion]
-version = "8"
+version = "9"
 optional = true
 
 [features]

rust/src/checkpoints.rs

@@ -12,7 +12,6 @@ use lazy_static::lazy_static;
 use log::*;
 use parquet::arrow::ArrowWriter;
 use parquet::errors::ParquetError;
-use parquet::file::writer::InMemoryWriteableCursor;
 use regex::Regex;
 use serde_json::Value;
 use std::collections::HashMap;
@@ -402,8 +401,8 @@ fn parquet_bytes_from_state(state: &DeltaTableState) -> Result<Vec<u8>, Checkpoi
 
     debug!("Writing to checkpoint parquet buffer...");
     // Write the Checkpoint parquet file.
-    let writeable_cursor = InMemoryWriteableCursor::default();
-    let mut writer = ArrowWriter::try_new(writeable_cursor.clone(), arrow_schema.clone(), None)?;
+    let mut bytes = vec![];
+    let mut writer = ArrowWriter::try_new(&mut bytes, arrow_schema.clone(), None)?;
     let options = DecoderOptions::new().with_batch_size(CHECKPOINT_RECORD_BATCH_SIZE);
     let decoder = Decoder::new(arrow_schema, options);
     while let Some(batch) = decoder.next_batch(&mut jsons)? {
@@ -412,7 +411,7 @@ fn parquet_bytes_from_state(state: &DeltaTableState) -> Result<Vec<u8>, Checkpoi
     let _ = writer.close()?;
     debug!("Finished writing checkpoint parquet buffer.");
 
-    Ok(writeable_cursor.data())
+    Ok(bytes)
 }
 
 fn checkpoint_add_from_state(

rust/src/delta_datafusion.rs

@@ -26,11 +26,12 @@ use std::sync::Arc;
 
 use arrow::datatypes::{DataType as ArrowDataType, Schema as ArrowSchema, TimeUnit};
 use async_trait::async_trait;
-use datafusion::datafusion_data_access::object_store::local::LocalFileSystem;
 use datafusion::datasource::file_format::parquet::ParquetFormat;
 use datafusion::datasource::file_format::FileFormat;
 use datafusion::datasource::listing::PartitionedFile;
+use datafusion::datasource::object_store::ObjectStoreUrl;
 use datafusion::datasource::{TableProvider, TableType};
+use datafusion::execution::context::SessionState;
 use datafusion::logical_plan::Expr;
 use datafusion::physical_plan::file_format::FileScanConfig;
 use datafusion::physical_plan::ExecutionPlan;
@@ -232,6 +233,7 @@ impl TableProvider for delta::DeltaTable {
 
     async fn scan(
         &self,
+        _: &SessionState,
         projection: &Option<Vec<usize>>,
         filters: &[Expr],
         limit: Option<usize>,
@@ -252,11 +254,13 @@ impl TableProvider for delta::DeltaTable {
             })
             .collect::<datafusion::error::Result<_>>()?;
 
-        let df_object_store = Arc::new(LocalFileSystem {});
+        let dt_object_store_url = ObjectStoreUrl::parse(&self.table_uri)
+            .unwrap_or_else(|_| ObjectStoreUrl::local_filesystem());
+
         ParquetFormat::default()
             .create_physical_plan(
                 FileScanConfig {
-                    object_store: df_object_store,
+                    object_store_url: dt_object_store_url,
                     file_schema: schema,
                     file_groups: partitions,
                     statistics: self.datafusion_table_statistics(),