bugfix: localfs

Author: parmesant

src/metastore/metastores/object_store_metastore.rs

@@ -733,7 +733,7 @@ impl Metastore for ObjectStoreMetastore {
 
     async fn list_streams(&self) -> Result<HashSet<String>, MetastoreError> {
         // using LocalFS list_streams because it doesn't implement list_with_delimiter
-        if PARSEABLE.get_storage_mode_string() == "drive" {
+        if PARSEABLE.storage.name() == "drive" {
             PARSEABLE
                 .storage
                 .get_object_store()

src/query/mod.rs

@@ -579,13 +579,15 @@ pub async fn get_manifest_list(
                 stream_name,
                 manifest_item.time_lower_bound,
                 manifest_item.time_upper_bound,
-                Some(manifest_item.manifest_path),
+                Some(manifest_item.manifest_path.clone()),
             )
             .await?;
         let manifest = manifest_opt.ok_or_else(|| {
             QueryError::CustomError(format!(
-                "Manifest not found for {stream_name} [{} - {}]",
-                manifest_item.time_lower_bound, manifest_item.time_upper_bound
+                "Manifest not found for {stream_name} [{} - {}], path- {}",
+                manifest_item.time_lower_bound,
+                manifest_item.time_upper_bound,
+                manifest_item.manifest_path
             ))
         })?;
         all_manifest_files.push(manifest);

src/storage/localfs.rs

@@ -139,7 +139,19 @@ impl ObjectStorage for LocalFS {
     }
     async fn get_object(&self, path: &RelativePath) -> Result<Bytes, ObjectStorageError> {
         let time = Instant::now();
-        let file_path = self.path_in_root(path);
+        let file_path = if path.to_string().contains(&self.root.to_str().unwrap()[1..]) {
+            #[cfg(windows)]
+            {
+                path.to_path("")
+            }
+            #[cfg(not(windows))]
+            {
+                path.to_path("/")
+            }
+        } else {
+            self.path_in_root(path)
+        };
+
         let res: Result<Bytes, ObjectStorageError> = match fs::read(file_path).await {
             Ok(x) => Ok(x.into()),
             Err(e) => match e.kind() {