test: port the Stage 1 callers to Operator<NodeId> payloads

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

Author: zzylol

crates/executor/tests/deployment_computation.rs

@@ -1,7 +1,6 @@
 //! Planner-selected PromQL computation compiles from the timed DAG alone;
 //! the deployment supplies only raw rows at the ingestion frontier.
 mod common;
-use planner_types::ir::ASAPOp;
 use asap_executor::{
     operators::Operator,
     physical_planner::{compile, promql_rows, CompiledPhysicalDAG, InputContract, Source},
@@ -11,6 +10,7 @@ use asap_executor::{
 use common::{compile_physical_asap_dag, selected_dag, stage1_candidates};
 use futures::{executor::block_on, StreamExt};
 use planner_types::ir::physical_export::{PhysicalASAPDAG, PhysicalASAPOperatorPayload};
+use planner_types::ir::ASAPOp;
 use planner_types::{ir::schema::*, types::AccuracyTarget, workload::*};
 use std::{collections::BTreeMap, rc::Rc, sync::Arc};
 
@@ -71,14 +71,18 @@ fn raw_inputs(dag: &PhysicalASAPDAG) -> Vec<(u64, Arc<Schema>, String)> {
     dag.nodes
         .iter()
         .filter_map(|node| match &node.payload {
-            PhysicalASAPOperatorPayload::NonASAP(planner_types::ir::NonASAPOp::TimeRange { .. }) => {
+            PhysicalASAPOperatorPayload::NonASAP(planner_types::ir::NonASAPOp::TimeRange {
+                ..
+            }) => {
                 let mut id = node.id;
                 loop {
                     let n = dag.nodes.iter().find(|n| n.id == id)?;
-                    if let PhysicalASAPOperatorPayload::NonASAP(planner_types::ir::NonASAPOp::Scan {
-                                source: planner_types::ir::operator::Source::TimeSeries { metric },
-                                ..
-                            }) = &n.payload
+                    if let PhysicalASAPOperatorPayload::NonASAP(
+                        planner_types::ir::NonASAPOp::Scan {
+                            source: planner_types::ir::operator::Source::TimeSeries { metric },
+                            ..
+                        },
+                    ) = &n.payload
                     {
                         return Some((
                             node.id as u64,
@@ -327,9 +331,10 @@ fn with_kind(
 ) -> PhysicalASAPDAG {
     for node in &mut dag.nodes {
         if let PhysicalASAPOperatorPayload::NonASAP(planner_types::ir::NonASAPOp::BinaryOp {
-                    operator,
-                    return_bool,
-                }) = &mut node.payload
+            operator,
+            return_bool,
+            ..
+        }) = &mut node.payload
         {
             operator.kind = kind.clone();
             *return_bool = bool_result;
@@ -526,9 +531,10 @@ fn with_vector_match(
 ) -> PhysicalASAPDAG {
     for node in &mut dag.nodes {
         if let PhysicalASAPOperatorPayload::NonASAP(planner_types::ir::NonASAPOp::BinaryOp {
-                    operator,
-                    return_bool: _,
-                }) = &mut node.payload
+            operator,
+            return_bool: _,
+            ..
+        }) = &mut node.payload
         {
             operator.vector_match = Some(planner_types::ir::operator::VectorMatch {
                 kind: kind.clone(),
@@ -614,7 +620,8 @@ fn stored_count_min_bare_count_compiles_to_a_evaluation() {
                 matches!(
                     &n.payload,
                     PhysicalASAPOperatorPayload::ASAP(ASAPOp::SummaryEstimate {
-                        query: SketchStatistic::PointCount { value: None, .. }
+                        query: SketchStatistic::PointCount { value: None, .. },
+                        ..
                     })
                 )
             });
@@ -629,7 +636,12 @@ fn stored_count_min_bare_count_compiles_to_a_evaluation() {
     let state = dag
         .nodes
         .iter()
-        .find(|n| matches!(n.payload, PhysicalASAPOperatorPayload::ASAP(ASAPOp::SummaryAgg { .. })))
+        .find(|n| {
+            matches!(
+                n.payload,
+                PhysicalASAPOperatorPayload::ASAP(ASAPOp::SummaryAgg { .. })
+            )
+        })
         .unwrap();
     let PhysicalASAPOperatorPayload::ASAP(ASAPOp::SummaryAgg {
         family: FieldDataType::Sketch(kind, _),
@@ -644,10 +656,7 @@ fn stored_count_min_bare_count_compiles_to_a_evaluation() {
     let schema = Arc::new(state.output_schema.clone());
     let program = compile(
         &dag,
-        BTreeMap::from([(
-            state.id as u64,
-            InputContract::bounded(schema.clone()),
-        )]),
+        BTreeMap::from([(state.id as u64, InputContract::bounded(schema.clone()))]),
         &[dag.roots[0] as u64],
     )
     .unwrap();

crates/executor/tests/weighted_topk_binding.rs

@@ -1,6 +1,5 @@
 //! Planner output binds directly to the shared runtime at a declared rate-value frontier.
 mod common;
-use planner_types::ir::ASAPOp;
 use asap_executor::dag::{
     operators::Operator,
     planner::{compile, InputContract, Source},
@@ -12,6 +11,7 @@ use futures::{executor::block_on, StreamExt};
 use planner_types::ir::physical_export::{PhysicalASAPDAG, PhysicalASAPOperatorPayload};
 use planner_types::ir::properties::*;
 use planner_types::ir::schema::{DataType, *};
+use planner_types::ir::ASAPOp;
 use planner_types::types::AccuracyTarget;
 use std::{collections::BTreeMap, rc::Rc, sync::Arc};
 #[test]
@@ -106,18 +106,18 @@ fn assert_weighted_binding(algorithm: SketchAlgorithm) {
             as Source<'static>;
         let compiled = compile(
             &placed,
-            BTreeMap::from([(rate_id.0 as u64, InputContract::bounded(rates.clone()))]),
-            &[dag.roots[0].0 as u64],
+            BTreeMap::from([(rate_id as u64, InputContract::bounded(rates.clone()))]),
+            &[dag.roots[0] as u64],
         )
         .unwrap();
         let physical_dag = compiled
-            .instantiate(BTreeMap::from([(rate_id.0 as u64, source)]))
+            .instantiate(BTreeMap::from([(rate_id as u64, source)]))
             .unwrap();
         let context = RunContext::new(scope, Limits::default()).unwrap();
         let output = block_on(async {
             let mut output = Vec::new();
             let mut stream = physical_dag
-                .execute(&[dag.roots[0].0 as u64], context)
+                .execute(&[dag.roots[0] as u64], context)
                 .unwrap()
                 .remove(0);
             while let Some(batch) = stream.next().await {
@@ -287,17 +287,16 @@ fn direct_rate_topk_preserves_dynamic_unreferenced_labels() {
         .find(|node| {
             matches!(
                 &node.payload,
-                PhysicalASAPOperatorPayload::NonASAP(planner_types::ir::NonASAPOp::TimeRange { .. })
+                PhysicalASAPOperatorPayload::NonASAP(
+                    planner_types::ir::NonASAPOp::TimeRange { .. }
+                )
             )
         })
         .unwrap_or_else(|| panic!("no raw counter source: {dag:?}"));
     let raw_schema = Arc::new(raw.output_schema.clone());
     let raw_compiled = compile(
         &dag,
-        BTreeMap::from([(
-            raw.id as u64,
-            InputContract::bounded(raw_schema.clone()),
-        )]),
+        BTreeMap::from([(raw.id as u64, InputContract::bounded(raw_schema.clone()))]),
         &[dag.roots[0] as u64],
     )
     .unwrap();
@@ -407,10 +406,7 @@ fn direct_rate_topk_preserves_dynamic_unreferenced_labels() {
     }
     let compiled = compile(
         &dag,
-        BTreeMap::from([(
-            input_id as u64,
-            InputContract::bounded(schema.clone()),
-        )]),
+        BTreeMap::from([(input_id as u64, InputContract::bounded(schema.clone()))]),
         &[dag.roots[0] as u64],
     )
     .unwrap();
@@ -510,7 +506,12 @@ fn continuously_maintained_dag(candidate: &Rc<planner_types::ir::OperatorNode>)
     let mut pending: Vec<_> = dag
         .nodes
         .iter()
-        .filter(|node| matches!(node.payload, PhysicalASAPOperatorPayload::ASAP(ASAPOp::SummaryAgg { .. })))
+        .filter(|node| {
+            matches!(
+                node.payload,
+                PhysicalASAPOperatorPayload::ASAP(ASAPOp::SummaryAgg { .. })
+            )
+        })
         .map(|node| node.id)
         .collect();
     let mut ingestion = std::collections::HashSet::new();

crates/integration-tests/tests/precompute_raw_samples.rs

@@ -93,7 +93,12 @@ fn candidates(query: &str, accuracy: AccuracyTarget) -> Vec<Rc<OperatorNode>> {
 fn raw_summaries(dag: &PhysicalASAPDAG) -> Vec<(u64, u64)> {
     dag.nodes
         .iter()
-        .filter(|node| matches!(node.payload, PhysicalASAPOperatorPayload::ASAP(ASAPOp::SummaryAgg { .. })))
+        .filter(|node| {
+            matches!(
+                node.payload,
+                PhysicalASAPOperatorPayload::ASAP(ASAPOp::SummaryAgg { .. })
+            )
+        })
         .filter_map(|node| {
             let inputs = dag
                 .edges
@@ -314,11 +319,7 @@ fn check(
     root: u64,
     rows: &[(Series, i64, f64)],
 ) -> Result<String, String> {
-    let node = dag
-        .nodes
-        .iter()
-        .find(|n| n.id as u64 == root)
-        .unwrap();
+    let node = dag.nodes.iter().find(|n| n.id as u64 == root).unwrap();
     let PhysicalASAPOperatorPayload::ASAP(ASAPOp::SummaryAgg {
         family,
         input,
@@ -329,11 +330,7 @@ fn check(
     else {
         unreachable!()
     };
-    let source_node = dag
-        .nodes
-        .iter()
-        .find(|n| n.id as u64 == source)
-        .unwrap();
+    let source_node = dag.nodes.iter().find(|n| n.id as u64 == source).unwrap();
     let keys = match reduction {
         Reduction::Reduce(keys) => keys
             .keys()
@@ -506,11 +503,7 @@ fn grouped_raw_summary(family: FieldDataType, input: SummaryUpdate) -> (Physical
     .unwrap();
     let mut dag = compile_maintained_physical_asap_dag(&candidate).unwrap();
     let (source, root) = raw_summaries(&dag)[0];
-    let node = dag
-        .nodes
-        .iter_mut()
-        .find(|n| n.id as u64 == root)
-        .unwrap();
+    let node = dag.nodes.iter_mut().find(|n| n.id as u64 == root).unwrap();
     let PhysicalASAPOperatorPayload::ASAP(ASAPOp::SummaryAgg {
         family: old,
         input: update,
@@ -527,11 +520,7 @@ fn grouped_raw_summary(family: FieldDataType, input: SummaryUpdate) -> (Physical
     *old = family;
     *update = input;
     let schema = node.output_schema.clone();
-    for edge in dag
-        .edges
-        .iter_mut()
-        .filter(|e| u64::from(e.producer.0) == root)
-    {
+    for edge in dag.edges.iter_mut().filter(|e| e.producer as u64 == root) {
         edge.intermediate_schema = schema.clone();
     }
     (dag, source, root)
@@ -638,12 +627,9 @@ fn raw_sample_without_grouping_drops_labels_and_name() {
         .iter()
         .position(|f| f.name == "service")
         .unwrap();
-    let node = dag
-        .nodes
-        .iter_mut()
-        .find(|n| n.id as u64 == root)
-        .unwrap();
-    let PhysicalASAPOperatorPayload::ASAP(ASAPOp::SummaryAgg { reduction, .. }) = &mut node.payload else {
+    let node = dag.nodes.iter_mut().find(|n| n.id as u64 == root).unwrap();
+    let PhysicalASAPOperatorPayload::ASAP(ASAPOp::SummaryAgg { reduction, .. }) = &mut node.payload
+    else {
         unreachable!()
     };
     *reduction = Reduction::Reduce(GroupKeys::without(vec![service]));