pub async fn groups_to_compact_and_files_to_upgrade(
        &self,
        partition_id: PartitionId,
        compaction_max_size_bytes: i64, // max size of files to get compacted
        compaction_max_file_count: i64, // max number of files to get compacted
    ) -> Result<CompactAndUpgrade> {
        let mut compact_and_upgrade = CompactAndUpgrade::new(None);

        // List all valid (not soft deletded) files of the partition
        let parquet_files = self
            .catalog
            .repositories()
            .await
            .parquet_files()
            .list_by_partition_not_to_delete_with_metadata(partition_id)
            .await
            .context(ListParquetFilesSnafu)?;
        if parquet_files.is_empty() {
            return Ok(compact_and_upgrade);
        }

        compact_and_upgrade.sequencer_id = Some(parquet_files[0].sequencer_id);

        // Group overlapped files
        let overlapped_file_groups = Self::overlapped_groups(parquet_files);

        // Group time-contiguous non-overlapped groups if their total size is smaller than a threshold
        let compact_file_groups = Self::group_small_contiguous_groups(
            overlapped_file_groups,
            compaction_max_size_bytes,
            compaction_max_file_count,
        );

        // Attach appropriate tombstones to each file
        let groups_with_tombstones = self.add_tombstones_to_groups(compact_file_groups).await?;
        info!("compacting {} groups", groups_with_tombstones.len());

        // File groups to compact and files to upgrade
        for group in groups_with_tombstones {
            // Only one file without tombstones, no need to compact.
            if group.parquet_files.len() == 1 && group.tombstones.is_empty() {
                // If it is level 0, upgrade it since it is non-overlapping
                if group.parquet_files[0].compaction_level == INITIAL_COMPACTION_LEVEL {
                    compact_and_upgrade
                        .files_to_upgrade
                        .push(group.parquet_files[0].parquet_file_id())
                }
            } else {
                compact_and_upgrade.groups_to_compact.push(group);
            }
        }

        Ok(compact_and_upgrade)
    }
/// Group files to be compacted together and level-0 files that will get upgraded
/// for a given partition.
/// All overlapped files will be in the same group no matter
/// how many and how large they are.
Group files to be compacted together and level-0 files that will get upgraded
for a given partition.
All overlapped files will be in the same group no matter
how many and how large they are.
Group files to be compacted together and level - 0 files that will get upgraded for a given partition . All overlapped files will be in the same group no matter how many and how large they are .
pub async fn groups_to_compact_and_files_to_upgrade(
        &self,
        partition_id: PartitionId,
        compaction_max_size_bytes: i64, // max size of files to get compacted
        compaction_max_file_count: i64, // max number of files to get compacted
    ) -> Result<CompactAndUpgrade> {
        let mut compact_and_upgrade = CompactAndUpgrade::new(None);
        // List all valid (not soft deletded) files of the partition
        let parquet_files = self
            .catalog
            .repositories()
            .await
            .parquet_files()
            .list_by_partition_not_to_delete_with_metadata(partition_id)
            .await
            .context(ListParquetFilesSnafu)?;
        if parquet_files.is_empty() {
            return Ok(compact_and_upgrade);
        }
        compact_and_upgrade.sequencer_id = Some(parquet_files[0].sequencer_id);
        // Group overlapped files
        let overlapped_file_groups = Self::overlapped_groups(parquet_files);
        // Group time-contiguous non-overlapped groups if their total size is smaller than a threshold
        let compact_file_groups = Self::group_small_contiguous_groups(
            overlapped_file_groups,
            compaction_max_size_bytes,
            compaction_max_file_count,
        );
        // Attach appropriate tombstones to each file
        let groups_with_tombstones = self.add_tombstones_to_groups(compact_file_groups).await?;
        info!("compacting {} groups", groups_with_tombstones.len());
        // File groups to compact and files to upgrade
        for group in groups_with_tombstones {
            // Only one file without tombstones, no need to compact.
            if group.parquet_files.len() == 1 && group.tombstones.is_empty() {
                // If it is level 0, upgrade it since it is non-overlapping
                if group.parquet_files[0].compaction_level == INITIAL_COMPACTION_LEVEL {
                    compact_and_upgrade
                        .files_to_upgrade
                        .push(group.parquet_files[0].parquet_file_id())
                }
            } else {
                compact_and_upgrade.groups_to_compact.push(group);
            }
        }
        Ok(compact_and_upgrade)
    }
pub async fn groups_to_compact_and_files_to_upgrade ( & self , partition_id : PartitionId , compaction_max_size_bytes : i64 , // max size of files to get compacted compaction_max_file_count : i64 , // max number of files to get compacted ) -> Result < CompactAndUpgrade > { let mut compact_and_upgrade = CompactAndUpgrade :: new ( None ) ; // List all valid (not soft deletded) files of the partition let parquet_files = self . catalog . repositories ( ) . await . parquet_files ( ) . list_by_partition_not_to_delete_with_metadata ( partition_id ) . await . context ( ListParquetFilesSnafu ) ? ; if parquet_files . is_empty ( ) { return Ok ( compact_and_upgrade ) ; } compact_and_upgrade . sequencer_id = Some ( parquet_files [ 0 ] . sequencer_id ) ; // Group overlapped files let overlapped_file_groups = Self :: overlapped_groups ( parquet_files ) ; // Group time-contiguous non-overlapped groups if their total size is smaller than a threshold let compact_file_groups = Self :: group_small_contiguous_groups ( overlapped_file_groups , compaction_max_size_bytes , compaction_max_file_count , ) ; // Attach appropriate tombstones to each file let groups_with_tombstones = self . add_tombstones_to_groups ( compact_file_groups ) . await ? ; info ! ( " " groups_with_tombstones len ( ) ) ; // File groups to compact and files to upgrade for group in groups_with_tombstones { // Only one file without tombstones, no need to compact. if group . parquet_files . len ( ) == 1 && group . tombstones . is_empty ( ) { // If it is level 0, upgrade it since it is non-overlapping if group . parquet_files [ 0 ] . compaction_level == INITIAL_COMPACTION_LEVEL { compact_and_upgrade . files_to_upgrade . push ( group . parquet_files [ 0 ] . parquet_file_id ( ) ) } } else { compact_and_upgrade . groups_to_compact . push ( group ) ; } } Ok ( compact_and_upgrade ) }
Group files to be compacted together and level-0 files that will get upgraded
for a given partition.
Group files to be compacted together and level - 0 files that will get upgraded for a given partition .
&self
partition_id
PartitionId
compaction_max_size_bytes
i64
compaction_max_file_count
i64
&self
partition_id
PartitionId
compaction_max_size_bytes
i64
compaction_max_file_count
i64