use std::collections::BTreeMap;
use crate::lpbound::ProductBound;
use crate::lpbound::UpperBound;
pub type AttributeId = u32;
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct AttributeDegree {
max_degree: u64,
}
impl AttributeDegree {
pub const fn unknown(rows: u64) -> Self {
Self { max_degree: rows }
}
pub const fn from_distinct(rows: u64, distinct: u64) -> Self {
if distinct == 0 || distinct > rows {
return Self::unknown(rows);
}
Self {
max_degree: rows - distinct + 1,
}
}
pub const fn from_upper_bound(rows: u64, upper_bound: u64) -> Self {
Self {
max_degree: if upper_bound < rows {
upper_bound
} else {
rows
},
}
}
pub fn from_hll_floor(rows: u64, sketch: &crate::sketches::HllSketch) -> Self {
Self::from_distinct(rows, sketch.nonzero_registers())
}
pub fn from_count_min(rows: u64, sketch: &crate::sketches::CountMinSketch) -> Self {
match sketch.max_frequency_bound() {
Some(bound) => Self::from_upper_bound(rows, u64::from(bound)),
None => Self::unknown(rows),
}
}
pub const fn max_degree(&self) -> u64 {
self.max_degree
}
}
#[derive(Debug, Clone)]
pub struct JoinRelation {
rows: u64,
degrees: BTreeMap<AttributeId, AttributeDegree>,
}
impl JoinRelation {
pub fn new(rows: u64) -> Self {
Self {
rows,
degrees: BTreeMap::new(),
}
}
pub fn with_degree(mut self, attribute: AttributeId, degree: AttributeDegree) -> Self {
self.degrees.insert(attribute, degree);
self
}
pub const fn rows(&self) -> u64 {
self.rows
}
pub fn degree(&self, attribute: AttributeId) -> AttributeDegree {
self.degrees
.get(&attribute)
.copied()
.unwrap_or_else(|| AttributeDegree::unknown(self.rows))
}
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct JoinEdge {
pub left: usize,
pub right: usize,
pub attribute: AttributeId,
}
#[derive(Debug, Clone, Default)]
pub struct JoinGraph {
relations: Vec<JoinRelation>,
edges: Vec<JoinEdge>,
}
impl JoinGraph {
pub fn new(relations: Vec<JoinRelation>) -> Self {
Self {
relations,
edges: Vec::new(),
}
}
pub fn with_edge(mut self, left: usize, right: usize, attribute: AttributeId) -> Self {
let n = self.relations.len();
if left < n && right < n && left != right {
self.edges.push(JoinEdge {
left,
right,
attribute,
});
}
self
}
pub fn relations(&self) -> &[JoinRelation] {
&self.relations
}
pub fn edges(&self) -> &[JoinEdge] {
&self.edges
}
pub fn ceiling(&self) -> u64 {
if self.relations.is_empty() {
return 0;
}
let mut total: u128 = 1;
for component in self.components() {
let component_ceiling = self.component_ceiling(&component);
total = total.saturating_mul(u128::from(component_ceiling));
if total >= u128::from(u64::MAX) {
return u64::MAX;
}
}
total as u64
}
fn components(&self) -> Vec<Vec<usize>> {
let n = self.relations.len();
let mut parent: Vec<usize> = (0..n).collect();
fn find(parent: &mut [usize], mut x: usize) -> usize {
while parent[x] != x {
parent[x] = parent[parent[x]];
x = parent[x];
}
x
}
for edge in &self.edges {
let a = find(&mut parent, edge.left);
let b = find(&mut parent, edge.right);
if a != b {
parent[a] = b;
}
}
let mut groups: BTreeMap<usize, Vec<usize>> = BTreeMap::new();
for v in 0..n {
let root = find(&mut parent, v);
groups.entry(root).or_default().push(v);
}
groups.into_values().collect()
}
fn component_ceiling(&self, component: &[usize]) -> u64 {
let rows: Vec<u64> = component.iter().map(|&r| self.relations[r].rows).collect();
let mut best = ProductBound.ceiling(&rows, &[]);
if component.len() == 1 {
return self.relations[component[0]].rows;
}
for &root in component {
let candidate = self.spanning_tree_ceiling(component, root);
if candidate < best {
best = candidate;
}
}
best
}
fn spanning_tree_ceiling(&self, component: &[usize], root: usize) -> u64 {
let mut visited: Vec<usize> = vec![root];
let mut bound: u128 = u128::from(self.relations[root].rows);
while visited.len() < component.len() {
let mut best: Option<(usize, u64)> = None;
for edge in &self.edges {
for (from, to) in [(edge.left, edge.right), (edge.right, edge.left)] {
if !visited.contains(&from) || visited.contains(&to) {
continue;
}
if !component.contains(&to) {
continue;
}
let factor = self.relations[to].degree(edge.attribute).max_degree();
if best.is_none_or(|(_, current)| factor < current) {
best = Some((to, factor));
}
}
}
let Some((next, factor)) = best else {
for &v in component {
if !visited.contains(&v) {
bound = bound.saturating_mul(u128::from(self.relations[v].rows));
visited.push(v);
}
}
break;
};
bound = bound.saturating_mul(u128::from(factor));
visited.push(next);
if bound >= u128::from(u64::MAX) {
return u64::MAX;
}
}
if bound >= u128::from(u64::MAX) {
u64::MAX
} else {
bound as u64
}
}
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn fk_join_is_exactly_tight() {
let orders = JoinRelation::new(10).with_degree(0, AttributeDegree::from_distinct(10, 10));
let lineitem =
JoinRelation::new(100).with_degree(0, AttributeDegree::from_distinct(100, 10));
let g = JoinGraph::new(vec![orders, lineitem]).with_edge(0, 1, 0);
assert_eq!(g.ceiling(), 100);
}
#[test]
fn all_rows_on_one_key_yields_the_product() {
let g = JoinGraph::new(vec![JoinRelation::new(4), JoinRelation::new(5)]).with_edge(0, 1, 0);
assert_eq!(g.ceiling(), 20);
}
#[test]
fn skewed_join_stays_above_truth() {
let rel = || JoinRelation::new(20).with_degree(0, AttributeDegree::from_distinct(20, 5));
let g = JoinGraph::new(vec![rel(), rel()]).with_edge(0, 1, 0);
assert_eq!(g.ceiling(), 320);
assert!(g.ceiling() >= 260);
}
#[test]
fn star_with_key_hub_is_tight() {
let hub = JoinRelation::new(2);
let spoke = || JoinRelation::new(4);
let g = JoinGraph::new(vec![hub, spoke(), spoke(), spoke()])
.with_edge(0, 1, 0)
.with_edge(0, 2, 1)
.with_edge(0, 3, 2);
assert_eq!(g.ceiling(), 128);
}
#[test]
fn key_star_collapses_to_the_hub() {
let hub = JoinRelation::new(1_000);
let dim = |n| JoinRelation::new(n).with_degree(0, AttributeDegree::from_distinct(n, n));
let g = JoinGraph::new(vec![hub, dim(50), dim(60), dim(70)])
.with_edge(0, 1, 0)
.with_edge(0, 2, 0)
.with_edge(0, 3, 0);
assert_eq!(g.ceiling(), 1_000);
}
#[test]
fn disconnected_components_multiply() {
let g = JoinGraph::new(vec![
JoinRelation::new(3),
JoinRelation::new(4),
JoinRelation::new(5),
])
.with_edge(0, 1, 0);
assert_eq!(g.ceiling(), 60);
}
#[test]
fn empty_graph_is_zero() {
assert_eq!(JoinGraph::new(Vec::new()).ceiling(), 0);
}
#[test]
fn single_relation_is_its_row_count() {
assert_eq!(JoinGraph::new(vec![JoinRelation::new(77)]).ceiling(), 77);
}
#[test]
fn ceiling_never_exceeds_the_product() {
let rel = |n| JoinRelation::new(n).with_degree(0, AttributeDegree::from_distinct(n, n));
let g = JoinGraph::new(vec![rel(10), rel(20), rel(30)])
.with_edge(0, 1, 0)
.with_edge(1, 2, 0);
assert!(g.ceiling() <= 10 * 20 * 30);
}
#[test]
fn saturates_instead_of_overflowing() {
let huge = || JoinRelation::new(u64::MAX);
let g = JoinGraph::new(vec![huge(), huge(), huge()])
.with_edge(0, 1, 0)
.with_edge(1, 2, 0);
assert_eq!(g.ceiling(), u64::MAX);
}
#[test]
fn degree_from_distinct_rejects_inconsistent_input() {
assert_eq!(AttributeDegree::from_distinct(10, 50).max_degree(), 10);
}
#[test]
fn unknown_degrees_degrade_to_the_product() {
let g = JoinGraph::new(vec![JoinRelation::new(6), JoinRelation::new(7)]).with_edge(0, 1, 0);
assert_eq!(g.ceiling(), 42);
}
}