Struct RecursiveQueryStream 

struct RecursiveQueryStream {
    task_context: Arc<TaskContext>,
    work_table: Arc<WorkTable>,
    recursive_term: Arc<dyn ExecutionPlan>,
    static_stream: Option<SendableRecordBatchStream>,
    recursive_stream: Option<SendableRecordBatchStream>,
    schema: SchemaRef,
    buffer: Vec<RecordBatch>,
    reservation: MemoryReservation,
    _baseline_metrics: BaselineMetrics,
}

The actual logic of the recursive queries happens during the streaming process. A simplified version of the algorithm is the following:

buffer = []

while batch := static_stream.next(): buffer.push(batch) yield buffer

while buffer.len() > 0: sender, receiver = Channel() register_continuation(handle_name, receiver) sender.send(buffer.drain()) recursive_stream = recursive_term.execute() while batch := recursive_stream.next(): buffer.append(batch) yield buffer

Fields

task_context: Arc<TaskContext>

The context to be used for managing handlers & executing new tasks

work_table: Arc<WorkTable>

The working table state, representing the self referencing cte table

recursive_term: Arc<dyn ExecutionPlan>

The dynamic part (recursive term) as is (without being executed)

static_stream: Option<SendableRecordBatchStream>

The static part (static term) as a stream. If the processing of this part is completed, then it will be None.

recursive_stream: Option<SendableRecordBatchStream>

The dynamic part (recursive term) as a stream. If the processing of this part has not started yet, or has been completed, then it will be None.

schema: SchemaRef

The schema of the output.

buffer: Vec<RecordBatch>

In-memory buffer for storing a copy of the current results. Will be cleared after each iteration.

reservation: MemoryReservation

Tracks the memory used by the buffer

_baseline_metrics: BaselineMetrics

Implementations

impl RecursiveQueryStream

fn new( task_context: Arc<TaskContext>, work_table: Arc<WorkTable>, recursive_term: Arc<dyn ExecutionPlan>, static_stream: SendableRecordBatchStream, baseline_metrics: BaselineMetrics, ) -> Self

Create a new recursive query stream

fn push_batch( self: Pin<&mut Self>, batch: RecordBatch, ) -> Poll<Option<Result<RecordBatch>>>

Push a clone of the given batch to the in memory buffer, and then return a poll with it.

fn poll_next_iteration( self: Pin<&mut Self>, cx: &mut Context<'_>, ) -> Poll<Option<Result<RecordBatch>>>

Start polling for the next iteration, will be called either after the static term is completed or another term is completed. It will follow the algorithm above on to check whether the recursion has ended.

Trait Implementations

impl RecordBatchStream for RecursiveQueryStream

fn schema(&self) -> SchemaRef

Get the schema

impl Stream for RecursiveQueryStream

type Item = Result<RecordBatch, DataFusionError>

Values yielded by the stream.

fn poll_next( self: Pin<&mut Self>, cx: &mut Context<'_>, ) -> Poll<Option<Self::Item>>

Attempt to pull out the next value of this stream, registering the current task for wakeup if the value is not yet available, and returning None if the stream is exhausted.

fn size_hint(&self) -> (usize, Option<usize>)

Returns the bounds on the remaining length of the stream.