[SPARK-23877][SQL]: Use filter predicates to prune partitions in metadata-only queries

[rdblue]

What changes were proposed in this pull request?

This updates the OptimizeMetadataOnlyQuery rule to use filter expressions when listing partitions, if there are filter nodes in the logical plan. This avoids listing all partitions for large tables on the driver.

This also fixes a minor bug where the partitions returned from fsRelation cannot be serialized without hitting a stack level too deep error. This is caused by serializing a stream to executors, where the stream is a recursive structure. If the stream is too long, the serialization stack reaches the maximum level of depth. The fix is to create a LocalRelation using an Array instead of the incoming Seq.

How was this patch tested?

Existing tests for metadata-only queries.

[SparkQA]

Test build #88959 has finished for PR 20988 at commit 2345896.

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

[rdblue]

@cloud-fan or @gatorsmile, could you review this?

[cloud-fan]

can we add a test? We can use HiveCatalogMetrics.METRIC_PARTITIONS_FETCHED.getCount() to check if this patch can really reduce the number of partitions being fetched.

[cloud-fan]

what about Filter(p > 1, Project(a, p, Table(a, b, p, partitioned by p)))? p > 1 should also be a partition filter.

[cloud-fan]

I'd propose something top-down like

def getPartitionedRelation(
    plan: LogicalPlan,
    predicates: Seq[Expression]): Option[(AttributeSet, Seq[Expression], LogicalPlan)] = {
  plan match {
    case Filter(condition, child) if condition.deterministic =>
      getPartitionedRelation(child, predicates ++ splitConjunctivePredicates(condition))
   
    case Project(projectList, child) if projectList.forall(_.deterministic) =>
      getPartitionedRelation(child, predicates.filter(_.references.subsetOf(child.outputSet)))
    
    case l @ LogicalRelation(fsRelation: HadoopFsRelation, _, _, _) if fsRelation.partitionSchema.nonEmpty =>
      val partAttrs = ...
      val partitionFilters = predicates.filter(_.references.subsetOf(partAttrs))
      Some(...)

    case _ => None
  }
}

[rdblue]

@cloud-fan, that is basically how this works already. Each matched node calls unapply(child) to get the result from the child node, then it adds the current node's conditions to that result. Using unapply instead of getPartitionedRelation makes this work in the matching rule:

  case a @ Aggregate(_, aggExprs, child @ PartitionedRelation(partAttrs, filters, relation)) =>

[rdblue]

@cloud-fan, I've added the test. Thanks for letting me know about HiveCatalogMetrics, that's exactly what I needed.

[SparkQA]

Test build #89535 has finished for PR 20988 at commit 6e0685e.

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

[cloud-fan]

there is a bug here. Think about Filter(x > 1, Project(p + 1 as x, Table(a, p, partitioned by p))), we will mistakenly report x > 1 as partition predicates and use it to list partitions and fail.

I think we should use PhysicalOperation here, which can help us to substitute the attributes in filter.

[rdblue]

Good catch. I've added that as a test case and updated the PartitionedRelation code to keep track of both original partition attributes -- that the filter needs -- and the top-most node's output that is used by the rule. One thing to note: the optimizer usually pushes Filter below Project` by the time this runs, so it is difficult to even construct this test case.

For using PhysicalOperation instead of PartitionedRelation, I don't see a compelling reason for such an invasive change. This currently adds a couple of results to unapply and keeps mostly the same logic. PhysicalOperation would lose the check that the references are a subset of the partition attributes and be a lot larger change. If you think this should be refactored, lets talk about that separately to understand the motivation for the change.

[SparkQA]

Test build #89588 has finished for PR 20988 at commit 93cf217.

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

[cloud-fan]

thanks, merging to master!