uqa_sql/schema/
schema_composition.rs

1//
2// Unified Query Algebra
3//
4// Copyright (c) 2023-2026 Cognica, Inc.
5//
6
7//! Schema append, join, view, and physical relayout operations.
8
9use super::{
10    ColumnIdentity, ColumnType, RowSchema, SchemaBuildMetadata, SchemaLayoutError,
11    SchemaLayoutResult, NULL_SLOT,
12};
13
14impl RowSchema {
15    /// Append freshly-computed values to an existing physical row. Reusing an
16    /// existing output name replaces its logical slot just like map insertion.
17    pub fn append(input: &Self, names: &[String]) -> Self {
18        let columns = names
19            .iter()
20            .cloned()
21            .map(|name| (name, None))
22            .collect::<Vec<_>>();
23        Self::append_typed(input, &columns)
24    }
25
26    /// Append freshly computed values with static SQL output types.
27    pub fn append_typed(input: &Self, names: &[(String, Option<ColumnType>)]) -> Self {
28        let mut columns = input.columns().to_vec();
29        let mut identities = input.identities().to_vec();
30        let mut types = input.column_types().to_vec();
31        let mut slots = input.index.slots.to_vec();
32        let mut wildcard_hidden = input.index.cold.wildcard_hidden.clone();
33        let base = input.physical_width();
34        for (offset, (name, ty)) in names.iter().enumerate() {
35            let slot = base + offset;
36            if let Some(position) = columns.iter().position(|column| column == name) {
37                slots[position] = slot;
38                identities[position] = ColumnIdentity::unqualified(name);
39                types[position].clone_from(ty);
40                wildcard_hidden.remove(&position);
41            } else {
42                columns.push(name.clone());
43                identities.push(ColumnIdentity::unqualified(name));
44                types.push(ty.clone());
45                slots.push(slot);
46            }
47        }
48        Self::from_typed_parts_with_aliases_and_exact_precedence(
49            columns,
50            identities,
51            types,
52            slots,
53            base + names.len(),
54            SchemaBuildMetadata {
55                aliases: input.index.aliases.clone(),
56                alias_types: input.index.cold.aliases.clone(),
57                internal: input.index.executor_attributes.clone(),
58                internal_types: input.index.cold.executor_attribute_types.clone(),
59                score_sources: input.index.cold.score_sources.clone(),
60                wildcard_hidden,
61                binding_only: input.index.cold.binding_only.clone(),
62                open_qualifiers: input.index.cold.open_qualifiers.clone(),
63                ..SchemaBuildMetadata::default()
64            },
65        )
66    }
67
68    /// Extend the physical row with anonymous values that have no SQL name or
69    /// wildcard presence. Callers may attach structured SQL identities or
70    /// internal relation attributes to the resulting physical slots.
71    pub fn append_hidden_typed(input: &Self, types: &[Option<ColumnType>]) -> Self {
72        Self::from_typed_parts_with_aliases_and_exact_precedence(
73            input.columns().to_vec(),
74            input.identities().to_vec(),
75            input.column_types().to_vec(),
76            input.index.slots.to_vec(),
77            input.physical_width() + types.len(),
78            SchemaBuildMetadata {
79                aliases: input.index.aliases.clone(),
80                alias_types: input.index.cold.aliases.clone(),
81                internal: input.index.executor_attributes.clone(),
82                internal_types: input.index.cold.executor_attribute_types.clone(),
83                score_sources: input.index.cold.score_sources.clone(),
84                wildcard_hidden: input.index.cold.wildcard_hidden.clone(),
85                binding_only: input.index.cold.binding_only.clone(),
86                open_qualifiers: input.index.cold.open_qualifiers.clone(),
87                ..SchemaBuildMetadata::default()
88            },
89        )
90    }
91
92    /// Append computed executor attributes under structural identities while
93    /// keeping them out of the SQL name and wildcard namespaces.
94    pub fn append_internal_typed(
95        input: &Self,
96        columns: &[(crate::ast::InternalColumnRef, Option<ColumnType>)],
97    ) -> Self {
98        let base = input.physical_width();
99        let mut internal = input.index.executor_attributes.clone();
100        let mut internal_types = input.index.cold.executor_attribute_types.clone();
101        for (offset, (column, ty)) in columns.iter().enumerate() {
102            internal.insert(*column, base + offset);
103            internal_types.insert(*column, ty.clone());
104        }
105        Self::from_typed_parts_with_aliases_and_exact_precedence(
106            input.columns().to_vec(),
107            input.identities().to_vec(),
108            input.column_types().to_vec(),
109            input.index.slots.to_vec(),
110            base + columns.len(),
111            SchemaBuildMetadata {
112                aliases: input.index.aliases.clone(),
113                alias_types: input.index.cold.aliases.clone(),
114                internal,
115                internal_types,
116                score_sources: input.index.cold.score_sources.clone(),
117                wildcard_hidden: input.index.cold.wildcard_hidden.clone(),
118                binding_only: input.index.cold.binding_only.clone(),
119                open_qualifiers: input.index.cold.open_qualifiers.clone(),
120                ..SchemaBuildMetadata::default()
121            },
122        )
123    }
124
125    /// Compose two child layouts while retaining duplicate logical labels.
126    /// Qualified and positional resolution can then distinguish both input
127    /// slots without copying either value fragment.
128    pub fn join(
129        left: &Self,
130        right: &Self,
131        extra_columns: impl IntoIterator<Item = String>,
132    ) -> Self {
133        let mut columns = left.columns().to_vec();
134        let mut identities = left.identities().to_vec();
135        let mut types = left.column_types().to_vec();
136        let mut slots = left.index.slots.to_vec();
137        let right_base = left.physical_width();
138        let mut aliases = left.index.aliases.clone();
139        let mut alias_types = left.index.cold.aliases.clone();
140        let mut internal = left.index.executor_attributes.clone();
141        let mut internal_types = left.index.cold.executor_attribute_types.clone();
142        let mut score_sources = left.index.cold.score_sources.clone();
143        let mut wildcard_hidden = left.index.cold.wildcard_hidden.clone();
144        let open_qualifiers = &left.index.cold.open_qualifiers | &right.index.cold.open_qualifiers;
145        let mut binding_only = left.index.cold.binding_only.clone();
146        aliases.extend(right.index.aliases.iter().map(|(name, slot)| {
147            (
148                name.clone(),
149                if *slot == NULL_SLOT {
150                    NULL_SLOT
151                } else {
152                    right_base + *slot
153                },
154            )
155        }));
156        alias_types.extend(
157            right
158                .index
159                .cold
160                .aliases
161                .iter()
162                .map(|(name, ty)| (name.clone(), ty.clone())),
163        );
164        for (column, slot) in &right.index.executor_attributes {
165            let shifted = if *slot == NULL_SLOT {
166                NULL_SLOT
167            } else {
168                right_base + *slot
169            };
170            assert!(
171                internal.insert(*column, shifted).is_none(),
172                "duplicate internal relation attribute in joined row"
173            );
174        }
175        for (column, ty) in &right.index.cold.executor_attribute_types {
176            assert!(
177                internal_types.insert(*column, ty.clone()).is_none(),
178                "duplicate internal relation attribute type in joined row"
179            );
180        }
181        score_sources.extend(right.index.cold.score_sources.iter().cloned());
182        wildcard_hidden.extend(
183            right
184                .index
185                .cold
186                .wildcard_hidden
187                .iter()
188                .map(|position| left.len() + *position),
189        );
190        binding_only.extend(
191            right
192                .index
193                .cold
194                .binding_only
195                .iter()
196                .map(|(identity, ty)| (identity.clone(), ty.clone())),
197        );
198        for (right_logical, column) in right.columns().iter().enumerate() {
199            let slot = right
200                .slot(right_logical)
201                .map_or(NULL_SLOT, |slot| right_base + slot);
202            columns.push(column.clone());
203            identities.push(right.identities()[right_logical].clone());
204            types.push(right.column_type(right_logical).cloned());
205            slots.push(slot);
206        }
207        for column in extra_columns {
208            if !columns.contains(&column) {
209                identities.push(ColumnIdentity::unqualified(column.clone()));
210                columns.push(column);
211                types.push(None);
212                slots.push(NULL_SLOT);
213            }
214        }
215        Self::from_typed_parts_with_aliases_and_exact_precedence(
216            columns,
217            identities,
218            types,
219            slots,
220            left.physical_width() + right.physical_width(),
221            SchemaBuildMetadata {
222                aliases,
223                alias_types,
224                internal,
225                internal_types,
226                score_sources,
227                wildcard_hidden,
228                binding_only,
229                open_qualifiers,
230                ..SchemaBuildMetadata::default()
231            },
232        )
233    }
234
235    /// Map logical and hidden identities into `target`'s complete slot layout. Visible columns are matched by logical position; hidden lookup aliases are matched by their structured identity. This is used when two equivalent operator pipelines expose the same logical row through different physical slot arrangements.
236    pub fn relayout_slots(&self, target: &Self) -> SchemaLayoutResult<Vec<usize>> {
237        fn assign(
238            source_slots: &mut [Option<usize>],
239            target_slot: usize,
240            source_slot: usize,
241        ) -> SchemaLayoutResult<()> {
242            if target_slot == NULL_SLOT {
243                return Ok(());
244            }
245            match source_slots[target_slot] {
246                Some(existing) if existing != source_slot => Err(SchemaLayoutError(format!(
247                    "physical relayout maps target slot {target_slot} to both source slots {existing} and {source_slot}"
248                ))),
249                Some(_) => Ok(()),
250                None => {
251                    source_slots[target_slot] = Some(source_slot);
252                    Ok(())
253                }
254            }
255        }
256
257        if self.len() != target.len() {
258            return Err(SchemaLayoutError(format!(
259                "cannot relayout {} logical columns as {} logical columns",
260                self.len(),
261                target.len()
262            )));
263        }
264
265        let mut source_slots = vec![None; target.physical_width()];
266
267        for logical in 0..target.len() {
268            assign(
269                &mut source_slots,
270                target.index.slots[logical],
271                self.index.slots[logical],
272            )?;
273        }
274
275        for (identity, target_slot) in &target.index.aliases {
276            if *target_slot == NULL_SLOT {
277                continue;
278            }
279            let mut matching_slots = self
280                .index
281                .identities
282                .iter()
283                .enumerate()
284                .filter_map(|(logical, candidate)| {
285                    (candidate == identity).then_some(self.index.slots[logical])
286                })
287                .chain(self.index.aliases.get(identity).copied())
288                .collect::<Vec<_>>();
289            matching_slots.sort_unstable();
290            matching_slots.dedup();
291            let source_slot = match matching_slots.as_slice() {
292                [source_slot] => *source_slot,
293                [] => {
294                    return Err(SchemaLayoutError(format!(
295                        "physical relayout source is missing lookup identity `{identity:?}`"
296                    )))
297                }
298                _ => {
299                    return Err(SchemaLayoutError(format!(
300                        "physical relayout source has ambiguous lookup identity `{identity:?}`"
301                    )))
302                }
303            };
304            assign(&mut source_slots, *target_slot, source_slot)?;
305        }
306
307        for (column, target_slot) in &target.index.executor_attributes {
308            if *target_slot == NULL_SLOT {
309                continue;
310            }
311            // An internal target entry may be another structural identity for a slot already mapped through the public target list. A rebuilt EvalPlanQual subtree receives fresh internal relation IDs, but its visible resno layout remains the same; the existing slot assignment is therefore already authoritative.
312            if source_slots[*target_slot].is_some() {
313                continue;
314            }
315            let source_slot = self
316                .internal_slot(*column)
317                .or_else(|| {
318                    target
319                        .index
320                        .cold
321                        .score_sources
322                        .iter()
323                        .find(|source| source.column == *column)
324                        .map(|source| source.qualifier.as_deref())
325                        .and_then(|qualifier| self.score_source_slot(qualifier))
326                })
327                .ok_or_else(|| {
328                    SchemaLayoutError(format!(
329                        "physical relayout source is missing internal relation attribute `{column:?}`"
330                    ))
331                })?;
332            assign(&mut source_slots, *target_slot, source_slot)?;
333        }
334
335        let source_slots = source_slots
336            .into_iter()
337            .map(|slot| slot.unwrap_or(NULL_SLOT))
338            .collect::<Vec<_>>();
339        Ok(source_slots)
340    }
341}
uqa_sql/schema/schema_composition.rs

uqa_sql/schema/
schema_composition.rs
