1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
//! Streaming aggregation for fast path execution
//!
//! This module provides ultra-fast execution for simple aggregate queries
//! that can accumulate results inline during a PK range scan.
use vibesql_ast::{SelectItem, SelectStmt};
use vibesql_storage::Row;
use vibesql_types::SqlValue;
use super::analysis::extract_simple_aggregate;
use crate::{
errors::ExecutorError,
select::{executor::builder::SelectExecutor, grouping::AggregateAccumulator},
};
impl SelectExecutor<'_> {
/// Execute a streaming aggregate query (#3815)
///
/// This provides ultra-fast execution for queries like:
/// `SELECT SUM(k) FROM sbtest1 WHERE id BETWEEN ? AND ?`
///
/// By accumulating aggregates inline during the PK range scan, we avoid:
/// - Materializing intermediate Row objects
/// - Going through the full pipeline infrastructure
/// - Multiple allocations per row
///
/// # Performance
/// This achieves SQLite-like performance (~4μs) compared to ~30μs
/// for the standard aggregation path.
pub fn execute_streaming_aggregate(
&self,
stmt: &SelectStmt,
) -> Result<Vec<Row>, ExecutorError> {
// Extract table name from FROM clause
let table_name = match &stmt.from {
Some(vibesql_ast::FromClause::Table { name, .. }) => name.as_str(),
_ => {
return Err(ExecutorError::Other(
"Streaming aggregate requires simple table FROM".to_string(),
))
}
};
// Get table
let table = match self.database.get_table(table_name) {
Some(t) => t,
None => {
return Err(ExecutorError::TableNotFound(table_name.to_string()));
}
};
// Need single-column PK for streaming
let pk_columns = match &table.schema.primary_key {
Some(cols) if cols.len() == 1 => cols,
_ => {
return Err(ExecutorError::Other(
"Streaming aggregate requires single-column PK".to_string(),
))
}
};
let pk_col = &pk_columns[0];
// Extract BETWEEN bounds from WHERE clause
let where_clause = stmt.where_clause.as_ref().ok_or_else(|| {
ExecutorError::Other("Streaming aggregate requires WHERE clause".to_string())
})?;
let (low_value, high_value) = match self.extract_between_bounds(where_clause, pk_col) {
Some(bounds) => bounds,
None => {
return Err(ExecutorError::Other(
"Streaming aggregate requires BETWEEN predicate on PK".to_string(),
))
}
};
// Find an index on the PK column
let index_names = self.database.list_indexes_for_table(table_name);
let pk_index_data = index_names.iter().find_map(|idx_name| {
let metadata = self.database.get_index(idx_name)?;
if metadata.columns.len() == 1
&& metadata.columns[0].expect_column_name().eq_ignore_ascii_case(pk_col)
// Partial indexes (CREATE INDEX ... WHERE expr) are usable only
// when the query WHERE clause structurally implies the index
// predicate: since PR #5323 the index body excludes
// predicate-false rows, so an ungated probe would silently
// drop matching rows (issue #5330).
&& crate::optimizer::predicate_implication::partial_index_usable(
self.database,
idx_name,
Some(where_clause),
)
{
self.database.get_index_data(idx_name)
} else {
None
}
});
let pk_index_data = match pk_index_data {
Some(idx) => idx,
None => {
return Err(ExecutorError::Other(
"Streaming aggregate requires index on PK".to_string(),
))
}
};
// Create accumulators for each aggregate in the SELECT list
let mut accumulators: Vec<(AggregateAccumulator, usize)> =
Vec::with_capacity(stmt.select_list.len());
for item in &stmt.select_list {
match item {
SelectItem::Expression { expr, .. } => {
let (func_name, col_idx) = extract_simple_aggregate(expr, &table.schema)
.ok_or_else(|| {
ExecutorError::Other(
"Streaming aggregate: invalid aggregate expression".to_string(),
)
})?;
let accumulator = AggregateAccumulator::new(&func_name, false)?;
accumulators.push((accumulator, col_idx));
}
_ => {
return Err(ExecutorError::Other(
"Streaming aggregate: SELECT must contain only aggregates".to_string(),
))
}
}
}
// Use streaming range scan to accumulate inline
if let Some(stream) = pk_index_data.range_scan_streaming(
Some(&low_value),
Some(&high_value),
true, // inclusive start (BETWEEN is inclusive)
true, // inclusive end
) {
for row_idx in stream {
if let Some(row) = table.get_row(row_idx) {
for (accumulator, col_idx) in &mut accumulators {
accumulator.accumulate(&row.values[*col_idx]);
}
}
}
}
// Finalize and return single row
let result_values: Vec<SqlValue> =
accumulators.iter().map(|(acc, _)| acc.finalize()).collect::<Result<Vec<_>, _>>()?;
Ok(vec![Row::from_vec(result_values)])
}
}