Skip to main content

uqa_execution/
batch.rs

1//
2// Unified Query Algebra
3//
4// Copyright (c) 2023-2026 Cognica, Inc.
5//
6
7//! Schema-bound, allocation-light physical rows and batches.
8//!
9//! Column names belong to [`RowSchema`], not to every row. A physical row is
10//! made from shared value fragments. Joins concatenate fragment handles while
11//! schemas remap `(qualifier, column)` identities to physical slots; neither
12//! operation rebuilds a string-keyed map or clones the contained values.
13
14use std::collections::{HashMap, HashSet};
15use std::sync::Arc;
16
17use smallvec::SmallVec;
18use uqa_core::Value;
19use uqa_sql::ast::ColumnType;
20use uqa_sql::expr::RowLookup;
21use uqa_sql::ResultRow;
22
23use crate::physical::{ExecError, ExecResult};
24
25mod materialization;
26mod outer_scope;
27mod owned_row;
28
29pub use owned_row::OwnedPhysicalRow;
30
31/// Default rows-per-batch hint.
32pub const DEFAULT_BATCH_SIZE: usize = 1024;
33
34const NULL_SLOT: usize = usize::MAX;
35/// Keep the optional row-lock lineage pointer inside the pre-lineage 64-bit row footprint while retaining seven allocation-free join/projection fragments.
36const INLINE_ROW_FRAGMENTS: usize = 7;
37static NULL_VALUE: Value = Value::Null;
38
39/// Structured SQL column identity. A qualifier is metadata, never a prefix encoded into the column name, so quoted names containing `.` remain intact.
40#[derive(Debug, Clone, PartialEq, Eq, Hash, PartialOrd, Ord)]
41pub struct ColumnIdentity {
42    qualifier: Option<Box<str>>,
43    column: Box<str>,
44}
45
46impl ColumnIdentity {
47    #[must_use]
48    pub fn unqualified(column: impl Into<String>) -> Self {
49        Self {
50            qualifier: None,
51            column: Box::<str>::from(column.into()),
52        }
53    }
54
55    #[must_use]
56    pub fn qualified(qualifier: impl Into<String>, column: impl Into<String>) -> Self {
57        Self {
58            qualifier: Some(Box::<str>::from(qualifier.into())),
59            column: Box::<str>::from(column.into()),
60        }
61    }
62
63    #[must_use]
64    pub fn qualifier(&self) -> Option<&str> {
65        self.qualifier.as_deref()
66    }
67
68    #[must_use]
69    pub fn column(&self) -> &str {
70        &self.column
71    }
72}
73
74#[derive(Debug, PartialEq, Eq)]
75struct SchemaIndex {
76    /// Public/materialized output labels in logical order.
77    columns: Box<[String]>,
78    /// SQL lookup identities aligned with `columns`.
79    identities: Box<[ColumnIdentity]>,
80    /// Logical column position -> flattened physical value position.
81    slots: Box<[usize]>,
82    physical_width: usize,
83    /// Structural lookup by physical/public label. SQL name binding uses `unqualified` or `qualified`, never this map.
84    exact: HashMap<Box<str>, usize>,
85    unqualified: HashMap<Box<str>, usize>,
86    qualified: HashMap<ColumnIdentity, usize>,
87    /// Additional lookup identities that point directly at an existing physical slot without becoming output columns. Correlated table aliases use this to expose `(alias, column)` without duplicating the value.
88    aliases: HashMap<ColumnIdentity, usize>,
89    /// Visible unqualified names with more than one logical owner.
90    ambiguous_unqualified: HashSet<Box<str>>,
91    /// Visible qualified identities with more than one logical owner.
92    ambiguous_qualified: HashSet<ColumnIdentity>,
93    /// Static type metadata stays behind a cold pointer so declared SQL identities do not enlarge or displace the cache-hot row lookup fields above.
94    cold: Box<SchemaColdMetadata>,
95}
96
97#[derive(Debug, PartialEq, Eq)]
98struct SchemaColdMetadata {
99    /// `None` is an as-yet unresolved type, not a runtime NULL value.
100    columns: Box<[Option<ColumnType>]>,
101    aliases: HashMap<ColumnIdentity, Option<ColumnType>>,
102    identity_layout: bool,
103}
104
105#[derive(Default)]
106struct SchemaBuildMetadata {
107    aliases: HashMap<ColumnIdentity, usize>,
108    alias_types: HashMap<ColumnIdentity, Option<ColumnType>>,
109    exact_unqualified_precedence: bool,
110    extra_ambiguous_unqualified: HashSet<Box<str>>,
111    extra_ambiguous_qualified: HashSet<ColumnIdentity>,
112}
113
114/// Immutable column layout shared by an operator and all of its batches.
115///
116/// `columns` are the logical output labels. `slots` may point into a wider
117/// composite physical row after a projection/rename, allowing those operators
118/// to change row shape without moving any values.
119#[derive(Debug, Clone, PartialEq, Eq)]
120pub struct RowSchema {
121    index: Arc<SchemaIndex>,
122}
123
124/// Physical source of one scalar-projection output. Direct input slots stay in the child row; only computed values extend its physical layout.
125#[derive(Debug, Clone, Copy, PartialEq, Eq)]
126pub(crate) enum ProjectedSlot {
127    Input(Option<usize>),
128    Computed,
129}
130
131impl Default for RowSchema {
132    fn default() -> Self {
133        Self::new(Vec::new())
134    }
135}
136
137impl From<Vec<String>> for RowSchema {
138    fn from(columns: Vec<String>) -> Self {
139        Self::new(columns)
140    }
141}
142
143impl RowSchema {
144    pub fn new(columns: Vec<String>) -> Self {
145        let width = columns.len();
146        let identities = columns
147            .iter()
148            .cloned()
149            .map(ColumnIdentity::unqualified)
150            .collect();
151        Self::from_parts(columns, identities, (0..width).collect(), width)
152    }
153
154    /// Build a positional schema with statically bound SQL types.
155    pub fn with_types(columns: Vec<String>, types: Vec<Option<ColumnType>>) -> Self {
156        let width = columns.len();
157        assert_eq!(width, types.len(), "row schema column/type width mismatch");
158        let identities = columns
159            .iter()
160            .cloned()
161            .map(ColumnIdentity::unqualified)
162            .collect();
163        Self::from_typed_parts_with_aliases_and_exact_precedence(
164            columns,
165            identities,
166            types,
167            (0..width).collect(),
168            width,
169            SchemaBuildMetadata::default(),
170        )
171    }
172
173    /// Build a positional schema whose visible columns all belong to one relation qualifier while retaining their public names verbatim.
174    pub fn with_qualified_types(
175        qualifier: &str,
176        columns: Vec<String>,
177        types: Vec<Option<ColumnType>>,
178    ) -> Self {
179        let identities = columns
180            .iter()
181            .cloned()
182            .map(|column| ColumnIdentity::qualified(qualifier, column))
183            .collect();
184        Self::with_identities(columns, identities, types)
185    }
186
187    /// Build a positional schema from explicit structured identities.
188    pub fn with_identities(
189        columns: Vec<String>,
190        identities: Vec<ColumnIdentity>,
191        types: Vec<Option<ColumnType>>,
192    ) -> Self {
193        let width = columns.len();
194        assert_eq!(
195            width,
196            identities.len(),
197            "row schema column/identity width mismatch"
198        );
199        assert_eq!(width, types.len(), "row schema column/type width mismatch");
200        Self::from_typed_parts_with_aliases_and_exact_precedence(
201            columns,
202            identities,
203            types,
204            (0..width).collect(),
205            width,
206            SchemaBuildMetadata::default(),
207        )
208    }
209
210    /// Build the lookup semantics of a named compatibility row. An exact bare
211    /// key in a map is authoritative even when qualified metadata keys share
212    /// its suffix; physical relational schemas continue to treat multiple
213    /// visible owners as ambiguous.
214    pub fn from_named_columns(columns: Vec<String>) -> Self {
215        let width = columns.len();
216        let identities = columns
217            .iter()
218            .cloned()
219            .map(ColumnIdentity::unqualified)
220            .collect();
221        Self::from_parts_with_aliases_and_exact_precedence(
222            columns,
223            identities,
224            (0..width).collect(),
225            width,
226            SchemaBuildMetadata {
227                exact_unqualified_precedence: true,
228                ..SchemaBuildMetadata::default()
229            },
230        )
231    }
232
233    fn from_parts(
234        columns: Vec<String>,
235        identities: Vec<ColumnIdentity>,
236        slots: Vec<usize>,
237        physical_width: usize,
238    ) -> Self {
239        Self::from_parts_with_aliases(columns, identities, slots, physical_width, HashMap::new())
240    }
241
242    fn from_parts_with_aliases(
243        columns: Vec<String>,
244        identities: Vec<ColumnIdentity>,
245        slots: Vec<usize>,
246        physical_width: usize,
247        aliases: HashMap<ColumnIdentity, usize>,
248    ) -> Self {
249        Self::from_parts_with_aliases_and_exact_precedence(
250            columns,
251            identities,
252            slots,
253            physical_width,
254            SchemaBuildMetadata {
255                aliases,
256                ..SchemaBuildMetadata::default()
257            },
258        )
259    }
260
261    fn from_typed_parts_with_aliases(
262        columns: Vec<String>,
263        identities: Vec<ColumnIdentity>,
264        types: Vec<Option<ColumnType>>,
265        slots: Vec<usize>,
266        physical_width: usize,
267        aliases: HashMap<ColumnIdentity, usize>,
268        alias_types: HashMap<ColumnIdentity, Option<ColumnType>>,
269    ) -> Self {
270        Self::from_typed_parts_with_aliases_and_exact_precedence(
271            columns,
272            identities,
273            types,
274            slots,
275            physical_width,
276            SchemaBuildMetadata {
277                aliases,
278                alias_types,
279                ..SchemaBuildMetadata::default()
280            },
281        )
282    }
283
284    fn from_parts_with_aliases_and_exact_precedence(
285        columns: Vec<String>,
286        identities: Vec<ColumnIdentity>,
287        slots: Vec<usize>,
288        physical_width: usize,
289        metadata: SchemaBuildMetadata,
290    ) -> Self {
291        let types = vec![None; columns.len()];
292        Self::from_typed_parts_with_aliases_and_exact_precedence(
293            columns,
294            identities,
295            types,
296            slots,
297            physical_width,
298            metadata,
299        )
300    }
301
302    fn from_typed_parts_with_aliases_and_exact_precedence(
303        columns: Vec<String>,
304        identities: Vec<ColumnIdentity>,
305        types: Vec<Option<ColumnType>>,
306        slots: Vec<usize>,
307        physical_width: usize,
308        metadata: SchemaBuildMetadata,
309    ) -> Self {
310        let SchemaBuildMetadata {
311            aliases,
312            alias_types,
313            exact_unqualified_precedence,
314            extra_ambiguous_unqualified,
315            extra_ambiguous_qualified,
316        } = metadata;
317        debug_assert_eq!(columns.len(), slots.len());
318        debug_assert_eq!(columns.len(), identities.len());
319        debug_assert_eq!(columns.len(), types.len());
320        let identity_layout = physical_width == columns.len()
321            && slots
322                .iter()
323                .enumerate()
324                .all(|(position, slot)| position == *slot);
325        let mut exact = HashMap::with_capacity(columns.len());
326        let mut unqualified = HashMap::with_capacity(columns.len());
327        let mut qualified = HashMap::with_capacity(columns.len());
328        let mut unqualified_counts: HashMap<Box<str>, usize> = HashMap::new();
329        let mut qualified_counts: HashMap<ColumnIdentity, usize> = HashMap::new();
330
331        for (logical, (name, identity)) in columns.iter().zip(&identities).enumerate() {
332            // Later writes to the same named field replace the value in a
333            // ResultRow. Schema transforms preserve that contract.
334            exact.insert(Box::<str>::from(name.as_str()), logical);
335            *unqualified_counts
336                .entry(identity.column.clone())
337                .or_default() += 1;
338            unqualified.insert(identity.column.clone(), logical);
339            if identity.qualifier.is_some() {
340                *qualified_counts.entry(identity.clone()).or_default() += 1;
341                qualified.insert(identity.clone(), logical);
342            }
343        }
344        for slot in aliases.values() {
345            debug_assert!(*slot == NULL_SLOT || *slot < physical_width);
346        }
347        let mut ambiguous_unqualified: HashSet<Box<str>> = unqualified_counts
348            .into_iter()
349            .filter_map(|(column, count)| {
350                (count > 1
351                    && !(exact_unqualified_precedence
352                        && identities.iter().any(|identity| {
353                            identity.qualifier.is_none() && identity.column == column
354                        })))
355                .then_some(column)
356            })
357            .collect();
358        ambiguous_unqualified.extend(extra_ambiguous_unqualified);
359        let mut ambiguous_qualified = qualified_counts
360            .into_iter()
361            .filter_map(|(identity, count)| (count > 1).then_some(identity))
362            .collect::<HashSet<_>>();
363        ambiguous_qualified.extend(extra_ambiguous_qualified);
364        Self {
365            index: Arc::new(SchemaIndex {
366                columns: columns.into_boxed_slice(),
367                identities: identities.into_boxed_slice(),
368                slots: slots.into_boxed_slice(),
369                physical_width,
370                exact,
371                unqualified,
372                qualified,
373                aliases,
374                ambiguous_unqualified,
375                ambiguous_qualified,
376                cold: Box::new(SchemaColdMetadata {
377                    columns: types.into_boxed_slice(),
378                    aliases: alias_types,
379                    identity_layout,
380                }),
381            }),
382        }
383    }
384
385    pub fn columns(&self) -> &[String] {
386        &self.index.columns
387    }
388
389    pub fn len(&self) -> usize {
390        self.index.columns.len()
391    }
392
393    pub fn is_empty(&self) -> bool {
394        self.index.columns.is_empty()
395    }
396
397    pub fn iter(&self) -> std::slice::Iter<'_, String> {
398        self.index.columns.iter()
399    }
400
401    /// Static SQL type at one logical output position.
402    pub fn column_type(&self, logical: usize) -> Option<&ColumnType> {
403        self.index
404            .cold
405            .columns
406            .get(logical)
407            .and_then(Option::as_ref)
408    }
409
410    /// Static SQL types aligned with [`Self::columns`].
411    pub fn column_types(&self) -> &[Option<ColumnType>] {
412        &self.index.cold.columns
413    }
414
415    /// Structured SQL identities aligned with [`Self::columns`].
416    pub fn identities(&self) -> &[ColumnIdentity] {
417        &self.index.identities
418    }
419
420    /// Whether a visible or hidden lookup identity belongs to `qualifier`.
421    #[must_use]
422    pub fn has_qualifier(&self, qualifier: &str) -> bool {
423        self.index
424            .identities
425            .iter()
426            .chain(self.index.aliases.keys())
427            .any(|identity| identity.qualifier() == Some(qualifier))
428    }
429
430    pub fn identity(&self, logical: usize) -> Option<&ColumnIdentity> {
431        self.index.identities.get(logical)
432    }
433
434    pub fn public_name(&self, logical: usize) -> Option<&str> {
435        self.identity(logical).map(ColumnIdentity::column)
436    }
437
438    pub fn position(&self, name: &str) -> Option<usize> {
439        self.index.exact.get(name).copied()
440    }
441
442    /// Resolve one visible unqualified SQL identity to its logical position. Ambiguous names deliberately do not select an arbitrary owner.
443    pub fn unqualified_position(&self, column: &str) -> Option<usize> {
444        if self.index.ambiguous_unqualified.contains(column) {
445            return None;
446        }
447        self.index.unqualified.get(column).copied()
448    }
449
450    /// Resolve one visible qualified identity to its logical position.
451    pub fn qualified_position(&self, qualifier: &str, column: &str) -> Option<usize> {
452        let identity = ColumnIdentity::qualified(qualifier, column);
453        if self.index.ambiguous_qualified.contains(&identity) {
454            return None;
455        }
456        self.index.qualified.get(&identity).copied()
457    }
458
459    pub fn physical_width(&self) -> usize {
460        self.index.physical_width
461    }
462
463    /// Resolve one logical output position to its flattened physical slot.
464    pub fn physical_slot(&self, logical: usize) -> Option<usize> {
465        self.slot(logical)
466    }
467
468    /// Resolve a structured SQL identity, including a hidden JOIN USING alias, to its flattened physical slot. Ambiguous identities deliberately return `None` rather than selecting an arbitrary source value.
469    pub fn physical_slot_for_identity(&self, identity: &ColumnIdentity) -> Option<usize> {
470        match identity.qualifier() {
471            Some(qualifier) => self.qualified_slot(qualifier, identity.column()),
472            None => self.column_slot(identity.column()),
473        }
474    }
475
476    pub(crate) fn slot(&self, logical: usize) -> Option<usize> {
477        self.index
478            .slots
479            .get(logical)
480            .copied()
481            .filter(|slot| *slot != NULL_SLOT)
482    }
483
484    fn exact_slot(&self, name: &str) -> Option<usize> {
485        self.index
486            .exact
487            .get(name)
488            .and_then(|logical| self.slot(*logical))
489            .filter(|slot| *slot != NULL_SLOT)
490    }
491
492    fn exact_type(&self, name: &str) -> Option<&ColumnType> {
493        self.index
494            .exact
495            .get(name)
496            .and_then(|logical| self.column_type(*logical))
497    }
498
499    fn column_slot(&self, name: &str) -> Option<usize> {
500        if self.index.ambiguous_unqualified.contains(name) {
501            return None;
502        }
503        self.index
504            .unqualified
505            .get(name)
506            .and_then(|logical| self.slot(*logical))
507            .or_else(|| {
508                self.index
509                    .aliases
510                    .get(&ColumnIdentity::unqualified(name))
511                    .copied()
512            })
513            .filter(|slot| *slot != NULL_SLOT)
514    }
515
516    /// Resolve an unqualified logical identity to its static type.
517    pub fn type_of(&self, name: &str) -> Option<&ColumnType> {
518        if self.index.ambiguous_unqualified.contains(name) {
519            return None;
520        }
521        self.index
522            .unqualified
523            .get(name)
524            .and_then(|logical| self.column_type(*logical))
525            .or_else(|| {
526                self.index
527                    .cold
528                    .aliases
529                    .get(&ColumnIdentity::unqualified(name))
530                    .and_then(Option::as_ref)
531            })
532    }
533
534    /// Resolve a qualified logical identity to its static type.
535    pub fn qualified_type(&self, qualifier: &str, column: &str) -> Option<&ColumnType> {
536        let identity = ColumnIdentity::qualified(qualifier, column);
537        if self.index.ambiguous_qualified.contains(&identity) {
538            return None;
539        }
540        self.index
541            .qualified
542            .get(&identity)
543            .and_then(|logical| self.column_type(*logical))
544            .or_else(|| {
545                self.index
546                    .cold
547                    .aliases
548                    .get(&identity)
549                    .and_then(Option::as_ref)
550            })
551    }
552
553    /// Physical projection layout for `qualifier.*` in relation-column order. Hidden identities introduced by `JOIN ... USING` remain selectable, so each side's wildcard retains its own merged-column value.
554    pub fn qualified_star_layout(
555        &self,
556        qualifier: &str,
557    ) -> Vec<(String, usize, Option<ColumnType>)> {
558        self.qualified_star_position_layout(qualifier)
559            .into_iter()
560            .map(|(column, _, slot, ty)| (column, slot, ty))
561            .collect()
562    }
563
564    /// Bound layout for `qualifier.*`. Visible columns retain their logical positions; hidden aliases such as the suppressed side of `JOIN ... USING` expose only their physical slot.
565    pub fn qualified_star_position_layout(
566        &self,
567        qualifier: &str,
568    ) -> Vec<(String, Option<usize>, usize, Option<ColumnType>)> {
569        let mut entries = Vec::new();
570        let mut visible_layout = HashSet::new();
571        for (logical, identity) in self.identities().iter().enumerate() {
572            if identity.qualifier() == Some(qualifier) {
573                let slot = self.slot(logical).unwrap_or(NULL_SLOT);
574                visible_layout.insert((identity.clone(), slot));
575                entries.push((
576                    identity.clone(),
577                    Some(logical),
578                    slot,
579                    self.column_type(logical).cloned(),
580                ));
581            }
582        }
583        for (identity, slot) in &self.index.aliases {
584            if identity.qualifier() == Some(qualifier)
585                && !visible_layout.contains(&(identity.clone(), *slot))
586            {
587                entries.push((
588                    identity.clone(),
589                    None,
590                    *slot,
591                    self.index.cold.aliases.get(identity).cloned().flatten(),
592                ));
593            }
594        }
595        entries.sort_by_key(|(_, _, slot, _)| *slot);
596        entries
597            .into_iter()
598            .map(|(identity, logical, slot, ty)| (identity.column().to_string(), logical, slot, ty))
599            .collect()
600    }
601
602    pub fn column_is_ambiguous(&self, name: &str) -> bool {
603        self.index.ambiguous_unqualified.contains(name)
604    }
605
606    pub fn qualified_column_is_ambiguous(&self, qualifier: &str, column: &str) -> bool {
607        self.index
608            .ambiguous_qualified
609            .contains(&ColumnIdentity::qualified(qualifier, column))
610    }
611
612    fn qualified_slot(&self, qualifier: &str, column: &str) -> Option<usize> {
613        let identity = ColumnIdentity::qualified(qualifier, column);
614        if self.index.ambiguous_qualified.contains(&identity) {
615            return None;
616        }
617        self.index
618            .qualified
619            .get(&identity)
620            .and_then(|logical| self.slot(*logical))
621            .or_else(|| self.index.aliases.get(&identity).copied())
622            .filter(|slot| *slot != NULL_SLOT)
623    }
624
625    /// Select and optionally rename logical columns while retaining the
626    /// child's physical fragments.
627    pub fn select(input: &Self, columns: &[(String, String)]) -> Self {
628        let output_names = columns
629            .iter()
630            .map(|(output, _)| output.clone())
631            .collect::<Vec<_>>();
632        let slots = columns
633            .iter()
634            .map(|(_, source)| input.exact_slot(source).unwrap_or(NULL_SLOT))
635            .collect();
636        let types = columns
637            .iter()
638            .map(|(_, source)| input.exact_type(source).cloned())
639            .collect();
640        let identities = output_names
641            .iter()
642            .cloned()
643            .map(ColumnIdentity::unqualified)
644            .collect();
645        Self::from_typed_parts_with_aliases(
646            output_names,
647            identities,
648            types,
649            slots,
650            input.physical_width(),
651            HashMap::new(),
652            HashMap::new(),
653        )
654    }
655
656    /// 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.
657    pub(crate) fn project_with_sources(
658        input: &Self,
659        projected: Vec<(String, Option<ColumnType>, ProjectedSlot)>,
660        pass_through: bool,
661    ) -> Self {
662        let mut next_computed_slot = input.physical_width();
663        let mut resolve_slot = |source: ProjectedSlot| match source {
664            ProjectedSlot::Input(slot) => slot.unwrap_or(NULL_SLOT),
665            ProjectedSlot::Computed => {
666                let slot = next_computed_slot;
667                next_computed_slot += 1;
668                slot
669            }
670        };
671
672        if pass_through {
673            let mut columns = input.columns().to_vec();
674            let mut identities = input.identities().to_vec();
675            let mut types = input.column_types().to_vec();
676            let mut slots = input.index.slots.to_vec();
677            for (name, ty, source) in projected {
678                let slot = resolve_slot(source);
679                if let Some(position) = columns.iter().position(|column| column == &name) {
680                    slots[position] = slot;
681                    identities[position] = ColumnIdentity::unqualified(name);
682                    types[position] = ty;
683                } else {
684                    identities.push(ColumnIdentity::unqualified(name.clone()));
685                    columns.push(name);
686                    types.push(ty);
687                    slots.push(slot);
688                }
689            }
690            return Self::from_typed_parts_with_aliases(
691                columns,
692                identities,
693                types,
694                slots,
695                next_computed_slot,
696                input.index.aliases.clone(),
697                input.index.cold.aliases.clone(),
698            );
699        }
700
701        let mut columns = Vec::with_capacity(projected.len());
702        let mut identities = Vec::with_capacity(projected.len());
703        let mut types = Vec::with_capacity(projected.len());
704        let mut slots = Vec::with_capacity(projected.len());
705        for (name, ty, source) in projected {
706            slots.push(resolve_slot(source));
707            identities.push(ColumnIdentity::unqualified(name.clone()));
708            columns.push(name);
709            types.push(ty);
710        }
711        Self::from_typed_parts_with_aliases(
712            columns,
713            identities,
714            types,
715            slots,
716            next_computed_slot,
717            HashMap::new(),
718            HashMap::new(),
719        )
720    }
721
722    /// Build a compact positional layout for a blocking or spill boundary.
723    /// Logical columns and hidden lookup aliases are remapped to a deduplicated
724    /// list of referenced physical slots; projecting a row through the returned
725    /// slot list shares its existing value fragments without cloning values.
726    pub(crate) fn canonical_projection(&self) -> (Self, Vec<usize>) {
727        fn remap_slot(
728            slot: usize,
729            source_slots: &mut Vec<usize>,
730            positions: &mut HashMap<usize, usize>,
731        ) -> usize {
732            if slot == NULL_SLOT {
733                return NULL_SLOT;
734            }
735            if let Some(position) = positions.get(&slot) {
736                return *position;
737            }
738            let position = source_slots.len();
739            source_slots.push(slot);
740            positions.insert(slot, position);
741            position
742        }
743
744        let mut source_slots = Vec::new();
745        let mut positions = HashMap::new();
746        let slots = self
747            .index
748            .slots
749            .iter()
750            .map(|slot| remap_slot(*slot, &mut source_slots, &mut positions))
751            .collect();
752        let mut source_aliases = self.index.aliases.iter().collect::<Vec<_>>();
753        source_aliases.sort_unstable_by(|(left, _), (right, _)| left.cmp(right));
754        let aliases = source_aliases
755            .into_iter()
756            .map(|(name, slot)| {
757                (
758                    name.clone(),
759                    remap_slot(*slot, &mut source_slots, &mut positions),
760                )
761            })
762            .collect();
763        (
764            Self::from_typed_parts_with_aliases(
765                self.columns().to_vec(),
766                self.identities().to_vec(),
767                self.column_types().to_vec(),
768                slots,
769                source_slots.len(),
770                aliases,
771                self.index.cold.aliases.clone(),
772            ),
773            source_slots,
774        )
775    }
776
777    /// Rebuild a schema decoded from a positional spill record.
778    ///
779    /// `None` represents a logical or alias identity whose source was absent
780    /// and therefore resolves to SQL NULL. Every physical slot is validated
781    /// before the derived lookup indexes are constructed.
782    pub(crate) fn from_physical_layout(
783        columns: Vec<String>,
784        identities: Vec<ColumnIdentity>,
785        types: Vec<Option<ColumnType>>,
786        slots: Vec<Option<usize>>,
787        physical_width: usize,
788        aliases: Vec<(ColumnIdentity, Option<usize>, Option<ColumnType>)>,
789    ) -> ExecResult<Self> {
790        if columns.len() != slots.len() {
791            return Err(ExecError::Other(format!(
792                "physical schema has {} columns but {} logical slots",
793                columns.len(),
794                slots.len()
795            )));
796        }
797        if columns.len() != types.len() {
798            return Err(ExecError::Other(format!(
799                "physical schema has {} columns but {} logical types",
800                columns.len(),
801                types.len()
802            )));
803        }
804        if columns.len() != identities.len() {
805            return Err(ExecError::Other(format!(
806                "physical schema has {} columns but {} logical identities",
807                columns.len(),
808                identities.len()
809            )));
810        }
811        let slots = slots
812            .into_iter()
813            .map(|slot| match slot {
814                Some(slot) if slot < physical_width => Ok(slot),
815                Some(slot) => Err(ExecError::Other(format!(
816                    "physical schema logical slot {slot} is outside width {physical_width}"
817                ))),
818                None => Ok(NULL_SLOT),
819            })
820            .collect::<ExecResult<Vec<_>>>()?;
821        let mut lookup_aliases = HashMap::with_capacity(aliases.len());
822        let mut alias_types = HashMap::with_capacity(aliases.len());
823        for (identity, slot, ty) in aliases {
824            let slot = match slot {
825                Some(slot) if slot < physical_width => slot,
826                Some(slot) => {
827                    return Err(ExecError::Other(format!(
828                        "physical schema alias `{identity:?}` slot {slot} is outside width {physical_width}"
829                    )))
830                }
831                None => NULL_SLOT,
832            };
833            if lookup_aliases.insert(identity.clone(), slot).is_some() {
834                return Err(ExecError::Other(format!(
835                    "physical schema contains duplicate alias `{identity:?}`"
836                )));
837            }
838            alias_types.insert(identity, ty);
839        }
840        Ok(Self::from_typed_parts_with_aliases(
841            columns,
842            identities,
843            types,
844            slots,
845            physical_width,
846            lookup_aliases,
847            alias_types,
848        ))
849    }
850
851    pub(crate) fn lookup_aliases(&self) -> Vec<(&ColumnIdentity, Option<usize>)> {
852        let mut aliases = self
853            .index
854            .aliases
855            .iter()
856            .map(|(identity, slot)| (identity, (*slot != NULL_SLOT).then_some(*slot)))
857            .collect::<Vec<_>>();
858        aliases.sort_unstable_by_key(|(identity, _)| *identity);
859        aliases
860    }
861
862    pub(crate) fn lookup_aliases_with_types(
863        &self,
864    ) -> Vec<(&ColumnIdentity, Option<usize>, Option<&ColumnType>)> {
865        self.lookup_aliases()
866            .into_iter()
867            .map(|(identity, slot)| {
868                (
869                    identity,
870                    slot,
871                    self.index
872                        .cold
873                        .aliases
874                        .get(identity)
875                        .and_then(Option::as_ref),
876                )
877            })
878            .collect()
879    }
880
881    /// Add hidden structured lookup identities for existing logical positions.
882    pub fn with_identity_aliases(input: &Self, aliases: &[(ColumnIdentity, usize)]) -> Self {
883        let mut lookup_aliases = input.index.aliases.clone();
884        let mut alias_types = input.index.cold.aliases.clone();
885        for (identity, logical) in aliases {
886            lookup_aliases.insert(identity.clone(), input.slot(*logical).unwrap_or(NULL_SLOT));
887            alias_types.insert(identity.clone(), input.column_type(*logical).cloned());
888        }
889        Self::from_typed_parts_with_aliases(
890            input.columns().to_vec(),
891            input.identities().to_vec(),
892            input.column_types().to_vec(),
893            input.index.slots.to_vec(),
894            input.physical_width(),
895            lookup_aliases,
896            alias_types,
897        )
898    }
899
900    /// 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.
901    pub(crate) fn remap_positions(
902        input: &Self,
903        columns: &[(String, usize)],
904        aliases: &[(ColumnIdentity, usize)],
905    ) -> Self {
906        let columns = columns
907            .iter()
908            .map(|(name, logical)| (name.clone(), *logical, input.column_type(*logical).cloned()))
909            .collect::<Vec<_>>();
910        Self::remap_typed_positions(input, &columns, aliases)
911    }
912
913    /// 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.
914    pub(crate) fn remap_typed_positions(
915        input: &Self,
916        columns: &[(String, usize, Option<ColumnType>)],
917        aliases: &[(ColumnIdentity, usize)],
918    ) -> Self {
919        let output_names = columns
920            .iter()
921            .map(|(output, _, _)| output.clone())
922            .collect::<Vec<_>>();
923        let slots = columns
924            .iter()
925            .map(|(_, logical, _)| input.slot(*logical).unwrap_or(NULL_SLOT))
926            .collect();
927        let types = columns.iter().map(|(_, _, ty)| ty.clone()).collect();
928        let identities = output_names
929            .iter()
930            .cloned()
931            .map(ColumnIdentity::unqualified)
932            .collect();
933        let mut lookup_aliases = input.index.aliases.clone();
934        let mut alias_types = input.index.cold.aliases.clone();
935        for (identity, logical) in aliases {
936            lookup_aliases.insert(identity.clone(), input.slot(*logical).unwrap_or(NULL_SLOT));
937            alias_types.insert(identity.clone(), input.column_type(*logical).cloned());
938        }
939        Self::from_typed_parts_with_aliases(
940            output_names,
941            identities,
942            types,
943            slots,
944            input.physical_width(),
945            lookup_aliases,
946            alias_types,
947        )
948    }
949
950    /// Select logical positions with explicit public labels, SQL identities, and types while preserving hidden aliases and physical fragments.
951    pub(crate) fn remap_typed_identities(
952        input: &Self,
953        columns: &[(String, ColumnIdentity, usize, Option<ColumnType>)],
954        aliases: &[(ColumnIdentity, usize)],
955    ) -> Self {
956        let output_names = columns
957            .iter()
958            .map(|(output, _, _, _)| output.clone())
959            .collect();
960        let identities = columns
961            .iter()
962            .map(|(_, identity, _, _)| identity.clone())
963            .collect();
964        let slots = columns
965            .iter()
966            .map(|(_, _, logical, _)| input.slot(*logical).unwrap_or(NULL_SLOT))
967            .collect();
968        let types = columns.iter().map(|(_, _, _, ty)| ty.clone()).collect();
969        let mut lookup_aliases = input.index.aliases.clone();
970        let mut alias_types = input.index.cold.aliases.clone();
971        for (identity, logical) in aliases {
972            lookup_aliases.insert(identity.clone(), input.slot(*logical).unwrap_or(NULL_SLOT));
973            alias_types.insert(identity.clone(), input.column_type(*logical).cloned());
974        }
975        Self::from_typed_parts_with_aliases(
976            output_names,
977            identities,
978            types,
979            slots,
980            input.physical_width(),
981            lookup_aliases,
982            alias_types,
983        )
984    }
985
986    /// Append freshly-computed values to an existing physical row. Reusing an
987    /// existing output name replaces its logical slot just like map insertion.
988    pub fn append(input: &Self, names: &[String]) -> Self {
989        let columns = names
990            .iter()
991            .cloned()
992            .map(|name| (name, None))
993            .collect::<Vec<_>>();
994        Self::append_typed(input, &columns)
995    }
996
997    /// Append freshly computed values with static SQL output types.
998    pub fn append_typed(input: &Self, names: &[(String, Option<ColumnType>)]) -> Self {
999        let mut columns = input.columns().to_vec();
1000        let mut identities = input.identities().to_vec();
1001        let mut types = input.column_types().to_vec();
1002        let mut slots = input.index.slots.to_vec();
1003        let base = input.physical_width();
1004        for (offset, (name, ty)) in names.iter().enumerate() {
1005            let slot = base + offset;
1006            if let Some(position) = columns.iter().position(|column| column == name) {
1007                slots[position] = slot;
1008                identities[position] = ColumnIdentity::unqualified(name);
1009                types[position].clone_from(ty);
1010            } else {
1011                columns.push(name.clone());
1012                identities.push(ColumnIdentity::unqualified(name));
1013                types.push(ty.clone());
1014                slots.push(slot);
1015            }
1016        }
1017        Self::from_typed_parts_with_aliases(
1018            columns,
1019            identities,
1020            types,
1021            slots,
1022            base + names.len(),
1023            input.index.aliases.clone(),
1024            input.index.cold.aliases.clone(),
1025        )
1026    }
1027
1028    /// Compose two child layouts while retaining duplicate logical labels.
1029    /// Qualified and positional resolution can then distinguish both input
1030    /// slots without copying either value fragment.
1031    pub fn join(
1032        left: &Self,
1033        right: &Self,
1034        extra_columns: impl IntoIterator<Item = String>,
1035    ) -> Self {
1036        let mut columns = left.columns().to_vec();
1037        let mut identities = left.identities().to_vec();
1038        let mut types = left.column_types().to_vec();
1039        let mut slots = left.index.slots.to_vec();
1040        let right_base = left.physical_width();
1041        let mut aliases = left.index.aliases.clone();
1042        let mut alias_types = left.index.cold.aliases.clone();
1043        aliases.extend(right.index.aliases.iter().map(|(name, slot)| {
1044            (
1045                name.clone(),
1046                if *slot == NULL_SLOT {
1047                    NULL_SLOT
1048                } else {
1049                    right_base + *slot
1050                },
1051            )
1052        }));
1053        alias_types.extend(
1054            right
1055                .index
1056                .cold
1057                .aliases
1058                .iter()
1059                .map(|(name, ty)| (name.clone(), ty.clone())),
1060        );
1061        for (right_logical, column) in right.columns().iter().enumerate() {
1062            let slot = right
1063                .slot(right_logical)
1064                .map_or(NULL_SLOT, |slot| right_base + slot);
1065            columns.push(column.clone());
1066            identities.push(right.identities()[right_logical].clone());
1067            types.push(right.column_type(right_logical).cloned());
1068            slots.push(slot);
1069        }
1070        for column in extra_columns {
1071            if !columns.contains(&column) {
1072                identities.push(ColumnIdentity::unqualified(column.clone()));
1073                columns.push(column);
1074                types.push(None);
1075                slots.push(NULL_SLOT);
1076            }
1077        }
1078        Self::from_typed_parts_with_aliases(
1079            columns,
1080            identities,
1081            types,
1082            slots,
1083            left.physical_width() + right.physical_width(),
1084            aliases,
1085            alias_types,
1086        )
1087    }
1088
1089    pub fn view<'a>(&'a self, row: &'a PhysicalRow) -> PhysicalRowView<'a> {
1090        PhysicalRowView { schema: self, row }
1091    }
1092
1093    /// 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.
1094    pub fn relayout_physical_row(
1095        &self,
1096        row: PhysicalRow,
1097        target: &Self,
1098    ) -> ExecResult<PhysicalRow> {
1099        if self.len() != target.len() {
1100            return Err(ExecError::Other(format!(
1101                "cannot relayout {} logical columns as {} logical columns",
1102                self.len(),
1103                target.len()
1104            )));
1105        }
1106
1107        let mut source_slots = vec![None; target.physical_width()];
1108        let mut assign = |target_slot: usize, source_slot: usize| -> ExecResult<()> {
1109            if target_slot == NULL_SLOT {
1110                return Ok(());
1111            }
1112            match source_slots[target_slot] {
1113                Some(existing) if existing != source_slot => Err(ExecError::Other(format!(
1114                    "physical relayout maps target slot {target_slot} to both source slots {existing} and {source_slot}"
1115                ))),
1116                Some(_) => Ok(()),
1117                None => {
1118                    source_slots[target_slot] = Some(source_slot);
1119                    Ok(())
1120                }
1121            }
1122        };
1123
1124        for logical in 0..target.len() {
1125            assign(target.index.slots[logical], self.index.slots[logical])?;
1126        }
1127
1128        for (identity, target_slot) in &target.index.aliases {
1129            if *target_slot == NULL_SLOT {
1130                continue;
1131            }
1132            let mut matching_slots = self
1133                .index
1134                .identities
1135                .iter()
1136                .enumerate()
1137                .filter_map(|(logical, candidate)| {
1138                    (candidate == identity).then_some(self.index.slots[logical])
1139                })
1140                .chain(self.index.aliases.get(identity).copied())
1141                .collect::<Vec<_>>();
1142            matching_slots.sort_unstable();
1143            matching_slots.dedup();
1144            let source_slot = match matching_slots.as_slice() {
1145                [source_slot] => *source_slot,
1146                [] => {
1147                    return Err(ExecError::Other(format!(
1148                        "physical relayout source is missing lookup identity `{identity:?}`"
1149                    )))
1150                }
1151                _ => {
1152                    return Err(ExecError::Other(format!(
1153                        "physical relayout source has ambiguous lookup identity `{identity:?}`"
1154                    )))
1155                }
1156            };
1157            assign(*target_slot, source_slot)?;
1158        }
1159
1160        let source_slots = source_slots
1161            .into_iter()
1162            .map(|slot| slot.unwrap_or(NULL_SLOT))
1163            .collect::<Vec<_>>();
1164        Ok(row.project_slots(&source_slots))
1165    }
1166}
1167
1168#[derive(Debug, Clone, PartialEq)]
1169struct RowFragment {
1170    values: Arc<Vec<Value>>,
1171    /// Fragment-local output slot -> stored value slot. `None` is the common
1172    /// contiguous case. A projection lets an in-memory scan share its stored
1173    /// row even when column pruning selects or reorders fields.
1174    projection: Option<Arc<[usize]>>,
1175}
1176
1177impl RowFragment {
1178    fn contiguous(values: Arc<Vec<Value>>) -> Self {
1179        Self {
1180            values,
1181            projection: None,
1182        }
1183    }
1184
1185    fn projected(values: Arc<Vec<Value>>, projection: Arc<[usize]>) -> Self {
1186        debug_assert!(projection
1187            .iter()
1188            .all(|slot| *slot == NULL_SLOT || *slot < values.len()));
1189        let identity = projection.len() == values.len()
1190            && projection
1191                .iter()
1192                .enumerate()
1193                .all(|(index, slot)| index == *slot);
1194        if identity {
1195            Self::contiguous(values)
1196        } else {
1197            Self {
1198                values,
1199                projection: Some(projection),
1200            }
1201        }
1202    }
1203
1204    fn len(&self) -> usize {
1205        self.projection
1206            .as_ref()
1207            .map_or(self.values.len(), |projection| projection.len())
1208    }
1209
1210    fn get(&self, slot: usize) -> Option<&Value> {
1211        match self.projection.as_ref() {
1212            Some(projection) => match projection.get(slot).copied()? {
1213                NULL_SLOT => Some(&NULL_VALUE),
1214                stored => self.values.get(stored),
1215            },
1216            None => self.values.get(slot),
1217        }
1218    }
1219
1220    fn stored_slot(&self, slot: usize) -> Option<usize> {
1221        match self.projection.as_ref() {
1222            Some(projection) => projection.get(slot).copied(),
1223            None => (slot < self.values.len()).then_some(slot),
1224        }
1225    }
1226
1227    fn into_prefix(mut self, width: usize) -> Self {
1228        debug_assert!(width <= self.len());
1229        if width == self.len() {
1230            return self;
1231        }
1232        if let Some(projection) = self.projection.as_ref() {
1233            self.projection = Some(Arc::from(&projection[..width]));
1234            return self;
1235        }
1236        if let Some(values) = Arc::get_mut(&mut self.values) {
1237            values.truncate(width);
1238        } else {
1239            self.projection = Some((0..width).collect::<Arc<[usize]>>());
1240        }
1241        self
1242    }
1243}
1244
1245type RowFragments = SmallVec<[RowFragment; INLINE_ROW_FRAGMENTS]>;
1246
1247/// A physical row owns no column names. Each fragment is created by a scan or
1248/// projection and shared thereafter; joining rows copies only `Arc` handles.
1249#[derive(Debug, Clone, Default, PartialEq)]
1250pub struct PhysicalRow {
1251    fragments: RowFragments,
1252    lock_origins: Option<Arc<Vec<RowLockOrigin>>>,
1253}
1254
1255/// One output position in a mixed physical projection.
1256#[derive(Debug, Clone, PartialEq)]
1257pub enum RowProjectionValue {
1258    /// Reuse one flattened slot from the input row.
1259    InputSlot(usize),
1260    /// Append a newly computed value.
1261    Owned(Value),
1262}
1263
1264impl PhysicalRow {
1265    pub fn from_values(values: Vec<Value>) -> Self {
1266        let mut fragments = RowFragments::new();
1267        if !values.is_empty() {
1268            fragments.push(RowFragment::contiguous(Arc::new(values)));
1269        }
1270        Self {
1271            fragments,
1272            lock_origins: None,
1273        }
1274    }
1275
1276    /// Build a row by sharing a stored positional value vector and applying a
1277    /// fragment-local slot projection. Neither the values nor contained
1278    /// strings are cloned.
1279    pub fn from_shared_values(values: Arc<Vec<Value>>, projection: Arc<[usize]>) -> Self {
1280        let mut fragments = RowFragments::new();
1281        if !projection.is_empty() {
1282            fragments.push(RowFragment::projected(values, projection));
1283        }
1284        Self {
1285            fragments,
1286            lock_origins: None,
1287        }
1288    }
1289
1290    pub fn from_result_row(schema: &RowSchema, mut row: ResultRow) -> Self {
1291        let values = schema
1292            .columns()
1293            .iter()
1294            .map(|column| row.remove(column).unwrap_or(Value::Null))
1295            .collect();
1296        Self::from_values(values)
1297    }
1298
1299    pub fn nulls(width: usize) -> Self {
1300        Self::from_values(vec![Value::Null; width])
1301    }
1302
1303    pub fn append_values(mut self, values: Vec<Value>) -> Self {
1304        if !values.is_empty() {
1305            self.fragments
1306                .push(RowFragment::contiguous(Arc::new(values)));
1307        }
1308        self
1309    }
1310
1311    pub fn concat(left: &Self, right: &Self) -> Self {
1312        let mut fragments =
1313            RowFragments::with_capacity(left.fragments.len() + right.fragments.len());
1314        fragments.extend(left.fragments.iter().cloned());
1315        fragments.extend(right.fragments.iter().cloned());
1316        let lock_origins =
1317            concat_lock_origins(left.lock_origins.as_ref(), right.lock_origins.as_ref());
1318        Self {
1319            fragments,
1320            lock_origins,
1321        }
1322    }
1323
1324    pub fn concat_left_owned(mut left: Self, right: &Self) -> Self {
1325        left.fragments.extend(right.fragments.iter().cloned());
1326        left.lock_origins =
1327            concat_lock_origins(left.lock_origins.as_ref(), right.lock_origins.as_ref());
1328        left
1329    }
1330
1331    pub fn concat_right_owned(left: &Self, mut right: Self) -> Self {
1332        let mut fragments =
1333            RowFragments::with_capacity(left.fragments.len() + right.fragments.len());
1334        fragments.extend(left.fragments.iter().cloned());
1335        fragments.append(&mut right.fragments);
1336        let lock_origins =
1337            concat_lock_origins(left.lock_origins.as_ref(), right.lock_origins.as_ref());
1338        Self {
1339            fragments,
1340            lock_origins,
1341        }
1342    }
1343
1344    pub(crate) fn value(&self, mut slot: usize) -> Option<&Value> {
1345        for fragment in &self.fragments {
1346            if slot < fragment.len() {
1347                return fragment.get(slot);
1348            }
1349            slot -= fragment.len();
1350        }
1351        None
1352    }
1353
1354    /// Re-express selected flattened slots as a compact positional row while
1355    /// sharing the underlying value vectors. Consecutive slots backed by the
1356    /// same source fragment share one projection fragment; no `Value` (and in
1357    /// particular no string payload) is cloned.
1358    pub(crate) fn project_slots(&self, slots: &[usize]) -> Self {
1359        let mut output = RowFragments::new();
1360        let null_values = Arc::new(Vec::new());
1361        let null_source = self.fragments.len();
1362        let mut current_source = None;
1363        let mut current_values: Option<Arc<Vec<Value>>> = None;
1364        let mut current_projection = Vec::new();
1365
1366        let flush = |output: &mut RowFragments,
1367                     values: &mut Option<Arc<Vec<Value>>>,
1368                     projection: &mut Vec<usize>| {
1369            if let Some(values) = values.take() {
1370                output.push(RowFragment::projected(
1371                    values,
1372                    Arc::from(std::mem::take(projection)),
1373                ));
1374            }
1375        };
1376
1377        for requested in slots {
1378            let mut remaining = *requested;
1379            let resolved = if remaining == NULL_SLOT {
1380                None
1381            } else {
1382                let mut found = None;
1383                for (fragment_index, fragment) in self.fragments.iter().enumerate() {
1384                    if remaining < fragment.len() {
1385                        found = fragment
1386                            .stored_slot(remaining)
1387                            .filter(|slot| *slot != NULL_SLOT)
1388                            .map(|stored| (fragment_index, Arc::clone(&fragment.values), stored));
1389                        break;
1390                    }
1391                    remaining -= fragment.len();
1392                }
1393                found
1394            };
1395            let (source, values, stored) = resolved.map_or_else(
1396                || (null_source, Arc::clone(&null_values), NULL_SLOT),
1397                |(source, values, stored)| (source, values, stored),
1398            );
1399            if current_source != Some(source) {
1400                flush(&mut output, &mut current_values, &mut current_projection);
1401                current_source = Some(source);
1402                current_values = Some(values);
1403            }
1404            current_projection.push(stored);
1405        }
1406        flush(&mut output, &mut current_values, &mut current_projection);
1407        Self {
1408            fragments: output,
1409            lock_origins: self.lock_origins.clone(),
1410        }
1411    }
1412
1413    /// Build an output row from shared input slots and newly computed values while preserving their requested order and sharing row metadata.
1414    pub fn project_with_values(
1415        &self,
1416        values: impl IntoIterator<Item = RowProjectionValue>,
1417    ) -> Self {
1418        fn flush_slots(source: &PhysicalRow, output: &mut RowFragments, slots: &mut Vec<usize>) {
1419            if slots.is_empty() {
1420                return;
1421            }
1422            let mut projected = source.project_slots(slots);
1423            output.append(&mut projected.fragments);
1424            slots.clear();
1425        }
1426
1427        fn flush_owned(output: &mut RowFragments, owned: &mut Vec<Value>) {
1428            if owned.is_empty() {
1429                return;
1430            }
1431            output.push(RowFragment::contiguous(Arc::new(std::mem::take(owned))));
1432        }
1433
1434        let mut fragments = RowFragments::new();
1435        let mut slots = Vec::new();
1436        let mut owned = Vec::new();
1437        for value in values {
1438            match value {
1439                RowProjectionValue::InputSlot(slot) => {
1440                    flush_owned(&mut fragments, &mut owned);
1441                    slots.push(slot);
1442                }
1443                RowProjectionValue::Owned(value) => {
1444                    flush_slots(self, &mut fragments, &mut slots);
1445                    owned.push(value);
1446                }
1447            }
1448        }
1449        flush_slots(self, &mut fragments, &mut slots);
1450        flush_owned(&mut fragments, &mut owned);
1451        Self {
1452            fragments,
1453            lock_origins: self.lock_origins.clone(),
1454        }
1455    }
1456
1457    pub fn fragment_count(&self) -> usize {
1458        self.fragments.len()
1459    }
1460
1461    pub(crate) fn into_prefix(self, width: usize) -> Self {
1462        let mut remaining = width;
1463        let mut fragments = RowFragments::new();
1464        for fragment in self.fragments {
1465            if remaining == 0 {
1466                break;
1467            }
1468            let fragment_width = fragment.len();
1469            if fragment_width <= remaining {
1470                fragments.push(fragment);
1471                remaining -= fragment_width;
1472            } else {
1473                fragments.push(fragment.into_prefix(remaining));
1474                remaining = 0;
1475            }
1476        }
1477        debug_assert_eq!(remaining, 0, "physical row prefix exceeds row width");
1478        Self {
1479            fragments,
1480            lock_origins: self.lock_origins,
1481        }
1482    }
1483}
1484
1485mod physical_row_view;
1486pub use physical_row_view::PhysicalRowView;
1487
1488mod batches;
1489pub use batches::Batch;
1490
1491mod row_lock_origins;
1492use row_lock_origins::concat_lock_origins;
1493pub use row_lock_origins::RowLockOrigin;
1494
1495#[cfg(test)]
1496mod tests;