venkat1701 · Oct 11, 2026 · Oct 5, 2026 · Oct 5, 2026 · Oct 5, 2026 · Oct 5, 2026
diff --git a/benchmarks/run.sh b/benchmarks/run.sh
@@ -9,10 +9,18 @@ heap="${HEAP:-4g}"
 history="${HISTORY:-64}"
 runs="${RUNS:-3}"
 cache="${CACHE_MB:-0}"
+segment="${PAGES_PER_SEGMENT:-0}"
+ratio="${COMPACTION_LIVE_RATIO:-0.5}"
 results="results/scale-$scale-$durability-history-$history"
 if [ "$cache" != 0 ]; then
 	results="$results-cache-$cache"
 fi
+if [ "$segment" != 0 ]; then
+	results="$results-segment-$segment"
+fi
+if [ "$ratio" != 0.5 ]; then
+	results="$results-ratio-$ratio"
+fi
 
 ../mvnw -q -f ../pom.xml -Pbenchmarks -DskipTests install
 classpath="target/classes:$(cat target/classpath.txt)"
@@ -26,7 +34,7 @@ rm -rf "$results" && mkdir -p "$results"
 for run in $(seq 1 "$runs"); do
 	for store in hstore hypergraphdb; do
 		java "${jvm[@]}" io.hstore.bench.Comparison run --store "$store" --scale "$scale" --durability "$durability" \
-			--threads "$threads" --history "$history" --cache-mb "$cache" --out "$results/$store-$run.json"
+			--threads "$threads" --history "$history" --cache-mb "$cache" --pages-per-segment "$segment" --compaction-live-ratio "$ratio" --out "$results/$store-$run.json"
 	done
 done
 java "${jvm[@]}" io.hstore.bench.Comparison report "$results"/hstore-*.json "$results"/hypergraphdb-*.json | tee "$results/report.md"
diff --git a/benchmarks/src/main/java/io/hstore/bench/Comparison.java b/benchmarks/src/main/java/io/hstore/bench/Comparison.java
@@ -1,6 +1,7 @@
 package io.hstore.bench;
 
 import io.hstore.db.value.Json;
+import io.hstore.engine.EngineOptions;
 
 import java.io.BufferedReader;
 import java.io.IOException;
@@ -21,6 +22,7 @@
 import java.util.function.IntToLongFunction;
 import java.util.function.LongSupplier;
 import java.util.function.Supplier;
+import java.util.concurrent.atomic.AtomicReference;
 import java.util.concurrent.locks.LockSupport;
 
 public final class Comparison {
@@ -70,6 +72,9 @@ private interface Slice {
     private static final int MIXED_BATCH = 10;
     private static final int MIXED_UPDATES_PER_SECOND = 1_000;
     private static final int SAMPLES_PER_THREAD = 200_000;
+    private static final long COMPACTION_WINDOW_NANOS = 10_000_000_000L;
+    private static final long COMPACTION_DELAY_NANOS = 1_000_000_000L;
+    private static final long COMPACTION_INTERVAL_NANOS = 20_000_000L;
     private static final long SEED = 20_260_904L;
 
     private Comparison() {
@@ -83,7 +88,7 @@ public static void main(String[] args) throws Exception {
             case "report" -> IO.println(Report.markdown(args));
             default -> {
                 IO.println("""
-                        usage: Comparison run --store hstore|hypergraphdb [--scale N] [--durability async|sync] [--threads N] [--history N] [--cache-mb N] --out FILE
+                        usage: Comparison run --store hstore|hypergraphdb [--scale N] [--durability async|sync] [--threads N] [--history N] [--cache-mb N] [--compaction-live-ratio R] [--pages-per-segment N] --out FILE
                                Comparison crash --store hstore|hypergraphdb --dir DIRECTORY (used by run; ingests until killed)
                                Comparison report FILE... (runs of two stores; cells show the median and range)""");
                 System.exit(2);
@@ -189,13 +194,18 @@ private static void run(Map<String, String> options) throws Exception {
                     }))));
             results.add(read("read.incidence.churned", "enumerate incidence sets after the deletes", nodes.length,
                     (from, to) -> store.incidence(nodes, from, to), true));
+            long[] churned = new long[1];
+            store.reopen(() -> churned[0] = store.diskBytes());
+            results.add(new Result("disk.churned", "bytes on disk after the deletes and a clean shutdown", churned[0], 1000.0, churned[0]));
+            results.add(compaction(store, dataset, false));
+            results.add(compaction(store, dataset, true));
             store.flush();
             totals.put("bytesWritten", number(store.bytesWritten()));
         } finally {
             store.close();
         }
