diff --git a/sidemantic-rs/examples/parity_adapter.rs b/sidemantic-rs/examples/parity_adapter.rs index 5b3fd81ea..74542f930 100644 --- a/sidemantic-rs/examples/parity_adapter.rs +++ b/sidemantic-rs/examples/parity_adapter.rs @@ -45,6 +45,28 @@ struct CompileRequest { parameter_values: std::collections::HashMap, } +// Native Ossie import already produces typed IR. Compile that graph directly; +// native YAML configuration has a different metric ownership/input contract. +#[derive(Debug, Deserialize)] +struct OssieCompileRequest { + content: String, + serialization: String, + #[serde(default = "default_ossie_consumer_profile")] + consumer_profile: String, + #[serde(default = "default_ossie_target")] + target: String, + scope_id: Option, + #[serde(default)] + metrics: Vec, + #[serde(default)] + dimensions: Vec, + #[serde(default)] + filters: Vec, + #[serde(default)] + order_by: Vec, + dialect: String, +} + #[derive(Debug, Deserialize)] #[serde(tag = "action", rename_all = "snake_case")] enum Request { @@ -52,6 +74,7 @@ enum Request { models_yaml: String, }, Compile(Box), + OssieCompile(Box), JoinPath { models_yaml: String, from_model: String, @@ -221,6 +244,31 @@ fn handle(request: Request) -> sidemantic::Result { value: None, }) } + Request::OssieCompile(request) => { + let graph = OssieForwardAdapter + .select_scope( + &request.content, + parse_ossie_serialization(&request.serialization)?, + parse_ossie_consumer(&request.consumer_profile)?, + parse_ossie_target(&request.target)?, + request.scope_id.as_deref(), + )? + .into_graph()?; + let query = SemanticQuery::new() + .with_metrics(request.metrics) + .with_dimensions(request.dimensions) + .with_filters(request.filters) + .with_order_by(request.order_by); + let sql = SqlGenerator::new(&graph) + .with_dialect(parse_dialect(&request.dialect)?) + .generate(&query)?; + Ok(Response::Ok { + sql: Some(sql), + path: None, + catalog: None, + value: None, + }) + } Request::JoinPath { models_yaml, from_model, diff --git a/sidemantic-rs/src/adapters/mod.rs b/sidemantic-rs/src/adapters/mod.rs index 05f2b0792..43f274618 100644 --- a/sidemantic-rs/src/adapters/mod.rs +++ b/sidemantic-rs/src/adapters/mod.rs @@ -14,6 +14,7 @@ use crate::error::{Result, SidemanticError}; pub mod cube; pub mod osi; pub mod ossie; +pub mod ossie_sql; pub use cube::CubeAdapter; pub use osi::OsiAdapter; diff --git a/sidemantic-rs/src/adapters/ossie.rs b/sidemantic-rs/src/adapters/ossie.rs index 015fa2640..6e26913fa 100644 --- a/sidemantic-rs/src/adapters/ossie.rs +++ b/sidemantic-rs/src/adapters/ossie.rs @@ -4,19 +4,39 @@ //! behavior is retained for compatibility. The forward adapter is strict, //! scope-preserving, target-aware, and fail-closed. +use std::cell::Cell; use std::collections::{BTreeMap, BTreeSet, HashMap}; -use polyglot_sql::{DialectType, Expression, ExpressionWalk}; -use serde::Serialize; +use polyglot_sql::{DialectType, Expression}; +use serde::de::{self, MapAccess, SeqAccess, Visitor}; +use serde::{Deserialize, Deserializer, Serialize}; use serde_json::{Map, Value}; -use crate::config::schema::metric_from_sql_expression; use crate::core::{ - Dimension, DimensionType, Metric, Model, Relationship, RelationshipType, Segment, + Dimension, DimensionType, Metric, MetricType, Model, Relationship, RelationshipType, Segment, + SemanticGraph, }; use crate::error::{Result, SidemanticError}; const VALIDATION_MODE: &str = "closed_structural_subset"; +const CURRENT_SCHEMA_REVISION: &str = "b6c702ed1c07e91382a69e870c875cbd19570828"; +const CURRENT_SCHEMA_VERSION: &str = "0.2.0.dev0-current"; +const MAX_SOURCE_BYTES: usize = 16 * 1024 * 1024; +const MAX_PARSE_DEPTH: usize = 256; +const MAX_PARSE_NODES: usize = 100_000; +thread_local! { + static PARSE_BUDGET: Cell<(usize, usize)> = const { Cell::new((0, 0)) }; +} + +struct ParseDepth; +impl Drop for ParseDepth { + fn drop(&mut self) { + PARSE_BUDGET.with(|budget| { + let (nodes, depth) = budget.get(); + budget.set((nodes, depth.saturating_sub(1))); + }); + } +} const TEMPORAL_TYPES: &[&str] = &["Date", "Time", "DateTime", "DateTimeTz"]; const NUMERIC_TYPES: &[&str] = &["Integer", "Decimal", "Float"]; const DATA_TYPES: &[&str] = &[ @@ -50,6 +70,137 @@ const DIALECTS_0_2_0: &[&str] = &[ "SIGMA", "THOUGHTSPOT", ]; +const DIALECTS_CURRENT: &[&str] = &[ + "ANSI_SQL", + "SNOWFLAKE", + "MDX", + "TABLEAU", + "DATABRICKS", + "MAQL", + "BIGQUERY", + "SIGMA", + "THOUGHTSPOT", + "DAX", + "OSSIE_SQL_2026", +]; + +/// Deserialize objects before converting to `Value`, which otherwise silently +/// overwrites duplicate JSON keys. The same contract applies to YAML mappings. +struct UniqueValue(Value); + +impl<'de> Deserialize<'de> for UniqueValue { + fn deserialize>(deserializer: D) -> std::result::Result { + let permitted = PARSE_BUDGET.with(|budget| { + let (nodes, depth) = budget.get(); + budget.set((nodes + 1, depth + 1)); + nodes < MAX_PARSE_NODES && depth < MAX_PARSE_DEPTH + }); + let _depth = ParseDepth; + if !permitted { + return Err(de::Error::custom("Ossie parser resource limit exceeded")); + } + struct UniqueVisitor; + impl<'de> Visitor<'de> for UniqueVisitor { + type Value = UniqueValue; + fn expecting(&self, formatter: &mut std::fmt::Formatter) -> std::fmt::Result { + formatter.write_str("a JSON-compatible value without duplicate keys") + } + fn visit_bool(self, value: bool) -> std::result::Result { + Ok(UniqueValue(Value::Bool(value))) + } + fn visit_i64(self, value: i64) -> std::result::Result { + Ok(UniqueValue(value.into())) + } + fn visit_u64(self, value: u64) -> std::result::Result { + Ok(UniqueValue(value.into())) + } + fn visit_f64(self, value: f64) -> std::result::Result { + serde_json::Number::from_f64(value) + .map(|value| UniqueValue(Value::Number(value))) + .ok_or_else(|| E::custom("non-finite numbers are not JSON values")) + } + fn visit_str(self, value: &str) -> std::result::Result { + Ok(UniqueValue(value.into())) + } + fn visit_string( + self, + value: String, + ) -> std::result::Result { + Ok(UniqueValue(value.into())) + } + fn visit_unit(self) -> std::result::Result { + Ok(UniqueValue(Value::Null)) + } + fn visit_none(self) -> std::result::Result { + Ok(UniqueValue(Value::Null)) + } + fn visit_seq>( + self, + mut sequence: A, + ) -> std::result::Result { + let mut values = Vec::new(); + while let Some(value) = sequence.next_element::()? { + values.push(value.0); + } + Ok(UniqueValue(Value::Array(values))) + } + fn visit_map>( + self, + mut mapping: A, + ) -> std::result::Result { + let mut values = Map::new(); + while let Some(key) = mapping.next_key::()? { + if values.contains_key(&key) { + return Err(de::Error::custom(format!("duplicate key {key:?}"))); + } + values.insert(key, mapping.next_value::()?.0); + } + Ok(UniqueValue(Value::Object(values))) + } + } + deserializer.deserialize_any(UniqueVisitor) + } +} + +fn expand_yaml_merges(value: &mut Value) -> std::result::Result<(), String> { + match value { + Value::Array(values) => { + for value in values { + expand_yaml_merges(value)?; + } + } + Value::Object(object) => { + for value in object.values_mut() { + expand_yaml_merges(value)?; + } + if let Some(merge) = object.remove("<<") { + let mappings = match merge { + Value::Object(mapping) => vec![mapping], + Value::Array(values) => values + .into_iter() + .map(|value| match value { + Value::Object(mapping) => Ok(mapping), + _ => Err("YAML merge sequences must contain mappings".to_string()), + }) + .collect::, _>>()?, + _ => { + return Err("YAML merge value must be a mapping or sequence of mappings" + .to_string()) + } + }; + // Explicit keys override merged defaults. In merge sequences, + // earlier mappings take precedence over subsequent mappings. + for mapping in mappings { + for (key, value) in mapping { + object.entry(key).or_insert(value); + } + } + } + } + _ => {} + } + Ok(()) +} #[derive(Debug, Clone, Copy, PartialEq, Eq)] enum DocumentKind { @@ -162,6 +313,8 @@ pub struct OssieProfile { pub consumer_profile: String, pub validation_schema_version: String, pub compatibility_alias_for: Option, + #[serde(skip_serializing_if = "Option::is_none")] + pub schema_revision: Option, } #[derive(Debug, Clone, PartialEq, Eq, Serialize)] @@ -194,6 +347,101 @@ pub struct OssieCompiledScope { pub metrics: Vec, } +impl OssieCompiledScope { + /// Build one graph without inferring dataset ownership from metric SQL. + /// Ossie metrics belong to the semantic-model namespace, and may reference + /// metrics declared later in the source document. + pub fn into_graph(self) -> Result { + crate::semantic_input::with_semantic_stack(|| { + let target = + OssieTarget::parse(&self.target_dialect).map_err(SidemanticError::Validation)?; + let mut graph = SemanticGraph::new(); + for model in self.models { + graph.add_model(model)?; + } + for metric in self.metrics { + graph.add_metric_unvalidated(metric)?; + } + graph.set_metric_scopes(HashMap::new())?; + validate_scope_dependencies(&graph, target)?; + Ok(graph) + }) + } +} + +fn validate_scope_dependencies(graph: &SemanticGraph, target: OssieTarget) -> Result<()> { + fn collect(value: &Value, graph: &SemanticGraph, dependencies: &mut Vec) -> Result<()> { + if let Some(column) = value.get("column") { + let field = column["name"]["name"].as_str().unwrap_or(""); + if let Some(model) = column["table"]["name"].as_str() { + if graph.get_model(model).is_none() { + return Err(SidemanticError::Validation(format!( + "Unknown dataset '{model}'" + ))); + } + // Logical fields and undeclared physical inputs were qualified + // by the importer; neither is a metric dependency. + } else if graph.get_metric(field).is_some() { + dependencies.push(field.to_string()); + } else { + return Err(SidemanticError::Validation(format!( + "Unresolved metric or ambiguous field '{field}'" + ))); + } + return Ok(()); + } + match value { + Value::Object(fields) => { + for child in fields.values() { + collect(child, graph, dependencies)?; + } + } + Value::Array(children) => { + for child in children { + collect(child, graph, dependencies)?; + } + } + _ => {} + } + Ok(()) + } + fn visit( + name: &str, + edges: &HashMap>, + active: &mut BTreeSet, + complete: &mut BTreeSet, + ) -> Result<()> { + if complete.contains(name) { + return Ok(()); + } + if !active.insert(name.to_string()) { + return Err(SidemanticError::CircularDependency(name.to_string())); + } + for child in &edges[name] { + visit(child, edges, active, complete)?; + } + active.remove(name); + complete.insert(name.to_string()); + Ok(()) + } + let mut edges = HashMap::new(); + for metric in graph.metrics() { + let expression = parse_scalar_sql(metric.sql.as_deref().unwrap_or(""), target) + .map_err(SidemanticError::SqlParse)?; + let ast = serde_json::to_value(expression) + .map_err(|error| SidemanticError::Validation(error.to_string()))?; + let mut dependencies = Vec::new(); + collect(&ast, graph, &mut dependencies)?; + edges.insert(metric.name.clone(), dependencies); + } + let mut active = BTreeSet::new(); + let mut complete = BTreeSet::new(); + for name in edges.keys() { + visit(name, &edges, &mut active, &mut complete)?; + } + Ok(()) +} + #[derive(Debug, Clone, Serialize)] pub struct OssieCatalog { pub profile: OssieProfile, @@ -209,6 +457,7 @@ struct InspectedDocument { kind: Option, scope_ids: Vec, diagnostics: Vec, + flat_root: bool, } impl InspectedDocument { @@ -237,8 +486,25 @@ impl OssieForwardAdapter { serialization: OssieSerialization, consumer: OssieConsumerProfile, ) -> OssieStatus { - self.inspect_document(content, serialization, consumer) - .status() + crate::semantic_input::with_semantic_stack(|| { + Ok(self + .inspect_document(content, serialization, consumer) + .status()) + }) + .unwrap_or_else(|error| OssieStatus { + valid: false, + executable: false, + validation_mode: VALIDATION_MODE, + profile: None, + document_kind: None, + scopes: Vec::new(), + diagnostics: vec![diagnostic( + "ossie.parse.resource_limit", + error.to_string(), + "", + None, + )], + }) } pub fn parse_catalog( @@ -247,6 +513,18 @@ impl OssieForwardAdapter { serialization: OssieSerialization, consumer: OssieConsumerProfile, target: OssieTarget, + ) -> Result { + crate::semantic_input::with_semantic_stack(|| { + self.parse_catalog_inner(content, serialization, consumer, target) + }) + } + + fn parse_catalog_inner( + &self, + content: &str, + serialization: OssieSerialization, + consumer: OssieConsumerProfile, + target: OssieTarget, ) -> Result { let mut inspected = self.inspect_document(content, serialization, consumer); if !inspected.diagnostics.is_empty() { @@ -281,7 +559,19 @@ impl OssieForwardAdapter { let mut scopes = Vec::new(); for (index, value) in semantic_models.iter().enumerate() { let scope_id = inspected.scope_ids[index].clone(); - if let Some(scope) = compile_scope(value, index, &scope_id, target, &mut diagnostics) { + let scope_pointer = if inspected.flat_root { + String::new() + } else { + format!("/semantic_model/{index}") + }; + if let Some(scope) = compile_scope( + value, + index, + &scope_pointer, + &scope_id, + target, + &mut diagnostics, + ) { scopes.push(scope); } } @@ -356,12 +646,20 @@ impl OssieForwardAdapter { serialization: OssieSerialization, consumer: OssieConsumerProfile, ) -> InspectedDocument { - let parsed = match serialization { - OssieSerialization::Json => { - serde_json::from_str(content).map_err(|error| error.to_string()) - } - OssieSerialization::Yaml => { - serde_yaml::from_str(content).map_err(|error| error.to_string()) + PARSE_BUDGET.with(|budget| budget.set((0, 0))); + let parsed = if content.len() > MAX_SOURCE_BYTES { + Err("Ossie parser resource limit exceeded: source is larger than 16 MiB".to_string()) + } else { + match serialization { + OssieSerialization::Json => serde_json::from_str::(content) + .map(|value| value.0) + .map_err(|error| error.to_string()), + OssieSerialization::Yaml => serde_yaml::from_str::(content) + .map_err(|error| error.to_string()) + .and_then(|mut value| { + expand_yaml_merges(&mut value.0)?; + Ok(value.0) + }), } }; let root: Value = match parsed { @@ -372,8 +670,15 @@ impl OssieForwardAdapter { profile: None, kind: None, scope_ids: Vec::new(), + flat_root: false, diagnostics: vec![diagnostic( - "ossie.parse.invalid_syntax", + if error.contains("resource limit") || error.contains("recursion limit") { + "ossie.parse.resource_limit" + } else if error.contains("duplicate key") { + "ossie.parse.duplicate_key" + } else { + "ossie.parse.invalid_syntax" + }, format!("Invalid Ossie input: {error}"), "", None, @@ -388,6 +693,7 @@ impl OssieForwardAdapter { profile: None, kind: None, scope_ids: Vec::new(), + flat_root: false, diagnostics: vec![diagnostic( "ossie.schema.type", "The Ossie document root must be an object.", @@ -398,8 +704,38 @@ impl OssieForwardAdapter { }; let kind = classify_document(object); + let flat_root = + kind == Some(DocumentKind::Logical) && !object.contains_key("semantic_model"); let mut diagnostics = Vec::new(); - let profile = resolve_profile(object.get("version"), consumer, &mut diagnostics); + let mut profile = resolve_profile(object.get("version"), consumer, &mut diagnostics); + if kind == Some(DocumentKind::Ontology) && current_ontology_shape(object) { + if let Some(profile) = profile.as_mut() { + profile.schema_revision = Some(CURRENT_SCHEMA_REVISION.to_string()); + } + } + if flat_root { + if let Some(profile) = profile.as_mut() { + if profile.schema_version != "0.2.0.dev0" { + diagnostics.push(diagnostic( + "ossie.schema.profile_unsupported", + "Flat logical documents require version 0.2.0.dev0.", + "/version", + None, + )); + } + profile.schema_revision = Some(CURRENT_SCHEMA_REVISION.to_string()); + } + } + // Keep the internal scope traversal shared. Source pointers are mapped + // back to the flat document before returning diagnostics. + let root = if flat_root { + let mut scope = object.clone(); + scope.remove("version"); + serde_json::json!({"version": object.get("version"), "semantic_model": [scope]}) + } else { + root + }; + let object = root.as_object().expect("checked object root"); let mut scope_ids = Vec::new(); match kind { @@ -416,6 +752,13 @@ impl OssieForwardAdapter { None, )), } + if flat_root { + for diagnostic in &mut diagnostics { + if let Some(pointer) = diagnostic.instance_path.strip_prefix("/semantic_model/0") { + diagnostic.instance_path = pointer.to_string(); + } + } + } sort_diagnostics(&mut diagnostics); InspectedDocument { @@ -424,12 +767,13 @@ impl OssieForwardAdapter { kind, scope_ids, diagnostics, + flat_root, } } } fn classify_document(root: &Map) -> Option { - let logical = root.contains_key("semantic_model"); + let logical = root.contains_key("semantic_model") || root.contains_key("datasets"); let ontology = root.contains_key("ontology") || root.contains_key("ontology_mappings"); match (logical, ontology) { (true, false) => Some(DocumentKind::Logical), @@ -438,6 +782,41 @@ fn classify_document(root: &Map) -> Option { } } +fn current_ontology_shape(root: &Map) -> bool { + if root.contains_key("prefixes") { + return true; + } + if root + .get("ontology") + .and_then(Value::as_array) + .is_some_and(|concepts| { + concepts.iter().any(|concept| { + concept.get("iri").is_some() + || concept + .get("relationships") + .and_then(Value::as_array) + .is_some_and(|relationships| { + relationships + .iter() + .any(|relationship| relationship.get("iri").is_some()) + }) + }) + }) + { + return true; + } + root.get("ontology_mappings") + .and_then(Value::as_array) + .is_some_and(|mappings| { + mappings.iter().any(|mapping| { + mapping + .get("semantic_model") + .and_then(Value::as_object) + .is_some_and(|model| model.contains_key("version")) + }) + }) +} + fn resolve_profile( version: Option<&Value>, consumer: OssieConsumerProfile, @@ -487,6 +866,7 @@ fn resolve_profile( consumer_profile: consumer.label().to_string(), validation_schema_version: validation_version.to_string(), compatibility_alias_for: alias.map(str::to_string), + schema_revision: None, }) } @@ -495,7 +875,13 @@ fn validate_logical_root( profile: Option<&OssieProfile>, diagnostics: &mut Vec, ) { - let validation_version = profile.map(|profile| profile.validation_schema_version.as_str()); + let validation_version = profile.map(|profile| { + if profile.schema_revision.is_some() { + CURRENT_SCHEMA_VERSION + } else { + profile.validation_schema_version.as_str() + } + }); reject_unknown( root, &["version", "dialects", "vendors", "semantic_model"], @@ -503,7 +889,9 @@ fn validate_logical_root( diagnostics, None, ); - let allowed_dialects = if validation_version == Some("0.2.0.dev0") { + let allowed_dialects = if validation_version == Some(CURRENT_SCHEMA_VERSION) { + DIALECTS_CURRENT + } else if validation_version == Some("0.2.0.dev0") { DIALECTS_0_2_0 } else { DIALECTS_0_1_1 @@ -629,7 +1017,25 @@ fn validate_dataset( None, ); required_string(dataset, "name", pointer, diagnostics, None); - required_string(dataset, "source", pointer, diagnostics, None); + if let Some(source) = required_string(dataset, "source", pointer, diagnostics, None) { + if ![ + OssieTarget::DuckDb, + OssieTarget::Postgres, + OssieTarget::Snowflake, + OssieTarget::Databricks, + OssieTarget::BigQuery, + ] + .iter() + .any(|target| classify_source(source, *target).is_some()) + { + diagnostics.push(diagnostic( + "ossie.lowering.source_ambiguous", + "Dataset source is neither a table reference nor one SQL query.", + format!("{pointer}/source"), + None, + )); + } + } validate_string_array( dataset.get("primary_key"), &format!("{pointer}/primary_key"), @@ -663,7 +1069,7 @@ fn validate_field_schema( let Some(field) = require_object(value, pointer, diagnostics, None) else { return; }; - let allowed = if version == Some("0.2.0.dev0") { + let allowed = if matches!(version, Some("0.2.0.dev0") | Some(CURRENT_SCHEMA_VERSION)) { &[ "name", "expression", @@ -735,7 +1141,7 @@ fn validate_metric_schema( let Some(metric) = require_object(value, pointer, diagnostics, None) else { return; }; - let allowed = if version == Some("0.2.0.dev0") { + let allowed = if matches!(version, Some("0.2.0.dev0") | Some(CURRENT_SCHEMA_VERSION)) { &[ "name", "expression", @@ -846,7 +1252,9 @@ fn validate_expression_schema( None, )); } - let allowed_dialects = if version == Some("0.2.0.dev0") { + let allowed_dialects = if version == Some(CURRENT_SCHEMA_VERSION) { + DIALECTS_CURRENT + } else if version == Some("0.2.0.dev0") { DIALECTS_0_2_0 } else { DIALECTS_0_1_1 @@ -868,7 +1276,7 @@ fn validate_expression_schema( required_string(variant, "expression", &variant_pointer, diagnostics, None); if let Some(dialect) = dialect { let normalized = dialect.to_ascii_uppercase(); - if !allowed_dialects.contains(&normalized.as_str()) { + if !allowed_dialects.contains(&dialect) { diagnostics.push(diagnostic( "ossie.schema.enum", format!("Unsupported expression dialect {dialect:?} for this profile."), @@ -922,12 +1330,53 @@ fn validate_ontology_root(root: &Map, diagnostics: &mut Vec, ) -> Option { let scope = value.as_object()?; let datasets = scope.get("datasets")?.as_array()?; + let dataset_names = runtime_names( + &datasets + .iter() + .filter_map(|dataset| dataset.get("name").and_then(Value::as_str)) + .map(str::to_string) + .collect::>(), + ); let mut models = Vec::new(); let mut model_index = HashMap::new(); @@ -1303,7 +1760,19 @@ fn compile_scope( let dataset = dataset.as_object()?; let name = dataset.get("name")?.as_str()?.to_string(); let source = dataset.get("source")?.as_str()?.to_string(); - let field_names = dataset + let runtime_name = dataset_names.get(&name)?.clone(); + let Some(is_query) = classify_source(&source, target) else { + diagnostics.push(diagnostic( + "ossie.lowering.source_ambiguous", + format!( + "Dataset source for {name:?} is neither a table reference nor one SQL query." + ), + format!("{scope_pointer}/datasets/{dataset_index}/source"), + Some(scope_id), + )); + continue; + }; + let source_field_names = dataset .get("fields") .and_then(Value::as_array) .into_iter() @@ -1311,7 +1780,17 @@ fn compile_scope( .filter_map(Value::as_object) .filter_map(|field| field.get("name")) .filter_map(Value::as_str) - .map(|name| (normalize_identifier(name), name.to_string())) + .map(str::to_string) + .collect::>(); + let field_runtime_names = runtime_names(&source_field_names); + let field_names = source_field_names + .iter() + .map(|name| { + ( + normalize_identifier(name), + field_runtime_names[name].clone(), + ) + }) .collect::>(); let canonical_columns = |values: &[Value]| { values @@ -1349,10 +1828,21 @@ fn compile_scope( let Some(field_name) = field.get("name").and_then(Value::as_str) else { continue; }; - let pointer = format!( - "/semantic_model/{scope_index}/datasets/{dataset_index}/fields/{field_index}/expression" - ); - let Some(sql) = select_expression(field.get("expression"), target) else { + let pointer = + format!("{scope_pointer}/datasets/{dataset_index}/fields/{field_index}/expression"); + let selected = match lower_selected_expression(field.get("expression"), target) { + Ok(selected) => selected, + Err(message) => { + diagnostics.push(diagnostic( + "ossie.lowering.expression_invalid", + message, + pointer, + Some(scope_id), + )); + continue; + } + }; + let Some(sql) = selected else { diagnostics.push(diagnostic( "ossie.lowering.expression_unavailable", format!( @@ -1364,7 +1854,7 @@ fn compile_scope( )); continue; }; - if let Err(message) = validate_scalar_sql(&sql, target) { + if let Err(message) = validate_row_sql(&sql, target) { diagnostics.push(diagnostic( "ossie.lowering.expression_invalid", message, @@ -1400,7 +1890,7 @@ fn compile_scope( DimensionType::Categorical }; dimensions.push(Dimension { - name: field_name.to_string(), + name: field_runtime_names[field_name].clone(), r#type: dimension_type, logical_data_type, declared_is_time, @@ -1409,7 +1899,7 @@ fn compile_scope( supported_granularities: None, label: optional_string(field, "label"), description: optional_string(field, "description"), - metadata: None, + metadata: Some(serde_json::json!({"ossie_source_name": field_name})), meta: None, format: None, value_format_name: None, @@ -1419,9 +1909,8 @@ fn compile_scope( }); } - let is_query = source_is_query(&source, target); let model = Model { - name: name.clone(), + name: runtime_name, table: (!is_query).then_some(source.clone()), sql: is_query.then_some(source), source_uri: None, @@ -1438,7 +1927,9 @@ fn compile_scope( default_grain: None, label: None, description: optional_string(dataset, "description"), - metadata: None, + metadata: Some( + serde_json::json!({"ossie_source_name": name, "ossie_source_kind": if is_query { "query" } else { "table" }}), + ), meta: None, }; model_index.insert(normalize_identifier(&name), models.len()); @@ -1472,12 +1963,22 @@ fn compile_scope( let from_field_names = models[from_index] .dimensions .iter() - .map(|field| (normalize_identifier(&field.name), field.name.clone())) + .map(|field| { + ( + normalize_identifier(source_name(&field.name, &field.metadata)), + field.name.clone(), + ) + }) .collect::>(); let to_field_names = models[to_index] .dimensions .iter() - .map(|field| (normalize_identifier(&field.name), field.name.clone())) + .map(|field| { + ( + normalize_identifier(source_name(&field.name, &field.metadata)), + field.name.clone(), + ) + }) .collect::>(); let from_columns = relationship .get("from_columns") @@ -1523,6 +2024,24 @@ fn compile_scope( } let mut metrics = Vec::new(); + let metric_source_names = scope + .get("metrics") + .and_then(Value::as_array) + .into_iter() + .flatten() + .filter_map(|metric| metric.get("name").and_then(Value::as_str)) + .map(str::to_string) + .collect::>(); + let metric_runtime_names = runtime_names(&metric_source_names); + let metric_names = metric_source_names + .iter() + .map(|name| { + ( + normalize_identifier(name), + metric_runtime_names[name].clone(), + ) + }) + .collect::>(); for (metric_index, metric) in scope .get("metrics") .and_then(Value::as_array) @@ -1536,8 +2055,20 @@ fn compile_scope( let Some(name) = metric.get("name").and_then(Value::as_str) else { continue; }; - let pointer = format!("/semantic_model/{scope_index}/metrics/{metric_index}/expression"); - let Some(sql) = select_expression(metric.get("expression"), target) else { + let pointer = format!("{scope_pointer}/metrics/{metric_index}/expression"); + let selected = match lower_selected_expression(metric.get("expression"), target) { + Ok(selected) => selected, + Err(message) => { + diagnostics.push(diagnostic( + "ossie.lowering.expression_invalid", + message, + pointer, + Some(scope_id), + )); + continue; + } + }; + let Some(sql) = selected else { diagnostics.push(diagnostic( "ossie.lowering.expression_unavailable", format!( @@ -1558,12 +2089,31 @@ fn compile_scope( )); continue; } - let mut parsed = metric_from_sql_expression( - name.to_string(), - Some(sql), - optional_string(metric, "description"), - None, - ); + let expression_dialect = + selected_expression_dialect(metric.get("expression"), target).unwrap_or_default(); + let sql = match bind_metric_sql(&sql, &models, &metric_names, target, &expression_dialect) { + Ok(sql) => sql, + Err(message) => { + diagnostics.push(diagnostic( + "ossie.lowering.metric_unexecutable", + message, + pointer, + Some(scope_id), + )); + continue; + } + }; + let mut parsed = Metric::new(&metric_runtime_names[name]); + parsed.r#type = MetricType::Derived; + parsed.agg = None; + parsed.sql = Some(sql); + parsed.sql_is_complete = true; + parsed.description = optional_string(metric, "description"); + parsed.metadata = Some(serde_json::json!({ + "ossie_expression_dialect": expression_dialect, + "ossie_target_dialect": target.label(), + "ossie_source_name": name, + })); parsed.logical_data_type = metric .get("datatype") .and_then(Value::as_str) @@ -1580,43 +2130,164 @@ fn compile_scope( }) } -fn source_is_query(source: &str, target: OssieTarget) -> bool { - polyglot_sql::parse_one(source, target.parser_dialect()).is_ok_and(|expression| { - matches!( - expression, +fn classify_source(source: &str, target: OssieTarget) -> Option { + if let Ok(expressions) = polyglot_sql::parse(source, target.parser_dialect()) { + if expressions.len() != 1 { + return None; + } + if matches!( + expressions[0], Expression::Select(_) | Expression::Union(_) | Expression::Intersect(_) | Expression::Except(_) | Expression::Subquery(_) | Expression::Values(_) + ) { + let ast = serde_json::to_value(&expressions[0]).ok()?; + fn empty_select(value: &Value) -> bool { + if value.get("select").is_some_and(|select| { + select + .get("expressions") + .and_then(Value::as_array) + .is_none_or(Vec::is_empty) + }) { + return true; + } + match value { + Value::Object(object) => object.values().any(empty_select), + Value::Array(values) => values.iter().any(empty_select), + _ => false, + } + } + if empty_select(&ast) { + return None; + } + return Some(true); + } + } + if source.split_whitespace().next().is_some_and(|word| { + matches!( + word.to_ascii_uppercase().as_str(), + "SELECT" + | "WITH" + | "VALUES" + | "INSERT" + | "UPDATE" + | "DELETE" + | "CREATE" + | "DROP" + | "ALTER" + | "TRUNCATE" + | "CALL" ) - }) + }) { + return None; + } + let parsed = + polyglot_sql::parse(&format!("SELECT * FROM {source}"), target.parser_dialect()).ok()?; + if parsed.len() != 1 { + return None; + } + let Expression::Select(select) = &parsed[0] else { + return None; + }; + let from = select.from.as_ref()?; + if from.expressions.len() != 1 + || !select.joins.is_empty() + || select.where_clause.is_some() + || select.group_by.is_some() + || select.having.is_some() + || select.order_by.is_some() + || select.limit.is_some() + || select.offset.is_some() + || select.qualify.is_some() + || select.with.is_some() + { + return None; + } + let Expression::Table(table) = &from.expressions[0] else { + return None; + }; + if table.alias.is_some() || !table.column_aliases.is_empty() || table.name.name.is_empty() { + return None; + } + Some(false) } -fn select_expression(value: Option<&Value>, target: OssieTarget) -> Option { +#[cfg(test)] +fn source_is_query(source: &str, target: OssieTarget) -> bool { + classify_source(source, target) == Some(true) +} + +fn select_variant(value: Option<&Value>, target: OssieTarget) -> Option<(String, String)> { let variants = value?.as_object()?.get("dialects")?.as_array()?; let mut exact = None; let mut ansi = None; + let mut portable = None; for variant in variants { let variant = variant.as_object()?; let dialect = variant.get("dialect")?.as_str()?.to_ascii_uppercase(); let expression = variant.get("expression")?.as_str()?.to_string(); - if dialect == target.label() && exact.is_none() { - exact = Some(expression.clone()); + if dialect == target.label() && dialect != "ANSI_SQL" && exact.is_none() { + exact = Some((expression.clone(), dialect.clone())); } if dialect == "ANSI_SQL" && ansi.is_none() { - ansi = Some(expression); + ansi = Some((expression.clone(), dialect.clone())); + } + if dialect == "OSSIE_SQL_2026" && portable.is_none() { + portable = Some((expression, dialect)); } } - exact.or(ansi) + exact.or(portable).or(ansi) +} + +#[cfg(test)] +fn select_expression(value: Option<&Value>, target: OssieTarget) -> Option { + select_variant(value, target).map(|(sql, _)| sql) +} + +fn selected_expression_dialect(value: Option<&Value>, target: OssieTarget) -> Option { + select_variant(value, target).map(|(_, dialect)| dialect) +} + +fn lower_selected_expression( + value: Option<&Value>, + target: OssieTarget, +) -> std::result::Result, String> { + let Some((sql, dialect)) = select_variant(value, target) else { + return Ok(None); + }; + if dialect == "OSSIE_SQL_2026" { + super::ossie_sql::lower_ossie_sql(&sql, target.parser_dialect()).map(Some) + } else { + Ok(Some(sql)) + } } fn validate_scalar_sql(sql: &str, target: OssieTarget) -> std::result::Result<(), String> { + parse_scalar_sql(sql, target).map(|_| ()) +} + +fn parse_scalar_sql(sql: &str, target: OssieTarget) -> std::result::Result { + crate::semantic_input::with_semantic_stack(|| { + parse_scalar_sql_inner(sql, target).map_err(SidemanticError::SqlParse) + }) + .map_err(|error| error.to_string()) +} + +fn parse_scalar_sql_inner( + sql: &str, + target: OssieTarget, +) -> std::result::Result { let wrapped = format!("SELECT {sql}"); - let parsed = polyglot_sql::parse_one(&wrapped, target.parser_dialect()) - .map_err(|error| format!("Invalid {} SQL expression: {error}", target.label()))?; - let Expression::Select(select) = parsed else { + let mut statements = + crate::semantic_input::dialects::parse_many(&wrapped, target.parser_dialect()) + .map_err(|error| format!("Invalid {} SQL expression: {error}", target.label()))?; + if statements.len() != 1 { + return Err("Expected exactly one scalar SQL expression.".into()); + } + let Expression::Select(mut select) = statements.remove(0) else { return Err(format!( "Expected one scalar {} SQL expression.", target.label() @@ -1631,36 +2302,200 @@ fn validate_scalar_sql(sql: &str, target: OssieTarget) -> std::result::Result<() || select.order_by.is_some() || select.limit.is_some() || select.offset.is_some() - || select.with.is_some(); + || select.with.is_some() + || select.distinct; if select.expressions.len() != 1 || has_query_clauses { return Err(format!( "Expected one scalar {} SQL expression without query clauses.", target.label() )); } - let contains_query = select.expressions[0].contains(|expression| { - matches!( - expression, - Expression::Select(_) - | Expression::Union(_) - | Expression::Intersect(_) - | Expression::Except(_) - | Expression::Subquery(_) - | Expression::Values(_) - | Expression::Insert(_) - | Expression::Update(_) - | Expression::Delete(_) - ) - }); + let expression = select.expressions.remove(0); + if matches!( + expression, + Expression::Alias(_) | Expression::Aliases(_) | Expression::Star(_) + ) { + return Err("Expected a scalar value without a projection alias or wildcard.".into()); + } + let ast = serde_json::to_value(&expression).map_err(|error| error.to_string())?; + let contains_query = ast_contains_query(&ast); if contains_query { return Err(format!( "Expected one scalar {} SQL expression without a nested query.", target.label() )); } + Ok(expression) +} + +fn ast_contains_query(value: &Value) -> bool { + match value { + Value::Object(object) => { + object.keys().any(|key| { + matches!( + key.as_str(), + "select" + | "subquery" + | "union" + | "intersect" + | "except" + | "values" + | "insert" + | "update" + | "delete" + | "create" + | "drop" + | "command" + ) + }) || object.values().any(ast_contains_query) + } + Value::Array(values) => values.iter().any(ast_contains_query), + _ => false, + } +} + +fn validate_row_sql(sql: &str, target: OssieTarget) -> std::result::Result<(), String> { + let expression = parse_scalar_sql(sql, target)?; + let value = serde_json::to_value(expression).map_err(|error| error.to_string())?; + fn has_aggregate(value: &Value) -> bool { + if value + .as_object() + .is_some_and(|object| object.contains_key("window_function")) + { + return false; + } + if crate::core::is_aggregate_ast_node(value) { + return true; + } + match value { + Value::Object(object) => object.values().any(has_aggregate), + Value::Array(values) => values.iter().any(has_aggregate), + _ => false, + } + } + if has_aggregate(&value) { + return Err( + "Ossie dataset fields are row-level expressions and cannot contain aggregates.".into(), + ); + } Ok(()) } +fn bind_metric_sql( + sql: &str, + models: &[Model], + metrics: &HashMap, + target: OssieTarget, + expression_dialect: &str, +) -> std::result::Result { + let expression = parse_scalar_sql(sql, target)?; + let mut value = serde_json::to_value(expression).map_err(|error| error.to_string())?; + fn key(identifier: &Value, expression_dialect: &str) -> String { + let name = identifier["name"].as_str().unwrap_or(""); + if identifier["quoted"].as_bool() == Some(true) + && !matches!(expression_dialect, "BIGQUERY" | "DATABRICKS") + { + name.to_string() + } else { + normalize_identifier(name) + } + } + fn runtime_identifier(name: &str) -> Value { + serde_json::json!({"name": name, "quoted": name.starts_with('"')}) + } + fn bind( + value: &mut Value, + models: &[Model], + metrics: &HashMap, + expression_dialect: &str, + ) -> std::result::Result { + if let Some(column) = value.get_mut("column").and_then(Value::as_object_mut) { + let field_key = key(&column["name"], expression_dialect); + if column.get("table").is_none_or(Value::is_null) { + if let Some(name) = metrics.get(&field_key) { + if column["name"]["name"].as_str() != Some(name.as_str()) { + column.insert("name".into(), runtime_identifier(name)); + return Ok(true); + } + return Ok(false); + } + } + let model = if column.get("table").is_some_and(|table| !table.is_null()) { + let table_key = key(&column["table"], expression_dialect); + Some( + models + .iter() + .find(|model| { + normalize_identifier(source_name(&model.name, &model.metadata)) + == table_key + }) + .ok_or_else(|| { + format!( + "Unknown logical dataset in column reference {:?}.", + column["table"]["name"] + ) + })?, + ) + } else { + let candidates = models + .iter() + .filter(|model| { + model.dimensions.iter().any(|field| { + normalize_identifier(source_name(&field.name, &field.metadata)) + == field_key + }) + }) + .collect::>(); + if candidates.len() == 1 { + Some(candidates[0]) + } else if candidates.is_empty() && models.len() == 1 { + Some(&models[0]) + } else { + None + } + }; + if let Some(model) = model { + let field = model.dimensions.iter().find(|field| { + normalize_identifier(source_name(&field.name, &field.metadata)) == field_key + }); + let field_name = field + .map(|field| field.name.as_str()) + .unwrap_or_else(|| column["name"]["name"].as_str().unwrap_or("")); + if column["table"]["name"].as_str() != Some(model.name.as_str()) + || column["name"]["name"].as_str() != Some(field_name) + { + let field_name = field_name.to_string(); + column.insert("table".into(), runtime_identifier(&model.name)); + column.insert("name".into(), runtime_identifier(&field_name)); + return Ok(true); + } + } + return Ok(false); + } + let mut changed = false; + match value { + Value::Object(object) => { + for child in object.values_mut() { + changed |= bind(child, models, metrics, expression_dialect)?; + } + } + Value::Array(values) => { + for child in values { + changed |= bind(child, models, metrics, expression_dialect)?; + } + } + _ => {} + } + Ok(changed) + } + if !bind(&mut value, models, metrics, expression_dialect)? { + return Ok(sql.to_string()); + } + let expression: Expression = + serde_json::from_value(value).map_err(|error| error.to_string())?; + polyglot_sql::generate(&expression, target.parser_dialect()).map_err(|error| error.to_string()) +} + fn duplicate_names( values: &[Value], pointer: &str, @@ -1677,6 +2512,17 @@ fn duplicate_names( else { continue; }; + if !identifier_syntax_valid(name) { + diagnostics.push(diagnostic("ossie.semantic.identifier.invalid", "Expected an ANSI regular identifier or a non-empty double-quoted identifier with doubled interior quotes.", format!("{pointer}/{index}/name"), scope)); + } + if identifier_length(name) > 128 { + diagnostics.push(diagnostic( + "ossie.semantic.identifier.length_exceeded", + "Ossie identifiers are limited to 128 decoded characters.", + format!("{pointer}/{index}/name"), + scope, + )); + } if let Some(first_index) = first.insert(normalize_identifier(name), index) { diagnostics.push(diagnostic( code, @@ -1696,6 +2542,98 @@ fn normalize_identifier(identifier: &str) -> String { } } +fn identifier_length(identifier: &str) -> usize { + if identifier.starts_with('"') && identifier.ends_with('"') && identifier.len() >= 2 { + normalize_identifier(identifier).chars().count() + } else { + identifier.chars().count() + } +} + +fn identifier_syntax_valid(identifier: &str) -> bool { + if identifier.starts_with('"') { + if identifier.len() < 3 || !identifier.ends_with('"') { + return false; + } + let mut body = identifier[1..identifier.len() - 1].chars(); + while let Some(character) = body.next() { + if character == '\0' || character == '"' && body.next() != Some('"') { + return false; + } + } + return true; + } + let mut characters = identifier.chars(); + characters + .next() + .is_some_and(|character| character == '_' || character.is_alphabetic()) + && characters.all(|character| character == '_' || character.is_alphanumeric()) +} + +fn source_name<'a>(name: &'a str, metadata: &'a Option) -> &'a str { + metadata + .as_ref() + .and_then(|metadata| metadata.get("ossie_source_name")) + .and_then(Value::as_str) + .unwrap_or(name) +} + +fn runtime_names(names: &[String]) -> HashMap { + let quoted = |name: &str| name.len() >= 2 && name.starts_with('"') && name.ends_with('"'); + let candidates = names + .iter() + .map(|name| { + ( + name.clone(), + if quoted(name) { + normalize_identifier(name) + } else { + name.clone() + }, + ) + }) + .collect::>(); + let mut counts = HashMap::new(); + for candidate in candidates.values() { + *counts + .entry(candidate.to_ascii_lowercase()) + .or_insert(0usize) += 1; + } + let mut used = candidates + .values() + .map(|candidate| candidate.to_ascii_lowercase()) + .collect::>(); + let mut ordered = names.to_vec(); + ordered.sort(); + let mut result = HashMap::new(); + for name in ordered { + let candidate = &candidates[&name]; + let mut characters = candidate.chars(); + let safe = characters + .next() + .is_some_and(|character| character.is_ascii_alphabetic() || character == '_') + && characters.all(|character| character.is_ascii_alphanumeric() || character == '_'); + if safe && (!quoted(&name) || counts[&candidate.to_ascii_lowercase()] == 1) { + result.insert(name, candidate.clone()); + continue; + } + // Stable non-security identifier fingerprint, shared with Python lowering. + let fingerprint = name.bytes().fold(0xcbf29ce484222325u64, |hash, byte| { + (hash ^ u64::from(byte)).wrapping_mul(0x100000001b3) + }); + let base = format!("__ossie_{fingerprint:016x}"); + let mut candidate = base.clone(); + let mut suffix = 1; + while used.contains(&candidate.to_ascii_lowercase()) { + candidate = format!("{base}_{suffix}"); + suffix += 1; + } + used.insert(candidate.to_ascii_lowercase()); + result.insert(name, candidate); + } + result +} + fn normalized_strings(values: &[Value]) -> Vec { values .iter() @@ -1723,6 +2661,101 @@ fn reject_unknown( scope, )); } + validate_metadata(object, pointer, diagnostics, scope); +} + +fn validate_metadata( + object: &Map, + pointer: &str, + diagnostics: &mut Vec, + scope: Option<&str>, +) { + for name in ["description", "label"] { + if object.get(name).is_some_and(|value| !value.is_string()) { + diagnostics.push(diagnostic( + "ossie.schema.type", + format!("{name} must be a string."), + format!("{pointer}/{name}"), + scope, + )); + } + } + if let Some(context) = object.get("ai_context") { + if let Some(context) = context.as_object() { + if context + .get("instructions") + .is_some_and(|value| !value.is_string()) + { + diagnostics.push(diagnostic( + "ossie.schema.type", + "AI instructions must be a string.", + format!("{pointer}/ai_context/instructions"), + scope, + )); + } + for name in ["synonyms", "examples"] { + if let Some(values) = context.get(name) { + if !values + .as_array() + .is_some_and(|values| values.iter().all(Value::is_string)) + { + diagnostics.push(diagnostic( + "ossie.schema.type", + format!("AI {name} must be an array of strings."), + format!("{pointer}/ai_context/{name}"), + scope, + )); + } + } + } + } else if !context.is_string() { + diagnostics.push(diagnostic( + "ossie.schema.type", + "ai_context must be a string or object.", + format!("{pointer}/ai_context"), + scope, + )); + } + } + if let Some(extensions) = + optional_array(object, "custom_extensions", pointer, diagnostics, scope) + { + for (index, extension) in extensions.iter().enumerate() { + let extension_pointer = format!("{pointer}/custom_extensions/{index}"); + if let Some(extension) = + require_object(extension, &extension_pointer, diagnostics, scope) + { + for key in extension + .keys() + .filter(|key| !matches!(key.as_str(), "vendor_name" | "data")) + { + diagnostics.push(diagnostic( + "ossie.schema.additional_properties", + format!("Unexpected extension property {key:?}."), + &extension_pointer, + scope, + )); + } + for key in ["vendor_name", "data"] { + match extension.get(key) { + None => diagnostics.push(diagnostic( + "ossie.schema.required", + format!("Required extension property {key:?} is missing."), + &extension_pointer, + scope, + )), + Some(value) if !value.is_string() => diagnostics.push(diagnostic( + "ossie.schema.type", + format!("Extension {key} must be a string."), + format!("{extension_pointer}/{key}"), + scope, + )), + _ => {} + } + } + } + } + } } fn require_object<'a>( @@ -1814,7 +2847,17 @@ fn required_string<'a>( return None; }; match value.as_str().filter(|value| !value.is_empty()) { - Some(value) => Some(value), + Some(value) => { + if matches!(key, "from" | "to") && identifier_length(value) > 128 { + diagnostics.push(diagnostic( + "ossie.semantic.identifier.length_exceeded", + "Ossie identifiers are limited to 128 decoded characters.", + format!("{pointer}/{key}"), + scope, + )); + } + Some(value) + } None => { diagnostics.push(diagnostic( "ossie.schema.type", @@ -1923,6 +2966,302 @@ fn status_error(status: &OssieStatus) -> SidemanticError { mod tests { use super::*; + fn flat_document() -> Value { + serde_json::json!({ + "version": "0.2.0.dev0", "name": "commerce", + "datasets": [{"name": "Orders", "source": "SELECT 10 AS amount UNION ALL SELECT 20 AS amount", + "fields": [{"name": "Amount", "expression": {"dialects": [{"dialect": "ANSI_SQL", "expression": "amount * 2"}]}}]}], + "metrics": [{"name": "total", "expression": {"dialects": [{"dialect": "ANSI_SQL", "expression": "SUM(orders.amount)"}]}}] + }) + } + + #[test] + fn flat_expression_import_preserves_logical_complete_sql() { + let scope = OssieForwardAdapter + .select_scope( + &flat_document().to_string(), + OssieSerialization::Json, + OssieConsumerProfile::OssieCore, + OssieTarget::DuckDb, + None, + ) + .unwrap(); + assert_eq!(scope.models[0].name, "Orders"); + assert_eq!(scope.metrics[0].sql.as_deref(), Some("SUM(Orders.Amount)")); + assert!(scope.metrics[0].sql_is_complete); + assert!(scope.metrics[0].agg.is_none()); + assert_eq!( + scope.metrics[0].metadata.as_ref().unwrap()["ossie_expression_dialect"], + "ANSI_SQL" + ); + } + + #[test] + fn unsafe_sources_fail_inspection_and_compilation() { + for source in [ + "DELETE FROM orders", + "1 + 2", + "SELECT", + "orders; DROP TABLE orders", + "orders AS aliased", + ] { + let mut document = flat_document(); + document["datasets"][0]["source"] = source.into(); + let content = document.to_string(); + let status = OssieForwardAdapter.inspect( + &content, + OssieSerialization::Json, + OssieConsumerProfile::OssieCore, + ); + assert!(!status.valid && !status.executable, "{source}: {status:?}"); + assert!( + OssieForwardAdapter + .parse_catalog( + &content, + OssieSerialization::Json, + OssieConsumerProfile::OssieCore, + OssieTarget::DuckDb + ) + .is_err(), + "{source}" + ); + } + } + + #[test] + fn graph_boundary_validates_complete_expressions_and_dependency_cycles() { + for sql in [ + "STDDEV_SAMP(orders.amount)", + "VAR_POP(orders.amount)", + "QUANTILE_CONT(orders.amount, 0.5 ORDER BY orders.amount DESC)", + ] { + let mut document = flat_document(); + document["metrics"][0]["expression"]["dialects"][0]["expression"] = sql.into(); + let scope = OssieForwardAdapter + .select_scope( + &document.to_string(), + OssieSerialization::Json, + OssieConsumerProfile::OssieCore, + OssieTarget::DuckDb, + None, + ) + .unwrap(); + assert!(scope.into_graph().is_ok(), "{sql}"); + } + for (second, valid) in [("SUM(orders.amount)", true), ("total + 1", false)] { + let mut document = flat_document(); + document["metrics"][0]["expression"]["dialects"][0]["expression"] = "later + 1".into(); + document["metrics"].as_array_mut().unwrap().push(serde_json::json!({"name":"later", "expression":{"dialects":[{"dialect":"ANSI_SQL", "expression":second}]}})); + let scope = OssieForwardAdapter + .select_scope( + &document.to_string(), + OssieSerialization::Json, + OssieConsumerProfile::OssieCore, + OssieTarget::DuckDb, + None, + ) + .unwrap(); + assert_eq!(scope.into_graph().is_ok(), valid, "{second}"); + } + } + + #[test] + fn row_aggregates_and_projection_aliases_fail_closed() { + for expression in [ + "SUM(amount)", + "STDDEV(amount)", + "COUNT(*)", + "amount AS renamed", + "*", + "SUM((SELECT 1))", + ] { + let mut document = flat_document(); + document["datasets"][0]["fields"][0]["expression"]["dialects"][0]["expression"] = + expression.into(); + let error = OssieForwardAdapter + .parse_catalog( + &document.to_string(), + OssieSerialization::Json, + OssieConsumerProfile::OssieCore, + OssieTarget::DuckDb, + ) + .unwrap_err(); + assert!( + error + .to_string() + .contains("ossie.lowering.expression_invalid"), + "{expression}: {error}" + ); + } + } + + #[test] + fn quoted_and_regular_source_names_have_distinct_runtime_bindings() { + let mut document = flat_document(); + let mut quoted = document["datasets"][0].clone(); + quoted["name"] = "\"Orders\"".into(); + document["datasets"].as_array_mut().unwrap().push(quoted); + document["metrics"].as_array_mut().unwrap().push(serde_json::json!({ + "name": "quoted", "expression": {"dialects": [{"dialect": "ANSI_SQL", "expression": "SUM(\"Orders\".amount)"}]} + })); + let scope = OssieForwardAdapter + .select_scope( + &document.to_string(), + OssieSerialization::Json, + OssieConsumerProfile::OssieCore, + OssieTarget::DuckDb, + None, + ) + .unwrap(); + assert_ne!( + scope.models[0].name.to_lowercase(), + scope.models[1].name.to_lowercase() + ); + assert_eq!( + scope.models[1].metadata.as_ref().unwrap()["ossie_source_name"], + "\"Orders\"" + ); + assert!(scope.metrics[1] + .sql + .as_ref() + .unwrap() + .contains(&scope.models[1].name)); + } + + #[test] + fn duplicate_json_keys_and_invalid_metadata_are_rejected() { + let duplicate = r#"{"version":"0.2.0.dev0","name":"first","name":"second","datasets":[{"name":"orders","source":"orders"}]}"#; + let status = OssieForwardAdapter.inspect( + duplicate, + OssieSerialization::Json, + OssieConsumerProfile::OssieCore, + ); + assert!(!status.valid); + assert_eq!(status.diagnostics[0].code, "ossie.parse.duplicate_key"); + for (name, value) in [ + ("description", serde_json::json!(42)), + ("ai_context", serde_json::json!([])), + ( + "custom_extensions", + serde_json::json!([{"vendor_name":"example","data":{}}]), + ), + ] { + let mut document = flat_document(); + document["datasets"][0][name] = value; + assert!( + !OssieForwardAdapter + .inspect( + &document.to_string(), + OssieSerialization::Json, + OssieConsumerProfile::OssieCore + ) + .valid, + "{name}" + ); + } + } + + #[test] + fn yaml_merge_overrides_are_preserved_and_expansion_is_bounded() { + let content = r#" +version: 0.2.0.dev0 +name: commerce +datasets: + - name: orders + source: orders + fields: + - &field + name: amount + expression: {dialects: [{dialect: ANSI_SQL, expression: amount}]} + - <<: *field + name: other_amount +"#; + let status = OssieForwardAdapter.inspect( + content, + OssieSerialization::Yaml, + OssieConsumerProfile::OssieCore, + ); + assert!(status.valid, "{status:?}"); + let oversized = format!("{}0{}", "[".repeat(300), "]".repeat(300)); + assert!( + !OssieForwardAdapter + .inspect( + &oversized, + OssieSerialization::Json, + OssieConsumerProfile::OssieCore + ) + .valid + ); + } + + #[test] + fn current_ontology_prefixes_are_valid_but_not_executable() { + let document = serde_json::json!({"version":"0.2.0.dev0", "name":"business", + "prefixes":{"ex":"https://example.com/"}, "requires":[], "ontology":[{"concept":"Order", "type":"EntityType", "iri":"ex:Order"}]}); + let status = OssieForwardAdapter.inspect( + &document.to_string(), + OssieSerialization::Json, + OssieConsumerProfile::OssieCore, + ); + assert!(status.valid, "{status:?}"); + assert!(!status.executable); + assert_eq!( + status.profile.unwrap().schema_revision.as_deref(), + Some(CURRENT_SCHEMA_REVISION) + ); + } + + #[test] + fn ontology_revision_detects_all_current_markers_without_prefixes() { + for marker in [ + "legacy", + "concept_iri", + "relationship_iri", + "embedded_model", + ] { + let mut document = serde_json::json!({ + "version": "0.2.0.dev0", "name": "business", + "ontology": [{"concept": "Order", "type": "EntityType"}] + }); + match marker { + "concept_iri" => { + document["ontology"][0]["iri"] = "https://example.com/Order".into(); + } + "relationship_iri" => { + document["ontology"][0]["relationships"] = serde_json::json!([{ + "name": "related", "roles": [{"concept": "Order", "name": "other"}], + "verbalizes": [], + "iri": "https://example.com/related" + }]); + } + "embedded_model" => { + document["ontology_mappings"] = serde_json::json!([{ + "semantic_model": flat_document(), "concept_mappings": [] + }]); + } + _ => {} + } + for serialization in [OssieSerialization::Json, OssieSerialization::Yaml] { + let content = match serialization { + OssieSerialization::Json => document.to_string(), + OssieSerialization::Yaml => serde_yaml::to_string(&document).unwrap(), + }; + let status = OssieForwardAdapter.inspect( + &content, + serialization, + OssieConsumerProfile::OssieCore, + ); + assert!(status.valid, "{marker}: {status:?}"); + assert!(!status.executable); + assert_eq!( + status.profile.unwrap().schema_revision.as_deref(), + (marker != "legacy").then_some(CURRENT_SCHEMA_REVISION), + "{marker}" + ); + } + } + } + const MULTI_SCOPE: &str = r#" version: 0.2.0.dev0 semantic_model: diff --git a/sidemantic-rs/src/adapters/ossie_sql.rs b/sidemantic-rs/src/adapters/ossie_sql.rs new file mode 100644 index 000000000..518f7586e --- /dev/null +++ b/sidemantic-rs/src/adapters/ossie_sql.rs @@ -0,0 +1,621 @@ +//! OSSIE_SQL_2026 source expressions and portable function normalization. +//! +//! Warehouse SQL alternatives do not use this module. Unknown function calls +//! remain extensions, as requested by the expression-language proposal. + +use polyglot_sql::expressions::{DateTimeField, ExtractFunc, Function, Literal}; +use polyglot_sql::tokens::TokenType; +use polyglot_sql::{Dialect, DialectType, Error, Expression}; + +fn invalid(message: impl Into) -> Error { + Error::Generate(message.into()) +} + +/// Parse the portable source grammar independently from the execution target. +pub fn parse_portable_expression(sql: &str) -> Result { + // Enter expression grammar before tokens such as TRUNCATE can be mistaken + // for statement keywords. The wrapper also prevents trailing SQL clauses. + let prepared = prepare_source_literals(sql)?; + let mut expressions = polyglot_sql::parse(&format!("({prepared})"), DialectType::Snowflake) + .map_err(|error| error.to_string())?; + if expressions.len() != 1 { + return Err("OSSIE_SQL_2026 requires exactly one expression".into()); + } + let expression = match expressions.remove(0) { + Expression::Paren(parentheses) => parentheses.this, + expression => expression, + }; + if matches!( + expression, + Expression::Alias(_) | Expression::Star(_) | Expression::Tuple(_) + ) { + return Err("OSSIE_SQL_2026 requires a scalar expression without an alias".into()); + } + fn validate_tree(value: &serde_json::Value) -> Result<(), String> { + // The pinned DFS walker does not enumerate typed aggregate arguments. + // Validate the same complete tree that the lowering pass transforms. + if let Ok(node) = serde_json::from_value::(value.clone()) { + validate_node(&node)?; + } + match value { + serde_json::Value::Object(fields) => { + for child in fields.values() { + validate_tree(child)?; + } + } + serde_json::Value::Array(values) => { + for child in values { + validate_tree(child)?; + } + } + _ => {} + } + Ok(()) + } + let value = serde_json::to_value(&expression).map_err(|error| error.to_string())?; + validate_tree(&value)?; + Ok(expression) +} + +fn prepare_source_literals(sql: &str) -> Result { + // Polyglot recognizes TIMESTAMP_NTZ in CAST but not as a typed literal. + // Rewrite only the lexer-confirmed type/literal pair, never quoted text. + let tokens = Dialect::get(DialectType::Snowflake) + .tokenize(sql) + .map_err(|error| error.to_string())?; + let offsets = sql + .char_indices() + .map(|(offset, _)| offset) + .chain([sql.len()]) + .collect::>(); + let mut prepared = sql.to_owned(); + for pair in tokens.windows(2).rev() { + if pair[0].text.eq_ignore_ascii_case("TIMESTAMP_NTZ") + && !matches!( + pair[0].token_type, + TokenType::String | TokenType::QuotedIdentifier + ) + && pair[1].token_type == TokenType::String + { + let start = offsets[pair[0].span.start]; + let literal_start = offsets[pair[1].span.start]; + let end = offsets[pair[1].span.end]; + prepared.replace_range( + start..end, + &format!("CAST({} AS TIMESTAMP_NTZ)", &sql[literal_start..end]), + ); + } + } + Ok(prepared) +} + +fn validate_node(node: &Expression) -> Result<(), String> { + if polyglot_sql::is_ddl(node) + || matches!( + node, + Expression::Select(_) + | Expression::Subquery(_) + | Expression::Union(_) + | Expression::Intersect(_) + | Expression::Except(_) + | Expression::Insert(_) + | Expression::Update(_) + | Expression::Delete(_) + | Expression::Copy(_) + | Expression::From(_) + | Expression::Join(_) + | Expression::Where(_) + | Expression::GroupBy(_) + | Expression::With(_) + | Expression::Array(_) + | Expression::ArrayFunc(_) + | Expression::Subscript(_) + | Expression::ArraySlice(_) + | Expression::Command(_) + ) + { + return Err( + "OSSIE_SQL_2026 expressions cannot contain queries, statements, or arrays".into(), + ); + } + if matches!(node, Expression::Dot(_)) { + // Two-part field references are Column nodes with a table qualifier; + // Dot is the parser's representation of additional path segments. + return Err("OSSIE_SQL_2026 field references have at most two identifiers".into()); + } + if let Expression::Identifier(identifier) = node { + if identifier.name.chars().count() > 128 { + return Err("OSSIE_SQL_2026 identifiers cannot exceed 128 characters".into()); + } + } + if let Expression::Column(column) = node { + if column.name.name.chars().count() > 128 + || column + .table + .as_ref() + .is_some_and(|name| name.name.chars().count() > 128) + { + return Err("OSSIE_SQL_2026 identifiers cannot exceed 128 characters".into()); + } + } + Ok(()) +} + +fn source_sql(expression: &Expression) -> polyglot_sql::Result { + Dialect::get(DialectType::Snowflake).generate(expression) +} + +fn parse_source(sql: &str) -> polyglot_sql::Result { + polyglot_sql::parse_one(sql, DialectType::Snowflake) +} + +fn extract(part: &str, value: Expression) -> polyglot_sql::Result { + let field = match part.to_uppercase().as_str() { + "YEAR" => DateTimeField::Year, + "QUARTER" => DateTimeField::Quarter, + "MONTH" => DateTimeField::Month, + "WEEK" => DateTimeField::Week, + "DAY" => DateTimeField::Day, + "DAYOFWEEK" => DateTimeField::DayOfWeek, + "DAYOFYEAR" => DateTimeField::DayOfYear, + "HOUR" => DateTimeField::Hour, + "MINUTE" => DateTimeField::Minute, + "SECOND" => DateTimeField::Second, + "MILLISECOND" => DateTimeField::Millisecond, + _ => { + return Err(invalid(format!( + "Unsupported OSSIE_SQL_2026 date part {part:?}" + ))) + } + }; + Ok(Expression::Extract(Box::new(ExtractFunc { + this: value, + field, + }))) +} + +fn truncate_sql(value: &str, decimals: &str) -> String { + // Preserve decimal literals instead of flooring an inexact POWER product. + let quantum = match decimals.parse::() { + Ok(places) if (1..=38).contains(&places) => { + format!("0.{}1", "0".repeat((places - 1) as usize)) + } + Ok(places) if (-38..=0).contains(&places) => format!("1{}", "0".repeat((-places) as usize)), + _ => format!("POWER(10, -({decimals}))"), + }; + format!("CASE WHEN ABS(ROUND({value}, {decimals})) > ABS({value}) THEN ROUND({value}, {decimals}) - SIGN({value}) * {quantum} ELSE ROUND({value}, {decimals}) END") +} + +fn duckdb_date_add(unit: &str, amount: &str, value: &str) -> polyglot_sql::Result { + let unit = unit.trim_matches('\'').to_uppercase(); + if !matches!( + unit.as_str(), + "YEAR" + | "QUARTER" + | "MONTH" + | "WEEK" + | "DAY" + | "HOUR" + | "MINUTE" + | "SECOND" + | "MILLISECOND" + ) { + return Err(invalid(format!( + "Unsupported OSSIE_SQL_2026 DATEADD unit {unit:?}" + ))); + } + // INTERVAL -1 MONTH is invalid DuckDB syntax. Multiplication handles + // negative literals and arbitrary numeric expressions without ambiguity. + polyglot_sql::parse_one( + &format!("(({value}) + ({amount}) * INTERVAL '1 {unit}')"), + DialectType::DuckDB, + ) +} + +fn normalize(node: Expression, target: DialectType) -> polyglot_sql::Result { + match node { + Expression::CurrentTime(_) if target == DialectType::DuckDB => { + polyglot_sql::parse_one("LOCALTIME", target) + } + Expression::DateAdd(ref function) if target == DialectType::DuckDB => duckdb_date_add( + &format!("{:?}", function.unit), + &source_sql(&function.interval)?, + &source_sql(&function.this)?, + ), + Expression::DateDiff(ref function) if target == DialectType::DuckDB => { + // The canonical node stores (end, start), while Polyglot's generic + // DuckDB emitter leaves that order unchanged and the unit unquoted. + let unit = format!( + "{:?}", + function + .unit + .as_ref() + .ok_or_else(|| invalid("DATEDIFF requires a unit"))? + ) + .to_uppercase(); + Ok(Expression::Function(Box::new(Function::new( + "DATE_DIFF", + vec![ + Expression::Literal(Literal::String(unit)), + function.expression.clone(), + function.this.clone(), + ], + )))) + } + Expression::DateDiff(ref function) if target == DialectType::PostgreSQL => { + let unit = format!( + "{:?}", + function + .unit + .as_ref() + .ok_or_else(|| invalid("DATEDIFF requires a unit"))? + ) + .to_uppercase(); + let start = source_sql(&function.expression)?; + let end = source_sql(&function.this)?; + let years = format!("(EXTRACT(YEAR FROM {end}) - EXTRACT(YEAR FROM {start}))"); + let sql = match unit.as_str() { + "YEAR" => years, + "MONTH" => format!( + "({years} * 12 + EXTRACT(MONTH FROM {end}) - EXTRACT(MONTH FROM {start}))" + ), + "QUARTER" => format!( + "({years} * 4 + EXTRACT(QUARTER FROM {end}) - EXTRACT(QUARTER FROM {start}))" + ), + "DAY" | "HOUR" | "MINUTE" | "SECOND" => { + let divisor = match unit.as_str() { + "DAY" => 86400, + "HOUR" => 3600, + "MINUTE" => 60, + _ => 1, + }; + format!("(EXTRACT(EPOCH FROM (DATE_TRUNC('{unit}', CAST({end} AS TIMESTAMP)) - DATE_TRUNC('{unit}', CAST({start} AS TIMESTAMP)))) / {divisor})") + } + _ => { + return Err(invalid(format!( + "Unsupported PostgreSQL DATEDIFF unit {unit}" + ))) + } + }; + polyglot_sql::parse_one(&sql, target) + } + Expression::Log(ref function) if function.base.is_some() => { + let value = source_sql(&function.this)?; + let base = source_sql(function.base.as_ref().unwrap())?; + parse_source(&format!("(LN({value}) / LN({base}))")) + } + Expression::ToTimestamp(ref function) if function.format.is_none() => parse_source( + &format!("CAST({} AS TIMESTAMP_NTZ)", source_sql(&function.this)?), + ), + Expression::ToDate(ref function) if function.format.is_none() => { + parse_source(&format!("CAST({} AS DATE)", source_sql(&function.this)?)) + } + Expression::RegexpLike(ref function) => { + let value = source_sql(&function.this)?; + let pattern = source_sql(&function.pattern)?; + match target { + DialectType::DuckDB => { + polyglot_sql::parse_one(&format!("REGEXP_MATCHES({value}, {pattern})"), target) + } + DialectType::Snowflake => { + parse_source(&format!("REGEXP_INSTR({value}, {pattern}) > 0")) + } + _ => Ok(node), + } + } + Expression::Extract(mut function) => { + if target == DialectType::PostgreSQL { + function.field = match function.field { + DateTimeField::DayOfYear => DateTimeField::Custom("DOY".into()), + DateTimeField::DayOfWeek => DateTimeField::Custom("DOW".into()), + part => part, + }; + } + Ok(Expression::Extract(function)) + } + Expression::DateTrunc(ref function) if function.unit == DateTimeField::Week => { + let value = source_sql(&function.this)?; + match target { + DialectType::BigQuery => { + polyglot_sql::parse_one(&format!("DATE_TRUNC({value}, WEEK(MONDAY))"), target) + } + DialectType::Snowflake => parse_source(&format!( + "DATEADD(day, 1 - DAYOFWEEKISO({value}), DATE_TRUNC('day', {value}))" + )), + _ => Ok(node), + } + } + Expression::Median(_) if target == DialectType::BigQuery => Err(invalid( + "BigQuery exact MEDIAN requires query-level lowering", + )), + Expression::WithinGroup(ref group) if target == DialectType::BigQuery => { + let sql = source_sql(&group.this)?.to_uppercase(); + if sql.starts_with("PERCENTILE_CONT(") || sql.starts_with("PERCENTILE_DISC(") { + Err(invalid( + "BigQuery exact ordered-set percentiles require query-level lowering", + )) + } else { + Ok(node) + } + } + Expression::Function(function) => { + let name = function.name.to_uppercase(); + let args = &function.args; + let rendered = args + .iter() + .map(source_sql) + .collect::>>()?; + if target == DialectType::DuckDB { + match (name.as_str(), rendered.as_slice()) { + ("CURRENT_TIME", []) => return polyglot_sql::parse_one("LOCALTIME", target), + ("CURRENT_TIMESTAMP", []) => { + return polyglot_sql::parse_one("CURRENT_TIMESTAMP", target) + } + ("CURRENT_DATE", []) => return polyglot_sql::parse_one("CURRENT_DATE", target), + ("DATEADD", [unit, amount, value]) => { + return duckdb_date_add(unit, amount, value) + } + _ => {} + } + } + let replacement = match (name.as_str(), rendered.as_slice()) { + ("TO_TIMESTAMP", [value]) => Some(format!("CAST({value} AS TIMESTAMP_NTZ)")), + ("TO_DATE", [value]) => Some(format!("CAST({value} AS DATE)")), + ("LOG10", [value]) => Some(format!("(LN({value}) / LN(10))")), + ("ZEROIFNULL", [value]) => Some(format!("COALESCE({value}, 0)")), + ("NULLIFZERO", [value]) => Some(format!("NULLIF({value}, 0)")), + ("TRUNC" | "TRUNCATE", [value, decimals]) => Some(truncate_sql(value, decimals)), + ("TRUNC" | "TRUNCATE", [value]) => Some(truncate_sql(value, "0")), + ("CONTAINS", [value, part]) => Some(format!("(POSITION({part} IN {value}) > 0)")), + ("STARTSWITH", [value, part]) => { + Some(format!("(LEFT({value}, LENGTH({part})) = {part})")) + } + ("ENDSWITH", [value, part]) => { + Some(format!("(RIGHT({value}, LENGTH({part})) = {part})")) + } + ("CHARINDEX", [part, value]) => Some(format!("POSITION({part} IN {value})")), + ( + "YEAR" | "QUARTER" | "MONTH" | "DAY" | "DAYOFYEAR" | "HOUR" | "MINUTE" + | "SECOND", + [_], + ) => { + return normalize(extract(&name, args[0].clone())?, target); + } + ("DATE_PART", [part, _]) => { + return normalize(extract(part.trim_matches('\''), args[1].clone())?, target); + } + _ => None, + }; + if let Some(sql) = replacement { + parse_source(&sql) + } else { + Ok(Expression::Function(function)) + } + } + _ => Ok(node), + } +} + +/// Lower the portable expression, preserving exact rather than approximate SQL. +pub fn lower_ossie_sql(sql: &str, target: DialectType) -> Result { + // Polyglot's recursive transformer has large stack frames. Reuse the + // established native semantic boundary, including its WASM-safe fallback. + crate::semantic_input::with_semantic_stack(|| { + lower_ossie_sql_inner(sql, target).map_err(crate::error::SidemanticError::Validation) + }) + .map_err(|error| error.to_string()) +} + +fn lower_ossie_sql_inner(sql: &str, target: DialectType) -> Result { + let expression = parse_portable_expression(sql)?; + let dialect = Dialect::get(target); + // The pinned Polyglot walker omits children of typed aggregates such as + // SUM/AVG. Follow the complete serialized AST, as semantic_input/dates does, + // so portable functions inside aggregates receive the same corrections. + fn rewrite( + value: &mut serde_json::Value, + target: DialectType, + dialect: &Dialect, + ) -> Result<(), String> { + match value { + serde_json::Value::Object(fields) => { + for child in fields.values_mut() { + rewrite(child, target, dialect)?; + } + } + serde_json::Value::Array(values) => { + for child in values { + rewrite(child, target, dialect)?; + } + } + _ => return Ok(()), + } + if let Ok(node) = serde_json::from_value::(value.clone()) { + let normalized = normalize(node, target).map_err(|error| error.to_string())?; + let transformed = dialect + .transform(normalized) + .map_err(|error| error.to_string())?; + *value = serde_json::to_value(transformed).map_err(|error| error.to_string())?; + } + Ok(()) + } + let mut value = serde_json::to_value(expression).map_err(|error| error.to_string())?; + rewrite(&mut value, target, &dialect)?; + let expression = serde_json::from_value(value).map_err(|error| error.to_string())?; + dialect + .generate_with_source(&expression, DialectType::Snowflake) + .map_err(|error| error.to_string()) +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn rejects_queries_and_multiple_expressions() { + for sql in [ + "SELECT x", + "x IN (SELECT x FROM t)", + "x; y", + "x AS y", + "DROP TABLE t", + ] { + assert!(lower_ossie_sql(sql, DialectType::DuckDB).is_err(), "{sql}"); + } + } + + #[test] + fn rejects_forbidden_nodes_inside_typed_aggregate_arguments() { + let mut failures = Vec::new(); + for (sql, reason) in [ + ("SUM((SELECT 1))".to_string(), "cannot contain queries"), + ( + "AVG(COALESCE((SELECT 1), 0))".to_string(), + "cannot contain queries", + ), + ("SUM([1, 2])".to_string(), "cannot contain queries"), + ( + "AVG(COALESCE([], [1]))".to_string(), + "cannot contain queries", + ), + ("SUM(values_array[1])".to_string(), "cannot contain queries"), + ("SUM(a.b.c)".to_string(), "at most two identifiers"), + ("AVG(ABS(a.b.c))".to_string(), "at most two identifiers"), + (format!("SUM(\"{}\")", "x".repeat(129)), "cannot exceed 128"), + ( + format!("AVG(\"{}\".x)", "t".repeat(129)), + "cannot exceed 128", + ), + ] { + match lower_ossie_sql(&sql, DialectType::DuckDB) { + Err(error) if error.contains(reason) => {} + outcome => { + failures.push(format!("{sql}: expected {reason:?}, received {outcome:?}")) + } + } + } + assert!(failures.is_empty(), "{}", failures.join("\n")); + } + + #[test] + fn preserves_valid_qualified_and_quoted_aggregate_arguments() { + for sql in ["SUM(ABS(t.x))", "AVG(COALESCE(t.x, 0))", "SUM(t.\"a.b\")"] { + let lowered = lower_ossie_sql(sql, DialectType::DuckDB).unwrap(); + assert!(lowered.contains("t."), "{sql}: {lowered}"); + } + } + + #[test] + fn required_portable_forms_produce_parseable_sql() { + for sql in [ + "LOG(2, 8)", + "LOG10(100)", + "TRUNC(-12.345, 2)", + "TRUNCATE(123.45, -1)", + "ZEROIFNULL(x)", + "NULLIFZERO(x)", + "DATEADD(day, 7, d)", + "DATEDIFF(day, a, b)", + "DAYOFYEAR(d)", + "DATE_PART('year', d)", + "TO_TIMESTAMP('2024-01-15 10:30:00')", + "TO_DATE('2024-01-15')", + "CONTAINS(s, 'x')", + "STARTSWITH(s, 'x')", + "ENDSWITH(s, 'x')", + "PERCENTILE_CONT(.5) WITHIN GROUP (ORDER BY x)", + "LAG(x, 1, 0) OVER (ORDER BY d)", + "CAST(x AS VARCHAR)", + ] { + for target in [ + DialectType::DuckDB, + DialectType::PostgreSQL, + DialectType::Snowflake, + DialectType::Databricks, + ] { + let lowered = + lower_ossie_sql(sql, target).unwrap_or_else(|error| panic!("{sql}: {error}")); + crate::semantic_input::with_semantic_stack(|| { + crate::semantic_input::dialects::parse(&lowered, target)?; + Ok(()) + }) + .unwrap_or_else(|error| panic!("{lowered}: {error}")); + assert!(!lowered.contains("APPROX"), "{sql}: {lowered}"); + } + } + } + + #[test] + fn keeps_logarithm_order_and_truncation_precision() { + let log = lower_ossie_sql("LOG(2, 8)", DialectType::DuckDB).unwrap(); + assert!(log.contains("LN(8) / LN(2)"), "{log}"); + let trunc = lower_ossie_sql("TRUNC(-12.345, 2)", DialectType::DuckDB).unwrap(); + assert!( + trunc.contains("ROUND(-12.345, 2)") && trunc.contains("0.01"), + "{trunc}" + ); + let timestamp = + lower_ossie_sql("TO_TIMESTAMP('2024-01-15 10:30:00')", DialectType::DuckDB).unwrap(); + assert!( + timestamp.contains("CAST(") && timestamp.contains("AS TIMESTAMP"), + "{timestamp}" + ); + } + + #[test] + fn normalizes_functions_inside_typed_aggregates() { + let sql = lower_ossie_sql( + "SUM(LOG(2, x)) + AVG(DATEDIFF(day, a, b))", + DialectType::DuckDB, + ) + .unwrap(); + assert!(sql.contains("LN(x) / LN(2)"), "{sql}"); + assert!(sql.contains("DATE_DIFF('DAY', a, b)"), "{sql}"); + } + + #[test] + fn normalizes_portable_time_and_timestamp_source_forms() { + for source in ["CURRENT_TIME", "CURRENT_TIME()"] { + assert_eq!( + lower_ossie_sql(source, DialectType::DuckDB).unwrap(), + "LOCALTIME" + ); + } + for source in ["CURRENT_TIMESTAMP", "CURRENT_TIMESTAMP()"] { + assert_eq!( + lower_ossie_sql(source, DialectType::DuckDB).unwrap(), + "CURRENT_TIMESTAMP" + ); + } + for part in ["HOUR", "MINUTE", "SECOND"] { + let source = format!("{part}(TIMESTAMP_NTZ '2024-03-01 12:34:56')"); + let lowered = lower_ossie_sql(&source, DialectType::DuckDB).unwrap(); + let explicit_cast = format!("{part}(CAST('2024-03-01 12:34:56' AS TIMESTAMP_NTZ))"); + assert_eq!( + lowered, + lower_ossie_sql(&explicit_cast, DialectType::DuckDB).unwrap() + ); + } + } + + #[test] + fn typed_timestamp_rewriting_preserves_literals_and_unicode_offsets() { + let source = + "CONCAT('é TIMESTAMP_NTZ', CAST(TIMESTAMP_NTZ '2024-03-01 12:34:56' AS VARCHAR))"; + assert_eq!(prepare_source_literals(source).unwrap(), "CONCAT('é TIMESTAMP_NTZ', CAST(CAST('2024-03-01 12:34:56' AS TIMESTAMP_NTZ) AS VARCHAR))"); + } + + #[test] + fn dateadd_negative_amount_uses_multiplication_not_an_unquoted_interval() { + let lowered = lower_ossie_sql( + "CAST(DATEADD(month, -1, DATE '2024-03-15') AS DATE)", + DialectType::DuckDB, + ) + .unwrap(); + assert!( + lowered.contains("-1") && lowered.contains("* INTERVAL"), + "{lowered}" + ); + assert!(!lowered.contains("INTERVAL -1 MONTH"), "{lowered}"); + } +} diff --git a/sidemantic-rs/src/sql/generator.rs b/sidemantic-rs/src/sql/generator.rs index 83c38e4dc..eee76e391 100644 --- a/sidemantic-rs/src/sql/generator.rs +++ b/sidemantic-rs/src/sql/generator.rs @@ -210,7 +210,15 @@ impl<'a> SqlGenerator<'a> { // Native callers can use this API without the SemanticInput host. Keep // parsing, AST transformation, serialization and destruction on the same // protected stack instead of returning a deep AST to the caller stack. - crate::semantic_input::with_semantic_stack(|| self.generate_with_options(query)) + crate::semantic_input::with_semantic_stack(|| { + if let Some(graph) = self.prepare_imported_window_dimensions()? { + return SqlGenerator::new(&graph) + .with_dialect(self.dialect) + .with_timezone(self.timezone.clone()) + .generate_with_options(query); + } + self.generate_with_options(query) + }) } /// Aggregate children retain their own source population independently of @@ -219,6 +227,15 @@ impl<'a> SqlGenerator<'a> { &self, query: &SemanticQuery, source_model: Option<&str>, + ) -> Result { + self.generate_from_model_with_aggregation(query, source_model, true) + } + + fn generate_from_model_with_aggregation( + &self, + query: &SemanticQuery, + source_model: Option<&str>, + plan_aggregates: bool, ) -> Result { if query.consumption_base_model.is_some() && query.metrics.is_empty() @@ -229,8 +246,10 @@ impl<'a> SqlGenerator<'a> { )); } self.validate_approximate_query(query)?; - if let Some(sql) = aggregate_plan::try_generate(self, query)? { - return Ok(sql); + if plan_aggregates { + if let Some(sql) = aggregate_plan::try_generate(self, query)? { + return Ok(sql); + } } if let Some(sql) = snapshots::try_generate(self, query)? { return Ok(sql); @@ -284,7 +303,7 @@ impl<'a> SqlGenerator<'a> { ); } - if self.needs_preaggregation_for_fanout(&metric_refs)? { + if plan_aggregates && self.needs_preaggregation_for_fanout(&metric_refs)? { self.reject_totals_route(query, "preaggregation")?; return self.generate_with_preaggregation( query, @@ -438,6 +457,10 @@ impl<'a> SqlGenerator<'a> { let metric = self.metric_for_model_with_source(&model_name, &metric_name, graph_metric)?; let raw_alias = self.metric_raw_alias(model, &metric_name, metric); + let bound_metric = graph_metric + .then(|| self.graph_metric_source_inputs(metric, Some(&model_name))) + .transpose()?; + let metric = bound_metric.as_ref().unwrap_or(metric); let mut raw_expr = self.normalize_cte_source_expression( &self.metric_raw_expression(metric, model)?, model, @@ -1462,6 +1485,9 @@ impl<'a> SqlGenerator<'a> { } } } + if owners.is_empty() { + owners.extend(self.logical_constant_owner(metric)?); + } if owners.is_empty() { return Err(SidemanticError::UnsupportedSemanticFeatures { capabilities: vec![format!("metric.graph_scope.{reference}")], @@ -1475,6 +1501,25 @@ impl<'a> SqlGenerator<'a> { Ok(owners) } + /// A column-free Ossie aggregate still has an unambiguous population when + /// its scope declares exactly one dataset. Never guess among several. + fn logical_constant_owner(&self, metric: &Metric) -> Result> { + if !metric.sql_is_complete || !aggregate_plan::has_logical_inputs(metric) { + return Ok(None); + } + let Some(sql) = metric.sql.as_deref() else { + return Ok(None); + }; + if !semantic_column_references(sql)?.is_empty() { + return Ok(None); + } + let mut models = self.graph.models(); + let first = models.next(); + Ok(first + .filter(|_| models.next().is_none()) + .map(|model| model.name.clone())) + } + fn metric_reference_tokens(&self, expression: &str) -> Result> { if self.graph.has_strict_metric_scope() { let expression = crate::core::replace_model_placeholder(expression, None)?; @@ -1810,6 +1855,57 @@ impl<'a> SqlGenerator<'a> { ordered } + /// Unowned graph measures reference semantic fields. Model-local measures + /// and graph-addressable model measures retain physical inputs, including + /// complete filtered SQL already lowered to a simple owned measure. + fn graph_metric_source_inputs(&self, metric: &Metric, owner: Option<&str>) -> Result { + let mut bound = metric.clone(); + if metric.r#type != MetricType::Simple + || metric.sql_is_complete + || self.graph.metric_owner(&metric.name).is_some() + { + return Ok(bound); + } + let Some(model) = owner.and_then(|owner| self.graph.get_model(owner)) else { + return Ok(bound); + }; + let bind = |sql: &str| -> Result { + let sql = crate::core::replace_model_placeholder(sql, Some(&model.name))?; + let mut replacements = HashMap::new(); + for column in crate::core::outer_semantic_column_references(&sql)? { + if column + .model + .as_deref() + .is_some_and(|source| source != model.name) + { + continue; + } + if let Some(dimension) = model.get_dimension(&column.field) { + let source = + self.normalize_cte_source_expression(dimension.sql_expr(), model)?; + replacements.insert((column.model, column.field), format!("({source})")); + } + } + self.emit_expression(&crate::core::replace_outer_semantic_columns( + parse_semantic_expression(&sql)?, + &replacements, + )?) + }; + if let Some(sql) = metric + .sql + .as_deref() + .filter(|sql| !sql.is_empty() && *sql != "*") + { + bound.sql = Some(bind(sql)?); + } + bound.filters = metric + .filters + .iter() + .map(|filter| bind(filter)) + .collect::>()?; + Ok(bound) + } + fn metric_raw_expression( &self, metric: &crate::core::Metric, diff --git a/sidemantic-rs/src/sql/generator/aggregate_plan.rs b/sidemantic-rs/src/sql/generator/aggregate_plan.rs index 82d19e421..76f259388 100644 --- a/sidemantic-rs/src/sql/generator/aggregate_plan.rs +++ b/sidemantic-rs/src/sql/generator/aggregate_plan.rs @@ -16,21 +16,36 @@ struct ResolvedMetric { #[derive(Clone)] struct Leaf { reference: String, - model: String, metric: Metric, alias: String, + population: String, +} + +/// Aggregate inputs share a child only when they have the same row grain. +/// A joined-row expression is distinct from either source's independent total. +struct Population { + name: String, + sources: Vec, } struct Plan<'a, 'g> { generator: &'a SqlGenerator<'g>, leaves: Vec, models: Vec, + populations: Vec, expressions: HashMap, active: HashSet, cross_source_calculation: bool, inline_aggregates: bool, } +pub(super) fn has_logical_inputs(metric: &Metric) -> bool { + metric + .metadata + .as_ref() + .is_some_and(|metadata| metadata.get("ossie_expression_dialect").is_some()) +} + fn unsupported(capability: &str) -> SidemanticError { SidemanticError::UnsupportedSemanticFeatures { capabilities: vec![format!("aggregation.{capability}")], @@ -50,13 +65,63 @@ fn count_aggregate(expression: &Expression) -> bool { } impl<'a, 'g> Plan<'a, 'g> { + fn population(&mut self, mut sources: Vec) -> String { + sources.sort(); + // Start joined rows at the finest declared grain. A many-to-one + // lookup must not discard unmatched fact rows by starting at its target. + if let Some(index) = sources.iter().position(|source| { + sources + .iter() + .filter(|target| *target != source) + .all(|target| { + self.generator + .graph + .find_join_path(source, target) + .is_ok_and(|path| !path.has_fan_out()) + }) + }) { + sources.swap(0, index); + } + if let Some(population) = self.populations.iter().find(|item| item.sources == sources) { + return population.name.clone(); + } + for source in &sources { + if !self.models.contains(source) { + self.models.push(source.clone()); + } + } + let name = if sources.len() == 1 { + sources[0].clone() + } else { + let mut index = self.populations.len(); + loop { + let candidate = format!("__sidemantic_joined_{index}"); + if self.generator.graph.get_model(&candidate).is_none() + && !self + .populations + .iter() + .any(|population| population.name == candidate) + { + break candidate; + } + index += 1; + } + }; + self.populations.push(Population { + name: name.clone(), + sources, + }); + name + } + /// Split authored aggregate calls before expanding scalar metric references. - /// Each call must read one source; arithmetic combines its grouped output. + /// Each call owns a source population; arithmetic combines grouped outputs. fn expand_calculation( &mut self, node: &mut serde_json::Value, context: Option<&str>, bindings: &HashSet, + logical_inputs: bool, ) -> Result<()> { let aggregate_node = crate::core::is_aggregate_ast_node(node); if let serde_json::Value::Object(fields) = node { @@ -71,7 +136,12 @@ impl<'a, 'g> Plan<'a, 'g> { .filter_map(|parameter| parameter["name"].as_str().map(str::to_owned)), ); } - return self.expand_calculation(&mut lambda["body"], context, &bindings); + return self.expand_calculation( + &mut lambda["body"], + context, + &bindings, + logical_inputs, + ); } if matches!( kind, @@ -107,25 +177,30 @@ impl<'a, 'g> Plan<'a, 'g> { if owners.is_empty() { owners.extend(context.map(str::to_string)); } - if owners.len() != 1 { + if owners.is_empty() || (owners.len() > 1 && !logical_inputs) { return Err(unsupported("cross_source_raw_input")); } - let model = owners.into_iter().next().unwrap(); - if !self.models.contains(&model) { - self.models.push(model.clone()); - } + let mut owners: Vec<_> = owners.into_iter().collect(); + owners.sort(); + let model = owners[0].clone(); + let population = self.population(owners); let alias = format!("__sidemantic_metric_{}", self.leaves.len()); let mut metric = Metric::derived(&alias, sql); metric.sql_is_complete = true; + if logical_inputs { + metric.metadata = + Some(serde_json::json!({"ossie_expression_dialect": "ANSI_SQL"})); + } self.leaves.push(Leaf { reference: format!("{model}.{alias}"), - model: model.clone(), metric, alias: alias.clone(), + population: population.clone(), }); let output = format!( "{}.{}", - self.generator.quote_identifier(&format!("{model}_preagg")), + self.generator + .quote_identifier(&format!("{population}_preagg")), self.generator.quote_identifier(&alias), ); if count_aggregate(&expression) { @@ -149,12 +224,12 @@ impl<'a, 'g> Plan<'a, 'g> { match node { serde_json::Value::Object(fields) => { for child in fields.values_mut() { - self.expand_calculation(child, context, bindings)?; + self.expand_calculation(child, context, bindings, logical_inputs)?; } } serde_json::Value::Array(children) => { for child in children { - self.expand_calculation(child, context, bindings)?; + self.expand_calculation(child, context, bindings, logical_inputs)?; } } _ => {} @@ -166,6 +241,9 @@ impl<'a, 'g> Plan<'a, 'g> { if let Some(owner) = self.generator.graph.metric_owner(reference) { return Ok(Some(owner.to_string())); } + if let Some(owner) = self.generator.logical_constant_owner(metric)? { + return Ok(Some(owner)); + } // Imported graph aggregates can bind their source in qualified SQL // without an explicit owner annotation. Scalar calculations stay unowned. if metric.agg.is_some() && !metric.sql_is_complete { @@ -183,10 +261,13 @@ impl<'a, 'g> Plan<'a, 'g> { let graph = self.generator.graph; if let Some((model_name, name)) = reference.split_once('.') { if let Some(metric) = graph.get_metric(reference) { + let context = self.graph_metric_context(reference, metric)?; return Ok(Some(ResolvedMetric { reference: reference.to_string(), - context: self.graph_metric_context(reference, metric)?, - metric: metric.clone(), + metric: self + .generator + .graph_metric_source_inputs(metric, context.as_deref())?, + context, })); } return Ok(graph.get_model(model_name).and_then(|model| { @@ -210,10 +291,13 @@ impl<'a, 'g> Plan<'a, 'g> { } } if let Some(metric) = graph.get_metric(reference) { + let context = self.graph_metric_context(reference, metric)?; return Ok(Some(ResolvedMetric { reference: reference.to_string(), - context: self.graph_metric_context(reference, metric)?, - metric: metric.clone(), + metric: self + .generator + .graph_metric_source_inputs(metric, context.as_deref())?, + context, })); } let owners: Vec<_> = graph @@ -292,15 +376,13 @@ impl<'a, 'g> Plan<'a, 'g> { return Err(unsupported("cross_source_raw_input")); } } - if !self.models.contains(&model) { - self.models.push(model.clone()); - } + let population = self.population(vec![model.clone()]); let alias = format!("__sidemantic_metric_{}", self.leaves.len()); self.leaves.push(Leaf { reference: resolved.reference.clone(), - model: model.clone(), metric: metric.clone(), alias: alias.clone(), + population, }); let output = format!( "{}.{}", @@ -350,7 +432,12 @@ impl<'a, 'g> Plan<'a, 'g> { let sql = crate::core::replace_model_placeholder(sql, resolved.context.as_deref())?; let mut ast = serde_json::to_value(parse_semantic_expression(&sql)?) .map_err(|error| SidemanticError::SqlGeneration(error.to_string()))?; - self.expand_calculation(&mut ast, resolved.context.as_deref(), &HashSet::new())?; + self.expand_calculation( + &mut ast, + resolved.context.as_deref(), + &HashSet::new(), + has_logical_inputs(metric), + )?; let expression = serde_json::from_value(ast) .map_err(|error| SidemanticError::SqlGeneration(error.to_string()))?; self.generator.emit_expression(&expression)? @@ -481,6 +568,7 @@ pub(super) fn try_generate( generator, leaves: Vec::new(), models: Vec::new(), + populations: Vec::new(), expressions: HashMap::new(), active: HashSet::new(), cross_source_calculation: false, @@ -534,7 +622,13 @@ pub(super) fn try_generate( // Inline splitting exists to combine independent sources. The ordinary // single-source compiler binds semantic dimension inputs and owns authored // aggregate/window expressions without manufacturing physical columns. - if plan.inline_aggregates && plan.models.len() < 2 { + if plan.inline_aggregates + && plan.models.len() < 2 + && !plan + .leaves + .iter() + .any(|leaf| has_logical_inputs(&leaf.metric)) + { return Ok(None); } // Cartesian products require every participating source even in a child @@ -628,12 +722,12 @@ pub(super) fn try_generate( .enumerate() .map(|(index, (reference, _))| { let columns = plan - .models + .populations .iter() - .map(|model| { + .map(|population| { format!( "{}.{}", - generator.quote_identifier(&format!("{model}_preagg")), + generator.quote_identifier(&format!("{}_preagg", population.name)), dimension_alias(index) ) }) @@ -764,14 +858,18 @@ pub(super) fn try_generate( } } let mut ctes = Vec::new(); - for model in &plan.models { + for population in &plan.populations { + let model = &population.sources[0]; let leaves: Vec<_> = plan .leaves .iter() - .filter(|leaf| &leaf.model == model) + .filter(|leaf| leaf.population == population.name) .collect(); let mut child = query.clone(); child.required_population_models = population_models.clone(); + child + .required_population_models + .extend(population.sources.iter().cloned()); child.metrics = leaves.iter().map(|leaf| leaf.reference.clone()).collect(); child.filters = row_filters.clone(); child @@ -795,11 +893,11 @@ pub(super) fn try_generate( super::fanout_complete::generate_entity_aggregates( generator, &child, - model, &dimensions, &metrics, fanout_models.contains(model), plan.models.len() > 1, + &population.sources, )? } else { let mut projection = child_projection(generator, &dimensions, &leaves)?; @@ -822,7 +920,7 @@ pub(super) fn try_generate( }; ctes.push(format!( "{} AS (\n{child_sql}\n)", - generator.quote_identifier(&format!("{model}_preagg")), + generator.quote_identifier(&format!("{}_preagg", population.name)), )); } let mut names = HashMap::new(); @@ -875,12 +973,12 @@ pub(super) fn try_generate( } if query.with_totals && !dimensions.is_empty() { let markers: Vec<_> = plan - .models + .populations .iter() - .map(|model| { + .map(|population| { format!( "{}._is_total", - generator.quote_identifier(&format!("{model}_preagg")) + generator.quote_identifier(&format!("{}_preagg", population.name)) ) }) .collect(); @@ -895,10 +993,10 @@ pub(super) fn try_generate( "WITH {}\nSELECT {}\nFROM {}", ctes.join(",\n"), selections.join(",\n"), - generator.quote_identifier(&format!("{}_preagg", plan.models[0])) + generator.quote_identifier(&format!("{}_preagg", plan.populations[0].name)) ); - for (index, model) in plan.models.iter().enumerate().skip(1) { - let table = generator.quote_identifier(&format!("{model}_preagg")); + for (index, population) in plan.populations.iter().enumerate().skip(1) { + let table = generator.quote_identifier(&format!("{}_preagg", population.name)); if dimensions.is_empty() { sql.push_str(&format!("\nCROSS JOIN {table}")); } else { @@ -907,12 +1005,12 @@ pub(super) fn try_generate( .enumerate() .map(|(dimension_index, _)| { let name = dimension_alias(dimension_index); - let previous: Vec<_> = plan.models[..index] + let previous: Vec<_> = plan.populations[..index] .iter() - .map(|model| { + .map(|population| { format!( "{}.{name}", - generator.quote_identifier(&format!("{model}_preagg")) + generator.quote_identifier(&format!("{}_preagg", population.name)) ) }) .collect(); @@ -925,12 +1023,12 @@ pub(super) fn try_generate( }) .collect::>(); if query.with_totals { - let previous: Vec<_> = plan.models[..index] + let previous: Vec<_> = plan.populations[..index] .iter() - .map(|model| { + .map(|population| { format!( "{}._is_total", - generator.quote_identifier(&format!("{model}_preagg")) + generator.quote_identifier(&format!("{}_preagg", population.name)) ) }) .collect(); @@ -1188,6 +1286,7 @@ mod tests { generator: &generator, leaves: Vec::new(), models: Vec::new(), + populations: Vec::new(), expressions: HashMap::new(), active: HashSet::new(), cross_source_calculation: false, diff --git a/sidemantic-rs/src/sql/generator/fanout_complete.rs b/sidemantic-rs/src/sql/generator/fanout_complete.rs index 24fcb4b5b..c1bec9fcc 100644 --- a/sidemantic-rs/src/sql/generator/fanout_complete.rs +++ b/sidemantic-rs/src/sql/generator/fanout_complete.rs @@ -14,15 +14,14 @@ fn unsupported(shape: &str) -> SidemanticError { struct Inputs<'a> { generator: &'a SqlGenerator<'a>, - owner: &'a str, - model: Model, + models: HashMap, names: HashSet, references: Vec, next: usize, } impl Inputs<'_> { - fn add(&mut self, sql: String, filters: &[String]) -> String { + fn add(&mut self, owner: &str, sql: String, filters: &[String]) -> String { let name = loop { let candidate = format!("sidemantic_input_{}", self.next); self.next += 1; @@ -32,8 +31,8 @@ impl Inputs<'_> { }; let mut metric = Metric::sum(&name, sql); metric.filters = filters.to_vec(); - self.model.metrics.push(metric); - self.references.push(format!("{}.{}", self.owner, name)); + self.models.get_mut(owner).unwrap().metrics.push(metric); + self.references.push(format!("{owner}.{name}")); self.generator.quote_identifier(&name) } } @@ -41,18 +40,34 @@ impl Inputs<'_> { pub(super) fn generate_entity_aggregates( generator: &SqlGenerator<'_>, query: &SemanticQuery, - owner: &str, dimensions: &[DimensionRef], metrics: &[(&Metric, &str)], deduplicate: bool, independent_source: bool, + sources: &[String], ) -> Result { + let owner = sources[0].as_str(); let model = generator.graph.get_model(owner).unwrap(); - if deduplicate && model.primary_keys().is_empty() { - return Err(SidemanticError::Validation(format!( - "Model '{owner}' has no primary key; cannot safely aggregate across a fanout join" - ))); - } + // Joined expressions own a tuple of source rows. Include every source key + // when another relationship multiplies that tuple, rather than restoring + // only one source's grain or deduplicating equal measure values. + let mut required = query.required_population_models.clone(); + required.extend(sources.iter().cloned()); + required.extend(dimensions.iter().map(|dimension| dimension.model.clone())); + required.extend(generator.find_filter_models(&query.filters)); + required.extend(query.prepared_policies.model_names().cloned()); + let anchor = query.consumption_base_model.as_deref().unwrap_or(owner); + let paths = generator.build_join_paths(anchor, &required)?; + let source_paths = generator.build_join_paths(anchor, &sources.iter().cloned().collect())?; + let population_edges: HashSet<_> = source_paths + .values() + .flat_map(|path| &path.steps) + .map(|step| (&step.from_model, &step.to_model)) + .collect(); + let extra_fanout = paths.values().flat_map(|path| &path.steps).any(|step| { + step.causes_fan_out() && !population_edges.contains(&(&step.from_model, &step.to_model)) + }); + let deduplicate = deduplicate || (sources.len() > 1 && extra_fanout); let names = generator .graph .models() @@ -72,15 +87,30 @@ pub(super) fn generate_entity_aggregates( .collect(); let mut inputs = Inputs { generator, - owner, - model: model.clone(), + models: sources + .iter() + .map(|source| { + ( + source.clone(), + generator.graph.get_model(source).unwrap().clone(), + ) + }) + .collect(), names, references: Vec::new(), next: 0, }; if deduplicate { - for key in model.primary_keys() { - inputs.add(generator.key_sql(model, &key, None)?, &[]); + for source in sources { + let source_model = generator.graph.get_model(source).unwrap(); + if source_model.primary_keys().is_empty() { + return Err(SidemanticError::Validation(format!( + "Model '{source}' has no primary key; cannot safely aggregate across a fanout join" + ))); + } + for key in source_model.primary_keys() { + inputs.add(source, generator.key_sql(source_model, &key, None)?, &[]); + } } } let mut selections = Vec::new(); @@ -97,19 +127,31 @@ pub(super) fn generate_entity_aggregates( } let mut replacements = HashMap::new(); for column in columns { - if column - .model - .as_deref() - .is_some_and(|name| name != owner && name != format!("{owner}_cte")) - { + let source = column.model.as_deref().unwrap_or(owner); + let source = source + .strip_suffix("_cte") + .filter(|source| sources.iter().any(|item| item == source)) + .unwrap_or(source); + if !sources.iter().any(|item| item == source) { return Err(unsupported("cross_source_raw_input")); } - let key = (column.model, column.field.clone()); + let key = (column.model.clone(), column.field.clone()); if let std::collections::hash_map::Entry::Vacant(entry) = replacements.entry(key) { - // Complete SQL names physical source columns. Do not expand - // a coincidentally named semantic metric as a dependency. - let input = - inputs.add(generator.quote_identifier(&column.field), &metric.filters); + // Ossie aggregate inputs prefer declared logical fields and + // preserve undeclared physical references. Native complete + // SQL retains its physical source-column contract. Project + // either expression under a fresh internal metric name, never + // beside SELECT * under a colliding source-column name. + let source_model = generator.graph.get_model(source).unwrap(); + let raw = if super::aggregate_plan::has_logical_inputs(metric) { + source_model + .get_dimension(&column.field) + .map(|dimension| dimension.sql_expr().to_string()) + .unwrap_or_else(|| generator.quote_identifier(&column.field)) + } else { + generator.quote_identifier(&column.field) + }; + let input = inputs.add(source, raw, &metric.filters); entry.insert(input); } } @@ -132,7 +174,7 @@ pub(super) fn generate_entity_aggregates( } else { generator.metric_raw_expression(metric, model)? }; - let raw = inputs.add(raw, &metric.filters); + let raw = inputs.add(owner, raw, &metric.filters); if implicit_distinct { format!("COUNT({raw})") } else { @@ -152,10 +194,12 @@ pub(super) fn generate_entity_aggregates( // COUNT(*) can be the only output. Retain its source in the row query even // when every grouping dimension belongs to a different model. if inputs.references.is_empty() { - inputs.add("1".into(), &[]); + inputs.add(owner, "1".into(), &[]); } let mut graph = generator.graph.clone(); - graph.replace_model(inputs.model)?; + for model in inputs.models.into_values() { + graph.replace_model(model)?; + } let row_generator = SqlGenerator::new(&graph) .with_dialect(generator.dialect) .with_timezone(generator.timezone.clone()); @@ -177,7 +221,8 @@ pub(super) fn generate_entity_aggregates( } else { row_generator.query_base_model(dimensions, &row_generator.parse_metric_refs(&rows.metrics)?) }; - let row_sql = row_generator.generate_from_model(&rows, source.as_deref())?; + let row_sql = + row_generator.generate_from_model_with_aggregation(&rows, source.as_deref(), false)?; let mut collisions = HashMap::new(); for dimension in dimensions { *collisions.entry(dimension.alias.clone()).or_insert(0usize) += 1; diff --git a/sidemantic-rs/src/sql/generator/temporal.rs b/sidemantic-rs/src/sql/generator/temporal.rs index 0e7fe13eb..a17766fdd 100644 --- a/sidemantic-rs/src/sql/generator/temporal.rs +++ b/sidemantic-rs/src/sql/generator/temporal.rs @@ -168,6 +168,54 @@ fn period_interval(value: &str) -> Result<(u32, String)> { } impl SqlGenerator<'_> { + /// Ossie fields can carry window SQL in the ordinary expression slot. + /// Materialize it before grouping or filtering, just like native window + /// dimensions, while leaving the caller's semantic graph untouched. + pub(super) fn prepare_imported_window_dimensions(&self) -> Result> { + fn contains_window(node: &serde_json::Value) -> bool { + match node { + serde_json::Value::Object(fields) => { + fields.contains_key("window") + || fields.contains_key("window_function") + || fields.values().any(contains_window) + } + serde_json::Value::Array(values) => values.iter().any(contains_window), + _ => false, + } + } + let mut graph = None; + for model in self.graph.models() { + if model + .metadata + .as_ref() + .is_none_or(|metadata| metadata.get("ossie_source_kind").is_none()) + { + continue; + } + let mut updated = None; + for (index, dimension) in model.dimensions.iter().enumerate() { + if dimension.window.is_some() { + continue; + } + let Some(sql) = dimension.sql.as_deref() else { + continue; + }; + let expression = serde_json::to_value(parse_semantic_expression(sql)?) + .map_err(|error| SidemanticError::SqlGeneration(error.to_string()))?; + if contains_window(&expression) { + let model = updated.get_or_insert_with(|| model.clone()); + model.dimensions[index].window = Some(sql.to_string()); + } + } + if let Some(model) = updated { + graph + .get_or_insert_with(|| self.graph.clone()) + .replace_model(model)?; + } + } + Ok(graph) + } + pub(super) fn window_dimension_alias(dimension: &crate::core::Dimension) -> String { format!("__sidemantic_window_{}", dimension.name) } diff --git a/tests/adapters/osi/test_rust_ossie_forward_parity.py b/tests/adapters/osi/test_rust_ossie_forward_parity.py index f53788bf1..b5de0a189 100644 --- a/tests/adapters/osi/test_rust_ossie_forward_parity.py +++ b/tests/adapters/osi/test_rust_ossie_forward_parity.py @@ -97,8 +97,13 @@ def test_selected_scope_shape_preserves_parity_fields_and_target_selection(tmp_p assert rust_dimension["declared_is_time"] is python_dimension.declared_is_time is False assert "declared_is_time" not in rust["models"][0]["dimensions"][1] assert rust["metrics"][0]["logical_data_type"] == "Decimal" - assert rust["metrics"][0]["agg"] == "sum" - assert rust["metrics"][0]["sql"] == "SAFE_CAST(orders.amount AS NUMERIC)" + assert rust["metrics"][0].get("agg") is None + assert rust["metrics"][0]["sql_is_complete"] + assert ( + rust["metrics"][0]["sql"] + == python_graph.get_metric("gross_amount").sql + == "SUM(SAFE_CAST(orders.amount AS NUMERIC))" + ) assert rust["models"][0]["relationships"][0]["edge_id"] == "orders_customer" assert python_graph.get_model("orders").relationships[0].edge_id == "orders_customer" assert rust["models"][0]["primary_key"] == "" diff --git a/tests/semantic_conformance/test_ossie_aggregate_inputs.py b/tests/semantic_conformance/test_ossie_aggregate_inputs.py new file mode 100644 index 000000000..3519812f6 --- /dev/null +++ b/tests/semantic_conformance/test_ossie_aggregate_inputs.py @@ -0,0 +1,284 @@ +"""Ossie logical fields and joined-row aggregate grains in the Rust runtime.""" + +import json + +import pytest + +from sidemantic import Dimension, Metric, Model, Relationship, SemanticLayer +from sidemantic.adapters.ossie import OssieAdapter +from sidemantic.rust_bridge import rewrite_semantic_input + + +def expression(sql): + return {"dialects": [{"dialect": "ANSI_SQL", "expression": sql}]} + + +def field(name, sql=None): + return {"name": name, "expression": expression(sql or name)} + + +def load_layer(tmp_path, datasets, metrics, relationships=()): + pytest.importorskip("sidemantic_rs", reason="Requires the source-built Rust runtime") + source = tmp_path / "aggregates.ossie.json" + source.write_text( + json.dumps( + { + "version": "0.2.0.dev0", + "semantic_model": [ + { + "name": "commerce", + "datasets": datasets, + "metrics": [{"name": name, "expression": expression(sql)} for name, sql in metrics.items()], + "relationships": list(relationships), + } + ], + } + ) + ) + layer = SemanticLayer(engine="rust", fallback=False, auto_register=False) + layer.graph = OssieAdapter().parse(source) + return layer + + +@pytest.mark.parametrize("name,reference", [("amount", "amount"), ("Amount", "amount"), ("gross", "gross")]) +@pytest.mark.parametrize("route", ["query", "rewrite"]) +def test_computed_input_cannot_be_shadowed_by_physical_column(tmp_path, name, reference, route): + layer = load_layer( + tmp_path, + [{"name": "orders", "source": "orders", "fields": [field(name, "amount * 2")]}], + {"total": f"SUM(orders.{reference})"}, + ) + try: + layer.adapter.execute("create table orders(amount int); insert into orders values (10),(20)") + if route == "query": + assert layer.query(metrics=["total"]).fetchall() == [(60,)] + assert layer.last_engine_selection["engine"] == "rust" + else: + sql = rewrite_semantic_input(layer.graph, "select total from metrics") + assert layer.adapter.execute(sql).fetchall() == [(60,)] + finally: + layer.adapter.close() + + +@pytest.mark.parametrize("dimensions", [[], ["orders.population"]]) +def test_column_free_aggregate_uses_sole_dataset_population(tmp_path, dimensions): + layer = load_layer( + tmp_path, + [{"name": "orders", "source": "orders", "fields": [field("population", "1")]}], + {"rows": "COUNT(*)"}, + ) + try: + layer.adapter.execute("create table orders(amount int); insert into orders values (10),(20),(null)") + expected = [(1, 3)] if dimensions else [(3,)] + assert layer.query(metrics=["rows"], dimensions=dimensions).fetchall() == expected + finally: + layer.adapter.close() + + +def test_imported_window_field_is_materialized_before_grouping_and_filtering(tmp_path): + layer = load_layer( + tmp_path, + [ + { + "name": "orders", + "source": "orders", + "fields": [field("id"), field("rank", "ROW_NUMBER() OVER (ORDER BY amount)")], + } + ], + {}, + ) + try: + layer.adapter.execute("create table orders(id int, amount int); insert into orders values (1,30),(2,10),(3,20)") + assert layer.query(dimensions=["orders.id", "orders.rank"], order_by=["orders.id"]).fetchall() == [ + (1, 3), + (2, 1), + (3, 2), + ] + assert layer.query( + dimensions=["orders.id", "orders.rank"], filters=["orders.rank <= 2"], order_by=["orders.id"] + ).fetchall() == [(2, 1), (3, 2)] + finally: + layer.adapter.close() + + +@pytest.fixture +def joined(tmp_path): + layer = load_layer( + tmp_path, + [ + { + "name": "orders", + "source": "orders", + "primary_key": ["id"], + "fields": [field("id"), field("customer_ref", "customer_id + 1"), field("amount")], + }, + { + "name": "customers", + "source": "customers", + "primary_key": ["id"], + "fields": [field("id"), field("name"), field("budget")], + }, + ], + { + "weighted": "SUM(orders.amount * customers.budget)", + "revenue": "SUM(orders.amount)", + "budget": "SUM(customers.budget)", + "customers": "COUNT(customers.id)", + "average": "AVG(customers.budget)", + "ratio": "SUM(orders.amount) / SUM(customers.budget)", + "combined": "weighted + budget", + }, + [ + { + "name": "customer", + "from": "orders", + "to": "customers", + "from_columns": ["customer_ref"], + "to_columns": ["id"], + } + ], + ) + layer.adapter.execute(""" + create table orders(id int, customer_id int, amount int); + insert into orders values (1,0,10),(2,0,20),(3,1,30); + create table customers(id int, name varchar, budget int); + insert into customers values (1,'A',100),(2,'B',200); + """) + try: + yield layer + finally: + layer.adapter.close() + + +def test_joined_row_aggregate_and_independent_aggregates_keep_separate_grains(joined): + assert joined.query(metrics=["weighted"]).fetchall() == [(9000,)] + assert joined.query( + metrics=["weighted", "revenue", "budget", "customers", "average", "ratio", "combined"] + ).fetchall() == [(9000, 60, 300, 2, 150, 0.2, 9300)] + assert joined.last_engine_selection["engine"] == "rust" + + +def test_joined_row_aggregate_groups_and_rewrites(joined): + sql = rewrite_semantic_input( + joined.graph, "select customers.name, weighted, budget, ratio from metrics order by customers.name" + ) + assert joined.adapter.execute(sql).fetchall() == [("A", 3000, 100, 0.3), ("B", 6000, 200, 0.15)] + + +def test_joined_rows_preserve_unmatched_facts_and_equal_value_entities(joined): + joined.adapter.execute("insert into orders values (4,0,10),(5,999,50)") + assert joined.query(metrics=["weighted", "revenue", "budget"]).fetchall() == [(10000, 120, 300)] + + +def test_joined_row_aggregate_does_not_require_unused_entity_keys(joined): + joined.graph.models["orders"].primary_key = None + assert joined.query(metrics=["weighted"]).fetchall() == [(9000,)] + + +def test_joined_tuple_grain_survives_an_additional_fanout(joined): + joined.add_model( + Model( + name="tags", + table="tags", + primary_key="id", + dimensions=[Dimension(name="label", type="categorical")], + relationships=[Relationship(name="orders", type="many_to_one", foreign_key="order_id")], + ) + ) + joined.adapter.execute(""" + create table tags(id int, order_id int, label varchar); + insert into tags values (1,1,'x'),(2,1,'x'),(3,2,'x'),(4,3,'x'); + """) + assert joined.query(metrics=["weighted", "budget"], dimensions=["tags.label"]).fetchall() == [("x", 9000, 300)] + + +def test_native_complete_sql_keeps_physical_input_semantics(joined): + joined.graph.models["orders"].get_dimension("amount").sql = "amount * 10" + joined.graph.models["orders"].metrics.append(Metric(name="physical", sql="SUM(amount)", sql_is_complete=True)) + assert joined.query(metrics=["orders.physical", "weighted"]).fetchall() == [(60, 90000)] + + +@pytest.mark.parametrize("plan", ["ordinary", "independent", "fanout"]) +@pytest.mark.parametrize("filtered", [False, True]) +@pytest.mark.parametrize("logical_name", ["amount", "gross"]) +@pytest.mark.parametrize("model_owned", [False, True]) +def test_graph_measures_preserve_input_scope(plan, filtered, logical_name, model_owned): + pytest.importorskip("sidemantic_rs") + layer = SemanticLayer(engine="rust", fallback=False, auto_register=False) + layer.add_model( + Model( + name="orders", + sql="SELECT * FROM (VALUES (1, 10, 1), (2, 5, 1), (3, NULL, 1)) AS t(id, amount, customer_id)", + primary_key="id", + dimensions=[Dimension(name=logical_name, type="numeric", sql="orders.amount * 2")], + metrics=[Metric(name="physical", agg="sum", sql="amount", filters=["amount > 7"] if filtered else None)], + ) + ) + input_name = "amount" if model_owned else logical_name + layer.graph.add_metric( + Metric( + name="value", + agg="sum", + sql=f"orders.{input_name}", + filters=[f"orders.{input_name} > 7"] if filtered else None, + ), + model_name="orders" if model_owned else None, + ) + metrics = ["value", "orders.physical"] + dimensions = [] + physical_value = 10 if filtered else 15 + expected = (physical_value if model_owned else 30, physical_value) + if plan == "independent": + layer.add_model( + Model( + name="customers", + sql="SELECT 1 AS id, 100 AS budget", + primary_key="id", + metrics=[Metric(name="budget", agg="sum", sql="budget")], + ) + ) + layer.graph.models["orders"].relationships.append( + Relationship(name="customers", type="many_to_one", foreign_key="customer_id") + ) + metrics.append("customers.budget") + expected += (100,) + elif plan == "fanout": + layer.add_model( + Model( + name="tags", + sql="SELECT * FROM (VALUES (1, 1), (2, 1), (3, 2), (4, 3)) AS t(id, order_id)", + primary_key="id", + dimensions=[Dimension(name="label", type="categorical", sql="'all'")], + relationships=[Relationship(name="orders", type="many_to_one", foreign_key="order_id")], + ) + ) + dimensions = ["tags.label"] + expected = ("all", *expected) + try: + assert layer.query(metrics=metrics, dimensions=dimensions).fetchall() == [expected] + assert layer.last_engine_selection["engine"] == "rust" + finally: + layer.adapter.close() + + +def test_joined_aggregate_preserves_compound_computed_relationship_keys(joined): + orders = joined.graph.models["orders"] + customers = joined.graph.models["customers"] + orders.dimensions.append(Dimension(name="tenant_ref", type="numeric", sql="tenant + 10")) + customers.dimensions.append(Dimension(name="tenant_key", type="numeric", sql="tenant + 10")) + customers.primary_key = ["tenant_key", "id"] + orders.relationships = [ + Relationship( + name="customers", + type="many_to_one", + foreign_key=["tenant_ref", "customer_ref"], + primary_key=["tenant_key", "id"], + ) + ] + joined.adapter.execute(""" + alter table customers add column tenant int default 1; + alter table orders add column tenant int default 1; + insert into customers values (1,'C',500,2); + insert into orders values (4,0,10,2); + """) + assert joined.query(metrics=["weighted", "budget", "ratio"]).fetchall() == [(14000, 800, 0.0875)]