chore: pr feedback and test fixes

Signed-off-by: Robert Pack <robstar.pack@gmail.com>

Author: roeap

crates/aws/src/logstore/default_logstore.rs

@@ -31,7 +31,7 @@ pub fn default_s3_logstore(
 /// Default [`LogStore`] implementation
 #[derive(Debug, Clone)]
 pub struct S3LogStore {
-    storage: ObjectStoreRef,
+    prefixed_store: ObjectStoreRef,
     root_store: ObjectStoreRef,
     config: LogStoreConfig,
 }
@@ -41,15 +41,18 @@ impl S3LogStore {
     ///
     /// # Arguments
     ///
-    /// * `storage` - A shared reference to an [`object_store::ObjectStore`] with "/" pointing at delta table root (i.e. where `_delta_log` is located).
+    /// * `prefixed_store` - A shared reference to an [`object_store::ObjectStore`]
+    ///   with "/" pointing at delta table root (i.e. where `_delta_log` is located).
+    /// * `root_store` - A shared reference to an [`object_store::ObjectStore`] with "/"
+    ///   pointing at root of the storage system.
     /// * `location` - A url corresponding to the storage location of `storage`.
     pub fn new(
-        storage: ObjectStoreRef,
+        prefixed_store: ObjectStoreRef,
         root_store: ObjectStoreRef,
         config: LogStoreConfig,
     ) -> Self {
         Self {
-            storage,
+            prefixed_store,
             root_store,
             config,
         }
@@ -120,7 +123,7 @@ impl LogStore for S3LogStore {
     }
 
     fn object_store(&self, _operation_id: Option<Uuid>) -> Arc<dyn ObjectStore> {
-        self.storage.clone()
+        self.prefixed_store.clone()
     }
 
     fn root_object_store(&self, _operation_id: Option<Uuid>) -> Arc<dyn ObjectStore> {

crates/aws/tests/integration_s3_dynamodb.rs

@@ -165,7 +165,8 @@ async fn test_repair_commit_entry() -> TestResult<()> {
         ensure_table_uri(table.table_uri())?,
         &options,
         &S3_OPTIONS,
-        std::sync::Arc::new(table.object_store()),
+        table.log_store().object_store(None),
+        table.log_store().root_object_store(None),
     )?;
 
     // create an incomplete log entry, commit file not yet moved from its temporary location
@@ -240,7 +241,8 @@ async fn test_abort_commit_entry() -> TestResult<()> {
         ensure_table_uri(table.table_uri())?,
         &options,
         &S3_OPTIONS,
-        std::sync::Arc::new(table.object_store()),
+        table.log_store().object_store(None),
+        table.log_store().root_object_store(None),
     )?;
 
     let entry = create_incomplete_commit_entry(&table, 1, "unfinished_commit").await?;
@@ -287,7 +289,8 @@ async fn test_abort_commit_entry_fail_to_delete_entry() -> TestResult<()> {
         ensure_table_uri(table.table_uri())?,
         &options,
         &S3_OPTIONS,
-        std::sync::Arc::new(table.object_store()),
+        table.log_store().object_store(None),
+        table.log_store().root_object_store(None),
     )?;
 
     let entry = create_incomplete_commit_entry(&table, 1, "finished_commit").await?;

crates/lakefs/src/execute.rs

@@ -1,6 +1,8 @@
 use async_trait::async_trait;
 use deltalake_core::{
-    logstore::LogStoreRef, operations::CustomExecuteHandler, DeltaResult, DeltaTableError,
+    logstore::{LogStore as _, LogStoreRef},
+    operations::CustomExecuteHandler,
+    DeltaResult, DeltaTableError,
 };
 use tracing::debug;
 use uuid::Uuid;
@@ -28,7 +30,7 @@ impl CustomExecuteHandler for LakeFSCustomExecuteHandler {
         if let Some(lakefs_store) = log_store.clone().as_any().downcast_ref::<LakeFSLogStore>() {
             let (repo, _, _) = lakefs_store
                 .client
-                .decompose_url(lakefs_store.config.location.to_string());
+                .decompose_url(lakefs_store.config().location.to_string());
             let result = lakefs_store
                 .client
                 .delete_branch(repo, lakefs_store.client.get_transaction(operation_id)?)
@@ -144,7 +146,7 @@ mod tests {
             .downcast_ref::<LakeFSLogStore>()
         {
             assert!(lakefs_store
-                .storage
+                .prefixed_registry
                 .get_store(
                     &Url::parse(format!("lakefs://repo/delta-tx-{operation_id}/table").as_str())
                         .unwrap()
@@ -185,7 +187,7 @@ mod tests {
             .downcast_ref::<LakeFSLogStore>()
         {
             assert!(lakefs_store
-                .storage
+                .prefixed_registry
                 .get_store(
                     &Url::parse(format!("lakefs://repo/delta-tx-{operation_id}/table").as_str())
                         .unwrap()

crates/lakefs/src/logstore.rs

@@ -56,9 +56,9 @@ pub fn lakefs_logstore(
 /// Default [`LogStore`] implementation
 #[derive(Debug, Clone)]
 pub(crate) struct LakeFSLogStore {
-    pub(crate) storage: DefaultObjectStoreRegistry,
+    pub(crate) prefixed_registry: DefaultObjectStoreRegistry,
     root_registry: DefaultObjectStoreRegistry,
-    pub(crate) config: LogStoreConfig,
+    config: LogStoreConfig,
     pub(crate) client: LakeFSClient,
 }
 
@@ -67,20 +67,23 @@ impl LakeFSLogStore {
     ///
     /// # Arguments
     ///
-    /// * `storage` - A shared reference to an [`object_store::ObjectStore`] with "/" pointing at delta table root (i.e. where `_delta_log` is located).
+    /// * `prefixed_store` - A shared reference to an [`object_store::ObjectStore`]
+    ///   with "/" pointing at delta table root (i.e. where `_delta_log` is located).
+    /// * `root_store` - A shared reference to an [`object_store::ObjectStore`] with "/"
+    ///   pointing at root of the storage system.
     /// * `location` - A url corresponding to the storage location of `storage`.
     pub fn new(
-        storage: ObjectStoreRef,
+        prefixed_store: ObjectStoreRef,
         root_store: ObjectStoreRef,
         config: LogStoreConfig,
         client: LakeFSClient,
     ) -> Self {
-        let registry = DefaultObjectStoreRegistry::new();
-        registry.register_store(&config.location, storage);
+        let prefixed_registry = DefaultObjectStoreRegistry::new();
+        prefixed_registry.register_store(&config.location, prefixed_store);
         let root_registry = DefaultObjectStoreRegistry::new();
         root_registry.register_store(&config.location, root_store);
         Self {
-            storage: registry,
+            prefixed_registry,
             root_registry,
             config,
             client,
@@ -112,7 +115,7 @@ impl LakeFSLogStore {
     }
 
     fn register_object_store(&self, url: &Url, store: ObjectStoreRef) {
-        self.storage.register_store(url, store);
+        self.prefixed_registry.register_store(url, store);
     }
 
     fn register_root_object_store(&self, url: &Url, store: ObjectStoreRef) {
@@ -131,7 +134,7 @@ impl LakeFSLogStore {
         let transaction_url = Url::parse(&string_url).unwrap();
         Ok((
             string_url,
-            self.storage.get_store(&transaction_url)?,
+            self.prefixed_registry.get_store(&transaction_url)?,
             self.root_registry.get_store(&transaction_url)?,
         ))
     }
@@ -236,7 +239,11 @@ impl LogStore for LakeFSLogStore {
     }
 
     async fn read_commit_entry(&self, version: i64) -> DeltaResult<Option<Bytes>> {
-        read_commit_entry(&self.storage.get_store(&self.config.location)?, version).await
+        read_commit_entry(
+            &self.prefixed_registry.get_store(&self.config.location)?,
+            version,
+        )
+        .await
     }
 
     /// Tries to commit a prepared commit file. Returns [`TransactionError`]
@@ -355,7 +362,10 @@ impl LogStore for LakeFSLogStore {
                 let (_, store, _) = self.get_transaction_objectstore(id).unwrap_or_else(|_| panic!("The object_store registry inside LakeFSLogstore didn't have a store for operation_id {id} Something went wrong."));
                 store
             }
-            _ => self.storage.get_store(&self.config.location).unwrap(),
+            _ => self
+                .prefixed_registry
+                .get_store(&self.config.location)
+                .unwrap(),
         }
     }