1use 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
252pub 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 let mut supplied = Vec::new();
259 for clause in &stmt.when_clauses {
260 match clause {
261 MergeWhenPlan::InsertNotMatched {
262 columns, values, ..
263 } => supplied.extend(merge_insert_columns(&table_columns, columns, values)),
264 MergeWhenPlan::UpdateMatched { assignments, .. }
265 | MergeWhenPlan::UpdateNotMatchedBySource { assignments, .. } => supplied.extend(
266 assignments
267 .iter()
268 .map(|assignment| assignment.target.column.clone()),
269 ),
270 _ => {}
271 }
272 }
273 Ok(supplied)
274}
275
276fn merge_insert_columns<T, V>(
278 table_columns: &[String],
279 columns: &[crate::ast::AssignmentTarget<T>],
280 values: &[V],
281) -> Vec<String> {
282 if columns.is_empty() {
283 table_columns.iter().take(values.len()).cloned().collect()
284 } else {
285 columns.iter().map(|target| target.column.clone()).collect()
286 }
287}
288
289pub fn ensure_merge_mutation_privileges(
290 catalog: &dyn super::mutation_privileges::MutationPrivilegeCatalog,
291 stmt: &MergePlan,
292) -> Result<(), SQLError> {
293 let mut column_privileges = BTreeSet::new();
294 let mut requires_delete = false;
295 let mut requires_any_insert = false;
296 let table_columns = catalog.bound_table_column_names(&stmt.target)?;
297 let privilege_subject = stmt
298 .target_privilege_subject
299 .clone()
300 .unwrap_or_else(|| catalog.current_role());
301 for clause in &stmt.when_clauses {
302 match clause {
303 MergeWhenPlan::InsertNotMatched {
304 columns, values, ..
305 } => {
306 if columns.is_empty() && values.is_empty() {
307 requires_any_insert = true;
308 } else {
309 let columns = merge_insert_columns(&table_columns, columns, values);
310 column_privileges.extend(columns.into_iter().map(|column| {
311 (
312 crate::catalog::security::table::TableAclPrivilege::Insert,
313 column,
314 )
315 }));
316 }
317 }
318 MergeWhenPlan::UpdateMatched { assignments, .. }
319 | MergeWhenPlan::UpdateNotMatchedBySource { assignments, .. } => {
320 column_privileges.extend(assignments.iter().map(|assignment| {
321 (
322 crate::catalog::security::table::TableAclPrivilege::Update,
323 assignment.target.column.clone(),
324 )
325 }));
326 }
327 MergeWhenPlan::DeleteMatched { .. }
328 | MergeWhenPlan::DeleteNotMatchedBySource { .. } => requires_delete = true,
329 _ => {}
330 }
331 }
332 if requires_delete {
333 catalog.ensure_table_privilege_for(
334 &stmt.target,
335 &privilege_subject,
336 crate::catalog::security::table::TableAclPrivilege::Delete,
337 )?;
338 }
339 if requires_any_insert {
340 catalog.ensure_any_column_privilege_for(
341 &stmt.target,
342 &privilege_subject,
343 crate::catalog::security::table::TableAclPrivilege::Insert,
344 )?;
345 }
346 for (privilege, column) in column_privileges {
347 catalog.ensure_column_privilege_for(
348 &stmt.target,
349 &column,
350 &privilege_subject,
351 privilege,
352 )?;
353 }
354 Ok(())
355}
356
357pub fn merge_returning_schema(
358 routines: &dyn crate::routines::RoutineResolution,
359 catalog: &dyn super::returning::ReturningCatalog,
360 stmt: &MergePlan,
361 params: &[SQLParam],
362 source_schema: &crate::RowSchema,
363 ctes: &crate::binding::context::BindingContext<'_>,
364) -> Result<Option<crate::RowSchema>, SQLError> {
365 if stmt.returning.is_empty() {
366 return Ok(None);
367 }
368 let source_relation = crate::ast::InternalRelationId::allocate();
369 let projections = expanded_merge_returning_projections(
370 catalog,
371 &stmt.target,
372 &stmt.target_qualifier,
373 &stmt.returning_aliases,
374 source_schema,
375 source_relation,
376 &stmt.returning,
377 )?;
378 let returning_source_schema = merge_returning_source_schema(source_schema, source_relation);
379 let star_schema = super::returning::returning_target_schema(catalog, &stmt.target)?;
380 let expression_schema = super::returning_expression_schema(
381 &star_schema,
382 &stmt.target_qualifier,
383 &stmt.returning_aliases,
384 Some(&returning_source_schema),
385 );
386 crate::binding::analyze_projection_output_schema(
387 routines,
388 &projections,
389 &expression_schema,
390 &star_schema,
391 &stmt.subqueries,
392 params,
393 ctes,
394 )
395 .map(Some)
396}
397
398pub fn merge_command_returning_schema(
399 routines: &dyn crate::routines::RoutineResolution,
400 catalog: &dyn super::returning::ReturningCatalog,
401 rows: &dyn super::mutation_rows::MutationRowCatalog,
402 stmt: &MergePlan,
403 params: &[SQLParam],
404 bindings: &crate::binding::context::BindingContext<'_>,
405) -> Result<Option<crate::RowSchema>, SQLError> {
406 if stmt.returning.is_empty() {
407 return Ok(None);
408 }
409 let source =
410 crate::binding::analyze_source_plan_schema(routines, &stmt.source, params, bindings, None)?;
411 super::returning::validate_returning_alias_relations(
412 &stmt.target_qualifier,
413 &stmt.returning_aliases,
414 Some(&source),
415 )?;
416 let target =
417 super::mutation_rows::null_target_schema(rows, &stmt.target, &stmt.target_qualifier)?;
418 validate_merge_action_scopes(routines, stmt, &target, &source, params, bindings)?;
419 merge_returning_schema(routines, catalog, stmt, params, &source, bindings)
420}
421
422pub fn validate_merge_target_columns(
423 catalog: &dyn crate::assignment::columns::AssignmentColumnCatalog,
424 stmt: &MergePlan,
425) -> Result<(), SQLError> {
426 use crate::assignment::columns::validate_mutation_targets;
427 for clause in &stmt.when_clauses {
428 match clause {
429 MergeWhenPlan::UpdateMatched { assignments, .. }
430 | MergeWhenPlan::UpdateNotMatchedBySource { assignments, .. } => {
431 validate_mutation_targets(
432 catalog,
433 &stmt.target,
434 assignments.iter().map(|assignment| &assignment.target),
435 "MERGE UPDATE",
436 false,
437 )?;
438 }
439 MergeWhenPlan::InsertNotMatched { columns, .. } => validate_mutation_targets(
440 catalog,
441 &stmt.target,
442 columns.iter(),
443 "MERGE INSERT",
444 true,
445 )?,
446 _ => {}
447 }
448 }
449 validate_merge_identity_targets(catalog, stmt)
450}
451
452fn validate_merge_identity_targets(
454 catalog: &dyn crate::assignment::columns::AssignmentColumnCatalog,
455 stmt: &MergePlan,
456) -> Result<(), SQLError> {
457 let identity = super::generated_values::GeneratedValueColumns::of(catalog, &stmt.target)?;
458 for clause in &stmt.when_clauses {
459 match clause {
460 MergeWhenPlan::UpdateMatched { assignments, .. }
461 | MergeWhenPlan::UpdateNotMatchedBySource { assignments, .. } => {
462 identity.validate_update(assignments.iter().map(|assignment| {
463 (
464 assignment.target.column.as_str(),
465 matches!(assignment.value, crate::ScalarExpr::Default),
466 )
467 }))?;
468 }
469 MergeWhenPlan::InsertNotMatched {
470 columns,
471 overriding,
472 values,
473 ..
474 } => {
475 let targets = if columns.is_empty() {
476 catalog
477 .try_describe_table(&stmt.target)
478 .map_err(|error| {
479 SQLError::Internal(format!("read MERGE target columns: {error}"))
480 })?
481 .unwrap_or_default()
482 .into_iter()
483 .map(|column| column.name)
484 .collect::<Vec<_>>()
485 } else {
486 columns.iter().map(|target| target.column.clone()).collect()
487 };
488 identity.validate_insert(
489 targets.iter().map(String::as_str).zip(
490 values
491 .iter()
492 .map(|value| !matches!(value, crate::ScalarExpr::Default)),
493 ),
494 *overriding,
495 )?;
496 }
497 _ => {}
498 }
499 }
500 Ok(())
501}