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 .map(|assignment| &assignment.value)
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 .map(|assignment| &assignment.value)
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, values, ..
137 } => (condition.as_ref(), values.iter().collect(), source_schema),
138 MergeWhenPlan::NothingNotMatched { condition } => {
139 (condition.as_ref(), Vec::new(), source_schema)
140 }
141 };
142 if let Some(condition) = condition {
143 validate_boolean(condition, schema, "WHEN")?;
144 }
145 for expression in expressions {
146 expression_type(expression, schema)?;
147 }
148 }
149 Ok(())
150}
151
152pub fn expanded_merge_returning_projections(
153 catalog: &dyn super::returning::ReturningCatalog,
154 target_table: &str,
155 target_qualifier: &str,
156 aliases: &crate::ast::ReturningAliases,
157 source_schema: &crate::RowSchema,
158 source_relation: crate::ast::InternalRelationId,
159 returning: &[ProjectionPlan],
160) -> Result<Vec<ProjectionPlan>, SQLError> {
161 let target_star = ProjectionPlan {
162 expr: crate::ScalarExpr::QualifiedStar(target_qualifier.into()),
163 alias: None,
164 };
165 let target_projections = expanded_returning_projections(
166 catalog,
167 target_table,
168 target_qualifier,
169 aliases,
170 std::slice::from_ref(&target_star),
171 )?;
172 let mut projections = Vec::new();
173 for projection in returning {
174 match &projection.expr {
175 crate::ScalarExpr::Star => {
176 projections.extend(
177 source_schema
178 .columns()
179 .iter()
180 .enumerate()
181 .filter(|(position, _)| {
182 super::projection::visible_projection_source_position(
183 source_schema,
184 *position,
185 )
186 })
187 .map(|(position, column)| ProjectionPlan {
188 expr: crate::ScalarExpr::InternalColumn(
189 source_relation.column(position),
190 ),
191 alias: Some(
192 source_schema
193 .public_name(position)
194 .unwrap_or(column)
195 .to_string(),
196 ),
197 }),
198 );
199 projections.extend(target_projections.iter().cloned());
200 }
201 crate::ScalarExpr::QualifiedStar(qualifier)
202 if qualifier == target_qualifier
203 || qualifier == &aliases.old
204 || qualifier == &aliases.new =>
205 {
206 projections.extend(expanded_returning_projections(
207 catalog,
208 target_table,
209 target_qualifier,
210 aliases,
211 std::slice::from_ref(projection),
212 )?);
213 }
214 _ => projections.push(projection.clone()),
215 }
216 }
217 Ok(projections)
218}
219
220pub fn merge_returning_source_schema(
221 source_schema: &crate::RowSchema,
222 source_relation: crate::ast::InternalRelationId,
223) -> crate::RowSchema {
224 let aliases = source_schema
225 .columns()
226 .iter()
227 .enumerate()
228 .map(|(position, _)| {
229 (
230 source_relation.column(position),
231 source_schema
232 .physical_slot(position)
233 .expect("source column has a physical slot"),
234 source_schema.column_type(position).cloned(),
235 )
236 })
237 .collect::<Vec<_>>();
238 crate::RowSchema::with_physical_internal_aliases(source_schema, &aliases)
239}
240
241pub fn ensure_merge_mutation_privileges(
242 catalog: &dyn super::mutation_privileges::MutationPrivilegeCatalog,
243 stmt: &MergePlan,
244) -> Result<(), SQLError> {
245 let mut column_privileges = BTreeSet::new();
246 let mut requires_delete = false;
247 let mut requires_any_insert = false;
248 let table_columns = catalog.bound_table_column_names(&stmt.target)?;
249 let privilege_subject = stmt
250 .target_privilege_subject
251 .clone()
252 .unwrap_or_else(|| catalog.current_user_name());
253 for clause in &stmt.when_clauses {
254 match clause {
255 MergeWhenPlan::InsertNotMatched {
256 columns, values, ..
257 } => {
258 if columns.is_empty() && values.is_empty() {
259 requires_any_insert = true;
260 } else {
261 let columns = if columns.is_empty() {
262 table_columns
263 .iter()
264 .take(values.len())
265 .cloned()
266 .collect::<Vec<_>>()
267 } else {
268 columns.clone()
269 };
270 column_privileges.extend(columns.into_iter().map(|column| {
271 (
272 crate::catalog::security::table::TableAclPrivilege::Insert,
273 column,
274 )
275 }));
276 }
277 }
278 MergeWhenPlan::UpdateMatched { assignments, .. }
279 | MergeWhenPlan::UpdateNotMatchedBySource { assignments, .. } => {
280 column_privileges.extend(assignments.iter().map(|assignment| {
281 (
282 crate::catalog::security::table::TableAclPrivilege::Update,
283 assignment.column.clone(),
284 )
285 }));
286 }
287 MergeWhenPlan::DeleteMatched { .. }
288 | MergeWhenPlan::DeleteNotMatchedBySource { .. } => requires_delete = true,
289 _ => {}
290 }
291 }
292 if requires_delete {
293 catalog.ensure_table_privilege_for(
294 &stmt.target,
295 &privilege_subject,
296 crate::catalog::security::table::TableAclPrivilege::Delete,
297 )?;
298 }
299 if requires_any_insert {
300 catalog.ensure_any_column_privilege_for(
301 &stmt.target,
302 &privilege_subject,
303 crate::catalog::security::table::TableAclPrivilege::Insert,
304 )?;
305 }
306 for (privilege, column) in column_privileges {
307 catalog.ensure_column_privilege_for(
308 &stmt.target,
309 &column,
310 &privilege_subject,
311 privilege,
312 )?;
313 }
314 Ok(())
315}
316
317pub fn merge_returning_schema(
318 routines: &dyn crate::routines::RoutineResolution,
319 catalog: &dyn super::returning::ReturningCatalog,
320 stmt: &MergePlan,
321 params: &[SQLParam],
322 source_schema: &crate::RowSchema,
323 ctes: &crate::binding::context::BindingContext<'_>,
324) -> Result<Option<crate::RowSchema>, SQLError> {
325 if stmt.returning.is_empty() {
326 return Ok(None);
327 }
328 let source_relation = crate::ast::InternalRelationId::allocate();
329 let projections = expanded_merge_returning_projections(
330 catalog,
331 &stmt.target,
332 &stmt.target_qualifier,
333 &stmt.returning_aliases,
334 source_schema,
335 source_relation,
336 &stmt.returning,
337 )?;
338 let returning_source_schema = merge_returning_source_schema(source_schema, source_relation);
339 let star_schema = super::returning::returning_target_schema(catalog, &stmt.target)?;
340 let expression_schema = super::returning_expression_schema(
341 &star_schema,
342 &stmt.target_qualifier,
343 &stmt.returning_aliases,
344 Some(&returning_source_schema),
345 );
346 crate::binding::analyze_projection_output_schema(
347 routines,
348 &projections,
349 &expression_schema,
350 &star_schema,
351 &stmt.subqueries,
352 params,
353 ctes,
354 )
355 .map(Some)
356}
357
358pub fn merge_command_returning_schema(
359 routines: &dyn crate::routines::RoutineResolution,
360 catalog: &dyn super::returning::ReturningCatalog,
361 rows: &dyn super::mutation_rows::MutationRowCatalog,
362 stmt: &MergePlan,
363 params: &[SQLParam],
364 bindings: &crate::binding::context::BindingContext<'_>,
365) -> Result<Option<crate::RowSchema>, SQLError> {
366 if stmt.returning.is_empty() {
367 return Ok(None);
368 }
369 let source =
370 crate::binding::analyze_source_plan_schema(routines, &stmt.source, params, bindings, None)?;
371 super::returning::validate_returning_alias_relations(
372 &stmt.target_qualifier,
373 &stmt.returning_aliases,
374 Some(&source),
375 )?;
376 let target =
377 super::mutation_rows::null_target_schema(rows, &stmt.target, &stmt.target_qualifier)?;
378 validate_merge_action_scopes(routines, stmt, &target, &source, params, bindings)?;
379 merge_returning_schema(routines, catalog, stmt, params, &source, bindings)
380}
381
382pub fn validate_merge_target_columns(
383 catalog: &dyn crate::assignment::columns::AssignmentColumnCatalog,
384 stmt: &MergePlan,
385) -> Result<(), SQLError> {
386 use crate::assignment::columns::validate_mutation_columns;
387 for clause in &stmt.when_clauses {
388 match clause {
389 MergeWhenPlan::UpdateMatched { assignments, .. }
390 | MergeWhenPlan::UpdateNotMatchedBySource { assignments, .. } => {
391 validate_mutation_columns(
392 catalog,
393 &stmt.target,
394 assignments
395 .iter()
396 .map(|assignment| assignment.column.as_str()),
397 "MERGE UPDATE",
398 )?;
399 }
400 MergeWhenPlan::InsertNotMatched { columns, .. } => validate_mutation_columns(
401 catalog,
402 &stmt.target,
403 columns.iter().map(String::as_str),
404 "MERGE INSERT",
405 )?,
406 _ => {}
407 }
408 }
409 Ok(())
410}