diff --git a/benchmarks/src/main/java/io/hstore/bench/HStoreStore.java b/benchmarks/src/main/java/io/hstore/bench/HStoreStore.java index 8611c144..e4519e77 100644 --- a/benchmarks/src/main/java/io/hstore/bench/HStoreStore.java +++ b/benchmarks/src/main/java/io/hstore/bench/HStoreStore.java @@ -16,7 +16,6 @@ import io.hstore.engine.txn.Durability; import java.nio.file.Path; -import java.util.ArrayList; import java.util.Arrays; import java.util.List; import java.util.Map; @@ -120,10 +119,10 @@ public void ingestEdges(Dataset dataset, int batch) { } } - private List members(int[] indexes) { - List members = new ArrayList<>(indexes.length); - for (int index : indexes) { - members.add(nodes[index]); + private long[] members(int[] indexes) { + long[] members = new long[indexes.length]; + for (int i = 0; i < indexes.length; i++) { + members[i] = nodes[indexes[i]]; } return members; } @@ -191,11 +190,7 @@ public void update(int[] probes, int from, int to, int round) { public void largeEdge(int count) { large = database.write(writer -> { long edge = writer.edge(EDGE); - List members = new ArrayList<>(count); - for (int i = 0; i < count; i++) { - members.add(nodes[i]); - } - writer.load(edge, members, MemberSpec.PLAIN); + writer.load(edge, Arrays.copyOf(nodes, count), MemberSpec.PLAIN); return edge; }); } diff --git a/database/src/main/java/io/hstore/db/Writer.java b/database/src/main/java/io/hstore/db/Writer.java index f6325a37..cc1f3b24 100644 --- a/database/src/main/java/io/hstore/db/Writer.java +++ b/database/src/main/java/io/hstore/db/Writer.java @@ -21,6 +21,7 @@ import io.hstore.db.value.TypeTag; import io.hstore.db.value.Value; import io.hstore.db.value.Values; +import io.hstore.engine.Dictionary; import io.hstore.engine.HStoreException; import io.hstore.engine.catalog.AtomRecord; import io.hstore.engine.catalog.EngineSlots; @@ -31,6 +32,7 @@ import io.hstore.engine.txn.Transaction; import java.nio.charset.StandardCharsets; +import java.util.Arrays; import java.util.List; import java.util.Map; import java.util.Optional; @@ -138,13 +140,11 @@ private TypeDef kindOf(String type, AtomKind kind) { } private void assign(long atom, TypeDef definition, Map properties) { - definition.properties().stream() - .filter(PropertyDef::required) - .filter(property -> !properties.containsKey(property.name())) - .findFirst() - .ifPresent(missing -> { - throw HStoreException.invalid(definition.name() + " requires property '" + missing.name() + "'"); - }); + for (PropertyDef property : definition.properties()) { + if (property.required() && !properties.containsKey(property.name())) { + throw HStoreException.invalid(definition.name() + " requires property '" + property.name() + "'"); + } + } properties.forEach((name, value) -> set(atom, name, Value.of(value))); } @@ -178,7 +178,19 @@ public void removeAt(long edge, long index) { } public void load(long edge, List members, MemberSpec spec) { - txn.load(edge, members.stream().map(member -> incidence(edge, member, spec)).toList()); + load(edge, members.stream().mapToLong(Long::longValue).toArray(), spec); + } + + public void load(long edge, long[] members, MemberSpec spec) { + Incidence[] incidences = new Incidence[members.length]; + if (members.length > 0) { + Incidence first = incidences[0] = incidence(edge, members[0], spec); + for (int i = 1; i < members.length; i++) { + requireVisible(members[i]); + incidences[i] = new Incidence(members[i], first.roleSet(), first.weight(), first.validFrom(), first.validTo(), first.dataRef(), first.qualifier()); + } + } + txn.load(edge, Arrays.asList(incidences)); } public void qualify(long edge, long member, long qualifier) { @@ -188,20 +200,30 @@ public void qualify(long edge, long member, long qualifier) { } private Incidence incidence(long edge, long member, MemberSpec spec) { + requireMembership(edge, spec); + requireVisible(member); + int roleSet = spec.roles().isEmpty() ? Dictionary.NO_ROLES : database.engine().dictionary().roleSet(spec.roles()); + long weight = spec.weight().isPresent() ? Weight.of(spec.weight().getAsDouble()) : Weight.ONE; + return new Incidence(member, roleSet, weight, spec.validity().from(), spec.validity().to(), spec.dataRef(), spec.qualifier()); + } + + private void requireMembership(long edge, MemberSpec spec) { principal().requireWrite(); Atom container = require(edge); if (!container.isEdge()) { throw HStoreException.invalid("atom " + edge + " is not a hyperedge"); } - spec.roles().stream().filter(role -> !container.type().allowsRole(role)).findFirst().ifPresent(role -> { - throw HStoreException.invalid(container.type().name() + " does not declare role '" + role + "'"); - }); + for (String role : spec.roles()) { + if (!container.type().allowsRole(role)) { + throw HStoreException.invalid(container.type().name() + " does not declare role '" + role + "'"); + } + } + } + + private void requireVisible(long member) { if (!visible(member)) { throw HStoreException.invalid("member atom " + member + " does not exist"); } - int roleSet = database.engine().dictionary().roleSet(spec.roles()); - return new Incidence(member, roleSet, spec.weight().stream().mapToLong(Weight::of).findFirst().orElse(Weight.ONE), - spec.validity().from(), spec.validity().to(), spec.dataRef(), spec.qualifier()); } public void set(long atom, String name, Value value) { diff --git a/database/src/test/java/io/hstore/db/DatabaseTest.java b/database/src/test/java/io/hstore/db/DatabaseTest.java index 1656ab4c..7754709a 100644 --- a/database/src/test/java/io/hstore/db/DatabaseTest.java +++ b/database/src/test/java/io/hstore/db/DatabaseTest.java @@ -136,6 +136,49 @@ void membersCarryRolesWeightsAndValidity() { }); } + @Test + void bulkLoadedMembersMatchMembersAddedOneAtATime() { + MemberSpec spec = MemberSpec.roles("author").withWeight(0.5).withValidity(new Validity(10, 20)); + long[] edges = database.write(writer -> { + long[] people = new long[5]; + for (int i = 0; i < people.length; i++) { + people[i] = writer.node("Person", "p" + i, Map.of("name", "Person " + i)); + } + long added = writer.edge("Coauthorship", Map.of("year", 2024)); + for (long person : people) { + writer.add(added, person, spec); + } + long fromArray = writer.edge("Coauthorship", Map.of("year", 2024)); + writer.load(fromArray, new long[]{people[3], people[0], people[4], people[1], people[2]}, spec); + long fromList = writer.edge("Coauthorship", Map.of("year", 2024)); + writer.load(fromList, List.of(people[4], people[2], people[0], people[3], people[1]), spec); + long empty = writer.edge("Coauthorship", Map.of("year", 2024)); + writer.load(empty, new long[0], spec); + return new long[]{added, fromArray, fromList, empty}; + }); + database.read(reader -> { + List expected = reader.members(edges[0]).toList(); + assertEquals(5, expected.size()); + assertEquals(expected, reader.members(edges[1]).toList()); + assertEquals(expected, reader.members(edges[2]).toList()); + assertEquals(0, reader.cardinality(edges[3])); + return null; + }); + long node = database.read(reader -> reader.resolve("Person", "p0")); + assertThrows(HStoreException.InvalidSchema.class, () -> database.write(writer -> { + writer.load(edges[3], new long[]{node}, MemberSpec.roles("reviewer")); + return null; + })); + assertThrows(HStoreException.class, () -> database.write(writer -> { + writer.load(edges[3], new long[]{node, 999_999_999L}, spec); + return null; + })); + assertThrows(HStoreException.class, () -> database.write(writer -> { + writer.load(node, new long[]{node}, spec); + return null; + })); + } + @Test void incidentEdgeVisitorsSeeTheSameEdgesAsTheIncidenceStream() { long[] people = database.write(writer -> { diff --git a/docs/database/data-model.md b/docs/database/data-model.md index 36a94834..15b8a85d 100644 --- a/docs/database/data-model.md +++ b/docs/database/data-model.md @@ -450,7 +450,7 @@ Optional.empty | `add(edge, member, spec)` | WRITER | upsert an incidence (roles validated against the type) | | `insertAt(edge, index, member, spec)` / `append(...)` | WRITER | ordered edges | | `remove(edge, member)` / `removeAt(edge, index)` | WRITER | membership removal | -| `load(edge, members, spec)` | WRITER | replaces the whole membership with a tree built bottom-up from `members` (`Hyperedge.build`) | +| `load(edge, members, spec)` | WRITER | replaces the whole membership with a tree built bottom-up from `members` (`Hyperedge.build`); `members` is a `List` or a `long[]`, and the edge and the roles are checked once per call | | `qualify(edge, member, qualifier)` | WRITER | attach an assertion id to an incidence | | `set(atom, name, value[, validity])` / `unset(atom, name)` | WRITER | per-key OCC on the property bag | | `document(atom, json)` | WRITER | stores `$document` | diff --git a/docs/storage/change-feed.md b/docs/storage/change-feed.md index 7436b2fc..415cdf46 100644 --- a/docs/storage/change-feed.md +++ b/docs/storage/change-feed.md @@ -77,7 +77,7 @@ offset size field **Open.** `restore()` opens every `*.feed`, scans all frames from the first segment's start to the end of the last segment, and stops at the first frame that is short, has a negative or overlong length, fails its CRC, or is not newer than the frame before it. The CRC covers only the payload, so a frame also counts as damaged unless its payload is non-empty and starts with the same generation as its header (`FeedCodec` always writes the generation first); this keeps a damaged header, or a run of zeros left where an unsynced frame never reached the disk, from passing as a frame. `cutAt(position)` then deletes every segment that starts after the failure point and truncates the segment containing it, so the feed always ends at a clean frame. -**Index.** Each `Segment` remembers `firstGeneration`, the generation of its first frame. An in-memory sparse index `generation -> global position` holds the first frame of **every segment** and every 64th frame thereafter (`INDEX_STRIDE = 64`). `startFor(g)` seeks to the floor entry (never before the first retained segment) and scans forward, so locating a generation costs at most 63 frame reads plus segment boundaries. +**Index.** Each `Segment` remembers `firstGeneration`, the generation of its first frame. An in-memory sparse index `generation -> global position` holds the first frame of **every segment** and every 64th frame thereafter (`INDEX_STRIDE = 64`). `startFor(g)` seeks to the floor entry (never before the first retained segment) and scans forward, so locating a generation costs at most 63 frame reads plus segment boundaries. `replay(after)` reads only the 16-byte header of a frame whose generation is at most `after` and steps over its payload, so a subscriber that wakes after every commit doesn't read the payloads of the frames it has already consumed again. ```text feed/ diff --git a/engine/src/main/java/io/hstore/engine/feed/ChangeFeed.java b/engine/src/main/java/io/hstore/engine/feed/ChangeFeed.java index 45610b3b..6f802c9c 100644 --- a/engine/src/main/java/io/hstore/engine/feed/ChangeFeed.java +++ b/engine/src/main/java/io/hstore/engine/feed/ChangeFeed.java @@ -10,8 +10,8 @@ import java.io.IOException; import java.io.UncheckedIOException; -import java.lang.foreign.MemorySegment; import java.nio.ByteBuffer; +import java.nio.ByteOrder; import java.nio.channels.FileChannel; import java.nio.file.Files; import java.nio.file.Path; @@ -307,7 +307,7 @@ public void truncateAfter(long generation) { } public Stream replay(long afterGeneration) { - Iterator frames = new FrameReader(startFor(afterGeneration), end); + Iterator frames = new FrameReader(startFor(afterGeneration), end, afterGeneration); return StreamSupport.stream(Spliterators.spliteratorUnknownSize(frames, Spliterator.ORDERED | Spliterator.NONNULL), false) .filter(frame -> frame.generation() > afterGeneration) .map(frame -> codec.decode(frame.payload())); @@ -351,13 +351,20 @@ private record Frame(long start, long next, long generation, byte[] payload) { } private final class FrameReader implements Iterator { + private final ByteBuffer header = ByteBuffer.allocate(FRAME_HEADER).order(ByteOrder.LITTLE_ENDIAN); private final long limit; + private final long skipThrough; private long position; private Frame next; FrameReader(long from, long limit) { + this(from, limit, Long.MIN_VALUE); + } + + FrameReader(long from, long limit, long skipThrough) { this.position = from; this.limit = limit; + this.skipThrough = skipThrough; } @Override @@ -389,32 +396,37 @@ private Frame read() { } private Frame readLocked() { - Map.Entry entry = segments.floorEntry(position); - if (entry == null || !entry.getValue().file.isOpen()) { - return null; - } - FileHandle file = entry.getValue().file; - long local = position - entry.getKey(); try { - ByteBuffer header = ByteBuffer.allocate(FRAME_HEADER); - if (!readFully(file, header, local)) { - return null; - } - ByteCursor in = ByteCursor.over(MemorySegment.ofBuffer(header.flip())); - int length = in.getInt(); - int crc = in.getInt(); - long generation = in.getLong(); - if (length < 0 || position + FRAME_HEADER + length > limit) { - return null; - } - ByteBuffer body = ByteBuffer.allocate(length); - if (!readFully(file, body, local + FRAME_HEADER)) { - return null; + while (position + FRAME_HEADER <= limit) { + Map.Entry entry = segments.floorEntry(position); + if (entry == null || !entry.getValue().file.isOpen()) { + return null; + } + FileHandle file = entry.getValue().file; + long local = position - entry.getKey(); + if (!readFully(file, header.clear(), local)) { + return null; + } + int length = header.getInt(0); + int crc = header.getInt(Integer.BYTES); + long generation = header.getLong(2 * Integer.BYTES); + if (length < 0 || position + FRAME_HEADER + length > limit) { + return null; + } + if (generation <= skipThrough) { + position += FRAME_HEADER + length; + continue; + } + ByteBuffer body = ByteBuffer.allocate(length); + if (!readFully(file, body, local + FRAME_HEADER)) { + return null; + } + byte[] payload = body.array(); + return Checksums.crc32c(payload) == crc && carries(payload, generation) + ? new Frame(position, position + FRAME_HEADER + length, generation, payload) + : null; } - byte[] payload = body.array(); - return Checksums.crc32c(payload) == crc && carries(payload, generation) - ? new Frame(position, position + FRAME_HEADER + length, generation, payload) - : null; + return null; } catch (IOException e) { throw new UncheckedIOException(e); } diff --git a/engine/src/main/java/io/hstore/engine/topology/IncidenceCodec.java b/engine/src/main/java/io/hstore/engine/topology/IncidenceCodec.java index 075a91dd..575a1513 100644 --- a/engine/src/main/java/io/hstore/engine/topology/IncidenceCodec.java +++ b/engine/src/main/java/io/hstore/engine/topology/IncidenceCodec.java @@ -79,8 +79,11 @@ public int maxSize(Incidence incidence) { public void encode(ByteCursor out, long[] keys, List values) { int flags = 0; for (Column column : COLUMNS) { - if (values.stream().anyMatch(column.present)) { - flags |= 1 << column.ordinal(); + for (int i = 0; i < values.size(); i++) { + if (column.present.test(values.get(i))) { + flags |= 1 << column.ordinal(); + break; + } } } out.putByte(flags); diff --git a/engine/src/main/java/io/hstore/engine/tree/BulkBuilder.java b/engine/src/main/java/io/hstore/engine/tree/BulkBuilder.java index 01793bb0..5c664b86 100644 --- a/engine/src/main/java/io/hstore/engine/tree/BulkBuilder.java +++ b/engine/src/main/java/io/hstore/engine/tree/BulkBuilder.java @@ -14,6 +14,7 @@ public final class BulkBuilder { private static final int SPILL_BATCH = 32; + private static final int INITIAL_CAPACITY = 8; private final TreeSchema schema; private final NodeSource source; @@ -21,8 +22,8 @@ public final class BulkBuilder { private final Layout layout; private final List leaves = new ArrayList<>(); private BinaryOperator merge = (_, latest) -> latest; - private long[] keys = new long[64]; - private Object[] values = new Object[64]; + private long[] keys = new long[0]; + private Object[] values = new Object[0]; private int size; private int bytes; private int unspilled; @@ -58,8 +59,9 @@ public BulkBuilder add(long key, V value) { } entryBytes += size == 0 ? ByteCursor.signedVarLongSize(key) : ByteCursor.varLongSize(key - keys[size - 1]); if (size == keys.length) { - keys = Arrays.copyOf(keys, size * 2); - values = Arrays.copyOf(values, size * 2); + int capacity = Math.max(INITIAL_CAPACITY, size * 2); + keys = Arrays.copyOf(keys, capacity); + values = Arrays.copyOf(values, capacity); } keys[size] = key; values[size] = value; diff --git a/engine/src/main/java/io/hstore/engine/tree/TreeDiff.java b/engine/src/main/java/io/hstore/engine/tree/TreeDiff.java index 44b37e18..c85d0644 100644 --- a/engine/src/main/java/io/hstore/engine/tree/TreeDiff.java +++ b/engine/src/main/java/io/hstore/engine/tree/TreeDiff.java @@ -88,7 +88,21 @@ private sealed interface Item { record Subtree(Ref ref) implements Item { } - record Element(long key, Object value) implements Item { + final class Run implements Item { + private final Leaf leaf; + private int index; + + Run(Leaf leaf) { + this.leaf = leaf; + } + + long key() { + return leaf.key(index); + } + + Object value() { + return leaf.value(index); + } } } @@ -139,23 +153,21 @@ private Delta step() { } return switch (a) { case Item.Subtree(Ref ra) when b instanceof Item.Subtree(Ref rb) -> compareSubtrees(ra, rb); - case Item.Subtree(Ref ra) when b instanceof Item.Element(long key, Object value) -> { - if (ra.summary().min() > key) { - right.pop(); - yield new Delta.Added<>(key, after.schema().cast(value)); + case Item.Subtree(Ref ra) when b instanceof Item.Run run -> { + if (ra.summary().min() > run.key()) { + yield new Delta.Added<>(run.key(), after.schema().cast(take(right, run))); } expand(left, before); yield null; } - case Item.Element(long key, Object value) when b instanceof Item.Subtree(Ref rb) -> { - if (rb.summary().min() > key) { - left.pop(); - yield new Delta.Removed<>(key, before.schema().cast(value)); + case Item.Run run when b instanceof Item.Subtree(Ref rb) -> { + if (rb.summary().min() > run.key()) { + yield new Delta.Removed<>(run.key(), before.schema().cast(take(left, run))); } expand(right, after); yield null; } - case Item.Element(long ka, Object va) when b instanceof Item.Element(long kb, Object vb) -> compareElements(ka, va, kb, vb); + case Item.Run ra when b instanceof Item.Run rb -> compareElements(ra, rb); default -> throw new IllegalStateException(); }; } @@ -183,17 +195,17 @@ private Delta compareSubtrees(Ref ra, Ref rb) { return null; } - private Delta compareElements(long ka, Object va, long kb, Object vb) { + private Delta compareElements(Item.Run ra, Item.Run rb) { + long ka = ra.key(); + long kb = rb.key(); if (ka < kb) { - left.pop(); - return new Delta.Removed<>(ka, before.schema().cast(va)); + return new Delta.Removed<>(ka, before.schema().cast(take(left, ra))); } if (kb < ka) { - right.pop(); - return new Delta.Added<>(kb, after.schema().cast(vb)); + return new Delta.Added<>(kb, after.schema().cast(take(right, rb))); } - left.pop(); - right.pop(); + Object va = take(left, ra); + Object vb = take(right, rb); return Objects.equals(va, vb) ? null : new Delta.Changed<>(ka, before.schema().cast(va), after.schema().cast(vb)); } @@ -201,20 +213,28 @@ private Delta drain(Deque side, Tree tree, boolean removed) { while (side.peek() instanceof Item.Subtree) { expand(side, tree); } - if (side.isEmpty()) { + if (!(side.peek() instanceof Item.Run run)) { return null; } - Item.Element element = (Item.Element) side.pop(); - V value = tree.schema().cast(element.value()); - return removed ? new Delta.Removed<>(element.key(), value) : new Delta.Added<>(element.key(), value); + long key = run.key(); + V value = tree.schema().cast(take(side, run)); + return removed ? new Delta.Removed<>(key, value) : new Delta.Added<>(key, value); + } + + private static Object take(Deque side, Item.Run run) { + Object value = run.value(); + if (++run.index == run.leaf.size()) { + side.pop(); + } + return value; } private void expand(Deque side, Tree tree) { Item.Subtree subtree = (Item.Subtree) side.pop(); switch (tree.node(subtree.ref())) { case Leaf leaf -> { - for (int i = leaf.size() - 1; i >= 0; i--) { - side.push(new Item.Element(leaf.key(i), leaf.value(i))); + if (leaf.size() > 0) { + side.push(new Item.Run(leaf)); } } case Branch branch -> { diff --git a/engine/src/test/java/io/hstore/engine/feed/ChangeFeedTest.java b/engine/src/test/java/io/hstore/engine/feed/ChangeFeedTest.java new file mode 100644 index 00000000..3f714d74 --- /dev/null +++ b/engine/src/test/java/io/hstore/engine/feed/ChangeFeedTest.java @@ -0,0 +1,57 @@ +// SPDX-FileCopyrightText: Metacog Labs +// SPDX-License-Identifier: PolyForm-Noncommercial-1.0.0 + +package io.hstore.engine.feed; + +import io.hstore.engine.catalog.SlotRegistry; + +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.io.TempDir; + +import java.io.IOException; +import java.io.RandomAccessFile; +import java.nio.ByteBuffer; +import java.nio.ByteOrder; +import java.nio.file.Files; +import java.nio.file.Path; +import java.util.List; +import java.util.stream.Stream; + +import static org.junit.jupiter.api.Assertions.assertEquals; + +class ChangeFeedTest { + + @TempDir + Path directory; + + @Test + void replayReadsOnlyTheHeaderOfAFrameItSkips() throws IOException { + FeedCodec codec = new FeedCodec(new SlotRegistry(List.of())); + try (ChangeFeed feed = ChangeFeed.open(directory, codec)) { + for (long generation = 1; generation <= 5; generation++) { + feed.append(new CommitEvent(generation, generation, generation * 1000, 0, List.of(), List.of())); + } + Path file; + try (Stream files = Files.list(directory)) { + file = files.filter(path -> path.toString().endsWith(".feed")).findFirst().orElseThrow(); + } + try (RandomAccessFile raw = new RandomAccessFile(file.toFile(), "rw")) { + long second = 16 + frameLength(raw, 0); + long payload = second + 16; + raw.seek(payload); + int value = raw.read(); + raw.seek(payload); + raw.write(value ^ 0xFF); + } + assertEquals(List.of(3L, 4L, 5L), feed.replay(2).map(CommitEvent::generation).toList()); + assertEquals(List.of(1L), feed.replay(0).map(CommitEvent::generation).toList()); + } + } + + private static int frameLength(RandomAccessFile raw, long at) throws IOException { + byte[] header = new byte[4]; + raw.seek(at); + raw.readFully(header); + return ByteBuffer.wrap(header).order(ByteOrder.LITTLE_ENDIAN).getInt(); + } +} 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 52dbcfcf..69957df8 100644 --- a/engine/src/test/java/io/hstore/engine/tree/TreeTest.java +++ b/engine/src/test/java/io/hstore/engine/tree/TreeTest.java @@ -230,6 +230,58 @@ void diffReportsExactDeltasAndSkipsSharedPages() { assertTrue(pages.loads < 40, "diff visited " + pages.loads + " pages"); } + @ParameterizedTest + @ValueSource(longs = {1, 2, 3, 4, 5}) + void diffMatchesAnOracleAcrossRandomEdits(long seed) { + RandomGenerator random = random(seed); + MemoryPages pages = new MemoryPages(1024); + TreeMap first = new TreeMap<>(); + for (int i = random.nextInt(3000); i > 0; i--) { + long key = random.nextLong(5000); + first.put(key, key); + } + Tree before = Tree.empty(LONGS, pages); + WriteScope scope = new WriteScope(); + for (Map.Entry entry : first.entrySet()) { + before = before.put(scope, entry.getKey(), entry.getValue()); + } + before = pages.persist(before); + TreeMap second = new TreeMap<>(first); + Tree after = before; + WriteScope edits = new WriteScope(); + for (int i = random.nextInt(400); i > 0; i--) { + long key = random.nextLong(6000); + if (random.nextBoolean() && !second.isEmpty()) { + long existing = second.ceilingKey(key) != null ? second.ceilingKey(key) : second.firstKey(); + second.remove(existing); + after = after.remove(edits, existing); + } else { + long value = random.nextLong(3); + second.put(key, value); + after = after.put(edits, key, value); + } + } + List> expected = new ArrayList<>(); + TreeSet keys = new TreeSet<>(first.keySet()); + keys.addAll(second.keySet()); + for (long key : keys) { + Long was = first.get(key); + Long now = second.get(key); + if (was == null) { + expected.add(new TreeDiff.Delta.Added<>(key, now)); + } else if (now == null) { + expected.add(new TreeDiff.Delta.Removed<>(key, was)); + } else if (!was.equals(now)) { + expected.add(new TreeDiff.Delta.Changed<>(key, was, now)); + } + } + assertEquals(expected, TreeDiff.diff(before, after).toList()); + assertEquals(expected.size(), TreeDiff.diff(after, before).count()); + assertEquals(List.of(), TreeDiff.diff(after, after).toList()); + assertEquals(second.size(), TreeDiff.diff(Tree.empty(LONGS, pages), after).count()); + assertEquals(second.size(), TreeDiff.diff(after, Tree.empty(LONGS, pages)).count()); + } + @Test void keysetPaginationIsStable() { MemoryPages pages = new MemoryPages(1024);