Remove Arc<LogicalPlan> from LogicalPlan, stop copying LogicalPlans

[tustvold]

Is your feature request related to a problem or challenge? Please describe what you are trying to do.

Related to #4627, the current representation of LogicalPlan contains Arc<LogicalPlan> at various points, whilst this does reduce the cost of copying a LogicalPlan tree, it:

• Complicates rewrites by necessitating clones
• Results in double-boxing - e.g. Vec<Arc<LogicalPlan>>
• Permits cycles and all the excitement that would entail
• Marginal overhead from additional atomics
• Unidiomatic is perhaps too strong, but it is strange for a tree datastructure to have shared ownership

Describe the solution you'd like

I would like to remove the Arc, replacing with Box where necessary. Methods that currently take Arc<LogicalPlan> should be updated to take LogicalPlan.

Describe alternatives you've considered

Additional context

This likely wants to wait until we are cloning LogicalPlan less frequently

[alamb]

I believe @mingmwang and @jackwener have noted this in the past -- the idea is good to me

[jackwener]

Hope to this change😍, it is very meaningful to optimizer.

[jackwener]

If we can remove Arc<LogicalPlan> and use LogicalPlan, wen can use pattern-match to match subtree pattern like:

match projection-filter

LogicalPlan::Projection(Projection{LogicalPlan::Filter, ..})

like this:

[mslapek]

How would OptimizerRule look after the change?

Currently we have:

/// Try and rewrite `plan` to an optimized form, returning None if the plan cannot be
/// optimized by this rule.
fn try_optimize(
    &self,
    plan: &LogicalPlan,
    config: &dyn OptimizerConfig,
) -> Result<Option<LogicalPlan>>;

With the current try_optimize signature, lack of Arc<LogicalPlan> would require to clone whole non-optimised subbranches.

[alamb]

How would OptimizerRule look after the change?

Maybe we could change the signature to something like

enum OptimizedPlan {
  // Optimizer did not make any changes to the original input pla
  NoChange(LogicalPlan),
  /// Optimizer rewrote the original plan
  Rewritten(LogicalPlan),
}


/// Try and rewrite `plan` to an optimized form
fn try_optimize(
    &self,
    plan: &LogicalPlan,
    config: &dyn OptimizerConfig,
) -> Result<OptimizedPlan>;

[tustvold]

I think you would want to make try_optimize take ownership not a borrow, and then return it

[alamb]

I think you would want to make try_optimize take ownership not a borrow, and then return it

I think there are cases (like deciding when a fixed point is reached) where the caller wants to distinguish between no more optimization and a new plan.

However, now that LogicalPlan supports PartialEq I think maybe we could just use that to check if any changes were made

https://github.com/apache/arrow-datafusion/blob/c37ddf72ec539bd39cce0dd4ff38db2e36ddb55f/datafusion/expr/src/logical_plan/plan.rs#L52

So the signature maybe could be

/// Try and rewrite `plan` to an optimized form
fn try_optimize(
    &self,
    plan: LogicalPlan,
    config: &dyn OptimizerConfig,
) -> Result<LogicalPlan>;

🤔

[mslapek]

To compare a new logical plan with the old one, we need to have both plans in the memory.

Without Arc<...>, it requires to have always doubled RAM usage for plans. With Arc<...> a slightly less? 🤔

[tustvold]

However, now that LogicalPlan supports PartialEq I think maybe we could just use that to check if any changes were made

This is not a particularly cheap operation, involving a lot of string comparisons. I think having the return value indicate if changes have been made as in your original example makes sense to me.

My point was if we remove the Arc we need to be careful to move LogicalPlan and avoid cloning, as any clone is then a deep clone of the entire tree. We therefore need to pass in an owned value so that it can be moved into the return type

[mslapek]

I think having the return value indicate if changes have been made as in your original example makes sense to me.

At the first sight the idea seems to be compelling...

Nevertheless I'm quite pessimistic about the idea. 😵‍💫

The main reason are optimisations, which UNDO other optimisations!

(sorry for the long post! but it's core architecture stuff...)

Examples of cancelling optimisations

Example 1. Commutative optimisations

Some operations are commutative, like projection or limit - it is used in both PushDownProjection and PushDownLimit - but in a different directions (grep commute in these files).

So for some plans they will always give Some(new_plan). 😕

Example 2. Undoing inside of an optimisation

Inside PushDownLimit we have:

1. Limit A -> Projection -> Limit B
2. Added a limit after the projection: Limit A -> Projection -> Limit C -> Limit B
3. Merge of the Limit C and Limit B in ANOTHER invocation of try_optimize: Limit A -> Projection -> Limit B

We have many invocations of try_optimize due to apply_order option.

The point is that for a fixed point (merge of Limit C + Limit B is Limit B) PushDownLimit will always yield Some(...).

Theoretical pain

Let's look at the issue formally... 😎

Consider small changes Δ_1, Δ_2, ... Δ_n - some signed integers.

The total change is:

Δ_all = Δ_1 + Δ_2 + ... + Δ_n

So Δ_all ≠ 0   implies   Δ_i ≠ 0 for some i.

The premise of Option<Plan> is that the implication holds in the another direction - which is NOT the case!

Just give Δ_1 = 1, Δ_2 = -1, Δ_3 = 0... (Δ_2 cancels changes from Δ_1!)

Alternatives?

Let's look again at the implication - how can we use it?

Δ_all ≠ 0   implies   Δ_i ≠ 0 for some i.

I would consider to use some stats about the tree - as a kind of heuristics. For example: number of nodes in the tree.

Each optimization could return (LogicalPlan, Δ of number of nodes). If sum of all Δ of number of nodes is nonzero - then we know the tree has changed.

I guess there could be other useful stats, like ∑ (node_type * node depth).