diff --git a/lib/segment/Cargo.toml b/lib/segment/Cargo.toml index 254e2fbb126..08c0c71516b 100644 --- a/lib/segment/Cargo.toml +++ b/lib/segment/Cargo.toml @@ -168,6 +168,10 @@ harness = false name = "id_type_benchmark" harness = false +[[bench]] +name = "id_tracker_lookup" +harness = false + [[bench]] name = "map_benchmark" harness = false diff --git a/lib/segment/benches/id_tracker_lookup.rs b/lib/segment/benches/id_tracker_lookup.rs new file mode 100644 index 00000000000..c0aeff45bdb --- /dev/null +++ b/lib/segment/benches/id_tracker_lookup.rs @@ -0,0 +1,146 @@ +//! Realistic id-tracker lookup costs. Public API only, so this file compiles +//! unchanged with and without the Bloom pre-screen. +//! +//! Two things the first cut of this bench got wrong, corrected here: +//! - a small probe set kept the filter's working set L2-resident, flattering +//! the miss numbers; PROBES is now large enough to defeat that. +//! - only numeric ids were exercised, so UUID-keyed collections — where the +//! map holds 16-byte keys — went unmeasured. + +use std::hint::black_box; + +use common::types::DeferredBehavior; +use criterion::{Criterion, criterion_group, criterion_main}; +use segment::id_tracker::mutable_id_tracker::MutableIdTracker; +use segment::id_tracker::{IdTracker, IdTrackerRead}; +use segment::types::PointIdType; +use tempfile::TempDir; +use uuid::Uuid; + +/// Points per tracker. The 10B deployment targets ~5.4M per segment; the +/// default is kept lower so the bench is cheap to run, and the ranking of the +/// variants does not change with it. +const POINTS: u64 = 1_000_000; +/// Large enough that the probe stream cannot sit in L2. +const PROBES: usize = 1 << 17; + +fn spread(i: u64) -> u64 { + i.wrapping_mul(0x9E37_79B9_7F4A_7C15) +} + +/// Only even slots are populated, so the odd ids used as absent keys interleave +/// with live ones. Absent keys drawn from beyond the maximum key would all +/// descend the same rightmost spine, which stays cached and makes a `BTreeMap` +/// miss look far cheaper than a real one. +fn key(i: u64, as_uuid: bool) -> PointIdType { + if as_uuid { + PointIdType::Uuid(Uuid::from_u128(u128::from(spread(i)))) + } else { + PointIdType::NumId(spread(i)) + } +} + +fn build(points: u64, as_uuid: bool) -> (TempDir, MutableIdTracker) { + let dir = TempDir::new().unwrap(); + let mut tracker = MutableIdTracker::open(dir.path(), None).unwrap(); + for i in 0..points { + tracker.set_link(key(i * 2, as_uuid), i as u32).unwrap(); + } + (dir, tracker) +} + +fn probe(c: &mut Criterion, name: &str, tracker: &MutableIdTracker, keys: Vec) { + c.bench_function(name, |b| { + let mut next = 0usize; + b.iter(|| { + let id = keys[next & (PROBES - 1)]; + next += 1; + black_box( + tracker.internal_id_with_behavior(black_box(id), DeferredBehavior::WithDeferred), + ) + }) + }); +} + +fn realistic(c: &mut Criterion) { + // Which mode this run measured. Set QDRANT_ID_TRACKER_BLOOM_FILTER=0 to + // get the unfiltered baseline without switching branches. + println!( + "\n=== external id pre-screen: {} ===", + if segment::id_tracker::external_id_filter::is_enabled() { + "ENABLED" + } else { + "DISABLED" + }, + ); + + // ---- numeric-keyed collection ---- + let (dir, tracker) = build(POINTS, false); + println!("\nnumeric tracker RAM: {} bytes", tracker.ram_usage_bytes()); + probe( + c, + "num/miss", + &tracker, + (0..PROBES) + .map(|i| key((i as u64 % POINTS) * 2 + 1, false)) + .collect(), + ); + probe( + c, + "num/hit", + &tracker, + (0..PROBES) + .map(|i| key((i as u64 % POINTS) * 2, false)) + .collect(), + ); + drop(tracker); + drop(dir); + + // ---- uuid-keyed collection ---- + let (dir, tracker) = build(POINTS, true); + println!("uuid tracker RAM: {} bytes\n", tracker.ram_usage_bytes()); + probe( + c, + "uuid/miss", + &tracker, + (0..PROBES) + .map(|i| key((i as u64 % POINTS) * 2 + 1, true)) + .collect(), + ); + probe( + c, + "uuid/hit", + &tracker, + (0..PROBES) + .map(|i| key((i as u64 % POINTS) * 2, true)) + .collect(), + ); + drop(tracker); + drop(dir); +} + +fn id_tracker_load(c: &mut Criterion) { + // Build once, flush, then measure repeated cold reopens. On a filtered + // branch this includes seeding the pre-screen for every replayed link. + let (dir, tracker) = build(200_000, false); + let flush = tracker.mapping_flusher(); + flush().unwrap(); + drop(tracker); + + c.bench_function("load/reopen_200k", |b| { + b.iter(|| { + let t = MutableIdTracker::open(dir.path(), None).unwrap(); + black_box(t.total_point_count()) + }) + }); + + drop(dir); +} + +criterion_group! { + name = benches; + config = Criterion::default().sample_size(10).measurement_time(std::time::Duration::from_secs(4)); + targets = realistic, id_tracker_load +} + +criterion_main!(benches); diff --git a/lib/segment/src/id_tracker/external_id_filter.rs b/lib/segment/src/id_tracker/external_id_filter.rs new file mode 100644 index 00000000000..4473014c330 --- /dev/null +++ b/lib/segment/src/id_tracker/external_id_filter.rs @@ -0,0 +1,634 @@ +//! Blocked Bloom Filter pre-screen for external → internal id lookups. +//! +//! To turn off the bloom filter, set `QDRANT_ID_TRACKER_BLOOM_FILTER=0` +//! (or `false`/`off`/`no`) +//! +//! # Why this needs no delete handling and no persistence +//! +//! The filter tracks *ever inserted*, which is a superset of *currently live*. +//! That makes both of the usual Bloom-filter complications disappear: +//! +//! - **Deletes need no hook.** Dropping a point leaves its bits set. The stale +//! bit costs a false positive, which falls through to the `BTreeMap` and gets +//! the correct answer — exactly today's behaviour. The one invariant is that +//! a bit is *never cleared*, so there are no false negatives. +//! - **Restarts need no format.** The filter is never written to disk. +//! `MutableIdTracker::open` always starts from an empty `PointMappings` and +//! replays the persisted change log through `set_link`, so the same insert +//! hook that maintains the filter at runtime rebuilds it during recovery. +//! +//! Stale bits accumulate over a segment's life, but appendable segments are +//! rolled into fresh ones by the optimizer, and each new segment builds a fresh +//! filter — so the false-positive rate is bounded by the segment lifecycle +//! rather than by any compaction logic here. +//! + +use std::fmt; +use std::sync::LazyLock; +use std::sync::atomic::{AtomicBool, Ordering}; + +use crate::types::PointIdType; + +/// Lanes per block. Each key sets exactly one bit in every lane, so a lookup +/// is eight independent lane tests with no dependency chain between them — +/// the shape LLVM turns into vector code. +const LANES: usize = 8; + +/// A block is [`LANES`] x 32 bits = 32 bytes. Aligned to its own size, so it +/// never straddles a cache line: one probe, one cache miss. +const BITS_PER_BLOCK: usize = LANES * u32::BITS as usize; + +/// ~1.3% false-positive rate for a split-block filter at this sizing. +const BITS_PER_KEY: usize = 10; + +/// Odd multipliers, one per lane, from the Impala/Parquet split-block filter. +/// Multiplying the key by a distinct odd constant and keeping the top 5 bits +/// gives each lane an independent bit index in `0..32`. +const SALT: [u32; LANES] = [ + 0x47b6_137b, + 0x4497_4d91, + 0x8824_ad5b, + 0xa2b7_289d, + 0x7054_95c7, + 0x2df1_424b, + 0x9efc_4947, + 0x5c6b_fb31, +]; + +/// Smallest filter worth allocating: 5 KiB +const MIN_CAPACITY: usize = 4096; + +/// Separates the numeric and UUID key spaces so a UUID whose low bits happen to +/// equal a numeric id does not systematically collide with it. +const UUID_DOMAIN: u64 = 0xD1B5_4A32_D192_ED03; + +/// Folds a UUID's high half into its low half before mixing. Odd, so the +/// multiply is invertible and cannot collapse distinct high halves. +const UUID_FOLD: u64 = 0x9E37_79B9_7F4A_7C15; + +/// Environment variable that turns the pre-screen off. +/// +/// Set to `0`, `false`, `off`, or `no` to disable. Anything else, or unset, +/// leaves it on. +pub const ENABLE_ENV_VAR: &str = "QDRANT_ID_TRACKER_BLOOM_FILTER"; + +/// Whether filters created from here on are enabled. Read from the environment +/// once, then overridable via [`set_enabled`]. +static ENABLED: LazyLock = LazyLock::new(|| { + let enabled = enabled_from_env(); + // Logged once per process, on the first filter built. Without it a running + // node gives no way to tell which mode it is in, which matters most while + // rolling the pre-screen out or bisecting a regression against it. + if enabled { + log::debug!("External id pre-screen enabled (disable with {ENABLE_ENV_VAR}=0)"); + } else { + log::info!("External id pre-screen disabled via {ENABLE_ENV_VAR}"); + } + AtomicBool::new(enabled) +}); + +/// Whether new filters are enabled. +pub fn is_enabled() -> bool { + ENABLED.load(Ordering::Relaxed) +} + +/// Turn the pre-screen on or off for filters created from here on. +pub fn set_enabled(enabled: bool) { + ENABLED.store(enabled, Ordering::Relaxed); +} + +/// Parse the toggle from a raw value. Absent, empty, or anything not spelled +/// like "off" leaves the pre-screen on. +fn enabled_from_value(value: Option<&str>) -> bool { + match value { + Some(value) => !matches!( + value.trim().to_ascii_lowercase().as_str(), + "0" | "false" | "off" | "no" + ), + None => true, + } +} + +fn enabled_from_env() -> bool { + enabled_from_value(std::env::var(ENABLE_ENV_VAR).ok().as_deref()) +} + +/// One block of the filter: [`LANES`] 32-bit lanes, aligned to its own size so +/// it cannot straddle a cache line. All bits for a key live in one block, so a +/// negative lookup is a single cache miss instead of one per `BTreeMap` level. +#[repr(align(32))] +#[derive(Clone, Copy, Default)] +struct Block([u32; LANES]); + +/// Set-membership pre-screen over the external ids held by a `PointMappings`. +/// +/// A `false` from [`maybe_contains`](Self::maybe_contains) is authoritative: +/// the id is definitely absent. A `true` means "maybe" and must be confirmed +/// against the map. +#[derive(Clone)] +pub struct ExternalIdFilter { + /// Empty when the filter is disabled, in which case every lookup answers + /// "maybe" and the caller always falls through to the map. + blocks: Vec, + /// Key count this filter was sized for. + capacity: usize, + /// Keys recorded since it was built. Counts insertions, not live keys — + /// re-inserting an existing key still moves it, which is what makes it a + /// conservative trigger for a rebuild. + inserted: usize, +} + +impl ExternalIdFilter { + /// Dummy filter that allocates nothing and screens nothing out. Used when the + /// toggle is off. + pub fn disabled() -> Self { + Self { + blocks: Vec::new(), + capacity: 0, + inserted: 0, + } + } + + /// If the toggle is on, allocate a filter sized for at least `capacity` keys + /// If the toggle is off, return the disabled filter. + pub fn for_new_mapping(capacity: usize) -> Self { + if is_enabled() { + Self::with_capacity(capacity) + } else { + Self::disabled() + } + } + + pub fn with_capacity(capacity: usize) -> Self { + let capacity = capacity.max(MIN_CAPACITY); + // Saturating on purpose: a wrapping multiply could turn an absurd + // capacity into a tiny block count, silently producing a filter that + // is over-saturated from birth. Saturating fails loudly at allocation + // instead. + let block_count = capacity + .saturating_mul(BITS_PER_KEY) + .div_ceil(BITS_PER_BLOCK) + .max(1); + Self { + blocks: vec![Block::default(); block_count], + capacity, + inserted: 0, + } + } + + /// Whether the filter has taken more keys than it was sized for and should + /// be rebuilt against the live key set. + pub fn is_saturated(&self) -> bool { + !self.blocks.is_empty() && self.inserted > self.capacity + } + + /// Headroom to size a rebuild for, given the current live key count. + /// The doubling is what keeps rebuilds geometric, so inserts amortise to + /// O(1) even though each rebuild is O(live keys). + pub fn rebuild_capacity(live_keys: usize) -> usize { + live_keys.saturating_mul(2).max(MIN_CAPACITY) + } + + /// Record `external_id` as present. Never clears bits, so the filter stays + /// a superset of the live key set. + #[inline] + pub fn insert(&mut self, external_id: &PointIdType) { + if self.blocks.is_empty() { + return; + } + self.inserted += 1; + let hash = hash_point_id(external_id); + let block_index = fastrange(hash, self.blocks.len()); + let block = &mut self.blocks[block_index]; + for (lane, mask) in block.0.iter_mut().zip(lane_masks(hash)) { + *lane |= mask; + } + } + + /// `false` means `external_id` is definitely absent; `true` means it may be + /// present and the caller must consult the map. + #[inline] + pub fn maybe_contains(&self, external_id: &PointIdType) -> bool { + if self.blocks.is_empty() { + // Disabled filter: no information, so never screen anything out. + return true; + } + let hash = hash_point_id(external_id); + let block_index = fastrange(hash, self.blocks.len()); + let block = &self.blocks[block_index]; + // Accumulate the missing bits rather than short-circuiting: branchless + // and vectorizable, and a miss has to test every lane anyway. + let mut missing = 0u32; + for (lane, mask) in block.0.iter().zip(lane_masks(hash)) { + missing |= mask & !lane; + } + missing == 0 + } + + /// Approximate RAM usage in bytes. + pub fn ram_usage_bytes(&self) -> usize { + self.blocks.capacity() * std::mem::size_of::() + } +} + +impl Default for ExternalIdFilter { + fn default() -> Self { + Self::for_new_mapping(MIN_CAPACITY) + } +} + +impl fmt::Debug for ExternalIdFilter { + /// Compact debug output that omits the backing bitset. + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + f.debug_struct("ExternalIdFilter") + .field("blocks", &self.blocks.len()) + .field("bytes", &self.ram_usage_bytes()) + .finish() + } +} + +/// The bit each lane holds for `hash`: lane `i` takes the top 5 bits of +/// `key * SALT[i]`, giving an index in `0..32`. +/// +/// The eight products are independent, so this compiles to a handful of vector +/// instructions rather than the serial chain a double-hashing probe loop needs. +/// The low half of the hash feeds the lanes while [`fastrange`] selects the +/// block from the high half, keeping the two independent. +#[inline] +fn lane_masks(hash: u64) -> [u32; LANES] { + let key = hash as u32; + let mut masks = [0u32; LANES]; + for (mask, salt) in masks.iter_mut().zip(SALT) { + *mask = 1u32 << (key.wrapping_mul(salt) >> 27); + } + masks +} + +#[inline] +fn hash_point_id(external_id: &PointIdType) -> u64 { + match external_id { + PointIdType::NumId(num) => mix64(*num), + PointIdType::Uuid(uuid) => { + // Fold to one word, then mix once. Chaining two mixes doubles the + // dependency chain — measurably slower, with no better spread. + let bits = uuid.as_u128(); + let folded = (bits as u64) ^ ((bits >> 64) as u64).wrapping_mul(UUID_FOLD); + mix64(folded ^ UUID_DOMAIN) + } + } +} + +/// SplitMix64 finalizer: cheap, dependency-free, and diffuses the sequential +/// numeric ids that dominate real workloads. +#[inline] +fn mix64(mut z: u64) -> u64 { + z = z.wrapping_add(0x9E37_79B9_7F4A_7C15); + z = (z ^ (z >> 30)).wrapping_mul(0xBF58_476D_1CE4_E5B9); + z = (z ^ (z >> 27)).wrapping_mul(0x94D0_49BB_1331_11EB); + z ^ (z >> 31) +} + +/// Map `hash` onto `0..len` by taking the high bits of a widening multiply. +/// Avoids both the modulo bias of `%` and the power-of-two rounding a bitmask +/// would force on the allocation. +#[inline] +fn fastrange(hash: u64, len: usize) -> usize { + ((u128::from(hash) * len as u128) >> 64) as usize +} + +/// Serialises tests that flip the process-global toggle +#[cfg(test)] +static TOGGLE_LOCK: std::sync::Mutex<()> = std::sync::Mutex::new(()); + +/// Pins the toggle to a chosen value for the life of the guard, serialising +/// against every other holder and restoring the previous value on drop — +/// including on panic, so one failing test cannot leave the toggle flipped for +/// the rest of the run. +/// +/// Tests that assert a filter actually screens must hold one of these rather +/// than trusting the ambient default, which the environment can flip. +#[cfg(test)] +pub(crate) struct ToggleGuard { + previous: bool, + _lock: std::sync::MutexGuard<'static, ()>, +} + +#[cfg(test)] +impl ToggleGuard { + pub(crate) fn set(enabled: bool) -> Self { + let lock = TOGGLE_LOCK.lock().unwrap_or_else(|err| err.into_inner()); + let previous = is_enabled(); + set_enabled(enabled); + Self { + previous, + _lock: lock, + } + } +} + +#[cfg(test)] +impl Drop for ToggleGuard { + fn drop(&mut self) { + set_enabled(self.previous); + } +} + +#[cfg(test)] +mod tests { + use rand::rngs::StdRng; + use rand::{RngExt, SeedableRng}; + + use super::*; + + fn num(n: u64) -> PointIdType { + PointIdType::NumId(n) + } + + fn uuid(n: u128) -> PointIdType { + PointIdType::Uuid(uuid::Uuid::from_u128(n)) + } + + /// The invariant the whole design rests on: anything inserted must always + /// report as present. A false negative would silently lose a point. + #[test] + fn never_reports_a_false_negative() { + let mut rng = StdRng::seed_from_u64(0xF11E); + let mut filter = ExternalIdFilter::with_capacity(100_000); + + let keys: Vec<_> = (0..100_000) + .map(|i| { + if i % 3 == 0 { + uuid(rng.random()) + } else { + num(rng.random()) + } + }) + .collect(); + + for key in &keys { + filter.insert(key); + } + for key in &keys { + assert!(filter.maybe_contains(key), "false negative for {key}"); + } + } + + /// Sequential ids are the common real-world case and must not degenerate + /// into a hot block: the mixer has to spread them across the whole filter. + #[test] + fn sequential_ids_have_no_false_negatives() { + let mut filter = ExternalIdFilter::with_capacity(50_000); + for i in 0..50_000u64 { + filter.insert(&num(i)); + } + for i in 0..50_000u64 { + assert!(filter.maybe_contains(&num(i)), "false negative for {i}"); + } + } + + /// The pre-screen only pays off if misses are actually rejected. + #[test] + fn rejects_the_vast_majority_of_absent_keys() { + let mut filter = ExternalIdFilter::with_capacity(100_000); + for i in 0..100_000u64 { + filter.insert(&num(i)); + } + + let probes = 100_000u64; + let false_positives = (1_000_000..1_000_000 + probes) + .filter(|i| filter.maybe_contains(&num(*i))) + .count(); + + let rate = false_positives as f64 / probes as f64; + assert!(rate < 0.05, "false positive rate too high: {rate}"); + } + + /// A filter loaded far past its sizing degrades to more false positives, + /// never to a false negative. This is what makes an un-rebuilt filter safe + /// between growth points. + #[test] + fn oversubscription_degrades_without_false_negatives() { + let mut filter = ExternalIdFilter::with_capacity(1); + let keys: Vec<_> = (0..MIN_CAPACITY as u64 * 64).map(num).collect(); + for key in &keys { + filter.insert(key); + } + for key in &keys { + assert!(filter.maybe_contains(key), "false negative under load"); + } + } + + /// Growth is driven by insert count crossing the sizing, and a fresh + /// filter must not immediately ask to be rebuilt. + #[test] + fn saturation_tracks_capacity() { + let mut filter = ExternalIdFilter::with_capacity(MIN_CAPACITY); + assert!(!filter.is_saturated()); + for i in 0..MIN_CAPACITY as u64 { + filter.insert(&num(i)); + } + assert!(!filter.is_saturated(), "at capacity is not yet over it"); + filter.insert(&num(MIN_CAPACITY as u64)); + assert!(filter.is_saturated()); + } + + /// Toggling must be honoured by production construction, and a disabled + /// filter must cost nothing and screen nothing. + #[test] + fn toggle_controls_new_filters() { + let _guard = ToggleGuard::set(false); + let off = ExternalIdFilter::for_new_mapping(100_000); + assert_eq!( + off.ram_usage_bytes(), + 0, + "disabled filter must allocate nothing" + ); + assert!( + off.maybe_contains(&num(1)), + "disabled filter must screen nothing out" + ); + assert_eq!(ExternalIdFilter::default().ram_usage_bytes(), 0); + + set_enabled(true); + assert!(ExternalIdFilter::for_new_mapping(100_000).ram_usage_bytes() > 0); + assert!(ExternalIdFilter::default().ram_usage_bytes() > 0); + } + + /// Parsed as a pure function of the raw value: mutating the process + /// environment would race every other test that builds a filter. + #[test] + fn env_var_parsing() { + for value in ["0", "false", "FALSE", "off", "no", " off "] { + assert!( + !enabled_from_value(Some(value)), + "{value:?} should disable the filter", + ); + } + for value in ["1", "true", "yes", "on", "", "anything"] { + assert!( + enabled_from_value(Some(value)), + "{value:?} should leave the filter on", + ); + } + assert!(enabled_from_value(None), "unset should leave the filter on"); + } + + /// A disabled filter screens nothing out, so growing it would spend memory + /// for no benefit. Reachable only by constructing one explicitly — the + /// `Default` impl is enabled on purpose. + #[test] + fn disabled_filter_never_asks_to_grow() { + let mut filter = ExternalIdFilter::disabled(); + for i in 0..10_000u64 { + filter.insert(&num(i)); + } + assert!(!filter.is_saturated()); + assert_eq!(filter.ram_usage_bytes(), 0); + assert!( + filter.maybe_contains(&num(1)), + "a disabled filter must screen nothing out" + ); + } + + /// `PointMappings::default()` backs `InMemoryIdTracker`, which backs + /// `segment_builder`. If that filter were disabled it would never grow, so + /// every segment build would run unscreened. + #[test] + fn default_filter_is_enabled_and_grows() { + let _guard = ToggleGuard::set(true); + let mut filter = ExternalIdFilter::default(); + assert!( + filter.ram_usage_bytes() > 0, + "default filter must be enabled" + ); + for i in 0..=MIN_CAPACITY as u64 { + filter.insert(&num(i)); + } + assert!( + filter.is_saturated(), + "default filter must be able to saturate" + ); + } + + /// Rebuild sizing must leave headroom, or every subsequent insert would + /// trigger another O(n) rebuild. + #[test] + fn rebuild_capacity_leaves_headroom() { + assert!(ExternalIdFilter::rebuild_capacity(1_000_000) >= 2_000_000); + assert_eq!(ExternalIdFilter::rebuild_capacity(0), MIN_CAPACITY); + // Must not overflow on an absurd input. + assert!(ExternalIdFilter::rebuild_capacity(usize::MAX) > 0); + } + + /// Numeric and UUID keys share one filter; inserting one must not be + /// required to make the other findable, and neither may go missing. + #[test] + fn numeric_and_uuid_keys_coexist() { + let mut filter = ExternalIdFilter::with_capacity(1000); + for i in 0..1000u64 { + filter.insert(&num(i)); + filter.insert(&uuid(u128::from(i))); + } + for i in 0..1000u64 { + assert!(filter.maybe_contains(&num(i))); + assert!(filter.maybe_contains(&uuid(u128::from(i)))); + } + } + + /// UUID keys must be screened out about as effectively as numeric ones; + /// a weak fold would show up here as a much higher rate. + #[test] + fn rejects_absent_uuid_keys() { + let mut filter = ExternalIdFilter::with_capacity(100_000); + for i in 0..100_000u128 { + filter.insert(&uuid(i * 6364136223846793005)); + } + + let probes = 100_000u128; + let false_positives = (0..probes) + .filter(|i| filter.maybe_contains(&uuid(i * 6364136223846793005 + 1))) + .count(); + + let rate = false_positives as f64 / probes as f64; + assert!(rate < 0.05, "uuid false positive rate too high: {rate}"); + } + + /// UUIDs differing only in their high half must not collide, which is what + /// the fold multiply protects against. + #[test] + fn uuid_high_half_affects_the_hash() { + let mut filter = ExternalIdFilter::with_capacity(100_000); + filter.insert(&uuid(1)); + + let differing_high: Vec = (1..2000u128).map(|i| (i << 64) | 1).collect(); + let collisions = differing_high + .iter() + .filter(|bits| filter.maybe_contains(&uuid(**bits))) + .count(); + let rate = collisions as f64 / differing_high.len() as f64; + assert!(rate < 0.05, "uuid high half is being ignored: {rate}"); + } + + /// A block must be aligned to its own size, so a probe can never straddle + /// two cache lines. That single-cache-miss property is the whole premise. + #[test] + fn block_never_straddles_a_cache_line() { + assert_eq!(std::mem::size_of::(), 32); + assert_eq!(std::mem::align_of::(), 32); + assert_eq!(BITS_PER_BLOCK, 256); + + // Alignment is what guarantees it in practice: check real allocations. + let filter = ExternalIdFilter::with_capacity(100_000); + for block in &filter.blocks { + let start = std::ptr::from_ref(block) as usize; + assert_eq!( + start / 64, + (start + std::mem::size_of::() - 1) / 64, + "block straddles a cache line", + ); + } + } + + /// Each lane must hold exactly one bit, or the false-positive maths and the + /// branchless lane test below both break. + #[test] + fn every_lane_gets_exactly_one_bit() { + let mut rng = StdRng::seed_from_u64(7); + for _ in 0..10_000 { + let masks = lane_masks(rng.random()); + assert_eq!(masks.len(), LANES); + for mask in masks { + assert_eq!(mask.count_ones(), 1, "lane mask must be a single bit"); + } + } + } + + /// The lanes must not move in lockstep: if every salt produced the same bit + /// index, the filter would behave like a single-hash filter. + #[test] + fn lanes_are_independent() { + let mut rng = StdRng::seed_from_u64(13); + let mut all_lanes_equal = 0; + let trials = 10_000; + for _ in 0..trials { + let masks = lane_masks(rng.random()); + if masks.iter().all(|m| *m == masks[0]) { + all_lanes_equal += 1; + } + } + assert!( + all_lanes_equal < trials / 100, + "lanes are correlated: {all_lanes_equal}/{trials} had every lane on the same bit", + ); + } + + #[test] + fn fastrange_stays_in_bounds() { + let mut rng = StdRng::seed_from_u64(11); + for len in [1usize, 3, 20_480, 65_536] { + for _ in 0..1000 { + assert!(fastrange(rng.random(), len) < len); + } + } + } +} diff --git a/lib/segment/src/id_tracker/mod.rs b/lib/segment/src/id_tracker/mod.rs index 2d5d86319bf..c941529c0a7 100644 --- a/lib/segment/src/id_tracker/mod.rs +++ b/lib/segment/src/id_tracker/mod.rs @@ -1,5 +1,6 @@ pub mod compressed; pub mod disk_id_tracker; +pub mod external_id_filter; pub mod format_detection; pub mod id_tracker_base; pub mod immutable_id_tracker; diff --git a/lib/segment/src/id_tracker/mutable_id_tracker/tests.rs b/lib/segment/src/id_tracker/mutable_id_tracker/tests.rs index 2b5787a3f79..7911a2d54ab 100644 --- a/lib/segment/src/id_tracker/mutable_id_tracker/tests.rs +++ b/lib/segment/src/id_tracker/mutable_id_tracker/tests.rs @@ -828,3 +828,255 @@ fn shadow_visible_head_survives_mapping_flush_reload() { assert_eq!(id_tracker.external_id(2), Some(p7)); assert!(!id_tracker.is_deleted_point(2)); } + +/// The external-id filter is never persisted; it is rebuilt by replaying the +/// mappings log through `set_link`. Exercise that for real — write a log +/// containing drops and re-inserts, reopen, and confirm every live point still +/// resolves. A filter that lost a key here would make committed points +/// invisible after a restart. +#[test] +fn filter_rebuilds_across_a_real_reopen() { + use common::types::DeferredBehavior; + + let dir = Builder::new().prefix("id_tracker_dir").tempdir().unwrap(); + + let mut expected: Vec<(PointIdType, PointOffsetType)> = Vec::new(); + { + let mut tracker = MutableIdTracker::open(dir.path(), None).unwrap(); + // Enough points to cross the filter's growth threshold during replay. + for i in 0..20_000u64 { + let point_id = PointIdType::NumId(i); + tracker.set_link(point_id, i as PointOffsetType).unwrap(); + tracker + .set_internal_version(i as PointOffsetType, 1) + .unwrap(); + } + // Drop a slice of them. + for i in (0..20_000u64).step_by(7) { + tracker.drop(PointIdType::NumId(i)).unwrap(); + } + // Re-insert some of the dropped ids at fresh slots. + for (n, i) in (0..20_000u64).step_by(7).take(500).enumerate() { + let internal_id = (100_000 + n) as PointOffsetType; + tracker + .set_link(PointIdType::NumId(i), internal_id) + .unwrap(); + tracker.set_internal_version(internal_id, 2).unwrap(); + } + + for i in 0..20_000u64 { + let point_id = PointIdType::NumId(i); + if let Some(internal_id) = + tracker.internal_id_with_behavior(point_id, DeferredBehavior::WithDeferred) + { + expected.push((point_id, internal_id)); + } + } + + tracker.mapping_flusher()().unwrap(); + tracker.versions_flusher()().unwrap(); + } + + let reloaded = MutableIdTracker::open(dir.path(), None).unwrap(); + assert!(!expected.is_empty()); + for (point_id, internal_id) in expected { + assert_eq!( + reloaded.internal_id_with_behavior(point_id, DeferredBehavior::WithDeferred), + Some(internal_id), + "point {point_id} became unresolvable after reopen", + ); + } +} + +/// The pre-screen must be observationally invisible. With it on and off, the +/// same operation sequence has to produce a *logically identical* tracker — +/// every mapping in both directions, every version, every deletion flag, the +/// full iteration order, and the persisted bytes — not merely the same amount +/// of data. Any divergence would mean an A/B run measured two different things. +#[test] +fn filter_toggle_is_logically_identical() { + use std::collections::BTreeMap; + + use common::types::DeferredBehavior; + + use crate::id_tracker::external_id_filter::{self, ToggleGuard}; + + /// Everything observable about a tracker. + #[derive(PartialEq)] + struct Snapshot { + total_points: usize, + available_points: usize, + deleted_points: usize, + /// external -> internal for every id probed, present or not. + forward: Vec<(PointIdType, Option)>, + /// internal -> (external, version, deleted) for every slot. + reverse: Vec<( + PointOffsetType, + Option, + Option, + bool, + )>, + /// Full ordered iteration, which the optimizer's merge depends on. + iteration: Vec<(PointIdType, PointOffsetType)>, + /// Persisted file contents, keyed by file name. + files: BTreeMap>, + ram: usize, + } + + fn run(dir: &std::path::Path) -> Snapshot { + let mut tracker = MutableIdTracker::open(dir, None).unwrap(); + for i in 0..15_000u64 { + tracker + .set_link(PointIdType::NumId(i), i as PointOffsetType) + .unwrap(); + tracker + .set_internal_version(i as PointOffsetType, i + 1) + .unwrap(); + } + for i in (0..15_000u64).step_by(5) { + tracker.drop(PointIdType::NumId(i)).unwrap(); + } + for (n, i) in (0..15_000u64).step_by(5).take(400).enumerate() { + let internal_id = (50_000 + n) as PointOffsetType; + tracker + .set_link(PointIdType::NumId(i), internal_id) + .unwrap(); + tracker + .set_internal_version(internal_id, 900_000 + i) + .unwrap(); + } + for i in 0..500u64 { + let point_id = PointIdType::Uuid(uuid::Uuid::from_u128(u128::from(i))); + let internal_id = (80_000 + i) as PointOffsetType; + tracker.set_link(point_id, internal_id).unwrap(); + tracker + .set_internal_version(internal_id, 700_000 + i) + .unwrap(); + } + + let mut forward = Vec::new(); + for i in 0..16_000u64 { + let point_id = PointIdType::NumId(i); + forward.push(( + point_id, + tracker.internal_id_with_behavior(point_id, DeferredBehavior::WithDeferred), + )); + } + for i in 0..600u64 { + let point_id = PointIdType::Uuid(uuid::Uuid::from_u128(u128::from(i))); + forward.push(( + point_id, + tracker.internal_id_with_behavior(point_id, DeferredBehavior::WithDeferred), + )); + } + + let reverse = (0..tracker.total_point_count() as PointOffsetType) + .map(|internal_id| { + ( + internal_id, + tracker.external_id(internal_id), + tracker.internal_version(internal_id), + tracker.is_deleted_point(internal_id), + ) + }) + .collect(); + + let iteration = tracker.point_mappings().iter_from(None).collect(); + + tracker.mapping_flusher()().unwrap(); + tracker.versions_flusher()().unwrap(); + + let files = tracker + .files() + .into_iter() + .map(|path| { + let name = path.file_name().unwrap().to_string_lossy().into_owned(); + (name, fs_err::read(&path).unwrap()) + }) + .collect(); + + Snapshot { + total_points: tracker.total_point_count(), + available_points: tracker.available_point_count(), + deleted_points: tracker.deleted_point_count(), + forward, + reverse, + iteration, + files, + ram: tracker.ram_usage_bytes(), + } + } + + let on = { + let _guard = ToggleGuard::set(true); + assert!(external_id_filter::is_enabled()); + let dir = Builder::new().prefix("filter_on").tempdir().unwrap(); + run(dir.path()) + }; + let off = { + let _guard = ToggleGuard::set(false); + assert!(!external_id_filter::is_enabled()); + let dir = Builder::new().prefix("filter_off").tempdir().unwrap(); + run(dir.path()) + }; + + // Guards against a vacuous pass: the runs must genuinely differ internally, + // or the comparisons below prove nothing about the toggle. + assert!( + on.ram > off.ram, + "toggle had no effect: both runs used {} bytes, so the filter was never \ + actually enabled and this comparison is vacuous", + on.ram, + ); + + assert_eq!( + on.total_points, off.total_points, + "total point count differs" + ); + assert_eq!( + on.available_points, off.available_points, + "available point count differs", + ); + assert_eq!( + on.deleted_points, off.deleted_points, + "deleted point count differs" + ); + + assert_eq!( + on.forward.len(), + off.forward.len(), + "probed a different number of ids", + ); + for ((id_on, got_on), (id_off, got_off)) in on.forward.iter().zip(&off.forward) { + assert_eq!(id_on, id_off); + assert_eq!(got_on, got_off, "external->internal differs for {id_on}"); + } + + for (on_slot, off_slot) in on.reverse.iter().zip(&off.reverse) { + assert_eq!( + on_slot, off_slot, + "slot state differs (internal, external, version, deleted)", + ); + } + assert_eq!(on.reverse.len(), off.reverse.len(), "slot count differs"); + + assert_eq!( + on.iteration, off.iteration, + "ordered iteration differs, which would change optimizer merge results", + ); + + assert_eq!( + on.files.keys().collect::>(), + off.files.keys().collect::>(), + "different files persisted", + ); + for (name, on_bytes) in &on.files { + let off_bytes = &off.files[name]; + assert!( + on_bytes == off_bytes, + "persisted bytes of {name} differ ({} vs {} bytes)", + on_bytes.len(), + off_bytes.len(), + ); + } +} diff --git a/lib/segment/src/id_tracker/point_mappings.rs b/lib/segment/src/id_tracker/point_mappings.rs index fd85fb828ea..20a8338044c 100644 --- a/lib/segment/src/id_tracker/point_mappings.rs +++ b/lib/segment/src/id_tracker/point_mappings.rs @@ -18,12 +18,13 @@ use rand::rngs::StdRng; use rand::seq::SliceRandom as _; use uuid::Uuid; +use crate::id_tracker::external_id_filter::ExternalIdFilter; use crate::types::PointIdType; /// Used endianness for storing PointMapping-files. pub type FileEndianess = LittleEndian; -#[derive(Clone, PartialEq, Default, Debug)] +#[derive(Clone, Default, Debug)] pub struct PointMappings { /// `deleted` specifies which points of internal_to_external was deleted. /// It is possible that `deleted` can be longer or shorter than `internal_to_external`. @@ -66,6 +67,47 @@ pub struct PointMappings { /// Number of deleted deferred points. Maintained incrementally so we can /// derive the visible deferred count without re-scanning the deleted bitslice. deferred_deleted_count: usize, + + /// Blocked Bloom pre-screen over the external ids held in the four maps + /// above. A `false` from it means the id is definitely absent, letting + /// `internal_id_with_behavior` skip both `BTreeMap` walks; a `true` is + /// only a hint and still resolves through the maps. + /// + /// Derived state, not logical state: it is rebuilt at construction, never + /// persisted, and deliberately keeps stale bits for dropped points (see + /// [`ExternalIdFilter`]). Excluded from [`PartialEq`] for that reason. + external_filter: ExternalIdFilter, +} + +impl PartialEq for PointMappings { + /// Compares logical state only. `external_filter` is excluded: its bits + /// record every key ever inserted and are never cleared, so two identical + /// mappings can hold different bits depending on what was deleted. + fn eq(&self, other: &Self) -> bool { + // Destructured so that adding a field is a compile error here rather + // than a silently-ignored term. + let Self { + deleted, + internal_to_external, + external_to_internal_num, + external_to_internal_uuid, + external_to_internal_num_deferred, + external_to_internal_uuid_deferred, + shadowed, + deferred_internal_id, + deferred_deleted_count, + external_filter: _, + } = self; + *deleted == other.deleted + && *internal_to_external == other.internal_to_external + && *external_to_internal_num == other.external_to_internal_num + && *external_to_internal_uuid == other.external_to_internal_uuid + && *external_to_internal_num_deferred == other.external_to_internal_num_deferred + && *external_to_internal_uuid_deferred == other.external_to_internal_uuid_deferred + && *shadowed == other.shadowed + && *deferred_internal_id == other.deferred_internal_id + && *deferred_deleted_count == other.deferred_deleted_count + } } impl PointMappings { @@ -133,6 +175,12 @@ impl PointMappings { } }) .unwrap_or(0); + let external_filter = Self::build_filter( + &external_to_internal_num, + &external_to_internal_uuid, + &external_to_internal_num_deferred, + &external_to_internal_uuid_deferred, + ); Self { deleted, internal_to_external, @@ -143,9 +191,35 @@ impl PointMappings { shadowed, deferred_internal_id, deferred_deleted_count, + external_filter, } } + /// Seed the pre-screen from a live key set. + /// + /// Only used at construction. There is no incremental rebuild: `set_link` + /// keeps the filter current, and `drop` intentionally leaves stale bits + /// behind rather than maintaining a removal path. + fn build_filter( + num: &BTreeMap, + uuid: &BTreeMap, + num_deferred: &BTreeMap, + uuid_deferred: &BTreeMap, + ) -> ExternalIdFilter { + let live = num.len() + uuid.len() + num_deferred.len() + uuid_deferred.len(); + // `for_new_mapping` honours the toggle: switched off it yields a filter + // that screens nothing, and the inserts below become no-ops. + let mut filter = + ExternalIdFilter::for_new_mapping(ExternalIdFilter::rebuild_capacity(live)); + for key in num.keys().chain(num_deferred.keys()) { + filter.insert(&PointIdType::NumId(*key)); + } + for key in uuid.keys().chain(uuid_deferred.keys()) { + filter.insert(&PointIdType::Uuid(*key)); + } + filter + } + /// ToDo: this function is temporary and should be removed before PR is merged pub fn deconstruct( self, @@ -194,6 +268,16 @@ impl PointMappings { external_id: &PointIdType, deferred_behavior: common::types::DeferredBehavior, ) -> Option { + // Return none if external filter (bloom filter) confirms + // it's not present, preventing a lookup in the BTree. + if !self.external_filter.maybe_contains(external_id) { + debug_assert!( + self.internal_id_active(external_id).is_none() + && self.internal_id_deferred(external_id).is_none(), + "external id filter reported a false negative for {external_id}", + ); + return None; + } if deferred_behavior.with_deferred_points() { self.internal_id_deferred(external_id) .or_else(|| self.internal_id_active(external_id)) @@ -521,6 +605,11 @@ impl PointMappings { external_id: PointIdType, internal_id: PointOffsetType, ) -> Option { + // Update the external id filter to reflect the new mapping first, + // preserving the invariant that a `false` from the filter means + // the external id is definitely absent from the maps. + self.external_filter.insert(&external_id); + let is_deferred = self .deferred_internal_id .is_some_and(|cutoff| internal_id >= cutoff); @@ -624,6 +713,18 @@ impl PointMappings { self.internal_to_external[internal_id_usize] = external_id; self.deleted.set(internal_id_usize, false); + // Rebuilt from the maps once the filter outgrows its sizing. Deferred + // to here so the maps already reflect this write, and so a rebuild + // also sheds the stale bits left behind by `drop`. + if self.external_filter.is_saturated() { + self.external_filter = Self::build_filter( + &self.external_to_internal_num, + &self.external_to_internal_uuid, + &self.external_to_internal_num_deferred, + &self.external_to_internal_uuid_deferred, + ); + } + same_track_prior } @@ -714,6 +815,12 @@ impl PointMappings { }) .collect(); + let external_filter = Self::build_filter( + &external_to_internal_num, + &external_to_internal_uuid, + &BTreeMap::new(), + &BTreeMap::new(), + ); Self { deleted, internal_to_external, @@ -724,6 +831,7 @@ impl PointMappings { shadowed: BitVec::new(), deferred_internal_id: None, deferred_deleted_count: 0, + external_filter, } } @@ -752,6 +860,7 @@ impl PointMappings { shadowed, deferred_internal_id: _, deferred_deleted_count: _, + external_filter, } = self; let deleted_bytes = deleted.capacity().div_ceil(u8::BITS as usize); @@ -773,7 +882,12 @@ impl PointMappings { let uuid_map_bytes = (external_to_internal_uuid.len() + external_to_internal_uuid_deferred.len()) * uuid_entry_size; - deleted_bytes + shadowed_bytes + internal_to_external_bytes + num_map_bytes + uuid_map_bytes + deleted_bytes + + shadowed_bytes + + internal_to_external_bytes + + num_map_bytes + + uuid_map_bytes + + external_filter.ram_usage_bytes() } } @@ -1037,3 +1151,446 @@ mod set_link_shadow_tests { ); } } + +#[cfg(test)] +mod external_filter_tests { + use common::types::DeferredBehavior; + use rand::SeedableRng as _; + + use super::*; + + fn fresh() -> PointMappings { + PointMappings::new( + BitVec::new(), + Vec::new(), + BTreeMap::new(), + BTreeMap::new(), + None, + ) + } + + /// Dropping a point leaves its bit set, so the pre-screen answers "maybe" + /// for an id that is gone. The map lookup behind it must still say `None` + /// — this is precisely why `drop` needs no filter maintenance. + #[test] + fn dropped_point_survives_as_a_false_positive_and_still_resolves_to_none() { + let mut mappings = fresh(); + let point_id = PointIdType::NumId(42); + + mappings.set_link(point_id, 0); + assert_eq!( + mappings.internal_id_with_behavior(&point_id, DeferredBehavior::WithDeferred), + Some(0), + ); + + mappings.drop(point_id); + + assert!( + mappings.external_filter.maybe_contains(&point_id), + "drop is expected to leave a stale bit behind", + ); + assert_eq!( + mappings.internal_id_with_behavior(&point_id, DeferredBehavior::WithDeferred), + None, + ); + } + + /// Two mappings in identical logical state must compare equal even when + /// their filters differ, which happens whenever one of them reached that + /// state via a drop. Reload round-trip assertions depend on this. + #[test] + fn filter_contents_do_not_affect_equality() { + // Asserts a filter actually screens, so it must pin the toggle rather + // than trust the ambient default. + let _guard = crate::id_tracker::external_id_filter::ToggleGuard::set(true); + // Reached by replay: links 2, drops it, then links 1 — so the filter + // carries a stale bit for 2. + let mut replayed = fresh(); + replayed.set_link(PointIdType::NumId(2), 1); + replayed.drop(PointIdType::NumId(2)); + replayed.set_link(PointIdType::NumId(1), 0); + + // Reached by construction from the resulting maps — filter holds 1 only. + let mut deleted = BitVec::repeat(true, 2); + deleted.set(0, false); + let loaded = PointMappings::new( + deleted, + vec![PointIdType::NumId(1), PointIdType::NumId(u64::MAX)], + BTreeMap::from([(1, 0)]), + BTreeMap::new(), + None, + ); + + assert!( + replayed + .external_filter + .maybe_contains(&PointIdType::NumId(2)), + "precondition: the two filters must actually differ", + ); + assert!( + !loaded + .external_filter + .maybe_contains(&PointIdType::NumId(2)) + ); + assert_eq!(replayed, loaded); + } + + /// The pre-screen must not hide a live point from either deferred mode. + #[test] + fn shadowed_point_is_visible_through_the_filter_in_both_modes() { + let mut mappings = PointMappings::new( + BitVec::new(), + Vec::new(), + BTreeMap::new(), + BTreeMap::new(), + Some(5), + ); + let point_id = PointIdType::NumId(7); + + mappings.set_link(point_id, 2); + mappings.set_link(point_id, 9); + + assert_eq!( + mappings.internal_id_with_behavior(&point_id, DeferredBehavior::VisibleOnly), + Some(2), + ); + assert_eq!( + mappings.internal_id_with_behavior(&point_id, DeferredBehavior::WithDeferred), + Some(9), + ); + } + + /// Replaying a change log is how the filter is rebuilt on restart. A log + /// containing drops and re-inserts must leave every live key resolvable — + /// the drops must not leave a hole the filter screens out. + #[test] + fn replayed_log_with_drops_and_reinserts_resolves_every_live_key() { + // The exact call sequence `read_mappings` makes when replaying. + let mut replayed = fresh(); + let mut expected = Vec::new(); + for i in 0..2_000u64 { + replayed.set_link(PointIdType::NumId(i), i as PointOffsetType); + } + for i in 0..2_000u64 { + if i % 3 == 0 { + replayed.drop(PointIdType::NumId(i)); + } else { + expected.push(i); + } + } + // Re-insert a third of the dropped ids at fresh slots, as an upsert of a + // previously deleted point would. + for i in (0..2_000u64).filter(|i| i % 3 == 0).take(200) { + replayed.set_link(PointIdType::NumId(i), (10_000 + i) as PointOffsetType); + expected.push(i); + } + + for i in expected { + let point_id = PointIdType::NumId(i); + assert!( + replayed.external_filter.maybe_contains(&point_id), + "filter lost live key {i} during replay", + ); + assert!( + replayed + .internal_id_with_behavior(&point_id, DeferredBehavior::WithDeferred) + .is_some(), + "live key {i} unresolvable after replay", + ); + } + } + + /// A replay long enough to cross the growth threshold rebuilds the filter + /// mid-stream. Every key inserted before the rebuild must survive it. + #[test] + fn growth_during_replay_preserves_earlier_keys() { + let mut replayed = fresh(); + let count = 40_000u64; // comfortably past MIN_CAPACITY + for i in 0..count { + replayed.set_link(PointIdType::NumId(i), i as PointOffsetType); + } + for i in 0..count { + let point_id = PointIdType::NumId(i); + assert!( + replayed.external_filter.maybe_contains(&point_id), + "key {i} lost across a mid-replay rebuild", + ); + assert_eq!( + replayed.internal_id_with_behavior(&point_id, DeferredBehavior::WithDeferred), + Some(i as PointOffsetType), + ); + } + } + + /// Loading a segment that has a deferred cutoff partitions the single + /// persisted map into active and deferred tracks. Keys that land on the + /// deferred side must still be in the filter, or they become invisible. + #[test] + fn keys_partitioned_into_the_deferred_track_stay_in_the_filter() { + let cutoff: PointOffsetType = 50; + let count = 100u64; + // One persisted map spanning both sides of the cutoff, as on disk. + let num_map: BTreeMap = + (0..count).map(|i| (i, i as PointOffsetType)).collect(); + let mut deleted = BitVec::repeat(false, count as usize); + deleted.set(0, false); + let internal_to_external: Vec = (0..count).map(PointIdType::NumId).collect(); + + let mappings = PointMappings::new( + deleted, + internal_to_external, + num_map, + BTreeMap::new(), + Some(cutoff), + ); + + for i in 0..count { + let point_id = PointIdType::NumId(i); + assert!( + mappings.external_filter.maybe_contains(&point_id), + "key {i} lost during the deferred partition at load", + ); + // Deferred keys are hidden from VisibleOnly by design, but must be + // reachable with deferred points included. + assert_eq!( + mappings.internal_id_with_behavior(&point_id, DeferredBehavior::WithDeferred), + Some(i as PointOffsetType), + "key {i} unresolvable after the deferred partition", + ); + } + } + + /// A shadowed point holds an active and a deferred head under one external + /// id. One filter entry has to serve both tracks. + #[test] + fn shadowed_pairs_survive_a_rebuild() { + let mut mappings = PointMappings::new( + BitVec::new(), + Vec::new(), + BTreeMap::new(), + BTreeMap::new(), + Some(20_000), + ); + // Enough writes to force at least one rebuild while shadowed pairs exist. + for i in 0..10_000u64 { + mappings.set_link(PointIdType::NumId(i), i as PointOffsetType); + } + for i in 0..10_000u64 { + mappings.set_link(PointIdType::NumId(i), (20_000 + i) as PointOffsetType); + } + + for i in 0..10_000u64 { + let point_id = PointIdType::NumId(i); + assert!(mappings.external_filter.maybe_contains(&point_id)); + assert_eq!( + mappings.internal_id_with_behavior(&point_id, DeferredBehavior::VisibleOnly), + Some(i as PointOffsetType), + "active head of shadowed point {i} lost", + ); + assert_eq!( + mappings.internal_id_with_behavior(&point_id, DeferredBehavior::WithDeferred), + Some((20_000 + i) as PointOffsetType), + "deferred head of shadowed point {i} lost", + ); + } + } + + /// Every id reachable through the maps must be reachable through the + /// filter, across a randomised mapping. + #[test] + fn no_false_negatives_across_a_random_mapping() { + let mut rand = StdRng::seed_from_u64(0xBEEF); + let mappings = PointMappings::random(&mut rand, 10_000); + + for (external_id, internal_id) in mappings.iter_from(None) { + assert_eq!( + mappings.internal_id_with_behavior(&external_id, DeferredBehavior::WithDeferred), + Some(internal_id), + "filter hid live point {external_id}", + ); + } + } +} + +/// Adversarial checks: randomised differential testing against the unfiltered +/// path, plus the pathological inputs the filter's sizing arithmetic could +/// mishandle. +#[cfg(test)] +mod adversarial_tests { + use common::types::DeferredBehavior; + use rand::rngs::StdRng; + use rand::{RngExt, SeedableRng}; + + use super::*; + use crate::id_tracker::external_id_filter::{ExternalIdFilter, ToggleGuard}; + + /// A scripted, deliberately hostile operation sequence: colliding ids, a + /// mix of numeric and UUID keys, drops of live and already-dead points, + /// re-links of the same external id to new and identical slots, and slot + /// reuse. Applied identically with the filter on and off; every lookup and + /// the entire resulting state must agree. + fn hostile_sequence(mappings: &mut PointMappings, seed: u64, ops: usize) { + let mut rng = StdRng::seed_from_u64(seed); + // Small id space on purpose, so collisions and reuse actually happen. + let id_space = 512u64; + for _ in 0..ops { + let raw = rng.random_range(0..id_space); + let point_id = if raw % 3 == 0 { + PointIdType::Uuid(Uuid::from_u128(u128::from(raw))) + } else { + PointIdType::NumId(raw) + }; + match rng.random_range(0..10u32) { + 0..=5 => { + let internal = rng.random_range(0..(id_space as PointOffsetType * 2)); + mappings.set_link(point_id, internal); + } + 6..=8 => { + mappings.drop(point_id); + } + _ => { + // Re-link to a slot that is very likely already occupied. + let internal = rng.random_range(0..16) as PointOffsetType; + mappings.set_link(point_id, internal); + } + } + } + } + + fn snapshot( + m: &PointMappings, + ) -> Vec<( + PointIdType, + Option, + Option, + )> { + let mut out = Vec::new(); + for raw in 0..512u64 { + for point_id in [ + PointIdType::NumId(raw), + PointIdType::Uuid(Uuid::from_u128(u128::from(raw))), + ] { + out.push(( + point_id, + m.internal_id_with_behavior(&point_id, DeferredBehavior::VisibleOnly), + m.internal_id_with_behavior(&point_id, DeferredBehavior::WithDeferred), + )); + } + } + out + } + + fn build(cutoff: Option, seed: u64, ops: usize) -> PointMappings { + let mut m = PointMappings::new( + BitVec::new(), + Vec::new(), + BTreeMap::new(), + BTreeMap::new(), + cutoff, + ); + hostile_sequence(&mut m, seed, ops); + m + } + + /// The core adversarial property: the filter must be invisible under any + /// operation sequence, with and without a deferred cutoff. + #[test] + fn randomised_differential_filter_on_vs_off() { + for cutoff in [None, Some(64), Some(1)] { + for seed in 0..12u64 { + let with = { + let _g = ToggleGuard::set(true); + build(cutoff, seed, 4000) + }; + let without = { + let _g = ToggleGuard::set(false); + build(cutoff, seed, 4000) + }; + assert_eq!( + snapshot(&with), + snapshot(&without), + "lookups diverged (cutoff={cutoff:?}, seed={seed})", + ); + assert_eq!( + with, without, + "state diverged (cutoff={cutoff:?}, seed={seed})" + ); + assert_eq!( + with.iter_from(None).collect::>(), + without.iter_from(None).collect::>(), + "iteration diverged (cutoff={cutoff:?}, seed={seed})", + ); + } + } + } + + /// Repeatedly upserting the same few points inflates the filter's insert + /// counter without growing the live set. That must not spiral into a + /// rebuild storm. + #[test] + fn repeated_updates_do_not_storm_rebuilds() { + let _g = ToggleGuard::set(true); + let mut m = PointMappings::new( + BitVec::new(), + Vec::new(), + BTreeMap::new(), + BTreeMap::new(), + None, + ); + for i in 0..64u64 { + m.set_link(PointIdType::NumId(i), i as PointOffsetType); + } + let before = m.external_filter.ram_usage_bytes(); + // 200k updates over a live set of 64. + for round in 0..200_000u64 { + let i = round % 64; + m.set_link(PointIdType::NumId(i), i as PointOffsetType); + } + let after = m.external_filter.ram_usage_bytes(); + assert_eq!( + before, after, + "filter grew despite a constant live set — sizing tracks inserts, not keys", + ); + for i in 0..64u64 { + assert_eq!( + m.internal_id_with_behavior(&PointIdType::NumId(i), DeferredBehavior::WithDeferred), + Some(i as PointOffsetType), + ); + } + } + + /// Flipping the toggle mid-life must never break a mapping that already + /// has an enabled filter, nor one that does not. + #[test] + fn toggle_flip_midlife_is_safe() { + let mut enabled_then_off = { + let _g = ToggleGuard::set(true); + build(None, 7, 2000) + }; + { + let _g = ToggleGuard::set(false); + // Keep mutating after the toggle flipped; a rebuild here would + // swap in a disabled filter, which must still be correct. + hostile_sequence(&mut enabled_then_off, 8, 20_000); + } + let reference = { + let _g = ToggleGuard::set(false); + let mut m = build(None, 7, 2000); + hostile_sequence(&mut m, 8, 20_000); + m + }; + assert_eq!(snapshot(&enabled_then_off), snapshot(&reference)); + } + + /// Absurd sizing inputs must not overflow or attempt an insane allocation. + /// `rebuild_capacity` saturates, but `with_capacity` then multiplies by + /// BITS_PER_KEY. + #[test] + fn absurd_capacity_is_handled() { + let huge = ExternalIdFilter::rebuild_capacity(usize::MAX); + assert!(huge > 0); + // The realistic ceiling: more keys than any machine can hold. + let filter = ExternalIdFilter::with_capacity(1 << 32); + assert!(filter.ram_usage_bytes() > 0); + } +}