async fn buffer_table_write(
        &mut self,
        sequence_number: SequenceNumber,
        batch: MutableBatch,
        sequencer_id: SequencerId,
        catalog: &dyn Catalog,
        lifecycle_handle: &dyn LifecycleHandle,
        partitioner: &dyn Partitioner,
    ) -> Result<bool> {
        let partition_key = partitioner
            .partition_key(&batch)
            .context(PartitioningSnafu)?;

        let partition_data = match self.partition_data.get_mut(&partition_key) {
            Some(p) => p,
            None => {
                self.insert_partition(&partition_key, sequencer_id, catalog)
                    .await?;
                self.partition_data.get_mut(&partition_key).unwrap()
            }
        };

        // skip the write if it has already been persisted
        if let Some(max) = partition_data.data.max_persisted_sequence_number {
            if max >= sequence_number {
                return Ok(false);
            }
        }

        let should_pause = lifecycle_handle.log_write(
            partition_data.id,
            sequencer_id,
            sequence_number,
            batch.size(),
        );
        partition_data.buffer_write(sequence_number, batch)?;

        Ok(should_pause)
    }
// buffers the table write and returns true if the lifecycle manager indicates that
// ingest should be paused.
buffers the table write and returns true if the lifecycle manager indicates that
ingest should be paused.
buffers the table write and returns true if the lifecycle manager indicates that ingest should be paused .
async fn buffer_table_write(
        &mut self,
        sequence_number: SequenceNumber,
        batch: MutableBatch,
        sequencer_id: SequencerId,
        catalog: &dyn Catalog,
        lifecycle_handle: &dyn LifecycleHandle,
        partitioner: &dyn Partitioner,
    ) -> Result<bool> {
        let partition_key = partitioner
            .partition_key(&batch)
            .context(PartitioningSnafu)?;
        let partition_data = match self.partition_data.get_mut(&partition_key) {
            Some(p) => p,
            None => {
                self.insert_partition(&partition_key, sequencer_id, catalog)
                    .await?;
                self.partition_data.get_mut(&partition_key).unwrap()
            }
        };
        // skip the write if it has already been persisted
        if let Some(max) = partition_data.data.max_persisted_sequence_number {
            if max >= sequence_number {
                return Ok(false);
            }
        }
        let should_pause = lifecycle_handle.log_write(
            partition_data.id,
            sequencer_id,
            sequence_number,
            batch.size(),
        );
        partition_data.buffer_write(sequence_number, batch)?;
        Ok(should_pause)
    }
async fn buffer_table_write ( & mut self , sequence_number : SequenceNumber , batch : MutableBatch , sequencer_id : SequencerId , catalog : & dyn Catalog , lifecycle_handle : & dyn LifecycleHandle , partitioner : & dyn Partitioner , ) -> Result < bool > { let partition_key = partitioner . partition_key ( & batch ) . context ( PartitioningSnafu ) ? ; let partition_data = match self . partition_data . get_mut ( & partition_key ) { Some ( p ) => p , None => { self . insert_partition ( & partition_key , sequencer_id , catalog ) . await ? ; self . partition_data . get_mut ( & partition_key ) . unwrap ( ) } } ; // skip the write if it has already been persisted if let Some ( max ) = partition_data . data . max_persisted_sequence_number { if max >= sequence_number { return Ok ( false ) ; } } let should_pause = lifecycle_handle . log_write ( partition_data . id , sequencer_id , sequence_number , batch . size ( ) , ) ; partition_data . buffer_write ( sequence_number , batch ) ? ; Ok ( should_pause ) }
buffers the table write and returns true if the lifecycle manager indicates that
ingest should be paused.
buffers the table write and returns true if the lifecycle manager indicates that ingest should be paused .
&mut self
sequence_number
SequenceNumber
batch
MutableBatch
sequencer_id
SequencerId
catalog
&dyn Catalog
lifecycle_handle
&dyn LifecycleHandle
partitioner
&dyn Partitioner
&mut self
sequence_number
SequenceNumber
batch
MutableBatch
sequencer_id
SequencerId
catalog
&dyn Catalog
lifecycle_handle
&dyn LifecycleHandle
partitioner
&dyn Partitioner