Wrap in CoalesceBatchesExec more operators

[akupchinskiy]

What is the problem the feature request solves?

Datafusion optimizer wraps some operators which might produce multiple small batches with CoalesceBatchesExec node. In Comet, this technique is already applied to SMJ with filter.

Describe the potential solution

Shouldn't it be extended to other plans (Filter/HashJoin)? It might be especially beneficial for workloads where sort merge join -> hash join replacement is enforced.

Additional context

No response

[andygrove]

I did experiment with this in the past (#495). My conclusion at the time was that when we write to a shuffle file, we are effectively coalescing batches as part of the process, so it was redundant to add additional CoalesceBatchesExec operators.

[akupchinskiy]

Ahh, make sense. I faced degradation where hash join has been chained with another comet operator (broadcast exchange), not the shuffle writer.

[akupchinskiy]

It is q14 from tpdcs benchmark with hash join over smj preference config on

[akupchinskiy]

I am thinking of making this logic conditional. We can infer chain patterns during CometExecRule optimization phase and pass the flag to the native planner to do coalesce wrapping only in cases like above. @andygrove what do you think?

[andygrove]

I'd like to bring @mbutrovich into the discussion since we have been discussing where optimization rules should live. I think we are leaning towards keeping the scala plan representation as close to the Spark plan as possible and handle adding additional nodes on the native side.

I do think we should continue exploring adding coalesce execs similar to how DataFusion does this in a physical optimizer rule.