uqa_sql/schema/inheritance/
alter.rs1use crate::ast::{
9 AutoIncrement, BinaryOp, ColumnDef, DetachedPartitionConstraint, Expr, ForeignKey,
10 PartitionBound, PartitionIdentityOverride, PartitionRangeDatum, PartitionSpec,
11 RelationPersistence, TableCheck, TableHierarchy, TableKeyConstraint,
12};
13use crate::SQLError;
14use uqa_core::Value;
15pub fn validate_row_type(
16 parent_columns: &[ColumnDef],
17 child_columns: &[ColumnDef],
18 parent: &str,
19 child: &str,
20 exact_columns: bool,
21 reject_child_identity: bool,
22) -> Result<(), SQLError> {
23 if reject_child_identity {
24 if let Some(column) = child_columns.iter().find(|column| {
25 column
26 .auto_increment
27 .as_ref()
28 .is_some_and(AutoIncrement::is_identity)
29 }) {
30 return Err(routine(
31 "55000",
32 format!(
33 "table \"{}\" being attached contains an identity column \"{}\"\nDETAIL: The new partition may not contain an identity column.",
34 local_relation_name(child),
35 column.name
36 ),
37 ));
38 }
39 }
40 for parent_column in parent_columns {
41 let Some(child_column) = child_columns
42 .iter()
43 .find(|column| column.name == parent_column.name)
44 else {
45 return Err(routine(
46 "42804",
47 format!("child table is missing column \"{}\"", parent_column.name),
48 ));
49 };
50 if parent_column.ty != child_column.ty {
51 return Err(routine(
52 "42804",
53 format!(
54 "child table \"{}\" has different type for column \"{}\"",
55 local_relation_name(child),
56 parent_column.name
57 ),
58 ));
59 }
60 if parent_column.not_null && !child_column.not_null {
61 return Err(routine(
62 "42804",
63 format!(
64 "column \"{}\" in child table \"{}\" must be marked NOT NULL",
65 parent_column.name,
66 local_relation_name(child)
67 ),
68 ));
69 }
70 match (&parent_column.generated, &child_column.generated) {
71 (None, None) | (Some(_), Some(_)) => {}
72 (Some(_), None) => {
73 return Err(routine(
74 "42804",
75 format!(
76 "column \"{}\" in child table must be a generated column",
77 parent_column.name
78 ),
79 ))
80 }
81 (None, Some(_)) => {
82 return Err(routine(
83 "42804",
84 format!(
85 "column \"{}\" in child table must not be a generated column",
86 parent_column.name
87 ),
88 ))
89 }
90 }
91 if let (Some(parent_generated), Some(child_generated)) =
92 (&parent_column.generated, &child_column.generated)
93 {
94 if parent_generated.kind != child_generated.kind {
95 return Err(routine(
96 "42804",
97 format!(
98 "column \"{}\" inherits from generated column of different kind",
99 parent_column.name
100 ),
101 ));
102 }
103 }
104 }
105 if exact_columns {
106 if let Some(extra) = child_columns.iter().find(|child_column| {
107 !parent_columns
108 .iter()
109 .any(|parent_column| parent_column.name == child_column.name)
110 }) {
111 return Err(routine(
112 "42804",
113 format!(
114 "table \"{}\" contains column \"{}\" not found in parent \"{}\"\nDETAIL: The new partition may contain only the columns present in parent.",
115 local_relation_name(child),
116 extra.name,
117 local_relation_name(parent)
118 ),
119 ));
120 }
121 }
122 Ok(())
123}
124
125pub fn validate_inherited_checks(
126 child: &str,
127 child_columns: &[ColumnDef],
128 parent_checks: &[TableCheck],
129 child_checks: &[TableCheck],
130) -> Result<(), SQLError> {
131 for parent_check in parent_checks
132 .iter()
133 .filter(|constraint| !constraint.no_inherit)
134 {
135 let Some(name) = parent_check.name.as_deref() else {
136 return Err(SQLError::Internal(
137 "persisted parent CHECK constraint has no name".into(),
138 ));
139 };
140 let Some(child_check) = child_checks
141 .iter()
142 .find(|constraint| constraint.name.as_deref() == Some(name))
143 else {
144 return Err(routine(
145 "42804",
146 format!("child table is missing constraint \"{name}\""),
147 ));
148 };
149 if !crate::schema::check_inheritance::same_check_expression(
150 &child_check.expr,
151 &parent_check.expr,
152 child_columns,
153 )? {
154 return Err(routine(
155 "42804",
156 format!(
157 "child table \"{}\" has different definition for check constraint \"{name}\"",
158 local_relation_name(child)
159 ),
160 ));
161 }
162 let conflict = if child_check.no_inherit {
163 Some("non-inherited")
164 } else if parent_check.validated && child_check.enforced && !child_check.validated {
165 Some("NOT VALID")
166 } else if parent_check.enforced && !child_check.enforced {
167 Some("NOT ENFORCED")
168 } else {
169 None
170 };
171 if let Some(conflict) = conflict {
172 return Err(routine("42P17", format!("constraint \"{name}\" conflicts with {conflict} constraint on child table \"{}\"", local_relation_name(child))));
173 }
174 }
175 Ok(())
176}
177
178pub fn install_inherited_identity(
179 columns: &mut [ColumnDef],
180 inherited: &[(String, AutoIncrement)],
181) -> Result<Vec<PartitionIdentityOverride>, SQLError> {
182 let mut overrides = Vec::with_capacity(inherited.len());
183 for (name, increment) in inherited {
184 let column = columns
185 .iter_mut()
186 .find(|column| column.name == *name)
187 .ok_or_else(|| SQLError::Internal(format!("partition lost column `{name}`")))?;
188 overrides.push(PartitionIdentityOverride {
189 column: name.clone(),
190 original: column.auto_increment.clone(),
191 });
192 column.auto_increment = Some(increment.clone());
193 }
194 Ok(overrides)
195}
196
197pub fn restore_identity_overrides(
198 columns: &mut [ColumnDef],
199 inherited: &[(String, AutoIncrement)],
200 overrides: &[PartitionIdentityOverride],
201) {
202 for (name, _) in inherited {
203 let Some(column) = columns.iter_mut().find(|column| column.name == *name) else {
204 continue;
205 };
206 column.auto_increment = overrides
207 .iter()
208 .find(|identity_override| identity_override.column == *name)
209 .and_then(|identity_override| identity_override.original.clone());
210 }
211}
212
213pub fn append_inherited_keys(
214 target: &mut Vec<TableKeyConstraint>,
215 inherited: &[TableKeyConstraint],
216) -> Vec<TableKeyConstraint> {
217 let mut appended = Vec::new();
218 for constraint in inherited {
219 if target
220 .iter()
221 .any(|candidate| key_equivalent(candidate, constraint))
222 {
223 continue;
224 }
225 let mut constraint = constraint.clone();
226 constraint.name = None;
227 target.push(constraint.clone());
228 appended.push(constraint);
229 }
230 appended
231}
232
233fn key_equivalent(left: &TableKeyConstraint, right: &TableKeyConstraint) -> bool {
234 left.kind == right.kind
235 && left.columns == right.columns
236 && left.nulls_not_distinct == right.nulls_not_distinct
237}
238
239pub fn append_inherited_foreign_keys(
240 target: &mut Vec<ForeignKey>,
241 inherited: &[ForeignKey],
242) -> Vec<ForeignKey> {
243 let mut appended = Vec::new();
244 for constraint in inherited {
245 if !target
246 .iter()
247 .any(|candidate| foreign_key_equivalent(candidate, constraint))
248 {
249 target.push(constraint.clone());
250 appended.push(constraint.clone());
251 }
252 }
253 appended
254}
255
256pub fn remove_partition_inherited_constraints(constraints: &mut crate::ast::TableConstraintSet) {
257 for inherited in &constraints.hierarchy.partition_inherited_key_constraints {
258 if let Some(index) = constraints
259 .key_constraints
260 .iter()
261 .position(|constraint| constraint == inherited)
262 {
263 constraints.key_constraints.remove(index);
264 }
265 }
266 for inherited in &constraints.hierarchy.partition_inherited_foreign_keys {
267 if let Some(index) = constraints
268 .foreign_keys
269 .iter()
270 .position(|constraint| constraint == inherited)
271 {
272 constraints.foreign_keys.remove(index);
273 }
274 }
275}
276
277fn foreign_key_equivalent(left: &ForeignKey, right: &ForeignKey) -> bool {
278 left.local_columns == right.local_columns
279 && left.ref_table == right.ref_table
280 && left.ref_columns == right.ref_columns
281 && left.on_update == right.on_update
282 && left.on_delete == right.on_delete
283 && left.on_delete_set_columns == right.on_delete_set_columns
284 && left.match_type == right.match_type
285 && left.enforced == right.enforced
286}
287
288pub fn detached_bound_check(
289 table: &str,
290 spec: &PartitionSpec,
291 bound: &PartitionBound,
292 existing: &[TableCheck],
293) -> TableCheck {
294 let expr = renderable_bound_expression(spec, bound);
295 let relation = local_relation_name(table);
296 let key = spec.keys.first().and_then(|key| match key {
297 Expr::Column(column) => Some(column.as_str()),
298 _ => None,
299 });
300 let base = key.map_or_else(
301 || format!("{relation}_check"),
302 |column| format!("{relation}_{column}_check"),
303 );
304 let name = unique_constraint_name(&base, existing);
305 TableCheck {
306 name: Some(name),
307 expr,
308 enforced: true,
309 validated: true,
310 no_inherit: false,
311 object_id: None,
312 is_local: true,
313 partition_constraint: Some(DetachedPartitionConstraint {
314 spec: spec.clone(),
315 bound: bound.clone(),
316 }),
317 }
318}
319
320fn unique_constraint_name(base: &str, existing: &[TableCheck]) -> String {
321 if !existing
322 .iter()
323 .any(|constraint| constraint.name.as_deref() == Some(base))
324 {
325 return base.to_string();
326 }
327 for suffix in 1_u64.. {
328 let candidate = format!("{base}{suffix}");
329 if !existing
330 .iter()
331 .any(|constraint| constraint.name.as_deref() == Some(candidate.as_str()))
332 {
333 return candidate;
334 }
335 }
336 unreachable!("u64 constraint suffix space is exhaustive")
337}
338
339fn renderable_bound_expression(spec: &PartitionSpec, bound: &PartitionBound) -> Expr {
340 let Some(key) = spec.keys.first().cloned().filter(|_| spec.keys.len() == 1) else {
341 return Expr::Literal(Value::Bool(true));
342 };
343 match bound {
344 PartitionBound::List(values) => {
345 let mut terms = Vec::new();
346 let mut non_null = Vec::new();
347 for value in values {
348 if matches!(value, Expr::Literal(Value::Null)) {
349 terms.push(Expr::IsNull {
350 expr: Box::new(key.clone()),
351 negated: false,
352 });
353 } else {
354 non_null.push(value.clone());
355 }
356 }
357 if !non_null.is_empty() {
358 terms.push(Expr::InList {
359 expr: Box::new(key),
360 list: non_null,
361 negated: false,
362 });
363 }
364 if terms.len() == 1 {
365 terms.pop().unwrap_or(Expr::Literal(Value::Bool(true)))
366 } else {
367 Expr::Or(terms)
368 }
369 }
370 PartitionBound::Range { lower, upper } if lower.len() == 1 && upper.len() == 1 => {
371 let mut terms = vec![Expr::IsNull {
372 expr: Box::new(key.clone()),
373 negated: true,
374 }];
375 if let PartitionRangeDatum::Value(lower) = &lower[0] {
376 terms.push(Expr::Binary {
377 op: BinaryOp::GreaterEqual,
378 lhs: Box::new(key.clone()),
379 rhs: Box::new(lower.clone()),
380 });
381 }
382 if let PartitionRangeDatum::Value(upper) = &upper[0] {
383 terms.push(Expr::Binary {
384 op: BinaryOp::Less,
385 lhs: Box::new(key),
386 rhs: Box::new(upper.clone()),
387 });
388 }
389 Expr::And(terms)
390 }
391 PartitionBound::Hash { .. } | PartitionBound::Range { .. } | PartitionBound::Default => {
392 Expr::Literal(Value::Bool(true))
393 }
394 }
395}
396
397pub fn validate_matching_persistence(
398 child: &str,
399 parent: &str,
400 operation: &str,
401 child_persistence: RelationPersistence,
402 parent_persistence: RelationPersistence,
403) -> Result<(), SQLError> {
404 if (child_persistence == RelationPersistence::Temporary)
405 != (parent_persistence == RelationPersistence::Temporary)
406 {
407 return Err(wrong_object(format!(
408 "cannot {operation} {} relation \"{}\" from {} relation \"{}\"",
409 persistence_label(child_persistence),
410 local_relation_name(child),
411 persistence_label(parent_persistence),
412 local_relation_name(parent)
413 )));
414 }
415 Ok(())
416}
417
418fn persistence_label(persistence: RelationPersistence) -> &'static str {
419 match persistence {
420 RelationPersistence::Temporary => "temporary",
421 RelationPersistence::Unlogged => "unlogged",
422 RelationPersistence::Permanent => "permanent",
423 }
424}
425
426pub fn normalize_parent_sequence_numbers(hierarchy: &mut TableHierarchy) {
427 if hierarchy.parent_sequence_numbers.len() == hierarchy.parents.len() {
428 return;
429 }
430 hierarchy.parent_sequence_numbers = hierarchy
431 .parents
432 .iter()
433 .enumerate()
434 .map(|(index, _)| i32::try_from(index + 1).unwrap_or(i32::MAX))
435 .collect();
436}
437
438fn local_relation_name(name: &str) -> &str {
439 name.rsplit('.').next().unwrap_or(name)
440}
441fn wrong_object(message: impl Into<String>) -> SQLError {
442 routine("42809", message)
443}
444fn routine(sqlstate: &str, message: impl Into<String>) -> SQLError {
445 SQLError::Routine {
446 sqlstate: sqlstate.into(),
447 message: message.into(),
448 }
449}