refactor: unify value and summary-state schemas

crates/asap-aware-mapping/src/accuracy/estimators/cms.rs

@@ -47,7 +47,7 @@ mod tests {
         let params = default_size_params(SketchAlgorithm::Cms, &c, 0.01, 0.001);
         let g = DefaultAccuracyModel
             .local_guarantee(
-                &SummaryFamilyType::Sketch(
+                &FieldDataType::Sketch(
                     SketchKind::new(SketchAlgorithm::Cms, params),
                     GroupingStrategy::default(),
                 ),
@@ -74,7 +74,7 @@ mod tests {
         };
         let topk_frequency = DefaultAccuracyModel
             .local_guarantee(
-                &SummaryFamilyType::Sketch(
+                &FieldDataType::Sketch(
                     SketchKind::new(SketchAlgorithm::CmsWithHeap, cms_heap),
                     GroupingStrategy::default(),
                 ),

crates/asap-aware-mapping/src/accuracy/estimators/count_sketch.rs

@@ -62,7 +62,7 @@ mod tests {
         let count_sketch = default_size_params(SketchAlgorithm::CountSketch, &intent, 0.01, 0.01);
         let guarantee = DefaultAccuracyModel
             .local_guarantee(
-                &SummaryFamilyType::Sketch(
+                &FieldDataType::Sketch(
                     SketchKind::new(SketchAlgorithm::CountSketch, count_sketch),
                     GroupingStrategy::default(),
                 ),

crates/asap-aware-mapping/src/accuracy/estimators/hll.rs

@@ -237,7 +237,7 @@ mod tests {
         let params = default_size_params(SketchAlgorithm::Hll, &c, 0.01, 0.01);
         let g = DefaultAccuracyModel
             .local_guarantee(
-                &SummaryFamilyType::Sketch(
+                &FieldDataType::Sketch(
                     SketchKind::new(SketchAlgorithm::Hll, params),
                     GroupingStrategy::default(),
                 ),

crates/asap-aware-mapping/src/accuracy/estimators/kll.rs

@@ -49,7 +49,7 @@ mod tests {
         let params = default_size_params(SketchAlgorithm::Kll, &q, 0.01, 0.01);
         let g = DefaultAccuracyModel
             .local_guarantee(
-                &SummaryFamilyType::Sketch(
+                &FieldDataType::Sketch(
                     SketchKind::new(SketchAlgorithm::Kll, params),
                     GroupingStrategy::default(),
                 ),

crates/asap-aware-mapping/src/accuracy/estimators/mod.rs

@@ -56,21 +56,19 @@ fn bounded_guarantee(
 }
 
 pub(super) fn local_guarantee(
-    family: &SummaryFamilyType,
+    family: &FieldDataType,
     query: &SketchQuery,
 ) -> Option<ResultGuarantee> {
     match family {
-        SummaryFamilyType::Plain(_) => Some(ResultGuarantee::exact("Plain value")),
-        SummaryFamilyType::ExactAggregate(kind, _) => {
+        FieldDataType::Plain(_) => Some(ResultGuarantee::exact("Plain value")),
+        FieldDataType::ExactAggregate(kind, _) => {
             Some(ResultGuarantee::exact(format!("ExactAggregate({kind:?})")))
         }
-        SummaryFamilyType::Sketch(kind, _) => {
-            sketch_guarantee(kind.algorithm(), kind.params(), query)
-        }
+        FieldDataType::Sketch(kind, _) => sketch_guarantee(kind.algorithm(), kind.params(), query),
         // No error model is registered for these families.
-        SummaryFamilyType::Sample(..)
-        | SummaryFamilyType::Wavelet(..)
-        | SummaryFamilyType::StatModel(..) => None,
+        FieldDataType::Sample(..) | FieldDataType::Wavelet(..) | FieldDataType::StatModel(..) => {
+            None
+        }
     }
 }
 pub(crate) fn size_params(
@@ -197,10 +195,10 @@ impl AccuracyModel for EstimatorAccuracy<'_> {
     }
     fn local_guarantee(
         &self,
-        family: &SummaryFamilyType,
+        family: &FieldDataType,
         query: &SketchQuery,
     ) -> Option<ResultGuarantee> {
-        if let (Some(_), SummaryFamilyType::Sketch(kind, grouping), SketchQuery::Cardinality) =
+        if let (Some(_), FieldDataType::Sketch(kind, grouping), SketchQuery::Cardinality) =
             (self.contract, family, query)
         {
             if let (SketchAlgorithm::Hll, SketchParams::Hll { precision }) =

crates/asap-aware-mapping/src/accuracy/evidence.rs

@@ -118,7 +118,7 @@ pub trait AccuracyEvidenceProvider {
     fn propagation_stats(
         &self,
         _op: &CompositionOperator,
-        _family: &SummaryFamilyType,
+        _family: &FieldDataType,
         _query: Option<&SketchQuery>,
     ) -> PropagationStats {
         PropagationStats::default()
@@ -142,7 +142,7 @@ impl AccuracyEvidenceProvider for WorkloadAccuracyEvidence<'_> {
     fn propagation_stats(
         &self,
         _op: &CompositionOperator,
-        _family: &SummaryFamilyType,
+        _family: &FieldDataType,
         _query: Option<&SketchQuery>,
     ) -> PropagationStats {
         PropagationStats {
@@ -180,7 +180,7 @@ mod tests {
         };
         let fresh = provider.propagation_stats(
             &CompositionOperator::ExactSum,
-            &SummaryFamilyType::ExactAggregate(
+            &FieldDataType::ExactAggregate(
                 asap_types::post_asap::ExactKind::Sum,
                 asap_types::post_asap::ExactParams::Sum,
             ),
@@ -195,7 +195,7 @@ mod tests {
         }
         .propagation_stats(
             &CompositionOperator::ExactSum,
-            &SummaryFamilyType::ExactAggregate(
+            &FieldDataType::ExactAggregate(
                 asap_types::post_asap::ExactKind::Sum,
                 asap_types::post_asap::ExactParams::Sum,
             ),

crates/asap-aware-mapping/src/accuracy/mod.rs

@@ -21,9 +21,8 @@ pub use evidence::{
 };
 
 use asap_types::post_asap::{
-    AccuracyError, BoundExpr, CompositionOperator, ErrorMetric, ExactOperation, GuaranteeSource,
-    ProbabilityExpr, ResultGuarantee, SketchAlgorithm, SketchParams, SketchQuery,
-    SummaryFamilyType,
+    AccuracyError, BoundExpr, CompositionOperator, ErrorMetric, ExactOperation, FieldDataType,
+    GuaranteeSource, ProbabilityExpr, ResultGuarantee, SketchAlgorithm, SketchParams, SketchQuery,
 };
 use asap_types::types::AccuracyTarget;
 
@@ -48,7 +47,7 @@ pub trait AccuracyModel {
     /// `Sample`/`Wavelet`/`StatModel`).
     fn local_guarantee(
         &self,
-        family: &SummaryFamilyType,
+        family: &FieldDataType,
         query: &SketchQuery,
     ) -> Option<ResultGuarantee>;
 
@@ -94,7 +93,7 @@ impl AccuracyModel for DefaultAccuracyModel {
     }
     fn local_guarantee(
         &self,
-        family: &SummaryFamilyType,
+        family: &FieldDataType,
         query: &SketchQuery,
     ) -> Option<ResultGuarantee> {
         estimators::local_guarantee(family, query)

crates/asap-aware-mapping/src/accuracy/reconciliation.rs

@@ -394,7 +394,7 @@ mod tests {
     use asap_types::post_asap::SketchAlgorithm;
     use asap_types::pre_asap::cse::share_common_sub_dags;
     use asap_types::pre_asap::query_expr::{GroupKeys, Source};
-    use asap_types::pre_asap::schema::{Column, ColumnId, DataType, Schema};
+    use asap_types::pre_asap::schema::{ColumnId, DataType, Field, Schema};
 
     /// `[ts(0), value(1), job(2)]`.
     /// A unique-keyed scan (`[ts]`) so `share_common_sub_dags` is actually
@@ -407,9 +407,9 @@ mod tests {
             predicates: vec![],
             schema: Schema::with_time_index(
                 vec![
-                    Column::new("ts", DataType::Timestamp, false),
-                    Column::new("value", DataType::Float64, false),
-                    Column::new("job", DataType::Utf8, true),
+                    Field::plain("ts", DataType::Timestamp, false),
+                    Field::plain("value", DataType::Float64, false),
+                    Field::plain("job", DataType::Utf8, true),
                 ],
                 0,
                 vec![vec![0]],

crates/asap-aware-mapping/src/cost_model.rs

@@ -49,8 +49,8 @@
 use std::rc::Rc;
 
 use asap_types::post_asap::{
-    ExactOperation, GroupingStrategy, HydraParams, ResultGuarantee, SketchAlgorithm, SketchParams,
-    SketchQuery, SummaryExpr, SummaryFamilyType, SummaryMaintenanceLifecycleGuarantee, SummaryNode,
+    ExactOperation, FieldDataType, GroupingStrategy, HydraParams, ResultGuarantee, SketchAlgorithm,
+    SketchParams, SketchQuery, SummaryExpr, SummaryMaintenanceLifecycleGuarantee, SummaryNode,
     SummaryWindowFramework,
 };
 use asap_types::pre_asap::agg_intent::AggIntent;
@@ -278,7 +278,7 @@ pub struct CseCandidate<'a> {
     /// The shared pre-ASAP sub-DAG itself.
     pub sub_dag: &'a QueryExpr,
     /// The `SummaryNode` this sub-DAG bound to — gives the cost model the
-    /// concrete `SummaryFamilyType`/`(kind, params)` actually at stake, not
+    /// concrete `FieldDataType`/`(kind, params)` actually at stake, not
     /// just the pre-ASAP shape.
     pub bound_summary: &'a SummaryNode,
     /// How many workload roots reference this exact shared sub-DAG, counted
@@ -387,7 +387,7 @@ pub fn default_cse_recompute_cost(sub_dag: &QueryExpr) -> Cost {
 }
 
 /// Default [`CostModel::cse_shared_maintenance_cost`]: a small
-/// per-[`SummaryFamilyType`] weight, scaled to the same order of magnitude
+/// per-[`FieldDataType`] weight, scaled to the same order of magnitude
 /// as [`default_cse_recompute_cost`]'s typical output (a small node
 /// count, not a byte length), reflecting that families differ in how
 /// expensive they are to keep *continuously updated* for the life of a
@@ -398,15 +398,15 @@ pub fn default_cse_recompute_cost(sub_dag: &QueryExpr) -> Cost {
 /// deployment with real memory/update-cost numbers should override
 /// [`CostModel::cse_shared_maintenance_cost`] instead of relying on this
 /// table.
-pub fn default_cse_shared_maintenance_cost(family: &SummaryFamilyType) -> Cost {
+pub fn default_cse_shared_maintenance_cost(family: &FieldDataType) -> Cost {
     const UNIT: f64 = 1.0;
     let weight = match family {
-        SummaryFamilyType::Plain(_) => 1.0,
-        SummaryFamilyType::ExactAggregate(..) => 1.0,
-        SummaryFamilyType::Sketch(..) => 3.0,
-        SummaryFamilyType::Sample(..) => 3.0,
-        SummaryFamilyType::Wavelet(..) => 5.0,
-        SummaryFamilyType::StatModel(..) => 6.0,
+        FieldDataType::Plain(_) => 1.0,
+        FieldDataType::ExactAggregate(..) => 1.0,
+        FieldDataType::Sketch(..) => 3.0,
+        FieldDataType::Sample(..) => 3.0,
+        FieldDataType::Wavelet(..) => 5.0,
+        FieldDataType::StatModel(..) => 6.0,
     };
     Cost(weight * UNIT)
 }
@@ -590,9 +590,9 @@ pub trait CostModel {
             .fields
             .iter()
             .map(|f| &f.dtype)
-            .find(|dtype| !matches!(dtype, SummaryFamilyType::Plain(_)))
+            .find(|dtype| !matches!(dtype, FieldDataType::Plain(_)))
             .cloned()
-            .unwrap_or(SummaryFamilyType::Plain(
+            .unwrap_or(FieldDataType::Plain(
                 asap_types::pre_asap::DataType::Float64,
             ));
         default_cse_shared_maintenance_cost(&family)
@@ -953,7 +953,7 @@ fn sketch_state(
     match &node.expr {
         SummaryExpr::SummaryEstimate { summary_input, .. } => sketch_state(summary_input),
         SummaryExpr::SummaryAgg {
-            family: SummaryFamilyType::Sketch(kind, grouping),
+            family: FieldDataType::Sketch(kind, grouping),
             ..
         } => Some((kind, grouping)),
         _ => None,
@@ -1347,36 +1347,32 @@ mod tests {
     // ── CSE sharing (issue #237, #223 stage 4) ──────────────────────────
 
     use asap_types::post_asap::{
-        ExactKind, ExactParams, GroupingStrategy, SketchKind, SummaryExpr, SummaryField,
-        SummarySchema,
+        ExactKind, ExactParams, Field, GroupingStrategy, Schema, SketchKind, SummaryExpr,
     };
     use asap_types::pre_asap::query_expr::Source;
-    use asap_types::pre_asap::schema::{Column, DataType, Schema};
+    use asap_types::pre_asap::schema::DataType;
 
     fn scan() -> QueryExpr {
         QueryExpr::Scan {
             source: Source::TimeSeries { metric: "m".into() },
             predicates: vec![],
             schema: Schema::with_time_index(
                 vec![
-                    Column::new("ts", DataType::Timestamp, false),
-                    Column::new("value", DataType::Float64, false),
+                    Field::plain("ts", DataType::Timestamp, false),
+                    Field::plain("value", DataType::Float64, false),
                 ],
                 0,
                 vec![],
             ),
         }
     }
 
-    fn summary_node(family: SummaryFamilyType) -> SummaryNode {
+    fn summary_node(family: FieldDataType) -> SummaryNode {
         SummaryNode {
             expr: SummaryExpr::SummaryAgg {
                 child: std::rc::Rc::new(SummaryNode {
                     expr: SummaryExpr::KeepPreAsap(Rc::new(scan())),
-                    schema: SummarySchema {
-                        fields: vec![],
-                        time_index: None,
-                    },
+                    schema: Schema::lifted(vec![], None),
                     guarantee: None,
                 }),
                 family: family.clone(),
@@ -1387,14 +1383,7 @@ mod tests {
                 grouping: GroupingStrategy::default(),
                 filter: None,
             },
-            schema: SummarySchema {
-                fields: vec![SummaryField {
-                    name: "state".into(),
-                    dtype: family,
-                    nullable: false,
-                }],
-                time_index: None,
-            },
+            schema: Schema::lifted(vec![Field::new("state", family, false)], None),
             guarantee: None,
         }
     }
@@ -1455,11 +1444,11 @@ mod tests {
 
     #[test]
     fn default_shared_maintenance_cost_orders_families_cheapest_to_priciest() {
-        let exact = default_cse_shared_maintenance_cost(&SummaryFamilyType::ExactAggregate(
+        let exact = default_cse_shared_maintenance_cost(&FieldDataType::ExactAggregate(
             ExactKind::Sum,
             ExactParams::Sum,
         ));
-        let sketch = default_cse_shared_maintenance_cost(&SummaryFamilyType::Sketch(
+        let sketch = default_cse_shared_maintenance_cost(&FieldDataType::Sketch(
             SketchKind::new(SketchAlgorithm::Hll, SketchParams::Hll { precision: 12 }),
             GroupingStrategy::default(),
         ));
@@ -1474,7 +1463,7 @@ mod tests {
     fn cse_share_decision_shares_when_recompute_dominates_maintenance() {
         let candidate = CseCandidate {
             sub_dag: &scan(),
-            bound_summary: &summary_node(SummaryFamilyType::ExactAggregate(
+            bound_summary: &summary_node(FieldDataType::ExactAggregate(
                 ExactKind::Sum,
                 ExactParams::Sum,
             )),
@@ -1492,7 +1481,7 @@ mod tests {
     fn cse_share_decision_recomputes_when_maintenance_dominates_recompute() {
         let candidate = CseCandidate {
             sub_dag: &scan(),
-            bound_summary: &summary_node(SummaryFamilyType::StatModel(
+            bound_summary: &summary_node(FieldDataType::StatModel(
                 asap_types::post_asap::StatModelKind::Parametric,
                 asap_types::post_asap::StatModelParams::Parametric {
                     family: "gaussian_mixture".into(),
@@ -1532,7 +1521,7 @@ mod tests {
         // hardcoding a comparison against its own defaults.
         let candidate = CseCandidate {
             sub_dag: &scan(),
-            bound_summary: &summary_node(SummaryFamilyType::StatModel(
+            bound_summary: &summary_node(FieldDataType::StatModel(
                 asap_types::post_asap::StatModelKind::Parametric,
                 asap_types::post_asap::StatModelParams::Parametric {
                     family: "gaussian_mixture".into(),
@@ -1570,7 +1559,7 @@ mod tests {
         let target = TargetSubDAG::new(&root);
         let candidate = ReplacementSubDAG {
             strategy: "TestStrategy",
-            replacement: Replacement::Summary(Rc::new(summary_node(SummaryFamilyType::Plain(
+            replacement: Replacement::Summary(Rc::new(summary_node(FieldDataType::Plain(
                 asap_types::pre_asap::DataType::Float64,
             )))),
             provenance: crate::replacement::ReplacementProvenance::SummaryRealization,
@@ -1595,14 +1584,14 @@ mod tests {
         let cheap = ReplacementSubDAG {
             strategy: "TestStrategy",
             replacement: Replacement::Summary(Rc::new(summary_node(
-                SummaryFamilyType::ExactAggregate(ExactKind::Sum, ExactParams::Sum),
+                FieldDataType::ExactAggregate(ExactKind::Sum, ExactParams::Sum),
             ))),
             provenance: crate::replacement::ReplacementProvenance::SummaryRealization,
             rationale: "exact accumulator".into(),
         };
         let pricey = ReplacementSubDAG {
             strategy: "TestStrategy",
-            replacement: Replacement::Summary(Rc::new(summary_node(SummaryFamilyType::StatModel(
+            replacement: Replacement::Summary(Rc::new(summary_node(FieldDataType::StatModel(
                 asap_types::post_asap::StatModelKind::Parametric,
                 asap_types::post_asap::StatModelParams::Parametric {
                     family: "gaussian_mixture".into(),

crates/asap-aware-mapping/src/empirical_cost.rs

@@ -3,7 +3,7 @@
 //! of an accuracy guarantee. CPU quantities are nanoseconds, never CPU operations.
 
 use asap_types::post_asap::{
-    GroupingStrategy, SketchAlgorithm, SketchParams, SummaryExpr, SummaryFamilyType, SummaryNode,
+    FieldDataType, GroupingStrategy, SketchAlgorithm, SketchParams, SummaryExpr, SummaryNode,
 };
 use asap_types::pre_asap::AggIntent;
 use serde::{Deserialize, Serialize};
@@ -217,7 +217,7 @@ impl EmpiricalEvidenceProvider {
         summary: &SummaryNode,
     ) -> SummaryMaintenanceLifecycleCostInputs {
         let SummaryExpr::SummaryAgg {
-            family: SummaryFamilyType::Sketch(kind, GroupingStrategy::PerSubpopulationInstance),
+            family: FieldDataType::Sketch(kind, GroupingStrategy::PerSubpopulationInstance),
             grouping: GroupingStrategy::PerSubpopulationInstance,
             ..
         } = &summary.expr