/// Runs compaction in a partition resolving any tombstones and compacting data so that parquet
/// files will be non-overlapping in time.
Runs compaction in a partition resolving any tombstones and compacting data so that parquet
files will be non-overlapping in time.
Runs compaction in a partition resolving any tombstones and compacting data so that parquet files will be non - overlapping in time .
pub async fn compact_partition(
        &self,
        partition_id: PartitionId,
        compact_and_upgrade: CompactAndUpgrade,
    ) -> Result<()> {
        if !compact_and_upgrade.compactable() {
            return Ok(());
        }
        info!("compacting partition {}", partition_id);
        let start_time = self.time_provider.now();
        let partition = self.get_partition_from_catalog(partition_id).await?;
        let mut file_count = 0;
        // Compact, persist,and update catalog accordingly for each overlapped file
        let mut tombstones = BTreeMap::new();
        let mut output_file_count = 0;
        let sequencer_id = compact_and_upgrade
            .sequencer_id()
            .expect("Should have sequencer ID");
        for group in compact_and_upgrade.groups_to_compact {
            file_count += group.parquet_files.len();
            // keep tombstone ids
            tombstones = Self::union_tombstones(tombstones, &group);
            // Collect all the parquet file IDs, to be able to set their catalog records to be
            // deleted. These should already be unique, no need to dedupe.
            let original_parquet_file_ids: Vec<_> =
                group.parquet_files.iter().map(|f| f.id).collect();
            let size: i64 = group.parquet_files.iter().map(|f| f.file_size_bytes).sum();
            info!(num_files=%group.parquet_files.len(), ?size, ?original_parquet_file_ids, "compacting group of files");
            // Compact the files concurrently.
            //
            // This builds the StreamSplitExec plan & executes both partitions
            // concurrently, streaming the resulting record batches into the
            // Parquet serializer and directly uploads them to object store.
            //
            // If an non-object-store upload error occurs, all
            // executions/uploads are aborted and the error is returned to the
            // caller. If an error occurs during object store upload, it will be
            // retried indefinitely.
            let catalog_update_info = self
                .compact(group.parquet_files, &partition)
                .await?
                .into_iter()
                .map(|v| async {
                    let CompactedData {
                        data,
                        meta,
                        tombstones,
                    } = v;
                    debug!(
                        ?partition_id,
                        ?meta,
                        "executing and uploading compaction StreamSplitExec"
                    );
                    let object_store_id = meta.object_store_id;
                    info!(?partition_id, %object_store_id, "streaming exec to object store");
                    // Stream the record batches from the compaction exec, serialize
                    // them, and directly upload the resulting Parquet files to
                    // object storage.
                    let (parquet_meta, file_size) =
                        self.store.upload(data, &meta).await.context(PersistSnafu)?;
                    debug!(?partition_id, %object_store_id, "file uploaded to object store");
                    Ok(CatalogUpdate::new(
                        partition_id,
                        meta,
                        file_size,
                        parquet_meta,
                        tombstones,
                    ))
                })
                .collect::<FuturesUnordered<_>>()
                .try_collect::<Vec<_>>()
                .await?;
            output_file_count += catalog_update_info.len();
            let mut txn = self
                .catalog
                .start_transaction()
                .await
                .context(TransactionSnafu)?;
            debug!(
                "updating catalog with {} updates",
                catalog_update_info.len()
            );
            self.update_catalog(
                catalog_update_info,
                original_parquet_file_ids,
                txn.deref_mut(),
            )
            .await?;
            txn.commit().await.context(TransactionCommitSnafu)?;
        }
        // Remove fully processed tombstones
        self.remove_fully_processed_tombstones(tombstones).await?;
        // Upgrade old level-0 to level 1
        self.update_to_level_1(&compact_and_upgrade.files_to_upgrade)
            .await?;
        let attributes = Attributes::from([("sequencer_id", format!("{}", sequencer_id).into())]);
        if !compact_and_upgrade.files_to_upgrade.is_empty() {
            let promotion_counter = self.level_promotion_counter.recorder(attributes.clone());
            promotion_counter.inc(compact_and_upgrade.files_to_upgrade.len() as u64);
        }
        if let Some(delta) = self.time_provider.now().checked_duration_since(start_time) {
            let duration_ms = self.compaction_duration_ms.recorder(attributes.clone());
            duration_ms.record(delta.as_millis() as _);
        }
        let compaction_counter = self.compaction_counter.recorder(attributes.clone());
        compaction_counter.inc(file_count as u64);
        let compaction_output_counter = self.compaction_output_counter.recorder(attributes);
        compaction_output_counter.inc(output_file_count as u64);
        Ok(())
    }
