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                collations: input.index.cold.collations.clone(),
44                record_fields: input.index.cold.record_fields.clone(),
45                aliases: input.index.aliases.clone(),
46                alias_types: input.index.cold.aliases.clone(),
47                internal,
48                internal_types,
49                score_sources,
50                wildcard_hidden: input.index.cold.wildcard_hidden.clone(),
51                binding_only: input.index.cold.binding_only.clone(),
52                row_types: input.index.cold.row_types.clone(),
53                open_qualifiers: input.index.cold.open_qualifiers.clone(),
54                ..SchemaBuildMetadata::default()
55            },
56        )
57    }
58
59    /// Select and optionally rename logical columns while retaining the
60    /// child's physical fragments.
61    pub fn select(input: &Self, columns: &[(String, String)]) -> Self {
62        let mut collations = input.index.cold.collations.clone();
63        collations.clear_columns();
64        for (position, (_, source)) in columns.iter().enumerate() {
65            let oid = input
66                .position(source)
67                .and_then(|position| input.inherited_column_collation(position));
68            collations.column(position, oid, input.exact_type(source));
69        }
70        let output_names = columns
71            .iter()
72            .map(|(output, _)| output.clone())
73            .collect::<Vec<_>>();
74        let slots = columns
75            .iter()
76            .map(|(_, source)| input.exact_slot(source).unwrap_or(NULL_SLOT))
77            .collect();
78        let types = columns
79            .iter()
80            .map(|(_, source)| input.exact_type(source).cloned())
81            .collect();
82        let identities = output_names
83            .iter()
84            .cloned()
85            .map(ColumnIdentity::unqualified)
86            .collect();
87        Self::from_typed_parts_with_aliases_and_exact_precedence(
88            output_names,
89            identities,
90            types,
91            slots,
92            input.physical_width(),
93            SchemaBuildMetadata {
94                collations,
95                record_fields: input.index.cold.record_fields.clone(),
96                aliases: HashMap::new(),
97                alias_types: HashMap::new(),
98                internal: input.index.executor_attributes.clone(),
99                internal_types: input.index.cold.executor_attribute_types.clone(),
100                score_sources: input.index.cold.score_sources.clone(),
101                binding_only: HashMap::new(),
102                ..SchemaBuildMetadata::default()
103            },
104        )
105    }
106
107    /// 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.
108    pub fn project_with_sources(
109        input: &Self,
110        projected: Vec<(String, Option<ColumnType>, ProjectedSlot)>,
111        projected_internal: Vec<(
112            crate::ast::InternalColumnRef,
113            Option<ColumnType>,
114            ProjectedSlot,
115        )>,
116        computed_count: usize,
117        pass_through: bool,
118    ) -> Self {
119        let mut collations = input.index.cold.collations.clone();
120        let resolve_slot = |source: ProjectedSlot| match source {
121            ProjectedSlot::Input(slot) | ProjectedSlot::BoundInput { physical: slot, .. } => {
122                slot.unwrap_or(NULL_SLOT)
123            }
124            ProjectedSlot::Computed { position, .. } => input.physical_width() + position,
125        };
126        let source_collation = |source: ProjectedSlot| match source {
127            ProjectedSlot::Input(Some(slot)) => input.inherited_physical_collation(slot),
128            ProjectedSlot::BoundInput { collation, .. }
129            | ProjectedSlot::Computed { collation, .. } => collation,
130            ProjectedSlot::Input(None) => None,
131        };
132        let physical_width = input.physical_width() + computed_count;
133        let mut internal = input.index.executor_attributes.clone();
134        let mut internal_types = input.index.cold.executor_attribute_types.clone();
135        for (column, ty, source) in projected_internal {
136            collations.internal(column, source_collation(source), ty.as_ref());
137            internal.insert(column, resolve_slot(source));
138            internal_types.insert(column, ty);
139        }
140
141        if pass_through {
142            let mut columns = input.columns().to_vec();
143            let mut identities = input.identities().to_vec();
144            let mut types = input.column_types().to_vec();
145            let mut slots = input.index.slots.to_vec();
146            let mut wildcard_hidden = input.index.cold.wildcard_hidden.clone();
147            for (name, ty, source) in projected {
148                let slot = resolve_slot(source);
149                if let Some(position) = columns.iter().position(|column| column == &name) {
150                    slots[position] = slot;
151                    identities[position] = ColumnIdentity::unqualified(name);
152                    collations.column(position, source_collation(source), ty.as_ref());
153                    types[position] = ty;
154                    wildcard_hidden.remove(&position);
155                } else {
156                    identities.push(ColumnIdentity::unqualified(name.clone()));
157                    columns.push(name);
158                    collations.column(types.len(), source_collation(source), ty.as_ref());
159                    types.push(ty);
160                    slots.push(slot);
161                }
162            }
163            return Self::from_typed_parts_with_aliases_and_exact_precedence(
164                columns,
165                identities,
166                types,
167                slots,
168                physical_width,
169                SchemaBuildMetadata {
170                    collations,
171                    record_fields: input.index.cold.record_fields.clone(),
172                    aliases: input.index.aliases.clone(),
173                    alias_types: input.index.cold.aliases.clone(),
174                    internal,
175                    internal_types,
176                    score_sources: input.index.cold.score_sources.clone(),
177                    wildcard_hidden,
178                    binding_only: input.index.cold.binding_only.clone(),
179                    row_types: input.index.cold.row_types.clone(),
180                    open_qualifiers: input.index.cold.open_qualifiers.clone(),
181                    ..SchemaBuildMetadata::default()
182                },
183            );
184        }
185
186        collations.clear_columns();
187        let mut columns = Vec::with_capacity(projected.len());
188        let mut identities = Vec::with_capacity(projected.len());
189        let mut types = Vec::with_capacity(projected.len());
190        let mut slots = Vec::with_capacity(projected.len());
191        for (name, ty, source) in projected {
192            slots.push(resolve_slot(source));
193            identities.push(ColumnIdentity::unqualified(name.clone()));
194            columns.push(name);
195            collations.column(types.len(), source_collation(source), ty.as_ref());
196            types.push(ty);
197        }
198        Self::from_typed_parts_with_aliases_and_exact_precedence(
199            columns,
200            identities,
201            types,
202            slots,
203            physical_width,
204            SchemaBuildMetadata {
205                collations,
206                record_fields: input.index.cold.record_fields.clone(),
207                aliases: HashMap::new(),
208                alias_types: HashMap::new(),
209                internal,
210                internal_types,
211                score_sources: input.index.cold.score_sources.clone(),
212                binding_only: HashMap::new(),
213                ..SchemaBuildMetadata::default()
214            },
215        )
216    }
217
218    /// Build a compact positional layout for a blocking or spill boundary.
219    /// Logical columns and hidden lookup aliases are remapped to a deduplicated
220    /// list of referenced physical slots; projecting a row through the returned
221    /// slot list shares its existing value fragments without cloning values.
222    pub fn canonical_projection(&self) -> (Self, Vec<usize>) {
223        fn remap_slot(
224            slot: usize,
225            source_slots: &mut Vec<usize>,
226            positions: &mut HashMap<usize, usize>,
227        ) -> usize {
228            if slot == NULL_SLOT {
229                return NULL_SLOT;
230            }
231            if let Some(position) = positions.get(&slot) {
232                return *position;
233            }
234            let position = source_slots.len();
235            source_slots.push(slot);
236            positions.insert(slot, position);
237            position
238        }
239
240        let mut source_slots = Vec::new();
241        let mut positions = HashMap::new();
242        let slots = self
243            .index
244            .slots
245            .iter()
246            .map(|slot| remap_slot(*slot, &mut source_slots, &mut positions))
247            .collect();
248        let mut source_aliases = self.index.aliases.iter().collect::<Vec<_>>();
249        source_aliases.sort_unstable_by_key(|(left, _)| *left);
250        let aliases = source_aliases
251            .into_iter()
252            .map(|(name, slot)| {
253                (
254                    name.clone(),
255                    remap_slot(*slot, &mut source_slots, &mut positions),
256                )
257            })
258            .collect();
259        let mut source_internal = self.index.executor_attributes.iter().collect::<Vec<_>>();
260        source_internal.sort_unstable_by_key(|(column, _)| **column);
261        let internal = source_internal
262            .into_iter()
263            .map(|(column, slot)| {
264                (
265                    *column,
266                    remap_slot(*slot, &mut source_slots, &mut positions),
267                )
268            })
269            .collect();
270        (
271            Self::from_typed_parts_with_aliases_and_exact_precedence(
272                self.columns().to_vec(),
273                self.identities().to_vec(),
274                self.column_types().to_vec(),
275                slots,
276                source_slots.len(),
277                SchemaBuildMetadata {
278                    collations: self.index.cold.collations.clone(),
279                    record_fields: self
280                        .index
281                        .cold
282                        .record_fields
283                        .iter()
284                        .filter_map(|(slot, fields)| {
285                            positions.get(slot).map(|mapped| (*mapped, fields.clone()))
286                        })
287                        .collect(),
288                    aliases,
289                    alias_types: self.index.cold.aliases.clone(),
290                    internal,
291                    internal_types: self.index.cold.executor_attribute_types.clone(),
292                    score_sources: self.index.cold.score_sources.clone(),
293                    wildcard_hidden: self.index.cold.wildcard_hidden.clone(),
294                    binding_only: self.index.cold.binding_only.clone(),
295                    row_types: self.index.cold.row_types.clone(),
296                    open_qualifiers: self.index.cold.open_qualifiers.clone(),
297                    ..SchemaBuildMetadata::default()
298                },
299            ),
300            source_slots,
301        )
302    }
303
304    /// Rebuild a schema decoded from a positional spill record.
305    ///
306    /// `None` represents a logical or alias identity whose source was absent
307    /// and therefore resolves to SQL NULL. Every physical slot is validated
308    /// before the derived lookup indexes are constructed.
309    #[expect(
310        clippy::too_many_lines,
311        reason = "projection keeps schema and physical column positions aligned"
312    )]
313    pub fn from_physical_layout(layout: PhysicalLayout) -> SchemaLayoutResult<Self> {
314        let PhysicalLayout {
315            columns,
316            identities,
317            types,
318            slots,
319            physical_width,
320            aliases,
321            internal,
322            score_sources,
323            wildcard_hidden,
324        } = layout;
325        if columns.len() != slots.len() {
326            return Err(SchemaLayoutError(format!(
327                "physical schema has {} columns but {} logical slots",
328                columns.len(),
329                slots.len()
330            )));
331        }
332        if columns.len() != types.len() {
333            return Err(SchemaLayoutError(format!(
334                "physical schema has {} columns but {} logical types",
335                columns.len(),
336                types.len()
337            )));
338        }
339        if columns.len() != identities.len() {
340            return Err(SchemaLayoutError(format!(
341                "physical schema has {} columns but {} logical identities",
342                columns.len(),
343                identities.len()
344            )));
345        }
346        if wildcard_hidden
347            .iter()
348            .any(|position| *position >= columns.len())
349        {
350            return Err(SchemaLayoutError(
351                "physical schema wildcard-hidden position is outside logical width".into(),
352            ));
353        }
354        let slots = slots
355            .into_iter()
356            .map(|slot| match slot {
357                Some(slot) if slot < physical_width => Ok(slot),
358                Some(slot) => Err(SchemaLayoutError(format!(
359                    "physical schema logical slot {slot} is outside width {physical_width}"
360                ))),
361                None => Ok(NULL_SLOT),
362            })
363            .collect::<SchemaLayoutResult<Vec<_>>>()?;
364        let mut lookup_aliases = HashMap::with_capacity(aliases.len());
365        let mut alias_types = HashMap::with_capacity(aliases.len());
366        for (identity, slot, ty) in aliases {
367            let slot = match slot {
368                Some(slot) if slot < physical_width => slot,
369                Some(slot) => {
370                    return Err(SchemaLayoutError(format!(
371                        "physical schema alias `{identity:?}` slot {slot} is outside width {physical_width}"
372                    )))
373                }
374                None => NULL_SLOT,
375            };
376            if lookup_aliases.insert(identity.clone(), slot).is_some() {
377                return Err(SchemaLayoutError(format!(
378                    "physical schema contains duplicate alias `{identity:?}`"
379                )));
380            }
381            alias_types.insert(identity, ty);
382        }
383        let mut internal_slots = HashMap::with_capacity(internal.len());
384        let mut internal_types = HashMap::with_capacity(internal.len());
385        for (column, slot, ty) in internal {
386            let slot = match slot {
387                Some(slot) if slot < physical_width => slot,
388                Some(slot) => {
389                    return Err(SchemaLayoutError(format!(
390                        "physical schema internal attribute `{column:?}` slot {slot} is outside width {physical_width}"
391                    )))
392                }
393                None => NULL_SLOT,
394            };
395            if internal_slots.insert(column, slot).is_some() {
396                return Err(SchemaLayoutError(format!(
397                    "physical schema contains duplicate internal attribute `{column:?}`"
398                )));
399            }
400            internal_types.insert(column, ty);
401        }
402        let score_sources = score_sources
403            .into_iter()
404            .map(|(qualifier, column)| {
405                if !internal_slots.contains_key(&column) {
406                    return Err(SchemaLayoutError(format!(
407                        "physical schema score source references missing internal attribute `{column:?}`"
408                    )));
409                }
410                Ok(ScoreSource {
411                    qualifier: qualifier.map(Box::<str>::from),
412                    column,
413                })
414            })
415            .collect::<SchemaLayoutResult<Vec<_>>>()?;
416        Ok(Self::from_typed_parts_with_aliases_and_exact_precedence(
417            columns,
418            identities,
419            types,
420            slots,
421            physical_width,
422            SchemaBuildMetadata {
423                aliases: lookup_aliases,
424                alias_types,
425                internal: internal_slots,
426                internal_types,
427                score_sources,
428                wildcard_hidden,
429                binding_only: HashMap::new(),
430                ..SchemaBuildMetadata::default()
431            },
432        ))
433    }
434
435    pub fn lookup_aliases(&self) -> Vec<(&ColumnIdentity, Option<usize>)> {
436        let mut aliases = self
437            .index
438            .aliases
439            .iter()
440            .map(|(identity, slot)| (identity, (*slot != NULL_SLOT).then_some(*slot)))
441            .collect::<Vec<_>>();
442        aliases.sort_unstable_by_key(|(identity, _)| *identity);
443        aliases
444    }
445
446    pub fn lookup_aliases_with_types(
447        &self,
448    ) -> Vec<(&ColumnIdentity, Option<usize>, Option<&ColumnType>)> {
449        self.lookup_aliases()
450            .into_iter()
451            .map(|(identity, slot)| {
452                (
453                    identity,
454                    slot,
455                    self.index
456                        .cold
457                        .aliases
458                        .get(identity)
459                        .and_then(Option::as_ref),
460                )
461            })
462            .collect()
463    }
464
465    pub fn internal_columns_with_types(
466        &self,
467    ) -> Vec<(
468        crate::ast::InternalColumnRef,
469        Option<usize>,
470        Option<&ColumnType>,
471    )> {
472        let mut columns = self
473            .index
474            .executor_attributes
475            .iter()
476            .map(|(column, slot)| {
477                (
478                    *column,
479                    (*slot != NULL_SLOT).then_some(*slot),
480                    self.index
481                        .cold
482                        .executor_attribute_types
483                        .get(column)
484                        .and_then(Option::as_ref),
485                )
486            })
487            .collect::<Vec<_>>();
488        columns.sort_unstable_by_key(|(column, _, _)| *column);
489        columns
490    }
491
492    pub fn score_sources(
493        &self,
494    ) -> impl Iterator<Item = (Option<&str>, crate::ast::InternalColumnRef)> {
495        self.index
496            .cold
497            .score_sources
498            .iter()
499            .map(|source| (source.qualifier.as_deref(), source.column))
500    }
501
502    pub fn wildcard_hidden_positions(&self) -> impl Iterator<Item = usize> + '_ {
503        self.index.cold.wildcard_hidden.iter().copied()
504    }
505
506    /// 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.
507    pub fn remap_positions(
508        input: &Self,
509        columns: &[(String, usize)],
510        aliases: &[(ColumnIdentity, usize)],
511    ) -> Self {
512        let columns = columns
513            .iter()
514            .map(|(name, logical)| (name.clone(), *logical, input.column_type(*logical).cloned()))
515            .collect::<Vec<_>>();
516        Self::remap_typed_positions(input, &columns, aliases)
517    }
518
519    /// 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.
520    pub fn remap_typed_positions(
521        input: &Self,
522        columns: &[(String, usize, Option<ColumnType>)],
523        aliases: &[(ColumnIdentity, usize)],
524    ) -> Self {
525        let mut collations = input.index.cold.collations.clone();
526        collations.clear_columns();
527        for (position, (_, source, ty)) in columns.iter().enumerate() {
528            collations.column(
529                position,
530                input.inherited_column_collation(*source),
531                ty.as_ref(),
532            );
533        }
534        let output_names = columns
535            .iter()
536            .map(|(output, _, _)| output.clone())
537            .collect::<Vec<_>>();
538        let slots = columns
539            .iter()
540            .map(|(_, logical, _)| input.slot(*logical).unwrap_or(NULL_SLOT))
541            .collect();
542        let types = columns.iter().map(|(_, _, ty)| ty.clone()).collect();
543        let identities = output_names
544            .iter()
545            .cloned()
546            .map(ColumnIdentity::unqualified)
547            .collect();
548        let wildcard_hidden = columns
549            .iter()
550            .enumerate()
551            .filter_map(|(output, (_, logical, _))| {
552                input
553                    .index
554                    .cold
555                    .wildcard_hidden
556                    .contains(logical)
557                    .then_some(output)
558            })
559            .collect();
560        let mut lookup_aliases = input.index.aliases.clone();
561        let mut alias_types = input.index.cold.aliases.clone();
562        for (identity, logical) in aliases {
563            collations.alias(
564                identity,
565                input.inherited_column_collation(*logical),
566                input.column_type(*logical),
567            );
568            lookup_aliases.insert(identity.clone(), input.slot(*logical).unwrap_or(NULL_SLOT));
569            alias_types.insert(identity.clone(), input.column_type(*logical).cloned());
570        }
571        Self::from_typed_parts_with_aliases_and_exact_precedence(
572            output_names,
573            identities,
574            types,
575            slots,
576            input.physical_width(),
577            SchemaBuildMetadata {
578                collations,
579                record_fields: input.index.cold.record_fields.clone(),
580                aliases: lookup_aliases,
581                alias_types,
582                internal: input.index.executor_attributes.clone(),
583                internal_types: input.index.cold.executor_attribute_types.clone(),
584                score_sources: input.index.cold.score_sources.clone(),
585                wildcard_hidden,
586                binding_only: input.index.cold.binding_only.clone(),
587                row_types: input.index.cold.row_types.clone(),
588                open_qualifiers: input.index.cold.open_qualifiers.clone(),
589                ..SchemaBuildMetadata::default()
590            },
591        )
592    }
593}
uqa_sql/schema/schema_projection.rs

uqa_sql/schema/
schema_projection.rs
