hawdb_executor/
profile.rs

1// Copyright 2026 Nowledge
2//
3// Licensed under the Apache License, Version 2.0 (the "License");
4// you may not use this file except in compliance with the License.
5// You may obtain a copy of the License at
6//
7//     http://www.apache.org/licenses/LICENSE-2.0
8//
9// Unless required by applicable law or agreed to in writing, software
10// distributed under the License is distributed on an "AS IS" BASIS,
11// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12// See the License for the specific language governing permissions and
13// limitations under the License.
14
15use crate::binding::value_payload_bytes;
16use crate::columnar::ColumnarRowRef;
17use hawdb_core::{HawDBError, Result, Value, ValueRef};
18use hawdb_plan_cypher::{PhysicalOperatorId, PhysicalPlanKind};
19use std::collections::BTreeMap;
20use std::ops::Index;
21use std::sync::{Arc, OnceLock};
22
23pub type Row = BTreeMap<String, Value>;
24
25/// Column names shared by every row of one query result.
26#[derive(Debug, Clone, PartialEq, Eq)]
27pub struct QuerySchema {
28    columns: Arc<[String]>,
29}
30
31impl QuerySchema {
32    pub fn try_new(columns: impl IntoIterator<Item = String>) -> Result<Self> {
33        let columns = columns.into_iter().collect::<Vec<_>>();
34        let mut unique = std::collections::BTreeSet::new();
35        if let Some(duplicate) = columns
36            .iter()
37            .find(|column| !unique.insert(column.as_str()))
38        {
39            return Err(HawDBError::Semantic(format!(
40                "query result schema contains duplicate column {duplicate}"
41            )));
42        }
43        Ok(Self {
44            columns: Arc::from(columns),
45        })
46    }
47
48    pub fn empty() -> Self {
49        Self {
50            columns: Arc::from([]),
51        }
52    }
53
54    pub fn columns(&self) -> &[String] {
55        &self.columns
56    }
57
58    pub fn len(&self) -> usize {
59        self.columns.len()
60    }
61
62    pub fn is_empty(&self) -> bool {
63        self.columns.is_empty()
64    }
65
66    pub fn position(&self, name: &str) -> Option<usize> {
67        self.columns.iter().position(|column| column == name)
68    }
69}
70
71#[derive(Debug)]
72struct QueryRowStorage {
73    schema: QuerySchema,
74    values: QueryStoredValues,
75    row_count: usize,
76}
77
78#[derive(Debug)]
79enum QueryStoredValues {
80    Flat(Vec<Value>),
81    Segmented(Vec<Vec<Value>>),
82}
83
84impl QueryStoredValues {
85    fn len(&self) -> usize {
86        match self {
87            Self::Flat(values) => values.len(),
88            Self::Segmented(segments) => segments.iter().map(Vec::len).sum(),
89        }
90    }
91}
92
93mod owned_result;
94
95/// One owned handle into a query's immutable row storage.
96#[derive(Clone)]
97pub struct QueryRow {
98    storage: Arc<QueryRowStorage>,
99    row: usize,
100}
101
102impl QueryRow {
103    pub fn schema(&self) -> &QuerySchema {
104        &self.storage.schema
105    }
106
107    pub fn len(&self) -> usize {
108        self.storage.schema.len()
109    }
110
111    pub fn is_empty(&self) -> bool {
112        self.len() == 0
113    }
114
115    pub fn value(&self, column: usize) -> Option<&Value> {
116        row_values(&self.storage, self.row)?.get(column)
117    }
118
119    pub fn get(&self, column: &str) -> Option<&Value> {
120        self.value(self.storage.schema.position(column)?)
121    }
122
123    pub fn contains_key(&self, column: &str) -> bool {
124        self.storage.schema.position(column).is_some()
125    }
126
127    pub fn iter(&self) -> impl ExactSizeIterator<Item = (&str, &Value)> {
128        self.storage
129            .schema
130            .columns()
131            .iter()
132            .map(String::as_str)
133            .zip(row_values(&self.storage, self.row).unwrap_or_default())
134    }
135
136    pub fn keys(&self) -> impl ExactSizeIterator<Item = &str> {
137        self.storage.schema.columns().iter().map(String::as_str)
138    }
139
140    pub fn values(&self) -> impl ExactSizeIterator<Item = &Value> {
141        row_values(&self.storage, self.row)
142            .unwrap_or_default()
143            .iter()
144    }
145
146    pub fn as_ref(&self) -> QueryRowRef<'_> {
147        QueryRowRef {
148            schema: &self.storage.schema,
149            values: row_values(&self.storage, self.row).unwrap_or_default(),
150        }
151    }
152
153    pub fn to_owned_row(&self) -> Row {
154        self.iter()
155            .map(|(name, value)| (name.to_owned(), value.clone()))
156            .collect()
157    }
158}
159
160impl std::fmt::Debug for QueryRow {
161    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
162        f.debug_map().entries(self.iter()).finish()
163    }
164}
165
166impl PartialEq for QueryRow {
167    fn eq(&self, other: &Self) -> bool {
168        self.schema() == other.schema()
169            && row_values(&self.storage, self.row) == row_values(&other.storage, other.row)
170    }
171}
172
173impl Eq for QueryRow {}
174
175impl PartialEq<Row> for QueryRow {
176    fn eq(&self, other: &Row) -> bool {
177        row_equals_map(self.as_ref(), other)
178    }
179}
180
181impl PartialEq<QueryRow> for Row {
182    fn eq(&self, other: &QueryRow) -> bool {
183        other == self
184    }
185}
186
187impl Index<&str> for QueryRow {
188    type Output = Value;
189
190    fn index(&self, column: &str) -> &Self::Output {
191        self.get(column)
192            .unwrap_or_else(|| panic!("query result has no column {column}"))
193    }
194}
195
196/// A borrowed row view over a query's immutable contiguous row buffer.
197#[derive(Debug, Clone, Copy)]
198pub struct QueryRowRef<'a> {
199    schema: &'a QuerySchema,
200    values: &'a [Value],
201}
202
203impl<'a> QueryRowRef<'a> {
204    pub fn schema(self) -> &'a QuerySchema {
205        self.schema
206    }
207
208    pub fn len(self) -> usize {
209        self.values.len()
210    }
211
212    pub fn is_empty(self) -> bool {
213        self.values.is_empty()
214    }
215
216    pub fn value(self, column: usize) -> Option<&'a Value> {
217        self.values.get(column)
218    }
219
220    pub fn get(self, column: &str) -> Option<&'a Value> {
221        self.value(self.schema.position(column)?)
222    }
223
224    pub fn contains_key(self, column: &str) -> bool {
225        self.schema.position(column).is_some()
226    }
227
228    pub fn iter(self) -> impl ExactSizeIterator<Item = (&'a str, &'a Value)> {
229        self.schema
230            .columns()
231            .iter()
232            .map(String::as_str)
233            .zip(self.values)
234    }
235
236    pub fn keys(self) -> impl ExactSizeIterator<Item = &'a str> {
237        self.schema.columns().iter().map(String::as_str)
238    }
239
240    pub fn values(self) -> impl ExactSizeIterator<Item = &'a Value> {
241        self.values.iter()
242    }
243
244    pub fn to_owned_row(self) -> Row {
245        self.iter()
246            .map(|(name, value)| (name.to_owned(), value.clone()))
247            .collect()
248    }
249}
250
251impl PartialEq<Row> for QueryRowRef<'_> {
252    fn eq(&self, other: &Row) -> bool {
253        row_equals_map(*self, other)
254    }
255}
256
257impl PartialEq<QueryRowRef<'_>> for Row {
258    fn eq(&self, other: &QueryRowRef<'_>) -> bool {
259        other == self
260    }
261}
262
263impl Index<&str> for QueryRowRef<'_> {
264    type Output = Value;
265
266    fn index(&self, column: &str) -> &Self::Output {
267        self.get(column)
268            .unwrap_or_else(|| panic!("query result has no column {column}"))
269    }
270}
271
272/// Schema-bearing result rows backed by immutable contiguous row buffers.
273pub struct QueryRows {
274    storage: Arc<QueryRowStorage>,
275    indexed_rows: OnceLock<Vec<QueryRow>>,
276}
277
278impl QueryRows {
279    pub fn empty() -> Self {
280        Self::from_flat_values_unchecked(QuerySchema::empty(), Vec::new(), 0)
281    }
282
283    fn from_flat_values_unchecked(
284        schema: QuerySchema,
285        values: Vec<Value>,
286        row_count: usize,
287    ) -> Self {
288        Self {
289            storage: Arc::new(QueryRowStorage {
290                schema,
291                values: QueryStoredValues::Flat(values),
292                row_count,
293            }),
294            indexed_rows: OnceLock::new(),
295        }
296    }
297
298    pub fn schema(&self) -> &QuerySchema {
299        &self.storage.schema
300    }
301
302    pub fn value_rows(&self) -> QueryValueRows<'_> {
303        QueryValueRows {
304            rows: self,
305            next: 0,
306        }
307    }
308
309    pub fn len(&self) -> usize {
310        self.storage.row_count
311    }
312
313    pub fn is_empty(&self) -> bool {
314        self.storage.row_count == 0
315    }
316
317    pub fn value(&self, row: usize, column: usize) -> Option<ValueRef<'_>> {
318        row_values(&self.storage, row)?
319            .get(column)
320            .map(Value::as_ref)
321    }
322
323    pub fn get(&self, row: usize, column: &str) -> Option<ValueRef<'_>> {
324        self.value(row, self.schema().position(column)?)
325    }
326
327    pub fn into_rows(self) -> Vec<Row> {
328        (0..self.len())
329            .map(|row| {
330                self.row(row)
331                    .expect("row ordinal is in range")
332                    .to_owned_row()
333            })
334            .collect()
335    }
336
337    pub fn row(&self, row: usize) -> Option<QueryRowRef<'_>> {
338        if row >= self.len() {
339            return None;
340        }
341        Some(QueryRowRef {
342            schema: &self.storage.schema,
343            values: row_values(&self.storage, row)?,
344        })
345    }
346
347    pub fn first(&self) -> Option<QueryRowRef<'_>> {
348        self.row(0)
349    }
350
351    pub fn last(&self) -> Option<QueryRowRef<'_>> {
352        self.len().checked_sub(1).and_then(|row| self.row(row))
353    }
354
355    pub fn iter(&self) -> QueryRowsIter<'_> {
356        QueryRowsIter {
357            rows: self,
358            next: 0,
359        }
360    }
361
362    pub fn retain(&mut self, mut keep: impl FnMut(QueryRowRef<'_>) -> bool) {
363        let width = self.schema().len();
364        let mut values = Vec::with_capacity(self.storage.values.len());
365        let mut row_count = 0usize;
366        for row in self.iter() {
367            if keep(row) {
368                values.extend_from_slice(row.values);
369                row_count = row_count.saturating_add(1);
370            }
371        }
372        let retained_row_count = values.len().checked_div(width).unwrap_or(row_count);
373        self.storage = Arc::new(QueryRowStorage {
374            schema: self.schema().clone(),
375            values: QueryStoredValues::Flat(values),
376            row_count: retained_row_count,
377        });
378        self.indexed_rows = OnceLock::new();
379    }
380
381    /// Returns stable owned row handles for APIs that require slice pattern
382    /// matching or indexing. No map or value is copied.
383    pub fn as_slice(&self) -> &[QueryRow] {
384        self.indexed_rows()
385    }
386
387    pub fn payload_bytes(&self) -> usize {
388        let names = self
389            .schema()
390            .columns()
391            .iter()
392            .fold(0usize, |total, name| total.saturating_add(name.len()));
393        stored_values(&self.storage).fold(names, |total, value| {
394            total.saturating_add(value_payload_bytes(value))
395        })
396    }
397
398    fn indexed_rows(&self) -> &[QueryRow] {
399        self.indexed_rows.get_or_init(|| {
400            (0..self.len())
401                .map(|row| QueryRow {
402                    storage: Arc::clone(&self.storage),
403                    row,
404                })
405                .collect()
406        })
407    }
408}
409
410impl Default for QueryRows {
411    fn default() -> Self {
412        Self::empty()
413    }
414}
415
416impl From<Vec<Row>> for QueryRows {
417    fn from(rows: Vec<Row>) -> Self {
418        let columns = rows
419            .iter()
420            .flat_map(|row| row.keys().cloned())
421            .collect::<std::collections::BTreeSet<_>>()
422            .into_iter()
423            .collect::<Vec<_>>();
424        let schema = QuerySchema::try_new(columns).expect("BTreeSet columns are unique");
425        let mut builder = QueryRowsBuilder::with_schema(schema.clone(), rows.len());
426        for mut row in rows {
427            builder
428                .push_values(
429                    schema
430                        .columns()
431                        .iter()
432                        .map(|column| row.remove(column).unwrap_or(Value::Null)),
433                )
434                .expect("row normalization always matches the inferred schema");
435        }
436        builder.finish()
437    }
438}
439
440impl FromIterator<Row> for QueryRows {
441    fn from_iter<T: IntoIterator<Item = Row>>(iter: T) -> Self {
442        iter.into_iter().collect::<Vec<_>>().into()
443    }
444}
445
446/// Builds immutable query results without retaining one map or vector per row.
447///
448/// The first named row binds column names to stable ordinals. Later rows are
449/// consumed in the same deterministic `BTreeMap` order and their values move
450/// directly into bounded row segments when cardinality is unknown. Explicit
451/// capacity constructors retain the flat-buffer path.
452pub struct QueryRowsBuilder {
453    schema: Option<QuerySchema>,
454    values: Vec<Value>,
455    row_count: usize,
456    row_capacity_hint: usize,
457    segments: Option<Vec<Vec<Value>>>,
458}
459
460impl QueryRowsBuilder {
461    pub fn new() -> Self {
462        let mut builder = Self::with_row_capacity(0);
463        builder.segments = Some(Vec::new());
464        builder
465    }
466
467    pub fn with_row_capacity(row_capacity: usize) -> Self {
468        Self {
469            schema: None,
470            values: Vec::new(),
471            row_count: 0,
472            row_capacity_hint: row_capacity,
473            segments: None,
474        }
475    }
476
477    pub fn with_schema(schema: QuerySchema, row_capacity: usize) -> Self {
478        Self {
479            values: Vec::with_capacity(row_capacity.saturating_mul(schema.len())),
480            schema: Some(schema),
481            row_count: 0,
482            row_capacity_hint: row_capacity,
483            segments: None,
484        }
485    }
486
487    pub fn push_values(&mut self, values: impl IntoIterator<Item = Value>) -> Result<()> {
488        self.try_push_values(values.into_iter().map(Ok))
489    }
490
491    /// Appends one fallible row into immutable-result storage.
492    ///
493    /// A conversion error or width mismatch rolls the partial row back.
494    pub fn try_push_values(
495        &mut self,
496        values: impl IntoIterator<Item = Result<Value>>,
497    ) -> Result<()> {
498        let schema = self.schema.as_ref().ok_or_else(|| {
499            HawDBError::Execution(
500                "query result values require a schema before ordinal insertion".to_string(),
501            )
502        })?;
503        if self.segments.is_some() {
504            let values = values.into_iter().collect::<Result<Vec<_>>>()?;
505            if values.len() != schema.len() {
506                return Err(HawDBError::Execution(format!(
507                    "query result row {} has width {}, expected {}",
508                    self.row_count,
509                    values.len(),
510                    schema.len()
511                )));
512            }
513            return self.append_named_values(values.into_iter());
514        }
515        let start = self.values.len();
516        for value in values {
517            match value {
518                Ok(value) => self.values.push(value),
519                Err(error) => {
520                    self.values.truncate(start);
521                    return Err(error);
522                }
523            }
524        }
525        let width = self.values.len().saturating_sub(start);
526        if width != schema.len() {
527            self.values.truncate(start);
528            return Err(HawDBError::Execution(format!(
529                "query result row {} has width {width}, expected {}",
530                self.row_count,
531                schema.len()
532            )));
533        }
534        self.row_count = self.row_count.saturating_add(1);
535        Ok(())
536    }
537
538    pub fn push_named_row(&mut self, row: Row) -> Result<()> {
539        if self.schema.is_none() {
540            self.schema = Some(QuerySchema::try_new(row.keys().cloned())?);
541            self.values.reserve(
542                row.len()
543                    .saturating_mul(self.row_capacity_hint.max(self.row_count.saturating_add(1))),
544            );
545        }
546        let schema = self.schema.as_ref().expect("query result schema is bound");
547        if row.len() != schema.len() {
548            return Err(HawDBError::Execution(format!(
549                "query result row {} has width {}, expected {}",
550                self.row_count,
551                row.len(),
552                schema.len()
553            )));
554        }
555        if let Some((ordinal, (name, expected))) = row
556            .keys()
557            .zip(schema.columns())
558            .enumerate()
559            .find(|(_, (name, expected))| *name != *expected)
560        {
561            return Err(HawDBError::Execution(format!(
562                "query result row {} column {ordinal} is {name}, expected {expected}",
563                self.row_count
564            )));
565        }
566        self.append_named_values(row.into_values())
567    }
568
569    fn append_named_values(&mut self, values: impl Iterator<Item = Value>) -> Result<()> {
570        if let Some(segments) = &mut self.segments {
571            let width = self
572                .schema
573                .as_ref()
574                .expect("query result schema is bound")
575                .len();
576            let rows_in_segment = self.row_count % OWNED_RESULT_SEGMENT_ROWS;
577            if rows_in_segment == 0 && self.row_count != 0 && width != 0 {
578                segments.push(std::mem::take(&mut self.values));
579            }
580            let needed = (rows_in_segment + 1).checked_mul(width).ok_or_else(|| {
581                HawDBError::Execution("query result segment capacity overflow".into())
582            })?;
583            if needed > self.values.capacity() {
584                let target_rows = (rows_in_segment + 1).next_power_of_two();
585                let target = target_rows.checked_mul(width).ok_or_else(|| {
586                    HawDBError::Execution("query result segment capacity overflow".into())
587                })?;
588                self.values.reserve_exact(target - self.values.len());
589            }
590        }
591        self.values.extend(values);
592        self.row_count = self.row_count.saturating_add(1);
593        Ok(())
594    }
595
596    pub fn finish(mut self) -> QueryRows {
597        if let Some(mut segments) = self.segments.take() {
598            if segments.is_empty() {
599                return QueryRows::from_flat_values_unchecked(
600                    self.schema.unwrap_or_else(QuerySchema::empty),
601                    self.values,
602                    self.row_count,
603                );
604            }
605            if !self.values.is_empty() {
606                segments.push(self.values);
607            }
608            return QueryRows {
609                storage: Arc::new(QueryRowStorage {
610                    schema: self.schema.unwrap_or_else(QuerySchema::empty),
611                    values: QueryStoredValues::Segmented(segments),
612                    row_count: self.row_count,
613                }),
614                indexed_rows: OnceLock::new(),
615            };
616        }
617        QueryRows::from_flat_values_unchecked(
618            self.schema.unwrap_or_else(QuerySchema::empty),
619            self.values,
620            self.row_count,
621        )
622    }
623}
624
625impl Default for QueryRowsBuilder {
626    fn default() -> Self {
627        Self::new()
628    }
629}
630
631impl Clone for QueryRows {
632    fn clone(&self) -> Self {
633        Self {
634            storage: Arc::clone(&self.storage),
635            indexed_rows: OnceLock::new(),
636        }
637    }
638}
639
640impl std::fmt::Debug for QueryRows {
641    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
642        f.debug_struct("QueryRows")
643            .field("schema", self.schema())
644            .field("values", &self.storage.values)
645            .finish()
646    }
647}
648
649impl PartialEq for QueryRows {
650    fn eq(&self, other: &Self) -> bool {
651        self.schema() == other.schema()
652            && self.len() == other.len()
653            && stored_values(&self.storage).eq(stored_values(&other.storage))
654    }
655}
656
657impl Eq for QueryRows {}
658
659impl PartialEq<Vec<Row>> for QueryRows {
660    fn eq(&self, other: &Vec<Row>) -> bool {
661        self.len() == other.len() && self.iter().zip(other).all(|(left, right)| left == *right)
662    }
663}
664
665impl PartialEq<QueryRows> for Vec<Row> {
666    fn eq(&self, other: &QueryRows) -> bool {
667        other == self
668    }
669}
670
671impl Index<usize> for QueryRows {
672    type Output = QueryRow;
673
674    fn index(&self, row: usize) -> &Self::Output {
675        &self.indexed_rows()[row]
676    }
677}
678
679impl<'a> IntoIterator for &'a QueryRows {
680    type Item = QueryRowRef<'a>;
681    type IntoIter = QueryRowsIter<'a>;
682
683    fn into_iter(self) -> Self::IntoIter {
684        self.iter()
685    }
686}
687
688impl IntoIterator for QueryRows {
689    type Item = QueryRow;
690    type IntoIter = QueryRowsIntoIter;
691
692    fn into_iter(self) -> Self::IntoIter {
693        QueryRowsIntoIter {
694            storage: self.storage,
695            next: 0,
696        }
697    }
698}
699
700pub struct QueryRowsIter<'a> {
701    rows: &'a QueryRows,
702    next: usize,
703}
704
705impl<'a> Iterator for QueryRowsIter<'a> {
706    type Item = QueryRowRef<'a>;
707
708    fn next(&mut self) -> Option<Self::Item> {
709        let row = self.rows.row(self.next)?;
710        self.next = self.next.saturating_add(1);
711        Some(row)
712    }
713
714    fn size_hint(&self) -> (usize, Option<usize>) {
715        let remaining = self.rows.len().saturating_sub(self.next);
716        (remaining, Some(remaining))
717    }
718}
719
720impl ExactSizeIterator for QueryRowsIter<'_> {}
721
722pub struct QueryValueRows<'a> {
723    rows: &'a QueryRows,
724    next: usize,
725}
726
727impl QueryValueRows<'_> {
728    pub fn is_empty(&self) -> bool {
729        self.len() == 0
730    }
731}
732
733impl<'a> Iterator for QueryValueRows<'a> {
734    type Item = &'a [Value];
735
736    fn next(&mut self) -> Option<Self::Item> {
737        let values = self.rows.row(self.next)?.values;
738        self.next = self.next.saturating_add(1);
739        Some(values)
740    }
741
742    fn size_hint(&self) -> (usize, Option<usize>) {
743        let remaining = self.rows.len().saturating_sub(self.next);
744        (remaining, Some(remaining))
745    }
746}
747
748impl ExactSizeIterator for QueryValueRows<'_> {}
749
750pub struct QueryRowsIntoIter {
751    storage: Arc<QueryRowStorage>,
752    next: usize,
753}
754
755impl Iterator for QueryRowsIntoIter {
756    type Item = QueryRow;
757
758    fn next(&mut self) -> Option<Self::Item> {
759        if self.next >= self.storage.row_count {
760            return None;
761        }
762        let row = QueryRow {
763            storage: Arc::clone(&self.storage),
764            row: self.next,
765        };
766        self.next = self.next.saturating_add(1);
767        Some(row)
768    }
769
770    fn size_hint(&self) -> (usize, Option<usize>) {
771        let remaining = self.storage.row_count.saturating_sub(self.next);
772        (remaining, Some(remaining))
773    }
774}
775
776impl ExactSizeIterator for QueryRowsIntoIter {}
777
778const OWNED_RESULT_SEGMENT_ROWS: usize = 256;
779
780fn stored_values(storage: &QueryRowStorage) -> impl Iterator<Item = &Value> {
781    let (flat, segments) = match &storage.values {
782        QueryStoredValues::Flat(values) => (Some(values), None),
783        QueryStoredValues::Segmented(segments) => (None, Some(segments)),
784    };
785    flat.into_iter().flatten().chain(
786        segments
787            .into_iter()
788            .flatten()
789            .flat_map(|segment| segment.iter()),
790    )
791}
792
793fn row_values(storage: &QueryRowStorage, row: usize) -> Option<&[Value]> {
794    if row >= storage.row_count {
795        return None;
796    }
797    let width = storage.schema.len();
798    if width == 0 {
799        return Some(&[]);
800    }
801    let (values, ordinal) = match &storage.values {
802        QueryStoredValues::Segmented(segments) => (
803            segments.get(row / OWNED_RESULT_SEGMENT_ROWS)?,
804            row % OWNED_RESULT_SEGMENT_ROWS,
805        ),
806        QueryStoredValues::Flat(values) => (values, row),
807    };
808    let start = ordinal.checked_mul(width)?;
809    let end = start.checked_add(width)?;
810    values.get(start..end)
811}
812
813fn row_equals_map(row: QueryRowRef<'_>, other: &Row) -> bool {
814    row.len() == other.len()
815        && row
816            .iter()
817            .all(|(name, value)| other.get(name).is_some_and(|other| other == value))
818}
819
820/// A row view whose values remain valid only for the current consumer call.
821///
822/// The map representation covers the scalar fallback executor. The columnar
823/// representation lets a downstream consumer pull selected values directly
824/// from an immutable batch without constructing an intermediate row map.
825#[derive(Debug, Clone, Copy)]
826pub enum RowRef<'a> {
827    Map(&'a Row),
828    Columnar(ColumnarRowRef<'a>),
829}
830
831impl<'a> RowRef<'a> {
832    pub fn len(self) -> usize {
833        match self {
834            Self::Map(row) => row.len(),
835            Self::Columnar(_) => self.iter().count(),
836        }
837    }
838
839    pub fn is_empty(self) -> bool {
840        self.len() == 0
841    }
842
843    pub fn get(self, name: &str) -> Option<ValueRef<'a>> {
844        match self {
845            Self::Map(row) => row.get(name).map(Value::as_ref),
846            Self::Columnar(row) => row.get(name),
847        }
848    }
849
850    pub fn column(self, index: usize) -> Option<(&'a str, ValueRef<'a>)> {
851        match self {
852            Self::Map(row) => row
853                .iter()
854                .nth(index)
855                .map(|(name, value)| (name.as_str(), value.as_ref())),
856            Self::Columnar(row) => row.column(index),
857        }
858    }
859
860    pub fn iter(self) -> RowRefIter<'a> {
861        match self {
862            Self::Map(row) => RowRefIter::Map(row.iter()),
863            Self::Columnar(row) => RowRefIter::Columnar { row, index: 0 },
864        }
865    }
866
867    pub fn to_owned_row(self) -> Row {
868        self.iter()
869            .map(|(name, value)| (name.to_owned(), value.to_owned_value()))
870            .collect()
871    }
872}
873
874impl<'a> From<&'a Row> for RowRef<'a> {
875    fn from(row: &'a Row) -> Self {
876        Self::Map(row)
877    }
878}
879
880impl<'a> From<ColumnarRowRef<'a>> for RowRef<'a> {
881    fn from(row: ColumnarRowRef<'a>) -> Self {
882        Self::Columnar(row)
883    }
884}
885
886pub enum RowRefIter<'a> {
887    Map(std::collections::btree_map::Iter<'a, String, Value>),
888    Columnar {
889        row: ColumnarRowRef<'a>,
890        index: usize,
891    },
892}
893
894impl<'a> Iterator for RowRefIter<'a> {
895    type Item = (&'a str, ValueRef<'a>);
896
897    fn next(&mut self) -> Option<Self::Item> {
898        match self {
899            Self::Map(iter) => iter
900                .next()
901                .map(|(name, value)| (name.as_str(), value.as_ref())),
902            Self::Columnar { row, index } => loop {
903                let current = *index;
904                *index = index.saturating_add(1);
905                if current >= row.schema().len() {
906                    return None;
907                }
908                if let Some(column) = row.column(current) {
909                    return Some(column);
910                }
911            },
912        }
913    }
914}
915
916#[derive(Debug, Clone, PartialEq, Eq)]
917pub struct BlockingOperatorMemoryReport {
918    pub operator: String,
919    pub budget_bytes: usize,
920    pub peak_tracked_bytes: usize,
921    pub input_rows: usize,
922    /// Equal-hash candidate pairs examined before domain-specific filtering.
923    pub candidate_rows: usize,
924    /// Records decoded from spill runs, including repartition and hot-key replay.
925    pub replay_rows: usize,
926    /// Successful spill-run partition splits.
927    pub repartitions: usize,
928    pub max_spill_bytes: u64,
929    pub max_spill_runs: usize,
930    pub spilled_bytes: u64,
931    pub spill_run_count: usize,
932    pub spilled_rows: usize,
933}
934
935#[derive(Debug, Clone, PartialEq, Eq, Default)]
936pub struct PipelineMemoryReport {
937    /// Query-owned runtime ledger budget across pipeline and blocking state.
938    pub query_memory_budget_bytes: usize,
939    /// Highest aggregate tracked resident bytes charged to the query ledger.
940    pub query_memory_peak_bytes: usize,
941    /// Tracked bytes still owned at the execution completion boundary.
942    pub query_memory_completion_bytes: usize,
943    /// Number of operator or transfer accounts created by the query.
944    pub query_memory_account_count: usize,
945    /// Sum of rows emitted at every physical operator boundary.
946    pub intermediate_rows: usize,
947    /// Sum of retained payload estimates emitted at every physical operator boundary.
948    pub intermediate_payload_bytes: usize,
949    pub peak_batch_rows: usize,
950    pub peak_batch_payload_bytes: usize,
951    /// Rows projected from source property references before owned output admission.
952    /// This counter does not claim source-to-host zero-copy delivery.
953    pub borrowed_node_projection_rows: usize,
954    /// Number of typed columnar batches evaluated by eligible pipeline fragments.
955    pub columnar_batches: usize,
956    /// Rows loaded into typed columns before selection.
957    pub columnar_input_rows: usize,
958    /// Rows retained by columnar selections.
959    pub columnar_selected_rows: usize,
960    /// Morsels consumed by columnar fragments.
961    pub morsel_count: usize,
962    /// Highest resource-admitted worker count across morsel pipelines.
963    pub morsel_max_admitted_workers: usize,
964    /// Highest worker count actually used by a morsel scheduler.
965    pub morsel_peak_active_workers: usize,
966    /// Highest number of completed morsel outputs awaiting or crossing the
967    /// coordinator's ordered consumer boundary.
968    pub morsel_peak_buffered_outputs: usize,
969    /// Highest estimated resident bytes held by those completed outputs.
970    pub morsel_peak_buffered_output_bytes: usize,
971    /// Highest number of out-of-order outputs held by the ordinal merger.
972    pub morsel_peak_reorder_entries: usize,
973    pub output_rows: usize,
974    pub output_payload_bytes: usize,
975    /// Resident memory before execution, when process sampling is supported.
976    pub start_resident_bytes: Option<u64>,
977    /// Process high-water resident memory before execution.
978    pub start_peak_resident_bytes: Option<u64>,
979    /// Resident memory after the output rows have been materialized.
980    pub steady_resident_bytes: Option<u64>,
981    /// Process high-water resident memory at completion.
982    pub peak_resident_bytes: Option<u64>,
983    /// Positive resident-memory delta retained at completion.
984    pub steady_resident_growth_bytes: Option<u64>,
985    /// Positive process high-water delta observed during execution.
986    pub lifetime_peak_resident_growth_bytes: Option<u64>,
987    /// Aggregate page-fault delta when the platform exposes it.
988    pub total_page_faults: Option<u64>,
989    /// Split page-fault deltas only on platforms that expose this distinction.
990    pub minor_page_faults: Option<u64>,
991    pub major_page_faults: Option<u64>,
992}
993
994/// The observed output cardinality for one physical operator.
995#[derive(Debug, Clone, PartialEq, Eq)]
996pub struct OperatorCardinalityProfile {
997    pub operator_id: PhysicalOperatorId,
998    pub operator: PhysicalPlanKind,
999    /// `None` means the operator was not invoked, while `Some(0)` means it ran
1000    /// and produced no output rows.
1001    pub actual_rows: Option<usize>,
1002}
1003
1004#[derive(Debug, Clone, PartialEq, Eq)]
1005pub struct ReadExecutionProfile<TScanPruningReport> {
1006    pub max_rows: Option<usize>,
1007    pub detection_row_cap: Option<usize>,
1008    pub row_limit_enforced_before_output: bool,
1009    pub operator_row_cap_enabled: bool,
1010    pub operator_cardinality_profiles: Vec<OperatorCardinalityProfile>,
1011    pub blocking_operator_kinds: Vec<String>,
1012    pub scan_pruning_reports: Vec<TScanPruningReport>,
1013    pub vector_execution_reports: Vec<crate::VectorExecutionReport>,
1014    pub graph_expansion_reports: Vec<crate::GraphExpansionExecutionReport>,
1015    pub blocking_operator_memory_reports: Vec<BlockingOperatorMemoryReport>,
1016    pub pipeline_memory_report: PipelineMemoryReport,
1017}
1018
1019impl<TScanPruningReport> ReadExecutionProfile<TScanPruningReport> {
1020    pub fn blocking_operator_count(&self) -> usize {
1021        self.blocking_operator_kinds.len()
1022    }
1023}
1024
1025#[derive(Debug, Clone, PartialEq, Eq)]
1026pub struct ProfiledQueryRows<TScanPruningReport> {
1027    pub rows: QueryRows,
1028    pub profile: ReadExecutionProfile<TScanPruningReport>,
1029}
1030
1031#[derive(Debug, Clone, PartialEq, Eq)]
1032pub struct ProfiledQueryStream<TScanPruningReport> {
1033    /// True when the physical plan emitted batches directly to the consumer.
1034    /// False means an unsupported operator still materialized bindings before
1035    /// the bounded consumer boundary.
1036    pub fully_streamed: bool,
1037    pub profile: ReadExecutionProfile<TScanPruningReport>,
1038}
1039
1040#[cfg(test)]
1041mod tests {
1042    use super::*;
1043    use crate::{
1044        BindingSchema, ColumnVector, ColumnarBatch, SlotDescriptor, SlotId, SlotType, Validity,
1045    };
1046    use hawdb_core::LogicalType;
1047    use std::sync::Arc;
1048
1049    #[test]
1050    fn segmented_results_preserve_access_ownership_and_flat_equality() {
1051        let expected = (0..1000)
1052            .map(|id| {
1053                Row::from([
1054                    ("id".into(), Value::Int(id)),
1055                    ("payload".into(), Value::String(format!("value-{id}\0"))),
1056                ])
1057            })
1058            .collect::<Vec<_>>();
1059        let mut builder = QueryRowsBuilder::new();
1060        for row in &expected {
1061            builder.push_named_row(row.clone()).unwrap();
1062        }
1063        let rows = builder.finish();
1064        assert_eq!(rows, QueryRows::from(expected.clone()));
1065        assert_eq!(
1066            rows.payload_bytes(),
1067            QueryRows::from(expected.clone()).payload_bytes()
1068        );
1069        for ordinal in [0, 255, 256, 511, 512, 999] {
1070            assert_eq!(rows.row(ordinal).unwrap(), expected[ordinal]);
1071            assert_eq!(rows[ordinal], expected[ordinal]);
1072        }
1073        let handle = rows[256].clone();
1074        let mut retained = rows.clone();
1075        retained.retain(|row| row["id"] == Value::Int(256));
1076        assert_eq!(retained.len(), 1);
1077        assert_eq!(retained.first().unwrap(), expected[256]);
1078        assert_eq!(rows.clone().into_rows(), expected);
1079        drop(rows);
1080        assert_eq!(handle["id"], Value::Int(256));
1081    }
1082
1083    #[test]
1084    fn segmented_results_bound_unknown_cardinality_capacity() {
1085        let mut builder = QueryRowsBuilder::new();
1086        for id in 0..100_000 {
1087            builder
1088                .push_named_row(Row::from([
1089                    ("id".into(), Value::Int(id)),
1090                    ("score".into(), Value::Float(id as f64)),
1091                    ("title".into(), Value::String("x".repeat(24))),
1092                ]))
1093                .unwrap();
1094        }
1095        let rows = builder.finish();
1096        let QueryStoredValues::Segmented(segments) = &rows.storage.values else {
1097            panic!("large results require segmented storage");
1098        };
1099        assert_eq!(
1100            segments.len(),
1101            100_000_usize.div_ceil(OWNED_RESULT_SEGMENT_ROWS)
1102        );
1103        let capacity = segments.iter().map(Vec::capacity).sum::<usize>();
1104        assert!(capacity >= 300_000);
1105        assert!(capacity <= 300_000 + 3 * OWNED_RESULT_SEGMENT_ROWS);
1106        assert!(segments
1107            .iter()
1108            .all(|segment| segment.capacity() <= 3 * OWNED_RESULT_SEGMENT_ROWS));
1109        assert_eq!(rows.get(99_999, "id"), Some(ValueRef::Int(99_999)));
1110    }
1111
1112    #[test]
1113    fn segmented_results_handle_empty_rows_and_boundary_rollback() {
1114        let mut empty = QueryRowsBuilder::new();
1115        for _ in 0..513 {
1116            empty.push_named_row(Row::new()).unwrap();
1117        }
1118        let empty = empty.finish();
1119        assert_eq!(empty.len(), 513);
1120        assert!(empty.row(512).unwrap().is_empty());
1121        let mut builder = QueryRowsBuilder::new();
1122        for id in 0..256 {
1123            builder
1124                .push_named_row(Row::from([("id".into(), Value::Int(id))]))
1125                .unwrap();
1126        }
1127        assert!(builder
1128            .push_named_row(Row::from([("wrong".into(), Value::Int(256))]))
1129            .is_err());
1130        assert!(builder
1131            .try_push_values([Err(HawDBError::Execution("failed".into()))])
1132            .is_err());
1133        assert!(builder.push_values([]).is_err());
1134        builder.push_values([Value::Int(256)]).unwrap();
1135        let rows = builder.finish();
1136        assert_eq!(rows.len(), 257);
1137        assert_eq!(rows.get(256, "id"), Some(ValueRef::Int(256)));
1138    }
1139
1140    #[test]
1141    fn counts_blocking_operator_kinds() {
1142        let profile = ReadExecutionProfile::<()> {
1143            max_rows: Some(10),
1144            detection_row_cap: Some(11),
1145            row_limit_enforced_before_output: true,
1146            operator_row_cap_enabled: true,
1147            operator_cardinality_profiles: Vec::new(),
1148            blocking_operator_kinds: vec!["sort".to_string(), "aggregate".to_string()],
1149            scan_pruning_reports: Vec::new(),
1150            vector_execution_reports: Vec::new(),
1151            graph_expansion_reports: Vec::new(),
1152            blocking_operator_memory_reports: Vec::new(),
1153            pipeline_memory_report: PipelineMemoryReport::default(),
1154        };
1155        assert_eq!(profile.blocking_operator_count(), 2);
1156    }
1157
1158    #[test]
1159    fn borrowed_map_row_materializes_only_on_request() {
1160        let row = Row::from([
1161            ("id".to_string(), Value::Int(7)),
1162            ("content".to_string(), Value::String("payload".into())),
1163        ]);
1164        let row_ref = RowRef::from(&row);
1165
1166        assert_eq!(row_ref.get("id"), Some(ValueRef::Int(7)));
1167        assert_eq!(row_ref.get("content").unwrap().as_str(), Some("payload"));
1168        assert_eq!(row_ref.to_owned_row(), row);
1169    }
1170
1171    #[test]
1172    fn schema_bearing_rows_keep_names_once_in_a_flat_value_buffer() {
1173        let schema = QuerySchema::try_new(["payload".to_string(), "id".to_string()]).unwrap();
1174        let mut builder = QueryRowsBuilder::with_schema(schema, 2);
1175        builder
1176            .push_values([Value::String("one".to_string()), Value::Int(1)])
1177            .unwrap();
1178        builder
1179            .push_values([Value::String("two".to_string()), Value::Int(2)])
1180            .unwrap();
1181        let rows = builder.finish();
1182
1183        assert_eq!(rows.schema().columns(), ["payload", "id"]);
1184        assert_eq!(rows.get(1, "payload").unwrap().as_str(), Some("two"));
1185        assert_eq!(rows.value(0, 1), Some(ValueRef::Int(1)));
1186        assert_eq!(rows.payload_bytes(), "payload".len() + "id".len() + 6 + 16);
1187        assert_eq!(rows.len(), 2);
1188        assert!(!rows.is_empty());
1189        assert_eq!(rows.storage.values.len(), 4);
1190        assert_eq!(rows[0]["payload"], Value::String("one".to_string()));
1191        assert_eq!(rows.iter().count(), 2);
1192        assert_eq!(rows.storage.values.len(), 4);
1193    }
1194
1195    #[test]
1196    fn schema_bearing_rows_reject_duplicate_columns_and_width_mismatch() {
1197        assert!(QuerySchema::try_new(["id".to_string(), "id".to_string()]).is_err());
1198        let schema = QuerySchema::try_new(["id".to_string()]).unwrap();
1199        let mut invalid = QueryRowsBuilder::with_schema(schema, 1);
1200        assert!(invalid.push_values([]).is_err());
1201
1202        let mut left = QueryRowsBuilder::with_schema(
1203            QuerySchema::try_new(["left".to_string(), "right".to_string()]).unwrap(),
1204            1,
1205        );
1206        left.push_values([Value::Int(1), Value::Int(2)]).unwrap();
1207        let left = left.finish();
1208        let mut right = QueryRowsBuilder::with_schema(
1209            QuerySchema::try_new(["right".to_string(), "left".to_string()]).unwrap(),
1210            1,
1211        );
1212        right.push_values([Value::Int(1), Value::Int(2)]).unwrap();
1213        let right = right.finish();
1214        assert_ne!(left, right);
1215    }
1216
1217    #[test]
1218    fn fallible_flat_row_append_rolls_back_partial_values() {
1219        let schema = QuerySchema::try_new(["left".to_string(), "right".to_string()]).unwrap();
1220        let mut builder = QueryRowsBuilder::with_schema(schema, 1);
1221        assert!(builder
1222            .try_push_values([
1223                Ok(Value::Int(1)),
1224                Err(HawDBError::Execution("projection failed".to_string())),
1225            ])
1226            .is_err());
1227        builder.push_values([Value::Int(2), Value::Int(3)]).unwrap();
1228        let rows = builder.finish();
1229
1230        assert_eq!(rows.len(), 1);
1231        assert_eq!(rows.value(0, 0), Some(ValueRef::Int(2)));
1232        assert_eq!(rows.value(0, 1), Some(ValueRef::Int(3)));
1233        assert_eq!(rows.storage.values.len(), 2);
1234    }
1235
1236    #[test]
1237    fn named_row_builder_binds_schema_once_and_moves_values_by_ordinal() {
1238        let mut builder = QueryRowsBuilder::with_row_capacity(2);
1239        builder
1240            .push_named_row(Row::from([
1241                ("id".to_string(), Value::Int(1)),
1242                ("payload".to_string(), Value::String("one".to_string())),
1243            ]))
1244            .unwrap();
1245        builder
1246            .push_named_row(Row::from([
1247                ("id".to_string(), Value::Int(2)),
1248                ("payload".to_string(), Value::String("two".to_string())),
1249            ]))
1250            .unwrap();
1251        let rows = builder.finish();
1252
1253        assert_eq!(rows.schema().columns(), ["id", "payload"]);
1254        assert_eq!(rows.value(1, 0), Some(ValueRef::Int(2)));
1255        assert_eq!(rows.get(0, "payload").unwrap().as_str(), Some("one"));
1256        assert_eq!(rows.storage.values.len(), 4);
1257    }
1258
1259    #[test]
1260    fn named_row_builder_rejects_schema_drift_without_partial_append() {
1261        let mut builder = QueryRowsBuilder::new();
1262        builder
1263            .push_named_row(Row::from([("id".to_string(), Value::Int(1))]))
1264            .unwrap();
1265        assert!(builder
1266            .push_named_row(Row::from([("name".to_string(), Value::Int(2))]))
1267            .is_err());
1268        let rows = builder.finish();
1269
1270        assert_eq!(rows.len(), 1);
1271        assert_eq!(rows.get(0, "id"), Some(ValueRef::Int(1)));
1272    }
1273
1274    #[test]
1275    fn borrowed_columnar_row_pulls_values_without_a_row_map() {
1276        let schema = Arc::new(
1277            BindingSchema::try_new(vec![
1278                SlotDescriptor {
1279                    id: SlotId(0),
1280                    name: "id".into(),
1281                    slot_type: SlotType::NodeId,
1282                },
1283                SlotDescriptor {
1284                    id: SlotId(1),
1285                    name: "content".into(),
1286                    slot_type: SlotType::logical(LogicalType::Text),
1287                },
1288            ])
1289            .unwrap(),
1290        );
1291        let batch = ColumnarBatch::try_new(
1292            schema,
1293            vec![
1294                Arc::new(ColumnVector::node_ids(vec![7])),
1295                Arc::new(ColumnVector::Utf8 {
1296                    values: vec!["payload".to_string()].into(),
1297                    validity: Validity::all(1),
1298                }),
1299            ],
1300        )
1301        .unwrap();
1302        let row_ref = RowRef::from(batch.rows().next().unwrap());
1303
1304        assert_eq!(row_ref.get("id"), Some(ValueRef::Int(7)));
1305        assert_eq!(row_ref.get("content").unwrap().as_str(), Some("payload"));
1306        assert_eq!(
1307            row_ref.to_owned_row(),
1308            Row::from([
1309                ("content".into(), Value::String("payload".into())),
1310                ("id".into(), Value::Int(7)),
1311            ])
1312        );
1313    }
1314}
hawdb_executor/profile.rs

hawdb_executor/
profile.rs
