Open alamb opened 2 years ago
I would love to implement this algorithm in DataFusion:
https://arxiv.org/abs/2010.00152 Sort-based grouping and aggregation Thanh Do, Goetz Graefe
I would love to implement this algorithm in DataFusion:
https://arxiv.org/abs/2010.00152 Sort-based grouping and aggregation Thanh Do, Goetz Graefe
Maybe you can share it at the next meeting ๐
I would love to implement this algorithm in DataFusion: https://arxiv.org/abs/2010.00152 Sort-based grouping and aggregation Thanh Do, Goetz Graefe
Maybe you can share it at the next meeting ๐
+1 well look forward ๐
I would love to implement this algorithm in DataFusion:
https://arxiv.org/abs/2010.00152 Sort-based grouping and aggregation Thanh Do, Goetz Graefe
Haha, that's great! I have talked with @houqp about this paper before.
I would propose we do this in several steps:
- [ ] Provide a classic SortMergeJoin implementation that is less memory bound itself (but move the need of memory management to the sort operator, which we already have memory controlled).
- [ ] Follow-up choice 1: Consolidate HashJoin and SortMergeJoin, providing a unified JoinExec, and do adaptive execution as @alamb suggested above.
- [ ] Follow-up choice 2: incorporate a cost-based join optimizer to choose the most suitable physical plan: sort-based or hash-based.
Moved to issue descriptions above. ๐
@yjshen sounds like a great plan to me -- and I see that @richox has an implementation for SortMergeJoin
already :)
Looks like now that we are able to fail query in case of breaching memory limit, it's the right time to start working on spills.
Taking into account what has been written above, I guess, next step could be to implement spilling for MergeJoin -- if our final intention to have runtime HJ -> MJ conversion it would be nice to have some guarantees that MJ won't fail for the same reason. I believe MJ spilling logic could be pretty straightforward without any pitfalls -- the naive approach would be to spill buffered-side data in .ipc batch by batch, more complex, and, probably, more effective way to think about would be spilling concatenation of all batches that fit in memory.
After that we could follow-up with what is mentioned in issue description -- HJ -> MJ conversion (I believe #2628 worth to be mentioned here, to unlock ability for more hash joins to be converted), and spilling mechanisms for other join implementations.
If this plan is fine, I'd like to take a stab at MJ spilling.
Looks like now that we are able to fail query in case of breaching memory limit, it's the right time to start working on spills.
I agree
I believe MJ spilling logic could be pretty straightforward without any pitfalls -- the naive approach would be to spill buffered-side data in .ipc batch by batch, more complex, and, probably, more effective way to think about would be spilling concatenation of all batches that fit in memory.
Unless there is a very compelling reason to have a separate implementation, I think we should leverage (reuse) the existing ExteraSorter
used in spilling sort:
After that we could follow-up with what is mentioned in issue description ๐
I think we should leverage (reuse) the existing ExteraSorter used in spilling sort
After some reading, it looks like that for MergeJoin
case it makes sense to split spilling part of ExternalSorter
(responsible for tracking spillable batch buffer) and sorting one -- we already have sorted streams in MJ, so spillable buffer is all that needed -- I believe this implementation could be acceptable and first I'll try to stick to this way.
And, further, ExternalSorter
also seems to be a perfect fit for HashJoin
-- this is the case where it can be reused as it is for resorting both build and probe sides of HJ.
After some reading, it looks like that for MergeJoin case it makes sense to split spilling part of ExternalSorter (responsible for tracking spillable batch buffer) and sorting one -- we already have sorted streams in MJ, so spillable buffer is all that needed -- I believe this implementation could be acceptable and first I'll try to stick to this way.
Thank you -- reusing the ExternalSorter will allow whatever spilling logic we develop to benefit from additional improvements in sort. For example, @jaylmiller has https://github.com/apache/arrow-datafusion/pull/5292 which makes substantial changes to ExternalSorter and hopefully it will be faster, but has some performance regressions that we haven't worked out yet,
It will be great if that work can directly benefit the spilling operators as well
And, further, ExternalSorter also seems to be a perfect fit for HashJoin -- this is the case where it can be reused as it is for resorting both build and probe sides of HJ.
Is your feature request related to a problem or challenge? Please describe what you are trying to do. Support Joining "arbitrarily" large inputs (e.g. when one or both of the inputs don't fit in the available RAM)
This ticket concerns the memory used the
JoinExec
operator -- it doesn't cover other potential targets (e.g. externalized sort or grouping). That will be covered by other tasks tracked by #587Describe the solution you'd like
There are many potential ways to Limit the memory used while joining. The classic way is "sort-merge-join" where the input data on both sides is sorted according to the equality predicates (using externalized sort, such as described in #1568 ) and then the two join inputs are streamed through and the output computed, depending on the type of Join required (INNER, LEFT, RIGHT, SEMI, etc)
I personally think the following would be the ideal behavior for DataFusion Joins:
The rationale for a runtime switch is that then the optimizer (which always has limited information) can't make the "wrong" choice related to join order
In case anyone wants some "light reading" this stuff is nicely described by Goetz Graffe in "Query evaluation techniques for large databases": https://scholar.google.com/citations?view_op=view_citation&hl=en&user=pdDeRScAAAAJ&citation_for_view=pdDeRScAAAAJ:u5HHmVD_uO8C
Online link: http://infolab.stanford.edu/~hyunjung/cs346/graefe.pdf
Describe alternatives you've considered Have the Optimizer (aka the
HashBuildProbeOrder
) pick both the order and the algorithm to use based on statistics or heuristicsContext This is follow on work from the great PR from @yjshen in https://github.com/apache/arrow-datafusion/pull/1526 and part of the story of limiting memory used by DataFusion https://github.com/apache/arrow-datafusion/issues/587
Task Tracking