From bd550994924844ae8b5681ba15ad70142020ce1b Mon Sep 17 00:00:00 2001
From: Daniel King <dan@spiraldb.com>
Date: Fri, 7 Aug 2026 16:16:19 -0400
Subject: [PATCH 1/7] Add local read-at support for file opens

---
 vortex-file/src/open.rs            | 99 +++++++++++++++++++++++++++++-
 vortex-file/src/segments/source.rs | 93 +++++++++++++++++++++++++++-
 vortex-io/src/read_at.rs           | 32 ++++++++++
 vortex-io/src/runtime/handle.rs    | 27 ++++++++
 vortex-io/src/runtime/mod.rs       | 12 ++++
 vortex-io/src/runtime/single.rs    | 42 +++++++++++++
 6 files changed, 303 insertions(+), 2 deletions(-)

diff --git a/vortex-file/src/open.rs b/vortex-file/src/open.rs
index 1bd42c7be41..63ca5163bfa 100644
--- a/vortex-file/src/open.rs
+++ b/vortex-file/src/open.rs
@@ -13,8 +13,8 @@ use vortex_buffer::ByteBuffer;
 use vortex_error::VortexError;
 use vortex_error::VortexExpect;
 use vortex_error::VortexResult;
-use vortex_io::VortexReadAt;
 use vortex_io::session::RuntimeSessionExt;
+use vortex_io::{VortexLocalReadAt, VortexReadAt};
 use vortex_layout::segments::InstrumentedSegmentCache;
 use vortex_layout::segments::NoOpSegmentCache;
 use vortex_layout::segments::SegmentCache;
@@ -283,6 +283,59 @@ impl VortexOpenOptions {
         })
     }
 
+    /// Open a [`VortexFile`] using a reader whose I/O futures must stay local to one runtime
+    /// thread.
+    pub async fn open_read_local<R: VortexLocalReadAt + Clone>(
+        self,
+        reader: R,
+    ) -> VortexResult<VortexFile> {
+        let segment_cache = self
+            .segment_cache
+            .clone()
+            .unwrap_or_else(|| Arc::new(NoOpSegmentCache));
+
+        let metrics_registry = self
+            .metrics_registry
+            .clone()
+            .unwrap_or_else(|| Arc::new(DefaultMetricsRegistry::default()));
+
+        let footer = if let Some(footer) = self.footer {
+            footer
+        } else {
+            self.read_footer_local(&reader).await?
+        };
+
+        let segment_cache = Arc::new(InstrumentedSegmentCache::new(
+            InitialReadSegmentCache {
+                initial: self.initial_read_segments,
+                fallback: segment_cache,
+            },
+            metrics_registry.as_ref(),
+            self.labels.clone(),
+        ));
+
+        let metrics = RequestMetrics::new(metrics_registry.as_ref(), self.labels);
+
+        let segment_source = Arc::new(SharedSegmentSource::new(FileSegmentSource::open_local(
+            Arc::clone(footer.segment_map()),
+            reader,
+            self.session.handle(),
+            metrics,
+        )));
+
+        let segment_source = Arc::new(SegmentCacheSourceAdapter::new(
+            segment_cache,
+            segment_source,
+        ));
+
+        let file = VortexFile::new(footer, segment_source, self.session.clone());
+        Ok(if self.cache_layout_reader {
+            file.with_caching()
+        } else {
+            file
+        })
+    }
+
     async fn read_footer(&self, read: &dyn VortexReadAt) -> VortexResult<Footer> {
         // Fetch the file size and perform the initial read.
         let file_size = match self.file_size {
@@ -331,6 +384,50 @@ impl VortexOpenOptions {
         Ok(footer)
     }
 
+    async fn read_footer_local(&self, read: &dyn VortexLocalReadAt) -> VortexResult<Footer> {
+        let file_size = match self.file_size {
+            None => read.size().await?,
+            Some(file_size) => file_size,
+        };
+        let mut initial_read_size = self
+            .initial_read_size
+            .max(MAX_POSTSCRIPT_SIZE as usize + EOF_SIZE);
+        if let Ok(file_size) = usize::try_from(file_size) {
+            initial_read_size = initial_read_size.min(file_size);
+        }
+
+        let initial_offset = file_size - initial_read_size as u64;
+        let initial_read: ByteBuffer = read
+            .read_at_local(initial_offset, initial_read_size, Alignment::none())
+            .await?
+            .try_into_host()?
+            .await?;
+
+        let mut deserializer = Footer::deserializer(initial_read, self.session.clone())
+            .with_size(file_size)
+            .with_some_dtype(self.dtype.clone());
+
+        let footer = loop {
+            match deserializer.deserialize()? {
+                DeserializeStep::NeedMoreData { offset, len } => {
+                    let more_data = read
+                        .read_at_local(offset, len, Alignment::none())
+                        .await?
+                        .try_into_host()?
+                        .await?;
+                    deserializer.prefix_data(more_data);
+                }
+                DeserializeStep::NeedFileSize => unreachable!("We passed file_size above"),
+                DeserializeStep::Done(footer) => break Ok::<_, VortexError>(footer),
+            }
+        }?;
+
+        let initial_offset = file_size - (deserializer.buffer().len() as u64);
+        self.populate_initial_segments(initial_offset, deserializer.buffer(), &footer);
+
+        Ok(footer)
+    }
+
     /// Populate segments in the cache that were covered by the initial read.
     fn populate_initial_segments(
         &self,
diff --git a/vortex-file/src/segments/source.rs b/vortex-file/src/segments/source.rs
index 3af33362b05..185419432a1 100644
--- a/vortex-file/src/segments/source.rs
+++ b/vortex-file/src/segments/source.rs
@@ -24,9 +24,9 @@ use vortex_error::VortexResult;
 use vortex_error::vortex_bail;
 use vortex_error::vortex_err;
 use vortex_error::vortex_panic;
-use vortex_io::VortexReadAt;
 use vortex_io::runtime::Handle;
 use vortex_io::runtime::JoinOutcome;
+use vortex_io::{VortexLocalReadAt, VortexReadAt};
 use vortex_layout::segments::SegmentFuture;
 use vortex_layout::segments::SegmentId;
 use vortex_layout::segments::SegmentSource;
@@ -193,6 +193,97 @@ impl FileSegmentSource {
             next_id: Arc::new(AtomicUsize::new(0)),
         }
     }
+
+    /// Open a file-backed segment source over a reader whose read futures are local to one I/O
+    /// runtime thread.
+    ///
+    /// Segment request futures remain `Send`; the local read futures are constructed and polled
+    /// inside one runtime-local driver spawned through [`Handle::spawn_local_io`].
+    pub fn open_local<R: VortexLocalReadAt + Clone>(
+        segments: Arc<[SegmentSpec]>,
+        reader: R,
+        handle: Handle,
+        metrics: RequestMetrics,
+    ) -> Self {
+        let (send, recv) = mpsc::unbounded();
+
+        let max_alignment = segments
+            .iter()
+            .map(|segment| segment.alignment)
+            .max()
+            .unwrap_or_else(Alignment::none);
+        let coalesce_config = reader.coalesce_config().map(|mut config| {
+            let extra = (*max_alignment as u64).saturating_sub(1);
+            config.max_size = config.max_size.saturating_add(extra);
+            config
+        });
+        let concurrency = reader.concurrency();
+        if concurrency == 0 {
+            vortex_panic!(
+                "VortexLocalReadAt::concurrency returned 0 (uri={:?}); this would stall I/O",
+                reader.uri()
+            );
+        }
+
+        let stream = IoRequestStream::new(
+            StreamExt::boxed(recv),
+            coalesce_config,
+            max_alignment,
+            metrics,
+        )
+        .boxed();
+
+        let mut task = handle.spawn_local_io(move || {
+            async move {
+                stream
+                    .map(move |req| {
+                        let reader = reader.clone();
+                        async move {
+                            let result = reader
+                                .read_at_local(req.offset(), req.len(), req.alignment())
+                                .await;
+                            let result = result.and_then(|buffer| {
+                                if req.len() != buffer.len() {
+                                    vortex_bail!(
+                                        "FileSegmentSource: expected buffer of length {} but received {}. {:?}",
+                                        req.len(),
+                                        buffer.len(),
+                                        req
+                                    )
+                                }
+                                Ok(buffer)
+                            });
+
+                            req.resolve(result);
+                        }
+                    })
+                    .buffer_unordered(concurrency)
+                    .collect::<()>()
+                    .await
+            }
+            .boxed_local()
+        });
+        let driver_panic: DriverPanic = Arc::new(Mutex::new(None));
+        let driver = {
+            let driver_panic = Arc::clone(&driver_panic);
+            async move {
+                if let JoinOutcome::Panicked(panic) = future::poll_fn(|cx| task.poll_join(cx)).await
+                {
+                    *driver_panic.lock() = Some(panic);
+                }
+            }
+            .boxed()
+            .shared()
+        };
+
+        Self {
+            segments,
+            events: send,
+            driver,
+            driver_panic,
+            next_id: Arc::new(AtomicUsize::new(0)),
+        }
+    }
 }
 
 impl SegmentSource for FileSegmentSource {
diff --git a/vortex-io/src/read_at.rs b/vortex-io/src/read_at.rs
index aa9a8a03abf..ee6ec949be1 100644
--- a/vortex-io/src/read_at.rs
+++ b/vortex-io/src/read_at.rs
@@ -5,6 +5,7 @@ use std::sync::Arc;
 
 use futures::FutureExt;
 use futures::future::BoxFuture;
+use futures::future::LocalBoxFuture;
 use vortex_array::buffer::BufferHandle;
 use vortex_buffer::Alignment;
 use vortex_buffer::ByteBuffer;
@@ -91,6 +92,37 @@ pub trait VortexReadAt: Send + Sync + 'static {
     ) -> BoxFuture<'static, VortexResult<BufferHandle>>;
 }
 
