uqa_execution/batch/
physical_row.rs

1//
2// Unified Query Algebra
3//
4// Copyright (c) 2023-2026 Cognica, Inc.
5//
6
7//! Allocation-light physical row fragments and projections.
8
9use super::{
10    concat_lock_origins, Arc, ResultRow, RowLockOrigin, RowSchema, SmallVec, Value,
11    INLINE_ROW_FRAGMENTS, NULL_SLOT, NULL_VALUE,
12};
13
14#[derive(Debug, Clone, PartialEq)]
15pub(super) struct RowFragment {
16    pub(super) values: Arc<Vec<Value>>,
17    /// Fragment-local output slot -> stored value slot. `None` is the common
18    /// contiguous case. A projection lets an in-memory scan share its stored
19    /// row even when column pruning selects or reorders fields.
20    pub(super) projection: Option<Arc<[usize]>>,
21}
22
23impl RowFragment {
24    fn contiguous(values: Arc<Vec<Value>>) -> Self {
25        Self {
26            values,
27            projection: None,
28        }
29    }
30
31    fn projected(values: Arc<Vec<Value>>, projection: Arc<[usize]>) -> Self {
32        debug_assert!(projection
33            .iter()
34            .all(|slot| *slot == NULL_SLOT || *slot < values.len()));
35        let identity = projection.len() == values.len()
36            && projection
37                .iter()
38                .enumerate()
39                .all(|(index, slot)| index == *slot);
40        if identity {
41            Self::contiguous(values)
42        } else {
43            Self {
44                values,
45                projection: Some(projection),
46            }
47        }
48    }
49
50    pub(super) fn len(&self) -> usize {
51        self.projection
52            .as_ref()
53            .map_or(self.values.len(), |projection| projection.len())
54    }
55
56    pub(super) fn get(&self, slot: usize) -> Option<&Value> {
57        match self.projection.as_ref() {
58            Some(projection) => match projection.get(slot).copied()? {
59                NULL_SLOT => Some(&NULL_VALUE),
60                stored => self.values.get(stored),
61            },
62            None => self.values.get(slot),
63        }
64    }
65
66    fn stored_slot(&self, slot: usize) -> Option<usize> {
67        match self.projection.as_ref() {
68            Some(projection) => projection.get(slot).copied(),
69            None => (slot < self.values.len()).then_some(slot),
70        }
71    }
72
73    fn into_prefix(mut self, width: usize) -> Self {
74        debug_assert!(width <= self.len());
75        if width == self.len() {
76            return self;
77        }
78        if let Some(projection) = self.projection.as_ref() {
79            self.projection = Some(Arc::from(&projection[..width]));
80            return self;
81        }
82        if let Some(values) = Arc::get_mut(&mut self.values) {
83            values.truncate(width);
84        } else {
85            self.projection = Some((0..width).collect::<Arc<[usize]>>());
86        }
87        self
88    }
89}
90
91pub(super) type RowFragments = SmallVec<[RowFragment; INLINE_ROW_FRAGMENTS]>;
92
93/// A physical row owns no column names. Each fragment is created by a scan or
94/// projection and shared thereafter; joining rows copies only `Arc` handles.
95#[derive(Debug, Clone, Default, PartialEq)]
96pub struct PhysicalRow {
97    pub(super) fragments: RowFragments,
98    pub(super) lock_origins: Option<Arc<Vec<RowLockOrigin>>>,
99}
100
101/// One output position in a mixed physical projection.
102#[derive(Debug, Clone, PartialEq)]
103pub enum RowProjectionValue {
104    /// Reuse one flattened slot from the input row.
105    InputSlot(usize),
106    /// Append a newly computed value.
107    Owned(Value),
108}
109
110impl PhysicalRow {
111    pub fn from_values(values: Vec<Value>) -> Self {
112        let mut fragments = RowFragments::new();
113        if !values.is_empty() {
114            fragments.push(RowFragment::contiguous(Arc::new(values)));
115        }
116        Self {
117            fragments,
118            lock_origins: None,
119        }
120    }
121
122    /// Build a row by sharing a stored positional value vector and applying a
123    /// fragment-local slot projection. Neither the values nor contained
124    /// strings are cloned.
125    pub fn from_shared_values(values: Arc<Vec<Value>>, projection: Arc<[usize]>) -> Self {
126        let mut fragments = RowFragments::new();
127        if !projection.is_empty() {
128            fragments.push(RowFragment::projected(values, projection));
129        }
130        Self {
131            fragments,
132            lock_origins: None,
133        }
134    }
135
136    pub fn from_result_row(schema: &RowSchema, mut row: ResultRow) -> Self {
137        let values = schema
138            .columns()
139            .iter()
140            .map(|column| row.remove(column).unwrap_or(Value::Null))
141            .collect();
142        Self::from_values(values)
143    }
144
145    pub fn nulls(width: usize) -> Self {
146        Self::from_values(vec![Value::Null; width])
147    }
148
149    pub fn append_values(mut self, values: Vec<Value>) -> Self {
150        if !values.is_empty() {
151            self.fragments
152                .push(RowFragment::contiguous(Arc::new(values)));
153        }
154        self
155    }
156
157    pub fn concat(left: &Self, right: &Self) -> Self {
158        let mut fragments =
159            RowFragments::with_capacity(left.fragments.len() + right.fragments.len());
160        fragments.extend(left.fragments.iter().cloned());
161        fragments.extend(right.fragments.iter().cloned());
162        let lock_origins =
163            concat_lock_origins(left.lock_origins.as_ref(), right.lock_origins.as_ref());
164        Self {
165            fragments,
166            lock_origins,
167        }
168    }
169
170    pub fn concat_left_owned(mut left: Self, right: &Self) -> Self {
171        left.fragments.extend(right.fragments.iter().cloned());
172        left.lock_origins =
173            concat_lock_origins(left.lock_origins.as_ref(), right.lock_origins.as_ref());
174        left
175    }
176
177    pub fn concat_right_owned(left: &Self, mut right: Self) -> Self {
178        let mut fragments =
179            RowFragments::with_capacity(left.fragments.len() + right.fragments.len());
180        fragments.extend(left.fragments.iter().cloned());
181        fragments.append(&mut right.fragments);
182        let lock_origins =
183            concat_lock_origins(left.lock_origins.as_ref(), right.lock_origins.as_ref());
184        Self {
185            fragments,
186            lock_origins,
187        }
188    }
189
190    pub(crate) fn value(&self, mut slot: usize) -> Option<&Value> {
191        for fragment in &self.fragments {
192            if slot < fragment.len() {
193                return fragment.get(slot);
194            }
195            slot -= fragment.len();
196        }
197        None
198    }
199
200    /// Re-express selected flattened slots as a compact positional row while
201    /// sharing the underlying value vectors. Consecutive slots backed by the
202    /// same source fragment share one projection fragment; no `Value` (and in
203    /// particular no string payload) is cloned.
204    pub(crate) fn project_slots(&self, slots: &[usize]) -> Self {
205        let mut output = RowFragments::new();
206        let null_values = Arc::new(Vec::new());
207        let null_source = self.fragments.len();
208        let mut current_source = None;
209        let mut current_values: Option<Arc<Vec<Value>>> = None;
210        let mut current_projection = Vec::new();
211
212        let flush = |output: &mut RowFragments,
213                     values: &mut Option<Arc<Vec<Value>>>,
214                     projection: &mut Vec<usize>| {
215            if let Some(values) = values.take() {
216                output.push(RowFragment::projected(
217                    values,
218                    Arc::from(std::mem::take(projection)),
219                ));
220            }
221        };
222
223        for requested in slots {
224            let mut remaining = *requested;
225            let resolved = if remaining == NULL_SLOT {
226                None
227            } else {
228                let mut found = None;
229                for (fragment_index, fragment) in self.fragments.iter().enumerate() {
230                    if remaining < fragment.len() {
231                        found = fragment
232                            .stored_slot(remaining)
233                            .filter(|slot| *slot != NULL_SLOT)
234                            .map(|stored| (fragment_index, Arc::clone(&fragment.values), stored));
235                        break;
236                    }
237                    remaining -= fragment.len();
238                }
239                found
240            };
241            let (source, values, stored) = resolved.map_or_else(
242                || (null_source, Arc::clone(&null_values), NULL_SLOT),
243                |(source, values, stored)| (source, values, stored),
244            );
245            if current_source != Some(source) {
246                flush(&mut output, &mut current_values, &mut current_projection);
247                current_source = Some(source);
248                current_values = Some(values);
249            }
250            current_projection.push(stored);
251        }
252        flush(&mut output, &mut current_values, &mut current_projection);
253        Self {
254            fragments: output,
255            lock_origins: self.lock_origins.clone(),
256        }
257    }
258
259    /// Build an output row from shared input slots and newly computed values while preserving their requested order and sharing row metadata.
260    pub fn project_with_values(
261        &self,
262        values: impl IntoIterator<Item = RowProjectionValue>,
263    ) -> Self {
264        fn flush_slots(source: &PhysicalRow, output: &mut RowFragments, slots: &mut Vec<usize>) {
265            if slots.is_empty() {
266                return;
267            }
268            let mut projected = source.project_slots(slots);
269            output.append(&mut projected.fragments);
270            slots.clear();
271        }
272
273        fn flush_owned(output: &mut RowFragments, owned: &mut Vec<Value>) {
274            if owned.is_empty() {
275                return;
276            }
277            output.push(RowFragment::contiguous(Arc::new(std::mem::take(owned))));
278        }
279
280        let mut fragments = RowFragments::new();
281        let mut slots = Vec::new();
282        let mut owned = Vec::new();
283        for value in values {
284            match value {
285                RowProjectionValue::InputSlot(slot) => {
286                    flush_owned(&mut fragments, &mut owned);
287                    slots.push(slot);
288                }
289                RowProjectionValue::Owned(value) => {
290                    flush_slots(self, &mut fragments, &mut slots);
291                    owned.push(value);
292                }
293            }
294        }
295        flush_slots(self, &mut fragments, &mut slots);
296        flush_owned(&mut fragments, &mut owned);
297        Self {
298            fragments,
299            lock_origins: self.lock_origins.clone(),
300        }
301    }
302
303    pub fn fragment_count(&self) -> usize {
304        self.fragments.len()
305    }
306
307    pub(crate) fn into_prefix(self, width: usize) -> Self {
308        let mut remaining = width;
309        let mut fragments = RowFragments::new();
310        for fragment in self.fragments {
311            if remaining == 0 {
312                break;
313            }
314            let fragment_width = fragment.len();
315            if fragment_width <= remaining {
316                fragments.push(fragment);
317                remaining -= fragment_width;
318            } else {
319                fragments.push(fragment.into_prefix(remaining));
320                remaining = 0;
321            }
322        }
323        debug_assert_eq!(remaining, 0, "physical row prefix exceeds row width");
324        Self {
325            fragments,
326            lock_origins: self.lock_origins,
327        }
328    }
329}
uqa_execution/batch/physical_row.rs

uqa_execution/batch/
physical_row.rs
