/// Collect all files under the given locations.
///
/// Returns a map of revisions to their UUIDs and a vector of file-metadata tuples.
///
/// The file listing is recursive.
Collect all files under the given locations.
Returns a map of revisions to their UUIDs and a vector of file-metadata tuples.

The file listing is recursive.
Collect all files under the given locations . Returns a map of revisions to their UUIDs and a vector of file - metadata tuples . The file listing is recursive .
async fn collect_revisions(
    object_store: &ObjectStore,
    search_location: &Path,
    ignore_metadata_read_failure: bool,
) -> Result<HashMap<u64, (Uuid, Vec<(Path, ParquetMetaData)>)>> {
    let mut stream = object_store
        .list(Some(search_location))
        .await
        .context(ReadFailure)?;
    // revision -> (uuid, [file])
    let mut revisions: HashMap<u64, (Uuid, Vec<(Path, ParquetMetaData)>)> = HashMap::new();
    while let Some(paths) = stream.try_next().await.context(ReadFailure)? {
        for path in paths.into_iter().filter(is_parquet) {
            let (iox_md, parquet_md) = match read_parquet(object_store, &path).await {
                Ok(res) => res,
                Err(e @ Error::MetadataReadFailure { .. }) if ignore_metadata_read_failure => {
                    error!("error while reading metdata from parquet, ignoring: {}", e);
                    continue;
                }
                Err(e) => return Err(e),
            };
            // revision 0 can never occur because it is always empty
            if iox_md.transaction_revision_counter == 0 {
                return Err(Error::RevisionZeroFailure { path });
            }
            match revisions.entry(iox_md.transaction_revision_counter) {
                Entry::Vacant(v) => {
                    // revision not known yet => create it
                    v.insert((iox_md.transaction_uuid, vec![(path, parquet_md)]));
                }
                Entry::Occupied(mut o) => {
                    // already exist => check UUID
                    let (uuid, entries) = o.get_mut();
                    if *uuid != iox_md.transaction_uuid {
                        // found multiple transactions for this revision => cannot rebuild cleanly
                        // sort UUIDs for deterministic error messages
                        let (uuid1, uuid2) = if *uuid < iox_md.transaction_uuid {
                            (*uuid, iox_md.transaction_uuid)
                        } else {
                            (iox_md.transaction_uuid, *uuid)
                        };
                        return Err(Error::MultipleTransactionsFailure {
                            revision_counter: iox_md.transaction_revision_counter,
                            uuid1,
                            uuid2,
                        });
                    }
                    entries.push((path, parquet_md));
                }
            }
        }
    }
    Ok(revisions)
}
async fn collect_revisions ( object_store : & ObjectStore , search_location : & Path , ignore_metadata_read_failure : bool , ) -> Result < HashMap < u64 , ( Uuid , Vec < ( Path , ParquetMetaData ) > ) > > { let mut stream = object_store . list ( Some ( search_location ) ) . await . context ( ReadFailure ) ? ; // revision -> (uuid, [file]) let mut revisions : HashMap < u64 , ( Uuid , Vec < ( Path , ParquetMetaData ) > ) > = HashMap :: new ( ) ; while let Some ( paths ) = stream . try_next ( ) . await . context ( ReadFailure ) ? { for path in paths . into_iter ( ) . filter ( is_parquet ) { let ( iox_md , parquet_md ) = match read_parquet ( object_store , & path ) . await { Ok ( res ) => res , Err ( e @ Error :: MetadataReadFailure { .. } ) if ignore_metadata_read_failure => { error ! ( " " e ) ; continue ; } Err ( e ) => return Err ( e ) , } ; // revision 0 can never occur because it is always empty if iox_md . transaction_revision_counter == 0 { return Err ( Error :: RevisionZeroFailure { path } ) ; } match revisions . entry ( iox_md . transaction_revision_counter ) { Entry :: Vacant ( v ) => { // revision not known yet => create it v . insert ( ( iox_md . transaction_uuid , vec ! [ ( path parquet_md ) ] ) ) ; } Entry :: Occupied ( mut o ) => { // already exist => check UUID let ( uuid , entries ) = o . get_mut ( ) ; if * uuid != iox_md . transaction_uuid { // found multiple transactions for this revision => cannot rebuild cleanly // sort UUIDs for deterministic error messages let ( uuid1 , uuid2 ) = if * uuid < iox_md . transaction_uuid { ( * uuid , iox_md . transaction_uuid ) } else { ( iox_md . transaction_uuid , * uuid ) } ; return Err ( Error :: MultipleTransactionsFailure { revision_counter : iox_md . transaction_revision_counter , uuid1 , uuid2 , } ) ; } entries . push ( ( path , parquet_md ) ) ; } } } } Ok ( revisions ) }
Collect all files under the given locations.
Collect all files under the given locations .
object_store
&ObjectStore
search_location
&Path
ignore_metadata_read_failure
bool