feat[cuda]: add direct I/O for local file scans

Cargo.toml

@@ -230,6 +230,7 @@ roaring = "0.11.0"
 rstest = "0.26.1"
 rstest_reuse = "0.7.0"
 rustc-hash = "2.1.1"
+rustix = { version = "1.1", features = ["fs"] }
 serde = "1.0.220"
 serde_json = "1.0.138"
 serde_test = "1.0.176"

vortex-cuda/Cargo.toml

@@ -44,6 +44,9 @@ vortex-cuda-macros = { workspace = true }
 vortex-error = { workspace = true, features = ["object_store"] }
 vortex-nvcomp = { path = "nvcomp" }
 
+[target.'cfg(target_os = "linux")'.dependencies]
+rustix = { workspace = true }
+
 [dev-dependencies]
 criterion = { workspace = true }
 futures = { workspace = true, features = ["executor"] }

vortex-cuda/ffi/README.md

@@ -23,3 +23,7 @@ Push host-resident arrays and close the sink using the standard `vx_array_sink_*
 Use `vx_cuda_scan_path_arrow_device_stream` to read such a local file through pinned host buffers
 and receive an Arrow C Device stream. Reuse the same CUDA session across scans so the pinned buffer
 pool and CUDA state are reused as well.
+
+On Linux, use `vx_cuda_scan_path_arrow_device_stream_with_options` with
+`vx_cuda_scan_options.flags = VX_CUDA_SCAN_FLAG_DIRECT_IO` to bypass the operating system page
+cache for pooled data-plane reads. Footer and zone-map reads remain buffered on the host.

vortex-cuda/ffi/cinclude/vortex_cuda.h

@@ -94,6 +94,22 @@ vx_array_sink *vx_cuda_array_sink_open_file_block_rows(const vx_session *session
                                                        size_t block_rows,
                                                        vx_error **error_out);
 
+/**
+ * Options for scanning a CUDA-compatible Vortex file.
+ *
+ * Zero-initialize this struct to use buffered file I/O and layout-derived batch splitting.
+ */
+/** Bypass the operating system page cache for pooled data-plane reads.
+ * Footer and zone-map reads remain buffered. Supported only on Linux. */
+#define VX_CUDA_SCAN_FLAG_DIRECT_IO (UINT32_C(1) << 0)
+
+typedef struct vx_cuda_scan_options {
+    /** Bitwise combination of `VX_CUDA_SCAN_FLAG_*` values. */
+    uint32_t flags;
+    /** Number of rows in each output ArrowDeviceArray. Zero uses layout-derived splitting. */
+    size_t batch_rows;
+} vx_cuda_scan_options;
+
 /**
  * Scan a local CUDA-compatible Vortex file as an Arrow C Device stream.
  *
@@ -125,6 +141,19 @@ int vx_cuda_scan_path_arrow_device_stream_batch_rows(const vx_session *session,
                                                      struct ArrowDeviceArrayStream *out_stream,
                                                      vx_error **error_out);
 
+/**
+ * Scan a local CUDA-compatible Vortex file with explicit options.
+ *
+ * This has the same ownership and file compatibility requirements as
+ * `vx_cuda_scan_path_arrow_device_stream`. Pass NULL or a zero-initialized options struct to use
+ * buffered file I/O and layout-derived batch splitting.
+ */
+int vx_cuda_scan_path_arrow_device_stream_with_options(const vx_session *session,
+                                                       vx_view path,
+                                                       const vx_cuda_scan_options *options,
+                                                       struct ArrowDeviceArrayStream *out_stream,
+                                                       vx_error **error_out);
+
 /**
  * Export a borrowed Vortex array for cuDF's Arrow Device import path.
  *

vortex-cuda/ffi/src/lib.rs

@@ -25,6 +25,7 @@ use vortex::session::SessionExt;
 use vortex::session::VortexSession;
 use vortex_cuda::CudaOpenOptionsExt;
 use vortex_cuda::CudaSession;
+use vortex_cuda::PooledFileReadAtOptions;
 use vortex_cuda::arrow::ArrowDeviceArray;
 use vortex_cuda::arrow::ArrowDeviceArrayStream;
 use vortex_cuda::arrow::DeviceArrayExt;
@@ -49,6 +50,22 @@ use vortex_ffi::vx_view;
 const VX_CUDA_OK: c_int = 0;
 const VX_CUDA_ERR: c_int = 1;
 
+/// Enable direct I/O for pooled CUDA file reads.
+pub const VX_CUDA_SCAN_FLAG_DIRECT_IO: u32 = 1 << 0;
+const VX_CUDA_SCAN_KNOWN_FLAGS: u32 = VX_CUDA_SCAN_FLAG_DIRECT_IO;
+
+/// Options for scanning a CUDA-compatible Vortex file.
+///
+/// Zero-initialize this struct to use buffered file I/O and layout-derived batch splitting.
+#[repr(C)]
+#[derive(Default)]
+pub struct vx_cuda_scan_options {
+    /// A bitwise combination of `VX_CUDA_SCAN_FLAG_*` values.
+    pub flags: u32,
+    /// Number of rows in each output batch. Zero uses layout-derived splitting.
+    pub batch_rows: usize,
+}
+
 /// Return a Vortex session with a [`CudaSession`] session variable.
 ///
 /// If `session` already has CUDA support, this returns a clone of it. Otherwise it
@@ -142,7 +159,7 @@ pub unsafe extern "C-unwind" fn vx_cuda_array_sink_open_file_block_rows(
     })
 }
 
-/// Scan a local Vortex file and export an Arrow C Device stream.
+/// Scan a local Vortex file with buffered I/O and export an Arrow C Device stream.
 ///
 /// Footer and zone-map reads remain on the host. Data segments are staged through pinned host
 /// buffers and transferred directly to the GPU.
@@ -171,7 +188,13 @@ pub unsafe extern "C-unwind" fn vx_cuda_scan_path_arrow_device_stream(
     error_out: *mut *mut vx_error,
 ) -> c_int {
     unsafe {
-        vx_cuda_scan_path_arrow_device_stream_batch_rows(session, path, 0, out_stream, error_out)
+        vx_cuda_scan_path_arrow_device_stream_with_options(
+            session,
+            path,
+            ptr::null(),
+            out_stream,
+            error_out,
+        )
     }
 }
 
@@ -195,19 +218,60 @@ pub unsafe extern "C-unwind" fn vx_cuda_scan_path_arrow_device_stream_batch_rows
     batch_rows: usize,
     out_stream: *mut ArrowDeviceArrayStream,
     error_out: *mut *mut vx_error,
+) -> c_int {
+    let options = vx_cuda_scan_options {
+        batch_rows,
+        ..Default::default()
+    };
+    unsafe {
+        vx_cuda_scan_path_arrow_device_stream_with_options(
+            session,
+            path,
+            &raw const options,
+            out_stream,
+            error_out,
+        )
+    }
+}
+
+/// Scan a local Vortex file with explicit options and export an Arrow C Device stream.
+///
+/// This has the same ownership and file compatibility requirements as
+/// [`vx_cuda_scan_path_arrow_device_stream`]. Pass a null `options` pointer or a zero-initialized
+/// [`vx_cuda_scan_options`] to use buffered file I/O and layout-derived batch splitting.
+///
+/// # Safety
+///
+/// `session` must be a valid borrowed handle created by `vortex-ffi`. `path` must be valid for the
+/// duration of this call and contain UTF-8. `options`, when non-null, must point to a valid
+/// [`vx_cuda_scan_options`]. `out_stream` must be a valid writable pointer. If `error_out` is
+/// non-null, it must be valid for writing one error pointer.
+#[unsafe(no_mangle)]
+pub unsafe extern "C-unwind" fn vx_cuda_scan_path_arrow_device_stream_with_options(
+    session: *const vx_session,
+    path: vx_view,
+    options: *const vx_cuda_scan_options,
+    out_stream: *mut ArrowDeviceArrayStream,
+    error_out: *mut *mut vx_error,
 ) -> c_int {
     try_or(error_out, VX_CUDA_ERR, || {
         vortex_ensure!(!out_stream.is_null(), "null ArrowDeviceArrayStream output");
 
         let path = unsafe { path.as_str() }?.to_owned();
         let session = session_with_cuda(unsafe { vx_session_ref(session) }?)?;
+        let options = unsafe { scan_options(options) }?;
         let array_stream = ffi_runtime().block_on(async {
-            let file = session.open_options().with_cuda().open_path(path).await?;
+            let file = session
+                .open_options()
+                .with_cuda()
+                .with_read_at_options(options.read_at_options)
+                .open_path(path)
+                .await?;
             let scan = file.scan()?;
-            let scan = if batch_rows == 0 {
+            let scan = if options.batch_rows == 0 {
                 scan
             } else {
-                scan.with_split_by(SplitBy::RowCount(batch_rows))
+                scan.with_split_by(SplitBy::RowCount(options.batch_rows))
             };
             Ok::<_, vortex::error::VortexError>(scan.into_array_stream()?.boxed())
         })?;
@@ -218,6 +282,46 @@ pub unsafe extern "C-unwind" fn vx_cuda_scan_path_arrow_device_stream_batch_rows
     })
 }
 
+struct CudaScanOptions {
+    read_at_options: PooledFileReadAtOptions,
+    batch_rows: usize,
+}
+
+unsafe fn scan_options(options: *const vx_cuda_scan_options) -> VortexResult<CudaScanOptions> {
+    let (flags, batch_rows) = if options.is_null() {
+        (0, 0)
+    } else {
+        let options = unsafe { &*options };
+        (options.flags, options.batch_rows)
+    };
+    vortex_ensure!(
+        flags & !VX_CUDA_SCAN_KNOWN_FLAGS == 0,
+        "unsupported CUDA scan option flags: {:#x}",
+        flags & !VX_CUDA_SCAN_KNOWN_FLAGS
+    );
+
+    let read_at_options = PooledFileReadAtOptions::default();
+    let read_at_options = if flags & VX_CUDA_SCAN_FLAG_DIRECT_IO == 0 {
+        read_at_options
+    } else {
+        #[cfg(target_os = "linux")]
+        {
+            read_at_options.with_direct_io()
+        }
+        #[cfg(not(target_os = "linux"))]
+        {
+            return Err(vortex::error::vortex_err!(
+                "direct CUDA file I/O is only supported on Linux"
+            ));
+        }
+    };
+
+    Ok(CudaScanOptions {
+        read_at_options,
+        batch_rows,
+    })
+}
+
 /// Export a borrowed Vortex array for cuDF's Arrow Device import path.
 ///
 /// On success returns `0` and writes independently releasable `out_schema` and `out_array`; the
@@ -321,6 +425,49 @@ mod tests {
 
     use super::*;
 
+    #[test]
+    fn scan_options_default_to_buffered_io() {
+        let options = vx_cuda_scan_options::default();
+        assert_eq!(options.flags, 0);
+        assert_eq!(options.batch_rows, 0);
+    }
+
+    #[test]
+    fn rejects_unknown_scan_option_flags() {
+        let options = vx_cuda_scan_options {
+            flags: 1 << 31,
+            ..Default::default()
+        };
+        assert!(unsafe { scan_options(&raw const options) }.is_err());
+    }
+
+    #[cfg(target_os = "linux")]
+    #[test]
+    fn maps_direct_io_scan_option_to_pooled_reader() -> VortexResult<()> {
+        let options = vx_cuda_scan_options {
+            flags: VX_CUDA_SCAN_FLAG_DIRECT_IO,
+            ..Default::default()
+        };
+        assert_eq!(
+            unsafe { scan_options(&raw const options) }?.read_at_options,
+            PooledFileReadAtOptions::default().with_direct_io()
+        );
+        Ok(())
+    }
+
+    #[test]
+    fn maps_batch_rows_scan_option() -> VortexResult<()> {
+        let options = vx_cuda_scan_options {
+            batch_rows: 8192,
+            ..Default::default()
+        };
+        assert_eq!(
+            unsafe { scan_options(&raw const options) }?.batch_rows,
+            8192
+        );
+        Ok(())
+    }
+
     fn test_session(session: VortexSession) -> *mut vx_session {
         Box::into_raw(Box::new(session)).cast::<vx_session>()
     }

vortex-cuda/src/file.rs

@@ -20,6 +20,7 @@ use vortex::layout::segments::SegmentSource;
 
 use crate::CudaSessionExt;
 use crate::PooledFileReadAt;
+use crate::PooledFileReadAtOptions;
 use crate::layout::register_cuda_layout;
 
 /// Extension trait for opening CUDA-readable files from [`VortexOpenOptions`].
@@ -31,7 +32,10 @@ pub trait CudaOpenOptionsExt {
 
 impl CudaOpenOptionsExt for VortexOpenOptions {
     fn with_cuda(self) -> CudaOpenOptions {
-        CudaOpenOptions { inner: self }
+        CudaOpenOptions {
+            inner: self,
+            read_at_options: PooledFileReadAtOptions::default(),
+        }
     }
 }
 
@@ -41,9 +45,18 @@ impl CudaOpenOptionsExt for VortexOpenOptions {
 /// configured before calling `with_cuda`.
 pub struct CudaOpenOptions {
     inner: VortexOpenOptions,
+    read_at_options: PooledFileReadAtOptions,
 }
 
 impl CudaOpenOptions {
+    /// Configure how the pooled data-plane reader opens and reads the local file.
+    ///
+    /// Footer and zone-map reads continue to use the standard buffered host reader.
+    pub fn with_read_at_options(mut self, options: PooledFileReadAtOptions) -> Self {
+        self.read_at_options = options;
+        self
+    }
+
     /// Open a local Vortex file for CUDA execution.
     ///
     /// The footer and zone-map segments are read through the ordinary host path. All other file
@@ -65,11 +78,12 @@ impl CudaOpenOptions {
         let pool = Arc::clone(cuda_session.pinned_buffer_pool());
         drop(cuda_session);
 
-        let reader = Arc::new(PooledFileReadAt::open(
+        let reader = Arc::new(PooledFileReadAt::open_with_options(
             &path,
             session.handle(),
             pool,
             stream,
+            self.read_at_options,
         )?);
         let data_file = data_options
             .with_footer(footer.clone())

vortex-cuda/src/lib.rs

@@ -62,6 +62,7 @@ pub use pinned::PinnedPoolStats;
 pub use pinned::PooledPinnedBuffer;
 pub use pooled_read_at::PooledByteBufferReadAt;
 pub use pooled_read_at::PooledFileReadAt;
+pub use pooled_read_at::PooledFileReadAtOptions;
 pub use pooled_read_at::PooledObjectStoreReadAt;
 pub use session::CudaSession;
 pub use session::CudaSessionExt;

vortex-cuda/src/pinned.rs

@@ -309,6 +309,19 @@ impl PooledPinnedBuffer {
             .unwrap_or_else(|e| vortex_panic!("failed to access pinned host buffer: {e}"))
     }
 
+    /// Shortens the logical length without reallocating the pinned buffer.
+    ///
+    /// This is an O(1) metadata update; the allocation capacity is unchanged.
+    #[cfg(target_os = "linux")]
+    pub(crate) fn truncate(&mut self, len: usize) {
+        let inner = self
+            .inner
+            .as_mut()
+            .unwrap_or_else(|| vortex_panic!("buffer already consumed"));
+        assert!(len <= inner.len());
+        inner.set_logical_len(len);
+    }
+
     /// Submits a non-blocking H2D DMA transfer and returns a device buffer.
     ///
     /// The pinned buffer is placed in the pool's inflight queue, gated on a `CudaEvent` marking