@@ -16,6 +16,7 @@
 #include "duckdb/function/table_function.hpp"
 #include "duckdb/main/capi/capi_internal.hpp"
 #include "duckdb/main/connection.hpp"
+#include "duckdb/parallel/async_result.hpp"
 #include "duckdb/parser/parsed_data/create_table_function_info.hpp"
 #include "duckdb/planner/operator/logical_get.hpp"
 
@@ -279,16 +280,31 @@ init_local(ExecutionContext &, TableFunctionInitInput &input, GlobalTableFunctio
     return make_uniq<CTableLocalData>(std::move(cdata));
 }
 
+struct VortexWaitTask final : AsyncTask {
+    explicit VortexWaitTask(void *ffi_local) : ffi_local(ffi_local) {
+    }
+    void Execute() override {
+        duckdb_table_function_wait_and_fetch(ffi_local);
+    }
+    void *ffi_local;
+};
+
 void function(ClientContext &, TableFunctionInput &input, DataChunk &output) {
     void *const ffi_global = input.global_state->Cast<CTableGlobalData>().ffi_data->DataPtr();
     void *const ffi_local = input.local_state->Cast<CTableLocalData>().ffi_data->DataPtr();
 
     duckdb_data_chunk chunk = reinterpret_cast<duckdb_data_chunk>(&output);
     duckdb_vx_error error_out = nullptr;
-    duckdb_table_function_scan(ffi_global, ffi_local, chunk, &error_out);
+    const bool blocked_on_io = duckdb_table_function_scan(ffi_global, ffi_local, chunk, &error_out);
+
     if (error_out) {
         throw InvalidInputException(IntoErrString(error_out));
     }
+    if (blocked_on_io) {
+        vector<unique_ptr<AsyncTask>> tasks(1);
+        tasks[0] = make_uniq<VortexWaitTask>(ffi_local);
+        input.async_result = AsyncResult(std::move(tasks));
+    }
 }
 
 using FilterVec = vector<unique_ptr<Expression>>;
 
@@ -48,11 +48,13 @@ bool duckdb_table_function_pushdown_projection_aggregates(void *bind_data,
                                                           duckdb_vx_error *error_out);
 
 extern
