diff --git a/vortex-file/src/open.rs b/vortex-file/src/open.rs
index 1bd42c7be41..f5f0ead7eca 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;
@@ -13,6 +14,7 @@ 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_layout::segments::InstrumentedSegmentCache;
@@ -283,6 +285,84 @@ 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 local_io_worker = self.session.handle().acquire_local_io_worker();
+        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 {
+            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_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
+        };
+
+        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(),
+            local_io_worker,
+            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 {
@@ -358,6 +438,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_local().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.
diff --git a/vortex-file/src/segments/source.rs b/vortex-file/src/segments/source.rs
index 3af33362b05..be7468966c5 100644
--- a/vortex-file/src/segments/source.rs
+++ b/vortex-file/src/segments/source.rs
@@ -24,9 +24,11 @@ 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::runtime::LocalIoWorker;
 use vortex_layout::segments::SegmentFuture;
 use vortex_layout::segments::SegmentId;
 use vortex_layout::segments::SegmentSource;
@@ -193,6 +195,98 @@ 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,
+        local_io_worker: LocalIoWorker,
+        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_on(local_io_worker, 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-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 aa9a8a03abf..5b2380c8476 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,42 @@ 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>>;
+
+    /// Asynchronously get the number of bytes of the underlying source on a local I/O worker.
+    fn size_local(&self) -> LocalBoxFuture<'static, VortexResult<u64>> {
+        self.size()
+    }
+
+    /// 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..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.
 ///
@@ -135,6 +136,64 @@ 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),
+        }
+    }
+
+    /// 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 864d3c8b2c6..aec4549668c 100644
--- a/vortex-io/src/runtime/mod.rs
+++ b/vortex-io/src/runtime/mod.rs
@@ -13,6 +13,21 @@
 //! * Tokio: work is driven on a Tokio runtime provided by the caller.
 
 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::*;
@@ -52,6 +67,34 @@ 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")
+    }
+
+    /// 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-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>,