diff --git a/docs/store-graphid-upgrade.md b/docs/store-graphid-upgrade.md new file mode 100644 index 0000000000..a77798deca --- /dev/null +++ b/docs/store-graphid-upgrade.md @@ -0,0 +1,34 @@ +# Store legacy GraphID preflight + +Some older Store writes can have the reserved `FFFE` graph prefix without a +persisted graph-to-ID mapping. Allocating a new graph ID does not rewrite those +rows, so their physical presence does not imply that normal graph reads can see +them. This condition is specific to affected old write paths; it does not mean +that every 1.7 database needs migration. + +Store checks existing graph-data column families before allocating a missing +graph ID, first opening a partition DB during metadata bootstrap, starting a +partition's Raft state machine, or activating a snapshot. +If it finds the reserved prefix, it rejects that operation and reports the +partition and column family. Metadata-only column families are excluded. The +checks use the configured column-family options from normal Store opens. They do +not assign an owner to legacy rows or change the persisted ID format. + +The partition preflight resolves existing DB versions without creating a DB, +allocating a store location or pruning older versions. It runs before partition +task/FSM construction and Raft replay. The first native DB open also validates +every existing version not already queued for removal before constructor pruning +or cache publication. Cached sessions are reused without repeating these scans. +Bootstrap can update Store/Raft location metadata before this check; the refusal +preserves data DB versions, not every file touched by the startup process. + +A rejected snapshot retains the active DB and its index, the incoming snapshot, +and the staged copy for diagnosis. Successful snapshots still use the existing +activation path. An incomplete selected DB version with an older valid DB is +rejected rather than silently skipping its preflight. + +After rejection, preserve the original database, metadata and snapshot evidence +and investigate the affected write/upgrade history before restarting writes or +replay. Do not delete mappings or assign all reserved-prefix rows to one graph: +different graphs may share that prefix. This change provides detection and +refusal, not automatic migration, repair or proof of a complete 1.7 upgrade. diff --git a/hugegraph-store/hg-store-core/src/main/java/org/apache/hugegraph/store/PartitionEngine.java b/hugegraph-store/hg-store-core/src/main/java/org/apache/hugegraph/store/PartitionEngine.java index 85e43fea02..588f0aebb1 100644 --- a/hugegraph-store/hg-store-core/src/main/java/org/apache/hugegraph/store/PartitionEngine.java +++ b/hugegraph-store/hg-store-core/src/main/java/org/apache/hugegraph/store/PartitionEngine.java @@ -22,6 +22,7 @@ import java.io.File; import java.io.IOException; import java.nio.ByteBuffer; +import java.nio.file.Paths; import java.util.ArrayList; import java.util.Collections; import java.util.HashMap; @@ -44,6 +45,8 @@ import org.apache.hugegraph.pd.grpc.MetaTask; import org.apache.hugegraph.pd.grpc.Metapb; import org.apache.hugegraph.pd.raft.RaftReflectionUtil; +import org.apache.hugegraph.rocksdb.access.RocksDBFactory; +import org.apache.hugegraph.rocksdb.access.RocksDBSession; import org.apache.hugegraph.store.business.BusinessHandler; import org.apache.hugegraph.store.business.BusinessHandlerImpl; import org.apache.hugegraph.store.cmd.HgCmdClient; @@ -52,6 +55,7 @@ import org.apache.hugegraph.store.cmd.request.DbCompactionRequest; import org.apache.hugegraph.store.cmd.request.UpdatePartitionRequest; import org.apache.hugegraph.store.listener.PartitionStateListener; +import org.apache.hugegraph.store.meta.GraphIdManager; import org.apache.hugegraph.store.meta.Partition; import org.apache.hugegraph.store.meta.PartitionManager; import org.apache.hugegraph.store.meta.Shard; @@ -175,6 +179,7 @@ public synchronized boolean init(PartitionEngineOptions opts) { return true; } + this.checkLegacyGraphIdData(opts.getGroupId()); log.info("PartitionEngine starting: {}", this); this.taskManager = new TaskManager(storeEngine.getBusinessHandler(), opts.getGroupId()); this.snapshotHandler = new SnapshotHandler(this); @@ -267,6 +272,30 @@ public LogStorage createLogStorage(final String uri, final RaftOptions raftOptio return this.started; } + private void checkLegacyGraphIdData(int partitionId) { + String dbName = BusinessHandlerImpl.getDbName(partitionId); + try (RocksDBSession cached = RocksDBFactory.getInstance().queryGraphDB(dbName)) { + if (cached != null) { + // Use the primary's view, including unflushed writes made with WAL disabled. + GraphIdManager.checkLegacyGraphIdData(cached, partitionId); + return; + } + } + Metapb.PartitionStore location = + this.partitionManager.getStoreMetadata().getPartitionStore(partitionId); + // Looking up or assigning a new store location would write metadata before inspection. + String[] roots = location != null ? new String[]{location.getStoreLocation()} : + this.storeEngine.getOption().getDataPath().split(","); + for (String root : roots) { + String dataPath = Paths.get(root, HgStoreEngineOptions.DB_Path_Prefix).toString(); + String existing = RocksDBSession.resolveExistingDBPath( + dataPath, dbName, this.storeEngine.getCommittedIndex(partitionId)); + if (existing != null) { + GraphIdManager.checkLegacyGraphIdData(existing, partitionId); + } + } + } + public HgStoreEngine getStoreEngine() { return this.storeEngine; } 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 bc446afb31..b3dd9f8c40 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 @@ -101,6 +101,7 @@ import org.apache.hugegraph.store.grpc.Graphpb.ScanPartitionRequest.Request; import org.apache.hugegraph.store.grpc.Graphpb.ScanPartitionRequest.ScanType; import org.apache.hugegraph.store.grpc.query.DeDupOption; +import org.apache.hugegraph.store.meta.GraphIdManager; import org.apache.hugegraph.store.meta.Partition; import org.apache.hugegraph.store.meta.PartitionManager; import org.apache.hugegraph.store.meta.asynctask.AsyncTaskState; @@ -1195,11 +1196,20 @@ public void saveSnapshot(String snapshotPath, String graph, int partId) throws public void loadSnapshot(String snapshotPath, String graph, int partId, long v1) throws HgStoreException { try (RocksDBSession dbSession = getSession(graph, partId)) { - dbSession.loadSnapshot(snapshotPath, v1); + dbSession.loadSnapshot(snapshotPath, v1, stagedPath -> { + try { + GraphIdManager.checkLegacyGraphIdData(stagedPath, partId); + } catch (HgStoreException e) { + throw new DBStoreException(e.getMessage(), e); + } + }); keyCreator.clearCache(partId); factory.destroyGraphDB(dbSession.getGraphName()); } catch (DBStoreException e) { - throw new HgStoreException(HgStoreException.EC_RKDB_IMPORT_SNAPSHOT_FAIL, e.toString()); + HgStoreException failure = + new HgStoreException(HgStoreException.EC_RKDB_IMPORT_SNAPSHOT_FAIL, e.toString()); + failure.initCause(e); + throw failure; } } @@ -1490,7 +1500,7 @@ public RocksDBSession getSession(int partId) throws HgStoreException { long version = HgStoreEngine.getInstance().getCommittedIndex(partId); dbSession = factory.createGraphDB(partitionManager.getDbDataPath(partId, dbName), dbName, - version); + version, path -> GraphIdManager.checkLegacyGraphIdData(path, partId)); if (dbSession == null) { log.info("failed to create a new graph db: {}", dbName); throw new HgStoreException(HgStoreException.EC_RKDB_CREATE_FAIL, 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 76eada307b..95fd438b0e 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 @@ -21,21 +21,35 @@ import java.nio.ByteBuffer; import java.nio.charset.StandardCharsets; +import java.util.ArrayList; import java.util.Arrays; import java.util.Collections; import java.util.HashMap; +import java.util.List; import java.util.Map; import java.util.Set; import java.util.concurrent.ConcurrentHashMap; import java.util.stream.Collectors; +import org.apache.hugegraph.config.HugeConfig; +import org.apache.hugegraph.rocksdb.access.RocksDBFactory; import org.apache.hugegraph.rocksdb.access.RocksDBSession; import org.apache.hugegraph.rocksdb.access.SessionOperator; +import org.apache.hugegraph.store.constant.HugeServerTables; import org.apache.hugegraph.store.meta.base.DBSessionBuilder; import org.apache.hugegraph.store.meta.base.PartitionMetaStore; import org.apache.hugegraph.store.term.Bits; import org.apache.hugegraph.store.util.Asserts; import org.apache.hugegraph.store.util.HgStoreException; +import org.rocksdb.ColumnFamilyDescriptor; +import org.rocksdb.ColumnFamilyHandle; +import org.rocksdb.ColumnFamilyOptions; +import org.rocksdb.DBOptions; +import org.rocksdb.InfoLogLevel; +import org.rocksdb.Logger; +import org.rocksdb.Options; +import org.rocksdb.RocksDB; +import org.rocksdb.RocksDBException; import com.google.protobuf.Int64Value; import com.google.protobuf.InvalidProtocolBufferException; @@ -99,6 +113,9 @@ public long getGraphIdOrCreate(String graphName) { byte[] key = MetadataKeyHelper.getGraphIDKey(graphName); Int64Value id = get(Int64Value.parser(), key); if (id == null) { + try (RocksDBSession session = this.sessionBuilder.getSession(this.partitionId)) { + checkLegacyGraphIdData(session, this.partitionId); + } id = Int64Value.of(getCId(GRAPH_ID_PREFIX, maxGraphID)); if (id.getValue() == -1) { throw new HgStoreException(HgStoreException.EC_FAIL, @@ -260,6 +277,90 @@ public long releaseGraphId(String graphName) { return gid; } + /** + * Reject unresolved rows written with the 1.7 missing-graph sentinel. + * Their graph ownership cannot be inferred from the shared physical prefix. + */ + public static void checkLegacyGraphIdData(RocksDBSession session, int partitionId) { + var lock = session.getCfHandleReadLock(); + lock.lock(); + try { + checkLegacyGraphIdData(session.getDB(), session.getTables(), partitionId); + } catch (RocksDBException e) { + throw new HgStoreException("Can't inspect legacy GraphID data in partition " + partitionId, e); + } finally { + lock.unlock(); + } + } + + public static void checkLegacyGraphIdData(String existingDbPath, int partitionId) { + try (Options options = new Options(); + ColumnFamilyOptions cfOptions = new ColumnFamilyOptions(); + DBOptions dbOptions = new DBOptions(); + Logger logger = new Logger(dbOptions) { + @Override + protected void log(InfoLogLevel level, String message) { + // Keep inspection from creating or replacing the DB's LOG file. + } + }) { + dbOptions.setLogger(logger); + HugeConfig config = RocksDBFactory.getInstance().getHugeConfig(); + if (config != null) { + RocksDBSession.initOptions(config, null, null, cfOptions, cfOptions); + } + List names = RocksDB.listColumnFamilies(options, existingDbPath); + List descriptors = new ArrayList<>(); + for (byte[] name : names) { + descriptors.add(new ColumnFamilyDescriptor(name, cfOptions)); + } + List handles = new ArrayList<>(); + RocksDB db = null; + try { + db = RocksDB.openReadOnly(dbOptions, existingDbPath, descriptors, handles); + Map tables = new HashMap<>(); + for (int i = 0; i < names.size(); i++) { + tables.put(new String(names.get(i), StandardCharsets.UTF_8), handles.get(i)); + } + checkLegacyGraphIdData(db, tables, partitionId); + } finally { + for (ColumnFamilyHandle handle : handles) { + handle.close(); + } + if (db != null) { + db.close(); + } + } + } catch (RocksDBException e) { + throw new HgStoreException("Can't inspect legacy GraphID data in partition " + partitionId + + " at " + existingDbPath, e); + } + } + + private static void checkLegacyGraphIdData(RocksDB db, + Map tables, + int partitionId) throws RocksDBException { + byte[] prefix = {(byte) 0xff, (byte) 0xfe}; + for (String table : HugeServerTables.TABLES) { + ColumnFamilyHandle handle = tables.get(table); + if (HugeServerTables.UNKNOWN_TABLE.equals(table) || handle == null) { + continue; + } + try (var iterator = db.newIterator(handle)) { + iterator.seek(prefix); + iterator.status(); + if (iterator.isValid()) { + byte[] key = iterator.key(); + if (key.length >= prefix.length && key[0] == prefix[0] && key[1] == prefix[1]) { + throw new HgStoreException(HgStoreException.EC_FAIL, + "Legacy GraphID data in partition " + partitionId + + ", table " + table + ", prefix 0xFFFE; " + + "stop upgrade and recover the affected data"); + } + } + } + } + } + /** * To maintain compatibility with affected graphs, ensure the g+v table contains no data * diff --git a/hugegraph-store/hg-store-core/src/main/java/org/apache/hugegraph/store/raft/PartitionStateMachine.java b/hugegraph-store/hg-store-core/src/main/java/org/apache/hugegraph/store/raft/PartitionStateMachine.java index cf74810aec..1ba2e7399c 100644 --- a/hugegraph-store/hg-store-core/src/main/java/org/apache/hugegraph/store/raft/PartitionStateMachine.java +++ b/hugegraph-store/hg-store-core/src/main/java/org/apache/hugegraph/store/raft/PartitionStateMachine.java @@ -222,11 +222,11 @@ public void onSnapshotSave(final SnapshotWriter writer, final Closure done) { @Override public boolean onSnapshotLoad(final SnapshotReader reader) { + final long snapshotIndex; try { RaftOutter.SnapshotMeta meta = reader.load(); if (meta != null) { - this.committedIndex = meta.getLastIncludedIndex(); - log.info("onSnapshotLoad committedIndex = {}", this.committedIndex); + snapshotIndex = meta.getLastIncludedIndex(); } else { log.error("onSnapshotLoad failed to get SnapshotMeta"); return false; @@ -241,8 +241,9 @@ public boolean onSnapshotLoad(final SnapshotReader reader) { return false; } try { - snapshotHandler.onSnapshotLoad(reader, this.committedIndex); - log.info("Raft {} onSnapshotLoad success", groupId); + snapshotHandler.onSnapshotLoad(reader, snapshotIndex); + this.committedIndex = snapshotIndex; + log.info("Raft {} onSnapshotLoad success, committedIndex = {}", groupId, snapshotIndex); return true; } catch (HgStoreException e) { log.error(String.format("Raft %s onSnapshotLoad failed. ", groupId), e); diff --git a/hugegraph-store/hg-store-rocksdb/README.md b/hugegraph-store/hg-store-rocksdb/README.md new file mode 100644 index 0000000000..86c034daae --- /dev/null +++ b/hugegraph-store/hg-store-rocksdb/README.md @@ -0,0 +1,46 @@ + + +# RocksDB path and snapshot validation + +`RocksDBSession.resolveExistingDBPath(dbDataPath, graphName, version)` uses the +same version-directory selection as the constructor, without opening, creating, +pruning or renaming databases. It returns `null` when the selected path has no +`CURRENT` and no existing database version is present. If an incomplete selected +version would hide another existing database, it fails without changing either +path. Directory listing or `CURRENT` inspection failures are rejected; unknown +access or I/O state is not treated as an absent database. Paths queued for removal retain the constructor's existing treatment. +The normal constructor continues to open read/write and prune old versions. + +`RocksDBFactory.createGraphDB(path, name, version, Consumer validator)` +validates existing DB versions before creating an uncached native session. It +uses the existing factory creation lock and cache: cached sessions skip validation. +Each existing candidate not queued for removal is passed to the validator through +the read-only resolver before constructor pruning or cache publication. Validation +failure leaves these DB paths intact. The original overloads keep their behavior; +this generic hook does not depend on Store graph-ID rules. + +`loadSnapshot(snapshotPath, version, Consumer validator)` first copies +and verifies the staging database with the session's configured column-family +options, then passes its path to the validator before +replacing the target version. The validator must inspect it read-only. A rejected +validation preserves active databases, target versions and staging evidence; +the exception cause is retained. The original two-argument overload remains +available and uses a no-op validator. + +These APIs provide generic inspection hooks. They do not validate graph ownership, +migrate historical data, or replace deployment-specific recovery and upgrade checks. diff --git a/hugegraph-store/hg-store-rocksdb/src/main/java/org/apache/hugegraph/rocksdb/access/RocksDBFactory.java b/hugegraph-store/hg-store-rocksdb/src/main/java/org/apache/hugegraph/rocksdb/access/RocksDBFactory.java index 2e8e0bae68..52b0b66cd9 100644 --- a/hugegraph-store/hg-store-rocksdb/src/main/java/org/apache/hugegraph/rocksdb/access/RocksDBFactory.java +++ b/hugegraph-store/hg-store-rocksdb/src/main/java/org/apache/hugegraph/rocksdb/access/RocksDBFactory.java @@ -31,6 +31,7 @@ import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.locks.ReentrantReadWriteLock; +import java.util.function.Consumer; import java.util.stream.Collectors; import org.apache.commons.io.FileUtils; @@ -185,6 +186,11 @@ public RocksDBSession createGraphDB(String dbPath, String dbName) { } public RocksDBSession createGraphDB(String dbPath, String dbName, long version) { + return createGraphDB(dbPath, dbName, version, null); + } + + public RocksDBSession createGraphDB(String dbPath, String dbName, long version, + Consumer existingDatabaseValidator) { if (closing.get()) { throw new RuntimeException("db closed"); } @@ -192,6 +198,10 @@ public RocksDBSession createGraphDB(String dbPath, String dbName, long version) try { RocksDBSession dbSession = dbSessionMap.get(dbName); if (dbSession == null) { + if (existingDatabaseValidator != null) { + RocksDBSession.resolveExistingDBPath(dbPath, dbName, version, + existingDatabaseValidator); + } log.info("create rocksdb for {}", dbName); dbSession = new RocksDBSession(this.hugeConfig, dbPath, dbName, version); dbSessionMap.put(dbName, dbSession); diff --git a/hugegraph-store/hg-store-rocksdb/src/main/java/org/apache/hugegraph/rocksdb/access/RocksDBSession.java b/hugegraph-store/hg-store-rocksdb/src/main/java/org/apache/hugegraph/rocksdb/access/RocksDBSession.java index f4e7605a7f..867b4ef17c 100644 --- a/hugegraph-store/hg-store-rocksdb/src/main/java/org/apache/hugegraph/rocksdb/access/RocksDBSession.java +++ b/hugegraph-store/hg-store-rocksdb/src/main/java/org/apache/hugegraph/rocksdb/access/RocksDBSession.java @@ -21,9 +21,13 @@ import java.io.File; import java.io.IOException; import java.io.Serializable; +import java.io.UncheckedIOException; import java.nio.charset.StandardCharsets; import java.nio.file.Files; +import java.nio.file.NoSuchFileException; +import java.nio.file.Path; import java.nio.file.Paths; +import java.nio.file.attribute.BasicFileAttributes; import java.util.ArrayList; import java.util.Arrays; import java.util.List; @@ -33,7 +37,9 @@ import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.atomic.AtomicInteger; import java.util.concurrent.locks.ReentrantReadWriteLock; +import java.util.function.Consumer; import java.util.stream.Collectors; +import java.util.stream.Stream; import org.apache.commons.io.FileUtils; import org.apache.hugegraph.config.HugeConfig; @@ -73,11 +79,11 @@ public class RocksDBSession implements AutoCloseable, Cloneable { private static final int CPUS = Runtime.getRuntime().availableProcessors(); + private static final String TEMP_SUFFIX = "_temp_"; final Statistics rocksDbStats; final WriteOptions writeOptions; final AtomicInteger refCount; final AtomicBoolean shutdown; - final String tempSuffix = "_temp_"; private final transient String graphName; private final HugeConfig hugeConfig; private final ReentrantReadWriteLock cfHandleLock; @@ -300,7 +306,44 @@ public ReentrantReadWriteLock.ReadLock getCfHandleReadLock() { return this.cfHandleLock.readLock(); } + /** + * Resolve the path the constructor would select without opening or pruning databases. + * Returns null for a new database without CURRENT. An incomplete selected version with + * another existing database is rejected rather than treated as a new database. + */ + public static String resolveExistingDBPath(String dbDataPath, String graphName, long version) { + return resolveExistingDBPath(dbDataPath, graphName, version, null); + } + + public static String resolveExistingDBPath(String dbDataPath, String graphName, long version, + Consumer existingDatabaseValidator) { + String path = Paths.get(dbDataPath, graphName).toString(); + String selected = findLatestDBPath(path, version, false, existingDatabaseValidator); + return hasCurrentFile(selected) ? selected : null; + } + + private static boolean hasCurrentFile(String dbPath) { + try { + return Files.readAttributes(Paths.get(dbPath, "CURRENT"), BasicFileAttributes.class).isRegularFile(); + } catch (NoSuchFileException e) { + return false; + } catch (IOException e) { + throw new DBStoreException(String.format("Unable to inspect database path %s", dbPath), e); + } + } + private String findLatestDBPath(String path, long version) { + String selected = findLatestDBPath(path, version, true); + log.info("{} latest db path {}", this.graphName, selected); + return selected; + } + + private static String findLatestDBPath(String path, long version, boolean prune) { + return findLatestDBPath(path, version, prune, null); + } + + private static String findLatestDBPath(String path, long version, boolean prune, + Consumer existingDatabaseValidator) { File file = new File(path); int strIndex = file.getName().indexOf("_"); @@ -313,7 +356,18 @@ private String findLatestDBPath(String path, long version) { String prefix = defaultName + "_"; File parentFile = new File(file.getParent()); ArrayList> dbs = new ArrayList<>(); - File[] files = parentFile.listFiles(); + File[] files; + if (prune) { + files = parentFile.listFiles(); + } else { + try (Stream paths = Files.list(parentFile.toPath())) { + files = paths.map(Path::toFile).toArray(File[]::new); + } catch (NoSuchFileException e) { + files = null; + } catch (IOException | UncheckedIOException e) { + throw new DBStoreException(String.format("Unable to list database directory %s", parentFile), e); + } + } if (files != null) { dbs.ensureCapacity(files.length); // search all db path @@ -322,7 +376,7 @@ private String findLatestDBPath(String path, long version) { if (!name.startsWith(prefix) && !name.equals(defaultName)) { continue; } - if (name.endsWith(tempSuffix)) { + if (name.endsWith(TEMP_SUFFIX)) { continue; } long v1 = -1L; @@ -345,6 +399,7 @@ private String findLatestDBPath(String path, long version) { RocksDBFactory factory = RocksDBFactory.getInstance(); // get last index db path String latestDBPath = ""; + boolean existingDatabase = false; if (!dbs.isEmpty()) { dbs.sort((o1, o2) -> o1.getKey().equals(o2.getKey()) ? @@ -363,11 +418,18 @@ private String findLatestDBPath(String path, long version) { String.format("%s_%d_%d", defaultName, pair.getKey(), pair.getValue()); } String curDBPath = Paths.get(parentFile.getPath(), curDBName).toString(); + if (!prune && !factory.findPathInRemovedList(curDBPath) && + hasCurrentFile(curDBPath)) { + existingDatabase = true; + if (existingDatabaseValidator != null) { + existingDatabaseValidator.accept(curDBPath); + } + } if (i == dbCount - 1) { latestDBPath = curDBPath; } else { // delete old db, do not delete files in the deletion queue - if (!factory.findPathInRemovedList(curDBPath)) { + if (prune && !factory.findPathInRemovedList(curDBPath)) { try { FileUtils.deleteDirectory(new File(curDBPath)); log.info("delete old dbpath {}", curDBPath); @@ -386,8 +448,10 @@ private String findLatestDBPath(String path, long version) { Paths.get(parentFile.getPath(), String.format("%s_%d", defaultName, version)) .toString(); } - // - log.info("{} latest db path {}", this.graphName, latestDBPath); + if (!prune && existingDatabase && !hasCurrentFile(latestDBPath)) { + throw new DBStoreException("Selected database path %s has no CURRENT while another version exists", + latestDBPath); + } return latestDBPath; } @@ -697,6 +761,7 @@ private boolean verifySnapshot(String snapshotPath) { try { try (final Options options = new Options(); final ColumnFamilyOptions cfOptions = new ColumnFamilyOptions()) { + RocksDBSession.initOptions(this.hugeConfig, null, null, cfOptions, cfOptions); List columnFamilyDescriptorList = new ArrayList<>(); List columnFamilyBytes = RocksDB.listColumnFamilies(options, snapshotPath); @@ -711,15 +776,27 @@ private boolean verifySnapshot(String snapshotPath) { } List columnFamilyHandleList = new ArrayList<>(); - try (final DBOptions dbOptions = new DBOptions(); - final RocksDB db = RocksDB.openReadOnly(dbOptions, snapshotPath, - columnFamilyDescriptorList, - columnFamilyHandleList)) { - for (ColumnFamilyHandle handle : columnFamilyHandleList) { - if (handle == null) { - log.error("verifySnapshot some ColumnFamilyHandle is null in {}", - snapshotPath); - return false; + try (final DBOptions dbOptions = new DBOptions()) { + RocksDB db = null; + try { + db = RocksDB.openReadOnly(dbOptions, snapshotPath, + columnFamilyDescriptorList, + columnFamilyHandleList); + for (ColumnFamilyHandle handle : columnFamilyHandleList) { + if (handle == null) { + log.error("verifySnapshot some ColumnFamilyHandle is null in {}", + snapshotPath); + return false; + } + } + } finally { + for (ColumnFamilyHandle handle : columnFamilyHandleList) { + if (handle != null) { + handle.close(); + } + } + if (db != null) { + db.close(); } } } @@ -733,6 +810,16 @@ private boolean verifySnapshot(String snapshotPath) { } public void loadSnapshot(String snapshotPath, long v1) throws DBStoreException { + this.loadSnapshot(snapshotPath, v1, path -> { }); + } + + /** + * Validate the verified staging directory before replacing a database version. The validator + * must inspect it read-only. Rejection preserves existing databases and the staging directory. + */ + public void loadSnapshot(String snapshotPath, long v1, + Consumer validator) throws DBStoreException { + E.checkArgumentNotNull(validator, "Snapshot validator can't be null"); long startTime = System.currentTimeMillis(); log.info("begin load snapshot from {}", snapshotPath); cfHandleLock.writeLock().lock(); @@ -750,7 +837,7 @@ public void loadSnapshot(String snapshotPath, long v1) throws DBStoreException { String defaultName = dbFile.getName().split("_")[0]; String defaultPath = Paths.get(parent, defaultName).toString(); String newDBPath = String.format("%s_%d", defaultPath, v1); - String tempDBPath = newDBPath + tempSuffix; + String tempDBPath = newDBPath + TEMP_SUFFIX; // first link snapshot to temp dir try { @@ -790,6 +877,8 @@ public void loadSnapshot(String snapshotPath, long v1) throws DBStoreException { String.format("failed to verify snapshot %s", tempDBPath)); } + validator.accept(tempDBPath); + // move temp to newDBPath try { FileUtils.deleteDirectory(new File(newDBPath)); @@ -804,7 +893,7 @@ public void loadSnapshot(String snapshotPath, long v1) throws DBStoreException { } catch (final Exception e) { log.error("failed to load snapshot from {}", snapshotPath); throw new DBStoreException( - String.format("Fail to write snapshot at path %s", snapshotPath)); + String.format("Fail to write snapshot at path %s", snapshotPath), e); } finally { cfHandleLock.writeLock().unlock(); } diff --git a/hugegraph-store/hg-store-rocksdb/src/test/java/org/apache/hugegraph/rocksdb/access/RocksDBSessionTest.java b/hugegraph-store/hg-store-rocksdb/src/test/java/org/apache/hugegraph/rocksdb/access/RocksDBSessionTest.java index 3f80191e4c..2245452490 100644 --- a/hugegraph-store/hg-store-rocksdb/src/test/java/org/apache/hugegraph/rocksdb/access/RocksDBSessionTest.java +++ b/hugegraph-store/hg-store-rocksdb/src/test/java/org/apache/hugegraph/rocksdb/access/RocksDBSessionTest.java @@ -17,26 +17,59 @@ package org.apache.hugegraph.rocksdb.access; +import static org.junit.Assert.assertArrayEquals; import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertFalse; +import static org.junit.Assert.assertNull; +import static org.junit.Assert.assertSame; +import static org.junit.Assert.assertThrows; import static org.junit.Assert.assertTrue; +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.nio.file.attribute.PosixFilePermission; +import java.nio.file.attribute.PosixFilePermissions; +import java.security.MessageDigest; +import java.util.ArrayList; import java.util.Arrays; import java.util.HashMap; +import java.util.List; import java.util.Map; +import java.util.Set; +import java.util.TreeMap; +import java.util.concurrent.atomic.AtomicInteger; +import java.util.concurrent.atomic.AtomicReference; +import java.util.stream.Stream; import org.apache.commons.configuration2.MapConfiguration; import org.apache.hugegraph.config.HugeConfig; import org.apache.hugegraph.config.OptionSpace; +import org.junit.Assume; import org.junit.BeforeClass; +import org.junit.Rule; +import org.junit.Test; +import org.junit.rules.TemporaryFolder; +import org.rocksdb.Checkpoint; +import org.rocksdb.ColumnFamilyDescriptor; +import org.rocksdb.ColumnFamilyHandle; +import org.rocksdb.ColumnFamilyOptions; +import org.rocksdb.DBOptions; +import org.rocksdb.InfoLogLevel; +import org.rocksdb.Logger; +import org.rocksdb.Options; +import org.rocksdb.RocksDB; public class RocksDBSessionTest { private final String graphName = "testDummy"; + @Rule + public TemporaryFolder temporary = new TemporaryFolder(); + @BeforeClass public static void init() { OptionSpace.register("rocksdb", @@ -51,6 +84,240 @@ public static void init() { rFactory.setHugeConfig(hConfig); } + @Test + public void resolveExistingPathPreservesVersionsAndConstructorStillPrunes() throws Exception { + Path root = this.temporary.newFolder("versions").toPath(); + for (String name : new String[]{"db", "db_2", "db_2_3", "db_4_1", "db_99_temp_"}) { + writeNative(root.resolve(name), bytes(name)); + } + Map before = fileHashes(root); + assertEquals(root.resolve("db_4_1").toString(), + RocksDBSession.resolveExistingDBPath(root.toString(), "db", 5L)); + assertEquals(before, fileHashes(root)); + assertArrayEquals(bytes("db_4_1"), readNative(root.resolve("db_4_1"))); + try (RocksDBSession session = new RocksDBSession(smallConfig(), root.toString(), "db", 5L)) { + assertEquals(root.resolve("db_4_1").toString(), session.getDbPath()); + assertArrayEquals(bytes("db_4_1"), session.getDB().get(bytes("key"))); + assertFalse(Files.exists(root.resolve("db"))); + assertFalse(Files.exists(root.resolve("db_2"))); + assertFalse(Files.exists(root.resolve("db_2_3"))); + assertTrue(Files.exists(root.resolve("db_99_temp_"))); + } + } + + @Test + public void resolveNewPathDoesNotCreateDatabase() throws Exception { + Path root = this.temporary.newFolder("missing").toPath(); + Path missing = root.resolve("not-created"); + assertNull(RocksDBSession.resolveExistingDBPath(missing.toString(), "db", 0L)); + assertFalse(Files.exists(missing)); + Files.createDirectory(root.resolve("db")); + assertNull(RocksDBSession.resolveExistingDBPath(root.toString(), "db", 0L)); + assertTrue(Files.isDirectory(root.resolve("db"))); + assertFalse(Files.exists(root.resolve("db/CURRENT"))); + } + + @Test + public void resolveIncompleteNewerVersionPreservesExistingDatabase() throws Exception { + Path root = this.temporary.newFolder("incomplete").toPath(); + writeNative(root.resolve("db_1"), bytes("existing")); + Files.createDirectory(root.resolve("db_2")); + Map before = fileHashes(root); + assertThrows(DBStoreException.class, + () -> RocksDBSession.resolveExistingDBPath(root.toString(), "db", 2L)); + assertTrue(Files.isDirectory(root.resolve("db_2"))); + assertEquals(before, fileHashes(root)); + assertArrayEquals(bytes("existing"), readNative(root.resolve("db_1"))); + } + + @Test + public void resolveRejectsUnreadableParent() throws Exception { + Path root = this.temporary.newFolder("unreadable-parent").toPath(); + assertRestrictedPathRejected(root, root, "-wx------", root); + } + + @Test + public void resolveRejectsUnsearchableDatabase() throws Exception { + Path root = this.temporary.newFolder("unsearchable-db").toPath(); + assertRestrictedPathRejected(root, root.resolve("db_2"), "r--------", root); + } + + @Test + public void resolveRejectsUnknownMissingPathUnderDeniedAncestor() throws Exception { + Path root = this.temporary.newFolder("denied-ancestor").toPath(); + assertRestrictedPathRejected(root, root, "---------", root.resolve("missing")); + } + + private static void assertRestrictedPathRejected(Path root, Path restricted, + String mode, Path requested) throws Exception { + Assume.assumeTrue("POSIX permissions are required for the access regression", + Files.getFileStore(root).supportsFileAttributeView("posix")); + Path database = root.resolve("db_2"); + writeNative(database, bytes("existing")); + Map before = fileHashes(root); + Set original = Files.getPosixFilePermissions(restricted); + try { + Files.setPosixFilePermissions(restricted, PosixFilePermissions.fromString(mode)); + boolean denied = restricted.equals(database) ? !Files.isExecutable(restricted) : + !Files.isReadable(restricted); + Assume.assumeTrue("Permission denial is required; root DAC override may bypass it", denied); + DBStoreException failure = assertThrows(DBStoreException.class, + () -> RocksDBSession.resolveExistingDBPath(requested.toString(), "db", 2L)); + assertTrue(failure.getCause() instanceof IOException); + } finally { + Files.setPosixFilePermissions(restricted, original); + } + assertEquals(before, fileHashes(root)); + assertArrayEquals(bytes("existing"), readNative(database)); + assertFalse(Files.exists(root.resolve("db"))); + assertFalse(Files.exists(root.resolve("missing"))); + } + + @Test + public void rejectedSnapshotPreservesActiveVersionAndStaging() throws Exception { + Path root = this.temporary.newFolder("rejected").toPath(); + Path snapshot = root.resolve("snapshot"); + writeNative(root.resolve("producer"), bytes("candidate")); + createNativeSnapshot(root.resolve("producer"), snapshot); + Path target = root.resolve("db_20"); + Path staging = root.resolve("db_20_temp_"); + IllegalStateException rejection = new IllegalStateException("candidate content rejected"); + AtomicInteger validated = new AtomicInteger(); + AtomicReference> stagedBefore = new AtomicReference<>(); + try (RocksDBSession active = new RocksDBSession(smallConfig(), root.toString(), "db", 0L)) { + active.getDB().put(bytes("key"), bytes("active")); + writeNative(target, bytes("previous-version")); + String activePath = active.getDbPath(); + long sequence = active.getLatestSequenceNumber(); + Map activeBefore = fileHashes(Path.of(activePath)); + Map targetBefore = fileHashes(target); + Map snapshotBefore = fileHashes(snapshot); + DBStoreException failed = assertThrows(DBStoreException.class, () -> + active.loadSnapshot(snapshot.toString(), 20L, path -> { + assertEquals(staging.toString(), path); + try { + assertArrayEquals(bytes("candidate"), readNative(Path.of(path))); + stagedBefore.set(fileHashes(Path.of(path))); + } catch (Exception e) { + throw new AssertionError(e); + } + validated.incrementAndGet(); + throw rejection; + })); + assertSame(rejection, failed.getCause()); + assertEquals(1, validated.get()); + assertEquals(activePath, active.getDbPath()); + assertEquals(sequence, active.getLatestSequenceNumber()); + assertArrayEquals(bytes("active"), active.getDB().get(bytes("key"))); + assertEquals(activeBefore, fileHashes(Path.of(activePath))); + assertEquals(targetBefore, fileHashes(target)); + assertEquals(snapshotBefore, fileHashes(snapshot)); + assertArrayEquals(bytes("previous-version"), readNative(target)); + assertArrayEquals(bytes("candidate"), readNative(staging)); + assertEquals(stagedBefore.get(), fileHashes(staging)); + } + } + + @Test + public void acceptedSnapshotAndOriginalOverloadActivateNativeData() throws Exception { + Path root = this.temporary.newFolder("accepted").toPath(); + Path snapshot = root.resolve("snapshot"); + writeNative(root.resolve("producer"), bytes("candidate")); + createNativeSnapshot(root.resolve("producer"), snapshot); + AtomicInteger validated = new AtomicInteger(); + try (RocksDBSession active = new RocksDBSession(smallConfig(), root.toString(), "db", 0L)) { + active.loadSnapshot(snapshot.toString(), 20L, path -> { + try { + assertArrayEquals(bytes("candidate"), readNative(Path.of(path))); + } catch (Exception e) { + throw new AssertionError(e); + } + validated.incrementAndGet(); + }); + assertEquals(1, validated.get()); + assertFalse(Files.exists(root.resolve("db_20_temp_"))); + assertArrayEquals(bytes("candidate"), readNative(root.resolve("db_20"))); + active.loadSnapshot(snapshot.toString(), 21L); + assertArrayEquals(bytes("candidate"), readNative(root.resolve("db_21"))); + } + try (RocksDBSession reopened = new RocksDBSession(smallConfig(), root.toString(), "db", 21L)) { + assertEquals(root.resolve("db_21").toString(), reopened.getDbPath()); + assertArrayEquals(bytes("candidate"), reopened.getDB().get(bytes("key"))); + } + } + + private static HugeConfig smallConfig() { + Map values = new HashMap<>(); + values.put("rocksdb.optimize_mode", false); + values.put("rocksdb.write_buffer_size", 65536L); + values.put("rocksdb.max_background_jobs", 1); + values.put("rocksdb.max_subcompactions", 1); + values.put("rocksdb.log_level", "ERROR"); + return new HugeConfig(new MapConfiguration(values)); + } + + private static byte[] bytes(String value) { + return value.getBytes(StandardCharsets.UTF_8); + } + + private static void writeNative(Path path, byte[] value) throws Exception { + try (Options options = new Options().setCreateIfMissing(true).setWriteBufferSize(65536L); + RocksDB db = RocksDB.open(options, path.toString())) { + db.put(bytes("key"), value); + } + } + + private static void createNativeSnapshot(Path source, Path snapshot) throws Exception { + try (Options options = new Options(); + RocksDB db = RocksDB.open(options, source.toString()); + Checkpoint checkpoint = Checkpoint.create(db)) { + checkpoint.createCheckpoint(snapshot.toString()); + } + } + + private static byte[] readNative(Path path) throws Exception { + try (Options options = new Options(); + ColumnFamilyOptions cfOptions = new ColumnFamilyOptions(); + DBOptions dbOptions = new DBOptions(); + Logger logger = new Logger(dbOptions) { + @Override + protected void log(InfoLogLevel level, String message) { + // Native readback must not create or replace the DB's LOG file. + } + }) { + dbOptions.setLogger(logger); + List descriptors = new ArrayList<>(); + for (byte[] name : RocksDB.listColumnFamilies(options, path.toString())) { + descriptors.add(new ColumnFamilyDescriptor(name, cfOptions)); + } + List handles = new ArrayList<>(); + try (RocksDB db = RocksDB.openReadOnly(dbOptions, path.toString(), descriptors, handles)) { + try { + return db.get(handles.get(0), bytes("key")); + } finally { + for (ColumnFamilyHandle handle : handles) { + handle.close(); + } + } + } + } + } + + private static Map fileHashes(Path root) throws Exception { + Map hashes = new TreeMap<>(); + try (Stream paths = Files.walk(root)) { + for (Path path : paths.filter(Files::isRegularFile).collect(java.util.stream.Collectors.toList())) { + byte[] digest = MessageDigest.getInstance("SHA-256").digest(Files.readAllBytes(path)); + StringBuilder encoded = new StringBuilder(); + for (byte value : digest) { + encoded.append(String.format("%02x", value)); + } + hashes.put(root.relativize(path).toString(), encoded.toString()); + } + } + return hashes; + } + private static byte[] intToBytesForPartId(int v) { short s = (short) v; ByteBuffer buffer = ByteBuffer.allocate(Short.BYTES).order(ByteOrder.BIG_ENDIAN); diff --git a/hugegraph-store/hg-store-test/pom.xml b/hugegraph-store/hg-store-test/pom.xml index 17dbdb32ae..7feeef0eda 100644 --- a/hugegraph-store/hg-store-test/pom.xml +++ b/hugegraph-store/hg-store-test/pom.xml @@ -242,6 +242,7 @@ **/PartitionEngineErrorTest.java **/BatchGraphIsolationTest.java **/GraphIDManagerTest.java + **/GraphIdUpgradePreflightTest.java **/SchemaDriverTest.java diff --git a/hugegraph-store/hg-store-test/src/main/java/org/apache/hugegraph/store/core/store/GraphIdUpgradePreflightTest.java b/hugegraph-store/hg-store-test/src/main/java/org/apache/hugegraph/store/core/store/GraphIdUpgradePreflightTest.java new file mode 100644 index 0000000000..91bd158b28 --- /dev/null +++ b/hugegraph-store/hg-store-test/src/main/java/org/apache/hugegraph/store/core/store/GraphIdUpgradePreflightTest.java @@ -0,0 +1,826 @@ +/* + * 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.store; + +import static org.junit.Assert.assertArrayEquals; +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertFalse; +import static org.junit.Assert.assertNull; +import static org.junit.Assert.assertNotNull; +import static org.junit.Assert.assertSame; +import static org.junit.Assert.assertThrows; +import static org.junit.Assert.assertTrue; +import static org.mockito.ArgumentMatchers.anyInt; +import static org.mockito.ArgumentMatchers.anyString; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.when; + +import java.lang.reflect.Field; +import java.nio.file.Files; +import java.nio.file.Path; +import java.security.MessageDigest; +import java.util.ArrayList; +import java.util.Arrays; +import java.util.Collections; +import java.util.HashMap; +import java.util.LinkedHashMap; +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 java.util.stream.Stream; + +import org.apache.hugegraph.config.HugeConfig; +import org.apache.hugegraph.rocksdb.access.RocksDBSession; +import org.apache.hugegraph.rocksdb.access.RocksDBFactory; +import org.apache.hugegraph.rocksdb.access.SessionOperator; +import org.apache.hugegraph.store.HgStoreEngine; +import org.apache.hugegraph.store.PartitionEngine; +import org.apache.hugegraph.store.business.BusinessHandlerImpl; +import org.apache.hugegraph.store.meta.PartitionManager; +import org.apache.hugegraph.store.meta.ShardGroup; +import org.apache.hugegraph.store.meta.StoreMetadata; +import org.apache.hugegraph.store.meta.Store; +import org.apache.hugegraph.store.meta.GraphIdManager; +import org.apache.hugegraph.store.meta.MetadataKeyHelper; +import org.apache.hugegraph.store.options.HgStoreEngineOptions; +import org.apache.hugegraph.store.options.PartitionEngineOptions; +import org.apache.hugegraph.store.raft.PartitionStateMachine; +import org.apache.hugegraph.store.pd.PdProvider; +import org.apache.hugegraph.store.snapshot.SnapshotHandler; +import org.apache.hugegraph.store.term.Bits; +import org.apache.hugegraph.store.util.HgStoreException; +import org.apache.hugegraph.pd.grpc.Metapb; +import org.junit.BeforeClass; +import org.junit.Rule; +import org.junit.Test; +import org.junit.rules.TemporaryFolder; +import org.rocksdb.Checkpoint; +import org.rocksdb.CompactRangeOptions; +import org.rocksdb.CompressionType; +import org.rocksdb.FlushOptions; +import org.rocksdb.ColumnFamilyDescriptor; +import org.rocksdb.ColumnFamilyHandle; +import org.rocksdb.ColumnFamilyOptions; +import org.rocksdb.DBOptions; +import org.rocksdb.Options; +import org.rocksdb.RocksDB; +import org.rocksdb.RocksDBException; + +import com.google.protobuf.Int64Value; + +import com.alipay.sofa.jraft.Status; +import com.alipay.sofa.jraft.closure.ClosureQueueImpl; +import com.alipay.sofa.jraft.closure.LoadSnapshotClosure; +import com.alipay.sofa.jraft.core.FSMCallerImpl; +import com.alipay.sofa.jraft.core.NodeImpl; +import com.alipay.sofa.jraft.core.NodeMetrics; +import com.alipay.sofa.jraft.entity.LogId; +import com.alipay.sofa.jraft.entity.RaftOutter; +import com.alipay.sofa.jraft.option.FSMCallerOptions; +import com.alipay.sofa.jraft.storage.LogManager; +import com.alipay.sofa.jraft.storage.snapshot.SnapshotReader; + +public class GraphIdUpgradePreflightTest { + + private static final int PARTITION_ID = 0; + private static final String DB_NAME = "00000"; + private static final String[] TABLES = {"g+v", "g+oe", "g+ie", "g+index"}; + private static final byte[] VALUE = {42}; + + @Rule + public TemporaryFolder temporary = new TemporaryFolder(); + + @BeforeClass + public static void initRocksdb() { + Map config = new HashMap<>(); + config.put("rocksdb.write_buffer_size", "1048576"); + config.put("rocksdb.bloom_filter_bits_per_key", "10"); + BusinessHandlerImpl.initRocksdb(config, null); + } + + @Test + public void testNativePreflightFailurePreservesCauseAndFiles() throws Exception { + Path data = this.temporary.newFolder("native-preflight-failure").toPath(); + writeDatabase(data, DB_NAME, 1); + Path database = data.resolve(DB_NAME); + Files.writeString(database.resolve("CURRENT"), "invalid-current"); + Map before = fileHashes(data); + RocksDBException nativeFailure; + try (Options options = new Options()) { + nativeFailure = assertThrows(RocksDBException.class, () -> { + try (RocksDB ignored = RocksDB.openReadOnly(options, database.toString())) { + // Close any unexpectedly successful open before reporting failure. + } + }); + } + HgStoreException rejected = assertThrows(HgStoreException.class, () -> + GraphIdManager.checkLegacyGraphIdData(database.toString(), PARTITION_ID)); + assertEquals(HgStoreException.EC_FAIL, rejected.getCode()); + assertEquals(before, fileHashes(data)); + assertTrue("Expected native cause " + nativeFailure.getStatus().getCode() + + ", got " + rejected.getCause(), + rejected.getCause() instanceof RocksDBException); + assertEquals(nativeFailure.getStatus().getCode(), + ((RocksDBException) rejected.getCause()).getStatus().getCode()); + } + + @Test + public void testEffectiveOptionsPreserveLevelSevenBootstrap() throws Exception { + RocksDBFactory factory = RocksDBFactory.getInstance(); + HugeConfig previous = factory.getHugeConfig(); + try { + useEightLevelConfiguration(); + Path location = this.temporary.newFolder("eight-level-bootstrap").toPath(); + Path data = Files.createDirectory(location.resolve("db")); + writeEightLevelDatabase(data.resolve(DB_NAME + "_1"), 1); + // The normal session can read this legal configuration before the preflight. + try (RocksDBSession normal = session(data, DB_NAME)) { + assertRows(normal, 1); + } + Map before = fileHashes(data); + GraphIdManager.checkLegacyGraphIdData(data.resolve(DB_NAME + "_1").toString(), PARTITION_ID); + assertEquals(before, fileHashes(data)); + try (Bootstrap bootstrap = new Bootstrap(location)) { + bootstrap.manager.loadPartition(); + assertTrue(bootstrap.manager.hasPartition("test", PARTITION_ID)); + assertEquals(1, bootstrap.handler.getKeyCreator().getGraphId(PARTITION_ID, "test")); + for (String table : TABLES) { + assertArrayEquals(VALUE, bootstrap.handler.doGet("test", 0, table, new byte[]{1})); + } + GraphIdManager ids = new GraphIdManager(bootstrap.handler::getSession, PARTITION_ID); + assertTrue(ids.getGraphIdOrCreate("additional") < 65534L); + } + } finally { + factory.setHugeConfig(previous); + } + } + + @Test + public void testEffectiveOptionsRejectLegacyLevelSevenData() throws Exception { + RocksDBFactory factory = RocksDBFactory.getInstance(); + HugeConfig previous = factory.getHugeConfig(); + try { + useEightLevelConfiguration(); + Path location = this.temporary.newFolder("eight-level-legacy").toPath(); + Path data = Files.createDirectory(location.resolve("db")); + writeEightLevelDatabase(data.resolve(DB_NAME + "_1"), 65534); + Map before = fileHashes(data); + HgStoreException rejected = assertThrows(HgStoreException.class, () -> + GraphIdManager.checkLegacyGraphIdData(data.resolve(DB_NAME + "_1").toString(), PARTITION_ID)); + assertTrue(rejected.getMessage().contains("prefix 0xFFFE")); + assertEquals(before, fileHashes(data)); + try (Bootstrap bootstrap = new Bootstrap(location)) { + assertThrows(HgStoreException.class, bootstrap.manager::loadPartition); + assertEquals(before, fileHashes(data)); + assertFalse(factory.getGraphNames().contains(DB_NAME)); + } + } finally { + factory.setHugeConfig(previous); + } + } + + @Test + public void testEffectiveOptionsAcceptLevelSevenSnapshot() throws Exception { + RocksDBFactory factory = RocksDBFactory.getInstance(); + HugeConfig previous = factory.getHugeConfig(); + try { + useEightLevelConfiguration(); + Path seed = this.temporary.newFolder("eight-level-seed").toPath(); + writeEightLevelDatabase(seed.resolve("seed"), 2); + Path snapshot = seed.resolve("snapshot"); + try (RocksDBSession source = session(seed, "seed"); + Checkpoint checkpoint = Checkpoint.create(source.getDB())) { + checkpoint.createCheckpoint(snapshot.toString()); + } + Map incoming = fileHashes(snapshot); + Path location = this.temporary.newFolder("eight-level-restore").toPath(); + Path data = Files.createDirectory(location.resolve("db")); + writeEightLevelDatabase(data.resolve(DB_NAME + "_1"), 1); + try (Bootstrap bootstrap = new Bootstrap(location)) { + bootstrap.handler.loadSnapshot(snapshot.toString(), "", PARTITION_ID, 5L); + try (RocksDBSession restored = bootstrap.handler.getSession(PARTITION_ID)) { + assertRows(restored, 2); + assertEquals(2, bootstrap.handler.getKeyCreator().getGraphId(PARTITION_ID, "test")); + } + } + assertEquals(incoming, fileHashes(snapshot)); + } finally { + factory.setHugeConfig(previous); + } + } + + private static void useEightLevelConfiguration() { + Map config = new HashMap<>(); + config.put("rocksdb.write_buffer_size", "1048576"); + config.put("rocksdb.bloom_filter_bits_per_key", "10"); + config.put("rocksdb.num_levels", "8"); + config.put("rocksdb.compression_per_level", "[none,none,none,none,none,none,none,none]"); + BusinessHandlerImpl.initRocksdb(config, null); + } + + private static void writeEightLevelDatabase(Path path, int graphId) throws Exception { + List handles = new ArrayList<>(); + try (DBOptions options = new DBOptions().setCreateIfMissing(true).setCreateMissingColumnFamilies(true); + ColumnFamilyOptions cfOptions = new ColumnFamilyOptions().setNumLevels(8) + .setCompressionPerLevel(Collections.nCopies(8, CompressionType.NO_COMPRESSION)) + .setDisableAutoCompactions(true).setWriteBufferSize(1048576); + FlushOptions flush = new FlushOptions().setWaitForFlush(true); + CompactRangeOptions compact = new CompactRangeOptions().setChangeLevel(true).setTargetLevel(7)) { + List descriptors = new ArrayList<>(); + descriptors.add(new ColumnFamilyDescriptor(RocksDB.DEFAULT_COLUMN_FAMILY, cfOptions)); + for (String table : TABLES) { + descriptors.add(new ColumnFamilyDescriptor(table.getBytes(java.nio.charset.StandardCharsets.UTF_8), + cfOptions)); + } + RocksDB database = null; + try { + database = RocksDB.open(options, path.toString(), descriptors, handles); + for (int i = 1; i < handles.size(); i++) { + database.put(handles.get(i), key(graphId), VALUE); + } + if (graphId != 65534) { + database.put(handles.get(0), MetadataKeyHelper.getGraphIDKey("test"), + Int64Value.of(graphId).toByteArray()); + database.put(handles.get(0), MetadataKeyHelper.getPartitionKey("test", PARTITION_ID), + Metapb.Partition.newBuilder().setId(PARTITION_ID).setGraphName("test") + .build().toByteArray()); + } + database.flush(flush, handles); + for (int i = 1; i < handles.size(); i++) { + database.compactRange(handles.get(i), null, null, compact); + assertFalse("A real L7 SST is required for this native compatibility regression", + database.getColumnFamilyMetaData(handles.get(i)).levels().get(7).files().isEmpty()); + } + } finally { + handles.forEach(ColumnFamilyHandle::close); + if (database != null) { + database.close(); + } + } + } + } + + @Test + public void testBootstrapPreservesOlderDatabaseWhenLatestHasNoCurrent() throws Exception { + Path location = this.temporary.newFolder("bootstrap").toPath(); + Path data = Files.createDirectory(location.resolve("db")); + writeDatabase(data, DB_NAME + "_1", 1); + Files.createDirectory(data.resolve(DB_NAME + "_2")); + Files.write(data.resolve(DB_NAME + "_2/incomplete-copy"), new byte[]{7}); + Map before = fileHashes(data); + try (Bootstrap bootstrap = new Bootstrap(location)) { + Throwable failure = null; + try { + bootstrap.manager.loadPartition(); + } catch (Throwable caught) { + failure = caught; + } + assertNotNull("Bootstrap must reject before creating/pruning native databases", failure); + assertTrue(failure.getMessage().contains("no CURRENT")); + assertEquals(before, fileHashes(data)); + assertFalse(RocksDBFactory.getInstance().getGraphNames().contains(DB_NAME)); + } + } + + @Test + public void testBootstrapPreservesLegacyOlderVersionBeforePruning() throws Exception { + Path location = this.temporary.newFolder("bootstrap-older-legacy").toPath(); + Path data = Files.createDirectory(location.resolve("db")); + writeDatabase(data, DB_NAME + "_1", 65534); + writeDatabase(data, DB_NAME + "_2", 1); + Map before = fileHashes(data); + try (Bootstrap bootstrap = new Bootstrap(location)) { + Throwable failure = null; + try { + bootstrap.manager.loadPartition(); + } catch (Throwable caught) { + failure = caught; + } + assertNotNull("Pruning must not erase an older legacy database", failure); + assertTrue(failure.getMessage().contains("GraphID")); + assertEquals(before, fileHashes(data)); + assertFalse(RocksDBFactory.getInstance().getGraphNames().contains(DB_NAME)); + } + } + + @Test + public void testBootstrapPreservesLegacyLatestVersion() throws Exception { + Path location = this.temporary.newFolder("bootstrap-latest-legacy").toPath(); + Path data = Files.createDirectory(location.resolve("db")); + writeDatabase(data, DB_NAME + "_1", 1); + writeDatabase(data, DB_NAME + "_2", 65534); + Map before = fileHashes(data); + try (Bootstrap bootstrap = new Bootstrap(location)) { + assertThrows(HgStoreException.class, bootstrap.manager::loadPartition); + assertEquals(before, fileHashes(data)); + assertFalse(RocksDBFactory.getInstance().getGraphNames().contains(DB_NAME)); + } + } + + @Test + public void testBootstrapLoadsCurrentValidMappingAndData() throws Exception { + Path location = this.temporary.newFolder("bootstrap-valid").toPath(); + Path data = Files.createDirectory(location.resolve("db")); + writeDatabase(data, DB_NAME + "_1", 1); + writeBootstrapMetadata(data.resolve(DB_NAME + "_1"), 1); + try (Bootstrap bootstrap = new Bootstrap(location)) { + bootstrap.manager.loadPartition(); + assertTrue(bootstrap.manager.hasPartition("test", PARTITION_ID)); + assertEquals(1, bootstrap.handler.getKeyCreator().getGraphId(PARTITION_ID, "test")); + for (String table : TABLES) { + assertArrayEquals(VALUE, bootstrap.handler.doGet("test", 0, table, new byte[]{1})); + } + } + } + + @Test + public void testFactoryValidatesColdOpenAndReusesCachedSession() throws Exception { + Path data = this.temporary.newFolder("factory-valid").toPath(); + writeDatabase(data, DB_NAME, 1); + java.util.concurrent.atomic.AtomicInteger checks = new java.util.concurrent.atomic.AtomicInteger(); + RocksDBFactory factory = RocksDBFactory.getInstance(); + try { + try (RocksDBSession first = factory.createGraphDB(data.toString(), DB_NAME, 0L, path -> { + checks.incrementAndGet(); + GraphIdManager.checkLegacyGraphIdData(path, PARTITION_ID); + }); + RocksDBSession cached = factory.createGraphDB(data.toString(), DB_NAME, 0L, + path -> org.junit.Assert.fail("Cached session must not be revalidated"))) { + assertEquals(1, checks.get()); + assertSame(first.getDB(), cached.getDB()); + assertRows(cached, 1); + } + } finally { + factory.releaseGraphDB(DB_NAME); + } + } + + @Test + public void testCachedMemtableLegacyDataRejectsBeforeRaft() throws Exception { + Path location = this.temporary.newFolder("cached-legacy-location").toPath(); + Path data = Files.createDirectory(location.resolve("db")); + RocksDBFactory factory = RocksDBFactory.getInstance(); + try (RocksDBSession cached = factory.createGraphDB(data.toString(), DB_NAME, 0L)) { + cached.setDisableWAL(true); + putRows(cached, 65534); + // A separate RO instance cannot see WAL-disabled writes still in this primary's memtable. + GraphIdManager.checkLegacyGraphIdData(cached.getDbPath(), PARTITION_ID); + assertThrows(HgStoreException.class, () -> GraphIdManager.checkLegacyGraphIdData(cached, PARTITION_ID)); + int references = cached.getRefCount(); + PartitionEngine engine = partitionEngine(location, true); + PartitionEngineOptions options = new PartitionEngineOptions(); + options.setGroupId(PARTITION_ID); + + HgStoreException failure = assertThrows(HgStoreException.class, () -> engine.init(options)); + + assertTrue(failure.getMessage().contains("GraphID")); + assertNull(engine.getRaftNode()); + assertNull(field(engine, "stateMachine")); + assertNull(engine.getTaskManager()); + assertEquals(references, cached.getRefCount()); + assertTrue(cached.getDB().isOwningHandle()); + assertRows(cached, 65534); + } finally { + factory.releaseGraphDB(DB_NAME); + } + } + + @Test + public void testCachedValidSessionPreflightAvoidsPathResolution() throws Exception { + Path location = this.temporary.newFolder("cached-valid-location").toPath(); + Path data = Files.createDirectory(location.resolve("db")); + RocksDBFactory factory = RocksDBFactory.getInstance(); + try (RocksDBSession cached = factory.createGraphDB(data.toString(), DB_NAME, 0L)) { + cached.setDisableWAL(true); + putRows(cached, 1); + int references = cached.getRefCount(); + PartitionEngine engine = partitionEngine(location, true); + // Cached native handles already identify the active DB; do not resolve another disk view. + when(engine.getStoreEngine().getPartitionManager().getStoreMetadata().getPartitionStore(PARTITION_ID)) + .thenThrow(new AssertionError("Cached preflight must not resolve a DB path")); + var check = PartitionEngine.class.getDeclaredMethod("checkLegacyGraphIdData", int.class); + check.setAccessible(true); + check.invoke(engine, PARTITION_ID); + + assertEquals(references, cached.getRefCount()); + assertTrue(cached.getDB().isOwningHandle()); + assertRows(cached, 1); + } finally { + factory.releaseGraphDB(DB_NAME); + } + } + + @Test + public void testNewPartitionDatabaseStillCreatesNormally() throws Exception { + Path location = this.temporary.newFolder("bootstrap-new").toPath(); + try (Bootstrap bootstrap = new Bootstrap(location); + RocksDBSession session = bootstrap.handler.getSession(PARTITION_ID)) { + assertTrue(Files.isRegularFile(Path.of(session.getDbPath(), "CURRENT"))); + assertTrue(session.getDB().isOwningHandle()); + } + } + + private static void writeBootstrapMetadata(Path databasePath, long graphId) throws Exception { + List handles = new ArrayList<>(); + try (DBOptions options = new DBOptions(); ColumnFamilyOptions cfOptions = new ColumnFamilyOptions()) { + List descriptors = new ArrayList<>(); + descriptors.add(new ColumnFamilyDescriptor(RocksDB.DEFAULT_COLUMN_FAMILY, cfOptions)); + for (String table : TABLES) { + descriptors.add(new ColumnFamilyDescriptor(table.getBytes(java.nio.charset.StandardCharsets.UTF_8), + cfOptions)); + } + try (RocksDB database = RocksDB.open(options, databasePath.toString(), descriptors, handles)) { + database.put(handles.get(0), MetadataKeyHelper.getGraphIDKey("test"), + Int64Value.of(graphId).toByteArray()); + database.put(handles.get(0), MetadataKeyHelper.getPartitionKey("test", PARTITION_ID), + Metapb.Partition.newBuilder().setId(PARTITION_ID).setGraphName("test") + .build().toByteArray()); + } finally { + handles.forEach(ColumnFamilyHandle::close); + } + } + } + + @Test + public void testBootstrapChecksActualRootAfterLocationUpdate() throws Exception { + Path location = this.temporary.newFolder("bootstrap-relocated").toPath(); + Path data = Files.createDirectory(location.resolve("db")); + writeDatabase(data, DB_NAME + "_1", 1); + Files.createDirectory(data.resolve(DB_NAME + "_2")); + Map before = fileHashes(data); + try (Bootstrap bootstrap = new Bootstrap(location, true)) { + assertThrows(org.apache.hugegraph.rocksdb.access.DBStoreException.class, + bootstrap.manager::loadPartition); + // Location discovery precedes native validation; it is not a startup transaction. + assertEquals(location.toString(), bootstrap.manager.getStoreMetadata() + .getPartitionStore(PARTITION_ID) + .getStoreLocation()); + org.mockito.Mockito.verify(bootstrap.manager.getStoreMetadata()) + .savePartitionStore(org.mockito.ArgumentMatchers.any()); + assertEquals(before, fileHashes(data)); + assertFalse(RocksDBFactory.getInstance().getGraphNames().contains(DB_NAME)); + } + } + + private static class Bootstrap implements AutoCloseable { + + private final PartitionManager manager; + private final BusinessHandlerImpl handler; + private final Object previousHandler; + private final Field handlerField; + + private Bootstrap(Path location) throws Exception { + this(location, false); + } + + private Bootstrap(Path location, boolean relocate) throws Exception { + Files.createDirectories(location.resolve("raft/" + DB_NAME + "/log")); + HgStoreEngineOptions options = new HgStoreEngineOptions(); + options.setDataPath(location.toString()); + options.setRaftPath(location.toString()); + PdProvider provider = mock(PdProvider.class); + when(provider.getShardGroup(PARTITION_ID)).thenReturn( + Metapb.ShardGroup.newBuilder().setId(PARTITION_ID) + .addShards(Metapb.Shard.newBuilder().setStoreId(1L).build()).build()); + Metapb.Partition partition = Metapb.Partition.newBuilder().setId(PARTITION_ID) + .setGraphName("test").build(); + when(provider.getPartitionByID(anyString(), anyInt())).thenReturn( + new org.apache.hugegraph.store.meta.Partition(partition)); + when(provider.getPartitionByCode(anyString(), anyInt())).thenReturn(partition); + this.manager = new PartitionManager(provider, options); + StoreMetadata metadata = mock(StoreMetadata.class); + when(metadata.getStore()).thenReturn(new Store(Metapb.Store.newBuilder().setId(1L).build())); + AtomicReference saved = new AtomicReference<>( + Metapb.PartitionStore.newBuilder().setPartitionId(PARTITION_ID) + .setStoreLocation(relocate ? location.resolve("previous").toString() : + location.toString()).build()); + when(metadata.getPartitionStore(PARTITION_ID)).thenAnswer(call -> saved.get()); + org.mockito.Mockito.doAnswer(call -> { + saved.set(call.getArgument(0)); + return null; + }).when(metadata).savePartitionStore(org.mockito.ArgumentMatchers.any()); + when(metadata.getPartitionRaft(PARTITION_ID)).thenReturn( + Metapb.PartitionRaft.newBuilder().setPartitionId(PARTITION_ID) + .setRaftLocation(location.toString()).build()); + when(metadata.getPartitionStoreLocation(PARTITION_ID, DB_NAME)) + .thenAnswer(call -> saved.get().getStoreLocation()); + Field metadataField = PartitionManager.class.getDeclaredField("storeMetadata"); + metadataField.setAccessible(true); + metadataField.set(this.manager, metadata); + this.handler = new BusinessHandlerImpl(this.manager); + this.handlerField = HgStoreEngine.class.getDeclaredField("businessHandler"); + this.handlerField.setAccessible(true); + this.previousHandler = this.handlerField.get(HgStoreEngine.getInstance()); + this.handlerField.set(HgStoreEngine.getInstance(), this.handler); + } + + @Override + public void close() throws Exception { + try { + RocksDBFactory.getInstance().releaseGraphDB(DB_NAME); + } finally { + this.handlerField.set(HgStoreEngine.getInstance(), this.previousHandler); + } + } + } + + @Test + public void testLegacyDataRejectsBeforeRaftAndPreservesVersions() throws Exception { + Path location = this.temporary.newFolder("existing-location").toPath(); + Path data = Files.createDirectory(location.resolve("db")); + writeDatabase(data, DB_NAME, 1); + writeDatabase(data, DB_NAME + "_4", 65534); + Map before = fileHashes(location); + PartitionEngine engine = partitionEngine(location, true); + PartitionEngineOptions options = new PartitionEngineOptions(); + options.setGroupId(PARTITION_ID); + + HgStoreException failure = assertThrows(HgStoreException.class, () -> engine.init(options)); + + assertTrue(failure.getMessage().contains("GraphID")); + assertEquals(false, field(engine, "started")); + assertNull(engine.getRaftNode()); + assertNull(field(engine, "stateMachine")); + assertNull(engine.getTaskManager()); + assertEquals(before, fileHashes(location)); + } + + @Test + public void testMissingLocationChecksAllConfiguredRootsWithoutWriting() throws Exception { + Path first = this.temporary.newFolder("first-root").toPath(); + Path second = this.temporary.newFolder("second-root").toPath(); + writeDatabase(Files.createDirectory(first.resolve("db")), DB_NAME, 1); + writeDatabase(Files.createDirectory(second.resolve("db")), DB_NAME, 65534); + Map firstBefore = fileHashes(first); + Map secondBefore = fileHashes(second); + PartitionEngine engine = partitionEngine(first, false); + engine.getStoreEngine().getOption().setDataPath(first + "," + second); + PartitionEngineOptions options = new PartitionEngineOptions(); + options.setGroupId(PARTITION_ID); + + assertThrows(HgStoreException.class, () -> engine.init(options)); + + assertNull(engine.getRaftNode()); + assertEquals(firstBefore, fileHashes(first)); + assertEquals(secondBefore, fileHashes(second)); + } + + @Test + public void testLegacySnapshotKeepsActiveAndTargetData() throws Exception { + Path data = this.temporary.newFolder("active-parent").toPath(); + try (RocksDBSession active = session(data, "active")) { + putRows(active, 1); + writeDatabase(data, "active_5", 2); + Path snapshot = snapshot(65534); + Map before = fileHashes(data); + Map incomingBefore = fileHashes(snapshot); + BusinessHandlerImpl handler = handler(active); + assertEquals(1, handler.getKeyCreator().getGraphId(PARTITION_ID, "test")); + Map cached = (Map) field(handler.getKeyCreator(), "graphIdCache"); + Object owner = cached.get(PARTITION_ID); + + HgStoreException failure = assertThrows(HgStoreException.class, + () -> handler.loadSnapshot(snapshot.toString(), "", PARTITION_ID, 5L)); + + assertTrue(failure.getMessage().contains("GraphID")); + assertNotNull(failure.getCause()); + assertSame(owner, cached.get(PARTITION_ID)); + assertEquals(before, fileHashes(data)); + assertEquals(incomingBefore, fileHashes(snapshot)); + assertTrue(Files.isDirectory(data.resolve("active_5_temp_"))); + assertRows(active, 1); + assertArrayEquals(Int64Value.of(1L).toByteArray(), active.sessionOp().get("default", + "HUGEGRAPH/GRAPH_ID/test".getBytes(java.nio.charset.StandardCharsets.UTF_8))); + } + } + + @Test + public void testRejectedNativeSnapshotDoesNotPublishEitherIndex() throws Exception { + assertSnapshotIndices(true); + } + + @Test + public void testAcceptedNativeSnapshotPublishesBothIndices() throws Exception { + assertSnapshotIndices(false); + } + + private void assertSnapshotIndices(boolean legacy) throws Exception { + Path data = this.temporary.newFolder("fsm-active").toPath(); + try (RocksDBSession active = session(data, "fsm")) { + putRows(active, 1); + Path snapshot = snapshot(legacy ? 65534 : 2); + Path snapshotRoot = snapshot.getParent(); + BusinessHandlerImpl handler = handler(active); + HgStoreEngine store = mock(HgStoreEngine.class); + PartitionEngine engine = mock(PartitionEngine.class); + when(engine.getStoreEngine()).thenReturn(store); + when(engine.getGroupId()).thenReturn(PARTITION_ID); + when(engine.loadPartitionsFromLocalDb()).thenReturn(Collections.emptyList()); + when(store.getBusinessHandler()).thenReturn(handler); + PartitionStateMachine machine = new PartitionStateMachine(PARTITION_ID, + new SnapshotHandler(engine)); + Field committed = PartitionStateMachine.class.getDeclaredField("committedIndex"); + committed.setAccessible(true); + committed.setLong(machine, 3L); + SnapshotReader reader = mock(SnapshotReader.class); + when(reader.getPath()).thenReturn(snapshotRoot.toString()); + when(reader.load()).thenReturn(RaftOutter.SnapshotMeta.newBuilder() + .setLastIncludedIndex(5L) + .setLastIncludedTerm(1L).build()); + NodeImpl node = mock(NodeImpl.class); + when(node.getNodeMetrics()).thenReturn(new NodeMetrics(false)); + when(node.getGroupId()).thenReturn("graph-id-snapshot-test"); + FSMCallerOptions options = new FSMCallerOptions(); + options.setNode(node); + options.setFsm(machine); + options.setLogManager(mock(LogManager.class)); + options.setClosureQueue(new ClosureQueueImpl("graph-id-snapshot-test")); + options.setBootstrapId(new LogId(3L, 1L)); + options.setDisruptorBufferSize(16); + FSMCallerImpl caller = new FSMCallerImpl(); + assertTrue(caller.init(options)); + AtomicReference result = new AtomicReference<>(); + CountDownLatch done = new CountDownLatch(1); + try { + assertTrue(caller.onSnapshotLoad(new LoadSnapshotClosure() { + @Override + public SnapshotReader start() { + return reader; + } + + @Override + public void run(Status status) { + result.set(status); + done.countDown(); + } + })); + assertTrue(done.await(5, TimeUnit.SECONDS)); + assertEquals(!legacy, result.get().isOk()); + assertEquals(legacy ? 3L : 5L, machine.getCommittedIndex()); + assertEquals(legacy ? 3L : 5L, caller.getLastAppliedIndex()); + assertEquals(legacy ? 3L : 5L, caller.getLastCommittedIndex()); + assertEquals(!legacy, Files.exists(snapshotRoot.resolve("should_not_load"))); + if (legacy) { + assertRows(active, 1); + assertFalse(Files.exists(data.resolve("fsm_5"))); + } else { + assertTrue(Files.isDirectory(data.resolve("fsm_5"))); + try (RocksDBSession restored = session(data, "fsm")) { + assertRows(restored, 2); + BusinessHandlerImpl restoredHandler = handler(restored); + assertEquals(2, restoredHandler.getKeyCreator().getGraphId(PARTITION_ID, "test")); + for (String table : TABLES) { + assertArrayEquals(VALUE, restoredHandler.doGet("test", 0, table, new byte[]{1})); + } + } + } + } finally { + caller.shutdown(); + caller.join(); + } + } + } + + private PartitionEngine partitionEngine(Path location, boolean persistedLocation) { + HgStoreEngine store = mock(HgStoreEngine.class); + PartitionManager manager = mock(PartitionManager.class); + StoreMetadata metadata = mock(StoreMetadata.class); + HgStoreEngineOptions options = new HgStoreEngineOptions(); + options.setDataPath(location.toString()); + when(store.getOption()).thenReturn(options); + when(store.getPartitionManager()).thenReturn(manager); + when(manager.getStoreMetadata()).thenReturn(metadata); + if (persistedLocation) { + when(metadata.getPartitionStore(PARTITION_ID)).thenReturn( + Metapb.PartitionStore.newBuilder().setPartitionId(PARTITION_ID) + .setStoreLocation(location.toString()).build()); + } + return new PartitionEngine(store, mock(ShardGroup.class)); + } + + private static BusinessHandlerImpl handler(RocksDBSession active) { + PartitionManager manager = mock(PartitionManager.class); + PdProvider provider = mock(PdProvider.class); + when(manager.getPdProvider()).thenReturn(provider); + when(manager.hasPartition(anyString(), anyInt())).thenReturn(true); + when(provider.getPartitionByCode(anyString(), anyInt())).thenReturn( + Metapb.Partition.newBuilder().setId(PARTITION_ID).build()); + return new BusinessHandlerImpl(manager) { + @Override + public RocksDBSession getSession(int partitionId) { + return active.clone(); + } + }; + } + + private Path snapshot(int graphId) throws Exception { + Path seed = this.temporary.newFolder("snapshot-seed").toPath(); + Path root = this.temporary.newFolder("snapshot-root").toPath(); + Path data = root.resolve("data"); + try (RocksDBSession session = session(seed, "seed")) { + putRows(session, graphId); + try (Checkpoint checkpoint = Checkpoint.create(session.getDB())) { + checkpoint.createCheckpoint(data.toString()); + } + } + return data; + } + + private static void writeDatabase(Path root, String name, int graphId) { + List handles = new ArrayList<>(); + try (DBOptions options = new DBOptions().setCreateIfMissing(true).setCreateMissingColumnFamilies(true); + ColumnFamilyOptions cfOptions = new ColumnFamilyOptions().setWriteBufferSize(1048576)) { + List descriptors = new ArrayList<>(); + descriptors.add(new ColumnFamilyDescriptor(RocksDB.DEFAULT_COLUMN_FAMILY, cfOptions)); + for (String table : TABLES) { + descriptors.add(new ColumnFamilyDescriptor(table.getBytes(java.nio.charset.StandardCharsets.UTF_8), + cfOptions)); + } + try (RocksDB database = RocksDB.open(options, root.resolve(name).toString(), descriptors, handles)) { + for (int i = 1; i < handles.size(); i++) { + database.put(handles.get(i), key(graphId), VALUE); + } + } finally { + handles.forEach(ColumnFamilyHandle::close); + } + } catch (Exception failure) { + throw new IllegalStateException(failure); + } + } + + private static Object field(Object object, String name) throws Exception { + Field field = object.getClass().getDeclaredField(name); + field.setAccessible(true); + return field.get(object); + } + + private static RocksDBSession session(Path root, String name) { + return new RocksDBSession(org.apache.hugegraph.rocksdb.access.RocksDBFactory + .getInstance().getHugeConfig(), root.toString(), name, 0L); + } + + private static void putRows(RocksDBSession session, int graphId) { + session.createTables(TABLES); + SessionOperator operator = session.sessionOp(); + operator.prepare(); + try { + for (String table : TABLES) { + operator.put(table, key(graphId), VALUE); + } + if (graphId != 65534) { + operator.put("default", "HUGEGRAPH/GRAPH_ID/test".getBytes(java.nio.charset.StandardCharsets.UTF_8), + Int64Value.of(graphId).toByteArray()); + } + operator.commit(); + } catch (RuntimeException failure) { + operator.rollback(); + throw failure; + } + } + + private static void assertRows(RocksDBSession session, int graphId) { + for (String table : TABLES) { + assertArrayEquals(VALUE, session.sessionOp().get(table, key(graphId))); + } + } + + private static byte[] key(int graphId) { + byte[] key = {0, 0, 1, 0, 0}; + Bits.putShort(key, 0, graphId); + return key; + } + + private static Map fileHashes(Path root) throws Exception { + Map hashes = new LinkedHashMap<>(); + try (Stream paths = Files.walk(root)) { + for (Path path : (Iterable) paths.filter(Files::isRegularFile) + .sorted()::iterator) { + // Rejected loads deliberately retain a new staging directory. + if (path.toString().contains("_temp_")) { + continue; + } + hashes.put(root.relativize(path).toString(), + Arrays.toString(MessageDigest.getInstance("SHA-256") + .digest(Files.readAllBytes(path)))); + } + } + return hashes; + } +} diff --git a/hugegraph-store/hg-store-test/src/main/java/org/apache/hugegraph/store/meta/GraphIDManagerTest.java b/hugegraph-store/hg-store-test/src/main/java/org/apache/hugegraph/store/meta/GraphIDManagerTest.java index ac0d5d0d1e..9b3314a722 100644 --- a/hugegraph-store/hg-store-test/src/main/java/org/apache/hugegraph/store/meta/GraphIDManagerTest.java +++ b/hugegraph-store/hg-store-test/src/main/java/org/apache/hugegraph/store/meta/GraphIDManagerTest.java @@ -91,6 +91,160 @@ public void clear() { } } + @Test + public void testLegacyVertexRowsBlockAllocation() { + this.assertLegacyRowsBlockAllocation(VERTEX_TABLE, false); + } + + @Test + public void testLegacyOutEdgeRowsBlockAllocationWithoutVertexTable() { + this.assertLegacyRowsBlockAllocation(OUT_EDGE_TABLE, false); + } + + @Test + public void testLegacyInEdgeRowsBlockAllocationWithoutVertexTable() { + this.assertLegacyRowsBlockAllocation(IN_EDGE_TABLE, false); + } + + @Test + public void testLegacyIndexRowsBlockAllocationWithoutVertexTable() { + this.assertLegacyRowsBlockAllocation(INDEX_TABLE, false); + } + + @Test + public void testLegacyRowsBlockAllocationDespiteExistingMapping() { + this.assertLegacyRowsBlockAllocation(INDEX_TABLE, true); + } + + @Test + public void testLegacyCheckPreservesCleanAndLegalBoundaryData() { + GraphIdManager.maxGraphID = this.previousMaxGraphId; + Set cleanTables = new HashSet<>(this.session.getTables().keySet()); + GraphIdManager.checkLegacyGraphIdData(this.session, PARTITION_ID); + Assert.assertEquals(cleanTables, this.session.getTables().keySet()); + Assert.assertFalse(this.session.tableIsExist(VERTEX_TABLE)); + + GraphIdManager manager = this.newManager(); + manager.updateGraphIds(Collections.singletonMap("boundary", 0xFFFDL)); + this.session.createTables(OUT_EDGE_TABLE); + byte[] key = {(byte) 0xff, (byte) 0xfd, 1, 0, 0}; + byte[] value = {43}; + SessionOperator operator = this.session.sessionOp(); + operator.prepare(); + try { + operator.put(OUT_EDGE_TABLE, key, value); + operator.commit(); + } catch (RuntimeException e) { + operator.rollback(); + throw e; + } + Set tablesBefore = new HashSet<>(this.session.getTables().keySet()); + this.session.flush(true); + Map filesBefore = this.databaseFiles(); + GraphIdManager.checkLegacyGraphIdData(this.session, PARTITION_ID); + GraphIdManager.checkLegacyGraphIdData(this.session.getDbPath(), PARTITION_ID); + Assert.assertEquals(filesBefore, this.databaseFiles()); + Assert.assertEquals(0L, manager.getGraphIdOrCreate("clean-control")); + Assert.assertEquals(0xFFFDL, this.newManager().getGraphId("boundary")); + Assert.assertArrayEquals(value, operator.get(OUT_EDGE_TABLE, key)); + Assert.assertEquals(tablesBefore, this.session.getTables().keySet()); + Assert.assertTrue(this.session.getIteratorMap().isEmpty()); + } + + private void assertLegacyRowsBlockAllocation(String table, boolean existingMapping) { + GraphIdManager.maxGraphID = this.previousMaxGraphId; + GraphIdManager manager = this.newManager(); + if (existingMapping) { + manager.updateGraphIds(Collections.singletonMap("existing", 1L)); + } + this.session.createTables(table); + byte[] key = {(byte) 0xff, (byte) 0xfe, 1, 0, 0}; + byte[] value = {42}; + SessionOperator operator = this.session.sessionOp(); + operator.prepare(); + try { + operator.put(table, key, value); + operator.commit(); + } catch (RuntimeException e) { + operator.rollback(); + throw e; + } + Map metadataBefore = this.metadataRows(); + Set tablesBefore = new HashSet<>(this.session.getTables().keySet()); + this.session.flush(true); + Map filesBefore = this.databaseFiles(); + Assert.assertThrows(HgStoreException.class, + () -> GraphIdManager.checkLegacyGraphIdData(this.session.getDbPath(), + PARTITION_ID)); + Assert.assertEquals(filesBefore, this.databaseFiles()); + HgStoreException failure = Assert.assertThrows(HgStoreException.class, + () -> manager.getGraphIdOrCreate("legacy-first-batch")); + Assert.assertTrue(failure.getMessage().contains("partition " + PARTITION_ID)); + Assert.assertTrue(failure.getMessage().contains("table " + table)); + Assert.assertTrue(failure.getMessage().contains("prefix 0xFFFE")); + Assert.assertEquals(filesBefore, this.databaseFiles()); + Assert.assertEquals(metadataBefore, this.metadataRows()); + Assert.assertEquals(tablesBefore, this.session.getTables().keySet()); + Assert.assertArrayEquals(value, operator.get(table, key)); + Assert.assertEquals(0xFFFE, manager.getGraphId("legacy-first-batch")); + Assert.assertNull(manager.get(Int64Value.parser(), + MetadataKeyHelper.getGraphIDKey("legacy-first-batch"))); + Assert.assertNull(manager.get(Int64Value.parser(), + manager.genCIDSlotKey(GraphIdManager.GRAPH_ID_PREFIX, 0L))); + Assert.assertNull(manager.get(Int64Value.parser(), + MetadataKeyHelper.getCidKey(GraphIdManager.GRAPH_ID_PREFIX))); + if (existingMapping) { + Assert.assertEquals(1L, this.newManager().getGraphId("existing")); + } + this.reopenDatabase(); + Assert.assertArrayEquals(value, this.session.sessionOp().get(table, key)); + Assert.assertEquals(metadataBefore, this.metadataRows()); + Assert.assertThrows(HgStoreException.class, + () -> GraphIdManager.checkLegacyGraphIdData(this.session, PARTITION_ID)); + Assert.assertTrue(this.session.getIteratorMap().isEmpty()); + } + + @Test + public void testLegacyPathCheckDoesNotCreateMissingDatabase() { + File missing = new File(this.dbPath, "not-created"); + Assert.assertFalse(missing.exists()); + Assert.assertThrows(HgStoreException.class, + () -> GraphIdManager.checkLegacyGraphIdData(missing.getAbsolutePath(), + PARTITION_ID)); + Assert.assertFalse(missing.exists()); + } + + private Map databaseFiles() { + Map files = new LinkedHashMap<>(); + java.nio.file.Path db = new File(this.session.getDbPath()).toPath(); + try (var paths = java.nio.file.Files.walk(db)) { + paths.filter(java.nio.file.Files::isRegularFile).sorted().forEach(path -> { + try { + byte[] digest = java.security.MessageDigest.getInstance("SHA-256") + .digest(java.nio.file.Files.readAllBytes(path)); + files.put(db.relativize(path).toString(), java.util.HexFormat.of().formatHex(digest)); + } catch (java.io.IOException | java.security.NoSuchAlgorithmException e) { + throw new AssertionError(e); + } + }); + } catch (java.io.IOException e) { + throw new AssertionError(e); + } + return files; + } + + private Map metadataRows() { + Map rows = new LinkedHashMap<>(); + try (var iterator = this.session.sessionOp().scan("default")) { + while (iterator.hasNext()) { + RocksDBSession.BackendColumn column = iterator.next(); + rows.put(java.util.HexFormat.of().formatHex(column.name), + java.util.HexFormat.of().formatHex(column.value)); + } + } + return rows; + } + @Test public void testAllocateBoundaryIdBeforeWrap() { GraphIdManager manager = this.newManager();