feat(control_plane): Phase 2 step 1 -- merge schema.rs onto asap-ir

control_plane/src/intent_algebra/agg_intent.rs

@@ -8,6 +8,12 @@
 //! ASAPController has no tagged releases yet) and holds only what's
 //! genuinely control_plane-specific:
 //!
+//! (Phase 2: `Column`/`DataType` are also re-exported from `asap_ir` now
+//! — `schema.rs` got the full swap, not a boundary conversion, since
+//! asap_ir's version turned out to be a purely additive, backward-
+//! compatible superset. `output_column` below no longer needs a
+//! conversion layer as a result.)
+//!
 //! - **`frequency()` / `as_frequency()`** — control_plane's standalone
 //!   point-frequency-via-CMS query (`count(*) WHERE key = k`), carried
 //!   through the shared `AggIntent` as an `Extension` rather than a
@@ -43,7 +49,6 @@
 //! in scope, not off the intent.
 
 use asap_ir::intent_algebra::agg_accuracy as asap_agg_accuracy;
-use asap_ir::intent_algebra::schema::{Column as AsapColumn, DataType as AsapDataType};
 pub use asap_ir::intent_algebra::{
     agg_is_exact, agg_is_mergeable, default_cardinality, default_quantile,
     is_frequency_heavy_hitter, ranking_measure, AggIntent, MathFunc, RankingMeasure, TimeFunc,
@@ -54,43 +59,6 @@ use crate::types_v2::AccuracyTarget;
 
 const FREQUENCY_EXT_KIND: &str = "frequency";
 
-/// `control_plane::Column`/`DataType` and `asap_ir::Column`/`DataType`
-/// are structurally identical but not the same type — merging `schema.rs`
-/// itself is Phase 2 scope (it cascades into `Schema`/`QueryExpr`, used
-/// pervasively; ~38 `Column{}` literals across this repo). Convert at
-/// this boundary instead of widening this change.
-fn to_asap_dtype(dt: &DataType) -> AsapDataType {
-    match dt {
-        DataType::Int64 => AsapDataType::Int64,
-        DataType::Float64 => AsapDataType::Float64,
-        DataType::Utf8 => AsapDataType::Utf8,
-        DataType::Bool => AsapDataType::Bool,
-        DataType::Timestamp => AsapDataType::Timestamp,
-    }
-}
-
-fn from_asap_dtype(dt: &AsapDataType) -> DataType {
-    match dt {
-        AsapDataType::Int64 => DataType::Int64,
-        AsapDataType::Float64 => DataType::Float64,
-        AsapDataType::Utf8 => DataType::Utf8,
-        AsapDataType::Bool => DataType::Bool,
-        AsapDataType::Timestamp => DataType::Timestamp,
-    }
-}
-
-fn to_asap_column(c: &Column) -> AsapColumn {
-    AsapColumn::new(c.name.clone(), to_asap_dtype(&c.dtype), c.nullable)
-}
-
-fn from_asap_column(c: AsapColumn) -> Column {
-    Column {
-        name: c.name,
-        dtype: from_asap_dtype(&c.dtype),
-        nullable: c.nullable,
-    }
-}
-
 /// Construct control_plane's point-frequency-via-CMS intent. See module
 /// docs for why this is an `Extension`, not a shared first-class variant.
 pub fn frequency(accuracy: AccuracyTarget) -> AggIntent {
@@ -190,11 +158,7 @@ pub fn archive_only(intent: &AggIntent) -> bool {
 /// `"frequency"` — that's control_plane-only knowledge).
 pub fn output_column(intent: &AggIntent, input: &Column) -> Column {
     if as_frequency(intent).is_some() {
-        return Column {
-            name: "frequency".into(),
-            dtype: DataType::Int64,
-            nullable: false,
-        };
+        return Column::new("frequency", DataType::Int64, false);
     }
-    from_asap_column(intent.output_column(&to_asap_column(input)))
+    intent.output_column(input)
 }

control_plane/src/intent_algebra/binder.rs

@@ -137,6 +137,7 @@ impl<C: SchemaCatalog> Binder<C> {
                     name,
                     dtype: DataType::Utf8, // labels / group keys are strings
                     nullable: true,
+                    table: None,
                 });
             }
         }
