Wrap immutable plan parts into Arc

[askalt]

Which issue does this PR close?

• Closes Wrap more immutable plan parts into Arc #19852

Rationale for this change

Improve performance of query planning and plan state re-set by making node clone cheap.

What changes are included in this PR?

• Store projection as Option<Arc<[usize]>> instead of Option<Vec<usize>> in FilterExec, HashJoinExec, NestedLoopJoinExec.
• Store exprs as Arc<[ProjectionExpr]> instead of Vec in ProjectionExprs.
• Store arced aggregation, filter, group by expressions within AggregateExec.

[askalt]

run benchmark reset_plan_states

[alamb-ghbot]

🤖 Hi @askalt, thanks for the request (#19893 (comment)). scrape_comments.py only responds to whitelisted users. Allowed users: Dandandan, Omega359, adriangb, alamb, comphead, gabotechs, geoffreyclaude, klion26, rluvaton, xudong963, zhuqi-lucas.

[askalt]

@xudong963 could you please run benchmark reset_plan_states?

[xudong963]

run benchmark reset_plan_states

[alamb-ghbot]

🤖 ./gh_compare_branch_bench.sh compare_branch_bench.sh Running
Linux aal-dev 6.14.0-1018-gcp #19~24.04.1-Ubuntu SMP Wed Sep 24 23:23:09 UTC 2025 x86_64 x86_64 x86_64 GNU/Linux
Comparing askalt/arc_immutable (bb97763) to d90d074 diff
BENCH_NAME=reset_plan_states
BENCH_COMMAND=cargo bench --features=parquet --bench reset_plan_states
BENCH_FILTER=
BENCH_BRANCH_NAME=askalt_arc_immutable
Results will be posted here when complete

[alamb-ghbot]

🤖: Benchmark completed

Details

group     askalt_arc_immutable                   main
-----     --------------------                   ----
query1    1.00     32.2±1.09ms        ? ?/sec    1.01     32.4±0.31ms        ? ?/sec
query2    1.00      9.2±0.08ms        ? ?/sec    1.03      9.5±0.11ms        ? ?/sec
query3    1.00     12.2±0.25ms        ? ?/sec    1.15     14.1±0.08ms        ? ?/sec

[crepererum]

I understand that we wanna avoid breaking changes and hence are kinda liberal with impl Into<...>/impl AsRef<...> parameters. However these methods are generic and will be compiled on the call-site, i.e. whoever calls these methods has to recompile them. Furthermore calling these methods with different types duplicates these methods in LLVM and likely also in the resulting binary. This means: longer compile times and binary size bloat.

Hence I would favor if we could avoid these impl parameters.

[askalt]

I understand that we wanna avoid breaking changes and hence are kinda liberal with impl Into<...>/impl AsRef<...> parameters. However these methods are generic and will be compiled on the call-site, i.e. whoever calls these methods has to recompile them. Furthermore calling these methods with different types duplicates these methods in LLVM and likely also in the resulting binary. This means: longer compile times and binary size bloat.

Hence I would favor if we could avoid these impl parameters.

Actually, I also think that using a generic here isn't justified, since &Vec<u8> can always be converted to a slice. I believe the API should be as simple as possible. Let's get @alamb opinion.

[alamb]

I understand that we wanna avoid breaking changes and hence are kinda liberal with impl Into<...>/impl AsRef<...> parameters. However these methods are generic and will be compiled on the call-site, i.e. whoever calls these methods has to recompile them. Furthermore calling these methods with different types duplicates these methods in LLVM and likely also in the resulting binary. This means: longer compile times and binary size bloat.
Hence I would favor if we could avoid these impl parameters.

Actually, I also think that using a generic here isn't justified, since &Vec<u8> can always be converted to a slice. I believe the API should be as simple as possible. Let's get @alamb opinion.

Indeed my concern is the impact on (all) downstream users during the upgrade. I believe the initial PR from @askalt was substantially larger because it required changing a bunch of callsites.

If we can avoid the use of impl and not require many changes (I understand some may be impossible to avoid) that sounds good to me

[alamb]

Thank you for this PR @askalt and for the review @crepererum and @martin-g

I think the performance results are quite compelling and this is quite close.

I agree it would be find to avoid the AsRef impls, if you are willing to do so @askalt . I also think it would be fine to leave this as is.

In any event, I think we should add an example to project_schema that shows how to call it with Option<&Vec<usize>> (what is passed by TableProvider::scan)

Before we merge this PR I think we

1. should also add a note to the upgrade guide with a note about this change (I can help with this, but I didn't want to push changes to this PR before we have consensus approach)
2. Clean up the API for OptionProjectionRef (move it, and make Option<ProjectionRef>)

[alamb]

Per @crepererum 's suggestion, I tried out what it would look like and came up with

• Change project_schema to use Option<&[usize]> askalt/datafusion#3

It does seem to be reasonable

The original signature is Option<&Vec<..>> I think to align with TableProvider::scan (which also shouldn't have a owned Vec, but that I think is a historic accident)

[alamb]

likewise here I think it would be good to note they are Arc to make clone/plan rewriting faster

[askalt]

Done.

[alamb]

I see this is basically refactored out of NestedLoops join (and maybe elsewhere)

[alamb]

I see why you need something like this (as it changes the signture of TableProvider) but this seems inconsistent with the other representations of projections in DataFusion/Arrow I am familiar with, which are represented with an Option rather than having the Option internally

I recommend:

1. Move this into datafusion/physical-expr/src/projection.rs (not in physical plan) so it is near ProjectionExprs
2. Move the Option outside (so the signature is Option<ProjectionRef> rather than OptionProjectionRef)

[alamb]

I also recommend documenting why this structure exists (because it is cheap to clone)

Something like

/// This structure represents projecting a set of columns by index.
/// It uses an `Arc` internally to make it cheap to clone

[askalt]

1. Move this into datafusion/physical-expr/src/projection.rs (not in physical plan) so it is near ProjectionExprs

Done

I also recommend documenting why this structure exists (because it is cheap to clone)

Done

2. Move the Option outside (so the signature is Option rather than OptionProjectionRef)

Initially I done in this way, but in this case, take a look at possible HashJoinExec::try_new signature. It seems we want something like

fn try_new(
    ...
   projection: Option<impl Into<ProjectionRef>>,
   ...
)

To be able to pass Some(vec![1,2,3]) and Some(projection_ref) as well. But Option<impl ...> forces us to explicitly specify None type, like None::<ProjectionRef> which looks annoying for me. What do you think?

[alamb]

I still think we should try it -- #19893 (review)

Specotying the None type does look annoying -- maybe we could make a version of the function that takes no projecton -- like try_new_without_projection 🤔

[crepererum]

A slightly more verbose trick to avoid the code bloat but still get the impl Into<...>/impl AsRef<...> would be to split the respective methods into a public and private part. Using an impl Into<...> example, then something like:

pub fn foo(&self, param: String) -> ... {
   ...
}

would become

pub fn foo(&self, param: impl Into<String>) -> ... {
    self.foo_impl(param.into())
}

fn foo_impl(&self, param: String) -> ... {
    // that's just the "old" `foo` method
    ...
}

i.e. you basically control the generic part a bit more manual. That was what once was planned (but no longer is) as an automatic step, see rust-lang/rust#124962

[askalt]

I decided to simplify Statistics::project and project_schema signature and take an Option slice. Added a note to upgrading.md. Anyway, @crepererum thank you for the mentioned trick!

[alamb]

I am sorry this is taking so long, but I a very worried about API churn and it takes me a non trivial amount of time to review the potential implications of this change

I have two major points:

1. I really think it is important to not force downstream users to change how they call project_schema as it is a very common operation and there is no technical reason we can't keep their code the same (though it is harder in Datafusion). I made a PR to do so (and backout many of the changes currently required in this PR): askalt#5 as suggested by @crepererum
2. As I mentioned previously OptionProjectionRef seems overly complicated to me to add to the public API -- and because the Option is wrapped, it means that the normal functions for Option (like match and .and_then, etc) don't work. I think it would be simpler, and follow the rest of the code more closely if it were Option<ProjectionRef> and the special projection application code is put into ProjectionRef

Thank you for your work on this @askalt

[askalt]

I am sorry this is taking so long, but I a very worried about API churn and it takes me a non trivial amount of time to review the potential implications of this change

I have two major points:

1. I really think it is important to not force downstream users to change how they call project_schema as it is a very common operation and there is no technical reason we can't keep their code the same (though it is harder in Datafusion). I made a PR to do so (and backout many of the changes currently required in this PR): askalt#5 as suggested by @crepererum
2. As I mentioned previously OptionProjectionRef seems overly complicated to me to add to the public API -- and because the Option is wrapped, it means that the normal functions for Option (like match and .and_then, etc) don't work. I think it would be simpler, and follow the rest of the code more closely if it were Option<ProjectionRef> and the special projection application code is put into ProjectionRef

Thank you for your work on this @askalt

Ok, I will do it.