feat: GH-5715 persist LMDB value hash codes

Author: hmottestad

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

@@ -59,7 +59,9 @@
 import java.util.Arrays;
 import java.util.Collection;
 import java.util.Collections;
+import java.util.HashMap;
 import java.util.HashSet;
+import java.util.Map;
 import java.util.Optional;
 import java.util.Set;
 import java.util.WeakHashMap;
@@ -200,6 +202,8 @@ class ValueStore extends AbstractValueFactory {
 	 */
 	private long nextId = 1;
 	private boolean freeIdsAvailable;
+	private ValueStoreHashFile hashFile;
+	private final Map<Long, Integer> pendingHashUpdates = new HashMap<>();
 
 	private volatile long nextValueEvictionTime = 0;
 
@@ -220,6 +224,7 @@ class ValueStore extends AbstractValueFactory {
 	private Object[] previousNamespaceEntry;
 
 	private final long valueEvictionInterval;
+	private final boolean valueHashCacheEnabled;
 
 	ValueStore(File dir, LmdbStoreConfig config) throws IOException {
 		this.dir = dir;
@@ -228,6 +233,7 @@ class ValueStore extends AbstractValueFactory {
 		this.autoGrow = config.getAutoGrow();
 		this.mapSize = config.getValueDBSize();
 		this.valueEvictionInterval = config.getValueEvictionInterval();
+		this.valueHashCacheEnabled = config.getValueHashCacheEnabled();
 		open();
 
 		int cacheSize = nextPowerOfTwo(config.getValueCacheSize());
@@ -278,6 +284,18 @@ class ValueStore extends AbstractValueFactory {
 		commit();
 	}
 
+	private void openHashFileQuietly() {
+		hashFile = null;
+		if (!valueHashCacheEnabled) {
+			return;
+		}
+		try {
+			hashFile = new ValueStoreHashFile(dir);
+		} catch (IOException e) {
+			logger.warn("Could not initialize LMDB hash cache", e);
+		}
+	}
+
 	private void logValues() throws IOException {
 		readTransaction(env, (stack, txn) -> {
 			long cursor = 0;
@@ -313,6 +331,7 @@ private void logValues() throws IOException {
 	private void open() throws IOException {
 		// create directory if it not exists
 		dir.mkdirs();
+		openHashFileQuietly();
 
 		try (MemoryStack stack = stackPush()) {
 			PointerBuffer pp = stack.mallocPointer(1);
@@ -404,8 +423,10 @@ private long nextId(byte type) throws IOException {
 					MDBVal keyData = MDBVal.calloc(stack);
 					MDBVal valueData = MDBVal.calloc(stack);
 					if (mdb_cursor_get(cursor, keyData, valueData, MDB_FIRST) == MDB_SUCCESS) {
+						long freedId = data2id(keyData.mv_data());
+						clearStoredHash(freedId);
 						// remove lower 2 type bits
-						long value = data2id(keyData.mv_data()) >> 2;
+						long value = freedId >> 2;
 						// delete entry
 						E(mdb_cursor_del(cursor, 0));
 						return value;
@@ -461,6 +482,51 @@ ValueStoreRevision getRevision() {
 		return revision;
 	}
 
+	int getStoredHash(long id) {
+		Integer pendingHash;
+		synchronized (pendingHashUpdates) {
+			pendingHash = pendingHashUpdates.get(id);
+		}
+		if (pendingHash != null) {
+			return pendingHash;
+		}
+		if (hashFile == null) {
+			return 0;
+		}
+		try {
+			return hashFile.get(id);
+		} catch (IOException e) {
+			resetHashFileQuietly("read", e);
+			return 0;
+		}
+	}
+
+	void storeHash(long id, int hash) {
+		if (id == LmdbValue.UNKNOWN_ID) {
+			return;
+		}
+		if (writeTxn != 0) {
+			synchronized (pendingHashUpdates) {
+				pendingHashUpdates.put(id, hash);
+			}
+			return;
+		}
+		writeHashNow(id, hash);
+	}
+
+	void clearStoredHash(long id) {
+		if (id == LmdbValue.UNKNOWN_ID) {
+			return;
+		}
+		if (writeTxn != 0) {
+			synchronized (pendingHashUpdates) {
+				pendingHashUpdates.put(id, 0);
+			}
+			return;
+		}
+		writeHashNow(id, 0);
+	}
+
 	protected byte[] getData(long id) throws IOException {
 		return readTransaction(env, (stack, txn) -> {
 			MDBVal keyData = MDBVal.calloc(stack);
@@ -590,6 +656,7 @@ public boolean resolveValue(long id, LmdbValue value) {
 		try {
 			byte[] data = getData(id);
 			if (data != null) {
+//				System.out.println(id);
 				data2value(id, data, value);
 				cacheValue(id, value);
 				return true;
@@ -963,6 +1030,7 @@ public long getId(Value value, boolean create) throws IOException {
 
 						valueIDCache.put(nv, id);
 					}
+					storeHashIfAbsent(id, value);
 				}
 
 				return id;
@@ -1174,6 +1242,7 @@ protected void freeUnusedIdsAndValues(MemoryStack stack, long txn, Set<Long> rev
 							E(mdb_put(txn, freeDbi, idVal, emptyVal, 0));
 							// delete id -> value association
 							E(mdb_del(txn, dbi, idVal, null));
+							clearStoredHash(data2id(idVal.mv_data().duplicate()));
 							// delete id and value from unused list
 							E(mdb_cursor_del(unusedIdsCursor, 0));
 
@@ -1228,6 +1297,7 @@ void endTransaction(boolean commit) throws IOException {
 					long stamp = revisionLock.writeLock();
 					try {
 						E(mdb_txn_commit(writeTxn));
+						flushPendingHashUpdates();
 						long revisionId = lazyRevision.getRevisionId();
 						cleaner.register(lazyRevision, () -> {
 							synchronized (unusedRevisionIds) {
@@ -1244,9 +1314,11 @@ void endTransaction(boolean commit) throws IOException {
 					}
 				} else {
 					E(mdb_txn_commit(writeTxn));
+					flushPendingHashUpdates();
 				}
 			} else {
 				mdb_txn_abort(writeTxn);
+				clearPendingHashUpdates();
 			}
 			writeTxn = 0;
 			invalidateRevisionOnCommit = false;
@@ -1298,6 +1370,7 @@ public void clear() throws IOException {
 
 		new File(dir, "data.mdb").delete();
 		new File(dir, "lock.mdb").delete();
+		ValueStoreHashFile.deleteIfPresent(dir);
 
 		clearCaches();
 		open();
@@ -1335,11 +1408,81 @@ private void closeReadTransactions() {
 	public void close() throws IOException {
 
 		if (env != 0) {
+			if (writeTxn == 0) {
+				flushPendingHashUpdates();
+			}
 			closeReadTransactions();
 			endTransaction(false);
 			mdb_env_close(env);
 			env = 0;
 		}
+		if (hashFile != null) {
+			try {
+				hashFile.close();
+			} catch (IOException e) {
+				logger.warn("Could not close LMDB hash cache", e);
+			}
+			hashFile = null;
+		}
+	}
+
+	private void storeHashIfAbsent(long id, Value value) {
+		if (getStoredHash(id) == 0) {
+			storeHash(id, value.hashCode());
+		}
+	}
+
+	private void writeHashNow(long id, int hash) {
+		if (hashFile == null) {
+			return;
+		}
+		try {
+			hashFile.put(id, hash);
+		} catch (IOException e) {
+			resetHashFileQuietly("write", e);
+		}
+	}
+
+	private void flushPendingHashUpdates() {
+		Map<Long, Integer> updates;
+		synchronized (pendingHashUpdates) {
+			if (pendingHashUpdates.isEmpty()) {
+				return;
+			}
+			updates = new HashMap<>(pendingHashUpdates);
+			pendingHashUpdates.clear();
+		}
+
+		for (Map.Entry<Long, Integer> entry : updates.entrySet()) {
+			writeHashNow(entry.getKey(), entry.getValue());
+		}
+	}
+
+	private void clearPendingHashUpdates() {
+		synchronized (pendingHashUpdates) {
+			pendingHashUpdates.clear();
+		}
+	}
+
+	private void resetHashFileQuietly(String operation, IOException cause) {
+		logger.warn("Resetting LMDB hash cache after {} failure", operation, cause);
+		clearPendingHashUpdates();
+		if (!valueHashCacheEnabled) {
+			hashFile = null;
+			return;
+		}
+		ValueStoreHashFile currentHashFile = hashFile;
+		hashFile = null;
+		try {
+			if (currentHashFile != null) {
+				currentHashFile.delete();
+			} else {
+				ValueStoreHashFile.deleteIfPresent(dir);
+			}
+		} catch (IOException deleteFailure) {
+			logger.warn("Could not delete LMDB hash cache", deleteFailure);
+		}
+		openHashFileQuietly();
 	}
 
 	private static final class ReadTxn {