/// Buffer the operation in the cache, adding any new partitions or delete tombstones to the
/// catalog. Returns true if ingest should be paused due to memory limits set in the passed
/// lifecycle manager.
Buffer the operation in the cache, adding any new partitions or delete tombstones to the
catalog. Returns true if ingest should be paused due to memory limits set in the passed
lifecycle manager.
Buffer the operation in the cache adding any new partitions or delete tombstones to the catalog . Returns true if ingest should be paused due to memory limits set in the passed lifecycle manager .
pub async fn buffer_operation(
        &self,
        dml_operation: DmlOperation,
        sequencer_id: SequencerId,
        catalog: &dyn Catalog,
        lifecycle_handle: &dyn LifecycleHandle,
        partitioner: &dyn Partitioner,
        executor: &Executor,
    ) -> Result<bool> {
        let sequence_number = dml_operation
            .meta()
            .sequence()
            .expect("must have sequence number")
            .sequence_number;
        let sequence_number = i64::try_from(sequence_number).expect("sequence out of bounds");
        let sequence_number = SequenceNumber::new(sequence_number);
        match dml_operation {
            DmlOperation::Write(write) => {
                let mut pause_writes = false;
                for (t, b) in write.into_tables() {
                    let table_data = match self.table_data(&t) {
                        Some(t) => t,
                        None => self.insert_table(sequencer_id, &t, catalog).await?,
                    };
                    let mut table_data = table_data.write().await;
                    let should_pause = table_data
                        .buffer_table_write(
                            sequence_number,
                            b,
                            sequencer_id,
                            catalog,
                            lifecycle_handle,
                            partitioner,
                        )
                        .await?;
                    pause_writes = pause_writes || should_pause;
                }
                Ok(pause_writes)
            }
            DmlOperation::Delete(delete) => {
                let table_name = delete.table_name().context(TableNotPresentSnafu)?;
                let table_data = match self.table_data(table_name) {
                    Some(t) => t,
                    None => self.insert_table(sequencer_id, table_name, catalog).await?,
                };
                let mut table_data = table_data.write().await;
                table_data
                    .buffer_delete(
                        table_name,
                        delete.predicate(),
                        sequencer_id,
                        sequence_number,
                        catalog,
                        executor,
                    )
                    .await?;
                // don't pause writes since deletes don't count towards memory limits
                Ok(false)
            }
        }
    }
pub async fn buffer_operation ( & self , dml_operation : DmlOperation , sequencer_id : SequencerId , catalog : & dyn Catalog , lifecycle_handle : & dyn LifecycleHandle , partitioner : & dyn Partitioner , executor : & Executor , ) -> Result < bool > { let sequence_number = dml_operation . meta ( ) . sequence ( ) . expect ( " " ) . sequence_number ; let sequence_number = i64 :: try_from ( sequence_number ) . expect ( " " ) ; let sequence_number = SequenceNumber :: new ( sequence_number ) ; match dml_operation { DmlOperation :: Write ( write ) => { let mut pause_writes = false ; for ( t , b ) in write . into_tables ( ) { let table_data = match self . table_data ( & t ) { Some ( t ) => t , None => self . insert_table ( sequencer_id , & t , catalog ) . await ? , } ; let mut table_data = table_data . write ( ) . await ; let should_pause = table_data . buffer_table_write ( sequence_number , b , sequencer_id , catalog , lifecycle_handle , partitioner , ) . await ? ; pause_writes = pause_writes || should_pause ; } Ok ( pause_writes ) } DmlOperation :: Delete ( delete ) => { let table_name = delete . table_name ( ) . context ( TableNotPresentSnafu ) ? ; let table_data = match self . table_data ( table_name ) { Some ( t ) => t , None => self . insert_table ( sequencer_id , table_name , catalog ) . await ? , } ; let mut table_data = table_data . write ( ) . await ; table_data . buffer_delete ( table_name , delete . predicate ( ) , sequencer_id , sequence_number , catalog , executor , ) . await ? ; // don't pause writes since deletes don't count towards memory limits Ok ( false ) } } }
Buffer the operation in the cache, adding any new partitions or delete tombstones to the
catalog.
Buffer the operation in the cache adding any new partitions or delete tombstones to the catalog .
&self
dml_operation
DmlOperation
sequencer_id
SequencerId
catalog
&dyn Catalog
lifecycle_handle
&dyn LifecycleHandle
partitioner
&dyn Partitioner
executor
&Executor