-        long churned = store.diskBytes();
-        results.add(new Result("disk.churned", "bytes on disk after the deletes and a clean shutdown", churned, 1000.0, churned));
+        long compacted = store.diskBytes();
+        results.add(new Result("disk.compacted", "bytes on disk after a full compaction and a clean shutdown", compacted, 1000.0, compacted));
         results.add(recover(options, dataset));
         write(out, store, scale, sync, threads, dataset, totals, results);
         results.forEach(result -> IO.println("  %-22s %,14.0f ops/s  %,10.1f ms  checksum %d"
@@ -206,7 +216,9 @@ private static Store open(Map<String, String> options, Path directory, boolean c
         boolean sync = options.getOrDefault("durability", "async").equals("sync");
         long cacheBytes = Long.parseLong(options.getOrDefault("cache-mb", "0")) << 20;
         return switch (options.getOrDefault("store", "hstore")) {
-            case "hstore" -> new HStoreStore(directory, sync, Integer.parseInt(options.getOrDefault("history", "64")), cacheBytes, create);
+            case "hstore" -> new HStoreStore(directory, sync, Integer.parseInt(options.getOrDefault("history", "64")), cacheBytes,
+                    Double.parseDouble(options.getOrDefault("compaction-live-ratio", String.valueOf(EngineOptions.defaults().compactionLiveRatio()))),
+                    Integer.parseInt(options.getOrDefault("pages-per-segment", "0")), create);
             case "hypergraphdb" -> new HyperGraphDbStore(directory, sync, cacheBytes);
             case String other -> throw new IllegalArgumentException("unknown store " + other);
         };
@@ -228,7 +240,7 @@ private static Result recover(Map<String, String> options, Dataset dataset) thro
         command.add(ProcessHandle.current().info().command().orElse("java"));
         command.addAll(ManagementFactory.getRuntimeMXBean().getInputArguments());
         command.addAll(List.of("-cp", System.getProperty("java.class.path"), Comparison.class.getName(), "crash", "--dir", directory.toString()));
-        for (String option : List.of("store", "scale", "durability", "history", "cache-mb")) {
+        for (String option : List.of("store", "scale", "durability", "history", "cache-mb", "compaction-live-ratio", "pages-per-segment")) {
             if (options.containsKey(option)) {
                 command.addAll(List.of("--" + option, options.get(option)));
             }
@@ -375,6 +387,68 @@ private static List<Result> mixed(Store store, Dataset dataset, int threads) thr
         return List.of(read, write);
     }
 
+    private static Result compaction(Store store, Dataset dataset, boolean compact) throws Exception {
+        int[] updates = dataset.updates();
+        Reservoir commits = new Reservoir(SAMPLES_PER_THREAD, SEED);
+        AtomicReference<Throwable> failure = new AtomicReference<>();
+        long[] compactNanos = new long[1];
+        long[] diskBefore = new long[1];
+        if (compact) {
+            store.reopen(() -> diskBefore[0] = store.diskBytes());
+        }
+        long before = store.bytesWritten();
+        JvmUsage usage = JvmUsage.now();
+        long started = System.nanoTime();
+        long deadline = started + COMPACTION_WINDOW_NANOS;
+        Thread compactor = Thread.ofPlatform().name("bench-compaction").start(() -> {
+            if (!compact) {
+                return;
+            }
+            for (long wait = started + COMPACTION_DELAY_NANOS - System.nanoTime(); wait > 0; wait = started + COMPACTION_DELAY_NANOS - System.nanoTime()) {
+                LockSupport.parkNanos(wait);
+            }
+            long begin = System.nanoTime();
+            try {
+                store.compact();
+            } catch (Throwable e) {
+                failure.set(e);
+            }
+            compactNanos[0] = System.nanoTime() - begin;
+        });
+        int committed = 0;
+        for (int k = 0; ; k++) {
+            long intended = started + k * COMPACTION_INTERVAL_NANOS;
+            if (intended >= deadline && !compactor.isAlive()) {
+                break;
+            }
+            for (long wait = intended - System.nanoTime(); wait > 0; wait = intended - System.nanoTime()) {
+                LockSupport.parkNanos(wait);
+            }
+            int index = k % updates.length;
+            store.update(updates, index, index + 1, (compact ? 4_000_000 : 3_000_000) + k);
+            commits.add(System.nanoTime() - intended);
+            committed++;
+        }
+        compactor.join();
+        if (failure.get() != null) {
+            throw new IllegalStateException("compaction failed", failure.get());
+        }
+        double millis = (System.nanoTime() - started) / 1e6;
+        JvmUsage used = JvmUsage.now().since(usage);
+        store.flush();
+        String schedule = "single-update commits at %d/s".formatted(1_000_000_000L / COMPACTION_INTERVAL_NANOS);
+        Result result = new Result(compact ? "compaction" : "compaction.baseline",
+                compact ? schedule + " while a full compaction runs" : schedule + " with no compaction, for comparison",
+                committed, millis, 0).using(used).with("latency", Reservoir.merge(commits).json()).with("checked", new Json.Bool(false))
+                .with("bytesWritten", number(store.bytesWritten() - before));
+        if (!compact) {
+            return result;
+        }
+        long[] diskAfter = new long[1];
+        store.reopen(() -> diskAfter[0] = store.diskBytes());
+        return result.with("compactionMillis", number(compactNanos[0] / 1_000_000)).with("reclaimed", number(diskBefore[0] - diskAfter[0]));
+    }
+
     private static Result concurrent(Store store, Dataset dataset, int threads) throws Exception {
         int[] probes = dataset.probeNodes();
         try (ExecutorService pool = Executors.newFixedThreadPool(threads)) {

diff --git a/benchmarks/src/main/java/io/hstore/bench/HStoreStore.java b/benchmarks/src/main/java/io/hstore/bench/HStoreStore.java
@@ -38,14 +38,15 @@ final class HStoreStore implements Store {
     private final int cachedNodes;
     private final long cacheBytes;
 
-    HStoreStore(Path directory, boolean sync, int history, long cacheBytes, boolean create) {
+    HStoreStore(Path directory, boolean sync, int history, long cacheBytes, double liveRatio, int pagesPerSegment, boolean create) {
         this.directory = directory;
         this.history = history;
         this.cacheBytes = cacheBytes;
         EngineOptions engine = EngineOptions.defaults();
         this.cachedNodes = cacheBytes > 0 ? Math.toIntExact(cacheBytes / engine.pageSize()) : CACHED_NODES;
         this.options = DatabaseOptions.defaults().withEngine(engine
-                .withDurability(sync ? Durability.SYNC : Durability.ASYNC).withCachedNodes(cachedNodes).withHistoryLimit(history));
+                .withDurability(sync ? Durability.SYNC : Durability.ASYNC).withCachedNodes(cachedNodes).withHistoryLimit(history)
+                .withCompactionLiveRatio(liveRatio).withPagesPerSegment(pagesPerSegment > 0 ? pagesPerSegment : engine.pagesPerSegment()));
         this.database = HypergraphDatabase.open(directory, options);
         if (create) {
             database.write(writer -> {
@@ -215,6 +216,11 @@ public void flush() {
         database.engine().checkpoint();
     }
 
+    @Override
+    public void compact() {
+        database.engine().compact();
+    }
+
     @Override
     public long bytesWritten() {
         EngineStats stats = database.engine().stats();

diff --git a/benchmarks/src/main/java/io/hstore/bench/HyperGraphDbStore.java b/benchmarks/src/main/java/io/hstore/bench/HyperGraphDbStore.java
@@ -1,5 +1,6 @@
 package io.hstore.bench;
 
+import com.sleepycat.je.CheckpointConfig;
 import com.sleepycat.je.Durability;
 import com.sleepycat.je.Environment;
 import com.sleepycat.je.EnvironmentConfig;
@@ -242,6 +243,15 @@ public void flush() {
         environment().sync();
     }
 
+    @Override
+    public void compact() {
+        Environment environment = environment();
+        while (environment.cleanLog() > 0) {
+            Thread.onSpinWait();
+        }
+        environment.checkpoint(new CheckpointConfig().setForce(true));
+    }
+
     @Override
     public long bytesWritten() {
         EnvironmentStats stats = environment().getStats(new StatsConfig());

diff --git a/benchmarks/src/main/java/io/hstore/bench/Report.java b/benchmarks/src/main/java/io/hstore/bench/Report.java
@@ -49,7 +49,7 @@ static String markdown(String[] args) {
         out.append("|---|---:|---:|---:|:---:|%n".formatted());
         for (Map.Entry<String, Map<String, Json>> entry : subject.results().entrySet()) {
             String workload = entry.getKey();
-            if (baselines.stream().anyMatch(run -> !run.results().containsKey(workload))) {
+            if (workload.startsWith("compaction") || baselines.stream().anyMatch(run -> !run.results().containsKey(workload))) {
                 continue;
             }
             boolean disk = workload.startsWith("disk");
@@ -71,13 +71,15 @@ static String markdown(String[] args) {
         return out.toString();
     }
 
-    private record Metric(String key, String name, boolean perOperation, boolean showZero, DoubleFunction<String> format) {
+    private record Metric(String key, String name, boolean perOperation, boolean showZero, boolean higherIsBetter, DoubleFunction<String> format) {
     }
 
-    private static final List<Metric> METRICS = List.of(new Metric("bytesWritten", "bytes written per operation", true, false, Report::bytes),
-            new Metric("allocated", "heap allocated per operation", true, false, Report::bytes),
-            new Metric("gcMillis", "GC pause time", false, false, value -> "%,.0f ms".formatted(value)),
-            new Metric("lost", "acknowledged nodes lost", false, true, value -> "%,.0f".formatted(value)));
+    private static final List<Metric> METRICS = List.of(new Metric("bytesWritten", "bytes written per operation", true, false, false, Report::bytes),
+            new Metric("allocated", "heap allocated per operation", true, false, false, Report::bytes),
+            new Metric("gcMillis", "GC pause time", false, false, false, value -> "%,.0f ms".formatted(value)),
+            new Metric("lost", "acknowledged nodes lost", false, true, false, value -> "%,.0f".formatted(value)),
+            new Metric("compactionMillis", "compaction time", false, true, false, value -> "%,.0f ms".formatted(value)),
+            new Metric("reclaimed", "disk space reclaimed", false, true, true, Report::bytes));
 
     private static void resources(StringBuilder out, List<Run> subjects, List<Run> baselines) {
         List<Run> all = Stream.concat(subjects.stream(), baselines.stream()).toList();
@@ -90,7 +92,7 @@ private static void resources(StringBuilder out, List<Run> subjects, List<Run> b
                     if (mine == 0 && theirs == 0 && !metric.showZero()) {
                         continue;
                     }
-                    double ratio = theirs / mine;
+                    double ratio = metric.higherIsBetter() ? (mine > 0 && theirs > 0 ? mine / theirs : Double.NaN) : theirs / mine;
                     rows.append("| `%s` | %s | %s | %s | %s |%n".formatted(workload, metric.name(), metric.format().apply(mine),
                             metric.format().apply(theirs), Double.isFinite(ratio) ? "**%.2f×**".formatted(ratio) : "n/a"));
                 }
@@ -127,10 +129,12 @@ private static double[] header(List<Run> runs, String key) {
     }
 
     private static String bytes(double value) {
-        if (value < 1024) {
-            return "%.0f B".formatted(value);
+        double size = Math.abs(value);
+        String sign = value < 0 ? "-" : "";
+        if (size < 1024) {
+            return sign + "%.0f B".formatted(size);
         }
-        return value < 1 << 20 ? "%.1f KiB".formatted(value / 1024) : "%,.1f MiB".formatted(value / (1 << 20));
+        return sign + (size < 1 << 20 ? "%.1f KiB".formatted(size / 1024) : "%,.1f MiB".formatted(size / (1 << 20)));
     }
 
     private static void latency(StringBuilder out, List<Run> subjects, List<Run> baselines) {

diff --git a/benchmarks/src/main/java/io/hstore/bench/Store.java b/benchmarks/src/main/java/io/hstore/bench/Store.java
@@ -55,6 +55,8 @@ default void reopen() {
 
     void flush();
 
+    void compact();
+
     long bytesWritten();
 
     Path directory();

diff --git a/docs/benchmarks.md b/docs/benchmarks.md
@@ -62,6 +62,9 @@ taken from a common hyperedge (so co-membership counts are non-zero), and 20,000
 | `churn.remove` | remove one member from each of 20,000 other hyperedges, 1,000 per transaction | `Writer.remove` | `graph.replace` with a new `HGPlainLink` (links are immutable) |
 | `read.incidence.churned` | `read.incidence` after the deletes | | |
 | `disk.churned` | bytes on disk after the deletes and a clean shutdown | | |
+| `compaction.baseline` | single-update commits at 50/s for 10 s, no compaction | `Writer.set` | `graph.replace` |
+| `compaction` | the same commits while a full compaction runs, starting 1 s in | `StorageEngine.compact` | `Environment.cleanLog` until it finds nothing, then a forced checkpoint |
+| `disk.compacted` | bytes on disk after the compaction and a clean shutdown | | |
 | `recover` | open the store after the process ingesting into it was killed with SIGKILL | `HypergraphDatabase.open` | `HGEnvironment.get` |
 
 Read workloads run a warm-up pass over 10% of the probes before timing, except `read.incidence.cold`.
@@ -93,6 +96,18 @@ the loaded store after a clean shutdown. The checksum of `read.incidence.churned
 both engines applied the same deletes. `disk.churned` shows whether the space came back: through compaction for
 HStore, and the log cleaner for JE.
 
+**Compaction.** After the churn, `disk.churned` is recorded at a reopen, and then two 10 s windows run. In both, one
+writer commits a single update every 20 ms on a fixed schedule, and each commit's latency is measured from when it
+was scheduled. In `compaction`, the store also runs a full compaction starting 1 s in, and the window lasts until
+the compaction has finished. HStore calls `StorageEngine.compact()`. JE calls `cleanLog()` until it cleans nothing,
+then forces a checkpoint so the cleaned files are deleted. The latency table compares the two windows, and the
+*Resources* table adds the compaction time and the space reclaimed. Disk is measured at a reopen just before the
+window and again at a reopen just after it, so segments that HStore only deletes once older readers are gone are
+counted too. That figure can be negative when compaction frees less than the writer adds. The two windows run at a
+fixed rate, so they are left out of the main table, where their ratio would always be about 1. HStore's default segments are 256 MiB, so
+at scale 1 most data shares one or two segments. `--pages-per-segment` (`PAGES_PER_SEGMENT`) and
+`--compaction-live-ratio` (`COMPACTION_LIVE_RATIO`) change HStore's settings for a run.
+
 **Crash recovery.** `reopen` closes the store cleanly first. `recover` doesn't:
 1. The run starts a second JVM (`Comparison crash`) with the same flags and classpath. It ingests nodes into a
    fresh directory and prints the count after every acknowledged commit.