Repository: qdrant/qdrant
File: lib/segment/src/id_tracker/mutable_id_tracker.rs (Rust)

Code before:
fn reconcile_persisted_version_changes(
    mut pending: MutexGuard<'_, BTreeMap<PointOffsetType, SeqNumberType>>,
    changes: BTreeMap<PointOffsetType, SeqNumberType>,
) {
    for (point_offset, persisted_version) in changes {
        // Remove persisted version
        let Some(pending_version) = pending.remove(&point_offset) else {
            continue;
        };

        // Reinsert if it has changed.
        if pending_version != persisted_version {
            pending.insert(point_offset, pending_version);
        }
    }
}

fn reconcile_persisted_mapping_changes(
    mut pending: MutexGuard<'_, Vec<MappingChange>>,
    changes: &Vec<MappingChange>,
) {
    let mut delete_up_to = 0;
    for (pending, persisted) in pending.iter().zip(changes) {
        if pending != persisted {
            // This should not happen, since flushers are supposed to be requested->executed
            // one at a time, but if it does, it should be fine.
            //
            // The only consequence is to persist some operations again.
            break;
        }
        delete_up_to += 1;
    }

    *pending = pending.split_off(delete_up_to);
}

#[cfg(test)]
pub(super) mod tests {
    use std::collections::{HashMap, HashSet};
    use std::io::Cursor;

    use fs_err as fs;
    use itertools::Itertools;
    #[cfg(feature = "rocksdb")]
    use rand::Rng;
    use rand::prelude::*;
    use tempfile::Builder;
    use uuid::Uuid;

    use super::*;
    use crate::id_tracker::compressed::compressed_point_mappings::CompressedPointMappings;

Review comment (performance):
We can probably use `retain` here too. As I remember it is very efficient on diffs like this. And it prevents us having to do this index magic.

Also, this indexed approach will break if we have concurrent flushers. We shouldn't have that, but you never know what happens in the future.

Code after:
) {
    pending.lock().retain(|point_offset, pending_version| {
        changes
            .get(point_offset)
            .is_none_or(|persisted_version| pending_version != persisted_version)
    });
}

fn reconcile_persisted_mapping_changes(
    pending: &Mutex<Vec<MappingChange>>,
    changes: &Vec<MappingChange>,
) {
    let mut pending = pending.lock();
    let mut delete_up_to = 0;
    for (pending, persisted) in pending.iter().zip(changes) {
        if pending != persisted {
            // This should not happen, since flushers are supposed to be requested->executed
            // one at a time, but if it does, it should be fine.
            //
            // The only consequence is to persist some operations again.
            break;
        }
        delete_up_to += 1;
    }

    *pending = pending.split_off(delete_up_to);
}

#[cfg(test)]
pub(super) mod tests {
    use std::collections::{HashMap, HashSet};
    use std::io::Cursor;

    use fs_err as fs;
    use itertools::Itertools;
    #[cfg(feature = "rocksdb")]
    use rand::Rng;
    use rand::prelude::*;
    use tempfile::Builder;
    use uuid::Uuid;

    use super::*;
    use crate::id_tracker::compressed::compressed_point_mappings::CompressedPointMappings;
    use crate::id_tracker::in_memory_id_tracker::InMemoryIdTracker;
    #[cfg(feature = "rocksdb")]
    use crate::id_tracker::simple_id_tracker::SimpleIdTracker;

    const RAND_SEED: u64 = 42;
    const DEFAULT_VERSION: SeqNumberType = 42;

    #[test]