use std::collections::HashMap;
use arrow_array::cast::AsArray;
use arrow_array::types::{
Float64Type, Int32Type, Int64Type, TimestampNanosecondType, UInt8Type, UInt16Type, UInt32Type,
UInt64Type,
};
use arrow_array::{Array, RecordBatch};
use arrow_schema::DataType;
use crate::query::Op;
use crate::schema::{ATTRS, AttrType};
pub const ZONE_IDX: &str = "zone.idx";
const MAGIC: [u8; 4] = *b"MZON";
const VERSION: u8 = 1;
const HEADER: usize = 16;
const ENTRY: usize = 40;
const MAX_KEYS: usize = 4096;
#[derive(Clone, Copy, Debug, PartialEq)]
pub struct Range {
pub int_min: i64,
pub int_max: i64,
pub dbl_min: f64,
pub dbl_max: f64,
}
impl Range {
const EMPTY: Range = Range {
int_min: i64::MAX,
int_max: i64::MIN,
dbl_min: f64::INFINITY,
dbl_max: f64::NEG_INFINITY,
};
pub const ANY: Range = Range {
int_min: i64::MIN,
int_max: i64::MAX,
dbl_min: f64::NEG_INFINITY,
dbl_max: f64::INFINITY,
};
fn int(&mut self, v: i64) {
self.int_min = self.int_min.min(v);
self.int_max = self.int_max.max(v);
}
fn float(&mut self, v: f64) {
if !v.is_nan() {
self.dbl_min = self.dbl_min.min(v);
self.dbl_max = self.dbl_max.max(v);
}
}
}
pub struct Probe {
pub key: u64,
pub op: Op,
pub int: Option<i64>,
pub float: Option<f64>,
}
impl Probe {
pub fn maybe(&self, m: &Map) -> bool {
match m.get(self.key) {
None => false,
Some(r) => {
self.int
.is_some_and(|t| reachable(self.op, r.int_min, r.int_max, t))
|| self
.float
.is_some_and(|t| reachable(self.op, r.dbl_min, r.dbl_max, t))
}
}
}
}
fn reachable<T: PartialOrd + Copy>(op: Op, min: T, max: T, target: T) -> bool {
match op {
Op::Eq => min <= target && target <= max,
Op::Lt => min < target,
Op::Lte => min <= target,
Op::Gt => max > target,
Op::Gte => max >= target,
Op::Ne | Op::Contains => true,
}
}
#[derive(Default)]
pub struct Builder {
keys: HashMap<u64, Range>,
full: bool,
}
impl Builder {
fn at(&mut self, key: u64) -> Option<&mut Range> {
if !self.keys.contains_key(&key) && self.keys.len() >= MAX_KEYS {
self.full = true;
return None;
}
Some(self.keys.entry(key).or_insert(Range::EMPTY))
}
pub fn int(&mut self, key: u64, v: i64) {
if let Some(r) = self.at(key) {
r.int(v);
}
}
pub fn float(&mut self, key: u64, v: f64) {
if let Some(r) = self.at(key) {
r.float(v);
}
}
pub fn any(&mut self, key: u64) {
if let Some(r) = self.at(key) {
*r = Range::ANY;
}
}
pub fn build(&self) -> Option<Vec<u8>> {
if self.full || self.keys.is_empty() {
return None;
}
let mut entries: Vec<(&u64, &Range)> = self.keys.iter().collect();
entries.sort_unstable_by_key(|(k, _)| **k);
let mut body = Vec::with_capacity(entries.len() * ENTRY);
for (k, r) in entries {
body.extend_from_slice(&k.to_le_bytes());
body.extend_from_slice(&r.int_min.to_le_bytes());
body.extend_from_slice(&r.int_max.to_le_bytes());
body.extend_from_slice(&r.dbl_min.to_le_bytes());
body.extend_from_slice(&r.dbl_max.to_le_bytes());
}
let mut out = Vec::with_capacity(HEADER + body.len());
out.extend_from_slice(&MAGIC);
out.push(VERSION);
out.extend_from_slice(&[0, 0, 0]);
out.extend_from_slice(&(self.keys.len() as u32).to_le_bytes());
out.extend_from_slice(&crc32fast::hash(&body).to_le_bytes());
out.extend_from_slice(&body);
Some(out)
}
}
pub struct Map<'a> {
body: &'a [u8],
n: usize,
}
impl<'a> Map<'a> {
pub fn open(file: &'a [u8]) -> Option<Map<'a>> {
if file.len() < HEADER || file[..4] != MAGIC || file[4] != VERSION {
return None;
}
let n = u32::from_le_bytes(file[8..12].try_into().expect("4 bytes")) as usize;
let crc = u32::from_le_bytes(file[12..16].try_into().expect("4 bytes"));
let body = &file[HEADER..];
if n == 0 || body.len() != n * ENTRY || crc32fast::hash(body) != crc {
return None;
}
Some(Map { body, n })
}
fn key_at(&self, i: usize) -> u64 {
u64::from_le_bytes(self.body[i * ENTRY..][..8].try_into().expect("8 bytes"))
}
fn get(&self, key: u64) -> Option<Range> {
let (mut lo, mut hi) = (0usize, self.n);
while lo < hi {
let mid = lo + (hi - lo) / 2;
match self.key_at(mid).cmp(&key) {
std::cmp::Ordering::Less => lo = mid + 1,
std::cmp::Ordering::Greater => hi = mid,
std::cmp::Ordering::Equal => {
lo = mid;
break;
}
}
}
if lo >= self.n || self.key_at(lo) != key {
return None;
}
let i = lo;
let f = |off: usize| {
self.body[i * ENTRY + off..][..8]
.try_into()
.expect("8 bytes")
};
Some(Range {
int_min: i64::from_le_bytes(f(8)),
int_max: i64::from_le_bytes(f(16)),
dbl_min: f64::from_le_bytes(f(24)),
dbl_max: f64::from_le_bytes(f(32)),
})
}
}
pub fn attr_key(key: &str) -> u64 {
mix(crate::identity::hash64(key.as_bytes()) ^ 0xa77b_a77b_a77b_a77b)
}
pub fn field_key(name: &str) -> u64 {
mix(crate::identity::hash64(name.as_bytes()) ^ 0xf1e1_f1e1_f1e1_f1e1)
}
fn mix(mut x: u64) -> u64 {
x ^= x >> 30;
x = x.wrapping_mul(0xbf58_476d_1ce4_e5b9);
x ^= x >> 27;
x = x.wrapping_mul(0x94d0_49bb_1331_11eb);
x ^ (x >> 31)
}
pub fn index(tables: &[(&'static str, RecordBatch)]) -> Option<Vec<u8>> {
let mut b = Builder::default();
if let Some((_, root)) = tables.first() {
index_root(&mut b, root);
}
for (_, t) in tables {
if std::sync::Arc::ptr_eq(&t.schema(), &ATTRS) {
index_attrs(&mut b, t);
}
}
b.build()
}
fn index_root(b: &mut Builder, root: &RecordBatch) {
macro_rules! ints {
($t:ty, $col:expr, $key:expr) => {{
let a = $col.as_primitive::<$t>();
for i in 0..a.len() {
if !a.is_null(i) {
b.int($key, a.value(i) as i64);
}
}
}};
}
for (f, col) in root.schema().fields().iter().zip(root.columns()) {
let key = field_key(f.name());
match col.data_type() {
DataType::Timestamp(_, _) => ints!(TimestampNanosecondType, col, key),
DataType::Int64 => ints!(Int64Type, col, key),
DataType::Int32 => ints!(Int32Type, col, key),
DataType::UInt64 => ints!(UInt64Type, col, key),
DataType::UInt32 => ints!(UInt32Type, col, key),
DataType::UInt16 => ints!(UInt16Type, col, key),
DataType::UInt8 => ints!(UInt8Type, col, key),
DataType::Float64 => {
let a = col.as_primitive::<Float64Type>();
for i in 0..a.len() {
if !a.is_null(i) {
b.float(key, a.value(i));
}
}
}
_ => {}
}
}
}
fn index_attrs(b: &mut Builder, t: &RecordBatch) {
let dict = t.column(1).as_dictionary::<UInt16Type>();
let names = dict.values().as_string::<i32>();
let codes = dict.keys().values();
let types = t.column(2).as_primitive::<UInt8Type>().values();
let strs = crate::attrs::str_column(t);
let ints = t.column(4).as_primitive::<Int64Type>();
let doubles = t.column(5).as_primitive::<Float64Type>();
const STR: u8 = AttrType::Str as u8;
const INT: u8 = AttrType::Int as u8;
const DOUBLE: u8 = AttrType::Double as u8;
let hashes: Vec<u64> = (0..names.len()).map(|i| attr_key(names.value(i))).collect();
for row in 0..t.num_rows() {
let key = hashes[codes[row] as usize];
match types[row] {
INT => b.int(key, ints.value(row)),
DOUBLE => b.float(key, doubles.value(row)),
STR => match strs.value(row).parse::<f64>() {
Ok(v) => b.float(key, v),
Err(_) => b.any(key),
},
_ => {}
}
}
}
#[cfg(test)]
mod tests {
use std::sync::Arc;
use arrow_array::{Float64Array, Int64Array, RecordBatch, StringArray, UInt64Array};
use arrow_schema::{Field, Schema};
use super::*;
fn map_of(b: &Builder) -> Vec<u8> {
b.build().expect("something to write")
}
fn probe(key: u64, op: Op, int: Option<i64>, float: Option<f64>) -> Probe {
Probe {
key,
op,
int,
float,
}
}
#[test]
fn a_range_answers_the_five_ordered_operators_and_nothing_else() {
let mut b = Builder::default();
let k = attr_key("http.status_code");
b.int(k, 200);
b.int(k, 404);
let bytes = map_of(&b);
let m = Map::open(&bytes).expect("readable");
let ask = |op, t: i64| probe(k, op, Some(t), Some(t as f64)).maybe(&m);
assert!(ask(Op::Eq, 200) && ask(Op::Eq, 300) && !ask(Op::Eq, 500));
assert!(ask(Op::Gte, 404) && !ask(Op::Gte, 405));
assert!(ask(Op::Gt, 403) && !ask(Op::Gt, 404));
assert!(ask(Op::Lte, 200) && !ask(Op::Lte, 199));
assert!(ask(Op::Lt, 201) && !ask(Op::Lt, 200));
assert!(ask(Op::Ne, 200) && ask(Op::Contains, 999));
}
#[test]
fn a_key_the_block_never_saw_prunes_and_an_unreadable_file_does_not() {
let mut b = Builder::default();
b.int(attr_key("present"), 1);
let bytes = map_of(&b);
let m = Map::open(&bytes).expect("readable");
assert!(probe(attr_key("present"), Op::Gte, Some(0), Some(0.0)).maybe(&m));
assert!(!probe(attr_key("absent"), Op::Gte, Some(0), Some(0.0)).maybe(&m));
assert!(Map::open(&[]).is_none());
assert!(Map::open(&bytes[..HEADER]).is_none());
let mut torn = bytes.clone();
torn.pop();
assert!(Map::open(&torn).is_none());
let mut flipped = bytes.clone();
*flipped.last_mut().expect("non-empty") ^= 0xff;
assert!(Map::open(&flipped).is_none(), "the crc has to catch this");
let mut version = bytes.clone();
version[4] = 2;
assert!(Map::open(&version).is_none());
}
#[test]
fn an_integer_past_two_to_the_fifty_three_keeps_its_own_number_line() {
let v = (1i64 << 53) + 1;
let mut b = Builder::default();
let k = attr_key("bytes");
b.int(k, v);
let bytes = map_of(&b);
let m = Map::open(&bytes).expect("readable");
assert!(probe(k, Op::Gte, Some(v), Some(v as f64)).maybe(&m));
assert!(!probe(k, Op::Gt, Some(v), Some(v as f64)).maybe(&m));
}
#[test]
fn a_fractional_scalar_cannot_reach_an_integer_only_key() {
let mut b = Builder::default();
let k = attr_key("retries");
b.int(k, 3);
let bytes = map_of(&b);
let m = Map::open(&bytes).expect("readable");
assert!(!probe(k, Op::Eq, None, Some(3.5)).maybe(&m));
assert!(!probe(k, Op::Lt, None, Some(3.5)).maybe(&m));
}
#[test]
fn text_that_parses_is_a_number_and_text_that_does_not_gives_up_the_key() {
let attrs = |vals: Vec<&str>| {
let mut a = crate::attrs::AttrsBuilder::new("t");
for v in vals {
a.append(
0,
"code",
Some(&mira_proto::common::v1::AnyValue {
value: Some(mira_proto::common::v1::any_value::Value::StringValue(
v.into(),
)),
}),
)
.expect("appends");
}
vec![("t", a.finish().expect("finishes"))]
};
let k = attr_key("code");
let numeric = index(&attrs(vec!["200", "503"])).expect("a map");
let m = Map::open(&numeric).expect("readable");
assert!(probe(k, Op::Gte, Some(500), Some(500.0)).maybe(&m));
assert!(!probe(k, Op::Gt, Some(503), Some(503.0)).maybe(&m));
let mixed = index(&attrs(vec!["200", "unset"])).expect("a map");
let m = Map::open(&mixed).expect("readable");
assert!(probe(k, Op::Gt, Some(9999), Some(9999.0)).maybe(&m));
}
#[test]
fn the_root_tables_numeric_columns_are_in_it_and_the_others_are_not() {
let schema = Arc::new(Schema::new(vec![
Field::new("duration_nano", DataType::UInt64, false),
Field::new("count", DataType::Int64, true),
Field::new("ratio", DataType::Float64, false),
Field::new("body", DataType::Utf8, false),
]));
let root = RecordBatch::try_new(
schema,
vec![
Arc::new(UInt64Array::from(vec![10u64, 2_000_000_000])),
Arc::new(Int64Array::from(vec![None, None] as Vec<Option<i64>>)),
Arc::new(Float64Array::from(vec![0.25, f64::NAN])),
Arc::new(StringArray::from(vec!["a", "b"])),
],
)
.expect("a batch");
let bytes = index(&[("root", root)]).expect("a map");
let m = Map::open(&bytes).expect("readable");
let d = field_key("duration_nano");
assert!(probe(d, Op::Gt, Some(1_000_000_000), Some(1e9)).maybe(&m));
assert!(!probe(d, Op::Gt, Some(2_000_000_000), Some(2e9)).maybe(&m));
assert!(!probe(field_key("count"), Op::Gte, Some(0), Some(0.0)).maybe(&m));
assert!(!probe(field_key("body"), Op::Gte, Some(0), Some(0.0)).maybe(&m));
let r = field_key("ratio");
assert!(probe(r, Op::Lte, None, Some(0.25)).maybe(&m));
assert!(!probe(r, Op::Gt, None, Some(0.25)).maybe(&m));
}
#[test]
fn too_many_keys_writes_nothing_rather_than_a_map_nobody_wants() {
let mut b = Builder::default();
for i in 0..=MAX_KEYS {
b.int(attr_key(&format!("k{i}")), i as i64);
}
assert!(b.build().is_none(), "over the cap, so no file");
assert!(Builder::default().build().is_none(), "nothing to say");
}
}