Implementation:Lance format Lance BackgroundIterator

Knowledge Sources	Lance
Domains	DataFusion_Integration, Query_Execution
Last Updated	2026-02-08 19:33 GMT

Overview

The BackgroundIterator module wraps a synchronous iterator as an asynchronous futures::Stream by executing iterator calls on a Tokio blocking thread.

Description

Many Arrow record batch readers implement Iterator with blocking I/O. Lance's async-first architecture requires Stream-based interfaces. This module bridges that gap with:

• BackgroundIterator -- A pin_project-annotated struct that wraps any Iterator + Send + 'static and implements futures::Stream. On each poll, it dispatches iter.next() to a Tokio spawn_blocking task, ensuring that blocking I/O does not stall the async runtime.

The internal state machine (BackgroundIterState) has three states:

• Current -- The iterator is available and ready for the next spawn_blocking call.
• Running -- A blocking task is in flight; the stream awaits its JoinHandle.
• Empty -- The iterator has been exhausted or the stream has completed.

Key properties:

• The size_hint from the underlying iterator is preserved.
• The stream is not fused -- callers should use .fuse() if needed.
• Panics in the blocking thread are propagated to the polling task via resume_unwind.
• The iterator ownership transfers between the main task and the blocking thread on each step, avoiding any shared mutable state.

Usage

Use BackgroundIterator when you need to convert a blocking RecordBatchReader or other synchronous iterator into a Stream for use with Lance's async pipelines. This is primarily used by reader_to_stream in the utils module.

Code Reference

Source Location

rust/lance-datafusion/src/utils/background_iterator.rs

Signature

#[pin_project]
pub struct BackgroundIterator<I: Iterator + Send + 'static> {
    state: BackgroundIterState<I>,
}

impl<I: Iterator + Send + 'static> BackgroundIterator<I> {
    pub fn new(iter: I) -> Self;
}

impl<I: Iterator + Send + 'static> Stream for BackgroundIterator<I>
where
    I::Item: Send + 'static,
{
    type Item = I::Item;
    fn poll_next(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>>;
    fn size_hint(&self) -> (usize, Option<usize>);
}

Import

use lance_datafusion::utils::background_iterator::BackgroundIterator;

I/O Contract

Input	Type	Description
iter	I: Iterator + Send + 'static	A synchronous iterator to be executed on a background blocking thread

Output	Type	Description
Stream	impl Stream<Item = I::Item>	An asynchronous stream that yields items from the iterator without blocking the async runtime

Usage Examples

use lance_datafusion::utils::background_iterator::BackgroundIterator;
use futures::StreamExt;

// Wrap a blocking RecordBatchReader as a stream
let reader: Box<dyn RecordBatchReader + Send> = get_reader();
let stream = BackgroundIterator::new(reader)
    .fuse()
    .map_err(DataFusionError::from);

// Consume the stream asynchronously
while let Some(batch_result) = stream.next().await {
    let batch = batch_result?;
    // Process batch...
}

Related Pages

• Lance_format_Lance_StreamingWriteSource -- Uses BackgroundIterator in reader_to_stream for converting readers to streams