53 increase csv ingest speed

bundles/io.github.linkedfactory.core/src/main/java/io/github/linkedfactory/core/kvin/util/KvinTupleGroupingIterator.java

@@ -0,0 +1,141 @@
+/*
+ * Copyright (c) 2024 Fraunhofer IWU.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ */
+package io.github.linkedfactory.core.kvin.util;
+
+import io.github.linkedfactory.core.kvin.KvinTuple;
+import net.enilink.commons.iterator.IExtendedIterator;
+import net.enilink.commons.iterator.NiceIterator;
+
+import java.util.ArrayDeque;
+import java.util.ArrayList;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+
+/**
+ * Reorders each bounded window of tuples into contiguous series. Series are
+ * emitted in the order in which their identities first occur in the window;
+ * tuples within a series retain their source order.
+ */
+public final class KvinTupleGroupingIterator extends NiceIterator<KvinTuple> {
+	private final IExtendedIterator<KvinTuple> source;
+	private final int windowSize;
+	private final ArrayDeque<KvinTuple> pending = new ArrayDeque<>();
+	private boolean finished;
+	private boolean sourceClosed;
+
+	public KvinTupleGroupingIterator(IExtendedIterator<KvinTuple> source, int windowSize) {
+		if (source == null) {
+			throw new NullPointerException("source");
+		}
+		if (windowSize <= 0) {
+			throw new IllegalArgumentException("windowSize must be positive");
+		}
+		this.source = source;
+		this.windowSize = windowSize;
+	}
+
+	@Override
+	public boolean hasNext() {
+		if (!pending.isEmpty()) {
+			return true;
+		}
+		if (finished) {
+			return false;
+		}
+
+		Map<Series, List<KvinTuple>> groups = new LinkedHashMap<>();
+		try {
+			int count = 0;
+			boolean sourceExhausted = false;
+			while (count < windowSize) {
+				if (!source.hasNext()) {
+					sourceExhausted = true;
+					break;
+				}
+				KvinTuple tuple = source.next();
+				groups.computeIfAbsent(new Series(tuple), key -> new ArrayList<>()).add(tuple);
+				count++;
+			}
+			for (List<KvinTuple> group : groups.values()) {
+				pending.addAll(group);
+			}
+			if (sourceExhausted) {
+				closeSource();
+			}
+			if (count == 0) {
+				finished = true;
+			}
+			return !pending.isEmpty();
+		} catch (RuntimeException error) {
+			finished = true;
+			closeSourceSuppressing(error);
+			throw error;
+		} catch (Error error) {
+			finished = true;
+			closeSourceSuppressing(error);
+			throw error;
+		}
+	}
+
+	@Override
+	public KvinTuple next() {
+		if (!hasNext()) {
+			throw new java.util.NoSuchElementException();
+		}
+		return pending.removeFirst();
+	}
+
+	@Override
+	public void close() {
+		finished = true;
+		pending.clear();
+		closeSource();
+	}
+
+	private void closeSource() {
+		if (!sourceClosed) {
+			sourceClosed = true;
+			source.close();
+		}
+	}
+
+	private void closeSourceSuppressing(Throwable original) {
+		try {
+			closeSource();
+		} catch (Throwable closeError) {
+			original.addSuppressed(closeError);
+		}
+	}
+
+	private static final class Series {
+		private final Object item;
+		private final Object property;
+		private final Object context;
+
+		private Series(KvinTuple tuple) {
+			item = tuple.item;
+			property = tuple.property;
+			context = tuple.context;
+		}
+
+		@Override
+		public boolean equals(Object other) {
+			if (!(other instanceof Series)) {
+				return false;
+			}
+			Series that = (Series) other;
+			return Objects.equals(item, that.item) && Objects.equals(property, that.property)
+					&& Objects.equals(context, that.context);
+		}
+
+		@Override
+		public int hashCode() {
+			return Objects.hash(item, property, context);
+		}
+	}
+}

bundles/io.github.linkedfactory.core/src/main/scala/io/github/linkedfactory/core/kvin/leveldb/KvinLevelDb.scala

