use std::cmp::Ordering;
use std::path::Path;
use std::sync::Arc;
use std::time::{Duration, Instant};
use rudb_common::bounds::{self, Bound};
use rudb_common::stat::Class;
use rudb_common::{LogicalType, Result, Value};
use rudb_encoding::sketch::{DEFAULT_K, Sketch};
use rudb_stats::{Order, STRIPE_K, Sketches, Summary, sketches::HEADER_BYTES as SKETCH_HEADER};
use rudb_storage::count::{Counts, countable};
use rudb_vector::{Data, Form, Validity, Vector};
use crate::section::{self, Attachment};
use crate::{Catalog, Reader, invalid};
pub const BUDGET_SHARE: u64 = 2;
pub const BUDGET_FLOOR: u64 = 64 * 1024;
#[derive(Debug, Clone)]
pub struct Built {
pub column: usize,
pub rows: u64,
pub distinct: u64,
pub exact: bool,
pub order: Order,
pub summary_bytes: usize,
pub sketch_bytes: usize,
pub stripes: usize,
pub column_bytes: u64,
pub built: bool,
pub build: Duration,
}
impl Built {
#[must_use]
pub fn bytes(&self) -> usize {
self.summary_bytes + self.sketch_bytes
}
}
#[derive(Debug, Clone)]
pub struct Stats {
pub summary: Summary,
pub sketches: Sketches,
}
pub fn build_summary(reader: &Reader, column: usize) -> Result<Stats> {
build_summary_for(reader, column, false)
}
pub fn build_summary_for(reader: &Reader, column: usize, per_stripe: bool) -> Result<Stats> {
let fields = reader.table().fields();
let Some(field) = fields.get(column) else {
return Err(invalid(&format!(
"column {column} is past the {} of table {}",
fields.len(),
reader.table().name()
)));
};
if !countable(&field.ty) {
return Err(invalid(&format!(
"a summary of {} needs a hash rule, and {} has none",
field.name, field.ty
)));
}
let blind = || {
invalid(&format!(
"column {} of {} holds a form with no hash rule, so it has no sketch",
field.name,
reader.table().name()
))
};
let mut whole = Counts::new(1);
let mut stripes = Vec::new();
let mut pass = Pass::new(&field.ty, reader.table().generation());
for (at, stripe) in reader.stripe_parts().into_iter().enumerate() {
pass.open_stripe((at as u64, 0));
let mut counted = per_stripe.then(|| Counts::new(1));
for part in stripe {
let chunk = reader.read(part, &[column])?;
match counted.as_mut() {
Some(counted) => counted.add(&chunk),
None => whole.add(&chunk),
}
pass.scan(chunk.column(0)?);
}
pass.close_stripe();
if let Some(counted) = counted {
stripes.push(counted.sketch(0).ok_or_else(blind)?);
}
}
if !per_stripe {
return Ok(pass.finish(whole.sketch(0).ok_or_else(blind)?, Vec::new()));
}
let mut merged = Sketch::new(DEFAULT_K)?;
for stripe in &stripes {
merged = merged.union(stripe)?;
}
let narrowed =
stripes.iter().map(|stripe| stripe.narrowed(STRIPE_K)).collect::<Result<Vec<_>>>()?;
Ok(pass.finish(merged, narrowed))
}
#[derive(Debug)]
struct Piece {
key: (u64, u64),
first: Option<Bound>,
last: Option<Bound>,
ascending: bool,
descending: bool,
runs: u64,
}
#[derive(Debug)]
struct Pass {
rows: u64,
nulls: u64,
low: Option<Bound>,
high: Option<Bound>,
bounded: bool,
ascending: bool,
descending: bool,
runs: u64,
previous: Option<Bound>,
bytes: u64,
widest: u64,
generation: u64,
stripe_low: Option<Bound>,
stripe_high: Option<Bound>,
stripes: Vec<(Bound, Bound)>,
key: (u64, u64),
first: Option<Bound>,
pieces: Vec<Piece>,
fixed: Option<u64>,
scale: Option<u8>,
coded: Option<Coded>,
}
impl Pass {
fn new(ty: &LogicalType, generation: u64) -> Self {
Self {
rows: 0,
nulls: 0,
low: None,
high: None,
bounded: true,
ascending: true,
descending: true,
runs: 0,
previous: None,
bytes: 0,
widest: 0,
generation,
stripe_low: None,
stripe_high: None,
stripes: Vec::new(),
key: (0, 0),
first: None,
pieces: Vec::new(),
fixed: fixed_width(ty),
scale: bounds::scale_of(ty),
coded: None,
}
}
fn scan(&mut self, vector: &Vector) {
if self.scan_flat(vector) || self.scan_dictionary(vector) {
return;
}
self.scan_rows(vector);
}
fn scan_flat(&mut self, vector: &Vector) -> bool {
if vector.form() != Form::Flat {
return false;
}
let Some(data) = vector.data() else { return false };
let rows = vector.len();
let validity = vector.validity();
macro_rules! signed {
($(($variant:ident, $native:ty, $zero:expr)),+ $(,)?) => {
match data {
$(Data::$variant(held) => {
let held: &[$native] = held;
if held.len() < rows {
return false;
}
let spread = spread(rows, validity, |row| i128::from(held[row]));
let width = self.fixed.unwrap_or(8);
self.fold(Reduced {
rows: spread.rows,
nulls: spread.nulls,
values: spread.values,
bytes: width.saturating_mul(spread.values),
widest: if spread.values > 0 { width } else { 0 },
ascents: spread.ascents,
descents: spread.descents,
ends: (spread.values > 0).then(|| Ends {
low: self.bound(spread.low),
high: self.bound(spread.high),
first: self.bound(spread.first),
last: self.bound(spread.last),
}),
});
return true;
})+
_ => false,
}
};
}
rudb_vector::for_each_layout!(signed, signed)
}
fn scan_dictionary(&mut self, vector: &Vector) -> bool {
let Some((codes, values)) = vector.shared_dictionary_parts() else { return false };
let rows = vector.len();
if codes.len() < rows {
return false;
}
let held = match self.coded.take() {
Some(held) if Arc::ptr_eq(&held.values, values) => held,
_ => {
if values.len() > rows {
return false;
}
match self.read_dictionary(values) {
Some(read) => read,
None => return false,
}
}
};
let out = held.reduce(codes, rows, vector.validity());
self.coded = Some(held);
self.fold(out);
true
}
fn read_dictionary(&self, values: &Arc<Vector>) -> Option<Coded> {
let mut entries = Vec::with_capacity(values.len());
for at in 0..values.len() {
if values.is_null_at(at) {
entries.push(None);
continue;
}
let entry = match values.signed_at(at) {
Some(signed) => Some((self.bound(signed), self.fixed.unwrap_or(8))),
None => match values.bytes_at(at) {
Some(bytes) => Some((Bound::Bytes(bytes.to_vec()), bytes.len() as u64)),
None => {
let value = values.value_at(at);
let wide = self.fixed.unwrap_or_else(|| width(&value));
Bound::of_value(&value).map(|bound| (bound, wide))
}
},
};
entries.push(Some(entry?));
}
let mut order = (0..entries.len()).filter(|&at| entries[at].is_some()).collect::<Vec<_>>();
order.sort_by(|&one, &other| {
bound_of(&entries, one).order(bound_of(&entries, other)).unwrap_or(Ordering::Equal)
});
let mut codes = vec![None; entries.len()];
let mut bounds = Vec::new();
for (at, &code) in order.iter().enumerate() {
if at > 0 {
match bound_of(&entries, order[at - 1]).order(bound_of(&entries, code)) {
Some(Ordering::Less) => bounds.push(bound_of(&entries, code).clone()),
Some(Ordering::Equal) => {}
Some(Ordering::Greater) | None => return None,
}
} else {
bounds.push(bound_of(&entries, code).clone());
}
let width = entries[code].as_ref().map_or(0, |(_, width)| *width);
codes[code] = Some(((bounds.len() - 1) as u32, width));
}
Some(Coded { values: Arc::clone(values), codes, bounds })
}
fn fold(&mut self, one: Reduced) {
self.rows += one.rows;
self.nulls += one.nulls;
self.bytes = self.bytes.saturating_add(one.bytes);
self.widest = self.widest.max(one.widest);
let Some(ends) = one.ends else { return };
match self.previous.take() {
None => {
self.runs = 1;
self.first = Some(ends.first.clone());
}
Some(previous) => self.run(Some(previous.order(&ends.first))),
}
self.runs += one.descents;
if one.descents > 0 {
self.ascending = false;
}
if one.ascents > 0 {
self.descending = false;
}
if takes(&self.low, &ends.low, Ordering::Less) {
self.low = Some(ends.low.clone());
}
if takes(&self.stripe_low, &ends.low, Ordering::Less) {
self.stripe_low = Some(ends.low);
}
if takes(&self.high, &ends.high, Ordering::Greater) {
self.high = Some(ends.high.clone());
}
if takes(&self.stripe_high, &ends.high, Ordering::Greater) {
self.stripe_high = Some(ends.high);
}
self.previous = Some(ends.last);
}
fn bound(&self, signed: i128) -> Bound {
match self.scale {
Some(scale) => Bound::Scaled { unscaled: signed, scale },
None => Bound::Int(signed),
}
}
fn scan_rows(&mut self, vector: &Vector) {
for row in 0..vector.len() {
self.rows += 1;
if vector.is_null_at(row) {
self.nulls += 1;
continue;
}
if let Some(signed) = vector.signed_at(row) {
let bound = match self.scale {
Some(scale) => Bound::Scaled { unscaled: signed, scale },
None => Bound::Int(signed),
};
self.value(bound, self.fixed.unwrap_or(8));
continue;
}
if let Some(bytes) = vector.bytes_at(row) {
self.bytes_value(bytes);
continue;
}
let value = vector.value_at(row);
let width = self.fixed.unwrap_or_else(|| width(&value));
match Bound::of_value(&value) {
Some(bound) => self.value(bound, width),
None => {
self.bytes = self.bytes.saturating_add(width);
self.widest = self.widest.max(width);
self.bounded = false;
self.ascending = false;
self.descending = false;
}
}
}
}
fn value(&mut self, bound: Bound, width: u64) {
self.measure(width);
let ordering = self.previous.as_ref().map(|previous| previous.order(&bound));
if ordering.is_none() {
self.first = Some(bound.clone());
}
self.run(ordering);
if takes(&self.low, &bound, Ordering::Less) {
self.low = Some(bound.clone());
}
if takes(&self.high, &bound, Ordering::Greater) {
self.high = Some(bound.clone());
}
if takes(&self.stripe_low, &bound, Ordering::Less) {
self.stripe_low = Some(bound.clone());
}
if takes(&self.stripe_high, &bound, Ordering::Greater) {
self.stripe_high = Some(bound.clone());
}
self.previous = Some(bound);
}
fn bytes_value(&mut self, bytes: &[u8]) {
self.measure(bytes.len() as u64);
let ordering = match &self.previous {
None => {
self.first = Some(Bound::Bytes(bytes.to_vec()));
None
}
Some(Bound::Bytes(previous)) => Some(Some(previous.as_slice().cmp(bytes))),
Some(_) => Some(None),
};
self.run(ordering);
if takes_bytes(&self.low, bytes, Ordering::Less) {
fill(&mut self.low, bytes);
}
if takes_bytes(&self.high, bytes, Ordering::Greater) {
fill(&mut self.high, bytes);
}
if takes_bytes(&self.stripe_low, bytes, Ordering::Less) {
fill(&mut self.stripe_low, bytes);
}
if takes_bytes(&self.stripe_high, bytes, Ordering::Greater) {
fill(&mut self.stripe_high, bytes);
}
fill(&mut self.previous, bytes);
}
fn measure(&mut self, width: u64) {
self.bytes = self.bytes.saturating_add(width);
self.widest = self.widest.max(width);
}
fn run(&mut self, ordering: Option<Option<Ordering>>) {
match ordering {
None => self.runs = 1,
Some(Some(Ordering::Less)) => self.descending = false,
Some(Some(Ordering::Greater)) => {
self.ascending = false;
self.runs += 1;
}
Some(Some(Ordering::Equal)) => {}
Some(None) => {
self.ascending = false;
self.descending = false;
}
}
}
fn open_stripe(&mut self, key: (u64, u64)) {
self.stripe_low = None;
self.stripe_high = None;
self.key = key;
self.first = None;
self.previous = None;
self.ascending = true;
self.descending = true;
self.runs = 0;
}
fn close_stripe(&mut self) {
if let (Some(low), Some(high)) = (self.stripe_low.take(), self.stripe_high.take()) {
self.stripes.push((low, high));
}
self.pieces.push(Piece {
key: self.key,
first: self.first.take(),
last: self.previous.take(),
ascending: self.ascending,
descending: self.descending,
runs: self.runs,
});
}
fn absorb(&mut self, later: Pass) {
self.rows += later.rows;
self.nulls += later.nulls;
self.bounded &= later.bounded;
self.bytes = self.bytes.saturating_add(later.bytes);
self.widest = self.widest.max(later.widest);
if let Some(low) = later.low {
if takes(&self.low, &low, Ordering::Less) {
self.low = Some(low);
}
}
if let Some(high) = later.high {
if takes(&self.high, &high, Ordering::Greater) {
self.high = Some(high);
}
}
self.stripes.extend(later.stripes);
self.pieces.extend(later.pieces);
}
fn settle(&mut self) {
if self.pieces.is_empty() {
return;
}
let mut pieces = std::mem::take(&mut self.pieces);
pieces.sort_by_key(|piece| piece.key);
let (mut ascending, mut descending, mut runs) = (true, true, 0_u64);
let mut previous: Option<Bound> = None;
for piece in pieces {
ascending &= piece.ascending;
descending &= piece.descending;
let (Some(first), Some(last)) = (piece.first, piece.last) else { continue };
runs += piece.runs;
if let Some(previous) = &previous {
match previous.order(&first) {
Some(Ordering::Less) => descending = false,
Some(Ordering::Greater) => ascending = false,
Some(Ordering::Equal) => {}
None => {
ascending = false;
descending = false;
}
}
if previous.order(&first) != Some(Ordering::Greater) {
runs = runs.saturating_sub(1);
}
}
previous = Some(last);
}
self.ascending = ascending;
self.descending = descending;
self.runs = runs;
}
fn finish(mut self, sketch: Sketch, stripes: Vec<Sketch>) -> Stats {
self.settle();
let present = self.rows - self.nulls;
let exact = sketch.is_exact();
let distinct = if exact {
sketch.len() as u64
} else {
#[allow(clippy::cast_possible_truncation, clippy::cast_sign_loss)]
let estimate = sketch.distinct().round().max(0.0) as u64;
estimate.min(present)
};
let summary = Summary {
rows: self.rows,
nulls: self.nulls,
low: if self.bounded { self.low } else { None },
high: if self.bounded { self.high } else { None },
ends_exact: self.bounded,
distinct,
distinct_class: if exact { Class::Exact } else { Class::Estimated },
unique: exact && distinct == present,
order: if present == 0 {
Order::Neither
} else if self.ascending {
Order::Ascending
} else if self.descending {
Order::Descending
} else {
Order::Neither
},
runs: self.runs,
overlapping: overlapping(&self.stripes),
bytes: self.bytes,
widest: self.widest,
newest: self.generation,
};
let sketches = match Sketches::new(sketch.clone(), stripes) {
Ok(sketches) => sketches,
Err(_) => Sketches::merged(sketch),
};
Stats { summary, sketches }
}
}
#[derive(Debug)]
struct Coded {
values: Arc<Vector>,
codes: Vec<Option<(u32, u64)>>,
bounds: Vec<Bound>,
}
impl Coded {
fn reduce(&self, codes: &[u32], rows: usize, validity: &Validity) -> Reduced {
let nullable = validity.has_nulls(rows);
let mut out = Reduced::empty(rows as u64);
let (mut low, mut high, mut first, mut last) = (0_u32, 0_u32, 0_u32, 0_u32);
for (row, &code) in codes.iter().take(rows).enumerate() {
let entry = if nullable && !validity.is_valid(row) {
None
} else {
self.codes.get(code as usize).copied().flatten()
};
let Some((position, width)) = entry else {
out.nulls += 1;
continue;
};
out.bytes = out.bytes.saturating_add(width);
out.widest = out.widest.max(width);
if out.values == 0 {
low = position;
high = position;
first = position;
} else if position < last {
out.descents += 1;
} else if position > last {
out.ascents += 1;
}
low = low.min(position);
high = high.max(position);
last = position;
out.values += 1;
}
let at = |position: u32| self.bounds[position as usize].clone();
out.ends = (out.values > 0).then(|| Ends {
low: at(low),
high: at(high),
first: at(first),
last: at(last),
});
out
}
}
#[derive(Debug)]
struct Reduced {
rows: u64,
nulls: u64,
values: u64,
bytes: u64,
widest: u64,
ascents: u64,
descents: u64,
ends: Option<Ends>,
}
impl Reduced {
fn empty(rows: u64) -> Self {
Self { rows, nulls: 0, values: 0, bytes: 0, widest: 0, ascents: 0, descents: 0, ends: None }
}
}
#[derive(Debug)]
struct Ends {
low: Bound,
high: Bound,
first: Bound,
last: Bound,
}
fn bound_of(entries: &[Option<(Bound, u64)>], at: usize) -> &Bound {
match &entries[at] {
Some((bound, _)) => bound,
None => &Bound::Int(i128::MIN),
}
}
#[derive(Debug)]
struct Spread {
rows: u64,
nulls: u64,
low: i128,
high: i128,
first: i128,
last: i128,
descents: u64,
ascents: u64,
values: u64,
}
fn spread(rows: usize, validity: &Validity, get: impl Fn(usize) -> i128) -> Spread {
let mut out = Spread {
rows: rows as u64,
nulls: 0,
low: 0,
high: 0,
first: 0,
last: 0,
descents: 0,
ascents: 0,
values: 0,
};
let nullable = validity.has_nulls(rows);
for row in 0..rows {
if nullable && !validity.is_valid(row) {
out.nulls += 1;
continue;
}
let value = get(row);
if out.values == 0 {
out.low = value;
out.high = value;
out.first = value;
} else {
if value < out.last {
out.descents += 1;
} else if value > out.last {
out.ascents += 1;
}
out.low = out.low.min(value);
out.high = out.high.max(value);
}
out.last = value;
out.values += 1;
}
out
}
fn takes(held: &Option<Bound>, bound: &Bound, want: Ordering) -> bool {
match held {
None => true,
Some(held) => bound.order(held) == Some(want),
}
}
fn takes_bytes(held: &Option<Bound>, bytes: &[u8], want: Ordering) -> bool {
match held {
None => true,
Some(Bound::Bytes(held)) => bytes.cmp(held.as_slice()) == want,
Some(_) => false,
}
}
fn fill(held: &mut Option<Bound>, bytes: &[u8]) {
match held {
Some(Bound::Bytes(held)) => {
held.clear();
held.extend_from_slice(bytes);
}
held => *held = Some(Bound::Bytes(bytes.to_vec())),
}
}
fn overlapping(stripes: &[(Bound, Bound)]) -> bool {
let mut order = (0..stripes.len()).collect::<Vec<_>>();
order
.sort_by(|&one, &other| stripes[one].0.order(&stripes[other].0).unwrap_or(Ordering::Equal));
order.windows(2).any(|pair| {
let before = &stripes[pair[0]].1;
let after = &stripes[pair[1]].0;
before.order(after) != Some(Ordering::Less)
})
}
fn fixed_width(ty: &LogicalType) -> Option<u64> {
Some(match ty {
LogicalType::Boolean | LogicalType::TinyInt | LogicalType::UTinyInt => 1,
LogicalType::SmallInt | LogicalType::USmallInt => 2,
LogicalType::Integer | LogicalType::UInteger | LogicalType::Float | LogicalType::Date => 4,
LogicalType::HugeInt | LogicalType::UHugeInt | LogicalType::Decimal { .. } => 16,
LogicalType::Varchar | LogicalType::Blob => return None,
_ => 8,
})
}
fn width(value: &Value) -> u64 {
match value {
Value::Null => 0,
Value::Boolean(_) | Value::TinyInt(_) | Value::UTinyInt(_) => 1,
Value::SmallInt(_) | Value::USmallInt(_) => 2,
Value::Integer(_) | Value::UInteger(_) | Value::Float(_) | Value::Date(_) => 4,
Value::HugeInt(_) | Value::UHugeInt(_) | Value::Decimal { .. } => 16,
Value::Varchar(text) => text.len() as u64,
Value::Blob(bytes) => bytes.len() as u64,
_ => 8,
}
}
#[derive(Debug)]
pub(crate) struct Gather {
pass: Pass,
counts: Counts,
}
impl Gather {
pub(crate) fn new(ty: &LogicalType, generation: u64) -> Option<Self> {
countable(ty).then(|| Self { pass: Pass::new(ty, generation), counts: Counts::new(1) })
}
pub(crate) fn stripe<'a>(&mut self, key: (u64, u64), parts: impl Iterator<Item = &'a Vector>) {
self.pass.open_stripe(key);
for vector in parts {
self.counts.add_column(0, vector);
self.pass.scan(vector);
}
self.pass.close_stripe();
}
pub(crate) fn absorb(&mut self, later: Gather) {
self.pass.absorb(later.pass);
self.counts.absorb(later.counts);
}
pub(crate) fn rows(&self) -> u64 {
self.pass.rows
}
pub(crate) fn finish(self) -> Option<Stats> {
let sketch = self.counts.sketch(0)?;
Some(self.pass.finish(sketch, Vec::new()))
}
}
pub(crate) fn column_bytes(table: &crate::Table) -> u64 {
(0..table.fields.len())
.map(|at| {
crate::sum(table.stripes.iter().map(|stripe| crate::span_bytes(&stripe.pages, at)))
.saturating_add(crate::sum(
table.stripes.iter().map(|stripe| stripe.memberships.bytes(at)),
))
.saturating_add(crate::sum(
table.stripes.iter().map(|stripe| stripe.sieves.bytes(at)),
))
.saturating_add(crate::sum(
table.stripes.iter().map(|stripe| stripe.part_ranges.bytes(at)),
))
.saturating_add(crate::dictionary_bytes(table, at))
})
.fold(0, u64::saturating_add)
}
pub(crate) fn within(costs: &[usize], allowance: u64, spent: u64) -> Vec<bool> {
let mut order = (0..costs.len()).collect::<Vec<_>>();
order.sort_by_key(|&at| costs[at]);
let mut spent = spent;
let mut keep = vec![false; costs.len()];
for at in order {
let cost = costs[at] as u64;
if spent.saturating_add(cost) <= allowance {
spent += cost;
keep[at] = true;
}
}
keep
}
pub(crate) fn allowance(column_bytes: u64, share: u64) -> u64 {
(column_bytes.saturating_mul(share) / 100).max(BUDGET_FLOOR)
}
pub fn build_stats(path: &Path, table: &str, columns: &[usize]) -> Result<Vec<Built>> {
build_stats_within(path, table, columns, BUDGET_SHARE)
}
#[must_use]
pub fn read_columns(reader: &Reader) -> Vec<usize> {
let generation = reader.table().generation();
let mut promoted = reader
.table()
.sections()
.iter()
.filter(|held| held.among(section::GRAPH_KINDS) && held.usable(generation))
.filter_map(|held| usize::try_from(held.id).ok())
.collect::<Vec<_>>();
promoted.sort_unstable();
promoted.dedup();
promoted
}
pub fn build_stats_within(
path: &Path,
table: &str,
columns: &[usize],
share: u64,
) -> Result<Vec<Built>> {
let promoted = read_columns(&Catalog::open(path)?.table(table)?);
build_stats_for(path, table, columns, &promoted, share)
}
pub fn build_stats_for(
path: &Path,
table: &str,
columns: &[usize],
per_stripe: &[usize],
share: u64,
) -> Result<Vec<Built>> {
let reader = Catalog::open(path)?.table(table)?;
let column_bytes = reader.layout().columns_total();
let allowance = allowance(column_bytes, share);
let spent = held_bytes(&reader, columns)?;
let mut report = Vec::with_capacity(columns.len());
let mut payloads = Vec::with_capacity(columns.len());
for &column in columns {
let start = Instant::now();
let stats = build_summary_for(&reader, column, per_stripe.contains(&column))?;
let mut summary = Vec::new();
stats.summary.encode(&mut summary)?;
let mut sketches = Vec::new();
stats.sketches.encode(&mut sketches)?;
report.push(Built {
column,
rows: stats.summary.rows,
distinct: stats.summary.distinct,
exact: stats.summary.distinct_class == Class::Exact,
order: stats.summary.order,
summary_bytes: summary.len(),
sketch_bytes: sketches.len(),
stripes: stats.sketches.stripes.len(),
column_bytes,
built: false,
build: start.elapsed(),
});
payloads.push((column, summary, sketches));
}
let costs = report.iter().map(Built::bytes).collect::<Vec<_>>();
let keep = within(&costs, allowance, spent);
for (one, &keep) in report.iter_mut().zip(&keep) {
one.built = keep;
}
drop(reader);
let mut attachments = Vec::with_capacity(payloads.len() * 2);
for ((column, summary, sketches), _) in payloads.iter().zip(&keep).filter(|&(_, &keep)| keep) {
let id = u64::try_from(*column).map_err(|_| invalid("column index overflow"))?;
attachments.push(Attachment {
kind: *section::SUMMARY,
id,
flags: 0,
header_bytes: u32::try_from(summary.len())
.map_err(|_| invalid("a summary longer than a u32 can count"))?,
bytes: summary,
});
attachments.push(Attachment {
kind: *section::SKETCHES,
id,
flags: 0,
header_bytes: SKETCH_HEADER,
bytes: sketches,
});
}
crate::attach(path, table, &attachments)?;
Ok(report)
}
fn held_bytes(reader: &Reader, replacing: &[usize]) -> Result<u64> {
let mut total = 0;
for held in reader.table().sections() {
if !held.among(section::STATISTICS_KINDS) {
continue;
}
let replaced = replacing.iter().any(|&column| u64::try_from(column) == Ok(held.id));
if replaced || !held.usable(reader.table().generation()) {
continue;
}
let Ok(extents) = reader.extents(held) else { continue };
total += extents.iter().map(|extent| u64::from(extent.length)).sum::<u64>();
}
Ok(total)
}
#[must_use]
pub fn summary(reader: &Reader, column: usize) -> Option<Summary> {
let bytes = payload(reader, column, section::SUMMARY)?;
Summary::decode(&bytes).ok()
}
#[must_use]
pub fn sketches(reader: &Reader, column: usize) -> Option<Sketches> {
let bytes = payload(reader, column, section::SKETCHES)?;
Sketches::decode(&bytes).ok()
}
fn payload(reader: &Reader, column: usize, kind: &[u8; 8]) -> Option<Vec<u8>> {
let table = reader.table();
let id = u64::try_from(column).ok()?;
let held = table.sections().iter().find(|section| section.kind == *kind && section.id == id)?;
if !held.usable(table.generation()) {
return None;
}
reader.payload(held).ok()
}
#[must_use]
pub fn summarizable(ty: &LogicalType) -> bool {
countable(ty)
}
#[cfg(test)]
mod tests {
use std::fs;
use std::path::PathBuf;
use std::sync::Arc;
use std::time::{SystemTime, UNIX_EPOCH};
use rudb_common::Field;
use rudb_encoding::sketch::hash64;
use rudb_storage::count::hash_value;
use rudb_vector::{Chunk, Vector};
use super::*;
use crate::Writer;
fn path(label: &str) -> PathBuf {
let stamp = SystemTime::now().duration_since(UNIX_EPOCH).expect("time advances").as_nanos();
std::env::temp_dir().join(format!("rudb-stats-{label}-{}-{stamp}.rdb", std::process::id()))
}
fn table_of(label: &str, values: &[Option<i64>]) -> PathBuf {
let path = path(label);
let mut writer =
Writer::create(&path, "t", vec![Field::new("v", LogicalType::BigInt)]).expect("new");
for part in values.chunks(1000) {
let held =
part.iter().map(|v| v.map_or(Value::Null, Value::BigInt)).collect::<Vec<_>>();
let chunk =
Chunk::new(vec![Vector::from_values(LogicalType::BigInt, &held).expect("values")])
.expect("one column");
writer.append(&chunk).expect("a part");
}
writer.finish().expect("commit");
path
}
fn table_of_parts(label: &str, values: &[Option<i64>], per_part: usize) -> PathBuf {
let path = path(label);
let mut writer =
Writer::create(&path, "t", vec![Field::new("v", LogicalType::BigInt)]).expect("new");
for part in values.chunks(per_part) {
let held =
part.iter().map(|v| v.map_or(Value::Null, Value::BigInt)).collect::<Vec<_>>();
let chunk =
Chunk::new(vec![Vector::from_values(LogicalType::BigInt, &held).expect("values")])
.expect("one column");
writer.append(&chunk).expect("a part");
}
writer.finish().expect("commit");
path
}
#[test]
fn stripes_that_arrive_out_of_order_are_summarized_in_the_order_they_are_read() {
let path = path("late-stripes");
let mut writer =
Writer::create(&path, "t", vec![Field::new("v", LogicalType::BigInt)]).expect("new");
let part = |from: i64| {
let held = (from..from + 10).map(|v| Value::BigInt(v / 2)).collect::<Vec<_>>();
Chunk::new(vec![Vector::from_values(LogicalType::BigInt, &held).expect("values")])
.expect("one column")
};
for stripe in [2_u64, 0, 1] {
let parts = (0..3)
.map(|at| {
(
(stripe * 3 + at, 0),
part(i64::try_from(stripe * 30 + at * 10).expect("small")),
)
})
.collect();
writer.append_stripe(parts).expect("a stripe");
}
writer.finish().expect("commit");
let reader = reopen(&path);
let summary = summary(&reader, 0).expect("the summary is in the file");
assert_eq!(summary.rows, 90);
assert_eq!(summary.order, Order::Ascending, "the stripes are in order once sorted");
assert_eq!(summary.runs, 1, "and the seams between them are not descents");
assert_eq!(crate::ascending(&reader), vec!["v".to_owned()]);
}
#[test]
fn the_vector_at_a_time_pass_says_what_the_row_at_a_time_pass_says() {
let shuffled = (0..500_i64).map(|at| Some(1 + at * 307 % 500)).collect::<Vec<_>>();
let shapes: [(&str, Vec<Option<i64>>); 7] = [
("ascending", (1..=500_i64).map(Some).collect()),
("descending", (1..=500_i64).rev().map(Some).collect()),
("constant", vec![Some(7); 500]),
("shuffled", shuffled),
("every third null", (1..=500_i64).map(|at| (at % 3 != 0).then_some(at)).collect()),
("all nulls", vec![None; 500]),
("twenty values over and over", (0..500_i64).map(|at| Some(at * 7 % 20)).collect()),
];
let types = [
LogicalType::SmallInt,
LogicalType::Integer,
LogicalType::BigInt,
LogicalType::Decimal { width: 18, scale: 2 },
LogicalType::Varchar,
];
for (label, values) in &shapes {
for ty in &types {
let held = values
.chunks(60)
.map(|part| {
let values = part.iter().map(|value| one(ty, *value)).collect::<Vec<_>>();
Vector::from_values(ty.clone(), &values).expect("values")
})
.collect::<Vec<_>>();
let coded = values
.chunks(60)
.map(|part| {
let mut distinct = part.to_vec();
distinct.sort_unstable();
distinct.dedup();
distinct.reverse();
let values =
distinct.iter().map(|value| one(ty, *value)).collect::<Vec<_>>();
let codes = part
.iter()
.map(|value| {
distinct.iter().position(|held| held == value).expect("a code")
as u32
})
.collect::<Vec<_>>();
Vector::dictionary(
codes,
Vector::from_values(ty.clone(), &values).expect("values"),
)
.expect("a dictionary")
})
.collect::<Vec<_>>();
let flat = drive(ty, &held, |pass, vector| {
assert!(pass.scan_flat(vector) || *ty == LogicalType::Varchar, "{label} {ty}");
if *ty == LogicalType::Varchar {
pass.scan_rows(vector);
}
});
let dictionary = drive(ty, &coded, |pass, vector| {
assert!(pass.scan_dictionary(vector), "{label} {ty} is dictionary coded");
});
let rows = drive(ty, &held, Pass::scan_rows);
assert_eq!(flat.summary, rows.summary, "flat: {label} {ty}");
assert_eq!(dictionary.summary, rows.summary, "dictionary: {label} {ty}");
let Some(shared) = shared(ty, values) else { continue };
let coded = values
.chunks(60)
.map(|part| {
let codes = part.iter().map(|value| code(values, *value)).collect();
Vector::dictionary_over(codes, Arc::clone(&shared)).expect("a dictionary")
})
.collect::<Vec<_>>();
let held = drive(ty, &coded, |pass, vector| {
assert!(pass.scan_dictionary(vector), "{label} {ty} is dictionary coded");
});
assert_eq!(held.summary, rows.summary, "one dictionary: {label} {ty}");
}
}
}
fn shared(ty: &LogicalType, values: &[Option<i64>]) -> Option<Arc<Vector>> {
let mut distinct = values.to_vec();
distinct.sort_unstable();
distinct.dedup();
if distinct.len() > 60 {
return None;
}
distinct.reverse();
let held = distinct.iter().map(|value| one(ty, *value)).collect::<Vec<_>>();
Some(Arc::new(Vector::from_values(ty.clone(), &held).expect("values")))
}
fn code(values: &[Option<i64>], value: Option<i64>) -> u32 {
let mut distinct = values.to_vec();
distinct.sort_unstable();
distinct.dedup();
distinct.reverse();
distinct.iter().position(|held| *held == value).expect("a code") as u32
}
fn one(ty: &LogicalType, value: Option<i64>) -> Value {
let Some(value) = value else { return Value::Null };
match ty {
LogicalType::SmallInt => Value::SmallInt(value as i16),
LogicalType::Integer => Value::Integer(value as i32),
LogicalType::BigInt => Value::BigInt(value),
LogicalType::Varchar => Value::Varchar(format!("v{value:04}")),
_ => Value::Decimal { unscaled: i128::from(value), width: 18, scale: 2 },
}
}
fn drive(ty: &LogicalType, held: &[Vector], mut scan: impl FnMut(&mut Pass, &Vector)) -> Stats {
let mut pass = Pass::new(ty, 1);
for (at, stripe) in held.chunks(5).enumerate() {
pass.open_stripe((at as u64, 0));
for vector in stripe {
scan(&mut pass, vector);
}
pass.close_stripe();
}
pass.finish(Sketch::of(&[]), Vec::new())
}
fn table_of_intervals(label: &str, months: &[i32]) -> PathBuf {
let path = path(label);
let mut writer =
Writer::create(&path, "t", vec![Field::new("v", LogicalType::Interval)]).expect("new");
for part in months.chunks(1000) {
let held = part
.iter()
.map(|months| Value::Interval { months: *months, days: 0, micros: 0 })
.collect::<Vec<_>>();
let chunk = Chunk::new(vec![
Vector::from_values(LogicalType::Interval, &held).expect("values"),
])
.expect("one column");
writer.append(&chunk).expect("a part");
}
writer.finish().expect("commit");
path
}
fn rows_of(reader: &Reader) -> Vec<Value> {
let mut out = Vec::new();
for part in 0..reader.parts() {
let chunk = reader.read(part, &[0]).expect("a part reads back");
for row in 0..chunk.len() {
out.push(chunk.value_at(0, row));
}
}
out
}
fn reopen(path: &PathBuf) -> Reader {
Catalog::open(path).expect("reopen").table("t").expect("the table")
}
#[test]
fn a_summary_built_over_a_file_says_what_the_column_holds() {
let values = (1..=3000_i64).map(Some).collect::<Vec<_>>();
let path = table_of("sorted", &values);
let built = build_stats(&path, "t", &[0]).expect("build");
assert_eq!(built.len(), 1);
assert!(built[0].built, "a one column table is nowhere near the budget");
assert_eq!(built[0].rows, 3000);
assert_eq!(built[0].distinct, 3000);
assert!(built[0].exact, "three thousand values is under the default k");
assert_eq!(built[0].order, Order::Ascending);
let reader = reopen(&path);
let summary = summary(&reader, 0).expect("the summary is in the file");
assert_eq!(summary.rows, 3000);
assert_eq!(summary.nulls, 0);
assert_eq!(summary.low, Some(Bound::Int(1)));
assert_eq!(summary.high, Some(Bound::Int(3000)));
assert!(summary.ends_exact);
assert!(summary.unique, "a sorted run of distinct values is a key candidate");
assert_eq!(summary.runs, 1, "one ascending run");
assert_eq!(summary.distinct_class, Class::Exact);
assert_eq!(summary.newest, reader.table().generation());
let sketches = sketches(&reader, 0).expect("the sketches are in the file");
assert!(sketches.merged.is_exact());
assert!(sketches.stripes.is_empty(), "the per stripe rule gives this column none");
fs::remove_file(&path).expect("clean up");
}
#[test]
fn nulls_are_counted_and_do_not_reach_the_ends_or_the_sketch() {
let values: Vec<Option<i64>> =
(0..2000).map(|at| if at % 3 == 0 { None } else { Some(at) }).collect();
let path = table_of("nulls", &values);
build_stats(&path, "t", &[0]).expect("build");
let reader = reopen(&path);
let summary = summary(&reader, 0).expect("the summary");
let nulls = values.iter().filter(|v| v.is_none()).count() as u64;
assert_eq!(summary.rows, 2000);
assert_eq!(summary.nulls, nulls);
assert_eq!(summary.present(), 2000 - nulls);
assert_eq!(summary.distinct, 2000 - nulls, "a null is not a distinct value");
assert_eq!(summary.low, Some(Bound::Int(1)), "zero is null here");
assert!(summary.unique);
fs::remove_file(&path).expect("clean up");
}
#[test]
fn a_column_that_repeats_is_not_reported_unique_and_a_descending_one_is_seen() {
let values = (0..2000_i64).map(|at| Some(-(at / 2))).collect::<Vec<_>>();
let path = table_of("repeats", &values);
build_stats(&path, "t", &[0]).expect("build");
let reader = reopen(&path);
let summary = summary(&reader, 0).expect("the summary");
assert_eq!(summary.distinct, 1000);
assert!(!summary.unique, "every value appears twice");
assert_eq!(summary.order, Order::Descending);
assert_eq!(summary.runs, 1000, "a descending column is a run per distinct value");
fs::remove_file(&path).expect("clean up");
}
#[test]
fn a_column_past_the_default_k_is_estimated_and_says_so() {
let values = (0..20_000_i64).map(Some).collect::<Vec<_>>();
let path = table_of("estimated", &values);
let built = build_stats(&path, "t", &[0]).expect("build");
assert!(!built[0].exact, "twenty thousand values is past the default k");
let reader = reopen(&path);
let summary = summary(&reader, 0).expect("the summary");
assert_eq!(summary.distinct_class, Class::Estimated);
assert!(!summary.unique, "uniqueness is never claimed off an estimate");
assert!(summary.distinct > 17_000 && summary.distinct <= 20_000, "{}", summary.distinct);
assert!(summary.distinct <= summary.present(), "more distinct values than rows");
fs::remove_file(&path).expect("clean up");
}
#[test]
fn a_shuffled_column_is_neither_ordered_nor_one_run() {
let values = (0..2000_i64).map(|at| Some((at * 7919) % 2000)).collect::<Vec<_>>();
let path = table_of("shuffled", &values);
build_stats(&path, "t", &[0]).expect("build");
let reader = reopen(&path);
let summary = summary(&reader, 0).expect("the summary");
assert_eq!(summary.order, Order::Neither);
assert!(summary.runs > 100, "a shuffle is many runs, not one: {}", summary.runs);
assert_eq!(summary.low, Some(Bound::Int(0)));
assert_eq!(summary.high, Some(Bound::Int(1999)));
fs::remove_file(&path).expect("clean up");
}
#[test]
fn a_column_something_declared_a_key_over_is_sketched_per_stripe_and_a_plain_one_is_not() {
let values = (1..=19_200_i64).map(Some).collect::<Vec<_>>();
let path = table_of_parts("promoted", &values, 100);
let plain = build_stats(&path, "t", &[0]).expect("build");
assert_eq!(plain[0].stripes, 0, "nothing has declared anything over this column yet");
crate::graph::build_key_maps(&path, "t", &[0]).expect("a key map declares it a key");
let promoted = build_stats(&path, "t", &[0]).expect("rebuild");
assert!(promoted[0].stripes > 1, "{} stripes, wanted more than one", promoted[0].stripes);
assert!(promoted[0].built, "and they fit");
assert_eq!(promoted[0].distinct, plain[0].distinct, "the merged count did not move");
let reader = reopen(&path);
let sketches = sketches(&reader, 0).expect("the sketches came back");
assert_eq!(sketches.stripes.len(), promoted[0].stripes);
assert!(
sketches.stripes.iter().all(|stripe| stripe.k() == STRIPE_K),
"a stripe sketch is written down at the smaller k"
);
let floor = sketches.floor(0, sketches.stripes.len()).expect("a floor over every stripe");
let actual = 19_200.0;
assert!(
(floor - actual).abs() / actual < 0.25,
"{floor:.0} over every stripe against {actual:.0}"
);
drop(reader);
fs::remove_file(&path).expect("clean up");
}
#[test]
fn a_file_from_before_the_section_table_opens_and_every_statistic_is_unknown() {
let months = (1..=3000_i32).collect::<Vec<_>>();
let older = table_of_intervals("before_sections", &months);
let current = table_of("with_sections", &(1..=3000_i64).map(Some).collect::<Vec<_>>());
let file = fs::OpenOptions::new().write(true).open(&older).expect("reopen to patch");
crate::write_at(&file, 8, &22_u32.to_le_bytes()).expect("stamp the older format");
drop(file);
let new = reopen(¤t);
assert!(summary(&new, 0).is_some(), "the file this build wrote says what it holds");
let old = reopen(&older);
assert!(old.table().sections().is_empty(), "an older file names no sections");
assert!(summary(&old, 0).is_none(), "and so says nothing about its columns");
assert!(sketches(&old, 0).is_none());
assert!(read_columns(&old).is_empty(), "nor promotes any of them");
assert_eq!(old.table().rows(), 3000, "and reads every row it holds");
assert_eq!(
rows_of(&old).first(),
Some(&Value::Interval { months: 1, days: 0, micros: 0 }),
"with the values it was written with"
);
drop(new);
drop(old);
fs::remove_file(¤t).expect("clean up");
fs::remove_file(&older).expect("clean up");
}
#[test]
fn the_stripe_ends_say_whether_a_scan_can_skip_and_a_shuffle_says_it_cannot() {
let sorted = (1..=19_200_i64).map(Some).collect::<Vec<_>>();
let ordered = table_of_parts("stripes_sorted", &sorted, 100);
build_stats(&ordered, "t", &[0]).expect("build");
let reader = reopen(&ordered);
let ordered_summary = summary(&reader, 0).expect("the summary");
assert!(!ordered_summary.overlapping, "a sorted column's stripes are disjoint");
assert_eq!(ordered_summary.low, Some(Bound::Int(1)));
assert_eq!(ordered_summary.high, Some(Bound::Int(19_200)));
drop(reader);
let shuffled = (0..19_200_i64).map(|at| Some(1 + at * 7919 % 19_200)).collect::<Vec<_>>();
let mixed = table_of_parts("stripes_shuffled", &shuffled, 100);
build_stats(&mixed, "t", &[0]).expect("build");
let reader = reopen(&mixed);
let mixed_summary = summary(&reader, 0).expect("the summary");
assert!(mixed_summary.overlapping, "a shuffled column's stripes all span it");
assert_eq!(mixed_summary.low, Some(Bound::Int(1)), "the same values in a different order");
assert_eq!(mixed_summary.high, Some(Bound::Int(19_200)));
drop(reader);
fs::remove_file(&ordered).expect("clean up");
fs::remove_file(&mixed).expect("clean up");
}
#[test]
fn the_graph_sections_do_not_count_against_the_statistics_budget() {
let values = (1..=3000_i64).map(Some).collect::<Vec<_>>();
let path = table_of("apart", &values);
crate::graph::build_key_maps(&path, "t", &[0]).expect("a key map first");
let reader = reopen(&path);
let graph = reader
.table()
.sections()
.iter()
.filter(|held| held.among(section::GRAPH_KINDS))
.count();
assert_eq!(graph, 1, "the key map is in the file");
assert_eq!(held_bytes(&reader, &[0]).expect("held"), 0, "and it is not the statistics'");
drop(reader);
fs::remove_file(&path).expect("clean up");
}
#[test]
fn deleting_the_sections_changes_nothing_but_whether_they_are_there() {
let values = (1..=1500_i64).map(Some).collect::<Vec<_>>();
let path = table_of("invariant", &values);
build_stats(&path, "t", &[0]).expect("build");
let reader = reopen(&path);
assert!(summary(&reader, 0).is_some());
let generation = reader.table().generation();
let held: Vec<_> = reader
.table()
.sections()
.iter()
.filter(|s| s.kind == *section::SUMMARY || s.kind == *section::SKETCHES)
.copied()
.collect();
assert_eq!(held.len(), 2, "a summary and a sketch section");
for section in &held {
assert!(section.usable(generation));
assert!(!section.usable(generation + 1), "a rewrite invalidates rather than corrupts");
}
let rows: usize =
(0..reader.parts()).map(|part| reader.read(part, &[0]).expect("a part").len()).sum();
assert_eq!(rows, 1500, "the scan is the scan whether the sections are read or not");
fs::remove_file(&path).expect("clean up");
}
#[test]
fn a_string_column_is_read_through_the_typed_path_and_measured_by_its_bytes() {
let path = path("strings");
let mut writer =
Writer::create(&path, "t", vec![Field::new("v", LogicalType::Varchar)]).expect("new");
let words = ["alpha", "bravo", "charlie", "delta", "alpha"];
let held = words.iter().map(|w| Value::Varchar((*w).into())).collect::<Vec<_>>();
let chunk =
Chunk::new(vec![Vector::from_values(LogicalType::Varchar, &held).expect("words")])
.expect("one column");
writer.append(&chunk).expect("a part");
writer.finish().expect("commit");
build_stats(&path, "t", &[0]).expect("build");
let reader = reopen(&path);
let summary = summary(&reader, 0).expect("the summary");
assert_eq!(summary.rows, 5);
assert_eq!(summary.distinct, 4, "alpha twice");
assert!(!summary.unique);
assert_eq!(summary.bytes, words.iter().map(|w| w.len() as u64).sum::<u64>());
assert_eq!(summary.widest, 7, "charlie");
assert_eq!(summary.low, Some(Bound::Bytes(b"alpha".to_vec())));
assert_eq!(summary.high, Some(Bound::Bytes(b"delta".to_vec())));
drop(reader);
fs::remove_file(&path).expect("clean up");
}
#[test]
fn a_type_with_no_hash_rule_is_refused_by_name_rather_than_summarized_as_empty() {
let path = table_of("refused", &[Some(1)]);
let reader = reopen(&path);
assert!(summarizable(&LogicalType::BigInt));
assert!(!summarizable(&LogicalType::Interval));
assert!(build_summary(&reader, 1).is_err(), "a column past the end");
drop(reader);
fs::remove_file(&path).expect("clean up");
}
#[test]
fn the_stored_sketch_depends_on_the_value_rule_and_not_only_on_the_hash() {
assert_eq!(hash_value(&Value::BigInt(1)), Some(hash64(&1_u128.to_le_bytes())));
assert_eq!(hash_value(&Value::Integer(1)), hash_value(&Value::BigInt(1)));
assert_eq!(hash_value(&Value::Varchar("a".into())), Some(hash64(b"a")));
assert_eq!(hash_value(&Value::Null), None);
}
}