multiple results in list enabled

Author: emrekenar

src/main/kotlin/no/uio/microobject/ast/stmt/MonitorStmt.kt

@@ -76,32 +76,35 @@ class MonitorObject(private val name: LiteralExpr, private val declaredType: Typ
     }
 
     fun getWindowResults(interpreter: Interpreter): LiteralExpr {
-        val rdfTable = interpreter.streamManager.getQueryResults(name)
+        val rdfTables = interpreter.streamManager.getQueryResults(name)
+        
         var list = LiteralExpr("null")
 
-        if (rdfTable != null) {
+        if (rdfTables != null) {
+            for (rdfTable in rdfTables) {
 
-            val resIt = rdfTable.rows().iterator()
-            val firstVar = rdfTable.varNames.get(0)
-            while (resIt.hasNext()) {
-                val rdfTuple = resIt.next()
-                try {
-                    // only consider first result for now
-                    var literal = iriToLiteral(rdfTuple.get(firstVar).toString(), interpreter)
-                    if (literal.tag != declaredType)
-                        throw Exception("Monitor parameter has incorrect type (expected ${declaredType}, got ${literal.tag})")
+                val resIt = rdfTable.rows().iterator()
+                val firstVar = rdfTable.varNames.get(0)
+                
+                while (resIt.hasNext()) {
+                    val rdfTuple = resIt.next()
+                    try {
+                        // only consider first result for now
+                        var literal = iriToLiteral(rdfTuple.get(firstVar).toString(), interpreter)
+                        if (literal.tag != declaredType)
+                            throw Exception("Monitor parameter has incorrect type (expected ${declaredType}, got ${literal.tag})")
 
-                    val name = Names.getObjName("List")
-                    val newMemory: Memory = mutableMapOf()                    
-                    newMemory["content"] = literal
-                    newMemory["next"] = list
-                    interpreter.heap[name] = newMemory
-                    list = name
+                        val name = Names.getObjName("List")
+                        val newMemory: Memory = mutableMapOf()
+                        newMemory["content"] = literal
+                        newMemory["next"] = list
+                        interpreter.heap[name] = newMemory
+                        list = name
 
-                } catch (e: Exception) {
-                    throw Exception("Error while processing query result: ${e.message}")
+                    } catch (e: Exception) {
+                        throw Exception("Error while processing query result: ${e.message}")
+                    }
                 }
-
             }
         }
         return list

src/main/kotlin/no/uio/microobject/data/StreamManager.kt

@@ -38,6 +38,9 @@ import org.streamreasoning.rsp4j.api.sds.SDSConfiguration
 import org.apache.jena.rdf.model.*;
 import org.apache.jena.atlas.lib.tuple.Tuple
 
+// Per-monitor state
+data class WindowBuf(var tick: Long? = null, val current: MutableList<Table> = mutableListOf())
+
 // Class managing streams
 class StreamManager(private val settings: Settings, val staticTable: StaticTable, private val interpreter: Interpreter?) {
 
@@ -50,12 +53,14 @@ class StreamManager(private val settings: Settings, val staticTable: StaticTable
     private var streams: MutableMap<String, MutableMap<LiteralExpr, StreamObject>> = mutableMapOf()
     private var monitors: MutableMap<LiteralExpr, MonitorObject> = mutableMapOf()
 
-    private var queryResults: MutableMap<LiteralExpr, Table?> = ConcurrentHashMap()
+    private val windowBufs = ConcurrentHashMap<LiteralExpr, WindowBuf>()
+    private val queryResults = ConcurrentHashMap<LiteralExpr, List<Table>>() // read-only snapshots
 
     var clockVar : String? = null
     var clockTimestampSec : String? = null
     var lastTimestamp: Long = 0
-    
+    var lastWindowTs: Long? = null
+
     var nStaticGraphsPushed = 0
 
     init {
@@ -89,12 +94,12 @@ class StreamManager(private val settings: Settings, val staticTable: StaticTable
                 sdsConfig = SDSConfiguration(defaultPath)
             }
 
-            engine = CSPARQLEngine(0, ec)
+            engine = CSPARQLEngine(ts, ec)
             engineInitialized = true
         }
     }
 
-    public fun getQueryResults(name: LiteralExpr): Table? {
+    public fun getQueryResults(name: LiteralExpr): List<Table>? {
         return queryResults[name]
     }
 
@@ -130,13 +135,27 @@ class StreamManager(private val settings: Settings, val staticTable: StaticTable
         val stream = streams[className]!![obj]!!
         val expressions = interpreter!!.staticInfo.streamersTable[className]!![methodName]!!
 
-        val subjIri = "${settings.runPrefix}${obj.literal}"
         val timestamp = getTimestamp()
 
         for (expr in expressions) {
             val res = interpreter.eval(expr, stackEntry)
 
-            val predIri = "${settings.progPrefix}${className}_${expr.toString().removePrefix("this.").replace('.', '_')}"
+            val exprParts = expr.toString().split('.')
+            var subjIri: String = "${settings.runPrefix}${obj.literal}"
+            var predIri = "${settings.progPrefix}${className}_${expr.toString().removePrefix("this.").replace('.', '_')}"
+            if (exprParts.size < 2) {
+                throw Exception("Streamer expression must be of the form this.field or this.obj.field")
+            }
+            if (exprParts.size > 2) {
+                // find the one before last part
+                var currentObj: LiteralExpr = obj
+                for (i in 1 until exprParts.size - 1) {
+                    val fieldName = exprParts[i]
+                    currentObj = interpreter.heap[currentObj]!![fieldName]!!
+                }
+                subjIri = "${settings.runPrefix}${currentObj.literal}"
+                predIri = "${settings.progPrefix}${(currentObj.tag as BaseType).name}_${exprParts.last()}"
+            }
 
             val m = ModelFactory.createDefaultModel()
             val stmt = m.createStatement(m.createResource(subjIri), m.createProperty(predIri), literalToIri(m, res, settings))
@@ -162,14 +181,25 @@ class StreamManager(private val settings: Settings, val staticTable: StaticTable
     public fun registerQuery(name: LiteralExpr, queryExpr : Expression, params: List<Expression>, stackMemory: Memory, heap: GlobalMemory, obj: LiteralExpr, SPARQL : Boolean = true, declaredType: Type): JenaContinuousQueryExecution {
         initEngineIfNeeded()
         val queryStr = prepareQuery(name, queryExpr, params, stackMemory, heap, obj, SPARQL)
+        if (settings.verbose) println("Registering query:\n$queryStr")
 
         val cqe = engine!!.register(queryStr, sdsConfig) as JenaContinuousQueryExecution
         monitors[name] = MonitorObject(name, declaredType)
 
+        windowBufs.computeIfAbsent(name) { WindowBuf() }
+
         val outputStream = cqe.outstream()
-        outputStream?.addConsumer { arg, ts -> 
-            val results: Table = arg as Table
-            queryResults[name] = results
+        outputStream?.addConsumer { arg, ts ->
+            val table = arg as Table
+            val b = windowBufs.getOrPut(name) { WindowBuf() }
+
+            if (b.tick != ts) {
+                b.tick = ts
+                b.current.clear()
+            }
+            b.current += table
+
+            queryResults[name] = b.current.toList()
         }
 
         return cqe
@@ -181,7 +211,7 @@ class StreamManager(private val settings: Settings, val staticTable: StaticTable
         var prefixes = ""
         for ((key, value) in settings.prefixMap()) prefixes += "PREFIX $key: <$value>\n"
 
-        val queryHeader = "REGISTER RSTREAM <${settings.runPrefix}${name.literal}> AS "
+        val queryHeader = "REGISTER RSTREAM <${settings.runPrefix}${name.literal}/out> AS "
 
         val queryBody = interpreter!!.prepareQuery(queryExpr, params, stackMemory, heap, obj, SPARQL)
             .removePrefix("\"").removeSuffix("\"")

src/main/resources/default-csparql2.properties

@@ -1,18 +1,17 @@
 rsp_engine.time=EventTime
-# rsp_engine.base_uri=http://streamreasoning.org/csparql/
 rsp_engine.stream.item.class=org.streamreasoning.rsp4j.csparql2.stream.GraphStreamSchema
 
 rsp_engine.response_format=JSON-LD
 
-rsp_engine.on_window_close=true
+rsp_engine.on_window_close=false
 rsp_engine.non_empty_content=true
 rsp_engine.periodic=false
 rsp_engine.on_content_change=false
 rsp_engine.tick=TIME_DRIVEN
 
 rsp_engine.report_grain=SINGLE
 rsp_engine.sds.mantainance=NAIVE
-rsp_engine.partialwindow=false
+rsp_engine.partialwindow=true
 
 # jasper.entailment=RDFS
 # rsp_engine.tbox_location=artist.tbox.owl

src/test/resources/csparql/04_no_aggregation.smol

@@ -1,16 +1,23 @@
-streamer class C(Int x, Monitor<Int> m)
+// reada src/test/resources/csparql/04_no_aggregation.smol
+
+class Observation(Int x, Int ts) end
+
+streamer class C(Int x, Observation obs, Monitor<Int> m)
 
     Unit register()
         Monitor<Int> m = monitor("
             SELECT ?x
-            FROM STREAM %1 [RANGE 5s STEP 2s]
-            WHERE { ?s prog:C_x ?x }
+            FROM NAMED WINDOW <win> ON %1 [RANGE PT5.001s STEP PT2s]
+            WHERE {
+                WINDOW ?w { ?s prog:Observation_x ?x }
+            }
         ", this);
         this.m = m;
     end
 
-    Unit doStream() emits(this.x)
+    Unit doStream(Int ts) emits(this.obs, this.obs.x, this.obs.ts)
         this.x = this.x + 1;
+        this.obs = new Observation(this.x, ts);
     end
 
     String windowToString()
@@ -25,16 +32,27 @@ streamer class C(Int x, Monitor<Int> m)
 end
 
 main
-    C o = new C(0, null);
-    o.register();
-
+    // configure simulation
     clock Int i = 100;
     Int endAt = 200;
 
+    // initialize stream
+    C o = new C(0, null, null);
+
+    // initialize monitor
+    o.register();
+
+    // run simulation
     while i < endAt do
-        o.doStream();
+
+        // put triples in the stream
+        o.doStream(i);
+
+        // read window contents
         String res = o.windowToString();
         print(">>" ++ intToString(i) ++ ": " ++ res);
+
+        // advance clock (timestamp)
         i = i + 1;
     end
 end
@@ -52,4 +70,5 @@ end
 // ..
 
 // Notes:
-// - Multiple elements in the window because there is no aggregation function
+// - Step 2 seconds = new window every 2 seconds
+// - Range 5.001s: Windows have {2, 4, 5, 5, 5, ...} elements