diff --git a/docs/architecture/overview.md b/docs/architecture/overview.md index b6e01a31..dd3793f1 100644 --- a/docs/architecture/overview.md +++ b/docs/architecture/overview.md @@ -154,7 +154,7 @@ sequenceDiagram Notes: -* Nodes are decoded once and then shared by every reader through `NodeCache`. It has 16 shards, each an open-addressing table keyed by page number plus a CLOCK (second-chance) ring, with a total byte budget of `node_cache_mb`, each node counted at an estimate of its heap size (`Node.heapBytes`). Lookups are lock-free; only admission takes a per-shard lock. Decoded nodes are immutable (`owner == null`), so sharing them needs no copying. See [pages.md](../storage/pages.md#caching-and-io-accounting). +* Nodes are decoded once and then shared by every reader through `NodeCache`. Catalog and posting-list leaves are kept packed and decode each record the first time it is read. It has 16 shards, each an open-addressing table keyed by page number plus a CLOCK (second-chance) ring, with a total byte budget of `node_cache_mb`, each node counted at an estimate of its heap size (`Node.heapBytes`). Lookups are lock-free; only admission takes a per-shard lock. Decoded nodes are immutable (`owner == null`), so sharing them needs no copying. See [pages.md](../storage/pages.md#caching-and-io-accounting). * `IoTrace` is bound with `ScopedValue.where(...)` around a query. `recordRead` throws `HStoreException.limit` once page reads plus cache hits exceed the query's page budget (`query_page_budget`). * A tree changed since the last checkpoint starts with in-memory nodes (`Ref.Pending`). A read follows those directly and goes through `PagedNodeSource` only once it reaches a stored reference. * Historical reads use `TransactionManager.snapshotAt(generation, branch)`. It looks the generation up in the retained `history` map; see [catalog-and-generations.md](../storage/catalog-and-generations.md#history-retention-and-time-travel). @@ -248,7 +248,7 @@ semantic/state 53 B persisted HNSW index state (datab | Path | Owner | Format | |---|---|---| -| `FORMAT` | `StorageEngine.verifyFormat` | `Properties` with `format=7` (commits logged as key changes and tree pages written at checkpoints, page numbers resolved through a page directory, reference counts per page, small member trees inline in their edge record) 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=8` (leaf records framed by length, commits logged as key changes and tree pages written at checkpoints, page numbers resolved through a page directory, reference counts per page, small member trees inline in their edge record) 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. | diff --git a/docs/operations/docker.md b/docs/operations/docker.md index 29b58883..cfdd6cc2 100644 --- a/docs/operations/docker.md +++ b/docs/operations/docker.md @@ -259,8 +259,8 @@ 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=7`, commits logged as key changes) and the page size. Both are checked at +records the on-disk format version (`format=8`, leaf records framed by length) 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 7 (commits logged as key changes); export and reload it`, +`database … uses storage format N; this build reads format 8 (leaf records framed by length); 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 volume. diff --git a/docs/operations/logging.md b/docs/operations/logging.md index 13195775..cbeaf4ff 100644 --- a/docs/operations/logging.md +++ b/docs/operations/logging.md @@ -115,7 +115,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 7 (commits logged as key changes); export and reload it` | `FORMAT` was written by an incompatible build. | +| `database uses storage format N; this build reads format 8 (leaf records framed by length); 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 7b306d9b..9b2ec348 100644 --- a/docs/storage/maintenance.md +++ b/docs/storage/maintenance.md @@ -268,6 +268,7 @@ History and pins are also what hold dead pages back. A page waits until every ge * magic, format, page identity and CRC-32C (`PageHeader.verify`); * the schema id against the tree being read (`NodeCodec.decode`); * strictly ascending keys in leaves; +* for a packed leaf, that the header's refers-to-pages flag matches its records once all of them are decoded. Loading a packed leaf skips that comparison, because it would decode every record; * separators that route each child; * uniform height (balance); * every stored child summary equal to its recomputed contents; diff --git a/docs/storage/pages.md b/docs/storage/pages.md index b2391575..96322d36 100644 --- a/docs/storage/pages.md +++ b/docs/storage/pages.md @@ -76,6 +76,8 @@ zigzag timeMax i64le roleBits keys: zigzag keys[0], then varint (keys[i] − keys[i−1]) for i = 1..slotCount−1 LEAF: schema.codec().encode(out, keys, values) (value format owned by the schema) + or, for a codec with independentValues(): + (varint length | value) × slotCount (catalog, posting lists, row codecs) INTERNAL: Ref × slotCount Ref := u8 0 empty tree | u8 1 | i64le pageId | varint units | Summary stored tree @@ -88,6 +90,8 @@ For an internal node, "keys" are the separators. `separators[0]` is the first ch **Reference encoding** (`tree/Ref.java`). Every reference starts with a tag byte. `0` is the empty tree and nothing follows. `1` is a stored tree: the page number, the image size in 64-byte units and the subtree's summary. `2` is an inline tree: the entry count `n` (1 to 32), then the leaf exactly as a leaf page payload encodes it, keys as a zigzag first key and varint gaps followed by `schema.codec().encode(...)`. There is no summary, because the reader recomputes it from the entries. An inline tree can only be decoded where the reader knows its schema, through `Ref.read(in, schema)`. `AtomCodec` passes the member schema of the edge's kind for `members` and `order-index` for `order`. The tree changes in a WAL `Changes` record are read with their schema as well, so a nested tree change can carry an inline tree. Everywhere else (branch children, catalog images, posting and property directories) `Ref.read(in)` rejects tag `2`, since only nested references in codecs that opt in with `inlinesSmallTrees()` are ever inline. An inline reference is never larger than `Ref.maxEncodedSize()` (108 bytes); see [persistent-tree.md](persistent-tree.md#inline-trees) for why that keeps pages from overflowing. +**Framed values.** A codec whose values are written one after another, each without reference to the others, says so with `ValueCodec.independentValues()`: `AtomCodec` (the catalog), `PostingsCodec` (posting directories) and every `ValueCodec.rows` codec. In their leaves `NodeCodec` writes each value's length in front of it, so a reader can find one value without decoding the ones before it. The length is counted in `TreeSchema.entryBytes`, so the leaf budget still keeps every image within a page. Columnar codecs, such as the member and incidence codecs, keep the plain layout. When a framed leaf is read from a page, `NodeCodec.decode` walks the lengths to record where each value starts, rejects the image if a value runs past the payload or the values stop short of its end, and returns a packed leaf ([caching](#nodecache-treenodecachejava)). An inline tree of a framed codec uses the same layout and is decoded in full. + An edge record with three plain members `{1000, 1001, 1005}` holds this as its `members` reference. It takes 7 bytes. A stored reference to the same tree would take 65 bytes, most of them summary, and point at an image of at least 128: ```text @@ -346,10 +350,10 @@ and the reference journal. ### `NodeCache` (`tree/NodeCache.java`) -The cache stores **decoded, frozen** `Node` objects, not bytes. A hit therefore costs no decoding and no allocation. The cache has 16 shards; a node goes to shard `mix(pageId) >>> 60`. Each shard has: +The cache stores **decoded, frozen** `Node` objects, not bytes. A hit therefore costs no decoding and no allocation. The exception is a leaf of framed values, which stays **packed**: it holds the page image, the decoded keys and the offset of every value, and decodes a value the first time it is read (`Leaf.value`). The decoded value is kept in the leaf, published with a compare-and-set so another thread sees it whole, and two threads that decode the same value at once keep whichever got there first. A catalog lookup on a cold page therefore decodes the keys and one record instead of every record and every inline member list on the page. The cache has 16 shards; a node goes to shard `mix(pageId) >>> 60`. Each shard has: * an open-addressing table of `Slot(pageId, node, volatile referenced)`, probed linearly from `mix(pageId)`, so a lookup needs no boxed key. It is kept at most half full, counting removed slots, and is rebuilt into a new array when it would pass that; -* a byte budget of `max(64 KiB, node_cache_mb / 16)` and a circular, doubly linked ring of slots with a CLOCK hand. Each slot weighs `Node.heapBytes()`, an estimate of the decoded node's heap size: 144 bytes for the node and its summary, its key and value or child arrays, then 96 bytes plus the encoded value per leaf entry, or 128 bytes per child reference in a branch. A leaf whose values are implied by its keys or member ids counts only its `long` arrays, and a catalog leaf adds the estimate of every inline tree it holds. Those constants come from measuring retained heap against cache contents at two dataset sizes; +* a byte budget of `max(64 KiB, node_cache_mb / 16)` and a circular, doubly linked ring of slots with a CLOCK hand. Each slot weighs `Node.heapBytes()`, an estimate of the decoded node's heap size: 144 bytes for the node and its summary, its key and value or child arrays, then 96 bytes plus the encoded value per leaf entry, or 128 bytes per child reference in a branch. A leaf whose values are implied by its keys or member ids counts only its `long` arrays, and a catalog leaf adds the estimate of every inline tree it holds. A packed leaf starts at its image size plus its key and offset arrays. Each value it decodes adds 96 bytes plus the encoded value and any inline tree in it, and charges that to the leaf's slot (`Node.chargeTo`), which evicts under the shard lock until the shard is back within budget. A charge for a slot that was evicted, replaced or cleared is dropped. Those constants come from measuring retained heap against cache contents at two dataset sizes; * a `ReentrantLock` taken only for admission. **Reads are lock-free.** `get` probes the table, reading each slot with acquire ordering, plus setting `referenced = true` if it was clear. Writers change the table only under the shard lock, publishing each slot with release ordering, and a rebuild publishes the new array through a volatile field. A reader still probing the old array can find a node that was just evicted; that is harmless, because a page number's contents never change. The check avoids a redundant volatile write on hot entries. @@ -390,7 +394,7 @@ Because images are never overwritten in place, a torn write can only damage an i | Version | Where | Meaning | |---|---|---| -| `format=7` in `/FORMAT` | `StorageEngine.verifyFormat` | Commits logged as key changes (`Changes` WAL records), tree pages written at checkpoints, no history in catalog images. Also tagged references with small member and order trees inline in their edge record, reference counts in `pages.ref`, 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, `format=4` no reference counts, `format=5` untagged references with every nested tree on its own page, `format=6` commits that logged their pages) fail to open: `uses storage format 6; this build reads format 7 (commits logged as key changes); export and reload it`. | +| `format=8` in `/FORMAT` | `StorageEngine.verifyFormat` | Leaf values of codecs with independent values framed by their length. Also commits logged as key changes (`Changes` WAL records), tree pages written at checkpoints, no history in catalog images, tagged references with small member and order trees inline in their edge record, reference counts in `pages.ref`, 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, `format=4` no reference counts, `format=5` untagged references with every nested tree on its own page, `format=6` commits that logged their pages, `format=7` unframed leaf values) fail to open: `uses storage format 7; this build reads format 8 (leaf records framed by length); 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 and by page numbers. | | `pages.dir` header | `PageDirectory` | `i32 magic 0x52494450`, `i32 format 1`. | diff --git a/docs/storage/persistent-tree.md b/docs/storage/persistent-tree.md index d29de88a..a0e1d4d8 100644 --- a/docs/storage/persistent-tree.md +++ b/docs/storage/persistent-tree.md @@ -204,7 +204,7 @@ An inline leaf is frozen (`owner == null`), so a writer copies it before changin **Equality.** `Ref.same` compares stored references by page id and pending ones by identity. A pending node stamped by a published epoch is also the same as the reference it was written as, and two such nodes are the same when they were written as the same reference. That matters to a transaction that began before a checkpoint: the checkpoint swaps the head generation's pending nodes for stored references, and without this rule every tree the transaction touched would look changed. Inline references have no identity worth comparing: two reads of the same catalog page decode two separate leaves. They are compared by content instead. Two inline references are the same when they belong to the same schema and have the same keys and equal values (`Leaf.sameEntries`). The same content test applies between a pending leaf and an inline reference, and between two pending leaves, when both are small enough that a checkpoint would store them inline (`Ref.fitsInline`). `Ref.Inline.equals` uses the content test, so `EdgeRecord.equals` compares inline member trees by content too. This is what the conflict checks in `Transaction` and `Op`, the shortcuts in `TreeAlgebra` and `TreeDiff`, and the read validation of serializable transactions rely on. A stored and an inline reference are never the same, even with equal entries. Every caller treats "not the same" as "look closer", so that only costs a slower path. -**Heap.** The node cache weighs a catalog leaf with `Leaf.heapBytes()`, which adds the heap estimate of every inline leaf it holds, so inline trees count against `node_cache_mb` like any other node. +**Heap.** The node cache weighs a catalog leaf with `Leaf.heapBytes()`, which adds the heap estimate of every inline leaf it holds, so inline trees count against `node_cache_mb` like any other node. A catalog leaf read from a page is packed: it decodes a record, and the inline trees in it, only when that record is read, and charges the cache as it does ([pages.md](pages.md#nodecache-treenodecachejava)). ## Reading diff --git a/docs/transactions/wal-and-recovery.md b/docs/transactions/wal-and-recovery.md index f117699d..dd16a697 100644 --- a/docs/transactions/wal-and-recovery.md +++ b/docs/transactions/wal-and-recovery.md @@ -27,7 +27,7 @@ Related: [transactions](transactions.md), [change feed](../storage/change-feed.m ```text / - FORMAT properties: format=7, page-size= + FORMAT properties: format=8, 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) diff --git a/engine/src/main/java/io/hstore/engine/StorageEngine.java b/engine/src/main/java/io/hstore/engine/StorageEngine.java index 964fdcce..f44a56ec 100644 --- a/engine/src/main/java/io/hstore/engine/StorageEngine.java +++ b/engine/src/main/java/io/hstore/engine/StorageEngine.java @@ -44,7 +44,7 @@ public final class StorageEngine implements AutoCloseable { private static final System.Logger LOG = System.getLogger("hstore.engine"); - private static final String FORMAT_VERSION = "7"; + private static final String FORMAT_VERSION = "8"; private static final int CHANGED_COUNTS_LIMIT = 1 << 20; private static final int MAX_ATTEMPTS = 8; @@ -135,7 +135,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 + " (commits logged as key changes); export and reload it"); + + "; this build reads format " + FORMAT_VERSION + " (leaf records framed by length); 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/catalog/AtomCodec.java b/engine/src/main/java/io/hstore/engine/catalog/AtomCodec.java index 71e7679e..f403a0dc 100644 --- a/engine/src/main/java/io/hstore/engine/catalog/AtomCodec.java +++ b/engine/src/main/java/io/hstore/engine/catalog/AtomCodec.java @@ -79,6 +79,11 @@ public boolean holdsRefs() { return true; } + @Override + public boolean independentValues() { + return true; + } + @Override public boolean inlinesSmallTrees() { return true; diff --git a/engine/src/main/java/io/hstore/engine/index/PostingsCodec.java b/engine/src/main/java/io/hstore/engine/index/PostingsCodec.java index f4404315..0aab2e26 100644 --- a/engine/src/main/java/io/hstore/engine/index/PostingsCodec.java +++ b/engine/src/main/java/io/hstore/engine/index/PostingsCodec.java @@ -97,6 +97,11 @@ public boolean holdsRefs() { return true; } + @Override + public boolean independentValues() { + return true; + } + @Override public boolean refersToPages(Postings

value) { return value instanceof Postings.Promoted

(Ref root) && root instanceof Ref.Stored; diff --git a/engine/src/main/java/io/hstore/engine/tree/Leaf.java b/engine/src/main/java/io/hstore/engine/tree/Leaf.java index 1da4499c..9c31d1d3 100644 --- a/engine/src/main/java/io/hstore/engine/tree/Leaf.java +++ b/engine/src/main/java/io/hstore/engine/tree/Leaf.java @@ -5,30 +5,67 @@ import io.hstore.engine.page.ByteCursor; +import java.lang.foreign.MemorySegment; +import java.lang.invoke.MethodHandles; +import java.lang.invoke.VarHandle; import java.util.AbstractList; import java.util.Arrays; import java.util.List; import java.util.Objects; +import java.util.function.IntConsumer; import java.util.function.LongConsumer; public final class Leaf extends Node { private static final int ENTRY_HEAP_BYTES = 96; + private static final Object ABSENT = new Object(); + private static final VarHandle UNPACKED_VALUES = MethodHandles.arrayElementVarHandle(Object[].class); + private static final VarHandle UNPACKED_BYTES; + + static { + try { + UNPACKED_BYTES = MethodHandles.lookup().findVarHandle(Leaf.class, "unpackedBytes", int.class); + } catch (ReflectiveOperationException e) { + throw new ExceptionInInitializerError(e); + } + } private long[] keys; private Object[] values; private final long[] words; + private final MemorySegment packed; + private final int[] offsets; + private final boolean flaggedAsReferring; private int size; private int keyBytes; private int valueBytes; + private volatile boolean measured; + private volatile int unpackedBytes; + private volatile IntConsumer charge; private Leaf(TreeSchema schema, Object owner, long[] keys, Object[] values, long[] words, int size, Summary summary) { super(schema, owner, summary); this.keys = keys; this.values = values; this.words = words; + this.packed = null; + this.offsets = null; + this.flaggedAsReferring = false; this.size = size; - remeasure(); + if (owner != null) { + remeasure(); + } + } + + private Leaf(TreeSchema schema, long[] keys, MemorySegment packed, int[] offsets, boolean flaggedAsReferring, Summary summary) { + super(schema, null, summary); + this.keys = keys; + this.values = new Object[keys.length]; + this.words = null; + this.packed = packed; + this.offsets = offsets; + this.flaggedAsReferring = flaggedAsReferring; + this.size = keys.length; } private Leaf(TreeSchema schema, Object owner, long[] keys, Object[] values, int size, Summary summary) { @@ -47,6 +84,14 @@ static Leaf implied(TreeSchema schema, long[] keys, long[] words, Summary sum return new Leaf(schema, null, keys, null, words, keys.length, summary); } + static Leaf packed(TreeSchema schema, long[] keys, MemorySegment page, int[] offsets, boolean flaggedAsReferring, Summary summary) { + return new Leaf(schema, keys, page, offsets, flaggedAsReferring, summary); + } + + boolean pageFlagMatches() { + return packed == null || flaggedAsReferring == NodeCodec.refersToPages(this); + } + static int keyStreamBytes(long[] keys, int size) { if (size == 0) { return 0; @@ -63,33 +108,58 @@ private static int gap(long low, long high) { } private void remeasure() { - keyBytes = keyStreamBytes(keys, size); - valueBytes = 0; + int valueTotal = 0; for (int i = 0; i < size; i++) { - valueBytes += values != null ? schema.entryBytes(values[i]) : schema.impliedBytes(words[i]); + valueTotal += values != null ? schema.entryBytes(peek(i)) : schema.impliedBytes(words[i]); + } + keyBytes = keyStreamBytes(keys, size); + valueBytes = valueTotal; + measured = true; + } + + private void measure() { + if (!measured) { + remeasure(); + } + } + + @Override + void chargeTo(IntConsumer charge) { + if (packed != null) { + this.charge = charge; } } @Override int heapBytes() { + if (packed != null) { + return NODE_HEAP_BYTES + 3 * ARRAY_HEAP_BYTES + Long.BYTES * keys.length + Integer.BYTES * (values.length + offsets.length) + + Math.toIntExact(packed.byteSize()) + unpackedBytes; + } if (values == null) { return NODE_HEAP_BYTES + 2 * ARRAY_HEAP_BYTES + Long.BYTES * keys.length + (words == keys ? 0 : Long.BYTES * words.length); } + measure(); int bytes = NODE_HEAP_BYTES + 2 * ARRAY_HEAP_BYTES + Long.BYTES * keys.length + Integer.BYTES * values.length + ENTRY_HEAP_BYTES * size + valueBytes; return schema.codec().holdsRefs() ? bytes + inlineHeapBytes(schema) : bytes; } private int inlineHeapBytes(TreeSchema typed) { + int total = 0; + for (int i = 0; i < size; i++) { + total += inlineHeapBytes(typed, value(i)); + } + return total; + } + + private static int inlineHeapBytes(TreeSchema typed, Object value) { int[] total = new int[1]; - ValueCodec.RefVisitor visitor = (ref, _) -> { + typed.codec().forEachRef(typed.cast(value), (ref, _) -> { if (ref instanceof Ref.Inline(Leaf inline)) { total[0] += inline.heapBytes(); } - }; - for (int i = 0; i < size; i++) { - typed.codec().forEachRef(typed.cast(value(i)), visitor); - } + }); return total[0]; } @@ -148,10 +218,55 @@ long[] impliedWords() { Object value(int index) { Object[] stored = values; - return stored != null ? stored[index] : schema.implied(words[index]); + if (stored == null) { + return schema.implied(words[index]); + } + if (packed == null) { + return stored[index]; + } + Object unpacked = UNPACKED_VALUES.getAcquire(stored, index); + if (unpacked == null) { + unpacked = unpack(stored, index); + } + return unpacked == ABSENT ? null : unpacked; + } + + private Object peek(int index) { + if (packed == null) { + return value(index); + } + Object unpacked = UNPACKED_VALUES.getAcquire(values, index); + if (unpacked == null) { + return decode(index); + } + return unpacked == ABSENT ? null : unpacked; + } + + private Object decode(int index) { + Object[] decoded = new Object[1]; + schema.codec().decode(ByteCursor.over(packed).position(offsets[index]), new long[]{keys[index]}, decoded); + return decoded[0]; + } + + private Object unpack(Object[] stored, int index) { + Object[] decoded = {decode(index)}; + Object value = decoded[0] == null ? ABSENT : decoded[0]; + Object raced = UNPACKED_VALUES.compareAndExchangeRelease(stored, index, null, value); + if (raced != null) { + return raced; + } + int grown = ENTRY_HEAP_BYTES + schema.entryBytes(decoded[0]) + + (schema.codec().holdsRefs() ? inlineHeapBytes(schema, decoded[0]) : 0); + UNPACKED_BYTES.getAndAdd(this, grown); + IntConsumer target = charge; + if (target != null) { + target.accept(grown); + } + return value; } int bytes() { + measure(); return keyBytes + valueBytes + schema.leafOverhead(size); } @@ -195,17 +310,18 @@ Leaf frozenWith(Object[] mappedValues) { } private Object[] valueArray(int capacity) { - if (values != null) { + if (values != null && packed == null) { return Arrays.copyOf(values, capacity); } Object[] copy = new Object[capacity]; for (int i = 0; i < size; i++) { - copy[i] = value(i); + copy[i] = peek(i); } return copy; } void insert(int position, long key, Object value) { + measure(); if (size == keys.length) { int capacity = Math.max(8, size + (size >> 1)); keys = Arrays.copyOf(keys, capacity); @@ -235,12 +351,14 @@ private int insertionDelta(int position, long key) { } void replace(int position, Object value) { + measure(); valueBytes += schema.entryBytes(value) - schema.entryBytes(values[position]); values[position] = value; invalidate(); } void remove(int position) { + measure(); long key = keys[position]; valueBytes -= schema.entryBytes(values[position]); System.arraycopy(keys, position + 1, keys, position, size - position - 1); diff --git a/engine/src/main/java/io/hstore/engine/tree/Node.java b/engine/src/main/java/io/hstore/engine/tree/Node.java index 0416ca3e..7a66b9ba 100644 --- a/engine/src/main/java/io/hstore/engine/tree/Node.java +++ b/engine/src/main/java/io/hstore/engine/tree/Node.java @@ -3,6 +3,8 @@ package io.hstore.engine.tree; +import java.util.function.IntConsumer; + public sealed abstract class Node permits Leaf, Branch { static final int NODE_HEAP_BYTES = 144; @@ -66,4 +68,7 @@ final boolean writableBy(WriteScope scope) { abstract int size(); abstract int heapBytes(); + + void chargeTo(IntConsumer charge) { + } } diff --git a/engine/src/main/java/io/hstore/engine/tree/NodeCache.java b/engine/src/main/java/io/hstore/engine/tree/NodeCache.java index b9997fbf..85ed75e8 100644 --- a/engine/src/main/java/io/hstore/engine/tree/NodeCache.java +++ b/engine/src/main/java/io/hstore/engine/tree/NodeCache.java @@ -69,21 +69,23 @@ private Shard shard(long pageId) { private static final class Slot { final long pageId; final Node node; - final int bytes; + final int generation; + int bytes; volatile boolean referenced; Slot previous; Slot next; - Slot(long pageId, Node node, int bytes) { + Slot(long pageId, Node node, int bytes, int generation) { this.pageId = pageId; this.node = node; this.bytes = bytes; + this.generation = generation; } } private static final class Shard { private static final int MIN_TABLE = 64; - private static final Slot GONE = new Slot(0, null, 0); + private static final Slot GONE = new Slot(0, null, 0, -1); private static final VarHandle SLOTS = MethodHandles.arrayElementVarHandle(Slot[].class); private final ReentrantLock admission = new ReentrantLock(); @@ -92,6 +94,7 @@ private static final class Shard { private volatile long bytes; private int live; private int gone; + private int generation; private Slot hand; Shard(long budget) { @@ -190,9 +193,26 @@ void put(long pageId, Node node, int weight) { while (bytes + weight > budget) { evict(); } - Slot slot = new Slot(pageId, node, weight); + Slot slot = new Slot(pageId, node, weight, generation); link(slot); insert(slot); + node.chargeTo(grown -> charge(slot, grown)); + } finally { + admission.unlock(); + } + } + + private void charge(Slot slot, int grown) { + admission.lock(); + try { + if (slot.next == null || slot.generation != generation) { + return; + } + slot.bytes += grown; + bytes += grown; + while (bytes > budget && hand != null) { + evict(); + } } finally { admission.unlock(); } @@ -243,6 +263,7 @@ void clear() { table = new Slot[MIN_TABLE]; live = 0; gone = 0; + generation++; hand = null; bytes = 0; } finally { diff --git a/engine/src/main/java/io/hstore/engine/tree/NodeCodec.java b/engine/src/main/java/io/hstore/engine/tree/NodeCodec.java index 373db8ca..17a79ace 100644 --- a/engine/src/main/java/io/hstore/engine/tree/NodeCodec.java +++ b/engine/src/main/java/io/hstore/engine/tree/NodeCodec.java @@ -9,6 +9,8 @@ import io.hstore.engine.page.PageType; import java.lang.foreign.MemorySegment; +import java.lang.foreign.ValueLayout; +import java.util.Collections; import java.util.List; public final class NodeCodec { @@ -72,10 +74,51 @@ private static Leaf decodeLeaf(ByteCursor in, TreeSchema schema, long[] keys, return Leaf.implied(schema, keys, words, summary); } Object[] values = new Object[keys.length]; - schema.codec().decode(in, keys, values); + if (schema.codec().independentValues()) { + decodeFramed(in, schema, keys, values); + } else { + schema.codec().decode(in, keys, values); + } return Leaf.decoded(schema, keys, values, summary); } + private static int[] offsets(ByteCursor in, int count, PageHeader header, long pageId) { + int[] offsets = new int[count]; + long end = PageHeader.SIZE + (long) header.payloadLength(); + for (int i = 0; i < count; i++) { + long length = in.getVarLong(); + offsets[i] = Math.toIntExact(in.position()); + if (length < 0 || in.position() + length > end) { + throw HStoreException.corrupt(pageId, "value " + i + " runs past the end of the page"); + } + in.skip(length); + } + if (in.position() != end) { + throw HStoreException.corrupt(pageId, "values end at byte " + in.position() + ", not at the end of the payload " + end); + } + return offsets; + } + + private static MemorySegment owned(MemorySegment page) { + return page.isNative() ? MemorySegment.ofArray(page.toArray(ValueLayout.JAVA_BYTE)) : page; + } + + private static void decodeFramed(ByteCursor in, TreeSchema schema, long[] keys, Object[] into) { + long[] key = new long[1]; + Object[] value = new Object[1]; + for (int i = 0; i < keys.length; i++) { + long length = in.getVarLong(); + long end = in.position() + length; + key[0] = keys[i]; + schema.codec().decode(in, key, value); + if (in.position() != end) { + throw new IllegalStateException("value " + i + " of a " + schema + " leaf is " + length + " bytes long but decoded from " + + (in.position() - end + length) + " bytes"); + } + into[i] = value[0]; + } + } + private static long[] decodeKeys(ByteCursor in, int count) { long[] keys = new long[count]; keys[0] = in.getSignedVarLong(); @@ -93,10 +136,22 @@ private static void encodeValues(ByteCursor out, TreeSchema schema, long[ } @SuppressWarnings("unchecked") List values = (List) leaf.valueList(); - schema.codec().encode(out, keys, values); + if (!schema.codec().independentValues()) { + schema.codec().encode(out, keys, values); + return; + } + ByteCursor value = ByteCursor.growable(256); + long[] key = new long[1]; + for (int i = 0; i < keys.length; i++) { + value.position(0); + key[0] = keys[i]; + schema.codec().encode(value, key, Collections.singletonList(values.get(i))); + MemorySegment written = value.written(); + out.putVarLong(written.byteSize()).putSegment(written); + } } - private static boolean refersToPages(Leaf leaf) { + static boolean refersToPages(Leaf leaf) { return leaf.schema.codec().holdsRefs() && refersToPages(leaf.schema, leaf.valueList()); } @@ -137,6 +192,9 @@ public static Node decode(MemorySegment page, PageHeader header, long pageId, Tr long[] keys = decodeKeys(in, count); return switch (header.type()) { case LEAF -> { + if (schema.codec().independentValues()) { + yield Leaf.packed(schema, keys, owned(page), offsets(in, count, header, pageId), header.refersToPages(), summary); + } Leaf leaf = decodeLeaf(in, schema, keys, summary); if (header.refersToPages() != refersToPages(leaf)) { throw HStoreException.corrupt(pageId, "page header and contents disagree on whether it refers to other pages"); diff --git a/engine/src/main/java/io/hstore/engine/tree/TreeSchema.java b/engine/src/main/java/io/hstore/engine/tree/TreeSchema.java index d5a28698..bf47ed38 100644 --- a/engine/src/main/java/io/hstore/engine/tree/TreeSchema.java +++ b/engine/src/main/java/io/hstore/engine/tree/TreeSchema.java @@ -3,6 +3,8 @@ package io.hstore.engine.tree; +import io.hstore.engine.page.ByteCursor; + import java.util.List; public record TreeSchema(int id, String name, FingerprintMode fingerprint, ValueCodec codec, EntryMeasure measure) { @@ -21,7 +23,8 @@ public V cast(Object value) { } int entryBytes(Object value) { - return codec.maxSize(cast(value)); + int size = codec.maxSize(cast(value)); + return codec.independentValues() ? size + ByteCursor.varLongSize(size) : size; } Object implied(long word) { diff --git a/engine/src/main/java/io/hstore/engine/tree/TreeVerifier.java b/engine/src/main/java/io/hstore/engine/tree/TreeVerifier.java index 2714d519..2a5b1d8c 100644 --- a/engine/src/main/java/io/hstore/engine/tree/TreeVerifier.java +++ b/engine/src/main/java/io/hstore/engine/tree/TreeVerifier.java @@ -60,6 +60,9 @@ private Visit visit(Ref ref, TreeSchema schema, int expectedHeight) { report = report.plus(nested(schema, leaf.value(i))); } } + if (!leaf.pageFlagMatches()) { + throw fault(ref, "page header and contents disagree on whether it refers to other pages in " + schema); + } report = report.plus(new Report(0, leaf.size(), 0, 0)); } case Branch branch -> { diff --git a/engine/src/main/java/io/hstore/engine/tree/ValueCodec.java b/engine/src/main/java/io/hstore/engine/tree/ValueCodec.java index 77171bd6..e54d08c8 100644 --- a/engine/src/main/java/io/hstore/engine/tree/ValueCodec.java +++ b/engine/src/main/java/io/hstore/engine/tree/ValueCodec.java @@ -47,6 +47,10 @@ default boolean holdsRefs() { return false; } + default boolean independentValues() { + return false; + } + default V mapRefs(V value, RefMapper mapper) { return value; } @@ -86,6 +90,11 @@ public void decode(ByteCursor in, long[] keys, Object[] into) { into[i] = reader.apply(in); } } + + @Override + public boolean independentValues() { + return true; + } }; } diff --git a/engine/src/test/java/io/hstore/engine/CompactionTest.java b/engine/src/test/java/io/hstore/engine/CompactionTest.java index 5e376477..d5ea5d91 100644 --- a/engine/src/test/java/io/hstore/engine/CompactionTest.java +++ b/engine/src/test/java/io/hstore/engine/CompactionTest.java @@ -73,7 +73,7 @@ void compactionMovesPagesWithoutWritingAGeneration() { @Test void retainedHistoryStaysReadableOnceItsSegmentsAreDeleted() { - EngineOptions options = options().withHistoryLimit(20); + EngineOptions options = options().withHistoryLimit(20).withCompactionLiveRatio(0.75); List written = Stream.iterate(500_000L, i -> i + 1).limit(2_000).toList(); long large; long edge; diff --git a/engine/src/test/java/io/hstore/engine/EngineTest.java b/engine/src/test/java/io/hstore/engine/EngineTest.java index 507ab8cb..e951cb60 100644 --- a/engine/src/test/java/io/hstore/engine/EngineTest.java +++ b/engine/src/test/java/io/hstore/engine/EngineTest.java @@ -614,10 +614,10 @@ void dictionaryInternsRolesAcrossRestarts() { } @Test - void directoriesFromTheInlineMemberFormatAreRefused() throws IOException { - Files.writeString(directory.resolve("FORMAT"), "format=6\npage-size=1024\n"); + void directoriesFromBeforeFramedLeafRecordsAreRefused() throws IOException { + Files.writeString(directory.resolve("FORMAT"), "format=7\npage-size=1024\n"); HStoreException refused = assertThrows(HStoreException.class, () -> StorageEngine.open(directory, small()).close()); - assertTrue(refused.getMessage().endsWith("uses storage format 6; this build reads format 7 (commits logged as key changes); export and reload it"), + assertTrue(refused.getMessage().endsWith("uses storage format 7; this build reads format 8 (leaf records framed by length); export and reload it"), refused.getMessage()); } } 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 679ca5cd..118b5070 100644 --- a/engine/src/test/java/io/hstore/engine/tree/MemoryPages.java +++ b/engine/src/test/java/io/hstore/engine/tree/MemoryPages.java @@ -37,6 +37,10 @@ public void accept(long pageId, MemorySegment image, Node frozen) { return tree.withRoot(materializer.materialize(tree.root(), tree.schema())); } + MemorySegment image(long pageId) { + return pages.get(pageId); + } + @Override public Node load(long pageId, TreeSchema schema) { loads++; diff --git a/engine/src/test/java/io/hstore/engine/tree/PackedLeafTest.java b/engine/src/test/java/io/hstore/engine/tree/PackedLeafTest.java new file mode 100644 index 00000000..5eb195b8 --- /dev/null +++ b/engine/src/test/java/io/hstore/engine/tree/PackedLeafTest.java @@ -0,0 +1,200 @@ +// SPDX-FileCopyrightText: Metacog Labs +// SPDX-License-Identifier: PolyForm-Noncommercial-1.0.0 + +package io.hstore.engine.tree; + +import io.hstore.engine.HStoreException; +import io.hstore.engine.catalog.AtomRecord; +import io.hstore.engine.catalog.EngineSlots; +import io.hstore.engine.page.ByteCursor; +import io.hstore.engine.page.PageHeader; +import io.hstore.engine.topology.EdgeKind; +import io.hstore.engine.topology.Incidence; +import io.hstore.engine.topology.TopologySchemas; + +import org.junit.jupiter.api.Test; + +import java.lang.foreign.MemorySegment; +import java.lang.foreign.ValueLayout; +import java.util.ArrayList; +import java.util.List; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.Future; +import java.util.concurrent.atomic.AtomicLong; +import java.util.random.RandomGenerator; +import java.util.random.RandomGeneratorFactory; + +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; + +class PackedLeafTest { + + private static final int PAGE = 16 * 1024; + private static final int RECORDS = 60; + + private final MemoryPages pages = new MemoryPages(PAGE); + private final List expected = new ArrayList<>(); + private final Ref root; + private final long rootPage; + + PackedLeafTest() { + WriteScope scope = new WriteScope(); + Tree catalog = Tree.empty(EngineSlots.ATOM_SCHEMA, pages); + for (int atom = 0; atom < RECORDS; atom++) { + AtomRecord record = atom % 3 == 0 ? AtomRecord.NodeRecord.of(1, "node-" + atom) : edge(scope, atom); + catalog = catalog.put(scope, atom, record); + } + Tree written = pages.persist(catalog); + root = written.root(); + rootPage = ((Ref.Stored) root).pageId(); + written.stream().forEach(entry -> expected.add(entry.value())); + } + + private AtomRecord edge(WriteScope scope, int atom) { + Tree members = Tree.empty(TopologySchemas.SET_MEMBERS, pages); + for (long member = 1; member <= 1 + atom % 6; member++) { + members = members.put(scope, 1_000L * atom + member, Incidence.of(1_000L * atom + member)); + } + return new AtomRecord.EdgeRecord(2, EdgeKind.SET, members.root(), Ref.EMPTY, atom, 0, 0, 0); + } + + private Leaf load() { + return (Leaf) NodeCodec.decode(pages.image(rootPage), rootPage, EngineSlots.ATOM_SCHEMA); + } + + @Test + void aPackedLeafDecodesEachRecordOnlyWhenItIsRead() { + Leaf leaf = load(); + assertEquals(RECORDS, leaf.size()); + int packedOnly = leaf.heapBytes(); + assertEquals(expected.get(17), leaf.value(17)); + int oneRecord = leaf.heapBytes(); + assertTrue(oneRecord > packedOnly); + assertEquals(expected.get(17), leaf.value(17)); + assertEquals(oneRecord, leaf.heapBytes(), "a record read twice is unpacked once"); + for (int i = 0; i < RECORDS; i++) { + assertEquals(expected.get(i), leaf.value(i)); + } + Leaf eager = Leaf.decoded(EngineSlots.ATOM_SCHEMA, leaf.keyArray(), expected.toArray(), null); + assertTrue(leaf.sameEntries(eager)); + assertEquals(eager.bytes(), leaf.bytes()); + } + + @Test + void aWriterCopiesAPackedLeafWithoutKeepingOrChargingItsRecords() { + Leaf leaf = load(); + int packedOnly = leaf.heapBytes(); + AtomicLong charged = new AtomicLong(); + leaf.chargeTo(charged::addAndGet); + Leaf copy = leaf.copyFor(new WriteScope()); + for (int i = 0; i < RECORDS; i++) { + assertEquals(expected.get(i), copy.value(i)); + } + assertEquals(leaf.bytes(), copy.bytes()); + assertEquals(packedOnly, leaf.heapBytes()); + assertEquals(0, charged.get()); + } + + @Test + void threadsReadingOneLeafSeeTheSameRecordsAndChargeEachOnce() throws Exception { + Leaf leaf = load(); + int packedOnly = leaf.heapBytes(); + AtomicLong charged = new AtomicLong(); + leaf.chargeTo(charged::addAndGet); + List> readers = new ArrayList<>(); + try (ExecutorService pool = Executors.newFixedThreadPool(8)) { + for (int thread = 0; thread < 8; thread++) { + RandomGenerator random = RandomGeneratorFactory.of("L64X128MixRandom").create(thread); + readers.add(pool.submit(() -> { + for (int round = 0; round < 2_000; round++) { + int i = random.nextInt(RECORDS); + assertEquals(expected.get(i), leaf.value(i)); + } + })); + } + for (Future reader : readers) { + reader.get(); + } + } + for (int i = 0; i < RECORDS; i++) { + leaf.value(i); + } + assertEquals(leaf.heapBytes() - packedOnly, charged.get()); + } + + @Test + void theCacheCountsRecordsAsTheyAreUnpackedAndStaysWithinBudget() { + Leaf leaf = load(); + NodeCache cache = new NodeCache(1 << 20, PAGE); + cache.put(rootPage, leaf, leaf.heapBytes()); + long before = cache.residentBytes(); + for (int i = 0; i < RECORDS; i++) { + leaf.value(i); + } + assertTrue(cache.residentBytes() > before); + assertEquals(leaf.heapBytes(), cache.residentBytes()); + + Leaf small = load(); + NodeCache tight = new NodeCache(small.heapBytes() + 2_000, PAGE); + tight.put(rootPage, small, small.heapBytes()); + for (int i = 0; i < RECORDS; i++) { + small.value(i); + assertTrue(tight.residentBytes() <= small.heapBytes() + 2_000, "resident " + tight.residentBytes()); + } + assertEquals(0, tight.residentBytes(), "the grown leaf no longer fits and is evicted"); + } + + @Test + void chargesForAnEvictedOrClearedLeafAreIgnored() { + NodeCache cache = new NodeCache(1 << 20, PAGE); + Leaf evicted = load(); + cache.put(rootPage, evicted, evicted.heapBytes()); + cache.put(rootPage, load(), 100); + long resident = cache.residentBytes(); + evicted.value(3); + assertEquals(resident, cache.residentBytes()); + + Leaf cleared = load(); + cache.put(rootPage + 1, cleared, cleared.heapBytes()); + cache.clear(); + cleared.value(4); + assertEquals(0, cache.residentBytes()); + } + + @Test + void checkingTheTreeFindsAPackedLeafWhoseHeaderClaimsPageReferencesItLacks() { + MemorySegment image = pages.image(rootPage); + PageHeader header = PageHeader.verify(image, rootPage); + assertTrue(!header.refersToPages()); + new TreeVerifier(pages).verify(root, EngineSlots.ATOM_SCHEMA); + image.set(ValueLayout.JAVA_BYTE, 7, (byte) (image.get(ValueLayout.JAVA_BYTE, 7) | 1)); + PageHeader.seal(image, header.payloadLength()); + RuntimeException fault = assertThrows(RuntimeException.class, () -> new TreeVerifier(pages).verify(root, EngineSlots.ATOM_SCHEMA)); + assertTrue(fault.getMessage().contains("disagree on whether it refers to other pages"), fault.getMessage()); + } + + @Test + void aLengthThatDoesNotMatchTheRecordsIsReportedAsCorruption() { + MemorySegment image = pages.image(rootPage); + PageHeader header = PageHeader.verify(image, rootPage); + ByteCursor in = ByteCursor.over(image).position(PageHeader.SIZE); + for (int field = 0; field < 5; field++) { + in.getSignedVarLong(); + } + in.getLong(); + in.getSignedVarLong(); + for (int key = 1; key < header.slotCount(); key++) { + in.getVarLong(); + } + long first = in.position(); + MemorySegment corrupt = MemorySegment.ofArray(image.toArray(ValueLayout.JAVA_BYTE)); + byte length = corrupt.get(ValueLayout.JAVA_BYTE, first); + assertNotEquals(0, length); + corrupt.set(ValueLayout.JAVA_BYTE, first, (byte) (length + 1)); + PageHeader.seal(corrupt, header.payloadLength()); + assertThrows(HStoreException.class, () -> NodeCodec.decode(corrupt, rootPage, EngineSlots.ATOM_SCHEMA)); + } +} diff --git a/engine/src/test/java/io/hstore/engine/tree/TreeTest.java b/engine/src/test/java/io/hstore/engine/tree/TreeTest.java index abc16953..52dbcfcf 100644 --- a/engine/src/test/java/io/hstore/engine/tree/TreeTest.java +++ b/engine/src/test/java/io/hstore/engine/tree/TreeTest.java @@ -406,6 +406,10 @@ void materializingAnEdgeRecordInlinesSmallTreesWithTheSameSummary() { assertEquals(edge.contentHash(), inlined.contentHash()); assertEquals(small.stream().toList(), new Tree<>(TopologySchemas.SET_MEMBERS, pages, inlined.members()).stream().toList()); Leaf root = (Leaf) written.node(written.root()); + int packedOnly = root.heapBytes(); + assertEquals(inlined, root.value(0)); + assertEquals(paged, root.value(1)); + assertTrue(root.heapBytes() > packedOnly); Leaf bare = Leaf.decoded(EngineSlots.ATOM_SCHEMA, new long[]{1}, new Object[]{AtomRecord.EdgeRecord.of(2, EdgeKind.SET)}, null); assertTrue(root.heapBytes() >= bare.heapBytes() + ((Ref.Inline) inlined.members()).leaf().heapBytes()); }