@@ -591,13 +591,28 @@ class KvinLevelDb(path: File) extends KvinLevelDbBase with Kvin {
 
     val idsBatch = ids.createWriteBatch()
     var batch = values.createWriteBatch()
+    var lastItem: URI = null
+    var lastProperty: URI = null
+    var lastContext: URI = null
+    var lastPrefix: Array[Byte] = null
+    var lastLock: ReentrantReadWriteLock = null
     activeWrites.incrementAndGet()
     try {
       entries.asScala.foreach { entry => // encode value first to circumvent problems with locks
         val encodedValue = encode(entry.value)
-        val lock = lockFor(entry.item)
+        val samePrefix = lastPrefix != null && entry.item == lastItem && entry.property == lastProperty &&
+          entry.context == lastContext
+        val lock = if (samePrefix) lastLock else lockFor(entry.item)
         readLock(lock) {
-          val prefix = toId(entry.item, entry.property, entry.context, generate = true, idsBatch)
+          val prefix = if (samePrefix) lastPrefix else {
+            val resolved = toId(entry.item, entry.property, entry.context, generate = true, idsBatch)
+            lastItem = entry.item
+            lastProperty = entry.property
+            lastContext = entry.context
+            lastPrefix = resolved
+            lastLock = lock
+            resolved
+          }
           val key = new Array[Byte](prefix.length + Varint.calcLengthUnsigned(entry.time) +
             Varint.calcLengthUnsigned(entry.seqNr))
           val bb = ByteBuffer.wrap(key).order(BYTE_ORDER)
@@ -879,4 +894,4 @@ class KvinLevelDb(path: File) extends KvinLevelDbBase with Kvin {
     executor.shutdown()
     errors.headOption.foreach(e => throw new UncheckedIOException(e))
   }
-}
+}

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,64 @@ public void shouldParseCsvDoubleValues() throws IOException {
 		assertFalse(tuples.hasNext());
 	}
 
+	@Test
+	public void shouldPreserveSparseRowsAndTupleFields() throws IOException {
+		String csv = String.join("\n",
+				"time,seqNr,<urn:item:a>@<urn:property>,<urn:item:b>@<urn:property>",
+				"100,1,10",
+				"101,2,,20");
+		CsvFormatParser parser = new CsvFormatParser(URIs.createURI("urn:base:"), ',',
+				new ByteArrayInputStream(csv.getBytes(StandardCharsets.UTF_8)));
+		List<KvinTuple> tuples = parser.parse().toList();
+
+		assertEquals(3, tuples.size());
+		assertTuple(tuples.get(0), "urn:item:a", 100, 1, 10L);
+		assertTuple(tuples.get(1), "urn:item:a", 101, 2, "");
+		assertTuple(tuples.get(2), "urn:item:b", 101, 2, 20L);
+	}
+
+	@Test
+	public void shouldRejectMalformedTimeAndSequenceNumberAndCloseInput() throws IOException {
+		assertMalformedCsv("time,value\ninvalid,1", "Invalid time format");
+		assertMalformedCsv("time,seqNr,value\n100,invalid,1", "Invalid seqNr format");
+	}
+
+	@Test
+	public void shouldCloseInputAtEndOfFile() throws IOException {
+		TrackingInputStream input = new TrackingInputStream("time,value\n100,1");
+		CsvFormatParser parser = new CsvFormatParser(URIs.createURI("urn:base:"), ',', input);
+		assertEquals(1, parser.parse().toList().size());
+		assertTrue(input.closed);
+	}
+
+	private static void assertMalformedCsv(String csv, String expectedMessage) throws IOException {
+		TrackingInputStream input = new TrackingInputStream(csv);
+		CsvFormatParser parser = new CsvFormatParser(URIs.createURI("urn:base:"), ',', input);
+		RuntimeException error = assertThrows(RuntimeException.class, () -> parser.parse().hasNext());
+		assertTrue(error.getCause().getMessage().contains(expectedMessage));
+		assertTrue(input.closed);
+	}
+
+	private static void assertTuple(KvinTuple tuple, String item, long time, int seqNr, Object value) {
+		assertEquals(URIs.createURI(item), tuple.item);
+		assertEquals(URIs.createURI("urn:property"), tuple.property);
+		assertEquals(time, tuple.time);
+		assertEquals(seqNr, tuple.seqNr);
+		assertEquals(value, tuple.value);
+	}
+
+	private static final class TrackingInputStream extends ByteArrayInputStream {
+		private boolean closed;
+
+		private TrackingInputStream(String content) {
+			super(content.getBytes(StandardCharsets.UTF_8));
+		}
+
+		@Override
+		public void close() throws IOException {
+			closed = true;
+			super.close();
+		}
+	}
+
 }

bundles/io.github.linkedfactory.core/src/test/java/io/github/linkedfactory/core/kvin/util/KvinTupleGroupingIteratorTest.java

@@ -0,0 +1,139 @@
+package io.github.linkedfactory.core.kvin.util;
+
+import io.github.linkedfactory.core.kvin.KvinTuple;
+import net.enilink.commons.iterator.IExtendedIterator;
+import net.enilink.commons.iterator.NiceIterator;
+import net.enilink.commons.iterator.WrappedIterator;
+import net.enilink.komma.core.URI;
+import net.enilink.komma.core.URIs;
+import org.junit.Test;
+
+import java.util.ArrayList;
+import java.util.List;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertSame;
+import static org.junit.Assert.assertTrue;
+
+public class KvinTupleGroupingIteratorTest {
+	private static final URI ITEM_A = URIs.createURI("urn:item:a");
+	private static final URI ITEM_B = URIs.createURI("urn:item:b");
+	private static final URI PROPERTY = URIs.createURI("urn:property");
+	private static final URI PROPERTY_B = URIs.createURI("urn:property:b");
+	private static final URI CONTEXT = URIs.createURI("urn:context");
+	private static final URI CONTEXT_B = URIs.createURI("urn:context:b");
+
+	@Test
+	public void groupsByIdentityWithinBoundedWindows() {
+		KvinTuple a1 = tuple(ITEM_A, 1, 1);
+		KvinTuple b1 = tuple(ITEM_B, 2, 1);
+		KvinTuple a2 = tuple(ITEM_A, 3, 2);
+		KvinTuple b2 = tuple(ITEM_B, 4, 2);
+		KvinTuple a3 = tuple(ITEM_A, 5, 3);
+
+		IExtendedIterator<KvinTuple> grouped = new KvinTupleGroupingIterator(
+				WrappedIterator.create(List.of(a1, b1, a2, b2, a3).iterator()), 4);
+		List<KvinTuple> actual = new ArrayList<>();
+		grouped.forEachRemaining(actual::add);
+
+		assertEquals(List.of(a1, a2, b1, b2, a3), actual);
+		assertFalse(grouped.hasNext());
+	}
+
+	@Test
+	public void closesSourceOnFailureAndExplicitClose() {
+		TrackingIterator source = new TrackingIterator(List.of(tuple(ITEM_A, 1, 1)));
+		KvinTupleGroupingIterator grouped = new KvinTupleGroupingIterator(source, 1);
+		assertTrue(grouped.hasNext());
+		assertSame(source.values.get(0), grouped.next());
+		assertFalse(grouped.hasNext());
+		assertTrue(source.closed);
+		grouped.close();
+		assertTrue(source.closed);
+	}
+
+	@Test
+	public void distinguishesPropertyAndContextInSeriesIdentity() {
+		KvinTuple first = new KvinTuple(ITEM_A, PROPERTY, CONTEXT, 1, 1, "first");
+		KvinTuple otherProperty = new KvinTuple(ITEM_A, PROPERTY_B, CONTEXT, 2, 1, "property");
+		KvinTuple otherContext = new KvinTuple(ITEM_A, PROPERTY, CONTEXT_B, 3, 1, "context");
+		KvinTuple second = new KvinTuple(ITEM_A, PROPERTY, CONTEXT, 4, 2, "second");
+
+		IExtendedIterator<KvinTuple> grouped = new KvinTupleGroupingIterator(
+				WrappedIterator.create(List.of(first, otherProperty, otherContext, second).iterator()), 4);
+		assertEquals(List.of(first, second, otherProperty, otherContext), grouped.toList());
+	}
+
+	@Test
+	public void closesSourceWhenSourceFails() {
+		TrackingIterator source = new TrackingIterator(List.of(tuple(ITEM_A, 1, 1)));
+		source.failure = new IllegalStateException("source failed");
+		KvinTupleGroupingIterator grouped = new KvinTupleGroupingIterator(source, 2);
+		try {
+			grouped.hasNext();
+			throw new AssertionError("expected source failure");
+		} catch (IllegalStateException expected) {
+			assertTrue(source.closed);
+		}
+	}
+
+	@Test
+	public void doesNotMaskSourceFailureWhenCloseFails() {
+		TrackingIterator source = new TrackingIterator(List.of(tuple(ITEM_A, 1, 1)));
+		IllegalStateException sourceFailure = new IllegalStateException("source failed");
+		RuntimeException closeFailure = new RuntimeException("close failed");
+		source.failure = sourceFailure;
+		source.closeFailure = closeFailure;
+		KvinTupleGroupingIterator grouped = new KvinTupleGroupingIterator(source, 2);
+		try {
+			grouped.hasNext();
+			throw new AssertionError("expected source failure");
+		} catch (IllegalStateException actual) {
+			assertSame(sourceFailure, actual);
+			assertEquals(List.of(closeFailure), List.of(actual.getSuppressed()));
+		}
+	}
+
+	@Test(expected = IllegalArgumentException.class)
+	public void rejectsNonPositiveWindow() {
+		new KvinTupleGroupingIterator(WrappedIterator.create(List.<KvinTuple>of().iterator()), 0);
+	}
+
+	private static KvinTuple tuple(URI item, long time, int seqNr) {
+		return new KvinTuple(item, PROPERTY, CONTEXT, time, seqNr, time);
+	}
+
+	private static final class TrackingIterator extends NiceIterator<KvinTuple> {
+		private final List<KvinTuple> values;
+		private int index;
+		private boolean closed;
+		private RuntimeException failure;
+		private RuntimeException closeFailure;
+
+		private TrackingIterator(List<KvinTuple> values) {
+			this.values = values;
+		}
+
+		@Override
+		public boolean hasNext() {
+			if (failure != null && index >= values.size()) {
+				throw failure;
+			}
+			return index < values.size();
+		}
+
+		@Override
+		public KvinTuple next() {
+			return values.get(index++);
+		}
+
+		@Override
+		public void close() {
+			closed = true;
+			if (closeFailure != null) {
+				throw closeFailure;
+			}
+		}
+	}
+}