uqa_sql/semantics/
merge.rs

1//
2// Unified Query Algebra
3//
4// Copyright (c) 2023-2026 Cognica, Inc.
5//
6
7//! MERGE clause scopes, target privileges, and positional RETURNING schemas.
8use super::returning::expanded_returning_projections;
9use crate::{
10    plan::{MergePlan, MergeWhenPlan, ProjectionPlan},
11    SQLError, SQLParam,
12};
13use std::collections::BTreeSet;
14
15#[expect(
16    clippy::too_many_lines,
17    reason = "validates clause-specific schemas and FULL JOIN requirements"
18)]
19pub fn validate_merge_action_scopes(
20    routines: &dyn crate::routines::RoutineResolution,
21    stmt: &MergePlan,
22    target_schema: &crate::RowSchema,
23    source_schema: &crate::RowSchema,
24    params: &[SQLParam],
25    bindings: &crate::binding::context::BindingContext<'_>,
26) -> Result<(), SQLError> {
27    let matched_schema = crate::RowSchema::join(target_schema, source_schema, std::iter::empty());
28    let expression_type = |expression: &crate::ScalarExpr, schema: &crate::RowSchema| {
29        crate::binding::analyze_projection_output_schema(
30            routines,
31            &[ProjectionPlan {
32                expr: expression.clone(),
33                alias: None,
34            }],
35            schema,
36            schema,
37            &stmt.subqueries,
38            params,
39            bindings,
40        )
41        .map(|output| output.column_type(0).cloned())
42    };
43    let validate_boolean = |expression: &crate::ScalarExpr,
44                            schema: &crate::RowSchema,
45                            label: &str|
46     -> Result<(), SQLError> {
47        if expression_type(expression, schema)?
48            .is_some_and(|ty| ty != crate::ast::ColumnType::Boolean)
49        {
50            return Err(SQLError::TypeMismatch(format!(
51                "argument of {label} must be type boolean"
52            )));
53        }
54        Ok(())
55    };
56    validate_boolean(&stmt.join_condition, &matched_schema, "MERGE ON")?;
57    let has_source_missing = stmt.when_clauses.iter().any(|clause| {
58        matches!(
59            clause,
60            MergeWhenPlan::UpdateNotMatchedBySource { .. }
61                | MergeWhenPlan::DeleteNotMatchedBySource { .. }
62                | MergeWhenPlan::NothingNotMatchedBySource { .. }
63        )
64    });
65    let has_target_missing = stmt.when_clauses.iter().any(|clause| {
66        matches!(
67            clause,
68            MergeWhenPlan::InsertNotMatched { .. } | MergeWhenPlan::NothingNotMatched { .. }
69        )
70    });
71    if has_source_missing
72        && has_target_missing
73        && !super::join_predicates::join_conjuncts(&stmt.join_condition)
74            .into_iter()
75            .any(|conjunct| {
76                matches!(
77                    conjunct,
78                    crate::ScalarExpr::Binary {
79                        op: crate::ast::BinaryOp::Equal,
80                        lhs,
81                        rhs,
82                    } if super::join_predicates::decide_join_sides(
83                        target_schema,
84                        source_schema,
85                        lhs,
86                        rhs,
87                    )
88                    .is_some()
89                )
90            })
91    {
92        return Err(SQLError::Routine {
93            sqlstate: "0A000".into(),
94            message:
95                "FULL JOIN is only supported with merge-joinable or hash-joinable join conditions"
96                    .into(),
97        });
98    }
99    for clause in &stmt.when_clauses {
100        let (condition, expressions, schema): (
101            Option<&crate::ScalarExpr>,
102            Vec<&crate::ScalarExpr>,
103            &crate::RowSchema,
104        ) = match clause {
105            MergeWhenPlan::UpdateMatched {
106                condition,
107                assignments,
108            } => (
109                condition.as_ref(),
110                assignments
111                    .iter()
112                    .flat_map(crate::plan::AssignmentPlan::expressions)
113                    .collect(),
114                &matched_schema,
115            ),
116            MergeWhenPlan::DeleteMatched { condition }
117            | MergeWhenPlan::NothingMatched { condition } => {
118                (condition.as_ref(), Vec::new(), &matched_schema)
119            }
120            MergeWhenPlan::UpdateNotMatchedBySource {
121                condition,
122                assignments,
123            } => (
124                condition.as_ref(),
125                assignments
126                    .iter()
127                    .flat_map(crate::plan::AssignmentPlan::expressions)
128                    .collect(),
129                target_schema,
130            ),
131            MergeWhenPlan::DeleteNotMatchedBySource { condition }
132            | MergeWhenPlan::NothingNotMatchedBySource { condition } => {
133                (condition.as_ref(), Vec::new(), target_schema)
134            }
135            MergeWhenPlan::InsertNotMatched {
136                condition,
137                columns,
138                values,
139                ..
140            } => (
141                condition.as_ref(),
142                columns
143                    .iter()
144                    .flat_map(crate::ast::AssignmentTarget::expressions)
145                    .chain(values)
146                    .collect(),
147                source_schema,
148            ),
149            MergeWhenPlan::NothingNotMatched { condition } => {
150                (condition.as_ref(), Vec::new(), source_schema)
151            }
152        };
153        if let Some(condition) = condition {
154            validate_boolean(condition, schema, "WHEN")?;
155        }
156        for expression in expressions {
157            expression_type(expression, schema)?;
158        }
159    }
160    Ok(())
161}
162
163pub fn expanded_merge_returning_projections(
164    catalog: &dyn super::returning::ReturningCatalog,
165    target_table: &str,
166    target_qualifier: &str,
167    aliases: &crate::ast::ReturningAliases,
168    source_schema: &crate::RowSchema,
169    source_relation: crate::ast::InternalRelationId,
170    returning: &[ProjectionPlan],
171) -> Result<Vec<ProjectionPlan>, SQLError> {
172    let target_star = ProjectionPlan {
173        expr: crate::ScalarExpr::QualifiedStar(target_qualifier.into()),
174        alias: None,
175    };
176    let target_projections = expanded_returning_projections(
177        catalog,
178        target_table,
179        target_qualifier,
180        aliases,
181        std::slice::from_ref(&target_star),
182    )?;
183    let mut projections = Vec::new();
184    for projection in returning {
185        match &projection.expr {
186            crate::ScalarExpr::Star => {
187                projections.extend(
188                    source_schema
189                        .columns()
190                        .iter()
191                        .enumerate()
192                        .filter(|(position, _)| {
193                            super::projection::visible_projection_source_position(
194                                source_schema,
195                                *position,
196                            )
197                        })
198                        .map(|(position, column)| ProjectionPlan {
199                            expr: crate::ScalarExpr::InternalColumn(
200                                source_relation.column(position),
201                            ),
202                            alias: Some(
203                                source_schema
204                                    .public_name(position)
205                                    .unwrap_or(column)
206                                    .to_string(),
207                            ),
208                        }),
209                );
210                projections.extend(target_projections.iter().cloned());
211            }
212            crate::ScalarExpr::QualifiedStar(qualifier)
213                if qualifier == target_qualifier
214                    || qualifier == &aliases.old
215                    || qualifier == &aliases.new =>
216            {
217                projections.extend(expanded_returning_projections(
218                    catalog,
219                    target_table,
220                    target_qualifier,
221                    aliases,
222                    std::slice::from_ref(projection),
223                )?);
224            }
225            _ => projections.push(projection.clone()),
226        }
227    }
228    Ok(projections)
229}
230
231pub fn merge_returning_source_schema(
232    source_schema: &crate::RowSchema,
233    source_relation: crate::ast::InternalRelationId,
234) -> crate::RowSchema {
235    let aliases = source_schema
236        .columns()
237        .iter()
238        .enumerate()
239        .map(|(position, _)| {
240            (
241                source_relation.column(position),
242                source_schema
243                    .physical_slot(position)
244                    .expect("source column has a physical slot"),
245                source_schema.column_type(position).cloned(),
246            )
247        })
248        .collect::<Vec<_>>();
249    crate::RowSchema::with_physical_internal_aliases(source_schema, &aliases)
250}
251
252/// The columns a `MERGE` supplies, as `PostgreSQL` records them in its target's `insertedCols` and `updatedCols`: those each `INSERT` action fills and those each `UPDATE` action sets.
253pub fn merge_target_columns(
254    catalog: &dyn super::mutation_privileges::MutationPrivilegeCatalog,
255    stmt: &MergePlan,
256) -> Result<Vec<String>, SQLError> {
257    let table_columns = catalog.bound_table_column_names(&stmt.target)?;
258    Ok(merge_supplied_columns(stmt, &table_columns))
259}
260
261/// The columns the actions of a `MERGE` supply, as [`merge_target_columns`] finds them, for a target whose columns `target_columns` lists in order.
262pub fn merge_supplied_columns(stmt: &MergePlan, target_columns: &[String]) -> Vec<String> {
263    let mut supplied = Vec::new();
264    for clause in &stmt.when_clauses {
265        match clause {
266            MergeWhenPlan::InsertNotMatched {
267                columns, values, ..
268            } => supplied.extend(merge_insert_columns(target_columns, columns, values)),
269            MergeWhenPlan::UpdateMatched { assignments, .. }
270            | MergeWhenPlan::UpdateNotMatchedBySource { assignments, .. } => {
271                supplied.extend(
272                    assignments
273                        .iter()
274                        .flat_map(|assignment| assignment.target.column_names())
275                        .map(str::to_owned),
276                );
277            }
278            _ => {}
279        }
280    }
281    supplied
282}
283
284/// The columns an `INSERT` action fills: those it names, or else the leading columns of the table, one for each value; `DEFAULT VALUES` fills none.
285fn merge_insert_columns<T, V>(
286    table_columns: &[String],
287    columns: &[crate::ast::AssignmentTarget<T>],
288    values: &[V],
289) -> Vec<String> {
290    if columns.is_empty() {
291        table_columns.iter().take(values.len()).cloned().collect()
292    } else {
293        columns.iter().map(|target| target.column.clone()).collect()
294    }
295}
296
297pub fn ensure_merge_mutation_privileges(
298    catalog: &dyn super::mutation_privileges::MutationPrivilegeCatalog,
299    stmt: &MergePlan,
300) -> Result<(), SQLError> {
301    let mut column_privileges = BTreeSet::new();
302    let mut requires_delete = false;
303    let mut requires_any_insert = false;
304    let table_columns = catalog.bound_table_column_names(&stmt.target)?;
305    let privilege_subject = stmt
306        .target_privilege_subject
307        .clone()
308        .unwrap_or_else(|| catalog.current_role());
309    for clause in &stmt.when_clauses {
310        match clause {
311            MergeWhenPlan::InsertNotMatched {
312                columns, values, ..
313            } => {
314                if columns.is_empty() && values.is_empty() {
315                    requires_any_insert = true;
316                } else {
317                    let columns = merge_insert_columns(&table_columns, columns, values);
318                    column_privileges.extend(columns.into_iter().map(|column| {
319                        (
320                            crate::catalog::security::table::TableAclPrivilege::Insert,
321                            column,
322                        )
323                    }));
324                }
325            }
326            MergeWhenPlan::UpdateMatched { assignments, .. }
327            | MergeWhenPlan::UpdateNotMatchedBySource { assignments, .. } => {
328                column_privileges.extend(
329                    assignments
330                        .iter()
331                        .flat_map(|assignment| assignment.target.targets())
332                        .map(|target| {
333                            (
334                                crate::catalog::security::table::TableAclPrivilege::Update,
335                                target.column.clone(),
336                            )
337                        }),
338                );
339            }
340            MergeWhenPlan::DeleteMatched { .. }
341            | MergeWhenPlan::DeleteNotMatchedBySource { .. } => requires_delete = true,
342            _ => {}
343        }
344    }
345    if requires_delete {
346        catalog.ensure_table_privilege_for(
347            &stmt.target,
348            &privilege_subject,
349            crate::catalog::security::table::TableAclPrivilege::Delete,
350        )?;
351    }
352    if requires_any_insert {
353        catalog.ensure_any_column_privilege_for(
354            &stmt.target,
355            &privilege_subject,
356            crate::catalog::security::table::TableAclPrivilege::Insert,
357        )?;
358    }
359    for (privilege, column) in column_privileges {
360        catalog.ensure_column_privilege_for(
361            &stmt.target,
362            &column,
363            &privilege_subject,
364            privilege,
365        )?;
366    }
367    Ok(())
368}
369
370pub fn merge_returning_schema(
371    routines: &dyn crate::routines::RoutineResolution,
372    catalog: &dyn super::returning::ReturningCatalog,
373    stmt: &MergePlan,
374    params: &[SQLParam],
375    source_schema: &crate::RowSchema,
376    ctes: &crate::binding::context::BindingContext<'_>,
377) -> Result<Option<crate::RowSchema>, SQLError> {
378    if stmt.returning.is_empty() {
379        return Ok(None);
380    }
381    let source_relation = crate::ast::InternalRelationId::allocate();
382    let projections = expanded_merge_returning_projections(
383        catalog,
384        &stmt.target,
385        &stmt.target_qualifier,
386        &stmt.returning_aliases,
387        source_schema,
388        source_relation,
389        &stmt.returning,
390    )?;
391    let returning_source_schema = merge_returning_source_schema(source_schema, source_relation);
392    let star_schema = super::returning::returning_target_schema(catalog, &stmt.target)?;
393    let expression_schema = super::returning_expression_schema(
394        &star_schema,
395        &stmt.target_qualifier,
396        &stmt.returning_aliases,
397        Some(&returning_source_schema),
398    );
399    crate::binding::analyze_projection_output_schema(
400        routines,
401        &projections,
402        &expression_schema,
403        &star_schema,
404        &stmt.subqueries,
405        params,
406        ctes,
407    )
408    .map(Some)
409}
410
411pub fn merge_command_returning_schema(
412    routines: &dyn crate::routines::RoutineResolution,
413    catalog: &dyn super::returning::ReturningCatalog,
414    rows: &dyn super::mutation_rows::MutationRowCatalog,
415    stmt: &MergePlan,
416    params: &[SQLParam],
417    bindings: &crate::binding::context::BindingContext<'_>,
418) -> Result<Option<crate::RowSchema>, SQLError> {
419    if stmt.returning.is_empty() {
420        return Ok(None);
421    }
422    let source =
423        crate::binding::analyze_source_plan_schema(routines, &stmt.source, params, bindings, None)?;
424    super::returning::validate_returning_alias_relations(
425        &stmt.target_qualifier,
426        &stmt.returning_aliases,
427        Some(&source),
428    )?;
429    let target =
430        super::mutation_rows::null_target_schema(rows, &stmt.target, &stmt.target_qualifier)?;
431    validate_merge_action_scopes(routines, stmt, &target, &source, params, bindings)?;
432    merge_returning_schema(routines, catalog, stmt, params, &source, bindings)
433}
434
435pub fn validate_merge_target_columns(
436    catalog: &dyn crate::assignment::columns::AssignmentColumnCatalog,
437    stmt: &MergePlan,
438) -> Result<(), SQLError> {
439    use crate::assignment::columns::validate_mutation_targets;
440    for clause in &stmt.when_clauses {
441        match clause {
442            MergeWhenPlan::UpdateMatched { assignments, .. }
443            | MergeWhenPlan::UpdateNotMatchedBySource { assignments, .. } => {
444                validate_mutation_targets(
445                    catalog,
446                    &stmt.target,
447                    assignments
448                        .iter()
449                        .flat_map(|assignment| assignment.target.targets()),
450                    "MERGE UPDATE",
451                    false,
452                )?;
453            }
454            MergeWhenPlan::InsertNotMatched { columns, .. } => validate_mutation_targets(
455                catalog,
456                &stmt.target,
457                columns.iter(),
458                "MERGE INSERT",
459                true,
460            )?,
461            _ => {}
462        }
463    }
464    validate_merge_identity_targets(catalog, stmt)
465}
466
467/// Reject a `MERGE` update that assigns a `GENERATED ALWAYS` identity column anything but `DEFAULT`, and an insert that supplies a value for one without an `OVERRIDING` clause, whether or not a row reaches the action.
468fn validate_merge_identity_targets(
469    catalog: &dyn crate::assignment::columns::AssignmentColumnCatalog,
470    stmt: &MergePlan,
471) -> Result<(), SQLError> {
472    let identity = super::generated_values::GeneratedValueColumns::of(catalog, &stmt.target)?;
473    for clause in &stmt.when_clauses {
474        match clause {
475            MergeWhenPlan::UpdateMatched { assignments, .. }
476            | MergeWhenPlan::UpdateNotMatchedBySource { assignments, .. } => {
477                identity.validate_update(assignments.iter().flat_map(|assignment| {
478                    assignment.target.targets().iter().map(move |target| {
479                        (
480                            target.column.as_str(),
481                            matches!(assignment.value, crate::ScalarExpr::Default),
482                        )
483                    })
484                }))?;
485            }
486            MergeWhenPlan::InsertNotMatched {
487                columns,
488                overriding,
489                values,
490                ..
491            } => {
492                let targets = if columns.is_empty() {
493                    catalog
494                        .try_describe_table(&stmt.target)
495                        .map_err(|error| {
496                            SQLError::Internal(format!("read MERGE target columns: {error}"))
497                        })?
498                        .unwrap_or_default()
499                        .into_iter()
500                        .map(|column| column.name)
501                        .collect::<Vec<_>>()
502                } else {
503                    columns.iter().map(|target| target.column.clone()).collect()
504                };
505                identity.validate_insert(
506                    targets.iter().map(String::as_str).zip(
507                        values
508                            .iter()
509                            .map(|value| !matches!(value, crate::ScalarExpr::Default)),
510                    ),
511                    *overriding,
512                )?;
513            }
514            _ => {}
515        }
516    }
517    Ok(())
518}
uqa_sql/semantics/merge.rs

uqa_sql/semantics/
merge.rs
