Skip to content

Commit ef19aea

Browse files
zzylolclaude
andcommitted
refactor(ir): use SummaryCoverage names in summary merges
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
1 parent a83fc91 commit ef19aea

4 files changed

Lines changed: 28 additions & 36 deletions

File tree

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

Lines changed: 6 additions & 14 deletions
Original file line numberDiff line numberDiff line change
@@ -6,6 +6,7 @@ use std::rc::Rc;
66
use serde::{Deserialize, Serialize};
77

88
use super::node::{OperatorNode, OperatorResultKind};
9+
use super::summary_coverage::{CoverageError, SummaryCoverage};
910
use crate::ir::operator_properties::Reduction;
1011
use crate::ir::SchemaDerivationError;
1112
use crate::post_asap::maintained_population::{MaintainedPopulation, PopulationStatistic};
@@ -213,27 +214,18 @@ impl ASAPOp {
213214
}
214215
}
215216

216-
/// Derive joint observation coverage; unknown or overlapping inputs fail closed.
217-
pub fn merged_extent(
218-
&self,
219-
) -> Result<super::observation_extent::ObservationExtent, SchemaDerivationError> {
217+
/// Derive the merged node's coverage; unknown or overlapping inputs fail closed.
218+
pub fn merged_coverage(&self) -> Result<SummaryCoverage, SchemaDerivationError> {
220219
let ASAPOp::SummaryMerge { children } = self else {
221220
return Err(SchemaDerivationError::InvalidScalarSignature(
222221
"coverage merge requires SummaryMerge".into(),
223222
));
224223
};
225224
let inputs = children
226225
.iter()
227-
.map(|child| {
228-
child.observation_extent.clone().ok_or_else(|| {
229-
SchemaDerivationError::InvalidScalarSignature(
230-
"summary merge requires known observation coverage".into(),
231-
)
232-
})
233-
})
226+
.map(|child| child.coverage.clone().ok_or(CoverageError::UnknownInput))
234227
.collect::<Result<Vec<_>, _>>()?;
235-
super::observation_extent::ObservationExtent::merge_disjoint(&inputs)
236-
.map_err(|error| SchemaDerivationError::InvalidScalarSignature(error.to_string()))
228+
Ok(SummaryCoverage::merge_disjoint(&inputs)?)
237229
}
238230

239231
/// Output schema derived from the operator and its children. Summary
@@ -506,7 +498,7 @@ impl ASAPOp {
506498
));
507499
}
508500
}
509-
self.merged_extent()?;
501+
self.merged_coverage()?;
510502
Ok(())
511503
}
512504
SummaryEstimate {

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

Lines changed: 7 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -112,7 +112,7 @@ impl OperatorNode {
112112
let schema = operator.output_schema()?;
113113
let mut node = Self::with_schema(operator, schema);
114114
if let Some(op @ ASAPOp::SummaryMerge { .. }) = node.asap() {
115-
node.observation_extent = Some(op.merged_extent()?);
115+
node.coverage = Some(op.merged_coverage()?);
116116
}
117117
Ok(node)
118118
}
@@ -165,7 +165,10 @@ impl OperatorNode {
165165

166166
/// Summary nodes whose state can be composed must declare coverage.
167167
pub fn requires_coverage(&self) -> bool {
168-
matches!(self.asap(), Some(ASAPOp::SummaryAgg { .. }))
168+
matches!(
169+
self.asap(),
170+
Some(ASAPOp::SummaryAgg { .. } | ASAPOp::SummaryMerge { .. })
171+
)
169172
}
170173

171174
pub fn non_asap(&self) -> Option<&NonASAPOp> {
@@ -325,10 +328,8 @@ impl OperatorNode {
325328
None => {}
326329
}
327330
if let Some(op @ ASAPOp::SummaryMerge { .. }) = node.asap() {
328-
if node.observation_extent.as_ref() != Some(&op.merged_extent()?) {
329-
return Err(SchemaDerivationError::InvalidScalarSignature(
330-
"retained merge coverage disagrees with input union".into(),
331-
));
331+
if node.coverage.as_ref() != Some(&op.merged_coverage()?) {
332+
return Err(CoverageError::MergeOutputMismatch.into());
332333
}
333334
}
334335
node.operator.validate_inputs()?;

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

Lines changed: 4 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -48,6 +48,10 @@ pub enum CoverageError {
4848
NotState,
4949
#[error("summary node requires coverage")]
5050
Missing,
51+
#[error("summary merge requires known coverage on every input")]
52+
UnknownInput,
53+
#[error("retained merge coverage disagrees with input union")]
54+
MergeOutputMismatch,
5155
}
5256

5357
impl SummaryCoverage {

‎crates/types/tests/summary_merge_structure.rs‎

Lines changed: 11 additions & 16 deletions
Original file line numberDiff line numberDiff line change
@@ -29,16 +29,12 @@ fn state(k: u32) -> Rc<OperatorNode> {
2929
.unwrap();
3030
std::rc::Rc::new(
3131
summary
32-
.with_observation_extent(asap_types::ir::observation_extent::ObservationExtent {
32+
.with_coverage(asap_types::ir::summary_coverage::SummaryCoverage {
3333
source: "latencies:timestamp".into(),
34-
revision: "fixture-1".into(),
3534
input: SummaryUpdate::column(ColumnRef::SampleValue),
36-
grouping: Reduction::by(vec![]),
37-
multiplicity:
38-
asap_types::ir::observation_extent::ObservationMultiplicity::OncePerObservation,
39-
regions: vec![asap_types::ir::observation_extent::ExtentRegion {
40-
start_ms: 0,
41-
end_ms: 1,
35+
reduction: Reduction::by(vec![]),
36+
regions: vec![asap_types::ir::summary_coverage::CoverageRegion {
37+
time_ms: Some(0..1),
4238
population: Default::default(),
4339
}],
4440
})
@@ -55,8 +51,8 @@ fn compatible_panes_merge_structurally() {
5551
root.validate_structure().unwrap();
5652
assert_eq!(root.schema.fields.len(), 1);
5753
assert_eq!(
58-
root.observation_extent.as_ref().unwrap().regions[0].end_ms,
59-
2
54+
root.coverage.as_ref().unwrap().regions[0].time_ms,
55+
Some(0..2)
6056
);
6157
}
6258
/// An empty merge, raw rows and differently sized state cannot masquerade as compatible panes.
@@ -75,16 +71,15 @@ fn incompatible_merge_inputs_fail() {
7571

7672
fn shifted_state(k: u32, start: i64, end: i64) -> Rc<OperatorNode> {
7773
let mut node = (*state(k)).clone();
78-
let region = &mut node.observation_extent.as_mut().unwrap().regions[0];
79-
region.start_ms = start;
80-
region.end_ms = end;
74+
let region = &mut node.coverage.as_mut().unwrap().regions[0];
75+
region.time_ms = Some(start..end);
8176
Rc::new(node)
8277
}
8378
/// Schema equality cannot authorize overlapping or unknown observation coverage.
8479
#[test]
8580
fn unsafe_coverage_merge_is_rejected() {
8681
let mut unknown = (*state(200)).clone();
87-
unknown.observation_extent = None;
82+
unknown.coverage = None;
8883
for children in [
8984
vec![state(200), state(200)],
9085
vec![state(200), Rc::new(unknown)],
@@ -102,8 +97,8 @@ fn merge_derives_coverage_and_validates_retained_metadata() {
10297
children: vec![state(200), shifted_state(200, 2, 3)],
10398
}))
10499
.unwrap();
105-
assert_eq!(root.observation_extent.as_ref().unwrap().regions.len(), 2);
100+
assert_eq!(root.coverage.as_ref().unwrap().regions.len(), 2);
106101
let mut forged = (*root).clone();
107-
forged.observation_extent.as_mut().unwrap().regions[0].end_ms = 2;
102+
forged.coverage.as_mut().unwrap().regions[0].time_ms = Some(0..2);
108103
assert!(Rc::new(forged).validate_structure().is_err());
109104
}

0 commit comments

Comments
 (0)