/// Filter out parquet files that contain "too new" data.
///
/// The caller may only use the returned parquet files.
///
/// This will remove files that are part of the catalog but that contain data that the ingester
/// persisted AFTER the querier contacted it. See module-level documentation about the order in
/// which the communication and the information processing should take place.
///
/// Note that the querier (and this method) do NOT care about the actual age of the parquet files,
/// since the compactor is free to to process files at any given moment (e.g. to combine them or to
/// materialize tombstones). However if the compactor combines files in a way that the querier
/// would need to split it into "desired" data and "too new" data then we will currently bail out
/// with [`ReconcileError`].
Filter out parquet files that contain "too new" data.
The caller may only use the returned parquet files.

This will remove files that are part of the catalog but that contain data that the ingester
persisted AFTER the querier contacted it. See module-level documentation about the order in
which the communication and the information processing should take place.

Note that the querier (and this method) do NOT care about the actual age of the parquet files,
since the compactor is free to to process files at any given moment . However if the compactor combines files in a way that the querier
would need to split it into "desired" data and "too new" data then we will currently bail out
with [`ReconcileError`].
Filter out parquet files that contain " too new " data . The caller may only use the returned parquet files . This will remove files that are part of the catalog but that contain data that the ingester persisted AFTER the querier contacted it . See module - level documentation about the order in which the communication and the information processing should take place . Note that the querier ( and this method ) do NOT care about the actual age of the parquet files since the compactor is free to to process files at any given moment . However if the compactor combines files in a way that the querier would need to split it into " desired " data and " too new " data then we will currently bail out with [ ` ReconcileError ` ] .
fn filter_parquet_files<I, P>(
    ingester_partitions: &[I],
    parquet_files: Vec<P>,
) -> Result<Vec<P>, ReconcileError>
where
    I: IngesterPartitionInfo,
    P: ParquetFileInfo,
{
    // Build partition-based lookup table.
    //
    // Note that we don't need to take the sequencer ID into account here because each partition is
    // not only bound to a table but also to a sequencer.
    let lookup_table: HashMap<PartitionId, &I> = ingester_partitions
        .iter()
        .map(|i| (i.partition_id(), i))
        .collect();
    // we assume that we filter out a minimal amount of files, so we can use the same capacity
    let mut result = Vec::with_capacity(parquet_files.len());
    for file in parquet_files {
        if let Some(ingester_partition) = lookup_table.get(&file.partition_id()) {
            if let Some(persisted_max) = ingester_partition.parquet_max_sequence_number() {
                debug!(
                    file_partition_id=%file.partition_id(),
                    file_max_seq_num=%file.max_sequence_number().get(),
                    file_min_seq_num=%file.min_sequence_number().get(),
                    persisted_max=%persisted_max.get(),
                    "Comparing parquet file and ingester parquet max"
                );
                if (file.max_sequence_number() > persisted_max)
                    && (file.min_sequence_number() <= persisted_max)
                {
                    return Err(ReconcileError::CompactorConflict);
                }
                if file.max_sequence_number() > persisted_max {
                    // filter out, file is newer
                    continue;
                }
            } else {
                debug!(
                    file_partition_id=%file.partition_id(),
                    file_max_seq_num=%file.max_sequence_number().get(),
                    file_min_seq_num=%file.min_sequence_number().get(),
                    "ingester thinks it doesn't have data persisted yet"
                );
                // ingester thinks it doesn't have any data persisted yet => can safely ignore file
                continue;
            }
        } else {
            debug!(
                file_partition_id=%file.partition_id(),
                file_max_seq_num=%file.max_sequence_number().get(),
                file_min_seq_num=%file.min_sequence_number().get(),
                "partition was not flagged by the ingester as unpersisted"
            );
            // partition was not flagged by the ingester as "unpersisted", so we can keep the
            // parquet file
        }
        result.push(file);
    }
    Ok(result)
}
fn filter_parquet_files < I , P > ( ingester_partitions : & [ I ] , parquet_files : Vec < P > , ) -> Result < Vec < P > , ReconcileError > where I : IngesterPartitionInfo , P : ParquetFileInfo , { // Build partition-based lookup table. // // Note that we don't need to take the sequencer ID into account here because each partition is // not only bound to a table but also to a sequencer. let lookup_table : HashMap < PartitionId , & I > = ingester_partitions . iter ( ) . map ( | i | ( i . partition_id ( ) , i ) ) . collect ( ) ; // we assume that we filter out a minimal amount of files, so we can use the same capacity let mut result = Vec :: with_capacity ( parquet_files . len ( ) ) ; for file in parquet_files { if let Some ( ingester_partition ) = lookup_table . get ( & file . partition_id ( ) ) { if let Some ( persisted_max ) = ingester_partition . parquet_max_sequence_number ( ) { debug ! ( file_partition_id file partition_id ( ) file_max_seq_num file max_sequence_number ( ) get ( ) file_min_seq_num file min_sequence_number ( ) get ( ) persisted_max persisted_max get ( ) " " ) ; if ( file . max_sequence_number ( ) > persisted_max ) && ( file . min_sequence_number ( ) <= persisted_max ) { return Err ( ReconcileError :: CompactorConflict ) ; } if file . max_sequence_number ( ) > persisted_max { // filter out, file is newer continue ; } } else { debug ! ( file_partition_id file partition_id ( ) file_max_seq_num file max_sequence_number ( ) get ( ) file_min_seq_num file min_sequence_number ( ) get ( ) " " ) ; // ingester thinks it doesn't have any data persisted yet => can safely ignore file continue ; } } else { debug ! ( file_partition_id file partition_id ( ) file_max_seq_num file max_sequence_number ( ) get ( ) file_min_seq_num file min_sequence_number ( ) get ( ) " " ) ; // partition was not flagged by the ingester as "unpersisted", so we can keep the // parquet file } result . push ( file ) ; } Ok ( result ) }
Filter out parquet files that contain "too new" data.
Filter out parquet files that contain " too new " data .
ingester_partitions
&[I]
parquet_files
Vec<P>