merge: Introduce schema for trace API (#88)

This pull request updates the OpenDC trace API to support proper specification 
of a schema of the tables exposed by the traces. This functionality makes it easier 
for the API consumer to understand the types exposed by the API.

## Implementation Notes :hammer_and_pick:

* Introduce type system for trace API
* Add benchmarks for odcvm trace format 
* Add benchmarks for Azure trace format
* Add conformance suite for OpenDC trace API 

## External Dependencies :four_leaf_clover:

* N/A

## Breaking API Changes :warning:

* Removal of typed `TableColumn`. Instead, `TableColumn` instances are now 
   used to describe the columns belonging to some table.
* `TableReader` and `TableWriter` do not support accessing arbitrary objects 
   anymore. Instead, only the types supported by the type system are exposed.

2 files changed, 80 insertions, 30 deletions

diff --git a/opendc-trace/opendc-trace-wfformat/src/main/kotlin/org/opendc/trace/wfformat/WfFormatTaskTableReader.kt b/opendc-trace/opendc-trace-wfformat/src/main/kotlin/org/opendc/trace/wfformat/WfFormatTaskTableReader.kt
index d8eafa9c..ca1a29d0 100644
--- a/opendc-trace/opendc-trace-wfformat/src/main/kotlin/org/opendc/trace/wfformat/WfFormatTaskTableReader.kt
+++ b/opendc-trace/opendc-trace-wfformat/src/main/kotlin/org/opendc/trace/wfformat/WfFormatTaskTableReader.kt
@@ -27,7 +27,10 @@ import com.fasterxml.jackson.core.JsonParser
 import com.fasterxml.jackson.core.JsonToken
 import org.opendc.trace.*
 import org.opendc.trace.conv.*
+import org.opendc.trace.util.convertTo
 import java.time.Duration
+import java.time.Instant
+import java.util.*
 import kotlin.math.roundToInt
 
 /**
@@ -51,6 +54,7 @@ internal class WfFormatTaskTableReader(private val parser: JsonParser) : TableRe
 
                     // Check whether the document is not empty and starts with an object
                     if (token == null) {
+                        parser.close()
                         break
                     } else if (token != JsonToken.START_OBJECT) {
                         throw JsonParseException(parser, "Expected object", parser.currentLocation)
@@ -61,6 +65,7 @@ internal class WfFormatTaskTableReader(private val parser: JsonParser) : TableRe
                 ParserLevel.TRACE -> {
                     // Seek for the workflow object in the file
                     if (!seekWorkflow()) {
+                        parser.close()
                         break
                     } else if (!parser.isExpectedStartObjectToken) {
                         throw JsonParseException(parser, "Expected object", parser.currentLocation)
@@ -95,41 +100,86 @@ internal class WfFormatTaskTableReader(private val parser: JsonParser) : TableRe
         return hasJob
     }
 
-    override fun resolve(column: TableColumn<*>): Int = columns[column] ?: -1
+    override fun resolve(name: String): Int {
+        return when (name) {
+            TASK_ID -> COL_ID
+            TASK_WORKFLOW_ID -> COL_WORKFLOW_ID
+            TASK_RUNTIME -> COL_RUNTIME
+            TASK_REQ_NCPUS -> COL_NPROC
+            TASK_PARENTS -> COL_PARENTS
+            TASK_CHILDREN -> COL_CHILDREN
+            else -> -1
+        }
+    }
 
     override fun isNull(index: Int): Boolean {
-        check(index in 0..columns.size) { "Invalid column value" }
+        require(index in 0..COL_CHILDREN) { "Invalid column value" }
         return false
     }
 
-    override fun get(index: Int): Any? {
+    override fun getBoolean(index: Int): Boolean {
+        throw IllegalArgumentException("Invalid column")
+    }
+
+    override fun getInt(index: Int): Int {
+        checkActive()
+        return when (index) {
+            COL_NPROC -> cores
+            else -> throw IllegalArgumentException("Invalid column")
+        }
+    }
+
+    override fun getLong(index: Int): Long {
+        throw IllegalArgumentException("Invalid column")
+    }
+
+    override fun getFloat(index: Int): Float {
+        throw IllegalArgumentException("Invalid column")
+    }
+
+    override fun getDouble(index: Int): Double {
+        throw IllegalArgumentException("Invalid column")
+    }
+
+    override fun getString(index: Int): String? {
+        checkActive()
         return when (index) {
             COL_ID -> id
             COL_WORKFLOW_ID -> workflowId
-            COL_RUNTIME -> runtime
-            COL_PARENTS -> parents
-            COL_CHILDREN -> children
-            COL_NPROC -> getInt(index)
             else -> throw IllegalArgumentException("Invalid column")
         }
     }
 
-    override fun getBoolean(index: Int): Boolean {
+    override fun getUUID(index: Int): UUID? {
         throw IllegalArgumentException("Invalid column")
     }
 
-    override fun getInt(index: Int): Int {
+    override fun getInstant(index: Int): Instant? {
+        throw IllegalArgumentException("Invalid column")
+    }
+
+    override fun getDuration(index: Int): Duration? {
+        checkActive()
         return when (index) {
-            COL_NPROC -> cores
+            COL_RUNTIME -> runtime
             else -> throw IllegalArgumentException("Invalid column")
         }
     }
 
-    override fun getLong(index: Int): Long {
+    override fun <T> getList(index: Int, elementType: Class<T>): List<T>? {
         throw IllegalArgumentException("Invalid column")
     }
 
-    override fun getDouble(index: Int): Double {
+    override fun <T> getSet(index: Int, elementType: Class<T>): Set<T>? {
+        checkActive()
+        return when (index) {
+            COL_PARENTS -> TYPE_PARENTS.convertTo(parents, elementType)
+            COL_CHILDREN -> TYPE_CHILDREN.convertTo(children, elementType)
+            else -> throw IllegalArgumentException("Invalid column")
+        }
+    }
+
+    override fun <K, V> getMap(index: Int, keyType: Class<K>, valueType: Class<V>): Map<K, V>? {
         throw IllegalArgumentException("Invalid column")
     }
 
@@ -138,10 +188,17 @@ internal class WfFormatTaskTableReader(private val parser: JsonParser) : TableRe
     }
 
     /**
+     * Helper method to check if the reader is active.
+     */
+    private fun checkActive() {
+        check(level != ParserLevel.TOP && !parser.isClosed) { "No active row. Did you call nextRow()?" }
+    }
+
+    /**
      * Parse the trace and seek until the workflow description.
      */
     private fun seekWorkflow(): Boolean {
-        while (parser.nextValue() != JsonToken.END_OBJECT) {
+        while (parser.nextValue() != JsonToken.END_OBJECT && !parser.isClosed) {
             when (parser.currentName) {
                 "name" -> workflowId = parser.text
                 "workflow" -> return true
@@ -232,12 +289,6 @@ internal class WfFormatTaskTableReader(private val parser: JsonParser) : TableRe
     private val COL_PARENTS = 5
     private val COL_CHILDREN = 6
 
-    private val columns = mapOf(
-        TASK_ID to COL_ID,
-        TASK_WORKFLOW_ID to COL_WORKFLOW_ID,
-        TASK_RUNTIME to COL_RUNTIME,
-        TASK_REQ_NCPUS to COL_NPROC,
-        TASK_PARENTS to COL_PARENTS,
-        TASK_CHILDREN to COL_CHILDREN,
-    )
+    private val TYPE_PARENTS = TableColumnType.Set(TableColumnType.String)
+    private val TYPE_CHILDREN = TableColumnType.Set(TableColumnType.String)
 }
diff --git a/opendc-trace/opendc-trace-wfformat/src/main/kotlin/org/opendc/trace/wfformat/WfFormatTraceFormat.kt b/opendc-trace/opendc-trace-wfformat/src/main/kotlin/org/opendc/trace/wfformat/WfFormatTraceFormat.kt
index 8db4c169..154fa061 100644
--- a/opendc-trace/opendc-trace-wfformat/src/main/kotlin/org/opendc/trace/wfformat/WfFormatTraceFormat.kt
+++ b/opendc-trace/opendc-trace-wfformat/src/main/kotlin/org/opendc/trace/wfformat/WfFormatTraceFormat.kt
@@ -50,20 +50,19 @@ public class WfFormatTraceFormat : TraceFormat {
         return when (table) {
             TABLE_TASKS -> TableDetails(
                 listOf(
-                    TASK_ID,
-                    TASK_WORKFLOW_ID,
-                    TASK_RUNTIME,
-                    TASK_REQ_NCPUS,
-                    TASK_PARENTS,
-                    TASK_CHILDREN
-                ),
-                emptyList()
+                    TableColumn(TASK_ID, TableColumnType.String),
+                    TableColumn(TASK_WORKFLOW_ID, TableColumnType.String),
+                    TableColumn(TASK_RUNTIME, TableColumnType.Duration),
+                    TableColumn(TASK_REQ_NCPUS, TableColumnType.Int),
+                    TableColumn(TASK_PARENTS, TableColumnType.Set(TableColumnType.String)),
+                    TableColumn(TASK_CHILDREN, TableColumnType.Set(TableColumnType.String))
+                )
             )
             else -> throw IllegalArgumentException("Table $table not supported")
         }
     }
 
-    override fun newReader(path: Path, table: String, projection: List<TableColumn<*>>?): TableReader {
+    override fun newReader(path: Path, table: String, projection: List<String>?): TableReader {
         return when (table) {
             TABLE_TASKS -> WfFormatTaskTableReader(factory.createParser(path.toFile()))
             else -> throw IllegalArgumentException("Table $table not supported")