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.

1 files changed, 39 insertions, 30 deletions

diff --git a/opendc-trace/opendc-trace-gwf/src/main/kotlin/org/opendc/trace/gwf/GwfTaskTableReader.kt b/opendc-trace/opendc-trace-gwf/src/main/kotlin/org/opendc/trace/gwf/GwfTaskTableReader.kt
index 39eb5520..aa4c543b 100644
--- a/opendc-trace/opendc-trace-gwf/src/main/kotlin/org/opendc/trace/gwf/GwfTaskTableReader.kt
+++ b/opendc-trace/opendc-trace-gwf/src/main/kotlin/org/opendc/trace/gwf/GwfTaskTableReader.kt
@@ -67,52 +67,43 @@ internal class GwfTaskTableReader(private val parser: CsvParser) : TableReader {
         return true
     }
 
-    override fun hasColumn(column: TableColumn<*>): Boolean {
-        return when (column) {
-            TASK_WORKFLOW_ID -> true
-            TASK_ID -> true
-            TASK_SUBMIT_TIME -> true
-            TASK_RUNTIME -> true
-            TASK_REQ_NCPUS -> true
-            TASK_ALLOC_NCPUS -> true
-            TASK_PARENTS -> 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" }
+        return false
     }
 
-    override fun <T> get(column: TableColumn<T>): T {
-        val res: Any? = when (column) {
-            TASK_WORKFLOW_ID -> workflowId
-            TASK_ID -> jobId
-            TASK_SUBMIT_TIME -> submitTime
-            TASK_RUNTIME -> runtime
-            TASK_REQ_NCPUS -> nProcs
-            TASK_ALLOC_NCPUS -> reqNProcs
-            TASK_PARENTS -> dependencies
+    override fun get(index: Int): Any? {
+        return when (index) {
+            COL_JOB_ID -> jobId
+            COL_WORKFLOW_ID -> workflowId
+            COL_SUBMIT_TIME -> submitTime
+            COL_RUNTIME -> runtime
+            COL_REQ_NPROC -> getInt(index)
+            COL_NPROC -> getInt(index)
+            COL_DEPS -> dependencies
             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) {
-            TASK_REQ_NCPUS -> nProcs
-            TASK_ALLOC_NCPUS -> reqNProcs
+    override fun getInt(index: Int): Int {
+        return when (index) {
+            COL_REQ_NPROC -> reqNProcs
+            COL_NPROC -> nProcs
             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 {
+    override fun getDouble(index: Int): Double {
         throw IllegalArgumentException("Invalid column")
     }
 
@@ -180,6 +171,24 @@ internal class GwfTaskTableReader(private val parser: CsvParser) : TableReader {
         dependencies = emptySet()
     }
 
+    private val COL_WORKFLOW_ID = 0
+    private val COL_JOB_ID = 1
+    private val COL_SUBMIT_TIME = 2
+    private val COL_RUNTIME = 3
+    private val COL_NPROC = 4
+    private val COL_REQ_NPROC = 5
+    private val COL_DEPS = 6
+
+    private val columns = mapOf(
+        TASK_ID to COL_JOB_ID,
+        TASK_WORKFLOW_ID to COL_WORKFLOW_ID,
+        TASK_SUBMIT_TIME to COL_SUBMIT_TIME,
+        TASK_RUNTIME to COL_RUNTIME,
+        TASK_ALLOC_NCPUS to COL_NPROC,
+        TASK_REQ_NCPUS to COL_REQ_NPROC,
+        TASK_PARENTS to COL_DEPS
+    )
+
     companion object {
         /**
          * The [CsvSchema] that is used to parse the trace.