[SPARK-8160][SQL]Support using external sorting to run aggregate

[lianhuiwang]

according to UnsafeFixedWidthAggregationMap, this PR implement external aggregation when each task has no enough memory. firstly when memory is not enough it spills UnsafeAppendOnlyMap to disk. when inserting is finished, merge in-memory map and disk's spill files and return final aggregation value of each key.

[SparkQA]

Test build #37375 has finished for PR 7423 at commit 4d7a310.

• This patch fails RAT tests.
• This patch merges cleanly.
• This patch adds the following public classes (experimental):
  • public class UnsafeRowLocation
  • public class UnsafeAppendOnlyMap
  • public static final class UnsafeMapSorterIterator implements Iterator<UnsafeRowLocation>
  • public final class UnsafeExternalAggregation
  • public class BufferedIterator
  • public class MapEntry
  • public class UnsafeSorterKVSpillWriter
  • final class UnsafeSorterKVSpillReader implements AbstractScalaIterator<UnsafeRowLocation>
  • public class UnsafeAppendOnlyMapIterator implements AbstractScalaIterator<MapEntry>
  • abstract class MutableProjection extends Projection

[SparkQA]

Test build #37377 has finished for PR 7423 at commit 688cdf4.

• This patch fails RAT tests.
• This patch merges cleanly.
• This patch adds the following public classes (experimental):
  • public class UnsafeRowLocation
  • public class UnsafeAppendOnlyMap
  • public static final class UnsafeMapSorterIterator implements Iterator<UnsafeRowLocation>
  • public final class UnsafeExternalAggregation
  • public class BufferedIterator
  • public class MapEntry
  • public class UnsafeSorterKVSpillWriter
  • final class UnsafeSorterKVSpillReader implements AbstractScalaIterator<UnsafeRowLocation>
  • public class UnsafeAppendOnlyMapIterator implements AbstractScalaIterator<MapEntry>
  • abstract class MutableProjection extends Projection

[davies]

@lianhuiwang This is cool, thanks for working on it. Recently, we have a refactor on aggregation (new code path), could you hold a few days until we finalize that?

[lianhuiwang]

ok,@davies thanks.

[SparkQA]

Test build #37378 has finished for PR 7423 at commit 9116810.

• This patch fails Scala style tests.
• This patch merges cleanly.
• This patch adds the following public classes (experimental):
  • public class UnsafeRowLocation
  • public class UnsafeAppendOnlyMap
  • public static final class UnsafeMapSorterIterator implements Iterator<UnsafeRowLocation>
  • public final class UnsafeExternalAggregation
  • public class BufferedIterator
  • public class MapEntry
  • public class UnsafeSorterKVSpillWriter
  • final class UnsafeSorterKVSpillReader implements AbstractScalaIterator<UnsafeRowLocation>
  • public class UnsafeAppendOnlyMapIterator implements AbstractScalaIterator<MapEntry>
  • abstract class MutableProjection extends Projection

[JoshRosen]

High-level question: why not wrap / re-use BytesToBytesMap instead of duplicating most of its code here?

[JoshRosen]

What are the key differences between this class and BytesToBytesMap?

[lianhuiwang]

because BytesToBytesMap has more complex, it did memory allocation's works.
it's differences between their are UnsafeAppendOnlyMap split memory allocation and RowLocation from BytesToBytesMap.
if based on BytesToBytesMap, UnsafeFixedWidthAggregationMap should change many. if it is necessary, i can did re-use BytesToBytesMap but it need to change some code of UnsafeFixedWidthAggregationMap.
moreover, i think UnsafeExternalAggregation could replace with UnsafeFixedWidthAggregationMap.

[lianhuiwang]

another point is we can remove UnsafeFixedWidthAggregationMap and BytesToBytesMap now, and just use UnsafeExternalAggregation. when it donot support Object schema, it do not did spills that is same as UnsafeFixedWidthAggregationMap.
i think it is better than do some changes on BytesToBytesMap and UnsafeFixedWidthAggregationMap. if it is ok, i think i can did.
@davies i think you have some opinions about it. thanks.

[SparkQA]

Test build #37484 has finished for PR 7423 at commit 68552cc.

• This patch fails to build.
• This patch merges cleanly.
• This patch adds the following public classes (experimental):
  • public class UnsafeRowLocation
  • public class UnsafeAppendOnlyMap
  • public static final class UnsafeMapSorterIterator implements Iterator<UnsafeRowLocation>
  • public final class UnsafeExternalAggregation
  • public class BufferedIterator
  • public class MapEntry
  • public class UnsafeSorterKVSpillWriter
  • final class UnsafeSorterKVSpillReader implements AbstractScalaIterator<UnsafeRowLocation>
  • public class UnsafeAppendOnlyMapIterator implements AbstractScalaIterator<MapEntry>
  • abstract class MutableProjection extends Projection

[SparkQA]

Test build #37501 has finished for PR 7423 at commit e651683.

• This patch passes all tests.
• This patch merges cleanly.
• This patch adds the following public classes (experimental):
  • public class UnsafeRowLocation
  • public class UnsafeAppendOnlyMap
  • public static final class UnsafeMapSorterIterator implements Iterator<UnsafeRowLocation>
  • public final class UnsafeExternalAggregation
  • public class BufferedIterator
  • public class MapEntry
  • public class UnsafeSorterKVSpillWriter
  • final class UnsafeSorterKVSpillReader implements AbstractScalaIterator<UnsafeRowLocation>
  • public class UnsafeAppendOnlyMapIterator implements AbstractScalaIterator<MapEntry>
  • abstract class MutableProjection extends Projection
  • case class Length(child: Expression) extends UnaryExpression with ExpectsInputTypes
  • case class FormatNumber(x: Expression, d: Expression)

[lianhuiwang]

@davies @JoshRosen i will close this PR because i have created new PR #7827 based on new aggregate interface.