Working on making the conversion to hdt faster

.gitignore

@@ -58,3 +58,6 @@ data
 wikidata
 qendpoint-store/wdbench-indexes
 wdbench-results
+testing
+indexing
+wdbench-indexes

pom.xml

@@ -63,10 +63,10 @@
 
     <properties>
         <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
-        <java.source.version>17</java.source.version>
-        <java.target.version>17</java.target.version>
-        <maven.compiler.source>17</maven.compiler.source>
-        <maven.compiler.target>17</maven.compiler.target>
+        <java.source.version>21</java.source.version>
+        <java.target.version>21</java.target.version>
+        <maven.compiler.source>21</maven.compiler.source>
+        <maven.compiler.target>21</maven.compiler.target>
     </properties>
 
     <repositories>

qendpoint-backend/pom.xml

@@ -146,6 +146,16 @@
                 <configuration>
                     <source>${java.source.version}</source>
                     <target>${java.target.version}</target>
+                    <compilerArgs>
+<!--                        <arg>&#45;&#45;enable-preview</arg>-->
+<!--                        <arg>&#45;&#45;add-opens=java.base/jdk.internal.misc=ALL-UNNAMED</arg>-->
+                        <arg>--add-exports</arg>
+                        <arg>java.base/jdk.internal.misc=ALL-UNNAMED</arg>
+                        <arg>--add-exports</arg>
+                        <arg>java.base/jdk.internal.util=ALL-UNNAMED</arg>
+                        <arg>--add-modules</arg>
+                        <arg>jdk.incubator.vector</arg>
+                    </compilerArgs>
                 </configuration>
             </plugin>
         </plugins>

qendpoint-backend/src/main/java/com/the_qa_company/qendpoint/controller/Sparql.java

@@ -34,8 +34,8 @@
 import org.springframework.stereotype.Component;
 import org.springframework.web.server.ServerWebInputException;
 
-import javax.annotation.PostConstruct;
-import javax.annotation.PreDestroy;
+import jakarta.annotation.PostConstruct;
+import jakarta.annotation.PreDestroy;
 import java.io.File;
 import java.io.FileOutputStream;
 import java.io.IOException;

qendpoint-core/pom.xml

@@ -17,8 +17,18 @@
                 <groupId>org.apache.maven.plugins</groupId>
                 <artifactId>maven-compiler-plugin</artifactId>
                 <configuration>
-                    <source>17</source>
-                    <target>17</target>
+                    <source>21</source>
+                    <target>21</target>
+                    <compilerArgs>
+<!--                        <arg>&#45;&#45;enable-preview</arg>-->
+<!--                        <arg>&#45;&#45;add-opens=java.base/jdk.internal.misc=ALL-UNNAMED</arg>-->
+                        <arg>--add-exports</arg>
+                        <arg>java.base/jdk.internal.misc=ALL-UNNAMED</arg>
+                        <arg>--add-exports</arg>
+                        <arg>java.base/jdk.internal.util=ALL-UNNAMED</arg>
+                        <arg>--add-modules</arg>
+                        <arg>jdk.incubator.vector</arg>
+                    </compilerArgs>
                 </configuration>
             </plugin>
         </plugins>
@@ -75,7 +85,7 @@
         <dependency>
             <groupId>org.apache.commons</groupId>
             <artifactId>commons-compress</artifactId>
-            <version>1.21</version>
+            <version>1.27.1</version>
         </dependency>
         <dependency>
             <groupId>org.apache.jena</groupId>
@@ -113,5 +123,22 @@
             <artifactId>RoaringBitmap</artifactId>
             <version>${roaringbitmap.version}</version>
         </dependency>
