feat: upgrade to arrow15 and datafusion9

rust/Cargo.toml

@@ -58,8 +58,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 }
 
@@ -71,7 +71,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,11 +12,11 @@ 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;
 use std::convert::TryFrom;
+use std::io::Cursor;
 use std::iter::Iterator;
 use std::ops::Add;
 
@@ -400,8 +400,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 writeable_cursor = Cursor::new(vec![]);
+    let mut writer = ArrowWriter::try_new(&mut writeable_cursor, 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)? {
@@ -410,7 +410,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(writeable_cursor.into_inner())
 }
 
 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(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(),

rust/src/writer/json.rs

@@ -16,14 +16,13 @@ use arrow::{
 };
 use log::{info, warn};
 use parquet::{
-    arrow::ArrowWriter,
-    basic::Compression,
-    errors::ParquetError,
-    file::{properties::WriterProperties, writer::InMemoryWriteableCursor},
+    arrow::ArrowWriter, basic::Compression, errors::ParquetError,
+    file::properties::WriterProperties,
 };
 use serde_json::Value;
 use std::collections::HashMap;
 use std::convert::TryFrom;
+use std::io::Cursor;
 use std::sync::Arc;
 
 type BadValue = (Value, ParquetError);
@@ -42,8 +41,8 @@ pub struct JsonWriter {
 pub(crate) struct DataArrowWriter {
     arrow_schema: Arc<ArrowSchema>,
     writer_properties: WriterProperties,
-    cursor: InMemoryWriteableCursor,
-    arrow_writer: ArrowWriter<InMemoryWriteableCursor>,
+    cursor: Cursor<Vec<u8>>,
+    arrow_writer: ArrowWriter<Cursor<Vec<u8>>>,
     partition_values: HashMap<String, Option<String>>,
     null_counts: NullCounts,
     buffered_record_batch_count: usize,
@@ -115,7 +114,7 @@ impl DataArrowWriter {
         }
 
         // Copy current cursor bytes so we can recover from failures
-        let current_cursor_bytes = self.cursor.data();
+        let current_cursor_bytes = self.cursor.clone().into_inner();
 
         let result = self.arrow_writer.write(&record_batch);
 
@@ -146,7 +145,7 @@ impl DataArrowWriter {
         arrow_schema: Arc<ArrowSchema>,
         writer_properties: WriterProperties,
     ) -> Result<Self, ParquetError> {
-        let cursor = InMemoryWriteableCursor::default();
+        let cursor = Cursor::new(vec![]);
         let arrow_writer = Self::new_underlying_writer(
             cursor.clone(),
             arrow_schema.clone(),
@@ -169,10 +168,10 @@ impl DataArrowWriter {
     }
 
     fn new_underlying_writer(
-        cursor: InMemoryWriteableCursor,
+        cursor: Cursor<Vec<u8>>,
         arrow_schema: Arc<ArrowSchema>,
         writer_properties: WriterProperties,
-    ) -> Result<ArrowWriter<InMemoryWriteableCursor>, ParquetError> {
+    ) -> Result<ArrowWriter<Cursor<Vec<u8>>>, ParquetError> {
         ArrowWriter::try_new(cursor, arrow_schema, Some(writer_properties))
     }
 }
@@ -259,7 +258,10 @@ impl JsonWriter {
     /// Returns the current byte length of the in memory buffer.
     /// This may be used by the caller to decide when to finalize the file write.
     pub fn buffer_len(&self) -> usize {
-        self.arrow_writers.values().map(|w| w.cursor.len()).sum()
+        self.arrow_writers
+            .values()
+            .map(|w| w.cursor.get_ref().len())
+            .sum()
     }
 
     /// Returns the number of records held in the current buffer.
@@ -370,7 +372,7 @@ impl DeltaWriter<Vec<Value>> for JsonWriter {
 
             let path = next_data_path(&self.partition_columns, &writer.partition_values, None)?;
 
-            let obj_bytes = writer.cursor.data();
+            let obj_bytes = writer.cursor.into_inner();
             let file_size = obj_bytes.len() as i64;
 
             let storage_path = self.storage.join_path(&self.table_uri, path.as_str());
@@ -420,8 +422,7 @@ fn quarantine_failed_parquet_rows(
 
     for value in values {
         let record_batch = record_batch_from_message(arrow_schema.clone(), &[value.clone()])?;
-
-        let cursor = InMemoryWriteableCursor::default();
+        let cursor = Cursor::new(vec![]);
         let mut writer = ArrowWriter::try_new(cursor.clone(), arrow_schema.clone(), None)?;
 
         match writer.write(&record_batch) {