wip

Author: hmottestad

core/queryalgebra/evaluation/src/main/java/org/eclipse/rdf4j/query/algebra/evaluation/optimizer/MinusOptimizer.java

@@ -236,6 +236,14 @@ public void meet(Var node) {
 					names.add(node.getName());
 				}
 			}
+
+			@Override
+			public void meet(Extension node) {
+				for (ExtensionElem elem : node.getElements()) {
+					names.add(elem.getName());
+				}
+				super.meet(node);
+			}
 		});
 		return names;
 	}

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

@@ -99,6 +99,31 @@ static <T> T readTransaction(long env, long writeTxn, Transaction<T> transaction
 		return ret;
 	}
 
+
+	static <T> T readTransaction(long env, long writeTxn, long id, TransactionWithId<T> transaction) throws IOException {
+		try (MemoryStack stack = stackPush()) {
+			long txn;
+			if (writeTxn == 0) {
+				PointerBuffer pp = stack.mallocPointer(1);
+				E(mdb_txn_begin(env, NULL, MDB_RDONLY, pp));
+				txn = pp.get(0);
+			} else {
+				txn = writeTxn;
+			}
+
+			try {
+				return transaction.exec(stack, txn, id);
+			} finally {
+				if (writeTxn == 0) {
+					mdb_txn_abort(txn);
+				}
+			}
+		}
+
+	}
+
+
+
 	static <T> T transaction(long env, Transaction<T> transaction) throws IOException {
 		T ret;
 		try (MemoryStack stack = stackPush()) {
@@ -202,4 +227,9 @@ interface Transaction<T> {
 		T exec(MemoryStack stack, long txn) throws IOException;
 	}
 
+	@FunctionalInterface
+	interface TransactionWithId<T> {
+		T exec(MemoryStack stack, long txn, long id) throws IOException;
+	}
+
 }

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

@@ -454,20 +454,45 @@ ValueStoreRevision getRevision() {
 		return revision;
 	}
 
+
+
+
+	private final LmdbUtil.TransactionWithId<byte[]> GET_DATA_FN = this::getDataInTxn;
+
 	protected byte[] getData(long id) throws IOException {
-		return readTransaction(env, (stack, txn) -> {
-			MDBVal keyData = MDBVal.calloc(stack);
-			keyData.mv_data(id2data(idBuffer(stack), id).flip());
-			MDBVal valueData = MDBVal.calloc(stack);
-			if (mdb_get(txn, dbi, keyData, valueData) == MDB_SUCCESS) {
-				byte[] valueBytes = new byte[valueData.mv_data().remaining()];
-				valueData.mv_data().get(valueBytes);
-				return valueBytes;
+		return readTransaction(env, id, GET_DATA_FN);
+	}
+
+	<T> T readTransaction(long env, long id, LmdbUtil.TransactionWithId<T> getDataFn) throws IOException {
+		txnLock.readLock().lock();
+		try {
+			if (writeTxn != 0) {
+				return LmdbUtil.readTransaction(env, writeTxn, id, getDataFn);
 			}
+			return threadLocalReadTxn.get().execute(getDataFn, env, id);
+		} finally {
+			txnLock.readLock().unlock();
+		}
+	}
+
+	private byte[] getDataInTxn(MemoryStack stack, long txn, long id) {
+		MDBVal keyData = MDBVal.malloc(stack);
+		ByteBuffer keyBuf = id2data(idBuffer(stack), id);
+		keyBuf.flip();
+		keyData.mv_data(keyBuf);
+
+		MDBVal valueData = MDBVal.malloc(stack);
+		if (mdb_get(txn, dbi, keyData, valueData) != MDB_SUCCESS) {
 			return null;
-		});
+		}
+
+		ByteBuffer src = valueData.mv_data();     // don’t call twice
+		byte[] out = new byte[src.remaining()];
+		src.get(out);
+		return out;
 	}
 
+
 	/**
 	 * Get value from cache by ID.
 	 * <p>
@@ -1412,6 +1437,38 @@ synchronized <T> T execute(Transaction<T> transaction, long env) throws IOExcept
 			}
 		}
 
+		synchronized <T> T execute(LmdbUtil.TransactionWithId<T> transaction, long env, long id) throws IOException {
+			try (MemoryStack stack = MemoryStack.stackPush()) {
+				try {
+					ensureTxn(env);
+					state.depth++;
+					try {
+						return transaction.exec(stack, state.txn, id);
+					} finally {
+						releaseTxn();
+					}
+				} catch (Exception e) {
+					// Retry once
+					try {
+						System.gc();
+						Thread.sleep(1);
+						System.gc();
+						Thread.sleep(1);
+					} catch (InterruptedException ie) {
+						Thread.currentThread().interrupt();
+					}
+
+					ensureTxn(env);
+					state.depth++;
+					try {
+						return transaction.exec(stack, state.txn, id);
+					} finally {
+						releaseTxn();
+					}
+				}
+			}
+		}
+
 		private void ensureTxn(long env) throws IOException {
 			registerIfNeeded();
 

core/sail/lmdb/src/test/java/org/eclipse/rdf4j/sail/lmdb/benchmark/ThemeQueryBenchmark.java

@@ -15,6 +15,7 @@
 
 import java.io.File;
 import java.io.IOException;
+import java.util.List;
 import java.util.concurrent.TimeUnit;
 
 import org.apache.commons.io.FileUtils;
@@ -197,19 +198,24 @@ public void testQueryExplanation() throws IOException {
 		String[] themeNames = paramValues("themeName");
 		for (String themeNameValue : themeNames) {
 			for (String queryIndexValue : queryIndexes) {
-				themeName = themeNameValue;
-				x_queryIndex = Integer.parseInt(queryIndexValue);
-				setup();
-				try (SailRepositoryConnection connection = repository.getConnection()) {
-					String explanation = connection
-							.prepareTupleQuery(query)
-							.explain(Explanation.Level.Executed)
-							.toString();
-					System.out.println("Query Explanation for theme " + themeName + " and query index " + x_queryIndex
-							+ ":\n" + explanation);
-				} finally {
-					tearDown();
+				for (Boolean b : List.of(false, true)) {
+					this.z_useSparqlUo = b;
+					themeName = themeNameValue;
+					x_queryIndex = Integer.parseInt(queryIndexValue);
+					setup();
+					try (SailRepositoryConnection connection = repository.getConnection()) {
+						String explanation = connection
+								.prepareTupleQuery(query)
+								.explain(Explanation.Level.Executed)
+								.toString();
+						System.out
+								.println("Query Explanation for theme " + themeName + " and query index " + x_queryIndex
+										+ " and z_useSparqlUo=" + b + " :\n" + explanation);
+					} finally {
+						tearDown();
+					}
 				}
+				System.out.println("----------------------------------------\n");
 			}
 		}
 	}