Skip to content

Commit 0150354

Browse files
zzylolclaude
andcommitted
refactor(planner): derive local candidate coverage; write logical stages as flat DAGs
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01W7qG9aFyPij5uWsyAJCxDW
1 parent 81af76d commit 0150354

5 files changed

Lines changed: 7889 additions & 15996 deletions

File tree

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

Lines changed: 3 additions & 32 deletions
Original file line numberDiff line numberDiff line change
@@ -4,10 +4,9 @@
44
//! target, not ranked plans or accuracy certificates. Workload composition and
55
//! physical planning consume this inventory later; empirical models belong to
66
//! selection. The legacy search API remains until planner cutover.
7-
use std::collections::{BTreeMap, HashMap, HashSet};
7+
use std::collections::{HashMap, HashSet};
88
use std::rc::Rc;
99

10-
use asap_types::ir::summary_coverage::{CoverageRegion, SummaryCoverage};
1110
use asap_types::ir::{ASAPOp, NonASAPOp, Operator, OperatorNode, QueryRoot, SchemaDerivationError};
1211
use asap_types::post_asap::{
1312
EntityIdentity, ExactKind, ExactParams, FieldDataType, GroupingStrategy,
@@ -197,7 +196,7 @@ fn rewrite(
197196
Some(realization) if **realization != Realization::PassThrough => {
198197
realize(node, realization, memo)?
199198
}
200-
_ if changed => Rc::new(node.map_children(|child| memo[&Rc::as_ptr(child)].clone())?),
199+
_ if changed => Rc::new(node.with_new_children(|child| memo[&Rc::as_ptr(child)].clone())?),
201200
_ => node.clone(),
202201
};
203202
memo.insert(Rc::as_ptr(node), rebuilt.clone());
@@ -253,16 +252,7 @@ fn realize(
253252
grouping: GroupingStrategy::default(),
254253
filter: None,
255254
}))?;
256-
// Whole-source coverage is declared, not proven: Pass 1 trusts that the
257-
// state holds every observation of its source that reaches it (#570).
258-
let coverage = SummaryCoverage {
259-
source: single_source(&child)?,
260-
regions: vec![CoverageRegion {
261-
time_ms: None,
262-
population: BTreeMap::new(),
263-
}],
264-
};
265-
let state = Rc::new(state.with_coverage(coverage)?);
255+
let state = Rc::new(state);
266256
let evaluation = match query {
267257
Some(query) => ASAPOp::SummaryEstimate {
268258
summary_input: state,
@@ -273,25 +263,6 @@ fn realize(
273263
Ok(OperatorNode::new_shared(Operator::ASAP(evaluation))?)
274264
}
275265

276-
fn single_source(
277-
node: &Rc<OperatorNode>,
278-
) -> Result<asap_types::pre_asap::Source, LogicalCandidateError> {
279-
let mut sources = Vec::new();
280-
for node in OperatorNode::reachable(node) {
281-
if let Some(NonASAPOp::Scan { source, .. }) = node.non_asap() {
282-
if !sources.contains(source) {
283-
sources.push(source.clone());
284-
}
285-
}
286-
}
287-
match <[_; 1]>::try_from(sources) {
288-
Ok([source]) => Ok(source),
289-
Err(_) => Err(LogicalCandidateError::Unsupported(
290-
"summary coverage needs exactly one source",
291-
)),
292-
}
293-
}
294-
295266
/// What each input row contributes, following the legacy realization rules:
296267
/// heap sketches rank series identities, frequency sketches count values, and
297268
/// every other family reads the measure's input column.

‎crates/asap-aware-mapping/tests/logical_candidates.rs‎

Lines changed: 10 additions & 11 deletions
Original file line numberDiff line numberDiff line change
@@ -259,17 +259,16 @@ fn composed_candidate_replaces_chosen_target_with_summary_evaluation() {
259259
let Some(ASAPOp::SummaryEstimate { summary_input, .. }) = estimate.asap() else {
260260
panic!("summary evaluation expected")
261261
};
262-
let coverage = summary_input.coverage.as_ref().unwrap();
263-
assert_eq!(
264-
coverage.source,
265-
Source::Table {
266-
table_ref: "flows".into()
267-
}
262+
// The state takes every row of its source: one unrestricted selection box.
263+
let coverage = summary_input.coverage().unwrap();
264+
assert!(OperatorNode::reachable(&coverage.definition).iter().any(|node| matches!(
265+
node.non_asap(),
266+
Some(NonASAPOp::Scan { source: Source::Table { table_ref }, .. }) if table_ref == "flows"
267+
)));
268+
assert_eq!(coverage.selection.len(), 1);
269+
assert!(
270+
coverage.selection[0].columns.is_empty() && coverage.selection[0].relative_time.is_none()
268271
);
269-
assert!(coverage.regions[0].time_ms.is_none() && coverage.regions[0].population.is_empty());
270-
asap_types::ir::export::compile_logical_asap_workload(&[hll[0].1.clone()])
271-
.unwrap()
272-
.validate()
273-
.unwrap();
272+
hll[0].1.validate_structure().unwrap();
274273
assert!(compose_logical_candidate(&inventory, &[99]).is_err());
275274
}

‎crates/devtools/src/bin/stage_pipeline.rs‎

Lines changed: 9 additions & 9 deletions
Original file line numberDiff line numberDiff line change
@@ -21,9 +21,7 @@ use asap_aware_mapping::logical_candidates::{
2121
compose_logical_candidate, enumerate_local_logical_candidates, LocalLogicalCandidates,
2222
};
2323
use asap_aware_mapping::Realization;
24-
use asap_types::ir::export::{
25-
compile_logical_asap_workload, LogicalASAPDAG, LogicalASAPDAGDocument,
26-
};
24+
use asap_types::ir::flat::{flatten, FlatDag};
2725
use asap_types::ir::{OperatorNode, QueryRoot};
2826
use asap_types::post_asap::SketchAlgorithm;
2927
use asap_types::types::AccuracyTarget;
@@ -130,12 +128,14 @@ fn stage_pipeline(workload: &PlanningWorkload, max_candidates: usize) -> Result<
130128
}))
131129
}
132130

