Fix file operations on deltalake with object_store

Author: skejserjensen

crates/modelardb_server/src/storage/data_transfer.rs

@@ -24,7 +24,6 @@ use dashmap::DashMap;
 use deltalake::arrow::array::RecordBatch;
 use futures::TryStreamExt;
 use modelardb_storage::data_folder::DataFolder;
-use object_store::path::Path;
 use tokio::sync::RwLock;
 use tokio::task::JoinHandle as TaskJoinHandle;
 use tracing::debug;
@@ -73,8 +72,7 @@ impl DataTransfer {
             let mut table_size_in_bytes = table_size_in_bytes.entry(table_name).or_insert(0);
 
             let object_store = delta_table.object_store();
-            for file_uri in delta_table.get_file_uris()? {
-                let file_path = Path::parse(file_uri)?;
+            for file_path in delta_table.get_files_by_partitions(&[]).await? {
                 let object_meta = object_store.head(&file_path).await?;
                 *table_size_in_bytes += object_meta.size;
             }
@@ -276,10 +274,11 @@ impl DataTransfer {
 mod tests {
     use super::*;
 
+    use deltalake::ObjectStore;
     use modelardb_test::table::{self, NORMAL_TABLE_NAME, TIME_SERIES_TABLE_NAME};
     use tempfile::{self, TempDir};
 
-    const EXPECTED_TIME_SERIES_TABLE_FILE_SIZE: u64 = 2038;
+    const EXPECTED_TIME_SERIES_TABLE_FILE_SIZE: u64 = 2128;
 
     // Tests for data transfer component.
     #[tokio::test]
@@ -532,8 +531,7 @@ mod tests {
         let delta_table = local_data_folder.delta_table(table_name).await.unwrap();
 
         let mut files_size = 0;
-        for file_uri in delta_table.get_file_uris().unwrap() {
-            let file_path = Path::parse(file_uri).unwrap();
+        for file_path in delta_table.get_files_by_partitions(&[]).await.unwrap() {
             let object_meta = delta_table.object_store().head(&file_path).await;
             files_size += object_meta.unwrap().size;
         }

crates/modelardb_storage/src/data_folder.rs

@@ -35,7 +35,6 @@ use datafusion::parquet::file::properties::WriterProperties;
 use datafusion::parquet::format::SortingColumn;
 use datafusion::prelude::{SessionContext, col};
 use datafusion_proto::bytes::Serializeable;
-use deltalake::table::builder::ensure_table_uri;
 use delta_kernel::engine::arrow_conversion::TryIntoKernel;
 use delta_kernel::table_properties::DataSkippingNumIndexedCols;
 use deltalake::delta_datafusion::DeltaDataChecker;
@@ -423,7 +422,7 @@ impl DataFolder {
             Ok(delta_table.clone())
         } else {
             // If the table is not in the cache, open it and add it to the cache before returning.
-            let table_url = ensure_table_uri(table_path)?;
+            let table_url = deltalake::ensure_table_uri(table_path)?;
             let delta_table = deltalake::open_table_with_storage_options(
                 table_url,
                 self.storage_options.clone(),