Skip to content

Commit 406ab92

Browse files
ate47ate47
authored
Merge pull request #629 from hmottestad/faster-indexing-parser-etc
2 parents d116e31 + 9c7f315 commit 406ab92

229 files changed

Lines changed: 9718 additions & 313 deletions

File tree

Some content is hidden

Large Commits have some content hidden by default. Use the searchbox below for content that may be hidden.

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

Lines changed: 100 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -16,10 +16,13 @@
1616
import com.the_qa_company.qendpoint.core.options.HDTOptionsFile;
1717
import com.the_qa_company.qendpoint.core.options.HDTOptionsKeys;
1818
import com.the_qa_company.qendpoint.core.options.HDTSpecification;
19+
import com.the_qa_company.qendpoint.core.rdf.parsers.BlankNodeIdMapper;
20+
import com.the_qa_company.qendpoint.core.rdf.parsers.NTriplesChunkedSource;
1921
import com.the_qa_company.qendpoint.core.rdf.RDFFluxStop;
2022
import com.the_qa_company.qendpoint.core.rdf.RDFParserCallback;
2123
import com.the_qa_company.qendpoint.core.rdf.RDFParserFactory;
2224
import com.the_qa_company.qendpoint.core.rdf.TripleWriter;
25+
import com.the_qa_company.qendpoint.core.iterator.utils.SizedSupplier;
2326
import com.the_qa_company.qendpoint.core.triples.TripleString;
2427
import com.the_qa_company.qendpoint.core.util.BitUtil;
2528
import com.the_qa_company.qendpoint.core.util.Profiler;
@@ -40,12 +43,14 @@
4043
import java.io.IOException;
4144
import java.io.InputStream;
4245
import java.io.OutputStream;
46+
import java.io.UncheckedIOException;
4347
import java.nio.file.Files;
4448
import java.nio.file.NoSuchFileException;
4549
import java.nio.file.Path;
4650
import java.nio.file.StandardOpenOption;
4751
import java.util.Iterator;
4852
import java.util.List;
53+
import java.util.NoSuchElementException;
4954

