SPARK-3462 push down filters and projections into Unions

[koeninger]

No description provided.

[SparkQA]

Can one of the admins verify this patch?

[marmbrus]

ok to test

[SparkQA]

QA tests have started for PR 2345 at commit ef47b3b.

• This patch merges cleanly.

[marmbrus]

Hey @koeninger, thanks for implementing this optimization!

Overall this looks pretty good. A few minor suggestions:

• I'm not sure that we want to check the names and qualifiers to find the corresponding Attributes. It is possible that either side could actually have tables that are aliased differently and thus the Attributes would have different qualifiers. Instead, I think that it safe to assume that the analysis phase has checked name and type matching and just find the corresponding Attribute by ordering. I'm thinking something like this:

/**
 * Pushes Project and Filter operations to either side of a Union.
 */
object UnionPushdown extends Rule[LogicalPlan] {

  /**
   * Rewrites an expression so that it can be pushed to the right side of a Union operator.
   * This method relies on the fact that the output attributes of a union are always equal to the
   * left child's output.
   */
  def pushToRight[A <: Expression](e: A, union: Union): A = {
    assert(union.left.output.size == union.right.output.size)

    // Maps Attributes from the left side to the corresponding Attribute on the right side.
    val rewrites = AttributeMap(union.left.output.zip(union.right.output))
    val result = e transform {
      case a: Attribute => rewrites(a)
    }

    // We must promise the compiler that we did not discard the names in the case of project
    // expressions.  This is safe since the only transformation is from Attribute => Attribute.
    result.asInstanceOf[A]
  }

  def apply(plan: LogicalPlan): LogicalPlan = plan transform {
    // Push down filter into union
    case Filter(condition, u @ Union(left, right)) =>
      Union(
        Filter(condition, left),
        Filter(pushToRight(condition, u), right))

    // Push down projection into union
    case Project(projectList, u @ Union(left, right)) =>
      Union(
        Project(projectList, left),
        Project(projectList.map(pushToRight(_, u)), right))
  }
}

• Would also be great to add some tests to FilterPushdownSuite and maybe create a similar ColumnPruningSuite.

[SparkQA]

QA tests have finished for PR 2345 at commit ef47b3b.

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

[koeninger]

@marbrus I see what you mean. Updated to basically what you suggested, aside from building the map once. Let me know, once it's finalized I can try to test one more time on live data.

[SparkQA]

QA tests have started for PR 2345 at commit 0788691.

• This patch merges cleanly.

[marmbrus]

Nit: union is not used.

[SparkQA]

QA tests have started for PR 2345 at commit 0788691.

• This patch merges cleanly.

[marmbrus]

LGTM to me once the tests pass.

[SparkQA]

QA tests have started for PR 2345 at commit 5c8d24d.

• This patch merges cleanly.

[SparkQA]

QA tests have finished for PR 2345 at commit 0788691.

• This patch passes unit tests.
• This patch merges cleanly.
• This patch adds the following public classes (experimental):
  • case class CreateTableAsSelect(
  • case class CreateTableAsSelect(

[SparkQA]

QA tests have finished for PR 2345 at commit 0788691.

• This patch passes unit tests.
• This patch merges cleanly.
• This patch adds the following public classes (experimental):
  • case class CreateTableAsSelect(
  • case class CreateTableAsSelect(

[marmbrus]

Thanks! I've merged to master.

[SparkQA]

QA tests have finished for PR 2345 at commit 5c8d24d.

• This patch passes unit tests.
• This patch merges cleanly.
• This patch adds the following public classes (experimental):
  • case class CreateTableAsSelect(
  • case class CreateTableAsSelect(