1use std::collections::HashMap;
33use std::sync::Arc;
34use xlog_core::{RelId, Schema};
35use xlog_ir::{CompareOp, Expr, JoinType, RirNode};
36use xlog_stats::StatsManager;
37
38#[derive(Debug, Clone)]
43#[non_exhaustive]
44pub struct OptimizerConfig {
45 pub dp_threshold: usize,
51
52 pub index_heat_threshold: f32,
57
58 pub enable_pushdown: bool,
63
64 pub default_filter_selectivity: f64,
69
70 pub transfer_cost_multiplier: f64,
75
76 pub default_bytes_per_row: u64,
80}
81
82impl Default for OptimizerConfig {
83 fn default() -> Self {
84 Self {
85 dp_threshold: 10,
86 index_heat_threshold: 0.7,
87 enable_pushdown: true,
88 default_filter_selectivity: 0.1,
89 transfer_cost_multiplier: 100.0,
90 default_bytes_per_row: 32,
91 }
92 }
93}
94
95#[derive(Debug, Clone, Default, PartialEq)]
100pub struct PlanCost {
101 pub rows: u64,
103
104 pub cpu_cost: f64,
109
110 pub gpu_mem: u64,
115
116 pub transfers: u32,
121}
122
123impl PlanCost {
124 pub fn with_rows(rows: u64) -> Self {
126 Self {
127 rows,
128 ..Default::default()
129 }
130 }
131
132 pub fn total_cost(&self, transfer_weight: f64) -> f64 {
143 self.cpu_cost + (self.gpu_mem as f64 * 0.001) + (self.transfers as f64 * transfer_weight)
144 }
145
146 pub fn then(self, other: PlanCost) -> PlanCost {
150 PlanCost {
151 rows: other.rows,
152 cpu_cost: self.cpu_cost + other.cpu_cost,
153 gpu_mem: self.gpu_mem.max(other.gpu_mem), transfers: self.transfers + other.transfers,
155 }
156 }
157}
158
159pub struct Optimizer {
165 stats: Arc<StatsManager>,
166 config: OptimizerConfig,
167 schemas: HashMap<RelId, Schema>,
169}
170
171impl Optimizer {
172 pub fn new(stats: Arc<StatsManager>) -> Self {
178 Self {
179 stats,
180 config: OptimizerConfig::default(),
181 schemas: HashMap::new(),
182 }
183 }
184
185 pub fn with_config(stats: Arc<StatsManager>, config: OptimizerConfig) -> Self {
192 Self {
193 stats,
194 config,
195 schemas: HashMap::new(),
196 }
197 }
198
199 pub fn set_schemas(&mut self, schemas: HashMap<RelId, Schema>) {
204 self.schemas = schemas;
205 }
206
207 pub fn config(&self) -> &OptimizerConfig {
209 &self.config
210 }
211
212 pub fn stats(&self) -> &Arc<StatsManager> {
214 &self.stats
215 }
216
217 pub fn optimize(&self, node: RirNode) -> RirNode {
235 if self.config.enable_pushdown {
236 self.predicate_pushdown(node)
237 } else {
238 node
239 }
240 }
241
242 fn predicate_pushdown(&self, node: RirNode) -> RirNode {
260 match node {
261 RirNode::Unit => RirNode::Unit,
263 RirNode::Scan { rel } => RirNode::Scan { rel },
264
265 RirNode::Filter { input, predicate } => {
267 let optimized_input = self.predicate_pushdown(*input);
269
270 match optimized_input {
271 RirNode::Filter {
273 input: inner_input,
274 predicate: inner_pred,
275 } => {
276 let merged = Expr::And(vec![inner_pred, predicate]);
277 RirNode::Filter {
278 input: inner_input,
279 predicate: merged,
280 }
281 }
282
283 RirNode::Project {
285 input: proj_input,
286 columns,
287 } => {
288 if let Some(remapped) =
290 self.remap_predicate_through_project(&predicate, &columns)
291 {
292 RirNode::Project {
294 input: Box::new(RirNode::Filter {
295 input: proj_input,
296 predicate: remapped,
297 }),
298 columns,
299 }
300 } else {
301 RirNode::Filter {
303 input: Box::new(RirNode::Project {
304 input: proj_input,
305 columns,
306 }),
307 predicate,
308 }
309 }
310 }
311
312 RirNode::Join {
314 left,
315 right,
316 left_keys,
317 right_keys,
318 join_type,
319 } => {
320 let left_width = self.estimate_width(&left);
321 let (left_preds, right_preds, remaining) =
322 self.split_predicate_for_join(&predicate, left_width);
323
324 let new_left = if !left_preds.is_empty() {
326 Box::new(RirNode::Filter {
327 input: left,
328 predicate: Self::conjoin(left_preds),
329 })
330 } else {
331 left
332 };
333
334 let new_right = if !right_preds.is_empty() {
335 Box::new(RirNode::Filter {
336 input: right,
337 predicate: Self::conjoin(right_preds),
338 })
339 } else {
340 right
341 };
342
343 let join_node = RirNode::Join {
344 left: new_left,
345 right: new_right,
346 left_keys,
347 right_keys,
348 join_type,
349 };
350
351 if !remaining.is_empty() {
353 RirNode::Filter {
354 input: Box::new(join_node),
355 predicate: Self::conjoin(remaining),
356 }
357 } else {
358 join_node
359 }
360 }
361
362 other => RirNode::Filter {
364 input: Box::new(other),
365 predicate,
366 },
367 }
368 }
369
370 RirNode::Project { input, columns } => {
375 let mut projections = vec![columns];
376 let mut base = *input;
377 while let RirNode::Project { input, columns } = base {
378 projections.push(columns);
379 base = *input;
380 }
381
382 let mut optimized = self.predicate_pushdown(base);
383 for columns in projections.into_iter().rev() {
384 optimized = RirNode::Project {
385 input: Box::new(optimized),
386 columns,
387 };
388 }
389 optimized
390 }
391
392 RirNode::Join {
394 left,
395 right,
396 left_keys,
397 right_keys,
398 join_type,
399 } => RirNode::Join {
400 left: Box::new(self.predicate_pushdown(*left)),
401 right: Box::new(self.predicate_pushdown(*right)),
402 left_keys,
403 right_keys,
404 join_type,
405 },
406
407 RirNode::GroupBy {
409 input,
410 key_cols,
411 aggs,
412 } => RirNode::GroupBy {
413 input: Box::new(self.predicate_pushdown(*input)),
414 key_cols,
415 aggs,
416 },
417
418 RirNode::Union { inputs } => RirNode::Union {
420 inputs: inputs
421 .into_iter()
422 .map(|i| self.predicate_pushdown(i))
423 .collect(),
424 },
425
426 RirNode::Distinct { input, key_cols } => RirNode::Distinct {
428 input: Box::new(self.predicate_pushdown(*input)),
429 key_cols,
430 },
431
432 RirNode::Diff { left, right } => RirNode::Diff {
434 left: Box::new(self.predicate_pushdown(*left)),
435 right: Box::new(self.predicate_pushdown(*right)),
436 },
437
438 RirNode::Fixpoint {
440 scc_id,
441 base,
442 recursive,
443 delta_rel,
444 full_rel,
445 } => RirNode::Fixpoint {
446 scc_id,
447 base: Box::new(self.predicate_pushdown(*base)),
448 recursive: Box::new(self.predicate_pushdown(*recursive)),
449 delta_rel,
450 full_rel,
451 },
452
453 RirNode::TensorMaskedJoin { .. } => node, RirNode::MultiWayJoin { .. } | RirNode::ChainJoin { .. } => node,
459 }
460 }
461
462 fn remap_predicate_through_project(
468 &self,
469 predicate: &Expr,
470 columns: &[xlog_ir::ProjectExpr],
471 ) -> Option<Expr> {
472 let mut output_to_input: std::collections::HashMap<usize, usize> =
475 std::collections::HashMap::new();
476
477 for (out_idx, proj_expr) in columns.iter().enumerate() {
478 if let xlog_ir::ProjectExpr::Column(in_idx) = proj_expr {
479 output_to_input.insert(out_idx, *in_idx);
480 }
481 }
482
483 self.remap_expr(predicate, &output_to_input)
484 }
485
486 fn remap_expr(
488 &self,
489 expr: &Expr,
490 mapping: &std::collections::HashMap<usize, usize>,
491 ) -> Option<Expr> {
492 match expr {
493 Expr::Column(idx) => mapping.get(idx).map(|&new_idx| Expr::Column(new_idx)),
494
495 Expr::Const(val) => Some(Expr::Const(val.clone())),
496
497 Expr::Compare { left, op, right } => {
498 let new_left = self.remap_expr(left, mapping)?;
499 let new_right = self.remap_expr(right, mapping)?;
500 Some(Expr::Compare {
501 left: Box::new(new_left),
502 op: *op,
503 right: Box::new(new_right),
504 })
505 }
506
507 Expr::And(exprs) => {
508 let remapped: Option<Vec<_>> =
509 exprs.iter().map(|e| self.remap_expr(e, mapping)).collect();
510 remapped.map(Expr::And)
511 }
512
513 Expr::Or(exprs) => {
514 let remapped: Option<Vec<_>> =
515 exprs.iter().map(|e| self.remap_expr(e, mapping)).collect();
516 remapped.map(Expr::Or)
517 }
518
519 Expr::Not(inner) => {
520 let remapped = self.remap_expr(inner, mapping)?;
521 Some(Expr::Not(Box::new(remapped)))
522 }
523
524 Expr::Add(l, r) => {
526 let new_l = self.remap_expr(l, mapping)?;
527 let new_r = self.remap_expr(r, mapping)?;
528 Some(Expr::Add(Box::new(new_l), Box::new(new_r)))
529 }
530 Expr::Sub(l, r) => {
531 let new_l = self.remap_expr(l, mapping)?;
532 let new_r = self.remap_expr(r, mapping)?;
533 Some(Expr::Sub(Box::new(new_l), Box::new(new_r)))
534 }
535 Expr::Mul(l, r) => {
536 let new_l = self.remap_expr(l, mapping)?;
537 let new_r = self.remap_expr(r, mapping)?;
538 Some(Expr::Mul(Box::new(new_l), Box::new(new_r)))
539 }
540 Expr::Div(l, r) => {
541 let new_l = self.remap_expr(l, mapping)?;
542 let new_r = self.remap_expr(r, mapping)?;
543 Some(Expr::Div(Box::new(new_l), Box::new(new_r)))
544 }
545 Expr::Mod(l, r) => {
546 let new_l = self.remap_expr(l, mapping)?;
547 let new_r = self.remap_expr(r, mapping)?;
548 Some(Expr::Mod(Box::new(new_l), Box::new(new_r)))
549 }
550
551 Expr::Abs(inner) => {
553 let remapped = self.remap_expr(inner, mapping)?;
554 Some(Expr::Abs(Box::new(remapped)))
555 }
556 Expr::Min(l, r) => {
557 let new_l = self.remap_expr(l, mapping)?;
558 let new_r = self.remap_expr(r, mapping)?;
559 Some(Expr::Min(Box::new(new_l), Box::new(new_r)))
560 }
561 Expr::Max(l, r) => {
562 let new_l = self.remap_expr(l, mapping)?;
563 let new_r = self.remap_expr(r, mapping)?;
564 Some(Expr::Max(Box::new(new_l), Box::new(new_r)))
565 }
566 Expr::Pow(l, r) => {
567 let new_l = self.remap_expr(l, mapping)?;
568 let new_r = self.remap_expr(r, mapping)?;
569 Some(Expr::Pow(Box::new(new_l), Box::new(new_r)))
570 }
571 Expr::Cast(inner, scalar_type) => {
572 let remapped = self.remap_expr(inner, mapping)?;
573 Some(Expr::Cast(Box::new(remapped), *scalar_type))
574 }
575 Expr::Conditional {
576 condition,
577 then_expr,
578 else_expr,
579 } => {
580 let new_condition = self.remap_expr(condition, mapping)?;
581 let new_then = self.remap_expr(then_expr, mapping)?;
582 let new_else = self.remap_expr(else_expr, mapping)?;
583 Some(Expr::Conditional {
584 condition: Box::new(new_condition),
585 then_expr: Box::new(new_then),
586 else_expr: Box::new(new_else),
587 })
588 }
589 }
590 }
591
592 fn estimate_width(&self, node: &RirNode) -> usize {
594 match node {
595 RirNode::Unit => 0,
596 RirNode::Scan { rel } => {
597 if let Some(schema) = self.schemas.get(rel) {
599 schema.arity()
600 } else if let Some(stats) = self.stats.get_relation_stats(*rel) {
601 stats.column_stats.len().max(1)
602 } else {
603 4 }
605 }
606 RirNode::Filter { input, .. } => self.estimate_width(input),
607 RirNode::Project { columns, .. } => columns.len(),
608 RirNode::Join { left, right, .. } => {
609 self.estimate_width(left) + self.estimate_width(right)
610 }
611 RirNode::ChainJoin { output_columns, .. } => output_columns.len(),
612 RirNode::GroupBy { key_cols, aggs, .. } => key_cols.len() + aggs.len(),
613 RirNode::Union { inputs } => {
614 inputs.first().map(|i| self.estimate_width(i)).unwrap_or(0)
615 }
616 RirNode::Distinct { input, .. } => self.estimate_width(input),
617 RirNode::Diff { left, .. } => self.estimate_width(left),
618 RirNode::Fixpoint { base, .. } => self.estimate_width(base),
619 RirNode::TensorMaskedJoin { head_rel_id, .. } => self
622 .schemas
623 .get(head_rel_id)
624 .map(|s| s.arity())
625 .unwrap_or(2),
626 RirNode::MultiWayJoin { output_columns, .. } => output_columns.len(),
629 }
630 }
631
632 fn split_predicate_for_join(
636 &self,
637 predicate: &Expr,
638 left_width: usize,
639 ) -> (Vec<Expr>, Vec<Expr>, Vec<Expr>) {
640 let mut left_preds = Vec::new();
641 let mut right_preds = Vec::new();
642 let mut remaining = Vec::new();
643
644 let conjuncts = Self::flatten_and(predicate);
646
647 for conj in conjuncts {
648 let cols = Self::collect_columns(&conj);
649 let max_col = cols.iter().copied().max().unwrap_or(0);
650 let min_col = cols.iter().copied().min().unwrap_or(0);
651
652 if cols.is_empty() {
653 left_preds.push(conj);
655 } else if max_col < left_width {
656 left_preds.push(conj);
658 } else if min_col >= left_width {
659 let remapped = Self::remap_columns(&conj, |c| c - left_width);
661 right_preds.push(remapped);
662 } else {
663 remaining.push(conj);
665 }
666 }
667
668 (left_preds, right_preds, remaining)
669 }
670
671 fn flatten_and(expr: &Expr) -> Vec<Expr> {
673 match expr {
674 Expr::And(exprs) => exprs.iter().flat_map(Self::flatten_and).collect(),
675 other => vec![other.clone()],
676 }
677 }
678
679 fn collect_columns(expr: &Expr) -> Vec<usize> {
681 match expr {
682 Expr::Column(idx) => vec![*idx],
683 Expr::Const(_) => vec![],
684 Expr::Compare { left, right, .. } => {
685 let mut cols = Self::collect_columns(left);
686 cols.extend(Self::collect_columns(right));
687 cols
688 }
689 Expr::And(exprs) | Expr::Or(exprs) => {
690 exprs.iter().flat_map(Self::collect_columns).collect()
691 }
692 Expr::Not(inner) | Expr::Abs(inner) | Expr::Cast(inner, _) => {
693 Self::collect_columns(inner)
694 }
695 Expr::Add(l, r)
696 | Expr::Sub(l, r)
697 | Expr::Mul(l, r)
698 | Expr::Div(l, r)
699 | Expr::Mod(l, r)
700 | Expr::Min(l, r)
701 | Expr::Max(l, r)
702 | Expr::Pow(l, r) => {
703 let mut cols = Self::collect_columns(l);
704 cols.extend(Self::collect_columns(r));
705 cols
706 }
707 Expr::Conditional {
708 condition,
709 then_expr,
710 else_expr,
711 } => {
712 let mut cols = Self::collect_columns(condition);
713 cols.extend(Self::collect_columns(then_expr));
714 cols.extend(Self::collect_columns(else_expr));
715 cols
716 }
717 }
718 }
719
720 fn remap_columns<F: Fn(usize) -> usize + Copy>(expr: &Expr, f: F) -> Expr {
722 match expr {
723 Expr::Column(idx) => Expr::Column(f(*idx)),
724 Expr::Const(v) => Expr::Const(v.clone()),
725 Expr::Compare { left, op, right } => Expr::Compare {
726 left: Box::new(Self::remap_columns(left, f)),
727 op: *op,
728 right: Box::new(Self::remap_columns(right, f)),
729 },
730 Expr::And(exprs) => {
731 Expr::And(exprs.iter().map(|e| Self::remap_columns(e, f)).collect())
732 }
733 Expr::Or(exprs) => Expr::Or(exprs.iter().map(|e| Self::remap_columns(e, f)).collect()),
734 Expr::Not(inner) => Expr::Not(Box::new(Self::remap_columns(inner, f))),
735 Expr::Add(l, r) => Expr::Add(
736 Box::new(Self::remap_columns(l, f)),
737 Box::new(Self::remap_columns(r, f)),
738 ),
739 Expr::Sub(l, r) => Expr::Sub(
740 Box::new(Self::remap_columns(l, f)),
741 Box::new(Self::remap_columns(r, f)),
742 ),
743 Expr::Mul(l, r) => Expr::Mul(
744 Box::new(Self::remap_columns(l, f)),
745 Box::new(Self::remap_columns(r, f)),
746 ),
747 Expr::Div(l, r) => Expr::Div(
748 Box::new(Self::remap_columns(l, f)),
749 Box::new(Self::remap_columns(r, f)),
750 ),
751 Expr::Mod(l, r) => Expr::Mod(
752 Box::new(Self::remap_columns(l, f)),
753 Box::new(Self::remap_columns(r, f)),
754 ),
755 Expr::Abs(inner) => Expr::Abs(Box::new(Self::remap_columns(inner, f))),
756 Expr::Min(l, r) => Expr::Min(
757 Box::new(Self::remap_columns(l, f)),
758 Box::new(Self::remap_columns(r, f)),
759 ),
760 Expr::Max(l, r) => Expr::Max(
761 Box::new(Self::remap_columns(l, f)),
762 Box::new(Self::remap_columns(r, f)),
763 ),
764 Expr::Pow(l, r) => Expr::Pow(
765 Box::new(Self::remap_columns(l, f)),
766 Box::new(Self::remap_columns(r, f)),
767 ),
768 Expr::Cast(inner, t) => Expr::Cast(Box::new(Self::remap_columns(inner, f)), *t),
769 Expr::Conditional {
770 condition,
771 then_expr,
772 else_expr,
773 } => Expr::Conditional {
774 condition: Box::new(Self::remap_columns(condition, f)),
775 then_expr: Box::new(Self::remap_columns(then_expr, f)),
776 else_expr: Box::new(Self::remap_columns(else_expr, f)),
777 },
778 }
779 }
780
781 fn conjoin(predicates: Vec<Expr>) -> Expr {
783 debug_assert!(!predicates.is_empty());
784 if predicates.len() == 1 {
785 predicates.into_iter().next().unwrap()
786 } else {
787 Expr::And(predicates)
788 }
789 }
790
791 pub fn estimate_cost(&self, node: &RirNode) -> PlanCost {
804 match node {
805 RirNode::Unit => PlanCost {
806 rows: 1,
807 cpu_cost: 0.0,
808 gpu_mem: 0,
809 transfers: 0,
810 },
811 RirNode::Scan { rel } => self.estimate_scan_cost(*rel),
812
813 RirNode::Filter { input, predicate } => {
814 let input_cost = self.estimate_cost(input);
815 self.estimate_filter_cost(input_cost, predicate, input)
816 }
817
818 RirNode::Project { input, columns } => {
819 let input_cost = self.estimate_cost(input);
820 self.estimate_project_cost(input_cost, columns)
821 }
822
823 RirNode::Join {
824 left,
825 right,
826 left_keys,
827 right_keys,
828 join_type,
829 } => {
830 let left_cost = self.estimate_cost(left);
831 let right_cost = self.estimate_cost(right);
832 self.estimate_join_cost(
833 left_cost, right_cost, left, right, left_keys, right_keys, *join_type,
834 )
835 }
836
837 RirNode::ChainJoin {
838 left,
839 right,
840 left_key,
841 right_key,
842 output_columns,
843 ..
844 } => {
845 let left_cost = self.estimate_cost(left);
846 let right_cost = self.estimate_cost(right);
847 let join_cost = self.estimate_join_cost(
848 left_cost,
849 right_cost,
850 left,
851 right,
852 &[*left_key],
853 &[*right_key],
854 JoinType::Inner,
855 );
856 self.estimate_project_cost(join_cost, output_columns)
857 }
858
859 RirNode::GroupBy {
860 input,
861 key_cols,
862 aggs,
863 } => {
864 let input_cost = self.estimate_cost(input);
865 self.estimate_groupby_cost(input_cost, key_cols, aggs)
866 }
867
868 RirNode::Union { inputs } => {
869 let costs: Vec<_> = inputs.iter().map(|i| self.estimate_cost(i)).collect();
870 self.estimate_union_cost(costs)
871 }
872
873 RirNode::Distinct { input, key_cols } => {
874 let input_cost = self.estimate_cost(input);
875 self.estimate_distinct_cost(input_cost, key_cols)
876 }
877
878 RirNode::Diff { left, right } => {
879 let left_cost = self.estimate_cost(left);
880 let right_cost = self.estimate_cost(right);
881 self.estimate_diff_cost(left_cost, right_cost)
882 }
883
884 RirNode::Fixpoint {
885 base, recursive, ..
886 } => {
887 let base_cost = self.estimate_cost(base);
888 let recursive_cost = self.estimate_cost(recursive);
889 self.estimate_fixpoint_cost(base_cost, recursive_cost)
890 }
891
892 RirNode::TensorMaskedJoin {
893 max_active_rules, ..
894 } => PlanCost {
895 rows: *max_active_rules as u64,
896 cpu_cost: *max_active_rules as f64 * 100.0,
897 gpu_mem: *max_active_rules as u64 * 1024,
898 transfers: 1,
899 },
900 RirNode::MultiWayJoin { inputs, .. } => {
905 let mut total = PlanCost::default();
906 for inp in inputs {
907 let c = self.estimate_cost(inp);
908 total.rows = total.rows.saturating_add(c.rows);
909 total.cpu_cost += c.cpu_cost;
910 total.gpu_mem = total.gpu_mem.saturating_add(c.gpu_mem);
911 total.transfers = total.transfers.saturating_add(c.transfers);
912 }
913 total
914 }
915 }
916 }
917
918 fn estimate_scan_cost(&self, rel: RelId) -> PlanCost {
920 if let Some(stats) = self.stats.get_relation_stats(rel) {
921 PlanCost {
922 rows: stats.cardinality,
923 cpu_cost: stats.cardinality as f64 * 0.01, gpu_mem: stats
925 .byte_size
926 .max(stats.cardinality * self.config.default_bytes_per_row),
927 transfers: 0, }
929 } else {
930 let default_rows = 1000;
932 PlanCost {
933 rows: default_rows,
934 cpu_cost: default_rows as f64 * 0.01,
935 gpu_mem: default_rows * self.config.default_bytes_per_row,
936 transfers: 0,
937 }
938 }
939 }
940
941 fn estimate_filter_cost(
943 &self,
944 input_cost: PlanCost,
945 predicate: &Expr,
946 input: &RirNode,
947 ) -> PlanCost {
948 let selectivity = self.estimate_predicate_selectivity(predicate, input);
949 let output_rows = ((input_cost.rows as f64 * selectivity) as u64).max(1);
950
951 PlanCost {
952 rows: output_rows,
953 cpu_cost: input_cost.cpu_cost + input_cost.rows as f64 * 0.02, gpu_mem: input_cost.gpu_mem, transfers: input_cost.transfers,
956 }
957 }
958
959 fn estimate_project_cost(
961 &self,
962 input_cost: PlanCost,
963 columns: &[xlog_ir::ProjectExpr],
964 ) -> PlanCost {
965 let computed_count = columns
967 .iter()
968 .filter(|c| matches!(c, xlog_ir::ProjectExpr::Computed(_, _)))
969 .count();
970
971 let compute_cost = computed_count as f64 * input_cost.rows as f64 * 0.05;
973
974 let output_width_ratio = columns.len() as f64 / (columns.len() + 2) as f64; PlanCost {
978 rows: input_cost.rows,
979 cpu_cost: input_cost.cpu_cost + compute_cost,
980 gpu_mem: (input_cost.gpu_mem as f64 * output_width_ratio) as u64,
981 transfers: input_cost.transfers,
982 }
983 }
984
985 #[allow(clippy::too_many_arguments)]
987 fn estimate_join_cost(
988 &self,
989 left_cost: PlanCost,
990 right_cost: PlanCost,
991 left: &RirNode,
992 right: &RirNode,
993 left_keys: &[usize],
994 right_keys: &[usize],
995 join_type: JoinType,
996 ) -> PlanCost {
997 let output_rows = match join_type {
1000 JoinType::Semi => {
1001 ((left_cost.rows as f64 * 0.5) as u64).max(1)
1003 }
1004 JoinType::Anti => {
1005 ((left_cost.rows as f64 * 0.5) as u64).max(1)
1007 }
1008 JoinType::Inner | JoinType::LeftOuter => {
1009 let left_rels = left.referenced_relations();
1011 let right_rels = right.referenced_relations();
1012
1013 if left_rels.len() == 1 && right_rels.len() == 1 {
1014 let estimated = self.stats.estimate_join_cardinality(
1016 left_rels[0],
1017 right_rels[0],
1018 left_keys,
1019 right_keys,
1020 );
1021
1022 match join_type {
1023 JoinType::LeftOuter => estimated.max(left_cost.rows),
1024 _ => estimated,
1025 }
1026 } else {
1027 match join_type {
1029 JoinType::Inner => {
1030 ((left_cost.rows as f64 * right_cost.rows as f64 * 0.1) as u64).max(1)
1032 }
1033 JoinType::LeftOuter => {
1034 left_cost.rows.max(
1036 ((left_cost.rows as f64 * right_cost.rows as f64 * 0.1) as u64)
1037 .max(1),
1038 )
1039 }
1040 _ => unreachable!(),
1041 }
1042 }
1043 }
1044 };
1045
1046 let build_cost = right_cost.rows as f64 * 1.0; let probe_cost = left_cost.rows as f64 * 0.5; let cpu_cost = left_cost.cpu_cost + right_cost.cpu_cost + build_cost + probe_cost;
1050
1051 let hash_table_overhead = right_cost.gpu_mem / 2; let gpu_mem = left_cost.gpu_mem + right_cost.gpu_mem + hash_table_overhead;
1054
1055 PlanCost {
1056 rows: output_rows,
1057 cpu_cost,
1058 gpu_mem,
1059 transfers: left_cost.transfers + right_cost.transfers,
1060 }
1061 }
1062
1063 fn estimate_groupby_cost(
1065 &self,
1066 input_cost: PlanCost,
1067 key_cols: &[usize],
1068 _aggs: &[(usize, xlog_core::AggOp)],
1069 ) -> PlanCost {
1070 let estimated_groups = if key_cols.is_empty() {
1073 1 } else {
1075 ((input_cost.rows as f64).sqrt() as u64).max(1)
1077 };
1078
1079 PlanCost {
1080 rows: estimated_groups,
1081 cpu_cost: input_cost.cpu_cost + input_cost.rows as f64 * 0.5, gpu_mem: input_cost.gpu_mem + estimated_groups * self.config.default_bytes_per_row,
1083 transfers: input_cost.transfers,
1084 }
1085 }
1086
1087 fn estimate_union_cost(&self, input_costs: Vec<PlanCost>) -> PlanCost {
1089 let total_rows: u64 = input_costs.iter().map(|c| c.rows).sum();
1090 let total_cpu: f64 = input_costs.iter().map(|c| c.cpu_cost).sum();
1091 let max_gpu: u64 = input_costs.iter().map(|c| c.gpu_mem).max().unwrap_or(0);
1092 let total_transfers: u32 = input_costs.iter().map(|c| c.transfers).sum();
1093
1094 PlanCost {
1095 rows: total_rows,
1096 cpu_cost: total_cpu + total_rows as f64 * 0.01, gpu_mem: max_gpu, transfers: total_transfers,
1099 }
1100 }
1101
1102 fn estimate_distinct_cost(&self, input_cost: PlanCost, _key_cols: &[usize]) -> PlanCost {
1104 let estimated_distinct = (input_cost.rows as f64 * 0.7) as u64;
1106
1107 PlanCost {
1108 rows: estimated_distinct.max(1),
1109 cpu_cost: input_cost.cpu_cost + input_cost.rows as f64 * 0.3, gpu_mem: input_cost.gpu_mem + input_cost.rows * 8, transfers: input_cost.transfers,
1112 }
1113 }
1114
1115 fn estimate_diff_cost(&self, left_cost: PlanCost, right_cost: PlanCost) -> PlanCost {
1117 let estimated_remaining = (left_cost.rows as f64 * 0.5) as u64;
1119
1120 PlanCost {
1121 rows: estimated_remaining.max(1),
1122 cpu_cost: left_cost.cpu_cost + right_cost.cpu_cost + right_cost.rows as f64 * 0.5,
1123 gpu_mem: left_cost.gpu_mem + right_cost.gpu_mem,
1124 transfers: left_cost.transfers + right_cost.transfers,
1125 }
1126 }
1127
1128 fn estimate_fixpoint_cost(&self, base_cost: PlanCost, recursive_cost: PlanCost) -> PlanCost {
1130 let estimated_iterations = ((base_cost.rows as f64).log2().ceil() as u64).max(1);
1133
1134 PlanCost {
1135 rows: base_cost.rows * estimated_iterations, cpu_cost: base_cost.cpu_cost + recursive_cost.cpu_cost * estimated_iterations as f64,
1137 gpu_mem: (base_cost.gpu_mem + recursive_cost.gpu_mem) * 2, transfers: base_cost.transfers + recursive_cost.transfers * estimated_iterations as u32,
1139 }
1140 }
1141
1142 fn estimate_predicate_selectivity(&self, predicate: &Expr, input: &RirNode) -> f64 {
1144 match predicate {
1145 Expr::Compare { left, op, right } => {
1146 self.estimate_compare_selectivity(left, *op, right, input)
1147 }
1148 Expr::And(exprs) => {
1149 exprs
1151 .iter()
1152 .map(|e| self.estimate_predicate_selectivity(e, input))
1153 .product()
1154 }
1155 Expr::Or(exprs) => {
1156 exprs
1159 .iter()
1160 .map(|e| self.estimate_predicate_selectivity(e, input))
1161 .fold(0.0, f64::max)
1162 }
1163 Expr::Not(inner) => 1.0 - self.estimate_predicate_selectivity(inner, input),
1164 _ => self.config.default_filter_selectivity,
1165 }
1166 }
1167
1168 fn estimate_compare_selectivity(
1170 &self,
1171 left: &Expr,
1172 op: CompareOp,
1173 right: &Expr,
1174 input: &RirNode,
1175 ) -> f64 {
1176 if let (Expr::Column(col_idx), Expr::Const(_)) | (Expr::Const(_), Expr::Column(col_idx)) =
1178 (left, right)
1179 {
1180 if let Some(rel_id) = self.find_column_relation(input, *col_idx) {
1182 if let Some(stats) = self.stats.get_relation_stats(rel_id) {
1183 if let Some(col_stats) = stats.get_column(*col_idx) {
1184 return match op {
1185 CompareOp::Eq => col_stats.equality_selectivity(stats.cardinality),
1186 CompareOp::Ne => {
1187 1.0 - col_stats.equality_selectivity(stats.cardinality)
1188 }
1189 CompareOp::Lt | CompareOp::Le | CompareOp::Gt | CompareOp::Ge => {
1190 0.33
1192 }
1193 };
1194 }
1195 }
1196 }
1197 }
1198
1199 match op {
1201 CompareOp::Eq => 0.1, CompareOp::Ne => 0.9, CompareOp::Lt | CompareOp::Le | CompareOp::Gt | CompareOp::Ge => 0.33, }
1205 }
1206
1207 fn find_column_relation(&self, node: &RirNode, col_idx: usize) -> Option<RelId> {
1209 match node {
1210 RirNode::Scan { rel } => Some(*rel),
1211 RirNode::Filter { input, .. } => self.find_column_relation(input, col_idx),
1212 RirNode::Project { input, columns } => {
1213 if col_idx < columns.len() {
1215 if let xlog_ir::ProjectExpr::Column(src_idx) = &columns[col_idx] {
1216 return self.find_column_relation(input, *src_idx);
1217 }
1218 }
1219 None
1220 }
1221 RirNode::Join { left, right, .. } => {
1222 let left_width = self.estimate_width(left);
1223 if col_idx < left_width {
1224 self.find_column_relation(left, col_idx)
1225 } else {
1226 self.find_column_relation(right, col_idx - left_width)
1227 }
1228 }
1229 RirNode::MultiWayJoin { .. } => None,
1235 _ => None, }
1237 }
1238
1239 pub fn recommend_indexes(&self) -> Vec<RelId> {
1244 self.stats.hot_relations(self.config.index_heat_threshold)
1245 }
1246
1247 pub fn should_use_greedy(&self, node: &RirNode) -> bool {
1251 let rels = node.referenced_relations();
1252 let unique_rels: std::collections::HashSet<_> = rels.iter().collect();
1253 unique_rels.len() > self.config.dp_threshold
1254 }
1255}
1256
1257pub mod selectivity_pass {
1270 use std::collections::HashMap;
1316 use xlog_core::RelId;
1317 use xlog_ir::ExecutionPlan;
1318 use xlog_stats::StatsManager;
1319
1320 pub fn run(plan: &mut ExecutionPlan, stats: &StatsManager, rel_ids: &HashMap<String, RelId>) {
1330 let _ = rel_ids;
1336 for rules in plan.rules_by_scc.iter_mut() {
1337 for rule in rules.iter_mut() {
1338 if let Some(rewritten) = super::reorder::try_reorder_triangle(&rule.body, stats) {
1339 rule.body = rewritten;
1340 continue;
1341 }
1342 if let Some(rewritten) = super::reorder::try_reorder_4cycle(&rule.body, stats) {
1343 rule.body = rewritten;
1344 }
1345 }
1346 }
1347 }
1348}
1349
1350pub mod helper_split_pass {
1352 use std::collections::{HashMap, HashSet};
1353
1354 use xlog_core::{RelId, ScalarType, Schema};
1355 use xlog_ir::rir::{HelperSplitSpec, KCliqueVariableOrder};
1356 use xlog_ir::{CompiledRule, ExecutionPlan, JoinType, ProjectExpr, RirMeta, RirNode, Scc};
1357 use xlog_stats::StatsManager;
1358
1359 const HEAVY_SKEW_RATIO: f64 = 10.0;
1360
1361 #[derive(Debug, Clone, PartialEq, Eq)]
1363 pub struct HelperRelationSpec {
1364 pub name: String,
1366 pub rel_id: RelId,
1368 pub schema: Schema,
1370 pub source_rels: [RelId; 2],
1372 }
1373
1374 struct JoinStep {
1375 left_keys: Vec<usize>,
1376 right_keys: Vec<usize>,
1377 }
1378
1379 struct LinearBody {
1380 leaves: Vec<RelId>,
1381 leaf_classes: Vec<Vec<u32>>,
1382 joins: Vec<JoinStep>,
1383 project: Vec<ProjectExpr>,
1384 final_classes: Vec<u32>,
1385 }
1386
1387 struct FlatJoin {
1388 leaves: Vec<RelId>,
1389 output_cols: Vec<usize>,
1390 equalities: Vec<(usize, usize)>,
1391 }
1392
1393 struct Candidate {
1394 pair_start: usize,
1395 helper_schema: Schema,
1396 helper_project: Vec<ProjectExpr>,
1397 helper_join_left_keys: Vec<usize>,
1398 helper_join_right_keys: Vec<usize>,
1399 exposed_classes: Vec<u32>,
1400 }
1401
1402 struct Rewrite {
1403 helper_body: RirNode,
1404 outer_body: RirNode,
1405 spec: HelperRelationSpec,
1406 }
1407
1408 #[derive(Clone, Copy)]
1409 struct KCliqueHelperEdge {
1410 slot: usize,
1411 rel: RelId,
1412 left: usize,
1413 right: usize,
1414 }
1415
1416 pub fn run<F>(
1418 plan: &mut ExecutionPlan,
1419 schemas: &HashMap<RelId, Schema>,
1420 stats: &StatsManager,
1421 mut allocate: F,
1422 ) -> Vec<HelperRelationSpec>
1423 where
1424 F: FnMut(Schema) -> (String, RelId),
1425 {
1426 let mut specs = Vec::new();
1427 for scc_idx in 0..plan.rules_by_scc.len() {
1428 let mut rule_idx = 0;
1429 while rule_idx < plan.rules_by_scc[scc_idx].len() {
1430 let rewrite = {
1431 let rule = &plan.rules_by_scc[scc_idx][rule_idx];
1432 try_rewrite_rule(rule, schemas, stats, &mut allocate)
1433 };
1434 if let Some(rewrite) = rewrite {
1435 let helper_rule = CompiledRule {
1436 head: rewrite.spec.name.clone(),
1437 body: rewrite.helper_body,
1438 meta: RirMeta::with_schema(rewrite.spec.schema.clone()),
1439 };
1440 plan.rules_by_scc[scc_idx].insert(rule_idx, helper_rule);
1441 rule_idx += 1;
1442 plan.rules_by_scc[scc_idx][rule_idx].body = rewrite.outer_body;
1443 add_helper_to_scc(&mut plan.sccs, scc_idx, &rewrite.spec.name);
1444 specs.push(rewrite.spec);
1445 }
1446 rule_idx += 1;
1447 }
1448 }
1449 specs
1450 }
1451
1452 pub fn run_kclique_specs<F>(
1458 plan: &mut ExecutionPlan,
1459 schemas: &HashMap<RelId, Schema>,
1460 mut allocate: F,
1461 ) -> Vec<HelperRelationSpec>
1462 where
1463 F: FnMut(Schema) -> (String, RelId),
1464 {
1465 let mut specs = Vec::new();
1466 for scc_idx in 0..plan.rules_by_scc.len() {
1467 let mut rule_idx = 0;
1468 while rule_idx < plan.rules_by_scc[scc_idx].len() {
1469 let rewrite = {
1470 let rule = &plan.rules_by_scc[scc_idx][rule_idx];
1471 try_rewrite_kclique_rule(rule, schemas, &mut allocate)
1472 };
1473 if let Some(rewrite) = rewrite {
1474 let helper_rule = CompiledRule {
1475 head: rewrite.spec.name.clone(),
1476 body: rewrite.helper_body,
1477 meta: RirMeta::with_schema(rewrite.spec.schema.clone()),
1478 };
1479 plan.rules_by_scc[scc_idx].insert(rule_idx, helper_rule);
1480 rule_idx += 1;
1481 plan.rules_by_scc[scc_idx][rule_idx].body = rewrite.outer_body;
1482 add_helper_to_scc(&mut plan.sccs, scc_idx, &rewrite.spec.name);
1483 specs.push(rewrite.spec);
1484 }
1485 rule_idx += 1;
1486 }
1487 }
1488 specs
1489 }
1490
1491 fn add_helper_to_scc(sccs: &mut [Scc], scc_idx: usize, helper: &str) {
1492 if let Some(scc) = sccs.get_mut(scc_idx) {
1493 if !scc.predicates.iter().any(|p| p == helper) {
1494 scc.predicates.push(helper.to_string());
1495 }
1496 }
1497 }
1498
1499 fn try_rewrite_rule<F>(
1500 rule: &CompiledRule,
1501 schemas: &HashMap<RelId, Schema>,
1502 stats: &StatsManager,
1503 allocate: &mut F,
1504 ) -> Option<Rewrite>
1505 where
1506 F: FnMut(Schema) -> (String, RelId),
1507 {
1508 let linear = linearize_project_body(&rule.body, schemas)?;
1509 let candidate = choose_candidate(&linear, schemas, stats)?;
1510 let (helper_name, helper_rel) = allocate(candidate.helper_schema.clone());
1511 let helper_body = build_helper_body(&linear, &candidate);
1512 let outer_body = build_outer_body(&linear, &candidate, helper_rel)?;
1513 Some(Rewrite {
1514 helper_body,
1515 outer_body,
1516 spec: HelperRelationSpec {
1517 name: helper_name,
1518 rel_id: helper_rel,
1519 schema: candidate.helper_schema,
1520 source_rels: [
1521 linear.leaves[candidate.pair_start],
1522 linear.leaves[candidate.pair_start + 1],
1523 ],
1524 },
1525 })
1526 }
1527
1528 fn try_rewrite_kclique_rule<F>(
1529 rule: &CompiledRule,
1530 schemas: &HashMap<RelId, Schema>,
1531 allocate: &mut F,
1532 ) -> Option<Rewrite>
1533 where
1534 F: FnMut(Schema) -> (String, RelId),
1535 {
1536 if !matches!(&rule.body, RirNode::MultiWayJoin { .. }) {
1537 return None;
1538 }
1539 let mut outer_body = rule.body.clone();
1540 let RirNode::MultiWayJoin {
1541 inputs, var_order, ..
1542 } = &mut outer_body
1543 else {
1544 return None;
1545 };
1546 let kclique = var_order.as_ref()?.kclique.as_ref()?;
1547 let spec = kclique.helper_split_specs.first()?;
1548 let (hot_left, hot_right, target) = kclique_helper_edges(inputs, kclique, spec)?;
1549 let helper_schema = schemas.get(&target.rel)?.clone();
1550 let (helper_name, helper_rel) = allocate(helper_schema.clone());
1551 let helper_body = build_kclique_helper_body(spec, hot_left, hot_right, target)?;
1552 *inputs.get_mut(target.slot)? = RirNode::Scan { rel: helper_rel };
1553 Some(Rewrite {
1554 helper_body,
1555 outer_body,
1556 spec: HelperRelationSpec {
1557 name: helper_name,
1558 rel_id: helper_rel,
1559 schema: helper_schema,
1560 source_rels: [hot_left.rel, hot_right.rel],
1561 },
1562 })
1563 }
1564
1565 fn kclique_helper_edges(
1566 inputs: &[RirNode],
1567 kclique: &KCliqueVariableOrder,
1568 spec: &HelperSplitSpec,
1569 ) -> Option<(KCliqueHelperEdge, KCliqueHelperEdge, KCliqueHelperEdge)> {
1570 let k = usize::from(kclique.k);
1571 let hot = usize::from(spec.variable);
1572 let mut hot_edges = Vec::new();
1573 let mut target = None;
1574 for &slot in &spec.edge_slots {
1575 let slot = usize::from(slot);
1576 let (left, right) = kclique_edge_pair(slot, k)?;
1577 let RirNode::Scan { rel } = inputs.get(slot)? else {
1578 return None;
1579 };
1580 let edge = KCliqueHelperEdge {
1581 slot,
1582 rel: *rel,
1583 left,
1584 right,
1585 };
1586 if left == hot || right == hot {
1587 hot_edges.push(edge);
1588 } else {
1589 target = Some(edge);
1590 }
1591 }
1592 if hot_edges.len() != 2 {
1593 return None;
1594 }
1595 Some((hot_edges[0], hot_edges[1], target?))
1596 }
1597
1598 fn build_kclique_helper_body(
1599 spec: &HelperSplitSpec,
1600 hot_left: KCliqueHelperEdge,
1601 hot_right: KCliqueHelperEdge,
1602 target: KCliqueHelperEdge,
1603 ) -> Option<RirNode> {
1604 let hot = usize::from(spec.variable);
1605 let target_left = target.left;
1606 let target_right = target.right;
1607 let first_other = kclique_other_endpoint(hot_left, hot)?;
1608 let second_other = kclique_other_endpoint(hot_right, hot)?;
1609 if ![first_other, second_other].contains(&target_left)
1610 || ![first_other, second_other].contains(&target_right)
1611 {
1612 return None;
1613 }
1614
1615 let first_scan = RirNode::Scan { rel: hot_left.rel };
1616 let second_scan = RirNode::Scan { rel: hot_right.rel };
1617 let target_scan = RirNode::Scan { rel: target.rel };
1618 let first_hot_col = kclique_endpoint_col(hot_left, hot)?;
1619 let second_hot_col = kclique_endpoint_col(hot_right, hot)?;
1620 let first_other_col = kclique_endpoint_col(hot_left, first_other)?;
1621 let second_other_col = 2 + kclique_endpoint_col(hot_right, second_other)?;
1622
1623 let target_left_in_join = if first_other == target_left {
1624 first_other_col
1625 } else {
1626 second_other_col
1627 };
1628 let target_right_in_join = if first_other == target_right {
1629 first_other_col
1630 } else {
1631 second_other_col
1632 };
1633 let target_left_col = kclique_endpoint_col(target, target_left)?;
1634 let target_right_col = kclique_endpoint_col(target, target_right)?;
1635
1636 let hot_join = RirNode::Join {
1637 left: Box::new(first_scan),
1638 right: Box::new(second_scan),
1639 left_keys: vec![first_hot_col],
1640 right_keys: vec![second_hot_col],
1641 join_type: JoinType::Inner,
1642 };
1643 let helper_join = RirNode::Join {
1644 left: Box::new(hot_join),
1645 right: Box::new(target_scan),
1646 left_keys: vec![target_left_in_join, target_right_in_join],
1647 right_keys: vec![target_left_col, target_right_col],
1648 join_type: JoinType::Inner,
1649 };
1650 Some(RirNode::Project {
1651 input: Box::new(helper_join),
1652 columns: vec![ProjectExpr::Column(4), ProjectExpr::Column(5)],
1653 })
1654 }
1655
1656 fn kclique_edge_pair(edge_idx: usize, k: usize) -> Option<(usize, usize)> {
1657 let mut idx = 0usize;
1658 for left in 0..k {
1659 for right in (left + 1)..k {
1660 if idx == edge_idx {
1661 return Some((left, right));
1662 }
1663 idx += 1;
1664 }
1665 }
1666 None
1667 }
1668
1669 fn kclique_endpoint_col(edge: KCliqueHelperEdge, variable: usize) -> Option<usize> {
1670 if edge.left == variable {
1671 Some(0)
1672 } else if edge.right == variable {
1673 Some(1)
1674 } else {
1675 None
1676 }
1677 }
1678
1679 fn kclique_other_endpoint(edge: KCliqueHelperEdge, variable: usize) -> Option<usize> {
1680 if edge.left == variable {
1681 Some(edge.right)
1682 } else if edge.right == variable {
1683 Some(edge.left)
1684 } else {
1685 None
1686 }
1687 }
1688
1689 fn linearize_project_body(
1690 body: &RirNode,
1691 schemas: &HashMap<RelId, Schema>,
1692 ) -> Option<LinearBody> {
1693 let RirNode::Project { input, columns } = body else {
1694 return None;
1695 };
1696 let flat = collect_join_graph(input, schemas)?;
1697 if flat.leaves.len() < 6 {
1698 return None;
1699 }
1700 let mut offsets = Vec::with_capacity(flat.leaves.len());
1701 let mut total_cols = 0usize;
1702 for rel in &flat.leaves {
1703 offsets.push(total_cols);
1704 total_cols += schemas.get(rel)?.arity();
1705 }
1706 let mut uf = UnionFind::new(total_cols);
1707 for (left, right) in flat.equalities {
1708 if left >= total_cols || right >= total_cols {
1709 return None;
1710 }
1711 uf.union(left, right);
1712 }
1713 let mut leaf_classes: Vec<Vec<u32>> = Vec::with_capacity(flat.leaves.len());
1714 for (leaf_idx, rel) in flat.leaves.iter().enumerate() {
1715 let arity = schemas.get(rel)?.arity();
1716 let offset = offsets[leaf_idx];
1717 leaf_classes.push((0..arity).map(|col| uf.find(offset + col) as u32).collect());
1718 }
1719 let final_classes = flat
1720 .output_cols
1721 .iter()
1722 .map(|col| uf.find(*col) as u32)
1723 .collect();
1724 let joins = derive_left_deep_steps(&leaf_classes)?;
1725 Some(LinearBody {
1726 leaves: flat.leaves,
1727 leaf_classes,
1728 joins,
1729 project: columns.clone(),
1730 final_classes,
1731 })
1732 }
1733
1734 fn collect_join_graph(node: &RirNode, schemas: &HashMap<RelId, Schema>) -> Option<FlatJoin> {
1735 match node {
1736 RirNode::Scan { rel } => Some(FlatJoin {
1737 leaves: vec![*rel],
1738 output_cols: (0..schemas.get(rel)?.arity()).collect(),
1739 equalities: Vec::new(),
1740 }),
1741 RirNode::Join {
1742 left,
1743 right,
1744 left_keys,
1745 right_keys,
1746 join_type,
1747 } if *join_type == JoinType::Inner => {
1748 let left_flat = collect_join_graph(left, schemas)?;
1749 let right_flat = collect_join_graph(right, schemas)?;
1750 if left_keys.len() != right_keys.len() {
1751 return None;
1752 }
1753 let right_shift = total_width(&left_flat.leaves, schemas)?;
1754 let mut leaves = left_flat.leaves;
1755 leaves.extend(right_flat.leaves);
1756 let right_output_cols: Vec<usize> = right_flat
1757 .output_cols
1758 .iter()
1759 .map(|col| col + right_shift)
1760 .collect();
1761 let mut equalities = left_flat.equalities;
1762 equalities.extend(
1763 right_flat
1764 .equalities
1765 .iter()
1766 .map(|(left, right)| (left + right_shift, right + right_shift)),
1767 );
1768 for (&left_key, &right_key) in left_keys.iter().zip(right_keys.iter()) {
1769 equalities.push((
1770 *left_flat.output_cols.get(left_key)?,
1771 *right_output_cols.get(right_key)?,
1772 ));
1773 }
1774 let mut output_cols = left_flat.output_cols;
1775 output_cols.extend(right_output_cols);
1776 Some(FlatJoin {
1777 leaves,
1778 output_cols,
1779 equalities,
1780 })
1781 }
1782 _ => None,
1783 }
1784 }
1785
1786 fn total_width(leaves: &[RelId], schemas: &HashMap<RelId, Schema>) -> Option<usize> {
1787 leaves
1788 .iter()
1789 .map(|rel| schemas.get(rel).map(Schema::arity))
1790 .try_fold(0usize, |acc, width| width.map(|width| acc + width))
1791 }
1792
1793 fn derive_left_deep_steps(leaf_classes: &[Vec<u32>]) -> Option<Vec<JoinStep>> {
1794 let mut joins = Vec::with_capacity(leaf_classes.len().saturating_sub(1));
1795 let mut current = leaf_classes.first()?.clone();
1796 for classes in leaf_classes.iter().skip(1) {
1797 let mut left_keys = Vec::new();
1798 let mut right_keys = Vec::new();
1799 for (right_col, class) in classes.iter().enumerate() {
1800 if let Some(left_col) = current
1801 .iter()
1802 .position(|current_class| current_class == class)
1803 {
1804 left_keys.push(left_col);
1805 right_keys.push(right_col);
1806 }
1807 }
1808 if left_keys.is_empty() {
1809 return None;
1810 }
1811 joins.push(JoinStep {
1812 left_keys,
1813 right_keys,
1814 });
1815 current.extend(classes.iter().copied());
1816 }
1817 Some(joins)
1818 }
1819
1820 fn choose_candidate(
1821 linear: &LinearBody,
1822 schemas: &HashMap<RelId, Schema>,
1823 stats: &StatsManager,
1824 ) -> Option<Candidate> {
1825 for pair_start in 3..linear.leaves.len().saturating_sub(1) {
1826 let candidate = build_candidate(linear, schemas, pair_start)?;
1827 if skew_ratio_for_candidate(linear, stats, &candidate) >= HEAVY_SKEW_RATIO {
1828 return Some(candidate);
1829 }
1830 }
1831 None
1832 }
1833
1834 fn build_candidate(
1835 linear: &LinearBody,
1836 schemas: &HashMap<RelId, Schema>,
1837 pair_start: usize,
1838 ) -> Option<Candidate> {
1839 let left_rel = linear.leaves[pair_start];
1840 let right_rel = linear.leaves[pair_start + 1];
1841 let left_schema = schemas.get(&left_rel)?;
1842 let right_schema = schemas.get(&right_rel)?;
1843 let internal_step = linear.joins.get(pair_start)?;
1844 let mut helper_left_keys = Vec::new();
1845 let mut helper_right_keys = Vec::new();
1846 for (&left_key, &right_key) in internal_step
1847 .left_keys
1848 .iter()
1849 .zip(internal_step.right_keys.iter())
1850 {
1851 let class = class_at_state(linear, pair_start + 1, left_key)?;
1852 let left_col = linear.leaf_classes[pair_start]
1853 .iter()
1854 .position(|c| *c == class)?;
1855 helper_left_keys.push(left_col);
1856 helper_right_keys.push(right_key);
1857 }
1858 let internal: HashSet<u32> = helper_left_keys
1859 .iter()
1860 .map(|col| linear.leaf_classes[pair_start][*col])
1861 .collect();
1862 let outside = outside_classes(linear, pair_start);
1863 let output = projected_classes(linear)?;
1864 let mut exposed_classes = Vec::new();
1865 let mut helper_project = Vec::new();
1866 let mut helper_columns = Vec::new();
1867 for (col, class) in linear.leaf_classes[pair_start].iter().copied().enumerate() {
1868 if !internal.contains(&class)
1869 && (outside.contains(&class) || output.contains(&class))
1870 && !exposed_classes.contains(&class)
1871 {
1872 exposed_classes.push(class);
1873 helper_project.push(ProjectExpr::Column(col));
1874 let ty = left_schema.column_type(col).unwrap_or(ScalarType::U32);
1875 helper_columns.push((format!("c{}", helper_columns.len()), ty));
1876 }
1877 }
1878 let right_offset = left_schema.arity();
1879 for (col, class) in linear.leaf_classes[pair_start + 1]
1880 .iter()
1881 .copied()
1882 .enumerate()
1883 {
1884 if !internal.contains(&class)
1885 && (outside.contains(&class) || output.contains(&class))
1886 && !exposed_classes.contains(&class)
1887 {
1888 exposed_classes.push(class);
1889 helper_project.push(ProjectExpr::Column(right_offset + col));
1890 let ty = right_schema.column_type(col).unwrap_or(ScalarType::U32);
1891 helper_columns.push((format!("c{}", helper_columns.len()), ty));
1892 }
1893 }
1894 if exposed_classes.len() != 2 {
1895 return None;
1896 }
1897 Some(Candidate {
1898 pair_start,
1899 helper_schema: Schema::new(helper_columns),
1900 helper_project,
1901 helper_join_left_keys: helper_left_keys,
1902 helper_join_right_keys: helper_right_keys,
1903 exposed_classes,
1904 })
1905 }
1906
1907 fn class_at_state(linear: &LinearBody, leaf_count: usize, col: usize) -> Option<u32> {
1908 let mut idx = col;
1909 for leaf_idx in 0..leaf_count {
1910 let classes = &linear.leaf_classes[leaf_idx];
1911 if idx < classes.len() {
1912 return Some(classes[idx]);
1913 }
1914 idx -= classes.len();
1915 }
1916 None
1917 }
1918
1919 fn outside_classes(linear: &LinearBody, pair_start: usize) -> HashSet<u32> {
1920 linear
1921 .leaf_classes
1922 .iter()
1923 .enumerate()
1924 .filter(|(idx, _)| *idx != pair_start && *idx != pair_start + 1)
1925 .flat_map(|(_, classes)| classes.iter().copied())
1926 .collect()
1927 }
1928
1929 fn projected_classes(linear: &LinearBody) -> Option<HashSet<u32>> {
1930 let mut out = HashSet::new();
1931 for expr in &linear.project {
1932 let ProjectExpr::Column(col) = expr else {
1933 return None;
1934 };
1935 out.insert(*linear.final_classes.get(*col)?);
1936 }
1937 Some(out)
1938 }
1939
1940 fn skew_ratio_for_candidate(
1941 linear: &LinearBody,
1942 stats: &StatsManager,
1943 candidate: &Candidate,
1944 ) -> f64 {
1945 let rel = linear.leaves[candidate.pair_start];
1946 let Some(rel_stats) = stats.get_relation_stats(rel) else {
1947 return 0.0;
1948 };
1949 let mut ratio: f64 = 0.0;
1950 for (col, class) in linear.leaf_classes[candidate.pair_start]
1951 .iter()
1952 .copied()
1953 .enumerate()
1954 {
1955 if !candidate.exposed_classes.contains(&class) {
1956 continue;
1957 }
1958 let Some(col_stats) = rel_stats.get_column(col) else {
1959 continue;
1960 };
1961 if col_stats.distinct_estimate == 0 {
1962 continue;
1963 }
1964 ratio = ratio.max(rel_stats.cardinality as f64 / col_stats.distinct_estimate as f64);
1965 }
1966 ratio
1967 }
1968
1969 fn build_helper_body(linear: &LinearBody, candidate: &Candidate) -> RirNode {
1970 let left = RirNode::Scan {
1971 rel: linear.leaves[candidate.pair_start],
1972 };
1973 let right = RirNode::Scan {
1974 rel: linear.leaves[candidate.pair_start + 1],
1975 };
1976 RirNode::Project {
1977 input: Box::new(RirNode::Join {
1978 left: Box::new(left),
1979 right: Box::new(right),
1980 left_keys: candidate.helper_join_left_keys.clone(),
1981 right_keys: candidate.helper_join_right_keys.clone(),
1982 join_type: JoinType::Inner,
1983 }),
1984 columns: candidate.helper_project.clone(),
1985 }
1986 }
1987
1988 fn build_outer_body(
1989 linear: &LinearBody,
1990 candidate: &Candidate,
1991 helper_rel: RelId,
1992 ) -> Option<RirNode> {
1993 let mut node = RirNode::Scan {
1994 rel: linear.leaves[0],
1995 };
1996 let mut classes = linear.leaf_classes[0].clone();
1997 for leaf_idx in 1..candidate.pair_start {
1998 let step = &linear.joins[leaf_idx - 1];
1999 node = RirNode::Join {
2000 left: Box::new(node),
2001 right: Box::new(RirNode::Scan {
2002 rel: linear.leaves[leaf_idx],
2003 }),
2004 left_keys: step.left_keys.clone(),
2005 right_keys: step.right_keys.clone(),
2006 join_type: JoinType::Inner,
2007 };
2008 classes.extend(linear.leaf_classes[leaf_idx].iter().copied());
2009 }
2010 let prefix_step = &linear.joins[candidate.pair_start - 1];
2011 let mut helper_right_keys = Vec::new();
2012 for &rk in &prefix_step.right_keys {
2013 let class = linear.leaf_classes[candidate.pair_start][rk];
2014 helper_right_keys.push(candidate.exposed_classes.iter().position(|c| *c == class)?);
2015 }
2016 node = RirNode::Join {
2017 left: Box::new(node),
2018 right: Box::new(RirNode::Scan { rel: helper_rel }),
2019 left_keys: prefix_step.left_keys.clone(),
2020 right_keys: helper_right_keys,
2021 join_type: JoinType::Inner,
2022 };
2023 classes.extend(candidate.exposed_classes.iter().copied());
2024 for leaf_idx in candidate.pair_start + 2..linear.leaves.len() {
2025 let step = &linear.joins[leaf_idx - 1];
2026 let mut left_keys = Vec::new();
2027 for &lk in &step.left_keys {
2028 let class = class_at_state(linear, leaf_idx, lk)?;
2029 left_keys.push(classes.iter().position(|c| *c == class)?);
2030 }
2031 node = RirNode::Join {
2032 left: Box::new(node),
2033 right: Box::new(RirNode::Scan {
2034 rel: linear.leaves[leaf_idx],
2035 }),
2036 left_keys,
2037 right_keys: step.right_keys.clone(),
2038 join_type: JoinType::Inner,
2039 };
2040 classes.extend(linear.leaf_classes[leaf_idx].iter().copied());
2041 }
2042 let mut project = Vec::with_capacity(linear.project.len());
2043 for expr in &linear.project {
2044 let ProjectExpr::Column(col) = expr else {
2045 return None;
2046 };
2047 let class = *linear.final_classes.get(*col)?;
2048 let mapped = classes.iter().position(|c| *c == class)?;
2049 project.push(ProjectExpr::Column(mapped));
2050 }
2051 Some(RirNode::Project {
2052 input: Box::new(node),
2053 columns: project,
2054 })
2055 }
2056
2057 struct UnionFind {
2058 parent: Vec<usize>,
2059 }
2060
2061 impl UnionFind {
2062 fn new(len: usize) -> Self {
2063 Self {
2064 parent: (0..len).collect(),
2065 }
2066 }
2067
2068 fn find(&mut self, x: usize) -> usize {
2069 let p = self.parent[x];
2070 if p == x {
2071 x
2072 } else {
2073 let root = self.find(p);
2074 self.parent[x] = root;
2075 root
2076 }
2077 }
2078
2079 fn union(&mut self, a: usize, b: usize) {
2080 let ra = self.find(a);
2081 let rb = self.find(b);
2082 if ra != rb {
2083 self.parent[rb] = ra;
2084 }
2085 }
2086 }
2087}
2088
2089#[path = "optimizer/stream_schedule_pass.rs"]
2090pub mod stream_schedule_pass;
2091
2092#[cfg(test)]
2093mod helper_split_pass_tests {
2094 use std::collections::HashMap;
2095
2096 use super::helper_split_pass;
2097 use xlog_core::{RelId, ScalarType, Schema};
2098 use xlog_ir::{CompiledRule, ExecutionPlan, JoinType, ProjectExpr, RirMeta, RirNode, Scc};
2099 use xlog_stats::{ColumnStats, StatsManager};
2100
2101 fn edge_schema() -> Schema {
2102 Schema::new(vec![
2103 ("c0".to_string(), ScalarType::U32),
2104 ("c1".to_string(), ScalarType::U32),
2105 ])
2106 }
2107
2108 fn helper_schema() -> Schema {
2109 Schema::new(vec![
2110 ("c0".to_string(), ScalarType::U32),
2111 ("c1".to_string(), ScalarType::U32),
2112 ])
2113 }
2114
2115 fn schemas() -> HashMap<RelId, Schema> {
2116 (0..6)
2117 .map(|idx| (RelId(idx), edge_schema()))
2118 .collect::<HashMap<_, _>>()
2119 }
2120
2121 fn left_deep_fixture_body() -> RirNode {
2122 let ab_bc = RirNode::Join {
2123 left: Box::new(RirNode::Scan { rel: RelId(0) }),
2124 right: Box::new(RirNode::Scan { rel: RelId(1) }),
2125 left_keys: vec![1],
2126 right_keys: vec![0],
2127 join_type: JoinType::Inner,
2128 };
2129 let with_cd = RirNode::Join {
2130 left: Box::new(ab_bc),
2131 right: Box::new(RirNode::Scan { rel: RelId(2) }),
2132 left_keys: vec![3],
2133 right_keys: vec![0],
2134 join_type: JoinType::Inner,
2135 };
2136 let with_de = RirNode::Join {
2137 left: Box::new(with_cd),
2138 right: Box::new(RirNode::Scan { rel: RelId(3) }),
2139 left_keys: vec![5],
2140 right_keys: vec![0],
2141 join_type: JoinType::Inner,
2142 };
2143 let with_ef = RirNode::Join {
2144 left: Box::new(with_de),
2145 right: Box::new(RirNode::Scan { rel: RelId(4) }),
2146 left_keys: vec![7],
2147 right_keys: vec![0],
2148 join_type: JoinType::Inner,
2149 };
2150 let with_af = RirNode::Join {
2151 left: Box::new(with_ef),
2152 right: Box::new(RirNode::Scan { rel: RelId(5) }),
2153 left_keys: vec![0, 9],
2154 right_keys: vec![0, 1],
2155 join_type: JoinType::Inner,
2156 };
2157 RirNode::Project {
2158 input: Box::new(with_af),
2159 columns: vec![
2160 ProjectExpr::Column(0),
2161 ProjectExpr::Column(1),
2162 ProjectExpr::Column(3),
2163 ProjectExpr::Column(5),
2164 ProjectExpr::Column(9),
2165 ],
2166 }
2167 }
2168
2169 fn plan() -> ExecutionPlan {
2170 ExecutionPlan {
2171 sccs: vec![Scc {
2172 id: 0,
2173 predicates: vec!["out".to_string()],
2174 is_recursive: false,
2175 }],
2176 strata: vec![],
2177 rules_by_scc: vec![vec![CompiledRule {
2178 head: "out".to_string(),
2179 body: left_deep_fixture_body(),
2180 meta: RirMeta::with_schema(Schema::new(vec![
2181 ("a".to_string(), ScalarType::U32),
2182 ("b".to_string(), ScalarType::U32),
2183 ("c".to_string(), ScalarType::U32),
2184 ("d".to_string(), ScalarType::U32),
2185 ("f".to_string(), ScalarType::U32),
2186 ])),
2187 }]],
2188 generated_query_rules: vec![],
2189 est_memory_peak: 0,
2190 rel_arities: std::collections::HashMap::new(),
2191 }
2192 }
2193
2194 fn stats_for_de(distinct_d: u64) -> StatsManager {
2195 let mut stats = StatsManager::new();
2196 for idx in 0..6 {
2197 stats.register_relation(RelId(idx));
2198 stats.update_cardinality(RelId(idx), 8192);
2199 }
2200 let mut d_col = ColumnStats::new(0, ScalarType::U32);
2201 d_col.update_distinct(distinct_d);
2202 stats.add_column_stats(RelId(3), d_col);
2203 stats
2204 }
2205
2206 fn contains_scan(node: &RirNode, rel: RelId) -> bool {
2207 match node {
2208 RirNode::Scan { rel: scan_rel } => *scan_rel == rel,
2209 RirNode::Join { left, right, .. } | RirNode::ChainJoin { left, right, .. } => {
2210 contains_scan(left, rel) || contains_scan(right, rel)
2211 }
2212 RirNode::Project { input, .. }
2213 | RirNode::Filter { input, .. }
2214 | RirNode::Distinct { input, .. }
2215 | RirNode::GroupBy { input, .. } => contains_scan(input, rel),
2216 RirNode::Union { inputs } => inputs.iter().any(|input| contains_scan(input, rel)),
2217 RirNode::Diff { left, right } => contains_scan(left, rel) || contains_scan(right, rel),
2218 RirNode::Fixpoint {
2219 base, recursive, ..
2220 } => contains_scan(base, rel) || contains_scan(recursive, rel),
2221 RirNode::MultiWayJoin { inputs, .. } => {
2222 inputs.iter().any(|input| contains_scan(input, rel))
2223 }
2224 RirNode::TensorMaskedJoin { rel_index, .. } => {
2225 rel_index.iter().any(|(input_rel, _)| *input_rel == rel)
2226 }
2227 RirNode::Unit => false,
2228 }
2229 }
2230
2231 #[test]
2232 fn helper_split_extracts_buried_pair() {
2233 let mut plan = plan();
2234 let schemas = schemas();
2235 let stats = stats_for_de(1);
2236 let specs = helper_split_pass::run(&mut plan, &schemas, &stats, |_| {
2237 ("__kclique_helper_6".to_string(), RelId(6))
2238 });
2239
2240 assert_eq!(specs.len(), 1);
2241 assert_eq!(specs[0].name, "__kclique_helper_6");
2242 assert_eq!(specs[0].rel_id, RelId(6));
2243 assert_eq!(specs[0].schema, helper_schema());
2244 assert_eq!(specs[0].source_rels, [RelId(3), RelId(4)]);
2245 assert_eq!(plan.rules_by_scc[0].len(), 2);
2246 assert_eq!(plan.rules_by_scc[0][0].head, "__kclique_helper_6");
2247 assert_eq!(plan.rules_by_scc[0][1].head, "out");
2248 assert!(contains_scan(&plan.rules_by_scc[0][1].body, RelId(6)));
2249 assert!(plan.sccs[0]
2250 .predicates
2251 .iter()
2252 .any(|predicate| predicate == "__kclique_helper_6"));
2253 }
2254
2255 #[test]
2256 fn helper_split_ignores_flat_distribution() {
2257 let mut plan = plan();
2258 let schemas = schemas();
2259 let stats = stats_for_de(8192);
2260 let specs = helper_split_pass::run(&mut plan, &schemas, &stats, |_| {
2261 ("__kclique_helper_6".to_string(), RelId(6))
2262 });
2263
2264 assert!(specs.is_empty());
2265 assert_eq!(plan.rules_by_scc[0].len(), 1);
2266 assert!(!contains_scan(&plan.rules_by_scc[0][0].body, RelId(6)));
2267 }
2268}
2269
2270mod reorder {
2273 use std::collections::HashMap;
2274 use xlog_core::RelId;
2275 use xlog_ir::rir::ProjectExpr;
2276 use xlog_ir::{JoinType, RirNode};
2277 use xlog_stats::StatsManager;
2278
2279 fn ac3(atom: u8, col: u8) -> u8 {
2280 atom * 2 + col
2281 }
2282 fn ac4(atom: u8, col: u8) -> u8 {
2283 atom * 2 + col
2284 }
2285 fn uf_find_n<const N: usize>(parent: &mut [u8; N], x: u8) -> u8 {
2286 let mut root = x;
2287 while parent[root as usize] != root {
2288 root = parent[root as usize];
2289 }
2290 let mut cur = x;
2291 while parent[cur as usize] != root {
2292 let next = parent[cur as usize];
2293 parent[cur as usize] = root;
2294 cur = next;
2295 }
2296 root
2297 }
2298 fn uf_union_n<const N: usize>(parent: &mut [u8; N], a: u8, b: u8) {
2299 let ra = uf_find_n(parent, a);
2300 let rb = uf_find_n(parent, b);
2301 if ra != rb {
2302 parent[rb as usize] = ra;
2303 }
2304 }
2305
2306 fn populated_card(stats: &StatsManager, rel: RelId) -> Option<u64> {
2307 stats
2308 .get_relation_stats(rel)
2309 .map(|s| s.cardinality)
2310 .filter(|c| *c > 0)
2311 }
2312
2313 struct TriangleSemantics {
2318 rel_xy: RelId,
2319 rel_yz: RelId,
2320 rel_xz: RelId,
2321 }
2322
2323 fn match_and_infer_triangle(body: &RirNode) -> Option<TriangleSemantics> {
2324 let RirNode::Project {
2325 input: outer_input,
2326 columns,
2327 } = body
2328 else {
2329 return None;
2330 };
2331 let RirNode::Join {
2332 left: l1,
2333 right: r1,
2334 left_keys: lk1,
2335 right_keys: rk1,
2336 join_type: jt1,
2337 } = outer_input.as_ref()
2338 else {
2339 return None;
2340 };
2341 if !matches!(jt1, JoinType::Inner) {
2342 return None;
2343 }
2344 let RirNode::Scan { rel: rel_third } = r1.as_ref() else {
2345 return None;
2346 };
2347 let RirNode::Join {
2348 left: l2,
2349 right: r2,
2350 left_keys: lk2,
2351 right_keys: rk2,
2352 join_type: jt2,
2353 } = l1.as_ref()
2354 else {
2355 return None;
2356 };
2357 if !matches!(jt2, JoinType::Inner) {
2358 return None;
2359 }
2360 let RirNode::Scan { rel: rel_inner_l } = l2.as_ref() else {
2361 return None;
2362 };
2363 let RirNode::Scan { rel: rel_inner_r } = r2.as_ref() else {
2364 return None;
2365 };
2366 if lk2.len() != 1 || rk2.len() != 1 || lk1.len() != 2 || rk1.len() != 2 {
2367 return None;
2368 }
2369 if columns.len() != 3 {
2370 return None;
2371 }
2372 if lk2[0] >= 2 || rk2[0] >= 2 {
2373 return None;
2374 }
2375 if lk1.iter().any(|k| *k >= 4) || rk1.iter().any(|k| *k >= 2) {
2376 return None;
2377 }
2378
2379 let mut parent = [0u8, 1, 2, 3, 4, 5];
2380 uf_union_n::<6>(&mut parent, ac3(0, lk2[0] as u8), ac3(1, rk2[0] as u8));
2381 for i in 0..2 {
2382 let inner_ac = match lk1[i] {
2383 0 => (0u8, 0u8),
2384 1 => (0, 1),
2385 2 => (1, 0),
2386 3 => (1, 1),
2387 _ => return None,
2388 };
2389 uf_union_n::<6>(
2390 &mut parent,
2391 ac3(inner_ac.0, inner_ac.1),
2392 ac3(2, rk1[i] as u8),
2393 );
2394 }
2395 let roots: [u8; 6] = std::array::from_fn(|i| uf_find_n::<6>(&mut parent, i as u8));
2396 let mut counts: HashMap<u8, u8> = HashMap::new();
2397 for r in &roots {
2398 *counts.entry(*r).or_insert(0) += 1;
2399 }
2400 if counts.len() != 3 || counts.values().any(|c| *c != 2) {
2401 return None;
2402 }
2403 let mut head_classes: [u8; 3] = [0; 3];
2404 for (i, pc) in columns.iter().enumerate() {
2405 let ProjectExpr::Column(k) = pc else {
2406 return None;
2407 };
2408 let outer_ac = match *k {
2409 0 => (0u8, 0u8),
2410 1 => (0, 1),
2411 2 => (1, 0),
2412 3 => (1, 1),
2413 4 => (2, 0),
2414 5 => (2, 1),
2415 _ => return None,
2416 };
2417 head_classes[i] = uf_find_n::<6>(&mut parent, ac3(outer_ac.0, outer_ac.1));
2418 }
2419 if head_classes[0] == head_classes[1]
2420 || head_classes[0] == head_classes[2]
2421 || head_classes[1] == head_classes[2]
2422 {
2423 return None;
2424 }
2425 let x_class = head_classes[0];
2426 let y_class = head_classes[1];
2427 let z_class = head_classes[2];
2428 let atom_classes = |a: u8| (roots[ac3(a, 0) as usize], roots[ac3(a, 1) as usize]);
2429 let atom_rels = [*rel_inner_l, *rel_inner_r, *rel_third];
2430 let mut rel_xy = None;
2431 let mut rel_yz = None;
2432 let mut rel_xz = None;
2433 for atom_idx in 0..3u8 {
2434 let (c0, c1) = atom_classes(atom_idx);
2435 let bx = c0 == x_class || c1 == x_class;
2436 let by = c0 == y_class || c1 == y_class;
2437 let bz = c0 == z_class || c1 == z_class;
2438 match (bx, by, bz) {
2439 (true, true, false) => rel_xy = Some(atom_rels[atom_idx as usize]),
2440 (false, true, true) => rel_yz = Some(atom_rels[atom_idx as usize]),
2441 (true, false, true) => rel_xz = Some(atom_rels[atom_idx as usize]),
2442 _ => return None,
2443 }
2444 }
2445 Some(TriangleSemantics {
2446 rel_xy: rel_xy?,
2447 rel_yz: rel_yz?,
2448 rel_xz: rel_xz?,
2449 })
2450 }
2451
2452 #[derive(Clone, Copy, Debug, PartialEq, Eq)]
2453 #[allow(clippy::enum_variant_names)]
2454 enum TriangleInnerPair {
2455 YShared,
2456 XShared,
2457 ZShared,
2458 }
2459
2460 fn build_triangle_body(s: &TriangleSemantics, inner_pair: TriangleInnerPair) -> RirNode {
2461 let mk_scan = |r: RelId| RirNode::Scan { rel: r };
2462 match inner_pair {
2463 TriangleInnerPair::YShared => {
2464 let inner = RirNode::Join {
2465 left: Box::new(mk_scan(s.rel_xy)),
2466 right: Box::new(mk_scan(s.rel_yz)),
2467 left_keys: vec![1],
2468 right_keys: vec![0],
2469 join_type: JoinType::Inner,
2470 };
2471 let outer = RirNode::Join {
2472 left: Box::new(inner),
2473 right: Box::new(mk_scan(s.rel_xz)),
2474 left_keys: vec![0, 3],
2475 right_keys: vec![0, 1],
2476 join_type: JoinType::Inner,
2477 };
2478 RirNode::Project {
2479 input: Box::new(outer),
2480 columns: vec![
2481 ProjectExpr::Column(0),
2482 ProjectExpr::Column(1),
2483 ProjectExpr::Column(3),
2484 ],
2485 }
2486 }
2487 TriangleInnerPair::XShared => {
2488 let inner = RirNode::Join {
2489 left: Box::new(mk_scan(s.rel_xy)),
2490 right: Box::new(mk_scan(s.rel_xz)),
2491 left_keys: vec![0],
2492 right_keys: vec![0],
2493 join_type: JoinType::Inner,
2494 };
2495 let outer = RirNode::Join {
2496 left: Box::new(inner),
2497 right: Box::new(mk_scan(s.rel_yz)),
2498 left_keys: vec![1, 3],
2499 right_keys: vec![0, 1],
2500 join_type: JoinType::Inner,
2501 };
2502 RirNode::Project {
2503 input: Box::new(outer),
2504 columns: vec![
2505 ProjectExpr::Column(0),
2506 ProjectExpr::Column(1),
2507 ProjectExpr::Column(3),
2508 ],
2509 }
2510 }
2511 TriangleInnerPair::ZShared => {
2512 let inner = RirNode::Join {
2513 left: Box::new(mk_scan(s.rel_xz)),
2514 right: Box::new(mk_scan(s.rel_yz)),
2515 left_keys: vec![1],
2516 right_keys: vec![1],
2517 join_type: JoinType::Inner,
2518 };
2519 let outer = RirNode::Join {
2520 left: Box::new(inner),
2521 right: Box::new(mk_scan(s.rel_xy)),
2522 left_keys: vec![0, 2],
2523 right_keys: vec![0, 1],
2524 join_type: JoinType::Inner,
2525 };
2526 RirNode::Project {
2527 input: Box::new(outer),
2528 columns: vec![
2529 ProjectExpr::Column(0),
2530 ProjectExpr::Column(2),
2531 ProjectExpr::Column(3),
2532 ],
2533 }
2534 }
2535 }
2536 }
2537
2538 pub fn try_reorder_triangle(body: &RirNode, stats: &StatsManager) -> Option<RirNode> {
2539 let s = match_and_infer_triangle(body)?;
2540 let _ = (
2541 populated_card(stats, s.rel_xy)?,
2542 populated_card(stats, s.rel_yz)?,
2543 populated_card(stats, s.rel_xz)?,
2544 );
2545 let est_y = stats.estimate_join_cardinality(s.rel_xy, s.rel_yz, &[1], &[0]);
2546 let est_x = stats.estimate_join_cardinality(s.rel_xy, s.rel_xz, &[0], &[0]);
2547 let est_z = stats.estimate_join_cardinality(s.rel_yz, s.rel_xz, &[1], &[1]);
2548 let mut best = (TriangleInnerPair::YShared, est_y);
2549 if est_x < best.1 {
2550 best = (TriangleInnerPair::XShared, est_x);
2551 }
2552 if est_z < best.1 {
2553 best = (TriangleInnerPair::ZShared, est_z);
2554 }
2555 let candidate = build_triangle_body(&s, best.0);
2556 if format!("{:?}", candidate) == format!("{:?}", body) {
2562 return None;
2563 }
2564 Some(candidate)
2565 }
2566
2567 struct Cycle4Semantics {
2572 rel_wx: RelId,
2573 rel_xy: RelId,
2574 rel_yz: RelId,
2575 rel_zw: RelId,
2576 }
2577
2578 fn match_and_infer_4cycle(body: &RirNode) -> Option<Cycle4Semantics> {
2579 let RirNode::Project {
2580 input: outer_input,
2581 columns,
2582 } = body
2583 else {
2584 return None;
2585 };
2586 let RirNode::Join {
2587 left: outer_l,
2588 right: outer_r,
2589 left_keys: olk,
2590 right_keys: ork,
2591 join_type: ojt,
2592 } = outer_input.as_ref()
2593 else {
2594 return None;
2595 };
2596 if !matches!(ojt, JoinType::Inner) {
2597 return None;
2598 }
2599 let RirNode::Join {
2600 left: ll,
2601 right: lr,
2602 left_keys: ilk_l,
2603 right_keys: irk_l,
2604 join_type: ijt_l,
2605 } = outer_l.as_ref()
2606 else {
2607 return None;
2608 };
2609 if !matches!(ijt_l, JoinType::Inner) {
2610 return None;
2611 }
2612 let RirNode::Scan { rel: rel_ll } = ll.as_ref() else {
2613 return None;
2614 };
2615 let RirNode::Scan { rel: rel_lr } = lr.as_ref() else {
2616 return None;
2617 };
2618 let RirNode::Join {
2619 left: rl,
2620 right: rr,
2621 left_keys: ilk_r,
2622 right_keys: irk_r,
2623 join_type: ijt_r,
2624 } = outer_r.as_ref()
2625 else {
2626 return None;
2627 };
2628 if !matches!(ijt_r, JoinType::Inner) {
2629 return None;
2630 }
2631 let RirNode::Scan { rel: rel_rl } = rl.as_ref() else {
2632 return None;
2633 };
2634 let RirNode::Scan { rel: rel_rr } = rr.as_ref() else {
2635 return None;
2636 };
2637 if ilk_l.len() != 1 || irk_l.len() != 1 || ilk_r.len() != 1 || irk_r.len() != 1 {
2638 return None;
2639 }
2640 if olk.len() != 2 || ork.len() != 2 || columns.len() != 4 {
2641 return None;
2642 }
2643 if ilk_l[0] >= 2 || irk_l[0] >= 2 || ilk_r[0] >= 2 || irk_r[0] >= 2 {
2644 return None;
2645 }
2646 if olk.iter().any(|k| *k >= 4) || ork.iter().any(|k| *k >= 4) {
2647 return None;
2648 }
2649
2650 let mut parent = [0u8, 1, 2, 3, 4, 5, 6, 7];
2651 uf_union_n::<8>(&mut parent, ac4(0, ilk_l[0] as u8), ac4(1, irk_l[0] as u8));
2652 uf_union_n::<8>(&mut parent, ac4(2, ilk_r[0] as u8), ac4(3, irk_r[0] as u8));
2653 for i in 0..2 {
2654 let l_ac = match olk[i] {
2655 0 => (0u8, 0u8),
2656 1 => (0, 1),
2657 2 => (1, 0),
2658 3 => (1, 1),
2659 _ => return None,
2660 };
2661 let r_ac = match ork[i] {
2662 0 => (2u8, 0u8),
2663 1 => (2, 1),
2664 2 => (3, 0),
2665 3 => (3, 1),
2666 _ => return None,
2667 };
2668 uf_union_n::<8>(&mut parent, ac4(l_ac.0, l_ac.1), ac4(r_ac.0, r_ac.1));
2669 }
2670 let roots: [u8; 8] = std::array::from_fn(|i| uf_find_n::<8>(&mut parent, i as u8));
2671 let mut counts: HashMap<u8, u8> = HashMap::new();
2672 for r in &roots {
2673 *counts.entry(*r).or_insert(0) += 1;
2674 }
2675 if counts.len() != 4 || counts.values().any(|c| *c != 2) {
2676 return None;
2677 }
2678
2679 let mut head_classes: [u8; 4] = [0; 4];
2680 for (i, pc) in columns.iter().enumerate() {
2681 let ProjectExpr::Column(k) = pc else {
2682 return None;
2683 };
2684 let ac = match *k {
2685 0 => (0u8, 0u8),
2686 1 => (0, 1),
2687 2 => (1, 0),
2688 3 => (1, 1),
2689 4 => (2, 0),
2690 5 => (2, 1),
2691 6 => (3, 0),
2692 7 => (3, 1),
2693 _ => return None,
2694 };
2695 head_classes[i] = uf_find_n::<8>(&mut parent, ac4(ac.0, ac.1));
2696 }
2697 for i in 0..4 {
2698 for j in (i + 1)..4 {
2699 if head_classes[i] == head_classes[j] {
2700 return None;
2701 }
2702 }
2703 }
2704 let w_class = head_classes[0];
2705 let x_class = head_classes[1];
2706 let y_class = head_classes[2];
2707 let z_class = head_classes[3];
2708 let atom_classes = |a: u8| (roots[ac4(a, 0) as usize], roots[ac4(a, 1) as usize]);
2709 let atom_rels = [*rel_ll, *rel_lr, *rel_rl, *rel_rr];
2710 let mut rel_wx = None;
2711 let mut rel_xy = None;
2712 let mut rel_yz = None;
2713 let mut rel_zw = None;
2714 for atom_idx in 0..4u8 {
2715 let (c0, c1) = atom_classes(atom_idx);
2716 let bw = c0 == w_class || c1 == w_class;
2717 let bx = c0 == x_class || c1 == x_class;
2718 let by = c0 == y_class || c1 == y_class;
2719 let bz = c0 == z_class || c1 == z_class;
2720 match (bw, bx, by, bz) {
2721 (true, true, false, false) => rel_wx = Some(atom_rels[atom_idx as usize]),
2722 (false, true, true, false) => rel_xy = Some(atom_rels[atom_idx as usize]),
2723 (false, false, true, true) => rel_yz = Some(atom_rels[atom_idx as usize]),
2724 (true, false, false, true) => rel_zw = Some(atom_rels[atom_idx as usize]),
2725 _ => return None,
2726 }
2727 }
2728 Some(Cycle4Semantics {
2729 rel_wx: rel_wx?,
2730 rel_xy: rel_xy?,
2731 rel_yz: rel_yz?,
2732 rel_zw: rel_zw?,
2733 })
2734 }
2735
2736 #[derive(Clone, Copy, Debug, PartialEq, Eq)]
2737 enum Cycle4Grouping {
2738 Default,
2739 Alt,
2740 }
2741
2742 fn build_4cycle_body(s: &Cycle4Semantics, g: Cycle4Grouping) -> RirNode {
2743 let mk_scan = |r: RelId| RirNode::Scan { rel: r };
2744 match g {
2745 Cycle4Grouping::Default => {
2746 let il = RirNode::Join {
2747 left: Box::new(mk_scan(s.rel_wx)),
2748 right: Box::new(mk_scan(s.rel_xy)),
2749 left_keys: vec![1],
2750 right_keys: vec![0],
2751 join_type: JoinType::Inner,
2752 };
2753 let ir = RirNode::Join {
2754 left: Box::new(mk_scan(s.rel_yz)),
2755 right: Box::new(mk_scan(s.rel_zw)),
2756 left_keys: vec![1],
2757 right_keys: vec![0],
2758 join_type: JoinType::Inner,
2759 };
2760 let outer = RirNode::Join {
2761 left: Box::new(il),
2762 right: Box::new(ir),
2763 left_keys: vec![0, 3],
2764 right_keys: vec![3, 0],
2765 join_type: JoinType::Inner,
2766 };
2767 RirNode::Project {
2768 input: Box::new(outer),
2769 columns: vec![
2770 ProjectExpr::Column(0),
2771 ProjectExpr::Column(1),
2772 ProjectExpr::Column(3),
2773 ProjectExpr::Column(5),
2774 ],
2775 }
2776 }
2777 Cycle4Grouping::Alt => {
2778 let il = RirNode::Join {
2779 left: Box::new(mk_scan(s.rel_xy)),
2780 right: Box::new(mk_scan(s.rel_yz)),
2781 left_keys: vec![1],
2782 right_keys: vec![0],
2783 join_type: JoinType::Inner,
2784 };
2785 let ir = RirNode::Join {
2786 left: Box::new(mk_scan(s.rel_zw)),
2787 right: Box::new(mk_scan(s.rel_wx)),
2788 left_keys: vec![1],
2789 right_keys: vec![0],
2790 join_type: JoinType::Inner,
2791 };
2792 let outer = RirNode::Join {
2793 left: Box::new(il),
2794 right: Box::new(ir),
2795 left_keys: vec![0, 3],
2796 right_keys: vec![3, 0],
2797 join_type: JoinType::Inner,
2798 };
2799 RirNode::Project {
2800 input: Box::new(outer),
2801 columns: vec![
2802 ProjectExpr::Column(5),
2803 ProjectExpr::Column(0),
2804 ProjectExpr::Column(1),
2805 ProjectExpr::Column(3),
2806 ],
2807 }
2808 }
2809 }
2810 }
2811
2812 pub fn try_reorder_4cycle(body: &RirNode, stats: &StatsManager) -> Option<RirNode> {
2813 let s = match_and_infer_4cycle(body)?;
2814 let _ = (
2815 populated_card(stats, s.rel_wx)?,
2816 populated_card(stats, s.rel_xy)?,
2817 populated_card(stats, s.rel_yz)?,
2818 populated_card(stats, s.rel_zw)?,
2819 );
2820 let est_default = stats
2821 .estimate_join_cardinality(s.rel_wx, s.rel_xy, &[1], &[0])
2822 .saturating_add(stats.estimate_join_cardinality(s.rel_yz, s.rel_zw, &[1], &[0]));
2823 let est_alt = stats
2824 .estimate_join_cardinality(s.rel_xy, s.rel_yz, &[1], &[0])
2825 .saturating_add(stats.estimate_join_cardinality(s.rel_zw, s.rel_wx, &[1], &[0]));
2826 let chosen = if est_alt < est_default {
2827 Cycle4Grouping::Alt
2828 } else {
2829 Cycle4Grouping::Default
2830 };
2831 let candidate = build_4cycle_body(&s, chosen);
2832 if format!("{:?}", candidate) == format!("{:?}", body) {
2833 return None;
2834 }
2835 Some(candidate)
2836 }
2837}
2838
2839#[cfg(test)]
2840mod selectivity_pass_tests {
2841 use super::selectivity_pass;
2842 use crate::Compiler;
2843 use xlog_stats::StatsManager;
2844
2845 fn body_snapshots(plan: &xlog_ir::ExecutionPlan) -> Vec<String> {
2846 plan.rules_by_scc
2847 .iter()
2848 .flatten()
2849 .map(|r| format!("{:?}", r.body))
2850 .collect()
2851 }
2852
2853 #[test]
2854 fn selectivity_pass_is_noop_for_triangle_plan() {
2855 let mut compiler = Compiler::new();
2856 let plan = compiler
2857 .compile("tri(X, Y, Z) :- e1(X, Y), e2(Y, Z), e3(X, Z).")
2858 .expect("compile");
2859 let before = body_snapshots(&plan);
2860 let stats = StatsManager::new();
2861 let mut plan2 = plan.clone();
2862 selectivity_pass::run(&mut plan2, &stats, &std::collections::HashMap::new());
2863 let after = body_snapshots(&plan2);
2864 assert_eq!(
2865 before, after,
2866 "selectivity_pass must preserve every triangle rule body byte-for-byte"
2867 );
2868 }
2869
2870 #[test]
2871 fn selectivity_pass_is_noop_for_4cycle_plan() {
2872 let mut compiler = Compiler::new();
2873 let plan = compiler
2874 .compile("cycle4(W, X, Y, Z) :- e1(W, X), e2(X, Y), e3(Y, Z), e4(Z, W).")
2875 .expect("compile");
2876 let before = body_snapshots(&plan);
2877 let stats = StatsManager::new();
2878 let mut plan2 = plan.clone();
2879 selectivity_pass::run(&mut plan2, &stats, &std::collections::HashMap::new());
2880 let after = body_snapshots(&plan2);
2881 assert_eq!(
2882 before, after,
2883 "selectivity_pass must preserve every 4-cycle rule body byte-for-byte"
2884 );
2885 }
2886
2887 #[test]
2888 fn selectivity_pass_is_noop_for_recursive_scc() {
2889 let mut compiler = Compiler::new();
2890 let plan = compiler
2891 .compile(
2892 "edge(1, 2). edge(2, 3). \
2893 reach(X, Y) :- edge(X, Y). \
2894 reach(X, Z) :- reach(X, Y), edge(Y, Z).",
2895 )
2896 .expect("compile");
2897 let before = body_snapshots(&plan);
2898 let stats = StatsManager::new();
2899 let mut plan2 = plan.clone();
2900 selectivity_pass::run(&mut plan2, &stats, &std::collections::HashMap::new());
2901 let after = body_snapshots(&plan2);
2902 assert_eq!(
2903 before, after,
2904 "selectivity_pass must preserve recursive SCC bodies byte-for-byte"
2905 );
2906 }
2907
2908 use xlog_core::RelId;
2913 use xlog_ir::plan::{CompiledRule, PlanBuilder, Scc};
2914 use xlog_ir::rir::ProjectExpr;
2915 use xlog_ir::{ExecutionPlan, JoinType, RirNode};
2916
2917 fn synth_triangle_plan() -> ExecutionPlan {
2926 let inner = RirNode::Join {
2927 left: Box::new(RirNode::Scan { rel: RelId(1) }),
2928 right: Box::new(RirNode::Scan { rel: RelId(2) }),
2929 left_keys: vec![1],
2930 right_keys: vec![0],
2931 join_type: JoinType::Inner,
2932 };
2933 let outer = RirNode::Join {
2934 left: Box::new(inner),
2935 right: Box::new(RirNode::Scan { rel: RelId(3) }),
2936 left_keys: vec![0, 3],
2937 right_keys: vec![0, 1],
2938 join_type: JoinType::Inner,
2939 };
2940 let body = RirNode::Project {
2941 input: Box::new(outer),
2942 columns: vec![
2943 ProjectExpr::Column(0),
2944 ProjectExpr::Column(1),
2945 ProjectExpr::Column(3),
2946 ],
2947 };
2948 let mut builder = PlanBuilder::new();
2949 builder.add_scc(Scc {
2950 id: 0,
2951 predicates: vec!["tri".to_string()],
2952 is_recursive: false,
2953 });
2954 builder.add_rule(
2955 0,
2956 CompiledRule {
2957 head: "tri".to_string(),
2958 body,
2959 meta: Default::default(),
2960 },
2961 );
2962 builder.build()
2963 }
2964
2965 fn seed_triangle_stats(c1: u64, c2: u64, c3: u64) -> StatsManager {
2969 let mut stats = StatsManager::new();
2970 for (rid, card) in [(RelId(1), c1), (RelId(2), c2), (RelId(3), c3)] {
2971 stats.register_relation(rid);
2972 stats.update_cardinality(rid, card);
2973 }
2974 stats
2975 }
2976
2977 fn inspect_triangle_inner_pair(plan: &xlog_ir::ExecutionPlan) -> Option<(RelId, RelId)> {
2988 let body = &plan.rules_by_scc.iter().flatten().next()?.body;
2989 let body = match body {
2990 xlog_ir::RirNode::MultiWayJoin { fallback, .. } => fallback.as_ref(),
2991 other => other,
2992 };
2993 let xlog_ir::RirNode::Project { input, .. } = body else {
2994 return None;
2995 };
2996 let xlog_ir::RirNode::Join { left, .. } = input.as_ref() else {
2997 return None;
2998 };
2999 let xlog_ir::RirNode::Join {
3000 left: l2,
3001 right: r2,
3002 ..
3003 } = left.as_ref()
3004 else {
3005 return None;
3006 };
3007 let xlog_ir::RirNode::Scan { rel: rel_l } = l2.as_ref() else {
3008 return None;
3009 };
3010 let xlog_ir::RirNode::Scan { rel: rel_r } = r2.as_ref() else {
3011 return None;
3012 };
3013 Some((*rel_l, *rel_r))
3014 }
3015
3016 #[test]
3023 fn selectivity_pass_picks_y_shared_inner_when_e1_e2_smallest() {
3024 let mut plan = synth_triangle_plan();
3025 let stats = seed_triangle_stats(10, 10, 100_000);
3027 selectivity_pass::run(&mut plan, &stats, &std::collections::HashMap::new());
3028 let pair = inspect_triangle_inner_pair(&plan).expect("inner pair");
3029 assert!(
3031 pair == (RelId(1), RelId(2)) || pair == (RelId(2), RelId(1)),
3032 "expected (RelId(1), RelId(2)) for Y-shared; got {:?}",
3033 pair
3034 );
3035 }
3036
3037 #[test]
3040 fn selectivity_pass_picks_x_shared_inner_when_e1_e3_smallest() {
3041 let mut plan = synth_triangle_plan();
3042 let stats = seed_triangle_stats(10, 100_000, 10);
3044 selectivity_pass::run(&mut plan, &stats, &std::collections::HashMap::new());
3045 let pair = inspect_triangle_inner_pair(&plan).expect("inner pair");
3046 assert!(
3048 pair == (RelId(1), RelId(3)) || pair == (RelId(3), RelId(1)),
3049 "expected (RelId(1), RelId(3)) for X-shared; got {:?}",
3050 pair
3051 );
3052 }
3053
3054 #[test]
3057 fn selectivity_pass_picks_z_shared_inner_when_e2_e3_smallest() {
3058 let mut plan = synth_triangle_plan();
3059 let stats = seed_triangle_stats(100_000, 10, 10);
3061 selectivity_pass::run(&mut plan, &stats, &std::collections::HashMap::new());
3062 let pair = inspect_triangle_inner_pair(&plan).expect("inner pair");
3063 assert!(
3065 pair == (RelId(2), RelId(3)) || pair == (RelId(3), RelId(2)),
3066 "expected (RelId(2), RelId(3)) for Z-shared; got {:?}",
3067 pair
3068 );
3069 }
3070
3071 #[test]
3076 fn selectivity_pass_two_snapshots_produce_different_inner_pairs() {
3077 let mut plan_a = synth_triangle_plan();
3078 let stats_a = seed_triangle_stats(10, 10, 100_000); selectivity_pass::run(&mut plan_a, &stats_a, &std::collections::HashMap::new());
3080 let pair_a = inspect_triangle_inner_pair(&plan_a).expect("snapshot A pair");
3081
3082 let mut plan_b = synth_triangle_plan();
3083 let stats_b = seed_triangle_stats(100_000, 10, 10); selectivity_pass::run(&mut plan_b, &stats_b, &std::collections::HashMap::new());
3085 let pair_b = inspect_triangle_inner_pair(&plan_b).expect("snapshot B pair");
3086
3087 let normalize = |(a, b): (RelId, RelId)| -> (RelId, RelId) {
3088 if a.0 <= b.0 {
3089 (a, b)
3090 } else {
3091 (b, a)
3092 }
3093 };
3094 assert_ne!(
3095 normalize(pair_a),
3096 normalize(pair_b),
3097 "two different stats snapshots must produce different inner pairs; \
3098 got A = {:?}, B = {:?}",
3099 pair_a,
3100 pair_b
3101 );
3102 }
3103
3104 #[test]
3112 fn selectivity_pass_with_only_relation_cards_may_pick_arbitrary_pair() {
3113 let mut plan = synth_triangle_plan();
3114 let stats = seed_triangle_stats(100, 100, 100);
3116 selectivity_pass::run(&mut plan, &stats, &std::collections::HashMap::new());
3117 let _ = inspect_triangle_inner_pair(&plan);
3120 }
3121
3122 fn synth_4cycle_plan() -> ExecutionPlan {
3133 let inner_left = RirNode::Join {
3134 left: Box::new(RirNode::Scan { rel: RelId(1) }),
3135 right: Box::new(RirNode::Scan { rel: RelId(2) }),
3136 left_keys: vec![1],
3137 right_keys: vec![0],
3138 join_type: JoinType::Inner,
3139 };
3140 let inner_right = RirNode::Join {
3141 left: Box::new(RirNode::Scan { rel: RelId(3) }),
3142 right: Box::new(RirNode::Scan { rel: RelId(4) }),
3143 left_keys: vec![1],
3144 right_keys: vec![0],
3145 join_type: JoinType::Inner,
3146 };
3147 let outer = RirNode::Join {
3148 left: Box::new(inner_left),
3149 right: Box::new(inner_right),
3150 left_keys: vec![0, 3],
3151 right_keys: vec![3, 0],
3152 join_type: JoinType::Inner,
3153 };
3154 let body = RirNode::Project {
3155 input: Box::new(outer),
3156 columns: vec![
3157 ProjectExpr::Column(0),
3158 ProjectExpr::Column(1),
3159 ProjectExpr::Column(3),
3160 ProjectExpr::Column(5),
3161 ],
3162 };
3163 let mut builder = PlanBuilder::new();
3164 builder.add_scc(Scc {
3165 id: 0,
3166 predicates: vec!["cyc".to_string()],
3167 is_recursive: false,
3168 });
3169 builder.add_rule(
3170 0,
3171 CompiledRule {
3172 head: "cyc".to_string(),
3173 body,
3174 meta: Default::default(),
3175 },
3176 );
3177 builder.build()
3178 }
3179
3180 fn seed_4cycle_stats(c1: u64, c2: u64, c3: u64, c4: u64) -> StatsManager {
3181 let mut stats = StatsManager::new();
3182 for (rid, card) in [
3183 (RelId(1), c1),
3184 (RelId(2), c2),
3185 (RelId(3), c3),
3186 (RelId(4), c4),
3187 ] {
3188 stats.register_relation(rid);
3189 stats.update_cardinality(rid, card);
3190 }
3191 stats
3192 }
3193
3194 fn inspect_4cycle_grouping(
3198 plan: &xlog_ir::ExecutionPlan,
3199 ) -> Option<(RelId, RelId, RelId, RelId)> {
3200 let body = &plan.rules_by_scc.iter().flatten().next()?.body;
3201 let body = match body {
3202 xlog_ir::RirNode::MultiWayJoin { fallback, .. } => fallback.as_ref(),
3203 other => other,
3204 };
3205 let xlog_ir::RirNode::Project { input, .. } = body else {
3206 return None;
3207 };
3208 let xlog_ir::RirNode::Join { left, right, .. } = input.as_ref() else {
3209 return None;
3210 };
3211 let xlog_ir::RirNode::Join {
3212 left: ll,
3213 right: lr,
3214 ..
3215 } = left.as_ref()
3216 else {
3217 return None;
3218 };
3219 let xlog_ir::RirNode::Join {
3220 left: rl,
3221 right: rr,
3222 ..
3223 } = right.as_ref()
3224 else {
3225 return None;
3226 };
3227 let xlog_ir::RirNode::Scan { rel: r_ll } = ll.as_ref() else {
3228 return None;
3229 };
3230 let xlog_ir::RirNode::Scan { rel: r_lr } = lr.as_ref() else {
3231 return None;
3232 };
3233 let xlog_ir::RirNode::Scan { rel: r_rl } = rl.as_ref() else {
3234 return None;
3235 };
3236 let xlog_ir::RirNode::Scan { rel: r_rr } = rr.as_ref() else {
3237 return None;
3238 };
3239 Some((*r_ll, *r_lr, *r_rl, *r_rr))
3240 }
3241
3242 #[test]
3264 fn selectivity_pass_4cycle_picks_default_grouping_when_corners_smallest() {
3265 let mut plan = synth_4cycle_plan();
3266 let stats = seed_4cycle_stats(10, 10_000, 10_000, 10);
3267 selectivity_pass::run(&mut plan, &stats, &std::collections::HashMap::new());
3268 let (ll, lr, rl, rr) = inspect_4cycle_grouping(&plan).expect("grouping");
3269 assert_eq!(
3271 (ll, lr, rl, rr),
3272 (RelId(1), RelId(2), RelId(3), RelId(4)),
3273 "expected Default grouping"
3274 );
3275 }
3276
3277 #[test]
3290 fn selectivity_pass_4cycle_picks_alt_grouping_when_diagonals_smallest() {
3291 let mut plan = synth_4cycle_plan();
3292 let stats = seed_4cycle_stats(10_000, 10_000, 10, 10);
3293 selectivity_pass::run(&mut plan, &stats, &std::collections::HashMap::new());
3294 let (ll, lr, rl, rr) = inspect_4cycle_grouping(&plan).expect("grouping");
3295 assert_eq!(
3297 (ll, lr, rl, rr),
3298 (RelId(2), RelId(3), RelId(4), RelId(1)),
3299 "expected Alt grouping"
3300 );
3301 }
3302
3303 #[test]
3307 fn selectivity_pass_4cycle_two_snapshots_produce_different_groupings() {
3308 let mut plan_a = synth_4cycle_plan();
3309 let stats_a = seed_4cycle_stats(10, 10_000, 10_000, 10); selectivity_pass::run(&mut plan_a, &stats_a, &std::collections::HashMap::new());
3311 let g_a = inspect_4cycle_grouping(&plan_a).expect("grouping a");
3312
3313 let mut plan_b = synth_4cycle_plan();
3314 let stats_b = seed_4cycle_stats(10_000, 10_000, 10, 10); selectivity_pass::run(&mut plan_b, &stats_b, &std::collections::HashMap::new());
3316 let g_b = inspect_4cycle_grouping(&plan_b).expect("grouping b");
3317
3318 assert_ne!(
3319 g_a, g_b,
3320 "two different stats snapshots must produce different 4-cycle groupings; \
3321 got A = {:?}, B = {:?}",
3322 g_a, g_b
3323 );
3324 }
3325
3326 #[test]
3329 fn selectivity_pass_4cycle_skips_when_card_missing() {
3330 let mut plan = synth_4cycle_plan();
3331 let mut stats = StatsManager::new();
3333 for rid in [RelId(1), RelId(2), RelId(3)] {
3334 stats.register_relation(rid);
3335 stats.update_cardinality(rid, 100);
3336 }
3337 let before = format!("{:?}", plan.rules_by_scc[0][0].body);
3338 selectivity_pass::run(&mut plan, &stats, &std::collections::HashMap::new());
3339 let after = format!("{:?}", plan.rules_by_scc[0][0].body);
3340 assert_eq!(
3341 before, after,
3342 "missing-stats safety floor must leave body unchanged"
3343 );
3344 }
3345}
3346
3347#[cfg(test)]
3348mod tests {
3349 use super::*;
3350 use xlog_core::ScalarType;
3351 use xlog_ir::{ConstValue, ProjectExpr};
3352 use xlog_stats::ColumnStats;
3353
3354 fn make_stats_manager() -> Arc<StatsManager> {
3355 let mut mgr = StatsManager::new();
3356
3357 mgr.register_relation(RelId(1));
3359 mgr.update_cardinality(RelId(1), 10_000);
3360 mgr.update_byte_size(RelId(1), 320_000); mgr.register_relation(RelId(2));
3363 mgr.update_cardinality(RelId(2), 5_000);
3364 mgr.update_byte_size(RelId(2), 160_000);
3365
3366 mgr.register_relation(RelId(3));
3367 mgr.update_cardinality(RelId(3), 1_000);
3368 mgr.update_byte_size(RelId(3), 32_000);
3369
3370 let mut col0 = ColumnStats::new(0, ScalarType::I64);
3372 col0.update_distinct(1000);
3373 col0.update_range(0, 10000);
3374 mgr.add_column_stats(RelId(1), col0);
3375
3376 let mut col1 = ColumnStats::new(1, ScalarType::I64);
3377 col1.update_distinct(100);
3378 mgr.add_column_stats(RelId(1), col1);
3379
3380 Arc::new(mgr)
3381 }
3382
3383 #[test]
3384 fn test_optimizer_new() {
3385 let stats = make_stats_manager();
3386 let optimizer = Optimizer::new(stats);
3387
3388 assert_eq!(optimizer.config().dp_threshold, 10);
3389 assert!(optimizer.config().enable_pushdown);
3390 }
3391
3392 #[test]
3393 fn test_optimizer_with_config() {
3394 let stats = make_stats_manager();
3395 let config = OptimizerConfig {
3396 dp_threshold: 5,
3397 enable_pushdown: false,
3398 ..Default::default()
3399 };
3400 let optimizer = Optimizer::with_config(stats, config);
3401
3402 assert_eq!(optimizer.config().dp_threshold, 5);
3403 assert!(!optimizer.config().enable_pushdown);
3404 }
3405
3406 #[test]
3407 fn test_estimate_scan_cost() {
3408 let stats = make_stats_manager();
3409 let optimizer = Optimizer::new(stats);
3410
3411 let scan = RirNode::Scan { rel: RelId(1) };
3412 let cost = optimizer.estimate_cost(&scan);
3413
3414 assert_eq!(cost.rows, 10_000);
3415 assert!(cost.gpu_mem > 0);
3416 assert_eq!(cost.transfers, 0); }
3418
3419 #[test]
3420 fn test_estimate_scan_cost_unknown_relation() {
3421 let stats = Arc::new(StatsManager::new());
3422 let optimizer = Optimizer::new(stats);
3423
3424 let scan = RirNode::Scan { rel: RelId(999) };
3425 let cost = optimizer.estimate_cost(&scan);
3426
3427 assert_eq!(cost.rows, 1000);
3429 }
3430
3431 #[test]
3432 fn test_estimate_filter_cost() {
3433 let stats = make_stats_manager();
3434 let optimizer = Optimizer::new(stats);
3435
3436 let filter = RirNode::Filter {
3437 input: Box::new(RirNode::Scan { rel: RelId(1) }),
3438 predicate: Expr::Compare {
3439 left: Box::new(Expr::Column(0)),
3440 op: CompareOp::Eq,
3441 right: Box::new(Expr::Const(ConstValue::I64(42))),
3442 },
3443 };
3444
3445 let cost = optimizer.estimate_cost(&filter);
3446
3447 assert!(cost.rows < 10_000);
3449 assert!(cost.rows >= 1);
3450 }
3451
3452 #[test]
3453 fn test_estimate_join_cost() {
3454 let stats = make_stats_manager();
3455 let optimizer = Optimizer::new(stats);
3456
3457 let join = RirNode::Join {
3458 left: Box::new(RirNode::Scan { rel: RelId(1) }),
3459 right: Box::new(RirNode::Scan { rel: RelId(2) }),
3460 left_keys: vec![0],
3461 right_keys: vec![0],
3462 join_type: JoinType::Inner,
3463 };
3464
3465 let cost = optimizer.estimate_cost(&join);
3466
3467 assert!(cost.rows > 0);
3469 assert!(cost.cpu_cost > 0.0);
3470 assert!(cost.gpu_mem > 0);
3471 }
3472
3473 #[test]
3474 fn test_estimate_join_cost_with_selectivity() {
3475 let mut mgr = StatsManager::new();
3476 mgr.register_relation(RelId(1));
3477 mgr.register_relation(RelId(2));
3478 mgr.update_cardinality(RelId(1), 1000);
3479 mgr.update_cardinality(RelId(2), 500);
3480
3481 mgr.record_join_result(RelId(1), RelId(2), vec![0], vec![0], 500_000, 2500);
3483
3484 let optimizer = Optimizer::new(Arc::new(mgr));
3485
3486 let join = RirNode::Join {
3487 left: Box::new(RirNode::Scan { rel: RelId(1) }),
3488 right: Box::new(RirNode::Scan { rel: RelId(2) }),
3489 left_keys: vec![0],
3490 right_keys: vec![0],
3491 join_type: JoinType::Inner,
3492 };
3493
3494 let cost = optimizer.estimate_cost(&join);
3495
3496 assert!(cost.rows > 0);
3498 }
3499
3500 #[test]
3501 fn test_predicate_pushdown_simple_scan() {
3502 let stats = make_stats_manager();
3503 let optimizer = Optimizer::new(stats);
3504
3505 let scan = RirNode::Scan { rel: RelId(1) };
3506 let optimized = optimizer.optimize(scan);
3507
3508 assert!(matches!(optimized, RirNode::Scan { rel: RelId(1) }));
3510 }
3511
3512 #[test]
3513 fn test_predicate_pushdown_filter_on_scan() {
3514 let stats = make_stats_manager();
3515 let optimizer = Optimizer::new(stats);
3516
3517 let filter = RirNode::Filter {
3518 input: Box::new(RirNode::Scan { rel: RelId(1) }),
3519 predicate: Expr::Compare {
3520 left: Box::new(Expr::Column(0)),
3521 op: CompareOp::Eq,
3522 right: Box::new(Expr::Const(ConstValue::I64(42))),
3523 },
3524 };
3525
3526 let optimized = optimizer.optimize(filter);
3527
3528 assert!(matches!(optimized, RirNode::Filter { .. }));
3530 }
3531
3532 #[test]
3533 fn test_predicate_pushdown_merges_filters() {
3534 let stats = make_stats_manager();
3535 let optimizer = Optimizer::new(stats);
3536
3537 let nested_filter = RirNode::Filter {
3538 input: Box::new(RirNode::Filter {
3539 input: Box::new(RirNode::Scan { rel: RelId(1) }),
3540 predicate: Expr::Compare {
3541 left: Box::new(Expr::Column(0)),
3542 op: CompareOp::Gt,
3543 right: Box::new(Expr::Const(ConstValue::I64(0))),
3544 },
3545 }),
3546 predicate: Expr::Compare {
3547 left: Box::new(Expr::Column(0)),
3548 op: CompareOp::Lt,
3549 right: Box::new(Expr::Const(ConstValue::I64(100))),
3550 },
3551 };
3552
3553 let optimized = optimizer.optimize(nested_filter);
3554
3555 if let RirNode::Filter { predicate, .. } = optimized {
3557 assert!(matches!(predicate, Expr::And(_)));
3558 } else {
3559 panic!("Expected Filter node");
3560 }
3561 }
3562
3563 #[test]
3564 fn test_predicate_pushdown_through_project() {
3565 let stats = make_stats_manager();
3566 let optimizer = Optimizer::new(stats);
3567
3568 let plan = RirNode::Filter {
3570 input: Box::new(RirNode::Project {
3571 input: Box::new(RirNode::Scan { rel: RelId(1) }),
3572 columns: vec![ProjectExpr::Column(0), ProjectExpr::Column(1)],
3573 }),
3574 predicate: Expr::Compare {
3575 left: Box::new(Expr::Column(0)),
3576 op: CompareOp::Eq,
3577 right: Box::new(Expr::Const(ConstValue::I64(42))),
3578 },
3579 };
3580
3581 let optimized = optimizer.optimize(plan);
3582
3583 assert!(matches!(optimized, RirNode::Project { .. }));
3585 if let RirNode::Project { input, .. } = optimized {
3586 assert!(matches!(*input, RirNode::Filter { .. }));
3587 }
3588 }
3589
3590 #[test]
3591 fn test_predicate_pushdown_into_join() {
3592 let stats = make_stats_manager();
3593 let optimizer = Optimizer::new(stats);
3594
3595 let plan = RirNode::Filter {
3597 input: Box::new(RirNode::Join {
3598 left: Box::new(RirNode::Scan { rel: RelId(1) }),
3599 right: Box::new(RirNode::Scan { rel: RelId(2) }),
3600 left_keys: vec![0],
3601 right_keys: vec![0],
3602 join_type: JoinType::Inner,
3603 }),
3604 predicate: Expr::Compare {
3605 left: Box::new(Expr::Column(0)), op: CompareOp::Eq,
3607 right: Box::new(Expr::Const(ConstValue::I64(42))),
3608 },
3609 };
3610
3611 let optimized = optimizer.optimize(plan);
3612
3613 if let RirNode::Join { left, .. } = optimized {
3615 assert!(matches!(*left, RirNode::Filter { .. }));
3616 } else {
3617 panic!("Expected Join node");
3618 }
3619 }
3620
3621 #[test]
3622 fn test_plan_cost_total() {
3623 let cost = PlanCost {
3624 rows: 1000,
3625 cpu_cost: 100.0,
3626 gpu_mem: 1_000_000,
3627 transfers: 2,
3628 };
3629
3630 let total = cost.total_cost(100.0);
3631
3632 assert!((total - 1300.0).abs() < 0.001);
3635 }
3636
3637 #[test]
3638 fn test_plan_cost_then() {
3639 let cost1 = PlanCost {
3640 rows: 1000,
3641 cpu_cost: 50.0,
3642 gpu_mem: 500,
3643 transfers: 1,
3644 };
3645
3646 let cost2 = PlanCost {
3647 rows: 500,
3648 cpu_cost: 25.0,
3649 gpu_mem: 800,
3650 transfers: 1,
3651 };
3652
3653 let combined = cost1.then(cost2);
3654
3655 assert_eq!(combined.rows, 500); assert_eq!(combined.cpu_cost, 75.0);
3657 assert_eq!(combined.gpu_mem, 800); assert_eq!(combined.transfers, 2);
3659 }
3660
3661 #[test]
3662 fn test_optimizer_config_default() {
3663 let config = OptimizerConfig::default();
3664
3665 assert_eq!(config.dp_threshold, 10);
3666 assert!((config.index_heat_threshold - 0.7).abs() < 0.001);
3667 assert!(config.enable_pushdown);
3668 assert!((config.default_filter_selectivity - 0.1).abs() < 0.001);
3669 }
3670
3671 #[test]
3672 fn test_should_use_greedy() {
3673 let stats = make_stats_manager();
3674 let config = OptimizerConfig {
3675 dp_threshold: 2,
3676 ..Default::default()
3677 };
3678 let optimizer = Optimizer::with_config(stats, config);
3679
3680 let single = RirNode::Scan { rel: RelId(1) };
3682 assert!(!optimizer.should_use_greedy(&single));
3683
3684 let multi = RirNode::Join {
3686 left: Box::new(RirNode::Join {
3687 left: Box::new(RirNode::Scan { rel: RelId(1) }),
3688 right: Box::new(RirNode::Scan { rel: RelId(2) }),
3689 left_keys: vec![0],
3690 right_keys: vec![0],
3691 join_type: JoinType::Inner,
3692 }),
3693 right: Box::new(RirNode::Scan { rel: RelId(3) }),
3694 left_keys: vec![0],
3695 right_keys: vec![0],
3696 join_type: JoinType::Inner,
3697 };
3698 assert!(optimizer.should_use_greedy(&multi));
3699 }
3700
3701 #[test]
3702 fn test_recommend_indexes() {
3703 let mut mgr = StatsManager::new();
3704 mgr.register_relation(RelId(1));
3705 mgr.register_relation(RelId(2));
3706
3707 for _ in 0..50 {
3709 mgr.record_access(RelId(1));
3710 }
3711
3712 let optimizer = Optimizer::new(Arc::new(mgr));
3713 let recommendations = optimizer.recommend_indexes();
3714
3715 assert!(recommendations.contains(&RelId(1)));
3716 assert!(!recommendations.contains(&RelId(2)));
3717 }
3718
3719 #[test]
3720 fn test_estimate_groupby_cost() {
3721 let stats = make_stats_manager();
3722 let optimizer = Optimizer::new(stats);
3723
3724 let groupby = RirNode::GroupBy {
3725 input: Box::new(RirNode::Scan { rel: RelId(1) }),
3726 key_cols: vec![0],
3727 aggs: vec![(1, xlog_core::AggOp::Sum)],
3728 };
3729
3730 let cost = optimizer.estimate_cost(&groupby);
3731
3732 assert!(cost.rows < 10_000);
3734 assert!(cost.rows >= 1);
3735 }
3736
3737 #[test]
3738 fn test_estimate_union_cost() {
3739 let stats = make_stats_manager();
3740 let optimizer = Optimizer::new(stats);
3741
3742 let union = RirNode::Union {
3743 inputs: vec![
3744 RirNode::Scan { rel: RelId(1) },
3745 RirNode::Scan { rel: RelId(2) },
3746 ],
3747 };
3748
3749 let cost = optimizer.estimate_cost(&union);
3750
3751 assert_eq!(cost.rows, 15_000); }
3754
3755 #[test]
3756 fn test_estimate_distinct_cost() {
3757 let stats = make_stats_manager();
3758 let optimizer = Optimizer::new(stats);
3759
3760 let distinct = RirNode::Distinct {
3761 input: Box::new(RirNode::Scan { rel: RelId(1) }),
3762 key_cols: vec![0],
3763 };
3764
3765 let cost = optimizer.estimate_cost(&distinct);
3766
3767 assert!(cost.rows <= 10_000);
3769 assert!(cost.rows >= 1);
3770 }
3771
3772 #[test]
3773 fn test_estimate_diff_cost() {
3774 let stats = make_stats_manager();
3775 let optimizer = Optimizer::new(stats);
3776
3777 let diff = RirNode::Diff {
3778 left: Box::new(RirNode::Scan { rel: RelId(1) }),
3779 right: Box::new(RirNode::Scan { rel: RelId(2) }),
3780 };
3781
3782 let cost = optimizer.estimate_cost(&diff);
3783
3784 assert!(cost.rows <= 10_000);
3786 assert!(cost.rows >= 1);
3787 }
3788
3789 #[test]
3790 fn test_estimate_fixpoint_cost() {
3791 let stats = make_stats_manager();
3792 let optimizer = Optimizer::new(stats);
3793
3794 let fixpoint = RirNode::Fixpoint {
3795 scc_id: 0,
3796 base: Box::new(RirNode::Scan { rel: RelId(1) }),
3797 recursive: Box::new(RirNode::Scan { rel: RelId(1) }),
3798 delta_rel: RelId(10),
3799 full_rel: RelId(11),
3800 };
3801
3802 let cost = optimizer.estimate_cost(&fixpoint);
3803
3804 assert!(cost.rows >= 10_000);
3806 }
3807
3808 #[test]
3809 fn test_predicate_selectivity_equality() {
3810 let stats = make_stats_manager();
3811 let optimizer = Optimizer::new(stats);
3812
3813 let scan = RirNode::Scan { rel: RelId(1) };
3814
3815 let eq_pred = Expr::Compare {
3817 left: Box::new(Expr::Column(0)),
3818 op: CompareOp::Eq,
3819 right: Box::new(Expr::Const(ConstValue::I64(42))),
3820 };
3821
3822 let selectivity = optimizer.estimate_predicate_selectivity(&eq_pred, &scan);
3823
3824 assert!(selectivity < 0.01);
3826 assert!(selectivity > 0.0);
3827 }
3828
3829 #[test]
3830 fn test_predicate_selectivity_and() {
3831 let stats = make_stats_manager();
3832 let optimizer = Optimizer::new(stats);
3833
3834 let scan = RirNode::Scan { rel: RelId(1) };
3835
3836 let and_pred = Expr::And(vec![
3838 Expr::Compare {
3839 left: Box::new(Expr::Column(0)),
3840 op: CompareOp::Gt,
3841 right: Box::new(Expr::Const(ConstValue::I64(0))),
3842 },
3843 Expr::Compare {
3844 left: Box::new(Expr::Column(0)),
3845 op: CompareOp::Lt,
3846 right: Box::new(Expr::Const(ConstValue::I64(100))),
3847 },
3848 ]);
3849
3850 let selectivity = optimizer.estimate_predicate_selectivity(&and_pred, &scan);
3851
3852 assert!(selectivity < 0.5);
3854 assert!(selectivity > 0.0);
3855 }
3856
3857 #[test]
3858 fn test_predicate_selectivity_not() {
3859 let stats = make_stats_manager();
3860 let optimizer = Optimizer::new(stats);
3861
3862 let scan = RirNode::Scan { rel: RelId(1) };
3863
3864 let not_pred = Expr::Not(Box::new(Expr::Compare {
3866 left: Box::new(Expr::Column(0)),
3867 op: CompareOp::Eq,
3868 right: Box::new(Expr::Const(ConstValue::I64(42))),
3869 }));
3870
3871 let selectivity = optimizer.estimate_predicate_selectivity(¬_pred, &scan);
3872
3873 assert!(selectivity > 0.9);
3875 }
3876
3877 #[test]
3878 fn test_join_type_semi() {
3879 let stats = make_stats_manager();
3880 let optimizer = Optimizer::new(stats);
3881
3882 let semi_join = RirNode::Join {
3883 left: Box::new(RirNode::Scan { rel: RelId(1) }),
3884 right: Box::new(RirNode::Scan { rel: RelId(2) }),
3885 left_keys: vec![0],
3886 right_keys: vec![0],
3887 join_type: JoinType::Semi,
3888 };
3889
3890 let cost = optimizer.estimate_cost(&semi_join);
3891
3892 assert!(cost.rows <= 10_000);
3894 }
3895
3896 #[test]
3897 fn test_join_type_anti() {
3898 let stats = make_stats_manager();
3899 let optimizer = Optimizer::new(stats);
3900
3901 let anti_join = RirNode::Join {
3902 left: Box::new(RirNode::Scan { rel: RelId(1) }),
3903 right: Box::new(RirNode::Scan { rel: RelId(2) }),
3904 left_keys: vec![0],
3905 right_keys: vec![0],
3906 join_type: JoinType::Anti,
3907 };
3908
3909 let cost = optimizer.estimate_cost(&anti_join);
3910
3911 assert!(cost.rows <= 10_000);
3913 }
3914
3915 #[test]
3916 fn test_pushdown_disabled() {
3917 let stats = make_stats_manager();
3918 let config = OptimizerConfig {
3919 enable_pushdown: false,
3920 ..Default::default()
3921 };
3922 let optimizer = Optimizer::with_config(stats, config);
3923
3924 let plan = RirNode::Filter {
3926 input: Box::new(RirNode::Filter {
3927 input: Box::new(RirNode::Scan { rel: RelId(1) }),
3928 predicate: Expr::Compare {
3929 left: Box::new(Expr::Column(0)),
3930 op: CompareOp::Gt,
3931 right: Box::new(Expr::Const(ConstValue::I64(0))),
3932 },
3933 }),
3934 predicate: Expr::Compare {
3935 left: Box::new(Expr::Column(0)),
3936 op: CompareOp::Lt,
3937 right: Box::new(Expr::Const(ConstValue::I64(100))),
3938 },
3939 };
3940
3941 let optimized = optimizer.optimize(plan.clone());
3942
3943 if let RirNode::Filter { input, .. } = optimized {
3946 assert!(matches!(*input, RirNode::Filter { .. }));
3947 } else {
3948 panic!("Expected Filter node");
3949 }
3950 }
3951
3952 #[test]
3953 fn test_collect_columns() {
3954 let expr = Expr::And(vec![
3955 Expr::Compare {
3956 left: Box::new(Expr::Column(0)),
3957 op: CompareOp::Eq,
3958 right: Box::new(Expr::Column(2)),
3959 },
3960 Expr::Compare {
3961 left: Box::new(Expr::Column(1)),
3962 op: CompareOp::Gt,
3963 right: Box::new(Expr::Const(ConstValue::I64(0))),
3964 },
3965 ]);
3966
3967 let cols = Optimizer::collect_columns(&expr);
3968
3969 assert!(cols.contains(&0));
3970 assert!(cols.contains(&1));
3971 assert!(cols.contains(&2));
3972 }
3973
3974 #[test]
3975 fn test_flatten_and() {
3976 let nested = Expr::And(vec![
3977 Expr::And(vec![
3978 Expr::Compare {
3979 left: Box::new(Expr::Column(0)),
3980 op: CompareOp::Eq,
3981 right: Box::new(Expr::Const(ConstValue::I64(1))),
3982 },
3983 Expr::Compare {
3984 left: Box::new(Expr::Column(1)),
3985 op: CompareOp::Eq,
3986 right: Box::new(Expr::Const(ConstValue::I64(2))),
3987 },
3988 ]),
3989 Expr::Compare {
3990 left: Box::new(Expr::Column(2)),
3991 op: CompareOp::Eq,
3992 right: Box::new(Expr::Const(ConstValue::I64(3))),
3993 },
3994 ]);
3995
3996 let flattened = Optimizer::flatten_and(&nested);
3997
3998 assert_eq!(flattened.len(), 3);
3999 }
4000
4001 #[test]
4002 fn test_conjoin_single() {
4003 let single = vec![Expr::Compare {
4004 left: Box::new(Expr::Column(0)),
4005 op: CompareOp::Eq,
4006 right: Box::new(Expr::Const(ConstValue::I64(42))),
4007 }];
4008
4009 let result = Optimizer::conjoin(single);
4010
4011 assert!(matches!(result, Expr::Compare { .. }));
4012 }
4013
4014 #[test]
4015 fn test_conjoin_multiple() {
4016 let multiple = vec![
4017 Expr::Compare {
4018 left: Box::new(Expr::Column(0)),
4019 op: CompareOp::Eq,
4020 right: Box::new(Expr::Const(ConstValue::I64(1))),
4021 },
4022 Expr::Compare {
4023 left: Box::new(Expr::Column(1)),
4024 op: CompareOp::Eq,
4025 right: Box::new(Expr::Const(ConstValue::I64(2))),
4026 },
4027 ];
4028
4029 let result = Optimizer::conjoin(multiple);
4030
4031 assert!(matches!(result, Expr::And(_)));
4032 }
4033
4034 #[test]
4035 fn test_predicate_pushdown_with_schemas() {
4036 let stats = make_stats_manager();
4039 let mut optimizer = Optimizer::new(stats);
4040
4041 let left_schema = Schema::new(vec![
4043 ("c0".to_string(), xlog_core::ScalarType::Symbol),
4044 ("c1".to_string(), xlog_core::ScalarType::Symbol),
4045 ("c2".to_string(), xlog_core::ScalarType::Symbol),
4046 ]);
4047 let right_schema = Schema::new(vec![
4048 ("c0".to_string(), xlog_core::ScalarType::Symbol),
4049 ("c1".to_string(), xlog_core::ScalarType::Symbol),
4050 ("c2".to_string(), xlog_core::ScalarType::U32),
4051 ]);
4052
4053 let mut schemas = HashMap::new();
4054 schemas.insert(RelId(1), left_schema);
4055 schemas.insert(RelId(2), right_schema);
4056 optimizer.set_schemas(schemas);
4057
4058 let plan = RirNode::Filter {
4060 input: Box::new(RirNode::Join {
4061 left: Box::new(RirNode::Scan { rel: RelId(1) }),
4062 right: Box::new(RirNode::Scan { rel: RelId(2) }),
4063 left_keys: vec![0],
4064 right_keys: vec![0],
4065 join_type: JoinType::Inner,
4066 }),
4067 predicate: Expr::Compare {
4068 left: Box::new(Expr::Column(5)), op: CompareOp::Ge,
4070 right: Box::new(Expr::Const(ConstValue::U32(4))),
4071 },
4072 };
4073
4074 let optimized = optimizer.optimize(plan);
4075
4076 if let RirNode::Join { right, .. } = optimized {
4078 if let RirNode::Filter { predicate, .. } = *right {
4079 if let Expr::Compare { left, .. } = predicate {
4080 if let Expr::Column(idx) = *left {
4081 assert_eq!(
4082 idx, 2,
4083 "Column should be remapped to 2 (5 - left_width(3) = 2)"
4084 );
4085 } else {
4086 panic!("Expected Column expression");
4087 }
4088 } else {
4089 panic!("Expected Compare predicate");
4090 }
4091 } else {
4092 panic!("Expected Filter on right side of join");
4093 }
4094 } else {
4095 panic!("Expected Join node");
4096 }
4097 }
4098
4099 fn build_canonical_triangle_multiway() -> RirNode {
4115 let scan_xy = RirNode::Scan { rel: RelId(1) };
4116 let scan_yz = RirNode::Scan { rel: RelId(2) };
4117 let scan_xz = RirNode::Scan { rel: RelId(3) };
4118 let inner_join = RirNode::Join {
4119 left: Box::new(scan_xy.clone()),
4120 right: Box::new(scan_yz.clone()),
4121 left_keys: vec![1],
4122 right_keys: vec![0],
4123 join_type: JoinType::Inner,
4124 };
4125 let outer_join = RirNode::Join {
4126 left: Box::new(inner_join),
4127 right: Box::new(scan_xz.clone()),
4128 left_keys: vec![0, 3],
4129 right_keys: vec![0, 1],
4130 join_type: JoinType::Inner,
4131 };
4132 let fallback = RirNode::Project {
4133 input: Box::new(outer_join),
4134 columns: vec![
4135 ProjectExpr::Column(0),
4136 ProjectExpr::Column(1),
4137 ProjectExpr::Column(3),
4138 ],
4139 };
4140 RirNode::MultiWayJoin {
4141 inputs: vec![scan_xy, scan_yz, scan_xz],
4142 slot_vars: vec![
4143 vec![Some(0), Some(1)],
4144 vec![Some(1), Some(2)],
4145 vec![Some(0), Some(2)],
4146 ],
4147 output_columns: vec![
4148 ProjectExpr::Column(0),
4149 ProjectExpr::Column(1),
4150 ProjectExpr::Column(3),
4151 ],
4152 fallback: Box::new(fallback),
4153 plan: None,
4154 var_order: None,
4155 }
4156 }
4157
4158 fn build_4input_multiway() -> RirNode {
4167 let scans = [RelId(1), RelId(2), RelId(3), RelId(1)]
4168 .map(|rel| RirNode::Scan { rel })
4169 .to_vec();
4170 let slot_vars = vec![
4172 vec![Some(0u32), Some(1)],
4173 vec![Some(1u32), Some(2)],
4174 vec![Some(2u32), Some(3)],
4175 vec![Some(0u32), Some(3)],
4176 ];
4177 let output_columns = vec![
4180 ProjectExpr::Column(0),
4181 ProjectExpr::Column(1),
4182 ProjectExpr::Column(2),
4183 ProjectExpr::Column(3),
4184 ];
4185 let fallback = RirNode::Unit;
4188 RirNode::MultiWayJoin {
4189 inputs: scans,
4190 slot_vars,
4191 output_columns,
4192 fallback: Box::new(fallback),
4193 plan: None,
4194 var_order: None,
4195 }
4196 }
4197
4198 #[test]
4199 fn optimize_returns_multiway_unchanged() {
4200 let optimizer = Optimizer::new(make_stats_manager());
4201 for node in [build_canonical_triangle_multiway(), build_4input_multiway()] {
4202 let optimized = optimizer.optimize(node.clone());
4203 match (&node, &optimized) {
4204 (
4205 RirNode::MultiWayJoin {
4206 inputs: a_in,
4207 output_columns: a_out,
4208 ..
4209 },
4210 RirNode::MultiWayJoin {
4211 inputs: b_in,
4212 output_columns: b_out,
4213 ..
4214 },
4215 ) => {
4216 assert_eq!(a_in.len(), b_in.len());
4217 assert_eq!(a_out.len(), b_out.len());
4218 }
4219 _ => panic!("optimize() must return a MultiWayJoin"),
4220 }
4221 }
4222 }
4223
4224 #[test]
4225 fn estimate_width_uses_output_columns_arity() {
4226 let optimizer = Optimizer::new(make_stats_manager());
4227 assert_eq!(
4229 optimizer.estimate_width(&build_canonical_triangle_multiway()),
4230 3
4231 );
4232 assert_eq!(optimizer.estimate_width(&build_4input_multiway()), 4);
4236 }
4237
4238 #[test]
4239 fn estimate_cost_sums_input_costs() {
4240 let optimizer = Optimizer::new(make_stats_manager());
4241
4242 let cost_tri = optimizer.estimate_cost(&build_canonical_triangle_multiway());
4245 assert!(
4246 cost_tri.rows >= 16_000,
4247 "expected cost.rows >= 16000, got {}",
4248 cost_tri.rows
4249 );
4250
4251 let cost_4 = optimizer.estimate_cost(&build_4input_multiway());
4256 assert!(
4257 cost_4.rows >= 26_000,
4258 "expected 4-input cost.rows >= 26000, got {}",
4259 cost_4.rows
4260 );
4261 assert!(
4262 cost_4.rows > cost_tri.rows,
4263 "4-input cost ({}) must exceed triangle cost ({})",
4264 cost_4.rows,
4265 cost_tri.rows
4266 );
4267 }
4268
4269 #[test]
4270 fn find_column_relation_returns_none_for_multiway() {
4271 let optimizer = Optimizer::new(make_stats_manager());
4272 for node in [build_canonical_triangle_multiway(), build_4input_multiway()] {
4277 for col in 0..node.referenced_relations().len() {
4278 assert!(
4279 optimizer.find_column_relation(&node, col).is_none(),
4280 "find_column_relation must return None for any \
4281 MultiWayJoin column (col={})",
4282 col,
4283 );
4284 }
4285 }
4286 }
4287}