pub async fn compact_partition ( & self , partition_id : PartitionId , compact_and_upgrade : CompactAndUpgrade , ) -> Result < ( ) > { if ! compact_and_upgrade . compactable ( ) { return Ok ( ( ) ) ; } info ! ( " " partition_id ) ; let start_time = self . time_provider . now ( ) ; let partition = self . get_partition_from_catalog ( partition_id ) . await ? ; let mut file_count = 0 ; // Compact, persist,and update catalog accordingly for each overlapped file let mut tombstones = BTreeMap :: new ( ) ; let mut output_file_count = 0 ; let sequencer_id = compact_and_upgrade . sequencer_id ( ) . expect ( " " ) ; for group in compact_and_upgrade . groups_to_compact { file_count += group . parquet_files . len ( ) ; // keep tombstone ids tombstones = Self :: union_tombstones ( tombstones , & group ) ; // Collect all the parquet file IDs, to be able to set their catalog records to be // deleted. These should already be unique, no need to dedupe. let original_parquet_file_ids : Vec < _ > = group . parquet_files . iter ( ) . map ( | f | f . id ) . collect ( ) ; let size : i64 = group . parquet_files . iter ( ) . map ( | f | f . file_size_bytes ) . sum ( ) ; info ! ( num_files group parquet_files len ( ) size original_parquet_file_ids " " ) ; // Compact the files concurrently. // // This builds the StreamSplitExec plan & executes both partitions // concurrently, streaming the resulting record batches into the // Parquet serializer and directly uploads them to object store. // // If an non-object-store upload error occurs, all // executions/uploads are aborted and the error is returned to the // caller. If an error occurs during object store upload, it will be // retried indefinitely. let catalog_update_info = self . compact ( group . parquet_files , & partition ) . await ? . into_iter ( ) . map ( | v | async { let CompactedData { data , meta , tombstones , } = v ; debug ! ( partition_id meta " " ) ; let object_store_id = meta . object_store_id ; info ! ( partition_id object_store_id " " ) ; // Stream the record batches from the compaction exec, serialize // them, and directly upload the resulting Parquet files to // object storage. let ( parquet_meta , file_size ) = self . store . upload ( data , & meta ) . await . context ( PersistSnafu ) ? ; debug ! ( partition_id object_store_id " " ) ; Ok ( CatalogUpdate :: new ( partition_id , meta , file_size , parquet_meta , tombstones , ) ) } ) . collect :: < FuturesUnordered < _ > > ( ) . try_collect :: < Vec < _ > > ( ) . await ? ; output_file_count += catalog_update_info . len ( ) ; let mut txn = self . catalog . start_transaction ( ) . await . context ( TransactionSnafu ) ? ; debug ! ( " " catalog_update_info len ( ) ) ; self . update_catalog ( catalog_update_info , original_parquet_file_ids , txn . deref_mut ( ) , ) . await ? ; txn . commit ( ) . await . context ( TransactionCommitSnafu ) ? ; } // Remove fully processed tombstones self . remove_fully_processed_tombstones ( tombstones ) . await ? ; // Upgrade old level-0 to level 1 self . update_to_level_1 ( & compact_and_upgrade . files_to_upgrade ) . await ? ; let attributes = Attributes :: from ( [ ( " " , format ! ( " " sequencer_id ) . into ( ) ) ] ) ; if ! compact_and_upgrade . files_to_upgrade . is_empty ( ) { let promotion_counter = self . level_promotion_counter . recorder ( attributes . clone ( ) ) ; promotion_counter . inc ( compact_and_upgrade . files_to_upgrade . len ( ) as u64 ) ; } if let Some ( delta ) = self . time_provider . now ( ) . checked_duration_since ( start_time ) { let duration_ms = self . compaction_duration_ms . recorder ( attributes . clone ( ) ) ; duration_ms . record ( delta . as_millis ( ) as _ ) ; } let compaction_counter = self . compaction_counter . recorder ( attributes . clone ( ) ) ; compaction_counter . inc ( file_count as u64 ) ; let compaction_output_counter = self . compaction_output_counter . recorder ( attributes ) ; compaction_output_counter . inc ( output_file_count as u64 ) ; Ok ( ( ) ) }
Runs compaction in a partition resolving any tombstones and compacting data so that parquet
files will be non-overlapping in time.
Runs compaction in a partition resolving any tombstones and compacting data so that parquet files will be non - overlapping in time .
&self
partition_id
PartitionId
compact_and_upgrade
CompactAndUpgrade