+        <dependency>
+            <groupId>it.unimi.dsi</groupId>
+            <artifactId>fastutil</artifactId>
+            <version>8.5.15</version>
+        </dependency>
+        <dependency>
+            <groupId>org.spf4j</groupId>
+            <artifactId>spf4j-core</artifactId>
+            <version>8.10.0</version>
+            <exclusions>
+                <exclusion>
+                    <groupId>org.apache.avro</groupId>
+                    <artifactId>avro</artifactId>
+                </exclusion>
+            </exclusions>
+        </dependency>
+
     </dependencies>
 </project>

qendpoint-core/src/main/java/com/the_qa_company/qendpoint/core/compact/bitmap/MultiRoaringBitmap.java

@@ -3,13 +3,13 @@
 import com.the_qa_company.qendpoint.core.exceptions.NotImplementedException;
 import com.the_qa_company.qendpoint.core.hdt.HDTVocabulary;
 import com.the_qa_company.qendpoint.core.listener.ProgressListener;
+import org.spf4j.io.BufferedInputStream;
+import org.spf4j.io.BufferedOutputStream;
 import com.the_qa_company.qendpoint.core.util.io.CloseMappedByteBuffer;
 import com.the_qa_company.qendpoint.core.util.io.Closer;
 import com.the_qa_company.qendpoint.core.util.io.IOUtil;
 import org.roaringbitmap.RoaringBitmap;
 
-import java.io.BufferedInputStream;
-import java.io.BufferedOutputStream;
 import java.io.Closeable;
 import java.io.DataInputStream;
 import java.io.DataOutputStream;

qendpoint-core/src/main/java/com/the_qa_company/qendpoint/core/compact/sequence/SequenceLog64Map.java

@@ -25,6 +25,7 @@
 import com.the_qa_company.qendpoint.core.exceptions.NotImplementedException;
 import com.the_qa_company.qendpoint.core.hdt.HDTVocabulary;
 import com.the_qa_company.qendpoint.core.listener.ProgressListener;
+import org.spf4j.io.BufferedInputStream;
 import com.the_qa_company.qendpoint.core.util.BitUtil;
 import com.the_qa_company.qendpoint.core.util.crc.CRC32;
 import com.the_qa_company.qendpoint.core.util.crc.CRC8;
@@ -35,7 +36,6 @@
 import com.the_qa_company.qendpoint.core.util.io.CountInputStream;
 import com.the_qa_company.qendpoint.core.util.io.IOUtil;
 
-import java.io.BufferedInputStream;
 import java.io.Closeable;
 import java.io.File;
 import java.io.FileInputStream;

qendpoint-core/src/main/java/com/the_qa_company/qendpoint/core/dictionary/impl/section/PFCDictionarySectionBig.java

@@ -19,8 +19,6 @@
 
 package com.the_qa_company.qendpoint.core.dictionary.impl.section;
 
-import java.io.BufferedInputStream;
-import java.io.BufferedOutputStream;
 import java.io.ByteArrayOutputStream;
 import java.io.File;
 import java.io.FileInputStream;
@@ -39,6 +37,8 @@
 import com.the_qa_company.qendpoint.core.exceptions.IllegalFormatException;
 import com.the_qa_company.qendpoint.core.listener.ProgressListener;
 import com.the_qa_company.qendpoint.core.options.HDTOptions;
+import org.spf4j.io.BufferedInputStream;
+import org.spf4j.io.BufferedOutputStream;
 import com.the_qa_company.qendpoint.core.util.BitUtil;
 import com.the_qa_company.qendpoint.core.util.Mutable;
 import com.the_qa_company.qendpoint.core.util.crc.CRC32;

qendpoint-core/src/main/java/com/the_qa_company/qendpoint/core/dictionary/impl/section/PFCDictionarySectionMap.java

@@ -25,6 +25,7 @@
 import com.the_qa_company.qendpoint.core.exceptions.IllegalFormatException;
 import com.the_qa_company.qendpoint.core.exceptions.NotImplementedException;
 import com.the_qa_company.qendpoint.core.listener.ProgressListener;
