Method render

src/storage/src/source/postgres.rs:126–268

Render the ingestion dataflow. This function only connects things together and contains no actual processing logic.

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

    /// Render the ingestion dataflow. This function only connects things together and contains no
    /// actual processing logic.
    fn render<'scope>(
        self,
        scope: Scope<'scope, MzOffset>,
        config: &RawSourceCreationConfig,
        resume_uppers: impl futures::Stream<Item = Antichain<MzOffset>> + 'static,
        _start_signal: impl std::future::Future<Output = ()> + 'static,
    ) -> (
        BTreeMap<
            GlobalId,
            StackedCollection<'scope, MzOffset, Result<SourceMessage, DataflowError>>,
        >,
        StreamVec<'scope, MzOffset, HealthStatusMessage>,
        StreamVec<'scope, MzOffset, Probe<MzOffset>>,
        Vec<PressOnDropButton>,
    ) {
        // Collect the source outputs that we will be exporting into a per-table map.
        let mut table_info = 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::Postgres(details) => details,
                // This is an export that doesn't need any data output to it.
                SourceExportDetails::None => continue,
                _ => panic!("unexpected source export details: {:?}", details),
            };
            let desc = details.table.clone();
            let casts = details.column_casts.clone();
            let resume_upper = Antichain::from_iter(
                config
                    .source_resume_uppers
                    .get(id)
                    .expect("all source exports must be present in source resume uppers")
                    .iter()
                    .map(MzOffset::decode_row),
            );
            let output = SourceOutputInfo {
                desc,
                projection: None,
                casts,
                resume_upper,
                export_id: id.clone(),
            };
            table_info
                .entry(output.desc.oid)
                .or_insert_with(BTreeMap::new)
                .insert(idx, output);
        }

        let metrics = config.metrics.get_postgres_source_metrics(config.id);

        let (snapshot_updates, rewinds, slot_ready, snapshot_err, snapshot_token) =
            snapshot::render(
                scope.clone(),
                config.clone(),