Skip to main content

vibesql_executor/select/columnar/
mod.rs

1//! Columnar execution for high-performance aggregation queries
2//!
3//! This module implements column-oriented query execution that avoids
4//! materializing full Row objects during table scans, providing 8-10x speedup
5//! for aggregation-heavy workloads.
6//!
7//! ## Architecture
8//!
9//! Instead of:
10//! ```text
11//! TableScan → Row{Vec<SqlValue>} → Filter(Row) → Aggregate(Row) → Vec<Row>
12//! ```
13//!
14//! We use:
15//! ```text
16//! TableScan → ColumnRefs → Filter(native types) → Aggregate → Row
17//! ```
18//!
19//! ## Benefits
20//!
21//! - **Zero-copy**: Work with `&SqlValue` references instead of cloning
22//! - **Cache-friendly**: Access contiguous column data instead of scattered row data
23//! - **Type-specialized**: Skip SqlValue enum matching overhead for filters/aggregates
24//! - **Minimal allocations**: Only allocate result rows, not intermediate data
25//!
26//! ## Usage
27//!
28//! This path is automatically selected for simple aggregate queries that:
29//! - Have a single table scan (no JOINs)
30//! - Use simple WHERE predicates
31//! - Compute aggregates (SUM, COUNT, AVG, MIN, MAX)
32//! - Don't use window functions or complex subqueries
33
34mod aggregate;
35pub mod batch;
36mod executor;
37pub mod filter;
38mod scan;
39mod string_ops;
40
41// Auto-vectorized SIMD operations - replaces the `wide` crate dependency
42// See simd_ops.rs for documentation on why these patterns are structured this way
43pub mod simd_ops;
44
45mod simd_aggregate;
46pub mod simd_filter;
47mod simd_join;
48
49pub use aggregate::{
50    columnar_group_by, columnar_group_by_batch, compute_aggregates_from_batch,
51    compute_multiple_aggregates, evaluate_expression_to_column,
52    evaluate_expression_with_cached_column, extract_aggregates, AggregateOp, AggregateSource,
53    AggregateSpec,
54};
55pub use batch::{ColumnArray, ColumnarBatch};
56pub use executor::execute_columnar_batch;
57pub use filter::{
58    apply_columnar_filter, apply_columnar_filter_simd_streaming, create_filter_bitmap,
59    create_filter_bitmap_tree, evaluate_predicate_tree, extract_column_predicates,
60    extract_predicate_tree, ColumnPredicate, PredicateTree,
61};
62use log;
63pub use scan::ColumnarScan;
64pub use simd_aggregate::{can_use_simd_for_column, simd_aggregate_f64, simd_aggregate_i64};
65pub use simd_filter::{
66    simd_create_filter_mask, simd_create_filter_mask_auto, simd_create_filter_mask_packed,
67    simd_filter_batch, simd_filter_to_indices,
68};
69#[cfg(feature = "parallel")]
70pub use simd_filter::{simd_create_filter_mask_parallel, simd_filter_batch_parallel};
71pub use simd_join::columnar_hash_join_inner;
72pub use simd_ops::PackedMask;
73use vibesql_storage::Row;
74use vibesql_types::SqlValue;
75
76use crate::{errors::ExecutorError, schema::CombinedSchema};
77
78/// Execute a columnar aggregate query with filtering
79///
80/// This is a simplified entry point for columnar execution that demonstrates
81/// the full pipeline: scan → filter → aggregate.
82///
83/// # Arguments
84///
85/// * `rows` - Input rows to process
86/// * `predicates` - Column predicates for filtering (optional)
87/// * `aggregates` - List of (column_index, aggregate_op) pairs to compute
88///
89/// # Returns
90///
91/// A single Row containing the computed aggregate values
92///
93/// # Example
94///
95/// ```text
96/// // Compute SUM(col0), AVG(col1) WHERE col2 < 100
97/// let predicates = vec![
98///     ColumnPredicate::LessThan {
99///         column_idx: 2,
100///         value: SqlValue::Integer(100),
101///     },
102/// ];
103/// let aggregates = vec![
104///     (0, AggregateOp::Sum),
105///     (1, AggregateOp::Avg),
106/// ];
107///
108/// let result = execute_columnar_aggregate(&rows, &predicates, &aggregates)?;
109/// ```
110///
111/// Note: This function provides SIMD-accelerated filtering and aggregation through
112/// LLVM auto-vectorization of batch-native operations.
113pub fn execute_columnar_aggregate(
114    rows: &[Row],
115    predicates: &[ColumnPredicate],
116    aggregates: &[aggregate::AggregateSpec],
117    schema: Option<&CombinedSchema>,
118) -> Result<Vec<Row>, ExecutorError> {
119    // Early return for empty input
120    // SQL standard: COUNT returns 0 for empty input, other aggregates return NULL
121    if rows.is_empty() {
122        let values: Vec<SqlValue> = aggregates
123            .iter()
124            .map(|spec| match spec.op {
125                aggregate::AggregateOp::Count => SqlValue::Integer(0),
126                _ => SqlValue::Null,
127            })
128            .collect();
129        return Ok(vec![Row::new(values)]);
130    }
131
132    // Phase 1: Convert to columnar batch for SIMD acceleration
133    #[cfg(feature = "profile-q6")]
134    let batch_start = std::time::Instant::now();
135
136    let batch = ColumnarBatch::from_rows(rows)?;
137
138    #[cfg(feature = "profile-q6")]
139    {
140        let batch_time = batch_start.elapsed();
141        eprintln!("[PROFILE-Q6]   Phase 1 - Convert to batch: {:?}", batch_time);
142    }
143
144    // Phase 2: Apply SIMD-accelerated filtering
145    #[cfg(feature = "profile-q6")]
146    let filter_start = std::time::Instant::now();
147
148    let filtered_batch =
149        if predicates.is_empty() { batch.clone() } else { simd_filter_batch(&batch, predicates)? };
150
151    #[cfg(feature = "profile-q6")]
152    {
153        let filter_time = filter_start.elapsed();
154        eprintln!(
155            "[PROFILE-Q6]   Phase 2 - SIMD filter: {:?} ({}/{} rows passed)",
156            filter_time,
157            filtered_batch.row_count(),
158            rows.len()
159        );
160    }
161
162    // Phase 3: Compute aggregates directly on batch (no row conversion!)
163    #[cfg(feature = "profile-q6")]
164    let agg_start = std::time::Instant::now();
165
166    // Use batch-native aggregation to avoid to_rows() conversion overhead
167    let results = compute_aggregates_from_batch(&filtered_batch, aggregates, schema)?;
168
169    #[cfg(feature = "profile-q6")]
170    {
171        let agg_time = agg_start.elapsed();
172        eprintln!(
173            "[PROFILE-Q6]   Phase 3 - Batch-native aggregate: {:?} ({} aggregates)",
174            agg_time,
175            aggregates.len()
176        );
177    }
178
179    // Return as single row
180    Ok(vec![Row::new(results)])
181}
182
183/// Fast single-pass aggregate on rows - avoids batch conversion overhead
184///
185/// This function performs filtering and aggregation in a single pass over the input rows,
186/// without converting to columnar format. It's 3-5x faster than `execute_columnar_aggregate`
187/// for queries that come from row-based storage.
188///
189/// # Use Cases
190///
191/// Best suited for:
192/// - Simple aggregate queries without GROUP BY
193/// - When data arrives as Vec<Row> (not native columnar)
194/// - TPC-H style queries: SUM(price * discount) WHERE ...
195///
196/// # Arguments
197///
198/// * `rows` - Input rows to process
199/// * `predicates` - Column predicates for filtering
200/// * `aggregates` - Aggregate specifications
201///
202/// # Returns
203///
204/// A single Row containing the computed aggregate values
205pub fn fast_aggregate_on_rows(
206    rows: &[Row],
207    predicates: &[ColumnPredicate],
208    aggregates: &[aggregate::AggregateSpec],
209) -> Result<Vec<Row>, ExecutorError> {
210    use aggregate::{AggregateOp, AggregateSource};
211
212    // Early return for empty input
213    if rows.is_empty() {
214        let values: Vec<SqlValue> = aggregates
215            .iter()
216            .map(|spec| match spec.op {
217                AggregateOp::Count => SqlValue::Integer(0),
218                _ => SqlValue::Null,
219            })
220            .collect();
221        return Ok(vec![Row::new(values)]);
222    }
223
224    // Initialize accumulators for each aggregate
225    struct Accumulator {
226        sum_f64: f64,
227        sum_i64: i64,
228        count: i64,
229        min_f64: Option<f64>,
230        max_f64: Option<f64>,
231        min_i64: Option<i64>,
232        max_i64: Option<i64>,
233        is_integer: bool,
234    }
235
236    let mut accumulators: Vec<Accumulator> = aggregates
237        .iter()
238        .map(|_| Accumulator {
239            sum_f64: 0.0,
240            sum_i64: 0,
241            count: 0,
242            min_f64: None,
243            max_f64: None,
244            min_i64: None,
245            max_i64: None,
246            is_integer: true,
247        })
248        .collect();
249
250    // Single pass: filter and accumulate
251    for row in rows {
252        // Check all predicates
253        let passes_filter = predicates.iter().all(|pred| evaluate_predicate(row, pred));
254
255        if !passes_filter {
256            continue;
257        }
258
259        // Accumulate values for each aggregate
260        for (i, spec) in aggregates.iter().enumerate() {
261            let acc = &mut accumulators[i];
262
263            match &spec.source {
264                AggregateSource::CountStar => {
265                    acc.count += 1;
266                }
267                AggregateSource::Column(col_idx) => {
268                    if let Some(value) = row.get(*col_idx) {
269                        if !matches!(value, SqlValue::Null) {
270                            acc.count += 1;
271                            match value {
272                                SqlValue::Integer(v) => {
273                                    acc.sum_i64 += v;
274                                    acc.sum_f64 += *v as f64;
275                                    acc.min_i64 = Some(acc.min_i64.map_or(*v, |m| m.min(*v)));
276                                    acc.max_i64 = Some(acc.max_i64.map_or(*v, |m| m.max(*v)));
277                                    acc.min_f64 =
278                                        Some(acc.min_f64.map_or(*v as f64, |m| m.min(*v as f64)));
279                                    acc.max_f64 =
280                                        Some(acc.max_f64.map_or(*v as f64, |m| m.max(*v as f64)));
281                                }
282                                SqlValue::Double(v) => {
283                                    acc.is_integer = false;
284                                    acc.sum_f64 += v;
285                                    acc.min_f64 = Some(acc.min_f64.map_or(*v, |m| m.min(*v)));
286                                    acc.max_f64 = Some(acc.max_f64.map_or(*v, |m| m.max(*v)));
287                                }
288                                SqlValue::Float(v) => {
289                                    acc.is_integer = false;
290                                    acc.sum_f64 += *v as f64;
291                                    acc.min_f64 =
292                                        Some(acc.min_f64.map_or(*v as f64, |m| m.min(*v as f64)));
293                                    acc.max_f64 =
294                                        Some(acc.max_f64.map_or(*v as f64, |m| m.max(*v as f64)));
295                                }
296                                SqlValue::Bigint(v) => {
297                                    acc.sum_i64 += v;
298                                    acc.sum_f64 += *v as f64;
299                                    acc.min_i64 = Some(acc.min_i64.map_or(*v, |m| m.min(*v)));
300                                    acc.max_i64 = Some(acc.max_i64.map_or(*v, |m| m.max(*v)));
301                                    acc.min_f64 =
302                                        Some(acc.min_f64.map_or(*v as f64, |m| m.min(*v as f64)));
303                                    acc.max_f64 =
304                                        Some(acc.max_f64.map_or(*v as f64, |m| m.max(*v as f64)));
305                                }
306                                SqlValue::Numeric(v) => {
307                                    acc.is_integer = false;
308                                    acc.sum_f64 += v;
309                                    acc.min_f64 = Some(acc.min_f64.map_or(*v, |m| m.min(*v)));
310                                    acc.max_f64 = Some(acc.max_f64.map_or(*v, |m| m.max(*v)));
311                                }
312                                _ => {}
313                            }
314                        }
315                    }
316                }
317                AggregateSource::Expression(expr) => {
318                    // For expression aggregates like SUM(a * b), evaluate the expression
319                    // This is a simplified evaluator for common binary operations
320                    if let Some(value) = eval_simple_expression(row, expr) {
321                        acc.count += 1;
322                        acc.is_integer = false;
323                        acc.sum_f64 += value;
324                        acc.min_f64 = Some(acc.min_f64.map_or(value, |m| m.min(value)));
325                        acc.max_f64 = Some(acc.max_f64.map_or(value, |m| m.max(value)));
326                    }
327                }
328            }
329        }
330    }
331
332    // Build result row from accumulators
333    let values: Vec<SqlValue> = aggregates
334        .iter()
335        .zip(accumulators.iter())
336        .map(|(spec, acc)| match spec.op {
337            AggregateOp::Count => SqlValue::Integer(acc.count),
338            AggregateOp::Sum => {
339                // SQLite's SUM() preserves integer type for integer inputs
340                if acc.count == 0 {
341                    SqlValue::Null
342                } else if acc.is_integer {
343                    SqlValue::Integer(acc.sum_i64)
344                } else {
345                    SqlValue::Double(acc.sum_f64)
346                }
347            }
348            AggregateOp::Avg => {
349                if acc.count == 0 {
350                    SqlValue::Null
351                } else {
352                    SqlValue::Double(acc.sum_f64 / acc.count as f64)
353                }
354            }
355            AggregateOp::Min => {
356                if acc.is_integer {
357                    acc.min_i64.map(SqlValue::Integer).unwrap_or(SqlValue::Null)
358                } else {
359                    acc.min_f64.map(SqlValue::Double).unwrap_or(SqlValue::Null)
360                }
361            }
362            AggregateOp::Max => {
363                if acc.is_integer {
364                    acc.max_i64.map(SqlValue::Integer).unwrap_or(SqlValue::Null)
365                } else {
366                    acc.max_f64.map(SqlValue::Double).unwrap_or(SqlValue::Null)
367                }
368            }
369        })
370        .collect();
371
372    Ok(vec![Row::new(values)])
373}
374
375/// Evaluate a column predicate against a row
376///
377/// Delegates to the canonical comparator in `filter::evaluation`. Issue
378/// #5335: this module previously carried a near-duplicate `compare_values`
379/// whose incomparable-types catch-all returned `Ordering::Equal`, turning
380/// equality/range predicates on temporal columns into tautologies or
381/// contradictions. Routing through the filter module keeps a single source
382/// of truth for predicate comparison semantics.
383fn evaluate_predicate(row: &Row, predicate: &ColumnPredicate) -> bool {
384    // Column-to-column comparison needs two values
385    if let ColumnPredicate::ColumnCompare { left_column_idx, op, right_column_idx } = predicate {
386        return filter::evaluate_column_compare(
387            *op,
388            row.get(*left_column_idx),
389            row.get(*right_column_idx),
390        );
391    }
392
393    let column_idx = match predicate {
394        ColumnPredicate::LessThan { column_idx, .. }
395        | ColumnPredicate::GreaterThan { column_idx, .. }
396        | ColumnPredicate::GreaterThanOrEqual { column_idx, .. }
397        | ColumnPredicate::LessThanOrEqual { column_idx, .. }
398        | ColumnPredicate::Equal { column_idx, .. }
399        | ColumnPredicate::NotEqual { column_idx, .. }
400        | ColumnPredicate::Between { column_idx, .. }
401        | ColumnPredicate::Like { column_idx, .. }
402        | ColumnPredicate::InList { column_idx, .. } => *column_idx,
403        ColumnPredicate::ColumnCompare { .. } => unreachable!(), // Handled above
404    };
405
406    match row.get(column_idx) {
407        Some(value) => filter::evaluate_predicate(predicate, value),
408        None => false,
409    }
410}
411
412/// Evaluate a simple expression (for expression aggregates)
413#[allow(clippy::only_used_in_recursion)]
414fn eval_simple_expression(row: &Row, expr: &vibesql_ast::Expression) -> Option<f64> {
415    use vibesql_ast::{BinaryOperator, Expression};
416
417    match expr {
418        Expression::BinaryOp { left, op, right } => {
419            let left_val = eval_simple_expression(row, left)?;
420            let right_val = eval_simple_expression(row, right)?;
421            match op {
422                BinaryOperator::Multiply => Some(left_val * right_val),
423                BinaryOperator::Divide => Some(left_val / right_val),
424                BinaryOperator::Plus => Some(left_val + right_val),
425                BinaryOperator::Minus => Some(left_val - right_val),
426                _ => None,
427            }
428        }
429        Expression::ColumnRef(col_id) => {
430            // Cannot resolve column names without a schema - return None to skip
431            // the fast path and fall back to the columnar execution path which
432            // properly handles expression aggregates with schema resolution.
433            log::debug!(
434                "fast_aggregate_on_rows: ColumnRef '{}' requires schema resolution, skipping fast path",
435                col_id.column_canonical()
436            );
437            None
438        }
439        Expression::Literal(val) => match val {
440            SqlValue::Integer(v) => Some(*v as f64),
441            SqlValue::Double(v) => Some(*v),
442            SqlValue::Float(v) => Some(*v as f64),
443            SqlValue::Bigint(v) => Some(*v as f64),
444            SqlValue::Numeric(v) => Some(*v),
445            _ => None,
446        },
447        _ => None,
448    }
449}
450
451/// Execute a query using columnar processing (AST-based interface)
452///
453/// This is the entry point for columnar execution that accepts AST expressions
454/// and converts them to the columnar execution pipeline.
455///
456/// # Arguments
457///
458/// * `rows` - The rows to process
459/// * `filter` - Optional WHERE clause expression
460/// * `aggregates` - SELECT list aggregate expressions
461/// * `schema` - Schema for resolving column names to indices
462///
463/// # Returns
464///
465/// Some(Result) if the query can be optimized using columnar execution,
466/// None if the expressions are too complex for columnar optimization.
467///
468/// Note: This function uses LLVM auto-vectorization for vectorized execution.
469pub fn execute_columnar(
470    rows: &[Row],
471    filter: Option<&vibesql_ast::Expression>,
472    aggregates: &[vibesql_ast::Expression],
473    schema: &CombinedSchema,
474    case_sensitive_like: bool,
475) -> Option<Result<Vec<Row>, ExecutorError>> {
476    log::debug!("  Executing columnar query with {} rows", rows.len());
477
478    // Extract column predicates from filter expression
479    let predicates = if let Some(filter_expr) = filter {
480        match extract_column_predicates(filter_expr, schema, case_sensitive_like) {
481            Some(preds) => {
482                log::debug!("    ✓ Extracted {} column predicates for SIMD filtering", preds.len());
483                preds
484            }
485            None => {
486                log::debug!("    ✗ Filter too complex for columnar optimization");
487                return None; // Too complex for columnar optimization
488            }
489        }
490    } else {
491        log::debug!("    No filter predicates");
492        vec![] // No filter
493    };
494
495    // Extract aggregates from SELECT list
496    let agg_specs = match extract_aggregates(aggregates, schema) {
497        Some(specs) => {
498            log::debug!("    ✓ Extracted {} aggregate operations", specs.len());
499            for (i, spec) in specs.iter().enumerate() {
500                log::debug!("      Aggregate {}: {:?}", i + 1, spec.op);
501            }
502            specs
503        }
504        None => {
505            log::debug!("    ✗ Aggregates too complex for columnar optimization");
506            return None; // Too complex for columnar optimization
507        }
508    };
509
510    // Call the simplified interface, passing schema if any aggregates use expressions
511    let needs_schema = agg_specs
512        .iter()
513        .any(|spec| matches!(spec.source, aggregate::AggregateSource::Expression(_)));
514    let schema_ref = if needs_schema { Some(schema) } else { None };
515
516    log::debug!("    Executing SIMD-accelerated columnar aggregation");
517    Some(execute_columnar_aggregate(rows, &predicates, &agg_specs, schema_ref))
518}
519
520#[cfg(test)]
521mod tests {
522    use vibesql_types::Date;
523
524    use super::*;
525
526    /// Test the full columnar pipeline: filter + aggregation
527    #[test]
528    fn test_columnar_pipeline_filtered_sum() {
529        // Create test data: TPC-H Q6 style query
530        // SELECT SUM(l_extendedprice * l_discount)
531        // WHERE l_shipdate >= '1994-01-01'
532        //   AND l_shipdate < '1995-01-01'
533        //   AND l_discount BETWEEN 0.05 AND 0.07
534        //   AND l_quantity < 24
535
536        let rows = vec![
537            Row::new(vec![
538                SqlValue::Integer(10),   // quantity
539                SqlValue::Double(100.0), // extendedprice
540                SqlValue::Double(0.06),  // discount
541                SqlValue::Date(Date::new(1994, 6, 1).unwrap()),
542            ]),
543            Row::new(vec![
544                SqlValue::Integer(25), // quantity (filtered out: > 24)
545                SqlValue::Double(200.0),
546                SqlValue::Double(0.06),
547                SqlValue::Date(Date::new(1994, 7, 1).unwrap()),
548            ]),
549            Row::new(vec![
550                SqlValue::Integer(15), // quantity
551                SqlValue::Double(150.0),
552                SqlValue::Double(0.05), // discount
553                SqlValue::Date(Date::new(1994, 8, 1).unwrap()),
554            ]),
555            Row::new(vec![
556                SqlValue::Integer(20), // quantity
557                SqlValue::Double(180.0),
558                SqlValue::Double(0.08), // discount (filtered out: > 0.07)
559                SqlValue::Date(Date::new(1994, 9, 1).unwrap()),
560            ]),
561        ];
562
563        // Predicates: quantity < 24 AND discount BETWEEN 0.05 AND 0.07
564        let predicates = vec![
565            ColumnPredicate::LessThan { column_idx: 0, value: SqlValue::Integer(24) },
566            ColumnPredicate::Between {
567                column_idx: 2,
568                low: SqlValue::Double(0.05),
569                high: SqlValue::Double(0.07),
570            },
571        ];
572
573        // Aggregates: SUM(extendedprice), COUNT(*)
574        let aggregates = vec![
575            AggregateSpec { op: AggregateOp::Sum, source: AggregateSource::Column(1) }, /* SUM(extendedprice) */
576            AggregateSpec { op: AggregateOp::Count, source: AggregateSource::Column(0) }, /* COUNT(*) */
577        ];
578
579        let result = execute_columnar_aggregate(&rows, &predicates, &aggregates, None).unwrap();
580
581        assert_eq!(result.len(), 1);
582        let result_row = &result[0];
583
584        // Only rows 0 and 2 pass the filter (quantity < 24 AND discount in range)
585        // SUM(extendedprice) = 100.0 + 150.0 = 250.0
586        assert!(
587            matches!(result_row.get(0), Some(&SqlValue::Double(sum)) if (sum - 250.0).abs() < 0.001)
588        );
589        // COUNT(*) = 2
590        assert_eq!(result_row.get(1), Some(&SqlValue::Integer(2)));
591    }
592
593    /// Test columnar execution with no filtering
594    #[test]
595    fn test_columnar_pipeline_no_filter() {
596        let rows = vec![
597            Row::new(vec![SqlValue::Integer(10), SqlValue::Double(1.5)]),
598            Row::new(vec![SqlValue::Integer(20), SqlValue::Double(2.5)]),
599            Row::new(vec![SqlValue::Integer(30), SqlValue::Double(3.5)]),
600        ];
601
602        let predicates = vec![];
603        let aggregates = vec![
604            AggregateSpec { op: AggregateOp::Sum, source: AggregateSource::Column(0) },
605            AggregateSpec { op: AggregateOp::Avg, source: AggregateSource::Column(1) },
606            AggregateSpec { op: AggregateOp::Max, source: AggregateSource::Column(0) },
607        ];
608
609        let result = execute_columnar_aggregate(&rows, &predicates, &aggregates, None).unwrap();
610
611        assert_eq!(result.len(), 1);
612        let result_row = &result[0];
613
614        // SUM(col0) = 60 (SQLite's SUM() returns REAL)
615        assert_eq!(result_row.get(0), Some(&SqlValue::Integer(60)));
616        // AVG(col1) = 2.5
617        assert!(
618            matches!(result_row.get(1), Some(&SqlValue::Double(avg)) if (avg - 2.5).abs() < 0.001)
619        );
620        // MAX(col0) = 30
621        assert_eq!(result_row.get(2), Some(&SqlValue::Integer(30)));
622    }
623
624    /// Test columnar execution with empty result set
625    #[test]
626    fn test_columnar_pipeline_empty_result() {
627        let rows =
628            vec![Row::new(vec![SqlValue::Integer(100)]), Row::new(vec![SqlValue::Integer(200)])];
629
630        // Filter that matches nothing
631        let predicates =
632            vec![ColumnPredicate::LessThan { column_idx: 0, value: SqlValue::Integer(50) }];
633
634        let aggregates = vec![
635            AggregateSpec { op: AggregateOp::Sum, source: AggregateSource::Column(0) },
636            AggregateSpec { op: AggregateOp::Count, source: AggregateSource::Column(0) },
637        ];
638
639        let result = execute_columnar_aggregate(&rows, &predicates, &aggregates, None).unwrap();
640
641        assert_eq!(result.len(), 1);
642        let result_row = &result[0];
643
644        // SUM of empty set = NULL
645        assert_eq!(result_row.get(0), Some(&SqlValue::Null));
646        // COUNT of empty set = 0
647        assert_eq!(result_row.get(1), Some(&SqlValue::Integer(0)));
648    }
649
650    // AST Integration Tests
651
652    use vibesql_ast::{BinaryOperator, Expression};
653    use vibesql_catalog::{ColumnSchema, TableSchema};
654    use vibesql_types::DataType;
655
656    use crate::schema::CombinedSchema;
657
658    fn make_test_schema() -> CombinedSchema {
659        let schema = TableSchema::new(
660            "test".to_string(),
661            vec![
662                ColumnSchema::new("quantity".to_string(), DataType::Integer, false),
663                ColumnSchema::new("price".to_string(), DataType::DoublePrecision, false),
664            ],
665        );
666        CombinedSchema::from_table("test".to_string(), schema)
667    }
668
669    fn make_test_rows_for_ast() -> Vec<Row> {
670        vec![
671            Row::new(vec![SqlValue::Integer(10), SqlValue::Double(1.5)]),
672            Row::new(vec![SqlValue::Integer(20), SqlValue::Double(2.5)]),
673            Row::new(vec![SqlValue::Integer(30), SqlValue::Double(3.5)]),
674            Row::new(vec![SqlValue::Integer(40), SqlValue::Double(4.5)]),
675        ]
676    }
677
678    #[test]
679    fn test_execute_columnar_simple_aggregate() {
680        let rows = make_test_rows_for_ast();
681        let schema = make_test_schema();
682
683        // SELECT SUM(price) FROM test
684        let aggregates = vec![Expression::AggregateFunction {
685            name: vibesql_ast::FunctionIdentifier::new("SUM"),
686            distinct: false,
687            args: vec![Expression::ColumnRef(vibesql_ast::ColumnIdentifier::simple(
688                "price", false,
689            ))],
690            order_by: None,
691            filter: None,
692        }];
693
694        let result = execute_columnar(&rows, None, &aggregates, &schema, false);
695        assert!(result.is_some());
696
697        let rows_result = result.unwrap();
698        assert!(rows_result.is_ok());
699
700        let result_rows = rows_result.unwrap();
701        assert_eq!(result_rows.len(), 1);
702        assert_eq!(result_rows[0].len(), 1);
703
704        // Sum should be 1.5 + 2.5 + 3.5 + 4.5 = 12.0
705        if let Some(SqlValue::Double(sum)) = result_rows[0].get(0) {
706            assert!((sum - 12.0).abs() < 0.001);
707        } else {
708            panic!("Expected Numeric value for SUM");
709        }
710    }
711
712    #[test]
713    fn test_execute_columnar_with_filter() {
714        let rows = make_test_rows_for_ast();
715        let schema = make_test_schema();
716
717        // SELECT SUM(price) FROM test WHERE quantity < 25
718        let filter = Expression::BinaryOp {
719            left: Box::new(Expression::ColumnRef(vibesql_ast::ColumnIdentifier::simple(
720                "quantity", false,
721            ))),
722            op: BinaryOperator::LessThan,
723            right: Box::new(Expression::Literal(SqlValue::Integer(25))),
724        };
725
726        let aggregates = vec![Expression::AggregateFunction {
727            name: vibesql_ast::FunctionIdentifier::new("SUM"),
728            distinct: false,
729            args: vec![Expression::ColumnRef(vibesql_ast::ColumnIdentifier::simple(
730                "price", false,
731            ))],
732            order_by: None,
733            filter: None,
734        }];
735
736        let result = execute_columnar(&rows, Some(&filter), &aggregates, &schema, false);
737        assert!(result.is_some());
738
739        let rows_result = result.unwrap();
740        assert!(rows_result.is_ok());
741
742        let result_rows = rows_result.unwrap();
743        assert_eq!(result_rows.len(), 1);
744        assert_eq!(result_rows[0].len(), 1);
745
746        // Sum of rows where quantity < 25: 1.5 + 2.5 = 4.0
747        if let Some(SqlValue::Double(sum)) = result_rows[0].get(0) {
748            assert!((sum - 4.0).abs() < 0.001);
749        } else {
750            panic!("Expected Numeric value for SUM");
751        }
752    }
753
754    #[test]
755    fn test_execute_columnar_multiple_aggregates() {
756        let rows = make_test_rows_for_ast();
757        let schema = make_test_schema();
758
759        // SELECT SUM(price), COUNT(*), AVG(quantity) FROM test
760        let aggregates = vec![
761            Expression::AggregateFunction {
762                name: vibesql_ast::FunctionIdentifier::new("SUM"),
763                distinct: false,
764                args: vec![Expression::ColumnRef(vibesql_ast::ColumnIdentifier::simple(
765                    "price", false,
766                ))],
767                order_by: None,
768                filter: None,
769            },
770            Expression::AggregateFunction {
771                name: vibesql_ast::FunctionIdentifier::new("COUNT"),
772                distinct: false,
773                args: vec![Expression::Wildcard],
774                order_by: None,
775                filter: None,
776            },
777            Expression::AggregateFunction {
778                name: vibesql_ast::FunctionIdentifier::new("AVG"),
779                distinct: false,
780                args: vec![Expression::ColumnRef(vibesql_ast::ColumnIdentifier::simple(
781                    "quantity", false,
782                ))],
783                order_by: None,
784                filter: None,
785            },
786        ];
787
788        let result = execute_columnar(&rows, None, &aggregates, &schema, false);
789        assert!(result.is_some());
790
791        let rows_result = result.unwrap();
792        assert!(rows_result.is_ok());
793
794        let result_rows = rows_result.unwrap();
795        assert_eq!(result_rows.len(), 1);
796        assert_eq!(result_rows[0].len(), 3);
797
798        // Check SUM(price) = 12.0
799        if let Some(SqlValue::Double(sum)) = result_rows[0].get(0) {
800            assert!((sum - 12.0).abs() < 0.001);
801        } else {
802            panic!("Expected Numeric value for SUM");
803        }
804
805        // Check COUNT(*) = 4
806        assert_eq!(result_rows[0].get(1), Some(&SqlValue::Integer(4)));
807
808        // Check AVG(quantity) = (10+20+30+40)/4 = 25.0
809        if let Some(SqlValue::Double(avg)) = result_rows[0].get(2) {
810            assert!((avg - 25.0).abs() < 0.001);
811        } else {
812            panic!("Expected Double value for AVG");
813        }
814    }
815
816    #[test]
817    fn test_execute_columnar_unsupported_distinct() {
818        let rows = make_test_rows_for_ast();
819        let schema = make_test_schema();
820
821        // SELECT SUM(DISTINCT price) FROM test - should return None
822        let aggregates = vec![Expression::AggregateFunction {
823            name: vibesql_ast::FunctionIdentifier::new("SUM"),
824            distinct: true,
825            args: vec![Expression::ColumnRef(vibesql_ast::ColumnIdentifier::simple(
826                "price", false,
827            ))],
828            order_by: None,
829            filter: None,
830        }];
831
832        let result = execute_columnar(&rows, None, &aggregates, &schema, false);
833        assert!(result.is_none());
834    }
835
836    #[test]
837    fn test_execute_columnar_unsupported_complex_filter() {
838        let rows = make_test_rows_for_ast();
839        let schema = make_test_schema();
840
841        // SELECT SUM(price) FROM test WHERE quantity IN (SELECT ...) - unsupported
842        let filter = Expression::ScalarSubquery(Box::new(vibesql_ast::SelectStmt {
843            with_clause: None,
844            distinct: false,
845            select_list: vec![],
846            into_table: None,
847            into_variables: None,
848            from: None,
849            where_clause: None,
850            group_by: None,
851            having: None,
852            window_definitions: None,
853            order_by: None,
854            limit: None,
855            offset: None,
856            set_operation: None,
857            values: None,
858        }));
859
860        let aggregates = vec![Expression::AggregateFunction {
861            name: vibesql_ast::FunctionIdentifier::new("SUM"),
862            distinct: false,
863            args: vec![Expression::ColumnRef(vibesql_ast::ColumnIdentifier::simple(
864                "price", false,
865            ))],
866            order_by: None,
867            filter: None,
868        }];
869
870        let result = execute_columnar(&rows, Some(&filter), &aggregates, &schema, false);
871        assert!(result.is_none());
872    }
873}