uqa_sql/schema/
schema_projection.rs

1//
2// Unified Query Algebra
3//
4// Copyright (c) 2023-2026 Cognica, Inc.
5//
6
7//! Projection, canonicalization, and position remapping for row schemas.
8
9use super::{
10    ColumnIdentity, ColumnType, HashMap, PhysicalLayout, ProjectedSlot, RowSchema,
11    SchemaBuildMetadata, SchemaLayoutError, SchemaLayoutResult, ScoreSource, NULL_SLOT,
12};
13
14impl RowSchema {
15    /// Remove selected executor-only identities after their consumer has run.
16    /// The physical fragments remain shareable until the next canonical
17    /// boundary, where now-unreferenced slots are naturally discarded.
18    pub fn without_internal_attributes(
19        input: &Self,
20        columns: &[crate::ast::InternalColumnRef],
21    ) -> Self {
22        let mut internal = input.index.executor_attributes.clone();
23        let mut internal_types = input.index.cold.executor_attribute_types.clone();
24        for column in columns {
25            internal.remove(column);
26            internal_types.remove(column);
27        }
28        let score_sources = input
29            .index
30            .cold
31            .score_sources
32            .iter()
33            .filter(|source| internal.contains_key(&source.column))
34            .cloned()
35            .collect();
36        Self::from_typed_parts_with_aliases_and_exact_precedence(
37            input.columns().to_vec(),
38            input.identities().to_vec(),
39            input.column_types().to_vec(),
40            input.index.slots.to_vec(),
41            input.physical_width(),
42            SchemaBuildMetadata {
43                aliases: input.index.aliases.clone(),
44                alias_types: input.index.cold.aliases.clone(),
45                internal,
46                internal_types,
47                score_sources,
48                wildcard_hidden: input.index.cold.wildcard_hidden.clone(),
49                binding_only: input.index.cold.binding_only.clone(),
50                open_qualifiers: input.index.cold.open_qualifiers.clone(),
51                ..SchemaBuildMetadata::default()
52            },
53        )
54    }
55
56    /// Select and optionally rename logical columns while retaining the
57    /// child's physical fragments.
58    pub fn select(input: &Self, columns: &[(String, String)]) -> Self {
59        let output_names = columns
60            .iter()
61            .map(|(output, _)| output.clone())
62            .collect::<Vec<_>>();
63        let slots = columns
64            .iter()
65            .map(|(_, source)| input.exact_slot(source).unwrap_or(NULL_SLOT))
66            .collect();
67        let types = columns
68            .iter()
69            .map(|(_, source)| input.exact_type(source).cloned())
70            .collect();
71        let identities = output_names
72            .iter()
73            .cloned()
74            .map(ColumnIdentity::unqualified)
75            .collect();
76        Self::from_typed_parts_with_aliases_and_exact_precedence(
77            output_names,
78            identities,
79            types,
80            slots,
81            input.physical_width(),
82            SchemaBuildMetadata {
83                aliases: HashMap::new(),
84                alias_types: HashMap::new(),
85                internal: input.index.executor_attributes.clone(),
86                internal_types: input.index.cold.executor_attribute_types.clone(),
87                score_sources: input.index.cold.score_sources.clone(),
88                binding_only: HashMap::new(),
89                ..SchemaBuildMetadata::default()
90            },
91        )
92    }
93
94    /// Build a scalar-projection schema without rebuilding direct input values. A non-pass-through projection hides child identities logically while retaining their physical fragments; an appending projection preserves the child schema and replaces duplicate labels with `PostgreSQL` map-insertion semantics.
95    pub fn project_with_sources(
96        input: &Self,
97        projected: Vec<(String, Option<ColumnType>, ProjectedSlot)>,
98        projected_internal: Vec<(
99            crate::ast::InternalColumnRef,
100            Option<ColumnType>,
101            ProjectedSlot,
102        )>,
103        computed_count: usize,
104        pass_through: bool,
105    ) -> Self {
106        let resolve_slot = |source: ProjectedSlot| match source {
107            ProjectedSlot::Input(slot) => slot.unwrap_or(NULL_SLOT),
108            ProjectedSlot::Computed(position) => input.physical_width() + position,
109        };
110        let physical_width = input.physical_width() + computed_count;
111        let mut internal = input.index.executor_attributes.clone();
112        let mut internal_types = input.index.cold.executor_attribute_types.clone();
113        for (column, ty, source) in projected_internal {
114            internal.insert(column, resolve_slot(source));
115            internal_types.insert(column, ty);
116        }
117
118        if pass_through {
119            let mut columns = input.columns().to_vec();
120            let mut identities = input.identities().to_vec();
121            let mut types = input.column_types().to_vec();
122            let mut slots = input.index.slots.to_vec();
123            let mut wildcard_hidden = input.index.cold.wildcard_hidden.clone();
124            for (name, ty, source) in projected {
125                let slot = resolve_slot(source);
126                if let Some(position) = columns.iter().position(|column| column == &name) {
127                    slots[position] = slot;
128                    identities[position] = ColumnIdentity::unqualified(name);
129                    types[position] = ty;
130                    wildcard_hidden.remove(&position);
131                } else {
132                    identities.push(ColumnIdentity::unqualified(name.clone()));
133                    columns.push(name);
134                    types.push(ty);
135                    slots.push(slot);
136                }
137            }
138            return Self::from_typed_parts_with_aliases_and_exact_precedence(
139                columns,
140                identities,
141                types,
142                slots,
143                physical_width,
144                SchemaBuildMetadata {
145                    aliases: input.index.aliases.clone(),
146                    alias_types: input.index.cold.aliases.clone(),
147                    internal,
148                    internal_types,
149                    score_sources: input.index.cold.score_sources.clone(),
150                    wildcard_hidden,
151                    binding_only: input.index.cold.binding_only.clone(),
152                    open_qualifiers: input.index.cold.open_qualifiers.clone(),
153                    ..SchemaBuildMetadata::default()
154                },
155            );
156        }
157
158        let mut columns = Vec::with_capacity(projected.len());
159        let mut identities = Vec::with_capacity(projected.len());
160        let mut types = Vec::with_capacity(projected.len());
161        let mut slots = Vec::with_capacity(projected.len());
162        for (name, ty, source) in projected {
163            slots.push(resolve_slot(source));
164            identities.push(ColumnIdentity::unqualified(name.clone()));
165            columns.push(name);
166            types.push(ty);
167        }
168        Self::from_typed_parts_with_aliases_and_exact_precedence(
169            columns,
170            identities,
171            types,
172            slots,
173            physical_width,
174            SchemaBuildMetadata {
175                aliases: HashMap::new(),
176                alias_types: HashMap::new(),
177                internal,
178                internal_types,
179                score_sources: input.index.cold.score_sources.clone(),
180                binding_only: HashMap::new(),
181                ..SchemaBuildMetadata::default()
182            },
183        )
184    }
185
186    /// Build a compact positional layout for a blocking or spill boundary.
187    /// Logical columns and hidden lookup aliases are remapped to a deduplicated
188    /// list of referenced physical slots; projecting a row through the returned
189    /// slot list shares its existing value fragments without cloning values.
190    pub fn canonical_projection(&self) -> (Self, Vec<usize>) {
191        fn remap_slot(
192            slot: usize,
193            source_slots: &mut Vec<usize>,
194            positions: &mut HashMap<usize, usize>,
195        ) -> usize {
196            if slot == NULL_SLOT {
197                return NULL_SLOT;
198            }
199            if let Some(position) = positions.get(&slot) {
200                return *position;
201            }
202            let position = source_slots.len();
203            source_slots.push(slot);
204            positions.insert(slot, position);
205            position
206        }
207
208        let mut source_slots = Vec::new();
209        let mut positions = HashMap::new();
210        let slots = self
211            .index
212            .slots
213            .iter()
214            .map(|slot| remap_slot(*slot, &mut source_slots, &mut positions))
215            .collect();
216        let mut source_aliases = self.index.aliases.iter().collect::<Vec<_>>();
217        source_aliases.sort_unstable_by(|(left, _), (right, _)| left.cmp(right));
218        let aliases = source_aliases
219            .into_iter()
220            .map(|(name, slot)| {
221                (
222                    name.clone(),
223                    remap_slot(*slot, &mut source_slots, &mut positions),
224                )
225            })
226            .collect();
227        let mut source_internal = self.index.executor_attributes.iter().collect::<Vec<_>>();
228        source_internal.sort_unstable_by_key(|(column, _)| **column);
229        let internal = source_internal
230            .into_iter()
231            .map(|(column, slot)| {
232                (
233                    *column,
234                    remap_slot(*slot, &mut source_slots, &mut positions),
235                )
236            })
237            .collect();
238        (
239            Self::from_typed_parts_with_aliases_and_exact_precedence(
240                self.columns().to_vec(),
241                self.identities().to_vec(),
242                self.column_types().to_vec(),
243                slots,
244                source_slots.len(),
245                SchemaBuildMetadata {
246                    aliases,
247                    alias_types: self.index.cold.aliases.clone(),
248                    internal,
249                    internal_types: self.index.cold.executor_attribute_types.clone(),
250                    score_sources: self.index.cold.score_sources.clone(),
251                    wildcard_hidden: self.index.cold.wildcard_hidden.clone(),
252                    binding_only: self.index.cold.binding_only.clone(),
253                    open_qualifiers: self.index.cold.open_qualifiers.clone(),
254                    ..SchemaBuildMetadata::default()
255                },
256            ),
257            source_slots,
258        )
259    }
260
261    /// Rebuild a schema decoded from a positional spill record.
262    ///
263    /// `None` represents a logical or alias identity whose source was absent
264    /// and therefore resolves to SQL NULL. Every physical slot is validated
265    /// before the derived lookup indexes are constructed.
266    #[expect(
267        clippy::too_many_lines,
268        reason = "projection keeps schema and physical column positions aligned"
269    )]
270    pub fn from_physical_layout(layout: PhysicalLayout) -> SchemaLayoutResult<Self> {
271        let PhysicalLayout {
272            columns,
273            identities,
274            types,
275            slots,
276            physical_width,
277            aliases,
278            internal,
279            score_sources,
280            wildcard_hidden,
281        } = layout;
282        if columns.len() != slots.len() {
283            return Err(SchemaLayoutError(format!(
284                "physical schema has {} columns but {} logical slots",
285                columns.len(),
286                slots.len()
287            )));
288        }
289        if columns.len() != types.len() {
290            return Err(SchemaLayoutError(format!(
291                "physical schema has {} columns but {} logical types",
292                columns.len(),
293                types.len()
294            )));
295        }
296        if columns.len() != identities.len() {
297            return Err(SchemaLayoutError(format!(
298                "physical schema has {} columns but {} logical identities",
299                columns.len(),
300                identities.len()
301            )));
302        }
303        if wildcard_hidden
304            .iter()
305            .any(|position| *position >= columns.len())
306        {
307            return Err(SchemaLayoutError(
308                "physical schema wildcard-hidden position is outside logical width".into(),
309            ));
310        }
311        let slots = slots
312            .into_iter()
313            .map(|slot| match slot {
314                Some(slot) if slot < physical_width => Ok(slot),
315                Some(slot) => Err(SchemaLayoutError(format!(
316                    "physical schema logical slot {slot} is outside width {physical_width}"
317                ))),
318                None => Ok(NULL_SLOT),
319            })
320            .collect::<SchemaLayoutResult<Vec<_>>>()?;
321        let mut lookup_aliases = HashMap::with_capacity(aliases.len());
322        let mut alias_types = HashMap::with_capacity(aliases.len());
323        for (identity, slot, ty) in aliases {
324            let slot = match slot {
325                Some(slot) if slot < physical_width => slot,
326                Some(slot) => {
327                    return Err(SchemaLayoutError(format!(
328                        "physical schema alias `{identity:?}` slot {slot} is outside width {physical_width}"
329                    )))
330                }
331                None => NULL_SLOT,
332            };
333            if lookup_aliases.insert(identity.clone(), slot).is_some() {
334                return Err(SchemaLayoutError(format!(
335                    "physical schema contains duplicate alias `{identity:?}`"
336                )));
337            }
338            alias_types.insert(identity, ty);
339        }
340        let mut internal_slots = HashMap::with_capacity(internal.len());
341        let mut internal_types = HashMap::with_capacity(internal.len());
342        for (column, slot, ty) in internal {
343            let slot = match slot {
344                Some(slot) if slot < physical_width => slot,
345                Some(slot) => {
346                    return Err(SchemaLayoutError(format!(
347                        "physical schema internal attribute `{column:?}` slot {slot} is outside width {physical_width}"
348                    )))
349                }
350                None => NULL_SLOT,
351            };
352            if internal_slots.insert(column, slot).is_some() {
353                return Err(SchemaLayoutError(format!(
354                    "physical schema contains duplicate internal attribute `{column:?}`"
355                )));
356            }
357            internal_types.insert(column, ty);
358        }
359        let score_sources = score_sources
360            .into_iter()
361            .map(|(qualifier, column)| {
362                if !internal_slots.contains_key(&column) {
363                    return Err(SchemaLayoutError(format!(
364                        "physical schema score source references missing internal attribute `{column:?}`"
365                    )));
366                }
367                Ok(ScoreSource {
368                    qualifier: qualifier.map(Box::<str>::from),
369                    column,
370                })
371            })
372            .collect::<SchemaLayoutResult<Vec<_>>>()?;
373        Ok(Self::from_typed_parts_with_aliases_and_exact_precedence(
374            columns,
375            identities,
376            types,
377            slots,
378            physical_width,
379            SchemaBuildMetadata {
380                aliases: lookup_aliases,
381                alias_types,
382                internal: internal_slots,
383                internal_types,
384                score_sources,
385                wildcard_hidden,
386                binding_only: HashMap::new(),
387                ..SchemaBuildMetadata::default()
388            },
389        ))
390    }
391
392    pub fn lookup_aliases(&self) -> Vec<(&ColumnIdentity, Option<usize>)> {
393        let mut aliases = self
394            .index
395            .aliases
396            .iter()
397            .map(|(identity, slot)| (identity, (*slot != NULL_SLOT).then_some(*slot)))
398            .collect::<Vec<_>>();
399        aliases.sort_unstable_by_key(|(identity, _)| *identity);
400        aliases
401    }
402
403    pub fn lookup_aliases_with_types(
404        &self,
405    ) -> Vec<(&ColumnIdentity, Option<usize>, Option<&ColumnType>)> {
406        self.lookup_aliases()
407            .into_iter()
408            .map(|(identity, slot)| {
409                (
410                    identity,
411                    slot,
412                    self.index
413                        .cold
414                        .aliases
415                        .get(identity)
416                        .and_then(Option::as_ref),
417                )
418            })
419            .collect()
420    }
421
422    pub fn internal_columns_with_types(
423        &self,
424    ) -> Vec<(
425        crate::ast::InternalColumnRef,
426        Option<usize>,
427        Option<&ColumnType>,
428    )> {
429        let mut columns = self
430            .index
431            .executor_attributes
432            .iter()
433            .map(|(column, slot)| {
434                (
435                    *column,
436                    (*slot != NULL_SLOT).then_some(*slot),
437                    self.index
438                        .cold
439                        .executor_attribute_types
440                        .get(column)
441                        .and_then(Option::as_ref),
442                )
443            })
444            .collect::<Vec<_>>();
445        columns.sort_unstable_by_key(|(column, _, _)| *column);
446        columns
447    }
448
449    pub fn score_sources(
450        &self,
451    ) -> impl Iterator<Item = (Option<&str>, crate::ast::InternalColumnRef)> {
452        self.index
453            .cold
454            .score_sources
455            .iter()
456            .map(|source| (source.qualifier.as_deref(), source.column))
457    }
458
459    pub fn wildcard_hidden_positions(&self) -> impl Iterator<Item = usize> + '_ {
460        self.index.cold.wildcard_hidden.iter().copied()
461    }
462
463    /// Select logical input positions and attach hidden lookup identities without copying their physical values. Existing hidden aliases are retained, so nested join qualification survives another remap.
464    pub fn remap_positions(
465        input: &Self,
466        columns: &[(String, usize)],
467        aliases: &[(ColumnIdentity, usize)],
468    ) -> Self {
469        let columns = columns
470            .iter()
471            .map(|(name, logical)| (name.clone(), *logical, input.column_type(*logical).cloned()))
472            .collect::<Vec<_>>();
473        Self::remap_typed_positions(input, &columns, aliases)
474    }
475
476    /// Select logical positions with explicit output types. This is used when a binder inserts an implicit coercion and the output identity no longer has the input slot's declared type.
477    pub fn remap_typed_positions(
478        input: &Self,
479        columns: &[(String, usize, Option<ColumnType>)],
480        aliases: &[(ColumnIdentity, usize)],
481    ) -> Self {
482        let output_names = columns
483            .iter()
484            .map(|(output, _, _)| output.clone())
485            .collect::<Vec<_>>();
486        let slots = columns
487            .iter()
488            .map(|(_, logical, _)| input.slot(*logical).unwrap_or(NULL_SLOT))
489            .collect();
490        let types = columns.iter().map(|(_, _, ty)| ty.clone()).collect();
491        let identities = output_names
492            .iter()
493            .cloned()
494            .map(ColumnIdentity::unqualified)
495            .collect();
496        let wildcard_hidden = columns
497            .iter()
498            .enumerate()
499            .filter_map(|(output, (_, logical, _))| {
500                input
501                    .index
502                    .cold
503                    .wildcard_hidden
504                    .contains(logical)
505                    .then_some(output)
506            })
507            .collect();
508        let mut lookup_aliases = input.index.aliases.clone();
509        let mut alias_types = input.index.cold.aliases.clone();
510        for (identity, logical) in aliases {
511            lookup_aliases.insert(identity.clone(), input.slot(*logical).unwrap_or(NULL_SLOT));
512            alias_types.insert(identity.clone(), input.column_type(*logical).cloned());
513        }
514        Self::from_typed_parts_with_aliases_and_exact_precedence(
515            output_names,
516            identities,
517            types,
518            slots,
519            input.physical_width(),
520            SchemaBuildMetadata {
521                aliases: lookup_aliases,
522                alias_types,
523                internal: input.index.executor_attributes.clone(),
524                internal_types: input.index.cold.executor_attribute_types.clone(),
525                score_sources: input.index.cold.score_sources.clone(),
526                wildcard_hidden,
527                binding_only: input.index.cold.binding_only.clone(),
528                open_qualifiers: input.index.cold.open_qualifiers.clone(),
529                ..SchemaBuildMetadata::default()
530            },
531        )
532    }
533}
uqa_sql/schema/schema_projection.rs

uqa_sql/schema/
schema_projection.rs
