feat(core): model schema completeness (Schema.closed, open/closed) — Calcite DynamicRecordType pattern

crates/core/src/intent_algebra/binder.rs

@@ -18,6 +18,15 @@ use crate::intent_algebra::schema::{Column, DataType, Schema};
 
 /// The DB / source-schema metadata source — resolves a source (metric /
 /// table) name to its known columns.
+/// Source of truth for a source's columns — the "catalog". `SqlCatalog` backs
+/// it for SQL; PromQL uses [`UsageDerivedCatalog`] (returns `None`) until a
+/// registry-backed impl (returning a metric's known label set) drops in here.
+/// Distinct from `Scan.schema`, which is the *resolved* binding schema this
+/// feeds — the catalog is the input, the schema is the result. Even a
+/// registry-backed PromQL catalog yields an **open** schema
+/// ([`Schema::closed`](super::schema::Schema::closed) `= false`): a metric's
+/// labels are per-series and time-varying, so the registry is a superset hint,
+/// not a per-row contract.
 pub trait SchemaCatalog {
     /// Columns known for `source`. `None` when unknown — the [`Binder`] then
     /// falls back to a usage-derived column set.
@@ -88,6 +97,9 @@ impl<C: SchemaCatalog> Binder<C> {
             columns,
             time_index,
             unique_keys: Vec::new(),
+            // Usage-derived (schemaless PromQL): the metric's full label set is
+            // open and runtime-only, so this lists only what the query references.
+            closed: false,
         }
     }
 }

crates/core/src/intent_algebra/column_resolution.rs

@@ -194,6 +194,9 @@ pub fn output_schema_for_aggregate(
         columns: out_cols,
         time_index: None,
         unique_keys,
+        // A cross-series aggregate fully determines its output columns, so the
+        // result is closed even over an open input (mirrors `output_schema_in`).
+        closed: true,
     }
 }
 

crates/core/src/intent_algebra/query_expr.rs

@@ -208,9 +208,15 @@ pub struct ProjectItem {
 
 #[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
 pub enum QueryExpr {
-    /// Outermost leaf. `schema` is the authoritative, self-contained output
-    /// schema (Binder-built); `predicates` are leaf-level row filters
-    /// (PromQL label matchers, pushed-down `WHERE` conjuncts).
+    /// Outermost leaf. `schema` is the **binding schema** — the resolved column
+    /// set every positional `ColumnId` in the tree indexes into, *not* a full
+    /// description of the runtime row. Complete when catalog-backed (SQL); for
+    /// schemaless PromQL it is usage-derived by the [`Binder`](super::binder)
+    /// (the `(ts, value)` floor + the labels the query references), since a
+    /// metric's label set is open and known only at runtime. That distinction is
+    /// carried explicitly by [`Schema::closed`](super::schema::Schema::closed)
+    /// (SQL leaf → `true`, PromQL leaf → `false`). `predicates` are leaf-level
+    /// row filters (PromQL label matchers, pushed-down `WHERE` conjuncts).
     Scan {
         source: Source,
         #[serde(default)]
@@ -431,6 +437,10 @@ impl QueryExpr {
                     columns: out_cols,
                     time_index: None,
                     unique_keys,
+                    // A cross-series aggregate enumerates exactly `by ++ aggs`,
+                    // so its output is closed even over an open input — this is
+                    // where an open schema freezes to closed.
+                    closed: true,
                 })
             }
 
@@ -475,6 +485,8 @@ impl QueryExpr {
                     columns,
                     time_index,
                     unique_keys: Vec::new(),
+                    // Projection enumerates exactly its items → closed.
+                    closed: true,
                 })
             }
 
@@ -528,6 +540,8 @@ impl QueryExpr {
                     columns,
                     time_index,
                     unique_keys: Vec::new(),
+                    // The concatenation is complete only if both sides are.
+                    closed: l.closed && r.closed,
                 })
             }
             // ψ-analytic — child schema + one appended window-output column.
@@ -591,6 +605,9 @@ fn per_series_reduction_schema(input: &Schema, agg: &AggIntent) -> Schema {
         columns,
         time_index: input.time_index,
         unique_keys: input.unique_keys.clone(),
+        // Per-series reduction is label-preserving: it inherits its input's
+        // completeness (an open scan stays open; a closed one stays closed).
+        closed: input.closed,
     }
 }
 
@@ -704,6 +721,7 @@ mod tests {
                 columns,
                 time_index,
                 unique_keys: uk,
+                closed: true,
             },
         }
     }
@@ -835,6 +853,55 @@ mod tests {
         );
     }
 
