uqa_sql/semantics/
join_using.rs

1//
2// Unified Query Algebra
3//
4// Copyright (c) 2023-2026 Cognica, Inc.
5//
6
7//! Schema binding and positional output shaping for `PostgreSQL` qualified joins.
8
9use std::collections::{BTreeMap, BTreeSet};
10
11use crate::ast::{BinaryOp, ColumnType, JoinKind, JoinUsing};
12use crate::schema::join_output::JoinOutputSource;
13use crate::SQLError;
14use crate::{ColumnIdentity, RowSchema, ScalarExpr};
15
16#[derive(Debug, Clone)]
17pub struct ResolvedJoinUsing {
18    columns: Vec<ResolvedJoinColumn>,
19    alias: Option<String>,
20}
21
22pub type JoinUsingLayout = (
23    Vec<(String, ColumnIdentity, JoinOutputSource)>,
24    Vec<(ColumnIdentity, JoinOutputSource)>,
25);
26
27#[derive(Debug, Clone)]
28struct ResolvedJoinColumn {
29    name: String,
30    left: usize,
31    right: usize,
32    comparison_type: Option<ColumnType>,
33    output_type: Option<ColumnType>,
34}
35
36fn visible_columns(schema: &RowSchema) -> Vec<(&str, usize)> {
37    schema
38        .columns()
39        .iter()
40        .enumerate()
41        .filter_map(|(position, _)| Some((schema.public_name(position)?, position)))
42        .collect()
43}
44
45fn positions_by_name<'a>(columns: &'a [(&'a str, usize)]) -> BTreeMap<&'a str, Vec<usize>> {
46    let mut positions = BTreeMap::<&str, Vec<usize>>::new();
47    for (name, position) in columns {
48        positions.entry(name).or_default().push(*position);
49    }
50    positions
51}
52
53fn missing_using_column(column: &str, side: &str) -> SQLError {
54    SQLError::Routine {
55        sqlstate: "42703".into(),
56        message: format!(
57            "column \"{column}\" specified in USING clause does not exist in {side} table"
58        ),
59    }
60}
61
62fn ambiguous_using_column(column: &str, side: &str) -> SQLError {
63    SQLError::Routine {
64        sqlstate: "42702".into(),
65        message: format!("common column name \"{column}\" appears more than once in {side} table"),
66    }
67}
68
69fn unique_position(
70    positions: &BTreeMap<&str, Vec<usize>>,
71    column: &str,
72    side: &str,
73) -> Result<usize, SQLError> {
74    match positions.get(column).map(Vec::as_slice) {
75        None | Some([]) => Err(missing_using_column(column, side)),
76        Some([position]) => Ok(*position),
77        Some(_) => Err(ambiguous_using_column(column, side)),
78    }
79}
80
81pub fn resolve_join_using(
82    explicit: Option<&JoinUsing>,
83    natural: bool,
84    left: &RowSchema,
85    right: &RowSchema,
86) -> Result<Option<ResolvedJoinUsing>, SQLError> {
87    if explicit.is_none() && !natural {
88        return Ok(None);
89    }
90    if explicit.is_some() && natural {
91        return Err(SQLError::Internal(
92            "JOIN cannot be both USING-qualified and NATURAL".into(),
93        ));
94    }
95
96    let left_columns = visible_columns(left);
97    let right_columns = visible_columns(right);
98    let left_positions = positions_by_name(&left_columns);
99    let right_positions = positions_by_name(&right_columns);
100    let (names, alias) = if let Some(using) = explicit {
101        let mut seen = BTreeSet::new();
102        for name in &using.columns {
103            if !seen.insert(name.as_str()) {
104                return Err(SQLError::Routine {
105                    sqlstate: "42701".into(),
106                    message: format!(
107                        "column name \"{name}\" appears more than once in USING clause"
108                    ),
109                });
110            }
111        }
112        (using.columns.clone(), using.alias.clone())
113    } else {
114        let mut names = Vec::new();
115        let mut seen = BTreeSet::new();
116        for (name, _) in &left_columns {
117            if right_positions.contains_key(name) && seen.insert(*name) {
118                names.push((*name).to_string());
119            }
120        }
121        (names, None)
122    };
123
124    if let Some(alias) = alias.as_deref() {
125        if left.has_qualifier(alias) || right.has_qualifier(alias) {
126            return Err(SQLError::Routine {
127                sqlstate: "42712".into(),
128                message: format!("table name \"{alias}\" specified more than once"),
129            });
130        }
131    }
132
133    let mut columns = Vec::with_capacity(names.len());
134    for name in names {
135        let left_position = unique_position(&left_positions, &name, "left")?;
136        let right_position = unique_position(&right_positions, &name, "right")?;
137        let (comparison_type, output_type) = match (
138            left.column_type(left_position),
139            right.column_type(right_position),
140        ) {
141            (Some(left_type), Some(right_type)) => (
142                Some(crate::equality_operand_type(left_type, right_type)?),
143                Some(crate::type_resolution::common_type_in(
144                    crate::type_resolution::CommonTypeContext::JoinUsing,
145                    left_type,
146                    right_type,
147                )?),
148            ),
149            _ => (None, None),
150        };
151        columns.push(ResolvedJoinColumn {
152            left: left_position,
153            right: right_position,
154            name,
155            comparison_type,
156            output_type,
157        });
158    }
159    Ok(Some(ResolvedJoinUsing { columns, alias }))
160}
161
162pub fn join_using_predicate(
163    using: &ResolvedJoinUsing,
164    left: &RowSchema,
165    right: &RowSchema,
166) -> Option<ScalarExpr> {
167    let mut predicates = using
168        .columns
169        .iter()
170        .map(|column| {
171            let lhs = coerced_join_column(
172                ScalarExpr::Position(column.left),
173                left.column_type(column.left),
174                column.comparison_type.as_ref(),
175            );
176            let rhs = coerced_join_column(
177                ScalarExpr::Position(left.len() + column.right),
178                right.column_type(column.right),
179                column.comparison_type.as_ref(),
180            );
181            ScalarExpr::Binary {
182                op: BinaryOp::Equal,
183                lhs: Box::new(lhs),
184                rhs: Box::new(rhs),
185            }
186        })
187        .collect::<Vec<_>>();
188    match predicates.len() {
189        0 => None,
190        1 => predicates.pop(),
191        _ => Some(ScalarExpr::And(predicates)),
192    }
193}
194
195fn coerced_join_column(
196    expression: ScalarExpr,
197    source: Option<&ColumnType>,
198    target: Option<&ColumnType>,
199) -> ScalarExpr {
200    match (source, target) {
201        (Some(source), Some(target)) if source != target => ScalarExpr::Cast {
202            implicit: true,
203            expr: Box::new(expression),
204            ty: target.catalog_name(),
205        },
206        _ => expression,
207    }
208}
209
210/// Bind the visible and hidden output identities of a qualified join without
211/// constructing or executing the join. Static source-schema binding and the
212/// physical operator share this layout so an empty or correlated right side
213/// cannot lose its declared types.
214pub fn join_using_output_schema(
215    kind: JoinKind,
216    left: &RowSchema,
217    right: &RowSchema,
218    using: &ResolvedJoinUsing,
219) -> Result<RowSchema, SQLError> {
220    let input = RowSchema::join(left, right, std::iter::empty());
221    let (columns, aliases) = join_using_layout(kind, left, right, using)?;
222    crate::schema::join_output::compile_layout(&input, &columns, &aliases)
223        .map(|(schema, _)| schema)
224        .map_err(|error| SQLError::Internal(error.0))
225}
226
227/// Add a parenthesized JOIN alias to the input-side schemas for null-rejection analysis. Merged FULL JOIN columns belong to neither side exclusively because their value is a coalesce of both inputs.
228#[expect(
229    clippy::too_many_lines,
230    reason = "preserves source schema and row identity"
231)]
232pub fn join_alias_input_schemas(
233    kind: JoinKind,
234    left: &RowSchema,
235    right: &RowSchema,
236    using: Option<&ResolvedJoinUsing>,
237    alias: &str,
238    column_aliases: &[String],
239) -> Result<(RowSchema, RowSchema), SQLError> {
240    #[derive(Clone, Copy)]
241    enum Owner {
242        Left(usize),
243        Right(usize),
244        Neither,
245    }
246
247    let mut output = Vec::<(String, Owner)>::new();
248    if let Some(using) = using {
249        let mut left_used = vec![false; left.len()];
250        let mut right_used = vec![false; right.len()];
251        for column in &using.columns {
252            left_used[column.left] = true;
253            right_used[column.right] = true;
254            let owner = match kind {
255                JoinKind::Inner | JoinKind::Left => Owner::Left(column.left),
256                JoinKind::Right => Owner::Right(column.right),
257                JoinKind::Full => Owner::Neither,
258                JoinKind::Cross => {
259                    return Err(SQLError::Internal(
260                        "CROSS JOIN cannot carry a USING qualification".into(),
261                    ));
262                }
263            };
264            output.push((column.name.clone(), owner));
265        }
266        output.extend(
267            left.columns()
268                .iter()
269                .enumerate()
270                .filter(|(position, _)| !left_used[*position])
271                .map(|(position, column)| {
272                    (
273                        left.public_name(position).unwrap_or(column).to_string(),
274                        Owner::Left(position),
275                    )
276                }),
277        );
278        output.extend(
279            right
280                .columns()
281                .iter()
282                .enumerate()
283                .filter(|(position, _)| !right_used[*position])
284                .map(|(position, column)| {
285                    (
286                        right.public_name(position).unwrap_or(column).to_string(),
287                        Owner::Right(position),
288                    )
289                }),
290        );
291    } else {
292        output.extend(left.columns().iter().enumerate().map(|(position, column)| {
293            (
294                left.public_name(position).unwrap_or(column).to_string(),
295                Owner::Left(position),
296            )
297        }));
298        output.extend(
299            right
300                .columns()
301                .iter()
302                .enumerate()
303                .map(|(position, column)| {
304                    (
305                        right.public_name(position).unwrap_or(column).to_string(),
306                        Owner::Right(position),
307                    )
308                }),
309        );
310    }
311
312    if column_aliases.len() > output.len() {
313        return Err(SQLError::Routine {
314            sqlstate: "42P10".into(),
315            message: format!(
316                "join expression \"{alias}\" has {} columns available but {} columns specified",
317                output.len(),
318                column_aliases.len()
319            ),
320        });
321    }
322    for (position, column_alias) in column_aliases.iter().enumerate() {
323        output[position].0.clone_from(column_alias);
324    }
325
326    let mut counts = BTreeMap::<String, usize>::new();
327    for (column, _) in &output {
328        *counts.entry(column.clone()).or_default() += 1;
329    }
330    let mut left_aliases = Vec::new();
331    let mut right_aliases = Vec::new();
332    for (column, owner) in output {
333        if counts.get(&column) != Some(&1) {
334            continue;
335        }
336        let identity = ColumnIdentity::qualified(alias, column);
337        match owner {
338            Owner::Left(position) => left_aliases.push((identity, position)),
339            Owner::Right(position) => right_aliases.push((identity, position)),
340            Owner::Neither => {}
341        }
342    }
343    Ok((
344        RowSchema::with_identity_aliases(left, &left_aliases),
345        RowSchema::with_identity_aliases(right, &right_aliases),
346    ))
347}
348
349pub fn join_using_layout(
350    kind: JoinKind,
351    left: &RowSchema,
352    right: &RowSchema,
353    using: &ResolvedJoinUsing,
354) -> Result<JoinUsingLayout, SQLError> {
355    let right_offset = left.len();
356    let mut left_used = vec![false; left.len()];
357    let mut right_used = vec![false; right.len()];
358    let mut columns = Vec::with_capacity(left.len() + right.len() - using.columns.len());
359    let mut merged_sources = Vec::with_capacity(using.columns.len());
360
361    for column in &using.columns {
362        left_used[column.left] = true;
363        right_used[column.right] = true;
364        let source = match kind {
365            JoinKind::Inner | JoinKind::Left => coerced_output_source(
366                column.left,
367                left.column_type(column.left),
368                column.output_type.as_ref(),
369            ),
370            JoinKind::Right => coerced_output_source(
371                right_offset + column.right,
372                right.column_type(column.right),
373                column.output_type.as_ref(),
374            ),
375            JoinKind::Full => {
376                column
377                    .output_type
378                    .as_ref()
379                    .map_or(JoinOutputSource::Input(column.left), |ty| {
380                        JoinOutputSource::Coalesce {
381                            left: column.left,
382                            right: right_offset + column.right,
383                            ty: ty.clone(),
384                        }
385                    })
386            }
387            JoinKind::Cross => {
388                return Err(SQLError::Internal(
389                    "CROSS JOIN cannot carry a USING qualification".into(),
390                ));
391            }
392        };
393        columns.push((
394            column.name.clone(),
395            ColumnIdentity::unqualified(&column.name),
396            source.clone(),
397        ));
398        merged_sources.push((column.name.clone(), source));
399    }
400    for (position, column) in left.columns().iter().enumerate() {
401        if !left_used[position] {
402            columns.push((
403                left.public_name(position).unwrap_or(column).to_string(),
404                left.identities()[position].clone(),
405                JoinOutputSource::Input(position),
406            ));
407        }
408    }
409    for (position, column) in right.columns().iter().enumerate() {
410        if !right_used[position] {
411            columns.push((
412                right.public_name(position).unwrap_or(column).to_string(),
413                right.identities()[position].clone(),
414                JoinOutputSource::Input(right_offset + position),
415            ));
416        }
417    }
418
419    let mut aliases = Vec::new();
420    for (position, identity) in left.identities().iter().enumerate() {
421        if identity.qualifier().is_some() {
422            aliases.push((identity.clone(), JoinOutputSource::Input(position)));
423        }
424    }
425    for (position, identity) in right.identities().iter().enumerate() {
426        if identity.qualifier().is_some() {
427            aliases.push((
428                identity.clone(),
429                JoinOutputSource::Input(right_offset + position),
430            ));
431        }
432    }
433    if let Some(alias) = using.alias.as_ref() {
434        aliases.extend(
435            merged_sources
436                .into_iter()
437                .map(|(column, source)| (ColumnIdentity::qualified(alias, column), source)),
438        );
439    }
440
441    Ok((columns, aliases))
442}
443
444fn coerced_output_source(
445    input: usize,
446    source: Option<&ColumnType>,
447    target: Option<&ColumnType>,
448) -> JoinOutputSource {
449    match (source, target) {
450        (Some(source), Some(target)) if source != target => JoinOutputSource::Cast {
451            input,
452            ty: target.clone(),
453        },
454        _ => JoinOutputSource::Input(input),
455    }
456}
457
458#[cfg(test)]
459mod tests {
460    use super::*;
461
462    struct Types;
463    impl crate::FunctionTypeResolver for Types {
464        fn resolve_function_type(
465            &self,
466            _: &str,
467            _: Option<&crate::ast::FunctionBinding>,
468            _: &[Option<String>],
469            _: &[Option<ColumnType>],
470            _: bool,
471        ) -> Result<Option<ColumnType>, SQLError> {
472            Ok(None)
473        }
474    }
475    impl crate::routines::RoutineResolution for Types {}
476
477    #[test]
478    fn natural_columns_follow_left_input_order() {
479        let left = RowSchema::with_qualified_types(
480            "l",
481            vec!["b".into(), "a".into(), "only".into()],
482            vec![None; 3],
483        );
484        let right = RowSchema::with_qualified_types(
485            "r",
486            vec!["a".into(), "b".into(), "other".into()],
487            vec![None; 3],
488        );
489        let resolved = resolve_join_using(None, true, &left, &right)
490            .unwrap()
491            .unwrap();
492        assert_eq!(
493            resolved
494                .columns
495                .iter()
496                .map(|column| column.name.as_str())
497                .collect::<Vec<_>>(),
498            ["b", "a"]
499        );
500    }
501
502    #[test]
503    fn nested_using_predicate_preserves_resolved_input_positions() {
504        let base = |alias: &str| {
505            RowSchema::with_qualified_types(
506                alias,
507                vec!["payload".into(), "id".into()],
508                vec![Some(ColumnType::Text), Some(ColumnType::Integer)],
509            )
510        };
511        let using = JoinUsing {
512            columns: vec!["id".into()],
513            alias: None,
514        };
515        let nested = |a: &str, b: &str, kind| {
516            let (a, b) = (base(a), base(b));
517            let resolved = resolve_join_using(Some(&using), false, &a, &b)
518                .unwrap()
519                .unwrap();
520            join_using_output_schema(kind, &a, &b, &resolved).unwrap()
521        };
522        for kind in [
523            JoinKind::Inner,
524            JoinKind::Left,
525            JoinKind::Right,
526            JoinKind::Full,
527        ] {
528            let left = nested("a", "b", kind);
529            let right = nested("c", "d", kind);
530            let resolved = resolve_join_using(Some(&using), false, &left, &right)
531                .unwrap()
532                .unwrap();
533            let predicate = join_using_predicate(&resolved, &left, &right).unwrap();
534            let joined = RowSchema::join(&left, &right, []);
535            let crate::plan::UnifiedPlan::Query(query) =
536                crate::plan::UnifiedPlan::lower(crate::compile("SELECT 1").unwrap().remove(0))
537            else {
538                panic!("query")
539            };
540            let crate::plan::RelationalPlan::QueryBlock(mut block) = query.root else {
541                panic!("block")
542            };
543            block.r#where = Some(predicate);
544            crate::binding::validate_query_block_references(
545                &Types,
546                &block,
547                &joined,
548                &[],
549                &crate::binding::fixture::empty_binding_context(),
550                None,
551            )
552            .unwrap();
553            assert_ne!(
554                left.physical_slot(0),
555                Some(0),
556                "logical positions must not be physical offsets"
557            );
558        }
559    }
560
561    #[test]
562    fn using_requires_one_column_on_each_side() {
563        let left = RowSchema::with_identities(
564            vec!["id".into(), "id".into()],
565            vec![
566                ColumnIdentity::qualified("l", "id"),
567                ColumnIdentity::qualified("other", "id"),
568            ],
569            vec![None; 2],
570        );
571        let right = RowSchema::with_qualified_types("r", vec!["id".into()], vec![None]);
572        let error = resolve_join_using(
573            Some(&JoinUsing {
574                columns: vec!["id".into()],
575                alias: None,
576            }),
577            false,
578            &left,
579            &right,
580        )
581        .unwrap_err();
582        assert_eq!(error.sqlstate(), Some("42702"));
583    }
584}
uqa_sql/semantics/join_using.rs

uqa_sql/semantics/
join_using.rs
