Function expand_counts

src/expr/src/relation/func.rs:2014–2022

Expands an iterator of `(datum, diff)` into one `datum` per unit of `diff`. A non-positive `diff` contributes no copies. This is used by aggregates that are sensitive to multiplicity (e.g. `sum`), to recover a flat datum stream from the count-aware surface.

(datums: I)

/// are sensitive to multiplicity (e.g. `sum`), to recover a flat datum stream
/// from the count-aware surface.
fn expand_counts<'a, I>(datums: I) -> impl Iterator<Item = Datum<'a>>
where
    I: IntoIterator<Item = (Datum<'a>, Diff)>,
{
    datums.into_iter().flat_map(|(datum, diff)| {
        let copies = usize::try_from(diff.into_inner()).unwrap_or(0);
        std::iter::repeat(datum).take(copies)
    })
}

impl AggregateFunc {
    /// Whether this aggregate's result is independent of the multiplicity of its