use crate::config::{BloomConstructionPolicy, FilterAdvisor};
use crate::table::Table;
use alloc::sync::Arc;
use alloc::vec::Vec;
use core::ops::Bound;
use core::sync::atomic::{AtomicBool, AtomicUsize, Ordering::Relaxed};
#[cfg(feature = "std")]
use parking_lot::Mutex;
use portable_atomic::{AtomicI64, AtomicU64};
#[cfg(not(feature = "std"))]
use spin::Mutex;
#[derive(Debug, Default)]
pub struct FilterBudget {
logged: AtomicBool,
held: AtomicI64,
published: AtomicU64,
window: Mutex<()>,
reserved: AtomicU64,
admission: Mutex<()>,
}
impl FilterBudget {
pub(crate) fn publish(&self, version: &crate::version::Version) {
let live: u64 = version
.iter_tables()
.map(|table| u64::from(table.filter_size()))
.sum();
let before = self.published.swap(live, Relaxed);
#[expect(
clippy::cast_possible_wrap,
reason = "filter byte counts far below 2^63"
)]
self.held.fetch_add(live as i64 - before as i64, Relaxed);
}
#[cfg(all(test, zstd_any))]
pub(crate) fn published(&self) -> u64 {
self.published.load(Relaxed)
}
pub(crate) fn held(&self) -> u64 {
u64::try_from(self.held.load(Relaxed)).unwrap_or(0)
}
fn slide_window(&self, live: &[Live<'_>], window: u64) -> u64 {
let _window = self.window.lock();
let observed: u64 = live
.iter()
.filter_map(|live| live.table.probe_stats())
.map(crate::table::probe_stats::ProbeStats::probes)
.sum();
if observed > window {
for stats in live.iter().filter_map(|live| live.table.probe_stats()) {
stats.decay();
}
}
observed
}
fn observe(&self, used: u64, budget: u64) {
if used > budget {
self.log_entering(used, budget);
} else if self.logged.swap(false, Relaxed) {
log::info!("Filters are back within their budget: {used} bytes against {budget}");
}
}
fn log_entering(&self, used: u64, budget: u64) {
if !self.logged.swap(true, Relaxed) {
log::warn!(
"Filters exceed their budget: {used} bytes against {budget}; \
narrow filters are written until rewrites bring them under it"
);
}
}
}
#[derive(Debug)]
pub struct FilterSizing {
widths: Vec<u8>,
fallback: BloomConstructionPolicy,
fallback_bits: u8,
price: f64,
budget: u64,
spent: AtomicU64,
replaced: u64,
replaced_released: AtomicBool,
spent_estimate: AtomicU64,
partition_keys: Option<usize>,
total_keys: AtomicU64,
admitted_keys: AtomicU64,
estimated_admitted: AtomicU64,
hashes_per_key: (AtomicU64, AtomicU64),
floor: AtomicU64,
reservation: AtomicU64,
typical_keys: AtomicU64,
table_keys: AtomicU64,
inputs: Vec<Table>,
input_densities: Vec<f64>,
input_keys: Vec<u64>,
input_distinct: Vec<u64>,
prior_density: f64,
observed: bool,
split: Option<Split>,
span: Option<Span>,
key_order: Option<KeyOrder>,
framing: Framing,
cursors: Mutex<Vec<RangeCursor>>,
owners: AtomicUsize,
ended: AtomicBool,
state: Arc<FilterBudget>,
}
#[derive(Debug)]
pub struct FilterPlan(Arc<FilterSizing>);
impl FilterPlan {
fn new(sizing: Arc<FilterSizing>) -> Self {
sizing.owners.fetch_add(1, Relaxed);
Self(sizing)
}
pub fn sizing(&self) -> Arc<FilterSizing> {
Arc::clone(&self.0)
}
}
impl Clone for FilterPlan {
fn clone(&self) -> Self {
Self::new(Arc::clone(&self.0))
}
}
impl core::ops::Deref for FilterPlan {
type Target = FilterSizing;
fn deref(&self) -> &FilterSizing {
&self.0
}
}
impl Drop for FilterPlan {
fn drop(&mut self) {
if self.0.owners.fetch_sub(1, Relaxed) == 1 {
self.0.end();
}
}
}
#[derive(Default)]
pub struct Rewrite {
pub inputs: Vec<Table>,
pub split: Option<Split>,
pub span: Option<Span>,
pub count: FilterCount,
pub comparator: Option<crate::comparator::SharedComparator>,
pub framing: Framing,
}
#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)]
pub struct FilterCount {
pub keys: u64,
pub hashes: u64,
}
pub struct Span {
pub lower: Bound<crate::UserKey>,
pub upper: Bound<crate::UserKey>,
pub comparator: crate::comparator::SharedComparator,
}
impl Span {
fn covers(&self, input: &Table) -> bool {
let max = input.metadata.key_range.max();
match &self.upper {
Bound::Unbounded => true,
Bound::Excluded(upper) => {
self.comparator.compare(max, upper) == core::cmp::Ordering::Less
}
Bound::Included(upper) => {
self.comparator.compare(max, upper) != core::cmp::Ordering::Greater
}
}
}
fn bounds(&self) -> (Bound<&[u8]>, Bound<&[u8]>) {
(
self.lower.as_ref().map(AsRef::as_ref),
self.upper.as_ref().map(AsRef::as_ref),
)
}
}
impl core::fmt::Debug for Span {
fn fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result {
f.debug_struct("Span")
.field("lower", &self.lower)
.field("upper", &self.upper)
.finish_non_exhaustive()
}
}
#[derive(Clone, Default)]
pub struct Framing {
pub encryption: Option<Arc<dyn crate::encryption::EncryptionProvider>>,
pub ecc: Option<crate::table::block::EccParams>,
}
impl Framing {
fn frame(&self, len: u64) -> u64 {
if len == 0 {
return 0;
}
crate::table::block::framed_len_bound(
len,
crate::table::block::BlockType::Filter,
crate::CompressionType::None,
self.encryption.as_deref(),
self.ecc,
)
}
}
impl core::fmt::Debug for Framing {
fn fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result {
f.debug_struct("Framing")
.field("encrypted", &self.encryption.is_some())
.field("ecc", &self.ecc)
.finish()
}
}
#[derive(Clone, Debug, Default)]
struct RangeCursor {
last: Option<Bound<crate::UserKey>>,
in_flight: Vec<Bound<crate::UserKey>>,
}
struct KeyOrder(crate::comparator::SharedComparator);
impl core::fmt::Debug for KeyOrder {
fn fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result {
f.write_str("KeyOrder")
}
}
impl KeyOrder {
fn settled_share(&self, input: &Table, bounds: (Bound<&[u8]>, Bound<&[u8]>)) -> Option<f64> {
use core::cmp::Ordering::{Greater, Less};
let order = &*self.0;
let before_lower = |key: &[u8]| match bounds.0 {
Bound::Unbounded => false,
Bound::Included(lower) => order.compare(key, lower) == Less,
Bound::Excluded(lower) => order.compare(key, lower) != Greater,
};
let past_upper = |key: &[u8]| match bounds.1 {
Bound::Unbounded => false,
Bound::Included(upper) => order.compare(key, upper) == Greater,
Bound::Excluded(upper) => order.compare(key, upper) != Less,
};
let range = &input.metadata.key_range;
let (min, max) = (range.min(), range.max());
if before_lower(max) || past_upper(min) {
Some(0.0)
} else if !before_lower(min) && !past_upper(max) {
Some(1.0)
} else {
None
}
}
}
pub struct Split {
pub boundaries: Vec<crate::UserKey>,
pub comparator: crate::comparator::SharedComparator,
}
impl core::fmt::Debug for Split {
fn fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result {
f.debug_struct("Split")
.field("boundaries", &self.boundaries)
.finish_non_exhaustive()
}
}
pub struct Live<'a> {
pub table: &'a Table,
pub fallback_bits: u8,
pub partition_keys: Option<usize>,
}
pub fn live(version: &crate::version::Version) -> Vec<Live<'_>> {
version
.iter_tables()
.filter(|table| table.filter_size() > 0)
.map(|table| {
let fallback_bits = table.metadata.filter_bits.unwrap_or_else(|| {
let bits = u64::from(table.filter_size()) * 8 / filter_keys(table).max(1);
u8::try_from(bits).unwrap_or(u8::MAX)
});
let partition_keys = table.regions.filter_tli.is_some().then(|| {
let keys = table
.metadata
.filter_partition_hashes
.unwrap_or_else(|| filter_keys(table));
usize::try_from(keys).unwrap_or(usize::MAX)
});
Live {
table,
fallback_bits,
partition_keys,
}
})
.collect()
}
pub fn plan(
advisor: &FilterAdvisor,
state: &Arc<FilterBudget>,
live: &[Live<'_>],
rewrite: Rewrite,
fallback: BloomConstructionPolicy,
partition_bytes: Option<u32>,
) -> Option<FilterPlan> {
if !fallback.is_active() {
return None;
}
let Rewrite {
inputs,
split,
span,
count,
comparator,
framing,
} = rewrite;
let ranges = split.as_ref().map_or(1, |split| split.boundaries.len() + 1);
let observed = state.slide_window(live, advisor.window_probes());
let raw: Vec<(f64, usize)> = live
.iter()
.map(|live| {
let negatives = live
.table
.probe_stats()
.map_or(0, crate::table::probe_stats::ProbeStats::negatives);
let served = usize::try_from(served_keys(live.table)).unwrap_or(usize::MAX);
(as_f64(negatives), served)
})
.collect();
let densities = shrunk_densities(&raw);
let loads: Vec<Load> = raw
.iter()
.zip(&densities)
.zip(live)
.map(|((&(_, served), &density), live)| Load {
negatives: density * as_f64(served as u64),
keys: key_count(live.table),
fallback_bits: live.fallback_bits,
partition_keys: live.partition_keys,
})
.collect();
let total_keys: usize = raw.iter().map(|(_, n)| n).sum();
let total_negatives: f64 = raw.iter().map(|(q, _)| q).sum();
let prior_density = if total_keys == 0 {
0.0
} else {
total_negatives / as_f64(total_keys as u64)
};
let input_densities: Vec<f64> = inputs
.iter()
.map(|input| {
live.iter()
.position(|live| live.table.id() == input.id())
.and_then(|index| densities.get(index).copied())
.unwrap_or(prior_density)
})
.collect();
let input_keys: Vec<u64> = inputs.iter().map(served_keys).collect();
let widths = advisor.bits_per_key().to_vec();
let fallback_bits = bits_of(fallback);
let budget = advisor.budget_bytes();
let live_bytes: u64 = live
.iter()
.map(|live| u64::from(live.table.filter_size()))
.sum();
state.observe(live_bytes, budget);
let replaced: u64 = live
.iter()
.filter(|live| {
inputs.iter().any(|input| {
input.id() == live.table.id() && span.as_ref().is_none_or(|span| span.covers(input))
})
})
.map(|live| u64::from(live.table.filter_size()))
.sum();
let input_distinct: Vec<u64> = inputs.iter().map(served_distinct_keys).collect();
let pending = if inputs.is_empty() {
count
} else {
let mut pending = FilterCount::default();
for ((input, &hashes), &distinct) in inputs.iter().zip(&input_keys).zip(&input_distinct) {
let share = span.as_ref().map_or(1.0, |span| {
crate::table::probe_stats::fraction_of(input, span.bounds()).unwrap_or(1.0)
});
#[expect(
clippy::cast_possible_truncation,
clippy::cast_sign_loss,
reason = "a share of the input's own key count"
)]
let covered = |count: u64| libm::ceil(as_f64(count) * share) as u64;
pending.keys += covered(distinct);
pending.hashes += covered(hashes);
}
pending
};
let partition_keys = partition_bytes.map(|bytes| partition_keys(fallback, bytes));
let price = price(&loads, &widths, budget, &|len| framing.frame(len));
let sizing = FilterPlan::new(Arc::new(FilterSizing {
price,
widths,
fallback,
fallback_bits,
budget,
spent: AtomicU64::new(0),
replaced,
replaced_released: AtomicBool::new(false),
spent_estimate: AtomicU64::new(0.0f64.to_bits()),
partition_keys,
total_keys: AtomicU64::new(pending.keys),
admitted_keys: AtomicU64::new(0),
estimated_admitted: AtomicU64::new(0),
hashes_per_key: if pending.keys == 0 {
(AtomicU64::new(1), AtomicU64::new(1))
} else {
(
AtomicU64::new(pending.hashes.max(pending.keys)),
AtomicU64::new(pending.keys),
)
},
floor: AtomicU64::new(0),
reservation: AtomicU64::new(0),
typical_keys: AtomicU64::new(0),
table_keys: AtomicU64::new(0),
inputs,
input_densities,
input_keys,
input_distinct,
prior_density,
observed: observed > 0,
split,
span,
key_order: comparator.map(KeyOrder),
framing,
cursors: Mutex::new(alloc::vec![RangeCursor::default(); ranges]),
owners: AtomicUsize::new(0),
ended: AtomicBool::new(false),
state: Arc::clone(state),
}));
let floor = sizing.floor_for(sizing.pending_hashes());
{
let _admission = state.admission.lock();
sizing.set_reservation(0, floor);
}
Some(sizing)
}
impl FilterSizing {
pub fn bound_policy(&self) -> BloomConstructionPolicy {
let widest = self.widths.last().copied().unwrap_or(self.fallback_bits);
if widest > self.fallback_bits {
BloomConstructionPolicy::BitsPerKey(f32::from(widest))
} else {
self.fallback
}
}
pub fn candidates(
&self,
bounds: (Bound<&[u8]>, Bound<&[u8]>),
n: usize,
) -> crate::Result<Vec<BloomConstructionPolicy>> {
let load = self.load(bounds, n)?;
self.enter(bounds.0);
let price = self.current_price(Some((n, bounds)))?;
let narrowest = self.narrowest();
if price.is_infinite() {
return Ok(alloc::vec![narrowest]);
}
let mut candidates = self.preference(load, n, price);
candidates.retain(|&policy| policy != narrowest);
candidates.push(narrowest);
Ok(candidates)
}
fn current_price(&self, priced: Option<Priced<'_>>) -> crate::Result<f64> {
if self.inputs.is_empty() {
return Ok(self.price);
}
let used = self.used();
let spent = as_f64(self.spent.load(Relaxed));
let estimated = f64::from_bits(self.spent_estimate.load(Relaxed));
let error = if estimated > 0.0 {
spent / estimated
} else {
1.0
};
let left = as_f64(self.budget.saturating_sub(used)) / error;
let remaining = self.remaining_loads(priced)?;
#[expect(
clippy::cast_possible_truncation,
clippy::cast_sign_loss,
reason = "a non-negative byte count below the budget"
)]
let left = left as u64;
Ok(price(&remaining, &self.widths, left, &|len| {
self.framing.frame(len)
}))
}
fn remaining_loads(&self, priced: Option<Priced<'_>>) -> crate::Result<Vec<Load>> {
let hashes_per_key = as_f64(self.hashes_per_key.0.load(Relaxed))
/ as_f64(self.hashes_per_key.1.load(Relaxed).max(1));
let admitted = as_f64(self.admitted_keys.load(Relaxed));
let estimated_admitted = as_f64(self.estimated_admitted.load(Relaxed));
let written = if estimated_admitted > 0.0 {
admitted / estimated_admitted
} else if let Some((n, bounds)) = priced {
let estimated = as_f64(self.estimated_keys(bounds)?);
if estimated > 0.0 {
as_f64(u64::try_from(n).unwrap_or(u64::MAX)) / hashes_per_key / estimated
} else {
1.0
}
} else {
1.0
};
let cursors = self.cursors();
let mut parts = Vec::with_capacity(self.inputs.len());
for ((input, &density), (&hashes, &distinct)) in self
.inputs
.iter()
.zip(&self.input_densities)
.zip(self.input_keys.iter().zip(&self.input_distinct))
{
let mut share = 0.0;
for (range, cursor) in cursors.iter().enumerate() {
let (low, high) = self.range(range);
let from = cursor
.as_ref()
.map_or(low, |bound| bound.as_ref().map(AsRef::as_ref));
share += self.share_of(input, (from, high))?;
}
let out = as_f64(distinct) * share * written * hashes_per_key;
if out >= 1.0 {
let negatives = as_f64(hashes) * share * written * density;
parts.push((input.metadata.key_range.min().as_ref(), out, negatives));
}
}
match &self.key_order {
Some(order) => parts.sort_by(|a, b| order.0.compare(a.0, b.0)),
None => parts.sort_by(|a, b| a.0.cmp(b.0)),
}
let table = match self.table_keys.load(Relaxed) {
0 => f64::INFINITY,
hashes => as_f64(hashes),
};
let mut loads = Vec::new();
let (mut hashes, mut negatives) = (0.0, 0.0);
let mut close = |hashes: f64, negatives: f64| {
if hashes >= 1.0 {
#[expect(
clippy::cast_possible_truncation,
clippy::cast_sign_loss,
reason = "a positive hash count below the inputs' own"
)]
loads.push(Load {
negatives,
keys: hashes as usize,
fallback_bits: self.fallback_bits,
partition_keys: self.partition_keys,
});
}
};
for (_, mut out, mut probes) in parts {
loop {
let taken = table - hashes;
if out <= taken {
break;
}
let taken_probes = probes * taken / out;
close(table, negatives + taken_probes);
(hashes, negatives) = (0.0, 0.0);
out -= taken;
probes -= taken_probes;
}
hashes += out;
negatives += probes;
}
close(hashes, negatives);
Ok(loads)
}
fn share_of(&self, input: &Table, bounds: (Bound<&[u8]>, Bound<&[u8]>)) -> crate::Result<f64> {
if let Some(share) = self
.key_order
.as_ref()
.and_then(|order| order.settled_share(input, bounds))
{
return Ok(share);
}
crate::table::probe_stats::fraction_of(input, bounds)
}
fn range_of(&self, lower: Bound<&[u8]>) -> usize {
match (&self.split, lower) {
(Some(split), Bound::Included(key) | Bound::Excluded(key)) => {
split.boundaries.partition_point(|boundary| {
split.comparator.compare(boundary, key) != core::cmp::Ordering::Greater
})
}
_ => 0,
}
}
fn lower_order(
&self,
a: &Bound<crate::UserKey>,
b: &Bound<crate::UserKey>,
) -> core::cmp::Ordering {
use core::cmp::Ordering::{Equal, Greater, Less};
let order = self.key_order.as_ref().map(|order| &*order.0);
match (a, b) {
(Bound::Unbounded, Bound::Unbounded) => Equal,
(Bound::Unbounded, _) => Less,
(_, Bound::Unbounded) => Greater,
(
Bound::Included(a_key) | Bound::Excluded(a_key),
Bound::Included(b_key) | Bound::Excluded(b_key),
) => {
let keys =
order.map_or_else(|| a_key.cmp(b_key), |order| order.compare(a_key, b_key));
keys.then(match (a, b) {
(Bound::Included(_), Bound::Excluded(_)) => Less,
(Bound::Excluded(_), Bound::Included(_)) => Greater,
_ => Equal,
})
}
}
}
fn enter(&self, lower: Bound<&[u8]>) {
let range = self.range_of(lower);
let lower = lower.map(crate::UserKey::from);
let mut cursors = self.cursors.lock();
if let Some(cursor) = cursors.get_mut(range) {
let later = cursor
.last
.as_ref()
.is_none_or(|last| self.lower_order(&lower, last) == core::cmp::Ordering::Greater);
if later {
cursor.last = Some(lower.clone());
}
cursor.in_flight.push(lower);
}
}
fn leave(&self, lower: Bound<&[u8]>) {
let range = self.range_of(lower);
let lower = lower.map(crate::UserKey::from);
let mut cursors = self.cursors.lock();
if let Some(cursor) = cursors.get_mut(range)
&& let Some(index) = cursor.in_flight.iter().position(|bound| *bound == lower)
{
cursor.in_flight.swap_remove(index);
}
}
fn cursors(&self) -> Vec<Option<Bound<crate::UserKey>>> {
let cursors = self.cursors.lock();
cursors
.iter()
.map(|cursor| {
cursor
.in_flight
.iter()
.min_by(|a, b| self.lower_order(a, b))
.or(cursor.last.as_ref())
.cloned()
})
.collect()
}
fn range(&self, index: usize) -> (Bound<&[u8]>, Bound<&[u8]>) {
let Some(split) = &self.split else {
return self
.span
.as_ref()
.map_or((Bound::Unbounded, Bound::Unbounded), Span::bounds);
};
let low = index
.checked_sub(1)
.and_then(|below| split.boundaries.get(below))
.map_or(Bound::Unbounded, |key| Bound::Included(key.as_ref()));
let high = split
.boundaries
.get(index)
.map_or(Bound::Unbounded, |key| Bound::Excluded(key.as_ref()));
(low, high)
}
pub fn narrowest(&self) -> BloomConstructionPolicy {
match self.widths.first() {
Some(&bits) if self.observed || bits < self.fallback_bits => {
BloomConstructionPolicy::BitsPerKey(f32::from(bits))
}
_ => self.fallback,
}
}
#[expect(
clippy::too_many_arguments,
reason = "the filter's place, counts and bytes, as the build produced them"
)]
pub fn admit(
&self,
bounds: (Bound<&[u8]>, Bound<&[u8]>),
n: usize,
keys: usize,
bytes: u64,
estimated: u64,
frame: &dyn Fn(u64) -> u64,
last: bool,
) -> bool {
let n = u64::try_from(n).unwrap_or(u64::MAX);
let keys = u64::try_from(keys).unwrap_or(u64::MAX);
let estimated_keys = self.estimated_keys(bounds).unwrap_or(keys);
let admission = self.state.admission.lock();
if self.ended.load(Relaxed) {
return true;
}
let per_filter = self.typical_keys.load(Relaxed).max(n);
let ratio = self.ratio_with(n, keys);
let admitted = self.admitted_keys.load(Relaxed) + keys;
let estimated_admitted = self.estimated_admitted.load(Relaxed) + estimated_keys;
let later = self.pending_at(admitted, estimated_admitted, ratio);
let floor = self.floor(later, per_filter, frame);
let used = self.used();
let mine = self.reservation.load(Relaxed);
let others = self.state.reserved.load(Relaxed) - mine;
let fits = bytes + floor + others <= self.budget.saturating_sub(used);
if !fits && !last {
return false;
}
#[expect(
clippy::cast_possible_wrap,
reason = "filter byte counts far below 2^63"
)]
self.state.held.fetch_add(bytes as i64, Relaxed);
self.spent.fetch_add(bytes, Relaxed);
self.set_reservation(mine, floor);
self.admitted_keys.store(admitted, Relaxed);
self.estimated_admitted.store(estimated_admitted, Relaxed);
self.hashes_per_key.0.store(ratio.0, Relaxed);
self.hashes_per_key.1.store(ratio.1, Relaxed);
self.typical_keys.fetch_max(n, Relaxed);
drop(admission);
self.leave(bounds.0);
if used + bytes > self.budget {
self.state.log_entering(used + bytes, self.budget);
}
let mut spent = self.spent_estimate.load(Relaxed);
while let Err(actual) = self.spent_estimate.compare_exchange_weak(
spent,
(f64::from_bits(spent) + as_f64(estimated)).to_bits(),
Relaxed,
Relaxed,
) {
spent = actual;
}
true
}
fn floor(&self, later: u64, per_filter: u64, frame: &dyn Fn(u64) -> u64) -> u64 {
if later == 0 || per_filter == 0 {
return 0;
}
let mean = |keys: u64| {
#[expect(
clippy::cast_possible_truncation,
clippy::cast_sign_loss,
reason = "a non-negative byte count of one filter"
)]
let bytes = libm::round(
self.narrowest()
.expected_filter_size(usize::try_from(keys).unwrap_or(usize::MAX)),
) as u64;
frame(bytes)
};
let narrowest_bytes = mean(per_filter);
let spread = frame(
self.narrowest()
.filter_size_bound(usize::try_from(per_filter).unwrap_or(usize::MAX))
as u64,
)
.saturating_sub(narrowest_bytes);
let tail_overhead = if self.partition_keys.is_some() {
mean(1)
} else {
0
};
#[expect(
clippy::cast_possible_truncation,
reason = "later * bytes / per_filter <= later * bytes, far below 2^64"
)]
let per_key = (u128::from(later) * u128::from(narrowest_bytes))
.div_ceil(u128::from(per_filter)) as u64;
let table_keys = self.table_keys.load(Relaxed);
let tables = if table_keys == 0 {
1
} else {
later.div_ceil(table_keys)
};
#[expect(
clippy::cast_possible_truncation,
clippy::cast_sign_loss,
reason = "the square root of a filter count, non-negative and small"
)]
let strays = libm::ceil(libm::sqrt(as_f64(later.div_ceil(per_filter)))) as u64;
let final_filter = frame(
self.narrowest()
.filter_size_bound(usize::try_from(later).unwrap_or(usize::MAX)) as u64,
);
(per_key + tables * tail_overhead + strays * spread).max(final_filter)
}
fn floor_for(&self, pending: u64) -> u64 {
let per_filter = match self.typical_keys.load(Relaxed) {
0 => pending,
keys => keys,
};
self.floor(pending, per_filter, &|len| self.framing.frame(len))
}
fn set_reservation(&self, mine: u64, floor: u64) {
let published = self.published_floor(floor);
let others = self.state.reserved.load(Relaxed) - mine;
self.state.reserved.store(others + published, Relaxed);
self.reservation.store(published, Relaxed);
self.floor.store(floor, Relaxed);
}
fn published_floor(&self, floor: u64) -> u64 {
if self.replaced_released.load(Relaxed) {
floor
} else {
floor.saturating_sub(self.replaced)
}
}
fn used(&self) -> u64 {
let held = self.state.held();
if self.replaced_released.load(Relaxed) {
held
} else {
held.saturating_sub(self.replaced)
}
}
pub fn release_replaced(&self) {
let _admission = self.state.admission.lock();
self.release_locked();
}
fn release_locked(&self) {
if !self.replaced_released.swap(true, Relaxed) {
self.set_reservation(self.reservation.load(Relaxed), self.floor.load(Relaxed));
}
}
pub fn table_finished(&self, keys: usize) {
self.table_keys
.fetch_max(u64::try_from(keys).unwrap_or(u64::MAX), Relaxed);
}
pub fn expect_keys(&self, keys: u64) {
let _admission = self.state.admission.lock();
self.total_keys.store(keys, Relaxed);
let pending = self.pending_hashes();
self.set_reservation(self.reservation.load(Relaxed), self.floor_for(pending));
}
fn pending_hashes(&self) -> u64 {
self.pending_at(
self.admitted_keys.load(Relaxed),
self.estimated_admitted.load(Relaxed),
(
self.hashes_per_key.0.load(Relaxed),
self.hashes_per_key.1.load(Relaxed),
),
)
}
fn pending_at(&self, admitted: u64, estimated_admitted: u64, ratio: (u64, u64)) -> u64 {
let total = self.total_keys.load(Relaxed);
let keys = if self.inputs.is_empty() || estimated_admitted == 0 {
total.saturating_sub(admitted)
} else {
let left = total.saturating_sub(estimated_admitted);
u64::try_from(
(u128::from(left) * u128::from(admitted)).div_ceil(u128::from(estimated_admitted)),
)
.unwrap_or(u64::MAX)
};
in_hashes(keys, ratio)
}
fn estimated_keys(&self, bounds: (Bound<&[u8]>, Bound<&[u8]>)) -> crate::Result<u64> {
let mut keys = 0.0;
for (input, &distinct) in self.inputs.iter().zip(&self.input_distinct) {
keys += as_f64(distinct) * self.share_of(input, bounds)?;
}
#[expect(
clippy::cast_possible_truncation,
clippy::cast_sign_loss,
reason = "a non-negative sum of shares of the inputs' key counts"
)]
let keys = libm::ceil(keys) as u64;
Ok(keys)
}
fn ratio_with(&self, n: u64, keys: u64) -> (u64, u64) {
let held = (
self.hashes_per_key.0.load(Relaxed),
self.hashes_per_key.1.load(Relaxed),
);
if keys == 0 {
return held;
}
let first = self.admitted_keys.load(Relaxed) == 0;
if first || u128::from(n) * u128::from(held.1) > u128::from(held.0) * u128::from(keys) {
(n, keys)
} else {
held
}
}
fn load(&self, bounds: (Bound<&[u8]>, Bound<&[u8]>), n: usize) -> crate::Result<f64> {
let (mut keys, mut probes) = (0.0, 0.0);
for ((input, density), &input_keys) in self
.inputs
.iter()
.zip(&self.input_densities)
.zip(&self.input_keys)
{
let covered = as_f64(input_keys) * self.share_of(input, bounds)?;
keys += covered;
probes += covered * density;
}
let density = if keys > 0.0 {
probes / keys
} else {
self.prior_density
};
Ok(density * as_f64(n as u64))
}
fn preference(&self, load: f64, n: usize, price: f64) -> Vec<BloomConstructionPolicy> {
if !self.observed {
let mut preference = alloc::vec![self.fallback];
preference.extend(
self.widths
.iter()
.rev()
.filter(|&&bits| bits < self.fallback_bits)
.map(|&bits| BloomConstructionPolicy::BitsPerKey(f32::from(bits))),
);
return preference;
}
let bytes = sizes(n, &self.widths);
let rates = rates(&self.widths);
let mut order: Vec<usize> = (0..self.widths.len()).collect();
order.sort_by(|&a, &b| {
rank(
load,
&bytes,
&rates,
&self.widths,
(a, b),
price,
self.fallback_bits,
)
});
order
.into_iter()
.filter_map(|index| self.widths.get(index))
.map(|&bits| BloomConstructionPolicy::BitsPerKey(f32::from(bits)))
.collect()
}
}
impl FilterSizing {
fn end(&self) {
let _admission = self.state.admission.lock();
if self.ended.swap(true, Relaxed) {
return;
}
self.release_locked();
self.set_reservation(self.reservation.load(Relaxed), 0);
#[expect(
clippy::cast_possible_wrap,
reason = "filter byte counts far below 2^63"
)]
self.state
.held
.fetch_sub(self.spent.load(Relaxed) as i64, Relaxed);
}
}
type Priced<'a> = (usize, (Bound<&'a [u8]>, Bound<&'a [u8]>));
fn in_hashes(keys: u64, ratio: (u64, u64)) -> u64 {
let (hashes, per) = ratio;
u64::try_from((u128::from(keys) * u128::from(hashes)).div_ceil(u128::from(per.max(1))))
.unwrap_or(u64::MAX)
}
#[derive(Clone, Copy, Debug)]
struct Load {
negatives: f64,
keys: usize,
fallback_bits: u8,
partition_keys: Option<usize>,
}
fn price(loads: &[Load], widths: &[u8], budget: u64, frame: &dyn Fn(u64) -> u64) -> f64 {
struct Entry {
filter_load: f64,
fallback_bits: u8,
partition: Option<usize>,
table_bytes: Vec<u64>,
}
fn filter_bytes<'a>(entry: &'a Entry, partitions: &'a [(usize, Vec<u64>)]) -> &'a [u64] {
entry
.partition
.and_then(|index| partitions.get(index))
.map_or(&entry.table_bytes[..], |(_, full)| &full[..])
}
let rates = rates(widths);
let sizes =
|n: usize, widths: &[u8]| -> Vec<u64> { sizes(n, widths).into_iter().map(frame).collect() };
let mut partitions: Vec<(usize, Vec<u64>)> = Vec::new();
let mut entries: Vec<Entry> = Vec::with_capacity(loads.len());
for load in loads {
let n = load.keys;
let load_of = |keys: usize| load.negatives * as_f64(keys as u64) / as_f64(n.max(1) as u64);
let partition = match load.partition_keys {
Some(keys) if keys > 0 => keys,
_ => usize::MAX,
};
if n <= partition {
entries.push(Entry {
filter_load: load_of(n.max(1)),
fallback_bits: load.fallback_bits,
partition: None,
table_bytes: sizes(n, widths),
});
continue;
}
let index = if let Some(index) = partitions.iter().position(|(keys, _)| *keys == partition)
{
index
} else {
partitions.push((partition, sizes(partition, widths)));
partitions.len() - 1
};
let count = (n / partition) as u64;
let full = partitions.get(index).map_or(&[][..], |(_, full)| full);
entries.push(Entry {
filter_load: load_of(partition),
fallback_bits: load.fallback_bits,
partition: Some(index),
table_bytes: full.iter().map(|&full| count * full).collect(),
});
let rest = n % partition;
if rest > 0 {
entries.push(Entry {
filter_load: load_of(rest),
fallback_bits: load.fallback_bits,
partition: None,
table_bytes: sizes(rest, widths),
});
}
}
let filled = |price: f64| -> u64 {
entries
.iter()
.map(|entry| {
let index = choose(
entry.filter_load,
filter_bytes(entry, &partitions),
&rates,
widths,
entry.fallback_bits,
price,
);
entry.table_bytes.get(index).copied().unwrap_or(0)
})
.sum()
};
if filled(0.0) <= budget {
return 0.0;
}
let mut total: u64 = 0;
let mut steps: Vec<(f64, i128)> = Vec::new();
let mut hull: Vec<usize> = Vec::with_capacity(widths.len());
for entry in &entries {
let filter_bytes = filter_bytes(entry, &partitions);
let point = |index: usize| {
(
filter_bytes
.get(index)
.copied()
.map_or(f64::INFINITY, as_f64),
entry.filter_load * rates.get(index).copied().unwrap_or(0.0),
)
};
let mut order: Vec<usize> = (0..widths.len()).collect();
order.sort_by(|&a, &b| {
let ((bytes_a, fp_a), (bytes_b, fp_b)) = (point(a), point(b));
bytes_a.total_cmp(&bytes_b).then(fp_a.total_cmp(&fp_b))
});
hull.clear();
for index in order {
let (bytes, fp) = point(index);
if let Some(&last) = hull.last() {
if fp >= point(last).1 || bytes <= point(last).0 {
continue;
}
}
while let [.., before, last] = hull[..] {
let ((bytes_a, fp_a), (bytes_b, fp_b)) = (point(before), point(last));
if (fp_b - fp) * (bytes_b - bytes_a) >= (fp_a - fp_b) * (bytes - bytes_b) {
hull.pop();
} else {
break;
}
}
hull.push(index);
}
let table = |index: usize| entry.table_bytes.get(index).copied().unwrap_or(0);
let Some(&first) = hull.first() else {
continue;
};
total += table(first);
let table = |index: usize| i128::from(table(index));
for pair in hull.windows(2) {
if let [a, b] = *pair {
let ((bytes_a, fp_a), (bytes_b, fp_b)) = (point(a), point(b));
steps.push(((fp_a - fp_b) / (bytes_b - bytes_a), table(b) - table(a)));
}
}
}
if total > budget {
return f64::INFINITY;
}
steps.sort_by(|a, b| b.0.total_cmp(&a.0));
let mut filled = i128::from(total);
let mut above = f64::INFINITY;
let mut index = 0;
while let Some(&(price, _)) = steps.get(index) {
let group = steps
.get(index..)
.unwrap_or_default()
.iter()
.take_while(|step| step.0.total_cmp(&price).is_eq());
let (count, grown) = group.fold((0, 0i128), |(count, sum), step| (count + 1, sum + step.1));
if filled + grown > i128::from(budget) {
let high = if above.is_finite() {
libm::log2(above)
} else {
libm::log2(price) + 1.0
};
return exp2(f64::midpoint(libm::log2(price), high));
}
filled += grown;
above = price;
index += count;
}
if above.is_finite() {
exp2(libm::log2(above) - 1.0)
} else {
1.0
}
}
#[cfg(test)]
fn cheapest(load: f64, n: usize, widths: &[u8], fallback_bits: u8, price: f64) -> u8 {
let bytes = sizes(n, widths);
widths
.get(choose(
load,
&bytes,
&rates(widths),
widths,
fallback_bits,
price,
))
.copied()
.unwrap_or(fallback_bits)
}
fn sizes(n: usize, widths: &[u8]) -> Vec<u64> {
let shape = crate::table::filter::ExpectedSize::of(n);
widths
.iter()
.map(|&bits| {
#[expect(
clippy::cast_possible_truncation,
clippy::cast_sign_loss,
reason = "a non-negative byte count of one filter"
)]
let bytes = libm::round(shape.at(bits)) as u64;
bytes
})
.collect()
}
fn rates(widths: &[u8]) -> Vec<f64> {
widths.iter().map(|&bits| exp2(-f64::from(bits))).collect()
}
fn choose(
load: f64,
bytes: &[u64],
rates: &[f64],
widths: &[u8],
fallback_bits: u8,
price: f64,
) -> usize {
let count = widths.len().min(bytes.len()).min(rates.len());
(0..count)
.min_by(|&a, &b| rank(load, bytes, rates, widths, (a, b), price, fallback_bits))
.unwrap_or(0)
}
#[expect(
clippy::indexing_slicing,
reason = "the indices address all three slices, which the callers size alike"
)]
fn rank(
load: f64,
bytes: &[u64],
rates: &[f64],
widths: &[u8],
(a, b): (usize, usize),
price: f64,
fallback_bits: u8,
) -> core::cmp::Ordering {
let by_cost = if price.is_infinite() {
bytes[a].cmp(&bytes[b])
} else {
cost(load, rates[a], bytes[a], price).total_cmp(&cost(load, rates[b], bytes[b], price))
};
by_cost.then(
widths[a]
.abs_diff(fallback_bits)
.cmp(&widths[b].abs_diff(fallback_bits)),
)
}
fn cost(load: f64, rate: f64, bytes: u64, price: f64) -> f64 {
#[expect(
clippy::suboptimal_flops,
reason = "the cost only orders widths, which a fused multiply-add's extra \
precision does not change, and a fused one is a software routine \
(`libm::fma` without std, `mul_add` on targets built without the FMA \
feature) on the price search's inner loop"
)]
let cost = load * rate + price * as_f64(bytes);
cost
}
fn shrunk_densities(counts: &[(f64, usize)]) -> Vec<f64> {
let keys: f64 = counts.iter().map(|&(_, n)| as_f64(n as u64)).sum();
if keys <= 0.0 {
return counts.iter().map(|_| 0.0).collect();
}
let mean = counts.iter().map(|&(q, _)| q).sum::<f64>() / keys;
let density = |&(q, n): &(f64, usize)| if n == 0 { mean } else { q / as_f64(n as u64) };
let observed = counts
.iter()
.map(|entry| {
let deviation = density(entry) - mean;
as_f64(entry.1 as u64) * deviation * deviation
})
.sum::<f64>()
/ keys;
let noise = as_f64(counts.len() as u64) * mean / keys;
let keep = if observed > noise {
1.0 - noise / observed
} else {
0.0
};
counts
.iter()
.map(|entry| mean + keep * (density(entry) - mean))
.collect()
}
fn partition_keys(policy: BloomConstructionPolicy, partition_bytes: u32) -> usize {
let target = partition_bytes as usize;
let (mut low, mut high) = (1usize, 1usize);
while policy.estimated_filter_size(high) < target && high < 1 << 30 {
low = high;
high *= 2;
}
while low + 1 < high {
let mid = low + (high - low) / 2;
if policy.estimated_filter_size(mid) < target {
low = mid;
} else {
high = mid;
}
}
high
}
#[cfg(test)]
fn estimate(n: usize, bits: u8) -> u64 {
sizes(n, &[bits]).first().copied().unwrap_or(0)
}
pub fn bits_of(policy: BloomConstructionPolicy) -> u8 {
match policy {
#[expect(
clippy::cast_possible_truncation,
clippy::cast_sign_loss,
reason = "an active policy's bits per key lie in 1..=64"
)]
BloomConstructionPolicy::BitsPerKey(bits) => crate::f32_ceil(bits) as u8,
BloomConstructionPolicy::FalsePositiveRate(rate) => (1..=64u8)
.find(|&bits| exp2(-f64::from(bits)) <= f64::from(rate))
.unwrap_or(64),
}
}
fn filter_keys(table: &Table) -> u64 {
table
.metadata
.filter_hashes
.or(table.metadata.key_count)
.unwrap_or(table.metadata.item_count)
}
fn served_keys(table: &Table) -> u64 {
served(table, filter_keys(table))
}
fn served_distinct_keys(table: &Table) -> u64 {
served(
table,
table
.metadata
.key_count
.unwrap_or(table.metadata.item_count),
)
}
fn served(table: &Table, keys: u64) -> u64 {
if table.restrict_lower_bound().is_none() {
return keys;
}
let Ok(Some(span)) = table.data_span(
(Bound::Unbounded, Bound::Unbounded),
crate::SeqNo::MAX,
crate::table::SpanEdge::ByLastKey,
) else {
return keys;
};
let live = span.data_end - span.live_start;
#[expect(
clippy::cast_possible_truncation,
reason = "live <= data_end, so the quotient never exceeds the u64 count"
)]
let served = (u128::from(keys) * u128::from(live) / u128::from(span.data_end)) as u64;
served
}
fn key_count(table: &Table) -> usize {
usize::try_from(filter_keys(table)).unwrap_or(usize::MAX)
}
fn exp2(x: f64) -> f64 {
libm::exp2(x)
}
#[expect(
clippy::cast_precision_loss,
reason = "counts past 2^53 lose only digits the cost comparison does not need"
)]
fn as_f64(count: u64) -> f64 {
count as f64
}
#[cfg(test)]
mod tests;