[SPARK-9241] [SQL] [WIP] Supporting multiple DISTINCT columns

[hvanhovell]

This PR adds support for multiple distinct columns to the new aggregation code path.

The implementation uses the OpenHashSet class and set expressions. As a result we can only use the slower sort based aggregation code path. This also means the code will be probably slower than the old hash aggregation.

The PR is currently in the proof of concept phase, and I have submitted it to get some feedback to see if I am headed in the right direction. I'll add more tests if this considered to be the way to go.

An example using the new code path:

val df = sqlContext
  .range(1 << 25)
  .select(
    $"id".as("employee_id"),
    (rand(6321782L) * 4 + 1).cast("int").as("department_id"),
    when(rand(981293L) >= 0.5, "M").otherwise("F").as("gender"),
    (rand(7123L) * 3 + 1).cast("int").as("education_level")
  )

df.registerTempTable("employee")

// Regular query.
sql("""
select   department_id as d,
         count(distinct gender, education_level) as c0,
         count(distinct gender) as c1,
         count(distinct education_level) as c2
from     employee
group by department_id
""").show()

cc @yhuai

[JoshRosen]

Jenkins, this is ok to test.

[SparkQA]

Test build #44363 has finished for PR 9280 at commit 3bd6db5.

• This patch fails Spark unit tests.
• This patch merges cleanly.
• This patch adds the following public classes (experimental):\n * case class DistinctAggregateFallback(function: AggregateFunction2) extends DeclarativeAggregate\n * case class ReduceSetUsingImperativeAggregate(left: Expression, right: ImperativeAggregate)\n * case class ReduceSetUsingDeclarativeAggregate(left: Expression, right: DeclarativeAggregate)\n

[SparkQA]

Test build #44376 has finished for PR 9280 at commit b76b83b.

• This patch passes all tests.
• This patch merges cleanly.
• This patch adds the following public classes (experimental):\n * case class DistinctAggregateFallback(function: AggregateFunction2) extends DeclarativeAggregate\n * case class ReduceSetUsingImperativeAggregate(left: Expression, right: ImperativeAggregate)\n * case class ReduceSetUsingDeclarativeAggregate(left: Expression, right: DeclarativeAggregate)\n * case class DropAnyNull(child: Expression) extends UnaryExpression\n

[hvanhovell]

I'll remove this in the next iteration...

[hvanhovell]

Removed....

[SparkQA]

Test build #44430 has finished for PR 9280 at commit bfbf829.

• This patch passes all tests.
• This patch merges cleanly.
• This patch adds the following public classes (experimental):\n * case class DistinctAggregateFallback(function: AggregateFunction2) extends DeclarativeAggregate\n * case class ReduceSetUsingImperativeAggregate(left: Expression, right: ImperativeAggregate)\n * case class ReduceSetUsingDeclarativeAggregate(left: Expression, right: DeclarativeAggregate)\n * case class DropAnyNull(child: Expression) extends UnaryExpression\n

[rxin]

@hvanhovell I think the approach we want to take is to either use the aggregate expansion (#9406), or use joins, and not this one. Do you mind closing this one? (We already have a record of it from JIRA in case we need to reference it).

[hvanhovell]

Closing PR. Some of the code in here will probably re-emerge if we ever want distincts in window functions.