Repository navigation
[DISCUSSION]: Unified approach for joins to output batches close to batch_size #14238
Description
`BatchCoalescer` is not used in joins yet, since CoalesceBatchesExec appears after the joins having filter, in case of the output batches might have a lower row count than target batch size. So, why cannot we follow the same pattern in SMJ? If collecting batches in the join itself is more performant, then we should also refactor the other joins as well?
On the other hand, BatchSplitter is used in other joins, and SMJ could (should) have it too, as there is no other way of splitting the batches according to target batch size.
I've thought about this, and I believe the most optimal solution is to make all join operators capable of performing both coalescing and splitting in a built-in manner. This is because the output of a join can either be smaller or larger than the target batch size. Ideally, there should be no need (or only minimal need) for CoalesceBatchesExec.
To achieve this built-in coalescing and splitting, we can leverage existing tools like BatchSplitter and BatchCoalescer (although there are no current examples of BatchCoalescer being used in joins). My suggestion is to generalize these tools so they can be utilized by any operator and applied wherever this mechanism is needed. As this pattern becomes more common, it will be easier to expand its usage and simplify its application.
Originally posted by @berkaysynnada in #14160 (comment)
Activity
comphead commentedon Jan 22, 2025
The direction proposed by @berkaysynnada is worth to discuss. The join specifics doesn't guarantee output batch size in records. It can much much smaller or even empty because of filtering, and it can be much larger because of join explosions.
The idea to discuss how we can make the output batches after joins to be more uniform and close to configured
batch_size.One of the options is to use
BatchSplitterorBatchCoalesceplan nodes after the join is called.Another is to align the batches in the join internally providing the coalescer/splitter or having custom implementation.
korowa commentedon Jan 24, 2025
I'd suggest to rename "splitting" part of the problem to "restricting" -- if join is able to produce a batch that needs to be splitted (event if this batch exists only internally), than it already may be issue, which may hurt on some specific cases. I also think that
BatchSplitterin it's current implementation (when it already has a batch to split) is not solving the problem, but just tries to fix/hide it (in addition if these batches for splitting are large enough, to start causing memory issues,BatchSplitterdoesn't seem to be able to help).In this case (for splitting / restricting), I think, what @berkaysynnada suggests:
is a better fit than separate operators on top of join -- each join operator should by itself be able to limit / restrict its internally created record batches to prevent excessive accumulation of data in memory (or at least, if it's required, to track them via memory reservations).
comphead commentedon Jan 24, 2025
thanks @korowa totally agree for the memory perspective, having splitter won't help as the memory already allocated for the batch.
However another path related to coalesce might help downstream nodes or direct consumer not to struggle because of swarm of small batches. More uniform method for all joins is to call
CoalesceBatchExecjust after the join execution however builtin approach might be more efficientkorowa commentedon Jan 24, 2025
I don't have a strong opinion here -- intuitively it seems like embedding coalescer into filtering operators (not only joins) could be beneficial for query execution time just because there will be less operators in the pipeline, but it still should be checked and somehow measured.
I'll try to come up with some POC during this weekend for coalescer in e.g. FilterExec (this one just seems to be the easiest to implement) -- the idea is that if it'll work well for filters, than joins would also benefit from it, otherwise -- having separate operator would make more sense (at least for now).
korowa commentedon Jan 26, 2025
Simple embedding of coalescer into filter (branch commit) gave following tpch results:
iterations = 50, target_partitions = 1
iterations = 50, target_partitions = 4
I tend to interpret these numbers as no difference between the two approaches so I don't see (yet) a rationale for embedding coalescer into the operators, and it seems to be fine to leave it as it is.
korowa commentedon Jan 26, 2025
As a further step, I'd like to also try it for HashJoin because of this point
Now hash join produces batches with less than or equal to batch_size rows, and in cases when the number of output rows is greater than the number of input rows (some non-unique keys on build side) it may end up with a sequence of batches sized as
where small batches are the result of key non-uinqueness. This sequence won't be compacted by CoalesceBatchesExec, due to full-sized batches, but having internal join buffer for the output may help with better output batch size alignment -> less number of batches.
comphead commentedon Jan 31, 2025
Thanks @korowa for running the benchmarks. 🚀
Speaking to
CoalescerIt is probably worth to think if we want to have a unified approach, like calling the coalescer on top of the join result stream. If I understand your message correctly there won't be performance drawbacks.The benefit of it is to reduce join complexity and separate out the coalesce step(which implemented in every type of join in its own way).