/// Generates "exclude" filter for tombstones.
///
/// Since tombstones are sequencer-wide but data persistence is partition-based (which are
/// sub-units of sequencers), we cannot just remove tombstones entirely but need to decide on a
/// per-partition basis. This function generates a lookup table of partition-tombstone tuples that
/// later need to be EXCLUDED/IGNORED when pairing tombstones with chunks.
Generates "exclude" filter for tombstones.
Since tombstones are sequencer-wide but data persistence is partition-based (which are
sub-units of sequencers), we cannot just remove tombstones entirely but need to decide on a
per-partition basis. This function generates a lookup table of partition-tombstone tuples that
later need to be EXCLUDED/IGNORED when pairing tombstones with chunks.
Generates " exclude " filter for tombstones . Since tombstones are sequencer - wide but data persistence is partition - based ( which are sub - units of sequencers ) we cannot just remove tombstones entirely but need to decide on a per - partition basis . This function generates a lookup table of partition - tombstone tuples that later need to be EXCLUDED / IGNORED when pairing tombstones with chunks .
fn tombstone_exclude_list<I, T>(
    ingester_partitions: &[I],
    tombstones: &[T],
) -> HashSet<(PartitionId, TombstoneId)>
where
    I: IngesterPartitionInfo,
    T: TombstoneInfo,
{
    // Build sequencer-based lookup table.
    let mut lookup_table: HashMap<SequencerId, Vec<&I>> = HashMap::default();
    for partition in ingester_partitions {
        lookup_table
            .entry(partition.sequencer_id())
            .or_default()
            .push(partition);
    }
    let mut exclude = HashSet::new();
    for t in tombstones {
        if let Some(partitions) = lookup_table.get(&t.sequencer_id()) {
            for p in partitions {
                if let Some(persisted_max) = p.tombstone_max_sequence_number() {
                    if t.sequence_number() > persisted_max {
                        // newer than persisted => exclude
                        exclude.insert((p.partition_id(), t.id()));
                    } else {
                        // in persisted range => keep
                    }
                } else {
                    // partition has no persisted data at all => need to exclude tombstone which is
                    // too new
                    exclude.insert((p.partition_id(), t.id()));
                }
            }
        }
    }
    exclude
}
fn tombstone_exclude_list < I , T > ( ingester_partitions : & [ I ] , tombstones : & [ T ] , ) -> HashSet < ( PartitionId , TombstoneId ) > where I : IngesterPartitionInfo , T : TombstoneInfo , { // Build sequencer-based lookup table. let mut lookup_table : HashMap < SequencerId , Vec < & I > > = HashMap :: default ( ) ; for partition in ingester_partitions { lookup_table . entry ( partition . sequencer_id ( ) ) . or_default ( ) . push ( partition ) ; } let mut exclude = HashSet :: new ( ) ; for t in tombstones { if let Some ( partitions ) = lookup_table . get ( & t . sequencer_id ( ) ) { for p in partitions { if let Some ( persisted_max ) = p . tombstone_max_sequence_number ( ) { if t . sequence_number ( ) > persisted_max { // newer than persisted => exclude exclude . insert ( ( p . partition_id ( ) , t . id ( ) ) ) ; } else { // in persisted range => keep } } else { // partition has no persisted data at all => need to exclude tombstone which is // too new exclude . insert ( ( p . partition_id ( ) , t . id ( ) ) ) ; } } } } exclude }
Generates "exclude" filter for tombstones.
Generates " exclude " filter for tombstones .
ingester_partitions
&[I]
tombstones
&[T]