fn maybe_persist_chunks<P: LockablePartition>(
        &mut self,
        db_name: &DatabaseName<'static>,
        partition: &P,
        rules: &LifecycleRules,
        now: Instant,
    ) -> bool {
        // TODO: Encapsulate locking into a CatalogTransaction type
        let partition = partition.read();

        if partition.is_persisted() {
            debug!(%db_name, %partition, "nothing to persist for partition");
            return false;
        }

        let persistable_age_seconds = partition
            .minimum_unpersisted_age()
            .and_then(|minimum_unpersisted_age| {
                // If writes happened between when the policy loop
                // started and this check is done, the duration may be
                // negative. Skip persistence in this case to avoid
                // panic in `duration_since`
                Some(
                    now.checked_duration_since(minimum_unpersisted_age)?
                        .as_secs(),
                )
            })
            .unwrap_or_default() as u32;

        let persistable_row_count = partition.persistable_row_count(now);
        debug!(%db_name, %partition,
               partition_persist_row_count=persistable_row_count,
               rules_persist_row_count=%rules.persist_row_threshold.get(),
               partition_persistable_age_seconds=persistable_age_seconds,
               rules_persist_age_threshold_seconds=%rules.persist_age_threshold_seconds.get(),
               "considering for persistence");

        if persistable_row_count >= rules.persist_row_threshold.get() {
            info!(%db_name, %partition, persistable_row_count, "persisting partition as exceeds row threshold");
        } else if persistable_age_seconds >= rules.persist_age_threshold_seconds.get() {
            info!(%db_name, %partition, persistable_age_seconds, "persisting partition as exceeds age threshold");
        } else {
            debug!(%db_name, %partition, persistable_row_count, "partition not eligible for persist");
            return false;
        }

        let mut chunks = LockablePartition::chunks(&partition);

        // Upgrade partition to be able to rotate persistence windows
        let mut partition = partition.upgrade();

        let persist_handle = match LockablePartition::prepare_persist(&mut partition, now) {
            Some(x) => x,
            None => {
                debug!(%db_name, %partition, "no persistable windows or previous outstanding persist");
                return false;
            }
        };

        // Sort by chunk ID to ensure a stable lock order
        chunks.sort_by_key(|x| x.0);
        let mut to_persist = Vec::new();
        for (_, chunk) in &chunks {
            let chunk = chunk.read();
            trace!(%db_name, %partition, chunk=%chunk.addr(), "considering chunk for persistence");

            // Check if chunk is eligible for persistence
            match chunk.storage() {
                ChunkStorage::OpenMutableBuffer
                | ChunkStorage::ClosedMutableBuffer
                | ChunkStorage::ReadBuffer => {}
                ChunkStorage::ReadBufferAndObjectStore | ChunkStorage::ObjectStoreOnly => {
                    debug!(%db_name, %partition, chunk=%chunk.addr(), storage=?chunk.storage(),
                           "chunk not eligible due to storage");
                    continue;
                }
            }

            // Chunk's data is entirely after the time we are flushing
            // up to, and thus there is reason to include it in the
            // plan
            if chunk.min_timestamp() > persist_handle.timestamp() {
                // Can safely ignore chunk
                debug!(%db_name, %partition, chunk=%chunk.addr(),
                       "chunk does not contain data eligible for persistence");
                continue;
            }

            // If the chunk has an outstanding lifecycle action
            if let Some(action) = chunk.lifecycle_action() {
                // see if we should stall subsequent pull it is
                // preventing us from persisting
                let stall = action.metadata() == &ChunkLifecycleAction::Compacting;
                info!(%db_name, ?action, chunk=%chunk.addr(), "Chunk to persist has outstanding action");
                return stall;
            }

            to_persist.push(chunk);
        }

        if to_persist.is_empty() {
            info!(%db_name, %partition, "expected to persist but found no eligible chunks");
            return false;
        }

        let chunks = to_persist
            .into_iter()
            .map(|chunk| chunk.upgrade())
            .collect();

        let tracker = LockablePartition::persist_chunks(partition, chunks, persist_handle)
            .expect("failed to persist chunks")
            .with_metadata(ChunkLifecycleAction::Persisting);

        self.trackers.push(tracker);
        false
    }
