#![allow(clippy::needless_range_loop, clippy::unnecessary_map_or)]
mod aggregate;
mod expression;
mod fused;
use aggregate::compute_batch_aggregates;
use fused::{can_use_fused_aggregation, execute_fused_filter_aggregate};
use vibesql_storage::Row;
use vibesql_types::SqlValue;
use super::{
aggregate::{AggregateOp, AggregateSpec},
batch::ColumnarBatch,
filter::ColumnPredicate,
simd_filter::simd_filter_batch,
};
use crate::{errors::ExecutorError, schema::CombinedSchema};
pub fn execute_columnar_batch(
batch: &ColumnarBatch,
predicates: &[ColumnPredicate],
aggregates: &[AggregateSpec],
_schema: Option<&CombinedSchema>,
) -> Result<Vec<Row>, ExecutorError> {
if batch.row_count() == 0 {
let values: Vec<SqlValue> = aggregates
.iter()
.map(|spec| match spec.op {
AggregateOp::Count => SqlValue::Integer(0),
_ => SqlValue::Null,
})
.collect();
return Ok(vec![Row::new(values)]);
}
if !predicates.is_empty() && can_use_fused_aggregation(aggregates) {
if let Ok(results) = execute_fused_filter_aggregate(batch, predicates, aggregates) {
return Ok(vec![Row::new(results)]);
}
}
#[cfg(feature = "profile-q6")]
let filter_start = std::time::Instant::now();
let filtered_batch =
if predicates.is_empty() { batch.clone() } else { simd_filter_batch(batch, predicates)? };
#[cfg(feature = "profile-q6")]
{
let filter_time = filter_start.elapsed();
eprintln!(
"[PROFILE-Q6] Phase 1 - SIMD Filter: {:?} ({}/{} rows passed)",
filter_time,
filtered_batch.row_count(),
batch.row_count()
);
}
#[cfg(feature = "profile-q6")]
let agg_start = std::time::Instant::now();
let results = compute_batch_aggregates(&filtered_batch, aggregates)?;
#[cfg(feature = "profile-q6")]
{
let agg_time = agg_start.elapsed();
eprintln!(
"[PROFILE-Q6] Phase 2 - SIMD Aggregate: {:?} ({} aggregates)",
agg_time,
aggregates.len()
);
}
Ok(vec![Row::new(results)])
}
#[cfg(test)]
mod tests {
use super::{
super::{aggregate::AggregateSource, batch::ColumnarBatch},
*,
};
fn make_test_batch() -> ColumnarBatch {
let rows = vec![
Row::new(vec![SqlValue::Integer(10), SqlValue::Double(1.5)]),
Row::new(vec![SqlValue::Integer(20), SqlValue::Double(2.5)]),
Row::new(vec![SqlValue::Integer(30), SqlValue::Double(3.5)]),
Row::new(vec![SqlValue::Integer(40), SqlValue::Double(4.5)]),
];
ColumnarBatch::from_rows(&rows).unwrap()
}
#[test]
fn test_execute_columnar_batch_sum() {
let batch = make_test_batch();
let aggregates =
vec![AggregateSpec { op: AggregateOp::Sum, source: AggregateSource::Column(0) }];
let result = execute_columnar_batch(&batch, &[], &aggregates, None).unwrap();
assert_eq!(result.len(), 1);
assert_eq!(result[0].get(0), Some(&SqlValue::Integer(100)));
}
#[test]
fn test_execute_columnar_batch_with_filter() {
let batch = make_test_batch();
let predicates =
vec![ColumnPredicate::LessThan { column_idx: 0, value: SqlValue::Integer(25) }];
let aggregates =
vec![AggregateSpec { op: AggregateOp::Sum, source: AggregateSource::Column(0) }];
let result = execute_columnar_batch(&batch, &predicates, &aggregates, None).unwrap();
assert_eq!(result.len(), 1);
assert_eq!(result[0].get(0), Some(&SqlValue::Integer(30)));
}
#[test]
fn test_execute_columnar_batch_multiple_aggregates() {
let batch = make_test_batch();
let aggregates = vec![
AggregateSpec { op: AggregateOp::Sum, source: AggregateSource::Column(0) },
AggregateSpec { op: AggregateOp::Avg, source: AggregateSource::Column(1) },
AggregateSpec { op: AggregateOp::Count, source: AggregateSource::CountStar },
];
let result = execute_columnar_batch(&batch, &[], &aggregates, None).unwrap();
assert_eq!(result.len(), 1);
assert_eq!(result[0].len(), 3);
assert_eq!(result[0].get(0), Some(&SqlValue::Integer(100)));
if let Some(SqlValue::Double(avg)) = result[0].get(1) {
assert!((avg - 3.0).abs() < 0.001);
} else {
panic!("Expected Double for AVG");
}
assert_eq!(result[0].get(2), Some(&SqlValue::Integer(4)));
}
#[test]
fn test_execute_columnar_batch_empty() {
let batch = ColumnarBatch::new(2);
let aggregates = vec![
AggregateSpec { op: AggregateOp::Sum, source: AggregateSource::Column(0) },
AggregateSpec { op: AggregateOp::Count, source: AggregateSource::CountStar },
];
let result = execute_columnar_batch(&batch, &[], &aggregates, None).unwrap();
assert_eq!(result.len(), 1);
assert_eq!(result[0].get(0), Some(&SqlValue::Null)); assert_eq!(result[0].get(1), Some(&SqlValue::Integer(0))); }
#[test]
fn test_execute_columnar_batch_with_nulls() {
let rows = vec![
Row::new(vec![SqlValue::Integer(10), SqlValue::Double(1.5)]),
Row::new(vec![SqlValue::Null, SqlValue::Double(2.5)]), Row::new(vec![SqlValue::Integer(30), SqlValue::Null]), Row::new(vec![SqlValue::Integer(40), SqlValue::Double(4.5)]),
];
let batch = ColumnarBatch::from_rows(&rows).unwrap();
let aggregates = vec![
AggregateSpec { op: AggregateOp::Sum, source: AggregateSource::Column(0) },
AggregateSpec { op: AggregateOp::Sum, source: AggregateSource::Column(1) },
AggregateSpec { op: AggregateOp::Count, source: AggregateSource::CountStar },
];
let result = execute_columnar_batch(&batch, &[], &aggregates, None).unwrap();
assert_eq!(result.len(), 1);
assert_eq!(result[0].get(0), Some(&SqlValue::Integer(80)));
if let Some(SqlValue::Double(sum)) = result[0].get(1) {
assert!((sum - 8.5).abs() < 0.001);
} else {
panic!("Expected Double for SUM(col1)");
}
assert_eq!(result[0].get(2), Some(&SqlValue::Integer(4)));
}
#[test]
fn test_execute_columnar_batch_with_nulls_and_filter() {
let rows = vec![
Row::new(vec![SqlValue::Integer(10), SqlValue::Double(1.0)]),
Row::new(vec![SqlValue::Integer(20), SqlValue::Null]),
Row::new(vec![SqlValue::Integer(30), SqlValue::Double(3.0)]),
Row::new(vec![SqlValue::Integer(40), SqlValue::Double(4.0)]),
];
let batch = ColumnarBatch::from_rows(&rows).unwrap();
let predicates =
vec![ColumnPredicate::LessThan { column_idx: 0, value: SqlValue::Integer(35) }];
let aggregates =
vec![AggregateSpec { op: AggregateOp::Sum, source: AggregateSource::Column(1) }];
let result = execute_columnar_batch(&batch, &predicates, &aggregates, None).unwrap();
assert_eq!(result.len(), 1);
if let Some(SqlValue::Double(sum)) = result[0].get(0) {
assert!((sum - 4.0).abs() < 0.001);
} else {
panic!("Expected Double for SUM(col1)");
}
}
}