diff --git a/docs/architecture/overview.md b/docs/architecture/overview.md index 53d6d31..bbe0f5d 100644 --- a/docs/architecture/overview.md +++ b/docs/architecture/overview.md @@ -91,7 +91,7 @@ The engine has no dependencies outside `java.base`. All three modules build into | `engine.txn` | `TransactionManager` (commit lock, append, publication, pins, history), `Transaction` (op log, read sets, rebase), `Snapshot`, `View` (read API shared by both), `Workspace` (mutable root vector plus derivations), `GroupCommitter`, `Op`/`EdgeAction`, `Derivation`/`EngineDerivations`, `CrashPoint`, `Durability`, `WalMode`, `Isolation` and `TxnOptions`. | | `engine.wal` | `WriteAheadLog` (segmented, checksummed records) and `WalRecord` (`Begin`, `Page`, `PageRef`, `Root`, `BranchMeta`, `Feed`, `Commit`, `Abort`, `Checkpoint`). | | `engine.feed` | `ChangeFeed` (segmented append-only `feed/*.feed` files, sparse generation index, retention, tailing subscriptions), `CommitEvent` and `FeedCodec`; see [change-feed.md](../storage/change-feed.md). | -| `engine.maintenance` | `Checkpointer`, `Compactor` (liveness, relocation, reclamation) and `Recovery` (catalog discovery, WAL redo, durable-prefix rule). | +| `engine.maintenance` | `Checkpointer`, `Compactor` (liveness, moving live pages out of victim segments, reclamation) and `Recovery` (catalog discovery, WAL redo, durable-prefix rule). | ### `io.hstore.db` diff --git a/docs/benchmarks.md b/docs/benchmarks.md index 87f3e80..f379651 100644 --- a/docs/benchmarks.md +++ b/docs/benchmarks.md @@ -229,7 +229,7 @@ not what drives HStore's footprint: * **Disk at scale 4 (0.54×).** Live data is comparable: about 1 GB of live node images in HStore against 1.27 GB in JE. The difference is garbage that has not been reclaimed yet. Both engines are log-structured with a 50% minimum-utilisation cleaner: HStore's `compaction_live_ratio = 0.5`, JE's - `je.cleaner.minUtilization = 50`. A sealed HStore segment is only relocated once fewer than half of its + `je.cleaner.minUtilization = 50`. A sealed HStore segment is only compacted once fewer than half of its images are live, and the background pass runs after every `checkpoint_wal_mb` of writes ([maintenance.md](storage/maintenance.md)). Raising `compaction_live_ratio` trades write amplification for space. diff --git a/docs/operations/configuration.md b/docs/operations/configuration.md index 4a07679..72bc767 100644 --- a/docs/operations/configuration.md +++ b/docs/operations/configuration.md @@ -94,7 +94,7 @@ max_connections = 200 | `cache_nodes` | `HSTORE_CACHE_NODES` | `65536` | Decoded tree nodes kept in the node cache. | | `checkpoint_wal_mb` | `HSTORE_CHECKPOINT_WAL_MB` | `256` | WAL volume since the last checkpoint that triggers a background checkpoint. | | `history_limit` | `HSTORE_HISTORY_LIMIT` | `64` | Committed generations kept addressable for `AT GENERATION`, `AS OF`, `HISTORY` and the Studio time slider. | -| `compaction_live_ratio` | `HSTORE_COMPACTION_LIVE_RATIO` | `0.5` | Sealed segments whose live bytes fall below this fraction of their allocated bytes are relocated by compaction: one segment (the emptiest) per background pass after every `checkpoint_wal_mb` of written node images, or all of them with `COMPACT`. Time-travel history is preserved; retired segments are deleted once `history_limit` has rolled past them. Must satisfy 0 < r < 1; other values fail at startup. | +| `compaction_live_ratio` | `HSTORE_COMPACTION_LIVE_RATIO` | `0.5` | Sealed segments whose live bytes fall below this fraction of their allocated bytes are emptied by compaction, which moves their live pages to the active segment: one segment (the emptiest) per background pass after every `checkpoint_wal_mb` of written node images, or all of them with `COMPACT`. Time-travel history is preserved. Emptied segments are deleted after the next checkpoint, once no reader that started before the move is still open. Must satisfy 0 < r < 1; other values fail at startup. | | `feed_retention_generations` | `HSTORE_FEED_RETENTION_GENERATIONS` | `100000` | Committed generations kept in the [change feed](../storage/change-feed.md) for `HISTORY`, `DIFF` and subscribers. After every checkpoint, whole feed segments that lie entirely below `current − N` are deleted. *Holds* lower that floor: the semantic index's last persisted generation and the generation of each continuous materialised view. Feed segments roll at the WAL segment size (`EngineOptions.walSegmentBytes`, 64 MiB). Must be ≥ 1. | ### Resource budgets diff --git a/docs/operations/logging.md b/docs/operations/logging.md index 1f74d2c..3c01209 100644 --- a/docs/operations/logging.md +++ b/docs/operations/logging.md @@ -90,8 +90,8 @@ This is every message the server emits. Placeholders are in `{braces}`. | Severity | Component | Message | |---|---|---| -| LOG | engine | `background compaction relocated segments {list}` (write-volume-triggered pass) | -| LOG | engine | `compaction relocated segments {list} and reclaimed {list}` | +| LOG | engine | `background compaction moved the live pages out of segments {list}` (write-volume-triggered pass) | +| LOG | engine | `compaction moved the live pages out of segments {list} and reclaimed {list}` | | LOG | engine | `reclaimed retired segments {list}` | | WARNING | hstore | `background maintenance failed` (with DETAIL) | | LOG | txn | `created branch {name} (#{id}) from generation {g}` | diff --git a/docs/storage/catalog-and-generations.md b/docs/storage/catalog-and-generations.md index 4d5b465..7fbc7d6 100644 --- a/docs/storage/catalog-and-generations.md +++ b/docs/storage/catalog-and-generations.md @@ -100,7 +100,7 @@ Published generations are kept in `TransactionManager.history` (a `ConcurrentSki * `snapshotAt(generation, branch)` pins the generation and opens a snapshot over its root vector. A generation that has been trimmed fails with `generation N is no longer retained`. * `generationAsOf(wallTime)` returns the newest retained generation with `wallTime <= t`. -Time travel needs no undo log. Copy-on-write never overwrites a page, so an old generation's root vector still points at valid pages, and reading the past is exactly as fast as reading the present. A historical page is only reclaimed once compaction has relocated the live data *and* both history trimming (`history_limit`) and every pinned reader have moved past the compaction generation ([maintenance.md](maintenance.md#reclaim)). Compaction therefore never shortens the time-travel window. +Time travel needs no undo log. Copy-on-write never overwrites a page, so an old generation's root vector still points at valid pages, and reading the past is exactly as fast as reading the present. Compaction moves the pages a retained generation still needs and keeps their page numbers, so old root vectors stay valid after their segments are deleted ([maintenance.md](maintenance.md#reclaim)). Compaction therefore never shortens the time-travel window. The retained history is also persisted. `Checkpointer` writes up to `historyLimit` past generations into the catalog image, and `Recovery` appends every replayed generation, so time travel survives restarts. @@ -124,7 +124,7 @@ Branches are named, independently writable lines of generations that share pages | merge | `markMerged(id)` → `closeBranch(id, MERGED)` | Sets the state to `MERGED` **and empties its roots**, so its pages are no longer reachable from the current generation. The actual three-way merge of contents happens in `db.temporal.BranchMerge` before this call. | | drop | `dropBranch(id)` → `closeBranch(id, DROPPED)` | The same, with state `DROPPED`. `Generation.branch` rejects a dropped id. | -`main` (id 0) cannot be closed. `StorageEngine.branches()` lists only `ACTIVE` branches. Because closed branches carry empty root vectors, compaction only has to relocate the active ones. +`main` (id 0) cannot be closed. `StorageEngine.branches()` lists only `ACTIVE` branches. Because closed branches carry empty root vectors, they keep no pages alive. ```sql CREATE BRANCH what_if; diff --git a/docs/storage/change-feed.md b/docs/storage/change-feed.md index 0cb2ac2..a956387 100644 --- a/docs/storage/change-feed.md +++ b/docs/storage/change-feed.md @@ -124,7 +124,7 @@ for each flagged column, in that order: A column is flagged only when it differs from the default (`roleSet != NO_ROLES`, `weight != Weight.ONE`, `validFrom != Long.MIN_VALUE`, `validTo != Long.MAX_VALUE`, `dataRef != 0`, `qualifier != 0`), so a plain membership costs 1 byte of flags plus the member id. -Slot values that contain tree references (codecs with `holdsRefs()`, for example an `EdgeRecord` with member and order roots) are **detached** before encoding: every embedded `Ref` is replaced with `Ref.EMPTY`. Feed events therefore never pin pages, and compaction may relocate or reclaim any page without invalidating the feed. Consumers that need the topology of an edge read it from a snapshot, or use the event's `members` list. +Slot values that contain tree references (codecs with `holdsRefs()`, for example an `EdgeRecord` with member and order roots) are **detached** before encoding: every embedded `Ref` is replaced with `Ref.EMPTY`. Feed events therefore never pin pages, and compaction may move or reclaim any page without invalidating the feed. Consumers that need the topology of an edge read it from a snapshot, or use the event's `members` list. ### Example diff --git a/docs/storage/maintenance.md b/docs/storage/maintenance.md index 5c15ff5..2b47e12 100644 --- a/docs/storage/maintenance.md +++ b/docs/storage/maintenance.md @@ -13,12 +13,12 @@ Sources are under `engine/src/main/java/io/hstore/engine/maintenance/`, plus `St | Trigger | Action | Code | |---|---|---| | Engine open, after recovery | checkpoint, then reclaim | `StorageEngine` constructor | -| `hstore-maintenance` virtual thread, every 500 ms (skipped when the `maintenance` lock is busy) | checkpoint if `walSinceCheckpoint() > checkpointWalBytes`; a compaction pass when more than `checkpointWalBytes` of node images were written since the previous pass (followed by a checkpoint if anything was relocated); then reclaim | `StorageEngine.maintain` | +| `hstore-maintenance` virtual thread, every 500 ms (skipped when the `maintenance` lock is busy) | checkpoint if `walSinceCheckpoint() > checkpointWalBytes`; a compaction pass when more than `checkpointWalBytes` of node images were written since the previous pass (followed by a checkpoint if anything was moved); then reclaim | `StorageEngine.maintain` | | `CHECKPOINT;` (HQL), `StorageEngine.checkpoint()` | checkpoint | `timedCheckpoint` | | `COMPACT;` (HQL), `StorageEngine.compact()` | compact, checkpoint, reclaim | `StorageEngine.compact` | | `StorageEngine.close()` | final checkpoint, then shutdown | `close` | -Checkpoint and compaction are serialized by `StorageEngine.maintenance`, a `ReentrantLock`. Background compaction plays the role of PostgreSQL's autovacuum. Its trigger is write volume, `pages.bytesWritten()` advancing by more than `checkpoint_wal_mb` since the last pass, so an idle database never pays for a liveness walk. The background pass relocates at most one segment, the emptiest one whose live bytes are below `compaction_live_ratio` of its allocated bytes. `COMPACT;` relocates every such segment on demand. +Checkpoint and compaction are serialized by `StorageEngine.maintenance`, a `ReentrantLock`. Background compaction plays the role of PostgreSQL's autovacuum. Its trigger is write volume, `pages.bytesWritten()` advancing by more than `checkpoint_wal_mb` since the last pass, so an idle database never pays for a liveness walk. The background pass empties at most one segment, the emptiest one whose live bytes are below `compaction_live_ratio` of its allocated bytes. `COMPACT;` empties every such segment on demand. Every checkpoint is logged, and then the checkpoint listeners run. The database registers `SemanticPlane.persist`, which saves the HNSW index alongside the catalog. A sample log line: @@ -51,9 +51,14 @@ Each commit writes fresh node images for every path it changes and references al Segments are append-only extents ([pages.md](pages.md#segment-files)), so garbage is never reused piecemeal. It is reclaimed one whole segment at a time: 1. measure how many images in each segment are still reachable; -2. copy the reachable ones out of mostly-dead segments; -3. publish roots that no longer reference those segments; -4. delete the files once no reader can still see the old roots. +2. copy the reachable ones out of mostly-dead segments, byte for byte; +3. point their entries in the page directory at the copies; +4. delete the files once a checkpoint has made the new entries durable and no reader can still be using an old + address. + +No tree is rewritten and no generation is committed. Pages keep their numbers when they move +([pages.md](pages.md#page-numbers-and-the-page-directory)), so every node that points at a moved page, in every +generation and branch, still points at it. ### Segment state machine @@ -61,13 +66,13 @@ Segments are append-only extents ([pages.md](pages.md#segment-files)), so garbag stateDiagram-v2 [*] --> ACTIVE: rollSegment() ACTIVE --> SEALED: next segment rolled (capacity reached) - SEALED --> COMPACTING: chosen as victim (live < pages × ratio) - COMPACTING --> SEALED: allowed transition (abandoned compaction) - COMPACTING --> RETIRED: roots rewritten, retiredAt = generation - RETIRED --> [*]: reclaim(): file deleted once oldestPinned ≥ retiredAt + SEALED --> COMPACTING: chosen as victim (live < bytes × ratio) + COMPACTING --> SEALED: reopened after a crash mid-compaction + COMPACTING --> RETIRED: live pages moved, retiredAt = current generation + RETIRED --> [*]: reclaim(): file deleted after a checkpoint, once oldestPinned > retiredAt ``` -`SegmentState.canTransitionTo` allows only the next state, plus `COMPACTING → SEALED`. `SegmentInfo.withState` throws on anything else. Only one segment is `ACTIVE` at a time: the one `PageStore.allocate` appends to. It is never chosen as a victim. +`SegmentState.canTransitionTo` allows only the next state, plus `COMPACTING → SEALED`. `SegmentInfo.withState` throws on anything else. `PageStore.open` turns a `COMPACTING` segment back into `SEALED`, so a compaction cut short by a crash leaves the segment a candidate for the next pass. Only one segment is `ACTIVE` at a time: the one `PageStore.allocate` appends to. It is never chosen as a victim. ### Liveness @@ -77,7 +82,7 @@ stateDiagram-v2 * **every branch** in each of those generations; * **every slot root** in each branch. -`TreeWalker.visit(root, schema, visited)` records each `Ref.Stored` page id it meets, with the image size the reference carries, in one shared `visited` map, so an image shared by several generations, branches or trees is counted once. The size comes from the reference itself, so no page has to be read to learn it. It descends into branches and, for codecs that hold references (catalog edge records, promoted posting lists), into nested trees. +`TreeWalker.visit(root, schema, firstVisit)` calls `firstVisit` with each `Ref.Stored` page id it meets and stops descending when that returns false. The compactor sets one bit per page number in a `BitSet`, so an image shared by several generations, branches or trees is visited once, and the whole walk costs one bit per page number. Sizes come from the page directory afterwards: each set bit is looked up and its length added to its segment. The walker descends into branches and, for codecs that hold references (catalog edge records, promoted posting lists), into nested trees. A detail that keeps this cheap: at height 0 of a tree whose codec has no nested references, the walker records the leaf's id **without loading it**. The summaries in the parent already prove that the leaf exists. Liveness is therefore proportional to the number of internal nodes plus the reference-holding leaves, not to total data size. @@ -87,36 +92,35 @@ Liveness is exposed as `StorageEngine.liveness()` (live bytes per segment). The `Compactor.compact(liveThreshold)` (default threshold `compaction_live_ratio = 0.5`): -1. **Choose victims.** A victim is a `SEALED` segment, not the active one, with `live bytes < allocated bytes × threshold`, so the ratio is *live bytes / bytes written*. Candidates are ordered from emptiest to fullest. The background pass takes only the first one (`compact(threshold, 1)`), which bounds how long a single relocation holds the commit lock; `COMPACT` takes all of them. If there are no victims, it reports and returns. -2. Mark the victims `COMPACTING`. -3. **Relocate.** For every **active** branch of the current generation, call `transactions.rewrite(branch, roots -> roots.map(relocate))`. `TreeWalker.relocate(ref, schema, moving, scope)` rebuilds as a fresh `Ref.Pending` copy: - * every stored node whose page id lies in a victim segment; - * every ancestor of such a node; - * every leaf whose nested trees changed. +1. **Take a cut.** Inside `TransactionManager.exclusive`, which holds the commit lock and drains the group committer, record the retained generations (history plus `current`) and the set of segments that are already `SEALED`. Only those segments can become victims. A commit writes its pages under the commit lock but publishes its generation a moment later, and the active segment can fill up and be sealed while the walk runs. Taking the cut with nothing in flight means every page the walk doesn't see sits in a segment that wasn't sealed yet. +2. **Walk** the recorded generations as described under *Liveness*. +3. **Choose victims.** A victim is a segment from the cut, still `SEALED`, with `live bytes < allocated bytes × threshold`, so the ratio is *live bytes / bytes written*. Candidates are ordered from emptiest to fullest. The background pass takes only the first one (`compact(threshold, 1)`); `COMPACT` takes all of them. If there are no victims, it reports and returns. +4. Mark the victims `COMPACTING`. +5. **Move.** For every set bit whose directory entry points into a victim, `PageStore.copy` reads the image, checks it with `PageHeader.verify`, and writes the same bytes to the active segment. The header holds the page number, not the address, so neither the header nor the checksum changes. Every 4 MiB of copies, `PageStore.sync()` forces the segments and only then are the directory entries pointed at the copies. A directory entry therefore never reaches the disk pointing at bytes that are not already there. +6. Count the compaction in `TransactionManager.compacted()` (see below) and mark the victims `RETIRED` with `retiredAt` set to the current generation. - Subtrees that touch no victim are returned unchanged and stay shared. `rewrite` publishes the new root vector as a normal commit: it takes the commit lock, materializes the pending nodes into the active segment, appends the WAL records and goes through the group-commit barrier. It also sets `relocationFence` to the new generation id. -4. Mark the victims `RETIRED` with `retiredAt = generation`. Retained history is **not** truncated. Generations older than `retiredAt` keep pointing at the old images in the victims, which stay on disk and readable, so `AT GENERATION` and `AS OF` keep working across a compaction. The victims are reclaimed once `history_limit` has rolled past `retiredAt`. +`StorageEngine.compact()` then checkpoints, which forces the page directory and persists the segment states, and calls `reclaim()`. -`StorageEngine.compact()` then checkpoints, which persists the new segment states and roots, and calls `reclaim()`. +Nothing in this takes the commit lock except the short cut in step 1, and nothing commits. Commits carry on while pages move, readers keep using the node cache, whose entries are keyed by page number and stay valid, and retained history keeps working because old generations reach the moved pages through the same directory. -Closed branches (`MERGED`, `DROPPED`) carry empty root vectors ([catalog-and-generations.md](catalog-and-generations.md#branches)), so they hold nothing alive and are not rewritten. +Moves are not written to the WAL. If the engine crashes before the checkpoint that follows a compaction, recovery replays the commits since the previous checkpoint and writes their original addresses back into the directory. Those addresses are in the victims, which are only deleted after that checkpoint, so they still hold the same bytes. The copies made before the crash become unreferenced space in the active segment, and the victims are picked again by a later pass. `CompactionTest` crashes at `COMPACTION_MOVE` and `COMPACTION_RETIRE`, with and without losing the directory writes made since the last checkpoint, and checks the data after reopening and after another compaction. #### Interaction with running transactions -A transaction that started before the compaction holds roots from its base generation. At commit, one of these applies: +Compaction doesn't change any root, so transactions that started before it commit exactly as they would have without it, on the fast path or by rebasing. -* **Fast path** (`Workspace.untouchedSince`). Every slot the transaction changed (and, for `SERIALIZABLE`, every slot it read) still has the same root in the latest generation. Compaction rewrote none of those slots, which means they contained no images in victim segments, so the transaction's versions of them cannot reference victims either. `rebasedOnto` combines them with the latest, relocated roots for every other slot. -* **Rebase path.** The op log is replayed onto the latest (relocated) roots, so the result references only relocated images. -* **Fence.** A transaction whose base generation predates `relocationFence` and that carries already-stored roots, such as spilled bulk-load subtrees (`Transaction.carriesStoredRoots`), cannot be safely rebased onto relocated data. It fails with a retryable conflict (`bulk-loaded roots of transaction N predate a compaction`), and `HypergraphDatabase.write` retries it. +The one exception is a bulk load. `Transaction.load` can spill finished subtrees to disk before commit (`Transaction.carriesStoredRoots`), and those pages are not reachable from any generation until the transaction commits, so the walk can't see them. If one of them sits in a victim, it isn't moved. Each transaction therefore records `TransactionManager.compactions()` when it begins. The count goes up when a compaction takes its cut, under the commit lock, so a transaction that carries stored roots and began before a cut that has since been taken fails with a retryable conflict, even when it commits while pages are still moving (`bulk-loaded roots of transaction N predate a compaction`). `HypergraphDatabase.write` retries it. The check applies on the fast path too, since compaction no longer changes the roots that would otherwise push such a transaction off it. ### Reclaim `Compactor.reclaim()` deletes the file and catalog entry of every `RETIRED` segment, but only when both of these hold: -* no generation is pinned, or `oldestPinned() ≥ retiredAt`; and -* the oldest generation still retained in history is `≥ retiredAt`, so no time-travel target can reference the segment. +* a checkpoint has completed since the segment was retired (`Checkpointer.completed()`), so the new directory entries are durable and the WAL no longer holds commits whose addresses point into it; and +* no generation is pinned, or `oldestPinned() > retiredAt`. + +Retained history no longer holds a segment back. Old generations reach moved pages through the directory like everything else, so `AT GENERATION` and `AS OF` keep working after the victims are gone. -**Why it must wait.** A `Snapshot` opened before compaction pinned a generation older than `retiredAt`. That generation's roots still point into the victim segments, and copy-on-write promises the reader those images stay valid until it closes. Deleting the file early would turn its next read into `segment does not exist`. Pins are reference counts per generation (`TransactionManager.pins`). Every `Snapshot.close()` and transaction end decrements them, and the maintenance thread retries `reclaim` every 500 ms. A retired segment therefore disappears within half a second of its last reader closing. +**Why it must wait for readers.** A reader looks up a page's address and then reads the page. One that started before the move may have looked up an old address and not read it yet. Every reader and writer pins its generation, and anyone pinned at `retiredAt` or earlier may have started before the moves finished. Deleting the file early would turn that reader's next read into `segment does not exist`. Pins are reference counts per generation (`TransactionManager.pins`). Every `Snapshot.close()` and transaction end decrements them, and the maintenance thread retries `reclaim` every 500 ms. A retired segment therefore disappears within half a second of its last reader closing. ### History retention and pins diff --git a/docs/storage/pages.md b/docs/storage/pages.md index 7b76aa7..c54d848 100644 --- a/docs/storage/pages.md +++ b/docs/storage/pages.md @@ -157,9 +157,9 @@ lookup is a load from the OS page cache and the directory takes no Java heap. A Trees, nested references, `NodeCache` and `Ref` use numbers and never see an address. Only `PageStore` resolves them, and the WAL logs each new page's address next to its number so that recovery can put the directory entry -back. That is what will let compaction move an image by changing one directory entry, without rewriting the nodes -that point at it. It doesn't do that yet; it still rebuilds the paths to the pages it moves -([maintenance.md](maintenance.md)). +back. That is what lets compaction move an image by copying its bytes and changing one directory entry, without +rewriting the nodes that point at it ([maintenance.md](maintenance.md#compaction)). Entries are written and read with +release and acquire ordering, so a reader racing a move sees either the old address or the new one, never a mix. Page `#1234`, stored as 3 units at unit 1234 of segment 1: diff --git a/docs/storage/persistent-tree.md b/docs/storage/persistent-tree.md index 66d4ba2..fe74b5a 100644 --- a/docs/storage/persistent-tree.md +++ b/docs/storage/persistent-tree.md @@ -20,7 +20,7 @@ This document describes the node structure, the write path, the summaries and th | `TreeCursor` | `TreeCursor.java` | Bidirectional cursor with seek and summary-based pruning. | | `TreeAlgebra`, `TreeDiff` | `TreeAlgebra.java`, `TreeDiff.java` | Intersection (leapfrog, synchronized descent, probing), union, difference, subset, equality and structural diff. | | `Materializer` | `Materializer.java` | Freezes and encodes pending nodes into pages, post-order. | -| `TreeWalker`, `TreeVerifier` | | Reachability (liveness), relocation (compaction) and invariant checking. | +| `TreeWalker`, `TreeVerifier` | | Reachability (liveness for compaction) and invariant checking. | ## Node structure diff --git a/docs/storage/topology.md b/docs/storage/topology.md index 09224a3..6de2322 100644 --- a/docs/storage/topology.md +++ b/docs/storage/topology.md @@ -36,9 +36,9 @@ EdgeRecord: u8 tag=1 | varint type | u8 kind (0=SET, 1=ORDERED) | Ref members | Ref: i64le pageId (0 = empty) | [Summary encoding, only when pageId != 0] ``` -Because `AtomCodec.holdsRefs()` is true, the materializer writes each edge's member and order trees *before* the catalog leaf that references them. `TreeWalker` and `TreeVerifier` descend into them, and compaction relocates them. The catalog therefore forms a two-level tree of trees: catalog leaves hold edge records, and each edge record roots its own membership tree. +Because `AtomCodec.holdsRefs()` is true, the materializer writes each edge's member and order trees *before* the catalog leaf that references them. `TreeWalker` and `TreeVerifier` descend into them, and compaction moves their pages like any other. The catalog therefore forms a two-level tree of trees: catalog leaves hold edge records, and each edge record roots its own membership tree. -A record's `contentHash()` feeds the catalog's SET fingerprint. The hash of an edge record is computed from `(1, type, kind, members.count, members.fingerprint, version, dataRef, tenant, isolation)`. It is a function of the edge's member *contents*, never of page ids, so relocating pages during compaction does not change catalog fingerprints. +A record's `contentHash()` feeds the catalog's SET fingerprint. The hash of an edge record is computed from `(1, type, kind, members.count, members.fingerprint, version, dataRef, tenant, isolation)`. It is a function of the edge's member *contents*, never of page ids, and compaction never changes page ids anyway, so it leaves catalog fingerprints alone. ## Hyperedges diff --git a/docs/transactions/transactions.md b/docs/transactions/transactions.md index 230e07f..fa66a8e 100644 --- a/docs/transactions/transactions.md +++ b/docs/transactions/transactions.md @@ -118,7 +118,7 @@ commit(txn): if txn has requestId and REQUESTS[hash(id)] exists -> DUPLICATE (prior txn id, prior generation) fast = branch.roots sameAs txn.baseRoots || (requestsStable && workspace.untouchedSince(branch.roots, isolation == SERIALIZABLE)) - if !fast and txn.generation < relocationFence and txn carries Load ops -> Conflict + if txn.compactionsAtStart < compactions and txn carries Load ops -> Conflict if fast: workspace = txn.workspace else: txn.validateReads(branch.roots); workspace = txn.replay(branch.roots) if requestId: workspace.write(REQUESTS, hash(id), RequestRecord(id, txn.id, latest.id + 1)) @@ -155,9 +155,9 @@ A footprint is *commutative* iff the edge was a `SET` edge at the base and every `TreeDiff.diff` uses the frontier algorithm over shared subtrees ([persistent tree](../storage/persistent-tree.md)), so the cost of the overlap check is proportional to the changed region, not the edge cardinality. -### 5.3 Relocation fence +### 5.3 Compaction fence -Compaction rewrites roots through `TransactionManager.rewrite`, which records `relocationFence = new generation id`. A transaction that bulk-loaded stored subtrees (`carriesStoredRoots()`: any `EdgeOp` with `EdgeAction.Load`) began before the fence and cannot take the fast path is rejected with a conflict, because its stored roots may point into segments that compaction is about to retire. Retrying rebuilds the subtree. +Compaction moves pages without changing any root, so it doesn't affect other commits. A bulk load is different: a transaction that bulk-loaded stored subtrees (`carriesStoredRoots()`: any `EdgeOp` with `EdgeAction.Load`) may have spilled pages that no generation reaches yet, so compaction can't see them and won't move them. Each transaction records `TransactionManager.compactions()` when it begins, and one that carries stored roots is rejected with a conflict if a compaction has finished since, on the fast path as well as the slow one. Retrying rebuilds the subtree. ### 5.4 Page budget @@ -206,7 +206,7 @@ Under `SNAPSHOT` both commits succeed (R replays `Insert(p)` into B, which L nev | Code | Class | Retryable | Raised by | |---|---|---|---| -| `RETRYABLE_CONFLICT` | `Conflict` | yes | replay preconditions, `verifyEdge`, `validateReads`, relocation fence, concurrent atom id/canonical key | +| `RETRYABLE_CONFLICT` | `Conflict` | yes | replay preconditions, `verifyEdge`, `validateReads`, compaction fence, concurrent atom id/canonical key | | `RETRYABLE_IO` | `TransientIo` | yes | WAL/page/feed I/O failures; a broken commit pipeline (`"the commit pipeline stopped after a durability failure; restart the engine"`) | | `ABORTED_RESOURCE_LIMIT` | `ResourceLimit` | no | page budget, quotas | | `INVALID_SCHEMA` | `InvalidSchema` | no | caller errors | @@ -344,10 +344,9 @@ Had T2 also touched `m1`, or had `E` been an `ORDERED` edge, the overlap check w ## 10. Branch operations and internal writes -`system(branchId, work)`, `createBranch`, `closeBranch` and `rewrite` reuse `append` and the group committer, so they are durable and ordered exactly like user commits: +`system(branchId, work)`, `createBranch` and `closeBranch` reuse `append` and the group committer, so they are durable and ordered exactly like user commits: * `createBranch` appends a `BranchMeta` record and a branch whose roots are the parent's current roots (copy-on-write makes this O(number of slots)). Names must be unique among active branches; the new id is one past the largest existing id. * `closeBranch` (`MERGED` or `DROPPED`, never for `MAIN`) empties the root vector so the branch's pages become unreachable. -* `rewrite` (compaction) replaces roots without member or slot changes and sets the relocation fence. These commits carry no member or slot changes, so their `CommitEvent` is empty and nothing is written to the change feed (see [change feed](../storage/change-feed.md)). diff --git a/docs/transactions/wal-and-recovery.md b/docs/transactions/wal-and-recovery.md index c095b0d..6e1fdb8 100644 --- a/docs/transactions/wal-and-recovery.md +++ b/docs/transactions/wal-and-recovery.md @@ -275,7 +275,7 @@ The checkpoint LSN is the position of the `Checkpoint` record itself, so the nex |---|---| | Every open, right after recovery | `StorageEngine` constructor | | WAL grew by more than `checkpointWalBytes` (default 256 MiB) since the last checkpoint; checked every 500 ms | `StorageEngine.maintain` | -| Compaction, between relocation and segment reclamation | `StorageEngine.compact` | +| Compaction, after moving pages and before deleting segments | `StorageEngine.compact` | | Clean shutdown | `StorageEngine.close` | | On demand | `StorageEngine.checkpoint()`, HQL `CHECKPOINT` | diff --git a/engine/src/main/java/io/hstore/engine/StorageEngine.java b/engine/src/main/java/io/hstore/engine/StorageEngine.java index fc2361f..7eab8a8 100644 --- a/engine/src/main/java/io/hstore/engine/StorageEngine.java +++ b/engine/src/main/java/io/hstore/engine/StorageEngine.java @@ -82,7 +82,7 @@ private StorageEngine(Path directory, EngineOptions options, FileChannel lockCha options.derivations(), options.durability(), options.walMode(), options.faults(), options.historyLimit()); this.transactions = new TransactionManager(storage, recovery.image().current(), recovery.image().history()); this.checkpointer = new Checkpointer(transactions, pages, wal, feed, catalog, options.historyLimit(), checkpoint.checkpointLsn()); - this.compactor = new Compactor(transactions, pages, slots); + this.compactor = new Compactor(transactions, pages, slots, checkpointer, options.faults()); this.dictionary = new Dictionary(transactions); checkpointer.checkpoint(); compactor.reclaim(); @@ -274,7 +274,7 @@ public Compactor.Report compact() { Compactor.Report report = compactor.compact(options.compactionLiveRatio()); timedCheckpoint(); List reclaimed = compactor.reclaim(); - LOG.log(System.Logger.Level.INFO, "compaction relocated segments {0} and reclaimed {1}", report.compacted(), reclaimed); + LOG.log(System.Logger.Level.INFO, "compaction moved the live pages out of segments {0} and reclaimed {1}", report.compacted(), reclaimed); return report; } finally { maintenance.unlock(); @@ -308,7 +308,7 @@ private void maintain() { Compactor.Report report = compactor.compact(options.compactionLiveRatio(), 1); if (!report.compacted().isEmpty()) { timedCheckpoint(); - LOG.log(System.Logger.Level.INFO, "background compaction relocated segments {0}", report.compacted()); + LOG.log(System.Logger.Level.INFO, "background compaction moved the live pages out of segments {0}", report.compacted()); } } List reclaimed = compactor.reclaim(); diff --git a/engine/src/main/java/io/hstore/engine/maintenance/Checkpointer.java b/engine/src/main/java/io/hstore/engine/maintenance/Checkpointer.java index ab97b75..f4b627b 100644 --- a/engine/src/main/java/io/hstore/engine/maintenance/Checkpointer.java +++ b/engine/src/main/java/io/hstore/engine/maintenance/Checkpointer.java @@ -23,6 +23,7 @@ public record Result(long generation, long lsn, int walSegments) { private final CatalogStore catalog; private final int historyLimit; private volatile long lastLsn; + private volatile long completed; public Checkpointer(TransactionManager transactions, PageStore pages, WriteAheadLog wal, ChangeFeed feed, CatalogStore catalog, int historyLimit, long recoveredLsn) { @@ -39,6 +40,10 @@ public long walSinceCheckpoint() { return wal.end() - lastLsn; } + public long completed() { + return completed; + } + public Result checkpoint() { return transactions.exclusive(() -> { long lsn = wal.end(); @@ -55,6 +60,7 @@ public Result checkpoint() { wal.sync(); wal.truncateBefore(lsn); lastLsn = lsn; + completed++; return new Result(current.id(), lsn, wal.segmentCount()); }); } diff --git a/engine/src/main/java/io/hstore/engine/maintenance/Compactor.java b/engine/src/main/java/io/hstore/engine/maintenance/Compactor.java index 3bb7932..ad8a5ab 100644 --- a/engine/src/main/java/io/hstore/engine/maintenance/Compactor.java +++ b/engine/src/main/java/io/hstore/engine/maintenance/Compactor.java @@ -1,24 +1,25 @@ package io.hstore.engine.maintenance; -import io.hstore.engine.catalog.Branch; import io.hstore.engine.catalog.Generation; import io.hstore.engine.catalog.SlotRegistry; +import io.hstore.engine.page.PageAddress; import io.hstore.engine.page.PageId; import io.hstore.engine.page.PageStore; import io.hstore.engine.page.SegmentInfo; import io.hstore.engine.page.SegmentState; import io.hstore.engine.tree.TreeWalker; -import io.hstore.engine.tree.WriteScope; +import io.hstore.engine.txn.CrashPoint; import io.hstore.engine.txn.TransactionManager; +import java.util.Arrays; +import java.util.BitSet; import java.util.Comparator; -import java.util.HashMap; import java.util.List; import java.util.Map; import java.util.OptionalLong; import java.util.Set; import java.util.TreeMap; -import java.util.function.LongPredicate; +import java.util.concurrent.ConcurrentHashMap; import java.util.stream.Collectors; import java.util.stream.Stream; @@ -27,25 +28,73 @@ public final class Compactor { public record Report(Map liveBytes, List compacted, long generation) { } + private record Cut(List generations, Set sealed) { + } + + private static final long PASS_BYTES = 4L << 20; + private final TransactionManager transactions; private final PageStore pages; private final SlotRegistry slots; + private final Checkpointer checkpointer; + private final CrashPoint.Injector faults; + private final Map retiredBeforeCheckpoint = new ConcurrentHashMap<>(); - public Compactor(TransactionManager transactions, PageStore pages, SlotRegistry slots) { + public Compactor(TransactionManager transactions, PageStore pages, SlotRegistry slots, Checkpointer checkpointer, + CrashPoint.Injector faults) { this.transactions = transactions; this.pages = pages; this.slots = slots; + this.checkpointer = checkpointer; + this.faults = faults; } public Map liveness() { - Map visited = new HashMap<>(); + return bytesBySegment(reachable(retained())); + } + + private List retained() { + return Stream.concat(transactions.history().stream(), Stream.of(transactions.current())).toList(); + } + + private Cut cut() { + return transactions.exclusive(() -> { + transactions.compacted(); + int active = pages.activeSegment(); + Set sealed = pages.segments().stream() + .filter(segment -> segment.state() == SegmentState.SEALED && segment.id() != active) + .map(SegmentInfo::id) + .collect(Collectors.toUnmodifiableSet()); + return new Cut(retained(), sealed); + }); + } + + private BitSet reachable(List generations) { + BitSet live = new BitSet(); TreeWalker walker = new TreeWalker(transactions.source()); - Stream.concat(transactions.history().stream(), Stream.of(transactions.current())) + generations.stream() .flatMap(generation -> generation.branches().values().stream()) .forEach(branch -> branch.roots().roots().forEach((slot, ref) -> - walker.visit(ref, slots.slot(slot).schema(), visited))); - return visited.entrySet().stream().collect(Collectors.groupingBy(entry -> pages.segmentOf(entry.getKey()), TreeMap::new, - Collectors.summingLong(entry -> (long) entry.getValue() * PageId.UNIT_BYTES))); + walker.visit(ref, slots.slot(slot).schema(), pageId -> mark(live, pageId)))); + return live; + } + + private static boolean mark(BitSet live, long pageId) { + int index = Math.toIntExact(PageId.indexOf(pageId)); + if (live.get(index)) { + return false; + } + live.set(index); + return true; + } + + private Map bytesBySegment(BitSet live) { + Map bytes = new TreeMap<>(); + live.stream().forEach(index -> { + long address = pages.addressOf(PageId.pack(index, 0)); + bytes.merge(PageAddress.segmentOf(address), (long) PageAddress.unitsOf(address) * PageId.UNIT_BYTES, Long::sum); + }); + return bytes; } public Report compact(double liveThreshold) { @@ -53,44 +102,85 @@ public Report compact(double liveThreshold) { } public Report compact(double liveThreshold, int maxVictims) { - Map live = liveness(); - int active = pages.activeSegment(); + Cut cut = cut(); + BitSet live = reachable(cut.generations()); + Map liveBytes = bytesBySegment(live); List victims = pages.segments().stream() - .filter(segment -> segment.state() == SegmentState.SEALED && segment.id() != active) - .filter(segment -> live.getOrDefault(segment.id(), 0L) < segment.bytes() * liveThreshold) - .sorted(Comparator.comparingDouble(segment -> (double) live.getOrDefault(segment.id(), 0L) / Math.max(1, segment.bytes()))) + .filter(segment -> segment.state() == SegmentState.SEALED && cut.sealed().contains(segment.id())) + .filter(segment -> liveBytes.getOrDefault(segment.id(), 0L) < segment.bytes() * liveThreshold) + .sorted(Comparator.comparingDouble(segment -> (double) liveBytes.getOrDefault(segment.id(), 0L) / Math.max(1, segment.bytes()))) .limit(maxVictims) .map(SegmentInfo::id) .toList(); - long generation = transactions.current().id(); if (victims.isEmpty()) { - return new Report(live, victims, generation); + return new Report(liveBytes, victims, transactions.current().id()); } victims.forEach(id -> pages.transition(id, SegmentState.COMPACTING, 0)); - Set victimSet = Set.copyOf(victims); - LongPredicate moving = pageId -> victimSet.contains(pages.segmentOf(pageId)); - TreeWalker walker = new TreeWalker(transactions.source()); - List branches = transactions.current().branches().values().stream() - .filter(Branch::isActive).map(Branch::id).toList(); - for (int branch : branches) { - generation = transactions.rewrite(branch, roots -> roots.map((slot, ref) -> - walker.relocate(ref, slots.slot(slot).schema(), moving, new WriteScope()))); + try { + move(live, Set.copyOf(victims)); + } catch (RuntimeException | Error failure) { + victims.forEach(id -> pages.transition(id, SegmentState.SEALED, 0)); + throw failure; + } + faults.reach(CrashPoint.COMPACTION_MOVE); + long retiredAt = transactions.current().id(); + long checkpoints = checkpointer.completed(); + victims.forEach(id -> { + pages.transition(id, SegmentState.RETIRED, retiredAt); + retiredBeforeCheckpoint.put(id, checkpoints); + }); + faults.reach(CrashPoint.COMPACTION_RETIRE); + return new Report(liveBytes, victims, retiredAt); + } + + private void move(BitSet live, Set victims) { + long[] moved = new long[64]; + int count = 0; + long bytes = 0; + for (int index = live.nextSetBit(0); index >= 0; index = live.nextSetBit(index + 1)) { + long pageId = PageId.pack(index, 0); + long from = pages.addressOf(pageId); + if (!victims.contains(PageAddress.segmentOf(from))) { + continue; + } + if (count + 2 > moved.length) { + moved = Arrays.copyOf(moved, moved.length * 2); + } + moved[count++] = pageId; + moved[count++] = pages.copy(pageId); + bytes += (long) PageAddress.unitsOf(from) * PageId.UNIT_BYTES; + if (bytes >= PASS_BYTES) { + repoint(moved, count); + count = 0; + bytes = 0; + } + } + repoint(moved, count); + } + + private void repoint(long[] moved, int count) { + if (count == 0) { + return; + } + pages.sync(); + for (int i = 0; i < count; i += 2) { + pages.place(moved[i], moved[i + 1]); } - long retiredAt = generation; - victims.forEach(id -> pages.transition(id, SegmentState.RETIRED, retiredAt)); - return new Report(live, victims, generation); } public List reclaim() { OptionalLong pinned = transactions.oldestPinned(); - long oldestRetained = transactions.history().stream().mapToLong(Generation::id).min().orElse(Long.MAX_VALUE); + long checkpoints = checkpointer.completed(); List reclaimable = pages.segments().stream() .filter(segment -> segment.state() == SegmentState.RETIRED) - .filter(segment -> pinned.isEmpty() || pinned.getAsLong() >= segment.retiredAt()) - .filter(segment -> oldestRetained >= segment.retiredAt()) + .filter(segment -> pinned.isEmpty() || pinned.getAsLong() > segment.retiredAt()) + .filter(segment -> checkpoints > retiredBeforeCheckpoint.getOrDefault(segment.id(), -1L)) .map(SegmentInfo::id) .toList(); - reclaimable.forEach(pages::delete); + reclaimable.forEach(id -> { + pages.delete(id); + retiredBeforeCheckpoint.remove(id); + }); return reclaimable; } } diff --git a/engine/src/main/java/io/hstore/engine/page/PageDirectory.java b/engine/src/main/java/io/hstore/engine/page/PageDirectory.java index 0518318..97ceb88 100644 --- a/engine/src/main/java/io/hstore/engine/page/PageDirectory.java +++ b/engine/src/main/java/io/hstore/engine/page/PageDirectory.java @@ -7,6 +7,7 @@ import java.lang.foreign.Arena; import java.lang.foreign.MemorySegment; import java.lang.foreign.ValueLayout; +import java.lang.invoke.VarHandle; import java.nio.ByteOrder; import java.nio.channels.FileChannel; import java.nio.file.Path; @@ -25,6 +26,7 @@ final class PageDirectory implements AutoCloseable { private static final long CHUNK_BYTES = 1L << CHUNK_SHIFT; private static final ValueLayout.OfInt INT = ValueLayout.JAVA_INT.withOrder(ByteOrder.LITTLE_ENDIAN); private static final ValueLayout.OfLong LONG = ValueLayout.JAVA_LONG.withOrder(ByteOrder.LITTLE_ENDIAN); + private static final VarHandle ENTRY = LONG.varHandle(); private final Path path; private final FileChannel channel; @@ -63,7 +65,7 @@ long get(long index) { long position = HEADER + index * Long.BYTES; MemorySegment[] mapped = chunks; int chunk = (int) (position >>> CHUNK_SHIFT); - return chunk < mapped.length ? mapped[chunk].get(LONG, position & (CHUNK_BYTES - 1)) : PageAddress.NONE; + return chunk < mapped.length ? (long) ENTRY.getAcquire(mapped[chunk], position & (CHUNK_BYTES - 1)) : PageAddress.NONE; } void put(long index, long address) { @@ -72,7 +74,7 @@ void put(long index, long address) { if (chunk >= chunks.length) { mapThrough(chunk); } - chunks[chunk].set(LONG, position & (CHUNK_BYTES - 1), address); + ENTRY.setRelease(chunks[chunk], position & (CHUNK_BYTES - 1), address); } long allocate() { diff --git a/engine/src/main/java/io/hstore/engine/page/PageStore.java b/engine/src/main/java/io/hstore/engine/page/PageStore.java index 62a7521..ebf4103 100644 --- a/engine/src/main/java/io/hstore/engine/page/PageStore.java +++ b/engine/src/main/java/io/hstore/engine/page/PageStore.java @@ -56,6 +56,9 @@ private void restore(Collection known) { files.put(id, segment); long units = segment.unitsOnDisk(); SegmentInfo info = byId.getOrDefault(id, new SegmentInfo(id, SegmentState.SEALED, units, units, 0)); + if (info.state() == SegmentState.COMPACTING) { + info = info.withState(SegmentState.SEALED, 0); + } segments.put(id, info.withAllocation(info.pages(), Math.max(info.units(), units))); } } @@ -87,20 +90,41 @@ public long allocate(int length) { int units = PageId.unitsFor(length); allocation.lock(); try { - if (activeSegment == 0 || nextUnit + units > segmentUnits) { - rollSegment(); - } - int offset = nextUnit; - nextUnit += units; - segments.computeIfPresent(activeSegment, (_, info) -> info.withAllocation(info.pages() + 1, Math.max(info.units(), offset + (long) units))); long index = addresses.allocate(); - addresses.put(index, new PageAddress(activeSegment, offset, units).pack()); + addresses.put(index, claim(units)); return PageId.pack(index, 0); } finally { allocation.unlock(); } } + public long copy(long pageId) { + long from = locate(pageId); + MemorySegment image = read(pageId); + PageHeader.verify(image, pageId); + long to; + allocation.lock(); + try { + to = claim(PageAddress.unitsOf(from)); + } finally { + allocation.unlock(); + } + files.get(PageAddress.segmentOf(to)).write((long) PageAddress.offsetOf(to) * PageId.UNIT_BYTES, image); + pagesWritten.incrementAndGet(); + bytesWritten.addAndGet(image.byteSize()); + return to; + } + + private long claim(int units) { + if (activeSegment == 0 || nextUnit + units > segmentUnits) { + rollSegment(); + } + int offset = nextUnit; + nextUnit += units; + segments.computeIfPresent(activeSegment, (_, info) -> info.withAllocation(info.pages() + 1, Math.max(info.units(), offset + (long) units))); + return new PageAddress(activeSegment, offset, units).pack(); + } + private void rollSegment() { if (activeSegment != 0) { segments.computeIfPresent(activeSegment, (_, info) -> info.withState(SegmentState.SEALED, 0)); diff --git a/engine/src/main/java/io/hstore/engine/tree/TreeWalker.java b/engine/src/main/java/io/hstore/engine/tree/TreeWalker.java index 67233f5..8e2ac4a 100644 --- a/engine/src/main/java/io/hstore/engine/tree/TreeWalker.java +++ b/engine/src/main/java/io/hstore/engine/tree/TreeWalker.java @@ -1,6 +1,5 @@ package io.hstore.engine.tree; -import java.util.Map; import java.util.function.LongPredicate; public final class TreeWalker { @@ -13,12 +12,12 @@ public TreeWalker(NodeSource source) { this.source = source; } - public void visit(Ref root, TreeSchema schema, Map visited) { - visit(root, schema, UNKNOWN_HEIGHT, visited); + public void visit(Ref root, TreeSchema schema, LongPredicate firstVisit) { + visit(root, schema, UNKNOWN_HEIGHT, firstVisit); } - private void visit(Ref ref, TreeSchema schema, int height, Map visited) { - if (!(ref instanceof Ref.Stored(long pageId, int units, Summary _)) || visited.putIfAbsent(pageId, units) != null) { + private void visit(Ref ref, TreeSchema schema, int height, LongPredicate firstVisit) { + if (!(ref instanceof Ref.Stored(long pageId, int _, Summary _)) || !firstVisit.test(pageId)) { return; } if (height == 0 && !schema.codec().holdsRefs()) { @@ -27,65 +26,23 @@ private void visit(Ref ref, TreeSchema schema, int height, Map switch (source.load(pageId, schema)) { case Leaf leaf -> { for (int i = 0; i < leaf.size(); i++) { - mapNested(schema, leaf.value(i), (nested, nestedSchema) -> { - visit(nested, nestedSchema, UNKNOWN_HEIGHT, visited); - return nested; - }); + visitNested(schema, leaf.value(i), firstVisit); } } case Branch branch -> { for (int i = 0; i < branch.size(); i++) { - visit(branch.child(i), schema, branch.height() - 1, visited); + visit(branch.child(i), schema, branch.height() - 1, firstVisit); } } } } - public Ref relocate(Ref root, TreeSchema schema, LongPredicate moving, WriteScope scope) { - return relocate(root, schema, UNKNOWN_HEIGHT, moving, scope); - } - - private Ref relocate(Ref ref, TreeSchema schema, int height, LongPredicate inVictim, WriteScope scope) { - if (!(ref instanceof Ref.Stored(long pageId, int _, Summary _))) { - return ref; - } - boolean moving = inVictim.test(pageId); - if (height == 0 && !moving && !schema.codec().holdsRefs()) { - return ref; - } - return switch (source.load(pageId, schema)) { - case Leaf leaf -> { - Object[] values = leaf.valueList().toArray(); - boolean changed = moving; - for (int i = 0; i < values.length; i++) { - Object mapped = mapNested(schema, values[i], (nested, nestedSchema) -> - relocate(nested, nestedSchema, UNKNOWN_HEIGHT, inVictim, scope)); - changed |= mapped != values[i]; - values[i] = mapped; - } - yield changed ? new Ref.Pending(Leaf.owned(schema, scope.token(), leaf.keyArray(), values, values.length)) : ref; - } - case Branch branch -> { - Ref[] children = branch.childArray(); - boolean changed = moving; - for (int i = 0; i < children.length; i++) { - Ref moved = relocate(children[i], schema, branch.height() - 1, inVictim, scope); - changed |= moved != children[i]; - children[i] = moved; - } - yield changed - ? new Ref.Pending(Branch.owned(schema, scope.token(), branch.separatorArray(), children, children.length, branch.height())) - : ref; - } - }; - } - - private static Object mapNested(TreeSchema schema, Object value, ValueCodec.RefMapper mapper) { - if (!schema.codec().holdsRefs()) { - return value; + private void visitNested(TreeSchema schema, Object value, LongPredicate firstVisit) { + if (schema.codec().holdsRefs()) { + schema.codec().mapRefs(schema.cast(value), (nested, nestedSchema) -> { + visit(nested, nestedSchema, UNKNOWN_HEIGHT, firstVisit); + return nested; + }); } - V typed = schema.cast(value); - V mapped = schema.codec().mapRefs(typed, mapper); - return mapped.equals(typed) ? value : mapped; } } diff --git a/engine/src/main/java/io/hstore/engine/txn/CrashPoint.java b/engine/src/main/java/io/hstore/engine/txn/CrashPoint.java index f6e4b18..56da57a 100644 --- a/engine/src/main/java/io/hstore/engine/txn/CrashPoint.java +++ b/engine/src/main/java/io/hstore/engine/txn/CrashPoint.java @@ -7,7 +7,9 @@ public enum CrashPoint { COMMIT_APPEND, DATA_SYNC, WAL_SYNC, - CATALOG_PUBLISH; + CATALOG_PUBLISH, + COMPACTION_MOVE, + COMPACTION_RETIRE; @FunctionalInterface public interface Injector { diff --git a/engine/src/main/java/io/hstore/engine/txn/Transaction.java b/engine/src/main/java/io/hstore/engine/txn/Transaction.java index ce0bfb2..1ea0b30 100644 --- a/engine/src/main/java/io/hstore/engine/txn/Transaction.java +++ b/engine/src/main/java/io/hstore/engine/txn/Transaction.java @@ -47,6 +47,7 @@ private record KeyRead(Slot slot, long key) { private final TransactionManager manager; private final long id; + private final long compactionsAtStart; private final Generation snapshot; private final Branch branch; private final TxnOptions options; @@ -62,6 +63,7 @@ private record KeyRead(Slot slot, long key) { Transaction(TransactionManager manager, long id, Generation snapshot, TxnOptions options) { this.manager = manager; this.id = id; + this.compactionsAtStart = manager.compactions(); this.snapshot = snapshot; this.branch = snapshot.branch(options.branch()); this.options = options; @@ -259,6 +261,10 @@ private void requireActive() { } } + long compactionsAtStart() { + return compactionsAtStart; + } + boolean carriesStoredRoots() { return ops.stream().anyMatch(op -> op instanceof Op.EdgeOp(long _, EdgeAction.Load _)); } diff --git a/engine/src/main/java/io/hstore/engine/txn/TransactionManager.java b/engine/src/main/java/io/hstore/engine/txn/TransactionManager.java index 16c39a5..8e9dae6 100644 --- a/engine/src/main/java/io/hstore/engine/txn/TransactionManager.java +++ b/engine/src/main/java/io/hstore/engine/txn/TransactionManager.java @@ -42,7 +42,6 @@ import java.util.concurrent.locks.ReentrantLock; import java.util.function.Function; import java.util.function.Supplier; -import java.util.function.UnaryOperator; import java.util.stream.Stream; public final class TransactionManager { @@ -80,7 +79,7 @@ private record PageWrite(long pageId, MemorySegment image) { private final LongAdder readOnly = new LongAdder(); private final LongAdder pagesWritten = new LongAdder(); private final LongAdder walBytes = new LongAdder(); - private volatile long relocationFence; + private final AtomicLong compactions = new AtomicLong(); public TransactionManager(Storage storage, Generation recovered, Collection retained) { this.storage = storage; @@ -197,7 +196,7 @@ CommitResult commit(Transaction txn) { || Ref.same(txn.baseRoots().get(EngineSlots.REQUESTS), branch.roots().get(EngineSlots.REQUESTS)); boolean serializable = txn.options().isolation() == Isolation.SERIALIZABLE; fast = branch.roots().sameAs(txn.baseRoots()) || (requestsStable && txn.workspace().untouchedSince(branch.roots(), serializable)); - if (!fast && txn.generation() < relocationFence && txn.carriesStoredRoots()) { + if (txn.compactionsAtStart() < compactions.get() && txn.carriesStoredRoots()) { conflicts.increment(); throw HStoreException.conflict("bulk-loaded roots of transaction " + txn.id() + " predate a compaction"); } @@ -305,22 +304,12 @@ private static WalRecord meta(long txn, Branch branch) { return new WalRecord.BranchMeta(txn, branch.id(), branch.name(), branch.parent(), branch.baseGeneration(), branch.createdAt(), branch.state()); } - public long rewrite(int branchId, UnaryOperator rewrite) { - Appended appended; - commitLock.lock(); - try { - Generation latest = head.get(); - Branch branch = latest.branch(branchId); - RootVector roots = rewrite.apply(branch.roots()); - if (roots.sameAs(branch.roots())) { - return latest.id(); - } - appended = append(latest, branch.withRoots(roots), txns.getAndIncrement(), List.of(), List.of(), List.of(), Long.MAX_VALUE); - relocationFence = appended.pending().generation().id(); - } finally { - commitLock.unlock(); - } - return committer.await(appended.pending()).id(); + public void compacted() { + compactions.incrementAndGet(); + } + + long compactions() { + return compactions.get(); } public T exclusive(Supplier action) { diff --git a/engine/src/test/java/io/hstore/engine/CompactionTest.java b/engine/src/test/java/io/hstore/engine/CompactionTest.java new file mode 100644 index 0000000..061a292 --- /dev/null +++ b/engine/src/test/java/io/hstore/engine/CompactionTest.java @@ -0,0 +1,285 @@ +package io.hstore.engine; + +import io.hstore.engine.maintenance.Compactor; +import io.hstore.engine.page.SegmentInfo; +import io.hstore.engine.page.SegmentState; +import io.hstore.engine.topology.EdgeKind; +import io.hstore.engine.topology.Incidence; +import io.hstore.engine.txn.CrashPoint; +import io.hstore.engine.txn.Snapshot; +import io.hstore.engine.txn.Transaction; + +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.io.TempDir; +import org.junit.jupiter.params.ParameterizedTest; +import org.junit.jupiter.params.provider.Arguments; +import org.junit.jupiter.params.provider.MethodSource; + +import java.nio.file.Files; +import java.nio.file.Path; +import java.util.ArrayList; +import java.util.List; +import java.util.Set; +import java.util.TreeSet; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.Future; +import java.util.concurrent.atomic.AtomicBoolean; +import java.util.concurrent.atomic.AtomicReference; +import java.util.stream.Collectors; +import java.util.stream.Stream; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertThrows; +import static org.junit.jupiter.api.Assertions.assertTrue; + +class CompactionTest { + + private static final int ROUNDS = 60; + + @TempDir + Path directory; + + private static EngineOptions options() { + return EngineTest.small().withPagesPerSegment(64).withHistoryLimit(2).withCheckpointWalBytes(Long.MAX_VALUE); + } + + @Test + void compactionMovesPagesWithoutWritingAGeneration() { + try (StorageEngine engine = StorageEngine.open(directory, options())) { + long edge = churn(engine); + long generation = engine.stats().generation(); + Set before = segmentIds(engine); + Compactor.Report report = engine.compact(); + assertFalse(report.compacted().isEmpty()); + assertEquals(generation, engine.stats().generation(), "compaction must not commit"); + Set after = segmentIds(engine); + report.compacted().forEach(victim -> assertFalse(after.contains(victim), "segment " + victim + " still exists")); + assertTrue(after.size() < before.size(), "segments " + before + " -> " + after); + assertEquals(expected(), members(engine, edge)); + } + try (StorageEngine engine = StorageEngine.open(directory, options())) { + assertTrue(engine.liveness().keySet().stream().allMatch(segmentIds(engine)::contains)); + assertEquals(expected(), members(engine, onlyEdge(engine))); + } + } + + @Test + void retainedHistoryStaysReadableOnceItsSegmentsAreDeleted() { + EngineOptions options = options().withHistoryLimit(20); + List written = Stream.iterate(500_000L, i -> i + 1).limit(2_000).toList(); + long large; + long edge; + long earlier; + try (StorageEngine engine = StorageEngine.open(directory, options)) { + large = engine.write(txn -> { + long created = txn.createEdge(EngineTest.EDGE, EdgeKind.SET); + written.forEach(member -> txn.insert(created, member)); + return created; + }); + Set early = segmentIds(engine); + edge = churn(engine); + earlier = engine.stats().generation() - 10; + Compactor.Report report = engine.compact(); + assertTrue(report.compacted().stream().anyMatch(early::contains), "no early segment was compacted: " + report.compacted()); + report.compacted().forEach(victim -> assertFalse(segmentIds(engine).contains(victim))); + } + try (StorageEngine engine = StorageEngine.open(directory, options); + Snapshot old = engine.snapshotAt(earlier, 0)) { + assertEquals(written, old.requireEdge(large).stream().map(Incidence::member).toList()); + assertEquals(afterRound(ROUNDS - 11), old.requireEdge(edge).stream().map(Incidence::member).toList()); + } + } + + @Test + void aReaderFromBeforeTheMoveKeepsTheOldSegmentAround() { + try (StorageEngine engine = StorageEngine.open(directory, options())) { + long edge = churn(engine); + Compactor.Report report; + try (Snapshot reader = engine.snapshot()) { + report = engine.compact(); + assertFalse(report.compacted().isEmpty()); + for (int victim : report.compacted()) { + SegmentInfo info = engine.stats().segments().stream().filter(segment -> segment.id() == victim).findFirst().orElseThrow(); + assertEquals(SegmentState.RETIRED, info.state()); + } + assertEquals(expected(), reader.requireEdge(edge).stream().map(Incidence::member).toList()); + } + engine.compact(); + report.compacted().forEach(victim -> assertFalse(segmentIds(engine).contains(victim))); + } + try (StorageEngine engine = StorageEngine.open(directory, options())) { + assertEquals(expected(), members(engine, onlyEdge(engine))); + } + } + + @Test + void commitsCarryOnWhilePagesMove() throws Exception { + List all; + try (StorageEngine engine = StorageEngine.open(directory, options())) { + long edge = churn(engine); + long conflicts = engine.stats().conflicts(); + AtomicBoolean compacting = new AtomicBoolean(true); + List written = new ArrayList<>(); + try (ExecutorService writer = Executors.newSingleThreadExecutor()) { + Future writes = writer.submit(() -> { + long next = 1_000_000; + while (compacting.get() || written.size() < 50) { + long member = next++; + engine.write(txn -> { + txn.insert(edge, member); + return null; + }); + written.add(member); + } + return null; + }); + for (int i = 0; i < 3; i++) { + engine.compact(); + } + compacting.set(false); + writes.get(); + } + assertEquals(conflicts, engine.stats().conflicts()); + TreeSet expected = new TreeSet<>(expected()); + expected.addAll(written); + all = List.copyOf(expected); + assertEquals(all, members(engine, edge)); + } + try (StorageEngine engine = StorageEngine.open(directory, options())) { + assertEquals(all, members(engine, onlyEdge(engine))); + } + } + + @Test + void aBulkLoadStartedBeforeCompactionHasToRetry() { + try (StorageEngine engine = StorageEngine.open(directory, options())) { + long edge = churn(engine); + long target = engine.write(txn -> txn.createEdge(EngineTest.EDGE, EdgeKind.SET)); + List members = Stream.iterate(2_000_000L, i -> i + 1).limit(5_000).map(Incidence::of).toList(); + Transaction early = engine.begin(); + early.load(target, members); + assertFalse(engine.compact().compacted().isEmpty()); + assertThrows(HStoreException.Conflict.class, early::commit); + engine.write(txn -> { + txn.load(target, members); + return null; + }); + assertEquals(5_000L, (long) engine.read(snapshot -> snapshot.requireEdge(target).size())); + assertEquals(expected(), members(engine, edge)); + } + } + + @Test + void aBulkLoadThatCommitsWhilePagesMoveHasToRetry() { + AtomicReference pending = new AtomicReference<>(); + AtomicReference refused = new AtomicReference<>(); + EngineOptions options = options().withFaults(point -> { + Transaction early = point == CrashPoint.COMPACTION_MOVE ? pending.getAndSet(null) : null; + if (early != null) { + refused.set(assertThrows(HStoreException.Conflict.class, early::commit)); + } + }); + List members = Stream.iterate(2_000_000L, i -> i + 1).limit(50_000).map(Incidence::of).toList(); + long target; + long edge; + try (StorageEngine engine = StorageEngine.open(directory, options)) { + target = engine.write(txn -> txn.createEdge(EngineTest.EDGE, EdgeKind.SET)); + Transaction early = engine.begin(); + early.load(target, members); + edge = churn(engine); + pending.set(early); + assertFalse(engine.compact().compacted().isEmpty()); + assertTrue(refused.get() != null, "the load committed while its pages were left behind"); + engine.write(txn -> { + txn.load(target, members); + return null; + }); + } + try (StorageEngine engine = StorageEngine.open(directory, options())) { + assertEquals(50_000L, (long) engine.read(snapshot -> snapshot.requireEdge(target).size())); + assertEquals(expected(), members(engine, edge)); + } + } + + static Stream crashes() { + return Stream.of(CrashPoint.COMPACTION_MOVE, CrashPoint.COMPACTION_RETIRE) + .flatMap(point -> Stream.of(Arguments.of(point, false), Arguments.of(point, true))); + } + + @ParameterizedTest + @MethodSource("crashes") + void aCrashDuringCompactionLosesNothing(CrashPoint point, boolean directoryWritesLost) throws Exception { + AtomicBoolean armed = new AtomicBoolean(false); + EngineOptions options = options().withFaults(reached -> { + if (reached == point && armed.compareAndSet(true, false)) { + throw new CrashPoint.SimulatedCrash(point); + } + }); + Path addresses = directory.resolve("data").resolve("segments").resolve("pages.dir"); + long edge; + try (StorageEngine engine = StorageEngine.open(directory, options)) { + edge = churn(engine); + } + byte[] checkpointed = Files.readAllBytes(addresses); + StorageEngine engine = StorageEngine.open(directory, options); + armed.set(true); + assertThrows(CrashPoint.SimulatedCrash.class, engine::compact); + engine.halt(); + if (directoryWritesLost) { + Files.write(addresses, checkpointed); + } + try (StorageEngine recovered = StorageEngine.open(directory, options())) { + assertEquals(expected(), members(recovered, edge)); + recovered.compact(); + assertEquals(expected(), members(recovered, edge)); + } + try (StorageEngine reopened = StorageEngine.open(directory, options())) { + assertEquals(expected(), members(reopened, edge)); + } + } + + private static long churn(StorageEngine engine) { + long edge = engine.write(txn -> txn.createEdge(EngineTest.EDGE, EdgeKind.SET)); + churn(engine, edge); + return edge; + } + + private static void churn(StorageEngine engine, long edge) { + for (int round = 0; round < ROUNDS; round++) { + int base = round * 50; + engine.write(txn -> { + for (int i = 0; i < 50; i++) { + txn.insert(edge, 10_000 + base + i); + } + if (base >= 100) { + for (int i = 0; i < 50; i++) { + txn.remove(edge, 10_000 + base - 100 + i); + } + } + return null; + }); + } + } + + private static List expected() { + return afterRound(ROUNDS - 1); + } + + private static List afterRound(int round) { + return Stream.iterate(10_000L + (round - 1) * 50L, i -> i + 1).limit(100).toList(); + } + + private static List members(StorageEngine engine, long edge) { + return engine.read(snapshot -> snapshot.requireEdge(edge).stream().map(Incidence::member).toList()); + } + + private static long onlyEdge(StorageEngine engine) { + return engine.read(snapshot -> snapshot.atomsOfType(0, EngineTest.EDGE).findFirst().orElseThrow()); + } + + private static Set segmentIds(StorageEngine engine) { + return engine.stats().segments().stream().map(SegmentInfo::id).collect(Collectors.toSet()); + } +} diff --git a/engine/src/test/java/io/hstore/engine/CrashRecoveryTest.java b/engine/src/test/java/io/hstore/engine/CrashRecoveryTest.java index a34fe4e..61cd5bf 100644 --- a/engine/src/test/java/io/hstore/engine/CrashRecoveryTest.java +++ b/engine/src/test/java/io/hstore/engine/CrashRecoveryTest.java @@ -18,6 +18,7 @@ import java.nio.file.Path; import java.nio.file.StandardOpenOption; import java.util.Comparator; +import java.util.EnumSet; import java.util.List; import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; @@ -35,7 +36,7 @@ class CrashRecoveryTest { Path directory; static Stream crashMatrix() { - return Stream.of(WalMode.values()).flatMap(mode -> Stream.of(CrashPoint.values()).map(point -> Arguments.of(mode, point))); + return Stream.of(WalMode.values()).flatMap(mode -> EnumSet.range(CrashPoint.PAGE, CrashPoint.CATALOG_PUBLISH).stream().map(point -> Arguments.of(mode, point))); } @ParameterizedTest