package org.alfiler

import frameless.{Injection, TypedDataset}
import org.alfiler.Algo.{Name, SecondName}
import org.apache.spark.sql.SparkSession
import shapeless.tag
import shapeless.tag.@@


object Algo {
  trait NameTag
  type Name = String @@ NameTag
  trait SecondNameTag
  type SecondName = String @@ SecondNameTag

  def asSecondName(s:String):SecondName = tag[SecondNameTag][String](s)

  def create(name: String, surname: String): Algo = new Algo(tag[NameTag][String](name), tag[SecondNameTag][String](surname))
}

case class Algo(name:Name, secondName:SecondName)

object Main {

  implicit def dateToLongInjection[T] = new Injection[String @@ T, String] {
    def apply(d: String @@ T): String = d.asInstanceOf[String]
    def invert(l: String): String @@ T = tag[T][String](l)
  }


  def main(args: Array[String]): Unit = {
    implicit val ss = SparkSession.builder().appName("frameless").master("local[*]").getOrCreate()

    import frameless.syntax._
    val data = Seq(
      Algo.create("Stephen","Hawking"),
      Algo.create("Albert","Einstein")
    )
    val aptTypedDs = TypedDataset.create(data)

    def doubleSurname(a:SecondName):SecondName = Algo.asSecondName(a+a)

    aptTypedDs.filter(_.secondName.startsWith("H")).map(c => c.copy(secondName = doubleSurname(c.secondName))).show().run()
    //aptTypedDs.filter(_.secondName.startsWith("H")).map(c => c.copy(secondName = doubleSurname(c.name))).show().run()
  }

}