Function window_aggr_no_list

src/expr/src/relation/func.rs:1313–1363

Like `window_aggr`, but doesn't perform the final wrapping in a list, returning an Iterator instead.

(
    input_datums: I,
    callers_temp_storage: &'b RowArena,
    wrapped_aggregate: &AggregateFunc,
    order_by: &[ColumnOrder],
    window_frame: &WindowFrame,
)

/// Like `window_aggr`, but doesn't perform the final wrapping in a list, returning an Iterator
/// instead.
fn window_aggr_no_list<'a: 'b, 'b, I, A>(
    input_datums: I,
    callers_temp_storage: &'b RowArena,
    wrapped_aggregate: &AggregateFunc,
    order_by: &[ColumnOrder],
    window_frame: &WindowFrame,
) -> impl Iterator<Item = Datum<'b>>
where
    I: IntoIterator<Item = Datum<'a>>,
    A: OneByOneAggr,
{
    // Sort the datums according to the ORDER BY expressions and return the ((OriginalRow, InputValue), OrderByRow) record
    // The OrderByRow is kept around because it is required to compute the peer groups in RANGE mode
    let datums = order_aggregate_datums_with_rank(input_datums, order_by);

    // Decode the input (OriginalRow, InputValue) into separate datums, while keeping the OrderByRow
    let size_hint = datums.size_hint().0;
    let mut args: Vec<Datum> = Vec::with_capacity(size_hint);
    let mut original_rows: Vec<Datum> = Vec::with_capacity(size_hint);
    let mut order_by_rows = Vec::with_capacity(size_hint);
    for (d, order_by_row) in datums.into_iter() {
        let mut iter = d.unwrap_list().iter();
        let original_row = iter.next().unwrap();
        let arg = iter.next().unwrap();
        order_by_rows.push(order_by_row);
        original_rows.push(original_row);
        args.push(arg);
    }

    let results = window_aggr_inner::<A>(
        args,
        &order_by_rows,
        wrapped_aggregate,
        order_by,
        window_frame,
        callers_temp_storage,
    );

    callers_temp_storage.reserve(results.len());
    results
        .into_iter()
        .zip_eq(original_rows)
        .map(|(result_value, original_row)| {
            callers_temp_storage.make_datum(|packer| {
                packer.push_list_with(|packer| {
                    packer.push(result_value);
                    packer.push(original_row);
                });
            })
        })
}

fn window_aggr_inner<'a, A>(
    mut args: Vec<Datum<'a>>,