Don't create local path when opening a repo (#2145)

Fix #2105 

- `get_client` in `ObjectStorage` returns an `StorageResult`, so we can
track errors during client instantiation.
- `create_location_if_needed` in the `Storage` trait. This is mostly
aimed at storages that are cheap/easy to create, like local FS.

Add tests for both Rust and Python.

Author: li-em

icechunk-arrow-object-store/src/lib.rs

@@ -48,7 +48,6 @@ use serde::{Deserialize, Serialize};
 use std::{
     collections::HashMap,
     fmt::{self, Debug, Display},
-    fs::create_dir_all,
     future::ready,
     num::{NonZeroU16, NonZeroU64},
     ops::Range,
@@ -264,15 +263,12 @@ impl ObjectStorage {
     /// client is not serializeable and must be initialized after deserialization. Under normal construction
     /// the original client is returned immediately.
     #[instrument(skip_all)]
-    async fn get_client(&self, settings: &Settings) -> &Arc<dyn ObjectStore> {
+    async fn get_client(
+        &self,
+        settings: &Settings,
+    ) -> StorageResult<&Arc<dyn ObjectStore>> {
         self.client
-            .get_or_init(|| async {
-                // TODO: handle error better?
-                #[expect(clippy::expect_used)]
-                self.backend
-                    .mk_object_store(settings)
-                    .expect("failed to create object store")
-            })
+            .get_or_try_init(|| async { self.backend.mk_object_store(settings) })
             .await
     }
 
@@ -287,7 +283,7 @@ impl ObjectStorage {
     /// Intended for testing and debugging purposes only.
     pub async fn all_keys(&self) -> StorageResult<Vec<String>> {
         self.get_client(&self.backend.default_settings())
-            .await
+            .await?
             .list(None)
             .map_ok(|obj| obj.location.to_string())
             .try_collect()
@@ -362,6 +358,10 @@ impl Storage for ObjectStorage {
         Ok(self.backend.can_write())
     }
 
+    async fn create_location_if_needed(&self) -> StorageResult<()> {
+        self.backend.create_location_if_needed()
+    }
+
     #[instrument(skip_all)]
     async fn default_settings(&self) -> StorageResult<Settings> {
         Ok(self.backend.default_settings())
@@ -394,7 +394,7 @@ impl Storage for ObjectStorage {
         let options = PutOptions { mode, attributes, ..PutOptions::default() };
         // FIXME: use multipart
         let res =
-            self.get_client(settings).await.put_opts(&path, bytes.into(), options).await;
+            self.get_client(settings).await?.put_opts(&path, bytes.into(), options).await;
         match res {
             Ok(res) => {
                 let new_version = VersionInfo {
@@ -431,7 +431,7 @@ impl Storage for ObjectStorage {
                 if_match: version.etag().map(|e| strip_quotes(e).into()),
                 ..Default::default()
             };
-            let result = self.get_client(settings).await.get_opts(&from, opts).await;
+            let result = self.get_client(settings).await?.get_opts(&from, opts).await;
             match result {
                 Ok(result) => {
                     let bytes = result
@@ -440,7 +440,7 @@ impl Storage for ObjectStorage {
                         .map_err(|e| StorageErrorKind::ObjectStore(Box::new(e)))
                         .capture()?;
                     self.get_client(settings)
-                        .await
+                        .await?
                         .put(&to, bytes.into())
                         .await
                         .map_err(|e| StorageErrorKind::ObjectStore(Box::new(e)))
@@ -454,7 +454,7 @@ impl Storage for ObjectStorage {
                 Err(err) => Err(obj_store_error(err)),
             }
         } else {
-            match self.get_client(settings).await.copy(&from, &to).await {
+            match self.get_client(settings).await?.copy(&from, &to).await {
                 Ok(_) => {
                     Ok(VersionedUpdateResult::Updated { new_version: version.clone() })
                 }
@@ -472,7 +472,7 @@ impl Storage for ObjectStorage {
     ) -> StorageResult<BoxStream<'a, StorageResult<ListInfo<String>>>> {
         let prefix = ObjectPath::from(format!("{}/{}", self.backend.prefix(), prefix));
         let stream =
-            self.get_client(settings).await.list(Some(&prefix)).map(move |object| {
+            self.get_client(settings).await?.list(Some(&prefix)).map(move |object| {
                 let prefix = prefix.clone();
                 object
                     .map_err(obj_store_error)
@@ -496,7 +496,7 @@ impl Storage for ObjectStorage {
             sizes.insert(path, size);
         }
         let results =
-            self.get_client(settings).await.delete_stream(stream::iter(ids).boxed());
+            self.get_client(settings).await?.delete_stream(stream::iter(ids).boxed());
         let res = results
             .fold(DeleteObjectsResult::default(), |mut res, delete_result| {
                 if let Ok(deleted_path) = delete_result {
@@ -525,7 +525,7 @@ impl Storage for ObjectStorage {
         let path = self.prefixed_path(path);
         let res = self
             .get_client(settings)
-            .await
+            .await?
             .head(&path)
             .await
             .map_err(Box::new)
@@ -598,7 +598,7 @@ impl ObjectStorage {
                 .and_then(|v| v.etag().map(|e| strip_quotes(e).into())),
             ..Default::default()
         };
-        let res = self.get_client(settings).await.get_opts(&full_key, opts).await;
+        let res = self.get_client(settings).await?.get_opts(&full_key, opts).await;
 
         match res {
             Ok(result) => {
@@ -640,6 +640,10 @@ pub trait ObjectStoreBackend: Debug + Display + Sync + Send {
     fn can_write(&self) -> bool {
         true
     }
+
+    fn create_location_if_needed(&self) -> Result<(), StorageError> {
+        Ok(())
+    }
 }
 
 #[derive(Debug, Serialize, Deserialize)]
@@ -717,8 +721,13 @@ impl ObjectStoreBackend for LocalFileSystemObjectStoreBackend {
         &self,
         _settings: &Settings,
     ) -> Result<Arc<dyn ObjectStore>, StorageError> {
-        create_dir_all(&self.path).capture()?;
-        let path = std::fs::canonicalize(&self.path).capture()?;
+        let path = std::fs::canonicalize(&self.path).map_err(|err| {
+            if err.kind() == std::io::ErrorKind::NotFound {
+                StorageError::capture(StorageErrorKind::ObjectNotFound)
+            } else {
+                StorageError::capture(StorageErrorKind::IOError(err))
+            }
+        })?;
         let fs = LocalFileSystem::new_with_prefix(path).capture_box()?;
         Ok(Arc::new(fs))
     }
@@ -751,6 +760,11 @@ impl ObjectStoreBackend for LocalFileSystemObjectStoreBackend {
             ..Default::default()
         }
     }
+
+    fn create_location_if_needed(&self) -> Result<(), StorageError> {
+        std::fs::create_dir_all(&self.path).capture()?;
+        Ok(())
+    }
 }
 
 #[cfg(feature = "http")]

icechunk-python/tests/test_session.py

@@ -2,6 +2,7 @@
 import tempfile
 import time
 from datetime import UTC, datetime
+from pathlib import Path
 from typing import Any, cast
 
 import pytest
@@ -504,6 +505,19 @@ async def test_repo_status_readonly_blocks_rearrange_session_async() -> None:
     await repo.garbage_collect_async(datetime.now(UTC))
 
 
+def test_open_doesnt_create_dir() -> None:
+    with tempfile.TemporaryDirectory() as tmpdir:
+        missing = Path(tmpdir) / "should_not_exist"
+        present = Path(tmpdir) / "present"
+
+        with pytest.raises(IcechunkError, match=r"the repository doesn.t exist"):
+            Repository.open(local_filesystem_storage(str(missing)))
+        assert not missing.exists()
+
+        Repository.create(local_filesystem_storage(str(present)))
+        assert present.exists()
+
+
 async def test_repo_status_readonly_blocks_writable_session_async() -> None:
     repo = await Repository.create_async(
         storage=in_memory_storage(),

icechunk-storage/src/storage.rs

@@ -464,6 +464,17 @@ pub trait Storage: fmt::Debug + Display + sealed::Sealed + Sync + Send {
 
     async fn can_write(&self) -> StorageResult<bool>;
 
+    /// Ensure the storage location is ready to receive writes.
+    ///
+    /// Called by [`Repository::create`] before any other I/O. The default
+    /// implementation is a no-op; backends that need to materialize the
+    /// location (e.g. the local filesystem creating the directory) override
+    /// this. Read paths like `open` never call it, so a missing location
+    /// stays missing.
+    async fn create_location_if_needed(&self) -> StorageResult<()> {
+        Ok(())
+    }
+
     async fn get_object(
         &self,
         settings: &Settings,
@@ -581,10 +592,14 @@ pub trait Storage: fmt::Debug + Display + sealed::Sealed + Sync + Send {
     }
 
     async fn root_is_clean(&self, settings: &Settings) -> StorageResult<bool> {
-        match self.list_objects(settings, "").await?.next().await {
-            None => Ok(true),
-            Some(Ok(_)) => Ok(false),
-            Some(Err(err)) => Err(err),
+        match self.list_objects(settings, "").await {
+            Ok(mut stream) => match stream.next().await {
+                None => Ok(true),
+                Some(Ok(_)) => Ok(false),
+                Some(Err(err)) => Err(err),
+            },
+            Err(StorageError { kind: StorageErrorKind::ObjectNotFound, .. }) => Ok(true),
+            Err(err) => Err(err),
         }
     }
 

icechunk/src/repository.rs

@@ -210,6 +210,7 @@ impl Repository {
     ) -> RepositoryResult<Self> {
         debug!("Creating Repository");
         raise_if_cant_write(storage.as_ref(), "Cannot create repository").await?;
+        storage.create_location_if_needed().await.inject()?;
 
         let has_overriden_config = match config {
             Some(ref config) => config != &RepositoryConfig::default(),
@@ -3908,6 +3909,38 @@ mod tests {
         Ok(())
     }
 
+    #[cfg(feature = "object-store-fs")]
+    #[tokio::test]
+    async fn open_doesnt_create_dir() -> Result<(), Box<dyn Error>> {
+        use crate::new_local_filesystem_storage;
+
+        let repo_dir = TempDir::new()?;
+        let missing = repo_dir.path().join("should_not_exist");
+        let present = repo_dir.path().join("present");
+
+        let storage: Arc<dyn Storage + Send + Sync> =
+            new_local_filesystem_storage(&missing)
+                .await
+                .expect("Creating local storage failed");
+
+        let open_err = Repository::open(None, Arc::clone(&storage), HashMap::new()).await;
+        assert!(matches!(
+            open_err,
+            Err(RepositoryError { kind: RepositoryErrorKind::RepositoryDoesntExist, .. })
+        ));
+        assert!(!missing.exists());
+
+        let storage: Arc<dyn Storage + Send + Sync> =
+            new_local_filesystem_storage(&present)
+                .await
+                .expect("Creating local storage failed");
+        Repository::create(None, Arc::clone(&storage), HashMap::new(), None, true)
+            .await?;
+        assert!(present.exists());
+
+        Ok(())
+    }
+
     #[tokio::test]
     async fn set_metadata() -> Result<(), Box<dyn Error>> {
         let storage: Arc<dyn Storage + Send + Sync> = new_in_memory_storage().await?;