fix: url parsing inconsistencies

Signed-off-by: Robert Pack <[email protected]>

Author: roeap

crates/core/src/logstore/config.rs

@@ -1,7 +1,7 @@
 use std::collections::HashMap;
 
 use ::object_store::RetryConfig;
-use object_store::{path::Path, prefix::PrefixStore, ObjectStore};
+use object_store::{path::Path, prefix::PrefixStore, ObjectStore, ObjectStoreScheme};
 use tokio::runtime::Handle;
 
 #[cfg(feature = "delta-cache")]
@@ -141,18 +141,14 @@ impl StorageConfig {
         Ok(inner)
     }
 
-    fn decorate_prefix<T: ObjectStore>(
+    pub(crate) fn decorate_prefix<T: ObjectStore>(
         store: T,
         table_root: &url::Url,
     ) -> DeltaResult<Box<dyn ObjectStore>> {
-        let prefix = if table_root.scheme() == "file" {
-            Path::from_filesystem_path(
-                table_root
-                    .to_file_path()
-                    .map_err(|_| DeltaTableError::generic("failed to convert fs"))?,
-            )?
-        } else {
-            Path::parse(table_root.path())?
+        let prefix = match ObjectStoreScheme::parse(table_root) {
+            Ok((ObjectStoreScheme::AmazonS3, _)) => Path::parse(table_root.path())?,
+            Ok((_, path)) => path,
+            _ => Path::parse(table_root.path())?,
         };
         Ok(if prefix != Path::from("/") {
             Box::new(PrefixStore::new(store, prefix)) as Box<dyn ObjectStore>

crates/gcp/tests/context.rs

@@ -129,6 +129,7 @@ pub mod gs_cli {
             .wait()
     }
 
+    #[allow(unused)]
     pub fn delete_bucket(container_name: impl AsRef<str>) -> std::io::Result<ExitStatus> {
         let endpoint = std::env::var("GOOGLE_ENDPOINT_URL")
             .expect("variable GOOGLE_ENDPOINT_URL must be set to connect to GCS Emulator");

crates/mount/src/file.rs

@@ -108,42 +108,20 @@ impl From<LocalFileSystemError> for ObjectStoreError {
 #[derive(Debug)]
 pub struct MountFileStorageBackend {
     inner: Arc<LocalFileSystem>,
-    root_url: Arc<Url>,
 }
 
 impl MountFileStorageBackend {
     /// Creates a new MountFileStorageBackend.
-    pub fn try_new(path: impl AsRef<std::path::Path>) -> ObjectStoreResult<Self> {
+    pub fn try_new() -> ObjectStoreResult<Self> {
         Ok(Self {
-            root_url: Arc::new(Self::path_to_root_url(path.as_ref())?),
-            inner: Arc::new(LocalFileSystem::new_with_prefix(path)?),
-        })
-    }
-
-    fn path_to_root_url(path: &std::path::Path) -> ObjectStoreResult<Url> {
-        let root_path =
-            std::fs::canonicalize(path).map_err(|e| object_store::Error::InvalidPath {
-                source: object_store::path::Error::Canonicalize {
-                    path: path.into(),
-                    source: e,
-                },
-            })?;
-
-        Url::from_file_path(root_path).map_err(|_| object_store::Error::InvalidPath {
-            source: object_store::path::Error::InvalidPath { path: path.into() },
+            inner: Arc::new(LocalFileSystem::new()),
         })
     }
 
     /// Return an absolute filesystem path of the given location
     fn path_to_filesystem(&self, location: &ObjectStorePath) -> String {
-        let mut url = self.root_url.as_ref().clone();
-        url.path_segments_mut()
-            .expect("url path")
-            // technically not necessary as Path ignores empty segments
-            // but avoids creating paths with "//" which look odd in error messages.
-            .pop_if_empty()
-            .extend(location.parts());
-
+        let mut url = url::Url::parse("file:///").unwrap();
+        url.set_path(location.as_ref());
         url.to_file_path().unwrap().to_str().unwrap().to_owned()
     }
 }
@@ -264,6 +242,19 @@ impl ObjectStore for MountFileStorageBackend {
     }
 }
 
+fn path_to_root_url(path: &std::path::Path) -> ObjectStoreResult<Url> {
+    let root_path = std::fs::canonicalize(path).map_err(|e| object_store::Error::InvalidPath {
+        source: object_store::path::Error::Canonicalize {
+            path: path.into(),
+            source: e,
+        },
+    })?;
+
+    Url::from_file_path(root_path).map_err(|_| object_store::Error::InvalidPath {
+        source: object_store::path::Error::InvalidPath { path: path.into() },
+    })
+}
+
 /// Regular renames `from` to `to`.
 /// `from` has to exist, but `to` is not, otherwise the operation will fail.
 /// It's not atomic and cannot be called in parallel with other operations on the same file.

crates/mount/src/lib.rs

@@ -60,17 +60,15 @@ impl ObjectStoreFactory for MountFactory {
                 }
                 // We need to convert the dbfs url to a file url
                 let new_url = Url::parse(&format!("file:///dbfs{}", url.path())).unwrap();
-                let store = Arc::new(file::MountFileStorageBackend::try_new(
-                    new_url.to_file_path().unwrap(),
-                )?) as ObjectStoreRef;
+                let store = Arc::new(file::MountFileStorageBackend::try_new()?) as ObjectStoreRef;
                 Ok((store, Path::from("/")))
             }
             "file" => {
                 if allow_unsafe_rename {
-                    let store = Arc::new(file::MountFileStorageBackend::try_new(
-                        url.to_file_path().unwrap(),
-                    )?) as ObjectStoreRef;
-                    Ok((store, Path::from("/")))
+                    let store =
+                        Arc::new(file::MountFileStorageBackend::try_new()?) as ObjectStoreRef;
+                    let prefix = Path::from_filesystem_path(url.to_file_path().unwrap())?;
+                    Ok((store, prefix))
                 } else {
                     let store = Arc::new(LocalFileSystem::new()) as ObjectStoreRef;
                     let prefix = Path::from_filesystem_path(url.to_file_path().unwrap())?;

crates/mount/tests/integration.rs

@@ -11,7 +11,7 @@ static TEST_PREFIXES: &[&str] = &["my table", "你好/😊"];
 
 #[tokio::test]
 #[serial]
-async fn test_integration_local() -> TestResult {
+async fn test_integration_mount() -> TestResult {
     let context = IntegrationContext::new(Box::<MountIntegration>::default())?;
 
     test_read_tables(&context).await?;