pub fn store_sequenced_entry(&self, sequenced_entry: Arc<SequencedEntry>) -> Result<()> {
        // Get all needed database rule values, then release the lock
        let rules = self.rules.read();
        let mutable_size_threshold = rules.lifecycle_rules.mutable_size_threshold;
        let immutable = rules.lifecycle_rules.immutable;
        let buffer_size_hard = rules.lifecycle_rules.buffer_size_hard;
        std::mem::drop(rules);

        // We may have gotten here through `store_entry`, in which case this is checking the
        // configuration again unnecessarily, but we may have come here by consuming records from
        // the write buffer, so this check is necessary in that case.
        if immutable {
            return DatabaseNotWriteable {}.fail();
        }
        if let Some(hard_limit) = buffer_size_hard {
            if self.preserved_catalog.state().metrics().memory().total() > hard_limit.get() {
                return HardLimitReached {}.fail();
            }
        }

        // TODO: Direct writes to closing chunks

        if let Some(partitioned_writes) = sequenced_entry.partition_writes() {
            for write in partitioned_writes {
                let partition_key = write.key();
                let partition = self
                    .preserved_catalog
                    .state()
                    .get_or_create_partition(partition_key);
                let mut partition = partition.write();
                partition.update_last_write_at();

                for table_batch in write.table_batches() {
                    match partition.open_chunk(table_batch.name()).ok().flatten() {
                        Some(chunk) => {
                            let mut chunk = chunk.write();
                            chunk.record_write();
                            let chunk_id = chunk.id();

                            let mb_chunk =
                                chunk.mutable_buffer().expect("cannot mutate open chunk");

                            mb_chunk
                                .write_table_batch(
                                    sequenced_entry.sequence().id,
                                    sequenced_entry.sequence().number,
                                    table_batch,
                                )
                                .context(WriteEntry {
                                    partition_key,
                                    chunk_id,
                                })?;

                            check_chunk_closed(&mut *chunk, mutable_size_threshold);
                        }
                        None => {
                            let metrics = self.metrics_registry.register_domain_with_labels(
                                "mutable_buffer",
                                self.metric_labels.clone(),
                            );
                            let mut mb_chunk = MutableBufferChunk::new(
                                table_batch.name(),
                                MutableBufferChunkMetrics::new(
                                    &metrics,
                                    self.preserved_catalog
                                        .state()
                                        .metrics()
                                        .memory()
                                        .mutable_buffer(),
                                ),
                            );

                            mb_chunk
                                .write_table_batch(
                                    sequenced_entry.sequence().id,
                                    sequenced_entry.sequence().number,
                                    table_batch,
                                )
                                .context(WriteEntryInitial { partition_key })?;

                            let new_chunk = partition
                                .create_open_chunk(mb_chunk)
                                .context(OpenEntry { partition_key })?;

                            check_chunk_closed(&mut *new_chunk.write(), mutable_size_threshold);
                        }
                    };
                }
            }
        }

        Ok(())
    }
/// Given a `SequencedEntry`, if the mutable buffer is configured, the `SequencedEntry` is then
/// written into the mutable buffer.
Given a `SequencedEntry`, if the mutable buffer is configured, the `SequencedEntry` is then
written into the mutable buffer.
Given a ` SequencedEntry ` if the mutable buffer is configured the ` SequencedEntry ` is then written into the mutable buffer .
pub fn store_sequenced_entry(&self, sequenced_entry: Arc<SequencedEntry>) -> Result<()> {
        // Get all needed database rule values, then release the lock
        let rules = self.rules.read();
        let mutable_size_threshold = rules.lifecycle_rules.mutable_size_threshold;
        let immutable = rules.lifecycle_rules.immutable;
        let buffer_size_hard = rules.lifecycle_rules.buffer_size_hard;
        std::mem::drop(rules);
        // We may have gotten here through `store_entry`, in which case this is checking the
        // configuration again unnecessarily, but we may have come here by consuming records from
        // the write buffer, so this check is necessary in that case.
        if immutable {
            return DatabaseNotWriteable {}.fail();
        }
        if let Some(hard_limit) = buffer_size_hard {
            if self.preserved_catalog.state().metrics().memory().total() > hard_limit.get() {
                return HardLimitReached {}.fail();
            }
        }
        // TODO: Direct writes to closing chunks
        if let Some(partitioned_writes) = sequenced_entry.partition_writes() {
            for write in partitioned_writes {
                let partition_key = write.key();
                let partition = self
                    .preserved_catalog
                    .state()
                    .get_or_create_partition(partition_key);
                let mut partition = partition.write();
                partition.update_last_write_at();
                for table_batch in write.table_batches() {
                    match partition.open_chunk(table_batch.name()).ok().flatten() {
                        Some(chunk) => {
                            let mut chunk = chunk.write();
                            chunk.record_write();
                            let chunk_id = chunk.id();
                            let mb_chunk =
                                chunk.mutable_buffer().expect("cannot mutate open chunk");
                            mb_chunk
                                .write_table_batch(
                                    sequenced_entry.sequence().id,
                                    sequenced_entry.sequence().number,
                                    table_batch,
                                )
                                .context(WriteEntry {
                                    partition_key,
                                    chunk_id,
                                })?;
                            check_chunk_closed(&mut *chunk, mutable_size_threshold);
                        }
                        None => {
                            let metrics = self.metrics_registry.register_domain_with_labels(
                                "mutable_buffer",
                                self.metric_labels.clone(),
                            );
                            let mut mb_chunk = MutableBufferChunk::new(
                                table_batch.name(),
                                MutableBufferChunkMetrics::new(
                                    &metrics,
                                    self.preserved_catalog
                                        .state()
                                        .metrics()
                                        .memory()
                                        .mutable_buffer(),
                                ),
                            );
                            mb_chunk
                                .write_table_batch(
                                    sequenced_entry.sequence().id,
                                    sequenced_entry.sequence().number,
                                    table_batch,
                                )
                                .context(WriteEntryInitial { partition_key })?;
                            let new_chunk = partition
                                .create_open_chunk(mb_chunk)
                                .context(OpenEntry { partition_key })?;
                            check_chunk_closed(&mut *new_chunk.write(), mutable_size_threshold);
                        }
                    };
                }
            }
        }
        Ok(())
    }
