JoinDataSet-WithKeys.scala

class JoinDataSet[L, R](lhs: Dataset[L], rhs: Dataset[R], joinType: String) {
  
  def withKey[K](column: Witness.Lt[Symbol])(implicit
                                             lhsExists: PropertyExists[L, column.T, K],
                                             rhsExists: PropertyExists[R, column.T, K]): DataFrame =
    doJoin(Seq(column.value.name))

  private def doJoin(columns: Seq[String]) = 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")
}