Function persist_source 

fn persist_source<'s>(
    scope: Scope<'s, Timestamp>,
    sink_id: GlobalId,
    target: CollectionMetadata,
    compute_state: &ComputeState,
    start_signal: StartSignal,
) -> (OkErr<StreamVec<'s, Timestamp, (Row, Timestamp, Diff)>, StreamVec<'s, Timestamp, (DataflowError, Timestamp, Diff)>>, Vec<PressOnDropButton>)

Instantiate a persist source reading back the target collection.