FIX : some benchmarks are failing

[getChan]

Which issue does this PR close?

• Closes distinct_query_sql benchmark is failing #15213 .

Rationale for this change

It is not certain, but it seems that plan creation and collect() should share the same runtime.. It is presumed that the issue occurred because RepartitionExec lazily polls within the runtime.

• https://github.com/apache/datafusion/blob/main/datafusion/physical-plan/src/repartition/mod.rs#L165-L177
• fix: reduce lock contention in RepartitionExec::execute #10009

I will add more details if I find anything additional.

What changes are included in this PR?

• Move the Runtime::new() intobench_function
• Ensure that plan creation (ctx.sql()) and collect() share the same runtime.

Are these changes tested?

yes. below test are succeded

• cargo bench -p datafusion --bench topk_aggregate
• cargo bench -p datafusion --bench distinct_query_sql

Are there any user-facing changes?

No.

[alamb]

Thanks for this @getChan

[alamb]

I think this means that the benchmark will include the time to create each tokio runtime (with a bunch of threads, etc)

To avoid this I think you can create the runtime once, and then use it for each tieration:

            let rt = Runtime::new().unwrap();
    c.bench_function(
        format!("distinct query with {} partitions and {} samples per partition with limit {}", partitions, samples, limit).as_str(),
        |b| b.iter(|| {

[getChan]

Got it. Changed the runtime to be shared between loops.

However, it seems that other benchmark codes still have runtime creation within the iteration.