Read input as JSON, enrich DFG with metadata

Author: segeljakt

build.sbt

@@ -18,7 +18,8 @@ lazy val compilerPipeline= (project in file("."))
        "com.lihaoyi" %% "pprint" % "0.5.5",
        "com.chuusai" %% "shapeless" % "2.3.3",
        "io.circe" %% "circe-core" % "0.11.1",
-       "io.circe" %% "circe-generic" % "0.11.1"
+       "io.circe" %% "circe-generic" % "0.11.1",
+       "io.circe" %% "circe-parser" % "0.11.1"
     )
   ).dependsOn(arc)
 
@@ -77,7 +78,7 @@ scalacOptions ++= Seq(
   "-language:implicitConversions", // Allow definition of implicit functions called views
   "-unchecked", // Enable additional warnings where generated code depends on assumptions.
   "-Xcheckinit", // Wrap field accessors to throw an exception on uninitialized access.
-  "-Xdev", // Indicates user is a developer - issue warnings about anything which seems amiss
+  //"-Xdev", // Indicates user is a developer - issue warnings about anything which seems amiss
   "-Xlint:_", // Enable all lint warnings
   "-Xfuture", // Turn on future language features.
   "-Yno-adapted-args", // Do not adapt an argument list (either by inserting () or creating a tuple) to match the receiver.

src/main/scala/se/kth/cda/compiler/Compiler.scala

@@ -1,32 +1,44 @@
 package se.kth.cda.compiler
 
+import io.circe.Json
 import org.antlr.v4.runtime.{CharStreams, CommonTokenStream}
 import se.kth.cda.arc.syntaxtree.PrettyPrint
 import se.kth.cda.arc.syntaxtree.parser.Translator
 import se.kth.cda.arc.syntaxtree.transformer.MacroExpansion
 import se.kth.cda.arc.syntaxtree.typer.TypeInference
 import se.kth.cda.arc.{ArcLexer, ArcParser}
 import se.kth.cda.compiler.dataflow.transform.ToDFG._
-import se.kth.cda.compiler.dataflow.JsonEncoder.encodeDFG
+import se.kth.cda.compiler.dataflow.encode.EncodeDFG.encodeDFG
+import se.kth.cda.compiler.dataflow.decode.DecodeMetadata.metadataDecoder
+import se.kth.cda.compiler.dataflow.Metadata
 import se.kth.cda.compiler.dataflow.optimize.OptimizeDFG._
+import se.kth.cda.compiler.dataflow.enrich.EnrichDFG._
 
 object Compiler {
-  def compile(code: String): String = {
+  def compile(input: String): String = {
 
-    val inputStream = CharStreams.fromString(code)
+    val metadata = metadataDecoder(io.circe.parser.parse(input).getOrElse(Json.Null).hcursor) match {
+    //val metadata = metadataDecoder.decodeJson(Json.fromString(input)) match {
+      case Left(value) => throw value
+      case Right(value) => value
+    }
+
+    val inputStream = CharStreams.fromString(metadata.arc_code)
     val lexer = new ArcLexer(inputStream)
     val tokenStream = new CommonTokenStream(lexer)
-    val parser = new ArcParser(tokenStream)
-    val translator = Translator(parser)
+    val arcparser = new ArcParser(tokenStream)
+    val translator = Translator(arcparser)
     val ast = translator.expr()
     val expanded = MacroExpansion.expand(ast).get
     val typed = TypeInference.solve(expanded).get
     val dfg = typed.toDFG
-    val optimized = dfg.optimize
+
+    val enriched_dfg = dfg.enrich(metadata)
+    val optimized_dfg = enriched_dfg.optimize
 
     //println(PrettyPrint.pretty(typed))
     //pprint.pprintln(optimized)
 
-    encodeDFG(optimized).noSpaces
+    encodeDFG(optimized_dfg).noSpaces
   }
 }

src/main/scala/se/kth/cda/compiler/dataflow/DFG.scala

@@ -3,7 +3,6 @@ package se.kth.cda.compiler.dataflow
 import se.kth.cda.arc.syntaxtree.AST.Expr
 import se.kth.cda.arc.syntaxtree.Type
 import se.kth.cda.compiler.dataflow.ChannelStrategy._
-import se.kth.cda.compiler.dataflow.DFG.newId
 import se.kth.cda.compiler.dataflow.SinkKind.Debug
 import se.kth.cda.compiler.dataflow.SourceKind.Socket
 import se.kth.cda.compiler.dataflow.TimeKind.Ingestion
@@ -28,17 +27,11 @@ import se.kth.cda.compiler.dataflow.WindowKind.All
 //  val trace: Option[Trace] = None
 //  val scope: Option[Scope] = None
 //}
+trait Trace
 
-object DFG {
-  var idCounter = 0
-  def newId: Int = {
-    val id = idCounter
-    idCounter += 1
-    id
-  }
-}
+final case class Metadata(nodes: List[Node], timestamp_extractor: Int, arc_code: String)
 
-object Node {
+object DFG {
   var idCounter = 0
   def newId: Int = {
     val id = idCounter
@@ -56,23 +49,65 @@ object Channel {
   }
 }
 
-trait Trace
-
-final case class DFG(id: String = s"dfg${DFG.newId}", var nodes: List[Node], target: String = "x86-64-unknown-linux-gnu")
+final case class DFG(id: String = s"dfg${DFG.newId}",
+                     var timestamp_extractor: Int = 1,
+                     var nodes: List[Node],
+                     target: String = "x86-64-unknown-linux-gnu")
 
 //case class Scope(depth: Long, parent: Option[Scope]) extends Id
 
-final case class Node(var id: String = s"node${Node.newId}", parallelism: Long = 1, kind: NodeKind)
+final case class Node(var id: String, parallelism: Long = 1, kind: NodeKind)
 
 sealed trait NodeKind
 
 object NodeKind {
-  final case class Source(sourceType: Type,
+  object Task {
+    var idCounter = 0
+    def newId: Int = {
+      val id = idCounter
+      idCounter += 1
+      id
+    }
+  }
+
+  object Source {
+    var idCounter = 0
+    def newId: Int = {
+      val id = idCounter
+      idCounter += 1
+      id
+    }
+  }
+
+  object Sink {
+    var idCounter = 0
+    def newId: Int = {
+      val id = idCounter
+      idCounter += 1
+      id
+    }
+  }
+
+  object Window {
+    var idCounter = 0
+    def newId: Int = {
+      val id = idCounter
+      idCounter += 1
+      id
+    }
+  }
+
+  final case class Source(sourceType: Type = null,
+                          var format: Format = null,
                           channelStrategy: ChannelStrategy = Forward,
                           var successors: Vector[ChannelKind] = Vector.empty,
-                          kind: SourceKind = Socket("localhost", 1337))
+                          var kind: SourceKind = null)
+      extends NodeKind
+  final case class Sink(sinkType: Type = null,
+                        var format: Format = null,
+                        var predecessor: Node = null,
+                        var kind: SinkKind = null)
       extends NodeKind
-  final case class Sink(sinkType: Type, var predecessor: Node = null, kind: SinkKind = Debug) extends NodeKind
   final case class Task(var weldFunc: Expr,
                         inputType: Type,
                         outputType: Type,
@@ -92,7 +127,12 @@ object NodeKind {
       extends NodeKind
 }
 
-final case class WindowFunction(inputType: Type, outputType: Type, builderType: Type, init: Expr, lift: Expr, lower: Expr)
+final case class WindowFunction(inputType: Type,
+                                outputType: Type,
+                                builderType: Type,
+                                init: Expr,
+                                lift: Expr,
+                                lower: Expr)
 
 sealed trait ChannelKind
 
@@ -104,13 +144,24 @@ object ChannelKind {
 sealed trait SourceKind
 
 object SourceKind {
-  final case class Socket(host: String, port: Long) extends SourceKind
+  final case class Socket(var host: String, var port: Long) extends SourceKind
+  final case class LocalFile(var path: String) extends SourceKind
 }
 
 sealed trait SinkKind
 
 object SinkKind {
   final case object Debug extends SinkKind
+  final case class Socket(var host: String, var port: Long) extends SinkKind
+  final case class LocalFile(var path: String) extends SinkKind
+}
+
+sealed trait Format
+
+object Format {
+  final case object JSON extends Format
+  final case object CSV extends Format
+  final case object UTF8 extends Format
 }
 
 sealed trait TaskKind

src/main/scala/se/kth/cda/compiler/dataflow/decode/DecodeMetadata.scala

@@ -0,0 +1,82 @@
+package se.kth.cda.compiler.dataflow.decode
+
+import se.kth.cda.compiler.dataflow.NodeKind.{Task, Window}
+import se.kth.cda.compiler.dataflow._
+
+object DecodeMetadata {
+  import io.circe._
+  import io.circe.generic.semiauto._
+
+  implicit val metadataDecoder: Decoder[Metadata] =
+    (cursor: HCursor) =>
+      for {
+        nodes <- cursor.get[List[Node]]("nodes")
+        timestamp_extractor <- cursor.get[Int]("timestamp_extractor")
+        age <- cursor.get[String]("arc_code")
+      } yield Metadata(nodes, timestamp_extractor, age)
+
+  implicit val nodeDecoder: Decoder[Node] =
+    (cursor: HCursor) =>
+      for {
+        id <- cursor.get[String]("id")
+        kind <- cursor.get[NodeKind]("kind")
+      } yield Node(id = id, kind = kind)
+  implicit val nodeKindDecoder: Decoder[NodeKind] = io.circe.generic.semiauto.deriveDecoder[NodeKind]
+
+  implicit val sourceDecoder: Decoder[NodeKind.Source] =
+    (cursor: HCursor) =>
+      for {
+        format <- cursor.get[String]("format").map{
+          case "CSV" => Format.CSV
+          case "UTF8" => Format.UTF8
+          case "JSON" => Format.JSON
+        }
+        kind <- cursor.get[SourceKind]("kind")
+      } yield NodeKind.Source(format = format, kind = kind)
+  implicit val sourceKindDecoder: Decoder[SourceKind] = deriveDecoder
+
+  implicit val localFileSourceDecoder: Decoder[SourceKind.LocalFile] =
+    (cursor: HCursor) =>
+      for {
+        path <- cursor.get[String]("path")
+      } yield SourceKind.LocalFile(path)
+
+  implicit val socketSourceDecoder: Decoder[SourceKind.Socket] =
+    (cursor: HCursor) =>
+      for {
+        host <- cursor.get[String]("host")
+        port <- cursor.get[Long]("port")
+      } yield SourceKind.Socket(host, port)
+
+  implicit val sinkDecoder: Decoder[NodeKind.Sink] =
+    (cursor: HCursor) =>
+      for {
+        format <- cursor.get[String]("format").map{
+          case "CSV" => Format.CSV
+          case "UTF8" => Format.UTF8
+          case "JSON" => Format.JSON
+        }
+        kind <- cursor.get[SinkKind]("kind")
+      } yield NodeKind.Sink(format = format, kind = kind)
+  implicit val sinkKindDecoder: Decoder[SinkKind] = deriveDecoder
+
+  implicit val localFileSinkDecoder: Decoder[SinkKind.LocalFile] =
+    (cursor: HCursor) => for { path <- cursor.get[String]("path") } yield SinkKind.LocalFile(path)
+
+  implicit val socketSinkDecoder: Decoder[SinkKind.Socket] =
+    (cursor: HCursor) =>
+      for {
+        host <- cursor.get[String]("host")
+        port <- cursor.get[Long]("port")
+      } yield SinkKind.Socket(host, port)
+
+  implicit val formatDecoder: Decoder[Format] = deriveDecoder
+
+  implicit val windowDecoder: Decoder[Window] = (cursor: HCursor) => for {
+    _ <- cursor.get[String]("name")
+  } yield Window(null, null, null, null, null, null, null)
+
+  implicit val taskDecoder: Decoder[Task] = (cursor: HCursor) => for {
+    _ <- cursor.get[String]("name")
+  } yield Task(null, null, null, null)
+}