pub fn store_sequenced_entry ( & self , sequenced_entry : Arc < SequencedEntry > ) -> Result < ( ) > { // Get all needed database rule values, then release the lock let rules = self . rules . read ( ) ; let mutable_size_threshold = rules . lifecycle_rules . mutable_size_threshold ; let immutable = rules . lifecycle_rules . immutable ; let buffer_size_hard = rules . lifecycle_rules . buffer_size_hard ; std :: mem :: drop ( rules ) ; // We may have gotten here through `store_entry`, in which case this is checking the // configuration again unnecessarily, but we may have come here by consuming records from // the write buffer, so this check is necessary in that case. if immutable { return DatabaseNotWriteable { } . fail ( ) ; } if let Some ( hard_limit ) = buffer_size_hard { if self . preserved_catalog . state ( ) . metrics ( ) . memory ( ) . total ( ) > hard_limit . get ( ) { return HardLimitReached { } . fail ( ) ; } } // TODO: Direct writes to closing chunks if let Some ( partitioned_writes ) = sequenced_entry . partition_writes ( ) { for write in partitioned_writes { let partition_key = write . key ( ) ; let partition = self . preserved_catalog . state ( ) . get_or_create_partition ( partition_key ) ; let mut partition = partition . write ( ) ; partition . update_last_write_at ( ) ; for table_batch in write . table_batches ( ) { match partition . open_chunk ( table_batch . name ( ) ) . ok ( ) . flatten ( ) { Some ( chunk ) => { let mut chunk = chunk . write ( ) ; chunk . record_write ( ) ; let chunk_id = chunk . id ( ) ; let mb_chunk = chunk . mutable_buffer ( ) . expect ( " " ) ; mb_chunk . write_table_batch ( sequenced_entry . sequence ( ) . id , sequenced_entry . sequence ( ) . number , table_batch , ) . context ( WriteEntry { partition_key , chunk_id , } ) ? ; check_chunk_closed ( & mut * chunk , mutable_size_threshold ) ; } None => { let metrics = self . metrics_registry . register_domain_with_labels ( " " , self . metric_labels . clone ( ) , ) ; let mut mb_chunk = MutableBufferChunk :: new ( table_batch . name ( ) , MutableBufferChunkMetrics :: new ( & metrics , self . preserved_catalog . state ( ) . metrics ( ) . memory ( ) . mutable_buffer ( ) , ) , ) ; mb_chunk . write_table_batch ( sequenced_entry . sequence ( ) . id , sequenced_entry . sequence ( ) . number , table_batch , ) . context ( WriteEntryInitial { partition_key } ) ? ; let new_chunk = partition . create_open_chunk ( mb_chunk ) . context ( OpenEntry { partition_key } ) ? ; check_chunk_closed ( & mut * new_chunk . write ( ) , mutable_size_threshold ) ; } } ; } } } Ok ( ( ) ) }
Given a `SequencedEntry`, if the mutable buffer is configured, the `SequencedEntry` is then
written into the mutable buffer.
Given a ` SequencedEntry ` if the mutable buffer is configured the ` SequencedEntry ` is then written into the mutable buffer .
&self
sequenced_entry
Arc<SequencedEntry>
&self
sequenced_entry
Arc<SequencedEntry>