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// LimitIterator - Applies LIMIT and OFFSET
445// ============================================================================
446
447/// Iterator that applies LIMIT and OFFSET constraints.
448///
449/// This iterator skips the first `offset` rows and yields at most `limit` rows.
450/// It provides early termination - once the limit is reached, no more rows
451/// are requested from the input.
452pub struct LimitIterator<I: RowIterator> {
453    input: I,
454    limit: Option<u64>,
455    offset: u64,
456    /// Number of rows skipped so far (for OFFSET).
457    skipped: u64,
458    /// Number of rows yielded so far (for LIMIT).
459    yielded: u64,
460}
461
462impl<I: RowIterator> LimitIterator<I> {
463    /// Creates a new limit iterator with the given LIMIT and OFFSET.
464    pub fn new(input: I, limit: Option<u64>, offset: Option<u64>) -> Self {
465        Self {
466            input,
467            limit,
468            offset: offset.unwrap_or(0),
469            skipped: 0,
470            yielded: 0,
471        }
472    }
473}
474
475impl<I: RowIterator> RowIterator for LimitIterator<I> {
476    fn next_row(&mut self) -> Option<Result<Row>> {
477        // Check if limit already reached
478        if let Some(limit) = self.limit
479            && self.yielded >= limit
480        {
481            return None;
482        }
483
484        loop {
485            match self.input.next_row()? {
486                Ok(row) => {
487                    // Skip rows for OFFSET
488                    if self.skipped < self.offset {
489                        self.skipped += 1;
490                        continue;
491                    }
492
493                    // Check limit again after skipping
494                    if let Some(limit) = self.limit
495                        && self.yielded >= limit
496                    {
497                        return None;
498                    }
499
500                    self.yielded += 1;
501                    return Some(Ok(row));
502                }
503                Err(e) => return Some(Err(e)),
504            }
505        }
506    }
507
508    fn schema(&self) -> &[ColumnMetadata] {
509        self.input.schema()
510    }
511}
512
513// ============================================================================
514// VecIterator - Wraps a Vec<Row> for testing and compatibility
515// ============================================================================
516
517/// Iterator that wraps a `Vec<Row>` for testing and compatibility.
518///
519/// This is useful for converting materialized results back into an iterator
520/// or for testing iterator-based code with fixed data.
521pub struct VecIterator {
522    rows: std::vec::IntoIter<Row>,
523    schema: Vec<ColumnMetadata>,
524}
525
526impl VecIterator {
527    /// Creates a new vec iterator from rows and schema.
528    pub fn new(rows: Vec<Row>, schema: Vec<ColumnMetadata>) -> Self {
529        Self {
530            rows: rows.into_iter(),
531            schema,
532        }
533    }
534}
535
536impl RowIterator for VecIterator {
537    fn next_row(&mut self) -> Option<Result<Row>> {
538        self.rows.next().map(Ok)
539    }
540
541    fn schema(&self) -> &[ColumnMetadata] {
542        &self.schema
543    }
544}
545
546// ============================================================================
547// Tests
548// ============================================================================
549
550#[cfg(test)]
551mod tests {
552    use super::*;
553    use crate::Span;
554    use crate::planner::types::ResolvedType;
555
556    fn sample_schema() -> Vec<ColumnMetadata> {
557        vec![
558            ColumnMetadata::new("id", ResolvedType::Integer),
559            ColumnMetadata::new("name", ResolvedType::Text),
560        ]
561    }
562
563    fn sample_rows() -> Vec<Row> {
564        vec![
565            Row::new(
566                1,
567                vec![SqlValue::Integer(1), SqlValue::Text("alice".into())],
568            ),
569            Row::new(2, vec![SqlValue::Integer(2), SqlValue::Text("bob".into())]),
570            Row::new(
571                3,
572                vec![SqlValue::Integer(3), SqlValue::Text("carol".into())],
573            ),
574            Row::new(4, vec![SqlValue::Integer(4), SqlValue::Text("dave".into())]),
575            Row::new(5, vec![SqlValue::Integer(5), SqlValue::Text("eve".into())]),
576        ]
577    }
578
579    #[test]
580    fn vec_iterator_returns_all_rows() {
581        let rows = sample_rows();
582        let expected_len = rows.len();
583        let mut iter = VecIterator::new(rows, sample_schema());
584
585        let mut count = 0;
586        while let Some(Ok(_)) = iter.next_row() {
587            count += 1;
588        }
589        assert_eq!(count, expected_len);
590    }
591
592    #[test]
593    fn filter_iterator_filters_rows() {
594        use crate::ast::expr::BinaryOp;
595        use crate::planner::typed_expr::{TypedExpr, TypedExprKind};
596
597        let rows = sample_rows();
598        let schema = sample_schema();
599        let input = VecIterator::new(rows, schema);
600
601        // Filter: id > 2
602        let predicate = TypedExpr {
603            kind: TypedExprKind::BinaryOp {
604                left: Box::new(TypedExpr {
605                    kind: TypedExprKind::ColumnRef {
606                        table: "test".into(),
607                        column: "id".into(),
608                        column_index: 0,
609                    },
610                    resolved_type: ResolvedType::Integer,
611                    span: Span::default(),
612                }),
613                op: BinaryOp::Gt,
614                right: Box::new(TypedExpr::literal(
615                    crate::ast::expr::Literal::Number("2".into()),
616                    ResolvedType::Integer,
617                    Span::default(),
618                )),
619            },
620            resolved_type: ResolvedType::Boolean,
621            span: Span::default(),
622        };
623
624        let mut filter = FilterIterator::new(input, predicate);
625
626        let mut results = Vec::new();
627        while let Some(Ok(row)) = filter.next_row() {
628            results.push(row);
629        }
630
631        assert_eq!(results.len(), 3);
632        assert_eq!(results[0].row_id, 3);
633        assert_eq!(results[1].row_id, 4);
634        assert_eq!(results[2].row_id, 5);
635    }
636
637    #[test]
638    fn limit_iterator_limits_rows() {
639        let rows = sample_rows();
640        let schema = sample_schema();
641        let input = VecIterator::new(rows, schema);
642
643        let mut limit = LimitIterator::new(input, Some(2), None);
644
645        let mut results = Vec::new();
646        while let Some(Ok(row)) = limit.next_row() {
647            results.push(row);
648        }
649
650        assert_eq!(results.len(), 2);
651        assert_eq!(results[0].row_id, 1);
652        assert_eq!(results[1].row_id, 2);
653    }
654
655    #[test]
656    fn limit_iterator_applies_offset() {
657        let rows = sample_rows();
658        let schema = sample_schema();
659        let input = VecIterator::new(rows, schema);
660
661        let mut limit = LimitIterator::new(input, Some(2), Some(2));
662
663        let mut results = Vec::new();
664        while let Some(Ok(row)) = limit.next_row() {
665            results.push(row);
666        }
667
668        assert_eq!(results.len(), 2);
669        assert_eq!(results[0].row_id, 3);
670        assert_eq!(results[1].row_id, 4);
671    }
672
673    #[test]
674    fn limit_iterator_offset_only() {
675        let rows = sample_rows();
676        let schema = sample_schema();
677        let input = VecIterator::new(rows, schema);
678
679        let mut limit = LimitIterator::new(input, None, Some(3));
680
681        let mut results = Vec::new();
682        while let Some(Ok(row)) = limit.next_row() {
683            results.push(row);
684        }
685
686        assert_eq!(results.len(), 2);
687        assert_eq!(results[0].row_id, 4);
688        assert_eq!(results[1].row_id, 5);
689    }
690
691    #[test]
692    fn sort_iterator_sorts_rows() {
693        use crate::planner::typed_expr::{SortExpr, TypedExpr, TypedExprKind};
694
695        let rows = vec![
696            Row::new(
697                1,
698                vec![SqlValue::Integer(3), SqlValue::Text("carol".into())],
699            ),
700            Row::new(
701                2,
702                vec![SqlValue::Integer(1), SqlValue::Text("alice".into())],
703            ),
704            Row::new(3, vec![SqlValue::Integer(2), SqlValue::Text("bob".into())]),
705        ];
706        let schema = sample_schema();
707        let input = VecIterator::new(rows, schema);
708
709        // Sort by id ASC
710        let order_by = vec![SortExpr {
711            expr: TypedExpr {
712                kind: TypedExprKind::ColumnRef {
713                    table: "test".into(),
714                    column: "id".into(),
715                    column_index: 0,
716                },
717                resolved_type: ResolvedType::Integer,
718                span: Span::default(),
719            },
720            asc: true,
721            nulls_first: false,
722        }];
723
724        let mut sort = SortIterator::new(input, &order_by).unwrap();
725
726        let mut results = Vec::new();
727        while let Some(Ok(row)) = sort.next_row() {
728            results.push(row);
729        }
730
731        assert_eq!(results.len(), 3);
732        assert_eq!(results[0].values[0], SqlValue::Integer(1));
733        assert_eq!(results[1].values[0], SqlValue::Integer(2));
734        assert_eq!(results[2].values[0], SqlValue::Integer(3));
735    }
736
737    #[test]
738    fn sort_iterator_memory_limit_exceeded_returns_resource_exhausted() {
739        use crate::executor::memory::SpillPolicy;
740
741        let rows = sample_rows();
742        let schema = sample_schema();
743        let input = VecIterator::new(rows, schema);
744        let policy = MemoryPolicy::new(Some(1), SpillPolicy::FailFast);
745
746        let err = match SortIterator::new_with_policy(input, &[], Some(policy)) {
747            Ok(_) => panic!("expected sort iterator memory limit error"),
748            Err(err) => err,
749        };
750
751        assert!(matches!(err, ExecutorError::ResourceExhausted { .. }));
752    }
753
754    #[test]
755    fn sort_iterator_sorts_descending() {
756        use crate::planner::typed_expr::{SortExpr, TypedExpr, TypedExprKind};
757
758        let rows = vec![
759            Row::new(
760                1,
761                vec![SqlValue::Integer(1), SqlValue::Text("alice".into())],
762            ),
763            Row::new(
764                2,
765                vec![SqlValue::Integer(3), SqlValue::Text("carol".into())],
766            ),
767            Row::new(3, vec![SqlValue::Integer(2), SqlValue::Text("bob".into())]),
768        ];
769        let schema = sample_schema();
770        let input = VecIterator::new(rows, schema);
771
772        // Sort by id DESC
773        let order_by = vec![SortExpr {
774            expr: TypedExpr {
775                kind: TypedExprKind::ColumnRef {
776                    table: "test".into(),
777                    column: "id".into(),
778                    column_index: 0,
779                },
780                resolved_type: ResolvedType::Integer,
781                span: Span::default(),
782            },
783            asc: false,
784            nulls_first: false,
785        }];
786
787        let mut sort = SortIterator::new(input, &order_by).unwrap();
788
789        let mut results = Vec::new();
790        while let Some(Ok(row)) = sort.next_row() {
791            results.push(row);
792        }
793
794        assert_eq!(results.len(), 3);
795        assert_eq!(results[0].values[0], SqlValue::Integer(3));
796        assert_eq!(results[1].values[0], SqlValue::Integer(2));
797        assert_eq!(results[2].values[0], SqlValue::Integer(1));
798    }
799
800    #[test]
801    fn composed_pipeline_filter_then_limit() {
802        use crate::ast::expr::BinaryOp;
803        use crate::planner::typed_expr::{TypedExpr, TypedExprKind};
804
805        let rows = sample_rows();
806        let schema = sample_schema();
807        let input = VecIterator::new(rows, schema);
808
809        // Filter: id > 1
810        let predicate = TypedExpr {
811            kind: TypedExprKind::BinaryOp {
812                left: Box::new(TypedExpr {
813                    kind: TypedExprKind::ColumnRef {
814                        table: "test".into(),
815                        column: "id".into(),
816                        column_index: 0,
817                    },
818                    resolved_type: ResolvedType::Integer,
819                    span: Span::default(),
820                }),
821                op: BinaryOp::Gt,
822                right: Box::new(TypedExpr::literal(
823                    crate::ast::expr::Literal::Number("1".into()),
824                    ResolvedType::Integer,
825                    Span::default(),
826                )),
827            },
828            resolved_type: ResolvedType::Boolean,
829            span: Span::default(),
830        };
831
832        let filtered = FilterIterator::new(input, predicate);
833        let mut limited = LimitIterator::new(filtered, Some(2), None);
834
835        let mut results = Vec::new();
836        while let Some(Ok(row)) = limited.next_row() {
837            results.push(row);
838        }
839
840        // Should get rows 2, 3 (id > 1, then limit 2)
841        assert_eq!(results.len(), 2);
842        assert_eq!(results[0].row_id, 2);
843        assert_eq!(results[1].row_id, 3);
844    }
845}
alopex_sql/executor/query/iterator.rs

alopex_sql/executor/query/
iterator.rs
