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

uqa_sql/schema/
schema_projection.rs