/// Check persistence
///
/// Looks for chunks to combine together in the "persist"
/// operation. The "persist" operation combines the data from a
/// list chunks and creates two new chunks: one persisted, with
/// all data that eligible for persistence, and the second with
/// all data that is not yet eligible for persistence (it was
/// written to recently)
///
/// A chunk will be chosen for the persist operation if either:
///
/// 1. it has more than `persist_row_threshold` rows
/// 2. it was last written to more than `late_arrive_window_seconds` ago
///
/// Returns true if persistence is being blocked by compaction,
/// signaling compaction should be stalled to allow persistence to
/// make progress
///
/// Returns a boolean to indicate if it should stall compaction to allow
/// persistence to make progress
///
/// The rationale for stalling compaction until a persist can start:
///
/// 1. It is a simple way to ensure a persist can start. Once the
/// persist has started (which might also effectively compact
/// several chunks as well) then compactions can start again
///
/// 2. It is not likely to change the number of compactions
/// significantly. Since the policy goal at time of writing is to
/// end up with ~ 2 unpersisted chunks at any time, any chunk that
/// is persistable is also likely to be one of the ones being
/// compacted.
Check persistence
Looks for chunks to combine together in the "persist"
operation. The "persist" operation combines the data from a
list chunks and creates two new chunks: one persisted, with
all data that eligible for persistence, and the second with
all data that is not yet eligible for persistence (it was
written to recently)

A chunk will be chosen for the persist operation if either.

1. it has more than `persist_row_threshold` rows
2. it was last written to more than `late_arrive_window_seconds` ago

Returns true if persistence is being blocked by compaction,
signaling compaction should be stalled to allow persistence to
make progress

Returns a boolean to indicate if it should stall compaction to allow
persistence to make progress

The rationale for stalling compaction until a persist can start.

1. It is a simple way to ensure a persist can start. Once the
persist has started (which might also effectively compact
several chunks as well) then compactions can start again