+/// A random-access source whose read futures are local to the I/O thread that polls them.
+///
+/// The reader value itself remains `Send + Sync` so it can be assigned to an I/O worker, but each
+/// concrete read future may borrow runtime-local state and does not need to implement `Send`.
+pub trait VortexLocalReadAt: Send + Sync + 'static {
+    /// URI for debugging/logging. Returns `None` for anonymous sources.
+    fn uri(&self) -> Option<&Arc<str>> {
+        None
+    }
+
+    /// Configuration for merging nearby I/O requests into fewer, larger reads.
+    fn coalesce_config(&self) -> Option<CoalesceConfig> {
+        None
+    }
+
+    /// Maximum number of concurrent local I/O requests for this source.
+    fn concurrency(&self) -> usize;
+
+    /// Asynchronously get the number of bytes of the underlying source.
+    fn size(&self) -> BoxFuture<'static, VortexResult<u64>>;
+
+    /// Request an asynchronous positional read whose future must be polled on the local I/O
+    /// runtime thread that owns the source.
+    fn read_at_local(
+        &self,
+        offset: u64,
+        length: usize,
+        alignment: Alignment,
+    ) -> LocalBoxFuture<'static, VortexResult<BufferHandle>>;
+}
+
 impl VortexReadAt for Arc<dyn VortexReadAt> {
     fn uri(&self) -> Option<&Arc<str>> {
         self.as_ref().uri()
diff --git a/vortex-io/src/runtime/handle.rs b/vortex-io/src/runtime/handle.rs
index bebbaac5f97..de9a4ec60ac 100644
--- a/vortex-io/src/runtime/handle.rs
+++ b/vortex-io/src/runtime/handle.rs
@@ -135,6 +135,33 @@ impl Handle {
         }
     }
 
+    /// Spawn an I/O future that must be constructed and polled on one runtime thread.
+    ///
+    /// This is intended for runtimes such as io_uring where request futures may carry
+    /// thread-local state and therefore cannot implement `Send`.
+    pub fn spawn_local_io<F, Fut, R>(&self, f: F) -> Task<R>
+    where
+        F: FnOnce() -> Fut + Send + 'static,
+        Fut: Future<Output = R> + 'static,
+        R: Send + 'static,
+    {
+        let (send, recv) = oneshot::channel();
+        let span = tracing::trace_span!(target: "vortex_io::spawn_local_io", "spawn_local_io");
+        let abort_handle = self.runtime().spawn_local_io(Box::new(move || {
+            async move {
+                // See `spawn`: catch a panic so it re-raises on the joining side.
+                let output = AssertUnwindSafe(f()).catch_unwind().await;
+                drop(send.send(output));
+            }
+            .instrument(span)
+            .boxed_local()
+        }));
+        Task {
+            recv: recv.into_future(),
+            abort_handle: Some(abort_handle),
+        }
+    }
+
     /// Spawn a CPU-bound task for execution on the runtime.
     ///
     /// Note that many runtimes will interleave this work on the same async runtime. See the
diff --git a/vortex-io/src/runtime/mod.rs b/vortex-io/src/runtime/mod.rs
index 864d3c8b2c6..e6bb1216a4c 100644
--- a/vortex-io/src/runtime/mod.rs
+++ b/vortex-io/src/runtime/mod.rs
@@ -13,6 +13,7 @@
 //! * Tokio: work is driven on a Tokio runtime provided by the caller.
 
 use futures::future::BoxFuture;
+use futures::future::LocalBoxFuture;
 
 mod blocking;
 pub use blocking::*;
@@ -52,6 +53,17 @@ pub trait Executor: Send + Sync {
         self.spawn(fut)
     }
 
+    /// Spawns a future doing I/O that must be constructed and polled on one runtime thread.
+    ///
+    /// The factory is `Send` so runtimes may route it to an I/O worker before constructing the
+    /// returned `!Send` future.
+    fn spawn_local_io(
+        &self,
+        _future: Box<dyn FnOnce() -> LocalBoxFuture<'static, ()> + Send + 'static>,
+    ) -> AbortHandleRef {
+        vortex_error::vortex_panic!("runtime does not support local I/O futures")
+    }
+
     /// Spawns a CPU-bound task for execution on the runtime.
     ///
     /// The returned `AbortHandle` may be used to optimistically cancel the task if it has not
diff --git a/vortex-io/src/runtime/single.rs b/vortex-io/src/runtime/single.rs
index 2a113e3715e..c5eefa77c27 100644
--- a/vortex-io/src/runtime/single.rs
+++ b/vortex-io/src/runtime/single.rs
@@ -8,6 +8,7 @@ use std::sync::Arc;
 use futures::Stream;
 use futures::StreamExt;
 use futures::future::BoxFuture;
+use futures::future::LocalBoxFuture;
 use futures::stream::LocalBoxStream;
 use parking_lot::Mutex;
 use smol::LocalExecutor;
@@ -39,6 +40,7 @@ impl Default for SingleThreadRuntime {
 
 struct Sender {
     scheduling: kanal::Sender<SpawnAsync<'static>>,
+    local_io: kanal::Sender<SpawnLocalAsync<'static>>,
     cpu: kanal::Sender<SpawnSync<'static>>,
     blocking: kanal::Sender<SpawnSync<'static>>,
 }
@@ -46,6 +48,7 @@ struct Sender {
 impl Sender {
     fn new(local: &Rc<LocalExecutor<'static>>) -> Self {
         let (scheduling_send, scheduling_recv) = kanal::unbounded::<SpawnAsync>();
+        let (local_io_send, local_io_recv) = kanal::unbounded::<SpawnLocalAsync>();
         let (cpu_send, cpu_recv) = kanal::unbounded::<SpawnSync>();
         let (blocking_send, blocking_recv) = kanal::unbounded::<SpawnSync>();
 
@@ -70,6 +73,23 @@ impl Sender {
             })
             .detach();
 
+        // Drive local I/O tasks. The future is constructed on this thread so it may capture
+        // runtime-local state and does not need to be Send.
+        let weak_local2 = RcWeak::clone(&weak_local);
+        local
+            .spawn(async move {
+                while let Ok(spawn) = local_io_recv.as_async().recv().await {
+                    if let Some(local) = weak_local2.upgrade() {
+                        drop(
+                            spawn
+                                .task_callback
+                                .send(SmolAbortHandle::new_handle(local.spawn((spawn.future)()))),
+                        );
+                    }
+                }
+            })
+            .detach();
+
         // Drive CPU tasks.
         let weak_local2 = RcWeak::clone(&weak_local);
         local
@@ -104,6 +124,7 @@ impl Sender {
 
         Self {
             scheduling: scheduling_send,
+            local_io: local_io_send,
             cpu: cpu_send,
             blocking: blocking_send,
         }
@@ -128,6 +149,22 @@ impl Executor for Sender {
         })
     }
 
+    fn spawn_local_io(
+        &self,
+        future: Box<dyn FnOnce() -> LocalBoxFuture<'static, ()> + Send + 'static>,
+    ) -> AbortHandleRef {
+        let (send, recv) = oneshot::channel();
+        if let Err(e) = self.local_io.send(SpawnLocalAsync {
+            future,
+            task_callback: send,
+        }) {
+            vortex_panic!("Executor missing: {}", e);
+        }
+        Box::new(LazyAbortHandle {
+            task: Mutex::new(recv),
+        })
+    }
+
     fn spawn_cpu(&self, cpu: Box<dyn FnOnce() + Send + 'static>) -> AbortHandleRef {
         let (send, recv) = oneshot::channel();
         if let Err(e) = self.cpu.send(SpawnSync {
@@ -223,6 +260,11 @@ struct SpawnAsync<'rt> {
     task_callback: oneshot::Sender<AbortHandleRef>,
 }
 
+struct SpawnLocalAsync<'rt> {
+    future: Box<dyn FnOnce() -> LocalBoxFuture<'rt, ()> + Send + 'rt>,
+    task_callback: oneshot::Sender<AbortHandleRef>,
+}
+
 // A spawn request for a synchronous job.
 struct SpawnSync<'rt> {
     sync: Box<dyn FnOnce() + Send + 'rt>,

From 0ba71baf9fe75ba02339be0c9a1518180cce789d Mon Sep 17 00:00:00 2001
From: Daniel King <dan@spiraldb.com>
Date: Fri, 7 Aug 2026 16:45:40 -0400
Subject: [PATCH 2/7] Keep local file open futures Send

---
 vortex-file/src/open.rs | 113 ++++++++++++++++++++++++----------------
 1 file changed, 68 insertions(+), 45 deletions(-)

diff --git a/vortex-file/src/open.rs b/vortex-file/src/open.rs
index 63ca5163bfa..91b515144eb 100644
--- a/vortex-file/src/open.rs
+++ b/vortex-file/src/open.rs
@@ -3,6 +3,7 @@
 
 use std::sync::Arc;
 
+use futures::FutureExt;
 use futures::executor::block_on;
 use parking_lot::RwLock;
 use vortex_array::dtype::DType;
@@ -302,7 +303,27 @@ impl VortexOpenOptions {
         let footer = if let Some(footer) = self.footer {
             footer
         } else {
-            self.read_footer_local(&reader).await?
+            let file_size = self.file_size;
+            let initial_read_size = self.initial_read_size;
+            let dtype = self.dtype.clone();
+            let session = self.session.clone();
+            let footer_reader = reader.clone();
+            let task = self.session.handle().spawn_local_io(move || {
+                async move {
+                    read_footer_local(
+                        footer_reader.clone(),
+                        file_size,
+                        initial_read_size,
+                        dtype,
+                        session,
+                    )
+                    .await
+                }
+                .boxed_local()
+            });
+            let (footer, initial_offset, initial_read) = task.await?;
+            self.populate_initial_segments(initial_offset, &initial_read, &footer);
+            footer
         };
 
         let segment_cache = Arc::new(InstrumentedSegmentCache::new(
@@ -384,50 +405,6 @@ impl VortexOpenOptions {
         Ok(footer)
     }
 
-    async fn read_footer_local(&self, read: &dyn VortexLocalReadAt) -> VortexResult<Footer> {
-        let file_size = match self.file_size {
-            None => read.size().await?,
-            Some(file_size) => file_size,
-        };
-        let mut initial_read_size = self
-            .initial_read_size
-            .max(MAX_POSTSCRIPT_SIZE as usize + EOF_SIZE);
-        if let Ok(file_size) = usize::try_from(file_size) {
-            initial_read_size = initial_read_size.min(file_size);
-        }
-
-        let initial_offset = file_size - initial_read_size as u64;
-        let initial_read: ByteBuffer = read
-            .read_at_local(initial_offset, initial_read_size, Alignment::none())
-            .await?
-            .try_into_host()?
-            .await?;
-
-        let mut deserializer = Footer::deserializer(initial_read, self.session.clone())
-            .with_size(file_size)
-            .with_some_dtype(self.dtype.clone());
-
-        let footer = loop {
-            match deserializer.deserialize()? {
-                DeserializeStep::NeedMoreData { offset, len } => {
-                    let more_data = read
-                        .read_at_local(offset, len, Alignment::none())
-                        .await?
-                        .try_into_host()?
-                        .await?;
-                    deserializer.prefix_data(more_data);
-                }
-                DeserializeStep::NeedFileSize => unreachable!("We passed file_size above"),
-                DeserializeStep::Done(footer) => break Ok::<_, VortexError>(footer),
-            }
-        }?;
-
-        let initial_offset = file_size - (deserializer.buffer().len() as u64);
-        self.populate_initial_segments(initial_offset, deserializer.buffer(), &footer);
-
-        Ok(footer)
-    }
-
     /// Populate segments in the cache that were covered by the initial read.
     fn populate_initial_segments(
         &self,
@@ -455,6 +432,52 @@ impl VortexOpenOptions {
     }
 }
 
+async fn read_footer_local<R: VortexLocalReadAt + Clone>(
+    read: R,
+    file_size: Option<u64>,
+    initial_read_size: usize,
+    dtype: Option<DType>,
+    session: VortexSession,
+) -> VortexResult<(Footer, u64, ByteBuffer)> {
+    let file_size = match file_size {
+        None => read.size().await?,
+        Some(file_size) => file_size,
+    };
+    let mut initial_read_size = initial_read_size.max(MAX_POSTSCRIPT_SIZE as usize + EOF_SIZE);
+    if let Ok(file_size) = usize::try_from(file_size) {
+        initial_read_size = initial_read_size.min(file_size);
+    }
+
+    let initial_offset = file_size - initial_read_size as u64;
+    let initial_read: ByteBuffer = read
+        .read_at_local(initial_offset, initial_read_size, Alignment::none())
+        .await?
+        .try_into_host()?
+        .await?;
+
+    let mut deserializer = Footer::deserializer(initial_read, session)
+        .with_size(file_size)
+        .with_some_dtype(dtype);
+
+    let footer = loop {
+        match deserializer.deserialize()? {
+            DeserializeStep::NeedMoreData { offset, len } => {
+                let more_data = read
+                    .read_at_local(offset, len, Alignment::none())
+                    .await?
+                    .try_into_host()?
+                    .await?;
+                deserializer.prefix_data(more_data);
+            }
+            DeserializeStep::NeedFileSize => unreachable!("We passed file_size above"),
+            DeserializeStep::Done(footer) => break Ok::<_, VortexError>(footer),
+        }
+    }?;
+
+    let initial_offset = file_size - (deserializer.buffer().len() as u64);
+    Ok((footer, initial_offset, deserializer.buffer().clone()))
+}
+
 #[cfg(feature = "object_store")]
 impl VortexOpenOptions {
     /// Open a Vortex file from an `object_store` backend and path.

From a9b5190d368282ea970770d5a4ec48b50843ff19 Mon Sep 17 00:00:00 2001
From: Daniel King <dan@spiraldb.com>
Date: Fri, 7 Aug 2026 17:20:31 -0400
Subject: [PATCH 3/7] Add local scan task factories

---
 vortex-file/src/open.rs                 |  3 ++-
 vortex-file/src/segments/source.rs      |  3 ++-
 vortex-layout/src/scan/mod.rs           |  2 ++
 vortex-layout/src/scan/repeated_scan.rs | 27 ++++++++++++++++++++++++-
 vortex-layout/src/scan/scan_builder.rs  | 14 +++++++++++++
 vortex-layout/src/scan/tasks.rs         | 19 +++++++++++++++++
 6 files changed, 65 insertions(+), 3 deletions(-)

diff --git a/vortex-file/src/open.rs b/vortex-file/src/open.rs
index 91b515144eb..127d2d00f89 100644
--- a/vortex-file/src/open.rs
+++ b/vortex-file/src/open.rs
@@ -14,8 +14,9 @@ use vortex_buffer::ByteBuffer;
 use vortex_error::VortexError;
 use vortex_error::VortexExpect;
 use vortex_error::VortexResult;
+use vortex_io::VortexLocalReadAt;
+use vortex_io::VortexReadAt;
 use vortex_io::session::RuntimeSessionExt;
-use vortex_io::{VortexLocalReadAt, VortexReadAt};
 use vortex_layout::segments::InstrumentedSegmentCache;
 use vortex_layout::segments::NoOpSegmentCache;
 use vortex_layout::segments::SegmentCache;
diff --git a/vortex-file/src/segments/source.rs b/vortex-file/src/segments/source.rs
index 185419432a1..b5e72e74c6a 100644
--- a/vortex-file/src/segments/source.rs
+++ b/vortex-file/src/segments/source.rs
@@ -24,9 +24,10 @@ use vortex_error::VortexResult;
 use vortex_error::vortex_bail;
 use vortex_error::vortex_err;
 use vortex_error::vortex_panic;
+use vortex_io::VortexLocalReadAt;
+use vortex_io::VortexReadAt;
 use vortex_io::runtime::Handle;
 use vortex_io::runtime::JoinOutcome;
-use vortex_io::{VortexLocalReadAt, VortexReadAt};
 use vortex_layout::segments::SegmentFuture;
 use vortex_layout::segments::SegmentId;
 use vortex_layout::segments::SegmentSource;
diff --git a/vortex-layout/src/scan/mod.rs b/vortex-layout/src/scan/mod.rs
index 98fd1918a42..867c1253d8b 100644
--- a/vortex-layout/src/scan/mod.rs
+++ b/vortex-layout/src/scan/mod.rs
@@ -13,6 +13,8 @@ mod tasks;
 #[cfg(test)]
 mod test;
 
+pub use tasks::LocalTaskFuture;
+
 /// A heuristic for an ideal split size.
 ///
 /// We don't actually know if this is right, but it is probably a good estimate.
diff --git a/vortex-layout/src/scan/repeated_scan.rs b/vortex-layout/src/scan/repeated_scan.rs
index 681f33639bc..2f2fc340416 100644
--- a/vortex-layout/src/scan/repeated_scan.rs
+++ b/vortex-layout/src/scan/repeated_scan.rs
@@ -28,8 +28,10 @@ use vortex_utils::parallelism::get_available_parallelism;
 use crate::LayoutReaderRef;
 use crate::scan::filter::FilterExpr;
 use crate::scan::splits::Splits;
+use crate::scan::tasks::LocalTaskFuture;
 use crate::scan::tasks::TaskContext;
 use crate::scan::tasks::split_exec;
+use crate::scan::tasks::split_exec_local;
 
 /// A projected subset (by indices, range, and filter) of rows from a Vortex data source.
 ///
@@ -123,6 +125,29 @@ impl<A: 'static + Send> RepeatedScan<A> {
         &self,
         row_range: Option<Range<u64>>,
     ) -> VortexResult<Vec<BoxFuture<'static, VortexResult<Option<A>>>>> {
+        self.execute_with(row_range, split_exec)
+    }
+
+    /// Constructs local-I/O task factories for each row split of the scan.
+    ///
+    /// These factories are `Send`, but each scan future they create is intended to be constructed
+    /// and polled on one local I/O worker via [`vortex_io::runtime::Handle::spawn_local_io`].
+    pub fn execute_local(
+        &self,
+        row_range: Option<Range<u64>>,
+    ) -> VortexResult<Vec<LocalTaskFuture<Option<A>>>> {
+        self.execute_with(row_range, split_exec_local)
+    }
+
+    fn execute_with<T>(
+        &self,
+        row_range: Option<Range<u64>>,
+        mut split_exec_fn: impl FnMut(
+            Arc<TaskContext<A>>,
+            vortex_scan::row_mask::RowMask,
+            Option<&mut u64>,
+        ) -> VortexResult<T>,
+    ) -> VortexResult<Vec<T>> {
         let selection_range: Option<Range<u64>> = match &self.selection {
             Selection::IncludeByIndex(buf) if !buf.is_empty() => {
                 Some(buf[0]..buf[buf.len() - 1] + 1)
@@ -186,7 +211,7 @@ impl<A: 'static + Send> RepeatedScan<A> {
                 continue;
             }
 
-            tasks.push(split_exec(Arc::clone(&ctx), row_mask, limit.as_mut())?);
+            tasks.push(split_exec_fn(Arc::clone(&ctx), row_mask, limit.as_mut())?);
             if limit.is_some_and(|l| l == 0) {
                 break;
             }
diff --git a/vortex-layout/src/scan/scan_builder.rs b/vortex-layout/src/scan/scan_builder.rs
index 11fd5c7b882..08047cae684 100644
--- a/vortex-layout/src/scan/scan_builder.rs
+++ b/vortex-layout/src/scan/scan_builder.rs
@@ -44,6 +44,7 @@ use crate::scan::repeated_scan::RepeatedScan;
 use crate::scan::split_by::SplitBy;
 use crate::scan::splits::Splits;
 use crate::scan::splits::attempt_split_ranges;
+use crate::scan::tasks::LocalTaskFuture;
 
 /// Builder for scanning a [`LayoutReader`] into arrays, streams, iterators, or mapped outputs.
 ///
@@ -335,6 +336,19 @@ impl<A: 'static + Send> ScanBuilder<A> {
         self.prepare()?.execute(None)
     }
 
+    /// Constructs local-I/O task factories, one per row split of the scan.
+    ///
+    /// Use this when the caller has a local runtime such as io_uring and wants scan futures to be
+    /// constructed and polled on one I/O worker instead of on the caller's worker thread.
+    pub fn build_local(self) -> VortexResult<Vec<LocalTaskFuture<Option<A>>>> {
+        // The ultimate short circuit
+        if self.limit.is_some_and(|l| l == 0) {
+            return Ok(vec![]);
+        }
+
+        self.prepare()?.execute_local(None)
+    }
+
     /// Returns a [`Stream`] with tasks spawned onto the session's runtime handle.
     pub fn into_stream(
         self,
diff --git a/vortex-layout/src/scan/tasks.rs b/vortex-layout/src/scan/tasks.rs
index a86546e15ef..4ee9c57954e 100644
--- a/vortex-layout/src/scan/tasks.rs
+++ b/vortex-layout/src/scan/tasks.rs
@@ -9,6 +9,7 @@ use std::sync::Arc;
 use bit_vec::BitVec;
 use futures::FutureExt;
 use futures::future::BoxFuture;
+use futures::future::LocalBoxFuture;
 use vortex_array::ArrayRef;
 use vortex_array::MaskFuture;
 use vortex_array::expr::Expression;
@@ -21,6 +22,10 @@ use crate::scan::filter::FilterExpr;
 
 pub type TaskFuture<A> = BoxFuture<'static, VortexResult<A>>;
 
+/// Factory for a split task that must be constructed and polled on one local I/O worker.
+pub type LocalTaskFuture<A> =
+    Box<dyn FnOnce() -> LocalBoxFuture<'static, VortexResult<A>> + Send + 'static>;
+
 /// Logic for executing a single split reading task.
 /// N.B. read_mask should be evaluated against all_false() before calling this
 /// method to avoid creating an empty TaskFuture.
@@ -150,6 +155,20 @@ pub fn split_exec<A: 'static + Send>(
     Ok(array_fut.boxed())
 }
 
+/// Local-I/O variant of [`split_exec`].
+///
+/// The returned value is `Send`, so callers may store it in shared scan state and route it to a
+/// local runtime worker when the split is claimed. The future itself is created by the factory and
+/// then polled on that worker, which supports runtimes where I/O futures carry thread-local state.
+pub fn split_exec_local<A: 'static + Send>(
+    ctx: Arc<TaskContext<A>>,
+    read_mask: RowMask,
+    limit: Option<&mut u64>,
+) -> VortexResult<LocalTaskFuture<Option<A>>> {
+    let task = split_exec(ctx, read_mask, limit)?;
+    Ok(Box::new(move || async move { task.await }.boxed_local()))
+}
+
 /// Information needed to execute a single split task.
 ///
 /// Row selection is evaluated before creating a split task so it's not included

From 881c0c7472505713eb383d71f4831813b6e34e5c Mon Sep 17 00:00:00 2001
From: Daniel King <dan@spiraldb.com>
Date: Fri, 7 Aug 2026 17:28:23 -0400
Subject: [PATCH 4/7] Use local size for local file opens

---
 vortex-file/src/open.rs  | 2 +-
 vortex-io/src/read_at.rs | 6 ++++++
 2 files changed, 7 insertions(+), 1 deletion(-)

diff --git a/vortex-file/src/open.rs b/vortex-file/src/open.rs
index 127d2d00f89..9d6c472dc2d 100644
--- a/vortex-file/src/open.rs
+++ b/vortex-file/src/open.rs
@@ -441,7 +441,7 @@ async fn read_footer_local<R: VortexLocalReadAt + Clone>(
     session: VortexSession,
 ) -> VortexResult<(Footer, u64, ByteBuffer)> {
     let file_size = match file_size {
-        None => read.size().await?,
+        None => read.size_local().await?,
         Some(file_size) => file_size,
     };
     let mut initial_read_size = initial_read_size.max(MAX_POSTSCRIPT_SIZE as usize + EOF_SIZE);
diff --git a/vortex-io/src/read_at.rs b/vortex-io/src/read_at.rs
index ee6ec949be1..63b00278ac3 100644
--- a/vortex-io/src/read_at.rs
+++ b/vortex-io/src/read_at.rs
@@ -113,6 +113,12 @@ pub trait VortexLocalReadAt: Send + Sync + 'static {
     /// Asynchronously get the number of bytes of the underlying source.
     fn size(&self) -> BoxFuture<'static, VortexResult<u64>>;
 
+    /// Asynchronously get the number of bytes of the underlying source on a local I/O worker.
+    fn size_local(&self) -> LocalBoxFuture<'static, VortexResult<u64>> {
+        let size = self.size();
+        async move { size.await }.boxed_local()
+    }
+
     /// Request an asynchronous positional read whose future must be polled on the local I/O
     /// runtime thread that owns the source.
     fn read_at_local(

From 6bb628ba614ab1c2d04b28cb65db39e2b03fa3ed Mon Sep 17 00:00:00 2001
From: Daniel King <dan@spiraldb.com>
Date: Fri, 7 Aug 2026 18:16:21 -0400
Subject: [PATCH 5/7] Keep local scans on their I/O worker

---
 vortex-file/src/file.rs                 |  8 +++---
 vortex-file/src/open.rs                 | 31 ++++++++++++----------
 vortex-file/src/segments/source.rs      | 11 +++++++-
 vortex-io/src/runtime/handle.rs         | 32 +++++++++++++++++++++++
 vortex-io/src/runtime/mod.rs            | 31 ++++++++++++++++++++++
 vortex-layout/src/scan/repeated_scan.rs |  9 ++++++-
 vortex-layout/src/scan/scan_builder.rs  | 11 ++++++++
 vortex-layout/src/scan/tasks.rs         | 34 ++++++++++++++++++++++---
 vortex-layout/src/segments/cache.rs     |  5 ++++
 vortex-layout/src/segments/shared.rs    |  5 ++++
 vortex-layout/src/segments/source.rs    |  6 +++++
 11 files changed, 161 insertions(+), 22 deletions(-)

diff --git a/vortex-file/src/file.rs b/vortex-file/src/file.rs
index c9a71f85c55..187590c2be3 100644
--- a/vortex-file/src/file.rs
+++ b/vortex-file/src/file.rs
@@ -176,10 +176,10 @@ impl VortexFile {
     /// Initiate a scan of the file, returning a builder for projection, filtering, selection, and
     /// execution options.
     pub fn scan(&self) -> VortexResult<ScanBuilder<ArrayRef>> {
-        Ok(ScanBuilder::new(
-            self.session.clone(),
-            self.layout_reader()?,
-        ))
+        Ok(
+            ScanBuilder::new(self.session.clone(), self.layout_reader()?)
+                .with_local_io_worker(self.segment_source.local_io_worker()),
+        )
     }
 
     /// Returns `true` if file-level statistics prove the expression cannot
diff --git a/vortex-file/src/open.rs b/vortex-file/src/open.rs
index 9d6c472dc2d..f5f0ead7eca 100644
--- a/vortex-file/src/open.rs
+++ b/vortex-file/src/open.rs
@@ -291,6 +291,7 @@ impl VortexOpenOptions {
         self,
         reader: R,
     ) -> VortexResult<VortexFile> {
+        let local_io_worker = self.session.handle().acquire_local_io_worker();
         let segment_cache = self
             .segment_cache
             .clone()
@@ -309,19 +310,22 @@ impl VortexOpenOptions {
             let dtype = self.dtype.clone();
             let session = self.session.clone();
             let footer_reader = reader.clone();
-            let task = self.session.handle().spawn_local_io(move || {
-                async move {
-                    read_footer_local(
-                        footer_reader.clone(),
-                        file_size,
-                        initial_read_size,
-                        dtype,
-                        session,
-                    )
-                    .await
-                }
-                .boxed_local()
-            });
+            let task = self
+                .session
+                .handle()
+                .spawn_local_io_on(local_io_worker, move || {
+                    async move {
+                        read_footer_local(
+                            footer_reader.clone(),
+                            file_size,
+                            initial_read_size,
+                            dtype,
+                            session,
+                        )
+                        .await
+                    }
+                    .boxed_local()
+                });
             let (footer, initial_offset, initial_read) = task.await?;
             self.populate_initial_segments(initial_offset, &initial_read, &footer);
             footer
@@ -342,6 +346,7 @@ impl VortexOpenOptions {
             Arc::clone(footer.segment_map()),
             reader,
             self.session.handle(),
+            local_io_worker,
             metrics,
         )));
 
diff --git a/vortex-file/src/segments/source.rs b/vortex-file/src/segments/source.rs
index b5e72e74c6a..33238b8e230 100644
--- a/vortex-file/src/segments/source.rs
+++ b/vortex-file/src/segments/source.rs
@@ -28,6 +28,7 @@ use vortex_io::VortexLocalReadAt;
 use vortex_io::VortexReadAt;
 use vortex_io::runtime::Handle;
 use vortex_io::runtime::JoinOutcome;
+use vortex_io::runtime::LocalIoWorker;
 use vortex_layout::segments::SegmentFuture;
 use vortex_layout::segments::SegmentId;
 use vortex_layout::segments::SegmentSource;
@@ -96,6 +97,7 @@ pub struct FileSegmentSource {
     driver_panic: DriverPanic,
     /// The next read request ID.
     next_id: Arc<AtomicUsize>,
+    local_io_worker: Option<LocalIoWorker>,
 }
 
 impl FileSegmentSource {
@@ -192,6 +194,7 @@ impl FileSegmentSource {
             driver,
             driver_panic,
             next_id: Arc::new(AtomicUsize::new(0)),
+            local_io_worker: None,
         }
     }
 
@@ -204,6 +207,7 @@ impl FileSegmentSource {
         segments: Arc<[SegmentSpec]>,
         reader: R,
         handle: Handle,
+        local_io_worker: LocalIoWorker,
         metrics: RequestMetrics,
     ) -> Self {
         let (send, recv) = mpsc::unbounded();
@@ -234,7 +238,7 @@ impl FileSegmentSource {
         )
         .boxed();
 
-        let mut task = handle.spawn_local_io(move || {
+        let mut task = handle.spawn_local_io_on(local_io_worker, move || {
             async move {
                 stream
                     .map(move |req| {
@@ -283,6 +287,7 @@ impl FileSegmentSource {
             driver,
             driver_panic,
             next_id: Arc::new(AtomicUsize::new(0)),
+            local_io_worker: Some(local_io_worker),
         }
     }
 }
@@ -332,6 +337,10 @@ impl SegmentSource for FileSegmentSource {
         // One allocation: we only box the returned SegmentFuture, not the inner ReadFuture.
         fut.boxed()
     }
+
+    fn local_io_worker(&self) -> Option<LocalIoWorker> {
+        self.local_io_worker
+    }
 }
 
 /// A future that resolves a read request from a [`FileSegmentSource`].
diff --git a/vortex-io/src/runtime/handle.rs b/vortex-io/src/runtime/handle.rs
index de9a4ec60ac..29d986d2cd8 100644
--- a/vortex-io/src/runtime/handle.rs
+++ b/vortex-io/src/runtime/handle.rs
@@ -16,6 +16,7 @@ use vortex_error::vortex_panic;
 
 use crate::runtime::AbortHandleRef;
 use crate::runtime::Executor;
+use crate::runtime::LocalIoWorker;
 
 /// A handle to an active Vortex runtime.
 ///
@@ -162,6 +163,37 @@ impl Handle {
         }
     }
 
+    /// Select a local I/O worker for related tasks that should retain affinity.
+    pub fn acquire_local_io_worker(&self) -> LocalIoWorker {
+        self.runtime().acquire_local_io_worker()
+    }
+
+    /// Spawn a local I/O future on a previously selected worker.
+    pub fn spawn_local_io_on<F, Fut, R>(&self, worker: LocalIoWorker, f: F) -> Task<R>
+    where
+        F: FnOnce() -> Fut + Send + 'static,
+        Fut: Future<Output = R> + 'static,
+        R: Send + 'static,
+    {
+        let (send, recv) = oneshot::channel();
+        let span = tracing::trace_span!(target: "vortex_io::spawn_local_io", "spawn_local_io");
+        let abort_handle = self.runtime().spawn_local_io_on(
+            worker,
+            Box::new(move || {
+                async move {
+                    let output = AssertUnwindSafe(f()).catch_unwind().await;
+                    drop(send.send(output));
+                }
+                .instrument(span)
+                .boxed_local()
+            }),
+        );
+        Task {
+            recv: recv.into_future(),
+            abort_handle: Some(abort_handle),
+        }
+    }
+
     /// Spawn a CPU-bound task for execution on the runtime.
     ///
     /// Note that many runtimes will interleave this work on the same async runtime. See the
diff --git a/vortex-io/src/runtime/mod.rs b/vortex-io/src/runtime/mod.rs
index e6bb1216a4c..aec4549668c 100644
--- a/vortex-io/src/runtime/mod.rs
+++ b/vortex-io/src/runtime/mod.rs
@@ -15,6 +15,20 @@
 use futures::future::BoxFuture;
 use futures::future::LocalBoxFuture;
 
+/// Opaque affinity token for one runtime-local I/O worker.
+#[derive(Clone, Copy, Debug, Default, Eq, Hash, PartialEq)]
+pub struct LocalIoWorker(usize);
+
+impl LocalIoWorker {
+    pub fn new(index: usize) -> Self {
+        Self(index)
+    }
+
+    pub fn index(self) -> usize {
+        self.0
+    }
+}
+
 mod blocking;
 pub use blocking::*;
 mod handle;
@@ -64,6 +78,23 @@ pub trait Executor: Send + Sync {
         vortex_error::vortex_panic!("runtime does not support local I/O futures")
     }
 
+    /// Select a local I/O worker for a group of related tasks.
+    fn acquire_local_io_worker(&self) -> LocalIoWorker {
+        LocalIoWorker::default()
+    }
+
+    /// Spawn a local I/O future on a previously selected worker.
+    fn spawn_local_io_on(
+        &self,
+        worker: LocalIoWorker,
+        future: Box<dyn FnOnce() -> LocalBoxFuture<'static, ()> + Send + 'static>,
+    ) -> AbortHandleRef {
+        if worker != LocalIoWorker::default() {
+            vortex_error::vortex_panic!("runtime does not support multiple local I/O workers")
+        }
+        self.spawn_local_io(future)
+    }
+
     /// Spawns a CPU-bound task for execution on the runtime.
     ///
     /// The returned `AbortHandle` may be used to optimistically cancel the task if it has not
diff --git a/vortex-layout/src/scan/repeated_scan.rs b/vortex-layout/src/scan/repeated_scan.rs
index 2f2fc340416..6fa4bf9655f 100644
--- a/vortex-layout/src/scan/repeated_scan.rs
+++ b/vortex-layout/src/scan/repeated_scan.rs
@@ -20,6 +20,7 @@ use vortex_array::stream::ArrayStreamAdapter;
 use vortex_error::VortexExpect;
 use vortex_error::VortexResult;
 use vortex_io::runtime::BlockingRuntime;
+use vortex_io::runtime::LocalIoWorker;
 use vortex_io::session::RuntimeSessionExt;
 use vortex_scan::selection::Selection;
 use vortex_session::VortexSession;
@@ -57,6 +58,7 @@ pub struct RepeatedScan<A: 'static + Send> {
     limit: Option<u64>,
     /// The dtype of the projected arrays.
     dtype: DType,
+    local_io_worker: Option<LocalIoWorker>,
 }
 
 impl RepeatedScan<ArrayRef> {
@@ -104,6 +106,7 @@ impl<A: 'static + Send> RepeatedScan<A> {
         map_fn: Arc<dyn Fn(ArrayRef) -> VortexResult<A> + Send + Sync>,
         limit: Option<u64>,
         dtype: DType,
+        local_io_worker: Option<LocalIoWorker>,
     ) -> Self {
         Self {
             session,
@@ -118,6 +121,7 @@ impl<A: 'static + Send> RepeatedScan<A> {
             map_fn,
             limit,
             dtype,
+            local_io_worker,
         }
     }
 
@@ -136,7 +140,10 @@ impl<A: 'static + Send> RepeatedScan<A> {
         &self,
         row_range: Option<Range<u64>>,
     ) -> VortexResult<Vec<LocalTaskFuture<Option<A>>>> {
-        self.execute_with(row_range, split_exec_local)
+        let worker = self.local_io_worker;
+        self.execute_with(row_range, move |ctx, read_mask, limit| {
+            split_exec_local(ctx, read_mask, limit, worker)
+        })
     }
 
     fn execute_with<T>(
diff --git a/vortex-layout/src/scan/scan_builder.rs b/vortex-layout/src/scan/scan_builder.rs
index 08047cae684..3a5bc32e335 100644
--- a/vortex-layout/src/scan/scan_builder.rs
+++ b/vortex-layout/src/scan/scan_builder.rs
@@ -30,6 +30,7 @@ use vortex_error::VortexResult;
 use vortex_error::vortex_bail;
 use vortex_io::runtime::BlockingRuntime;
 use vortex_io::runtime::Handle;
+use vortex_io::runtime::LocalIoWorker;
 use vortex_io::runtime::Task;
 use vortex_io::session::RuntimeSessionExt;
 use vortex_metrics::MetricsRegistry;
@@ -83,6 +84,7 @@ pub struct ScanBuilder<A> {
     /// The row-offset assigned to the first row of the file. Used by the `row_idx` expression,
     /// but not by the scan [`Selection`] which remains relative.
     row_offset: u64,
+    local_io_worker: Option<LocalIoWorker>,
 }
 
 impl ScanBuilder<ArrayRef> {
@@ -105,6 +107,7 @@ impl ScanBuilder<ArrayRef> {
             file_stats: None,
             limit: None,
             row_offset: 0,
+            local_io_worker: None,
         }
     }
 
@@ -132,6 +135,12 @@ impl ScanBuilder<ArrayRef> {
 }
 
 impl<A: 'static + Send> ScanBuilder<A> {
+    /// Pin local scan tasks to the worker that owns the underlying I/O driver.
+    pub fn with_local_io_worker(mut self, worker: Option<LocalIoWorker>) -> Self {
+        self.local_io_worker = worker;
+        self
+    }
+
     /// Add a filter expression evaluated against the projected row ranges.
     pub fn with_filter(mut self, filter: Expression) -> Self {
         self.filter = Some(filter);
@@ -258,6 +267,7 @@ impl<A: 'static + Send> ScanBuilder<A> {
             file_stats: self.file_stats,
             limit: self.limit,
             row_offset: self.row_offset,
+            local_io_worker: self.local_io_worker,
             map_fn: Arc::new(move |a| old_map_fn(a).and_then(&map_fn)),
         }
     }
@@ -323,6 +333,7 @@ impl<A: 'static + Send> ScanBuilder<A> {
             self.map_fn,
             self.limit,
             dtype,
+            self.local_io_worker,
         ))
     }
 
diff --git a/vortex-layout/src/scan/tasks.rs b/vortex-layout/src/scan/tasks.rs
index 4ee9c57954e..eb458ab9de0 100644
--- a/vortex-layout/src/scan/tasks.rs
+++ b/vortex-layout/src/scan/tasks.rs
@@ -14,6 +14,9 @@ use vortex_array::ArrayRef;
 use vortex_array::MaskFuture;
 use vortex_array::expr::Expression;
 use vortex_error::VortexResult;
+use vortex_io::runtime::Handle;
+use vortex_io::runtime::LocalIoWorker;
+use vortex_io::runtime::Task;
 use vortex_mask::Mask;
 use vortex_scan::row_mask::RowMask;
 
@@ -23,8 +26,30 @@ use crate::scan::filter::FilterExpr;
 pub type TaskFuture<A> = BoxFuture<'static, VortexResult<A>>;
 
 /// Factory for a split task that must be constructed and polled on one local I/O worker.
-pub type LocalTaskFuture<A> =
-    Box<dyn FnOnce() -> LocalBoxFuture<'static, VortexResult<A>> + Send + 'static>;
+pub struct LocalTaskFuture<A> {
+    local_io_worker: Option<LocalIoWorker>,
+    future: Box<dyn FnOnce() -> LocalBoxFuture<'static, VortexResult<A>> + Send + 'static>,
+}
+
+impl<A: Send + 'static> LocalTaskFuture<A> {
+    pub fn new(
+        local_io_worker: Option<LocalIoWorker>,
+        future: impl FnOnce() -> LocalBoxFuture<'static, VortexResult<A>> + Send + 'static,
+    ) -> Self {
+        Self {
+            local_io_worker,
+            future: Box::new(future),
+        }
+    }
+
+    /// Spawn this task on the I/O worker that owns its segment source.
+    pub fn spawn(self, handle: &Handle) -> Task<VortexResult<A>> {
+        match self.local_io_worker {
+            Some(worker) => handle.spawn_local_io_on(worker, self.future),
+            None => handle.spawn_local_io(self.future),
+        }
+    }
+}
 
 /// Logic for executing a single split reading task.
 /// N.B. read_mask should be evaluated against all_false() before calling this
@@ -164,9 +189,12 @@ pub fn split_exec_local<A: 'static + Send>(
     ctx: Arc<TaskContext<A>>,
     read_mask: RowMask,
     limit: Option<&mut u64>,
+    local_io_worker: Option<LocalIoWorker>,
 ) -> VortexResult<LocalTaskFuture<Option<A>>> {
     let task = split_exec(ctx, read_mask, limit)?;
-    Ok(Box::new(move || async move { task.await }.boxed_local()))
+    Ok(LocalTaskFuture::new(local_io_worker, move || {
+        async move { task.await }.boxed_local()
+    }))
 }
 
 /// Information needed to execute a single split task.
diff --git a/vortex-layout/src/segments/cache.rs b/vortex-layout/src/segments/cache.rs
index 1f7f5e91f5c..bbeb58e6d4c 100644
--- a/vortex-layout/src/segments/cache.rs
+++ b/vortex-layout/src/segments/cache.rs
@@ -13,6 +13,7 @@ use vortex_array::buffer::BufferHandle;
 use vortex_buffer::ByteBuffer;
 use vortex_error::VortexExpect;
 use vortex_error::VortexResult;
+use vortex_io::runtime::LocalIoWorker;
 use vortex_metrics::Counter;
 use vortex_metrics::Label;
 use vortex_metrics::MetricBuilder;
@@ -166,4 +167,8 @@ impl SegmentSource for SegmentCacheSourceAdapter {
         }
         .boxed()
     }
+
+    fn local_io_worker(&self) -> Option<LocalIoWorker> {
+        self.source.local_io_worker()
+    }
 }
diff --git a/vortex-layout/src/segments/shared.rs b/vortex-layout/src/segments/shared.rs
index c794daf608e..b85e9d6f96d 100644
--- a/vortex-layout/src/segments/shared.rs
+++ b/vortex-layout/src/segments/shared.rs
@@ -11,6 +11,7 @@ use vortex_array::buffer::BufferHandle;
 use vortex_error::SharedVortexResult;
 use vortex_error::VortexError;
 use vortex_error::VortexExpect;
+use vortex_io::runtime::LocalIoWorker;
 use vortex_utils::aliases::dash_map::DashMap;
 use vortex_utils::aliases::dash_map::Entry;
 
@@ -61,6 +62,10 @@ impl<S: SegmentSource> SegmentSource for SharedSegmentSource<S> {
             }
         }
     }
+
+    fn local_io_worker(&self) -> Option<LocalIoWorker> {
+        self.inner.local_io_worker()
+    }
 }
 
 #[cfg(test)]
