Struct PartSortStream

struct PartSortStream {

    reservation: MemoryReservation,
    buffer: PartSortBuffer,
    expression: PhysicalSortExpr,
    limit: Option<usize>,
    produced: usize,
    input: DfSendableRecordBatchStream,
    input_complete: bool,
    schema: SchemaRef,
    partition_ranges: Vec<PartitionRange>,
    partition: usize,
    cur_part_idx: usize,
    evaluating_batch: Option<DfRecordBatch>,
    metrics: BaselineMetrics,
    context: Arc<TaskContext>,
    root_metrics: ExecutionPlanMetricsSet,
}

Fields

reservation: MemoryReservation

Memory pool for this stream

buffer: PartSortBuffer

expression: PhysicalSortExpr

limit: Option<usize>

produced: usize

input: DfSendableRecordBatchStream

input_complete: bool

schema: SchemaRef

partition_ranges: Vec<PartitionRange>

partition: usize

cur_part_idx: usize

evaluating_batch: Option<DfRecordBatch>

metrics: BaselineMetrics

context: Arc<TaskContext>

root_metrics: ExecutionPlanMetricsSet

Implementations

impl PartSortStream

fn new( context: Arc<TaskContext>, sort: &PartSortExec, limit: Option<usize>, input: DfSendableRecordBatchStream, partition_ranges: Vec<PartitionRange>, partition: usize, ) -> Result<Self>

impl PartSortStream

fn check_in_range( &self, sort_column: &ArrayRef, min_max_idx: (usize, usize), ) -> Result<()>

check whether the sort column’s min/max value is within the partition range

fn try_find_next_range(&self, sort_column: &ArrayRef) -> Result<Option<usize>>

Try find data whose value exceeds the current partition range.

Returns None if no such data is found, and Some(idx) where idx points to the first data that exceeds the current partition range.

fn push_buffer(&mut self, batch: DfRecordBatch) -> Result<()>

fn sort_buffer(&mut self) -> Result<DfRecordBatch>

Sort and clear the buffer and return the sorted record batch

this function will return a empty record batch if the buffer is empty

fn sort_all_buffer(&mut self) -> Result<DfRecordBatch>

Internal method for sorting All buffer (without limit).

fn sort_top_buffer(&mut self) -> Result<DfRecordBatch>

Internal method for sorting Top buffer (with limit).

fn split_batch(&mut self, batch: DfRecordBatch) -> Result<Option<DfRecordBatch>>

Try to split the input batch if it contains data that exceeds the current partition range.

When the input batch contains data that exceeds the current partition range, this function will split the input batch into two parts, the first part is within the current partition range will be merged and sorted with previous buffer, and the second part will be registered to evaluating_batch for next polling.

Returns None if the input batch is empty or fully within the current partition range, and Some(batch) otherwise.

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

Trait Implementations

impl RecordBatchStream for PartSortStream

fn schema(&self) -> SchemaRef

Returns the schema of this RecordBatchStream.

impl Stream for PartSortStream

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.