JoinDataSet-on.scala

class JoinDataSet[L, R](lhs: Dataset[L], rhs: Dataset[R], joinType: String) {
  object on extends SingletonProductArgs {
    def applyProduct[V <: HList, K](columns: V)(implicit
                                                i0: ToTraversable.Aux[V, List, Symbol],
                                                lhsExists: PropertiesExists[L, V, K],
                                                rhsExists: PropertiesExists[R, V, K]): DataFrame =
      doJoin(columns.toList[Symbol].map(_.name))

  }

  private def doJoin(columns: Seq[String]) = lhs.join(rhs, columns, joinType)
  private def doJoin(columns: Column)      = lhs.join(rhs, columns, joinType)

}

object JoinDataSet {
  def apply[L, R](lhs: Dataset[L], rhs: Dataset[R], joinType: String) = new JoinDataSet(lhs, rhs, joinType)
  def leftJoin[L, R](lhs: Dataset[L], rhs: Dataset[R])                = JoinDataSet(lhs, rhs, "leftOuter")
  def rightJoin[L, R](lhs: Dataset[L], rhs: Dataset[R])               = JoinDataSet(lhs, rhs, "rightOuter")
  def fullOuterJoin[L, R](lhs: Dataset[L], rhs: Dataset[R])           = JoinDataSet(lhs, rhs, "fullOuter")
}