diff --git a/vortex-layout/src/segments/source.rs b/vortex-layout/src/segments/source.rs
index 5c709f5a7ad..c1a1c9706ca 100644
--- a/vortex-layout/src/segments/source.rs
+++ b/vortex-layout/src/segments/source.rs
@@ -4,6 +4,7 @@
 use futures::future::BoxFuture;
 use vortex_array::buffer::BufferHandle;
 use vortex_error::VortexResult;
+use vortex_io::runtime::LocalIoWorker;
 
 use crate::segments::SegmentId;
 /// Static future resolving to a segment byte buffer.
@@ -16,4 +17,9 @@ pub type SegmentFuture = BoxFuture<'static, VortexResult<BufferHandle>>;
 pub trait SegmentSource: 'static + Send + Sync {
     /// Request a segment, returning a future that will eventually resolve to the segment data.
     fn request(&self, id: SegmentId) -> SegmentFuture;
+
+    /// Worker that owns runtime-local I/O for this source, when applicable.
+    fn local_io_worker(&self) -> Option<LocalIoWorker> {
+        None
+    }
 }

From 5f148cd571150cb3a051f06928f2543870954d4a Mon Sep 17 00:00:00 2001
From: Daniel King <dan@spiraldb.com>
Date: Fri, 7 Aug 2026 18:58:44 -0400
Subject: [PATCH 6/7] Keep scan processing off local I/O workers

