Method try_ship_dataflow

src/adapter/src/coord.rs:4240–4258

Call into the compute controller to install a finalized dataflow, and initialize the read policies for its exported readable objects.

(
        &mut self,
        dataflow: DataflowDescription<Plan>,
        instance: ComputeInstanceId,
        target_replica: Option<ReplicaId>,
    )

    /// Call into the compute controller to install a finalized dataflow, and
    /// initialize the read policies for its exported readable objects.
    pub(crate) async fn try_ship_dataflow(
        &mut self,
        dataflow: DataflowDescription<Plan>,
        instance: ComputeInstanceId,
        target_replica: Option<ReplicaId>,
    ) -> Result<(), DataflowCreationError> {
        // We must only install read policies for indexes, not for sinks.
        // Sinks are write-only compute collections that don't have read policies.
        let export_ids = dataflow.exported_index_ids().collect();

        self.controller
            .compute
            .create_dataflow(instance, dataflow, target_replica)?;

        self.initialize_compute_read_policies(export_ids, instance, CompactionWindow::Default)
            .await;

        Ok(())
    }

    /// Call into the compute controller to allow writes to the specified IDs
    /// from the specified instance. Calling this function multiple times and