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

uqa_execution/batch/
schema_composition.rs