---
 vortex-file/src/file.rs                 |  8 ++---
 vortex-file/src/segments/source.rs      |  7 ----
 vortex-layout/src/scan/mod.rs           |  2 --
 vortex-layout/src/scan/repeated_scan.rs | 34 +-----------------
 vortex-layout/src/scan/scan_builder.rs  | 25 -------------
 vortex-layout/src/scan/tasks.rs         | 47 -------------------------
 vortex-layout/src/segments/cache.rs     |  5 ---
 vortex-layout/src/segments/shared.rs    |  5 ---
 vortex-layout/src/segments/source.rs    |  6 ----
 9 files changed, 5 insertions(+), 134 deletions(-)

diff --git a/vortex-file/src/file.rs b/vortex-file/src/file.rs
index 187590c2be3..c9a71f85c55 100644
--- a/vortex-file/src/file.rs
+++ b/vortex-file/src/file.rs
@@ -176,10 +176,10 @@ impl VortexFile {
     /// Initiate a scan of the file, returning a builder for projection, filtering, selection, and
     /// execution options.
     pub fn scan(&self) -> VortexResult<ScanBuilder<ArrayRef>> {
-        Ok(
-            ScanBuilder::new(self.session.clone(), self.layout_reader()?)
-                .with_local_io_worker(self.segment_source.local_io_worker()),
-        )
+        Ok(ScanBuilder::new(
+            self.session.clone(),
+            self.layout_reader()?,
+        ))
     }
 
     /// Returns `true` if file-level statistics prove the expression cannot
diff --git a/vortex-file/src/segments/source.rs b/vortex-file/src/segments/source.rs
index 33238b8e230..be7468966c5 100644
--- a/vortex-file/src/segments/source.rs
+++ b/vortex-file/src/segments/source.rs
@@ -97,7 +97,6 @@ pub struct FileSegmentSource {
     driver_panic: DriverPanic,
     /// The next read request ID.
     next_id: Arc<AtomicUsize>,
-    local_io_worker: Option<LocalIoWorker>,
 }
 
 impl FileSegmentSource {
@@ -194,7 +193,6 @@ impl FileSegmentSource {
             driver,
             driver_panic,
             next_id: Arc::new(AtomicUsize::new(0)),
-            local_io_worker: None,
         }
     }
 
@@ -287,7 +285,6 @@ impl FileSegmentSource {
             driver,
             driver_panic,
             next_id: Arc::new(AtomicUsize::new(0)),
-            local_io_worker: Some(local_io_worker),
         }
     }
 }
