diff --git a/README.md b/README.md
index addfbb37..18ec1c78 100644
--- a/README.md
+++ b/README.md
@@ -101,6 +101,7 @@ select ?time ?value {
## Building
* This is a plain Maven project
* a full build can be executed via `mvn package`
+* KVIN ingestion benchmark instructions are in [docs/benchmarks/kvin-ingestion.md](docs/benchmarks/kvin-ingestion.md)
## Running
* change to the folder `launch/equinox`
diff --git a/bundles/io.github.linkedfactory.service/pom.xml b/bundles/io.github.linkedfactory.service/pom.xml
index 069651de..dd6ccbcc 100644
--- a/bundles/io.github.linkedfactory.service/pom.xml
+++ b/bundles/io.github.linkedfactory.service/pom.xml
@@ -222,4 +222,55 @@
+
+
+
+ jmh
+
+ io.github.linkedfactory.service.benchmark.KvinIngestionBenchmark
+ json
+ ${project.build.directory}/jmh-result.json
+ 3
+ 3s
+ 5
+ 3s
+ 2
+
+
+
+
+
+ org.codehaus.mojo
+ exec-maven-plugin
+ 3.5.0
+
+ ${java.home}/bin/java
+ test
+
+ -Djmh.temp.root=${jmh.temp.root}
+ -cp
+
+ org.openjdk.jmh.Main
+ ${jmh.includes}
+ -wi
+ ${jmh.warmups}
+ -w
+ ${jmh.warmup.time}
+ -i
+ ${jmh.measurements}
+ -r
+ ${jmh.measurement.time}
+ -f
+ ${jmh.forks}
+ -rf
+ ${jmh.result.format}
+ -rff
+ ${jmh.result.file}
+
+
+
+
+
+
+
diff --git a/bundles/io.github.linkedfactory.service/src/test/java/io/github/linkedfactory/core/kvin/KvinHttpTest.java b/bundles/io.github.linkedfactory.service/src/test/java/io/github/linkedfactory/core/kvin/KvinHttpTest.java
index 37803c67..86d90d75 100644
--- a/bundles/io.github.linkedfactory.service/src/test/java/io/github/linkedfactory/core/kvin/KvinHttpTest.java
+++ b/bundles/io.github.linkedfactory.service/src/test/java/io/github/linkedfactory/core/kvin/KvinHttpTest.java
@@ -92,10 +92,12 @@ public Function0> apply(Req in) {
}
@BeforeClass
- public static void setupClass() throws ClassNotFoundException {
+ public static void setupClass() {
// create configuration and a model set factory
- KommaModule module = ModelPlugin.createModelSetModule(Class.forName("net.enilink.komma.model.ModelPlugin").getClassLoader());
- IModelSetFactory factory = (IModelSetFactory) Guice.createInjector(new ModelSetModule(module)).getInstance(Class.forName("net.enilink.komma.model.IModelSetFactory"));
+ KommaModule module = ModelPlugin.createModelSetModule(ModelPlugin.class.getClassLoader());
+ IModelSetFactory factory =
+ Guice.createInjector(new ModelSetModule(module))
+ .getInstance(IModelSetFactory.class);
// create a model set with an in-memory repository
modelSet = factory.createModelSet(MODELS.NAMESPACE_URI.appendFragment("MemoryModelSet"));
Globals.contextModelSet().theDefault().set(VendorJ.vendor(new Full(modelSet)));
@@ -315,4 +317,4 @@ private static List generateTuples(int numberOfItems, int numberOfPro
.generate()
.toList();
}
-}
\ No newline at end of file
+}
diff --git a/bundles/io.github.linkedfactory.service/src/test/java/io/github/linkedfactory/service/benchmark/KvinIngestionBenchmark.java b/bundles/io.github.linkedfactory.service/src/test/java/io/github/linkedfactory/service/benchmark/KvinIngestionBenchmark.java
new file mode 100644
index 00000000..7135b47b
--- /dev/null
+++ b/bundles/io.github.linkedfactory.service/src/test/java/io/github/linkedfactory/service/benchmark/KvinIngestionBenchmark.java
@@ -0,0 +1,312 @@
+package io.github.linkedfactory.service.benchmark;
+
+import com.google.inject.Guice;
+import io.github.linkedfactory.core.kvin.KvinTuple;
+import io.github.linkedfactory.core.kvin.leveldb.KvinLevelDb;
+import io.github.linkedfactory.core.kvin.util.CsvFormatParser;
+import io.github.linkedfactory.core.kvin.util.JsonFormatParser;
+import io.github.linkedfactory.service.KvinService;
+import io.github.linkedfactory.service.MockHttpServletRequest;
+import net.enilink.commons.iterator.IExtendedIterator;
+import net.enilink.komma.core.KommaModule;
+import net.enilink.komma.core.URI;
+import net.enilink.komma.core.URIs;
+import net.enilink.komma.model.IModelSet;
+import net.enilink.komma.model.IModelSetFactory;
+import net.enilink.komma.model.MODELS;
+import net.enilink.komma.model.ModelPlugin;
+import net.enilink.komma.model.ModelSetModule;
+import net.enilink.platform.lift.util.Globals;
+import net.liftweb.common.Box;
+import net.liftweb.common.Empty$;
+import net.liftweb.common.Full;
+import net.liftweb.http.CurrentReq$;
+import net.liftweb.http.LiftResponse;
+import net.liftweb.http.Req;
+import net.liftweb.http.provider.servlet.HTTPRequestServlet;
+import net.liftweb.util.VendorJ;
+import org.json4s.JValue;
+import org.openjdk.jmh.annotations.Benchmark;
+import org.openjdk.jmh.annotations.BenchmarkMode;
+import org.openjdk.jmh.annotations.Fork;
+import org.openjdk.jmh.annotations.Level;
+import org.openjdk.jmh.annotations.Measurement;
+import org.openjdk.jmh.annotations.Mode;
+import org.openjdk.jmh.annotations.OutputTimeUnit;
+import org.openjdk.jmh.annotations.OperationsPerInvocation;
+import org.openjdk.jmh.annotations.Scope;
+import org.openjdk.jmh.annotations.Setup;
+import org.openjdk.jmh.annotations.State;
+import org.openjdk.jmh.annotations.TearDown;
+import org.openjdk.jmh.annotations.Threads;
+import org.openjdk.jmh.annotations.Warmup;
+import org.junit.Assert;
+import scala.Function0;
+import scala.PartialFunction;
+import scala.collection.immutable.Nil$;
+
+import java.io.ByteArrayInputStream;
+import java.io.File;
+import java.io.IOException;
+import java.io.InputStream;
+import java.nio.file.FileVisitResult;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.nio.file.SimpleFileVisitor;
+import java.nio.file.attribute.BasicFileAttributes;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Set;
+import java.util.concurrent.TimeUnit;
+
+@BenchmarkMode(Mode.Throughput)
+@OutputTimeUnit(TimeUnit.SECONDS)
+@Warmup(iterations = 3)
+@Measurement(iterations = 5)
+@Fork(2)
+@Threads(1)
+public class KvinIngestionBenchmark {
+ private static final int SEQUENTIAL_CSV_FILE_COUNT = 10;
+
+ @State(Scope.Thread)
+ public static class BenchmarkState {
+ private List workloads;
+ private List jsonPayloadVariants;
+ private List csvPayloadVariants;
+ private List> csvPayloadPartitionVariants;
+ private int nextVariantIndex;
+ private KvinIngestionWorkload workload;
+ private IModelSet modelSet;
+ private KvinLevelDb store;
+ private File storeDirectory;
+ private KvinService service;
+ private KvinService parseOnlyService;
+ private byte[] jsonPayload;
+ private byte[] csvPayload;
+ private List csvPayloads;
+ private boolean measuredWrites;
+
+ @Setup(Level.Trial)
+ public void setupTrial() {
+ workloads = KvinIngestionWorkload.variants();
+ jsonPayloadVariants = workloads.stream().map(KvinIngestionWorkload::jsonPayload).toList();
+ csvPayloadVariants = workloads.stream().map(KvinIngestionWorkload::csvPayload).toList();
+ csvPayloadPartitionVariants = workloads.stream()
+ .map(variant -> variant.csvPayloads(SEQUENTIAL_CSV_FILE_COUNT)).toList();
+ nextVariantIndex = 0;
+ KommaModule module = ModelPlugin.createModelSetModule(getClass().getClassLoader());
+ IModelSetFactory factory = Guice.createInjector(new ModelSetModule(module))
+ .getInstance(IModelSetFactory.class);
+ modelSet = factory.createModelSet(MODELS.NAMESPACE_URI.appendFragment("MemoryModelSet"));
+ Globals.contextModelSet().theDefault().set(VendorJ.vendor(new Full(modelSet)));
+ }
+
+ @Setup(Level.Invocation)
+ public void setupInvocation() throws IOException {
+ int variantIndex = nextVariantIndex;
+ nextVariantIndex = (nextVariantIndex + 1) % KvinIngestionWorkload.VARIANT_COUNT;
+ workload = workloads.get(variantIndex);
+ jsonPayload = jsonPayloadVariants.get(variantIndex);
+ csvPayload = csvPayloadVariants.get(variantIndex);
+ csvPayloads = csvPayloadPartitionVariants.get(variantIndex);
+ String tempRoot = System.getProperty("jmh.temp.root", "");
+ Path directory = tempRoot.isEmpty()
+ ? Files.createTempDirectory("kvin-ingestion-jmh-")
+ : Files.createTempDirectory(Path.of(tempRoot), "kvin-ingestion-jmh-");
+ storeDirectory = directory.toFile();
+ store = new KvinLevelDb(storeDirectory);
+ store.put(workload.preseedTuples());
+ service = new BenchmarkService(false);
+ parseOnlyService = new BenchmarkService(true);
+ measuredWrites = false;
+ }
+
+ @TearDown(Level.Invocation)
+ public void teardownInvocation() throws IOException {
+ try {
+ validateStore();
+ } finally {
+ if (store != null) {
+ store.close();
+ store = null;
+ }
+ if (storeDirectory != null) {
+ deleteDirectory(storeDirectory.toPath());
+ storeDirectory = null;
+ }
+ }
+ }
+
+ @TearDown(Level.Trial)
+ public void teardownTrial() {
+ if (modelSet != null) {
+ modelSet.dispose();
+ modelSet = null;
+ }
+ }
+
+ public void putBatch() {
+ store.put(workload.tuples());
+ measuredWrites = true;
+ }
+
+ public void postJson() throws IOException {
+ post(jsonPayload, "application/json", service, true);
+ }
+
+ public void postJsonParseOnly() throws IOException {
+ post(jsonPayload, "application/json", parseOnlyService, false);
+ }
+
+ public void postCsv() throws IOException {
+ post(csvPayload, "text/csv", service, true);
+ }
+
+ public void putCsvDirect() throws IOException {
+ CsvFormatParser parser = new CsvFormatParser(
+ URIs.createURI("http://foo.com/linkedfactory/"), ',',
+ new ByteArrayInputStream(csvPayload));
+ parser.setContext(KvinIngestionWorkload.CONTEXT);
+ try (IExtendedIterator tuples = parser.parse()) {
+ store.put(tuples);
+ }
+ measuredWrites = true;
+ }
+
+ public void postCsvSequentialFiles() throws IOException {
+ for (byte[] payload : csvPayloads) {
+ post(payload, "text/csv", service, true);
+ }
+ }
+
+ private void post(byte[] payload, String contentType, KvinService targetService,
+ boolean writesMeasuredTuples) throws IOException {
+ MockHttpServletRequest request = new MockHttpServletRequest("http://foo.com/linkedfactory/values");
+ request.method_$eq("POST");
+ request.body_$eq(payload);
+ request.contentType_$eq(contentType);
+ Req req = Req.apply(new HTTPRequestServlet(request, null),
+ Nil$.MODULE$.$colon$colon(PartialFunction.empty()), System.nanoTime());
+ Box result = targetService.apply(req).apply();
+ LiftResponse response = result.openOr(null);
+ if (response == null || response.toResponse().code() != 200) {
+ int status = response == null ? -1 : response.toResponse().code();
+ throw new IOException("KVIN ingestion request failed with HTTP status " + status);
+ }
+ if (writesMeasuredTuples) {
+ measuredWrites = true;
+ }
+ }
+
+ private class BenchmarkService extends KvinService {
+ private final boolean parseOnly;
+
+ private BenchmarkService(boolean parseOnly) {
+ super(Nil$.MODULE$.$colon$colon("linkedfactory"), store);
+ this.parseOnly = parseOnly;
+ }
+
+ @Override
+ public URI contextModelUri() {
+ return KvinIngestionWorkload.CONTEXT;
+ }
+
+ @Override
+ public Box saveJsonValues(InputStream in, scala.collection.immutable.List path, long currentTime) {
+ if (!parseOnly) {
+ return super.saveJsonValues(in, path, currentTime);
+ }
+ try {
+ new JsonFormatParser(in).parse(currentTime).toList(); // parse and discard the tuples
+ } catch (IOException e) {
+ throw new RuntimeException(e);
+ }
+ return Empty$.MODULE$;
+ }
+
+ @Override
+ public Function0> apply(Req in) {
+ IModelSet currentModelSet = Globals.contextModelSet().vend().openOr(null);
+ return CurrentReq$.MODULE$.doWith(in, () -> {
+ try {
+ currentModelSet.getUnitOfWork().begin();
+ if (isDefinedAt(in)) {
+ return super.apply(in);
+ }
+ return (Function0) (() -> Box.legacyNullTest((LiftResponse) null));
+ } finally {
+ currentModelSet.getUnitOfWork().end();
+ }
+ });
+ }
+ }
+
+ private void validateStore() {
+ Set expected = new HashSet<>(workload.preseedTuples());
+ if (measuredWrites) {
+ expected.addAll(workload.tuples());
+ }
+ Set actual = new HashSet<>();
+ for (URI item : workload.items()) {
+ try (IExtendedIterator iterator = store.fetch(item, workload.property(),
+ KvinIngestionWorkload.CONTEXT, 0)) {
+ while (iterator.hasNext()) {
+ actual.add(iterator.next());
+ }
+ }
+ }
+ Assert.assertEquals("Unexpected persisted KVIN tuples", expected, actual);
+ Assert.assertEquals((measuredWrites ? KvinIngestionWorkload.TUPLE_COUNT : 0)
+ + KvinIngestionWorkload.CHANNEL_COUNT, actual.size());
+ }
+
+ private static void deleteDirectory(Path directory) throws IOException {
+ if (!Files.exists(directory)) {
+ return;
+ }
+ Files.walkFileTree(directory, new SimpleFileVisitor<>() {
+ @Override
+ public FileVisitResult visitFile(Path file, BasicFileAttributes attrs) throws IOException {
+ Files.deleteIfExists(file);
+ return FileVisitResult.CONTINUE;
+ }
+
+ @Override
+ public FileVisitResult postVisitDirectory(Path dir, IOException exception) throws IOException {
+ Files.deleteIfExists(dir);
+ return FileVisitResult.CONTINUE;
+ }
+ });
+ }
+ }
+
+ @Benchmark
+ @OperationsPerInvocation(KvinIngestionWorkload.TUPLE_COUNT)
+ public void putBatch(BenchmarkState state) {
+ state.putBatch();
+ }
+
+ @Benchmark
+ @OperationsPerInvocation(KvinIngestionWorkload.TUPLE_COUNT)
+ public void postJson(BenchmarkState state) throws IOException {
+ state.postJson();
+ }
+
+ @Benchmark
+ @OperationsPerInvocation(KvinIngestionWorkload.TUPLE_COUNT)
+ public void postCsv(BenchmarkState state) throws IOException {
+ state.postCsv();
+ }
+
+ @Benchmark
+ @OperationsPerInvocation(KvinIngestionWorkload.TUPLE_COUNT)
+ public void putCsvDirect(BenchmarkState state) throws IOException {
+ state.putCsvDirect();
+ }
+
+ @Benchmark
+ @OperationsPerInvocation(KvinIngestionWorkload.TUPLE_COUNT)
+ public void postCsvSequentialFiles(BenchmarkState state) throws IOException {
+ state.postCsvSequentialFiles();
+ }
+}
diff --git a/bundles/io.github.linkedfactory.service/src/test/java/io/github/linkedfactory/service/benchmark/KvinIngestionCsvDiagnosticBenchmark.java b/bundles/io.github.linkedfactory.service/src/test/java/io/github/linkedfactory/service/benchmark/KvinIngestionCsvDiagnosticBenchmark.java
new file mode 100644
index 00000000..ccdca880
--- /dev/null
+++ b/bundles/io.github.linkedfactory.service/src/test/java/io/github/linkedfactory/service/benchmark/KvinIngestionCsvDiagnosticBenchmark.java
@@ -0,0 +1,287 @@
+package io.github.linkedfactory.service.benchmark;
+
+import com.google.inject.Guice;
+import com.opencsv.CSVParser;
+import com.opencsv.CSVParserBuilder;
+import com.opencsv.CSVReader;
+import com.opencsv.CSVReaderBuilder;
+import com.opencsv.exceptions.CsvValidationException;
+import io.github.linkedfactory.core.kvin.DelegatingKvin;
+import io.github.linkedfactory.core.kvin.Kvin;
+import io.github.linkedfactory.core.kvin.KvinListener;
+import io.github.linkedfactory.core.kvin.KvinTuple;
+import io.github.linkedfactory.core.kvin.util.CsvFormatParser;
+import io.github.linkedfactory.service.KvinService;
+import io.github.linkedfactory.service.MockHttpServletRequest;
+import net.enilink.commons.iterator.IExtendedIterator;
+import net.enilink.komma.core.KommaModule;
+import net.enilink.komma.core.URI;
+import net.enilink.komma.core.URIs;
+import net.enilink.komma.model.IModelSet;
+import net.enilink.komma.model.IModelSetFactory;
+import net.enilink.komma.model.MODELS;
+import net.enilink.komma.model.ModelPlugin;
+import net.enilink.komma.model.ModelSetModule;
+import net.enilink.platform.lift.util.Globals;
+import net.liftweb.common.Box;
+import net.liftweb.http.CurrentReq$;
+import net.liftweb.http.LiftResponse;
+import net.liftweb.http.Req;
+import net.liftweb.http.provider.servlet.HTTPRequestServlet;
+import net.liftweb.util.VendorJ;
+import org.openjdk.jmh.annotations.Benchmark;
+import org.openjdk.jmh.annotations.BenchmarkMode;
+import org.openjdk.jmh.annotations.Fork;
+import org.openjdk.jmh.annotations.Level;
+import org.openjdk.jmh.annotations.Measurement;
+import org.openjdk.jmh.annotations.Mode;
+import org.openjdk.jmh.annotations.OutputTimeUnit;
+import org.openjdk.jmh.annotations.Scope;
+import org.openjdk.jmh.annotations.Setup;
+import org.openjdk.jmh.annotations.State;
+import org.openjdk.jmh.annotations.TearDown;
+import org.openjdk.jmh.annotations.Threads;
+import org.openjdk.jmh.annotations.Warmup;
+import org.openjdk.jmh.infra.Blackhole;
+import scala.Function0;
+import scala.PartialFunction;
+import scala.collection.immutable.Nil$;
+
+import java.io.ByteArrayInputStream;
+import java.io.IOException;
+import java.io.InputStreamReader;
+import java.nio.charset.StandardCharsets;
+import java.util.concurrent.TimeUnit;
+
+@BenchmarkMode(Mode.SingleShotTime)
+@OutputTimeUnit(TimeUnit.MILLISECONDS)
+@Warmup(iterations = 3)
+@Measurement(iterations = 5)
+@Fork(2)
+@Threads(1)
+public class KvinIngestionCsvDiagnosticBenchmark {
+ private static final URI BASE = URIs.createURI("http://foo.com/linkedfactory/");
+
+ @State(Scope.Thread)
+ public static class BenchmarkState {
+ private static final int EXPECTED_FIELD_COUNT = KvinIngestionWorkload.ROW_COUNT + 1;
+ private static final int EXPECTED_FIELDS_PER_ROW = KvinIngestionWorkload.CHANNEL_COUNT + 2;
+
+ private KvinIngestionWorkload workload;
+ private byte[] csvPayload;
+ private IModelSet modelSet;
+ private ConsumingKvin sink;
+ private KvinService service;
+ private String measuredStage;
+ private int rowCount;
+ private int fieldCount;
+ private int tupleCount;
+
+ @Setup(Level.Trial)
+ public void setupTrial() {
+ workload = new KvinIngestionWorkload();
+ csvPayload = workload.csvPayload();
+
+ KommaModule module = ModelPlugin.createModelSetModule(ModelPlugin.class.getClassLoader());
+ IModelSetFactory factory =
+ Guice.createInjector(new ModelSetModule(module))
+ .getInstance(IModelSetFactory.class);
+ modelSet = factory.createModelSet(MODELS.NAMESPACE_URI.appendFragment("MemoryModelSet"));
+
+ Globals.contextModelSet().theDefault().set(VendorJ.vendor(new net.liftweb.common.Full(modelSet)));
+ sink = new ConsumingKvin();
+ service = new BenchmarkService(sink);
+ }
+
+ @TearDown(Level.Invocation)
+ public void validateInvocation() {
+ switch (measuredStage) {
+ case "consumePrebuilt", "parseCsvAndConsumeTuples", "postCsvParseOnly" ->
+ org.junit.Assert.assertEquals(KvinIngestionWorkload.TUPLE_COUNT, tupleCount);
+ case "decodeCsvAndConsumeFields" -> {
+ org.junit.Assert.assertEquals(EXPECTED_FIELD_COUNT, rowCount);
+ org.junit.Assert.assertEquals(EXPECTED_FIELD_COUNT * EXPECTED_FIELDS_PER_ROW, fieldCount);
+ }
+ default -> throw new AssertionError("Unknown CSV diagnostic stage " + measuredStage);
+ }
+ }
+
+ @TearDown(Level.Trial)
+ public void teardownTrial() {
+ if (modelSet != null) {
+ modelSet.dispose();
+ modelSet = null;
+ }
+ }
+
+ public void consumePrebuilt(Blackhole blackhole) {
+ measuredStage = "consumePrebuilt";
+ tupleCount = 0;
+ for (KvinTuple tuple : workload.tuples()) {
+ blackhole.consume(tuple);
+ tupleCount++;
+ }
+ }
+
+ public void decodeCsvAndConsumeFields(Blackhole blackhole) throws IOException {
+ measuredStage = "decodeCsvAndConsumeFields";
+ rowCount = 0;
+ fieldCount = 0;
+ CSVParser parser = new CSVParserBuilder()
+ .withSeparator(',')
+ .withIgnoreQuotations(true)
+ .build();
+ try (CSVReader reader = new CSVReaderBuilder(new InputStreamReader(
+ new ByteArrayInputStream(csvPayload), StandardCharsets.UTF_8))
+ .withSkipLines(0)
+ .withCSVParser(parser)
+ .build()) {
+ String[] row;
+ while ((row = readNext(reader)) != null) {
+ rowCount++;
+ for (String field : row) {
+ blackhole.consume(field);
+ fieldCount++;
+ }
+ }
+ }
+ }
+
+ public void parseCsvAndConsumeTuples(Blackhole blackhole) throws IOException {
+ measuredStage = "parseCsvAndConsumeTuples";
+ tupleCount = 0;
+ CsvFormatParser parser = new CsvFormatParser(BASE, ',',
+ new ByteArrayInputStream(csvPayload));
+ parser.setContext(KvinIngestionWorkload.CONTEXT);
+ try (IExtendedIterator tuples = parser.parse()) {
+ while (tuples.hasNext()) {
+ blackhole.consume(tuples.next());
+ tupleCount++;
+ }
+ }
+ }
+
+ public void postCsvParseOnly(Blackhole blackhole) throws IOException {
+ measuredStage = "postCsvParseOnly";
+ tupleCount = 0;
+ sink.startInvocation(blackhole);
+ MockHttpServletRequest request = new MockHttpServletRequest("http://foo.com/linkedfactory/values");
+ request.method_$eq("POST");
+ request.body_$eq(csvPayload);
+ request.contentType_$eq("text/csv");
+ Req req = Req.apply(new HTTPRequestServlet(request, null),
+ Nil$.MODULE$.$colon$colon(PartialFunction.empty()), System.nanoTime());
+ Box result = service.apply(req).apply();
+ LiftResponse response = result.openOr(null);
+ if (response == null || response.toResponse().code() != 200) {
+ int status = response == null ? -1 : response.toResponse().code();
+ throw new IOException("CSV parse-only request failed with HTTP status " + status);
+ }
+ tupleCount = sink.tupleCount();
+ }
+
+ private static String[] readNext(CSVReader reader) throws IOException {
+ try {
+ return reader.readNext();
+ } catch (CsvValidationException e) {
+ throw new IOException(e);
+ }
+ }
+
+ private class BenchmarkService extends KvinService {
+ private BenchmarkService(Kvin sink) {
+ super(Nil$.MODULE$.$colon$colon("linkedfactory"), sink);
+ }
+
+ @Override
+ public URI contextModelUri() {
+ return KvinIngestionWorkload.CONTEXT;
+ }
+
+ @Override
+ public Function0> apply(Req in) {
+ IModelSet currentModelSet = Globals.contextModelSet().vend().openOr(null);
+ return CurrentReq$.MODULE$.doWith(in, () -> {
+ try {
+ currentModelSet.getUnitOfWork().begin();
+ if (isDefinedAt(in)) {
+ return super.apply(in);
+ }
+ return () -> Box.legacyNullTest((LiftResponse) null);
+ } finally {
+ currentModelSet.getUnitOfWork().end();
+ }
+ });
+ }
+ }
+ }
+
+ static class ConsumingKvin extends DelegatingKvin {
+ private Blackhole blackhole;
+ private int iterablePutCount;
+ private int scalarPutCount;
+ private int tupleCount;
+
+ ConsumingKvin() {
+ super(() -> null);
+ }
+
+ void startInvocation(Blackhole blackhole) {
+ this.blackhole = blackhole;
+ iterablePutCount = 0;
+ scalarPutCount = 0;
+ tupleCount = 0;
+ }
+
+ @Override
+ public void put(KvinTuple... tuples) {
+ scalarPutCount++;
+ throw new IllegalStateException("CSV ingestion used scalar KVIN persistence");
+ }
+
+ @Override
+ public void put(Iterable tuples) {
+ iterablePutCount++;
+ for (KvinTuple tuple : tuples) {
+ consumeTuple(tuple);
+ tupleCount++;
+ }
+ }
+
+ protected void consumeTuple(KvinTuple tuple) {
+ blackhole.consume(tuple);
+ }
+
+ int iterablePutCount() {
+ return iterablePutCount;
+ }
+
+ int scalarPutCount() {
+ return scalarPutCount;
+ }
+
+ int tupleCount() {
+ return tupleCount;
+ }
+ }
+
+ @Benchmark
+ public void consumePrebuilt(BenchmarkState state, Blackhole blackhole) {
+ state.consumePrebuilt(blackhole);
+ }
+
+ @Benchmark
+ public void decodeCsvAndConsumeFields(BenchmarkState state, Blackhole blackhole) throws IOException {
+ state.decodeCsvAndConsumeFields(blackhole);
+ }
+
+ @Benchmark
+ public void parseCsvAndConsumeTuples(BenchmarkState state, Blackhole blackhole) throws IOException {
+ state.parseCsvAndConsumeTuples(blackhole);
+ }
+
+ @Benchmark
+ public void postCsvParseOnly(BenchmarkState state, Blackhole blackhole) throws IOException {
+ state.postCsvParseOnly(blackhole);
+ }
+}
diff --git a/bundles/io.github.linkedfactory.service/src/test/java/io/github/linkedfactory/service/benchmark/KvinIngestionCsvDiagnosticBenchmarkTest.java b/bundles/io.github.linkedfactory.service/src/test/java/io/github/linkedfactory/service/benchmark/KvinIngestionCsvDiagnosticBenchmarkTest.java
new file mode 100644
index 00000000..3afe3c3b
--- /dev/null
+++ b/bundles/io.github.linkedfactory.service/src/test/java/io/github/linkedfactory/service/benchmark/KvinIngestionCsvDiagnosticBenchmarkTest.java
@@ -0,0 +1,34 @@
+package io.github.linkedfactory.service.benchmark;
+
+import io.github.linkedfactory.core.kvin.KvinTuple;
+import org.junit.Test;
+
+import java.util.HashSet;
+import java.util.Set;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertTrue;
+
+public class KvinIngestionCsvDiagnosticBenchmarkTest {
+ @Test
+ public void consumingSinkUsesTheIterableContract() {
+ Set observed = new HashSet<>();
+ KvinIngestionCsvDiagnosticBenchmark.ConsumingKvin sink =
+ new KvinIngestionCsvDiagnosticBenchmark.ConsumingKvin() {
+ @Override
+ protected void consumeTuple(KvinTuple tuple) {
+ observed.add(tuple);
+ }
+ };
+ KvinIngestionWorkload workload = new KvinIngestionWorkload();
+
+ sink.startInvocation(null);
+ sink.put(workload.tuples());
+
+ assertEquals(1, sink.iterablePutCount());
+ assertEquals(0, sink.scalarPutCount());
+ assertEquals(KvinIngestionWorkload.TUPLE_COUNT, sink.tupleCount());
+ assertEquals(new HashSet<>(workload.tuples()), observed);
+ assertTrue(observed.stream().allMatch(tuple -> tuple.context.equals(KvinIngestionWorkload.CONTEXT)));
+ }
+}
\ No newline at end of file
diff --git a/bundles/io.github.linkedfactory.service/src/test/java/io/github/linkedfactory/service/benchmark/KvinIngestionJsonDiagnosticBenchmark.java b/bundles/io.github.linkedfactory.service/src/test/java/io/github/linkedfactory/service/benchmark/KvinIngestionJsonDiagnosticBenchmark.java
new file mode 100644
index 00000000..bc4f288a
--- /dev/null
+++ b/bundles/io.github.linkedfactory.service/src/test/java/io/github/linkedfactory/service/benchmark/KvinIngestionJsonDiagnosticBenchmark.java
@@ -0,0 +1,27 @@
+package io.github.linkedfactory.service.benchmark;
+
+import java.io.IOException;
+import java.util.concurrent.TimeUnit;
+
+import org.openjdk.jmh.annotations.Benchmark;
+import org.openjdk.jmh.annotations.BenchmarkMode;
+import org.openjdk.jmh.annotations.Fork;
+import org.openjdk.jmh.annotations.Measurement;
+import org.openjdk.jmh.annotations.Mode;
+import org.openjdk.jmh.annotations.OutputTimeUnit;
+import org.openjdk.jmh.annotations.Threads;
+import org.openjdk.jmh.annotations.Warmup;
+
+@BenchmarkMode(Mode.SingleShotTime)
+@OutputTimeUnit(TimeUnit.MILLISECONDS)
+@Warmup(iterations = 3)
+@Measurement(iterations = 5)
+@Fork(2)
+@Threads(1)
+public class KvinIngestionJsonDiagnosticBenchmark {
+ @Benchmark
+ public void postJsonParseOnly(KvinIngestionBenchmark.BenchmarkState state) throws IOException {
+ state.postJsonParseOnly();
+ }
+
+}
diff --git a/bundles/io.github.linkedfactory.service/src/test/java/io/github/linkedfactory/service/benchmark/KvinIngestionWorkload.java b/bundles/io.github.linkedfactory.service/src/test/java/io/github/linkedfactory/service/benchmark/KvinIngestionWorkload.java
new file mode 100644
index 00000000..e9187934
--- /dev/null
+++ b/bundles/io.github.linkedfactory.service/src/test/java/io/github/linkedfactory/service/benchmark/KvinIngestionWorkload.java
@@ -0,0 +1,187 @@
+package io.github.linkedfactory.service.benchmark;
+
+import io.github.linkedfactory.core.kvin.KvinTuple;
+import io.github.linkedfactory.core.kvin.util.JsonFormatWriter;
+import net.enilink.commons.iterator.WrappedIterator;
+import net.enilink.komma.core.URI;
+import net.enilink.komma.core.URIs;
+
+import java.io.IOException;
+import java.io.UncheckedIOException;
+import java.nio.charset.StandardCharsets;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.Comparator;
+import java.util.List;
+import java.util.Locale;
+import java.util.Random;
+
+public final class KvinIngestionWorkload {
+ public static final int ROW_COUNT = 5_000;
+ public static final int CHANNEL_COUNT = 6;
+ public static final int TUPLE_COUNT = ROW_COUNT * CHANNEL_COUNT;
+ public static final int TIMESTAMP_COUNT = 1_000;
+ public static final int SEQUENCES_PER_TIMESTAMP = 5;
+ public static final int VARIANT_COUNT = 10;
+ public static final int CHANNEL_POOL_SIZE = 100;
+ public static final int PROPERTY_POOL_SIZE = 10;
+ public static final long BASE_START_TIME = 1_710_000_000_000L;
+ public static final long TIMESTAMP_STEP = 1_000L;
+ public static final long TIMESTAMP_WINDOW_SIZE = TIMESTAMP_COUNT * TIMESTAMP_STEP;
+ public static final long SHUFFLE_SEED = 0x4B56494E_20260721L;
+
+ public static final URI CONTEXT = URIs.createURI("http://iwu.lf.de/ecc4p/models/emag");
+
+ private static final List CHANNEL_POOL = createUriPool("http://iwu.lf.de/ecc4p/emag/channel-", CHANNEL_POOL_SIZE, 3);
+ private static final List PROPERTY_POOL = createUriPool("http://iwu.lf.de/ecc4p/property-", PROPERTY_POOL_SIZE, 2);
+ private static final List SHUFFLED_CHANNELS = shuffledChannels();
+
+ private final int variantIndex;
+ private final List items;
+ private final URI property;
+ private final long startTime;
+ private final List tuples;
+ private final List preseedTuples;
+ private final byte[] jsonPayload;
+ private final byte[] csvPayload;
+
+ public KvinIngestionWorkload() {
+ this(0);
+ }
+
+ KvinIngestionWorkload(int variantIndex) {
+ if (variantIndex < 0 || variantIndex >= VARIANT_COUNT) {
+ throw new IllegalArgumentException("variantIndex must be in [0, " + VARIANT_COUNT + "): " + variantIndex);
+ }
+ this.variantIndex = variantIndex;
+ int firstChannel = variantIndex * CHANNEL_COUNT;
+ this.items = List.copyOf(SHUFFLED_CHANNELS.subList(firstChannel, firstChannel + CHANNEL_COUNT));
+ this.property = PROPERTY_POOL.get(variantIndex);
+ this.startTime = BASE_START_TIME + variantIndex * TIMESTAMP_WINDOW_SIZE;
+ this.tuples = createTuples();
+ this.preseedTuples = createPreseedTuples();
+ this.jsonPayload = createJsonPayload(tuples);
+ this.csvPayload = createCsvPayload(0, ROW_COUNT);
+ }
+
+ static List variants() {
+ List variants = new ArrayList<>(VARIANT_COUNT);
+ for (int variant = 0; variant < VARIANT_COUNT; variant++) {
+ variants.add(new KvinIngestionWorkload(variant));
+ }
+ return List.copyOf(variants);
+ }
+
+ int variantIndex() {
+ return variantIndex;
+ }
+
+ List items() {
+ return items;
+ }
+
+ URI property() {
+ return property;
+ }
+
+ public List tuples() {
+ return tuples;
+ }
+
+ public List preseedTuples() {
+ return preseedTuples;
+ }
+
+ public byte[] jsonPayload() {
+ return jsonPayload.clone();
+ }
+
+ public byte[] csvPayload() {
+ return csvPayload.clone();
+ }
+
+ public List csvPayloads(int fileCount) {
+ if (fileCount <= 0 || ROW_COUNT % fileCount != 0) {
+ throw new IllegalArgumentException("fileCount must divide " + ROW_COUNT + ": " + fileCount);
+ }
+ int rowsPerFile = ROW_COUNT / fileCount;
+ List payloads = new ArrayList<>(fileCount);
+ for (int file = 0; file < fileCount; file++) {
+ int startRow = file * rowsPerFile;
+ payloads.add(createCsvPayload(startRow, startRow + rowsPerFile));
+ }
+ return List.copyOf(payloads);
+ }
+
+ private List createTuples() {
+ List result = new ArrayList<>(TUPLE_COUNT);
+ for (int row = 0; row < ROW_COUNT; row++) {
+ for (int channel = 0; channel < CHANNEL_COUNT; channel++) {
+ long time = startTime + (row / SEQUENCES_PER_TIMESTAMP) * TIMESTAMP_STEP;
+ int seqNr = row % SEQUENCES_PER_TIMESTAMP + 1;
+ result.add(new KvinTuple(items.get(channel), property, CONTEXT, time, seqNr, value(channel, row)));
+ }
+ }
+ return List.copyOf(result);
+ }
+
+ private List createPreseedTuples() {
+ List result = new ArrayList<>(CHANNEL_COUNT);
+ for (int channel = 0; channel < CHANNEL_COUNT; channel++) {
+ result.add(new KvinTuple(items.get(channel), property, CONTEXT, startTime - TIMESTAMP_STEP, 0,
+ value(channel, -1)));
+ }
+ return List.copyOf(result);
+ }
+
+ private static byte[] createJsonPayload(List tuples) {
+ try {
+ String json = JsonFormatWriter.toJsonString(WrappedIterator.create(tuples.stream()
+ .sorted(Comparator.comparing((KvinTuple t) -> t.item.toString())
+ .thenComparing(t -> t.property.toString())
+ .thenComparingLong(t -> t.time)
+ .thenComparingInt(t -> t.seqNr))
+ .toList().iterator()));
+ return json.getBytes(StandardCharsets.UTF_8);
+ } catch (IOException e) {
+ throw new UncheckedIOException(e);
+ }
+ }
+
+ private byte[] createCsvPayload(int startRow, int endRow) {
+ StringBuilder csv = new StringBuilder((endRow - startRow) * CHANNEL_COUNT * 12);
+ csv.append("time,seqNr");
+ for (URI item : items) {
+ csv.append(',').append(item).append('@').append(property);
+ }
+ csv.append('\n');
+
+ for (int row = startRow; row < endRow; row++) {
+ csv.append(startTime + (row / SEQUENCES_PER_TIMESTAMP) * TIMESTAMP_STEP)
+ .append(',').append(row % SEQUENCES_PER_TIMESTAMP + 1);
+ for (int channel = 0; channel < CHANNEL_COUNT; channel++) {
+ csv.append(',').append(value(channel, row));
+ }
+ csv.append('\n');
+ }
+ return csv.toString().getBytes(StandardCharsets.UTF_8);
+ }
+
+ private static List createUriPool(String prefix, int size, int digits) {
+ List uris = new ArrayList<>(size);
+ for (int index = 0; index < size; index++) {
+ uris.add(URIs.createURI(prefix + String.format(Locale.ROOT, "%0" + digits + "d", index)));
+ }
+ return List.copyOf(uris);
+ }
+
+ private static List shuffledChannels() {
+ List channels = new ArrayList<>(CHANNEL_POOL);
+ Collections.shuffle(channels, new Random(SHUFFLE_SEED));
+ return List.copyOf(channels);
+ }
+
+ private static double value(int channel, int row) {
+ return channel * 100_000.0 + row + 0.25;
+ }
+}
diff --git a/bundles/io.github.linkedfactory.service/src/test/java/io/github/linkedfactory/service/benchmark/KvinIngestionWorkloadTest.java b/bundles/io.github.linkedfactory.service/src/test/java/io/github/linkedfactory/service/benchmark/KvinIngestionWorkloadTest.java
new file mode 100644
index 00000000..9f5520dc
--- /dev/null
+++ b/bundles/io.github.linkedfactory.service/src/test/java/io/github/linkedfactory/service/benchmark/KvinIngestionWorkloadTest.java
@@ -0,0 +1,156 @@
+package io.github.linkedfactory.service.benchmark;
+
+import io.github.linkedfactory.core.kvin.KvinTuple;
+import io.github.linkedfactory.core.kvin.util.CsvFormatParser;
+import io.github.linkedfactory.core.kvin.util.JsonFormatParser;
+import net.enilink.commons.iterator.IExtendedIterator;
+import net.enilink.komma.core.URI;
+import net.enilink.komma.core.URIs;
+import org.junit.Test;
+
+import java.io.ByteArrayInputStream;
+import java.nio.charset.StandardCharsets;
+import java.util.ArrayList;
+import java.util.Comparator;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+
+import static org.junit.Assert.assertArrayEquals;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertNotEquals;
+import static org.junit.Assert.assertTrue;
+
+public class KvinIngestionWorkloadTest {
+ @Test
+ public void suitesAreDeterministicAndSelectDisjointChannelsAndProperties() {
+ List first = KvinIngestionWorkload.variants();
+ List second = KvinIngestionWorkload.variants();
+
+ assertEquals(KvinIngestionWorkload.VARIANT_COUNT, first.size());
+ assertEquals(first.stream().map(KvinIngestionWorkload::tuples).toList(),
+ second.stream().map(KvinIngestionWorkload::tuples).toList());
+
+ Set properties = new HashSet<>();
+ Set selectedChannels = new HashSet<>();
+ for (int index = 0; index < first.size(); index++) {
+ KvinIngestionWorkload workload = first.get(index);
+ KvinIngestionWorkload copy = second.get(index);
+ assertEquals(index, workload.variantIndex());
+ assertEquals(workload.items(), copy.items());
+ assertEquals(workload.property(), copy.property());
+ assertEquals(workload.preseedTuples(), copy.preseedTuples());
+ assertArrayEquals(workload.csvPayload(), copy.csvPayload());
+ assertArrayEquals(workload.jsonPayload(), copy.jsonPayload());
+ assertEquals(KvinIngestionWorkload.CHANNEL_COUNT, new HashSet<>(workload.items()).size());
+ assertTrue("Channels overlap at variant " + index, selectedChannels.addAll(workload.items()));
+ assertTrue("Property repeated at variant " + index, properties.add(workload.property()));
+ }
+ assertEquals(KvinIngestionWorkload.VARIANT_COUNT, properties.size());
+ assertEquals(KvinIngestionWorkload.VARIANT_COUNT * KvinIngestionWorkload.CHANNEL_COUNT,
+ selectedChannels.size());
+ assertEquals(first.get(0).tuples(), new KvinIngestionWorkload().tuples());
+ }
+
+ @Test
+ public void everyVariantHasTheExpectedShapeAndDisjointKeysAndWindows() {
+ Set allKeys = new HashSet<>();
+ Set allTimestamps = new HashSet<>();
+
+ for (KvinIngestionWorkload workload : KvinIngestionWorkload.variants()) {
+ long expectedStartTime = KvinIngestionWorkload.BASE_START_TIME
+ + workload.variantIndex() * KvinIngestionWorkload.TIMESTAMP_WINDOW_SIZE;
+ assertEquals(KvinIngestionWorkload.TUPLE_COUNT, workload.tuples().size());
+ assertEquals(KvinIngestionWorkload.CHANNEL_COUNT, workload.preseedTuples().size());
+
+ Map> sequencesByTimestamp = new HashMap<>();
+ Set variantTimestamps = new HashSet<>();
+ for (KvinTuple tuple : workload.tuples()) {
+ assertEquals(workload.property(), tuple.property);
+ assertEquals(KvinIngestionWorkload.CONTEXT, tuple.context);
+ assertTrue(tuple.seqNr >= 1 && tuple.seqNr <= KvinIngestionWorkload.SEQUENCES_PER_TIMESTAMP);
+ assertTrue("Duplicate tuple key", allKeys.add(key(tuple)));
+ variantTimestamps.add(tuple.time);
+ sequencesByTimestamp.computeIfAbsent(tuple.time, ignored -> new HashSet<>()).add(tuple.seqNr);
+ }
+
+ assertEquals(KvinIngestionWorkload.TIMESTAMP_COUNT, variantTimestamps.size());
+ assertTrue("Timestamp windows overlap", allTimestamps.stream().noneMatch(variantTimestamps::contains));
+ allTimestamps.addAll(variantTimestamps);
+ assertEquals(KvinIngestionWorkload.TIMESTAMP_COUNT, sequencesByTimestamp.size());
+ assertTrue(sequencesByTimestamp.values().stream()
+ .allMatch(sequences -> sequences.size() == KvinIngestionWorkload.SEQUENCES_PER_TIMESTAMP));
+ assertEquals(expectedStartTime, variantTimestamps.stream().mapToLong(Long::longValue).min().orElseThrow());
+ assertEquals(expectedStartTime + (KvinIngestionWorkload.TIMESTAMP_COUNT - 1)
+ * KvinIngestionWorkload.TIMESTAMP_STEP,
+ variantTimestamps.stream().mapToLong(Long::longValue).max().orElseThrow());
+ }
+ }
+
+ @Test
+ public void allPayloadFormsNormalizeToEachVariantsCanonicalTupleSet() throws Exception {
+ for (KvinIngestionWorkload workload : KvinIngestionWorkload.variants()) {
+ Set expected = new HashSet<>(workload.tuples());
+
+ List csv = parseCsv(workload.csvPayload());
+ assertEquals("CSV tuple order for variant " + workload.variantIndex(), workload.tuples(), csv);
+ assertEquals(expected, new HashSet<>(csv));
+
+ List json = parseJson(workload.jsonPayload());
+ assertEquals(KvinIngestionWorkload.TUPLE_COUNT, json.size());
+ assertEquals(expected, new HashSet<>(json));
+ assertJsonOrder(json);
+
+ List partitionedCsv = new ArrayList<>();
+ List payloads = workload.csvPayloads(10);
+ assertEquals(10, payloads.size());
+ for (byte[] payload : payloads) {
+ partitionedCsv.addAll(parseCsv(payload));
+ }
+ assertEquals(KvinIngestionWorkload.TUPLE_COUNT, partitionedCsv.size());
+ assertEquals(workload.tuples(), partitionedCsv);
+ assertEquals(expected, new HashSet<>(partitionedCsv));
+
+ assertEquals(KvinIngestionWorkload.ROW_COUNT + 1,
+ new String(workload.csvPayload(), StandardCharsets.UTF_8).split("\\n").length);
+ assertNotEquals(0, workload.jsonPayload().length);
+ }
+ }
+
+ private static List parseCsv(byte[] payload) throws Exception {
+ List tuples = new ArrayList<>();
+ CsvFormatParser parser = new CsvFormatParser(URIs.createURI("http://foo.com/linkedfactory/"), ',',
+ new ByteArrayInputStream(payload));
+ parser.setContext(KvinIngestionWorkload.CONTEXT);
+ try (IExtendedIterator iterator = parser.parse()) {
+ while (iterator.hasNext()) {
+ tuples.add(iterator.next());
+ }
+ }
+ return tuples;
+ }
+
+ private static List parseJson(byte[] payload) throws Exception {
+ return new JsonFormatParser(new ByteArrayInputStream(payload))
+ .setContext(KvinIngestionWorkload.CONTEXT)
+ .parse().toList();
+ }
+
+ private static void assertJsonOrder(List tuples) {
+ Comparator order = Comparator.comparing((KvinTuple tuple) -> tuple.item.toString())
+ .thenComparing(tuple -> tuple.property.toString())
+ .thenComparingLong(tuple -> tuple.time)
+ .thenComparingInt(tuple -> tuple.seqNr);
+ for (int index = 1; index < tuples.size(); index++) {
+ assertFalse("JSON is out of order at tuple " + index,
+ order.compare(tuples.get(index - 1), tuples.get(index)) > 0);
+ }
+ }
+
+ private static String key(KvinTuple tuple) {
+ return tuple.context + "|" + tuple.item + "|" + tuple.property + "|" + tuple.time + "|" + tuple.seqNr;
+ }
+}
diff --git a/bundles/io.github.linkedfactory.service/src/test/scala/io/github/linkedfactory/service/benchmark/KvinServiceBenchmark.scala b/bundles/io.github.linkedfactory.service/src/test/scala/io/github/linkedfactory/service/benchmark/KvinServiceBenchmark.scala
deleted file mode 100644
index faed154e..00000000
--- a/bundles/io.github.linkedfactory.service/src/test/scala/io/github/linkedfactory/service/benchmark/KvinServiceBenchmark.scala
+++ /dev/null
@@ -1,168 +0,0 @@
-package io.github.linkedfactory.service.benchmark
-
-import com.google.inject.Guice
-import io.github.linkedfactory.core.kvin.leveldb.KvinLevelDb
-import io.github.linkedfactory.core.kvin.util.JsonFormatWriter
-import io.github.linkedfactory.core.kvin.{Kvin, KvinTuple}
-import io.github.linkedfactory.service.{KvinService, MockHttpServletRequest}
-import net.enilink.commons.iterator.WrappedIterator
-import net.enilink.komma.core.{KommaModule, URI, URIs}
-import net.enilink.komma.model._
-import net.enilink.platform.lift.util.Globals
-import net.liftweb.common.{Box, Full}
-import net.liftweb.http.provider.servlet.HTTPRequestServlet
-import net.liftweb.http.{CurrentReq, LiftResponse, Req}
-import org.junit.{AfterClass, BeforeClass, Ignore, Test}
-import sun.invoke.util.ValueConversions
-
-import java.io.{File, IOException}
-import java.nio.file.attribute.BasicFileAttributes
-import java.nio.file.{FileVisitResult, Files, Path, SimpleFileVisitor}
-import java.util
-import java.util.concurrent.LinkedBlockingQueue
-import jakarta.servlet.http.HttpServletRequest
-import scala.util.Random
-import scala.compiletime.uninitialized
-
-/**
- * Companion object of unit tests for the KVIN service endpoint
- */
-object KvinServiceBenchmark {
- var modelSet: IModelSet = null
- var storeDirectory: File = uninitialized
- var store: Kvin = uninitialized
-
- @BeforeClass
- def setup(): Unit = {
- // create configuration and a model set factory
- val module: KommaModule = ModelPlugin.createModelSetModule(classOf[ModelPlugin].getClassLoader)
- val factory: IModelSetFactory = Guice.createInjector(new ModelSetModule(module)).getInstance(classOf[IModelSetFactory])
-
- // create a model set with an in-memory repository
- modelSet = factory.createModelSet(MODELS.NAMESPACE_URI.appendFragment("MemoryModelSet"))
- Globals.contextModelSet.default.set(Full(modelSet))
-
- createStore()
- }
-
- @AfterClass
- def tearDown(): Unit = {
- modelSet.dispose()
- modelSet = null
-
- store.close
- store = null
- deleteDirectory(storeDirectory.toPath)
- }
-
- def createStore(): Unit = {
- storeDirectory = new File("/tmp/leveldb-test-" + System.currentTimeMillis + "-" + Random.nextInt(1000) + "/")
- storeDirectory.deleteOnExit
- store = new KvinLevelDb(storeDirectory)
- }
-
- def deleteDirectory(dir: Path): Unit = {
- // delete store directory
- Files.walkFileTree(dir, new SimpleFileVisitor[Path]() {
- override def visitFile(file: Path, attrs: BasicFileAttributes): FileVisitResult = {
- Files.delete(file)
- FileVisitResult.CONTINUE
- }
-
- override def postVisitDirectory(dir: Path, ex: IOException): FileVisitResult = {
- Files.delete(dir)
- FileVisitResult.CONTINUE
- }
- })
- }
-}
-
-/**
- * Unit tests for the KVIN service endpoint
- */
-class KvinServiceBenchmark {
- val kvinService = new KvinService("linkedfactory" :: Nil, KvinServiceBenchmark.store) {
- override def apply(in: Req): () => Box[LiftResponse] = {
- try {
- Globals.contextModelSet.vend.map(_.getUnitOfWork.begin)
- // S.request is used in Data.pathToURI therefore the request needs to be initialized here
- CurrentReq.doWith(in) {
- super.apply(in)
- }
- } finally {
- Globals.contextModelSet.vend.map(_.getUnitOfWork.end)
- }
- }
- }
-
- def kvinRest(req: Req): () => Box[LiftResponse] = {
- kvinService(req)
- }
-
- val baseUrl = "http://foo.com/linkedfactory/values"
-
- def toReq(httpRequest: HttpServletRequest): Req = {
- Req(new HTTPRequestServlet(httpRequest, null), Nil, System.nanoTime)
- }
-
- @Test
- @Ignore
- def postValues(): Unit = {
- val valueProperty = URIs.createURI("property:value")
-
- val seed = 200
- val writeValues = 1000000
-
- val benchmarkStart = System.currentTimeMillis
-
- val startTimeValues = 1478252048736L
- val nrs = Array.fill(100)(Random.nextInt(Integer.MAX_VALUE))
- val rand = new Random(seed)
-
- // decouples client-side serialization and server-side parsing and insertion
- val queue = new LinkedBlockingQueue[Option[String]](2)
- val inserter = new Thread() {
- override def run() : Unit = {
- var finished = false
- while (!finished) {
- queue.take() match {
- case None => finished = true
- case Some(json) =>
- // support post request
- val postReq = new MockHttpServletRequest(baseUrl) {
- method = "POST"
- body_=(json, "application/json")
- }
- kvinRest(toReq(postReq))().map(_.toResponse.code)
- }
- }
- }
- }
- inserter.start()
-
- var tuples = new util.ArrayList[KvinTuple]()
- var currentTime = startTimeValues
- (0 to writeValues).foreach { i =>
- val randomNr = nrs(rand.nextInt(nrs.length))
- val uri = URIs.createURI("http://linkedfactory.github.io/" + randomNr + "/e3fabrik/rollex/" + randomNr + "/measured-point-1")
- val ctx = URIs.createURI("ctx:" + randomNr)
-
- val value = if (randomNr % 2 == 0) rand.nextGaussian() else rand.nextLong(100000)
-
- tuples.add(new KvinTuple(uri, valueProperty, ctx, currentTime, value))
- currentTime += rand.nextInt(1000)
-
- if (i % 10000 == 0) {
- println(" at: " + i)
- val json = JsonFormatWriter.toJsonString(WrappedIterator.create(tuples.iterator()))
- queue.put(Some(json))
- tuples = new util.ArrayList[KvinTuple]()
- }
- }
- queue.put(None)
- inserter.join()
-
- val seconds = (System.currentTimeMillis - benchmarkStart) / 1000.0
- println(s"Wrote $writeValues in %1$$,.2f seconds: %2$$,.2f ops per second".format(seconds, writeValues / seconds))
- }
-}
\ No newline at end of file
diff --git a/docs/benchmarks/kvin-ingestion.md b/docs/benchmarks/kvin-ingestion.md
new file mode 100644
index 00000000..d92d558f
--- /dev/null
+++ b/docs/benchmarks/kvin-ingestion.md
@@ -0,0 +1,184 @@
+# KVIN Ingestion Benchmarks
+
+This JMH suite measures **KVIN ingestion throughput** through the direct LevelDB API and the in-process JSON and CSV service routes.
+
+Each primary benchmark processes **30,000 `KvinTuple`s per invocation**. JMH reports throughput as `ops/s`; `@OperationsPerInvocation(30000)` normalizes the result so that:
+
+> **1 op = 1 KvinTuple**
+
+The displayed `ops/s` score therefore directly represents tuple throughput.
+
+Test-data generation, validation, cleanup, network transport, TLS, and server startup are outside the timed methods.
+
+## What is measured
+
+```mermaid
+flowchart LR
+ PB["putBatch"] --> W["KVIN batch writer"]
+ CD["putCsvDirect"] --> CP["CSV parser"] --> W
+ PC["postCsv"] --> L["Lift route"] --> CP
+ PJ["postJson"] --> L --> JP["JSON parser"] --> W
+ PS["postCsvSequentialFiles
10 requests"] --> L
+ W --> DB[("LevelDB")]
+```
+
+| Benchmark | Measures |
+| ------------------------ | ----------------------------------------------------------------------- |
+| `putBatch` | Prebuilt tuples → KVIN encoding / ID resolution / batch write → LevelDB |
+| `putCsvDirect` | CSV parsing + tuple creation + persistence, without Lift routing |
+| `postCsv` | In-process CSV request → route → parsing → persistence |
+| `postJson` | In-process JSON request → route → parsing → persistence |
+| `postCsvSequentialFiles` | Ten sequential CSV requests to the same service and store |
+
+`putBatch` is the persistence baseline. `putCsvDirect` adds CSV parsing. `postCsv` and `postJson` represent the main in-process service ingestion paths.
+
+### Test data
+
+Before measurement starts, the benchmark creates **10 fixed workload variants**. Each variant contains **5,000 rows across 6 channels, producing 30,000 `KvinTuple`s**.
+
+A channel represents one KVIN **item URI**, i.e. one time-series source. In CSV input, the six channels correspond to six value columns. Each resulting tuple contains:
+
+```text
+(item, property, context, timestamp, sequence number, value)
+```
+
+For each workload variant:
+
+* **6 channels** are selected from a pool of 100 item URIs. The pool is shuffled once with a fixed seed, so the same variant always receives the same channels across benchmark runs.
+* **1 property** is selected from a pool of 10 properties, with a different property assigned to each variant.
+* **1 context** is shared by all tuples.
+* **1,000 timestamps** form a fixed time window for that variant.
+* **5 sequence numbers per timestamp** produce the 5,000 rows.
+* Each row contains values for all 6 channels, resulting in `5,000 × 6 = 30,000` tuples.
+
+During measurement, the benchmark cycles through the 10 workload variants on repeat. Each warmup or measurement iteration runs the benchmark repeatedly for 3 seconds, so the ten variants are normally cycled through many times. The exact number of repetitions depends on how many benchmark invocations complete during that time.
+
+The variant sequence and contents remain fixed across runs; each persistence invocation writes its selected variant to a fresh temporary LevelDB store.
+
+## Run
+
+From the repository root, build once:
+
+```sh
+mvn clean install -DskipTests
+```
+
+Run the five primary benchmarks:
+
+```sh
+mvn -pl bundles/io.github.linkedfactory.service -Pjmh \
+ -Djmh.result.file=target/jmh-primary.json \
+ test-compile exec:exec
+```
+
+The default run uses 3 × 3 s warmups, 5 × 3 s measurements, 2 forks, and 1 thread.
+
+When comparing a change, repeat the full run with a separate result file.
+
+## Results
+
+A full run produces standard JMH throughput output like:
+
+```text
+Benchmark Mode Cnt Score Error Units
+KvinIngestionBenchmark.postCsv thrpt 10 712702.231 ± 234440.177 ops/s
+KvinIngestionBenchmark.postCsvSequentialFiles thrpt 10 692147.445 ± 279633.790 ops/s
+KvinIngestionBenchmark.postJson thrpt 10 504792.294 ± 42078.535 ops/s
+KvinIngestionBenchmark.putBatch thrpt 10 1673947.197 ± 132642.643 ops/s
+KvinIngestionBenchmark.putCsvDirect thrpt 10 880491.470 ± 126678.908 ops/s
+```
+
+For the primary benchmarks, these values are already normalized to `KvinTuple`s per second. For example, `880491 ops/s` means approximately **880,491 tuples/s**.
+
+Results are also written as JSON. Print a compact summary with:
+
+```sh
+jq -r '.[] |
+ [(.benchmark | split(".")[-1]),
+ .primaryMetric.score,
+ .primaryMetric.scoreError,
+ .primaryMetric.scoreUnit] |
+ @tsv' \
+ bundles/io.github.linkedfactory.service/target/jmh-primary.json
+```
+
+Use the score together with its error interval when comparing runs. The values above are example measurements, not performance thresholds.
+
+## Diagnostics
+
+Diagnostics go **one level deeper** when a primary benchmark shows that CSV or JSON ingestion needs investigation. Unlike the primary suite, they time complete diagnostic invocations in `ms/op`.
+
+### CSV diagnostics
+
+Use these when the difference between `putBatch`, `putCsvDirect`, and `postCsv` suggests that CSV parsing or routing is responsible for significant overhead.
+
+```sh
+mvn -pl bundles/io.github.linkedfactory.service -Pjmh \
+ -Djmh.includes=io.github.linkedfactory.service.benchmark.KvinIngestionCsvDiagnosticBenchmark \
+ -Djmh.result.file=target/jmh-csv-diagnostic.json \
+ test-compile exec:exec
+```
+
+With example result:
+
+```text
+Benchmark Mode Cnt Score Error Units
+KvinIngestionCsvDiagnosticBenchmark.consumePrebuilt ss 10 0.384 ± 0.290 ms/op
+KvinIngestionCsvDiagnosticBenchmark.decodeCsvAndConsumeFields ss 10 12.234 ± 3.119 ms/op
+KvinIngestionCsvDiagnosticBenchmark.parseCsvAndConsumeTuples ss 10 29.207 ± 17.566 ms/op
+KvinIngestionCsvDiagnosticBenchmark.postCsvParseOnly ss 10 30.864 ± 22.291 ms/op
+```
+
+They progressively inspect the time taken up to each gate:
+
+```text
+prebuilt tuple iteration ~0.4 ms
+ ↓
+CSV decoding ~12.2 ms
+ ↓
+full CSV → KvinTuple parsing ~29.2 ms
+ ↓
+Lift route + full CSV parsing ~30.9 ms
+```
+
+This separates tuple-consumption overhead, CSV tokenization, tuple construction, and routing without involving LevelDB persistence.
+
+### JSON diagnostics
+
+Use this diagnostic when `postJson` is slow and you want to distinguish **JSON request/parsing overhead** from persistence.
+
+```sh
+mvn -pl bundles/io.github.linkedfactory.service -Pjmh \
+ -Djmh.includes=io.github.linkedfactory.service.benchmark.KvinIngestionJsonDiagnosticBenchmark \
+ -Djmh.result.file=target/jmh-json-diagnostic.json \
+ test-compile exec:exec
+```
+
+Example result:
+
+```text
+Benchmark Mode Cnt Score Error Units
+KvinIngestionJsonDiagnosticBenchmark.postJsonParseOnly ss 10 37.008 ± 3.228 ms/op
+```
+
+`postJsonParseOnly` measures the in-process JSON request and parsing path without persistence, so this run spends about **37 ms** parsing one 30,000-tuple payload.
+
+For comparison, convert the throughput results from the primary benchmarks to equivalent 30,000-tuple batch times:
+
+```text
+postJson 504,792 ops/s → ~59.4 ms per 30,000 tuples
+putBatch 1,673,947 ops/s → ~17.9 ms per 30,000 tuples
+```
+
+This gives the following conceptual breakdown:
+
+```text
+JSON request + parsing only ~37.0 ms
+ ↓
+shared KVIN/LevelDB persistence
+baseline ~17.9 ms
+ ↓
+full JSON ingestion (postJson) ~59.4 ms
+```
+
+The values come from separate benchmark boundaries and are useful for locating overhead, but they should not be treated as exactly additive stage timings.