1use std::collections::{HashMap, HashSet};
29
30use super::super::ast::{
31 CteDefinition, Expr, Filter, Projection, QueryExpr, QueryWithCte, SelectItem, TableSource,
32};
33use super::super::unified::{ExecutionError, UnifiedRecord, UnifiedResult};
34use crate::storage::schema::Value;
35
36const MAX_RECURSION_DEPTH: usize = 1000;
38
39const MAX_RECURSIVE_ROWS: usize = 100_000;
41
42#[derive(Debug, Clone, Default)]
44pub struct CteContext {
45 tables: HashMap<String, UnifiedResult>,
47 evaluating: HashSet<String>,
49 stats: CteStats,
51}
52
53impl CteContext {
54 pub fn new() -> Self {
56 Self::default()
57 }
58
59 pub fn get(&self, name: &str) -> Option<&UnifiedResult> {
61 self.tables.get(name)
62 }
63
64 pub fn store(&mut self, name: String, result: UnifiedResult) {
66 self.tables.insert(name, result);
67 }
68
69 pub fn is_evaluating(&self, name: &str) -> bool {
71 self.evaluating.contains(name)
72 }
73
74 pub fn start_evaluating(&mut self, name: &str) {
76 self.evaluating.insert(name.to_string());
77 }
78
79 pub fn done_evaluating(&mut self, name: &str) {
81 self.evaluating.remove(name);
82 }
83
84 pub fn stats(&self) -> &CteStats {
86 &self.stats
87 }
88}
89
90#[derive(Debug, Clone, Default)]
92pub struct CteStats {
93 pub ctes_executed: usize,
95 pub recursive_iterations: usize,
97 pub rows_produced: usize,
99 pub exec_time_us: u64,
101}
102
103pub struct CteExecutor<F>
105where
106 F: Fn(&QueryExpr, &CteContext) -> Result<UnifiedResult, ExecutionError>,
107{
108 execute_fn: F,
110}
111
112impl<F> CteExecutor<F>
113where
114 F: Fn(&QueryExpr, &CteContext) -> Result<UnifiedResult, ExecutionError>,
115{
116 pub fn new(execute_fn: F) -> Self {
118 Self { execute_fn }
119 }
120
121 pub fn execute(&self, query: &QueryWithCte) -> Result<UnifiedResult, ExecutionError> {
123 let start = std::time::Instant::now();
124 let mut ctx = CteContext::new();
125
126 if let Some(ref with_clause) = query.with_clause {
128 for cte in &with_clause.ctes {
129 self.materialize_cte(cte, &mut ctx)?;
130 }
131 }
132
133 let result = (self.execute_fn)(&query.query, &ctx)?;
135
136 ctx.stats.exec_time_us = start.elapsed().as_micros() as u64;
137 Ok(result)
138 }
139
140 fn materialize_cte(
142 &self,
143 cte: &CteDefinition,
144 ctx: &mut CteContext,
145 ) -> Result<(), ExecutionError> {
146 if ctx.is_evaluating(&cte.name) {
147 return Err(ExecutionError::new(format!(
148 "Circular CTE reference: {}",
149 cte.name
150 )));
151 }
152
153 if ctx.get(&cte.name).is_some() {
155 return Ok(());
156 }
157
158 ctx.start_evaluating(&cte.name);
159
160 let result = if cte.recursive {
161 self.execute_recursive_cte(cte, ctx)?
162 } else {
163 let result = (self.execute_fn)(&cte.query, ctx)?;
165 self.project_columns(&result, &cte.columns)
166 };
167
168 ctx.stats.ctes_executed += 1;
169 ctx.stats.rows_produced += result.len();
170 ctx.store(cte.name.clone(), result);
171 ctx.done_evaluating(&cte.name);
172
173 Ok(())
174 }
175
176 fn execute_recursive_cte(
178 &self,
179 cte: &CteDefinition,
180 ctx: &mut CteContext,
181 ) -> Result<UnifiedResult, ExecutionError> {
182 let mut all_results = UnifiedResult::with_columns(cte.columns.clone());
199 let mut working_table = UnifiedResult::with_columns(cte.columns.clone());
200 let mut seen_rows: HashSet<u64> = HashSet::new();
201 let mut iteration = 0;
202
203 let initial = (self.execute_fn)(&cte.query, ctx)?;
205 let initial = self.project_columns(&initial, &cte.columns);
206
207 for record in &initial.records {
208 let hash = self.hash_record(record);
209 if seen_rows.insert(hash) {
210 working_table.push(record.clone());
211 all_results.push(record.clone());
212 }
213 }
214
215 ctx.store(cte.name.clone(), working_table.clone());
217
218 while !working_table.is_empty() && iteration < MAX_RECURSION_DEPTH {
220 iteration += 1;
221 ctx.stats.recursive_iterations += 1;
222
223 if all_results.len() > MAX_RECURSIVE_ROWS {
224 return Err(ExecutionError::new(format!(
225 "Recursive CTE '{}' exceeded maximum rows ({})",
226 cte.name, MAX_RECURSIVE_ROWS
227 )));
228 }
229
230 let new_results = (self.execute_fn)(&cte.query, ctx)?;
232 let new_results = self.project_columns(&new_results, &cte.columns);
233
234 let mut new_working_table = UnifiedResult::with_columns(cte.columns.clone());
236 for record in &new_results.records {
237 let hash = self.hash_record(record);
238 if seen_rows.insert(hash) {
239 new_working_table.push(record.clone());
240 all_results.push(record.clone());
241 }
242 }
243
244 working_table = new_working_table;
245
246 ctx.store(cte.name.clone(), all_results.clone());
248 }
249
250 if iteration >= MAX_RECURSION_DEPTH && !working_table.is_empty() {
251 return Err(ExecutionError::new(format!(
252 "Recursive CTE '{}' exceeded maximum recursion depth ({})",
253 cte.name, MAX_RECURSION_DEPTH
254 )));
255 }
256
257 Ok(all_results)
258 }
259
260 fn project_columns(&self, result: &UnifiedResult, columns: &[String]) -> UnifiedResult {
262 if columns.is_empty() {
263 return result.clone();
264 }
265
266 let mut projected = UnifiedResult::with_columns(columns.to_vec());
267
268 for record in &result.records {
269 let mut new_record = UnifiedRecord::new();
270
271 for (i, col) in columns.iter().enumerate() {
273 let value = result
275 .columns
276 .get(i)
277 .and_then(|orig_col| record.get(orig_col))
278 .cloned()
279 .or_else(|| record.get(col).cloned())
280 .unwrap_or(Value::Null);
281
282 new_record.set(col, value);
283 }
284
285 projected.push(new_record);
286 }
287
288 projected
289 }
290
291 fn hash_record(&self, record: &UnifiedRecord) -> u64 {
293 use std::collections::hash_map::DefaultHasher;
294 use std::hash::{Hash, Hasher};
295
296 let mut hasher = DefaultHasher::new();
297
298 let mut keys = record.column_names();
300 keys.sort();
301
302 for key in &keys {
303 (**key).hash(&mut hasher);
304 if let Some(value) = record.get(key) {
305 Self::hash_value(value, &mut hasher);
306 }
307 }
308
309 hasher.finish()
310 }
311
312 fn hash_value(value: &Value, hasher: &mut impl std::hash::Hasher) {
314 use std::hash::Hash;
315
316 match value {
317 Value::Null => 0u8.hash(hasher),
318 Value::Boolean(b) => {
319 1u8.hash(hasher);
320 b.hash(hasher);
321 }
322 Value::Integer(i) => {
323 2u8.hash(hasher);
324 i.hash(hasher);
325 }
326 Value::UnsignedInteger(u) => {
327 3u8.hash(hasher);
328 u.hash(hasher);
329 }
330 Value::Float(f) => {
331 4u8.hash(hasher);
332 f.to_bits().hash(hasher);
333 }
334 Value::Text(s) => {
335 5u8.hash(hasher);
336 s.hash(hasher);
337 }
338 Value::Blob(b) => {
339 6u8.hash(hasher);
340 b.hash(hasher);
341 }
342 Value::Timestamp(t) => {
343 7u8.hash(hasher);
344 t.hash(hasher);
345 }
346 Value::Duration(d) => {
347 8u8.hash(hasher);
348 d.hash(hasher);
349 }
350 Value::IpAddr(addr) => {
351 9u8.hash(hasher);
352 match addr {
353 std::net::IpAddr::V4(v4) => v4.octets().hash(hasher),
354 std::net::IpAddr::V6(v6) => v6.octets().hash(hasher),
355 }
356 }
357 Value::MacAddr(mac) => {
358 10u8.hash(hasher);
359 mac.hash(hasher);
360 }
361 Value::Vector(v) => {
362 11u8.hash(hasher);
363 v.len().hash(hasher);
364 for f in v {
365 f.to_bits().hash(hasher);
366 }
367 }
368 Value::Json(j) => {
369 12u8.hash(hasher);
370 j.hash(hasher);
371 }
372 Value::Uuid(u) => {
373 13u8.hash(hasher);
374 u.hash(hasher);
375 }
376 Value::NodeRef(n) => {
377 14u8.hash(hasher);
378 n.hash(hasher);
379 }
380 Value::EdgeRef(e) => {
381 15u8.hash(hasher);
382 e.hash(hasher);
383 }
384 Value::VectorRef(coll, id) => {
385 16u8.hash(hasher);
386 coll.hash(hasher);
387 id.hash(hasher);
388 }
389 Value::RowRef(table, id) => {
390 17u8.hash(hasher);
391 table.hash(hasher);
392 id.hash(hasher);
393 }
394 Value::Color(rgb) => {
395 18u8.hash(hasher);
396 rgb.hash(hasher);
397 }
398 Value::Email(s) => {
399 19u8.hash(hasher);
400 s.hash(hasher);
401 }
402 Value::Url(s) => {
403 20u8.hash(hasher);
404 s.hash(hasher);
405 }
406 Value::Phone(n) => {
407 21u8.hash(hasher);
408 n.hash(hasher);
409 }
410 Value::Semver(v) => {
411 22u8.hash(hasher);
412 v.hash(hasher);
413 }
414 Value::Cidr(ip, prefix) => {
415 23u8.hash(hasher);
416 ip.hash(hasher);
417 prefix.hash(hasher);
418 }
419 Value::Date(d) => {
420 24u8.hash(hasher);
421 d.hash(hasher);
422 }
423 Value::Time(t) => {
424 25u8.hash(hasher);
425 t.hash(hasher);
426 }
427 Value::Decimal(v) => {
428 26u8.hash(hasher);
429 v.hash(hasher);
430 }
431 Value::DecimalText(v) => {
432 52u8.hash(hasher);
433 v.hash(hasher);
434 }
435 Value::EnumValue(i) => {
436 27u8.hash(hasher);
437 i.hash(hasher);
438 }
439 Value::Array(elems) => {
440 28u8.hash(hasher);
441 elems.len().hash(hasher);
442 for elem in elems {
443 Self::hash_value(elem, hasher);
444 }
445 }
446 Value::TimestampMs(v) => {
447 29u8.hash(hasher);
448 v.hash(hasher);
449 }
450 Value::Ipv4(v) => {
451 30u8.hash(hasher);
452 v.hash(hasher);
453 }
454 Value::Ipv6(bytes) => {
455 31u8.hash(hasher);
456 bytes.hash(hasher);
457 }
458 Value::Subnet(ip, mask) => {
459 32u8.hash(hasher);
460 ip.hash(hasher);
461 mask.hash(hasher);
462 }
463 Value::Port(v) => {
464 33u8.hash(hasher);
465 v.hash(hasher);
466 }
467 Value::Latitude(v) => {
468 34u8.hash(hasher);
469 v.hash(hasher);
470 }
471 Value::Longitude(v) => {
472 35u8.hash(hasher);
473 v.hash(hasher);
474 }
475 Value::GeoPoint(lat, lon) => {
476 36u8.hash(hasher);
477 lat.hash(hasher);
478 lon.hash(hasher);
479 }
480 Value::Country2(c) => {
481 37u8.hash(hasher);
482 c.hash(hasher);
483 }
484 Value::Country3(c) => {
485 38u8.hash(hasher);
486 c.hash(hasher);
487 }
488 Value::Lang2(c) => {
489 39u8.hash(hasher);
490 c.hash(hasher);
491 }
492 Value::Lang5(c) => {
493 40u8.hash(hasher);
494 c.hash(hasher);
495 }
496 Value::Currency(c) => {
497 41u8.hash(hasher);
498 c.hash(hasher);
499 }
500 Value::AssetCode(code) => {
501 50u8.hash(hasher);
502 code.hash(hasher);
503 }
504 Value::Money {
505 asset_code,
506 minor_units,
507 scale,
508 } => {
509 51u8.hash(hasher);
510 asset_code.hash(hasher);
511 minor_units.hash(hasher);
512 scale.hash(hasher);
513 }
514 Value::ColorAlpha(rgba) => {
515 42u8.hash(hasher);
516 rgba.hash(hasher);
517 }
518 Value::BigInt(v) => {
519 43u8.hash(hasher);
520 v.hash(hasher);
521 }
522 Value::KeyRef(col, key) => {
523 44u8.hash(hasher);
524 col.hash(hasher);
525 key.hash(hasher);
526 }
527 Value::DocRef(col, id) => {
528 45u8.hash(hasher);
529 col.hash(hasher);
530 id.hash(hasher);
531 }
532 Value::TableRef(name) => {
533 46u8.hash(hasher);
534 name.hash(hasher);
535 }
536 Value::PageRef(page_id) => {
537 47u8.hash(hasher);
538 page_id.hash(hasher);
539 }
540 Value::Secret(bytes) => {
541 48u8.hash(hasher);
542 bytes.hash(hasher);
543 }
544 Value::Password(hash) => {
545 49u8.hash(hasher);
546 hash.hash(hasher);
547 }
548 }
549 }
550}
551
552pub fn split_union_parts(query: &QueryExpr) -> Option<(QueryExpr, QueryExpr)> {
554 let _ = query;
557 None
558}
559
560pub fn inline_ctes(query: QueryWithCte) -> Result<QueryExpr, ExecutionError> {
579 let Some(with_clause) = query.with_clause else {
580 return Ok(query.query);
581 };
582 if with_clause.has_recursive {
583 return Err(ExecutionError::new(
584 "WITH RECURSIVE is not yet supported by the executor; \
585 non-recursive WITH clauses run today, recursive support \
586 is tracked separately"
587 .to_string(),
588 ));
589 }
590
591 let mut resolved: HashMap<String, QueryExpr> = HashMap::new();
595 for cte in &with_clause.ctes {
596 let mut body = (*cte.query).clone();
597 rewrite(&mut body, &resolved);
598 resolved.insert(cte.name.clone(), body);
599 }
600
601 let mut outer = query.query;
602 rewrite(&mut outer, &resolved);
603 Ok(outer)
604}
605
606fn rewrite(expr: &mut QueryExpr, ctes: &HashMap<String, QueryExpr>) {
616 match expr {
617 QueryExpr::Table(tq) => {
618 let lookup_name = match &tq.source {
619 Some(TableSource::Subquery(_)) => None,
620 Some(TableSource::Name(n)) => Some(n.clone()),
621 Some(TableSource::Function { .. } | TableSource::InlineGraphFunction { .. }) => {
624 None
625 }
626 None => Some(tq.table.clone()),
627 };
628
629 if let Some(name) = lookup_name {
630 if let Some(body) = ctes.get(&name) {
631 let outer_has_constraints = tq.filter.is_some()
632 || tq.where_expr.is_some()
633 || tq.limit.is_some()
634 || tq.offset.is_some()
635 || !tq.columns.is_empty()
636 || !tq.select_items.is_empty()
637 || !tq.group_by.is_empty()
638 || !tq.order_by.is_empty();
639
640 if outer_has_constraints {
641 tq.source = Some(TableSource::Subquery(Box::new(body.clone())));
647 tq.table = format!("__cte_{name}");
648 } else {
649 *expr = body.clone();
653 }
654 return;
655 }
656 }
657
658 match tq.source.as_mut() {
659 Some(TableSource::Subquery(body)) => rewrite(body, ctes),
660 Some(TableSource::InlineGraphFunction { nodes, edges, .. }) => {
661 rewrite(nodes, ctes);
662 rewrite(edges, ctes);
663 }
664 _ => {}
665 }
666 rewrite_table_query_parts(tq, ctes);
667 }
668 QueryExpr::Join(jq) => {
669 rewrite(&mut jq.left, ctes);
670 rewrite(&mut jq.right, ctes);
671 if let Some(filter) = jq.filter.as_mut() {
672 rewrite_filter(filter, ctes);
673 }
674 for item in &mut jq.order_by {
675 if let Some(expr) = item.expr.as_mut() {
676 rewrite_expr(expr, ctes);
677 }
678 }
679 for item in &mut jq.return_items {
680 rewrite_select_item(item, ctes);
681 }
682 for projection in &mut jq.return_ {
683 rewrite_projection(projection, ctes);
684 }
685 }
686 QueryExpr::Graph(gq) => {
687 if let Some(filter) = gq.filter.as_mut() {
688 rewrite_filter(filter, ctes);
689 }
690 for projection in &mut gq.return_ {
691 rewrite_projection(projection, ctes);
692 }
693 }
694 _ => {}
695 }
696}
697
698fn rewrite_table_query_parts(
699 tq: &mut super::super::ast::TableQuery,
700 ctes: &HashMap<String, QueryExpr>,
701) {
702 for item in &mut tq.select_items {
703 rewrite_select_item(item, ctes);
704 }
705 for projection in &mut tq.columns {
706 rewrite_projection(projection, ctes);
707 }
708 if let Some(expr) = tq.where_expr.as_mut() {
709 rewrite_expr(expr, ctes);
710 }
711 if let Some(filter) = tq.filter.as_mut() {
712 rewrite_filter(filter, ctes);
713 }
714 for expr in &mut tq.group_by_exprs {
715 rewrite_expr(expr, ctes);
716 }
717 if let Some(expr) = tq.having_expr.as_mut() {
718 rewrite_expr(expr, ctes);
719 }
720 if let Some(filter) = tq.having.as_mut() {
721 rewrite_filter(filter, ctes);
722 }
723 for item in &mut tq.order_by {
724 if let Some(expr) = item.expr.as_mut() {
725 rewrite_expr(expr, ctes);
726 }
727 }
728}
729
730fn rewrite_select_item(item: &mut SelectItem, ctes: &HashMap<String, QueryExpr>) {
731 if let SelectItem::Expr { expr, .. } = item {
732 rewrite_expr(expr, ctes);
733 }
734}
735
736fn rewrite_projection(projection: &mut Projection, ctes: &HashMap<String, QueryExpr>) {
737 match projection {
738 Projection::Function(_, args) => {
739 for arg in args {
740 rewrite_projection(arg, ctes);
741 }
742 }
743 Projection::Expression(filter, _) => rewrite_filter(filter, ctes),
744 Projection::Window { args, window, .. } => {
745 for arg in args {
746 rewrite_projection(arg, ctes);
747 }
748 for expr in &mut window.partition_by {
749 rewrite_expr(expr, ctes);
750 }
751 for item in &mut window.order_by {
752 rewrite_expr(&mut item.expr, ctes);
753 }
754 }
755 Projection::All
756 | Projection::Column(_)
757 | Projection::Alias(_, _)
758 | Projection::Field(_, _) => {}
759 }
760}
761
762fn rewrite_filter(filter: &mut Filter, ctes: &HashMap<String, QueryExpr>) {
763 match filter {
764 Filter::CompareExpr { lhs, rhs, .. } => {
765 rewrite_expr(lhs, ctes);
766 rewrite_expr(rhs, ctes);
767 }
768 Filter::And(left, right) | Filter::Or(left, right) => {
769 rewrite_filter(left, ctes);
770 rewrite_filter(right, ctes);
771 }
772 Filter::Not(inner) => rewrite_filter(inner, ctes),
773 Filter::Compare { .. }
774 | Filter::CompareFields { .. }
775 | Filter::IsNull(_)
776 | Filter::IsNotNull(_)
777 | Filter::In { .. }
778 | Filter::Between { .. }
779 | Filter::Like { .. }
780 | Filter::StartsWith { .. }
781 | Filter::EndsWith { .. }
782 | Filter::Contains { .. } => {}
783 }
784}
785
786fn rewrite_expr(expr: &mut Expr, ctes: &HashMap<String, QueryExpr>) {
787 match expr {
788 Expr::BinaryOp { lhs, rhs, .. } => {
789 rewrite_expr(lhs, ctes);
790 rewrite_expr(rhs, ctes);
791 }
792 Expr::UnaryOp { operand, .. } => rewrite_expr(operand, ctes),
793 Expr::Cast { inner, .. } => rewrite_expr(inner, ctes),
794 Expr::FunctionCall { args, .. } => {
795 for arg in args {
796 rewrite_expr(arg, ctes);
797 }
798 }
799 Expr::Case {
800 branches, else_, ..
801 } => {
802 for (condition, value) in branches {
803 rewrite_expr(condition, ctes);
804 rewrite_expr(value, ctes);
805 }
806 if let Some(value) = else_ {
807 rewrite_expr(value, ctes);
808 }
809 }
810 Expr::IsNull { operand, .. } => rewrite_expr(operand, ctes),
811 Expr::InList { target, values, .. } => {
812 rewrite_expr(target, ctes);
813 for value in values {
814 rewrite_expr(value, ctes);
815 }
816 }
817 Expr::Between {
818 target, low, high, ..
819 } => {
820 rewrite_expr(target, ctes);
821 rewrite_expr(low, ctes);
822 rewrite_expr(high, ctes);
823 }
824 Expr::Subquery { query, .. } => rewrite(&mut query.query, ctes),
825 Expr::WindowFunctionCall { args, window, .. } => {
826 for arg in args {
827 rewrite_expr(arg, ctes);
828 }
829 for expr in &mut window.partition_by {
830 rewrite_expr(expr, ctes);
831 }
832 for item in &mut window.order_by {
833 rewrite_expr(&mut item.expr, ctes);
834 }
835 }
836 Expr::Literal { .. } | Expr::Column { .. } | Expr::Parameter { .. } => {}
837 }
838}
839
840#[cfg(test)]
845mod tests {
846 use super::*;
847 use crate::storage::query::ast::CteQueryBuilder;
848 use crate::storage::query::WithClause;
849
850 fn mock_execute(
851 _query: &QueryExpr,
852 _ctx: &CteContext,
853 ) -> Result<UnifiedResult, ExecutionError> {
854 Ok(UnifiedResult::empty())
856 }
857
858 #[test]
859 fn test_cte_context() {
860 let mut ctx = CteContext::new();
861
862 assert!(ctx.get("test").is_none());
864 assert!(!ctx.is_evaluating("test"));
865
866 let result = UnifiedResult::with_columns(vec!["col1".to_string()]);
868 ctx.store("test".to_string(), result);
869 assert!(ctx.get("test").is_some());
870
871 ctx.start_evaluating("other");
873 assert!(ctx.is_evaluating("other"));
874 ctx.done_evaluating("other");
875 assert!(!ctx.is_evaluating("other"));
876 }
877
878 #[test]
879 fn test_simple_cte_execution() {
880 let executor = CteExecutor::new(|_query, _ctx| {
881 let mut result = UnifiedResult::with_columns(vec!["id".to_string()]);
882 let mut record = UnifiedRecord::new();
883 record.set("id", Value::Integer(1));
884 result.push(record);
885 Ok(result)
886 });
887
888 let cte = CteDefinition {
890 name: "test_cte".to_string(),
891 columns: vec!["id".to_string()],
892 query: Box::new(QueryExpr::table("dummy").build()),
893 recursive: false,
894 };
895
896 let with_clause = WithClause::new().add(cte);
897 let query = QueryWithCte::with_ctes(with_clause, QueryExpr::table("test_cte").build());
898
899 let result = executor.execute(&query);
900 assert!(result.is_ok());
901 }
902
903 #[test]
904 fn test_cte_builder() {
905 let query = CteQueryBuilder::new()
906 .cte_with_columns(
907 "nums",
908 vec!["n".to_string()],
909 QueryExpr::table("numbers").build(),
910 )
911 .build(QueryExpr::table("nums").build());
912
913 assert!(query.with_clause.is_some());
914 let with_clause = query.with_clause.unwrap();
915 assert_eq!(with_clause.ctes.len(), 1);
916 assert_eq!(with_clause.ctes[0].name, "nums");
917 }
918
919 #[test]
920 fn test_recursive_cte_builder() {
921 let query = CteQueryBuilder::new()
922 .recursive_cte("paths", QueryExpr::table("connections").build())
923 .build(QueryExpr::table("paths").build());
924
925 assert!(query.with_clause.is_some());
926 let with_clause = query.with_clause.unwrap();
927 assert!(with_clause.has_recursive);
928 assert!(with_clause.ctes[0].recursive);
929 }
930
931 #[test]
932 fn test_circular_reference_detection() {
933 let mut ctx = CteContext::new();
934 ctx.start_evaluating("cte_a");
935
936 assert!(ctx.is_evaluating("cte_a"));
938 }
939
940 #[test]
941 fn test_cte_stats() {
942 let ctx = CteContext::new();
943 let stats = ctx.stats();
944
945 assert_eq!(stats.ctes_executed, 0);
946 assert_eq!(stats.recursive_iterations, 0);
947 assert_eq!(stats.rows_produced, 0);
948 }
949
950 #[test]
951 fn test_hash_record() {
952 let executor = CteExecutor::new(mock_execute);
953
954 let mut record1 = UnifiedRecord::new();
955 record1.set("id", Value::Integer(1));
956 record1.set("name", Value::text("test".to_string()));
957
958 let mut record2 = UnifiedRecord::new();
959 record2.set("id", Value::Integer(1));
960 record2.set("name", Value::text("test".to_string()));
961
962 let mut record3 = UnifiedRecord::new();
963 record3.set("id", Value::Integer(2));
964 record3.set("name", Value::text("test".to_string()));
965
966 assert_eq!(
968 executor.hash_record(&record1),
969 executor.hash_record(&record2)
970 );
971
972 assert_ne!(
974 executor.hash_record(&record1),
975 executor.hash_record(&record3)
976 );
977 }
978
979 #[test]
980 fn test_hash_various_value_types() {
981 let executor = CteExecutor::new(mock_execute);
982
983 let mut record = UnifiedRecord::new();
985 record.set("null_val", Value::Null);
986 record.set("bool_val", Value::Boolean(true));
987 record.set("int_val", Value::Integer(42));
988 record.set("float_val", Value::Float(2.5));
989 record.set("text_val", Value::text("hello".to_string()));
990 record.set("blob_val", Value::Blob(vec![1, 2, 3]));
991 record.set("timestamp_val", Value::Timestamp(1234567890));
992 record.set("duration_val", Value::Duration(5000));
993
994 let hash = executor.hash_record(&record);
996 assert!(hash > 0);
997 }
998
999 #[test]
1000 fn test_project_columns() {
1001 let executor = CteExecutor::new(mock_execute);
1002
1003 let mut original =
1004 UnifiedResult::with_columns(vec!["a".to_string(), "b".to_string(), "c".to_string()]);
1005
1006 let mut record = UnifiedRecord::new();
1007 record.set("a", Value::Integer(1));
1008 record.set("b", Value::Integer(2));
1009 record.set("c", Value::Integer(3));
1010 original.push(record);
1011
1012 let projected = executor.project_columns(&original, &["x".to_string(), "y".to_string()]);
1014
1015 assert_eq!(projected.columns, vec!["x", "y"]);
1016 assert_eq!(projected.len(), 1);
1017 }
1018
1019 #[test]
1020 fn test_empty_columns_projection() {
1021 let executor = CteExecutor::new(mock_execute);
1022
1023 let original = UnifiedResult::with_columns(vec!["a".to_string()]);
1024
1025 let projected = executor.project_columns(&original, &[]);
1027 assert_eq!(projected.columns, original.columns);
1028 }
1029
1030 #[test]
1031 fn test_cte_with_multiple_definitions() {
1032 let executor = CteExecutor::new(|query, ctx| {
1033 match query {
1035 QueryExpr::Table(t) if t.table == "base" => {
1036 let mut result = UnifiedResult::with_columns(vec!["id".to_string()]);
1037 let mut record = UnifiedRecord::new();
1038 record.set("id", Value::Integer(1));
1039 result.push(record);
1040 Ok(result)
1041 }
1042 QueryExpr::Table(t) if t.table == "cte1" => {
1043 if ctx.get("cte1").is_some() {
1045 Ok(ctx.get("cte1").unwrap().clone())
1046 } else {
1047 Ok(UnifiedResult::empty())
1048 }
1049 }
1050 _ => Ok(UnifiedResult::empty()),
1051 }
1052 });
1053
1054 let cte1 = CteDefinition {
1055 name: "cte1".to_string(),
1056 columns: vec!["id".to_string()],
1057 query: Box::new(QueryExpr::table("base").build()),
1058 recursive: false,
1059 };
1060
1061 let cte2 = CteDefinition {
1062 name: "cte2".to_string(),
1063 columns: vec!["id".to_string()],
1064 query: Box::new(QueryExpr::table("cte1").build()),
1065 recursive: false,
1066 };
1067
1068 let with_clause = WithClause::new().add(cte1).add(cte2);
1069 let query = QueryWithCte::with_ctes(with_clause, QueryExpr::table("cte2").build());
1070
1071 let result = executor.execute(&query);
1072 assert!(result.is_ok());
1073 }
1074}