[SPARK-24985][SQL][WIP] Fix OOM in Full Outer Join in case of data skew

[sujithjay]

What issue does this pull request address ?

JIRA: https://issues.apache.org/jira/browse/SPARK-24985
In the case of Full Outer Joins of large tables, in the presence of data skew around the join keys for either of the joined tables, OOMs exceptions occur. While its possible to increase the heap size to workaround, Spark should be resilient to such issues as skews can happen arbitrarily.

What changes were proposed in this pull request?

#16909 introduced ExternalAppendOnlyUnsafeRowArray & changed SortMergeJoinExec to use ExternalAppendOnlyUnsafeRowArray for every join, except 'Full Outer Join'. This PR makes changes to make 'Full Outer Joins' to use ExternalAppendOnlyUnsafeRowArray.

How was this patch tested?

Unit testing

• Changed a test-case in JoinSuite.
• Existing tests in OuterJoinSuite were used to verify correctness.

Stress testing

• This is still a work in progress. I plan to verify this patch using a production workload.

[sujithjay]

Hi @tejasapatil, @viirya, @hvanhovell, & @kiszk, can you please review this pull request?

[viirya]

Can we keep the scanning left and right iterators? Because if they are spilled, obtaining the iterator from spilled data needs to loop over spill writers and create readers. We may avoid calling generateIterator every time for obtaining the iterators. However it might make the code a bit complicated than now.

[sujithjay]

Hi @viirya,
Thank you for reviewing the code.

I agree the code will be a bit complicated if we had to keep scanning the iterators. In this particular case, I was following a pattern observed throughout the rest of the class.
For eg.,

spark/sql/core/src/main/scala/org/apache/spark/sql/execution/joins/SortMergeJoinExec.scala

Lines 307 to 329 in 35f7f5c

	override def advanceNext(): Boolean = {
	while (smjScanner.findNextOuterJoinRows()) {
	currentLeftRow = smjScanner.getStreamedRow
	val currentRightMatches = smjScanner.getBufferedMatches
	if (currentRightMatches == null || currentRightMatches.length == 0) {
	numOutputRows += 1
	return true
	}
	var found = false
	val rightMatchesIterator = currentRightMatches.generateIterator()
	while (!found && rightMatchesIterator.hasNext) {
	joinRow(currentLeftRow, rightMatchesIterator.next())
	if (boundCondition(joinRow)) {
	found = true
	}
	}
	if (!found) {
	numOutputRows += 1
	return true
	}
	}
	false
	}

Having said that, I do feel the penalty for following a similar approach in case of full outer joins could be higher. I will try & see what I can do.

[sujithjay]

Hi @viirya ,
After some deliberation, I figure it would not be possible to avoid the reinitialisation of the right iterator. Please share your thought on this.