/// Delete all unreferenced parquet files.
///
/// This will hold the transaction lock while the list of files is being gathered. To limit the time the lock is held
/// use `max_files` which will limit the number of files to delete in this cleanup round.
Delete all unreferenced parquet files.
This will hold the transaction lock while the list of files is being gathered. To limit the time the lock is held
use `max_files` which will limit the number of files to delete in this cleanup round.
Delete all unreferenced parquet files . This will hold the transaction lock while the list of files is being gathered . To limit the time the lock is held use ` max_files ` which will limit the number of files to delete in this cleanup round .
pub async fn cleanup_unreferenced_parquet_files<S>(
    catalog: &PreservedCatalog<S>,
    max_files: usize,
) -> Result<()>
where
    S: CatalogState,
{
    // Create a transaction to prevent parallel modifications of the catalog. This avoids that we delete files there
    // that are about to get added to the catalog.
    let transaction = catalog.open_transaction().await;
    let store = catalog.object_store();
    let server_id = catalog.server_id();
    let db_name = catalog.db_name();
    let all_known = {
        // replay catalog transactions to track ALL (even dropped) files that are referenced
        let catalog = PreservedCatalog::<TracerCatalogState>::load(
            Arc::clone(&store),
            server_id,
            db_name.to_string(),
            (),
        )
        .await
        .context(CatalogLoadError)?
        .expect("catalog gone while reading it?");
        catalog
            .state()
            .files
            .lock()
            .expect("lock poissened?")
            .clone()
    };
    let prefix = data_location(&store, server_id, db_name);
    // gather a list of "files to remove" eagerly so we do not block transactions on the catalog for too long
    let mut to_remove = vec![];
    let mut stream = store.list(Some(&prefix)).await.context(ReadError)?;
    'outer: while let Some(paths) = stream.try_next().await.context(ReadError)? {
        for path in paths {
            if to_remove.len() >= max_files {
                info!(%max_files, "reached limit of number of files to cleanup in one go");
                break 'outer;
            }
            let path_parsed: DirsAndFileName = path.clone().into();
            // only delete if all of the following conditions are met:
            // - filename ends with `.parquet`
            // - file is not tracked by the catalog
            if path_parsed
                .file_name
                .as_ref()
                .map(|part| part.encoded().ends_with(".parquet"))
                .unwrap_or(false)
                && !all_known.contains(&path_parsed)
            {
                to_remove.push(path);
            }
        }
    }
    // abort transaction cleanly to avoid warnings about uncommited transactions
    transaction.abort();
    // now that the transaction lock is dropped, perform the actual (and potentially slow) delete operation
    let n_files = to_remove.len();
    info!(%n_files, "Found files to delete, start deletion.");
    for path in to_remove {
        info!(path = %path.display(), "Delete file");
        store.delete(&path).await.context(WriteError)?;
    }
    info!(%n_files, "Finished deletion, removed files.");
    Ok(())
}
pub async fn cleanup_unreferenced_parquet_files < S > ( catalog : & PreservedCatalog < S > , max_files : usize , ) -> Result < ( ) > where S : CatalogState , { // Create a transaction to prevent parallel modifications of the catalog. This avoids that we delete files there // that are about to get added to the catalog. let transaction = catalog . open_transaction ( ) . await ; let store = catalog . object_store ( ) ; let server_id = catalog . server_id ( ) ; let db_name = catalog . db_name ( ) ; let all_known = { // replay catalog transactions to track ALL (even dropped) files that are referenced let catalog = PreservedCatalog :: < TracerCatalogState > :: load ( Arc :: clone ( & store ) , server_id , db_name . to_string ( ) , ( ) , ) . await . context ( CatalogLoadError ) ? . expect ( " " ) ; catalog . state ( ) . files . lock ( ) . expect ( " " ) . clone ( ) } ; let prefix = data_location ( & store , server_id , db_name ) ; // gather a list of "files to remove" eagerly so we do not block transactions on the catalog for too long let mut to_remove = vec ! [ ] ; let mut stream = store . list ( Some ( & prefix ) ) . await . context ( ReadError ) ? ; ' outer : while let Some ( paths ) = stream . try_next ( ) . await . context ( ReadError ) ? { for path in paths { if to_remove . len ( ) >= max_files { info ! ( max_files " " ) ; break ' outer ; } let path_parsed : DirsAndFileName = path . clone ( ) . into ( ) ; // only delete if all of the following conditions are met: // - filename ends with `.parquet` // - file is not tracked by the catalog if path_parsed . file_name . as_ref ( ) . map ( | part | part . encoded ( ) . ends_with ( " " ) ) . unwrap_or ( false ) && ! all_known . contains ( & path_parsed ) { to_remove . push ( path ) ; } } } // abort transaction cleanly to avoid warnings about uncommited transactions transaction . abort ( ) ; // now that the transaction lock is dropped, perform the actual (and potentially slow) delete operation let n_files = to_remove . len ( ) ; info ! ( n_files " " ) ; for path in to_remove { info ! ( path path display ( ) " " ) ; store . delete ( & path ) . await . context ( WriteError ) ? ; } info ! ( n_files " " ) ; Ok ( ( ) ) }
Delete all unreferenced parquet files.
Delete all unreferenced parquet files .
catalog
&PreservedCatalog<S>
max_files
usize