alopex_sql/executor/query/
mod.rs

1use alopex_core::kv::KVStore;
2
3use crate::ast::LITERAL_TABLE;
4use crate::catalog::{Catalog, StorageType};
5use crate::executor::evaluator::EvalContext;
6use crate::executor::memory::MemoryPolicy;
7use crate::executor::{ExecutionResult, ExecutorError, QueryResult, QueryRowIterator, Result};
8use crate::planner::logical_plan::LogicalPlan;
9use crate::planner::typed_expr::{Projection, SortExpr};
10use crate::storage::{SqlTxn, SqlValue};
11
12use super::{ColumnInfo, Row};
13
14pub mod aggregate;
15pub mod columnar_scan;
16pub mod iterator;
17pub mod join;
18mod knn;
19mod project;
20mod scan;
21pub mod subquery;
22
23pub use columnar_scan::{ColumnarScanIterator, create_columnar_scan_iterator};
24pub use iterator::{FilterIterator, LimitIterator, RowIterator, ScanIterator, SortIterator};
25pub use scan::create_scan_iterator;
26
27/// Execute a SELECT logical plan and return a query result.
28///
29/// This function uses an iterator-based execution model that processes rows
30/// through a pipeline of operators. This approach:
31/// - Enables early termination for LIMIT queries
32/// - Provides streaming execution after the initial scan
33/// - Allows composable query operators
34///
35/// Note: The Scan stage reads all matching rows into memory, but subsequent
36/// operators (Filter, Sort, Limit) process rows through an iterator pipeline.
37/// Sort operations additionally require materializing all input rows.
38pub fn execute_query<'txn, S: KVStore + 'txn, C: Catalog + ?Sized, T: SqlTxn<'txn, S>>(
39    txn: &mut T,
40    catalog: &C,
41    plan: LogicalPlan,
42) -> Result<ExecutionResult> {
43    execute_query_with_policy(txn, catalog, plan, None)
44}
45
46pub fn execute_query_with_policy<
47    'txn,
48    S: KVStore + 'txn,
49    C: Catalog + ?Sized,
50    T: SqlTxn<'txn, S>,
51>(
52    txn: &mut T,
53    catalog: &C,
54    plan: LogicalPlan,
55    memory: Option<&MemoryPolicy>,
56) -> Result<ExecutionResult> {
57    if let Some((pattern, projection, filter)) = knn::extract_knn_context(&plan) {
58        return knn::execute_knn_query(txn, catalog, &pattern, &projection, filter.as_ref());
59    }
60
61    let result = execute_query_result_with_outer_and_policy(txn, catalog, plan, None, memory)?;
62    Ok(ExecutionResult::Query(result))
63}
64
65pub(crate) fn execute_query_result_with_outer<
66    'txn,
67    S: KVStore + 'txn,
68    C: Catalog + ?Sized,
69    T: SqlTxn<'txn, S>,
70>(
71    txn: &mut T,
72    catalog: &C,
73    plan: LogicalPlan,
74    outer: Option<&Row>,
75) -> Result<QueryResult> {
76    execute_query_result_with_outer_and_policy(txn, catalog, plan, outer, None)
77}
78
79fn execute_query_result_with_outer_and_policy<
80    'txn,
81    S: KVStore + 'txn,
82    C: Catalog + ?Sized,
83    T: SqlTxn<'txn, S>,
84>(
85    txn: &mut T,
86    catalog: &C,
87    plan: LogicalPlan,
88    outer: Option<&Row>,
89    memory: Option<&MemoryPolicy>,
90) -> Result<QueryResult> {
91    let (mut iter, projection, schema) =
92        build_iterator_pipeline_with_outer(txn, catalog, plan, memory, outer)?;
93    let mut rows = Vec::new();
94    while let Some(result) = iter.next_row() {
95        rows.push(result?);
96    }
97    execute_project_with_subqueries(txn, catalog, rows, &projection, &schema, outer)
98}
99
100/// Execute a SELECT logical plan and return a streaming query result.
101///
102/// This function returns a `QueryRowIterator` that yields rows one at a time,
103/// enabling true streaming output without materializing all rows upfront.
104///
105/// # FR-7 Streaming Output
106///
107/// This function implements the FR-7 requirement for streaming output.
108/// Rows are yielded through an iterator interface, and projection is applied
109/// on-the-fly as each row is consumed.
110///
111/// # Note
112///
113/// KNN queries currently fall back to the non-streaming path as they require
114/// specialized handling.
115pub fn execute_query_streaming<'txn, S: KVStore + 'txn, C: Catalog + ?Sized, T: SqlTxn<'txn, S>>(
116    txn: &mut T,
117    catalog: &C,
118    plan: LogicalPlan,
119) -> Result<QueryRowIterator<'static>> {
120    execute_query_streaming_with_policy(txn, catalog, plan, None)
121}
122
123pub fn execute_query_streaming_with_policy<
124    'txn,
125    S: KVStore + 'txn,
126    C: Catalog + ?Sized,
127    T: SqlTxn<'txn, S>,
128>(
129    txn: &mut T,
130    catalog: &C,
131    plan: LogicalPlan,
132    memory: Option<&MemoryPolicy>,
133) -> Result<QueryRowIterator<'static>> {
134    // KNN queries not yet supported for streaming - fall back would need different handling
135    if knn::extract_knn_context(&plan).is_some() {
136        // For KNN, we materialize and wrap in VecIterator
137        let result = execute_query_with_policy(txn, catalog, plan, memory)?;
138        if let ExecutionResult::Query(qr) = result {
139            let (iter, projection, schema) = materialize_query_result(qr);
140            return Ok(QueryRowIterator::new(iter, projection, schema));
141        }
142        return Err(ExecutorError::InvalidOperation {
143            operation: "execute_query_streaming".into(),
144            reason: "KNN query did not return Query result".into(),
145        });
146    }
147
148    // Subqueries need transaction access during evaluation, which streaming
149    // iterators borrow exclusively. Execute through the materializing path
150    // (the same one used by `execute_query`) so results are identical to the
151    // non-streaming API instead of failing or silently dropping rows.
152    if subquery::plan_contains_subquery(&plan) {
153        let result = execute_query_result_with_outer_and_policy(txn, catalog, plan, None, memory)?;
154        let (iter, projection, schema) = materialize_query_result(result);
155        return Ok(QueryRowIterator::new(iter, projection, schema));
156    }
157
158    let (iter, projection, schema) = build_iterator_pipeline(txn, catalog, plan, memory)?;
159
160    Ok(QueryRowIterator::new(iter, projection, schema))
161}
162
163/// Convert a materialized `QueryResult` into pipeline outputs.
164///
165/// The resulting rows are already fully projected, so the returned projection
166/// is `Projection::All` over the output column names.
167fn materialize_query_result(
168    result: QueryResult,
169) -> (
170    Box<dyn RowIterator>,
171    Projection,
172    Vec<crate::catalog::ColumnMetadata>,
173) {
174    let column_names: Vec<String> = result.columns.iter().map(|c| c.name.clone()).collect();
175    let schema: Vec<crate::catalog::ColumnMetadata> = result
176        .columns
177        .iter()
178        .map(|c| crate::catalog::ColumnMetadata::new(&c.name, c.data_type.clone()))
179        .collect();
180    let rows: Vec<Row> = result
181        .rows
182        .into_iter()
183        .enumerate()
184        .map(|(i, values)| Row::new(i as u64, values))
185        .collect();
186    let iter = iterator::VecIterator::new(rows, schema.clone());
187    (Box::new(iter), Projection::All(column_names), schema)
188}
189
190/// Build an iterator pipeline from a logical plan.
191///
192/// This recursively constructs a tree of iterators that mirrors the logical plan
193/// structure. The scan phase reads rows into memory, then subsequent operators
194/// process them through an iterator pipeline enabling streaming execution and
195/// early termination.
196fn build_iterator_pipeline<'txn, S: KVStore + 'txn, C: Catalog + ?Sized, T: SqlTxn<'txn, S>>(
197    txn: &mut T,
198    catalog: &C,
199    plan: LogicalPlan,
200    memory: Option<&MemoryPolicy>,
201) -> Result<(
202    Box<dyn RowIterator>,
203    Projection,
204    Vec<crate::catalog::ColumnMetadata>,
205)> {
206    build_iterator_pipeline_with_outer(txn, catalog, plan, memory, None)
207}
208
209fn build_iterator_pipeline_with_outer<
210    'txn,
211    S: KVStore + 'txn,
212    C: Catalog + ?Sized,
213    T: SqlTxn<'txn, S>,
214>(
215    txn: &mut T,
216    catalog: &C,
217    plan: LogicalPlan,
218    memory: Option<&MemoryPolicy>,
219    outer: Option<&Row>,
220) -> Result<(
221    Box<dyn RowIterator>,
222    Projection,
223    Vec<crate::catalog::ColumnMetadata>,
224)> {
225    match plan {
226        LogicalPlan::Scan { table, projection } => {
227            if table == LITERAL_TABLE {
228                let schema = Vec::new();
229                let rows = vec![Row::new(0, Vec::new())];
230                let iter = iterator::VecIterator::new(rows, schema.clone());
231                return Ok((Box::new(iter), projection, schema));
232            }
233            let table_meta = catalog
234                .get_table(&table)
235                .cloned()
236                .ok_or_else(|| ExecutorError::TableNotFound(table.clone()))?;
237
238            if table_meta.storage_options.storage_type == StorageType::Columnar {
239                let columnar_scan = columnar_scan::build_columnar_scan(&table_meta, &projection);
240                let rows = columnar_scan::execute_columnar_scan(txn, &table_meta, &columnar_scan)?;
241                let schema = table_meta.columns.clone();
242                let iter = iterator::VecIterator::new(rows, schema.clone());
243                return Ok((Box::new(iter), projection, schema));
244            }
245
246            // TODO: 現状は Scan で一度全件をメモリに載せてから iterator に渡しています。
247            // 将来ストリーミングを徹底する場合は、ScanIterator を活用できるよう
248            // トランザクションのライフタイム設計を見直すとよいです。
249            let rows = scan::execute_scan(txn, &table_meta)?;
250            let schema = table_meta.columns.clone();
251
252            // Wrap in VecIterator for consistent iterator-based processing
253            let iter = iterator::VecIterator::new(rows, schema.clone());
254            Ok((Box::new(iter), projection, schema))
255        }
256        LogicalPlan::Filter { input, predicate } => {
257            if let LogicalPlan::Scan { table, projection } = input.as_ref()
258                && let Some(table_meta) = catalog.get_table(table)
259                && table_meta.storage_options.storage_type == StorageType::Columnar
260            {
261                let columnar_scan = columnar_scan::build_columnar_scan_for_filter(
262                    table_meta,
263                    projection.clone(),
264                    &predicate,
265                );
266                let rows = columnar_scan::execute_columnar_scan(txn, table_meta, &columnar_scan)?;
267                let schema = table_meta.columns.clone();
268                let iter = iterator::VecIterator::new(rows, schema.clone());
269                return Ok((Box::new(iter), projection.clone(), schema));
270            }
271            let (mut input_iter, projection, schema) =
272                build_iterator_pipeline_with_outer(txn, catalog, *input, memory, outer)?;
273            if outer.is_some() || subquery::contains_subquery(&predicate) {
274                let mut rows = Vec::new();
275                while let Some(result) = input_iter.next_row() {
276                    let row = result?;
277                    let eval_row = combine_outer_for_eval(&row, outer);
278                    if let SqlValue::Boolean(true) = subquery::evaluate_expr_with_subqueries(
279                        txn, catalog, &predicate, &eval_row,
280                    )? {
281                        rows.push(row);
282                    }
283                }
284                let iter = iterator::VecIterator::new(rows, schema.clone());
285                return Ok((Box::new(iter), projection, schema));
286            }
287            let filter_iter = FilterIterator::new(input_iter, predicate);
288            Ok((Box::new(filter_iter), projection, schema))
289        }
290        LogicalPlan::Project { input, projection } => {
291            let (mut input_iter, _input_projection, schema) =
292                build_iterator_pipeline_with_outer(txn, catalog, *input, memory, outer)?;
293            let mut rows = Vec::new();
294            while let Some(result) = input_iter.next_row() {
295                rows.push(result?);
296            }
297            let projected =
298                execute_project_with_subqueries(txn, catalog, rows, &projection, &schema, outer)?;
299            let output_schema = projected
300                .columns
301                .iter()
302                .map(|col| crate::catalog::ColumnMetadata::new(&col.name, col.data_type.clone()))
303                .collect::<Vec<_>>();
304            let rows = projected
305                .rows
306                .into_iter()
307                .enumerate()
308                .map(|(idx, values)| Row::new(idx as u64, values))
309                .collect::<Vec<_>>();
310            let output_projection =
311                Projection::All(output_schema.iter().map(|col| col.name.clone()).collect());
312            let iter = iterator::VecIterator::new(rows, output_schema.clone());
313            Ok((Box::new(iter), output_projection, output_schema))
314        }
315        LogicalPlan::Join {
316            left,
317            right,
318            join_type,
319            condition,
320            using: _,
321        } => {
322            let (mut left_iter, _left_projection, left_schema) =
323                build_iterator_pipeline_with_outer(txn, catalog, *left, memory, outer)?;
324            let (mut right_iter, _right_projection, right_schema) =
325                build_iterator_pipeline_with_outer(txn, catalog, *right, memory, outer)?;
326            let mut left_rows = Vec::new();
327            while let Some(result) = left_iter.next_row() {
328                left_rows.push(result?);
329            }
330            let mut right_rows = Vec::new();
331            while let Some(result) = right_iter.next_row() {
332                right_rows.push(result?);
333            }
334            let left_width = left_schema.len();
335            let right_width = right_schema.len();
336            let rows = join::execute_join_with_widths(
337                left_rows,
338                right_rows,
339                join_type,
340                condition.as_ref(),
341                left_width,
342                right_width,
343            )?;
344            let mut schema = left_schema;
345            schema.extend(right_schema);
346            let projection = Projection::All(schema.iter().map(|col| col.name.clone()).collect());
347            let iter = iterator::VecIterator::new(rows, schema.clone());
348            Ok((Box::new(iter), projection, schema))
349        }
350        LogicalPlan::Aggregate {
351            input,
352            group_keys,
353            aggregates,
354            having,
355            projection,
356        } => {
357            let (input_iter, _projection, _schema) =
358                build_iterator_pipeline_with_outer(txn, catalog, *input, memory, outer)?;
359            let schema = aggregate::build_aggregate_schema(&group_keys, &aggregates);
360            if let Some(policy) = memory
361                && policy.spill_directory().is_some()
362            {
363                if group_keys.is_empty() {
364                    let iter = aggregate::StreamingAggregateIterator::new(
365                        input_iter,
366                        group_keys,
367                        aggregates,
368                        having,
369                        schema.clone(),
370                    );
371                    return Ok((Box::new(iter), projection, schema));
372                }
373                let order_by = group_keys
374                    .iter()
375                    .cloned()
376                    .map(|expr| SortExpr {
377                        expr,
378                        asc: true,
379                        nulls_first: false,
380                    })
381                    .collect::<Vec<_>>();
382                let sort_iter =
383                    SortIterator::new_with_policy(input_iter, &order_by, Some(policy.clone()))?;
384                let iter = aggregate::StreamingAggregateIterator::new(
385                    Box::new(sort_iter),
386                    group_keys,
387                    aggregates,
388                    having,
389                    schema.clone(),
390                );
391                return Ok((Box::new(iter), projection, schema));
392            }
393
394            let parallelism = std::thread::available_parallelism()
395                .map(usize::from)
396                .unwrap_or(1);
397            if !aggregate::should_use_single_for_parallel(parallelism, &aggregates) {
398                let rows = aggregate::execute_parallel_aggregate_rows_with_policy(
399                    input_iter,
400                    group_keys,
401                    aggregates,
402                    having,
403                    schema.clone(),
404                    parallelism,
405                    memory.cloned(),
406                    1_000_000,
407                )?;
408                let iter = iterator::VecIterator::new(rows, schema.clone());
409                return Ok((Box::new(iter), projection, schema));
410            }
411
412            let mut iter = aggregate::AggregateIterator::new(
413                input_iter,
414                group_keys,
415                aggregates,
416                having,
417                schema.clone(),
418            );
419            if let Some(policy) = memory {
420                iter = iter.with_memory_policy(Some(policy.clone()));
421            }
422            Ok((Box::new(iter), projection, schema))
423        }
424        LogicalPlan::Sort { input, order_by } => {
425            let (input_iter, projection, schema) =
426                build_iterator_pipeline_with_outer(txn, catalog, *input, memory, outer)?;
427            let sort_iter = if let Some(policy) = memory {
428                SortIterator::new_with_policy(input_iter, &order_by, Some(policy.clone()))?
429            } else {
430                SortIterator::new(input_iter, &order_by)?
431            };
432            Ok((Box::new(sort_iter), projection, schema))
433        }
434        LogicalPlan::Limit {
435            input,
436            limit,
437            offset,
438        } => {
439            let (input_iter, projection, schema) =
440                build_iterator_pipeline_with_outer(txn, catalog, *input, memory, outer)?;
441            let limit_iter = LimitIterator::new(input_iter, limit, offset);
442            Ok((Box::new(limit_iter), projection, schema))
443        }
444        other => Err(ExecutorError::UnsupportedOperation(format!(
445            "unsupported query plan: {other:?}"
446        ))),
447    }
448}
449
450/// Build a streaming iterator pipeline from a logical plan (FR-7).
451///
452/// This version uses `ScanIterator` for row-based tables to enable true
453/// streaming without materializing all rows upfront. The returned iterator
454/// has lifetime `'a` tied to the transaction borrow.
455///
456/// # Limitations
457///
458/// - Columnar storage still materializes rows (uses VecIterator)
459/// - Sort operations materialize all input rows
460/// - KNN queries are not supported (use `build_iterator_pipeline` instead)
461pub fn build_streaming_pipeline<
462    'a,
463    'txn: 'a,
464    S: KVStore + 'txn,
465    C: Catalog + ?Sized,
466    T: SqlTxn<'txn, S>,
467>(
468    txn: &'a mut T,
469    catalog: &C,
470    plan: LogicalPlan,
471) -> Result<(
472    Box<dyn RowIterator + 'a>,
473    Projection,
474    Vec<crate::catalog::ColumnMetadata>,
475)> {
476    build_streaming_pipeline_with_policy(txn, catalog, plan, None)
477}
478
479pub fn build_streaming_pipeline_with_policy<
480    'a,
481    'txn: 'a,
482    S: KVStore + 'txn,
483    C: Catalog + ?Sized,
484    T: SqlTxn<'txn, S>,
485>(
486    txn: &'a mut T,
487    catalog: &C,
488    plan: LogicalPlan,
489    memory: Option<&MemoryPolicy>,
490) -> Result<(
491    Box<dyn RowIterator + 'a>,
492    Projection,
493    Vec<crate::catalog::ColumnMetadata>,
494)> {
495    // Subqueries need transaction access during evaluation, which streaming
496    // iterators borrow exclusively. Execute through the materializing path
497    // (the same one used by `execute_query`) so results are identical to the
498    // non-streaming API instead of failing or silently dropping rows
499    // (GitHub issues #23 / #24).
500    if subquery::plan_contains_subquery(&plan) {
501        let result = execute_query_result_with_outer_and_policy(txn, catalog, plan, None, memory)?;
502        return Ok(materialize_query_result(result));
503    }
504
505    build_streaming_pipeline_inner(txn, catalog, plan, memory)
506}
507
508/// Inner implementation of streaming pipeline builder.
509fn build_streaming_pipeline_inner<
510    'a,
511    'txn: 'a,
512    S: KVStore + 'txn,
513    C: Catalog + ?Sized,
514    T: SqlTxn<'txn, S>,
515>(
516    txn: &'a mut T,
517    catalog: &C,
518    plan: LogicalPlan,
519    memory: Option<&MemoryPolicy>,
520) -> Result<(
521    Box<dyn RowIterator + 'a>,
522    Projection,
523    Vec<crate::catalog::ColumnMetadata>,
524)> {
525    match plan {
526        LogicalPlan::Scan { table, projection } => {
527            if table == LITERAL_TABLE {
528                let schema = Vec::new();
529                let rows = vec![Row::new(0, Vec::new())];
530                let iter = iterator::VecIterator::new(rows, schema.clone());
531                return Ok((Box::new(iter), projection, schema));
532            }
533            let table_meta = catalog
534                .get_table(&table)
535                .cloned()
536                .ok_or_else(|| ExecutorError::TableNotFound(table.clone()))?;
537
538            if table_meta.storage_options.storage_type == StorageType::Columnar {
539                // Columnar storage: use ColumnarScanIterator for FR-7 streaming
540                let columnar_scan = columnar_scan::build_columnar_scan(&table_meta, &projection);
541                let schema = table_meta.columns.clone();
542                let iter =
543                    columnar_scan::create_columnar_scan_iterator(txn, &table_meta, &columnar_scan)?;
544                return Ok((Box::new(iter), projection, schema));
545            }
546
547            // Row-based storage: use ScanIterator for true streaming (FR-7)
548            let schema = table_meta.columns.clone();
549            let scan_iter = scan::create_scan_iterator(txn, &table_meta)?;
550            Ok((Box::new(scan_iter), projection, schema))
551        }
552        LogicalPlan::Filter { input, predicate } => {
553            if let LogicalPlan::Scan { table, projection } = input.as_ref()
554                && let Some(table_meta) = catalog.get_table(table)
555                && table_meta.storage_options.storage_type == StorageType::Columnar
556            {
557                // Columnar storage with filter: use ColumnarScanIterator for FR-7 streaming
558                let columnar_scan = columnar_scan::build_columnar_scan_for_filter(
559                    table_meta,
560                    projection.clone(),
561                    &predicate,
562                );
563                let schema = table_meta.columns.clone();
564                let iter =
565                    columnar_scan::create_columnar_scan_iterator(txn, table_meta, &columnar_scan)?;
566                return Ok((Box::new(iter), projection.clone(), schema));
567            }
568            let (input_iter, projection, schema) =
569                build_streaming_pipeline_inner(txn, catalog, *input, memory)?;
570            let filter_iter = FilterIterator::new(input_iter, predicate);
571            Ok((Box::new(filter_iter), projection, schema))
572        }
573        LogicalPlan::Project { input, projection } => {
574            let (mut input_iter, _input_projection, schema) =
575                build_streaming_pipeline_inner(txn, catalog, *input, memory)?;
576            let mut rows = Vec::new();
577            while let Some(result) = input_iter.next_row() {
578                rows.push(result?);
579            }
580            let projected = project::execute_project(rows, &projection, &schema)?;
581            let output_schema = projected
582                .columns
583                .iter()
584                .map(|col| crate::catalog::ColumnMetadata::new(&col.name, col.data_type.clone()))
585                .collect::<Vec<_>>();
586            let rows = projected
587                .rows
588                .into_iter()
589                .enumerate()
590                .map(|(idx, values)| Row::new(idx as u64, values))
591                .collect::<Vec<_>>();
592            let output_projection =
593                Projection::All(output_schema.iter().map(|col| col.name.clone()).collect());
594            let iter = iterator::VecIterator::new(rows, output_schema.clone());
595            Ok((Box::new(iter), output_projection, output_schema))
596        }
597        LogicalPlan::Join {
598            left,
599            right,
600            join_type,
601            condition,
602            using: _,
603        } => {
604            let (mut left_iter, _left_projection, left_schema) =
605                build_streaming_pipeline_inner(txn, catalog, *left, memory)?;
606            let mut left_rows = Vec::new();
607            while let Some(result) = left_iter.next_row() {
608                left_rows.push(result?);
609            }
610            drop(left_iter);
611            let (mut right_iter, _right_projection, right_schema) =
612                build_streaming_pipeline_inner(txn, catalog, *right, memory)?;
613            let mut right_rows = Vec::new();
614            while let Some(result) = right_iter.next_row() {
615                right_rows.push(result?);
616            }
617            let rows = join::execute_join_with_widths(
618                left_rows,
619                right_rows,
620                join_type,
621                condition.as_ref(),
622                left_schema.len(),
623                right_schema.len(),
624            )?;
625            let mut schema = left_schema;
626            schema.extend(right_schema);
627            let projection = Projection::All(schema.iter().map(|col| col.name.clone()).collect());
628            let iter = iterator::VecIterator::new(rows, schema.clone());
629            Ok((Box::new(iter), projection, schema))
630        }
631        LogicalPlan::Aggregate {
632            input,
633            group_keys,
634            aggregates,
635            having,
636            projection,
637        } => {
638            let (input_iter, _projection, _schema) =
639                build_streaming_pipeline_inner(txn, catalog, *input, memory)?;
640            let schema = aggregate::build_aggregate_schema(&group_keys, &aggregates);
641            if let Some(policy) = memory
642                && policy.spill_directory().is_some()
643            {
644                if group_keys.is_empty() {
645                    let iter = aggregate::StreamingAggregateIterator::new(
646                        input_iter,
647                        group_keys,
648                        aggregates,
649                        having,
650                        schema.clone(),
651                    );
652                    return Ok((Box::new(iter), projection, schema));
653                }
654                let order_by = group_keys
655                    .iter()
656                    .cloned()
657                    .map(|expr| SortExpr {
658                        expr,
659                        asc: true,
660                        nulls_first: false,
661                    })
662                    .collect::<Vec<_>>();
663                let sort_iter =
664                    SortIterator::new_with_policy(input_iter, &order_by, Some(policy.clone()))?;
665                let iter = aggregate::StreamingAggregateIterator::new(
666                    Box::new(sort_iter),
667                    group_keys,
668                    aggregates,
669                    having,
670                    schema.clone(),
671                );
672                return Ok((Box::new(iter), projection, schema));
673            }
674
675            let parallelism = std::thread::available_parallelism()
676                .map(usize::from)
677                .unwrap_or(1);
678            if !aggregate::should_use_single_for_parallel(parallelism, &aggregates) {
679                let rows = aggregate::execute_parallel_aggregate_rows_with_policy(
680                    input_iter,
681                    group_keys,
682                    aggregates,
683                    having,
684                    schema.clone(),
685                    parallelism,
686                    memory.cloned(),
687                    1_000_000,
688                )?;
689                let iter = iterator::VecIterator::new(rows, schema.clone());
690                return Ok((Box::new(iter), projection, schema));
691            }
692
693            let mut iter = aggregate::AggregateIterator::new(
694                input_iter,
695                group_keys,
696                aggregates,
697                having,
698                schema.clone(),
699            );
700            if let Some(policy) = memory {
701                iter = iter.with_memory_policy(Some(policy.clone()));
702            }
703            Ok((Box::new(iter), projection, schema))
704        }
705        LogicalPlan::Sort { input, order_by } => {
706            let (input_iter, projection, schema) =
707                build_streaming_pipeline_inner(txn, catalog, *input, memory)?;
708            let sort_iter = if let Some(policy) = memory {
709                SortIterator::new_with_policy(input_iter, &order_by, Some(policy.clone()))?
710            } else {
711                SortIterator::new(input_iter, &order_by)?
712            };
713            Ok((Box::new(sort_iter), projection, schema))
714        }
715        LogicalPlan::Limit {
716            input,
717            limit,
718            offset,
719        } => {
720            let (input_iter, projection, schema) =
721                build_streaming_pipeline_inner(txn, catalog, *input, memory)?;
722            let limit_iter = LimitIterator::new(input_iter, limit, offset);
723            Ok((Box::new(limit_iter), projection, schema))
724        }
725        other => Err(ExecutorError::UnsupportedOperation(format!(
726            "unsupported query plan: {other:?}"
727        ))),
728    }
729}
730
731/// Evaluate a typed expression against a row, returning SqlValue.
732fn eval_expr(expr: &crate::planner::typed_expr::TypedExpr, row: &Row) -> Result<SqlValue> {
733    let ctx = EvalContext::new(&row.values);
734    crate::executor::evaluator::evaluate(expr, &ctx)
735}
736
737fn combine_outer_for_eval(row: &Row, outer: Option<&Row>) -> Row {
738    let Some(outer) = outer else {
739        return row.clone();
740    };
741    let mut values = Vec::with_capacity(row.len() + outer.len());
742    values.extend(row.values.clone());
743    values.extend(outer.values.clone());
744    Row::new(row.row_id, values)
745}
746
747fn execute_project_with_subqueries<
748    'txn,
749    S: KVStore + 'txn,
750    C: Catalog + ?Sized,
751    T: SqlTxn<'txn, S>,
752>(
753    txn: &mut T,
754    catalog: &C,
755    rows: Vec<Row>,
756    projection: &Projection,
757    schema: &[crate::catalog::ColumnMetadata],
758    outer: Option<&Row>,
759) -> Result<QueryResult> {
760    match projection {
761        Projection::All(_) => project::execute_project(rows, projection, schema),
762        Projection::Columns(cols)
763            if outer.is_some() || cols.iter().any(|c| subquery::contains_subquery(&c.expr)) =>
764        {
765            let columns: Vec<_> = cols
766                .iter()
767                .enumerate()
768                .map(|(i, c)| column_info_from_projection(c, i))
769                .collect();
770            let mut projected_rows = Vec::with_capacity(rows.len());
771            for row in rows {
772                let eval_row = combine_outer_for_eval(&row, outer);
773                let mut values = Vec::with_capacity(cols.len());
774                for col in cols {
775                    values.push(subquery::evaluate_expr_with_subqueries(
776                        txn, catalog, &col.expr, &eval_row,
777                    )?);
778                }
779                projected_rows.push(values);
780            }
781            Ok(QueryResult::new(columns, projected_rows))
782        }
783        Projection::Columns(_) => project::execute_project(rows, projection, schema),
784    }
785}
786
787/// Build column info name using alias fallback.
788fn column_name_from_projection(
789    projected: &crate::planner::typed_expr::ProjectedColumn,
790    idx: usize,
791) -> String {
792    projected
793        .alias
794        .clone()
795        .or_else(|| match &projected.expr.kind {
796            crate::planner::typed_expr::TypedExprKind::ColumnRef { column, .. } => {
797                Some(column.clone())
798            }
799            _ => None,
800        })
801        .unwrap_or_else(|| format!("col_{idx}"))
802}
803
804/// Build ColumnInfo from projection.
805fn column_info_from_projection(
806    projected: &crate::planner::typed_expr::ProjectedColumn,
807    idx: usize,
808) -> ColumnInfo {
809    ColumnInfo::new(
810        column_name_from_projection(projected, idx),
811        projected.expr.resolved_type.clone(),
812    )
813}
814
815/// Build ColumnInfo for Projection::All using schema.
816fn column_infos_from_all(
817    schema: &[crate::catalog::ColumnMetadata],
818    names: &[String],
819) -> Result<Vec<ColumnInfo>> {
820    names
821        .iter()
822        .map(|name| {
823            let col = schema
824                .iter()
825                .find(|c| &c.name == name)
826                .ok_or_else(|| ExecutorError::ColumnNotFound(name.clone()))?;
827            Ok(ColumnInfo::new(name.clone(), col.data_type.clone()))
828        })
829        .collect()
830}
831
832#[cfg(test)]
833mod tests {
834    use super::*;
835    use crate::catalog::{ColumnMetadata, MemoryCatalog, TableMetadata};
836    use crate::executor::ddl::create_table::execute_create_table;
837    use crate::planner::typed_expr::TypedExpr;
838    use crate::planner::types::ResolvedType;
839    use crate::storage::TxnBridge;
840    use alopex_core::kv::memory::MemoryKV;
841    use std::sync::Arc;
842
843    #[test]
844    fn execute_query_scan_only_returns_rows() {
845        let bridge = TxnBridge::new(Arc::new(MemoryKV::new()));
846        let mut catalog = MemoryCatalog::new();
847        let table = TableMetadata::new(
848            "users",
849            vec![
850                ColumnMetadata::new("id", ResolvedType::Integer),
851                ColumnMetadata::new("name", ResolvedType::Text),
852            ],
853        );
854        let mut ddl_txn = bridge.begin_write().unwrap();
855        execute_create_table(&mut ddl_txn, &mut catalog, table.clone(), vec![], false).unwrap();
856        ddl_txn.commit().unwrap();
857
858        let mut txn = bridge.begin_write().unwrap();
859        crate::executor::dml::execute_insert(
860            &mut txn,
861            &catalog,
862            "users",
863            vec!["id".into(), "name".into()],
864            vec![vec![
865                TypedExpr::literal(
866                    crate::ast::expr::Literal::Number("1".into()),
867                    ResolvedType::Integer,
868                    crate::Span::default(),
869                ),
870                TypedExpr::literal(
871                    crate::ast::expr::Literal::String("alice".into()),
872                    ResolvedType::Text,
873                    crate::Span::default(),
874                ),
875            ]],
876        )
877        .unwrap();
878
879        let result = execute_query(
880            &mut txn,
881            &catalog,
882            LogicalPlan::scan(
883                "users".into(),
884                Projection::All(vec!["id".into(), "name".into()]),
885            ),
886        )
887        .unwrap();
888
889        match result {
890            ExecutionResult::Query(q) => {
891                assert_eq!(q.rows.len(), 1);
892                assert_eq!(q.columns.len(), 2);
893                assert_eq!(
894                    q.rows[0],
895                    vec![SqlValue::Integer(1), SqlValue::Text("alice".into())]
896                );
897            }
898            other => panic!("unexpected result {other:?}"),
899        }
900    }
901}
alopex_sql/executor/query/mod.rs

alopex_sql/executor/query/
mod.rs
