Skip to content

Commit b044f4f

Browse files
committed
feat(planner): enumerate phase-free local logical alternatives
1 parent 2918b78 commit b044f4f

4 files changed

Lines changed: 432 additions & 0 deletions

File tree

‎crates/asap-aware-mapping/src/lib.rs‎

Lines changed: 3 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -234,3 +234,6 @@ pub use summary_maintenance_lifecycle::{
234234
pub use topk_reuse::TopKLimitReuseStrategy;
235235

236236
pub mod maintained_population;
237+
238+
/// Phase-free local candidate generation over the unified IR.
239+
pub mod logical_candidates;
Lines changed: 147 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,147 @@
1+
//! Pass 1 local alternatives over the unified logical IR.
2+
//!
3+
//! Alternatives are nominal realization descriptors attached to their original
4+
//! target, not ranked plans or accuracy certificates. Workload composition and
5+
//! physical planning consume this inventory later; empirical models belong to
6+
//! selection. The legacy search API remains until planner cutover.
7+
use std::collections::HashSet;
8+
use std::rc::Rc;
9+
10+
use asap_types::ir::{NonASAPOp, OperatorNode, QueryRoot, SchemaDerivationError};
11+
use asap_types::post_asap::{ExactKind, ExactParams, SketchKind};
12+
use asap_types::pre_asap::AggIntent;
13+
use asap_types::types::AccuracyTarget;
14+
use thiserror::Error;
15+
16+
use crate::replacement::{
17+
accuracy_budget, accuracy_target, default_size_params, summary_candidates, Realization,
18+
};
19+
20+
/// All local realizations of one single-measure aggregate. The target retains
21+
/// source, grouping, filters, input expressions and evaluation context.
22+
#[derive(Debug, Clone)]
23+
pub struct LocalLogicalTarget {
24+
pub target: Rc<OperatorNode>,
25+
pub alternatives: Vec<Realization>,
26+
}
27+
28+
/// Compact Pass 1 inventory; roots and nested producer dependencies are retained.
29+
#[derive(Debug, Clone)]
30+
pub struct LocalLogicalCandidates<Id> {
31+
pub roots: Vec<(Id, QueryRoot)>,
32+
pub targets: Vec<LocalLogicalTarget>,
33+
}
34+
35+
#[derive(Debug, Error)]
36+
pub enum LogicalCandidateError {
37+
#[error(transparent)]
38+
Structure(#[from] SchemaDerivationError),
39+
#[error("logical candidate input already has assigned execution timing")]
40+
AssignedTiming,
41+
#[error("approximate accuracy requires finite positive epsilon and delta in (0, 1)")]
42+
InvalidAccuracy,
43+
}
44+
45+
/// Enumerate exact and summary choices in stable catalog order, without ranking
46+
/// or empirical assessment. Parameters are candidate dimensions, not a claim
47+
/// that a deployment meets the request's accuracy requirement.
48+
pub fn local_realizations_for_intent(
49+
intent: &AggIntent,
50+
) -> Result<Vec<Realization>, LogicalCandidateError> {
51+
let mut choices = vec![Realization::PassThrough];
52+
let exact = match intent {
53+
AggIntent::Count { .. } => Some((ExactKind::Count, ExactParams::Count)),
54+
AggIntent::Sum { .. } => Some((ExactKind::Sum, ExactParams::Sum)),
55+
AggIntent::Min { .. } => Some((ExactKind::Min, ExactParams::Min)),
56+
AggIntent::Max { .. } => Some((ExactKind::Max, ExactParams::Max)),
57+
AggIntent::Rate => Some((ExactKind::Rate, ExactParams::Rate)),
58+
AggIntent::IRate => Some((ExactKind::IRate, ExactParams::IRate)),
59+
AggIntent::Increase => Some((ExactKind::Increase, ExactParams::Increase)),
60+
_ => None,
61+
};
62+
if let Some((kind, params)) = exact {
63+
choices.push(Realization::ExactAggregate { kind, params });
64+
}
65+
if let Some(target) = accuracy_target(intent) {
66+
if *target != AccuracyTarget::Exact {
67+
let (epsilon, delta) = accuracy_budget(target);
68+
if !epsilon.is_finite()
69+
|| epsilon <= 0.0
70+
|| !delta.is_finite()
71+
|| !(0.0..1.0).contains(&delta)
72+
|| delta == 0.0
73+
{
74+
return Err(LogicalCandidateError::InvalidAccuracy);
75+
}
76+
for algorithm in summary_candidates(intent) {
77+
choices.push(Realization::Sketch(SketchKind::new(
78+
algorithm.clone(),
79+
default_size_params(algorithm.clone(), intent, epsilon, delta),
80+
)));
81+
}
82+
}
83+
}
84+
Ok(choices)
85+
}
86+
87+
/// Discover single-measure targets, including operator plans read by scalar roots.
88+
/// Multi-measure aggregates remain intact pending a semantics-preserving split.
89+
pub fn enumerate_local_logical_candidates<Id>(
90+
roots: Vec<(Id, QueryRoot)>,
91+
) -> Result<LocalLogicalCandidates<Id>, LogicalCandidateError> {
92+
let mut seen = HashSet::new();
93+
let mut targets = Vec::new();
94+
for (_, root) in &roots {
95+
root.validate_structure()?;
96+
let operators = match root {
97+
QueryRoot::Operator(node) => vec![node],
98+
QueryRoot::Scalar(expr) => expr.operator_refs(),
99+
};
100+
for root in operators {
101+
for node in OperatorNode::reachable(root) {
102+
if !seen.insert(Rc::as_ptr(&node)) {
103+
continue;
104+
}
105+
if node.timing.is_some() {
106+
return Err(LogicalCandidateError::AssignedTiming);
107+
}
108+
if let Some(NonASAPOp::Aggregate { measures, .. }) = node.non_asap() {
109+
if let [intent] = measures.as_slice() {
110+
targets.push(LocalLogicalTarget {
111+
alternatives: local_realizations_for_intent(intent)?,
112+
target: node,
113+
});
114+
}
115+
}
116+
}
117+
}
118+
}
119+
Ok(LocalLogicalCandidates { roots, targets })
120+
}
121+
122+
#[cfg(test)]
123+
mod tests {
124+
use super::*;
125+
/// Approximate requests must retain the exact execution alternative too.
126+
#[test]
127+
fn approximate_count_keeps_exact_and_universal_choices() {
128+
let choices = local_realizations_for_intent(&AggIntent::Count {
129+
accuracy: AccuracyTarget::EpsilonDelta {
130+
epsilon: 0.05,
131+
delta: 0.01,
132+
},
133+
})
134+
.unwrap();
135+
assert!(choices
136+
.iter()
137+
.any(|choice| matches!(choice, Realization::PassThrough)));
138+
assert!(choices.iter().any(|choice| matches!(
139+
choice,
140+
Realization::ExactAggregate {
141+
kind: ExactKind::Count,
142+
..
143+
}
144+
)));
145+
assert!(choices.iter().any(|choice| matches!(choice, Realization::Sketch(kind) if *kind.algorithm() == asap_types::post_asap::SketchAlgorithm::UnivMon)));
146+
}
147+
}
Lines changed: 243 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,243 @@
1+
//! Frontend-to-Pass-1 acceptance: candidate discovery precedes empirical selection.
2+
use asap_aware_mapping::{
3+
logical_candidates::{
4+
enumerate_local_logical_candidates, local_realizations_for_intent, LogicalCandidateError,
5+
},
6+
Realization,
7+
};
8+
use asap_types::{
9+
ir::operator_properties::{Reduction, Source},
10+
ir::{NonASAPOp, Operator, OperatorNode, QueryRoot, ScalarExpr},
11+
post_asap::{ExactKind, SketchAlgorithm},
12+
pre_asap::{AggIntent, DataType, Field, Schema},
13+
types::AccuracyTarget,
14+
};
15+
use std::rc::Rc;
16+
17+
fn approximate() -> AccuracyTarget {
18+
AccuracyTarget::EpsilonDelta {
19+
epsilon: 0.05,
20+
delta: 0.01,
21+
}
22+
}
23+
fn algorithms(choices: &[Realization]) -> Vec<SketchAlgorithm> {
24+
choices
25+
.iter()
26+
.filter_map(|choice| match choice {
27+
Realization::Sketch(kind) => Some(kind.algorithm().clone()),
28+
_ => None,
29+
})
30+
.collect()
31+
}
32+
fn aggregate(intent: AggIntent) -> Rc<OperatorNode> {
33+
let child = OperatorNode::new_shared(Operator::NonASAP(NonASAPOp::Scan {
34+
source: Source::Table {
35+
table_ref: "flows".into(),
36+
},
37+
predicates: vec![],
38+
schema: Schema::lifted(vec![Field::plain("src_ip", DataType::Utf8, false)], None),
39+
}))
40+
.unwrap();
41+
OperatorNode::new_shared(Operator::NonASAP(NonASAPOp::Aggregate {
42+
child,
43+
reduction: Reduction::by(vec![]),
44+
measures: vec![intent],
45+
output_names: vec![],
46+
filters: vec![],
47+
having: None,
48+
}))
49+
.unwrap()
50+
}
51+
52+
/// Example 2 preserves specialized distinct summaries and the universal alternative.
53+
#[test]
54+
fn cardinality_keeps_exact_specialized_and_universal_alternatives() {
55+
let choices = local_realizations_for_intent(&AggIntent::Cardinality {
56+
cols: vec![0],
57+
accuracy: approximate(),
58+
})
59+
.unwrap();
60+
assert!(matches!(choices[0], Realization::PassThrough));
61+
assert_eq!(
62+
algorithms(&choices),
63+
vec![
64+
SketchAlgorithm::Hll,
65+
SketchAlgorithm::Theta,
66+
SketchAlgorithm::Kmv,
67+
SketchAlgorithm::UnivMon
68+
]
69+
);
70+
let tuple = local_realizations_for_intent(&AggIntent::Cardinality {
71+
cols: vec![0, 1],
72+
accuracy: approximate(),
73+
})
74+
.unwrap();
75+
assert!(!algorithms(&tuple).contains(&SketchAlgorithm::UnivMon));
76+
}
77+
78+
/// Frequency moments retain exact execution and a universal sketch without certification.
79+
#[test]
80+
fn frequency_statistics_keep_universal_choices() {
81+
for intent in [
82+
AggIntent::FrequencyL2 {
83+
col: Some(0),
84+
accuracy: approximate(),
85+
},
86+
AggIntent::FrequencyEntropy {
87+
col: Some(0),
88+
accuracy: approximate(),
89+
},
90+
] {
91+
let choices = local_realizations_for_intent(&intent).unwrap();
92+
assert!(matches!(choices[0], Realization::PassThrough));
93+
assert_eq!(algorithms(&choices), vec![SketchAlgorithm::UnivMon]);
94+
}
95+
}
96+
97+
/// An exact request cannot acquire an approximate sketch merely because one is available.
98+
#[test]
99+
fn exact_quantile_stays_exact_and_approximate_keeps_both_families() {
100+
let choices = local_realizations_for_intent(&AggIntent::Quantile {
101+
col: Some(0),
102+
q: 0.99,
103+
accuracy: approximate(),
104+
})
105+
.unwrap();
106+
assert_eq!(
107+
algorithms(&choices),
108+
vec![SketchAlgorithm::Kll, SketchAlgorithm::DDSketch]
109+
);
110+
let exact = local_realizations_for_intent(&AggIntent::Quantile {
111+
col: Some(0),
112+
q: 0.99,
113+
accuracy: AccuracyTarget::Exact,
114+
})
115+
.unwrap();
116+
assert_eq!(exact, vec![Realization::PassThrough]);
117+
}
118+
119+
/// Scalar roots expose their producer targets; repeated references retain one target identity.
120+
#[test]
121+
fn scalar_root_producers_are_discovered_once() {
122+
let producer = aggregate(AggIntent::Cardinality {
123+
cols: vec![0],
124+
accuracy: approximate(),
125+
});
126+
let roots = vec![
127+
(
128+
"scalar",
129+
QueryRoot::Scalar(ScalarExpr::ScalarSubquery(producer.clone())),
130+
),
131+
("relation", QueryRoot::Operator(producer.clone())),
132+
];
133+
let candidates = enumerate_local_logical_candidates(roots).unwrap();
134+
assert_eq!(candidates.roots.len(), 2);
135+
for (_, root) in &candidates.roots {
136+
asap_types::ir::export::compile_logical_asap_query(root)
137+
.unwrap()
138+
.validate()
139+
.unwrap();
140+
}
141+
assert_eq!(candidates.targets.len(), 1);
142+
assert!(Rc::ptr_eq(&candidates.targets[0].target, &producer));
143+
assert!(producer.timing.is_none());
144+
assert!(producer.guarantee.is_none());
145+
asap_types::ir::export::compile_logical_asap_dag(&producer)
146+
.unwrap()
147+
.validate()
148+
.unwrap();
149+
}
150+
151+
/// Example 1 rate lowering reaches the exact accumulator choice without a cost model.
152+
#[test]
153+
fn promql_lowering_reaches_phase_free_local_candidates() {
154+
use asap_types::workload::{
155+
AccuracyRequirement, BatchEntry, PlanningWorkload, Query, QueryLanguage, QueryRequirements,
156+
QueryWorkload,
157+
};
158+
let workload = PlanningWorkload {
159+
query_workload: QueryWorkload {
160+
language: QueryLanguage::PromQL,
161+
query_batch: Some(vec![BatchEntry {
162+
query: Query("sum by (job) (rate(http_requests_total[1m]))".into()),
163+
requirements: QueryRequirements {
164+
accuracy: AccuracyRequirement::Explicit(approximate()),
165+
..Default::default()
166+
},
167+
predictability: Default::default(),
168+
invocations: 1,
169+
execute_at: None,
170+
time_selection: Default::default(),
171+
}]),
172+
repeating_queries: None,
173+
},
174+
data_workload: Some(asap_types::workload::DataWorkload {
175+
data_ingestion_interval: asap_types::workload::Evidence {
176+
value: Some(asap_types::workload::DurationMs(1000)),
177+
..Default::default()
178+
},
179+
..Default::default()
180+
}),
181+
};
182+
let roots = asap_frontend_promql::unified::lower_promql_query_workload(&workload, 0).unwrap();
183+
let candidates =
184+
enumerate_local_logical_candidates(roots.into_iter().enumerate().collect()).unwrap();
185+
assert!(candidates
186+
.targets
187+
.iter()
188+
.any(|target| target.alternatives.iter().any(|choice| matches!(
189+
choice,
190+
Realization::ExactAggregate {
191+
kind: ExactKind::Rate,
192+
..
193+
}
194+
))));
195+
assert!(candidates
196+
.targets
197+
.iter()
198+
.all(|target| target.target.timing.is_none()));
199+
}
200+
201+
/// Physical annotations and invalid probability requirements fail at the stage boundary.
202+
#[test]
203+
fn assigned_timing_and_invalid_accuracy_are_rejected() {
204+
let mut producer = (*aggregate(AggIntent::Count {
205+
accuracy: approximate(),
206+
}))
207+
.clone();
208+
producer.timing = Some(asap_types::post_asap::ExecutionTiming::QueryTime);
209+
assert!(matches!(
210+
enumerate_local_logical_candidates(vec![(0, QueryRoot::Operator(Rc::new(producer)))]),
211+
Err(LogicalCandidateError::AssignedTiming)
212+
));
213+
for target in [
214+
AccuracyTarget::Epsilon(f64::NAN),
215+
AccuracyTarget::EpsilonDelta {
216+
epsilon: 0.1,
217+
delta: 0.0,
218+
},
219+
] {
220+
assert!(matches!(
221+
local_realizations_for_intent(&AggIntent::Count { accuracy: target }),
222+
Err(LogicalCandidateError::InvalidAccuracy)
223+
));
224+
}
225+
}
226+
227+
/// Local TopK keeps both declared heap substrates without choosing an implementation.
228+
#[test]
229+
fn topk_keeps_both_specialized_heap_choices() {
230+
let choices = local_realizations_for_intent(&AggIntent::TopK {
231+
k: 10,
232+
accuracy: approximate(),
233+
})
234+
.unwrap();
235+
assert!(matches!(choices[0], Realization::PassThrough));
236+
assert_eq!(
237+
algorithms(&choices),
238+
vec![
239+
SketchAlgorithm::CmsWithHeap,
240+
SketchAlgorithm::CountSketchWithHeap
241+
]
242+
);
243+
}

0 commit comments

Comments
 (0)