[SPARK-9996][SPARK-9997][SQL]Add local expand and NestedLoopJoin operators

[zsxwing]

This PR is in conflict with #8535 and #8573. Will update this one when they are merged.

[SparkQA]

Test build #42095 has finished for PR 8642 at commit 9055d8a.

• 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 ExpandNode(
  • case class FilterNode(conf: SQLConf, condition: Expression, child: LocalNode)
  • 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 NestedLoopJoinNode(
  • 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]

hm, at this point input hasn't been initialized yet has it? Did you intend to pass null into the projection?

[SparkQA]

Test build #42330 has finished for PR 8642 at commit 514d3d2.

• This patch passes all tests.
• This patch merges cleanly.
• This patch adds the following public classes (experimental):
  • case class ExpandNode(
  • case class NestedLoopJoinNode(

[SparkQA]

Test build #42336 has finished for PR 8642 at commit 4b18418.

• This patch passes all tests.
• This patch merges cleanly.
• This patch adds the following public classes (experimental):
  • case class ExpandNode(
  • case class NestedLoopJoinNode(

[andrewor14]

how about:

if (idx < 0 || idx >= groups.length) {
  if (child.next()) {
    input = child.fetch()
    idx = 0
  } else {
    return false
  }
}
result = groups(idx)(input)
idx += 1
true

a little less duplication

[andrewor14]

@zsxwing I left some suggestions on reducing code duplication, but otherwise this looks pretty good.

[zsxwing]

@andrewor14 thanks for your reviewing. I updated this PR as per your comments.

[SparkQA]

Test build #42421 has finished for PR 8642 at commit c6e80a2.

• This patch passes all tests.
• This patch merges cleanly.
• This patch adds the following public classes (experimental):
  • case class ExpandNode(
  • case class NestedLoopJoinNode(

[andrewor14]

LGTM, thanks I'm merging this into master.