diff --git a/docs/architecture/overview.md b/docs/architecture/overview.md
index 308d446..d02f451 100644
--- a/docs/architecture/overview.md
+++ b/docs/architecture/overview.md
@@ -83,7 +83,7 @@ The engine has no dependencies outside `java.base`. All three modules build into
| Package | Purpose |
|---|---|
| `io.hstore.engine` | `StorageEngine` (open/close, maintenance thread, checkpoint/compact entry points), `EngineOptions`, `EngineStats`, `Dictionary` (interned role names and role sets), `HStoreException` (error codes, retryable conflicts). |
-| `engine.page` | Byte-level storage. `PageStore` and `SegmentFile` hold segment files and allocation. `PageHeader` is the 80-byte header and crc32c seal. `PageId` packs page addresses. `ByteCursor` provides little-endian and varint codecs over `MemorySegment`. `IoTrace` counts page visits per query through a `ScopedValue`. |
+| `engine.page` | Byte-level storage. `PageStore` and `SegmentFile` hold segment files and allocation. `PageHeader` is the 80-byte header and crc32c seal. `PageId` is a page's permanent number, `PageAddress` its place in a segment, and `PageDirectory` the memory-mapped `pages.dir` that maps one to the other. `ByteCursor` provides little-endian and varint codecs over `MemorySegment`. `IoTrace` counts page visits per query through a `ScopedValue`. |
| `engine.tree` | The copy-on-write counted B+tree. `Tree` is the immutable handle and `TreeWriter` does path-copy updates. Nodes are `Leaf` and `Branch`, referenced by `Ref` (`Empty`, `Pending` or `Stored`). Also here: `Summary` (monoid annotations), `TreeCursor`, `TreeAlgebra` (leapfrog, set algebra), `TreeDiff`, `BulkBuilder`, `NodeCodec`, `NodeCache`, `Materializer`, `TreeWalker` and `TreeVerifier`. |
| `engine.topology` | Hyperedge representation: `Hyperedge` (members tree plus order index), `Incidence` (one member's role set, weight, validity, payload and qualifier), `IncidenceCodec` (columnar leaf encoding), `MemberChange`, `Weight` (fixed point) and `Validity`. |
| `engine.index` | `PostingIndex`, a two-level index from a key to a postings set. Small sets are inlined and large ones are promoted to their own tree. Used by the reverse incidence, type, canonical-key and symbol indexes. |
@@ -184,7 +184,7 @@ sequenceDiagram
TM->>TM: fast path when base roots or touched slots are unchanged,
otherwise validateReads and replay the op log on the latest roots (rebase)
TM->>M: materialize each pending root (post-order)
M->>M: encode node image (pageId 0)
- M->>PS: allocate(epoch = generation, length): 64-byte units
+ M->>PS: allocate(length): next page number, 64-byte units, directory entry
M->>M: PageHeader.assign: patch pageId, reseal crc32c
M-->>TM: Stored refs, WAL Page / PageRef records, node images
TM->>WAL: append Begin, Page/PageRef*, Root*, BranchMeta*, Feed
@@ -201,7 +201,7 @@ sequenceDiagram
GC-->>TX: await() returns published generation
```
-The ordering above is what makes recovery safe. Under `PAGE_REFERENCES` the WAL carries only `(pageId, length, crc32c)` for each new page. The commit leader therefore makes data pages durable *before* the log records that reference them. Recovery replays committed transactions in order and stops at the first commit whose referenced pages fail verification (the *durable-prefix* rule in `Recovery.intact`). The full protocol and the crash-point matrix are in [wal-and-recovery.md](../transactions/wal-and-recovery.md).
+The ordering above is what makes recovery safe. Under `PAGE_REFERENCES` the WAL carries only `(pageId, address, length, crc32c)` for each new page. The commit leader therefore makes data pages durable *before* the log records that reference them. Recovery replays committed transactions in order and stops at the first commit whose referenced pages fail verification (the *durable-prefix* rule in `Recovery.intact`). The full protocol and the crash-point matrix are in [wal-and-recovery.md](../transactions/wal-and-recovery.md).
## Threading model
@@ -227,13 +227,14 @@ Readers take no locks at all.
The following is real output from `hstore init` followed by loading `examples/clinical-claims.hql` and running `CHECKPOINT;` (default 16 KiB pages):
```text
-FORMAT 55 B format=3, page-size=16384 (java.util.Properties)
+FORMAT 55 B format=4, page-size=16384 (java.util.Properties)
LOCK 0 B FileChannel.tryLock() held while open
hstore.conf 2397 B server configuration template written by `hstore init`
catalog/manifest 20 B name of the latest catalog file
catalog/generations/0000000000000000.cat 37 B checkpoint of generation 0
catalog/generations/000000000000009d.cat 49147 B checkpoint of generation 157 (current + retained history)
data/segments/00000001.seg 1210439 B segment 1: packed node images
+data/segments/pages.dir 8388608 B page directory, one 8 MiB chunk mapped so far (sparse)
wal/segments/0000000000000000.wal 156180 B WAL segment starting at LSN 0
feed/0000000000000000.feed 29846 B change feed segment starting at byte 0
payload/000001.pay 0 B payload segment (database layer)
@@ -242,11 +243,12 @@ semantic/state 53 B persisted HNSW index state (datab
| Path | Owner | Format |
|---|---|---|
-| `FORMAT` | `StorageEngine.verifyFormat` | `Properties` with `format=3` (packed extents, sized references) and `page-size` (the maximum node size). Opening a directory written in an older format, or opening with a different page size, fails. |
+| `FORMAT` | `StorageEngine.verifyFormat` | `Properties` with `format=4` (page numbers resolved through a page directory) and `page-size` (the maximum node size). Opening a directory written in an older format, or opening with a different page size, fails. |
| `LOCK` | `StorageEngine.open` | An exclusive OS file lock. A second process gets `database ... is opened by another process`. |
| `catalog/generations/%016x.cat` | `CatalogStore.publish` | `int magic 0x54414348`, `int crc32c(payload)`, `long length`, then the `CatalogImage` payload. Written to `*.tmp`, fsynced, atomically renamed, and the directory fsynced. The three newest files are kept. |
| `catalog/manifest` | `CatalogStore` | The file name of the latest `.cat`, written durably the same way. |
| `data/segments/%08x.seg` | `PageStore` / `SegmentFile` | Append-only extents of node images packed at 64-byte alignment, up to 256 MiB per segment by default (1 GiB maximum); see [pages.md](../storage/pages.md). |
+| `data/segments/pages.dir` | `PageDirectory` | 64-byte header, then one `i64` address (segment, offset, units) per page number, memory-mapped in 8 MiB chunks; see [pages.md](../storage/pages.md#page-numbers-and-the-page-directory). |
| `wal/segments/%016x.wal` | `WriteAheadLog` | Each segment is named by its starting LSN; see [wal-and-recovery.md](../transactions/wal-and-recovery.md). |
| `feed/%016x.feed` | `ChangeFeed` | Segments (64 MiB, named by starting byte offset) of frames `int length`, `int crc32c`, `long generation`, `payload`, trimmed by retention after checkpoints; see [change-feed.md](../storage/change-feed.md). |
| `payload/%06x.pay` | `db.payload.PayloadStore` | Append-only payload extents. |
diff --git a/docs/development.md b/docs/development.md
index f9268a3..a2677ca 100644
--- a/docs/development.md
+++ b/docs/development.md
@@ -24,7 +24,7 @@ flowchart LR
| Package | Contents |
|---|---|
-| `engine.page` | Segment files, 80-byte page header with CRC32C, `PageId` (segment 24 / offset 24 / generation 16 bits), page store and cache |
+| `engine.page` | Segment files, 80-byte page header with CRC32C, `PageId` page numbers, `PageAddress` (segment 24 / offset 24 / units 16 bits), the `pages.dir` page directory, page store and cache |
| `engine.tree` | Copy-on-write counted B+tree, monoid summaries, write scopes, cursors, `TreeAlgebra` (leapfrog intersection, synchronized traversal), `TreeDiff`, `TreeVerifier` |
| `engine.topology` | Hyperedge membership and incidence encodings, `Incidence`, `Validity` |
| `engine.catalog` | Slots, root vectors, branches, generations, dictionary |
diff --git a/docs/operations/configuration.md b/docs/operations/configuration.md
index be4b152..76df7e7 100644
--- a/docs/operations/configuration.md
+++ b/docs/operations/configuration.md
@@ -90,7 +90,7 @@ max_connections = 200
|---|---|---|---|
| `page_size` | `HSTORE_PAGE_SIZE` | `16384` | Page size in bytes. Must be a power of two from 1 KiB to 1 MiB. It is recorded in `/FORMAT` at creation, and opening with a different value fails with `database uses N byte pages, options request M`. |
| `durability` | `HSTORE_DURABILITY` | `sync` | `sync`: a commit is acknowledged after the group-commit barrier has `fsync`ed it. `async`: acknowledged once appended, so a crash can lose a suffix of acknowledged commits but never corrupts the database. |
-| `wal_mode` | `HSTORE_WAL_MODE` | `references` | `references`: the WAL logs `PageRef(pageId, length, crc32c)` records and data pages are synced before the WAL. `images`: the WAL logs full page images. See [WAL and recovery](../transactions/wal-and-recovery.md). |
+| `wal_mode` | `HSTORE_WAL_MODE` | `references` | `references`: the WAL logs `PageRef(pageId, address, length, crc32c)` records and data pages are synced before the WAL. `images`: the WAL logs full page images. See [WAL and recovery](../transactions/wal-and-recovery.md). |
| `node_cache_mb` | `HSTORE_NODE_CACHE_MB` | `256` | Megabytes of Java heap for decoded tree nodes, each counted at an estimate of the memory it takes. |
| `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. |
diff --git a/docs/operations/docker.md b/docs/operations/docker.md
index d93b415..1b409a8 100644
--- a/docs/operations/docker.md
+++ b/docs/operations/docker.md
@@ -259,7 +259,7 @@ skipped because `FORMAT` exists. To verify a backup, run
`docker run --rm -v restored:/var/lib/hstore/data hstore check /var/lib/hstore/data`.
To upgrade, stop the container, back up the volume, and start the new image on the same volume. `FORMAT`
-records the on-disk format version (`format=3`, sized page references) and the page size. Both are checked at
+records the on-disk format version (`format=4`, page directory) and the page size. Both are checked at
open. A directory written in another format is refused with
`database … uses storage format N; this build reads format 2 (packed segment extents); export and reload it`,
so the server never misreads it. Export the data with HQL against the old version and reload it into a fresh
diff --git a/docs/operations/logging.md b/docs/operations/logging.md
index b972412..084e181 100644
--- a/docs/operations/logging.md
+++ b/docs/operations/logging.md
@@ -109,7 +109,7 @@ and in Docker it is the last line of `docker logs`. The storage-integrity failur
| Message | Meaning |
|---|---|
-| `database uses storage format N; this build reads format 3 (sized page references); export and reload it` | `FORMAT` was written by an incompatible build. |
+| `database uses storage format N; this build reads format 4 (page directory); export and reload it` | `FORMAT` was written by an incompatible build. |
| `database uses N byte pages, options request M` | `page_size` differs from the value fixed at `init`. |
| `write-ahead log segment is damaged () but later segments exist; replay would silently drop committed transactions` | `CORRUPT_LOG`: damage in a non-tail WAL segment. A torn record at the end of the *last* segment is normal after a crash. Replay simply ends there. Damage before later segments means storage corruption, so recovery refuses to guess. Restore from backup. |
diff --git a/docs/storage/maintenance.md b/docs/storage/maintenance.md
index 6eaf812..5c15ff5 100644
--- a/docs/storage/maintenance.md
+++ b/docs/storage/maintenance.md
@@ -38,7 +38,7 @@ Every checkpoint is logged, and then the checkpoint listeners run. The database
6. `wal.truncateBefore(lsn)` deletes WAL segment files that end before the checkpoint LSN. They are no longer needed, because the catalog image now covers them.
7. It remembers `lastLsn = lsn`.
-**Trigger threshold.** `walSinceCheckpoint() = wal.end() − lastLsn` is compared with `checkpoint_wal_mb` (default 256 MiB, `EngineOptions.checkpointWalBytes`). In the default `PAGE_REFERENCES` WAL mode a commit logs about 20 bytes per new node image instead of the image itself. The threshold therefore corresponds to far more committed work than in `PAGE_IMAGES` mode, and replay after a crash stays short.
+**Trigger threshold.** `walSinceCheckpoint() = wal.end() − lastLsn` is compared with `checkpoint_wal_mb` (default 256 MiB, `EngineOptions.checkpointWalBytes`). In the default `PAGE_REFERENCES` WAL mode a commit logs about 47 bytes per new node image instead of the image itself. The threshold therefore corresponds to far more committed work than in `PAGE_IMAGES` mode, and replay after a crash stays short.
**What recovery gains.** Recovery loads the newest valid catalog image and replays only WAL records after `checkpointLsn` ([wal-and-recovery.md](../transactions/wal-and-recovery.md)). The image's persisted history makes time travel survive restarts. Its segment table restores the allocator position (`units`) and the compaction state machine.
diff --git a/docs/storage/pages.md b/docs/storage/pages.md
index dfb7f73..8c5511e 100644
--- a/docs/storage/pages.md
+++ b/docs/storage/pages.md
@@ -37,7 +37,7 @@ Sizes are computed exactly before writing (`varLongSize`, `signedVarLongSize`, `
| 5 | 1 | `type` | `PageType`: `1` = `INTERNAL` (branch), `2` = `LEAF` |
| 6 | 1 | `schema` | `TreeSchema.id` (1–255) of the tree the node belongs to |
| 7 | 1 | `flags` | Currently always `0` |
-| 8 | 8 | `pageId` | The image's own packed address (below). It is written last, after allocation (`PageHeader.assign`). |
+| 8 | 8 | `pageId` | The image's own page number (below). It is written last, after allocation (`PageHeader.assign`). |
| 16 | 8 | `creationEpoch` | Id of the generation whose commit wrote the image. Spilled pages use `head + 1`. |
| 24 | 4 | `payloadLength` | Payload bytes after the header. Readers use it to size the second read. |
| 28 | 4 | `slotCount` | Number of entries (leaf) or children (branch) |
@@ -57,7 +57,7 @@ Sizes are computed exactly before writing (`varLongSize`, `signedVarLongSize`, `
* the magic is wrong;
* the format is not `1`;
* the type is unknown;
-* `pageId != expectedPageId` (a misdirected read, or a stale address pointing into reused bytes);
+* `pageId != expectedPageId` (a directory entry that points at the wrong image);
* `80 + payloadLength` exceeds the bytes read;
* the CRC does not match.
@@ -96,14 +96,14 @@ Leaf value formats in use:
| posting directories (#5, #7, #9, #12, …) | `index/PostingsCodec` | `u8 0 \| varint n \| delta keys \| values` (inline) or `u8 1 \| Ref` (promoted) |
| simple rows | `ValueCodec.rows(size, writer, reader)` | values written one after another |
-A worked example, produced by running `Materializer` over a two-entry `order-index` leaf (`{17 → 16777216, 18 → 33554432}`) allocated at `1:1234@157`. The image is 139 bytes, so it occupies 3 units (192 bytes):
+A worked example, produced by running `Materializer` over a two-entry `order-index` leaf (`{17 → 16777216, 18 → 33554432}`) written as page `#1234` by generation 157. The image is 139 bytes, so it occupies 3 units (192 bytes):
```text
-00 48 53 50 47 01 02 04 00 9d 00 d2 04 00 01 00 00 magic HSPG | format 1 | LEAF | schema 4 | flags 0 | pageId (le) = 0x10004d2009d
+00 48 53 50 47 01 02 04 00 00 00 d2 04 00 00 00 00 magic HSPG | format 1 | LEAF | schema 4 | flags 0 | pageId (le) = 0x4d20000
10 9d 00 00 00 00 00 00 00 3b 00 00 00 02 00 00 00 epoch 157 | payloadLength 59 | slotCount 2
20 02 00 00 00 00 00 00 00 11 00 00 00 00 00 00 00 count 2 | lower 17
30 12 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 upper 18 | height 0 | reserved
-40 0a 2a 2f 32 04 9f 42 eb 15 bc b1 19 00 00 00 00 fingerprint | crc32c | reserved
+40 0a 2a 2f 32 04 9f 42 eb da c5 41 7e 00 00 00 00 fingerprint | crc32c | reserved
50 00 weightSum = zigzag(0)
51 fe ff ff ff ff ff ff ff ff 01 weightMin = zigzag(Long.MAX_VALUE) (EMPTY sentinel)
5b ff ff ff ff ff ff ff ff ff 01 weightMax = zigzag(Long.MIN_VALUE)
@@ -116,34 +116,65 @@ A worked example, produced by running `Materializer` over a two-entry `order-ind
An `EntryMeasure.keyed` measure never feeds weight or validity, so those fields keep the `Summary.EMPTY` sentinels. They round-trip exactly at a fixed cost of 40 bytes per node. Incidence trees, whose measure does feed them, encode real ranges there instead.
-## Addressing: packed `PageId`
+## Page numbers and the page directory
-A `PageId` (`page/PageId.java`) is a 64-bit value:
+A `PageId` (`page/PageId.java`) is a page's permanent number, not its location:
+
+```text
+ 63 16 15 0
+┌───────────────────────────────────────────────┬────────────────┐
+│ index (48 bits) │ reuse (16 bits)│
+│ 1 .. 2⁴⁸ − 1 │ │
+└───────────────────────────────────────────────┴────────────────┘
+```
+
+* `index` starts at 1 and goes up by one for every image allocated. `0` is reserved, so `PageId.NONE = 0` means
+ "no page".
+* `reuse` is always 0 for now. It exists so a number can be handed out again later, once nothing can reach its old
+ image, and still differ from the id the old image was written with. `PageId.toString()` prints `#index`, or
+ `#index.reuse` when `reuse` isn't 0.
+
+Where an image is stored is a `PageAddress` (`page/PageAddress.java`):
```text
63 40 39 16 15 0
-┌───────────────────────┬───────────────────────┬───────────────┐
-│ segment (24 bits) │ offset (24 bits) │ generation │
-│ 1 .. 16 777 215 │ 64-byte units │ (16 bits) │
-└───────────────────────┴───────────────────────┴───────────────┘
+┌───────────────────────┬───────────────────────┬────────────────┐
+│ segment (24 bits) │ offset (24 bits) │ units (16 bits)│
+│ 1 .. 16 777 215 │ 64-byte units │ │
+└───────────────────────┴───────────────────────┴────────────────┘
```
-* `segment` is the segment file id; `0` is reserved, so `PageId.NONE = 0` means "no page".
-* `offset` counts **64-byte units** (`PageId.UNIT_BYTES = 64`) from the start of the segment file. The byte position is `offset × 64`. With 24 bits, one segment can address at most 2²⁴ × 64 B = 1 GiB.
-* `generation` is `creationEpoch & 0xFFFF`. It is not needed for addressing. It makes the id of an image written in a different epoch differ from one written at the same location, which the identity check in `verify` exploits, and it makes ids readable in diagnostics (`PageId.toString()` prints `segment:offset@generation`).
+* `segment` is the segment file id.
+* `offset` counts **64-byte units** (`PageId.UNIT_BYTES = 64`) from the start of the file, so the byte position is
+ `offset × 64`. With 24 bits, one segment holds at most 2²⁴ × 64 B = 1 GiB.
+* `units` is the image length rounded up to whole units. 16 bits allow images up to 4 MiB, far above any page size.
+
+The **page directory** (`page/PageDirectory.java`) maps numbers to addresses. It is the file
+`/data/segments/pages.dir`: a 64-byte header (`i32 magic 0x52494450`, `i32 format 1`, `i64` next index),
+then one little-endian `i64` address per index, entry `i` at byte `64 + 8 × i`. An entry of 0 means the number has
+no image. The file is memory-mapped in 8 MiB chunks, and a new chunk is mapped when the file grows into it, so a
+lookup is a load from the OS page cache and the directory takes no Java heap. A million pages need 8 MB of it.
-Worked encode/decode: segment 1, unit offset 1234, generation 157.
+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)).
+
+Page `#1234`, stored as 3 units at unit 1234 of segment 1:
```text
-pack = (1 << 40) | (1234 << 16) | 157
- = 0x0000_0100_04D2_009D = 1 099 592 499 357
-byte offset in 00000001.seg = 1234 × 64 = 78 976
-segmentOf(p) = p >>> 40 = 1
-offsetOf(p) = (p >>> 16) & 0xFFFFFF = 1234
-generationOf(p) = p & 0xFFFF = 157
+pageId = 1234 << 16 = 0x0000_0000_04D2_0000
+address = (1 << 40) | (1234 << 16) | 3 = 0x0000_0100_04D2_0003
+directory entry = byte 64 + 8 × 1234 = byte 9 936 of pages.dir
+image = byte 1234 × 64 = byte 78 976 of 00000001.seg
```
-A generation above 65 535 wraps: generation 70 000 is stored as 4 464.
+Entries are written when a page is allocated and forced to disk only at checkpoints (`PageStore.checkpoint`). The
+commit path never syncs the directory, because every WAL record for a page carries its address too. Recovery
+writes each logged address back into the directory before it checks or replays the page, so an entry that a crash
+lost comes back from the log. The header's next index is updated on every allocation, and recovery raises it past
+every number it restores, so a restored number is never handed out a second time.
## Segment files
@@ -164,7 +195,7 @@ A generation above 65 535 wraps: generation 70 000 is stored as 4 464.
The list is persisted in each catalog image ([catalog-and-generations.md](catalog-and-generations.md#catalog-images-checkpoints)) and restored by `PageStore.open`. On open, each `.seg` file found on disk gets `units = max(recorded units, ⌈file size / 64⌉)`. A file with no catalog entry (for example, one created after the last checkpoint and abandoned by a crash) is registered as `SEALED` with `pages = units = ⌈file size / 64⌉`. That is an upper bound on its image count, so its live ratio is underestimated and the segment stays a compaction candidate instead of being stranded. The newest `ACTIVE` segment resumes allocation at its recorded `units`.
-### Allocation (`PageStore.allocate(epoch, length)`)
+### Allocation (`PageStore.allocate(length)`)
Allocation happens **after** encoding, because the exact length is only known then:
@@ -176,23 +207,27 @@ sequenceDiagram
participant H as PageHeader
M->>NC: encode(frozen, PageId.NONE, epoch, pageSize)
NC-->>M: image (80 + payloadLength bytes, sealed with pageId 0)
- M->>PS: allocate(epoch, image.byteSize())
+ M->>PS: allocate(image.byteSize())
PS->>PS: units = ⌈len / 64⌉; roll the segment if nextUnit + units > segmentUnits
- PS-->>M: pack(activeSegment, nextUnit, epoch); nextUnit += units
+ PS->>PS: index = next index; directory[index] = (activeSegment, nextUnit, units)
+ PS-->>M: pack(index, 0); nextUnit += units
M->>H: assign(image, pageId): write pageId at offset 8, recompute the crc
M->>M: sink.accept(pageId, image, frozen): WAL record, queued write, cache admission
```
* `allocate` rejects `length > pageSize`.
-* Allocation is serialized by a `ReentrantLock`, and only the counter update runs under it.
+* Allocation is serialized by a `ReentrantLock`, and only the counter and directory updates run under it.
* Rolling a segment marks the previous one `SEALED`, creates the next id, and resets `nextUnit` to 0.
-* `PageStore.write(pageId, image)` performs a positional `FileChannel.write` at `offset × 64` and marks the file dirty. The write happens later, inside `TransactionManager.append`, after the WAL records have been appended (see [wal-and-recovery.md](../transactions/wal-and-recovery.md)).
-* `PageStore.sync()` calls `FileChannel.force(false)` on every dirty segment. The group committer calls it once per batch in `PAGE_REFERENCES` WAL mode, before the WAL sync.
+* `PageStore.write(pageId, image)` looks the number up in the directory and performs a positional `FileChannel.write` at `offset × 64`, then marks the file dirty. It refuses an image longer than the units allocated for it. The write happens later, inside `TransactionManager.append`, after the WAL records have been appended (see [wal-and-recovery.md](../transactions/wal-and-recovery.md)).
+* `PageStore.sync()` calls `FileChannel.force(false)` on every dirty segment. The group committer calls it once per batch in `PAGE_REFERENCES` WAL mode, before the WAL sync. `PageStore.checkpoint()` does the same and then forces the page directory; only `Checkpointer` calls it.
Compared with fixed page slots, packing removes almost all internal fragmentation for the many small nodes a hypergraph produces. A two-member hyperedge's member tree is one leaf of roughly 100–150 bytes, which now occupies 2–3 units instead of a full 16 KiB page.
### Reading (`SegmentFile.read(position, maximum)`)
+`PageStore.read(pageId)` looks the number up in the directory, then asks that segment for at most `units × 64` bytes
+at `offset × 64`. A number with no entry fails with `corrupt(pageId, "page has no address")`.
+
Each segment file has a read-only memory map. If the map covers the image, `read` copies exactly
`80 + payloadLength` bytes out of it into a new heap array and returns that. There is no system call, and the raw
bytes stay in the OS page cache rather than on the Java heap, which only holds the decoded nodes in `NodeCache`.
@@ -220,9 +255,9 @@ ever appended to or deleted, never truncated, so a map can't end up pointing pas
`pread` doesn't know how long an image is until it has the header, so it takes one or two system calls:
-1. Allocate a buffer of `min(pageSize, 4096)` bytes and read it at `offset × 64`.
+1. Allocate a buffer of `min(units × 64, 4096)` bytes and read it at `offset × 64`.
2. If at least 80 bytes came back, compute `length = 80 + payloadLength` from header offset 24:
- * if `length` is larger than `pageSize`, the header is corrupt; return the buffer and let `verify` fail it;
+ * if `length` is larger than `units × 64`, the header is corrupt; return the buffer and let `verify` fail it;
* if the first read already has `length` bytes, return those;
* otherwise allocate a buffer of exactly `length` bytes, copy in what was read, and read the rest.
@@ -272,8 +307,8 @@ When `pagesRead + cacheHits` exceeds the budget, it throws `HStoreException.limi
For each materialized image, the commit appends one WAL record ([wal-and-recovery.md](../transactions/wal-and-recovery.md)):
-* `PAGE_REFERENCES` (the default) appends `PageRef(txnId, pageId, length, crc32c(image))`, about 20 bytes. The image itself is written only to its segment, and it must be durable before the WAL record that refers to it (the data-before-log ordering in `TransactionManager.makeDurable`). Recovery accepts a committed transaction only if every referenced image reads back with a valid header and matching CRC (`Recovery.intact`). Replay stops at the first commit that fails this check.
-* `PAGE_IMAGES` appends `Page(txnId, pageId, image bytes)`. Recovery rewrites the image at its address, and the data file needs no sync before the WAL.
+* `PAGE_REFERENCES` (the default) appends `PageRef(txnId, pageId, address, length, crc32c(image))`, 47 or 48 bytes. The image itself is written only to its segment, and it must be durable before the WAL record that refers to it (the data-before-log ordering in `TransactionManager.makeDurable`). Recovery puts the address back into the directory, then accepts a committed transaction only if every referenced image reads back with a valid header and matching CRC (`Recovery.intact`). Replay stops at the first commit that fails this check.
+* `PAGE_IMAGES` appends `Page(txnId, pageId, address, image bytes)`. Recovery puts the address back into the directory and rewrites the image there, and the data file needs no sync before the WAL.
Because images are never overwritten in place, a torn write can only damage an image that no durable commit references yet. Reference mode is safe without full-page images.
@@ -281,18 +316,22 @@ Because images are never overwritten in place, a torn write can only damage an i
| Version | Where | Meaning |
|---|---|---|
-| `format=3` in `/FORMAT` | `StorageEngine.verifyFormat` | Packed 64-byte-unit extents, with image sizes in every stored reference. Directories written by older builds (`format=1` fixed page slots, `format=2` references without sizes) fail to open: `uses storage format 2; this build reads format 3 (sized page references); export and reload it`. |
+| `format=4` in `/FORMAT` | `StorageEngine.verifyFormat` | Page numbers resolved through `pages.dir`, images packed in 64-byte units, image sizes in every stored reference. Directories written by older builds (`format=1` fixed page slots, `format=2` references without sizes, `format=3` references by address) fail to open: `uses storage format 3; this build reads format 4 (page directory); export and reload it`. |
| `page-size=N` in `/FORMAT` | same | Fixed at creation. A mismatching `--page_size` is rejected. |
-| header byte 4 = `1` | `PageHeader.FORMAT` | Node image layout; unchanged by packing. |
+| header byte 4 = `1` | `PageHeader.FORMAT` | Node image layout; unchanged by packing and by page numbers. |
+| `pages.dir` header | `PageDirectory` | `i32 magic 0x52494450`, `i32 format 1`. |
| `SegmentInfo` in catalog images | `CatalogImage` | Now `(id, u8 state, pages, units, retiredAt)` |
## Inspecting images by hand
-The layout is simple enough to inspect with standard tools. To print the header of the image at `1:1234@157` (byte 1234 × 64 = 78 976 of segment 1):
+The layout is simple enough to inspect with standard tools. To print the header of page `#1234`, read its
+directory entry first, then the image it points at:
```bash
+xxd -s $((64 + 1234 * 8)) -l 8 -g 8 -e data/segments/pages.dir
+# 000026d0: 0000010004d20003 segment 1, unit 1234, 3 units
xxd -s $((1234 * 64)) -l 80 data/segments/00000001.seg
-# 00013480: 4853 5047 0102 0400 9d00 d204 0001 0000 HSPG, format 1, LEAF, schema 4, flags 0, pageId…
+# 00013480: 4853 5047 0102 0400 0000 d204 0000 0000 HSPG, format 1, LEAF, schema 4, flags 0, pageId…
```
`hstore check ` decodes and verifies every reachable image, including checksums, identity, ordering, balance and summaries. See [maintenance.md](maintenance.md#verification-hstore-check).
diff --git a/docs/transactions/wal-and-recovery.md b/docs/transactions/wal-and-recovery.md
index 8390f06..349591e 100644
--- a/docs/transactions/wal-and-recovery.md
+++ b/docs/transactions/wal-and-recovery.md
@@ -23,11 +23,12 @@ Related: [transactions](transactions.md), [change feed](../storage/change-feed.m
```text
/
- FORMAT properties: format=1, page-size=
+ FORMAT properties: format=4, page-size=
LOCK exclusive FileLock held while open
catalog/manifest name of the current catalog image file
catalog/generations/%016x.cat checkpointed catalog images (3 newest retained)
data/segments/*.seg page segments (see pages.md)
+ data/segments/pages.dir page directory: the address of every page number (see pages.md)
wal/segments/%016x.wal WAL segments, named by their starting LSN in hex
feed/%016x.feed change feed segments, named by global byte position (see change-feed.md)
```
@@ -90,37 +91,38 @@ Only the tail can legitimately be torn, because every earlier segment was forced
| Type | Record | Payload layout | Emitted by |
|---|---|---|---|
| 1 | `Begin(txnId, branch, baseGeneration)` | `varint branch`, `varlong baseGeneration` | first record of every commit (`append`) |
-| 2 | `Page(txnId, pageId, image)` | `i64 pageId`, `blob image` | `PAGE_IMAGES` mode, one per materialized page |
+| 2 | `Page(txnId, pageId, address, image)` | `i64 pageId`, `i64 address`, `blob image` | `PAGE_IMAGES` mode, one per materialized page |
| 3 | `Root(txnId, branch, slot, root)` | `varint branch`, `varint slot`, `Ref root` | one per slot whose root changed (`Ref.same` false), in ascending slot order |
| 4 | `BranchMeta(txnId, id, name, parent, baseGeneration, createdAt, state)` | `varint id`, `string name`, `varint parent`, `varlong baseGeneration`, `varlong createdAt`, `u8 state` (`ACTIVE=0`, `MERGED=1`, `DROPPED=2`) | `createBranch`, `closeBranch` |
| 5 | `Feed(txnId, payload)` | `blob` containing a `FeedCodec`-encoded `CommitEvent` | commits with at least one member or slot change |
| 6 | `Commit(txnId, generation, wallTime, nextAtom, nextTxn)` | `varlong generation`, `varlong wallTime`, `varlong nextAtom`, `varlong nextTxn` | last record of every commit, appended separately |
| 7 | `Abort(txnId)` | empty | `Transaction.abort()` via `TransactionManager.logAbort`, only for transactions that spilled pages (the only case in which a transaction has WAL records before its commit) |
| 8 | `Checkpoint(0, generation, lsn)` | `varlong generation`, `varlong lsn` | `Checkpointer`, transaction id is always 0 |
-| 9 | `PageRef(txnId, pageId, length, checksum)` | `i64 pageId`, `varint length`, `i32 checksum` (CRC-32C of the page image's first `length` bytes) | `PAGE_REFERENCES` mode, one per materialized page |
+| 9 | `PageRef(txnId, pageId, address, length, checksum)` | `i64 pageId`, `i64 address` (a packed `PageAddress`), `varint length`, `i32 checksum` (CRC-32C of the page image's first `length` bytes) | `PAGE_REFERENCES` mode, one per materialized page |
Frame sizes that follow directly from the layouts (25-byte header included):
| Record | Bytes |
|---|---|
-| `PageRef` with `128 <= length < 16384` | 25 + 8 + 2 + 4 = **39** |
-| `PageRef` with `length = 16384` | 25 + 8 + 3 + 4 = **40** |
-| `Page` with image length `L` (128 <= L < 16384) | 25 + 8 + 2 + L = **35 + L** |
+| `PageRef` with `128 <= length < 16384` | 25 + 8 + 8 + 2 + 4 = **47** |
+| `PageRef` with `length = 16384` | 25 + 8 + 8 + 3 + 4 = **48** |
+| `Page` with image length `L` (128 <= L < 16384) | 25 + 8 + 8 + 2 + L = **43 + L** |
| `Commit` | 25 + 4 varlongs; with a 6-byte millisecond wall clock typically 34 to 42 |
| `Root` | 25 + 2 + 8 + summary; empty tree 35, otherwise typically 60 to 130 |
| `Checkpoint` | 25 + 2 varlongs |
### Example frame
-A `PageRef` for the (arbitrary) page id `0x0000000300000101`, image length 3,912, checksum `0x1A2B3C4D`, transaction 77, at LSN 4096. The payload is 8 + 2 + 4 = 14 bytes, so the frame is 39 bytes and the next frame starts at LSN 4135:
+A `PageRef` for the (arbitrary) page id `0x0000000300000101`, stored at unit 1234 of segment 1, image length 3,912 (62 units), checksum `0x1A2B3C4D`, transaction 77, at LSN 4096. The payload is 8 + 8 + 2 + 4 = 22 bytes, so the frame is 47 bytes and the next frame starts at LSN 4143:
```text
-0e 00 00 00 payload length 14
-xx xx xx xx CRC-32C of frame bytes [8, 39)
+16 00 00 00 payload length 22
+xx xx xx xx CRC-32C of frame bytes [8, 47)
00 10 00 00 00 00 00 00 LSN 4096
09 type 9 (PageRef)
4d 00 00 00 00 00 00 00 txn 77
01 01 00 00 03 00 00 00 pageId (i64 LE)
+3e 00 d2 04 00 01 00 00 address (i64 LE): segment 1, offset 1234, 62 units
c8 1e varint 3912 (0x0F48): low 7 bits 0x48 | 0x80, then 0x1E
4d 3c 2b 1a checksum (i32 LE)
```
@@ -160,7 +162,7 @@ PAGE < WAL_APPEND < DATA_WRITE < COMMIT_APPEND < DATA_SYNC < WAL_SYNC < CATALOG_
The `Commit` record is appended in a separate call **after** the data pages were written. It is the atomic commit point: a transaction is part of the recovered state if and only if its `Commit` frame is valid in the recovered log and the durability-prefix rule (section 8) admits it. Consequently a crash at any point `>= COMMIT_APPEND` that preserves the operating system's page cache recovers the commit, and a crash at any point `< COMMIT_APPEND` recovers the prior state.
-Spilled subtrees (`TransactionManager.spill`, only for bulk loads) are handled earlier, outside the commit lock: the page images are written to their segments **first**, then the `Page`/`PageRef` records are appended, carrying the transaction id but no `Begin`. Recovery attaches them to the transaction's body by id. The data-before-log order closes a checkpoint race in `PAGE_IMAGES` mode: a checkpoint can only observe an LSN past the spill records after the pages are already in the page cache, so its `pages.sync()` makes them durable before `truncateBefore` drops the records. If the transaction is aborted, `WalRecord.Abort` follows the spill records.
+Spilled subtrees (`TransactionManager.spill`, only for bulk loads) are handled earlier, outside the commit lock: the page images are written to their segments **first**, then the `Page`/`PageRef` records are appended, carrying the transaction id but no `Begin`. Recovery attaches them to the transaction's body by id. The data-before-log order matters for checkpoints, which can run while a spill is in progress. A checkpoint reads its LSN *before* it syncs pages and the page directory (section 9), so every spill record below that LSN was appended after its page was written and its directory entry set, and both are made durable before `truncateBefore` drops the record. If the transaction is aborted, `WalRecord.Abort` follows the spill records.
## 6. WAL modes
@@ -172,9 +174,9 @@ Every materialized page is logged in full (`Page`). At commit (`SYNC`) only the
### PAGE_REFERENCES
-Every materialized page is logged as a 39 or 40 byte `PageRef` holding its id, length and CRC-32C. The image itself lives only in the data segment. This is safe because of two invariants:
+Every materialized page is logged as a 47 or 48 byte `PageRef` holding its id, address, length and CRC-32C. The image itself lives only in the data segment. This is safe because of two invariants:
-1. **Pages are never overwritten in place while reachable.** The tree is copy-on-write: a commit writes new versions of changed nodes to *freshly allocated* `PageId`s and never modifies a page that any generation, pinned snapshot or retained catalog image can reach. Compaction copies live pages to new ids and only reclaims a segment after a checkpoint has moved the WAL past every record that could reference it ([maintenance](../storage/maintenance.md)). A page id additionally embeds the allocation epoch, and `PageHeader.verify` rejects an image whose stored id differs from the requested one.
+1. **Pages are never overwritten in place while reachable.** The tree is copy-on-write: a commit writes new versions of changed nodes to *freshly allocated* `PageId`s and never modifies a page that any generation, pinned snapshot or retained catalog image can reach. Compaction copies live pages to new ids and only reclaims a segment after a checkpoint has moved the WAL past every record that could reference it ([maintenance](../storage/maintenance.md)). A page number is never handed out twice, and `PageHeader.verify` rejects an image whose stored number differs from the requested one.
2. **Data is forced before the log.** `makeDurable` calls `pages.sync()` (every dirty segment file, `force(false)`) and only then `wal.sync()`. When the commit's `Commit` frame is durable, so is every page it references.
Invariant 2 alone is not enough: the OS may write the WAL segment back to disk on its own *before* `pages.sync()` runs, and a power failure at that moment leaves a durable `Commit` frame whose pages are missing. That commit was never acknowledged (acknowledgement happens after `WAL_SYNC` and publication). Recovery detects the situation by verifying every `PageRef` (section 8) and discards the commit together with everything after it.
@@ -185,22 +187,22 @@ Let a commit materialize `N` pages with mean image length `L` (images are `PageH
| | `PAGE_IMAGES` | `PAGE_REFERENCES` |
|---|---|---|
-| WAL bytes | `N (35 + L)` | `39 N` |
+| WAL bytes | `N (43 + L)` | `47 N` |
| Data segment bytes | `N L` | `N L` |
-| Total bytes written per commit | `N (2L + 35)` | `N (L + 39)` |
+| Total bytes written per commit | `N (2L + 43)` | `N (L + 47)` |
| Forced at commit (`SYNC`) | WAL only | dirty data segments, then WAL |
-| Forced later | data segments at checkpoint | nothing |
+| Forced later | data segments and the page directory at checkpoint | the page directory at checkpoint |
| Recovery source of page bytes | WAL | data segments (verified) |
Worked numbers for the default page size `P = 16 KiB` and a commit that rewrites a root-to-leaf path in three trees of height 4 (`N = 12`):
| `L` | `PAGE_IMAGES` total | `PAGE_REFERENCES` total | Ratio |
|---|---|---|---|
-| 16,384 (full pages) | 12 x (32,768 + 35) = 393,636 B | 12 x (16,384 + 40) = 197,088 B | 2.00x |
-| 8,000 | 12 x 16,035 = 192,420 B | 12 x 8,039 = 96,468 B | 1.99x |
-| 1,024 | 12 x 2,083 = 24,996 B | 12 x 1,063 = 12,756 B | 1.96x |
+| 16,384 (full pages) | 12 x (32,768 + 44) = 393,744 B | 12 x (16,384 + 48) = 197,184 B | 2.00x |
+| 8,000 | 12 x 16,043 = 192,516 B | 12 x 8,047 = 96,564 B | 1.99x |
+| 1,024 | 12 x 2,091 = 25,092 B | 12 x 1,071 = 12,852 B | 1.95x |
-`PAGE_REFERENCES` halves device write volume and shrinks the log by roughly `L / 39`, which also shortens recovery scans and makes checkpoints rarer for the same `checkpointWalBytes`. The price is a second `fsync` target per group-commit batch (the dirty segment files), amortised across the batch like the WAL sync.
+`PAGE_REFERENCES` halves device write volume and shrinks the log by roughly `L / 47`, which also shortens recovery scans and makes checkpoints rarer for the same `checkpointWalBytes`. The price is a second `fsync` target per group-commit batch (the dirty segment files), amortised across the batch like the WAL sync.
## 7. Durability levels
@@ -242,9 +244,9 @@ sequenceDiagram
`Recovery.recover(checkpoint, wal, pages, feed, codec)`:
1. **Scan.** Read the WAL from `checkpoint.checkpointLsn()` (may throw `CORRUPT_LOG`, section 3). `Commit` frames are collected in log order; `Abort` drops the transaction's buffered spill records; `Checkpoint` is ignored; every other record is buffered in `pending[txnId]`.
-2. **Filter.** For each commit in log order, remove its body from `pending`. Skip it if `commit.generation <= checkpoint.current().id()` (already in the image).
+2. **Filter.** For each commit in log order, remove its body from `pending`. Skip it if `commit.generation <= checkpoint.current().id()` (already in the image). Otherwise write the address of every `Page` and `PageRef` in the body back into the page directory (`PageStore.place`), which also moves the directory's next index past that number. Entries written after the last checkpoint may not have reached the disk; this restores them.
3. **Durability prefix.** If any earlier commit was discarded, discard this one too. Otherwise check `intact(body, pages)`: for every `PageRef` in the body, read the page, run `PageHeader.verify(page, pageId)` (magic, format, identity, header checksum), and require `page.byteSize() >= length` and `crc32c(page[0, length)) == checksum`. Any exception or mismatch means the page never reached storage; log a warning once (`durable prefix ends before generation g`) and discard this and every later commit. `Page` records need no verification because they carry their own bytes, protected by the frame CRC.
-4. **Redo.** Apply the body in order: `Page` writes the image to its page id; `BranchMeta` inserts or updates the branch (keeping its current roots); `Root` sets one slot root of one branch; `Feed` appends the decoded event to the change feed (a no-op if the feed already holds that generation). Build `Generation(commit.generation, commit.wallTime, commit.txnId, max(nextAtom), max(nextTxn), branches)` and append it to the history.
+4. **Redo.** Apply the body in order: `Page` writes the image at the address it carries; `BranchMeta` inserts or updates the branch (keeping its current roots); `Root` sets one slot root of one branch; `Feed` appends the decoded event to the change feed (a no-op if the feed already holds that generation). Build `Generation(commit.generation, commit.wallTime, commit.txnId, max(nextAtom), max(nextTxn), branches)` and append it to the history.
5. **Seal.** `pages.sync()`; `feed.truncateAfter(recovered generation)` drops feed frames for commits that did not survive (possible under `ASYNC`, or after the prefix rule fired); `feed.sync()`.
6. **Result.** `CatalogImage(recovered generation, history, segments, wal.end(), feed.size())` and `Outcome(image, replayedCommits, discardedTransactions = pending.size(), discardedCommits)`. `discardedTransactions` counts transactions that wrote records but neither a `Commit` nor an `Abort` (crashed mid-commit, or spilled and then lost to a crash before committing or aborting).
@@ -262,8 +264,8 @@ LOG: [recovery] recovered generation 158: replayed 0 commits from lsn 156,490,
`Checkpointer.checkpoint()` runs inside `TransactionManager.exclusive(...)`, which takes the commit lock **and** drains the group committer, so `current == head` and nothing is in flight:
-1. `pages.sync()`, `feed.sync()` (forces the active feed segment; earlier feed segments were forced when the feed rolled).
-2. `lsn = wal.end()`; `history` = retained generations other than `current`, newest first, at most `historyLimit`.
+1. `lsn = wal.end()`. It is read first so that everything logged below it was written before the syncs that follow.
+2. `pages.checkpoint()` (every dirty segment, then the page directory), `feed.sync()` (forces the active feed segment; earlier feed segments were forced when the feed rolled). `history` = retained generations other than `current`, newest first, at most `historyLimit`.
3. `catalog.publish(CatalogImage(current, history, segments, lsn, feed.size()))`. The image is framed as `i32 magic 0x54414348`, `i32 crc32c(payload)`, `i64 length`, payload; written to `generations/%016x.cat.tmp`, forced, atomically renamed, the directory forced; then `manifest` is rewritten the same way. The three newest `.cat` files are kept; `loadLatest()` falls back to the newest valid file if the manifest is missing or names a corrupt file.
4. `wal.append(Checkpoint(0, current.id, lsn))`, `wal.sync()`, `wal.truncateBefore(lsn)`.
diff --git a/engine/src/main/java/io/hstore/engine/StorageEngine.java b/engine/src/main/java/io/hstore/engine/StorageEngine.java
index 1918ec3..5ec45a6 100644
--- a/engine/src/main/java/io/hstore/engine/StorageEngine.java
+++ b/engine/src/main/java/io/hstore/engine/StorageEngine.java
@@ -38,7 +38,7 @@
public final class StorageEngine implements AutoCloseable {
private static final System.Logger LOG = System.getLogger("hstore.engine");
- private static final String FORMAT_VERSION = "3";
+ private static final String FORMAT_VERSION = "4";
private static final int MAX_ATTEMPTS = 8;
private static final Duration MAINTENANCE_INTERVAL = Duration.ofMillis(500);
@@ -116,7 +116,7 @@ private static void verifyFormat(Path directory, EngineOptions options) throws I
}
if (!format.getProperty("format", "1").equals(FORMAT_VERSION)) {
throw HStoreException.invalid("database " + directory + " uses storage format " + format.getProperty("format", "1")
- + "; this build reads format " + FORMAT_VERSION + " (sized page references); export and reload it");
+ + "; this build reads format " + FORMAT_VERSION + " (page directory); export and reload it");
}
int stored = Integer.parseInt(format.getProperty("page-size"));
if (stored != options.pageSize()) {
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 befdc25..ab97b75 100644
--- a/engine/src/main/java/io/hstore/engine/maintenance/Checkpointer.java
+++ b/engine/src/main/java/io/hstore/engine/maintenance/Checkpointer.java
@@ -41,10 +41,10 @@ public long walSinceCheckpoint() {
public Result checkpoint() {
return transactions.exclusive(() -> {
- pages.sync();
+ long lsn = wal.end();
+ pages.checkpoint();
feed.sync();
Generation current = transactions.current();
- long lsn = wal.end();
List history = transactions.history().stream()
.filter(generation -> generation.id() != current.id())
.sorted((a, b) -> Long.compare(b.id(), a.id()))
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 89f0a6a..3bb7932 100644
--- a/engine/src/main/java/io/hstore/engine/maintenance/Compactor.java
+++ b/engine/src/main/java/io/hstore/engine/maintenance/Compactor.java
@@ -44,7 +44,7 @@ public Map liveness() {
.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 -> PageId.segmentOf(entry.getKey()), TreeMap::new,
+ return visited.entrySet().stream().collect(Collectors.groupingBy(entry -> pages.segmentOf(entry.getKey()), TreeMap::new,
Collectors.summingLong(entry -> (long) entry.getValue() * PageId.UNIT_BYTES)));
}
@@ -68,7 +68,7 @@ public Report compact(double liveThreshold, int maxVictims) {
}
victims.forEach(id -> pages.transition(id, SegmentState.COMPACTING, 0));
Set victimSet = Set.copyOf(victims);
- LongPredicate moving = pageId -> victimSet.contains(PageId.segmentOf(pageId));
+ 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();
diff --git a/engine/src/main/java/io/hstore/engine/maintenance/Recovery.java b/engine/src/main/java/io/hstore/engine/maintenance/Recovery.java
index e339e0f..298da98 100644
--- a/engine/src/main/java/io/hstore/engine/maintenance/Recovery.java
+++ b/engine/src/main/java/io/hstore/engine/maintenance/Recovery.java
@@ -56,6 +56,7 @@ public static Outcome recover(CatalogImage checkpoint, WriteAheadLog wal, PageSt
continue;
}
List body = records == null ? List.of() : records;
+ body.forEach(record -> place(record, pages));
if (discarded > 0 || !intact(body, pages)) {
if (discarded == 0) {
LOG.log(System.Logger.Level.WARNING, "durable prefix ends before generation {0}: transaction {1} references pages that never reached storage",
@@ -77,9 +78,18 @@ public static Outcome recover(CatalogImage checkpoint, WriteAheadLog wal, PageSt
return new Outcome(recovered, replayed, pending.size(), discarded);
}
+ private static void place(WalRecord record, PageStore pages) {
+ switch (record) {
+ case WalRecord.Page page -> pages.place(page.pageId(), page.address());
+ case WalRecord.PageRef ref -> pages.place(ref.pageId(), ref.address());
+ default -> {
+ }
+ }
+ }
+
private static boolean intact(List records, PageStore pages) {
for (WalRecord record : records) {
- if (record instanceof WalRecord.PageRef(long _, long pageId, int length, int checksum)) {
+ if (record instanceof WalRecord.PageRef(long _, long pageId, long _, int length, int checksum)) {
try {
MemorySegment page = pages.read(pageId);
PageHeader.verify(page, pageId);
@@ -99,7 +109,7 @@ private static Generation redo(Generation base, WalRecord.Commit commit, List branches = new TreeMap<>(base.branches());
for (WalRecord record : records) {
switch (record) {
- case WalRecord.Page(long _, long pageId, byte[] image) -> pages.write(pageId, MemorySegment.ofArray(image));
+ case WalRecord.Page(long _, long pageId, long _, byte[] image) -> pages.write(pageId, MemorySegment.ofArray(image));
case WalRecord.BranchMeta(long _, int id, String name, int parent, long baseGeneration, long created, Branch.State state) -> {
RootVector roots = branches.containsKey(id) ? branches.get(id).roots() : RootVector.EMPTY;
branches.put(id, new Branch(id, name, parent, baseGeneration, created, state, roots));
diff --git a/engine/src/main/java/io/hstore/engine/page/PageAddress.java b/engine/src/main/java/io/hstore/engine/page/PageAddress.java
new file mode 100644
index 0000000..2d79b86
--- /dev/null
+++ b/engine/src/main/java/io/hstore/engine/page/PageAddress.java
@@ -0,0 +1,52 @@
+package io.hstore.engine.page;
+
+public record PageAddress(int segment, int offset, int units) {
+
+ static final int SEGMENT_BITS = 24;
+ static final int OFFSET_BITS = 24;
+ static final int UNITS_BITS = 16;
+
+ public static final long NONE = 0L;
+ public static final int MAX_SEGMENT = (1 << SEGMENT_BITS) - 1;
+ public static final int MAX_OFFSET = (1 << OFFSET_BITS) - 1;
+ public static final int MAX_UNITS = (1 << UNITS_BITS) - 1;
+
+ public PageAddress {
+ if (segment < 1 || segment > MAX_SEGMENT || offset < 0 || offset > MAX_OFFSET || units < 1 || units > MAX_UNITS) {
+ throw new IllegalArgumentException("page address out of range: " + segment + ":" + offset + "+" + units);
+ }
+ }
+
+ public static PageAddress unpack(long packed) {
+ return new PageAddress(segmentOf(packed), offsetOf(packed), unitsOf(packed));
+ }
+
+ public static int segmentOf(long packed) {
+ return (int) (packed >>> (OFFSET_BITS + UNITS_BITS));
+ }
+
+ public static int offsetOf(long packed) {
+ return (int) (packed >>> UNITS_BITS) & MAX_OFFSET;
+ }
+
+ public static int unitsOf(long packed) {
+ return (int) packed & MAX_UNITS;
+ }
+
+ public long position() {
+ return (long) offset * PageId.UNIT_BYTES;
+ }
+
+ public long bytes() {
+ return (long) units * PageId.UNIT_BYTES;
+ }
+
+ public long pack() {
+ return ((long) segment << (OFFSET_BITS + UNITS_BITS)) | ((long) offset << UNITS_BITS) | units;
+ }
+
+ @Override
+ public String toString() {
+ return segment + ":" + offset + "+" + units;
+ }
+}
diff --git a/engine/src/main/java/io/hstore/engine/page/PageDirectory.java b/engine/src/main/java/io/hstore/engine/page/PageDirectory.java
new file mode 100644
index 0000000..0518318
--- /dev/null
+++ b/engine/src/main/java/io/hstore/engine/page/PageDirectory.java
@@ -0,0 +1,128 @@
+package io.hstore.engine.page;
+
+import io.hstore.engine.HStoreException;
+
+import java.io.IOException;
+import java.io.UncheckedIOException;
+import java.lang.foreign.Arena;
+import java.lang.foreign.MemorySegment;
+import java.lang.foreign.ValueLayout;
+import java.nio.ByteOrder;
+import java.nio.channels.FileChannel;
+import java.nio.file.Path;
+import java.nio.file.StandardOpenOption;
+import java.util.Arrays;
+
+final class PageDirectory implements AutoCloseable {
+
+ static final String FILE = "pages.dir";
+
+ private static final int MAGIC = 0x52494450;
+ private static final int FORMAT = 1;
+ private static final long HEADER = 64;
+ private static final long NEXT_OFFSET = 8;
+ private static final int CHUNK_SHIFT = 23;
+ 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 final Path path;
+ private final FileChannel channel;
+ private volatile MemorySegment[] chunks = new MemorySegment[0];
+ private long next;
+
+ private PageDirectory(Path path, FileChannel channel) {
+ this.path = path;
+ this.channel = channel;
+ }
+
+ static PageDirectory open(Path path) {
+ try {
+ FileChannel channel = FileChannel.open(path, StandardOpenOption.CREATE, StandardOpenOption.READ, StandardOpenOption.WRITE);
+ PageDirectory directory = new PageDirectory(path, channel);
+ long size = channel.size();
+ directory.mapThrough(size == 0 ? 0 : (int) ((size - 1) >>> CHUNK_SHIFT));
+ MemorySegment header = directory.chunks[0];
+ if (header.get(INT, 0) == 0 && header.get(INT, 4) == 0 && header.get(LONG, NEXT_OFFSET) == 0) {
+ header.set(INT, 0, MAGIC);
+ header.set(INT, 4, FORMAT);
+ header.set(LONG, NEXT_OFFSET, 1L);
+ header.force();
+ } else if (header.get(INT, 0) != MAGIC || header.get(INT, 4) != FORMAT) {
+ channel.close();
+ throw HStoreException.invalid("page directory " + path + " has an unknown format");
+ }
+ directory.next = header.get(LONG, NEXT_OFFSET);
+ return directory;
+ } catch (IOException e) {
+ throw new UncheckedIOException("cannot open page directory " + path, e);
+ }
+ }
+
+ 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;
+ }
+
+ void put(long index, long address) {
+ long position = HEADER + index * Long.BYTES;
+ int chunk = (int) (position >>> CHUNK_SHIFT);
+ if (chunk >= chunks.length) {
+ mapThrough(chunk);
+ }
+ chunks[chunk].set(LONG, position & (CHUNK_BYTES - 1), address);
+ }
+
+ long allocate() {
+ long index = next;
+ if (index > PageId.MAX_INDEX) {
+ throw HStoreException.limit("page numbers exhausted");
+ }
+ reserve(index);
+ return index;
+ }
+
+ void reserve(long index) {
+ if (index >= next) {
+ next = index + 1;
+ chunks[0].set(LONG, NEXT_OFFSET, next);
+ }
+ }
+
+ long next() {
+ return next;
+ }
+
+ void force() {
+ for (MemorySegment chunk : chunks) {
+ chunk.force();
+ }
+ }
+
+ private void mapThrough(int last) {
+ MemorySegment[] mapped = chunks;
+ if (last < mapped.length) {
+ return;
+ }
+ MemorySegment[] grown = Arrays.copyOf(mapped, last + 1);
+ try {
+ for (int chunk = mapped.length; chunk <= last; chunk++) {
+ grown[chunk] = channel.map(FileChannel.MapMode.READ_WRITE, chunk * CHUNK_BYTES, CHUNK_BYTES, Arena.ofAuto());
+ }
+ } catch (IOException e) {
+ throw new UncheckedIOException("cannot grow page directory " + path, e);
+ }
+ chunks = grown;
+ }
+
+ @Override
+ public void close() {
+ try {
+ channel.close();
+ } catch (IOException e) {
+ throw new UncheckedIOException("cannot close page directory " + path, e);
+ }
+ }
+}
diff --git a/engine/src/main/java/io/hstore/engine/page/PageId.java b/engine/src/main/java/io/hstore/engine/page/PageId.java
index e27c92a..383a2bc 100644
--- a/engine/src/main/java/io/hstore/engine/page/PageId.java
+++ b/engine/src/main/java/io/hstore/engine/page/PageId.java
@@ -1,53 +1,46 @@
package io.hstore.engine.page;
-public record PageId(int segment, int offset, int generation) {
+public record PageId(long index, int reuse) {
- static final int SEGMENT_BITS = 24;
- static final int OFFSET_BITS = 24;
- static final int GENERATION_BITS = 16;
+ static final int REUSE_BITS = 16;
public static final long NONE = 0L;
public static final int UNIT_BYTES = 64;
- public static final int MAX_SEGMENT = (1 << SEGMENT_BITS) - 1;
- public static final int MAX_OFFSET = (1 << OFFSET_BITS) - 1;
+ public static final long MAX_INDEX = (1L << (Long.SIZE - REUSE_BITS)) - 1;
public PageId {
- if (segment < 1 || segment > MAX_SEGMENT || offset < 0 || offset > MAX_OFFSET) {
- throw new IllegalArgumentException("page address out of range: " + segment + ":" + offset);
+ if (index < 1 || index > MAX_INDEX) {
+ throw new IllegalArgumentException("page number out of range: " + index);
}
- generation &= (1 << GENERATION_BITS) - 1;
+ reuse &= (1 << REUSE_BITS) - 1;
}
public static int unitsFor(long bytes) {
return Math.toIntExact((bytes + UNIT_BYTES - 1) / UNIT_BYTES);
}
- public static long pack(int segment, int offset, long epoch) {
- return new PageId(segment, offset, (int) epoch).pack();
+ public static long pack(long index, int reuse) {
+ return new PageId(index, reuse).pack();
}
public static PageId unpack(long packed) {
- return new PageId(segmentOf(packed), offsetOf(packed), generationOf(packed));
+ return new PageId(indexOf(packed), reuseOf(packed));
}
- public static int segmentOf(long packed) {
- return (int) (packed >>> (OFFSET_BITS + GENERATION_BITS));
+ public static long indexOf(long packed) {
+ return packed >>> REUSE_BITS;
}
- public static int offsetOf(long packed) {
- return (int) (packed >>> GENERATION_BITS) & MAX_OFFSET;
- }
-
- public static int generationOf(long packed) {
- return (int) packed & ((1 << GENERATION_BITS) - 1);
+ public static int reuseOf(long packed) {
+ return (int) packed & ((1 << REUSE_BITS) - 1);
}
public long pack() {
- return ((long) segment << (OFFSET_BITS + GENERATION_BITS)) | ((long) offset << GENERATION_BITS) | generation;
+ return (index << REUSE_BITS) | reuse;
}
@Override
public String toString() {
- return segment + ":" + offset + "@" + generation;
+ return reuse == 0 ? "#" + index : "#" + index + "." + reuse;
}
}
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 b79a5b2..4de53de 100644
--- a/engine/src/main/java/io/hstore/engine/page/PageStore.java
+++ b/engine/src/main/java/io/hstore/engine/page/PageStore.java
@@ -28,13 +28,14 @@ public final class PageStore implements AutoCloseable {
private final AtomicLong bytesWritten = new AtomicLong();
private final AtomicLong pagesRead = new AtomicLong();
private final AtomicLong bytesRead = new AtomicLong();
+ private PageDirectory addresses;
private int activeSegment;
private int nextUnit;
private PageStore(Path directory, int pageSize, int pagesPerSegment) {
this.directory = directory;
this.pageSize = pageSize;
- this.segmentUnits = (int) Math.min(PageId.MAX_OFFSET + 1L, (long) pagesPerSegment * pageSize / PageId.UNIT_BYTES);
+ this.segmentUnits = (int) Math.min(PageAddress.MAX_OFFSET + 1L, (long) pagesPerSegment * pageSize / PageId.UNIT_BYTES);
}
public static PageStore open(Path directory, int pageSize, int pagesPerSegment, Collection known) {
@@ -46,6 +47,7 @@ public static PageStore open(Path directory, int pageSize, int pagesPerSegment,
private void restore(Collection known) {
try {
Files.createDirectories(directory);
+ addresses = PageDirectory.open(directory.resolve(PageDirectory.FILE));
Map byId = new ConcurrentHashMap<>();
known.forEach(info -> byId.put(info.id(), info));
try (Stream listing = Files.list(directory)) {
@@ -79,7 +81,7 @@ public int pageSize() {
return pageSize;
}
- public long allocate(long epoch, int length) {
+ public long allocate(int length) {
if (length > pageSize) {
throw new IllegalArgumentException("page image exceeds page size: " + length);
}
@@ -92,7 +94,9 @@ public long allocate(long epoch, int length) {
int offset = nextUnit;
nextUnit += units;
segments.computeIfPresent(activeSegment, (_, info) -> info.withAllocation(info.pages() + 1, Math.max(info.units(), offset + (long) units)));
- return PageId.pack(activeSegment, offset, epoch);
+ long index = addresses.allocate();
+ addresses.put(index, new PageAddress(activeSegment, offset, units).pack());
+ return PageId.pack(index, 0);
} finally {
allocation.unlock();
}
@@ -117,8 +121,12 @@ public void write(long pageId, MemorySegment image) {
if (image.byteSize() > pageSize) {
throw new IllegalArgumentException("page image exceeds page size: " + image.byteSize());
}
- int segment = PageId.segmentOf(pageId);
- int offset = PageId.offsetOf(pageId);
+ long address = locate(pageId);
+ if (PageId.unitsFor(image.byteSize()) > PageAddress.unitsOf(address)) {
+ throw new IllegalArgumentException("page image exceeds its allocation of " + PageAddress.unitsOf(address) + " units");
+ }
+ int segment = PageAddress.segmentOf(address);
+ int offset = PageAddress.offsetOf(address);
SegmentFile file = files.computeIfAbsent(segment, id -> {
segments.putIfAbsent(id, new SegmentInfo(id, SegmentState.SEALED, 0, 0, 0));
return SegmentFile.open(id, SegmentFile.pathFor(directory, id));
@@ -136,20 +144,56 @@ long mappedBytes(int segment) {
}
public MemorySegment read(long pageId) {
- SegmentFile file = files.get(PageId.segmentOf(pageId));
+ long address = locate(pageId);
+ SegmentFile file = files.get(PageAddress.segmentOf(address));
if (file == null) {
throw HStoreException.corrupt(pageId, "segment does not exist");
}
pagesRead.incrementAndGet();
- MemorySegment page = file.read((long) PageId.offsetOf(pageId) * PageId.UNIT_BYTES, pageSize);
+ MemorySegment page = file.read((long) PageAddress.offsetOf(address) * PageId.UNIT_BYTES, PageAddress.unitsOf(address) * PageId.UNIT_BYTES);
bytesRead.addAndGet(page.byteSize());
return page;
}
+ public long addressOf(long pageId) {
+ return locate(pageId);
+ }
+
+ public int segmentOf(long pageId) {
+ return PageAddress.segmentOf(locate(pageId));
+ }
+
+ public void place(long pageId, long address) {
+ allocation.lock();
+ try {
+ long index = PageId.indexOf(pageId);
+ addresses.put(index, address);
+ addresses.reserve(index);
+ if (PageAddress.segmentOf(address) == activeSegment) {
+ nextUnit = Math.max(nextUnit, PageAddress.offsetOf(address) + PageAddress.unitsOf(address));
+ }
+ } finally {
+ allocation.unlock();
+ }
+ }
+
+ private long locate(long pageId) {
+ long address = addresses.get(PageId.indexOf(pageId));
+ if (address == PageAddress.NONE) {
+ throw HStoreException.corrupt(pageId, "page has no address");
+ }
+ return address;
+ }
+
public void sync() {
files.values().forEach(SegmentFile::force);
}
+ public void checkpoint() {
+ sync();
+ addresses.force();
+ }
+
public List segments() {
return List.copyOf(segments.values());
}
@@ -195,5 +239,6 @@ public long bytesRead() {
public void close() {
files.values().forEach(SegmentFile::close);
files.clear();
+ addresses.close();
}
}
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 443f9bc..16c39a5 100644
--- a/engine/src/main/java/io/hstore/engine/txn/TransactionManager.java
+++ b/engine/src/main/java/io/hstore/engine/txn/TransactionManager.java
@@ -361,14 +361,14 @@ private Materializer materializer(long epoch, long txnId, List record
return new Materializer(new Materializer.Sink() {
@Override
public long allocate(int length) {
- return storage.pages().allocate(epoch, length);
+ return storage.pages().allocate(length);
}
@Override
public void accept(long pageId, MemorySegment image, Node frozen) {
records.add(storage.walMode() == WalMode.PAGE_IMAGES
- ? new WalRecord.Page(txnId, pageId, image.toArray(ValueLayout.JAVA_BYTE))
- : new WalRecord.PageRef(txnId, pageId, Math.toIntExact(image.byteSize()), Checksums.crc32c(image)));
+ ? new WalRecord.Page(txnId, pageId, storage.pages().addressOf(pageId), image.toArray(ValueLayout.JAVA_BYTE))
+ : new WalRecord.PageRef(txnId, pageId, storage.pages().addressOf(pageId), Math.toIntExact(image.byteSize()), Checksums.crc32c(image)));
writes.add(new PageWrite(pageId, image));
storage.source().admit(pageId, frozen);
}
diff --git a/engine/src/main/java/io/hstore/engine/wal/WalRecord.java b/engine/src/main/java/io/hstore/engine/wal/WalRecord.java
index 2e00ec3..4d9fef8 100644
--- a/engine/src/main/java/io/hstore/engine/wal/WalRecord.java
+++ b/engine/src/main/java/io/hstore/engine/wal/WalRecord.java
@@ -11,10 +11,10 @@ public sealed interface WalRecord {
record Begin(long txnId, int branch, long baseGeneration) implements WalRecord {
}
- record Page(long txnId, long pageId, byte[] image) implements WalRecord {
+ record Page(long txnId, long pageId, long address, byte[] image) implements WalRecord {
}
- record PageRef(long txnId, long pageId, int length, int checksum) implements WalRecord {
+ record PageRef(long txnId, long pageId, long address, int length, int checksum) implements WalRecord {
}
record Root(long txnId, int branch, int slot, Ref root) implements WalRecord {
@@ -53,7 +53,7 @@ static int typeOf(WalRecord record) {
static void writePayload(ByteCursor out, WalRecord record) {
switch (record) {
case Begin(long _, int branch, long base) -> out.putVarInt(branch).putVarLong(base);
- case Page(long _, long pageId, byte[] image) -> out.putLong(pageId).putBlob(image);
+ case Page(long _, long pageId, long address, byte[] image) -> out.putLong(pageId).putLong(address).putBlob(image);
case Root(long _, int branch, int slot, Ref root) -> {
out.putVarInt(branch).putVarInt(slot);
Ref.write(out, root);
@@ -66,14 +66,14 @@ case Commit(long _, long generation, long wallTime, long nextAtom, long nextTxn)
case Abort _ -> {
}
case Checkpoint(long _, long generation, long lsn) -> out.putVarLong(generation).putVarLong(lsn);
- case PageRef(long _, long pageId, int length, int checksum) -> out.putLong(pageId).putVarInt(length).putInt(checksum);
+ case PageRef(long _, long pageId, long address, int length, int checksum) -> out.putLong(pageId).putLong(address).putVarInt(length).putInt(checksum);
}
}
static WalRecord readPayload(int type, long txnId, ByteCursor in) {
return switch (type) {
case 1 -> new Begin(txnId, in.getVarInt(), in.getVarLong());
- case 2 -> new Page(txnId, in.getLong(), in.getBlob());
+ case 2 -> new Page(txnId, in.getLong(), in.getLong(), in.getBlob());
case 3 -> new Root(txnId, in.getVarInt(), in.getVarInt(), Ref.read(in));
case 4 -> new BranchMeta(txnId, in.getVarInt(), in.getString(), in.getVarInt(), in.getVarLong(), in.getVarLong(),
Branch.State.values()[in.getUnsignedByte()]);
@@ -81,7 +81,7 @@ static WalRecord readPayload(int type, long txnId, ByteCursor in) {
case 6 -> new Commit(txnId, in.getVarLong(), in.getVarLong(), in.getVarLong(), in.getVarLong());
case 7 -> new Abort(txnId);
case 8 -> new Checkpoint(txnId, in.getVarLong(), in.getVarLong());
- case 9 -> new PageRef(txnId, in.getLong(), in.getVarInt(), in.getInt());
+ case 9 -> new PageRef(txnId, in.getLong(), in.getLong(), in.getVarInt(), in.getInt());
default -> throw new IllegalStateException("unknown WAL record type " + type);
};
}
diff --git a/engine/src/test/java/io/hstore/engine/CrashRecoveryTest.java b/engine/src/test/java/io/hstore/engine/CrashRecoveryTest.java
index 2a93987..a34fe4e 100644
--- a/engine/src/test/java/io/hstore/engine/CrashRecoveryTest.java
+++ b/engine/src/test/java/io/hstore/engine/CrashRecoveryTest.java
@@ -10,6 +10,7 @@
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.EnumSource;
import org.junit.jupiter.params.provider.MethodSource;
import java.nio.channels.FileChannel;
@@ -21,6 +22,7 @@
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.stream.IntStream;
import java.util.stream.Stream;
import static org.junit.jupiter.api.Assertions.assertEquals;
@@ -110,6 +112,37 @@ void recoveryStopsAtTheFirstCommitWhosePagesAreMissing() throws Exception {
}
}
+ @ParameterizedTest
+ @EnumSource(WalMode.class)
+ void directoryEntriesLostInACrashAreRebuiltFromTheLog(WalMode mode) throws Exception {
+ EngineOptions options = EngineTest.small().withWalMode(mode).withCheckpointWalBytes(Long.MAX_VALUE);
+ Path addresses = directory.resolve("data").resolve("segments").resolve("pages.dir");
+ try (StorageEngine engine = StorageEngine.open(directory, options)) {
+ engine.write(txn -> txn.createNode(EngineTest.NODE, "checkpointed"));
+ }
+ byte[] checkpointed = Files.readAllBytes(addresses);
+ StorageEngine engine = StorageEngine.open(directory, options);
+ for (int i = 0; i < 50; i++) {
+ String name = "logged-" + i;
+ engine.write(txn -> txn.createNode(EngineTest.NODE, name));
+ }
+ engine.halt();
+ Files.write(addresses, checkpointed);
+ try (StorageEngine recovered = StorageEngine.open(directory, options)) {
+ assertEquals(51L, (long) recovered.read(snapshot -> snapshot.countOfType(0, EngineTest.NODE)));
+ recovered.write(txn -> txn.createNode(EngineTest.NODE, "after-crash"));
+ }
+ try (StorageEngine reopened = StorageEngine.open(directory, options)) {
+ reopened.read(snapshot -> {
+ assertEquals(52L, snapshot.countOfType(0, EngineTest.NODE));
+ for (String name : Stream.concat(Stream.of("checkpointed", "after-crash"), IntStream.range(0, 50).mapToObj(i -> "logged-" + i)).toList()) {
+ assertTrue(snapshot.resolve(0, EngineTest.NODE, name).isPresent(), name);
+ }
+ return null;
+ });
+ }
+ }
+
@Test
void concurrentCommitsShareDurabilityBarriers() throws Exception {
try (StorageEngine engine = StorageEngine.open(directory, EngineTest.small())) {
@@ -126,7 +159,7 @@ void concurrentCommitsShareDurabilityBarriers() throws Exception {
private Path lastSegment() throws Exception {
try (Stream files = Files.list(directory.resolve("data").resolve("segments"))) {
- return files.max(Comparator.naturalOrder()).orElseThrow();
+ return files.filter(file -> file.getFileName().toString().endsWith(".seg")).max(Comparator.naturalOrder()).orElseThrow();
}
}
}
diff --git a/engine/src/test/java/io/hstore/engine/page/PageDirectoryTest.java b/engine/src/test/java/io/hstore/engine/page/PageDirectoryTest.java
new file mode 100644
index 0000000..8344c54
--- /dev/null
+++ b/engine/src/test/java/io/hstore/engine/page/PageDirectoryTest.java
@@ -0,0 +1,75 @@
+package io.hstore.engine.page;
+
+import io.hstore.engine.HStoreException;
+
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
+
+import java.nio.file.Files;
+import java.nio.file.Path;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+
+class PageDirectoryTest {
+
+ @TempDir
+ Path directory;
+
+ @Test
+ void addressesAndTheNextNumberSurviveReopenAcrossChunks() {
+ Path file = directory.resolve(PageDirectory.FILE);
+ int count = 1_500_000;
+ try (PageDirectory pages = PageDirectory.open(file)) {
+ for (long expected = 1; expected <= count; expected++) {
+ long index = pages.allocate();
+ assertEquals(expected, index);
+ pages.put(index, address(index));
+ }
+ pages.force();
+ }
+ try (PageDirectory reopened = PageDirectory.open(file)) {
+ assertEquals(count + 1, reopened.next());
+ for (long index = 1; index <= count; index++) {
+ assertEquals(address(index), reopened.get(index), "entry " + index);
+ }
+ assertEquals(PageAddress.NONE, reopened.get(count + 1));
+ assertEquals(PageAddress.NONE, reopened.get(50_000_000));
+ }
+ }
+
+ @Test
+ void aReservedNumberIsNeverHandedOut() {
+ try (PageDirectory pages = PageDirectory.open(directory.resolve(PageDirectory.FILE))) {
+ pages.reserve(41);
+ assertEquals(42, pages.allocate());
+ pages.reserve(7);
+ assertEquals(43, pages.allocate());
+ }
+ }
+
+ @Test
+ void aFileInAnotherFormatIsRefused() throws Exception {
+ Path file = directory.resolve(PageDirectory.FILE);
+ byte[] header = new byte[64];
+ header[0] = 0x2A;
+ Files.write(file, header);
+ assertThrows(HStoreException.InvalidSchema.class, () -> PageDirectory.open(file));
+ }
+
+ @Test
+ void aFileWhoseHeaderNeverReachedTheDiskStartsOver() throws Exception {
+ Path file = directory.resolve(PageDirectory.FILE);
+ Files.write(file, new byte[1 << 20]);
+ try (PageDirectory pages = PageDirectory.open(file)) {
+ assertEquals(1, pages.allocate());
+ }
+ try (PageDirectory pages = PageDirectory.open(file)) {
+ assertEquals(2, pages.allocate());
+ }
+ }
+
+ private static long address(long index) {
+ return new PageAddress(1 + (int) (index % 1000), (int) (index % PageAddress.MAX_OFFSET), 1 + (int) (index % 200)).pack();
+ }
+}
diff --git a/engine/src/test/java/io/hstore/engine/page/PageStoreTest.java b/engine/src/test/java/io/hstore/engine/page/PageStoreTest.java
index cb5628b..c2eccce 100644
--- a/engine/src/test/java/io/hstore/engine/page/PageStoreTest.java
+++ b/engine/src/test/java/io/hstore/engine/page/PageStoreTest.java
@@ -1,5 +1,7 @@
package io.hstore.engine.page;
+import io.hstore.engine.HStoreException;
+
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.io.TempDir;
@@ -18,6 +20,7 @@
import static org.junit.jupiter.api.Assertions.assertArrayEquals;
import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNotEquals;
import static org.junit.jupiter.api.Assertions.assertThrows;
import static org.junit.jupiter.api.Assertions.assertTrue;
@@ -39,7 +42,7 @@ void pagesReadBackExactlyAsWrittenAcrossSegmentsAndReopens() {
try (PageStore store = PageStore.open(directory, PAGE_SIZE, PAGES_PER_SEGMENT, List.of())) {
for (int i = 0; i < 16_000; i++) {
byte[] image = image(random, PageHeader.SIZE + random.nextInt(PAGE_SIZE - PageHeader.SIZE + 1));
- long pageId = store.allocate(1, image.length);
+ long pageId = store.allocate(image.length);
store.write(pageId, MemorySegment.ofArray(image));
written.put(pageId, image);
ids.add(pageId);
@@ -56,7 +59,7 @@ void pagesReadBackExactlyAsWrittenAcrossSegmentsAndReopens() {
written.forEach((pageId, image) -> assertArrayEquals(image, bytes(reopened.read(pageId))));
long first = ids.getFirst();
reopened.read(first);
- assertTrue(reopened.mappedBytes(PageId.segmentOf(first)) > 0);
+ assertTrue(reopened.mappedBytes(reopened.segmentOf(first)) > 0);
}
}
@@ -81,14 +84,44 @@ void finishedSegmentsAreMappedAndTheGrowingOneIsNotUntilItGrows() {
long growing = write(store, PAGE_SIZE);
store.read(finished);
store.read(growing);
- assertTrue(store.mappedBytes(PageId.segmentOf(finished)) >= (long) PAGES_PER_SEGMENT * PAGE_SIZE - PAGE_SIZE);
- assertEquals(0, store.mappedBytes(PageId.segmentOf(growing)));
+ assertTrue(store.mappedBytes(store.segmentOf(finished)) >= (long) PAGES_PER_SEGMENT * PAGE_SIZE - PAGE_SIZE);
+ assertEquals(0, store.mappedBytes(store.segmentOf(growing)));
for (int i = 0; i < 1100; i++) {
write(store, PAGE_SIZE);
}
long later = write(store, PAGE_SIZE);
store.read(later);
- assertTrue(store.mappedBytes(PageId.segmentOf(later)) > 4L << 20);
+ assertTrue(store.mappedBytes(store.segmentOf(later)) > 4L << 20);
+ }
+ }
+
+ @Test
+ void pagesAreNumberedInOrderWhicheverSegmentHoldsThem() {
+ try (PageStore store = PageStore.open(directory, PAGE_SIZE, PAGES_PER_SEGMENT, List.of())) {
+ List ids = new ArrayList<>();
+ while (store.segments().size() < 3) {
+ ids.add(write(store, PAGE_SIZE / 2));
+ }
+ for (int i = 0; i < ids.size(); i++) {
+ assertEquals(i + 1, PageId.indexOf(ids.get(i)));
+ }
+ assertNotEquals(store.segmentOf(ids.getFirst()), store.segmentOf(ids.getLast()));
+ }
+ }
+
+ @Test
+ void aNumberWithoutAnAddressIsReportedAsCorrupt() {
+ try (PageStore store = PageStore.open(directory, PAGE_SIZE, PAGES_PER_SEGMENT, List.of())) {
+ write(store, 200);
+ assertThrows(HStoreException.CorruptPage.class, () -> store.read(PageId.pack(99, 0)));
+ }
+ }
+
+ @Test
+ void aPageCannotBeWrittenPastWhatWasAllocatedForIt() {
+ try (PageStore store = PageStore.open(directory, PAGE_SIZE, PAGES_PER_SEGMENT, List.of())) {
+ long pageId = store.allocate(100);
+ assertThrows(IllegalArgumentException.class, () -> store.write(pageId, MemorySegment.ofArray(image(new Random(1), 200))));
}
}
@@ -119,7 +152,7 @@ void readersKeepSeeingWholePagesWhileOldMapsAreClosed() throws Exception {
Random random = new Random(11);
for (int i = 0; i < pages; i++) {
images[i] = image(random, PAGE_SIZE - random.nextInt(64));
- ids[i] = store.allocate(1, images[i].length);
+ ids[i] = store.allocate(images[i].length);
store.write(ids[i], MemorySegment.ofArray(images[i]));
written.incrementAndGet();
}
@@ -131,7 +164,7 @@ void readersKeepSeeingWholePagesWhileOldMapsAreClosed() throws Exception {
private static long write(PageStore store, int length) {
byte[] image = image(new Random(length), length);
- long pageId = store.allocate(1, image.length);
+ long pageId = store.allocate(image.length);
store.write(pageId, MemorySegment.ofArray(image));
return pageId;
}
diff --git a/engine/src/test/java/io/hstore/engine/tree/MemoryPages.java b/engine/src/test/java/io/hstore/engine/tree/MemoryPages.java
index 00fe36a..5a546cd 100644
--- a/engine/src/test/java/io/hstore/engine/tree/MemoryPages.java
+++ b/engine/src/test/java/io/hstore/engine/tree/MemoryPages.java
@@ -21,9 +21,7 @@ Tree persist(Tree tree) {
Materializer materializer = new Materializer(new Materializer.Sink() {
@Override
public long allocate(int length) {
- long pageId = PageId.pack(1, next, 1);
- next += PageId.unitsFor(length);
- return pageId;
+ return PageId.pack(++next, 0);
}
@Override
diff --git a/engine/src/test/java/io/hstore/engine/wal/WriteAheadLogTest.java b/engine/src/test/java/io/hstore/engine/wal/WriteAheadLogTest.java
index f7d7589..5f33ae0 100644
--- a/engine/src/test/java/io/hstore/engine/wal/WriteAheadLogTest.java
+++ b/engine/src/test/java/io/hstore/engine/wal/WriteAheadLogTest.java
@@ -33,7 +33,7 @@ private static List records(int count, long seed) {
random.nextBytes(bytes);
records.add(switch (i % 4) {
case 0 -> new WalRecord.Begin(i, random.nextInt(4), random.nextLong(1_000_000));
- case 1 -> new WalRecord.Page(i, random.nextLong(), bytes);
+ case 1 -> new WalRecord.Page(i, random.nextLong(), random.nextLong(), bytes);
case 2 -> new WalRecord.Feed(i, bytes);
default -> new WalRecord.Commit(i, i, random.nextLong(), random.nextLong(1 << 30), random.nextLong(1 << 30));
});
@@ -43,7 +43,7 @@ private static List records(int count, long seed) {
private static String describe(WalRecord record) {
return switch (record) {
- case WalRecord.Page(long txn, long page, byte[] image) -> "page " + txn + " " + page + " " + HexFormat.of().formatHex(image);
+ case WalRecord.Page(long txn, long page, long address, byte[] image) -> "page " + txn + " " + page + " " + address + " " + HexFormat.of().formatHex(image);
case WalRecord.Feed(long txn, byte[] payload) -> "feed " + txn + " " + HexFormat.of().formatHex(payload);
default -> record.toString();
};
@@ -94,7 +94,7 @@ void aTornFinalFrameIsDroppedAndEverythingBeforeItKept() throws IOException {
@Test
void aSmallFrameAfterALargeOneReadsBackIntact() {
- List written = List.of(new WalRecord.Feed(1, new byte[]{1, 2, 3}), new WalRecord.Page(2, 9, new byte[30_000]),
+ List written = List.of(new WalRecord.Feed(1, new byte[]{1, 2, 3}), new WalRecord.Page(2, 9, 10, new byte[30_000]),
new WalRecord.Feed(3, new byte[]{4, 5}), new WalRecord.Commit(4, 4, 5, 6, 7), new WalRecord.Feed(5, new byte[0]));
try (WriteAheadLog wal = WriteAheadLog.open(directory, 1 << 20)) {
written.forEach(wal::append);