Struct ChannelStreamSource 

pub struct ChannelStreamSource { /* private fields */ }

A streaming source that receives data through a channel.

This is the primary integration point between LaminarDB’s push-based Reactor and DataFusion’s pull-based query execution. Data is pushed into the source via BridgeSender, and DataFusion pulls it through the stream.

Important Usage Pattern

The sender must be taken (not cloned) to ensure proper channel closure:

// Create the source and take the sender
let source = ChannelStreamSource::new(schema);
let sender = source.take_sender().expect("sender available");

// Register with `DataFusion`
let provider = StreamingTableProvider::new("events", Arc::new(source));
ctx.register_table("events", Arc::new(provider))?;

// Push data from Reactor
sender.send(batch).await?;

// IMPORTANT: Drop the sender to close the channel before querying
drop(sender);

// Execute query
let df = ctx.sql("SELECT * FROM events").await?;
let results = df.collect().await?;

Thread Safety

The source is thread-safe and can be shared across threads. The sender can be cloned after being taken to allow multiple producers.

Implementations

impl ChannelStreamSource

pub fn new(schema: SchemaRef) -> Self

Creates a new channel stream source with default capacity.

Arguments

• schema - Schema of the RecordBatch instances that will be pushed

pub fn with_capacity(schema: SchemaRef, capacity: usize) -> Self

Creates a new channel stream source with specified capacity.

Arguments

• schema - Schema of the RecordBatch instances that will be pushed
• capacity - Maximum number of batches that can be buffered

pub fn with_ordering(self, ordering: Vec<SortColumn>) -> Self

Declares that this source produces data in the given sort order.

When set, DataFusion can elide SortExec for ORDER BY queries that match the declared ordering.

Arguments

• ordering - The columns that the output is sorted by

pub fn take_sender(&self) -> Option<BridgeSender>

Takes the sender for pushing batches into this source.

This method can only be called once. The sender is moved out of the source to ensure the caller has full ownership and can close the channel by dropping the sender.

The returned sender can be cloned to allow multiple producers.

Returns None if the sender was already taken.

pub fn sender(&self) -> Option<BridgeSender>

Returns a clone of the sender if it hasn’t been taken yet.

Warning: Using this method can lead to channel leak issues if the original sender is never dropped. Prefer take_sender() for proper channel lifecycle management.

pub fn reset(&self) -> BridgeSender

Resets the source with a new bridge and sender.

This is useful when you need to reuse the source after the previous stream has been consumed. Any data sent before the reset but not yet consumed will be lost.

Returns the new sender.

Trait Implementations

impl Debug for ChannelStreamSource

fn fmt(&self, f: &mut Formatter<'_>) -> Result

Formats the value using the given formatter.

impl StreamSource for ChannelStreamSource

fn schema(&self) -> SchemaRef

Returns the schema of records produced by this source.

fn output_ordering(&self) -> Option<Vec<SortColumn>>

Returns the output ordering of this source, if any.

fn stream( &self, projection: Option<Vec<usize>>, _filters: Vec<Expr>, ) -> Result<SendableRecordBatchStream, DataFusionError>

Creates a stream of RecordBatch instances.

fn supports_filters(&self, filters: &[Expr]) -> Vec<bool>

Returns which filters this source can apply.