/// Perform a pass of garbage collection, querying the catalog for all files marked to be
/// deleted earlier than the specified time. Remove the catalog entries, then remove the
/// associated object store files.
/// Meant to be invoked in a background loop.
Perform a pass of garbage collection, querying the catalog for all files marked to be
deleted earlier than the specified time. Remove the catalog entries, then remove the
associated object store files.
Meant to be invoked in a background loop.
Perform a pass of garbage collection querying the catalog for all files marked to be deleted earlier than the specified time . Remove the catalog entries then remove the associated object store files . Meant to be invoked in a background loop .
pub async fn cleanup(&self, older_than: Timestamp) -> Result<()> {
        let deleted_catalog_records = self
            .catalog
            .repositories()
            .await
            .parquet_files()
            .delete_old(older_than)
            .await
            .context(DeletingCatalogRecordsSnafu)?;
        let mut object_store_errors = Vec::with_capacity(deleted_catalog_records.len());
        for catalog_record in deleted_catalog_records {
            let path = ParquetFilePath::new(
                catalog_record.namespace_id,
                catalog_record.table_id,
                catalog_record.sequencer_id,
                catalog_record.partition_id,
                catalog_record.object_store_id,
            );
            let path = path.object_store_path();
            if let Err(e) = self.object_store.delete(&path).await {
                object_store_errors.push(e);
            }
        }
        if object_store_errors.is_empty() {
            Ok(())
        } else {
            DeletingObjectStoreFilesSnafu {
                sources: object_store_errors,
            }
            .fail()
        }
    }
pub async fn cleanup ( & self , older_than : Timestamp ) -> Result < ( ) > { let deleted_catalog_records = self . catalog . repositories ( ) . await . parquet_files ( ) . delete_old ( older_than ) . await . context ( DeletingCatalogRecordsSnafu ) ? ; let mut object_store_errors = Vec :: with_capacity ( deleted_catalog_records . len ( ) ) ; for catalog_record in deleted_catalog_records { let path = ParquetFilePath :: new ( catalog_record . namespace_id , catalog_record . table_id , catalog_record . sequencer_id , catalog_record . partition_id , catalog_record . object_store_id , ) ; let path = path . object_store_path ( ) ; if let Err ( e ) = self . object_store . delete ( & path ) . await { object_store_errors . push ( e ) ; } } if object_store_errors . is_empty ( ) { Ok ( ( ) ) } else { DeletingObjectStoreFilesSnafu { sources : object_store_errors , } . fail ( ) } }
Perform a pass of garbage collection, querying the catalog for all files marked to be
deleted earlier than the specified time.
Perform a pass of garbage collection querying the catalog for all files marked to be deleted earlier than the specified time .
&self
older_than
Timestamp