/// Shard `writes` and dispatch the resultant DML operations.
Shard `writes` and dispatch the resultant DML operations.
Shard ` writes ` and dispatch the resultant DML operations .
async fn write(
        &self,
        namespace: &DatabaseName<'static>,
        writes: Self::WriteInput,
        span_ctx: Option<SpanContext>,
    ) -> Result<Self::WriteOutput, ShardError> {
        let mut collated: HashMap<_, HashMap<String, MutableBatch>> = HashMap::new();
        // Extract the partition key & DML writes.
        let (partition_key, writes) = writes.into_parts();
        // Shard each entry in `writes` and collate them into one DML operation
        // per shard to maximise the size of each write, and therefore increase
        // the effectiveness of compression of ops in the write buffer.
        for (table, batch) in writes.into_iter() {
            let sequencer = self.sharder.shard(&table, namespace, &batch);
            let existing = collated
                .entry(sequencer)
                .or_default()
                .insert(table, batch.clone());
            assert!(existing.is_none());
        }
        let iter = collated.into_iter().map(|(sequencer, batch)| {
            let dml = DmlWrite::new(namespace, batch, DmlMeta::unsequenced(span_ctx.clone()));
            trace!(
                %partition_key,
                kafka_partition=%sequencer.id(),
                tables=%dml.table_count(),
                %namespace,
                approx_size=%dml.size(),
                "routing writes to shard"
            );
            (sequencer, DmlOperation::from(dml))
        });
        parallel_enqueue(iter).await
    }
async fn write ( & self , namespace : & DatabaseName < ' static > , writes : Self :: WriteInput , span_ctx : Option < SpanContext > , ) -> Result < Self :: WriteOutput , ShardError > { let mut collated : HashMap < _ , HashMap < String , MutableBatch > > = HashMap :: new ( ) ; // Extract the partition key & DML writes. let ( partition_key , writes ) = writes . into_parts ( ) ; // Shard each entry in `writes` and collate them into one DML operation // per shard to maximise the size of each write, and therefore increase // the effectiveness of compression of ops in the write buffer. for ( table , batch ) in writes . into_iter ( ) { let sequencer = self . sharder . shard ( & table , namespace , & batch ) ; let existing = collated . entry ( sequencer ) . or_default ( ) . insert ( table , batch . clone ( ) ) ; assert ! ( existing is_none ( ) ) ; } let iter = collated . into_iter ( ) . map ( | ( sequencer , batch ) | { let dml = DmlWrite :: new ( namespace , batch , DmlMeta :: unsequenced ( span_ctx . clone ( ) ) ) ; trace ! ( partition_key kafka_partition sequencer id ( ) tables dml table_count ( ) namespace approx_size dml size ( ) " " ) ; ( sequencer , DmlOperation :: from ( dml ) ) } ) ; parallel_enqueue ( iter ) . await }
Shard `writes` and dispatch the resultant DML operations.
Shard ` writes ` and dispatch the resultant DML operations .
&self
namespace
&DatabaseName<'static>
writes
Self::WriteInput
span_ctx
Option<SpanContext>