fn build_topk_negated_stage<G, Tr>(
    input: &Collection<G, (Row, Row), Diff>,
    order_key: Vec<ColumnOrder>,
    offset: usize,
    limit: Option<MirScalarExpr>,
    arity: usize
) -> Arranged<G, TraceAgent<Tr>>where
    G: Scope,
    G::Timestamp: Lattice + Columnation,
    Tr::ValOwned: MaybeValidatingRow<Row, Row>,
    Tr: Trace + for<'a> TraceReader<Key<'a> = DatumSeq<'a>, Time = G::Timestamp, Diff = Diff> + 'static,
    Tr::Batch: Batch,
    Tr::Batcher: Batcher<Item = ((Row, Tr::ValOwned), G::Timestamp, Diff)>,
    Arranged<G, TraceAgent<Tr>>: ArrangementSize,