-void duckdb_table_function_scan(void *global_init_data,
+bool duckdb_table_function_scan(void *global_init_data,
                                 void *local_init_data,
                                 duckdb_data_chunk output,
                                 duckdb_vx_error *error_out);
 
+extern void duckdb_table_function_wait_and_fetch(void *local_init_data);
+
 extern bool duckdb_table_function_pushdown_expression(duckdb_vx_expr expr);
 
 extern
 
@@ -30,6 +30,7 @@ use crate::duckdb::TableInitInput;
 use crate::duckdb::try_or;
 use crate::duckdb::try_or_null;
 use crate::table_function::Cardinality;
+use crate::table_function::ScanResult;
 use crate::table_function::TableFunctionBind;
 use crate::table_function::TableFunctionGlobal;
 use crate::table_function::TableFunctionLocal;
@@ -45,6 +46,7 @@ use crate::table_function::scan;
 use crate::table_function::statistics;
 use crate::table_function::table_scan_progress;
 use crate::table_function::to_string;
+use crate::table_function::wait_and_fetch;
 
 #[unsafe(no_mangle)]
 unsafe extern "C-unwind" fn duckdb_table_function_to_string(
@@ -148,27 +150,34 @@ unsafe extern "C-unwind" fn duckdb_table_function_scan(
     local_init_data: *mut c_void,
     output: cpp::duckdb_data_chunk,
     error_out: *mut cpp::duckdb_vx_error,
-) {
+) -> bool {
     let global_init_data = unsafe { global_init_data.cast::<TableFunctionGlobal>().as_ref() }
         .vortex_expect("global_init_data null pointer");
     let local_init_data = unsafe { local_init_data.cast::<TableFunctionLocal>().as_mut() }
         .vortex_expect("local_init_data null pointer");
     let data_chunk = unsafe { DataChunk::borrow_mut(output) };
 
     match scan(local_init_data, global_init_data, data_chunk) {
-        Ok(()) => {
-            // The data chunk is already filled by the function.
-            // No need to do anything here.
+        Ok(result) => result == ScanResult::BlockedOnIO,
+        Err(e) => {
+            unsafe {
+                error_out.write(cpp::duckdb_vx_error_create(
+                    e.to_string().as_ptr().cast(),
+                    e.to_string().len(),
+                ));
+            }
+            false
         }
-        Err(e) => unsafe {
-            error_out.write(cpp::duckdb_vx_error_create(
-                e.to_string().as_ptr().cast(),
-                e.to_string().len(),
-            ));
-        },
     }
 }
 
+#[unsafe(no_mangle)]
+unsafe extern "C-unwind" fn duckdb_table_function_wait_and_fetch(local_init_data: *mut c_void) {
+    let local_init_data = unsafe { local_init_data.cast::<TableFunctionLocal>().as_mut() }
+        .vortex_expect("local_init_data null pointer");
+    wait_and_fetch(local_init_data);
+}
+
 #[unsafe(no_mangle)]
 pub unsafe extern "C-unwind" fn duckdb_table_function_pushdown_expression(
     expr: cpp::duckdb_vx_expr,
 
@@ -41,6 +41,7 @@ use vortex::file::v2::FileStatsLayoutReader;
 use vortex::io::kanal_ext::KanalExt as _;
 use vortex::io::runtime::BlockingRuntime as _;
 use vortex::io::runtime::current::ThreadSafeIterator;
+use vortex::io::runtime::current::TryRecv;
 use vortex::layout::scan::multi::MultiLayoutChild;
 use vortex::layout::scan::multi::MultiLayoutDataSource;
 use vortex::metrics::tracing::get_global_labels;
@@ -131,6 +132,7 @@ impl<'a> TableInitInput<'a> {
 
 type ScanItem = VortexResult<(ArrayRef, Arc<ConversionCache>)>;
 type DataSourceIterator = ThreadSafeIterator<ScanItem>;
+type ScanItemFuture = BoxFuture<'static, Option<ScanItem>>;
 
 pub struct TableFunctionGlobal {
     iterator: DataSourceIterator,
@@ -158,8 +160,10 @@ pub struct TableFunctionLocal {
     exporter: Option<ArrayExporter>,
     partition_index: u64,
     file_index: usize,
-    // Aggregate scan accumulated partials. Empty for non-aggregate scan
+    /// Aggregate scan accumulated partials. Empty for non-aggregate scan
     partials: Vec<Box<dyn DynAccumulator>>,
+    /// In progress next item
+    pending: Option<ScanItemFuture>,
 }
 
 pub struct PartitionData {
@@ -422,6 +426,7 @@ pub fn init_local(
         partition_index: 0,
         file_index: 0,
         partials,
+        pending: None,
     }
 }
 
@@ -443,11 +448,43 @@ fn convert_result(array: ArrayRef, ctx: &mut ExecutionCtx) -> VortexResult<Struc
     })
 }
 
+/// Write aggregation output row to "chunk"
+fn finalize_aggregate(
+    global_state: &TableFunctionGlobal,
+    chunk: &mut DataChunkRef,
+) -> VortexResult<()> {
+    // 0 means we're the last thread, u64::MAX means output is written.
+    // is_err() means CAS didn't succeed
+    if global_state
+        .pending
+        .compare_exchange(0, u64::MAX, Ordering::AcqRel, Ordering::Relaxed)
+        .is_err()
+    {
+        return Ok(());
+    }
+
+    let mut accumulators = global_state.partials.lock();
+    let row_count = global_state.row_count.load(Ordering::Acquire) as i64;
+    let mut accum_iter = accumulators.iter_mut();
+    for (idx, aggregate) in global_state.aggregates.iter().enumerate() {
+        let value = match aggregate {
+            ColumnAggregate::Real { .. } => {
+                let accum = accum_iter.next().vortex_expect("partial for real agg");
+                Value::try_from(accum.finish()?)?
+            }
+            ColumnAggregate::CountStar => Value::from(row_count),
+        };
+        chunk.get_vector_mut(idx).reference_value(&value);
+    }
+    chunk.set_len(1);
+    Ok(())
+}
+
 fn scan_aggregate(
     local_state: &mut TableFunctionLocal,
     global_state: &TableFunctionGlobal,
     chunk: &mut DataChunkRef,
-) -> VortexResult<()> {
+) -> VortexResult<ScanResult> {
     let aggregates_len = global_state.aggregates.len();
     // seen[k] = output column for requested column k.
     // If min(x), max(x), avg(y) are requested, seen = { 0: 0, 1: 1}
@@ -468,32 +505,13 @@ fn scan_aggregate(
 
     let mut ctx = SESSION.create_execution_ctx();
     loop {
-        let Some(result) = local_state.iterator.next() else {
-            // 0 means we're the last thread, u64::MAX means output is written.
-            // is_err() means CAS didn't succeed
-            if global_state
-                .pending
-                .compare_exchange(0, u64::MAX, Ordering::AcqRel, Ordering::Relaxed)
-                .is_err()
-            {
-                return Ok(());
+        let result = match poll_next_item(local_state) {
+            NextItem::Ready(item) => item,
+            NextItem::BlockedOnIO => return Ok(ScanResult::BlockedOnIO),
+            NextItem::None => {
+                finalize_aggregate(global_state, chunk)?;
+                return Ok(ScanResult::Exported);
             }
-
-            let mut accumulators = global_state.partials.lock();
-            let row_count = global_state.row_count.load(Ordering::Acquire) as i64;
-            let mut accum_iter = accumulators.iter_mut();
-            for (idx, aggregate) in global_state.aggregates.iter().enumerate() {
-                let value = match aggregate {
-                    ColumnAggregate::Real { .. } => {
-                        let accum = accum_iter.next().vortex_expect("partial for real agg");
-                        Value::try_from(accum.finish()?)?
-                    }
-                    ColumnAggregate::CountStar => Value::from(row_count),
-                };
-                chunk.get_vector_mut(idx).reference_value(&value);
-            }
-            chunk.set_len(1);
-            return Ok(());
         };
         let array = convert_result(result?.0, &mut ctx)?;
 
@@ -517,22 +535,77 @@ fn scan_aggregate(
     }
 }
 
+enum NextItem {
+    Ready(ScanItem),
+    None,
+    BlockedOnIO,
+}
+
+/// Poll next item without blocking caller thread
+fn poll_next_item(local_state: &mut TableFunctionLocal) -> NextItem {
+    if let Some(future) = local_state.pending.as_mut() {
+        let mut ctx = Context::from_waker(std::task::Waker::noop());
+        return match future.poll_unpin(&mut ctx) {
+            Poll::Ready(item) => {
+                local_state.pending = None;
+                if let Some(item) = item {
+                    NextItem::Ready(item)
+                } else {
+                    NextItem::None
+                }
+            }
+            Poll::Pending => NextItem::BlockedOnIO,
+        };
+    }
+
+    match local_state.iterator.try_recv() {
+        TryRecv::Item(item) => NextItem::Ready(item),
+        TryRecv::Closed => NextItem::None,
+        TryRecv::Empty => {
+            let rx = local_state.iterator.receiver();
+            local_state.pending = Some(Box::pin(async move { rx.recv().await.ok() }));
+            NextItem::BlockedOnIO
+        }
+    }
+}
+
+/// Finish array receive started by scan()
+pub fn wait_and_fetch(local_state: &mut TableFunctionLocal) {
+    if let Some(mut future) = local_state.pending.take() {
+        // Duckdb calls this on a background task thread pool which is separate
+        // from table function processing pool. This call therefore doesn't
+        // block the query worker
+        let item = RUNTIME.block_on(&mut future);
+        // When this assignment is happening, thread related to "local_state"
+        // is suspended. It will be resumed only when wait_and_fetch returns
+        // so there's no race condition
+        local_state.pending = Some(Box::pin(std::future::ready(item)));
+    };
+}
+
+#[derive(PartialEq, Eq)]
+pub enum ScanResult {
+    Exported,
+    BlockedOnIO,
+}
+
 pub fn scan(
     local_state: &mut TableFunctionLocal,
     global_state: &TableFunctionGlobal,
     chunk: &mut DataChunkRef,
-) -> VortexResult<()> {
+) -> VortexResult<ScanResult> {
     if !local_state.partials.is_empty() {
         return scan_aggregate(local_state, global_state, chunk);
     }
 
     loop {
         if local_state.exporter.is_none() {
             let mut ctx = SESSION.create_execution_ctx();
-            let Some(result) = local_state.iterator.next() else {
-                return Ok(());
+            let (array_result, conversion_cache) = match poll_next_item(local_state) {
+                NextItem::Ready(item) => item?,
+                NextItem::None => return Ok(ScanResult::Exported),
+                NextItem::BlockedOnIO => return Ok(ScanResult::BlockedOnIO),
             };
-            let (array_result, conversion_cache) = result?;
             local_state.file_index = conversion_cache.file_index;
             let array_result = convert_result(array_result, &mut ctx)?;
 
@@ -576,7 +649,7 @@ pub fn scan(
             .reference_value(&Value::from(local_state.file_index as u64));
     }
 
-    Ok(())
+    Ok(ScanResult::Exported)
 }
 
 /// Scan progress as a percentage (0.0–100.0).
 
@@ -139,6 +139,26 @@ pub struct ThreadSafeIterator<T> {
     driver: Arc<Mutex<Option<smol::Task<()>>>>,
 }
 
+pub enum TryRecv<T> {
+    Item(T),
+    Empty,
+    Closed,
+}
+
+impl<T> ThreadSafeIterator<T> {
+    pub fn receiver(&self) -> kanal::AsyncReceiver<T> {
+        self.results.clone()
+    }
+
+    pub fn try_recv(&self) -> TryRecv<T> {
+        match self.results.try_recv() {
+            Ok(Some(v)) => TryRecv::Item(v),
+            Ok(None) => TryRecv::Empty,
+            Err(_) => TryRecv::Closed,
+        }
+    }
+}
+
 // Manual clone implementation since `T` does not need to be `Clone`.
 impl<T> Clone for ThreadSafeIterator<T> {
     fn clone(&self) -> Self {