Method optimize

datafusion/optimizer/src/optimizer.rs:365–495

Optimizes the logical plan by applying optimizer rules, and invoking observer function after each call

(
        &self,
        plan: LogicalPlan,
        config: &dyn OptimizerConfig,
        mut observer: F,
    )

    /// Optimizes the logical plan by applying optimizer rules, and
    /// invoking observer function after each call
    pub fn optimize<F>(
        &self,
        plan: LogicalPlan,
        config: &dyn OptimizerConfig,
        mut observer: F,
    ) -> Result<LogicalPlan>
    where
        F: FnMut(&LogicalPlan, &dyn OptimizerRule),
    {
        // verify LP is valid, before the first LP optimizer pass.
        plan.check_invariants(InvariantLevel::Executable)
            .map_err(|e| e.context("Invalid input plan before LP Optimizers"))?;

        let start_time = Instant::now();
        let options = config.options();
        let mut new_plan = plan;

        let mut previous_plans = HashSet::with_capacity(16);
        previous_plans.insert(LogicalPlanSignature::new(&new_plan));

        let starting_schema = Arc::clone(new_plan.schema());

        let mut i = 0;
        while i < options.optimizer.max_passes {
            log_plan(&format!("Optimizer input (pass {i})"), &new_plan);

            for rule in &self.rules {
                // If skipping failed rules, copy plan before attempting to rewrite
                // as rewriting is destructive
                let prev_plan = options
                    .optimizer
                    .skip_failed_rules
                    .then(|| new_plan.clone());

                let starting_schema = Arc::clone(new_plan.schema());

                let result = match rule.apply_order() {
                    // optimizer handles recursion
                    Some(apply_order) => new_plan.rewrite_with_subqueries(
                        &mut Rewriter::new(apply_order, rule.as_ref(), config),
                    ),
                    // rule handles recursion itself
                    None => {
                        rule.rewrite(new_plan, config)
                    },
                }
                .and_then(|tnr| {
                    // run checks optimizer invariant checks, per optimizer rule applied
                    assert_valid_optimization(&tnr.data, &starting_schema)
                        .map_err(|e| e.context(format!("Check optimizer-specific invariants after optimizer rule: {}", rule.name())))?;

                    // run LP invariant checks only in debug mode for performance reasons
                    #[cfg(debug_assertions)]
                    tnr.data.check_invariants(InvariantLevel::Executable)
                        .map_err(|e| e.context(format!("Invalid (non-executable) plan after Optimizer rule: {}", rule.name())))?;

                    Ok(tnr)
                });