@@ -337,10 +334,6 @@ impl SegmentSource for FileSegmentSource {
         // One allocation: we only box the returned SegmentFuture, not the inner ReadFuture.
         fut.boxed()
     }
-
-    fn local_io_worker(&self) -> Option<LocalIoWorker> {
-        self.local_io_worker
-    }
 }
 
 /// A future that resolves a read request from a [`FileSegmentSource`].
diff --git a/vortex-layout/src/scan/mod.rs b/vortex-layout/src/scan/mod.rs
index 867c1253d8b..98fd1918a42 100644
--- a/vortex-layout/src/scan/mod.rs
+++ b/vortex-layout/src/scan/mod.rs
@@ -13,8 +13,6 @@ mod tasks;
 #[cfg(test)]
 mod test;
 
-pub use tasks::LocalTaskFuture;
-
 /// A heuristic for an ideal split size.
 ///
 /// We don't actually know if this is right, but it is probably a good estimate.
diff --git a/vortex-layout/src/scan/repeated_scan.rs b/vortex-layout/src/scan/repeated_scan.rs
index 6fa4bf9655f..681f33639bc 100644
--- a/vortex-layout/src/scan/repeated_scan.rs
+++ b/vortex-layout/src/scan/repeated_scan.rs
@@ -20,7 +20,6 @@ use vortex_array::stream::ArrayStreamAdapter;
 use vortex_error::VortexExpect;
 use vortex_error::VortexResult;
 use vortex_io::runtime::BlockingRuntime;
