SPARK-1380: Add sort-merge based cogroup/joins.

[ueshin]

I've written cogroup/joins based on 'Sort-Merge' algorithm.

[AmplabJenkins]

Merged build triggered.

[AmplabJenkins]

Merged build started.

[AmplabJenkins]

Merged build finished. All automated tests passed.

[AmplabJenkins]

All automated tests passed.
Refer to this link for build results: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/13626/

[rxin]

Is there a specific use case you are trying to address that cannot be handled by the hash join?

[mridulm]

I have not done a detailed review - but looks pretty expensive in terms of memory.
Is it making assumptions about lack of skew w.r.t a key and amount of data per partition (that it can be held entirely in memory) ?
Would be good to document what are the constraints of the solution.

[ueshin]

@rxin Thank you for your reply.

There are some case to use merge join for optimization:

1. If data to be joined are already sorted by join keys, merge join would be done more efficiently than hash join. In my test case, both algorithms were almost same speed, but merge join was scalable.
2. Merge join for sorted data by the same keys would be pipelined (each output can be produced immediately for arrived input tuples) even if N-way join (N>2). Hash join blocks due to building a hash-table before output are produced.

I think it is useful for users to choose ways to optimize their processing.

[ueshin]

@mridulm Thank you for your reply.

There are 2 points I have to mention about memory:

• Before shuffle
  If data are sorted, no more memory is needed because no sort operation is needed, and if not sorted, merge join needs some amount of memory to sort data in each partition.
• After shuffle
  Merge join needs at most the same amount of memory as hash join while fetching data, but it does not need more memory because it can produce output immediately from input. Hash join needs some more memory to build a hash table.

[nchammas]

@pwendell @rxin @mateiz What is the status of this PR? It looks pretty substantial, but it hasn't been updated in a while.

[pwendell]

I'd suggest we close this issue for now and go to the JIRA to discuss whether the feature is needed and how high of a priority it is.