Implementation:Lance format Lance StreamingWriteSource

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

Overview

The StreamingWriteSource module defines a trait and utility functions for converting various record batch sources into DataFusion streams, along with extension traits for execution plan metrics.

Description

This module provides the abstraction layer between different record batch input types and the SendableRecordBatchStream type required by Lance's write pipeline. Key components include:

• StreamingWriteSource trait -- An async trait that unifies multiple record batch sources behind a common interface:
  • into_stream_and_schema(self) -- Consumes the source, peeks the first batch to infer dictionary encodings, validates the schema, and returns both the stream and a Lance Schema.
  • arrow_schema(&self) -- Returns the Arrow schema without consuming the source.
  • into_stream(self) -- Converts the source into a SendableRecordBatchStream, running blocking iteration on a background thread.

Implementations are provided for:

• ArrowArrayStreamReader -- FFI stream readers from Arrow C Data Interface.
• RecordBatchIterator<I> -- Any iterator of Result<RecordBatch, ArrowError>.
• Box<T: StreamingWriteSource> -- Boxed write sources.
• Box<dyn RecordBatchReader + Send> -- Trait object readers.
• SendableRecordBatchStream -- Pass-through for already-streaming sources.

• reader_to_stream -- A public function that converts a Box<dyn RecordBatchReader + Send> into a SendableRecordBatchStream using a BackgroundIterator for non-blocking execution.

• MetricsExt trait -- Extends DataFusion's MetricsSet with convenience methods: find_count, iter_counts, iter_gauges.

• ExecutionPlanMetricsSetExt trait -- Extends ExecutionPlanMetricsSet with factory methods: new_count, new_time, new_gauge.

• Metric constants -- Predefined metric names used throughout Lance's execution plans: IOPS_METRIC, REQUESTS_METRIC, BYTES_READ_METRIC, INDICES_LOADED_METRIC, PARTS_LOADED_METRIC, INDEX_COMPARISONS_METRIC, FRAGMENTS_SCANNED_METRIC, ROWS_SCANNED_METRIC, and others.

Usage

Use StreamingWriteSource when you need to:

• Accept diverse record batch sources (iterators, readers, FFI streams) in Lance's write APIs
• Convert blocking record batch readers to async streams for Lance's async-first architecture
• Work with execution plan metrics in custom Lance execution nodes

Code Reference

Source Location

rust/lance-datafusion/src/utils.rs

Signature

#[async_trait]
pub trait StreamingWriteSource: Send {
    async fn into_stream_and_schema(self) -> Result<(SendableRecordBatchStream, Schema)>
    where Self: Sized;
    fn arrow_schema(&self) -> SchemaRef;
    fn into_stream(self) -> SendableRecordBatchStream;
}

pub fn reader_to_stream(batches: Box<dyn RecordBatchReader + Send>) -> SendableRecordBatchStream;

pub trait MetricsExt {
    fn find_count(&self, name: &str) -> Option<Count>;
    fn iter_counts(&self) -> impl Iterator<Item = (impl AsRef<str>, &Count)>;
    fn iter_gauges(&self) -> impl Iterator<Item = (impl AsRef<str>, &Gauge)>;
}

pub trait ExecutionPlanMetricsSetExt {
    fn new_count(&self, name: &'static str, partition: usize) -> Count;
    fn new_time(&self, name: &'static str, partition: usize) -> Time;
    fn new_gauge(&self, name: &'static str, partition: usize) -> Gauge;
}

Import

use lance_datafusion::utils::{StreamingWriteSource, reader_to_stream, MetricsExt};

I/O Contract

Input	Type	Description
self	Various (RecordBatchReader, RecordBatchIterator, etc.)	A record batch source to be converted to a stream

Output	Type	Description
into_stream	SendableRecordBatchStream	An async stream of record batches
into_stream_and_schema	Result<(SendableRecordBatchStream, Schema)>	A stream paired with the inferred Lance schema (including dictionary info)

Usage Examples

use lance_datafusion::utils::StreamingWriteSource;
use arrow_array::{RecordBatch, RecordBatchIterator};

// Convert a RecordBatchIterator to a stream
let batches = vec![Ok(batch1), Ok(batch2)];
let reader = RecordBatchIterator::new(batches, schema.clone());
let stream = reader.into_stream();

// Get stream and inferred schema for writing
let (stream, lance_schema) = reader.into_stream_and_schema().await?;

Related Pages

• Lance_format_Lance_BackgroundIterator -- Background thread iterator used by reader_to_stream
• Lance_format_Lance_ExecPlans -- Execution plans that use the metric extension traits
• Lance_format_Lance_Chunker -- Batch chunking applied to streams produced by this module