Skip to content

Commit 20d2923

Browse files
authored
chore: fix the small nits collected from clean reviews (#745)
refactor(types)!: make fixed_interval_ms a QueryRecurrence method The free function workload_candidates::fixed_interval_ms(&recurrence) read only its argument, and Stage 2 imported it from Stage 1. It is now QueryRecurrence::fixed_interval_ms(&self) in asap-types, next to the type, with a test of every recurrence form. refactor(logical-asap-optimization)!: keep the HLL failure statistic format crate-private hll::classic_failure_statistic was public only so a Stage 3 test could build the statistic it binds. It is pub(crate) again, and the test uses test_support::hll_classic_failure_statistic, compiled only with the test-support feature. The crate docs said Stage 1 never depends on a front end; with test-support on it depends on asap-frontend-promql to lower test queries. Say so. refactor(logical-asap-optimization): name capability sharing candidates by what they are The locals holding CapabilitySharingCandidate values were still called groups/group/group_options; name them candidates/candidate. fix(executor): look through every time shift below a precompute range boundary_schema skipped at most one TimeShift between a TimeRange and its time series scan, so a range over stacked shifts was taken for a stored summary population and failed with "stored population requires one typed summary state". Follow every shift. fix(dag-viewer): judge costlier vs rejected by the rejection cause The Stage 3 list called a candidate "costlier" when the document had its cost and "rejected" otherwise, not by the cause Stage 3 wrote. It now reads the cause: a failed check (FAILED_CHECK_CAUSES) is rejected, a loss under the cost model is costlier. The failed-check cause list was also copied into the stage_pipeline and viewer tests. The Rust test now deserializes the cause into RejectionCause (which derives Deserialize) and asks failed_a_check; the viewer test reads FAILED_CHECK_CAUSES from stages.js. refactor(plan-selection): key raw data by source, not its Debug text Stage 3 kept the largest query-time scan per source in a map keyed by format!("{source:?}"). Key it by the Source itself (which now derives Hash). Restore "CPU" in the measured_operation_costs module doc: the table prices a sketch node's CPU, not the whole node.
1 parent 58999df commit 20d2923

16 files changed

Lines changed: 182 additions & 77 deletions

File tree

‎crates/devtools/tests/stage_pipeline.rs‎

Lines changed: 3 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -9,6 +9,7 @@ use asap_devtools::planner_layering_examples::{
99
use asap_devtools::planning_stages::{
1010
lower_promql_roots, run_planning_stages, PlanningStagesError,
1111
};
12+
use asap_plan_selection::RejectionCause;
1213
use asap_types::ir::flat::FlatDag;
1314
use asap_types::ir::physical_export::{PhysicalASAPDAG, PhysicalASAPDAGDocument};
1415
use asap_types::ir::QueryRoot;
@@ -135,14 +136,8 @@ fn example1_document_is_valid_and_committed_fixture_is_current() {
135136
assert!(!rejection["reason"].as_str().unwrap().is_empty());
136137
// A candidate that failed a check is not priced; one that lost
137138
// under the cost model is.
138-
let failed_a_check = [
139-
"failure_probability_resize_failed",
140-
"accuracy_target_missed",
141-
"raw_data_not_retained",
142-
"latency_bound_exceeded",
143-
"over_cluster_capacity",
144-
]
145-
.contains(&rejection["cause"].as_str().unwrap());
139+
let cause: RejectionCause = serde_json::from_value(rejection["cause"].clone()).unwrap();
140+
let failed_a_check = cause.failed_a_check();
146141
assert_eq!(failed_a_check, !costs.contains_key(id), "{rejection}");
147142
}
148143
assert!(costs.contains_key(selected));

‎crates/executor/src/physical_planner/precompute.rs‎

Lines changed: 3 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -82,8 +82,8 @@ pub fn raw_sample_row(
8282
}
8383

8484
/// Input contract of a precompute boundary: raw sample rows for a raw time
85-
/// series scan (or a `TimeRange`, optionally over a `TimeShift`, of one),
86-
/// otherwise the stored population of its summary state.
85+
/// series scan (or a `TimeRange`, optionally over stacked `TimeShift`s, of
86+
/// one), otherwise the stored population of its summary state.
8787
pub fn boundary_schema(
8888
node: &PhysicalASAPDAGNode,
8989
nodes: &BTreeMap<u64, &PhysicalASAPDAGNode>,
@@ -93,7 +93,7 @@ pub fn boundary_schema(
9393
scan = nodes
9494
.get(&(*child as u64))
9595
.ok_or_else(|| invalid("missing precompute node"))?;
96-
if let Payload::NonASAP(NonASAPOp::TimeShift { child, .. }) = &scan.payload {
96+
while let Payload::NonASAP(NonASAPOp::TimeShift { child, .. }) = &scan.payload {
9797
scan = nodes
9898
.get(&(*child as u64))
9999
.ok_or_else(|| invalid("missing precompute node"))?;

‎crates/executor/tests/precompute.rs‎

Lines changed: 33 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -2,7 +2,7 @@
22
mod common;
33
use asap_executor::physical_planner::precompute::{boundary_schema, raw_sample_schema};
44
use common::compile_physical_asap_dag;
5-
use planner_types::ir::operator::Source;
5+
use planner_types::ir::operator::{Source, TimeShift};
66
use planner_types::ir::schema::{DataType, Field, Schema};
77
use planner_types::ir::{NonASAPOp, Operator, OperatorNode, TimeRangeKind};
88
use std::{collections::BTreeMap, rc::Rc, time::Duration};
@@ -53,3 +53,35 @@ fn only_a_time_series_range_reads_raw_samples() {
5353
// A table range is not raw samples: it must be a stored population.
5454
assert_ne!(table.ok(), Some(raw_sample_schema()));
5555
}
56+
57+
/// A range over stacked time shifts of a time series scan still reads raw
58+
/// samples: the boundary looks through every shift, not only the first.
59+
#[test]
60+
fn a_range_over_stacked_shifts_reads_raw_samples() {
61+
let range = time_range_over(Source::TimeSeries { metric: "m".into() });
62+
let NonASAPOp::TimeRange {
63+
range: window,
64+
kind,
65+
child: scan,
66+
} = range.non_asap().unwrap().clone()
67+
else {
68+
unreachable!()
69+
};
70+
let shifted = |offset_ms, child| {
71+
OperatorNode::new_shared(Operator::NonASAP(NonASAPOp::TimeShift {
72+
shift: TimeShift {
73+
offset_ms,
74+
at: None,
75+
},
76+
child,
77+
}))
78+
.unwrap()
79+
};
80+
let stacked = OperatorNode::new_shared(Operator::NonASAP(NonASAPOp::TimeRange {
81+
range: window,
82+
kind,
83+
child: shifted(60_000, shifted(300_000, scan)),
84+
}))
85+
.unwrap();
86+
assert_eq!(boundary(&stacked).unwrap(), raw_sample_schema());
87+
}

‎crates/logical-asap-optimization/Cargo.toml‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -5,8 +5,8 @@ edition = "2021"
55

66
# #509 Stage 1, logical ASAP-aware optimization: Pass 1 per-computation
77
# candidates and Pass 2 ASAP-aware CSE over asap-types' operator DAG, with
8-
# the analytical accuracy model. Depends on no front end and on no later
9-
# stage.
8+
# the analytical accuracy model. Depends on no later stage, and on no
9+
# query-language frontend except through the `test-support` feature.
1010
[dependencies]
1111
asap_sketchlib = { workspace = true }
1212
asap-types = { path = "../types" }

‎crates/logical-asap-optimization/src/accuracy/estimators/hll.rs‎

Lines changed: 9 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -47,14 +47,18 @@ pub fn classic_relative_error(precision: u8, relative_error: Option<f64>) -> f64
4747
/// `relative_error` over `source` is left as ([`classic_guarantee`]).
4848
/// `relative_error` is written in Rust's shortest round-trip form, so it
4949
/// parses back to the same value.
50-
pub fn classic_failure_statistic(precision: u8, relative_error: f64, source: &str) -> String {
50+
pub(crate) fn classic_failure_statistic(
51+
precision: u8,
52+
relative_error: f64,
53+
source: &str,
54+
) -> String {
5155
format!(
5256
"{CLASSIC_FAILURE_PROBABILITY}[p={precision},eps={relative_error}]({MAX_DISTINCT}[{source}])"
5357
)
5458
}
5559

56-
/// The precision, relative error and source a [`classic_failure_statistic`]
57-
/// names.
60+
/// The precision, relative error and source a classic HLL failure
61+
/// statistic (`classic_failure_statistic`) names.
5862
pub fn parse_classic_failure_statistic(statistic: &str) -> Option<(u8, f64, &str)> {
5963
let rest = statistic
6064
.strip_prefix(CLASSIC_FAILURE_PROBABILITY)?
@@ -112,7 +116,8 @@ pub(crate) fn classic_guarantee(
112116
))
113117
}
114118

115-
/// The failure probability a [`classic_failure_statistic`] names, given the
119+
/// The failure probability a classic HLL failure statistic
120+
/// (`classic_failure_statistic`) names, given the
116121
/// source's distinct bound: `1` (no certificate) when the bound is outside
117122
/// the contract's range for this precision and relative error.
118123
pub fn classic_failure_probability(precision: u8, relative_error: f64, max_distinct: u64) -> f64 {

‎crates/logical-asap-optimization/src/asap_aware_cse.rs‎

Lines changed: 23 additions & 20 deletions
Original file line numberDiff line numberDiff line change
@@ -393,24 +393,24 @@ pub fn capability_sharing_candidates(roots: &[QueryRoot]) -> Vec<CapabilityShari
393393
readers.entry(Rc::as_ptr(child)).or_default().push(node);
394394
}
395395
}
396-
let mut groups: Vec<CapabilitySharingCandidate> = Vec::new();
396+
let mut candidates: Vec<CapabilitySharingCandidate> = Vec::new();
397397
for node in &nodes {
398398
let Some(family) = shareable_family(node, readers.get(&Rc::as_ptr(node))) else {
399399
continue;
400400
};
401-
match groups
401+
match candidates
402402
.iter_mut()
403-
.find(|g| g.family == family && same_input_data(&g.members[0], node))
403+
.find(|c| c.family == family && same_input_data(&c.members[0], node))
404404
{
405-
Some(group) => group.members.push(Rc::clone(node)),
406-
None => groups.push(CapabilitySharingCandidate {
405+
Some(candidate) => candidate.members.push(Rc::clone(node)),
406+
None => candidates.push(CapabilitySharingCandidate {
407407
family,
408408
members: vec![Rc::clone(node)],
409409
}),
410410
}
411411
}
412-
groups.retain(|g| !all_equal(g.members.iter()));
413-
groups
412+
candidates.retain(|c| !all_equal(c.members.iter()));
413+
candidates
414414
}
415415

416416
fn collect_children_first(
@@ -510,25 +510,28 @@ pub fn summary_capability_pass2_candidates(
510510
roots: &[QueryRoot],
511511
candidates: &[CapabilitySharingCandidate],
512512
) -> Result<Vec<SummaryCapabilityPass2Candidate>, PerComputationCandidateError> {
513-
// Every subset of a group is a candidate: 2^n of them. Past
514-
// MAX_CAPABILITY_SHARING_MEMBERS that is no longer a list plan selection can price.
513+
// Every subset of a capability sharing candidate is a Pass 2
514+
// candidate: 2^n of them. Past MAX_CAPABILITY_SHARING_MEMBERS that is
515+
// no longer a list plan selection can price.
515516
if candidates
516517
.iter()
517-
.any(|group| group.members.len() > MAX_CAPABILITY_SHARING_MEMBERS)
518+
.any(|candidate| candidate.members.len() > MAX_CAPABILITY_SHARING_MEMBERS)
518519
{
519520
return Err(PerComputationCandidateError::Unsupported(
520521
"a capability sharing candidate of more than 20 summary builds (2^n sharing subsets)",
521522
));
522523
}
523-
// Each group's options: `None` (no sharing), then its subsets.
524+
// Each capability sharing candidate's options: `None` (no sharing),
525+
// then its subsets.
524526
let options: Vec<Vec<Option<Vec<usize>>>> = candidates
525527
.iter()
526-
.map(|group| {
527-
let n = group.members.len();
528+
.map(|candidate| {
529+
let n = candidate.members.len();
528530
let mut options = vec![None];
529531
for mask in 1u64..(1u64 << n) {
530532
let members: Vec<usize> = (0..n).filter(|&m| mask >> m & 1 == 1).collect();
531-
if members.len() >= 2 && !all_equal(members.iter().map(|&m| &group.members[m])) {
533+
if members.len() >= 2 && !all_equal(members.iter().map(|&m| &candidate.members[m]))
534+
{
532535
options.push(Some(members));
533536
}
534537
}
@@ -538,11 +541,11 @@ pub fn summary_capability_pass2_candidates(
538541
let mut choice = vec![0usize; candidates.len()];
539542
let mut pass2_candidates = Vec::new();
540543
loop {
541-
// Mixed-radix increment, the last group varying fastest.
544+
// Mixed-radix increment, the last candidate varying fastest.
542545
let mut carried = true;
543-
for (digit, group_options) in choice.iter_mut().zip(&options).rev() {
546+
for (digit, candidate_options) in choice.iter_mut().zip(&options).rev() {
544547
*digit += 1;
545-
if *digit < group_options.len() {
548+
if *digit < candidate_options.len() {
546549
carried = false;
547550
break;
548551
}
@@ -554,12 +557,12 @@ pub fn summary_capability_pass2_candidates(
554557
let shared_builds: Vec<Vec<Rc<OperatorNode>>> = choice
555558
.iter()
556559
.zip(candidates.iter().zip(&options))
557-
.filter_map(|(&option, (group, group_options))| {
558-
let members = group_options[option].as_ref()?;
560+
.filter_map(|(&option, (candidate, candidate_options))| {
561+
let members = candidate_options[option].as_ref()?;
559562
Some(
560563
members
561564
.iter()
562-
.map(|&m| Rc::clone(&group.members[m]))
565+
.map(|&m| Rc::clone(&candidate.members[m]))
563566
.collect(),
564567
)
565568
})

‎crates/logical-asap-optimization/src/lib.rs‎

Lines changed: 7 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -7,10 +7,13 @@
77
//! invisible at the logical level ([`asap_aware_cse`], Pass 2), e.g.
88
//! `Quantile(x, 0.99)` and `Quantile(x, 0.95)` reading one built sketch.
99
//!
10-
//! It depends only on `asap-types`, never on a front end or a later stage:
11-
//! Stage 2 (`asap-physical-asap-optimization`) and Stage 3
12-
//! (`asap-plan-selection`) depend on it, and Cargo enforces that one-way
13-
//! flow.
10+
//! It depends only on `asap-types`, never on a later stage: Stage 2
11+
//! (`asap-physical-asap-optimization`) and Stage 3 (`asap-plan-selection`)
12+
//! depend on it, and Cargo enforces that one-way flow. It depends on no
13+
//! query-language frontend either, except with the `test-support` feature,
14+
//! whose test helpers (`test_support`) lower PromQL test queries through
15+
//! `asap-frontend-promql`; the later stages enable that feature only for
16+
//! their tests.
1417
//!
1518
//! Post-lowering **canonicalization** is *not* here: it lives in
1619
//! `asap_types::ir::canonicalize`, run inside `asap_frontend_common`'s shared

‎crates/logical-asap-optimization/src/test_support.rs‎

Lines changed: 7 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -11,6 +11,13 @@ use asap_types::workload::{
1111
QueryRequirements, QueryWorkload,
1212
};
1313

14+
/// The statistic classic HLL's failure probability at `precision` and
15+
/// `relative_error` over `source` is left as, for Stage 3 tests that bind
16+
/// it. The estimator keeps the format crate-private.
17+
pub fn hll_classic_failure_statistic(precision: u8, relative_error: f64, source: &str) -> String {
18+
crate::accuracy::estimators::hll::classic_failure_statistic(precision, relative_error, source)
19+
}
20+
1421
pub fn lower_promql(query: &str, accuracy: AccuracyTarget) -> Rc<OperatorNode> {
1522
let workload = PlanningWorkload {
1623
query_workload: QueryWorkload {

‎crates/logical-asap-optimization/src/workload_candidates.rs‎

Lines changed: 5 additions & 13 deletions
Original file line numberDiff line numberDiff line change
@@ -8,7 +8,7 @@ use std::rc::Rc;
88

99
use asap_types::ir::flat::flatten;
1010
use asap_types::ir::{OperatorNode, QueryRoot};
11-
use asap_types::workload::{QueryRecurrence, RepeatedDemand};
11+
use asap_types::workload::QueryRecurrence;
1212

1313
use crate::asap_aware_cse::{
1414
capability_sharing_candidates, identical_expression_pass2_candidates, pane_candidates,
@@ -79,7 +79,10 @@ pub fn stage1_logical_asap_optimization<'a, Id: Clone>(
7979
per_computation: &'a PerComputationCandidates<Id>,
8080
recurrence: &[QueryRecurrence],
8181
) -> impl Iterator<Item = Stage1Candidate> + 'a {
82-
let interval_ms: Vec<Option<u64>> = recurrence.iter().map(fixed_interval_ms).collect();
82+
let interval_ms: Vec<Option<u64>> = recurrence
83+
.iter()
84+
.map(QueryRecurrence::fixed_interval_ms)
85+
.collect();
8386
let mut choice = Some(vec![0; per_computation.targets.len()]);
8487
let mut queue = std::collections::VecDeque::new();
8588
let mut position = 0usize;
@@ -355,17 +358,6 @@ fn next_choice<Id>(inventory: &PerComputationCandidates<Id>, choice: &mut [usize
355358
false
356359
}
357360

358-
/// The fixed interval at which a query repeats, in milliseconds.
359-
pub fn fixed_interval_ms(recurrence: &QueryRecurrence) -> Option<u64> {
360-
match recurrence {
361-
QueryRecurrence::Repeated(
362-
RepeatedDemand::FixedInterval(interval)
363-
| RepeatedDemand::FixedIntervalAt { interval, .. },
364-
) => Some(u64::from(interval.0)).filter(|&ms| ms > 0),
365-
_ => None,
366-
}
367-
}
368-
369361
#[cfg(test)]
370362
mod tests {
371363
use super::*;

‎crates/physical-asap-optimization/src/lib.rs‎

Lines changed: 1 addition & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -93,7 +93,6 @@ use asap_types::workload::{
9393
use thiserror::Error;
9494

9595
use asap_logical_asap_optimization::deployment_capabilities::DeploymentCapabilities;
96-
use asap_logical_asap_optimization::workload_candidates::fixed_interval_ms;
9796

9897
/// One CandidatePhysicalASAPDAG (Stage 2 output), derived from exactly one
9998
/// CandidateLogicalASAPDAG (Stage 1 output). It holds planning data only;
@@ -241,7 +240,7 @@ pub(crate) fn builds_one_pane_per_evaluation<'a>(
241240
let mut any_reader = false;
242241
for recurrence in readers {
243242
any_reader = true;
244-
if fixed_interval_ms(recurrence) != Some(pane_width_ms) {
243+
if recurrence.fixed_interval_ms() != Some(pane_width_ms) {
245244
return false;
246245
}
247246
if let QueryRecurrence::Repeated(RepeatedDemand::FixedIntervalAt {

0 commit comments

Comments
 (0)