merge: Integrate v2.1 progress into public repository

This pull request integrates the changes planned for the v2.1 release of
OpenDC into the public Github repository in order to sync the progress
of both repositories.

12 files changed, 811 insertions, 0 deletions

diff --git a/opendc-trace/opendc-trace-opendc/build.gradle.kts b/opendc-trace/opendc-trace-opendc/build.gradle.kts
new file mode 100644
index 00000000..b9c242a1
--- /dev/null
+++ b/opendc-trace/opendc-trace-opendc/build.gradle.kts
@@ -0,0 +1,39 @@
+/*
+ * Copyright (c) 2021 AtLarge Research
+ *
+ * Permission is hereby granted, free of charge, to any person obtaining a copy
+ * of this software and associated documentation files (the "Software"), to deal
+ * in the Software without restriction, including without limitation the rights
+ * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+ * copies of the Software, and to permit persons to whom the Software is
+ * furnished to do so, subject to the following conditions:
+ *
+ * The above copyright notice and this permission notice shall be included in all
+ * copies or substantial portions of the Software.
+ *
+ * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+ * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+ * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+ * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+ * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+ * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
+ * SOFTWARE.
+ */
+
+description = "Support for OpenDC-specific trace formats"
+
+/* Build configuration */
+plugins {
+    `kotlin-library-conventions`
+    `testing-conventions`
+    `jacoco-conventions`
+}
+
+dependencies {
+    api(platform(projects.opendcPlatform))
+    api(projects.opendcTrace.opendcTraceApi)
+
+    implementation(projects.opendcTrace.opendcTraceParquet)
+
+    testRuntimeOnly(libs.slf4j.simple)
+}
diff --git a/opendc-trace/opendc-trace-opendc/src/main/kotlin/org/opendc/trace/opendc/OdcVmResourceStateTableReader.kt b/opendc-trace/opendc-trace-opendc/src/main/kotlin/org/opendc/trace/opendc/OdcVmResourceStateTableReader.kt
new file mode 100644
index 00000000..b5043f82
--- /dev/null
+++ b/opendc-trace/opendc-trace-opendc/src/main/kotlin/org/opendc/trace/opendc/OdcVmResourceStateTableReader.kt
@@ -0,0 +1,143 @@
+/*
+ * Copyright (c) 2021 AtLarge Research
+ *
+ * Permission is hereby granted, free of charge, to any person obtaining a copy
+ * of this software and associated documentation files (the "Software"), to deal
+ * in the Software without restriction, including without limitation the rights
+ * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+ * copies of the Software, and to permit persons to whom the Software is
+ * furnished to do so, subject to the following conditions:
+ *
+ * The above copyright notice and this permission notice shall be included in all
+ * copies or substantial portions of the Software.
+ *
+ * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+ * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+ * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+ * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+ * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+ * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
+ * SOFTWARE.
+ */
+
+package org.opendc.trace.opendc
+
+import org.apache.avro.Schema
+import org.apache.avro.generic.GenericRecord
+import org.opendc.trace.*
+import org.opendc.trace.util.parquet.LocalParquetReader
+import java.time.Duration
+import java.time.Instant
+
+/**
+ * A [TableReader] implementation for the OpenDC virtual machine trace format.
+ */
+internal class OdcVmResourceStateTableReader(private val reader: LocalParquetReader<GenericRecord>) : TableReader {
+    /**
+     * The current record.
+     */
+    private var record: GenericRecord? = null
+
+    /**
+     * A flag to indicate that the columns have been initialized.
+     */
+    private var hasInitializedColumns = false
+
+    override fun nextRow(): Boolean {
+        val record = reader.read()
+        this.record = record
+
+        if (!hasInitializedColumns && record != null) {
+            initColumns(record.schema)
+            hasInitializedColumns = true
+        }
+
+        return record != null
+    }
+
+    override fun resolve(column: TableColumn<*>): Int = columns[column] ?: -1
+
+    override fun isNull(index: Int): Boolean {
+        check(index in 0..columns.size) { "Invalid column index" }
+        return get(index) == null
+    }
+
+    override fun get(index: Int): Any? {
+        val record = checkNotNull(record) { "Reader in invalid state" }
+
+        return when (index) {
+            COL_ID -> record[AVRO_COL_ID].toString()
+            COL_TIMESTAMP -> Instant.ofEpochMilli(record[AVRO_COL_TIMESTAMP] as Long)
+            COL_DURATION -> Duration.ofMillis(record[AVRO_COL_DURATION] as Long)
+            COL_CPU_COUNT -> getInt(index)
+            COL_CPU_USAGE -> getDouble(index)
+            else -> throw IllegalArgumentException("Invalid column")
+        }
+    }
+
+    override fun getBoolean(index: Int): Boolean {
+        throw IllegalArgumentException("Invalid column")
+    }
+
+    override fun getInt(index: Int): Int {
+        val record = checkNotNull(record) { "Reader in invalid state" }
+        return when (index) {
+            COL_CPU_COUNT -> record[AVRO_COL_CPU_COUNT] as Int
+            else -> throw IllegalArgumentException("Invalid column")
+        }
+    }
+
+    override fun getLong(index: Int): Long {
+        throw IllegalArgumentException("Invalid column")
+    }
+
+    override fun getDouble(index: Int): Double {
+        val record = checkNotNull(record) { "Reader in invalid state" }
+        return when (index) {
+            COL_CPU_USAGE -> (record[AVRO_COL_CPU_USAGE] as Number).toDouble()
+            else -> throw IllegalArgumentException("Invalid column")
+        }
+    }
+
+    override fun close() {
+        reader.close()
+    }
+
+    override fun toString(): String = "OdcVmResourceStateTableReader"
+
+    /**
+     * Initialize the columns for the reader based on [schema].
+     */
+    private fun initColumns(schema: Schema) {
+        try {
+            AVRO_COL_ID = schema.getField("id").pos()
+            AVRO_COL_TIMESTAMP = (schema.getField("timestamp") ?: schema.getField("time")).pos()
+            AVRO_COL_DURATION = schema.getField("duration").pos()
+            AVRO_COL_CPU_COUNT = (schema.getField("cpu_count") ?: schema.getField("cores")).pos()
+            AVRO_COL_CPU_USAGE = (schema.getField("cpu_usage") ?: schema.getField("cpuUsage")).pos()
+        } catch (e: NullPointerException) {
+            // This happens when the field we are trying to access does not exist
+            throw IllegalArgumentException("Invalid schema", e)
+        }
+    }
+
+    private var AVRO_COL_ID = -1
+    private var AVRO_COL_TIMESTAMP = -1
+    private var AVRO_COL_DURATION = -1
+    private var AVRO_COL_CPU_COUNT = -1
+    private var AVRO_COL_CPU_USAGE = -1
+
+    private val COL_ID = 0
+    private val COL_TIMESTAMP = 1
+    private val COL_DURATION = 2
+    private val COL_CPU_COUNT = 3
+    private val COL_CPU_USAGE = 4
+
+    private val columns = mapOf(
+        RESOURCE_ID to COL_ID,
+        RESOURCE_STATE_TIMESTAMP to COL_TIMESTAMP,
+        RESOURCE_STATE_DURATION to COL_DURATION,
+        RESOURCE_CPU_COUNT to COL_CPU_COUNT,
+        RESOURCE_STATE_CPU_USAGE to COL_CPU_USAGE,
+    )
+}
diff --git a/opendc-trace/opendc-trace-opendc/src/main/kotlin/org/opendc/trace/opendc/OdcVmResourceStateTableWriter.kt b/opendc-trace/opendc-trace-opendc/src/main/kotlin/org/opendc/trace/opendc/OdcVmResourceStateTableWriter.kt
new file mode 100644
index 00000000..15a8cb85
--- /dev/null
+++ b/opendc-trace/opendc-trace-opendc/src/main/kotlin/org/opendc/trace/opendc/OdcVmResourceStateTableWriter.kt
@@ -0,0 +1,123 @@
+/*
+ * Copyright (c) 2021 AtLarge Research
+ *
+ * Permission is hereby granted, free of charge, to any person obtaining a copy
+ * of this software and associated documentation files (the "Software"), to deal
+ * in the Software without restriction, including without limitation the rights
+ * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+ * copies of the Software, and to permit persons to whom the Software is
+ * furnished to do so, subject to the following conditions:
+ *
+ * The above copyright notice and this permission notice shall be included in all
+ * copies or substantial portions of the Software.
+ *
+ * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+ * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+ * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+ * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+ * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+ * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
+ * SOFTWARE.
+ */
+
+package org.opendc.trace.opendc
+
+import org.apache.avro.Schema
+import org.apache.avro.generic.GenericRecord
+import org.apache.avro.generic.GenericRecordBuilder
+import org.apache.parquet.hadoop.ParquetWriter
+import org.opendc.trace.*
+import java.time.Duration
+import java.time.Instant
+
+/**
+ * A [TableWriter] implementation for the OpenDC virtual machine trace format.
+ */
+internal class OdcVmResourceStateTableWriter(
+    private val writer: ParquetWriter<GenericRecord>,
+    private val schema: Schema
+) : TableWriter {
+    /**
+     * The current builder for the record that is being written.
+     */
+    private var builder: GenericRecordBuilder? = null
+
+    /**
+     * The fields belonging to the resource state schema.
+     */
+    private val fields = schema.fields
+
+    override fun startRow() {
+        builder = GenericRecordBuilder(schema)
+    }
+
+    override fun endRow() {
+        val builder = checkNotNull(builder) { "No active row" }
+        this.builder = null
+
+        val record = builder.build()
+        val id = record[COL_ID] as String
+        val timestamp = record[COL_TIMESTAMP] as Long
+
+        check(lastId != id || timestamp >= lastTimestamp) { "Records need to be ordered by (id, timestamp)" }
+
+        writer.write(builder.build())
+
+        lastId = id
+        lastTimestamp = timestamp
+    }
+
+    override fun resolve(column: TableColumn<*>): Int {
+        val schema = schema
+        return when (column) {
+            RESOURCE_ID -> schema.getField("id").pos()
+            RESOURCE_STATE_TIMESTAMP -> (schema.getField("timestamp") ?: schema.getField("time")).pos()
+            RESOURCE_STATE_DURATION -> schema.getField("duration").pos()
+            RESOURCE_CPU_COUNT -> (schema.getField("cpu_count") ?: schema.getField("cores")).pos()
+            RESOURCE_STATE_CPU_USAGE -> (schema.getField("cpu_usage") ?: schema.getField("cpuUsage")).pos()
+            else -> -1
+        }
+    }
+
+    override fun set(index: Int, value: Any) {
+        val builder = checkNotNull(builder) { "No active row" }
+
+        builder.set(
+            fields[index],
+            when (index) {
+                COL_TIMESTAMP -> (value as Instant).toEpochMilli()
+                COL_DURATION -> (value as Duration).toMillis()
+                else -> value
+            }
+        )
+    }
+
+    override fun setBoolean(index: Int, value: Boolean) = set(index, value)
+
+    override fun setInt(index: Int, value: Int) = set(index, value)
+
+    override fun setLong(index: Int, value: Long) = set(index, value)
+
+    override fun setDouble(index: Int, value: Double) = set(index, value)
+
+    override fun flush() {
+        // Not available
+    }
+
+    override fun close() {
+        writer.close()
+    }
+
+    /**
+     * Last column values that are used to check for correct partitioning.
+     */
+    private var lastId: String? = null
+    private var lastTimestamp: Long = Long.MIN_VALUE
+
+    /**
+     * Columns with special behavior.
+     */
+    private val COL_ID = resolve(RESOURCE_ID)
+    private val COL_TIMESTAMP = resolve(RESOURCE_STATE_TIMESTAMP)
+    private val COL_DURATION = resolve(RESOURCE_STATE_DURATION)
+}
diff --git a/opendc-trace/opendc-trace-opendc/src/main/kotlin/org/opendc/trace/opendc/OdcVmResourceTableReader.kt b/opendc-trace/opendc-trace-opendc/src/main/kotlin/org/opendc/trace/opendc/OdcVmResourceTableReader.kt
new file mode 100644
index 00000000..d93929aa
--- /dev/null
+++ b/opendc-trace/opendc-trace-opendc/src/main/kotlin/org/opendc/trace/opendc/OdcVmResourceTableReader.kt
@@ -0,0 +1,144 @@
+/*
+ * Copyright (c) 2021 AtLarge Research
+ *
+ * Permission is hereby granted, free of charge, to any person obtaining a copy
+ * of this software and associated documentation files (the "Software"), to deal
+ * in the Software without restriction, including without limitation the rights
+ * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+ * copies of the Software, and to permit persons to whom the Software is
+ * furnished to do so, subject to the following conditions:
+ *
+ * The above copyright notice and this permission notice shall be included in all
+ * copies or substantial portions of the Software.
+ *
+ * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+ * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+ * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+ * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+ * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+ * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
+ * SOFTWARE.
+ */
+
+package org.opendc.trace.opendc
+
+import org.apache.avro.Schema
+import org.apache.avro.generic.GenericRecord
+import org.opendc.trace.*
+import org.opendc.trace.util.parquet.LocalParquetReader
+import java.time.Instant
+
+/**
+ * A [TableReader] implementation for the resources table in the OpenDC virtual machine trace format.
+ */
+internal class OdcVmResourceTableReader(private val reader: LocalParquetReader<GenericRecord>) : TableReader {
+    /**
+     * The current record.
+     */
+    private var record: GenericRecord? = null
+
+    /**
+     * A flag to indicate that the columns have been initialized.
+     */
+    private var hasInitializedColumns = false
+
+    override fun nextRow(): Boolean {
+        val record = reader.read()
+        this.record = record
+
+        if (!hasInitializedColumns && record != null) {
+            initColumns(record.schema)
+            hasInitializedColumns = true
+        }
+
+        return record != null
+    }
+
+    override fun resolve(column: TableColumn<*>): Int = columns[column] ?: -1
+
+    override fun isNull(index: Int): Boolean {
+        check(index in 0..columns.size) { "Invalid column index" }
+        return get(index) == null
+    }
+
+    override fun get(index: Int): Any? {
+        val record = checkNotNull(record) { "Reader in invalid state" }
+
+        return when (index) {
+            COL_ID -> record[AVRO_COL_ID].toString()
+            COL_START_TIME -> Instant.ofEpochMilli(record[AVRO_COL_START_TIME] as Long)
+            COL_STOP_TIME -> Instant.ofEpochMilli(record[AVRO_COL_STOP_TIME] as Long)
+            COL_CPU_COUNT -> getInt(index)
+            COL_MEM_CAPACITY -> getDouble(index)
+            else -> throw IllegalArgumentException("Invalid column")
+        }
+    }
+
+    override fun getBoolean(index: Int): Boolean {
+        throw IllegalArgumentException("Invalid column")
+    }
+
+    override fun getInt(index: Int): Int {
+        val record = checkNotNull(record) { "Reader in invalid state" }
+
+        return when (index) {
+            COL_CPU_COUNT -> record[AVRO_COL_CPU_COUNT] as Int
+            else -> throw IllegalArgumentException("Invalid column")
+        }
+    }
+
+    override fun getLong(index: Int): Long {
+        throw IllegalArgumentException("Invalid column")
+    }
+
+    override fun getDouble(index: Int): Double {
+        val record = checkNotNull(record) { "Reader in invalid state" }
+
+        return when (index) {
+            COL_MEM_CAPACITY -> (record[AVRO_COL_MEM_CAPACITY] as Number).toDouble()
+            else -> throw IllegalArgumentException("Invalid column")
+        }
+    }
+
+    override fun close() {
+        reader.close()
+    }
+
+    override fun toString(): String = "OdcVmResourceTableReader"
+
+    /**
+     * Initialize the columns for the reader based on [schema].
+     */
+    private fun initColumns(schema: Schema) {
+        try {
+            AVRO_COL_ID = schema.getField("id").pos()
+            AVRO_COL_START_TIME = (schema.getField("start_time") ?: schema.getField("submissionTime")).pos()
+            AVRO_COL_STOP_TIME = (schema.getField("stop_time") ?: schema.getField("endTime")).pos()
+            AVRO_COL_CPU_COUNT = (schema.getField("cpu_count") ?: schema.getField("maxCores")).pos()
+            AVRO_COL_MEM_CAPACITY = (schema.getField("mem_capacity") ?: schema.getField("requiredMemory")).pos()
+        } catch (e: NullPointerException) {
+            // This happens when the field we are trying to access does not exist
+            throw IllegalArgumentException("Invalid schema")
+        }
+    }
+
+    private var AVRO_COL_ID = -1
+    private var AVRO_COL_START_TIME = -1
+    private var AVRO_COL_STOP_TIME = -1
+    private var AVRO_COL_CPU_COUNT = -1
+    private var AVRO_COL_MEM_CAPACITY = -1
+
+    private val COL_ID = 0
+    private val COL_START_TIME = 1
+    private val COL_STOP_TIME = 2
+    private val COL_CPU_COUNT = 3
+    private val COL_MEM_CAPACITY = 4
+
+    private val columns = mapOf(
+        RESOURCE_ID to COL_ID,
+        RESOURCE_START_TIME to COL_START_TIME,
+        RESOURCE_STOP_TIME to COL_STOP_TIME,
+        RESOURCE_CPU_COUNT to COL_CPU_COUNT,
+        RESOURCE_MEM_CAPACITY to COL_MEM_CAPACITY,
+    )
+}
diff --git a/opendc-trace/opendc-trace-opendc/src/main/kotlin/org/opendc/trace/opendc/OdcVmResourceTableWriter.kt b/opendc-trace/opendc-trace-opendc/src/main/kotlin/org/opendc/trace/opendc/OdcVmResourceTableWriter.kt
new file mode 100644
index 00000000..9cc6ca7d
--- /dev/null
+++ b/opendc-trace/opendc-trace-opendc/src/main/kotlin/org/opendc/trace/opendc/OdcVmResourceTableWriter.kt
@@ -0,0 +1,106 @@
+/*
+ * Copyright (c) 2021 AtLarge Research
+ *
+ * Permission is hereby granted, free of charge, to any person obtaining a copy
+ * of this software and associated documentation files (the "Software"), to deal
+ * in the Software without restriction, including without limitation the rights
+ * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+ * copies of the Software, and to permit persons to whom the Software is
+ * furnished to do so, subject to the following conditions:
+ *
+ * The above copyright notice and this permission notice shall be included in all
+ * copies or substantial portions of the Software.
+ *
+ * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+ * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+ * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+ * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+ * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+ * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
+ * SOFTWARE.
+ */
+
+package org.opendc.trace.opendc
+
+import org.apache.avro.Schema
+import org.apache.avro.generic.GenericRecord
+import org.apache.avro.generic.GenericRecordBuilder
+import org.apache.parquet.hadoop.ParquetWriter
+import org.opendc.trace.*
+import java.time.Instant
+import kotlin.math.roundToLong
+
+/**
+ * A [TableWriter] implementation for the OpenDC virtual machine trace format.
+ */
+internal class OdcVmResourceTableWriter(
+    private val writer: ParquetWriter<GenericRecord>,
+    private val schema: Schema
+) : TableWriter {
+    /**
+     * The current builder for the record that is being written.
+     */
+    private var builder: GenericRecordBuilder? = null
+
+    /**
+     * The fields belonging to the resource schema.
+     */
+    private val fields = schema.fields
+
+    override fun startRow() {
+        builder = GenericRecordBuilder(schema)
+    }
+
+    override fun endRow() {
+        val builder = checkNotNull(builder) { "No active row" }
+        this.builder = null
+        writer.write(builder.build())
+    }
+
+    override fun resolve(column: TableColumn<*>): Int {
+        val schema = schema
+        return when (column) {
+            RESOURCE_ID -> schema.getField("id").pos()
+            RESOURCE_START_TIME -> (schema.getField("start_time") ?: schema.getField("submissionTime")).pos()
+            RESOURCE_STOP_TIME -> (schema.getField("stop_time") ?: schema.getField("endTime")).pos()
+            RESOURCE_CPU_COUNT -> (schema.getField("cpu_count") ?: schema.getField("maxCores")).pos()
+            RESOURCE_MEM_CAPACITY -> (schema.getField("mem_capacity") ?: schema.getField("requiredMemory")).pos()
+            else -> -1
+        }
+    }
+
+    override fun set(index: Int, value: Any) {
+        val builder = checkNotNull(builder) { "No active row" }
+        builder.set(
+            fields[index],
+            when (index) {
+                COL_START_TIME, COL_STOP_TIME -> (value as Instant).toEpochMilli()
+                COL_MEM_CAPACITY -> (value as Double).roundToLong()
+                else -> value
+            }
+        )
+    }
+
+    override fun setBoolean(index: Int, value: Boolean) = set(index, value)
+
+    override fun setInt(index: Int, value: Int) = set(index, value)
+
+    override fun setLong(index: Int, value: Long) = set(index, value)
+
+    override fun setDouble(index: Int, value: Double) = set(index, value)
+
+    override fun flush() {
+        // Not available
+    }
+
+    override fun close() {
+        writer.close()
+    }
+
+    /**
+     * Columns with special behavior.
+     */
+    private val COL_START_TIME = resolve(RESOURCE_START_TIME)
+    private val COL_STOP_TIME = resolve(RESOURCE_STOP_TIME)
+    private val COL_MEM_CAPACITY = resolve(RESOURCE_MEM_CAPACITY)
+}
diff --git a/opendc-trace/opendc-trace-opendc/src/main/kotlin/org/opendc/trace/opendc/OdcVmTraceFormat.kt b/opendc-trace/opendc-trace-opendc/src/main/kotlin/org/opendc/trace/opendc/OdcVmTraceFormat.kt
new file mode 100644
index 00000000..9b32f8fd
--- /dev/null
+++ b/opendc-trace/opendc-trace-opendc/src/main/kotlin/org/opendc/trace/opendc/OdcVmTraceFormat.kt
@@ -0,0 +1,159 @@
+/*
+ * Copyright (c) 2021 AtLarge Research
+ *
+ * Permission is hereby granted, free of charge, to any person obtaining a copy
+ * of this software and associated documentation files (the "Software"), to deal
+ * in the Software without restriction, including without limitation the rights
+ * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+ * copies of the Software, and to permit persons to whom the Software is
+ * furnished to do so, subject to the following conditions:
+ *
+ * The above copyright notice and this permission notice shall be included in all
+ * copies or substantial portions of the Software.
+ *
+ * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+ * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+ * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+ * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+ * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+ * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
+ * SOFTWARE.
+ */
+
+package org.opendc.trace.opendc
+
+import org.apache.avro.Schema
+import org.apache.avro.SchemaBuilder
+import org.apache.avro.generic.GenericRecord
+import org.apache.parquet.avro.AvroParquetWriter
+import org.apache.parquet.hadoop.ParquetFileWriter
+import org.apache.parquet.hadoop.metadata.CompressionCodecName
+import org.opendc.trace.*
+import org.opendc.trace.spi.TableDetails
+import org.opendc.trace.spi.TraceFormat
+import org.opendc.trace.util.parquet.LocalOutputFile
+import org.opendc.trace.util.parquet.LocalParquetReader
+import org.opendc.trace.util.parquet.TIMESTAMP_SCHEMA
+import java.nio.file.Files
+import java.nio.file.Path
+
+/**
+ * A [TraceFormat] implementation of the OpenDC virtual machine trace format.
+ */
+public class OdcVmTraceFormat : TraceFormat {
+    /**
+     * The name of this trace format.
+     */
+    override val name: String = "opendc-vm"
+
+    override fun create(path: Path) {
+        // Construct directory containing the trace files
+        Files.createDirectory(path)
+
+        val tables = getTables(path)
+
+        for (table in tables) {
+            val writer = newWriter(path, table)
+            writer.close()
+        }
+    }
+
+    override fun getTables(path: Path): List<String> = listOf(TABLE_RESOURCES, TABLE_RESOURCE_STATES)
+
+    override fun getDetails(path: Path, table: String): TableDetails {
+        return when (table) {
+            TABLE_RESOURCES -> TableDetails(
+                listOf(
+                    RESOURCE_ID,
+                    RESOURCE_START_TIME,
+                    RESOURCE_STOP_TIME,
+                    RESOURCE_CPU_COUNT,
+                    RESOURCE_MEM_CAPACITY,
+                )
+            )
+            TABLE_RESOURCE_STATES -> TableDetails(
+                listOf(
+                    RESOURCE_ID,
+                    RESOURCE_STATE_TIMESTAMP,
+                    RESOURCE_STATE_DURATION,
+                    RESOURCE_CPU_COUNT,
+                    RESOURCE_STATE_CPU_USAGE,
+                ),
+                listOf(RESOURCE_ID, RESOURCE_STATE_TIMESTAMP)
+            )
+            else -> throw IllegalArgumentException("Table $table not supported")
+        }
+    }
+
+    override fun newReader(path: Path, table: String): TableReader {
+        return when (table) {
+            TABLE_RESOURCES -> {
+                val reader = LocalParquetReader<GenericRecord>(path.resolve("meta.parquet"))
+                OdcVmResourceTableReader(reader)
+            }
+            TABLE_RESOURCE_STATES -> {
+                val reader = LocalParquetReader<GenericRecord>(path.resolve("trace.parquet"))
+                OdcVmResourceStateTableReader(reader)
+            }
+            else -> throw IllegalArgumentException("Table $table not supported")
+        }
+    }
+
+    override fun newWriter(path: Path, table: String): TableWriter {
+        return when (table) {
+            TABLE_RESOURCES -> {
+                val schema = RESOURCES_SCHEMA
+                val writer = AvroParquetWriter.builder<GenericRecord>(LocalOutputFile(path.resolve("meta.parquet")))
+                    .withSchema(schema)
+                    .withCompressionCodec(CompressionCodecName.ZSTD)
+                    .withWriteMode(ParquetFileWriter.Mode.OVERWRITE)
+                    .build()
+                OdcVmResourceTableWriter(writer, schema)
+            }
+            TABLE_RESOURCE_STATES -> {
+                val schema = RESOURCE_STATES_SCHEMA
+                val writer = AvroParquetWriter.builder<GenericRecord>(LocalOutputFile(path.resolve("trace.parquet")))
+                    .withSchema(schema)
+                    .withCompressionCodec(CompressionCodecName.ZSTD)
+                    .withDictionaryEncoding("id", true)
+                    .withBloomFilterEnabled("id", true)
+                    .withWriteMode(ParquetFileWriter.Mode.OVERWRITE)
+                    .build()
+                OdcVmResourceStateTableWriter(writer, schema)
+            }
+            else -> throw IllegalArgumentException("Table $table not supported")
+        }
+    }
+
+    public companion object {
+        /**
+         * Schema for the resources table in the trace.
+         */
+        @JvmStatic
+        public val RESOURCES_SCHEMA: Schema = SchemaBuilder
+            .record("resource")
+            .namespace("org.opendc.trace.opendc")
+            .fields()
+            .requiredString("id")
+            .name("start_time").type(TIMESTAMP_SCHEMA).noDefault()
+            .name("stop_time").type(TIMESTAMP_SCHEMA).noDefault()
+            .requiredInt("cpu_count")
+            .requiredLong("mem_capacity")
+            .endRecord()
+
+        /**
+         * Schema for the resource states table in the trace.
+         */
+        @JvmStatic
+        public val RESOURCE_STATES_SCHEMA: Schema = SchemaBuilder
+            .record("resource_state")
+            .namespace("org.opendc.trace.opendc")
+            .fields()
+            .requiredString("id")
+            .name("timestamp").type(TIMESTAMP_SCHEMA).noDefault()
+            .requiredLong("duration")
+            .requiredInt("cpu_count")
+            .requiredDouble("cpu_usage")
+            .endRecord()
+    }
+}
diff --git a/opendc-trace/opendc-trace-opendc/src/main/resources/META-INF/services/org.opendc.trace.spi.TraceFormat b/opendc-trace/opendc-trace-opendc/src/main/resources/META-INF/services/org.opendc.trace.spi.TraceFormat
new file mode 100644
index 00000000..94094af4
--- /dev/null
+++ b/opendc-trace/opendc-trace-opendc/src/main/resources/META-INF/services/org.opendc.trace.spi.TraceFormat
@@ -0,0 +1 @@
+org.opendc.trace.opendc.OdcVmTraceFormat
diff --git a/opendc-trace/opendc-trace-opendc/src/test/kotlin/org/opendc/trace/opendc/OdcVmTraceFormatTest.kt b/opendc-trace/opendc-trace-opendc/src/test/kotlin/org/opendc/trace/opendc/OdcVmTraceFormatTest.kt
new file mode 100644
index 00000000..bfe0f881
--- /dev/null
+++ b/opendc-trace/opendc-trace-opendc/src/test/kotlin/org/opendc/trace/opendc/OdcVmTraceFormatTest.kt
@@ -0,0 +1,96 @@
+/*
+ * Copyright (c) 2021 AtLarge Research
+ *
+ * Permission is hereby granted, free of charge, to any person obtaining a copy
+ * of this software and associated documentation files (the "Software"), to deal
+ * in the Software without restriction, including without limitation the rights
+ * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+ * copies of the Software, and to permit persons to whom the Software is
+ * furnished to do so, subject to the following conditions:
+ *
+ * The above copyright notice and this permission notice shall be included in all
+ * copies or substantial portions of the Software.
+ *
+ * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+ * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+ * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+ * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+ * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+ * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
+ * SOFTWARE.
+ */
+
+package org.opendc.trace.opendc
+
+import org.junit.jupiter.api.Assertions.*
+import org.junit.jupiter.api.Test
+import org.junit.jupiter.api.assertDoesNotThrow
+import org.junit.jupiter.api.assertThrows
+import org.junit.jupiter.params.ParameterizedTest
+import org.junit.jupiter.params.provider.ValueSource
+import org.opendc.trace.*
+import java.nio.file.Paths
+
+/**
+ * Test suite for the [OdcVmTraceFormat] implementation.
+ */
+internal class OdcVmTraceFormatTest {
+    private val format = OdcVmTraceFormat()
+
+    @Test
+    fun testTables() {
+        val path = Paths.get("src/test/resources/trace-v2.1")
+
+        assertEquals(listOf(TABLE_RESOURCES, TABLE_RESOURCE_STATES), format.getTables(path))
+    }
+
+    @Test
+    fun testTableExists() {
+        val path = Paths.get("src/test/resources/trace-v2.1")
+
+        assertDoesNotThrow { format.getDetails(path, TABLE_RESOURCE_STATES) }
+    }
+
+    @Test
+    fun testTableDoesNotExist() {
+        val path = Paths.get("src/test/resources/trace-v2.1")
+        assertThrows<IllegalArgumentException> { format.getDetails(path, "test") }
+    }
+
+    @ParameterizedTest
+    @ValueSource(strings = ["trace-v2.0", "trace-v2.1"])
+    fun testResources(name: String) {
+        val path = Paths.get("src/test/resources/$name")
+        val reader = format.newReader(path, TABLE_RESOURCES)
+
+        assertAll(
+            { assertTrue(reader.nextRow()) },
+            { assertEquals("1019", reader.get(RESOURCE_ID)) },
+            { assertTrue(reader.nextRow()) },
+            { assertEquals("1023", reader.get(RESOURCE_ID)) },
+            { assertTrue(reader.nextRow()) },
+            { assertEquals("1052", reader.get(RESOURCE_ID)) },
+            { assertTrue(reader.nextRow()) },
+            { assertEquals("1073", reader.get(RESOURCE_ID)) },
+            { assertFalse(reader.nextRow()) }
+        )
+
+        reader.close()
+    }
+
+    @ParameterizedTest
+    @ValueSource(strings = ["trace-v2.0", "trace-v2.1"])
+    fun testSmoke(name: String) {
+        val path = Paths.get("src/test/resources/$name")
+        val reader = format.newReader(path, TABLE_RESOURCE_STATES)
+
+        assertAll(
+            { assertTrue(reader.nextRow()) },
+            { assertEquals("1019", reader.get(RESOURCE_ID)) },
+            { assertEquals(1376314846, reader.get(RESOURCE_STATE_TIMESTAMP).epochSecond) },
+            { assertEquals(0.0, reader.getDouble(RESOURCE_STATE_CPU_USAGE), 0.01) }
+        )
+
+        reader.close()
+    }
+}
diff --git a/opendc-trace/opendc-trace-opendc/src/test/resources/trace-v2.0/meta.parquet b/opendc-trace/opendc-trace-opendc/src/test/resources/trace-v2.0/meta.parquet
new file mode 100644
index 00000000..d6ff09d8
--- /dev/null
+++ b/opendc-trace/opendc-trace-opendc/src/test/resources/trace-v2.0/meta.parquet
diff --git a/opendc-trace/opendc-trace-opendc/src/test/resources/trace-v2.0/trace.parquet b/opendc-trace/opendc-trace-opendc/src/test/resources/trace-v2.0/trace.parquet
new file mode 100644
index 00000000..5b6fa6b7
--- /dev/null
+++ b/opendc-trace/opendc-trace-opendc/src/test/resources/trace-v2.0/trace.parquet
diff --git a/opendc-trace/opendc-trace-opendc/src/test/resources/trace-v2.1/meta.parquet b/opendc-trace/opendc-trace-opendc/src/test/resources/trace-v2.1/meta.parquet
new file mode 100644
index 00000000..d8184945
--- /dev/null
+++ b/opendc-trace/opendc-trace-opendc/src/test/resources/trace-v2.1/meta.parquet
diff --git a/opendc-trace/opendc-trace-opendc/src/test/resources/trace-v2.1/trace.parquet b/opendc-trace/opendc-trace-opendc/src/test/resources/trace-v2.1/trace.parquet
new file mode 100644
index 00000000..00ab5835
--- /dev/null
+++ b/opendc-trace/opendc-trace-opendc/src/test/resources/trace-v2.1/trace.parquet