@@ -146,6 +147,7 @@ impl<C: SchemaCatalog> Binder<C> {
             columns,
             time_index,
             unique_keys: Vec::new(),
+            closed: false,
         }
     }
 }
@@ -157,11 +159,13 @@ fn default_leaf_columns() -> Vec<Column> {
             name: "ts".into(),
             dtype: DataType::Timestamp,
             nullable: false,
+            table: None,
         },
         Column {
             name: "value".into(),
             dtype: DataType::Float64,
             nullable: false,
+            table: None,
         },
     ]
 }
@@ -269,16 +273,19 @@ mod tests {
                             name: "ts".into(),
                             dtype: DataType::Timestamp,
                             nullable: false,
+                            table: None,
                         },
                         Column {
                             name: "value".into(),
                             dtype: DataType::Float64,
                             nullable: false,
+                            table: None,
                         },
                         Column {
                             name: "datacenter".into(),
                             dtype: DataType::Utf8,
                             nullable: false,
+                            table: None,
                         },
                     ])
                 } else {

control_plane/src/intent_algebra/column_resolution.rs

@@ -91,11 +91,13 @@ pub fn infer_source_schema(_metric_or_table_name: &str) -> Schema {
                 name: "ts".into(),
                 dtype: DataType::Timestamp,
                 nullable: false,
+                table: None,
             },
             Column {
                 name: "value".into(),
                 dtype: DataType::Float64,
                 nullable: false,
+                table: None,
             },
         ],
         0,
@@ -265,6 +267,7 @@ pub fn output_schema_for_aggregate(input: &Schema, by: &[ColumnId], aggs: &[AggI
             name: "value".into(),
             dtype: DataType::Float64,
             nullable: false,
+            table: None,
         });
     for intent in aggs {
         out_cols.push(crate::intent_algebra::output_column(intent, &probe));
@@ -280,6 +283,7 @@ pub fn output_schema_for_aggregate(input: &Schema, by: &[ColumnId], aggs: &[AggI
         columns: out_cols,
         time_index: None,
         unique_keys,
+        closed: false,
     }
 }
 
@@ -402,11 +406,13 @@ mod tests {
             name: "host".into(),
             dtype: DataType::Utf8,
             nullable: false,
+            table: None,
         });
         input.columns.push(Column {
             name: "region".into(),
             dtype: DataType::Utf8,
             nullable: false,
+            table: None,
         });
         // Group by host, region (positions 2 and 3).
         let by = vec![2usize, 3usize];
@@ -439,6 +445,7 @@ mod tests {
             name: "host".into(),
             dtype: DataType::Utf8,
             nullable: false,
+            table: None,
         });
         let ids = resolve_named_keys(&["host".to_string()], &s).unwrap();
         assert_eq!(ids, vec![2usize]);

control_plane/src/intent_algebra/cse.rs

@@ -170,6 +170,7 @@ mod tests {
             name: name.into(),
             dtype,
             nullable: false,
+            table: None,
         }
     }
 

control_plane/src/intent_algebra/query_expr.rs

@@ -560,6 +560,7 @@ impl QueryExpr {
                         name: "value".into(),
                         dtype: DataType::Float64,
                         nullable: false,
+                        table: None,
                     });
                 for intent in aggs {
                     out_cols.push(crate::intent_algebra::output_column(intent, &probe));
@@ -578,6 +579,7 @@ impl QueryExpr {
                     columns: out_cols,
                     time_index: None,
                     unique_keys,
+                    closed: false,
                 })
             }
             QueryExpr::LetBinding { name, expr, child } => {
@@ -794,6 +796,7 @@ mod tests {
             name: name.into(),
             dtype,
             nullable: false,
+            table: None,
         }
     }