uqa_execution/batch/
outer_scope.rs

1//
2// Unified Query Algebra
3//
4// Copyright (c) 2023-2026 Cognica, Inc.
5//
6
7//! Correlated outer-scope schema composition.
8
9use std::collections::{HashMap, HashSet};
10
11use uqa_sql::ast::ColumnType;
12
13use super::{ColumnIdentity, RowSchema, SchemaBuildMetadata, NULL_SLOT};
14
15impl RowSchema {
16    /// Overlay a statically typed correlated outer scope. The outer columns remain hidden from schema iteration and star expansion, but both qualified and unqualified lookup aliases retain their declared SQL types for expression binding.
17    pub fn with_typed_outer_scope(
18        input: &Self,
19        outer_columns: &[(String, Option<ColumnType>)],
20    ) -> Self {
21        let identities = outer_columns
22            .iter()
23            .map(|(column, ty)| (ColumnIdentity::unqualified(column), ty.clone()))
24            .collect::<Vec<_>>();
25        Self::with_typed_outer_identities(input, &identities)
26    }
27
28    /// Overlay a correlated outer scope carried as structured identities.
29    pub fn with_typed_outer_identities(
30        input: &Self,
31        outer_columns: &[(ColumnIdentity, Option<ColumnType>)],
32    ) -> Self {
33        let outer_base = input.physical_width();
34        let current_qualifiers = input
35            .index
36            .qualified
37            .keys()
38            .chain(input.index.aliases.keys())
39            .filter_map(ColumnIdentity::qualifier)
40            .collect::<HashSet<_>>();
41        let mut current_unqualified = input
42            .identities()
43            .iter()
44            .map(ColumnIdentity::column)
45            .collect::<HashSet<_>>();
46        current_unqualified.extend(
47            input
48                .index
49                .aliases
50                .keys()
51                .filter(|identity| identity.qualifier().is_none())
52                .map(ColumnIdentity::column),
53        );
54
55        let mut aliases = input.index.aliases.clone();
56        let mut alias_types = input.index.cold.aliases.clone();
57        let mut ambiguous_qualified = input.index.ambiguous_qualified.clone();
58        let mut outer_exact = HashSet::<&str>::new();
59        let mut outer_qualified = HashMap::<&str, Vec<usize>>::new();
60        let mut outer_qualified_counts = HashMap::<ColumnIdentity, usize>::new();
61        for (identity, _) in outer_columns {
62            if identity
63                .qualifier()
64                .is_some_and(|qualifier| !current_qualifiers.contains(qualifier))
65            {
66                *outer_qualified_counts.entry(identity.clone()).or_default() += 1;
67            }
68        }
69        for (position, (identity, ty)) in outer_columns.iter().enumerate() {
70            let slot = outer_base + position;
71            if let Some(qualifier) = identity.qualifier() {
72                if !current_qualifiers.contains(qualifier) {
73                    if outer_qualified_counts.get(identity) == Some(&1) {
74                        aliases.insert(identity.clone(), slot);
75                        alias_types.insert(identity.clone(), ty.clone());
76                    } else {
77                        aliases.remove(identity);
78                        alias_types.remove(identity);
79                        ambiguous_qualified.insert(identity.clone());
80                    }
81                }
82                outer_qualified
83                    .entry(identity.column())
84                    .or_default()
85                    .push(position);
86            } else {
87                outer_exact.insert(identity.column());
88                if !current_unqualified.contains(identity.column()) {
89                    aliases.insert(identity.clone(), slot);
90                    alias_types.insert(identity.clone(), ty.clone());
91                }
92            }
93        }
94
95        let mut outer_ambiguous = HashSet::new();
96        for (column, positions) in outer_qualified {
97            if current_unqualified.contains(column) || outer_exact.contains(column) {
98                continue;
99            }
100            match positions.as_slice() {
101                [position] => {
102                    let identity = ColumnIdentity::unqualified(column);
103                    aliases.insert(identity.clone(), outer_base + position);
104                    alias_types.insert(identity, outer_columns[*position].1.clone());
105                }
106                [_, _, ..] => {
107                    outer_ambiguous.insert(Box::<str>::from(column));
108                }
109                [] => {}
110            }
111        }
112
113        Self::from_typed_parts_with_aliases_and_exact_precedence(
114            input.columns().to_vec(),
115            input.identities().to_vec(),
116            input.column_types().to_vec(),
117            input.index.slots.to_vec(),
118            input.physical_width() + outer_columns.len(),
119            SchemaBuildMetadata {
120                aliases,
121                alias_types,
122                internal: input.index.executor_attributes.clone(),
123                internal_types: input.index.cold.executor_attribute_types.clone(),
124                score_sources: input.index.cold.score_sources.clone(),
125                wildcard_hidden: input.index.cold.wildcard_hidden.clone(),
126                binding_only: input.index.cold.binding_only.clone(),
127                extra_ambiguous_unqualified: outer_ambiguous,
128                extra_ambiguous_qualified: ambiguous_qualified,
129                ..SchemaBuildMetadata::default()
130            },
131        )
132    }
133
134    /// Overlay an existing positional outer scope while preserving its complete structured lookup layout and sharing its physical value fragments.
135    #[expect(
136        clippy::too_many_lines,
137        reason = "outer-scope merge keeps schema and value precedence aligned"
138    )]
139    pub fn with_outer_schema(input: &Self, outer: &Self) -> Self {
140        let outer_base = input.physical_width();
141        let current_qualifiers = input
142            .index
143            .qualified
144            .keys()
145            .chain(input.index.aliases.keys())
146            .filter_map(ColumnIdentity::qualifier)
147            .collect::<HashSet<_>>();
148        let mut current_unqualified = input
149            .index
150            .unqualified
151            .keys()
152            .map(std::convert::AsRef::as_ref)
153            .collect::<HashSet<_>>();
154        current_unqualified.extend(
155            input
156                .index
157                .aliases
158                .keys()
159                .filter(|identity| identity.qualifier().is_none())
160                .map(ColumnIdentity::column),
161        );
162
163        let shifted_slot = |slot: usize| {
164            if slot == NULL_SLOT {
165                NULL_SLOT
166            } else {
167                outer_base + slot
168            }
169        };
170        let mut aliases = input.index.aliases.clone();
171        let mut alias_types = input.index.cold.aliases.clone();
172        let mut ambiguous_unqualified = input.index.ambiguous_unqualified.clone();
173        let mut ambiguous_qualified = input.index.ambiguous_qualified.clone();
174
175        let qualified_identities = outer
176            .index
177            .qualified
178            .keys()
179            .chain(
180                outer
181                    .index
182                    .aliases
183                    .keys()
184                    .filter(|identity| identity.qualifier().is_some()),
185            )
186            .chain(outer.index.ambiguous_qualified.iter())
187            .cloned()
188            .collect::<HashSet<_>>();
189        for identity in qualified_identities {
190            let qualifier = identity
191                .qualifier()
192                .expect("qualified outer identity has a qualifier");
193            if current_qualifiers.contains(qualifier) {
194                continue;
195            }
196            if outer.qualified_column_is_ambiguous(qualifier, identity.column()) {
197                aliases.remove(&identity);
198                alias_types.remove(&identity);
199                ambiguous_qualified.insert(identity);
200                continue;
201            }
202            let slot = outer
203                .qualified_slot(qualifier, identity.column())
204                .map_or(NULL_SLOT, shifted_slot);
205            let ty = outer.qualified_type(qualifier, identity.column()).cloned();
206            aliases.insert(identity.clone(), slot);
207            alias_types.insert(identity, ty);
208        }
209
210        let outer_unqualified = outer
211            .index
212            .unqualified
213            .keys()
214            .map(std::convert::AsRef::as_ref)
215            .chain(
216                outer
217                    .index
218                    .aliases
219                    .keys()
220                    .filter(|identity| identity.qualifier().is_none())
221                    .map(ColumnIdentity::column),
222            )
223            .chain(
224                outer
225                    .index
226                    .ambiguous_unqualified
227                    .iter()
228                    .map(std::convert::AsRef::as_ref),
229            )
230            .collect::<HashSet<_>>();
231        for column in outer_unqualified {
232            if current_unqualified.contains(column) {
233                continue;
234            }
235            let identity = ColumnIdentity::unqualified(column);
236            if outer.column_is_ambiguous(column) {
237                aliases.remove(&identity);
238                alias_types.remove(&identity);
239                ambiguous_unqualified.insert(Box::<str>::from(column));
240                continue;
241            }
242            let slot = outer.column_slot(column).map_or(NULL_SLOT, shifted_slot);
243            aliases.insert(identity.clone(), slot);
244            alias_types.insert(identity, outer.type_of(column).cloned());
245        }
246
247        let mut internal = input.index.executor_attributes.clone();
248        let mut internal_types = input.index.cold.executor_attribute_types.clone();
249        for (column, slot) in &outer.index.executor_attributes {
250            internal.insert(*column, shifted_slot(*slot));
251        }
252        for (column, ty) in &outer.index.cold.executor_attribute_types {
253            if let Some(existing) = internal_types.get(column) {
254                assert_eq!(
255                    existing, ty,
256                    "structural identity type changed across correlated scopes"
257                );
258            } else {
259                internal_types.insert(*column, ty.clone());
260            }
261        }
262        let mut score_sources = input.index.cold.score_sources.clone();
263        score_sources.extend(outer.index.cold.score_sources.iter().cloned());
264
265        let schema = Self::from_typed_parts_with_aliases_and_exact_precedence(
266            input.columns().to_vec(),
267            input.identities().to_vec(),
268            input.column_types().to_vec(),
269            input.index.slots.to_vec(),
270            input.physical_width() + outer.physical_width(),
271            SchemaBuildMetadata {
272                aliases,
273                alias_types,
274                internal,
275                internal_types,
276                score_sources,
277                wildcard_hidden: input.index.cold.wildcard_hidden.clone(),
278                binding_only: input.index.cold.binding_only.clone(),
279                extra_ambiguous_unqualified: ambiguous_unqualified,
280                extra_ambiguous_qualified: ambiguous_qualified,
281                ..SchemaBuildMetadata::default()
282            },
283        );
284        let virtual_identities = outer
285            .typed_virtual_identities()
286            .filter(|(identity, _)| {
287                identity
288                    .qualifier()
289                    .is_none_or(|qualifier| !current_qualifiers.contains(qualifier))
290            })
291            .map(|(identity, ty)| (identity.clone(), ty.cloned()))
292            .collect::<Vec<_>>();
293        Self::with_typed_virtual_identities(&schema, &virtual_identities)
294    }
295}
uqa_execution/batch/outer_scope.rs

uqa_execution/batch/
outer_scope.rs
