Skip to content

Commit 0ebeb9d

Browse files
zzylolclaude
andauthored
feat(ir): define compatible logical summary merges (#560)
* feat(ir): define compatible logical summary merges * feat(ir): require disjoint coverage for summary merges * refactor(ir): use SummaryCoverage names in summary merges Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> * feat(ir): check merge update/reduction on producers; test documented coverage examples SummaryCoverage no longer repeats input/reduction, so SummaryMerge compares them through OperatorNode::summary_update. summary_coverage_examples.rs builds each example in docs/develop_docs/summary-coverage.md as a SummaryAgg -> SummaryMerge plan. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> * test(ir): point coverage examples at the design document Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> * test(ir): point coverage examples at the ASAP primitive schema design doc Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> * feat(ir): reject summary merges of heap-based sketches A merged top-k heap can miss an item that is heavy in only one input, so CmsWithHeap, CountSketchWithHeap and UnivMon states do not merge. Also correct the merge_disjoint doc: SummaryMerge checks schema, update, reduction and heap families, not accuracy. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> * refactor(ir): compute merge coverage with SummaryCoverage::of_merge `ASAPOp::merged_coverage` only applied to SummaryMerge and returned an error for every other operator. Replace it with `SummaryCoverage::of_merge`, which takes the merge's inputs; the callers already have them. Also explain what `OperatorNode::summary_update` returns and why merging compares it. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> * refactor(ir): leave summary merge coverage to #646; allow heap-based sketches SummaryMerge no longer derives or checks coverage: of_merge, UnknownInput and MergeOutputMismatch are removed and summary_coverage.rs matches main. Coverage for all summary nodes will be derived by one function in #646. The heap-based sketch restriction is also dropped. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> * refactor(ir): rename summary_update to summary_input_data It reads the producing SummaryAgg's update expression and reduction; it does not update the summary. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> * feat(ir): record summarized columns in SummaryCoverage SummaryCoverage now records which columns a state summarizes as well as which rows: `input` (the SummaryAgg update expression) and `group_by` (its reduction). with_coverage rejects a SummaryAgg declaration whose columns differ from the node's own (ColumnMismatch), and SummaryMerge requires coverage on every input (UnknownInput) with identical columns. merge_disjoint checks columns too. summary_input_data is removed. A nested SummaryMerge carries no coverage until #646, so it is rejected. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> * docs(ir): attach the merge_disjoint doc comment to merge_disjoint Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> * refactor(ir): keep SummaryMerge structural; leave coverage to #646 Remove the coverage columns (input, group_by), check_columns and the merge's coverage checks. SummaryMerge now only checks structure: at least one input, every input is State with one state column and an identical schema. Whether a structurally valid merge is semantically valid (same computation, disjoint selections) is decided by summary coverage in #646, following the design in #573. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01W7qG9aFyPij5uWsyAJCxDW --------- Co-authored-by: Claude Opus 5.5 <noreply@anthropic.com>
1 parent f394faa commit 0ebeb9d

2 files changed

Lines changed: 143 additions & 8 deletions

File tree

‎crates/types/src/ir/asap.rs‎

Lines changed: 50 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -54,10 +54,11 @@ pub enum ASAPOp<C = Rc<OperatorNode>> {
5454
child: C,
5555
evaluation: PopulationStatistic,
5656
},
57-
// ── Reserved: migrated but unimplemented (§1.3 of the proposal) ──
57+
/// Merge compatible partial states for the same grouping and family.
5858
SummaryMerge {
5959
children: Vec<C>,
6060
},
61+
// ── Reserved: migrated but unimplemented ──
6162
SummarySubtract {
6263
left: C,
6364
right: C,
@@ -192,11 +193,7 @@ impl ASAPOp {
192193
use ASAPOp::*;
193194
matches!(
194195
self,
195-
SummaryMerge { .. }
196-
| SummarySubtract { .. }
197-
| SummaryDelete { .. }
198-
| SummaryJoin { .. }
199-
| Extension { .. }
196+
SummarySubtract { .. } | SummaryDelete { .. } | SummaryJoin { .. } | Extension { .. }
200197
)
201198
}
202199

@@ -208,6 +205,14 @@ impl ASAPOp {
208205
pub fn produced_state(&self) -> Option<&FieldDataType> {
209206
match self {
210207
ASAPOp::SummaryAgg { family, .. } | ASAPOp::SummaryJoin { family, .. } => Some(family),
208+
ASAPOp::SummaryMerge { children } => children.first().and_then(|child| {
209+
child
210+
.schema
211+
.fields
212+
.iter()
213+
.find(|field| !field.is_plain())
214+
.map(|field| &field.dtype)
215+
}),
211216
_ => None,
212217
}
213218
}
@@ -411,8 +416,11 @@ impl ASAPOp {
411416
.output_schema()?
412417
}
413418
}
414-
SummaryMerge { .. }
415-
| SummarySubtract { .. }
419+
SummaryMerge { children } => {
420+
self.validate_inputs()?;
421+
children[0].schema.clone()
422+
}
423+
SummarySubtract { .. }
416424
| SummaryDelete { .. }
417425
| SummaryJoin { .. }
418426
| Extension { .. } => return Err(Self::unimplemented()),
@@ -450,6 +458,40 @@ impl ASAPOp {
450458
}
451459
};
452460
match self {
461+
SummaryMerge { children } => {
462+
let Some(first) = children.first() else {
463+
return Err(SchemaDerivationError::InvalidScalarSignature(
464+
"summary merge requires at least one state input".into(),
465+
));
466+
};
467+
// Matching state parameters and grouping positions are necessary;
468+
// matching names alone cannot prove two states compatible.
469+
if first
470+
.schema
471+
.fields
472+
.iter()
473+
.filter(|field| !field.is_plain())
474+
.count()
475+
!= 1
476+
{
477+
return Err(SchemaDerivationError::InvalidScalarSignature(
478+
"summary merge requires exactly one state column".into(),
479+
));
480+
}
481+
for child in children {
482+
needs_state(child, "SummaryMerge")?;
483+
if child.schema != first.schema {
484+
return Err(SchemaDerivationError::InvalidScalarSignature(
485+
"summary merge inputs must have identical state and grouping schemas"
486+
.into(),
487+
));
488+
}
489+
}
490+
// Equal schemas cannot tell a KLL over `latency` from one over
491+
// `size`, nor prove the inputs disjoint; summary coverage (#646)
492+
// decides whether a structurally valid merge is semantically valid.
493+
Ok(())
494+
}
453495
SummaryEstimate {
454496
summary_input,
455497
query,
Lines changed: 93 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,93 @@
1+
//! Window composition merges compatible summary states without consuming raw rows.
2+
use asap_types::{
3+
ir::operator_properties::{Reduction, Source},
4+
ir::{ASAPOp, NonASAPOp, Operator, OperatorNode, SchemaDerivationError},
5+
post_asap::{GroupingStrategy, SketchAlgorithm, SketchKind, SketchParams, SummaryUpdate},
6+
pre_asap::{ColumnRef, DataType, Field, FieldDataType, Schema},
7+
};
8+
use std::rc::Rc;
9+
fn state(k: u32) -> Rc<OperatorNode> {
10+
state_with(FieldDataType::Sketch(
11+
SketchKind::new(SketchAlgorithm::Kll, SketchParams::Kll { k }),
12+
Default::default(),
13+
))
14+
}
15+
16+
fn state_with(family: FieldDataType) -> Rc<OperatorNode> {
17+
let scan = OperatorNode::new_shared(Operator::NonASAP(NonASAPOp::Scan {
18+
source: Source::Table {
19+
table_ref: "latencies".into(),
20+
},
21+
predicates: vec![],
22+
schema: Schema::new(vec![Field::plain("value", DataType::Float64, false)]),
23+
}))
24+
.unwrap();
25+
let summary = OperatorNode::new(Operator::ASAP(ASAPOp::SummaryAgg {
26+
child: scan,
27+
family,
28+
input: SummaryUpdate::column(ColumnRef::SampleValue),
29+
reduction: Reduction::by(vec![]),
30+
grouping: GroupingStrategy::default(),
31+
filter: None,
32+
}))
33+
.unwrap();
34+
std::rc::Rc::new(
35+
summary
36+
.with_coverage(asap_types::ir::summary_coverage::SummaryCoverage {
37+
source: Source::Table {
38+
table_ref: "latencies".into(),
39+
},
40+
regions: vec![asap_types::ir::summary_coverage::CoverageRegion {
41+
time_ms: Some(0..1),
42+
population: Default::default(),
43+
}],
44+
})
45+
.unwrap(),
46+
)
47+
}
48+
/// Two KLL panes compose into one typed logical state without timing assignment.
49+
#[test]
50+
fn compatible_panes_merge_structurally() {
51+
let root = OperatorNode::new_shared(Operator::ASAP(ASAPOp::SummaryMerge {
52+
children: vec![state(200), shifted_state(200, 1, 2)],
53+
}))
54+
.unwrap();
55+
root.validate_structure().unwrap();
56+
assert_eq!(root.schema.fields.len(), 1);
57+
}
58+
/// An empty merge, raw rows and differently sized state cannot masquerade as compatible panes.
59+
#[test]
60+
fn incompatible_merge_inputs_fail() {
61+
for children in [
62+
vec![],
63+
vec![state(200), state(300)],
64+
vec![state(200).children()[0].clone()],
65+
] {
66+
assert!(
67+
OperatorNode::new_shared(Operator::ASAP(ASAPOp::SummaryMerge { children })).is_err()
68+
);
69+
}
70+
}
71+
72+
fn shifted_state(k: u32, start: i64, end: i64) -> Rc<OperatorNode> {
73+
shifted(&state(k), start, end)
74+
}
75+
76+
fn shifted(state: &Rc<OperatorNode>, start: i64, end: i64) -> Rc<OperatorNode> {
77+
let mut node = (**state).clone();
78+
let region = &mut node.coverage.as_mut().unwrap().regions[0];
79+
region.time_ms = Some(start..end);
80+
Rc::new(node)
81+
}
82+
83+
fn merge(children: Vec<Rc<OperatorNode>>) -> Result<Rc<OperatorNode>, SchemaDerivationError> {
84+
OperatorNode::new_shared(Operator::ASAP(ASAPOp::SummaryMerge { children }))
85+
}
86+
87+
/// A merge is a state, so merges nest structurally.
88+
#[test]
89+
fn merges_nest() {
90+
let inner = merge(vec![state(200)]).unwrap();
91+
let outer = merge(vec![inner, shifted_state(200, 1, 2)]).unwrap();
92+
outer.validate_structure().unwrap();
93+
}

0 commit comments

Comments
 (0)