1use anyhow::{anyhow, Result};
2use fxhash::FxHashSet;
3use std::cmp::min;
4use std::collections::HashMap;
5use std::sync::Arc;
6use std::time::{Duration, Instant};
7use tracing::{debug, info};
8
9use crate::config::config::BehaviorConfig;
10use crate::config::global::get_date_notation;
11use crate::data::arithmetic_evaluator::ArithmeticEvaluator;
12use crate::data::data_view::DataView;
13use crate::data::datatable::{DataColumn, DataRow, DataTable, DataValue};
14use crate::data::evaluation_context::EvaluationContext;
15use crate::data::group_by_expressions::GroupByExpressions;
16use crate::data::hash_join::HashJoinExecutor;
17use crate::data::recursive_where_evaluator::RecursiveWhereEvaluator;
18use crate::data::row_expanders::RowExpanderRegistry;
19use crate::data::subquery_executor::SubqueryExecutor;
20use crate::data::temp_table_registry::TempTableRegistry;
21use crate::execution_plan::{ExecutionPlan, ExecutionPlanBuilder, StepType};
22use crate::sql::aggregates::{contains_aggregate, is_aggregate_compatible};
23use crate::sql::parser::ast::ColumnRef;
24use crate::sql::parser::ast::SetOperation;
25use crate::sql::parser::ast::TableSource;
26use crate::sql::parser::ast::WindowSpec;
27use crate::sql::recursive_parser::{
28 CTEType, OrderByItem, Parser, SelectItem, SelectStatement, SortDirection, SqlExpression,
29 TableFunction,
30};
31
32fn resolve_cte<'a>(
37 context: &'a HashMap<String, Arc<DataView>>,
38 name: &str,
39) -> Option<&'a Arc<DataView>> {
40 if let Some(v) = context.get(name) {
41 return Some(v);
42 }
43 let lower = name.to_lowercase();
44 context
45 .iter()
46 .find(|(k, _)| k.to_lowercase() == lower)
47 .map(|(_, v)| v)
48}
49
50#[derive(Debug, Clone)]
52pub struct ExecutionContext {
53 alias_map: HashMap<String, String>,
56}
57
58impl ExecutionContext {
59 pub fn new() -> Self {
61 Self {
62 alias_map: HashMap::new(),
63 }
64 }
65
66 pub fn register_alias(&mut self, alias: String, table_name: String) {
68 debug!("Registering alias: {} -> {}", alias, table_name);
69 self.alias_map.insert(alias, table_name);
70 }
71
72 pub fn resolve_alias(&self, name: &str) -> String {
75 self.alias_map
76 .get(name)
77 .cloned()
78 .unwrap_or_else(|| name.to_string())
79 }
80
81 pub fn is_alias(&self, name: &str) -> bool {
83 self.alias_map.contains_key(name)
84 }
85
86 pub fn get_aliases(&self) -> HashMap<String, String> {
88 self.alias_map.clone()
89 }
90
91 pub fn resolve_column_index(&self, table: &DataTable, column_ref: &ColumnRef) -> Result<usize> {
106 if let Some(table_prefix) = &column_ref.table_prefix {
107 let actual_table = self.resolve_alias(table_prefix);
109
110 let qualified_name = format!("{}.{}", actual_table, column_ref.name);
112 if let Some(idx) = table.find_column_by_qualified_name(&qualified_name) {
113 debug!(
114 "Resolved {}.{} -> qualified column '{}' at index {}",
115 table_prefix, column_ref.name, qualified_name, idx
116 );
117 return Ok(idx);
118 }
119
120 if let Some(idx) = table.get_column_index(&column_ref.name) {
122 debug!(
123 "Resolved {}.{} -> unqualified column '{}' at index {}",
124 table_prefix, column_ref.name, column_ref.name, idx
125 );
126 return Ok(idx);
127 }
128
129 Err(anyhow!(
131 "Column '{}' not found. Table '{}' may not support qualified column names",
132 qualified_name,
133 actual_table
134 ))
135 } else {
136 if let Some(idx) = table.get_column_index(&column_ref.name) {
138 debug!(
139 "Resolved unqualified column '{}' at index {}",
140 column_ref.name, idx
141 );
142 return Ok(idx);
143 }
144
145 if column_ref.name.contains('.') {
147 if let Some(idx) = table.find_column_by_qualified_name(&column_ref.name) {
148 debug!(
149 "Resolved '{}' as qualified column at index {}",
150 column_ref.name, idx
151 );
152 return Ok(idx);
153 }
154 }
155
156 let suggestion = self.find_similar_column(table, &column_ref.name);
158 match suggestion {
159 Some(similar) => Err(anyhow!(
160 "Column '{}' not found. Did you mean '{}'?",
161 column_ref.name,
162 similar
163 )),
164 None => Err(anyhow!("Column '{}' not found", column_ref.name)),
165 }
166 }
167 }
168
169 fn find_similar_column(&self, table: &DataTable, name: &str) -> Option<String> {
171 let columns = table.column_names();
172 let mut best_match: Option<(String, usize)> = None;
173
174 for col in columns {
175 let distance = edit_distance(name, &col);
176 if distance <= 2 {
177 match best_match {
179 Some((_, best_dist)) if distance < best_dist => {
180 best_match = Some((col.clone(), distance));
181 }
182 None => {
183 best_match = Some((col.clone(), distance));
184 }
185 _ => {}
186 }
187 }
188 }
189
190 best_match.map(|(name, _)| name)
191 }
192}
193
194impl Default for ExecutionContext {
195 fn default() -> Self {
196 Self::new()
197 }
198}
199
200fn edit_distance(a: &str, b: &str) -> usize {
202 let len_a = a.chars().count();
203 let len_b = b.chars().count();
204
205 if len_a == 0 {
206 return len_b;
207 }
208 if len_b == 0 {
209 return len_a;
210 }
211
212 let mut matrix = vec![vec![0; len_b + 1]; len_a + 1];
213
214 for i in 0..=len_a {
215 matrix[i][0] = i;
216 }
217 for j in 0..=len_b {
218 matrix[0][j] = j;
219 }
220
221 let a_chars: Vec<char> = a.chars().collect();
222 let b_chars: Vec<char> = b.chars().collect();
223
224 for i in 1..=len_a {
225 for j in 1..=len_b {
226 let cost = if a_chars[i - 1] == b_chars[j - 1] {
227 0
228 } else {
229 1
230 };
231 matrix[i][j] = min(
232 min(matrix[i - 1][j] + 1, matrix[i][j - 1] + 1),
233 matrix[i - 1][j - 1] + cost,
234 );
235 }
236 }
237
238 matrix[len_a][len_b]
239}
240
241#[derive(Clone)]
243pub struct QueryEngine {
244 case_insensitive: bool,
245 date_notation: String,
246 _behavior_config: Option<BehaviorConfig>,
247}
248
249impl Default for QueryEngine {
250 fn default() -> Self {
251 Self::new()
252 }
253}
254
255impl QueryEngine {
256 #[must_use]
257 pub fn new() -> Self {
258 Self {
259 case_insensitive: false,
260 date_notation: get_date_notation(),
261 _behavior_config: None,
262 }
263 }
264
265 #[must_use]
266 pub fn with_behavior_config(config: BehaviorConfig) -> Self {
267 let case_insensitive = config.case_insensitive_default;
268 let date_notation = get_date_notation();
270 Self {
271 case_insensitive,
272 date_notation,
273 _behavior_config: Some(config),
274 }
275 }
276
277 #[must_use]
278 pub fn with_date_notation(_date_notation: String) -> Self {
279 Self {
280 case_insensitive: false,
281 date_notation: get_date_notation(), _behavior_config: None,
283 }
284 }
285
286 #[must_use]
287 pub fn with_case_insensitive(case_insensitive: bool) -> Self {
288 Self {
289 case_insensitive,
290 date_notation: get_date_notation(),
291 _behavior_config: None,
292 }
293 }
294
295 #[must_use]
296 pub fn with_case_insensitive_and_date_notation(
297 case_insensitive: bool,
298 _date_notation: String, ) -> Self {
300 Self {
301 case_insensitive,
302 date_notation: get_date_notation(), _behavior_config: None,
304 }
305 }
306
307 fn find_similar_column(&self, table: &DataTable, name: &str) -> Option<String> {
309 let columns = table.column_names();
310 let mut best_match: Option<(String, usize)> = None;
311
312 for col in columns {
313 let distance = self.edit_distance(&col.to_lowercase(), &name.to_lowercase());
314 let max_distance = if name.len() > 10 { 3 } else { 2 };
317 if distance <= max_distance {
318 match &best_match {
319 None => best_match = Some((col, distance)),
320 Some((_, best_dist)) if distance < *best_dist => {
321 best_match = Some((col, distance));
322 }
323 _ => {}
324 }
325 }
326 }
327
328 best_match.map(|(name, _)| name)
329 }
330
331 fn edit_distance(&self, s1: &str, s2: &str) -> usize {
333 let len1 = s1.len();
334 let len2 = s2.len();
335 let mut matrix = vec![vec![0; len2 + 1]; len1 + 1];
336
337 for i in 0..=len1 {
338 matrix[i][0] = i;
339 }
340 for j in 0..=len2 {
341 matrix[0][j] = j;
342 }
343
344 for (i, c1) in s1.chars().enumerate() {
345 for (j, c2) in s2.chars().enumerate() {
346 let cost = usize::from(c1 != c2);
347 matrix[i + 1][j + 1] = std::cmp::min(
348 matrix[i][j + 1] + 1, std::cmp::min(
350 matrix[i + 1][j] + 1, matrix[i][j] + cost, ),
353 );
354 }
355 }
356
357 matrix[len1][len2]
358 }
359
360 fn contains_unnest(expr: &SqlExpression) -> bool {
362 match expr {
363 SqlExpression::Unnest { .. } => true,
365 SqlExpression::FunctionCall { name, args, .. } => {
366 if name.to_uppercase() == "UNNEST" {
367 return true;
368 }
369 args.iter().any(Self::contains_unnest)
371 }
372 SqlExpression::BinaryOp { left, right, .. } => {
373 Self::contains_unnest(left) || Self::contains_unnest(right)
374 }
375 SqlExpression::Not { expr } => Self::contains_unnest(expr),
376 SqlExpression::CaseExpression {
377 when_branches,
378 else_branch,
379 } => {
380 when_branches.iter().any(|branch| {
381 Self::contains_unnest(&branch.condition)
382 || Self::contains_unnest(&branch.result)
383 }) || else_branch
384 .as_ref()
385 .map_or(false, |e| Self::contains_unnest(e))
386 }
387 SqlExpression::SimpleCaseExpression {
388 expr,
389 when_branches,
390 else_branch,
391 } => {
392 Self::contains_unnest(expr)
393 || when_branches.iter().any(|branch| {
394 Self::contains_unnest(&branch.value)
395 || Self::contains_unnest(&branch.result)
396 })
397 || else_branch
398 .as_ref()
399 .map_or(false, |e| Self::contains_unnest(e))
400 }
401 SqlExpression::InList { expr, values } => {
402 Self::contains_unnest(expr) || values.iter().any(Self::contains_unnest)
403 }
404 SqlExpression::NotInList { expr, values } => {
405 Self::contains_unnest(expr) || values.iter().any(Self::contains_unnest)
406 }
407 SqlExpression::Between { expr, lower, upper } => {
408 Self::contains_unnest(expr)
409 || Self::contains_unnest(lower)
410 || Self::contains_unnest(upper)
411 }
412 SqlExpression::InSubquery { expr, .. } => Self::contains_unnest(expr),
413 SqlExpression::NotInSubquery { expr, .. } => Self::contains_unnest(expr),
414 SqlExpression::ScalarSubquery { .. } => false, SqlExpression::WindowFunction { args, .. } => args.iter().any(Self::contains_unnest),
416 SqlExpression::MethodCall { args, .. } => args.iter().any(Self::contains_unnest),
417 SqlExpression::ChainedMethodCall { base, args, .. } => {
418 Self::contains_unnest(base) || args.iter().any(Self::contains_unnest)
419 }
420 _ => false,
421 }
422 }
423
424 fn collect_window_specs(expr: &SqlExpression, specs: &mut Vec<WindowSpec>) {
426 match expr {
427 SqlExpression::WindowFunction {
428 window_spec, args, ..
429 } => {
430 specs.push(window_spec.clone());
432 for arg in args {
434 Self::collect_window_specs(arg, specs);
435 }
436 }
437 SqlExpression::BinaryOp { left, right, .. } => {
438 Self::collect_window_specs(left, specs);
439 Self::collect_window_specs(right, specs);
440 }
441 SqlExpression::Not { expr } => {
442 Self::collect_window_specs(expr, specs);
443 }
444 SqlExpression::FunctionCall { args, .. } => {
445 for arg in args {
446 Self::collect_window_specs(arg, specs);
447 }
448 }
449 SqlExpression::CaseExpression {
450 when_branches,
451 else_branch,
452 } => {
453 for branch in when_branches {
454 Self::collect_window_specs(&branch.condition, specs);
455 Self::collect_window_specs(&branch.result, specs);
456 }
457 if let Some(else_expr) = else_branch {
458 Self::collect_window_specs(else_expr, specs);
459 }
460 }
461 SqlExpression::SimpleCaseExpression {
462 expr,
463 when_branches,
464 else_branch,
465 } => {
466 Self::collect_window_specs(expr, specs);
467 for branch in when_branches {
468 Self::collect_window_specs(&branch.value, specs);
469 Self::collect_window_specs(&branch.result, specs);
470 }
471 if let Some(else_expr) = else_branch {
472 Self::collect_window_specs(else_expr, specs);
473 }
474 }
475 SqlExpression::InList { expr, values, .. } => {
476 Self::collect_window_specs(expr, specs);
477 for item in values {
478 Self::collect_window_specs(item, specs);
479 }
480 }
481 SqlExpression::ChainedMethodCall { base, args, .. } => {
482 Self::collect_window_specs(base, specs);
483 for arg in args {
484 Self::collect_window_specs(arg, specs);
485 }
486 }
487 SqlExpression::Column(_)
489 | SqlExpression::NumberLiteral(_)
490 | SqlExpression::StringLiteral(_)
491 | SqlExpression::BooleanLiteral(_)
492 | SqlExpression::Null
493 | SqlExpression::DateTimeToday { .. }
494 | SqlExpression::DateTimeConstructor { .. }
495 | SqlExpression::MethodCall { .. } => {}
496 _ => {}
498 }
499 }
500
501 fn contains_window_function(expr: &SqlExpression) -> bool {
503 match expr {
504 SqlExpression::WindowFunction { .. } => true,
505 SqlExpression::BinaryOp { left, right, .. } => {
506 Self::contains_window_function(left) || Self::contains_window_function(right)
507 }
508 SqlExpression::Not { expr } => Self::contains_window_function(expr),
509 SqlExpression::FunctionCall { args, .. } => {
510 args.iter().any(Self::contains_window_function)
511 }
512 SqlExpression::CaseExpression {
513 when_branches,
514 else_branch,
515 } => {
516 when_branches.iter().any(|branch| {
517 Self::contains_window_function(&branch.condition)
518 || Self::contains_window_function(&branch.result)
519 }) || else_branch
520 .as_ref()
521 .map_or(false, |e| Self::contains_window_function(e))
522 }
523 SqlExpression::SimpleCaseExpression {
524 expr,
525 when_branches,
526 else_branch,
527 } => {
528 Self::contains_window_function(expr)
529 || when_branches.iter().any(|branch| {
530 Self::contains_window_function(&branch.value)
531 || Self::contains_window_function(&branch.result)
532 })
533 || else_branch
534 .as_ref()
535 .map_or(false, |e| Self::contains_window_function(e))
536 }
537 SqlExpression::InList { expr, values } => {
538 Self::contains_window_function(expr)
539 || values.iter().any(Self::contains_window_function)
540 }
541 SqlExpression::NotInList { expr, values } => {
542 Self::contains_window_function(expr)
543 || values.iter().any(Self::contains_window_function)
544 }
545 SqlExpression::Between { expr, lower, upper } => {
546 Self::contains_window_function(expr)
547 || Self::contains_window_function(lower)
548 || Self::contains_window_function(upper)
549 }
550 SqlExpression::InSubquery { expr, .. } => Self::contains_window_function(expr),
551 SqlExpression::NotInSubquery { expr, .. } => Self::contains_window_function(expr),
552 SqlExpression::MethodCall { args, .. } => {
553 args.iter().any(Self::contains_window_function)
554 }
555 SqlExpression::ChainedMethodCall { base, args, .. } => {
556 Self::contains_window_function(base)
557 || args.iter().any(Self::contains_window_function)
558 }
559 _ => false,
560 }
561 }
562
563 fn extract_window_specs(
565 items: &[SelectItem],
566 ) -> Vec<crate::data::batch_window_evaluator::WindowFunctionSpec> {
567 let mut specs = Vec::new();
568 for (idx, item) in items.iter().enumerate() {
569 if let SelectItem::Expression { expr, .. } = item {
570 Self::collect_window_function_specs(expr, idx, &mut specs);
571 }
572 }
573 specs
574 }
575
576 fn collect_window_function_specs(
578 expr: &SqlExpression,
579 output_column_index: usize,
580 specs: &mut Vec<crate::data::batch_window_evaluator::WindowFunctionSpec>,
581 ) {
582 match expr {
583 SqlExpression::WindowFunction {
584 name,
585 args,
586 window_spec,
587 } => {
588 specs.push(crate::data::batch_window_evaluator::WindowFunctionSpec {
589 spec: window_spec.clone(),
590 function_name: name.clone(),
591 args: args.clone(),
592 output_column_index,
593 });
594 }
595 SqlExpression::BinaryOp { left, right, .. } => {
596 Self::collect_window_function_specs(left, output_column_index, specs);
597 Self::collect_window_function_specs(right, output_column_index, specs);
598 }
599 SqlExpression::Not { expr } => {
600 Self::collect_window_function_specs(expr, output_column_index, specs);
601 }
602 SqlExpression::FunctionCall { args, .. } => {
603 for arg in args {
604 Self::collect_window_function_specs(arg, output_column_index, specs);
605 }
606 }
607 SqlExpression::CaseExpression {
608 when_branches,
609 else_branch,
610 } => {
611 for branch in when_branches {
612 Self::collect_window_function_specs(
613 &branch.condition,
614 output_column_index,
615 specs,
616 );
617 Self::collect_window_function_specs(&branch.result, output_column_index, specs);
618 }
619 if let Some(e) = else_branch {
620 Self::collect_window_function_specs(e, output_column_index, specs);
621 }
622 }
623 SqlExpression::SimpleCaseExpression {
624 expr,
625 when_branches,
626 else_branch,
627 } => {
628 Self::collect_window_function_specs(expr, output_column_index, specs);
629 for branch in when_branches {
630 Self::collect_window_function_specs(&branch.value, output_column_index, specs);
631 Self::collect_window_function_specs(&branch.result, output_column_index, specs);
632 }
633 if let Some(e) = else_branch {
634 Self::collect_window_function_specs(e, output_column_index, specs);
635 }
636 }
637 SqlExpression::InList { expr, values } => {
638 Self::collect_window_function_specs(expr, output_column_index, specs);
639 for val in values {
640 Self::collect_window_function_specs(val, output_column_index, specs);
641 }
642 }
643 SqlExpression::NotInList { expr, values } => {
644 Self::collect_window_function_specs(expr, output_column_index, specs);
645 for val in values {
646 Self::collect_window_function_specs(val, output_column_index, specs);
647 }
648 }
649 SqlExpression::Between { expr, lower, upper } => {
650 Self::collect_window_function_specs(expr, output_column_index, specs);
651 Self::collect_window_function_specs(lower, output_column_index, specs);
652 Self::collect_window_function_specs(upper, output_column_index, specs);
653 }
654 SqlExpression::InSubquery { expr, .. } => {
655 Self::collect_window_function_specs(expr, output_column_index, specs);
656 }
657 SqlExpression::NotInSubquery { expr, .. } => {
658 Self::collect_window_function_specs(expr, output_column_index, specs);
659 }
660 SqlExpression::MethodCall { args, .. } => {
661 for arg in args {
662 Self::collect_window_function_specs(arg, output_column_index, specs);
663 }
664 }
665 SqlExpression::ChainedMethodCall { base, args, .. } => {
666 Self::collect_window_function_specs(base, output_column_index, specs);
667 for arg in args {
668 Self::collect_window_function_specs(arg, output_column_index, specs);
669 }
670 }
671 _ => {} }
673 }
674
675 pub fn execute(&self, table: Arc<DataTable>, sql: &str) -> Result<DataView> {
677 let (view, _plan) = self.execute_with_plan(table, sql)?;
678 Ok(view)
679 }
680
681 pub fn execute_with_temp_tables(
683 &self,
684 table: Arc<DataTable>,
685 sql: &str,
686 temp_tables: Option<&TempTableRegistry>,
687 ) -> Result<DataView> {
688 let (view, _plan) = self.execute_with_plan_and_temp_tables(table, sql, temp_tables)?;
689 Ok(view)
690 }
691
692 pub fn execute_statement(
694 &self,
695 table: Arc<DataTable>,
696 statement: SelectStatement,
697 ) -> Result<DataView> {
698 self.execute_statement_with_temp_tables(table, statement, None)
699 }
700
701 pub fn execute_statement_with_temp_tables(
703 &self,
704 table: Arc<DataTable>,
705 statement: SelectStatement,
706 temp_tables: Option<&TempTableRegistry>,
707 ) -> Result<DataView> {
708 let mut cte_context = HashMap::new();
710
711 if let Some(temp_registry) = temp_tables {
713 for table_name in temp_registry.list_tables() {
714 if let Some(temp_table) = temp_registry.get(&table_name) {
715 debug!("Adding temp table {} to CTE context", table_name);
716 let view = DataView::new(temp_table);
717 cte_context.insert(table_name, Arc::new(view));
718 }
719 }
720 }
721
722 for cte in &statement.ctes {
723 debug!("QueryEngine: Pre-processing CTE '{}'...", cte.name);
724 let cte_result = match &cte.cte_type {
726 CTEType::Standard(query) => {
727 let view = self.build_view_with_context(
729 table.clone(),
730 query.clone(),
731 &mut cte_context,
732 )?;
733
734 let mut materialized = self.materialize_view(view)?;
736
737 for column in materialized.columns_mut() {
739 column.qualified_name = Some(format!("{}.{}", cte.name, column.name));
740 column.source_table = Some(cte.name.clone());
741 }
742
743 DataView::new(Arc::new(materialized))
744 }
745 CTEType::Web(web_spec) => {
746 use crate::web::http_fetcher::WebDataFetcher;
748
749 let fetcher = WebDataFetcher::new()?;
750 let mut data_table = fetcher.fetch(web_spec, &cte.name, None)?;
752
753 for column in data_table.columns_mut() {
755 column.qualified_name = Some(format!("{}.{}", cte.name, column.name));
756 column.source_table = Some(cte.name.clone());
757 }
758
759 DataView::new(Arc::new(data_table))
761 }
762 CTEType::File(file_spec) => {
763 let mut data_table =
764 crate::data::file_walker::walk_filesystem(file_spec, &cte.name)?;
765
766 for column in data_table.columns_mut() {
767 column.qualified_name = Some(format!("{}.{}", cte.name, column.name));
768 column.source_table = Some(cte.name.clone());
769 }
770
771 DataView::new(Arc::new(data_table))
772 }
773 };
774 cte_context.insert(cte.name.clone(), Arc::new(cte_result));
776 debug!(
777 "QueryEngine: CTE '{}' pre-processed, stored in context",
778 cte.name
779 );
780 }
781
782 let mut subquery_executor =
784 SubqueryExecutor::with_cte_context(self.clone(), table.clone(), cte_context.clone());
785 let processed_statement = subquery_executor.execute_subqueries(&statement)?;
786
787 self.build_view_with_context(table, processed_statement, &mut cte_context)
789 }
790
791 pub fn execute_statement_with_cte_context(
793 &self,
794 table: Arc<DataTable>,
795 statement: SelectStatement,
796 cte_context: &HashMap<String, Arc<DataView>>,
797 ) -> Result<DataView> {
798 let mut local_context = cte_context.clone();
800
801 for cte in &statement.ctes {
803 debug!("QueryEngine: Processing nested CTE '{}'...", cte.name);
804 let cte_result = match &cte.cte_type {
805 CTEType::Standard(query) => {
806 let view = self.build_view_with_context(
807 table.clone(),
808 query.clone(),
809 &mut local_context,
810 )?;
811
812 let mut materialized = self.materialize_view(view)?;
814
815 for column in materialized.columns_mut() {
817 column.qualified_name = Some(format!("{}.{}", cte.name, column.name));
818 column.source_table = Some(cte.name.clone());
819 }
820
821 DataView::new(Arc::new(materialized))
822 }
823 CTEType::Web(web_spec) => {
824 use crate::web::http_fetcher::WebDataFetcher;
826
827 let fetcher = WebDataFetcher::new()?;
828 let mut data_table = fetcher.fetch(web_spec, &cte.name, None)?;
830
831 for column in data_table.columns_mut() {
833 column.qualified_name = Some(format!("{}.{}", cte.name, column.name));
834 column.source_table = Some(cte.name.clone());
835 }
836
837 DataView::new(Arc::new(data_table))
839 }
840 CTEType::File(file_spec) => {
841 let mut data_table =
842 crate::data::file_walker::walk_filesystem(file_spec, &cte.name)?;
843
844 for column in data_table.columns_mut() {
845 column.qualified_name = Some(format!("{}.{}", cte.name, column.name));
846 column.source_table = Some(cte.name.clone());
847 }
848
849 DataView::new(Arc::new(data_table))
850 }
851 };
852 local_context.insert(cte.name.clone(), Arc::new(cte_result));
853 }
854
855 let mut subquery_executor =
857 SubqueryExecutor::with_cte_context(self.clone(), table.clone(), local_context.clone());
858 let processed_statement = subquery_executor.execute_subqueries(&statement)?;
859
860 self.build_view_with_context(table, processed_statement, &mut local_context)
862 }
863
864 pub fn execute_with_plan(
866 &self,
867 table: Arc<DataTable>,
868 sql: &str,
869 ) -> Result<(DataView, ExecutionPlan)> {
870 self.execute_with_plan_and_temp_tables(table, sql, None)
871 }
872
873 pub fn execute_with_plan_and_temp_tables(
875 &self,
876 table: Arc<DataTable>,
877 sql: &str,
878 temp_tables: Option<&TempTableRegistry>,
879 ) -> Result<(DataView, ExecutionPlan)> {
880 let mut plan_builder = ExecutionPlanBuilder::new();
881 let start_time = Instant::now();
882
883 plan_builder.begin_step(StepType::Parse, "Parse SQL query".to_string());
885 plan_builder.add_detail(format!("Query: {}", sql));
886 let mut parser = Parser::new(sql);
887 let statement = parser
888 .parse()
889 .map_err(|e| anyhow::anyhow!("Parse error: {}", e))?;
890 plan_builder.add_detail(format!("Parsed successfully"));
891 if let Some(ref from_source) = statement.from_source {
892 match from_source {
893 TableSource::Table(name) => {
894 plan_builder.add_detail(format!("FROM: {}", name));
895 }
896 TableSource::DerivedTable { alias, .. } => {
897 plan_builder.add_detail(format!("FROM: derived table (alias: {})", alias));
898 }
899 TableSource::Pivot { .. } => {
900 plan_builder.add_detail("FROM: PIVOT".to_string());
901 }
902 }
903 }
904 if statement.where_clause.is_some() {
905 plan_builder.add_detail("WHERE clause present".to_string());
906 }
907 plan_builder.end_step();
908
909 let mut cte_context = HashMap::new();
911
912 if let Some(temp_registry) = temp_tables {
914 for table_name in temp_registry.list_tables() {
915 if let Some(temp_table) = temp_registry.get(&table_name) {
916 debug!("Adding temp table {} to CTE context", table_name);
917 let view = DataView::new(temp_table);
918 cte_context.insert(table_name, Arc::new(view));
919 }
920 }
921 }
922
923 if !statement.ctes.is_empty() {
924 plan_builder.begin_step(
925 StepType::CTE,
926 format!("Process {} CTEs", statement.ctes.len()),
927 );
928
929 for cte in &statement.ctes {
930 let cte_start = Instant::now();
931 plan_builder.begin_step(StepType::CTE, format!("CTE '{}'", cte.name));
932
933 let cte_result = match &cte.cte_type {
934 CTEType::Standard(query) => {
935 if let Some(ref from_source) = query.from_source {
937 match from_source {
938 TableSource::Table(name) => {
939 plan_builder.add_detail(format!("Source: {}", name));
940 }
941 TableSource::DerivedTable { alias, .. } => {
942 plan_builder
943 .add_detail(format!("Source: derived table ({})", alias));
944 }
945 TableSource::Pivot { .. } => {
946 plan_builder.add_detail("Source: PIVOT".to_string());
947 }
948 }
949 }
950 if query.where_clause.is_some() {
951 plan_builder.add_detail("Has WHERE clause".to_string());
952 }
953 if query.group_by.is_some() {
954 plan_builder.add_detail("Has GROUP BY".to_string());
955 }
956
957 debug!(
958 "QueryEngine: Processing CTE '{}' with existing context: {:?}",
959 cte.name,
960 cte_context.keys().collect::<Vec<_>>()
961 );
962
963 let mut subquery_executor = SubqueryExecutor::with_cte_context(
966 self.clone(),
967 table.clone(),
968 cte_context.clone(),
969 );
970 let processed_query = subquery_executor.execute_subqueries(query)?;
971
972 let view = self.build_view_with_context(
973 table.clone(),
974 processed_query,
975 &mut cte_context,
976 )?;
977
978 let mut materialized = self.materialize_view(view)?;
980
981 for column in materialized.columns_mut() {
983 column.qualified_name = Some(format!("{}.{}", cte.name, column.name));
984 column.source_table = Some(cte.name.clone());
985 }
986
987 DataView::new(Arc::new(materialized))
988 }
989 CTEType::Web(web_spec) => {
990 plan_builder.add_detail(format!("URL: {}", web_spec.url));
991 if let Some(format) = &web_spec.format {
992 plan_builder.add_detail(format!("Format: {:?}", format));
993 }
994 if let Some(cache) = web_spec.cache_seconds {
995 plan_builder.add_detail(format!("Cache: {} seconds", cache));
996 }
997
998 use crate::web::http_fetcher::WebDataFetcher;
1000
1001 let fetcher = WebDataFetcher::new()?;
1002 let mut data_table = fetcher.fetch(web_spec, &cte.name, None)?;
1004
1005 for column in data_table.columns_mut() {
1007 column.qualified_name = Some(format!("{}.{}", cte.name, column.name));
1008 column.source_table = Some(cte.name.clone());
1009 }
1010
1011 DataView::new(Arc::new(data_table))
1013 }
1014 CTEType::File(file_spec) => {
1015 plan_builder.add_detail(format!("PATH: {}", file_spec.path));
1016 if file_spec.recursive {
1017 plan_builder.add_detail("RECURSIVE".to_string());
1018 }
1019 if let Some(ref g) = file_spec.glob {
1020 plan_builder.add_detail(format!("GLOB: {}", g));
1021 }
1022 if let Some(d) = file_spec.max_depth {
1023 plan_builder.add_detail(format!("MAX_DEPTH: {}", d));
1024 }
1025
1026 let mut data_table =
1027 crate::data::file_walker::walk_filesystem(file_spec, &cte.name)?;
1028
1029 for column in data_table.columns_mut() {
1030 column.qualified_name = Some(format!("{}.{}", cte.name, column.name));
1031 column.source_table = Some(cte.name.clone());
1032 }
1033
1034 DataView::new(Arc::new(data_table))
1035 }
1036 };
1037
1038 plan_builder.set_rows_out(cte_result.row_count());
1040 plan_builder.add_detail(format!(
1041 "Result: {} rows, {} columns",
1042 cte_result.row_count(),
1043 cte_result.column_count()
1044 ));
1045 plan_builder.add_detail(format!(
1046 "Execution time: {:.3}ms",
1047 cte_start.elapsed().as_secs_f64() * 1000.0
1048 ));
1049
1050 debug!(
1051 "QueryEngine: Storing CTE '{}' in context with {} rows",
1052 cte.name,
1053 cte_result.row_count()
1054 );
1055 cte_context.insert(cte.name.clone(), Arc::new(cte_result));
1056 plan_builder.end_step();
1057 }
1058
1059 plan_builder.add_detail(format!(
1060 "All {} CTEs cached in context",
1061 statement.ctes.len()
1062 ));
1063 plan_builder.end_step();
1064 }
1065
1066 plan_builder.begin_step(StepType::Subquery, "Process subqueries".to_string());
1068 let mut subquery_executor =
1069 SubqueryExecutor::with_cte_context(self.clone(), table.clone(), cte_context.clone());
1070
1071 let has_subqueries = statement.where_clause.as_ref().map_or(false, |w| {
1073 format!("{:?}", w).contains("Subquery")
1075 });
1076
1077 if has_subqueries {
1078 plan_builder.add_detail("Evaluating subqueries in WHERE clause".to_string());
1079 }
1080
1081 let processed_statement = subquery_executor.execute_subqueries(&statement)?;
1082
1083 if has_subqueries {
1084 plan_builder.add_detail("Subqueries replaced with materialized values".to_string());
1085 } else {
1086 plan_builder.add_detail("No subqueries to process".to_string());
1087 }
1088
1089 plan_builder.end_step();
1090 let result = self.build_view_with_context_and_plan(
1091 table,
1092 processed_statement,
1093 &mut cte_context,
1094 &mut plan_builder,
1095 )?;
1096
1097 let total_duration = start_time.elapsed();
1098 info!(
1099 "Query execution complete: total={:?}, rows={}",
1100 total_duration,
1101 result.row_count()
1102 );
1103
1104 let plan = plan_builder.build();
1105 Ok((result, plan))
1106 }
1107
1108 fn build_view(&self, table: Arc<DataTable>, statement: SelectStatement) -> Result<DataView> {
1110 let mut cte_context = HashMap::new();
1111 self.build_view_with_context(table, statement, &mut cte_context)
1112 }
1113
1114 fn build_view_with_context(
1116 &self,
1117 table: Arc<DataTable>,
1118 statement: SelectStatement,
1119 cte_context: &mut HashMap<String, Arc<DataView>>,
1120 ) -> Result<DataView> {
1121 let mut dummy_plan = ExecutionPlanBuilder::new();
1122 let mut exec_context = ExecutionContext::new();
1123 self.build_view_with_context_and_plan_and_exec(
1124 table,
1125 statement,
1126 cte_context,
1127 &mut dummy_plan,
1128 &mut exec_context,
1129 )
1130 }
1131
1132 fn build_view_with_context_and_plan(
1134 &self,
1135 table: Arc<DataTable>,
1136 statement: SelectStatement,
1137 cte_context: &mut HashMap<String, Arc<DataView>>,
1138 plan: &mut ExecutionPlanBuilder,
1139 ) -> Result<DataView> {
1140 let mut exec_context = ExecutionContext::new();
1141 self.build_view_with_context_and_plan_and_exec(
1142 table,
1143 statement,
1144 cte_context,
1145 plan,
1146 &mut exec_context,
1147 )
1148 }
1149
1150 fn build_view_with_context_and_plan_and_exec(
1152 &self,
1153 table: Arc<DataTable>,
1154 statement: SelectStatement,
1155 cte_context: &mut HashMap<String, Arc<DataView>>,
1156 plan: &mut ExecutionPlanBuilder,
1157 exec_context: &mut ExecutionContext,
1158 ) -> Result<DataView> {
1159 for cte in &statement.ctes {
1161 if cte_context.contains_key(&cte.name) {
1163 debug!(
1164 "QueryEngine: CTE '{}' already in context, skipping",
1165 cte.name
1166 );
1167 continue;
1168 }
1169
1170 debug!("QueryEngine: Processing CTE '{}'...", cte.name);
1171 debug!(
1172 "QueryEngine: Available CTEs for '{}': {:?}",
1173 cte.name,
1174 cte_context.keys().collect::<Vec<_>>()
1175 );
1176
1177 let cte_result = match &cte.cte_type {
1179 CTEType::Standard(query) => {
1180 let view =
1181 self.build_view_with_context(table.clone(), query.clone(), cte_context)?;
1182
1183 let mut materialized = self.materialize_view(view)?;
1185
1186 for column in materialized.columns_mut() {
1188 column.qualified_name = Some(format!("{}.{}", cte.name, column.name));
1189 column.source_table = Some(cte.name.clone());
1190 }
1191
1192 DataView::new(Arc::new(materialized))
1193 }
1194 CTEType::Web(_web_spec) => {
1195 return Err(anyhow!(
1197 "Web CTEs should be processed in execute_select method"
1198 ));
1199 }
1200 CTEType::File(_file_spec) => {
1201 return Err(anyhow!(
1203 "FILE CTEs should be processed in execute_select method"
1204 ));
1205 }
1206 };
1207
1208 cte_context.insert(cte.name.clone(), Arc::new(cte_result));
1210 debug!(
1211 "QueryEngine: CTE '{}' processed, stored in context",
1212 cte.name
1213 );
1214 }
1215
1216 let source_table = if let Some(ref from_source) = statement.from_source {
1218 match from_source {
1219 TableSource::Table(table_name) => {
1220 if let Some(cte_view) = resolve_cte(cte_context, table_name) {
1222 debug!("QueryEngine: Using CTE '{}' as source table", table_name);
1223 let mut materialized = self.materialize_view((**cte_view).clone())?;
1225
1226 #[allow(deprecated)]
1228 if let Some(ref alias) = statement.from_alias {
1229 debug!(
1230 "QueryEngine: Applying alias '{}' to CTE '{}' qualified column names",
1231 alias, table_name
1232 );
1233 for column in materialized.columns_mut() {
1234 if let Some(ref qualified_name) = column.qualified_name {
1236 if qualified_name.starts_with(&format!("{}.", table_name)) {
1237 column.qualified_name = Some(qualified_name.replace(
1238 &format!("{}.", table_name),
1239 &format!("{}.", alias),
1240 ));
1241 }
1242 }
1243 if column.source_table.as_ref() == Some(table_name) {
1245 column.source_table = Some(alias.clone());
1246 }
1247 }
1248 }
1249
1250 Arc::new(materialized)
1251 } else {
1252 table.clone()
1254 }
1255 }
1256 TableSource::DerivedTable { query, alias } => {
1257 debug!(
1259 "QueryEngine: Processing FROM derived table (alias: {})",
1260 alias
1261 );
1262 let subquery_result =
1263 self.build_view_with_context(table.clone(), *query.clone(), cte_context)?;
1264
1265 let mut materialized = self.materialize_view(subquery_result)?;
1268
1269 for column in materialized.columns_mut() {
1273 column.source_table = Some(alias.clone());
1274 }
1275
1276 Arc::new(materialized)
1277 }
1278 TableSource::Pivot { .. } => {
1279 return Err(anyhow!(
1281 "PIVOT in FROM clause should have been expanded by preprocessing pipeline"
1282 ));
1283 }
1284 }
1285 } else {
1286 #[allow(deprecated)]
1288 if let Some(ref table_func) = statement.from_function {
1289 debug!("QueryEngine: Processing table function (deprecated field)...");
1291 match table_func {
1292 TableFunction::Generator { name, args } => {
1293 use crate::sql::generators::GeneratorRegistry;
1295
1296 let registry = GeneratorRegistry::new();
1298
1299 if let Some(generator) = registry.get(name) {
1300 let mut evaluator = ArithmeticEvaluator::with_date_notation(
1302 &table,
1303 self.date_notation.clone(),
1304 );
1305 let dummy_row = 0;
1306
1307 let mut evaluated_args = Vec::new();
1308 for arg in args {
1309 evaluated_args.push(evaluator.evaluate(arg, dummy_row)?);
1310 }
1311
1312 generator.generate(evaluated_args)?
1314 } else {
1315 return Err(anyhow!("Unknown generator function: {}", name));
1316 }
1317 }
1318 }
1319 } else {
1320 #[allow(deprecated)]
1321 if let Some(ref subquery) = statement.from_subquery {
1322 debug!("QueryEngine: Processing FROM subquery (deprecated field)...");
1324 let subquery_result = self.build_view_with_context(
1325 table.clone(),
1326 *subquery.clone(),
1327 cte_context,
1328 )?;
1329
1330 let materialized = self.materialize_view(subquery_result)?;
1333 Arc::new(materialized)
1334 } else {
1335 #[allow(deprecated)]
1336 if let Some(ref table_name) = statement.from_table {
1337 if let Some(cte_view) = resolve_cte(cte_context, table_name) {
1339 debug!(
1340 "QueryEngine: Using CTE '{}' as source table (deprecated field)",
1341 table_name
1342 );
1343 let mut materialized = self.materialize_view((**cte_view).clone())?;
1345
1346 #[allow(deprecated)]
1348 if let Some(ref alias) = statement.from_alias {
1349 debug!(
1350 "QueryEngine: Applying alias '{}' to CTE '{}' qualified column names",
1351 alias, table_name
1352 );
1353 for column in materialized.columns_mut() {
1354 if let Some(ref qualified_name) = column.qualified_name {
1356 if qualified_name.starts_with(&format!("{}.", table_name)) {
1357 column.qualified_name = Some(qualified_name.replace(
1358 &format!("{}.", table_name),
1359 &format!("{}.", alias),
1360 ));
1361 }
1362 }
1363 if column.source_table.as_ref() == Some(table_name) {
1365 column.source_table = Some(alias.clone());
1366 }
1367 }
1368 }
1369
1370 Arc::new(materialized)
1371 } else {
1372 table.clone()
1374 }
1375 } else {
1376 Arc::new(DataTable::dual())
1381 }
1382 }
1383 }
1384 };
1385
1386 #[allow(deprecated)]
1388 if let Some(ref alias) = statement.from_alias {
1389 #[allow(deprecated)]
1390 if let Some(ref table_name) = statement.from_table {
1391 exec_context.register_alias(alias.clone(), table_name.clone());
1392 }
1393 }
1394
1395 let final_table = if !statement.joins.is_empty() {
1397 plan.begin_step(
1398 StepType::Join,
1399 format!("Process {} JOINs", statement.joins.len()),
1400 );
1401 plan.set_rows_in(source_table.row_count());
1402
1403 let join_executor = HashJoinExecutor::new(self.case_insensitive);
1404
1405 #[allow(deprecated)]
1409 let base_table_name = match statement.from_source {
1410 Some(TableSource::Table(ref n)) => Some(n.clone()),
1411 _ => statement.from_table.clone(),
1412 };
1413
1414 let mut current_table = source_table;
1415
1416 for (idx, join_clause) in statement.joins.iter().enumerate() {
1417 let join_start = Instant::now();
1418 plan.begin_step(StepType::Join, format!("JOIN #{}", idx + 1));
1419 plan.add_detail(format!("Type: {:?}", join_clause.join_type));
1420 plan.add_detail(format!("Left table: {} rows", current_table.row_count()));
1421 plan.add_detail(format!(
1422 "Executing {:?} JOIN on {} condition(s)",
1423 join_clause.join_type,
1424 join_clause.condition.conditions.len()
1425 ));
1426
1427 let right_table = match &join_clause.table {
1429 TableSource::Table(name) => {
1430 if let Some(cte_view) = resolve_cte(cte_context, name) {
1432 let mut materialized = self.materialize_view((**cte_view).clone())?;
1433
1434 if let Some(ref alias) = join_clause.alias {
1436 debug!("QueryEngine: Applying JOIN alias '{}' to CTE '{}' qualified column names", alias, name);
1437 for column in materialized.columns_mut() {
1438 if let Some(ref qualified_name) = column.qualified_name {
1440 if qualified_name.starts_with(&format!("{}.", name)) {
1441 column.qualified_name = Some(qualified_name.replace(
1442 &format!("{}.", name),
1443 &format!("{}.", alias),
1444 ));
1445 }
1446 }
1447 if column.source_table.as_ref() == Some(name) {
1449 column.source_table = Some(alias.clone());
1450 }
1451 }
1452 }
1453
1454 Arc::new(materialized)
1455 } else if base_table_name.as_deref().is_some_and(|base| {
1456 if self.case_insensitive {
1457 base.eq_ignore_ascii_case(name)
1458 } else {
1459 base == name
1460 }
1461 }) {
1462 let mut materialized = (*table).clone();
1468 if let Some(ref alias) = join_clause.alias {
1469 for column in materialized.columns_mut() {
1470 if let Some(ref qualified_name) = column.qualified_name {
1471 if qualified_name.starts_with(&format!("{}.", name)) {
1472 column.qualified_name = Some(qualified_name.replace(
1473 &format!("{}.", name),
1474 &format!("{}.", alias),
1475 ));
1476 }
1477 }
1478 if column.source_table.as_ref() == Some(name) {
1479 column.source_table = Some(alias.clone());
1480 }
1481 }
1482 }
1483 Arc::new(materialized)
1484 } else {
1485 return Err(anyhow!("Cannot resolve table '{}' for JOIN", name));
1488 }
1489 }
1490 TableSource::DerivedTable { query, alias: _ } => {
1491 let subquery_result = self.build_view_with_context(
1493 table.clone(),
1494 *query.clone(),
1495 cte_context,
1496 )?;
1497 let materialized = self.materialize_view(subquery_result)?;
1498 Arc::new(materialized)
1499 }
1500 TableSource::Pivot { .. } => {
1501 return Err(anyhow!("PIVOT in JOIN clause is not yet supported"));
1503 }
1504 };
1505
1506 let joined = join_executor.execute_join(
1508 current_table.clone(),
1509 join_clause,
1510 right_table.clone(),
1511 )?;
1512
1513 plan.add_detail(format!("Right table: {} rows", right_table.row_count()));
1514 plan.set_rows_out(joined.row_count());
1515 plan.add_detail(format!("Result: {} rows", joined.row_count()));
1516 plan.add_detail(format!(
1517 "Join time: {:.3}ms",
1518 join_start.elapsed().as_secs_f64() * 1000.0
1519 ));
1520 plan.end_step();
1521
1522 current_table = Arc::new(joined);
1523 }
1524
1525 plan.set_rows_out(current_table.row_count());
1526 plan.add_detail(format!(
1527 "Final result after all joins: {} rows",
1528 current_table.row_count()
1529 ));
1530 plan.end_step();
1531 current_table
1532 } else {
1533 source_table
1534 };
1535
1536 self.build_view_internal_with_plan_and_exec(
1538 final_table,
1539 statement,
1540 plan,
1541 Some(exec_context),
1542 )
1543 }
1544
1545 pub fn materialize_view(&self, view: DataView) -> Result<DataTable> {
1547 let source = view.source();
1548 let mut result_table = DataTable::new("derived");
1549
1550 let visible_cols = view.visible_column_indices().to_vec();
1552
1553 for col_idx in &visible_cols {
1555 let col = &source.columns[*col_idx];
1556 let new_col = DataColumn {
1557 name: col.name.clone(),
1558 data_type: col.data_type.clone(),
1559 nullable: col.nullable,
1560 unique_values: col.unique_values,
1561 null_count: col.null_count,
1562 metadata: col.metadata.clone(),
1563 qualified_name: col.qualified_name.clone(), source_table: col.source_table.clone(), };
1566 result_table.add_column(new_col);
1567 }
1568
1569 for row_idx in view.visible_row_indices() {
1571 let source_row = &source.rows[*row_idx];
1572 let mut new_row = DataRow { values: Vec::new() };
1573
1574 for col_idx in &visible_cols {
1575 new_row.values.push(source_row.values[*col_idx].clone());
1576 }
1577
1578 result_table.add_row(new_row);
1579 }
1580
1581 Ok(result_table)
1582 }
1583
1584 fn build_view_internal(
1585 &self,
1586 table: Arc<DataTable>,
1587 statement: SelectStatement,
1588 ) -> Result<DataView> {
1589 let mut dummy_plan = ExecutionPlanBuilder::new();
1590 self.build_view_internal_with_plan(table, statement, &mut dummy_plan)
1591 }
1592
1593 fn build_view_internal_with_plan(
1594 &self,
1595 table: Arc<DataTable>,
1596 statement: SelectStatement,
1597 plan: &mut ExecutionPlanBuilder,
1598 ) -> Result<DataView> {
1599 self.build_view_internal_with_plan_and_exec(table, statement, plan, None)
1600 }
1601
1602 fn build_view_internal_with_plan_and_exec(
1603 &self,
1604 table: Arc<DataTable>,
1605 statement: SelectStatement,
1606 plan: &mut ExecutionPlanBuilder,
1607 exec_context: Option<&ExecutionContext>,
1608 ) -> Result<DataView> {
1609 debug!(
1610 "QueryEngine::build_view - select_items: {:?}",
1611 statement.select_items
1612 );
1613 debug!(
1614 "QueryEngine::build_view - where_clause: {:?}",
1615 statement.where_clause
1616 );
1617
1618 let mut visible_rows: Vec<usize> = (0..table.row_count()).collect();
1620
1621 if let Some(where_clause) = &statement.where_clause {
1623 let total_rows = table.row_count();
1624 debug!("QueryEngine: Applying WHERE clause to {} rows", total_rows);
1625 debug!("QueryEngine: WHERE clause = {:?}", where_clause);
1626
1627 plan.begin_step(StepType::Filter, "WHERE clause filtering".to_string());
1628 plan.set_rows_in(total_rows);
1629 plan.add_detail(format!("Input: {} rows", total_rows));
1630
1631 for condition in &where_clause.conditions {
1633 plan.add_detail(format!("Condition: {:?}", condition.expr));
1634 }
1635
1636 let filter_start = Instant::now();
1637 let mut eval_context = EvaluationContext::new(self.case_insensitive);
1639
1640 let mut evaluator = if let Some(exec_ctx) = exec_context {
1642 RecursiveWhereEvaluator::with_both_contexts(&table, &mut eval_context, exec_ctx)
1644 } else {
1645 RecursiveWhereEvaluator::with_context(&table, &mut eval_context)
1646 };
1647
1648 let mut filtered_rows = Vec::new();
1650 for row_idx in visible_rows {
1651 if row_idx < 3 {
1653 debug!("QueryEngine: Evaluating WHERE clause for row {}", row_idx);
1654 }
1655
1656 match evaluator.evaluate(where_clause, row_idx) {
1657 Ok(result) => {
1658 if row_idx < 3 {
1659 debug!("QueryEngine: Row {} WHERE result: {}", row_idx, result);
1660 }
1661 if result {
1662 filtered_rows.push(row_idx);
1663 }
1664 }
1665 Err(e) => {
1666 if row_idx < 3 {
1667 debug!(
1668 "QueryEngine: WHERE evaluation error for row {}: {}",
1669 row_idx, e
1670 );
1671 }
1672 return Err(e);
1674 }
1675 }
1676 }
1677
1678 let (compilations, cache_hits) = eval_context.get_stats();
1680 if compilations > 0 || cache_hits > 0 {
1681 debug!(
1682 "LIKE pattern cache: {} compilations, {} cache hits",
1683 compilations, cache_hits
1684 );
1685 }
1686 visible_rows = filtered_rows;
1687 let filter_duration = filter_start.elapsed();
1688 info!(
1689 "WHERE clause filtering: {} rows -> {} rows in {:?}",
1690 total_rows,
1691 visible_rows.len(),
1692 filter_duration
1693 );
1694
1695 plan.set_rows_out(visible_rows.len());
1696 plan.add_detail(format!("Output: {} rows", visible_rows.len()));
1697 plan.add_detail(format!(
1698 "Filter time: {:.3}ms",
1699 filter_duration.as_secs_f64() * 1000.0
1700 ));
1701 plan.end_step();
1702 }
1703
1704 let mut view = DataView::new(table.clone());
1706 view = view.with_rows(visible_rows);
1707
1708 if let Some(group_by_exprs) = &statement.group_by {
1710 if !group_by_exprs.is_empty() {
1711 debug!("QueryEngine: Processing GROUP BY: {:?}", group_by_exprs);
1712
1713 plan.begin_step(
1714 StepType::GroupBy,
1715 format!("GROUP BY {} expressions", group_by_exprs.len()),
1716 );
1717 plan.set_rows_in(view.row_count());
1718 plan.add_detail(format!("Input: {} rows", view.row_count()));
1719 for expr in group_by_exprs {
1720 plan.add_detail(format!("Group by: {:?}", expr));
1721 }
1722
1723 let group_start = Instant::now();
1724 view = self.apply_group_by(
1725 view,
1726 group_by_exprs,
1727 &statement.select_items,
1728 statement.having.as_ref(),
1729 plan,
1730 )?;
1731
1732 use crate::query_plan::having_alias_transformer::HIDDEN_AGG_PREFIX;
1735 let hidden_indices: Vec<usize> = view
1736 .source()
1737 .columns
1738 .iter()
1739 .enumerate()
1740 .filter_map(|(i, c)| {
1741 if c.name.starts_with(HIDDEN_AGG_PREFIX) {
1742 Some(i)
1743 } else {
1744 None
1745 }
1746 })
1747 .collect();
1748 for &idx in hidden_indices.iter().rev() {
1749 view.hide_column(idx);
1750 }
1751
1752 plan.set_rows_out(view.row_count());
1753 plan.add_detail(format!("Output: {} groups", view.row_count()));
1754 plan.add_detail(format!(
1755 "Overall time: {:.3}ms",
1756 group_start.elapsed().as_secs_f64() * 1000.0
1757 ));
1758 plan.end_step();
1759 }
1760 } else {
1761 if !statement.select_items.is_empty() {
1763 let has_non_star_items = statement
1765 .select_items
1766 .iter()
1767 .any(|item| !matches!(item, SelectItem::Star { .. }));
1768
1769 if has_non_star_items || statement.select_items.len() > 1 {
1773 view = self.apply_select_items(
1774 view,
1775 &statement.select_items,
1776 &statement,
1777 exec_context,
1778 plan,
1779 )?;
1780 }
1781 } else if !statement.columns.is_empty() && statement.columns[0] != "*" {
1783 debug!("QueryEngine: Using legacy columns path");
1784 let source_table = view.source();
1787 let column_indices =
1788 self.resolve_column_indices(source_table, &statement.columns)?;
1789 view = view.with_columns(column_indices);
1790 }
1791 }
1792
1793 if statement.distinct {
1795 plan.begin_step(StepType::Distinct, "Remove duplicate rows".to_string());
1796 plan.set_rows_in(view.row_count());
1797 plan.add_detail(format!("Input: {} rows", view.row_count()));
1798
1799 let distinct_start = Instant::now();
1800 view = self.apply_distinct(view)?;
1801
1802 plan.set_rows_out(view.row_count());
1803 plan.add_detail(format!("Output: {} unique rows", view.row_count()));
1804 plan.add_detail(format!(
1805 "Distinct time: {:.3}ms",
1806 distinct_start.elapsed().as_secs_f64() * 1000.0
1807 ));
1808 plan.end_step();
1809 }
1810
1811 if let Some(order_by_columns) = &statement.order_by {
1813 if !order_by_columns.is_empty() {
1814 plan.begin_step(
1815 StepType::Sort,
1816 format!("ORDER BY {} columns", order_by_columns.len()),
1817 );
1818 plan.set_rows_in(view.row_count());
1819 for col in order_by_columns {
1820 let expr_str = match &col.expr {
1822 SqlExpression::Column(col_ref) => col_ref.name.clone(),
1823 _ => "expr".to_string(),
1824 };
1825 plan.add_detail(format!("{} {:?}", expr_str, col.direction));
1826 }
1827
1828 let sort_start = Instant::now();
1829 view =
1830 self.apply_multi_order_by_with_context(view, order_by_columns, exec_context)?;
1831
1832 plan.add_detail(format!(
1833 "Sort time: {:.3}ms",
1834 sort_start.elapsed().as_secs_f64() * 1000.0
1835 ));
1836 plan.end_step();
1837 }
1838 }
1839
1840 {
1845 use crate::query_plan::order_by_alias_transformer::HIDDEN_ORDERBY_PREFIX;
1846 let hidden_indices: Vec<usize> = view
1847 .source()
1848 .columns
1849 .iter()
1850 .enumerate()
1851 .filter_map(|(i, c)| {
1852 if c.name.starts_with(HIDDEN_ORDERBY_PREFIX) {
1853 Some(i)
1854 } else {
1855 None
1856 }
1857 })
1858 .collect();
1859 for &idx in hidden_indices.iter().rev() {
1860 view.hide_column(idx);
1861 }
1862 }
1863
1864 if let Some(limit) = statement.limit {
1866 let offset = statement.offset.unwrap_or(0);
1867 plan.begin_step(StepType::Limit, format!("LIMIT {}", limit));
1868 plan.set_rows_in(view.row_count());
1869 if offset > 0 {
1870 plan.add_detail(format!("OFFSET: {}", offset));
1871 }
1872 view = view.with_limit(limit, offset);
1873 plan.set_rows_out(view.row_count());
1874 plan.add_detail(format!("Output: {} rows", view.row_count()));
1875 plan.end_step();
1876 }
1877
1878 if !statement.set_operations.is_empty() {
1880 plan.begin_step(
1881 StepType::SetOperation,
1882 format!("Process {} set operations", statement.set_operations.len()),
1883 );
1884 plan.set_rows_in(view.row_count());
1885
1886 let mut combined_table = self.materialize_view(view)?;
1888 let first_columns = combined_table.column_names();
1889 let first_column_count = first_columns.len();
1890
1891 let mut needs_deduplication = false;
1893
1894 for (idx, (operation, next_statement)) in statement.set_operations.iter().enumerate() {
1896 let op_start = Instant::now();
1897 plan.begin_step(
1898 StepType::SetOperation,
1899 format!("{:?} operation #{}", operation, idx + 1),
1900 );
1901
1902 let next_view = if let Some(exec_ctx) = exec_context {
1905 self.build_view_internal_with_plan_and_exec(
1906 table.clone(),
1907 *next_statement.clone(),
1908 plan,
1909 Some(exec_ctx),
1910 )?
1911 } else {
1912 self.build_view_internal_with_plan(
1913 table.clone(),
1914 *next_statement.clone(),
1915 plan,
1916 )?
1917 };
1918
1919 let next_table = self.materialize_view(next_view)?;
1921 let next_columns = next_table.column_names();
1922 let next_column_count = next_columns.len();
1923
1924 if first_column_count != next_column_count {
1926 return Err(anyhow!(
1927 "UNION queries must have the same number of columns: first query has {} columns, but query #{} has {} columns",
1928 first_column_count,
1929 idx + 2,
1930 next_column_count
1931 ));
1932 }
1933
1934 for (col_idx, (first_col, next_col)) in
1936 first_columns.iter().zip(next_columns.iter()).enumerate()
1937 {
1938 if !first_col.eq_ignore_ascii_case(next_col) {
1939 debug!(
1940 "UNION column name mismatch at position {}: '{}' vs '{}' (using first query's name)",
1941 col_idx + 1,
1942 first_col,
1943 next_col
1944 );
1945 }
1946 }
1947
1948 plan.add_detail(format!("Left: {} rows", combined_table.row_count()));
1949 plan.add_detail(format!("Right: {} rows", next_table.row_count()));
1950
1951 match operation {
1953 SetOperation::UnionAll => {
1954 for row in next_table.rows.iter() {
1956 combined_table.add_row(row.clone());
1957 }
1958 plan.add_detail(format!(
1959 "Result: {} rows (no deduplication)",
1960 combined_table.row_count()
1961 ));
1962 }
1963 SetOperation::Union => {
1964 for row in next_table.rows.iter() {
1966 combined_table.add_row(row.clone());
1967 }
1968 needs_deduplication = true;
1969 plan.add_detail(format!(
1970 "Combined: {} rows (deduplication pending)",
1971 combined_table.row_count()
1972 ));
1973 }
1974 SetOperation::Intersect => {
1975 let right_keys: std::collections::HashSet<String> = next_table
1980 .rows
1981 .iter()
1982 .map(|r| format!("{:?}", r.values))
1983 .collect();
1984 let mut seen = std::collections::HashSet::new();
1985 let retained: Vec<_> = combined_table
1986 .rows
1987 .iter()
1988 .filter(|r| {
1989 let key = format!("{:?}", r.values);
1990 right_keys.contains(&key) && seen.insert(key)
1991 })
1992 .cloned()
1993 .collect();
1994 combined_table.rows = retained;
1995 plan.add_detail(format!(
1996 "Result: {} rows (intersection, deduplicated)",
1997 combined_table.row_count()
1998 ));
1999 }
2000 SetOperation::Except => {
2001 let right_keys: std::collections::HashSet<String> = next_table
2004 .rows
2005 .iter()
2006 .map(|r| format!("{:?}", r.values))
2007 .collect();
2008 let mut seen = std::collections::HashSet::new();
2009 let retained: Vec<_> = combined_table
2010 .rows
2011 .iter()
2012 .filter(|r| {
2013 let key = format!("{:?}", r.values);
2014 !right_keys.contains(&key) && seen.insert(key)
2015 })
2016 .cloned()
2017 .collect();
2018 combined_table.rows = retained;
2019 plan.add_detail(format!(
2020 "Result: {} rows (difference, deduplicated)",
2021 combined_table.row_count()
2022 ));
2023 }
2024 }
2025
2026 plan.add_detail(format!(
2027 "Operation time: {:.3}ms",
2028 op_start.elapsed().as_secs_f64() * 1000.0
2029 ));
2030 plan.set_rows_out(combined_table.row_count());
2031 plan.end_step();
2032 }
2033
2034 plan.set_rows_out(combined_table.row_count());
2035 plan.add_detail(format!(
2036 "Combined result: {} rows after {} operations",
2037 combined_table.row_count(),
2038 statement.set_operations.len()
2039 ));
2040 plan.end_step();
2041
2042 view = DataView::new(Arc::new(combined_table));
2044
2045 if needs_deduplication {
2047 plan.begin_step(
2048 StepType::Distinct,
2049 "UNION deduplication - remove duplicate rows".to_string(),
2050 );
2051 plan.set_rows_in(view.row_count());
2052 plan.add_detail(format!("Input: {} rows", view.row_count()));
2053
2054 let distinct_start = Instant::now();
2055 view = self.apply_distinct(view)?;
2056
2057 plan.set_rows_out(view.row_count());
2058 plan.add_detail(format!("Output: {} unique rows", view.row_count()));
2059 plan.add_detail(format!(
2060 "Deduplication time: {:.3}ms",
2061 distinct_start.elapsed().as_secs_f64() * 1000.0
2062 ));
2063 plan.end_step();
2064 }
2065 }
2066
2067 Ok(view)
2068 }
2069
2070 fn resolve_column_indices(&self, table: &DataTable, columns: &[String]) -> Result<Vec<usize>> {
2072 let mut indices = Vec::new();
2073 let table_columns = table.column_names();
2074
2075 for col_name in columns {
2076 let index = table_columns
2077 .iter()
2078 .position(|c| c.eq_ignore_ascii_case(col_name))
2079 .ok_or_else(|| {
2080 let suggestion = self.find_similar_column(table, col_name);
2081 match suggestion {
2082 Some(similar) => anyhow::anyhow!(
2083 "Column '{}' not found. Did you mean '{}'?",
2084 col_name,
2085 similar
2086 ),
2087 None => anyhow::anyhow!("Column '{}' not found", col_name),
2088 }
2089 })?;
2090 indices.push(index);
2091 }
2092
2093 Ok(indices)
2094 }
2095
2096 fn apply_select_items(
2098 &self,
2099 view: DataView,
2100 select_items: &[SelectItem],
2101 _statement: &SelectStatement,
2102 exec_context: Option<&ExecutionContext>,
2103 plan: &mut ExecutionPlanBuilder,
2104 ) -> Result<DataView> {
2105 debug!(
2106 "QueryEngine::apply_select_items - items: {:?}",
2107 select_items
2108 );
2109 debug!(
2110 "QueryEngine::apply_select_items - input view has {} rows",
2111 view.row_count()
2112 );
2113
2114 let has_window_functions = select_items.iter().any(|item| match item {
2116 SelectItem::Expression { expr, .. } => Self::contains_window_function(expr),
2117 _ => false,
2118 });
2119
2120 let window_func_count: usize = select_items
2122 .iter()
2123 .filter(|item| match item {
2124 SelectItem::Expression { expr, .. } => Self::contains_window_function(expr),
2125 _ => false,
2126 })
2127 .count();
2128
2129 let window_start = if has_window_functions {
2131 debug!(
2132 "QueryEngine::apply_select_items - detected {} window functions",
2133 window_func_count
2134 );
2135
2136 let window_specs = Self::extract_window_specs(select_items);
2138 debug!("Extracted {} window function specs", window_specs.len());
2139
2140 Some(Instant::now())
2141 } else {
2142 None
2143 };
2144
2145 let has_unnest = select_items.iter().any(|item| match item {
2147 SelectItem::Expression { expr, .. } => Self::contains_unnest(expr),
2148 _ => false,
2149 });
2150
2151 if has_unnest {
2152 debug!("QueryEngine::apply_select_items - UNNEST detected, using row expansion");
2153 return self.apply_select_with_row_expansion(view, select_items);
2154 }
2155
2156 let has_aggregates = select_items.iter().any(|item| match item {
2160 SelectItem::Expression { expr, .. } => contains_aggregate(expr),
2161 SelectItem::Column { .. } => false,
2162 SelectItem::Star { .. } => false,
2163 SelectItem::StarExclude { .. } => false,
2164 });
2165
2166 let all_aggregate_compatible = select_items.iter().all(|item| match item {
2167 SelectItem::Expression { expr, .. } => is_aggregate_compatible(expr),
2168 SelectItem::Column { .. } => false, SelectItem::Star { .. } => false, SelectItem::StarExclude { .. } => false, });
2172
2173 if has_aggregates && all_aggregate_compatible && view.row_count() > 0 {
2174 debug!("QueryEngine::apply_select_items - detected aggregate query with constants");
2177 return self.apply_aggregate_select(view, select_items);
2178 }
2179
2180 let has_computed_expressions = select_items
2182 .iter()
2183 .any(|item| matches!(item, SelectItem::Expression { .. }));
2184
2185 debug!(
2186 "QueryEngine::apply_select_items - has_computed_expressions: {}",
2187 has_computed_expressions
2188 );
2189
2190 if !has_computed_expressions {
2191 let column_indices = self.resolve_select_columns(view.source(), select_items)?;
2193 return Ok(view.with_columns(column_indices));
2194 }
2195
2196 let source_table = view.source();
2201 let visible_rows = view.visible_row_indices();
2202
2203 let mut computed_table = DataTable::new("query_result");
2206
2207 let mut expanded_items = Vec::new();
2209 for item in select_items {
2210 match item {
2211 SelectItem::Star { table_prefix, .. } => {
2212 if let Some(prefix) = table_prefix {
2213 debug!("QueryEngine::apply_select_items - expanding {}.*", prefix);
2215 for col in &source_table.columns {
2216 if Self::column_matches_table(col, prefix) {
2217 expanded_items.push(SelectItem::Column {
2218 column: ColumnRef::unquoted(col.name.clone()),
2219 leading_comments: vec![],
2220 trailing_comment: None,
2221 });
2222 }
2223 }
2224 } else {
2225 debug!("QueryEngine::apply_select_items - expanding *");
2227 for col_name in source_table.column_names() {
2228 expanded_items.push(SelectItem::Column {
2229 column: ColumnRef::unquoted(col_name.to_string()),
2230 leading_comments: vec![],
2231 trailing_comment: None,
2232 });
2233 }
2234 }
2235 }
2236 _ => expanded_items.push(item.clone()),
2237 }
2238 }
2239
2240 let mut column_name_counts: std::collections::HashMap<String, usize> =
2242 std::collections::HashMap::new();
2243
2244 for item in &expanded_items {
2245 let base_name = match item {
2246 SelectItem::Column {
2247 column: col_ref, ..
2248 } => col_ref.name.clone(),
2249 SelectItem::Expression { alias, .. } => alias.clone(),
2250 SelectItem::Star { .. } => unreachable!("Star should have been expanded"),
2251 SelectItem::StarExclude { .. } => {
2252 unreachable!("StarExclude should have been expanded")
2253 }
2254 };
2255
2256 let count = column_name_counts.entry(base_name.clone()).or_insert(0);
2258 let column_name = if *count == 0 {
2259 base_name.clone()
2261 } else {
2262 format!("{base_name}_{count}")
2264 };
2265 *count += 1;
2266
2267 computed_table.add_column(DataColumn::new(&column_name));
2268 }
2269
2270 let can_use_batch = expanded_items.iter().all(|item| {
2274 match item {
2275 SelectItem::Expression { expr, .. } => {
2276 matches!(expr, SqlExpression::WindowFunction { .. })
2279 || !Self::contains_window_function(expr)
2280 }
2281 _ => true, }
2283 });
2284
2285 let use_batch_evaluation = can_use_batch
2288 && std::env::var("SQL_CLI_BATCH_WINDOW")
2289 .map(|v| v != "0" && v.to_lowercase() != "false")
2290 .unwrap_or(true);
2291
2292 let batch_window_specs = if use_batch_evaluation && has_window_functions {
2294 debug!("BATCH window function evaluation flag is enabled");
2295 let specs = Self::extract_window_specs(&expanded_items);
2297 debug!(
2298 "Extracted {} window function specs for batch evaluation",
2299 specs.len()
2300 );
2301 Some(specs)
2302 } else {
2303 None
2304 };
2305
2306 let mut evaluator =
2313 ArithmeticEvaluator::with_date_notation(source_table, self.date_notation.clone())
2314 .with_visible_rows(view.visible_row_indices().to_vec());
2315
2316 if let Some(exec_ctx) = exec_context {
2318 let aliases = exec_ctx.get_aliases();
2319 if !aliases.is_empty() {
2320 debug!(
2321 "Applying {} aliases to evaluator: {:?}",
2322 aliases.len(),
2323 aliases
2324 );
2325 evaluator = evaluator.with_table_aliases(aliases);
2326 }
2327 }
2328
2329 if has_window_functions {
2332 let preload_start = Instant::now();
2333
2334 let mut window_specs = Vec::new();
2336 for item in &expanded_items {
2337 if let SelectItem::Expression { expr, .. } = item {
2338 Self::collect_window_specs(expr, &mut window_specs);
2339 }
2340 }
2341
2342 for spec in &window_specs {
2344 let _ = evaluator.get_or_create_window_context(spec);
2345 }
2346
2347 debug!(
2348 "Pre-created {} WindowContext(s) in {:.2}ms",
2349 window_specs.len(),
2350 preload_start.elapsed().as_secs_f64() * 1000.0
2351 );
2352 }
2353
2354 if let Some(window_specs) = batch_window_specs {
2356 debug!("Starting batch window function evaluation");
2357 let batch_start = Instant::now();
2358
2359 let mut batch_results: Vec<Vec<DataValue>> =
2361 vec![vec![DataValue::Null; expanded_items.len()]; visible_rows.len()];
2362
2363 let detailed_window_specs = &window_specs;
2365
2366 let mut specs_by_window: HashMap<
2368 u64,
2369 Vec<&crate::data::batch_window_evaluator::WindowFunctionSpec>,
2370 > = HashMap::new();
2371 for spec in detailed_window_specs {
2372 let hash = spec.spec.compute_hash();
2373 specs_by_window
2374 .entry(hash)
2375 .or_insert_with(Vec::new)
2376 .push(spec);
2377 }
2378
2379 for (_window_hash, specs) in specs_by_window {
2381 let context = evaluator.get_or_create_window_context(&specs[0].spec)?;
2383
2384 for spec in specs {
2386 match spec.function_name.as_str() {
2387 "LAG" => {
2388 if let Some(SqlExpression::Column(col_ref)) = spec.args.get(0) {
2390 let column_name = col_ref.name.as_str();
2391 let offset = if let Some(SqlExpression::NumberLiteral(n)) =
2392 spec.args.get(1)
2393 {
2394 n.parse::<i64>().unwrap_or(1)
2395 } else {
2396 1 };
2398
2399 let values = context.evaluate_lag_batch(
2400 visible_rows,
2401 column_name,
2402 offset,
2403 )?;
2404
2405 for (row_idx, value) in values.into_iter().enumerate() {
2407 batch_results[row_idx][spec.output_column_index] = value;
2408 }
2409 }
2410 }
2411 "LEAD" => {
2412 if let Some(SqlExpression::Column(col_ref)) = spec.args.get(0) {
2414 let column_name = col_ref.name.as_str();
2415 let offset = if let Some(SqlExpression::NumberLiteral(n)) =
2416 spec.args.get(1)
2417 {
2418 n.parse::<i64>().unwrap_or(1)
2419 } else {
2420 1 };
2422
2423 let values = context.evaluate_lead_batch(
2424 visible_rows,
2425 column_name,
2426 offset,
2427 )?;
2428
2429 for (row_idx, value) in values.into_iter().enumerate() {
2431 batch_results[row_idx][spec.output_column_index] = value;
2432 }
2433 }
2434 }
2435 "ROW_NUMBER" => {
2436 let values = context.evaluate_row_number_batch(visible_rows)?;
2437
2438 for (row_idx, value) in values.into_iter().enumerate() {
2440 batch_results[row_idx][spec.output_column_index] = value;
2441 }
2442 }
2443 "RANK" => {
2444 let values = context.evaluate_rank_batch(visible_rows)?;
2445
2446 for (row_idx, value) in values.into_iter().enumerate() {
2448 batch_results[row_idx][spec.output_column_index] = value;
2449 }
2450 }
2451 "DENSE_RANK" => {
2452 let values = context.evaluate_dense_rank_batch(visible_rows)?;
2453
2454 for (row_idx, value) in values.into_iter().enumerate() {
2456 batch_results[row_idx][spec.output_column_index] = value;
2457 }
2458 }
2459 "SUM" => {
2460 if let Some(SqlExpression::Column(col_ref)) = spec.args.get(0) {
2461 let column_name = col_ref.name.as_str();
2462 let values =
2463 context.evaluate_sum_batch(visible_rows, column_name)?;
2464
2465 for (row_idx, value) in values.into_iter().enumerate() {
2466 batch_results[row_idx][spec.output_column_index] = value;
2467 }
2468 }
2469 }
2470 "AVG" => {
2471 if let Some(SqlExpression::Column(col_ref)) = spec.args.get(0) {
2472 let column_name = col_ref.name.as_str();
2473 let values =
2474 context.evaluate_avg_batch(visible_rows, column_name)?;
2475
2476 for (row_idx, value) in values.into_iter().enumerate() {
2477 batch_results[row_idx][spec.output_column_index] = value;
2478 }
2479 }
2480 }
2481 "MIN" => {
2482 if let Some(SqlExpression::Column(col_ref)) = spec.args.get(0) {
2483 let column_name = col_ref.name.as_str();
2484 let values =
2485 context.evaluate_min_batch(visible_rows, column_name)?;
2486
2487 for (row_idx, value) in values.into_iter().enumerate() {
2488 batch_results[row_idx][spec.output_column_index] = value;
2489 }
2490 }
2491 }
2492 "MAX" => {
2493 if let Some(SqlExpression::Column(col_ref)) = spec.args.get(0) {
2494 let column_name = col_ref.name.as_str();
2495 let values =
2496 context.evaluate_max_batch(visible_rows, column_name)?;
2497
2498 for (row_idx, value) in values.into_iter().enumerate() {
2499 batch_results[row_idx][spec.output_column_index] = value;
2500 }
2501 }
2502 }
2503 "COUNT" => {
2504 let column_name = match spec.args.get(0) {
2506 Some(SqlExpression::Column(col_ref)) => Some(col_ref.name.as_str()),
2507 Some(SqlExpression::StringLiteral(s)) if s == "*" => None,
2508 _ => None,
2509 };
2510
2511 let values = context.evaluate_count_batch(visible_rows, column_name)?;
2512
2513 for (row_idx, value) in values.into_iter().enumerate() {
2514 batch_results[row_idx][spec.output_column_index] = value;
2515 }
2516 }
2517 "FIRST_VALUE" => {
2518 if let Some(SqlExpression::Column(col_ref)) = spec.args.get(0) {
2519 let column_name = col_ref.name.as_str();
2520 let values = context
2521 .evaluate_first_value_batch(visible_rows, column_name)?;
2522
2523 for (row_idx, value) in values.into_iter().enumerate() {
2524 batch_results[row_idx][spec.output_column_index] = value;
2525 }
2526 }
2527 }
2528 "LAST_VALUE" => {
2529 if let Some(SqlExpression::Column(col_ref)) = spec.args.get(0) {
2530 let column_name = col_ref.name.as_str();
2531 let values =
2532 context.evaluate_last_value_batch(visible_rows, column_name)?;
2533
2534 for (row_idx, value) in values.into_iter().enumerate() {
2535 batch_results[row_idx][spec.output_column_index] = value;
2536 }
2537 }
2538 }
2539 _ => {
2540 debug!(
2542 "Window function {} not supported in batch mode, using per-row",
2543 spec.function_name
2544 );
2545 }
2546 }
2547 }
2548 }
2549
2550 for (result_row_idx, &source_row_idx) in visible_rows.iter().enumerate() {
2552 for (col_idx, item) in expanded_items.iter().enumerate() {
2553 if !matches!(batch_results[result_row_idx][col_idx], DataValue::Null) {
2555 continue;
2556 }
2557
2558 let value = match item {
2559 SelectItem::Column {
2560 column: col_ref, ..
2561 } => {
2562 match evaluator
2563 .evaluate(&SqlExpression::Column(col_ref.clone()), source_row_idx)
2564 {
2565 Ok(val) => val,
2566 Err(e) => {
2567 return Err(anyhow!(
2568 "Failed to evaluate column {}: {}",
2569 col_ref.to_sql(),
2570 e
2571 ));
2572 }
2573 }
2574 }
2575 SelectItem::Expression { expr, .. } => {
2576 if matches!(expr, SqlExpression::WindowFunction { .. }) {
2579 continue;
2581 }
2582 evaluator.evaluate(&expr, source_row_idx)?
2585 }
2586 SelectItem::Star { .. } => unreachable!("Star should have been expanded"),
2587 SelectItem::StarExclude { .. } => {
2588 unreachable!("StarExclude should have been expanded")
2589 }
2590 };
2591 batch_results[result_row_idx][col_idx] = value;
2592 }
2593 }
2594
2595 for row_values in batch_results {
2597 computed_table
2598 .add_row(DataRow::new(row_values))
2599 .map_err(|e| anyhow::anyhow!("Failed to add row: {}", e))?;
2600 }
2601
2602 debug!(
2603 "Batch window evaluation completed in {:.3}ms",
2604 batch_start.elapsed().as_secs_f64() * 1000.0
2605 );
2606 } else {
2607 for &row_idx in visible_rows {
2609 let mut row_values = Vec::new();
2610
2611 for item in &expanded_items {
2612 let value = match item {
2613 SelectItem::Column {
2614 column: col_ref, ..
2615 } => {
2616 match evaluator
2618 .evaluate(&SqlExpression::Column(col_ref.clone()), row_idx)
2619 {
2620 Ok(val) => val,
2621 Err(e) => {
2622 return Err(anyhow!(
2623 "Failed to evaluate column {}: {}",
2624 col_ref.to_sql(),
2625 e
2626 ));
2627 }
2628 }
2629 }
2630 SelectItem::Expression { expr, .. } => {
2631 evaluator.evaluate(&expr, row_idx)?
2633 }
2634 SelectItem::Star { .. } => unreachable!("Star should have been expanded"),
2635 SelectItem::StarExclude { .. } => {
2636 unreachable!("StarExclude should have been expanded")
2637 }
2638 };
2639 row_values.push(value);
2640 }
2641
2642 computed_table
2643 .add_row(DataRow::new(row_values))
2644 .map_err(|e| anyhow::anyhow!("Failed to add row: {}", e))?;
2645 }
2646 }
2647
2648 if let Some(start) = window_start {
2650 let window_duration = start.elapsed();
2651 info!(
2652 "Window function evaluation took {:.2}ms for {} rows ({} window functions)",
2653 window_duration.as_secs_f64() * 1000.0,
2654 visible_rows.len(),
2655 window_func_count
2656 );
2657
2658 plan.begin_step(
2660 StepType::WindowFunction,
2661 format!("Evaluate {} window function(s)", window_func_count),
2662 );
2663 plan.set_rows_in(visible_rows.len());
2664 plan.set_rows_out(visible_rows.len());
2665 plan.add_detail(format!("Input: {} rows", visible_rows.len()));
2666 plan.add_detail(format!("{} window functions evaluated", window_func_count));
2667 plan.add_detail(format!(
2668 "Evaluation time: {:.3}ms",
2669 window_duration.as_secs_f64() * 1000.0
2670 ));
2671 plan.end_step();
2672 }
2673
2674 Ok(DataView::new(Arc::new(computed_table)))
2677 }
2678
2679 fn apply_select_with_row_expansion(
2681 &self,
2682 view: DataView,
2683 select_items: &[SelectItem],
2684 ) -> Result<DataView> {
2685 debug!("QueryEngine::apply_select_with_row_expansion - expanding rows");
2686
2687 let source_table = view.source();
2688 let visible_rows = view.visible_row_indices();
2689 let expander_registry = RowExpanderRegistry::new();
2690
2691 let mut result_table = DataTable::new("unnest_result");
2693
2694 let mut expanded_items = Vec::new();
2696 for item in select_items {
2697 match item {
2698 SelectItem::Star { table_prefix, .. } => {
2699 if let Some(prefix) = table_prefix {
2700 debug!(
2702 "QueryEngine::apply_select_with_row_expansion - expanding {}.*",
2703 prefix
2704 );
2705 for col in &source_table.columns {
2706 if Self::column_matches_table(col, prefix) {
2707 expanded_items.push(SelectItem::Column {
2708 column: ColumnRef::unquoted(col.name.clone()),
2709 leading_comments: vec![],
2710 trailing_comment: None,
2711 });
2712 }
2713 }
2714 } else {
2715 debug!("QueryEngine::apply_select_with_row_expansion - expanding *");
2717 for col_name in source_table.column_names() {
2718 expanded_items.push(SelectItem::Column {
2719 column: ColumnRef::unquoted(col_name.to_string()),
2720 leading_comments: vec![],
2721 trailing_comment: None,
2722 });
2723 }
2724 }
2725 }
2726 _ => expanded_items.push(item.clone()),
2727 }
2728 }
2729
2730 for item in &expanded_items {
2732 let column_name = match item {
2733 SelectItem::Column {
2734 column: col_ref, ..
2735 } => col_ref.name.clone(),
2736 SelectItem::Expression { alias, .. } => alias.clone(),
2737 SelectItem::Star { .. } => unreachable!("Star should have been expanded"),
2738 SelectItem::StarExclude { .. } => {
2739 unreachable!("StarExclude should have been expanded")
2740 }
2741 };
2742 result_table.add_column(DataColumn::new(&column_name));
2743 }
2744
2745 let mut evaluator =
2747 ArithmeticEvaluator::with_date_notation(source_table, self.date_notation.clone());
2748
2749 for &row_idx in visible_rows {
2750 let mut unnest_expansions = Vec::new();
2752 let mut unnest_indices = Vec::new();
2753
2754 for (col_idx, item) in expanded_items.iter().enumerate() {
2755 if let SelectItem::Expression { expr, .. } = item {
2756 if let Some(expansion_result) = self.try_expand_unnest(
2757 &expr,
2758 source_table,
2759 row_idx,
2760 &mut evaluator,
2761 &expander_registry,
2762 )? {
2763 unnest_expansions.push(expansion_result);
2764 unnest_indices.push(col_idx);
2765 }
2766 }
2767 }
2768
2769 let expansion_count = if unnest_expansions.is_empty() {
2771 1 } else {
2773 unnest_expansions
2774 .iter()
2775 .map(|exp| exp.row_count())
2776 .max()
2777 .unwrap_or(1)
2778 };
2779
2780 for output_idx in 0..expansion_count {
2782 let mut row_values = Vec::new();
2783
2784 for (col_idx, item) in expanded_items.iter().enumerate() {
2785 let unnest_position = unnest_indices.iter().position(|&idx| idx == col_idx);
2787
2788 let value = if let Some(unnest_idx) = unnest_position {
2789 let expansion = &unnest_expansions[unnest_idx];
2791 expansion
2792 .values
2793 .get(output_idx)
2794 .cloned()
2795 .unwrap_or(DataValue::Null)
2796 } else {
2797 match item {
2799 SelectItem::Column {
2800 column: col_ref, ..
2801 } => {
2802 let col_idx =
2803 source_table.get_column_index(&col_ref.name).ok_or_else(
2804 || anyhow::anyhow!("Column '{}' not found", col_ref.name),
2805 )?;
2806 let row = source_table
2807 .get_row(row_idx)
2808 .ok_or_else(|| anyhow::anyhow!("Row {} not found", row_idx))?;
2809 row.get(col_idx)
2810 .ok_or_else(|| {
2811 anyhow::anyhow!("Column {} not found in row", col_idx)
2812 })?
2813 .clone()
2814 }
2815 SelectItem::Expression { expr, .. } => {
2816 evaluator.evaluate(&expr, row_idx)?
2818 }
2819 SelectItem::Star { .. } => unreachable!(),
2820 SelectItem::StarExclude { .. } => {
2821 unreachable!("StarExclude should have been expanded")
2822 }
2823 }
2824 };
2825
2826 row_values.push(value);
2827 }
2828
2829 result_table
2830 .add_row(DataRow::new(row_values))
2831 .map_err(|e| anyhow::anyhow!("Failed to add expanded row: {}", e))?;
2832 }
2833 }
2834
2835 debug!(
2836 "QueryEngine::apply_select_with_row_expansion - input rows: {}, output rows: {}",
2837 visible_rows.len(),
2838 result_table.row_count()
2839 );
2840
2841 Ok(DataView::new(Arc::new(result_table)))
2842 }
2843
2844 fn try_expand_unnest(
2847 &self,
2848 expr: &SqlExpression,
2849 _source_table: &DataTable,
2850 row_idx: usize,
2851 evaluator: &mut ArithmeticEvaluator,
2852 expander_registry: &RowExpanderRegistry,
2853 ) -> Result<Option<crate::data::row_expanders::ExpansionResult>> {
2854 if let SqlExpression::Unnest { column, delimiter } = expr {
2856 let column_value = evaluator.evaluate(column, row_idx)?;
2858
2859 let delimiter_value = DataValue::String(delimiter.clone());
2861
2862 let expander = expander_registry
2864 .get("UNNEST")
2865 .ok_or_else(|| anyhow::anyhow!("UNNEST expander not found"))?;
2866
2867 let expansion = expander.expand(&column_value, &[delimiter_value])?;
2869 return Ok(Some(expansion));
2870 }
2871
2872 if let SqlExpression::FunctionCall { name, args, .. } = expr {
2874 if name.to_uppercase() == "UNNEST" {
2875 if args.len() != 2 {
2877 return Err(anyhow::anyhow!(
2878 "UNNEST requires exactly 2 arguments: UNNEST(column, delimiter)"
2879 ));
2880 }
2881
2882 let column_value = evaluator.evaluate(&args[0], row_idx)?;
2884
2885 let delimiter_value = evaluator.evaluate(&args[1], row_idx)?;
2887
2888 let expander = expander_registry
2890 .get("UNNEST")
2891 .ok_or_else(|| anyhow::anyhow!("UNNEST expander not found"))?;
2892
2893 let expansion = expander.expand(&column_value, &[delimiter_value])?;
2895 return Ok(Some(expansion));
2896 }
2897 }
2898
2899 Ok(None)
2900 }
2901
2902 fn apply_aggregate_select(
2904 &self,
2905 view: DataView,
2906 select_items: &[SelectItem],
2907 ) -> Result<DataView> {
2908 debug!("QueryEngine::apply_aggregate_select - creating single row aggregate result");
2909
2910 let source_table = view.source();
2911 let mut result_table = DataTable::new("aggregate_result");
2912
2913 for item in select_items {
2915 let column_name = match item {
2916 SelectItem::Expression { alias, .. } => alias.clone(),
2917 _ => unreachable!("Should only have expressions in aggregate-only query"),
2918 };
2919 result_table.add_column(DataColumn::new(&column_name));
2920 }
2921
2922 let visible_rows = view.visible_row_indices().to_vec();
2924 let mut evaluator =
2925 ArithmeticEvaluator::with_date_notation(source_table, self.date_notation.clone())
2926 .with_visible_rows(visible_rows);
2927
2928 let mut row_values = Vec::new();
2930 for item in select_items {
2931 match item {
2932 SelectItem::Expression { expr, .. } => {
2933 let value = evaluator.evaluate(expr, 0)?;
2936 row_values.push(value);
2937 }
2938 _ => unreachable!("Should only have expressions in aggregate-only query"),
2939 }
2940 }
2941
2942 result_table
2944 .add_row(DataRow::new(row_values))
2945 .map_err(|e| anyhow::anyhow!("Failed to add aggregate result row: {}", e))?;
2946
2947 Ok(DataView::new(Arc::new(result_table)))
2948 }
2949
2950 fn column_matches_table(col: &DataColumn, table_name: &str) -> bool {
2962 if let Some(ref source) = col.source_table {
2964 if source == table_name || source.ends_with(&format!(".{}", table_name)) {
2966 return true;
2967 }
2968 }
2969
2970 if let Some(ref qualified) = col.qualified_name {
2972 if qualified.starts_with(&format!("{}.", table_name)) {
2974 return true;
2975 }
2976 }
2977
2978 false
2979 }
2980
2981 fn resolve_select_columns(
2983 &self,
2984 table: &DataTable,
2985 select_items: &[SelectItem],
2986 ) -> Result<Vec<usize>> {
2987 let mut indices = Vec::new();
2988 let table_columns = table.column_names();
2989
2990 for item in select_items {
2991 match item {
2992 SelectItem::Column {
2993 column: col_ref, ..
2994 } => {
2995 let index = if let Some(table_prefix) = &col_ref.table_prefix {
2997 let qualified_name = format!("{}.{}", table_prefix, col_ref.name);
3006 table.find_column_by_qualified_name(&qualified_name)
3007 .or_else(|| {
3008 table_columns
3009 .iter()
3010 .position(|c| c.eq_ignore_ascii_case(&col_ref.name))
3011 })
3012 .ok_or_else(|| {
3013 let has_qualified = table.columns.iter()
3015 .any(|c| c.qualified_name.is_some());
3016 if !has_qualified {
3017 anyhow::anyhow!(
3018 "Column '{}' not found. Note: Table '{}' may not support qualified column names",
3019 qualified_name, table_prefix
3020 )
3021 } else {
3022 anyhow::anyhow!("Column '{}' not found", qualified_name)
3023 }
3024 })?
3025 } else {
3026 table_columns
3028 .iter()
3029 .position(|c| c.eq_ignore_ascii_case(&col_ref.name))
3030 .ok_or_else(|| {
3031 let suggestion = self.find_similar_column(table, &col_ref.name);
3032 match suggestion {
3033 Some(similar) => anyhow::anyhow!(
3034 "Column '{}' not found. Did you mean '{}'?",
3035 col_ref.name,
3036 similar
3037 ),
3038 None => anyhow::anyhow!("Column '{}' not found", col_ref.name),
3039 }
3040 })?
3041 };
3042 indices.push(index);
3043 }
3044 SelectItem::Star { table_prefix, .. } => {
3045 if let Some(prefix) = table_prefix {
3046 for (i, col) in table.columns.iter().enumerate() {
3048 if Self::column_matches_table(col, prefix) {
3049 indices.push(i);
3050 }
3051 }
3052 } else {
3053 for i in 0..table_columns.len() {
3055 indices.push(i);
3056 }
3057 }
3058 }
3059 SelectItem::StarExclude {
3060 table_prefix,
3061 excluded_columns,
3062 ..
3063 } => {
3064 if let Some(prefix) = table_prefix {
3066 for (i, col) in table.columns.iter().enumerate() {
3068 if Self::column_matches_table(col, prefix)
3069 && !excluded_columns.contains(&col.name)
3070 {
3071 indices.push(i);
3072 }
3073 }
3074 } else {
3075 for (i, col_name) in table_columns.iter().enumerate() {
3077 if !excluded_columns
3078 .iter()
3079 .any(|exc| exc.eq_ignore_ascii_case(col_name))
3080 {
3081 indices.push(i);
3082 }
3083 }
3084 }
3085 }
3086 SelectItem::Expression { .. } => {
3087 return Err(anyhow::anyhow!(
3088 "Computed expressions require new table creation"
3089 ));
3090 }
3091 }
3092 }
3093
3094 Ok(indices)
3095 }
3096
3097 fn apply_distinct(&self, view: DataView) -> Result<DataView> {
3099 use std::collections::HashSet;
3100
3101 let source = view.source();
3102 let visible_cols = view.visible_column_indices();
3103 let visible_rows = view.visible_row_indices();
3104
3105 let mut seen_rows = HashSet::new();
3107 let mut unique_row_indices = Vec::new();
3108
3109 for &row_idx in visible_rows {
3110 let mut row_key = Vec::new();
3112 for &col_idx in visible_cols {
3113 let value = source
3114 .get_value(row_idx, col_idx)
3115 .ok_or_else(|| anyhow!("Invalid cell reference"))?;
3116 row_key.push(format!("{:?}", value));
3118 }
3119
3120 if seen_rows.insert(row_key) {
3122 unique_row_indices.push(row_idx);
3124 }
3125 }
3126
3127 Ok(view.with_rows(unique_row_indices))
3129 }
3130
3131 fn apply_multi_order_by(
3133 &self,
3134 view: DataView,
3135 order_by_columns: &[OrderByItem],
3136 ) -> Result<DataView> {
3137 self.apply_multi_order_by_with_context(view, order_by_columns, None)
3138 }
3139
3140 fn apply_multi_order_by_with_context(
3142 &self,
3143 mut view: DataView,
3144 order_by_columns: &[OrderByItem],
3145 _exec_context: Option<&ExecutionContext>,
3146 ) -> Result<DataView> {
3147 let mut sort_columns = Vec::new();
3149
3150 for order_col in order_by_columns {
3151 let column_name = match &order_col.expr {
3153 SqlExpression::Column(col_ref) => col_ref.name.clone(),
3154 _ => {
3155 return Err(anyhow!(
3157 "ORDER BY expressions not yet supported - only simple columns allowed"
3158 ));
3159 }
3160 };
3161
3162 let col_index = if column_name.contains('.') {
3164 if let Some(dot_pos) = column_name.rfind('.') {
3166 let col_name = &column_name[dot_pos + 1..];
3167
3168 debug!(
3171 "ORDER BY: Extracting unqualified column '{}' from '{}'",
3172 col_name, column_name
3173 );
3174 view.source().get_column_index(col_name)
3175 } else {
3176 view.source().get_column_index(&column_name)
3177 }
3178 } else {
3179 view.source().get_column_index(&column_name)
3181 }
3182 .ok_or_else(|| {
3183 let suggestion = self.find_similar_column(view.source(), &column_name);
3185 match suggestion {
3186 Some(similar) => anyhow::anyhow!(
3187 "Column '{}' not found. Did you mean '{}'?",
3188 column_name,
3189 similar
3190 ),
3191 None => {
3192 let available_cols = view.source().column_names().join(", ");
3194 anyhow::anyhow!(
3195 "Column '{}' not found. Available columns: {}",
3196 column_name,
3197 available_cols
3198 )
3199 }
3200 }
3201 })?;
3202
3203 let ascending = matches!(order_col.direction, SortDirection::Asc);
3204 sort_columns.push((col_index, ascending));
3205 }
3206
3207 view.apply_multi_sort(&sort_columns)?;
3209 Ok(view)
3210 }
3211
3212 fn apply_group_by(
3214 &self,
3215 view: DataView,
3216 group_by_exprs: &[SqlExpression],
3217 select_items: &[SelectItem],
3218 having: Option<&SqlExpression>,
3219 plan: &mut ExecutionPlanBuilder,
3220 ) -> Result<DataView> {
3221 let (result_view, phase_info) = self.apply_group_by_expressions(
3223 view,
3224 group_by_exprs,
3225 select_items,
3226 having,
3227 self.case_insensitive,
3228 self.date_notation.clone(),
3229 )?;
3230
3231 plan.add_detail(format!("=== GROUP BY Phase Breakdown ==="));
3233 plan.add_detail(format!(
3234 "Phase 1 - Group Building: {:.3}ms",
3235 phase_info.phase2_key_building.as_secs_f64() * 1000.0
3236 ));
3237 plan.add_detail(format!(
3238 " • Processing {} rows into {} groups",
3239 phase_info.total_rows, phase_info.num_groups
3240 ));
3241 plan.add_detail(format!(
3242 "Phase 2 - Aggregation: {:.3}ms",
3243 phase_info.phase4_aggregation.as_secs_f64() * 1000.0
3244 ));
3245 if phase_info.phase4_having_evaluation > Duration::ZERO {
3246 plan.add_detail(format!(
3247 "Phase 3 - HAVING Filter: {:.3}ms",
3248 phase_info.phase4_having_evaluation.as_secs_f64() * 1000.0
3249 ));
3250 plan.add_detail(format!(
3251 " • Filtered {} groups",
3252 phase_info.groups_filtered_by_having
3253 ));
3254 }
3255 plan.add_detail(format!(
3256 "Total GROUP BY time: {:.3}ms",
3257 phase_info.total_time.as_secs_f64() * 1000.0
3258 ));
3259
3260 Ok(result_view)
3261 }
3262
3263 pub fn estimate_group_cardinality(
3266 &self,
3267 view: &DataView,
3268 group_by_exprs: &[SqlExpression],
3269 ) -> usize {
3270 let row_count = view.get_visible_rows().len();
3272 if row_count <= 100 {
3273 return row_count;
3274 }
3275
3276 let sample_size = min(1000, row_count / 10).max(100);
3278 let mut seen = FxHashSet::default();
3279
3280 let visible_rows = view.get_visible_rows();
3281 for (i, &row_idx) in visible_rows.iter().enumerate() {
3282 if i >= sample_size {
3283 break;
3284 }
3285
3286 let mut key_values = Vec::new();
3288 for expr in group_by_exprs {
3289 let mut evaluator = ArithmeticEvaluator::new(view.source());
3290 let value = evaluator.evaluate(expr, row_idx).unwrap_or(DataValue::Null);
3291 key_values.push(value);
3292 }
3293
3294 seen.insert(key_values);
3295 }
3296
3297 let sample_cardinality = seen.len();
3299 let estimated = (sample_cardinality * row_count) / sample_size;
3300
3301 estimated.min(row_count).max(sample_cardinality)
3303 }
3304}
3305
3306#[cfg(test)]
3307mod tests {
3308 use super::*;
3309 use crate::data::datatable::{DataColumn, DataRow, DataValue};
3310
3311 fn create_test_table() -> Arc<DataTable> {
3312 let mut table = DataTable::new("test");
3313
3314 table.add_column(DataColumn::new("id"));
3316 table.add_column(DataColumn::new("name"));
3317 table.add_column(DataColumn::new("age"));
3318
3319 table
3321 .add_row(DataRow::new(vec![
3322 DataValue::Integer(1),
3323 DataValue::String("Alice".to_string()),
3324 DataValue::Integer(30),
3325 ]))
3326 .unwrap();
3327
3328 table
3329 .add_row(DataRow::new(vec![
3330 DataValue::Integer(2),
3331 DataValue::String("Bob".to_string()),
3332 DataValue::Integer(25),
3333 ]))
3334 .unwrap();
3335
3336 table
3337 .add_row(DataRow::new(vec![
3338 DataValue::Integer(3),
3339 DataValue::String("Charlie".to_string()),
3340 DataValue::Integer(35),
3341 ]))
3342 .unwrap();
3343
3344 Arc::new(table)
3345 }
3346
3347 #[test]
3348 fn test_select_all() {
3349 let table = create_test_table();
3350 let engine = QueryEngine::new();
3351
3352 let view = engine
3353 .execute(table.clone(), "SELECT * FROM users")
3354 .unwrap();
3355 assert_eq!(view.row_count(), 3);
3356 assert_eq!(view.column_count(), 3);
3357 }
3358
3359 #[test]
3360 fn test_select_columns() {
3361 let table = create_test_table();
3362 let engine = QueryEngine::new();
3363
3364 let view = engine
3365 .execute(table.clone(), "SELECT name, age FROM users")
3366 .unwrap();
3367 assert_eq!(view.row_count(), 3);
3368 assert_eq!(view.column_count(), 2);
3369 }
3370
3371 #[test]
3372 fn test_select_with_limit() {
3373 let table = create_test_table();
3374 let engine = QueryEngine::new();
3375
3376 let view = engine
3377 .execute(table.clone(), "SELECT * FROM users LIMIT 2")
3378 .unwrap();
3379 assert_eq!(view.row_count(), 2);
3380 }
3381
3382 #[test]
3383 fn test_type_coercion_contains() {
3384 let _ = tracing_subscriber::fmt()
3386 .with_max_level(tracing::Level::DEBUG)
3387 .try_init();
3388
3389 let mut table = DataTable::new("test");
3390 table.add_column(DataColumn::new("id"));
3391 table.add_column(DataColumn::new("status"));
3392 table.add_column(DataColumn::new("price"));
3393
3394 table
3396 .add_row(DataRow::new(vec![
3397 DataValue::Integer(1),
3398 DataValue::String("Pending".to_string()),
3399 DataValue::Float(99.99),
3400 ]))
3401 .unwrap();
3402
3403 table
3404 .add_row(DataRow::new(vec![
3405 DataValue::Integer(2),
3406 DataValue::String("Confirmed".to_string()),
3407 DataValue::Float(150.50),
3408 ]))
3409 .unwrap();
3410
3411 table
3412 .add_row(DataRow::new(vec![
3413 DataValue::Integer(3),
3414 DataValue::String("Pending".to_string()),
3415 DataValue::Float(75.00),
3416 ]))
3417 .unwrap();
3418
3419 let table = Arc::new(table);
3420 let engine = QueryEngine::new();
3421
3422 println!("\n=== Testing WHERE clause with Contains ===");
3423 println!("Table has {} rows", table.row_count());
3424 for i in 0..table.row_count() {
3425 let status = table.get_value(i, 1);
3426 println!("Row {i}: status = {status:?}");
3427 }
3428
3429 println!("\n--- Test 1: status.Contains('pend') ---");
3431 let result = engine.execute(
3432 table.clone(),
3433 "SELECT * FROM test WHERE status.Contains('pend')",
3434 );
3435 match result {
3436 Ok(view) => {
3437 println!("SUCCESS: Found {} matching rows", view.row_count());
3438 assert_eq!(view.row_count(), 2); }
3440 Err(e) => {
3441 panic!("Query failed: {e}");
3442 }
3443 }
3444
3445 println!("\n--- Test 2: price.Contains('9') ---");
3447 let result = engine.execute(
3448 table.clone(),
3449 "SELECT * FROM test WHERE price.Contains('9')",
3450 );
3451 match result {
3452 Ok(view) => {
3453 println!(
3454 "SUCCESS: Found {} matching rows with price containing '9'",
3455 view.row_count()
3456 );
3457 assert!(view.row_count() >= 1);
3459 }
3460 Err(e) => {
3461 panic!("Numeric coercion query failed: {e}");
3462 }
3463 }
3464
3465 println!("\n=== All tests passed! ===");
3466 }
3467
3468 #[test]
3469 fn test_not_in_clause() {
3470 let _ = tracing_subscriber::fmt()
3472 .with_max_level(tracing::Level::DEBUG)
3473 .try_init();
3474
3475 let mut table = DataTable::new("test");
3476 table.add_column(DataColumn::new("id"));
3477 table.add_column(DataColumn::new("country"));
3478
3479 table
3481 .add_row(DataRow::new(vec![
3482 DataValue::Integer(1),
3483 DataValue::String("CA".to_string()),
3484 ]))
3485 .unwrap();
3486
3487 table
3488 .add_row(DataRow::new(vec![
3489 DataValue::Integer(2),
3490 DataValue::String("US".to_string()),
3491 ]))
3492 .unwrap();
3493
3494 table
3495 .add_row(DataRow::new(vec![
3496 DataValue::Integer(3),
3497 DataValue::String("UK".to_string()),
3498 ]))
3499 .unwrap();
3500
3501 let table = Arc::new(table);
3502 let engine = QueryEngine::new();
3503
3504 println!("\n=== Testing NOT IN clause ===");
3505 println!("Table has {} rows", table.row_count());
3506 for i in 0..table.row_count() {
3507 let country = table.get_value(i, 1);
3508 println!("Row {i}: country = {country:?}");
3509 }
3510
3511 println!("\n--- Test: country NOT IN ('CA') ---");
3513 let result = engine.execute(
3514 table.clone(),
3515 "SELECT * FROM test WHERE country NOT IN ('CA')",
3516 );
3517 match result {
3518 Ok(view) => {
3519 println!("SUCCESS: Found {} rows not in ('CA')", view.row_count());
3520 assert_eq!(view.row_count(), 2); }
3522 Err(e) => {
3523 panic!("NOT IN query failed: {e}");
3524 }
3525 }
3526
3527 println!("\n=== NOT IN test complete! ===");
3528 }
3529
3530 #[test]
3531 fn test_case_insensitive_in_and_not_in() {
3532 let _ = tracing_subscriber::fmt()
3534 .with_max_level(tracing::Level::DEBUG)
3535 .try_init();
3536
3537 let mut table = DataTable::new("test");
3538 table.add_column(DataColumn::new("id"));
3539 table.add_column(DataColumn::new("country"));
3540
3541 table
3543 .add_row(DataRow::new(vec![
3544 DataValue::Integer(1),
3545 DataValue::String("CA".to_string()), ]))
3547 .unwrap();
3548
3549 table
3550 .add_row(DataRow::new(vec![
3551 DataValue::Integer(2),
3552 DataValue::String("us".to_string()), ]))
3554 .unwrap();
3555
3556 table
3557 .add_row(DataRow::new(vec![
3558 DataValue::Integer(3),
3559 DataValue::String("UK".to_string()), ]))
3561 .unwrap();
3562
3563 let table = Arc::new(table);
3564
3565 println!("\n=== Testing Case-Insensitive IN clause ===");
3566 println!("Table has {} rows", table.row_count());
3567 for i in 0..table.row_count() {
3568 let country = table.get_value(i, 1);
3569 println!("Row {i}: country = {country:?}");
3570 }
3571
3572 println!("\n--- Test: country IN ('ca') with case_insensitive=true ---");
3574 let engine = QueryEngine::with_case_insensitive(true);
3575 let result = engine.execute(table.clone(), "SELECT * FROM test WHERE country IN ('ca')");
3576 match result {
3577 Ok(view) => {
3578 println!(
3579 "SUCCESS: Found {} rows matching 'ca' (case-insensitive)",
3580 view.row_count()
3581 );
3582 assert_eq!(view.row_count(), 1); }
3584 Err(e) => {
3585 panic!("Case-insensitive IN query failed: {e}");
3586 }
3587 }
3588
3589 println!("\n--- Test: country NOT IN ('ca') with case_insensitive=true ---");
3591 let result = engine.execute(
3592 table.clone(),
3593 "SELECT * FROM test WHERE country NOT IN ('ca')",
3594 );
3595 match result {
3596 Ok(view) => {
3597 println!(
3598 "SUCCESS: Found {} rows not matching 'ca' (case-insensitive)",
3599 view.row_count()
3600 );
3601 assert_eq!(view.row_count(), 2); }
3603 Err(e) => {
3604 panic!("Case-insensitive NOT IN query failed: {e}");
3605 }
3606 }
3607
3608 println!("\n--- Test: country IN ('ca') with case_insensitive=false ---");
3610 let engine_case_sensitive = QueryEngine::new(); let result = engine_case_sensitive
3612 .execute(table.clone(), "SELECT * FROM test WHERE country IN ('ca')");
3613 match result {
3614 Ok(view) => {
3615 println!(
3616 "SUCCESS: Found {} rows matching 'ca' (case-sensitive)",
3617 view.row_count()
3618 );
3619 assert_eq!(view.row_count(), 0); }
3621 Err(e) => {
3622 panic!("Case-sensitive IN query failed: {e}");
3623 }
3624 }
3625
3626 println!("\n=== Case-insensitive IN/NOT IN test complete! ===");
3627 }
3628
3629 #[test]
3630 #[ignore = "Parentheses in WHERE clause not yet implemented"]
3631 fn test_parentheses_in_where_clause() {
3632 let _ = tracing_subscriber::fmt()
3634 .with_max_level(tracing::Level::DEBUG)
3635 .try_init();
3636
3637 let mut table = DataTable::new("test");
3638 table.add_column(DataColumn::new("id"));
3639 table.add_column(DataColumn::new("status"));
3640 table.add_column(DataColumn::new("priority"));
3641
3642 table
3644 .add_row(DataRow::new(vec![
3645 DataValue::Integer(1),
3646 DataValue::String("Pending".to_string()),
3647 DataValue::String("High".to_string()),
3648 ]))
3649 .unwrap();
3650
3651 table
3652 .add_row(DataRow::new(vec![
3653 DataValue::Integer(2),
3654 DataValue::String("Complete".to_string()),
3655 DataValue::String("High".to_string()),
3656 ]))
3657 .unwrap();
3658
3659 table
3660 .add_row(DataRow::new(vec![
3661 DataValue::Integer(3),
3662 DataValue::String("Pending".to_string()),
3663 DataValue::String("Low".to_string()),
3664 ]))
3665 .unwrap();
3666
3667 table
3668 .add_row(DataRow::new(vec![
3669 DataValue::Integer(4),
3670 DataValue::String("Complete".to_string()),
3671 DataValue::String("Low".to_string()),
3672 ]))
3673 .unwrap();
3674
3675 let table = Arc::new(table);
3676 let engine = QueryEngine::new();
3677
3678 println!("\n=== Testing Parentheses in WHERE clause ===");
3679 println!("Table has {} rows", table.row_count());
3680 for i in 0..table.row_count() {
3681 let status = table.get_value(i, 1);
3682 let priority = table.get_value(i, 2);
3683 println!("Row {i}: status = {status:?}, priority = {priority:?}");
3684 }
3685
3686 println!("\n--- Test: (status = 'Pending' AND priority = 'High') OR (status = 'Complete' AND priority = 'Low') ---");
3688 let result = engine.execute(
3689 table.clone(),
3690 "SELECT * FROM test WHERE (status = 'Pending' AND priority = 'High') OR (status = 'Complete' AND priority = 'Low')",
3691 );
3692 match result {
3693 Ok(view) => {
3694 println!(
3695 "SUCCESS: Found {} rows with parenthetical logic",
3696 view.row_count()
3697 );
3698 assert_eq!(view.row_count(), 2); }
3700 Err(e) => {
3701 panic!("Parentheses query failed: {e}");
3702 }
3703 }
3704
3705 println!("\n=== Parentheses test complete! ===");
3706 }
3707
3708 #[test]
3709 #[ignore = "Numeric type coercion needs fixing"]
3710 fn test_numeric_type_coercion() {
3711 let _ = tracing_subscriber::fmt()
3713 .with_max_level(tracing::Level::DEBUG)
3714 .try_init();
3715
3716 let mut table = DataTable::new("test");
3717 table.add_column(DataColumn::new("id"));
3718 table.add_column(DataColumn::new("price"));
3719 table.add_column(DataColumn::new("quantity"));
3720
3721 table
3723 .add_row(DataRow::new(vec![
3724 DataValue::Integer(1),
3725 DataValue::Float(99.50), DataValue::Integer(100),
3727 ]))
3728 .unwrap();
3729
3730 table
3731 .add_row(DataRow::new(vec![
3732 DataValue::Integer(2),
3733 DataValue::Float(150.0), DataValue::Integer(200),
3735 ]))
3736 .unwrap();
3737
3738 table
3739 .add_row(DataRow::new(vec![
3740 DataValue::Integer(3),
3741 DataValue::Integer(75), DataValue::Integer(50),
3743 ]))
3744 .unwrap();
3745
3746 let table = Arc::new(table);
3747 let engine = QueryEngine::new();
3748
3749 println!("\n=== Testing Numeric Type Coercion ===");
3750 println!("Table has {} rows", table.row_count());
3751 for i in 0..table.row_count() {
3752 let price = table.get_value(i, 1);
3753 let quantity = table.get_value(i, 2);
3754 println!("Row {i}: price = {price:?}, quantity = {quantity:?}");
3755 }
3756
3757 println!("\n--- Test: price.Contains('.') ---");
3759 let result = engine.execute(
3760 table.clone(),
3761 "SELECT * FROM test WHERE price.Contains('.')",
3762 );
3763 match result {
3764 Ok(view) => {
3765 println!(
3766 "SUCCESS: Found {} rows with decimal points in price",
3767 view.row_count()
3768 );
3769 assert_eq!(view.row_count(), 2); }
3771 Err(e) => {
3772 panic!("Numeric Contains query failed: {e}");
3773 }
3774 }
3775
3776 println!("\n--- Test: quantity.Contains('0') ---");
3778 let result = engine.execute(
3779 table.clone(),
3780 "SELECT * FROM test WHERE quantity.Contains('0')",
3781 );
3782 match result {
3783 Ok(view) => {
3784 println!(
3785 "SUCCESS: Found {} rows with '0' in quantity",
3786 view.row_count()
3787 );
3788 assert_eq!(view.row_count(), 2); }
3790 Err(e) => {
3791 panic!("Integer Contains query failed: {e}");
3792 }
3793 }
3794
3795 println!("\n=== Numeric type coercion test complete! ===");
3796 }
3797
3798 #[test]
3799 fn test_datetime_comparisons() {
3800 let _ = tracing_subscriber::fmt()
3802 .with_max_level(tracing::Level::DEBUG)
3803 .try_init();
3804
3805 let mut table = DataTable::new("test");
3806 table.add_column(DataColumn::new("id"));
3807 table.add_column(DataColumn::new("created_date"));
3808
3809 table
3811 .add_row(DataRow::new(vec![
3812 DataValue::Integer(1),
3813 DataValue::String("2024-12-15".to_string()),
3814 ]))
3815 .unwrap();
3816
3817 table
3818 .add_row(DataRow::new(vec![
3819 DataValue::Integer(2),
3820 DataValue::String("2025-01-15".to_string()),
3821 ]))
3822 .unwrap();
3823
3824 table
3825 .add_row(DataRow::new(vec![
3826 DataValue::Integer(3),
3827 DataValue::String("2025-02-15".to_string()),
3828 ]))
3829 .unwrap();
3830
3831 let table = Arc::new(table);
3832 let engine = QueryEngine::new();
3833
3834 println!("\n=== Testing DateTime Comparisons ===");
3835 println!("Table has {} rows", table.row_count());
3836 for i in 0..table.row_count() {
3837 let date = table.get_value(i, 1);
3838 println!("Row {i}: created_date = {date:?}");
3839 }
3840
3841 println!("\n--- Test: created_date > DateTime(2025,1,1) ---");
3843 let result = engine.execute(
3844 table.clone(),
3845 "SELECT * FROM test WHERE created_date > DateTime(2025,1,1)",
3846 );
3847 match result {
3848 Ok(view) => {
3849 println!("SUCCESS: Found {} rows after 2025-01-01", view.row_count());
3850 assert_eq!(view.row_count(), 2); }
3852 Err(e) => {
3853 panic!("DateTime comparison query failed: {e}");
3854 }
3855 }
3856
3857 println!("\n=== DateTime comparison test complete! ===");
3858 }
3859
3860 #[test]
3861 fn test_not_with_method_calls() {
3862 let _ = tracing_subscriber::fmt()
3864 .with_max_level(tracing::Level::DEBUG)
3865 .try_init();
3866
3867 let mut table = DataTable::new("test");
3868 table.add_column(DataColumn::new("id"));
3869 table.add_column(DataColumn::new("status"));
3870
3871 table
3873 .add_row(DataRow::new(vec![
3874 DataValue::Integer(1),
3875 DataValue::String("Pending Review".to_string()),
3876 ]))
3877 .unwrap();
3878
3879 table
3880 .add_row(DataRow::new(vec![
3881 DataValue::Integer(2),
3882 DataValue::String("Complete".to_string()),
3883 ]))
3884 .unwrap();
3885
3886 table
3887 .add_row(DataRow::new(vec![
3888 DataValue::Integer(3),
3889 DataValue::String("Pending Approval".to_string()),
3890 ]))
3891 .unwrap();
3892
3893 let table = Arc::new(table);
3894 let engine = QueryEngine::with_case_insensitive(true);
3895
3896 println!("\n=== Testing NOT with Method Calls ===");
3897 println!("Table has {} rows", table.row_count());
3898 for i in 0..table.row_count() {
3899 let status = table.get_value(i, 1);
3900 println!("Row {i}: status = {status:?}");
3901 }
3902
3903 println!("\n--- Test: NOT status.Contains('pend') ---");
3905 let result = engine.execute(
3906 table.clone(),
3907 "SELECT * FROM test WHERE NOT status.Contains('pend')",
3908 );
3909 match result {
3910 Ok(view) => {
3911 println!(
3912 "SUCCESS: Found {} rows NOT containing 'pend'",
3913 view.row_count()
3914 );
3915 assert_eq!(view.row_count(), 1); }
3917 Err(e) => {
3918 panic!("NOT Contains query failed: {e}");
3919 }
3920 }
3921
3922 println!("\n--- Test: NOT status.StartsWith('Pending') ---");
3924 let result = engine.execute(
3925 table.clone(),
3926 "SELECT * FROM test WHERE NOT status.StartsWith('Pending')",
3927 );
3928 match result {
3929 Ok(view) => {
3930 println!(
3931 "SUCCESS: Found {} rows NOT starting with 'Pending'",
3932 view.row_count()
3933 );
3934 assert_eq!(view.row_count(), 1); }
3936 Err(e) => {
3937 panic!("NOT StartsWith query failed: {e}");
3938 }
3939 }
3940
3941 println!("\n=== NOT with method calls test complete! ===");
3942 }
3943
3944 #[test]
3945 #[ignore = "Complex logical expressions with parentheses not yet implemented"]
3946 fn test_complex_logical_expressions() {
3947 let _ = tracing_subscriber::fmt()
3949 .with_max_level(tracing::Level::DEBUG)
3950 .try_init();
3951
3952 let mut table = DataTable::new("test");
3953 table.add_column(DataColumn::new("id"));
3954 table.add_column(DataColumn::new("status"));
3955 table.add_column(DataColumn::new("priority"));
3956 table.add_column(DataColumn::new("assigned"));
3957
3958 table
3960 .add_row(DataRow::new(vec![
3961 DataValue::Integer(1),
3962 DataValue::String("Pending".to_string()),
3963 DataValue::String("High".to_string()),
3964 DataValue::String("John".to_string()),
3965 ]))
3966 .unwrap();
3967
3968 table
3969 .add_row(DataRow::new(vec![
3970 DataValue::Integer(2),
3971 DataValue::String("Complete".to_string()),
3972 DataValue::String("High".to_string()),
3973 DataValue::String("Jane".to_string()),
3974 ]))
3975 .unwrap();
3976
3977 table
3978 .add_row(DataRow::new(vec![
3979 DataValue::Integer(3),
3980 DataValue::String("Pending".to_string()),
3981 DataValue::String("Low".to_string()),
3982 DataValue::String("John".to_string()),
3983 ]))
3984 .unwrap();
3985
3986 table
3987 .add_row(DataRow::new(vec![
3988 DataValue::Integer(4),
3989 DataValue::String("In Progress".to_string()),
3990 DataValue::String("Medium".to_string()),
3991 DataValue::String("Jane".to_string()),
3992 ]))
3993 .unwrap();
3994
3995 let table = Arc::new(table);
3996 let engine = QueryEngine::new();
3997
3998 println!("\n=== Testing Complex Logical Expressions ===");
3999 println!("Table has {} rows", table.row_count());
4000 for i in 0..table.row_count() {
4001 let status = table.get_value(i, 1);
4002 let priority = table.get_value(i, 2);
4003 let assigned = table.get_value(i, 3);
4004 println!(
4005 "Row {i}: status = {status:?}, priority = {priority:?}, assigned = {assigned:?}"
4006 );
4007 }
4008
4009 println!("\n--- Test: status = 'Pending' AND (priority = 'High' OR assigned = 'John') ---");
4011 let result = engine.execute(
4012 table.clone(),
4013 "SELECT * FROM test WHERE status = 'Pending' AND (priority = 'High' OR assigned = 'John')",
4014 );
4015 match result {
4016 Ok(view) => {
4017 println!(
4018 "SUCCESS: Found {} rows with complex logic",
4019 view.row_count()
4020 );
4021 assert_eq!(view.row_count(), 2); }
4023 Err(e) => {
4024 panic!("Complex logic query failed: {e}");
4025 }
4026 }
4027
4028 println!("\n--- Test: NOT (status.Contains('Complete') OR priority = 'Low') ---");
4030 let result = engine.execute(
4031 table.clone(),
4032 "SELECT * FROM test WHERE NOT (status.Contains('Complete') OR priority = 'Low')",
4033 );
4034 match result {
4035 Ok(view) => {
4036 println!(
4037 "SUCCESS: Found {} rows with NOT complex logic",
4038 view.row_count()
4039 );
4040 assert_eq!(view.row_count(), 2); }
4042 Err(e) => {
4043 panic!("NOT complex logic query failed: {e}");
4044 }
4045 }
4046
4047 println!("\n=== Complex logical expressions test complete! ===");
4048 }
4049
4050 #[test]
4051 fn test_mixed_data_types_and_edge_cases() {
4052 let _ = tracing_subscriber::fmt()
4054 .with_max_level(tracing::Level::DEBUG)
4055 .try_init();
4056
4057 let mut table = DataTable::new("test");
4058 table.add_column(DataColumn::new("id"));
4059 table.add_column(DataColumn::new("value"));
4060 table.add_column(DataColumn::new("nullable_field"));
4061
4062 table
4064 .add_row(DataRow::new(vec![
4065 DataValue::Integer(1),
4066 DataValue::String("123.45".to_string()),
4067 DataValue::String("present".to_string()),
4068 ]))
4069 .unwrap();
4070
4071 table
4072 .add_row(DataRow::new(vec![
4073 DataValue::Integer(2),
4074 DataValue::Float(678.90),
4075 DataValue::Null,
4076 ]))
4077 .unwrap();
4078
4079 table
4080 .add_row(DataRow::new(vec![
4081 DataValue::Integer(3),
4082 DataValue::Boolean(true),
4083 DataValue::String("also present".to_string()),
4084 ]))
4085 .unwrap();
4086
4087 table
4088 .add_row(DataRow::new(vec![
4089 DataValue::Integer(4),
4090 DataValue::String("false".to_string()),
4091 DataValue::Null,
4092 ]))
4093 .unwrap();
4094
4095 let table = Arc::new(table);
4096 let engine = QueryEngine::new();
4097
4098 println!("\n=== Testing Mixed Data Types and Edge Cases ===");
4099 println!("Table has {} rows", table.row_count());
4100 for i in 0..table.row_count() {
4101 let value = table.get_value(i, 1);
4102 let nullable = table.get_value(i, 2);
4103 println!("Row {i}: value = {value:?}, nullable_field = {nullable:?}");
4104 }
4105
4106 println!("\n--- Test: value.Contains('true') (boolean to string coercion) ---");
4108 let result = engine.execute(
4109 table.clone(),
4110 "SELECT * FROM test WHERE value.Contains('true')",
4111 );
4112 match result {
4113 Ok(view) => {
4114 println!(
4115 "SUCCESS: Found {} rows with boolean coercion",
4116 view.row_count()
4117 );
4118 assert_eq!(view.row_count(), 1); }
4120 Err(e) => {
4121 panic!("Boolean coercion query failed: {e}");
4122 }
4123 }
4124
4125 println!("\n--- Test: id IN (1, 3) ---");
4127 let result = engine.execute(table.clone(), "SELECT * FROM test WHERE id IN (1, 3)");
4128 match result {
4129 Ok(view) => {
4130 println!("SUCCESS: Found {} rows with IN clause", view.row_count());
4131 assert_eq!(view.row_count(), 2); }
4133 Err(e) => {
4134 panic!("Multiple IN values query failed: {e}");
4135 }
4136 }
4137
4138 println!("\n=== Mixed data types test complete! ===");
4139 }
4140
4141 #[test]
4143 fn test_aggregate_only_single_row() {
4144 let table = create_test_stock_data();
4145 let engine = QueryEngine::new();
4146
4147 let result = engine
4149 .execute(
4150 table.clone(),
4151 "SELECT COUNT(*), MIN(close), MAX(close), AVG(close) FROM stock",
4152 )
4153 .expect("Query should succeed");
4154
4155 assert_eq!(
4156 result.row_count(),
4157 1,
4158 "Aggregate-only query should return exactly 1 row"
4159 );
4160 assert_eq!(result.column_count(), 4, "Should have 4 aggregate columns");
4161
4162 let source = result.source();
4164 let row = source.get_row(0).expect("Should have first row");
4165
4166 assert_eq!(row.values[0], DataValue::Integer(5));
4168
4169 assert_eq!(row.values[1], DataValue::Float(99.5));
4171
4172 assert_eq!(row.values[2], DataValue::Float(105.0));
4174
4175 if let DataValue::Float(avg) = &row.values[3] {
4177 assert!(
4178 (avg - 102.4).abs() < 0.01,
4179 "Average should be approximately 102.4, got {}",
4180 avg
4181 );
4182 } else {
4183 panic!("AVG should return a Float value");
4184 }
4185 }
4186
4187 #[test]
4189 fn test_single_aggregate_single_row() {
4190 let table = create_test_stock_data();
4191 let engine = QueryEngine::new();
4192
4193 let result = engine
4194 .execute(table.clone(), "SELECT COUNT(*) FROM stock")
4195 .expect("Query should succeed");
4196
4197 assert_eq!(
4198 result.row_count(),
4199 1,
4200 "Single aggregate query should return exactly 1 row"
4201 );
4202 assert_eq!(result.column_count(), 1, "Should have 1 column");
4203
4204 let source = result.source();
4205 let row = source.get_row(0).expect("Should have first row");
4206 assert_eq!(row.values[0], DataValue::Integer(5));
4207 }
4208
4209 #[test]
4211 fn test_aggregate_with_where_single_row() {
4212 let table = create_test_stock_data();
4213 let engine = QueryEngine::new();
4214
4215 let result = engine
4217 .execute(
4218 table.clone(),
4219 "SELECT COUNT(*), MIN(close), MAX(close) FROM stock WHERE close >= 103.0",
4220 )
4221 .expect("Query should succeed");
4222
4223 assert_eq!(
4224 result.row_count(),
4225 1,
4226 "Filtered aggregate query should return exactly 1 row"
4227 );
4228 assert_eq!(result.column_count(), 3, "Should have 3 aggregate columns");
4229
4230 let source = result.source();
4231 let row = source.get_row(0).expect("Should have first row");
4232
4233 assert_eq!(row.values[0], DataValue::Integer(2));
4235 assert_eq!(row.values[1], DataValue::Float(103.5)); assert_eq!(row.values[2], DataValue::Float(105.0)); }
4238
4239 #[test]
4240 fn test_not_in_parsing() {
4241 use crate::sql::recursive_parser::Parser;
4242
4243 let query = "SELECT * FROM test WHERE country NOT IN ('CA')";
4244 println!("\n=== Testing NOT IN parsing ===");
4245 println!("Parsing query: {query}");
4246
4247 let mut parser = Parser::new(query);
4248 match parser.parse() {
4249 Ok(statement) => {
4250 println!("Parsed statement: {statement:#?}");
4251 if let Some(where_clause) = statement.where_clause {
4252 println!("WHERE conditions: {:#?}", where_clause.conditions);
4253 if let Some(first_condition) = where_clause.conditions.first() {
4254 println!("First condition expression: {:#?}", first_condition.expr);
4255 }
4256 }
4257 }
4258 Err(e) => {
4259 panic!("Parse error: {e}");
4260 }
4261 }
4262 }
4263
4264 fn create_test_stock_data() -> Arc<DataTable> {
4266 let mut table = DataTable::new("stock");
4267
4268 table.add_column(DataColumn::new("symbol"));
4269 table.add_column(DataColumn::new("close"));
4270 table.add_column(DataColumn::new("volume"));
4271
4272 let test_data = vec![
4274 ("AAPL", 99.5, 1000),
4275 ("AAPL", 101.2, 1500),
4276 ("AAPL", 103.5, 2000),
4277 ("AAPL", 105.0, 1200),
4278 ("AAPL", 102.8, 1800),
4279 ];
4280
4281 for (symbol, close, volume) in test_data {
4282 table
4283 .add_row(DataRow::new(vec![
4284 DataValue::String(symbol.to_string()),
4285 DataValue::Float(close),
4286 DataValue::Integer(volume),
4287 ]))
4288 .expect("Should add row successfully");
4289 }
4290
4291 Arc::new(table)
4292 }
4293}
4294
4295#[cfg(test)]
4296#[path = "query_engine_tests.rs"]
4297mod query_engine_tests;