133-
fn export(roots: &[QueryRoot]) -> Result<LogicalASAPDAG, String> {
134-
let dag = compile_logical_asap_workload(roots).map_err(|e| format!("export: {e}"))?;
135-
LogicalASAPDAGDocument::new(dag.clone())
136-
.validate()
137-
.map_err(|e| format!("export validation: {e}"))?;
138-
Ok(dag)
131+
/// A logical DAG as a flat node list: node `i` is `nodes[i]`, and children are
132+
/// node indices inside each operator.
133+
fn export(roots: &[QueryRoot]) -> Result<FlatDag, String> {
134+
for root in roots {
135+
root.validate_structure()
136+
.map_err(|e| format!("validation: {e}"))?;
137+
}
138+
Ok(flatten(roots).0)
139139
}
140140

141141
/// The first query whose plan reaches each target (enumeration order).

‎crates/devtools/tests/stage_pipeline.rs‎

Lines changed: 17 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -3,7 +3,8 @@ use std::collections::HashSet;
33
use std::path::PathBuf;
44
use std::process::Command;
55

6-
use asap_types::ir::export::{LogicalASAPDAG, LogicalASAPDAGDocument};
6+
use asap_types::ir::flat::FlatDag;
7+
use asap_types::ir::QueryRoot;
78
use serde_json::Value;
89

910
const COMMITTED: &str = "../../tools/dag-viewer/examples/planner-layering-example1.json";
@@ -27,12 +28,24 @@ fn generate(extra: &[&str]) -> Value {
2728
}
2829

2930
fn validated_dag(value: &Value, queries: usize) {
30-
let dag: LogicalASAPDAG = serde_json::from_value(value.clone()).unwrap();
31+
let dag: FlatDag = serde_json::from_value(value.clone()).unwrap();
3132
assert_eq!(dag.roots.len(), queries);
32-
LogicalASAPDAGDocument::new(dag).validate().unwrap();
33+
// Children come before their parents.
34+
for (id, node) in dag.nodes.iter().enumerate() {
35+
assert!(node
36+
.operator
37+
.children()
38+
.into_iter()
39+
.all(|child| *child < id));
40+
}
41+
for root in &dag.roots {
42+
if let QueryRoot::Operator(id) = root {
43+
assert!(*id < dag.nodes.len());
44+
}
45+
}
3346
}
3447

35-
/// Every exported DAG validates and has one root per query; ids are unique;
48+
/// Every DAG is a well-formed flat DAG with one root per query; ids are unique;
3649
/// only Stages 0 and 1 are present; the committed fixture is current.
3750
#[test]
3851
fn example1_document_is_valid_and_committed_fixture_is_current() {

0 commit comments

Comments
 (0)