uqa_sql/expr/composites/
retained.rs

1//
2// Unified Query Algebra
3//
4// Copyright (c) 2023-2026 Cognica, Inc.
5//
6
7//! Original composite inputs shared by prepared plans and durable expression constants.
8
9use super::{datum, descriptor, CompositeTypeDescriptor};
10use crate::{ColumnType, SQLError};
11use std::{collections::BTreeMap, sync::Arc};
12use uqa_core::{ArrayValue, Value};
13
14mod tuple;
15
16#[cfg(test)]
17mod tests;
18
19pub(crate) type Descriptors = BTreeMap<u32, Arc<CompositeTypeDescriptor>>;
20
21/// The original admitted datum and its field types, retained independently of subsequent descriptor interpretations.
22#[derive(Clone, Debug, serde::Serialize, serde::Deserialize)]
23pub struct CompositeConstantSource {
24    pub value: Value,
25    pub descriptors: Vec<CompositeTypeDescriptor>,
26}
27
28impl PartialEq for CompositeConstantSource {
29    fn eq(&self, other: &Self) -> bool {
30        self.descriptors == other.descriptors && self.value.has_same_representation(&other.value)
31    }
32}
33
34impl CompositeConstantSource {
35    pub(crate) fn capture(
36        value: &Value,
37        ty: &ColumnType,
38        catalog: Option<&dyn super::CompositeTypeCatalog>,
39        enums: Option<&dyn crate::expr::enums::EnumLabelCatalog>,
40    ) -> Result<Self, SQLError> {
41        let mut descriptors = Descriptors::new();
42        capture_type(ty, catalog, &mut descriptors)?;
43        Ok(Self {
44            value: crate::expr::enums::retain_enum_oids(enums, value)?,
45            descriptors: descriptors
46                .into_values()
47                .map(|value| value.as_ref().clone())
48                .collect(),
49        })
50    }
51
52    pub(crate) fn retain_enum_oids(
53        &mut self,
54        catalog: Option<&dyn crate::expr::enums::EnumLabelCatalog>,
55    ) -> Result<(), SQLError> {
56        if crate::expr::enums::has_missing_enum_oid(&self.value) {
57            self.value = crate::expr::enums::retain_enum_oids(catalog, &self.value)?;
58        }
59        Ok(())
60    }
61
62    pub(crate) fn project_type_change(
63        &self,
64        ty: &ColumnType,
65        target: u32,
66        name: &str,
67        to: &ColumnType,
68        catalog: &dyn super::CompositeTypeCatalog,
69    ) -> Result<Value, SQLError> {
70        let original = self
71            .descriptors
72            .iter()
73            .map(|descriptor| (descriptor.type_oid, Arc::new(descriptor.clone())))
74            .collect::<Descriptors>();
75        let mut current = Descriptors::new();
76        for oid in original.keys() {
77            if let Some(descriptor) = catalog.composite_type(*oid)? {
78                current.insert(*oid, descriptor);
79            }
80        }
81        if let Some(descriptor) = current.get_mut(&target) {
82            if let Some(attribute) = Arc::make_mut(descriptor)
83                .attributes
84                .iter_mut()
85                .find(|attribute| attribute.name == name)
86            {
87                attribute.ty = to.clone();
88            }
89        }
90        project_value(&self.value, ty, &original, &current, true)
91    }
92}
93
94pub(crate) fn capture_type(
95    ty: &ColumnType,
96    catalog: Option<&dyn super::CompositeTypeCatalog>,
97    output: &mut Descriptors,
98) -> Result<(), SQLError> {
99    match ty {
100        ColumnType::Domain { base, .. } | ColumnType::Array(base) => {
101            capture_type(base, catalog, output)
102        }
103        ColumnType::Composite(reference) => capture_descriptor(reference.oid, catalog, output),
104        _ => Ok(()),
105    }
106}
107
108pub(crate) fn capture_descriptor(
109    oid: u32,
110    catalog: Option<&dyn super::CompositeTypeCatalog>,
111    output: &mut Descriptors,
112) -> Result<(), SQLError> {
113    if output.contains_key(&oid) {
114        return Ok(());
115    }
116    let descriptor = descriptor(catalog, oid)?;
117    output.insert(oid, descriptor.clone());
118    for attribute in &descriptor.attributes {
119        capture_type(&attribute.ty, catalog, output)?;
120    }
121    Ok(())
122}
123
124pub(crate) fn project_value(
125    value: &Value,
126    ty: &ColumnType,
127    original: &Descriptors,
128    current: &Descriptors,
129    interpret_datums: bool,
130) -> Result<Value, SQLError> {
131    match (ty, value) {
132        (_, Value::Null) => Ok(Value::Null),
133        (ColumnType::Domain { base, .. }, _) => {
134            project_value(value, base, original, current, interpret_datums)
135        }
136        (ColumnType::Composite(reference), Value::Record(fields)) => {
137            let type_oid = fields.type_oid().unwrap_or(reference.oid);
138            let (Some(before), Some(after)) = (original.get(&type_oid), current.get(&type_oid))
139            else {
140                return Ok(value.clone());
141            };
142            if interpret_datums {
143                if let Some(Value::Record(mut projected)) =
144                    tuple::project(fields, before, after, original)
145                {
146                    for (attribute, (_, value)) in after.attributes.iter().zip(&mut projected) {
147                        let Some(old) = before
148                            .attributes
149                            .iter()
150                            .find(|old| old.number == attribute.number)
151                        else {
152                            continue;
153                        };
154                        let mut leaf = &old.ty;
155                        while let ColumnType::Domain { base, .. } | ColumnType::Array(base) = leaf {
156                            leaf = base;
157                        }
158                        if matches!(leaf, ColumnType::Composite(_)) {
159                            // Unchanged-position records and arrays still observe changes to their nested descriptors. Their original type identifies the retained payload even if the outer declaration changed.
160                            *value = project_value(value, &old.ty, original, current, true)?;
161                        }
162                    }
163                    return Ok(Value::Record(projected));
164                }
165            }
166            after
167                .attributes
168                .iter()
169                .map(|attribute| {
170                    let retained = before
171                        .attributes
172                        .iter()
173                        .find(|old| old.number == attribute.number)
174                        .and_then(|old| {
175                            fields
176                                .iter()
177                                .find(|(name, _)| *name == old.name)
178                                .map(|(_, value)| (old, value))
179                        });
180                    let value = retained.map_or(Ok(Value::Null), |(old, value)| {
181                        let reinterpreted = interpret_datums
182                            .then(|| datum::reinterpret(value, &old.ty, &attribute.ty))
183                            .flatten();
184                        project_value(
185                            reinterpreted.as_ref().unwrap_or(value),
186                            &attribute.ty,
187                            original,
188                            current,
189                            interpret_datums,
190                        )
191                    })?;
192                    Ok((attribute.name.clone(), value))
193                })
194                .collect::<Result<Vec<_>, _>>()
195                .map(|output| {
196                    Value::Record(uqa_core::RecordValue::from_parts(
197                        output,
198                        Some(fields.type_oid().unwrap_or(reference.oid)),
199                    ))
200                })
201        }
202        (ColumnType::Array(element), Value::Array(array)) => {
203            let values = array
204                .elements()
205                .iter()
206                .map(|value| {
207                    project_array_element(value, element, original, current, interpret_datums)
208                })
209                .collect::<Result<Vec<_>, _>>()?;
210            ArrayValue::with_lower_bounds(values, array.lower_bounds().to_vec())
211                .map(|converted| converted.with_element_type_oid(array.element_type_oid()))
212                .map(Value::Array)
213                .ok_or_else(|| {
214                    SQLError::Internal("retained composite projection changed array shape".into())
215                })
216        }
217        _ => Ok(value.clone()),
218    }
219}
220
221fn project_array_element(
222    value: &Value,
223    element: &ColumnType,
224    original: &Descriptors,
225    current: &Descriptors,
226    interpret_datums: bool,
227) -> Result<Value, SQLError> {
228    match value {
229        Value::List(values) => values
230            .iter()
231            .map(|value| project_array_element(value, element, original, current, interpret_datums))
232            .collect::<Result<Vec<_>, _>>()
233            .map(Value::List),
234        value => project_value(value, element, original, current, interpret_datums),
235    }
236}
uqa_sql/expr/composites/retained.rs

uqa_sql/expr/composites/
retained.rs
