fn maybe_compact_chunks<P: LockablePartition>(
        &mut self,
        partition: &P,
        rules: &LifecycleRules,
        now: DateTime<Utc>,
    ) {
        let mut rows_left = rules.persist_row_threshold.get();

        // TODO: Encapsulate locking into a CatalogTransaction type
        let partition = partition.read();
        if partition.is_persisted() {
            debug!(db_name = %self.db.name(), %partition, "nothing to be compacted for partition");
            return;
        }

        let mut chunks = LockablePartition::chunks(&partition);
        // Sort by chunk ID to ensure a stable lock order
        chunks.sort_by_key(|x| x.0);

        let mut has_mub_snapshot = false;
        let mut to_compact = Vec::new();
        for (_, chunk) in &chunks {
            let chunk = chunk.read();
            if chunk.lifecycle_action().is_some() {
                // This likely means it is being persisted
                continue;
            }

            let to_compact_len_before = to_compact.len();
            let storage = chunk.storage();
            match storage {
                ChunkStorage::OpenMutableBuffer => {
                    if can_move(rules, &*chunk, now) {
                        has_mub_snapshot = true;
                        to_compact.push(chunk);
                    }
                }
                ChunkStorage::ClosedMutableBuffer => {
                    has_mub_snapshot = true;
                    to_compact.push(chunk);
                }
                ChunkStorage::ReadBuffer => {
                    let row_count = chunk.row_count();
                    if row_count >= rows_left {
                        continue;
                    }
                    rows_left = rows_left.saturating_sub(row_count);
                    to_compact.push(chunk);
                }
                _ => {}
            }
            let has_added_to_compact = to_compact.len() > to_compact_len_before;
            debug!(db_name = %self.db.name(),
                   %partition,
                   ?has_added_to_compact,
                   chunk_storage = ?storage,
                   ?has_mub_snapshot,
                   "maybe compacting chunks");
        }

        if to_compact.len() >= 2 || has_mub_snapshot {
            // caller's responsibility to determine if we can maybe compact.
            assert!(self.active_compactions < rules.max_active_compactions.get() as usize);

            // Upgrade partition first
            let partition = partition.upgrade();
            let chunks = to_compact
                .into_iter()
                .map(|chunk| chunk.upgrade())
                .collect();

            let tracker = LockablePartition::compact_chunks(partition, chunks)
                .expect("failed to compact chunks")
                .with_metadata(ChunkLifecycleAction::Compacting);

            self.active_compactions += 1;
            self.trackers.push(tracker);
        }
    }
/// Find chunks to compact together
///
/// Finds unpersisted chunks with no in-progress lifecycle actions
/// and compacts them into a single RUB chunk
///
/// Will include the open chunk if it is cold for writes as determined
/// by the mutable linger threshold
Find chunks to compact together
Finds unpersisted chunks with no in-progress lifecycle actions
and compacts them into a single RUB chunk

