From 3edf8f6be1fc912f169e8abc644045287a6f4a2f Mon Sep 17 00:00:00 2001 From: zz_y Date: Tue, 6 Oct 2026 18:43:14 +0000 Subject: [PATCH 1/3] atomic-costs: record each row's measurement conditions (measured_at) Each AtomicCostEntry now carries the conditions it was measured under, read off the record's workload: items per instance, keys per instance, the value column's range and distribution, and the merge operand size. Optional on read so tables written before it still load. Part of #147. Co-Authored-By: Claude Opus 5.5 --- aqpbm-core/src/atomic_costs.rs | 135 ++++++++++++++++++++++++++++++-- aqpbm-core/src/lib.rs | 4 +- rqe-optimizer/src/autosketch.rs | 2 + rqe-optimizer/src/candidates.rs | 1 + rqe-optimizer/src/enumerate.rs | 1 + rqe-optimizer/src/lib.rs | 1 + 6 files changed, 137 insertions(+), 7 deletions(-) diff --git a/aqpbm-core/src/atomic_costs.rs b/aqpbm-core/src/atomic_costs.rs index 8e799ed2..aa7f1701 100644 --- a/aqpbm-core/src/atomic_costs.rs +++ b/aqpbm-core/src/atomic_costs.rs @@ -11,8 +11,10 @@ use std::collections::BTreeMap; use serde::{Deserialize, Serialize}; +use aqpbm_datagen::{DataDistribution, TableDescription}; + use crate::accuracy::aggregate::GROUPS_PER_INSTANCE; -use crate::benchmark_result::{CpuTime, MergedRecord, RunStats}; +use crate::benchmark_result::{CpuTime, MergedRecord, RunStats, WorkloadDescription}; /// One row: measured atomic costs for one (sketch algorithm, construction /// params) point, at the grid resolution ASAPQuery's `candidate_gen.rs` @@ -34,6 +36,31 @@ pub struct AtomicCostEntry { /// rank-error's `mean_rank_err`). No single scalar covers all of them, so /// this stays a map rather than picking one field to promote. pub query_accuracy: BTreeMap, + /// The conditions the row was measured under. Optional so tables written + /// before it was added still load; [`reduce_one`] always fills it. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub measured_at: Option, +} + +/// What one benchmark instance saw. A consumer applying the row to an instance +/// of a different size, key count or value range can tell how far it is +/// extrapolating (sketch-bench#147). Read off the record's workload, so a +/// field the workload doesn't determine is `None`. +#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)] +pub struct MeasuredAt { + /// Items (rows) inserted into the instance. + pub items_per_instance: u64, + /// The measured group count for a grouped accumulator; otherwise the + /// product of the key columns' populations, an upper bound on the distinct + /// keys actually drawn. + pub keys_per_instance: Option, + /// `[lower, upper]` of the value column's distribution, plus its shift. + pub value_range: Option<[f64; 2]>, + /// Items in each sketch the merge benchmark folded in: + /// `items_per_instance / merge_shards`. + pub merge_operand_items: Option, + /// The value column's distribution as generated, e.g. Zipf with its skew. + pub distribution: Option, } pub type AtomicCostTable = Vec; @@ -169,11 +196,8 @@ pub fn reduce_one(record: &MergedRecord) -> Result // Priced per group here, so the optimizer's `card(group-by) ×` counts each // group once. Query is already per group (one probe per group); insert is // per item. - let groups = query_accuracy - .get(GROUPS_PER_INSTANCE) - .copied() - .unwrap_or(1.0) - .max(1.0); + let groups_measured = query_accuracy.get(GROUPS_PER_INSTANCE).copied(); + let groups = groups_measured.unwrap_or(1.0).max(1.0); let mem_bytes_per_instance = mem_bytes_per_instance / groups; let merge_cpu_secs = merge_cpu_secs / groups; @@ -187,10 +211,51 @@ pub fn reduce_one(record: &MergedRecord) -> Result insert_cpu_secs, merge_cpu_secs, query_cpu_secs, + measured_at: Some(measured_at(record, groups_measured)), query_accuracy, }) } +/// The record's measurement conditions. In a table description the last +/// column holds the values and the ones before it are the keys; a +/// single-column table's one column is both. +fn measured_at(record: &MergedRecord, groups_measured: Option) -> MeasuredAt { + let (items, columns) = match &record.input_dataset { + WorkloadDescription::Synthetic { description, .. } => { + (description.row_num, Some(description)) + } + WorkloadDescription::External(external) => (external.records_loaded, None), + }; + let value = columns.and_then(|d: &TableDescription| d.column_spec.last()); + let keys = columns.and_then(|d| match d.column_spec.split_last() { + Some((value, [])) => value.distribution.domain().map(|d| d.size), + Some((_, labels)) => labels + .iter() + .map(|c| c.distribution.domain().map(|d| d.size)) + .product(), + None => None, + }); + MeasuredAt { + items_per_instance: items, + keys_per_instance: groups_measured.map(|g| g as u64).or(keys), + value_range: value.and_then(|c| { + let shift = c.shift.unwrap_or(0.0); + match &c.distribution { + DataDistribution::Zipf(p) => Some([1.0, p.population_size as f64]), + DataDistribution::Uniform(p) => Some([p.lower_bound, p.upper_bound]), + DataDistribution::Normal(_) | DataDistribution::Pareto(_) => None, + } + .map(|[lo, hi]| [lo + shift, hi + shift]) + }), + merge_operand_items: record + .merge + .merge_shards + .filter(|&m| m > 0) + .map(|m| items / m as u64), + distribution: value.map(|c| c.distribution.clone()), + } +} + /// `MergedRecord::query.accuracy` down to the flat numeric map every /// comparator's `GroundTruth::score` actually produces. Required like the /// cost fields (missing skips the row, `SkipReason` says which): an entry @@ -336,6 +401,63 @@ mod tests { assert!((entry.query_cpu_secs - 4e-6).abs() < 1e-12); } + /// A one-column table: its column gives the keys and the values, and the + /// merge operand is the stream split `merge_shards` ways. + #[test] + fn measured_at_reads_a_single_column_workload() { + let mut record = full_record(); + record.merge.merge_shards = Some(16); + let entry = reduce_one(&record).expect("fully populated record"); + let at = entry.measured_at.expect("always filled"); + assert_eq!(at.items_per_instance, 1_000_000); + assert_eq!(at.keys_per_instance, Some(100_000)); + assert_eq!(at.value_range, Some([0.0, 100_000.0])); + assert_eq!(at.merge_operand_items, Some(62_500)); + assert_eq!( + at.distribution, + Some(dataset().column_spec[0].distribution.clone()) + ); + } + + /// A multi-column table: the last column is the value, the rest are keys. + /// A measured group count wins over the key populations' bound. + #[test] + fn measured_at_reads_a_labelled_workload() { + let zipf = |population_size| aqpbm_datagen::ColumnSpec { + distribution: aqpbm_datagen::DataDistribution::Zipf(aqpbm_datagen::ZipfParameter { + skewness: 1.1, + population_size, + seed: 1, + }), + ..dataset().column_spec[0].clone() + }; + let mut description = dataset(); + description.column_num = 3; + description.column_label = vec!["a".into(), "b".into(), "value".into()]; + description.column_spec = vec![zipf(50), zipf(1000), dataset().column_spec[0].clone()]; + description.column_spec[2].shift = Some(10.0); + let mut record = full_record(); + record.input_dataset = description.into(); + + let at = reduce_one(&record).unwrap().measured_at.unwrap(); + assert_eq!(at.keys_per_instance, Some(50_000)); + assert_eq!(at.value_range, Some([10.0, 100_010.0])); + assert_eq!(at.merge_operand_items, None); + + record.query.accuracy = + Some(serde_json::json!({"relative_error": 0.0, "groups_per_instance": 9_876.0})); + let at = reduce_one(&record).unwrap().measured_at.unwrap(); + assert_eq!(at.keys_per_instance, Some(9_876)); + } + + /// Tables written before `measured_at` existed still load. + #[test] + fn a_row_without_measured_at_still_loads() { + let json = r#"{"sketch":"cms","sketch_config":null,"mem_bytes_per_instance":1.0,"insert_cpu_secs":1.0,"merge_cpu_secs":1.0,"query_cpu_secs":1.0,"query_accuracy":{}}"#; + let entry: AtomicCostEntry = serde_json::from_str(json).unwrap(); + assert_eq!(entry.measured_at, None); + } + #[test] fn reduces_a_full_record_to_per_op_seconds() { let entry = reduce_one(&full_record()).expect("fully populated record"); @@ -454,6 +576,7 @@ mod tests { merge_cpu_secs: 1e-2, query_cpu_secs: 4e-6, query_accuracy: BTreeMap::from([("relative_error_mean".to_string(), 0.01)]), + measured_at: None, }; let json = serde_json::to_string(&entry).unwrap(); assert_eq!( diff --git a/aqpbm-core/src/lib.rs b/aqpbm-core/src/lib.rs index da048d69..67e72d73 100644 --- a/aqpbm-core/src/lib.rs +++ b/aqpbm-core/src/lib.rs @@ -19,7 +19,9 @@ pub use aqpbm_datagen::{ NormalParameter, ParetoParameter, StringOpts, TableDescription, UniformParameter, ZipfParameter, RULE_MONOTONIC_INCREASE, RULE_NONE, }; -pub use atomic_costs::{reduce_all, reduce_one, AtomicCostEntry, AtomicCostTable, SkipReason}; +pub use atomic_costs::{ + reduce_all, reduce_one, AtomicCostEntry, AtomicCostTable, MeasuredAt, SkipReason, +}; pub use benchmark_result::{ BenchReport, BenchSection, CpuTime, ExternalWorkload, InsertMetrics, LatencySummary, MergeMetrics, MergedRecord, Mode, PrepareMetrics, QueryMetrics, Record, RunStats, Source, diff --git a/rqe-optimizer/src/autosketch.rs b/rqe-optimizer/src/autosketch.rs index 72fb4e6d..d67482a1 100644 --- a/rqe-optimizer/src/autosketch.rs +++ b/rqe-optimizer/src/autosketch.rs @@ -354,6 +354,7 @@ mod tests { query_cpu_secs: 1e-6, // Error falls with width and depth, so feasibility is monotone. query_accuracy: BTreeMap::from([(ERR.into(), 1.0 / (rows * cols) as f64)]), + measured_at: None, } } @@ -488,6 +489,7 @@ mod tests { merge_cpu_secs: 1e-6, query_cpu_secs: 1e-6, query_accuracy: BTreeMap::from([(ERR.into(), 1.0 / k as f64)]), + measured_at: None, } } diff --git a/rqe-optimizer/src/candidates.rs b/rqe-optimizer/src/candidates.rs index 12d58698..8f4ab736 100644 --- a/rqe-optimizer/src/candidates.rs +++ b/rqe-optimizer/src/candidates.rs @@ -351,6 +351,7 @@ mod tests { merge_cpu_secs: 1.0, query_cpu_secs: 1.0, query_accuracy: BTreeMap::from([("err".into(), 0.1)]), + measured_at: None, } } #[test] diff --git a/rqe-optimizer/src/enumerate.rs b/rqe-optimizer/src/enumerate.rs index abcaf7cd..ba9e5f3c 100644 --- a/rqe-optimizer/src/enumerate.rs +++ b/rqe-optimizer/src/enumerate.rs @@ -126,6 +126,7 @@ mod tests { merge_cpu_secs: 1.0, query_cpu_secs: 1.0, query_accuracy, + measured_at: None, } } diff --git a/rqe-optimizer/src/lib.rs b/rqe-optimizer/src/lib.rs index 912c3181..52e8cd65 100644 --- a/rqe-optimizer/src/lib.rs +++ b/rqe-optimizer/src/lib.rs @@ -385,6 +385,7 @@ pub(crate) mod test_support { merge_cpu_secs: merge, query_cpu_secs: query, query_accuracy: BTreeMap::from([("err".into(), 0.0)]), + measured_at: None, }, window_ms, slide_ms, From af5545d319de2358e281122c7d001a8496ab5890 Mon Sep 17 00:00:00 2001 From: zz_y Date: Tue, 6 Oct 2026 19:07:24 +0000 Subject: [PATCH 2/3] measured_at: no value range for a monotonic (counter) column Its values are a running sum of the draws, so the draws' bounds don't bound them. Co-Authored-By: Claude Opus 5.5 --- aqpbm-core/src/atomic_costs.rs | 37 +++++++++++++++++++++++++--------- 1 file changed, 27 insertions(+), 10 deletions(-) diff --git a/aqpbm-core/src/atomic_costs.rs b/aqpbm-core/src/atomic_costs.rs index aa7f1701..86dd8106 100644 --- a/aqpbm-core/src/atomic_costs.rs +++ b/aqpbm-core/src/atomic_costs.rs @@ -11,7 +11,7 @@ use std::collections::BTreeMap; use serde::{Deserialize, Serialize}; -use aqpbm_datagen::{DataDistribution, TableDescription}; +use aqpbm_datagen::{DataDistribution, TableDescription, RULE_MONOTONIC_INCREASE}; use crate::accuracy::aggregate::GROUPS_PER_INSTANCE; use crate::benchmark_result::{CpuTime, MergedRecord, RunStats, WorkloadDescription}; @@ -55,6 +55,8 @@ pub struct MeasuredAt { /// keys actually drawn. pub keys_per_instance: Option, /// `[lower, upper]` of the value column's distribution, plus its shift. + /// `None` for a monotonic (counter) column, whose values are a running sum + /// of draws rather than the draws themselves. pub value_range: Option<[f64; 2]>, /// Items in each sketch the merge benchmark folded in: /// `items_per_instance / merge_shards`. @@ -238,15 +240,17 @@ fn measured_at(record: &MergedRecord, groups_measured: Option) -> MeasuredA MeasuredAt { items_per_instance: items, keys_per_instance: groups_measured.map(|g| g as u64).or(keys), - value_range: value.and_then(|c| { - let shift = c.shift.unwrap_or(0.0); - match &c.distribution { - DataDistribution::Zipf(p) => Some([1.0, p.population_size as f64]), - DataDistribution::Uniform(p) => Some([p.lower_bound, p.upper_bound]), - DataDistribution::Normal(_) | DataDistribution::Pareto(_) => None, - } - .map(|[lo, hi]| [lo + shift, hi + shift]) - }), + value_range: value + .filter(|c| c.special_rule & RULE_MONOTONIC_INCREASE == 0) + .and_then(|c| { + let shift = c.shift.unwrap_or(0.0); + match &c.distribution { + DataDistribution::Zipf(p) => Some([1.0, p.population_size as f64]), + DataDistribution::Uniform(p) => Some([p.lower_bound, p.upper_bound]), + DataDistribution::Normal(_) | DataDistribution::Pareto(_) => None, + } + .map(|[lo, hi]| [lo + shift, hi + shift]) + }), merge_operand_items: record .merge .merge_shards @@ -450,6 +454,19 @@ mod tests { assert_eq!(at.keys_per_instance, Some(9_876)); } + /// A counter column emits a running sum of its draws, so the draws' + /// bounds say nothing about the values. + #[test] + fn measured_at_has_no_value_range_for_a_counter_column() { + let mut description = dataset(); + description.column_spec[0].special_rule = aqpbm_datagen::RULE_MONOTONIC_INCREASE; + let mut record = full_record(); + record.input_dataset = description.into(); + let at = reduce_one(&record).unwrap().measured_at.unwrap(); + assert_eq!(at.value_range, None); + assert_eq!(at.items_per_instance, 1_000_000); + } + /// Tables written before `measured_at` existed still load. #[test] fn a_row_without_measured_at_still_loads() { From c6c0bf91f6b7bf7c27a0ced7b2c678fed81c6ba4 Mon Sep 17 00:00:00 2001 From: zz_y Date: Tue, 6 Oct 2026 20:27:58 +0000 Subject: [PATCH 3/3] measured_at: saturate the key-population product Co-Authored-By: Claude Opus 5.5 --- aqpbm-core/src/atomic_costs.rs | 8 ++++---- 1 file changed, 4 insertions(+), 4 deletions(-) diff --git a/aqpbm-core/src/atomic_costs.rs b/aqpbm-core/src/atomic_costs.rs index 86dd8106..6164c23a 100644 --- a/aqpbm-core/src/atomic_costs.rs +++ b/aqpbm-core/src/atomic_costs.rs @@ -231,10 +231,10 @@ fn measured_at(record: &MergedRecord, groups_measured: Option) -> MeasuredA let value = columns.and_then(|d: &TableDescription| d.column_spec.last()); let keys = columns.and_then(|d| match d.column_spec.split_last() { Some((value, [])) => value.distribution.domain().map(|d| d.size), - Some((_, labels)) => labels - .iter() - .map(|c| c.distribution.domain().map(|d| d.size)) - .product(), + // Saturating: wide label domains multiply past u64; the bound stays a bound. + Some((_, labels)) => labels.iter().try_fold(1u64, |keys, c| { + Some(keys.saturating_mul(c.distribution.domain()?.size)) + }), None => None, }); MeasuredAt {