refactor(ir): follow derived summary coverage in timing and physical export

Coverage is derived from the node (#646), so timed copies no longer carry
it and PhysicalASAPDAGNode drops its coverage field.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01W7qG9aFyPij5uWsyAJCxDW

Author: zzylol

crates/types/src/ir/physical_export.rs

@@ -67,8 +67,6 @@ pub struct PhysicalASAPDAGNode {
     pub output_state: ExecutionDataState,
     pub output_schema: Schema,
     pub guarantee: Option<ResultGuarantee>,
-    #[serde(default)]
-    pub coverage: Option<super::summary_coverage::SummaryCoverage>,
 }
 
 #[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
@@ -413,7 +411,6 @@ pub fn compile_physical_asap_workload_with_node_ids(
             output_state,
             output_schema: flat_node.schema,
             guarantee: flat_node.guarantee,
-            coverage: flat_node.coverage,
         });
     }
     let dag = PhysicalASAPDAG {

crates/types/src/ir/timing.rs

@@ -264,15 +264,11 @@ fn write(
     if let Some(e) = error {
         return Err(e);
     }
-    let timed = Rc::new(OperatorNode {
-        operator,
-        result_kind: node.result_kind,
-        schema: node.schema.clone(),
-        guarantee: node.guarantee.clone(),
-        timing: Some(timing),
-        // Timing copies the same logical sub-DAG; its observations are unchanged.
-        coverage: node.coverage.clone(),
-    });
+    let timed = Rc::new(
+        OperatorNode::with_schema(operator, node.schema.clone())
+            .with_guarantee(node.guarantee.clone())
+            .with_timing(Some(timing)),
+    );
     memo.done.insert(Rc::as_ptr(node), Rc::clone(&timed));
     Ok(timed)
 }
@@ -618,14 +614,11 @@ pub fn split_shared_by_phase(
             rebuilt
         });
         let out = if changed {
-            Rc::new(OperatorNode {
-                operator,
-                result_kind: node.result_kind,
-                schema: node.schema.clone(),
-                guarantee: node.guarantee.clone(),
-                timing: node.timing,
-                coverage: node.coverage.clone(),
-            })
+            Rc::new(
+                OperatorNode::with_schema(operator, node.schema.clone())
+                    .with_guarantee(node.guarantee.clone())
+                    .with_timing(node.timing),
+            )
         } else {
             Rc::clone(node)
         };

crates/types/tests/physical_export.rs

@@ -1,8 +1,7 @@
-//! A timed plan exports as a PhysicalASAPDAG that keeps timing and coverage.
+//! A timed plan exports as a PhysicalASAPDAG that keeps timing.
 use asap_types::ir::physical_export::{
     compile_physical_asap_dag, PhysicalASAPDAGDocument, PhysicalASAPDAGValidationError,
 };
-use asap_types::ir::summary_coverage::{CoverageRegion, SummaryCoverage};
 use asap_types::ir::{
     apply_lifecycle_timings, ASAPOp, LifecycleAssignment, NonASAPOp, Operator, OperatorNode,
     TimingMemo,
@@ -13,11 +12,10 @@ use std::rc::Rc;
 
 /// Scan(t) → SummaryAgg(sum by key) → FinalizeExactAccumulator, untimed.
 fn plan() -> Rc<OperatorNode> {
-    let source = Source::Table {
-        table_ref: "t".into(),
-    };
     let scan = OperatorNode::new_shared(Operator::NonASAP(NonASAPOp::Scan {
-        source: source.clone(),
+        source: Source::Table {
+            table_ref: "t".into(),
+        },
         predicates: vec![],
         schema: Schema::new(vec![
             Field::plain("key", DataType::Utf8, false),
@@ -33,14 +31,6 @@ fn plan() -> Rc<OperatorNode> {
         grouping: Default::default(),
         filter: None,
     }))
-    .unwrap()
-    .with_coverage(SummaryCoverage {
-        source,
-        regions: vec![CoverageRegion {
-            time_ms: Some(0..60_000),
-            population: Default::default(),
-        }],
-    })
     .unwrap();
     OperatorNode::new_shared(Operator::ASAP(ASAPOp::FinalizeExactAccumulator {
         child: Rc::new(state),
@@ -50,7 +40,7 @@ fn plan() -> Rc<OperatorNode> {
 
 /// Default lifecycle: the summary is maintained at ingestion time and read at query time.
 #[test]
-fn timed_plan_exports_with_timing_and_coverage() {
+fn timed_plan_exports_with_timing() {
     let timed = apply_lifecycle_timings(
         &plan(),
         &LifecycleAssignment::default_maintained(),
@@ -70,7 +60,8 @@ fn timed_plan_exports_with_timing_and_coverage() {
             ExecutionTiming::QueryTime
         ]
     );
-    assert!(dag.nodes[1].coverage.is_some());
+    // Timing keeps what the summary covers.
+    assert!(timed.children()[0].coverage().is_some());
 
     let decoded: PhysicalASAPDAGDocument =
         serde_json::from_str(&serde_json::to_string(&document).unwrap()).unwrap();