Creates the described dataflow and initializes state for its output. Only materialized views and subscribes are allowed to have a `target_replica`. Panics if called with a dataflow description that has index exports when `target_replica` is set.
(
&mut self,
instance_id: ComputeInstanceId,
mut dataflow: DataflowDescription<mz_compute_types::plan::Plan, ()>,
target_replica: Option<ReplicaId>,
)
| 779 | /// Panics if called with a dataflow description that has index exports |
| 780 | /// when `target_replica` is set. |
| 781 | pub fn create_dataflow( |
| 782 | &mut self, |
| 783 | instance_id: ComputeInstanceId, |
| 784 | mut dataflow: DataflowDescription<mz_compute_types::plan::Plan, ()>, |
| 785 | target_replica: Option<ReplicaId>, |
| 786 | ) -> Result<(), DataflowCreationError> { |
| 787 | use DataflowCreationError::*; |
| 788 | |
| 789 | let instance = self.instance(instance_id)?; |
| 790 | |
| 791 | // Validation: target replica |
| 792 | if let Some(replica_id) = target_replica { |
| 793 | if !instance.replicas.contains(&replica_id) { |
| 794 | return Err(ReplicaMissing(replica_id)); |
| 795 | } |
| 796 | assert!( |
| 797 | dataflow.exported_index_ids().next().is_none(), |
| 798 | "Replica-targeted indexes are not supported" |
| 799 | ); |
| 800 | } |
| 801 | |
| 802 | // Validation: as_of |
| 803 | let as_of = dataflow.as_of.as_ref().ok_or(MissingAsOf)?; |
| 804 | if as_of.is_empty() && dataflow.subscribe_ids().next().is_some() { |
| 805 | return Err(EmptyAsOfForSubscribe); |
| 806 | } |
| 807 | if as_of.is_empty() && dataflow.copy_to_ids().next().is_some() { |
| 808 | return Err(EmptyAsOfForCopyTo); |
| 809 | } |
| 810 | |
| 811 | // Validation: input collections |
| 812 | let storage_ids = dataflow.imported_source_ids().collect(); |
| 813 | let mut import_read_holds = self.storage_collections.acquire_read_holds(storage_ids)?; |
| 814 | for id in dataflow.imported_index_ids() { |
| 815 | let read_hold = instance.acquire_read_hold(id)?; |
| 816 | import_read_holds.push(read_hold); |
| 817 | } |
| 818 | for hold in &import_read_holds { |
| 819 | if PartialOrder::less_than(as_of, hold.since()) { |
| 820 | return Err(SinceViolation(hold.id())); |
| 821 | } |
| 822 | } |
| 823 | |
| 824 | // Validation: storage sink collections |
| 825 | for id in dataflow.persist_sink_ids() { |
| 826 | if self.storage_collections.check_exists(id).is_err() { |
| 827 | return Err(CollectionMissing(id)); |
| 828 | } |
| 829 | } |
| 830 | let time_dependence = self |
| 831 | .determine_time_dependence(instance_id, &dataflow) |
| 832 | .expect("must exist"); |
| 833 | |
| 834 | let instance = self.instance_mut(instance_id).expect("validated"); |
| 835 | |
| 836 | let mut shared_collection_state = BTreeMap::new(); |
| 837 | for id in dataflow.export_ids() { |
| 838 | let shared = SharedCollectionState::new(as_of.clone()); |
no test coverage detected