1#![allow(missing_docs)]
37
38use crate::error::{IoError, Result};
39use serde::{Deserialize, Serialize};
40use std::cmp::Ordering;
41use std::collections::HashMap;
42
43#[derive(Debug, Clone, PartialEq, Eq, Hash, Serialize, Deserialize)]
47pub enum ColumnType {
48 Int64,
50 Float64,
52 Utf8,
54 Boolean,
56 Nullable(Box<ColumnType>),
58}
59
60impl ColumnType {
61 pub fn as_str(&self) -> &str {
63 match self {
64 ColumnType::Int64 => "int64",
65 ColumnType::Float64 => "float64",
66 ColumnType::Utf8 => "utf8",
67 ColumnType::Boolean => "boolean",
68 ColumnType::Nullable(_) => "nullable",
69 }
70 }
71}
72
73#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
75pub enum ColumnValue {
76 Null,
78 Int(i64),
80 Float(f64),
82 Utf8(String),
84 Boolean(bool),
86}
87
88impl ColumnValue {
89 pub fn column_type(&self) -> ColumnType {
91 match self {
92 ColumnValue::Null => ColumnType::Nullable(Box::new(ColumnType::Utf8)),
93 ColumnValue::Int(_) => ColumnType::Int64,
94 ColumnValue::Float(_) => ColumnType::Float64,
95 ColumnValue::Utf8(_) => ColumnType::Utf8,
96 ColumnValue::Boolean(_) => ColumnType::Boolean,
97 }
98 }
99
100 pub fn as_f64(&self) -> Option<f64> {
102 match self {
103 ColumnValue::Float(f) => Some(*f),
104 ColumnValue::Int(i) => Some(*i as f64),
105 _ => None,
106 }
107 }
108
109 pub fn as_i64(&self) -> Option<i64> {
111 match self {
112 ColumnValue::Int(i) => Some(*i),
113 ColumnValue::Float(f) if f.fract() == 0.0 => Some(*f as i64),
114 _ => None,
115 }
116 }
117
118 pub fn partial_cmp_value(&self, other: &Self) -> Option<Ordering> {
120 match (self, other) {
121 (ColumnValue::Int(a), ColumnValue::Int(b)) => a.partial_cmp(b),
122 (ColumnValue::Float(a), ColumnValue::Float(b)) => a.partial_cmp(b),
123 (ColumnValue::Int(a), ColumnValue::Float(b)) => (*a as f64).partial_cmp(b),
124 (ColumnValue::Float(a), ColumnValue::Int(b)) => a.partial_cmp(&(*b as f64)),
125 (ColumnValue::Utf8(a), ColumnValue::Utf8(b)) => a.partial_cmp(b),
126 (ColumnValue::Boolean(a), ColumnValue::Boolean(b)) => a.partial_cmp(b),
127 (ColumnValue::Null, ColumnValue::Null) => Some(Ordering::Equal),
128 (ColumnValue::Null, _) => Some(Ordering::Less),
129 (_, ColumnValue::Null) => Some(Ordering::Greater),
130 _ => None,
131 }
132 }
133
134 pub fn to_json(&self) -> serde_json::Value {
136 match self {
137 ColumnValue::Null => serde_json::Value::Null,
138 ColumnValue::Int(i) => serde_json::json!(i),
139 ColumnValue::Float(f) => serde_json::json!(f),
140 ColumnValue::Utf8(s) => serde_json::json!(s),
141 ColumnValue::Boolean(b) => serde_json::json!(b),
142 }
143 }
144
145 pub fn from_json(v: &serde_json::Value, col_type: &ColumnType) -> Self {
147 match (col_type, v) {
148 (_, serde_json::Value::Null) => ColumnValue::Null,
149 (ColumnType::Int64, serde_json::Value::Number(n)) => {
150 ColumnValue::Int(n.as_i64().unwrap_or_default())
151 }
152 (ColumnType::Float64, serde_json::Value::Number(n)) => {
153 ColumnValue::Float(n.as_f64().unwrap_or_default())
154 }
155 (ColumnType::Utf8, serde_json::Value::String(s)) => ColumnValue::Utf8(s.clone()),
156 (ColumnType::Boolean, serde_json::Value::Bool(b)) => ColumnValue::Boolean(*b),
157 (ColumnType::Nullable(inner), val) => Self::from_json(val, inner),
158 _ => ColumnValue::Null,
159 }
160 }
161}
162
163impl std::fmt::Display for ColumnValue {
164 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
165 match self {
166 ColumnValue::Null => write!(f, "NULL"),
167 ColumnValue::Int(i) => write!(f, "{i}"),
168 ColumnValue::Float(v) => write!(f, "{v}"),
169 ColumnValue::Utf8(s) => write!(f, "{s}"),
170 ColumnValue::Boolean(b) => write!(f, "{b}"),
171 }
172 }
173}
174
175#[derive(Debug, Clone, Serialize, Deserialize)]
179pub struct ColumnSchema {
180 pub name: String,
182 pub col_type: ColumnType,
184}
185
186#[derive(Debug, Clone, Default, Serialize, Deserialize)]
193pub struct InMemoryTable {
194 pub columns: Vec<ColumnSchema>,
196 pub rows: Vec<Vec<ColumnValue>>,
198 pub name: Option<String>,
200}
201
202impl InMemoryTable {
203 pub fn new(columns: Vec<(String, ColumnType)>) -> Self {
205 Self {
206 columns: columns
207 .into_iter()
208 .map(|(n, t)| ColumnSchema {
209 name: n,
210 col_type: t,
211 })
212 .collect(),
213 rows: Vec::new(),
214 name: None,
215 }
216 }
217
218 pub fn with_name(mut self, name: impl Into<String>) -> Self {
220 self.name = Some(name.into());
221 self
222 }
223
224 pub fn row_count(&self) -> usize {
226 self.rows.len()
227 }
228
229 pub fn column_count(&self) -> usize {
231 self.columns.len()
232 }
233
234 pub fn column_index(&self, name: &str) -> Option<usize> {
236 self.columns.iter().position(|c| c.name == name)
237 }
238
239 pub fn push_row(&mut self, row: &[ColumnValue]) -> Result<()> {
241 if row.len() != self.columns.len() {
242 return Err(IoError::ValidationError(format!(
243 "Row has {} values but table has {} columns",
244 row.len(),
245 self.columns.len()
246 )));
247 }
248 self.rows.push(row.to_vec());
249 Ok(())
250 }
251
252 pub fn push_row_map(&mut self, map: HashMap<String, ColumnValue>) -> Result<()> {
254 let row: Vec<ColumnValue> = self
255 .columns
256 .iter()
257 .map(|col| map.get(&col.name).cloned().unwrap_or(ColumnValue::Null))
258 .collect();
259 self.rows.push(row);
260 Ok(())
261 }
262
263 pub fn get(&self, row: usize, col: usize) -> Option<&ColumnValue> {
265 self.rows.get(row)?.get(col)
266 }
267
268 pub fn get_column(&self, name: &str) -> Option<Vec<&ColumnValue>> {
270 let idx = self.column_index(name)?;
271 Some(self.rows.iter().map(|r| &r[idx]).collect())
272 }
273
274 pub fn row_as_map(&self, row_idx: usize) -> Option<HashMap<String, ColumnValue>> {
276 let row = self.rows.get(row_idx)?;
277 Some(
278 self.columns
279 .iter()
280 .zip(row.iter())
281 .map(|(col, val)| (col.name.clone(), val.clone()))
282 .collect(),
283 )
284 }
285
286 pub fn to_json_rows(&self) -> Vec<serde_json::Value> {
288 self.rows
289 .iter()
290 .map(|row| {
291 let obj: serde_json::Map<String, serde_json::Value> = self
292 .columns
293 .iter()
294 .zip(row.iter())
295 .map(|(col, val)| (col.name.clone(), val.to_json()))
296 .collect();
297 serde_json::Value::Object(obj)
298 })
299 .collect()
300 }
301
302 pub fn append(&mut self, other: &InMemoryTable) -> Result<()> {
304 if self.columns.len() != other.columns.len() {
305 return Err(IoError::ValidationError(
306 "Column count mismatch in append".to_string(),
307 ));
308 }
309 for (a, b) in self.columns.iter().zip(other.columns.iter()) {
310 if a.name != b.name {
311 return Err(IoError::ValidationError(format!(
312 "Column name mismatch: '{}' vs '{}'",
313 a.name, b.name
314 )));
315 }
316 }
317 self.rows.extend(other.rows.clone());
318 Ok(())
319 }
320}
321
322#[derive(Debug, Clone)]
326pub enum Predicate {
327 Eq(String, ColumnValue),
329 Ne(String, ColumnValue),
331 Greater(String, ColumnValue),
333 GreaterEq(String, ColumnValue),
335 Less(String, ColumnValue),
337 LessEq(String, ColumnValue),
339 IsNull(String),
341 IsNotNull(String),
343 Like(String, String),
345 In(String, Vec<ColumnValue>),
347 And(Box<Predicate>, Box<Predicate>),
349 Or(Box<Predicate>, Box<Predicate>),
351 Not(Box<Predicate>),
353}
354
355impl Predicate {
356 fn eval(&self, row: &[ColumnValue], columns: &[ColumnSchema]) -> bool {
357 match self {
358 Predicate::Eq(col, val) => get_col_val(row, columns, col)
359 .map(|v| v == val)
360 .unwrap_or(false),
361 Predicate::Ne(col, val) => get_col_val(row, columns, col)
362 .map(|v| v != val)
363 .unwrap_or(false),
364 Predicate::Greater(col, val) => get_col_val(row, columns, col)
365 .and_then(|v| v.partial_cmp_value(val))
366 .map(|o| o == Ordering::Greater)
367 .unwrap_or(false),
368 Predicate::GreaterEq(col, val) => get_col_val(row, columns, col)
369 .and_then(|v| v.partial_cmp_value(val))
370 .map(|o| o != Ordering::Less)
371 .unwrap_or(false),
372 Predicate::Less(col, val) => get_col_val(row, columns, col)
373 .and_then(|v| v.partial_cmp_value(val))
374 .map(|o| o == Ordering::Less)
375 .unwrap_or(false),
376 Predicate::LessEq(col, val) => get_col_val(row, columns, col)
377 .and_then(|v| v.partial_cmp_value(val))
378 .map(|o| o != Ordering::Greater)
379 .unwrap_or(false),
380 Predicate::IsNull(col) => get_col_val(row, columns, col)
381 .map(|v| matches!(v, ColumnValue::Null))
382 .unwrap_or(true),
383 Predicate::IsNotNull(col) => get_col_val(row, columns, col)
384 .map(|v| !matches!(v, ColumnValue::Null))
385 .unwrap_or(false),
386 Predicate::Like(col, pattern) => get_col_val(row, columns, col)
387 .and_then(|v| {
388 if let ColumnValue::Utf8(s) = v {
389 Some(like_match(s, pattern))
390 } else {
391 None
392 }
393 })
394 .unwrap_or(false),
395 Predicate::In(col, values) => get_col_val(row, columns, col)
396 .map(|v| values.contains(v))
397 .unwrap_or(false),
398 Predicate::And(a, b) => a.eval(row, columns) && b.eval(row, columns),
399 Predicate::Or(a, b) => a.eval(row, columns) || b.eval(row, columns),
400 Predicate::Not(p) => !p.eval(row, columns),
401 }
402 }
403}
404
405fn get_col_val<'a>(
406 row: &'a [ColumnValue],
407 columns: &[ColumnSchema],
408 name: &str,
409) -> Option<&'a ColumnValue> {
410 let idx = columns.iter().position(|c| c.name == name)?;
411 row.get(idx)
412}
413
414fn like_match(s: &str, pattern: &str) -> bool {
416 like_match_recursive(s.as_bytes(), pattern.as_bytes())
417}
418
419fn like_match_recursive(s: &[u8], p: &[u8]) -> bool {
420 match (s, p) {
421 (_, []) => s.is_empty(),
422 (_, [b'%', rest @ ..]) => {
423 for i in 0..=s.len() {
425 if like_match_recursive(&s[i..], rest) {
426 return true;
427 }
428 }
429 false
430 }
431 ([], _) => false,
432 ([sc, s_rest @ ..], [b'_', p_rest @ ..]) => {
433 like_match_recursive(s_rest, p_rest)
434 || (sc.is_ascii() && like_match_recursive(s_rest, p_rest))
435 }
436 ([sc, s_rest @ ..], [pc, p_rest @ ..]) => {
437 sc.eq_ignore_ascii_case(pc) && like_match_recursive(s_rest, p_rest)
438 }
439 }
440}
441
442pub struct TableFilter<'a> {
444 table: &'a InMemoryTable,
445 predicates: Vec<Predicate>,
446}
447
448impl<'a> TableFilter<'a> {
449 pub fn new(table: &'a InMemoryTable) -> Self {
451 Self {
452 table,
453 predicates: Vec::new(),
454 }
455 }
456
457 pub fn predicate(mut self, p: Predicate) -> Self {
459 self.predicates.push(p);
460 self
461 }
462
463 pub fn apply(&self) -> Result<InMemoryTable> {
465 let mut result = InMemoryTable {
466 columns: self.table.columns.clone(),
467 rows: Vec::new(),
468 name: self.table.name.clone(),
469 };
470 for row in &self.table.rows {
471 let matches = self
472 .predicates
473 .iter()
474 .all(|p| p.eval(row, &self.table.columns));
475 if matches {
476 result.rows.push(row.clone());
477 }
478 }
479 Ok(result)
480 }
481}
482
483#[derive(Debug, Clone, Copy, PartialEq, Eq)]
487pub enum SortDirection {
488 Asc,
490 Desc,
492}
493
494#[derive(Debug, Clone)]
496pub struct SortKey {
497 pub column: String,
499 pub direction: SortDirection,
501 pub nulls_first: bool,
503}
504
505impl SortKey {
506 pub fn asc(column: impl Into<String>) -> Self {
508 Self {
509 column: column.into(),
510 direction: SortDirection::Asc,
511 nulls_first: false,
512 }
513 }
514
515 pub fn desc(column: impl Into<String>) -> Self {
517 Self {
518 column: column.into(),
519 direction: SortDirection::Desc,
520 nulls_first: false,
521 }
522 }
523}
524
525pub struct TableSort;
527
528impl TableSort {
529 pub fn sort(table: &InMemoryTable, keys: &[SortKey]) -> Result<InMemoryTable> {
531 for key in keys {
533 if table.column_index(&key.column).is_none() {
534 return Err(IoError::ValidationError(format!(
535 "Sort column '{}' not found in table",
536 key.column
537 )));
538 }
539 }
540
541 let mut rows = table.rows.clone();
542 rows.sort_by(|a, b| {
543 for key in keys {
544 let idx = table
545 .columns
546 .iter()
547 .position(|c| c.name == key.column)
548 .unwrap_or(0);
549 let va = &a[idx];
550 let vb = &b[idx];
551
552 let ord = match (va, vb) {
553 (ColumnValue::Null, ColumnValue::Null) => Ordering::Equal,
554 (ColumnValue::Null, _) => {
555 if key.nulls_first {
556 Ordering::Less
557 } else {
558 Ordering::Greater
559 }
560 }
561 (_, ColumnValue::Null) => {
562 if key.nulls_first {
563 Ordering::Greater
564 } else {
565 Ordering::Less
566 }
567 }
568 _ => va.partial_cmp_value(vb).unwrap_or(Ordering::Equal),
569 };
570
571 let ord = if key.direction == SortDirection::Desc {
572 ord.reverse()
573 } else {
574 ord
575 };
576
577 if ord != Ordering::Equal {
578 return ord;
579 }
580 }
581 Ordering::Equal
582 });
583
584 Ok(InMemoryTable {
585 columns: table.columns.clone(),
586 rows,
587 name: table.name.clone(),
588 })
589 }
590}
591
592pub struct TableProjection<'a> {
596 table: &'a InMemoryTable,
597 selections: Vec<(String, Option<String>)>, }
599
600impl<'a> TableProjection<'a> {
601 pub fn new(table: &'a InMemoryTable) -> Self {
603 Self {
604 table,
605 selections: Vec::new(),
606 }
607 }
608
609 pub fn column(mut self, name: impl Into<String>) -> Self {
611 self.selections.push((name.into(), None));
612 self
613 }
614
615 pub fn column_as(mut self, name: impl Into<String>, alias: impl Into<String>) -> Self {
617 self.selections.push((name.into(), Some(alias.into())));
618 self
619 }
620
621 pub fn apply(&self) -> Result<InMemoryTable> {
623 let mut indices: Vec<(usize, String)> = Vec::new();
625 for (orig, alias) in &self.selections {
626 let idx = self.table.column_index(orig).ok_or_else(|| {
627 IoError::ValidationError(format!("Projection column '{}' not found in table", orig))
628 })?;
629 let out_name = alias.as_deref().unwrap_or(orig.as_str()).to_string();
630 indices.push((idx, out_name));
631 }
632
633 let new_columns: Vec<ColumnSchema> = indices
634 .iter()
635 .map(|(idx, name)| ColumnSchema {
636 name: name.clone(),
637 col_type: self.table.columns[*idx].col_type.clone(),
638 })
639 .collect();
640
641 let new_rows: Vec<Vec<ColumnValue>> = self
642 .table
643 .rows
644 .iter()
645 .map(|row| indices.iter().map(|(idx, _)| row[*idx].clone()).collect())
646 .collect();
647
648 Ok(InMemoryTable {
649 columns: new_columns,
650 rows: new_rows,
651 name: self.table.name.clone(),
652 })
653 }
654}
655
656#[derive(Debug, Clone)]
660pub enum AggFunc {
661 Count,
663 Sum(String),
665 Mean(String),
667 Min(String),
669 Max(String),
671 Std(String),
673 CountDistinct(String),
675}
676
677impl AggFunc {
678 pub fn output_name(&self) -> String {
680 match self {
681 AggFunc::Count => "count".to_string(),
682 AggFunc::Sum(col) => format!("sum_{col}"),
683 AggFunc::Mean(col) => format!("mean_{col}"),
684 AggFunc::Min(col) => format!("min_{col}"),
685 AggFunc::Max(col) => format!("max_{col}"),
686 AggFunc::Std(col) => format!("std_{col}"),
687 AggFunc::CountDistinct(col) => format!("count_distinct_{col}"),
688 }
689 }
690
691 fn compute(&self, rows: &[&Vec<ColumnValue>], columns: &[ColumnSchema]) -> ColumnValue {
692 match self {
693 AggFunc::Count => ColumnValue::Int(rows.len() as i64),
694 AggFunc::Sum(col) => {
695 let sum: f64 = rows
696 .iter()
697 .filter_map(|r| get_col_val(r, columns, col)?.as_f64())
698 .sum();
699 ColumnValue::Float(sum)
700 }
701 AggFunc::Mean(col) => {
702 let vals: Vec<f64> = rows
703 .iter()
704 .filter_map(|r| get_col_val(r, columns, col)?.as_f64())
705 .collect();
706 if vals.is_empty() {
707 ColumnValue::Null
708 } else {
709 ColumnValue::Float(vals.iter().sum::<f64>() / vals.len() as f64)
710 }
711 }
712 AggFunc::Min(col) => rows
713 .iter()
714 .filter_map(|r| get_col_val(r, columns, col))
715 .filter(|v| !matches!(v, ColumnValue::Null))
716 .min_by(|a, b| a.partial_cmp_value(b).unwrap_or(Ordering::Equal))
717 .cloned()
718 .unwrap_or(ColumnValue::Null),
719 AggFunc::Max(col) => rows
720 .iter()
721 .filter_map(|r| get_col_val(r, columns, col))
722 .filter(|v| !matches!(v, ColumnValue::Null))
723 .max_by(|a, b| a.partial_cmp_value(b).unwrap_or(Ordering::Equal))
724 .cloned()
725 .unwrap_or(ColumnValue::Null),
726 AggFunc::Std(col) => {
727 let vals: Vec<f64> = rows
728 .iter()
729 .filter_map(|r| get_col_val(r, columns, col)?.as_f64())
730 .collect();
731 if vals.len() < 2 {
732 ColumnValue::Float(0.0)
733 } else {
734 let n = vals.len() as f64;
735 let mean = vals.iter().sum::<f64>() / n;
736 let variance = vals.iter().map(|v| (v - mean).powi(2)).sum::<f64>() / (n - 1.0);
737 ColumnValue::Float(variance.sqrt())
738 }
739 }
740 AggFunc::CountDistinct(col) => {
741 let mut seen: Vec<String> = Vec::new();
742 for r in rows {
743 if let Some(v) = get_col_val(r, columns, col) {
744 let s = v.to_string();
745 if !seen.contains(&s) {
746 seen.push(s);
747 }
748 }
749 }
750 ColumnValue::Int(seen.len() as i64)
751 }
752 }
753 }
754}
755
756pub struct GroupBy<'a> {
758 table: &'a InMemoryTable,
759 group_cols: Vec<String>,
760 agg_funcs: Vec<AggFunc>,
761}
762
763impl<'a> GroupBy<'a> {
764 pub fn new(table: &'a InMemoryTable, group_cols: Vec<String>) -> Self {
766 Self {
767 table,
768 group_cols,
769 agg_funcs: Vec::new(),
770 }
771 }
772
773 pub fn agg(mut self, func: AggFunc) -> Self {
775 self.agg_funcs.push(func);
776 self
777 }
778
779 pub fn apply(&self) -> Result<InMemoryTable> {
781 let group_indices: Vec<usize> = self
783 .group_cols
784 .iter()
785 .map(|col| {
786 self.table.column_index(col).ok_or_else(|| {
787 IoError::ValidationError(format!("Group-by column '{col}' not found"))
788 })
789 })
790 .collect::<Result<Vec<_>>>()?;
791
792 let mut groups: HashMap<Vec<String>, Vec<&Vec<ColumnValue>>> = HashMap::new();
794 for row in &self.table.rows {
795 let key: Vec<String> = group_indices.iter().map(|&i| row[i].to_string()).collect();
796 groups.entry(key).or_default().push(row);
797 }
798
799 let mut out_columns: Vec<ColumnSchema> = self
801 .group_cols
802 .iter()
803 .map(|name| {
804 let idx = self.table.column_index(name).unwrap_or(0);
805 ColumnSchema {
806 name: name.clone(),
807 col_type: self.table.columns[idx].col_type.clone(),
808 }
809 })
810 .collect();
811 for agg in &self.agg_funcs {
812 out_columns.push(ColumnSchema {
813 name: agg.output_name(),
814 col_type: ColumnType::Float64,
815 });
816 }
817
818 let mut out_rows: Vec<Vec<ColumnValue>> = Vec::new();
820 let mut keys: Vec<Vec<String>> = groups.keys().cloned().collect();
822 keys.sort();
823
824 for key in &keys {
825 let group_rows = &groups[key];
826 let first_row = group_rows[0];
828 let mut out_row: Vec<ColumnValue> = group_indices
829 .iter()
830 .map(|&i| first_row[i].clone())
831 .collect();
832 for agg in &self.agg_funcs {
833 out_row.push(agg.compute(group_rows, &self.table.columns));
834 }
835 out_rows.push(out_row);
836 }
837
838 Ok(InMemoryTable {
839 columns: out_columns,
840 rows: out_rows,
841 name: None,
842 })
843 }
844}
845
846#[derive(Debug, Clone, Copy, PartialEq, Eq)]
850pub enum JoinType {
851 Inner,
853 Left,
855 Right,
857 Cross,
859}
860
861pub struct TableJoin;
863
864impl TableJoin {
865 pub fn hash_join(
867 left: &InMemoryTable,
868 right: &InMemoryTable,
869 left_key: &str,
870 right_key: &str,
871 join_type: JoinType,
872 ) -> Result<InMemoryTable> {
873 let left_key_idx = left.column_index(left_key).ok_or_else(|| {
874 IoError::ValidationError(format!("Left join key '{left_key}' not found"))
875 })?;
876 let right_key_idx = right.column_index(right_key).ok_or_else(|| {
877 IoError::ValidationError(format!("Right join key '{right_key}' not found"))
878 })?;
879
880 let mut out_columns: Vec<ColumnSchema> = left.columns.clone();
882 for (i, col) in right.columns.iter().enumerate() {
883 if i != right_key_idx {
884 let name = if left.column_index(&col.name).is_some() {
886 format!("right_{}", col.name)
887 } else {
888 col.name.clone()
889 };
890 out_columns.push(ColumnSchema {
891 name,
892 col_type: col.col_type.clone(),
893 });
894 }
895 }
896
897 let null_left: Vec<ColumnValue> = left.columns.iter().map(|_| ColumnValue::Null).collect();
898 let null_right_no_key: Vec<ColumnValue> = right
899 .columns
900 .iter()
901 .enumerate()
902 .filter(|(i, _)| *i != right_key_idx)
903 .map(|_| ColumnValue::Null)
904 .collect();
905
906 match join_type {
907 JoinType::Cross => {
908 let mut rows = Vec::new();
909 for l_row in &left.rows {
910 for r_row in &right.rows {
911 let mut out_row = l_row.clone();
912 for (i, v) in r_row.iter().enumerate() {
913 if i != right_key_idx {
914 out_row.push(v.clone());
915 }
916 }
917 rows.push(out_row);
918 }
919 }
920 return Ok(InMemoryTable {
921 columns: out_columns,
922 rows,
923 name: None,
924 });
925 }
926 JoinType::Inner | JoinType::Left | JoinType::Right => {}
927 }
928
929 let mut right_map: HashMap<String, Vec<usize>> = HashMap::new();
931 for (i, r_row) in right.rows.iter().enumerate() {
932 let key = r_row[right_key_idx].to_string();
933 right_map.entry(key).or_default().push(i);
934 }
935
936 let mut rows: Vec<Vec<ColumnValue>> = Vec::new();
937 let mut right_matched: Vec<bool> = vec![false; right.rows.len()];
938
939 for l_row in &left.rows {
940 let key = l_row[left_key_idx].to_string();
941 match right_map.get(&key) {
942 Some(r_indices) => {
943 for &ri in r_indices {
944 right_matched[ri] = true;
945 let mut out_row = l_row.clone();
946 for (i, v) in right.rows[ri].iter().enumerate() {
947 if i != right_key_idx {
948 out_row.push(v.clone());
949 }
950 }
951 rows.push(out_row);
952 }
953 }
954 None => {
955 if join_type == JoinType::Left {
956 let mut out_row = l_row.clone();
957 out_row.extend(null_right_no_key.iter().cloned());
958 rows.push(out_row);
959 }
960 }
961 }
962 }
963
964 if join_type == JoinType::Right {
966 for (i, r_row) in right.rows.iter().enumerate() {
967 if !right_matched[i] {
968 let mut out_row = null_left.clone();
969 for (j, v) in r_row.iter().enumerate() {
970 if j != right_key_idx {
971 out_row.push(v.clone());
972 }
973 }
974 rows.push(out_row);
975 }
976 }
977 }
978
979 Ok(InMemoryTable {
980 columns: out_columns,
981 rows,
982 name: None,
983 })
984 }
985
986 pub fn merge_join(
989 left: &InMemoryTable,
990 right: &InMemoryTable,
991 left_key: &str,
992 right_key: &str,
993 ) -> Result<InMemoryTable> {
994 let left_key_idx = left.column_index(left_key).ok_or_else(|| {
995 IoError::ValidationError(format!("Left merge key '{left_key}' not found"))
996 })?;
997 let right_key_idx = right.column_index(right_key).ok_or_else(|| {
998 IoError::ValidationError(format!("Right merge key '{right_key}' not found"))
999 })?;
1000
1001 let mut out_columns: Vec<ColumnSchema> = left.columns.clone();
1002 for (i, col) in right.columns.iter().enumerate() {
1003 if i != right_key_idx {
1004 let name = if left.column_index(&col.name).is_some() {
1005 format!("right_{}", col.name)
1006 } else {
1007 col.name.clone()
1008 };
1009 out_columns.push(ColumnSchema {
1010 name,
1011 col_type: col.col_type.clone(),
1012 });
1013 }
1014 }
1015
1016 let mut rows = Vec::new();
1017 let mut li = 0usize;
1018 let mut ri = 0usize;
1019
1020 while li < left.rows.len() && ri < right.rows.len() {
1021 let lk = &left.rows[li][left_key_idx];
1022 let rk = &right.rows[ri][right_key_idx];
1023
1024 match lk.partial_cmp_value(rk).unwrap_or(Ordering::Equal) {
1025 Ordering::Equal => {
1026 let mut rj = ri;
1028 while rj < right.rows.len() {
1029 let rk2 = &right.rows[rj][right_key_idx];
1030 if lk.partial_cmp_value(rk2) != Some(Ordering::Equal) {
1031 break;
1032 }
1033 let mut out_row = left.rows[li].clone();
1034 for (k, v) in right.rows[rj].iter().enumerate() {
1035 if k != right_key_idx {
1036 out_row.push(v.clone());
1037 }
1038 }
1039 rows.push(out_row);
1040 rj += 1;
1041 }
1042 li += 1;
1043 }
1044 Ordering::Less => li += 1,
1045 Ordering::Greater => ri += 1,
1046 }
1047 }
1048
1049 Ok(InMemoryTable {
1050 columns: out_columns,
1051 rows,
1052 name: None,
1053 })
1054 }
1055}
1056
1057#[cfg(test)]
1060mod tests {
1061 use super::*;
1062
1063 fn make_table() -> InMemoryTable {
1064 let mut t = InMemoryTable::new(vec![
1065 ("id".to_string(), ColumnType::Int64),
1066 ("name".to_string(), ColumnType::Utf8),
1067 ("score".to_string(), ColumnType::Float64),
1068 ("dept".to_string(), ColumnType::Utf8),
1069 ]);
1070 t.push_row(&[
1071 ColumnValue::Int(1),
1072 ColumnValue::Utf8("Alice".to_string()),
1073 ColumnValue::Float(95.0),
1074 ColumnValue::Utf8("eng".to_string()),
1075 ])
1076 .unwrap();
1077 t.push_row(&[
1078 ColumnValue::Int(2),
1079 ColumnValue::Utf8("Bob".to_string()),
1080 ColumnValue::Float(82.5),
1081 ColumnValue::Utf8("eng".to_string()),
1082 ])
1083 .unwrap();
1084 t.push_row(&[
1085 ColumnValue::Int(3),
1086 ColumnValue::Utf8("Carol".to_string()),
1087 ColumnValue::Float(91.0),
1088 ColumnValue::Utf8("hr".to_string()),
1089 ])
1090 .unwrap();
1091 t.push_row(&[
1092 ColumnValue::Int(4),
1093 ColumnValue::Utf8("Dave".to_string()),
1094 ColumnValue::Float(78.0),
1095 ColumnValue::Utf8("hr".to_string()),
1096 ])
1097 .unwrap();
1098 t
1099 }
1100
1101 #[test]
1102 fn test_filter_greater() {
1103 let t = make_table();
1104 let filtered = TableFilter::new(&t)
1105 .predicate(Predicate::Greater(
1106 "score".to_string(),
1107 ColumnValue::Float(90.0),
1108 ))
1109 .apply()
1110 .unwrap();
1111 assert_eq!(filtered.row_count(), 2); }
1113
1114 #[test]
1115 fn test_filter_eq() {
1116 let t = make_table();
1117 let filtered = TableFilter::new(&t)
1118 .predicate(Predicate::Eq(
1119 "dept".to_string(),
1120 ColumnValue::Utf8("eng".to_string()),
1121 ))
1122 .apply()
1123 .unwrap();
1124 assert_eq!(filtered.row_count(), 2);
1125 }
1126
1127 #[test]
1128 fn test_sort_asc() {
1129 let t = make_table();
1130 let sorted = TableSort::sort(&t, &[SortKey::asc("score")]).unwrap();
1131 let scores: Vec<f64> = sorted
1132 .get_column("score")
1133 .unwrap()
1134 .into_iter()
1135 .filter_map(|v| v.as_f64())
1136 .collect();
1137 assert_eq!(scores, vec![78.0, 82.5, 91.0, 95.0]);
1138 }
1139
1140 #[test]
1141 fn test_sort_desc() {
1142 let t = make_table();
1143 let sorted = TableSort::sort(&t, &[SortKey::desc("score")]).unwrap();
1144 let scores: Vec<f64> = sorted
1145 .get_column("score")
1146 .unwrap()
1147 .into_iter()
1148 .filter_map(|v| v.as_f64())
1149 .collect();
1150 assert_eq!(scores, vec![95.0, 91.0, 82.5, 78.0]);
1151 }
1152
1153 #[test]
1154 fn test_projection() {
1155 let t = make_table();
1156 let projected = TableProjection::new(&t)
1157 .column("id")
1158 .column_as("name", "full_name")
1159 .apply()
1160 .unwrap();
1161 assert_eq!(projected.column_count(), 2);
1162 assert_eq!(projected.columns[1].name, "full_name");
1163 assert_eq!(projected.row_count(), 4);
1164 }
1165
1166 #[test]
1167 fn test_group_by_sum_mean() {
1168 let t = make_table();
1169 let grouped = GroupBy::new(&t, vec!["dept".to_string()])
1170 .agg(AggFunc::Count)
1171 .agg(AggFunc::Sum("score".to_string()))
1172 .agg(AggFunc::Mean("score".to_string()))
1173 .apply()
1174 .unwrap();
1175
1176 assert_eq!(grouped.row_count(), 2); let eng_row = grouped
1179 .rows
1180 .iter()
1181 .find(|r| r[0] == ColumnValue::Utf8("eng".to_string()))
1182 .expect("eng group missing");
1183 assert_eq!(eng_row[1], ColumnValue::Int(2)); if let ColumnValue::Float(sum) = eng_row[2] {
1185 assert!((sum - 177.5).abs() < 1e-9);
1186 } else {
1187 panic!("Expected float sum");
1188 }
1189 }
1190
1191 #[test]
1192 fn test_inner_join() {
1193 let mut left = InMemoryTable::new(vec![
1194 ("id".to_string(), ColumnType::Int64),
1195 ("val".to_string(), ColumnType::Float64),
1196 ]);
1197 left.push_row(&[ColumnValue::Int(1), ColumnValue::Float(1.0)])
1198 .unwrap();
1199 left.push_row(&[ColumnValue::Int(2), ColumnValue::Float(2.0)])
1200 .unwrap();
1201 left.push_row(&[ColumnValue::Int(3), ColumnValue::Float(3.0)])
1202 .unwrap();
1203
1204 let mut right = InMemoryTable::new(vec![
1205 ("id".to_string(), ColumnType::Int64),
1206 ("label".to_string(), ColumnType::Utf8),
1207 ]);
1208 right
1209 .push_row(&[ColumnValue::Int(1), ColumnValue::Utf8("one".to_string())])
1210 .unwrap();
1211 right
1212 .push_row(&[ColumnValue::Int(2), ColumnValue::Utf8("two".to_string())])
1213 .unwrap();
1214
1215 let joined = TableJoin::hash_join(&left, &right, "id", "id", JoinType::Inner).unwrap();
1216 assert_eq!(joined.row_count(), 2);
1217 }
1218
1219 #[test]
1220 fn test_left_join() {
1221 let mut left = InMemoryTable::new(vec![("id".to_string(), ColumnType::Int64)]);
1222 left.push_row(&[ColumnValue::Int(1)]).unwrap();
1223 left.push_row(&[ColumnValue::Int(2)]).unwrap();
1224 left.push_row(&[ColumnValue::Int(3)]).unwrap(); let mut right = InMemoryTable::new(vec![
1227 ("id".to_string(), ColumnType::Int64),
1228 ("x".to_string(), ColumnType::Float64),
1229 ]);
1230 right
1231 .push_row(&[ColumnValue::Int(1), ColumnValue::Float(10.0)])
1232 .unwrap();
1233 right
1234 .push_row(&[ColumnValue::Int(2), ColumnValue::Float(20.0)])
1235 .unwrap();
1236
1237 let joined = TableJoin::hash_join(&left, &right, "id", "id", JoinType::Left).unwrap();
1238 assert_eq!(joined.row_count(), 3);
1239 let row3 = joined
1241 .rows
1242 .iter()
1243 .find(|r| r[0] == ColumnValue::Int(3))
1244 .expect("row 3 missing");
1245 assert_eq!(row3[1], ColumnValue::Null);
1246 }
1247
1248 #[test]
1249 fn test_cross_join() {
1250 let mut a = InMemoryTable::new(vec![("a".to_string(), ColumnType::Int64)]);
1251 a.push_row(&[ColumnValue::Int(1)]).unwrap();
1252 a.push_row(&[ColumnValue::Int(2)]).unwrap();
1253
1254 let mut b = InMemoryTable::new(vec![("b".to_string(), ColumnType::Int64)]);
1255 b.push_row(&[ColumnValue::Int(10)]).unwrap();
1256 b.push_row(&[ColumnValue::Int(20)]).unwrap();
1257 b.push_row(&[ColumnValue::Int(30)]).unwrap();
1258
1259 let crossed = TableJoin::hash_join(&a, &b, "a", "b", JoinType::Cross).unwrap();
1260 assert_eq!(crossed.row_count(), 6); }
1262
1263 #[test]
1264 fn test_like_match() {
1265 assert!(like_match("hello world", "%world"));
1266 assert!(like_match("hello world", "hello%"));
1267 assert!(like_match("hello world", "%lo w%"));
1268 assert!(!like_match("hello world", "xyz%"));
1269 assert!(like_match("abc", "a_c"));
1270 assert!(!like_match("axyz", "a_c"));
1271 }
1272}