Function write_batches 

fn write_batches<'scope>(
    scope: Scope<'scope, Timestamp>,
    collection_id: GlobalId,
    operator_name: &str,
    target: &CollectionMetadata,
    batch_descriptions: Stream<'scope, Timestamp, Vec<(Antichain<Timestamp>, Antichain<Timestamp>)>>,
    desired_collection: VecCollection<'scope, Timestamp, Result<Row, DataflowError>, Diff>,
    persist_clients: Arc<PersistClientCache>,
    storage_state: &StorageState,
    busy_signal: Arc<Semaphore>,
) -> (StreamVec<'scope, Timestamp, HollowBatchAndMetadata<Timestamp>>, PressOnDropButton)

Writes desired_collection to persist, but only for updates that fall into batch a description that we get via batch_descriptions. This forwards a HollowBatch (with additional metadata) for any batch of updates that was written.

This operator assumes that the desired_collection comes pre-sharded.

This also and updates various metrics.