+import org.spf4j.io.BufferedInputStream;
 import com.the_qa_company.qendpoint.core.util.io.BigMappedByteBuffer;
 import com.the_qa_company.qendpoint.core.compact.integer.VByte;
 import com.the_qa_company.qendpoint.core.compact.sequence.Sequence;
@@ -40,7 +41,6 @@
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
-import java.io.BufferedInputStream;
 import java.io.Closeable;
 import java.io.File;
 import java.io.FileInputStream;

qendpoint-core/src/main/java/com/the_qa_company/qendpoint/core/hdt/HDTManagerImpl.java

@@ -20,6 +20,7 @@
 import com.the_qa_company.qendpoint.core.rdf.RDFParserCallback;
 import com.the_qa_company.qendpoint.core.rdf.RDFParserFactory;
 import com.the_qa_company.qendpoint.core.rdf.TripleWriter;
+import org.spf4j.io.BufferedOutputStream;
 import com.the_qa_company.qendpoint.core.triples.TripleString;
 import com.the_qa_company.qendpoint.core.util.BitUtil;
 import com.the_qa_company.qendpoint.core.util.Profiler;
@@ -35,7 +36,6 @@
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
-import java.io.BufferedOutputStream;
 import java.io.File;
 import java.io.IOException;
 import java.io.InputStream;

qendpoint-core/src/main/java/com/the_qa_company/qendpoint/core/hdt/impl/HDTDiskImporter.java

@@ -13,6 +13,7 @@
 import com.the_qa_company.qendpoint.core.hdt.impl.diskimport.TripleCompressionResult;
 import com.the_qa_company.qendpoint.core.header.HeaderPrivate;
 import com.the_qa_company.qendpoint.core.iterator.utils.AsyncIteratorFetcher;
+import com.the_qa_company.qendpoint.core.iterator.utils.AsyncIteratorFetcherUnordered;
 import com.the_qa_company.qendpoint.core.listener.MultiThreadListener;
 import com.the_qa_company.qendpoint.core.listener.ProgressListener;
 import com.the_qa_company.qendpoint.core.options.HDTOptions;
@@ -100,10 +101,14 @@ public HDTDiskImporter(HDTOptions hdtFormat, ProgressListener progressListener,
 			throw new IllegalArgumentException("Number of workers should be positive!");
 		}
 		// maximum size of a chunk
-		chunkSize = hdtFormat.getInt(HDTOptionsKeys.LOADER_DISK_CHUNK_SIZE_KEY, () -> getMaxChunkSize(this.workers));
+		long chunkSize = hdtFormat.getInt(HDTOptionsKeys.LOADER_DISK_CHUNK_SIZE_KEY,
+				() -> getMaxChunkSize(this.workers));
 		if (chunkSize < 0) {
 			throw new IllegalArgumentException("Negative chunk size!");
 		}
+		System.err.println("chunkSize: " + chunkSize);
+		this.chunkSize = ((((chunkSize / 1024 / 1024) / 32) * 32) * 1024 * 1024);
+		System.err.println("this.chunkSize: " + this.chunkSize);
 		long maxFileOpenedLong = hdtFormat.getInt(HDTOptionsKeys.LOADER_DISK_MAX_FILE_OPEN_KEY, 1024);
 		int maxFileOpened;
 		if (maxFileOpenedLong < 0 || maxFileOpenedLong > Integer.MAX_VALUE) {
@@ -178,11 +183,11 @@ public CompressTripleMapper compressDictionary(Iterator<TripleString> iterator)
 			throw new IllegalArgumentException("Dictionary already built! Use another importer instance!");
 		}
 		listener.notifyProgress(0,
-				"Sorting sections with chunk of size: " + StringUtil.humanReadableByteCount(chunkSize, true) + "B with "
-						+ ways + "ways and " + workers + " worker(s)");
-
-		AsyncIteratorFetcher<TripleString> source = new AsyncIteratorFetcher<>(iterator);
+				"Sorting sections with chunk of size: " + StringUtil.humanReadableByteCount(chunkSize, false)
+						+ "iB with " + ways + "ways and " + workers + " worker(s)");
 
