feat(query-expr): add opt-in Concat discriminator key

crates/asap-aware-mapping/src/explanation.rs

@@ -438,7 +438,7 @@ fn visit_children(
         | SQLWindowFunc { child, .. }
         | Sort { child, .. }
         | Limit { child, .. } => visit(child, format!("{label} > child"), locations),
-        Concat { children } => {
+        Concat { children, .. } => {
             for (i, c) in children.iter().enumerate() {
                 visit_children(c, &format!("{label} > concat[{i}]"), locations);
             }

crates/asap-aware-mapping/src/query_physical_lowering.rs

@@ -693,7 +693,7 @@ pub fn lower_query_physical_dag(
                     }
                     self.lower_unary(query, occurrence, PhysicalOperator::PassThrough, child)
                 }
-                QueryExpr::Concat { children } => {
+                QueryExpr::Concat { children, .. } => {
                     let child_ids = children
                         .iter()
                         .map(|child| self.lower(child))
@@ -1979,6 +1979,7 @@ mod tests {
 
         let concat = Rc::new(QueryExpr::Concat {
             children: vec![scan("a"), scan("b")],
+            discriminator_unique_key: None,
         });
         let dag = lower_query_physical_dag(&concat, &scope, &scripted(&provided)).unwrap();
         assert_eq!(dag.nodes.last().unwrap().operator, PhysicalOperator::Concat);

crates/asap-aware-mapping/src/replacement.rs

@@ -3597,6 +3597,7 @@ fn direct_child_counts(node: &QueryExpr) -> Vec<(*const QueryExpr, usize)> {
             }
             Concat {
                 children: concat_children,
+                ..
             } => {
                 for c in concat_children {
                     collect(c, children);
@@ -4183,7 +4184,7 @@ fn walk_children(
         | SQLWindowFunc { child, .. }
         | Sort { child, .. }
         | Limit { child, .. } => walk(child, order, nodes, counts),
-        Concat { children } => {
+        Concat { children, .. } => {
             for c in children {
                 walk_children(c, order, nodes, counts);
             }
@@ -5124,7 +5125,7 @@ mod tests {
                 | SQLWindowFunc { child, .. }
                 | Sort { child, .. }
                 | Limit { child, .. } => walk(child, counts),
-                Concat { children } => {
+                Concat { children, .. } => {
                     for c in children {
                         walk_children(c, counts);
                     }

crates/asap-aware-mapping/src/summary_maintenance_cost/model.rs

@@ -87,7 +87,7 @@ pub(super) fn query_source_selections(
         | PromqlSeriesSample { child, .. }
         | Sort { child, .. }
         | Limit { child, .. } => query_source_selections(child, out)?,
-        Concat { children } => {
+        Concat { children, .. } => {
             for child in children {
                 query_source_selections(child, out)?;
             }

crates/devtools/src/bin/variant_coverage.rs

@@ -88,7 +88,7 @@ fn walk(e: &QueryExpr, seen: &mut BTreeSet<&'static str>) {
             seen.insert("Dedup");
             walk(child, seen);
         }
-        QueryExpr::Concat { children } => {
+        QueryExpr::Concat { children, .. } => {
             seen.insert("Concat");
             children.iter().for_each(|c| walk(c, seen));
         }

crates/frontend-promql/src/promql.rs

@@ -742,7 +742,13 @@ fn walk_histogram_quantiles(call: &Call) -> Result<Unresolved> {
             })
         })
         .collect::<Result<Vec<_>>>()?;
-    Ok(Unresolved::Concat { children: branches })
+    // No discriminator asserted here today (issue #228): the φ value each
+    // branch carries via `PromqlRelabel` *is* structurally a distinct
+    // per-branch discriminator, but nothing downstream currently needs the
+    // resulting compound unique key — see
+    // `docs/design_docs/concat-unique-keys-decision.md`. `Unresolved::concat`
+    // keeps `output_schema`'s default (drop `unique_keys` entirely).
+    Ok(Unresolved::concat(branches))
 }
 
 /// Prometheus's `labels.FormatOpenMetricsFloat` — how `histogram_quantiles`

crates/frontend-promql/tests/observability/awesome_prometheus_alerts.rs

@@ -83,7 +83,7 @@ fn intents(e: &QueryExpr) -> Vec<AggIntent> {
                 go(lhs, out);
                 go(rhs, out);
             }
-            QueryExpr::Concat { children } => children.iter().for_each(|c| go(c, out)),
+            QueryExpr::Concat { children, .. } => children.iter().for_each(|c| go(c, out)),
             QueryExpr::PromqlVectorFromScalar(inner) | QueryExpr::PromqlScalarFromVector(inner) => {
                 go(inner, out)
             }

crates/frontend-promql/tests/promql_conformance.rs

@@ -92,7 +92,7 @@ fn collect(e: &QueryExpr, out: &mut Vec<AggIntent>) {
             collect(left, out);
             collect(right, out);
         }
-        QueryExpr::Concat { children } => children.iter().for_each(|c| collect(c, out)),
+        QueryExpr::Concat { children, .. } => children.iter().for_each(|c| collect(c, out)),
         QueryExpr::PromqlVectorFromScalar(inner) | QueryExpr::PromqlScalarFromVector(inner) => {
             collect(inner, out)
         }

crates/frontend-promql/tests/promql_lowering.rs

@@ -920,7 +920,7 @@ fn topk_over_bare_selector_ranks_raw_samples() {
 
 /// The `(label value, intent)` of each `histogram_quantiles` branch.
 fn quantile_branches(q: &QueryExpr) -> Vec<(String, AggIntent)> {
-    let QueryExpr::Concat { children } = q else {
+    let QueryExpr::Concat { children, .. } = q else {
         panic!("expected a Concat at the root, got {q:?}");
     };
     children
@@ -978,7 +978,7 @@ fn histogram_quantiles_branches_are_union_compatible() {
     // `Concat` derives its schema from the first child, so every branch must
     // agree on column names — the φ lives in the label, not the column name.
     let q = lower(r#"histogram_quantiles(testhistogram3, "q", 0.5, 0.9)"#);
-    let QueryExpr::Concat { children } = &q else {
+    let QueryExpr::Concat { children, .. } = &q else {
         panic!("expected Concat");
     };
     let shapes: Vec<Vec<String>> = children
@@ -1004,7 +1004,7 @@ fn histogram_quantiles_branches_are_union_compatible() {
 #[test]
 fn histogram_quantiles_uses_the_given_label_name() {
     let q = lower(r#"histogram_quantiles(h, "phi", 0.5)"#);
-    let QueryExpr::Concat { children } = &q else {
+    let QueryExpr::Concat { children, .. } = &q else {
         panic!("expected Concat");
     };
     let QueryExpr::PromqlRelabel { dst, .. } = &children[0] else {

crates/frontend-sql/src/sql/mod.rs

@@ -865,7 +865,14 @@ impl<'a> SqlLowerer<'a> {
             })
             .collect::<Result<Vec<_>, LoweringError>>()?;
 
-        Ok(Unresolved::Concat { children: branches })
+        // No discriminator asserted here today (issue #228): DataFusion's own
+        // `__grouping_id` would be the natural one, but this front end
+        // already discards it (see above — `GROUPING()` itself is rejected),
+        // so there is no distinct-per-branch column available to name yet.
+        // `Unresolved::concat` keeps `output_schema`'s default (drop
+        // `unique_keys` entirely). See
+        // `docs/design_docs/concat-unique-keys-decision.md`.
+        Ok(Unresolved::concat(branches))
     }
 
     fn lower_sort(&self, sort: &logical_expr::Sort) -> Result<Unresolved, LoweringError> {

crates/frontend-sql/tests/data_quality_check/synthetic_packet_trace.rs

@@ -105,7 +105,7 @@ fn intents(e: &QueryExpr) -> Vec<AggIntent> {
                 go(lhs, out);
                 go(rhs, out);
             }
-            QueryExpr::Concat { children } => children.iter().for_each(|c| go(c, out)),
+            QueryExpr::Concat { children, .. } => children.iter().for_each(|c| go(c, out)),
             QueryExpr::PromqlVectorFromScalar(inner) | QueryExpr::PromqlScalarFromVector(inner) => {
                 go(inner, out)
             }

crates/frontend-sql/tests/netflow/netflow.rs

@@ -290,7 +290,7 @@ fn visit(qe: &QueryExpr, f: &mut impl FnMut(&QueryExpr)) {
             visit(lhs, f);
             visit(rhs, f);
         }
-        QueryExpr::Concat { children } => {
+        QueryExpr::Concat { children, .. } => {
             for child in children {
                 visit(child, f);
             }

crates/frontend-sql/tests/sql_lowering.rs

@@ -1292,7 +1292,7 @@ async fn a_shared_expression_is_materialized_once() {
 fn merge_branches(qe: &QueryExpr) -> &Vec<QueryExpr> {
     fn find(qe: &QueryExpr) -> Option<&Vec<QueryExpr>> {
         match qe {
-            QueryExpr::Concat { children } => Some(children),
+            QueryExpr::Concat { children, .. } => Some(children),
             QueryExpr::Project { child, .. }
             | QueryExpr::Filter { child, .. }
             | QueryExpr::Sort { child, .. }

crates/types/src/dag_export.rs

@@ -1205,13 +1205,18 @@ fn build_no_recheck(
                 vec![c],
             )
         }
-        QueryExpr::Concat { children } => {
+        QueryExpr::Concat {
+            children,
+            discriminator_unique_key,
+        } => {
             let ids: Vec<u32> = children
                 .iter()
                 .map(|c| build(c, nodes, cache, find_winner))
                 .collect();
             let label = format!("Concat({} branches)", ids.len());
-            push_node(nodes, expr, cache, label, serde_json::json!({}), ids)
+            let detail =
+                serde_json::json!({ "discriminator_unique_key": discriminator_unique_key });
+            push_node(nodes, expr, cache, label, detail, ids)
         }
         QueryExpr::Join {
             kind,
@@ -1442,6 +1447,7 @@ mod tests {
         };
         let root = QueryExpr::Concat {
             children: vec![left_branch, right_branch],
+            discriminator_unique_key: None,
         };
         let graph = export_post_asap(&root, &mut |_| None);
 
@@ -1531,13 +1537,11 @@ mod tests {
 
     #[test]
     fn merge_keeps_every_branch_as_a_child() {
-        let expr = QueryExpr::Concat {
-            children: vec![
-                scan("a", value_col()),
-                scan("b", value_col()),
-                scan("c", value_col()),
-            ],
-        };
+        let expr = QueryExpr::concat(vec![
+            scan("a", value_col()),
+            scan("b", value_col()),
+            scan("c", value_col()),
+        ]);
         let graph = export(&expr);
         assert_eq!(graph.nodes.len(), 4, "3 branches + the Concat node");
         let merge = &graph.nodes[graph.root as usize];

crates/types/src/pre_asap/binder.rs

@@ -163,7 +163,7 @@ fn leftmost_scan_name(tree: &UnresolvedQueryExpr) -> Option<&str> {
         | QE::TimeRange { child, .. }
         | QE::TimeShift { child, .. }
         | QE::SQLWindowFunc { child, .. } => leftmost_scan_name(child),
-        QE::Concat { children } => children.first().and_then(leftmost_scan_name),
+        QE::Concat { children, .. } => children.first().and_then(leftmost_scan_name),
         QE::Join { left, .. } | QE::SetOp { left, .. } | QE::BinaryOp { lhs: left, .. } => {
             leftmost_scan_name(left)
         }
@@ -310,7 +310,24 @@ pub(crate) fn collect_referenced_columns(tree: &UnresolvedQueryExpr) -> Vec<Stri
             | QE::PromqlSubquery { child, .. }
             | QE::TimeRange { child, .. }
             | QE::TimeShift { child, .. } => walk(child, out),
-            QE::Concat { children } => children.iter().for_each(|c| walk(c, out)),
+            QE::Concat {
+                children,
+                discriminator_unique_key,
+            } => {
+                // Same treatment as `Dedup.cols` above: an own-field
+                // `ColumnRef` must be seeded here too, or a discriminator
+                // column that isn't otherwise referenced anywhere else in
+                // the tree (plausible — a raw usage-derived label, not one a
+                // `Project`/relabel freshly created) is absent from the
+                // Binder's usage-derived fallback schema, and
+                // `resolve.rs`'s later `resolve_column_ref` call fails with
+                // `NotFound` for a column the caller correctly named.
+                if let Some(key) = discriminator_unique_key {
+                    push_ref_name(key.discriminator(), out);
+                    key.inner_key().iter().for_each(|c| push_ref_name(c, out));
+                }
+                children.iter().for_each(|c| walk(c, out));
+            }
             QE::SetOp { left, right, .. } => {
                 walk(left, out);
                 walk(right, out);
@@ -390,6 +407,29 @@ mod tests {
         assert!(schema.column_id("host").is_some());
     }
 
+    /// Issue #228 review: a `Concat`'s `discriminator_unique_key` columns —
+    /// even one referenced nowhere else in the tree — must be seeded into
+    /// the usage-derived fallback schema, exactly like `Dedup.cols`, or
+    /// `resolve.rs`'s later `resolve_column_ref` fails `NotFound` for a
+    /// column the caller correctly named.
+    #[test]
+    fn concat_discriminator_key_is_seeded_into_the_binder_schema() {
+        let tree = UnresolvedQueryExpr::concat_with_discriminator(
+            vec![src("m")],
+            ColumnRef::Named("phi".into()),
+            vec![ColumnRef::Named("host".into())],
+        );
+        let schema = Binder::new().bind(&tree);
+        assert!(
+            schema.column_id("phi").is_some(),
+            "discriminator column must be seeded"
+        );
+        assert!(
+            schema.column_id("host").is_some(),
+            "inner_key column must be seeded"
+        );
+    }
+
     #[test]
     fn inherited_names_are_seeded_alongside_referenced() {
         // A `BinaryOp` side re-binds against its own sub-tree, but must still see