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..dd6ccbcc 100644 --- a/bundles/io.github.linkedfactory.service/pom.xml +++ b/bundles/io.github.linkedfactory.service/pom.xml @@ -222,4 +222,55 @@ + + + + jmh + + io.github.linkedfactory.service.benchmark.KvinIngestionBenchmark + json + ${project.build.directory}/jmh-result.json + 3 + 3s + 5 + 3s + 2 + + + + + + org.codehaus.mojo + exec-maven-plugin + 3.5.0 + + ${java.home}/bin/java + test + + -Djmh.temp.root=${jmh.temp.root} + -cp + + org.openjdk.jmh.Main + ${jmh.includes} + -wi + ${jmh.warmups} + -w + ${jmh.warmup.time} + -i + ${jmh.measurements} + -r + ${jmh.measurement.time} + -f + ${jmh.forks} + -rf + ${jmh.result.format} + -rff + ${jmh.result.file} + + + + + + + diff --git a/bundles/io.github.linkedfactory.service/src/test/java/io/github/linkedfactory/core/kvin/KvinHttpTest.java b/bundles/io.github.linkedfactory.service/src/test/java/io/github/linkedfactory/core/kvin/KvinHttpTest.java index 37803c67..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/KvinIngestionBenchmark.java b/bundles/io.github.linkedfactory.service/src/test/java/io/github/linkedfactory/service/benchmark/KvinIngestionBenchmark.java new file mode 100644 index 00000000..3decded6 --- /dev/null +++ b/bundles/io.github.linkedfactory.service/src/test/java/io/github/linkedfactory/service/benchmark/KvinIngestionBenchmark.java @@ -0,0 +1,339 @@ +package io.github.linkedfactory.service.benchmark; + +import com.google.inject.Guice; +import io.github.linkedfactory.core.kvin.KvinTuple; +import io.github.linkedfactory.core.kvin.leveldb.KvinLevelDb; +import io.github.linkedfactory.core.kvin.util.CsvFormatParser; +import io.github.linkedfactory.core.kvin.util.JsonFormatParser; +import io.github.linkedfactory.service.KvinService; +import io.github.linkedfactory.service.MockHttpServletRequest; +import net.enilink.commons.iterator.IExtendedIterator; +import net.enilink.komma.core.KommaModule; +import net.enilink.komma.core.URI; +import net.enilink.komma.core.URIs; +import net.enilink.komma.model.IModelSet; +import net.enilink.komma.model.IModelSetFactory; +import net.enilink.komma.model.MODELS; +import net.enilink.komma.model.ModelPlugin; +import net.enilink.komma.model.ModelSetModule; +import net.enilink.platform.lift.util.Globals; +import net.liftweb.common.Box; +import net.liftweb.common.Empty$; +import net.liftweb.common.Full; +import net.liftweb.http.CurrentReq$; +import net.liftweb.http.LiftResponse; +import net.liftweb.http.Req; +import net.liftweb.http.provider.servlet.HTTPRequestServlet; +import net.liftweb.util.VendorJ; +import org.json4s.JValue; +import org.openjdk.jmh.annotations.Benchmark; +import org.openjdk.jmh.annotations.BenchmarkMode; +import org.openjdk.jmh.annotations.Fork; +import org.openjdk.jmh.annotations.Level; +import org.openjdk.jmh.annotations.Measurement; +import org.openjdk.jmh.annotations.Mode; +import org.openjdk.jmh.annotations.OutputTimeUnit; +import org.openjdk.jmh.annotations.OperationsPerInvocation; +import org.openjdk.jmh.annotations.Scope; +import org.openjdk.jmh.annotations.Setup; +import org.openjdk.jmh.annotations.State; +import org.openjdk.jmh.annotations.TearDown; +import org.openjdk.jmh.annotations.Threads; +import org.openjdk.jmh.annotations.Warmup; +import org.junit.Assert; +import 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$; + +import java.io.ByteArrayInputStream; +import java.io.File; +import java.io.IOException; +import java.io.InputStream; +import java.nio.file.FileVisitResult; +import java.nio.file.Files; +import java.nio.file.Path; +import java.nio.file.SimpleFileVisitor; +import java.nio.file.attribute.BasicFileAttributes; +import java.util.HashSet; +import java.util.List; +import java.util.Set; +import java.util.concurrent.TimeUnit; + +@BenchmarkMode(Mode.Throughput) +@OutputTimeUnit(TimeUnit.SECONDS) +@Warmup(iterations = 3) +@Measurement(iterations = 5) +@Fork(2) +@Threads(1) +public class KvinIngestionBenchmark { + private static final int SEQUENTIAL_CSV_FILE_COUNT = 10; + + @State(Scope.Thread) + public static class BenchmarkState { + private List workloads; + private List jsonPayloadVariants; + private List csvPayloadVariants; + private List> csvPayloadPartitionVariants; + private int nextVariantIndex; + private KvinIngestionWorkload workload; + private IModelSet modelSet; + private KvinLevelDb store; + private File storeDirectory; + private KvinService service; + private KvinService parseOnlyService; + private byte[] jsonPayload; + private byte[] csvPayload; + private List csvPayloads; + private boolean measuredWrites; + + @Setup(Level.Trial) + public void setupTrial() { + workloads = KvinIngestionWorkload.variants(); + jsonPayloadVariants = workloads.stream().map(KvinIngestionWorkload::jsonPayload).toList(); + csvPayloadVariants = workloads.stream().map(KvinIngestionWorkload::csvPayload).toList(); + csvPayloadPartitionVariants = workloads.stream() + .map(variant -> variant.csvPayloads(SEQUENTIAL_CSV_FILE_COUNT)).toList(); + nextVariantIndex = 0; + KommaModule module = ModelPlugin.createModelSetModule(getClass().getClassLoader()); + IModelSetFactory factory = Guice.createInjector(new ModelSetModule(module)) + .getInstance(IModelSetFactory.class); + modelSet = factory.createModelSet(MODELS.NAMESPACE_URI.appendFragment("MemoryModelSet")); + Globals.contextModelSet().theDefault().set(VendorJ.vendor(new Full(modelSet))); + } + + @Setup(Level.Invocation) + public void setupInvocation() throws IOException { + int variantIndex = nextVariantIndex; + nextVariantIndex = (nextVariantIndex + 1) % KvinIngestionWorkload.VARIANT_COUNT; + workload = workloads.get(variantIndex); + jsonPayload = jsonPayloadVariants.get(variantIndex); + csvPayload = csvPayloadVariants.get(variantIndex); + csvPayloads = csvPayloadPartitionVariants.get(variantIndex); + String tempRoot = System.getProperty("jmh.temp.root", ""); + Path directory = tempRoot.isEmpty() + ? Files.createTempDirectory("kvin-ingestion-jmh-") + : Files.createTempDirectory(Path.of(tempRoot), "kvin-ingestion-jmh-"); + storeDirectory = directory.toFile(); + store = new KvinLevelDb(storeDirectory); + store.put(workload.preseedTuples()); + service = 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); + try (IExtendedIterator tuples = parser.parse()) { + store.put(tuples); + } + measuredWrites = true; + } + + public void postCsvSequentialFiles() throws IOException { + for (byte[] payload : csvPayloads) { + post(payload, "text/csv", service, true); + } + } + + private void post(byte[] payload, String contentType, KvinService targetService, + boolean writesMeasuredTuples) throws IOException { + MockHttpServletRequest request = new MockHttpServletRequest("http://foo.com/linkedfactory/values"); + request.method_$eq("POST"); + request.body_$eq(payload); + request.contentType_$eq(contentType); + Req req = Req.apply(new HTTPRequestServlet(request, null), + Nil$.MODULE$.$colon$colon(PartialFunction.empty()), System.nanoTime()); + Box result = targetService.apply(req).apply(); + LiftResponse response = result.openOr(null); + if (response == null || response.toResponse().code() != 200) { + int status = response == null ? -1 : response.toResponse().code(); + throw new IOException("KVIN ingestion request failed with HTTP status " + status); + } + if (writesMeasuredTuples) { + measuredWrites = true; + } + } + + private 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 saveJsonValues(InputStream in, scala.collection.immutable.List path, long currentTime) { + if (!parseOnly) { + return super.saveJsonValues(in, path, currentTime); + } + try { + new JsonFormatParser(in).parse(currentTime).toList(); // parse and discard the tuples + } catch (IOException e) { + throw new RuntimeException(e); + } + return Empty$.MODULE$; + } + + @Override + public Function0> apply(Req in) { + IModelSet currentModelSet = Globals.contextModelSet().vend().openOr(null); + return CurrentReq$.MODULE$.doWith(in, () -> { + try { + currentModelSet.getUnitOfWork().begin(); + if (isDefinedAt(in)) { + return super.apply(in); + } + return (Function0) (() -> Box.legacyNullTest((LiftResponse) null)); + } finally { + currentModelSet.getUnitOfWork().end(); + } + }); + } + } + + private void validateStore() { + Set expected = new HashSet<>(workload.preseedTuples()); + if (measuredWrites) { + expected.addAll(workload.tuples()); + } + Set actual = new HashSet<>(); + for (URI item : workload.items()) { + try (IExtendedIterator iterator = store.fetch(item, workload.property(), + KvinIngestionWorkload.CONTEXT, 0)) { + while (iterator.hasNext()) { + actual.add(iterator.next()); + } + } + } + Assert.assertEquals("Unexpected persisted KVIN tuples", expected, actual); + Assert.assertEquals((measuredWrites ? KvinIngestionWorkload.TUPLE_COUNT : 0) + + KvinIngestionWorkload.CHANNEL_COUNT, actual.size()); + } + + private static void deleteDirectory(Path directory) throws IOException { + if (!Files.exists(directory)) { + return; + } + Files.walkFileTree(directory, new SimpleFileVisitor<>() { + @Override + public FileVisitResult visitFile(Path file, BasicFileAttributes attrs) throws IOException { + Files.deleteIfExists(file); + return FileVisitResult.CONTINUE; + } + + @Override + public FileVisitResult postVisitDirectory(Path dir, IOException exception) throws IOException { + Files.deleteIfExists(dir); + return FileVisitResult.CONTINUE; + } + }); + } + } + + 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 + @OperationsPerInvocation(KvinIngestionWorkload.TUPLE_COUNT) + public void putBatch(BenchmarkState state) { + state.putBatch(); + } + + @Benchmark + @OperationsPerInvocation(KvinIngestionWorkload.TUPLE_COUNT) + public void postJson(BenchmarkState state) throws IOException { + state.postJson(); + } + + @Benchmark + @OperationsPerInvocation(KvinIngestionWorkload.TUPLE_COUNT) + public void postCsv(BenchmarkState state) throws IOException { + state.postCsv(); + } + + @Benchmark + @OperationsPerInvocation(KvinIngestionWorkload.TUPLE_COUNT) + public void putCsvDirect(BenchmarkState state) throws IOException { + state.putCsvDirect(); + } + + @Benchmark + @OperationsPerInvocation(KvinIngestionWorkload.TUPLE_COUNT) + public void postCsvSequentialFiles(BenchmarkState state) throws IOException { + state.postCsvSequentialFiles(); + } +} diff --git a/bundles/io.github.linkedfactory.service/src/test/java/io/github/linkedfactory/service/benchmark/KvinIngestionCsvDiagnosticBenchmark.java b/bundles/io.github.linkedfactory.service/src/test/java/io/github/linkedfactory/service/benchmark/KvinIngestionCsvDiagnosticBenchmark.java new file mode 100644 index 00000000..8c0a9c0a --- /dev/null +++ b/bundles/io.github.linkedfactory.service/src/test/java/io/github/linkedfactory/service/benchmark/KvinIngestionCsvDiagnosticBenchmark.java @@ -0,0 +1,291 @@ +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(); + + KommaModule module = ModelPlugin.createModelSetModule(ModelPlugin.class.getClassLoader()); + IModelSetFactory factory = + Guice.createInjector(new ModelSetModule(module)) + .getInstance(IModelSetFactory.class); + modelSet = factory.createModelSet(MODELS.NAMESPACE_URI.appendFragment("MemoryModelSet")); + + Globals.contextModelSet().theDefault().set(VendorJ.vendor(new net.liftweb.common.Full(modelSet))); + sink = new ConsumingKvin(); + service = new BenchmarkService(sink); + } + + @TearDown(Level.Invocation) + public void validateInvocation() { + switch (measuredStage) { + case "consumePrebuilt", "parseCsvAndConsumeTuples", "postCsvParseOnly" -> + org.junit.Assert.assertEquals(KvinIngestionWorkload.TUPLE_COUNT, tupleCount); + case "decodeCsvAndConsumeFields" -> { + org.junit.Assert.assertEquals(EXPECTED_FIELD_COUNT, rowCount); + org.junit.Assert.assertEquals(EXPECTED_FIELD_COUNT * EXPECTED_FIELDS_PER_ROW, fieldCount); + } + default -> throw new AssertionError("Unknown CSV diagnostic stage " + measuredStage); + } + } + + @TearDown(Level.Trial) + public void teardownTrial() { + if (modelSet != null) { + modelSet.dispose(); + modelSet = null; + } + } + + public void consumePrebuilt(Blackhole blackhole) { + measuredStage = "consumePrebuilt"; + tupleCount = 0; + for (KvinTuple tuple : workload.tuples()) { + blackhole.consume(tuple); + tupleCount++; + } + } + + public void decodeCsvAndConsumeFields(Blackhole blackhole) throws IOException { + measuredStage = "decodeCsvAndConsumeFields"; + rowCount = 0; + fieldCount = 0; + CSVParser parser = new CSVParserBuilder() + .withSeparator(',') + .withIgnoreQuotations(true) + .build(); + try (CSVReader reader = new CSVReaderBuilder(new InputStreamReader( + new ByteArrayInputStream(csvPayload), StandardCharsets.UTF_8)) + .withSkipLines(0) + .withCSVParser(parser) + .build()) { + String[] row; + while ((row = readNext(reader)) != null) { + rowCount++; + for (String field : row) { + blackhole.consume(field); + fieldCount++; + } + } + } + } + + public void parseCsvAndConsumeTuples(Blackhole blackhole) throws IOException { + measuredStage = "parseCsvAndConsumeTuples"; + tupleCount = 0; + CsvFormatParser parser = new CsvFormatParser(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..e9187934 --- /dev/null +++ b/bundles/io.github.linkedfactory.service/src/test/java/io/github/linkedfactory/service/benchmark/KvinIngestionWorkload.java @@ -0,0 +1,187 @@ +package io.github.linkedfactory.service.benchmark; + +import io.github.linkedfactory.core.kvin.KvinTuple; +import io.github.linkedfactory.core.kvin.util.JsonFormatWriter; +import net.enilink.commons.iterator.WrappedIterator; +import net.enilink.komma.core.URI; +import net.enilink.komma.core.URIs; + +import java.io.IOException; +import java.io.UncheckedIOException; +import java.nio.charset.StandardCharsets; +import java.util.ArrayList; +import java.util.Collections; +import java.util.Comparator; +import java.util.List; +import java.util.Locale; +import java.util.Random; + +public final class KvinIngestionWorkload { + public static final int ROW_COUNT = 5_000; + public static final int CHANNEL_COUNT = 6; + public static final int TUPLE_COUNT = ROW_COUNT * CHANNEL_COUNT; + public static final int TIMESTAMP_COUNT = 1_000; + public static final int SEQUENCES_PER_TIMESTAMP = 5; + public static final int VARIANT_COUNT = 10; + public static final int CHANNEL_POOL_SIZE = 100; + public static final int PROPERTY_POOL_SIZE = 10; + public static final long BASE_START_TIME = 1_710_000_000_000L; + public static final long TIMESTAMP_STEP = 1_000L; + public static final long TIMESTAMP_WINDOW_SIZE = TIMESTAMP_COUNT * TIMESTAMP_STEP; + public static final long SHUFFLE_SEED = 0x4B56494E_20260721L; + + public static final URI CONTEXT = URIs.createURI("http://iwu.lf.de/ecc4p/models/emag"); + + private static final List CHANNEL_POOL = createUriPool("http://iwu.lf.de/ecc4p/emag/channel-", CHANNEL_POOL_SIZE, 3); + private static final List PROPERTY_POOL = createUriPool("http://iwu.lf.de/ecc4p/property-", PROPERTY_POOL_SIZE, 2); + private static final List SHUFFLED_CHANNELS = shuffledChannels(); + + private final int variantIndex; + private final List items; + private final URI property; + private final long startTime; + private final List tuples; + private final List preseedTuples; + private final byte[] jsonPayload; + private final byte[] csvPayload; + + public KvinIngestionWorkload() { + this(0); + } + + KvinIngestionWorkload(int variantIndex) { + if (variantIndex < 0 || variantIndex >= VARIANT_COUNT) { + throw new IllegalArgumentException("variantIndex must be in [0, " + VARIANT_COUNT + "): " + variantIndex); + } + this.variantIndex = variantIndex; + int firstChannel = variantIndex * CHANNEL_COUNT; + this.items = List.copyOf(SHUFFLED_CHANNELS.subList(firstChannel, firstChannel + CHANNEL_COUNT)); + this.property = PROPERTY_POOL.get(variantIndex); + this.startTime = BASE_START_TIME + variantIndex * TIMESTAMP_WINDOW_SIZE; + this.tuples = createTuples(); + this.preseedTuples = createPreseedTuples(); + this.jsonPayload = createJsonPayload(tuples); + this.csvPayload = createCsvPayload(0, ROW_COUNT); + } + + static List variants() { + List variants = new ArrayList<>(VARIANT_COUNT); + for (int variant = 0; variant < VARIANT_COUNT; variant++) { + variants.add(new KvinIngestionWorkload(variant)); + } + return List.copyOf(variants); + } + + int variantIndex() { + return variantIndex; + } + + List items() { + return items; + } + + URI property() { + return property; + } + + public List tuples() { + return tuples; + } + + public List preseedTuples() { + return preseedTuples; + } + + public byte[] jsonPayload() { + return jsonPayload.clone(); + } + + public byte[] csvPayload() { + return csvPayload.clone(); + } + + public List csvPayloads(int fileCount) { + if (fileCount <= 0 || ROW_COUNT % fileCount != 0) { + throw new IllegalArgumentException("fileCount must divide " + ROW_COUNT + ": " + fileCount); + } + int rowsPerFile = ROW_COUNT / fileCount; + List payloads = new ArrayList<>(fileCount); + for (int file = 0; file < fileCount; file++) { + int startRow = file * rowsPerFile; + payloads.add(createCsvPayload(startRow, startRow + rowsPerFile)); + } + return List.copyOf(payloads); + } + + private List createTuples() { + List result = new ArrayList<>(TUPLE_COUNT); + for (int row = 0; row < ROW_COUNT; row++) { + for (int channel = 0; channel < CHANNEL_COUNT; channel++) { + long time = startTime + (row / SEQUENCES_PER_TIMESTAMP) * TIMESTAMP_STEP; + int seqNr = row % SEQUENCES_PER_TIMESTAMP + 1; + result.add(new KvinTuple(items.get(channel), property, CONTEXT, time, seqNr, value(channel, row))); + } + } + return List.copyOf(result); + } + + private List createPreseedTuples() { + List result = new ArrayList<>(CHANNEL_COUNT); + for (int channel = 0; channel < CHANNEL_COUNT; channel++) { + result.add(new KvinTuple(items.get(channel), property, CONTEXT, startTime - TIMESTAMP_STEP, 0, + value(channel, -1))); + } + return List.copyOf(result); + } + + private static byte[] createJsonPayload(List tuples) { + try { + String json = JsonFormatWriter.toJsonString(WrappedIterator.create(tuples.stream() + .sorted(Comparator.comparing((KvinTuple t) -> t.item.toString()) + .thenComparing(t -> t.property.toString()) + .thenComparingLong(t -> t.time) + .thenComparingInt(t -> t.seqNr)) + .toList().iterator())); + return json.getBytes(StandardCharsets.UTF_8); + } catch (IOException e) { + throw new UncheckedIOException(e); + } + } + + private byte[] createCsvPayload(int startRow, int endRow) { + StringBuilder csv = new StringBuilder((endRow - startRow) * CHANNEL_COUNT * 12); + csv.append("time,seqNr"); + for (URI item : items) { + csv.append(',').append(item).append('@').append(property); + } + csv.append('\n'); + + for (int row = startRow; row < endRow; row++) { + csv.append(startTime + (row / SEQUENCES_PER_TIMESTAMP) * TIMESTAMP_STEP) + .append(',').append(row % SEQUENCES_PER_TIMESTAMP + 1); + for (int channel = 0; channel < CHANNEL_COUNT; channel++) { + csv.append(',').append(value(channel, row)); + } + csv.append('\n'); + } + return csv.toString().getBytes(StandardCharsets.UTF_8); + } + + private static List createUriPool(String prefix, int size, int digits) { + List uris = new ArrayList<>(size); + for (int index = 0; index < size; index++) { + uris.add(URIs.createURI(prefix + String.format(Locale.ROOT, "%0" + digits + "d", index))); + } + return List.copyOf(uris); + } + + private static List shuffledChannels() { + List channels = new ArrayList<>(CHANNEL_POOL); + Collections.shuffle(channels, new Random(SHUFFLE_SEED)); + return List.copyOf(channels); + } + + private static double value(int channel, int row) { + return channel * 100_000.0 + row + 0.25; + } +} diff --git a/bundles/io.github.linkedfactory.service/src/test/java/io/github/linkedfactory/service/benchmark/KvinIngestionWorkloadTest.java b/bundles/io.github.linkedfactory.service/src/test/java/io/github/linkedfactory/service/benchmark/KvinIngestionWorkloadTest.java new file mode 100644 index 00000000..9f5520dc --- /dev/null +++ b/bundles/io.github.linkedfactory.service/src/test/java/io/github/linkedfactory/service/benchmark/KvinIngestionWorkloadTest.java @@ -0,0 +1,156 @@ +package io.github.linkedfactory.service.benchmark; + +import io.github.linkedfactory.core.kvin.KvinTuple; +import io.github.linkedfactory.core.kvin.util.CsvFormatParser; +import io.github.linkedfactory.core.kvin.util.JsonFormatParser; +import net.enilink.commons.iterator.IExtendedIterator; +import net.enilink.komma.core.URI; +import net.enilink.komma.core.URIs; +import org.junit.Test; + +import java.io.ByteArrayInputStream; +import java.nio.charset.StandardCharsets; +import java.util.ArrayList; +import java.util.Comparator; +import java.util.HashMap; +import java.util.HashSet; +import java.util.List; +import java.util.Map; +import java.util.Set; + +import static org.junit.Assert.assertArrayEquals; +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertFalse; +import static org.junit.Assert.assertNotEquals; +import static org.junit.Assert.assertTrue; + +public class KvinIngestionWorkloadTest { + @Test + public void suitesAreDeterministicAndSelectDisjointChannelsAndProperties() { + List first = KvinIngestionWorkload.variants(); + List second = KvinIngestionWorkload.variants(); + + assertEquals(KvinIngestionWorkload.VARIANT_COUNT, first.size()); + assertEquals(first.stream().map(KvinIngestionWorkload::tuples).toList(), + second.stream().map(KvinIngestionWorkload::tuples).toList()); + + Set properties = new HashSet<>(); + Set selectedChannels = new HashSet<>(); + for (int index = 0; index < first.size(); index++) { + KvinIngestionWorkload workload = first.get(index); + KvinIngestionWorkload copy = second.get(index); + assertEquals(index, workload.variantIndex()); + assertEquals(workload.items(), copy.items()); + assertEquals(workload.property(), copy.property()); + assertEquals(workload.preseedTuples(), copy.preseedTuples()); + assertArrayEquals(workload.csvPayload(), copy.csvPayload()); + assertArrayEquals(workload.jsonPayload(), copy.jsonPayload()); + assertEquals(KvinIngestionWorkload.CHANNEL_COUNT, new HashSet<>(workload.items()).size()); + assertTrue("Channels overlap at variant " + index, selectedChannels.addAll(workload.items())); + assertTrue("Property repeated at variant " + index, properties.add(workload.property())); + } + assertEquals(KvinIngestionWorkload.VARIANT_COUNT, properties.size()); + assertEquals(KvinIngestionWorkload.VARIANT_COUNT * KvinIngestionWorkload.CHANNEL_COUNT, + selectedChannels.size()); + assertEquals(first.get(0).tuples(), new KvinIngestionWorkload().tuples()); + } + + @Test + public void everyVariantHasTheExpectedShapeAndDisjointKeysAndWindows() { + Set allKeys = new HashSet<>(); + Set allTimestamps = new HashSet<>(); + + for (KvinIngestionWorkload workload : KvinIngestionWorkload.variants()) { + long expectedStartTime = KvinIngestionWorkload.BASE_START_TIME + + workload.variantIndex() * KvinIngestionWorkload.TIMESTAMP_WINDOW_SIZE; + assertEquals(KvinIngestionWorkload.TUPLE_COUNT, workload.tuples().size()); + assertEquals(KvinIngestionWorkload.CHANNEL_COUNT, workload.preseedTuples().size()); + + Map> sequencesByTimestamp = new HashMap<>(); + Set variantTimestamps = new HashSet<>(); + for (KvinTuple tuple : workload.tuples()) { + assertEquals(workload.property(), tuple.property); + assertEquals(KvinIngestionWorkload.CONTEXT, tuple.context); + assertTrue(tuple.seqNr >= 1 && tuple.seqNr <= KvinIngestionWorkload.SEQUENCES_PER_TIMESTAMP); + assertTrue("Duplicate tuple key", allKeys.add(key(tuple))); + variantTimestamps.add(tuple.time); + sequencesByTimestamp.computeIfAbsent(tuple.time, ignored -> new HashSet<>()).add(tuple.seqNr); + } + + assertEquals(KvinIngestionWorkload.TIMESTAMP_COUNT, variantTimestamps.size()); + assertTrue("Timestamp windows overlap", allTimestamps.stream().noneMatch(variantTimestamps::contains)); + allTimestamps.addAll(variantTimestamps); + assertEquals(KvinIngestionWorkload.TIMESTAMP_COUNT, sequencesByTimestamp.size()); + assertTrue(sequencesByTimestamp.values().stream() + .allMatch(sequences -> sequences.size() == KvinIngestionWorkload.SEQUENCES_PER_TIMESTAMP)); + assertEquals(expectedStartTime, variantTimestamps.stream().mapToLong(Long::longValue).min().orElseThrow()); + assertEquals(expectedStartTime + (KvinIngestionWorkload.TIMESTAMP_COUNT - 1) + * KvinIngestionWorkload.TIMESTAMP_STEP, + variantTimestamps.stream().mapToLong(Long::longValue).max().orElseThrow()); + } + } + + @Test + public void allPayloadFormsNormalizeToEachVariantsCanonicalTupleSet() throws Exception { + for (KvinIngestionWorkload workload : KvinIngestionWorkload.variants()) { + Set expected = new HashSet<>(workload.tuples()); + + List csv = parseCsv(workload.csvPayload()); + assertEquals("CSV tuple order for variant " + workload.variantIndex(), workload.tuples(), csv); + assertEquals(expected, new HashSet<>(csv)); + + List json = parseJson(workload.jsonPayload()); + assertEquals(KvinIngestionWorkload.TUPLE_COUNT, json.size()); + assertEquals(expected, new HashSet<>(json)); + assertJsonOrder(json); + + List partitionedCsv = new ArrayList<>(); + List payloads = workload.csvPayloads(10); + assertEquals(10, payloads.size()); + for (byte[] payload : payloads) { + partitionedCsv.addAll(parseCsv(payload)); + } + assertEquals(KvinIngestionWorkload.TUPLE_COUNT, partitionedCsv.size()); + assertEquals(workload.tuples(), partitionedCsv); + assertEquals(expected, new HashSet<>(partitionedCsv)); + + assertEquals(KvinIngestionWorkload.ROW_COUNT + 1, + new String(workload.csvPayload(), StandardCharsets.UTF_8).split("\\n").length); + assertNotEquals(0, workload.jsonPayload().length); + } + } + + private static List parseCsv(byte[] payload) throws Exception { + List tuples = new ArrayList<>(); + CsvFormatParser parser = new CsvFormatParser(URIs.createURI("http://foo.com/linkedfactory/"), ',', + new ByteArrayInputStream(payload)); + parser.setContext(KvinIngestionWorkload.CONTEXT); + try (IExtendedIterator iterator = parser.parse()) { + while (iterator.hasNext()) { + tuples.add(iterator.next()); + } + } + return tuples; + } + + private static List parseJson(byte[] payload) throws Exception { + return new JsonFormatParser(new ByteArrayInputStream(payload)) + .setContext(KvinIngestionWorkload.CONTEXT) + .parse().toList(); + } + + private static void assertJsonOrder(List tuples) { + Comparator order = Comparator.comparing((KvinTuple tuple) -> tuple.item.toString()) + .thenComparing(tuple -> tuple.property.toString()) + .thenComparingLong(tuple -> tuple.time) + .thenComparingInt(tuple -> tuple.seqNr); + for (int index = 1; index < tuples.size(); index++) { + assertFalse("JSON is out of order at tuple " + index, + order.compare(tuples.get(index - 1), tuples.get(index)) > 0); + } + } + + private static String key(KvinTuple tuple) { + return tuple.context + "|" + tuple.item + "|" + tuple.property + "|" + tuple.time + "|" + tuple.seqNr; + } +} diff --git a/bundles/io.github.linkedfactory.service/src/test/scala/io/github/linkedfactory/service/benchmark/KvinServiceBenchmark.scala b/bundles/io.github.linkedfactory.service/src/test/scala/io/github/linkedfactory/service/benchmark/KvinServiceBenchmark.scala deleted file mode 100644 index faed154e..00000000 --- a/bundles/io.github.linkedfactory.service/src/test/scala/io/github/linkedfactory/service/benchmark/KvinServiceBenchmark.scala +++ /dev/null @@ -1,168 +0,0 @@ -package io.github.linkedfactory.service.benchmark - -import com.google.inject.Guice -import io.github.linkedfactory.core.kvin.leveldb.KvinLevelDb -import io.github.linkedfactory.core.kvin.util.JsonFormatWriter -import io.github.linkedfactory.core.kvin.{Kvin, KvinTuple} -import io.github.linkedfactory.service.{KvinService, MockHttpServletRequest} -import net.enilink.commons.iterator.WrappedIterator -import net.enilink.komma.core.{KommaModule, URI, URIs} -import net.enilink.komma.model._ -import net.enilink.platform.lift.util.Globals -import net.liftweb.common.{Box, Full} -import net.liftweb.http.provider.servlet.HTTPRequestServlet -import net.liftweb.http.{CurrentReq, LiftResponse, Req} -import org.junit.{AfterClass, BeforeClass, Ignore, Test} -import sun.invoke.util.ValueConversions - -import java.io.{File, IOException} -import java.nio.file.attribute.BasicFileAttributes -import java.nio.file.{FileVisitResult, Files, Path, SimpleFileVisitor} -import java.util -import java.util.concurrent.LinkedBlockingQueue -import jakarta.servlet.http.HttpServletRequest -import scala.util.Random -import scala.compiletime.uninitialized - -/** - * Companion object of unit tests for the KVIN service endpoint - */ -object KvinServiceBenchmark { - var modelSet: IModelSet = null - var storeDirectory: File = uninitialized - var store: Kvin = uninitialized - - @BeforeClass - def setup(): Unit = { - // create configuration and a model set factory - val module: KommaModule = ModelPlugin.createModelSetModule(classOf[ModelPlugin].getClassLoader) - val factory: IModelSetFactory = Guice.createInjector(new ModelSetModule(module)).getInstance(classOf[IModelSetFactory]) - - // create a model set with an in-memory repository - modelSet = factory.createModelSet(MODELS.NAMESPACE_URI.appendFragment("MemoryModelSet")) - Globals.contextModelSet.default.set(Full(modelSet)) - - createStore() - } - - @AfterClass - def tearDown(): Unit = { - modelSet.dispose() - modelSet = null - - store.close - store = null - deleteDirectory(storeDirectory.toPath) - } - - def createStore(): Unit = { - storeDirectory = new File("/tmp/leveldb-test-" + System.currentTimeMillis + "-" + Random.nextInt(1000) + "/") - storeDirectory.deleteOnExit - store = new KvinLevelDb(storeDirectory) - } - - def deleteDirectory(dir: Path): Unit = { - // delete store directory - Files.walkFileTree(dir, new SimpleFileVisitor[Path]() { - override def visitFile(file: Path, attrs: BasicFileAttributes): FileVisitResult = { - Files.delete(file) - FileVisitResult.CONTINUE - } - - override def postVisitDirectory(dir: Path, ex: IOException): FileVisitResult = { - Files.delete(dir) - FileVisitResult.CONTINUE - } - }) - } -} - -/** - * Unit tests for the KVIN service endpoint - */ -class KvinServiceBenchmark { - val kvinService = new KvinService("linkedfactory" :: Nil, KvinServiceBenchmark.store) { - override def apply(in: Req): () => Box[LiftResponse] = { - try { - Globals.contextModelSet.vend.map(_.getUnitOfWork.begin) - // S.request is used in Data.pathToURI therefore the request needs to be initialized here - CurrentReq.doWith(in) { - super.apply(in) - } - } finally { - Globals.contextModelSet.vend.map(_.getUnitOfWork.end) - } - } - } - - def kvinRest(req: Req): () => Box[LiftResponse] = { - kvinService(req) - } - - val baseUrl = "http://foo.com/linkedfactory/values" - - def toReq(httpRequest: HttpServletRequest): Req = { - Req(new HTTPRequestServlet(httpRequest, null), Nil, System.nanoTime) - } - - @Test - @Ignore - def postValues(): Unit = { - val valueProperty = URIs.createURI("property:value") - - val seed = 200 - val writeValues = 1000000 - - val benchmarkStart = System.currentTimeMillis - - val startTimeValues = 1478252048736L - val nrs = Array.fill(100)(Random.nextInt(Integer.MAX_VALUE)) - val rand = new Random(seed) - - // decouples client-side serialization and server-side parsing and insertion - val queue = new LinkedBlockingQueue[Option[String]](2) - val inserter = new Thread() { - override def run() : Unit = { - var finished = false - while (!finished) { - queue.take() match { - case None => finished = true - case Some(json) => - // support post request - val postReq = new MockHttpServletRequest(baseUrl) { - method = "POST" - body_=(json, "application/json") - } - kvinRest(toReq(postReq))().map(_.toResponse.code) - } - } - } - } - inserter.start() - - var tuples = new util.ArrayList[KvinTuple]() - var currentTime = startTimeValues - (0 to writeValues).foreach { i => - val randomNr = nrs(rand.nextInt(nrs.length)) - val uri = URIs.createURI("http://linkedfactory.github.io/" + randomNr + "/e3fabrik/rollex/" + randomNr + "/measured-point-1") - val ctx = URIs.createURI("ctx:" + randomNr) - - val value = if (randomNr % 2 == 0) rand.nextGaussian() else rand.nextLong(100000) - - tuples.add(new KvinTuple(uri, valueProperty, ctx, currentTime, value)) - currentTime += rand.nextInt(1000) - - if (i % 10000 == 0) { - println(" at: " + i) - val json = JsonFormatWriter.toJsonString(WrappedIterator.create(tuples.iterator())) - queue.put(Some(json)) - tuples = new util.ArrayList[KvinTuple]() - } - } - queue.put(None) - inserter.join() - - val seconds = (System.currentTimeMillis - benchmarkStart) / 1000.0 - println(s"Wrote $writeValues in %1$$,.2f seconds: %2$$,.2f ops per second".format(seconds, writeValues / seconds)) - } -} \ No newline at end of file diff --git a/docs/benchmarks/kvin-ingestion.md b/docs/benchmarks/kvin-ingestion.md new file mode 100644 index 00000000..983eba45 --- /dev/null +++ b/docs/benchmarks/kvin-ingestion.md @@ -0,0 +1,331 @@ +# KVIN Ingestion Benchmarks + +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. Primary results are reported directly as tuples per second. + +## 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 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 +``` + +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. + +**3. Run attribution diagnostics when needed:** + +```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 + +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 +``` + +**4. Read the primary results** (method, tuples/s, JMH error): + +```sh +jq -r '.[] | [(.benchmark | split(".")[-1]), .primaryMetric.score, .primaryMetric.scoreError, .primaryMetric.scoreUnit] | @tsv' \ + bundles/io.github.linkedfactory.service/target/jmh-primary-run-a.json +``` + +Diagnostic suites deliberately remain single-shot batch timings in `ms/op` and +do not use `OperationsPerInvocation`. + +#### Smoke test only + +> not a valid measurement, just checks that everything starts: + +`-Djmh.warmups=0 -Djmh.measurements=1 -Djmh.measurement.time=1s -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.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 +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 the selected six item/property IDs with six preseed tuples, validates the +persisted set, closes the store, and removes the temporary directory. + +--- + +## Reference + +### Workload + +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 + +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: 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 +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 + +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. + +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. + +| 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: + +- `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 + +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 +```