1use 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
31pub const DEFAULT_BATCH_SIZE: usize = 1024;
33
34const NULL_SLOT: usize = usize::MAX;
35const INLINE_ROW_FRAGMENTS: usize = 7;
37static NULL_VALUE: Value = Value::Null;
38
39#[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 columns: Box<[String]>,
78 identities: Box<[ColumnIdentity]>,
80 slots: Box<[usize]>,
82 physical_width: usize,
83 exact: HashMap<Box<str>, usize>,
85 unqualified: HashMap<Box<str>, usize>,
86 qualified: HashMap<ColumnIdentity, usize>,
87 aliases: HashMap<ColumnIdentity, usize>,
89 ambiguous_unqualified: HashSet<Box<str>>,
91 ambiguous_qualified: HashSet<ColumnIdentity>,
93 cold: Box<SchemaColdMetadata>,
95}
96
97#[derive(Debug, PartialEq, Eq)]
98struct SchemaColdMetadata {
99 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#[derive(Debug, Clone, PartialEq, Eq)]
120pub struct RowSchema {
121 index: Arc<SchemaIndex>,
122}
123
124#[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 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 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 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 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 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 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 pub fn column_types(&self) -> &[Option<ColumnType>] {
412 &self.index.cold.columns
413 }
414
415 pub fn identities(&self) -> &[ColumnIdentity] {
417 &self.index.identities
418 }
419
420 #[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 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 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 pub fn physical_slot(&self, logical: usize) -> Option<usize> {
465 self.slot(logical)
466 }
467
468 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 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 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 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 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 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 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 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 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 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 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 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 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 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 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 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 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 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#[derive(Debug, Clone, Default, PartialEq)]
1250pub struct PhysicalRow {
1251 fragments: RowFragments,
1252 lock_origins: Option<Arc<Vec<RowLockOrigin>>>,
1253}
1254
1255#[derive(Debug, Clone, PartialEq)]
1257pub enum RowProjectionValue {
1258 InputSlot(usize),
1260 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 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 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 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;