Will include the open chunk if it is cold for writes as determined
by the mutable linger threshold
Find chunks to compact together Finds unpersisted chunks with no in - progress lifecycle actions and compacts them into a single RUB chunk Will include the open chunk if it is cold for writes as determined by the mutable linger threshold
fn maybe_compact_chunks<P: LockablePartition>(
        &mut self,
        partition: &P,
        rules: &LifecycleRules,
        now: DateTime<Utc>,
    ) {
        let mut rows_left = rules.persist_row_threshold.get();
        // TODO: Encapsulate locking into a CatalogTransaction type
        let partition = partition.read();
        if partition.is_persisted() {
            debug!(db_name = %self.db.name(), %partition, "nothing to be compacted for partition");
            return;
        }
        let mut chunks = LockablePartition::chunks(&partition);
        // Sort by chunk ID to ensure a stable lock order
        chunks.sort_by_key(|x| x.0);
        let mut has_mub_snapshot = false;
        let mut to_compact = Vec::new();
        for (_, chunk) in &chunks {
            let chunk = chunk.read();
            if chunk.lifecycle_action().is_some() {
                // This likely means it is being persisted
                continue;
            }
            let to_compact_len_before = to_compact.len();
            let storage = chunk.storage();
            match storage {
                ChunkStorage::OpenMutableBuffer => {
                    if can_move(rules, &*chunk, now) {
                        has_mub_snapshot = true;
                        to_compact.push(chunk);
                    }
                }
                ChunkStorage::ClosedMutableBuffer => {
                    has_mub_snapshot = true;
                    to_compact.push(chunk);
                }
                ChunkStorage::ReadBuffer => {
                    let row_count = chunk.row_count();
                    if row_count >= rows_left {
                        continue;
                    }
                    rows_left = rows_left.saturating_sub(row_count);
                    to_compact.push(chunk);
                }
                _ => {}
            }
            let has_added_to_compact = to_compact.len() > to_compact_len_before;
            debug!(db_name = %self.db.name(),
                   %partition,
                   ?has_added_to_compact,
                   chunk_storage = ?storage,
                   ?has_mub_snapshot,
                   "maybe compacting chunks");
        }
        if to_compact.len() >= 2 || has_mub_snapshot {
            // caller's responsibility to determine if we can maybe compact.
            assert!(self.active_compactions < rules.max_active_compactions.get() as usize);
            // Upgrade partition first
            let partition = partition.upgrade();
            let chunks = to_compact
                .into_iter()
                .map(|chunk| chunk.upgrade())
                .collect();
            let tracker = LockablePartition::compact_chunks(partition, chunks)
                .expect("failed to compact chunks")
                .with_metadata(ChunkLifecycleAction::Compacting);
            self.active_compactions += 1;
            self.trackers.push(tracker);
        }
    }
fn maybe_compact_chunks < P : LockablePartition > ( & mut self , partition : & P , rules : & LifecycleRules , now : DateTime < Utc > , ) { let mut rows_left = rules . persist_row_threshold . get ( ) ; // TODO: Encapsulate locking into a CatalogTransaction type let partition = partition . read ( ) ; if partition . is_persisted ( ) { debug ! ( db_name self db name ( ) partition " " ) ; return ; } let mut chunks = LockablePartition :: chunks ( & partition ) ; // Sort by chunk ID to ensure a stable lock order chunks . sort_by_key ( | x | x . 0 ) ; let mut has_mub_snapshot = false ; let mut to_compact = Vec :: new ( ) ; for ( _ , chunk ) in & chunks { let chunk = chunk . read ( ) ; if chunk . lifecycle_action ( ) . is_some ( ) { // This likely means it is being persisted continue ; } let to_compact_len_before = to_compact . len ( ) ; let storage = chunk . storage ( ) ; match storage { ChunkStorage :: OpenMutableBuffer => { if can_move ( rules , & * chunk , now ) { has_mub_snapshot = true ; to_compact . push ( chunk ) ; } } ChunkStorage :: ClosedMutableBuffer => { has_mub_snapshot = true ; to_compact . push ( chunk ) ; } ChunkStorage :: ReadBuffer => { let row_count = chunk . row_count ( ) ; if row_count >= rows_left { continue ; } rows_left = rows_left . saturating_sub ( row_count ) ; to_compact . push ( chunk ) ; } _ => { } } let has_added_to_compact = to_compact . len ( ) > to_compact_len_before ; debug ! ( db_name self db name ( ) partition has_added_to_compact chunk_storage storage has_mub_snapshot " " ) ; } if to_compact . len ( ) >= 2 || has_mub_snapshot { // caller's responsibility to determine if we can maybe compact. assert ! ( self active_compactions rules max_active_compactions get ( ) as usize ) ; // Upgrade partition first let partition = partition . upgrade ( ) ; let chunks = to_compact . into_iter ( ) . map ( | chunk | chunk . upgrade ( ) ) . collect ( ) ; let tracker = LockablePartition :: compact_chunks ( partition , chunks ) . expect ( " " ) . with_metadata ( ChunkLifecycleAction :: Compacting ) ; self . active_compactions += 1 ; self . trackers . push ( tracker ) ; } }
Find chunks to compact together
Finds unpersisted chunks with no in-progress lifecycle actions
and compacts them into a single RUB chunk
Find chunks to compact together Finds unpersisted chunks with no in - progress lifecycle actions and compacts them into a single RUB chunk
&mut self
partition
&P
rules
&LifecycleRules
now
DateTime<Utc>
&mut self
partition
&P
rules
&LifecycleRules
now
DateTime<Utc>