Generic Node

Node.scala

package org.opencypher.caps.impl.spark.physical


case class Pipeline[I <: Node[I], O <: Node[O]](
  sequence: TreeTransformation[I, O]) {

  def chain[A](t: TreeTransformation[O, A]): Pipeline[I, A] = {

  }
}


case class PrintingTransformation[I, O](t: TreeTransformation[I, O]) extends TreeTransformation[I, O] {
  override def apply(root: Node[I]): Node[O] = {
    val msg = s"${t.getClass.getSimpleName} started ..."
    val padded = msg.padTo(30, " ").mkString("")
    print(padded)
    val r = t(root)
    println("done!")
    r
  }
}

case class Intercept[I, O](
  f: I => O,
  input: I => Unit = (_) => {},
  output: O => Unit = (_) => {}
) extends ((I) => O) {
  override def apply(i: I): O = {
    ???
  }
}

trait TreeTransformation[I, O] extends ((Node[I]) => Node[O]) {
  def apply(root: Node[I]): Node[O]

  def withPrinting: TreeTransformation[I, O] = PrintingTransformation(this)
}

case class Node[T](value: T, children: Seq[Node[T]] = Seq.empty) {
  def pretty: String = ???

  def map[O](f: T => O): Node[O] = {
    Node(f(value), children.map(c => c.map(f)))
  }

  def fold[O](initial: O)(f: (O, T) => O): O = {
    children.foldLeft(f(initial, value)) { case (agg, nextChild) =>
      nextChild.fold(agg)(f)
    }
  }

  def flatten: Seq[T] = fold[List[T]](List.empty) { case (r, n) => n :: r }
}