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
346fn compare_key_values(a: &[SqlValue], b: &[SqlValue], order_by: &[SortExpr]) -> Ordering {
347    for (i, sort_expr) in order_by.iter().enumerate() {
348        let left = &a[i];
349        let right = &b[i];
350        let cmp = compare_single(left, right, sort_expr.asc, sort_expr.nulls_first);
351        if cmp != Ordering::Equal {
352            return cmp;
353        }
354    }
355    Ordering::Equal
356}
357
358/// Compare two SqlValues according to sort direction and NULL ordering.
359fn compare_single(left: &SqlValue, right: &SqlValue, asc: bool, nulls_first: bool) -> Ordering {
360    match (left, right) {
361        (SqlValue::Null, SqlValue::Null) => Ordering::Equal,
362        (SqlValue::Null, _) => {
363            if nulls_first {
364                Ordering::Less
365            } else {
366                Ordering::Greater
367            }
368        }
369        (_, SqlValue::Null) => {
370            if nulls_first {
371                Ordering::Greater
372            } else {
373                Ordering::Less
374            }
375        }
376        _ => match left.partial_cmp(right).unwrap_or(Ordering::Equal) {
377            Ordering::Equal => Ordering::Equal,
378            ord if asc => ord,
379            ord => ord.reverse(),
380        },
381    }
382}
383
384// ============================================================================
385// LimitIterator - Applies LIMIT and OFFSET
386// ============================================================================
387
388/// Iterator that applies LIMIT and OFFSET constraints.
389///
390/// This iterator skips the first `offset` rows and yields at most `limit` rows.
391/// It provides early termination - once the limit is reached, no more rows
392/// are requested from the input.
393pub struct LimitIterator<I: RowIterator> {
394    input: I,
395    limit: Option<u64>,
396    offset: u64,
397    /// Number of rows skipped so far (for OFFSET).
398    skipped: u64,
399    /// Number of rows yielded so far (for LIMIT).
400    yielded: u64,
401}
402
403impl<I: RowIterator> LimitIterator<I> {
404    /// Creates a new limit iterator with the given LIMIT and OFFSET.
405    pub fn new(input: I, limit: Option<u64>, offset: Option<u64>) -> Self {
406        Self {
407            input,
408            limit,
409            offset: offset.unwrap_or(0),
410            skipped: 0,
411            yielded: 0,
412        }
413    }
414}
415
416impl<I: RowIterator> RowIterator for LimitIterator<I> {
417    fn next_row(&mut self) -> Option<Result<Row>> {
418        // Check if limit already reached
419        if let Some(limit) = self.limit
420            && self.yielded >= limit
421        {
422            return None;
423        }
424
425        loop {
426            match self.input.next_row()? {
427                Ok(row) => {
428                    // Skip rows for OFFSET
429                    if self.skipped < self.offset {
430                        self.skipped += 1;
431                        continue;
432                    }
433
434                    // Check limit again after skipping
435                    if let Some(limit) = self.limit
436                        && self.yielded >= limit
437                    {
438                        return None;
439                    }
440
441                    self.yielded += 1;
442                    return Some(Ok(row));
443                }
444                Err(e) => return Some(Err(e)),
445            }
446        }
447    }
448
449    fn schema(&self) -> &[ColumnMetadata] {
450        self.input.schema()
451    }
452}
453
454// ============================================================================
455// VecIterator - Wraps a Vec<Row> for testing and compatibility
456// ============================================================================
457
458/// Iterator that wraps a `Vec<Row>` for testing and compatibility.
459///
460/// This is useful for converting materialized results back into an iterator
461/// or for testing iterator-based code with fixed data.
462pub struct VecIterator {
463    rows: std::vec::IntoIter<Row>,
464    schema: Vec<ColumnMetadata>,
465}
466
467impl VecIterator {
468    /// Creates a new vec iterator from rows and schema.
469    pub fn new(rows: Vec<Row>, schema: Vec<ColumnMetadata>) -> Self {
470        Self {
471            rows: rows.into_iter(),
472            schema,
473        }
474    }
475}
476
477impl RowIterator for VecIterator {
478    fn next_row(&mut self) -> Option<Result<Row>> {
479        self.rows.next().map(Ok)
480    }
481
482    fn schema(&self) -> &[ColumnMetadata] {
483        &self.schema
484    }
485}
486
487// ============================================================================
488// Tests
489// ============================================================================
490
491#[cfg(test)]
492mod tests {
493    use super::*;
494    use crate::Span;
495    use crate::planner::types::ResolvedType;
496
497    fn sample_schema() -> Vec<ColumnMetadata> {
498        vec![
499            ColumnMetadata::new("id", ResolvedType::Integer),
500            ColumnMetadata::new("name", ResolvedType::Text),
501        ]
502    }
503
504    fn sample_rows() -> Vec<Row> {
505        vec![
506            Row::new(
507                1,
508                vec![SqlValue::Integer(1), SqlValue::Text("alice".into())],
509            ),
510            Row::new(2, vec![SqlValue::Integer(2), SqlValue::Text("bob".into())]),
511            Row::new(
512                3,
513                vec![SqlValue::Integer(3), SqlValue::Text("carol".into())],
514            ),
515            Row::new(4, vec![SqlValue::Integer(4), SqlValue::Text("dave".into())]),
516            Row::new(5, vec![SqlValue::Integer(5), SqlValue::Text("eve".into())]),
517        ]
518    }
519
520    #[test]
521    fn vec_iterator_returns_all_rows() {
522        let rows = sample_rows();
523        let expected_len = rows.len();
524        let mut iter = VecIterator::new(rows, sample_schema());
525
526        let mut count = 0;
527        while let Some(Ok(_)) = iter.next_row() {
528            count += 1;
529        }
530        assert_eq!(count, expected_len);
531    }
532
533    #[test]
534    fn filter_iterator_filters_rows() {
535        use crate::ast::expr::BinaryOp;
536        use crate::planner::typed_expr::{TypedExpr, TypedExprKind};
537
538        let rows = sample_rows();
539        let schema = sample_schema();
540        let input = VecIterator::new(rows, schema);
541
542        // Filter: id > 2
543        let predicate = TypedExpr {
544            kind: TypedExprKind::BinaryOp {
545                left: Box::new(TypedExpr {
546                    kind: TypedExprKind::ColumnRef {
547                        table: "test".into(),
548                        column: "id".into(),
549                        column_index: 0,
550                    },
551                    resolved_type: ResolvedType::Integer,
552                    span: Span::default(),
553                }),
554                op: BinaryOp::Gt,
555                right: Box::new(TypedExpr::literal(
556                    crate::ast::expr::Literal::Number("2".into()),
557                    ResolvedType::Integer,
558                    Span::default(),
559                )),
560            },
561            resolved_type: ResolvedType::Boolean,
562            span: Span::default(),
563        };
564
565        let mut filter = FilterIterator::new(input, predicate);
566
567        let mut results = Vec::new();
568        while let Some(Ok(row)) = filter.next_row() {
569            results.push(row);
570        }
571
572        assert_eq!(results.len(), 3);
573        assert_eq!(results[0].row_id, 3);
574        assert_eq!(results[1].row_id, 4);
575        assert_eq!(results[2].row_id, 5);
576    }
577
578    #[test]
579    fn limit_iterator_limits_rows() {
580        let rows = sample_rows();
581        let schema = sample_schema();
582        let input = VecIterator::new(rows, schema);
583
584        let mut limit = LimitIterator::new(input, Some(2), None);
585
586        let mut results = Vec::new();
587        while let Some(Ok(row)) = limit.next_row() {
588            results.push(row);
589        }
590
591        assert_eq!(results.len(), 2);
592        assert_eq!(results[0].row_id, 1);
593        assert_eq!(results[1].row_id, 2);
594    }
595
596    #[test]
597    fn limit_iterator_applies_offset() {
598        let rows = sample_rows();
599        let schema = sample_schema();
600        let input = VecIterator::new(rows, schema);
601
602        let mut limit = LimitIterator::new(input, Some(2), Some(2));
603
604        let mut results = Vec::new();
605        while let Some(Ok(row)) = limit.next_row() {
606            results.push(row);
607        }
608
609        assert_eq!(results.len(), 2);
610        assert_eq!(results[0].row_id, 3);
611        assert_eq!(results[1].row_id, 4);
612    }
613
614    #[test]
615    fn limit_iterator_offset_only() {
616        let rows = sample_rows();
617        let schema = sample_schema();
618        let input = VecIterator::new(rows, schema);
619
620        let mut limit = LimitIterator::new(input, None, Some(3));
621
622        let mut results = Vec::new();
623        while let Some(Ok(row)) = limit.next_row() {
624            results.push(row);
625        }
626
627        assert_eq!(results.len(), 2);
628        assert_eq!(results[0].row_id, 4);
629        assert_eq!(results[1].row_id, 5);
630    }
631
632    #[test]
633    fn sort_iterator_sorts_rows() {
634        use crate::planner::typed_expr::{SortExpr, TypedExpr, TypedExprKind};
635
636        let rows = vec![
637            Row::new(
638                1,
639                vec![SqlValue::Integer(3), SqlValue::Text("carol".into())],
640            ),
641            Row::new(
642                2,
643                vec![SqlValue::Integer(1), SqlValue::Text("alice".into())],
644            ),
645            Row::new(3, vec![SqlValue::Integer(2), SqlValue::Text("bob".into())]),
646        ];
647        let schema = sample_schema();
648        let input = VecIterator::new(rows, schema);
649
650        // Sort by id ASC
651        let order_by = vec![SortExpr {
652            expr: TypedExpr {
653                kind: TypedExprKind::ColumnRef {
654                    table: "test".into(),
655                    column: "id".into(),
656                    column_index: 0,
657                },
658                resolved_type: ResolvedType::Integer,
659                span: Span::default(),
660            },
661            asc: true,
662            nulls_first: false,
663        }];
664
665        let mut sort = SortIterator::new(input, &order_by).unwrap();
666
667        let mut results = Vec::new();
668        while let Some(Ok(row)) = sort.next_row() {
669            results.push(row);
670        }
671
672        assert_eq!(results.len(), 3);
673        assert_eq!(results[0].values[0], SqlValue::Integer(1));
674        assert_eq!(results[1].values[0], SqlValue::Integer(2));
675        assert_eq!(results[2].values[0], SqlValue::Integer(3));
676    }
677
678    #[test]
679    fn sort_iterator_memory_limit_exceeded_returns_resource_exhausted() {
680        use crate::executor::memory::SpillPolicy;
681
682        let rows = sample_rows();
683        let schema = sample_schema();
684        let input = VecIterator::new(rows, schema);
685        let policy = MemoryPolicy::new(Some(1), SpillPolicy::FailFast);
686
687        let err = match SortIterator::new_with_policy(input, &[], Some(policy)) {
688            Ok(_) => panic!("expected sort iterator memory limit error"),
689            Err(err) => err,
690        };
691
692        assert!(matches!(err, ExecutorError::ResourceExhausted { .. }));
693    }
694
695    #[test]
696    fn sort_iterator_sorts_descending() {
697        use crate::planner::typed_expr::{SortExpr, TypedExpr, TypedExprKind};
698
699        let rows = vec![
700            Row::new(
701                1,
702                vec![SqlValue::Integer(1), SqlValue::Text("alice".into())],
703            ),
704            Row::new(
705                2,
706                vec![SqlValue::Integer(3), SqlValue::Text("carol".into())],
707            ),
708            Row::new(3, vec![SqlValue::Integer(2), SqlValue::Text("bob".into())]),
709        ];
710        let schema = sample_schema();
711        let input = VecIterator::new(rows, schema);
712
713        // Sort by id DESC
714        let order_by = vec![SortExpr {
715            expr: TypedExpr {
716                kind: TypedExprKind::ColumnRef {
717                    table: "test".into(),
718                    column: "id".into(),
719                    column_index: 0,
720                },
721                resolved_type: ResolvedType::Integer,
722                span: Span::default(),
723            },
724            asc: false,
725            nulls_first: false,
726        }];
727
728        let mut sort = SortIterator::new(input, &order_by).unwrap();
729
730        let mut results = Vec::new();
731        while let Some(Ok(row)) = sort.next_row() {
732            results.push(row);
733        }
734
735        assert_eq!(results.len(), 3);
736        assert_eq!(results[0].values[0], SqlValue::Integer(3));
737        assert_eq!(results[1].values[0], SqlValue::Integer(2));
738        assert_eq!(results[2].values[0], SqlValue::Integer(1));
739    }
740
741    #[test]
742    fn composed_pipeline_filter_then_limit() {
743        use crate::ast::expr::BinaryOp;
744        use crate::planner::typed_expr::{TypedExpr, TypedExprKind};
745
746        let rows = sample_rows();
747        let schema = sample_schema();
748        let input = VecIterator::new(rows, schema);
749
750        // Filter: id > 1
751        let predicate = TypedExpr {
752            kind: TypedExprKind::BinaryOp {
753                left: Box::new(TypedExpr {
754                    kind: TypedExprKind::ColumnRef {
755                        table: "test".into(),
756                        column: "id".into(),
757                        column_index: 0,
758                    },
759                    resolved_type: ResolvedType::Integer,
760                    span: Span::default(),
761                }),
762                op: BinaryOp::Gt,
763                right: Box::new(TypedExpr::literal(
764                    crate::ast::expr::Literal::Number("1".into()),
765                    ResolvedType::Integer,
766                    Span::default(),
767                )),
768            },
769            resolved_type: ResolvedType::Boolean,
770            span: Span::default(),
771        };
772
773        let filtered = FilterIterator::new(input, predicate);
774        let mut limited = LimitIterator::new(filtered, Some(2), None);
775
776        let mut results = Vec::new();
777        while let Some(Ok(row)) = limited.next_row() {
778            results.push(row);
779        }
780
781        // Should get rows 2, 3 (id > 1, then limit 2)
782        assert_eq!(results.len(), 2);
783        assert_eq!(results[0].row_id, 2);
784        assert_eq!(results[1].row_id, 3);
785    }
786}
alopex_sql/executor/query/iterator.rs

alopex_sql/executor/query/
iterator.rs
