[SPARK-17712][SQL] Fix invalid pushdown of data-independent filters beneath aggregates

[JoshRosen]

What changes were proposed in this pull request?

This patch fixes a minor correctness issue impacting the pushdown of filters beneath aggregates. Specifically, if a filter condition references no grouping or aggregate columns (e.g. WHERE false) then it would be incorrectly pushed beneath an aggregate.

Intuitively, the only case where you can push a filter beneath an aggregate is when that filter is deterministic and is defined over the grouping columns / expressions, since in that case the filter is acting to exclude entire groups from the query (like a HAVING clause). The existing code would only push deterministic filters beneath aggregates when all of the filter's references were grouping columns, but this logic missed the case where a filter has no references. For example, WHERE false is deterministic but is independent of the actual data.

This patch fixes this minor bug by adding a new check to ensure that we don't push filters beneath aggregates when those filters don't reference any columns.

How was this patch tested?

New regression test in FilterPushdownSuite.

[JoshRosen]

If you use replaced.references here then this will break the "aggregate: push down filters with literal" test case. Here's that test:

  test("aggregate: push down filters with literal") {
    val originalQuery = testRelation
      .select('a, 'b)
      .groupBy('a)('a, count('b) as 'c, "s" as 'd)
      .where('c === 2L && 'd === "s")

    val optimized = Optimize.execute(originalQuery.analyze)

    val correctAnswer = testRelation
      .where("s" === "s")
      .select('a, 'b)
      .groupBy('a)('a, count('b) as 'c, "s" as 'd)
      .where('c === 2L)
      .analyze

    comparePlans(optimized, correctAnswer)
  }

If you use replaced.references then the 'd in the filter is replaced with "s" and the resulting expression no longer contains any references and thus isn't pushed.

[hvanhovell]

LGTM - pending Jenkins.

[SparkQA]

Test build #66074 has finished for PR 15289 at commit 87504e4.

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

[hvanhovell]

Merging to master. Thanks!

[hvanhovell]

@JoshRosen I cannot cherry-pick this one into 2.0. Could you open a PR against 2.0 I feel you that this is required?

[ioana-delaney]

@JoshRosen Wouldn't be a better design to push down the predicate but keep the original predicate as well? If the aggregate is above a complex join, not pushing down the predicate may have significant performance implications.

[JoshRosen]

@ioana-delaney, to clarify, are you suggesting that there are predicates which we could correctly be pushing down past aggregates but are not pushing? I believe that the current logic encompasses all of the cases that are safe to push.

Or are you suggesting that we're affecting correctness by not keeping the original predicate following the pushdown?

[JoshRosen]

@hvanhovell, I've cherry-picked this fix to branch-2.0.

[ioana-delaney]

@JoshRosen The original predicate has to be kept above the aggregation. An optimization would be to also push down the predicate below the aggregation, lower in the plan for early filtering. In theory an always-false predicate should eliminate an entire sub-plan.

[ioana-delaney]

@JoshRosen In your example, we don't want to first count one million rows coming from the base table and then to return zero rows based on the false predicate in the outer query block. Instead, by pushing down the predicate to the base table, you do a pre-filtering and return zero rows early in the plan. Then you apply the aggregate followed by the the original predicate that will do the final filtering. Anyway, just some thought for further optimizations of predicates pushed down through aggregation. Also, a more realistic query would imply false predicates through predicate transitivity e.g. a != b and a =1 and b =1 => 1 != 1 So there might be some real customer queries that can take advantage of these optimizations.