alopex_sql/executor/query/
iterator.rs

1//! Iterator-based query execution pipeline.
2//!
3//! This module provides an iterator-based execution model for SQL queries,
4//! enabling streaming execution and reduced memory usage for large datasets.
5//!
6//! # Architecture
7//!
8//! The execution pipeline is built from composable iterators:
9//! - [`ScanIterator`]: Reads rows from storage
10//! - [`FilterIterator`]: Filters rows based on predicates
11//! - [`SortIterator`]: Sorts rows (requires materialization)
12//! - [`LimitIterator`]: Applies LIMIT/OFFSET constraints
13//!
14//! Each iterator implements the [`RowIterator`] trait, allowing them to be
15//! composed into a pipeline that processes rows one at a time.
16
17use std::cmp::Ordering;
18use std::marker::PhantomData;
19use std::path::PathBuf;
20
21use crate::catalog::{ColumnMetadata, TableMetadata};
22use crate::executor::evaluator::EvalContext;
23use crate::executor::memory::{MemoryPolicy, MemoryTracker, map_core_memory_error};
24use crate::executor::{ExecutorError, Result, Row};
25use crate::planner::typed_expr::{SortExpr, TypedExpr};
26use crate::storage::{RowCodec, SqlValue, TableScanIterator};
27use alopex_core::Error as CoreError;
28use alopex_core::sql::spill::{
29    SpillMergeIterator, spill_io_error as core_spill_io_error, spill_run as core_spill_run,
30};
31
32/// A trait for row-producing iterators in the query execution pipeline.
33///
34/// This trait abstracts over different types of iterators (scan, filter, sort, etc.)
35/// allowing them to be composed into execution pipelines.
36pub trait RowIterator {
37    /// Advances the iterator and returns the next row, or `None` if exhausted.
38    ///
39    /// # Errors
40    ///
41    /// Returns an error if the underlying operation fails (e.g., storage errors,
42    /// evaluation errors).
43    fn next_row(&mut self) -> Option<Result<Row>>;
44
45    /// Returns the schema of rows produced by this iterator.
46    fn schema(&self) -> &[ColumnMetadata];
47}
48
49// Implement RowIterator for Box<dyn RowIterator> to allow dynamic dispatch.
50impl RowIterator for Box<dyn RowIterator + '_> {
51    fn next_row(&mut self) -> Option<Result<Row>> {
52        (**self).next_row()
53    }
54
55    fn schema(&self) -> &[ColumnMetadata] {
56        (**self).schema()
57    }
58}
59
60// ============================================================================
61// ScanIterator - Reads rows from storage for true streaming execution
62// ============================================================================
63
64/// Iterator that reads rows from table storage.
65///
66/// This is the leaf node in the iterator tree, providing rows from the
67/// underlying storage layer. Used for FR-7 streaming output compliance.
68pub struct ScanIterator<'a> {
69    inner: TableScanIterator<'a>,
70    schema: Vec<ColumnMetadata>,
71}
72
73impl<'a> ScanIterator<'a> {
74    /// Creates a new scan iterator from a table scan iterator and metadata.
75    pub fn new(inner: TableScanIterator<'a>, table_meta: &TableMetadata) -> Self {
76        Self {
77            inner,
78            schema: table_meta.columns.clone(),
79        }
80    }
81}
82
83impl RowIterator for ScanIterator<'_> {
84    fn next_row(&mut self) -> Option<Result<Row>> {
85        self.inner.next().map(|result| {
86            result
87                .map(|(row_id, values)| Row::new(row_id, values))
88                .map_err(ExecutorError::from)
89        })
90    }
91
92    fn schema(&self) -> &[ColumnMetadata] {
93        &self.schema
94    }
95}
96
97// ============================================================================
98// FilterIterator - Filters rows based on a predicate
99// ============================================================================
100
101/// Iterator that filters rows based on a predicate expression.
102///
103/// Only rows where the predicate evaluates to `true` are yielded.
104/// Rows where the predicate evaluates to `false` or `NULL` are skipped.
105pub struct FilterIterator<I: RowIterator> {
106    input: I,
107    predicate: TypedExpr,
108}
109
110impl<I: RowIterator> FilterIterator<I> {
111    /// Creates a new filter iterator with the given input and predicate.
112    pub fn new(input: I, predicate: TypedExpr) -> Self {
113        Self { input, predicate }
114    }
115}
116
117impl<I: RowIterator> RowIterator for FilterIterator<I> {
118    fn next_row(&mut self) -> Option<Result<Row>> {
119        loop {
120            match self.input.next_row()? {
121                Ok(row) => {
122                    let ctx = EvalContext::new(&row.values);
123                    match crate::executor::evaluator::evaluate(&self.predicate, &ctx) {
124                        Ok(SqlValue::Boolean(true)) => return Some(Ok(row)),
125                        Ok(_) => continue, // false or null - skip this row
126                        Err(e) => return Some(Err(e)),
127                    }
128                }
129                Err(e) => return Some(Err(e)),
130            }
131        }
132    }
133
134    fn schema(&self) -> &[ColumnMetadata] {
135        self.input.schema()
136    }
137}
138
139// ============================================================================
140// SortIterator - Sorts rows (materializes all input)
141// ============================================================================
142
143/// Iterator that sorts rows according to ORDER BY expressions.
144///
145/// **Note**: Sorting requires materializing all input rows into memory.
146/// This iterator collects all rows from its input, sorts them, and then
147/// yields them one at a time.
148pub struct SortIterator<I: RowIterator> {
149    output: SortOutput,
150    /// Schema from input.
151    schema: Vec<ColumnMetadata>,
152    /// Marker for input iterator type.
153    _marker: PhantomData<I>,
154}
155
156enum SortOutput {
157    InMemory(std::vec::IntoIter<Row>),
158    External(ExternalSortState),
159}
160
161impl<I: RowIterator> SortIterator<I> {
162    /// Creates a new sort iterator.
163    ///
164    /// This constructor immediately materializes all input rows and sorts them.
165    ///
166    /// # Errors
167    ///
168    /// Returns an error if reading from input fails or if sort key evaluation fails.
169    pub fn new(input: I, order_by: &[SortExpr]) -> Result<Self> {
170        Self::new_with_policy(input, order_by, None)
171    }
172
173    /// Creates a new sort iterator with an optional memory policy.
174    pub fn new_with_policy(
175        mut input: I,
176        order_by: &[SortExpr],
177        policy: Option<MemoryPolicy>,
178    ) -> Result<Self> {
179        let schema = input.schema().to_vec();
180        let mut tracker = policy.clone().map(MemoryTracker::new);
181
182        if order_by.is_empty() {
183            let mut rows = Vec::new();
184            while let Some(result) = input.next_row() {
185                rows.push(result?);
186                if let Some(tracker) = &mut tracker {
187                    let row = rows.last().expect("row just pushed");
188                    tracker
189                        .add_row(&row.values)
190                        .map_err(map_core_memory_error)?;
191                }
192            }
193            return Ok(Self {
194                output: SortOutput::InMemory(rows.into_iter()),
195                schema,
196                _marker: PhantomData,
197            });
198        }
199
200        let allow_spill = policy
201            .as_ref()
202            .and_then(|policy| policy.spill_directory())
203            .is_some();
204        let mut runs: Vec<PathBuf> = Vec::new();
205        let mut keyed: Vec<(Row, Vec<SqlValue>)> = Vec::new();
206
207        while let Some(result) = input.next_row() {
208            let row = result?;
209            let mut keys = Vec::with_capacity(order_by.len());
210            for expr in order_by {
211                let ctx = EvalContext::new(&row.values);
212                keys.push(crate::executor::evaluator::evaluate(&expr.expr, &ctx)?);
213            }
214            if let Some(tracker) = &mut tracker {
215                tracker
216                    .add_row(&row.values)
217                    .map_err(map_core_memory_error)?;
218                tracker.add_values(&keys).map_err(map_core_memory_error)?;
219            }
220            keyed.push((row, keys));
221
222            if allow_spill && tracker.as_ref().map(|t| t.over_limit()).unwrap_or(false) {
223                let policy = policy
224                    .as_ref()
225                    .ok_or_else(|| ExecutorError::InvalidOperation {
226                        operation: "sort spill".into(),
227                        reason: "spill policy missing".into(),
228                    })?;
229                let path = spill_run(&mut keyed, order_by, policy)?;
230                runs.push(path);
231                if let Some(tracker) = &mut tracker {
232                    tracker.reset();
233                }
234            }
235        }
236
237        if runs.is_empty() {
238            keyed.sort_by(|a, b| compare_key_values(&a.1, &b.1, order_by));
239            let sorted: Vec<Row> = keyed.into_iter().map(|(row, _)| row).collect();
240            return Ok(Self {
241                output: SortOutput::InMemory(sorted.into_iter()),
242                schema,
243                _marker: PhantomData,
244            });
245        }
246
247        if !keyed.is_empty() {
248            let policy = policy
249                .as_ref()
250                .ok_or_else(|| ExecutorError::InvalidOperation {
251                    operation: "sort spill".into(),
252                    reason: "spill policy missing".into(),
253                })?;
254            let path = spill_run(&mut keyed, order_by, policy)?;
255            runs.push(path);
256        }
257
258        let external = ExternalSortState::new(order_by.to_vec(), runs)?;
259
260        Ok(Self {
261            output: SortOutput::External(external),
262            schema,
263            _marker: PhantomData,
264        })
265    }
266}
267
268impl<I: RowIterator> RowIterator for SortIterator<I> {
269    fn next_row(&mut self) -> Option<Result<Row>> {
270        match &mut self.output {
271            SortOutput::InMemory(iter) => iter.next().map(Ok),
272            SortOutput::External(state) => state.next_row(),
273        }
274    }
275
276    fn schema(&self) -> &[ColumnMetadata] {
277        &self.schema
278    }
279}
280
281fn spill_run(
282    entries: &mut Vec<(Row, Vec<SqlValue>)>,
283    order_by: &[SortExpr],
284    policy: &MemoryPolicy,
285) -> Result<PathBuf> {
286    core_spill_run(
287        entries,
288        policy,
289        "sort-run",
290        |left, right| compare_key_values(left, right, order_by),
291        |row| row.row_id,
292        |keys| RowCodec::encode(keys),
293        |row| RowCodec::encode(&row.values),
294    )
295    .map_err(map_core_spill_error)
296}
297
298struct ExternalSortState {
299    iter: SpillMergeIterator<Row, Vec<SqlValue>>,
300}
301
302impl ExternalSortState {
303    fn new(order_by: Vec<SortExpr>, runs: Vec<PathBuf>) -> Result<Self> {
304        let iter = SpillMergeIterator::new(
305            runs,
306            move |left, right| compare_key_values(left, right, &order_by),
307            |row: &Row| row.row_id,
308            decode_spill_values,
309            |row_id, bytes| decode_spill_values(bytes).map(|values| Row::new(row_id, values)),
310        )
311        .map_err(map_core_spill_error)?;
312
313        Ok(Self { iter })
314    }
315
316    fn next_row(&mut self) -> Option<Result<Row>> {
317        self.iter
318            .next_item()
319            .map(|result| result.map_err(map_core_spill_error))
320    }
321}
322
323fn decode_spill_values(bytes: &[u8]) -> alopex_core::Result<Vec<SqlValue>> {
324    RowCodec::decode(bytes).map_err(|err| CoreError::SpillFailed {
325        reason: format!("sort spill: {err}"),
326    })
327}
328
329fn map_core_spill_error(err: CoreError) -> ExecutorError {
330    match err {
331        CoreError::SpillFailed { reason } => ExecutorError::InvalidOperation {
332            operation: "sort spill".into(),
333            reason,
334        },
335        CoreError::Io(err) => ExecutorError::InvalidOperation {
336            operation: "sort spill".into(),
337            reason: core_spill_io_error("sort spill", err).to_string(),
338        },
339        CoreError::MemoryLimitExceeded { limit, requested } => ExecutorError::ResourceExhausted {
340            message: format!("query memory limit exceeded: {requested} bytes (limit {limit})"),
341        },
342        other => ExecutorError::Core(other),
343    }
344}
345
346pub(super) fn compare_key_values(
347    a: &[SqlValue],
348    b: &[SqlValue],
349    order_by: &[SortExpr],
350) -> Ordering {
351    for (i, sort_expr) in order_by.iter().enumerate() {
352        let left = &a[i];
353        let right = &b[i];
354        let cmp = compare_single(left, right, sort_expr.asc, sort_expr.nulls_first);
355        if cmp != Ordering::Equal {
356            return cmp;
357        }
358    }
359    Ordering::Equal
360}
361
362/// Compare two SqlValues according to sort direction and NULL ordering.
363pub(super) fn compare_single(
364    left: &SqlValue,
365    right: &SqlValue,
366    asc: bool,
367    nulls_first: bool,
368) -> Ordering {
369    match (left, right) {
370        (SqlValue::Null, SqlValue::Null) => Ordering::Equal,
371        (SqlValue::Null, _) => {
372            if nulls_first {
373                Ordering::Less
374            } else {
375                Ordering::Greater
376            }
377        }
378        (_, SqlValue::Null) => {
379            if nulls_first {
380                Ordering::Greater
381            } else {
382                Ordering::Less
383            }
384        }
385        _ => match compare_non_null_values(left, right) {
386            Ordering::Equal => Ordering::Equal,
387            ord if asc => ord,
388            ord => ord.reverse(),
389        },
390    }
391}
392
393fn compare_non_null_values(left: &SqlValue, right: &SqlValue) -> Ordering {
394    if let Some(ordering) = left.partial_cmp(right) {
395        return ordering;
396    }
397    match (left, right) {
398        (SqlValue::Float(left), SqlValue::Float(right)) => compare_f32(*left, *right),
399        (SqlValue::Double(left), SqlValue::Double(right)) => compare_f64(*left, *right),
400        (SqlValue::Vector(left), SqlValue::Vector(right)) => compare_vectors(left, right),
401        // A typed sort expression normally makes this unreachable, but a total
402        // type-tag fallback prevents inconsistent inputs from becoming peers.
403        _ => left.type_tag().cmp(&right.type_tag()),
404    }
405}
406
407fn compare_f32(left: f32, right: f32) -> Ordering {
408    if left.partial_cmp(&right) == Some(Ordering::Equal) || (left.is_nan() && right.is_nan()) {
409        Ordering::Equal
410    } else if left.is_nan() {
411        Ordering::Greater
412    } else if right.is_nan() {
413        Ordering::Less
414    } else {
415        left.partial_cmp(&right)
416            .expect("non-NaN f32 values are totally ordered")
417    }
418}
419
420fn compare_f64(left: f64, right: f64) -> Ordering {
421    if left.partial_cmp(&right) == Some(Ordering::Equal) || (left.is_nan() && right.is_nan()) {
422        Ordering::Equal
423    } else if left.is_nan() {
424        Ordering::Greater
425    } else if right.is_nan() {
426        Ordering::Less
427    } else {
428        left.partial_cmp(&right)
429            .expect("non-NaN f64 values are totally ordered")
430    }
431}
432
433fn compare_vectors(left: &[f32], right: &[f32]) -> Ordering {
434    for (left, right) in left.iter().zip(right) {
435        let ordering = compare_f32(*left, *right);
436        if ordering != Ordering::Equal {
437            return ordering;
438        }
439    }
440    left.len().cmp(&right.len())
441}
442
443// ============================================================================
444// DistinctOnIterator - SELECT DISTINCT ON deduplication (issue #150)
445// ============================================================================
446
447/// Iterator that implements `SELECT DISTINCT ON (expr, ...)`.
448///
449/// The input is first sorted by the complete effective sort specification the
450/// planner synthesized (ON keys, user ORDER BY tail, and an all-column
451/// tie-breaker; see docs/sql-distinct-on.md D2-D4). Sorting reuses
452/// [`SortIterator`], so spill runs are merged under the same full comparator
453/// and determinism survives external sorting. The iterator then emits only
454/// the first row of each group whose leading `key_count` sort keys compare
455/// equal; NULL keys compare equal to NULL (D5).
456pub struct DistinctOnIterator<I: RowIterator> {
457    input: SortIterator<I>,
458    key_exprs: Vec<SortExpr>,
459    last_keys: Option<Vec<SqlValue>>,
460}
461
462impl<I: RowIterator> DistinctOnIterator<I> {
463    /// Creates a new DISTINCT ON iterator.
464    ///
465    /// `order_by` is the complete effective sort specification whose leading
466    /// `key_count` entries form the distinctness key.
467    ///
468    /// # Errors
469    ///
470    /// Returns an error if sorting the input fails.
471    pub fn new(
472        input: I,
473        order_by: &[SortExpr],
474        key_count: usize,
475        policy: Option<MemoryPolicy>,
476    ) -> Result<Self> {
477        debug_assert!((1..=order_by.len()).contains(&key_count));
478        let sorted = SortIterator::new_with_policy(input, order_by, policy)?;
479        Ok(Self {
480            input: sorted,
481            key_exprs: order_by[..key_count.min(order_by.len())].to_vec(),
482            last_keys: None,
483        })
484    }
485}
486
487impl<I: RowIterator> RowIterator for DistinctOnIterator<I> {
488    fn next_row(&mut self) -> Option<Result<Row>> {
489        loop {
490            let row = match self.input.next_row()? {
491                Ok(row) => row,
492                Err(err) => return Some(Err(err)),
493            };
494            let mut keys = Vec::with_capacity(self.key_exprs.len());
495            for sort in &self.key_exprs {
496                let ctx = EvalContext::new(&row.values);
497                match crate::executor::evaluator::evaluate(&sort.expr, &ctx) {
498                    Ok(value) => keys.push(value),
499                    Err(err) => return Some(Err(err)),
500                }
501            }
502            let starts_new_group = match &self.last_keys {
503                None => true,
504                Some(previous) => previous.iter().zip(&keys).any(|(left, right)| {
505                    // Direction and NULL placement are irrelevant to key
506                    // equality: equal values stay equal under ASC/DESC and
507                    // (NULL, NULL) compares Equal (D5).
508                    compare_single(left, right, true, false) != Ordering::Equal
509                }),
510            };
511            if starts_new_group {
512                self.last_keys = Some(keys);
513                return Some(Ok(row));
514            }
515        }
516    }
517
518    fn schema(&self) -> &[ColumnMetadata] {
519        self.input.schema()
520    }
521}
522
523// ============================================================================
524// LimitIterator - Applies LIMIT and OFFSET
525// ============================================================================
526
527/// Iterator that applies LIMIT and OFFSET constraints.
528///
529/// This iterator skips the first `offset` rows and yields at most `limit` rows.
530/// It provides early termination - once the limit is reached, no more rows
531/// are requested from the input.
532///
533/// With `tie_keys` (FETCH ... WITH TIES, issue #152) the iterator keeps
534/// yielding rows past the limit while their ORDER BY sort key compares equal
535/// to the final counted row's key. Rows discarded by OFFSET never revive,
536/// even when they are peers of the boundary row.
537pub struct LimitIterator<I: RowIterator> {
538    input: I,
539    limit: Option<u64>,
540    offset: u64,
541    /// Number of rows skipped so far (for OFFSET).
542    skipped: u64,
543    /// Number of rows yielded so far (for LIMIT).
544    yielded: u64,
545    /// Sort keys copied from the Sort node beneath the Limit (WITH TIES).
546    tie_keys: Vec<SortExpr>,
547    /// Sort-key values of the most recently yielded row (WITH TIES).
548    boundary: Option<Vec<SqlValue>>,
549    /// Set once a non-peer row ends the tie scan.
550    done: bool,
551}
552
553impl<I: RowIterator> LimitIterator<I> {
554    /// Creates a new limit iterator with the given LIMIT and OFFSET.
555    pub fn new(input: I, limit: Option<u64>, offset: Option<u64>) -> Self {
556        Self::with_ties(input, limit, offset, Vec::new())
557    }
558
559    /// Creates a limit iterator that keeps boundary peers (FETCH ... WITH TIES).
560    ///
561    /// `tie_keys` must be the sort expressions of the input ordering; an empty
562    /// vector degrades to plain LIMIT/OFFSET semantics.
563    pub fn with_ties(
564        input: I,
565        limit: Option<u64>,
566        offset: Option<u64>,
567        tie_keys: Vec<SortExpr>,
568    ) -> Self {
569        Self {
570            input,
571            limit,
572            offset: offset.unwrap_or(0),
573            skipped: 0,
574            yielded: 0,
575            tie_keys,
576            boundary: None,
577            done: false,
578        }
579    }
580
581    fn row_keys(&self, row: &Row) -> Result<Vec<SqlValue>> {
582        let ctx = EvalContext::new(&row.values);
583        self.tie_keys
584            .iter()
585            .map(|sort| crate::executor::evaluator::evaluate(&sort.expr, &ctx))
586            .collect()
587    }
588}
589
590impl<I: RowIterator> RowIterator for LimitIterator<I> {
591    fn next_row(&mut self) -> Option<Result<Row>> {
592        if self.done {
593            return None;
594        }
595
596        // Check if limit already reached
597        if let Some(limit) = self.limit
598            && self.yielded >= limit
599        {
600            if self.tie_keys.is_empty() {
601                return None;
602            }
603            // WITH TIES: keep yielding while the sort key equals the final
604            // counted row's key. LIMIT 0 (no boundary) yields nothing.
605            let Some(boundary) = self.boundary.clone() else {
606                self.done = true;
607                return None;
608            };
609            return match self.input.next_row() {
610                Some(Ok(row)) => {
611                    let keys = match self.row_keys(&row) {
612                        Ok(keys) => keys,
613                        Err(error) => return Some(Err(error)),
614                    };
615                    if compare_key_values(&keys, &boundary, &self.tie_keys) == Ordering::Equal {
616                        Some(Ok(row))
617                    } else {
618                        self.done = true;
619                        None
620                    }
621                }
622                Some(Err(error)) => Some(Err(error)),
623                None => {
624                    self.done = true;
625                    None
626                }
627            };
628        }
629
630        loop {
631            match self.input.next_row()? {
632                Ok(row) => {
633                    // Skip rows for OFFSET
634                    if self.skipped < self.offset {
635                        self.skipped += 1;
636                        continue;
637                    }
638
639                    // Check limit again after skipping
640                    if let Some(limit) = self.limit
641                        && self.yielded >= limit
642                    {
643                        return None;
644                    }
645
646                    if !self.tie_keys.is_empty() {
647                        match self.row_keys(&row) {
648                            Ok(keys) => self.boundary = Some(keys),
649                            Err(error) => return Some(Err(error)),
650                        }
651                    }
652                    self.yielded += 1;
653                    return Some(Ok(row));
654                }
655                Err(e) => return Some(Err(e)),
656            }
657        }
658    }
659
660    fn schema(&self) -> &[ColumnMetadata] {
661        self.input.schema()
662    }
663}
664
665// ============================================================================
666// VecIterator - Wraps a Vec<Row> for testing and compatibility
667// ============================================================================
668
669/// Iterator that wraps a `Vec<Row>` for testing and compatibility.
670///
671/// This is useful for converting materialized results back into an iterator
672/// or for testing iterator-based code with fixed data.
673pub struct VecIterator {
674    rows: std::vec::IntoIter<Row>,
675    schema: Vec<ColumnMetadata>,
676}
677
678impl VecIterator {
679    /// Creates a new vec iterator from rows and schema.
680    pub fn new(rows: Vec<Row>, schema: Vec<ColumnMetadata>) -> Self {
681        Self {
682            rows: rows.into_iter(),
683            schema,
684        }
685    }
686}
687
688impl RowIterator for VecIterator {
689    fn next_row(&mut self) -> Option<Result<Row>> {
690        self.rows.next().map(Ok)
691    }
692
693    fn schema(&self) -> &[ColumnMetadata] {
694        &self.schema
695    }
696}
697
698// ============================================================================
699// ValuesIterator - Lazily evaluates inline VALUES rows
700// ============================================================================
701
702/// Streaming iterator for a type-checked VALUES relation.
703pub struct ValuesIterator {
704    rows: std::vec::IntoIter<Vec<TypedExpr>>,
705    schema: Vec<ColumnMetadata>,
706    context_values: Vec<SqlValue>,
707    next_row_id: u64,
708}
709
710impl ValuesIterator {
711    /// Creates a VALUES iterator without materializing every evaluated row.
712    ///
713    /// Correlated VALUES subqueries evaluate column references against a copy
714    /// of the current outer row. Top-level VALUES queries use an empty context.
715    pub fn new(
716        rows: Vec<Vec<TypedExpr>>,
717        schema: Vec<ColumnMetadata>,
718        outer: Option<&Row>,
719    ) -> Self {
720        Self {
721            rows: rows.into_iter(),
722            schema,
723            context_values: outer.map(|row| row.values.clone()).unwrap_or_default(),
724            next_row_id: 0,
725        }
726    }
727}
728
729impl RowIterator for ValuesIterator {
730    fn next_row(&mut self) -> Option<Result<Row>> {
731        let expressions = self.rows.next()?;
732        let context = EvalContext::new(&self.context_values);
733        let values = expressions
734            .iter()
735            .map(|expression| crate::executor::evaluator::evaluate(expression, &context))
736            .collect::<Result<Vec<_>>>();
737        let row_id = self.next_row_id;
738        self.next_row_id = self.next_row_id.saturating_add(1);
739        Some(values.map(|values| Row::new(row_id, values)))
740    }
741
742    fn schema(&self) -> &[ColumnMetadata] {
743        &self.schema
744    }
745}
746
747// ============================================================================
748// Tests
749// ============================================================================
750
751#[cfg(test)]
752mod tests {
753    use super::*;
754    use crate::Span;
755    use crate::planner::types::ResolvedType;
756
757    fn sample_schema() -> Vec<ColumnMetadata> {
758        vec![
759            ColumnMetadata::new("id", ResolvedType::Integer),
760            ColumnMetadata::new("name", ResolvedType::Text),
761        ]
762    }
763
764    fn sample_rows() -> Vec<Row> {
765        vec![
766            Row::new(
767                1,
768                vec![SqlValue::Integer(1), SqlValue::Text("alice".into())],
769            ),
770            Row::new(2, vec![SqlValue::Integer(2), SqlValue::Text("bob".into())]),
771            Row::new(
772                3,
773                vec![SqlValue::Integer(3), SqlValue::Text("carol".into())],
774            ),
775            Row::new(4, vec![SqlValue::Integer(4), SqlValue::Text("dave".into())]),
776            Row::new(5, vec![SqlValue::Integer(5), SqlValue::Text("eve".into())]),
777        ]
778    }
779
780    #[test]
781    fn vec_iterator_returns_all_rows() {
782        let rows = sample_rows();
783        let expected_len = rows.len();
784        let mut iter = VecIterator::new(rows, sample_schema());
785
786        let mut count = 0;
787        while let Some(Ok(_)) = iter.next_row() {
788            count += 1;
789        }
790        assert_eq!(count, expected_len);
791    }
792
793    #[test]
794    fn filter_iterator_filters_rows() {
795        use crate::ast::expr::BinaryOp;
796        use crate::planner::typed_expr::{TypedExpr, TypedExprKind};
797
798        let rows = sample_rows();
799        let schema = sample_schema();
800        let input = VecIterator::new(rows, schema);
801
802        // Filter: id > 2
803        let predicate = TypedExpr {
804            kind: TypedExprKind::BinaryOp {
805                left: Box::new(TypedExpr {
806                    kind: TypedExprKind::ColumnRef {
807                        table: "test".into(),
808                        column: "id".into(),
809                        column_index: 0,
810                    },
811                    resolved_type: ResolvedType::Integer,
812                    span: Span::default(),
813                }),
814                op: BinaryOp::Gt,
815                right: Box::new(TypedExpr::literal(
816                    crate::ast::expr::Literal::Number("2".into()),
817                    ResolvedType::Integer,
818                    Span::default(),
819                )),
820            },
821            resolved_type: ResolvedType::Boolean,
822            span: Span::default(),
823        };
824
825        let mut filter = FilterIterator::new(input, predicate);
826
827        let mut results = Vec::new();
828        while let Some(Ok(row)) = filter.next_row() {
829            results.push(row);
830        }
831
832        assert_eq!(results.len(), 3);
833        assert_eq!(results[0].row_id, 3);
834        assert_eq!(results[1].row_id, 4);
835        assert_eq!(results[2].row_id, 5);
836    }
837
838    #[test]
839    fn limit_iterator_limits_rows() {
840        let rows = sample_rows();
841        let schema = sample_schema();
842        let input = VecIterator::new(rows, schema);
843
844        let mut limit = LimitIterator::new(input, Some(2), None);
845
846        let mut results = Vec::new();
847        while let Some(Ok(row)) = limit.next_row() {
848            results.push(row);
849        }
850
851        assert_eq!(results.len(), 2);
852        assert_eq!(results[0].row_id, 1);
853        assert_eq!(results[1].row_id, 2);
854    }
855
856    #[test]
857    fn limit_iterator_applies_offset() {
858        let rows = sample_rows();
859        let schema = sample_schema();
860        let input = VecIterator::new(rows, schema);
861
862        let mut limit = LimitIterator::new(input, Some(2), Some(2));
863
864        let mut results = Vec::new();
865        while let Some(Ok(row)) = limit.next_row() {
866            results.push(row);
867        }
868
869        assert_eq!(results.len(), 2);
870        assert_eq!(results[0].row_id, 3);
871        assert_eq!(results[1].row_id, 4);
872    }
873
874    #[test]
875    fn limit_iterator_offset_only() {
876        let rows = sample_rows();
877        let schema = sample_schema();
878        let input = VecIterator::new(rows, schema);
879
880        let mut limit = LimitIterator::new(input, None, Some(3));
881
882        let mut results = Vec::new();
883        while let Some(Ok(row)) = limit.next_row() {
884            results.push(row);
885        }
886
887        assert_eq!(results.len(), 2);
888        assert_eq!(results[0].row_id, 4);
889        assert_eq!(results[1].row_id, 5);
890    }
891
892    fn score_sort_key() -> SortExpr {
893        use crate::planner::typed_expr::{TypedExpr, TypedExprKind};
894        SortExpr {
895            expr: TypedExpr {
896                kind: TypedExprKind::ColumnRef {
897                    table: "test".into(),
898                    column: "id".into(),
899                    column_index: 0,
900                },
901                resolved_type: ResolvedType::Integer,
902                span: Span::default(),
903            },
904            asc: true,
905            nulls_first: false,
906        }
907    }
908
909    #[test]
910    fn limit_iterator_with_ties_keeps_boundary_peers() {
911        // Keys: 10, 20, 20, 20, 30 — limit 2 keeps 10, 20 plus both 20 peers.
912        let rows = vec![
913            Row::new(1, vec![SqlValue::Integer(10)]),
914            Row::new(2, vec![SqlValue::Integer(20)]),
915            Row::new(3, vec![SqlValue::Integer(20)]),
916            Row::new(4, vec![SqlValue::Integer(20)]),
917            Row::new(5, vec![SqlValue::Integer(30)]),
918        ];
919        let schema = vec![ColumnMetadata::new("id", ResolvedType::Integer)];
920        let input = VecIterator::new(rows, schema);
921        let mut limit = LimitIterator::with_ties(input, Some(2), None, vec![score_sort_key()]);
922
923        let mut results = Vec::new();
924        while let Some(Ok(row)) = limit.next_row() {
925            results.push(row.row_id);
926        }
927        assert_eq!(results, vec![1, 2, 3, 4]);
928    }
929
930    #[test]
931    fn limit_iterator_with_ties_stops_on_the_first_non_peer() {
932        let rows = vec![
933            Row::new(1, vec![SqlValue::Integer(10)]),
934            Row::new(2, vec![SqlValue::Integer(20)]),
935            Row::new(3, vec![SqlValue::Integer(30)]),
936        ];
937        let schema = vec![ColumnMetadata::new("id", ResolvedType::Integer)];
938        let input = VecIterator::new(rows, schema);
939        let mut limit = LimitIterator::with_ties(input, Some(2), None, vec![score_sort_key()]);
940
941        let mut results = Vec::new();
942        while let Some(Ok(row)) = limit.next_row() {
943            results.push(row.row_id);
944        }
945        assert_eq!(results, vec![1, 2]);
946        // The iterator is fused after the tie scan ends.
947        assert!(limit.next_row().is_none());
948    }
949
950    #[test]
951    fn limit_iterator_with_ties_null_keys_are_peers() {
952        let rows = vec![
953            Row::new(1, vec![SqlValue::Null]),
954            Row::new(2, vec![SqlValue::Null]),
955            Row::new(3, vec![SqlValue::Integer(1)]),
956        ];
957        let schema = vec![ColumnMetadata::new("id", ResolvedType::Integer)];
958        let input = VecIterator::new(rows, schema);
959        let mut limit = LimitIterator::with_ties(input, Some(1), None, vec![score_sort_key()]);
960
961        let mut results = Vec::new();
962        while let Some(Ok(row)) = limit.next_row() {
963            results.push(row.row_id);
964        }
965        assert_eq!(results, vec![1, 2]);
966    }
967
968    #[test]
969    fn limit_iterator_with_ties_zero_limit_yields_nothing() {
970        let rows = sample_rows();
971        let schema = sample_schema();
972        let input = VecIterator::new(rows, schema);
973        let mut limit = LimitIterator::with_ties(input, Some(0), None, vec![score_sort_key()]);
974        assert!(limit.next_row().is_none());
975    }
976
977    #[test]
978    fn sort_iterator_sorts_rows() {
979        use crate::planner::typed_expr::{SortExpr, TypedExpr, TypedExprKind};
980
981        let rows = vec![
982            Row::new(
983                1,
984                vec![SqlValue::Integer(3), SqlValue::Text("carol".into())],
985            ),
986            Row::new(
987                2,
988                vec![SqlValue::Integer(1), SqlValue::Text("alice".into())],
989            ),
990            Row::new(3, vec![SqlValue::Integer(2), SqlValue::Text("bob".into())]),
991        ];
992        let schema = sample_schema();
993        let input = VecIterator::new(rows, schema);
994
995        // Sort by id ASC
996        let order_by = vec![SortExpr {
997            expr: TypedExpr {
998                kind: TypedExprKind::ColumnRef {
999                    table: "test".into(),
1000                    column: "id".into(),
1001                    column_index: 0,
1002                },
1003                resolved_type: ResolvedType::Integer,
1004                span: Span::default(),
1005            },
1006            asc: true,
1007            nulls_first: false,
1008        }];
1009
1010        let mut sort = SortIterator::new(input, &order_by).unwrap();
1011
1012        let mut results = Vec::new();
1013        while let Some(Ok(row)) = sort.next_row() {
1014            results.push(row);
1015        }
1016
1017        assert_eq!(results.len(), 3);
1018        assert_eq!(results[0].values[0], SqlValue::Integer(1));
1019        assert_eq!(results[1].values[0], SqlValue::Integer(2));
1020        assert_eq!(results[2].values[0], SqlValue::Integer(3));
1021    }
1022
1023    #[test]
1024    fn sort_iterator_memory_limit_exceeded_returns_resource_exhausted() {
1025        use crate::executor::memory::SpillPolicy;
1026
1027        let rows = sample_rows();
1028        let schema = sample_schema();
1029        let input = VecIterator::new(rows, schema);
1030        let policy = MemoryPolicy::new(Some(1), SpillPolicy::FailFast);
1031
1032        let err = match SortIterator::new_with_policy(input, &[], Some(policy)) {
1033            Ok(_) => panic!("expected sort iterator memory limit error"),
1034            Err(err) => err,
1035        };
1036
1037        assert!(matches!(err, ExecutorError::ResourceExhausted { .. }));
1038    }
1039
1040    #[test]
1041    fn sort_iterator_sorts_descending() {
1042        use crate::planner::typed_expr::{SortExpr, TypedExpr, TypedExprKind};
1043
1044        let rows = vec![
1045            Row::new(
1046                1,
1047                vec![SqlValue::Integer(1), SqlValue::Text("alice".into())],
1048            ),
1049            Row::new(
1050                2,
1051                vec![SqlValue::Integer(3), SqlValue::Text("carol".into())],
1052            ),
1053            Row::new(3, vec![SqlValue::Integer(2), SqlValue::Text("bob".into())]),
1054        ];
1055        let schema = sample_schema();
1056        let input = VecIterator::new(rows, schema);
1057
1058        // Sort by id DESC
1059        let order_by = vec![SortExpr {
1060            expr: TypedExpr {
1061                kind: TypedExprKind::ColumnRef {
1062                    table: "test".into(),
1063                    column: "id".into(),
1064                    column_index: 0,
1065                },
1066                resolved_type: ResolvedType::Integer,
1067                span: Span::default(),
1068            },
1069            asc: false,
1070            nulls_first: false,
1071        }];
1072
1073        let mut sort = SortIterator::new(input, &order_by).unwrap();
1074
1075        let mut results = Vec::new();
1076        while let Some(Ok(row)) = sort.next_row() {
1077            results.push(row);
1078        }
1079
1080        assert_eq!(results.len(), 3);
1081        assert_eq!(results[0].values[0], SqlValue::Integer(3));
1082        assert_eq!(results[1].values[0], SqlValue::Integer(2));
1083        assert_eq!(results[2].values[0], SqlValue::Integer(1));
1084    }
1085
1086    #[test]
1087    fn composed_pipeline_filter_then_limit() {
1088        use crate::ast::expr::BinaryOp;
1089        use crate::planner::typed_expr::{TypedExpr, TypedExprKind};
1090
1091        let rows = sample_rows();
1092        let schema = sample_schema();
1093        let input = VecIterator::new(rows, schema);
1094
1095        // Filter: id > 1
1096        let predicate = TypedExpr {
1097            kind: TypedExprKind::BinaryOp {
1098                left: Box::new(TypedExpr {
1099                    kind: TypedExprKind::ColumnRef {
1100                        table: "test".into(),
1101                        column: "id".into(),
1102                        column_index: 0,
1103                    },
1104                    resolved_type: ResolvedType::Integer,
1105                    span: Span::default(),
1106                }),
1107                op: BinaryOp::Gt,
1108                right: Box::new(TypedExpr::literal(
1109                    crate::ast::expr::Literal::Number("1".into()),
1110                    ResolvedType::Integer,
1111                    Span::default(),
1112                )),
1113            },
1114            resolved_type: ResolvedType::Boolean,
1115            span: Span::default(),
1116        };
1117
1118        let filtered = FilterIterator::new(input, predicate);
1119        let mut limited = LimitIterator::new(filtered, Some(2), None);
1120
1121        let mut results = Vec::new();
1122        while let Some(Ok(row)) = limited.next_row() {
1123            results.push(row);
1124        }
1125
1126        // Should get rows 2, 3 (id > 1, then limit 2)
1127        assert_eq!(results.len(), 2);
1128        assert_eq!(results[0].row_id, 2);
1129        assert_eq!(results[1].row_id, 3);
1130    }
1131}
alopex_sql/executor/query/iterator.rs

alopex_sql/executor/query/
iterator.rs
