Skip to main content

xlog_logic/
optimizer.rs

1//! Query optimizer for join ordering and predicate pushdown.
2//!
3//! This module provides cost-based query optimization for XLOG's relational IR.
4//! It uses GPU-resident statistics from [`xlog_stats::StatsManager`] to make
5//! informed decisions about:
6//!
7//! - **Predicate pushdown**: Moving filter predicates closer to base scans to
8//!   reduce intermediate result sizes early in the pipeline.
9//! - **Cost estimation**: Computing expected row counts, CPU costs, GPU memory
10//!   usage, and data transfer counts for plan nodes.
11//! - **Join ordering**: (Future) Reordering joins based on selectivity estimates
12//!   to minimize intermediate result sizes.
13//!
14//! # Usage
15//!
16//! ```ignore
17//! use std::sync::Arc;
18//! use xlog_logic::optimizer::{Optimizer, OptimizerConfig, PlanCost};
19//! use xlog_stats::StatsManager;
20//!
21//! let stats = Arc::new(StatsManager::new());
22//! let optimizer = Optimizer::new(stats);
23//!
24//! // Optimize a query plan
25//! let optimized_plan = optimizer.optimize(original_plan);
26//!
27//! // Get cost estimates
28//! let cost = optimizer.estimate_cost(&optimized_plan);
29//! println!("Estimated rows: {}, GPU memory: {} bytes", cost.rows, cost.gpu_mem);
30//! ```
31
32use 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/// Configuration for query optimization.
39///
40/// Controls optimizer behavior including thresholds for algorithm selection
41/// and feature toggles.
42#[derive(Debug, Clone)]
43#[non_exhaustive]
44pub struct OptimizerConfig {
45    /// Maximum number of relations for exhaustive dynamic programming.
46    ///
47    /// When a query involves more relations than this threshold, the optimizer
48    /// switches to a greedy algorithm for join ordering to avoid exponential
49    /// time complexity. Default: 10 relations.
50    pub dp_threshold: usize,
51
52    /// Heat threshold for recommending index creation.
53    ///
54    /// Relations with access heat above this threshold are candidates for
55    /// index building to accelerate future queries. Default: 0.7.
56    pub index_heat_threshold: f32,
57
58    /// Enable predicate pushdown optimization.
59    ///
60    /// When enabled, filter predicates are pushed down through projections
61    /// and joins to be applied as early as possible. Default: true.
62    pub enable_pushdown: bool,
63
64    /// Default selectivity for filters when no statistics are available.
65    ///
66    /// Used as a fallback when column statistics cannot provide a better
67    /// estimate. Default: 0.1 (10% selectivity).
68    pub default_filter_selectivity: f64,
69
70    /// Cost multiplier for GPU-to-host data transfers.
71    ///
72    /// Transfers are expensive operations; this multiplier reflects the
73    /// relative cost compared to local GPU operations. Default: 100.0.
74    pub transfer_cost_multiplier: f64,
75
76    /// Bytes per row used for GPU memory estimation when schema is unknown.
77    ///
78    /// Default: 32 bytes (assumes 4 columns at 8 bytes each on average).
79    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/// Cost estimate for a query plan node.
96///
97/// Captures the multi-dimensional cost of executing a plan node, enabling
98/// the optimizer to make informed decisions based on available resources.
99#[derive(Debug, Clone, Default, PartialEq)]
100pub struct PlanCost {
101    /// Estimated number of output rows.
102    pub rows: u64,
103
104    /// Estimated CPU cost (arbitrary units, relative comparisons only).
105    ///
106    /// This represents processing overhead that cannot be parallelized on
107    /// the GPU, such as coordination, scheduling, and result materialization.
108    pub cpu_cost: f64,
109
110    /// Estimated GPU memory usage in bytes.
111    ///
112    /// Includes both input buffers and intermediate storage required for
113    /// the operation.
114    pub gpu_mem: u64,
115
116    /// Number of GPU-to-host or host-to-GPU data transfers.
117    ///
118    /// Transfers are typically the most expensive operations in GPU computing
119    /// and should be minimized.
120    pub transfers: u32,
121}
122
123impl PlanCost {
124    /// Creates a new cost estimate with specified row count.
125    pub fn with_rows(rows: u64) -> Self {
126        Self {
127            rows,
128            ..Default::default()
129        }
130    }
131
132    /// Computes a scalar cost value for comparison purposes.
133    ///
134    /// The formula weights different cost components:
135    /// - CPU cost is taken directly
136    /// - GPU memory is scaled by 0.001 (1GB = 1M cost units)
137    /// - Transfers are heavily weighted due to their high latency
138    ///
139    /// # Arguments
140    ///
141    /// * `transfer_weight` - Weight multiplier for transfer costs
142    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    /// Combines two costs representing sequential operations.
147    ///
148    /// Row count comes from the second (later) operation; other costs are summed.
149    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), // Peak memory usage
154            transfers: self.transfers + other.transfers,
155        }
156    }
157}
158
159/// Query optimizer using statistics for cost-based decisions.
160///
161/// The optimizer transforms query plans to improve execution efficiency
162/// by applying rewrites like predicate pushdown and using statistics to
163/// estimate costs for different plan alternatives.
164pub struct Optimizer {
165    stats: Arc<StatsManager>,
166    config: OptimizerConfig,
167    /// Schemas for relations, keyed by RelId
168    schemas: HashMap<RelId, Schema>,
169}
170
171impl Optimizer {
172    /// Creates a new optimizer with default configuration.
173    ///
174    /// # Arguments
175    ///
176    /// * `stats` - Shared statistics manager for cardinality and selectivity estimates
177    pub fn new(stats: Arc<StatsManager>) -> Self {
178        Self {
179            stats,
180            config: OptimizerConfig::default(),
181            schemas: HashMap::new(),
182        }
183    }
184
185    /// Creates a new optimizer with custom configuration.
186    ///
187    /// # Arguments
188    ///
189    /// * `stats` - Shared statistics manager
190    /// * `config` - Custom optimizer configuration
191    pub fn with_config(stats: Arc<StatsManager>, config: OptimizerConfig) -> Self {
192        Self {
193            stats,
194            config,
195            schemas: HashMap::new(),
196        }
197    }
198
199    /// Sets the schemas for relations.
200    ///
201    /// This information is used by the optimizer to accurately determine
202    /// column widths during predicate pushdown.
203    pub fn set_schemas(&mut self, schemas: HashMap<RelId, Schema>) {
204        self.schemas = schemas;
205    }
206
207    /// Returns a reference to the current configuration.
208    pub fn config(&self) -> &OptimizerConfig {
209        &self.config
210    }
211
212    /// Returns a reference to the statistics manager.
213    pub fn stats(&self) -> &Arc<StatsManager> {
214        &self.stats
215    }
216
217    /// Optimizes an execution plan by applying transformation rules.
218    ///
219    /// Currently applies:
220    /// - Predicate pushdown (if enabled)
221    ///
222    /// Future optimizations may include:
223    /// - Join reordering based on cardinality estimates
224    /// - Projection pushdown
225    /// - Common subexpression elimination
226    ///
227    /// # Arguments
228    ///
229    /// * `node` - The plan to optimize
230    ///
231    /// # Returns
232    ///
233    /// An optimized plan that is semantically equivalent to the input
234    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    /// Pushes filter predicates closer to scan nodes.
243    ///
244    /// This transformation reduces intermediate result sizes by applying
245    /// filters as early as possible in the query pipeline. The rules are:
246    ///
247    /// - Filters can be pushed through projections (with column remapping)
248    /// - Filters can be pushed into one or both sides of a join if the
249    ///   predicate references only columns from that side
250    /// - Filters on join keys can inform join selectivity estimates
251    ///
252    /// # Arguments
253    ///
254    /// * `node` - The plan node to transform
255    ///
256    /// # Returns
257    ///
258    /// The transformed plan with predicates pushed down where beneficial
259    fn predicate_pushdown(&self, node: RirNode) -> RirNode {
260        match node {
261            // Base case: scan nodes cannot be transformed further
262            RirNode::Unit => RirNode::Unit,
263            RirNode::Scan { rel } => RirNode::Scan { rel },
264
265            // Filter on top of another node: try to push down
266            RirNode::Filter { input, predicate } => {
267                // First, recursively optimize the input
268                let optimized_input = self.predicate_pushdown(*input);
269
270                match optimized_input {
271                    // Filter on Filter: merge predicates
272                    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                    // Filter on Project: push through if possible
284                    RirNode::Project {
285                        input: proj_input,
286                        columns,
287                    } => {
288                        // Check if predicate only references pass-through columns
289                        if let Some(remapped) =
290                            self.remap_predicate_through_project(&predicate, &columns)
291                        {
292                            // Push the remapped predicate below the projection
293                            RirNode::Project {
294                                input: Box::new(RirNode::Filter {
295                                    input: proj_input,
296                                    predicate: remapped,
297                                }),
298                                columns,
299                            }
300                        } else {
301                            // Cannot push: keep filter above
302                            RirNode::Filter {
303                                input: Box::new(RirNode::Project {
304                                    input: proj_input,
305                                    columns,
306                                }),
307                                predicate,
308                            }
309                        }
310                    }
311
312                    // Filter on Join: try to push to appropriate side
313                    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                        // Apply pushed predicates to each side
325                        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                        // Apply remaining predicates that couldn't be pushed
352                        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                    // Default: cannot push further
363                    other => RirNode::Filter {
364                        input: Box::new(other),
365                        predicate,
366                    },
367                }
368            }
369
370            // Project: unwrap consecutive projections before visiting the base.
371            // Function expansion can produce one projection per generated
372            // binding, so recursing through this unary chain can exhaust the
373            // native stack at the supported expansion-depth boundary.
374            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            // Join: recursively optimize both sides
393            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            // GroupBy: recursively optimize input
408            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            // Union: recursively optimize all inputs
419            RirNode::Union { inputs } => RirNode::Union {
420                inputs: inputs
421                    .into_iter()
422                    .map(|i| self.predicate_pushdown(i))
423                    .collect(),
424            },
425
426            // Distinct: recursively optimize input
427            RirNode::Distinct { input, key_cols } => RirNode::Distinct {
428                input: Box::new(self.predicate_pushdown(*input)),
429                key_cols,
430            },
431
432            // Diff: recursively optimize both sides
433            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            // Fixpoint: recursively optimize base and recursive parts
439            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, // Leaf-like: no pushdown
454
455            // Promoted physical-shape nodes are produced after the
456            // optimizer runs. Required for compile safety and as a
457            // no-op fallback if the call order ever changes.
458            RirNode::MultiWayJoin { .. } | RirNode::ChainJoin { .. } => node,
459        }
460    }
461
462    /// Attempts to remap a predicate through a projection.
463    ///
464    /// Returns `Some(remapped_predicate)` if all column references in the
465    /// predicate can be traced back through pass-through columns.
466    /// Returns `None` if the predicate references computed columns.
467    fn remap_predicate_through_project(
468        &self,
469        predicate: &Expr,
470        columns: &[xlog_ir::ProjectExpr],
471    ) -> Option<Expr> {
472        // Build a mapping from output column index to input column index
473        // Only for pass-through columns
474        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    /// Recursively remaps column references in an expression.
487    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            // Arithmetic operations
525            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            // Built-in functions
552            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    /// Estimates the output width (number of columns) of a plan node.
593    fn estimate_width(&self, node: &RirNode) -> usize {
594        match node {
595            RirNode::Unit => 0,
596            RirNode::Scan { rel } => {
597                // Use schema if available, otherwise stats, otherwise default
598                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 // Default assumption
604                }
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            // TensorMaskedJoin schemas are keyed by RelId.
620            // Use head_rel_id (not head_rel_name) for lookup.
621            RirNode::TensorMaskedJoin { head_rel_id, .. } => self
622                .schemas
623                .get(head_rel_id)
624                .map(|s| s.arity())
625                .unwrap_or(2),
626            // MultiWayJoin is produced after promotion; width equals the
627            // head projection arity, mirroring the Project arm.
628            RirNode::MultiWayJoin { output_columns, .. } => output_columns.len(),
629        }
630    }
631
632    /// Splits a predicate into parts pushable to left, right, or neither side of a join.
633    ///
634    /// Returns (left_predicates, right_predicates, remaining_predicates).
635    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        // Flatten AND expressions
645        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                // No columns referenced, can push to either side
654                left_preds.push(conj);
655            } else if max_col < left_width {
656                // All columns from left side
657                left_preds.push(conj);
658            } else if min_col >= left_width {
659                // All columns from right side - need to remap
660                let remapped = Self::remap_columns(&conj, |c| c - left_width);
661                right_preds.push(remapped);
662            } else {
663                // References both sides, cannot push
664                remaining.push(conj);
665            }
666        }
667
668        (left_preds, right_preds, remaining)
669    }
670
671    /// Flattens nested AND expressions into a list of conjuncts.
672    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    /// Collects all column indices referenced in an expression.
680    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    /// Remaps column references in an expression using a transformation function.
721    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    /// Combines a list of predicates into a single AND expression.
782    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    /// Estimates the cost of executing a plan node.
792    ///
793    /// Recursively computes cost estimates for the entire plan tree,
794    /// using statistics when available and falling back to heuristics.
795    ///
796    /// # Arguments
797    ///
798    /// * `node` - The plan node to estimate
799    ///
800    /// # Returns
801    ///
802    /// A [`PlanCost`] with estimated rows, CPU cost, GPU memory, and transfers
803    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            // MultiWayJoin heuristic cost is the sum of input scan costs.
901            // Post-promoter dispatch decides whether to run the WCOJ kernel
902            // or fall back; full multiway cost-model integration is separate
903            // planner work.
904            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    /// Estimates cost for a base relation scan.
919    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, // Minimal per-row CPU cost
924                gpu_mem: stats
925                    .byte_size
926                    .max(stats.cardinality * self.config.default_bytes_per_row),
927                transfers: 0, // Data already on GPU
928            }
929        } else {
930            // Default estimates for unknown relations
931            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    /// Estimates cost for a filter operation.
942    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, // Predicate eval cost
954            gpu_mem: input_cost.gpu_mem, // Filter reuses input memory
955            transfers: input_cost.transfers,
956        }
957    }
958
959    /// Estimates cost for a projection operation.
960    fn estimate_project_cost(
961        &self,
962        input_cost: PlanCost,
963        columns: &[xlog_ir::ProjectExpr],
964    ) -> PlanCost {
965        // Count computed vs pass-through columns
966        let computed_count = columns
967            .iter()
968            .filter(|c| matches!(c, xlog_ir::ProjectExpr::Computed(_, _)))
969            .count();
970
971        // Computed columns add CPU cost
972        let compute_cost = computed_count as f64 * input_cost.rows as f64 * 0.05;
973
974        // Output size may be smaller if fewer columns
975        let output_width_ratio = columns.len() as f64 / (columns.len() + 2) as f64; // Rough estimate
976
977        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    /// Estimates cost for a join operation.
986    #[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        // Semi and Anti joins always produce at most left_cost.rows
998        // Handle these specially before checking stats
999        let output_rows = match join_type {
1000            JoinType::Semi => {
1001                // At most left side rows, estimate 50% match
1002                ((left_cost.rows as f64 * 0.5) as u64).max(1)
1003            }
1004            JoinType::Anti => {
1005                // At most left side rows, estimate 50% don't match
1006                ((left_cost.rows as f64 * 0.5) as u64).max(1)
1007            }
1008            JoinType::Inner | JoinType::LeftOuter => {
1009                // Get relation IDs for selectivity lookup
1010                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                    // Simple join between two base relations
1015                    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                    // Multi-way or complex join: use heuristic
1028                    match join_type {
1029                        JoinType::Inner => {
1030                            // Assume 10% selectivity for inner joins
1031                            ((left_cost.rows as f64 * right_cost.rows as f64 * 0.1) as u64).max(1)
1032                        }
1033                        JoinType::LeftOuter => {
1034                            // At least left side rows
1035                            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        // Join CPU cost: hash build + probe
1047        let build_cost = right_cost.rows as f64 * 1.0; // Build hash table
1048        let probe_cost = left_cost.rows as f64 * 0.5; // Probe operations
1049        let cpu_cost = left_cost.cpu_cost + right_cost.cpu_cost + build_cost + probe_cost;
1050
1051        // GPU memory: both inputs plus hash table overhead
1052        let hash_table_overhead = right_cost.gpu_mem / 2; // Approximate hash table size
1053        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    /// Estimates cost for a group-by with aggregation.
1064    fn estimate_groupby_cost(
1065        &self,
1066        input_cost: PlanCost,
1067        key_cols: &[usize],
1068        _aggs: &[(usize, xlog_core::AggOp)],
1069    ) -> PlanCost {
1070        // Estimate distinct groups based on key columns
1071        // Heuristic: sqrt(input_rows) for unknown cardinality
1072        let estimated_groups = if key_cols.is_empty() {
1073            1 // No grouping = single result
1074        } else {
1075            // Rough estimate: assume good reduction
1076            ((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, // Aggregation cost
1082            gpu_mem: input_cost.gpu_mem + estimated_groups * self.config.default_bytes_per_row,
1083            transfers: input_cost.transfers,
1084        }
1085    }
1086
1087    /// Estimates cost for a union operation.
1088    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, // Concatenation cost
1097            gpu_mem: max_gpu,                               // Can process sequentially
1098            transfers: total_transfers,
1099        }
1100    }
1101
1102    /// Estimates cost for a distinct operation.
1103    fn estimate_distinct_cost(&self, input_cost: PlanCost, _key_cols: &[usize]) -> PlanCost {
1104        // Heuristic: distinct reduces rows by some factor
1105        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, // Hash-based dedup
1110            gpu_mem: input_cost.gpu_mem + input_cost.rows * 8,            // Hash set overhead
1111            transfers: input_cost.transfers,
1112        }
1113    }
1114
1115    /// Estimates cost for a set difference operation.
1116    fn estimate_diff_cost(&self, left_cost: PlanCost, right_cost: PlanCost) -> PlanCost {
1117        // Diff removes matching rows from left
1118        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    /// Estimates cost for a fixpoint (recursive) operation.
1129    fn estimate_fixpoint_cost(&self, base_cost: PlanCost, recursive_cost: PlanCost) -> PlanCost {
1130        // Fixpoint cost depends on number of iterations
1131        // Heuristic: assume log2(base_rows) iterations
1132        let estimated_iterations = ((base_cost.rows as f64).log2().ceil() as u64).max(1);
1133
1134        PlanCost {
1135            rows: base_cost.rows * estimated_iterations, // Output accumulates
1136            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, // Need delta and full
1138            transfers: base_cost.transfers + recursive_cost.transfers * estimated_iterations as u32,
1139        }
1140    }
1141
1142    /// Estimates selectivity of a predicate expression.
1143    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                // Multiply selectivities (independence assumption)
1150                exprs
1151                    .iter()
1152                    .map(|e| self.estimate_predicate_selectivity(e, input))
1153                    .product()
1154            }
1155            Expr::Or(exprs) => {
1156                // P(A or B) = P(A) + P(B) - P(A)P(B) for independent events
1157                // Simplified: max of selectivities as lower bound
1158                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    /// Estimates selectivity for a comparison predicate.
1169    fn estimate_compare_selectivity(
1170        &self,
1171        left: &Expr,
1172        op: CompareOp,
1173        right: &Expr,
1174        input: &RirNode,
1175    ) -> f64 {
1176        // Try to get column statistics if comparing column to constant
1177        if let (Expr::Column(col_idx), Expr::Const(_)) | (Expr::Const(_), Expr::Column(col_idx)) =
1178            (left, right)
1179        {
1180            // Find the base relation for this column
1181            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                                // Range predicates: estimate ~33% selectivity
1191                                0.33
1192                            }
1193                        };
1194                    }
1195                }
1196            }
1197        }
1198
1199        // Default selectivities by operator
1200        match op {
1201            CompareOp::Eq => 0.1, // 10% for equality
1202            CompareOp::Ne => 0.9, // 90% for inequality
1203            CompareOp::Lt | CompareOp::Le | CompareOp::Gt | CompareOp::Ge => 0.33, // 33% for ranges
1204        }
1205    }
1206
1207    /// Finds the base relation that provides a given column.
1208    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                // Trace column through projection
1214                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            // MultiWayJoin has no stable column-to-input mapping here.
1230            // The promoter runs after the optimizer, so this arm is
1231            // unreachable in production. A half-mapped implementation that
1232            // walked `inputs` via `slot_vars` would be more dangerous than
1233            // returning None for this optimizer fallback.
1234            RirNode::MultiWayJoin { .. } => None,
1235            _ => None, // Complex cases: give up
1236        }
1237    }
1238
1239    /// Returns relations that should have indexes built based on access heat.
1240    ///
1241    /// This is useful for adaptive query processing where frequently accessed
1242    /// relations benefit from index structures.
1243    pub fn recommend_indexes(&self) -> Vec<RelId> {
1244        self.stats.hot_relations(self.config.index_heat_threshold)
1245    }
1246
1247    /// Returns true if the query involves more relations than the DP threshold.
1248    ///
1249    /// Used to decide between exhaustive and greedy join ordering algorithms.
1250    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
1257// Selectivity-aware optimizer pass.
1258//
1259// No-op by default for unrecognized shapes. This pass owns the
1260// selectivity-driven join reordering hook; broader planner work may add
1261// reordering logic that consults `stats` for more shapes.
1262//
1263// Walks `plan.rules_by_scc[*].body` and rewrites nodes in place. The default
1264// no-op preserves every existing plan tree byte-for-byte. Tests assert
1265// structural equality for triangle, 4-cycle, and recursive-SCC plans.
1266//
1267// Compile-pipeline ordering: runs between `Optimizer::optimize` and
1268// `xlog_logic::promote::promote_multiway`.
1269pub mod selectivity_pass {
1270    //! Selectivity-driven join reordering for canonical lowered triangle and
1271    //! 4-cycle bodies.
1272    //!
1273    //! ## Behavior
1274    //!
1275    //! For each rule body that matches the canonical lowered
1276    //! triangle or 4-cycle shape, the pass enumerates the valid
1277    //! candidate inner pairings (3 for triangle, 2 for 4-cycle),
1278    //! computes each candidate's
1279    //! `StatsManager::estimate_join_cardinality` with
1280    //! **pair-derived join keys from the shared-variable
1281    //! mapping**, and rewrites the body so the smallest-cost
1282    //! choice is materialized first. Tie → keep the optimizer's
1283    //! existing order (deterministic no-op).
1284    //!
1285    //! ## Safety floor
1286    //!
1287    //! If any input atom for a recognized body has no
1288    //! `StatsManager` entry OR `cardinality == 0`, the body is
1289    //! left unchanged. Recursive deltas / freshly-uploaded
1290    //! relations / unseeded predicates therefore stay on the
1291    //! optimizer's default order until stats are populated.
1292    //!
1293    //! ## Default-fallback edge case
1294    //!
1295    //! `StatsManager::estimate_join_cardinality` returns `u64`
1296    //! with no provenance — the caller cannot tell whether the
1297    //! estimate came from the cached `JoinSelectivity` table,
1298    //! the column-distinct heuristic, or the 10% default
1299    //! fallback. When all input atoms have populated
1300    //! cardinalities but no column statistics, the per-pair
1301    //! estimates may all collapse to the same fallback ratio,
1302    //! making the chosen pairing uninformative. **This is an
1303    //! accepted trade-off**: row-set parity holds regardless of
1304    //! selectivity quality (the rewrite preserves semantics);
1305    //! the integration checks gate on row-set + WCOJ-dispatch
1306    //! correctness, not on optimal pair choice.
1307    //!
1308    //! ## Promoter coordination
1309    //!
1310    //! The triangle and 4-cycle promoters accept the canonical *semantic*
1311    //! shape with any valid key combination; they emit `MultiWayJoin.inputs`
1312    //! and `slot_vars` in canonical semantic order regardless of the body's
1313    //! positional layout. Reordered bodies therefore still promote and still
1314    //! dispatch the WCOJ kernel correctly.
1315    use std::collections::HashMap;
1316    use xlog_core::RelId;
1317    use xlog_ir::ExecutionPlan;
1318    use xlog_stats::StatsManager;
1319
1320    /// Selectivity-driven join reordering for canonical triangle and 4-cycle
1321    /// bodies. See module-level doc.
1322    ///
1323    /// `rel_ids` is the predicate-name → RelId map used to
1324    /// resolve body Scans against `StatsManager` lookups.
1325    /// Production callers pass `Compiler::lowerer().rel_ids()`.
1326    /// Test callers can pass an empty map; with no
1327    /// `StatsManager` entries either, the safety floor leaves
1328    /// every body unchanged (legacy no-op behavior preserved).
1329    pub fn run(plan: &mut ExecutionPlan, stats: &StatsManager, rel_ids: &HashMap<String, RelId>) {
1330        // `rel_ids` is reserved for future shape-extension
1331        // work; the current rewriters operate on RelIds
1332        // directly from the body's Scans, so the map isn't
1333        // consulted here. Production callers still pass it
1334        // so the API surface is forward-compatible.
1335        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
1350/// Ahead-of-time helper-relation splitting for deep joins with buried skew.
1351pub 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    /// Description of a helper relation introduced by the pass.
1362    #[derive(Debug, Clone, PartialEq, Eq)]
1363    pub struct HelperRelationSpec {
1364        /// Predicate name allocated for the helper relation.
1365        pub name: String,
1366        /// Relation identifier allocated for the helper relation.
1367        pub rel_id: RelId,
1368        /// Output schema of the helper relation.
1369        pub schema: Schema,
1370        /// Pair of source relations extracted into the helper body.
1371        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    /// Rewrite eligible rules in-place and return the helper relations introduced.
1417    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    /// K-clique helper split entry for K-clique plans that already carry
1453    /// planner-produced `HelperSplitSpec`s. The pass reuses the compiler-owned
1454    /// helper-relation lifecycle: emit a helper rule before the consumer rule,
1455    /// allocate a compiler-owned helper relation, and rewrite the consumer to
1456    /// scan that helper.
1457    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
2270/// Selectivity-driven body rewriters for triangle and 4-cycle canonical lowered
2271/// shapes. `pub(super)` so `selectivity_pass::run` can dispatch into them.
2272mod 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    // ---------------------------------------------------------
2314    // Triangle rewriter
2315    // ---------------------------------------------------------
2316
2317    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        // Skip when the candidate is structurally identical to
2557        // the input (no-op rewrite). RirNode doesn't impl
2558        // PartialEq, so compare via Debug — bodies are small
2559        // (≤ 6 Scans + 2 Joins + 1 Project) so the cost is
2560        // negligible relative to the optimizer's broader work.
2561        if format!("{:?}", candidate) == format!("{:?}", body) {
2562            return None;
2563        }
2564        Some(candidate)
2565    }
2566
2567    // ---------------------------------------------------------
2568    // 4-cycle rewriter
2569    // ---------------------------------------------------------
2570
2571    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    // ---------------------------------------------------------
2909    // Selectivity-driven reordering tests
2910    // ---------------------------------------------------------
2911
2912    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    /// Build a hand-crafted canonical lowered triangle plan
2918    /// with three Scans at RelId(1), RelId(2), RelId(3) for
2919    /// (e_xy, e_yz, e_xz). Bypasses the optimizer entirely so
2920    /// the reordering check is a clean stats-→-pair-choice
2921    /// observation, not a confounded test of optimizer plus the rewriter.
2922    ///
2923    /// Default canonical shape (Y-shared inner): inner keys
2924    /// `[1]/[0]`, outer keys `[0,3]/[0,1]`, project `[0,1,3]`.
2925    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    /// Seed a `StatsManager` with three triangle-edge
2966    /// cardinalities at the conventional RelIds (1, 2, 3) used
2967    /// by `synth_triangle_plan`.
2968    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    /// Inspect the (left RelId, right RelId) of the inner Join
2978    /// in a canonical lowered triangle body. Used by selectivity reordering
2979    /// checks.
2980    ///
2981    /// After `compile()` the body is a `MultiWayJoin` whose
2982    /// `fallback` field holds the post-selectivity-pass
2983    /// pre-promotion shape — that's where the inner-pair
2984    /// signature lives. The helper unwraps `MultiWayJoin →
2985    /// fallback` if needed before drilling into the binary
2986    /// Join structure.
2987    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    /// Snapshot 1: cards favor `(e1, e2)` Y-shared inner.
3017    /// Triangle rule: `tri(X, Y, Z) :- e1(X, Y), e2(Y, Z), e3(X, Z)`.
3018    /// To make Y-shared smallest, give e1 + e2 small cards and
3019    /// e3 a large card so all pair products are dominated by
3020    /// pairs containing e3 — except the pair (e1, e2) which
3021    /// is the smallest product.
3022    #[test]
3023    fn selectivity_pass_picks_y_shared_inner_when_e1_e2_smallest() {
3024        let mut plan = synth_triangle_plan();
3025        // e1=10, e2=10, e3=100_000 → Y-shared (e1⋈e2) smallest.
3026        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        // Y-shared inner = (e_xy, e_yz) = (RelId(1), RelId(2)).
3030        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    /// Snapshot 2: cards favor `(e1, e3)` X-shared inner.
3038    /// e1 + e3 small, e2 large.
3039    #[test]
3040    fn selectivity_pass_picks_x_shared_inner_when_e1_e3_smallest() {
3041        let mut plan = synth_triangle_plan();
3042        // e1=10, e2=100_000, e3=10 → X-shared (e1⋈e3) smallest.
3043        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        // X-shared inner = (e_xy, e_xz) = (RelId(1), RelId(3)).
3047        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    /// Snapshot 3: cards favor `(e2, e3)` Z-shared inner.
3055    /// e2 + e3 small, e1 large.
3056    #[test]
3057    fn selectivity_pass_picks_z_shared_inner_when_e2_e3_smallest() {
3058        let mut plan = synth_triangle_plan();
3059        // e1=100_000, e2=10, e3=10 → Z-shared (e2⋈e3) smallest.
3060        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        // Z-shared inner = (e_yz, e_xz) = (RelId(2), RelId(3)).
3064        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    /// Two snapshots produce different inner pairs. Pins
3072    /// "stats drive the order, not deterministic
3073    /// canonicalization." Deterministic canonicalization that
3074    /// ignores stats CANNOT pass this gate.
3075    #[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); // Y-shared
3079        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); // Z-shared
3084        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    /// Fallback edge case: relation cards present but no
3105    /// column statistics. The 10% default fallback inside
3106    /// `estimate_join_cardinality` means all three pair
3107    /// estimates collapse to roughly the same ratio. The pass
3108    /// either picks SOME pair or leaves the body unchanged;
3109    /// the test is tolerant by design and documents the
3110    /// uninformative-fallback case explicitly.
3111    #[test]
3112    fn selectivity_pass_with_only_relation_cards_may_pick_arbitrary_pair() {
3113        let mut plan = synth_triangle_plan();
3114        // All three cards equal — no column stats to break ties.
3115        let stats = seed_triangle_stats(100, 100, 100);
3116        selectivity_pass::run(&mut plan, &stats, &std::collections::HashMap::new());
3117        // Either a triangle inner pair is identifiable (any of
3118        // the three) or the body stays unchanged. Both are OK.
3119        let _ = inspect_triangle_inner_pair(&plan);
3120    }
3121
3122    // ---------------------------------------------------------
3123    // 4-cycle compile-time reordering tests
3124    // ---------------------------------------------------------
3125
3126    /// Build a hand-crafted canonical lowered 4-cycle plan
3127    /// with four Scans at RelId(1), RelId(2), RelId(3), RelId(4)
3128    /// for (e_wx, e_xy, e_yz, e_zw). Bypasses the optimizer.
3129    /// Default canonical bushy shape: inner-left
3130    /// `(e_wx ⋈ e_xy)` on X, inner-right `(e_yz ⋈ e_zw)` on Z,
3131    /// outer keys `[0, 3] / [3, 0]`, project `[0, 1, 3, 5]`.
3132    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    /// Recover the 4-cycle inner-grouping signature: `(left_left,
3195    /// left_right, right_left, right_right)` Scan RelIds. Used
3196    /// to identify which grouping the rewriter chose.
3197    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    /// 4-cycle: cards favor Default grouping
3243    /// `(e_wx⋈e_xy on X) + (e_yz⋈e_zw on Z)`. Default cost is
3244    /// `est(WX⋈XY)+est(YZ⋈ZW) = 0.1*c1*c2 + 0.1*c3*c4`.
3245    /// Alt cost is `0.1*c2*c3 + 0.1*c4*c1`. Default smaller
3246    /// when `c1*c2 + c3*c4 < c2*c3 + c4*c1`. With
3247    /// (c1=10, c2=10, c3=100_000, c4=100_000):
3248    ///   default = 100 + 10^10 ≈ 10^10.
3249    ///   alt = 10^6 + 10^6 ≈ 2*10^6.
3250    /// → alt is smaller, so this fixture actually favors Alt.
3251    /// Use (c1=10, c2=10, c3=10, c4=10_000_000) instead:
3252    ///   default = 100 + 10^8 = 10^8.
3253    ///   alt = 100 + 10^8 = 10^8 (same).
3254    /// Need uneven c4 vs others: (c1=10, c2=10, c3=10_000_000, c4=10):
3255    ///   default = 100 + 10^8 = 10^8.
3256    ///   alt = 10^8 + 100 = 10^8 (same).
3257    /// Default favored when c1*c2 << c2*c3 AND c3*c4 << c4*c1.
3258    /// I.e., c1 small and c4 small relative to c2 and c3.
3259    /// (c1=10, c2=10_000, c3=10_000, c4=10):
3260    ///   default = 0.1*100_000 + 0.1*100_000 = 20_000.
3261    ///   alt = 0.1*100_000_000 + 0.1*100 = 10_000_010.
3262    /// → Default smaller. ✓
3263    #[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        // Default: (e_wx, e_xy, e_yz, e_zw) = (RelId(1..4)).
3270        assert_eq!(
3271            (ll, lr, rl, rr),
3272            (RelId(1), RelId(2), RelId(3), RelId(4)),
3273            "expected Default grouping"
3274        );
3275    }
3276
3277    /// 4-cycle: cards favor Alt grouping
3278    /// `(e_xy⋈e_yz on Y) + (e_zw⋈e_wx on W)`. Alt smaller when
3279    /// `c2*c3 + c4*c1 < c1*c2 + c3*c4`. Use
3280    /// (c1=10_000, c2=10, c3=10, c4=10_000):
3281    ///   default = 0.1*100_000 + 0.1*100_000 = 20_000.
3282    ///   alt = 0.1*100 + 0.1*10^8 = 10_000_010.
3283    /// → Default still wins. Need c1*c2 LARGE and c3*c4 LARGE
3284    /// while c2*c3 SMALL and c4*c1 SMALL. Try
3285    /// (c1=10_000, c2=10_000, c3=10, c4=10):
3286    ///   default = 0.1*10^8 + 0.1*100 = 10_000_010.
3287    ///   alt = 0.1*100_000 + 0.1*100_000 = 20_000.
3288    /// → Alt smaller. ✓
3289    #[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        // Alt: (e_xy, e_yz, e_zw, e_wx) = (RelId(2), RelId(3), RelId(4), RelId(1)).
3296        assert_eq!(
3297            (ll, lr, rl, rr),
3298            (RelId(2), RelId(3), RelId(4), RelId(1)),
3299            "expected Alt grouping"
3300        );
3301    }
3302
3303    /// Same plan, two stats snapshots → two different
3304    /// 4-cycle groupings. Pins "stats drive the choice" for
3305    /// 4-cycle.
3306    #[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); // Default.
3310        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); // Alt.
3315        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    /// 4-cycle missing-stats safety floor: any unseeded
3327    /// relation → body unchanged.
3328    #[test]
3329    fn selectivity_pass_4cycle_skips_when_card_missing() {
3330        let mut plan = synth_4cycle_plan();
3331        // Only seed 3 of 4.
3332        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        // Register test relations with realistic statistics
3358        mgr.register_relation(RelId(1));
3359        mgr.update_cardinality(RelId(1), 10_000);
3360        mgr.update_byte_size(RelId(1), 320_000); // ~32 bytes per row
3361
3362        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        // Add column statistics for relation 1
3371        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); // Data on GPU
3417    }
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        // Should use defaults
3428        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        // Filter should reduce row count
3448        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        // Should have positive estimates
3468        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        // Record a join result to cache selectivity
3482        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        // Should use cached selectivity for estimate
3497        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        // Scan should pass through unchanged
3509        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        // Filter on scan should stay in place
3529        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        // Filters should be merged into AND
3556        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        // Filter on projected column that's a pass-through
3569        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        // Filter should be pushed below project
3584        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        // Filter on left side column only
3596        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)), // Left side column
3606                op: CompareOp::Eq,
3607                right: Box::new(Expr::Const(ConstValue::I64(42))),
3608            },
3609        };
3610
3611        let optimized = optimizer.optimize(plan);
3612
3613        // Filter should be pushed into left side of join
3614        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        // cpu_cost + gpu_mem*0.001 + transfers*100
3633        // 100.0 + 1000.0 + 200.0 = 1300.0
3634        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); // Takes output rows from second
3656        assert_eq!(combined.cpu_cost, 75.0);
3657        assert_eq!(combined.gpu_mem, 800); // Peak memory
3658        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        // Single relation: should NOT use greedy
3681        let single = RirNode::Scan { rel: RelId(1) };
3682        assert!(!optimizer.should_use_greedy(&single));
3683
3684        // Three relations: should use greedy (threshold is 2)
3685        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        // Heat up relation 1 extensively
3708        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        // GroupBy should reduce row count
3733        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        // Union sums row counts
3752        assert_eq!(cost.rows, 15_000); // 10000 + 5000
3753    }
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        // Distinct reduces rows
3768        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        // Diff reduces left side
3785        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        // Fixpoint accumulates rows across iterations
3805        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        // Equality predicate
3816        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        // With 1000 distinct values, selectivity should be ~0.001
3825        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        // AND of two predicates
3837        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        // Product of individual selectivities (0.33 * 0.33 ≈ 0.11)
3853        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        // NOT of equality
3865        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(&not_pred, &scan);
3872
3873        // NOT(equality) should have high selectivity
3874        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        // Semi join outputs at most left side rows
3893        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        // Anti join outputs at most left side rows
3912        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        // Filter that could be pushed
3925        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        // With pushdown disabled, structure should remain the same
3944        // (outer filter, inner filter, scan)
3945        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        // Regression test: ensure predicate pushdown uses schemas for accurate width estimation.
4037        // Without schemas, the optimizer could incorrectly remap column indices.
4038        let stats = make_stats_manager();
4039        let mut optimizer = Optimizer::new(stats);
4040
4041        // Set up schemas: left has 3 columns, right has 3 columns
4042        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        // Filter on Column(5) which is in the right side (left_width=3, so column 5-3=2 in right)
4059        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)), // Right side column (index 5 = 3 + 2)
4069                op: CompareOp::Ge,
4070                right: Box::new(Expr::Const(ConstValue::U32(4))),
4071            },
4072        };
4073
4074        let optimized = optimizer.optimize(plan);
4075
4076        // Filter should be pushed into right side of join with Column(2) (remapped from 5-3=2)
4077        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    /// Optimizer fallback arms for `MultiWayJoin`.
4100    ///
4101    /// The promoter runs after `Optimizer::optimize` in `Compiler`, so
4102    /// these arms are unreachable in production. They exist for compile
4103    /// safety and to pin the documented semantics: `optimize` returns
4104    /// the node unchanged, `estimate_width` reports the head arity from
4105    /// `output_columns`, `estimate_cost` is the sum of input costs, and
4106    /// `find_column_relation` returns `None` under the optimizer fallback.
4107    ///
4108    /// Shape-agnostic coverage extends each test below to also exercise a
4109    /// synthesized four-input `MultiWayJoin` via [`build_4input_multiway`].
4110    /// This pins shape-agnosticism: the arms must NOT hard-code
4111    /// `inputs.len() == 3` or `output_columns.len() == 3`. The four-way
4112    /// promoter path produces real four-input bodies; these tests are the
4113    /// load-bearing guard against silent regression.
4114    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    /// Synthesized four-input `MultiWayJoin` for shape-agnosticism testing.
4159    /// The original promoter shape is triangle-only, so this shape never
4160    /// reaches `Optimizer` through the production pipeline; the tests below
4161    /// exercise the optimizer arms directly.
4162    ///
4163    /// Inputs reuse `RelId(1, 2, 3, 1)` — RelId(1) repeats — so the
4164    /// stats manager registered in `make_stats_manager` covers all
4165    /// four scans. Cost floor is `2*10_000 + 5_000 + 1_000 = 26_000`.
4166    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        // 4-cycle slot_vars [[A,B],[B,C],[C,D],[A,D]].
4171        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        // 4-arity head projection (no real semantic meaning — the
4178        // synthesized fallback is a stub).
4179        let output_columns = vec![
4180            ProjectExpr::Column(0),
4181            ProjectExpr::Column(1),
4182            ProjectExpr::Column(2),
4183            ProjectExpr::Column(3),
4184        ];
4185        // Stub fallback: the optimizer arms do not execute fallback,
4186        // so any RirNode is fine. Use Unit to keep the fixture small.
4187        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        // Canonical triangle: 3 head columns.
4228        assert_eq!(
4229            optimizer.estimate_width(&build_canonical_triangle_multiway()),
4230            3
4231        );
4232        // 4-input synthesized: 4 head columns. Locks shape-
4233        // agnosticism — the arm must use output_columns.len(),
4234        // not a hard-coded 3.
4235        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        // Canonical triangle: rels 1, 2, 3 with cardinalities
4243        // 10_000 + 5_000 + 1_000 = 16_000.
4244        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        // 4-input synthesized: rels 1, 2, 3, 1 → 2*10_000 + 5_000 +
4252        // 1_000 = 26_000. The arm sums all four inputs; cost grows.
4253        // Locks shape-agnosticism — the arm must walk every entry
4254        // in `inputs`, not a hard-coded 3.
4255        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        // Optimizer fallback guardrail: no column-to-input mapping is exposed
4273        // here. Half-mapped is more dangerous than None. The arm must return
4274        // None regardless of arity; the synthesized four-input shape catches a
4275        // future "let's just return inputs[col_idx % len]" patch.
4276        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}