Method add_collection

src/compute-client/src/controller/instance.rs:282–337

Add a collection to the instance state. # Panics Panics if a collection with the same ID exists already.

(
        &mut self,
        id: GlobalId,
        as_of: Antichain<Timestamp>,
        shared: SharedCollectionState,
        storage_dependencies: BTreeMap<GlobalId, ReadHold>,
        compute_depen

    ///
    /// Panics if a collection with the same ID exists already.
    fn add_collection(
        &mut self,
        id: GlobalId,
        as_of: Antichain<Timestamp>,
        shared: SharedCollectionState,
        storage_dependencies: BTreeMap<GlobalId, ReadHold>,
        compute_dependencies: BTreeMap<GlobalId, ReadHold>,
        replica_input_read_holds: Vec<ReadHold>,
        write_only: bool,
        storage_sink: bool,
        initial_as_of: Option<Antichain<Timestamp>>,
        refresh_schedule: Option<RefreshSchedule>,
        target_replica: Option<ReplicaId>,
    ) {
        // Add global collection state.
        let dependency_ids: Vec<GlobalId> = compute_dependencies
            .keys()
            .chain(storage_dependencies.keys())
            .copied()
            .collect();
        let introspection = CollectionIntrospection::new(
            id,
            self.introspection_tx.clone(),
            as_of.clone(),
            storage_sink,
            initial_as_of,
            refresh_schedule,
            dependency_ids,
        );
        let mut state = CollectionState::new(
            id,
            as_of.clone(),
            shared,
            storage_dependencies,
            compute_dependencies,
            Arc::clone(&self.read_hold_tx),
            introspection,
        );
        state.target_replica = target_replica;
        // If the collection is write-only, clear its read policy to reflect that.
        if write_only {
            state.read_policy = None;
        }

        if let Some(previous) = self.collections.insert(id, state) {
            panic!("attempt to add a collection with existing ID {id} (previous={previous:?}");
        }

        // Add per-replica collection state.
        for replica in self.replicas.values_mut() {
            if target_replica.is_some_and(|id| id != replica.id) {
                continue;
            }
            replica.add_collection(id, as_of.clone(), replica_input_read_holds.clone());
        }
    }

    fn remove_collection(&mut self, id: GlobalId) {