2. It is not likely to change the number of compactions
significantly. Since the policy goal at time of writing is to
end up with ~ 2 unpersisted chunks at any time, any chunk that
is persistable is also likely to be one of the ones being
compacted.
Check persistence Looks for chunks to combine together in the " persist " operation . The " persist " operation combines the data from a list chunks and creates two new chunks : one persisted with all data that eligible for persistence and the second with all data that is not yet eligible for persistence ( it was written to recently ) A chunk will be chosen for the persist operation if either . 1 . it has more than ` persist_row_threshold ` rows 2 . it was last written to more than ` late_arrive_window_seconds ` ago Returns true if persistence is being blocked by compaction signaling compaction should be stalled to allow persistence to make progress Returns a boolean to indicate if it should stall compaction to allow persistence to make progress The rationale for stalling compaction until a persist can start . 1 . It is a simple way to ensure a persist can start . Once the persist has started ( which might also effectively compact several chunks as well ) then compactions can start again 2 . It is not likely to change the number of compactions significantly . Since the policy goal at time of writing is to end up with ~ 2 unpersisted chunks at any time any chunk that is persistable is also likely to be one of the ones being compacted .
fn maybe_persist_chunks<P: LockablePartition>(
        &mut self,
        db_name: &DatabaseName<'static>,
        partition: &P,
        rules: &LifecycleRules,
        now: Instant,
    ) -> bool {
        // TODO: Encapsulate locking into a CatalogTransaction type
        let partition = partition.read();
        if partition.is_persisted() {
            debug!(%db_name, %partition, "nothing to persist for partition");
            return false;
        }
        let persistable_age_seconds = partition
            .minimum_unpersisted_age()
            .and_then(|minimum_unpersisted_age| {
                // If writes happened between when the policy loop
                // started and this check is done, the duration may be
                // negative. Skip persistence in this case to avoid
                // panic in `duration_since`
                Some(
                    now.checked_duration_since(minimum_unpersisted_age)?
                        .as_secs(),
                )
            })
            .unwrap_or_default() as u32;
        let persistable_row_count = partition.persistable_row_count(now);
        debug!(%db_name, %partition,
               partition_persist_row_count=persistable_row_count,
               rules_persist_row_count=%rules.persist_row_threshold.get(),
               partition_persistable_age_seconds=persistable_age_seconds,
               rules_persist_age_threshold_seconds=%rules.persist_age_threshold_seconds.get(),
               "considering for persistence");
        if persistable_row_count >= rules.persist_row_threshold.get() {
            info!(%db_name, %partition, persistable_row_count, "persisting partition as exceeds row threshold");
        } else if persistable_age_seconds >= rules.persist_age_threshold_seconds.get() {
            info!(%db_name, %partition, persistable_age_seconds, "persisting partition as exceeds age threshold");
        } else {
            debug!(%db_name, %partition, persistable_row_count, "partition not eligible for persist");
            return false;
        }
        let mut chunks = LockablePartition::chunks(&partition);
        // Upgrade partition to be able to rotate persistence windows
        let mut partition = partition.upgrade();
        let persist_handle = match LockablePartition::prepare_persist(&mut partition, now) {
            Some(x) => x,
            None => {
                debug!(%db_name, %partition, "no persistable windows or previous outstanding persist");
                return false;
            }
        };
        // Sort by chunk ID to ensure a stable lock order
        chunks.sort_by_key(|x| x.0);
        let mut to_persist = Vec::new();
        for (_, chunk) in &chunks {
            let chunk = chunk.read();
            trace!(%db_name, %partition, chunk=%chunk.addr(), "considering chunk for persistence");
            // Check if chunk is eligible for persistence
            match chunk.storage() {
                ChunkStorage::OpenMutableBuffer
                | ChunkStorage::ClosedMutableBuffer
                | ChunkStorage::ReadBuffer => {}
                ChunkStorage::ReadBufferAndObjectStore | ChunkStorage::ObjectStoreOnly => {
                    debug!(%db_name, %partition, chunk=%chunk.addr(), storage=?chunk.storage(),
                           "chunk not eligible due to storage");
                    continue;
                }
            }
            // Chunk's data is entirely after the time we are flushing
            // up to, and thus there is reason to include it in the
            // plan
            if chunk.min_timestamp() > persist_handle.timestamp() {
                // Can safely ignore chunk
                debug!(%db_name, %partition, chunk=%chunk.addr(),
                       "chunk does not contain data eligible for persistence");
                continue;
            }
            // If the chunk has an outstanding lifecycle action
            if let Some(action) = chunk.lifecycle_action() {
                // see if we should stall subsequent pull it is
                // preventing us from persisting
                let stall = action.metadata() == &ChunkLifecycleAction::Compacting;
                info!(%db_name, ?action, chunk=%chunk.addr(), "Chunk to persist has outstanding action");
                return stall;
            }
            to_persist.push(chunk);
        }
        if to_persist.is_empty() {
            info!(%db_name, %partition, "expected to persist but found no eligible chunks");
            return false;
        }
        let chunks = to_persist
            .into_iter()
            .map(|chunk| chunk.upgrade())
            .collect();
        let tracker = LockablePartition::persist_chunks(partition, chunks, persist_handle)
            .expect("failed to persist chunks")
            .with_metadata(ChunkLifecycleAction::Persisting);
        self.trackers.push(tracker);
        false
    }
