From ecbd0d3f1c22808e9d9547030d9fb49af8d969c7 Mon Sep 17 00:00:00 2001 From: Mikhail Kot Date: Fri, 4 Sep 2026 17:25:20 +0100 Subject: [PATCH] FileSystem for local reads Signed-off-by: Mikhail Kot --- vortex-duckdb/src/file_reader.rs | 3 +- vortex-io/src/std_file/filesystem.rs | 107 +++++++++++++++++++++++++++ vortex-io/src/std_file/mod.rs | 2 + 3 files changed, 111 insertions(+), 1 deletion(-) create mode 100644 vortex-io/src/std_file/filesystem.rs diff --git a/vortex-duckdb/src/file_reader.rs b/vortex-duckdb/src/file_reader.rs index 5048aa8a025..f4c9bb67995 100644 --- a/vortex-duckdb/src/file_reader.rs +++ b/vortex-duckdb/src/file_reader.rs @@ -23,6 +23,7 @@ use vortex::io::compat::Compat; use vortex::io::filesystem::FileSystemRef; use vortex::io::object_store::ObjectStoreFileSystem; use vortex::io::runtime::BlockingRuntime as _; +use vortex::io::std_file::StdFileSystem; use vortex::layout::LayoutReaderRef; use vortex::layout::scan::scan_builder::ScanBuilder; use vortex::mask::Mask; @@ -80,7 +81,7 @@ fn resolve_filesystem(url: &Url) -> VortexResult<(FileSystemRef, String)> { // high-core machines because reads go into blocking pool if url.scheme() == "file" { return Ok(( - Arc::new(ObjectStoreFileSystem::local(RUNTIME.handle())), + Arc::new(StdFileSystem::new(RUNTIME.handle())), url.path().to_string(), )); } diff --git a/vortex-io/src/std_file/filesystem.rs b/vortex-io/src/std_file/filesystem.rs new file mode 100644 index 00000000000..9c71be47516 --- /dev/null +++ b/vortex-io/src/std_file/filesystem.rs @@ -0,0 +1,107 @@ +// SPDX-License-Identifier: Apache-2.0 +// SPDX-FileCopyrightText: Copyright the Vortex contributors + +use std::fmt::Debug; +use std::fmt::Formatter; +use std::fs; +use std::io; +use std::path::Path; +use std::path::PathBuf; +use std::sync::Arc; + +use async_trait::async_trait; +use futures::StreamExt; +use futures::stream; +use futures::stream::BoxStream; +use vortex_error::VortexResult; + +use crate::VortexReadAt; +use crate::filesystem::FileListing; +use crate::filesystem::FileSystem; +use crate::runtime::Handle; +use crate::std_file::FileReadAt; + +/// A FileSystem over local filesystem. +pub struct StdFileSystem { + handle: Handle, +} + +impl Debug for StdFileSystem { + fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result { + f.debug_struct("StdFileSystem").finish() + } +} + +impl StdFileSystem { + pub fn new(handle: Handle) -> Self { + Self { handle } + } +} + +fn walk(dir: &Path, out: &mut Vec) -> io::Result<()> { + for entry in fs::read_dir(dir)? { + let entry = entry?; + let file_type = entry.file_type()?; + if file_type.is_dir() { + walk(&entry.path(), out)?; + } else { + let metadata = entry.metadata()?; + out.push(FileListing { + path: entry.path().to_string_lossy().into_owned(), + size: Some(metadata.len()), + }); + } + } + Ok(()) +} + +#[async_trait] +impl FileSystem for StdFileSystem { + fn list(&self, prefix: &str) -> BoxStream<'_, VortexResult> { + let dir = PathBuf::from(prefix); + let listing = self.handle.spawn_blocking(move || { + let mut out = Vec::new(); + walk(&dir, &mut out)?; + Ok::<_, io::Error>(out) + }); + stream::once(listing) + .flat_map(|result| match result { + Ok(listings) => stream::iter(listings.into_iter().map(Ok)).boxed(), + Err(e) => stream::once(async move { Err(e.into()) }).boxed(), + }) + .boxed() + } + + async fn head(&self, path: &str) -> VortexResult> { + let path = path.to_owned(); + self.handle + .spawn_blocking(move || match fs::metadata(&path) { + Ok(metadata) if metadata.is_file() => Ok(Some(FileListing { + path, + size: Some(metadata.len()), + })), + Ok(_) => Ok(None), + Err(e) if e.kind() == io::ErrorKind::NotFound => Ok(None), + Err(e) => Err(e.into()), + }) + .await + } + + async fn open_read(&self, path: &str) -> VortexResult> { + let handle = self.handle.clone(); + let path = path.to_owned(); + let reader = self + .handle + .spawn_blocking(move || FileReadAt::open(path, handle)) + .await?; + Ok(Arc::new(reader)) + } + + async fn delete(&self, path: &str) -> VortexResult<()> { + let path = path.to_owned(); + self.handle + .spawn_blocking(move || fs::remove_file(path)) + .await?; + Ok(()) + } +} diff --git a/vortex-io/src/std_file/mod.rs b/vortex-io/src/std_file/mod.rs index c30248ed496..3345089a023 100644 --- a/vortex-io/src/std_file/mod.rs +++ b/vortex-io/src/std_file/mod.rs @@ -1,8 +1,10 @@ // SPDX-License-Identifier: Apache-2.0 // SPDX-FileCopyrightText: Copyright the Vortex contributors +mod filesystem; mod read_at; mod write; +pub use filesystem::*; pub use read_at::*; pub use write::*;