pub async fn partitions_to_compact(&self) -> Result<Vec<PartitionCompactionCandidate>> {
        let mut candidates = vec![];

        for sequencer_id in &self.sequencers {
            // Read level-0 parquet files
            let level_0_files = self.level_0_parquet_files(*sequencer_id).await?;

            let mut partitions = BTreeMap::new();
            for f in level_0_files {
                let mut p = partitions.entry(f.partition_id).or_insert_with(|| {
                    PartitionCompactionCandidate {
                        sequencer_id: *sequencer_id,
                        table_id: f.table_id,
                        partition_id: f.partition_id,
                        level_0_file_count: 0,
                        file_size_bytes: 0,
                        oldest_file: f.created_at,
                    }
                });
                p.file_size_bytes += f.file_size_bytes;
                p.level_0_file_count += 1;
                p.oldest_file = p.oldest_file.min(f.created_at);
            }

            let mut partitions: Vec<_> = partitions.into_values().collect();

            let total_size = partitions.iter().fold(0, |t, p| t + p.file_size_bytes);
            let attributes =
                Attributes::from([("sequencer_id", format!("{}", *sequencer_id).into())]);

            let number_gauge = self.compaction_candidate_gauge.recorder(attributes.clone());
            number_gauge.set(partitions.len() as u64);
            let size_gauge = self.compaction_candidate_bytes_gauge.recorder(attributes);
            size_gauge.set(total_size as u64);

            candidates.append(&mut partitions);
        }

        candidates.sort_by(
            |a, b| match b.level_0_file_count.cmp(&a.level_0_file_count) {
                Ordering::Equal => b.file_size_bytes.cmp(&a.file_size_bytes),
                o => o,
            },
        );

        debug!(
            candidate_num = candidates.len(),
            "Number of candidate partitions to be considered to compact"
        );

        Ok(candidates)
    }
/// Returns a list of partitions that have level 0 files to compact along with some summary
/// statistics that the scheduler can use to decide which partitions to prioritize. Orders
/// them by the number of level 0 files and then size.
Returns a list of partitions that have level 0 files to compact along with some summary
statistics that the scheduler can use to decide which partitions to prioritize. Orders
them by the number of level 0 files and then size.
Returns a list of partitions that have level 0 files to compact along with some summary statistics that the scheduler can use to decide which partitions to prioritize . Orders them by the number of level 0 files and then size .
pub async fn partitions_to_compact(&self) -> Result<Vec<PartitionCompactionCandidate>> {
        let mut candidates = vec![];
        for sequencer_id in &self.sequencers {
            // Read level-0 parquet files
            let level_0_files = self.level_0_parquet_files(*sequencer_id).await?;
            let mut partitions = BTreeMap::new();
            for f in level_0_files {
                let mut p = partitions.entry(f.partition_id).or_insert_with(|| {
                    PartitionCompactionCandidate {
                        sequencer_id: *sequencer_id,
                        table_id: f.table_id,
                        partition_id: f.partition_id,
                        level_0_file_count: 0,
                        file_size_bytes: 0,
                        oldest_file: f.created_at,
                    }
                });
                p.file_size_bytes += f.file_size_bytes;
                p.level_0_file_count += 1;
                p.oldest_file = p.oldest_file.min(f.created_at);
            }
            let mut partitions: Vec<_> = partitions.into_values().collect();
            let total_size = partitions.iter().fold(0, |t, p| t + p.file_size_bytes);
            let attributes =
                Attributes::from([("sequencer_id", format!("{}", *sequencer_id).into())]);
            let number_gauge = self.compaction_candidate_gauge.recorder(attributes.clone());
            number_gauge.set(partitions.len() as u64);
            let size_gauge = self.compaction_candidate_bytes_gauge.recorder(attributes);
            size_gauge.set(total_size as u64);
            candidates.append(&mut partitions);
        }
        candidates.sort_by(
            |a, b| match b.level_0_file_count.cmp(&a.level_0_file_count) {
                Ordering::Equal => b.file_size_bytes.cmp(&a.file_size_bytes),
                o => o,
            },
        );
        debug!(
            candidate_num = candidates.len(),
            "Number of candidate partitions to be considered to compact"
        );
        Ok(candidates)
    }
pub async fn partitions_to_compact ( & self ) -> Result < Vec < PartitionCompactionCandidate > > { let mut candidates = vec ! [ ] ; for sequencer_id in & self . sequencers { // Read level-0 parquet files let level_0_files = self . level_0_parquet_files ( * sequencer_id ) . await ? ; let mut partitions = BTreeMap :: new ( ) ; for f in level_0_files { let mut p = partitions . entry ( f . partition_id ) . or_insert_with ( | | { PartitionCompactionCandidate { sequencer_id : * sequencer_id , table_id : f . table_id , partition_id : f . partition_id , level_0_file_count : 0 , file_size_bytes : 0 , oldest_file : f . created_at , } } ) ; p . file_size_bytes += f . file_size_bytes ; p . level_0_file_count += 1 ; p . oldest_file = p . oldest_file . min ( f . created_at ) ; } let mut partitions : Vec < _ > = partitions . into_values ( ) . collect ( ) ; let total_size = partitions . iter ( ) . fold ( 0 , | t , p | t + p . file_size_bytes ) ; let attributes = Attributes :: from ( [ ( " " , format ! ( " " sequencer_id ) . into ( ) ) ] ) ; let number_gauge = self . compaction_candidate_gauge . recorder ( attributes . clone ( ) ) ; number_gauge . set ( partitions . len ( ) as u64 ) ; let size_gauge = self . compaction_candidate_bytes_gauge . recorder ( attributes ) ; size_gauge . set ( total_size as u64 ) ; candidates . append ( & mut partitions ) ; } candidates . sort_by ( | a , b | match b . level_0_file_count . cmp ( & a . level_0_file_count ) { Ordering :: Equal => b . file_size_bytes . cmp ( & a . file_size_bytes ) , o => o , } , ) ; debug ! ( candidate_num candidates len ( ) " " ) ; Ok ( candidates ) }
Returns a list of partitions that have level 0 files to compact along with some summary
statistics that the scheduler can use to decide which partitions to prioritize.
Returns a list of partitions that have level 0 files to compact along with some summary statistics that the scheduler can use to decide which partitions to prioritize .
&self
&self