refactor: remove dead and test-only code, part 1 (#1046)

crates/cost/src/resources.rs

@@ -1,47 +1,17 @@
-//! Resource dimensions shared by analytical estimates and benchmark evidence.
+//! Resource dimensions of analytical estimates.
 //!
 //! The CPU payload establishes its unit and operation scope; CPU operations
-//! must never be interpreted as nanoseconds. Byte values can be exact integers
-//! or measurements carrying uncertainty. `None` means unavailable, not zero.
+//! must never be interpreted as nanoseconds. `None` means unavailable, not zero.
 
 pub mod cpu;
-pub mod measurement;
 pub mod physical;
 
-pub use cpu::{MeasuredCpu, ModeledCpu};
-pub use measurement::Measurement;
-pub use physical::{MeasuredResources, PhysicalResources};
+pub use cpu::ModeledCpu;
+pub use physical::PhysicalResources;
 #[cfg(test)]
 mod tests {
     use super::*;
 
-    /// Distinct CPU payloads reject each other's units at the wire boundary;
-    /// measured phase metadata survives without filling unknown phases.
-    #[test]
-    fn cpu_units_and_measurement_metadata_remain_distinct() {
-        let modeled = ModeledCpu { cpu_ops: 12.0 };
-        assert!(
-            serde_json::from_value::<MeasuredCpu>(serde_json::to_value(modeled).unwrap()).is_err()
-        );
-        let measured = MeasuredCpu {
-            update_cpu_ns: Some(Measurement {
-                value: 12.0,
-                stddev: Some(0.5),
-                samples: 5,
-                method: Some("test process CPU clock".into()),
-            }),
-            ..Default::default()
-        };
-        let encoded = serde_json::to_value(&measured).unwrap();
-        assert!(serde_json::from_value::<ModeledCpu>(encoded.clone()).is_err());
-        assert_eq!(
-            serde_json::from_value::<MeasuredCpu>(encoded).unwrap(),
-            measured
-        );
-        assert!(measured.build_cpu_ns.is_none());
-        assert!(measured.read_cpu_ns.is_none());
-    }
-
     /// Default resources preserve unavailable dimensions instead of inventing zeros.
     #[test]
     fn missing_dimensions_remain_unavailable() {

crates/cost/src/resources/cpu.rs

@@ -1,26 +1,10 @@
-//! CPU quantities with explicit modeled-work or measured-time units.
+//! CPU quantities in modeled-work units.
 
 use serde::{Deserialize, Serialize};
 
-use super::measurement::Measurement;
-
 /// Modeled CPU work, never measured elapsed or process CPU time.
 #[derive(Debug, Clone, Copy, PartialEq, Serialize, Deserialize)]
 #[serde(deny_unknown_fields)]
 pub struct ModeledCpu {
     pub cpu_ops: f64,
 }
-
-/// Process CPU nanoseconds per named operation, never modeled CPU operations.
-/// An absent phase was not measured; it does not imply a free operation.
-#[derive(Debug, Clone, Default, PartialEq, Serialize, Deserialize)]
-#[serde(deny_unknown_fields)]
-pub struct MeasuredCpu {
-    /// Empty construction; ingestion is charged separately.
-    pub build_cpu_ns: Option<Measurement>,
-    pub update_cpu_ns: Option<Measurement>,
-    pub merge_cpu_ns: Option<Measurement>,
-    /// One prepare pass after ingestion and before reads.
-    pub prepare_cpu_ns: Option<Measurement>,
-    pub read_cpu_ns: Option<Measurement>,
-}

crates/cost/src/resources/physical.rs

@@ -2,10 +2,6 @@
 
 use serde::{Deserialize, Serialize};
 
-use super::{cpu::MeasuredCpu, measurement::Measurement};
-
-pub type MeasuredResources = PhysicalResources<MeasuredCpu, Measurement>;
-
 #[derive(Debug, Clone, Copy, PartialEq, Serialize, Deserialize)]
 #[serde(deny_unknown_fields)]
 pub struct PhysicalResources<Cpu, Bytes> {

crates/executor/src/operators/aggregate/temporal.rs

@@ -402,9 +402,11 @@ mod tests {
     use super::*;
     use crate::{
         operators::Operator,
-        runtime::{batch_execution::evaluate_batch, Limits, RunContext, Scope},
+        plan::PhysicalDAG,
+        runtime::{Limits, RunContext, Scope},
         values::Batch,
     };
+    use futures::StreamExt;
     use planner_types::ir::properties::AccuracyTarget;
     use planner_types::ir::schema::{
         DataType, Field as SummaryField, FieldDataType as SummaryFamilyType, Schema,
@@ -464,14 +466,19 @@ mod tests {
                 let operator =
                     Operator::window(schema.clone(), intent, 0, 1, vec![], Some((0, 2000)))
                         .unwrap();
-                let result = evaluate_batch(
-                    batch,
-                    vec![operator],
-                    RunContext::new(scope.clone(), Limits::default()).unwrap(),
+                let mut dag = PhysicalDAG::default();
+                dag.add(
+                    0,
+                    vec![],
+                    Operator::source(schema.clone(), vec![batch]).unwrap(),
                 )
                 .unwrap();
+                dag.add(1, vec![0], operator).unwrap();
+                let context = RunContext::new(scope.clone(), Limits::default()).unwrap();
+                let mut output = dag.execute(&[1], context).unwrap().remove(0);
+                let result = futures::executor::block_on(output.next()).unwrap().unwrap();
                 assert_eq!(
-                    format!("{:?}", result[0].to_rows()[0][0]),
+                    format!("{:?}", result.to_rows()[0][0]),
                     format!("{expected:?}")
                 );
             }

crates/executor/src/runtime/mod.rs

@@ -264,7 +264,6 @@ impl<V> Stream for Input<'_, V> {
     }
 }
 
-pub mod batch_execution;
 #[cfg(test)]
 mod tests;
 

crates/logical-asap-optimization/src/pass2/memo.rs

@@ -12,8 +12,6 @@ mod groups;
 pub(crate) mod rules;
 /// Shared finite state configuration enumeration.
 pub mod shared_configurations;
-#[cfg(test)]
-mod shared_producer_snapshot;
 /// Concrete summary-state groups exposed by sharing rules.
 pub mod state_groups;
 /// Local pane candidates with selected-reader grid configurations.