[SPARK-9990][SQL]Create local hash join operator

[zsxwing]

This PR includes the following changes:

• Add SQLConf to LocalNode
• Add HashJoinNode
• Add ConvertToUnsafeNode and ConvertToSafeNode.scala to test unsafe hash join.

[zsxwing]

Need to use a new method name because the default parameter is in conflict with overloading.

[andrewor14]

can you add some java doc to explain what this method's doing?

[SparkQA]

Test build #41824 has finished for PR 8535 at commit 2ca5778.

• This patch passes all tests.
• This patch merges cleanly.
• This patch adds the following public classes (experimental):
  • case class ConvertToSafeNode(conf: SQLConf, child: LocalNode) extends UnaryLocalNode(conf)
  • case class ConvertToUnsafeNode(conf: SQLConf, child: LocalNode) extends UnaryLocalNode(conf)
  • case class FilterNode(conf: SQLConf, condition: Expression, child: LocalNode)
  • case class HashJoinNode (
  • case class LimitNode(conf: SQLConf, limit: Int, child: LocalNode) extends UnaryLocalNode(conf)
  • abstract class LocalNode(conf: SQLConf) extends TreeNode[LocalNode] with Logging
  • abstract class LeafLocalNode(conf: SQLConf) extends LocalNode(conf)
  • abstract class UnaryLocalNode(conf: SQLConf) extends LocalNode(conf)
  • abstract class BinaryLocalNode(conf: SQLConf) extends LocalNode(conf)
  • case class ProjectNode(conf: SQLConf, projectList: Seq[NamedExpression], child: LocalNode)
  • case class SeqScanNode(conf: SQLConf, output: Seq[Attribute], data: Seq[InternalRow])
  • case class UnionNode(conf: SQLConf, children: Seq[LocalNode]) extends LocalNode(conf)

[zsxwing]

@rxin do we need to make these local classes private[sql]?

[cloud-fan]

shall we just use tungstenEnabled here?

[cloud-fan]

and the ProjectNode always uses unsafe projection, should we control that by this config?

[zsxwing]

and the ProjectNode always uses unsafe projection, should we control that by this config?

Agreed.

[zsxwing]

shall we just use tungstenEnabled here?

Just followed SparkPlan.

[SparkQA]

Test build #41902 has finished for PR 8535 at commit aa928fd.

• This patch passes all tests.
• This patch merges cleanly.
• This patch adds the following public classes (experimental):
  • case class ConvertToSafeNode(conf: SQLConf, child: LocalNode) extends UnaryLocalNode(conf)
  • case class ConvertToUnsafeNode(conf: SQLConf, child: LocalNode) extends UnaryLocalNode(conf)
  • case class FilterNode(conf: SQLConf, condition: Expression, child: LocalNode)
  • case class HashJoinNode (
  • case class LimitNode(conf: SQLConf, limit: Int, child: LocalNode) extends UnaryLocalNode(conf)
  • abstract class LocalNode(conf: SQLConf) extends TreeNode[LocalNode] with Logging
  • abstract class LeafLocalNode(conf: SQLConf) extends LocalNode(conf)
  • abstract class UnaryLocalNode(conf: SQLConf) extends LocalNode(conf)
  • abstract class BinaryLocalNode(conf: SQLConf) extends LocalNode(conf)
  • case class ProjectNode(conf: SQLConf, projectList: Seq[NamedExpression], child: LocalNode)
  • case class SeqScanNode(conf: SQLConf, output: Seq[Attribute], data: Seq[InternalRow])
  • case class UnionNode(conf: SQLConf, children: Seq[LocalNode]) extends LocalNode(conf)

[andrewor14]

I actually think implementing the wrapper is better since it's not very complicated. Duplicate code in general is really bad and hard to maintain. We can have something like the following in LocalNode

def asIterator: Iterator[InternalRow] = new LocalNodeIterator(this)

then provide the dummy SQLMetrics.nullLongMetric

[andrewor14]

I wrote some code for this. Feel free to steal or come up with something better. (not tested!)

/**
 * An thin wrapper around a [[LocalNode]] that provides an iterator interface.
 */
private class LocalNodeIterator(localNode: LocalNode) extends Iterator[InternalRow] {
  private var nextRow: InternalRow = _

  override def hasNext: Boolean = {
    if (nextRow == null) {
      val res = localNode.next()
      if (res) {
        nextRow = localNode.fetch()
      }
      res
    } else {
      true
    }
  }

  override def next(): InternalRow = {
    if (hasNext) {
      val res = nextRow
      nextRow = null
      res
    } else {
      throw new NoSuchElementException
    }
  }
}

[zsxwing]

Thanks. Added LocalNodeIterator to this PR :)