mod aggregate;
pub mod batch;
mod executor;
pub mod filter;
mod scan;
mod string_ops;
pub mod simd_ops;
mod simd_aggregate;
pub mod simd_filter;
mod simd_join;
pub use aggregate::{
columnar_group_by, columnar_group_by_batch, compute_aggregates_from_batch,
compute_multiple_aggregates, evaluate_expression_to_column,
evaluate_expression_with_cached_column, extract_aggregates, AggregateOp, AggregateSource,
AggregateSpec,
};
pub use batch::{ColumnArray, ColumnarBatch};
pub use executor::execute_columnar_batch;
pub use filter::{
apply_columnar_filter, apply_columnar_filter_simd_streaming, create_filter_bitmap,
create_filter_bitmap_tree, evaluate_predicate_tree, extract_column_predicates,
extract_predicate_tree, ColumnPredicate, PredicateTree,
};
use log;
pub use scan::ColumnarScan;
pub use simd_aggregate::{can_use_simd_for_column, simd_aggregate_f64, simd_aggregate_i64};
pub use simd_filter::{
simd_create_filter_mask, simd_create_filter_mask_auto, simd_create_filter_mask_packed,
simd_filter_batch, simd_filter_to_indices,
};
#[cfg(feature = "parallel")]
pub use simd_filter::{simd_create_filter_mask_parallel, simd_filter_batch_parallel};
pub use simd_join::columnar_hash_join_inner;
pub use simd_ops::PackedMask;
use vibesql_storage::Row;
use vibesql_types::SqlValue;
use crate::{errors::ExecutorError, schema::CombinedSchema};
pub fn execute_columnar_aggregate(
rows: &[Row],
predicates: &[ColumnPredicate],
aggregates: &[aggregate::AggregateSpec],
schema: Option<&CombinedSchema>,
) -> Result<Vec<Row>, ExecutorError> {
if rows.is_empty() {
let values: Vec<SqlValue> = aggregates
.iter()
.map(|spec| match spec.op {
aggregate::AggregateOp::Count => SqlValue::Integer(0),
_ => SqlValue::Null,
})
.collect();
return Ok(vec![Row::new(values)]);
}
#[cfg(feature = "profile-q6")]
let batch_start = std::time::Instant::now();
let batch = ColumnarBatch::from_rows(rows)?;
#[cfg(feature = "profile-q6")]
{
let batch_time = batch_start.elapsed();
eprintln!("[PROFILE-Q6] Phase 1 - Convert to batch: {:?}", batch_time);
}
#[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 2 - SIMD filter: {:?} ({}/{} rows passed)",
filter_time,
filtered_batch.row_count(),
rows.len()
);
}
#[cfg(feature = "profile-q6")]
let agg_start = std::time::Instant::now();
let results = compute_aggregates_from_batch(&filtered_batch, aggregates, schema)?;
#[cfg(feature = "profile-q6")]
{
let agg_time = agg_start.elapsed();
eprintln!(
"[PROFILE-Q6] Phase 3 - Batch-native aggregate: {:?} ({} aggregates)",
agg_time,
aggregates.len()
);
}
Ok(vec![Row::new(results)])
}
pub fn fast_aggregate_on_rows(
rows: &[Row],
predicates: &[ColumnPredicate],
aggregates: &[aggregate::AggregateSpec],
) -> Result<Vec<Row>, ExecutorError> {
use aggregate::{AggregateOp, AggregateSource};
if rows.is_empty() {
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)]);
}
struct Accumulator {
sum_f64: f64,
sum_i64: i64,
count: i64,
min_f64: Option<f64>,
max_f64: Option<f64>,
min_i64: Option<i64>,
max_i64: Option<i64>,
is_integer: bool,
}
let mut accumulators: Vec<Accumulator> = aggregates
.iter()
.map(|_| Accumulator {
sum_f64: 0.0,
sum_i64: 0,
count: 0,
min_f64: None,
max_f64: None,
min_i64: None,
max_i64: None,
is_integer: true,
})
.collect();
for row in rows {
let passes_filter = predicates.iter().all(|pred| evaluate_predicate(row, pred));
if !passes_filter {
continue;
}
for (i, spec) in aggregates.iter().enumerate() {
let acc = &mut accumulators[i];
match &spec.source {
AggregateSource::CountStar => {
acc.count += 1;
}
AggregateSource::Column(col_idx) => {
if let Some(value) = row.get(*col_idx) {
if !matches!(value, SqlValue::Null) {
acc.count += 1;
match value {
SqlValue::Integer(v) => {
acc.sum_i64 += v;
acc.sum_f64 += *v as f64;
acc.min_i64 = Some(acc.min_i64.map_or(*v, |m| m.min(*v)));
acc.max_i64 = Some(acc.max_i64.map_or(*v, |m| m.max(*v)));
acc.min_f64 =
Some(acc.min_f64.map_or(*v as f64, |m| m.min(*v as f64)));
acc.max_f64 =
Some(acc.max_f64.map_or(*v as f64, |m| m.max(*v as f64)));
}
SqlValue::Double(v) => {
acc.is_integer = false;
acc.sum_f64 += v;
acc.min_f64 = Some(acc.min_f64.map_or(*v, |m| m.min(*v)));
acc.max_f64 = Some(acc.max_f64.map_or(*v, |m| m.max(*v)));
}
SqlValue::Float(v) => {
acc.is_integer = false;
acc.sum_f64 += *v as f64;
acc.min_f64 =
Some(acc.min_f64.map_or(*v as f64, |m| m.min(*v as f64)));
acc.max_f64 =
Some(acc.max_f64.map_or(*v as f64, |m| m.max(*v as f64)));
}
SqlValue::Bigint(v) => {
acc.sum_i64 += v;
acc.sum_f64 += *v as f64;
acc.min_i64 = Some(acc.min_i64.map_or(*v, |m| m.min(*v)));
acc.max_i64 = Some(acc.max_i64.map_or(*v, |m| m.max(*v)));
acc.min_f64 =
Some(acc.min_f64.map_or(*v as f64, |m| m.min(*v as f64)));
acc.max_f64 =
Some(acc.max_f64.map_or(*v as f64, |m| m.max(*v as f64)));
}
SqlValue::Numeric(v) => {
acc.is_integer = false;
acc.sum_f64 += v;
acc.min_f64 = Some(acc.min_f64.map_or(*v, |m| m.min(*v)));
acc.max_f64 = Some(acc.max_f64.map_or(*v, |m| m.max(*v)));
}
_ => {}
}
}
}
}
AggregateSource::Expression(expr) => {
if let Some(value) = eval_simple_expression(row, expr) {
acc.count += 1;
acc.is_integer = false;
acc.sum_f64 += value;
acc.min_f64 = Some(acc.min_f64.map_or(value, |m| m.min(value)));
acc.max_f64 = Some(acc.max_f64.map_or(value, |m| m.max(value)));
}
}
}
}
}
let values: Vec<SqlValue> = aggregates
.iter()
.zip(accumulators.iter())
.map(|(spec, acc)| match spec.op {
AggregateOp::Count => SqlValue::Integer(acc.count),
AggregateOp::Sum => {
if acc.count == 0 {
SqlValue::Null
} else if acc.is_integer {
SqlValue::Integer(acc.sum_i64)
} else {
SqlValue::Double(acc.sum_f64)
}
}
AggregateOp::Avg => {
if acc.count == 0 {
SqlValue::Null
} else {
SqlValue::Double(acc.sum_f64 / acc.count as f64)
}
}
AggregateOp::Min => {
if acc.is_integer {
acc.min_i64.map(SqlValue::Integer).unwrap_or(SqlValue::Null)
} else {
acc.min_f64.map(SqlValue::Double).unwrap_or(SqlValue::Null)
}
}
AggregateOp::Max => {
if acc.is_integer {
acc.max_i64.map(SqlValue::Integer).unwrap_or(SqlValue::Null)
} else {
acc.max_f64.map(SqlValue::Double).unwrap_or(SqlValue::Null)
}
}
})
.collect();
Ok(vec![Row::new(values)])
}
fn evaluate_predicate(row: &Row, predicate: &ColumnPredicate) -> bool {
if let ColumnPredicate::ColumnCompare { left_column_idx, op, right_column_idx } = predicate {
return filter::evaluate_column_compare(
*op,
row.get(*left_column_idx),
row.get(*right_column_idx),
);
}
let column_idx = match predicate {
ColumnPredicate::LessThan { column_idx, .. }
| ColumnPredicate::GreaterThan { column_idx, .. }
| ColumnPredicate::GreaterThanOrEqual { column_idx, .. }
| ColumnPredicate::LessThanOrEqual { column_idx, .. }
| ColumnPredicate::Equal { column_idx, .. }
| ColumnPredicate::NotEqual { column_idx, .. }
| ColumnPredicate::Between { column_idx, .. }
| ColumnPredicate::Like { column_idx, .. }
| ColumnPredicate::InList { column_idx, .. } => *column_idx,
ColumnPredicate::ColumnCompare { .. } => unreachable!(), };
match row.get(column_idx) {
Some(value) => filter::evaluate_predicate(predicate, value),
None => false,
}
}
#[allow(clippy::only_used_in_recursion)]
fn eval_simple_expression(row: &Row, expr: &vibesql_ast::Expression) -> Option<f64> {
use vibesql_ast::{BinaryOperator, Expression};
match expr {
Expression::BinaryOp { left, op, right } => {
let left_val = eval_simple_expression(row, left)?;
let right_val = eval_simple_expression(row, right)?;
match op {
BinaryOperator::Multiply => Some(left_val * right_val),
BinaryOperator::Divide => Some(left_val / right_val),
BinaryOperator::Plus => Some(left_val + right_val),
BinaryOperator::Minus => Some(left_val - right_val),
_ => None,
}
}
Expression::ColumnRef(col_id) => {
log::debug!(
"fast_aggregate_on_rows: ColumnRef '{}' requires schema resolution, skipping fast path",
col_id.column_canonical()
);
None
}
Expression::Literal(val) => match val {
SqlValue::Integer(v) => Some(*v as f64),
SqlValue::Double(v) => Some(*v),
SqlValue::Float(v) => Some(*v as f64),
SqlValue::Bigint(v) => Some(*v as f64),
SqlValue::Numeric(v) => Some(*v),
_ => None,
},
_ => None,
}
}
pub fn execute_columnar(
rows: &[Row],
filter: Option<&vibesql_ast::Expression>,
aggregates: &[vibesql_ast::Expression],
schema: &CombinedSchema,
case_sensitive_like: bool,
) -> Option<Result<Vec<Row>, ExecutorError>> {
log::debug!(" Executing columnar query with {} rows", rows.len());
let predicates = if let Some(filter_expr) = filter {
match extract_column_predicates(filter_expr, schema, case_sensitive_like) {
Some(preds) => {
log::debug!(" ✓ Extracted {} column predicates for SIMD filtering", preds.len());
preds
}
None => {
log::debug!(" ✗ Filter too complex for columnar optimization");
return None; }
}
} else {
log::debug!(" No filter predicates");
vec![] };
let agg_specs = match extract_aggregates(aggregates, schema) {
Some(specs) => {
log::debug!(" ✓ Extracted {} aggregate operations", specs.len());
for (i, spec) in specs.iter().enumerate() {
log::debug!(" Aggregate {}: {:?}", i + 1, spec.op);
}
specs
}
None => {
log::debug!(" ✗ Aggregates too complex for columnar optimization");
return None; }
};
let needs_schema = agg_specs
.iter()
.any(|spec| matches!(spec.source, aggregate::AggregateSource::Expression(_)));
let schema_ref = if needs_schema { Some(schema) } else { None };
log::debug!(" Executing SIMD-accelerated columnar aggregation");
Some(execute_columnar_aggregate(rows, &predicates, &agg_specs, schema_ref))
}
#[cfg(test)]
mod tests {
use vibesql_types::Date;
use super::*;
#[test]
fn test_columnar_pipeline_filtered_sum() {
let rows = vec![
Row::new(vec![
SqlValue::Integer(10), SqlValue::Double(100.0), SqlValue::Double(0.06), SqlValue::Date(Date::new(1994, 6, 1).unwrap()),
]),
Row::new(vec![
SqlValue::Integer(25), SqlValue::Double(200.0),
SqlValue::Double(0.06),
SqlValue::Date(Date::new(1994, 7, 1).unwrap()),
]),
Row::new(vec![
SqlValue::Integer(15), SqlValue::Double(150.0),
SqlValue::Double(0.05), SqlValue::Date(Date::new(1994, 8, 1).unwrap()),
]),
Row::new(vec![
SqlValue::Integer(20), SqlValue::Double(180.0),
SqlValue::Double(0.08), SqlValue::Date(Date::new(1994, 9, 1).unwrap()),
]),
];
let predicates = vec![
ColumnPredicate::LessThan { column_idx: 0, value: SqlValue::Integer(24) },
ColumnPredicate::Between {
column_idx: 2,
low: SqlValue::Double(0.05),
high: SqlValue::Double(0.07),
},
];
let aggregates = vec![
AggregateSpec { op: AggregateOp::Sum, source: AggregateSource::Column(1) },
AggregateSpec { op: AggregateOp::Count, source: AggregateSource::Column(0) },
];
let result = execute_columnar_aggregate(&rows, &predicates, &aggregates, None).unwrap();
assert_eq!(result.len(), 1);
let result_row = &result[0];
assert!(
matches!(result_row.get(0), Some(&SqlValue::Double(sum)) if (sum - 250.0).abs() < 0.001)
);
assert_eq!(result_row.get(1), Some(&SqlValue::Integer(2)));
}
#[test]
fn test_columnar_pipeline_no_filter() {
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)]),
];
let predicates = vec![];
let aggregates = vec![
AggregateSpec { op: AggregateOp::Sum, source: AggregateSource::Column(0) },
AggregateSpec { op: AggregateOp::Avg, source: AggregateSource::Column(1) },
AggregateSpec { op: AggregateOp::Max, source: AggregateSource::Column(0) },
];
let result = execute_columnar_aggregate(&rows, &predicates, &aggregates, None).unwrap();
assert_eq!(result.len(), 1);
let result_row = &result[0];
assert_eq!(result_row.get(0), Some(&SqlValue::Integer(60)));
assert!(
matches!(result_row.get(1), Some(&SqlValue::Double(avg)) if (avg - 2.5).abs() < 0.001)
);
assert_eq!(result_row.get(2), Some(&SqlValue::Integer(30)));
}
#[test]
fn test_columnar_pipeline_empty_result() {
let rows =
vec![Row::new(vec![SqlValue::Integer(100)]), Row::new(vec![SqlValue::Integer(200)])];
let predicates =
vec![ColumnPredicate::LessThan { column_idx: 0, value: SqlValue::Integer(50) }];
let aggregates = vec![
AggregateSpec { op: AggregateOp::Sum, source: AggregateSource::Column(0) },
AggregateSpec { op: AggregateOp::Count, source: AggregateSource::Column(0) },
];
let result = execute_columnar_aggregate(&rows, &predicates, &aggregates, None).unwrap();
assert_eq!(result.len(), 1);
let result_row = &result[0];
assert_eq!(result_row.get(0), Some(&SqlValue::Null));
assert_eq!(result_row.get(1), Some(&SqlValue::Integer(0)));
}
use vibesql_ast::{BinaryOperator, Expression};
use vibesql_catalog::{ColumnSchema, TableSchema};
use vibesql_types::DataType;
use crate::schema::CombinedSchema;
fn make_test_schema() -> CombinedSchema {
let schema = TableSchema::new(
"test".to_string(),
vec![
ColumnSchema::new("quantity".to_string(), DataType::Integer, false),
ColumnSchema::new("price".to_string(), DataType::DoublePrecision, false),
],
);
CombinedSchema::from_table("test".to_string(), schema)
}
fn make_test_rows_for_ast() -> Vec<Row> {
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)]),
]
}
#[test]
fn test_execute_columnar_simple_aggregate() {
let rows = make_test_rows_for_ast();
let schema = make_test_schema();
let aggregates = vec![Expression::AggregateFunction {
name: vibesql_ast::FunctionIdentifier::new("SUM"),
distinct: false,
args: vec![Expression::ColumnRef(vibesql_ast::ColumnIdentifier::simple(
"price", false,
))],
order_by: None,
filter: None,
}];
let result = execute_columnar(&rows, None, &aggregates, &schema, false);
assert!(result.is_some());
let rows_result = result.unwrap();
assert!(rows_result.is_ok());
let result_rows = rows_result.unwrap();
assert_eq!(result_rows.len(), 1);
assert_eq!(result_rows[0].len(), 1);
if let Some(SqlValue::Double(sum)) = result_rows[0].get(0) {
assert!((sum - 12.0).abs() < 0.001);
} else {
panic!("Expected Numeric value for SUM");
}
}
#[test]
fn test_execute_columnar_with_filter() {
let rows = make_test_rows_for_ast();
let schema = make_test_schema();
let filter = Expression::BinaryOp {
left: Box::new(Expression::ColumnRef(vibesql_ast::ColumnIdentifier::simple(
"quantity", false,
))),
op: BinaryOperator::LessThan,
right: Box::new(Expression::Literal(SqlValue::Integer(25))),
};
let aggregates = vec![Expression::AggregateFunction {
name: vibesql_ast::FunctionIdentifier::new("SUM"),
distinct: false,
args: vec![Expression::ColumnRef(vibesql_ast::ColumnIdentifier::simple(
"price", false,
))],
order_by: None,
filter: None,
}];
let result = execute_columnar(&rows, Some(&filter), &aggregates, &schema, false);
assert!(result.is_some());
let rows_result = result.unwrap();
assert!(rows_result.is_ok());
let result_rows = rows_result.unwrap();
assert_eq!(result_rows.len(), 1);
assert_eq!(result_rows[0].len(), 1);
if let Some(SqlValue::Double(sum)) = result_rows[0].get(0) {
assert!((sum - 4.0).abs() < 0.001);
} else {
panic!("Expected Numeric value for SUM");
}
}
#[test]
fn test_execute_columnar_multiple_aggregates() {
let rows = make_test_rows_for_ast();
let schema = make_test_schema();
let aggregates = vec![
Expression::AggregateFunction {
name: vibesql_ast::FunctionIdentifier::new("SUM"),
distinct: false,
args: vec![Expression::ColumnRef(vibesql_ast::ColumnIdentifier::simple(
"price", false,
))],
order_by: None,
filter: None,
},
Expression::AggregateFunction {
name: vibesql_ast::FunctionIdentifier::new("COUNT"),
distinct: false,
args: vec![Expression::Wildcard],
order_by: None,
filter: None,
},
Expression::AggregateFunction {
name: vibesql_ast::FunctionIdentifier::new("AVG"),
distinct: false,
args: vec![Expression::ColumnRef(vibesql_ast::ColumnIdentifier::simple(
"quantity", false,
))],
order_by: None,
filter: None,
},
];
let result = execute_columnar(&rows, None, &aggregates, &schema, false);
assert!(result.is_some());
let rows_result = result.unwrap();
assert!(rows_result.is_ok());
let result_rows = rows_result.unwrap();
assert_eq!(result_rows.len(), 1);
assert_eq!(result_rows[0].len(), 3);
if let Some(SqlValue::Double(sum)) = result_rows[0].get(0) {
assert!((sum - 12.0).abs() < 0.001);
} else {
panic!("Expected Numeric value for SUM");
}
assert_eq!(result_rows[0].get(1), Some(&SqlValue::Integer(4)));
if let Some(SqlValue::Double(avg)) = result_rows[0].get(2) {
assert!((avg - 25.0).abs() < 0.001);
} else {
panic!("Expected Double value for AVG");
}
}
#[test]
fn test_execute_columnar_unsupported_distinct() {
let rows = make_test_rows_for_ast();
let schema = make_test_schema();
let aggregates = vec![Expression::AggregateFunction {
name: vibesql_ast::FunctionIdentifier::new("SUM"),
distinct: true,
args: vec![Expression::ColumnRef(vibesql_ast::ColumnIdentifier::simple(
"price", false,
))],
order_by: None,
filter: None,
}];
let result = execute_columnar(&rows, None, &aggregates, &schema, false);
assert!(result.is_none());
}
#[test]
fn test_execute_columnar_unsupported_complex_filter() {
let rows = make_test_rows_for_ast();
let schema = make_test_schema();
let filter = Expression::ScalarSubquery(Box::new(vibesql_ast::SelectStmt {
with_clause: None,
distinct: false,
select_list: vec![],
into_table: None,
into_variables: None,
from: None,
where_clause: None,
group_by: None,
having: None,
window_definitions: None,
order_by: None,
limit: None,
offset: None,
set_operation: None,
values: None,
}));
let aggregates = vec![Expression::AggregateFunction {
name: vibesql_ast::FunctionIdentifier::new("SUM"),
distinct: false,
args: vec![Expression::ColumnRef(vibesql_ast::ColumnIdentifier::simple(
"price", false,
))],
order_by: None,
filter: None,
}];
let result = execute_columnar(&rows, Some(&filter), &aggregates, &schema, false);
assert!(result.is_none());
}
}