ddb_core/engine/
executor.rs

1// Query executor for SELECT, INSERT, UPDATE, DELETE statements
2use crate::config::{Table, TableCatalog};
3use crate::engine::{evaluator::Evaluator, row::Row};
4use crate::error::{DdbError, Result};
5use crate::file_io::{CsvReader, FileLock};
6use crate::functions::Value;
7use crate::parser::{DeleteStatement, InsertStatement, JoinClause, JoinType, OrderDirection, SelectColumn, SelectStatement, UpdateStatement, UpsertStatement};
8use std::fs::{File, OpenOptions};
9use std::io::{BufRead, BufReader, BufWriter, Write};
10
11pub struct QueryExecutor {
12    evaluator: Evaluator,
13}
14
15impl QueryExecutor {
16    pub fn new() -> Self {
17        QueryExecutor {
18            evaluator: Evaluator::new(),
19        }
20    }
21
22    /// Execute a SELECT statement with optional catalog for JOIN support
23    pub fn execute_select(&self, stmt: &SelectStatement, table: &Table) -> Result<Vec<Row>> {
24        self.execute_select_with_catalog(stmt, table, None)
25    }
26
27    /// Execute a SELECT statement with catalog support for JOINs
28    pub fn execute_select_with_catalog(
29        &self,
30        stmt: &SelectStatement,
31        table: &Table,
32        catalog: Option<&TableCatalog>,
33    ) -> Result<Vec<Row>> {
34        // Check if query uses aggregate functions or GROUP BY
35        let has_aggregates = self.has_aggregate_functions(&stmt.columns);
36        let has_group_by = !stmt.group_by.is_empty();
37
38        if has_aggregates || has_group_by {
39            return self.execute_aggregate_select_with_catalog(stmt, table, catalog);
40        }
41
42        // Read data from main table
43        let mut reader = CsvReader::new(&table.data_file, table.field_delimiter, true)?;
44        let mut rows = Vec::new();
45        while let Some(row) = reader.next_row()? {
46            rows.push(row);
47        }
48
49        // Process JOINs if any
50        if !stmt.joins.is_empty() {
51            rows = self.execute_joins(rows, &stmt.joins, catalog)?;
52        }
53
54        // Filter rows based on WHERE clause
55        rows.retain(|row| {
56            if let Some(where_expr) = &stmt.where_clause {
57                if let Ok(result) = self.evaluator.evaluate(where_expr, row) {
58                    if let Ok(matches) = self.to_boolean(&result) {
59                        return matches;
60                    }
61                }
62            }
63            stmt.where_clause.is_none()
64        });
65
66        // Project columns (SELECT clause)
67        rows = rows
68            .into_iter()
69            .filter_map(|row| self.project_row(&row, &stmt.columns).ok())
70            .collect();
71
72        // Apply DISTINCT
73        if stmt.distinct {
74            rows = self.deduplicate_rows(rows);
75        }
76
77        // Apply ORDER BY with optional LIMIT optimization
78        if !stmt.order_by.is_empty() {
79            // Heap-based optimization: If we have LIMIT without OFFSET, use a heap to avoid full sort
80            if let Some(limit) = &stmt.limit {
81                if limit.offset.is_none() || limit.offset == Some(0) {
82                    // Use heap-based top-K algorithm for better performance
83                    rows = self.sort_rows_with_limit(rows, &stmt.order_by, limit.count)?;
84                } else {
85                    // Has OFFSET - need full sort
86                    rows = self.sort_rows(rows, &stmt.order_by)?;
87                    let start = limit.offset.unwrap_or(0);
88                    rows = rows.into_iter().skip(start).take(limit.count).collect();
89                }
90            } else {
91                // No LIMIT - full sort required
92                rows = self.sort_rows(rows, &stmt.order_by)?;
93            }
94        } else if let Some(limit) = &stmt.limit {
95            // LIMIT without ORDER BY - just take first N rows
96            let start = limit.offset.unwrap_or(0);
97            rows = rows.into_iter().skip(start).take(limit.count).collect();
98        }
99
100        Ok(rows)
101    }
102
103    /// Check if SELECT columns contain aggregate functions
104    fn has_aggregate_functions(&self, columns: &[SelectColumn]) -> bool {
105        columns.iter().any(|col| match col {
106            SelectColumn::Function { name, .. } => {
107                matches!(
108                    name.to_uppercase().as_str(),
109                    "COUNT" | "SUM" | "AVG" | "MIN" | "MAX" | "STDDEV" | "VARIANCE" | "GROUP_CONCAT"
110                )
111            }
112            _ => false,
113        })
114    }
115
116    /// Execute JOINs on the result set
117    fn execute_joins(
118        &self,
119        left_rows: Vec<Row>,
120        joins: &[JoinClause],
121        catalog: Option<&TableCatalog>,
122    ) -> Result<Vec<Row>> {
123        let mut result = left_rows;
124
125        for join in joins {
126            // Get the table to join with
127            let catalog = catalog.ok_or_else(|| {
128                DdbError::ExecutionError(
129                    "JOIN requires table catalog. Use execute_select_with_catalog.".to_string(),
130                )
131            })?;
132
133            let right_table = catalog.get_table(&join.table).ok_or_else(|| {
134                DdbError::ExecutionError(format!("Table '{}' not found for JOIN", join.table))
135            })?;
136
137            // Read all rows from the right table
138            let mut right_reader =
139                CsvReader::new(&right_table.data_file, right_table.field_delimiter, true)?;
140            let mut right_rows = Vec::new();
141            while let Some(row) = right_reader.next_row()? {
142                right_rows.push(row);
143            }
144
145            // Perform the join with index optimization
146            result = self.perform_join_indexed(result, right_rows, join)?;
147        }
148
149        Ok(result)
150    }
151
152    /// Perform a single join operation
153    fn perform_join(
154        &self,
155        left_rows: Vec<Row>,
156        right_rows: Vec<Row>,
157        join: &JoinClause,
158    ) -> Result<Vec<Row>> {
159        let mut joined_rows = Vec::new();
160
161        match join.join_type {
162            JoinType::Inner => {
163                // INNER JOIN: Only rows where ON condition matches
164                for left_row in &left_rows {
165                    for right_row in &right_rows {
166                        let combined = self.merge_rows(left_row, right_row);
167                        let result = self.evaluator.evaluate(&join.on_condition, &combined)?;
168                        if self.to_boolean(&result)? {
169                            joined_rows.push(combined);
170                        }
171                    }
172                }
173            }
174            JoinType::Left => {
175                // LEFT JOIN: All rows from left, with NULLs for right if no match
176                for left_row in &left_rows {
177                    let mut matched = false;
178                    for right_row in &right_rows {
179                        let combined = self.merge_rows(left_row, right_row);
180                        let result = self.evaluator.evaluate(&join.on_condition, &combined)?;
181                        if self.to_boolean(&result)? {
182                            joined_rows.push(combined);
183                            matched = true;
184                        }
185                    }
186                    if !matched {
187                        // No match found, add left row with NULLs for right columns
188                        let row_with_nulls = left_row.clone();
189                        // Add NULL values for right table columns
190                        // (In practice, the projection step will handle missing columns)
191                        joined_rows.push(row_with_nulls);
192                    }
193                }
194            }
195            JoinType::Right => {
196                // RIGHT JOIN: All rows from right, with NULLs for left if no match
197                for right_row in &right_rows {
198                    let mut matched = false;
199                    for left_row in &left_rows {
200                        let combined = self.merge_rows(left_row, right_row);
201                        let result = self.evaluator.evaluate(&join.on_condition, &combined)?;
202                        if self.to_boolean(&result)? {
203                            joined_rows.push(combined);
204                            matched = true;
205                        }
206                    }
207                    if !matched {
208                        // No match found, add right row with NULLs for left columns
209                        joined_rows.push(right_row.clone());
210                    }
211                }
212            }
213            JoinType::Full => {
214                // FULL OUTER JOIN: All rows from both tables
215                let mut matched_left = vec![false; left_rows.len()];
216                let mut matched_right = vec![false; right_rows.len()];
217
218                for (left_idx, left_row) in left_rows.iter().enumerate() {
219                    for (right_idx, right_row) in right_rows.iter().enumerate() {
220                        let combined = self.merge_rows(left_row, right_row);
221                        let result = self.evaluator.evaluate(&join.on_condition, &combined)?;
222                        if self.to_boolean(&result)? {
223                            joined_rows.push(combined);
224                            matched_left[left_idx] = true;
225                            matched_right[right_idx] = true;
226                        }
227                    }
228                }
229
230                // Add unmatched left rows
231                for (idx, left_row) in left_rows.iter().enumerate() {
232                    if !matched_left[idx] {
233                        joined_rows.push(left_row.clone());
234                    }
235                }
236
237                // Add unmatched right rows
238                for (idx, right_row) in right_rows.iter().enumerate() {
239                    if !matched_right[idx] {
240                        joined_rows.push(right_row.clone());
241                    }
242                }
243            }
244        }
245
246        Ok(joined_rows)
247    }
248
249    /// Perform a join with index optimization
250    /// Tries to use hash index for equality joins, falls back to nested loop for complex conditions
251    fn perform_join_indexed(
252        &self,
253        left_rows: Vec<Row>,
254        right_rows: Vec<Row>,
255        join: &JoinClause,
256    ) -> Result<Vec<Row>> {
257        use crate::engine::index::HashIndex;
258
259        // Try to detect simple equality condition for index optimization
260        // Pattern: left_col = right_col (or right_col = left_col)
261        let index_info = self.try_extract_equality_join(&join.on_condition);
262
263        if let Some((left_col, right_col)) = index_info {
264            // We can use index optimization!
265            // Build hash index on the right table's join column
266            let right_index = HashIndex::build(&right_col, &right_rows);
267
268            let mut joined_rows = Vec::new();
269
270            match join.join_type {
271                JoinType::Inner => {
272                    // For each left row, use index to find matching right rows
273                    for left_row in &left_rows {
274                        if let Some(left_value) = left_row.get(&left_col) {
275                            if let Some(matching_indices) = right_index.lookup(left_value) {
276                                for &right_idx in matching_indices {
277                                    let right_row = &right_rows[right_idx];
278                                    let combined = self.merge_rows(left_row, right_row);
279
280                                    // Still need to evaluate full condition in case there are additional clauses
281                                    let result = self.evaluator.evaluate(&join.on_condition, &combined)?;
282                                    if self.to_boolean(&result)? {
283                                        joined_rows.push(combined);
284                                    }
285                                }
286                            }
287                        }
288                    }
289                }
290                JoinType::Left => {
291                    // LEFT JOIN: All rows from left, with NULLs for right if no match
292                    for left_row in &left_rows {
293                        let mut matched = false;
294                        if let Some(left_value) = left_row.get(&left_col) {
295                            if let Some(matching_indices) = right_index.lookup(left_value) {
296                                for &right_idx in matching_indices {
297                                    let right_row = &right_rows[right_idx];
298                                    let combined = self.merge_rows(left_row, right_row);
299
300                                    let result = self.evaluator.evaluate(&join.on_condition, &combined)?;
301                                    if self.to_boolean(&result)? {
302                                        joined_rows.push(combined);
303                                        matched = true;
304                                    }
305                                }
306                            }
307                        }
308                        if !matched {
309                            joined_rows.push(left_row.clone());
310                        }
311                    }
312                }
313                JoinType::Right => {
314                    // RIGHT JOIN: Build index on left table instead
315                    let left_index = HashIndex::build(&left_col, &left_rows);
316                    let mut matched_right = vec![false; right_rows.len()];
317
318                    for (right_idx, right_row) in right_rows.iter().enumerate() {
319                        if let Some(right_value) = right_row.get(&right_col) {
320                            if let Some(matching_indices) = left_index.lookup(right_value) {
321                                for &left_idx in matching_indices {
322                                    let left_row = &left_rows[left_idx];
323                                    let combined = self.merge_rows(left_row, right_row);
324
325                                    let result = self.evaluator.evaluate(&join.on_condition, &combined)?;
326                                    if self.to_boolean(&result)? {
327                                        joined_rows.push(combined);
328                                        matched_right[right_idx] = true;
329                                    }
330                                }
331                            }
332                        }
333                    }
334
335                    // Add unmatched right rows
336                    for (idx, right_row) in right_rows.iter().enumerate() {
337                        if !matched_right[idx] {
338                            joined_rows.push(right_row.clone());
339                        }
340                    }
341                }
342                JoinType::Full => {
343                    // FULL OUTER JOIN: Need to track both sides
344                    let mut matched_left = vec![false; left_rows.len()];
345                    let mut matched_right = vec![false; right_rows.len()];
346
347                    // Use index on right table
348                    for (left_idx, left_row) in left_rows.iter().enumerate() {
349                        if let Some(left_value) = left_row.get(&left_col) {
350                            if let Some(matching_indices) = right_index.lookup(left_value) {
351                                for &right_idx in matching_indices {
352                                    let right_row = &right_rows[right_idx];
353                                    let combined = self.merge_rows(left_row, right_row);
354
355                                    let result = self.evaluator.evaluate(&join.on_condition, &combined)?;
356                                    if self.to_boolean(&result)? {
357                                        joined_rows.push(combined);
358                                        matched_left[left_idx] = true;
359                                        matched_right[right_idx] = true;
360                                    }
361                                }
362                            }
363                        }
364                    }
365
366                    // Add unmatched left rows
367                    for (idx, left_row) in left_rows.iter().enumerate() {
368                        if !matched_left[idx] {
369                            joined_rows.push(left_row.clone());
370                        }
371                    }
372
373                    // Add unmatched right rows
374                    for (idx, right_row) in right_rows.iter().enumerate() {
375                        if !matched_right[idx] {
376                            joined_rows.push(right_row.clone());
377                        }
378                    }
379                }
380            }
381
382            Ok(joined_rows)
383        } else {
384            // Can't use index - fall back to nested loop
385            self.perform_join(left_rows, right_rows, join)
386        }
387    }
388
389    /// Try to extract simple equality join condition (e.g., "a.id = b.user_id")
390    /// Returns Some((left_column, right_column)) if it's a simple equality, None otherwise
391    fn try_extract_equality_join(&self, expr: &crate::parser::Expression) -> Option<(String, String)> {
392        use crate::parser::{BinaryOperator, Expression};
393
394        if let Expression::BinaryOp { left, op, right } = expr {
395            // Check if it's an equality operator
396            if matches!(op, BinaryOperator::Equal) {
397                // Both sides must be column references
398                if let (Expression::Column(left_col), Expression::Column(right_col)) = (left.as_ref(), right.as_ref()) {
399                    return Some((left_col.clone(), right_col.clone()));
400                }
401            }
402        }
403
404        None
405    }
406
407    /// Merge two rows into one (for JOINs)
408    fn merge_rows(&self, left: &Row, right: &Row) -> Row {
409        let mut merged = left.clone();
410        for (name, value) in right.as_map() {
411            merged.set(name, value.clone());
412        }
413        merged
414    }
415
416    /// Execute SELECT with aggregate functions
417    #[allow(dead_code)]
418    fn execute_aggregate_select(&self, stmt: &SelectStatement, table: &Table) -> Result<Vec<Row>> {
419        self.execute_aggregate_select_with_catalog(stmt, table, None)
420    }
421
422    /// Execute SELECT with aggregate functions and catalog support
423    fn execute_aggregate_select_with_catalog(
424        &self,
425        stmt: &SelectStatement,
426        table: &Table,
427        catalog: Option<&TableCatalog>,
428    ) -> Result<Vec<Row>> {
429        // Read all rows from main table
430        let mut reader = CsvReader::new(&table.data_file, table.field_delimiter, true)?;
431        let mut all_rows = Vec::new();
432        while let Some(row) = reader.next_row()? {
433            all_rows.push(row);
434        }
435
436        // Process JOINs if any
437        if !stmt.joins.is_empty() {
438            all_rows = self.execute_joins(all_rows, &stmt.joins, catalog)?;
439        }
440
441        // Filter rows based on WHERE clause
442        all_rows.retain(|row| {
443            if let Some(where_expr) = &stmt.where_clause {
444                if let Ok(result) = self.evaluator.evaluate(where_expr, row) {
445                    if let Ok(matches) = self.to_boolean(&result) {
446                        return matches;
447                    }
448                }
449                return false;
450            }
451            true
452        });
453
454        // If GROUP BY is present, group rows and aggregate
455        if !stmt.group_by.is_empty() {
456            return self.execute_group_by_aggregation(all_rows, stmt);
457        }
458
459        // No GROUP BY - aggregate all rows into one result
460        let mut result_row = Row::new();
461
462        for col_spec in &stmt.columns {
463            match col_spec {
464                SelectColumn::Function { name, args, alias } => {
465                    // Special handling for COUNT(*)
466                    let aggregate_result = if self.is_count_star(name, args) {
467                        // COUNT(*) - return row count directly
468                        Value::Integer(all_rows.len() as i64)
469                    } else {
470                        // Collect values for this column
471                        let values: Result<Vec<Value>> = all_rows
472                            .iter()
473                            .map(|row| {
474                                if args.len() == 1 {
475                                    self.evaluator.evaluate(&args[0], row)
476                                } else {
477                                    Err(DdbError::ExecutionError(
478                                        "Aggregate functions require exactly one argument".to_string(),
479                                    ))
480                                }
481                            })
482                            .collect();
483
484                        let values = values?;
485
486                        // Apply aggregate function
487                        match name.to_uppercase().as_str() {
488                            "COUNT" => crate::functions::aggregate::count(&values, true)?,
489                            "SUM" => crate::functions::aggregate::sum(&values)?,
490                            "AVG" => crate::functions::aggregate::avg(&values)?,
491                            "MIN" => crate::functions::aggregate::min(&values)?,
492                            "MAX" => crate::functions::aggregate::max(&values)?,
493                            "STDDEV" | "STDDEV_POP" => crate::functions::aggregate::stddev_pop(&values)?,
494                            "VARIANCE" | "VAR_POP" => crate::functions::aggregate::var_pop(&values)?,
495                            _ => {
496                                return Err(DdbError::ExecutionError(format!(
497                                    "Unknown aggregate function: {}",
498                                    name
499                                )))
500                            }
501                        }
502                    };
503
504                    let default_name = format!("{}(...)", name);
505                    let output_name = alias.as_ref().unwrap_or(&default_name);
506                    result_row.set(output_name, aggregate_result);
507                }
508                SelectColumn::Column { name, alias } => {
509                    // Non-aggregate column in aggregate query - use first value
510                    // (This is technically not standard SQL without GROUP BY, but we'll allow it)
511                    let value = all_rows
512                        .first()
513                        .and_then(|row| row.get(name).cloned())
514                        .unwrap_or(Value::Null);
515                    let output_name = alias.as_ref().unwrap_or(name);
516                    result_row.set(output_name, value);
517                }
518                SelectColumn::Wildcard => {
519                    return Err(DdbError::ExecutionError(
520                        "Cannot use * with aggregate functions (use COUNT(*) instead)".to_string(),
521                    ));
522                }
523            }
524        }
525
526        Ok(vec![result_row])
527    }
528
529    /// Execute GROUP BY aggregation
530    fn execute_group_by_aggregation(
531        &self,
532        all_rows: Vec<Row>,
533        stmt: &SelectStatement,
534    ) -> Result<Vec<Row>> {
535        use std::collections::HashMap;
536
537        // Group rows by GROUP BY columns
538        let mut groups: HashMap<Vec<String>, Vec<Row>> = HashMap::new();
539
540        for row in all_rows {
541            // Build group key from GROUP BY columns
542            let mut group_key = Vec::new();
543            for col_name in &stmt.group_by {
544                let value = row.get(col_name)
545                    .map(|v| v.to_string())
546                    .unwrap_or_else(|| "NULL".to_string());
547                group_key.push(value);
548            }
549
550            groups.entry(group_key).or_insert_with(Vec::new).push(row);
551        }
552
553        // Build result rows - one per group
554        let mut result_rows = Vec::new();
555
556        for (_group_key, group_rows) in groups {
557            let mut result_row = Row::new();
558
559            // Add GROUP BY columns to result
560            for (_idx, col_name) in stmt.group_by.iter().enumerate() {
561                if let Some(value) = group_rows.first().and_then(|row| row.get(col_name)) {
562                    result_row.set(col_name, value.clone());
563                }
564            }
565
566            // Compute aggregates for this group
567            for col_spec in &stmt.columns {
568                match col_spec {
569                    SelectColumn::Function { name, args, alias } => {
570                        let aggregate_result = if self.is_count_star(name, args) {
571                            Value::Integer(group_rows.len() as i64)
572                        } else {
573                            // Collect values for this column in this group
574                            let values: Result<Vec<Value>> = group_rows
575                                .iter()
576                                .map(|row| {
577                                    if args.len() == 1 {
578                                        self.evaluator.evaluate(&args[0], row)
579                                    } else {
580                                        Err(DdbError::ExecutionError(
581                                            "Aggregate functions require exactly one argument".to_string(),
582                                        ))
583                                    }
584                                })
585                                .collect();
586
587                            let values = values?;
588
589                            // Apply aggregate function
590                            match name.to_uppercase().as_str() {
591                                "COUNT" => crate::functions::aggregate::count(&values, true)?,
592                                "SUM" => crate::functions::aggregate::sum(&values)?,
593                                "AVG" => crate::functions::aggregate::avg(&values)?,
594                                "MIN" => crate::functions::aggregate::min(&values)?,
595                                "MAX" => crate::functions::aggregate::max(&values)?,
596                                "STDDEV" | "STDDEV_POP" => crate::functions::aggregate::stddev_pop(&values)?,
597                                "VARIANCE" | "VAR_POP" => crate::functions::aggregate::var_pop(&values)?,
598                                _ => {
599                                    return Err(DdbError::ExecutionError(format!(
600                                        "Unknown aggregate function: {}",
601                                        name
602                                    )))
603                                }
604                            }
605                        };
606
607                        let default_name = format!("{}(...)", name);
608                        let output_name = alias.as_ref().unwrap_or(&default_name);
609                        result_row.set(output_name, aggregate_result);
610                    }
611                    SelectColumn::Column { name, alias } => {
612                        // Non-aggregate column - should be in GROUP BY
613                        if !stmt.group_by.contains(name) {
614                            return Err(DdbError::ExecutionError(format!(
615                                "Column '{}' must appear in GROUP BY clause or be used in an aggregate function",
616                                name
617                            )));
618                        }
619                        // Already added above in GROUP BY section
620                        if let Some(alias_name) = alias {
621                            if let Some(value) = result_row.get(name) {
622                                result_row.set(alias_name, value.clone());
623                            }
624                        }
625                    }
626                    SelectColumn::Wildcard => {
627                        return Err(DdbError::ExecutionError(
628                            "Cannot use * with GROUP BY (select specific columns)".to_string(),
629                        ));
630                    }
631                }
632            }
633
634            // Apply HAVING clause to filter groups
635            if let Some(having_expr) = &stmt.having {
636                let result = self.evaluator.evaluate(having_expr, &result_row)?;
637                let matches = self.to_boolean(&result)?;
638                if !matches {
639                    continue; // Skip this group
640                }
641            }
642
643            result_rows.push(result_row);
644        }
645
646        Ok(result_rows)
647    }
648
649    /// Check if this is COUNT(*)
650    fn is_count_star(&self, name: &str, args: &[crate::parser::Expression]) -> bool {
651        name.to_uppercase() == "COUNT"
652            && args.len() == 1
653            && matches!(&args[0], crate::parser::Expression::Column(col) if col == "*")
654    }
655
656    /// Project columns from a row based on SELECT clause
657    fn project_row(&self, row: &Row, columns: &[SelectColumn]) -> Result<Row> {
658        let mut new_row = Row::new();
659
660        for col_spec in columns {
661            match col_spec {
662                SelectColumn::Wildcard => {
663                    // Include all columns
664                    for (name, value) in row.as_map() {
665                        new_row.set(name, value.clone());
666                    }
667                }
668                SelectColumn::Column { name, alias } => {
669                    // Check if this is a system variable (@@VARIABLE)
670                    let value = if name.starts_with("@@") {
671                        let var_name = &name[2..];
672                        let var_expr = crate::parser::Expression::SystemVariable(var_name.to_string());
673                        self.evaluator.evaluate(&var_expr, row)?
674                    } else {
675                        row.get(name)
676                            .cloned()
677                            .ok_or_else(|| DdbError::ColumnNotFound(name.clone()))?
678                    };
679                    let output_name = alias.as_ref().unwrap_or(name);
680                    new_row.set(output_name, value);
681                }
682                SelectColumn::Function { name, args, alias } => {
683                    // Evaluate function
684                    let func_expr = crate::parser::Expression::Function {
685                        name: name.clone(),
686                        args: args.clone(),
687                    };
688                    let result = self.evaluator.evaluate(&func_expr, row)?;
689
690                    let default_name = format!("{}(...)", name);
691                    let output_name = alias.as_ref().unwrap_or(&default_name);
692                    new_row.set(output_name, result);
693                }
694            }
695        }
696
697        Ok(new_row)
698    }
699
700    /// Remove duplicate rows
701    fn deduplicate_rows(&self, mut rows: Vec<Row>) -> Vec<Row> {
702        let mut seen = std::collections::HashSet::new();
703        rows.retain(|row| {
704            let key = self.row_to_key(row);
705            seen.insert(key)
706        });
707        rows
708    }
709
710    /// Convert row to a hashable key for deduplication
711    fn row_to_key(&self, row: &Row) -> Vec<String> {
712        let mut keys: Vec<_> = row.column_names();
713        keys.sort();
714        keys.iter()
715            .map(|name| {
716                row.get(name)
717                    .map(|v| v.to_string())
718                    .unwrap_or_else(|| "NULL".to_string())
719            })
720            .collect()
721    }
722
723    /// Sort rows based on ORDER BY clause
724    fn sort_rows(
725        &self,
726        mut rows: Vec<Row>,
727        order_by: &[crate::parser::OrderByColumn],
728    ) -> Result<Vec<Row>> {
729        rows.sort_by(|a, b| {
730            for order_col in order_by {
731                let a_val = a.get(&order_col.column).cloned().unwrap_or(Value::Null);
732                let b_val = b.get(&order_col.column).cloned().unwrap_or(Value::Null);
733
734                let cmp = self.compare_values(&a_val, &b_val);
735
736                let cmp = match order_col.direction {
737                    OrderDirection::Asc => cmp,
738                    OrderDirection::Desc => cmp.reverse(),
739                };
740
741                if cmp != std::cmp::Ordering::Equal {
742                    return cmp;
743                }
744            }
745            std::cmp::Ordering::Equal
746        });
747
748        Ok(rows)
749    }
750
751    /// Sort rows with LIMIT optimization using partial sorting
752    /// This uses select_nth_unstable for O(n) partitioning + O(k log k) sorting = O(n + k log k)
753    fn sort_rows_with_limit(
754        &self,
755        mut rows: Vec<Row>,
756        order_by: &[crate::parser::OrderByColumn],
757        limit: usize,
758    ) -> Result<Vec<Row>> {
759        use std::cmp::Ordering;
760
761        // If limit is 0, return empty
762        if limit == 0 {
763            return Ok(Vec::new());
764        }
765
766        // If we have fewer rows than the limit, just do a full sort
767        if rows.len() <= limit {
768            return self.sort_rows(rows, order_by);
769        }
770
771        // If limit is very close to total size, full sort is better
772        if limit * 2 > rows.len() {
773            return self.sort_rows(rows, order_by);
774        }
775
776        // Create a comparator closure
777        let cmp_fn = |a: &Row, b: &Row| -> Ordering {
778            for order_col in order_by {
779                let a_val = a.get(&order_col.column).cloned().unwrap_or(Value::Null);
780                let b_val = b.get(&order_col.column).cloned().unwrap_or(Value::Null);
781
782                let cmp = self.compare_values(&a_val, &b_val);
783
784                let cmp = match order_col.direction {
785                    OrderDirection::Asc => cmp,
786                    OrderDirection::Desc => cmp.reverse(),
787                };
788
789                if cmp != Ordering::Equal {
790                    return cmp;
791                }
792            }
793            Ordering::Equal
794        };
795
796        // Use select_nth_unstable to partition the array so that the first `limit` elements
797        // are the smallest, without fully sorting them. This is O(n) average case.
798        rows.select_nth_unstable_by(limit, |a, b| cmp_fn(a, b));
799
800        // Now take the first `limit` elements and sort them properly
801        rows.truncate(limit);
802        rows.sort_by(|a, b| cmp_fn(a, b));
803
804        Ok(rows)
805    }
806
807    /// Compare two values for sorting
808    fn compare_values(&self, left: &Value, right: &Value) -> std::cmp::Ordering {
809        use std::cmp::Ordering;
810
811        match (left, right) {
812            (Value::Null, Value::Null) => Ordering::Equal,
813            (Value::Null, _) => Ordering::Less,
814            (_, Value::Null) => Ordering::Greater,
815            (Value::Integer(a), Value::Integer(b)) => a.cmp(b),
816            (Value::Float(a), Value::Float(b)) => a.partial_cmp(b).unwrap_or(Ordering::Equal),
817            (Value::Integer(a), Value::Float(b)) => {
818                (*a as f64).partial_cmp(b).unwrap_or(Ordering::Equal)
819            }
820            (Value::Float(a), Value::Integer(b)) => {
821                a.partial_cmp(&(*b as f64)).unwrap_or(Ordering::Equal)
822            }
823            (Value::String(a), Value::String(b)) => a.cmp(b),
824            (Value::Boolean(a), Value::Boolean(b)) => a.cmp(b),
825            (Value::Date(a), Value::Date(b)) => a.cmp(b),
826            (Value::DateTime(a), Value::DateTime(b)) => a.cmp(b),
827            (Value::Time(a), Value::Time(b)) => a.cmp(b),
828            _ => Ordering::Equal,
829        }
830    }
831
832    /// Convert value to boolean
833    fn to_boolean(&self, value: &Value) -> Result<bool> {
834        match value {
835            Value::Boolean(b) => Ok(*b),
836            Value::Integer(i) => Ok(*i != 0),
837            Value::Float(f) => Ok(*f != 0.0),
838            Value::String(s) => Ok(!s.is_empty()),
839            Value::Null => Ok(false),
840            _ => Ok(true),
841        }
842    }
843
844    /// Execute an INSERT statement
845    pub fn execute_insert(&self, stmt: &InsertStatement, table: &Table) -> Result<usize> {
846        // Acquire exclusive lock on the file
847        let lock = FileLock::new(&table.data_file)?;
848        lock.lock_exclusive()?;
849
850        // Open file in append mode
851        let file = OpenOptions::new()
852            .append(true)
853            .open(&table.data_file)?;
854
855        let mut writer = BufWriter::new(file);
856        let mut rows_inserted = 0;
857
858        // Get column names from table or from INSERT statement
859        let _column_names = if stmt.columns.is_empty() {
860            // No columns specified - use table columns
861            table.columns.iter().map(|c| c.name.clone()).collect::<Vec<_>>()
862        } else {
863            stmt.columns.clone()
864        };
865
866        // Process each row of values
867        for value_row in &stmt.values {
868            if !stmt.columns.is_empty() && value_row.len() != stmt.columns.len() {
869                return Err(DdbError::ExecutionError(format!(
870                    "Column count ({}) doesn't match value count ({})",
871                    stmt.columns.len(),
872                    value_row.len()
873                )));
874            }
875
876            // Build CSV row
877            let mut csv_row = Vec::new();
878            for expr in value_row {
879                // Evaluate expression to get value
880                let empty_row = Row::new();
881                let value = self.evaluator.evaluate(expr, &empty_row)?;
882                csv_row.push(self.value_to_csv_string(&value));
883            }
884
885            // Write row to file
886            writeln!(writer, "{}", csv_row.join(&table.field_delimiter.to_string()))?;
887
888            rows_inserted += 1;
889        }
890
891        writer.flush()?;
892
893        // Lock is automatically released when dropped
894        Ok(rows_inserted)
895    }
896
897    /// Execute an UPSERT statement
898    /// If a row with the key_column value exists, update it; otherwise insert it
899    pub fn execute_upsert(&self, stmt: &UpsertStatement, table: &Table) -> Result<(usize, usize)> {
900        // Acquire exclusive lock on the file
901        let lock = FileLock::new(&table.data_file)?;
902        lock.lock_exclusive()?;
903
904        // Read all rows
905        let file = File::open(&table.data_file)?;
906        let reader = BufReader::new(file);
907        let mut lines: Vec<String> = reader.lines().collect::<std::result::Result<Vec<_>, _>>()?;
908
909        if lines.is_empty() {
910            return Err(DdbError::ExecutionError("Cannot UPSERT into empty file".to_string()));
911        }
912
913        // Parse header
914        let header = lines[0].clone();
915        let column_names: Vec<String> = header
916            .split(table.field_delimiter)
917            .map(|s| s.trim().to_string())
918            .collect();
919
920        // Verify key column exists
921        let key_col_idx = column_names
922            .iter()
923            .position(|c| c == &stmt.key_column)
924            .ok_or_else(|| {
925                DdbError::ExecutionError(format!(
926                    "Key column '{}' not found in table",
927                    stmt.key_column
928                ))
929            })?;
930
931        // Verify stmt columns match
932        if stmt.columns.len() != column_names.len() {
933            return Err(DdbError::ExecutionError(format!(
934                "Column count mismatch: UPSERT specifies {} columns, table has {}",
935                stmt.columns.len(),
936                column_names.len()
937            )));
938        }
939
940        let mut rows_inserted = 0;
941        let mut rows_updated = 0;
942
943        // Process each value row in the UPSERT
944        for value_row in &stmt.values {
945            if value_row.len() != stmt.columns.len() {
946                return Err(DdbError::ExecutionError(format!(
947                    "Value count ({}) doesn't match column count ({})",
948                    value_row.len(),
949                    stmt.columns.len()
950                )));
951            }
952
953            // Evaluate expressions to get values
954            let empty_row = Row::new();
955            let mut values = Vec::new();
956            for expr in value_row {
957                let value = self.evaluator.evaluate(expr, &empty_row)?;
958                values.push(value);
959            }
960
961            // Get key value (value corresponding to key_column)
962            let key_col_pos = stmt
963                .columns
964                .iter()
965                .position(|c| c == &stmt.key_column)
966                .ok_or_else(|| {
967                    DdbError::ExecutionError(format!(
968                        "Key column '{}' not found in UPSERT columns",
969                        stmt.key_column
970                    ))
971                })?;
972            let key_value_str = self.value_to_csv_string(&values[key_col_pos]);
973
974            // Search for existing row with this key
975            let mut found_row_idx = None;
976            for i in (table.data_starts_on as usize + 1)..lines.len() {
977                let line = &lines[i];
978                if line.trim().is_empty() {
979                    continue;
980                }
981
982                let row_values: Vec<String> = line
983                    .split(table.field_delimiter)
984                    .map(|s| s.trim().to_string())
985                    .collect();
986
987                if key_col_idx < row_values.len() && row_values[key_col_idx] == key_value_str {
988                    found_row_idx = Some(i);
989                    break;
990                }
991            }
992
993            // Build CSV row
994            let csv_row: Vec<String> = values.iter().map(|v| self.value_to_csv_string(v)).collect();
995            let csv_line = csv_row.join(&table.field_delimiter.to_string());
996
997            if let Some(idx) = found_row_idx {
998                // Update existing row
999                lines[idx] = csv_line;
1000                rows_updated += 1;
1001            } else {
1002                // Insert new row
1003                lines.push(csv_line);
1004                rows_inserted += 1;
1005            }
1006        }
1007
1008        // Rewrite file
1009        let file = OpenOptions::new()
1010            .write(true)
1011            .truncate(true)
1012            .open(&table.data_file)?;
1013
1014        let mut writer = BufWriter::new(file);
1015        for line in &lines {
1016            writeln!(writer, "{}", line)?;
1017        }
1018        writer.flush()?;
1019
1020        Ok((rows_inserted, rows_updated))
1021    }
1022
1023    /// Execute an UPDATE statement
1024    pub fn execute_update(&self, stmt: &UpdateStatement, table: &Table) -> Result<usize> {
1025        // Acquire exclusive lock on the file
1026        let lock = FileLock::new(&table.data_file)?;
1027        lock.lock_exclusive()?;
1028
1029        // Read all rows
1030        let file = File::open(&table.data_file)?;
1031        let reader = BufReader::new(file);
1032        let mut lines: Vec<String> = reader.lines().collect::<std::result::Result<Vec<_>, _>>()?;
1033
1034        if lines.is_empty() {
1035            return Ok(0);
1036        }
1037
1038        // Parse header
1039        let header = lines[0].clone();
1040        let column_names: Vec<String> = header
1041            .split(table.field_delimiter)
1042            .map(|s| s.trim().to_string())
1043            .collect();
1044
1045        let mut rows_updated = 0;
1046
1047        // Process data rows
1048        for i in (table.data_starts_on as usize + 1)..lines.len() {
1049            let line = &lines[i];
1050
1051            // Skip empty lines
1052            if line.trim().is_empty() {
1053                continue;
1054            }
1055
1056            // Parse row
1057            let values: Vec<String> = line
1058                .split(table.field_delimiter)
1059                .map(|s| s.trim().to_string())
1060                .collect();
1061
1062            let mut row = Row::new();
1063            for (idx, col_name) in column_names.iter().enumerate() {
1064                if idx < values.len() {
1065                    row.set(col_name, self.parse_value(&values[idx]));
1066                }
1067            }
1068
1069            // Check if row matches WHERE clause
1070            let matches = if let Some(where_expr) = &stmt.where_clause {
1071                let result = self.evaluator.evaluate(where_expr, &row)?;
1072                self.to_boolean(&result)?
1073            } else {
1074                true // No WHERE clause, update all rows
1075            };
1076
1077            if matches {
1078                // Update the row
1079                for (column, value_expr) in &stmt.assignments {
1080                    let new_value = self.evaluator.evaluate(value_expr, &row)?;
1081                    row.set(column, new_value);
1082                }
1083
1084                // Rebuild CSV line
1085                let mut csv_values = Vec::new();
1086                for col_name in &column_names {
1087                    if let Some(value) = row.get(col_name) {
1088                        csv_values.push(self.value_to_csv_string(value));
1089                    } else {
1090                        csv_values.push(String::new());
1091                    }
1092                }
1093
1094                lines[i] = csv_values.join(&table.field_delimiter.to_string());
1095                rows_updated += 1;
1096            }
1097        }
1098
1099        // Rewrite file
1100        let file = OpenOptions::new()
1101            .write(true)
1102            .truncate(true)
1103            .open(&table.data_file)?;
1104
1105        let mut writer = BufWriter::new(file);
1106        for line in &lines {
1107            writeln!(writer, "{}", line)?;
1108        }
1109        writer.flush()?;
1110
1111        Ok(rows_updated)
1112    }
1113
1114    /// Execute a DELETE statement
1115    pub fn execute_delete(&self, stmt: &DeleteStatement, table: &Table) -> Result<usize> {
1116        // Acquire exclusive lock on the file
1117        let lock = FileLock::new(&table.data_file)?;
1118        lock.lock_exclusive()?;
1119
1120        // Read all rows
1121        let file = File::open(&table.data_file)?;
1122        let reader = BufReader::new(file);
1123        let lines: Vec<String> = reader.lines().collect::<std::result::Result<Vec<_>, _>>()?;
1124
1125        if lines.is_empty() {
1126            return Ok(0);
1127        }
1128
1129        // Parse header
1130        let header = lines[0].clone();
1131        let column_names: Vec<String> = header
1132            .split(table.field_delimiter)
1133            .map(|s| s.trim().to_string())
1134            .collect();
1135
1136        let mut kept_lines = vec![header]; // Keep header
1137        let mut rows_deleted = 0;
1138
1139        // Process data rows
1140        for i in (table.data_starts_on as usize + 1)..lines.len() {
1141            let line = &lines[i];
1142
1143            // Skip empty lines
1144            if line.trim().is_empty() {
1145                continue;
1146            }
1147
1148            // Parse row
1149            let values: Vec<String> = line
1150                .split(table.field_delimiter)
1151                .map(|s| s.trim().to_string())
1152                .collect();
1153
1154            let mut row = Row::new();
1155            for (idx, col_name) in column_names.iter().enumerate() {
1156                if idx < values.len() {
1157                    row.set(col_name, self.parse_value(&values[idx]));
1158                }
1159            }
1160
1161            // Check if row matches WHERE clause
1162            let should_delete = if let Some(where_expr) = &stmt.where_clause {
1163                let result = self.evaluator.evaluate(where_expr, &row)?;
1164                self.to_boolean(&result)?
1165            } else {
1166                true // No WHERE clause, delete all rows
1167            };
1168
1169            if should_delete {
1170                rows_deleted += 1;
1171            } else {
1172                kept_lines.push(line.clone());
1173            }
1174        }
1175
1176        // Rewrite file with kept lines only
1177        let file = OpenOptions::new()
1178            .write(true)
1179            .truncate(true)
1180            .open(&table.data_file)?;
1181
1182        let mut writer = BufWriter::new(file);
1183        for line in &kept_lines {
1184            writeln!(writer, "{}", line)?;
1185        }
1186        writer.flush()?;
1187
1188        Ok(rows_deleted)
1189    }
1190
1191    /// Convert a Value to CSV string representation
1192    fn value_to_csv_string(&self, value: &Value) -> String {
1193        match value {
1194            Value::String(s) => {
1195                // Escape quotes and wrap in quotes if contains delimiter or quotes
1196                if s.contains(',') || s.contains('"') || s.contains('\n') {
1197                    format!("\"{}\"", s.replace('"', "\"\""))
1198                } else {
1199                    s.clone()
1200                }
1201            }
1202            Value::Integer(i) => i.to_string(),
1203            Value::Float(f) => f.to_string(),
1204            Value::Boolean(b) => b.to_string(),
1205            Value::Null => String::new(),
1206            Value::Date(d) => d.format("%Y-%m-%d").to_string(),
1207            Value::DateTime(dt) => dt.format("%Y-%m-%d %H:%M:%S").to_string(),
1208            Value::Time(t) => t.format("%H:%M:%S").to_string(),
1209        }
1210    }
1211
1212    /// Parse a string value into a Value
1213    fn parse_value(&self, s: &str) -> Value {
1214        if s.is_empty() {
1215            return Value::Null;
1216        }
1217
1218        // Try parsing as integer
1219        if let Ok(i) = s.parse::<i64>() {
1220            return Value::Integer(i);
1221        }
1222
1223        // Try parsing as float
1224        if let Ok(f) = s.parse::<f64>() {
1225            return Value::Float(f);
1226        }
1227
1228        // Try parsing as boolean
1229        match s.to_lowercase().as_str() {
1230            "true" | "t" | "yes" | "y" | "1" => return Value::Boolean(true),
1231            "false" | "f" | "no" | "n" | "0" => return Value::Boolean(false),
1232            _ => {}
1233        }
1234
1235        // Default to string
1236        Value::String(s.to_string())
1237    }
1238}
1239
1240impl Default for QueryExecutor {
1241    fn default() -> Self {
1242        Self::new()
1243    }
1244}
1245
1246#[cfg(test)]
1247mod tests {
1248    use super::*;
1249    use crate::parser::{BinaryOperator, Expression, Literal};
1250    use std::io::Write;
1251    use tempfile::NamedTempFile;
1252
1253    fn create_test_table() -> (Table, NamedTempFile) {
1254        let mut file = NamedTempFile::new().unwrap();
1255        writeln!(file, "id,name,age").unwrap();
1256        writeln!(file, "1,Alice,30").unwrap();
1257        writeln!(file, "2,Bob,25").unwrap();
1258        writeln!(file, "3,Charlie,35").unwrap();
1259        file.flush().unwrap();
1260
1261        let table = Table {
1262            name: "users".to_string(),
1263            database: "test".to_string(),
1264            data_file: file.path().to_string_lossy().to_string(),
1265            columns: vec![],
1266            field_delimiter: ',',
1267            data_starts_on: 0,
1268            comment_char: None,
1269        };
1270
1271        (table, file)
1272    }
1273
1274    #[test]
1275    fn test_execute_select_all() {
1276        let (table, _file) = create_test_table();
1277        let executor = QueryExecutor::new();
1278
1279        let stmt = SelectStatement {
1280            distinct: false,
1281            columns: vec![SelectColumn::Wildcard],
1282            from: Some("users".to_string()),
1283            joins: vec![],
1284            where_clause: None,
1285            group_by: vec![],
1286            having: None,
1287            order_by: vec![],
1288            limit: None,
1289        };
1290
1291        let results = executor.execute_select(&stmt, &table).unwrap();
1292        assert_eq!(results.len(), 3);
1293    }
1294
1295    #[test]
1296    fn test_execute_select_with_where() {
1297        let (table, _file) = create_test_table();
1298        let executor = QueryExecutor::new();
1299
1300        // WHERE age > 25
1301        let where_expr = Expression::BinaryOp {
1302            left: Box::new(Expression::Column("age".to_string())),
1303            op: BinaryOperator::GreaterThan,
1304            right: Box::new(Expression::Literal(Literal::Integer(25))),
1305        };
1306
1307        let stmt = SelectStatement {
1308            distinct: false,
1309            columns: vec![SelectColumn::Wildcard],
1310            from: Some("users".to_string()),
1311            joins: vec![],
1312            where_clause: Some(where_expr),
1313            group_by: vec![],
1314            having: None,
1315            order_by: vec![],
1316            limit: None,
1317        };
1318
1319        let results = executor.execute_select(&stmt, &table).unwrap();
1320        assert_eq!(results.len(), 2); // Alice (30) and Charlie (35)
1321    }
1322
1323    #[test]
1324    fn test_execute_select_with_columns() {
1325        let (table, _file) = create_test_table();
1326        let executor = QueryExecutor::new();
1327
1328        let stmt = SelectStatement {
1329            distinct: false,
1330            columns: vec![
1331                SelectColumn::Column {
1332                    name: "name".to_string(),
1333                    alias: None,
1334                },
1335                SelectColumn::Column {
1336                    name: "age".to_string(),
1337                    alias: None,
1338                },
1339            ],
1340            from: Some("users".to_string()),
1341            joins: vec![],
1342            where_clause: None,
1343            group_by: vec![],
1344            having: None,
1345            order_by: vec![],
1346            limit: None,
1347        };
1348
1349        let results = executor.execute_select(&stmt, &table).unwrap();
1350        assert_eq!(results.len(), 3);
1351
1352        // Check that only selected columns are present
1353        let row = &results[0];
1354        assert!(row.get("name").is_some());
1355        assert!(row.get("age").is_some());
1356        // id should not be present
1357        assert!(row.get("id").is_none());
1358    }
1359
1360    #[test]
1361    fn test_execute_select_with_limit() {
1362        let (table, _file) = create_test_table();
1363        let executor = QueryExecutor::new();
1364
1365        let stmt = SelectStatement {
1366            distinct: false,
1367            columns: vec![SelectColumn::Wildcard],
1368            from: Some("users".to_string()),
1369            joins: vec![],
1370            where_clause: None,
1371            group_by: vec![],
1372            having: None,
1373            order_by: vec![],
1374            limit: Some(crate::parser::Limit {
1375                offset: None,
1376                count: 2,
1377            }),
1378        };
1379
1380        let results = executor.execute_select(&stmt, &table).unwrap();
1381        assert_eq!(results.len(), 2);
1382    }
1383
1384    #[test]
1385    fn test_execute_select_with_order_by() {
1386        let (table, _file) = create_test_table();
1387        let executor = QueryExecutor::new();
1388
1389        let stmt = SelectStatement {
1390            distinct: false,
1391            columns: vec![SelectColumn::Wildcard],
1392            from: Some("users".to_string()),
1393            joins: vec![],
1394            where_clause: None,
1395            group_by: vec![],
1396            having: None,
1397            order_by: vec![crate::parser::OrderByColumn {
1398                column: "age".to_string(),
1399                direction: OrderDirection::Desc,
1400            }],
1401            limit: None,
1402        };
1403
1404        let results = executor.execute_select(&stmt, &table).unwrap();
1405        assert_eq!(results.len(), 3);
1406
1407        // Check order: Charlie (35), Alice (30), Bob (25)
1408        assert_eq!(
1409            results[0].get("name"),
1410            Some(&Value::String("Charlie".to_string()))
1411        );
1412        assert_eq!(
1413            results[2].get("name"),
1414            Some(&Value::String("Bob".to_string()))
1415        );
1416    }
1417
1418    #[test]
1419    fn test_indexed_join() {
1420        use crate::file_io::CsvReader;
1421
1422        // Create left table (users)
1423        let mut users_file = NamedTempFile::new().unwrap();
1424        writeln!(users_file, "id,name").unwrap();
1425        writeln!(users_file, "1,Alice").unwrap();
1426        writeln!(users_file, "2,Bob").unwrap();
1427        writeln!(users_file, "3,Charlie").unwrap();
1428        users_file.flush().unwrap();
1429
1430        let users_table = Table {
1431            name: "users".to_string(),
1432            database: "test".to_string(),
1433            data_file: users_file.path().to_string_lossy().to_string(),
1434            columns: vec![],
1435            field_delimiter: ',',
1436            data_starts_on: 0,
1437            comment_char: None,
1438        };
1439
1440        // Create right table (orders)
1441        let mut orders_file = NamedTempFile::new().unwrap();
1442        writeln!(orders_file, "order_id,user_id,amount").unwrap();
1443        writeln!(orders_file, "101,1,50").unwrap();
1444        writeln!(orders_file, "102,2,75").unwrap();
1445        writeln!(orders_file, "103,1,100").unwrap();
1446        orders_file.flush().unwrap();
1447
1448        // Read users data
1449        let mut users_reader = CsvReader::new(&users_table.data_file, ',', true).unwrap();
1450        let mut left_rows = Vec::new();
1451        while let Some(row) = users_reader.next_row().unwrap() {
1452            left_rows.push(row);
1453        }
1454
1455        // Read orders data
1456        let orders_path = orders_file.path().to_string_lossy().to_string();
1457        let mut orders_reader = CsvReader::new(&orders_path, ',', true).unwrap();
1458        let mut right_rows = Vec::new();
1459        while let Some(row) = orders_reader.next_row().unwrap() {
1460            right_rows.push(row);
1461        }
1462
1463        // Create JOIN condition: id = user_id
1464        let join_condition = Expression::BinaryOp {
1465            left: Box::new(Expression::Column("id".to_string())),
1466            op: BinaryOperator::Equal,
1467            right: Box::new(Expression::Column("user_id".to_string())),
1468        };
1469
1470        let join_clause = JoinClause {
1471            join_type: JoinType::Inner,
1472            table: "orders".to_string(),
1473            on_condition: join_condition,
1474        };
1475
1476        // Test the indexed join directly
1477        let executor = QueryExecutor::new();
1478        let results = executor
1479            .perform_join_indexed(left_rows, right_rows, &join_clause)
1480            .unwrap();
1481
1482        // Should have 3 rows (Alice has 2 orders, Bob has 1, Charlie has 0)
1483        assert_eq!(results.len(), 3);
1484
1485        // Verify JOIN worked correctly
1486        let alice_orders: Vec<_> = results
1487            .iter()
1488            .filter(|r| r.get("name") == Some(&Value::String("Alice".to_string())))
1489            .collect();
1490        assert_eq!(alice_orders.len(), 2);
1491
1492        let bob_orders: Vec<_> = results
1493            .iter()
1494            .filter(|r| r.get("name") == Some(&Value::String("Bob".to_string())))
1495            .collect();
1496        assert_eq!(bob_orders.len(), 1);
1497
1498        // Charlie should have 0 orders (INNER JOIN excludes unmatched)
1499        let charlie_orders: Vec<_> = results
1500            .iter()
1501            .filter(|r| r.get("name") == Some(&Value::String("Charlie".to_string())))
1502            .collect();
1503        assert_eq!(charlie_orders.len(), 0);
1504    }
1505}
ddb_core/engine/executor.rs

ddb_core/engine/
executor.rs
