vortex_io/object_store/
filesystem.rs

1// SPDX-License-Identifier: Apache-2.0
2// SPDX-FileCopyrightText: Copyright the Vortex contributors
3
4//! [`FileSystem`] implementation backed by an [`ObjectStore`].
5
6use std::fmt::Debug;
7use std::fmt::Formatter;
8use std::sync::Arc;
9
10use async_trait::async_trait;
11use futures::StreamExt;
12use futures::stream::BoxStream;
13use object_store::ObjectStore;
14use object_store::ObjectStoreExt;
15use object_store::path::Path;
16use vortex_error::VortexResult;
17
18use crate::VortexReadAt;
19use crate::filesystem::FileListing;
20use crate::filesystem::FileSystem;
21use crate::object_store::ObjectStoreReadAt;
22use crate::object_store::object_path_from_literal;
23use crate::runtime::Handle;
24
25/// A [`FileSystem`] backed by an [`ObjectStore`].
26// TODO(ngates): we could consider spawning a driver task inside this file system such that we can
27//  apply concurrency limits to the overall object store, rather than on a per-file basis.
28pub struct ObjectStoreFileSystem {
29    store: Arc<dyn ObjectStore>,
30    handle: Handle,
31}
32
33impl Debug for ObjectStoreFileSystem {
34    fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
35        f.debug_struct("ObjectStoreFileSystem")
36            .field("store", &self.store)
37            .finish()
38    }
39}
40
41impl ObjectStoreFileSystem {
42    /// Create a new filesystem backed by the given object store and runtime handle.
43    pub fn new(store: Arc<dyn ObjectStore>, handle: Handle) -> Self {
44        Self { store, handle }
45    }
46
47    /// Create a new filesystem backed by a local file system object store and the given runtime
48    /// handle.
49    #[deprecated = "use StdFileSystem::new"]
50    pub fn local(handle: Handle) -> Self {
51        Self::new(
52            Arc::new(object_store::local::LocalFileSystem::new()),
53            handle,
54        )
55    }
56}
57
58fn listing_from_meta(location: &Path, size: u64) -> FileListing {
59    FileListing {
60        path: location.to_string(),
61        size: Some(size),
62    }
63}
64
65#[async_trait]
66impl FileSystem for ObjectStoreFileSystem {
67    fn list(&self, prefix: &str) -> BoxStream<'_, VortexResult<FileListing>> {
68        let path = if prefix.is_empty() {
69            None
70        } else {
71            Some(object_path_from_literal(prefix))
72        };
73        self.store
74            .list(path.as_ref())
75            .map(|result| {
76                result
77                    .map(|meta| listing_from_meta(&meta.location, meta.size))
78                    .map_err(Into::into)
79            })
80            .boxed()
81    }
82
83    async fn head(&self, path: &str) -> VortexResult<Option<FileListing>> {
84        // `head` issues a single metadata lookup (e.g. an S3 HEAD) for the exact key, unlike
85        // `list`, which enumerates by path-segment prefix and never returns the key itself.
86        match self.store.head(&object_path_from_literal(path)).await {
87            Ok(meta) => Ok(Some(listing_from_meta(&meta.location, meta.size))),
88            Err(object_store::Error::NotFound { .. }) => Ok(None),
89            Err(e) => Err(e.into()),
90        }
91    }
92
93    async fn open_read(&self, path: &str) -> VortexResult<Arc<dyn VortexReadAt>> {
94        Ok(Arc::new(ObjectStoreReadAt::new(
95            Arc::clone(&self.store),
96            object_path_from_literal(path),
97            self.handle.clone(),
98        )))
99    }
100
101    async fn delete(&self, path: &str) -> VortexResult<()> {
102        self.store.delete(&object_path_from_literal(path)).await?;
103        Ok(())
104    }
105}
106
107// Exercises the fix against a real object store, whose `list` excludes the exact-path match.
108// `Handle::find` only resolves a runtime under the `tokio` feature, so gate these tests on it.
109#[cfg(test)]
110#[cfg(feature = "tokio")]
111mod tests {
112    use futures::TryStreamExt;
113    use object_store::ObjectStoreExt;
114    use object_store::local::LocalFileSystem;
115    use object_store::memory::InMemory;
116    use rstest::rstest;
117
118    use super::*;
119    use crate::filesystem::FileSystem;
120    use crate::runtime::Handle;
121
122    /// Build an [`ObjectStoreFileSystem`] over an in-memory store seeded with `(path, size)` files.
123    ///
124    /// Keys are written with [`object_path_from_literal`] so the store holds the same literal keys
125    /// a real backend would (e.g. `a~b.vortex`, not the percent-encoded `a%7Eb.vortex`).
126    async fn memory_fs(files: &[(&str, usize)]) -> VortexResult<ObjectStoreFileSystem> {
127        let store = Arc::new(InMemory::new()) as Arc<dyn ObjectStore>;
128        for &(path, size) in files {
129            store
130                .put(&object_path_from_literal(path), vec![0u8; size].into())
131                .await?;
132        }
133        let handle = Handle::find().expect("tokio runtime available within #[tokio::test]");
134        Ok(ObjectStoreFileSystem::new(store, handle))
135    }
136
137    /// Regression test for #6599: globbing an exact path that exists must return that one file.
138    /// `ObjectStore::list` never yields the prefix itself, so this would return nothing if the
139    /// exact-path branch used `list`.
140    #[tokio::test]
141    async fn test_glob_exact_existing_path() -> VortexResult<()> {
142        let fs = memory_fs(&[("data/file.vortex", 1024)]).await?;
143        let fs_dyn: &dyn FileSystem = &fs;
144        let results: Vec<FileListing> = fs_dyn.glob("data/file.vortex")?.try_collect().await?;
145        assert_eq!(results.len(), 1);
146        assert_eq!(results[0].path, "data/file.vortex");
147        assert_eq!(results[0].size, Some(1024));
148        Ok(())
149    }
150
151    #[tokio::test]
152    async fn test_glob_exact_missing_path_is_empty() -> VortexResult<()> {
153        let fs = memory_fs(&[("data/other.vortex", 1)]).await?;
154        let fs_dyn: &dyn FileSystem = &fs;
155        let results: Vec<FileListing> = fs_dyn.glob("data/missing.vortex")?.try_collect().await?;
156        assert!(results.is_empty());
157        Ok(())
158    }
159
160    /// `list("foo.vortex")` would surface the prefix-sibling `foo.vortex.backup`; `head` does not.
161    #[tokio::test]
162    async fn test_glob_exact_path_ignores_prefix_siblings() -> VortexResult<()> {
163        let fs = memory_fs(&[("foo.vortex", 10), ("foo.vortex.backup", 20)]).await?;
164        let fs_dyn: &dyn FileSystem = &fs;
165        let results: Vec<FileListing> = fs_dyn.glob("foo.vortex")?.try_collect().await?;
166        assert_eq!(results.len(), 1);
167        assert_eq!(results[0].path, "foo.vortex");
168        assert_eq!(results[0].size, Some(10));
169        Ok(())
170    }
171
172    /// Paths containing characters that `object_store` percent-encodes (`~ % [ ] # { } ^`, …)
173    /// must round-trip on a single literal-path convention: `head` returns the literal key, and
174    /// that exact string must reopen the file via `open_read` (what multi-file scan does).
175    /// Previously these were converted with `Path::from`, which encoded them into a key no real
176    /// object has, so the file was silently lost.
177    #[tokio::test]
178    #[rstest]
179    #[case::tilde("dir/a~b.vortex")]
180    #[case::percent("dir/a%20b.vortex")]
181    #[case::brackets("dir/a[1].vortex")]
182    #[case::hash("dir/a#b.vortex")]
183    #[case::braces("dir/a{x}.vortex")]
184    #[case::caret("dir/a^b.vortex")]
185    #[case::backslash_tilde("dir/a\\~b.vortex")]
186    #[case::space("dir/a b.vortex")]
187    async fn test_head_open_read_round_trip_special_chars(#[case] path: &str) -> VortexResult<()> {
188        let fs = memory_fs(&[(path, 5)]).await?;
189        // `head` returns the literal key, matching the caller's input.
190        assert_eq!(
191            fs.head(path).await?,
192            Some(FileListing {
193                path: path.to_string(),
194                size: Some(5),
195            })
196        );
197        // and that literal path reopens the file (`size()` issues a `head` under the hood).
198        assert_eq!(fs.open_read(path).await?.size().await?, 5);
199        Ok(())
200    }
201
202    /// Glob (both branches) over paths with encoded—but not glob-metacharacter—characters returns
203    /// the literal path, which must reopen the file. (`*`, `?`, `[` are excluded here: they make
204    /// the input a pattern rather than an exact path.)
205    #[tokio::test]
206    #[rstest]
207    #[case::tilde("dir/a~b.vortex")]
208    #[case::percent("dir/a%20b.vortex")]
209    #[case::hash("dir/a#b.vortex")]
210    #[case::backslash_tilde("dir/a\\~b.vortex")]
211    #[case::space("dir/a b.vortex")]
212    #[case::plain("dir/plain.vortex")]
213    async fn test_glob_round_trip_special_chars(#[case] path: &str) -> VortexResult<()> {
214        let fs = memory_fs(&[(path, 5)]).await?;
215        let fs_dyn: &dyn FileSystem = &fs;
216        let expected = FileListing {
217            path: path.to_string(),
218            size: Some(5),
219        };
220
221        // Exact-path glob returns the literal path.
222        let exact: Vec<FileListing> = fs_dyn.glob(path)?.try_collect().await?;
223        assert_eq!(exact, vec![expected.clone()]);
224
225        // Wildcard glob over the directory lists the same literal path (not an encoded one).
226        let wild: Vec<FileListing> = fs_dyn.glob("dir/*.vortex")?.try_collect().await?;
227        assert!(
228            wild.contains(&expected),
229            "wildcard glob should list {path:?}, got {wild:?}"
230        );
231
232        // The returned path reopens the file.
233        assert_eq!(fs.open_read(path).await?.size().await?, 5);
234        Ok(())
235    }
236
237    /// The same round-trip against the real local-filesystem backend: a `~` in an on-disk filename
238    /// must be addressed literally (converting it with `Path::from` would percent-encode it and
239    /// miss the file). Confirms the literal-path convention holds across backends, not just for the
240    /// in-memory store.
241    #[tokio::test]
242    async fn test_local_filesystem_special_char_round_trip() -> anyhow::Result<()> {
243        let dir = tempfile::tempdir()?;
244        std::fs::write(dir.path().join("a~b.vortex"), [0u8; 5])?;
245        let store = Arc::new(LocalFileSystem::new_with_prefix(dir.path())?) as Arc<dyn ObjectStore>;
246        let fs = ObjectStoreFileSystem::new(store, Handle::find().expect("tokio runtime"));
247        let fs_dyn: &dyn FileSystem = &fs;
248        let expected = FileListing {
249            path: "a~b.vortex".to_string(),
250            size: Some(5),
251        };
252
253        let wild: Vec<FileListing> = fs_dyn.glob("*.vortex")?.try_collect().await?;
254        assert!(wild.contains(&expected), "wildcard glob got {wild:?}");
255
256        let exact: Vec<FileListing> = fs_dyn.glob("a~b.vortex")?.try_collect().await?;
257        assert_eq!(exact, vec![expected]);
258
259        assert_eq!(fs.open_read("a~b.vortex").await?.size().await?, 5);
260        Ok(())
261    }
262
263    /// `delete` must address the same literal key as the rest of the trait. It used
264    /// `from_url_path`, which percent-*decodes*: given the real object `a%20b.vortex` it resolved
265    /// to `a b.vortex` and deleted a *different* object that happened to exist.
266    #[tokio::test]
267    async fn test_delete_addresses_the_literal_key() -> VortexResult<()> {
268        let fs = memory_fs(&[("dir/a%20b.vortex", 1), ("dir/a b.vortex", 2)]).await?;
269
270        fs.delete("dir/a%20b.vortex").await?;
271
272        assert_eq!(fs.head("dir/a%20b.vortex").await?, None);
273        assert_eq!(
274            fs.head("dir/a b.vortex").await?,
275            Some(FileListing {
276                path: "dir/a b.vortex".to_string(),
277                size: Some(2),
278            }),
279            "deleting the percent-literal key must not touch its decoded neighbour"
280        );
281        Ok(())
282    }
283
284    /// The keys `head`/`open_read` accept must also be the keys `delete` accepts.
285    #[tokio::test]
286    #[rstest]
287    #[case::tilde("dir/a~b.vortex")]
288    #[case::brackets("dir/a[1].vortex")]
289    #[case::hash("dir/a#b.vortex")]
290    #[case::space("dir/a b.vortex")]
291    #[case::plain("dir/plain.vortex")]
292    async fn test_delete_round_trip_special_chars(#[case] path: &str) -> VortexResult<()> {
293        let fs = memory_fs(&[(path, 5)]).await?;
294        assert!(fs.head(path).await?.is_some());
295        fs.delete(path).await?;
296        assert_eq!(fs.head(path).await?, None);
297        Ok(())
298    }
299
300    #[tokio::test]
301    async fn test_head_existing_and_missing() -> VortexResult<()> {
302        let fs = memory_fs(&[("a/b.vortex", 7)]).await?;
303        assert_eq!(
304            fs.head("a/b.vortex").await?,
305            Some(FileListing {
306                path: "a/b.vortex".to_string(),
307                size: Some(7),
308            })
309        );
310        assert_eq!(fs.head("a/missing.vortex").await?, None);
311        Ok(())
312    }
313}
vortex_io/object_store/filesystem.rs

vortex_io/object_store/
filesystem.rs
