From 6126b95981ab7b5b4a393b3d5263b02809da6fbc Mon Sep 17 00:00:00 2001 From: contrueCT Date: Wed, 12 Aug 2026 17:09:20 +0800 Subject: [PATCH 1/4] fix(store): allocate graph id before batch writes --- .../store/business/BusinessHandlerImpl.java | 4 +- hugegraph-store/hg-store-test/pom.xml | 1 + .../store/core/BatchGraphIsolationTest.java | 175 ++++++++++++++++++ 3 files changed, 178 insertions(+), 2 deletions(-) create mode 100644 hugegraph-store/hg-store-test/src/main/java/org/apache/hugegraph/store/core/BatchGraphIsolationTest.java diff --git a/hugegraph-store/hg-store-core/src/main/java/org/apache/hugegraph/store/business/BusinessHandlerImpl.java b/hugegraph-store/hg-store-core/src/main/java/org/apache/hugegraph/store/business/BusinessHandlerImpl.java index 9287bfe267..266d90ad9f 100644 --- a/hugegraph-store/hg-store-core/src/main/java/org/apache/hugegraph/store/business/BusinessHandlerImpl.java +++ b/hugegraph-store/hg-store-core/src/main/java/org/apache/hugegraph/store/business/BusinessHandlerImpl.java @@ -1577,7 +1577,7 @@ private TxBuilderImpl(String graph, int partId, RocksDBSession dbSession) { public TxBuilder put(int code, String table, byte[] key, byte[] value) throws HgStoreException { try { - byte[] targetKey = keyCreator.getKey(this.partId, graph, code, key); + byte[] targetKey = keyCreator.getKeyOrCreate(this.partId, graph, code, key); this.op.put(table, targetKey, value); } catch (DBStoreException e) { throw new HgStoreException(HgStoreException.EC_RKDB_DOPUT_FAIL, e.toString()); @@ -1642,7 +1642,7 @@ public TxBuilder merge(int code, String table, byte[] key, byte[] value) throws HgStoreException { try { - byte[] targetKey = keyCreator.getKey(this.partId, graph, code, key); + byte[] targetKey = keyCreator.getKeyOrCreate(this.partId, graph, code, key); op.merge(table, targetKey, value); } catch (DBStoreException e) { throw new HgStoreException(HgStoreException.EC_RKDB_DOMERGE_FAIL, e.toString()); diff --git a/hugegraph-store/hg-store-test/pom.xml b/hugegraph-store/hg-store-test/pom.xml index 36308f449d..bd5e5a0cf2 100644 --- a/hugegraph-store/hg-store-test/pom.xml +++ b/hugegraph-store/hg-store-test/pom.xml @@ -236,6 +236,7 @@ **/CoreSuiteTest.java + **/BatchGraphIsolationTest.java diff --git a/hugegraph-store/hg-store-test/src/main/java/org/apache/hugegraph/store/core/BatchGraphIsolationTest.java b/hugegraph-store/hg-store-test/src/main/java/org/apache/hugegraph/store/core/BatchGraphIsolationTest.java new file mode 100644 index 0000000000..e037e99780 --- /dev/null +++ b/hugegraph-store/hg-store-test/src/main/java/org/apache/hugegraph/store/core/BatchGraphIsolationTest.java @@ -0,0 +1,175 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.hugegraph.store.core; + +import static org.apache.hugegraph.store.constant.HugeServerTables.TABLES_MAP; +import static org.apache.hugegraph.store.constant.HugeServerTables.VERTEX_TABLE; + +import java.io.IOException; +import java.nio.ByteBuffer; +import java.nio.ByteOrder; +import java.nio.charset.StandardCharsets; +import java.nio.file.Files; +import java.nio.file.Path; +import java.util.Collections; +import java.util.HashMap; +import java.util.List; +import java.util.Map; + +import org.apache.hugegraph.store.UnitTestBase; +import org.apache.hugegraph.store.business.BusinessHandler; +import org.apache.hugegraph.store.business.BusinessHandlerImpl; +import org.apache.hugegraph.store.grpc.common.Key; +import org.apache.hugegraph.store.grpc.common.OpType; +import org.apache.hugegraph.store.grpc.session.BatchEntry; +import org.apache.hugegraph.store.meta.PartitionManager; +import org.apache.hugegraph.store.options.HgStoreEngineOptions; +import org.apache.hugegraph.store.options.RaftRocksdbOptions; +import org.apache.hugegraph.store.pd.FakePdServiceProvider; +import org.apache.hugegraph.store.pd.PdProvider; +import org.junit.AfterClass; +import org.junit.Assert; +import org.junit.BeforeClass; +import org.junit.Test; + +import com.alipay.sofa.jraft.util.StorageOptionsFactory; +import com.google.protobuf.ByteString; + +public class BatchGraphIsolationTest { + + private static final int PARTITION_ID = 0; + private static final int KEY_CODE = 0; + private static final byte[] SHARED_KEY = + "shared-key".getBytes(StandardCharsets.UTF_8); + + private static Path databasePath; + private static BusinessHandler handler; + + @BeforeClass + public static void setup() throws IOException { + databasePath = Files.createTempDirectory("hugegraph-batch-graph-isolation-"); + + Map rocksdbConfig = new HashMap<>(); + rocksdbConfig.put("rocksdb.write_buffer_size", "1048576"); + StorageOptionsFactory.releaseAllOptions(); + RaftRocksdbOptions.initRocksdbGlobalConfig(rocksdbConfig); + BusinessHandlerImpl.initRocksdb(rocksdbConfig, null); + + HgStoreEngineOptions options = new HgStoreEngineOptions(); + options.setDataPath(databasePath.toString()); + options.setRaftPath(databasePath.toString()); + + HgStoreEngineOptions.FakePdOptions fakePdOptions = + new HgStoreEngineOptions.FakePdOptions(); + fakePdOptions.setPartitionCount(1); + fakePdOptions.setPeersList("127.0.0.1"); + fakePdOptions.setStoreList("127.0.0.1"); + options.setFakePdOptions(fakePdOptions); + + PdProvider pdProvider = new FakePdServiceProvider(fakePdOptions); + PartitionManager partitionManager = new PartitionManager(pdProvider, options) { + + @Override + public String getDbDataPath(int partitionId, String dbName) { + return databasePath.resolve("data").toString(); + } + + @Override + public boolean hasPartition(String graphName, int partitionId) { + return partitionId == PARTITION_ID; + } + + @Override + public List getLeaderPartitionIds(String graph) { + return Collections.singletonList(PARTITION_ID); + } + }; + handler = new BusinessHandlerImpl(partitionManager); + handler.createTable("setup", PARTITION_ID, VERTEX_TABLE); + } + + @AfterClass + public static void teardown() { + if (handler != null) { + handler.closeAll(); + } + if (databasePath != null) { + UnitTestBase.deleteDir(databasePath.toFile()); + } + } + + @Test + public void testBatchPutKeepsGraphsIsolated() { + String graph1 = "batch-put-graph-1"; + String graph2 = "batch-put-graph-2"; + byte[] value1 = "graph-1-value".getBytes(StandardCharsets.UTF_8); + byte[] value2 = "graph-2-value".getBytes(StandardCharsets.UTF_8); + + writeBatch(graph1, OpType.OP_TYPE_PUT, value1); + writeBatch(graph2, OpType.OP_TYPE_PUT, value2); + + Assert.assertArrayEquals(value1, read(graph1)); + Assert.assertArrayEquals(value2, read(graph2)); + + handler.truncate(graph2, PARTITION_ID); + Assert.assertArrayEquals(value1, read(graph1)); + } + + @Test + public void testBatchMergeKeepsGraphsIsolated() { + String graph1 = "batch-merge-graph-1"; + String graph2 = "batch-merge-graph-2"; + + writeBatch(graph1, OpType.OP_TYPE_MERGE, longToBytes(10L)); + writeBatch(graph2, OpType.OP_TYPE_MERGE, longToBytes(20L)); + + Assert.assertEquals(10L, bytesToLong(read(graph1))); + Assert.assertEquals(20L, bytesToLong(read(graph2))); + } + + private static void writeBatch(String graph, OpType type, byte[] value) { + Key key = Key.newBuilder() + .setCode(KEY_CODE) + .setKey(ByteString.copyFrom(SHARED_KEY)) + .build(); + BatchEntry entry = BatchEntry.newBuilder() + .setOpType(type) + .setTable(TABLES_MAP.get(VERTEX_TABLE)) + .setStartKey(key) + .setValue(ByteString.copyFrom(value)) + .build(); + handler.doBatch(graph, PARTITION_ID, Collections.singletonList(entry)); + } + + private static byte[] read(String graph) { + return handler.doGet(graph, KEY_CODE, VERTEX_TABLE, SHARED_KEY); + } + + private static byte[] longToBytes(long value) { + return ByteBuffer.allocate(Long.BYTES) + .order(ByteOrder.LITTLE_ENDIAN) + .putLong(value) + .array(); + } + + private static long bytesToLong(byte[] value) { + return ByteBuffer.wrap(value) + .order(ByteOrder.LITTLE_ENDIAN) + .getLong(); + } +} From ec8094ad36b4b0ea878663341cced4eeba061773 Mon Sep 17 00:00:00 2001 From: contrueCT Date: Sun, 16 Aug 2026 23:54:45 +0800 Subject: [PATCH 2/4] fix(store): avoid graph id allocation deadlock --- .../apache/hugegraph/store/meta/GraphIdManager.java | 11 +++++++++-- .../hugegraph/store/core/BatchGraphIsolationTest.java | 11 +++++++++++ 2 files changed, 20 insertions(+), 2 deletions(-) diff --git a/hugegraph-store/hg-store-core/src/main/java/org/apache/hugegraph/store/meta/GraphIdManager.java b/hugegraph-store/hg-store-core/src/main/java/org/apache/hugegraph/store/meta/GraphIdManager.java index 3a9e2c18e7..3c95919591 100644 --- a/hugegraph-store/hg-store-core/src/main/java/org/apache/hugegraph/store/meta/GraphIdManager.java +++ b/hugegraph-store/hg-store-core/src/main/java/org/apache/hugegraph/store/meta/GraphIdManager.java @@ -17,6 +17,8 @@ package org.apache.hugegraph.store.meta; +import static org.apache.hugegraph.store.constant.HugeServerTables.VERTEX_TABLE; + import java.nio.ByteBuffer; import java.util.Arrays; import java.util.List; @@ -128,8 +130,13 @@ public long releaseGraphId(String graphName) { private boolean checkCount(long l) { var start = new byte[2]; Bits.putShort(start, 0, (short) l); - try (var itr = sessionBuilder.getSession(partitionId).sessionOp().scan("g+v", start)) { - return itr == null || !itr.hasNext(); + try (var session = sessionBuilder.getSession(partitionId)) { + if (!session.tableIsExist(VERTEX_TABLE)) { + return true; + } + try (var itr = session.sessionOp().scan(VERTEX_TABLE, start)) { + return itr == null || !itr.hasNext(); + } } } diff --git a/hugegraph-store/hg-store-test/src/main/java/org/apache/hugegraph/store/core/BatchGraphIsolationTest.java b/hugegraph-store/hg-store-test/src/main/java/org/apache/hugegraph/store/core/BatchGraphIsolationTest.java index e037e99780..9c8fb4494d 100644 --- a/hugegraph-store/hg-store-test/src/main/java/org/apache/hugegraph/store/core/BatchGraphIsolationTest.java +++ b/hugegraph-store/hg-store-test/src/main/java/org/apache/hugegraph/store/core/BatchGraphIsolationTest.java @@ -53,6 +53,7 @@ public class BatchGraphIsolationTest { private static final int PARTITION_ID = 0; + private static final int EMPTY_PARTITION_ID = 1; private static final int KEY_CODE = 0; private static final byte[] SHARED_KEY = "shared-key".getBytes(StandardCharsets.UTF_8); @@ -113,6 +114,16 @@ public static void teardown() { } } + @Test + public void testGraphIdAllocationDoesNotCreateVertexTable() { + String graph = "graph-id-allocation"; + + Assert.assertFalse(handler.existsTable(graph, EMPTY_PARTITION_ID, VERTEX_TABLE)); + ((BusinessHandlerImpl) handler).getKeyCreator() + .getGraphIdOrCreate(EMPTY_PARTITION_ID, graph); + Assert.assertFalse(handler.existsTable(graph, EMPTY_PARTITION_ID, VERTEX_TABLE)); + } + @Test public void testBatchPutKeepsGraphsIsolated() { String graph1 = "batch-put-graph-1"; From 2f2765f82b97463b07e3b437a8af94288ac215a9 Mon Sep 17 00:00:00 2001 From: contrueCT Date: Tue, 25 Aug 2026 19:23:39 +0800 Subject: [PATCH 3/4] fix(store): guard batch graph id lifecycle --- .../store/business/BusinessHandler.java | 9 +- .../store/business/BusinessHandlerImpl.java | 87 ++++++++-- .../store/business/DataManagerImpl.java | 18 +- .../store/business/DefaultDataMover.java | 18 +- .../store/core/BatchGraphIsolationTest.java | 154 ++++++++++++++++-- 5 files changed, 253 insertions(+), 33 deletions(-) diff --git a/hugegraph-store/hg-store-core/src/main/java/org/apache/hugegraph/store/business/BusinessHandler.java b/hugegraph-store/hg-store-core/src/main/java/org/apache/hugegraph/store/business/BusinessHandler.java index 8133654387..58e0056e5b 100644 --- a/hugegraph-store/hg-store-core/src/main/java/org/apache/hugegraph/store/business/BusinessHandler.java +++ b/hugegraph-store/hg-store-core/src/main/java/org/apache/hugegraph/store/business/BusinessHandler.java @@ -152,6 +152,7 @@ boolean cleanPartition(String graph, int partId, long startKey, long endKey, default void doBatch(String graph, int partId, List entryList) { BusinessHandler.TxBuilder builder = txBuilder(graph, partId); + BusinessHandler.Tx transaction = builder.build(); try { for (BatchEntry b : entryList) { Key start = b.getStartKey(); @@ -185,12 +186,16 @@ default void doBatch(String graph, int partId, List entryList) { } } } - builder.build().commit(); + transaction.commit(); } catch (Throwable e) { String msg = String.format("graph data %s-%s do batch insert with error:", graph, partId); log.error(msg, e); - builder.build().rollback(); + try { + transaction.rollback(); + } catch (Throwable rollbackError) { + e.addSuppressed(rollbackError); + } throw e; } } diff --git a/hugegraph-store/hg-store-core/src/main/java/org/apache/hugegraph/store/business/BusinessHandlerImpl.java b/hugegraph-store/hg-store-core/src/main/java/org/apache/hugegraph/store/business/BusinessHandlerImpl.java index 266d90ad9f..d66093e290 100644 --- a/hugegraph-store/hg-store-core/src/main/java/org/apache/hugegraph/store/business/BusinessHandlerImpl.java +++ b/hugegraph-store/hg-store-core/src/main/java/org/apache/hugegraph/store/business/BusinessHandlerImpl.java @@ -38,6 +38,9 @@ import java.util.concurrent.ThreadPoolExecutor; import java.util.concurrent.TimeoutException; import java.util.concurrent.atomic.AtomicInteger; +import java.util.concurrent.locks.Lock; +import java.util.concurrent.locks.ReadWriteLock; +import java.util.concurrent.locks.ReentrantReadWriteLock; import java.util.function.BiFunction; import java.util.function.Consumer; import java.util.function.Function; @@ -121,6 +124,8 @@ public class BusinessHandlerImpl implements BusinessHandler { private static final Map GRAPH_SUPPLIER_CACHE = new ConcurrentHashMap<>(); + private static final int GRAPH_LOCK_STRIPES = 1024; + private static final ReadWriteLock[] GRAPH_LOCKS = createGraphLocks(); private static final int batchSize = 10000; private static Long indexDataSize = 50 * 1024L; private static final RocksDBFactory factory = RocksDBFactory.getInstance(); @@ -172,6 +177,20 @@ public void onDBSessionReleased(RocksDBSession dbSession) { }); } + private static ReadWriteLock[] createGraphLocks() { + ReadWriteLock[] locks = new ReadWriteLock[GRAPH_LOCK_STRIPES]; + for (int i = 0; i < locks.length; i++) { + locks[i] = new ReentrantReadWriteLock(true); + } + return locks; + } + + private static ReadWriteLock graphLock(String graph, int partId) { + int hash = 31 * partId + graph.hashCode(); + hash ^= hash >>> 16; + return GRAPH_LOCKS[hash & (GRAPH_LOCK_STRIPES - 1)]; + } + public static HugeConfig initRocksdb(Map rocksdbConfig, RocksdbChangedListener listener) { // Register rocksdb configuration @@ -1014,11 +1033,17 @@ public void batchGet(String graph, String table, Supplier truncate = executor.submit(() -> { + truncateStarted.countDown(); + handler.truncate(graph, PARTITION_ID); + }); + + Assert.assertTrue(truncateStarted.await(1L, TimeUnit.SECONDS)); + try { + truncate.get(500L, TimeUnit.MILLISECONDS); + Assert.fail("truncate must wait for the in-flight batch"); + } catch (TimeoutException expected) { + // Expected until the transaction releases its graph ID reservation. + } + + transaction.commit(); + committed = true; + truncate.get(5L, TimeUnit.SECONDS); + + writeBatch(nextGraph, OpType.OP_TYPE_PUT, nextValue); + Assert.assertArrayEquals(nextValue, read(nextGraph)); + } finally { + if (!committed) { + transaction.rollback(); + } + executor.shutdownNow(); + } + } + + @Test + public void testDataManagerRollsBackFailedBatch() { + BusinessHandler mockHandler = Mockito.mock(BusinessHandler.class); + BusinessHandler.TxBuilder mockBuilder = Mockito.mock(BusinessHandler.TxBuilder.class); + BusinessHandler.Tx mockTransaction = Mockito.mock(BusinessHandler.Tx.class); + BatchPutRequest request = batchPutRequest(); + BatchPutRequest.KV entry = request.getEntries().get(0); + RuntimeException failure = new RuntimeException("injected put failure"); + Mockito.when(mockHandler.txBuilder(request.getGraphName(), request.getPartitionId())) + .thenReturn(mockBuilder); + Mockito.when(mockBuilder.build()).thenReturn(mockTransaction); + Mockito.doThrow(failure).when(mockBuilder) + .put(entry.getCode(), entry.getTable(), entry.getKey(), entry.getValue()); + DataManagerImpl dataManager = new DataManagerImpl(); + dataManager.setBusinessHandler(mockHandler); + + assertWriteFails(failure, () -> dataManager.write(request)); + + Mockito.verify(mockTransaction).rollback(); + } + + @SuppressWarnings("deprecation") + @Test + public void testDefaultDataMoverRollsBackFailedBatch() { + BusinessHandler mockHandler = Mockito.mock(BusinessHandler.class); + BusinessHandler.TxBuilder mockBuilder = Mockito.mock(BusinessHandler.TxBuilder.class); + BusinessHandler.Tx mockTransaction = Mockito.mock(BusinessHandler.Tx.class); + BatchPutRequest request = batchPutRequest(); + BatchPutRequest.KV entry = request.getEntries().get(0); + RuntimeException failure = new RuntimeException("injected put failure"); + Mockito.when(mockHandler.txBuilder(request.getGraphName(), request.getPartitionId())) + .thenReturn(mockBuilder); + Mockito.when(mockBuilder.build()).thenReturn(mockTransaction); + Mockito.doThrow(failure).when(mockBuilder) + .put(entry.getCode(), entry.getTable(), entry.getKey(), entry.getValue()); + DefaultDataMover dataMover = new DefaultDataMover(); + dataMover.setBusinessHandler(mockHandler); + + assertWriteFails(failure, () -> dataMover.doWriteData(request)); + + Mockito.verify(mockTransaction).rollback(); } @Test @@ -154,8 +250,17 @@ public void testBatchMergeKeepsGraphsIsolated() { } private static void writeBatch(String graph, OpType type, byte[] value) { + writeBatch(graph, PARTITION_ID, type, value); + } + + private static void writeBatch(String graph, int partitionId, OpType type, byte[] value) { + writeBatch(graph, partitionId, KEY_CODE, type, value); + } + + private static void writeBatch(String graph, int partitionId, int code, OpType type, + byte[] value) { Key key = Key.newBuilder() - .setCode(KEY_CODE) + .setCode(code) .setKey(ByteString.copyFrom(SHARED_KEY)) .build(); BatchEntry entry = BatchEntry.newBuilder() @@ -164,11 +269,38 @@ private static void writeBatch(String graph, OpType type, byte[] value) { .setStartKey(key) .setValue(ByteString.copyFrom(value)) .build(); - handler.doBatch(graph, PARTITION_ID, Collections.singletonList(entry)); + handler.doBatch(graph, partitionId, Collections.singletonList(entry)); + } + + private static void put(BusinessHandler.TxBuilder builder, String table, byte[] value) { + builder.put(KEY_CODE, table, SHARED_KEY, value); } private static byte[] read(String graph) { - return handler.doGet(graph, KEY_CODE, VERTEX_TABLE, SHARED_KEY); + return readByCode(graph, KEY_CODE); + } + + private static byte[] readByCode(String graph, int code) { + return handler.doGet(graph, code, VERTEX_TABLE, SHARED_KEY); + } + + private static BatchPutRequest batchPutRequest() { + BatchPutRequest request = new BatchPutRequest(); + request.setGraphName("failed-transfer-graph"); + request.setPartitionId(PARTITION_ID); + request.getEntries() + .add(BatchPutRequest.KV.of(VERTEX_TABLE, KEY_CODE, SHARED_KEY, + "value".getBytes(StandardCharsets.UTF_8))); + return request; + } + + private static void assertWriteFails(RuntimeException expected, Runnable writer) { + try { + writer.run(); + Assert.fail("batch write must propagate its failure"); + } catch (RuntimeException actual) { + Assert.assertSame(expected, actual); + } } private static byte[] longToBytes(long value) { From a0c34a77f607c66da32d2ca5aa6a2d026a8c237a Mon Sep 17 00:00:00 2001 From: imbajin Date: Tue, 25 Aug 2026 21:20:02 +0800 Subject: [PATCH 4/4] fix(store): verify graph removal after truncate - assert the truncated graph no longer returns data - retain the neighboring graph isolation assertion - prevent false-positive truncate regression coverage --- .../org/apache/hugegraph/store/core/BatchGraphIsolationTest.java | 1 + 1 file changed, 1 insertion(+) diff --git a/hugegraph-store/hg-store-test/src/main/java/org/apache/hugegraph/store/core/BatchGraphIsolationTest.java b/hugegraph-store/hg-store-test/src/main/java/org/apache/hugegraph/store/core/BatchGraphIsolationTest.java index ec271edcec..c222557962 100644 --- a/hugegraph-store/hg-store-test/src/main/java/org/apache/hugegraph/store/core/BatchGraphIsolationTest.java +++ b/hugegraph-store/hg-store-test/src/main/java/org/apache/hugegraph/store/core/BatchGraphIsolationTest.java @@ -234,6 +234,7 @@ public void testBatchPutKeepsGraphsIsolated() { Assert.assertArrayEquals(value2, read(graph2)); handler.truncate(graph2, PARTITION_ID); + Assert.assertNull(read(graph2)); Assert.assertArrayEquals(value1, read(graph1)); }