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

uqa_sql/schema/
schema_composition.rs
