diff --git a/database/src/main/java/io/hstore/db/stats/Statistics.java b/database/src/main/java/io/hstore/db/stats/Statistics.java index c650447..69f2680 100644 --- a/database/src/main/java/io/hstore/db/stats/Statistics.java +++ b/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 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 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); } diff --git a/database/src/test/java/io/hstore/db/stats/StatisticsTest.java b/database/src/test/java/io/hstore/db/stats/StatisticsTest.java new file mode 100644 index 0000000..f437f7f --- /dev/null +++ b/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()); + } +} diff --git a/docs/database/planner.md b/docs/database/planner.md index d2d0da8..fd14a63 100644 --- a/docs/database/planner.md +++ b/docs/database/planner.md @@ -203,7 +203,8 @@ The corrections are process-local and visible in `STATS` as `planner feedback = 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.