Function persist_source_core

pub fn persist_source_core<'g, G>(
    scope: &Child<'g, G, (<G as ScopeParent>::Timestamp, Subtime)>,
    source_id: GlobalId,
    persist_clients: Arc<PersistClientCache>,
    metadata: CollectionMetadata,
    read_schema: Option<RelationDesc>,
    as_of: Option<Antichain<Timestamp>>,
    snapshot_mode: SnapshotMode,
    until: Antichain<Timestamp>,
    map_filter_project: Option<&mut MfpPlan>,
    flow_control: Option<FlowControl<Child<'g, G, (<G as ScopeParent>::Timestamp, Subtime)>>>,
    listen_sleep: Option<impl Fn() -> RetryParameters + 'static>,
    start_signal: impl Future<Output = ()> + 'static,
    error_handler: impl FnOnce(String) -> Pin<Box<dyn Future<Output = ()>>> + 'static,
) -> (Stream<Child<'g, G, (<G as ScopeParent>::Timestamp, Subtime)>, (Result<Row, DataflowError>, (Timestamp, Subtime), Diff)>, Vec<PressOnDropButton>)
where
    G: Scope<Timestamp = Timestamp>,

Creates a new source that reads from a persist shard, distributing the work of reading data to all timely workers.

All times emitted will have been advanced by the given as_of frontier.