+		AsyncIteratorFetcherUnordered<TripleString> source = new AsyncIteratorFetcherUnordered<>(iterator);
+//		AsyncIteratorFetcher<TripleString> source = new AsyncIteratorFetcher<>(iterator);
 		profiler.pushSection("section compression");
 		CompressionResult compressionResult;
 		try {

qendpoint-core/src/main/java/com/the_qa_company/qendpoint/core/hdt/impl/HDTImpl.java

@@ -52,6 +52,8 @@
 import com.the_qa_company.qendpoint.core.options.HDTOptions;
 import com.the_qa_company.qendpoint.core.options.HDTOptionsKeys;
 import com.the_qa_company.qendpoint.core.options.HDTSpecification;
+import org.spf4j.io.BufferedInputStream;
+import org.spf4j.io.BufferedOutputStream;
 import com.the_qa_company.qendpoint.core.triples.DictionaryEntriesDiff;
 import com.the_qa_company.qendpoint.core.triples.IteratorTripleID;
 import com.the_qa_company.qendpoint.core.triples.IteratorTripleString;
@@ -74,8 +76,6 @@
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
-import java.io.BufferedInputStream;
-import java.io.BufferedOutputStream;
 import java.io.File;
 import java.io.FileInputStream;
 import java.io.FileNotFoundException;
@@ -528,7 +528,7 @@ public void loadOrCreateIndex(ProgressListener listener, HDTOptions spec) throws
 
 			// SAVE
 			if (this.hdtFileName != null) {
-				BufferedOutputStream out = null;
+				OutputStream out = null;
 				try {
 					out = new BufferedOutputStream(new FileOutputStream(versionName));
 					ci.clear();

qendpoint-core/src/main/java/com/the_qa_company/qendpoint/core/hdt/impl/WriteHDTImpl.java

@@ -8,13 +8,13 @@
 import com.the_qa_company.qendpoint.core.header.HeaderPrivate;
 import com.the_qa_company.qendpoint.core.listener.ProgressListener;
 import com.the_qa_company.qendpoint.core.options.HDTOptions;
+import org.spf4j.io.BufferedOutputStream;
 import com.the_qa_company.qendpoint.core.triples.IteratorTripleString;
 import com.the_qa_company.qendpoint.core.triples.TriplesPrivate;
 import com.the_qa_company.qendpoint.core.triples.impl.WriteBitmapTriples;
 import com.the_qa_company.qendpoint.core.util.io.CloseSuppressPath;
 import com.the_qa_company.qendpoint.core.util.io.IOUtil;
 
-import java.io.BufferedOutputStream;
 import java.io.File;
 import java.io.IOException;
 import java.io.InputStream;

qendpoint-core/src/main/java/com/the_qa_company/qendpoint/core/hdt/impl/diskimport/SectionCompressor.java

@@ -18,6 +18,7 @@
 import com.the_qa_company.qendpoint.core.util.listener.IntermediateListener;
 import com.the_qa_company.qendpoint.core.util.string.ByteString;
 import com.the_qa_company.qendpoint.core.util.string.CompactString;
+import org.apache.jena.base.Sys;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -51,14 +52,16 @@ public class SectionCompressor implements KWayMerger.KWayMergerImpl<TripleString
 	private final int k;
 	private final boolean debugSleepKwayDict;
 	private final boolean quads;
+	private final long start = System.currentTimeMillis();
 
 	public SectionCompressor(CloseSuppressPath baseFileName, AsyncIteratorFetcher<TripleString> source,
 			MultiThreadListener listener, int bufferSize, long chunkSize, int k, boolean debugSleepKwayDict,
 			boolean quads) {
 		this.source = source;
 		this.listener = listener;
 		this.baseFileName = baseFileName;
-		this.bufferSize = bufferSize;
+		this.bufferSize = (Math.max(4 * 1024 * 1024, bufferSize) / (4 * 1024 * 1024)) * 4 * 1024 * 1024;
+		System.out.println("Buffer size: " + this.bufferSize);
 		this.chunkSize = chunkSize;
 		this.k = k;
 		this.debugSleepKwayDict = debugSleepKwayDict;
@@ -250,7 +253,10 @@ public void createChunk(SizeFetcher<TripleString> fetcher, CloseSuppressPath out
 			}
 
 			if (tripleID % 100_000 == 0) {
-				listener.notifyProgress(10, "reading triples " + tripleID);
+				// use start to measure how many triples are read per second
+				int triplesPerSecond = (int) (tripleID / ((System.currentTimeMillis() - start) / 1000.0));
+
+				listener.notifyProgress(10, "reading triples " + tripleID + " triples per second: " + triplesPerSecond);
 			}
 			// too much ram allowed?
 			if (subjects.size() == Integer.MAX_VALUE - 6) {

qendpoint-core/src/main/java/com/the_qa_company/qendpoint/core/hdt/writer/TripleWriterHDT.java

@@ -1,6 +1,5 @@
 package com.the_qa_company.qendpoint.core.hdt.writer;
 
-import java.io.BufferedOutputStream;
 import java.io.FileOutputStream;
 import java.io.IOException;
 import java.io.OutputStream;
@@ -18,6 +17,7 @@
 import com.the_qa_company.qendpoint.core.listener.ProgressListener;
 import com.the_qa_company.qendpoint.core.options.HDTOptions;
 import com.the_qa_company.qendpoint.core.rdf.TripleWriter;
+import org.spf4j.io.BufferedOutputStream;
 import com.the_qa_company.qendpoint.core.triples.TempTriples;
 import com.the_qa_company.qendpoint.core.triples.TripleString;
 import com.the_qa_company.qendpoint.core.util.StopWatch;
@@ -42,7 +42,7 @@ public TripleWriterHDT(String baseUri, HDTOptions spec, String outFile, boolean
 			this.out = new BufferedOutputStream(
 					new GZIPOutputStream(new BufferedOutputStream(new FileOutputStream(outFile))));
 		} else {
-			this.out = new BufferedOutputStream(new FileOutputStream(outFile));
+			this.out = new BufferedOutputStream(new FileOutputStream(outFile), 4 * 1024 * 1024);
 		}
 		close = true;
 		init();

qendpoint-core/src/main/java/com/the_qa_company/qendpoint/core/hdt/writer/TripleWriterNtriples.java

@@ -1,6 +1,5 @@
 package com.the_qa_company.qendpoint.core.hdt.writer;
 
-import java.io.BufferedOutputStream;
 import java.io.BufferedWriter;
 import java.io.FileOutputStream;
 import java.io.FileWriter;
@@ -11,6 +10,7 @@
 import java.util.zip.GZIPOutputStream;
 
 import com.the_qa_company.qendpoint.core.rdf.TripleWriter;
+import org.spf4j.io.BufferedOutputStream;
 import com.the_qa_company.qendpoint.core.triples.TripleString;
 
 public class TripleWriterNtriples implements TripleWriter {

qendpoint-core/src/main/java/com/the_qa_company/qendpoint/core/iterator/utils/AsyncExceptionIterator.java

@@ -0,0 +1,12 @@
+package com.the_qa_company.qendpoint.core.iterator.utils;
+
+import java.util.concurrent.CompletableFuture;
+
+public /**
+		 * The asynchronous iterator interface. Its nextFuture() returns a
+		 * CompletableFuture that completes with null when the iterator is
+		 * exhausted or an exception if something goes wrong.
+		 */
+interface AsyncExceptionIterator<T, E extends Exception> {
+	CompletableFuture<T> nextFuture();
+}