working on new ID based join iterator

Author: hmottestad

core/sail/lmdb/src/main/java/org/eclipse/rdf4j/sail/lmdb/LmdbDupRecordIterator.java

@@ -54,26 +54,26 @@ RecordIterator get(long[] quadReuse, ByteBuffer minKeyBuf, ByteBuffer maxKeyBuf,
 	/** Size in bytes of one (v3,v4) tuple. */
 	private static final int DUP_PAIR_BYTES = Long.BYTES * 2;
 
-	private final Pool pool;
-	private final DupIndex index;
-	private final int dupDbi;
-	private final long cursor;
+	private final Pool pool = Pool.get();
+	private DupIndex index;
+	private int dupDbi;
+	private long cursor;
 
-	private final Txn txnRef;
+	private Txn txnRef;
 	private long txn; // refreshed on txn version changes
 	private long txnRefVersion;
-	private final StampedLongAdderLockManager txnLockManager;
+	private StampedLongAdderLockManager txnLockManager;
 	private final Thread ownerThread = Thread.currentThread();
 
-	private final MDBVal keyData;
-	private final MDBVal valueData;
+	private MDBVal keyData;
+	private MDBVal valueData;
 
 	/** Reused output buffer required by the RecordIterator API. */
-	private final long[] quad;
+	private long[] quad;
 
 	/** Scalars defining the prefix to scan (subject, predicate). */
-	private final long prefixSubj;
-	private final long prefixPred;
+	private long prefixSubj;
+	private long prefixPred;
 
 	private ByteBuffer prefixKeyBuf;
 
@@ -83,33 +83,67 @@ RecordIterator get(long[] quadReuse, ByteBuffer minKeyBuf, ByteBuffer maxKeyBuf,
 	private int dupLimit;
 
 	private int lastResult;
-	private boolean closed = false;
+	private boolean closed = true;
 
-	private final RecordIterator fallback;
-	private final FallbackSupplier fallbackSupplier;
+	private RecordIterator fallback;
+	private FallbackSupplier fallbackSupplier;
 
 	LmdbDupRecordIterator(DupIndex index, long subj, long pred,
 			boolean explicit, Txn txnRef, FallbackSupplier fallbackSupplier) throws IOException {
+		initialize(index, subj, pred, explicit, txnRef, null, fallbackSupplier);
+	}
+
+	LmdbDupRecordIterator(DupIndex index, long subj, long pred,
+			boolean explicit, Txn txnRef, long[] quadReuse, FallbackSupplier fallbackSupplier) throws IOException {
+		initialize(index, subj, pred, explicit, txnRef, quadReuse, fallbackSupplier);
+	}
+
+	void initialize(DupIndex index, long subj, long pred, boolean explicit, Txn txnRef, long[] quadReuse,
+			FallbackSupplier fallbackSupplier) throws IOException {
+		if (!closed || fallback != null) {
+			throw new IllegalStateException("Cannot initialize LMDB dup iterator while it is open");
+		}
+
 		this.index = index;
+		this.dupDbi = index.getDupDB(explicit);
+		this.txnRef = txnRef;
+		this.txnLockManager = txnRef.lockManager();
+		this.fallbackSupplier = fallbackSupplier;
+		this.fallback = null;
+
+		this.prefixSubj = subj;
+		this.prefixPred = pred;
 
-		// Output buffer (s,p are constant for the life of this iterator)
-		this.quad = new long[4];
+		if (quadReuse != null && quadReuse.length >= 4) {
+			this.quad = quadReuse;
+		} else if (this.quad == null || this.quad.length < 4) {
+			this.quad = new long[4];
+		}
 		this.quad[0] = subj;
 		this.quad[1] = pred;
 		this.quad[2] = -1L;
 		this.quad[3] = -1L;
 
-		this.prefixSubj = subj;
-		this.prefixPred = pred;
+		if (this.keyData == null) {
+			this.keyData = pool.getVal();
+		}
+		if (this.valueData == null) {
+			this.valueData = pool.getVal();
+		}
 
-		this.fallbackSupplier = fallbackSupplier;
+		if (this.prefixKeyBuf == null) {
+			this.prefixKeyBuf = pool.getKeyBuffer();
+		}
+		prefixKeyBuf.clear();
+		Varint.writeUnsigned(prefixKeyBuf, prefixSubj);
+		Varint.writeUnsigned(prefixKeyBuf, prefixPred);
+		prefixKeyBuf.flip();
 
-		this.pool = Pool.get();
-		this.keyData = pool.getVal();
-		this.valueData = pool.getVal();
-		this.dupDbi = index.getDupDB(explicit);
-		this.txnRef = txnRef;
-		this.txnLockManager = txnRef.lockManager();
+		this.dupBuf = null;
+		this.dupPos = 0;
+		this.dupLimit = 0;
+		this.lastResult = MDB_SUCCESS;
+		this.closed = false;
 
 		RecordIterator fallbackIterator = null;
 
@@ -125,13 +159,6 @@ RecordIterator get(long[] quadReuse, ByteBuffer minKeyBuf, ByteBuffer maxKeyBuf,
 
 			cursor = openCursor(txn, dupDbi, txnRef.isReadOnly());
 
-			prefixKeyBuf = pool.getKeyBuffer();
-			prefixKeyBuf.clear();
-			Varint.writeUnsigned(prefixKeyBuf, prefixSubj);
-			Varint.writeUnsigned(prefixKeyBuf, prefixPred);
-			// index.toDupKeyPrefix(prefixKeyBuf, subj, pred, 0, 0);
-			prefixKeyBuf.flip();
-
 			boolean positioned = positionOnPrefix();
 			if (positioned) {
 				positioned = primeDuplicateBlock();
@@ -340,22 +367,35 @@ private void closeInternal(boolean maybeCalledAsync) {
 			}
 			try {
 				if (!closed) {
-					if (txnRef.isReadOnly()) {
-						pool.freeCursor(dupDbi, index, cursor);
-					} else {
-						mdb_cursor_close(cursor);
+					if (cursor != 0L && txnRef != null) {
+						if (txnRef.isReadOnly()) {
+							pool.freeCursor(dupDbi, index, cursor);
+						} else {
+							mdb_cursor_close(cursor);
+						}
+					}
+					cursor = 0L;
+					if (keyData != null) {
+						pool.free(keyData);
+						keyData = null;
+					}
+					if (valueData != null) {
+						pool.free(valueData);
+						valueData = null;
 					}
-					pool.free(keyData);
-					pool.free(valueData);
 					if (prefixKeyBuf != null) {
 						pool.free(prefixKeyBuf);
+						prefixKeyBuf = null;
 					}
 				}
 			} finally {
 				closed = true;
 				if (writeLocked) {
 					txnLockManager.unlockWrite(writeStamp);
 				}
+				txnRef = null;
+				dupBuf = null;
+				dupPos = dupLimit = 0;
 			}
 		}
 	}
@@ -364,9 +404,14 @@ private void closeInternal(boolean maybeCalledAsync) {
 	public void close() {
 		if (fallback != null) {
 			fallback.close();
+			fallback = null;
+			fallbackSupplier = null;
 		} else {
 			closeInternal(true);
+			fallbackSupplier = null;
 		}
+		txnLockManager = null;
+		txnRef = null;
 	}
 
 	private long openCursor(long txn, int dbi, boolean tryReuse) throws IOException {

core/sail/lmdb/src/main/java/org/eclipse/rdf4j/sail/lmdb/LmdbSailStore.java

@@ -1377,9 +1377,8 @@ public RecordIterator getRecordIterator(StatementPattern pattern, BindingSet bin
 
 				ByteBuffer minKeyBuf = keyBuffers != null ? keyBuffers.minKey() : null;
 				ByteBuffer maxKeyBuf = keyBuffers != null ? keyBuffers.maxKey() : null;
-				LmdbRecordIterator reuse = (iteratorReuse instanceof LmdbRecordIterator)
-						? (LmdbRecordIterator) iteratorReuse
-						: null;
+				RecordIterator reuse = (iteratorReuse instanceof LmdbRecordIterator
+						|| iteratorReuse instanceof LmdbDupRecordIterator) ? iteratorReuse : null;
 				return tripleStore.getTriples(txn, subjID, predID, objID, contextID, explicit, minKeyBuf, maxKeyBuf,
 						null, reuse);
 			} catch (IOException e) {
@@ -1438,8 +1437,10 @@ public RecordIterator getRecordIterator(long[] binding, int subjIndex, int predI
 				BindingProjectingIterator projectingReuse = iteratorReuse instanceof BindingProjectingIterator
 						? (BindingProjectingIterator) iteratorReuse
 						: null;
-				LmdbRecordIterator baseReuse = projectingReuse != null ? projectingReuse.getBase()
-						: (iteratorReuse instanceof LmdbRecordIterator ? (LmdbRecordIterator) iteratorReuse : null);
+				RecordIterator baseReuse = projectingReuse != null ? projectingReuse.getReusableBase()
+						: (iteratorReuse instanceof LmdbRecordIterator || iteratorReuse instanceof LmdbDupRecordIterator
+								? iteratorReuse
+								: null);
 
 				RecordIterator raw = tripleStore.getTriples(txn, subjQuery, predQuery, objQuery, ctxQuery, explicit,
 						minKeyBuf, maxKeyBuf, quadReuse, baseReuse);
@@ -1476,8 +1477,11 @@ public void close() {
 
 				BindingProjectingIterator result = projectingReuse != null ? projectingReuse
 						: new BindingProjectingIterator();
-				result.configure(raw, raw instanceof LmdbRecordIterator ? (LmdbRecordIterator) raw : null, binding,
-						subjIndex, predIndex, objIndex, ctxIndex, reuse);
+				RecordIterator reusableBase = (raw instanceof LmdbRecordIterator
+						|| raw instanceof LmdbDupRecordIterator)
+								? raw
+								: null;
+				result.configure(raw, reusableBase, binding, subjIndex, predIndex, objIndex, ctxIndex, reuse);
 				return result;
 			} catch (IOException e) {
 				throw new QueryEvaluationException("Unable to create LMDB record iterator", e);
@@ -1486,19 +1490,19 @@ public void close() {
 
 		private final class BindingProjectingIterator implements RecordIterator {
 			private RecordIterator base;
-			private LmdbRecordIterator lmdbBase;
+			private RecordIterator reusableBase;
 			private long[] binding;
 			private int subjIndex;
 			private int predIndex;
 			private int objIndex;
 			private int ctxIndex;
 			private long[] scratch;
 
-			void configure(RecordIterator base, LmdbRecordIterator lmdbBase, long[] binding, int subjIndex,
+			void configure(RecordIterator base, RecordIterator reusableBase, long[] binding, int subjIndex,
 					int predIndex, int objIndex,
 					int ctxIndex, long[] bindingReuse) {
 				this.base = base;
-				this.lmdbBase = lmdbBase;
+				this.reusableBase = reusableBase;
 				this.binding = binding;
 				this.subjIndex = subjIndex;
 				this.predIndex = predIndex;
@@ -1515,8 +1519,8 @@ void configure(RecordIterator base, LmdbRecordIterator lmdbBase, long[] binding,
 				}
 			}
 
-			LmdbRecordIterator getBase() {
-				return lmdbBase;
+			RecordIterator getReusableBase() {
+				return reusableBase;
 			}
 
 			@Override
@@ -1582,6 +1586,7 @@ public void close() {
 				if (base != null) {
 					base.close();
 					base = null;
+					reusableBase = null;
 				}
 			}
 		}

core/sail/lmdb/src/main/java/org/eclipse/rdf4j/sail/lmdb/TripleStore.java

@@ -613,7 +613,7 @@ public RecordIterator getTriples(Txn txn, long subj, long pred, long obj, long c
 	}
 
 	public RecordIterator getTriples(Txn txn, long subj, long pred, long obj, long context, boolean explicit,
-			ByteBuffer minKeyBuf, ByteBuffer maxKeyBuf, long[] quadReuse, LmdbRecordIterator iteratorReuse)
+			ByteBuffer minKeyBuf, ByteBuffer maxKeyBuf, long[] quadReuse, RecordIterator iteratorReuse)
 			throws IOException {
 		TripleIndex index = getBestIndex(subj, pred, obj, context);
 		// System.out.println("get triples: " + Arrays.asList(subj, pred, obj,context));
@@ -627,15 +627,25 @@ public RecordIterator getTriples(Txn txn, long subj, long pred, long obj, long c
 			return new LmdbRecordIterator(index, null, doRangeSearch, subj, pred, obj, context, explicit, txn, quad,
 					minBuf, maxBuf);
 		};
+		LmdbRecordIterator recordReuse = iteratorReuse instanceof LmdbRecordIterator
+				? (LmdbRecordIterator) iteratorReuse
+				: null;
+		LmdbDupRecordIterator dupReuse = iteratorReuse instanceof LmdbDupRecordIterator
+				? (LmdbDupRecordIterator) iteratorReuse
+				: null;
 		if (dupsortRead && subjectPredicateIndex != null && subj >= 0 && pred >= 0 && obj == -1 && context == -1) {
 			assert context == -1 && obj == -1 : "subject-predicate index can only be used for (s,p,?,?) patterns";
 			// Use SP dup iterator, but union with the standard iterator to guard against any edge cases
 			// in SP storage/retrieval; de-duplicate at the record level.
-			return new LmdbDupRecordIterator(subjectPredicateIndex, subj, pred, explicit, txn,
+			if (dupReuse != null) {
+				dupReuse.initialize(subjectPredicateIndex, subj, pred, explicit, txn, quadReuse, fallbackSupplier);
+				return dupReuse;
+			}
+			return new LmdbDupRecordIterator(subjectPredicateIndex, subj, pred, explicit, txn, quadReuse,
 					fallbackSupplier);
 		}
 		return getTriplesUsingIndex(txn, subj, pred, obj, context, explicit, index, doRangeSearch, fallbackSupplier,
-				minKeyBuf, maxKeyBuf, quadReuse, iteratorReuse);
+				minKeyBuf, maxKeyBuf, quadReuse, recordReuse);
 	}
 
 	boolean hasTriples(boolean explicit) throws IOException {