Function window_aggr_inner

src/expr/src/relation/func.rs:1365–1692

(
    mut args: Vec<Datum<'a>>,
    order_by_rows: &Vec<Row>,
    wrapped_aggregate: &AggregateFunc,
    order_by: &[ColumnOrder],
    window_frame: &WindowFrame,
    temp_storage: &'a RowArena,
)

}

fn window_aggr_inner<'a, A>(
    mut args: Vec<Datum<'a>>,
    order_by_rows: &Vec<Row>,
    wrapped_aggregate: &AggregateFunc,
    order_by: &[ColumnOrder],
    window_frame: &WindowFrame,
    temp_storage: &'a RowArena,
) -> Vec<Datum<'a>>
where
    A: OneByOneAggr,
{
    let length = args.len();
    let mut result: Vec<Datum> = Vec::with_capacity(length);

    // In this degenerate case, all results would be `wrapped_aggregate.default()` (usually null).
    // However, this currently can't happen, because
    // - Groups frame mode is currently not supported;
    // - Range frame mode is currently supported only for the default frame, which includes the
    //   current row.
    soft_assert_or_log!(
        !((matches!(window_frame.units, WindowFrameUnits::Groups)
            || matches!(window_frame.units, WindowFrameUnits::Range))
            && !window_frame.includes_current_row()),
        "window frame without current row"
    );

    if (matches!(
        window_frame.start_bound,
        WindowFrameBound::UnboundedPreceding
    ) && matches!(window_frame.end_bound, WindowFrameBound::UnboundedFollowing))
        || (order_by.is_empty()
            && (matches!(window_frame.units, WindowFrameUnits::Groups)
                || matches!(window_frame.units, WindowFrameUnits::Range))
            && window_frame.includes_current_row())
    {
        // Either
        //  - UNBOUNDED frame in both directions, or
        //  - There is no ORDER BY and the frame is such that the current peer group is included.
        //    (The current peer group will be the whole partition if there is no ORDER BY.)
        // We simply need to compute the aggregate once, on the entire partition, and each input
        // row will get this one aggregate value as result.
        let result_value =
            wrapped_aggregate.eval(args.into_iter().map(|d| (d, Diff::ONE)), temp_storage);
        // Every row will get the above aggregate as result.
        for _ in 0..length {
            result.push(result_value);
        }
    } else {
        fn rows_between_unbounded_preceding_and_current_row<'a, A>(
            args: Vec<Datum<'a>>,
            result: &mut Vec<Datum<'a>>,
            mut one_by_one_aggr: A,
            temp_storage: &'a RowArena,
        ) where
            A: OneByOneAggr,
        {
            for current_arg in args.into_iter() {
                one_by_one_aggr.give(&current_arg);