feat(trace): Support column lookup via index

This change adds support for looking up the column value through the
column index. This enables faster lookup when processing very large
traces.

5 files changed, 125 insertions, 214 deletions

diff --git a/opendc-trace/opendc-trace-bitbrains/src/main/kotlin/org/opendc/trace/bitbrains/BitbrainsExResourceStateTable.kt b/opendc-trace/opendc-trace-bitbrains/src/main/kotlin/org/opendc/trace/bitbrains/BitbrainsExResourceStateTable.kt
index 44a6c26e..ab768608 100644
--- a/opendc-trace/opendc-trace-bitbrains/src/main/kotlin/org/opendc/trace/bitbrains/BitbrainsExResourceStateTable.kt
+++ b/opendc-trace/opendc-trace-bitbrains/src/main/kotlin/org/opendc/trace/bitbrains/BitbrainsExResourceStateTable.kt
@@ -23,6 +23,7 @@
 package org.opendc.trace.bitbrains
 
 import org.opendc.trace.*
+import org.opendc.trace.util.CompositeTableReader
 import java.nio.file.Files
 import java.nio.file.Path
 import java.util.stream.Collectors
@@ -64,57 +65,8 @@ internal class BitbrainsExResourceStateTable(path: Path) : Table {
     override fun newReader(): TableReader {
         val it = partitions.iterator()
 
-        return object : TableReader {
-            var delegate: TableReader? = nextDelegate()
-
-            override fun nextRow(): Boolean {
-                var delegate = delegate
-
-                while (delegate != null) {
-                    if (delegate.nextRow()) {
-                        break
-                    }
-
-                    delegate.close()
-                    delegate = nextDelegate()
-                    this.delegate = delegate
-                }
-
-                return delegate != null
-            }
-
-            override fun hasColumn(column: TableColumn<*>): Boolean = delegate?.hasColumn(column) ?: false
-
-            override fun <T> get(column: TableColumn<T>): T {
-                val delegate = checkNotNull(delegate) { "Invalid reader state" }
-                return delegate.get(column)
-            }
-
-            override fun getBoolean(column: TableColumn<Boolean>): Boolean {
-                val delegate = checkNotNull(delegate) { "Invalid reader state" }
-                return delegate.getBoolean(column)
-            }
-
-            override fun getInt(column: TableColumn<Int>): Int {
-                val delegate = checkNotNull(delegate) { "Invalid reader state" }
-                return delegate.getInt(column)
-            }
-
-            override fun getLong(column: TableColumn<Long>): Long {
-                val delegate = checkNotNull(delegate) { "Invalid reader state" }
-                return delegate.getLong(column)
-            }
-
-            override fun getDouble(column: TableColumn<Double>): Double {
-                val delegate = checkNotNull(delegate) { "Invalid reader state" }
-                return delegate.getDouble(column)
-            }
-
-            override fun close() {
-                delegate?.close()
-            }
-
-            private fun nextDelegate(): TableReader? {
+        return object : CompositeTableReader() {
+            override fun nextReader(): TableReader? {
                 return if (it.hasNext()) {
                     val (_, path) = it.next()
                     val reader = path.bufferedReader()
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 5619e839..c1b6f5ba 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
@@ -88,70 +88,53 @@ internal class BitbrainsExResourceStateTableReader(private val reader: BufferedR
         return true
     }
 
-    override fun hasColumn(column: TableColumn<*>): Boolean {
-        return when (column) {
-            RESOURCE_ID -> true
-            RESOURCE_CLUSTER_ID -> true
-            RESOURCE_STATE_TIMESTAMP -> true
-            RESOURCE_CPU_COUNT -> true
-            RESOURCE_CPU_CAPACITY -> true
-            RESOURCE_STATE_CPU_USAGE -> true
-            RESOURCE_STATE_CPU_USAGE_PCT -> true
-            RESOURCE_STATE_CPU_DEMAND -> true
-            RESOURCE_STATE_CPU_READY_PCT -> true
-            RESOURCE_MEM_CAPACITY -> true
-            RESOURCE_STATE_DISK_READ -> true
-            RESOURCE_STATE_DISK_WRITE -> true
-            else -> false
-        }
+    override fun resolve(column: TableColumn<*>): Int = columns[column] ?: -1
+
+    override fun isNull(index: Int): Boolean {
+        require(index in 0..COL_MAX) { "Invalid column index" }
+        return false
     }
 
-    override fun <T> get(column: TableColumn<T>): T {
-        val res: Any? = when (column) {
-            RESOURCE_ID -> id
-            RESOURCE_CLUSTER_ID -> cluster
-            RESOURCE_STATE_TIMESTAMP -> timestamp
-            RESOURCE_CPU_COUNT -> getInt(RESOURCE_CPU_COUNT)
-            RESOURCE_CPU_CAPACITY -> getDouble(RESOURCE_CPU_CAPACITY)
-            RESOURCE_STATE_CPU_USAGE -> getDouble(RESOURCE_STATE_CPU_USAGE)
-            RESOURCE_STATE_CPU_USAGE_PCT -> getDouble(RESOURCE_STATE_CPU_USAGE_PCT)
-            RESOURCE_MEM_CAPACITY -> getDouble(RESOURCE_MEM_CAPACITY)
-            RESOURCE_STATE_DISK_READ -> getDouble(RESOURCE_STATE_DISK_READ)
-            RESOURCE_STATE_DISK_WRITE -> getDouble(RESOURCE_STATE_DISK_WRITE)
+    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")
         }
-
-        @Suppress("UNCHECKED_CAST")
-        return res as T
     }
 
-    override fun getBoolean(column: TableColumn<Boolean>): Boolean {
-        return when (column) {
-            RESOURCE_STATE_POWERED_ON -> poweredOn
+    override fun getBoolean(index: Int): Boolean {
+        return when (index) {
+            COL_POWERED_ON -> poweredOn
             else -> throw IllegalArgumentException("Invalid column")
         }
     }
 
-    override fun getInt(column: TableColumn<Int>): Int {
-        return when (column) {
-            RESOURCE_CPU_COUNT -> cpuCores
+    override fun getInt(index: Int): Int {
+        return when (index) {
+            COL_NCPUS -> cpuCores
             else -> throw IllegalArgumentException("Invalid column")
         }
     }
 
-    override fun getLong(column: TableColumn<Long>): Long {
+    override fun getLong(index: Int): Long {
         throw IllegalArgumentException("Invalid column")
     }
 
-    override fun getDouble(column: TableColumn<Double>): Double {
-        return when (column) {
-            RESOURCE_CPU_CAPACITY -> cpuCapacity
-            RESOURCE_STATE_CPU_USAGE -> cpuUsage
-            RESOURCE_STATE_CPU_USAGE_PCT -> cpuUsage / cpuCapacity
-            RESOURCE_STATE_CPU_DEMAND -> cpuDemand
-            RESOURCE_MEM_CAPACITY -> memCapacity
-            RESOURCE_STATE_DISK_READ -> diskRead
-            RESOURCE_STATE_DISK_WRITE -> diskWrite
+    override fun getDouble(index: Int): Double {
+        return when (index) {
+            COL_CPU_CAPACITY -> cpuCapacity
+            COL_CPU_USAGE -> cpuUsage
+            COL_CPU_USAGE_PCT -> cpuUsage / cpuCapacity
+            COL_CPU_READY_PCT -> cpuReadyPct
+            COL_CPU_DEMAND -> cpuDemand
+            COL_MEM_CAPACITY -> memCapacity
+            COL_DISK_READ -> diskRead
+            COL_DISK_WRITE -> diskWrite
             else -> throw IllegalArgumentException("Invalid column")
         }
     }
@@ -209,4 +192,21 @@ internal class BitbrainsExResourceStateTableReader(private val reader: BufferedR
     private val COL_CPU_CAPACITY = 18
     private val COL_ID = 19
     private val COL_MEM_CAPACITY = 20
+    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
+    )
 }
diff --git a/opendc-trace/opendc-trace-bitbrains/src/main/kotlin/org/opendc/trace/bitbrains/BitbrainsResourceStateTable.kt b/opendc-trace/opendc-trace-bitbrains/src/main/kotlin/org/opendc/trace/bitbrains/BitbrainsResourceStateTable.kt
index f68e61dc..6b6ac9da 100644
--- a/opendc-trace/opendc-trace-bitbrains/src/main/kotlin/org/opendc/trace/bitbrains/BitbrainsResourceStateTable.kt
+++ b/opendc-trace/opendc-trace-bitbrains/src/main/kotlin/org/opendc/trace/bitbrains/BitbrainsResourceStateTable.kt
@@ -24,6 +24,7 @@ package org.opendc.trace.bitbrains
 
 import com.fasterxml.jackson.dataformat.csv.CsvFactory
 import org.opendc.trace.*
+import org.opendc.trace.util.CompositeTableReader
 import java.nio.file.Files
 import java.nio.file.Path
 import java.util.stream.Collectors
@@ -65,57 +66,8 @@ internal class BitbrainsResourceStateTable(private val factory: CsvFactory, path
     override fun newReader(): TableReader {
         val it = partitions.iterator()
 
-        return object : TableReader {
-            var delegate: TableReader? = nextDelegate()
-
-            override fun nextRow(): Boolean {
-                var delegate = delegate
-
-                while (delegate != null) {
-                    if (delegate.nextRow()) {
-                        break
-                    }
-
-                    delegate.close()
-                    delegate = nextDelegate()
-                    this.delegate = delegate
-                }
-
-                return delegate != null
-            }
-
-            override fun hasColumn(column: TableColumn<*>): Boolean = delegate?.hasColumn(column) ?: false
-
-            override fun <T> get(column: TableColumn<T>): T {
-                val delegate = checkNotNull(delegate) { "Invalid reader state" }
-                return delegate.get(column)
-            }
-
-            override fun getBoolean(column: TableColumn<Boolean>): Boolean {
-                val delegate = checkNotNull(delegate) { "Invalid reader state" }
-                return delegate.getBoolean(column)
-            }
-
-            override fun getInt(column: TableColumn<Int>): Int {
-                val delegate = checkNotNull(delegate) { "Invalid reader state" }
-                return delegate.getInt(column)
-            }
-
-            override fun getLong(column: TableColumn<Long>): Long {
-                val delegate = checkNotNull(delegate) { "Invalid reader state" }
-                return delegate.getLong(column)
-            }
-
-            override fun getDouble(column: TableColumn<Double>): Double {
-                val delegate = checkNotNull(delegate) { "Invalid reader state" }
-                return delegate.getDouble(column)
-            }
-
-            override fun close() {
-                delegate?.close()
-            }
-
-            private fun nextDelegate(): TableReader? {
+        return object : CompositeTableReader() {
+            override fun nextReader(): TableReader? {
                 return if (it.hasNext()) {
                     val (partition, path) = it.next()
                     return BitbrainsResourceStateTableReader(partition, factory.createParser(path.toFile()))
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 54be5dea..3a8839b4 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
@@ -111,71 +111,49 @@ internal class BitbrainsResourceStateTableReader(private val partition: String,
         return true
     }
 
-    override fun hasColumn(column: TableColumn<*>): Boolean {
-        return when (column) {
-            RESOURCE_ID -> true
-            RESOURCE_STATE_TIMESTAMP -> true
-            RESOURCE_CPU_COUNT -> true
-            RESOURCE_CPU_CAPACITY -> true
-            RESOURCE_STATE_CPU_USAGE -> true
-            RESOURCE_STATE_CPU_USAGE_PCT -> true
-            RESOURCE_MEM_CAPACITY -> true
-            RESOURCE_STATE_MEM_USAGE -> true
-            RESOURCE_STATE_DISK_READ -> true
-            RESOURCE_STATE_DISK_WRITE -> true
-            RESOURCE_STATE_NET_RX -> true
-            RESOURCE_STATE_NET_TX -> true
-            else -> false
-        }
+    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 false
     }
 
-    override fun <T> get(column: TableColumn<T>): T {
-        val res: Any? = when (column) {
-            RESOURCE_ID -> partition
-            RESOURCE_STATE_TIMESTAMP -> timestamp
-            RESOURCE_CPU_COUNT -> cpuCores
-            RESOURCE_CPU_CAPACITY -> cpuCapacity
-            RESOURCE_STATE_CPU_USAGE -> cpuUsage
-            RESOURCE_STATE_CPU_USAGE_PCT -> cpuUsagePct
-            RESOURCE_MEM_CAPACITY -> memCapacity
-            RESOURCE_STATE_MEM_USAGE -> memUsage
-            RESOURCE_STATE_DISK_READ -> diskRead
-            RESOURCE_STATE_DISK_WRITE -> diskWrite
-            RESOURCE_STATE_NET_RX -> netReceived
-            RESOURCE_STATE_NET_TX -> netTransmitted
+    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")
         }
-
-        @Suppress("UNCHECKED_CAST")
-        return res as T
     }
 
-    override fun getBoolean(column: TableColumn<Boolean>): Boolean {
+    override fun getBoolean(index: Int): Boolean {
         throw IllegalArgumentException("Invalid column")
     }
 
-    override fun getInt(column: TableColumn<Int>): Int {
-        return when (column) {
-            RESOURCE_CPU_COUNT -> cpuCores
+    override fun getInt(index: Int): Int {
+        return when (index) {
+            COL_CPU_COUNT -> cpuCores
             else -> throw IllegalArgumentException("Invalid column")
         }
     }
 
-    override fun getLong(column: TableColumn<Long>): Long {
+    override fun getLong(index: Int): Long {
         throw IllegalArgumentException("Invalid column")
     }
 
-    override fun getDouble(column: TableColumn<Double>): Double {
-        return when (column) {
-            RESOURCE_CPU_CAPACITY -> cpuCapacity
-            RESOURCE_STATE_CPU_USAGE -> cpuUsage
-            RESOURCE_STATE_CPU_USAGE_PCT -> cpuUsagePct
-            RESOURCE_MEM_CAPACITY -> memCapacity
-            RESOURCE_STATE_MEM_USAGE -> memUsage
-            RESOURCE_STATE_DISK_READ -> diskRead
-            RESOURCE_STATE_DISK_WRITE -> diskWrite
-            RESOURCE_STATE_NET_RX -> netReceived
-            RESOURCE_STATE_NET_TX -> netTransmitted
+    override fun getDouble(index: Int): Double {
+        return when (index) {
+            COL_CPU_CAPACITY -> cpuCapacity
+            COL_CPU_USAGE -> cpuUsage
+            COL_CPU_USAGE_PCT -> cpuUsagePct
+            COL_MEM_CAPACITY -> memCapacity
+            COL_MEM_USAGE -> memUsage
+            COL_DISK_READ -> diskRead
+            COL_DISK_WRITE -> diskWrite
+            COL_NET_RX -> netReceived
+            COL_NET_TX -> netTransmitted
             else -> throw IllegalArgumentException("Invalid column")
         }
     }
@@ -249,6 +227,34 @@ 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 146c04f0..3701994a 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
@@ -43,13 +43,14 @@ internal class BitbrainsResourceTableReader(private val factory: CsvFactory, vms
 
             val parser = factory.createParser(path.toFile())
             val reader = BitbrainsResourceStateTableReader(name, parser)
+            val idCol = reader.resolve(RESOURCE_ID)
 
             try {
                 if (!reader.nextRow()) {
                     continue
                 }
 
-                id = reader.get(RESOURCE_ID)
+                id = reader.get(idCol) as String
                 return true
             } finally {
                 reader.close()
@@ -59,36 +60,33 @@ internal class BitbrainsResourceTableReader(private val factory: CsvFactory, vms
         return false
     }
 
-    override fun hasColumn(column: TableColumn<*>): Boolean {
-        return when (column) {
-            RESOURCE_ID -> true
-            else -> false
-        }
+    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 false
     }
 
-    override fun <T> get(column: TableColumn<T>): T {
-        val res: Any? = when (column) {
-            RESOURCE_ID -> id
+    override fun get(index: Int): Any? {
+        return when (index) {
+            COL_ID -> id
             else -> throw IllegalArgumentException("Invalid column")
         }
-
-        @Suppress("UNCHECKED_CAST")
-        return res as T
     }
 
-    override fun getBoolean(column: TableColumn<Boolean>): Boolean {
+    override fun getBoolean(index: Int): Boolean {
         throw IllegalArgumentException("Invalid column")
     }
 
-    override fun getInt(column: TableColumn<Int>): Int {
+    override fun getInt(index: Int): Int {
         throw IllegalArgumentException("Invalid column")
     }
 
-    override fun getLong(column: TableColumn<Long>): Long {
+    override fun getLong(index: Int): Long {
         throw IllegalArgumentException("Invalid column")
     }
 
-    override fun getDouble(column: TableColumn<Double>): Double {
+    override fun getDouble(index: Int): Double {
         throw IllegalArgumentException("Invalid column")
     }
 
@@ -105,4 +103,7 @@ internal class BitbrainsResourceTableReader(private val factory: CsvFactory, vms
     private fun reset() {
         id = null
     }
+
+    private val COL_ID = 0
+    private val columns = mapOf(RESOURCE_ID to COL_ID)
 }