Add FileSystem trait and make DuckDB configurable (#6564)

We switched over to using DuckDB's own file system for I/O. For some
reason we have really binary performance benchmarks, either the same
performance as before, or really really slow. We are going to
investigate, but in the meantime this change abstracts the I/O behind a
file system and allows us to configure whether to use Vortex I/O or
DuckDB I/O. This abstraction was planned and built in #6391 anyway, so
just making use of it a little sooner.

As part of this change, we have a `FileSystem::list -> Stream` API. We
used to use the async-compat crate to provide Tokio support within
DuckDB, but that crate doesn't support streams. This PR inlines the core
piece of logic from that crate and provides wrappers for I/O.

To summarise:

1. Pulls out the FileSystem trait as used in #6391 
1. Uses the trait to make fs configurable in DuckDB. Defaults to Vortex
I/O as per before #5767 , with extension option to enable DuckDB native
file system.
1. Clean up of the vortex-io crate for consistent structure + naming, as
well as providing a "compat" module with compatibility wrappers.

---------

Signed-off-by: Nicholas Gates <nick@nickgates.com>

Author: gatesn

Cargo.lock

@@ -96,7 +96,6 @@ arrow-ord = "57.1"
 arrow-schema = "57.1"
 arrow-select = "57.1"
 arrow-string = "57.1"
-async-compat = "0.2.5"
 async-fs = "2.2.0"
 async-lock = "3.4"
 async-stream = "0.3.6"

Cargo.toml

@@ -79,7 +79,7 @@ impl DuckClient {
         }?;
 
         let connection = db.connect()?;
-        vortex_duckdb::register_table_functions(&connection)?;
+        vortex_duckdb::initialize(&db)?;
 
         // Install and load httpfs so DuckDB can access remote files (S3, GCS, HTTP).
         connection.query("INSTALL httpfs;")?;

benchmarks/duckdb-bench/src/lib.rs

@@ -40,6 +40,26 @@ without it, DuckDB defaults to CSV.
 COPY (SELECT * FROM my_table) TO 'output.vortex' (FORMAT vortex);
 ```
 
+## Extension Options
+
+### `vortex_filesystem`
+
+Controls which filesystem implementation is used for reading and writing Vortex files.
+
+| Value | Description |
+|-------|-------------|
+| `'vortex'` (default) | Uses Vortex's built-in object store filesystem. Supports `file://` and `s3://` schemes. |
+| `'duckdb'` | Uses DuckDB's built-in filesystem, including any filesystem extensions such as `httpfs`. |
+
+```sql
+SET vortex_filesystem = 'duckdb';
+```
+
+Use `'duckdb'` when you want to leverage DuckDB's filesystem extensions (e.g., `httpfs` for HTTP
+or S3 access with DuckDB's credential management). Use `'vortex'` (the default) for direct
+object store access via Vortex's own S3 integration, which reads credentials from environment
+variables.
+
 ## Python
 
 The DuckDB Python client works with `read_vortex` the same way:

docs/user-guide/duckdb.md

@@ -70,8 +70,8 @@ mod common_tests {
     use vortex::array::ArrayRef;
     use vortex::array::arrow::FromArrowArray;
     use vortex::file::WriteOptionsSessionExt;
-    use vortex::io::ObjectStoreWriter;
     use vortex::io::VortexWrite;
+    use vortex::io::object_store::ObjectStoreWrite;
     use vortex::session::VortexSession;
 
     use crate::VortexFormatFactory;
@@ -123,7 +123,7 @@ mod common_tests {
             P: Into<object_store::path::Path>,
         {
             let array = ArrayRef::from_arrow(batch, false)?;
-            let mut write = ObjectStoreWriter::new(self.store.clone(), &path.into()).await?;
+            let mut write = ObjectStoreWrite::new(self.store.clone(), &path.into()).await?;
             VX_SESSION
                 .write_options()
                 .write(&mut write, array.to_array_stream())

vortex-datafusion/src/lib.rs

@@ -55,7 +55,7 @@ use vortex::file::EOF_SIZE;
 use vortex::file::MAX_POSTSCRIPT_SIZE;
 use vortex::file::OpenOptionsSessionExt;
 use vortex::file::VORTEX_FILE_EXTENSION;
-use vortex::io::file::object_store::ObjectStoreSource;
+use vortex::io::object_store::ObjectStoreReadAt;
 use vortex::io::session::RuntimeSessionExt;
 use vortex::scalar::Scalar;
 use vortex::session::VortexSession;
@@ -261,7 +261,7 @@ impl FileFormat for VortexFormat {
                     }
 
                     // Not cached or invalid - open the file
-                    let reader = Arc::new(ObjectStoreSource::new(
+                    let reader = Arc::new(ObjectStoreReadAt::new(
                         store,
                         object.location.clone(),
                         session.handle(),
@@ -328,7 +328,7 @@ impl FileFormat for VortexFormat {
                 Some(metadata) => metadata,
                 None => {
                     // Not cached - open the file
-                    let reader = Arc::new(ObjectStoreSource::new(
+                    let reader = Arc::new(ObjectStoreReadAt::new(
                         store,
                         object.location.clone(),
                         session.handle(),

vortex-datafusion/src/persistent/format.rs

@@ -33,8 +33,8 @@ mod tests {
     use vortex::array::validity::Validity;
     use vortex::buffer::buffer;
     use vortex::file::WriteOptionsSessionExt;
-    use vortex::io::ObjectStoreWriter;
     use vortex::io::VortexWrite;
+    use vortex::io::object_store::ObjectStoreWrite;
     use vortex::session::VortexSession;
 
     use crate::common_tests::TestSessionContext;
@@ -65,7 +65,7 @@ mod tests {
             Validity::NonNullable,
         )?;
 
-        let mut writer = ObjectStoreWriter::new(ctx.store.clone(), &"test.vortex".into()).await?;
+        let mut writer = ObjectStoreWrite::new(ctx.store.clone(), &"test.vortex".into()).await?;
 
         let summary = session
             .write_options()

vortex-datafusion/src/persistent/mod.rs

@@ -477,8 +477,8 @@ mod tests {
     use vortex::array::arrow::FromArrowArray;
     use vortex::buffer::Buffer;
     use vortex::file::WriteOptionsSessionExt;
-    use vortex::io::ObjectStoreWriter;
     use vortex::io::VortexWrite;
+    use vortex::io::object_store::ObjectStoreWrite;
     use vortex::metrics::DefaultMetricsRegistry;
     use vortex::scan::Selection;
     use vortex::session::VortexSession;
@@ -540,7 +540,7 @@ mod tests {
         let array = ArrayRef::from_arrow(rb, false)?;
         let path = Path::parse(path)?;
 
-        let mut write = ObjectStoreWriter::new(object_store, &path).await?;
+        let mut write = ObjectStoreWrite::new(object_store, &path).await?;
         let summary = SESSION
             .write_options()
             .write(&mut write, array.to_array_stream())

vortex-datafusion/src/persistent/opener.rs

@@ -7,7 +7,7 @@ use std::sync::Arc;
 use datafusion_common::Result as DFResult;
 use object_store::ObjectStore;
 use vortex::io::VortexReadAt;
-use vortex::io::file::object_store::ObjectStoreSource;
+use vortex::io::object_store::ObjectStoreReadAt;
 use vortex::io::session::RuntimeSessionExt;
 use vortex::session::VortexSession;
 
@@ -38,7 +38,7 @@ impl VortexReaderFactory for DefaultVortexReaderFactory {
         path: &str,
         session: &VortexSession,
     ) -> DFResult<Arc<dyn VortexReadAt>> {
-        Ok(Arc::new(ObjectStoreSource::new(
+        Ok(Arc::new(ObjectStoreReadAt::new(
             self.object_store.clone(),
             path.into(),
             session.handle(),

vortex-datafusion/src/persistent/reader.rs

@@ -32,8 +32,8 @@ use vortex::dtype::DType;
 use vortex::dtype::arrow::FromArrowType;
 use vortex::file::WriteOptionsSessionExt;
 use vortex::file::WriteSummary;
-use vortex::io::ObjectStoreWriter;
 use vortex::io::VortexWrite;
+use vortex::io::object_store::ObjectStoreWrite;
 use vortex::session::VortexSession;
 
 pub struct VortexSink {
@@ -124,9 +124,9 @@ impl FileSink for VortexSink {
 
                 let stream_adapter = ArrayStreamAdapter::new(dtype, stream);
 
-                let mut object_writer = ObjectStoreWriter::new(object_store, &path)
+                let mut object_writer = ObjectStoreWrite::new(object_store, &path)
                     .await
-                    .map_err(|e| exec_datafusion_err!("Failed to create ObjectStoreWriter: {e}"))?;
+                    .map_err(|e| exec_datafusion_err!("Failed to create ObjectStoreWrite: {e}"))?;
 
                 let summary = session
                     .write_options()