Function build_export_dataflow

src/storage/src/render.rs:439–480

do the export dataflow thing

(
    timely_worker: &mut TimelyWorker,
    storage_state: &mut StorageState,
    id: GlobalId,
    description: StorageSinkDesc<CollectionMetadata, mz_repr::Timestamp>,
)

/// do the export dataflow thing
pub fn build_export_dataflow(
    timely_worker: &mut TimelyWorker,
    storage_state: &mut StorageState,
    id: GlobalId,
    description: StorageSinkDesc<CollectionMetadata, mz_repr::Timestamp>,
) {
    let worker_logging = timely_worker.logger_for("timely").map(Into::into);
    let debug_name = id.to_string();
    let name = format!("Source dataflow: {debug_name}");
    timely_worker.dataflow_core(&name, worker_logging, Box::new(()), |_, scope| {
        let scope = scope.with_label();

        let mut tokens = vec![];
        let (health_stream, sink_tokens) =
            crate::render::sinks::render_sink(scope, storage_state, id, &description);
        tokens.extend(sink_tokens);

        // Note that sinks also have only 1 active worker, which simplifies the work that
        // `health_operator` has to do internally.
        let health_token = crate::healthcheck::health_operator(
            scope,
            storage_state.now.clone(),
            [id].into_iter().collect(),
            id,
            "sink",
            health_stream,
            crate::healthcheck::DefaultWriter {
                command_tx: storage_state.internal_cmd_tx.clone(),
                updates: Rc::clone(&storage_state.shared_status_updates),
            },
            storage_state
                .storage_configuration
                .parameters
                .record_namespaced_errors,
            dyncfgs::STORAGE_SUSPEND_AND_RESTART_DELAY
                .get(storage_state.storage_configuration.config_set()),
        );
        tokens.push(health_token);

        storage_state.sink_tokens.insert(id, tokens);
    });
}

pub(crate) fn build_oneshot_ingestion_dataflow(
    timely_worker: &mut TimelyWorker,