-use vortex_io::runtime::LocalIoWorker;
 use vortex_io::session::RuntimeSessionExt;
 use vortex_scan::selection::Selection;
 use vortex_session::VortexSession;
@@ -29,10 +28,8 @@ use vortex_utils::parallelism::get_available_parallelism;
 use crate::LayoutReaderRef;
 use crate::scan::filter::FilterExpr;
 use crate::scan::splits::Splits;
-use crate::scan::tasks::LocalTaskFuture;
 use crate::scan::tasks::TaskContext;
 use crate::scan::tasks::split_exec;
-use crate::scan::tasks::split_exec_local;
 
 /// A projected subset (by indices, range, and filter) of rows from a Vortex data source.
 ///
@@ -58,7 +55,6 @@ pub struct RepeatedScan<A: 'static + Send> {
     limit: Option<u64>,
     /// The dtype of the projected arrays.
     dtype: DType,
-    local_io_worker: Option<LocalIoWorker>,
 }
 
 impl RepeatedScan<ArrayRef> {
@@ -106,7 +102,6 @@ impl<A: 'static + Send> RepeatedScan<A> {
         map_fn: Arc<dyn Fn(ArrayRef) -> VortexResult<A> + Send + Sync>,
         limit: Option<u64>,
         dtype: DType,
-        local_io_worker: Option<LocalIoWorker>,
     ) -> Self {
         Self {
             session,
@@ -121,7 +116,6 @@ impl<A: 'static + Send> RepeatedScan<A> {
             map_fn,
             limit,
             dtype,
-            local_io_worker,
         }
     }
 
