Method render

src/storage/src/source/sql_server.rs:107–230

(
        self,
        scope: Scope<'scope, Lsn>,
        config: &RawSourceCreationConfig,
        resume_uppers: impl futures::Stream<Item = Antichain<Lsn>> + 'static,
        _start_signal: impl F

    const STATUS_NAMESPACE: StatusNamespace = StatusNamespace::SqlServer;

    fn render<'scope>(
        self,
        scope: Scope<'scope, Lsn>,
        config: &RawSourceCreationConfig,
        resume_uppers: impl futures::Stream<Item = Antichain<Lsn>> + 'static,
        _start_signal: impl Future<Output = ()> + 'static,
    ) -> (
        // Timely Collection for each Source Export defined in the provided `config`.
        BTreeMap<GlobalId, StackedCollection<'scope, Lsn, Result<SourceMessage, DataflowError>>>,
        StreamVec<'scope, Lsn, HealthStatusMessage>,
        StreamVec<'scope, Lsn, Probe<Lsn>>,
        Vec<PressOnDropButton>,
    ) {
        // Collect the source outputs that we will be exporting.
        let mut source_outputs = BTreeMap::new();
        for (idx, (id, export)) in config.source_exports.iter().enumerate() {
            let SourceExport {
                details,
                storage_metadata,
                data_config: _,
            } = export;

            let details = match details {
                SourceExportDetails::SqlServer(details) => details,
                // This is an export that doesn't need any data output to it.
                SourceExportDetails::None => continue,
                other => unreachable!("unexpected source export details: {other:?}"),
            };

            let decoder = details
                .table
                .decoder(&storage_metadata.relation_desc)
                .expect("TODO handle errors");
            let upstream_desc = Arc::new(details.table.clone());
            let resume_upper = config
                .source_resume_uppers
                .get(id)
                .expect("missing resume upper")
                .iter()
                .map(Lsn::decode_row);

            let output_info = SourceOutputInfo {
                capture_instance: Arc::clone(&details.capture_instance),
                upstream_desc,
                decoder: Arc::new(decoder),
                resume_upper: Antichain::from_iter(resume_upper),
                partition_index: u64::cast_from(idx),
                initial_lsn: details.initial_lsn,
            };
            source_outputs.insert(*id, output_info);
        }

        let metrics = config
            .metrics
            .get_sql_server_source_metrics(config.id, config.worker_id);

        let (repl_updates, repl_errs, repl_token) = replication::render(
            scope.clone(),