From aa4cd188d56d3e03ea57629d1fd440a9e24c1349 Mon Sep 17 00:00:00 2001 From: Willy Mehling Date: Fri, 17 Jul 2026 15:42:57 +0200 Subject: [PATCH 1/5] Add KVIN ingestion benchmarks for CSV and JSON processing - Implement KvinIngestionCsvDiagnosticBenchmark for performance testing of CSV ingestion. - Create KvinIngestionJsonDiagnosticBenchmark for benchmarking JSON ingestion. - Introduce KvinIngestionWorkload to generate test data for benchmarks. - Add unit tests for KvinIngestionCsvDiagnosticBenchmark and KvinIngestionWorkload. - Remove obsolete KvinServiceBenchmark Scala file. - Document benchmark setup and execution in markdown format. --- README.md | 1 + .../core/kvin/util/CsvFormatParserTest.java | 23 ++ .../io.github.linkedfactory.service/pom.xml | 45 +++ .../benchmark/KvinIngestionBenchmark.java | 313 ++++++++++++++++++ .../KvinIngestionCsvDiagnosticBenchmark.java | 293 ++++++++++++++++ ...inIngestionCsvDiagnosticBenchmarkTest.java | 34 ++ .../KvinIngestionJsonDiagnosticBenchmark.java | 31 ++ .../benchmark/KvinIngestionWorkload.java | 131 ++++++++ .../benchmark/KvinIngestionWorkloadTest.java | 154 +++++++++ .../benchmark/KvinServiceBenchmark.scala | 168 ---------- docs/benchmarks/kvin-ingestion.md | 294 ++++++++++++++++ 11 files changed, 1319 insertions(+), 168 deletions(-) create mode 100644 bundles/io.github.linkedfactory.service/src/test/java/io/github/linkedfactory/service/benchmark/KvinIngestionBenchmark.java create mode 100644 bundles/io.github.linkedfactory.service/src/test/java/io/github/linkedfactory/service/benchmark/KvinIngestionCsvDiagnosticBenchmark.java create mode 100644 bundles/io.github.linkedfactory.service/src/test/java/io/github/linkedfactory/service/benchmark/KvinIngestionCsvDiagnosticBenchmarkTest.java create mode 100644 bundles/io.github.linkedfactory.service/src/test/java/io/github/linkedfactory/service/benchmark/KvinIngestionJsonDiagnosticBenchmark.java create mode 100644 bundles/io.github.linkedfactory.service/src/test/java/io/github/linkedfactory/service/benchmark/KvinIngestionWorkload.java create mode 100644 bundles/io.github.linkedfactory.service/src/test/java/io/github/linkedfactory/service/benchmark/KvinIngestionWorkloadTest.java delete mode 100644 bundles/io.github.linkedfactory.service/src/test/scala/io/github/linkedfactory/service/benchmark/KvinServiceBenchmark.scala create mode 100644 docs/benchmarks/kvin-ingestion.md 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.core/src/test/java/io/github/linkedfactory/core/kvin/util/CsvFormatParserTest.java b/bundles/io.github.linkedfactory.core/src/test/java/io/github/linkedfactory/core/kvin/util/CsvFormatParserTest.java index b796eec6..2ef1afe5 100644 --- a/bundles/io.github.linkedfactory.core/src/test/java/io/github/linkedfactory/core/kvin/util/CsvFormatParserTest.java +++ b/bundles/io.github.linkedfactory.core/src/test/java/io/github/linkedfactory/core/kvin/util/CsvFormatParserTest.java @@ -22,6 +22,7 @@ import java.io.ByteArrayInputStream; import java.io.IOException; +import java.nio.charset.StandardCharsets; import java.util.List; import static org.junit.Assert.*; @@ -148,4 +149,26 @@ public void shouldParseCsvDoubleValues() throws IOException { assertFalse(tuples.hasNext()); } + @Test + public void shouldInterpretCsvValueTypes() throws IOException { + CsvFormatParser csvParser = new CsvFormatParser(URIs.createURI("urn:base:"), ';', + new ByteArrayInputStream("time;value\n123;42\n124;3.25\n125;TrUe\n126;\"quoted\"\n127;plain\n128;1.234,56\n" + .getBytes(StandardCharsets.UTF_8))); + IExtendedIterator tuples = csvParser.parse(); + assertNotNull(tuples); + + assertValue(tuples.next(), Long.class, 42L); + assertValue(tuples.next(), Double.class, 3.25d); + assertValue(tuples.next(), Boolean.class, true); + assertValue(tuples.next(), String.class, "quoted"); + assertValue(tuples.next(), String.class, "plain"); + assertValue(tuples.next(), Double.class, 1234.56d); + assertFalse(tuples.hasNext()); + } + + private static void assertValue(KvinTuple tuple, Class type, Object expected) { + assertEquals(type, tuple.value.getClass()); + assertEquals(expected, tuple.value); + } + } diff --git a/bundles/io.github.linkedfactory.service/pom.xml b/bundles/io.github.linkedfactory.service/pom.xml index 069651de..57f433d2 100644 --- a/bundles/io.github.linkedfactory.service/pom.xml +++ b/bundles/io.github.linkedfactory.service/pom.xml @@ -222,4 +222,49 @@ + + + + jmh + + io.github.linkedfactory.service.benchmark.KvinIngestionBenchmark + json + ${project.build.directory}/jmh-result.json + 3 + 5 + 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} + -i + ${jmh.measurements} + -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/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..0e9751d7 --- /dev/null +++ b/bundles/io.github.linkedfactory.service/src/test/java/io/github/linkedfactory/service/benchmark/KvinIngestionBenchmark.java @@ -0,0 +1,313 @@ +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.service.KvinService; +import io.github.linkedfactory.service.MockHttpServletRequest; +import io.github.linkedfactory.service.util.JsonFormatParser$; +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.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.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.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.SingleShotTime) +@OutputTimeUnit(TimeUnit.MILLISECONDS) +@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 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() { + workload = new KvinIngestionWorkload(); + jsonPayload = workload.jsonPayload(); + csvPayload = workload.csvPayload(); + csvPayloads = workload.csvPayloads(SEQUENTIAL_CSV_FILE_COUNT); + try { + 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")); + modelSet = factory.createModelSet(MODELS.NAMESPACE_URI.appendFragment("MemoryModelSet")); + } catch (ClassNotFoundException e) { + throw new IllegalStateException("Could not initialize the Komma model set", e); + } + Globals.contextModelSet().theDefault().set(VendorJ.vendor(new Full(modelSet))); + } + + @Setup(Level.Invocation) + public void setupInvocation() throws IOException { + 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 = createService(); + parseOnlyService = createParseOnlyService(); + 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 putScalar() { + for (KvinTuple tuple : workload.tuples()) { + store.put(tuple); + } + 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); + IExtendedIterator tuples = parser.parse(); + try { + store.put(tuples); + } finally { + tuples.close(); + } + 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 KvinService createService() { + return new BenchmarkService(false); + } + + private KvinService createParseOnlyService() { + return new BenchmarkService(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 saveValues(JValue json, scala.collection.immutable.List path, long currentTime) { + if (!parseOnly) { + return super.saveValues(json, path, currentTime); + } + return JsonFormatParser$.MODULE$.parseItem(URIs.createURI("http://foo.com/linkedfactory/"), + contextModelUri(), json, currentTime); + } + + @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(); + } + }); + } + } + + private void validateStore() { + Set expected = new HashSet<>(workload.preseedTuples()); + if (measuredWrites) { + expected.addAll(workload.tuples()); + } + Set actual = new HashSet<>(); + for (URI item : KvinIngestionWorkload.ITEMS) { + IExtendedIterator iterator = store.fetch(item, KvinIngestionWorkload.PROPERTY, + KvinIngestionWorkload.CONTEXT, 0); + try { + while (iterator.hasNext()) { + actual.add(iterator.next()); + } + } finally { + iterator.close(); + } + } + 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 + public void putBatch(BenchmarkState state) { + state.putBatch(); + } + + @Benchmark + public void postJson(BenchmarkState state) throws IOException { + state.postJson(); + } + + @Benchmark + public void postCsv(BenchmarkState state) throws IOException { + state.postCsv(); + } + + @Benchmark + public void putCsvDirect(BenchmarkState state) throws IOException { + state.putCsvDirect(); + } + + @Benchmark + public void postCsvSequentialFiles(BenchmarkState state) throws IOException { + state.postCsvSequentialFiles(); + } +} \ No newline at end of file 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..ce4e7ca9 --- /dev/null +++ b/bundles/io.github.linkedfactory.service/src/test/java/io/github/linkedfactory/service/benchmark/KvinIngestionCsvDiagnosticBenchmark.java @@ -0,0 +1,293 @@ +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.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 { + @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(); + try { + 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")); + modelSet = factory.createModelSet(MODELS.NAMESPACE_URI.appendFragment("MemoryModelSet")); + } catch (ClassNotFoundException e) { + throw new IllegalStateException("Could not initialize the Komma model set", e); + } + 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(URIsForBenchmark.BASE, ',', + new ByteArrayInputStream(csvPayload)); + parser.setContext(KvinIngestionWorkload.CONTEXT); + IExtendedIterator tuples = parser.parse(); + try { + while (tuples.hasNext()) { + blackhole.consume(tuples.next()); + tupleCount++; + } + } finally { + tuples.close(); + } + } + + 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; + } + } + + private static final class URIsForBenchmark { + private static final URI BASE = net.enilink.komma.core.URIs.createURI("http://foo.com/linkedfactory/"); + } + + @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); + } +} \ No newline at end of file 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..e5163312 --- /dev/null +++ b/bundles/io.github.linkedfactory.service/src/test/java/io/github/linkedfactory/service/benchmark/KvinIngestionJsonDiagnosticBenchmark.java @@ -0,0 +1,31 @@ +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(); + } + + @Benchmark + public void putScalar(KvinIngestionBenchmark.BenchmarkState state) { + state.putScalar(); + } +} \ No newline at end of file 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..366acd11 --- /dev/null +++ b/bundles/io.github.linkedfactory.service/src/test/java/io/github/linkedfactory/service/benchmark/KvinIngestionWorkload.java @@ -0,0 +1,131 @@ +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.List; + +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 long START_TIME = 1_710_000_000_000L; + + public static final URI PROPERTY = URIs.createURI("http://iwu.lf.de/ecc4p/values"); + public static final URI CONTEXT = URIs.createURI("http://iwu.lf.de/ecc4p/models/emag"); + public static final List ITEMS = List.of( + URIs.createURI("http://iwu.lf.de/ecc4p/emag/channel-1"), + URIs.createURI("http://iwu.lf.de/ecc4p/emag/channel-2"), + URIs.createURI("http://iwu.lf.de/ecc4p/emag/channel-3"), + URIs.createURI("http://iwu.lf.de/ecc4p/emag/channel-4"), + URIs.createURI("http://iwu.lf.de/ecc4p/emag/channel-5"), + URIs.createURI("http://iwu.lf.de/ecc4p/emag/channel-6")); + + private final List tuples; + private final List preseedTuples; + private final byte[] jsonPayload; + private final byte[] csvPayload; + + public KvinIngestionWorkload() { + this.tuples = createTuples(); + this.preseedTuples = createPreseedTuples(); + this.jsonPayload = createJsonPayload(tuples); + this.csvPayload = createCsvPayload(); + } + + 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 static List createTuples() { + List tuples = new ArrayList<>(TUPLE_COUNT); + for (int row = 0; row < ROW_COUNT; row++) { + for (int channel = 0; channel < CHANNEL_COUNT; channel++) { + long time = START_TIME + row / SEQUENCES_PER_TIMESTAMP; + int seqNr = row % SEQUENCES_PER_TIMESTAMP + 1; + double value = value(channel, row); + tuples.add(new KvinTuple(ITEMS.get(channel), PROPERTY, CONTEXT, time, seqNr, value)); + } + } + return List.copyOf(tuples); + } + + private static List createPreseedTuples() { + List tuples = new ArrayList<>(CHANNEL_COUNT); + for (int channel = 0; channel < CHANNEL_COUNT; channel++) { + tuples.add(new KvinTuple(ITEMS.get(channel), PROPERTY, CONTEXT, START_TIME - 1, 0, + value(channel, -1))); + } + return List.copyOf(tuples); + } + + private static byte[] createJsonPayload(List tuples) { + try { + String json = JsonFormatWriter.toJsonString(WrappedIterator.create(tuples.iterator())); + return json.getBytes(StandardCharsets.UTF_8); + } catch (IOException e) { + throw new UncheckedIOException(e); + } + } + + private static byte[] createCsvPayload() { + return createCsvPayload(0, ROW_COUNT); + } + + private static 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(START_TIME + row / SEQUENCES_PER_TIMESTAMP) + .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 double value(int channel, int row) { + return channel * 100_000.0 + row + 0.25; + } +} \ No newline at end of file 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..b7b2c9f9 --- /dev/null +++ b/bundles/io.github.linkedfactory.service/src/test/java/io/github/linkedfactory/service/benchmark/KvinIngestionWorkloadTest.java @@ -0,0 +1,154 @@ +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.service.util.JsonFormatParser; +import net.enilink.commons.iterator.IExtendedIterator; +import net.enilink.komma.core.URIs; +import net.liftweb.common.Box; +import org.json4s.AsJsonInput; +import org.json4s.JValue; +import org.junit.Test; +import scala.collection.immutable.List; + +import java.io.ByteArrayInputStream; +import java.io.StringReader; +import java.lang.reflect.Method; +import java.nio.charset.StandardCharsets; +import java.util.HashMap; +import java.util.HashSet; +import java.util.Map; +import java.util.Set; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertFalse; +import static org.junit.Assert.assertTrue; + +public class KvinIngestionWorkloadTest { + @Test + public void hasTheExpectedDeterministicShape() { + KvinIngestionWorkload workload = new KvinIngestionWorkload(); + assertEquals(KvinIngestionWorkload.TUPLE_COUNT, workload.tuples().size()); + assertEquals(KvinIngestionWorkload.CHANNEL_COUNT, new HashSet<>(workload.tuples().stream() + .map(tuple -> tuple.item).toList()).size()); + assertEquals(KvinIngestionWorkload.TUPLE_COUNT + KvinIngestionWorkload.CHANNEL_COUNT, + workload.tuples().size() + workload.preseedTuples().size()); + + Map> sequencesByTimestamp = new HashMap<>(); + Set keys = new HashSet<>(); + for (KvinTuple tuple : workload.tuples()) { + assertEquals(KvinIngestionWorkload.PROPERTY, tuple.property); + assertEquals(KvinIngestionWorkload.CONTEXT, tuple.context); + assertTrue(tuple.seqNr >= 1 && tuple.seqNr <= KvinIngestionWorkload.SEQUENCES_PER_TIMESTAMP); + assertTrue(keys.add(key(tuple))); + sequencesByTimestamp.computeIfAbsent(tuple.time, ignored -> new HashSet<>()).add(tuple.seqNr); + } + assertEquals(KvinIngestionWorkload.TIMESTAMP_COUNT, sequencesByTimestamp.size()); + assertTrue(sequencesByTimestamp.values().stream() + .allMatch(sequences -> sequences.size() == KvinIngestionWorkload.SEQUENCES_PER_TIMESTAMP)); + assertEquals(workload.tuples(), new KvinIngestionWorkload().tuples()); + } + + @Test + public void payloadsNormalizeToTheCanonicalTupleSet() throws Exception { + KvinIngestionWorkload workload = new KvinIngestionWorkload(); + Set expected = new HashSet<>(workload.tuples()); + + Set csvTuples = new HashSet<>(); + int csvCount = 0; + CsvFormatParser csvParser = new CsvFormatParser(URIs.createURI("http://foo.com/linkedfactory/"), ',', + new ByteArrayInputStream(workload.csvPayload())); + csvParser.setContext(KvinIngestionWorkload.CONTEXT); + IExtendedIterator csvIterator = csvParser.parse(); + try { + while (csvIterator.hasNext()) { + KvinTuple tuple = csvIterator.next(); + assertEquals("CSV tuple order", workload.tuples().get(csvCount), tuple); + csvTuples.add(tuple); + csvCount++; + } + } finally { + csvIterator.close(); + } + assertEquals("CSV tuple count", KvinIngestionWorkload.TUPLE_COUNT, csvCount); + assertEquals(expected, csvTuples); + + JValue json = parseJson(new String(workload.jsonPayload(), StandardCharsets.UTF_8)); + Box> parsed = parseProductionJson(json); + assertTrue(parsed.isDefined()); + Set jsonTuples = new HashSet<>(); + @SuppressWarnings("unchecked") + List parsedTuples = (List) parsed.openOr(null); + assertEquals("JSON tuple count", KvinIngestionWorkload.TUPLE_COUNT, parsedTuples.size()); + scala.collection.Iterator jsonIterator = parsedTuples.iterator(); + while (jsonIterator.hasNext()) { + jsonTuples.add(jsonIterator.next()); + } + Set missing = new HashSet<>(expected); + missing.removeAll(jsonTuples); + Set extra = new HashSet<>(jsonTuples); + extra.removeAll(expected); + assertEquals(expected.size(), jsonTuples.size()); + assertTrue("JSON missing=" + sample(missing) + ", extra=" + sample(extra), missing.isEmpty() && extra.isEmpty()); + assertEquals(KvinIngestionWorkload.ROW_COUNT + 1, + new String(workload.csvPayload(), StandardCharsets.UTF_8).split("\\n").length); + assertFalse(workload.jsonPayload().length == 0); + } + + @Test + public void csvFilePartitionsNormalizeToTheCanonicalTupleSet() throws Exception { + KvinIngestionWorkload workload = new KvinIngestionWorkload(); + Set expected = new HashSet<>(workload.tuples()); + + for (int fileCount : java.util.List.of(1, 2, 5, 10)) { + Set actual = new HashSet<>(); + int tupleCount = 0; + java.util.List payloads = workload.csvPayloads(fileCount); + for (byte[] payload : payloads) { + CsvFormatParser parser = new CsvFormatParser(URIs.createURI("http://foo.com/linkedfactory/"), ',', + new ByteArrayInputStream(payload)); + parser.setContext(KvinIngestionWorkload.CONTEXT); + IExtendedIterator tuples = parser.parse(); + try { + while (tuples.hasNext()) { + actual.add(tuples.next()); + tupleCount++; + } + } finally { + tuples.close(); + } + } + + assertEquals(fileCount, payloads.size()); + assertEquals(KvinIngestionWorkload.TUPLE_COUNT, tupleCount); + assertEquals(expected, actual); + } + } + + private static String key(KvinTuple tuple) { + return tuple.context + "|" + tuple.item + "|" + tuple.property + "|" + tuple.time + "|" + tuple.seqNr; + } + + private static String sample(Set tuples) { + return tuples.stream().limit(3).toList().toString(); + } + + private static JValue parseJson(String json) throws Exception { + Class parserClass = Class.forName("org.json4s.native.JsonParser$"); + Object parser = parserClass.getField("MODULE$").get(null); + Method parse = parserClass.getMethod("parse", java.io.Reader.class, boolean.class, boolean.class, + boolean.class); + return (JValue) parse.invoke(parser, new StringReader(json), true, false, true); + } + + @SuppressWarnings("unchecked") + private static Box> parseProductionJson(JValue json) throws Exception { + Class parserClass = Class.forName("io.github.linkedfactory.service.util.JsonFormatParser$"); + Object parser = parserClass.getField("MODULE$").get(null); + Method parseItem = parserClass.getMethod("parseItem", net.enilink.komma.core.URI.class, + net.enilink.komma.core.URI.class, JValue.class, long.class); + return (Box>) parseItem.invoke(parser, + URIs.createURI("http://foo.com/linkedfactory/"), KvinIngestionWorkload.CONTEXT, + json, KvinIngestionWorkload.START_TIME); + } +} \ No newline at end of file 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..e9abda15 --- /dev/null +++ b/docs/benchmarks/kvin-ingestion.md @@ -0,0 +1,294 @@ +# KVIN Ingestion Benchmarks + +This benchmark compares one deterministic 30,000-tuple ingestion batch through +the direct KVIN LevelDB API and the in-process JSON and CSV service endpoints. +It is an opt-in test benchmark and does not start the POD or open network +sockets. + +## Quick Start + +All commands run from the repository root. Results are written as JSON under the +relevant module `target/` directory. + +**1. Build once:** + +```sh +mvn -U clean install -DskipTests +``` + +This clean reactor build compiles the benchmark sources and generates JMH's +benchmark registry. + +**2. Run the benchmarks with the comparison protocol** (3 warmups, 5 measurements, 2 forks, 1 thread): + +```sh + +### Primary ingestion benchmarks (putBatch, postJson, postCsv, putCsvDirect, postCsvSequentialFiles) + +mvn -pl bundles/io.github.linkedfactory.service -Pjmh \ + -Djmh.warmups=3 -Djmh.measurements=5 -Djmh.forks=2 \ + -Djmh.result.file=target/jmh-primary.json test-compile exec:exec + +### CSV attribution diagnostics + +mvn -pl bundles/io.github.linkedfactory.service -Pjmh \ + -Djmh.includes=io.github.linkedfactory.service.benchmark.KvinIngestionCsvDiagnosticBenchmark \ + -Djmh.warmups=3 -Djmh.measurements=5 -Djmh.forks=2 \ + -Djmh.result.file=target/jmh-csv-diagnostic.json test-compile exec:exec + +### JSON diagnostics + +mvn -pl bundles/io.github.linkedfactory.service -Pjmh \ + -Djmh.includes=io.github.linkedfactory.service.benchmark.KvinIngestionJsonDiagnosticBenchmark \ + -Djmh.warmups=3 -Djmh.measurements=5 -Djmh.forks=2 \ + -Djmh.result.file=target/jmh-json-diagnostic.json test-compile exec:exec +``` + +**3. Read the results** (method, ms/op, JMH error, derived tuples/s): + +```sh +jq -r '.[] | [(.benchmark | split(".")[-1]), .primaryMetric.score, .primaryMetric.scoreError, (30000000 / .primaryMetric.score)] | @tsv' \ + bundles/io.github.linkedfactory.service/target/jmh-primary.json +``` + +Gives output like +```sh +(method) (ms/op) (JMH error) (tuples/s) +postCsv 72.29846979999999 24.080653592587588 414946.5415103433 +postCsvSequentialFiles 80.8543817 17.14687428124646 371037.40538529155 +postJson 235.88979579999994 37.99118818211922 127178.03200540143 +putBatch 43.564955499999996 27.169511316820625 688626.8941557854 +putCsvDirect 62.84361260000001 53.04789937599848 477375.4842986222 +```` + +**4. Before drawing any conclusion,** repeat the same command as an independent +Run B with a different `-Djmh.result.file` and compare. + +#### Smoke test only + +> not a valid measurement, just checks that everything starts: + +`-Djmh.warmups=0 -Djmh.measurements=1 -Djmh.forks=1` + + +### Optional flags + +- `-Djmh.includes=`: run only the benchmarks matching the pattern. +- `-Djmh.temp.root=/tmp`: override the temporary LevelDB root without changing + production storage behavior. +- `-Djmh.warmups`, `-Djmh.measurements`, `-Djmh.forks`, `-Djmh.result.file`: + standard JMH run controls used above. + +The JMH profile never runs during a normal build or `mvn test`. Benchmarks only +start when you explicitly pass `-Pjmh` and call the `exec:exec` goal. This keeps +them out of standard build and CI pipelines. + +## Prerequisites + +- JDK 21. The project release is Java 21. +- Maven 3.9.x or another version new enough for `scala-maven-plugin:4.9.10`. + Maven 3.9.9 is known to work in this repository. +- A quiet host with a stable power mode. Use the same commit, workload source, + JDK, filesystem, and JMH settings when comparing runs. + +The benchmark creates a fresh temporary LevelDB store for every invocation, +warms six URI IDs with six preseed tuples, validates the persisted set, closes +the store, and removes the temporary directory. + +--- + +## Reference + +### Workload + +Each operation is one request or one direct batch containing exactly 30,000 +`KvinTuple` values: 5,000 rows across six stable item URIs, one property, one +context, 1,000 timestamps, and sequence numbers 1 through 5. JSON, CSV, and +direct input normalize to the same tuple set. Payload generation and +serialization happen during JMH trial setup, outside measured methods. The +prebuilt tuple list uses the same row-major order emitted by the CSV parser, so +the direct/CSV comparison does not also compare insertion orders. + +#### KVIN Tuples + +A tuple is one KVIN value with its identity and ordering metadata: + +```text +(item URI, property URI, context URI, timestamp, sequence number, value) +``` + +For example, the first canonical value is: + +```text +item: http://iwu.lf.de/ecc4p/emag/channel-1 +property: http://iwu.lf.de/ecc4p/values +context: http://iwu.lf.de/ecc4p/models/emag +time: 1710000000000 +seqNr: 1 +value: 0.25 +``` + +One CSV data row contains one `time`, one `seqNr`, and six channel values, so +it expands to six tuples. The 5,000-row workload therefore contains exactly +30,000 tuples. + +### Benchmarks + +The measured benchmarks are: + +- `putBatch`: calls `KvinLevelDb.put(Iterable)` with prebuilt tuples. + It measures KVIN encoding, warm ID resolution, batching, and LevelDB writes. +- `postCsv`: sends cached CSV bytes through an in-process Lift request and the + production CSV route. It includes request creation/routing, OpenCSV decoding, + value interpretation, tuple construction, and iterable LevelDB persistence. +- `postJson`: sends cached JSON bytes through the in-process Lift request and + production JSON route. The current route materializes JSON tuples and writes + them through scalar `store.put(tuple)` calls. +- `putCsvDirect`: connects the production CSV parser directly to the iterable + LevelDB writer, excluding Lift request creation and routing. +- `postCsvSequentialFiles`: sends ten separate 3,000-tuple CSV requests to one + service and store during the same measured invocation. + +Endpoint methods include Lift request creation, routing, request decoding, and +persistence. They exclude sockets, TLS, authentication, server startup, +fixture generation, serialization, and post-run correctness scans. + +CSV parsing and persistence are interleaved sequentially on the one JMH thread: +LevelDB pulls the next tuple from the lazy parser, processes it, then pulls the +next. Parser and writer are not parallel. + +#### Benchmark entry points visualized + +```mermaid +%%{init: {"flowchart": {"defaultRenderer": "elk"}}}%% +flowchart LR + classDef bench fill:#ffe0b2,stroke:#e65100,color:#000; + classDef prim fill:#bbdefb,stroke:#0d47a1,color:#000; + classDef store fill:#c8e6c9,stroke:#1b5e20,color:#000; + classDef prep fill:#eeeeee,stroke:#9e9e9e,color:#555,stroke-dasharray:4 3; + + subgraph PREP["Setup · not measured (JMH @Setup)"] + direction LR + G1["Payload gen. + serialization"]:::prep + G2["Fresh preseeded LevelDB"]:::prep + G3["Warm 6 URI IDs & preseed tuples"]:::prep + end + + subgraph TIMED["Measured (Benchmark case in orange)"] + direction LR + PB["putBatch"]:::bench + PD["putCsvDirect"]:::bench + PC["postCsv (1x)"]:::bench + PS["postCsvSequentialFiles (10x)"]:::bench + PJ["postJson"]:::bench + + LIFT["Lift request + routing"]:::prim + CSVP["CsvFormatParser.parse()
lazy iterator"]:::prim + JSONP["JSON route
materialize tuples"]:::prim + IW["Iterable LevelDB writer
encode + IDs + keys + WriteBatch"]:::prim + SP["Scalar store.put(tuple)"]:::prim + DB[("LevelDB store")]:::store + + PB --> IW + PD --> CSVP + PC --> LIFT + PS --> LIFT + PJ --> LIFT + LIFT --> CSVP + LIFT --> JSONP + CSVP -. "lazy pull · same thread" .-> IW + JSONP --> SP + IW --> DB + SP --> DB + end + + PREP -.-> TIMED +``` + +#### Measured steps in benchmarks + +| Benchmark | Lift routing | Parse | Writer | LevelDB | +|---|---|---|---|---| +| `putBatch` | – | – | ✓ Iterable writer | ✓ | +| `putCsvDirect` | – | ✓ (csv) | ✓ Iterable writer | ✓ | +| `postCsv` | ✓ | ✓ (csv) | ✓ Iterable writer | ✓ | +| `postCsvSequentialFiles` | ✓ ×10 | ✓ (10x csv) | ✓ Iterable writer (shared) | ✓ (shared) | +| `postJson` | ✓ | ✓ (json) | ✓ Scalar put | ✓ | + +### Running a subset + +The class-wide default runs `putBatch`, `postJson`, `postCsv`, `putCsvDirect`, +and `postCsvSequentialFiles`. To run only the four current direct/CSV controls: + +```sh +mvn -pl bundles/io.github.linkedfactory.service -Pjmh \ + -Djmh.includes='io.github.linkedfactory.service.benchmark.KvinIngestionBenchmark\.(putBatch|putCsvDirect|postCsv|postCsvSequentialFiles)' \ + -Djmh.warmups=3 -Djmh.measurements=5 -Djmh.forks=2 \ + -Djmh.result.file=target/jmh-csv-controls.json test-compile exec:exec +``` + +### In-depth attribution + +#### CSV attribution diagnostics + +This suite attributes CSV ingestion cost to its individual stages: decoding, +parsing, and routing. Each benchmark adds one more layer on top of the previous +one, so read them as nested boundaries, not additive stages. You cannot subtract +one score from another to get an exact per-stage time (see "Reading Results"), +but you can use them to decide where to profile or optimize. + +| Benchmark | Adds on top of the previous stage | Measures | +|---|---|---| +| `consumePrebuilt` | Nothing; consumes prebuilt tuples via `Blackhole` | The consumption floor / baseline iteration cost | +| `decodeCsvAndConsumeFields` | OpenCSV reader, tokenization, field string creation (production parser config) | Raw CSV decoding cost | +| `parseCsvAndConsumeTuples` | Header mapping, trimming, type interpretation, `KvinTuple` creation (no LevelDB) | Turning fields into real tuples | +| `postCsvParseOnly` | Real in-process Lift route + iterable sink that discards tuples | Routing cost, without persistence | + +#### JSON diagnostics + +The JSON diagnostic class isolates the JSON path in the same way: + +| Benchmark | Measures | +|---|---| +| `postJsonParseOnly` | The production JSON request path against a non-persistent result | +| `putScalar` | Prebuilt tuples written through the scalar KVIN API, matching the persistence style the JSON route currently uses | + +### Reading Results + +JMH reports a mean and an error interval for each single-shot batch. Use the +raw JSON for the values, not a hand-timed loop. The derived rates are: + +```text +tuples/s = 30,000,000 / batch_ms +payload MiB/s = payload_bytes / 1,048,576 / (batch_ms / 1,000) +``` + +The `jq` command in the Quick Start prints method, mean milliseconds per +operation, JMH error, and derived tuples per second. Include the raw JMH +environment header and error intervals when publishing results. + +Interpret the controls as boundaries, not additive stages: + +- `putBatch` is the shared prebuilt persistence baseline. +- `putCsvDirect` adds lazy CSV decoding, conversion, and tuple allocation, but + excludes Lift routing. +- `postCsv` is the authoritative in-process CSV endpoint result. +- `postCsvSequentialFiles - postCsv` is a directional request-splitting signal + for ten files, not multipart or network overhead. +- `parseCsvAndConsumeTuples` isolates production tuple parsing without LevelDB; + `decodeCsvAndConsumeFields` is a tokenizer/field-allocation control. + +Independent benchmark scores cannot be subtracted into exact code-stage times, +because sources and sinks interact and the uncertainty intervals may overlap. +Use diagnostics to choose a profiler or optimization candidate, then require the +complete `postCsv` result to improve in two independent runs. + +### Local Evolution Log + +Machine-specific history is deliberately not tracked. When doing performance +work, create this append-only file: + +```sh +mkdir -p .cache-main/benchmarks +touch .cache-main/benchmarks/kvin-ingestion-evolution.md +``` \ No newline at end of file From 54097733888fa0fbba92b10a9387c94edc2f058f Mon Sep 17 00:00:00 2001 From: Ken Wenzel Date: Mon, 20 Jul 2026 13:08:33 +0200 Subject: [PATCH 2/5] Adapt to pull-based JsonFormatParser --- .../benchmark/KvinIngestionBenchmark.java | 62 ++++++++++--------- .../benchmark/KvinIngestionWorkloadTest.java | 59 ++++-------------- 2 files changed, 46 insertions(+), 75 deletions(-) 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 index 0e9751d7..83985ac1 100644 --- 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 @@ -4,9 +4,9 @@ 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 io.github.linkedfactory.service.util.JsonFormatParser$; import net.enilink.commons.iterator.IExtendedIterator; import net.enilink.komma.core.KommaModule; import net.enilink.komma.core.URI; @@ -18,6 +18,7 @@ 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; @@ -39,6 +40,10 @@ import org.openjdk.jmh.annotations.Threads; import org.openjdk.jmh.annotations.Warmup; import org.junit.Assert; +import org.openjdk.jmh.runner.Runner; +import org.openjdk.jmh.runner.RunnerException; +import org.openjdk.jmh.runner.options.Options; +import org.openjdk.jmh.runner.options.OptionsBuilder; import scala.Function0; import scala.PartialFunction; import scala.collection.immutable.Nil$; @@ -46,6 +51,7 @@ 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; @@ -56,11 +62,10 @@ import java.util.Set; import java.util.concurrent.TimeUnit; -@BenchmarkMode(Mode.SingleShotTime) -@OutputTimeUnit(TimeUnit.MILLISECONDS) +@BenchmarkMode(Mode.Throughput) @Warmup(iterations = 3) -@Measurement(iterations = 5) -@Fork(2) +@Measurement(iterations = 3) +@Fork(1) @Threads(1) public class KvinIngestionBenchmark { private static final int SEQUENTIAL_CSV_FILE_COUNT = 10; @@ -84,15 +89,10 @@ public void setupTrial() { jsonPayload = workload.jsonPayload(); csvPayload = workload.csvPayload(); csvPayloads = workload.csvPayloads(SEQUENTIAL_CSV_FILE_COUNT); - try { - 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")); - modelSet = factory.createModelSet(MODELS.NAMESPACE_URI.appendFragment("MemoryModelSet")); - } catch (ClassNotFoundException e) { - throw new IllegalStateException("Could not initialize the Komma model set", e); - } + 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))); } @@ -163,11 +163,8 @@ public void putCsvDirect() throws IOException { URIs.createURI("http://foo.com/linkedfactory/"), ',', new ByteArrayInputStream(csvPayload)); parser.setContext(KvinIngestionWorkload.CONTEXT); - IExtendedIterator tuples = parser.parse(); - try { + try (IExtendedIterator tuples = parser.parse()) { store.put(tuples); - } finally { - tuples.close(); } measuredWrites = true; } @@ -219,12 +216,16 @@ public URI contextModelUri() { } @Override - public Box saveValues(JValue json, scala.collection.immutable.List path, long currentTime) { + public Box saveJsonValues(InputStream in, scala.collection.immutable.List path, long currentTime) { if (!parseOnly) { - return super.saveValues(json, path, currentTime); + return super.saveJsonValues(in, path, currentTime); } - return JsonFormatParser$.MODULE$.parseItem(URIs.createURI("http://foo.com/linkedfactory/"), - contextModelUri(), json, currentTime); + try { + new JsonFormatParser(in).parse(currentTime).toList(); // parse and discard the tuples + } catch (IOException e) { + throw new RuntimeException(e); + } + return Empty$.MODULE$; } @Override @@ -236,7 +237,7 @@ public Function0> apply(Req in) { if (isDefinedAt(in)) { return super.apply(in); } - return () -> Box.legacyNullTest((LiftResponse) null); + return (Function0) (() -> Box.legacyNullTest((LiftResponse) null)); } finally { currentModelSet.getUnitOfWork().end(); } @@ -251,14 +252,11 @@ private void validateStore() { } Set actual = new HashSet<>(); for (URI item : KvinIngestionWorkload.ITEMS) { - IExtendedIterator iterator = store.fetch(item, KvinIngestionWorkload.PROPERTY, - KvinIngestionWorkload.CONTEXT, 0); - try { + try (IExtendedIterator iterator = store.fetch(item, KvinIngestionWorkload.PROPERTY, + KvinIngestionWorkload.CONTEXT, 0)) { while (iterator.hasNext()) { actual.add(iterator.next()); } - } finally { - iterator.close(); } } Assert.assertEquals("Unexpected persisted KVIN tuples", expected, actual); @@ -286,6 +284,14 @@ public FileVisitResult postVisitDirectory(Path dir, IOException exception) throw } } + public static void main(String[] args) throws RunnerException { + Options opt = new OptionsBuilder() + .include(KvinIngestionBenchmark.class.getSimpleName() + "\\.") // adapt to control which benchmark tests to run + .forks(1) + .build(); + new Runner(opt).run(); + } + @Benchmark public void putBatch(BenchmarkState state) { state.putBatch(); 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 index b7b2c9f9..c4e1f072 100644 --- 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 @@ -2,27 +2,19 @@ import io.github.linkedfactory.core.kvin.KvinTuple; import io.github.linkedfactory.core.kvin.util.CsvFormatParser; -import io.github.linkedfactory.service.util.JsonFormatParser; +import io.github.linkedfactory.core.kvin.util.JsonFormatParser; import net.enilink.commons.iterator.IExtendedIterator; import net.enilink.komma.core.URIs; -import net.liftweb.common.Box; -import org.json4s.AsJsonInput; -import org.json4s.JValue; import org.junit.Test; -import scala.collection.immutable.List; import java.io.ByteArrayInputStream; -import java.io.StringReader; -import java.lang.reflect.Method; import java.nio.charset.StandardCharsets; import java.util.HashMap; import java.util.HashSet; import java.util.Map; import java.util.Set; -import static org.junit.Assert.assertEquals; -import static org.junit.Assert.assertFalse; -import static org.junit.Assert.assertTrue; +import static org.junit.Assert.*; public class KvinIngestionWorkloadTest { @Test @@ -59,31 +51,20 @@ public void payloadsNormalizeToTheCanonicalTupleSet() throws Exception { CsvFormatParser csvParser = new CsvFormatParser(URIs.createURI("http://foo.com/linkedfactory/"), ',', new ByteArrayInputStream(workload.csvPayload())); csvParser.setContext(KvinIngestionWorkload.CONTEXT); - IExtendedIterator csvIterator = csvParser.parse(); - try { + try (IExtendedIterator csvIterator = csvParser.parse()) { while (csvIterator.hasNext()) { KvinTuple tuple = csvIterator.next(); assertEquals("CSV tuple order", workload.tuples().get(csvCount), tuple); csvTuples.add(tuple); csvCount++; } - } finally { - csvIterator.close(); } assertEquals("CSV tuple count", KvinIngestionWorkload.TUPLE_COUNT, csvCount); assertEquals(expected, csvTuples); - JValue json = parseJson(new String(workload.jsonPayload(), StandardCharsets.UTF_8)); - Box> parsed = parseProductionJson(json); - assertTrue(parsed.isDefined()); - Set jsonTuples = new HashSet<>(); - @SuppressWarnings("unchecked") - List parsedTuples = (List) parsed.openOr(null); - assertEquals("JSON tuple count", KvinIngestionWorkload.TUPLE_COUNT, parsedTuples.size()); - scala.collection.Iterator jsonIterator = parsedTuples.iterator(); - while (jsonIterator.hasNext()) { - jsonTuples.add(jsonIterator.next()); - } + java.util.List json = parseJson(new String(workload.jsonPayload(), StandardCharsets.UTF_8)); + assertEquals("JSON tuple count", KvinIngestionWorkload.TUPLE_COUNT, json.size()); + Set jsonTuples = new HashSet<>(json); Set missing = new HashSet<>(expected); missing.removeAll(jsonTuples); Set extra = new HashSet<>(jsonTuples); @@ -92,7 +73,7 @@ public void payloadsNormalizeToTheCanonicalTupleSet() throws Exception { assertTrue("JSON missing=" + sample(missing) + ", extra=" + sample(extra), missing.isEmpty() && extra.isEmpty()); assertEquals(KvinIngestionWorkload.ROW_COUNT + 1, new String(workload.csvPayload(), StandardCharsets.UTF_8).split("\\n").length); - assertFalse(workload.jsonPayload().length == 0); + assertNotEquals(0, workload.jsonPayload().length); } @Test @@ -108,14 +89,11 @@ public void csvFilePartitionsNormalizeToTheCanonicalTupleSet() throws Exception CsvFormatParser parser = new CsvFormatParser(URIs.createURI("http://foo.com/linkedfactory/"), ',', new ByteArrayInputStream(payload)); parser.setContext(KvinIngestionWorkload.CONTEXT); - IExtendedIterator tuples = parser.parse(); - try { + try (IExtendedIterator tuples = parser.parse()) { while (tuples.hasNext()) { actual.add(tuples.next()); tupleCount++; } - } finally { - tuples.close(); } } @@ -133,22 +111,9 @@ private static String sample(Set tuples) { return tuples.stream().limit(3).toList().toString(); } - private static JValue parseJson(String json) throws Exception { - Class parserClass = Class.forName("org.json4s.native.JsonParser$"); - Object parser = parserClass.getField("MODULE$").get(null); - Method parse = parserClass.getMethod("parse", java.io.Reader.class, boolean.class, boolean.class, - boolean.class); - return (JValue) parse.invoke(parser, new StringReader(json), true, false, true); - } - - @SuppressWarnings("unchecked") - private static Box> parseProductionJson(JValue json) throws Exception { - Class parserClass = Class.forName("io.github.linkedfactory.service.util.JsonFormatParser$"); - Object parser = parserClass.getField("MODULE$").get(null); - Method parseItem = parserClass.getMethod("parseItem", net.enilink.komma.core.URI.class, - net.enilink.komma.core.URI.class, JValue.class, long.class); - return (Box>) parseItem.invoke(parser, - URIs.createURI("http://foo.com/linkedfactory/"), KvinIngestionWorkload.CONTEXT, - json, KvinIngestionWorkload.START_TIME); + private static java.util.List parseJson(String json) throws Exception { + return new JsonFormatParser(new ByteArrayInputStream(json.getBytes(StandardCharsets.UTF_8))) + .setContext(KvinIngestionWorkload.CONTEXT) + .parse().toList(); } } \ No newline at end of file From 2c99e8be1ec2632ea67027c6ef0d88767a8dcd91 Mon Sep 17 00:00:00 2001 From: Ken Wenzel Date: Mon, 20 Jul 2026 16:26:27 +0200 Subject: [PATCH 3/5] Use sorted JSON payload. --- .../service/benchmark/KvinIngestionWorkload.java | 8 +++++++- 1 file changed, 7 insertions(+), 1 deletion(-) 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 index 366acd11..9d012f12 100644 --- 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 @@ -10,6 +10,7 @@ import java.io.UncheckedIOException; import java.nio.charset.StandardCharsets; import java.util.ArrayList; +import java.util.Comparator; import java.util.List; public final class KvinIngestionWorkload { @@ -95,7 +96,12 @@ private static List createPreseedTuples() { private static byte[] createJsonPayload(List tuples) { try { - String json = JsonFormatWriter.toJsonString(WrappedIterator.create(tuples.iterator())); + 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); From 733a33d632aaa2c9071bedc34e682a842a077ac9 Mon Sep 17 00:00:00 2001 From: Willy Mehling Date: Tue, 21 Jul 2026 14:09:27 +0200 Subject: [PATCH 4/5] Replace reflection for direct class names --- .../linkedfactory/core/kvin/KvinHttpTest.java | 10 ++++++---- .../KvinIngestionCsvDiagnosticBenchmark.java | 16 +++++++--------- 2 files changed, 13 insertions(+), 13 deletions(-) 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..4776507b 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,15 +92,17 @@ 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))); } - + @AfterClass public static void tearDownClass() { modelSet.dispose(); 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 index ce4e7ca9..8c0a9c0a 100644 --- 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 @@ -78,15 +78,13 @@ public static class BenchmarkState { public void setupTrial() { workload = new KvinIngestionWorkload(); csvPayload = workload.csvPayload(); - try { - 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")); - modelSet = factory.createModelSet(MODELS.NAMESPACE_URI.appendFragment("MemoryModelSet")); - } catch (ClassNotFoundException e) { - throw new IllegalStateException("Could not initialize the Komma model set", e); - } + + 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); From 0d444e0183eff3029c282e34a68fccfd81c47df1 Mon Sep 17 00:00:00 2001 From: Willy Mehling Date: Tue, 21 Jul 2026 15:28:29 +0200 Subject: [PATCH 5/5] Add deterministic KVIN benchmark variants --- .../io.github.linkedfactory.service/pom.xml | 6 + .../benchmark/KvinIngestionBenchmark.java | 40 +++- .../benchmark/KvinIngestionWorkload.java | 108 +++++++--- .../benchmark/KvinIngestionWorkloadTest.java | 197 +++++++++++------- docs/benchmarks/kvin-ingestion.md | 139 +++++++----- 5 files changed, 320 insertions(+), 170 deletions(-) diff --git a/bundles/io.github.linkedfactory.service/pom.xml b/bundles/io.github.linkedfactory.service/pom.xml index 57f433d2..dd6ccbcc 100644 --- a/bundles/io.github.linkedfactory.service/pom.xml +++ b/bundles/io.github.linkedfactory.service/pom.xml @@ -231,7 +231,9 @@ json ${project.build.directory}/jmh-result.json 3 + 3s 5 + 3s 2 @@ -252,8 +254,12 @@ ${jmh.includes} -wi ${jmh.warmups} + -w + ${jmh.warmup.time} -i ${jmh.measurements} + -r + ${jmh.measurement.time} -f ${jmh.forks} -rf 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 index 83985ac1..3decded6 100644 --- 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 @@ -33,6 +33,7 @@ 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; @@ -63,15 +64,21 @@ import java.util.concurrent.TimeUnit; @BenchmarkMode(Mode.Throughput) +@OutputTimeUnit(TimeUnit.SECONDS) @Warmup(iterations = 3) -@Measurement(iterations = 3) -@Fork(1) +@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; @@ -85,10 +92,12 @@ public static class BenchmarkState { @Setup(Level.Trial) public void setupTrial() { - workload = new KvinIngestionWorkload(); - jsonPayload = workload.jsonPayload(); - csvPayload = workload.csvPayload(); - csvPayloads = workload.csvPayloads(SEQUENTIAL_CSV_FILE_COUNT); + 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); @@ -98,6 +107,12 @@ public void setupTrial() { @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-") @@ -243,7 +258,7 @@ public Function0> apply(Req in) { } }); } - } + } private void validateStore() { Set expected = new HashSet<>(workload.preseedTuples()); @@ -251,8 +266,8 @@ private void validateStore() { expected.addAll(workload.tuples()); } Set actual = new HashSet<>(); - for (URI item : KvinIngestionWorkload.ITEMS) { - try (IExtendedIterator iterator = store.fetch(item, KvinIngestionWorkload.PROPERTY, + for (URI item : workload.items()) { + try (IExtendedIterator iterator = store.fetch(item, workload.property(), KvinIngestionWorkload.CONTEXT, 0)) { while (iterator.hasNext()) { actual.add(iterator.next()); @@ -293,27 +308,32 @@ public static void main(String[] args) throws RunnerException { } @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(); } -} \ No newline at end of file +} 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 index 9d012f12..e9187934 100644 --- 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 @@ -10,8 +10,11 @@ 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; @@ -19,28 +22,66 @@ public final class KvinIngestionWorkload { 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 long START_TIME = 1_710_000_000_000L; + 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 PROPERTY = URIs.createURI("http://iwu.lf.de/ecc4p/values"); public static final URI CONTEXT = URIs.createURI("http://iwu.lf.de/ecc4p/models/emag"); - public static final List ITEMS = List.of( - URIs.createURI("http://iwu.lf.de/ecc4p/emag/channel-1"), - URIs.createURI("http://iwu.lf.de/ecc4p/emag/channel-2"), - URIs.createURI("http://iwu.lf.de/ecc4p/emag/channel-3"), - URIs.createURI("http://iwu.lf.de/ecc4p/emag/channel-4"), - URIs.createURI("http://iwu.lf.de/ecc4p/emag/channel-5"), - URIs.createURI("http://iwu.lf.de/ecc4p/emag/channel-6")); + 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(); + 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() { @@ -72,26 +113,25 @@ public List csvPayloads(int fileCount) { return List.copyOf(payloads); } - private static List createTuples() { - List tuples = new ArrayList<>(TUPLE_COUNT); + 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 = START_TIME + row / SEQUENCES_PER_TIMESTAMP; + long time = startTime + (row / SEQUENCES_PER_TIMESTAMP) * TIMESTAMP_STEP; int seqNr = row % SEQUENCES_PER_TIMESTAMP + 1; - double value = value(channel, row); - tuples.add(new KvinTuple(ITEMS.get(channel), PROPERTY, CONTEXT, time, seqNr, value)); + result.add(new KvinTuple(items.get(channel), property, CONTEXT, time, seqNr, value(channel, row))); } } - return List.copyOf(tuples); + return List.copyOf(result); } - private static List createPreseedTuples() { - List tuples = new ArrayList<>(CHANNEL_COUNT); + private List createPreseedTuples() { + List result = new ArrayList<>(CHANNEL_COUNT); for (int channel = 0; channel < CHANNEL_COUNT; channel++) { - tuples.add(new KvinTuple(ITEMS.get(channel), PROPERTY, CONTEXT, START_TIME - 1, 0, + result.add(new KvinTuple(items.get(channel), property, CONTEXT, startTime - TIMESTAMP_STEP, 0, value(channel, -1))); } - return List.copyOf(tuples); + return List.copyOf(result); } private static byte[] createJsonPayload(List tuples) { @@ -108,20 +148,16 @@ private static byte[] createJsonPayload(List tuples) { } } - private static byte[] createCsvPayload() { - return createCsvPayload(0, ROW_COUNT); - } - - private static byte[] createCsvPayload(int startRow, int endRow) { + 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); + for (URI item : items) { + csv.append(',').append(item).append('@').append(property); } csv.append('\n'); for (int row = startRow; row < endRow; row++) { - csv.append(START_TIME + row / SEQUENCES_PER_TIMESTAMP) + 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)); @@ -131,7 +167,21 @@ private static byte[] createCsvPayload(int startRow, int endRow) { 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; } -} \ No newline at end of file +} 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 index c4e1f072..9f5520dc 100644 --- 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 @@ -4,116 +4,153 @@ 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.*; +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 hasTheExpectedDeterministicShape() { - KvinIngestionWorkload workload = new KvinIngestionWorkload(); - assertEquals(KvinIngestionWorkload.TUPLE_COUNT, workload.tuples().size()); - assertEquals(KvinIngestionWorkload.CHANNEL_COUNT, new HashSet<>(workload.tuples().stream() - .map(tuple -> tuple.item).toList()).size()); - assertEquals(KvinIngestionWorkload.TUPLE_COUNT + KvinIngestionWorkload.CHANNEL_COUNT, - workload.tuples().size() + workload.preseedTuples().size()); - - Map> sequencesByTimestamp = new HashMap<>(); - Set keys = new HashSet<>(); - for (KvinTuple tuple : workload.tuples()) { - assertEquals(KvinIngestionWorkload.PROPERTY, tuple.property); - assertEquals(KvinIngestionWorkload.CONTEXT, tuple.context); - assertTrue(tuple.seqNr >= 1 && tuple.seqNr <= KvinIngestionWorkload.SEQUENCES_PER_TIMESTAMP); - assertTrue(keys.add(key(tuple))); - sequencesByTimestamp.computeIfAbsent(tuple.time, ignored -> new HashSet<>()).add(tuple.seqNr); + 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.TIMESTAMP_COUNT, sequencesByTimestamp.size()); - assertTrue(sequencesByTimestamp.values().stream() - .allMatch(sequences -> sequences.size() == KvinIngestionWorkload.SEQUENCES_PER_TIMESTAMP)); - assertEquals(workload.tuples(), new KvinIngestionWorkload().tuples()); + 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 payloadsNormalizeToTheCanonicalTupleSet() throws Exception { - KvinIngestionWorkload workload = new KvinIngestionWorkload(); - Set expected = new HashSet<>(workload.tuples()); - - Set csvTuples = new HashSet<>(); - int csvCount = 0; - CsvFormatParser csvParser = new CsvFormatParser(URIs.createURI("http://foo.com/linkedfactory/"), ',', - new ByteArrayInputStream(workload.csvPayload())); - csvParser.setContext(KvinIngestionWorkload.CONTEXT); - try (IExtendedIterator csvIterator = csvParser.parse()) { - while (csvIterator.hasNext()) { - KvinTuple tuple = csvIterator.next(); - assertEquals("CSV tuple order", workload.tuples().get(csvCount), tuple); - csvTuples.add(tuple); - csvCount++; + 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()); } - assertEquals("CSV tuple count", KvinIngestionWorkload.TUPLE_COUNT, csvCount); - assertEquals(expected, csvTuples); - - java.util.List json = parseJson(new String(workload.jsonPayload(), StandardCharsets.UTF_8)); - assertEquals("JSON tuple count", KvinIngestionWorkload.TUPLE_COUNT, json.size()); - Set jsonTuples = new HashSet<>(json); - Set missing = new HashSet<>(expected); - missing.removeAll(jsonTuples); - Set extra = new HashSet<>(jsonTuples); - extra.removeAll(expected); - assertEquals(expected.size(), jsonTuples.size()); - assertTrue("JSON missing=" + sample(missing) + ", extra=" + sample(extra), missing.isEmpty() && extra.isEmpty()); - assertEquals(KvinIngestionWorkload.ROW_COUNT + 1, - new String(workload.csvPayload(), StandardCharsets.UTF_8).split("\\n").length); - assertNotEquals(0, workload.jsonPayload().length); } @Test - public void csvFilePartitionsNormalizeToTheCanonicalTupleSet() throws Exception { - KvinIngestionWorkload workload = new KvinIngestionWorkload(); - Set expected = new HashSet<>(workload.tuples()); - - for (int fileCount : java.util.List.of(1, 2, 5, 10)) { - Set actual = new HashSet<>(); - int tupleCount = 0; - java.util.List payloads = workload.csvPayloads(fileCount); + 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) { - CsvFormatParser parser = new CsvFormatParser(URIs.createURI("http://foo.com/linkedfactory/"), ',', - new ByteArrayInputStream(payload)); - parser.setContext(KvinIngestionWorkload.CONTEXT); - try (IExtendedIterator tuples = parser.parse()) { - while (tuples.hasNext()) { - actual.add(tuples.next()); - tupleCount++; - } - } + partitionedCsv.addAll(parseCsv(payload)); } + assertEquals(KvinIngestionWorkload.TUPLE_COUNT, partitionedCsv.size()); + assertEquals(workload.tuples(), partitionedCsv); + assertEquals(expected, new HashSet<>(partitionedCsv)); - assertEquals(fileCount, payloads.size()); - assertEquals(KvinIngestionWorkload.TUPLE_COUNT, tupleCount); - assertEquals(expected, actual); + assertEquals(KvinIngestionWorkload.ROW_COUNT + 1, + new String(workload.csvPayload(), StandardCharsets.UTF_8).split("\\n").length); + assertNotEquals(0, workload.jsonPayload().length); } } - private static String key(KvinTuple tuple) { - return tuple.context + "|" + tuple.item + "|" + tuple.property + "|" + tuple.time + "|" + tuple.seqNr; - } - - private static String sample(Set tuples) { - return tuples.stream().limit(3).toList().toString(); + 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 java.util.List parseJson(String json) throws Exception { - return new JsonFormatParser(new ByteArrayInputStream(json.getBytes(StandardCharsets.UTF_8))) + private static List parseJson(byte[] payload) throws Exception { + return new JsonFormatParser(new ByteArrayInputStream(payload)) .setContext(KvinIngestionWorkload.CONTEXT) .parse().toList(); } -} \ No newline at end of file + + 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/docs/benchmarks/kvin-ingestion.md b/docs/benchmarks/kvin-ingestion.md index e9abda15..983eba45 100644 --- a/docs/benchmarks/kvin-ingestion.md +++ b/docs/benchmarks/kvin-ingestion.md @@ -1,9 +1,9 @@ # KVIN Ingestion Benchmarks -This benchmark compares one deterministic 30,000-tuple ingestion batch through +This benchmark compares deterministic 30,000-tuple ingestion batches through the direct KVIN LevelDB API and the in-process JSON and CSV service endpoints. It is an opt-in test benchmark and does not start the POD or open network -sockets. +sockets. Primary results are reported directly as tuples per second. ## Quick Start @@ -19,24 +19,31 @@ mvn -U clean install -DskipTests This clean reactor build compiles the benchmark sources and generates JMH's benchmark registry. -**2. Run the benchmarks with the comparison protocol** (3 warmups, 5 measurements, 2 forks, 1 thread): +**2. Run the primary benchmarks with the comparison protocol** (3 three-second +warmups, 5 three-second measurements, 2 forks, 1 thread): ```sh +mvn -pl bundles/io.github.linkedfactory.service -Pjmh \ + -Djmh.result.file=target/jmh-primary-run-a.json test-compile exec:exec +``` -### Primary ingestion benchmarks (putBatch, postJson, postCsv, putCsvDirect, postCsvSequentialFiles) +The defaults exercise five methods for 240 timed seconds +(`5 methods × 8 iterations × 2 forks × 3 seconds`); including trial setup, +invocation setup, validation, and store cleanup, allow approximately five +minutes. Repeat the command with `jmh-primary-run-b.json` before comparing +implementations. -mvn -pl bundles/io.github.linkedfactory.service -Pjmh \ - -Djmh.warmups=3 -Djmh.measurements=5 -Djmh.forks=2 \ - -Djmh.result.file=target/jmh-primary.json test-compile exec:exec +**3. Run attribution diagnostics when needed:** -### CSV attribution diagnostics +```sh +# CSV attribution diagnostics mvn -pl bundles/io.github.linkedfactory.service -Pjmh \ -Djmh.includes=io.github.linkedfactory.service.benchmark.KvinIngestionCsvDiagnosticBenchmark \ -Djmh.warmups=3 -Djmh.measurements=5 -Djmh.forks=2 \ -Djmh.result.file=target/jmh-csv-diagnostic.json test-compile exec:exec -### JSON diagnostics +# JSON diagnostics mvn -pl bundles/io.github.linkedfactory.service -Pjmh \ -Djmh.includes=io.github.linkedfactory.service.benchmark.KvinIngestionJsonDiagnosticBenchmark \ @@ -44,31 +51,21 @@ mvn -pl bundles/io.github.linkedfactory.service -Pjmh \ -Djmh.result.file=target/jmh-json-diagnostic.json test-compile exec:exec ``` -**3. Read the results** (method, ms/op, JMH error, derived tuples/s): +**4. Read the primary results** (method, tuples/s, JMH error): ```sh -jq -r '.[] | [(.benchmark | split(".")[-1]), .primaryMetric.score, .primaryMetric.scoreError, (30000000 / .primaryMetric.score)] | @tsv' \ - bundles/io.github.linkedfactory.service/target/jmh-primary.json +jq -r '.[] | [(.benchmark | split(".")[-1]), .primaryMetric.score, .primaryMetric.scoreError, .primaryMetric.scoreUnit] | @tsv' \ + bundles/io.github.linkedfactory.service/target/jmh-primary-run-a.json ``` -Gives output like -```sh -(method) (ms/op) (JMH error) (tuples/s) -postCsv 72.29846979999999 24.080653592587588 414946.5415103433 -postCsvSequentialFiles 80.8543817 17.14687428124646 371037.40538529155 -postJson 235.88979579999994 37.99118818211922 127178.03200540143 -putBatch 43.564955499999996 27.169511316820625 688626.8941557854 -putCsvDirect 62.84361260000001 53.04789937599848 477375.4842986222 -```` - -**4. Before drawing any conclusion,** repeat the same command as an independent -Run B with a different `-Djmh.result.file` and compare. +Diagnostic suites deliberately remain single-shot batch timings in `ms/op` and +do not use `OperationsPerInvocation`. -#### Smoke test only +#### Smoke test only > not a valid measurement, just checks that everything starts: -`-Djmh.warmups=0 -Djmh.measurements=1 -Djmh.forks=1` +`-Djmh.warmups=0 -Djmh.measurements=1 -Djmh.measurement.time=1s -Djmh.forks=1` ### Optional flags @@ -76,8 +73,8 @@ Run B with a different `-Djmh.result.file` and compare. - `-Djmh.includes=`: run only the benchmarks matching the pattern. - `-Djmh.temp.root=/tmp`: override the temporary LevelDB root without changing production storage behavior. -- `-Djmh.warmups`, `-Djmh.measurements`, `-Djmh.forks`, `-Djmh.result.file`: - standard JMH run controls used above. +- `-Djmh.warmups`, `-Djmh.warmup.time`, `-Djmh.measurements`, + `-Djmh.measurement.time`, `-Djmh.forks`, `-Djmh.result.file`: JMH run controls. The JMH profile never runs during a normal build or `mvn test`. Benchmarks only start when you explicitly pass `-Pjmh` and call the `exec:exec` goal. This keeps @@ -92,8 +89,8 @@ them out of standard build and CI pipelines. JDK, filesystem, and JMH settings when comparing runs. The benchmark creates a fresh temporary LevelDB store for every invocation, -warms six URI IDs with six preseed tuples, validates the persisted set, closes -the store, and removes the temporary directory. +warms the selected six item/property IDs with six preseed tuples, validates the +persisted set, closes the store, and removes the temporary directory. --- @@ -101,13 +98,25 @@ the store, and removes the temporary directory. ### Workload -Each operation is one request or one direct batch containing exactly 30,000 -`KvinTuple` values: 5,000 rows across six stable item URIs, one property, one -context, 1,000 timestamps, and sequence numbers 1 through 5. JSON, CSV, and -direct input normalize to the same tuple set. Payload generation and -serialization happen during JMH trial setup, outside measured methods. The -prebuilt tuple list uses the same row-major order emitted by the CSV parser, so -the direct/CSV comparison does not also compare insertion orders. +Each invocation processes one of ten immutable variants. Every variant contains +exactly 30,000 `KvinTuple` values: 5,000 rows across six item URIs, one property, +one context, 1,000 timestamps, and sequence numbers 1 through 5. The channel +pool contains 100 zero-padded URIs and is shuffled once with seed +`0x4B56494E_20260721L`. Variants use non-overlapping six-channel slices from +that order and ten distinct properties from a pool of ten. + +Timestamps advance in one-second steps. Variant `n` starts at +`1710000000000 + n × 1000000`, giving each variant a disjoint deterministic +timestamp window. The numeric value formula remains relative to the six CSV +columns, keeping value parsing and payload sizes comparable across variants. + +All ten tuple lists, JSON payloads, CSV payloads, and ten-file CSV partitions +are generated and cached during JMH trial setup, outside measured methods. +Invocation setup rotates through variants zero through nine, then repeats. Each +trial and fork begins at variant zero. JSON, CSV, partitioned CSV, and direct +input normalize to the same tuple set for their selected variant. The prebuilt +tuple list uses the same row-major order emitted by the CSV parser, so the +direct/CSV comparison does not also compare insertion orders. #### KVIN Tuples @@ -120,8 +129,8 @@ A tuple is one KVIN value with its identity and ordering metadata: For example, the first canonical value is: ```text -item: http://iwu.lf.de/ecc4p/emag/channel-1 -property: http://iwu.lf.de/ecc4p/values +item: one deterministic selection from .../emag/channel-000 through channel-099 +property: one of .../property-00 through property-09 context: http://iwu.lf.de/ecc4p/models/emag time: 1710000000000 seqNr: 1 @@ -255,17 +264,45 @@ The JSON diagnostic class isolates the JSON path in the same way: ### Reading Results -JMH reports a mean and an error interval for each single-shot batch. Use the -raw JSON for the values, not a hand-timed loop. The derived rates are: +For the five primary methods, `OperationsPerInvocation(30000)` makes JMH report +the mean and error directly in `ops/s`, where one operation is one tuple. Read +these tuple/s values from the raw JSON rather than deriving them from rounded +batch times. An equivalent 30,000-tuple batch time can be calculated for a +presentation table as `30,000 / tuples_per_second × 1,000` milliseconds. -```text -tuples/s = 30,000,000 / batch_ms -payload MiB/s = payload_bytes / 1,048,576 / (batch_ms / 1,000) -``` +Diagnostics report single-shot batch time in `ms/op`; their unit and semantic +boundary are intentionally different. Include the JMH environment header, +score unit, and error intervals when publishing results. + +Use the three-second defaults for comparisons and require two independent full +runs. A short smoke run or a duration-sensitivity check is validation, not a +publishable performance result. + +#### Run A reference result + +Run A used the protocol above on 2026-07-21. The commit column names the base +commit; the benchmark included the workload-variant changes in this patch. -The `jq` command in the Quick Start prints method, mean milliseconds per -operation, JMH error, and derived tuples per second. Include the raw JMH -environment header and error intervals when publishing results. +| Benchmark | Tuples/s | JMH error | Equivalent 30,000-tuple batch | +|---|---:|---:|---:| +| `postCsv` | 737,270 | ±23,597 | 40.69 ms | +| `postCsvSequentialFiles` | 669,339 | ±65,246 | 44.82 ms | +| `postJson` | 499,159 | ±109,763 | 60.10 ms | +| `putBatch` | 1,468,205 | ±133,007 | 20.43 ms | +| `putCsvDirect` | 840,222 | ±196,984 | 35.70 ms | + +| Environment | Value | +|---|---| +| Base commit | `733a33d` | +| Date | 2026-07-21 | +| CPU | Intel Core i7-1185G7 @ 3.00 GHz, 4 vCPUs | +| OS | Linux 5.15.123.1-microsoft-standard-WSL2, x86_64 | +| JDK | Eclipse Temurin 21.0.11+10 LTS | + +The independent Run B scores were 761,544, 640,135, 468,624, 1,672,301, +and 920,589 tuples/s in the table's method order. Every Run A and Run B JMH +confidence interval overlapped; Run B serves as the consistency check rather +than a second result to average into Run A. Interpret the controls as boundaries, not additive stages: @@ -285,10 +322,10 @@ complete `postCsv` result to improve in two independent runs. ### Local Evolution Log -Machine-specific history is deliberately not tracked. When doing performance -work, create this append-only file: +Beyond the reference run above, machine-specific history is deliberately not +tracked. When doing performance work, create this append-only file: ```sh mkdir -p .cache-main/benchmarks touch .cache-main/benchmarks/kvin-ingestion-evolution.md -``` \ No newline at end of file +```