You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@hugegraph.apache.org by ji...@apache.org on 2026/10/08 15:34:55 UTC
(hugegraph) 05/15: fix(rocksdb): release native session owners
This is an automated email from the ASF dual-hosted git repository.
imbajin pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/hugegraph.git
commit 20003d6af04061847213031ea7b33aebccbcadcb
Author: imbajin <ji...@apache.org>
AuthorDate: Thu Oct 8 20:31:40 2026 +0800
fix(rocksdb): release native session owners
- close session and database native resources reliably
- preserve commits and clear reusable write batches
- cover native session disposal and reuse
---
.../backend/store/rocksdb/RocksDBOptions.java | 8 +
.../backend/store/rocksdb/RocksDBSessions.java | 36 +++
.../backend/store/rocksdb/RocksDBStdSessions.java | 21 +-
.../backend/store/rocksdb/RocksDBStore.java | 59 +++-
.../unit/rocksdb/RocksDBSessionsTest.java | 354 +++++++++++++++++++++
5 files changed, 466 insertions(+), 12 deletions(-)
diff --git a/hugegraph-server/hugegraph-rocksdb/src/main/java/org/apache/hugegraph/backend/store/rocksdb/RocksDBOptions.java b/hugegraph-server/hugegraph-rocksdb/src/main/java/org/apache/hugegraph/backend/store/rocksdb/RocksDBOptions.java
index 17148e094..0f0d18f1a 100644
--- a/hugegraph-server/hugegraph-rocksdb/src/main/java/org/apache/hugegraph/backend/store/rocksdb/RocksDBOptions.java
+++ b/hugegraph-server/hugegraph-rocksdb/src/main/java/org/apache/hugegraph/backend/store/rocksdb/RocksDBOptions.java
@@ -61,6 +61,14 @@ public class RocksDBOptions extends OptionHolder {
"rocksdb-data/data"
);
+ public static final ConfigOption<String> PROVIDER =
+ new ConfigOption<>(
+ "rocksdb.provider",
+ "The RocksDB-compatible runtime provider.",
+ allowValues("rocksdb", "topling"),
+ "rocksdb"
+ );
+
public static final ConfigListOption<String> DATA_DISKS =
new ConfigListOption<>(
"rocksdb.data_disks",
diff --git a/hugegraph-server/hugegraph-rocksdb/src/main/java/org/apache/hugegraph/backend/store/rocksdb/RocksDBSessions.java b/hugegraph-server/hugegraph-rocksdb/src/main/java/org/apache/hugegraph/backend/store/rocksdb/RocksDBSessions.java
index 7d0576e36..7c06081f7 100644
--- a/hugegraph-server/hugegraph-rocksdb/src/main/java/org/apache/hugegraph/backend/store/rocksdb/RocksDBSessions.java
+++ b/hugegraph-server/hugegraph-rocksdb/src/main/java/org/apache/hugegraph/backend/store/rocksdb/RocksDBSessions.java
@@ -17,6 +17,9 @@
package org.apache.hugegraph.backend.store.rocksdb;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collection;
import java.util.List;
import java.util.Set;
@@ -59,6 +62,39 @@ public abstract class RocksDBSessions extends BackendSessionPool {
public abstract void forceCloseRocksDB();
+ public boolean databaseOpened() {
+ return this.opened();
+ }
+
+ public void clearTables(Collection<String> tables) {
+ Session session = (Session) this.useSession();
+ try {
+ List<Pair<String, Pair<byte[], byte[]>>> ranges = new ArrayList<>();
+ for (String table : tables) {
+ Pair<byte[], byte[]> range = session.keyRange(table);
+ if (range != null) {
+ ranges.add(Pair.of(table, range));
+ }
+ }
+
+ for (Pair<String, Pair<byte[], byte[]>> entry : ranges) {
+ String table = entry.getLeft();
+ byte[] first = entry.getRight().getLeft();
+ byte[] last = entry.getRight().getRight();
+ if (!Arrays.equals(first, last)) {
+ session.deleteRange(table, first, last);
+ }
+ session.delete(table, last);
+ }
+ session.commit();
+ } catch (Throwable e) {
+ session.rollback();
+ throw e;
+ } finally {
+ this.close();
+ }
+ }
+
@Override
public abstract Session session();
diff --git a/hugegraph-server/hugegraph-rocksdb/src/main/java/org/apache/hugegraph/backend/store/rocksdb/RocksDBStdSessions.java b/hugegraph-server/hugegraph-rocksdb/src/main/java/org/apache/hugegraph/backend/store/rocksdb/RocksDBStdSessions.java
index 467de5a27..7dca2fc0e 100644
--- a/hugegraph-server/hugegraph-rocksdb/src/main/java/org/apache/hugegraph/backend/store/rocksdb/RocksDBStdSessions.java
+++ b/hugegraph-server/hugegraph-rocksdb/src/main/java/org/apache/hugegraph/backend/store/rocksdb/RocksDBStdSessions.java
@@ -43,6 +43,7 @@ import org.apache.hugegraph.config.HugeConfig;
import org.apache.hugegraph.util.Bytes;
import org.apache.hugegraph.util.E;
import org.apache.hugegraph.util.Log;
+import org.apache.hugegraph.util.RocksDBRuntime;
import org.apache.hugegraph.util.StringEncoding;
import org.rocksdb.BlockBasedTableConfig;
import org.rocksdb.BloomFilter;
@@ -367,6 +368,7 @@ public class RocksDBStdSessions extends RocksDBSessions {
private static OpenedRocksDB openRocksDB(HugeConfig config, String dataPath,
String walPath) throws RocksDBException {
+ RocksDBRuntime.verify(config.get(RocksDBOptions.PROVIDER));
// Init options
Options options = new Options();
RocksDBStdSessions.initOptions(config, options, options, options, options);
@@ -385,6 +387,7 @@ public class RocksDBStdSessions extends RocksDBSessions {
private static OpenedRocksDB openRocksDB(HugeConfig config,
List<String> cfNames, String dataPath,
String walPath) throws RocksDBException {
+ RocksDBRuntime.verify(config.get(RocksDBOptions.PROVIDER));
// Old CFs should always be opened
Set<String> mergedCFs = RocksDBStdSessions.mergeOldCFs(dataPath,
cfNames);
@@ -731,7 +734,15 @@ public class RocksDBStdSessions extends RocksDBSessions {
@Override
public void close() {
assert this.closeable();
- this.opened = false;
+ try {
+ this.batch.close();
+ } finally {
+ try {
+ this.writeOptions.close();
+ } finally {
+ this.opened = false;
+ }
+ }
}
@Override
@@ -741,7 +752,7 @@ public class RocksDBStdSessions extends RocksDBSessions {
@Override
public void reset() {
- this.batch = new WriteBatch();
+ this.batch.clear();
}
/**
@@ -782,14 +793,18 @@ public class RocksDBStdSessions extends RocksDBSessions {
RocksIterator iter = rocksdb().newIterator(cf.get())) {
iter.seekToFirst();
if (!iter.isValid()) {
+ iter.status();
return null;
}
startKey = iter.key();
iter.seekToLast();
if (!iter.isValid()) {
- return Pair.of(startKey, null);
+ iter.status();
+ throw new BackendException("Missing last key in table '%s'", table);
}
endKey = iter.key();
+ } catch (RocksDBException e) {
+ throw new BackendException("Failed to read key range of table '%s'", e, table);
}
return Pair.of(startKey, endKey);
}
diff --git a/hugegraph-server/hugegraph-rocksdb/src/main/java/org/apache/hugegraph/backend/store/rocksdb/RocksDBStore.java b/hugegraph-server/hugegraph-rocksdb/src/main/java/org/apache/hugegraph/backend/store/rocksdb/RocksDBStore.java
index eb7abcf6b..535101c80 100644
--- a/hugegraph-server/hugegraph-rocksdb/src/main/java/org/apache/hugegraph/backend/store/rocksdb/RocksDBStore.java
+++ b/hugegraph-server/hugegraph-rocksdb/src/main/java/org/apache/hugegraph/backend/store/rocksdb/RocksDBStore.java
@@ -60,6 +60,7 @@ import org.apache.hugegraph.backend.store.BackendTable;
import org.apache.hugegraph.config.CoreOptions;
import org.apache.hugegraph.config.HugeConfig;
import org.apache.hugegraph.exception.ConnectionException;
+import org.apache.hugegraph.exception.NotSupportException;
import org.apache.hugegraph.type.HugeType;
import org.apache.hugegraph.util.Consumers;
import org.apache.hugegraph.util.E;
@@ -103,6 +104,7 @@ public abstract class RocksDBStore extends AbstractBackendStore<RocksDBSessions.
*/
private static final int OPEN_POOL_THREADS = 8;
private boolean isGraphStore;
+ private boolean toplingProvider;
public RocksDBStore(final BackendStoreProvider provider,
final String database, final String store) {
@@ -212,6 +214,7 @@ public abstract class RocksDBStore extends AbstractBackendStore<RocksDBSessions.
E.checkNotNull(config, "config");
String graphStore = config.get(CoreOptions.STORE_GRAPH);
this.isGraphStore = this.store.equals(graphStore);
+ this.toplingProvider = "topling".equals(config.get(RocksDBOptions.PROVIDER));
this.dataPath = config.get(RocksDBOptions.DATA_PATH);
if (this.sessions != null && !this.sessions.closed()) {
@@ -417,7 +420,7 @@ public abstract class RocksDBStore extends AbstractBackendStore<RocksDBSessions.
@Override
public boolean opened() {
this.checkDbOpened();
- return this.sessions.session().opened();
+ return this.sessions.databaseOpened();
}
@Override
@@ -630,16 +633,48 @@ public abstract class RocksDBStore extends AbstractBackendStore<RocksDBSessions.
try {
this.checkOpened();
- this.clear(false);
- this.init();
- // Clear write-batch
- this.dbs.values().forEach(BackendSessionPool::forceResetSessions);
+ if (this.toplingProvider) {
+ if (!(this.sessions instanceof RocksDBStdSessions)) {
+ throw new NotSupportException("Topling does not support truncate for SST generation");
+ }
+ // Discard pending writes before committing the truncate batch.
+ this.dbs.values().forEach(BackendSessionPool::forceResetSessions);
+ this.clearTables();
+ this.init();
+ } else {
+ this.clear(false);
+ this.init();
+ this.dbs.values().forEach(BackendSessionPool::forceResetSessions);
+ }
LOG.debug("Store truncated: {}", this.store);
} finally {
writeLock.unlock();
}
}
+ private void clearTables() {
+ Map<String, RocksDBSessions> tableDBMap = this.tableDBMapping();
+ List<String> mainTables = new ArrayList<>(this.tableNames());
+ for (String table : tableDBMap.keySet()) {
+ if (table.equals(HugeType.OLAP.string())) {
+ mainTables.removeAll(this.olapTables());
+ } else {
+ mainTables.remove(table);
+ }
+ }
+ this.sessions.clearTables(mainTables);
+ for (Map.Entry<String, RocksDBSessions> entry :
+ tableDBMap.entrySet()) {
+ Collection<String> tables;
+ if (entry.getKey().equals(HugeType.OLAP.string())) {
+ tables = this.olapTables();
+ } else {
+ tables = Collections.singletonList(entry.getKey());
+ }
+ entry.getValue().clearTables(tables);
+ }
+ }
+
@Override
public void beginTx() {
Lock readLock = this.storeLock.readLock();
@@ -1104,10 +1139,16 @@ public abstract class RocksDBStore extends AbstractBackendStore<RocksDBSessions.
Lock writeLock = this.storeLock().writeLock();
writeLock.lock();
try {
- RocksDBSessions.Session session = super.session(HugeType.META);
- String driverVersion = this.provider().driverVersion();
- this.meta.writeVersion(session, driverVersion);
- LOG.info("Write down the backend version: {}", driverVersion);
+ RocksDBSessions sessions = super.db(HugeType.META);
+ sessions.useSession();
+ try {
+ RocksDBSessions.Session session = super.session(HugeType.META);
+ String driverVersion = this.provider().driverVersion();
+ this.meta.writeVersion(session, driverVersion);
+ LOG.info("Write down the backend version: {}", driverVersion);
+ } finally {
+ sessions.close();
+ }
} finally {
writeLock.unlock();
}
diff --git a/hugegraph-server/hugegraph-test/src/main/java/org/apache/hugegraph/unit/rocksdb/RocksDBSessionsTest.java b/hugegraph-server/hugegraph-test/src/main/java/org/apache/hugegraph/unit/rocksdb/RocksDBSessionsTest.java
index 5dbc96c5c..5c8b14c71 100644
--- a/hugegraph-server/hugegraph-test/src/main/java/org/apache/hugegraph/unit/rocksdb/RocksDBSessionsTest.java
+++ b/hugegraph-server/hugegraph-test/src/main/java/org/apache/hugegraph/unit/rocksdb/RocksDBSessionsTest.java
@@ -19,24 +19,378 @@ package org.apache.hugegraph.unit.rocksdb;
import java.io.File;
import java.io.IOException;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicReference;
import org.apache.commons.io.FileUtils;
import org.apache.hugegraph.backend.store.rocksdb.RocksDBMetrics;
import org.apache.hugegraph.backend.store.rocksdb.RocksDBOptions;
import org.apache.hugegraph.backend.store.rocksdb.RocksDBSessions;
import org.apache.hugegraph.backend.store.rocksdb.RocksDBStdSessions;
+import org.apache.hugegraph.exception.BackendException;
+import org.apache.hugegraph.backend.store.rocksdb.RocksDBStore;
+import org.apache.hugegraph.backend.store.rocksdb.RocksDBStoreProvider;
import org.apache.hugegraph.backend.store.rocksdbsst.RocksDBSstSessions;
import org.apache.hugegraph.config.HugeConfig;
import org.apache.hugegraph.testutil.Assert;
+import org.apache.hugegraph.testutil.Whitebox;
+import org.apache.hugegraph.type.HugeType;
import org.apache.hugegraph.unit.FakeObjects;
import org.junit.Test;
+import org.mockito.AdditionalAnswers;
+import org.mockito.Mockito;
+import org.rocksdb.ColumnFamilyHandle;
+import org.rocksdb.RocksDB;
+import org.rocksdb.RocksIterator;
import org.rocksdb.RocksDBException;
+import org.rocksdb.WriteBatch;
+import org.rocksdb.WriteOptions;
import com.google.common.collect.ImmutableList;
import com.google.common.collect.ImmutableSet;
public class RocksDBSessionsTest extends BaseRocksDBUnitTest {
+ @Test
+ public void testResetDiscardsPendingWritesWithoutLeakingNativeBatch() throws Exception {
+ RocksDBSessions.Session session = this.rocks.session();
+ WriteBatch previous = Whitebox.getInternalState(session, "batch");
+ for (int i = 0; i < 32; i++) {
+ session.put(TABLE, getBytes("pending"), getBytes("value"));
+ this.rocks.forceResetSessions();
+ WriteBatch current = Whitebox.getInternalState(session, "batch");
+ Assert.assertFalse(session.hasChanges());
+ Assert.assertTrue(current.isOwningHandle());
+ Assert.assertTrue(current == previous || !previous.isOwningHandle());
+ previous = current;
+ }
+ session.put(TABLE, getBytes("retained"), getBytes("value"));
+ session.commit();
+ Assert.assertNull(this.get("pending"));
+ Assert.assertEquals("value", this.get("retained"));
+ }
+
+ @Test
+ public void testFinalDetachDisposesNativeOwners() throws Exception {
+ RocksDBSessions.Session session = this.rocks.session();
+ WriteBatch batch = Whitebox.getInternalState(session, "batch");
+ WriteOptions options = Whitebox.getInternalState(session, "writeOptions");
+ // Keep the DB alive while this worker releases its final request lease.
+ CountDownLatch attached = new CountDownLatch(1);
+ CountDownLatch release = new CountDownLatch(1);
+ AtomicReference<Throwable> failure = new AtomicReference<>();
+ Thread keeper = new Thread(() -> {
+ try {
+ this.rocks.session();
+ attached.countDown();
+ release.await();
+ } catch (Throwable e) {
+ failure.set(e);
+ } finally {
+ attached.countDown();
+ try {
+ this.rocks.close();
+ } catch (Throwable e) {
+ failure.compareAndSet(null, e);
+ }
+ }
+ });
+ keeper.start();
+ try {
+ Assert.assertTrue(attached.await(30, TimeUnit.SECONDS));
+ Assert.assertNull(failure.get());
+ Assert.assertSame(session, this.rocks.useSession());
+ Assert.assertFalse(this.rocks.close());
+ Assert.assertTrue(batch.isOwningHandle());
+ Assert.assertTrue(options.isOwningHandle());
+ session.put(TABLE, getBytes("lease"), getBytes("retained"));
+ session.commit();
+ Assert.assertFalse(this.rocks.close());
+ Assert.assertFalse(batch.isOwningHandle());
+ Assert.assertFalse(options.isOwningHandle());
+ // Native close is idempotent even after the pool removed this session.
+ session.close();
+ Assert.assertFalse(batch.isOwningHandle());
+ Assert.assertFalse(options.isOwningHandle());
+ for (int i = 0; i < 32; i++) {
+ RocksDBSessions.Session request = this.rocks.session();
+ Assert.assertNotSame(session, request);
+ WriteBatch requestBatch = Whitebox.getInternalState(request, "batch");
+ WriteOptions requestOptions = Whitebox.getInternalState(request, "writeOptions");
+ Assert.assertTrue(requestBatch.isOwningHandle());
+ Assert.assertTrue(requestOptions.isOwningHandle());
+ Assert.assertArrayEquals(getBytes("retained"), request.get(TABLE, getBytes("lease")));
+ request.put(TABLE, getBytes("request"), getBytes("value"));
+ request.commit();
+ Assert.assertFalse(this.rocks.close());
+ Assert.assertFalse(requestBatch.isOwningHandle());
+ Assert.assertFalse(requestOptions.isOwningHandle());
+ }
+ } finally {
+ // Leave the fixture's main-worker session open for ordinary teardown.
+ this.rocks.session();
+ release.countDown();
+ keeper.join(TimeUnit.SECONDS.toMillis(30));
+ }
+ Assert.assertFalse(keeper.isAlive());
+ Assert.assertNull(failure.get());
+ Assert.assertEquals("retained", this.get("lease"));
+ }
+
+ @Test
+ public void testAdapterToplingTruncateWithMultipleKeys() throws Exception {
+ this.assertAdapterTruncate(true);
+ }
+
+ @Test
+ public void testAdapterStandardTruncateWithMultipleKeys() throws Exception {
+ this.assertAdapterTruncate(false);
+ }
+
+ @Test
+ public void testAdapterToplingTruncateDiscardsPendingWritesInEmptyTable() throws Exception {
+ this.assertAdapterTruncateDiscardsPendingWrites(true, false);
+ }
+
+ @Test
+ public void testAdapterToplingTruncateDiscardsPendingWritesOutsideRange() throws Exception {
+ this.assertAdapterTruncateDiscardsPendingWrites(true, true);
+ }
+
+ @Test
+ public void testAdapterStandardTruncateDiscardsPendingWritesInEmptyTable() throws Exception {
+ this.assertAdapterTruncateDiscardsPendingWrites(false, false);
+ }
+
+ @Test
+ public void testAdapterStandardTruncateDiscardsPendingWritesOutsideRange() throws Exception {
+ this.assertAdapterTruncateDiscardsPendingWrites(false, true);
+ }
+
+ @Test
+ public void testAdapterToplingTruncatePreservesBackendVersion() throws Exception {
+ this.assertAdapterTruncatePreservesBackendVersion(true);
+ }
+
+ @Test
+ public void testAdapterStandardTruncatePreservesBackendVersion() throws Exception {
+ this.assertAdapterTruncatePreservesBackendVersion(false);
+ }
+
+ @Test
+ public void testToplingTruncateRoutesDynamicOlapTablesToTheirDatabase() throws Exception {
+ String dynamic = "graph+ap_123";
+ String olapPath = DB_PATH + "/independent-olap";
+ RocksDBStdSessions olap = new RocksDBStdSessions(FakeObjects.newConfig(), "db", "olap", olapPath, olapPath);
+ try {
+ olap.createTable(dynamic);
+ olap.session().put(dynamic, getBytes("key"), getBytes("olap value"));
+ olap.session().commit();
+ this.put("main", "main value");
+ RocksDBStore store = new RocksDBStore.RocksDBGraphStore(null, "db", "graph") {
+ @Override
+ protected List<String> tableNames() {
+ return ImmutableList.of(TABLE, dynamic);
+ }
+
+ @Override
+ protected List<String> olapTables() {
+ return ImmutableList.of(dynamic);
+ }
+
+ @Override
+ protected Map<String, RocksDBSessions> tableDBMapping() {
+ return java.util.Collections.singletonMap(HugeType.OLAP.string(), olap);
+ }
+ };
+ Whitebox.setInternalState(store, "sessions", this.rocks);
+ Map<String, RocksDBSessions> databases = Whitebox.getInternalState(store, "dbs");
+ databases.put(DB_PATH, this.rocks);
+ databases.put(olapPath, olap);
+ Whitebox.setInternalState(store, "toplingProvider", true);
+ // Real standard JNI exercises the Topling Java routing branch.
+ Assert.assertFalse(this.rocks.existsTable(dynamic));
+ store.truncate();
+ Assert.assertNull(this.get("main"));
+ Assert.assertTrue(olap.existsTable(dynamic));
+ Assert.assertNull(olap.session().get(dynamic, getBytes("key")));
+ olap.session().put(dynamic, getBytes("next"), getBytes("after truncate"));
+ olap.session().commit();
+ Assert.assertArrayEquals(getBytes("after truncate"), olap.session().get(dynamic, getBytes("next")));
+ } finally {
+ olap.close();
+ }
+ }
+
+ private void assertAdapterTruncatePreservesBackendVersion(boolean topling) throws Exception {
+ RocksDBStoreProvider provider = new RocksDBStoreProvider();
+ RocksDBStore store = new RocksDBStore.RocksDBSystemStore(provider, "db", "store") {
+ @Override
+ protected List<String> tableNames() {
+ return ImmutableList.<String>builder().addAll(super.tableNames()).add(TABLE).build();
+ }
+ };
+ Whitebox.setInternalState(store, "sessions", this.rocks);
+ Map<String, RocksDBSessions> databases = Whitebox.getInternalState(store, "dbs");
+ databases.put(DB_PATH, this.rocks);
+ Whitebox.setInternalState(store, "toplingProvider", topling);
+ store.init();
+ this.put("old", "before truncate");
+ Assert.assertEquals(provider.driverVersion(), store.storedVersion());
+
+ store.truncate();
+
+ Assert.assertNull(this.get("old"));
+ Assert.assertEquals(provider.driverVersion(), store.storedVersion());
+ this.put("new", "after truncate");
+ this.rocks.close();
+ this.rocks = new RocksDBStdSessions(FakeObjects.newConfig(), "db", "store",
+ DB_PATH, DB_PATH, ImmutableList.of(TABLE));
+ this.rocks.session();
+ Whitebox.setInternalState(store, "sessions", this.rocks);
+ databases.clear();
+ databases.put(DB_PATH, this.rocks);
+ // Reopening must read the persisted version without init repairing it.
+ Assert.assertEquals(provider.driverVersion(), store.storedVersion());
+ Assert.assertEquals("after truncate", this.get("new"));
+ }
+
+ private void assertAdapterTruncateDiscardsPendingWrites(boolean topling, boolean committedRange) throws Exception {
+ if (committedRange) {
+ this.put("m", "committed first");
+ this.put("n", "committed last");
+ }
+ RocksDBSessions.Session session = this.rocks.session();
+ // Both inserts fall outside the committed range; an empty CF has no range at all.
+ session.put(TABLE, getBytes("a"), getBytes("pending first"));
+ session.put(TABLE, getBytes("z"), getBytes("pending last"));
+ Assert.assertTrue(session.hasChanges());
+
+ RocksDBStore store = this.adapterStore(topling, ImmutableList.of(TABLE));
+ store.truncate();
+
+ Assert.assertFalse(session.hasChanges());
+ Assert.assertNull(session.keyRange(TABLE));
+ Assert.assertNull(session.get(TABLE, getBytes("a")));
+ Assert.assertNull(session.get(TABLE, getBytes("z")));
+ // The retained session must not replay pending writes after truncate either.
+ session.commit();
+ Assert.assertNull(session.keyRange(TABLE));
+ this.put("new", "after truncate");
+ Assert.assertEquals("after truncate", this.get("new"));
+ }
+
+ private void assertAdapterTruncate(boolean topling) throws Exception {
+ List<String> tables = ImmutableList.of(TABLE, "single", "empty", "multiple");
+ this.rocks.createTable("single", "empty", "multiple");
+ // Deliberately unordered, including empty and unsigned byte boundaries.
+ byte[][] keys = {new byte[]{(byte) 0xff}, new byte[]{0}, new byte[]{},
+ new byte[]{(byte) 0x80}, new byte[]{0, (byte) 0xff}, new byte[]{0x7f}};
+ for (String table : ImmutableList.of(TABLE, "multiple")) {
+ for (byte[] key : keys) {
+ this.rocks.session().put(table, key, new byte[]{42});
+ }
+ }
+ this.rocks.session().put("single", new byte[]{(byte) 0xff}, new byte[]{43});
+ this.commit();
+ Object opened = Whitebox.getInternalState(this.rocks, "rocksdb");
+ Map<String, ?> handles = Whitebox.getInternalState(opened, "cfHandles");
+ Map<String, ?> before = new HashMap<>(handles);
+ RocksDBStore store = this.adapterStore(topling, tables);
+ store.truncate();
+ for (String table : tables) {
+ Assert.assertTrue(this.rocks.existsTable(table));
+ Assert.assertNull(this.rocks.session().keyRange(table));
+ for (byte[] key : keys) {
+ Assert.assertNull(this.rocks.session().get(table, key));
+ }
+ if (topling) {
+ Assert.assertSame(before.get(table), handles.get(table));
+ } else {
+ Assert.assertNotSame(before.get(table), handles.get(table));
+ }
+ }
+ for (String table : tables) {
+ this.rocks.session().put(table, new byte[]{(byte) 0xff}, new byte[]{44});
+ }
+ this.commit();
+ for (String table : tables) {
+ Assert.assertArrayEquals(new byte[]{44}, this.rocks.session().get(table, new byte[]{(byte) 0xff}));
+ }
+ // A second truncate covers the single-key path for every formerly empty CF.
+ store.truncate();
+ for (String table : tables) {
+ Assert.assertNull(this.rocks.session().keyRange(table));
+ }
+ }
+
+ @Test
+ public void testTruncatePropagatesFirstKeyReadFailure() throws Exception {
+ this.assertTruncateKeyReadFailure(false);
+ }
+
+ @Test
+ public void testTruncatePropagatesLastKeyReadFailure() throws Exception {
+ this.assertTruncateKeyReadFailure(true);
+ }
+
+ private void assertTruncateKeyReadFailure(boolean lastKey) throws Exception {
+ String brokenTable = "broken";
+ this.rocks.createTable(brokenTable);
+ this.put("healthy", "retained");
+ this.rocks.session().put(brokenTable, getBytes("a"), getBytes("first"));
+ this.rocks.session().put(brokenTable, getBytes("z"), getBytes("last"));
+ this.commit();
+ RocksDBStore store = this.adapterStore(true, ImmutableList.of(TABLE, brokenTable));
+ Object opened = Whitebox.getInternalState(this.rocks, "rocksdb");
+ RocksDB realDB = Whitebox.getInternalState(opened, "rocksdb");
+ Map<String, ?> handles = Whitebox.getInternalState(opened, "cfHandles");
+ ColumnFamilyHandle brokenHandle = Whitebox.getInternalState(handles.get(brokenTable), "handle");
+ RocksIterator brokenIterator = Mockito.mock(RocksIterator.class);
+ Mockito.when(brokenIterator.isValid()).thenReturn(lastKey, false);
+ Mockito.when(brokenIterator.key()).thenReturn(getBytes("a"));
+ RocksDBException readFailure = new RocksDBException("injected iterator read failure");
+ Mockito.doThrow(readFailure).when(brokenIterator).status();
+ // Keep a real DB and healthy iterator; replace only the failing native
+ // boundary. The production truncate -> clearTables -> keyRange chain runs.
+ RocksDB faultDB = Mockito.mock(RocksDB.class, AdditionalAnswers.delegatesTo(realDB));
+ Mockito.doReturn(brokenIterator).when(faultDB).newIterator(brokenHandle);
+ Whitebox.setInternalState(opened, "rocksdb", faultDB);
+ try {
+ Throwable failure = Assert.assertThrows(BackendException.class, store::truncate);
+ Assert.assertSame(readFailure, failure.getCause());
+ Mockito.verify(brokenIterator).status();
+ Mockito.verify(brokenIterator).close();
+ } finally {
+ Whitebox.setInternalState(opened, "rocksdb", realDB);
+ }
+ Assert.assertEquals("retained", this.get("healthy"));
+ Assert.assertArrayEquals(getBytes("first"), this.rocks.session().get(brokenTable, getBytes("a")));
+ Assert.assertArrayEquals(getBytes("last"), this.rocks.session().get(brokenTable, getBytes("z")));
+ Assert.assertFalse(this.rocks.session().hasChanges());
+ Assert.assertSame(brokenHandle, Whitebox.getInternalState(handles.get(brokenTable), "handle"));
+ }
+
+ private RocksDBStore adapterStore(boolean topling, List<String> tables) {
+ RocksDBStore store = new RocksDBStore.RocksDBGraphStore(new RocksDBStoreProvider(), "db", "store") {
+ @Override
+ protected List<String> tableNames() {
+ return tables;
+ }
+ };
+ Whitebox.setInternalState(store, "sessions", this.rocks);
+ Map<String, RocksDBSessions> databases = Whitebox.getInternalState(store, "dbs");
+ databases.put(DB_PATH, this.rocks);
+ // Select the actual adapter branch, independently of the loaded JNI.
+ // This fixture also runs unchanged with the externally loaded TP JNI on Linux.
+ Whitebox.setInternalState(store, "toplingProvider", topling);
+ return store;
+ }
+
@Test
public void testTable() throws RocksDBException {
final String TABLE2 = "test-table2";