vortex_io/std_file/
filesystem.rs

1// SPDX-License-Identifier: Apache-2.0
2// SPDX-FileCopyrightText: Copyright the Vortex contributors
3
4use std::fmt::Debug;
5use std::fmt::Formatter;
6use std::fs;
7use std::io;
8use std::path::Path;
9use std::path::PathBuf;
10use std::sync::Arc;
11
12use async_trait::async_trait;
13use futures::StreamExt;
14use futures::stream;
15use futures::stream::BoxStream;
16use vortex_error::VortexResult;
17
18use crate::VortexReadAt;
19use crate::filesystem::FileListing;
20use crate::filesystem::FileSystem;
21use crate::runtime::Handle;
22use crate::std_file::FileReadAt;
23
24/// A FileSystem over local filesystem.
25pub struct StdFileSystem {
26    handle: Handle,
27}
28
29impl Debug for StdFileSystem {
30    fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
31        f.debug_struct("StdFileSystem").finish()
32    }
33}
34
35impl StdFileSystem {
36    pub fn new(handle: Handle) -> Self {
37        Self { handle }
38    }
39}
40
41fn walk(dir: &Path, out: &mut Vec<FileListing>) -> io::Result<()> {
42    for entry in fs::read_dir(dir)? {
43        let entry = entry?;
44        let file_type = entry.file_type()?;
45        if file_type.is_dir() {
46            walk(&entry.path(), out)?;
47        } else {
48            let metadata = entry.metadata()?;
49            out.push(FileListing {
50                path: entry.path().to_string_lossy().into_owned(),
51                size: Some(metadata.len()),
52            });
53        }
54    }
55    Ok(())
56}
57
58#[async_trait]
59impl FileSystem for StdFileSystem {
60    fn list(&self, prefix: &str) -> BoxStream<'_, VortexResult<FileListing>> {
61        let dir = PathBuf::from(prefix);
62        let listing = self.handle.spawn_blocking(move || {
63            let mut out = Vec::new();
64            walk(&dir, &mut out)?;
65            Ok::<_, io::Error>(out)
66        });
67        stream::once(listing)
68            .flat_map(|result| match result {
69                Ok(listings) => stream::iter(listings.into_iter().map(Ok)).boxed(),
70                Err(e) => stream::once(async move { Err(e.into()) }).boxed(),
71            })
72            .boxed()
73    }
74
75    async fn head(&self, path: &str) -> VortexResult<Option<FileListing>> {
76        let path = path.to_owned();
77        self.handle
78            .spawn_blocking(move || match fs::metadata(&path) {
79                Ok(metadata) if metadata.is_file() => Ok(Some(FileListing {
80                    path,
81                    size: Some(metadata.len()),
82                })),
83                Ok(_) => Ok(None),
84                Err(e) if e.kind() == io::ErrorKind::NotFound => Ok(None),
85                Err(e) => Err(e.into()),
86            })
87            .await
88    }
89
90    async fn open_read(&self, path: &str) -> VortexResult<Arc<dyn VortexReadAt>> {
91        let handle = self.handle.clone();
92        let path = path.to_owned();
93        let reader = self
94            .handle
95            .spawn_blocking(move || FileReadAt::open(path, handle))
96            .await?;
97        Ok(Arc::new(reader))
98    }
99
100    async fn delete(&self, path: &str) -> VortexResult<()> {
101        let path = path.to_owned();
102        self.handle
103            .spawn_blocking(move || fs::remove_file(path))
104            .await?;
105        Ok(())
106    }
107}
vortex_io/std_file/filesystem.rs

vortex_io/std_file/
filesystem.rs