fn maybe_persist_chunks < P : LockablePartition > ( & mut self , db_name : & DatabaseName < ' static > , partition : & P , rules : & LifecycleRules , now : Instant , ) -> bool { // TODO: Encapsulate locking into a CatalogTransaction type let partition = partition . read ( ) ; if partition . is_persisted ( ) { debug ! ( db_name partition " " ) ; return false ; } let persistable_age_seconds = partition . minimum_unpersisted_age ( ) . and_then ( | minimum_unpersisted_age | { // If writes happened between when the policy loop // started and this check is done, the duration may be // negative. Skip persistence in this case to avoid // panic in `duration_since` Some ( now . checked_duration_since ( minimum_unpersisted_age ) ? . as_secs ( ) , ) } ) . unwrap_or_default ( ) as u32 ; let persistable_row_count = partition . persistable_row_count ( now ) ; debug ! ( db_name partition partition_persist_row_count persistable_row_count rules_persist_row_count rules persist_row_threshold get ( ) partition_persistable_age_seconds persistable_age_seconds rules_persist_age_threshold_seconds rules persist_age_threshold_seconds get ( ) " " ) ; if persistable_row_count >= rules . persist_row_threshold . get ( ) { info ! ( db_name partition persistable_row_count " " ) ; } else if persistable_age_seconds >= rules . persist_age_threshold_seconds . get ( ) { info ! ( db_name partition persistable_age_seconds " " ) ; } else { debug ! ( db_name partition persistable_row_count " " ) ; return false ; } let mut chunks = LockablePartition :: chunks ( & partition ) ; // Upgrade partition to be able to rotate persistence windows let mut partition = partition . upgrade ( ) ; let persist_handle = match LockablePartition :: prepare_persist ( & mut partition , now ) { Some ( x ) => x , None => { debug ! ( db_name partition " " ) ; return false ; } } ; // Sort by chunk ID to ensure a stable lock order chunks . sort_by_key ( | x | x . 0 ) ; let mut to_persist = Vec :: new ( ) ; for ( _ , chunk ) in & chunks { let chunk = chunk . read ( ) ; trace ! ( db_name partition chunk chunk addr ( ) " " ) ; // Check if chunk is eligible for persistence match chunk . storage ( ) { ChunkStorage :: OpenMutableBuffer | ChunkStorage :: ClosedMutableBuffer | ChunkStorage :: ReadBuffer => { } ChunkStorage :: ReadBufferAndObjectStore | ChunkStorage :: ObjectStoreOnly => { debug ! ( db_name partition chunk chunk addr ( ) storage chunk storage ( ) " " ) ; continue ; } } // Chunk's data is entirely after the time we are flushing // up to, and thus there is reason to include it in the // plan if chunk . min_timestamp ( ) > persist_handle . timestamp ( ) { // Can safely ignore chunk debug ! ( db_name partition chunk chunk addr ( ) " " ) ; continue ; } // If the chunk has an outstanding lifecycle action if let Some ( action ) = chunk . lifecycle_action ( ) { // see if we should stall subsequent pull it is // preventing us from persisting let stall = action . metadata ( ) == & ChunkLifecycleAction :: Compacting ; info ! ( db_name action chunk chunk addr ( ) " " ) ; return stall ; } to_persist . push ( chunk ) ; } if to_persist . is_empty ( ) { info ! ( db_name partition " " ) ; return false ; } let chunks = to_persist . into_iter ( ) . map ( | chunk | chunk . upgrade ( ) ) . collect ( ) ; let tracker = LockablePartition :: persist_chunks ( partition , chunks , persist_handle ) . expect ( " " ) . with_metadata ( ChunkLifecycleAction :: Persisting ) ; self . trackers . push ( tracker ) ; false }
Check persistence
Looks for chunks to combine together in the "persist"
operation.
Check persistence Looks for chunks to combine together in the " persist " operation .
&mut self
db_name
&DatabaseName<'static>
partition
&P
rules
&LifecycleRules
now
Instant
&mut self
db_name
&DatabaseName<'static>
partition
&P
rules
&LifecycleRules
now
Instant