perf(db): refresh statistics after a tenth of the store changes

database/src/main/java/io/hstore/db/stats/Statistics.java

@@ -45,17 +45,20 @@ Feedback absorb(double sample) {
 
     private static final int SAMPLE_LIMIT = 200_000;
     private static final long REFRESH_AFTER_CHANGES = 10_000;
+    private static final long REFRESH_FRACTION = 10;
 
     private final StorageEngine engine;
     private final Map<String, Feedback> feedback = new ConcurrentHashMap<>();
     private final AtomicLong changesSinceRefresh = new AtomicLong();
+    private final AtomicLong refreshes = new AtomicLong();
+    private volatile long refreshAfter = REFRESH_AFTER_CHANGES;
     private final ChangeFeed.Subscription subscription;
     private volatile Catalog current;
 
     public Statistics(StorageEngine engine) {
         this.engine = engine;
         this.subscription = engine.feed().subscribe(engine.transactions().current().id(), event -> {
-            if (changesSinceRefresh.addAndGet(event.members().size() + event.slots().size()) > REFRESH_AFTER_CHANGES) {
+            if (changesSinceRefresh.addAndGet(event.members().size() + event.slots().size()) > refreshAfter) {
                 refresh();
             }
         }, (_, _) -> refresh());
@@ -87,14 +90,25 @@ public Catalog refresh() {
             Map<Integer, Long> types = snapshot.scan(EngineSlots.TYPES).stream()
                     .collect(Collectors.toMap(entry -> (int) entry.key(), entry -> entry.value().size(), Long::sum, TreeMap::new));
             long edges = cardinalities.length * stride;
-            Catalog refreshed = new Catalog(snapshot.generation(), total, edges, sampled, Histogram.of(cardinalities),
+            Histogram cardinality = Histogram.of(cardinalities);
+            Catalog refreshed = new Catalog(snapshot.generation(), total, edges, sampled, cardinality,
                     Histogram.of(degrees), byCardinality, types);
             current = refreshed;
+            refreshAfter = Math.max(REFRESH_AFTER_CHANGES, (total + cardinality.sum() * stride) / REFRESH_FRACTION);
             changesSinceRefresh.set(0);
+            refreshes.incrementAndGet();
             return refreshed;
         }
     }
 
+    long refreshes() {
+        return refreshes.get();
+    }
+
+    long acknowledged() {
+        return subscription.acknowledged();
+    }
+
     public static long degree(View view, long atom) {
         return view.degree(atom);
     }

database/src/test/java/io/hstore/db/stats/StatisticsTest.java

@@ -0,0 +1,113 @@
+// SPDX-FileCopyrightText: Metacog Labs
+// SPDX-License-Identifier: PolyForm-Noncommercial-1.0.0
+
+package io.hstore.db.stats;
+
+import io.hstore.engine.EngineOptions;
+import io.hstore.engine.StorageEngine;
+import io.hstore.engine.topology.EdgeKind;
+import io.hstore.engine.topology.Incidence;
+import io.hstore.engine.txn.Durability;
+
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
+
+import java.nio.file.Path;
+import java.util.stream.LongStream;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+class StatisticsTest {
+
+    static final int EDGE = 2;
+    static final int MEMBERS = 8;
+
+    @TempDir
+    Path directory;
+
+    static EngineOptions options() {
+        return EngineOptions.defaults().withPageSize(4096).withHistoryLimit(64).withDurability(Durability.ASYNC);
+    }
+
+    @Test
+    void aSmallStoreGetsStatisticsFromItsFirstChanges() throws InterruptedException {
+        try (StorageEngine engine = StorageEngine.open(directory, options());
+             Statistics statistics = new Statistics(engine)) {
+            long[] next = {1L << 40};
+            long changes = load(engine, next, 5, 100);
+            caughtUp(engine, statistics);
+            assertTrue(changes < 10_000, "changes " + changes);
+            assertEquals(0, statistics.refreshes());
+            changes += load(engine, next, 10, 100);
+            caughtUp(engine, statistics);
+            assertTrue(changes > 10_000, "changes " + changes);
+            assertEquals(1, statistics.refreshes());
+            assertEquals(1_200, statistics.catalog().edges());
+            assertEquals(1, statistics.refreshes());
+        }
+    }
+
+    @Test
+    void aLargeStoreWaitsForATenthOfItsSizeToChange() throws InterruptedException {
+        try (StorageEngine engine = StorageEngine.open(directory, options());
+             Statistics statistics = new Statistics(engine)) {
+            long[] next = {1L << 40};
+            load(engine, next, 50, 1_000);
+            caughtUp(engine, statistics);
+            Statistics.Catalog catalog = statistics.refresh();
+            long refreshes = statistics.refreshes();
+            long size = catalog.atoms() + catalog.cardinality().sum();
+            assertEquals(50_000L * (MEMBERS + 1), size);
+            long changes = 0;
+            while (changes + 1_000L * (MEMBERS + 1) <= size / 10) {
+                changes += load(engine, next, 1, 1_000);
+            }
+            caughtUp(engine, statistics);
+            assertTrue(changes > 30_000, "changes " + changes);
+            assertEquals(refreshes, statistics.refreshes());
+            load(engine, next, 2, 1_000);
+            caughtUp(engine, statistics);
+            assertEquals(refreshes + 1, statistics.refreshes());
+        }
+    }
+
+    @Test
+    void aBulkLoadRefreshesLogarithmicallyOften() throws InterruptedException {
+        try (StorageEngine engine = StorageEngine.open(directory, options());
+             Statistics statistics = new Statistics(engine)) {
+            long[] next = {1L << 40};
+            long changes = load(engine, next, 150, 1_000);
+            caughtUp(engine, statistics);
+            long everyTenThousand = changes / 10_000;
+            long everyTenth = 10 + Math.round(Math.ceil(Math.log(changes / 100_000.0) / Math.log(1.1)));
+            String message = statistics.refreshes() + " refreshes for " + changes + " changes";
+            assertTrue(statistics.refreshes() <= everyTenth, message);
+            assertTrue(statistics.refreshes() < everyTenThousand / 4, message);
+        }
+    }
+
+    private static long load(StorageEngine engine, long[] next, int transactions, int edges) {
+        for (int t = 0; t < transactions; t++) {
+            engine.write(txn -> {
+                for (int e = 0; e < edges; e++) {
+                    long edge = txn.createEdge(EDGE, EdgeKind.SET);
+                    long first = next[0];
+                    next[0] += MEMBERS;
+                    txn.load(edge, LongStream.range(first, first + MEMBERS).mapToObj(Incidence::of).toList());
+                }
+                return null;
+            });
+        }
+        return (long) transactions * edges * (MEMBERS + 1);
+    }
+
+    private static void caughtUp(StorageEngine engine, Statistics statistics) throws InterruptedException {
+        long generation = engine.transactions().current().id();
+        long deadline = System.currentTimeMillis() + 10_000;
+        while (statistics.acknowledged() < generation && System.currentTimeMillis() < deadline) {
+            Thread.sleep(10);
+        }
+        assertEquals(generation, statistics.acknowledged());
+    }
+}

docs/database/planner.md

@@ -203,7 +203,8 @@ The corrections are process-local and visible in `STATS` as `planner feedback <k
 
 ### Catalog statistics
 
-`Statistics.refresh` (also triggered after 10,000 member/slot changes from the change feed) scans the catalog
+`Statistics.refresh` (also triggered from the change feed once the member/slot changes since the last refresh pass
+10,000 or a tenth of the store, whichever is larger) scans the catalog
 snapshot and builds cardinality and degree histograms (`Histogram`, 65 power-of-two buckets), counts of edges by
 exact cardinality (`Count_k`) and per-`(tenant, type)` counts. Above `SAMPLE_LIMIT` = 200,000 atoms, only atoms
 with `id % stride == 0` are sampled (`stride = ceil(total / 200000)`); the edge total and every per-cardinality count

docs/database/views-signals-stats.md

@@ -151,8 +151,17 @@ record Catalog(long generation, long atoms, long edges, boolean sampled, Histogr
 | `edges` | `#sampled edges × stride` |
 | `typeCounts` | posting count per `typeKey(tenant, type)` from the type index (exact) |
 
-Refresh happens on first use, on `STATS`, and from a change-feed subscriber whenever more than
-`REFRESH_AFTER_CHANGES` = 10,000 member and slot changes accumulated.
+Refresh happens on first use, on `STATS`, and from a change-feed subscriber once the member and slot changes since
+the last refresh exceed `max(REFRESH_AFTER_CHANGES, size / REFRESH_FRACTION)`, with `REFRESH_AFTER_CHANGES` = 10,000
+and `REFRESH_FRACTION` = 10. `size` is what the last refresh counted: `atoms` plus the memberships of all edges
+(`cardinality.sum() × stride`). These are the units the change counter uses: in the engine, creating an edge with
+`k` members is one catalog change plus `k` member changes.
+
+A new or small store refreshes after its first 10,000 changes. Once the store holds more than about 100,000 atoms and
+memberships, the subscriber waits for roughly a tenth of it to change. A bulk load of `N` changes then triggers
+`O(log N)` full scans instead of `N / 10,000`, and the total scan work grows linearly with the load instead of with
+its square. The cost is that statistics on a large store can lag by up to a tenth of its size. `STATS` always
+refreshes first.
 
 ### Bounded-error sampling
 

docs/storage/change-feed.md

@@ -193,7 +193,7 @@ stateDiagram-v2
 |---|---|---|
 | `SemanticPlane` (`database/.../semantic/SemanticPlane.java`) | the generation of its persisted vector index if `covers(persisted)` is true, otherwise it rebuilds from a snapshot and subscribes from that snapshot's generation | For `main`-branch events, upserts or removes HNSW entries for every `Embedding.EMBEDDINGS` slot change, compacts an index whose tombstone ratio exceeds 0.3, and advances `indexedGeneration` for every event. `FRESH` similarity queries wait (bounded by 5 s) for `indexedGeneration >= min(snapshot generation, feed.lastGeneration())`; because the feed is appended before `current` advances, and empty commits have no feed entry, this target is always reached promptly. Holds the feed at its persisted generation. |
 | `MaterializedViews` (`database/.../view/MaterializedViews.java`) | `current` at database open | For `main`-branch events with member changes, refreshes every `CONTINUOUS` view. `refresh` itself uses `replay(descriptor.lastGeneration())` up to `lastGeneration()` and applies degree or cardinality deltas; `staleness` counts pending events; `ACTIVITY` views are built from `replay(0)`, i.e. from the oldest **retained** event. Refresh commits only write view slots (slot changes, no member changes), so they do not retrigger the subscriber. Holds the feed at the minimum `lastGeneration` over all view descriptors. |
-| `Statistics` (`database/.../stats/Statistics.java`) | `current` at database open | Counts member and slot changes and recomputes the planner's cardinality/degree histograms after more than 10,000 changes. |
+| `Statistics` (`database/.../stats/Statistics.java`) | `current` at database open | Counts member and slot changes and recomputes the planner's cardinality/degree histograms once they exceed 10,000 or a tenth of the store's atoms and memberships at the last refresh, whichever is larger. |
 | `HISTORY` statement (`database/.../query/Executor.java`) | n/a (one-shot `replay`) | Lists the last N events: generation, transaction, time, branch, number of member and slot changes. Only retained events can be listed. |
 
 Signals (`database/.../signal/Signals.java`) and the temporal layer read trees and snapshots directly and do not consume the feed. `Temporal.events(CommitEvent)` converts an event into per-edge `TopologyEvent`s but is not currently called.