use yo_common::num::{parse_f64, parse_i64};
use yo_common::{Code, Error, Result};
use yo_kv::{Foreign, Keyspace};
use yo_sketch::bloom::{Added, Bloom, Load, MAX_CAPACITY, MAX_EXPANSION, MIN_CAPACITY};
use super::args::{self, Args};
use super::table::Spec;
use crate::reply::Out;
const DEFAULT_ERROR: f64 = 0.01;
const DEFAULT_CAPACITY: u64 = 100;
const DEFAULT_EXPANSION: u32 = 2;
const NOT_FOUND: &str = "not found";
const ITEM_EXISTS: &str = "item exists";
const BAD_ERROR_RATE: &str = "bad error rate";
const ERROR_RANGE: &str = "error rate must be in the range (0.000000, 1.000000)";
const BAD_CAPACITY: &str = "bad capacity";
const CAPACITY_RANGE: &str = "capacity must be in the range [1, 1073741824]";
const NO_EXPANSION: &str = "no expansion";
const BAD_EXPANSION: &str = "bad expansion";
const EXPANSION_RANGE: &str = "expansion must be in the range [0, 32768]";
const FULL: &str = "non scaling filter is full";
const BAD_DATA: &str = "received bad data";
const NO_LINK: &str = "invalid offset - no link found";
const TOO_BIG: &str = "invalid chunk - Too big for current filter";
const LOAD_NOT_NUMERIC: &str = "Second argument must be numeric";
const SCAN_NOT_NUMERIC: &[u8] = b"Second argument must be numeric";
const CANNOT_EXPAND: &[u8] = b"Nonscaling filters cannot expand";
const BAD_INFO: &[u8] = b"Invalid information value";
const INSERT_CAPACITY: &[u8] = b"Bad capacity";
const INSERT_ERROR: &[u8] = b"Bad error rate";
const INSERT_EXPANSION: &[u8] = b"Bad expansion";
const UNKNOWN_ARG: &[u8] = b"Unknown argument received";
#[derive(Debug)]
pub(super) struct BloomBody {
b: Bloom,
}
impl Foreign for BloomBody {
fn type_name(&self) -> &'static str {
"MBbloom--"
}
fn encoding(&self) -> &'static str {
"raw"
}
fn memory_bytes(&self) -> usize {
self.b.memory_bytes()
}
fn is_empty(&self) -> bool {
false
}
}
pub(super) fn execute(db: &mut Keyspace, spec: &Spec, args: Args<'_>, out: &mut Out) -> Result<()> {
match spec.name {
"bf.reserve" => reserve(db, args, out),
"bf.add" => add(db, args, out),
"bf.madd" => madd(db, args, out),
"bf.insert" => insert(db, args, out),
"bf.exists" => exists(db, args, out),
"bf.mexists" => mexists(db, args, out),
"bf.scandump" => scandump(db, args, out),
"bf.loadchunk" => loadchunk(db, args, out),
"bf.info" => info(db, args, out),
"bf.card" => card(db, args, out),
"bf.debug" => debug(db, args, out),
other => unreachable!("{other} is not a bloom filter command"),
}
}
fn reserve(db: &mut Keyspace, args: Args<'_>, out: &mut Out) -> Result<()> {
let capacity = capacity(args.get(3))?;
let error = rate(args.get(2))?;
let mut growth = DEFAULT_EXPANSION;
let mut fixed = false;
let mut asked_to_grow = false;
let mut i = 4;
while i < args.len() {
let arg = args.get(i);
if args::is(arg, b"nonscaling") {
fixed = true;
i += 1;
} else if args::is(arg, b"expansion") {
let Some(n) = args.opt(i + 1) else {
return Err(bf(NO_EXPANSION));
};
growth = expansion(n)?;
asked_to_grow = true;
i += 2;
} else {
i += 1;
}
}
if fixed && asked_to_grow {
out.error(CANNOT_EXPAND);
return Ok(());
}
let key = args.get(1);
if write(db, key)?.is_some() {
return Err(bf(ITEM_EXISTS));
}
put(db, key, Bloom::new(capacity, error, growth, fixed));
out.ok();
Ok(())
}
fn add(db: &mut Keyspace, args: Args<'_>, out: &mut Out) -> Result<()> {
let body = open(db, args.get(1))?;
match body.b.add(args.get(2)) {
Added::Yes => out.bool(true),
Added::Already => out.bool(false),
Added::Full => out.error_line(b"ERR ", FULL.as_bytes()),
}
Ok(())
}
fn madd(db: &mut Keyspace, args: Args<'_>, out: &mut Out) -> Result<()> {
let body = open(db, args.get(1))?;
each(&mut body.b, args, 2, out);
Ok(())
}
fn insert(db: &mut Keyspace, args: Args<'_>, out: &mut Out) -> Result<()> {
let mut capacity = DEFAULT_CAPACITY;
let mut error = DEFAULT_ERROR;
let mut growth = DEFAULT_EXPANSION;
let mut fixed = false;
let mut create = true;
let mut items = None;
let mut i = 2;
while i < args.len() {
match option(args.get(i)) {
Some(Opt::Items) => {
items = Some(i + 1);
break;
}
Some(Opt::NoCreate) => {
create = false;
i += 1;
}
Some(Opt::NonScaling) => {
fixed = true;
i += 1;
}
Some(Opt::Capacity) => {
let Some(n) = args.opt(i + 1).and_then(number) else {
out.error(INSERT_CAPACITY);
return Ok(());
};
capacity = n as u64;
i += 2;
}
Some(Opt::Error) => {
let Some(e) = args.opt(i + 1).and_then(fraction) else {
out.error(INSERT_ERROR);
return Ok(());
};
error = e;
i += 2;
}
Some(Opt::Expansion) => {
let Some(n) = args.opt(i + 1).and_then(factor) else {
out.error(INSERT_EXPANSION);
return Ok(());
};
growth = n;
i += 2;
}
None => {
out.error(UNKNOWN_ARG);
return Ok(());
}
}
}
let Some(first) = items.filter(|&at| at < args.len()) else {
return Err(args::wrong_arity("bf.insert"));
};
let key = args.get(1);
let body = match write(db, key)? {
Some(body) => body,
None if create => {
put(db, key, Bloom::new(capacity, error, growth, fixed));
write(db, key)?.expect("the filter was just created")
}
None => return Err(bf(NOT_FOUND)),
};
each(&mut body.b, args, first, out);
Ok(())
}
enum Opt {
Capacity,
Error,
Expansion,
NoCreate,
NonScaling,
Items,
}
fn option(arg: &[u8]) -> Option<Opt> {
let rest = arg.get(1).copied().unwrap_or(0).to_ascii_uppercase();
match arg.first().copied().unwrap_or(0).to_ascii_uppercase() {
b'C' => Some(Opt::Capacity),
b'E' if rest == b'R' => Some(Opt::Error),
b'E' => Some(Opt::Expansion),
b'I' => Some(Opt::Items),
b'N' if arg.len() >= 3 && arg[..3].eq_ignore_ascii_case(b"noc") => Some(Opt::NoCreate),
b'N' => Some(Opt::NonScaling),
_ => None,
}
}
fn each(b: &mut Bloom, args: Args<'_>, from: usize, out: &mut Out) {
let start = out.len();
let mut n = 0;
for i in from..args.len() {
n += 1;
match b.add(args.get(i)) {
Added::Yes => out.bool(true),
Added::Already => out.bool(false),
Added::Full => {
out.error_line(b"ERR ", FULL.as_bytes());
break;
}
}
}
out.close_array(start, n);
}
fn exists(db: &mut Keyspace, args: Args<'_>, out: &mut Out) -> Result<()> {
let found = peek(db, args.get(1)).is_some_and(|b| b.b.contains(args.get(2)));
out.bool(found);
Ok(())
}
fn mexists(db: &mut Keyspace, args: Args<'_>, out: &mut Out) -> Result<()> {
let body = peek(db, args.get(1));
out.array(args.len() - 2);
for i in 2..args.len() {
out.bool(body.is_some_and(|b| b.b.contains(args.get(i))));
}
Ok(())
}
fn scandump(db: &mut Keyspace, args: Args<'_>, out: &mut Out) -> Result<()> {
let Some(body) = read(db, args.get(1))? else {
return Err(bf(NOT_FOUND));
};
let Some(iter) = parse_i64(args.get(2)) else {
out.error(SCAN_NOT_NUMERIC);
return Ok(());
};
out.array(2);
if iter == 0 {
let header = body.b.header();
out.int(1);
out.bulk(&header);
} else {
let (next, data) = body.b.chunk(iter);
out.int(next);
out.bulk(data);
}
Ok(())
}
fn loadchunk(db: &mut Keyspace, args: Args<'_>, out: &mut Out) -> Result<()> {
let Some(iter) = parse_i64(args.get(2)) else {
return Err(bf(LOAD_NOT_NUMERIC));
};
let key = args.get(1);
let data = args.get(3);
let Some(body) = write(db, key)? else {
if iter != 1 {
return Err(bf(NOT_FOUND));
}
let Some(b) = Bloom::from_header(data) else {
return Err(bf(BAD_DATA));
};
put(db, key, b);
out.ok();
return Ok(());
};
match body.b.load(iter, data) {
Ok(()) => out.ok(),
Err(Load::BadData) => return Err(bf(BAD_DATA)),
Err(Load::NoLink) => return Err(bf(NO_LINK)),
Err(Load::TooBig) => return Err(bf(TOO_BIG)),
}
Ok(())
}
fn info(db: &mut Keyspace, args: Args<'_>, out: &mut Out) -> Result<()> {
if args.len() > 3 {
return Err(args::wrong_arity("bf.info"));
}
let Some(body) = read(db, args.get(1))? else {
return Err(bf(NOT_FOUND));
};
let b = &body.b;
let Some(field) = args.opt(2) else {
out.map(5);
out.simple(b"Capacity");
out.uint(b.capacity());
out.simple(b"Size");
out.uint(b.reported_size());
out.simple(b"Number of filters");
out.uint(b.filters() as u64);
out.simple(b"Number of items inserted");
out.uint(b.len());
out.simple(b"Expansion rate");
match b.expansion() {
Some(n) => out.uint(u64::from(n)),
None => out.nil(),
}
return Ok(());
};
let (name, value): (&[u8], Option<u64>) = if args::is(field, b"capacity") {
(b"Capacity", Some(b.capacity()))
} else if args::is(field, b"size") {
(b"Size", Some(b.reported_size()))
} else if args::is(field, b"filters") {
(b"Number of filters", Some(b.filters() as u64))
} else if args::is(field, b"items") {
(b"Number of items inserted", Some(b.len()))
} else if args::is(field, b"expansion") {
(b"Expansion rate", b.expansion().map(u64::from))
} else {
out.error(BAD_INFO);
return Ok(());
};
if out.proto().is_resp3() {
out.map(1);
out.simple(name);
} else {
out.array(1);
}
match value {
Some(n) => out.uint(n),
None => out.nil(),
}
Ok(())
}
fn card(db: &mut Keyspace, args: Args<'_>, out: &mut Out) -> Result<()> {
let n = read(db, args.get(1))?.map_or(0, |b| b.b.len());
out.uint(n);
Ok(())
}
fn debug(db: &mut Keyspace, args: Args<'_>, out: &mut Out) -> Result<()> {
let Some(body) = read(db, args.get(1))? else {
return Err(bf(NOT_FOUND));
};
let lines = yo_alloc::allow(|| {
let mut lines = Vec::with_capacity(body.b.filters() + 1);
lines.push(format!("size:{}", body.b.len()));
for l in body.b.links() {
lines.push(format!(
"bytes:{} bits:{} hashes:{} hashwidth:64 capacity:{} size:{} ratio:{}",
l.bytes,
l.bits,
l.hashes,
l.capacity,
l.size,
significant(l.error),
));
}
lines
});
out.array(lines.len());
for line in &lines {
out.bulk(line.as_bytes());
}
Ok(())
}
fn significant(d: f64) -> String {
let sci = format!("{d:.5e}");
let (mantissa, exponent) = sci.split_once('e').expect("a scientific form has an e");
let exponent: i32 = exponent.parse().expect("and a whole number after it");
if !(-4..6).contains(&exponent) {
let m = trim(mantissa);
let sign = if exponent < 0 { '-' } else { '+' };
format!("{m}e{sign}{:02}", exponent.abs())
} else {
let places = (5 - exponent).max(0) as usize;
trim(&format!("{d:.places$}")).to_string()
}
}
fn trim(s: &str) -> &str {
match s.contains('.') {
true => s.trim_end_matches('0').trim_end_matches('.'),
false => s,
}
}
fn rate(arg: &[u8]) -> Result<f64> {
match parse_f64(arg) {
None => Err(bf(BAD_ERROR_RATE)),
Some(e) if e <= 0.0 || e >= 1.0 => Err(bf(ERROR_RANGE)),
Some(e) => Ok(e),
}
}
fn capacity(arg: &[u8]) -> Result<u64> {
match parse_i64(arg) {
None => Err(bf(BAD_CAPACITY)),
Some(n) if !(MIN_CAPACITY..=MAX_CAPACITY).contains(&n) => Err(bf(CAPACITY_RANGE)),
Some(n) => Ok(n as u64),
}
}
fn expansion(arg: &[u8]) -> Result<u32> {
match parse_i64(arg) {
None => Err(bf(BAD_EXPANSION)),
Some(n) if !(0..=MAX_EXPANSION).contains(&n) => Err(bf(EXPANSION_RANGE)),
Some(n) => Ok(n as u32),
}
}
fn number(arg: &[u8]) -> Option<i64> {
parse_i64(arg).filter(|n| (MIN_CAPACITY..=MAX_CAPACITY).contains(n))
}
fn fraction(arg: &[u8]) -> Option<f64> {
parse_f64(arg).filter(|e| *e > 0.0 && *e < 1.0)
}
fn factor(arg: &[u8]) -> Option<u32> {
parse_i64(arg)
.filter(|n| (0..=MAX_EXPANSION).contains(n))
.map(|n| n as u32)
}
fn bf(msg: &'static str) -> Error {
Error::new(Code::Invalid, msg)
}
fn put(db: &mut Keyspace, key: &[u8], b: Bloom) {
db.put_foreign(key, Box::new(BloomBody { b }));
}
fn open<'d>(db: &'d mut Keyspace, key: &[u8]) -> Result<&'d mut BloomBody> {
if write(db, key)?.is_none() {
put(
db,
key,
Bloom::new(DEFAULT_CAPACITY, DEFAULT_ERROR, DEFAULT_EXPANSION, false),
);
}
Ok(write(db, key)?.expect("the filter is there either way"))
}
fn write<'d>(db: &'d mut Keyspace, key: &[u8]) -> Result<Option<&'d mut BloomBody>> {
match db.foreign_mut(key)? {
Some(body) => match body.downcast_mut::<BloomBody>() {
Some(body) => Ok(Some(body)),
None => Err(Error::new(Code::WrongType, WRONG_KIND)),
},
None => Ok(None),
}
}
fn read<'d>(db: &'d mut Keyspace, key: &[u8]) -> Result<Option<&'d BloomBody>> {
match db.foreign(key)? {
Some(body) => match body.downcast_ref::<BloomBody>() {
Some(body) => Ok(Some(body)),
None => Err(Error::new(Code::WrongType, WRONG_KIND)),
},
None => Ok(None),
}
}
fn peek<'d>(db: &'d mut Keyspace, key: &[u8]) -> Option<&'d BloomBody> {
db.foreign(key)
.ok()
.flatten()
.and_then(<dyn Foreign>::downcast_ref::<BloomBody>)
}
const WRONG_KIND: &str = "Operation against a key holding the wrong kind of value";