fix: thread safety, crash safety, and code cleanup for S3 SAIL

- Catalog: volatile copy-on-write for thread-safe concurrent reads
- S3SailStore: save catalog before values for crash-safe ordering;
  persist namespaces/values even when memTable is empty;
  delete old compaction files only after catalog is saved
- QuadIndex: fix context=0 treated as wildcard in range scans
- QuadStats: filter tombstones from stats computation
- FileSystemObjectStore: atomic writes via temp-file-then-rename
- L2DiskCache: volatile lastAccessNanos, synchronized eviction
- Remove dead config fields (quadIndexes, blockSize, valueCacheSize,
  valueIdCacheSize) from S3StoreConfig and S3StoreSchema
- Unify ALL_INDEXES with SortOrder.values() single source of truth
- Fix inline FQNs and wildcard imports across main and test sources
- Fix stale javadocs in CompactionPolicy, MemTable, S3SailDataset

Author: odysa

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

@@ -14,6 +14,7 @@
 import java.nio.file.Path;
 import java.util.ArrayList;
 import java.util.Comparator;
+import java.util.HashSet;
 import java.util.Iterator;
 import java.util.List;
 import java.util.Map;
@@ -55,6 +56,7 @@
 import org.eclipse.rdf4j.sail.s3.storage.ParquetFileBuilder;
 import org.eclipse.rdf4j.sail.s3.storage.ParquetQuadSource;
 import org.eclipse.rdf4j.sail.s3.storage.ParquetSchemas;
+import org.eclipse.rdf4j.sail.s3.storage.QuadEntry;
 import org.eclipse.rdf4j.sail.s3.storage.QuadIndex;
 import org.eclipse.rdf4j.sail.s3.storage.QuadStats;
 import org.eclipse.rdf4j.sail.s3.storage.RawEntrySource;