@@ -129,32 +123,6 @@ impl<A: 'static + Send> RepeatedScan<A> {
         &self,
         row_range: Option<Range<u64>>,
     ) -> VortexResult<Vec<BoxFuture<'static, VortexResult<Option<A>>>>> {
-        self.execute_with(row_range, split_exec)
-    }
-
-    /// Constructs local-I/O task factories for each row split of the scan.
-    ///
-    /// These factories are `Send`, but each scan future they create is intended to be constructed
-    /// and polled on one local I/O worker via [`vortex_io::runtime::Handle::spawn_local_io`].
-    pub fn execute_local(
-        &self,
-        row_range: Option<Range<u64>>,
-    ) -> VortexResult<Vec<LocalTaskFuture<Option<A>>>> {
-        let worker = self.local_io_worker;
-        self.execute_with(row_range, move |ctx, read_mask, limit| {
-            split_exec_local(ctx, read_mask, limit, worker)
-        })
-    }
-
-    fn execute_with<T>(
-        &self,
-        row_range: Option<Range<u64>>,
-        mut split_exec_fn: impl FnMut(
-            Arc<TaskContext<A>>,
-            vortex_scan::row_mask::RowMask,
-            Option<&mut u64>,
-        ) -> VortexResult<T>,
-    ) -> VortexResult<Vec<T>> {
         let selection_range: Option<Range<u64>> = match &self.selection {
             Selection::IncludeByIndex(buf) if !buf.is_empty() => {
                 Some(buf[0]..buf[buf.len() - 1] + 1)
@@ -218,7 +186,7 @@ impl<A: 'static + Send> RepeatedScan<A> {
                 continue;
             }
 
-            tasks.push(split_exec_fn(Arc::clone(&ctx), row_mask, limit.as_mut())?);
+            tasks.push(split_exec(Arc::clone(&ctx), row_mask, limit.as_mut())?);
             if limit.is_some_and(|l| l == 0) {
                 break;
             }
diff --git a/vortex-layout/src/scan/scan_builder.rs b/vortex-layout/src/scan/scan_builder.rs
index 3a5bc32e335..11fd5c7b882 100644
--- a/vortex-layout/src/scan/scan_builder.rs
+++ b/vortex-layout/src/scan/scan_builder.rs
@@ -30,7 +30,6 @@ use vortex_error::VortexResult;
 use vortex_error::vortex_bail;
 use vortex_io::runtime::BlockingRuntime;
 use vortex_io::runtime::Handle;
-use vortex_io::runtime::LocalIoWorker;
 use vortex_io::runtime::Task;
 use vortex_io::session::RuntimeSessionExt;
 use vortex_metrics::MetricsRegistry;
@@ -45,7 +44,6 @@ use crate::scan::repeated_scan::RepeatedScan;
 use crate::scan::split_by::SplitBy;
 use crate::scan::splits::Splits;
 use crate::scan::splits::attempt_split_ranges;
-use crate::scan::tasks::LocalTaskFuture;
 
 /// Builder for scanning a [`LayoutReader`] into arrays, streams, iterators, or mapped outputs.
 ///
@@ -84,7 +82,6 @@ pub struct ScanBuilder<A> {
     /// The row-offset assigned to the first row of the file. Used by the `row_idx` expression,
     /// but not by the scan [`Selection`] which remains relative.
     row_offset: u64,
-    local_io_worker: Option<LocalIoWorker>,
 }
 
 impl ScanBuilder<ArrayRef> {
@@ -107,7 +104,6 @@ impl ScanBuilder<ArrayRef> {
             file_stats: None,
             limit: None,
             row_offset: 0,
-            local_io_worker: None,
         }
     }
 
@@ -135,12 +131,6 @@ impl ScanBuilder<ArrayRef> {
 }
 
 impl<A: 'static + Send> ScanBuilder<A> {
-    /// Pin local scan tasks to the worker that owns the underlying I/O driver.
-    pub fn with_local_io_worker(mut self, worker: Option<LocalIoWorker>) -> Self {
-        self.local_io_worker = worker;
-        self
-    }
-
     /// Add a filter expression evaluated against the projected row ranges.
     pub fn with_filter(mut self, filter: Expression) -> Self {
         self.filter = Some(filter);
@@ -267,7 +257,6 @@ impl<A: 'static + Send> ScanBuilder<A> {
             file_stats: self.file_stats,
             limit: self.limit,
             row_offset: self.row_offset,
-            local_io_worker: self.local_io_worker,
             map_fn: Arc::new(move |a| old_map_fn(a).and_then(&map_fn)),
         }
     }
@@ -333,7 +322,6 @@ impl<A: 'static + Send> ScanBuilder<A> {
             self.map_fn,
             self.limit,
             dtype,
-            self.local_io_worker,
         ))
     }
 
@@ -347,19 +335,6 @@ impl<A: 'static + Send> ScanBuilder<A> {
         self.prepare()?.execute(None)
     }
 
-    /// Constructs local-I/O task factories, one per row split of the scan.
-    ///
-    /// Use this when the caller has a local runtime such as io_uring and wants scan futures to be
-    /// constructed and polled on one I/O worker instead of on the caller's worker thread.
-    pub fn build_local(self) -> VortexResult<Vec<LocalTaskFuture<Option<A>>>> {
-        // The ultimate short circuit
-        if self.limit.is_some_and(|l| l == 0) {
-            return Ok(vec![]);
-        }
-
-        self.prepare()?.execute_local(None)
-    }
-
     /// Returns a [`Stream`] with tasks spawned onto the session's runtime handle.
     pub fn into_stream(
         self,
diff --git a/vortex-layout/src/scan/tasks.rs b/vortex-layout/src/scan/tasks.rs
index eb458ab9de0..a86546e15ef 100644
--- a/vortex-layout/src/scan/tasks.rs
+++ b/vortex-layout/src/scan/tasks.rs
@@ -9,14 +9,10 @@ use std::sync::Arc;
 use bit_vec::BitVec;
 use futures::FutureExt;
 use futures::future::BoxFuture;
-use futures::future::LocalBoxFuture;
 use vortex_array::ArrayRef;
 use vortex_array::MaskFuture;
 use vortex_array::expr::Expression;
 use vortex_error::VortexResult;
-use vortex_io::runtime::Handle;
-use vortex_io::runtime::LocalIoWorker;
-use vortex_io::runtime::Task;
 use vortex_mask::Mask;
 use vortex_scan::row_mask::RowMask;
 
@@ -25,32 +21,6 @@ use crate::scan::filter::FilterExpr;
 
 pub type TaskFuture<A> = BoxFuture<'static, VortexResult<A>>;
 
-/// Factory for a split task that must be constructed and polled on one local I/O worker.
-pub struct LocalTaskFuture<A> {
-    local_io_worker: Option<LocalIoWorker>,
-    future: Box<dyn FnOnce() -> LocalBoxFuture<'static, VortexResult<A>> + Send + 'static>,
-}
-
-impl<A: Send + 'static> LocalTaskFuture<A> {
-    pub fn new(
-        local_io_worker: Option<LocalIoWorker>,
-        future: impl FnOnce() -> LocalBoxFuture<'static, VortexResult<A>> + Send + 'static,
-    ) -> Self {
-        Self {
-            local_io_worker,
-            future: Box::new(future),
-        }
-    }
-
-    /// Spawn this task on the I/O worker that owns its segment source.
-    pub fn spawn(self, handle: &Handle) -> Task<VortexResult<A>> {
-        match self.local_io_worker {
-            Some(worker) => handle.spawn_local_io_on(worker, self.future),
-            None => handle.spawn_local_io(self.future),
-        }
-    }
-}
-
 /// Logic for executing a single split reading task.
 /// N.B. read_mask should be evaluated against all_false() before calling this
 /// method to avoid creating an empty TaskFuture.
@@ -180,23 +150,6 @@ pub fn split_exec<A: 'static + Send>(
     Ok(array_fut.boxed())
 }
 
-/// Local-I/O variant of [`split_exec`].
-///
-/// The returned value is `Send`, so callers may store it in shared scan state and route it to a
-/// local runtime worker when the split is claimed. The future itself is created by the factory and
-/// then polled on that worker, which supports runtimes where I/O futures carry thread-local state.
-pub fn split_exec_local<A: 'static + Send>(
-    ctx: Arc<TaskContext<A>>,
-    read_mask: RowMask,
-    limit: Option<&mut u64>,
-    local_io_worker: Option<LocalIoWorker>,
-) -> VortexResult<LocalTaskFuture<Option<A>>> {
-    let task = split_exec(ctx, read_mask, limit)?;
-    Ok(LocalTaskFuture::new(local_io_worker, move || {
-        async move { task.await }.boxed_local()
-    }))
-}
-
 /// Information needed to execute a single split task.
 ///
 /// Row selection is evaluated before creating a split task so it's not included
diff --git a/vortex-layout/src/segments/cache.rs b/vortex-layout/src/segments/cache.rs
index bbeb58e6d4c..1f7f5e91f5c 100644
--- a/vortex-layout/src/segments/cache.rs
+++ b/vortex-layout/src/segments/cache.rs
@@ -13,7 +13,6 @@ use vortex_array::buffer::BufferHandle;
 use vortex_buffer::ByteBuffer;
 use vortex_error::VortexExpect;
 use vortex_error::VortexResult;
