You are viewing a plain text version of this content. The canonical link for it is here.
Posted to oak-commits@jackrabbit.apache.org by js...@apache.org on 2026/09/14 13:09:12 UTC
(jackrabbit-oak) branch record-repacker-prototype created (now d86e9aa2bb)
This is an automated email from the ASF dual-hosted git repository.
jsedding pushed a change to branch record-repacker-prototype
in repository https://gitbox.apache.org/repos/asf/jackrabbit-oak.git
at d86e9aa2bb RecordRepacker: research prototype for record-level compaction
This branch includes the following new commits:
new d86e9aa2bb RecordRepacker: research prototype for record-level compaction
The 1 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
(jackrabbit-oak) 01/01: RecordRepacker: research prototype for record-level compaction
Posted by js...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
jsedding pushed a commit to branch record-repacker-prototype
in repository https://gitbox.apache.org/repos/asf/jackrabbit-oak.git
commit d86e9aa2bb3f19c5e45ab9bad1faf8d07498261d
Author: Julian Sedding <js...@apache.org>
AuthorDate: Wed Sep 9 22:40:27 2026 +0200
RecordRepacker: research prototype for record-level compaction
Repacks the records reachable from a root into densely-packed segments of a
target generation. Adds content deduplication, off-heap memory-mapped
alias/dedup maps, and optional parallel repacking, plus RecordRepackerTest
with round-trip tests and compaction/read-locality benchmarks against
CheckpointCompactor and ParallelCompactor.
Co-Authored-By: Claude Opus 4.8 <no...@anthropic.com>
---
.../docs/record-repacker-research-slides.md | 216 ++
.../apache/jackrabbit/oak/segment/DedupCache.java | 76 +
.../oak/segment/MmapWindowDedupCache.java | 323 ++
.../jackrabbit/oak/segment/RecordRepacker.java | 1384 ++++++++
.../jackrabbit/oak/segment/ShardedRecordIdMap.java | 172 +
.../oak/segment/SpillableRecordIdMap.java | 304 ++
.../jackrabbit/oak/segment/WindowDedupCache.java | 187 ++
.../jackrabbit/oak/segment/RecordRepackerTest.java | 3295 ++++++++++++++++++++
8 files changed, 5957 insertions(+)
diff --git a/oak-segment-tar/docs/record-repacker-research-slides.md b/oak-segment-tar/docs/record-repacker-research-slides.md
new file mode 100644
index 0000000000..6833ca1256
--- /dev/null
+++ b/oak-segment-tar/docs/record-repacker-research-slides.md
@@ -0,0 +1,216 @@
+---
+marp: true
+paginate: true
+style: |
+ section { font-size: 18px; }
+ h1 { font-size: 30px; }
+ h2 { font-size: 24px; }
+ table { font-size: 0.85em; }
+ section > p { margin: 0.4em 0; }
+---
+
+# Record-level Compaction for Oak Segment-Stores<br><span style="font-size:0.7em">*Research Summary*</span>
+
+Evaluation of a record-level compactor prototype, benchmarked against existing compactors.
+
+---
+
+## Core Idea
+
+- **Today:** semantic compaction - *copies the entire live repo* into a new GC generation at the **NodeState** level
+- **Insight:** persisted data is a *DAG of records*
+- **Idea:** structural compaction - *copy reachable records to a new GC generation*
+- **Caveat:** `RecordId` is `(SegmentId, recordNumber)`. Moving a record changes its id; records are not freely mobile.
+- **Solution:** `RecordId` substitution using a temporary alias table
+- **Bonus:** record deduplication to reduce number of records and store size
+
+---
+
+## Highlights
+
+- **~1.9× faster**
+- **~25% smaller and ~40% fewer records**
+- **~35% less heap** - but requires memory-mapped off-heap storage
+
+Numbers are based on a real segmentstore with a compacted size of ~823 MB (tar-files) downloaded from a customer's RDE
+environment. Garbage is created artificially. Results are compared to `CheckpointCompactor` and `ParallelCompactor`.
+
+---
+
+## At a glance: record-repacker vs compactor
+
+<div style="display:flex; gap:1.5rem; align-items:flex-start;">
+<div>
+
+**Compaction**
+
+| scenario | time | store size | record count | peak heap |
+|----------------------|----------:|------------:|-------------:|-----------:|
+| compactor 1x | 17.9 s | 824 MB | 16.8 M | 1508 MB |
+| compactor 16x | 8.8 s | 843 MB | 17.0 M | 1681 MB |
+| record-repacker 1x | 9.3 s | 629 MB | 10.08 M | 997 MB |
+| record-repacker 16x | **4.6 s** | 629 MB | 10.12 M | 1073 MB |
+
+</div>
+<div>
+
+**Read locality** (cold reads, 32 MB cache — segment fetches)
+
+| scenario | full (distinct) | list-widest | deep-path |
+|---------------------|----------------:|------------:|----------:|
+| compactor 1x | 3264 | 164 | 45 |
+| compactor 16x | 3344 | 230 | 54 |
+| record-repacker 1x | 2488 | 142 | 40 |
+| record-repacker 16x | 2510 | **152** | 46 |
+
+</div>
+</div>
+
+**~1.9× faster · ~35% less heap · ~25% smaller · ~40% fewer records — and best-in-class read locality**
+
+Note
+`compactor` refers to `CheckpointCompactor` or `ParallelCompactor`
+`record-repacker` numbers refer to the `hot-dedup+idx` configuration of the repacker
+
+---
+
+## The approach: `RecordRepacker`
+
+- walk the **reachable DAG**
+- read record bytes
+- replace `RecordId` references in the bytes with their aliases
+- write patched record bytes
+- record the new `RecordId` in the alias table
+
+**bounded heap:** the alias table is stored in a memory-mapped file
+
+---
+
+## Record deduplication
+
+Records can be **byte-identical after reference translation** yet start as separate records.
+
+Before writing a record's bytes, check if the same bytes are in the deduplication cache. If
+present, only record an alias mapping from the old `RecordId` to the already written `RecordId`.
+
+Three modes benchmarked:
+
+- **preserve** — no deduplication, only existing physical sharing preserved
+- **full-dedup** — *every* content-equal record collapsed into one (exact, unbounded off-heap dedup table)
+- **hot-dedup** — deduplicates only records that recur within a bounded **recency window** (memory-mapped, off-heap)
+
+**Examples of content-equal records:**
+- property-index mirrors
+- empty version-storage buckets
+- permission entries / ACLs
+
+**hot-dedup is a good default** — a window of ~10% of the records recovers full-dedup's size within ~2%
+
+---
+
+## Compaction benchmark setup & scenarios
+
+**Store:**
+- real compacted ~823 MB segment store
+- garbage injected, resulting in ~1 GB (20.6 M records, several garbage generations, 2 checkpoints)
+
+**Comparison Matrix** = five approaches × {single-threaded, 16 threads}:
+
+| approach | what it is |
+|-------------------|--------------------------------------------------------------------------|
+| **preserve** | repacker preserves existing record structure (physical sharing) |
+| **full-dedup** | repacker collapsing *every* content-equal record (exact) |
+| **hot-dedup** | repacker collapsing content-equal records seen within a window (default) |
+| **hot-dedup+idx** | like `hot-dedup`, but repacked in three stages |
+| **compactor** | existing `CheckpointCompactor` / `ParallelCompactor` |
+
+- **16 threads (concurrent):** DAG split into sub-roots; repacked in parallel; requires shared alias and deduplication data
+
+---
+
+## Results: compaction (speed / size / heap)
+
+| scenario | time | store size | record count | peak heap |
+|-------------------------------|----------:|-----------:|-------------:|-----------:|
+| input (uncompacted) | — | 1014 MB | 20.6 M | — |
+| preserve 1x | 7.2 s | 770 MB | 15.2 M | 955 MB |
+| preserve 16x | 4.7 s | 872 MB | 17.7 M | 1126 MB |
+| full-dedup 1x | 9.7 s | **617 MB** | **9.81 M** | 983 MB |
+| full-dedup 16x | 5.1 s | **619 MB** | **9.82 M** | 1147 MB |
+| hot-dedup 1x | 9.3 s | 629 MB | 10.07 M | 970 MB |
+| hot-dedup 16x | **4.3 s** | 647 MB | 10.40 M | 1149 MB |
+| hot-dedup+idx 1x *(default)* | 9.3 s | 629 MB | 10.08 M | 997 MB |
+| hot-dedup+idx 16x *(default)* | **4.6 s** | 629 MB | 10.12 M | 1073 MB |
+| compactor 1x (Checkpoint) | 17.9 s | 824 MB | 16.8 M | 1508 MB |
+| compactor 16x (Parallel) | 8.8 s | 843 MB | 17.0 M | 1681 MB |
+
+---
+
+## Read Locality Benchmark Setup
+
+The store after each compaction scenario is tested.
+
+**Read-locality:**
+- read from a **cold** store
+- count **segment fetches** and **distinct segments**
+
+**Access patterns:**
+- **full** — depth first traversal of the whole tree
+- **list-widest** — list children of the widest node (uuid index, **165,060** nodes)
+- **deep-path** — traverse one root-to-leaf path (depth 47)
+
+---
+
+## Lessons on locality
+
+Early benchmark results showed `list-widest` to be an order of magnitude worse when deduplication
+was used in a parallel setup. `list-widest` iterates over the children of the UUID index.
+
+The key insight here was that the record holding the UUID in a deduplicated repository can either
+be colocated with the index or with the node that has the `jcr:uuid` property - but not both.
+
+Tests confirmed that forcing `/oak:index` to be repacked last during any single-threaded
+run using deduplication yielded the same poor result.
+
+Conversely, forcing `/oak:index` to be repacked first, even in parallel, got rid of this
+artifact.
+
+Extrapolating from this learning, a three-stage approach was chosen:
+
+1. repack `/oak:index` (internal path `/root/oak:index`)
+2. repack the live root (internal path `/root`)
+3. repack all checkpoints (internal path `/`)
+
+The underlying assumption is that it should be beneficial to pack the live root more densely
+at the expense of slightly more fragmented reads from checkpoints.
+
+---
+
+## Results: read locality (cold reads, 32MB segment cache)
+
+**fetch** = segments loaded (includes reloads due to cache eviction);
+**distinct** = distinct segments loaded.
+For list-widest and deep-path fetch==distinct, because all repeat accesses are cached.
+
+| scenario | full (fetch→distinct) | list-widest | deep-path |
+|------------------------------|----------------------:|------------:|----------:|
+| preserve 1x | 3188 → 3052 | 143 | 39 |
+| preserve 16x | 3698 → 3398 | 202 | 40 |
+| full-dedup 1x | 4434 → **2441** | 143 | **36** |
+| full-dedup 16x | 4391 → **2457** | 1094 | 38 |
+| hot-dedup 1x | 3984 → **2488** | 143 | 40 |
+| hot-dedup 16x | 4578 → **2515** | 1046 | 38 |
+| hot-dedup+idx 1x | 4010 → **2488** | **142** | 40 |
+| hot-dedup+idx 16x | 4676 → **2510** | **152** | 46 |
+| compactor 1x | 3552 → 3264 | 164 | 45 |
+| compactor 16x | 4326 → 3344 | 230 | 54 |
+
+---
+
+## Conclusions & status
+
+- ~25% smaller, ~40% fewer records, faster - record-level repacking beats the semantic compactors on this store
+- hot-dedup+idx achieves the best overall results
+- tradeoff: deduplication vs locality - subtlety: denser stores have inherently higher locality
+- a deduplication window covering ~10% of the records nearly ties with full deduplication (+ ~2% size)
+- parallel compaction/repacking results in slightly worse locality, independent of the mode
diff --git a/oak-segment-tar/src/main/java/org/apache/jackrabbit/oak/segment/DedupCache.java b/oak-segment-tar/src/main/java/org/apache/jackrabbit/oak/segment/DedupCache.java
new file mode 100644
index 0000000000..91fff024f1
--- /dev/null
+++ b/oak-segment-tar/src/main/java/org/apache/jackrabbit/oak/segment/DedupCache.java
@@ -0,0 +1,76 @@
+/*
+ * 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.jackrabbit.oak.segment;
+
+import java.io.Closeable;
+import java.io.IOException;
+
+import org.jetbrains.annotations.NotNull;
+import org.jetbrains.annotations.Nullable;
+
+/**
+ * Content-addressed cache mapping a record's content-hash key to the {@link RecordId} of an
+ * already-emitted, byte-identical record, so the {@link RecordRepacker} can reuse it instead of
+ * emitting a duplicate. Two implementations:
+ * <ul>
+ * <li>{@link ShardedRecordIdMap} - exact and unbounded (memory-mapped/off-heap): every duplicate is
+ * collapsed, at the cost of a whole-store filesystem spill and a canonical copy that may live far
+ * from its referrers;</li>
+ * <li>{@link MmapWindowDedupCache} - a bounded recency window in a fixed-size memory-mapped file
+ * (the default): only records whose content recurs while still in the window are deduplicated,
+ * older entries are forgotten. Off-heap and its spill is bounded to the window size;</li>
+ * <li>{@link WindowDedupCache} - the same bounded recency window kept in the Java heap: no spill at
+ * all, but the window's entries count against the heap.</li>
+ * </ul>
+ * In both cases a miss means the caller emits a fresh record; the repacker's exact alias still
+ * preserves the input's physical record sharing regardless of this cache.
+ */
+interface DedupCache extends Closeable {
+
+ /** @return the mapped id for {@code key}, or {@code null} if none is (currently) held. */
+ @Nullable
+ RecordId get(@NotNull byte[] key);
+
+ /**
+ * Map {@code key} to {@code value} unless already mapped. Returns the existing mapping if one is
+ * present (leaving it unchanged), otherwise {@code null} (the caller's {@code value} was stored).
+ */
+ @Nullable
+ RecordId putIfAbsent(@NotNull byte[] key, @NotNull RecordId value) throws IOException;
+
+ /**
+ * Return the id mapped to {@code key}, computing and storing it via {@code writer} if absent. The
+ * computation runs while the (shard) lock is held, so two callers with the same key never both
+ * emit.
+ */
+ @NotNull
+ RecordId computeIfAbsent(@NotNull byte[] key, @NotNull ShardedRecordIdMap.RecordIdSupplier writer)
+ throws IOException;
+
+ /** Number of deduplication hits (content found), or {@code -1} if this cache does not track it. */
+ default long getHits() {
+ return -1;
+ }
+
+ /** Number of deduplication misses (fresh emits), or {@code -1} if this cache does not track it. */
+ default long getMisses() {
+ return -1;
+ }
+
+ @Override
+ void close();
+}
diff --git a/oak-segment-tar/src/main/java/org/apache/jackrabbit/oak/segment/MmapWindowDedupCache.java b/oak-segment-tar/src/main/java/org/apache/jackrabbit/oak/segment/MmapWindowDedupCache.java
new file mode 100644
index 0000000000..6f1720ddac
--- /dev/null
+++ b/oak-segment-tar/src/main/java/org/apache/jackrabbit/oak/segment/MmapWindowDedupCache.java
@@ -0,0 +1,323 @@
+/*
+ * 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.jackrabbit.oak.segment;
+
+import static java.nio.channels.FileChannel.MapMode.READ_WRITE;
+import static java.util.Objects.requireNonNull;
+
+import java.io.IOException;
+import java.io.RandomAccessFile;
+import java.lang.reflect.Field;
+import java.lang.reflect.Method;
+import java.nio.ByteBuffer;
+import java.nio.MappedByteBuffer;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.util.concurrent.atomic.LongAdder;
+
+import org.jetbrains.annotations.NotNull;
+import org.jetbrains.annotations.Nullable;
+
+/**
+ * An <em>off-heap</em>, bounded, sliding-window {@link DedupCache}: it has the same inexact recency
+ * semantics as {@link WindowDedupCache} - a record is deduplicated only against recently emitted
+ * records still resident in the window, older ones are forgotten and re-emitted - but the window
+ * lives in a fixed-size memory-mapped file instead of the Java heap. This keeps the heap at the
+ * exact cache's floor (the window entries do not count against it) while bounding the on-disk
+ * footprint to the window size, unlike the exact {@link ShardedRecordIdMap}, whose spill grows with
+ * the whole store.
+ * <p>
+ * The table is a fixed-capacity, never-growing, {@value #WAYS}-way set-associative hash table. A key
+ * hashes to a set of {@value #WAYS} consecutive slots; a lookup returns the value of the slot whose
+ * full key matches (comparing the whole content hash, so a false deduplication is impossible) and
+ * marks it "recently used". An insert fills the first free slot in the set or, when the set is full,
+ * evicts a not-recently-used slot chosen by a CLOCK second-chance scan (a spare bit in the occupancy
+ * byte tracks reuse), approximating in-heap LRU without any per-entry recency list. Eviction never
+ * clears a slot (occupied stays occupied), so the effective window is "a reused key survives, a
+ * one-off ('cold') record falls out and is re-emitted" - exactly the recency semantics of
+ * {@link WindowDedupCache}, at ~capacity most-recent entries.
+ * <p>
+ * Sharded and thread-safe: the slot array is partitioned into {@code shardCount} regions, each
+ * guarded by its own monitor, and {@link #computeIfAbsent} emits while holding it, so concurrent
+ * workers with the same content converge on one copy (for content still in the window). A single
+ * mapping backs all shards; the backing descriptor is released right after mapping and the file is
+ * deleted on {@link #close()}.
+ */
+final class MmapWindowDedupCache implements DedupCache {
+
+ /** Slots probed per key (set associativity). Small, so every lookup/insert touches only a few slots. */
+ private static final int WAYS = 8;
+
+ /** Occupancy-byte flag: the slot holds an entry. */
+ private static final byte OCCUPIED = 1;
+
+ /** Occupancy-byte flag: the entry was reused since it was last considered for eviction (CLOCK). */
+ private static final byte REFERENCED = 2;
+
+ /** Serialised value width: msb (8) + lsb (8) + recordNumber (4). */
+ private static final int VALUE_WIDTH = 20;
+
+ @NotNull
+ private final SegmentIdProvider idProvider;
+
+ private final int keyWidth;
+
+ private final int slotWidth;
+
+ /** Slots per shard, a power of two (so the intra-shard home index is a mask). */
+ private final int perShard;
+
+ private final int shardCount;
+
+ private final Object[] locks;
+
+ private final LongAdder hits = new LongAdder();
+
+ private final LongAdder misses = new LongAdder();
+
+ private Path file;
+
+ private MappedByteBuffer buffer;
+
+ /**
+ * @param window desired number of most-recent entries to retain (the window size); split evenly
+ * across shards and rounded up per shard to a power of two.
+ * @param shardCount number of independently-locked shards (match the repacker's concurrency).
+ */
+ MmapWindowDedupCache(@NotNull SegmentIdProvider idProvider, int keyWidth, int window, int shardCount)
+ throws IOException {
+ this.idProvider = requireNonNull(idProvider);
+ this.keyWidth = keyWidth;
+ this.slotWidth = 1 + keyWidth + VALUE_WIDTH;
+ this.shardCount = Math.max(1, shardCount);
+ this.perShard = perShardSlots(window, this.shardCount, this.slotWidth);
+ this.locks = new Object[this.shardCount];
+ for (int i = 0; i < this.shardCount; i++) {
+ locks[i] = new Object();
+ }
+ long bytes = (long) perShard * this.shardCount * slotWidth;
+ this.file = Files.createTempFile("oak-repack-dedup-window-", ".map");
+ this.file.toFile().deleteOnExit();
+ this.buffer = mapFile(file, bytes);
+ }
+
+ @Override
+ @Nullable
+ public RecordId get(@NotNull byte[] key) {
+ int shard = shardFor(key);
+ synchronized (locks[shard]) {
+ return getInShard(shard, key);
+ }
+ }
+
+ @Override
+ @Nullable
+ public RecordId putIfAbsent(@NotNull byte[] key, @NotNull RecordId value) {
+ int shard = shardFor(key);
+ synchronized (locks[shard]) {
+ RecordId existing = getInShard(shard, key);
+ if (existing != null) {
+ return existing;
+ }
+ putInShard(shard, key, value);
+ return null;
+ }
+ }
+
+ @Override
+ @NotNull
+ public RecordId computeIfAbsent(@NotNull byte[] key,
+ @NotNull ShardedRecordIdMap.RecordIdSupplier writer) throws IOException {
+ int shard = shardFor(key);
+ synchronized (locks[shard]) {
+ RecordId existing = getInShard(shard, key);
+ if (existing != null) {
+ hits.increment();
+ return existing;
+ }
+ RecordId created = writer.get();
+ misses.increment();
+ putInShard(shard, key, created);
+ return created;
+ }
+ }
+
+ @Override
+ public void close() {
+ unmap(buffer);
+ buffer = null;
+ if (file != null) {
+ file.toFile().delete();
+ file = null;
+ }
+ }
+
+ /** Deduplication hits (content found in the window). */
+ @Override
+ public long getHits() {
+ return hits.sum();
+ }
+
+ /** Deduplication misses (content not in the window - emitted fresh). */
+ @Override
+ public long getMisses() {
+ return misses.sum();
+ }
+
+ // -- internals -----------------------------------------------------------
+
+ private int shardFor(@NotNull byte[] key) {
+ // The key is a content hash; route by its leading eight bytes, disjoint from the slot bits.
+ return (int) Math.floorMod(WindowDedupCache.readLong(key, 0), shardCount);
+ }
+
+ @Nullable
+ private RecordId getInShard(int shard, @NotNull byte[] key) {
+ int home = homeIndex(key);
+ for (int w = 0; w < WAYS; w++) {
+ int pos = slotPos(shard, (home + w) & (perShard - 1));
+ byte occ = buffer.get(pos);
+ if (occ != 0 && keyEquals(pos, key)) {
+ if ((occ & REFERENCED) == 0) {
+ buffer.put(pos, (byte) (occ | REFERENCED)); // reused: give it a second chance
+ }
+ return readValue(pos);
+ }
+ }
+ return null;
+ }
+
+ private void putInShard(int shard, @NotNull byte[] key, @NotNull RecordId value) {
+ int home = homeIndex(key);
+ int emptyPos = -1;
+ for (int w = 0; w < WAYS; w++) {
+ int pos = slotPos(shard, (home + w) & (perShard - 1));
+ byte occ = buffer.get(pos);
+ if (occ == 0) {
+ if (emptyPos < 0) {
+ emptyPos = pos;
+ }
+ } else if (keyEquals(pos, key)) {
+ buffer.put(pos, (byte) (occ | REFERENCED)); // present already: mark reused, refresh id
+ writeValue(pos, value);
+ return;
+ }
+ }
+ // A new key: take the first free slot in the set, or, if the set is full, evict a
+ // not-recently-used slot (CLOCK). Eviction never clears occupancy, so no probe hole is made.
+ writeSlot(emptyPos >= 0 ? emptyPos : clockVictim(shard, home), key, value);
+ }
+
+ /**
+ * CLOCK second-chance victim for a full set: return the first slot whose {@link #REFERENCED} bit
+ * is clear (a cold, not-recently-reused entry), clearing (aging) the bit of each recently-used
+ * slot it skips. If every slot was recently used they are all aged and the home slot is evicted.
+ */
+ private int clockVictim(int shard, int home) {
+ for (int w = 0; w < WAYS; w++) {
+ int pos = slotPos(shard, (home + w) & (perShard - 1));
+ byte occ = buffer.get(pos);
+ if ((occ & REFERENCED) == 0) {
+ return pos;
+ }
+ buffer.put(pos, (byte) (occ & ~REFERENCED)); // second chance: age this slot
+ }
+ return slotPos(shard, home);
+ }
+
+ private int homeIndex(@NotNull byte[] key) {
+ // Use the trailing eight bytes for the intra-shard index, decorrelated from the shard routing.
+ return (int) (WindowDedupCache.readLong(key, 8) & (perShard - 1));
+ }
+
+ private int slotPos(int shard, int localIndex) {
+ return (shard * perShard + localIndex) * slotWidth;
+ }
+
+ private boolean keyEquals(int pos, @NotNull byte[] key) {
+ int base = pos + 1;
+ for (int i = 0; i < keyWidth; i++) {
+ if (buffer.get(base + i) != key[i]) {
+ return false;
+ }
+ }
+ return true;
+ }
+
+ private void writeSlot(int pos, @NotNull byte[] key, @NotNull RecordId value) {
+ buffer.put(pos, OCCUPIED); // fresh entry starts un-referenced: a cold one-off is evicted first
+ int base = pos + 1;
+ for (int i = 0; i < keyWidth; i++) {
+ buffer.put(base + i, key[i]);
+ }
+ writeValue(pos, value);
+ }
+
+ private void writeValue(int pos, @NotNull RecordId value) {
+ SegmentId segmentId = value.getSegmentId();
+ int vbase = pos + 1 + keyWidth;
+ buffer.putLong(vbase, segmentId.getMostSignificantBits());
+ buffer.putLong(vbase + 8, segmentId.getLeastSignificantBits());
+ buffer.putInt(vbase + 16, value.getRecordNumber());
+ }
+
+ private RecordId readValue(int pos) {
+ int vbase = pos + 1 + keyWidth;
+ long msb = buffer.getLong(vbase);
+ long lsb = buffer.getLong(vbase + 8);
+ int recordNumber = buffer.getInt(vbase + 16);
+ return new RecordId(idProvider.newSegmentId(msb, lsb), recordNumber);
+ }
+
+ /**
+ * Slots per shard: a power of two of at least {@value #WAYS}, sized to hold ~{@code window} entries
+ * across all shards, and capped so the whole table fits a single {@code int}-addressed mapping.
+ */
+ private static int perShardSlots(int window, int shardCount, int slotWidth) {
+ int want = Math.max(WAYS, (Math.max(1, window) + shardCount - 1) / shardCount);
+ int pow2 = Integer.highestOneBit(want - 1) << 1;
+ long maxTotalSlots = Integer.MAX_VALUE / slotWidth;
+ int maxPerShard = Integer.highestOneBit((int) Math.max(WAYS, maxTotalSlots / shardCount));
+ return Math.min(pow2, maxPerShard);
+ }
+
+ private static MappedByteBuffer mapFile(Path file, long bytes) throws IOException {
+ // The mapping outlives the channel/descriptor, so the backing file is only opened long enough
+ // to size and map it. This keeps the descriptor count low regardless of shard count.
+ try (RandomAccessFile raf = new RandomAccessFile(file.toFile(), "rw")) {
+ raf.setLength(bytes);
+ return raf.getChannel().map(READ_WRITE, 0, bytes);
+ }
+ }
+
+ /** Best-effort unmap so the (deleted) file's pages are released promptly rather than at GC. */
+ private static void unmap(MappedByteBuffer buffer) {
+ if (buffer == null) {
+ return;
+ }
+ try {
+ Class<?> unsafeClass = Class.forName("sun.misc.Unsafe");
+ Field theUnsafe = unsafeClass.getDeclaredField("theUnsafe");
+ theUnsafe.setAccessible(true);
+ Object unsafe = theUnsafe.get(null);
+ Method invokeCleaner = unsafeClass.getMethod("invokeCleaner", ByteBuffer.class);
+ invokeCleaner.invoke(unsafe, buffer);
+ } catch (Throwable t) {
+ // The mapping will be released when the buffer is garbage collected.
+ }
+ }
+}
diff --git a/oak-segment-tar/src/main/java/org/apache/jackrabbit/oak/segment/RecordRepacker.java b/oak-segment-tar/src/main/java/org/apache/jackrabbit/oak/segment/RecordRepacker.java
new file mode 100644
index 0000000000..aff8143161
--- /dev/null
+++ b/oak-segment-tar/src/main/java/org/apache/jackrabbit/oak/segment/RecordRepacker.java
@@ -0,0 +1,1384 @@
+/*
+ * 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.jackrabbit.oak.segment;
+
+import static java.util.Objects.requireNonNull;
+import static org.apache.jackrabbit.oak.segment.SegmentStream.BLOCK_SIZE;
+
+import java.io.IOException;
+import java.nio.charset.StandardCharsets;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.HashSet;
+import java.util.LinkedHashMap;
+import java.util.LinkedHashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.TreeMap;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+
+import org.apache.commons.codec.digest.MurmurHash3;
+import org.apache.jackrabbit.oak.api.Type;
+import org.apache.jackrabbit.oak.segment.spi.persistence.GCGeneration;
+import org.apache.jackrabbit.oak.spi.blob.BlobStore;
+import org.apache.jackrabbit.oak.spi.state.ChildNodeEntry;
+import org.apache.jackrabbit.oak.spi.state.NodeState;
+import org.jetbrains.annotations.NotNull;
+import org.jetbrains.annotations.Nullable;
+
+/**
+ * Prototype of a <em>record-level</em> compactor. In contrast to the semantic
+ * compactors ({@link ClassicCompactor}, {@code ParallelCompactor}) which rebuild
+ * the content tree from {@link org.apache.jackrabbit.oak.spi.state.NodeState}
+ * instances, this class repacks the raw records reachable from a root into new,
+ * densely packed segments of a target {@link GCGeneration}.
+ * <p>
+ * The segment format guarantees that a record only ever references records that
+ * were written before it, so the reachable record set forms a DAG with edges
+ * pointing backwards in time. This class walks that DAG in <em>post-order</em>:
+ * a record is only emitted once all the records it references have been emitted,
+ * at which point their new {@link RecordId}s are known and can be substituted for
+ * the old ones (a record cannot be copied verbatim, because moving it changes its
+ * {@code RecordId}). The old → new mapping is kept in an
+ * <em>alias table</em> ({@link #alias}) which also deduplicates records that were
+ * already physically shared in the source (i.e. reachable via the same source
+ * record id through multiple referrers).
+ * <p>
+ * In addition, a content-addressed cache ({@link #dedupCache}) performs
+ * <em>record deduplication</em>: records that are byte-for-byte identical after
+ * their references have been translated are emitted only once, even when they had
+ * distinct source record ids (i.e. were <em>not</em> physically shared in the
+ * source). Nodes are keyed by their stable id — which uniquely identifies
+ * a node's content — so two records for the same logical node collapse to
+ * one and their (identical) subtree is not walked twice; templates, values, strings,
+ * blocks and lists are keyed by their translated content. This is the same kind of
+ * new sharing the semantic compactors discover through content equality.
+ * <p>
+ * How nodes are keyed and how their stable ids are handled is selected by {@link Mode}:
+ * {@link Mode#STABLE_ID} (the default) keys nodes by stable id and preserves every stable
+ * id, {@link Mode#NODE_DATA} keys nodes by their translated record data and keeps stable ids
+ * at their self-reference default unless already set, {@link Mode#DEEP_DEDUP} keys nodes by
+ * their translated record data alone (ignoring stable ids) and {@link Mode#NO_DEDUP} disables
+ * content deduplication altogether, relocating every record faithfully.
+ * <p>
+ * This first increment keeps the full alias table in memory (simplest, exact) so
+ * that the transcoder itself can be validated; a reference-counted, memory-bounded
+ * alias table is a follow-up. Only whole-graph (full) repacking is supported.
+ * <p>
+ * Byte payloads (value/string/block records) are copied verbatim without decoding
+ * them into Java objects. Structural records (maps and lists) are rebuilt through
+ * {@link DefaultSegmentWriter}, and binaries are relinked through
+ * {@link DefaultSegmentWriter#writeBlob}, so bulk segments are never rewritten.
+ */
+public class RecordRepacker {
+
+ /**
+ * Strategy for handling a node's stable id and for deduplicating nodes.
+ */
+ public enum Mode {
+ /**
+ * Preserve every stable id by materialising it as an explicit block, and
+ * deduplicate nodes by their stable id. A stable-id hit prunes the (identical)
+ * subtree from a second walk, mirroring the semantic compactors' node cache.
+ */
+ STABLE_ID,
+
+ /**
+ * Keep a node's stable id at its self-reference default unless the source node
+ * already has an explicit one, and deduplicate nodes by their translated record
+ * data (the template, child and property references). Because references are
+ * record ids, equal translated references mean byte-identical node records, so no
+ * subtree hashing is needed. This can collapse content-equal nodes that carry
+ * distinct <em>default</em> stable ids, at the cost of not preserving those default
+ * stable ids and of walking a duplicate node's subtree before deduplicating it.
+ */
+ NODE_DATA,
+
+ /**
+ * Like {@link #NODE_DATA}, but the node dedup key ignores stable ids entirely and every
+ * stable id is dropped to its self-reference default. Because deduplication runs
+ * bottom-up in post-order, this collapses whole content-equal subtrees —
+ * including nodes that carry <em>distinct explicit</em> stable ids, which {@link #NODE_DATA}
+ * keeps apart. On a store whose nodes are all at their self-reference default this
+ * coincides with {@link #NODE_DATA}; it only finds more when explicit stable ids are present.
+ */
+ DEEP_DEDUP,
+
+ /**
+ * Faithfully relocate every record without discovering any new sharing: keep only
+ * the physical sharing already present in the source (via the alias table) and
+ * disable the content-addressed {@link #dedupCache}. A node's stable id is kept
+ * exactly as the source encodes it (a self-reference default stays a self-reference
+ * default). This is the un-deduplicated baseline the other modes improve on.
+ */
+ NO_DEDUP
+ }
+
+ @NotNull
+ private final SegmentStore store;
+
+ @NotNull
+ private final SegmentReader reader;
+
+ @NotNull
+ private final SegmentIdProvider idProvider;
+
+ @NotNull
+ private final GCGeneration targetGeneration;
+
+ @NotNull
+ private final Mode mode;
+
+ /** Number of worker threads to repack with; {@code 1} means fully single-threaded. */
+ private final int concurrency;
+
+ @Nullable
+ private final BlobStore blobStore;
+
+ private final int binariesInlineThreshold;
+
+ /**
+ * Handles all writes to the target generation. Both the raw record emits of this class (see
+ * {@link #emit}) and the {@link #structuralWriter} route through it, so all repacked records are
+ * co-packed into the same segment stream. A plain {@link SegmentBufferWriter} single-threaded;
+ * when repacking in parallel either a thread-specific {@link SegmentBufferWriterPool} (a writer
+ * per worker thread, the default) or a single {@link SegmentBufferWriter} behind a synchronized
+ * wrapper (all workers serialise on one writer) depending on {@link #singleWriter}. Rebuilt by
+ * {@link #withSingleWriter(boolean)}, hence not {@code final}.
+ */
+ @NotNull
+ private WriteOperationHandler writeHandler;
+
+ /**
+ * Used for rebuilding maps and lists and for relinking binaries. Shares
+ * {@link #writeHandler} as its write operation handler, so it is thread-safe when the handler is.
+ * Rebuilt together with {@link #writeHandler}, hence not {@code final}.
+ */
+ @NotNull
+ private DefaultSegmentWriter structuralWriter;
+
+ /**
+ * Old {@link RecordId} to new {@link RecordId}, kept off the Java heap in a sharded, memory
+ * mapped table (see {@link ShardedRecordIdMap}). Deduplicates records that were already
+ * physically shared in the source.
+ */
+ @NotNull
+ private final ShardedRecordIdMap alias;
+
+ /**
+ * Canonical record content (a {@value #KEY_HASH_WIDTH}-byte hash built by {@link HashKey})
+ * to the {@link RecordId} of the (single) emitted copy. Enables record-level deduplication of
+ * records that are content-identical after reference translation but had distinct source record
+ * ids. Keys are tagged by record type (see the {@code T_*} constants); nodes are keyed by their
+ * stable id or translated data (per {@link Mode}), everything else by its translated content.
+ * {@code null} when the mode disables deduplication.
+ * <p>
+ * A bounded recency window by default (auto-sized to {@link #AUTO_WINDOW_FRACTION} of
+ * {@link #expectedRecords}): off-heap in a bounded memory-mapped file ({@link MmapWindowDedupCache})
+ * or, per {@link #withDedupCacheImpl}, in-heap ({@link WindowDedupCache}); either deduplicates only
+ * records whose content recurs while still in the window and needs no whole-store spill. The exact,
+ * unbounded, off-heap {@link ShardedRecordIdMap} is used when the window resolves to {@code 0} (see
+ * {@link #withDedupWindow}). Built lazily in {@link #repack} (so the window vs exact choice can be
+ * made after construction).
+ */
+ @Nullable
+ private DedupCache dedupCache;
+
+ /** Number of shards for the {@link #alias} / {@link #dedupCache} tables (matches concurrency). */
+ private final int shardCount;
+
+ /** Estimated record count, used to pre-size the exact tables; see the constructor. */
+ private final int expectedRecords;
+
+ /**
+ * Selects the {@link #dedupCache} window size. {@code -1} (the default) auto-sizes a bounded recency
+ * window (off-heap {@link MmapWindowDedupCache} by default) to {@link #AUTO_WINDOW_FRACTION} of
+ * {@link #expectedRecords} (recovering nearly all of the exact deduplication at a small, bounded
+ * footprint and no whole-store spill); {@code > 0} uses a window of exactly that many entries;
+ * {@code 0} uses the exact, unbounded, off-heap {@link ShardedRecordIdMap}. Set via {@link #withDedupWindow}.
+ */
+ private int dedupWindow = -1;
+
+ /**
+ * Fraction of {@link #expectedRecords} used to auto-size the default recency window: a window of
+ * ~10% of the input's record count recovers the exact deduplication within a few percent (the
+ * evicted entries are one-off records that never recur), at a small fraction of the exact cache's
+ * footprint. Below a known record count ({@code expectedRecords <= 0}) the exact cache is used.
+ */
+ private static final int AUTO_WINDOW_FRACTION = 10;
+
+ /** Which bounded-dedup implementation to use for a windowed (non-exact) {@link #dedupCache}. */
+ public enum DedupCacheImpl {
+ /** Off-heap (memory-mapped) bounded recency window that forgets evicted entries ({@link MmapWindowDedupCache}). */
+ MMAP_WINDOW,
+ /** Hand-rolled in-heap sharded access-order LRU that forgets evicted entries ({@link WindowDedupCache}). */
+ HEAP_WINDOW
+ }
+
+ /**
+ * Bounded-dedup implementation; only used when {@link #resolveDedupWindow} resolves to a window.
+ * The default {@link DedupCacheImpl#MMAP_WINDOW} keeps the window off the Java heap (in a bounded
+ * memory-mapped file), which is lower-heap and faster than the in-heap {@link DedupCacheImpl#HEAP_WINDOW}
+ * at a marginally larger output.
+ */
+ private DedupCacheImpl dedupCacheImpl = DedupCacheImpl.MMAP_WINDOW;
+
+ /** Width in bytes of a {@link #dedupCache} content-hash key. */
+ private static final int KEY_HASH_WIDTH = 16;
+
+ /** Target number of parallel subtree roots to explore for, per worker thread. */
+ private static final int SPLIT_POINTS_PER_THREAD = 32;
+
+ /** Upper bound on the number of parallel subtree roots (bounds task count and exploration). */
+ private static final int MAX_SPLIT_POINTS = 16384;
+
+ /**
+ * Maximum number of direct children a node may have and still be expanded into per-child split
+ * roots during {@link #findSplitPoints}. A node wider than this is kept whole as a single split
+ * root, so its entire child set is repacked by one worker into a contiguous run of segments
+ * rather than being scattered across the workers' separate (thread-specific) segment streams.
+ * Scattering fragments read locality: listing such a node then has to fetch a segment from every
+ * worker instead of one dense run. Keeping wide nodes whole preserves the serial layout's read
+ * density at the cost of repacking those (typically leaf-heavy) subtrees single-threaded.
+ */
+ private static final int SPLIT_MAX_FANOUT = 1024;
+
+ /** Per-thread reused builder for {@link #dedupCache} keys; see {@link #hash(int)}. */
+ @NotNull
+ private final ThreadLocal<HashKey> hashKey = ThreadLocal.withInitial(HashKey::new);
+
+ private static final int T_NODE = 0;
+ private static final int T_TEMPLATE = 1;
+ private static final int T_VALUE = 2;
+ private static final int T_VALUE_LONG = 3;
+ private static final int T_BLOCK = 4;
+ private static final int T_LIST = 5;
+ private static final int T_ARRAY = 6;
+ private static final int T_EMPTY_ARRAY = 7;
+ private static final int T_MAP = 8;
+
+ /**
+ * Whether to content-deduplicate child-node maps. On by default, so that whenever content
+ * deduplication is enabled at all (any mode except {@link Mode#NO_DEDUP}) every structural
+ * record type is deduplicated consistently. A child map whose translated
+ * {@code name -> child id} entries match an already emitted map is not written again.
+ * Because the entries are the (canonical) translated child ids, this also lets the
+ * multi-child nodes referencing content-equal maps deduplicate. Can be turned off (e.g. to
+ * measure its effect) via {@link #withMapDeduplication(boolean)}.
+ */
+ private boolean deduplicateMaps = true;
+
+ /**
+ * Target number of parallel subtree roots for {@link #findSplitPoints} (parallel repacking only).
+ * {@code 0} (the default) derives the target from the worker count
+ * ({@link #concurrency} × {@link #SPLIT_POINTS_PER_THREAD}), coupling the split depth to the
+ * thread count. A positive value decouples the two: the frontier is explored to this many roots
+ * regardless of {@link #concurrency}, so a deeper split (which pushes shared subtrees into the
+ * exactly-deduplicated single-threaded spine, reducing cross-worker re-emission) can be measured
+ * independently of the degree of parallelism. Set via {@link #withSplitTarget(int)}.
+ */
+ private int splitTarget = 0;
+
+ /**
+ * Whether concurrent content deduplication is <em>exact</em>. When {@code true} (the default),
+ * the dedup-cache check and the record emit are performed as one atomic operation
+ * ({@link ShardedRecordIdMap#computeIfAbsent}), so two workers reaching the same content never
+ * both emit: the loser blocks, observes the winner's id and skips its write. When {@code false},
+ * the check and emit race ({@code get} then {@code emit} then {@code putIfAbsent}) and a losing
+ * worker's already-emitted record is left as unreferenced garbage - writes are not serialised
+ * on the shard lock, at the cost of duplicated records under contention. Single-threaded the two
+ * are equivalent. Applies only to the deduplicating modes; {@link Mode#NO_DEDUP} has no dedup
+ * cache and so is always inexact under concurrency. Set via {@link #withExactDedup(boolean)}.
+ */
+ private boolean exactDedup = true;
+
+ /**
+ * When repacking in parallel ({@link #concurrency} > 1), whether all worker threads serialise
+ * their writes through a single {@link SegmentBufferWriter} (behind a synchronized wrapper) instead
+ * of each writing to its own thread-specific writer. A single writer packs every record into one
+ * segment stream (marginally denser, no partially-filled trailing segment per worker) but
+ * serialises the physical write; the default (per-thread writers) writes in parallel. Only affects
+ * the physical write path, not which records are emitted, so output is content-identical either
+ * way. Set via {@link #withSingleWriter(boolean)}.
+ */
+ private boolean singleWriter = false;
+
+ /**
+ * Ordered list of subtree roots to repack first, each as a separate stage, before the rest of the
+ * graph. Used to control emission order: repacking {@code /oak:index} first means its records - and
+ * any records it shares with the rest of the repository - get their canonical copies emitted up
+ * front, keeping the (typically wide) index subtrees packed into a contiguous run of segments
+ * rather than scattered across the parallel workers' separate streams. Listing a wide index then
+ * reads densely instead of fetching a segment per worker. Each stage runs at the configured
+ * {@link #concurrency} and completes (a barrier) before the next begins; the final stage is always
+ * the overall root passed to {@link #repack}, which alias-hits everything the earlier stages
+ * already emitted. Stages may be nested (e.g. {@code /root/oak:index}, then {@code /root}): an inner
+ * stage's records are simply alias hits when the enclosing stage runs. Empty (the default) disables
+ * staging (a single stage over the whole graph). Set via {@link #withStages} / {@link #withIndexFirst}.
+ */
+ @NotNull
+ private List<RecordId> stageRoots = List.of();
+
+ public RecordRepacker(
+ @NotNull SegmentStore store,
+ @NotNull SegmentReader reader,
+ @NotNull SegmentIdProvider idProvider,
+ @Nullable BlobStore blobStore,
+ int binariesInlineThreshold,
+ @NotNull GCGeneration targetGeneration) throws IOException {
+ this(store, reader, idProvider, blobStore, binariesInlineThreshold, targetGeneration,
+ Mode.STABLE_ID);
+ }
+
+ public RecordRepacker(
+ @NotNull SegmentStore store,
+ @NotNull SegmentReader reader,
+ @NotNull SegmentIdProvider idProvider,
+ @Nullable BlobStore blobStore,
+ int binariesInlineThreshold,
+ @NotNull GCGeneration targetGeneration,
+ @NotNull Mode mode) throws IOException {
+ this(store, reader, idProvider, blobStore, binariesInlineThreshold, targetGeneration, mode, 0);
+ }
+
+ /**
+ * @param expectedRecords an estimate of the number of records that will be repacked, used to
+ * pre-size the {@link #alias} and {@link #dedupCache} tables so they do not grow and rehash
+ * while repacking a large store. {@code 0} (or a negative value) falls back to a small
+ * default. Over-estimating is cheap - the tables are sparse memory-mapped files - and the
+ * value is capped at the tables' maximum size.
+ */
+ public RecordRepacker(
+ @NotNull SegmentStore store,
+ @NotNull SegmentReader reader,
+ @NotNull SegmentIdProvider idProvider,
+ @Nullable BlobStore blobStore,
+ int binariesInlineThreshold,
+ @NotNull GCGeneration targetGeneration,
+ @NotNull Mode mode,
+ int expectedRecords) throws IOException {
+ this(store, reader, idProvider, blobStore, binariesInlineThreshold, targetGeneration, mode,
+ expectedRecords, 1);
+ }
+
+ /**
+ * @param concurrency number of worker threads to repack with. {@code 1} repacks single-threaded;
+ * a higher value partitions the record graph into subtrees repacked in parallel (only the
+ * deduplicating modes are safe to parallelize). The alias and dedup tables are sharded
+ * accordingly so concurrent workers rarely contend.
+ */
+ public RecordRepacker(
+ @NotNull SegmentStore store,
+ @NotNull SegmentReader reader,
+ @NotNull SegmentIdProvider idProvider,
+ @Nullable BlobStore blobStore,
+ int binariesInlineThreshold,
+ @NotNull GCGeneration targetGeneration,
+ @NotNull Mode mode,
+ int expectedRecords,
+ int concurrency) throws IOException {
+ this.store = requireNonNull(store);
+ this.reader = requireNonNull(reader);
+ this.idProvider = requireNonNull(idProvider);
+ this.targetGeneration = requireNonNull(targetGeneration);
+ this.mode = requireNonNull(mode);
+ this.concurrency = Math.max(1, concurrency);
+ this.blobStore = blobStore;
+ this.binariesInlineThreshold = binariesInlineThreshold;
+ this.writeHandler = newWriteHandler();
+ this.structuralWriter = newStructuralWriter();
+ this.shardCount = this.concurrency <= 1 ? 1 : Math.min(128, this.concurrency * 4);
+ this.expectedRecords = expectedRecords;
+ this.alias = new ShardedRecordIdMap(idProvider, RecordId.SERIALIZED_RECORD_ID_BYTES,
+ expectedRecords, shardCount, "alias");
+ // dedupCache is built lazily in repack(), so withDedupWindow() can pick window vs exact.
+ }
+
+ /**
+ * Override the content-deduplication cache sizing. By default ({@link #dedupWindow} {@code == -1}) a
+ * bounded in-heap recency window auto-sized to {@link #AUTO_WINDOW_FRACTION} of {@link #expectedRecords}
+ * is used. Pass {@code maxEntries > 0} for a window of exactly that many entries, or {@code 0} for the
+ * exact, unbounded, off-heap-spilled cache. Only records whose content recurs while still in the window
+ * are deduplicated; the alias (physical-sharing memo) stays exact regardless. Package-private prototype
+ * knob; takes effect for the deduplicating modes.
+ *
+ * @return {@code this}, for chaining.
+ */
+ @NotNull
+ RecordRepacker withDedupWindow(int maxEntries) {
+ this.dedupWindow = maxEntries;
+ return this;
+ }
+
+ /**
+ * Select the bounded-dedup implementation used when a window is in effect (see {@link DedupCacheImpl}).
+ * The default {@link DedupCacheImpl#MMAP_WINDOW} keeps the window off-heap in a bounded memory-mapped
+ * file; {@link DedupCacheImpl#HEAP_WINDOW} keeps it in the Java heap. No effect on the exact
+ * ({@code window <= 0}) path. Package-private prototype knob.
+ *
+ * @return {@code this}, for chaining.
+ */
+ @NotNull
+ RecordRepacker withDedupCacheImpl(@NotNull DedupCacheImpl impl) {
+ this.dedupCacheImpl = impl;
+ return this;
+ }
+
+ /**
+ * Enable or disable content deduplication of child-node maps (see {@link #deduplicateMaps}).
+ * Package-private prototype knob used to measure the effect of map deduplication.
+ *
+ * @return {@code this}, for chaining.
+ */
+ @NotNull
+ RecordRepacker withMapDeduplication(boolean enabled) {
+ this.deduplicateMaps = enabled;
+ return this;
+ }
+
+ /**
+ * Set the target number of parallel subtree roots, decoupling the split depth from the worker
+ * count (see {@link #splitTarget}). {@code 0} restores the default (derive from concurrency).
+ * Package-private prototype knob used to measure the split depth in isolation.
+ *
+ * @return {@code this}, for chaining.
+ */
+ @NotNull
+ RecordRepacker withSplitTarget(int target) {
+ this.splitTarget = Math.max(0, target);
+ return this;
+ }
+
+ /**
+ * Enable or disable exact concurrent content deduplication (see {@link #exactDedup}).
+ * Package-private prototype knob used to measure the cost of serialising record emits on the
+ * shard lock (exact) against the un-serialised racing emit (inexact, which leaves loser garbage).
+ *
+ * @return {@code this}, for chaining.
+ */
+ @NotNull
+ RecordRepacker withExactDedup(boolean enabled) {
+ this.exactDedup = enabled;
+ return this;
+ }
+
+ /**
+ * Switch the parallel write path between per-thread writers and a single serialised writer (see
+ * {@link #singleWriter}) and rebuild {@link #writeHandler} / {@link #structuralWriter} accordingly.
+ * Package-private prototype knob used to measure whether serialising all writes through one writer
+ * (rather than one per worker thread) changes throughput - i.e. whether writing is a bottleneck.
+ * Has no effect when repacking single-threaded. Must be called before {@link #repack(RecordId)}.
+ *
+ * @return {@code this}, for chaining.
+ */
+ @NotNull
+ RecordRepacker withSingleWriter(boolean enabled) {
+ this.singleWriter = enabled;
+ this.writeHandler = newWriteHandler();
+ this.structuralWriter = newStructuralWriter();
+ return this;
+ }
+
+ /**
+ * Repack {@code subtreeRoot} first, as a separate stage, before repacking the rest of the graph
+ * (see {@link #stageRoots}). Intended for the {@code /oak:index} node, whose wide child sets read
+ * best when kept together in one contiguous run rather than scattered across parallel workers.
+ * Both stages run at the configured {@link #concurrency}. {@code null} (the default) disables the
+ * two-stage ordering. Convenience for the single-stage case; see {@link #withStages}.
+ * Package-private prototype knob.
+ *
+ * @return {@code this}, for chaining.
+ */
+ @NotNull
+ RecordRepacker withIndexFirst(@Nullable RecordId subtreeRoot) {
+ return withStages(subtreeRoot == null ? List.of() : List.of(subtreeRoot));
+ }
+
+ /**
+ * Repack each of {@code stages} first, in order, each as a separate parallel stage separated by a
+ * barrier, before repacking the rest of the graph (see {@link #stageRoots}). The overall root
+ * passed to {@link #repack} is always repacked last. Stages may be nested and given innermost-first
+ * (e.g. {@code [/root/oak:index, /root]}): an inner stage's records are alias hits when the
+ * enclosing stage runs, so each stage only emits what the earlier stages did not. An empty list
+ * (the default) disables staging. Package-private prototype knob.
+ *
+ * @return {@code this}, for chaining.
+ */
+ @NotNull
+ RecordRepacker withStages(@NotNull List<RecordId> stages) {
+ this.stageRoots = List.copyOf(stages);
+ return this;
+ }
+
+ @NotNull
+ private WriteOperationHandler newWriteHandler() {
+ if (concurrency <= 1) {
+ return new SegmentBufferWriter(idProvider, "repack", targetGeneration);
+ }
+ if (singleWriter) {
+ return new SynchronizedWriteHandler(
+ new SegmentBufferWriter(idProvider, "repack", targetGeneration));
+ }
+ return SegmentBufferWriterPool.factory(idProvider, "repack", () -> targetGeneration)
+ .newPool(SegmentBufferWriterPool.PoolType.THREAD_SPECIFIC);
+ }
+
+ @NotNull
+ private DefaultSegmentWriter newStructuralWriter() {
+ return new DefaultSegmentWriter(store, reader, idProvider, blobStore,
+ WriterCacheManager.Empty.INSTANCE, writeHandler, binariesInlineThreshold);
+ }
+
+ /**
+ * Repack the whole record graph reachable from {@code rootId} into the target
+ * generation and return the {@link RecordId} of the repacked root node.
+ */
+ @NotNull
+ public RecordId repack(@NotNull RecordId rootId) throws IOException {
+ this.dedupCache = newDedupCache();
+ try {
+ RecordId newRoot = concurrency <= 1 ? repackSerial(rootId) : repackParallel(rootId);
+ writeHandler.flush(store);
+ return newRoot;
+ } finally {
+ alias.close();
+ if (dedupCache != null) {
+ dedupCache.close();
+ }
+ }
+ }
+
+ /**
+ * Build the content-dedup cache for a repack run: {@code null} when the mode disables dedup, the exact
+ * off-heap {@link ShardedRecordIdMap} when the window resolves to {@code 0} (full-dedup), otherwise the
+ * bounded implementation selected by {@link #withDedupCacheImpl} (default {@link DedupCacheImpl#MMAP_WINDOW}).
+ */
+ @Nullable
+ private DedupCache newDedupCache() throws IOException {
+ if (!dedupEnabled()) {
+ return null;
+ }
+ int window = resolveDedupWindow();
+ if (window <= 0) {
+ return new ShardedRecordIdMap(idProvider, KEY_HASH_WIDTH, expectedRecords, shardCount, "dedup");
+ }
+ switch (dedupCacheImpl) {
+ case HEAP_WINDOW:
+ return new WindowDedupCache(window, shardCount);
+ case MMAP_WINDOW:
+ default:
+ return new MmapWindowDedupCache(idProvider, KEY_HASH_WIDTH, window, shardCount);
+ }
+ }
+
+ /**
+ * Resolve the effective recency-window size: an explicit {@link #dedupWindow} ({@code >= 0}) as set,
+ * or - by default ({@code -1}) - {@link #AUTO_WINDOW_FRACTION} of {@link #expectedRecords}. Falls back
+ * to {@code 0} (the exact cache) when the record count is unknown or too small to size a window.
+ */
+ private int resolveDedupWindow() {
+ if (dedupWindow >= 0) {
+ return dedupWindow;
+ }
+ return expectedRecords > 0 ? expectedRecords / AUTO_WINDOW_FRACTION : 0;
+ }
+
+ /**
+ * Single-threaded repack. Each {@link #stageRoots} subtree is repacked first, in order, so its
+ * records are emitted as one contiguous run up front; then the whole graph, which alias-hits them.
+ */
+ @NotNull
+ private RecordId repackSerial(@NotNull RecordId rootId) throws IOException {
+ for (RecordId stageRoot : stageRoots) {
+ copyNode(stageRoot);
+ }
+ return copyNode(rootId);
+ }
+
+ /**
+ * Repack the graph reachable from {@code rootId} using {@link #concurrency} worker threads. When
+ * {@link #stageRoots} is non-empty this runs in stages, each parallel and separated by a barrier:
+ * each stage subtree is repacked to completion in order, then the whole graph (whose records under
+ * the earlier stages are now alias hits), so each stage's (typically wide) child sets are emitted
+ * as a contiguous run before the next stage races. Otherwise a single stage repacks the whole graph.
+ */
+ @NotNull
+ private RecordId repackParallel(@NotNull RecordId rootId) throws IOException {
+ for (RecordId stageRoot : stageRoots) {
+ // Repack this stage subtree first, in parallel, to completion (a barrier).
+ parallelCopy(stageRoot);
+ }
+ // Final stage: repack the whole graph in parallel; the earlier stages are now aliased.
+ return parallelCopy(rootId);
+ }
+
+ /**
+ * Repack the graph reachable from {@code subtreeRoot} in parallel. The graph is explored top-down
+ * to a frontier of subtree roots, each repacked concurrently ({@link #copyNode} is thread-safe:
+ * the alias and dedup tables are sharded, the writers thread-specific). The frontier is chosen
+ * locality-aware (see {@link #findSplitPoints}): a wide node's whole child set stays within one
+ * worker, so its children are repacked into a contiguous segment run and reads of that node stay
+ * dense. A final single-threaded walk from {@code subtreeRoot} emits the spine above the frontier;
+ * each frontier subtree is already in the alias table, so descending into it is a hit that returns
+ * the concurrently repacked result.
+ */
+ @NotNull
+ private RecordId parallelCopy(@NotNull RecordId subtreeRoot) throws IOException {
+ int target = splitTarget > 0 ? splitTarget : concurrency * SPLIT_POINTS_PER_THREAD;
+ List<RecordId> splitRoots = findSplitPoints(subtreeRoot, target);
+ ExecutorService executor = Executors.newFixedThreadPool(concurrency);
+ try {
+ List<Future<RecordId>> futures = new ArrayList<>(splitRoots.size());
+ for (RecordId splitRoot : splitRoots) {
+ futures.add(executor.submit(() -> copyNode(splitRoot)));
+ }
+ for (Future<RecordId> future : futures) {
+ future.get();
+ }
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ throw new IOException("interrupted while repacking in parallel", e);
+ } catch (ExecutionException e) {
+ Throwable cause = e.getCause();
+ throw cause instanceof IOException ? (IOException) cause
+ : new IOException("parallel repack failed", cause);
+ } finally {
+ executor.shutdown();
+ }
+ // Spine: the frontier subtrees are now aliased, so this only walks the nodes above them.
+ return copyNode(subtreeRoot);
+ }
+
+ /**
+ * Explore the node graph below {@code root} breadth-first (through child maps) until at least
+ * {@code target} subtree roots are found or the frontier can no longer be expanded, and return
+ * that (distinct) frontier. Growth is capped at {@value #MAX_SPLIT_POINTS} so a very wide node
+ * does not produce an unbounded number of tasks.
+ * <p>
+ * The split is locality-aware: a node with more than {@link #SPLIT_MAX_FANOUT} children is
+ * <em>not</em> expanded but kept whole as a single split root (see {@link #SPLIT_MAX_FANOUT}), so
+ * one worker repacks its entire child set contiguously and reads of that node stay dense. Once a
+ * node is kept whole (or is a leaf) it is not re-examined on later passes.
+ */
+ @NotNull
+ private List<RecordId> findSplitPoints(@NotNull RecordId root, int target) {
+ List<RecordId> frontier = new ArrayList<>();
+ frontier.add(root);
+ // Split roots that must not be expanded further: leaves, kept-as-is records and nodes kept
+ // whole because they are wider than SPLIT_MAX_FANOUT. Tracked so they are not re-materialised.
+ Set<RecordId> closed = new HashSet<>();
+ while (frontier.size() < target) {
+ List<RecordId> next = new ArrayList<>();
+ boolean progressed = false;
+ for (RecordId nodeId : frontier) {
+ if (closed.contains(nodeId)) {
+ next.add(nodeId);
+ continue;
+ }
+ List<RecordId> children = next.size() < MAX_SPLIT_POINTS ? childNodeIds(nodeId) : List.of();
+ if (children.isEmpty() || children.size() > SPLIT_MAX_FANOUT) {
+ // Leaf/kept-as-is, or too wide to scatter: keep this node as a single split root.
+ next.add(nodeId);
+ closed.add(nodeId);
+ } else {
+ next.addAll(children);
+ progressed = true;
+ }
+ }
+ if (!progressed) {
+ break;
+ }
+ frontier = next;
+ if (frontier.size() >= MAX_SPLIT_POINTS) {
+ break;
+ }
+ }
+ return new ArrayList<>(new LinkedHashSet<>(frontier));
+ }
+
+ /** The record ids of {@code nodeId}'s child nodes, or empty if it is a leaf or kept as-is. */
+ @NotNull
+ private List<RecordId> childNodeIds(@NotNull RecordId nodeId) {
+ if (keepAsIs(nodeId)) {
+ return List.of();
+ }
+ List<RecordId> children = new ArrayList<>();
+ for (ChildNodeEntry entry : reader.readNode(nodeId).getChildNodeEntries()) {
+ NodeState child = entry.getNodeState();
+ if (child instanceof SegmentNodeState) {
+ children.add(((SegmentNodeState) child).getRecordId());
+ }
+ }
+ return children;
+ }
+
+ /**
+ * @return {@code true} if the record must be kept as-is (never repacked):
+ * records in bulk segments (binaries) and records that are already in the
+ * target generation.
+ */
+ private boolean keepAsIs(@NotNull RecordId id) {
+ SegmentId segmentId = id.getSegmentId();
+ if (!segmentId.isDataSegmentId()) {
+ return true;
+ }
+ return targetGeneration.equals(segmentId.getGcGeneration());
+ }
+
+ private RecordId emit(@NotNull RecordWriters.RecordWriter recordWriter) throws IOException {
+ return writeHandler.execute(targetGeneration, writer -> recordWriter.write(writer, store));
+ }
+
+ /**
+ * Emit the record produced by {@code recordWriter}, unless a content-identical
+ * record (same {@code key}) has already been emitted, in which case its
+ * {@link RecordId} is reused.
+ */
+ private RecordId dedup(@NotNull byte[] key, @NotNull RecordWriters.RecordWriter recordWriter)
+ throws IOException {
+ if (!dedupEnabled()) {
+ return emit(recordWriter);
+ }
+ if (exactDedup) {
+ return dedupCache.computeIfAbsent(key, () -> emit(recordWriter));
+ }
+ RecordId existing = dedupCache.get(key);
+ if (existing != null) {
+ return existing;
+ }
+ RecordId newId = emit(recordWriter);
+ return dedupPut(key, newId);
+ }
+
+ /**
+ * Record {@code newId} as the emitted copy for content {@code key} and return it, unless a
+ * competing worker already emitted a copy of the same content - in which case that (winning)
+ * {@link RecordId} is returned and {@code newId}'s record is left as unreferenced garbage.
+ * Single-threaded this always returns {@code newId}.
+ */
+ private RecordId dedupPut(@NotNull byte[] key, @NotNull RecordId newId) throws IOException {
+ RecordId won = dedupCache.putIfAbsent(key, newId);
+ return won != null ? won : newId;
+ }
+
+ /**
+ * Emit a node record deduplicated by {@code nodeKey}: reuse an already-emitted content-equal
+ * node if present, otherwise emit via {@code emitNode} and record it, then map
+ * {@code sourceId -> result} in the alias table. Exact under concurrency (see {@link #exactDedup}):
+ * the check and the emit are atomic, so two workers with the same node content never both emit.
+ * The caller must already have handled any pre-recursion dedup short-circuit (a stable-id hit that
+ * prunes the subtree walk); this only guards the emit itself.
+ */
+ private RecordId dedupNode(@NotNull RecordId sourceId, @NotNull byte[] nodeKey,
+ @NotNull ShardedRecordIdMap.RecordIdSupplier emitNode) throws IOException {
+ RecordId newId;
+ if (exactDedup) {
+ newId = dedupCache.computeIfAbsent(nodeKey, emitNode);
+ } else {
+ RecordId existing = dedupCache.get(nodeKey);
+ newId = existing != null ? existing : dedupPut(nodeKey, emitNode.get());
+ }
+ alias.putIfAbsent(sourceId, newId);
+ return newId;
+ }
+
+ /**
+ * Write a list record for the given (already translated) entries, deduplicating
+ * lists that have an identical entry sequence.
+ */
+ private RecordId writeListDeduped(@NotNull List<RecordId> entries) throws IOException {
+ if (!dedupEnabled()) {
+ return structuralWriter.writeList(entries);
+ }
+ byte[] key = hash(T_LIST).putIds(entries).digest();
+ if (exactDedup) {
+ return dedupCache.computeIfAbsent(key, () -> structuralWriter.writeList(entries));
+ }
+ RecordId existing = dedupCache.get(key);
+ if (existing != null) {
+ return existing;
+ }
+ RecordId newId = structuralWriter.writeList(entries);
+ return dedupPut(key, newId);
+ }
+
+ // -- per record type -----------------------------------------------------
+
+ @Nullable
+ private RecordId aliasGet(@NotNull RecordId id) {
+ return alias.get(id);
+ }
+
+ private void aliasPut(@NotNull RecordId id, @NotNull RecordId newId) throws IOException {
+ alias.putIfAbsent(id, newId);
+ }
+
+ private RecordId copyNode(@NotNull RecordId id) throws IOException {
+ if (keepAsIs(id)) {
+ return id;
+ }
+ RecordId cached = aliasGet(id);
+ if (cached != null) {
+ return cached;
+ }
+ return copyNodeCore(id);
+ }
+
+ private RecordId copyNodeCore(@NotNull RecordId id) throws IOException {
+ Segment segment = id.getSegment();
+ int recordNumber = id.getRecordNumber();
+ RecordId stableSlot = segment.readRecordId(recordNumber);
+ switch (mode) {
+ case STABLE_ID:
+ return copyNodeByStableId(id, segment, recordNumber, stableSlot);
+ case DEEP_DEDUP:
+ return copyNodeDeep(id, segment, recordNumber);
+ default: // NODE_DATA and NO_DEDUP share the self-reference-default stable id handling
+ return copyNodeByData(id, segment, recordNumber, stableSlot);
+ }
+ }
+
+ /** @return {@code true} unless the mode disables content deduplication. */
+ private boolean dedupEnabled() {
+ return mode != Mode.NO_DEDUP;
+ }
+
+ /**
+ * Deduplicate by stable id. A node's stable id uniquely identifies its content, so
+ * two node records that share a stable id are the same logical node and can share a
+ * single repacked record, pruning the (identical) subtree from a second walk. This
+ * mirrors the stable-id keyed node cache of the semantic compactors, and preserves
+ * every stable id by materialising it as an explicit block.
+ */
+ private RecordId copyNodeByStableId(@NotNull RecordId id, @NotNull Segment segment,
+ int recordNumber, @NotNull RecordId stableSlot) throws IOException {
+ byte[] stableId = readStableId(id, stableSlot);
+ byte[] nodeKey = hash(T_NODE).putBytes(stableId).digest();
+ RecordId deduped = dedupCache.get(nodeKey);
+ if (deduped != null) {
+ alias.putIfAbsent(id, deduped);
+ return deduped;
+ }
+
+ RecordId newStableId = copyStableId(stableId);
+ List<RecordId> ids = translateNodeBody(segment, recordNumber);
+
+ return dedupNode(id, nodeKey, () -> emit(RecordWriters.newNodeStateWriter(newStableId, ids)));
+ }
+
+ /**
+ * Handle a node with the self-reference-default stable id policy shared by
+ * {@link Mode#NODE_DATA} and {@link Mode#NO_DEDUP}: the stable id is left at its
+ * self-reference default (written from the new record id) unless the source node already
+ * carries an explicit one, which is preserved. When deduplication is enabled
+ * ({@link Mode#NODE_DATA}) the node is keyed by its translated record data and its
+ * stable-id treatment, so two nodes with the same translated references collapse to one;
+ * when it is disabled ({@link Mode#NO_DEDUP}) the node is emitted faithfully.
+ */
+ private RecordId copyNodeByData(@NotNull RecordId id, @NotNull Segment segment,
+ int recordNumber, @NotNull RecordId stableSlot) throws IOException {
+ boolean explicit = !stableSlot.equals(id);
+ RecordId newStableId = explicit ? copyStableId(readStableId(id, stableSlot)) : null;
+ List<RecordId> ids = translateNodeBody(segment, recordNumber);
+
+ if (dedupEnabled()) {
+ HashKey hk = hash(T_NODE);
+ if (explicit) {
+ hk.putByte(1).putId(newStableId);
+ } else {
+ hk.putByte(0);
+ }
+ byte[] nodeKey = hk.putIds(ids).digest();
+ return dedupNode(id, nodeKey, () -> emit(RecordWriters.newNodeStateWriter(newStableId, ids)));
+ }
+
+ RecordId newId = emit(RecordWriters.newNodeStateWriter(newStableId, ids));
+ aliasPut(id, newId);
+ return newId;
+ }
+
+ /**
+ * Deduplicate by the node's translated record data alone, ignoring stable ids, and drop
+ * the stable id to its self-reference default. Because children are deduplicated first
+ * (post-order), equal translated references mean the whole subtree is content-equal, so
+ * this collapses content-equal nodes even when they carried distinct explicit stable ids
+ * (which {@link #copyNodeByData} keeps apart).
+ */
+ private RecordId copyNodeDeep(@NotNull RecordId id, @NotNull Segment segment, int recordNumber)
+ throws IOException {
+ List<RecordId> ids = translateNodeBody(segment, recordNumber);
+
+ byte[] nodeKey = hash(T_NODE).putIds(ids).digest();
+ return dedupNode(id, nodeKey, () -> emit(RecordWriters.newNodeStateWriter(null, ids)));
+ }
+
+ /**
+ * Translate a node's body records (template, children and property list) to the
+ * target generation and return their new ids in node-record order.
+ */
+ private List<RecordId> translateNodeBody(@NotNull Segment segment, int recordNumber)
+ throws IOException {
+ RecordId templateId = segment.readRecordId(recordNumber, 0, 1);
+ Template template = reader.readTemplate(templateId);
+
+ List<RecordId> ids = new ArrayList<>();
+ ids.add(copyTemplate(templateId));
+
+ int propertyListIndex = 2;
+ String childName = template.getChildName();
+ if (childName == Template.MANY_CHILD_NODES) {
+ ids.add(copyChildMap(segment.readRecordId(recordNumber, 0, 2)));
+ propertyListIndex = 3;
+ } else if (childName != Template.ZERO_CHILD_NODES) {
+ ids.add(copyNode(segment.readRecordId(recordNumber, 0, 2)));
+ propertyListIndex = 3;
+ }
+
+ PropertyTemplate[] propertyTemplates = template.getPropertyTemplates();
+ if (propertyTemplates.length > 0) {
+ RecordId propertyListId = segment.readRecordId(recordNumber, 0, propertyListIndex);
+ ids.add(copyPropertyList(propertyListId, propertyTemplates));
+ }
+ return ids;
+ }
+
+ /**
+ * Read the 20-byte stable id of a node. The first record id of a node points to
+ * its stable id; when that slot is equal to the node's own record id the stable
+ * id is not stored explicitly but is the serialised form of that record id.
+ */
+ private byte[] readStableId(@NotNull RecordId nodeId, @NotNull RecordId stableSlot) {
+ byte[] data = new byte[RecordId.SERIALIZED_RECORD_ID_BYTES];
+ if (stableSlot.equals(nodeId)) {
+ nodeId.getBytes().get(data);
+ } else {
+ stableSlot.getSegment().readBytes(stableSlot.getRecordNumber(), 0, data, 0, data.length);
+ }
+ return data;
+ }
+
+ /**
+ * Materialise the stable id as an explicit block record so that it survives the
+ * change of record id caused by repacking (otherwise it would silently change).
+ * Identical stable-id blocks are deduplicated.
+ */
+ private RecordId copyStableId(@NotNull byte[] stableId) throws IOException {
+ return dedup(hash(T_BLOCK).putBytes(stableId).digest(),
+ RecordWriters.newBlockWriter(stableId, 0, stableId.length));
+ }
+
+ private RecordId copyTemplate(@NotNull RecordId id) throws IOException {
+ RecordId cached = alias.get(id);
+ if (cached != null) {
+ return cached;
+ }
+
+ Segment segment = id.getSegment();
+ int recordNumber = id.getRecordNumber();
+
+ int head = segment.readInt(recordNumber, 0);
+ boolean hasPrimaryType = (head & (1 << 31)) != 0;
+ boolean hasMixinTypes = (head & (1 << 30)) != 0;
+ boolean zeroChildNodes = (head & (1 << 29)) != 0;
+ boolean manyChildNodes = (head & (1 << 28)) != 0;
+ int mixinCount = (head >> 18) & ((1 << 10) - 1);
+ int propertyCount = head & ((1 << 18) - 1);
+
+ int offset = 4;
+
+ List<RecordId> ids = new ArrayList<>();
+
+ RecordId primaryId = null;
+ if (hasPrimaryType) {
+ primaryId = copyString(segment.readRecordId(recordNumber, offset));
+ ids.add(primaryId);
+ offset += Segment.RECORD_ID_BYTES;
+ }
+
+ List<RecordId> mixinIds = null;
+ if (hasMixinTypes) {
+ mixinIds = new ArrayList<>(mixinCount);
+ for (int i = 0; i < mixinCount; i++) {
+ RecordId mixinId = copyString(segment.readRecordId(recordNumber, offset));
+ mixinIds.add(mixinId);
+ offset += Segment.RECORD_ID_BYTES;
+ }
+ ids.addAll(mixinIds);
+ }
+
+ RecordId childNameId = null;
+ if (!zeroChildNodes && !manyChildNodes) {
+ childNameId = copyString(segment.readRecordId(recordNumber, offset));
+ ids.add(childNameId);
+ offset += Segment.RECORD_ID_BYTES;
+ }
+
+ RecordId propNamesId = null;
+ byte[] propertyTypes = new byte[propertyCount];
+ if (propertyCount > 0) {
+ RecordId propertyNamesListId = segment.readRecordId(recordNumber, offset);
+ offset += Segment.RECORD_ID_BYTES;
+ propNamesId = copyStringList(propertyNamesListId, propertyCount);
+ ids.add(propNamesId);
+ for (int i = 0; i < propertyCount; i++) {
+ propertyTypes[i] = segment.readByte(recordNumber, offset + i);
+ }
+ }
+
+ RecordId newId = dedup(
+ hash(T_TEMPLATE).putInt(head).putIds(ids).putBytes(propertyTypes).digest(),
+ RecordWriters.newTemplateWriter(ids, new RecordId[propertyCount],
+ propertyTypes, head, primaryId, mixinIds, childNameId, propNamesId));
+ alias.putIfAbsent(id, newId);
+ return newId;
+ }
+
+ private RecordId copyChildMap(@NotNull RecordId id) throws IOException {
+ RecordId cached = aliasGet(id);
+ if (cached != null) {
+ return cached;
+ }
+ MapRecord map = reader.readMap(id);
+ Map<String, RecordId> changes = new LinkedHashMap<>();
+ for (MapEntry entry : map.getEntries()) {
+ changes.put(entry.getName(), copyNode(entry.getValue()));
+ }
+
+ RecordId newId;
+ if (dedupEnabled() && deduplicateMaps) {
+ // Key on the translated (canonical) entries, order-independent, so content-equal
+ // maps collapse to one record - which in turn lets their referencing nodes dedup.
+ byte[] key = hashMap(changes);
+ if (exactDedup) {
+ newId = dedupCache.computeIfAbsent(key, () -> structuralWriter.writeMap(null, changes));
+ } else {
+ RecordId existing = dedupCache.get(key);
+ if (existing != null) {
+ alias.putIfAbsent(id, existing);
+ return existing;
+ }
+ newId = dedupPut(key, structuralWriter.writeMap(null, changes));
+ }
+ } else {
+ newId = structuralWriter.writeMap(null, changes);
+ }
+ aliasPut(id, newId);
+ return newId;
+ }
+
+ private RecordId copyPropertyList(@NotNull RecordId id, @NotNull PropertyTemplate[] templates)
+ throws IOException {
+ RecordId cached = aliasGet(id);
+ if (cached != null) {
+ return cached;
+ }
+ List<RecordId> entries = new ListRecord(id, templates.length).getEntries();
+ List<RecordId> newEntries = new ArrayList<>(templates.length);
+ for (int i = 0; i < templates.length; i++) {
+ newEntries.add(copyProperty(entries.get(i), templates[i]));
+ }
+ RecordId newId = writeListDeduped(newEntries);
+ aliasPut(id, newId);
+ return newId;
+ }
+
+ private RecordId copyProperty(@NotNull RecordId id, @NotNull PropertyTemplate template)
+ throws IOException {
+ Type<?> type = template.getType();
+ if (!type.isArray()) {
+ return type == Type.BINARY ? copyBinary(id) : copyString(id);
+ }
+
+ RecordId cached = aliasGet(id);
+ if (cached != null) {
+ return cached;
+ }
+
+ Segment segment = id.getSegment();
+ int recordNumber = id.getRecordNumber();
+ int count = segment.readInt(recordNumber);
+
+ RecordId newId;
+ if (count == 0) {
+ newId = dedup(hash(T_EMPTY_ARRAY).digest(), RecordWriters.newListWriter());
+ } else {
+ RecordId listId = segment.readRecordId(recordNumber, 4);
+ List<RecordId> entries = new ListRecord(listId, count).getEntries();
+ Type<?> baseType = type.getBaseType();
+ List<RecordId> newEntries = new ArrayList<>(count);
+ for (RecordId entry : entries) {
+ newEntries.add(baseType == Type.BINARY ? copyBinary(entry) : copyString(entry));
+ }
+ RecordId newListId = writeListDeduped(newEntries);
+ newId = dedup(hash(T_ARRAY).putInt(count).putId(newListId).digest(),
+ RecordWriters.newListWriter(count, newListId));
+ }
+ aliasPut(id, newId);
+ return newId;
+ }
+
+ private RecordId copyBinary(@NotNull RecordId id) throws IOException {
+ RecordId cached = alias.get(id);
+ if (cached != null) {
+ return cached;
+ }
+ // Relink through the segment writer: external blobs re-emit just the blob
+ // id, inline/long binaries re-link the existing bulk block records without
+ // copying the bulk data.
+ RecordId newId = structuralWriter.writeBlob(reader.readBlob(id));
+ alias.putIfAbsent(id, newId);
+ return newId;
+ }
+
+ private RecordId copyString(@NotNull RecordId id) throws IOException {
+ if (keepAsIs(id)) {
+ return id;
+ }
+ RecordId cached = aliasGet(id);
+ if (cached != null) {
+ return cached;
+ }
+
+ Segment segment = id.getSegment();
+ int recordNumber = id.getRecordNumber();
+ long length = segment.readLength(recordNumber);
+
+ RecordId newId;
+ if (length < Segment.SMALL_LIMIT) {
+ byte[] data = readBytes(segment, recordNumber, 1, (int) length);
+ newId = dedup(hash(T_VALUE).putBytes(data).digest(),
+ RecordWriters.newValueWriter((int) length, data));
+ } else if (length < Segment.MEDIUM_LIMIT) {
+ byte[] data = readBytes(segment, recordNumber, 2, (int) length);
+ newId = dedup(hash(T_VALUE).putBytes(data).digest(),
+ RecordWriters.newValueWriter((int) length, data));
+ } else {
+ int count = (int) ((length + BLOCK_SIZE - 1) / BLOCK_SIZE);
+ RecordId listId = segment.readRecordId(recordNumber, 8);
+ List<RecordId> blocks = new ListRecord(listId, count).getEntries();
+ List<RecordId> newBlocks = new ArrayList<>(count);
+ for (int i = 0; i < count; i++) {
+ RecordId block = blocks.get(i);
+ if (keepAsIs(block)) {
+ newBlocks.add(block);
+ } else {
+ int blockLength = (int) Math.min(BLOCK_SIZE, length - (long) i * BLOCK_SIZE);
+ newBlocks.add(copyBlock(block, blockLength));
+ }
+ }
+ RecordId newListId = writeListDeduped(newBlocks);
+ long len = (length - Segment.MEDIUM_LIMIT) | (0x3L << 62);
+ newId = dedup(hash(T_VALUE_LONG).putLong(len).putId(newListId).digest(),
+ RecordWriters.newValueWriter(newListId, len));
+ }
+ aliasPut(id, newId);
+ return newId;
+ }
+
+ /** Content-dedup hits when a bounded (windowed) cache is used, else {@code -1}. */
+ long getDedupWindowHits() {
+ return dedupCache != null ? dedupCache.getHits() : -1;
+ }
+
+ /** Content-dedup misses (fresh emits) when a bounded (windowed) cache is used, else {@code -1}. */
+ long getDedupWindowMisses() {
+ return dedupCache != null ? dedupCache.getMisses() : -1;
+ }
+
+ private RecordId copyBlock(@NotNull RecordId id, int length) throws IOException {
+ RecordId cached = aliasGet(id);
+ if (cached != null) {
+ return cached;
+ }
+ byte[] data = readBytes(id.getSegment(), id.getRecordNumber(), 0, length);
+ RecordId newId = dedup(hash(T_BLOCK).putBytes(data).digest(),
+ RecordWriters.newBlockWriter(data, 0, length));
+ aliasPut(id, newId);
+ return newId;
+ }
+
+ private RecordId copyStringList(@NotNull RecordId id, int count) throws IOException {
+ RecordId cached = aliasGet(id);
+ if (cached != null) {
+ return cached;
+ }
+ List<RecordId> entries = new ListRecord(id, count).getEntries();
+ List<RecordId> newEntries = new ArrayList<>(count);
+ for (RecordId entry : entries) {
+ newEntries.add(copyString(entry));
+ }
+ RecordId newId = writeListDeduped(newEntries);
+ aliasPut(id, newId);
+ return newId;
+ }
+
+ private static byte[] readBytes(@NotNull Segment segment, int recordNumber, int position, int length) {
+ byte[] data = new byte[length];
+ segment.readBytes(recordNumber, position, data, 0, length);
+ return data;
+ }
+
+ /**
+ * Order-independent content hash of a child map's translated {@code name -> id} entries.
+ */
+ @NotNull
+ private byte[] hashMap(@NotNull Map<String, RecordId> changes) {
+ HashKey hk = hash(T_MAP).putInt(changes.size());
+ for (Map.Entry<String, RecordId> entry : new TreeMap<>(changes).entrySet()) {
+ hk.putBytes(entry.getKey().getBytes(StandardCharsets.UTF_8)).putId(entry.getValue());
+ }
+ return hk.digest();
+ }
+
+ /** Start a new {@link #dedupCache} key, tagged by record type, over the reused {@link HashKey}. */
+ @NotNull
+ private HashKey hash(int tag) {
+ return hashKey.get().reset(tag);
+ }
+
+ /**
+ * A {@link WriteOperationHandler} that serialises all writes onto a single (not thread-safe)
+ * {@link SegmentBufferWriter} by synchronising every operation. Used by {@link #singleWriter} so
+ * that parallel workers co-pack their records into one segment stream instead of one per thread.
+ */
+ private static final class SynchronizedWriteHandler implements WriteOperationHandler {
+ @NotNull
+ private final SegmentBufferWriter writer;
+
+ SynchronizedWriteHandler(@NotNull SegmentBufferWriter writer) {
+ this.writer = requireNonNull(writer);
+ }
+
+ @NotNull
+ @Override
+ public synchronized RecordId execute(@NotNull GCGeneration gcGeneration,
+ @NotNull WriteOperation writeOperation) throws IOException {
+ return writer.execute(gcGeneration, writeOperation);
+ }
+
+ @Override
+ public synchronized void flush(@NotNull SegmentStore store) throws IOException {
+ writer.flush(store);
+ }
+
+ @NotNull
+ @Override
+ public GCGeneration getGCGeneration() {
+ return writer.getGCGeneration();
+ }
+ }
+
+ /**
+ * Builder for a fixed-width ({@value #KEY_HASH_WIDTH}-byte) content hash used as a
+ * {@link #dedupCache} key. Fields are fed in a canonical, length-prefixed order so that
+ * structurally different records cannot collide, then hashed with the 128-bit MurmurHash3
+ * ({@link MurmurHash3#hash128x64}). A non-cryptographic hash is deliberate: the input is
+ * trusted internal segment data (not adversarial), and 128 bits keeps the collision
+ * probability negligible (~10^-26 at 15M records) at a fraction of a cryptographic hash's cost.
+ * The builder reuses its byte buffer (cleared on {@link #reset}), so at most one key may be in
+ * flight at a time - which holds because the repacker is single-threaded and each key is fully
+ * built and digested before the next {@link #reset}.
+ */
+ private static final class HashKey {
+ private byte[] buffer = new byte[256];
+ private int length;
+
+ @NotNull
+ HashKey reset(int tag) {
+ length = 0;
+ return putInt(tag);
+ }
+
+ private void ensure(int extra) {
+ if (length + extra > buffer.length) {
+ buffer = Arrays.copyOf(buffer, Math.max(buffer.length * 2, length + extra));
+ }
+ }
+
+ @NotNull
+ HashKey putByte(int b) {
+ ensure(1);
+ buffer[length++] = (byte) b;
+ return this;
+ }
+
+ @NotNull
+ HashKey putInt(int value) {
+ ensure(4);
+ buffer[length++] = (byte) (value >>> 24);
+ buffer[length++] = (byte) (value >>> 16);
+ buffer[length++] = (byte) (value >>> 8);
+ buffer[length++] = (byte) value;
+ return this;
+ }
+
+ @NotNull
+ HashKey putLong(long value) {
+ ensure(8);
+ for (int i = 0; i < 8; i++) {
+ buffer[length++] = (byte) (value >>> (56 - 8 * i));
+ }
+ return this;
+ }
+
+ @NotNull
+ HashKey putBytes(@NotNull byte[] bytes) {
+ putInt(bytes.length);
+ ensure(bytes.length);
+ System.arraycopy(bytes, 0, buffer, length, bytes.length);
+ length += bytes.length;
+ return this;
+ }
+
+ @NotNull
+ HashKey putId(@NotNull RecordId id) {
+ SegmentId segmentId = id.getSegmentId();
+ putLong(segmentId.getMostSignificantBits());
+ putLong(segmentId.getLeastSignificantBits());
+ putInt(id.getRecordNumber());
+ return this;
+ }
+
+ @NotNull
+ HashKey putIds(@NotNull List<RecordId> ids) {
+ putInt(ids.size());
+ for (RecordId id : ids) {
+ putId(id);
+ }
+ return this;
+ }
+
+ @NotNull
+ byte[] digest() {
+ long[] hash = MurmurHash3.hash128x64(buffer, 0, length, 0);
+ byte[] out = new byte[KEY_HASH_WIDTH];
+ writeLong(out, 0, hash[0]);
+ writeLong(out, 8, hash[1]);
+ return out;
+ }
+
+ private static void writeLong(byte[] out, int offset, long value) {
+ for (int i = 0; i < 8; i++) {
+ out[offset + i] = (byte) (value >>> (56 - 8 * i));
+ }
+ }
+ }
+}
diff --git a/oak-segment-tar/src/main/java/org/apache/jackrabbit/oak/segment/ShardedRecordIdMap.java b/oak-segment-tar/src/main/java/org/apache/jackrabbit/oak/segment/ShardedRecordIdMap.java
new file mode 100644
index 0000000000..4f2675e572
--- /dev/null
+++ b/oak-segment-tar/src/main/java/org/apache/jackrabbit/oak/segment/ShardedRecordIdMap.java
@@ -0,0 +1,172 @@
+/*
+ * 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.jackrabbit.oak.segment;
+
+import java.io.IOException;
+
+import org.jetbrains.annotations.NotNull;
+import org.jetbrains.annotations.Nullable;
+
+/**
+ * A thread-safe {@link RecordId} map that partitions its entries across a fixed (power-of-two)
+ * number of independently locked {@link SpillableRecordIdMap} shards. A key is routed to a shard by
+ * a hash of the key, so concurrent operations on different shards proceed without contention; only
+ * operations hitting the same shard serialise. Used by {@link RecordRepacker} for the alias and
+ * content-deduplication tables when repacking with more than one thread.
+ * <p>
+ * With a single shard it degenerates to one {@link SpillableRecordIdMap} behind an (uncontended)
+ * lock, matching single-threaded behaviour. Each shard keeps its entries off the Java heap in a
+ * memory-mapped file (see {@link SpillableRecordIdMap}); the shards' backing descriptors are
+ * released right after mapping, so a large shard count does not exhaust file descriptors.
+ */
+final class ShardedRecordIdMap implements DedupCache {
+
+ private final SpillableRecordIdMap[] shards;
+
+ private final Object[] locks;
+
+ private final int mask;
+
+ /**
+ * @param expectedRecords estimate of the total number of entries, split evenly across shards to
+ * pre-size each shard; {@code 0} (or negative) uses a small default.
+ * @param shardCount desired number of shards; rounded down to a power of two (at least one).
+ */
+ ShardedRecordIdMap(@NotNull SegmentIdProvider idProvider, int keyWidth, int expectedRecords,
+ int shardCount, @NotNull String name) throws IOException {
+ int count = Integer.highestOneBit(Math.max(1, shardCount));
+ this.shards = new SpillableRecordIdMap[count];
+ this.locks = new Object[count];
+ this.mask = count - 1;
+ int perShard = perShardCapacity(expectedRecords, count);
+ for (int i = 0; i < count; i++) {
+ shards[i] = new SpillableRecordIdMap(idProvider, keyWidth, perShard, name + "-" + i);
+ locks[i] = new Object();
+ }
+ }
+
+ @Nullable
+ RecordId get(@NotNull RecordId key) {
+ int i = shardOf(key);
+ synchronized (locks[i]) {
+ return shards[i].get(key);
+ }
+ }
+
+ @Override
+ @Nullable
+ public RecordId get(@NotNull byte[] key) {
+ int i = shardOf(key);
+ synchronized (locks[i]) {
+ return shards[i].get(key);
+ }
+ }
+
+ /**
+ * Insert {@code (key, value)} if the key is absent, and return the value already mapped (a
+ * competing insert won) or {@code null} if this call inserted. The get and the conditional put
+ * happen atomically under the shard lock, so concurrent callers converge on a single value.
+ */
+ @Nullable
+ RecordId putIfAbsent(@NotNull RecordId key, @NotNull RecordId value) throws IOException {
+ int i = shardOf(key);
+ synchronized (locks[i]) {
+ RecordId existing = shards[i].get(key);
+ if (existing != null) {
+ return existing;
+ }
+ shards[i].put(key, value);
+ return null;
+ }
+ }
+
+ @Override
+ @Nullable
+ public RecordId putIfAbsent(@NotNull byte[] key, @NotNull RecordId value) throws IOException {
+ int i = shardOf(key);
+ synchronized (locks[i]) {
+ RecordId existing = shards[i].get(key);
+ if (existing != null) {
+ return existing;
+ }
+ shards[i].put(key, value);
+ return null;
+ }
+ }
+
+ /** A supplier of a {@link RecordId} that may fail with an {@link IOException}; see {@link #computeIfAbsent}. */
+ @FunctionalInterface
+ interface RecordIdSupplier {
+ @NotNull
+ RecordId get() throws IOException;
+ }
+
+ /**
+ * Atomically return the value mapped to {@code key}, or, if absent, invoke {@code writer} to
+ * produce a value, store it and return it - all while holding the shard lock. A concurrent
+ * caller with the same key therefore blocks until this call completes and then observes the
+ * stored value; it never invokes its own {@code writer}. Used to make record emission exactly
+ * deduplicated under concurrency: the "is this content already emitted?" check and the emit
+ * happen as one critical section, so racing workers never both emit (and so never leave a loser
+ * copy as unreferenced garbage). The trade-off is that {@code writer} (the record emit) runs
+ * under the shard lock, serialising emits whose keys land in the same shard.
+ */
+ @Override
+ @NotNull
+ public RecordId computeIfAbsent(@NotNull byte[] key, @NotNull RecordIdSupplier writer)
+ throws IOException {
+ int i = shardOf(key);
+ synchronized (locks[i]) {
+ RecordId existing = shards[i].get(key);
+ if (existing != null) {
+ return existing;
+ }
+ RecordId created = writer.get();
+ shards[i].put(key, created);
+ return created;
+ }
+ }
+
+ @Override
+ public void close() {
+ for (SpillableRecordIdMap shard : shards) {
+ shard.close();
+ }
+ }
+
+ private int shardOf(@NotNull RecordId key) {
+ long lsb = key.getSegmentId().getLeastSignificantBits();
+ return (int) (lsb ^ (lsb >>> 32) ^ key.getRecordNumber()) & mask;
+ }
+
+ private int shardOf(@NotNull byte[] key) {
+ // The keys are already uniformly distributed (serialised random segment ids, or a content
+ // hash), so folding the leading four bytes into an int gives a well-spread shard index.
+ int h = ((key[0] & 0xff) << 24) | ((key[1] & 0xff) << 16)
+ | ((key[2] & 0xff) << 8) | (key[3] & 0xff);
+ return (h ^ (h >>> 16)) & mask;
+ }
+
+ /** Initial slot count for each shard: ~twice the per-shard expected entries, or a small default. */
+ private static int perShardCapacity(int expectedRecords, int shardCount) {
+ if (expectedRecords <= 0) {
+ return 1 << 12;
+ }
+ long perShard = ((long) expectedRecords / shardCount) * 2;
+ return (int) Math.min(Integer.MAX_VALUE / 2, Math.max(16, perShard));
+ }
+}
diff --git a/oak-segment-tar/src/main/java/org/apache/jackrabbit/oak/segment/SpillableRecordIdMap.java b/oak-segment-tar/src/main/java/org/apache/jackrabbit/oak/segment/SpillableRecordIdMap.java
new file mode 100644
index 0000000000..9f9ef6cb15
--- /dev/null
+++ b/oak-segment-tar/src/main/java/org/apache/jackrabbit/oak/segment/SpillableRecordIdMap.java
@@ -0,0 +1,304 @@
+/*
+ * 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.jackrabbit.oak.segment;
+
+import static java.nio.channels.FileChannel.MapMode.READ_WRITE;
+import static java.util.Objects.requireNonNull;
+
+import java.io.Closeable;
+import java.io.IOException;
+import java.io.RandomAccessFile;
+import java.lang.reflect.Field;
+import java.lang.reflect.Method;
+import java.nio.ByteBuffer;
+import java.nio.MappedByteBuffer;
+import java.nio.file.Files;
+import java.nio.file.Path;
+
+import org.jetbrains.annotations.NotNull;
+import org.jetbrains.annotations.Nullable;
+
+/**
+ * A disk-backed (memory mapped) hash map from a fixed-width byte key to a
+ * {@link RecordId}, used by {@link RecordRepacker} to keep the alias table and the
+ * content-deduplication cache off the Java heap. The mapping is stored in a temporary
+ * file which is deleted when the map is {@link #close() closed}, so the entries do not
+ * count against the heap and can grow well beyond it.
+ * <p>
+ * The table is an open-addressing hash table with linear probing. Each slot holds a
+ * one-byte occupancy flag, the {@code keyWidth}-byte key and the 20-byte serialised
+ * value ({@code msb, lsb, recordNumber}). Values are reconstructed through a
+ * {@link SegmentIdProvider}, which interns the (few) target-generation segment ids.
+ * Entries are never removed; the table grows (and rehashes into a fresh file) once the
+ * load factor is exceeded.
+ * <p>
+ * This is a prototype helper: it is single-threaded, keeps at most one live mapping and
+ * therefore caps a single table at 2 GB. That is ample for repacking the stores this
+ * prototype targets; a larger table would need a paged mapping.
+ */
+final class SpillableRecordIdMap implements Closeable {
+
+ private static final int VALUE_WIDTH = 20; // msb (8) + lsb (8) + recordNumber (4)
+
+ private static final float LOAD_FACTOR = 0.6f;
+
+ @NotNull
+ private final SegmentIdProvider idProvider;
+
+ private final int keyWidth;
+
+ private final int slotWidth;
+
+ /** Largest power-of-two capacity whose byte size still fits a single {@code int}-addressed mapping. */
+ private final int maxCapacity;
+
+ private final String name;
+
+ private Path file;
+
+ private MappedByteBuffer buffer;
+
+ private int capacity;
+
+ private long size;
+
+ /** Reusable scratch for the serialised key of the {@link RecordId} key overloads. */
+ private final byte[] keyScratch;
+
+ SpillableRecordIdMap(@NotNull SegmentIdProvider idProvider, int keyWidth, int initialCapacity,
+ @NotNull String name) throws IOException {
+ this.idProvider = requireNonNull(idProvider);
+ this.keyWidth = keyWidth;
+ this.slotWidth = 1 + keyWidth + VALUE_WIDTH;
+ this.maxCapacity = Integer.highestOneBit(Integer.MAX_VALUE / slotWidth);
+ this.name = requireNonNull(name);
+ this.keyScratch = new byte[keyWidth];
+ this.capacity = Math.min(tableSize(initialCapacity), maxCapacity);
+ this.file = Files.createTempFile("oak-repack-" + name + "-", ".map");
+ this.file.toFile().deleteOnExit();
+ this.buffer = mapFile(file, (long) capacity * slotWidth);
+ }
+
+ // -- RecordId key overloads (alias table) --------------------------------
+
+ @Nullable
+ RecordId get(@NotNull RecordId key) {
+ serializeId(key, keyScratch);
+ return get(keyScratch);
+ }
+
+ void put(@NotNull RecordId key, @NotNull RecordId value) throws IOException {
+ serializeId(key, keyScratch);
+ put(keyScratch, value);
+ }
+
+ // -- byte[] key overloads (dedup cache) ----------------------------------
+
+ @Nullable
+ RecordId get(@NotNull byte[] key) {
+ int index = slotFor(key, capacity);
+ while (true) {
+ int pos = index * slotWidth;
+ if (buffer.get(pos) == 0) {
+ return null;
+ }
+ if (keyEquals(pos, key)) {
+ return readValue(pos);
+ }
+ index = (index + 1) & (capacity - 1);
+ }
+ }
+
+ void put(@NotNull byte[] key, @NotNull RecordId value) throws IOException {
+ if (size + 1 > (long) (capacity * LOAD_FACTOR)) {
+ grow();
+ }
+ insert(buffer, capacity, key, value.getSegmentId().getMostSignificantBits(),
+ value.getSegmentId().getLeastSignificantBits(), value.getRecordNumber());
+ }
+
+ long size() {
+ return size;
+ }
+
+ @Override
+ public void close() {
+ unmap(buffer);
+ buffer = null;
+ if (file != null) {
+ file.toFile().delete();
+ file = null;
+ }
+ }
+
+ // -- internals -----------------------------------------------------------
+
+ /**
+ * Insert into {@code buf} (of {@code cap} slots), incrementing {@link #size} only when a
+ * new key is added. An existing key is overwritten in place.
+ */
+ private void insert(MappedByteBuffer buf, int cap, byte[] key, long msb, long lsb, int recordNumber) {
+ int index = slotFor(key, cap);
+ while (true) {
+ int pos = index * slotWidth;
+ if (buf.get(pos) == 0) {
+ buf.put(pos, (byte) 1);
+ int base = pos + 1;
+ for (int i = 0; i < keyWidth; i++) {
+ buf.put(base + i, key[i]);
+ }
+ int vbase = base + keyWidth;
+ buf.putLong(vbase, msb);
+ buf.putLong(vbase + 8, lsb);
+ buf.putInt(vbase + 16, recordNumber);
+ size++;
+ return;
+ }
+ if (bufKeyEquals(buf, pos, key)) {
+ int vbase = pos + 1 + keyWidth;
+ buf.putLong(vbase, msb);
+ buf.putLong(vbase + 8, lsb);
+ buf.putInt(vbase + 16, recordNumber);
+ return;
+ }
+ index = (index + 1) & (cap - 1);
+ }
+ }
+
+ private void grow() throws IOException {
+ if (capacity >= maxCapacity) {
+ throw new IllegalStateException("Spillable map '" + name + "' exceeded the maximum size of "
+ + ((long) maxCapacity * slotWidth) + " bytes");
+ }
+ int newCapacity = capacity * 2;
+ Path newFile = Files.createTempFile("oak-repack-" + name + "-", ".map");
+ newFile.toFile().deleteOnExit();
+ MappedByteBuffer newBuffer = mapFile(newFile, (long) newCapacity * slotWidth);
+
+ byte[] key = new byte[keyWidth];
+ long moved = size;
+ size = 0;
+ for (int index = 0; index < capacity; index++) {
+ int pos = index * slotWidth;
+ if (buffer.get(pos) != 0) {
+ int base = pos + 1;
+ for (int i = 0; i < keyWidth; i++) {
+ key[i] = buffer.get(base + i);
+ }
+ int vbase = base + keyWidth;
+ insert(newBuffer, newCapacity, key, buffer.getLong(vbase),
+ buffer.getLong(vbase + 8), buffer.getInt(vbase + 16));
+ }
+ }
+ assert size == moved : "rehash lost entries";
+
+ unmap(buffer);
+ file.toFile().delete();
+
+ this.buffer = newBuffer;
+ this.file = newFile;
+ this.capacity = newCapacity;
+ }
+
+ private int slotFor(byte[] key, int cap) {
+ return (int) (hash(key) & (cap - 1));
+ }
+
+ private boolean keyEquals(int pos, byte[] key) {
+ return bufKeyEquals(buffer, pos, key);
+ }
+
+ private boolean bufKeyEquals(MappedByteBuffer buf, int pos, byte[] key) {
+ int base = pos + 1;
+ for (int i = 0; i < keyWidth; i++) {
+ if (buf.get(base + i) != key[i]) {
+ return false;
+ }
+ }
+ return true;
+ }
+
+ private RecordId readValue(int pos) {
+ int vbase = pos + 1 + keyWidth;
+ long msb = buffer.getLong(vbase);
+ long lsb = buffer.getLong(vbase + 8);
+ int recordNumber = buffer.getInt(vbase + 16);
+ return new RecordId(idProvider.newSegmentId(msb, lsb), recordNumber);
+ }
+
+ private static void serializeId(RecordId id, byte[] out) {
+ SegmentId segmentId = id.getSegmentId();
+ writeLong(out, 0, segmentId.getMostSignificantBits());
+ writeLong(out, 8, segmentId.getLeastSignificantBits());
+ writeInt(out, 16, id.getRecordNumber());
+ }
+
+ private static void writeLong(byte[] out, int offset, long value) {
+ for (int i = 0; i < 8; i++) {
+ out[offset + i] = (byte) (value >>> (56 - 8 * i));
+ }
+ }
+
+ private static void writeInt(byte[] out, int offset, int value) {
+ for (int i = 0; i < 4; i++) {
+ out[offset + i] = (byte) (value >>> (24 - 8 * i));
+ }
+ }
+
+ /** FNV-1a over the key bytes with a final avalanche mix. */
+ private long hash(byte[] key) {
+ long h = 0xcbf29ce484222325L;
+ for (int i = 0; i < keyWidth; i++) {
+ h ^= (key[i] & 0xffL);
+ h *= 0x100000001b3L;
+ }
+ h ^= (h >>> 32);
+ return h;
+ }
+
+ private static int tableSize(int expectedEntries) {
+ int min = Math.max(16, expectedEntries);
+ int size = Integer.highestOneBit(min - 1) << 1;
+ return size < 16 ? 16 : size;
+ }
+
+ private static MappedByteBuffer mapFile(Path file, long bytes) throws IOException {
+ // The mapping outlives the channel/descriptor, so the backing file is only opened long
+ // enough to size and map it. This keeps the descriptor count low even with many shards.
+ try (RandomAccessFile raf = new RandomAccessFile(file.toFile(), "rw")) {
+ raf.setLength(bytes);
+ return raf.getChannel().map(READ_WRITE, 0, bytes);
+ }
+ }
+
+ /** Best-effort unmap so the (deleted) file's pages are released promptly rather than at GC. */
+ private static void unmap(MappedByteBuffer buffer) {
+ if (buffer == null) {
+ return;
+ }
+ try {
+ Class<?> unsafeClass = Class.forName("sun.misc.Unsafe");
+ Field theUnsafe = unsafeClass.getDeclaredField("theUnsafe");
+ theUnsafe.setAccessible(true);
+ Object unsafe = theUnsafe.get(null);
+ Method invokeCleaner = unsafeClass.getMethod("invokeCleaner", ByteBuffer.class);
+ invokeCleaner.invoke(unsafe, buffer);
+ } catch (Throwable t) {
+ // The mapping will be released when the buffer is garbage collected.
+ }
+ }
+}
diff --git a/oak-segment-tar/src/main/java/org/apache/jackrabbit/oak/segment/WindowDedupCache.java b/oak-segment-tar/src/main/java/org/apache/jackrabbit/oak/segment/WindowDedupCache.java
new file mode 100644
index 0000000000..e1e3db553e
--- /dev/null
+++ b/oak-segment-tar/src/main/java/org/apache/jackrabbit/oak/segment/WindowDedupCache.java
@@ -0,0 +1,187 @@
+/*
+ * 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.jackrabbit.oak.segment;
+
+import java.io.IOException;
+import java.util.LinkedHashMap;
+import java.util.Map;
+import java.util.concurrent.atomic.LongAdder;
+
+import org.jetbrains.annotations.NotNull;
+import org.jetbrains.annotations.Nullable;
+
+/**
+ * A bounded, in-heap, sharded LRU {@link DedupCache}: a record is deduplicated only against the most
+ * recently emitted records still resident in the window, and older ones are forgotten. Unlike the exact
+ * {@link ShardedRecordIdMap} it needs no off-heap spill and its footprint is capped at {@code maxEntries}
+ * entries, trading a little size - duplicates whose recurrences are farther apart than the window are
+ * re-emitted - for bounded heap and better read locality: a reused copy is always recent, so it sits near
+ * its referrer rather than in a remote canonical segment.
+ * <p>
+ * This is the "sliding window" hot-record detector: content that recurs while still in the window is
+ * "hot" (deduplicated); a one-off "cold" record falls out and is emitted fresh (the repacker's alias
+ * still preserves the input's physical record sharing). It self-identifies hot vs cold for every record
+ * type with no pre-pass, ranking or explicit hot set. Keys are the repacker's fixed-width (128-bit)
+ * content hashes, compared in full, so there is no risk of a false deduplication.
+ * <p>
+ * Thread-safe: each shard is guarded by its own monitor and {@link #computeIfAbsent} emits while holding
+ * it, so concurrent workers with the same content converge on one copy (for content still in the window).
+ */
+final class WindowDedupCache implements DedupCache {
+
+ private final Shard[] shards;
+ private final LongAdder hits = new LongAdder();
+ private final LongAdder misses = new LongAdder();
+
+ /**
+ * @param maxEntries total number of most-recent entries to retain across all shards (the window
+ * size); split evenly per shard.
+ * @param shardCount number of independently-locked shards (match the repacker's concurrency).
+ */
+ WindowDedupCache(int maxEntries, int shardCount) {
+ int n = Math.max(1, shardCount);
+ int perShard = Math.max(1, maxEntries / n);
+ shards = new Shard[n];
+ for (int i = 0; i < n; i++) {
+ shards[i] = new Shard(perShard);
+ }
+ }
+
+ private Shard shardFor(@NotNull byte[] key) {
+ // The key is a content hash; its bits are already well distributed.
+ return shards[(int) Math.floorMod(readLong(key, 0), shards.length)];
+ }
+
+ @Override
+ @Nullable
+ public RecordId get(@NotNull byte[] key) {
+ Shard s = shardFor(key);
+ Key k = new Key(key);
+ synchronized (s) {
+ return s.map.get(k);
+ }
+ }
+
+ @Override
+ @Nullable
+ public RecordId putIfAbsent(@NotNull byte[] key, @NotNull RecordId value) {
+ Shard s = shardFor(key);
+ Key k = new Key(key);
+ synchronized (s) {
+ RecordId existing = s.map.get(k);
+ if (existing != null) {
+ return existing;
+ }
+ s.map.put(k, value);
+ return null;
+ }
+ }
+
+ @Override
+ @NotNull
+ public RecordId computeIfAbsent(@NotNull byte[] key,
+ @NotNull ShardedRecordIdMap.RecordIdSupplier writer) throws IOException {
+ Shard s = shardFor(key);
+ Key k = new Key(key);
+ synchronized (s) {
+ RecordId existing = s.map.get(k);
+ if (existing != null) {
+ hits.increment();
+ return existing;
+ }
+ RecordId created = writer.get();
+ s.map.put(k, created);
+ misses.increment();
+ return created;
+ }
+ }
+
+ @Override
+ public void close() {
+ for (Shard s : shards) {
+ synchronized (s) {
+ s.map.clear();
+ }
+ }
+ }
+
+ /** Deduplication hits (content found in the window). */
+ @Override
+ public long getHits() {
+ return hits.sum();
+ }
+
+ /** Deduplication misses (content not in the window - emitted fresh). */
+ @Override
+ public long getMisses() {
+ return misses.sum();
+ }
+
+ /** Access-ordered LRU map bounded to {@code capacity} entries, guarded by its own monitor. */
+ private static final class Shard {
+ final LinkedHashMap<Key, RecordId> map;
+
+ Shard(int capacity) {
+ this.map = new LinkedHashMap<Key, RecordId>(16, 0.75f, true) {
+ @Override
+ protected boolean removeEldestEntry(Map.Entry<Key, RecordId> eldest) {
+ return size() > capacity;
+ }
+ };
+ }
+ }
+
+ /** Immutable 128-bit content-hash key (a copy, since callers reuse their hash buffer). */
+ private static final class Key {
+ private final long hi;
+ private final long lo;
+
+ Key(@NotNull byte[] b) {
+ this.hi = readLong(b, 0);
+ this.lo = b.length >= 16 ? readLong(b, 8) : 0L;
+ }
+
+ @Override
+ public int hashCode() {
+ long h = hi ^ lo;
+ return (int) (h ^ (h >>> 32));
+ }
+
+ @Override
+ public boolean equals(Object o) {
+ if (this == o) {
+ return true;
+ }
+ if (!(o instanceof Key)) {
+ return false;
+ }
+ Key k = (Key) o;
+ return hi == k.hi && lo == k.lo;
+ }
+ }
+
+ static long readLong(@NotNull byte[] b, int off) {
+ return ((long) (b[off] & 0xff) << 56)
+ | ((long) (b[off + 1] & 0xff) << 48)
+ | ((long) (b[off + 2] & 0xff) << 40)
+ | ((long) (b[off + 3] & 0xff) << 32)
+ | ((long) (b[off + 4] & 0xff) << 24)
+ | ((long) (b[off + 5] & 0xff) << 16)
+ | ((long) (b[off + 6] & 0xff) << 8)
+ | ((long) (b[off + 7] & 0xff));
+ }
+}
diff --git a/oak-segment-tar/src/test/java/org/apache/jackrabbit/oak/segment/RecordRepackerTest.java b/oak-segment-tar/src/test/java/org/apache/jackrabbit/oak/segment/RecordRepackerTest.java
new file mode 100644
index 0000000000..84eacaedc6
--- /dev/null
+++ b/oak-segment-tar/src/test/java/org/apache/jackrabbit/oak/segment/RecordRepackerTest.java
@@ -0,0 +1,3295 @@
+/*
+ * 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.jackrabbit.oak.segment;
+
+import static org.apache.jackrabbit.oak.api.Type.BINARIES;
+import static org.apache.jackrabbit.oak.api.Type.BINARY;
+import static org.apache.jackrabbit.oak.api.Type.LONGS;
+import static org.apache.jackrabbit.oak.api.Type.NAME;
+import static org.apache.jackrabbit.oak.api.Type.NAMES;
+import static org.apache.jackrabbit.oak.api.Type.STRINGS;
+import static org.apache.jackrabbit.oak.plugins.memory.EmptyNodeState.EMPTY_NODE;
+import static org.apache.jackrabbit.oak.segment.DefaultSegmentWriterBuilder.defaultSegmentWriterBuilder;
+import static org.apache.jackrabbit.oak.segment.file.FileStoreBuilder.fileStoreBuilder;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNotEquals;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.junit.jupiter.api.Assumptions.assumeFalse;
+import static org.junit.jupiter.api.Assumptions.assumeTrue;
+
+import java.io.ByteArrayInputStream;
+import java.io.File;
+import java.io.IOException;
+import java.io.InputStream;
+import java.lang.management.ManagementFactory;
+import java.lang.management.MemoryMXBean;
+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.ArrayDeque;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.BitSet;
+import java.util.Collections;
+import java.util.Comparator;
+import java.util.Deque;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicLong;
+
+import org.apache.jackrabbit.oak.api.Blob;
+import org.apache.jackrabbit.oak.api.CommitFailedException;
+import org.apache.jackrabbit.oak.api.PropertyState;
+import org.apache.jackrabbit.oak.api.Type;
+import org.apache.jackrabbit.oak.plugins.blob.datastore.DataStoreBlobStore;
+import org.apache.jackrabbit.oak.segment.file.CompactedNodeState;
+import org.apache.jackrabbit.oak.segment.file.CompactionWriter;
+import org.apache.jackrabbit.oak.segment.file.FileStore;
+import org.apache.jackrabbit.oak.segment.file.FileStoreBuilder;
+import org.apache.jackrabbit.oak.segment.file.GCIncrement;
+import org.apache.jackrabbit.oak.segment.file.GCNodeWriteMonitor;
+import org.apache.jackrabbit.oak.segment.file.ReadOnlyFileStore;
+import org.apache.jackrabbit.oak.segment.file.cancel.Canceller;
+import org.apache.jackrabbit.oak.segment.memory.MemoryStore;
+import org.apache.jackrabbit.oak.segment.spi.monitor.IOMonitor;
+import org.apache.jackrabbit.oak.segment.spi.persistence.GCGeneration;
+import org.apache.jackrabbit.oak.spi.blob.BlobOptions;
+import org.apache.jackrabbit.oak.spi.blob.BlobStore;
+import org.apache.jackrabbit.oak.spi.commit.CommitInfo;
+import org.apache.jackrabbit.oak.spi.commit.EmptyHook;
+import org.apache.jackrabbit.oak.spi.gc.GCMonitor;
+import org.apache.jackrabbit.oak.spi.state.ChildNodeEntry;
+import org.apache.jackrabbit.oak.spi.state.NodeBuilder;
+import org.apache.jackrabbit.oak.spi.state.NodeState;
+import org.jetbrains.annotations.NotNull;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
+
+class RecordRepackerTest {
+
+ /**
+ * Directory of a sample segment store to run the {@link #repackSampleStore}
+ * test against.
+ */
+ private static final String SAMPLE_STORE = System.getProperty("oak.repack.sampleStore");
+
+ private MemoryStore store;
+
+ private SegmentWriter writer;
+
+ @BeforeEach
+ void setup() throws IOException {
+ store = new MemoryStore();
+ writer = defaultSegmentWriterBuilder("test").build(store);
+ }
+
+ private RecordId write(NodeState state) throws IOException {
+ RecordId id = writer.writeNode(state);
+ writer.flush();
+ return id;
+ }
+
+ private RecordId repack(RecordId rootId) throws IOException {
+ GCGeneration target = rootId.getSegmentId().getGcGeneration().nextFull();
+ RecordRepacker repacker = new RecordRepacker(store, store.getReader(),
+ store.getSegmentIdProvider(), store.getBlobStore(), Segment.MEDIUM_LIMIT, target);
+ RecordId newRoot = repacker.repack(rootId);
+ assertEquals(target, newRoot.getSegmentId().getGcGeneration(),
+ "repacked root must be in the target generation");
+ return newRoot;
+ }
+
+ /**
+ * Repack {@code before}, then assert that the repacked tree is content-equal
+ * to the original, has an identical stable id, and lives in a fresh generation.
+ */
+ private void assertRoundTrips(NodeState before) throws IOException {
+ RecordId rootId = write(before);
+ SegmentNodeState beforeNode = store.getReader().readNode(rootId);
+
+ RecordId newRootId = repack(rootId);
+ SegmentNodeState afterNode = store.getReader().readNode(newRootId);
+
+ assertNotEquals(rootId, newRootId, "repacking must move the root to a new record");
+ // Structural (deep) comparison against the in-memory state: the SegmentNodeState
+ // fast-path on stable id is bypassed because 'before' is not a SegmentNodeState.
+ assertEquals(before, afterNode);
+ assertEquals(beforeNode.getStableId(), afterNode.getStableId(),
+ "stable id must be preserved across record-level compaction");
+ }
+
+ @Test
+ void emptyNode() throws IOException {
+ assertRoundTrips(EMPTY_NODE);
+ }
+
+ @Test
+ void simpleProperties() throws IOException {
+ NodeState before = EMPTY_NODE.builder()
+ .setProperty("string", "abc")
+ .setProperty("long", 123L)
+ .setProperty("double", Math.PI)
+ .setProperty("boolean", true)
+ .getNodeState();
+ assertRoundTrips(before);
+ }
+
+ @Test
+ void primaryTypeAndMixins() throws IOException {
+ NodeBuilder builder = EMPTY_NODE.builder();
+ builder.setProperty("jcr:primaryType", "nt:unstructured", NAME);
+ builder.setProperty("jcr:mixinTypes", List.of("mix:versionable", "mix:lockable"), NAMES);
+ builder.setProperty("prop", "value");
+ assertRoundTrips(builder.getNodeState());
+ }
+
+ @Test
+ void singleChild() throws IOException {
+ NodeBuilder builder = EMPTY_NODE.builder();
+ builder.child("only").setProperty("p", "v");
+ assertRoundTrips(builder.getNodeState());
+ }
+
+ @Test
+ void deeplyNested() throws IOException {
+ NodeBuilder root = EMPTY_NODE.builder();
+ NodeBuilder builder = root;
+ for (int i = 0; i < 500; i++) {
+ builder = builder.child("level" + i);
+ builder.setProperty("depth", (long) i);
+ }
+ assertRoundTrips(root.getNodeState());
+ }
+
+ @Test
+ void manyChildrenMap() throws IOException {
+ NodeBuilder builder = EMPTY_NODE.builder();
+ for (int i = 0; i < 400; i++) {
+ builder.child("child" + i).setProperty("index", (long) i);
+ }
+ assertRoundTrips(builder.getNodeState());
+ }
+
+ @Test
+ void multiValuedProperties() throws IOException {
+ List<Long> longs = new ArrayList<>();
+ List<String> strings = new ArrayList<>();
+ for (int i = 0; i < 50; i++) {
+ longs.add((long) i);
+ strings.add("value-" + i);
+ }
+ NodeBuilder builder = EMPTY_NODE.builder();
+ builder.setProperty("longs", longs, LONGS);
+ builder.setProperty("strings", strings, STRINGS);
+ builder.setProperty("empty", List.<String>of(), STRINGS);
+ assertRoundTrips(builder.getNodeState());
+ }
+
+ @Test
+ void longStringProperty() throws IOException {
+ StringBuilder sb = new StringBuilder();
+ while (sb.length() < Segment.MEDIUM_LIMIT * 4) {
+ sb.append("The quick brown fox jumps over the lazy dog. ");
+ }
+ NodeState before = EMPTY_NODE.builder()
+ .setProperty("long-string", sb.toString())
+ .getNodeState();
+ assertRoundTrips(before);
+ }
+
+ @Test
+ void binaryProperties() throws IOException {
+ byte[] small = randomBytes(0x40);
+ byte[] medium = randomBytes(Segment.MEDIUM_LIMIT + 100);
+ byte[] large = randomBytes(Segment.MAX_SEGMENT_SIZE + 4096);
+
+ NodeBuilder builder = EMPTY_NODE.builder();
+ builder.setProperty("small", blob(small), BINARY);
+ builder.setProperty("medium", blob(medium), BINARY);
+ builder.setProperty("large", blob(large), BINARY);
+ assertRoundTrips(builder.getNodeState());
+ }
+
+ @Test
+ void sharedSubtreeIsDeduplicated() throws IOException {
+ // The same subtree referenced under two names must be repacked once and
+ // remain shared (same record id) in the output.
+ NodeBuilder shared = EMPTY_NODE.builder();
+ for (int i = 0; i < 20; i++) {
+ shared.child("c" + i).setProperty("i", (long) i);
+ }
+ RecordId sharedId = write(shared.getNodeState());
+ SegmentNodeState sharedNode = store.getReader().readNode(sharedId);
+
+ NodeBuilder root = EMPTY_NODE.builder();
+ root.setChildNode("a", sharedNode);
+ root.setChildNode("b", sharedNode);
+ RecordId rootId = write(root.getNodeState());
+
+ RecordId newRootId = repack(rootId);
+ SegmentNodeState afterRoot = store.getReader().readNode(newRootId);
+ SegmentNodeState a = (SegmentNodeState) afterRoot.getChildNode("a");
+ SegmentNodeState b = (SegmentNodeState) afterRoot.getChildNode("b");
+ assertEquals(a.getRecordId(), b.getRecordId(),
+ "shared subtree must stay shared after repacking");
+ assertEquals(root.getNodeState(), afterRoot);
+ }
+
+ @Test
+ void nodeDataModeDeduplicatesContentEqualNodes() throws IOException {
+ // Two content-identical subtrees written independently get distinct (default)
+ // stable ids. NODE_DATA mode collapses them (their translated record data is
+ // equal), while STABLE_ID mode keeps them apart (distinct stable ids).
+ RecordId id1 = write(contentEqualSubtree());
+ RecordId id2 = write(contentEqualSubtree());
+ SegmentNodeState n1 = store.getReader().readNode(id1);
+ SegmentNodeState n2 = store.getReader().readNode(id2);
+ assertNotEquals(n1.getStableId(), n2.getStableId(),
+ "independently written nodes must have distinct stable ids");
+
+ NodeBuilder root = EMPTY_NODE.builder();
+ root.setChildNode("a", n1);
+ root.setChildNode("b", n2);
+ RecordId rootId = write(root.getNodeState());
+
+ SegmentNodeState byData = repackHead(rootId, RecordRepacker.Mode.NODE_DATA);
+ assertEquals(root.getNodeState(), byData);
+ assertEquals(childRecordId(byData, "a"), childRecordId(byData, "b"),
+ "NODE_DATA must collapse content-equal nodes with distinct stable ids");
+
+ SegmentNodeState byStableId = repackHead(rootId, RecordRepacker.Mode.STABLE_ID);
+ assertEquals(root.getNodeState(), byStableId);
+ assertNotEquals(childRecordId(byStableId, "a"), childRecordId(byStableId, "b"),
+ "STABLE_ID must keep nodes with distinct stable ids apart");
+ }
+
+ @Test
+ void noDedupModeKeepsContentEqualNodesSeparate() throws IOException {
+ // A faithful, un-deduplicated repack preserves only existing physical sharing, so two
+ // independently written (content-equal) subtrees stay two separate records.
+ NodeBuilder root = EMPTY_NODE.builder();
+ root.setChildNode("a", store.getReader().readNode(write(contentEqualSubtree())));
+ root.setChildNode("b", store.getReader().readNode(write(contentEqualSubtree())));
+ RecordId rootId = write(root.getNodeState());
+
+ SegmentNodeState repacked = repackHead(rootId, RecordRepacker.Mode.NO_DEDUP);
+ assertEquals(root.getNodeState(), repacked);
+ assertNotEquals(childRecordId(repacked, "a"), childRecordId(repacked, "b"),
+ "NO_DEDUP must not collapse content-equal nodes");
+ }
+
+ @Test
+ void deepDedupModeCollapsesAcrossDistinctStableIds() throws IOException {
+ // Two independently written, content-equal subtrees with distinct stable ids.
+ NodeBuilder root = EMPTY_NODE.builder();
+ root.setChildNode("a", store.getReader().readNode(write(contentEqualSubtree())));
+ root.setChildNode("b", store.getReader().readNode(write(contentEqualSubtree())));
+ RecordId rootId = write(root.getNodeState());
+
+ // STABLE_ID materialises the (distinct) stable ids as explicit blocks and preserves them,
+ // producing a tree where a and b carry distinct explicit stable ids.
+ RecordId explicitRoot = repackHead(rootId, RecordRepacker.Mode.STABLE_ID).getRecordId();
+
+ // NODE_DATA keys nodes by data and their stable-id treatment, so the distinct explicit
+ // ids keep a and b apart.
+ SegmentNodeState byNodeData = repackHead(explicitRoot, RecordRepacker.Mode.NODE_DATA);
+ assertEquals(store.getReader().readNode(explicitRoot), byNodeData);
+ assertNotEquals(childRecordId(byNodeData, "a"), childRecordId(byNodeData, "b"),
+ "NODE_DATA must keep nodes with distinct explicit stable ids apart");
+
+ // DEEP_DEDUP ignores stable ids, so the content-equal a and b collapse to one record.
+ SegmentNodeState byDeep = repackHead(explicitRoot, RecordRepacker.Mode.DEEP_DEDUP);
+ assertEquals(store.getReader().readNode(explicitRoot), byDeep);
+ assertEquals(childRecordId(byDeep, "a"), childRecordId(byDeep, "b"),
+ "DEEP_DEDUP must collapse content-equal nodes across distinct stable ids");
+ }
+
+ private static NodeState contentEqualSubtree() {
+ NodeBuilder builder = EMPTY_NODE.builder();
+ builder.setProperty("p", "v");
+ builder.child("x").setProperty("i", 1L);
+ return builder.getNodeState();
+ }
+
+ private SegmentNodeState repackHead(RecordId rootId, RecordRepacker.Mode mode) throws IOException {
+ return repackHead(rootId, mode, 1);
+ }
+
+ private SegmentNodeState repackHead(RecordId rootId, RecordRepacker.Mode mode, int concurrency)
+ throws IOException {
+ return repackHead(rootId, mode, concurrency, false);
+ }
+
+ private SegmentNodeState repackHead(RecordId rootId, RecordRepacker.Mode mode, int concurrency,
+ boolean singleWriter) throws IOException {
+ GCGeneration target = rootId.getSegmentId().getGcGeneration().nextFull();
+ RecordRepacker repacker = new RecordRepacker(store, store.getReader(),
+ store.getSegmentIdProvider(), store.getBlobStore(), Segment.MEDIUM_LIMIT, target, mode,
+ 0, concurrency).withSingleWriter(singleWriter);
+ RecordId newRoot = repacker.repack(rootId);
+ assertEquals(target, newRoot.getSegmentId().getGcGeneration(),
+ "repacked root must be in the target generation");
+ return store.getReader().readNode(newRoot);
+ }
+
+ @Test
+ void parallelRepackPreservesContent() throws IOException {
+ // A store with enough branches to split into many parallel subtrees, whose leaves share
+ // content across branches so cross-thread deduplication is exercised too.
+ NodeBuilder root = EMPTY_NODE.builder();
+ for (int i = 0; i < 200; i++) {
+ NodeBuilder branch = root.child("branch" + i);
+ branch.setProperty("i", (long) i);
+ for (int j = 0; j < 20; j++) {
+ branch.child("leaf" + j).setProperty("v", "value-" + (j % 5));
+ }
+ }
+ RecordId rootId = write(root.getNodeState());
+ NodeState expected = root.getNodeState();
+
+ SegmentNodeState serial = repackHead(rootId, RecordRepacker.Mode.NODE_DATA, 1);
+ SegmentNodeState parallel = repackHead(rootId, RecordRepacker.Mode.NODE_DATA, 4);
+ SegmentNodeState parallelSingleWriter =
+ repackHead(rootId, RecordRepacker.Mode.NODE_DATA, 4, true);
+
+ assertSameContent("/", expected, serial);
+ assertSameContent("/", expected, parallel);
+ assertSameContent("/", expected, parallelSingleWriter);
+ }
+
+ @Test
+ void indexFirstRepackPreservesContent() throws IOException {
+ // A tree with an "oak:index"-like subtree (its leaves share content with the rest of the
+ // tree, so the two-stage repack exercises cross-stage deduplication) plus other branches.
+ NodeBuilder root = EMPTY_NODE.builder();
+ NodeBuilder index = root.child("oak:index");
+ for (int i = 0; i < 50; i++) {
+ index.child("idx" + i).setProperty("match", "value-" + (i % 5));
+ }
+ for (int i = 0; i < 100; i++) {
+ NodeBuilder branch = root.child("branch" + i);
+ for (int j = 0; j < 10; j++) {
+ branch.child("leaf" + j).setProperty("match", "value-" + (j % 5));
+ }
+ }
+ RecordId rootId = write(root.getNodeState());
+ NodeState expected = root.getNodeState();
+ RecordId indexId = ((SegmentNodeState) store.getReader().readNode(rootId)
+ .getChildNode("oak:index")).getRecordId();
+ GCGeneration target = rootId.getSegmentId().getGcGeneration().nextFull();
+
+ // Two-stage index-first (/oak:index repacked first, then the rest), serial and parallel;
+ // both must round-trip content.
+ for (int concurrency : new int[] {1, 4}) {
+ RecordRepacker repacker = new RecordRepacker(store, store.getReader(),
+ store.getSegmentIdProvider(), store.getBlobStore(), Segment.MEDIUM_LIMIT, target,
+ RecordRepacker.Mode.NODE_DATA, 0, concurrency).withIndexFirst(indexId);
+ SegmentNodeState repacked = store.getReader().readNode(repacker.repack(rootId));
+ assertSameContent("/", expected, repacked);
+ }
+ }
+
+ @Test
+ void multiStageRepackPreservesContent() throws IOException {
+ // Nested stages mirroring the real superroot: / -> root -> oak:index. Repacking with the ordered
+ // pre-stages [oak:index, root] (each a barrier) before the whole superroot must round-trip.
+ NodeBuilder superRoot = EMPTY_NODE.builder();
+ NodeBuilder contentRoot = superRoot.child("root");
+ NodeBuilder index = contentRoot.child("oak:index");
+ for (int i = 0; i < 50; i++) {
+ index.child("idx" + i).setProperty("match", "value-" + (i % 5));
+ }
+ for (int i = 0; i < 100; i++) {
+ NodeBuilder branch = contentRoot.child("branch" + i);
+ for (int j = 0; j < 10; j++) {
+ branch.child("leaf" + j).setProperty("match", "value-" + (j % 5));
+ }
+ }
+ RecordId superRootId = write(superRoot.getNodeState());
+ NodeState expected = superRoot.getNodeState();
+ SegmentNodeState superHead = store.getReader().readNode(superRootId);
+ RecordId contentRootId = ((SegmentNodeState) superHead.getChildNode("root")).getRecordId();
+ RecordId indexId = ((SegmentNodeState) superHead.getChildNode("root")
+ .getChildNode("oak:index")).getRecordId();
+ GCGeneration target = superRootId.getSegmentId().getGcGeneration().nextFull();
+
+ for (int concurrency : new int[] {1, 4}) {
+ RecordRepacker repacker = new RecordRepacker(store, store.getReader(),
+ store.getSegmentIdProvider(), store.getBlobStore(), Segment.MEDIUM_LIMIT, target,
+ RecordRepacker.Mode.NODE_DATA, 0, concurrency)
+ .withStages(List.of(indexId, contentRootId));
+ SegmentNodeState repacked = store.getReader().readNode(repacker.repack(superRootId));
+ assertSameContent("/", expected, repacked);
+ }
+ }
+
+ @Test
+ void windowDedupPreservesContent() throws IOException {
+ // The bounded recency window (withDedupWindow) deduplicates only content still in the window and
+ // emits the rest fresh; it must round-trip content at any window size and concurrency - including
+ // a window of 1 (almost everything a miss) and a window larger than the store (≈ exact dedup).
+ NodeBuilder root = EMPTY_NODE.builder();
+ for (int i = 0; i < 200; i++) {
+ NodeBuilder branch = root.child("branch" + i);
+ branch.setProperty("i", (long) i);
+ for (int j = 0; j < 20; j++) {
+ // Repeated shared strings/subtrees (dedup targets) plus per-branch unique ones.
+ branch.child("leaf" + j).setProperty("v", "shared-" + (j % 5));
+ branch.child("leaf" + j).setProperty("u", "unique-" + i + "-" + j);
+ }
+ }
+ RecordId rootId = write(root.getNodeState());
+ NodeState expected = root.getNodeState();
+ GCGeneration target = rootId.getSegmentId().getGcGeneration().nextFull();
+
+ for (int window : new int[] {1, 16, 1_000_000}) {
+ for (int concurrency : new int[] {1, 4}) {
+ RecordRepacker repacker = new RecordRepacker(store, store.getReader(),
+ store.getSegmentIdProvider(), store.getBlobStore(), Segment.MEDIUM_LIMIT, target,
+ RecordRepacker.Mode.NODE_DATA, 0, concurrency).withDedupWindow(window);
+ SegmentNodeState repacked = store.getReader().readNode(repacker.repack(rootId));
+ assertSameContent("/", expected, repacked);
+ assertTrue(repacker.getDedupWindowMisses() > 0,
+ "window cache must record emitted (missed) records");
+ }
+ }
+ }
+
+ @Test
+ void dedupCacheImplsPreserveContent() throws IOException {
+ // Every bounded-dedup implementation - the off-heap memory-mapped window (MMAP_WINDOW) and the
+ // in-heap window (HEAP_WINDOW) - must round-trip content at a small size and any concurrency.
+ NodeBuilder root = EMPTY_NODE.builder();
+ for (int i = 0; i < 200; i++) {
+ NodeBuilder branch = root.child("branch" + i);
+ branch.setProperty("i", (long) i);
+ for (int j = 0; j < 20; j++) {
+ branch.child("leaf" + j).setProperty("v", "shared-" + (j % 5));
+ branch.child("leaf" + j).setProperty("u", "unique-" + i + "-" + j);
+ }
+ }
+ RecordId rootId = write(root.getNodeState());
+ NodeState expected = root.getNodeState();
+ GCGeneration target = rootId.getSegmentId().getGcGeneration().nextFull();
+
+ for (RecordRepacker.DedupCacheImpl impl : RecordRepacker.DedupCacheImpl.values()) {
+ for (int concurrency : new int[] {1, 4}) {
+ RecordRepacker repacker = new RecordRepacker(store, store.getReader(),
+ store.getSegmentIdProvider(), store.getBlobStore(), Segment.MEDIUM_LIMIT, target,
+ RecordRepacker.Mode.NODE_DATA, 0, concurrency)
+ .withDedupWindow(16) // force the bounded path (expectedRecords is 0 here)
+ .withDedupCacheImpl(impl);
+ SegmentNodeState repacked = store.getReader().readNode(repacker.repack(rootId));
+ assertSameContent("/", expected, repacked);
+ assertTrue(repacker.getDedupWindowMisses() > 0,
+ impl + " cache must record emitted (missed) records");
+ }
+ }
+ }
+
+ private static RecordId childRecordId(SegmentNodeState node, String name) {
+ return ((SegmentNodeState) node.getChildNode(name)).getRecordId();
+ }
+
+ /**
+ * Repack a copy of a real, on-disk sample segment store and assert that the
+ * repacked head is content-identical to the original head. Skipped when the
+ * sample store is not present (e.g. on CI).
+ */
+ @Test
+ void repackSampleStore(@TempDir Path tempDir) throws Exception {
+ File sample = getSampleStore();
+
+ File copy = tempDir.resolve("segmentstore").toFile();
+ copySegmentStore(sample, copy);
+
+ try (FileStore fileStore = fileStoreBuilder(copy).build()) {
+ SegmentReader reader = fileStore.getReader();
+ SegmentNodeState originalHead = fileStore.getHead();
+ RecordId originalRoot = originalHead.getRecordId();
+
+ GCGeneration target = originalRoot.getSegmentId().getGcGeneration().nextFull();
+ RecordRepacker repacker = new RecordRepacker(fileStore, reader,
+ fileStore.getSegmentIdProvider(), fileStore.getBlobStore(),
+ fileStore.getBinariesInlineThreshold(), target);
+
+ RecordId repackedRoot = repacker.repack(originalRoot);
+
+ assertNotEquals(originalRoot, repackedRoot, "repacking must move the root");
+ assertEquals(target, repackedRoot.getSegmentId().getGcGeneration(),
+ "repacked root must be in the target generation");
+
+ SegmentNodeState repackedHead = reader.readNode(repackedRoot);
+ assertEquals(originalHead.getStableId(), repackedHead.getStableId(),
+ "stable id of the head must be preserved");
+ // Deep content comparison that bypasses SegmentNodeState's stable-id fast path.
+ assertSameContent("/", originalHead, repackedHead);
+ }
+ }
+
+ private static @NotNull File getSampleStore() {
+ assumeFalse(SAMPLE_STORE == null, "no sample segment store path configured");
+ File sample = new File(SAMPLE_STORE);
+ assumeTrue(sample.isDirectory(), "sample segment store not present at " + sample.getAbsolutePath());
+ return sample;
+ }
+
+ /**
+ * Analyse which nodes the record-level content deduplication ({@code NODE_DATA} /
+ * {@code DEEP_DEDUP}) collapses. Every node in the sample store's content tree is grouped by
+ * its <em>content signature</em>: recursively, its property names/types/values plus its child
+ * names and their content signatures. Two nodes share a signature iff they translate to the
+ * same repacked record (ignoring stable ids) — exactly the key those modes dedupe on. The
+ * largest clusters are reported with a few example repository paths and the content rendered as
+ * JSON, so we can see what kind of content is duplicated. Skipped when the sample store is
+ * absent.
+ */
+ @Test
+ void analyzeDedupClusters() throws Exception {
+ File sample = getSampleStore();
+
+ try (ReadOnlyFileStore store = fileStoreBuilder(sample).buildReadOnly()) {
+ SegmentNodeState superRoot = store.getReader().readNode(store.getRevisions().getHead());
+ NodeState root = superRoot.getChildNode("root");
+ assertTrue(root.exists(), "content root not found under the superroot");
+
+ MessageDigest digest = MessageDigest.getInstance("SHA-256");
+
+ // Pass 1: count nodes per content signature.
+ Map<String, long[]> counts = new HashMap<>();
+ walk(root, "/", digest, (sig, node, path) ->
+ counts.computeIfAbsent(sig, k -> new long[1])[0]++);
+
+ long totalNodes = 0;
+ long duplicatedNodes = 0;
+ long recordsSaved = 0;
+ for (long[] c : counts.values()) {
+ totalNodes += c[0];
+ if (c[0] > 1) {
+ duplicatedNodes += c[0];
+ recordsSaved += c[0] - 1;
+ }
+ }
+
+ // Top clusters by node records saved (copies - 1).
+ List<Map.Entry<String, long[]>> top = new ArrayList<>();
+ for (Map.Entry<String, long[]> e : counts.entrySet()) {
+ if (e.getValue()[0] > 1) {
+ top.add(e);
+ }
+ }
+ top.sort(Comparator.comparingLong(
+ (Map.Entry<String, long[]> e) -> e.getValue()[0]).reversed());
+ int topN = Math.min(25, top.size());
+ Set<String> topHashes = new HashSet<>();
+ for (int i = 0; i < topN; i++) {
+ topHashes.add(top.get(i).getKey());
+ }
+
+ // Pass 2: collect a few example paths and a representative node per top cluster.
+ Map<String, List<String>> examples = new HashMap<>();
+ Map<String, NodeState> repr = new HashMap<>();
+ walk(root, "/", digest, (sig, node, path) -> {
+ if (topHashes.contains(sig)) {
+ List<String> paths = examples.computeIfAbsent(sig, k -> new ArrayList<>());
+ if (paths.size() < 5) {
+ paths.add(path);
+ }
+ repr.putIfAbsent(sig, node);
+ }
+ });
+
+ System.out.printf("%n=== Node content-dedup analysis (%s) ===%n", sample.getAbsolutePath());
+ System.out.printf("content-tree nodes: %,d%n", totalNodes);
+ System.out.printf("distinct content shapes: %,d%n", counts.size());
+ System.out.printf("nodes in duplicated clusters: %,d (%.1f%%)%n",
+ duplicatedNodes, pct(duplicatedNodes, totalNodes));
+ System.out.printf("node records saved by dedup: %,d (%.1f%%)%n",
+ recordsSaved, pct(recordsSaved, totalNodes));
+
+ System.out.printf("%n--- top %d content clusters (by node records saved) ---%n", topN);
+ for (int i = 0; i < topN; i++) {
+ Map.Entry<String, long[]> e = top.get(i);
+ long copies = e.getValue()[0];
+ NodeState example = repr.get(e.getKey());
+ System.out.printf("%n[%d] copies=%,d saved=%,d subtreeHeight=%d%n",
+ i + 1, copies, copies - 1, subtreeHeight(example, 32));
+ System.out.println(" example paths:");
+ for (String p : examples.getOrDefault(e.getKey(), List.of())) {
+ System.out.println(" " + p);
+ }
+ System.out.println(" content:");
+ System.out.println(indent(toJson(example, 4, 20), " "));
+ }
+ }
+ }
+
+ /**
+ * Dump the full record tree of a few representative nodes from the sample store so the
+ * NodeState → segment-record mapping is visible: the node record's id slots (stable id,
+ * template, child/child-map, property-values list), the shared template record (primary type /
+ * mixins / child name / property-name list) and the string/value/list/map records they
+ * reference. Skipped when the sample store is absent.
+ */
+ @Test
+ void dumpRecordTree() throws Exception {
+ File sample = getSampleStore();
+
+ try (ReadOnlyFileStore store = fileStoreBuilder(sample).buildReadOnly()) {
+ SegmentReader reader = store.getReader();
+ NodeState root = reader.readNode(store.getRevisions().getHead()).getChildNode("root");
+
+ dumpNode(reader, "empty rep:versionStorage bucket (leaf, no properties/children)",
+ root, "jcr:system", "jcr:versionStorage", "00", "00");
+ dumpNode(reader, "rep:system (many children -> child map)",
+ root, "jcr:system");
+ dumpNode(reader, "rep:Permissions entry (leaf with a single-valued and a multi-valued property)",
+ root, "jcr:system", "rep:permissionStore", "crx.default",
+ "analytics-administrators", "941565749", "0");
+ }
+ }
+
+ private static void dumpNode(SegmentReader reader, String label, NodeState root, String... path) {
+ NodeState node = root;
+ for (String name : path) {
+ node = node.getChildNode(name);
+ }
+ String p = "/" + String.join("/", path);
+ if (!node.exists() || !(node instanceof SegmentNodeState)) {
+ System.out.printf("%n### %s%n# path %s not found in this sample store%n", label, p);
+ return;
+ }
+ RecordId id = ((SegmentNodeState) node).getRecordId();
+ System.out.printf("%n### %s%n# path: %s%n# root record: %s%n%s",
+ label, p, id, new RecordTreePrinter(reader).dump(id));
+ }
+
+ /**
+ * Count how many records pertain to the {@code /oak:index} subtree, with a per-index breakdown
+ * (one row per direct child of {@code /oak:index}). "Record" here means a distinct data-segment
+ * record reachable by reference: node + stable-id block, template, map (leaf/branch/diff), list,
+ * bucket, string and inlined blob-header records; bulk binary block data is shared and excluded,
+ * matching the "records" metric used elsewhere.
+ * <p>
+ * Three views are reported, because records are shared:
+ * <ul>
+ * <li><b>per-child reachable</b> — distinct records reachable from that index child (fresh
+ * dedup per child). Shared records (templates, strings, …) are counted in every child that
+ * reaches them, so these rows overlap and sum to more than the union.</li>
+ * <li><b>/oak:index union</b> — distinct records reachable from {@code /oak:index} as a whole.</li>
+ * <li><b>/oak:index exclusive</b> — records reachable from {@code /oak:index} but from no other
+ * part of {@code /root}; i.e. what would become unreferenced if {@code /oak:index} were
+ * removed. The truest "records that pertain to the index".</li>
+ * </ul>
+ * The denominator is the record count reachable from {@code /root} (the live content tree).
+ * Skipped when the sample store is absent.
+ */
+ @Test
+ void analyzeIndexRecordCounts() throws Exception {
+ File sample = getSampleStore();
+
+ try (ReadOnlyFileStore store = fileStoreBuilder(sample).buildReadOnly()) {
+ SegmentReader reader = store.getReader();
+ SegmentNodeState superRoot = reader.readNode(store.getRevisions().getHead());
+ NodeState root = superRoot.getChildNode("root");
+ assertTrue(root.exists(), "content root not found under the superroot");
+ NodeState index = root.getChildNode("oak:index");
+ assertTrue(index.exists(), "/oak:index not found under /root");
+ RecordId rootId = ((SegmentNodeState) root).getRecordId();
+ RecordId indexId = ((SegmentNodeState) index).getRecordId();
+
+ List<String> names = new ArrayList<>();
+ for (ChildNodeEntry e : index.getChildNodeEntries()) {
+ names.add(e.getName());
+ }
+ Collections.sort(names);
+
+ // (1) Per-child reachable (fresh dedup each) — the requested breakdown.
+ System.out.printf("%n=== records per direct child of /oak:index (reachable, deduped per child) ===%n");
+ System.out.printf("%-48s %14s%n", "index (direct child of /oak:index)", "records");
+ System.out.printf("%-48s %14s%n",
+ "------------------------------------------------", "--------------");
+ long sumPerChild = 0;
+ for (String name : names) {
+ RecordId childId = ((SegmentNodeState) index.getChildNode(name)).getRecordId();
+ RecordCounter c = new RecordCounter(reader);
+ c.count(childId);
+ sumPerChild += c.total();
+ System.out.printf("%-48s %,14d%n", name, c.total());
+ }
+ System.out.printf("%-48s %,14d%n",
+ "(sum of rows — overlaps due to shared records)", sumPerChild);
+
+ // (2) /oak:index union reachable, with a record-type breakdown.
+ RecordCounter unionCounter = new RecordCounter(reader);
+ unionCounter.count(indexId);
+ long indexUnion = unionCounter.total();
+
+ // (3) /oak:index exclusive + /root denominator, in one shared walk (non-index first, so
+ // any record shared with the rest of /root is seen before the /oak:index phase).
+ RecordCounter all = new RecordCounter(reader);
+ for (ChildNodeEntry e : root.getChildNodeEntries()) {
+ if (!"oak:index".equals(e.getName())) {
+ all.count(((SegmentNodeState) e.getNodeState()).getRecordId());
+ }
+ }
+ long nonIndexReachable = all.total();
+ all.count(indexId);
+ long indexExclusive = all.total() - nonIndexReachable;
+ all.count(rootId); // fold in /root's own node/template/child-map records
+ long rootReachable = all.total();
+
+ System.out.printf("%n=== /oak:index aggregate ===%n");
+ System.out.printf("%-48s %,14d%n", "/oak:index union (reachable)", indexUnion);
+ System.out.printf(" %s%n", unionCounter.breakdown());
+ System.out.printf("%-48s %,14d%n", "/oak:index exclusive (only via /oak:index)", indexExclusive);
+ System.out.printf("%-48s %,14d%n", "/root reachable (denominator)", rootReachable);
+ System.out.printf("%-48s %13.1f%%%n", "union / root", 100.0 * indexUnion / rootReachable);
+ System.out.printf("%-48s %13.1f%%%n", "exclusive / root", 100.0 * indexExclusive / rootReachable);
+ }
+ }
+
+ /**
+ * Estimate how much larger the store would be if template records were not shared. Oak writes
+ * one {@link RecordType#TEMPLATE template} record per distinct node "shape" (primary type,
+ * mixins, child-node arity and the set of property names/types), and every node of that shape
+ * references it. This walks the whole reachable record graph (content root plus checkpoints),
+ * measures each distinct template's own record size and counts how many node records reference
+ * it, then reports the extra bytes that would be written if each node instead carried a private
+ * copy of its template. Only the template record payload is counted: the strings and
+ * property-name list a template references have their own deduplication and would stay shared.
+ * The figure is a lower bound on the on-disk increase (record alignment and record-table
+ * overhead are not counted). Skipped when the sample store is absent.
+ */
+ @Test
+ void analyzeTemplateReuse() throws Exception {
+ File sample = getSampleStore();
+
+ try (ReadOnlyFileStore store = fileStoreBuilder(sample).buildReadOnly()) {
+ SegmentReader reader = store.getReader();
+ RecordId head = store.getRevisions().getHead();
+
+ TemplateReuseAnalyser analyser = new TemplateReuseAnalyser(reader);
+ analyser.analyseNode(head);
+
+ Map<RecordId, long[]> templates = analyser.templates;
+ long distinctTemplates = templates.size();
+ long nodeRecords = analyser.getNodeCount();
+ long currentTemplateBytes = 0;
+ long noDedupTemplateBytes = 0;
+ long refSum = 0;
+ for (long[] t : templates.values()) {
+ currentTemplateBytes += t[0];
+ noDedupTemplateBytes += t[0] * t[1];
+ refSum += t[1];
+ }
+ long extraBytes = noDedupTemplateBytes - currentTemplateBytes;
+
+ // Cross-checks against the RecordUsageAnalyser accounting.
+ assertEquals(nodeRecords, refSum, "every node record references exactly one template");
+ assertEquals(analyser.getTemplateSize(), currentTemplateBytes,
+ "per-template sizes must sum to the total template bytes");
+ assertTrue(extraBytes > 0, "templates are shared, so removing sharing must add bytes");
+
+ long contentRecordBytes = analyser.getMapSize() + analyser.getListSize()
+ + analyser.getValueSize() + analyser.getTemplateSize() + analyser.getNodeSize();
+ long storeOnDisk = directorySize(sample);
+
+ System.out.printf("%n=== Template reuse analysis (%s) ===%n", sample.getAbsolutePath());
+ System.out.printf("reachable node records: %,d%n", nodeRecords);
+ System.out.printf("distinct templates: %,d%n", distinctTemplates);
+ System.out.printf("avg node records per template: %.1f%n",
+ distinctTemplates == 0 ? 0.0 : (double) nodeRecords / distinctTemplates);
+ System.out.printf("current template bytes (shared): %s%n", mb(currentTemplateBytes));
+ System.out.printf("template bytes if not deduplicated: %s%n", mb(noDedupTemplateBytes));
+ System.out.printf("extra bytes from duplication: %s%n", mb(extraBytes));
+ System.out.printf(" = %.0fx the current template bytes%n",
+ currentTemplateBytes == 0 ? 0.0 : (double) noDedupTemplateBytes / currentTemplateBytes);
+ System.out.printf(" = +%.1f%% of content-record bytes (%s of node/template/map/list/value records)%n",
+ pct(extraBytes, contentRecordBytes), mb(contentRecordBytes));
+ System.out.printf(" = +%.1f%% of the on-disk store (%s, incl. bulk binaries)%n",
+ pct(extraBytes, storeOnDisk), mb(storeOnDisk));
+
+ // Top templates by the bytes their duplication would add (size x (refs - 1)).
+ List<Map.Entry<RecordId, long[]>> top = new ArrayList<>(templates.entrySet());
+ top.sort(Comparator.comparingLong(
+ (Map.Entry<RecordId, long[]> e) -> e.getValue()[0] * (e.getValue()[1] - 1)).reversed());
+ int topN = Math.min(15, top.size());
+ System.out.printf("%n--- top %d templates by duplication cost ---%n", topN);
+ for (int i = 0; i < topN; i++) {
+ Map.Entry<RecordId, long[]> e = top.get(i);
+ long size = e.getValue()[0];
+ long refs = e.getValue()[1];
+ String shape = reader.readTemplate(e.getKey()).toString();
+ if (shape.length() > 160) {
+ shape = shape.substring(0, 160) + "…";
+ }
+ System.out.printf("%n[%d] template size=%dB nodes=%,d extra=%s%n",
+ i + 1, size, refs, mb(size * (refs - 1)));
+ System.out.println(" " + shape);
+ }
+ }
+ }
+
+ /**
+ * Measure how much content-deduplicating child-node maps would save. Child maps are the one
+ * structural record type the repacker does not yet content-deduplicate; because a fresh (not
+ * shared) map id also prevents the multi-child node that references it from deduplicating,
+ * closing this gap can save both the duplicate map records and the multi-child nodes (and their
+ * ancestors) whose deduplication the distinct map ids blocked. The same self-marker store is
+ * repacked in {@code DEEP_DEDUP} mode with map deduplication off and on, and the deduplicated
+ * output is analysed per record type with {@link RecordUsageAnalyser}. Skipped when the sample
+ * store is absent. Run with a generous heap (e.g. {@code -Dtest.opts.memory=-Xmx4g}).
+ */
+ @Test
+ void measureMapDedup(@TempDir Path tempDir) throws Exception {
+ File sample = getSampleStore();
+
+ File prepared = tempDir.resolve("prepared").toFile();
+ prepareStoreWithGarbage(sample, prepared);
+
+ Usage off = repackAndMeasure(tempDir.resolve("map-off").toFile(), prepared, false);
+ Usage on = repackAndMeasure(tempDir.resolve("map-on").toFile(), prepared, true);
+
+ System.out.printf("%n=== Child-map deduplication effect (DEEP_DEDUP, %s) ===%n",
+ prepared.getAbsolutePath());
+ printUsageRow("nodes", off.nodeCount, off.nodeSize, on.nodeCount, on.nodeSize);
+ printUsageRow("templates", off.templateCount, off.templateSize, on.templateCount, on.templateSize);
+ printUsageRow("maps", off.mapCount, off.mapSize, on.mapCount, on.mapSize);
+ printUsageRow("lists", off.listCount, off.listSize, on.listCount, on.listSize);
+ printUsageRow("values", -1, off.valueSize, -1, on.valueSize);
+
+ long offTotal = off.total();
+ long onTotal = on.total();
+ System.out.printf("%nTOTAL content bytes: off=%s on=%s saved=%s (%.1f%%)%n",
+ mb(offTotal), mb(onTotal), mb(offTotal - onTotal), pct(offTotal - onTotal, offTotal));
+ System.out.printf("structural records (node+template+map+list): off=%,d on=%,d saved=%,d (%.1f%%)%n",
+ off.records(), on.records(), off.records() - on.records(),
+ pct(off.records() - on.records(), off.records()));
+
+ assertTrue(onTotal <= offTotal, "map dedup must not increase the content-record bytes");
+ assertTrue(on.mapCount < off.mapCount, "map dedup must reduce the number of map records");
+ assertTrue(on.nodeCount < off.nodeCount,
+ "map dedup should unlock deduplication of the multi-child nodes referencing the maps");
+ }
+
+ /** Per-record-type usage of a repacked (deduplicated) output, from {@link RecordUsageAnalyser}. */
+ private static final class Usage {
+ final long nodeCount, nodeSize, templateCount, templateSize, mapCount, mapSize,
+ listCount, listSize, valueSize;
+
+ Usage(RecordUsageAnalyser a) {
+ nodeCount = a.getNodeCount();
+ nodeSize = a.getNodeSize();
+ templateCount = a.getTemplateCount();
+ templateSize = a.getTemplateSize();
+ mapCount = a.getMapCount();
+ mapSize = a.getMapSize();
+ listCount = a.getListCount();
+ listSize = a.getListSize();
+ valueSize = a.getValueSize();
+ }
+
+ long total() {
+ return nodeSize + templateSize + mapSize + listSize + valueSize;
+ }
+
+ long records() {
+ return nodeCount + templateCount + mapCount + listCount;
+ }
+ }
+
+ /** Repack a fresh copy of {@code prepared} in DEEP_DEDUP mode and analyse the deduplicated output. */
+ private Usage repackAndMeasure(File dest, File prepared, boolean mapDedup) throws Exception {
+ copySegmentStore(prepared, dest);
+ try (FileStore fs = fileStoreBuilder(dest).build()) {
+ SegmentNodeState originalHead = fs.getHead();
+ RecordId root = originalHead.getRecordId();
+ GCGeneration target = root.getSegmentId().getGcGeneration().nextFull();
+ RecordRepacker repacker = new RecordRepacker(fs, fs.getReader(),
+ fs.getSegmentIdProvider(), fs.getBlobStore(),
+ fs.getBinariesInlineThreshold(), target, RecordRepacker.Mode.DEEP_DEDUP)
+ .withMapDeduplication(mapDedup);
+ RecordId repacked = repacker.repack(root);
+ fs.flush();
+ // Content must be preserved regardless of map deduplication (bypasses the stable-id
+ // fast path, which DEEP_DEDUP does not preserve).
+ assertSameContent("/", originalHead, fs.getReader().readNode(repacked));
+ RecordUsageAnalyser usage = new RecordUsageAnalyser(fs.getReader());
+ usage.analyseNode(repacked);
+ return new Usage(usage);
+ }
+ }
+
+ private static void printUsageRow(String type, long offCount, long offBytes, long onCount, long onBytes) {
+ String off = offCount < 0 ? "" : String.format("%,d recs, ", offCount);
+ String on = onCount < 0 ? "" : String.format("%,d recs, ", onCount);
+ System.out.printf(" %-10s off: %s%-16s on: %s%-16s saved %s%n",
+ type, off, mb(offBytes), on, mb(onBytes), mb(offBytes - onBytes));
+ }
+
+ /** Visitor invoked once per node (post-order) with its content signature and repository path. */
+ @FunctionalInterface
+ private interface NodeVisitor {
+ void visit(String signature, NodeState node, String path);
+ }
+
+ /**
+ * Post-order walk that computes each node's content signature (hashing property
+ * names/types/values and, recursively, child names and their signatures) and invokes
+ * {@code visitor}. A node's own name is deliberately excluded — it is stored in the parent,
+ * not the node record — so content-equal nodes under different names share a signature,
+ * matching record-level dedup. Returns the visited node's signature.
+ */
+ private static String walk(NodeState node, String path, MessageDigest digest, NodeVisitor visitor) {
+ StringBuilder sb = new StringBuilder();
+ List<PropertyState> props = new ArrayList<>();
+ node.getProperties().forEach(props::add);
+ props.sort(Comparator.comparing(PropertyState::getName));
+ for (PropertyState p : props) {
+ sb.append('P').append(p.getName()).append('#').append(p.getType().tag()).append('=');
+ appendSignatureValues(sb, p);
+ sb.append('\u0002');
+ }
+ List<ChildNodeEntry> children = new ArrayList<>();
+ node.getChildNodeEntries().forEach(children::add);
+ children.sort(Comparator.comparing(ChildNodeEntry::getName));
+ for (ChildNodeEntry c : children) {
+ String childPath = path.endsWith("/") ? path + c.getName() : path + "/" + c.getName();
+ String childSig = walk(c.getNodeState(), childPath, digest, visitor);
+ sb.append('C').append(c.getName()).append('=').append(childSig).append('\u0002');
+ }
+ String sig = toHex(digest.digest(sb.toString().getBytes(StandardCharsets.UTF_8)), 16);
+ visitor.visit(sig, node, path);
+ return sig;
+ }
+
+ private static void appendSignatureValues(StringBuilder sb, PropertyState p) {
+ Type<?> base = p.getType().isArray() ? p.getType().getBaseType() : p.getType();
+ for (int i = 0; i < p.count(); i++) {
+ if (base == BINARY) {
+ sb.append("bin:").append(p.getValue(BINARY, i).length());
+ } else {
+ sb.append(p.getValue(Type.STRING, i));
+ }
+ sb.append('\u0001');
+ }
+ }
+
+ private static String toHex(byte[] bytes, int len) {
+ StringBuilder sb = new StringBuilder(len * 2);
+ for (int i = 0; i < len; i++) {
+ sb.append(Character.forDigit((bytes[i] >> 4) & 0xf, 16));
+ sb.append(Character.forDigit(bytes[i] & 0xf, 16));
+ }
+ return sb.toString();
+ }
+
+ private static double pct(long part, long total) {
+ return total == 0 ? 0.0 : 100.0 * part / total;
+ }
+
+ private static String mb(long bytes) {
+ return String.format("%,d B (%.1f MB)", bytes, bytes / (1024.0 * 1024.0));
+ }
+
+ private static int subtreeHeight(NodeState node, int limit) {
+ if (limit <= 0) {
+ return 0;
+ }
+ int h = 0;
+ for (ChildNodeEntry c : node.getChildNodeEntries()) {
+ h = Math.max(h, 1 + subtreeHeight(c.getNodeState(), limit - 1));
+ }
+ return h;
+ }
+
+ /** Render a node's content as indented JSON, bounded to {@code maxDepth} and {@code maxChildren}. */
+ private static String toJson(NodeState node, int maxDepth, int maxChildren) {
+ StringBuilder sb = new StringBuilder();
+ appendJson(node, sb, 0, maxDepth, maxChildren);
+ return sb.toString();
+ }
+
+ private static void appendJson(NodeState node, StringBuilder sb, int depth, int maxDepth, int maxChildren) {
+ String pad = " ".repeat(depth + 1);
+ String closePad = " ".repeat(depth);
+ sb.append('{');
+ boolean first = true;
+ List<PropertyState> props = new ArrayList<>();
+ node.getProperties().forEach(props::add);
+ props.sort(Comparator.comparing(PropertyState::getName));
+ for (PropertyState p : props) {
+ sb.append(first ? "\n" : ",\n").append(pad).append(jsonString(p.getName())).append(": ");
+ appendJsonValue(sb, p);
+ first = false;
+ }
+ List<ChildNodeEntry> children = new ArrayList<>();
+ node.getChildNodeEntries().forEach(children::add);
+ children.sort(Comparator.comparing(ChildNodeEntry::getName));
+ int shown = 0;
+ for (ChildNodeEntry c : children) {
+ if (shown >= maxChildren) {
+ sb.append(",\n").append(pad).append(jsonString("…"))
+ .append(": ").append(jsonString("(" + (children.size() - shown) + " more children)"));
+ break;
+ }
+ sb.append(first ? "\n" : ",\n").append(pad).append(jsonString(c.getName())).append(": ");
+ if (depth + 1 >= maxDepth) {
+ sb.append(jsonString("{…}"));
+ } else {
+ appendJson(c.getNodeState(), sb, depth + 1, maxDepth, maxChildren);
+ }
+ first = false;
+ shown++;
+ }
+ if (!first) {
+ sb.append('\n').append(closePad);
+ }
+ sb.append('}');
+ }
+
+ private static void appendJsonValue(StringBuilder sb, PropertyState p) {
+ if (p.isArray()) {
+ sb.append('[');
+ for (int i = 0; i < p.count(); i++) {
+ sb.append(i == 0 ? "" : ", ").append(jsonScalar(p, i));
+ }
+ sb.append(']');
+ } else {
+ sb.append(jsonScalar(p, 0));
+ }
+ }
+
+ private static String jsonScalar(PropertyState p, int i) {
+ Type<?> base = p.getType().isArray() ? p.getType().getBaseType() : p.getType();
+ if (base == BINARY) {
+ return jsonString("<binary " + p.getValue(BINARY, i).length() + " bytes>");
+ }
+ String v = p.getValue(Type.STRING, i);
+ if (v.length() > 200) {
+ v = v.substring(0, 200) + "…(" + v.length() + " chars)";
+ }
+ return jsonString(v);
+ }
+
+ private static String jsonString(String s) {
+ StringBuilder sb = new StringBuilder(s.length() + 2);
+ sb.append('"');
+ for (int i = 0; i < s.length(); i++) {
+ char ch = s.charAt(i);
+ switch (ch) {
+ case '"': sb.append("\\\""); break;
+ case '\\': sb.append("\\\\"); break;
+ case '\n': sb.append("\\n"); break;
+ case '\r': sb.append("\\r"); break;
+ case '\t': sb.append("\\t"); break;
+ default:
+ if (ch < 0x20) {
+ sb.append(String.format("\\u%04x", (int) ch));
+ } else {
+ sb.append(ch);
+ }
+ }
+ }
+ sb.append('"');
+ return sb.toString();
+ }
+
+ private static String indent(String text, String prefix) {
+ return prefix + text.replace("\n", "\n" + prefix);
+ }
+
+ /**
+ * Prints the record tree rooted at a node using {@link SegmentParser}'s type-aware, root-anchored
+ * walk. Child nodes are not expanded (they are listed by their parent's child-map entries as
+ * {@code name -> NODE <id>}), so the output shows one node's own records: the node record's id
+ * slots, its (shared) template, and the string/value/list/map records they reference.
+ */
+ private static final class RecordTreePrinter extends SegmentParser {
+ private final SegmentReader reader;
+ private final StringBuilder out = new StringBuilder();
+ private int depth;
+ private int lines;
+
+ RecordTreePrinter(SegmentReader reader) {
+ super(reader);
+ this.reader = reader;
+ }
+
+ String dump(RecordId nodeId) {
+ onNode(null, nodeId);
+ return out.toString();
+ }
+
+ private void line(String s) {
+ if (lines++ > 200) {
+ return;
+ }
+ for (int i = 0; i < depth; i++) {
+ out.append(" ");
+ }
+ out.append(s).append('\n');
+ }
+
+ @Override
+ protected void onNode(RecordId parentId, RecordId nodeId) {
+ if (parentId != null) {
+ // A child node: already listed by its parent's child-map entry; don't expand it.
+ return;
+ }
+ SegmentNodeState node = reader.readNode(nodeId);
+ RecordId stableRec = nodeId.getSegment().readRecordId(nodeId.getRecordNumber(), 0, 0);
+ line(String.format("NODE %s (childNodes=%d, properties=%d)",
+ nodeId, node.getChildNodeCount(Long.MAX_VALUE), node.getPropertyCount()));
+ depth++;
+ if (stableRec.equals(nodeId)) {
+ line("[id 0] stableId = SELF (self-reference default; no separate record)");
+ } else {
+ line(String.format("[id 0] stableId -> BLOCK %s value=%s", stableRec, node.getStableId()));
+ }
+ parseNode(nodeId);
+ depth--;
+ }
+
+ @Override
+ protected void onTemplate(RecordId parentId, RecordId templateId) {
+ Segment seg = templateId.getSegment();
+ int head = seg.readInt(templateId.getRecordNumber(), 0);
+ boolean hasPrimaryType = (head & (1 << 31)) != 0;
+ int mixins = (head >> 18) & ((1 << 10) - 1);
+ int props = head & ((1 << 18) - 1);
+ String children = (head & (1 << 29)) != 0 ? "zero"
+ : (head & (1 << 28)) != 0 ? "many" : "single";
+ line(String.format(
+ "[id 1] template -> TEMPLATE %s {primaryType=%b, mixins=%d, children=%s, properties=%d}",
+ templateId, hasPrimaryType, mixins, children, props));
+ depth++;
+ parseTemplate(templateId);
+ depth--;
+ }
+
+ @Override
+ protected void onProperty(RecordId parentId, RecordId propertyId, PropertyTemplate template) {
+ line(String.format("property \"%s\" (%s) -> %s",
+ template.getName(), template.getType(), propertyId));
+ depth++;
+ parseProperty(parentId, propertyId, template);
+ depth--;
+ }
+
+ @Override
+ protected void onString(RecordId parentId, RecordId stringId) {
+ String v = reader.readString(stringId);
+ if (v.length() > 80) {
+ v = v.substring(0, 80) + "…(" + v.length() + " chars)";
+ }
+ line(String.format("STRING %s = \"%s\"", stringId, v));
+ }
+
+ @Override
+ protected void onBlob(RecordId parentId, RecordId blobId) {
+ BlobInfo info = parseBlob(blobId);
+ line(String.format("BLOB %s type=%s size=%d", blobId, info.blobType, info.size));
+ }
+
+ @Override
+ protected void onList(RecordId parentId, RecordId listId, int count) {
+ line(String.format("LIST %s (count=%d)", listId, count));
+ }
+
+ @Override
+ protected void onMapLeaf(RecordId parentId, RecordId mapId, MapRecord map) {
+ line(String.format("MAP-LEAF %s (entries=%d) [name -> child node record]", mapId, map.size()));
+ depth++;
+ int shown = 0;
+ for (MapEntry e : map.getEntries()) {
+ if (shown++ >= 12) {
+ line(String.format("… (%d more entries)", map.size() - 12));
+ break;
+ }
+ line(String.format("\"%s\" -> NODE %s", e.getName(), e.getValue()));
+ }
+ depth--;
+ }
+
+ @Override
+ protected void onMapBranch(RecordId parentId, RecordId mapId, MapRecord map) {
+ line(String.format("MAP-BRANCH %s", mapId));
+ depth++;
+ parseMapBranch(mapId, map);
+ depth--;
+ }
+ }
+
+ /**
+ * Extends {@link RecordUsageAnalyser} to additionally record, per distinct template record, its
+ * own size ({@link SegmentParser.TemplateInfo#size}: head + reference slots + per-property type
+ * bytes, excluding the shared strings / property-name list it points at) and the number of
+ * distinct node records that reference it. The base class deduplicates records by id while
+ * walking, so each node is parsed once and {@link #onTemplate} fires exactly once per node
+ * record; the size is measured on first sight and subsequent references only bump the count.
+ */
+ private static final class TemplateReuseAnalyser extends RecordUsageAnalyser {
+ /** template record id -> [size in bytes, number of referencing node records]. */
+ private final Map<RecordId, long[]> templates = new HashMap<>();
+
+ TemplateReuseAnalyser(SegmentReader reader) {
+ super(reader);
+ }
+
+ @Override
+ protected void onTemplate(RecordId parentId, RecordId templateId) {
+ long[] entry = templates.get(templateId);
+ if (entry == null) {
+ entry = new long[] { parseTemplate(templateId).size, 0 };
+ templates.put(templateId, entry);
+ }
+ entry[1]++;
+ super.onTemplate(parentId, templateId);
+ }
+ }
+
+ /**
+ * Benchmark helper: compact a copy of the sample store with the record-level
+ * {@link RecordRepacker} and the two semantic compactors ({@link ClassicCompactor} via
+ * {@link CheckpointCompactor}, and {@link ParallelCompactor}) and print a side-by-side
+ * comparison of wall-clock time, peak heap and the size of the written (compacted)
+ * generation. Skipped when the sample store is absent.
+ * <p>
+ * This is a rough, single-JVM measurement meant for local benchmarking. Run it with a
+ * generous heap (e.g. {@code -Dtest.opts.memory=-Xmx4g}), because the semantic compactors
+ * build NodeState graphs and dedup caches that the record-level repacker avoids.
+ */
+ @Test
+ void compareWithCheckpointCompactor(@TempDir Path tempDir) throws Exception {
+ File sample = getSampleStore();
+
+ // Build the benchmark input once by copying the sample's content tree through the
+ // NodeStore API. Copying node-by-node re-writes every record from scratch, dropping
+ // the source stable ids, and intermittently deleting and re-copying already-copied
+ // branches creates garbage (an uncompacted store, as it would look before GC). All
+ // arms then run on identical raw copies of this single prepared store.
+ File prepared = tempDir.resolve("prepared").toFile();
+ prepareStoreWithGarbage(sample, prepared);
+ verifyStableIdsDropped(prepared);
+
+ // The input (pre-compaction) store's own record count and non-bulk record bytes, for a
+ // baseline row and to pre-size the repacker's alias/dedup tables so they do not rehash.
+ long[] inputStats = countStoreContent(prepared);
+ int expectedRecords = (int) Math.min(Integer.MAX_VALUE, inputStats[0]);
+
+ int runs = Integer.getInteger("oak.repack.runs", 1);
+
+ // The arm list can be driven from the command line via -Doak.repack.arms=<spec>[,<spec>...]
+ // so a split-level / thread-count sweep can be bisected across runs without recompiling. Each
+ // spec is one of:
+ // dedup:<threads>:<split>[:inexact][:full|:heap] hot-dedup (NODE_DATA, auto off-heap mmap
+ // window) repack; :full = full-dedup (exact off-heap);
+ // :heap = the in-heap window instead of the mmap window
+ // none:<threads>:<split>[:inexact] preserve (NO_DEDUP) repack (always inexact under >1 thread)
+ // checkpoint CheckpointCompactor (Classic, full up-compaction)
+ // parallel:<n> ParallelCompactor at concurrency n
+ // where <threads> 1 means serial, <split> 0 means the coupled default (concurrency*32 roots).
+ // With no property set, the default decoupling matrix below runs.
+ List<Arm> arms = new ArrayList<>();
+ String armsSpec = System.getProperty("oak.repack.arms");
+ if (armsSpec != null && !armsSpec.isBlank()) {
+ for (String spec : armsSpec.split(",")) {
+ if (!spec.isBlank()) {
+ arms.add(buildArm(spec.trim(), expectedRecords));
+ }
+ }
+ } else {
+ int[] matrixThreads = { 4, 8 };
+ int[] splitTargets = { 128, 512, 2048 };
+ for (RecordRepacker.Mode mode : new RecordRepacker.Mode[] {
+ RecordRepacker.Mode.NO_DEDUP, RecordRepacker.Mode.NODE_DATA }) {
+ String name = mode == RecordRepacker.Mode.NO_DEDUP ? "preserve" : "hot-dedup";
+ String slug = mode == RecordRepacker.Mode.NO_DEDUP ? "none" : "dedup";
+ arms.add(new Arm("RecordRepacker(" + name + ")", "repack-" + slug,
+ fs -> runRecordRepacker(fs, mode, expectedRecords)));
+ for (int c : new int[] { 8, 16 }) {
+ final int cc = c;
+ arms.add(new Arm("RecordRepacker(" + name + "|c" + cc + ")", "repack-" + slug + "-c" + cc,
+ fs -> runRecordRepacker(fs, mode, expectedRecords, cc)));
+ }
+ for (int t : matrixThreads) {
+ for (int st : splitTargets) {
+ final int tt = t;
+ final int sst = st;
+ arms.add(new Arm("RecordRepacker(" + name + "|" + tt + "/" + sst + ")",
+ "repack-" + slug + "-" + tt + "-" + sst,
+ fs -> runRecordRepacker(fs, mode, expectedRecords, tt, sst)));
+ }
+ }
+ }
+ }
+
+ System.out.printf("%n=== Sample store compaction comparison (%s), %d runs each ===%n",
+ prepared.getAbsolutePath(), runs);
+
+ // Pass-major: one full pass over every arm before starting the next pass, so aborting
+ // between passes still leaves a complete, balanced dataset (N completed passes = N runs per
+ // arm). Incremental medians (over the passes completed so far) are printed after every pass.
+ int nArms = arms.size();
+ long[][] times = new long[nArms][runs];
+ long[][] heaps = new long[nArms][runs];
+ long[][] outputs = new long[nArms][runs];
+ long[][] records = new long[nArms][runs];
+ for (int i = 0; i < runs; i++) {
+ System.out.printf("%n=== pass %d/%d ===%n", i + 1, runs);
+ for (int a = 0; a < nArms; a++) {
+ Arm arm = arms.get(a);
+ File dest = tempDir.resolve(arm.dir + "-" + i).toFile();
+ CompactionResult r = measureCompaction(arm.label, dest, prepared, arm.compaction);
+ assertTrue(r.outputBytes > 0, arm.label + " wrote no output");
+ times[a][i] = r.elapsedMillis;
+ heaps[a][i] = r.peakHeapBytes / (1024 * 1024);
+ outputs[a][i] = r.outputBytes;
+ records[a][i] = r.recordCount;
+ deleteRecursively(dest);
+ }
+ printMedians(arms, times, heaps, outputs, records, i + 1, inputStats);
+ }
+
+ // Full per-arm run breakdown once all passes have completed.
+ for (int a = 0; a < nArms; a++) {
+ printRuns(arms.get(a).label, times[a], heaps[a], outputs[a], records[a]);
+ }
+ }
+
+ /**
+ * Print the medians table over the first {@code completed} passes (incremental, abort-safe). Also
+ * appended to {@code -Doak.repack.resultFile} if set, flushed each pass, so an early abort (see
+ * the pass-major loop) still leaves the completed passes' medians on disk - surefire buffers
+ * System.out into the report file and loses it on kill.
+ */
+ private static void printMedians(List<Arm> arms, long[][] times, long[][] heaps,
+ long[][] outputs, long[][] records, int completed, long[] inputStats) {
+ StringBuilder sb = new StringBuilder();
+ sb.append(String.format("%n--- medians over %d run(s) ---%n", completed));
+ sb.append(String.format("%-40s %12s %16s %16s %14s%n",
+ "compactor", "time (ms)", "peak heap (MB)", "output (bytes)", "records"));
+ // Baseline: the input store before any compaction (non-bulk record bytes and record count,
+ // measured the same way as each arm's output; includes the garbage the compactors remove).
+ sb.append(String.format("%-40s %12s %16s %16d %14d%n",
+ "(input, pre-compaction)", "-", "-", inputStats[1], inputStats[0]));
+ for (int a = 0; a < arms.size(); a++) {
+ sb.append(String.format("%-40s %12d %16d %16d %14d%n", arms.get(a).label,
+ median(times[a], completed), median(heaps[a], completed),
+ median(outputs[a], completed), median(records[a], completed)));
+ }
+ System.out.print(sb);
+ appendResultFile(sb.toString());
+ }
+
+ /** Append {@code text} to {@code -Doak.repack.resultFile} (if set), flushed immediately so an early
+ * abort still leaves completed rows on disk (surefire buffers System.out and loses it on kill). */
+ private static void appendResultFile(String text) {
+ String resultFile = System.getProperty("oak.repack.resultFile");
+ if (resultFile == null || resultFile.isBlank()) {
+ return;
+ }
+ try {
+ Files.write(Path.of(resultFile), text.getBytes(StandardCharsets.UTF_8),
+ StandardOpenOption.CREATE, StandardOpenOption.APPEND);
+ } catch (IOException e) {
+ System.out.println("could not append to " + resultFile + ": " + e);
+ }
+ }
+
+ /** Parse one {@code -Doak.repack.arms} spec into an {@link Arm}; see {@link #compareWithCheckpointCompactor}. */
+ private Arm buildArm(String spec, int expectedRecords) {
+ String[] p = spec.split(":");
+ String kind = p[0];
+ if (kind.equals("checkpoint")) {
+ return new Arm("CheckpointCompactor", "checkpoint", this::runCheckpointCompactor);
+ }
+ if (kind.equals("parallel")) {
+ int n = Integer.parseInt(p[1]);
+ return new Arm("ParallelCompactor(" + n + ")", "parallel-" + n, fs -> runParallelCompactor(fs, n));
+ }
+ RecordRepacker.Mode mode = kind.equals("none")
+ ? RecordRepacker.Mode.NO_DEDUP : RecordRepacker.Mode.NODE_DATA;
+ int threads = p.length > 1 ? Integer.parseInt(p[1]) : 1;
+ int split = p.length > 2 ? Integer.parseInt(p[2]) : 0;
+ // Tokens after the split are flags: "inexact" (racy dedup), "single" (one synchronized writer),
+ // "full" (the exact, unbounded off-heap dedup cache instead of the default recency window),
+ // "heap" (in-heap window instead of the default off-heap mmap window), or "idx" (three-stage
+ // index-first repack: /oak:index, then /root, then / - each a parallel barrier; see
+ // RecordRepacker#withStages).
+ boolean exact = true;
+ boolean single = false;
+ boolean full = false;
+ int stages = 0;
+ RecordRepacker.DedupCacheImpl cacheImpl = RecordRepacker.DedupCacheImpl.MMAP_WINDOW;
+ for (int i = 3; i < p.length; i++) {
+ if (p[i].equals("inexact")) {
+ exact = false;
+ } else if (p[i].equals("single")) {
+ single = true;
+ } else if (p[i].equals("full")) {
+ full = true;
+ } else if (p[i].equals("heap")) {
+ cacheImpl = RecordRepacker.DedupCacheImpl.HEAP_WINDOW;
+ } else if (p[i].equals("idx")) {
+ stages = 2;
+ }
+ }
+ // full-dedup pins the exact cache (window 0); otherwise the default auto window (hot-dedup).
+ int window = full ? 0 : -1;
+ String implTag = cacheImpl == RecordRepacker.DedupCacheImpl.HEAP_WINDOW ? "-heap" : "";
+ String idxTag = stages == 2 ? "+idx" : "";
+ String name = (mode == RecordRepacker.Mode.NO_DEDUP ? "preserve"
+ : (full ? "full-dedup" : "hot-dedup" + implTag)) + idxTag;
+ String flags = (exact ? "" : "|inexact") + (single ? "|1w" : "");
+ String cfg = threads <= 1 ? ""
+ : "|" + threads + "/" + (split > 0 ? Integer.toString(split) : "c") + flags;
+ String label = "RecordRepacker(" + name + cfg + ")";
+ String dir = "repack-" + kind + (full ? "-full" : "") + implTag
+ + (stages == 2 ? "-idx" : "")
+ + "-" + threads + "-" + split
+ + (exact ? "" : "-inexact") + (single ? "-1w" : "");
+ final boolean ex = exact;
+ final boolean sw = single;
+ final int win = window;
+ final int st = stages;
+ final RecordRepacker.DedupCacheImpl impl = cacheImpl;
+ return new Arm(label, dir,
+ fs -> runRecordRepacker(fs, mode, expectedRecords, threads, split, ex, sw, win, impl, st));
+ }
+
+ /** A named compaction to benchmark, run on a raw copy of the prepared store in {@code dir}. */
+ private static final class Arm {
+ final String label;
+ final String dir;
+ final Compaction compaction;
+
+ Arm(String label, String dir, Compaction compaction) {
+ this.label = label;
+ this.dir = dir;
+ this.compaction = compaction;
+ }
+ }
+
+ private static void printRuns(String label, long[] times, long[] heaps, long[] outputs, long[] records) {
+ System.out.printf("%n%s%n", label);
+ System.out.printf(" %-5s %12s %16s %16s %14s%n",
+ "run", "time (ms)", "peak heap (MB)", "output (bytes)", "records");
+ for (int i = 0; i < times.length; i++) {
+ System.out.printf(" %-5d %12d %16d %16d %14d%n", i + 1, times[i], heaps[i], outputs[i], records[i]);
+ }
+ System.out.printf(" %-5s %12d %16d %16d %14d%n", "min", min(times), min(heaps), min(outputs), min(records));
+ System.out.printf(" %-5s %12d %16d %16d %14d%n", "med", median(times), median(heaps), median(outputs), median(records));
+ System.out.printf(" %-5s %12d %16d %16d %14d%n", "mean", mean(times), mean(heaps), mean(outputs), mean(records));
+ System.out.printf(" %-5s %12d %16d %16d %14d%n", "max", max(times), max(heaps), max(outputs), max(records));
+ }
+
+ private static long min(long[] a) {
+ long m = a[0];
+ for (long x : a) {
+ m = Math.min(m, x);
+ }
+ return m;
+ }
+
+ private static long max(long[] a) {
+ long m = a[0];
+ for (long x : a) {
+ m = Math.max(m, x);
+ }
+ return m;
+ }
+
+ private static long mean(long[] a) {
+ long sum = 0;
+ for (long x : a) {
+ sum += x;
+ }
+ return sum / a.length;
+ }
+
+ private static long median(long[] a) {
+ long[] sorted = a.clone();
+ Arrays.sort(sorted);
+ int n = sorted.length;
+ return n % 2 == 1 ? sorted[n / 2] : (sorted[n / 2 - 1] + sorted[n / 2]) / 2;
+ }
+
+ /** Median over the first {@code n} elements of {@code a}. */
+ private static long median(long[] a, int n) {
+ long[] sorted = Arrays.copyOf(a, n);
+ Arrays.sort(sorted);
+ return n % 2 == 1 ? sorted[n / 2] : (sorted[n / 2 - 1] + sorted[n / 2]) / 2;
+ }
+
+ private static void deleteRecursively(File file) {
+ File[] children = file.listFiles();
+ if (children != null) {
+ for (File child : children) {
+ deleteRecursively(child);
+ }
+ }
+ file.delete();
+ }
+
+ /**
+ * Build a fresh segment store at {@code prepared} by copying the content tree of
+ * {@code sample} through the {@link org.apache.jackrabbit.oak.spi.state.NodeStore} API.
+ * Copying node-by-node re-writes every record from scratch, so the copies carry
+ * self-reference-default stable ids instead of the source's (verified by
+ * {@link #verifyStableIdsDropped}). Garbage is introduced by intermittently deleting an
+ * already-copied top-level branch and copying it again, so its first copy becomes
+ * unreferenced, and a couple of intermediate states are retained as checkpoints so the
+ * repacked graph has real checkpoint history.
+ */
+ private void prepareStoreWithGarbage(File sample, File prepared) throws Exception {
+ Files.createDirectories(prepared.toPath());
+ int garbageEvents = 0;
+ int checkpoints = 0;
+ List<String> names = new ArrayList<>();
+ // Only stores with external binaries (blobs in a - here absent - DataStore) need a
+ // BlobStore, and only to relink references by id; stores with inline binaries must be
+ // copied without one, so the binaries stay inlined instead of being externalised.
+ BlobStore blobStore = hasExternalBlobs(sample) ? new IdentityBlobStore() : null;
+ FileStoreBuilder sampleBuilder = fileStoreBuilder(sample);
+ FileStoreBuilder destBuilder = fileStoreBuilder(prepared);
+ if (blobStore != null) {
+ sampleBuilder.withBlobStore(blobStore);
+ destBuilder.withBlobStore(blobStore);
+ }
+ try (FileStore sampleStore = sampleBuilder.build();
+ FileStore destStore = destBuilder.build()) {
+ SegmentNodeStore sampleNs = SegmentNodeStoreBuilders.builder(sampleStore).build();
+ SegmentNodeStore destNs = SegmentNodeStoreBuilders.builder(destStore).build();
+ NodeState src = sampleNs.getRoot();
+ for (ChildNodeEntry entry : src.getChildNodeEntries()) {
+ names.add(entry.getName());
+ }
+
+ NodeBuilder rootProps = destNs.getRoot().builder();
+ for (PropertyState p : src.getProperties()) {
+ rootProps.setProperty(p);
+ }
+ destNs.merge(rootProps, EmptyHook.INSTANCE, CommitInfo.EMPTY);
+
+ String previous = null;
+ for (int i = 0; i < names.size(); i++) {
+ String name = names.get(i);
+ NodeBuilder root = destNs.getRoot().builder();
+ copyTree(src.getChildNode(name), root.child(name));
+ destNs.merge(root, EmptyHook.INSTANCE, CommitInfo.EMPTY);
+
+ // Intermittently turn an already-copied branch into garbage.
+ if (previous != null && i % 3 == 2) {
+ recreateBranch(destNs, src, previous);
+ garbageEvents++;
+ }
+ // Retain a couple of intermediate states as checkpoints.
+ if (i == names.size() / 3 || i == 2 * names.size() / 3) {
+ destNs.checkpoint(TimeUnit.DAYS.toMillis(1));
+ checkpoints++;
+ }
+ previous = name;
+ }
+ // Guarantee at least one garbage event even for stores with few top-level branches.
+ if (garbageEvents == 0 && previous != null) {
+ recreateBranch(destNs, src, previous);
+ garbageEvents++;
+ }
+ destStore.flush();
+ }
+ System.out.printf(
+ "prepared store from %d top-level branches, %d garbage delete/recopy events, %d checkpoints%n",
+ names.size(), garbageEvents, checkpoints);
+ assertTrue(garbageEvents > 0, "expected to create garbage while preparing the store");
+ assertTrue(checkpoints > 0, "expected to create checkpoints while preparing the store");
+ }
+
+ /** Delete branch {@code name} from {@code destNs}, then copy it again from {@code src}. */
+ private static void recreateBranch(SegmentNodeStore destNs, NodeState src, String name)
+ throws CommitFailedException {
+ NodeBuilder delete = destNs.getRoot().builder();
+ delete.getChildNode(name).remove();
+ destNs.merge(delete, EmptyHook.INSTANCE, CommitInfo.EMPTY);
+
+ NodeBuilder recopy = destNs.getRoot().builder();
+ copyTree(src.getChildNode(name), recopy.child(name));
+ destNs.merge(recopy, EmptyHook.INSTANCE, CommitInfo.EMPTY);
+ }
+
+ private static void copyTree(NodeState src, NodeBuilder dest) {
+ for (PropertyState p : src.getProperties()) {
+ dest.setProperty(p);
+ }
+ for (ChildNodeEntry e : src.getChildNodeEntries()) {
+ copyTree(e.getNodeState(), dest.child(e.getName()));
+ }
+ }
+
+ /**
+ * Verify that copying through the NodeStore API dropped the source stable ids: the
+ * copied nodes should carry the self-reference-default stable id (the record id at index
+ * 0 equals the node's own record id). Walks a bounded sample of the prepared store.
+ */
+ private void verifyStableIdsDropped(File prepared) throws Exception {
+ try (FileStore fileStore = fileStoreBuilder(prepared).build()) {
+ long self = 0;
+ long explicit = 0;
+ long cap = 50_000;
+ Deque<SegmentNodeState> queue = new ArrayDeque<>();
+ queue.add(fileStore.getHead());
+ while (!queue.isEmpty() && self + explicit < cap) {
+ SegmentNodeState node = queue.removeFirst();
+ if (hasSelfMarkerStableId(node)) {
+ self++;
+ } else {
+ explicit++;
+ }
+ for (ChildNodeEntry e : node.getChildNodeEntries()) {
+ NodeState child = e.getNodeState();
+ if (child instanceof SegmentNodeState) {
+ queue.add((SegmentNodeState) child);
+ }
+ }
+ }
+ System.out.printf("prepared store stable ids (sampled %d nodes): self-marker=%d, explicit=%d%n",
+ self + explicit, self, explicit);
+ assertTrue(self > explicit,
+ "NodeStore-level copy should drop stable ids (self-marker nodes should dominate)");
+ }
+ }
+
+ private static boolean hasSelfMarkerStableId(SegmentNodeState node) {
+ // Mirrors SegmentNodeState.getStableIdBytes: when the record id at index 0 equals the
+ // node's own record id, the node carries no explicit (source) stable id.
+ return node.getSegment().readRecordId(node.getRecordNumber()).equals(node.getRecordId());
+ }
+
+ /**
+ * Whether {@code sample} stores any binary as an external blob reference (data in a
+ * DataStore) rather than inline. Walks the content tree, stopping at the first external
+ * binary, so it is cheap on external-blob stores and a full (read-only) scan on inline ones.
+ */
+ private static boolean hasExternalBlobs(File sample) throws Exception {
+ try (FileStore store = fileStoreBuilder(sample).build()) {
+ Deque<NodeState> queue = new ArrayDeque<>();
+ queue.add(SegmentNodeStoreBuilders.builder(store).build().getRoot());
+ while (!queue.isEmpty()) {
+ NodeState node = queue.removeFirst();
+ for (PropertyState p : node.getProperties()) {
+ if (p.getType() == BINARY) {
+ if (isExternalBlob(p.getValue(BINARY))) {
+ return true;
+ }
+ } else if (p.getType() == BINARIES) {
+ for (Blob b : p.getValue(BINARIES)) {
+ if (isExternalBlob(b)) {
+ return true;
+ }
+ }
+ }
+ }
+ for (ChildNodeEntry e : node.getChildNodeEntries()) {
+ queue.add(e.getNodeState());
+ }
+ }
+ }
+ return false;
+ }
+
+ private static boolean isExternalBlob(Blob blob) {
+ return blob instanceof SegmentBlob && ((SegmentBlob) blob).isExternal();
+ }
+
+ /**
+ * Identity {@link BlobStore} used only to copy external blob <em>references</em> across
+ * stores in {@link #prepareStoreWithGarbage}: a blob reference equals its blob id, so an
+ * external binary is relinked by id without its data (which lives in an absent DataStore).
+ * Compaction never reads blob data, so this faithfully preserves external references.
+ * <p>
+ * The blob <em>length</em> is derivable from the blob id (which ends in {@code #<length>}, a
+ * DataStore convention), so {@link #getBlobLength} answers without the data - unlike the
+ * genuinely data-backed methods, which are unreachable on this path and therefore unsupported.
+ */
+ private static final class IdentityBlobStore implements BlobStore {
+
+ private final Map<String, Long> referenceToLength = new ConcurrentHashMap<>();
+
+ @Override
+ public String getReference(String blobId) {
+ DataStoreBlobStore.BlobId id = DataStoreBlobStore.BlobId.of(blobId);
+ String reference = id.getBlobId();
+ referenceToLength.putIfAbsent(reference, id.getLength());
+ return reference;
+ }
+
+ @Override
+ public String getBlobId(String reference) {
+ Long length = referenceToLength.get(reference);
+ if (length != null) {
+ return reference + "#" + length;
+ }
+ return reference;
+ }
+
+ @Override
+ public long getBlobLength(String blobId) {
+ int separator = blobId.lastIndexOf('#');
+ if (separator >= 0) {
+ return Long.parseLong(blobId.substring(separator + 1));
+ }
+ throw new UnsupportedOperationException("blob id carries no length: " + blobId);
+ }
+
+ @Override
+ public String writeBlob(InputStream in) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public String writeBlob(InputStream in, BlobOptions options) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public int readBlob(String blobId, long pos, byte[] buff, int off, int length) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public InputStream getInputStream(String blobId) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public void close() {
+ }
+ }
+
+ private void runRecordRepacker(FileStore fileStore, RecordRepacker.Mode mode, int expectedRecords)
+ throws Exception {
+ runRecordRepacker(fileStore, mode, expectedRecords, 1);
+ }
+
+ private void runRecordRepacker(FileStore fileStore, RecordRepacker.Mode mode, int expectedRecords,
+ int concurrency) throws Exception {
+ runRecordRepacker(fileStore, mode, expectedRecords, concurrency, 0);
+ }
+
+ private void runRecordRepacker(FileStore fileStore, RecordRepacker.Mode mode, int expectedRecords,
+ int concurrency, int splitTarget) throws Exception {
+ runRecordRepacker(fileStore, mode, expectedRecords, concurrency, splitTarget, true);
+ }
+
+ private void runRecordRepacker(FileStore fileStore, RecordRepacker.Mode mode, int expectedRecords,
+ int concurrency, int splitTarget, boolean exactDedup) throws Exception {
+ runRecordRepacker(fileStore, mode, expectedRecords, concurrency, splitTarget, exactDedup, false);
+ }
+
+ private void runRecordRepacker(FileStore fileStore, RecordRepacker.Mode mode, int expectedRecords,
+ int concurrency, int splitTarget, boolean exactDedup, boolean singleWriter)
+ throws Exception {
+ // -1 = the default auto-sized recency window (hot-dedup).
+ runRecordRepacker(fileStore, mode, expectedRecords, concurrency, splitTarget, exactDedup,
+ singleWriter, -1);
+ }
+
+ private void runRecordRepacker(FileStore fileStore, RecordRepacker.Mode mode, int expectedRecords,
+ int concurrency, int splitTarget, boolean exactDedup, boolean singleWriter, int dedupWindow)
+ throws Exception {
+ runRecordRepacker(fileStore, mode, expectedRecords, concurrency, splitTarget, exactDedup,
+ singleWriter, dedupWindow, RecordRepacker.DedupCacheImpl.MMAP_WINDOW, 0);
+ }
+
+ private void runRecordRepacker(FileStore fileStore, RecordRepacker.Mode mode, int expectedRecords,
+ int concurrency, int splitTarget, boolean exactDedup, boolean singleWriter, int dedupWindow,
+ RecordRepacker.DedupCacheImpl cacheImpl, int stages) throws Exception {
+ SegmentNodeState head = fileStore.getHead();
+ RecordId root = head.getRecordId();
+ GCGeneration target = root.getSegmentId().getGcGeneration().nextFull();
+ RecordRepacker repacker = new RecordRepacker(fileStore, fileStore.getReader(),
+ fileStore.getSegmentIdProvider(), fileStore.getBlobStore(),
+ fileStore.getBinariesInlineThreshold(), target, mode, expectedRecords, concurrency)
+ .withSplitTarget(splitTarget)
+ .withExactDedup(exactDedup)
+ .withSingleWriter(singleWriter)
+ .withDedupWindow(dedupWindow)
+ .withDedupCacheImpl(cacheImpl)
+ .withStages(stageRoots(head, stages));
+ RecordId repacked = repacker.repack(root);
+ assertEquals(target, repacked.getSegmentId().getGcGeneration(),
+ "repacked root must be in the target generation");
+ }
+
+ /**
+ * Record id of {@code /oak:index} under the superroot {@code head}, or {@code null} if absent.
+ * Drives {@link RecordRepacker#withIndexFirst} for the two-stage index-first repack.
+ */
+ private static RecordId indexRoot(SegmentNodeState head) {
+ NodeState index = head.getChildNode("root").getChildNode("oak:index");
+ return index instanceof SegmentNodeState ? ((SegmentNodeState) index).getRecordId() : null;
+ }
+
+ /**
+ * Ordered pre-stage roots for {@link RecordRepacker#withStages}, given the superroot {@code head}:
+ * {@code stages<=0} none (single-stage repack); {@code 1} {@code [/root/oak:index]} (index-first);
+ * {@code 2} {@code [/root/oak:index, /root]} (the whole superroot {@code /} is always the final
+ * stage, so this yields three parallel barriers: {@code /root/oak:index}, {@code /root}, {@code /}).
+ */
+ private static List<RecordId> stageRoots(SegmentNodeState head, int stages) {
+ if (stages <= 0) {
+ return List.of();
+ }
+ RecordId indexId = indexRoot(head);
+ if (indexId == null) {
+ return List.of();
+ }
+ if (stages == 1) {
+ return List.of(indexId);
+ }
+ return List.of(indexId, ((SegmentNodeState) head.getChildNode("root")).getRecordId());
+ }
+
+ private void runCheckpointCompactor(FileStore fileStore) throws Exception {
+ SegmentNodeState head = fileStore.getHead();
+ GCGeneration base = head.getGcGeneration();
+ GCGeneration target = base.nextFull();
+ GCIncrement increment = new GCIncrement(base, base.nextPartial(), target);
+ SegmentWriterFactory writerFactory = generation -> defaultSegmentWriterBuilder("c")
+ .withGeneration(generation).build(fileStore);
+ CompactionWriter compactionWriter = new CompactionWriter(fileStore.getReader(),
+ fileStore.getBlobStore(), increment, writerFactory);
+ GCNodeWriteMonitor monitor = new GCNodeWriteMonitor(-1, GCMonitor.EMPTY);
+ CheckpointCompactor compactor = new CheckpointCompactor(GCMonitor.EMPTY,
+ new ClassicCompactor(compactionWriter, monitor));
+ CompactedNodeState compacted = compactor.compactUp(head, Canceller.newCanceller());
+ assertNotNull(compacted, "checkpoint compaction must not be cancelled");
+ assertEquals(target, compacted.getGcGeneration(),
+ "compacted head must be in the target generation");
+ compactionWriter.flush();
+ }
+
+ private void runParallelCompactor(FileStore fileStore, int concurrency) throws Exception {
+ SegmentNodeState head = fileStore.getHead();
+ GCGeneration base = head.getGcGeneration();
+ GCGeneration target = base.nextFull();
+ GCIncrement increment = new GCIncrement(base, base.nextPartial(), target);
+ SegmentWriterFactory writerFactory = generation -> defaultSegmentWriterBuilder("c")
+ .withGeneration(generation)
+ .withWriterPool(SegmentBufferWriterPool.PoolType.THREAD_SPECIFIC)
+ .build(fileStore);
+ CompactionWriter compactionWriter = new CompactionWriter(fileStore.getReader(),
+ fileStore.getBlobStore(), increment, writerFactory);
+ GCNodeWriteMonitor monitor = new GCNodeWriteMonitor(-1, GCMonitor.EMPTY);
+ CheckpointCompactor compactor = new CheckpointCompactor(GCMonitor.EMPTY,
+ new ParallelCompactor(GCMonitor.EMPTY, compactionWriter, monitor, concurrency));
+ CompactedNodeState compacted = compactor.compactUp(head, Canceller.newCanceller());
+ assertNotNull(compacted, "parallel compaction must not be cancelled");
+ assertEquals(target, compacted.getGcGeneration(),
+ "compacted head must be in the target generation");
+ compactionWriter.flush();
+ }
+
+ @FunctionalInterface
+ private interface Compaction {
+ void run(FileStore fileStore) throws Exception;
+ }
+
+ /**
+ * Run {@code compaction} on a fresh copy of {@code sample}, measuring the wall-clock
+ * time, the peak heap used (sampled on a background thread, relative to a settled
+ * baseline) and the size of the compacted generation (growth of the store directory
+ * after flushing). Time includes the final flush so both compactors are measured
+ * end-to-end.
+ */
+ private CompactionResult measureCompaction(String label, File dest, File sample,
+ Compaction compaction) throws Exception {
+ copySegmentStore(sample, dest);
+ long elapsedMillis;
+ long peakHeapBytes;
+ long outputBytes;
+ GCGeneration target;
+ try (FileStore fileStore = fileStoreBuilder(dest).build()) {
+ fileStore.flush();
+ long sizeBefore = directorySize(dest);
+ // Each compaction writes the next full generation on top of the head; count the
+ // records it produced once it has been flushed.
+ target = fileStore.getHead().getGcGeneration().nextFull();
+
+ MemoryMXBean memoryBean = ManagementFactory.getMemoryMXBean();
+ settle();
+ long baseHeap = memoryBean.getHeapMemoryUsage().getUsed();
+ AtomicLong peak = new AtomicLong(baseHeap);
+ AtomicBoolean sampling = new AtomicBoolean(true);
+ Thread sampler = new Thread(() -> {
+ while (sampling.get()) {
+ peak.accumulateAndGet(memoryBean.getHeapMemoryUsage().getUsed(), Math::max);
+ try {
+ Thread.sleep(10);
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ return;
+ }
+ }
+ });
+ sampler.setDaemon(true);
+ sampler.start();
+
+ long start = System.nanoTime();
+ compaction.run(fileStore);
+ fileStore.flush();
+ elapsedMillis = (System.nanoTime() - start) / 1_000_000;
+
+ sampling.set(false);
+ sampler.join();
+
+ peakHeapBytes = Math.max(0, peak.get() - baseHeap);
+ outputBytes = directorySize(dest) - sizeBefore;
+ }
+ long records = countRecordsInGeneration(dest, target);
+ return new CompactionResult(label, elapsedMillis, peakHeapBytes, outputBytes, records);
+ }
+
+ /**
+ * Count the records the compaction wrote into the {@code target} generation of the store
+ * at {@code dir}. Reopens the store read-only and sums the record counts of all data
+ * segments in that generation; bulk binary segments are shared and carry no generation,
+ * so they are excluded (matching the {@code output (bytes)} column, which measures the
+ * growth of the store directory).
+ */
+ private static long countRecordsInGeneration(File dir, GCGeneration target) throws Exception {
+ long[] records = {0};
+ try (ReadOnlyFileStore store = fileStoreBuilder(dir).buildReadOnly()) {
+ for (SegmentId id : store.getSegmentIds()) {
+ if (id.isDataSegmentId() && target.equals(id.getGcGeneration())) {
+ id.getSegment().forEachRecord((number, type, offset) -> records[0]++);
+ }
+ }
+ }
+ return records[0];
+ }
+
+ /**
+ * Total record count and non-bulk (data-segment) byte size of the whole store at {@code dir},
+ * summed over every data segment regardless of generation. Bulk binary segments are excluded, so
+ * the byte figure is comparable to each arm's {@code output (bytes)}. Unlike
+ * {@link #countRecordsInGeneration}, this includes unreferenced (garbage) records still present
+ * on disk, giving the true pre-compaction size.
+ *
+ * @return {@code [recordCount, dataSegmentBytes]}.
+ */
+ private static long[] countStoreContent(File dir) throws Exception {
+ long[] stats = {0, 0};
+ try (ReadOnlyFileStore store = fileStoreBuilder(dir).buildReadOnly()) {
+ for (SegmentId id : store.getSegmentIds()) {
+ if (id.isDataSegmentId()) {
+ Segment segment = id.getSegment();
+ segment.forEachRecord((number, type, offset) -> stats[0]++);
+ stats[1] += segment.size();
+ }
+ }
+ }
+ return stats;
+ }
+
+ private static void settle() throws InterruptedException {
+ for (int i = 0; i < 3; i++) {
+ System.gc();
+ Thread.sleep(100);
+ }
+ }
+
+ private static long directorySize(File dir) {
+ long total = 0;
+ File[] files = dir.listFiles();
+ if (files != null) {
+ for (File file : files) {
+ total += file.isDirectory() ? directorySize(file) : file.length();
+ }
+ }
+ return total;
+ }
+
+ private static final class CompactionResult {
+ final String label;
+ final long elapsedMillis;
+ final long peakHeapBytes;
+ final long outputBytes;
+ final long recordCount;
+
+ CompactionResult(String label, long elapsedMillis, long peakHeapBytes, long outputBytes,
+ long recordCount) {
+ this.label = label;
+ this.elapsedMillis = elapsedMillis;
+ this.peakHeapBytes = peakHeapBytes;
+ this.outputBytes = outputBytes;
+ this.recordCount = recordCount;
+ }
+ }
+
+ /**
+ * Recursively assert that {@code actual} has exactly the same properties and
+ * children (by content) as {@code expected}. Deliberately avoids
+ * {@link SegmentNodeState#equals}, which short-circuits on the (preserved)
+ * stable id and would not detect a corrupted repack.
+ */
+ private static void assertSameContent(String path, NodeState expected, NodeState actual) {
+ assertEquals(expected.getPropertyCount(), actual.getPropertyCount(),
+ "property count at " + path);
+ for (PropertyState p : expected.getProperties()) {
+ PropertyState q = actual.getProperty(p.getName());
+ assertNotNull(q, "missing property " + p.getName() + " at " + path);
+ assertSameProperty(path, p, q);
+ }
+
+ assertEquals(expected.getChildNodeCount(Long.MAX_VALUE),
+ actual.getChildNodeCount(Long.MAX_VALUE), "child count at " + path);
+ for (ChildNodeEntry entry : expected.getChildNodeEntries()) {
+ NodeState child = actual.getChildNode(entry.getName());
+ assertTrue(child.exists(), "missing child " + entry.getName() + " at " + path);
+ assertSameContent(path + entry.getName() + "/", entry.getNodeState(), child);
+ }
+ }
+
+ private static void assertSameProperty(String path, PropertyState p, PropertyState q) {
+ assertEquals(p.getType(), q.getType(), "type of " + p.getName() + " at " + path);
+ assertEquals(p.count(), q.count(), "value count of " + p.getName() + " at " + path);
+ Type<?> type = p.getType();
+ if (type == BINARY || type == BINARIES) {
+ // Binaries are relinked (bulk blocks are shared): comparing lengths is a
+ // cheap and sufficient check that avoids streaming large binary content.
+ for (int i = 0; i < p.count(); i++) {
+ assertEquals(p.getValue(BINARY, i).length(), q.getValue(BINARY, i).length(),
+ "binary length of " + p.getName() + "[" + i + "] at " + path);
+ }
+ } else {
+ assertEquals(p, q, "property " + p.getName() + " at " + path);
+ }
+ }
+
+ private static void copySegmentStore(File from, File to) throws IOException {
+ Files.createDirectories(to.toPath());
+ File[] files = from.listFiles();
+ assertNotNull(files, "cannot list " + from);
+ for (File file : files) {
+ if (file.isFile() && !"repo.lock".equals(file.getName())) {
+ Files.copy(file.toPath(), to.toPath().resolve(file.getName()));
+ }
+ }
+ }
+
+ private byte[] randomBytes(int size) {
+ byte[] data = new byte[size];
+ new java.util.Random(size).nextBytes(data);
+ return data;
+ }
+
+ private Blob blob(byte[] data) throws IOException {
+ return new SegmentBlob(store.getBlobStore(), writer.writeStream(new ByteArrayInputStream(data)));
+ }
+
+ // ------------------------------------------------------------------------
+ // Read locality benchmark
+ //
+ // The repacker writes the record graph in post-order DFS, so a node's whole subtree is emitted
+ // as one contiguous run of records (and hence into a small set of consecutive segments). This
+ // benchmark measures how well that on-disk layout serves reads: it lays out the same content
+ // three ways (serial repack, parallel repack, semantic CheckpointCompactor), then traverses the
+ // content tree of each with a deliberately small segment cache (so the working set does not fit
+ // and locality shows up as re-fetches) in both depth-first and breadth-first read order. The
+ // metric is the number of segment fetches (archive reads = segment-cache misses), counted via
+ // an {@link IOMonitor} on the read path - a structural figure independent of how fast the local
+ // TAR store is. A "simulated ms" column applies a configurable per-fetch/per-byte latency to
+ // model a network-backed store (Azure/S3), where each fetch is a round-trip.
+ // ------------------------------------------------------------------------
+
+ @Test
+ void compareReadLocality(@TempDir Path tempDir) throws Exception {
+ File sample = getSampleStore();
+
+ File prepared = tempDir.resolve("prepared").toFile();
+ prepareStoreWithGarbage(sample, prepared);
+ int expectedRecords = (int) Math.min(Integer.MAX_VALUE, countStoreContent(prepared)[0]);
+
+ // Small cache relative to the compacted tree, so a full traversal must re-fetch under
+ // eviction and the layout's locality is exposed. Latency is applied arithmetically to the
+ // fetch/byte counts (no real sleeping) to keep the run fast and deterministic.
+ int[] cacheSizes = parseCacheSizes(System.getProperty("oak.repack.readCacheMB", "32"));
+ long fetchLatencyNanos = Long.getLong("oak.repack.readLatencyMicros", 1000) * 1000L;
+ long byteLatencyNanos = Long.getLong("oak.repack.readBandwidthNanosPerKiB", 0) / 1024L;
+ int threads = Integer.getInteger("oak.repack.readThreads", 16);
+ int splitTarget = Integer.getInteger("oak.repack.readSplitTarget", 2048);
+
+ String header = "--- read locality (caches " + Arrays.toString(cacheSizes) + " MB, sim latency "
+ + (fetchLatencyNanos / 1000) + " us/fetch) ---" + System.lineSeparator()
+ + String.format("%-30s %-14s %8s %12s %12s %12s %12s%n",
+ "layout", "workload", "cacheMB", "fetches", "distinct", "MB fetched", "sim ms");
+ System.out.print(header);
+ appendResultFile(header);
+
+ // Produce and measure one layout at a time, deleting its store before the next, so peak disk
+ // stays bounded even for a large sample. Scenarios are grouped so each one's serial and
+ // parallel (‖threads) rows are adjacent, serial first.
+ //
+ // preserve: no content deduplication (only physical source sharing via the alias). Serially
+ // this writes every record fresh and contiguous; in parallel each worker keeps its own copy of
+ // shared records (no cross-worker dedup). Isolates whether dedup - pointing a node's children
+ // at records another writer emitted elsewhere - is what fragments reads.
+ measureLayout("repack(preserve)", tempDir.resolve("serial-nodedup").toFile(), prepared,
+ dir -> produceRepackLayout(prepared, dir, RecordRepacker.Mode.NO_DEDUP, 1, 0,
+ expectedRecords),
+ cacheSizes, fetchLatencyNanos, byteLatencyNanos);
+ measureLayout("repack(preserve‖" + threads + ")", tempDir.resolve("parallel-nodedup").toFile(),
+ prepared,
+ dir -> produceRepackLayout(prepared, dir, RecordRepacker.Mode.NO_DEDUP, threads,
+ splitTarget, expectedRecords),
+ cacheSizes, fetchLatencyNanos, byteLatencyNanos);
+ // full-dedup (exact off-heap cache) vs hot-dedup (default auto-sized recency window): the
+ // window layout must read as well as the exact one to confirm the window is a safe default.
+ measureLayout("repack(full-dedup)", tempDir.resolve("serial-exact").toFile(), prepared,
+ dir -> produceRepackLayout(prepared, dir, RecordRepacker.Mode.NODE_DATA, 1, 0,
+ expectedRecords, 0),
+ cacheSizes, fetchLatencyNanos, byteLatencyNanos);
+ measureLayout("repack(full-dedup‖" + threads + ")", tempDir.resolve("parallel-exact").toFile(),
+ prepared,
+ dir -> produceRepackLayout(prepared, dir, RecordRepacker.Mode.NODE_DATA, threads,
+ splitTarget, expectedRecords, 0),
+ cacheSizes, fetchLatencyNanos, byteLatencyNanos);
+ measureLayout("repack(hot-dedup)", tempDir.resolve("serial").toFile(), prepared,
+ dir -> produceRepackLayout(prepared, dir, RecordRepacker.Mode.NODE_DATA, 1, 0,
+ expectedRecords, -1),
+ cacheSizes, fetchLatencyNanos, byteLatencyNanos);
+ measureLayout("repack(hot-dedup‖" + threads + ")", tempDir.resolve("parallel").toFile(),
+ prepared,
+ dir -> produceRepackLayout(prepared, dir, RecordRepacker.Mode.NODE_DATA, threads,
+ splitTarget, expectedRecords, -1),
+ cacheSizes, fetchLatencyNanos, byteLatencyNanos);
+ // hot-dedup+idx: the three-stage index-first repack - /oak:index, then /root, then / - each a
+ // parallel barrier, keeping the wide index child sets in one contiguous run. The parallel row
+ // is the point of the mode; the serial row is the staged ordering applied single-threaded.
+ measureLayout("repack(hot-dedup+idx)", tempDir.resolve("serial-idx").toFile(), prepared,
+ dir -> produceRepackLayout(prepared, dir, RecordRepacker.Mode.NODE_DATA, 1, 0,
+ expectedRecords, -1, 2),
+ cacheSizes, fetchLatencyNanos, byteLatencyNanos);
+ measureLayout("repack(hot-dedup+idx‖" + threads + ")", tempDir.resolve("parallel-idx").toFile(),
+ prepared,
+ dir -> produceRepackLayout(prepared, dir, RecordRepacker.Mode.NODE_DATA, threads,
+ splitTarget, expectedRecords, -1, 2),
+ cacheSizes, fetchLatencyNanos, byteLatencyNanos);
+ measureLayout("semantic(checkpoint)", tempDir.resolve("semantic").toFile(), prepared,
+ dir -> produceSemanticLayout(prepared, dir),
+ cacheSizes, fetchLatencyNanos, byteLatencyNanos);
+ measureLayout("semantic(parallel‖" + threads + ")", tempDir.resolve("semantic-parallel").toFile(),
+ prepared,
+ dir -> produceParallelSemanticLayout(prepared, dir, threads),
+ cacheSizes, fetchLatencyNanos, byteLatencyNanos);
+ }
+
+ /**
+ * SIZE decomposition of the bounded recency-window content-dedup ({@link RecordRepacker#withDedupWindow})
+ * that replaces the exact off-heap-spilled dedup cache. Contrasts the exact record-dedup floor against
+ * a sweep of window sizes (default {@code 65536,262144,1048576,4194304}, override with
+ * {@code -Doak.repack.dedupWindows=...}), serial and parallel, reporting records + bytes + per-type +
+ * the window hit/miss counts. A window that recovers the exact floor's size proves the hot duplicates
+ * (index mirrors, version buckets, type names) recur within the window, so the filesystem spill for
+ * duplicates can be dropped. Additive and test-only.
+ */
+ @Test
+ void measureDedupWindow(@TempDir Path tempDir) throws Exception {
+ File sample = getSampleStore();
+
+ File prepared = tempDir.resolve("prepared").toFile();
+ prepareStoreWithGarbage(sample, prepared);
+ long[] content = countStoreContent(prepared);
+ int expectedRecords = (int) Math.min(Integer.MAX_VALUE, content[0]);
+ int threads = Integer.getInteger("oak.repack.readThreads", 16);
+ int splitTarget = Integer.getInteger("oak.repack.readSplitTarget", 2048);
+ boolean analyzeTypes = Boolean.parseBoolean(System.getProperty("oak.repack.analyzeTypes", "true"));
+ int[] windows = parseThresholds(
+ System.getProperty("oak.repack.dedupWindows", "65536,262144,1048576,4194304"));
+
+ System.out.println("--- recency-window content-dedup SIZE decomposition ---");
+ System.out.printf("%-32s %14s %10s | %12s %10s %10s %10s %12s | %s%n",
+ "arm", "records", "bytes", "nodes", "templates", "maps", "lists", "strings",
+ "win hits/misses");
+ System.out.printf("%-32s %,14d %10s%n", "(input, pre-compaction)", content[0], mb(content[1]));
+
+ windowArm("full-dedup(serial)", prepared, tempDir, 0, 1, 0, expectedRecords, analyzeTypes);
+ for (int w : windows) {
+ windowArm("window " + w + "(serial)", prepared, tempDir, w, 1, 0, expectedRecords, analyzeTypes);
+ }
+ windowArm("full-dedup(‖" + threads + ")", prepared, tempDir, 0, threads, splitTarget,
+ expectedRecords, analyzeTypes);
+ for (int w : windows) {
+ windowArm("window " + w + "(‖" + threads + ")", prepared, tempDir, w, threads, splitTarget,
+ expectedRecords, analyzeTypes);
+ }
+ }
+
+ /**
+ * One recency-window repack arm: repack the prepared store in {@link RecordRepacker.Mode#NODE_DATA}
+ * with the given content-dedup window ({@code 0} = the exact off-heap cache), print records + bytes +
+ * per-type breakdown + window hit/miss counts, then delete it.
+ */
+ private void windowArm(String label, File prepared, Path tempDir, int window, int concurrency,
+ int splitTarget, int expectedRecords, boolean analyzeTypes) throws Exception {
+ File dir = tempDir.resolve("win-" + label.replaceAll("[^a-zA-Z0-9]", "_")).toFile();
+ copySegmentStore(prepared, dir);
+ long sizeBefore = directorySize(dir);
+ GCGeneration target;
+ long[] ref;
+ long hits;
+ long misses;
+ try (FileStore fs = fileStoreBuilder(dir).build()) {
+ SegmentNodeState head = fs.getHead();
+ RecordId root = head.getRecordId();
+ target = root.getSegmentId().getGcGeneration().nextFull();
+ RecordRepacker repacker = new RecordRepacker(fs, fs.getReader(), fs.getSegmentIdProvider(),
+ fs.getBlobStore(), fs.getBinariesInlineThreshold(), target,
+ RecordRepacker.Mode.NODE_DATA, expectedRecords, concurrency).withSplitTarget(splitTarget);
+ // 0 pins the exact off-heap cache; > 0 a fixed window. Unconditional so the "exact" arm is
+ // exact even though the repacker's default is now an auto-sized window.
+ repacker.withDedupWindow(window);
+ RecordId repacked = repacker.repack(root);
+ fs.flush();
+ hits = repacker.getDedupWindowHits();
+ misses = repacker.getDedupWindowMisses();
+ ref = toRef(repacked);
+ }
+ long bytes = directorySize(dir) - sizeBefore;
+ long records = countRecordsInGeneration(dir, target);
+
+ long nodes = -1, templates = -1, maps = -1, lists = -1, strings = -1;
+ if (analyzeTypes) {
+ try (ReadOnlyFileStore ro = fileStoreBuilder(dir).buildReadOnly()) {
+ SegmentId sid = ro.getSegmentIdProvider().newSegmentId(ref[0], ref[1]);
+ RecordUsageAnalyser a = new RecordUsageAnalyser(ro.getReader());
+ a.analyseNode(new RecordId(sid, (int) ref[2]));
+ nodes = a.getNodeCount();
+ templates = a.getTemplateCount();
+ maps = a.getMapCount();
+ lists = a.getListCount();
+ strings = a.getSmallStringCount() + a.getMediumStringCount() + a.getLongStringCount();
+ } catch (Exception | OutOfMemoryError e) {
+ System.out.println(" (per-type analysis skipped: " + e + ")");
+ }
+ }
+ String hm = window <= 0 ? "-" : String.format("%,d / %,d", hits, misses);
+ System.out.printf("%-32s %,14d %10s | %12s %10s %10s %10s %12s | %s%n",
+ label, records, mb(bytes),
+ col(nodes), col(templates), col(maps), col(lists), col(strings), hm);
+ deleteDir(dir);
+ }
+
+ private static String col(long v) {
+ return v < 0 ? "-" : String.format("%,d", v);
+ }
+
+ /**
+ * Decompose <em>why</em> the list-widest fetch count differs across layouts (in particular why
+ * parallel record-dedup reads so much worse than serial). For each layout it reads exactly the
+ * widest node's immediate children from a cold store and reports, in one pass:
+ * <ul>
+ * <li><b>fetches</b> - archive reads (cache misses), i.e. the headline number;</li>
+ * <li><b>distinct</b> - distinct segments the workload touches (the true footprint; the
+ * {@link SegmentFetchTracker} set is cache-independent);</li>
+ * <li><b>refetch</b> = fetches - distinct - segments re-read because the footprint exceeds the
+ * {@code cacheMB} cache and eviction unloads them (thrash), <em>not</em> footprint;</li>
+ * <li><b>nodeSegs</b> / <b>tmplSegs</b> - distinct segments holding the children's own NODE
+ * records resp. their TEMPLATE records, read straight from the record slots. This splits
+ * the footprint into "the children themselves are scattered" (worker/frontier split) vs
+ * "the records the children point at are scattered" (dedup: one canonical copy lives in
+ * whichever worker emitted it first).</li>
+ * </ul>
+ * Additive and test-only; leaves {@link #compareReadLocality} untouched.
+ */
+ @Test
+ void analyzeWideListingLocality(@TempDir Path tempDir) throws Exception {
+ File sample = getSampleStore();
+
+ File prepared = tempDir.resolve("prepared").toFile();
+ prepareStoreWithGarbage(sample, prepared);
+ int expectedRecords = (int) Math.min(Integer.MAX_VALUE, countStoreContent(prepared)[0]);
+
+ int cacheMB = Integer.getInteger("oak.repack.readCacheMB", 32);
+ int threads = Integer.getInteger("oak.repack.readThreads", 16);
+ int splitTarget = Integer.getInteger("oak.repack.readSplitTarget", 2048);
+
+ System.out.println("--- wide-listing locality decomposition (cache " + cacheMB + " MB) ---");
+ System.out.printf("%-24s %10s %10s %10s %10s %10s %10s%n",
+ "layout", "children", "fetches", "distinct", "refetch", "nodeSegs", "tmplSegs");
+
+ analyzeOneLayout("record-dedup(serial)", tempDir, prepared,
+ RecordRepacker.Mode.NODE_DATA, 1, 0, expectedRecords, cacheMB);
+ analyzeOneLayout("record-dedup(‖" + threads + ")", tempDir, prepared,
+ RecordRepacker.Mode.NODE_DATA, threads, splitTarget, expectedRecords, cacheMB);
+ analyzeOneLayout("no-dedup(‖" + threads + ")", tempDir, prepared,
+ RecordRepacker.Mode.NO_DEDUP, threads, splitTarget, expectedRecords, cacheMB);
+ analyzeOneLayout("no-dedup(serial)", tempDir, prepared,
+ RecordRepacker.Mode.NO_DEDUP, 1, 0, expectedRecords, cacheMB);
+ }
+
+ private void analyzeOneLayout(String label, Path tempDir, File prepared,
+ RecordRepacker.Mode mode, int concurrency, int splitTarget, int expectedRecords, int cacheMB)
+ throws Exception {
+ File dir = tempDir.resolve("layout-" + label.replaceAll("[^a-zA-Z0-9]", "_")).toFile();
+ long[] ref = produceRepackLayout(prepared, dir, mode, concurrency, splitTarget, expectedRecords);
+ try {
+ if (widestPath == null) {
+ discoverTargets(dir, ref);
+ }
+ List<String> widest = widestPath;
+
+ // (A) cold read of exactly list-widest, tracking total + distinct segment fetches.
+ SegmentFetchTracker tracker = new SegmentFetchTracker();
+ try (ReadOnlyFileStore store = fileStoreBuilder(dir)
+ .withSegmentCacheSize(cacheMB).withIOMonitor(tracker).buildReadOnly()) {
+ listImmediateChildren(resolvePath(readContentRoot(store, ref), widest));
+ }
+ long total = tracker.fetches.get();
+ long distinct = tracker.distinct.size();
+
+ // (B) warm pass: distinct segments of the children's own NODE and TEMPLATE records
+ // (slot 1 of a node record is its template; slot 0 is the stable id).
+ Set<String> nodeSegs = new HashSet<>();
+ Set<String> tmplSegs = new HashSet<>();
+ try (ReadOnlyFileStore store = fileStoreBuilder(dir).buildReadOnly()) {
+ NodeState widestNode = resolvePath(readContentRoot(store, ref), widest);
+ for (ChildNodeEntry e : widestNode.getChildNodeEntries()) {
+ RecordId nid = ((SegmentNodeState) e.getNodeState()).getRecordId();
+ nodeSegs.add(segKey(nid.getSegmentId()));
+ RecordId tmpl = nid.getSegment().readRecordId(nid.getRecordNumber(), 0, 1);
+ tmplSegs.add(segKey(tmpl.getSegmentId()));
+ }
+ }
+ System.out.printf("%-24s %10d %10d %10d %10d %10d %10d%n",
+ label, widestChildCount, total, distinct, total - distinct,
+ nodeSegs.size(), tmplSegs.size());
+
+ // (C) per-record-type segment attribution of the same list-widest footprint, so the
+ // scatter can be pinned to a specific layer (node / template / value / string / ...).
+ try (ReadOnlyFileStore store = fileStoreBuilder(dir).buildReadOnly()) {
+ RecordId widestId = ((SegmentNodeState) resolvePath(readContentRoot(store, ref), widest))
+ .getRecordId();
+ ListingSegCollector collector = new ListingSegCollector(store.getReader());
+ collector.collect(widestId);
+ System.out.printf(" by type (segs): node=%d template=%d map=%d value=%d "
+ + "string=%d list=%d blob=%d | union=%d%n",
+ collector.count("node"), collector.count("template"), collector.count("map"),
+ collector.count("value"), collector.count("string"), collector.count("list"),
+ collector.count("blob"), collector.all.size());
+ System.out.printf(" records: node=%d template=%d map=%d property=%d "
+ + "string=%d list=%d%n",
+ collector.records("node"), collector.records("template"),
+ collector.records("map"), collector.records("property"),
+ collector.records("string"), collector.records("list"));
+ System.out.printf(" bytes (MB): node=%.1f template=%.1f map=%.1f property=%.1f "
+ + "string=%.1f list=%.1f | union=%.1f MB%n",
+ collector.mb("node"), collector.mb("template"), collector.mb("map"),
+ collector.mb("property"), collector.mb("string"), collector.mb("list"),
+ collector.unionMB());
+ double segMB = 256.0 / 1024.0; // 256 KB per data segment
+ double valueLayerMB = collector.mb("property") + collector.mb("string")
+ + collector.mb("list");
+ System.out.printf(" ideal segs (bytes/256KB): value-layer=%.0f (%.1f MB) "
+ + "union=%.0f (%.1f MB) | ACTUAL union segs=%d%n",
+ Math.ceil(valueLayerMB / segMB), valueLayerMB,
+ Math.ceil(collector.unionMB() / segMB), collector.unionMB(),
+ collector.all.size());
+ }
+ } finally {
+ deleteDir(dir);
+ }
+ }
+
+ /**
+ * Same wide-listing segment attribution as {@link #analyzeWideListingLocality}, but for the
+ * <b>top-K widest nodes</b> (default 5, {@code -Doak.repack.topWidest}) and comparing only
+ * <b>hot-dedup serial vs hot-dedup ‖threads</b>. Each layout is produced once and every
+ * target node is attributed against it. Rows are labelled by the node's path so the per-node
+ * scatter can be read off directly. The record counts / bytes touched are layout-independent
+ * (parallelism only relocates records, it does not change what is reachable), so they are printed
+ * once per node from the serial pass. Additive and test-only.
+ */
+ @Test
+ void analyzeTopWideListingLocality(@TempDir Path tempDir) throws Exception {
+ File sample = getSampleStore();
+
+ File prepared = tempDir.resolve("prepared").toFile();
+ prepareStoreWithGarbage(sample, prepared);
+ int expectedRecords = (int) Math.min(Integer.MAX_VALUE, countStoreContent(prepared)[0]);
+
+ int cacheMB = Integer.getInteger("oak.repack.readCacheMB", 32);
+ int threads = Integer.getInteger("oak.repack.readThreads", 16);
+ int splitTarget = Integer.getInteger("oak.repack.readSplitTarget", 2048);
+ int topK = Integer.getInteger("oak.repack.topWidest", 5);
+
+ File serialDir = tempDir.resolve("hot-serial").toFile();
+ long[] serialRef = produceRepackLayout(prepared, serialDir, RecordRepacker.Mode.NODE_DATA,
+ 1, 0, expectedRecords);
+ File parDir = tempDir.resolve("hot-par").toFile();
+ long[] parRef = produceRepackLayout(prepared, parDir, RecordRepacker.Mode.NODE_DATA,
+ threads, splitTarget, expectedRecords);
+ try {
+ List<WideNode> targets = discoverTopWidest(serialDir, serialRef, topK);
+ System.out.println("--- top-" + topK + " widest-node listing locality (hot-dedup, cache "
+ + cacheMB + " MB); serial vs ‖" + threads + " ---");
+ long[] sStat = generationStats(serialDir, serialRef);
+ long[] pStat = generationStats(parDir, parRef);
+ System.out.printf("new-gen size serial=%,d recs / %.1f MB ‖%d=%,d / %.1f MB%n",
+ sStat[0], sStat[1] / 1048576.0, threads, pStat[0], pStat[1] / 1048576.0);
+ for (WideNode t : targets) {
+ System.out.println();
+ System.out.println("=== " + t.display() + " (" + t.children + " children) ===");
+ System.out.printf("%-24s %8s %9s %6s %5s %5s %6s %7s %6s %7s %7s%n",
+ "layout", "fetches", "distinct", "node", "tmpl", "map", "value", "string",
+ "list", "ideal", "union");
+ analyzeNode(t, "hot-dedup(serial)", serialDir, serialRef, cacheMB, true);
+ analyzeNode(t, "hot-dedup(‖" + threads + ")", parDir, parRef, cacheMB, false);
+ }
+ } finally {
+ deleteDir(serialDir);
+ deleteDir(parDir);
+ }
+ }
+
+ /**
+ * Read/size Pareto frontier as a function of the dedup <b>window size</b> (the existing
+ * {@link RecordRepacker#withDedupWindow(int)} knob), on the parallel layout. A smaller window
+ * forgets shared records sooner, so a wide/bucketed node's aggregates re-emit their strings
+ * locally (denser reads) at the cost of a larger store. Reports, per window, the store-wide new-gen
+ * size and the top-K widest nodes' list-widest fetches - so we can see how much plain windowing
+ * recovers before deciding whether a selective (look-ahead) aggregate duplication is needed.
+ * Windows via {@code -Doak.repack.windowSweep} (comma-separated; -1 auto, 0 exact-unbounded,
+ * >0 fixed). Additive and test-only.
+ */
+ @Test
+ void analyzeDedupWindowSweep(@TempDir Path tempDir) throws Exception {
+ File sample = getSampleStore();
+
+ File prepared = tempDir.resolve("prepared").toFile();
+ prepareStoreWithGarbage(sample, prepared);
+ int expectedRecords = (int) Math.min(Integer.MAX_VALUE, countStoreContent(prepared)[0]);
+
+ int cacheMB = Integer.getInteger("oak.repack.readCacheMB", 32);
+ int threads = Integer.getInteger("oak.repack.readThreads", 16);
+ int splitTarget = Integer.getInteger("oak.repack.readSplitTarget", 2048);
+ int topK = Integer.getInteger("oak.repack.topWidest", 3);
+ int[] windows = parseCacheSizes(System.getProperty("oak.repack.windowSweep",
+ "-1,1000000,200000,50000"));
+
+ // Serial reference (auto window) - the read locality target.
+ File serialDir = tempDir.resolve("serial").toFile();
+ long[] serialRef = produceRepackLayout(prepared, serialDir, RecordRepacker.Mode.NODE_DATA,
+ 1, 0, expectedRecords);
+ List<WideNode> targets = discoverTopWidest(serialDir, serialRef, topK);
+ long[] sStat = generationStats(serialDir, serialRef);
+ System.out.println("--- dedup-window sweep (‖" + threads + ", cache " + cacheMB
+ + " MB): read/size frontier ---");
+ System.out.printf("serial(ref) new-gen %,d recs / %.1f MB%n", sStat[0], sStat[1] / 1048576.0);
+ for (WideNode t : targets) {
+ System.out.printf(" serial %-40s fetches=%d%n", t.display(),
+ listWidestFetches(serialDir, serialRef, t, cacheMB));
+ }
+ deleteDir(serialDir);
+
+ for (int window : windows) {
+ File dir = tempDir.resolve("win-" + window).toFile();
+ long[] ref = produceRepackLayout(prepared, dir, RecordRepacker.Mode.NODE_DATA,
+ threads, splitTarget, expectedRecords, window);
+ try {
+ long[] gStat = generationStats(dir, ref);
+ System.out.printf("window=%-9s new-gen %,d recs / %.1f MB%n",
+ window == -1 ? "auto" : window == 0 ? "exact" : Integer.toString(window),
+ gStat[0], gStat[1] / 1048576.0);
+ for (WideNode t : targets) {
+ System.out.printf(" ‖%d %-40s fetches=%d%n", threads, t.display(),
+ listWidestFetches(dir, ref, t, cacheMB));
+ }
+ } finally {
+ deleteDir(dir);
+ }
+ }
+ }
+
+ /** Cold list-widest fetch count for one node against one layout. */
+ private long listWidestFetches(File dir, long[] ref, WideNode target, int cacheMB) throws Exception {
+ SegmentFetchTracker tracker = new SegmentFetchTracker();
+ try (ReadOnlyFileStore store = fileStoreBuilder(dir)
+ .withSegmentCacheSize(cacheMB).withIOMonitor(tracker).buildReadOnly()) {
+ listImmediateChildren(resolvePath(readContentRoot(store, ref), target.path));
+ }
+ return tracker.fetches.get();
+ }
+
+ /** A wide node found during discovery: its path under the content root and its child count. */
+ private static final class WideNode {
+ final List<String> path;
+ final int children;
+
+ WideNode(List<String> path, int children) {
+ this.path = path;
+ this.children = children;
+ }
+
+ String display() {
+ return path.isEmpty() ? "/" : "/" + String.join("/", path);
+ }
+ }
+
+ private List<WideNode> topWidest;
+
+ /** One streaming O(depth) walk retaining the {@code k} nodes with the most immediate children. */
+ private List<WideNode> discoverTopWidest(File dir, long[] ref, int k) throws Exception {
+ topWidest = new ArrayList<>();
+ try (ReadOnlyFileStore store = fileStoreBuilder(dir).buildReadOnly()) {
+ discoverWide(readContentRoot(store, ref), new ArrayDeque<>(), k);
+ }
+ topWidest.sort((a, b) -> Integer.compare(b.children, a.children));
+ return topWidest;
+ }
+
+ private void discoverWide(NodeState node, ArrayDeque<String> path, int k) {
+ int children = (int) node.getChildNodeCount(Long.MAX_VALUE);
+ if (topWidest.size() < k || children > topWidest.get(topWidest.size() - 1).children) {
+ topWidest.add(new WideNode(new ArrayList<>(path), children));
+ topWidest.sort((a, b) -> Integer.compare(b.children, a.children));
+ if (topWidest.size() > k) {
+ topWidest.remove(topWidest.size() - 1);
+ }
+ }
+ for (ChildNodeEntry entry : node.getChildNodeEntries()) {
+ path.addLast(entry.getName());
+ discoverWide(entry.getNodeState(), path, k);
+ path.removeLast();
+ }
+ }
+
+ /** Attribute one wide node's list-widest footprint against one already-produced layout. */
+ private void analyzeNode(WideNode target, String label, File dir, long[] ref, int cacheMB,
+ boolean printRecords) throws Exception {
+ SegmentFetchTracker tracker = new SegmentFetchTracker();
+ try (ReadOnlyFileStore store = fileStoreBuilder(dir)
+ .withSegmentCacheSize(cacheMB).withIOMonitor(tracker).buildReadOnly()) {
+ listImmediateChildren(resolvePath(readContentRoot(store, ref), target.path));
+ }
+ try (ReadOnlyFileStore store = fileStoreBuilder(dir).buildReadOnly()) {
+ RecordId nodeId = ((SegmentNodeState) resolvePath(readContentRoot(store, ref), target.path))
+ .getRecordId();
+ ListingSegCollector c = new ListingSegCollector(store.getReader());
+ c.collect(nodeId);
+ int ideal = (int) Math.ceil(c.unionMB() / (256.0 / 1024.0));
+ System.out.printf("%-24s %8d %9d %6d %5d %5d %6d %7d %6d %7d %7d%n",
+ label, tracker.fetches.get(), tracker.distinct.size(),
+ c.count("node"), c.count("template"), c.count("map"), c.count("value"),
+ c.count("string"), c.count("list"), ideal, c.all.size());
+ if (printRecords) {
+ System.out.printf(" records touched: node=%d template=%d map=%d property=%d "
+ + "string=%d list=%d ; union=%.1f MB (ideal %d segs)%n",
+ c.records("node"), c.records("template"), c.records("map"),
+ c.records("property"), c.records("string"), c.records("list"),
+ c.unionMB(), ideal);
+ }
+ }
+ }
+
+ private static NodeState readContentRoot(ReadOnlyFileStore store, long[] ref) {
+ SegmentId sid = store.getSegmentIdProvider().newSegmentId(ref[0], ref[1]);
+ return store.getReader().readNode(new RecordId(sid, (int) ref[2])).getChildNode("root");
+ }
+
+ private static String segKey(SegmentId id) {
+ return id.getMostSignificantBits() + "," + id.getLeastSignificantBits();
+ }
+
+ /** Like {@link LatencyIOMonitor} but also records the set of distinct segments fetched. */
+ private static final class SegmentFetchTracker implements IOMonitor {
+ final AtomicLong fetches = new AtomicLong();
+ final Set<String> distinct = ConcurrentHashMap.newKeySet();
+
+ @Override
+ public void beforeSegmentRead(File file, long msb, long lsb, int length) {
+ fetches.incrementAndGet();
+ distinct.add(msb + "," + lsb);
+ }
+
+ @Override
+ public void afterSegmentRead(File file, long msb, long lsb, int length, long elapsed) {
+ }
+
+ @Override
+ public void beforeSegmentWrite(File file, long msb, long lsb, int length) {
+ }
+
+ @Override
+ public void afterSegmentWrite(File file, long msb, long lsb, int length, long elapsed) {
+ }
+ }
+
+ /**
+ * Counts distinct data-segment records reachable from a node by reference: node + stable-id
+ * block, template, map, list, bucket, string and inlined blob-header records. A single dedup set
+ * shared across all record types means each physical record is counted exactly once, which also
+ * folds a degenerate single-element list into its value record. Bulk binary block data is not
+ * visited (shared, excluded), matching the "records" metric used by the repacker benchmark.
+ */
+ private static final class RecordCounter extends SegmentParser {
+ private final Map<String, BitSet> seen = new HashMap<>();
+ long nodes, stableIdBlocks, templates, maps, lists, buckets, strings, blobs;
+
+ RecordCounter(SegmentReader reader) {
+ super(reader);
+ }
+
+ void count(RecordId nodeId) {
+ onNode(null, nodeId);
+ }
+
+ long total() {
+ return nodes + stableIdBlocks + templates + maps + lists + buckets + strings + blobs;
+ }
+
+ String breakdown() {
+ return String.format(
+ "nodes %,d, stable-id blocks %,d, templates %,d, maps %,d, lists %,d, "
+ + "buckets %,d, strings %,d, blobs %,d",
+ nodes, stableIdBlocks, templates, maps, lists, buckets, strings, blobs);
+ }
+
+ /** @return {@code true} if {@code id} was already seen; otherwise marks it seen and returns {@code false}. */
+ private boolean seen(RecordId id) {
+ BitSet bs = seen.computeIfAbsent(segKey(id.getSegmentId()), k -> new BitSet());
+ int rn = id.getRecordNumber();
+ if (bs.get(rn)) {
+ return true;
+ }
+ bs.set(rn);
+ return false;
+ }
+
+ @Override
+ protected void onNode(RecordId parentId, RecordId nodeId) {
+ if (seen(nodeId)) {
+ return;
+ }
+ nodes++;
+ // Stable-id block (node id-slot 0). A self-marker points at the node itself (no record).
+ RecordId stableId = nodeId.getSegment().readRecordId(nodeId.getRecordNumber(), 0, 0);
+ if (!stableId.equals(nodeId) && !seen(stableId)) {
+ stableIdBlocks++;
+ }
+ parseNode(nodeId);
+ }
+
+ @Override
+ protected void onTemplate(RecordId parentId, RecordId templateId) {
+ if (seen(templateId)) {
+ return;
+ }
+ templates++;
+ parseTemplate(templateId);
+ }
+
+ @Override
+ protected void onMapLeaf(RecordId parentId, RecordId mapId, MapRecord map) {
+ if (seen(mapId)) {
+ return;
+ }
+ maps++;
+ parseMapLeaf(mapId, map);
+ }
+
+ @Override
+ protected void onMapBranch(RecordId parentId, RecordId mapId, MapRecord map) {
+ if (seen(mapId)) {
+ return;
+ }
+ maps++;
+ parseMapBranch(mapId, map);
+ }
+
+ @Override
+ protected void onMapDiff(RecordId parentId, RecordId mapId, MapRecord map) {
+ if (seen(mapId)) {
+ return;
+ }
+ maps++;
+ parseMapDiff(mapId, map);
+ }
+
+ @Override
+ protected void onList(RecordId parentId, RecordId listId, int count) {
+ if (seen(listId)) {
+ return;
+ }
+ lists++;
+ parseList(parentId, listId, count);
+ }
+
+ @Override
+ protected void onListBucket(RecordId parentId, RecordId listId, int index, int count, int capacity) {
+ // The top-level call passes the LIST record id (already counted in onList); only the
+ // recursive sub-bucket calls carry genuine BUCKET record ids.
+ if (seen(listId)) {
+ return;
+ }
+ buckets++;
+ parseListBucket(listId, index, count, capacity);
+ }
+
+ @Override
+ protected void onString(RecordId parentId, RecordId stringId) {
+ if (seen(stringId)) {
+ return;
+ }
+ strings++;
+ parseString(stringId);
+ }
+
+ @Override
+ protected void onBlob(RecordId parentId, RecordId blobId) {
+ if (seen(blobId)) {
+ return;
+ }
+ blobs++;
+ parseBlob(blobId);
+ }
+ }
+
+ /**
+ * Walks exactly the records a {@code list-widest} read touches - the widest node, its child map,
+ * and each immediate child's template + properties + property values, but NOT the children's own
+ * children - and buckets the distinct segments touched by record type. This attributes the
+ * list-widest footprint (node vs template vs value/string/list/...) so the parallel-dedup
+ * blow-up can be pinned to a specific record layer. Segment sets are cache-independent.
+ */
+ private static final class ListingSegCollector extends SegmentParser {
+ private final Map<String, Set<String>> byType = new HashMap<>();
+ private final Set<String> all = new HashSet<>();
+ // distinct records (not segments) + their byte sizes, per type. Answers: if the records this
+ // read touches were packed contiguously, how few segments could they occupy (bytes / 256 KB)?
+ private final Map<String, Set<String>> recsByType = new HashMap<>();
+ private final Map<String, Long> bytesByType = new HashMap<>();
+ private final Set<String> seenRecords = new HashSet<>();
+ private long unionBytes;
+ private int depth;
+
+ ListingSegCollector(SegmentReader reader) {
+ super(reader);
+ }
+
+ void collect(RecordId widestNodeId) {
+ onNode(null, widestNodeId);
+ }
+
+ int count(String type) {
+ Set<String> s = byType.get(type);
+ return s == null ? 0 : s.size();
+ }
+
+ int records(String type) {
+ Set<String> s = recsByType.get(type);
+ return s == null ? 0 : s.size();
+ }
+
+ double mb(String type) {
+ return bytesByType.getOrDefault(type, 0L) / (1024.0 * 1024.0);
+ }
+
+ double unionMB() {
+ return unionBytes / (1024.0 * 1024.0);
+ }
+
+ private void add(String type, RecordId id) {
+ String key = segKey(id.getSegmentId());
+ byType.computeIfAbsent(type, k -> new HashSet<>()).add(key);
+ all.add(key);
+ }
+
+ /** Record one distinct record's byte size once (deduped by record id across the whole walk). */
+ private void noteRecord(String type, RecordId id, long size) {
+ String rk = segKey(id.getSegmentId()) + "#" + id.getRecordNumber();
+ if (seenRecords.add(rk)) {
+ recsByType.computeIfAbsent(type, k -> new HashSet<>()).add(rk);
+ bytesByType.merge(type, size, Long::sum);
+ unionBytes += size;
+ }
+ }
+
+ @Override
+ protected void onNode(RecordId parentId, RecordId nodeId) {
+ add("node", nodeId);
+ if (depth >= 2) {
+ return; // parent (depth 0->1) and its children (1->2); grandchildren are not read
+ }
+ depth++;
+ NodeInfo info = parseNode(nodeId);
+ depth--;
+ noteRecord("node", nodeId, info.size);
+ }
+
+ @Override
+ protected void onTemplate(RecordId parentId, RecordId templateId) {
+ add("template", templateId);
+ TemplateInfo info = parseTemplate(templateId); // property-name / primary-type strings
+ noteRecord("template", templateId, info.size);
+ }
+
+ @Override
+ protected void onMap(RecordId parentId, RecordId mapId, MapRecord map) {
+ if (depth >= 2) {
+ return; // don't descend a child's own child map (grandchildren)
+ }
+ add("map", mapId);
+ MapInfo info = parseMap(parentId, mapId, map);
+ noteRecord("map", mapId, info.size);
+ }
+
+ @Override
+ protected void onMapLeaf(RecordId parentId, RecordId mapId, MapRecord map) {
+ add("map", mapId);
+ MapInfo info = parseMapLeaf(mapId, map);
+ noteRecord("map", mapId, info.size);
+ }
+
+ @Override
+ protected void onMapBranch(RecordId parentId, RecordId mapId, MapRecord map) {
+ add("map", mapId);
+ MapInfo info = parseMapBranch(mapId, map);
+ noteRecord("map", mapId, info.size);
+ }
+
+ @Override
+ protected void onProperty(RecordId parentId, RecordId propertyId, PropertyTemplate template) {
+ add("value", propertyId);
+ PropertyInfo info = parseProperty(parentId, propertyId, template);
+ noteRecord("property", propertyId, info.size);
+ }
+
+ @Override
+ protected void onValue(RecordId parentId, RecordId valueId, Type<?> type) {
+ add("value", valueId);
+ parseValue(parentId, valueId, type);
+ }
+
+ @Override
+ protected void onString(RecordId parentId, RecordId stringId) {
+ add("string", stringId);
+ BlobInfo info = parseString(stringId);
+ noteRecord("string", stringId, info.size);
+ }
+
+ @Override
+ protected void onBlob(RecordId parentId, RecordId blobId) {
+ add("blob", blobId);
+ }
+
+ @Override
+ protected void onList(RecordId parentId, RecordId listId, int count) {
+ add("list", listId);
+ ListInfo info = parseList(parentId, listId, count);
+ noteRecord("list", listId, info.size);
+ }
+
+ @Override
+ protected void onListBucket(RecordId parentId, RecordId listId, int index, int count,
+ int capacity) {
+ add("list", listId);
+ ListBucketInfo info = parseListBucket(listId, index, count, capacity);
+ noteRecord("list", listId, info.size);
+ }
+ }
+
+ @FunctionalInterface
+ private interface LayoutProducer {
+ /** Copy the prepared store into {@code dir}, compact/repack it, and return the new root ref. */
+ long[] produce(File dir) throws Exception;
+ }
+
+ private void measureLayout(String label, File dir, File prepared, LayoutProducer producer,
+ int[] cacheSizes, long fetchLatencyNanos, long byteLatencyNanos) throws Exception {
+ if (!layoutSelected(label)) {
+ return;
+ }
+ long[] ref = producer.produce(dir);
+ try {
+ if (widestPath == null) {
+ discoverTargets(dir, ref);
+ System.out.println(" (widest node /root/" + String.join("/", widestPath) + " with "
+ + widestChildCount + " children; deepest path depth " + deepestPath.size() + ")");
+ }
+ List<String> widest = widestPath;
+ List<String> deepest = deepestPath;
+ printWorkload(label, "full-dfs", dir, ref, cacheSizes, fetchLatencyNanos, byteLatencyNanos,
+ RecordRepackerTest::traverseDepthFirst);
+ printWorkload(label, "list-widest", dir, ref, cacheSizes, fetchLatencyNanos, byteLatencyNanos,
+ content -> listImmediateChildren(resolvePath(content, widest)));
+ printWorkload(label, "subtree-widest", dir, ref, cacheSizes, fetchLatencyNanos, byteLatencyNanos,
+ content -> traverseDepthFirst(resolvePath(content, widest)));
+ printWorkload(label, "deep-path", dir, ref, cacheSizes, fetchLatencyNanos, byteLatencyNanos,
+ content -> readAlongPath(content, deepest));
+ } finally {
+ deleteDir(dir);
+ }
+ }
+
+ private void printWorkload(String label, String workload, File dir, long[] ref, int[] cacheSizes,
+ long fetchLatencyNanos, long byteLatencyNanos, ReadWorkload workloadFn) throws Exception {
+ for (int cacheMB : cacheSizes) {
+ long[] counts = runReadWorkload(dir, ref, cacheMB, workloadFn);
+ long fetches = counts[0];
+ long bytes = counts[1];
+ long distinct = counts[2];
+ double simMillis = (fetches * fetchLatencyNanos + bytes * byteLatencyNanos) / 1_000_000.0;
+ String row = String.format("%-30s %-14s %8d %12d %12d %12.1f %12.1f%n",
+ label, workload, cacheMB, fetches, distinct, bytes / (1024.0 * 1024.0), simMillis);
+ System.out.print(row);
+ appendResultFile(row);
+ }
+ }
+
+ /**
+ * Optional comma-separated substring filter ({@code -Doak.repack.readLayouts}) selecting which layouts
+ * to produce and measure; all layouts when unset. Lets a run target e.g. only the semantic scenarios.
+ */
+ private static boolean layoutSelected(String label) {
+ String filter = System.getProperty("oak.repack.readLayouts");
+ if (filter == null || filter.isBlank()) {
+ return true;
+ }
+ for (String token : filter.split(",")) {
+ if (!token.isBlank() && label.contains(token.trim())) {
+ return true;
+ }
+ }
+ return false;
+ }
+
+ /** Parse a comma-separated list of segment-cache sizes (MB) for the read-locality cache sweep. */
+ private static int[] parseCacheSizes(String spec) {
+ String[] parts = spec.split(",");
+ int[] sizes = new int[parts.length];
+ for (int i = 0; i < parts.length; i++) {
+ sizes[i] = Integer.parseInt(parts[i].trim());
+ }
+ return sizes;
+ }
+
+ /**
+ * Traverse the whole content tree ({@code /root}) of the store at {@code dir} from a cold
+ * segment cache, forcing every node, template and value record to be read, and return
+ * {@code [segmentFetches, bytesFetched]}. A fresh {@link ReadOnlyFileStore} with a small segment
+ * cache and a counting {@link IOMonitor} is opened per call, so the counts reflect exactly the
+ * archive reads (cache misses) this one traversal incurs.
+ */
+ private long[] runReadWorkload(File dir, long[] rootRef, int cacheMB, ReadWorkload workload)
+ throws Exception {
+ LatencyIOMonitor monitor = new LatencyIOMonitor();
+ try (ReadOnlyFileStore store = fileStoreBuilder(dir)
+ .withSegmentCacheSize(cacheMB)
+ .withIOMonitor(monitor)
+ .buildReadOnly()) {
+ SegmentId sid = store.getSegmentIdProvider().newSegmentId(rootRef[0], rootRef[1]);
+ NodeState superRoot = store.getReader().readNode(new RecordId(sid, (int) rootRef[2]));
+ workload.read(superRoot.getChildNode("root"));
+ }
+ return new long[] {monitor.fetches.get(), monitor.bytes.get(), monitor.distinctSegs.size()};
+ }
+
+ @FunctionalInterface
+ private interface ReadWorkload {
+ void read(NodeState content) throws Exception;
+ }
+
+ private List<String> widestPath;
+ private List<String> deepestPath;
+ private int widestChildCount;
+
+ /**
+ * One warm pass over the content tree to locate the widest node (most immediate children) and a
+ * deepest path. The structure is identical across layouts, so this runs once and the discovered
+ * paths are reused (resolved by name) against every layout's cold store.
+ */
+ private void discoverTargets(File dir, long[] ref) throws Exception {
+ try (ReadOnlyFileStore store = fileStoreBuilder(dir).buildReadOnly()) {
+ SegmentId sid = store.getSegmentIdProvider().newSegmentId(ref[0], ref[1]);
+ NodeState content = store.getReader().readNode(new RecordId(sid, (int) ref[2]))
+ .getChildNode("root");
+ widestPath = new ArrayList<>();
+ deepestPath = new ArrayList<>();
+ widestChildCount = -1;
+ // Streaming depth-first walk: only the current root-to-node path is retained (O(depth)),
+ // so this scales to a many-million-node tree without holding the BFS frontier.
+ discover(content, new ArrayDeque<>());
+ }
+ }
+
+ private void discover(NodeState node, ArrayDeque<String> path) {
+ long children = node.getChildNodeCount(Long.MAX_VALUE);
+ if (children > widestChildCount) {
+ widestChildCount = (int) children;
+ widestPath = new ArrayList<>(path);
+ }
+ if (path.size() > deepestPath.size()) {
+ deepestPath = new ArrayList<>(path);
+ }
+ for (ChildNodeEntry entry : node.getChildNodeEntries()) {
+ path.addLast(entry.getName());
+ discover(entry.getNodeState(), path);
+ path.removeLast();
+ }
+ }
+
+ private static NodeState resolvePath(NodeState content, List<String> path) {
+ NodeState node = content;
+ for (String name : path) {
+ node = node.getChildNode(name);
+ }
+ return node;
+ }
+
+ /** Read a node and only its immediate children (no recursion) - the wide-listing pattern. */
+ private static void listImmediateChildren(NodeState parent) {
+ touchNode(parent);
+ for (ChildNodeEntry entry : parent.getChildNodeEntries()) {
+ touchNode(entry.getNodeState());
+ }
+ }
+
+ /** Resolve a single deep path from the content root, reading each node along the way. */
+ private static void readAlongPath(NodeState content, List<String> path) {
+ NodeState node = content;
+ touchNode(node);
+ for (String name : path) {
+ node = node.getChildNode(name);
+ touchNode(node);
+ }
+ }
+
+ private long[] produceRepackLayout(File prepared, File dir, RecordRepacker.Mode mode,
+ int concurrency, int splitTarget, int expectedRecords) throws Exception {
+ return produceRepackLayout(prepared, dir, mode, concurrency, splitTarget, expectedRecords, -1);
+ }
+
+ /**
+ * As {@link #produceRepackLayout(File, File, RecordRepacker.Mode, int, int, int)}, but pin the
+ * content-dedup cache: {@code window == -1} the default (auto-sized recency window), {@code 0} the
+ * exact off-heap cache, {@code > 0} a fixed-size window. Lets the read-locality benchmark contrast
+ * the default window layout against the exact one.
+ */
+ private long[] produceRepackLayout(File prepared, File dir, RecordRepacker.Mode mode,
+ int concurrency, int splitTarget, int expectedRecords, int window) throws Exception {
+ return produceRepackLayout(prepared, dir, mode, concurrency, splitTarget, expectedRecords,
+ window, 0);
+ }
+
+ /**
+ * As {@link #produceRepackLayout(File, File, RecordRepacker.Mode, int, int, int, int)}, but with
+ * {@code stages} pre-stages driving the staged repack ({@link RecordRepacker#withStages} via
+ * {@link #stageRoots}): {@code 1} = index-first ({@code /oak:index}, then the rest); {@code 2} =
+ * three-stage ({@code /oak:index}, then {@code /root}, then {@code /}).
+ */
+ private long[] produceRepackLayout(File prepared, File dir, RecordRepacker.Mode mode,
+ int concurrency, int splitTarget, int expectedRecords, int window, int stages)
+ throws Exception {
+ copySegmentStore(prepared, dir);
+ try (FileStore fs = fileStoreBuilder(dir).build()) {
+ SegmentNodeState head = fs.getHead();
+ RecordId root = head.getRecordId();
+ GCGeneration target = root.getSegmentId().getGcGeneration().nextFull();
+ RecordRepacker repacker = new RecordRepacker(fs, fs.getReader(),
+ fs.getSegmentIdProvider(), fs.getBlobStore(), fs.getBinariesInlineThreshold(),
+ target, mode, expectedRecords, concurrency)
+ .withSplitTarget(splitTarget).withDedupWindow(window)
+ .withStages(stageRoots(head, stages));
+ RecordId repacked = repacker.repack(root);
+ fs.flush();
+ return toRef(repacked);
+ }
+ }
+
+ /** Records + non-bulk bytes written into the generation of {@code ref} (the repacked output). */
+ private static long[] generationStats(File dir, long[] ref) throws Exception {
+ long[] stats = {0, 0};
+ try (ReadOnlyFileStore store = fileStoreBuilder(dir).buildReadOnly()) {
+ GCGeneration target = store.getSegmentIdProvider()
+ .newSegmentId(ref[0], ref[1]).getGcGeneration();
+ for (SegmentId id : store.getSegmentIds()) {
+ if (id.isDataSegmentId() && target.equals(id.getGcGeneration())) {
+ Segment segment = id.getSegment();
+ segment.forEachRecord((number, type, offset) -> stats[0]++);
+ stats[1] += segment.size();
+ }
+ }
+ }
+ return stats;
+ }
+
+ private long[] produceSemanticLayout(File prepared, File dir) throws Exception {
+ copySegmentStore(prepared, dir);
+ try (FileStore fs = fileStoreBuilder(dir).build()) {
+ SegmentNodeState head = fs.getHead();
+ GCGeneration base = head.getGcGeneration();
+ GCGeneration target = base.nextFull();
+ GCIncrement increment = new GCIncrement(base, base.nextPartial(), target);
+ SegmentWriterFactory writerFactory = generation -> defaultSegmentWriterBuilder("c")
+ .withGeneration(generation).build(fs);
+ CompactionWriter compactionWriter = new CompactionWriter(fs.getReader(),
+ fs.getBlobStore(), increment, writerFactory);
+ GCNodeWriteMonitor monitor = new GCNodeWriteMonitor(-1, GCMonitor.EMPTY);
+ CheckpointCompactor compactor = new CheckpointCompactor(GCMonitor.EMPTY,
+ new ClassicCompactor(compactionWriter, monitor));
+ CompactedNodeState compacted = compactor.compactUp(head, Canceller.newCanceller());
+ assertNotNull(compacted, "checkpoint compaction must not be cancelled");
+ compactionWriter.flush();
+ fs.flush();
+ return toRef(compacted.getRecordId());
+ }
+ }
+
+ private long[] produceParallelSemanticLayout(File prepared, File dir, int concurrency)
+ throws Exception {
+ copySegmentStore(prepared, dir);
+ try (FileStore fs = fileStoreBuilder(dir).build()) {
+ SegmentNodeState head = fs.getHead();
+ GCGeneration base = head.getGcGeneration();
+ GCGeneration target = base.nextFull();
+ GCIncrement increment = new GCIncrement(base, base.nextPartial(), target);
+ SegmentWriterFactory writerFactory = generation -> defaultSegmentWriterBuilder("c")
+ .withGeneration(generation)
+ .withWriterPool(SegmentBufferWriterPool.PoolType.THREAD_SPECIFIC)
+ .build(fs);
+ CompactionWriter compactionWriter = new CompactionWriter(fs.getReader(),
+ fs.getBlobStore(), increment, writerFactory);
+ GCNodeWriteMonitor monitor = new GCNodeWriteMonitor(-1, GCMonitor.EMPTY);
+ CheckpointCompactor compactor = new CheckpointCompactor(GCMonitor.EMPTY,
+ new ParallelCompactor(GCMonitor.EMPTY, compactionWriter, monitor, concurrency));
+ CompactedNodeState compacted = compactor.compactUp(head, Canceller.newCanceller());
+ assertNotNull(compacted, "parallel checkpoint compaction must not be cancelled");
+ compactionWriter.flush();
+ fs.flush();
+ return toRef(compacted.getRecordId());
+ }
+ }
+
+ private static double[] parseFractions(String spec) {
+ String[] parts = spec.split(",");
+ double[] out = new double[parts.length];
+ for (int i = 0; i < parts.length; i++) {
+ out[i] = Double.parseDouble(parts[i].trim());
+ }
+ return out;
+ }
+
+ private static int[] parseThresholds(String spec) {
+ String[] parts = spec.split(",");
+ int[] out = new int[parts.length];
+ for (int i = 0; i < parts.length; i++) {
+ out[i] = Integer.parseInt(parts[i].trim());
+ }
+ return out;
+ }
+
+ /**
+ * Depth-first traversal reading every node (recursive, so only the current root-to-node path is
+ * retained). An iterative stack would materialise and pin a wide node's whole child frontier
+ * (165K children on the large store), retaining ~all segments at once and exhausting the heap; the
+ * recursion holds O(depth) nodes, matching {@link #discover}.
+ */
+ private static void traverseDepthFirst(NodeState node) {
+ touchNode(node);
+ for (ChildNodeEntry entry : node.getChildNodeEntries()) {
+ traverseDepthFirst(entry.getNodeState());
+ }
+ }
+
+ /** Read every property value of {@code node}, forcing its template and value records to load. */
+ private static void touchNode(NodeState node) {
+ for (PropertyState p : node.getProperties()) {
+ Type<?> type = p.getType();
+ if (type == BINARY || type == BINARIES) {
+ // Materialise the in-segment blob-id record but do NOT resolve the external blob
+ // (length()/stream() needs a BlobStore and hits the DataStore, not the segment store,
+ // which is irrelevant to segment-read locality and unavailable on the read-only store).
+ for (int i = 0; i < p.count(); i++) {
+ p.getValue(BINARY, i);
+ }
+ } else {
+ p.getValue(type);
+ }
+ }
+ }
+
+ private static long[] toRef(RecordId id) {
+ return new long[] {id.getSegmentId().getMostSignificantBits(),
+ id.getSegmentId().getLeastSignificantBits(), id.getRecordNumber()};
+ }
+
+ private static void deleteDir(File dir) {
+ File[] files = dir.listFiles();
+ if (files != null) {
+ for (File f : files) {
+ if (f.isDirectory()) {
+ deleteDir(f);
+ } else {
+ f.delete();
+ }
+ }
+ }
+ dir.delete();
+ }
+
+ /** Counts segment fetches (archive reads = cache misses), distinct segments, and bytes read. */
+ private static final class LatencyIOMonitor implements IOMonitor {
+ final AtomicLong fetches = new AtomicLong();
+ final AtomicLong bytes = new AtomicLong();
+ // Distinct segments that missed at least once: fetches - distinctSegs.size() = re-reads
+ // forced by cache eviction (thrash), so the two columns separate footprint from re-fetch.
+ final Set<String> distinctSegs = ConcurrentHashMap.newKeySet();
+
+ @Override
+ public void beforeSegmentRead(File file, long msb, long lsb, int length) {
+ fetches.incrementAndGet();
+ bytes.addAndGet(length);
+ distinctSegs.add(msb + "," + lsb);
+ }
+
+ @Override
+ public void afterSegmentRead(File file, long msb, long lsb, int length, long elapsed) {
+ }
+
+ @Override
+ public void beforeSegmentWrite(File file, long msb, long lsb, int length) {
+ }
+
+ @Override
+ public void afterSegmentWrite(File file, long msb, long lsb, int length, long elapsed) {
+ }
+ }
+}