在 Spark Catalyst 中从一个逻辑计划转换为另一个逻辑计划答案

【问题标题】：Convert from a logical plan to another logical plan in Spark Catalyst在 Spark Catalyst 中从一个逻辑计划转换为另一个逻辑计划
【发布时间】：2018-01-21 17:57:08
【问题描述】：

我使用 Spark Catalyst 表示 openCypher 查询引擎 ingraph 的查询计划。在查询计划过程中，我想从某个逻辑计划（Plan1）转换为另一个逻辑计划（Plan2）。（我尽量让问题保持简单，所以在这里省略了一些细节。该项目是完全开源的，所以如果需要，我很乐意提供更多关于why this is necessary的信息。）

我能找到的最佳方法是递归使用transformDown。这是一个小示例，它通过将每个 OpA1 实例替换为 OpA2 并将每个 OpB1 实例替换为 OpB2 来从 Plan1Nodes 转换为 Plan2Nodes。

import org.apache.spark.sql.catalyst.expressions.Attribute
import org.apache.spark.sql.catalyst.plans.logical.{LeafNode, LogicalPlan, UnaryNode}

trait Plan1Node extends LogicalPlan

case class OpA1() extends LeafNode with Plan1Node {
  override def output: Seq[Attribute] = Seq()
}
case class OpB1(child: Plan1Node) extends UnaryNode with Plan1Node {
  override def output: Seq[Attribute] = Seq()
}

trait Plan2Node extends LogicalPlan

case class OpA2() extends LeafNode with Plan2Node {
  override def output: Seq[Attribute] = Seq()
}
case class OpB2(child: Plan2Node) extends UnaryNode with Plan2Node {
  override def output: Seq[Attribute] = Seq()
}

object Plan1ToPlan2 {
  def transform(plan: Plan1Node): Plan2Node = {
    plan.transformDown {
      case OpA1() => OpA2()
      case OpB1(child) => OpB2(transform(child))
    }
  }.asInstanceOf[Plan2Node]
}

这种方法可以完成工作。这段代码：

val p1 = OpB1(OpA1())
val p2 = Plan1ToPlan2.transform(p1)

结果：

p1: OpB1 = OpB1
+- OpA1

p2: Plan2Node = OpB2
+- OpA2

但是，在代码中使用asInstanceOf[Plan2Node] 绝对是一种难闻的气味。我考虑使用Strategy 来定义转换规则，但该类用于从物理计划转换为逻辑计划。

有没有更优雅的方式来定义逻辑计划之间的转换？还是使用多个逻辑计划被视为反模式？

【问题讨论】：

标签： scala apache-spark apache-spark-sql

【解决方案1】：

（回答我自己的问题。）

将 transformDown 更改为简单的模式匹配 (match) 和递归调用解决了类型问题：

object Plan1ToPlan2 {
  def transform(plan: Plan1Node): Plan2Node = {
    plan match {
      case OpA1() => OpA2()
      case OpB1(child) => OpB2(transform(child))
    }
  }
}

这似乎是一种类型安全且（基于我对 Catalyst 的有限理解）惯用的解决方案。

【讨论】：