@@ -79,12 +81,19 @@
  */
 class S3SailStore implements SailStore {
 
-	final Logger logger = LoggerFactory.getLogger(S3SailStore.class);
+	private static final Logger logger = LoggerFactory.getLogger(S3SailStore.class);
 
 	private static final QuadIndex SPOC_INDEX = new QuadIndex("spoc");
-	private static final QuadIndex OPSC_INDEX = new QuadIndex("opsc");
-	private static final QuadIndex CSPO_INDEX = new QuadIndex("cspo");
-	private static final List<QuadIndex> ALL_INDEXES = List.of(SPOC_INDEX, OPSC_INDEX, CSPO_INDEX);
+	private static final List<QuadIndex> ALL_INDEXES;
+
+	static {
+		ParquetSchemas.SortOrder[] orders = ParquetSchemas.SortOrder.values();
+		List<QuadIndex> indexes = new ArrayList<>(orders.length);
+		for (ParquetSchemas.SortOrder order : orders) {
+			indexes.add(new QuadIndex(order.suffix()));
+		}
+		ALL_INDEXES = List.copyOf(indexes);
+	}
 
 	private static final int DEFAULT_ROW_GROUP_SIZE = 8 * 1024 * 1024; // 8 MiB
 	private static final int DEFAULT_PAGE_SIZE = 64 * 1024; // 64 KiB
@@ -218,18 +227,24 @@ private void flushToObjectStore() {
 			return;
 		}
 
+		// Always persist namespaces and values (they may have changed without any quad writes)
+		valueStore.serialize(objectStore);
+		namespaceStore.serialize(objectStore, jsonMapper);
+
+		if (memTable.size() == 0) {
+			return; // no quads to flush — avoid wasting epoch numbers and S3 writes
+		}
+
 		long epoch = epochCounter.getAndIncrement();
 
-		if (memTable.size() > 0) {
-			// Freeze active MemTable and swap in fresh one
-			MemTable frozen = memTable;
-			frozen.freeze();
-			memTable = new MemTable(SPOC_INDEX);
+		// Freeze active MemTable and swap in fresh one
+		MemTable frozen = memTable;
+		frozen.freeze();
+		memTable = new MemTable(SPOC_INDEX);
 
-			List<long[]> allQuads = collectQuads(frozen);
-			QuadStats stats = QuadStats.fromQuads(allQuads);
-			writeParquetFiles(epoch, allQuads, stats);
-		}
+		List<long[]> allQuads = collectQuads(frozen);
+		QuadStats stats = QuadStats.fromQuads(allQuads);
+		writeParquetFiles(epoch, allQuads, stats);
 
 		persistMetadata(epoch);
 		runCompactionIfNeeded();
@@ -254,7 +269,7 @@ private static List<long[]> collectQuads(MemTable frozen) {
 	private void writeParquetFiles(long epoch, List<long[]> allQuads, QuadStats stats) {
 		for (QuadIndex sortIndex : ALL_INDEXES) {
 			String sortSuffix = sortIndex.getFieldSeqString();
-			List<ParquetFileBuilder.QuadEntry> sorted = sortQuadEntries(allQuads, sortIndex);
+			List<QuadEntry> sorted = sortQuadEntries(allQuads, sortIndex);
 
 			ParquetSchemas.SortOrder sortOrder = ParquetSchemas.SortOrder.fromSuffix(sortSuffix);
 			byte[] parquetData = ParquetFileBuilder.build(sorted, ParquetSchemas.QUAD_SCHEMA,
@@ -273,17 +288,20 @@ private void writeParquetFiles(long epoch, List<long[]> allQuads, QuadStats stat
 	}
 
 	private void persistMetadata(long epoch) {
-		valueStore.serialize(objectStore);
-		namespaceStore.serialize(objectStore, jsonMapper);
+		// Save catalog first: if we crash after catalog but before values,
+		// on restart we have new nextValueId but old values — IDs are gaps (safe).
+		// The reverse (values first, catalog second) risks ID reuse on crash (corruption).
 		catalog.setNextValueId(valueStore.getNextId());
 		catalog.setEpoch(epoch);
 		catalog.save(objectStore, jsonMapper, epoch);
+		valueStore.serialize(objectStore);
+		namespaceStore.serialize(objectStore, jsonMapper);
 	}
 
 	/**
 	 * Sorts quad entries according to the given sort index.
 	 */
-	private static List<ParquetFileBuilder.QuadEntry> sortQuadEntries(List<long[]> quads, QuadIndex sortIndex) {
+	private static List<QuadEntry> sortQuadEntries(List<long[]> quads, QuadIndex sortIndex) {
 		List<long[]> sorted = new ArrayList<>(quads);
 		String seq = sortIndex.getFieldSeqString();
 		sorted.sort((a, b) -> {
@@ -297,9 +315,9 @@ private static List<ParquetFileBuilder.QuadEntry> sortQuadEntries(List<long[]> q
 			return 0;
 		});
 
-		List<ParquetFileBuilder.QuadEntry> result = new ArrayList<>(sorted.size());
+		List<QuadEntry> result = new ArrayList<>(sorted.size());
 		for (long[] q : sorted) {
-			result.add(new ParquetFileBuilder.QuadEntry(q[0], q[1], q[2], q[3], (byte) q[4]));
+			result.add(new QuadEntry(q[0], q[1], q[2], q[3], (byte) q[4]));
 		}
 		return result;
 	}
@@ -312,27 +330,40 @@ private void runCompactionIfNeeded() {
 			return;
 		}
 
+		List<Compactor.CompactionResult> results = new ArrayList<>();
 		List<Catalog.ParquetFileInfo> files = catalog.getFiles();
 
 		// L0→L1 compaction
 		if (compactionPolicy.shouldCompact(files, 0)) {
 			List<Catalog.ParquetFileInfo> l0Files = CompactionPolicy.filesAtLevel(files, 0);
 			long compactEpoch = epochCounter.getAndIncrement();
-			compactor.compact(l0Files, 0, 1, compactEpoch, catalog);
+			results.add(compactor.compact(l0Files, 0, 1, compactEpoch, catalog));
 			files = catalog.getFiles();
 		}
 
 		// L1→L2 compaction
 		if (compactionPolicy.shouldCompact(files, 1)) {
 			List<Catalog.ParquetFileInfo> l1Files = CompactionPolicy.filesAtLevel(files, 1);
 			long compactEpoch = epochCounter.getAndIncrement();
-			compactor.compact(l1Files, 1, 2, compactEpoch, catalog);
+			results.add(compactor.compact(l1Files, 1, 2, compactEpoch, catalog));
 		}
 
-		// Save catalog after compaction
-		long epoch = epochCounter.getAndIncrement();
-		catalog.setEpoch(epoch);
-		catalog.save(objectStore, jsonMapper, epoch);
+		if (!results.isEmpty()) {
+			// Save catalog BEFORE deleting old files — crash-safe ordering
+			long epoch = epochCounter.getAndIncrement();
+			catalog.setEpoch(epoch);
+			catalog.save(objectStore, jsonMapper, epoch);
+
+			// Now safe to delete old files
+			for (Compactor.CompactionResult result : results) {
+				for (String key : result.getDeletedKeys()) {
+					objectStore.delete(key);
+					if (cache != null) {
+						cache.invalidate(key);
+					}
+				}
+			}
+		}
 	}
 
 	private boolean hasPersistence() {
@@ -575,24 +606,9 @@ public void approve(Resource subj, IRI pred, Value obj, Resource ctx) throws Sai
 		public void approveAll(Set<Statement> approved, Set<Resource> approvedContexts) {
 			sinkStoreAccessLock.lock();
 			try {
-				for (Statement statement : approved) {
-					Resource subj = statement.getSubject();
-					IRI pred = statement.getPredicate();
-					Value obj = statement.getObject();
-					Resource context = statement.getContext();
-
-					long s = valueStore.storeValue(subj);
-					long p = valueStore.storeValue(pred);
-					long o = valueStore.storeValue(obj);
-					long c = context == null ? 0 : valueStore.storeValue(context);
-
-					if (!explicit) {
-						mayHaveInferred = true;
-					}
-
-					memTable.put(s, p, o, c, explicit);
+				for (Statement st : approved) {
+					storeQuad(st.getSubject(), st.getPredicate(), st.getObject(), explicit, st.getContext());
 				}
-
 				// Size-triggered flush
 				if (objectStore != null && memTable.approximateSizeInBytes() >= memTableFlushSize) {
 					flushToObjectStore();
@@ -621,21 +637,23 @@ public boolean supportsDeprecateByQuery() {
 		private void addStatement(Resource subj, IRI pred, Value obj, boolean explicit, Resource context) {
 			sinkStoreAccessLock.lock();
 			try {
-				long s = valueStore.storeValue(subj);
-				long p = valueStore.storeValue(pred);
-				long o = valueStore.storeValue(obj);
-				long c = context == null ? 0 : valueStore.storeValue(context);
-
-				if (!explicit) {
-					mayHaveInferred = true;
-				}
-
-				memTable.put(s, p, o, c, explicit);
+				storeQuad(subj, pred, obj, explicit, context);
 			} finally {
 				sinkStoreAccessLock.unlock();
 			}
 		}
 
+		private void storeQuad(Resource subj, IRI pred, Value obj, boolean explicit, Resource context) {
+			long s = valueStore.storeValue(subj);
+			long p = valueStore.storeValue(pred);
+			long o = valueStore.storeValue(obj);
+			long c = context == null ? 0 : valueStore.storeValue(context);
+			if (!explicit) {
+				mayHaveInferred = true;
+			}
+			memTable.put(s, p, o, c, explicit);
+		}
+
 		private long removeStatements(Resource subj, IRI pred, Value obj, boolean explicit, Resource... contexts) {
 			Objects.requireNonNull(contexts,
 					"contexts argument may not be null; either the value should be cast to Resource or an empty array should be supplied");
@@ -735,7 +753,7 @@ protected Resource convert(long[] quad) {
 							return val instanceof Resource ? (Resource) val : null;
 						}
 					}) {
-				private final java.util.Set<Resource> seen = new java.util.HashSet<>();
+				private final Set<Resource> seen = new HashSet<>();
 
 				@Override
 				protected boolean accept(Resource ctx) {
@@ -755,10 +773,13 @@ public CloseableIteration<? extends Statement> getStatements(Resource subj, IRI
 			return createStatementIterator(subj, pred, obj, explicit, contexts);
 		}
 
+		/**
+		 * @throws UnsupportedOperationException always — ordered iteration is not supported
+		 */
 		@Override
 		public CloseableIteration<? extends Statement> getStatements(StatementOrder statementOrder, Resource subj,
 				IRI pred, Value obj, Resource... contexts) throws SailException {
-			throw new UnsupportedOperationException("Not implemented yet");
+			throw new UnsupportedOperationException("Ordered iteration is not supported by S3Store");
 		}
 
 		@Override
@@ -796,6 +817,11 @@ public Statement next() {
 			Resource subj = (Resource) valueStore.getValue(quad[0]);
 			IRI pred = (IRI) valueStore.getValue(quad[1]);
 			Value obj = valueStore.getValue(quad[2]);
+			if (subj == null || pred == null || obj == null) {
+				// Value ID exists in Parquet but not in value store — can happen after
+				// crash recovery when catalog was saved but value file was not.
+				return null;
+			}
 			Resource ctx = quad[3] == 0 ? null : (Resource) valueStore.getValue(quad[3]);
 			return valueStore.createStatement(subj, pred, obj, ctx);
 		}

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

@@ -18,6 +18,7 @@
 import org.eclipse.rdf4j.common.transaction.IsolationLevel;
 import org.eclipse.rdf4j.common.transaction.IsolationLevels;
 import org.eclipse.rdf4j.model.ValueFactory;
+import org.eclipse.rdf4j.model.impl.LinkedHashModel;
 import org.eclipse.rdf4j.query.algebra.evaluation.EvaluationStrategyFactory;
 import org.eclipse.rdf4j.query.algebra.evaluation.federation.FederatedServiceResolver;
 import org.eclipse.rdf4j.query.algebra.evaluation.federation.FederatedServiceResolverClient;
@@ -135,7 +136,7 @@ protected void initializeInternal() throws SailException {
 
 		try {
 			backingStore = new S3SailStore(config);
-			this.store = new SnapshotSailStore(backingStore, () -> new org.eclipse.rdf4j.model.impl.LinkedHashModel()) {
+			this.store = new SnapshotSailStore(backingStore, LinkedHashModel::new) {
 
 				@Override
 				public SailSource getExplicitSailSource() {

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

@@ -106,60 +106,26 @@ void serialize(ObjectStore objectStore) {
 		try {
 			ByteArrayOutputStream baos = new ByteArrayOutputStream();
 			DataOutputStream out = new DataOutputStream(baos);
+			ByteBuffer buf = ByteBuffer.allocate(9); // reusable scratch for varints
 
-			int count = idToValue.size();
-			// Write count as varint
-			ByteBuffer countBuf = ByteBuffer.allocate(9);
-			Varint.writeUnsigned(countBuf, count);
-			out.write(countBuf.array(), 0, countBuf.position());
+			writeVarint(out, buf, idToValue.size());
 
 			for (Map.Entry<Long, Value> entry : idToValue.entrySet()) {
-				long id = entry.getKey();
+				writeVarint(out, buf, entry.getKey());
 				Value val = entry.getValue();
 
-				// Write id as varint
-				ByteBuffer idBuf = ByteBuffer.allocate(9);
-				Varint.writeUnsigned(idBuf, id);
-				out.write(idBuf.array(), 0, idBuf.position());
-
 				if (val instanceof IRI) {
-					out.writeByte(0); // type = IRI
-					byte[] payload = val.stringValue().getBytes(StandardCharsets.UTF_8);
-					ByteBuffer lenBuf = ByteBuffer.allocate(9);
-					Varint.writeUnsigned(lenBuf, payload.length);
-					out.write(lenBuf.array(), 0, lenBuf.position());
-					out.write(payload);
+					out.writeByte(0);
+					writeBytes(out, buf, val.stringValue().getBytes(StandardCharsets.UTF_8));
 				} else if (val instanceof Literal) {
-					out.writeByte(1); // type = Literal
+					out.writeByte(1);
 					Literal lit = (Literal) val;
-					byte[] label = lit.getLabel().getBytes(StandardCharsets.UTF_8);
-					byte[] dt = lit.getDatatype().stringValue().getBytes(StandardCharsets.UTF_8);
-					String langStr = lit.getLanguage().orElse("");
-					byte[] lang = langStr.getBytes(StandardCharsets.UTF_8);
-
-					ByteBuffer buf = ByteBuffer.allocate(9);
-
-					buf.clear();
-					Varint.writeUnsigned(buf, label.length);
-					out.write(buf.array(), 0, buf.position());
-					out.write(label);
-
-					buf.clear();
-					Varint.writeUnsigned(buf, dt.length);
-					out.write(buf.array(), 0, buf.position());
-					out.write(dt);
-
-					buf.clear();
-					Varint.writeUnsigned(buf, lang.length);
-					out.write(buf.array(), 0, buf.position());
-					out.write(lang);
+					writeBytes(out, buf, lit.getLabel().getBytes(StandardCharsets.UTF_8));
+					writeBytes(out, buf, lit.getDatatype().stringValue().getBytes(StandardCharsets.UTF_8));
+					writeBytes(out, buf, lit.getLanguage().orElse("").getBytes(StandardCharsets.UTF_8));
 				} else if (val instanceof BNode) {
-					out.writeByte(2); // type = BNode
-					byte[] payload = ((BNode) val).getID().getBytes(StandardCharsets.UTF_8);
-					ByteBuffer lenBuf = ByteBuffer.allocate(9);
-					Varint.writeUnsigned(lenBuf, payload.length);
-					out.write(lenBuf.array(), 0, lenBuf.position());
-					out.write(payload);
+					out.writeByte(2);
+					writeBytes(out, buf, ((BNode) val).getID().getBytes(StandardCharsets.UTF_8));
 				} else {
 					throw new IllegalStateException("Unsupported value type: " + val.getClass());
 				}
@@ -249,4 +215,15 @@ void deserialize(ObjectStore objectStore, long nextValueId) {
 	public void close() {
 		clear();
 	}
+
+	private static void writeVarint(DataOutputStream out, ByteBuffer buf, long value) throws IOException {
+		buf.clear();
+		Varint.writeUnsigned(buf, value);
+		out.write(buf.array(), 0, buf.position());
+	}
+
+	private static void writeBytes(DataOutputStream out, ByteBuffer buf, byte[] data) throws IOException {
+		writeVarint(out, buf, data.length);
+		out.write(data);
+	}
 }

core/sail/s3/src/main/java/org/eclipse/rdf4j/sail/s3/cache/L2DiskCache.java

@@ -108,7 +108,7 @@ public void remove(String s3Key) {
 		}
 	}
 
-	private void evictIfNeeded(long incomingSize) {
+	private synchronized void evictIfNeeded(long incomingSize) {
 		while (currentSizeBytes.get() + incomingSize > maxSizeBytes && !index.isEmpty()) {
 			// Find LRU entry
 			String lruKey = null;
@@ -196,7 +196,7 @@ static class CacheEntry {
 		public long sizeBytes;
 
 		@JsonProperty("lastAccessNanos")
-		public long lastAccessNanos;
+		public volatile long lastAccessNanos;
 
 		public CacheEntry() {
 			// for Jackson deserialization