Struct WindowedSortStream

pub struct WindowedSortStream {

    memory_pool: Arc<dyn MemoryPool>,
    in_progress: Vec<DfRecordBatch>,
    last_value: Option<Timestamp>,
    sorted_input_runs: Vec<DfSendableRecordBatchStream>,
    merge_stream: VecDeque<DfSendableRecordBatchStream>,
    merge_count: usize,
    working_idx: usize,
    input: DfSendableRecordBatchStream,
    is_terminated: bool,
    schema: SchemaRef,
    expression: PhysicalSortExpr,
    fetch: Option<usize>,
    produced: usize,
    batch_size: usize,
    all_avail_working_range: Vec<(TimeRange, BTreeSet<usize>)>,
    ranges: Vec<PartitionRange>,
    metrics: BaselineMetrics,
}

The core logic of merging sort multiple sorted ranges

the flow of data is:

input --check if sorted--> in_progress --find sorted run--> sorted_input_runs --call merge sort--> merge_stream --> output

Fields

memory_pool: Arc<dyn MemoryPool>

Memory pool for this stream

in_progress: Vec<DfRecordBatch>

currently assembling RecordBatches, will be put to sort_partition_rbs when it’s done

last_value: Option<Timestamp>

last Timestamp of the last input RecordBatch in in_progress, use to found partial sorted run’s boundary

sorted_input_runs: Vec<DfSendableRecordBatchStream>

Current working set of PartitionRange sorted RecordBatches

merge_stream: VecDeque<DfSendableRecordBatchStream>

Merge-sorted result streams, should be polled to end before start a new merge sort again

merge_count: usize

The number of times merge sort has been called

working_idx: usize

Index into current working_range in all_avail_working_range

input: DfSendableRecordBatchStream

input stream assumed reading in order of PartitionRange

is_terminated: bool

Whether this stream is terminated. For reasons like limit reached or input stream is done.

schema: SchemaRef

Output Schema, which is the same as input schema, since this is a sort plan

expression: PhysicalSortExpr

Physical sort expressions(that is, sort by timestamp)

fetch: Option<usize>

Optional number of rows to fetch. Stops producing rows after this fetch

produced: usize

number of rows produced

batch_size: usize

Resulting Stream(merge_stream)’s batch size, merely a suggestion

all_avail_working_range: Vec<(TimeRange, BTreeSet<usize>)>

All available working ranges and their corresponding working set

working ranges promise once input stream get a value out of current range, future values will never be in this range

ranges: Vec<PartitionRange>

The input partition ranges

metrics: BaselineMetrics

Execution metrics

Implementations

impl WindowedSortStream

pub fn new( context: Arc<TaskContext>, exec: &WindowedSortExec, input: DfSendableRecordBatchStream, partition: usize, ) -> Self

impl WindowedSortStream

fn check_subset_ranges(&self, cur_range: &TimeRange)

fn poll_result_stream( &mut self, cx: &mut Context<'_>, ) -> Poll<Option<Result<DfRecordBatch>>>

Poll the next RecordBatch from the merge-sort’s output stream

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

The core logic of merging sort multiple sorted ranges

We try to maximize the number of sorted runs we can merge in one go, while emit the result as soon as possible.

fn push_batch(&mut self, batch: DfRecordBatch)

fn try_concat_batch( &mut self, batch: DfRecordBatch, run_info: &SucRun<Timestamp>, opt: Option<SortOptions>, ) -> Result<()>

Try to concat the input batch to the current in_progress run

if the input batch is not sorted, build old run to stream and start a new run with new batch

fn get_working_range(&self) -> Option<TimeRange>

Get the current working range

fn set_next_working_range(&mut self)

Set current working range to the next working range

fn build_sorted_stream(&mut self) -> Result<()>

make in_progress as a new DfSendableRecordBatchStream and put into sorted_input_runs

fn start_new_merge_sort(&mut self) -> Result<()>

Start merging sort the current working set

fn remaining_fetch(&self) -> Option<usize>

Remaining number of rows to fetch, if no fetch limit, return None if fetch limit is reached, return Some(0)

Trait Implementations

impl RecordBatchStream for WindowedSortStream

fn schema(&self) -> SchemaRef

Returns the schema of this RecordBatchStream.

impl Stream for WindowedSortStream

type Item = Result<RecordBatch, DataFusionError>

Values yielded by the stream.

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

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.