[SPARK-11055][SQL] Use mixing hash-based and sort-based aggregation in TungstenAggregationIterator

[viirya]

JIRA: https://issues.apache.org/jira/browse/SPARK-11055

In TungstenAggregationIterator we switch to sort-based aggregation when we can't allocate more memory for hashmap.

However, using external sorter-based aggregation will write too much key-value pairs into disk. We should use mixing hash-based and sort-based aggregation to reduce the key-value pairs needed to write to disk.

[SparkQA]

Test build #43546 has finished for PR 9067 at commit dd98d57.

• This patch fails Spark unit tests.
• This patch merges cleanly.
• This patch adds no public classes.

[SparkQA]

Test build #43557 has finished for PR 9067 at commit 787846b.

• This patch fails Spark unit tests.
• This patch merges cleanly.
• This patch adds no public classes.

[viirya]

retest this please.

[SparkQA]

Test build #43559 has finished for PR 9067 at commit 787846b.

• This patch passes all tests.
• This patch merges cleanly.
• This patch adds no public classes.

[rxin]

Can you explain what you mean by "mixing"?

[JoshRosen]

@rxin, my understanding of this patch is that it lets us continue to perform hash-based pre-aggregation on the remainder of the iterator after we've decided to spill and switch to sort-based aggregation. After we've destructed and freed the original hashMap, we'll now loop back around and continue to use a new hash map to aggregate the remainder of the iterator, spilling that map if it also becomes too large.

After this patch, records will have more opportunities to be pre-aggregated before being spilled to disk or sent across the network. I wonder about the case where you're doing map-side pre-aggregation and are experiencing a very low reduction factor: in that case, this patch means that we'll do more work per record, since we'll be hashing and sorting every record, whereas before there was a chance that we'd skip hashing of some records. OTOH, if you have an input consisting of all unique keys then it's best to skip both the hashing AND the sorting and just push the records straight to the reduce side, but that optimization is kind of orthogonal to this one.

[JoshRosen]

I think you might actually want to re-assign to the original hashMap by making it into a var. This will let us ensure that UnsafeAggregationIterator.free() is able to free all memory after failures.

[rxin]

Ideally we should be able to turn partial aggregation off when we don't see reduction. We had that in Shark, and a lot of query engines do this.

[viirya]

@JoshRosen Thanks for explaining this patch. It does exactly what you said.

@rxin Do you think it is ok to add a configuration for turning on/off this feature?

[rxin]

It's best to have a feature flag for this, in case it yields worse performance. Eventually we should find a way to make this the default.

Can you do some performance measure to quantify the gain from this change?

[SparkQA]

Test build #44359 has finished for PR 9067 at commit aa51d8a.

• This patch passes all tests.
• This patch merges cleanly.
• This patch adds no public classes.

[viirya]

@rxin I ran a simple performance measure as following.

Record count: 1333635318
Record after group by: 259200

SQL query looks like: SELECT SUM(a) as a , SUM(b) as b , SUM(c) as c , SUM(d) as d from table GROUP BY e

4 workers (8 cores), executor memory: 512 MB.

With pre-aggregation enabled:

67720191 microseconds
66424539 microseconds
62959275 microseconds

With pre-aggregation disabled:

69934956 microseconds                                                           
70351959 microseconds                                                           
68437353 microseconds   

So looks like it roughly gains about 5% improvement in average.

[davies]

@viirya Did not realized that you had don similar things, i created davies@5707f5b, could you review that?

[viirya]

@davies ok.