fn build_distinct_retractions<G, T>(collection: Collection<G, (Row, Row)>) -> Collection<G, Row>
where
    G: Scope,
    G::Timestamp: Lattice + Refines<T>,
    T: Timestamp + Lattice,
{
    let negated_result = collection.reduce_named("DistinctBy Retractions", {
        |key, input, output| {
            output.push((key.clone(), -1));
            output.extend(
                input
                    .iter()
                    .map(|(values, count)| ((*values).clone(), *count)),
            );
        }
    });
    use timely::dataflow::operators::Map;
    negated_result
        .negate()
        .concat(&collection)
        .consolidate()
        .inner
        .map(|((k, _), time, count)| (k, time, count))
        .as_collection()
}
/// Build the dataflow to compute the set of distinct keys.
///
/// This implementation maintains the rows that don't appear in the output.
Build the dataflow to compute the set of distinct keys.
This implementation maintains the rows that don't appear in the output.
Build the dataflow to compute the set of distinct keys . This implementation maintains the rows that don ' t appear in the output .
fn build_distinct_retractions<G, T>(collection: Collection<G, (Row, Row)>) -> Collection<G, Row>
where
    G: Scope,
    G::Timestamp: Lattice + Refines<T>,
    T: Timestamp + Lattice,
{
    let negated_result = collection.reduce_named("DistinctBy Retractions", {
        |key, input, output| {
            output.push((key.clone(), -1));
            output.extend(
                input
                    .iter()
                    .map(|(values, count)| ((*values).clone(), *count)),
            );
        }
    });
    use timely::dataflow::operators::Map;
    negated_result
        .negate()
        .concat(&collection)
        .consolidate()
        .inner
        .map(|((k, _), time, count)| (k, time, count))
        .as_collection()
}
fn build_distinct_retractions < G , T > ( collection : Collection < G , ( Row , Row ) > ) -> Collection < G , Row > where G : Scope , G :: Timestamp : Lattice + Refines < T > , T : Timestamp + Lattice , { let negated_result = collection . reduce_named ( " " , { | key , input , output | { output . push ( ( key . clone ( ) , - 1 ) ) ; output . extend ( input . iter ( ) . map ( | ( values , count ) | ( ( * values ) . clone ( ) , * count ) ) , ) ; } } ) ; use timely :: dataflow :: operators :: Map ; negated_result . negate ( ) . concat ( & collection ) . consolidate ( ) . inner . map ( | ( ( k , _ ) , time , count ) | ( k , time , count ) ) . as_collection ( ) }
Build the dataflow to compute the set of distinct keys.
Build the dataflow to compute the set of distinct keys .
collection
Collection<G, (Row, Row)>
collection
Collection<G, (Row, Row)>