diff --git a/crates/devtools/src/planner_fuzz/relations.rs b/crates/devtools/src/planner_fuzz/relations.rs index 3bc274793..9b6d71ae9 100644 --- a/crates/devtools/src/planner_fuzz/relations.rs +++ b/crates/devtools/src/planner_fuzz/relations.rs @@ -954,12 +954,7 @@ fn with_root_estimate( estimate: Rc, dropped: &[String], ) -> Result, String> { - let Some(NonASAPOp::Project { - cols, - qualifier, - child, - }) = root.non_asap() - else { + let Some(NonASAPOp::Project { cols, child }) = root.non_asap() else { return Ok(estimate); }; let mut kept = Vec::new(); @@ -980,7 +975,6 @@ fn with_root_estimate( let cols = kept; OperatorNode::new_shared(Operator::NonASAP(NonASAPOp::Project { cols, - qualifier: qualifier.clone(), child: estimate, })) .map_err(|e| format!("projection over the new estimate: {e}")) diff --git a/crates/devtools/tests/viewer_contract.rs b/crates/devtools/tests/viewer_contract.rs index fc7d6c89f..0783f2e05 100644 --- a/crates/devtools/tests/viewer_contract.rs +++ b/crates/devtools/tests/viewer_contract.rs @@ -42,7 +42,6 @@ fn non_asap_samples() -> Vec> { }, NonASAPOp::Project { cols: vec![], - qualifier: None, child: (), }, NonASAPOp::Aggregate { diff --git a/crates/executor/src/expressions/columnar.rs b/crates/executor/src/expressions/columnar.rs index 9c15bc96b..6192a59c4 100644 --- a/crates/executor/src/expressions/columnar.rs +++ b/crates/executor/src/expressions/columnar.rs @@ -812,7 +812,6 @@ mod tests { name: name.into(), dtype: FieldDataType::Plain(dtype), nullable, - table: None, } } diff --git a/crates/executor/src/operators/aggregate/temporal.rs b/crates/executor/src/operators/aggregate/temporal.rs index 0ea8cedfd..55e2b628e 100644 --- a/crates/executor/src/operators/aggregate/temporal.rs +++ b/crates/executor/src/operators/aggregate/temporal.rs @@ -341,13 +341,11 @@ mod tests { name: "time".into(), dtype: SummaryFamilyType::Plain(DataType::Timestamp), nullable: false, - table: None, }, SummaryField { name: "value".into(), dtype: SummaryFamilyType::Plain(DataType::Float64), nullable: false, - table: None, }, ], time_index: Some(0), diff --git a/crates/executor/src/operators/common.rs b/crates/executor/src/operators/common.rs index f0182757d..0a9732dc5 100644 --- a/crates/executor/src/operators/common.rs +++ b/crates/executor/src/operators/common.rs @@ -15,7 +15,6 @@ pub(super) fn result_field(name: &str, dtype: DataType, nullable: bool) -> Summa name: name.into(), dtype: SummaryFamilyType::Plain(dtype), nullable, - table: None, } } diff --git a/crates/executor/src/operators/group_table.rs b/crates/executor/src/operators/group_table.rs index 64a173c97..53e6a702d 100644 --- a/crates/executor/src/operators/group_table.rs +++ b/crates/executor/src/operators/group_table.rs @@ -476,7 +476,6 @@ mod tests { name: name.into(), dtype: FieldDataType::Plain(dtype), nullable: true, - table: None, }; let schema = Arc::new(Schema { fields: vec![ @@ -548,7 +547,6 @@ mod tests { name: name.into(), dtype: FieldDataType::Plain(dtype), nullable: false, - table: None, }; let schema = Arc::new(Schema { fields: vec![ diff --git a/crates/executor/src/operators/series_labels.rs b/crates/executor/src/operators/series_labels.rs index f44a87933..fdf8b8b73 100644 --- a/crates/executor/src/operators/series_labels.rs +++ b/crates/executor/src/operators/series_labels.rs @@ -1130,7 +1130,6 @@ mod columnar_tests { name: name.into(), dtype: SummaryFamilyType::Plain(dtype), nullable, - table: None, } } diff --git a/crates/executor/src/operators/summary/mod.rs b/crates/executor/src/operators/summary/mod.rs index 4f40f2a06..2eee3373d 100644 --- a/crates/executor/src/operators/summary/mod.rs +++ b/crates/executor/src/operators/summary/mod.rs @@ -92,7 +92,6 @@ impl Operator { name: "state".into(), dtype: family.clone(), nullable: false, - table: None, }); Ok(Self { kind: Kind::AllGroupsSummaryBuild { @@ -177,7 +176,6 @@ impl Operator { name: "state".into(), dtype: family.clone(), nullable: false, - table: None, }); Ok(Self { kind: Kind::PerGroupSummaryBuild { diff --git a/crates/executor/src/values/batch.rs b/crates/executor/src/values/batch.rs index 12f06cf90..d9b9374c2 100644 --- a/crates/executor/src/values/batch.rs +++ b/crates/executor/src/values/batch.rs @@ -616,7 +616,6 @@ mod tests { name: name.into(), dtype: FieldDataType::Plain(dtype), nullable, - table: None, } } @@ -662,13 +661,11 @@ mod tests { name: "a".into(), dtype: DataType::Int64, nullable: false, - table: None, }, Field:: { name: "b".into(), dtype: DataType::Utf8, nullable: true, - table: None, }, ]; let labels = DataType::Map { @@ -786,7 +783,6 @@ mod tests { name: "state".into(), dtype: dd_family(), nullable: false, - table: None, }, ], time_index: None, @@ -868,7 +864,6 @@ mod tests { name: "state".into(), dtype: family.clone(), nullable: false, - table: None, }, ], time_index: None, @@ -1010,7 +1005,6 @@ mod tests { name: "state".into(), dtype: family, nullable: false, - table: None, }, ], time_index: None, @@ -1063,7 +1057,6 @@ mod tests { name: "state".into(), dtype: family.clone(), nullable: false, - table: None, }], time_index: None, unique_keys: vec![], @@ -1142,7 +1135,6 @@ mod tests { name: "state".into(), dtype: family, nullable: false, - table: None, }], time_index: None, unique_keys: vec![], diff --git a/crates/executor/tests/aggregate_memory_limit.rs b/crates/executor/tests/aggregate_memory_limit.rs index 75e6268b9..ac4a75085 100644 --- a/crates/executor/tests/aggregate_memory_limit.rs +++ b/crates/executor/tests/aggregate_memory_limit.rs @@ -53,7 +53,6 @@ fn run_aggregate( name: name.into(), dtype: FieldDataType::Plain(dtype), nullable: false, - table: None, }; let input = Arc::new(Schema { fields: vec![ diff --git a/crates/executor/tests/blocking_resources.rs b/crates/executor/tests/blocking_resources.rs index e397b15d8..d72749462 100644 --- a/crates/executor/tests/blocking_resources.rs +++ b/crates/executor/tests/blocking_resources.rs @@ -20,7 +20,6 @@ fn schema(width: usize) -> SchemaRef { closed: false, fields: (0..width) .map(|i| Field { - table: None, name: format!("v{i}"), dtype: FieldDataType::Plain(DataType::Int64), nullable: false, @@ -193,13 +192,11 @@ fn weighted_summary_build_yields_within_a_batch() { closed: false, fields: vec![ Field { - table: None, name: "item".into(), dtype: FieldDataType::Plain(DataType::Int64), nullable: false, }, Field { - table: None, name: "weight".into(), dtype: FieldDataType::Plain(DataType::Float64), nullable: false, diff --git a/crates/executor/tests/cheap_measures_memory_limit.rs b/crates/executor/tests/cheap_measures_memory_limit.rs index 070950a98..873286fef 100644 --- a/crates/executor/tests/cheap_measures_memory_limit.rs +++ b/crates/executor/tests/cheap_measures_memory_limit.rs @@ -90,7 +90,6 @@ fn many_groups_of_cheap_measures_fail_near_the_memory_limit() { name: name.into(), dtype: FieldDataType::Plain(dtype), nullable: false, - table: None, }; let input = Arc::new(Schema { fields: vec![field("k", DataType::Int64), field("v", DataType::Float64)], diff --git a/crates/executor/tests/current_series_heap.rs b/crates/executor/tests/current_series_heap.rs index 26d727944..804a6d4f7 100644 --- a/crates/executor/tests/current_series_heap.rs +++ b/crates/executor/tests/current_series_heap.rs @@ -28,7 +28,6 @@ fn schema() -> Arc { ] .into_iter() .map(|(name, dtype)| Field { - table: None, name: name.into(), dtype: FieldDataType::Plain(dtype), nullable: false, diff --git a/crates/executor/tests/deployment.rs b/crates/executor/tests/deployment.rs index e79652f97..9e0cc7bee 100644 --- a/crates/executor/tests/deployment.rs +++ b/crates/executor/tests/deployment.rs @@ -255,7 +255,6 @@ fn heap_sketch_states_load_as_dag_summary_values() { name: "state".into(), dtype: family.clone(), nullable: false, - table: None, }], unique_keys: vec![], closed: false, diff --git a/crates/executor/tests/exact_state_columns.rs b/crates/executor/tests/exact_state_columns.rs index 27cf337b7..c9fa8aaec 100644 --- a/crates/executor/tests/exact_state_columns.rs +++ b/crates/executor/tests/exact_state_columns.rs @@ -17,7 +17,6 @@ fn field(name: &str, dtype: DataType) -> Field { name: name.into(), dtype: FieldDataType::Plain(dtype), nullable: false, - table: None, } } diff --git a/crates/executor/tests/grouped_limit_memory_limit.rs b/crates/executor/tests/grouped_limit_memory_limit.rs index 10f5ceecf..e1b03188a 100644 --- a/crates/executor/tests/grouped_limit_memory_limit.rs +++ b/crates/executor/tests/grouped_limit_memory_limit.rs @@ -43,7 +43,6 @@ fn many_groups_in_one_batch_fail_near_the_memory_limit() { name: "k".into(), dtype: FieldDataType::Plain(DataType::Int64), nullable: false, - table: None, }], time_index: None, unique_keys: vec![], diff --git a/crates/executor/tests/hydra_cms_execution.rs b/crates/executor/tests/hydra_cms_execution.rs index fc23257c5..a71188167 100644 --- a/crates/executor/tests/hydra_cms_execution.rs +++ b/crates/executor/tests/hydra_cms_execution.rs @@ -121,7 +121,6 @@ fn summary_agg(update: SummaryUpdate, filter: Option>) -> Physi name: "state".into(), dtype: family(), nullable: false, - table: None, }, ]), ) diff --git a/crates/executor/tests/keyed_summary_row_weight.rs b/crates/executor/tests/keyed_summary_row_weight.rs index 19ecf85b9..0ab6e245d 100644 --- a/crates/executor/tests/keyed_summary_row_weight.rs +++ b/crates/executor/tests/keyed_summary_row_weight.rs @@ -160,7 +160,6 @@ fn query_time_count_heap_binds_a_constant_row_weight() { name: "state".into(), dtype: heap_family(), nullable: false, - table: None, }, ])) .clone(), diff --git a/crates/executor/tests/physical_dag.rs b/crates/executor/tests/physical_dag.rs index 7a0595e9e..996aecc58 100644 --- a/crates/executor/tests/physical_dag.rs +++ b/crates/executor/tests/physical_dag.rs @@ -25,7 +25,6 @@ fn schema(fields: &[(&str, DataType, bool)]) -> SchemaRef { fields: fields .iter() .map(|(name, dtype, nullable)| Field { - table: None, name: (*name).into(), dtype: FieldDataType::Plain(dtype.clone()), nullable: *nullable, @@ -468,7 +467,6 @@ fn exact_state_and_family_validation() { unique_keys: vec![], closed: false, fields: vec![Field { - table: None, name: "state".into(), dtype: family.clone(), nullable: false, @@ -564,7 +562,6 @@ fn bind_physical_asap_dag_before_execution() { )), }, }], - qualifier: None, child: 0, }), ), diff --git a/crates/executor/tests/physical_semantics.rs b/crates/executor/tests/physical_semantics.rs index 37d826fbe..5d002b2bd 100644 --- a/crates/executor/tests/physical_semantics.rs +++ b/crates/executor/tests/physical_semantics.rs @@ -25,7 +25,6 @@ fn schema(fields: &[(&str, DataType, bool)]) -> SchemaRef { fields: fields .iter() .map(|(name, dtype, nullable)| Field { - table: None, name: (*name).into(), dtype: FieldDataType::Plain(dtype.clone()), nullable: *nullable, @@ -924,7 +923,6 @@ fn planner_node( fields .iter() .map(|(name, dtype, nullable)| Field { - table: None, name: (*name).into(), dtype: dtype.clone(), nullable: *nullable, diff --git a/crates/executor/tests/precompute_population.rs b/crates/executor/tests/precompute_population.rs index 50cef473b..abc87ef26 100644 --- a/crates/executor/tests/precompute_population.rs +++ b/crates/executor/tests/precompute_population.rs @@ -22,7 +22,6 @@ fn logical_schema(family: FieldDataType) -> Schema { unique_keys: vec![], closed: false, fields: vec![Field { - table: None, name: "value".into(), dtype: family, nullable: false, diff --git a/crates/executor/tests/promql_binary.rs b/crates/executor/tests/promql_binary.rs index 47a13aebd..7f868f2c5 100644 --- a/crates/executor/tests/promql_binary.rs +++ b/crates/executor/tests/promql_binary.rs @@ -25,7 +25,6 @@ fn schema() -> SchemaRef { closed: false, fields: vec![ Field { - table: None, name: "labels".into(), dtype: FieldDataType::Plain(DataType::Map { key: Box::new(DataType::Utf8), @@ -35,7 +34,6 @@ fn schema() -> SchemaRef { nullable: false, }, Field { - table: None, name: "value".into(), dtype: FieldDataType::Plain(DataType::Float64), nullable: false, @@ -343,13 +341,11 @@ fn stored_series_evaluations_support_filters_and_sets() { closed: false, fields: vec![ Field { - table: None, name: PROMQL_SERIES_IDENTITY.into(), dtype: FieldDataType::Plain(DataType::Utf8), nullable: false, }, Field { - table: None, name: "value".into(), dtype: family.clone(), nullable: false, diff --git a/crates/executor/tests/raw_scan.rs b/crates/executor/tests/raw_scan.rs index 7a2ff24bf..671ba113d 100644 --- a/crates/executor/tests/raw_scan.rs +++ b/crates/executor/tests/raw_scan.rs @@ -34,7 +34,6 @@ fn fixture() -> (planner_types::ir::NonASAPOp, SchemaRef, Vec) { unique_keys: vec![], closed: false, fields: vec![Field { - table: None, name: "value".into(), dtype: FieldDataType::Plain(DataType::Int64), nullable: true, @@ -824,7 +823,6 @@ fn windowed_count_plan(shift_ms: Option) -> PhysicalASAPDAG { dag.edges.retain(|edge| edge.consumer < 3); let family = FieldDataType::ExactAggregate(ExactKind::Count, ExactParams::Count); let state = planner_types::ir::schema::Schema::new(vec![Field { - table: None, name: "count".into(), dtype: family.clone(), nullable: false, diff --git a/crates/executor/tests/series_window_memory_limit.rs b/crates/executor/tests/series_window_memory_limit.rs index e8c83bcd5..07ca7d29f 100644 --- a/crates/executor/tests/series_window_memory_limit.rs +++ b/crates/executor/tests/series_window_memory_limit.rs @@ -40,7 +40,6 @@ fn a_long_subquery_over_many_series_fails_near_the_memory_limit() { name: name.into(), dtype: FieldDataType::Plain(dtype), nullable: false, - table: None, }; let input = Arc::new(Schema { fields: vec![ diff --git a/crates/executor/tests/state_column_updater.rs b/crates/executor/tests/state_column_updater.rs index 3957844d8..52f8ce9fc 100644 --- a/crates/executor/tests/state_column_updater.rs +++ b/crates/executor/tests/state_column_updater.rs @@ -27,7 +27,6 @@ fn plain(name: &str, dtype: DataType) -> Field { name: name.into(), dtype: FieldDataType::Plain(dtype), nullable: false, - table: None, } } diff --git a/crates/executor/tests/stored_tumbling_window_memory_limit.rs b/crates/executor/tests/stored_tumbling_window_memory_limit.rs index 946e786fb..d02cf1e46 100644 --- a/crates/executor/tests/stored_tumbling_window_memory_limit.rs +++ b/crates/executor/tests/stored_tumbling_window_memory_limit.rs @@ -32,13 +32,11 @@ fn tumbling_window_of(series_count: u32) -> (Arc, Batch) { name: "series".into(), dtype: FieldDataType::Plain(DataType::Utf8), nullable: false, - table: None, }, Field { name: "state".into(), dtype: family.clone(), nullable: false, - table: None, }, ], time_index: None, diff --git a/crates/executor/tests/summary_build_memory_limit.rs b/crates/executor/tests/summary_build_memory_limit.rs index a14d3c713..d4c54924b 100644 --- a/crates/executor/tests/summary_build_memory_limit.rs +++ b/crates/executor/tests/summary_build_memory_limit.rs @@ -40,7 +40,6 @@ fn one_large_batch_fails_near_the_memory_limit() { name: name.into(), dtype: FieldDataType::Plain(dtype), nullable: false, - table: None, }; let input = Arc::new(Schema { fields: vec![ diff --git a/crates/executor/tests/summary_projection.rs b/crates/executor/tests/summary_projection.rs index 3e64a23c3..d291eb3b0 100644 --- a/crates/executor/tests/summary_projection.rs +++ b/crates/executor/tests/summary_projection.rs @@ -82,13 +82,11 @@ fn summary_projection_survives_recovery() { closed: false, fields: vec![ Field { - table: None, name: "state".into(), dtype: family.clone(), nullable: false, }, Field { - table: None, name: "service".into(), dtype: FieldDataType::Plain(DataType::Utf8), nullable: false, @@ -148,7 +146,6 @@ fn summary_projection_survives_recovery() { expr: planner_types::ir::ScalarExpr::Column(index), }) .collect(), - qualifier: None, child: 0, }), output_schema: output.clone(), @@ -246,7 +243,6 @@ fn summary_merge_without_inputs_is_rejected() { unique_keys: vec![], closed: false, fields: vec![Field { - table: None, name: "state".into(), dtype: family, nullable: false, diff --git a/crates/executor/tests/text_extremum_memory.rs b/crates/executor/tests/text_extremum_memory.rs index 7aec609c3..0cc056fa4 100644 --- a/crates/executor/tests/text_extremum_memory.rs +++ b/crates/executor/tests/text_extremum_memory.rs @@ -65,7 +65,6 @@ fn text_max_charges_only_the_current_value() { name: name.into(), dtype: FieldDataType::Plain(dtype), nullable: false, - table: None, }; let input = Arc::new(Schema { fields: vec![field("k", DataType::Int64), field("t", DataType::Utf8)], diff --git a/crates/executor/tests/univmon_execution.rs b/crates/executor/tests/univmon_execution.rs index cbc8e9128..730d5fb02 100644 --- a/crates/executor/tests/univmon_execution.rs +++ b/crates/executor/tests/univmon_execution.rs @@ -157,7 +157,6 @@ fn planner_unit_frequency_update_binds_as_a_value_frequency_build() -> Result<() false, )])); let state = Arc::new(Schema::new(vec![Field { - table: None, name: "state".into(), dtype: family(), nullable: false, diff --git a/crates/frontend-common/src/canonicalize.rs b/crates/frontend-common/src/canonicalize.rs index 43495769f..f99739f56 100644 --- a/crates/frontend-common/src/canonicalize.rs +++ b/crates/frontend-common/src/canonicalize.rs @@ -324,7 +324,6 @@ fn try_rewrite_row_number_top_k( ) -> Result>, SchemaDerivationError> { let Some(NonASAPOp::Project { cols: select_list, - qualifier, child: filter, }) = node.non_asap() else { @@ -437,14 +436,12 @@ fn try_rewrite_row_number_top_k( (Some(partition), Some(order)) if !partition_by.is_without() => { let selected = new(NonASAPOp::Project { cols: kept, - qualifier: qualifier.clone(), child: Rc::clone(ranked), })?; limit(GroupKeys::by(partition), order, selected)? } _ => new(NonASAPOp::Project { cols: kept, - qualifier: qualifier.clone(), child: limit(partition_by.clone(), order_by.clone(), Rc::clone(ranked))?, })?, }; @@ -648,7 +645,6 @@ mod tests { expr: ScalarExpr::Column(1), }, ], - qualifier: None, child, }) } @@ -766,7 +762,6 @@ mod tests { alias: None, expr: ScalarExpr::Column(0), }], - qualifier: None, child: Rc::clone(&ranking), }); let out = canonicalize(Rc::clone(&q)).unwrap(); @@ -994,7 +989,6 @@ mod tests { expr: ScalarExpr::Column(col), }) .collect(), - qualifier: None, child, }) } @@ -1232,7 +1226,6 @@ mod tests { let sub = one_column_subquery(); let root = node(NonASAPOp::Project { child: scan(), - qualifier: None, cols: vec![ ProjectItem { alias: Some("a".into()), diff --git a/crates/frontend-common/src/resolve.rs b/crates/frontend-common/src/resolve.rs index 1fb82f5a9..645a1a6f6 100644 --- a/crates/frontend-common/src/resolve.rs +++ b/crates/frontend-common/src/resolve.rs @@ -20,8 +20,8 @@ use std::rc::Rc; use thiserror::Error; -use asap_types::ir::operator::{AggIntent, GroupKeys, Reduction}; -use asap_types::ir::scalar::{resolve_column_ref, resolve_column_refs, ColumnRef, ResolveError}; +use asap_types::ir::operator::{AggIntent, GroupKeys, JoinKind, Reduction}; +use asap_types::ir::scalar::{resolve_column_ref, ColumnRef, ResolveError}; use asap_types::ir::schema::{ColumnId, Schema}; use asap_types::ir::SchemaDerivationError; use asap_types::ir::{NonASAPOp, OperatorNode, Predicate, ProjectItem, ScalarExpr, SortKey}; @@ -43,23 +43,102 @@ pub enum ResolveDAGError { use crate::canonicalize::canonicalize; +/// The names an operator's parent resolves against: the operator's output +/// schema and, per column, the SQL table or alias that qualifies it (`None` +/// when unqualified; an empty list qualifies no column). A qualified +/// reference (`f.src`) binds here, while lowering; the IR keeps no qualifier. +#[derive(Clone, Copy)] +struct Scope<'a> { + schema: &'a Schema, + qualifiers: &'a [Option], +} + +impl<'a> Scope<'a> { + /// A scope whose columns have no qualifier. + fn unqualified(schema: &'a Schema) -> Self { + Self { + schema, + qualifiers: &[], + } + } + + /// The column `column` names: by qualifier and name when it is + /// qualified and a column has both, else by its unique name. + fn resolve(&self, column: &ColumnRef) -> Result { + if let ColumnRef::Qualified { table, name } = column { + let qualified = + self.schema + .fields + .iter() + .zip(self.qualifiers) + .position(|(field, qualifier)| { + field.name == *name && qualifier.as_deref() == Some(table.as_str()) + }); + if let Some(id) = qualified { + return Ok(id); + } + } + resolve_column_ref(column, self.schema) + } + + fn resolve_all(&self, columns: &[ColumnRef]) -> Result, ResolveError> { + columns.iter().map(|column| self.resolve(column)).collect() + } + + /// The qualifier of column `column`. + fn qualifier(&self, column: ColumnId) -> Option { + self.qualifiers.get(column).cloned().flatten() + } +} + +/// A resolved operator with the qualifier of each output column: the scope +/// its parent resolves names in. +struct Bound { + node: Rc, + qualifiers: Vec>, +} + +impl Bound { + /// `node` with `qualifiers`; unqualified unless there is one per column. + fn new(node: Rc, qualifiers: Vec>) -> Self { + let qualifiers = if qualifiers.len() == node.schema.fields.len() { + qualifiers + } else { + Vec::new() + }; + Self { node, qualifiers } + } + + fn unqualified(node: Rc) -> Self { + Self::new(node, Vec::new()) + } + + fn scope(&self) -> Scope<'_> { + Scope { + schema: &self.node.schema, + qualifiers: &self.qualifiers, + } + } +} + /// Resolve the whole tree rooted at `tree`: bind every `ColumnRef` to a /// `ColumnId` via the usage-derived schema resolver, then canonicalize the result. pub fn resolve_root(tree: &UnresolvedOp) -> Result, ResolveDAGError> { - resolve_root_with_inherited(tree, &[]) + Ok(resolve_root_with_inherited(tree, &[])?.node) } /// [`resolve_root`] with label names inherited from an enclosing scope seeded /// into the leaf schema (a `BinaryOp` side, a scalar operand's operator). +/// Canonicalizing keeps the root's output columns, so their qualifiers. fn resolve_root_with_inherited( tree: &UnresolvedOp, inherited: &[String], -) -> Result, ResolveDAGError> { +) -> Result { let usage_derived_schema = resolve_schema_with_inherited(tree, inherited); let root = resolve(tree, &usage_derived_schema)?; - let root = canonicalize(root)?; - root.validate_structure()?; - Ok(root) + let node = canonicalize(root.node)?; + node.validate_structure()?; + Ok(Bound::new(node, root.qualifiers)) } /// Bind `tree` as a root in its own scope, inheriting from `enclosing` the @@ -73,7 +152,7 @@ fn resolve_nested_root( .into_iter() .filter(|n| !own.contains(n)) .collect(); - resolve_root_with_inherited(tree, &inherited) + Ok(resolve_root_with_inherited(tree, &inherited)?.node) } fn node(op: NonASAPOp) -> Result, ResolveDAGError> { @@ -83,31 +162,32 @@ fn node(op: NonASAPOp) -> Result, ResolveDAGError> { } /// The generic substitution walk. `usage_derived_schema` is the schema a -/// schemaless `Scan` in this scope binds to. -fn resolve( - tree: &UnresolvedOp, - usage_derived_schema: &Schema, -) -> Result, ResolveDAGError> { +/// schemaless `Scan` in this scope binds to. Qualifiers follow the columns: +/// a `Scan` qualifies its columns by its own; an operator that passes +/// columns through, a join and an aggregate's group keys keep them; a +/// `Project` qualifies all its columns by its own or none. +fn resolve(tree: &UnresolvedOp, usage_derived_schema: &Schema) -> Result { use UnresolvedOp as U; - let expr = - |e: &UnresolvedScalar, schema: &Schema| resolve_expr_in(e, schema, usage_derived_schema); - let pred = |p: &UnresolvedScalar, schema: &Schema| { - Ok::<_, ResolveDAGError>(Predicate(expr(p, schema)?)) - }; - let sort_keys = |keys: &[UnresolvedSortKey], schema: &Schema| { + let expr = |e: &UnresolvedScalar, scope: Scope| resolve_expr_in(e, scope, usage_derived_schema); + let pred = + |p: &UnresolvedScalar, scope: Scope| Ok::<_, ResolveDAGError>(Predicate(expr(p, scope)?)); + let sort_keys = |keys: &[UnresolvedSortKey], scope: Scope| { keys.iter() .map(|k| { Ok::<_, ResolveDAGError>(SortKey { - expr: expr(&k.expr, schema)?, + expr: expr(&k.expr, scope)?, ascending: k.ascending, nulls_first: k.nulls_first, }) }) .collect::, _>>() }; + // A node over `child` with `child`'s columns and qualifiers. + let passing = |op: NonASAPOp, child: Bound| Ok(Bound::new(node(op)?, child.qualifiers)); + let unqualified = |op: NonASAPOp| Ok(Bound::unqualified(node(op)?)); // A PromQL child over an open schema gets an opaque series-identity column. let promql_series = |child: &UnresolvedOp| -> Result, ResolveDAGError> { - let child = resolve(child, usage_derived_schema)?; + let child = resolve(child, usage_derived_schema)?.node; if child.schema.closed { return Ok(child); } @@ -119,19 +199,28 @@ fn resolve( source, predicates, schema, + qualifier, } => { let schema = schema .clone() .unwrap_or_else(|| usage_derived_schema.clone()); + let qualifiers = vec![qualifier.clone(); schema.fields.len()]; + let scope = Scope { + schema: &schema, + qualifiers: &qualifiers, + }; let predicates = predicates .iter() - .map(|p| pred(&p.0, &schema)) + .map(|p| pred(&p.0, scope)) .collect::, _>>()?; - node(NonASAPOp::Scan { - source: source.clone(), - predicates, - schema, - }) + Ok(Bound::new( + node(NonASAPOp::Scan { + source: source.clone(), + predicates, + schema, + })?, + qualifiers, + )) } // Row expressions have no input-column scope. @@ -141,11 +230,11 @@ fn resolve( .iter() .map(|row| { row.iter() - .map(|e| expr(e, &empty)) + .map(|e| expr(e, Scope::unqualified(&empty))) .collect::, _>>() }) .collect::, _>>()?; - node(NonASAPOp::Values { + unqualified(NonASAPOp::Values { rows, schema: schema.clone(), }) @@ -162,7 +251,13 @@ fn resolve( } => { let child = promql_series(child)?; let scalar = resolve_expr(scalar, &Schema::default())?; - lower_scalar_vector(child, scalar, op, *scalar_left, *return_bool) + Ok(Bound::unqualified(lower_scalar_vector( + child, + scalar, + op, + *scalar_left, + *return_bool, + )?)) } U::PromqlMap { child, @@ -171,42 +266,61 @@ fn resolve( } => { let child = promql_series(child)?; let sample = resolve_expr(sample, &child.schema)?; - project_sample(child, sample, *drop_metric_name) + Ok(Bound::unqualified(project_sample( + child, + sample, + *drop_metric_name, + )?)) } - U::PromqlVectorFromScalar(inner) => node(NonASAPOp::PromqlVectorFromScalar(expr( + U::PromqlVectorFromScalar(inner) => unqualified(NonASAPOp::PromqlVectorFromScalar(expr( inner, - usage_derived_schema, + Scope::unqualified(usage_derived_schema), )?)), U::PromqlRelabel { dst, value, child } => { let child = resolve(child, usage_derived_schema)?; - let value = expr(value, &child.schema)?; - node(NonASAPOp::PromqlRelabel { + let value = expr(value, child.scope())?; + unqualified(NonASAPOp::PromqlRelabel { dst: dst.clone(), value, - child, + child: child.node, }) } - U::PromqlInfoEnrich { selector, child } => node(NonASAPOp::PromqlInfoEnrich { - selector: selector.clone(), - child: resolve(child, usage_derived_schema)?, - }), + U::PromqlInfoEnrich { selector, child } => { + let child = resolve(child, usage_derived_schema)?; + passing( + NonASAPOp::PromqlInfoEnrich { + selector: selector.clone(), + child: Rc::clone(&child.node), + }, + child, + ) + } U::PromqlSeriesSample { by, kind, child } => { let child = resolve(child, usage_derived_schema)?; - let by = resolve_group_keys(by, &child.schema)?; - node(NonASAPOp::PromqlSeriesSample { - by, - kind: *kind, + let by = resolve_group_keys(by, child.scope())?; + passing( + NonASAPOp::PromqlSeriesSample { + by, + kind: *kind, + child: Rc::clone(&child.node), + }, child, - }) + ) } U::Filter { pred: p, child } => { let child = resolve(child, usage_derived_schema)?; - let pred = pred(&p.0, &child.schema)?; - node(NonASAPOp::Filter { pred, child }) + let pred = pred(&p.0, child.scope())?; + passing( + NonASAPOp::Filter { + pred, + child: Rc::clone(&child.node), + }, + child, + ) } U::Project { @@ -220,15 +334,18 @@ fn resolve( .map(|item| { Ok::<_, ResolveDAGError>(ProjectItem { alias: item.alias.clone(), - expr: expr(&item.expr, &child.schema)?, + expr: expr(&item.expr, child.scope())?, }) }) .collect::, _>>()?; - node(NonASAPOp::Project { - cols, - qualifier: qualifier.clone(), - child, - }) + let qualifiers = vec![qualifier.clone(); cols.len()]; + Ok(Bound::new( + node(NonASAPOp::Project { + cols, + child: child.node, + })?, + qualifiers, + )) } U::Aggregate { @@ -240,42 +357,76 @@ fn resolve( child, } => { let child = resolve(child, usage_derived_schema)?; - let reduction = resolve_reduction(reduction, &child.schema)?; + let scope = child.scope(); + let reduction = resolve_reduction(reduction, scope)?; let measures = measures .iter() - .map(|m| resolve_agg_intent(m, &child.schema)) + .map(|m| resolve_agg_intent(m, scope)) .collect::, ResolveError>>()?; let filters = filters .iter() - .map(|p| p.as_ref().map(|p| pred(&p.0, &child.schema)).transpose()) + .map(|p| p.as_ref().map(|p| pred(&p.0, scope)).transpose()) .collect::, _>>()?; + // A `by` key keeps its qualifier; a per-series reduction keeps + // its input's columns in place. A top-k emits ranked rows. + let qualifiers: Vec> = match &reduction { + _ if matches!(measures.as_slice(), [AggIntent::TopK { .. }]) => Vec::new(), + Reduction::PerEntity => child.qualifiers.clone(), + Reduction::Reduce(by) if by.is_without() => Vec::new(), + Reduction::Reduce(by) => by + .keys() + .iter() + .map(|&key| scope.qualifier(key)) + .chain(std::iter::repeat_n(None, measures.len())) + .collect(), + }; // HAVING is evaluated over the aggregate's own output. let having = having .as_ref() .map(|h| { let out_schema = aggregate_output_schema( - &child.schema, + &child.node.schema, &reduction, &measures, output_names, )?; - pred(&h.0, &out_schema) + let out_qualifiers = if qualifiers.len() == out_schema.fields.len() { + &qualifiers[..] + } else { + &[] + }; + pred( + &h.0, + Scope { + schema: &out_schema, + qualifiers: out_qualifiers, + }, + ) }) .transpose()?; - node(NonASAPOp::Aggregate { - reduction, - measures, - output_names: output_names.clone(), - filters, - having, - child, - }) + Ok(Bound::new( + node(NonASAPOp::Aggregate { + reduction, + measures, + output_names: output_names.clone(), + filters, + having, + child: child.node, + })?, + qualifiers, + )) } U::Dedup { cols, child } => { let child = resolve(child, usage_derived_schema)?; - let cols = resolve_column_refs(cols, &child.schema)?; - node(NonASAPOp::Dedup { cols, child }) + let cols = child.scope().resolve_all(cols)?; + passing( + NonASAPOp::Dedup { + cols, + child: Rc::clone(&child.node), + }, + child, + ) } U::Concat { children } => { @@ -283,7 +434,15 @@ fn resolve( .iter() .map(|c| resolve(c, usage_derived_schema)) .collect::, _>>()?; - node(NonASAPOp::Concat { children }) + let qualifiers = children + .first() + .map_or_else(Vec::new, |first| first.qualifiers.clone()); + Ok(Bound::new( + node(NonASAPOp::Concat { + children: children.into_iter().map(|child| child.node).collect(), + })?, + qualifiers, + )) } U::Join { @@ -296,15 +455,37 @@ fn resolve( // sets); the predicate sees left ++ right. let left = resolve_root_with_inherited(left, &[])?; let right = resolve_root_with_inherited(right, &[])?; - let mut concat = left.schema.clone(); - concat.fields.extend(right.schema.fields.iter().cloned()); - let pred = pred(&p.0, &concat)?; - node(NonASAPOp::Join { - kind: kind.clone(), - pred, - left, - right, - }) + let qualify = |side: &Bound| { + (0..side.node.schema.fields.len()) + .map(|column| side.scope().qualifier(column)) + .collect::>() + }; + let mut concat = left.node.schema.clone(); + concat + .fields + .extend(right.node.schema.fields.iter().cloned()); + let mut qualifiers = qualify(&left); + qualifiers.extend(qualify(&right)); + let pred = pred( + &p.0, + Scope { + schema: &concat, + qualifiers: &qualifiers, + }, + )?; + // A semi or anti join outputs the left side's columns. + if matches!(kind, JoinKind::Semi | JoinKind::Anti) { + qualifiers.truncate(left.node.schema.fields.len()); + } + Ok(Bound::new( + node(NonASAPOp::Join { + kind: kind.clone(), + pred, + left: left.node, + right: right.node, + })?, + qualifiers, + )) } U::SetOp { @@ -312,12 +493,19 @@ fn resolve( all, left, right, - } => node(NonASAPOp::SetOp { - kind: kind.clone(), - all: *all, - left: resolve_root_with_inherited(left, &[])?, - right: resolve_root_with_inherited(right, &[])?, - }), + } => { + let left = resolve_root_with_inherited(left, &[])?; + let right = resolve_root_with_inherited(right, &[])?; + Ok(Bound::new( + node(NonASAPOp::SetOp { + kind: kind.clone(), + all: *all, + left: left.node, + right: right.node, + })?, + left.qualifiers, + )) + } U::Sort { keys, @@ -325,13 +513,16 @@ fn resolve( child, } => { let child = resolve(child, usage_derived_schema)?; - let keys = sort_keys(keys, &child.schema)?; - let partition_by = resolve_group_keys(partition_by, &child.schema)?; - node(NonASAPOp::Sort { - keys, - partition_by, + let keys = sort_keys(keys, child.scope())?; + let partition_by = resolve_group_keys(partition_by, child.scope())?; + passing( + NonASAPOp::Sort { + keys, + partition_by, + child: Rc::clone(&child.node), + }, child, - }) + ) } U::Limit { @@ -341,35 +532,56 @@ fn resolve( child, } => { let child = resolve(child, usage_derived_schema)?; - let partition_by = resolve_group_keys(partition_by, &child.schema)?; - node(NonASAPOp::Limit { - n: *n, - offset: *offset, - partition_by, + let partition_by = resolve_group_keys(partition_by, child.scope())?; + passing( + NonASAPOp::Limit { + n: *n, + offset: *offset, + partition_by, + child: Rc::clone(&child.node), + }, child, - }) + ) } U::PromqlSubquery { range, resolution, child, - } => node(NonASAPOp::PromqlSubquery { - range: *range, - resolution: *resolution, - child: resolve(child, usage_derived_schema)?, - }), - - U::TimeRange { range, kind, child } => node(NonASAPOp::TimeRange { - range: *range, - kind: *kind, - child: resolve(child, usage_derived_schema)?, - }), - - U::TimeShift { shift, child } => node(NonASAPOp::TimeShift { - shift: *shift, - child: resolve(child, usage_derived_schema)?, - }), + } => { + let child = resolve(child, usage_derived_schema)?; + passing( + NonASAPOp::PromqlSubquery { + range: *range, + resolution: *resolution, + child: Rc::clone(&child.node), + }, + child, + ) + } + + U::TimeRange { range, kind, child } => { + let child = resolve(child, usage_derived_schema)?; + passing( + NonASAPOp::TimeRange { + range: *range, + kind: *kind, + child: Rc::clone(&child.node), + }, + child, + ) + } + + U::TimeShift { shift, child } => { + let child = resolve(child, usage_derived_schema)?; + passing( + NonASAPOp::TimeShift { + shift: *shift, + child: Rc::clone(&child.node), + }, + child, + ) + } U::SQLWindowFunc { func, @@ -383,19 +595,27 @@ fn resolve( let child = resolve(child, usage_derived_schema)?; let args = args .iter() - .map(|a| expr(a, &child.schema)) + .map(|a| expr(a, child.scope())) .collect::, _>>()?; - let partition_by = resolve_group_keys(partition_by, &child.schema)?; - let order_by = sort_keys(order_by, &child.schema)?; - node(NonASAPOp::SQLWindowFunc { - func: func.clone(), - args, - partition_by, - order_by, - frame: frame.clone(), - output_name: output_name.clone(), - child, - }) + let partition_by = resolve_group_keys(partition_by, child.scope())?; + let order_by = sort_keys(order_by, child.scope())?; + // The window function's own column comes last, unqualified. + let mut qualifiers = child.qualifiers.clone(); + if !qualifiers.is_empty() { + qualifiers.push(None); + } + Ok(Bound::new( + node(NonASAPOp::SQLWindowFunc { + func: func.clone(), + args, + partition_by, + order_by, + frame: frame.clone(), + output_name: output_name.clone(), + child: child.node, + })?, + qualifiers, + )) } U::BinaryOp { @@ -414,11 +634,11 @@ fn resolve( .into_iter() .filter(|n| !own.contains(n)) .collect(); - node(NonASAPOp::BinaryOp { + unqualified(NonASAPOp::BinaryOp { operator: operator.clone(), return_bool: *return_bool, - lhs: resolve_root_with_inherited(lhs, &inherited)?, - rhs: resolve_root_with_inherited(rhs, &inherited)?, + lhs: resolve_root_with_inherited(lhs, &inherited)?.node, + rhs: resolve_root_with_inherited(rhs, &inherited)?.node, }) } } @@ -442,28 +662,29 @@ pub(crate) fn resolve_expr( expr: &UnresolvedScalar, schema: &Schema, ) -> Result { - resolve_expr_in(expr, schema, schema) + resolve_expr_in(expr, Scope::unqualified(schema), schema) } -/// [`resolve_expr`] where the operators the expression reads inherit from -/// `enclosing` (the owning root's usage-derived schema) rather than from `schema`. +/// [`resolve_expr`] in `scope`, where the operators the expression reads +/// inherit from `enclosing` (the owning root's usage-derived schema) rather +/// than from the scope's schema. fn resolve_expr_in( expr: &UnresolvedScalar, - schema: &Schema, + scope: Scope, enclosing: &Schema, ) -> Result { use UnresolvedScalar as S; let bx = |e: &UnresolvedScalar| -> Result, ResolveDAGError> { - Ok(Box::new(resolve_expr_in(e, schema, enclosing)?)) + Ok(Box::new(resolve_expr_in(e, scope, enclosing)?)) }; let each = |es: &[UnresolvedScalar]| -> Result, ResolveDAGError> { es.iter() - .map(|e| resolve_expr_in(e, schema, enclosing)) + .map(|e| resolve_expr_in(e, scope, enclosing)) .collect() }; let op = |o: &UnresolvedOp| resolve_nested_root(o, enclosing); Ok(match expr { - S::Column(c) => ScalarExpr::Column(resolve_column_ref(c, schema)?), + S::Column(c) => ScalarExpr::Column(scope.resolve(c)?), S::Literal(s) => ScalarExpr::Literal(s.clone()), S::EvalTimestamp => ScalarExpr::EvalTimestamp, S::CurrentTimestamp => ScalarExpr::CurrentTimestamp, @@ -526,8 +747,8 @@ fn resolve_expr_in( .iter() .map(|(w, t)| { Ok(( - resolve_expr_in(w, schema, enclosing)?, - resolve_expr_in(t, schema, enclosing)?, + resolve_expr_in(w, scope, enclosing)?, + resolve_expr_in(t, scope, enclosing)?, )) }) .collect::, ResolveDAGError>>()?, @@ -554,9 +775,9 @@ fn resolve_expr_in( /// Resolve name-based group keys positionally, preserving `by`/`without`. fn resolve_group_keys( keys: &GroupKeys, - schema: &Schema, + scope: Scope, ) -> Result, ResolveError> { - let ids = resolve_column_refs(keys.keys(), schema)?; + let ids = scope.resolve_all(keys.keys())?; Ok(if keys.is_without() { GroupKeys::without(ids) } else { @@ -577,15 +798,15 @@ fn resolve_group_keys( /// may exist at runtime, and PromQL leaves seed every referenced label via the /// schema resolver, so an unresolved key over an open schema indicates a resolution bug, /// not an absent label. (SQL is unaffected — its `GROUP BY` resolves through -/// the strict [`resolve_column_refs`], and DataFusion has already validated +/// the strict [`resolve_group_keys`], and DataFusion has already validated /// the columns anyway.) fn resolve_group_keys_promql( cols: &[ColumnRef], - schema: &Schema, + scope: Scope, ) -> Result, ResolveError> { cols.iter() - .filter_map(|c| match resolve_column_ref(c, schema) { - Err(ResolveError::NotFound { .. }) if schema.closed => None, + .filter_map(|c| match scope.resolve(c) { + Err(ResolveError::NotFound { .. }) if scope.schema.closed => None, other => Some(other), }) .collect() @@ -600,11 +821,11 @@ fn resolve_group_keys_promql( /// path is a no-op difference there. fn resolve_reduction( reduction: &Reduction, - schema: &Schema, + scope: Scope, ) -> Result, ResolveError> { Ok(match reduction { Reduction::Reduce(by) => { - let ids = resolve_group_keys_promql(by.keys(), schema)?; + let ids = resolve_group_keys_promql(by.keys(), scope)?; Reduction::Reduce(if by.is_without() { GroupKeys::without(ids) } else { @@ -620,20 +841,18 @@ fn resolve_reduction( /// convention); every other field carries through unchanged. fn resolve_agg_intent( intent: &AggIntent, - schema: &Schema, + scope: Scope, ) -> Result, ResolveError> { let col = |c: &Option| -> Result, ResolveError> { - c.as_ref() - .map(|r| resolve_column_ref(r, schema)) - .transpose() + c.as_ref().map(|r| scope.resolve(r)).transpose() }; Ok(match intent { AggIntent::Count { accuracy } => AggIntent::Count { accuracy: accuracy.clone(), }, AggIntent::PearsonCorr { left, right } => AggIntent::PearsonCorr { - left: resolve_column_ref(left, schema)?, - right: resolve_column_ref(right, schema)?, + left: scope.resolve(left)?, + right: scope.resolve(right)?, }, AggIntent::Sum { col: c } => AggIntent::Sum { col: col(c)? }, AggIntent::Min { col: c } => AggIntent::Min { col: col(c)? }, @@ -661,10 +880,7 @@ fn resolve_agg_intent( accuracy: accuracy.clone(), }, AggIntent::Cardinality { cols, accuracy } => AggIntent::Cardinality { - cols: cols - .iter() - .map(|c| resolve_column_ref(c, schema)) - .collect::>()?, + cols: scope.resolve_all(cols)?, accuracy: accuracy.clone(), }, AggIntent::FrequencyL2 { col: c, accuracy } => AggIntent::FrequencyL2 { @@ -699,7 +915,7 @@ fn resolve_agg_intent( }, AggIntent::HistogramQuantile { q, le } => AggIntent::HistogramQuantile { q: *q, - le: resolve_column_ref(le, schema)?, + le: scope.resolve(le)?, }, AggIntent::Math(f) => AggIntent::Math(f.clone()), AggIntent::Absent { labels } => AggIntent::Absent { @@ -839,11 +1055,7 @@ fn project_sample( } }) .collect(); - node(NonASAPOp::Project { - child, - cols, - qualifier: None, - }) + node(NonASAPOp::Project { child, cols }) } #[cfg(test)] @@ -866,6 +1078,7 @@ mod tests { }, predicates: vec![], schema: None, + qualifier: None, } } @@ -895,9 +1108,14 @@ mod tests { #[test] fn resolve_pearson_corr_inputs() { let schema = Schema::new(vec![ - Field::plain("x", DataType::Float64, true).with_table("a"), - Field::plain("x", DataType::Float64, true).with_table("b"), + Field::plain("x", DataType::Float64, true), + Field::plain("x", DataType::Float64, true), ]); + let qualifiers = [Some("a".to_string()), Some("b".to_string())]; + let scope = Scope { + schema: &schema, + qualifiers: &qualifiers, + }; let intent = AggIntent::PearsonCorr { left: ColumnRef::Qualified { table: "a".into(), @@ -909,7 +1127,7 @@ mod tests { }, }; assert_eq!( - resolve_agg_intent(&intent, &schema).unwrap(), + resolve_agg_intent(&intent, scope).unwrap(), AggIntent::PearsonCorr { left: 0, right: 1 } ); let missing = AggIntent::PearsonCorr { @@ -919,7 +1137,7 @@ mod tests { }, right: ColumnRef::Named("missing".into()), }; - assert!(resolve_agg_intent(&missing, &schema).is_err()); + assert!(resolve_agg_intent(&missing, scope).is_err()); } // Every leg resolves independently, qualifiers included; one unknown leg @@ -927,9 +1145,14 @@ mod tests { #[test] fn resolve_distinct_tuple_columns() { let schema = Schema::new(vec![ - Field::plain("k", DataType::Int64, true).with_table("a"), - Field::plain("k", DataType::Int64, true).with_table("b"), + Field::plain("k", DataType::Int64, true), + Field::plain("k", DataType::Int64, true), ]); + let qualifiers = [Some("a".to_string()), Some("b".to_string())]; + let scope = Scope { + schema: &schema, + qualifiers: &qualifiers, + }; let qualified = |table: &str| ColumnRef::Qualified { table: table.into(), name: "k".into(), @@ -939,7 +1162,7 @@ mod tests { accuracy: AccuracyTarget::exact(), }; assert_eq!( - resolve_agg_intent(&intent, &schema).unwrap(), + resolve_agg_intent(&intent, scope).unwrap(), AggIntent::Cardinality { cols: vec![1, 0], accuracy: AccuracyTarget::exact(), @@ -949,7 +1172,7 @@ mod tests { cols: vec![qualified("a"), ColumnRef::Named("missing".into())], accuracy: AccuracyTarget::exact(), }; - assert!(resolve_agg_intent(&missing, &schema).is_err()); + assert!(resolve_agg_intent(&missing, scope).is_err()); } // ` > `: the bridged literal comes through unchanged, the @@ -1040,6 +1263,7 @@ mod tests { }, predicates: vec![], schema: Some(input), + qualifier: None, }), }; let resolved = resolve_root(&unresolved).expect("resolves"); @@ -1063,9 +1287,8 @@ mod tests { table_ref: table.into(), }, predicates: vec![], - schema: Some(Schema::new(vec![ - Field::plain("k", DataType::Int64, false).with_table(table) - ])), + schema: Some(Schema::new(vec![Field::plain("k", DataType::Int64, false)])), + qualifier: Some(table.into()), }; let qualified = |table: &str| { UnresolvedScalar::Column(ColumnRef::Qualified { @@ -1105,6 +1328,7 @@ mod tests { source: Source::TimeSeries { metric: "x".into() }, predicates: vec![UnresolvedPredicate(eq_lit(named("a"), "1"))], schema: None, + qualifier: None, }; let unresolved = UnresolvedOp::PromqlScalarOp { child: Rc::new(scan("m")), @@ -1194,7 +1418,7 @@ mod tests { closed: true, }; assert_eq!( - resolve_group_keys_promql(&[ColumnRef::Named("job".into())], &s), + resolve_group_keys_promql(&[ColumnRef::Named("job".into())], Scope::unqualified(&s)), Ok(vec![]) ); // Present keys still resolve positionally; absent ones drop around them. @@ -1204,7 +1428,7 @@ mod tests { ColumnRef::Named("job".into()), ColumnRef::Named("group".into()) ], - &s + Scope::unqualified(&s) ), Ok(vec![0]) ); @@ -1217,7 +1441,7 @@ mod tests { // absent label. Keep the strict error. let open = ts_value_schema(); // closed: false assert!(matches!( - resolve_group_keys_promql(&[ColumnRef::Named("job".into())], &open), + resolve_group_keys_promql(&[ColumnRef::Named("job".into())], Scope::unqualified(&open)), Err(ResolveError::NotFound { .. }) )); } diff --git a/crates/frontend-common/src/schema_resolver.rs b/crates/frontend-common/src/schema_resolver.rs index 5462304cc..96be18f6e 100644 --- a/crates/frontend-common/src/schema_resolver.rs +++ b/crates/frontend-common/src/schema_resolver.rs @@ -222,6 +222,7 @@ mod tests { }, predicates: vec![], schema: None, + qualifier: None, } } diff --git a/crates/frontend-common/src/unresolved.rs b/crates/frontend-common/src/unresolved.rs index 52255beeb..0e55023d4 100644 --- a/crates/frontend-common/src/unresolved.rs +++ b/crates/frontend-common/src/unresolved.rs @@ -114,6 +114,10 @@ pub enum UnresolvedOp { /// `Some` for a catalog-backed (SQL) leaf; `None` defers to the /// usage-derived schema resolver. schema: Option, + /// The SQL table or alias qualifying the leaf's columns, which + /// qualified references (`t.col`) above resolve by; `None` for a + /// PromQL leaf. + qualifier: Option, }, Values { rows: Vec>, @@ -125,6 +129,9 @@ pub enum UnresolvedOp { }, Project { cols: Vec, + /// The alias of a derived table (`FROM (SELECT …) t`) qualifying + /// every output column, which qualified references above resolve + /// by; `None` for an ordinary SELECT list. qualifier: Option, child: Rc, }, diff --git a/crates/frontend-metricsql/src/lib.rs b/crates/frontend-metricsql/src/lib.rs index 4a3172506..13523dd0c 100644 --- a/crates/frontend-metricsql/src/lib.rs +++ b/crates/frontend-metricsql/src/lib.rs @@ -166,6 +166,7 @@ impl Lowerer<'_> { .map(|f| UnresolvedPredicate(matcher(f))) .collect(), schema: None, + qualifier: None, }) } diff --git a/crates/frontend-promql/src/promql.rs b/crates/frontend-promql/src/promql.rs index 6c8a33f2a..57134ef34 100644 --- a/crates/frontend-promql/src/promql.rs +++ b/crates/frontend-promql/src/promql.rs @@ -1651,6 +1651,7 @@ fn filtered_source(metric: String, matchers: Vec, shift: TimeShift) -> U predicates: matchers.into_iter().map(UnresolvedPredicate).collect(), // Usage-derived (PromQL is schemaless) — the schema resolver fills this in. schema: None, + qualifier: None, }; if shift.is_identity() { scan diff --git a/crates/frontend-sql/src/frequency.rs b/crates/frontend-sql/src/frequency.rs index c95f4af1a..819174ba6 100644 --- a/crates/frontend-sql/src/frequency.rs +++ b/crates/frontend-sql/src/frequency.rs @@ -349,7 +349,7 @@ fn sql_frequency_result( name: &str, value: ScalarExpr, ) -> Option> { - let NonASAPOp::Project { qualifier, .. } = root.non_asap()? else { + let NonASAPOp::Project { .. } = root.non_asap()? else { return None; }; let aggregate = |measure, name: &str| { @@ -398,7 +398,6 @@ fn sql_frequency_result( else_expr: Some(Box::new(value)), }, }], - qualifier: qualifier.clone(), child, })) .ok()?; diff --git a/crates/frontend-sql/src/sql/mod.rs b/crates/frontend-sql/src/sql/mod.rs index e684874c6..bf76881c0 100644 --- a/crates/frontend-sql/src/sql/mod.rs +++ b/crates/frontend-sql/src/sql/mod.rs @@ -536,7 +536,7 @@ impl<'a> SqlLowerer<'a> { self.scan_source(&table, &table) } - /// A `Scan` over catalog table `table`, with its columns qualified by + /// A `Scan` over catalog table `table`, its columns qualified by /// `qualifier` (the table name, or an alias from a `SubqueryAlias`) so /// `Qualified` column refs resolve to the right side across a join. fn scan_source(&self, table: &str, qualifier: &str) -> Result { @@ -545,24 +545,17 @@ impl<'a> SqlLowerer<'a> { .tables .get(table) .ok_or_else(|| LoweringError::TableNotFound(table.to_string()))?; - let qualified = Schema { - fields: schema - .fields - .iter() - .cloned() - .map(|c| c.with_table(qualifier)) - .collect(), - time_index: schema.time_index, - unique_keys: schema.unique_keys.clone(), - // Catalog-backed: the table's columns are fully declared → closed. - closed: true, - }; Ok(Unresolved::Scan { source: Source::Table { table_ref: table.to_string(), }, predicates: vec![], - schema: Some(qualified), + schema: Some(Schema { + // Catalog-backed: the table's columns are fully declared → closed. + closed: true, + ..schema.clone() + }), + qualifier: Some(qualifier.to_string()), }) } @@ -1945,12 +1938,14 @@ fn filter_or_fold(pred: Scalar, child: Unresolved) -> Unresolved { source, mut predicates, schema, + qualifier, } => { predicates.push(Predicate(pred)); Unresolved::Scan { source, predicates, schema, + qualifier, } } other => Unresolved::Filter { diff --git a/crates/frontend-sql/tests/sql_lowering.rs b/crates/frontend-sql/tests/sql_lowering.rs index fcb2b731d..18ae33b86 100644 --- a/crates/frontend-sql/tests/sql_lowering.rs +++ b/crates/frontend-sql/tests/sql_lowering.rs @@ -2947,6 +2947,34 @@ async fn lower_err(sql: &str) -> String { .to_string() } +/// Both join sides have a `service` column; the front end binds `h.service` +/// with its own scope while lowering, so the group key is the right side's +/// column, the IR schema carrying no qualifier. +#[tokio::test] +async fn a_qualified_group_key_binds_its_join_side() { + let qe = lower( + "SELECT h.service, SUM(m.bytes) FROM metrics m JOIN hosts h ON m.service = h.service \ + GROUP BY h.service", + ) + .await; + let aggregate = find_aggregate_node(&qe).expect("aggregate"); + let NonASAPOp::Aggregate { + reduction, child, .. + } = op(aggregate) + else { + unreachable!() + }; + let [key] = reduction.expect_reduce().keys() else { + panic!("one key: {reduction:?}") + }; + let metrics_columns = 4; + assert!( + *key >= metrics_columns, + "{key} is a metrics column: {child:#?}" + ); + assert_eq!(child.schema.fields[*key].name, "service"); +} + /// After a join the `Project` drops table qualifiers, so `h.service` must not /// fall back to the first bare `service` (`m.service`): the reference is /// ambiguous and rejected rather than sorting by the wrong column. @@ -2957,7 +2985,7 @@ async fn qualified_order_by_after_a_join_does_not_bind_the_wrong_column() { ORDER BY h.service", ) .await; - assert!(err.contains("several unqualified columns"), "{err}"); + assert!(err.contains("several columns"), "{err}"); } /// `WITHIN GROUP (ORDER BY x DESC)` reads the percentile from the top, as diff --git a/crates/integration-tests/tests/sql_to_post_asap.rs b/crates/integration-tests/tests/sql_to_post_asap.rs index 411f5b612..2d8829f76 100644 --- a/crates/integration-tests/tests/sql_to_post_asap.rs +++ b/crates/integration-tests/tests/sql_to_post_asap.rs @@ -268,23 +268,16 @@ async fn sql_full_query_retains_project_and_binds_inner_aggregate() { .await; let Some(NonASAPOp::Project { cols: expected_cols, - qualifier: expected_qualifier, .. }) = pre_asap.non_asap() else { panic!("sanity: a SQL root is a Project, unlike lower_promql's bare Aggregate"); }; let root = realize(&pre_asap, AccuracyTarget::max_error(0.01)); - let Some(NonASAPOp::Project { - child, - cols, - qualifier, - }) = root.non_asap() - else { + let Some(NonASAPOp::Project { child, cols }) = root.non_asap() else { panic!("expected retained Project root, got {:?}", root.operator); }; assert_eq!(cols, expected_cols, "projection expressions and aliases"); - assert_eq!(qualifier, expected_qualifier, "projection qualifier"); assert_eq!(root.schema.fields[0].name, "p99", "project output schema"); assert_eq!( root.schema.fields[0].dtype, diff --git a/crates/planner/src/lib.rs b/crates/planner/src/lib.rs index db7d1f891..400e42e7d 100644 --- a/crates/planner/src/lib.rs +++ b/crates/planner/src/lib.rs @@ -60,7 +60,8 @@ pub use asap_types::ir::materialized_view::{ MaterializedViewRefresh, RawDataRetention, RawDataWindow, RefreshWindow, ViewCoverage, }; pub use asap_types::ir::physical_export::{PhysicalASAPDAG, PhysicalASAPNodeId}; -pub use asap_types::ir::properties::summary_coverage::{ColumnIdentity, Constraint, SelectionBox}; +pub use asap_types::ir::properties::column_lineage::SourceColumn; +pub use asap_types::ir::properties::summary_coverage::{Constraint, SelectionBox}; pub use asap_types::ir::properties::ExecutionTiming; pub use asap_types::ir::schema::{SketchAlgorithm, SketchParams, StatisticKind}; pub use asap_types::workload::PlanningWorkload; diff --git a/crates/stage0-memo/src/lib.rs b/crates/stage0-memo/src/lib.rs index 80697373c..4521b07a3 100644 --- a/crates/stage0-memo/src/lib.rs +++ b/crates/stage0-memo/src/lib.rs @@ -459,7 +459,6 @@ mod tests { alias: Some("v".into()), expr: ScalarExpr::literal_f64(value), }], - qualifier: None, child: scan(), }) } diff --git a/crates/stage1-pass1-per-computation-candidates/src/per_computation_candidates.rs b/crates/stage1-pass1-per-computation-candidates/src/per_computation_candidates.rs index 3ca44bfd4..e59abc541 100644 --- a/crates/stage1-pass1-per-computation-candidates/src/per_computation_candidates.rs +++ b/crates/stage1-pass1-per-computation-candidates/src/per_computation_candidates.rs @@ -1469,7 +1469,7 @@ fn with_names_of( // A top-k's ranked rows may carry their item columns after the // aggregate's ([`ranked_as_topk`]). if names(&output).starts_with(&names(target)) { - return Ok(with_aggregate_qualifiers(target, output)); + return Ok(output); } if output.schema.fields.len() != target.schema.fields.len() { return Err(PerComputationCandidateError::Unsupported( @@ -1484,34 +1484,12 @@ fn with_names_of( expr: ScalarExpr::Column(column), }) .collect(); - Ok(with_aggregate_qualifiers( - target, - OperatorNode::new_shared(Operator::NonASAP(NonASAPOp::Project { + Ok(OperatorNode::new_shared(Operator::NonASAP( + NonASAPOp::Project { cols, - qualifier: None, child: output, - }))?, - )) -} - -/// An equivalent aggregate replacement must retain SQL field identity. These -/// explicit qualifiers survive child rebinding; types and keys remain derived -/// from executable operators rather than copied from the requested output. -fn with_aggregate_qualifiers(target: &OperatorNode, output: Rc) -> Rc { - if output - .schema - .fields - .iter() - .zip(&target.schema.fields) - .all(|(field, expected)| field.table == expected.table) - { - return output; - } - let mut output = output.as_ref().clone(); - for (field, expected) in output.schema.fields.iter_mut().zip(&target.schema.fields) { - field.table = expected.table.clone(); - } - Rc::new(output) + }, + ))?) } /// A whole-expression top-k for `target` ([`whole_expression_input`]): one @@ -1589,7 +1567,6 @@ fn ranked_as_topk( Ok(OperatorNode::new_shared(Operator::NonASAP( NonASAPOp::Project { cols, - qualifier: None, child: estimate, }, ))?) @@ -1799,11 +1776,7 @@ fn last_as_float64( }, }); Ok(OperatorNode::new_shared(Operator::NonASAP( - NonASAPOp::Project { - cols, - qualifier: None, - child: node, - }, + NonASAPOp::Project { cols, child: node }, ))?) } @@ -1865,7 +1838,6 @@ fn sql_avg( Ok(OperatorNode::new_shared(Operator::NonASAP( NonASAPOp::Project { cols, - qualifier: None, child: join_on_keys(keys, sum, count)?, }, ))?) @@ -1960,7 +1932,6 @@ fn join_measures( Ok(OperatorNode::new_shared(Operator::NonASAP( NonASAPOp::Project { cols, - qualifier: None, child: joined, }, ))?) diff --git a/crates/types/src/ir/flat.rs b/crates/types/src/ir/flat.rs index f5c940069..731def21b 100644 --- a/crates/types/src/ir/flat.rs +++ b/crates/types/src/ir/flat.rs @@ -129,7 +129,6 @@ mod tests { alias: Some("s".into()), expr: ScalarExpr::ScalarSubquery(Rc::clone(&v)), }], - qualifier: None, child: Rc::clone(&v), })) .unwrap(); diff --git a/crates/types/src/ir/operator/agg_intent.rs b/crates/types/src/ir/operator/agg_intent.rs index 4bde9c1b2..f035d23bc 100644 --- a/crates/types/src/ir/operator/agg_intent.rs +++ b/crates/types/src/ir/operator/agg_intent.rs @@ -508,9 +508,11 @@ impl AggIntent { ), // Ranked rows retain the ranked value's type; the enclosing // operator also preserves item identity via schema::topk. - AggIntent::TopK { k, .. } => { - col(&format!("topk_{k}"), input.dtype.clone(), input.nullable) - } + AggIntent::TopK { k, .. } => col( + &crate::ir::schema::topk::topk_output_name(*k, &[]), + input.dtype.clone(), + input.nullable, + ), AggIntent::Cardinality { .. } => col("cardinality", DataType::Int64, false), AggIntent::FrequencyL2 { .. } => col("frequency_l2", DataType::Float64, false), AggIntent::FrequencyEntropy { .. } => { diff --git a/crates/types/src/ir/operator/node.rs b/crates/types/src/ir/operator/node.rs index 956a85dc6..6aba895d7 100644 --- a/crates/types/src/ir/operator/node.rs +++ b/crates/types/src/ir/operator/node.rs @@ -121,7 +121,7 @@ impl OperatorNode { Ok(Self::with_schema(operator, schema)) } - /// Build a node with caller-supplied output names and qualifiers. For + /// Build a node with caller-supplied output names. For /// either operator category, `validate_structure` requires all other /// schema metadata to agree with derivation; output kind is derived here. pub fn with_schema(operator: Operator, schema: Schema) -> Self { @@ -190,10 +190,9 @@ impl OperatorNode { /// Replace the children and rebuild: each child is replaced by `f(child)` /// and the node is rebuilt over the new children, re-deriving all - /// structural schema metadata. Only names and qualifiers that override - /// the old derived schema are retained, for either operator category. A - /// change in output arity with such overrides needs an explicit new - /// naming assignment. + /// structural schema metadata. Only names that override the old derived + /// schema are retained, for either operator category. A change in output + /// arity with such overrides needs an explicit new naming assignment. /// `guarantee` depends on the inputs and is cleared. pub fn with_new_children( &self, @@ -203,22 +202,15 @@ impl OperatorNode { let mut rebuilt = Self::new(self.operator.map_children(f))?; for (i, (derived, retained)) in previous.fields.iter().zip(&self.schema.fields).enumerate() { - let renamed = retained.name != derived.name; - let requalified = retained.table != derived.table; - if !renamed && !requalified { + if retained.name == derived.name { continue; } if rebuilt.schema.fields.len() != previous.fields.len() { return Err(SchemaDerivationError::InvalidScalarSignature( - "output arity changed; reassign explicit output names and qualifiers".into(), + "output arity changed; reassign explicit output names".into(), )); } - if renamed { - rebuilt.schema.fields[i].name = retained.name.clone(); - } - if requalified { - rebuilt.schema.fields[i].table = retained.table.clone(); - } + rebuilt.schema.fields[i].name = retained.name.clone(); } Ok(rebuilt) } @@ -262,7 +254,7 @@ impl OperatorNode { /// input contract, scalar typing against the owning operator's input /// schema, and agreement between each retained schema and the one /// derived from the operator. Both operator categories may override field - /// names and qualifiers; all structural metadata must match derivation. + /// names; all structural metadata must match derivation. pub fn validate_structure(self: &Rc) -> Result<(), SchemaDerivationError> { for node in Self::reachable(self) { if node.schema.time_index.is_some_and(|i| { @@ -291,7 +283,6 @@ impl OperatorNode { // so new structural metadata cannot accidentally escape validation. for (field, retained) in derived.fields.iter_mut().zip(&node.schema.fields) { field.name = retained.name.clone(); - field.table = retained.table.clone(); } let agree = derived == node.schema; if !agree { diff --git a/crates/types/src/ir/operator/non_asap.rs b/crates/types/src/ir/operator/non_asap.rs index dd1755456..f064cf622 100644 --- a/crates/types/src/ir/operator/non_asap.rs +++ b/crates/types/src/ir/operator/non_asap.rs @@ -74,14 +74,7 @@ pub enum NonASAPOp> { /// σ — row-level filter. Output schema = child schema. Filter { pred: Predicate, child: C }, /// π — projection. - Project { - cols: Vec>, - /// Re-qualifies every output column with this table alias (a derived - /// table / inline view). `None` for an ordinary SELECT list. - #[serde(default)] - qualifier: Option, - child: C, - }, + Project { cols: Vec>, child: C }, /// γ + α — grouping + aggregate intents. Aggregate { reduction: Reduction, @@ -299,11 +292,7 @@ impl NonASAPOp { child: f(child), } } - Project { - cols, - qualifier, - child, - } => { + Project { cols, child } => { let cols = cols .iter() .map(|c| ProjectItem { @@ -313,7 +302,6 @@ impl NonASAPOp { .collect(); Project { cols, - qualifier: qualifier.clone(), child: f(child), } } @@ -596,16 +584,11 @@ impl NonASAPOp { .. } => { if let [AggIntent::TopK { k, .. }] = measures.as_slice() { - let name = output_names - .first() - .filter(|name| !name.is_empty()) - .cloned() - .unwrap_or_else(|| format!("topk_{k}")); + let name = crate::ir::schema::topk::topk_output_name(*k, output_names); let cols = crate::ir::schema::topk::ranked_row_projection(child, reduction, &name)?; return NonASAPOp::Project { cols, - qualifier: None, child: child.clone(), } .output_schema(); @@ -694,11 +677,7 @@ impl NonASAPOp { // keeps its field verbatim, so an `ExactAggregate` state column // can pass through a projection unchanged; any other expression // is typed against the input and must read plain values. - Project { - cols, - qualifier, - child, - } => { + Project { cols, child } => { let in_schema = child.schema(); let fields: Vec = cols .iter() @@ -706,9 +685,7 @@ impl NonASAPOp { .map(|(i, item)| { let mut field = match &item.expr { ScalarExpr::Column(id) if in_schema.fields.get(*id).is_some() => { - let mut f = in_schema.fields[*id].clone(); - f.table = None; - f + in_schema.fields[*id].clone() } expr => { let (dtype, nullable) = expr.scalar_type(in_schema)?; @@ -719,10 +696,7 @@ impl NonASAPOp { .alias .clone() .unwrap_or_else(|| default_proj_name(&item.expr, i, in_schema)); - Ok(match qualifier { - Some(q) => field.with_table(q), - None => field, - }) + Ok(field) }) .collect::, SchemaDerivationError>>()?; let time_index = fields.iter().position(|c| c.name == "ts"); @@ -1403,7 +1377,6 @@ mod tests { fn project(cols: Vec, child: NonASAPOp) -> NonASAPOp { NonASAPOp::Project { cols, - qualifier: None, child: node(child), } } diff --git a/crates/types/src/ir/operator/operator_properties.rs b/crates/types/src/ir/operator/operator_properties.rs index e2a6ef1da..a6c052e62 100644 --- a/crates/types/src/ir/operator/operator_properties.rs +++ b/crates/types/src/ir/operator/operator_properties.rs @@ -513,7 +513,7 @@ mod tests { use super::*; use crate::ir::schema::{DataType, Field, Schema}; - /// A key moves to the input's column of the same name and qualifier; a + /// A key moves to the input's column of the same name; a /// key that is missing or ambiguous there, or a `without` grouping, has no /// partitions over the input. #[test] @@ -534,11 +534,6 @@ mod tests { Reduction::by(vec![0]).partitions_over(&inner, &ambiguous), None ); - let qualified = Schema::new(vec![field("job").with_table("t")]); - assert_eq!( - Reduction::by(vec![0]).partitions_over(&inner, &qualified), - None - ); assert_eq!( Reduction::Reduce(GroupKeys::without(vec![0])).partitions_over(&inner, &input), None diff --git a/crates/types/src/ir/operator/tumbling_window_chain.rs b/crates/types/src/ir/operator/tumbling_window_chain.rs index add4528bf..8d1ea5811 100644 --- a/crates/types/src/ir/operator/tumbling_window_chain.rs +++ b/crates/types/src/ir/operator/tumbling_window_chain.rs @@ -272,7 +272,6 @@ fn named_as_chain( let mut schema = summary_agg_schema(child.schema(), family, reduction)?; for (field, kept) in schema.fields.iter_mut().zip(&chain.fields) { field.name = kept.name.clone(); - field.table = kept.table.clone(); } Ok((build, schema)) } diff --git a/crates/types/src/ir/properties/accuracy.rs b/crates/types/src/ir/properties/accuracy.rs index 45f50b6b9..1e704bc61 100644 --- a/crates/types/src/ir/properties/accuracy.rs +++ b/crates/types/src/ir/properties/accuracy.rs @@ -750,7 +750,6 @@ mod tests { else_expr: Some(Box::new(otherwise)), }, }], - qualifier: None, child: joined.clone(), })) }; diff --git a/crates/types/src/ir/properties/column_lineage.rs b/crates/types/src/ir/properties/column_lineage.rs new file mode 100644 index 000000000..28f754f61 --- /dev/null +++ b/crates/types/src/ir/properties/column_lineage.rs @@ -0,0 +1,335 @@ +//! Column lineage: where each output column of a memo group comes from. +//! Design: concepts doc `memo-groups-properties-requirements.md`, section 3, +//! "Column lineage". +//! +//! A group's lineage is derived once from its input groups' lineage, one +//! entry per output column in schema order. A column copied unchanged from a +//! source column (a `Scan`'s table or metric column), however renamed on the +//! way, has that source column's lineage; any other column is `Computed`. +//! It identifies columns where a schema field's name cannot: two columns of +//! one name from two join sides have different lineage, and a renamed column +//! keeps its lineage. + +use crate::ir::operator::asap::ASAPOp; +use crate::ir::operator::input::OperatorInput; +use crate::ir::operator::node::Operator; +use crate::ir::operator::non_asap::NonASAPOp; +use crate::ir::operator::operator_properties::{ + BinaryOpKind, GroupSide, JoinKind, PromQLVectorSetOpKind, Reduction, Source, +}; +use crate::ir::operator::AggIntent; +use crate::ir::scalar::{resolve_column_ref, ColumnRef, ProjectItem, ScalarExpr}; +use crate::ir::schema::Schema; +use serde::{Deserialize, Serialize}; + +/// A column of a source (a `Scan`'s table or metric), by name. +#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord, Serialize, Deserialize)] +#[serde(deny_unknown_fields)] +pub struct SourceColumn { + pub(crate) source: String, + pub(crate) column: String, +} + +#[cfg(any(test, feature = "test-support"))] +impl SourceColumn { + pub fn new(source: impl Into, column: impl Into) -> Self { + Self { + source: source.into(), + column: column.into(), + } + } +} + +/// Where an output column's values come from. +#[derive(Debug, Clone, PartialEq, Eq)] +pub enum ColumnLineage { + /// Copied unchanged from a column of a source. + SourceColumn(SourceColumn), + /// Computed by an operator: a measure, an expression, a summary state or readout. + Computed, +} + +impl Operator { + /// The lineage of each column of `output`, this operator's output + /// schema, from its inputs' lineage (`lineage_of`). + pub(crate) fn output_column_lineage( + &self, + output: &Schema, + lineage_of: &dyn Fn(&I) -> Vec, + ) -> Vec { + let lineage = match self { + Operator::NonASAP(op) => non_asap_lineage(op, output, lineage_of), + Operator::ASAP(op) => asap_lineage(op, output, lineage_of), + }; + debug_assert_eq!( + lineage.len(), + output.fields.len(), + "one lineage entry per output column of {}", + self.kind_name() + ); + lineage + } +} + +fn non_asap_lineage( + op: &NonASAPOp, + output: &Schema, + lineage_of: &dyn Fn(&I) -> Vec, +) -> Vec { + use NonASAPOp::*; + let computed = || vec![ColumnLineage::Computed; output.fields.len()]; + match op { + Scan { source, schema, .. } => { + let source = match source { + Source::TimeSeries { metric } => metric, + Source::Table { table_ref } => table_ref, + }; + schema + .fields + .iter() + .map(|field| { + ColumnLineage::SourceColumn(SourceColumn { + source: source.clone(), + column: field.name.clone(), + }) + }) + .collect() + } + Values { .. } | PromqlVectorFromScalar(_) => computed(), + Filter { child, .. } + | Sort { child, .. } + | Limit { child, .. } + | Dedup { child, .. } + | TimeRange { child, .. } + | TimeShift { child, .. } + | PromqlSubquery { child, .. } + | PromqlSeriesSample { child, .. } + | PromqlInfoEnrich { child, .. } => lineage_of(child), + // The relabeled column is written; every other column is copied. + PromqlRelabel { dst, child, .. } => { + let mut lineage = lineage_of(child); + match child.schema().column_id(dst) { + Some(written) => lineage[written] = ColumnLineage::Computed, + None => lineage.push(ColumnLineage::Computed), + } + lineage + } + Project { cols, child } => project_lineage(cols, &lineage_of(child)), + Aggregate { + reduction, + measures, + output_names, + child, + .. + } => { + if let [AggIntent::TopK { k, .. }] = measures.as_slice() { + let name = crate::ir::schema::topk::topk_output_name(*k, output_names); + return match crate::ir::schema::topk::ranked_row_projection(child, reduction, &name) + { + Ok(cols) => project_lineage(&cols, &lineage_of(child)), + Err(_) => computed(), + }; + } + aggregate_lineage(child, reduction, measures, output, lineage_of) + } + Join { + kind, left, right, .. + } => { + let mut lineage = lineage_of(left); + if !matches!(kind, JoinKind::Semi | JoinKind::Anti) { + lineage.extend(lineage_of(right)); + } + lineage + } + SetOp { left, right, .. } => agreed(&[lineage_of(left), lineage_of(right)]), + Concat { children } => agreed(&children.iter().map(lineage_of).collect::>()), + SQLWindowFunc { child, .. } => { + let mut lineage = lineage_of(child); + lineage.push(ColumnLineage::Computed); + lineage + } + // The output is the left side's columns, its value computed, then the + // right side's labels the match adds. A label that may come from + // either side keeps its lineage only if both sides agree. + BinaryOp { + operator, lhs, rhs, .. + } => { + let mut lineage = lineage_of(lhs); + if let Ok(value) = resolve_column_ref(&ColumnRef::SampleValue, lhs.schema()) { + lineage[value] = ColumnLineage::Computed; + } + let right = lineage_of(rhs); + let right_named = |name: &str| { + rhs.schema() + .column_id(name) + .map_or(ColumnLineage::Computed, |column| right[column].clone()) + }; + let right_rows = matches!(operator.kind, BinaryOpKind::Set(PromQLVectorSetOpKind::Or)) + || operator + .vector_match + .as_ref() + .and_then(|m| m.grouping.as_ref()) + .is_some_and(|grouping| grouping.side == GroupSide::Right); + if right_rows { + for (column, field) in lhs.schema().fields.iter().enumerate() { + if right_named(&field.name) != lineage[column] { + lineage[column] = ColumnLineage::Computed; + } + } + } + let added = output.fields.get(lineage.len()..).unwrap_or_default(); + lineage.extend(added.iter().map(|field| right_named(&field.name))); + lineage + } + } +} + +fn asap_lineage( + op: &ASAPOp, + output: &Schema, + lineage_of: &dyn Fn(&I) -> Vec, +) -> Vec { + use ASAPOp::*; + match op { + // Laid out as an aggregate of one measure, the state in its place. + SummaryAgg { + child, reduction, .. + } + | SummaryTumblingWindowChain { + child, reduction, .. + } => aggregate_lineage( + child, + reduction, + &[AggIntent::Sum { col: None }], + output, + lineage_of, + ), + SummaryEstimate { + summary_input: child, + .. + } + | FinalizeExactAccumulator { child } => readout_lineage(child, output, lineage_of), + MaintainPopulation { child, .. } => lineage_of(child), + EvaluatePopulation { child, .. } if output == child.schema() => lineage_of(child), + SummaryMerge { + children, group_by, .. + } => { + let merged = agreed(&children.iter().map(lineage_of).collect::>()); + match group_by { + None => merged, + Some(keys) => keys + .iter() + .map(|&key| merged.get(key).cloned().unwrap_or(ColumnLineage::Computed)) + .chain([ColumnLineage::Computed]) + .collect(), + } + } + EvaluatePopulation { .. } + | SummarySubtract { .. } + | SummaryDelete { .. } + | SummaryJoin { .. } + | Extension { .. } => vec![ColumnLineage::Computed; output.fields.len()], + } +} + +/// A column item, renamed or not, keeps its input column's lineage; an +/// expression item is computed. +fn project_lineage(cols: &[ProjectItem], input: &[ColumnLineage]) -> Vec { + cols.iter() + .map(|item| match item.expr { + ScalarExpr::Column(column) => input + .get(column) + .cloned() + .unwrap_or(ColumnLineage::Computed), + _ => ColumnLineage::Computed, + }) + .collect() +} + +/// An aggregate's (`aggregate_output_schema`'s layout): a per-entity +/// reduction keeps its input's columns in place, its value computed; a +/// `by` reduction lists its keys, then its measures; a `without` reduction +/// the input's other label columns, then its measures. +fn aggregate_lineage( + child: &I, + reduction: &Reduction, + measures: &[AggIntent], + output: &Schema, + lineage_of: &dyn Fn(&I) -> Vec, +) -> Vec { + let input = lineage_of(child); + let schema = child.schema(); + let sample_value = resolve_column_ref(&ColumnRef::SampleValue, schema).ok(); + let keys: Vec = match reduction { + Reduction::PerEntity => { + let mut lineage = input; + let value = measures + .first() + .and_then(|measure| measure.input_cols().first().copied()) + .or(sample_value); + if let Some(value) = value.and_then(|value| lineage.get_mut(value)) { + *value = ColumnLineage::Computed; + } + return lineage; + } + Reduction::Reduce(by) if by.is_without() => (0..schema.fields.len()) + .filter(|&column| { + schema.time_index != Some(column) + && sample_value != Some(column) + && !by.keys().contains(&column) + }) + .map(|column| input[column].clone()) + .collect(), + Reduction::Reduce(by) => by + .keys() + .iter() + .map(|&key| input.get(key).cloned().unwrap_or(ColumnLineage::Computed)) + .collect(), + }; + let measured = output.fields.len().saturating_sub(keys.len()); + keys.into_iter() + .chain(std::iter::repeat_n(ColumnLineage::Computed, measured)) + .collect() +} + +/// A readout of summary state: the state's plain (key) columns keep their +/// lineage in place; the value read out is computed. +fn readout_lineage( + state: &I, + output: &Schema, + lineage_of: &dyn Fn(&I) -> Vec, +) -> Vec { + let input = state.schema(); + if input.fields.len() != output.fields.len() { + return vec![ColumnLineage::Computed; output.fields.len()]; + } + lineage_of(state) + .into_iter() + .zip(input.fields.iter().zip(&output.fields)) + .map(|(lineage, (at_state, at_output))| { + if at_state.is_plain() && at_output.is_plain() { + lineage + } else { + ColumnLineage::Computed + } + }) + .collect() +} + +/// Per column, the inputs' lineage if they all agree, else `Computed`. +fn agreed(inputs: &[Vec]) -> Vec { + let Some((first, rest)) = inputs.split_first() else { + return Vec::new(); + }; + first + .iter() + .enumerate() + .map(|(column, lineage)| { + if rest.iter().all(|other| other.get(column) == Some(lineage)) { + lineage.clone() + } else { + ColumnLineage::Computed + } + }) + .collect() +} diff --git a/crates/types/src/ir/properties/mod.rs b/crates/types/src/ir/properties/mod.rs index c7730ba8f..0da36a16c 100644 --- a/crates/types/src/ir/properties/mod.rs +++ b/crates/types/src/ir/properties/mod.rs @@ -8,7 +8,9 @@ //! - [`execution_data_state`] — when a node's value is produced //! ([`ExecutionTiming`]) and what its edge carries ([`ExecutionDataState`]). //! - [`summary_coverage`] — what a summary state covers. +//! - [`column_lineage`] — where each output column of a memo group comes from. pub mod accuracy; +pub mod column_lineage; pub mod execution_data_state; pub mod guarantee; pub mod summary_coverage; diff --git a/crates/types/src/ir/properties/summary_coverage.rs b/crates/types/src/ir/properties/summary_coverage.rs index 0e2dec55e..63a5945ef 100644 --- a/crates/types/src/ir/properties/summary_coverage.rs +++ b/crates/types/src/ir/properties/summary_coverage.rs @@ -9,7 +9,8 @@ //! predicate conjunct moves into `selection` when it can be lifted to the //! `SummaryAgg` (through `Filter`, a range `TimeRange`, a `TimeShift` without //! `@` and direct-column `Project` items) and it is a value set or an -//! interval on one column. Everything else stays in `definition` as a +//! interval on one column with a source column's lineage (a selection names +//! the column by its lineage). Everything else stays in `definition` as a //! residual, so two states are merged only when they compute the same thing //! over disjoint rows. Node trees carry no coverage. use std::cmp::Ordering; @@ -25,11 +26,12 @@ use crate::ir::operator::node::{Operator, OperatorResultKind}; use crate::ir::operator::non_asap::{NonASAPOp, TimeRangeKind}; use crate::ir::operator::operator_properties::Reduction; use crate::ir::operator::tumbling_window_chain::window_build; +use crate::ir::properties::column_lineage::{ColumnLineage, SourceColumn}; use crate::ir::scalar::expr_ir::{CompareOpKind, ScalarValue}; use crate::ir::scalar::{Predicate, ScalarExpr}; use crate::ir::schema::family_capability::{FamilyCapability, MergeInputRule}; use crate::ir::schema::sketch::GroupingStrategy; -use crate::ir::schema::{ColumnId, Schema}; +use crate::ir::schema::ColumnId; use crate::ir::SchemaDerivationError; use crate::stages::{MemoGroupAsInput, MemoGroupId}; @@ -79,7 +81,7 @@ impl SummaryCoverage { #[serde(deny_unknown_fields)] pub struct SelectionBox { #[serde(with = "columns_as_list")] - pub(crate) columns: BTreeMap, + pub(crate) columns: BTreeMap, /// Offsets from the evaluation time in milliseconds: a PromQL range /// `TimeRange(w)` over `TimeShift(s)` is `(-(s + w), -s]`. Absolute /// time will be an ordinary interval on the timestamp column once the IR @@ -90,19 +92,19 @@ pub struct SelectionBox { /// `SelectionBox::columns` as a list of `{column, constraint}`. mod columns_as_list { - use super::{ColumnIdentity, Constraint}; + use super::{Constraint, SourceColumn}; use serde::{Deserialize, Deserializer, Serialize, Serializer}; use std::collections::BTreeMap; #[derive(Serialize, Deserialize)] #[serde(deny_unknown_fields)] struct Entry { - column: ColumnIdentity, + column: SourceColumn, constraint: Constraint, } pub(super) fn serialize( - columns: &BTreeMap, + columns: &BTreeMap, serializer: S, ) -> Result { let entries: Vec = columns @@ -117,7 +119,7 @@ mod columns_as_list { pub(super) fn deserialize<'de, D: Deserializer<'de>>( deserializer: D, - ) -> Result, D::Error> { + ) -> Result, D::Error> { let entries = Vec::::deserialize(deserializer)?; Ok(entries .into_iter() @@ -182,26 +184,6 @@ mod relative_time_bounds { } } -/// A column of the definition child's output, by `(table, name)`. A -/// `Project` below renames its columns and sets their table to its -/// qualifier (none by default). -#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord, Serialize, Deserialize)] -#[serde(deny_unknown_fields)] -pub struct ColumnIdentity { - pub(crate) table: Option, - pub(crate) name: String, -} - -#[cfg(any(test, feature = "test-support"))] -impl ColumnIdentity { - pub fn new(table: Option, name: impl Into) -> Self { - Self { - table, - name: name.into(), - } - } -} - #[derive(Debug, Clone, PartialEq, Serialize, Deserialize)] #[serde(rename_all = "snake_case")] pub enum Constraint { @@ -517,8 +499,11 @@ fn regrouped( return Err(CoverageError::FamilyCannotRegroup); } let state = first.schema(); - let rows = build_input_schema(first).ok_or(CoverageError::NotSummary)?; - // The build's input column each key of the state names. + let state_lineage = first.column_lineage(); + let rows = build_input(first).ok_or(CoverageError::NotSummary)?; + let rows_lineage = rows.column_lineage(); + // The build's input column each key of the state names: the column of + // the key's lineage. let mapped = keys .iter() .map(|&key| { @@ -530,20 +515,15 @@ fn regrouped( }; column .filter(|&column| { - let (Some(at_state), Some(at_input)) = - (state.fields.get(key), rows.fields.get(column)) - else { - return false; - }; - at_state.is_plain() - && at_state.name == at_input.name - && at_state.table == at_input.table + state.fields.get(key).is_some_and(|field| field.is_plain()) + && state_lineage.get(key).is_some() + && state_lineage.get(key) == rows_lineage.get(column) }) .ok_or(CoverageError::NotAGroupKey { key }) }) .collect::, _>>()?; let reduction = Reduction::by(mapped); - summary_agg_schema(&rows, family, &reduction).map_err(|error| { + summary_agg_schema(rows.schema(), family, &reduction).map_err(|error| { CoverageError::RegroupedBuild { reason: error.to_string(), } @@ -564,21 +544,19 @@ fn regrouped( }) } -/// The schema of the rows the build of summary state `state` reads (its +/// The input group of the rows the build of summary state `state` reads (its /// `SummaryAgg`'s input; one build over a chain's window reads its range of -/// the chain's input, which has the input's schema). `None` when `state` -/// is no summary state. -fn build_input_schema(state: &I) -> Option { +/// the chain's input, which has the input's schema and lineage). `None` when +/// `state` is no summary state. +fn build_input<'m>(state: &MemoGroupAsInput<'m>) -> Option> { if let Some(states) = selected_from(state) { - return build_input_schema(&states); + return build_input(&states); } match state.operator() { Operator::ASAP( ASAPOp::SummaryAgg { child, .. } | ASAPOp::SummaryTumblingWindowChain { child, .. }, - ) => Some(child.schema().clone()), - Operator::ASAP(ASAPOp::SummaryMerge { children, .. }) => { - build_input_schema(children.first()?) - } + ) => Some(child), + Operator::ASAP(ASAPOp::SummaryMerge { children, .. }) => build_input(children.first()?), _ => None, } } @@ -653,19 +631,19 @@ fn of_summary_agg(build: &ASAPOp>) -> Result>>> = chain .iter() .enumerate() .map(|(depth, link)| match link { Operator::NonASAP(NonASAPOp::Filter { pred, .. }) => { - residual(pred, top, &chain[..depth], &mut lifted) + residual(pred, &top, &chain[..depth], &mut lifted) } _ => None, }) @@ -682,7 +660,7 @@ fn of_summary_agg(build: &ASAPOp>) -> Result>) -> Result>) -> Result>, - top: &Schema, + top: &[ColumnLineage], above: &[Operator>], lifted: &mut BTreeMap, ) -> Option>> { @@ -771,13 +747,10 @@ fn residual( for conjunct in pred.0.conjuncts() { let lift = constraint_of(conjunct).and_then(|(column, constraint)| { let column = column_at_top(column, above)?; - let field = &top.fields[column]; - let namesakes = top - .fields - .iter() - .filter(|f| f.name == field.name && f.table == field.table) - .count(); - if namesakes != 1 { + let lineage = &top[column]; + if *lineage == ColumnLineage::Computed + || top.iter().filter(|other| *other == lineage).count() != 1 + { return None; } let combined = match lifted.get(&column) { @@ -800,6 +773,17 @@ fn residual( } } +/// The column lineage of `input`: a memo group's stored lineage, or a +/// rule's temporary build's, derived from its inputs'. +fn column_lineage(input: &GroupInput<'_>) -> Vec { + match input { + InputOrTemporaryBuild::Input(group) => group.column_lineage().to_vec(), + InputOrTemporaryBuild::TemporaryBuild(_) => input + .operator() + .output_column_lineage(input.schema(), &column_lineage), + } +} + /// The memo group `input` is; `None` for a rule's temporary build. fn group_id(input: &GroupInput<'_>) -> Option { match input { @@ -1087,7 +1071,7 @@ impl Constraint { #[cfg(any(test, feature = "test-support"))] impl SelectionBox { pub fn new( - columns: BTreeMap, + columns: BTreeMap, relative_time: Option<(Bound, Bound)>, ) -> Self { Self { @@ -1097,7 +1081,7 @@ impl SelectionBox { } /// The per-column constraints. - pub fn columns(&self) -> &BTreeMap { + pub fn columns(&self) -> &BTreeMap { &self.columns } diff --git a/crates/types/src/ir/scalar/column_resolution.rs b/crates/types/src/ir/scalar/column_resolution.rs index 8822c6e85..e5bfc9d32 100644 --- a/crates/types/src/ir/scalar/column_resolution.rs +++ b/crates/types/src/ir/scalar/column_resolution.rs @@ -2,8 +2,9 @@ //! //! Front ends emit `ColumnRef` (name-based, optionally table-qualified); the //! IR uses positional [`ColumnId`] resolved against a per-node [`Schema`]. -//! These helpers turn name-based refs into positional ids, qualifier-aware. -//! Front-end name resolution (`asap_frontend_common::resolve`) calls them. +//! These helpers turn name-based refs into positional ids. A schema carries +//! no qualifier, so front-end name resolution (`asap_frontend_common::resolve`) +//! binds a qualified name with its own scope before it calls them. use thiserror::Error; @@ -22,38 +23,24 @@ pub enum ResolveError { NoSampleValue { available: Vec }, #[error("ColumnRef::Wildcard cannot be resolved to a single ColumnId")] WildcardNotPositional, - #[error("column `{name}` matches several unqualified columns")] + #[error("column `{name}` matches several columns")] Ambiguous { name: String }, } /// Resolve a single [`ColumnRef`] to a positional [`ColumnId`]. pub fn resolve_column_ref(col: &ColumnRef, schema: &Schema) -> Result { match col { - // A bare name must be unique: after a join `k` could be `a.k` or `b.k`. - ColumnRef::Named(name) => match schema.fields.iter().filter(|c| &c.name == name).count() { - 0 => Err(ResolveError::NotFound { - name: name.clone(), - available: schema.fields.iter().map(|c| c.name.clone()).collect(), - }), - 1 => Ok(schema.column_id(name).expect("one field has this name")), - _ => Err(ResolveError::Ambiguous { name: name.clone() }), - }, - // Prefer the (table, name) match; fall back to the bare name for - // schemas whose columns carry no qualifier, but only when that name - // is unique: after a join `b.k` must not bind to `a.k`. - ColumnRef::Qualified { table, name } => { - if let Some(id) = schema.column_id_qualified(table, name) { - return Ok(id); - } + // A name must be unique: after a join `k` could be either side's. + // A schema carries no qualifier; the front end binds a qualified + // name with its own scope before it calls this. + ColumnRef::Named(name) | ColumnRef::Qualified { name, .. } => { match schema.fields.iter().filter(|c| &c.name == name).count() { 0 => Err(ResolveError::NotFound { - name: format!("{table}.{name}"), + name: name.clone(), available: schema.fields.iter().map(|c| c.name.clone()).collect(), }), 1 => Ok(schema.column_id(name).expect("one field has this name")), - _ => Err(ResolveError::Ambiguous { - name: format!("{table}.{name}"), - }), + _ => Err(ResolveError::Ambiguous { name: name.clone() }), } } ColumnRef::SampleValue => schema diff --git a/crates/types/src/ir/scalar/expr_ir.rs b/crates/types/src/ir/scalar/expr_ir.rs index 50606f185..18cb41afd 100644 --- a/crates/types/src/ir/scalar/expr_ir.rs +++ b/crates/types/src/ir/scalar/expr_ir.rs @@ -18,9 +18,10 @@ use serde::{Deserialize, Serialize}; #[derive(Debug, Clone, PartialEq, Eq, Hash, Serialize, Deserialize)] pub enum ColumnRef { Named(String), - /// Table-qualified reference (`t.col` / `alias.col`). Resolved by - /// `(table, name)` so a column name shared across a join (`a.k` vs `b.k`) - /// binds to the correct side. + /// Table-qualified reference (`t.col` / `alias.col`), as a front end + /// emits it. The front end binds it by `(table, name)` with its own + /// scope while it lowers, so a column name shared across a join (`a.k` + /// vs `b.k`) binds to the correct side; a schema carries no qualifier. Qualified { table: String, name: String, diff --git a/crates/types/src/ir/scalar/mod.rs b/crates/types/src/ir/scalar/mod.rs index 1b4753b89..5eff1b016 100644 --- a/crates/types/src/ir/scalar/mod.rs +++ b/crates/types/src/ir/scalar/mod.rs @@ -839,7 +839,6 @@ mod tests { alias: Some("result".into()), expr, }], - qualifier: None, child: OperatorNode::new_shared(crate::ir::Operator::NonASAP(NonASAPOp::Scan { source: Source::Table { table_ref: "t".into(), diff --git a/crates/types/src/ir/schema/mod.rs b/crates/types/src/ir/schema/mod.rs index cce079379..3ae0481b7 100644 --- a/crates/types/src/ir/schema/mod.rs +++ b/crates/types/src/ir/schema/mod.rs @@ -42,8 +42,10 @@ use serde::{Deserialize, Serialize}; /// joins, and does not own data or prescribe a row/column-oriented storage layout. pub type ColumnId = usize; -/// One field of a [`Schema`]: `name + dtype + nullable`, plus an optional -/// table qualifier. The struct describes a column and holds none of its data. +/// One field of a [`Schema`]: `name + dtype + nullable`. The struct describes +/// a column and holds none of its data. Where a column's values come from is +/// its memo group's column lineage, not part of the field; a front end binds +/// qualified names (`t.col`) with its own scope while it lowers. /// /// `T` is the type vocabulary: [`FieldDataType`] on an operator edge (the /// default, and what [`Schema::fields`] holds), plain [`DataType`] for the @@ -59,46 +61,25 @@ pub struct Field { /// Whether NULL values are allowed in this field. PromQL value /// columns are non-nullable; SQL columns inherit their DDL nullability. pub nullable: bool, - /// Optional table/alias qualifier (SQL `t.col` / `t AS a` → `a`). Travels - /// with the field through joins so a `ColumnRef::Qualified` can pick the - /// right side when both carry the same `name`. `None` for PromQL labels and - /// unqualified columns. - #[serde(default)] - pub table: Option, } impl Field { - /// An unqualified field (`table = None`). pub fn new(name: impl Into, dtype: T, nullable: bool) -> Self { Self { name: name.into(), dtype, nullable, - table: None, } } - /// This field re-qualified under `table` (e.g. by a `SubqueryAlias`). - pub fn with_table(mut self, table: impl Into) -> Self { - self.table = Some(table.into()); - self - } - - /// A reference to this field: qualified by its table when it has one. + /// A reference to this field by its name. pub fn column_ref(&self) -> crate::ir::scalar::ColumnRef { - use crate::ir::scalar::ColumnRef; - match &self.table { - Some(table) => ColumnRef::Qualified { - table: table.clone(), - name: self.name.clone(), - }, - None => ColumnRef::Named(self.name.clone()), - } + crate::ir::scalar::ColumnRef::Named(self.name.clone()) } } impl Field { - /// An unqualified field carrying an ordinary readable value. + /// A field carrying an ordinary readable value. pub fn plain(name: impl Into, dtype: DataType, nullable: bool) -> Self { Self::new(name, FieldDataType::Plain(dtype), nullable) } @@ -135,7 +116,6 @@ impl From> for Field { name: field.name, dtype: FieldDataType::Plain(field.dtype), nullable: field.nullable, - table: field.table, } } } @@ -247,7 +227,7 @@ pub enum DataType { /// catalog tables. Date, /// Variable-length sequence. The existing column contract preserves the - /// element field name, type, and nullability. Nested fields are unqualified. + /// element field name, type, and nullability. List { element: Box> }, /// Ordered named fields, including each field's independent nullability. Struct { fields: Vec> }, @@ -312,7 +292,6 @@ impl Schema { field.name == PROMQL_SERIES_IDENTITY && field.dtype == DataType::Utf8 && !field.nullable - && field.table.is_none() }) } @@ -365,15 +344,6 @@ impl Schema { self.fields.iter().position(|c| c.name == name) } - /// Look up a field by `(table, name)` qualifier — disambiguates columns - /// that share a `name` across a join (`a.k` vs `b.k`). `None` if no field - /// has both that qualifier and name. - pub(crate) fn column_id_qualified(&self, table: &str, name: &str) -> Option { - self.fields - .iter() - .position(|c| c.name == name && c.table.as_deref() == Some(table)) - } - /// Whether this schema has *any* provable unique key. #[cfg(any(test, feature = "test-support"))] pub fn has_unique_key(&self) -> bool { @@ -454,12 +424,4 @@ mod tests { fn schema_without_fields_is_rejected() { assert!(serde_json::from_str::(r#"{"closed":true}"#).is_err()); } - - #[test] - fn qualified_field_serde_roundtrip() { - let c = col("service", DataType::Utf8).with_table("hosts"); - let back: Field = serde_json::from_str(&serde_json::to_string(&c).unwrap()).unwrap(); - assert_eq!(back, c); - assert_eq!(back.table.as_deref(), Some("hosts")); - } } diff --git a/crates/types/src/ir/schema/topk.rs b/crates/types/src/ir/schema/topk.rs index 6600fd1ca..f2e4cb0d7 100644 --- a/crates/types/src/ir/schema/topk.rs +++ b/crates/types/src/ir/schema/topk.rs @@ -55,6 +55,16 @@ pub fn ranked_column(child: &I) -> Result String { + output_names + .first() + .filter(|name| !name.is_empty()) + .cloned() + .unwrap_or_else(|| format!("topk_{k}")) +} + /// Ordered projection of ranked rows: partition keys, ranked value, then /// item identity. TopK retains rows; partition keys alone are not unique. pub(crate) fn ranked_row_projection( diff --git a/crates/types/src/stages/candidates.rs b/crates/types/src/stages/candidates.rs index 17d72ada0..ffc57336a 100644 --- a/crates/types/src/stages/candidates.rs +++ b/crates/types/src/stages/candidates.rs @@ -264,9 +264,13 @@ fn derive_group_properties( } Err(_) => None, }; + let column_lineage = operator.output_column_lineage(&schema, &|input: &MemoGroupAsInput| { + input.column_lineage().to_vec() + }); Ok(GroupProperties { schema, result_kind: operator.output_kind(), + column_lineage, summary_coverage, shareable_when_equal: operator.is_shareable_when_equal(), }) @@ -1370,7 +1374,6 @@ mod memo_tests { alias: None, expr: crate::ir::ScalarExpr::Column(0), }], - qualifier: None, child: input, }) } diff --git a/crates/types/src/stages/memo.rs b/crates/types/src/stages/memo.rs index d6842846f..294be5284 100644 --- a/crates/types/src/stages/memo.rs +++ b/crates/types/src/stages/memo.rs @@ -5,6 +5,7 @@ use std::collections::BTreeSet; use super::{CandidateForMemoGroup, Memo}; use crate::ir::operator::input::OperatorInput; +use crate::ir::properties::column_lineage::ColumnLineage; use crate::ir::properties::summary_coverage::SummaryCoverage; use crate::ir::schema::Schema; use crate::ir::{Operator, OperatorResultKind, QueryRoot}; @@ -43,8 +44,11 @@ pub struct MemoGroup { /// operator nodes. #[derive(Debug, Clone)] pub struct GroupProperties { + /// Fields: name, type, nullability; no table qualifier. pub schema: Schema, pub result_kind: OperatorResultKind, + /// One entry per output column, in schema order. + pub column_lineage: Vec, /// What a summary-state group covers; `None` for any other group. pub summary_coverage: Option, /// Whether two equal expressions return the same rows for the same @@ -64,6 +68,11 @@ impl<'m> MemoGroupAsInput<'m> { fn group(&self) -> &'m MemoGroup { &self.memo.groups()[self.group_id.0 as usize] } + + /// The group's `column_lineage`. + pub(crate) fn column_lineage(&self) -> &'m [ColumnLineage] { + &self.group().group_properties.column_lineage + } } impl OperatorInput for MemoGroupAsInput<'_> { diff --git a/crates/types/src/test_support.rs b/crates/types/src/test_support.rs index 720812263..684fa2433 100644 --- a/crates/types/src/test_support.rs +++ b/crates/types/src/test_support.rs @@ -102,16 +102,10 @@ pub fn population_over_selector( ) -> Rc { use crate::ir::operator::maintained_population::*; use crate::ir::operator::operator_properties::Source; - if let Some(NonASAPOp::Project { - cols, - qualifier, - child, - }) = root.non_asap() - { + if let Some(NonASAPOp::Project { cols, child }) = root.non_asap() { let child = population_over_selector(child, statistic.clone(), grouping); let project = Operator::NonASAP(NonASAPOp::Project { cols: cols.clone(), - qualifier: qualifier.clone(), child, }); return Rc::new(OperatorNode::with_schema(project, root.schema.clone())); diff --git a/crates/types/tests/column_lineage.rs b/crates/types/tests/column_lineage.rs new file mode 100644 index 000000000..110550056 --- /dev/null +++ b/crates/types/tests/column_lineage.rs @@ -0,0 +1,324 @@ +//! Column lineage, derived once per memo group from its input groups' +//! (concepts doc `memo-groups-properties-requirements.md`, section 3, +//! "Column lineage"): the doc's examples 1 to 3. Example 4 (PromQL label +//! matchers) is `scan_predicates_lift_and_their_states_merge` in +//! `memo_summary_coverage.rs`. +use asap_types::ir::operator::operator_properties::{JoinKind, Reduction, Source}; +use asap_types::ir::operator::AggIntent; +use asap_types::ir::properties::column_lineage::{ColumnLineage, SourceColumn}; +use asap_types::ir::properties::summary_coverage::Constraint; +use asap_types::ir::properties::AccuracyTarget; +use asap_types::ir::scalar::expr_ir::{CompareOpKind, ScalarValue}; +use asap_types::ir::scalar::ColumnRef; +use asap_types::ir::schema::{ + DataType, Field, FieldDataType, GroupingStrategy, Schema, SketchAlgorithm, SketchStatistic, + SummaryUpdate, +}; +use asap_types::ir::{ + ASAPOp, ExprSemantics, NonASAPOp, Operator, Predicate, ProjectItem, ScalarExpr, +}; +use asap_types::stages::{ + CandidateExpression, CandidateForMemoGroup, LogicalProperties, LogicalRequirements, Memo, + MemoGroupId, OriginOfCandidateForMemoGroup, +}; + +/// `Scan table` of Utf8 `columns`. +fn scan(table: &str, columns: &[&str]) -> Operator { + Operator::NonASAP(NonASAPOp::Scan { + source: Source::Table { + table_ref: table.into(), + }, + predicates: vec![], + schema: Schema::new( + columns + .iter() + .map(|name| Field::plain(*name, DataType::Utf8, false)) + .collect(), + ), + }) +} + +/// `column = value`. +fn equals(column: usize, value: &str) -> Predicate { + Predicate(ScalarExpr::Compare { + left: Box::new(ScalarExpr::Column(column)), + op: CompareOpKind::Eq, + right: Box::new(ScalarExpr::Literal(ScalarValue::Utf8(value.into()))), + semantics: ExprSemantics::Sql, + }) +} + +fn filter(pred: Predicate, child: MemoGroupId) -> Operator { + Operator::NonASAP(NonASAPOp::Filter { pred, child }) +} + +/// `SummaryAgg[HLL](by keys; column)`: the distinct values of `column`. +fn hll(child: MemoGroupId, keys: Vec, column: &str) -> Operator { + Operator::ASAP(ASAPOp::SummaryAgg { + child, + family: FieldDataType::Sketch(SketchAlgorithm::Hll, GroupingStrategy::default()), + input: SummaryUpdate::column(ColumnRef::Named(column.into())), + reduction: Reduction::by(keys), + grouping: GroupingStrategy::default(), + filter: None, + }) +} + +fn source_column(source: &str, column: &str) -> ColumnLineage { + ColumnLineage::SourceColumn(SourceColumn::new(source, column)) +} + +fn lineage(memo: &Memo, group: MemoGroupId) -> &[ColumnLineage] { + &memo.groups()[group.0 as usize] + .group_properties + .column_lineage +} + +/// Example 1: `SELECT f.src, approx_distinct(f.dst) FROM flows f WHERE +/// f.proto = 'tcp' GROUP BY f.src`. The group-by key keeps `flows.src` +/// through the aggregate, the HLL state and its readout; the readout +/// renamed to the aggregate's column name has the aggregate's schema, so +/// `add_candidate` accepts it. +#[test] +fn an_hll_readout_renamed_to_the_aggregate_column_is_its_candidate() { + let mut memo = Memo::default(); + let g1 = memo + .find_or_add_group(scan("flows", &["src", "dst", "proto", "ts"])) + .unwrap(); + let g2 = memo + .find_or_add_group(filter(equals(2, "tcp"), g1)) + .unwrap(); + let g3 = memo + .find_or_add_group(Operator::NonASAP(NonASAPOp::Aggregate { + reduction: Reduction::by(vec![0]), + measures: vec![AggIntent::Cardinality { + cols: vec![1], + accuracy: AccuracyTarget::max_error(0.01), + }], + output_names: vec!["approx_distinct(dst)".into()], + filters: vec![], + having: None, + child: g2, + })) + .unwrap(); + let g4 = memo.find_or_add_group(hll(g2, vec![0], "dst")).unwrap(); + let g5 = memo + .find_or_add_group(Operator::ASAP(ASAPOp::SummaryEstimate { + summary_input: g4, + query: SketchStatistic::Cardinality, + })) + .unwrap(); + let flows = ["src", "dst", "proto", "ts"].map(|column| source_column("flows", column)); + assert_eq!(lineage(&memo, g1), flows); + assert_eq!(lineage(&memo, g2), flows); + let key_then_computed = [source_column("flows", "src"), ColumnLineage::Computed]; + for group in [g3, g4, g5] { + assert_eq!(lineage(&memo, group), key_then_computed); + } + // The state's selection names the filtered column by its lineage. + let coverage = memo.groups()[g4.0 as usize] + .group_properties + .summary_coverage + .as_ref() + .expect("a summary state"); + assert_eq!( + coverage.selection()[0].columns()[&SourceColumn::new("flows", "proto")], + Constraint::In(vec![ScalarValue::Utf8("tcp".into())]) + ); + + let operator = Operator::NonASAP(NonASAPOp::Project { + cols: vec![ + ProjectItem { + alias: Some("src".into()), + expr: ScalarExpr::Column(0), + }, + ProjectItem { + alias: Some("approx_distinct(dst)".into()), + expr: ScalarExpr::Column(1), + }, + ], + child: g5, + }); + let expression = CandidateExpression { operator }; + let candidate = CandidateForMemoGroup { + logical_properties: LogicalProperties::new(&expression, None, None).unwrap(), + expression, + input_requirements: vec![LogicalRequirements::new(None)], + fused_groups: vec![], + origin: OriginOfCandidateForMemoGroup::Pass1, + }; + let id = memo.add_candidate(g3, candidate).unwrap(); + assert_eq!(id.group, g3); +} + +/// Example 2: `SELECT source, count(*) FROM (SELECT f.src AS source FROM +/// flows f) GROUP BY source`. The renaming `Project` keeps `flows.src`; a +/// state grouped by `source` has the key lineage of one grouped by `src` +/// over the scan, a selection on the renamed `area` names `flows.region`, +/// and the rollup's group-key check, which compares lineage, passes. +#[test] +fn a_rename_keeps_lineage_and_the_rollup_key_check_compares_it() { + let mut memo = Memo::default(); + let g1 = memo + .find_or_add_group(scan("flows", &["src", "dst", "region"])) + .unwrap(); + let renamed = memo + .find_or_add_group(Operator::NonASAP(NonASAPOp::Project { + cols: [("source", 0), ("dst", 1), ("area", 2)] + .map(|(alias, column)| ProjectItem { + alias: Some(alias.into()), + expr: ScalarExpr::Column(column), + }) + .to_vec(), + child: g1, + })) + .unwrap(); + assert_eq!( + lineage(&memo, renamed), + [ + source_column("flows", "src"), + source_column("flows", "dst"), + source_column("flows", "region"), + ] + ); + + let by_source = memo + .find_or_add_group(hll(renamed, vec![0], "dst")) + .unwrap(); + let by_src = memo.find_or_add_group(hll(g1, vec![0], "dst")).unwrap(); + assert_eq!(lineage(&memo, by_source)[0], lineage(&memo, by_src)[0]); + + let in_us = memo + .find_or_add_group(filter(equals(2, "us"), renamed)) + .unwrap(); + let by_source_and_area = memo + .find_or_add_group(hll(in_us, vec![0, 2], "dst")) + .unwrap(); + let coverage = memo.groups()[by_source_and_area.0 as usize] + .group_properties + .summary_coverage + .as_ref() + .expect("a summary state"); + assert_eq!( + coverage.selection()[0].columns()[&SourceColumn::new("flows", "region")], + Constraint::In(vec![ScalarValue::Utf8("us".into())]) + ); + + let rolled_up = memo + .find_or_add_group(Operator::ASAP(ASAPOp::SummaryMerge { + children: vec![by_source_and_area], + newest_tumbling_windows: None, + group_by: Some(vec![0]), + })) + .unwrap(); + let rolled_up = &memo.groups()[rolled_up.0 as usize].group_properties; + assert_eq!( + rolled_up.column_lineage, + [source_column("flows", "src"), ColumnLineage::Computed] + ); + assert!(rolled_up.summary_coverage.is_some()); +} + +/// Example 3: `SELECT s.region, count(*) FROM sales s JOIN users u ON +/// s.user = u.id GROUP BY s.region`. Both join sides have a `region` +/// column; their lineage tells them apart, so selections on them are +/// different columns. +#[test] +fn join_columns_of_one_name_have_distinct_lineage() { + let mut memo = Memo::default(); + let sales = memo + .find_or_add_group(scan("sales", &["user", "region"])) + .unwrap(); + let users = memo + .find_or_add_group(scan("users", &["id", "region"])) + .unwrap(); + let joined = memo + .find_or_add_group(Operator::NonASAP(NonASAPOp::Join { + kind: JoinKind::Inner, + pred: Predicate(ScalarExpr::Compare { + left: Box::new(ScalarExpr::Column(0)), + op: CompareOpKind::Eq, + right: Box::new(ScalarExpr::Column(2)), + semantics: ExprSemantics::Sql, + }), + left: sales, + right: users, + })) + .unwrap(); + assert_eq!( + lineage(&memo, joined), + [ + source_column("sales", "user"), + source_column("sales", "region"), + source_column("users", "id"), + source_column("users", "region"), + ] + ); + + let selected_column = |memo: &mut Memo, region_column: usize| { + let in_us = memo + .find_or_add_group(filter(equals(region_column, "us"), joined)) + .unwrap(); + let state = memo.find_or_add_group(hll(in_us, vec![], "id")).unwrap(); + let coverage = memo.groups()[state.0 as usize] + .group_properties + .summary_coverage + .as_ref() + .expect("a summary state"); + let [selection] = coverage.selection() else { + panic!("one box: {coverage:?}") + }; + selection.columns().keys().cloned().collect::>() + }; + assert_eq!( + selected_column(&mut memo, 1), + [SourceColumn::new("sales", "region")] + ); + assert_eq!( + selected_column(&mut memo, 3), + [SourceColumn::new("users", "region")] + ); +} + +/// A predicate on a computed column is not lifted: no selection can name +/// the column, so it stays in the definition's unlifted operators. +#[test] +fn a_predicate_on_a_computed_column_stays_unlifted() { + let mut memo = Memo::default(); + let g1 = memo + .find_or_add_group(scan("flows", &["src", "dst"])) + .unwrap(); + let computed = memo + .find_or_add_group(Operator::NonASAP(NonASAPOp::Project { + cols: vec![ + ProjectItem { + alias: Some("src".into()), + expr: ScalarExpr::Column(0), + }, + ProjectItem { + alias: Some("tag".into()), + expr: ScalarExpr::Literal(ScalarValue::Utf8("x".into())), + }, + ], + child: g1, + })) + .unwrap(); + assert_eq!( + lineage(&memo, computed), + [source_column("flows", "src"), ColumnLineage::Computed] + ); + let tagged = memo + .find_or_add_group(filter(equals(1, "x"), computed)) + .unwrap(); + let state = memo.find_or_add_group(hll(tagged, vec![], "src")).unwrap(); + let coverage = memo.groups()[state.0 as usize] + .group_properties + .summary_coverage + .as_ref() + .expect("a summary state"); + assert!(coverage.selection()[0].columns().is_empty()); + assert!(coverage + .definition() + .unlifted_operators + .contains(&filter(equals(1, "x"), MemoGroupId(0)).map_children(|_| None))); +} diff --git a/crates/types/tests/main.rs b/crates/types/tests/main.rs index 8c1477bf9..91b007dfc 100644 --- a/crates/types/tests/main.rs +++ b/crates/types/tests/main.rs @@ -2,6 +2,7 @@ //! its own copy of every dependency, which outgrew CI's disk. //! One file: `cargo test -p asap-types --test types ::`. +mod column_lineage; mod memo_summary_coverage; mod physical_export; mod planner_vocabulary; diff --git a/crates/types/tests/memo_summary_coverage.rs b/crates/types/tests/memo_summary_coverage.rs index cf96a6c45..7f0ca1387 100644 --- a/crates/types/tests/memo_summary_coverage.rs +++ b/crates/types/tests/memo_summary_coverage.rs @@ -12,9 +12,8 @@ use std::ops::Bound; use std::time::Duration; use asap_types::ir::operator::operator_properties::{GroupKeys, Reduction, Source}; -use asap_types::ir::properties::summary_coverage::{ - ColumnIdentity, Constraint, CoverageError, SummaryCoverage, -}; +use asap_types::ir::properties::column_lineage::{ColumnLineage, SourceColumn}; +use asap_types::ir::properties::summary_coverage::{Constraint, CoverageError, SummaryCoverage}; use asap_types::ir::scalar::expr_ir::{ArithmeticOpKind, CompareOpKind, ScalarValue}; use asap_types::ir::scalar::ColumnRef; use asap_types::ir::schema::{ @@ -213,7 +212,7 @@ fn a_summary_agg_over_a_job_filter_lifts_it_into_the_selection() { panic!("one box: {coverage:?}") }; assert_eq!( - selection.columns()[&ColumnIdentity::new(None, "job")], + selection.columns()[&SourceColumn::new("latency", "job")], jobs(&["api"]) ); assert_eq!(selection.relative_time(), last_minute()); @@ -242,7 +241,7 @@ fn states_with_equal_definitions_and_disjoint_selections_merge() { panic!("one box: {coverage:?}") }; assert_eq!( - selection.columns()[&ColumnIdentity::new(None, "job")], + selection.columns()[&SourceColumn::new("latency", "job")], jobs(&["api", "web"]) ); } @@ -288,7 +287,7 @@ fn a_filter_that_is_no_selection_stays_unlifted() { ] ); assert_eq!( - coverage_of_g10.selection()[0].columns()[&ColumnIdentity::new(None, "job")], + coverage_of_g10.selection()[0].columns()[&SourceColumn::new("latency", "job")], jobs(&["api"]) ); assert!(matches!( @@ -343,19 +342,28 @@ fn states_whose_definitions_read_the_same_subquery_merge() { assert_eq!(definition, coverage(&memo, web).definition()); let merged = memo.find_or_add_group(merge(vec![api, web])).unwrap(); assert_eq!( - coverage(&memo, merged).selection()[0].columns()[&ColumnIdentity::new(None, "job")], + coverage(&memo, merged).selection()[0].columns()[&SourceColumn::new("latency", "job")], jobs(&["api", "web"]) ); } /// PromQL label matchers are the scan's predicates: they are lifted too, so /// `latency{job="api"}` and `latency{job="web"}` states merge. The scan, -/// without them, is the last unlifted operator. +/// without them, is the last unlifted operator. Each scan's `job` column has +/// lineage `latency.job`, so the selections `latency.job ∈ {api}` and +/// `latency.job ∈ {web}` are disjoint selections on one column (concepts doc, +/// section 3, column lineage example 4). #[test] fn scan_predicates_lift_and_their_states_merge() { let mut memo = Memo::default(); let state_of_job = |memo: &mut Memo, job: &str| { let series = memo.find_or_add_group(scan(vec![job_is(job)])).unwrap(); + assert_eq!( + memo.groups()[series.0 as usize] + .group_properties + .column_lineage[JOB], + ColumnLineage::SourceColumn(SourceColumn::new("latency", "job")) + ); let last_minute = memo.find_or_add_group(range(MINUTE_MS, series)).unwrap(); memo.find_or_add_group(kll(last_minute)).unwrap() }; @@ -368,12 +376,12 @@ fn scan_predicates_lift_and_their_states_merge() { vec![kll_unlifted(), scan(vec![]).map_children(|_| None)] ); assert_eq!( - coverage(&memo, api).selection()[0].columns()[&ColumnIdentity::new(None, "job")], + coverage(&memo, api).selection()[0].columns()[&SourceColumn::new("latency", "job")], jobs(&["api"]) ); let merged = memo.find_or_add_group(merge(vec![api, web])).unwrap(); assert_eq!( - coverage(&memo, merged).selection()[0].columns()[&ColumnIdentity::new(None, "job")], + coverage(&memo, merged).selection()[0].columns()[&SourceColumn::new("latency", "job")], jobs(&["api", "web"]) ); } diff --git a/crates/types/tests/physical_export.rs b/crates/types/tests/physical_export.rs index 10a46370a..7324878b5 100644 --- a/crates/types/tests/physical_export.rs +++ b/crates/types/tests/physical_export.rs @@ -176,7 +176,6 @@ fn a_reference_read_twice_is_one_scalar_ref_edge() { semantics: ExprSemantics::Promql, }, }], - qualifier: None, child: series("a"), })) .unwrap(); diff --git a/crates/types/tests/schema_rebuilding.rs b/crates/types/tests/schema_rebuilding.rs index 611b0b316..745c1df65 100644 --- a/crates/types/tests/schema_rebuilding.rs +++ b/crates/types/tests/schema_rebuilding.rs @@ -65,7 +65,6 @@ fn rebuilding_preserves_only_explicit_naming_overrides() { let original = aggregate(scan(DataType::Int64, "key"), asap); let mut schema = original.schema.clone(); schema.fields[0].name = "alias".into(); - schema.fields[0].table = Some("result".into()); let renamed = OperatorNode::with_schema(original.operator.clone(), schema) .with_guarantee(Some(AccuracyGuarantee::exact("fixture"))); let original = Rc::new(renamed); @@ -73,7 +72,6 @@ fn rebuilding_preserves_only_explicit_naming_overrides() { let replacement = scan(DataType::Utf8, "new_key"); let rebuilt = Rc::new(original.with_new_children(|_| replacement.clone()).unwrap()); assert_eq!(rebuilt.schema.fields[0].name, "alias"); - assert_eq!(rebuilt.schema.fields[0].table.as_deref(), Some("result")); assert_eq!( rebuilt.schema.fields[0].plain_dtype(), Some(&DataType::Utf8) diff --git a/crates/types/tests/structure_contract.rs b/crates/types/tests/structure_contract.rs index 47ce849bc..2dcace30a 100644 --- a/crates/types/tests/structure_contract.rs +++ b/crates/types/tests/structure_contract.rs @@ -112,7 +112,6 @@ fn state_evaluations_and_passthrough_keep_their_contracts() { expr: ScalarExpr::Column(0), alias: None, }], - qualifier: None, })) .unwrap(); pass.validate_structure().unwrap(); @@ -166,7 +165,6 @@ fn shared_construction_derives_both_operator_categories() { expr: ScalarExpr::Column(99), alias: None }], - qualifier: None, })) .is_err() ); diff --git a/docs/design_docs/proposals/asap-primitive-schema.md b/docs/design_docs/proposals/asap-primitive-schema.md index ac0c7c1ac..aa6dc15fd 100644 --- a/docs/design_docs/proposals/asap-primitive-schema.md +++ b/docs/design_docs/proposals/asap-primitive-schema.md @@ -48,8 +48,8 @@ This section fixes the words used below. They follow relational databases and Ap |---|---|---| | **Relation / table** | A set (bag) of rows with the same columns. A base table is stored; a derived relation is the output of a query operator. | Every edge in the DAG carries a relation. A `Scan` reads a base table (SQL table or PromQL metric); every other operator outputs a derived relation. | | **Row / tuple** | One element of a relation: one value per column. | One output row of a node. For PromQL, one sample of one series at one time. | -| **Column** | One position in every row, with a name and a type. Qualified as `table.column` when names can collide (DataFusion `Column { relation, name }`). | `ColumnId` refers to a column of the input schema; `(table, name)` identifies it across nodes (§4.2.2). | -| **Schema** | The ordered list of columns of a relation: name, data type, nullability (Arrow `Schema` of `Field { name, data_type, nullable }`; DataFusion `DFSchema` adds the table qualifier). The schema is *metadata*: it describes rows, it contains none. | `Schema` of `Field { name, dtype, nullable, table }` in `crates/types/src/ir/schema/mod.rs`. Unlike Arrow, `dtype` can be a summary state type (§3). | +| **Column** | One position in every row, with a name and a type. Qualified as `table.column` when names can collide (DataFusion `Column { relation, name }`). | `ColumnId` refers to a column of the input schema; its memo group's column lineage, the source column `(source, column)` its values are copied from, identifies it across nodes (§4.2.2). A front end binds `table.column` with its own scope while it lowers. | +| **Schema** | The ordered list of columns of a relation: name, data type, nullability (Arrow `Schema` of `Field { name, data_type, nullable }`; DataFusion `DFSchema` adds the table qualifier). The schema is *metadata*: it describes rows, it contains none. | `Schema` of `Field { name, dtype, nullable }` in `crates/types/src/ir/schema/mod.rs`, with no table qualifier. Unlike Arrow, `dtype` can be a summary state type (§3). | | **Data type** | The type of a column's values (`Int64`, `Utf8`, `Timestamp`, …). | `DataType`, wrapped as `FieldDataType::Plain`. | | **Aggregate state** | The intermediate value of an aggregate function before its final result, e.g. `(sum, count)` for `AVG` (DataFusion `Accumulator::state`, partial/final aggregation). It is never exposed as a column type to users. | Summary state *is* a column type here (`FieldDataType::Sketch`, `ExactAggregate`, …), so state can flow along edges and be merged, stored and read by later operators. | | **View / materialized view** | A view is a named query (its *definition*). A materialized view also stores the query's result rows; a query can then be answered from it when its definition matches (view matching, §4.1). | A built summary state is a materialized aggregation view whose aggregate is a summary family. Its definition and which rows it took are its coverage (§4). | @@ -216,9 +216,9 @@ With an error target, the planner answers each quantile from a KLL state. All th | State | Sub-DAG | `definition` | `selection` | |---|---|---|---| -| `S_us` | `KLL(latency) by[job]` over `Filter(region = 'us', Scan t)` | `KLL(latency) by[job]` over `Scan t` | `region ∈ {us}` | -| `S_eu` | `KLL(latency) by[job]` over `Filter(region = 'eu', Scan t)` | `KLL(latency) by[job]` over `Scan t` | `region ∈ {eu}` | -| `S_size` | `KLL(size) by[job]` over `Filter(region = 'eu', Scan t)` | `KLL(size) by[job]` over `Scan t` | `region ∈ {eu}` | +| `S_us` | `KLL(latency) by[job]` over `Filter(region = 'us', Scan t)` | `KLL(latency) by[job]` over `Scan t` | `t.region ∈ {us}` | +| `S_eu` | `KLL(latency) by[job]` over `Filter(region = 'eu', Scan t)` | `KLL(latency) by[job]` over `Scan t` | `t.region ∈ {eu}` | +| `S_size` | `KLL(size) by[job]` over `Filter(region = 'eu', Scan t)` | `KLL(size) by[job]` over `Scan t` | `t.region ∈ {eu}` | - `S_us + S_eu` ✓: same computation, different rows. The result is the KLL of latency by job for US and EU, with every row counted once. - `S_us + S_us` ✗: same computation, same rows. Every US row would be counted twice. @@ -293,11 +293,13 @@ G4.summary_coverage = SummaryCoverage { input_group: Some(G1), unlifted_operators: [SummaryAgg[KLL]], }, - selection: [ { job ∈ {api}, time (−60 s, 0] } ], + selection: [ { latency.job ∈ {api}, time (−60 s, 0] } ], } ``` -**Case 2, G7: the same definition, a disjoint selection.** The walk over G6 and G5 lifts `job = "web"` and the same minute, so G7's definition is G4's (`input_group: Some(G1)`, `unlifted_operators: [SummaryAgg[KLL]]`) and its `definition_id()` is the same; its selection is `{ job ∈ {web}, time (−60 s, 0] }`. The two selections are disjoint, so `SummaryMerge(G4, G7)` is a valid candidate (and group), covering `{ job ∈ {api, web}, time (−60 s, 0] }`. +A selection names a column by its lineage: `latency.job` is the `job` column of the series `latency` (see *Which column a condition is on* below). + +**Case 2, G7: the same definition, a disjoint selection.** The walk over G6 and G5 lifts `job = "web"` and the same minute, so G7's definition is G4's (`input_group: Some(G1)`, `unlifted_operators: [SummaryAgg[KLL]]`) and its `definition_id()` is the same; its selection is `{ latency.job ∈ {web}, time (−60 s, 0] }`. The two selections are disjoint, so `SummaryMerge(G4, G7)` is a valid candidate (and group), covering `{ latency.job ∈ {api, web}, time (−60 s, 0] }`. **Case 3, G10: `a + b > 10` cannot be lifted.** It spans two columns, so it is no selection. At G8 the walk lifts `job = "api"` and keeps the rest: @@ -307,7 +309,7 @@ G10.summary_coverage = SummaryCoverage { input_group: Some(G1), unlifted_operators: [SummaryAgg[KLL], Filter(a + b > 10)], }, - selection: [ { job ∈ {api}, time (−60 s, 0] } ], + selection: [ { latency.job ∈ {api}, time (−60 s, 0] } ], } ``` @@ -368,11 +370,11 @@ GROUP BY job; | Condition | Found at | Rule 1: same rows at the `SummaryAgg`? | Rule 2: simple? | Result | |---|---|---|---|---| -| `latency < 100` | `SummaryAgg.filter` | yes, it is already there | yes, an interval | `selection`: `latency ∈ (−∞, 100)` | -| `region = 'us'` | `Scan.predicates` | yes: `Project` passes `region` through (renamed `r`) | yes, a value set | `selection`: `r ∈ {us}` | +| `latency < 100` | `SummaryAgg.filter` | yes, it is already there | yes, an interval | `selection`: `t.latency ∈ (−∞, 100)` | +| `region = 'us'` | `Scan.predicates` | yes: `Project` passes `region` through (renamed `r`) | yes, a value set | `selection`: `t.region ∈ {us}` | | `latency * 2 > 10` | `Scan.predicates` | yes | **no**: it is on an expression, not a column | stays in `definition` | -So the `selection` is `latency ∈ (−∞, 100)`, `r ∈ {us}`. A PromQL example with a time window is under **Time** below. +So the `selection` is `t.latency ∈ (−∞, 100)`, `t.region ∈ {us}`. A PromQL example with a time window is under **Time** below. **Rule 1 in detail.** Imagine moving the condition up, one operator at a time, until it is just below the `SummaryAgg`. Every operator it passes must leave the picked rows unchanged. Whether it can pass depends on what the operator does: @@ -380,16 +382,16 @@ So the `selection` is `latency ∈ (−∞, 100)`, `r ∈ {us}`. A PromQL exampl |---|---|---|---| | another `Filter` (also `Scan.predicates`, `SummaryAgg.filter`) | yes | filters only drop rows, so their order does not matter | `region = 'us'` below `Filter(value < 100)` ✓ | | `TimeRange` (range) or `TimeShift` (without `@`) | yes | they choose a time window, but do not change any row's values | `region = 'us'` below `TimeRange(1m)` ✓ | -| `Project` | only if its column is passed through unchanged (a rename is fine) | the column must still be there, with the same values, above the `Project` | `Project [job, region AS r]`: `region = 'us'` ✓, it becomes `r = 'us'`. `Project [job, value * 2 AS v2]`: `value > 5` ✗, `value` is gone | +| `Project` | only if its column is passed through unchanged (a rename is fine) | the column must still be there, with the same values, above the `Project` | `Project [job, region AS r]`: `region = 'us'` ✓, on `r`, whose lineage is still `t.region`. `Project [job, value * 2 AS v2]`: `value > 5` ✗, `value` is gone | | `Aggregate` (later) | only if it uses group columns | a group column has one value per group, so filtering before or after grouping keeps the same groups | below `SUM(value) by job`: `job = 'api'` ✓; `value > 5` ✗, it changes the sums | | `rate`, a window function, or an instant `TimeRange` (later) | only if it uses series labels | a label is the same for every sample of a series | below `rate(...)`: `job = 'api'` ✓; `value > 5` ✗, dropping raw samples changes the rate. Below an instant selector: `job = 'api'` ✓; `value > 5` ✗, the latest sample with value > 5 is not the latest sample | | any other operator, e.g. `Join`, a vector/vector `BinaryOp`, `Limit`, a `TimeShift` with `@` | no | | | -A `Filter` *above* `rate` has nothing to pass: it keeps some rate outputs, and those are exactly the rows the `SummaryAgg` reads. How a PromQL comparison is lowered decides whether it is such a `Filter`: +A `Filter` *above* `rate` has nothing to pass: it keeps some rate outputs, and those are exactly the rows the `SummaryAgg` reads. But its condition is on the rate value, a computed column with no source column to name it by, so it stays in `definition` too. How a PromQL comparison is lowered decides whether it is such a `Filter`: | PromQL | Lowered to | Coverage | |---|---|---| -| `rate(m[5m]) > 0` (vector and scalar) | `Filter` | `selection`: `value ∈ (0, ∞)` | +| `rate(m[5m]) > 0` (vector and scalar) | `Filter` | stays in `definition`: `value` is computed by `rate` | | `rate(m[5m]) > bool 0` | `Project` (values become 0 or 1, no row is dropped) | stays in `definition`: it does not pick rows | | `rate(a[5m]) > rate(b[5m])` (vector and vector) | `BinaryOp` (matches the two sides by labels, like a join) | stays in `definition`: the walk does not enter a `BinaryOp` | @@ -415,11 +417,11 @@ Any other shape stays in `definition`: **Which column a condition is on.** -- A column is named by its table and name, `(table, name)`, **as the `SummaryAgg` reads it** (in the schema of its child). Directly above a join, `shipping.region = 'us'` and `billing.region = 'us'` are different conditions. -- A `Project` gives its columns new names and replaces their table with its own qualifier (none by default): after `region AS r`, the column is `(none, r)`, so the condition becomes `r ∈ {us}`. Since the `definition` contains the same `Project`, two states that rename the same way still compare equal. -- Only a `Project` may rename. Any other operator whose output names a column differently from its input (allowed for every operator, e.g. a `Filter` with an explicit schema) stays in `definition`, and so do the conditions at and below it. Removing it would drop the rename: if a `Filter` calls the `tier` column `region`, its `tier = 'eu'` would read as `region ∈ {eu}` and look disjoint from another state's `region ∈ {us}`, though a row with `region = 'us', tier = 'eu'` is in both. -- If two columns the `SummaryAgg` reads have the same `(table, name)` (for example `Project [a AS k, b AS k]`), a condition on `k` cannot tell them apart and stays in `definition`. -- PromQL labels have no table, so a label is named by its name alone. +- A column is named by its **lineage** ([concepts doc, section 3](../concepts/memo-groups-properties-requirements.md#column-lineage)) in the group the `SummaryAgg` reads: the source column `(source, column)` its values are copied from, a `Scan`'s table or metric and its column there. Directly above a join, `shipping.region = 'us'` and `billing.region = 'us'` are different conditions, though both columns are named `region`. +- A rename keeps lineage: after `Project [region AS r]`, `r` is still `t.region`, so the condition is `t.region ∈ {us}`. A memo group's schema is derived, so no other operator renames a column. Since the `definition` contains the same `Project`, two states that rename the same way still compare equal. +- A condition on a computed column (a `Project` expression item, an aggregate's measure, a readout's value) has no source column to name and stays in `definition`. +- If two columns the `SummaryAgg` reads have the same lineage (for example `Project [a AS k, a AS k2]`), the selection cannot name one of them apart, and their conditions stay in `definition`. +- A PromQL label is named by its metric and label: `latency.job`. - Values of different types are never treated as different: `1` and `1.0` might be equal, so `x = 1` and `x = 1.0` are treated as possibly overlapping. - A constant may come from a cast: type coercion turns `latency < 100` on a `DOUBLE` column into `latency < CAST(100 AS DOUBLE)`, and the SQL front end folds a numeric cast of a literal as DataFusion's constant folding does, so the condition reads `latency < 100.0`. @@ -440,11 +442,11 @@ Any other shape stays in `definition`: | Query | Sub-DAG below `SummaryAgg` | `definition` keeps | `selection` takes | |---|---|---|---| -| SQL `… FROM t WHERE region = 'us'` | `Scan t {region = 'us'}` | `Scan t` | `region ∈ {us}` | -| SQL `… FROM t WHERE latency < 100` | `Scan t {latency < 100}` | `Scan t` | `latency ∈ (−∞, 100)` | +| SQL `… FROM t WHERE region = 'us'` | `Scan t {region = 'us'}` | `Scan t` | `t.region ∈ {us}` | +| SQL `… FROM t WHERE latency < 100` | `Scan t {latency < 100}` | `Scan t` | `t.latency ∈ (−∞, 100)` | | PromQL `quantile_over_time(0.99, m[1m] offset 2m)` | `TimeRange(1m, TimeShift(2m, Scan m))` | `Scan m` | the last 3 to 2 minutes before evaluation, `(−3m, −2m]` | | PromQL `quantile(0.99, rate(m{job="api"}[5m]))` | `rate(TimeRange(5m, Scan m {job = 'api'}))` | all of it | nothing: `job = 'api'` is under `rate`, which the walk cannot pass yet (Rule 1, later) | -| PromQL `quantile(0.99, rate(m[5m]) > 0)` | `Filter(value > 0, rate(TimeRange(5m, Scan m)))` | `rate(TimeRange(5m, Scan m))` | `value ∈ (0, ∞)`: the `Filter` is above `rate` | +| PromQL `quantile(0.99, rate(m[5m]) > 0)` | `Filter(value > 0, rate(TimeRange(5m, Scan m)))` | all of it | nothing: the `Filter` is above `rate`, but `value` is computed by `rate`, so no selection can name it | | PromQL `quantile(0.99, rate(a[5m]) > rate(b[5m]))` | `BinaryOp(>, rate(TimeRange(5m, Scan a)), rate(TimeRange(5m, Scan b)))` | all of it | nothing: a vector/vector comparison is a `BinaryOp`, not a `Filter` | `Scan t {…}` is a `Scan` with those predicates. In the last three rows `TimeRange(5m)` stays in `definition` in any case: it is the input window of `rate` and changes the rate values, so it does not just pick rows. @@ -542,7 +544,7 @@ With an error target, the planner answers the quantile from a KLL sketch with `k coverage() of the SummaryAgg ┌──────────────────────────────────────────────────────────────────┐ │ definition: this SummaryAgg over Scan t (the Filter removed) │ -│ selection: region ∈ {us}, latency ∈ (−∞, 10000) │ +│ selection: t.region ∈ {us}, t.latency ∈ (−∞, 10000) │ └──────────────────────────────────────────────────────────────────┘ ``` @@ -781,13 +783,13 @@ The first two build `KLL(latency) by[job]` states. The third can be answered by │ State: job Utf8, state Sketch(KLL k=200) │ [[ SummaryMerge ]] - ▲ selection: region ∈ {us, eu} + ▲ selection: t.region ∈ {us, eu} │ ┌─────────────────┴─────────────────┐ │ │ [[ SummaryAgg ]] [[ SummaryAgg ]] KLL k=200, by job KLL k=200, by job - selection: region ∈ {us} selection: region ∈ {eu} + selection: t.region ∈ {us} selection: t.region ∈ {eu} ▲ ▲ │ │ [ Filter region = 'us' ] [ Filter region = 'eu' ] @@ -1128,9 +1130,6 @@ impl Schema { pub fn is_all_plain(&self) -> bool; /// Position of the first field with this name, or `None`. pub fn column_id(&self, name: &str) -> Option; - /// Position of the field with this table and name, or `None`. Tells apart - /// `a.k` and `b.k` after a join. - pub fn column_id_qualified(&self, table: &str, name: &str) -> Option; } /// One column of a `Schema`. `T` is `FieldDataType` on DAG edges, and plain @@ -1143,13 +1142,9 @@ pub struct Field { pub dtype: T, /// Whether the column may contain NULL. PromQL value columns are never NULL. pub nullable: bool, - /// Table or alias the column comes from (`t` in `t.col`), so two columns - /// with the same name from a join stay apart. `None` for PromQL labels and - /// unqualified columns. - pub table: Option, } impl Field { - /// An unqualified field holding a plain value of type `dtype`. + /// A field holding a plain value of type `dtype`. pub fn plain(name: impl Into, dtype: DataType, nullable: bool) -> Self; /// The value type if the field is plain; `None` for summary state. pub fn plain_dtype(&self) -> Option<&DataType>; @@ -1407,19 +1402,21 @@ pub struct SummaryCoverage { pub struct SelectionBox { /// One constraint per restricted column. A column not in the map is /// unrestricted. - pub columns: BTreeMap, + pub columns: BTreeMap, /// The time window relative to the evaluation time, in ms, e.g. /// `(Excluded(-120000), Included(-60000))` for `(−2m, −1m]`. /// `None`: no time restriction. pub relative_time: Option<(Bound, Bound)>, } -/// Names a column across nodes, independent of its position. -pub struct ColumnIdentity { - /// The table or alias the column comes from; `None` if unqualified. - pub table: Option, - /// The column name. - pub name: String, +/// Names a column across nodes, independent of its position and name: the +/// source column of its lineage (`ColumnLineage::SourceColumn`), the same +/// type the lineage holds. +pub struct SourceColumn { + /// The table or metric a `Scan` reads. + pub source: String, + /// The column's name in that source. + pub column: String, } /// A constraint on one column.