store results from ResponseBodyProcessor

* also updates CSVStorage * adds Storable interface
dice-group · Aug 30, 2023 · d242722 · d242722
1 parent 79cd8fd
commit d242722
Show file tree

Hide file tree

Showing 11 changed files with 271 additions and 60 deletions.
diff --git a/src/main/java/org/aksw/iguana/cc/lang/LanguageProcessor.java b/src/main/java/org/aksw/iguana/cc/lang/LanguageProcessor.java
@@ -26,10 +26,11 @@ public abstract class LanguageProcessor {
     }
 
     public interface LanguageProcessingData {
+        long hash();
         Class<? extends LanguageProcessor> processor();
     }
 
-    public abstract LanguageProcessingData process(InputStream inputStream);
+    public abstract LanguageProcessingData process(InputStream inputStream, long hash);
 
     final private static Map<String, Class<? extends LanguageProcessor>> processors = new HashMap<>();
 

diff --git a/src/main/java/org/aksw/iguana/cc/lang/impl/SaxSparqlJsonResultCountingParser.java b/src/main/java/org/aksw/iguana/cc/lang/impl/SaxSparqlJsonResultCountingParser.java
@@ -1,6 +1,13 @@
 package org.aksw.iguana.cc.lang.impl;
 
 import org.aksw.iguana.cc.lang.LanguageProcessor;
+import org.aksw.iguana.cc.storage.Storable;
+import org.aksw.iguana.commons.rdf.IPROP;
+import org.aksw.iguana.commons.rdf.IRES;
+import org.apache.jena.rdf.model.Model;
+import org.apache.jena.rdf.model.ModelFactory;
+import org.apache.jena.rdf.model.Resource;
+import org.apache.jena.rdf.model.ResourceFactory;
 import org.json.simple.parser.ContentHandler;
 import org.json.simple.parser.JSONParser;
 import org.json.simple.parser.ParseException;