5055
public class HDTManagerImpl extends HDTManager {
5156
private static final Logger logger = LoggerFactory.getLogger(HDTManagerImpl.class);
@@ -334,13 +339,23 @@ public HDTResult doGenerateHDT(String rdfFileName, String baseURI, RDFNotation r
334339
@Override
335340
public HDTResult doGenerateHDT(InputStream fileStream, String baseURI, RDFNotation rdfNotation,
336341
CompressionType compressionType, HDTOptions hdtFormat, ProgressListener listener) throws IOException {
342+
boolean keepBNode = hdtFormat.getBoolean(HDTOptionsKeys.PARSER_KEEP_BNODE_KEY, true);
343+
if (HDTOptionsKeys.LOADER_TYPE_VALUE_DISK.equals(hdtFormat.get(HDTOptionsKeys.LOADER_TYPE_KEY))
344+
&& (rdfNotation == RDFNotation.NTRIPLES || rdfNotation == RDFNotation.NQUAD)
345+
&& RDFParserFactory.useSimple(hdtFormat)) {
346+
try {
347+
return doGenerateHDTDisk(fileStream, baseURI, rdfNotation, compressionType, hdtFormat, listener);
348+
} catch (ParserException e) {
349+
throw new IOException(e);
350+
}
351+
}
337352
// uncompress the stream if required
338353
fileStream = IOUtil.asUncompressed(fileStream, compressionType);
339354
// create a parser for this rdf stream
340-
RDFParserCallback parser = RDFParserFactory.getParserCallback(rdfNotation);
355+
RDFParserCallback parser = RDFParserFactory.getParserCallback(rdfNotation, hdtFormat);
341356
// read the stream as triples
342357
try (PipedCopyIterator<TripleString> iterator = RDFParserFactory.readAsIterator(parser, fileStream, baseURI,
343-
true, rdfNotation, hdtFormat)) {
358+
keepBNode, rdfNotation, hdtFormat)) {
344359
return doGenerateHDT(iterator, baseURI, hdtFormat, listener);
345360
}
346361
}
@@ -410,10 +425,24 @@ public HDTResult doGenerateHDT(Iterator<TripleString> triples, String baseURI, H
410425
public HDTResult doGenerateHDTDisk(String rdfFileName, String baseURI, RDFNotation rdfNotation,
411426
CompressionType compressionType, HDTOptions hdtFormat, ProgressListener listener)
412427
throws IOException, ParserException {
428+
boolean keepBNode = hdtFormat.getBoolean(HDTOptionsKeys.PARSER_KEEP_BNODE_KEY, true);
413429
if (compressionType == CompressionType.NONE) {
430+
if ((rdfNotation == RDFNotation.NTRIPLES || rdfNotation == RDFNotation.NQUAD)
431+
&& RDFParserFactory.useSimple(hdtFormat)) {
432+
if (!IOUtil.isRemoteURL(rdfFileName) && !"-".equals(rdfFileName)) {
433+
long chunkBudget = resolveDiskChunkBudget(hdtFormat);
434+
try (NTriplesChunkedSource chunked = new NTriplesChunkedSource(Path.of(rdfFileName), rdfNotation,
435+
chunkBudget, createBNodeMapper(keepBNode))) {
436+
return doGenerateHDTDisk0(asIterator(chunked), false, baseURI, hdtFormat, listener);
437+
}
438+
}
439+
try (InputStream stream = IOUtil.getFileInputStream(rdfFileName, false)) {
440+
return doGenerateHDTDisk(stream, baseURI, rdfNotation, compressionType, hdtFormat, listener);
441+
}
442+
}
414443
RDFParserCallback parser = RDFParserFactory.getParserCallback(rdfNotation, hdtFormat);
415444
try (PipedCopyIterator<TripleString> iterator = RDFParserFactory.readAsIterator(parser, rdfFileName,
416-
baseURI, true, rdfNotation, hdtFormat)) {
445+
baseURI, keepBNode, rdfNotation, hdtFormat)) {
417446
return doGenerateHDTDisk0(iterator, true, baseURI, hdtFormat, listener);
418447
}
419448
}
@@ -428,15 +457,82 @@ public HDTResult doGenerateHDTDisk(InputStream fileStream, String baseURI, RDFNo
428457
throws IOException, ParserException {
429458
// uncompress the stream if required
430459
fileStream = IOUtil.asUncompressed(fileStream, compressionType);
460+
boolean keepBNode = hdtFormat.getBoolean(HDTOptionsKeys.PARSER_KEEP_BNODE_KEY, true);
461+
if ((rdfNotation == RDFNotation.NTRIPLES || rdfNotation == RDFNotation.NQUAD)
462+
&& RDFParserFactory.useSimple(hdtFormat)) {
463+
long chunkBudget = resolveDiskChunkBudget(hdtFormat);
464+
try (NTriplesChunkedSource chunked = new NTriplesChunkedSource(fileStream, rdfNotation, chunkBudget,
465+
8L * 1024 * 1024, 8192, createBNodeMapper(keepBNode))) {
466+
return doGenerateHDTDisk0(asIterator(chunked), false, baseURI, hdtFormat, listener);
467+
}
468+
}
431469
// create a parser for this rdf stream
432470
RDFParserCallback parser = RDFParserFactory.getParserCallback(rdfNotation, hdtFormat);
433471
// read the stream as triples
434472
try (PipedCopyIterator<TripleString> iterator = RDFParserFactory.readAsIterator(parser, fileStream, baseURI,
435-
true, rdfNotation, hdtFormat)) {
473+
keepBNode, rdfNotation, hdtFormat)) {
436474
return doGenerateHDTDisk0(iterator, true, baseURI, hdtFormat, listener);
437475
}
438476
}
439477

478+
private static long resolveDiskChunkBudget(HDTOptions hdtFormat) {
479+
return hdtFormat.getInt(HDTOptionsKeys.LOADER_DISK_CHUNK_SIZE_KEY, HDTManagerImpl::getMaxChunkSize);
480+
}
481+
482+
private static BlankNodeIdMapper createBNodeMapper(boolean keepBNode) {
483+
return keepBNode ? null : BlankNodeIdMapper.create();
484+
}
485+
486+
private static Iterator<TripleString> asIterator(NTriplesChunkedSource chunked) {
487+
return new Iterator<>() {
488+
private SizedSupplier<TripleString> currentChunk;
489+
private TripleString next;
490+
private boolean finished;
491+
492+
@Override
493+
public boolean hasNext() {
494+
pullNext();
495+
return next != null;
496+
}
497+
498+
@Override
499+
public TripleString next() {
500+
pullNext();
501+
if (next == null) {
502+
throw new NoSuchElementException();
503+
}
504+
TripleString value = next;
505+
next = null;
506+
return value;
507+
}
508+
509+
private void pullNext() {
510+
if (finished || next != null) {
511+
return;
512+
}
513+
try {
514+
while (true) {
515+
if (currentChunk == null) {
516+
currentChunk = chunked.get();
517+
if (currentChunk == null) {
518+
finished = true;
519+
return;
520+
}
521+
}
522+
TripleString value = currentChunk.get();
523+
if (value != null) {
524+
next = value.tripleToString();
525+
return;
526+
}
527+
currentChunk = null;
528+
}
529+
} catch (IOException e) {
530+
throw new UncheckedIOException(e);
531+
}
532+
}
533+
};
534+
}
535+
440536
/**
441537
* @return a theoretical maximum amount of memory the JVM will attempt to
442538
* use

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

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -53,7 +53,7 @@ public TripleAppender(TempDictionary dict, TempTriples triples, ProgressListener
5353
}
5454

5555
@Override
56-
public void processTriple(TripleString triple, long pos) {
56+
synchronized public void processTriple(TripleString triple, long pos) {
5757
long s = dict.insert(triple.getSubject(), TripleComponentRole.SUBJECT);
5858
long p = dict.insert(triple.getPredicate(), TripleComponentRole.PREDICATE);
5959
long o = dict.insert(triple.getObject(), TripleComponentRole.OBJECT);

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

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -48,7 +48,7 @@ static class DictionaryAppender implements RDFCallback {
4848
}
4949

5050
@Override
51-
public void processTriple(TripleString triple, long pos) {
51+
synchronized public void processTriple(TripleString triple, long pos) {
5252
dict.insert(triple.getSubject(), TripleComponentRole.SUBJECT);
5353
dict.insert(triple.getPredicate(), TripleComponentRole.PREDICATE);
5454
dict.insert(triple.getObject(), TripleComponentRole.OBJECT);
@@ -80,7 +80,7 @@ public TripleAppender2(TempDictionary dict, TempTriples triples, ProgressListene
8080
}
8181

8282
@Override
83-
public void processTriple(TripleString triple, long pos) {
83+
synchronized public void processTriple(TripleString triple, long pos) {
8484
triples.insert(dict.stringToId(triple.getSubject(), TripleComponentRole.SUBJECT),
8585
dict.stringToId(triple.getPredicate(), TripleComponentRole.PREDICATE),
8686
dict.stringToId(triple.getObject(), TripleComponentRole.OBJECT));

‎qendpoint-core/src/main/java/com/the_qa_company/qendpoint/core/hdt/impl/diskindex/DiskIndexSort.java‎

Lines changed: 43 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -9,7 +9,10 @@
99
import com.the_qa_company.qendpoint.core.iterator.utils.AsyncIteratorFetcher;
1010
import com.the_qa_company.qendpoint.core.iterator.utils.ExceptionIterator;
1111
import com.the_qa_company.qendpoint.core.iterator.utils.SizeFetcher;
12+
import com.the_qa_company.qendpoint.core.iterator.utils.SizedSupplier;
13+
import com.the_qa_company.qendpoint.core.util.concurrent.ExceptionSupplier;
1214
import com.the_qa_company.qendpoint.core.util.concurrent.KWayMerger;
15+
import com.the_qa_company.qendpoint.core.util.concurrent.KWayMergerChunked;
1316
import com.the_qa_company.qendpoint.core.util.io.CloseSuppressPath;
1417
import com.the_qa_company.qendpoint.core.util.io.IOUtil;
1518
import com.the_qa_company.qendpoint.core.util.listener.IntermediateListener;
@@ -26,9 +29,10 @@
2629
*
2730
* @author Antoine Willerval
2831
*/
29-
public class DiskIndexSort implements KWayMerger.KWayMergerImpl<Pair, SizeFetcher<Pair>> {
32+
public class DiskIndexSort implements KWayMerger.KWayMergerImpl<Pair, SizedSupplier<Pair>>,
33+
KWayMergerChunked.KWayMergerChunkedImpl<Pair, SizedSupplier<Pair>> {
3034
private final CloseSuppressPath baseFileName;
31-
private final AsyncIteratorFetcher<Pair> source;
35+
private final AsyncIteratorFetcher<Pair> source; // may be null in pull-mode
3236
private final MultiThreadListener listener;
3337
private final int bufferSize;
3438
private final long chunkSize;
@@ -47,9 +51,15 @@ public DiskIndexSort(CloseSuppressPath baseFileName, AsyncIteratorFetcher<Pair>
4751
this.comparator = comparator;
4852
}
4953

54+
public DiskIndexSort(CloseSuppressPath baseFileName, MultiThreadListener listener, int bufferSize, long chunkSize,
55+
int k, Comparator<Pair> comparator) {
56+
this(baseFileName, null, listener, bufferSize, chunkSize, k, comparator);
57+
}
58+
5059
@Override
51-
public void createChunk(SizeFetcher<Pair> flux, CloseSuppressPath output) throws KWayMerger.KWayMergerException {
52-
ParallelSortableArrayList<Pair> pairs = new ParallelSortableArrayList<>(Pair[].class);
60+
public void createChunk(SizedSupplier<Pair> flux, CloseSuppressPath output) throws KWayMerger.KWayMergerException {
61+
int expectedCapacity = (int) Math.min(Integer.MAX_VALUE - 5L, Math.max(16L, chunkSize / (3L * Long.BYTES)));
62+
ParallelSortableArrayList<Pair> pairs = new ParallelSortableArrayList<>(Pair[].class, expectedCapacity);
5363

5464
Pair pair;
5565
// loading the pairs
@@ -123,7 +133,7 @@ public void mergeChunks(List<CloseSuppressPath> inputs, CloseSuppressPath output
123133
}
124134

125135
@Override
126-
public SizeFetcher<Pair> newStopFlux(Supplier<Pair> flux) {
136+
public SizedSupplier<Pair> newStopFlux(Supplier<Pair> flux) {
127137
return SizeFetcher.of(flux, p -> 3 * Long.BYTES, chunkSize);
128138
}
129139

@@ -139,9 +149,12 @@ public SizeFetcher<Pair> newStopFlux(Supplier<Pair> flux) {
139149
*/
140150
public ExceptionIterator<Pair, IOException> sort(int workers)
141151
throws InterruptedException, IOException, KWayMerger.KWayMergerException {
152+
if (source == null) {
153+
throw new IllegalStateException("sort(workers) requires a source; use sortPull(...) instead.");
154+
}
142155
listener.notifyProgress(0, "Pair sort asked in " + baseFileName.toAbsolutePath());
143156
// force to create the first file
144-
KWayMerger<Pair, SizeFetcher<Pair>> merger = new KWayMerger<>(baseFileName, source, this,
157+
KWayMerger<Pair, SizedSupplier<Pair>> merger = new KWayMerger<>(baseFileName, source, this,
145158
Math.max(1, workers - 1), k);
146159
merger.start();
147160
// wait for the workers to merge the sections and create the triples
@@ -162,4 +175,28 @@ public void close() throws IOException {
162175
};
163176
}
164177

178+
public ExceptionIterator<Pair, IOException> sortPull(int workers,
179+
ExceptionSupplier<SizedSupplier<Pair>, IOException> chunkSupplier)
180+
throws InterruptedException, IOException, KWayMerger.KWayMergerException {
181+
listener.notifyProgress(0, "Pair sort asked in " + baseFileName.toAbsolutePath());
182+
KWayMergerChunked<Pair, SizedSupplier<Pair>> merger = new KWayMergerChunked<>(baseFileName, chunkSupplier, this,
183+
Math.max(1, workers - 1), k);
184+
merger.start();
185+
Optional<CloseSuppressPath> sections = merger.waitResult();
186+
if (sections.isEmpty()) {
187+
return ExceptionIterator.empty();
188+
}
189+
CloseSuppressPath path = sections.get();
190+
return new PairReader(path.openInputStream(bufferSize)) {
191+
@Override
192+
public void close() throws IOException {
193+
try {
194+
super.close();
195+
} finally {
196+
IOUtil.closeObject(path);
197+
}
198+
}
199+
};
200+
}
201+
165202
}

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

Lines changed: 8 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -1,9 +1,7 @@
11
package com.the_qa_company.qendpoint.core.iterator.utils;
22

33
import com.the_qa_company.qendpoint.core.triples.TripleString;
4-
import com.the_qa_company.qendpoint.core.util.string.ByteString;
54

6-
import java.io.IOException;
75
import java.util.Iterator;
86

97
/**
@@ -16,11 +14,15 @@
1614
*/
1715
public class FileTripleIterator extends FileChunkIterator<TripleString> {
1816
public static long estimateSize(TripleString tripleString) {
19-
try {
20-
return ByteString.of(tripleString.asNtriple()).getBuffer().length;
21-
} catch (IOException e) {
22-
throw new RuntimeException("Can't estimate the size of the triple " + tripleString, e);
17+
// Approximation used for chunk sizing: cheap and monotonic-ish.
18+
long size = tripleString.getSubject().length() + tripleString.getPredicate().length()
19+
+ tripleString.getObject().length();
20+
CharSequence graph = tripleString.getGraph();
21+
if (graph != null && !graph.isEmpty()) {
22+
size += graph.length();
2323
}
24+
// Rough overhead for separators and N-Triples syntax.
25+
return size + 16;
2426
}
2527

2628
/**

0 commit comments

Comments
 (0)