Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
3 changes: 1 addition & 2 deletions aqpbm-cli/src/atomic_costs_cmd.rs
Original file line number Diff line number Diff line change
Expand Up @@ -29,8 +29,7 @@ pub fn run(args: AtomicCostsArgs) -> Result<()> {
.context("reading stdin")?;
buf
} else {
std::fs::read_to_string(&args.input)
.with_context(|| format!("reading {}", args.input))?
std::fs::read_to_string(&args.input).with_context(|| format!("reading {}", args.input))?
};

let records: Vec<MergedRecord> = raw
Expand Down
3 changes: 2 additions & 1 deletion aqpbm-cli/src/cli.rs
Original file line number Diff line number Diff line change
Expand Up @@ -22,6 +22,7 @@ pub struct Cli {
}

#[derive(Subcommand, Debug)]
#[allow(clippy::large_enum_variant)]
pub enum Cmd {
/// Measure one target of the sketch bundle.
Sketchbench(SketchbenchArgs),
Expand Down Expand Up @@ -95,7 +96,7 @@ pub struct SketchbenchArgs {
/// Item type the value column is generated at: `i64`, `u64`, `f64` or
/// `string` — every type the generator renders. The one item-type choice
/// left, since a row's own wrapper fixes what it can ingest and refuses the
/// rest by name; today the ordered rows (`kll-percall`, `kll-cdf`) take
/// rest by name; today the ordered rows (`kll-percall`, `kll-cdf`, `dd`) take
/// either numeric width and every other row takes `i64`. Encoding only.
#[arg(long, default_value = "i64", help_heading = "Dataset")]
pub dtype: String,
Expand Down
71 changes: 71 additions & 0 deletions aqpbm-cli/src/rows.rs
Original file line number Diff line number Diff line change
Expand Up @@ -36,6 +36,7 @@ use aqpbm_core::{ColumnItem, GeneratedTable, TableDescription};

use sketch_bench::wrappers::cms::{datasketches as cd, oxide as co, polars as cp, sketchlib as cl};
use sketch_bench::wrappers::cs::{oxide as so, polars as sp, sketchlib as sl};
use sketch_bench::wrappers::dd::{oxide as ddo, sketchlib as ddl};
use sketch_bench::wrappers::hll::{
datasketches as hd, oxide as ho, polars as hpo, sketchlib as hl,
};
Expand Down Expand Up @@ -122,6 +123,8 @@ fn binding(algorithm: &str, impl_name: &str) -> Option<RowBinding> {
("kll-cdf", "oxide") => row_kll_cdf_oxide,
("kll-cdf", "lib") => row_kll_cdf_lib,
("kll-cdf", "polars") => row_kll_cdf_polars,
("dd", "lib") => row_dd_lib,
("dd", "oxide") => row_dd_oxide,
("hydra-cms", "lib") => row_hydra_cms_lib,
("hydra-cms", "polars") => row_hydra_cms_polars,
("hydra-hll", "lib") => row_hydra_hll_lib,
Expand Down Expand Up @@ -613,6 +616,74 @@ pub(crate) fn row_kll_cdf_polars(
)
}

// -------- DDSketch (quantile) --------

pub(crate) fn row_dd_lib(
req: &Requirement,
description: &TableDescription,
table: GeneratedTable,
want: &[(Operation, Metric)],
) -> Result<Measurements, RunError> {
match req.width {
Dtype::I64 => quantile_row::<i64>(
req,
description,
table,
want,
ddl::insert_dd_lib::<i64>,
ddl::insert_step_dd_lib::<i64>,
ddl::query_dd_lib::<i64>,
Some((ddl::merge_dd_lib::<i64>, ddl::merge_step_dd_lib::<i64>)),
None,
),
Dtype::F64 => quantile_row::<f64>(
req,
description,
table,
want,
ddl::insert_dd_lib::<f64>,
ddl::insert_step_dd_lib::<f64>,
ddl::query_dd_lib::<f64>,
Some((ddl::merge_dd_lib::<f64>, ddl::merge_step_dd_lib::<f64>)),
None,
),
other => Err(no_build_at(req, other)),
}
}

pub(crate) fn row_dd_oxide(
req: &Requirement,
description: &TableDescription,
table: GeneratedTable,
want: &[(Operation, Metric)],
) -> Result<Measurements, RunError> {
match req.width {
Dtype::I64 => quantile_row::<i64>(
req,
description,
table,
want,
ddo::insert_dd_oxide::<i64>,
ddo::insert_step_dd_oxide::<i64>,
ddo::query_dd_oxide::<i64>,
Some((ddo::merge_dd_oxide::<i64>, ddo::merge_step_dd_oxide::<i64>)),
None,
),
Dtype::F64 => quantile_row::<f64>(
req,
description,
table,
want,
ddo::insert_dd_oxide::<f64>,
ddo::insert_step_dd_oxide::<f64>,
ddo::query_dd_oxide::<f64>,
Some((ddo::merge_dd_oxide::<f64>, ddo::merge_step_dd_oxide::<f64>)),
None,
),
other => Err(no_build_at(req, other)),
}
}

// -------- Hydra (per-subpopulation statistics over labelled records) --------

pub(crate) fn row_hydra_cms_lib(
Expand Down
9 changes: 4 additions & 5 deletions aqpbm-core/src/atomic_costs.rs
Original file line number Diff line number Diff line change
Expand Up @@ -238,7 +238,9 @@ mod tests {
mode: Mode::Bench,
runs: 5,
source: Source::Cli,
sketch_config: Some(serde_json::json!({"algorithm": "cms", "params": {"rows": 3, "cols": 1024}})),
sketch_config: Some(
serde_json::json!({"algorithm": "cms", "params": {"rows": 3, "cols": 1024}}),
),
input_dataset: dataset(),
memory_bytes: Some(12_288),
heap_bytes_net: None,
Expand Down Expand Up @@ -311,10 +313,7 @@ mod tests {
fn zero_elapsed_time_is_skipped_not_divided() {
let mut record = full_record();
record.insert.wall_time_ms = Some(stats(0.0));
assert_eq!(
reduce_one(&record),
Err(SkipReason::ZeroWork("insert"))
);
assert_eq!(reduce_one(&record), Err(SkipReason::ZeroWork("insert")));
}

#[test]
Expand Down
2 changes: 1 addition & 1 deletion aqpbm-core/src/lib.rs
Original file line number Diff line number Diff line change
Expand Up @@ -13,12 +13,12 @@ pub mod metrics;
// sketches, and no file format — it hands back in-memory columns.
// Re-exported so callers find it here.
pub use accuracy::GroundTruth;
pub use atomic_costs::{reduce_all, reduce_one, AtomicCostEntry, AtomicCostTable, SkipReason};
pub use aqpbm_datagen::{
ColumnData, ColumnItem, ColumnSpec, DataDistribution, DataGenError, GeneratedTable,
NormalParameter, StringOpts, TableDescription, UniformParameter, ZipfParameter,
RULE_MONOTONIC_INCREASE, RULE_NONE,
};
pub use atomic_costs::{reduce_all, reduce_one, AtomicCostEntry, AtomicCostTable, SkipReason};
pub use benchmark_result::{
BenchReport, BenchSection, CpuTime, InsertMetrics, LatencySummary, MergeMetrics, MergedRecord,
Mode, PrepareMetrics, QueryMetrics, Record, RunStats, Source, SCHEMA_VERSION,
Expand Down
11 changes: 11 additions & 0 deletions sketch-bench/src/params.rs
Original file line number Diff line number Diff line change
Expand Up @@ -370,6 +370,13 @@ pub struct KllParams {
}
sketch_params!(KllParams, "kll", KllParams { k: 100 });

#[derive(Debug, Clone, Copy, PartialEq, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct DdParams {
pub alpha: f64,
}
sketch_params!(DdParams, "dd", DdParams { alpha: 0.01 });

#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct CmsParams {
Expand Down Expand Up @@ -546,6 +553,7 @@ mod tests {
}
check::<HllParams>();
check::<KllParams>();
check::<DdParams>();
check::<CmsParams>();
check::<CountSketchParams>();
check::<HydraCmsParams>();
Expand Down Expand Up @@ -578,6 +586,9 @@ mod tests {
assert!(KllParams::owns("kll-cdf"));
assert!(!KllParams::owns("hydra-kll"));

assert!(DdParams::owns("dd"));
assert!(!DdParams::owns("kll"));

// The three Hydra cell types take different knobs, so they are three
// families and none of them owns another's name.
assert!(HydraCmsParams::owns("hydra-cms"));
Expand Down
33 changes: 33 additions & 0 deletions sketch-bench/src/registry.rs
Original file line number Diff line number Diff line change
Expand Up @@ -419,6 +419,39 @@ pub const REGISTRY: &[SketchId] = &[
.union(MetricsMask::MEMORY)
.union(MetricsMask::ACCURACY),
},
// -------- DDSketch (quantile) --------
SketchId {
family: "dd",
algorithm: "dd",
impl_name: "lib",
description: "asap_sketchlib::DDSketch: relative-error buckets, alpha in (0, 1)",
capability: Capability::Quantile,
comparator: Some("rank-error"),
operations: OperationMask::INSERT
.union(OperationMask::QUERY)
.union(OperationMask::MERGE),
metrics: MetricsMask::THROUGHPUT
.union(MetricsMask::LATENCY)
.union(MetricsMask::CPU)
.union(MetricsMask::MEMORY)
.union(MetricsMask::ACCURACY),
},
SketchId {
family: "dd",
algorithm: "dd",
impl_name: "oxide",
description: "sketch_oxide::quantiles::DDSketch: relative-error buckets, alpha in (0, 1)",
capability: Capability::Quantile,
comparator: Some("rank-error"),
operations: OperationMask::INSERT
.union(OperationMask::QUERY)
.union(OperationMask::MERGE),
metrics: MetricsMask::THROUGHPUT
.union(MetricsMask::LATENCY)
.union(MetricsMask::CPU)
.union(MetricsMask::MEMORY)
.union(MetricsMask::ACCURACY),
},
// -------- Hydra (per-subpopulation statistics over labelled records) --------
// Three families, not one: what sits in a cell decides which statistic the
// grid answers, so each cell type gets its own params vocabulary and its own
Expand Down
32 changes: 32 additions & 0 deletions sketch-bench/src/wrappers/dd/mod.rs
Original file line number Diff line number Diff line change
@@ -0,0 +1,32 @@
use crate::params::*;
use crate::wrappers::BuildError;

pub mod oxide;
pub mod sketchlib;

pub use super::kll::{QuantileValue, ToF64};

fn require_alpha(what: &str, alpha: f64) -> Result<(), BuildError> {
if !(alpha > 0.0 && alpha < 1.0) {
return Err(BuildError(format!(
"{what}: alpha={alpha} outside (0, 1), which is the relative accuracy \
this library accepts"
)));
}
Ok(())
}

fn dd_footprint(alpha: f64, min: Option<f64>, max: Option<f64>, count: u64) -> usize {
let (Some(min), Some(max)) = (min, max) else {
return 0;
};
let (lo, hi) = (min.abs().min(max.abs()), min.abs().max(max.abs()));
let log_gamma = ((1.0 + alpha) / (1.0 - alpha)).ln();
let span = if lo > 0.0 && hi.is_finite() && log_gamma > 0.0 {
((hi.ln() - lo.ln()) / log_gamma).floor() as u64 + 1
} else {
count
};
let stores = if min < 0.0 && max > 0.0 { 2 } else { 1 };
(span.saturating_mul(stores).min(count) as usize) * std::mem::size_of::<u64>()
}
Loading
Loading