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.

7 files changed, 331 insertions, 119 deletions

diff --git a/opendc-trace/opendc-trace-bitbrains/src/main/kotlin/org/opendc/trace/bitbrains/BitbrainsExResourceStateTableReader.kt b/opendc-trace/opendc-trace-bitbrains/src/main/kotlin/org/opendc/trace/bitbrains/BitbrainsExResourceStateTableReader.kt
index 1e1d1a09..df7a1c91 100644
--- a/opendc-trace/opendc-trace-bitbrains/src/main/kotlin/org/opendc/trace/bitbrains/BitbrainsExResourceStateTableReader.kt
+++ b/opendc-trace/opendc-trace-bitbrains/src/main/kotlin/org/opendc/trace/bitbrains/BitbrainsExResourceStateTableReader.kt
@@ -25,20 +25,37 @@ package org.opendc.trace.bitbrains
 import org.opendc.trace.*
 import org.opendc.trace.conv.*
 import java.io.BufferedReader
+import java.time.Duration
 import java.time.Instant
+import java.util.*
 
 /**
  * A [TableReader] for the Bitbrains resource state table.
  */
 internal class BitbrainsExResourceStateTableReader(private val reader: BufferedReader) : TableReader {
+    private var state = State.Pending
+
     override fun nextRow(): Boolean {
+        val state = state
+        if (state == State.Closed) {
+            return false
+        } else if (state == State.Pending) {
+            this.state = State.Active
+        }
+
         reset()
 
-        var line: String
+        var line: String?
         var num = 0
 
         while (true) {
-            line = reader.readLine() ?: return false
+            line = reader.readLine()
+
+            if (line == null) {
+                this.state = State.Closed
+                return false
+            }
+
             num++
 
             if (line[0] == '#' || line.isBlank()) {
@@ -49,7 +66,7 @@ internal class BitbrainsExResourceStateTableReader(private val reader: BufferedR
             break
         }
 
-        line = line.trim()
+        line = line!!.trim()
 
         val length = line.length
         var col = 0
@@ -89,26 +106,31 @@ internal class BitbrainsExResourceStateTableReader(private val reader: BufferedR
         return true
     }
 
-    override fun resolve(column: TableColumn<*>): Int = columns[column] ?: -1
+    override fun resolve(name: String): Int {
+        return when (name) {
+            RESOURCE_ID -> COL_ID
+            RESOURCE_CLUSTER_ID -> COL_CLUSTER_ID
+            RESOURCE_STATE_TIMESTAMP -> COL_TIMESTAMP
+            RESOURCE_CPU_COUNT -> COL_NCPUS
+            RESOURCE_CPU_CAPACITY -> COL_CPU_CAPACITY
+            RESOURCE_STATE_CPU_USAGE -> COL_CPU_USAGE
+            RESOURCE_STATE_CPU_USAGE_PCT -> COL_CPU_USAGE_PCT
+            RESOURCE_STATE_CPU_DEMAND -> COL_CPU_DEMAND
+            RESOURCE_STATE_CPU_READY_PCT -> COL_CPU_READY_PCT
+            RESOURCE_MEM_CAPACITY -> COL_MEM_CAPACITY
+            RESOURCE_STATE_DISK_READ -> COL_DISK_READ
+            RESOURCE_STATE_DISK_WRITE -> COL_DISK_WRITE
+            else -> -1
+        }
+    }
 
     override fun isNull(index: Int): Boolean {
-        require(index in 0..COL_MAX) { "Invalid column index" }
+        require(index in 0 until COL_MAX) { "Invalid column index" }
         return false
     }
 
-    override fun get(index: Int): Any? {
-        return when (index) {
-            COL_ID -> id
-            COL_CLUSTER_ID -> cluster
-            COL_TIMESTAMP -> timestamp
-            COL_NCPUS -> getInt(index)
-            COL_POWERED_ON -> getInt(index)
-            COL_CPU_CAPACITY, COL_CPU_USAGE, COL_CPU_USAGE_PCT, COL_CPU_READY_PCT, COL_CPU_DEMAND, COL_MEM_CAPACITY, COL_DISK_READ, COL_DISK_WRITE -> getDouble(index)
-            else -> throw IllegalArgumentException("Invalid column")
-        }
-    }
-
     override fun getBoolean(index: Int): Boolean {
+        check(state == State.Active) { "No active row" }
         return when (index) {
             COL_POWERED_ON -> poweredOn
             else -> throw IllegalArgumentException("Invalid column")
@@ -116,6 +138,7 @@ internal class BitbrainsExResourceStateTableReader(private val reader: BufferedR
     }
 
     override fun getInt(index: Int): Int {
+        check(state == State.Active) { "No active row" }
         return when (index) {
             COL_NCPUS -> cpuCores
             else -> throw IllegalArgumentException("Invalid column")
@@ -126,7 +149,12 @@ internal class BitbrainsExResourceStateTableReader(private val reader: BufferedR
         throw IllegalArgumentException("Invalid column")
     }
 
+    override fun getFloat(index: Int): Float {
+        throw IllegalArgumentException("Invalid column")
+    }
+
     override fun getDouble(index: Int): Double {
+        check(state == State.Active) { "No active row" }
         return when (index) {
             COL_CPU_CAPACITY -> cpuCapacity
             COL_CPU_USAGE -> cpuUsage
@@ -140,8 +168,47 @@ internal class BitbrainsExResourceStateTableReader(private val reader: BufferedR
         }
     }
 
+    override fun getString(index: Int): String? {
+        check(state == State.Active) { "No active row" }
+        return when (index) {
+            COL_ID -> id
+            COL_CLUSTER_ID -> cluster
+            else -> throw IllegalArgumentException("Invalid column")
+        }
+    }
+
+    override fun getUUID(index: Int): UUID? {
+        throw IllegalArgumentException("Invalid column")
+    }
+
+    override fun getInstant(index: Int): Instant? {
+        check(state == State.Active) { "No active row" }
+        return when (index) {
+            COL_TIMESTAMP -> timestamp
+            else -> throw IllegalArgumentException("Invalid column")
+        }
+    }
+
+    override fun getDuration(index: Int): Duration? {
+        throw IllegalArgumentException("Invalid column")
+    }
+
+    override fun <T> getList(index: Int, elementType: Class<T>): List<T>? {
+        throw IllegalArgumentException("Invalid column")
+    }
+
+    override fun <T> getSet(index: Int, elementType: Class<T>): Set<T>? {
+        throw IllegalArgumentException("Invalid column")
+    }
+
+    override fun <K, V> getMap(index: Int, keyType: Class<K>, valueType: Class<V>): Map<K, V>? {
+        throw IllegalArgumentException("Invalid column")
+    }
+
     override fun close() {
         reader.close()
+        reset()
+        state = State.Closed
     }
 
     /**
@@ -196,18 +263,7 @@ internal class BitbrainsExResourceStateTableReader(private val reader: BufferedR
     private val COL_CPU_USAGE_PCT = 21
     private val COL_MAX = COL_CPU_USAGE_PCT + 1
 
-    private val columns = mapOf(
-        RESOURCE_ID to COL_ID,
-        RESOURCE_CLUSTER_ID to COL_CLUSTER_ID,
-        RESOURCE_STATE_TIMESTAMP to COL_TIMESTAMP,
-        RESOURCE_CPU_COUNT to COL_NCPUS,
-        RESOURCE_CPU_CAPACITY to COL_CPU_CAPACITY,
-        RESOURCE_STATE_CPU_USAGE to COL_CPU_USAGE,
-        RESOURCE_STATE_CPU_USAGE_PCT to COL_CPU_USAGE_PCT,
-        RESOURCE_STATE_CPU_DEMAND to COL_CPU_DEMAND,
-        RESOURCE_STATE_CPU_READY_PCT to COL_CPU_READY_PCT,
-        RESOURCE_MEM_CAPACITY to COL_MEM_CAPACITY,
-        RESOURCE_STATE_DISK_READ to COL_DISK_READ,
-        RESOURCE_STATE_DISK_WRITE to COL_DISK_WRITE
-    )
+    private enum class State {
+        Pending, Active, Closed
+    }
 }
diff --git a/opendc-trace/opendc-trace-bitbrains/src/main/kotlin/org/opendc/trace/bitbrains/BitbrainsExTraceFormat.kt b/opendc-trace/opendc-trace-bitbrains/src/main/kotlin/org/opendc/trace/bitbrains/BitbrainsExTraceFormat.kt
index 82e454ad..31c4f1e2 100644
--- a/opendc-trace/opendc-trace-bitbrains/src/main/kotlin/org/opendc/trace/bitbrains/BitbrainsExTraceFormat.kt
+++ b/opendc-trace/opendc-trace-bitbrains/src/main/kotlin/org/opendc/trace/bitbrains/BitbrainsExTraceFormat.kt
@@ -53,26 +53,25 @@ public class BitbrainsExTraceFormat : TraceFormat {
         return when (table) {
             TABLE_RESOURCE_STATES -> TableDetails(
                 listOf(
-                    RESOURCE_ID,
-                    RESOURCE_CLUSTER_ID,
-                    RESOURCE_STATE_TIMESTAMP,
-                    RESOURCE_CPU_COUNT,
-                    RESOURCE_CPU_CAPACITY,
-                    RESOURCE_STATE_CPU_USAGE,
-                    RESOURCE_STATE_CPU_USAGE_PCT,
-                    RESOURCE_STATE_CPU_DEMAND,
-                    RESOURCE_STATE_CPU_READY_PCT,
-                    RESOURCE_MEM_CAPACITY,
-                    RESOURCE_STATE_DISK_READ,
-                    RESOURCE_STATE_DISK_WRITE
-                ),
-                listOf(RESOURCE_ID, RESOURCE_STATE_TIMESTAMP)
+                    TableColumn(RESOURCE_ID, TableColumnType.String),
+                    TableColumn(RESOURCE_CLUSTER_ID, TableColumnType.String),
+                    TableColumn(RESOURCE_STATE_TIMESTAMP, TableColumnType.Instant),
+                    TableColumn(RESOURCE_CPU_COUNT, TableColumnType.Int),
+                    TableColumn(RESOURCE_CPU_CAPACITY, TableColumnType.Double),
+                    TableColumn(RESOURCE_STATE_CPU_USAGE, TableColumnType.Double),
+                    TableColumn(RESOURCE_STATE_CPU_USAGE_PCT, TableColumnType.Double),
+                    TableColumn(RESOURCE_STATE_CPU_DEMAND, TableColumnType.Double),
+                    TableColumn(RESOURCE_STATE_CPU_READY_PCT, TableColumnType.Double),
+                    TableColumn(RESOURCE_MEM_CAPACITY, TableColumnType.Double),
+                    TableColumn(RESOURCE_STATE_DISK_READ, TableColumnType.Double),
+                    TableColumn(RESOURCE_STATE_DISK_WRITE, TableColumnType.Double),
+                )
             )
             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_RESOURCE_STATES -> newResourceStateReader(path)
             else -> throw IllegalArgumentException("Table $table not supported")
diff --git a/opendc-trace/opendc-trace-bitbrains/src/main/kotlin/org/opendc/trace/bitbrains/BitbrainsResourceStateTableReader.kt b/opendc-trace/opendc-trace-bitbrains/src/main/kotlin/org/opendc/trace/bitbrains/BitbrainsResourceStateTableReader.kt
index 214fd749..4d8cf758 100644
--- a/opendc-trace/opendc-trace-bitbrains/src/main/kotlin/org/opendc/trace/bitbrains/BitbrainsResourceStateTableReader.kt
+++ b/opendc-trace/opendc-trace-bitbrains/src/main/kotlin/org/opendc/trace/bitbrains/BitbrainsResourceStateTableReader.kt
@@ -29,6 +29,7 @@ import com.fasterxml.jackson.dataformat.csv.CsvSchema
 import org.opendc.trace.*
 import org.opendc.trace.conv.*
 import java.text.NumberFormat
+import java.time.Duration
 import java.time.Instant
 import java.time.LocalDateTime
 import java.time.ZoneOffset
@@ -41,6 +42,11 @@ import java.util.*
  */
 internal class BitbrainsResourceStateTableReader(private val partition: String, private val parser: CsvParser) : TableReader {
     /**
+     * A flag to indicate whether a single row has been read already.
+     */
+    private var isStarted = false
+
+    /**
      * The [DateTimeFormatter] used to parse the timestamps in case of the Materna trace.
      */
     private val formatter = DateTimeFormatter.ofPattern("dd.MM.yyyy HH:mm:ss")
@@ -65,6 +71,10 @@ internal class BitbrainsResourceStateTableReader(private val partition: String,
     }
 
     override fun nextRow(): Boolean {
+        if (!isStarted) {
+            isStarted = true
+        }
+
         // Reset the row state
         reset()
 
@@ -112,21 +122,40 @@ internal class BitbrainsResourceStateTableReader(private val partition: String,
         return true
     }
 
-    override fun resolve(column: TableColumn<*>): Int = columns[column] ?: -1
+    private val COL_TIMESTAMP = 0
+    private val COL_CPU_COUNT = 1
+    private val COL_CPU_CAPACITY = 2
+    private val COL_CPU_USAGE = 3
+    private val COL_CPU_USAGE_PCT = 4
+    private val COL_MEM_CAPACITY = 5
+    private val COL_MEM_USAGE = 6
+    private val COL_DISK_READ = 7
+    private val COL_DISK_WRITE = 8
+    private val COL_NET_RX = 9
+    private val COL_NET_TX = 10
+    private val COL_ID = 11
 
-    override fun isNull(index: Int): Boolean {
-        check(index in 0..columns.size) { "Invalid column index" }
-        return false
+    override fun resolve(name: String): Int {
+        return when (name) {
+            RESOURCE_ID -> COL_ID
+            RESOURCE_STATE_TIMESTAMP -> COL_TIMESTAMP
+            RESOURCE_CPU_COUNT -> COL_CPU_COUNT
+            RESOURCE_CPU_CAPACITY -> COL_CPU_CAPACITY
+            RESOURCE_STATE_CPU_USAGE -> COL_CPU_USAGE
+            RESOURCE_STATE_CPU_USAGE_PCT -> COL_CPU_USAGE_PCT
+            RESOURCE_MEM_CAPACITY -> COL_MEM_CAPACITY
+            RESOURCE_STATE_MEM_USAGE -> COL_MEM_USAGE
+            RESOURCE_STATE_DISK_READ -> COL_DISK_READ
+            RESOURCE_STATE_DISK_WRITE -> COL_DISK_WRITE
+            RESOURCE_STATE_NET_RX -> COL_NET_RX
+            RESOURCE_STATE_NET_TX -> COL_NET_TX
+            else -> -1
+        }
     }
 
-    override fun get(index: Int): Any? {
-        return when (index) {
-            COL_ID -> partition
-            COL_TIMESTAMP -> timestamp
-            COL_CPU_COUNT -> getInt(index)
-            COL_CPU_CAPACITY, COL_CPU_USAGE, COL_CPU_USAGE_PCT, COL_MEM_CAPACITY, COL_MEM_USAGE, COL_DISK_READ, COL_DISK_WRITE, COL_NET_RX, COL_NET_TX -> getDouble(index)
-            else -> throw IllegalArgumentException("Invalid column")
-        }
+    override fun isNull(index: Int): Boolean {
+        require(index in 0..COL_ID) { "Invalid column index" }
+        return false
     }
 
     override fun getBoolean(index: Int): Boolean {
@@ -134,6 +163,7 @@ internal class BitbrainsResourceStateTableReader(private val partition: String,
     }
 
     override fun getInt(index: Int): Int {
+        checkActive()
         return when (index) {
             COL_CPU_COUNT -> cpuCores
             else -> throw IllegalArgumentException("Invalid column")
@@ -144,7 +174,12 @@ internal class BitbrainsResourceStateTableReader(private val partition: String,
         throw IllegalArgumentException("Invalid column")
     }
 
+    override fun getFloat(index: Int): Float {
+        throw IllegalArgumentException("Invalid column")
+    }
+
     override fun getDouble(index: Int): Double {
+        checkActive()
         return when (index) {
             COL_CPU_CAPACITY -> cpuCapacity
             COL_CPU_USAGE -> cpuUsage
@@ -159,11 +194,54 @@ internal class BitbrainsResourceStateTableReader(private val partition: String,
         }
     }
 
+    override fun getString(index: Int): String {
+        checkActive()
+        return when (index) {
+            COL_ID -> partition
+            else -> throw IllegalArgumentException("Invalid column")
+        }
+    }
+
+    override fun getUUID(index: Int): UUID? {
+        throw IllegalArgumentException("Invalid column")
+    }
+
+    override fun getInstant(index: Int): Instant? {
+        checkActive()
+        return when (index) {
+            COL_TIMESTAMP -> timestamp
+            else -> throw IllegalArgumentException("Invalid column")
+        }
+    }
+
+    override fun getDuration(index: Int): Duration? {
+        throw IllegalArgumentException("Invalid column")
+    }
+
+    override fun <T> getList(index: Int, elementType: Class<T>): List<T>? {
+        throw IllegalArgumentException("Invalid column")
+    }
+
+    override fun <T> getSet(index: Int, elementType: Class<T>): Set<T>? {
+        throw IllegalArgumentException("Invalid column")
+    }
+
+    override fun <K, V> getMap(index: Int, keyType: Class<K>, valueType: Class<V>): Map<K, V>? {
+        throw IllegalArgumentException("Invalid column")
+    }
+
     override fun close() {
         parser.close()
     }
 
     /**
+     * Helper method to check if the reader is active.
+     */
+    private fun checkActive() {
+        check(isStarted && !parser.isClosed) { "No active row. Did you call nextRow()?" }
+    }
+
+    /**
      * Advance the parser until the next object start.
      */
     private fun nextStart(): Boolean {
@@ -228,34 +306,6 @@ internal class BitbrainsResourceStateTableReader(private val partition: String,
         netTransmitted = Double.NaN
     }
 
-    private val COL_TIMESTAMP = 0
-    private val COL_CPU_COUNT = 1
-    private val COL_CPU_CAPACITY = 2
-    private val COL_CPU_USAGE = 3
-    private val COL_CPU_USAGE_PCT = 4
-    private val COL_MEM_CAPACITY = 5
-    private val COL_MEM_USAGE = 6
-    private val COL_DISK_READ = 7
-    private val COL_DISK_WRITE = 8
-    private val COL_NET_RX = 9
-    private val COL_NET_TX = 10
-    private val COL_ID = 11
-
-    private val columns = mapOf(
-        RESOURCE_ID to COL_ID,
-        RESOURCE_STATE_TIMESTAMP to COL_TIMESTAMP,
-        RESOURCE_CPU_COUNT to COL_CPU_COUNT,
-        RESOURCE_CPU_CAPACITY to COL_CPU_CAPACITY,
-        RESOURCE_STATE_CPU_USAGE to COL_CPU_USAGE,
-        RESOURCE_STATE_CPU_USAGE_PCT to COL_CPU_USAGE_PCT,
-        RESOURCE_MEM_CAPACITY to COL_MEM_CAPACITY,
-        RESOURCE_STATE_MEM_USAGE to COL_MEM_USAGE,
-        RESOURCE_STATE_DISK_READ to COL_DISK_READ,
-        RESOURCE_STATE_DISK_WRITE to COL_DISK_WRITE,
-        RESOURCE_STATE_NET_RX to COL_NET_RX,
-        RESOURCE_STATE_NET_TX to COL_NET_TX
-    )
-
     /**
      * The type of the timestamp in the trace.
      */
diff --git a/opendc-trace/opendc-trace-bitbrains/src/main/kotlin/org/opendc/trace/bitbrains/BitbrainsResourceTableReader.kt b/opendc-trace/opendc-trace-bitbrains/src/main/kotlin/org/opendc/trace/bitbrains/BitbrainsResourceTableReader.kt
index 55f09f43..7454c40f 100644
--- a/opendc-trace/opendc-trace-bitbrains/src/main/kotlin/org/opendc/trace/bitbrains/BitbrainsResourceTableReader.kt
+++ b/opendc-trace/opendc-trace-bitbrains/src/main/kotlin/org/opendc/trace/bitbrains/BitbrainsResourceTableReader.kt
@@ -26,6 +26,9 @@ import com.fasterxml.jackson.dataformat.csv.CsvFactory
 import org.opendc.trace.*
 import org.opendc.trace.conv.RESOURCE_ID
 import java.nio.file.Path
+import java.time.Duration
+import java.time.Instant
+import java.util.*
 
 /**
  * A [TableReader] for the Bitbrains resource table.
@@ -36,7 +39,16 @@ internal class BitbrainsResourceTableReader(private val factory: CsvFactory, vms
      */
     private val it = vms.iterator()
 
+    /**
+     * The state of the reader.
+     */
+    private var state = State.Pending
+
     override fun nextRow(): Boolean {
+        if (state == State.Pending) {
+            state = State.Active
+        }
+
         reset()
 
         while (it.hasNext()) {
@@ -51,47 +63,87 @@ internal class BitbrainsResourceTableReader(private val factory: CsvFactory, vms
                     continue
                 }
 
-                id = reader.get(idCol) as String
+                id = reader.getString(idCol)
                 return true
             } finally {
                 reader.close()
             }
         }
 
+        state = State.Closed
         return false
     }
 
-    override fun resolve(column: TableColumn<*>): Int = columns[column] ?: -1
+    private val COL_ID = 0
+
+    override fun resolve(name: String): Int {
+        return when (name) {
+            RESOURCE_ID -> COL_ID
+            else -> -1
+        }
+    }
 
     override fun isNull(index: Int): Boolean {
-        check(index in 0..columns.size) { "Invalid column index" }
+        require(index in 0..COL_ID) { "Invalid column index" }
         return false
     }
 
-    override fun get(index: Int): Any? {
+    override fun getBoolean(index: Int): Boolean {
+        throw IllegalArgumentException("Invalid column")
+    }
+
+    override fun getInt(index: Int): Int {
+        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? {
+        check(state == State.Active) { "No active row" }
         return when (index) {
             COL_ID -> id
             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 getLong(index: Int): Long {
+    override fun getDuration(index: Int): Duration? {
         throw IllegalArgumentException("Invalid column")
     }
 
-    override fun getDouble(index: Int): Double {
+    override fun <T> getList(index: Int, elementType: Class<T>): List<T>? {
         throw IllegalArgumentException("Invalid column")
     }
 
-    override fun close() {}
+    override fun <T> getSet(index: Int, elementType: Class<T>): Set<T>? {
+        throw IllegalArgumentException("Invalid column")
+    }
+
+    override fun <K, V> getMap(index: Int, keyType: Class<K>, valueType: Class<V>): Map<K, V>? {
+        throw IllegalArgumentException("Invalid column")
+    }
+
+    override fun close() {
+        reset()
+        state = State.Closed
+    }
 
     /**
      * State fields of the reader.
@@ -105,6 +157,7 @@ internal class BitbrainsResourceTableReader(private val factory: CsvFactory, vms
         id = null
     }
 
-    private val COL_ID = 0
-    private val columns = mapOf(RESOURCE_ID to COL_ID)
+    private enum class State {
+        Pending, Active, Closed
+    }
 }
diff --git a/opendc-trace/opendc-trace-bitbrains/src/main/kotlin/org/opendc/trace/bitbrains/BitbrainsTraceFormat.kt b/opendc-trace/opendc-trace-bitbrains/src/main/kotlin/org/opendc/trace/bitbrains/BitbrainsTraceFormat.kt
index a374e951..f3030893 100644
--- a/opendc-trace/opendc-trace-bitbrains/src/main/kotlin/org/opendc/trace/bitbrains/BitbrainsTraceFormat.kt
+++ b/opendc-trace/opendc-trace-bitbrains/src/main/kotlin/org/opendc/trace/bitbrains/BitbrainsTraceFormat.kt
@@ -59,29 +59,32 @@ public class BitbrainsTraceFormat : TraceFormat {
 
     override fun getDetails(path: Path, table: String): TableDetails {
         return when (table) {
-            TABLE_RESOURCES -> TableDetails(listOf(RESOURCE_ID))
+            TABLE_RESOURCES -> TableDetails(
+                listOf(
+                    TableColumn(RESOURCE_ID, TableColumnType.String)
+                )
+            )
             TABLE_RESOURCE_STATES -> TableDetails(
                 listOf(
-                    RESOURCE_ID,
-                    RESOURCE_STATE_TIMESTAMP,
-                    RESOURCE_CPU_COUNT,
-                    RESOURCE_CPU_CAPACITY,
-                    RESOURCE_STATE_CPU_USAGE,
-                    RESOURCE_STATE_CPU_USAGE_PCT,
-                    RESOURCE_MEM_CAPACITY,
-                    RESOURCE_STATE_MEM_USAGE,
-                    RESOURCE_STATE_DISK_READ,
-                    RESOURCE_STATE_DISK_WRITE,
-                    RESOURCE_STATE_NET_RX,
-                    RESOURCE_STATE_NET_TX,
+                    TableColumn(RESOURCE_ID, TableColumnType.String),
+                    TableColumn(RESOURCE_STATE_TIMESTAMP, TableColumnType.Instant),
+                    TableColumn(RESOURCE_CPU_COUNT, TableColumnType.Int),
+                    TableColumn(RESOURCE_CPU_CAPACITY, TableColumnType.Double),
+                    TableColumn(RESOURCE_STATE_CPU_USAGE, TableColumnType.Double),
+                    TableColumn(RESOURCE_STATE_CPU_USAGE_PCT, TableColumnType.Double),
+                    TableColumn(RESOURCE_MEM_CAPACITY, TableColumnType.Double),
+                    TableColumn(RESOURCE_STATE_MEM_USAGE, TableColumnType.Double),
+                    TableColumn(RESOURCE_STATE_DISK_READ, TableColumnType.Double),
+                    TableColumn(RESOURCE_STATE_DISK_WRITE, TableColumnType.Double),
+                    TableColumn(RESOURCE_STATE_NET_RX, TableColumnType.Double),
+                    TableColumn(RESOURCE_STATE_NET_TX, TableColumnType.Double),
                 ),
-                listOf(RESOURCE_ID, RESOURCE_STATE_TIMESTAMP)
             )
             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_RESOURCES -> {
                 val vms = Files.walk(path, 1)
diff --git a/opendc-trace/opendc-trace-bitbrains/src/test/kotlin/org/opendc/trace/bitbrains/BitbrainsExTraceFormatTest.kt b/opendc-trace/opendc-trace-bitbrains/src/test/kotlin/org/opendc/trace/bitbrains/BitbrainsExTraceFormatTest.kt
index c944cb98..dbb75c50 100644
--- a/opendc-trace/opendc-trace-bitbrains/src/test/kotlin/org/opendc/trace/bitbrains/BitbrainsExTraceFormatTest.kt
+++ b/opendc-trace/opendc-trace-bitbrains/src/test/kotlin/org/opendc/trace/bitbrains/BitbrainsExTraceFormatTest.kt
@@ -22,12 +22,15 @@
 
 package org.opendc.trace.bitbrains
 
+import org.junit.jupiter.api.*
 import org.junit.jupiter.api.Assertions.*
-import org.junit.jupiter.api.Test
-import org.junit.jupiter.api.assertThrows
+import org.junit.jupiter.api.Assertions.assertAll
+import org.opendc.trace.TableColumn
+import org.opendc.trace.TableReader
 import org.opendc.trace.conv.RESOURCE_STATE_CPU_USAGE
 import org.opendc.trace.conv.RESOURCE_STATE_TIMESTAMP
 import org.opendc.trace.conv.TABLE_RESOURCE_STATES
+import org.opendc.trace.testkit.TableReaderTestKit
 import java.nio.file.Paths
 
 /**
@@ -63,10 +66,25 @@ internal class BitbrainsExTraceFormatTest {
 
         assertAll(
             { assertTrue(reader.nextRow()) },
-            { assertEquals(1631911500, reader.get(RESOURCE_STATE_TIMESTAMP).epochSecond) },
+            { assertEquals(1631911500, reader.getInstant(RESOURCE_STATE_TIMESTAMP)?.epochSecond) },
             { assertEquals(21.2, reader.getDouble(RESOURCE_STATE_CPU_USAGE), 0.01) }
         )
 
         reader.close()
     }
+
+    @DisplayName("TableReader for Resource States")
+    @Nested
+    inner class ResourceStatesTableReaderTest : TableReaderTestKit() {
+        override lateinit var reader: TableReader
+        override lateinit var columns: List<TableColumn>
+
+        @BeforeEach
+        fun setUp() {
+            val path = Paths.get("src/test/resources/vm.txt")
+
+            columns = format.getDetails(path, TABLE_RESOURCE_STATES).columns
+            reader = format.newReader(path, TABLE_RESOURCE_STATES, null)
+        }
+    }
 }
diff --git a/opendc-trace/opendc-trace-bitbrains/src/test/kotlin/org/opendc/trace/bitbrains/BitbrainsTraceFormatTest.kt b/opendc-trace/opendc-trace-bitbrains/src/test/kotlin/org/opendc/trace/bitbrains/BitbrainsTraceFormatTest.kt
index 841801e6..712e1fcb 100644
--- a/opendc-trace/opendc-trace-bitbrains/src/test/kotlin/org/opendc/trace/bitbrains/BitbrainsTraceFormatTest.kt
+++ b/opendc-trace/opendc-trace-bitbrains/src/test/kotlin/org/opendc/trace/bitbrains/BitbrainsTraceFormatTest.kt
@@ -22,10 +22,13 @@
 
 package org.opendc.trace.bitbrains
 
+import org.junit.jupiter.api.*
 import org.junit.jupiter.api.Assertions.*
-import org.junit.jupiter.api.Test
-import org.junit.jupiter.api.assertThrows
+import org.junit.jupiter.api.Assertions.assertAll
+import org.opendc.trace.TableColumn
+import org.opendc.trace.TableReader
 import org.opendc.trace.conv.*
+import org.opendc.trace.testkit.TableReaderTestKit
 import java.nio.file.Paths
 
 /**
@@ -61,7 +64,7 @@ class BitbrainsTraceFormatTest {
 
         assertAll(
             { assertTrue(reader.nextRow()) },
-            { assertEquals("bitbrains", reader.get(RESOURCE_ID)) },
+            { assertEquals("bitbrains", reader.getString(RESOURCE_ID)) },
             { assertFalse(reader.nextRow()) }
         )
 
@@ -75,10 +78,40 @@ class BitbrainsTraceFormatTest {
 
         assertAll(
             { assertTrue(reader.nextRow()) },
-            { assertEquals(1376314846, reader.get(RESOURCE_STATE_TIMESTAMP).epochSecond) },
+            { assertEquals(1376314846, reader.getInstant(RESOURCE_STATE_TIMESTAMP)?.epochSecond) },
             { assertEquals(19.066, reader.getDouble(RESOURCE_STATE_CPU_USAGE), 0.01) }
         )
 
         reader.close()
     }
+
+    @DisplayName("TableReader for Resources")
+    @Nested
+    inner class ResourcesTableReaderTest : TableReaderTestKit() {
+        override lateinit var reader: TableReader
+        override lateinit var columns: List<TableColumn>
+
+        @BeforeEach
+        fun setUp() {
+            val path = Paths.get("src/test/resources/bitbrains.csv")
+
+            columns = format.getDetails(path, TABLE_RESOURCES).columns
+            reader = format.newReader(path, TABLE_RESOURCES, null)
+        }
+    }
+
+    @DisplayName("TableReader for Resource States")
+    @Nested
+    inner class ResourceStatesTableReaderTest : TableReaderTestKit() {
+        override lateinit var reader: TableReader
+        override lateinit var columns: List<TableColumn>
+
+        @BeforeEach
+        fun setUp() {
+            val path = Paths.get("src/test/resources/bitbrains.csv")
+
+            columns = format.getDetails(path, TABLE_RESOURCE_STATES).columns
+            reader = format.newReader(path, TABLE_RESOURCE_STATES, null)
+        }
+    }
 }