Function reduce_distinct_subgraph

fn reduce_distinct_subgraph(
    arrange: &ArrangeHandler,
    kv: impl IntoIterator<Item = ((Row, Row), i64, i64)>,
    _: SubgraphArg<'_>,
)

eval distinct reduce plan, output the distinct, and update the arrangement

invariant: it’is assumed kv’s time is always <= now, since it’s from a Collection Bundle, where future inserts are stored in arrange