-use vortex_io::runtime::LocalIoWorker;
 use vortex_metrics::Counter;
 use vortex_metrics::Label;
 use vortex_metrics::MetricBuilder;
@@ -167,8 +166,4 @@ impl SegmentSource for SegmentCacheSourceAdapter {
         }
         .boxed()
     }
-
-    fn local_io_worker(&self) -> Option<LocalIoWorker> {
-        self.source.local_io_worker()
-    }
 }
diff --git a/vortex-layout/src/segments/shared.rs b/vortex-layout/src/segments/shared.rs
index b85e9d6f96d..c794daf608e 100644
--- a/vortex-layout/src/segments/shared.rs
+++ b/vortex-layout/src/segments/shared.rs
@@ -11,7 +11,6 @@ use vortex_array::buffer::BufferHandle;
 use vortex_error::SharedVortexResult;
 use vortex_error::VortexError;
 use vortex_error::VortexExpect;
-use vortex_io::runtime::LocalIoWorker;
 use vortex_utils::aliases::dash_map::DashMap;
 use vortex_utils::aliases::dash_map::Entry;
 
@@ -62,10 +61,6 @@ impl<S: SegmentSource> SegmentSource for SharedSegmentSource<S> {
             }
         }
     }
-
-    fn local_io_worker(&self) -> Option<LocalIoWorker> {
-        self.inner.local_io_worker()
-    }
 }
 
 #[cfg(test)]
diff --git a/vortex-layout/src/segments/source.rs b/vortex-layout/src/segments/source.rs
index c1a1c9706ca..5c709f5a7ad 100644
--- a/vortex-layout/src/segments/source.rs
+++ b/vortex-layout/src/segments/source.rs
@@ -4,7 +4,6 @@
 use futures::future::BoxFuture;
 use vortex_array::buffer::BufferHandle;
 use vortex_error::VortexResult;
-use vortex_io::runtime::LocalIoWorker;
 
 use crate::segments::SegmentId;
 /// Static future resolving to a segment byte buffer.
@@ -17,9 +16,4 @@ pub type SegmentFuture = BoxFuture<'static, VortexResult<BufferHandle>>;
 pub trait SegmentSource: 'static + Send + Sync {
     /// Request a segment, returning a future that will eventually resolve to the segment data.
     fn request(&self, id: SegmentId) -> SegmentFuture;
-
-    /// Worker that owns runtime-local I/O for this source, when applicable.
-    fn local_io_worker(&self) -> Option<LocalIoWorker> {
-        None
-    }
 }

From 3e30f104ec3ea7aed164feaa2874967b019d5bd3 Mon Sep 17 00:00:00 2001
From: Daniel King <dan@spiraldb.com>
Date: Fri, 7 Aug 2026 19:36:45 -0400
Subject: [PATCH 7/7] Test non-Send local file scans

---
 vortex-file/src/tests.rs | 123 +++++++++++++++++++++++++++++++++++++++
 vortex-io/src/read_at.rs |   3 +-
 2 files changed, 124 insertions(+), 2 deletions(-)

diff --git a/vortex-file/src/tests.rs b/vortex-file/src/tests.rs
index e665ab18d50..cea69bd2659 100644
--- a/vortex-file/src/tests.rs
+++ b/vortex-file/src/tests.rs
@@ -3,12 +3,19 @@
 
 #![expect(clippy::cast_possible_truncation)]
 use std::iter;
+use std::rc::Rc;
 use std::sync::Arc;
 use std::sync::LazyLock;
+use std::sync::atomic::AtomicUsize;
+use std::sync::atomic::Ordering;
+use std::thread;
 
 use bytes::Bytes;
+use futures::FutureExt;
 use futures::StreamExt;
 use futures::TryStreamExt;
+use futures::future::BoxFuture;
+use futures::future::LocalBoxFuture;
 use futures::pin_mut;
 use rstest::rstest;
 use vortex_array::ArrayRef;
@@ -62,11 +69,18 @@ use vortex_array::validity::Validity;
 use vortex_btrblocks::BtrBlocksCompressorBuilder;
 use vortex_btrblocks::SchemeExt;
 use vortex_btrblocks::schemes::string::StringDictScheme;
+use vortex_buffer::Alignment;
 use vortex_buffer::Buffer;
+use vortex_buffer::ByteBuffer;
 use vortex_buffer::ByteBufferMut;
 use vortex_buffer::buffer;
 use vortex_error::VortexResult;
+use vortex_error::vortex_bail;
+use vortex_io::VortexLocalReadAt;
+use vortex_io::runtime::BlockingRuntime;
+use vortex_io::runtime::single::SingleThreadRuntime;
 use vortex_io::session::RuntimeSession;
+use vortex_io::session::RuntimeSessionExt;
 use vortex_layout::Layout;
 use vortex_layout::layouts::flat::writer::FlatLayoutStrategy;
 use vortex_layout::layouts::zoned::LegacyStats;
@@ -143,6 +157,115 @@ async fn test_read_simple() {
     assert_eq!(row_count, 8);
 }
 
+#[derive(Clone)]
+struct StrictLocalReadAt {
+    bytes: ByteBuffer,
+    worker: thread::ThreadId,
+    local_sizes: Arc<AtomicUsize>,
+    local_reads: Arc<AtomicUsize>,
+    send_sizes: Arc<AtomicUsize>,
+}
+
+impl VortexLocalReadAt for StrictLocalReadAt {
+    fn concurrency(&self) -> usize {
+        4
+    }
+
+    fn size(&self) -> BoxFuture<'static, VortexResult<u64>> {
+        self.send_sizes.fetch_add(1, Ordering::Relaxed);
+        async { vortex_bail!("local file open used the Send size fallback") }.boxed()
+    }
+
+    fn size_local(&self) -> LocalBoxFuture<'static, VortexResult<u64>> {
+        let marker = Rc::new(());
+        let bytes = self.bytes.clone();
+        let worker = self.worker;
+        let local_sizes = Arc::clone(&self.local_sizes);
+        async move {
+            let _marker = marker;
+            assert_eq!(thread::current().id(), worker);
+            local_sizes.fetch_add(1, Ordering::Relaxed);
+            Ok(bytes.len() as u64)
+        }
+        .boxed_local()
+    }
+
+    fn read_at_local(
+        &self,
+        offset: u64,
+        length: usize,
+        alignment: Alignment,
+    ) -> LocalBoxFuture<'static, VortexResult<vortex_array::buffer::BufferHandle>> {
+        let marker = Rc::new(());
+        let bytes = self.bytes.clone();
+        let worker = self.worker;
+        let local_reads = Arc::clone(&self.local_reads);
+        async move {
+            let _marker = marker;
+            assert_eq!(thread::current().id(), worker);
+            local_reads.fetch_add(1, Ordering::Relaxed);
+            let start = usize::try_from(offset).expect("offset fits usize");
+            let end = start.checked_add(length).expect("range does not overflow");
+            if end > bytes.len() {
+                vortex_bail!(
+                    "requested range {start}..{end} exceeds buffer length {}",
+                    bytes.len()
+                );
+            }
+            Ok(vortex_array::buffer::BufferHandle::new_host(
+                bytes.slice_unaligned(start..end).aligned(alignment),
+            ))
+        }
+        .boxed_local()
+    }
+}
+
+#[test]
+fn local_file_scan_keeps_non_send_reads_on_runtime_worker() -> VortexResult<()> {
+    const ROW_COUNT: usize = 32_768;
+
+    let runtime = SingleThreadRuntime::default();
+    let session = array_session()
+        .with::<LayoutSession>()
+        .with::<RuntimeSession>()
+        .with_handle(runtime.handle());
+    crate::register_default_encodings(&session);
+
+    runtime.block_on(async move {
+        let input = StructArray::from_fields(&[(
+            "value",
+            PrimitiveArray::from_iter(0..ROW_COUNT as u32).into_array(),
+        )])?
+        .into_array();
+        let mut bytes = ByteBufferMut::empty();
+        session
+            .write_options()
+            .with_strategy(Arc::new(FlatLayoutStrategy::default()))
+            .write(&mut bytes, input.to_array_stream())
+            .await?;
+
+        let local_sizes = Arc::new(AtomicUsize::new(0));
+        let local_reads = Arc::new(AtomicUsize::new(0));
+        let send_sizes = Arc::new(AtomicUsize::new(0));
+        let reader = StrictLocalReadAt {
+            bytes: bytes.freeze(),
+            worker: thread::current().id(),
+            local_sizes: Arc::clone(&local_sizes),
+            local_reads: Arc::clone(&local_reads),
+            send_sizes: Arc::clone(&send_sizes),
+        };
+
+        let file = session.open_options().open_read_local(reader).await?;
+        let result = file.scan()?.into_array_stream()?.read_all().await?;
+
+        assert_eq!(result.len(), ROW_COUNT);
+        assert_eq!(local_sizes.load(Ordering::Relaxed), 1);
+        assert!(local_reads.load(Ordering::Relaxed) > 1);
+        assert_eq!(send_sizes.load(Ordering::Relaxed), 0);
+        Ok(())
+    })
+}
+
 #[tokio::test]
 #[cfg_attr(miri, ignore)]
 async fn test_round_trip_many_types() {
diff --git a/vortex-io/src/read_at.rs b/vortex-io/src/read_at.rs
index 63b00278ac3..5b2380c8476 100644
--- a/vortex-io/src/read_at.rs
+++ b/vortex-io/src/read_at.rs
@@ -115,8 +115,7 @@ pub trait VortexLocalReadAt: Send + Sync + 'static {
 
     /// Asynchronously get the number of bytes of the underlying source on a local I/O worker.
     fn size_local(&self) -> LocalBoxFuture<'static, VortexResult<u64>> {
-        let size = self.size();
-        async move { size.await }.boxed_local()
+        self.size()
     }
 
     /// Request an asynchronous positional read whose future must be polled on the local I/O