diff --git a/crates/asap-physical-operators/src/expressions/planner.rs b/crates/asap-physical-operators/src/expressions/planner.rs index 92042fa8e..ae04b3f7c 100644 --- a/crates/asap-physical-operators/src/expressions/planner.rs +++ b/crates/asap-physical-operators/src/expressions/planner.rs @@ -69,7 +69,7 @@ pub(super) fn evaluate( Value::Null ))), QueryExpr::FunctionCall { name, args } => { - use planner_types::pre_asap::scalar_signature::MapScalarFunction; + use planner_types::pre_asap::scalar_type_rules::MapScalarFunction; if name.eq_ignore_ascii_case("asap_struct_field") { expr.scalar_type(schema) .map_err(|error| Error::Invalid(error.to_string()))?; @@ -478,7 +478,7 @@ fn validate(expr: &QueryExpr, schema: &planner_types::pre_asap::Schema) -> Resul QueryExpr::FunctionCall { name, args } => { if name != "asap_struct_field" && name != "asap_element_access" - && planner_types::pre_asap::scalar_signature::MapScalarFunction::from_name(name) + && planner_types::pre_asap::scalar_type_rules::MapScalarFunction::from_name(name) .is_none() { return Err(invalid()); diff --git a/crates/frontend-sql/src/sql/collection_planning.rs b/crates/frontend-sql/src/sql/collection_planning.rs index 05f309470..dd7106564 100644 --- a/crates/frontend-sql/src/sql/collection_planning.rs +++ b/crates/frontend-sql/src/sql/collection_planning.rs @@ -1,7 +1,7 @@ //! DataFusion planning adapters. Types come from the canonical signature rules; //! physical evaluation deliberately remains the query engine's responsibility. use super::types::{arrow_to_dtype, dtype_to_arrow, scalar_value_to_asap}; -use asap_types::pre_asap::scalar_signature::{ +use asap_types::pre_asap::scalar_type_rules::{ element_access_type, struct_field_type, MapScalarFunction, }; use asap_types::pre_asap::{Field, QueryExpr, Schema}; diff --git a/crates/frontend-sql/src/sql/types.rs b/crates/frontend-sql/src/sql/types.rs index 282d07711..1d178daf4 100644 --- a/crates/frontend-sql/src/sql/types.rs +++ b/crates/frontend-sql/src/sql/types.rs @@ -364,7 +364,7 @@ mod bottom_map_tests { use super::*; #[test] fn empty_map_bottom_types_roundtrip_without_string_defaults() { - let (map, nullable) = asap_types::pre_asap::scalar_signature::MapScalarFunction::Construct + let (map, nullable) = asap_types::pre_asap::scalar_type_rules::MapScalarFunction::Construct .output_type(&[]) .unwrap(); assert!(!nullable); diff --git a/crates/types/src/ir/aggregate_schema.rs b/crates/types/src/ir/aggregate_schema.rs new file mode 100644 index 000000000..10b33d2ae --- /dev/null +++ b/crates/types/src/ir/aggregate_schema.rs @@ -0,0 +1,359 @@ +//! Derive aggregate output columns and types from input schema and reduction. +//! +//! [`aggregate_output_schema`] handles grouping keys and aggregate results. +//! For example, SQL `GROUP BY host` retains the grouping column and adds the +//! aggregate result; a PromQL per-series range reduction preserves labels +//! and produces a Float64 sample value. This module derives schemas, not +//! aggregate values or summary candidates. +use super::operator_properties::*; +use super::SchemaDerivationError; +use crate::pre_asap::{AggIntent, ColumnId, ColumnRef, DataType, Field, FieldDataType, Schema}; +/// Output schema of a *per-series* window/range reduction (`rate`/`increase`, +/// or an `*_over_time` reducer under a time `Window`). Such a reduction emits +/// one value per series, so every label column of `input` is preserved and only +/// the sample value is replaced — kept named `value` so the PromQL sample-value +/// convention (and any outer `SampleValue` reference) still resolves it by name. +fn per_series_reduction_schema( + input: &Schema, + agg: &AggIntent, +) -> Result { + let vi = if let Some(index) = agg.input_cols().first() { + *index + } else { + crate::pre_asap::column_resolution::resolve_column_ref(&ColumnRef::SampleValue, input) + .map_err(|error| SchemaDerivationError::InvalidSampleColumn(error.to_string()))? + }; + if !matches!( + input.fields.get(vi).map(|column| &column.dtype), + Some(FieldDataType::Plain(DataType::Float64 | DataType::Int64)) + ) { + return Err(SchemaDerivationError::InvalidSampleColumn(format!( + "column {vi} is not numeric" + ))); + } + let mut columns = input.fields.clone(); + { + let mut out = agg.output_column(&columns[vi]); + out.name = "value".into(); + // A per-series range reduction produces a PromQL sample value, which is + // always `float64` — override the reducer's own output dtype so + // `count_over_time` (whose `Count` intent types `Int64`) matches every + // other range reducer instead of leaking an `Int64` value column (#69). + out.dtype = FieldDataType::Plain(DataType::Float64); + columns[vi] = out; + } + Ok(Schema { + fields: 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, + }) +} + +/// The output schema of an `Aggregate { reduction, measures }` over `in_schema` — +/// the **single** canonical derivation shared by +/// [`NonASAPOp::output_schema`](crate::ir::NonASAPOp::output_schema)'s +/// `Aggregate` arm and the HAVING-resolution path (`column_resolution::output_schema_for_aggregate`), +/// so the two can never drift (issue #41). +/// +/// `Reduction::PerEntity` selects the label-preserving +/// [`per_series_reduction_schema`] (`rate`/`increase`/`*_over_time`) instead +/// of the cross-series `by ++ measures` shape. Which one applies is read directly +/// off `reduction` — decided once, at construction, by whoever built the +/// `Aggregate` node (issue #165) — not re-derived here from `by`/child shape. +pub fn aggregate_output_schema( + in_schema: &Schema, + reduction: &Reduction, + measures: &[AggIntent], + output_names: &[String], +) -> Result { + let by = match reduction { + Reduction::PerEntity => { + debug_assert_eq!( + measures.len(), + 1, + "a per-entity reduction is single-aggregate" + ); + return per_series_reduction_schema(in_schema, &measures[0]); + } + Reduction::Reduce(by) => by, + }; + + // `without(excluded)` groups by every label *except* those listed: the kept + // labels are the input's label columns minus the excluded positions (and the + // ts / sample-value columns), and the schema stays **open** because the full + // runtime label set isn't known. The `by(...)` path instead enumerates its + // kept columns and freezes to closed (issue #39). + if by.is_without() { + return without_output_schema(in_schema, by.keys(), measures, output_names); + } + + let mut out_cols: Vec = Vec::with_capacity(by.len() + measures.len()); + for &id in by.keys() { + let c = in_schema + .fields + .get(id) + .ok_or(SchemaDerivationError::InvalidGroupByColumn( + id, + in_schema.fields.len(), + ))?; + out_cols.push(c.clone()); + } + let value_col_idx = + crate::pre_asap::column_resolution::resolve_column_ref(&ColumnRef::SampleValue, in_schema) + .ok() + .or_else(|| (0..in_schema.fields.len()).find(|i| !by.contains(i))); + let probe = value_col_idx + .and_then(|i| in_schema.fields.get(i)) + .cloned() + .unwrap_or_else(|| Field::plain("value", DataType::Float64, false)); + // Each reducer types off its own input column (`SUM(bytes)` vs `AVG(latency)` + // in one node); `None` falls back to the sample-value probe (PromQL's + // single-column convention). A non-empty `output_names[i]` overrides the + // synthetic output column name. + for (i, intent) in measures.iter().enumerate() { + // `count_values("l", v)` emits TWO columns: the synthesized `Utf8` label + // `l` (the stringified sample value it groups by) and the per-value + // count. If `l` collides with a group-by key of the same name, PromQL's + // synthesized label takes precedence — emit a single column, never a + // duplicate. + if let AggIntent::CountValues { label } = intent { + if !out_cols.iter().any(|c| c.name == *label) { + out_cols.push(Field::plain(label.clone(), DataType::Utf8, false)); + } + let mut cnt = intent.output_column(&probe); + if let Some(name) = output_names.get(i).filter(|s| !s.is_empty()) { + cnt.name = name.clone(); + } + out_cols.push(cnt); + continue; + } + // Only the output *type* is read from here, so the leading column is + // enough for the multi-column intents: `Cardinality` and `PearsonCorr` + // both have a fixed output type that ignores it. + let in_col = intent + .input_cols() + .first() + .and_then(|id| in_schema.fields.get(*id)) + .unwrap_or(&probe); + let mut out = intent.output_column(in_col); + // A global extremum emits NULL for an empty input, even if its input + // column is non-nullable. Grouped extrema only emit existing groups. + if by.is_empty() && matches!(intent, AggIntent::Min { .. } | AggIntent::Max { .. }) { + out.nullable = true; + } + if let Some((arg, _)) = intent + .arg_selector_columns(in_schema) + .map_err(SchemaDerivationError::InvalidScalarSignature)? + { + out.dtype = in_schema.fields[arg].dtype.clone(); + out.nullable = in_schema.fields[arg].nullable; + } + if let Some(name) = output_names.get(i).filter(|s| !s.is_empty()) { + out.name = name.clone(); + } + out_cols.push(out); + } + // `count_values` groups by (by-keys ∪ the synthesized value label), so the + // by-keys alone are not a unique key — be conservative and claim none. + let has_count_values = measures + .iter() + .any(|a| matches!(a, AggIntent::CountValues { .. })); + let unique_keys = if by.is_empty() || has_count_values { + Vec::new() + } else { + vec![(0..by.len()).collect()] + }; + Ok(Schema { + fields: out_cols, + time_index: None, + unique_keys, + // A cross-series aggregate enumerates exactly `by ++ measures`, so its output + // is closed even over an open input — this is where an open schema + // freezes to closed. + closed: true, + }) +} + +/// Output schema of a `without(excluded)` aggregate: the kept labels (every +/// input label column except the `excluded` positions, the time axis, and the +/// sample-value column) followed by the aggregate output column(s). Unlike the +/// `by` path this stays **open** — the excluded set is enumerable but the kept +/// set is not (the runtime carries labels the usage-derived schema never saw), +/// so the schema can't freeze to closed and claims no unique key (issue #39). +fn without_output_schema( + in_schema: &Schema, + excluded: &[ColumnId], + measures: &[AggIntent], + output_names: &[String], +) -> Result { + for &id in excluded { + if id >= in_schema.fields.len() { + return Err(SchemaDerivationError::InvalidGroupByColumn( + id, + in_schema.fields.len(), + )); + } + } + let mut out_cols: Vec = Vec::new(); + for (i, col) in in_schema.fields.iter().enumerate() { + let is_time = in_schema.time_index == Some(i); + let is_value = crate::pre_asap::column_resolution::resolve_column_ref( + &ColumnRef::SampleValue, + in_schema, + ) + .ok() + == Some(i); + if !is_time && !is_value && !excluded.contains(&i) { + out_cols.push(col.clone()); + } + } + let probe = in_schema + .column_id("value") + .and_then(|i| in_schema.fields.get(i)) + .cloned() + .unwrap_or_else(|| Field::plain("value", DataType::Float64, false)); + for (i, intent) in measures.iter().enumerate() { + // Only the output *type* is read from here, so the leading column is + // enough for the multi-column intents: `Cardinality` and `PearsonCorr` + // both have a fixed output type that ignores it. + let in_col = intent + .input_cols() + .first() + .and_then(|id| in_schema.fields.get(*id)) + .unwrap_or(&probe); + let mut out = intent.output_column(in_col); + if let Some((arg, _)) = intent + .arg_selector_columns(in_schema) + .map_err(SchemaDerivationError::InvalidScalarSignature)? + { + out.dtype = in_schema.fields[arg].dtype.clone(); + out.nullable = in_schema.fields[arg].nullable; + } + if let Some(name) = output_names.get(i).filter(|s| !s.is_empty()) { + out.name = name.clone(); + } + out_cols.push(out); + } + Ok(Schema { + fields: out_cols, + time_index: None, + unique_keys: Vec::new(), + // The kept label set is runtime-only, so — unlike `by` — this does not + // freeze the open schema to closed. + closed: false, + }) +} + +#[cfg(test)] +mod tests { + use super::*; + + fn col(name: &str, dtype: DataType, nullable: bool) -> Field { + Field::plain(name, dtype, nullable) + } + + #[test] + fn group_keys_by_vs_without_semantics() { + let by = GroupKeys::by(vec![1, 2]); + let without = GroupKeys::without(vec![1, 2]); + assert!(!by.is_without()); + assert!(without.is_without()); + // Deref / iteration expose the stored keys regardless of mode. + assert_eq!(by.len(), 2); + assert_eq!(without.keys(), &[1, 2]); + // A `by` compares equal to its bare vec; a `without` never does. + assert_eq!(by, vec![1, 2]); + assert_ne!(without, vec![1, 2]); + assert_ne!(by, without); + } + + #[test] + fn group_keys_serde_by_is_bare_array_without_is_tagged() { + // `by` keeps the pre-#39 bare-array wire format; `without` uses an object. + let by = serde_json::to_string(&GroupKeys::by(vec![2, 3])).unwrap(); + assert_eq!(by, "[2,3]"); + let without = serde_json::to_string(&GroupKeys::without(vec![2])).unwrap(); + assert_eq!(without, r#"{"without":[2]}"#); + // Round-trip both. + for g in [GroupKeys::by(vec![2, 3]), GroupKeys::without(vec![2])] { + let json = serde_json::to_string(&g).unwrap(); + let back: GroupKeys = serde_json::from_str(&json).unwrap(); + assert_eq!(back, g); + } + } + + #[test] + fn time_shift_identity_and_serde() { + let offset_only = TimeShift { + offset_ms: 1, + at: None, + }; + let at_only = TimeShift { + offset_ms: 0, + at: Some(AtModifier::End), + }; + assert!(TimeShift::default().is_identity()); + assert!(!offset_only.is_identity()); + assert!(!at_only.is_identity()); + // Round-trip the shift + anchor. + let s = TimeShift { + offset_ms: -300_000, + at: Some(AtModifier::Timestamp(60_000)), + }; + let back: TimeShift = serde_json::from_str(&serde_json::to_string(&s).unwrap()).unwrap(); + assert_eq!(back, s); + } + + // Nested temporal aggregation must replace the sample, never the grouping label. + #[test] + fn temporal_reduction_of_grouped_sum_preserves_job() { + let input = Schema::new(vec![ + col("job", DataType::Utf8, true), + col("sum", DataType::Float64, false), + ]); + for aggregate in [ + AggIntent::Avg { col: None }, + AggIntent::Avg { col: Some(1) }, + AggIntent::Rate, + ] { + let output = + aggregate_output_schema(&input, &Reduction::PerEntity, &[aggregate], &[]).unwrap(); + assert_eq!(output.fields[0], input.fields[0]); + assert_eq!(output.fields[1].name, "value"); + assert_eq!(output.fields[1].dtype, DataType::Float64); + } + } + + #[test] + fn discriminator_assertion_rejects_unknown_wire_fields() { + let json = r#"{"discriminator":1,"inner_key":[0],"unverified":true}"#; + assert!(serde_json::from_str::(json).is_err()); + } + + #[test] + fn aggregate_strips_time_and_keeps_unique_keys() { + let input = Schema::with_time_index( + vec![ + col("ts", DataType::Timestamp, false), + col("value", DataType::Float64, false), + col("host", DataType::Utf8, false), + ], + 0, + Vec::new(), + ); + let out = aggregate_output_schema( + &input, + &Reduction::by(vec![2]), + &[AggIntent::Sum { col: None }], + &[], + ) + .expect("valid group-by column"); + let names: Vec<_> = out.fields.iter().map(|c| c.name.as_str()).collect(); + assert_eq!(names, vec!["host", "sum"]); + assert!(out.time_index.is_none()); + assert_eq!(out.unique_keys, vec![vec![0]]); + } +} diff --git a/crates/types/src/ir/asap.rs b/crates/types/src/ir/asap.rs new file mode 100644 index 000000000..faee35c5f --- /dev/null +++ b/crates/types/src/ir/asap.rs @@ -0,0 +1,592 @@ +//! ASAP operators: summary-state construction, state operations and evaluations. +//! The summary family, kind/algorithm and parameters are committed here. + +use std::rc::Rc; + +use serde::{Deserialize, Serialize}; + +use super::node::{OperatorNode, OperatorResultKind}; +use crate::ir::operator_properties::Reduction; +use crate::ir::SchemaDerivationError; +use crate::post_asap::maintained_population::{MaintainedPopulation, PopulationStatistic}; +use crate::post_asap::sketch::{GroupingStrategy, SketchStatistic, SummaryUpdate}; +use crate::pre_asap::schema::{ColumnId, DataType, Field, FieldDataType, Schema}; + +/// Why an ASAP operator cannot be used yet. +pub const UNIMPLEMENTED_ASAP_OP: &str = + "this ASAP operator is reserved: schema, accuracy, timing and export are not implemented"; + +#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)] +pub enum ASAPOp { + /// Summary aggregation. Output: grouping columns + one field carrying + /// partial summary state per group, typed `family`. + SummaryAgg { + child: Rc, + /// Which summary family realizes this aggregation. Never + /// `FieldDataType::Plain`. + family: FieldDataType, + input: SummaryUpdate, + reduction: Reduction, + grouping: GroupingStrategy, + #[serde(default)] + filter: Option, + }, + /// Read out a query result from built summary state. Output is a + /// row-shaped schema. + SummaryEstimate { + summary_input: Rc, + query: SketchStatistic, + }, + /// Read an exact accumulator's state as its finalized value: the + /// maintenance-to-read boundary before query-time operators. + FinalizeExactAccumulator { + child: Rc, + }, + /// Maintain the full declared population, including membership changes. + MaintainPopulation { + child: Rc, + population: MaintainedPopulation, + }, + /// Read an aggregate or TopK prefix from the maintained population. + EvaluatePopulation { + child: Rc, + evaluation: PopulationStatistic, + }, + // ── Reserved: migrated but unimplemented (§1.3 of the proposal) ── + SummaryMerge { + children: Vec>, + }, + SummarySubtract { + left: Rc, + right: Rc, + }, + SummaryDelete { + summary_input: Rc, + key: ColumnId, + }, + SummaryJoin { + outer: Rc, + inner: Rc, + key: ColumnId, + family: FieldDataType, + }, + Extension { + child: Rc, + name: String, + }, +} + +impl ASAPOp { + pub fn children(&self) -> Vec<&Rc> { + use ASAPOp::*; + match self { + SummaryAgg { child, filter, .. } => { + let mut inputs = vec![child]; + if let Some(filter) = filter { + inputs.extend(filter.0.operator_refs()); + } + inputs + } + FinalizeExactAccumulator { child } + | MaintainPopulation { child, .. } + | EvaluatePopulation { child, .. } + | Extension { child, .. } => vec![child], + SummaryEstimate { summary_input, .. } | SummaryDelete { summary_input, .. } => { + vec![summary_input] + } + SummarySubtract { left, right } => vec![left, right], + SummaryJoin { outer, inner, .. } => vec![outer, inner], + SummaryMerge { children } => children.iter().collect(), + } + } + + pub fn map_children(&self, mut f: impl FnMut(&Rc) -> Rc) -> Self { + use ASAPOp::*; + match self { + SummaryAgg { + child, + family, + input, + reduction, + grouping, + filter, + } => SummaryAgg { + child: f(child), + family: family.clone(), + input: input.clone(), + reduction: reduction.clone(), + grouping: grouping.clone(), + filter: filter + .as_ref() + .map(|p| super::scalar::Predicate(p.0.map_operator_refs(&mut f))), + }, + SummaryEstimate { + summary_input, + query, + } => SummaryEstimate { + summary_input: f(summary_input), + query: query.clone(), + }, + FinalizeExactAccumulator { child } => FinalizeExactAccumulator { child: f(child) }, + MaintainPopulation { child, population } => MaintainPopulation { + child: f(child), + population: population.clone(), + }, + EvaluatePopulation { child, evaluation } => EvaluatePopulation { + child: f(child), + evaluation: evaluation.clone(), + }, + SummaryMerge { children } => SummaryMerge { + children: children.iter().map(&mut f).collect(), + }, + SummarySubtract { left, right } => SummarySubtract { + left: f(left), + right: f(right), + }, + SummaryDelete { summary_input, key } => SummaryDelete { + summary_input: f(summary_input), + key: *key, + }, + SummaryJoin { + outer, + inner, + key, + family, + } => SummaryJoin { + outer: f(outer), + inner: f(inner), + key: *key, + family: family.clone(), + }, + Extension { child, name } => Extension { + child: f(child), + name: name.clone(), + }, + } + } + + pub fn kind_name(&self) -> &'static str { + use ASAPOp::*; + match self { + SummaryAgg { .. } => "SummaryAgg", + SummaryEstimate { .. } => "SummaryEstimate", + FinalizeExactAccumulator { .. } => "FinalizeExactAccumulator", + MaintainPopulation { .. } => "MaintainPopulation", + EvaluatePopulation { .. } => "EvaluatePopulation", + SummaryMerge { .. } => "SummaryMerge", + SummarySubtract { .. } => "SummarySubtract", + SummaryDelete { .. } => "SummaryDelete", + SummaryJoin { .. } => "SummaryJoin", + Extension { .. } => "Extension", + } + } + + /// Reserved variants that are migrated but not implemented. + pub fn is_unimplemented(&self) -> bool { + use ASAPOp::*; + matches!( + self, + SummaryMerge { .. } + | SummarySubtract { .. } + | SummaryDelete { .. } + | SummaryJoin { .. } + | Extension { .. } + ) + } + + fn unimplemented() -> SchemaDerivationError { + SchemaDerivationError::InvalidScalarSignature(UNIMPLEMENTED_ASAP_OP.into()) + } + + /// The summary state this operator produces, if it produces state. + pub fn produced_state(&self) -> Option<&FieldDataType> { + match self { + ASAPOp::SummaryAgg { family, .. } | ASAPOp::SummaryJoin { family, .. } => Some(family), + _ => None, + } + } + + /// Output schema derived from the operator and its children. Summary + /// planning may retain a more specific schema (evaluation column naming) + /// through [`OperatorNode::with_schema`]; all structural metadata must + /// still agree with this derivation. + pub fn output_schema(&self) -> Result { + use ASAPOp::*; + Ok(match self { + SummaryAgg { + child, + family, + reduction, + .. + } => { + let mut schema = crate::ir::aggregate_schema::aggregate_output_schema( + &child.schema, + reduction, + &[crate::pre_asap::AggIntent::Sum { col: None }], + &[], + )?; + let index = match reduction { + Reduction::PerEntity => crate::pre_asap::resolve_column_ref( + &crate::pre_asap::ColumnRef::SampleValue, + &schema, + ) + .map_err(|e| SchemaDerivationError::InvalidScalarSignature(e.to_string()))?, + Reduction::Reduce(_) => schema.fields.len() - 1, + }; + schema.fields[index] = Field::new("state", family.clone(), false); + schema + } + SummaryEstimate { + summary_input, + query, + } => { + let input = &summary_input.schema; + let (name, mut dtype) = match query { + SketchStatistic::Quantile { .. } => ("quantile", DataType::Float64), + SketchStatistic::Cardinality => ("cardinality", DataType::Int64), + SketchStatistic::PointCount { .. } => ("count", DataType::Int64), + SketchStatistic::FrequencyL2 => ("frequency_l2", DataType::Float64), + SketchStatistic::FrequencyEntropy => ("frequency_entropy", DataType::Float64), + SketchStatistic::TopK { .. } => ("topk", DataType::Utf8), + }; + if matches!( + summary_input.asap(), + Some(SummaryAgg { + reduction: Reduction::PerEntity, + .. + }) + ) && dtype == DataType::Int64 + { + dtype = DataType::Float64; + } + let mut schema = input.clone(); + for field in &mut schema.fields { + if !field.is_plain() { + *field = Field::plain(name, dtype.clone(), false); + } + } + schema + } + FinalizeExactAccumulator { child } => { + let value_result = if let Some(ASAPOp::SummaryAgg { + child: source, + family: FieldDataType::ExactAggregate(kind, _), + input, + reduction, + .. + }) = child.asap() + { + { + use crate::post_asap::{ExactKind, SummaryInputExpr}; + use crate::pre_asap::AggIntent; + let column = match &input.weight { + SummaryInputExpr::Column(col) => Some( + crate::pre_asap::column_resolution::resolve_column_ref( + col, + &source.schema, + ) + .map_err(|e| { + SchemaDerivationError::InvalidScalarSignature(e.to_string()) + })?, + ), + _ => None, + }; + let measure = match kind { + ExactKind::Sum => Some(AggIntent::Sum { col: column }), + ExactKind::Min => Some(AggIntent::Min { col: column }), + ExactKind::Max => Some(AggIntent::Max { col: column }), + ExactKind::Count => Some(AggIntent::Count { + accuracy: crate::types::AccuracyTarget::Exact, + }), + ExactKind::Rate => Some(AggIntent::Rate), + ExactKind::IRate => Some(AggIntent::IRate), + ExactKind::Increase => Some(AggIntent::Increase), + }; + measure + .map(|measure| { + super::NonASAPOp::Aggregate { + child: Rc::clone(source), + reduction: reduction.clone(), + measures: vec![measure], + output_names: vec![], + filters: vec![], + having: None, + } + .output_schema() + }) + .transpose()? + .and_then(|s| match reduction { + Reduction::PerEntity => { + s.column_id("value").and_then(|i| s.fields.get(i).cloned()) + } + Reduction::Reduce(_) => s.fields.last().cloned(), + }) + } + } else { + None + }; + let mut out = child.schema.clone(); + for f in &mut out.fields { + if let FieldDataType::ExactAggregate(kind, _) = &f.dtype { + if let Some(result) = &value_result { + f.dtype = result.dtype.clone(); + f.nullable = result.nullable; + } else { + f.dtype = FieldDataType::Plain(finalized_data_type(kind)); + } + } + } + out + } + MaintainPopulation { child, .. } => child.schema.clone(), + EvaluatePopulation { child, evaluation } => { + use crate::post_asap::maintained_population::PopulationInput; + use crate::pre_asap::{AggIntent, GroupKeys}; + let Some(MaintainPopulation { + child: source, + population, + }) = child.asap() + else { + return Err(SchemaDerivationError::InvalidScalarSignature( + "population evaluation requires maintained membership".into(), + )); + }; + if matches!(evaluation, PopulationStatistic::TopK { .. }) { + source.schema.clone() + } else { + let (keys, column) = match &population.input { + PopulationInput::Rows { + grouping, + value_column, + .. + } => (grouping.clone(), Some(*value_column)), + PopulationInput::CurrentSeries(spec) => { + let keys = spec + .grouping + .iter() + .map(|name| { + source.schema.column_id(name).ok_or_else(|| { + SchemaDerivationError::InvalidScalarSignature( + "population grouping column is absent".into(), + ) + }) + }) + .collect::, _>>()?; + ( + if spec.without { + GroupKeys::without(keys) + } else { + GroupKeys::by(keys) + }, + None, + ) + } + }; + let accuracy = crate::types::AccuracyTarget::Exact; + let measure = match evaluation { + PopulationStatistic::Quantile { q } => AggIntent::Quantile { + q: *q, + col: column, + accuracy, + }, + PopulationStatistic::Sum => AggIntent::Sum { col: column }, + PopulationStatistic::Count => AggIntent::Count { accuracy }, + PopulationStatistic::Average => AggIntent::Avg { col: column }, + PopulationStatistic::TopK { .. } => unreachable!(), + }; + super::NonASAPOp::Aggregate { + child: source.clone(), + reduction: Reduction::Reduce(keys), + measures: vec![measure], + output_names: vec![], + filters: vec![], + having: None, + } + .output_schema()? + } + } + SummaryMerge { .. } + | SummarySubtract { .. } + | SummaryDelete { .. } + | SummaryJoin { .. } + | Extension { .. } => return Err(Self::unimplemented()), + }) + } + + pub fn output_kind(&self) -> OperatorResultKind { + use ASAPOp::*; + match self { + SummaryAgg { .. } + | MaintainPopulation { .. } + | SummaryMerge { .. } + | SummarySubtract { .. } + | SummaryDelete { .. } + | SummaryJoin { .. } + | Extension { .. } => OperatorResultKind::State, + SummaryEstimate { summary_input, .. } => source_kind(summary_input), + FinalizeExactAccumulator { child } | EvaluatePopulation { child, .. } => { + source_kind(child) + } + } + } + + /// Local input-contract checks. + pub fn validate_inputs(&self) -> Result<(), SchemaDerivationError> { + use ASAPOp::*; + let needs_state = |node: &OperatorNode, what: &str| { + if node.result_kind != OperatorResultKind::State { + Err(SchemaDerivationError::InvalidScalarSignature(format!( + "{what} requires summary state as input, got {:?}", + node.result_kind + ))) + } else { + Ok(()) + } + }; + match self { + SummaryEstimate { + summary_input, + query, + } => { + needs_state(summary_input, "SummaryEstimate")?; + use crate::post_asap::sketch::SketchCategory as C; + let states: Vec<_> = summary_input + .schema + .fields + .iter() + .filter(|f| !f.is_plain()) + .collect(); + let valid = match states.as_slice() { + [field] => match &field.dtype { + FieldDataType::Sketch(kind, _) => matches!( + (kind.category(), query), + (C::Quantile, SketchStatistic::Quantile { .. }) + | (C::Cardinality | C::Universal, SketchStatistic::Cardinality) + | ( + C::Frequency | C::Universal, + SketchStatistic::PointCount { .. } + ) + | ( + C::Universal, + SketchStatistic::FrequencyL2 + | SketchStatistic::FrequencyEntropy + ) + | (C::TopK | C::Universal, SketchStatistic::TopK { .. }) + ), + _ => false, + }, + _ => false, + }; + if !valid { + return Err(SchemaDerivationError::InvalidScalarSignature( + "evaluation does not match its summary family".into(), + )); + } + Ok(()) + } + FinalizeExactAccumulator { child } => { + needs_state(child, "FinalizeExactAccumulator")?; + if child + .schema + .fields + .iter() + .all(|f| !matches!(f.dtype, FieldDataType::ExactAggregate(..))) + { + return Err(SchemaDerivationError::InvalidScalarSignature( + "FinalizeExactAccumulator requires exact accumulator state".into(), + )); + } + Ok(()) + } + EvaluatePopulation { child, evaluation } => { + needs_state(child, "EvaluatePopulation")?; + if !matches!(child.asap(), Some(MaintainPopulation { population, .. }) if population.supports(evaluation)) + { + return Err(SchemaDerivationError::InvalidScalarSignature( + "population evaluation requires compatible maintained membership".into(), + )); + } + Ok(()) + } + SummaryAgg { + child, + family, + input, + filter, + .. + } => { + if family.is_plain() || child.result_kind == OperatorResultKind::State { + return Err(SchemaDerivationError::InvalidScalarSignature( + "summary aggregation requires values and produces a state family".into(), + )); + } + fn check( + expr: &crate::post_asap::SummaryInputExpr, + schema: &Schema, + ) -> Result<(), SchemaDerivationError> { + use crate::post_asap::SummaryInputExpr; + match expr { + SummaryInputExpr::Column(col) => { + crate::pre_asap::resolve_column_ref(col, schema).map_err(|e| { + SchemaDerivationError::InvalidScalarSignature(e.to_string()) + })?; + } + SummaryInputExpr::Tuple(items) => { + for item in items { + check(item, schema)?; + } + } + _ => {} + } + Ok(()) + } + check(&input.weight, &child.schema)?; + if let Some(item) = &input.item { + check(item, &child.schema)?; + } + if let Some(filter) = filter { + if filter.0.scalar_type(&child.schema)?.0 != DataType::Bool { + return Err(SchemaDerivationError::InvalidScalarSignature( + "summary filter must be boolean".into(), + )); + } + } + Ok(()) + } + MaintainPopulation { child, population } => { + if !population.matches_node(child) { + return Err(SchemaDerivationError::InvalidScalarSignature( + "population input differs from its membership contract".into(), + )); + } + Ok(()) + } + _ => Err(Self::unimplemented()), + } + } +} + +/// The plain value an exact accumulator finalizes to. +fn finalized_data_type(kind: &crate::post_asap::sketch::ExactKind) -> DataType { + use crate::post_asap::sketch::ExactKind; + match kind { + ExactKind::Count => DataType::Int64, + _ => DataType::Float64, + } +} + +/// The category of the values a evaluation of `state` produces: the category +/// of the relational input the state was built from. +fn source_kind(node: &OperatorNode) -> OperatorResultKind { + match &node.operator { + super::node::Operator::ASAP(op) => match op.children().first() { + Some(child) => source_kind(child), + None => OperatorResultKind::Relation, + }, + super::node::Operator::NonASAP(_) => match node.result_kind { + OperatorResultKind::RangeVector => OperatorResultKind::InstantVector, + OperatorResultKind::State => OperatorResultKind::Relation, + other => other, + }, + } +} diff --git a/crates/types/src/ir/error.rs b/crates/types/src/ir/error.rs new file mode 100644 index 000000000..4b6849965 --- /dev/null +++ b/crates/types/src/ir/error.rs @@ -0,0 +1,19 @@ +//! Typed failures from operator/scalar schema and type derivation. +//! +//! [`SchemaDerivationError`] distinguishes invalid scalar signatures, out-of-range +//! grouping columns, empty concatenations, and invalid sample columns. +//! Structural DAG and execution-timing validation have separate error types. +use crate::pre_asap::ColumnId; +use thiserror::Error; +/// Errors from schema and type derivation over an operator DAG. +#[derive(Debug, Error)] +pub enum SchemaDerivationError { + #[error("invalid scalar function signature: {0}")] + InvalidScalarSignature(String), + #[error("by-column id {0} out of range (input has {1} columns)")] + InvalidGroupByColumn(ColumnId, usize), + #[error("Concat requires at least one child")] + EmptyConcat, + #[error("invalid per-series sample column: {0}")] + InvalidSampleColumn(String), +} diff --git a/crates/types/src/ir/mod.rs b/crates/types/src/ir/mod.rs new file mode 100644 index 000000000..d20bfd075 --- /dev/null +++ b/crates/types/src/ir/mod.rs @@ -0,0 +1,17 @@ +//! Unified operator and scalar representation from #511. +//! Graph algorithms are added in the next stack layer; legacy consumers +//! remain on their existing representation until the planner cutover. +pub mod aggregate_schema; +pub mod asap; +pub mod error; +pub mod node; +pub mod non_asap; +pub mod operator_properties; +pub mod query; +pub mod scalar; +pub use asap::{ASAPOp, UNIMPLEMENTED_ASAP_OP}; +pub use error::SchemaDerivationError; +pub use node::{Operator, OperatorNode, OperatorResultKind}; +pub use non_asap::{BinaryOperator, NonASAPOp, TimeRangeKind}; +pub use query::QueryRoot; +pub use scalar::{ExprSemantics, Predicate, ProjectItem, ScalarExpr, SortKey}; diff --git a/crates/types/src/ir/node.rs b/crates/types/src/ir/node.rs new file mode 100644 index 000000000..d68f5aa03 --- /dev/null +++ b/crates/types/src/ir/node.rs @@ -0,0 +1,312 @@ +//! The unified operator node: one node type before and after ASAP +//! optimization. A node is an operator plus the common planning properties +//! every traversal needs (output category and schema, accuracy guarantee, +//! execution timing). + +use std::collections::HashSet; +use std::rc::Rc; + +use serde::{Deserialize, Serialize}; + +use super::asap::ASAPOp; +use super::non_asap::NonASAPOp; +use crate::ir::SchemaDerivationError; +use crate::post_asap::execution_data_state::ExecutionTiming; +use crate::post_asap::guarantee::ResultGuarantee; +use crate::pre_asap::schema::Schema; + +/// The output category of an operator, derived from the operation and its +/// inputs. Matching column schemas do not make categories interchangeable. +#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, Serialize, Deserialize)] +#[serde(rename_all = "snake_case")] +pub enum OperatorResultKind { + Relation, + InstantVector, + RangeVector, + /// Unfinalized summary / accumulator state. + State, +} + +/// The operation a node performs: an ordinary query operator or an ASAP +/// summary operator. Either category can consume the other's output. +#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)] +pub enum Operator { + NonASAP(NonASAPOp), + ASAP(ASAPOp), +} + +impl Operator { + pub fn children(&self) -> Vec<&Rc> { + match self { + Operator::NonASAP(op) => op.children(), + Operator::ASAP(op) => op.children(), + } + } + + pub fn map_children(&self, f: impl FnMut(&Rc) -> Rc) -> Self { + match self { + Operator::NonASAP(op) => Operator::NonASAP(op.map_children(f)), + Operator::ASAP(op) => Operator::ASAP(op.map_children(f)), + } + } + + pub fn output_schema(&self) -> Result { + match self { + Operator::NonASAP(op) => op.output_schema(), + Operator::ASAP(op) => op.output_schema(), + } + } + + pub fn output_kind(&self) -> OperatorResultKind { + match self { + Operator::NonASAP(op) => op.output_kind(), + Operator::ASAP(op) => op.output_kind(), + } + } + + pub fn validate_inputs(&self) -> Result<(), SchemaDerivationError> { + match self { + Operator::NonASAP(op) => op.validate_inputs(), + Operator::ASAP(op) => op.validate_inputs(), + } + } + + pub fn kind_name(&self) -> &'static str { + match self { + Operator::NonASAP(op) => op.kind_name(), + Operator::ASAP(op) => op.kind_name(), + } + } +} + +/// A node of the logical DAG. Nodes are immutable and shared through `Rc`; +/// a structurally identical sub-DAG referenced from several parents is one +/// node. +/// +/// `schema` and `result_kind` are derived from `operator` and its children +/// at construction and retained. `guarantee` is `None` until accuracy +/// assessment establishes one (`None` never means exact). `timing` is `None` +/// until a lifecycle assignment is applied; export rejects an executable +/// node without one. +#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)] +pub struct OperatorNode { + pub operator: Operator, + pub result_kind: OperatorResultKind, + pub schema: Schema, + pub guarantee: Option, + pub timing: Option, +} + +impl OperatorNode { + /// Build a node, deriving its schema and output category. Fails when the + /// schema cannot be derived (a column reference out of range, a reserved + /// ASAP operator, ...). + pub fn new(operator: Operator) -> Result { + let schema = operator.output_schema()?; + Ok(Self::with_schema(operator, schema)) + } + + /// Build a node with caller-supplied output names and qualifiers. For + /// either operator category, `validate_structure` requires all other + /// schema metadata to agree with derivation; output kind is derived here. + pub fn with_schema(operator: Operator, schema: Schema) -> Self { + let result_kind = operator.output_kind(); + Self { + operator, + result_kind, + schema, + guarantee: None, + timing: None, + } + } + + /// Build a shared node for either operator category, deriving its schema + /// and output category. Use `with_schema` before wrapping in `Rc` when + /// planning supplies a more specific output schema. + pub fn new_shared(operator: Operator) -> Result, SchemaDerivationError> { + Self::new(operator).map(Rc::new) + } + + pub fn with_guarantee(mut self, guarantee: Option) -> Self { + self.guarantee = guarantee; + self + } + + pub fn with_timing(mut self, timing: Option) -> Self { + self.timing = timing; + self + } + + pub fn non_asap(&self) -> Option<&NonASAPOp> { + match &self.operator { + Operator::NonASAP(op) => Some(op), + Operator::ASAP(_) => None, + } + } + + pub fn asap(&self) -> Option<&ASAPOp> { + match &self.operator { + Operator::ASAP(op) => Some(op), + Operator::NonASAP(_) => None, + } + } + + pub fn is_asap(&self) -> bool { + matches!(self.operator, Operator::ASAP(_)) + } + + /// The non-ASAP operator of a node that is known to be one; a front-end + /// DAG never contains an ASAP node, so one here is a caller bug. + pub fn expect_non_asap(&self) -> &NonASAPOp { + self.non_asap().unwrap_or_else(|| { + panic!( + "expected a non-ASAP operator, found {}", + self.operator.kind_name() + ) + }) + } + + /// Direct inputs, including the operator nodes referenced from this + /// node's scalar expressions. + pub fn children(&self) -> Vec<&Rc> { + self.operator.children() + } + + /// Rebuild with new inputs, re-deriving all structural schema metadata. + /// Only names and qualifiers that override the old derived schema are + /// retained, for either operator category. A change in output arity with + /// such overrides needs an explicit new naming assignment. + /// `guarantee` and `timing` depend on the inputs and are cleared. + pub fn map_children( + &self, + f: impl FnMut(&Rc) -> Rc, + ) -> Result { + let previous = self.operator.output_schema()?; + let mut rebuilt = Self::new(self.operator.map_children(f))?; + for (i, (derived, retained)) in previous.fields.iter().zip(&self.schema.fields).enumerate() + { + let renamed = retained.name != derived.name; + let requalified = retained.table != derived.table; + if !renamed && !requalified { + continue; + } + if rebuilt.schema.fields.len() != previous.fields.len() { + return Err(SchemaDerivationError::InvalidScalarSignature( + "output arity changed; reassign explicit output names and qualifiers".into(), + )); + } + if renamed { + rebuilt.schema.fields[i].name = retained.name.clone(); + } + if requalified { + rebuilt.schema.fields[i].table = retained.table.clone(); + } + } + Ok(rebuilt) + } + + /// Whether any node reachable from this one (including itself) is an + /// ASAP operator. Visits each shared node once. + pub fn contains_asap(&self) -> bool { + fn walk(node: &OperatorNode, seen: &mut HashSet<*const OperatorNode>) -> bool { + if node.is_asap() { + return true; + } + node.children() + .iter() + .any(|child| seen.insert(Rc::as_ptr(child)) && walk(child, seen)) + } + walk(self, &mut HashSet::new()) + } + + /// Every unique node reachable from `root`, parents before children + /// (pre-order, deduplicated by pointer identity). + pub fn reachable(root: &Rc) -> Vec> { + fn walk( + node: &Rc, + seen: &mut HashSet<*const OperatorNode>, + out: &mut Vec>, + ) { + if !seen.insert(Rc::as_ptr(node)) { + return; + } + out.push(Rc::clone(node)); + for child in node.children() { + walk(child, seen, out); + } + } + let mut out = Vec::new(); + walk(root, &mut HashSet::new(), &mut out); + out + } + + /// Validate assigned phases without imposing any particular runtime implementation. + pub fn validate_execution_timing(self: &Rc) -> Result<(), SchemaDerivationError> { + self.validate_structure()?; + for node in Self::reachable(self) { + let timing = node.timing.ok_or_else(|| { + SchemaDerivationError::InvalidScalarSignature( + "execution timing is unassigned".into(), + ) + })?; + if timing == crate::post_asap::ExecutionTiming::IngestionTime + && node.children().iter().any(|child| { + child.timing != Some(crate::post_asap::ExecutionTiming::IngestionTime) + }) + { + return Err(SchemaDerivationError::InvalidScalarSignature( + "ingestion-time operation depends on a query-time or unassigned input".into(), + )); + } + } + Ok(()) + } + + /// Validate the whole DAG reachable from this node: every operator's + /// input contract, scalar typing against the owning operator's input + /// schema, and agreement between each retained schema and the one + /// derived from the operator. Both operator categories may override field + /// names and qualifiers; all structural metadata must match derivation. + /// `timing` may be `None`. + pub fn validate_structure(self: &Rc) -> Result<(), SchemaDerivationError> { + for node in Self::reachable(self) { + if node.schema.time_index.is_some_and(|i| { + node.schema + .fields + .get(i) + .is_none_or(|f| f.plain_dtype() != Some(&crate::pre_asap::DataType::Timestamp)) + }) || node + .schema + .unique_keys + .iter() + .flatten() + .any(|i| *i >= node.schema.fields.len()) + { + return Err(SchemaDerivationError::InvalidScalarSignature( + "invalid time or identity column in schema".into(), + )); + } + node.operator.validate_inputs()?; + if node.result_kind != node.operator.output_kind() { + return Err(SchemaDerivationError::InvalidScalarSignature( + "retained result kind disagrees with operation".into(), + )); + } + let mut derived = node.operator.output_schema()?; + // Normalize only naming overrides, then compare the whole schema + // so new structural metadata cannot accidentally escape validation. + for (field, retained) in derived.fields.iter_mut().zip(&node.schema.fields) { + field.name = retained.name.clone(); + field.table = retained.table.clone(); + } + let agree = derived == node.schema; + if !agree { + return Err(SchemaDerivationError::InvalidScalarSignature(format!( + "retained schema of {} disagrees with its derived schema", + node.operator.kind_name() + ))); + } + } + Ok(()) + } +} diff --git a/crates/types/src/ir/non_asap.rs b/crates/types/src/ir/non_asap.rs new file mode 100644 index 000000000..0daf3a0ce --- /dev/null +++ b/crates/types/src/ir/non_asap.rs @@ -0,0 +1,1624 @@ +//! Ordinary (non-ASAP) query operators: everything a front end emits and +//! everything that survives ASAP optimization unchanged. + +use std::rc::Rc; +use std::time::Duration; + +use serde::{Deserialize, Serialize}; + +use super::node::{OperatorNode, OperatorResultKind}; +use super::scalar::{Predicate, ProjectItem, ScalarExpr, SortKey}; +use crate::ir::aggregate_schema::aggregate_output_schema; +use crate::ir::operator_properties::{ + BinaryOpKind, ConcatDiscriminatorKey, GroupKeys, InfoMatcher, JoinKind, Reduction, + RelationalSetOpKind, SampleKind, Source, TimeShift, VectorMatch, WindowFrame, WindowFuncKind, +}; +use crate::ir::SchemaDerivationError; +use crate::pre_asap::agg_intent::AggIntent; +use crate::pre_asap::schema::{ColumnId, DataType, Field, FieldDataType, Schema}; + +/// All semantics owned by a binary operator. +#[derive(Debug, Clone, PartialEq, serde::Serialize, serde::Deserialize)] +pub struct BinaryOperator { + /// Execute division only for finite operands, a nonzero divisor, and a + /// normal finite result; otherwise use exact execution. Required by the + /// relative-value division certificate, including floating-point range. + #[serde(default)] + pub checked_relative_division: bool, + /// Conditional exact rewrites (such as temporal average from sum/count) + /// require finite operands and quotient. Zero/subnormal results are valid; + /// overflow must fall back to the original query rather than emit infinity. + #[serde(default)] + pub checked_finite_division: bool, + pub kind: BinaryOpKind, + /// `None` is the only currently supported vector/vector matching mode. + /// The field is retained so execution never has to recover semantics by + /// re-parsing PromQL. + pub vector_match: Option, +} + +/// Which samples a PromQL selector reads. +#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, Serialize, Deserialize)] +#[serde(rename_all = "snake_case")] +pub enum TimeRangeKind { + /// An instant selector: `range` is the lookback horizon and the latest + /// eligible sample per series is selected. + Instant, + /// A range selector (`m[5m]`): every sample in the window. + Range, +} + +/// The non-ASAP operator vocabulary. Children are [`Rc`], so an +/// ordinary operator can read a summary evaluation and a summary can read any +/// relational sub-DAG. +#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)] +pub enum NonASAPOp { + /// Leaf. `schema` is the binding schema every positional `ColumnId` in + /// the tree indexes into; `predicates` are leaf-level row filters (PromQL + /// label matchers, pushed-down `WHERE` conjuncts). + Scan { + source: Source, + #[serde(default)] + predicates: Vec, + schema: Schema, + }, + /// SQL `VALUES` rows, or the one empty row of a `SELECT` without `FROM`. + /// Row expressions have no input-column scope. + Values { + rows: Vec>, + schema: Schema, + }, + /// σ — row-level filter. Output schema = child schema. + Filter { + pred: Predicate, + child: Rc, + }, + /// π — projection. + Project { + cols: Vec, + /// Re-qualifies every output column with this table alias (a derived + /// table / inline view). `None` for an ordinary SELECT list. + #[serde(default)] + qualifier: Option, + child: Rc, + }, + /// γ + α — grouping + aggregate intents. + Aggregate { + reduction: Reduction, + measures: Vec, + /// Output column names parallel to `measures`; an empty entry falls + /// back to the intent's synthetic name. + #[serde(default)] + output_names: Vec, + #[serde(default)] + filters: Vec>, + #[serde(default)] + having: Option, + child: Rc, + }, + Join { + kind: JoinKind, + pred: Predicate, + left: Rc, + right: Rc, + }, + SetOp { + kind: RelationalSetOpKind, + all: bool, + left: Rc, + right: Rc, + }, + /// ⊕ — exact n-ary `UNION ALL` of union-compatible branches. The output + /// schema is the first child's. + Concat { + children: Vec>, + #[serde(default)] + discriminator_unique_key: Option, + }, + /// δ — deduplication; empty `cols` = all columns. + Dedup { + cols: Vec, + child: Rc, + }, + /// Order-by, per `partition_by` group when non-empty. + Sort { + keys: Vec, + #[serde(default)] + partition_by: GroupKeys, + child: Rc, + }, + /// Row selection; `n = None` is offset-only. `partition_by` applies the + /// limit per group (PromQL `topk by (..)`). + Limit { + n: Option, + offset: usize, + #[serde(default)] + partition_by: GroupKeys, + child: Rc, + }, + /// Arithmetic / comparison / set composition of two operands (PromQL + /// binary operators). Mixed scalar/vector operations use Project or Filter. + BinaryOp { + operator: BinaryOperator, + /// PromQL `bool` modifier: a comparison returns `0`/`1` instead of + /// filtering. Valid only for comparison operators. + #[serde(default)] + return_bool: bool, + lhs: Rc, + rhs: Rc, + }, + /// SQL analytic window function. Output schema = child schema + one + /// column named `output_name`. + SQLWindowFunc { + func: WindowFuncKind, + args: Vec, + partition_by: GroupKeys, + order_by: Vec, + #[serde(default)] + frame: Option, + output_name: String, + child: Rc, + }, + /// Temporal selection over a time-series input. + TimeRange { + range: Duration, + kind: TimeRangeKind, + child: Rc, + }, + /// PromQL `offset` / `@`: moves when `child` is evaluated. + TimeShift { + shift: TimeShift, + child: Rc, + }, + /// PromQL `vector(s)`: a label-less instant vector carrying a scalar. + PromqlVectorFromScalar(ScalarExpr), + /// ρ — PromQL `label_replace` / `label_join`. + PromqlRelabel { + dst: String, + value: ScalarExpr, + child: Rc, + }, + /// PromQL `info(v, selector)` label enrichment. + PromqlInfoEnrich { + #[serde(default)] + selector: Vec, + child: Rc, + }, + /// PromQL `limitk` / `limit_ratio`. + PromqlSeriesSample { + #[serde(default)] + by: GroupKeys, + kind: SampleKind, + child: Rc, + }, + /// PromQL subquery `[range:resolution]`. + PromqlSubquery { + range: Duration, + #[serde(default)] + resolution: Option, + child: Rc, + }, +} + +impl NonASAPOp { + /// The direct operator inputs, in field order, followed by the operator + /// nodes referenced from this operator's scalar expressions. + pub fn children(&self) -> Vec<&Rc> { + use NonASAPOp::*; + let mut out: Vec<&Rc> = match self { + Scan { .. } | Values { .. } | PromqlVectorFromScalar(_) => vec![], + Filter { child, .. } + | Project { child, .. } + | Aggregate { child, .. } + | Dedup { child, .. } + | Sort { child, .. } + | Limit { child, .. } + | SQLWindowFunc { child, .. } + | TimeRange { child, .. } + | TimeShift { child, .. } + | PromqlRelabel { child, .. } + | PromqlInfoEnrich { child, .. } + | PromqlSeriesSample { child, .. } + | PromqlSubquery { child, .. } => vec![child], + Join { left, right, .. } | SetOp { left, right, .. } => vec![left, right], + BinaryOp { lhs, rhs, .. } => vec![lhs, rhs], + Concat { children, .. } => children.iter().collect(), + }; + for expr in self.scalar_exprs() { + out.extend(expr.operator_refs()); + } + out + } + + /// Every scalar expression this operator owns. + pub fn scalar_exprs(&self) -> Vec<&ScalarExpr> { + use NonASAPOp::*; + match self { + Scan { predicates, .. } => predicates.iter().map(|p| &p.0).collect(), + Values { rows, .. } => rows.iter().flatten().collect(), + Filter { pred, .. } | Join { pred, .. } => vec![&pred.0], + Project { cols, .. } => cols.iter().map(|c| &c.expr).collect(), + Aggregate { + filters, having, .. + } => filters + .iter() + .flatten() + .chain(having.iter()) + .map(|p| &p.0) + .collect(), + Sort { keys, .. } => keys.iter().map(|k| &k.expr).collect(), + SQLWindowFunc { args, order_by, .. } => args + .iter() + .chain(order_by.iter().map(|k| &k.expr)) + .collect(), + PromqlVectorFromScalar(e) => vec![e], + PromqlRelabel { value, .. } => vec![value], + SetOp { .. } + | Concat { .. } + | Dedup { .. } + | Limit { .. } + | BinaryOp { .. } + | TimeRange { .. } + | TimeShift { .. } + | PromqlInfoEnrich { .. } + | PromqlSeriesSample { .. } + | PromqlSubquery { .. } => vec![], + } + } + + /// Rebuild this operator with `f` applied to every child, including the + /// operator nodes referenced from scalar expressions. Every other field + /// is cloned. + pub fn map_children(&self, mut f: impl FnMut(&Rc) -> Rc) -> Self { + use NonASAPOp::*; + let mut map_scalar = |e: &ScalarExpr| e.map_operator_refs(&mut f); + fn map_pred(p: &Predicate, f: &mut impl FnMut(&ScalarExpr) -> ScalarExpr) -> Predicate { + Predicate(f(&p.0)) + } + fn map_keys( + keys: &[SortKey], + f: &mut impl FnMut(&ScalarExpr) -> ScalarExpr, + ) -> Vec { + keys.iter() + .map(|k| SortKey { + expr: f(&k.expr), + ascending: k.ascending, + nulls_first: k.nulls_first, + }) + .collect() + } + match self { + Scan { + source, + predicates, + schema, + } => Scan { + source: source.clone(), + predicates: predicates + .iter() + .map(|p| map_pred(p, &mut map_scalar)) + .collect(), + schema: schema.clone(), + }, + Values { rows, schema } => Values { + rows: rows + .iter() + .map(|row| row.iter().map(&mut map_scalar).collect()) + .collect(), + schema: schema.clone(), + }, + PromqlVectorFromScalar(e) => PromqlVectorFromScalar(map_scalar(e)), + Filter { pred, child } => { + let pred = map_pred(pred, &mut map_scalar); + Filter { + pred, + child: f(child), + } + } + Project { + cols, + qualifier, + child, + } => { + let cols = cols + .iter() + .map(|c| ProjectItem { + alias: c.alias.clone(), + expr: map_scalar(&c.expr), + }) + .collect(); + Project { + cols, + qualifier: qualifier.clone(), + child: f(child), + } + } + Aggregate { + reduction, + measures, + output_names, + filters, + having, + child, + } => { + let filters = filters + .iter() + .map(|p| p.as_ref().map(|p| map_pred(p, &mut map_scalar))) + .collect(); + let having = having.as_ref().map(|p| map_pred(p, &mut map_scalar)); + Aggregate { + reduction: reduction.clone(), + measures: measures.clone(), + output_names: output_names.clone(), + filters, + having, + child: f(child), + } + } + Join { + kind, + pred, + left, + right, + } => { + let pred = map_pred(pred, &mut map_scalar); + Join { + kind: kind.clone(), + pred, + left: f(left), + right: f(right), + } + } + SetOp { + kind, + all, + left, + right, + } => SetOp { + kind: kind.clone(), + all: *all, + left: f(left), + right: f(right), + }, + Concat { + children, + discriminator_unique_key, + } => Concat { + children: children.iter().map(&mut f).collect(), + discriminator_unique_key: discriminator_unique_key.clone(), + }, + Dedup { cols, child } => Dedup { + cols: cols.clone(), + child: f(child), + }, + Sort { + keys, + partition_by, + child, + } => { + let keys = map_keys(keys, &mut map_scalar); + Sort { + keys, + partition_by: partition_by.clone(), + child: f(child), + } + } + Limit { + n, + offset, + partition_by, + child, + } => Limit { + n: *n, + offset: *offset, + partition_by: partition_by.clone(), + child: f(child), + }, + BinaryOp { + operator, + return_bool, + lhs, + rhs, + } => BinaryOp { + operator: operator.clone(), + return_bool: *return_bool, + lhs: f(lhs), + rhs: f(rhs), + }, + SQLWindowFunc { + func, + args, + partition_by, + order_by, + frame, + output_name, + child, + } => { + let args = args.iter().map(&mut map_scalar).collect(); + let order_by = map_keys(order_by, &mut map_scalar); + SQLWindowFunc { + func: func.clone(), + args, + partition_by: partition_by.clone(), + order_by, + frame: frame.clone(), + output_name: output_name.clone(), + child: f(child), + } + } + TimeRange { range, kind, child } => TimeRange { + range: *range, + kind: *kind, + child: f(child), + }, + TimeShift { shift, child } => TimeShift { + shift: *shift, + child: f(child), + }, + PromqlRelabel { dst, value, child } => { + let value = map_scalar(value); + PromqlRelabel { + dst: dst.clone(), + value, + child: f(child), + } + } + PromqlInfoEnrich { selector, child } => PromqlInfoEnrich { + selector: selector.clone(), + child: f(child), + }, + PromqlSeriesSample { by, kind, child } => PromqlSeriesSample { + by: by.clone(), + kind: *kind, + child: f(child), + }, + PromqlSubquery { + range, + resolution, + child, + } => PromqlSubquery { + range: *range, + resolution: *resolution, + child: f(child), + }, + } + } + + /// The variant name, for diagnostics and export. + pub fn kind_name(&self) -> &'static str { + use NonASAPOp::*; + match self { + Scan { .. } => "Scan", + Values { .. } => "Values", + Filter { .. } => "Filter", + Project { .. } => "Project", + Aggregate { .. } => "Aggregate", + Join { .. } => "Join", + SetOp { .. } => "SetOp", + Concat { .. } => "Concat", + Dedup { .. } => "Dedup", + Sort { .. } => "Sort", + Limit { .. } => "Limit", + BinaryOp { .. } => "BinaryOp", + SQLWindowFunc { .. } => "SQLWindowFunc", + TimeRange { .. } => "TimeRange", + TimeShift { .. } => "TimeShift", + PromqlVectorFromScalar(_) => "PromqlVectorFromScalar", + PromqlRelabel { .. } => "PromqlRelabel", + PromqlInfoEnrich { .. } => "PromqlInfoEnrich", + PromqlSeriesSample { .. } => "PromqlSeriesSample", + PromqlSubquery { .. } => "PromqlSubquery", + } + } + + /// Output schema derived from this operator's parameters and its + /// children's (already derived) schemas. + pub fn output_schema(&self) -> Result { + use NonASAPOp::*; + Ok(match self { + Scan { schema, .. } | Values { schema, .. } => schema.clone(), + + Aggregate { + reduction, + measures, + output_names, + child, + filters, + .. + } => { + let mut output = + aggregate_output_schema(&child.schema, reduction, measures, output_names)?; + if child.result_kind == OperatorResultKind::Relation { + let offset = reduction.group_keys().map_or(0, |keys| keys.len()); + for (index, measure) in measures.iter().enumerate() { + if matches!( + measure, + AggIntent::Sum { .. } + | AggIntent::Avg { .. } + | AggIntent::Min { .. } + | AggIntent::Max { .. } + ) { + let nullable = offset == 0 + || filters.get(index).is_some_and(Option::is_some) + || measure + .input_cols() + .iter() + .any(|i| child.schema.fields[*i].nullable); + if let Some(field) = output.fields.get_mut(offset + index) { + field.nullable = nullable; + } + } + } + } + output + } + + Filter { child, .. } + | Sort { child, .. } + | Limit { child, .. } + | PromqlSubquery { child, .. } + | PromqlSeriesSample { child, .. } + | PromqlInfoEnrich { child, .. } + | TimeRange { child, .. } + | TimeShift { child, .. } => child.schema.clone(), + + // ρ — relabel preserves every input column and writes one label + // `dst` (Utf8): overwritten in place if it already exists, else + // appended (nullable). Row-uniqueness is no longer provable. + PromqlRelabel { dst, child, .. } => { + let mut out = child.schema.clone(); + if let Some(existing) = out.fields.iter_mut().find(|c| c.name == *dst) { + existing.dtype = FieldDataType::Plain(DataType::Utf8); + existing.nullable = true; + } else { + out.fields + .push(Field::plain(dst.clone(), DataType::Utf8, true)); + } + out.unique_keys.clear(); + out + } + + // π — one output column per projection item. A bare column item + // keeps its field verbatim, so an `ExactAggregate` state column + // can pass through a projection unchanged; any other expression + // is typed against the input and must read plain values. + Project { + cols, + qualifier, + child, + } => { + let in_schema = &child.schema; + let fields: Vec = cols + .iter() + .enumerate() + .map(|(i, item)| { + let mut field = match &item.expr { + ScalarExpr::Column(id) if in_schema.fields.get(*id).is_some() => { + let mut f = in_schema.fields[*id].clone(); + f.table = None; + f + } + expr => { + let (dtype, nullable) = expr.scalar_type(in_schema)?; + Field::plain(String::new(), dtype, nullable) + } + }; + field.name = item + .alias + .clone() + .unwrap_or_else(|| default_proj_name(&item.expr, i, in_schema)); + Ok(match qualifier { + Some(q) => field.with_table(q), + None => field, + }) + }) + .collect::, SchemaDerivationError>>()?; + let time_index = fields.iter().position(|c| c.name == "ts"); + let unique_keys = in_schema + .unique_keys + .iter() + .filter_map(|key| { + key.iter() + .map(|input_col| { + cols.iter().position(|item| { + matches!(&item.expr, ScalarExpr::Column(col) if col == input_col) + }) + }) + .collect::>>() + }) + .collect(); + Schema { + fields, + time_index, + unique_keys, + closed: true, + } + } + + Dedup { cols, child } => { + let mut out = child.schema.clone(); + if !cols.is_empty() { + out.add_unique_key(cols.clone()); + } + out + } + + Concat { + children, + discriminator_unique_key, + } => { + let mut s = children + .first() + .ok_or(SchemaDerivationError::EmptyConcat)? + .schema + .clone(); + s.unique_keys.clear(); + if let Some(key) = discriminator_unique_key { + let mut compound = vec![*key.discriminator()]; + compound.extend(key.inner_key().iter().copied()); + s.add_unique_key(compound); + } + s + } + SetOp { left, .. } => { + let mut s = left.schema.clone(); + s.unique_keys.clear(); + s + } + Join { + kind, left, right, .. + } => { + let l = &left.schema; + let r = &right.schema; + if matches!(kind, JoinKind::Semi | JoinKind::Anti) { + return Ok(Schema { + unique_keys: Vec::new(), + ..l.clone() + }); + } + let (left_null, right_null) = match kind { + JoinKind::Left => (false, true), + JoinKind::Right => (true, false), + JoinKind::Full => (true, true), + JoinKind::Inner | JoinKind::Cross => (false, false), + JoinKind::Semi | JoinKind::Anti => unreachable!("handled above"), + }; + let l_len = l.fields.len(); + let mut fields = Vec::with_capacity(l_len + r.fields.len()); + fields.extend(l.fields.iter().cloned().map(|mut c| { + c.nullable |= left_null; + c + })); + fields.extend(r.fields.iter().cloned().map(|mut c| { + c.nullable |= right_null; + c + })); + let time_index = l.time_index.or(r.time_index.map(|i| i + l_len)); + Schema { + fields, + time_index, + unique_keys: Vec::new(), + closed: l.closed && r.closed, + } + } + SQLWindowFunc { + func, + args, + output_name, + child, + .. + } => { + let mut out = child.schema.clone(); + let arg = args.first().map(|a| a.scalar_type(&out)).transpose()?; + let arg_dtype = || { + arg.as_ref() + .map(|(ty, _)| ty.clone()) + .unwrap_or(DataType::Null) + }; + let (dtype, nullable) = match func { + WindowFuncKind::RowNumber + | WindowFuncKind::Rank + | WindowFuncKind::DenseRank + | WindowFuncKind::Count => (DataType::Int64, false), + WindowFuncKind::Sum => (arg_dtype(), true), + WindowFuncKind::Avg => (DataType::Float64, true), + WindowFuncKind::Lag + | WindowFuncKind::Lead + | WindowFuncKind::LagInFrame + | WindowFuncKind::LeadInFrame + | WindowFuncKind::FirstValue + | WindowFuncKind::LastValue + | WindowFuncKind::NthValue(_) => (arg_dtype(), true), + WindowFuncKind::Min | WindowFuncKind::Max => (arg_dtype(), true), + }; + out.fields + .push(Field::plain(output_name.clone(), dtype, nullable)); + out + } + + // `vector(s)` yields a label-less instant vector: the (ts, value) + // floor and nothing else; its full label set (empty) is known. + PromqlVectorFromScalar(_) => Schema { + fields: vec![ + Field::plain("ts", DataType::Timestamp, false), + Field::plain("value", DataType::Float64, false), + ], + time_index: Some(0), + unique_keys: Vec::new(), + closed: true, + }, + + // The output shape of ` op ` is the vector side's: + // a scalar operand contributes only its value, no labels. A `bool` + // comparison still produces the vector's shape (values 0/1). + BinaryOp { + lhs, rhs, operator, .. + } => { + let mut output = lhs.schema.clone(); + let grouping = operator + .vector_match + .as_ref() + .and_then(|m| m.grouping.as_ref()); + let right_rows = matches!( + operator.kind, + BinaryOpKind::Set(crate::pre_asap::PromQLVectorSetOpKind::Or) + ) || matches!(grouping, Some(g) if g.side == crate::pre_asap::GroupSide::Right); + let mut additions = Vec::new(); + if right_rows { + additions.extend( + rhs.schema + .fields + .iter() + .filter(|c| c.plain_dtype() == Some(&DataType::Utf8)) + .cloned(), + ); + } + if let Some(grouping) = grouping { + additions.extend( + grouping + .labels + .iter() + .map(|name| Field::plain(name.clone(), DataType::Utf8, true)), + ); + } + for column in additions { + if !output.fields.iter().any(|c| c.name == column.name) { + output.fields.push(column); + } + } + output + } + }) + } + + /// The output category derived from this operator and its children. + pub fn output_kind(&self) -> OperatorResultKind { + use NonASAPOp::*; + match self { + Scan { source, .. } => match source { + Source::TimeSeries { .. } => OperatorResultKind::InstantVector, + Source::Table { .. } => OperatorResultKind::Relation, + }, + Values { .. } | SQLWindowFunc { .. } => OperatorResultKind::Relation, + TimeRange { child, .. } if child.result_kind == OperatorResultKind::Relation => { + OperatorResultKind::Relation + } + TimeRange { kind, .. } => match kind { + TimeRangeKind::Instant => OperatorResultKind::InstantVector, + TimeRangeKind::Range => OperatorResultKind::RangeVector, + }, + PromqlSubquery { .. } => OperatorResultKind::RangeVector, + PromqlVectorFromScalar(_) => OperatorResultKind::InstantVector, + // A per-entity range reduction turns a range vector into an + // instant vector; a cross-series reduction keeps its input's + // category (a SQL GROUP BY stays a relation). + Aggregate { + reduction, child, .. + } => match (reduction, child.result_kind) { + (Reduction::PerEntity, OperatorResultKind::RangeVector) + | (_, OperatorResultKind::State) => OperatorResultKind::InstantVector, + (_, kind) => kind, + }, + Project { child, cols, .. } if cols.iter().any(|item| matches!(item.expr, ScalarExpr::Column(i) if child.schema.fields.get(i).is_some_and(|f| !f.is_plain()))) => OperatorResultKind::State, + Filter { child, .. } + | Project { child, .. } + | Dedup { child, .. } + | Sort { child, .. } + | Limit { child, .. } + | TimeShift { child, .. } + | PromqlRelabel { child, .. } + | PromqlInfoEnrich { child, .. } + | PromqlSeriesSample { child, .. } => readable(child.result_kind), + Join { left, .. } | SetOp { left, .. } => readable(left.result_kind), + Concat { children, .. } => children + .first() + .map_or(OperatorResultKind::Relation, |c| readable(c.result_kind)), + BinaryOp { lhs, .. } => readable(lhs.result_kind), + } + } + + /// Local producer/consumer contract checks that need only this operator + /// and its children's output categories. + pub fn validate_inputs(&self) -> Result<(), SchemaDerivationError> { + use NonASAPOp::*; + let no_state = |node: &OperatorNode, what: &str| { + if node.result_kind == OperatorResultKind::State { + Err(SchemaDerivationError::InvalidScalarSignature(format!( + "{what} consumes summary state; read it out first" + ))) + } else { + Ok(()) + } + }; + let invalid = |message: &str| SchemaDerivationError::InvalidScalarSignature(message.into()); + let predicate = |pred: &Predicate, scope: &Schema| -> Result<(), SchemaDerivationError> { + if matches!( + pred.0.scalar_type(scope)?.0, + DataType::Bool | DataType::Null + ) { + Ok(()) + } else { + Err(invalid("predicate must be boolean")) + } + }; + let instant = |child: &OperatorNode| -> Result<(), SchemaDerivationError> { + if child.result_kind == OperatorResultKind::InstantVector { + Ok(()) + } else { + Err(invalid("operation requires an instant vector")) + } + }; + let columns = |cols: &[usize], scope: &Schema| -> Result<(), SchemaDerivationError> { + if cols.iter().any(|i| *i >= scope.fields.len()) { + Err(invalid("column outside operator input scope")) + } else { + Ok(()) + } + }; + match self { + Scan { + predicates, schema, .. + } => { + for pred in predicates { + predicate(pred, schema)?; + } + } + Values { rows, schema } => { + for row in rows { + if row.len() != schema.fields.len() { + return Err(invalid("Values row width differs from its schema")); + } + for (expr, field) in row.iter().zip(&schema.fields) { + let (ty, nullable) = expr.scalar_type(&Schema::default())?; + if field + .plain_dtype() + .is_none_or(|declared| *declared != ty && ty != DataType::Null) + || nullable && !field.nullable + { + return Err(invalid( + "Values expression differs from declared type/nullability", + )); + } + } + } + } + Filter { child, pred } => { + no_state(child, "Filter")?; + predicate(pred, &child.schema)?; + } + Project { child, cols, .. } => { + for col in cols { + if let ScalarExpr::Column(index) = col.expr { + columns(&[index], &child.schema)?; + } else { + col.expr.scalar_type(&child.schema)?; + } + } + } + BinaryOp { + lhs, + rhs, + operator, + return_bool, + } => { + no_state(lhs, "BinaryOp")?; + no_state(rhs, "BinaryOp")?; + if lhs.result_kind != rhs.result_kind + || lhs.result_kind == OperatorResultKind::RangeVector + { + return Err(invalid("binary operands have incompatible result kinds")); + } + if *return_bool && !matches!(operator.kind, BinaryOpKind::Compare(_)) { + return Err(invalid("bool mode requires a comparison")); + } + } + Join { + left, right, pred, .. + } => { + no_state(left, "Join")?; + no_state(right, "Join")?; + if left.result_kind != OperatorResultKind::Relation + || right.result_kind != OperatorResultKind::Relation + { + return Err(invalid("SQL join requires relations")); + } + let scope = Schema::new( + left.schema + .fields + .iter() + .chain(&right.schema.fields) + .cloned() + .collect(), + ); + predicate(pred, &scope)?; + } + SetOp { left, right, .. } => { + no_state(left, "SetOp")?; + no_state(right, "SetOp")?; + if left.result_kind != OperatorResultKind::Relation + || right.result_kind != OperatorResultKind::Relation + { + return Err(invalid("SQL set operation requires relations")); + } + } + Concat { children, .. } => { + for child in children { + no_state(child, "Concat")?; + } + } + Aggregate { + child, + filters, + measures, + having, + reduction, + .. + } => { + no_state(child, "Aggregate")?; + if let Reduction::Reduce(keys) = reduction { + columns(keys.keys(), &child.schema)?; + } + for measure in measures { + columns(&measure.input_cols(), &child.schema)?; + } + if !filters.is_empty() && filters.len() != measures.len() { + return Err(invalid("aggregate filter count differs from measure count")); + } + for pred in filters.iter().flatten() { + predicate(pred, &child.schema)?; + } + if let Some(pred) = having { + predicate(pred, &self.output_schema()?)?; + } + } + Dedup { child, cols } => { + no_state(child, "Dedup")?; + columns(cols, &child.schema)?; + } + Sort { + child, + keys, + partition_by, + } => { + columns(partition_by.keys(), &child.schema)?; + for key in keys { + key.expr.scalar_type(&child.schema)?; + } + } + Limit { + child, + partition_by, + .. + } => columns(partition_by.keys(), &child.schema)?, + SQLWindowFunc { + child, + args, + order_by, + partition_by, + .. + } => { + if child.result_kind != OperatorResultKind::Relation { + return Err(invalid("SQL window requires a relation")); + } + columns(partition_by.keys(), &child.schema)?; + for expr in args.iter().chain(order_by.iter().map(|k| &k.expr)) { + expr.scalar_type(&child.schema)?; + } + } + PromqlVectorFromScalar(expr) => { + if expr.scalar_type(&Schema::default())? != (DataType::Float64, false) { + return Err(invalid("vector() requires a non-null float scalar")); + } + } + PromqlSubquery { child, .. } + | PromqlInfoEnrich { child, .. } + | PromqlSeriesSample { child, .. } => instant(child)?, + PromqlRelabel { child, value, .. } => { + instant(child)?; + if value.scalar_type(&child.schema)?.0 != DataType::Utf8 { + return Err(invalid("label expression requires a string")); + } + } + TimeRange { child, .. } => { + if child.result_kind != OperatorResultKind::Relation { + instant(child)?; + } + } + TimeShift { child, .. } => { + if !matches!( + child.result_kind, + OperatorResultKind::InstantVector | OperatorResultKind::RangeVector + ) { + return Err(invalid("time shift requires a vector")); + } + } + } + Ok(()) + } +} + +/// A evaluation-shaped category for a value-level operator over `kind`: state +/// never flows through an ordinary operator unchanged in category. +fn readable(kind: OperatorResultKind) -> OperatorResultKind { + match kind { + OperatorResultKind::State => OperatorResultKind::Relation, + other => other, + } +} + +/// Default output-column name for a projection item with no explicit alias: +/// a bare column keeps its (schema) name; anything else gets `col_{i}`. +fn default_proj_name(expr: &ScalarExpr, idx: usize, schema: &Schema) -> String { + match expr { + ScalarExpr::Column(id) => schema + .fields + .get(*id) + .map(|c| c.name.clone()) + .unwrap_or_else(|| format!("col_{idx}")), + _ => format!("col_{idx}"), + } +} + +/// Whether any aggregate measure has its own input predicate. +pub fn any_measure_filtered(filters: &[Option]) -> bool { + filters.iter().any(Option::is_some) +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::ir::operator_properties::{ + AtModifier, VectorMatchKind, WindowFrameBound, WindowFrameOffset, WindowFrameUnits, + }; + use crate::ir::scalar::ExprSemantics; + use crate::pre_asap::expr_ir::{ArithmeticOpKind, CompareOpKind, ScalarValue}; + + fn col(name: &str, dtype: DataType, nullable: bool) -> Field { + Field::plain(name, dtype, nullable) + } + + fn node(op: NonASAPOp) -> Rc { + OperatorNode::new_shared(crate::ir::Operator::NonASAP(op)).unwrap() + } + + fn scan( + columns: Vec, + time_index: Option, + uk: Vec>, + ) -> NonASAPOp { + NonASAPOp::Scan { + source: Source::Table { + table_ref: "t".into(), + }, + predicates: vec![], + schema: Schema { + fields: columns, + time_index, + unique_keys: uk, + closed: true, + }, + } + } + + /// `[ts, value, job]` time-series leaf; open, as a PromQL leaf is. + fn series_scan() -> NonASAPOp { + NonASAPOp::Scan { + source: Source::TimeSeries { metric: "m".into() }, + predicates: vec![], + schema: Schema::with_time_index( + vec![ + col("ts", DataType::Timestamp, false), + col("value", DataType::Float64, false), + col("job", DataType::Utf8, true), + ], + 0, + vec![], + ), + } + } + + fn item(alias: Option<&str>, expr: ScalarExpr) -> ProjectItem { + ProjectItem { + alias: alias.map(Into::into), + expr, + } + } + + fn add(left: ScalarExpr, right: ScalarExpr) -> ScalarExpr { + ScalarExpr::Arithmetic { + op: ArithmeticOpKind::Add, + left: Box::new(left), + right: Box::new(right), + semantics: ExprSemantics::Sql, + } + } + + fn project(cols: Vec, child: NonASAPOp) -> NonASAPOp { + NonASAPOp::Project { + cols, + qualifier: None, + child: node(child), + } + } + + fn dedup_branch(columns: Vec) -> Rc { + node(NonASAPOp::Dedup { + cols: vec![0], + child: node(scan(columns, None, vec![])), + }) + } + + fn concat( + children: Vec>, + discriminator_unique_key: Option, + ) -> NonASAPOp { + NonASAPOp::Concat { + children, + discriminator_unique_key, + } + } + + fn rate_over(child: NonASAPOp) -> NonASAPOp { + NonASAPOp::Aggregate { + reduction: Reduction::PerEntity, + measures: vec![AggIntent::Rate], + output_names: vec![], + filters: vec![], + having: None, + child: node(child), + } + } + + #[test] + fn project_preserves_unique_keys_that_are_passed_through() { + let input = scan( + vec![ + col("tenant", DataType::Utf8, false), + col("region", DataType::Utf8, false), + col("value", DataType::Int64, false), + ], + None, + vec![vec![0, 1]], + ); + let projected = project( + vec![ + item(Some("r"), ScalarExpr::Column(1)), + item(Some("t"), ScalarExpr::Column(0)), + item( + None, + add( + ScalarExpr::Column(2), + ScalarExpr::Literal(ScalarValue::Int64(1)), + ), + ), + ], + input, + ); + assert_eq!( + projected.output_schema().unwrap().unique_keys, + vec![vec![1, 0]] + ); + } + + #[test] + fn project_drops_a_unique_key_when_a_key_column_is_omitted() { + let input = scan( + vec![ + col("tenant", DataType::Utf8, false), + col("region", DataType::Utf8, false), + ], + None, + vec![vec![0, 1]], + ); + let projected = project(vec![item(None, ScalarExpr::Column(0))], input); + assert!(projected.output_schema().unwrap().unique_keys.is_empty()); + } + + #[test] + fn project_retypes_and_renames_per_item() { + let child = scan( + vec![ + col("ts", DataType::Timestamp, false), + col("host", DataType::Utf8, false), + col("value", DataType::Float64, false), + ], + Some(0), + vec![vec![0, 1]], + ); + let q = project( + vec![ + // A bare column keeps its name and type. + item(None, ScalarExpr::Column(1)), + item( + Some("dbl"), + add(ScalarExpr::Column(2), ScalarExpr::Column(2)), + ), + // A comparison is a nullable Bool under 3-valued logic. + item( + Some("flag"), + ScalarExpr::Compare { + left: Box::new(ScalarExpr::Column(2)), + op: CompareOpKind::Gt, + right: Box::new(ScalarExpr::Literal(ScalarValue::Float64(0.0))), + semantics: ExprSemantics::Sql, + }, + ), + ], + child, + ); + let s = q.output_schema().unwrap(); + assert_eq!(s.fields.len(), 3); + assert_eq!(s.fields[0], col("host", DataType::Utf8, false)); + assert_eq!(s.fields[1], col("dbl", DataType::Float64, false)); + assert_eq!(s.fields[2], col("flag", DataType::Bool, false)); + // `ts` is not retained: no time axis, and the key is lost. + assert!(s.time_index.is_none()); + assert!(s.unique_keys.is_empty()); + } + + #[test] + fn project_keeps_time_index_when_ts_passed_through() { + let child = scan( + vec![ + col("ts", DataType::Timestamp, false), + col("value", DataType::Float64, false), + ], + Some(0), + vec![], + ); + let q = project( + vec![ + item(None, ScalarExpr::Column(1)), + item(None, ScalarExpr::Column(0)), + ], + child, + ); + let s = q.output_schema().unwrap(); + assert_eq!(s.fields[0].name, "value"); + assert_eq!(s.fields[1].name, "ts"); + assert_eq!(s.time_index, Some(1)); + } + + #[test] + fn legacy_window_json_without_frame_deserializes_as_unspecified() { + let window = NonASAPOp::SQLWindowFunc { + func: WindowFuncKind::RowNumber, + args: vec![], + partition_by: GroupKeys::by(vec![]), + order_by: vec![], + frame: Some(WindowFrame { + units: WindowFrameUnits::Range, + start_bound: WindowFrameBound::Preceding(WindowFrameOffset::Scalar( + ScalarValue::Null, + )), + end_bound: WindowFrameBound::CurrentRow, + }), + output_name: "row_number".into(), + child: node(scan(vec![col("v", DataType::Int64, false)], None, vec![])), + }; + let mut json = serde_json::to_value(window).unwrap(); + json.get_mut("SQLWindowFunc") + .and_then(serde_json::Value::as_object_mut) + .unwrap() + .remove("frame"); + let decoded: NonASAPOp = serde_json::from_value(json).unwrap(); + assert!(matches!( + decoded, + NonASAPOp::SQLWindowFunc { frame: None, .. } + )); + } + + /// A row can appear in more than one branch, so no branch's unique key is + /// a key of the union — `unique_keys` feeds CSE's sharing legality check. + #[test] + fn merge_drops_the_branches_unique_keys() { + let branch = || { + dedup_branch(vec![ + col("k", DataType::Utf8, false), + col("v", DataType::Int64, false), + ]) + }; + assert_eq!( + branch().schema.unique_keys, + vec![vec![0]], + "a Dedup branch does have a unique key on its own" + ); + let schema = concat(vec![branch(), branch()], None) + .output_schema() + .unwrap(); + assert!( + schema.unique_keys.is_empty(), + "the union of two deduplicated branches is not deduplicated" + ); + assert_eq!(schema.fields.len(), 2, "column shape is the first branch's"); + } + + #[test] + fn merge_and_setop_agree_on_unique_keys() { + let branch = || dedup_branch(vec![col("k", DataType::Utf8, false)]); + let merged = concat(vec![branch(), branch()], None); + let setop = NonASAPOp::SetOp { + kind: RelationalSetOpKind::Union, + all: true, + left: branch(), + right: branch(), + }; + assert_eq!( + merged.output_schema().unwrap().unique_keys, + setop.output_schema().unwrap().unique_keys, + ); + } + + #[test] + fn an_empty_merge_has_no_schema() { + assert!(matches!( + concat(vec![], None).output_schema(), + Err(SchemaDerivationError::EmptyConcat) + )); + } + + /// Issue #228: an asserted discriminator yields the compound + /// `(discriminator, inner_key)` unique key, although each branch's own + /// `inner_key` repeats across branches. + #[test] + fn discriminator_override_produces_a_compound_unique_key() { + let branch = || { + dedup_branch(vec![ + col("k", DataType::Utf8, false), + col("branch_id", DataType::Int64, false), + ]) + }; + let schema = concat( + vec![branch(), branch()], + Some(ConcatDiscriminatorKey::new(1, vec![0])), + ) + .output_schema() + .unwrap(); + assert_eq!(schema.unique_keys, vec![vec![1, 0]]); + assert_eq!(schema.fields.len(), 2, "column shape is the first branch's"); + } + + /// Without a named discriminator a `Concat` never claims a unique key. + #[test] + fn no_way_to_fabricate_a_unique_key_without_naming_a_discriminator() { + let branch = || dedup_branch(vec![col("k", DataType::Utf8, false)]); + assert!(concat(vec![branch(), branch()], None) + .output_schema() + .unwrap() + .unique_keys + .is_empty()); + } + + #[test] + fn without_aggregate_keeps_open_schema_minus_excluded() { + // `sum without (instance) (m)` over `[ts, value, instance, job]`. + let leaf = NonASAPOp::Scan { + source: Source::TimeSeries { metric: "m".into() }, + predicates: vec![], + schema: Schema::with_time_index( + vec![ + col("ts", DataType::Timestamp, false), + col("value", DataType::Float64, false), + col("instance", DataType::Utf8, true), + col("job", DataType::Utf8, true), + ], + 0, + vec![], + ), + }; + let agg = NonASAPOp::Aggregate { + reduction: Reduction::Reduce(GroupKeys::without(vec![2])), + measures: vec![AggIntent::Sum { col: None }], + output_names: vec![], + filters: vec![], + having: None, + child: node(leaf), + }; + let s = agg.output_schema().unwrap(); + let names: Vec<_> = s.fields.iter().map(|c| c.name.as_str()).collect(); + assert_eq!(names, vec!["job", "sum"], "kept `job`, dropped `instance`"); + assert!(!s.closed, "a `without` result stays open"); + assert!(s.time_index.is_none()); + assert!(s.unique_keys.is_empty(), "kept set unknown → no unique key"); + } + + #[test] + fn time_shift_is_schema_pass_through() { + let leaf = node(scan( + vec![ + col("ts", DataType::Timestamp, false), + col("value", DataType::Float64, false), + col("job", DataType::Utf8, true), + ], + Some(0), + vec![], + )); + let shifted = NonASAPOp::TimeShift { + shift: TimeShift { + offset_ms: 3_600_000, + at: Some(AtModifier::Timestamp(1_609_746_000_000)), + }, + child: Rc::clone(&leaf), + }; + assert_eq!(shifted.output_schema().unwrap(), leaf.schema); + } + + /// `rate` and `*_over_time` are per-series: every label survives and only + /// the sample value is replaced (kept named `value`). + #[test] + fn per_series_reductions_preserve_labels() { + for measure in [AggIntent::Rate, AggIntent::Avg { col: None }] { + let reduced = NonASAPOp::Aggregate { + reduction: Reduction::PerEntity, + measures: vec![measure], + output_names: vec![], + filters: vec![], + having: None, + child: node(NonASAPOp::TimeRange { + range: Duration::from_secs(300), + kind: TimeRangeKind::Range, + child: node(series_scan()), + }), + }; + let s = reduced.output_schema().unwrap(); + let names: Vec<_> = s.fields.iter().map(|c| c.name.as_str()).collect(); + assert_eq!(names, vec!["ts", "value", "job"]); + assert_eq!(s.time_index, Some(0)); + } + } + + /// An open leaf stays open through a per-series `rate` and is frozen to + /// closed by a cross-series aggregate. + #[test] + fn completeness_open_leaf_freezes_to_closed_at_cross_series_aggregate() { + let leaf = series_scan(); + assert!( + !leaf.output_schema().unwrap().closed, + "schemaless leaf is open" + ); + let rate = rate_over(leaf); + assert!(!rate.output_schema().unwrap().closed, "rate stays open"); + let sum_by_job = NonASAPOp::Aggregate { + reduction: Reduction::by(vec![2]), + measures: vec![AggIntent::Sum { col: None }], + output_names: vec![], + filters: vec![], + having: None, + child: node(rate), + }; + assert!(sum_by_job.output_schema().unwrap().closed); + } + + fn join(kind: JoinKind) -> NonASAPOp { + NonASAPOp::Join { + kind, + pred: Predicate(ScalarExpr::Literal(ScalarValue::Boolean(true))), + left: node(scan( + vec![col("a", DataType::Int64, false)], + None, + vec![vec![0]], + )), + right: node(scan(vec![col("b", DataType::Utf8, false)], None, vec![])), + } + } + + #[test] + fn inner_join_concatenates_both_sides() { + let s = join(JoinKind::Inner).output_schema().unwrap(); + assert_eq!( + s.fields, + vec![ + col("a", DataType::Int64, false), + col("b", DataType::Utf8, false) + ] + ); + assert!( + s.unique_keys.is_empty(), + "post-join row identity not provable" + ); + } + + #[test] + fn left_join_makes_right_side_nullable() { + let s = join(JoinKind::Left).output_schema().unwrap(); + assert!(!s.fields[0].nullable); + assert!(s.fields[1].nullable); + } + + #[test] + fn full_join_makes_both_sides_nullable() { + let s = join(JoinKind::Full).output_schema().unwrap(); + assert!(s.fields[0].nullable); + assert!(s.fields[1].nullable); + } + + #[test] + fn setop_takes_left_shape_and_drops_unique_keys() { + let side = || { + node(scan( + vec![ + col("k", DataType::Utf8, false), + col("v", DataType::Int64, false), + ], + None, + vec![vec![0]], + )) + }; + let s = NonASAPOp::SetOp { + kind: RelationalSetOpKind::Union, + all: false, + left: side(), + right: side(), + } + .output_schema() + .unwrap(); + assert_eq!(s.fields.len(), 2); + assert_eq!(s.fields[0].name, "k"); + assert!( + s.unique_keys.is_empty(), + "UNION does not preserve row identity" + ); + } + + /// Standalone constants are scalar roots and carry no operator schema. + #[test] + fn constant_is_a_scalar_root() { + let root = crate::ir::QueryRoot::Scalar(ScalarExpr::literal_f64(42.0)); + assert!(root.as_operator().is_none()); + } + + /// ` op ` takes the vector side's schema; the vector + /// match modifier is kept on the operator. + #[test] + fn binary_op_schema_follows_the_vector_side_over_a_scalar_bridge() { + let vector = node(scan( + vec![ + col("host", DataType::Utf8, false), + col("value", DataType::Float64, false), + ], + None, + vec![], + )); + let vm = VectorMatch { + kind: VectorMatchKind::On, + labels: vec!["host".into()], + grouping: None, + }; + let op = NonASAPOp::BinaryOp { + operator: BinaryOperator { + checked_relative_division: false, + checked_finite_division: false, + kind: BinaryOpKind::Compare(CompareOpKind::Gt), + vector_match: Some(vm.clone()), + }, + return_bool: false, + lhs: Rc::clone(&vector), + rhs: Rc::clone(&vector), + }; + assert_eq!(op.output_schema().unwrap(), vector.schema); + let NonASAPOp::BinaryOp { operator, .. } = &op else { + unreachable!() + }; + assert_eq!(operator.vector_match.as_ref(), Some(&vm)); + } +} diff --git a/crates/types/src/ir/operator_properties.rs b/crates/types/src/ir/operator_properties.rs new file mode 100644 index 000000000..4737e6278 --- /dev/null +++ b/crates/types/src/ir/operator_properties.rs @@ -0,0 +1,8 @@ +//! Operator parameters shared with the existing dag during migration. +//! Definitions move here when legacy dag consumers are removed. +pub use crate::pre_asap::query_expr::{ + AtModifier, BinaryOpKind, ColState, ConcatDiscriminatorKey, DataModel, GroupKeys, GroupSide, + InfoMatcher, JoinKind, PromQLVectorSetOpKind, Reduction, RelationalSetOpKind, SampleKind, + Source, TimeShift, VectorGrouping, VectorMatch, VectorMatchKind, WindowFrame, WindowFrameBound, + WindowFrameOffset, WindowFrameUnits, WindowFuncKind, +}; diff --git a/crates/types/src/ir/query.rs b/crates/types/src/ir/query.rs new file mode 100644 index 000000000..682c59465 --- /dev/null +++ b/crates/types/src/ir/query.rs @@ -0,0 +1,36 @@ +//! Query results are either an operator result or a standalone scalar expression. +//! The root discriminator is not an operator and never creates a dag node. +use super::{OperatorNode, ScalarExpr}; +use serde::{Deserialize, Serialize}; +use std::rc::Rc; +#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)] +pub enum QueryRoot { + Operator(Rc), + Scalar(ScalarExpr), +} +impl From> for QueryRoot { + fn from(node: Rc) -> Self { + Self::Operator(node) + } +} +impl QueryRoot { + pub fn validate_structure(&self) -> Result<(), crate::ir::SchemaDerivationError> { + match self { + Self::Operator(node) => node.validate_structure(), + Self::Scalar(expr) => { + expr.scalar_type(&crate::pre_asap::Schema::default())?; + for node in expr.operator_refs() { + node.validate_structure()?; + } + Ok(()) + } + } + } + + pub fn as_operator(&self) -> Option<&Rc> { + match self { + Self::Operator(node) => Some(node), + Self::Scalar(_) => None, + } + } +} diff --git a/crates/types/src/ir/scalar.rs b/crates/types/src/ir/scalar.rs new file mode 100644 index 000000000..c1a0b90a1 --- /dev/null +++ b/crates/types/src/ir/scalar.rs @@ -0,0 +1,990 @@ +//! Scalar expressions: value computation evaluated within the schema chosen by +//! the operator that owns them. +//! +//! A [`ScalarExpr`] never produces a table. It is owned by value by an operator +//! field (`Filter.pred`, `ProjectItem.expr`, `SortKey.expr`, `HAVING`, window +//! arguments, relabel values) or by a [`super::QueryRoot::Scalar`] query root. +//! The only operator references inside a scalar tree are the explicit +//! plan-reading variants (`PromqlScalarFromVector`, `ScalarSubquery`, `Exists`, +//! `InSubquery`); every traversal of the operator DAG follows them. + +use std::rc::Rc; + +use serde::{Deserialize, Serialize}; + +use super::node::OperatorNode; +use crate::ir::SchemaDerivationError; +use crate::pre_asap::expr_ir::{ArithmeticOpKind, CompareOpKind, ScalarValue}; +use crate::pre_asap::scalar_type_rules::MapScalarFunction; +use crate::pre_asap::schema::{ColumnId, DataType, Schema}; + +/// Which language's numeric and comparison rules an expression follows. +/// Both languages use `Float64`, so a result type alone does not preserve +/// NaN, ordering or error rules. +#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, Serialize, Deserialize)] +#[serde(rename_all = "snake_case")] +pub enum ExprSemantics { + Sql, + Promql, +} + +/// A scalar expression over the owning operator's input schema. Column +/// references are positional [`ColumnId`]s. +#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)] +pub enum ScalarExpr { + Column(ColumnId), + Literal(ScalarValue), + /// Unary minus. + Negative { + expr: Box, + semantics: ExprSemantics, + }, + Compare { + left: Box, + op: CompareOpKind, + right: Box, + semantics: ExprSemantics, + }, + /// Flat conjunction (logical AND). An empty list is vacuously true. + BoolAnd(Vec), + /// Flat disjunction (logical OR). An empty list is vacuously false. + BoolOr(Vec), + Not(Box), + IsNull(Box), + IsNotNull(Box), + /// `CAST(expr AS to)`; `try_cast` for SQL `TRY_CAST` (NULL on failure). + Cast { + expr: Box, + to: DataType, + try_cast: bool, + }, + /// `expr [NOT] IN (v1, v2, …)`. + InList { + expr: Box, + list: Vec, + negated: bool, + }, + /// Scalar function call, e.g. `LOWER(col)`, `ABS(x)`. + FunctionCall { + name: String, + args: Vec, + }, + Arithmetic { + op: ArithmeticOpKind, + left: Box, + right: Box, + semantics: ExprSemantics, + }, + /// SQL `CASE` (both searched and simple forms). `operand` present for the + /// simple form (`CASE expr WHEN …`), absent for searched. + Case { + operand: Option>, + branches: Vec<(ScalarExpr, ScalarExpr)>, + else_expr: Option>, + }, + /// SQL `NOW()` / `CURRENT_TIMESTAMP`: the statement evaluation time. + CurrentTimestamp, + /// PromQL `time()`: the evaluation instant as Unix seconds (`Float64`). + EvalTimestamp, + /// PromQL `scalar(v)`: the single sample of an instant vector, NaN + /// otherwise. The referenced operator is a real plan dependency. + PromqlScalarFromVector(Rc), + /// An uncorrelated SQL scalar subquery: one column; zero rows is NULL, + /// more than one row is an error. + ScalarSubquery(Rc), + /// SQL `[NOT] EXISTS (subquery)`. + Exists { + subquery: Rc, + negated: bool, + }, + /// SQL `expr [NOT] IN (subquery)` over a one-column relation. + InSubquery { + expr: Box, + subquery: Rc, + negated: bool, + }, +} + +/// A row-level filter predicate (WHERE clause / PromQL label matcher). +#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)] +pub struct Predicate(pub ScalarExpr); + +/// One item in a SELECT projection list. +#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)] +pub struct ProjectItem { + pub alias: Option, + pub expr: ScalarExpr, +} + +#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)] +pub struct SortKey { + pub expr: ScalarExpr, + pub ascending: bool, + pub nulls_first: bool, +} + +impl ScalarExpr { + pub fn literal_f64(v: f64) -> Self { + ScalarExpr::Literal(ScalarValue::Float64(v)) + } + + pub fn column(id: ColumnId) -> Self { + ScalarExpr::Column(id) + } + + /// If this expression is a `BoolAnd`, its elements; otherwise `self` alone. + pub fn conjuncts(&self) -> &[ScalarExpr] { + match self { + ScalarExpr::BoolAnd(v) => v.as_slice(), + _ => std::slice::from_ref(self), + } + } + + /// If this expression is a `BoolOr`, its elements; otherwise `self` alone. + pub fn disjuncts(&self) -> &[ScalarExpr] { + match self { + ScalarExpr::BoolOr(v) => v.as_slice(), + _ => std::slice::from_ref(self), + } + } + + /// The direct scalar sub-expressions. + pub fn children(&self) -> Vec<&ScalarExpr> { + match self { + ScalarExpr::Column(_) + | ScalarExpr::Literal(_) + | ScalarExpr::CurrentTimestamp + | ScalarExpr::EvalTimestamp + | ScalarExpr::PromqlScalarFromVector(_) + | ScalarExpr::ScalarSubquery(_) + | ScalarExpr::Exists { .. } => vec![], + ScalarExpr::Negative { expr, .. } + | ScalarExpr::Not(expr) + | ScalarExpr::IsNull(expr) + | ScalarExpr::IsNotNull(expr) + | ScalarExpr::Cast { expr, .. } + | ScalarExpr::InSubquery { expr, .. } => vec![expr], + ScalarExpr::Compare { left, right, .. } + | ScalarExpr::Arithmetic { left, right, .. } => { + vec![left, right] + } + ScalarExpr::BoolAnd(parts) | ScalarExpr::BoolOr(parts) => parts.iter().collect(), + ScalarExpr::InList { expr, list, .. } => { + let mut v = vec![expr.as_ref()]; + v.extend(list.iter()); + v + } + ScalarExpr::FunctionCall { args, .. } => args.iter().collect(), + ScalarExpr::Case { + operand, + branches, + else_expr, + } => { + let mut v = Vec::new(); + if let Some(op) = operand { + v.push(op.as_ref()); + } + for (when, then) in branches { + v.push(when); + v.push(then); + } + if let Some(e) = else_expr { + v.push(e.as_ref()); + } + v + } + } + } + + /// The operator nodes this expression (transitively) reads: the explicit + /// plan-reading variants. Every DAG traversal must follow these. + pub fn operator_refs(&self) -> Vec<&Rc> { + let mut out = Vec::new(); + self.collect_operator_refs(&mut out); + out + } + + fn collect_operator_refs<'a>(&'a self, out: &mut Vec<&'a Rc>) { + match self { + ScalarExpr::PromqlScalarFromVector(node) | ScalarExpr::ScalarSubquery(node) => { + out.push(node) + } + ScalarExpr::Exists { subquery, .. } => out.push(subquery), + ScalarExpr::InSubquery { subquery, .. } => out.push(subquery), + _ => {} + } + for child in self.children() { + child.collect_operator_refs(out); + } + } + + /// Rebuild this expression with `f` applied to every operator node it + /// reads (recursively through scalar children). + pub fn map_operator_refs( + &self, + f: &mut impl FnMut(&Rc) -> Rc, + ) -> ScalarExpr { + fn map_box) -> Rc>( + e: &ScalarExpr, + f: &mut F, + ) -> Box { + Box::new(e.map_operator_refs(f)) + } + match self { + ScalarExpr::Column(_) + | ScalarExpr::Literal(_) + | ScalarExpr::CurrentTimestamp + | ScalarExpr::EvalTimestamp => self.clone(), + ScalarExpr::PromqlScalarFromVector(node) => ScalarExpr::PromqlScalarFromVector(f(node)), + ScalarExpr::ScalarSubquery(node) => ScalarExpr::ScalarSubquery(f(node)), + ScalarExpr::Exists { subquery, negated } => ScalarExpr::Exists { + subquery: f(subquery), + negated: *negated, + }, + ScalarExpr::InSubquery { + expr, + subquery, + negated, + } => ScalarExpr::InSubquery { + expr: map_box(expr, f), + subquery: f(subquery), + negated: *negated, + }, + ScalarExpr::Negative { expr, semantics } => ScalarExpr::Negative { + expr: map_box(expr, f), + semantics: *semantics, + }, + ScalarExpr::Compare { + left, + op, + right, + semantics, + } => ScalarExpr::Compare { + left: map_box(left, f), + op: op.clone(), + right: map_box(right, f), + semantics: *semantics, + }, + ScalarExpr::BoolAnd(parts) => { + ScalarExpr::BoolAnd(parts.iter().map(|p| p.map_operator_refs(f)).collect()) + } + ScalarExpr::BoolOr(parts) => { + ScalarExpr::BoolOr(parts.iter().map(|p| p.map_operator_refs(f)).collect()) + } + ScalarExpr::Not(e) => ScalarExpr::Not(map_box(e, f)), + ScalarExpr::IsNull(e) => ScalarExpr::IsNull(map_box(e, f)), + ScalarExpr::IsNotNull(e) => ScalarExpr::IsNotNull(map_box(e, f)), + ScalarExpr::Cast { expr, to, try_cast } => ScalarExpr::Cast { + expr: map_box(expr, f), + to: to.clone(), + try_cast: *try_cast, + }, + ScalarExpr::InList { + expr, + list, + negated, + } => ScalarExpr::InList { + expr: map_box(expr, f), + list: list.iter().map(|p| p.map_operator_refs(f)).collect(), + negated: *negated, + }, + ScalarExpr::FunctionCall { name, args } => ScalarExpr::FunctionCall { + name: name.clone(), + args: args.iter().map(|p| p.map_operator_refs(f)).collect(), + }, + ScalarExpr::Arithmetic { + op, + left, + right, + semantics, + } => ScalarExpr::Arithmetic { + op: op.clone(), + left: map_box(left, f), + right: map_box(right, f), + semantics: *semantics, + }, + ScalarExpr::Case { + operand, + branches, + else_expr, + } => ScalarExpr::Case { + operand: operand.as_ref().map(|e| map_box(e, f)), + branches: branches + .iter() + .map(|(w, t)| (w.map_operator_refs(f), t.map_operator_refs(f))) + .collect(), + else_expr: else_expr.as_ref().map(|e| map_box(e, f)), + }, + } + } + + /// Every column referenced anywhere in this expression (not inside + /// referenced operator subgraphs, which have their own scope). + pub fn columns_referenced(&self) -> Vec { + let mut out = Vec::new(); + self.collect_columns(&mut out); + out + } + + fn collect_columns(&self, out: &mut Vec) { + if let ScalarExpr::Column(id) = self { + out.push(*id); + } + for child in self.children() { + child.collect_columns(out); + } + } + + /// Infer the `(DataType, nullable)` this expression produces against the + /// input schema its owner evaluates it in. Unregistered functions are + /// rejected. A reference to a field carrying summary state is an error: + /// state must be read out before an expression can use it. + pub fn scalar_type(&self, schema: &Schema) -> Result<(DataType, bool), SchemaDerivationError> { + Ok(match self { + ScalarExpr::CurrentTimestamp => (DataType::Timestamp, false), + ScalarExpr::EvalTimestamp => (DataType::Float64, false), + ScalarExpr::Column(id) => match schema.fields.get(*id) { + Some(c) => match c.plain_dtype() { + Some(dtype) => (dtype.clone(), c.nullable), + None => { + return Err(SchemaDerivationError::InvalidScalarSignature(format!( + "column `{}` carries summary state and cannot be read as a value", + c.name + ))) + } + }, + None => return Err(signature("column outside scalar input scope")), + }, + ScalarExpr::Literal(s) => match s { + ScalarValue::Int64(_) => (DataType::Int64, false), + ScalarValue::Float64(_) => (DataType::Float64, false), + ScalarValue::Utf8(_) => (DataType::Utf8, false), + ScalarValue::Boolean(_) => (DataType::Bool, false), + ScalarValue::Null => (DataType::Null, true), + ScalarValue::Interval { .. } => (DataType::Interval, false), + }, + ScalarExpr::Compare { + left, right, op, .. + } => { + let (lt, ln) = left.scalar_type(schema)?; + let (rt, rn) = right.scalar_type(schema)?; + common_scalar_type(<, &rt)?; + if matches!( + op, + CompareOpKind::Regex + | CompareOpKind::NotRegex + | CompareOpKind::Like + | CompareOpKind::NotLike + | CompareOpKind::ILike + | CompareOpKind::NotILike + ) && (lt != DataType::Utf8 || rt != DataType::Utf8) + { + return Err(signature("pattern comparison requires strings")); + } + (DataType::Bool, ln || rn) + } + ScalarExpr::BoolAnd(parts) | ScalarExpr::BoolOr(parts) => { + let mut nullable = false; + for part in parts { + let (ty, n) = part.scalar_type(schema)?; + require_bool(&ty)?; + nullable |= n; + } + (DataType::Bool, nullable) + } + ScalarExpr::Not(expr) => { + let (ty, n) = expr.scalar_type(schema)?; + require_bool(&ty)?; + (DataType::Bool, n) + } + ScalarExpr::IsNull(expr) | ScalarExpr::IsNotNull(expr) => { + expr.scalar_type(schema)?; + (DataType::Bool, false) + } + ScalarExpr::InList { expr, list, .. } => { + let (ty, mut nullable) = expr.scalar_type(schema)?; + for item in list { + let (other, n) = item.scalar_type(schema)?; + common_scalar_type(&ty, &other)?; + nullable |= n; + } + (DataType::Bool, nullable) + } + ScalarExpr::Exists { subquery, .. } => { + relation(subquery)?; + (DataType::Bool, false) + } + ScalarExpr::InSubquery { expr, subquery, .. } => { + let (ty, _) = expr.scalar_type(schema)?; + let field = scalar_subquery_field(subquery)?; + common_scalar_type( + &ty, + field + .plain_dtype() + .ok_or_else(|| signature("subquery returns state"))?, + )?; + (DataType::Bool, true) + } + ScalarExpr::Negative { expr, .. } => { + let (dtype, nullable) = expr.scalar_type(schema)?; + if !numeric(&dtype) && dtype != DataType::Interval { + return Err(signature("negation requires a number or interval")); + } + (dtype, nullable) + } + ScalarExpr::Arithmetic { + op, left, right, .. + } => { + let (lt, ln) = left.scalar_type(schema)?; + let (rt, rn) = right.scalar_type(schema)?; + // Temporal subtraction yields a fixed duration with a unit, not a + // calendar interval or a floating-point number. Until the IR can + // preserve that unit, fail instead of publishing a numeric schema. + if matches!(op, ArithmeticOpKind::Sub) + && matches!(lt, DataType::Date | DataType::Timestamp) + && matches!(rt, DataType::Date | DataType::Timestamp) + { + return Err(SchemaDerivationError::InvalidScalarSignature( + "temporal subtraction produces an unsupported duration type".into(), + )); + } + let dtype = match (<, &rt) { + (DataType::Int64, DataType::Interval) + | (DataType::Interval, DataType::Int64) + if matches!(op, ArithmeticOpKind::Mul) => + { + DataType::Interval + } + (DataType::Timestamp, DataType::Interval) + if matches!(op, ArithmeticOpKind::Add | ArithmeticOpKind::Sub) => + { + DataType::Timestamp + } + (DataType::Interval, DataType::Timestamp) + if matches!(op, ArithmeticOpKind::Add) => + { + DataType::Timestamp + } + (DataType::Date, DataType::Interval) + if matches!(op, ArithmeticOpKind::Add | ArithmeticOpKind::Sub) => + { + DataType::Date + } + (DataType::Interval, DataType::Date) if matches!(op, ArithmeticOpKind::Add) => { + DataType::Date + } + (DataType::Interval, DataType::Interval) + if matches!(op, ArithmeticOpKind::Add | ArithmeticOpKind::Sub) => + { + DataType::Interval + } + (DataType::Int64, DataType::Int64) => DataType::Int64, + _ if numeric(<) && numeric(&rt) => common_scalar_type(<, &rt)?, + _ => return Err(signature("invalid arithmetic operand types")), + }; + (dtype, ln || rn) + } + ScalarExpr::Cast { to, try_cast, expr } => { + let (_, nullable) = expr.scalar_type(schema)?; + (to.clone(), *try_cast || nullable) + } + ScalarExpr::FunctionCall { name, args } => { + if let Some(arity) = crate::pre_asap::scalar_type_rules::promql_function_arity(name) + { + if args.len() != arity + || args + .iter() + .map(|a| a.scalar_type(schema)) + .collect::, _>>()? + .iter() + .any(|t| *t != (DataType::Float64, false)) + { + return Err(signature("PromQL function requires non-null float arguments of the declared arity")); + } + (DataType::Float64, false) + } else if name == "promql_drop_metric_name" { + if args.len() != 1 || args[0].scalar_type(schema)? != (DataType::Utf8, false) { + return Err(SchemaDerivationError::InvalidScalarSignature( + "metric-name removal requires one non-null series identity".into(), + )); + } + (DataType::Utf8, false) + } else if name == "asap_element_access" { + element_access_type(args, schema) + .map_err(SchemaDerivationError::InvalidScalarSignature)? + } else if name == "asap_struct_field" { + struct_field_type(args, schema) + .map_err(SchemaDerivationError::InvalidScalarSignature)? + } else if let Some(function) = MapScalarFunction::from_name(name) { + let arguments = args + .iter() + .map(|arg| arg.scalar_type(schema)) + .collect::, _>>()?; + function + .output_type(&arguments) + .map_err(SchemaDerivationError::InvalidScalarSignature)? + } else { + sql_function_type(name, args, schema)? + } + } + ScalarExpr::Case { + operand, + branches, + else_expr, + } => { + let operand = operand + .as_ref() + .map(|e| e.scalar_type(schema)) + .transpose()?; + let mut dtype = DataType::Null; + let mut nullable = else_expr.is_none(); + for (condition, value) in branches { + let (condition, _) = condition.scalar_type(schema)?; + if let Some((ty, _)) = &operand { + common_scalar_type(ty, &condition)?; + } else { + require_bool(&condition)?; + } + let (ty, n) = value.scalar_type(schema)?; + dtype = common_scalar_type(&dtype, &ty)?; + nullable |= n; + } + if let Some(value) = else_expr { + let (ty, n) = value.scalar_type(schema)?; + dtype = common_scalar_type(&dtype, &ty)?; + nullable |= n; + } + (dtype, nullable) + } + // `scalar(v)` is one float sample (NaN when the vector is not + // exactly one series); a scalar subquery is its single column. + ScalarExpr::PromqlScalarFromVector(node) => { + if node.result_kind != super::OperatorResultKind::InstantVector { + return Err(signature("scalar() requires an instant vector")); + } + (DataType::Float64, false) + } + ScalarExpr::ScalarSubquery(node) => { + let field = scalar_subquery_field(node)?; + ( + field + .plain_dtype() + .ok_or_else(|| signature("scalar subquery returns state"))? + .clone(), + true, + ) + } + }) + } +} + +fn signature(message: &str) -> SchemaDerivationError { + SchemaDerivationError::InvalidScalarSignature(message.into()) +} +fn numeric(ty: &DataType) -> bool { + matches!(ty, DataType::Int64 | DataType::Float64 | DataType::Null) +} +fn require_bool(ty: &DataType) -> Result<(), SchemaDerivationError> { + if matches!(ty, DataType::Bool | DataType::Null) { + Ok(()) + } else { + Err(signature("boolean expression required")) + } +} +fn common_scalar_type(a: &DataType, b: &DataType) -> Result { + if a == b || *b == DataType::Null { + Ok(a.clone()) + } else if *a == DataType::Null { + Ok(b.clone()) + } else if numeric(a) && numeric(b) { + Ok(DataType::Float64) + } else { + Err(signature("incompatible scalar types")) + } +} +fn relation(node: &OperatorNode) -> Result<(), SchemaDerivationError> { + if node.result_kind == super::OperatorResultKind::Relation { + Ok(()) + } else { + Err(signature("SQL subquery requires a relation")) + } +} +fn scalar_subquery_field( + node: &OperatorNode, +) -> Result<&crate::pre_asap::Field, SchemaDerivationError> { + relation(node)?; + match node.schema.fields.as_slice() { + [field] => Ok(field), + _ => Err(signature("scalar subquery requires exactly one column")), + } +} +fn sql_function_type( + name: &str, + args: &[ScalarExpr], + schema: &Schema, +) -> Result<(DataType, bool), SchemaDerivationError> { + let types = args + .iter() + .map(|a| a.scalar_type(schema)) + .collect::, _>>()?; + let nullable = types.iter().any(|(_, n)| *n); + let name = name.to_ascii_lowercase(); + match (name.as_str(), types.as_slice()) { + ("abs" | "ceil" | "floor" | "round", [(ty, _)]) if numeric(ty) => { + Ok((ty.clone(), nullable)) + } + ("sqrt" | "exp" | "ln" | "log2" | "log10" | "sin" | "cos" | "tan", [(ty, _)]) + if numeric(ty) => + { + Ok((DataType::Float64, nullable)) + } + ("lower" | "upper" | "trim" | "btrim" | "ltrim" | "rtrim", [(DataType::Utf8, _)]) => { + Ok((DataType::Utf8, nullable)) + } + ("length" | "char_length" | "character_length" | "octet_length", [(DataType::Utf8, _)]) => { + Ok((DataType::Int64, nullable)) + } + ("label_replace", [(DataType::Utf8, _), (DataType::Utf8, _), (DataType::Utf8, _)]) => { + Ok((DataType::Utf8, false)) + } + ("label_join" | "concat", [_, ..]) if types.iter().all(|(t, _)| *t == DataType::Utf8) => { + Ok((DataType::Utf8, nullable)) + } + ("date_trunc", [(DataType::Utf8, _), (DataType::Timestamp, _)]) => { + Ok((DataType::Timestamp, nullable)) + } + ("regexp_like", [(DataType::Utf8, _), (DataType::Utf8, _)]) => { + Ok((DataType::Bool, nullable)) + } + ("nullif", [(a, _), (b, _)]) => { + common_scalar_type(a, b)?; + Ok((a.clone(), true)) + } + ("coalesce", [_, ..]) => { + let mut ty = DataType::Null; + for (arg, _) in &types { + ty = common_scalar_type(&ty, arg)?; + } + Ok((ty, types.iter().all(|(_, n)| *n))) + } + _ => Err(signature(&format!( + "unregistered scalar function or invalid signature: {name}" + ))), + } +} + +/// Resolve the bounded canonical `asap_struct_field(struct, selector)` operation. +/// Selectors are positive 1-based literal ordinals or exact literal field names. +pub fn struct_field_type(args: &[ScalarExpr], schema: &Schema) -> Result<(DataType, bool), String> { + let [input, selector] = args else { + return Err("struct field access requires a struct and constant selector".into()); + }; + let (dtype, nullable) = input + .scalar_type(schema) + .map_err(|error| error.to_string())?; + if nullable { + return Err("nullable struct container access is unsupported".into()); + } + let DataType::Struct { fields } = dtype else { + return Err("struct field access requires a Struct input".into()); + }; + let field = match selector { + ScalarExpr::Literal(ScalarValue::Int64(index)) if *index > 0 => usize::try_from(*index - 1) + .ok() + .and_then(|index| fields.get(index)) + .ok_or("struct field ordinal is out of bounds")?, + ScalarExpr::Literal(ScalarValue::Utf8(name)) => { + let mut matches = fields.iter().filter(|field| field.name == *name); + let field = matches.next().ok_or("struct field name does not exist")?; + if matches.next().is_some() { + return Err("struct field name is ambiguous".into()); + } + field + } + _ => { + return Err( + "struct field selector must be a positive ordinal or field-name literal".into(), + ) + } + }; + Ok((field.dtype.clone(), field.nullable)) +} + +/// Resolve `asap_element_access(collection, index)`: Map access through the +/// map function contract, List access with integer indices. +pub fn element_access_type( + args: &[ScalarExpr], + schema: &Schema, +) -> Result<(DataType, bool), String> { + let [input, index] = args else { + return Err("element access requires a collection and index".into()); + }; + let source = input.scalar_type(schema).map_err(|e| e.to_string())?; + let key = index.scalar_type(schema).map_err(|e| e.to_string())?; + match &source.0 { + DataType::Map { .. } => MapScalarFunction::Access.output_type(&[source, key]), + DataType::List { element } => { + if source.1 { + return Err("nullable List container access is unsupported".into()); + } + if !matches!(key.0, DataType::Int64 | DataType::Null) { + return Err("List index must have integer type".into()); + } + if matches!(index, ScalarExpr::Literal(ScalarValue::Int64(0))) { + return Err( + "literal zero List index is unsupported without constant-array proof".into(), + ); + } + Ok(( + element.dtype.clone(), + element.nullable || key.1 || key.0 == DataType::Null, + )) + } + _ => Err("element access requires a Map or List".into()), + } +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::ir::operator_properties::Source; + use crate::ir::{NonASAPOp, OperatorNode}; + use crate::pre_asap::schema::{Field, FieldDataType}; + + fn call(name: &str, args: Vec) -> ScalarExpr { + ScalarExpr::FunctionCall { + name: name.into(), + args, + } + } + + fn int(v: i64) -> ScalarExpr { + ScalarExpr::Literal(ScalarValue::Int64(v)) + } + + fn utf8(v: &str) -> ScalarExpr { + ScalarExpr::Literal(ScalarValue::Utf8(v.into())) + } + + /// Shifting an instant by a duration stays an instant and shifting a date + /// stays a date (`l_shipdate + INTERVAL '30' DAY`), never numeric. + #[test] + fn interval_arithmetic_keeps_the_temporal_type() { + let schema = Schema::new(vec![ + Field::plain("ts", DataType::Timestamp, false), + Field::plain("d", DataType::Date, false), + ]); + let thirty_days = || { + Box::new(ScalarExpr::Literal(ScalarValue::Interval { + months: 0, + days: 30, + nanos: 0, + })) + }; + let shift = |left: Box, op| ScalarExpr::Arithmetic { + op, + left, + right: thirty_days(), + semantics: ExprSemantics::Sql, + }; + let ty = |e: ScalarExpr| e.scalar_type(&schema).unwrap().0; + assert_eq!( + ty(shift( + Box::new(ScalarExpr::Column(0)), + ArithmeticOpKind::Add + )), + DataType::Timestamp + ); + assert_eq!( + ty(shift( + Box::new(ScalarExpr::Column(1)), + ArithmeticOpKind::Sub + )), + DataType::Date + ); + assert_eq!( + ty(shift(thirty_days(), ArithmeticOpKind::Add)), + DataType::Interval + ); + } + + #[test] + fn canonical_projection_uses_map_signature_and_rejects_invalid_arity() { + let project = |expr: ScalarExpr| NonASAPOp::Project { + cols: vec![ProjectItem { + alias: Some("result".into()), + expr, + }], + qualifier: None, + child: OperatorNode::new_shared(crate::ir::Operator::NonASAP(NonASAPOp::Scan { + source: Source::Table { + table_ref: "t".into(), + }, + predicates: vec![], + schema: Schema::new(vec![ + Field::plain("k", DataType::Utf8, false), + Field::plain("v", DataType::Int64, true), + ]), + })) + .unwrap(), + }; + let map = call("map", vec![ScalarExpr::Column(0), ScalarExpr::Column(1)]); + let schema = project(map.clone()).output_schema().unwrap(); + assert_eq!( + schema.fields[0].dtype, + DataType::Map { + key: Box::new(DataType::Utf8), + value: Box::new(DataType::Int64), + value_nullable: true + } + ); + assert!(!schema.fields[0].nullable); + let lookup = call("asap_map_access", vec![map, utf8("missing")]); + assert_eq!( + project(lookup).output_schema().unwrap().fields[0], + Field::plain("result", DataType::Int64, true) + ); + assert!(project(call("map", vec![ScalarExpr::Column(0)])) + .output_schema() + .is_err()); + } + + fn record_schema() -> Schema { + Schema::new(vec![Field::plain( + "record", + DataType::Struct { + fields: vec![ + Field::new("ts", DataType::Int64, false), + Field::new( + "values", + DataType::List { + element: Box::new(Field::new("item", DataType::Float64, true)), + }, + true, + ), + ], + }, + false, + )]) + } + + fn field_access(selector: ScalarExpr) -> ScalarExpr { + call("asap_struct_field", vec![ScalarExpr::Column(0), selector]) + } + + #[test] + fn field_access_reuses_nested_field_type_and_nullability() { + let schema = record_schema(); + assert_eq!( + field_access(int(1)).scalar_type(&schema).unwrap(), + (DataType::Int64, false) + ); + let named = field_access(utf8("values")); + let ordinal = field_access(int(2)); + assert_eq!( + named.scalar_type(&schema).unwrap(), + ordinal.scalar_type(&schema).unwrap() + ); + assert_eq!( + named.scalar_type(&schema).unwrap(), + ( + DataType::List { + element: Box::new(Field::new("item", DataType::Float64, true)) + }, + true + ) + ); + let roundtrip: ScalarExpr = + serde_json::from_str(&serde_json::to_string(&named).unwrap()).unwrap(); + assert_eq!(roundtrip, named); + } + + #[test] + fn unsupported_field_access_is_an_error_not_placeholder_typing() { + for selector in [ + ScalarExpr::Column(0), + int(0), + int(-1), + int(3), + utf8("missing"), + ] { + assert!(field_access(selector) + .scalar_type(&record_schema()) + .is_err()); + } + let mut ambiguous = record_schema(); + if let FieldDataType::Plain(DataType::Struct { fields }) = &mut ambiguous.fields[0].dtype { + fields.push(Field::new("ts", DataType::Utf8, false)); + } + assert!(field_access(utf8("ts")).scalar_type(&ambiguous).is_err()); + let mut nullable = record_schema(); + nullable.fields[0].nullable = true; + assert!(field_access(int(1)).scalar_type(&nullable).is_err()); + } + + fn element_access(index: ScalarExpr) -> ScalarExpr { + call("asap_element_access", vec![ScalarExpr::Column(0), index]) + } + + #[test] + fn list_index_preserves_nested_element_metadata() { + let element = DataType::Struct { + fields: vec![ + Field::new("ts", DataType::Int64, false), + Field::new("value", DataType::Float64, true), + ], + }; + let schema = Schema::new(vec![ + Field::plain( + "samples", + DataType::List { + element: Box::new(Field::new("item", element.clone(), false)), + }, + false, + ), + Field::plain("i", DataType::Int64, true), + ]); + for index in [1, -1, 100] { + assert_eq!( + element_access(int(index)).scalar_type(&schema).unwrap(), + (element.clone(), false) + ); + } + assert_eq!( + element_access(ScalarExpr::Column(1)) + .scalar_type(&schema) + .unwrap(), + (element.clone(), true) + ); + assert!(element_access(int(0)).scalar_type(&schema).is_err()); + assert!(element_access(ScalarExpr::literal_f64(1.0)) + .scalar_type(&schema) + .is_err()); + let nested = call("asap_struct_field", vec![element_access(int(1)), int(2)]); + assert_eq!( + nested.scalar_type(&schema).unwrap(), + (DataType::Float64, true) + ); + let roundtrip: ScalarExpr = + serde_json::from_value(serde_json::to_value(&nested).unwrap()).unwrap(); + assert_eq!(roundtrip, nested); + } + + #[test] + fn generic_map_lookup_reuses_legacy_signature() { + let schema = Schema::new(vec![Field::plain( + "m", + DataType::Map { + key: Box::new(DataType::Utf8), + value: Box::new(DataType::Int64), + value_nullable: false, + }, + false, + )]); + let legacy = call("asap_map_access", vec![ScalarExpr::Column(0), utf8("k")]); + assert_eq!( + element_access(utf8("k")).scalar_type(&schema).unwrap(), + legacy.scalar_type(&schema).unwrap() + ); + } +} diff --git a/crates/types/src/lib.rs b/crates/types/src/lib.rs index caacfd441..9fc450c7f 100644 --- a/crates/types/src/lib.rs +++ b/crates/types/src/lib.rs @@ -28,3 +28,5 @@ pub mod resources; pub mod serde_f64; pub mod types; pub mod workload; + +pub mod ir; diff --git a/crates/types/src/post_asap/maintained_population.rs b/crates/types/src/post_asap/maintained_population.rs index 1357fa63d..73ddd38ba 100644 --- a/crates/types/src/post_asap/maintained_population.rs +++ b/crates/types/src/post_asap/maintained_population.rs @@ -114,18 +114,18 @@ impl CurrentSeriesInput { /// Membership is part of state identity. Table rows must never acquire implicit /// latest-per-series selection, stale markers, or a PromQL lookback. #[derive(Debug, Clone, PartialEq, Serialize, Deserialize)] -pub enum PopulationInput { +pub enum PopulationInput { CurrentSeries(CurrentSeriesInput), Rows { - input: std::rc::Rc, + input: std::rc::Rc, value_column: usize, grouping: crate::pre_asap::GroupKeys, }, } #[derive(Debug, Clone, PartialEq, Serialize, Deserialize)] -pub struct MaintainedPopulation { - pub input: PopulationInput, +pub struct MaintainedPopulation { + pub input: PopulationInput, pub max_k: usize, pub quantiles: bool, } @@ -158,3 +158,130 @@ impl MaintainedPopulation { } } } + +impl CurrentSeriesInput { + /// Verify the named contract against the canonical maintenance input. + pub fn matches_node(&self, input: &crate::ir::OperatorNode) -> bool { + use crate::ir::{NonASAPOp, Operator, ScalarExpr, TimeRangeKind}; + use crate::pre_asap::{CompareOpKind, DataType, ScalarValue, Source}; + let input = match &input.operator { + Operator::NonASAP(NonASAPOp::TimeRange { + range, + kind: TimeRangeKind::Instant, + child, + }) if self.lookback_ms > 0 + && *range == std::time::Duration::from_millis(self.lookback_ms) => + { + child.as_ref() + } + Operator::NonASAP(NonASAPOp::TimeRange { .. }) => return false, + _ if self.lookback_ms == 300_000 => input, + _ => return false, + }; + let Operator::NonASAP(NonASAPOp::Scan { + source: Source::TimeSeries { metric }, + predicates, + schema, + }) = &input.operator + else { + return false; + }; + if self.metric.is_empty() + || *metric != self.metric + || (schema.closed && !schema.has_promql_series_identity()) + || schema.time_index.is_none() + { + return false; + } + if self.grouping.iter().any(|label| { + !schema + .fields + .iter() + .any(|c| c.name == *label && c.dtype == DataType::Utf8) + }) { + return false; + } + let mut matchers = Vec::new(); + for predicate in predicates { + let ScalarExpr::Compare { + left, op, right, .. + } = &predicate.0 + else { + return false; + }; + let (ScalarExpr::Column(col), ScalarExpr::Literal(ScalarValue::Utf8(value))) = + (left.as_ref(), right.as_ref()) + else { + return false; + }; + let Some(column) = schema.fields.get(*col) else { + return false; + }; + if column.dtype != DataType::Utf8 { + return false; + } + let operation = match op { + CompareOpKind::Eq => CurrentSeriesMatch::Equal, + CompareOpKind::Ne => CurrentSeriesMatch::NotEqual, + CompareOpKind::Regex => CurrentSeriesMatch::Regex, + CompareOpKind::NotRegex => CurrentSeriesMatch::NotRegex, + _ => return false, + }; + matchers.push(CurrentSeriesMatcher { + label: column.name.clone(), + value: value.clone(), + operation, + }); + } + matchers.sort(); + matchers.dedup(); + self.matchers == matchers && self.grouping.windows(2).all(|w| w[0] < w[1]) + } +} + +impl MaintainedPopulation { + /// Whether `input` is the maintenance input this population declares. + pub fn matches_node(&self, input: &crate::ir::OperatorNode) -> bool { + use crate::ir::{NonASAPOp, Operator}; + use crate::pre_asap::{DataType, Source}; + match &self.input { + PopulationInput::CurrentSeries(spec) => spec.matches_node(input), + PopulationInput::Rows { + input: expected, + value_column, + grouping, + } => { + let Operator::NonASAP(NonASAPOp::Scan { + source: Source::Table { .. }, + schema, + .. + }) = &input.operator + else { + return false; + }; + // The same computation, whatever accuracy or timing has + // been attached to the node since. + let same_source = + expected.operator == input.operator && expected.schema == input.schema; + same_source + && schema.closed + && schema + .fields + .get(*value_column) + .is_some_and(|c| c.dtype == DataType::Float64 && !c.nullable) + && !grouping.is_without() + && grouping.keys().iter().all(|k| *k < schema.fields.len()) + } + } + } + + pub fn supports(&self, evaluation: &PopulationStatistic) -> bool { + match evaluation { + PopulationStatistic::Quantile { q } => self.quantiles && q.is_finite(), + PopulationStatistic::TopK { k } => *k <= self.max_k, + PopulationStatistic::Sum + | PopulationStatistic::Count + | PopulationStatistic::Average => true, + } + } +} diff --git a/crates/types/src/pre_asap/mod.rs b/crates/types/src/pre_asap/mod.rs index 4cfcfbd96..f434eb154 100644 --- a/crates/types/src/pre_asap/mod.rs +++ b/crates/types/src/pre_asap/mod.rs @@ -37,7 +37,7 @@ pub mod cse; pub mod expr_ir; pub mod query_expr; pub mod resolve; -pub mod scalar_signature; +pub mod scalar_type_rules; pub mod schema; pub mod schema_resolver; diff --git a/crates/types/src/pre_asap/query_expr.rs b/crates/types/src/pre_asap/query_expr.rs index 1051157d5..e389f84a2 100644 --- a/crates/types/src/pre_asap/query_expr.rs +++ b/crates/types/src/pre_asap/query_expr.rs @@ -1811,13 +1811,13 @@ fn infer_expr_type( } QueryExpr::FunctionCall { name, args } => { if name == "asap_element_access" { - super::scalar_signature::element_access_type(args, schema) + super::scalar_type_rules::element_access_type(args, schema) .map_err(QueryExprError::InvalidScalarSignature)? } else if name == "asap_struct_field" { - super::scalar_signature::struct_field_type(args, schema) + super::scalar_type_rules::struct_field_type(args, schema) .map_err(QueryExprError::InvalidScalarSignature)? } else if let Some(function) = - super::scalar_signature::MapScalarFunction::from_name(name) + super::scalar_type_rules::MapScalarFunction::from_name(name) { let arguments = args .iter() diff --git a/crates/types/src/pre_asap/scalar_signature.rs b/crates/types/src/pre_asap/scalar_type_rules.rs similarity index 95% rename from crates/types/src/pre_asap/scalar_signature.rs rename to crates/types/src/pre_asap/scalar_type_rules.rs index 6e316169c..44eaa250c 100644 --- a/crates/types/src/pre_asap/scalar_signature.rs +++ b/crates/types/src/pre_asap/scalar_type_rules.rs @@ -502,3 +502,18 @@ mod element_access_tests { ); } } + +/// Closed, namespaced contracts for PromQL pointwise float functions. +/// Date functions consume Unix seconds; `timestamp` remains a sample-selection +/// operation because its operand is a sample timestamp rather than its value. +pub fn promql_function_arity(name: &str) -> Option { + Some(match name.strip_prefix("promql_")? { + "abs" | "ceil" | "floor" | "exp" | "ln" | "log2" | "log10" | "sqrt" | "sgn" | "sin" + | "cos" | "tan" | "asin" | "acos" | "atan" | "sinh" | "cosh" | "tanh" | "asinh" + | "acosh" | "atanh" | "deg" | "rad" | "minute" | "hour" | "day_of_week" + | "day_of_month" | "day_of_year" | "month" | "year" | "days_in_month" => 1, + "round" | "clamp_min" | "clamp_max" => 2, + "clamp" => 3, + _ => return None, + }) +} diff --git a/crates/types/tests/schema_rebuilding.rs b/crates/types/tests/schema_rebuilding.rs new file mode 100644 index 000000000..1f1b57dbc --- /dev/null +++ b/crates/types/tests/schema_rebuilding.rs @@ -0,0 +1,203 @@ +use asap_types::ir::{ASAPOp, NonASAPOp, Operator, OperatorNode}; +use asap_types::post_asap::{ + ExactKind, ExactParams, ExecutionTiming, GroupingStrategy, ResultGuarantee, SummaryUpdate, +}; +use asap_types::pre_asap::{ + AggIntent, ColumnRef, DataType, Field, FieldDataType, Reduction, Schema, Source, +}; +use std::rc::Rc; + +fn scan(key_type: DataType, name: &str) -> Rc { + OperatorNode::new_shared(Operator::NonASAP(NonASAPOp::Scan { + source: Source::Table { + table_ref: "t".into(), + }, + predicates: vec![], + schema: Schema::new(vec![ + Field::plain(name, key_type, false), + Field::plain("value", DataType::Float64, false), + ]), + })) + .unwrap() +} +fn aggregate(child: Rc, asap: bool) -> Rc { + let operator = if asap { + Operator::ASAP(ASAPOp::SummaryAgg { + child, + family: FieldDataType::ExactAggregate(ExactKind::Sum, ExactParams::Sum), + input: SummaryUpdate::column(ColumnRef::Named("value".into())), + reduction: Reduction::by(vec![0]), + grouping: GroupingStrategy::default(), + filter: None, + }) + } else { + Operator::NonASAP(NonASAPOp::Aggregate { + child, + reduction: Reduction::by(vec![0]), + measures: vec![AggIntent::Sum { col: Some(1) }], + output_names: vec![], + filters: vec![], + having: None, + }) + }; + OperatorNode::new_shared(operator).unwrap() +} + +/// Rewrites follow changed input types and inherited names for either category. +#[test] +fn rebuilding_rederives_schema_for_both_categories() { + for asap in [false, true] { + let original = aggregate(scan(DataType::Int64, "key"), asap); + original.validate_structure().unwrap(); + let replacement = scan(DataType::Utf8, "new_key"); + let rebuilt = Rc::new(original.map_children(|_| replacement.clone()).unwrap()); + assert_eq!(rebuilt.schema, rebuilt.operator.output_schema().unwrap()); + rebuilt.validate_structure().unwrap(); + } +} + +/// Naming overrides survive rewrites without freezing types or assessed properties. +#[test] +fn rebuilding_preserves_only_explicit_naming_overrides() { + for asap in [false, true] { + let original = aggregate(scan(DataType::Int64, "key"), asap); + let mut schema = original.schema.clone(); + schema.fields[0].name = "alias".into(); + schema.fields[0].table = Some("result".into()); + let original = Rc::new( + OperatorNode::with_schema(original.operator.clone(), schema) + .with_guarantee(Some(ResultGuarantee::exact("fixture"))) + .with_timing(Some(ExecutionTiming::QueryTime)), + ); + original.validate_structure().unwrap(); + let replacement = scan(DataType::Utf8, "new_key"); + let rebuilt = Rc::new(original.map_children(|_| replacement.clone()).unwrap()); + assert_eq!(rebuilt.schema.fields[0].name, "alias"); + assert_eq!(rebuilt.schema.fields[0].table.as_deref(), Some("result")); + assert_eq!( + rebuilt.schema.fields[0].plain_dtype(), + Some(&DataType::Utf8) + ); + assert!(rebuilt.guarantee.is_none()); + assert!(rebuilt.timing.is_none()); + rebuilt.validate_structure().unwrap(); + } +} + +/// Custom names never authorize changes to structural schema metadata. +#[test] +fn validation_rejects_structural_overrides_for_both_categories() { + for asap in [false, true] { + let original = aggregate(scan(DataType::Timestamp, "key"), asap); + original.validate_structure().unwrap(); + let mut invalid = vec![]; + let mut schema = original.schema.clone(); + schema.unique_keys = vec![vec![1]]; + invalid.push(schema); + let mut schema = original.schema.clone(); + schema.time_index = Some(0); // In-range Timestamp, but not the derived time axis. + invalid.push(schema); + let mut schema = original.schema.clone(); + schema.closed = !schema.closed; + invalid.push(schema); + let mut schema = original.schema.clone(); + schema.fields[0].nullable = !schema.fields[0].nullable; + invalid.push(schema); + let mut schema = original.schema.clone(); + schema.fields[0].dtype = FieldDataType::Plain(DataType::Utf8); + invalid.push(schema); + let mut schema = original.schema.clone(); + schema.fields.pop(); + invalid.push(schema); + for schema in invalid { + let forged = Rc::new(OperatorNode::with_schema(original.operator.clone(), schema)); + assert!( + forged.validate_structure().is_err(), + "accepted structural override: {:?}", + forged.schema + ); + } + } +} + +/// Passthrough rewrites derive metadata and arity, but cannot guess alias positions. +#[test] +fn rebuilding_updates_metadata_and_requires_new_aliases_after_arity_changes() { + use asap_types::pre_asap::GroupKeys; + let input = scan(DataType::Timestamp, "key"); + let original = OperatorNode::new_shared(Operator::NonASAP(NonASAPOp::Limit { + n: Some(10), + offset: 0, + partition_by: GroupKeys::by(vec![]), + child: input, + })) + .unwrap(); + let mut replacement_schema = original.schema.clone(); + replacement_schema.time_index = Some(0); + replacement_schema.unique_keys = vec![vec![0]]; + replacement_schema.closed = false; + replacement_schema + .fields + .push(Field::plain("extra", DataType::Int64, true)); + let replacement = OperatorNode::new_shared(Operator::NonASAP(NonASAPOp::Scan { + source: Source::Table { + table_ref: "t".into(), + }, + predicates: vec![], + schema: replacement_schema.clone(), + })) + .unwrap(); + let rebuilt = Rc::new(original.map_children(|_| replacement.clone()).unwrap()); + assert_eq!(rebuilt.schema, replacement_schema); + rebuilt.validate_structure().unwrap(); + + let mut names = original.schema.clone(); + names.fields[0].name = "alias".into(); + let named = OperatorNode::with_schema(original.operator.clone(), names); + assert!(named.map_children(|_| replacement.clone()).is_err()); +} + +/// Maintaining membership and finalizing values preserve identity/time metadata. +#[test] +fn summary_transitions_preserve_structural_metadata() { + use asap_types::post_asap::maintained_population::{MaintainedPopulation, PopulationInput}; + use asap_types::pre_asap::GroupKeys; + let mut schema = scan(DataType::Timestamp, "key").schema.clone(); + schema.closed = true; + schema.time_index = Some(0); + schema.unique_keys = vec![vec![0]]; + let source = OperatorNode::new_shared(Operator::NonASAP(NonASAPOp::Scan { + source: Source::Table { + table_ref: "t".into(), + }, + predicates: vec![], + schema: schema.clone(), + })) + .unwrap(); + let maintained = OperatorNode::new_shared(Operator::ASAP(ASAPOp::MaintainPopulation { + child: source.clone(), + population: MaintainedPopulation { + input: PopulationInput::Rows { + input: source.clone(), + value_column: 1, + grouping: GroupKeys::by(vec![0]), + }, + max_k: 1, + quantiles: false, + }, + })) + .unwrap(); + assert_eq!(maintained.schema, schema); + maintained.validate_structure().unwrap(); + + let state = aggregate(source, true); + let finalized = OperatorNode::new_shared(Operator::ASAP(ASAPOp::FinalizeExactAccumulator { + child: state.clone(), + })) + .unwrap(); + assert_eq!(finalized.schema.unique_keys, state.schema.unique_keys); + assert_eq!(finalized.schema.closed, state.schema.closed); + assert_eq!(finalized.schema.time_index, state.schema.time_index); + assert!(finalized.schema.is_all_plain()); + finalized.validate_structure().unwrap(); +} diff --git a/crates/types/tests/structure_contract.rs b/crates/types/tests/structure_contract.rs new file mode 100644 index 000000000..d4e164f71 --- /dev/null +++ b/crates/types/tests/structure_contract.rs @@ -0,0 +1,207 @@ +use asap_types::ir::{ + ExprSemantics, NonASAPOp, OperatorNode, OperatorResultKind, Predicate, ScalarExpr, +}; +use asap_types::pre_asap::{DataType, Field, ScalarValue, Schema, Source}; +use std::rc::Rc; +fn scan() -> Rc { + OperatorNode::new_shared(asap_types::ir::Operator::NonASAP(NonASAPOp::Scan { + source: Source::Table { + table_ref: "t".into(), + }, + predicates: vec![], + schema: Schema::new(vec![Field::plain("x", DataType::Float64, false)]), + })) + .unwrap() +} +/// Resolved filters cannot hide invalid scalar types or out-of-scope columns. +#[test] +fn invalid_predicates_are_rejected() { + for expr in [ + ScalarExpr::Column(7), + ScalarExpr::literal_f64(1.0), + ScalarExpr::Not(Box::new(ScalarExpr::literal_f64(1.0))), + ] { + let op = NonASAPOp::Filter { + child: scan(), + pred: Predicate(expr), + }; + assert!(op.validate_inputs().is_err()); + } +} +/// Common metadata must agree with the actual operation, including result kind. +#[test] +fn retained_result_kind_is_checked() { + let mut node = (*scan()).clone(); + node.result_kind = OperatorResultKind::RangeVector; + assert!(Rc::new(node).validate_structure().is_err()); +} +/// Values validates arity and declared nullability without inventing columns. +#[test] +fn values_contract_is_checked() { + for row in [ + vec![], + vec![ScalarExpr::Literal(ScalarValue::Null)], + vec![ScalarExpr::Literal(ScalarValue::Utf8("x".into()))], + ] { + let op = NonASAPOp::Values { + rows: vec![row], + schema: scan().schema.clone(), + }; + assert!(op.validate_inputs().is_err()); + } +} +/// Scalar typing validates every branch and never assigns placeholder types. +#[test] +fn scalar_type_ruless_fail_closed() { + for expr in [ + ScalarExpr::Column(99), + ScalarExpr::FunctionCall { + name: "not_registered".into(), + args: vec![], + }, + ScalarExpr::Negative { + expr: Box::new(ScalarExpr::Literal(ScalarValue::Utf8("x".into()))), + semantics: ExprSemantics::Sql, + }, + ScalarExpr::Case { + operand: None, + branches: vec![(ScalarExpr::literal_f64(1.0), ScalarExpr::literal_f64(2.0))], + else_expr: None, + }, + ScalarExpr::ScalarSubquery( + OperatorNode::new_shared(asap_types::ir::Operator::NonASAP(NonASAPOp::Values { + rows: vec![], + schema: Schema::default(), + })) + .unwrap(), + ), + ScalarExpr::PromqlScalarFromVector(scan()), + ] { + assert!(expr.scalar_type(&scan().schema).is_err(), "{expr:?}"); + } +} + +/// A state family is not interchangeable with another sketch or a scalar field. +#[test] +fn state_evaluations_and_passthrough_keep_their_contracts() { + use asap_types::ir::{ASAPOp, Operator, ProjectItem}; + use asap_types::post_asap::{ + FieldDataType, GroupingStrategy, SketchAlgorithm, SketchKind, SketchParams, + SketchStatistic, SummaryUpdate, + }; + use asap_types::pre_asap::{ColumnRef, Reduction}; + let family = FieldDataType::Sketch( + SketchKind::new(SketchAlgorithm::Kll, SketchParams::Kll { k: 100 }), + GroupingStrategy::default(), + ); + let state = Rc::new( + OperatorNode::new(Operator::ASAP(ASAPOp::SummaryAgg { + child: scan(), + family, + input: SummaryUpdate::column(ColumnRef::Named("x".into())), + reduction: Reduction::by(vec![]), + grouping: GroupingStrategy::default(), + filter: None, + })) + .unwrap(), + ); + state.validate_structure().unwrap(); + let pass = OperatorNode::new_shared(asap_types::ir::Operator::NonASAP(NonASAPOp::Project { + child: state.clone(), + cols: vec![ProjectItem { + expr: ScalarExpr::Column(0), + alias: None, + }], + qualifier: None, + })) + .unwrap(); + pass.validate_structure().unwrap(); + assert_eq!(pass.result_kind, OperatorResultKind::State); + assert!(ScalarExpr::Column(0).scalar_type(&pass.schema).is_err()); + assert!(ASAPOp::SummaryEstimate { + summary_input: state.clone(), + query: SketchStatistic::Cardinality + } + .validate_inputs() + .is_err()); + assert!(ASAPOp::SummaryEstimate { + summary_input: state.clone(), + query: SketchStatistic::Quantile { q: 0.99 } + } + .validate_inputs() + .is_ok()); + assert!(ASAPOp::FinalizeExactAccumulator { child: state } + .validate_inputs() + .is_err()); +} + +/// Phase validation checks dependencies, without declaring a computation query-only. +#[test] +fn execution_timing_checks_edges_not_function_names() { + use asap_types::ir::ProjectItem; + use asap_types::post_asap::ExecutionTiming::{IngestionTime, QueryTime}; + let input = Rc::new((*scan()).clone().with_timing(Some(IngestionTime))); + let mut project = OperatorNode::new(asap_types::ir::Operator::NonASAP(NonASAPOp::Project { + child: input, + cols: vec![ProjectItem { + expr: ScalarExpr::FunctionCall { + name: "promql_abs".into(), + args: vec![ScalarExpr::Column(0)], + }, + alias: None, + }], + qualifier: None, + })) + .unwrap() + .with_timing(Some(IngestionTime)); + Rc::new(project.clone()) + .validate_execution_timing() + .unwrap(); + if let asap_types::ir::Operator::NonASAP(NonASAPOp::Project { child, .. }) = + &mut project.operator + { + *child = Rc::new(child.as_ref().clone().with_timing(Some(QueryTime))); + } + assert!(Rc::new(project).validate_execution_timing().is_err()); +} + +/// Both operator categories use the same fallible schema-deriving constructor. +#[test] +fn shared_construction_derives_both_operator_categories() { + use asap_types::ir::{ASAPOp, Operator, ProjectItem}; + use asap_types::post_asap::{ExactKind, ExactParams, GroupingStrategy, SummaryUpdate}; + use asap_types::pre_asap::{ColumnRef, FieldDataType, Reduction}; + + let input = scan(); + let state = OperatorNode::new_shared(Operator::ASAP(ASAPOp::SummaryAgg { + child: input.clone(), + family: FieldDataType::ExactAggregate(ExactKind::Sum, ExactParams::Sum), + input: SummaryUpdate::column(ColumnRef::Named("x".into())), + reduction: Reduction::by(vec![]), + grouping: GroupingStrategy::default(), + filter: None, + })) + .unwrap(); + assert_eq!(state.result_kind, OperatorResultKind::State); + assert!(!state.schema.fields.last().unwrap().is_plain()); + assert!(state.guarantee.is_none()); + assert!(state.timing.is_none()); + state.validate_structure().unwrap(); + + assert!( + OperatorNode::new_shared(Operator::NonASAP(NonASAPOp::Project { + child: input.clone(), + cols: vec![ProjectItem { + expr: ScalarExpr::Column(99), + alias: None + }], + qualifier: None, + })) + .is_err() + ); + assert!(OperatorNode::new_shared(Operator::ASAP(ASAPOp::Extension { + child: input, + name: "unregistered".into(), + })) + .is_err()); +}