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

uqa_sql/expr/composites/
retained.rs