+    #[test]
+    fn completeness_open_leaf_freezes_to_closed_at_cross_series_aggregate() {
+        // A schemaless (PromQL-style) leaf is *open*; it stays open through a
+        // per-series reduction (`rate`), then is **frozen to closed** by a
+        // cross-series aggregate (which enumerates exactly its output columns).
+        let open_leaf = QueryExpr::Scan {
+            source: Source::TimeSeries { metric: "m".into() },
+            predicates: vec![],
+            // `with_time_index` defaults to `closed: false` (open).
+            schema: Schema::with_time_index(
+                vec![
+                    col("ts", DataType::Timestamp, false),
+                    col("value", DataType::Float64, false),
+                    col("job", DataType::Utf8, true),
+                ],
+                0,
+                vec![],
+            ),
+        };
+        assert!(
+            !open_leaf.output_schema().unwrap().closed,
+            "schemaless leaf is open"
+        );
+
+        let rate = QueryExpr::Aggregate {
+            by: vec![],
+            aggs: vec![AggIntent::Rate],
+            output_names: vec![],
+            having: None,
+            child: Box::new(open_leaf),
+        };
+        assert!(
+            !rate.output_schema().unwrap().closed,
+            "per-series rate is label-preserving → stays open"
+        );
+
+        let sum_by_job = QueryExpr::Aggregate {
+            by: vec![2], // `job`
+            aggs: vec![AggIntent::Sum { col: None }],
+            output_names: vec![],
+            having: None,
+            child: Box::new(rate),
+        };
+        assert!(
+            sum_by_job.output_schema().unwrap().closed,
+            "cross-series aggregate enumerates `by ++ aggs` → frozen to closed"
+        );
+    }
+
     #[test]
     fn project_keeps_time_index_when_ts_passed_through() {
         let child = scan(

crates/core/src/intent_algebra/schema.rs

@@ -113,6 +113,24 @@ pub struct Schema {
     /// "no provable unique constraint" (the conservative default).
     #[serde(default)]
     pub unique_keys: Vec<Vec<ColumnId>>,
+    /// Whether this schema **completely enumerates** the columns at this point.
+    ///
+    /// - `true` (**closed**): there are no columns beyond these — a catalog-backed
+    ///   SQL source, or an output fully determined by an `Aggregate`/`Project`.
+    /// - `false` (**open**): a dynamic / superset schema — the runtime row may
+    ///   carry more columns than are listed (a schemaless PromQL leaf lists only
+    ///   the `(ts, value)` floor + the labels the query references).
+    ///
+    /// This mirrors Apache Calcite's `DynamicRecordType` (schema-on-read): the
+    /// schema starts open at a schemaless leaf and is **frozen to closed** by the
+    /// first operator that fully determines its output columns (`Aggregate` /
+    /// `Project`). Consumers needing completeness (label validation, full-output
+    /// enumeration, cardinality for the cost model) must check this; positional
+    /// resolution does not care. **Invariant: open ⇒ do not apply closed-world
+    /// validation** (PromQL tolerates unknown labels). Defaults to `false` (open)
+    /// — the conservative choice when completeness is unknown.
+    #[serde(default)]
+    pub closed: bool,
 }
 
 impl Schema {
@@ -124,6 +142,7 @@ impl Schema {
             columns,
             time_index: None,
             unique_keys: Vec::new(),
+            closed: false,
         }
     }
 
@@ -138,6 +157,7 @@ impl Schema {
             columns,
             time_index: Some(time_index),
             unique_keys,
+            closed: false,
         }
     }
 
@@ -359,6 +379,17 @@ mod tests {
         assert_eq!(s, back);
     }
 
+    #[test]
+    fn schema_closed_defaults_to_open_when_absent() {
+        // `closed` is `#[serde(default)]` so schemas serialized before the field
+        // existed deserialize to `closed: false` (open) — the conservative
+        // default (don't claim completeness you can't prove).
+        let mut v = serde_json::to_value(Schema::new(vec![col("a", DataType::Utf8)])).unwrap();
+        assert!(v.as_object_mut().unwrap().remove("closed").is_some());
+        let back: Schema = serde_json::from_value(v).unwrap();
+        assert!(!back.closed, "absent `closed` ⇒ open");
+    }
+
     #[test]
     fn column_table_defaults_to_none_when_absent() {
         // `Column.table` is `#[serde(default)]` so schemas serialized before the

crates/e2e/src/lib.rs

@@ -33,6 +33,9 @@ pub mod fixtures {
             columns: cols,
             time_index: Some(0),
             unique_keys: vec![],
+            // Schemaless PromQL leaf: open (the metric's full label set is
+            // runtime-only; this lists just the referenced labels).
+            closed: false,
         }
     }
 }

crates/lower/src/sql/mod.rs

@@ -151,6 +151,8 @@ impl<'a> SqlLowerer<'a> {
                 .collect(),
             time_index: schema.time_index,
             unique_keys: schema.unique_keys.clone(),
+            // Catalog-backed: the table's columns are fully declared → closed.
+            closed: true,
         };
         Ok(L2::Source(SourceSpec::with_schema(
             table.to_string(),

crates/lower/tests/promql_conformance.rs

@@ -137,6 +137,21 @@ fn instant_vector_selector() {
     assert_eq!(preds, 0, "no label matchers → no predicates");
 }
 
+#[test]
+fn promql_scan_schema_is_open() {
+    // A schemaless PromQL leaf is *open*: the metric's full label set is
+    // runtime-only, so the binding schema lists only the (ts, value) floor +
+    // referenced labels and may be a subset of the runtime row.
+    let qe = ok("node_cpu_seconds_total");
+    let QueryExpr::Scan { schema, .. } = &qe else {
+        panic!("expected a Scan for a bare selector, got {qe:?}");
+    };
+    assert!(
+        !schema.closed,
+        "a schemaless PromQL scan has an open schema"
+    );
+}
+
 #[test]
 fn label_matchers_become_scan_predicates() {
     // SEMANTICS: `=`, `!=`, `=~`, `!~` filter series; one conjunct per matcher.

crates/lower/tests/sql_lowering.rs

@@ -100,13 +100,19 @@ async fn select_star_with_where_folds_predicate_onto_scan() {
     // SELECT * elides the projection; WHERE folds onto the Scan predicates.
     let qe = lower("SELECT * FROM metrics WHERE service = 'api'").await;
     let QueryExpr::Scan {
-        source, predicates, ..
+        source,
+        predicates,
+        schema,
     } = &qe
     else {
         panic!("expected Scan at root, got {qe:?}");
     };
     assert!(matches!(source, Source::Table { table_ref } if table_ref == "metrics"));
     assert_eq!(predicates.len(), 1, "WHERE clause folded onto the scan");
+    assert!(
+        schema.closed,
+        "a catalog-backed SQL scan has a closed schema"
+    );
 }
 
 #[tokio::test]