refactor: deduplicate bloom filter logic and simplify serialization

Consolidate duplicated buildBloomFilter and leading-component switch
logic from S3SailStore and Compactor into BloomFilter. Replace
hand-rolled byte serialization with ByteBuffer. Merge queryQuads
overloads and eliminate double shouldCompact evaluation in compaction.

Author: odysa

core/sail/s3/src/main/java/org/eclipse/rdf4j/sail/s3/S3SailStore.java

@@ -287,7 +287,7 @@ private void writeParquetFiles(long epoch, List<QuadEntry> allEntries, QuadStats
 				ParquetSchemas.SortOrder sortOrder = ParquetSchemas.SortOrder.fromSuffix(sortSuffix);
 				byte[] parquetData = ParquetFileBuilder.build(sorted, sortOrder);
 
-				BloomFilter bloom = buildBloomFilter(sorted, sortSuffix);
+				BloomFilter bloom = BloomFilter.buildForEntries(sorted, sortSuffix);
 
 				String s3Key = Catalog.dataKey(0, epoch, sortSuffix);
 				objectStore.put(s3Key, parquetData);
@@ -303,32 +303,6 @@ private void writeParquetFiles(long epoch, List<QuadEntry> allEntries, QuadStats
 		CompletableFuture.allOf(futures.toArray(CompletableFuture[]::new)).join();
 	}
 
-	/**
-	 * Builds a bloom filter for the leading component of the given sort order.
-	 */
-	static BloomFilter buildBloomFilter(List<QuadEntry> entries, String sortSuffix) {
-		BloomFilter bloom = new BloomFilter(Math.max(1, entries.size()), 0.01);
-		for (QuadEntry entry : entries) {
-			bloom.add(leadingComponent(entry, sortSuffix));
-		}
-		return bloom;
-	}
-
-	private static long leadingComponent(QuadEntry entry, String sortSuffix) {
-		switch (sortSuffix.charAt(0)) {
-		case 's':
-			return entry.subject;
-		case 'o':
-			return entry.object;
-		case 'c':
-			return entry.context;
-		case 'p':
-			return entry.predicate;
-		default:
-			return entry.subject;
-		}
-	}
-
 	private void persistMetadata(long epoch) {
 		// Save catalog first: if we crash after catalog but before values,
 		// on restart we have new nextValueId but old values — IDs are gaps (safe).
@@ -365,14 +339,14 @@ private void runCompactionIfNeeded() {
 
 		pendingCompaction = CompletableFuture.runAsync(() -> {
 			try {
-				doCompaction();
+				doCompaction(needsL0, needsL1);
 			} catch (Exception e) {
 				logger.error("Background compaction failed", e);
 			}
 		}, compactionExecutor);
 	}
 
-	private void doCompaction() {
+	private void doCompaction(boolean compactL0, boolean compactL1) {
 		List<Compactor.CompactionResult> results = new ArrayList<>();
 
 		// Snapshot file list under synchronization
@@ -382,7 +356,7 @@ private void doCompaction() {
 		}
 
 		// L0→L1 compaction
-		if (compactionPolicy.shouldCompact(files, 0)) {
+		if (compactL0) {
 			List<Catalog.ParquetFileInfo> l0Files = CompactionPolicy.filesAtLevel(files, 0);
 			long compactEpoch = epochCounter.getAndIncrement();
 			results.add(compactor.compact(l0Files, 0, 1, compactEpoch, catalog));
@@ -391,8 +365,8 @@ private void doCompaction() {
 			}
 		}
 
-		// L1→L2 compaction
-		if (compactionPolicy.shouldCompact(files, 1)) {
+		// L1→L2 compaction (re-check after L0 compaction may have produced new L1 files)
+		if (compactL1 || (compactL0 && compactionPolicy.shouldCompact(files, 1))) {
 			List<Catalog.ParquetFileInfo> l1Files = CompactionPolicy.filesAtLevel(files, 1);
 			long compactEpoch = epochCounter.getAndIncrement();
 			results.add(compactor.compact(l1Files, 1, 2, compactEpoch, catalog));
@@ -423,16 +397,8 @@ private boolean hasPersistence() {
 	}
 
 	/**
-	 * Queries quads using the best available source (merged Parquet + MemTable, or MemTable only).
-	 */
-	private Iterator<long[]> queryQuads(long s, long p, long o, long c, boolean explicit) {
-		return hasPersistence()
-				? createMergedIterator(s, p, o, c, explicit, null)
-				: memTable.scan(s, p, o, c, explicit);
-	}
-
-	/**
-	 * Queries quads with a preferred index hint.
+	 * Queries quads using the best available source (merged Parquet + MemTable, or MemTable only). If
+	 * {@code preferredIndex} is non-null, it is used instead of automatic index selection.
 	 */
 	private Iterator<long[]> queryQuads(long s, long p, long o, long c, boolean explicit,
 			QuadIndex preferredIndex) {
@@ -500,7 +466,7 @@ CloseableIteration<? extends Statement> createStatementIterator(
 		ArrayList<CloseableIteration<? extends Statement>> perContextIterList = new ArrayList<>(contextIDList.size());
 
 		for (long contextID : contextIDList) {
-			Iterator<long[]> quads = queryQuads(subjID, predID, objID, contextID, explicit);
+			Iterator<long[]> quads = queryQuads(subjID, predID, objID, contextID, explicit, null);
 			perContextIterList.add(new QuadToStatementIteration(quads, valueStore));
 		}
 
@@ -746,7 +712,7 @@ private long removeStatements(Resource subj, IRI pred, Value obj, boolean explic
 
 				long removeCount = 0;
 				for (long contextId : contextIds) {
-					Iterator<long[]> iter = queryQuads(subjID, predID, objID, contextId, explicit);
+					Iterator<long[]> iter = queryQuads(subjID, predID, objID, contextId, explicit, null);
 
 					// Buffer results before removing to avoid ConcurrentModificationException
 					// when the iterator is backed by the MemTable's own map

core/sail/s3/src/main/java/org/eclipse/rdf4j/sail/s3/storage/BloomFilter.java

@@ -10,7 +10,9 @@
  *******************************************************************************/
 package org.eclipse.rdf4j.sail.s3.storage;
 
+import java.nio.ByteBuffer;
 import java.util.Base64;
+import java.util.List;
 
 /**
  * A simple bit-array bloom filter for long values. Uses two independent hash functions derived from a single
@@ -78,31 +80,81 @@ public boolean mightContain(long value) {
 		return true;
 	}
 
+	/**
+	 * Builds a bloom filter for the leading component of the given sort order.
+	 */
+	public static BloomFilter buildForEntries(List<QuadEntry> entries, String sortSuffix) {
+		BloomFilter bloom = new BloomFilter(Math.max(1, entries.size()), 0.01);
+		for (QuadEntry entry : entries) {
+			bloom.add(leadingComponent(entry, sortSuffix));
+		}
+		return bloom;
+	}
+
+	/**
+	 * Extracts the leading component value from a quad entry based on the sort order suffix.
+	 */
+	static long leadingComponent(QuadEntry entry, String sortSuffix) {
+		switch (sortSuffix.charAt(0)) {
+		case 's':
+			return entry.subject;
+		case 'o':
+			return entry.object;
+		case 'c':
+			return entry.context;
+		case 'p':
+			return entry.predicate;
+		default:
+			return entry.subject;
+		}
+	}
+
+	/**
+	 * Extracts the leading component value from raw quad IDs based on the sort order suffix.
+	 */
+	static long leadingComponent(long s, long p, long o, long c, String sortOrder) {
+		if (sortOrder == null) {
+			return -1;
+		}
+		switch (sortOrder.charAt(0)) {
+		case 's':
+			return s;
+		case 'p':
+			return p;
+		case 'o':
+			return o;
+		case 'c':
+			return c;
+		default:
+			return -1;
+		}
+	}
+
 	/**
 	 * Serializes this bloom filter to a Base64-encoded string for JSON storage.
 	 */
 	public String toBase64() {
 		// Format: [numBits (4 bytes)] [numHashFunctions (4 bytes)] [bits array (8 bytes each)]
-		byte[] data = new byte[8 + bits.length * 8];
-		writeInt(data, 0, numBits);
-		writeInt(data, 4, numHashFunctions);
-		for (int i = 0; i < bits.length; i++) {
-			writeLong(data, 8 + i * 8, bits[i]);
+		ByteBuffer buf = ByteBuffer.allocate(8 + bits.length * 8);
+		buf.putInt(numBits);
+		buf.putInt(numHashFunctions);
+		for (long word : bits) {
+			buf.putLong(word);
 		}
-		return Base64.getEncoder().encodeToString(data);
+		return Base64.getEncoder().encodeToString(buf.array());
 	}
 
 	/**
 	 * Deserializes a bloom filter from a Base64-encoded string.
 	 */
 	public static BloomFilter fromBase64(String encoded) {
-		byte[] data = Base64.getDecoder().decode(encoded);
-		int numBits = readInt(data, 0);
-		int numHash = readInt(data, 4);
-		int arrayLen = (data.length - 8) / 8;
+		ByteBuffer buf = ByteBuffer.wrap(Base64.getDecoder().decode(encoded));
+		int numBits = buf.getInt();
+		int numHash = buf.getInt();
+		int arrayLen = buf.remaining() / 8;
 		long[] bits = new long[arrayLen];
 		for (int i = 0; i < arrayLen; i++) {
-			bits[i] = readLong(data, 8 + i * 8);
+			bits[i] = buf.getLong();
 		}
 		return new BloomFilter(bits, numBits, numHash);
 	}
@@ -125,33 +177,4 @@ private static int optimalNumHashFunctions(int n, int m) {
 		return Math.max(1, (int) Math.round((double) m / n * Math.log(2)));
 	}
 
-	private static void writeInt(byte[] buf, int offset, int value) {
-		buf[offset] = (byte) (value >>> 24);
-		buf[offset + 1] = (byte) (value >>> 16);
-		buf[offset + 2] = (byte) (value >>> 8);
-		buf[offset + 3] = (byte) value;
-	}
-
-	private static void writeLong(byte[] buf, int offset, long value) {
-		buf[offset] = (byte) (value >>> 56);
-		buf[offset + 1] = (byte) (value >>> 48);
-		buf[offset + 2] = (byte) (value >>> 40);
-		buf[offset + 3] = (byte) (value >>> 32);
-		buf[offset + 4] = (byte) (value >>> 24);
-		buf[offset + 5] = (byte) (value >>> 16);
-		buf[offset + 6] = (byte) (value >>> 8);
-		buf[offset + 7] = (byte) value;
-	}
-
-	private static int readInt(byte[] buf, int offset) {
-		return ((buf[offset] & 0xFF) << 24) | ((buf[offset + 1] & 0xFF) << 16)
-				| ((buf[offset + 2] & 0xFF) << 8) | (buf[offset + 3] & 0xFF);
-	}
-
-	private static long readLong(byte[] buf, int offset) {
-		return ((long) (buf[offset] & 0xFF) << 56) | ((long) (buf[offset + 1] & 0xFF) << 48)
-				| ((long) (buf[offset + 2] & 0xFF) << 40) | ((long) (buf[offset + 3] & 0xFF) << 32)
-				| ((long) (buf[offset + 4] & 0xFF) << 24) | ((long) (buf[offset + 5] & 0xFF) << 16)
-				| ((long) (buf[offset + 6] & 0xFF) << 8) | (long) (buf[offset + 7] & 0xFF);
-	}
 }

core/sail/s3/src/main/java/org/eclipse/rdf4j/sail/s3/storage/Catalog.java

@@ -415,28 +415,6 @@ public void setBloomFilter(BloomFilter filter) {
 			this.bloomFilterBase64 = filter != null ? filter.toBase64() : null;
 		}
 
-		/**
-		 * Returns the leading component's filter value for this file's sort order. SPOC → subject, OPSC → object, CSPO
-		 * → context.
-		 */
-		private long getLeadingFilterValue(long s, long p, long o, long c) {
-			if (sortOrder == null) {
-				return -1;
-			}
-			switch (sortOrder.charAt(0)) {
-			case 's':
-				return s;
-			case 'o':
-				return o;
-			case 'c':
-				return c;
-			case 'p':
-				return p;
-			default:
-				return -1;
-			}
-		}
-
 		/**
 		 * Tests whether this file's statistics allow it to contain a quad matching the given pattern. Bound components
 		 * (>= 0) are checked against the file's min/max range; unbound components (< 0) are wildcards. Also checks the
@@ -458,7 +436,7 @@ public boolean mayContain(long s, long p, long o, long c) {
 			// Check bloom filter for the leading component
 			BloomFilter bf = getBloomFilter();
 			if (bf != null) {
-				long leadingVal = getLeadingFilterValue(s, p, o, c);
+				long leadingVal = BloomFilter.leadingComponent(s, p, o, c, sortOrder);
 				if (leadingVal >= 0 && !bf.mightContain(leadingVal)) {
 					return false;
 				}

core/sail/s3/src/main/java/org/eclipse/rdf4j/sail/s3/storage/Compactor.java

@@ -112,8 +112,7 @@ public CompactionResult compact(List<Catalog.ParquetFileInfo> sourceFiles,
 				cache.writeThrough(s3Key, parquetData);
 			}
 
-			// Build bloom filter for the leading component
-			BloomFilter bloom = buildBloomFilter(merged, suffix);
+			BloomFilter bloom = BloomFilter.buildForEntries(merged, suffix);
 
 			QuadStats stats = QuadStats.fromEntries(merged);
 			newFiles.add(new Catalog.ParquetFileInfo(s3Key, targetLevel, suffix, merged.size(),
@@ -159,32 +158,6 @@ private List<QuadEntry> mergeEntries(List<RawEntrySource> sources, QuadIndex qua
 		return new ArrayList<>(deduped.values());
 	}
 
-	private static BloomFilter buildBloomFilter(List<QuadEntry> entries, String sortSuffix) {
-		BloomFilter bloom = new BloomFilter(Math.max(1, entries.size()), 0.01);
-		for (QuadEntry entry : entries) {
-			long val;
-			switch (sortSuffix.charAt(0)) {
-			case 's':
-				val = entry.subject;
-				break;
-			case 'o':
-				val = entry.object;
-				break;
-			case 'c':
-				val = entry.context;
-				break;
-			case 'p':
-				val = entry.predicate;
-				break;
-			default:
-				val = entry.subject;
-				break;
-			}
-			bloom.add(val);
-		}
-		return bloom;
-	}
-
 	private static class CompactKey implements Comparable<CompactKey> {
 		final byte[] key;