diff --git a/.github/workflows/graphid-official-source.yml b/.github/workflows/graphid-official-source.yml new file mode 100644 index 0000000000..b84ad9431a --- /dev/null +++ b/.github/workflows/graphid-official-source.yml @@ -0,0 +1,50 @@ +# 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. +name: GraphID official source fixture +on: + pull_request: + paths: + - 'hugegraph-store/hg-store-test/src/test/graphid-rekey/official-17-source.py' + - '.github/workflows/graphid-official-source.yml' +permissions: + contents: read +jobs: + official-source: + runs-on: ubuntu-24.04 + timeout-minutes: 15 + steps: + - uses: actions/checkout@v7 + with: + ref: ${{ github.event.pull_request.head.sha }} + persist-credentials: false + - uses: actions/setup-java@v6 + with: + distribution: zulu + java-version: '17' + - uses: actions/setup-java@v6 + with: + distribution: zulu + java-version: '11' + - name: Generate stopped official 1.7 source in isolated directories + shell: bash + run: | + set -euo pipefail + python3 -u hugegraph-store/hg-store-test/src/test/graphid-rekey/official-17-source.py \ + "$RUNNER_TEMP/official-17-source" + - name: Preserve fixture state and diagnostics + if: always() + uses: actions/upload-artifact@v7 + with: + name: graphid-official-source-${{ github.event.pull_request.head.sha }} + retention-days: 7 + path: ${{ runner.temp }}/official-17-source/evidence/ diff --git a/.github/workflows/graphid-rekey-experiment.yml b/.github/workflows/graphid-rekey-experiment.yml new file mode 100644 index 0000000000..ec08988575 --- /dev/null +++ b/.github/workflows/graphid-rekey-experiment.yml @@ -0,0 +1,78 @@ +# +# 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. +# +name: GraphID rekey experiment +on: + pull_request: + paths: + - 'hugegraph-store/hg-store-core/src/main/java/org/apache/hugegraph/store/upgrade/GraphIdMigrationPlan.java' + - 'hugegraph-store/hg-store-core/src/main/java/org/apache/hugegraph/store/upgrade/GraphIdRewriter.java' + - 'hugegraph-store/hg-store-test/src/test/graphid-rekey/**' + - '.github/workflows/graphid-rekey-experiment.yml' +permissions: + contents: read +jobs: + native-rekey: + runs-on: ubuntu-24.04 + timeout-minutes: 12 + steps: + - uses: actions/checkout@v7 + with: + ref: ${{ github.event.pull_request.head.sha }} + persist-credentials: false + - uses: actions/setup-java@v6 + with: + distribution: zulu + java-version: '17' + - name: Run bounded native database experiments + shell: bash + run: | + set -euo pipefail + WORK="$RUNNER_TEMP/graphid-rekey" + mkdir -p "$WORK/classes" "$WORK/deps" + exec > >(tee "$WORK/run.log") 2>&1 + git rev-parse HEAD | tee "$WORK/source-sha.txt" + java -version + cat > "$WORK/pom.xml" <<'POM' + + 4.0.0 + org.apache.hugegraphgraphid-rekey-probe1 + + org.rocksdbrocksdbjni8.10.2 + com.google.protobufprotobuf-java3.22.3 + + + POM + mvn -B -ntp -f "$WORK/pom.xml" org.apache.maven.plugins:maven-dependency-plugin:3.8.1:copy-dependencies -DoutputDirectory="$WORK/deps" + sha256sum "$WORK"/deps/*.jar | tee "$WORK/dependencies.sha256" + CORE=hugegraph-store/hg-store-core/src/main/java/org/apache/hugegraph/store/upgrade + TEST=hugegraph-store/hg-store-test/src/test/graphid-rekey + javac --release 17 -cp "$WORK/deps/*" -d "$WORK/classes" \ + "$CORE/GraphIdMigrationPlan.java" "$CORE/GraphIdRewriter.java" "$TEST"/*.java + java -Xmx512m -cp "$WORK/classes:$WORK/deps/*" org.apache.hugegraph.store.upgrade.GraphIdMigrationPlanProbe + timeout 180 java -Xmx512m -cp "$WORK/classes:$WORK/deps/*" org.apache.hugegraph.store.upgrade.GraphIdRewriterProbe "$WORK/cases" + - name: Retain exact-head evidence and native dependencies + if: always() + uses: actions/upload-artifact@v7 + with: + name: graphid-rekey-${{ github.event.pull_request.head.sha }} + retention-days: 7 + path: | + ${{ runner.temp }}/graphid-rekey/run.log + ${{ runner.temp }}/graphid-rekey/source-sha.txt + ${{ runner.temp }}/graphid-rekey/dependencies.sha256 + ${{ runner.temp }}/graphid-rekey/deps/ + ${{ runner.temp }}/graphid-rekey/cases/ diff --git a/docs/graphid-offline-upgrade-design.md b/docs/graphid-offline-upgrade-design.md new file mode 100644 index 0000000000..8c7367682e --- /dev/null +++ b/docs/graphid-offline-upgrade-design.md @@ -0,0 +1,148 @@ +# Offline GraphID conversion: design and experimental kernel + +Status: **experimental; not a complete cluster upgrade tool; do not activate its output**. +Related: apache/hugegraph#3224 and hugegraph/hugegraph#325. The latter is a +startup/snapshot guard, not the conversion implementation. + +## Goal + +Permit a stopped HStore cluster to upgrade without exporting all vertices and +edges through Server APIs and rebuilding them through a logical restore. A +physical rewrite still reads and writes data; it is not an O(1) metadata change. +Do not add a permanent legacy read path or guess ownership of shared FFFE keys. + +## Alternatives and decision + +| Candidate | Benefits | Failure boundary | Decision | +| --- | --- | --- | --- | +| Change only graph-to-ID metadata | Smallest change and no data rewrite | Existing keys retain FFFE; changing the mapping hides them. Assigning FFFE permanently or aliasing multiple owners reintroduces legacy semantics | Reject as the upgrade route | +| Rewrite keys in the existing DB | Avoids a second complete DB copy | A crash between batches leaves mixed prefixes. Requires a durable migration journal and a startup barrier throughout the transition | Keep as a comparison candidate, not the first implementation | +| Rewrite an isolated copy, validate, then switch while stopped | Original state remains available; bounded batches need no in-place rollback protocol | Needs additional disk, ownership proof, and a coordinated Raft/snapshot cutover | Preferred baseline; kernel implemented here | + +Only immutable SST files may be hard-linked on the same filesystem. CURRENT, +MANIFEST, OPTIONS and WAL files are copied. The destination `LOCK` is created as +a fresh file: reopening the source `LOCK` while its process-wide POSIX lock is +held can release that lock on some systems. Cross-filesystem staging falls back +to copying SSTs. No file in the original database is opened for writing by the +kernel. Native tests validate the source lock throughout copying, rather than +assuming read-only open alone is sufficient. + +## Implemented kernel + +`GraphIdMigrationPlan` validates explicit nonoverlapping column-family/logical-key +prefix rules. It deterministically allocates legal target IDs, excluding existing +physical prefixes, graph mappings and allocator slots. A rule is an externally +established ownership assertion, **not proof produced by this algorithm**. + +`GraphIdRewriter.plan()` reads a stopped RocksDB under its existing lock, +checks all FFFE records against the rules, and binds the plan to a framed SHA-256 +digest of all logical records. `convert()` reacquires the lock, revalidates source +state and allocation before creating output, and writes only to a new staging +DB. It rewrites the first two key bytes; the remaining key and value bytes are +unchanged. Mappings and allocator slots are written in a final batch. The source +is neither deleted nor renamed. + +The output is closed and reopened, every expected record is compared, additional +records are detected by total count, metadata/slot values are checked, remaining +FFFE records are rejected, and RocksDB checksums are verified. Interrupted outputs +retain `INCOMPLETE`; retry means making another fresh output from the unchanged +source, not resuming an unqualified partial database. Even a successful result is +labelled `DB_VERIFIED_NOT_CLUSTER_READY`. + +The persisted slot layout intentionally preserves GraphIdManager.genCIDSlotKey's +use of `Long.SIZE` bytes including padding. Consolidating that codec with the +existing helper and adding a cross-implementation regression is a merge gate. + +This kernel is not wired into startup, snapshot installation, a released CLI, +or automatic directory activation. It cannot yet establish that Raft will not +replay or install older state. Its explicit output label is not a durable +cluster-ready protocol and must not be treated as one. + +## Ownership is a separate correctness gate + +A single configured user graph does not imply a single historical owner. System +graphs, variables and tasks must be inventoried. Negative/hidden label IDs, +successful deserialization of a sample, or absence of another current mapping do +not by themselves prove ownership. Two graphs may have had identical encodings; +physical overwrites cannot be recovered by a prefix rewrite. + +For the actual 1.7 fixture, first inventory all graph namespaces and schema, +including hidden state. Compare complete candidate decoding/encoding and index +references, and ground a partitioning rule in independent input/provenance. +Every affected record must have exactly one supported owner. A fully matching +rule set is necessary but not sufficient: a wrong human-supplied rule can still +be wrong. Do not promote these synthetic test prefixes into a production +heuristic. Prefix rules are intentionally a restricted interface; an arbitrary +schema-aware classifier may require a different ownership representation. + +Mixed allocated/legacy prefixes are rejected, including an existing valid mapping +for a migrating graph. A union is unsafe without deletion/truncate history; +equal current values or an empty new prefix do not establish that history. + +## Cluster cutover contract (not implemented yet) + +1. Stop writers, schema changes, TTL cleanup and background mutation sources. + Take matched PD, every Store, configuration, data and Raft backups. On the old + version, finish replay and take successful snapshots at a verified committed + boundary for each partition. A stopped process alone does not flush WAL-disabled + writes. Do not discard Raft logs to make a dirty state appear clean. +2. Obtain a common ownership and target-ID plan per partition across all its + replicas. IDs need not be globally equal across different partitions. Verify + logical state and committed index/term before converting any candidate. +3. Prefer a **snapshot-aligned/no-unapplied-tail** first release. Verify the log + boundary mechanically; a tail requires old-version recovery and a fresh + snapshot, or a separately implemented replay-aware path. An operator saying + 'snapshot completed' is not enough. +4. Convert both the selected DB and the current snapshot in staging. Rebuild + snapshot file lists/checksums using JRaft's APIs and HugeGraph's checksum + convention. Preserve index, term, membership and unrelated local metadata. + Old snapshots must not remain eligible for later local or remote installation. + Never hand-edit only the leader or replace files underneath running Raft. +5. Validate all prepared replicas against the common plan. Keep all services + stopped until every required partition/replica is prepared. Directory renames + are local operations, not an atomic distributed transaction. The activation + wrapper must fail closed on a partial cutover and preserve matched old roots. +6. Start 1.8, verify reads and new writes, cold restart, leader changes and a real + remote snapshot to a lagging replica. Rollback before traffic uses the complete + matched old state; rollback after new writes requires reconciling those writes. + +These gates must precede a public upgrade CLI. This PR does not implement them +by deleting logs, resetting indexes, suppressing checks, or allowing FFFE aliases. + +## Reproducible evidence + +`.github/workflows/graphid-rekey-experiment.yml` compiles the exact head's kernel +with Java 17, RocksDB JNI 8.10.2 and protobuf 3.22.3. It runs the dependency-free +planner probe and the native synthetic DB probe. It retains dependency hashes, +exact commit, logs and small DB fixtures in a seven-day artifact. + +The native probe is designed to cover five data column families, two explicitly +known owners, existing legal data, opaque value preservation, ID/slot publication, +close/reopen verification, deterministic fresh-output retry, injected failure +after a durable batch, stale plans, mixed prefixes and a separate-process native +lock. It additionally checks that cold-copying does not release the held source +lock and that interruption at an exact final batch cannot publish a success +certificate. A mapping-only counterexample is also exercised against an actual DB. + +These probes are not official 1.7 Server data, actual TTL expiration, Raft replay, +snapshot installation, a throughput benchmark, or a three-Store upgrade test. +Test presence is not test execution: use the actual workflow outcome and retained +logs. The local environment used for the initial planner probe had Java 21 with +`javac --release 17`, no Maven/Docker/RocksDB JNI and unavailable external DNS. + +## Remaining acceptance gates + +- Run the fixed issue fixture from SebastianGruza/hugegraph-validation at + 52c52778263e4ba81686a86de8eac2cf4b0df7ae with the official 1.7 release, not only + handcrafted keys. Validate its 16 queries, complete schema/data/index identity, + hidden state and proven owner separation. +- Implement and fault-test the cluster contract above on one Store and three + Stores. Cover no-tail/tail boundaries, partial conversion, interrupted activation, + old/current snapshots, remote snapshot checksums and mixed-version refusal. +- Compare in-place versus isolated conversion on the same representative dataset; + report elapsed time, actual bytes read/written, peak disk and downtime. The + baseline choice here is based on failure isolation, not a completed cost benchmark. +- Add a user-facing plan/convert/verify/activate entry point only after qualification; + preserve original TTL timestamps and state, rather than re-creating graph objects. +- Full reactor build, project formatting, codec compatibility and independent review + remain required in addition to the isolated native probe. diff --git a/hugegraph-store/hg-store-core/src/main/java/org/apache/hugegraph/store/upgrade/GraphIdMigrationPlan.java b/hugegraph-store/hg-store-core/src/main/java/org/apache/hugegraph/store/upgrade/GraphIdMigrationPlan.java new file mode 100644 index 0000000000..e6921ae75e --- /dev/null +++ b/hugegraph-store/hg-store-core/src/main/java/org/apache/hugegraph/store/upgrade/GraphIdMigrationPlan.java @@ -0,0 +1,139 @@ +/* + * 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.upgrade; + +import java.util.Arrays; +import java.util.Collections; +import java.util.HexFormat; +import java.util.LinkedHashMap; +import java.util.List; +import java.util.Map; +import java.util.Set; +import java.util.TreeSet; + +/** A source-bound plan, not an inference of historical graph ownership. */ +public final class GraphIdMigrationPlan { + + public static final int LEGACY_ID = 0xFFFE; + private final String sourceDigest; + private final List rules; + private final Map targets; + + /** Prefix is hexadecimal logical-key bytes, excluding the two-byte GraphID. */ + public record Rule(String columnFamily, String prefixHex, String graph) { + public Rule { + if (columnFamily == null || columnFamily.isBlank() || graph == null || + graph.isBlank() || prefixHex == null) { + throw new IllegalArgumentException("A rule needs a column family, prefix and graph"); + } + prefixHex = HexFormat.of().formatHex(HexFormat.of().parseHex(prefixHex)); + } + + public boolean matches(String family, byte[] key) { + byte[] prefix = HexFormat.of().parseHex(this.prefixHex); + return this.columnFamily.equals(family) && isLegacy(key) && + key.length >= prefix.length + 2 && + Arrays.equals(key, 2, 2 + prefix.length, prefix, 0, prefix.length); + } + } + + public GraphIdMigrationPlan(String sourceDigest, List rules, + Map targets) { + if (sourceDigest == null || !sourceDigest.matches("[0-9a-f]{64}")) { + throw new IllegalArgumentException("Expected a SHA-256 source digest"); + } + validateRules(rules); + Set owners = new TreeSet<>(); + rules.forEach(rule -> owners.add(rule.graph())); + if (!owners.equals(targets.keySet()) || + targets.values().stream().anyMatch(id -> id == null || id < 0 || id >= LEGACY_ID) || + new TreeSet<>(targets.values()).size() != targets.size()) { + throw new IllegalArgumentException("Targets must be legal, unique and cover exactly the owners"); + } + this.sourceDigest = sourceDigest; + this.rules = List.copyOf(rules); + this.targets = Collections.unmodifiableMap(new LinkedHashMap<>(targets)); + } + + public static void validateRules(List rules) { + if (rules == null || rules.isEmpty()) { + throw new IllegalArgumentException("Explicit independently verified ownership rules are required"); + } + for (int i = 0; i < rules.size(); i++) { + Rule a = rules.get(i); + for (int j = i + 1; j < rules.size(); j++) { + Rule b = rules.get(j); + if (a.columnFamily().equals(b.columnFamily()) && + (a.prefixHex().startsWith(b.prefixHex()) || + b.prefixHex().startsWith(a.prefixHex()))) { + throw new IllegalArgumentException("Overlapping ownership rules: " + a + " / " + b); + } + } + } + } + + public static Map allocate(List rules, Set occupied) { + validateRules(rules); + Set owners = new TreeSet<>(); + rules.forEach(rule -> owners.add(rule.graph())); + Map targets = new LinkedHashMap<>(); + int next = 0; + for (String owner : owners) { + while (next < LEGACY_ID && occupied.contains(next)) { + next++; + } + if (next >= LEGACY_ID) { + throw new IllegalStateException("No free GraphID for " + owner); + } + targets.put(owner, next++); + } + return targets; + } + + public static boolean isLegacy(byte[] key) { + return key.length >= 2 && (key[0] & 0xFF) == 0xFF && (key[1] & 0xFF) == 0xFE; + } + + public byte[] rewrite(String family, byte[] key) { + if (!isLegacy(key)) { + return key.clone(); + } + for (Rule rule : this.rules) { + if (rule.matches(family, key)) { + int target = this.targets.get(rule.graph()); + byte[] converted = key.clone(); + converted[0] = (byte) (target >>> 8); + converted[1] = (byte) target; + return converted; + } + } + throw new IllegalStateException("No proven owner for legacy key in " + family + + ": " + HexFormat.of().formatHex(key)); + } + + public String sourceDigest() { + return this.sourceDigest; + } + + public List rules() { + return this.rules; + } + + public Map targets() { + return this.targets; + } +} diff --git a/hugegraph-store/hg-store-core/src/main/java/org/apache/hugegraph/store/upgrade/GraphIdRewriter.java b/hugegraph-store/hg-store-core/src/main/java/org/apache/hugegraph/store/upgrade/GraphIdRewriter.java new file mode 100644 index 0000000000..37e2512d84 --- /dev/null +++ b/hugegraph-store/hg-store-core/src/main/java/org/apache/hugegraph/store/upgrade/GraphIdRewriter.java @@ -0,0 +1,464 @@ +/* + * 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.upgrade; + +import java.io.IOException; +import java.nio.ByteBuffer; +import java.nio.channels.FileChannel; +import java.nio.channels.FileLock; +import java.nio.charset.StandardCharsets; +import java.nio.file.Files; +import java.nio.file.LinkOption; +import java.nio.file.Path; +import java.nio.file.StandardOpenOption; +import java.security.MessageDigest; +import java.util.ArrayList; +import java.util.Arrays; +import java.util.HashSet; +import java.util.HexFormat; +import java.util.LinkedHashMap; +import java.util.List; +import java.util.Map; +import java.util.Set; +import java.util.function.LongConsumer; + +import org.rocksdb.ColumnFamilyDescriptor; +import org.rocksdb.ColumnFamilyHandle; +import org.rocksdb.ColumnFamilyOptions; +import org.rocksdb.DBOptions; +import org.rocksdb.FlushOptions; +import org.rocksdb.InfoLogLevel; +import org.rocksdb.Logger; +import org.rocksdb.Options; +import org.rocksdb.RocksDB; +import org.rocksdb.RocksIterator; +import org.rocksdb.WriteBatch; +import org.rocksdb.WriteOptions; + +import com.google.protobuf.Int64Value; + +/** + * Experimental cold-DB conversion kernel. Never activates a Store or changes + * its Raft state. A verified result is NOT yet a cluster-upgrade certificate. + */ +public final class GraphIdRewriter { + + private static final String GRAPH_KEY = "HUGEGRAPH/GRAPH_ID/"; + private static final String SLOT_KEY = "HUGEGRAPH/CID_SLOT/@GRAPH_ID@/"; + private static final String INTERRUPTED = + "Conversion interrupted; discard unpublished output"; + private static final long BATCH_BYTES = 8L * 1024 * 1024; + + private GraphIdRewriter() { + } + + public record Result(long converted, long records, String sourceDigest, + String targetDigest, Path database) { + } + + public static GraphIdMigrationPlan plan(Path source, ColumnFamilyOptions options, + List rules) throws Exception { + try (SourceLock lock = new SourceLock(source); + Database database = new Database(lock.path, options, true)) { + return plan(database, rules); + } + } + + private static GraphIdMigrationPlan plan(Database source, + List rules) throws Exception { + GraphIdMigrationPlan.validateRules(rules); + Set occupied = new HashSet<>(); + Map mappings = new LinkedHashMap<>(); + Set assigned = new HashSet<>(); + for (var entry : source.families.entrySet()) { + try (RocksIterator iterator = source.db.newIterator(entry.getValue())) { + for (iterator.seekToFirst(); iterator.isValid(); iterator.next()) { + checkInterrupted(); + byte[] key = iterator.key(); + if (key.length >= 2) { + occupied.add(((key[0] & 255) << 8) | (key[1] & 255)); + } + if (!entry.getKey().equals("default")) { + continue; + } + if (startsWith(key, utf8(GRAPH_KEY)) || startsWith(key, utf8(SLOT_KEY))) { + long id = Int64Value.parseFrom(iterator.value()).getValue(); + if (id < 0 || id > GraphIdMigrationPlan.LEGACY_ID) { + throw new IOException("Invalid GraphID metadata"); + } + occupied.add((int) id); + if (startsWith(key, utf8(SLOT_KEY)) && + !Arrays.equals(key, slotKey((int) id))) { + throw new IOException("Allocator slot key/value mismatch"); + } + if (startsWith(key, utf8(GRAPH_KEY))) { + String graph = new String(key, utf8(GRAPH_KEY).length, + key.length - utf8(GRAPH_KEY).length, + StandardCharsets.UTF_8); + if (!Arrays.equals(key, utf8(GRAPH_KEY + graph))) { + throw new IOException("Invalid UTF-8 graph metadata"); + } + mappings.put(graph, id); + if (id != GraphIdMigrationPlan.LEGACY_ID && !assigned.add(id)) { + throw new IOException("Duplicate allocated GraphID"); + } + } + } + } + iterator.status(); + } + } + Map targets = GraphIdMigrationPlan.allocate(rules, occupied); + for (var entry : mappings.entrySet()) { + if (targets.containsKey(entry.getKey()) && entry.getValue() != GraphIdMigrationPlan.LEGACY_ID) { + throw new IOException("Mixed/already allocated graph requires separate recovery: " + entry.getKey()); + } + if (entry.getValue() == GraphIdMigrationPlan.LEGACY_ID && !targets.containsKey(entry.getKey())) { + throw new IOException("Legacy mapping lacks an ownership plan: " + entry.getKey()); + } + } + GraphIdMigrationPlan plan = new GraphIdMigrationPlan(digest(source), rules, targets); + for (var entry : source.families.entrySet()) { + try (RocksIterator iterator = source.db.newIterator(entry.getValue())) { + for (iterator.seek(new byte[]{-1, -2}); iterator.isValid() && + GraphIdMigrationPlan.isLegacy(iterator.key()); iterator.next()) { + checkInterrupted(); + plan.rewrite(entry.getKey(), iterator.key()); + } + iterator.status(); + } + } + return plan; + } + + public static Result convert(Path source, Path output, ColumnFamilyOptions options, + GraphIdMigrationPlan plan) throws Exception { + return convert(source, output, options, plan, ignored -> { }); + } + + static Result convert(Path source, Path output, ColumnFamilyOptions options, + GraphIdMigrationPlan plan, LongConsumer progress) throws Exception { + try (SourceLock lock = new SourceLock(source); + Database original = new Database(lock.path, options, true)) { + GraphIdMigrationPlan actual = plan(original, plan.rules()); + if (!actual.sourceDigest().equals(plan.sourceDigest()) || + !actual.targets().equals(plan.targets())) { + throw new IOException("Stale source or target allocation: regenerate the plan"); + } + Path destination = output.toAbsolutePath().normalize(); + Path parent = destination.getParent().toRealPath(); + destination = parent.resolve(destination.getFileName()); + if (destination.startsWith(lock.path) || Files.exists(destination, LinkOption.NOFOLLOW_LINKS)) { + throw new IOException("Output must be new and outside the source database"); + } + Files.createDirectory(destination); + Files.writeString(destination.resolve("INCOMPLETE"), "Never activate this output\n", + StandardOpenOption.CREATE_NEW); + Path data = Files.createDirectory(destination.resolve("database")); + copyColdDatabase(lock.path, data); + long converted = 0; + try (Database target = new Database(data, options, false); + WriteOptions writes = new WriteOptions().setSync(true).setDisableWAL(false); + WriteBatch batch = new WriteBatch()) { + long bytes = 0; + for (var entry : original.families.entrySet()) { + try (RocksIterator iterator = original.db.newIterator(entry.getValue())) { + for (iterator.seek(new byte[]{-1, -2}); iterator.isValid() && + GraphIdMigrationPlan.isLegacy(iterator.key()); iterator.next()) { + checkInterrupted(); + byte[] key = iterator.key(); + byte[] value = iterator.value(); + ColumnFamilyHandle family = target.families.get(entry.getKey()); + batch.put(family, plan.rewrite(entry.getKey(), key), value); + batch.delete(family, key); + converted++; + bytes += 2L * key.length + value.length; + if (bytes >= BATCH_BYTES || converted % 256 == 0) { + target.db.write(writes, batch); + batch.clear(); + bytes = 0; + progress.accept(converted); + checkInterrupted(); + } + } + iterator.status(); + } + } + checkInterrupted(); + target.db.write(writes, batch); + batch.clear(); + for (var entry : plan.targets().entrySet()) { + byte[] value = Int64Value.of(entry.getValue()).toByteArray(); + batch.put(utf8(GRAPH_KEY + entry.getKey()), value); + batch.put(slotKey(entry.getValue()), value); + } + target.db.write(writes, batch); + try (FlushOptions flush = new FlushOptions().setWaitForFlush(true)) { + target.db.flush(flush, new ArrayList<>(target.families.values())); + } + } + // Reopen: verify durable bytes, not only the writer's memtables. + try (Database target = new Database(data, options, true)) { + long count = verify(original, target, plan); + checkInterrupted(); + target.db.verifyChecksum(); + checkInterrupted(); + String resultDigest = digest(target); + if (!digest(original).equals(plan.sourceDigest())) { + throw new IOException("Source changed during conversion"); + } + checkInterrupted(); + Files.writeString(destination.resolve("DB_VERIFIED_NOT_CLUSTER_READY"), + "source=" + plan.sourceDigest() + "\ntarget=" + resultDigest + + "\nconverted=" + converted + "\nrecords=" + count + "\n", + StandardOpenOption.CREATE_NEW); + Files.delete(destination.resolve("INCOMPLETE")); + return new Result(converted, count, plan.sourceDigest(), resultDigest, data); + } + } + } + + private static long verify(Database source, Database target, GraphIdMigrationPlan plan) throws Exception { + long expectedCount = 0; + for (var entry : source.families.entrySet()) { + try (RocksIterator iterator = source.db.newIterator(entry.getValue())) { + for (iterator.seekToFirst(); iterator.isValid(); iterator.next()) { + checkInterrupted(); + byte[] key = iterator.key(); + byte[] value = iterator.value(); + if (entry.getKey().equals("default") && startsWith(key, utf8(GRAPH_KEY))) { + String graph = new String(key, utf8(GRAPH_KEY).length, + key.length - utf8(GRAPH_KEY).length, + StandardCharsets.UTF_8); + Integer targetId = plan.targets().get(graph); + if (targetId != null) { + value = Int64Value.of(targetId).toByteArray(); + } + } + byte[] actual = target.db.get(target.families.get(entry.getKey()), + plan.rewrite(entry.getKey(), key)); + if (!Arrays.equals(value, actual)) { + throw new IOException("Reopened output differs in " + entry.getKey()); + } + expectedCount++; + } + iterator.status(); + } + } + for (var mapping : plan.targets().entrySet()) { + checkInterrupted(); + byte[] key = utf8(GRAPH_KEY + mapping.getKey()); + byte[] value = Int64Value.of(mapping.getValue()).toByteArray(); + if (source.db.get(key) == null) { + expectedCount++; + } + if (!Arrays.equals(value, target.db.get(key)) || + !Arrays.equals(value, target.db.get(slotKey(mapping.getValue())))) { + throw new IOException("GraphID mapping/allocator slot mismatch"); + } + expectedCount++; // Allocation explicitly excludes every existing slot. + } + long actualCount = 0; + for (ColumnFamilyHandle family : target.families.values()) { + try (RocksIterator iterator = target.db.newIterator(family)) { + for (iterator.seekToFirst(); iterator.isValid(); iterator.next()) { + checkInterrupted(); + if (GraphIdMigrationPlan.isLegacy(iterator.key())) { + throw new IOException("Legacy prefix remains in output"); + } + actualCount++; + } + iterator.status(); + } + } + if (actualCount != expectedCount) { + throw new IOException("Output contains missing or unexpected records"); + } + return actualCount; + } + + private static void copyColdDatabase(Path source, Path target) throws IOException { + try (var files = Files.walk(source)) { + for (Path path : files.toList()) { + if (path.equals(source)) { + continue; + } + if (Files.isSymbolicLink(path)) { + throw new IOException("Symlinks in database are not supported: " + path); + } + Path relative = source.relativize(path); + Path copy = target.resolve(relative); + if (Files.isDirectory(path, LinkOption.NOFOLLOW_LINKS)) { + Files.createDirectory(copy); + } else if (Files.isRegularFile(path, LinkOption.NOFOLLOW_LINKS)) { + String fileName = path.getFileName().toString(); + // Closing any descriptor for a POSIX-locked file can release this + // process' lock. Never reopen the source LOCK while SourceLock is held. + if (relative.equals(Path.of("LOCK"))) { + Files.createFile(copy); + continue; + } + // Only immutable SST files may share inodes. Never hard-link manifests/WAL. + if (fileName.endsWith(".sst")) { + try { + Files.createLink(copy, path); + continue; + } catch (IOException | UnsupportedOperationException unavailable) { + // A different filesystem requires copying; never weaken isolation. + } + } + Files.copy(path, copy); + } else { + throw new IOException("Unsupported database entry: " + path); + } + } + } + } + + private static String digest(Database database) throws Exception { + MessageDigest digest = MessageDigest.getInstance("SHA-256"); + for (var entry : database.families.entrySet()) { + checkInterrupted(); + digest.update((byte) 1); + update(digest, utf8(entry.getKey())); + try (RocksIterator iterator = database.db.newIterator(entry.getValue())) { + for (iterator.seekToFirst(); iterator.isValid(); iterator.next()) { + checkInterrupted(); + digest.update((byte) 2); + update(digest, iterator.key()); + update(digest, iterator.value()); + } + iterator.status(); + } + digest.update((byte) 0); + } + return HexFormat.of().formatHex(digest.digest()); + } + + private static void checkInterrupted() throws InterruptedException { + if (Thread.currentThread().isInterrupted()) { + throw new InterruptedException(INTERRUPTED); + } + } + + private static void update(MessageDigest digest, byte[] bytes) { + digest.update(ByteBuffer.allocate(Integer.BYTES).putInt(bytes.length).array()); + digest.update(bytes); + } + + // Existing GraphIdManager.genCIDSlotKey uses Long.SIZE, not Long.BYTES. + // Preserve that persisted layout, including its zero padding. + static byte[] slotKey(int id) { + byte[] prefix = utf8(SLOT_KEY); + return ByteBuffer.allocate(prefix.length + Long.SIZE).put(prefix).putLong(id).array(); + } + + private static byte[] utf8(String text) { + return text.getBytes(StandardCharsets.UTF_8); + } + + private static boolean startsWith(byte[] key, byte[] prefix) { + return key.length >= prefix.length && Arrays.equals(key, 0, prefix.length, prefix, 0, prefix.length); + } + + private static final class SourceLock implements AutoCloseable { + private final Path path; + private final FileChannel channel; + private final FileLock lock; + + private SourceLock(Path source) throws IOException { + this.path = source.toRealPath(); + if (!Files.isRegularFile(this.path.resolve("CURRENT"), LinkOption.NOFOLLOW_LINKS) || + !Files.isRegularFile(this.path.resolve("LOCK"), LinkOption.NOFOLLOW_LINKS)) { + throw new IOException("Expected a complete, stopped database with CURRENT and LOCK"); + } + this.channel = FileChannel.open(this.path.resolve("LOCK"), StandardOpenOption.READ); + FileLock acquired; + try { + acquired = this.channel.tryLock(0L, Long.MAX_VALUE, true); + if (acquired == null) { + throw new IOException("Source database is running or locked"); + } + } catch (IOException | RuntimeException error) { + this.channel.close(); + throw error; + } + this.lock = acquired; + } + + @Override + public void close() throws IOException { + try { + this.lock.release(); + } finally { + this.channel.close(); + } + } + } + + private static final class Database implements AutoCloseable { + private final DBOptions options = new DBOptions(); + private final ColumnFamilyOptions familyOptions; + private final List handles = new ArrayList<>(); + private final Map families = new LinkedHashMap<>(); + private final Logger logger = new Logger(this.options) { + @Override + protected void log(InfoLogLevel level, String message) { + // Read-only source inspection must not create or rotate LOG files. + } + }; + private RocksDB db; + + private Database(Path path, ColumnFamilyOptions template, boolean readOnly) throws Exception { + this.familyOptions = new ColumnFamilyOptions(template).setDisableAutoCompactions(true); + try { + this.options.setCreateIfMissing(false).setCreateMissingColumnFamilies(false).setLogger(this.logger); + List names; + try (Options listing = new Options().setLogger(this.logger)) { + names = RocksDB.listColumnFamilies(listing, path.toString()); + } + names.sort(Arrays::compareUnsigned); + List descriptors = new ArrayList<>(); + for (byte[] name : names) { + if (!Arrays.equals(name, utf8(new String(name, StandardCharsets.UTF_8)))) { + throw new IOException("Non-UTF-8 column family name"); + } + descriptors.add(new ColumnFamilyDescriptor(name, this.familyOptions)); + } + this.db = readOnly ? RocksDB.openReadOnly(this.options, path.toString(), descriptors, this.handles) : + RocksDB.open(this.options, path.toString(), descriptors, this.handles); + for (int i = 0; i < names.size(); i++) { + this.families.put(new String(names.get(i), StandardCharsets.UTF_8), this.handles.get(i)); + } + } catch (Exception | Error error) { + this.close(); + throw error; + } + } + + @Override + public void close() { + this.handles.forEach(ColumnFamilyHandle::close); + if (this.db != null) { + this.db.close(); + } + this.familyOptions.close(); + this.logger.close(); + this.options.close(); + } + } +} diff --git a/hugegraph-store/hg-store-test/src/test/graphid-rekey/GraphIdMigrationPlanProbe.java b/hugegraph-store/hg-store-test/src/test/graphid-rekey/GraphIdMigrationPlanProbe.java new file mode 100644 index 0000000000..02d1b30059 --- /dev/null +++ b/hugegraph-store/hg-store-test/src/test/graphid-rekey/GraphIdMigrationPlanProbe.java @@ -0,0 +1,77 @@ +/* + * 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.upgrade; + +import java.util.*; + +public class GraphIdMigrationPlanProbe { + static int assertions; + static void require(boolean result) { + assertions++; + if (!result) throw new AssertionError("assertion " + assertions); + } + static void refused(Runnable operation) { + assertions++; + try { operation.run(); } catch (IllegalArgumentException | IllegalStateException expected) { return; } + throw new AssertionError("Expected refusal " + assertions); + } + static GraphIdMigrationPlan plan(List rules, Set used) { + return new GraphIdMigrationPlan("0".repeat(64), rules, GraphIdMigrationPlan.allocate(rules, used)); + } + public static void main(String[] args) throws Exception { + var business = new GraphIdMigrationPlan.Rule("g+v", "01", "DEFAULT/business/g"); + var system = new GraphIdMigrationPlan.Rule("g+v", "02", "DEFAULT/~sys_graph/g"); + var rules = List.of(system, business); + var p = plan(rules, Set.of(0, 2, 4)); + require(p.targets().get(business.graph()) == 1); + require(p.targets().get(system.graph()) == 3); + require(plan(List.of(business, system), Set.of(0, 2, 4)).targets().equals(p.targets())); + byte[] old = {(byte)255, (byte)254, 1, 8, 0, 9}; + byte[] converted = p.rewrite("g+v", old); + require(converted[0] == 0 && converted[1] == 1); + require(Arrays.equals(Arrays.copyOfRange(old,2,old.length), Arrays.copyOfRange(converted,2,converted.length))); + require(GraphIdMigrationPlan.isLegacy(old)); + require(!GraphIdMigrationPlan.isLegacy(converted)); + require(Arrays.equals(converted, p.rewrite("g+v", converted))); + refused(() -> p.rewrite("g+v", new byte[]{-1,-2,3})); + refused(() -> p.rewrite("unknown", old)); + refused(() -> plan(List.of(business, new GraphIdMigrationPlan.Rule("g+v", "0101", system.graph())), Set.of())); + refused(() -> plan(List.of(business, new GraphIdMigrationPlan.Rule("g+v", "", system.graph())), Set.of())); + refused(() -> plan(List.of(business, business), Set.of())); + refused(() -> new GraphIdMigrationPlan("bad", rules, p.targets())); + refused(() -> new GraphIdMigrationPlan("0".repeat(64), rules, Map.of(business.graph(),1,system.graph(),1))); + refused(() -> new GraphIdMigrationPlan("0".repeat(64), rules, Map.of(business.graph(),65534,system.graph(),1))); + refused(() -> new GraphIdMigrationPlan.Rule("g+v", "0", business.graph())); + Set exhausted = new HashSet<>(); + for(int i=0;i<65534;i++) exhausted.add(i); + refused(() -> plan(rules, exhausted)); + Random random = new Random(3224L); + for(int n=0;n<10000;n++) { + byte[] key = new byte[4+random.nextInt(100)]; random.nextBytes(key); + key[0]=-1; key[1]=-2; key[2]=(byte)(1+random.nextInt(2)); + byte[] result=p.rewrite("g+v",key); + require(Arrays.equals(Arrays.copyOfRange(key,2,key.length),Arrays.copyOfRange(result,2,result.length))); + require(result[1]==(key[2]==1?1:3)); + require(Arrays.equals(result,p.rewrite("g+v",result))); + } + System.out.printf("PASS: %d deterministic planner assertions, including 10000 randomized keys. JVM=%s%n",assertions,System.getProperty("java.version")); + // Counterexample: changing metadata alone never moves bytes to the new prefix. + Map rows = new HashMap<>(); rows.put("fffe01080009", "value"); + require(rows.get("000101080009")==null); + System.out.println("PASS: mapping-only counterexample retains unreachable legacy row"); + } +} diff --git a/hugegraph-store/hg-store-test/src/test/graphid-rekey/GraphIdRewriterProbe.java b/hugegraph-store/hg-store-test/src/test/graphid-rekey/GraphIdRewriterProbe.java new file mode 100644 index 0000000000..07eb569534 --- /dev/null +++ b/hugegraph-store/hg-store-test/src/test/graphid-rekey/GraphIdRewriterProbe.java @@ -0,0 +1,340 @@ +/* + * 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.upgrade; + +import java.nio.ByteBuffer; +import java.nio.channels.FileChannel; +import java.nio.channels.FileLock; +import java.nio.charset.StandardCharsets; +import java.nio.file.Files; +import java.nio.file.Path; +import java.nio.file.StandardOpenOption; +import java.security.MessageDigest; +import java.util.ArrayList; +import java.util.Arrays; +import java.util.HexFormat; +import java.util.List; +import java.util.Map; +import java.util.TreeMap; +import java.util.concurrent.TimeUnit; + +import org.rocksdb.ColumnFamilyDescriptor; +import org.rocksdb.ColumnFamilyHandle; +import org.rocksdb.ColumnFamilyOptions; +import org.rocksdb.DBOptions; +import org.rocksdb.FlushOptions; +import org.rocksdb.Options; +import org.rocksdb.RocksDB; + +import com.google.protobuf.Int64Value; + +/** Native synthetic fixtures. These are deliberately NOT called 1.7 cluster upgrade tests. */ +public final class GraphIdRewriterProbe { + private static final String BUSINESS = "DEFAULT/business/g"; + private static final String SYSTEM = "DEFAULT/~sys_graph/g"; + private static final List TABLES = List.of("g+v", "g+ie", "g+oe", "g+index", "custom"); + private static int assertions; + + @FunctionalInterface + private interface Checked { + void run() throws Exception; + } + + private static void check(boolean value, String message) { + assertions++; + if (!value) { + throw new AssertionError(message); + } + } + + private static void refused(Checked operation, String message) throws Exception { + assertions++; + try { + operation.run(); + } catch (Exception expected) { + check(expected.toString().contains(message), expected.toString()); + return; + } + throw new AssertionError("Expected refusal: " + message); + } + + private static byte[] key(int id, int row) { + return ByteBuffer.allocate(9).putShort((short) id).put((byte) (1 + row % 2)) + .putInt(row).putShort((short) (row % 65536)).array(); + } + + private static byte[] value(int row) { + // Include opaque expiry-like bytes to catch accidental value re-serialization. + return ByteBuffer.allocate(16).putLong(1800000000000L + row).putLong(row).array(); + } + + private static List rules() { + List rules = new ArrayList<>(); + for (String table : TABLES) { + rules.add(new GraphIdMigrationPlan.Rule(table, "01", BUSINESS)); + rules.add(new GraphIdMigrationPlan.Rule(table, "02", SYSTEM)); + } + return rules; + } + + private static void seed(Path path, int rows) throws Exception { + Files.createDirectory(path); + try (Options options = new Options().setCreateIfMissing(true).setNumLevels(8); + ColumnFamilyOptions cf = new ColumnFamilyOptions().setNumLevels(8); + RocksDB db = RocksDB.open(options, path.toString())) { + db.put("HUGEGRAPH/GRAPH_ID/DEFAULT/existing/g".getBytes(StandardCharsets.UTF_8), + Int64Value.of(7).toByteArray()); + db.put(GraphIdRewriter.slotKey(7), Int64Value.of(7).toByteArray()); + db.put("store-private-metadata".getBytes(StandardCharsets.UTF_8), new byte[]{5, 6}); + for (String table : TABLES) { + try (ColumnFamilyHandle handle = db.createColumnFamily( + new ColumnFamilyDescriptor(table.getBytes(StandardCharsets.UTF_8), cf))) { + for (int row = 0; row < rows; row++) { + db.put(handle, key(65534, row), value(row)); + } + db.put(handle, key(7, rows + 1), value(rows + 1)); + try (FlushOptions flush = new FlushOptions().setWaitForFlush(true)) { + db.flush(flush, handle); + } + } + } + try (FlushOptions flush = new FlushOptions().setWaitForFlush(true)) { + db.flush(flush); + } + } + } + + private static void seedExactBatch(Path path, int rows) throws Exception { + Files.createDirectory(path); + try (Options options = new Options().setCreateIfMissing(true).setNumLevels(8); + ColumnFamilyOptions cf = new ColumnFamilyOptions().setNumLevels(8); + RocksDB db = RocksDB.open(options, path.toString()); + ColumnFamilyHandle handle = db.createColumnFamily( + new ColumnFamilyDescriptor("g+v".getBytes(StandardCharsets.UTF_8), cf))) { + for (int row = 0; row < rows; row++) { + db.put(handle, key(65534, row * 2), value(row)); + } + try (FlushOptions flush = new FlushOptions().setWaitForFlush(true)) { + db.flush(flush, handle); + db.flush(flush); + } + } + } + + private static boolean canAcquireSourceLock(Path source, Path log) { + try { + Process contender = new ProcessBuilder( + Path.of(System.getProperty("java.home"), "bin", "java").toString(), + "-cp", System.getProperty("java.class.path"), + GraphIdRewriterProbe.class.getName(), "try-lock", source.toString()) + .redirectErrorStream(true).redirectOutput(log.toFile()).start(); + if (!contender.waitFor(10, TimeUnit.SECONDS)) { + contender.destroyForcibly(); + contender.waitFor(10, TimeUnit.SECONDS); + throw new IllegalStateException("Timed out waiting for lock contender"); + } + int exit = contender.exitValue(); + if (exit == 0) { + return true; + } + if (exit == 2) { + return false; + } + throw new IllegalStateException("Lock contender failed with exit " + exit); + } catch (Exception error) { + throw new IllegalStateException("Failed to run lock contender", error); + } + } + + private static Map physical(Path path) throws Exception { + Map result = new TreeMap<>(); + try (var files = Files.walk(path)) { + for (Path file : files.filter(Files::isRegularFile).toList()) { + result.put(path.relativize(file).toString(), Files.getLastModifiedTime(file) + ":" + + HexFormat.of().formatHex(MessageDigest.getInstance("SHA-256") + .digest(Files.readAllBytes(file)))); + } + } + return result; + } + + private static final class OpenDb implements AutoCloseable { + private final Options listing = new Options(); + private final DBOptions options = new DBOptions(); + private final ColumnFamilyOptions cf = new ColumnFamilyOptions().setNumLevels(8); + private final List handles = new ArrayList<>(); + private final Map families = new TreeMap<>(); + private final RocksDB db; + + private OpenDb(Path path, boolean readOnly) throws Exception { + List names = RocksDB.listColumnFamilies(this.listing, path.toString()); + List descriptors = new ArrayList<>(); + names.forEach(name -> descriptors.add(new ColumnFamilyDescriptor(name, this.cf))); + this.db = readOnly ? RocksDB.openReadOnly(this.options, path.toString(), descriptors, this.handles) : + RocksDB.open(this.options, path.toString(), descriptors, this.handles); + for (int i = 0; i < names.size(); i++) { + this.families.put(new String(names.get(i), StandardCharsets.UTF_8), this.handles.get(i)); + } + } + + @Override + public void close() { + this.handles.forEach(ColumnFamilyHandle::close); + this.db.close(); + this.cf.close(); + this.options.close(); + this.listing.close(); + } + } + + public static void main(String[] args) throws Exception { + RocksDB.loadLibrary(); + if (args.length > 0 && args[0].equals("try-lock")) { + try (FileChannel channel = FileChannel.open(Path.of(args[1]).resolve("LOCK"), + StandardOpenOption.WRITE); + FileLock lock = channel.tryLock()) { + if (lock == null) { + System.exit(2); + } + return; + } catch (Exception error) { + System.err.println(error); + System.exit(3); + } + } + if (args.length > 0 && args[0].equals("hold")) { + try (OpenDb db = new OpenDb(Path.of(args[1]), false)) { + Files.writeString(Path.of(args[2]), "locked"); + System.in.read(); + } + return; + } + Path root = Path.of(args[0]).toAbsolutePath(); + Files.createDirectories(root); + int rows = 2000; + Path source = root.resolve("source"); + seed(source, rows); + Map original = physical(source); + try (ColumnFamilyOptions cf = new ColumnFamilyOptions().setNumLevels(8).setMergeOperatorName("uint64add")) { + GraphIdMigrationPlan plan = GraphIdRewriter.plan(source, cf, rules()); + check(original.equals(physical(source)), "Planning changed source files/mtimes"); + check(plan.targets().get(BUSINESS) == 0 && plan.targets().get(SYSTEM) == 1, "Allocation"); + long started = System.nanoTime(); + GraphIdRewriter.Result result = GraphIdRewriter.convert(source, root.resolve("converted"), cf, plan); + long millis = TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - started); + check(result.converted() == (long) rows * TABLES.size(), "Converted count"); + check(original.equals(physical(source)), "Conversion changed source files/mtimes"); + try (OpenDb target = new OpenDb(result.database(), true)) { + for (String table : TABLES) { + for (int row = 0; row < rows; row++) { + check(Arrays.equals(value(row), target.db.get(target.families.get(table), + key(row % 2, row))), "Value/key mismatch"); + check(target.db.get(target.families.get(table), key(65534, row)) == null, + "Legacy key remains"); + } + check(Arrays.equals(value(rows + 1), target.db.get(target.families.get(table), + key(7, rows + 1))), "Unrelated data changed"); + } + check(Int64Value.parseFrom(target.db.get(GraphIdRewriter.slotKey(0))).getValue() == 0, + "Missing allocator slot 0"); + } + GraphIdRewriter.Result retry = GraphIdRewriter.convert(source, root.resolve("retry"), cf, plan); + check(result.targetDigest().equals(retry.targetDigest()), "Non-deterministic retry"); + refused(() -> GraphIdRewriter.convert(source, root.resolve("converted"), cf, plan), "Output must be new"); + refused(() -> GraphIdRewriter.convert(source, source.resolve("nested"), cf, plan), "Output must be new"); + List missing = new ArrayList<>(rules()); + missing.remove(0); + refused(() -> GraphIdRewriter.plan(source, cf, missing), "No proven owner"); + refused(() -> GraphIdRewriter.convert(source, root.resolve("interrupted"), cf, plan, + n -> { throw new IllegalStateException("injected after durable batch"); }), + "injected after durable batch"); + check(Files.exists(root.resolve("interrupted/INCOMPLETE")), "Missing incomplete marker"); + check(!Files.exists(root.resolve("interrupted/DB_VERIFIED_NOT_CLUSTER_READY")), "False certificate"); + check(original.equals(physical(source)), "Interrupted conversion changed original"); + refused(() -> GraphIdRewriter.convert(source, root.resolve("lock-protected"), cf, plan, n -> { + check(!canAcquireSourceLock(source, root.resolve("lock-contender.log")), + "Copying the cold DB released the source lock"); + throw new IllegalStateException("stop after lock retention check"); + }), "stop after lock retention check"); + check(Files.exists(root.resolve("lock-protected/INCOMPLETE")), + "Lock test output lost incomplete marker"); + + Path exactSource = root.resolve("exact-batch-source"); + seedExactBatch(exactSource, 256); + List exactRules = List.of( + new GraphIdMigrationPlan.Rule("g+v", "01", BUSINESS)); + GraphIdMigrationPlan exactPlan = GraphIdRewriter.plan(exactSource, cf, exactRules); + try { + GraphIdRewriter.convert(exactSource, root.resolve("exact-batch-interrupted"), cf, + exactPlan, n -> Thread.currentThread().interrupt()); + throw new AssertionError("Interrupted final batch returned success"); + } catch (InterruptedException expected) { + check(expected.getMessage().contains("Conversion interrupted"), expected.toString()); + } finally { + Thread.interrupted(); + } + check(Files.exists(root.resolve("exact-batch-interrupted/INCOMPLETE")), + "Final-batch interruption lost incomplete marker"); + check(!Files.exists(root.resolve( + "exact-batch-interrupted/DB_VERIFIED_NOT_CLUSTER_READY")), + "Final-batch interruption published a false certificate"); + + GraphIdRewriter.convert(source, root.resolve("after-interruption"), cf, plan); + check(original.equals(physical(source)), "Retry changed original"); + try (OpenDb changed = new OpenDb(source, false)) { + changed.db.put("new-metadata".getBytes(StandardCharsets.UTF_8), new byte[]{9}); + } + refused(() -> GraphIdRewriter.convert(source, root.resolve("stale"), cf, plan), "Stale source"); + check(!Files.exists(root.resolve("stale")), "Stale plan created output"); + try (OpenDb changed = new OpenDb(source, false)) { + changed.db.put(("HUGEGRAPH/GRAPH_ID/" + BUSINESS).getBytes(StandardCharsets.UTF_8), + Int64Value.of(12).toByteArray()); + } + refused(() -> GraphIdRewriter.plan(source, cf, rules()), "Mixed/already allocated"); + // This is the actual mapping-only counterexample, not an in-memory simulation. + try (OpenDb alias = new OpenDb(source, true)) { + check(alias.db.get(alias.families.get("g+v"), key(12, 0)) == null, "Alias unexpectedly moved data"); + check(alias.db.get(alias.families.get("g+v"), key(65534, 0)) != null, "Legacy source vanished"); + } + Path locked = root.resolve("locked"); + seed(locked, 2); + Path ready = root.resolve("lock-ready"); + Process holder = new ProcessBuilder(Path.of(System.getProperty("java.home"), "bin", "java").toString(), + "-cp", System.getProperty("java.class.path"), GraphIdRewriterProbe.class.getName(), + "hold", locked.toString(), ready.toString()).redirectErrorStream(true) + .redirectOutput(root.resolve("holder.log").toFile()).start(); + try { + long deadline = System.nanoTime() + TimeUnit.SECONDS.toNanos(10); + while (!Files.exists(ready) && holder.isAlive() && System.nanoTime() < deadline) { + Thread.sleep(10); + } + check(Files.exists(ready), "Native lock holder did not start"); + refused(() -> GraphIdRewriter.plan(locked, cf, rules()), "locked"); + } finally { + holder.getOutputStream().close(); + if (!holder.waitFor(10, TimeUnit.SECONDS)) { + holder.destroyForcibly(); + holder.waitFor(10, TimeUnit.SECONDS); + } + } + GraphIdRewriter.plan(locked, cf, rules()); + System.out.printf("PASS: %d native assertions; rekeyed=%d; elapsed_ms=%d; JVM=%s%n", + assertions, result.converted(), millis, System.getProperty("java.version")); + System.out.println("NOT TESTED: official 1.7 fixture, ownership inference, Raft, real snapshots, cluster activation."); + } + } +} diff --git a/hugegraph-store/hg-store-test/src/test/graphid-rekey/official-17-source.py b/hugegraph-store/hg-store-test/src/test/graphid-rekey/official-17-source.py new file mode 100644 index 0000000000..ed169c0fb9 --- /dev/null +++ b/hugegraph-store/hg-store-test/src/test/graphid-rekey/official-17-source.py @@ -0,0 +1,251 @@ +#!/usr/bin/env python3 +# 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. +"""Generate a bounded, fresh official 1.7 HStore source; NOT an upgrade test.""" +import hashlib +import json +import os +from pathlib import Path +import shutil +import socket +import subprocess +import sys +import tarfile +import time +import urllib.request +import zipfile + +RELEASE = "https://github.com/apache/hugegraph/releases/download/1.7.0/apache-hugegraph-incubating-1.7.0.tar.gz" +RELEASE_SHA = "e9b95b5afac156f251a12134a03a27579cca9ed7cca7e42ad0529e639b1cde1e" +FIXTURE_SHA = "52c52778263e4ba81686a86de8eac2cf4b0df7ae" +FIXTURE = f"https://raw.githubusercontent.com/SebastianGruza/hugegraph-validation/{FIXTURE_SHA}/results/upgrade-170-to-master/scripts/hg_data.py" + + +def http(path, body=None): + payload = None if body is None else json.dumps(body).encode() + request = urllib.request.Request(path, data=payload, headers={"Content-Type": "application/json"}) + with urllib.request.urlopen(request, timeout=60) as reply: + raw = reply.read() + try: + return json.loads(raw) + except json.JSONDecodeError: + return raw.decode() + + +def download(url, target, expected=None): + h = hashlib.sha256() + with urllib.request.urlopen(url, timeout=120) as response, target.open("xb") as output: + while chunk := response.read(1024 * 1024): + h.update(chunk) + output.write(chunk) + digest = h.hexdigest() + if expected and digest != expected: + raise RuntimeError(f"Checksum mismatch for {url}: {digest}") + return digest + + +def property_file(path, updates): + lines = path.read_text().splitlines() + lines = [line for line in lines if line.partition("=")[0].strip() not in updates] + path.write_text("\n".join(lines + [f"{k}={v}" for k, v in updates.items()]) + "\n") + + +def main(): + root = Path(sys.argv[1]).resolve() + root.mkdir(parents=True, exist_ok=False) + evidence = root / "evidence" + evidence.mkdir() + processes = [] + handles = [] + components = {} + forced_stops = [] + + def launch(name, command, cwd): + log = (evidence / f"{name}.log").open("wb") + handles.append(log) + process = subprocess.Popen(command, cwd=cwd, stdin=subprocess.DEVNULL, stdout=log, + stderr=subprocess.STDOUT) + processes.append((name, process)) + return process + + def wait_port(process, port): + deadline = time.monotonic() + 150 + while time.monotonic() < deadline and process.poll() is None: + try: + with socket.create_connection(("127.0.0.1", port), timeout=1): + return + except OSError: + time.sleep(1) + raise RuntimeError(f"Service did not listen on {port}; exit={process.poll()}") + + def stop(process): + if process.poll() is not None: + return + process.terminate() + try: + process.wait(timeout=35) + except subprocess.TimeoutExpired: + process.kill() + process.wait(timeout=10) + forced_stops.append(process.pid) + print("Stopped a stuck old-version process with SIGKILL; retain Raft for validation", flush=True) + + try: + archive = root / "release.tar.gz" + download(RELEASE, archive, RELEASE_SHA) + with tarfile.open(archive) as tar: + members = [entry for entry in tar.getmembers() if not Path(entry.name).name.startswith("._")] + (evidence / "excluded-appledouble.txt").write_text("\n".join( + entry.name for entry in tar.getmembers() if Path(entry.name).name.startswith("._"))) + tar.extractall(root / "release", members=members, filter="data") + for name in ("pd", "store", "server"): + matches = list((root / "release").glob(f"**/apache-hugegraph-{name}-incubating-1.7.0")) + if len(matches) != 1: + raise RuntimeError(f"Expected one {name} distribution, found {matches}") + components[name] = matches[0] + pd, store, server = (components[name] for name in ("pd", "store", "server")) + config = pd / "conf/application.yml" + config.write_text(config.read_text().replace("store-max-shard-count: 12", "store-max-shard-count: 2")) + property_file(server / "conf/rest-server.properties", { + "usePD": "true", "pd.peers": "127.0.0.1:8686", "memory_monitor.threshold": "1", + "batch.max_write_threads": "2"}) + property_file(server / "conf/graphs/hugegraph.properties", { + "backend": "hstore", "pd.peers": "127.0.0.1:8686", "vertex.cache_capacity": "1000", + "edge.cache_capacity": "1000"}) + java17 = str(Path(os.environ["JAVA_HOME_17_X64"]) / "bin/java") + java11 = str(Path(os.environ["JAVA_HOME_11_X64"]) / "bin/java") + jvm = ["-Xms128m", "-Xmx768m", "-XX:ActiveProcessorCount=2", + "--add-opens=java.base/java.lang=ALL-UNNAMED", + "--add-opens=java.base/java.nio=ALL-UNNAMED", + "--add-exports=java.base/sun.nio.ch=ALL-UNNAMED", + "--add-exports=java.base/jdk.internal.reflect=ALL-UNNAMED"] + for name, component, pattern, port in (("pd", pd, "hg-pd-service-*.jar", 8686), + ("store", store, "hg-store-node-*.jar", 8500)): + jars = list((component / "lib").glob(pattern)) + if len(jars) != 1: + raise RuntimeError(f"Ambiguous {name} executable: {jars}") + process = launch(name, [java17, *jvm, f"-Dspring.config.location={component}/conf/application.yml", + "-jar", str(jars[0])], component) + wait_port(process, port) + store_process = processes[-1][1] + jars = sorted((server / "lib").rglob("*.jar")) + # Prefer released modular Netty classes over an older bundled fat netty-all. + modular_netty = [p for p in jars if p.name.startswith("netty-") and not p.name.startswith("netty-all-")] + hugegraph = [p for p in jars if p.name.startswith("hugegraph")] + ordered = list(dict.fromkeys(modular_netty + hugegraph + jars)) + ordered += sorted((server / "ext").rglob("*.jar")) + sorted((server / "plugins").rglob("*.jar")) + classpath = os.pathsep.join(map(str, ordered)) + (evidence / "server-classpath.txt").write_text("\n".join(p.name for p in ordered)) + for attempt in range(3): + with (evidence / f"init-{attempt}.log").open("wb") as output: + result = subprocess.run([java11, *jvm, "-cp", classpath, "org.apache.hugegraph.cmd.InitStore", + "conf/rest-server.properties"], cwd=server, input=b"\n", + stdout=output, stderr=subprocess.STDOUT, timeout=120) + if result.returncode == 0: + break + time.sleep(3) + else: + raise RuntimeError("Official 1.7 InitStore failed") + sv = launch("server", [java11, *jvm, "-Dlog4j.configurationFile=conf/log4j2.xml", "-cp", classpath, + "org.apache.hugegraph.dist.HugeGraphServer", "conf/gremlin-server.yaml", + "conf/rest-server.properties"], server) + wait_port(sv, 8080) + deadline = time.monotonic() + 90 + while True: + try: + http("http://127.0.0.1:8080/graphs") + break + except Exception: + if sv.poll() is not None or time.monotonic() > deadline: + raise + time.sleep(1) + fixture = root / "hg_data.py" + fixture_digest = download(FIXTURE, fixture) + for mode in ("write", "read"): + with (evidence / f"fixture-{mode}.log").open("wb") as output: + subprocess.run([sys.executable, str(fixture), mode, "http://127.0.0.1:8080", + str(evidence / f"fixture-{mode}.json")], stdout=output, stderr=subprocess.STDOUT, + check=True, timeout=180) + baseline = json.loads((evidence / "fixture-read.json").read_text()) + checks = [entry for entry in baseline if entry["step"].startswith("gremlin ")] + if len(checks) != 16 or any(entry["status"] != 200 for entry in checks): + raise RuntimeError("The official source did not pass all 16 fixture requests") + by_name = {entry["step"]: entry["data"] for entry in checks} + if by_name["gremlin g.V().count()"] != [29] or by_name["gremlin g.E().count()"] != [63]: + raise RuntimeError(f"Unexpected baseline counts: {by_name}") + for graph in ("hugegraph", "~sys_graph"): + query = "['~variables','~task','~server'].collectEntries { n -> [(n): graph.existsVertexLabel(n) ? [id:graph.vertexLabel(n).id().asLong(),name:n] : null] }" + try: + result = http("http://127.0.0.1:8080/gremlin", { + "gremlin": query, "language": "gremlin-groovy", "bindings": {}, + "aliases": {"graph": "DEFAULT-" + graph, "g": "__g_DEFAULT-" + graph}}) + except Exception as error: + result = {"error": str(error)} + (evidence / f"hidden-labels-{graph}.json").write_text(json.dumps(result, default=str)) + stop(sv) + snapshot_response = http("http://127.0.0.1:8520/test/snapshot") + (evidence / "snapshot-response.json").write_text(json.dumps(snapshot_response)) + deadline = time.monotonic() + 120 + while True: + snapshots = list((store / "storage/raft").glob("*/snapshot/snapshot_*/data/CURRENT")) + if len(snapshots) >= 2: + break + if store_process.poll() is not None or time.monotonic() > deadline: + raise RuntimeError("Expected two persisted partition snapshots") + time.sleep(1) + time.sleep(3) + stop(store_process) + stop(processes[0][1]) + shutil.copytree(store / "storage", evidence / "store-data") + if (pd / "pd_data").exists(): + shutil.copytree(pd / "pd_data", evidence / "pd-data") + (evidence / "SOURCE_GENERATED_NOT_UPGRADED.json").write_text(json.dumps({ + "release_sha256": RELEASE_SHA, "fixture_revision": FIXTURE_SHA, + "fixture_script_sha256": fixture_digest, "vertices": 29, "edges": 63, + "gremlin_requests": 16, "snapshots_found": len(snapshots), "forced_stops": forced_stops, + "warning": "Source generation only; no conversion, no version upgrade tested"}, indent=2)) + print("PASS: official 1.7 source created, 29 V / 63 E, 16 successful Gremlin requests; NOT UPGRADED", flush=True) + finally: + for _, process in reversed(processes): + if process.poll() is None: + try: + stop(process) + except Exception as error: + print(f"Cleanup: {error}", flush=True) + for name, component in components.items(): + for folder in ("conf", "logs"): + if (component / folder).exists(): + shutil.copytree(component / folder, evidence / name / folder, dirs_exist_ok=True) + if "server" in components: + libraries = evidence / "released-codecs" + libraries.mkdir(exist_ok=True) + prefixes = ("hugegraph-core-", "hugegraph-common-", "hg-store-core-", "hg-store-rocksdb-", + "hg-pd-grpc-", "hugegraph-struct-", "jraft-core-", "slf4j-api-", "commons-io-") + for jar in (components["server"] / "lib").glob("*.jar"): + if jar.name.startswith(prefixes): + shutil.copy2(jar, libraries / jar.name) + for executable in (components["store"] / "lib").glob("hg-store-node-*.jar"): + with zipfile.ZipFile(executable) as jar: + for entry in jar.namelist(): + if entry.startswith("BOOT-INF/lib/") and Path(entry).name.startswith(prefixes): + (libraries / Path(entry).name).write_bytes(jar.read(entry)) + repo = Path(__file__).resolve().parents[5] + subprocess.run(["git", "archive", "--format=tar.gz", "--output=" + str(evidence / "source-tree.tar.gz"), + "HEAD", "hugegraph-store", "hugegraph-struct", "hugegraph-server/hugegraph-core", + "hugegraph-commons", "hugegraph-pd"], cwd=repo, check=True) + for handle in handles: + handle.close() + + +if __name__ == "__main__": + main()