pub fn new_from(source: Collection<G, D, R, C>, step: <G::Timestamp as Timestamp>::Summary) -> Self {
        let (feedback, updates) = source.inner.scope().feedback(step.clone());
        let collection = Collection::<G, D, R, C>::new(updates).concat(&source);
        Variable { collection, feedback, source: Some(source), step }
    }
Creates a new `Variable` from a supplied `source` stream.
differential-dataflow/src/operators/iterate.rs