1use super::{
8 add_check_option, automatic_view_layer, combine_view_predicate, display_relation,
9 dml_analysis_scope, duplicate_assignment, duplicate_insert_column, finalize_source_returning,
10 merge_action_capability_error, merge_matched_subquery_ids, merge_target_only_subquery_ids,
11 merge_view_target_path, not_automatically_updatable, returning_subquery_ids,
12 rewrite_correlated_dml_context, rewrite_existing_view_checks, rewrite_merge_returning,
13 rewrite_target_expression, validate_mapped_columns, validate_merge_expressions,
14 validate_merge_targets, validate_public_merge_contract, validate_public_merge_targets,
15 view_updatability, writable_column, BTreeSet, CorrelatedDmlContext, ExpressionScope, MergePlan,
16 MergeViewTargetPath, MergeWhenPlan, SQLError, StoredViewKind, ViewMutationCapabilities,
17 ViewRewriteContext,
18};
19
20#[expect(
21 clippy::too_many_lines,
22 reason = "preserves view qualifier and row identity"
23)]
24pub fn rewrite_merge_to_base(
25 services: ViewRewriteContext<'_>,
26 statement: &MergePlan,
27 params: &[crate::SQLParam],
28 inherited_ctes: Option<&super::CteScope>,
29) -> Result<MergePlan, SQLError> {
30 if services.catalog.target_view_kind(&statement.target)? == Some(StoredViewKind::Materialized) {
31 return Err(SQLError::Routine {
32 sqlstate: "0A000".into(),
33 message: format!(
34 "cannot execute MERGE on relation \"{}\"",
35 display_relation(&statement.target)
36 ),
37 });
38 }
39 let analysis_scope = dml_analysis_scope(
40 services,
41 &statement.ctes,
42 &statement.subqueries,
43 inherited_ctes,
44 )?;
45 let source_schema = crate::semantics::view_rewrite::context::analyze_source_plan_schema(
46 services,
47 &statement.source,
48 params,
49 &analysis_scope,
50 None,
51 )?;
52 validate_public_merge_targets(services, statement)?;
53 validate_public_merge_contract(services, statement, &source_schema)?;
54 if merge_view_target_path(services, statement)? != MergeViewTargetPath::AutomaticRewrite {
55 return Err(SQLError::Internal(
56 "automatic MERGE rewrite selected for a view-trigger target".into(),
57 ));
58 }
59 let Some(initial_layer) = automatic_view_layer(services, &statement.target)? else {
60 return Err(merge_action_capability_error(
61 &statement.target,
62 &statement.when_clauses,
63 ViewMutationCapabilities::default(),
64 )
65 .unwrap_or_else(|| not_automatically_updatable(&statement.target, "MERGE")));
66 };
67 validate_merge_targets(&initial_layer, statement)?;
68 validate_merge_expressions(
69 services,
70 statement,
71 &initial_layer,
72 &source_schema,
73 params,
74 inherited_ctes,
75 )?;
76 if let Some(error) = merge_action_capability_error(
77 &statement.target,
78 &statement.when_clauses,
79 view_updatability(services, &statement.target)?.automatic,
80 ) {
81 return Err(error);
82 }
83
84 let mut plan = statement.clone();
85 let next_privilege_subject =
86 crate::semantics::view_privileges::ensure_merge(services.authorization, &plan)?;
87 plan.target_privilege_subject = Some(next_privilege_subject);
88 let mut cascaded = false;
89 let mut visited = BTreeSet::new();
90 let mut source_star_boundaries = Vec::new();
91 loop {
92 if !visited.is_empty()
93 && merge_view_target_path(services, &plan)? == MergeViewTargetPath::ViewTriggers
94 {
95 break;
96 }
97 let Some(layer) = automatic_view_layer(services, &plan.target)? else {
98 return Err(merge_action_capability_error(
99 &plan.target,
100 &plan.when_clauses,
101 ViewMutationCapabilities::default(),
102 )
103 .unwrap_or_else(|| not_automatically_updatable(&plan.target, "MERGE")));
104 };
105 if !visited.insert(layer.canonical_name.clone()) {
106 return Err(SQLError::Internal(format!(
107 "cycle while rewriting automatically updatable view `{}`",
108 layer.canonical_name
109 )));
110 }
111 validate_merge_targets(&layer, &plan)?;
112 if visited.len() > 1 {
113 let next_privilege_subject =
114 crate::semantics::view_privileges::ensure_merge(services.authorization, &plan)?;
115 plan.target_privilege_subject = Some(next_privilege_subject);
116 }
117
118 let matched_subqueries = merge_matched_subquery_ids(&plan);
119 rewrite_correlated_dml_context(
120 CorrelatedDmlContext {
121 inherited_ctes,
122 services,
123 layer: &layer,
124 target_qualifier: &plan.target_qualifier,
125 source: Some(&source_schema),
126 returning_aliases: None,
127 include_excluded: false,
128 ctes: &plan.ctes,
129 ids: &matched_subqueries,
130 params,
131 },
132 &mut plan.subqueries,
133 )?;
134 let target_only_subqueries = merge_target_only_subquery_ids(&plan);
135 rewrite_correlated_dml_context(
136 CorrelatedDmlContext {
137 inherited_ctes,
138 services,
139 layer: &layer,
140 target_qualifier: &plan.target_qualifier,
141 source: None,
142 returning_aliases: None,
143 include_excluded: false,
144 ctes: &plan.ctes,
145 ids: &target_only_subqueries,
146 params,
147 },
148 &mut plan.subqueries,
149 )?;
150 let returning_subqueries = returning_subquery_ids(&plan.returning);
151 rewrite_correlated_dml_context(
152 CorrelatedDmlContext {
153 inherited_ctes,
154 services,
155 layer: &layer,
156 target_qualifier: &plan.target_qualifier,
157 source: Some(&source_schema),
158 returning_aliases: Some(&plan.returning_aliases),
159 include_excluded: false,
160 ctes: &plan.ctes,
161 ids: &returning_subqueries,
162 params,
163 },
164 &mut plan.subqueries,
165 )?;
166 let matched_scope = ExpressionScope {
167 target_qualifier: &plan.target_qualifier,
168 returning_aliases: None,
169 source: Some(&source_schema),
170 include_excluded: false,
171 };
172 let target_only_scope = ExpressionScope {
173 source: None,
174 ..matched_scope
175 };
176 rewrite_target_expression(
177 services,
178 &mut plan.join_condition,
179 &layer,
180 matched_scope,
181 &mut plan.subqueries,
182 )?;
183 if let Some(predicate) = &mut plan.target_predicate {
184 rewrite_target_expression(
185 services,
186 predicate,
187 &layer,
188 target_only_scope,
189 &mut plan.subqueries,
190 )?;
191 }
192 for clause in &mut plan.when_clauses {
193 match clause {
194 MergeWhenPlan::UpdateMatched {
195 condition,
196 assignments,
197 } => {
198 if let Some(condition) = condition {
199 rewrite_target_expression(
200 services,
201 condition,
202 &layer,
203 matched_scope,
204 &mut plan.subqueries,
205 )?;
206 }
207 for assignment in assignments.iter_mut() {
208 rewrite_target_expression(
209 services,
210 &mut assignment.value,
211 &layer,
212 matched_scope,
213 &mut plan.subqueries,
214 )?;
215 assignment.column =
216 writable_column(&layer, &assignment.column, "MERGE INTO")?;
217 }
218 validate_mapped_columns(
219 &assignments
220 .iter()
221 .map(|assignment| assignment.column.clone())
222 .collect::<Vec<_>>(),
223 duplicate_assignment,
224 )?;
225 }
226 MergeWhenPlan::DeleteMatched { condition }
227 | MergeWhenPlan::NothingMatched { condition } => {
228 if let Some(condition) = condition {
229 rewrite_target_expression(
230 services,
231 condition,
232 &layer,
233 matched_scope,
234 &mut plan.subqueries,
235 )?;
236 }
237 }
238 MergeWhenPlan::UpdateNotMatchedBySource {
239 condition,
240 assignments,
241 } => {
242 if let Some(condition) = condition {
243 rewrite_target_expression(
244 services,
245 condition,
246 &layer,
247 target_only_scope,
248 &mut plan.subqueries,
249 )?;
250 }
251 for assignment in assignments.iter_mut() {
252 rewrite_target_expression(
253 services,
254 &mut assignment.value,
255 &layer,
256 target_only_scope,
257 &mut plan.subqueries,
258 )?;
259 assignment.column =
260 writable_column(&layer, &assignment.column, "MERGE INTO")?;
261 }
262 validate_mapped_columns(
263 &assignments
264 .iter()
265 .map(|assignment| assignment.column.clone())
266 .collect::<Vec<_>>(),
267 duplicate_assignment,
268 )?;
269 }
270 MergeWhenPlan::DeleteNotMatchedBySource { condition }
271 | MergeWhenPlan::NothingNotMatchedBySource { condition } => {
272 if let Some(condition) = condition {
273 rewrite_target_expression(
274 services,
275 condition,
276 &layer,
277 target_only_scope,
278 &mut plan.subqueries,
279 )?;
280 }
281 }
282 MergeWhenPlan::InsertNotMatched {
283 columns, values, ..
284 } => {
285 let supplied_columns = if columns.is_empty() {
286 layer
287 .columns
288 .iter()
289 .take(values.len())
290 .map(|column| column.name.clone())
291 .collect::<Vec<_>>()
292 } else {
293 columns.clone()
294 };
295 *columns = supplied_columns
296 .iter()
297 .map(|column| writable_column(&layer, column, "MERGE INTO"))
298 .collect::<Result<Vec<_>, SQLError>>()?;
299 validate_mapped_columns(columns, duplicate_insert_column)?;
300 }
301 MergeWhenPlan::NothingNotMatched { .. } => {}
302 }
303 }
304 rewrite_existing_view_checks(
305 services,
306 &mut plan.view_checks,
307 &layer,
308 &plan.target_qualifier,
309 &mut plan.subqueries,
310 )?;
311 let (returning, boundaries) = rewrite_merge_returning(
312 services,
313 plan.returning,
314 &layer,
315 &plan.target_qualifier,
316 &plan.returning_aliases,
317 &source_schema,
318 &mut plan.subqueries,
319 )?;
320 plan.returning = returning;
321 if visited.len() == 1 {
322 source_star_boundaries = boundaries;
323 }
324 plan.target_predicate = combine_view_predicate(
325 services,
326 plan.target_predicate,
327 &layer,
328 &plan.target_qualifier,
329 &mut plan.subqueries,
330 )?;
331 add_check_option(
332 services,
333 &mut plan.view_checks,
334 &layer,
335 &plan.target_qualifier,
336 &mut cascaded,
337 &mut plan.subqueries,
338 )?;
339 plan.target = layer.source_name;
340 plan.include_descendants = layer.source_include_descendants;
341 if !super::context::target_is_view(services, &plan.target)? {
342 break;
343 }
344 }
345 plan.returning = finalize_source_returning(
346 services,
347 &plan.target,
348 plan.returning,
349 Some(&source_schema),
350 &source_star_boundaries,
351 )?;
352 Ok(plan)
353}