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

uqa_sql/schema/
schema_composition.rs