@@ -24,26 +31,65 @@
 public class SaxSparqlJsonResultCountingParser extends LanguageProcessor {
 
     @Override
-    public LanguageProcessingData process(InputStream inputStream) {
+    public LanguageProcessingData process(InputStream inputStream, long hash) {
         var parser = new JSONParser();
         var handler = new SaxSparqlJsonResultContentHandler();
         try {
             parser.parse(new BufferedReader(new InputStreamReader(inputStream, StandardCharsets.UTF_8)), handler);
-            return new SaxSparqlJsonResultData(handler.solutions(), handler.boundValues(), handler.variables(), null);
+            return new SaxSparqlJsonResultData(hash, handler.solutions(), handler.boundValues(), handler.variables(), null);
         } catch (IOException e) {
             throw new RuntimeException(e);
         } catch (ParseException e) {
-            return new SaxSparqlJsonResultData(-1, -1, null, e);
+            return new SaxSparqlJsonResultData(hash, -1, -1, null, e);
         }
     }
 
-    record SaxSparqlJsonResultData(long results, long bindings,
-                                   List<String> variables, Exception exception) implements LanguageProcessingData {
+    record SaxSparqlJsonResultData(
+            long hash,
+            long results,
+            long bindings,
+            List<String> variables,
+            Exception exception
+    ) implements LanguageProcessingData, Storable.AsCSV, Storable.AsRDF {
 
         @Override
         public Class<? extends LanguageProcessor> processor() {
             return SaxSparqlJsonResultCountingParser.class;
         }
+
+        @Override
+        public List<Storable.CSVFileData> toCSV() {
+            String variablesString = "";
+            String exceptionString = "";
+            if (variables != null)
+                variablesString = String.join("; ", variables);
+            if (exception != null)
+                exceptionString = exception().toString();
+
+            String[] header = new String[]{ "responseBodyHash", "results", "bindings", "variables", "exception" };
+            String[] content = new String[]{ String.valueOf(hash), String.valueOf(results), String.valueOf(bindings), variablesString, exceptionString};
+            String[][] data = new String[][]{ header, content };
+            return List.of(new Storable.CSVFileData("sax-sparql-result-data.csv", data));
+        }
+
+        @Override
+        public Model toRDF() {
+            Model m = ModelFactory.createDefaultModel();
+            Resource responseBodyRes = IRES.getResponsebodyResource(this.hash);
+            m.add(responseBodyRes, IPROP.results, ResourceFactory.createTypedLiteral(this.results))
+                    .add(responseBodyRes, IPROP.bindings, ResourceFactory.createTypedLiteral(this.bindings));
+
+            if (this.variables != null) {
+                for (String variable : this.variables) {
+                    m.add(responseBodyRes, IPROP.variable, ResourceFactory.createTypedLiteral(variable));
+                }
+            }
+            if (this.exception != null) {
+                m.add(responseBodyRes, IPROP.exception, ResourceFactory.createTypedLiteral(this.exception.toString()));
+            }
+
+            return m;
+        }
     }
 
     private static class SaxSparqlJsonResultContentHandler implements ContentHandler {

diff --git a/src/main/java/org/aksw/iguana/cc/metrics/impl/AggregatedExecutionStatistics.java b/src/main/java/org/aksw/iguana/cc/metrics/impl/AggregatedExecutionStatistics.java
@@ -79,8 +79,7 @@ private static Model createAggregatedModel(List<HttpWorker.ExecutionStats> data,
 
         m.add(queryRes, IPROP.succeeded, ResourceFactory.createTypedLiteral(succeeded));
         m.add(queryRes, IPROP.failed, ResourceFactory.createTypedLiteral(failed));
-        if (resultSize.isPresent())
-            m.add(queryRes, IPROP.resultSize, ResourceFactory.createTypedLiteral(resultSize.get()));
+        m.add(queryRes, IPROP.resultSize, ResourceFactory.createTypedLiteral(resultSize.orElse(BigInteger.valueOf(-1))));
         m.add(queryRes, IPROP.timeOuts, ResourceFactory.createTypedLiteral(timeOuts));
         m.add(queryRes, IPROP.wrongCodes, ResourceFactory.createTypedLiteral(wrongCodes));
         m.add(queryRes, IPROP.unknownException, ResourceFactory.createTypedLiteral(unknownExceptions));

diff --git a/src/main/java/org/aksw/iguana/cc/metrics/impl/EachExecutionStatistic.java b/src/main/java/org/aksw/iguana/cc/metrics/impl/EachExecutionStatistic.java
@@ -5,6 +5,7 @@
 import org.aksw.iguana.cc.worker.HttpWorker;
 import org.aksw.iguana.commons.rdf.IPROP;
 import org.aksw.iguana.commons.rdf.IRES;
+import org.aksw.iguana.commons.time.TimeUtils;
 import org.apache.jena.rdf.model.Model;
 import org.apache.jena.rdf.model.ModelFactory;
 import org.apache.jena.rdf.model.Resource;
@@ -32,14 +33,19 @@ public Model createMetricModel(List<HttpWorker> workers, List<HttpWorker.Executi
                 for (HttpWorker.ExecutionStats exec : data[(int) worker.getWorkerID()][i]) {
                     Resource runRes = iresFactory.getWorkerQueryRunResource(worker, i, run);
                     m.add(workerQueryResource, IPROP.queryExecution, runRes);
-                    m.add(runRes, IPROP.time, ResourceFactory.createTypedLiteral(exec.duration()));
+                    m.add(runRes, IPROP.time, TimeUtils.createTypedDurationLiteral(exec.duration()));
                     m.add(runRes, IPROP.success, ResourceFactory.createTypedLiteral(exec.successful()));
                     m.add(runRes, IPROP.run, ResourceFactory.createTypedLiteral(run));
                     m.add(runRes, IPROP.code, ResourceFactory.createTypedLiteral(exec.endState().value));
-                    // TODO: maybe add http status code
-                    if (exec.contentLength().isPresent())
-                        m.add(runRes, IPROP.resultSize, ResourceFactory.createTypedLiteral(exec.contentLength().getAsLong()));
+                    m.add(runRes, IPROP.resultSize, ResourceFactory.createTypedLiteral(exec.contentLength().orElse(-1)));
                     m.add(runRes, IPROP.queryID, queryRes);
+                    if (exec.responseBodyHash().isPresent()) {
+                        Resource responseBodyRes = IRES.getResponsebodyResource(exec.responseBodyHash().getAsLong());
+                        m.add(runRes, IPROP.responseBody, responseBodyRes);
+                        m.add(responseBodyRes, IPROP.responseBodyHash, ResourceFactory.createTypedLiteral(exec.responseBodyHash().getAsLong()));
+                    }
+                    // TODO: maybe add http status code and qps
+
                     run = run.add(BigInteger.ONE);
                 }
             }

diff --git a/src/main/java/org/aksw/iguana/cc/storage/Storable.java b/src/main/java/org/aksw/iguana/cc/storage/Storable.java
@@ -0,0 +1,34 @@
+package org.aksw.iguana.cc.storage;
+
+import org.apache.jena.rdf.model.Model;
+
+import java.util.List;
+
+public interface Storable {
+
+    record CSVFileData (
+        String filename,
+        String[][] data
+    ) {}
+
+    interface AsCSV {
+
+        /**
+         * Converts the data into CSV files. The key of the map contains the file name for the linked entries.
+         *
+         * @return CSVFileData list which contains all the files and their data that should be created and stored
+         */
+        List<CSVFileData> toCSV();
+    }
+
+    interface AsRDF {
+
+        /**
+         * Converts the data into an RDF model, which will be added to the appropriate storages.
+         *
+         * @return RDF model that contains the data
+         */
+        Model toRDF();
+    }
+
+}