From 0d47c4c2cb3a7d3e6c03085712053e6d8ee33240 Mon Sep 17 00:00:00 2001 From: venkat1701 Date: Sun, 25 Oct 2026 21:44:46 +0530 Subject: [PATCH 1/3] fix(txn): validate predicate reads under SERIALIZABLE SERIALIZABLE only recorded reads by atom id or slot key, so reads that went through a posting list or an index range were never validated. Write skew through an indexed predicate and phantoms through incidence or type scans committed on both sides. Scan-style reads are now recorded at the granularity they read: an atom's reverse-index postings, a type's postings, a canonical-key hash bucket, the key range of a property-index scan, and whole slots for unbounded scans (promoted to after 8192 reads in one slot). They are validated by content at commit, so checkpoints do not cause conflicts. Reads that only steer a plan (planner estimates, the single-tenant shortcut, the eager quota check whose merge re-checks at commit, HORA domain estimates) run in an explicit unrecorded scope, which also stops every pair of inserting SERIALIZABLE transactions from conflicting on the tenant usage counter. --- .../src/main/java/io/hstore/db/Reader.java | 9 +- .../io/hstore/db/hora/PatternMatcher.java | 6 +- .../io/hstore/db/property/PropertyIndex.java | 7 +- .../main/java/io/hstore/db/query/Planner.java | 14 +- .../java/io/hstore/db/security/Security.java | 5 +- .../io/hstore/db/query/SerializableTest.java | 257 +++++++++++++ docs/database/hql.md | 14 +- docs/transactions/transactions.md | 61 ++- .../java/io/hstore/engine/tree/TreeDiff.java | 80 +++- .../io/hstore/engine/txn/Transaction.java | 155 +++++++- .../main/java/io/hstore/engine/txn/View.java | 53 ++- .../main/java/io/hstore/engine/txn/Views.java | 6 +- .../engine/txn/SerializableReadsTest.java | 363 ++++++++++++++++++ 13 files changed, 992 insertions(+), 38 deletions(-) create mode 100644 database/src/test/java/io/hstore/db/query/SerializableTest.java create mode 100644 engine/src/test/java/io/hstore/engine/txn/SerializableReadsTest.java diff --git a/database/src/main/java/io/hstore/db/Reader.java b/database/src/main/java/io/hstore/db/Reader.java index 7540cda4..ce814d89 100644 --- a/database/src/main/java/io/hstore/db/Reader.java +++ b/database/src/main/java/io/hstore/db/Reader.java @@ -62,8 +62,10 @@ public int tenant() { private boolean onlyDefaultTenant() { Boolean known = onlyDefaultTenant; if (known == null) { - known = tenant() == Principal.DEFAULT_TENANT - && view.scan(EngineSlots.TYPES).last().map(entry -> entry.key() >>> 32 == Principal.DEFAULT_TENANT).orElse(true); + try (View.Unrecorded _ = view.unrecorded()) { + known = tenant() == Principal.DEFAULT_TENANT + && view.scan(EngineSlots.TYPES).last().map(entry -> entry.key() >>> 32 == Principal.DEFAULT_TENANT).orElse(true); + } onlyDefaultTenant = known; } return known; @@ -287,8 +289,7 @@ public LongStream range(String type, String property, Optional low, Optio Optional indexedTag = indexedTag(definition, property); if (indexedTag.isPresent()) { TypeTag tag = indexedTag.get(); - return PropertyIndex.candidates(view.scan(PropertySlots.PROPERTY_INDEX), tenant(), key.getAsInt(), tag, - PropertyIndex.lowerKey(tag, low), PropertyIndex.upperKey(tag, high)) + return PropertyIndex.candidates(view, tenant(), key.getAsInt(), tag, PropertyIndex.lowerKey(tag, low), PropertyIndex.upperKey(tag, high)) .filter(entry -> accept.test(entry.value())) .mapToLong(Entry::key) .filter(owner -> view.atom(owner).map(record -> record.type() == definition.id() && record.tenant() == tenant()).orElse(false)) diff --git a/database/src/main/java/io/hstore/db/hora/PatternMatcher.java b/database/src/main/java/io/hstore/db/hora/PatternMatcher.java index 57f3dfe9..95895eb4 100644 --- a/database/src/main/java/io/hstore/db/hora/PatternMatcher.java +++ b/database/src/main/java/io/hstore/db/hora/PatternMatcher.java @@ -12,6 +12,7 @@ import io.hstore.engine.tree.Tree; import io.hstore.engine.tree.TreeAlgebra; import io.hstore.engine.txn.IncidentEdge; +import io.hstore.engine.txn.View; import java.util.ArrayList; import java.util.Comparator; @@ -94,7 +95,10 @@ private long estimate(Pattern pattern, String name, Set bound) { } private long typeDomain(Variable variable) { - return variable.type().flatMap(reader::type).map(type -> reader.countOfType(type.id())).orElse(UNKNOWN_DOMAIN); + Optional type = variable.type().flatMap(reader::type); + try (View.Unrecorded _ = reader.view().unrecorded()) { + return type.map(found -> reader.countOfType(found.id())).orElse(UNKNOWN_DOMAIN); + } } private LongStream candidates(Pattern pattern, String name, Map binding) { diff --git a/database/src/main/java/io/hstore/db/property/PropertyIndex.java b/database/src/main/java/io/hstore/db/property/PropertyIndex.java index 2a5717b3..84701683 100644 --- a/database/src/main/java/io/hstore/db/property/PropertyIndex.java +++ b/database/src/main/java/io/hstore/db/property/PropertyIndex.java @@ -11,6 +11,7 @@ import io.hstore.engine.tree.Entry; import io.hstore.engine.tree.Ref; import io.hstore.engine.tree.Tree; +import io.hstore.engine.txn.View; import io.hstore.engine.txn.Workspace; import java.util.Optional; @@ -53,8 +54,10 @@ public static Tree> values(Tree directory, int tenant return new Tree<>(PropertySlots.postingSchema(), directory.source(), root); } - public static Stream> candidates(Tree directory, int tenant, int propertyKey, TypeTag tag, long lowKey, long highKey) { - Tree> values = values(directory, tenant, propertyKey, tag); + public static Stream> candidates(View view, int tenant, int propertyKey, TypeTag tag, long lowKey, long highKey) { + Ref root = view.nested(PropertySlots.PROPERTY_INDEX, PropertySlots.directoryKey(tenant, propertyKey, tag), lowKey, highKey) + .map(IndexRoot::root).orElse(Ref.EMPTY); + Tree> values = new Tree<>(PropertySlots.postingSchema(), view.source(), root); return values.range(lowKey, highKey) .flatMap(bucket -> PropertySlots.VALUE_POSTINGS.stream(values, bucket.key())); } diff --git a/database/src/main/java/io/hstore/db/query/Planner.java b/database/src/main/java/io/hstore/db/query/Planner.java index cae87e10..4ea56cab 100644 --- a/database/src/main/java/io/hstore/db/query/Planner.java +++ b/database/src/main/java/io/hstore/db/query/Planner.java @@ -13,6 +13,7 @@ import io.hstore.db.value.TypeTag; import io.hstore.db.value.Value; import io.hstore.engine.HStoreException; +import io.hstore.engine.txn.View; import java.util.ArrayList; import java.util.Comparator; @@ -75,9 +76,12 @@ Plan plan(Ast.Match match) { }); List conjuncts = match.where().map(Planner::conjuncts).orElse(List.of()); List candidates = new ArrayList<>(); - universe = reader.population(match.edges()); + double population; + try (View.Unrecorded _ = reader.view().unrecorded()) { + universe = reader.population(match.edges()); + population = type.map(found -> (double) reader.countOfType(found.id())).orElse(universe); + } double catalogSize = universe; - double population = type.map(found -> (double) reader.countOfType(found.id())).orElse(catalogSize); Candidate catalogScan = candidate(new AccessPath.AllAtoms(match.edges()), List.of(), catalogSize, conjuncts); candidates.add(new Candidate(catalogScan.path(), List.of(), catalogScan.rows(), catalogScan.cost() + VERIFY_WEIGHT * catalogScan.rows())); type.ifPresent(found -> candidates.add(candidate(new AccessPath.TypeScan(found), List.of(), population, conjuncts))); @@ -157,6 +161,12 @@ private double scanCost(AccessPath path, double rows) { } private double estimate(AccessPath path, double population) { + try (View.Unrecorded _ = reader.view().unrecorded()) { + return unrecordedEstimate(path, population); + } + } + + private double unrecordedEstimate(AccessPath path, double population) { return switch (path) { case AccessPath.Incident(List atoms) -> { double[] degrees = atoms.stream().mapToDouble(reader::degree).sorted().toArray(); diff --git a/database/src/main/java/io/hstore/db/security/Security.java b/database/src/main/java/io/hstore/db/security/Security.java index 5ed547d3..935d1296 100644 --- a/database/src/main/java/io/hstore/db/security/Security.java +++ b/database/src/main/java/io/hstore/db/security/Security.java @@ -123,7 +123,10 @@ public static Usage usage(View view, int tenant) { public static void charge(Transaction txn, int tenant, Usage delta) { Quota quota = quota(txn, tenant); - Usage projected = usage(txn, tenant).plus(delta); + Usage projected; + try (View.Unrecorded _ = txn.unrecorded()) { + projected = usage(txn, tenant).plus(delta); + } boolean growing = delta.atoms() > 0 || delta.edges() > 0 || delta.payloadBytes() > 0; if (growing && !quota.admits(projected)) { throw HStoreException.limit("tenant #" + tenant + " would exceed its quota: usage " + projected + " against " + quota); diff --git a/database/src/test/java/io/hstore/db/query/SerializableTest.java b/database/src/test/java/io/hstore/db/query/SerializableTest.java new file mode 100644 index 00000000..d3b4cce8 --- /dev/null +++ b/database/src/test/java/io/hstore/db/query/SerializableTest.java @@ -0,0 +1,257 @@ +// SPDX-FileCopyrightText: MetaCognition India Pvt Limited +// SPDX-License-Identifier: PolyForm-Noncommercial-1.0.0 + +package io.hstore.db.query; + +import io.hstore.db.DatabaseOptions; +import io.hstore.db.HypergraphDatabase; +import io.hstore.engine.EngineOptions; +import io.hstore.engine.HStoreException; + +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.io.TempDir; + +import java.nio.file.Path; +import java.util.ArrayList; +import java.util.List; +import java.util.Optional; + +import static org.junit.jupiter.api.Assertions.assertDoesNotThrow; +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertThrows; +import static org.junit.jupiter.api.Assertions.assertTrue; + +class SerializableTest { + + private static final String ON_CALL = "MATCH NODE n:Account WHERE n.oncall = true RETURN count(*)"; + private static final String BOOKED_R1 = "MATCH EDGE e:Booking WHERE e HAS @Room:'r1' RETURN count(*)"; + + @TempDir + Path directory; + + HypergraphDatabase database; + Session admin; + Session a; + Session b; + + @BeforeEach + void open() { + database = HypergraphDatabase.open(directory, DatabaseOptions.defaults().withEngine(EngineOptions.defaults().withPageSize(4096))); + admin = new Session(database); + admin.executeAll(""" + CREATE NODE TYPE Account (balance INT, oncall BOOL INDEXED); + CREATE NODE TYPE Room; + CREATE NODE TYPE Note; + CREATE SET EDGE TYPE Booking ROLES (room, guest); + INSERT NODE Account 'x' {balance: 100, oncall: true}; + INSERT NODE Account 'y' {balance: 100, oncall: true}; + INSERT NODE Room 'r1'; + INSERT NODE Room 'r2'; + INSERT EDGE Booking MEMBERS (@Account:'x' AS guest) AS $bx; + INSERT EDGE Booking MEMBERS (@Account:'y' AS guest) AS $by; + """); + a = new Session(database); + b = new Session(database); + } + + @AfterEach + void close() { + a.close(); + b.close(); + admin.close(); + database.close(); + } + + private static String cell(QueryResult result) { + return result.rows().getFirst().getFirst().render(); + } + + private String count(String query) { + return cell(admin.execute(query)); + } + + private void begin(boolean serializable) { + a.execute(serializable ? "BEGIN SERIALIZABLE" : "BEGIN"); + b.execute(serializable ? "BEGIN SERIALIZABLE" : "BEGIN"); + } + + private List> commitBoth() { + List> outcomes = new ArrayList<>(); + for (Session session : List.of(a, b)) { + try { + session.execute("COMMIT"); + outcomes.add(Optional.empty()); + } catch (HStoreException.Conflict conflict) { + assertTrue(conflict.retryable()); + assertEquals(HStoreException.Code.RETRYABLE_CONFLICT, conflict.code()); + outcomes.add(Optional.of(conflict)); + } + } + return outcomes; + } + + private static String onlySecondFails(List> outcomes) { + assertTrue(outcomes.get(0).isEmpty(), "the first commit should succeed"); + assertTrue(outcomes.get(1).isPresent(), "the second commit should fail with a retryable conflict"); + return outcomes.get(1).get().getMessage(); + } + + private static void bothCommit(List> outcomes) { + assertEquals(List.of(Optional.empty(), Optional.empty()), outcomes); + } + + @Test + void writeSkewThroughAnIndexedPredicateIsRejected() { + begin(true); + assertEquals("2", cell(a.execute(ON_CALL))); + assertEquals("2", cell(b.execute(ON_CALL))); + a.execute("SET @Account:'x'.oncall = false"); + b.execute("SET @Account:'y'.oncall = false"); + String message = onlySecondFails(commitBoth()); + assertTrue(message.contains("serializable read of property-index"), message); + assertEquals("1", count(ON_CALL)); + } + + @Test + void indexedPredicatesIgnoreChangesOutsideTheirRange() { + a.execute("BEGIN SERIALIZABLE"); + assertEquals("2", cell(a.execute(ON_CALL))); + a.execute("INSERT NODE Note 'checked'"); + admin.execute("INSERT NODE Account 'z' {balance: 5, oncall: false}"); + admin.execute("SET @Account:'x'.balance = 50"); + assertDoesNotThrow(() -> a.execute("COMMIT")); + } + + @Test + void snapshotAllowsWriteSkewThroughAnIndexedPredicate() { + begin(false); + assertEquals("2", cell(a.execute(ON_CALL))); + assertEquals("2", cell(b.execute(ON_CALL))); + a.execute("SET @Account:'x'.oncall = false"); + b.execute("SET @Account:'y'.oncall = false"); + bothCommit(commitBoth()); + assertEquals("0", count(ON_CALL)); + } + + @Test + void snapshotStillPreventsLostUpdates() { + begin(false); + long first = Long.parseLong(cell(a.execute("MATCH NODE n:Account WHERE n = @Account:'x' RETURN n.balance"))); + long second = Long.parseLong(cell(b.execute("MATCH NODE n:Account WHERE n = @Account:'x' RETURN n.balance"))); + a.execute("SET @Account:'x'.balance = " + (first + 10)); + b.execute("SET @Account:'x'.balance = " + (second + 20)); + onlySecondFails(commitBoth()); + assertEquals("110", count("MATCH NODE n:Account WHERE n = @Account:'x' RETURN n.balance")); + } + + @Test + void phantomThroughIncidenceIsRejected() { + long bx = admin.variables().get("bx"); + long by = admin.variables().get("by"); + begin(true); + assertEquals("0", cell(a.execute(BOOKED_R1))); + assertEquals("0", cell(b.execute(BOOKED_R1))); + a.execute("ADD @Room:'r1' TO @" + bx + " AS room"); + b.execute("ADD @Room:'r1' TO @" + by + " AS room"); + String message = onlySecondFails(commitBoth()); + assertTrue(message.contains("incident edges of atom"), message); + assertEquals("1", count(BOOKED_R1)); + } + + @Test + void countThenInsertThroughIncidenceIsRejected() { + begin(true); + assertEquals("0", cell(a.execute(BOOKED_R1))); + assertEquals("0", cell(b.execute(BOOKED_R1))); + a.execute("INSERT EDGE Booking MEMBERS (@Room:'r1' AS room, @Account:'x' AS guest)"); + b.execute("INSERT EDGE Booking MEMBERS (@Room:'r1' AS room, @Account:'y' AS guest)"); + onlySecondFails(commitBoth()); + assertEquals("1", count(BOOKED_R1)); + } + + @Test + void countThenInsertByTypeIsRejected() { + begin(true); + assertEquals("2", cell(a.execute("MATCH NODE n:Room RETURN count(*)"))); + assertEquals("2", cell(b.execute("MATCH NODE n:Room RETURN count(*)"))); + a.execute("INSERT NODE Room 'ra'"); + b.execute("INSERT NODE Room 'rb'"); + String message = onlySecondFails(commitBoth()); + assertTrue(message.contains("atoms of type"), message); + assertEquals("3", count("MATCH NODE n:Room RETURN count(*)")); + } + + @Test + void snapshotAllowsCountThenInsertPhantoms() { + begin(false); + assertEquals("2", cell(a.execute("MATCH NODE n:Room RETURN count(*)"))); + assertEquals("2", cell(b.execute("MATCH NODE n:Room RETURN count(*)"))); + a.execute("INSERT NODE Room 'ra'"); + b.execute("INSERT NODE Room 'rb'"); + bothCommit(commitBoth()); + assertEquals("4", count("MATCH NODE n:Room RETURN count(*)")); + } + + @Test + void absentCanonicalKeyReadsAreInvalidatedByTheirInsert() { + begin(true); + assertThrows(HStoreException.InvalidSchema.class, () -> a.execute("MATCH NODE n:Room WHERE n = @Room:'r4' RETURN count(*)")); + assertThrows(HStoreException.InvalidSchema.class, () -> b.execute("MATCH NODE n:Room WHERE n = @Room:'r3' RETURN count(*)")); + a.execute("INSERT NODE Room 'r3'"); + b.execute("INSERT NODE Room 'r4'"); + String message = onlySecondFails(commitBoth()); + assertTrue(message.contains("canonical key"), message); + assertEquals("3", count("MATCH NODE n:Room RETURN count(*)")); + } + + @Test + void insertsOfTypesNeitherTransactionScannedDoNotConflict() { + begin(true); + a.execute("INSERT NODE Room 'ra'"); + b.execute("INSERT NODE Account 'z' {balance: 5, oncall: false}"); + bothCommit(commitBoth()); + assertEquals("3", count("MATCH NODE n:Room RETURN count(*)")); + assertEquals("3", count("MATCH NODE n:Account RETURN count(*)")); + } + + @Test + void disjointPostingListsDoNotConflict() { + begin(true); + assertEquals("0", cell(a.execute(BOOKED_R1))); + assertEquals("0", cell(b.execute("MATCH EDGE e:Booking WHERE e HAS @Room:'r2' RETURN count(*)"))); + a.execute("INSERT EDGE Booking MEMBERS (@Room:'r1' AS room, @Account:'x' AS guest)"); + b.execute("INSERT EDGE Booking MEMBERS (@Room:'r2' AS room, @Account:'y' AS guest)"); + bothCommit(commitBoth()); + assertEquals("4", count("MATCH EDGE e:Booking RETURN count(*)")); + } + + @Test + void planEstimatesAreNotRecorded() { + long bx = admin.variables().get("bx"); + a.execute("BEGIN SERIALIZABLE"); + assertTrue(a.execute("EXPLAIN " + ON_CALL).message().contains("IndexRange")); + assertTrue(a.execute("EXPLAIN " + BOOKED_R1).message().contains("IncidentIntersection")); + String plan = a.execute("EXPLAIN MATCH NODE n:Account WHERE n = @Account:'x' AND n.oncall = true RETURN n.balance").message(); + assertTrue(plan.startsWith("Lookup") && plan.contains("rejected IndexRange"), plan); + assertEquals("100", cell(a.execute("MATCH NODE n:Account WHERE n = @Account:'x' AND n.oncall = true RETURN n.balance"))); + a.execute("SET @Account:'x'.balance = 7"); + admin.execute("SET @Account:'y'.oncall = false"); + admin.execute("ADD @Room:'r1' TO @" + bx + " AS room"); + admin.execute("INSERT NODE Account 'w' {balance: 1, oncall: true}"); + assertDoesNotThrow(() -> a.execute("COMMIT")); + assertEquals("7", count("MATCH NODE n:Account WHERE n = @Account:'x' RETURN n.balance")); + } + + @Test + void resultsThatUseTheEstimatedReadsAreStillRecorded() { + long bx = admin.variables().get("bx"); + a.execute("BEGIN SERIALIZABLE"); + assertEquals("0", cell(a.execute("MATCH NODE n:Room WHERE n = @Room:'r1' RETURN degree(n)"))); + a.execute("SET @Account:'x'.balance = 7"); + admin.execute("ADD @Room:'r1' TO @" + bx + " AS room"); + HStoreException.Conflict conflict = assertThrows(HStoreException.Conflict.class, () -> a.execute("COMMIT")); + assertTrue(conflict.getMessage().contains("incident edges of atom"), conflict.getMessage()); + } +} diff --git a/docs/database/hql.md b/docs/database/hql.md index d94f971c..6ee400fd 100644 --- a/docs/database/hql.md +++ b/docs/database/hql.md @@ -589,9 +589,17 @@ Session semantics (`Session`): - Without `BEGIN`, every statement runs in its own transaction (`HypergraphDatabase.write`, retried up to 8 times on conflict) or snapshot. -- `BEGIN` opens a session transaction with `SNAPSHOT` isolation (`SERIALIZABLE` adds read-set validation, see - [transactions](../transactions/transactions.md)); subsequent reads see its writes. `COMMIT` publishes it, - `ROLLBACK` aborts it; closing a session aborts an open transaction. +- `BEGIN` opens a session transaction with `SNAPSHOT` isolation; subsequent reads see its writes. `COMMIT` + publishes it, `ROLLBACK` aborts it; closing a session aborts an open transaction. Under `SNAPSHOT`, two + transactions can each read a predicate, write something the other's predicate depends on, and both commit + (write skew). +- `BEGIN SERIALIZABLE` also records what each statement read: the atoms and properties it looked at, the posting + lists behind `HAS`, `CONTAINS`, `degree()` and type scans, the property index range of an indexed `WHERE`, and + canonical key lookups, including ones that found nothing. `COMMIT` fails with a retryable conflict + (`RETRYABLE_CONFLICT`, "serializable read of ... was invalidated") if a transaction that committed in the + meantime changed any of it. The session transaction is closed either way; run it again from `BEGIN`. Planner + estimates and `EXPLAIN` are not recorded. See [transactions](../transactions/transactions.md#6-isolation) for the + exact granularity. - `USE BRANCH b` / `USE MAIN` / `USE TENANT t` are rejected inside a transaction. - `AT GENERATION g s` and `AS OF t s` run the single statement `s` against a retained historical generation (`Database.readAt`); writes inside are rejected (`historical generations are read-only`). `AS OF` resolves the diff --git a/docs/transactions/transactions.md b/docs/transactions/transactions.md index 0957f230..91cf52dd 100644 --- a/docs/transactions/transactions.md +++ b/docs/transactions/transactions.md @@ -187,14 +187,65 @@ The same walk adds up the heap those nodes take, which the commit adds to the wr `TxnOptions(branch, isolation, requestId, maxPages)`; defaults are `Branch.MAIN`, `Isolation.SNAPSHOT`, no request id, `Long.MAX_VALUE` pages. -**SNAPSHOT.** Reads come from the pinned base generation plus the transaction's own writes. Write-write conflicts are detected by replay (section 5.2). Read-write anomalies (write skew) are possible, as in any snapshot-isolation system. +**SNAPSHOT.** Reads come from the pinned base generation plus the transaction's own writes. Write-write conflicts are detected by replay (section 5.2). Read-write anomalies (write skew, and phantoms where a commit rests on a predicate that a concurrent commit changed) are possible, as in any snapshot-isolation system. -**SERIALIZABLE.** `View` calls `observeAtom` / `observeKey` on every read; under `SERIALIZABLE` the transaction records the value as of the **base** snapshot for each atom (`atomReads`) and each `(slot, key)` (`keyReads`) on first observation. On the slow path, `validateReads(latest)` re-reads each recorded key in the latest roots and raises a conflict if: +**SERIALIZABLE.** The transaction records what it read and, on the slow path, `validateReads(latest)` checks every recorded read against the latest roots under the commit lock. A read is valid when the latest committed state still gives the value the transaction saw in its base snapshot. If every read is valid the transaction can be placed at its commit point in a serial order, so the serial order is the commit order. Recording goes through hooks on `View` (`observeAtom`, `observeKey`, `observeRange`, `observeSlot`). They are empty methods on `Snapshot`, and `Transaction` returns from them at once unless the isolation is `SERIALIZABLE` and no unrecorded scope is open, so `SNAPSHOT` transactions and snapshots don't pay for them. -* a key read: `now.equals(seen)` is false; -* an atom read: presence differs, or the record differs **unless** both are `EdgeRecord`s with the same member root (`Ref.same(was.members(), is.members())`). Records compare inline member trees by content, so they count as equal when the members are. Membership changes made by replay are still validated through `verifyEdge`. +Each read is recorded once, on first observation, with the value from the **base** roots (`snapshotValue`), not the value the transaction saw through its own writes. Validation compares that base value with the latest committed value, so a transaction's own writes never invalidate its own reads; replay then applies its writes on top of the latest roots. -The fast path never needs read validation: it is only taken when every touched slot, including every read slot, is byte-for-byte the same root as at the base. Validation is therefore at key granularity when slots moved and free when they did not. +| Read on `View` | Recorded as | Valid at commit when | +|---|---|---| +| `atom`, `exists`, `edge`, `requireEdge`, `contains`, `incidence`, `cardinality` | atom read (`atomReads`) of the atom | presence is unchanged and the record is equal, or both are `EdgeRecord`s with the same member root (`Ref.same`) | +| `get(slot, key)` | key read (`keyReads`) of `(slot, key)` | the value has the same content (below) | +| `incident`, `forEachIncidentEdge`, `incidentTree`, `degree` | key read of `REVERSE` at the atom, which is the atom's whole incidence posting list | the posting list has the same content | +| `atomsOfType`, `typeTree`, `countOfType` | key read of `TYPES` at `typeKey(tenant, type)`, the posting list of that type in that tenant | the posting list has the same content | +| `resolve(tenant, type, key)` | key read of `CANONICAL` at `canonicalKey(tenant, type, key)` | the posting list of that hash is unchanged, so a lookup that found nothing fails once the key is inserted, and one that found an atom fails once it is deleted or renamed | +| `nested(slot, key, low, high)` | range read (`rangeReads`) | the trees nested in the value at `key` have the same entries between `low` and `high`; an absent value counts as empty nested trees | +| `scan(slot)`, `atoms()` | slot read (`slotReads`) of the whole slot | the slot has the same content | + +The database module reads the property index through `nested`: `Reader.range` passes the directory key `directoryKey(tenant, property, tag)` of `PROPERTY_INDEX` and the order-key range `[lowKey, highKey]` of the predicate. `MATCH NODE n:Account WHERE n.oncall = true` therefore depends on the `true` bucket of `oncall` only, and a commit that changes some other account's `oncall` from `true` to `false` invalidates it, while a commit that only adds `oncall = false` accounts does not. + +Same content means `TreeDiff.sameValue` for values and `TreeDiff.sameContent` for trees. Equal values are the same. Values that hold nested trees (`Postings.Promoted`, the property index's `IndexRoot`, edge records) are the same when their other fields are equal and each nested tree holds the same entries. Inline posting lists compare by their keys and values. Tree comparison walks `TreeDiff.diff`, which skips subtrees that are `Ref.same`, so it costs time proportional to the changed region, and a range read prunes subtrees whose summary lies outside the range. Because `Ref.same` treats a pending node as the same tree as the stored reference a published checkpoint wrote it as, a checkpoint that finishes between a read and the commit doesn't cause a conflict, for inline and promoted posting lists alike. + +The fast path never needs read validation: it is only taken when every touched slot, including every slot that was only read, has the same root as at the base. Validation is therefore at the granularity above when slots moved and free when they did not. + +**Granularity and false positives.** Validation never accepts a transaction whose recorded reads changed, but it can reject transactions that a finer check would accept. Such conflicts are retryable like any other. + +* Posting lists are validated whole. Any change to an atom's incidence list invalidates `incident(atom)`, including an edge that a role filter or tenant filter would have skipped; any new atom of a type invalidates a scan of that type, even if a residual predicate would have rejected it. +* Canonical keys are validated per hash bucket of `(tenant, type, key)`. +* `scan(slot)` and `atoms()` are validated per slot, so any change to the slot conflicts. Database code that scans whole slots (schema listings, tenant and user listings, materialized view data, signals, exact semantic search) is protected at that granularity. +* Promotion bounds the read set, in the spirit of predicate-lock promotion in PostgreSQL's serializable snapshot isolation: when a transaction records more than `Transaction.PROMOTE_AFTER` (8192) key and range reads in one slot, or that many atom reads, they are replaced by one read of the whole slot (`CATALOG` for atoms). +* Validation fixes the serial order to the commit order. A transaction that read something a concurrent transaction changed and committed first fails, even when the opposite order would have been serial, for example after reading that a key is absent while the other transaction inserts it. +* A value that a concurrent commit changed and changed back counts as unchanged. + +A read-only transaction commits with `READ_ONLY` and validates nothing. Its reads come from one committed generation, which is a prefix of the commit order, so it is serializable as it stands. + +**Unrecorded reads.** `View.unrecorded()` opens a scope in which the transaction records nothing: + +```java +try (View.Unrecorded _ = view.unrecorded()) { + estimate = reader.countOfType(type); +} +``` + +`Snapshot` and `SNAPSHOT` transactions return one shared no-op handle, so the call allocates nothing. A `SERIALIZABLE` transaction counts open scopes, so scopes nest, and returns a handle it allocated when it began. Reads inside a scope still touch their slots, so they can push a commit onto the slow path, but they never cause a conflict. The scope is for reads whose value cannot change a result. It is used in these places and nowhere else: + +| Place | Reads | Why it doesn't affect results | +|---|---|---| +| `Planner.plan` and `Planner.estimate` (database) | population, type counts, degrees, cardinalities, `PropertyIndex.estimate` | they only rank access paths; the chosen path reads its rows with recording on | +| `PatternMatcher.typeDomain` (database) | type counts | they only order the variables of a pattern | +| `Reader.onlyDefaultTenant` (database) | the last key of `TYPES` | it only decides whether tenant filtering can be skipped; any atom of another tenant that a reader can reach shows up in a posting list or record that the reader did record. Recorded, it would make every `SERIALIZABLE` transaction a reader of the whole `TYPES` slot | +| `Security.charge` (database) | the tenant's usage counter | it only fails early when a quota would be exceeded; the commutative `merge` on `USAGE` checks the quota again at commit. Recorded, it would make any two `SERIALIZABLE` transactions that insert in one tenant conflict | + +The same methods called anywhere else are recorded: an HQL `degree()` or `count(*)` over a type scan, or the access path a query runs, is validated. `Statistics` reads through its own `Snapshot`, so it records nothing. + +| Anomaly | `SNAPSHOT` | `SERIALIZABLE` | +|---|---|---| +| Dirty read | prevented: reads see committed generations and the transaction's own writes only | prevented | +| Non-repeatable read | prevented: every read in a transaction comes from the same pinned generation | prevented | +| Lost update | prevented for slot writes: `put`, `update` and `delete(slot, key)` replay only if the value is still the one the transaction saw, and HQL `SET` and `UNSET` use a `merge` whose precondition requires the property to be unchanged; edge actions follow section 5.2 | prevented | +| Write skew | allowed | prevented: one of the transactions fails at commit with `RETRYABLE_CONFLICT` | +| Phantom (a predicate result changed by a concurrent insert, update or member change) | allowed: repeated reads in the transaction see the same snapshot, but a commit based on a stale predicate result is accepted | prevented for predicates evaluated through the recorded reads above | Write skew rejected under `SERIALIZABLE` (this is `EngineTest.serializableReadsDetectWriteSkew`): 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 4544902a..c4333b00 100644 --- a/engine/src/main/java/io/hstore/engine/tree/TreeDiff.java +++ b/engine/src/main/java/io/hstore/engine/tree/TreeDiff.java @@ -32,7 +32,52 @@ private TreeDiff() { } public static Stream> diff(Tree before, Tree after) { - return Tree.stream(new Walker<>(before, after)).map(Entry::value); + return diff(before, after, Long.MIN_VALUE, Long.MAX_VALUE); + } + + public static Stream> diff(Tree before, Tree after, long low, long high) { + return Tree.stream(new Walker<>(before, after, low, high)).map(Entry::value); + } + + public static boolean sameContent(Tree a, Tree b) { + return sameContent(a, b, Long.MIN_VALUE, Long.MAX_VALUE); + } + + public static boolean sameContent(Tree a, Tree b, long low, long high) { + return Ref.same(a.root(), b.root()) || diff(a, b, low, high).allMatch(delta -> + delta instanceof Delta.Changed(long _, V before, V after) && sameValue(a.schema(), a.source(), before, after)); + } + + public static boolean sameValue(TreeSchema schema, NodeSource source, V a, V b) { + if (Objects.equals(a, b)) { + return true; + } + ValueCodec codec = schema.codec(); + if (!codec.holdsRefs() || !Objects.equals(codec.mapRefs(a, (_, _) -> Ref.EMPTY), codec.mapRefs(b, (_, _) -> Ref.EMPTY))) { + return false; + } + List> left = nested(source, codec, a); + List> right = nested(source, codec, b); + if (left.size() != right.size()) { + return false; + } + for (int i = 0; i < left.size(); i++) { + if (!left.get(i).schema().equals(right.get(i).schema()) || !sameNested(left.get(i), right.get(i), Long.MIN_VALUE, Long.MAX_VALUE)) { + return false; + } + } + return true; + } + + public static List> nested(NodeSource source, ValueCodec codec, V value) { + List> trees = new ArrayList<>(); + codec.forEachRef(value, (ref, schema) -> trees.add(new Tree<>(schema, source, ref))); + return trees; + } + + @SuppressWarnings("unchecked") + public static boolean sameNested(Tree a, Tree b, long low, long high) { + return sameContent(a, (Tree) b, low, high); } public static Optional difference(Tree a, Tree b) { @@ -109,13 +154,19 @@ Object value() { private static final class Walker implements Iterator>> { private final Tree before; private final Tree after; + private final long low; + private final long high; + private final boolean bounded; private final Deque left = new ArrayDeque<>(); private final Deque right = new ArrayDeque<>(); private Delta pending; - Walker(Tree before, Tree after) { + Walker(Tree before, Tree after, long low, long high) { this.before = before; this.after = after; + this.low = low; + this.high = high; + this.bounded = low != Long.MIN_VALUE || high != Long.MAX_VALUE; if (!before.isEmpty()) { left.push(new Item.Subtree(before.root())); } @@ -145,6 +196,14 @@ public Entry> next() { private Delta step() { Item a = left.peek(); Item b = right.peek(); + if (bounded && a != null && outside(a)) { + skip(left, a); + return null; + } + if (bounded && b != null && outside(b)) { + skip(right, b); + return null; + } if (a == null) { return drain(right, after, false); } @@ -212,6 +271,9 @@ private Delta compareElements(Item.Run ra, Item.Run rb) { private Delta drain(Deque side, Tree tree, boolean removed) { while (side.peek() instanceof Item.Subtree) { expand(side, tree); + if (bounded) { + return null; + } } if (!(side.peek() instanceof Item.Run run)) { return null; @@ -221,6 +283,20 @@ private Delta drain(Deque side, Tree tree, boolean removed) { return removed ? new Delta.Removed<>(key, value) : new Delta.Added<>(key, value); } + private boolean outside(Item item) { + return switch (item) { + case Item.Subtree(Ref ref) -> !ref.summary().overlaps(low, high); + case Item.Run run -> run.key() < low || run.key() > high; + }; + } + + private static void skip(Deque side, Item item) { + switch (item) { + case Item.Subtree _ -> side.pop(); + case Item.Run run -> take(side, run); + } + } + private static Object take(Deque side, Item.Run run) { Object value = run.value(); if (++run.index == run.leaf.size()) { diff --git a/engine/src/main/java/io/hstore/engine/txn/Transaction.java b/engine/src/main/java/io/hstore/engine/txn/Transaction.java index 114753e9..9d83b556 100644 --- a/engine/src/main/java/io/hstore/engine/txn/Transaction.java +++ b/engine/src/main/java/io/hstore/engine/txn/Transaction.java @@ -20,6 +20,7 @@ import io.hstore.engine.tree.Ref; import io.hstore.engine.tree.Tree; import io.hstore.engine.tree.TreeDiff; +import io.hstore.engine.tree.ValueCodec; import io.hstore.engine.tree.WriteScope; import java.util.ArrayList; @@ -49,6 +50,11 @@ Footprint touch(EdgeAction action) { private record KeyRead(Slot slot, long key) { } + private record RangeRead(Slot slot, long key, long low, long high) { + } + + static final int PROMOTE_AFTER = 8192; + private final TransactionManager manager; private final long id; private final Retained retained; @@ -61,6 +67,12 @@ private record KeyRead(Slot slot, long key) { private final Map footprints = new HashMap<>(); private final Map> atomReads = new HashMap<>(); private final Map> keyReads = new HashMap<>(); + private final Map> rangeReads = new HashMap<>(); + private final Map, Integer> readsPerSlot = new HashMap<>(); + private final Set> slotReads = new HashSet<>(); + private final boolean serializable; + private final Unrecorded resume; + private int unrecorded; private State state = State.ACTIVE; private long[] spilled = new long[0]; private int spilledPages; @@ -72,6 +84,8 @@ private record KeyRead(Slot slot, long key) { this.snapshot = snapshot; this.branch = snapshot.branch(options.branch()); this.options = options; + this.serializable = options.isolation() == Isolation.SERIALIZABLE; + this.resume = serializable ? () -> unrecorded = Math.max(0, unrecorded - 1) : NOT_RECORDING; this.scope = new WriteScope((ref, schema) -> manager.spill(this.id, ref, schema, this::spilled)); this.workspace = manager.workspace(scope, branch.roots(), Workspace.EdgeGuard.NONE, false); } @@ -113,18 +127,69 @@ void exposeLazily() { scope.freeze(); } + @Override + public Unrecorded unrecorded() { + if (serializable) { + unrecorded++; + } + return resume; + } + + private boolean recording() { + return serializable && unrecorded == 0; + } + @Override void observeAtom(long atom, Optional record) { - if (options.isolation() == Isolation.SERIALIZABLE && !atomReads.containsKey(atom)) { - atomReads.put(atom, snapshotValue(EngineSlots.CATALOG, atom)); + if (!recording() || slotReads.contains(EngineSlots.CATALOG) || atomReads.containsKey(atom)) { + return; + } + atomReads.put(atom, snapshotValue(EngineSlots.CATALOG, atom)); + if (atomReads.size() > PROMOTE_AFTER) { + observeSlot(EngineSlots.CATALOG); } } @Override - void observeKey(Slot slot, long key, Optional value) { + void observeKey(Slot slot, long key) { + if (!recording() || slotReads.contains(slot)) { + return; + } KeyRead read = new KeyRead(slot, key); - if (options.isolation() == Isolation.SERIALIZABLE && !keyReads.containsKey(read)) { + if (!keyReads.containsKey(read)) { keyReads.put(read, snapshotValue(slot, key)); + counted(slot); + } + } + + @Override + void observeRange(Slot slot, long key, long low, long high) { + if (!recording() || slotReads.contains(slot) || keyReads.containsKey(new KeyRead(slot, key))) { + return; + } + RangeRead read = new RangeRead(slot, key, low, high); + if (!rangeReads.containsKey(read)) { + rangeReads.put(read, snapshotValue(slot, key)); + counted(slot); + } + } + + @Override + void observeSlot(Slot slot) { + if (!recording() || !slotReads.add(slot)) { + return; + } + keyReads.keySet().removeIf(read -> read.slot().equals(slot)); + rangeReads.keySet().removeIf(read -> read.slot().equals(slot)); + readsPerSlot.remove(slot); + if (slot.equals(EngineSlots.CATALOG)) { + atomReads.clear(); + } + } + + private void counted(Slot slot) { + if (readsPerSlot.merge(slot, 1, Integer::sum) > PROMOTE_AFTER) { + observeSlot(slot); } } @@ -227,7 +292,11 @@ public void merge(Slot slot, long key, Predicate> preconditio } private Optional snapshotValue(Slot slot, long key) { - return new Tree<>(slot.schema(), source(), branch.roots().get(slot)).get(key); + return snapshotTree(slot).get(key); + } + + private Tree snapshotTree(Slot slot) { + return new Tree<>(slot.schema(), source(), branch.roots().get(slot)); } private void edgeAction(long edge, EdgeAction action) { @@ -361,9 +430,14 @@ private void verifyEdge(long edge, Workspace target) { } void validateReads(RootVector latest) { - if (options.isolation() != Isolation.SERIALIZABLE) { + if (!serializable) { return; } + slotReads.forEach(slot -> { + if (!sameSlot(slot, latest)) { + throw HStoreException.conflict("serializable scan of " + slot + " was invalidated"); + } + }); Tree catalog = new Tree<>(EngineSlots.ATOM_SCHEMA, source(), latest.get(EngineSlots.CATALOG)); atomReads.forEach((atom, seen) -> { Optional now = catalog.get(atom); @@ -375,10 +449,73 @@ void validateReads(RootVector latest) { } }); keyReads.forEach((read, seen) -> { - Optional now = new Tree<>(read.slot().schema(), source(), latest.get(read.slot())).get(read.key()); - if (!now.equals(seen)) { - throw HStoreException.conflict("serializable read of " + read.slot() + " key " + read.key() + " was invalidated"); + if (!sameValue(read.slot(), seen, latestValue(read.slot(), read.key(), latest))) { + throw HStoreException.conflict("serializable read of " + describe(read.slot(), read.key()) + " was invalidated"); + } + }); + rangeReads.forEach((read, seen) -> { + if (!sameRange(read, seen, latestValue(read.slot(), read.key(), latest))) { + throw HStoreException.conflict("serializable read of " + describe(read.slot(), read.key()) + + " between " + read.low() + " and " + read.high() + " was invalidated"); } }); } + + private boolean sameSlot(Slot slot, RootVector latest) { + return TreeDiff.sameContent(snapshotTree(slot), new Tree<>(slot.schema(), source(), latest.get(slot))); + } + + private Optional latestValue(Slot slot, long key, RootVector latest) { + return new Tree<>(slot.schema(), source(), latest.get(slot)).get(key); + } + + @SuppressWarnings("unchecked") + private boolean sameValue(Slot slot, Optional seen, Optional now) { + if (seen.isEmpty() || now.isEmpty()) { + return seen.isEmpty() && now.isEmpty(); + } + return TreeDiff.sameValue(slot.schema(), source(), (V) seen.get(), (V) now.get()); + } + + @SuppressWarnings("unchecked") + private boolean sameRange(RangeRead read, Optional seen, Optional now) { + Slot slot = (Slot) read.slot(); + ValueCodec codec = slot.schema().codec(); + if (seen.isEmpty() && now.isEmpty()) { + return true; + } + if (!codec.holdsRefs()) { + return sameValue(slot, seen, now); + } + if (seen.isPresent() && now.isPresent() + && !Objects.equals(codec.mapRefs((V) seen.get(), (_, _) -> Ref.EMPTY), codec.mapRefs((V) now.get(), (_, _) -> Ref.EMPTY))) { + return false; + } + List> before = seen.map(value -> TreeDiff.nested(source(), codec, (V) value)).orElse(List.of()); + List> after = now.map(value -> TreeDiff.nested(source(), codec, (V) value)).orElse(List.of()); + if (seen.isPresent() && now.isPresent() ? before.size() != after.size() : before.isEmpty() && after.isEmpty()) { + return false; + } + for (int i = 0; i < Math.max(before.size(), after.size()); i++) { + Tree was = before.isEmpty() ? after.get(i).withRoot(Ref.EMPTY) : before.get(i); + Tree is = after.isEmpty() ? before.get(i).withRoot(Ref.EMPTY) : after.get(i); + if (!TreeDiff.sameNested(was, is, read.low(), read.high())) { + return false; + } + } + return true; + } + + private static String describe(Slot slot, long key) { + if (slot.equals(EngineSlots.REVERSE)) { + return "the incident edges of atom " + key; + } + if (slot.equals(EngineSlots.TYPES)) { + return "the atoms of type " + (int) key + " in tenant " + (key >>> 32); + } + if (slot.equals(EngineSlots.CANONICAL)) { + return "canonical key hash " + Long.toHexString(key); + } + return slot + " key " + key; + } } diff --git a/engine/src/main/java/io/hstore/engine/txn/View.java b/engine/src/main/java/io/hstore/engine/txn/View.java index 8d7c0412..0d802d87 100644 --- a/engine/src/main/java/io/hstore/engine/txn/View.java +++ b/engine/src/main/java/io/hstore/engine/txn/View.java @@ -24,6 +24,14 @@ public abstract sealed class View permits Snapshot, Transaction { + public interface Unrecorded extends AutoCloseable { + @Override + void close(); + } + + static final Unrecorded NOT_RECORDING = () -> { + }; + abstract Tree tree(Slot slot); void exposeLazily() { @@ -37,7 +45,17 @@ private Tree lazyTree(Slot slot) { void observeAtom(long id, Optional record) { } - void observeKey(Slot slot, long key, Optional value) { + void observeKey(Slot slot, long key) { + } + + void observeRange(Slot slot, long key, long low, long high) { + } + + void observeSlot(Slot slot) { + } + + public Unrecorded unrecorded() { + return NOT_RECORDING; } public abstract long generation(); @@ -88,38 +106,51 @@ public long cardinality(long edge) { } public Stream incident(long atom) { + observeKey(EngineSlots.REVERSE, atom); return Views.incident(lazyTree(EngineSlots.REVERSE), atom); } public void forEachIncidentEdge(long atom, LongConsumer action) { + observeKey(EngineSlots.REVERSE, atom); EngineSlots.REVERSE_INDEX.forEachPosting(lazyTree(EngineSlots.REVERSE), atom, action); } public Tree incidentTree(long atom) { + observeKey(EngineSlots.REVERSE, atom); return EngineSlots.REVERSE_INDEX.tree(lazyTree(EngineSlots.REVERSE), atom); } public long degree(long atom) { + observeKey(EngineSlots.REVERSE, atom); return EngineSlots.REVERSE_INDEX.count(tree(EngineSlots.REVERSE), atom); } public OptionalLong resolve(int tenant, int type, String canonicalKey) { - return Views.resolve(tree(EngineSlots.CANONICAL), tenant, type, canonicalKey); + long hash = EngineSlots.canonicalKey(tenant, type, canonicalKey); + observeKey(EngineSlots.CANONICAL, hash); + return Views.resolve(tree(EngineSlots.CANONICAL), tenant, type, canonicalKey, hash); } public LongStream atomsOfType(int tenant, int type) { - return EngineSlots.TYPE_INDEX.stream(lazyTree(EngineSlots.TYPES), EngineSlots.typeKey(tenant, type)).mapToLong(Entry::key); + long key = EngineSlots.typeKey(tenant, type); + observeKey(EngineSlots.TYPES, key); + return EngineSlots.TYPE_INDEX.stream(lazyTree(EngineSlots.TYPES), key).mapToLong(Entry::key); } public Tree typeTree(int tenant, int type) { - return EngineSlots.TYPE_INDEX.tree(lazyTree(EngineSlots.TYPES), EngineSlots.typeKey(tenant, type)); + long key = EngineSlots.typeKey(tenant, type); + observeKey(EngineSlots.TYPES, key); + return EngineSlots.TYPE_INDEX.tree(lazyTree(EngineSlots.TYPES), key); } public long countOfType(int tenant, int type) { - return EngineSlots.TYPE_INDEX.count(tree(EngineSlots.TYPES), EngineSlots.typeKey(tenant, type)); + long key = EngineSlots.typeKey(tenant, type); + observeKey(EngineSlots.TYPES, key); + return EngineSlots.TYPE_INDEX.count(tree(EngineSlots.TYPES), key); } public Stream> atoms() { + observeSlot(EngineSlots.CATALOG); return lazyTree(EngineSlots.CATALOG).stream(); } @@ -128,12 +159,18 @@ public Optional symbol(int id) { } public Optional get(Slot slot, long key) { - Optional value = tree(slot).get(key); - observeKey(slot, key, value); - return value; + observeKey(slot, key); + return tree(slot).get(key); + } + + public Optional nested(Slot slot, long key, long low, long high) { + observeRange(slot, key, low, high); + exposeLazily(); + return tree(slot).get(key); } public Tree scan(Slot slot) { + observeSlot(slot); return lazyTree(slot); } diff --git a/engine/src/main/java/io/hstore/engine/txn/Views.java b/engine/src/main/java/io/hstore/engine/txn/Views.java index f828ae58..8af1d184 100644 --- a/engine/src/main/java/io/hstore/engine/txn/Views.java +++ b/engine/src/main/java/io/hstore/engine/txn/Views.java @@ -19,8 +19,12 @@ private Views() { } static OptionalLong resolve(Tree> canonical, int tenant, int type, String key) { + return resolve(canonical, tenant, type, key, EngineSlots.canonicalKey(tenant, type, key)); + } + + static OptionalLong resolve(Tree> canonical, int tenant, int type, String key, long hash) { CanonicalName wanted = new CanonicalName(tenant, type, key); - return EngineSlots.CANONICAL_INDEX.stream(canonical, EngineSlots.canonicalKey(tenant, type, key)) + return EngineSlots.CANONICAL_INDEX.stream(canonical, hash) .filter(entry -> entry.value().equals(wanted)) .mapToLong(Entry::key) .findFirst(); diff --git a/engine/src/test/java/io/hstore/engine/txn/SerializableReadsTest.java b/engine/src/test/java/io/hstore/engine/txn/SerializableReadsTest.java new file mode 100644 index 00000000..ef12208f --- /dev/null +++ b/engine/src/test/java/io/hstore/engine/txn/SerializableReadsTest.java @@ -0,0 +1,363 @@ +// SPDX-FileCopyrightText: MetaCognition India Pvt Limited +// SPDX-License-Identifier: PolyForm-Noncommercial-1.0.0 + +package io.hstore.engine.txn; + +import io.hstore.engine.EngineOptions; +import io.hstore.engine.HStoreException; +import io.hstore.engine.StorageEngine; +import io.hstore.engine.catalog.EngineSlots; +import io.hstore.engine.catalog.Slot; +import io.hstore.engine.index.Postings; +import io.hstore.engine.page.ByteCursor; +import io.hstore.engine.topology.EdgeKind; +import io.hstore.engine.tree.FingerprintMode; +import io.hstore.engine.tree.Hashing; +import io.hstore.engine.tree.TreeSchema; +import io.hstore.engine.tree.ValueCodec; + +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.io.TempDir; + +import java.nio.file.Path; +import java.util.LinkedHashMap; +import java.util.List; +import java.util.Map; +import java.util.function.Consumer; +import java.util.function.ObjLongConsumer; +import java.util.stream.LongStream; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertInstanceOf; +import static org.junit.jupiter.api.Assertions.assertSame; +import static org.junit.jupiter.api.Assertions.assertThrows; +import static org.junit.jupiter.api.Assertions.assertTrue; + +class SerializableReadsTest { + + private static final int NODE = 1; + private static final int EDGE = 2; + private static final int OTHER = 3; + private static final int FEW = 4; + private static final TxnOptions SERIALIZABLE = TxnOptions.defaults().withIsolation(Isolation.SERIALIZABLE); + private static final Slot OBSERVED = Slot.primary(120, "observed", counters(240)); + private static final Slot WRITTEN = Slot.primary(121, "written", counters(241)); + + @TempDir + Path directory; + + private int names; + + private static TreeSchema counters(int id) { + return new TreeSchema<>(id, "counters-" + id, FingerprintMode.SET, + ValueCodec.rows(ByteCursor::signedVarLongSize, ByteCursor::putSignedVarLong, ByteCursor::getSignedVarLong), + (key, value, into) -> into.entry(key, Hashing.of(key, value))); + } + + private StorageEngine open() { + return StorageEngine.open(directory, EngineOptions.defaults().withPageSize(1024).withPagesPerSegment(256).withHistoryLimit(1000) + .withExtensions(List.of(OBSERVED, WRITTEN), List.of())); + } + + private String name() { + return "n" + names++; + } + + private long node(StorageEngine engine, int type) { + String key = name(); + return engine.write(txn -> txn.createNode(type, key)); + } + + private long joinNewEdge(StorageEngine engine, long member) { + return engine.write(txn -> { + long edge = txn.createEdge(EDGE, EdgeKind.SET); + txn.insert(edge, member); + return edge; + }); + } + + private void unrelatedCommit(StorageEngine engine) { + long key = names++; + engine.write(txn -> { + txn.put(WRITTEN, 1_000_000 + key, key); + long edge = txn.createEdge(EDGE, EdgeKind.SET); + txn.insert(edge, txn.createNode(OTHER, "unrelated-" + key)); + return null; + }); + } + + private void ownWrite(Transaction txn) { + txn.put(WRITTEN, names++, 1L); + } + + private CommitResult.Outcome commitAfterCheckpoint(StorageEngine engine, Transaction txn) { + engine.checkpoint(); + return txn.commit().outcome(); + } + + private void assertConflict(StorageEngine engine, Transaction txn, String expected) { + engine.checkpoint(); + HStoreException.Conflict conflict = assertThrows(HStoreException.Conflict.class, txn::commit); + assertTrue(conflict.retryable()); + assertTrue(conflict.getMessage().contains(expected), conflict.getMessage()); + } + + private static Map> incidenceReads() { + Map> reads = new LinkedHashMap<>(); + reads.put("incident", (txn, atom) -> txn.incident(atom).count()); + reads.put("forEachIncidentEdge", (txn, atom) -> txn.forEachIncidentEdge(atom, _ -> { + })); + reads.put("incidentTree", (txn, atom) -> txn.incidentTree(atom).size()); + reads.put("degree", Transaction::degree); + return reads; + } + + @Test + void incidenceReadsAreValidatedAcrossACheckpoint() { + try (StorageEngine engine = open()) { + long hub = node(engine, NODE); + long leaf = node(engine, NODE); + for (int i = 0; i < 60; i++) { + joinNewEdge(engine, hub); + } + joinNewEdge(engine, leaf); + engine.read(snapshot -> { + assertInstanceOf(Postings.Promoted.class, snapshot.scan(EngineSlots.REVERSE).get(hub).orElseThrow()); + assertInstanceOf(Postings.Inline.class, snapshot.scan(EngineSlots.REVERSE).get(leaf).orElseThrow()); + return null; + }); + for (Map.Entry> read : incidenceReads().entrySet()) { + for (long atom : new long[]{hub, leaf}) { + String what = read.getKey() + " of " + atom; + joinNewEdge(engine, atom); + Transaction unchanged = engine.begin(SERIALIZABLE); + read.getValue().accept(unchanged, atom); + ownWrite(unchanged); + unrelatedCommit(engine); + assertEquals(CommitResult.Outcome.REBASED, commitAfterCheckpoint(engine, unchanged), what); + + Transaction changed = engine.begin(SERIALIZABLE); + read.getValue().accept(changed, atom); + ownWrite(changed); + joinNewEdge(engine, atom); + assertConflict(engine, changed, "incident edges of atom " + atom); + } + } + } + } + + @Test + void typeIndexReadsAreValidatedAcrossACheckpoint() { + try (StorageEngine engine = open()) { + for (int i = 0; i < 40; i++) { + node(engine, NODE); + } + node(engine, FEW); + Map> reads = new LinkedHashMap<>(); + for (int type : new int[]{NODE, FEW}) { + reads.put("atomsOfType " + type, txn -> txn.atomsOfType(0, type).count()); + reads.put("typeTree " + type, txn -> txn.typeTree(0, type).size()); + reads.put("countOfType " + type, txn -> txn.countOfType(0, type)); + } + engine.read(snapshot -> { + assertInstanceOf(Postings.Promoted.class, snapshot.scan(EngineSlots.TYPES).get(EngineSlots.typeKey(0, NODE)).orElseThrow()); + assertInstanceOf(Postings.Inline.class, snapshot.scan(EngineSlots.TYPES).get(EngineSlots.typeKey(0, FEW)).orElseThrow()); + return null; + }); + for (Map.Entry> read : reads.entrySet()) { + int type = read.getKey().endsWith(" " + NODE) ? NODE : FEW; + node(engine, type); + Transaction unchanged = engine.begin(SERIALIZABLE); + read.getValue().accept(unchanged); + ownWrite(unchanged); + unrelatedCommit(engine); + assertEquals(CommitResult.Outcome.REBASED, commitAfterCheckpoint(engine, unchanged), read.getKey()); + + Transaction changed = engine.begin(SERIALIZABLE); + read.getValue().accept(changed); + ownWrite(changed); + node(engine, type); + assertConflict(engine, changed, "atoms of type " + type + " in tenant 0"); + } + } + } + + @Test + void canonicalKeyLookupsAreValidatedIncludingAbsentKeys() { + try (StorageEngine engine = open()) { + long alice = engine.write(txn -> txn.createNode(NODE, "alice")); + + Transaction absent = engine.begin(SERIALIZABLE); + assertTrue(absent.resolve(0, NODE, "ghost").isEmpty()); + ownWrite(absent); + engine.write(txn -> txn.createNode(NODE, "someone-else")); + assertEquals(CommitResult.Outcome.REBASED, commitAfterCheckpoint(engine, absent)); + + Transaction appears = engine.begin(SERIALIZABLE); + assertTrue(appears.resolve(0, NODE, "ghost").isEmpty()); + ownWrite(appears); + engine.write(txn -> txn.createNode(NODE, "ghost")); + assertConflict(engine, appears, "canonical key"); + + Transaction disappears = engine.begin(SERIALIZABLE); + assertEquals(alice, disappears.resolve(0, NODE, "alice").orElseThrow()); + ownWrite(disappears); + engine.write(txn -> { + txn.delete(alice); + return null; + }); + assertConflict(engine, disappears, "canonical key"); + } + } + + @Test + void wholeSlotScansAreValidatedByContent() { + try (StorageEngine engine = open()) { + node(engine, NODE); + engine.write(txn -> { + txn.put(OBSERVED, 1, 1L); + return null; + }); + + Transaction catalog = engine.begin(SERIALIZABLE); + catalog.atoms().count(); + ownWrite(catalog); + engine.write(txn -> { + txn.put(WRITTEN, 9_999, 1L); + return null; + }); + assertEquals(CommitResult.Outcome.REBASED, commitAfterCheckpoint(engine, catalog)); + + Transaction grows = engine.begin(SERIALIZABLE); + grows.atoms().count(); + ownWrite(grows); + node(engine, OTHER); + assertConflict(engine, grows, "serializable scan of catalog"); + + Transaction slot = engine.begin(SERIALIZABLE); + slot.scan(OBSERVED).stream().count(); + ownWrite(slot); + unrelatedCommit(engine); + assertEquals(CommitResult.Outcome.REBASED, commitAfterCheckpoint(engine, slot)); + + Transaction moved = engine.begin(SERIALIZABLE); + moved.scan(OBSERVED).stream().count(); + ownWrite(moved); + engine.write(txn -> { + txn.put(OBSERVED, 2, 2L); + return null; + }); + assertConflict(engine, moved, "serializable scan of observed"); + } + } + + @Test + void nestedRangeReadsIgnoreChangesOutsideTheRange() { + try (StorageEngine engine = open()) { + long hub = node(engine, NODE); + long[] edges = LongStream.range(0, 60).map(_ -> joinNewEdge(engine, hub)).toArray(); + long low = edges[10]; + long high = edges[20]; + + Transaction outside = engine.begin(SERIALIZABLE); + assertInstanceOf(Postings.Promoted.class, outside.nested(EngineSlots.REVERSE, hub, low, high).orElseThrow()); + ownWrite(outside); + joinNewEdge(engine, hub); + engine.write(txn -> { + txn.remove(edges[5], hub); + return null; + }); + assertEquals(CommitResult.Outcome.REBASED, commitAfterCheckpoint(engine, outside)); + + Transaction inside = engine.begin(SERIALIZABLE); + inside.nested(EngineSlots.REVERSE, hub, low, high); + ownWrite(inside); + engine.write(txn -> { + txn.remove(edges[15], hub); + return null; + }); + assertConflict(engine, inside, "between " + low + " and " + high); + } + } + + @Test + void unrecordedReadsAreNotValidated() { + try (StorageEngine engine = open()) { + node(engine, NODE); + + Transaction estimated = engine.begin(SERIALIZABLE); + try (View.Unrecorded _ = estimated.unrecorded()) { + try (View.Unrecorded _ = estimated.unrecorded()) { + estimated.countOfType(0, NODE); + } + estimated.degree(1); + } + ownWrite(estimated); + node(engine, NODE); + assertEquals(CommitResult.Outcome.REBASED, commitAfterCheckpoint(engine, estimated)); + + Transaction recorded = engine.begin(SERIALIZABLE); + try (View.Unrecorded _ = recorded.unrecorded()) { + recorded.countOfType(0, OTHER); + } + recorded.countOfType(0, NODE); + ownWrite(recorded); + node(engine, NODE); + assertConflict(engine, recorded, "atoms of type " + NODE); + + try (Transaction first = engine.begin(); Transaction second = engine.begin(); Snapshot snapshot = engine.snapshot()) { + assertSame(View.NOT_RECORDING, first.unrecorded()); + assertSame(View.NOT_RECORDING, second.unrecorded()); + assertSame(View.NOT_RECORDING, snapshot.unrecorded()); + } + } + } + + @Test + void ownWritesDoNotInvalidateOwnReads() { + try (StorageEngine engine = open()) { + node(engine, NODE); + long hub = node(engine, NODE); + Transaction txn = engine.begin(SERIALIZABLE); + long before = txn.countOfType(0, NODE); + long degree = txn.degree(hub); + txn.createNode(NODE, "mine"); + long edge = txn.createEdge(EDGE, EdgeKind.SET); + txn.insert(edge, hub); + assertEquals(before + 1, txn.countOfType(0, NODE)); + assertEquals(degree + 1, txn.degree(hub)); + long observed = txn.scan(OBSERVED).size(); + txn.put(OBSERVED, 5, 5L); + assertEquals(observed + 1, txn.scan(OBSERVED).size()); + unrelatedCommit(engine); + assertEquals(CommitResult.Outcome.REBASED, commitAfterCheckpoint(engine, txn)); + } + } + + @Test + void manyKeyReadsInOneSlotArePromotedToAScanOfTheSlot() { + try (StorageEngine engine = open()) { + Transaction few = engine.begin(SERIALIZABLE); + for (long key = 0; key < Transaction.PROMOTE_AFTER; key++) { + few.get(OBSERVED, key); + } + ownWrite(few); + engine.write(txn -> { + txn.put(OBSERVED, Transaction.PROMOTE_AFTER + 10, 1L); + return null; + }); + assertEquals(CommitResult.Outcome.REBASED, few.commit().outcome()); + + Transaction many = engine.begin(SERIALIZABLE); + for (long key = 0; key <= Transaction.PROMOTE_AFTER; key++) { + many.get(OBSERVED, key); + } + ownWrite(many); + engine.write(txn -> { + txn.put(OBSERVED, Transaction.PROMOTE_AFTER + 20, 1L); + return null; + }); + assertConflict(engine, many, "serializable scan of observed"); + } + } +} From 7df8a1dab20af352fac67664d882141b218c3e6c Mon Sep 17 00:00:00 2001 From: venkat1701 Date: Tue, 27 Oct 2026 14:57:21 +0530 Subject: [PATCH 2/3] Record reads of property names that have no key yet A serializable read of a property no transaction had set returned before reading anything, so a concurrent SET of that property did not invalidate it. Reader.property now reads the atom's property record and Reader.range reads the type's atoms when the view records reads. Reader.range also scanned nothing for an unindexed JSON path, since only indexed paths have a key; it now scans the documents. --- .../src/main/java/io/hstore/db/Reader.java | 16 +++++++-- .../test/java/io/hstore/db/DatabaseTest.java | 29 ++++++++++++++++ .../io/hstore/db/query/SerializableTest.java | 33 +++++++++++++++++++ docs/transactions/transactions.md | 4 ++- .../io/hstore/engine/txn/Transaction.java | 5 +++ .../main/java/io/hstore/engine/txn/View.java | 4 +++ 6 files changed, 87 insertions(+), 4 deletions(-) diff --git a/database/src/main/java/io/hstore/db/Reader.java b/database/src/main/java/io/hstore/db/Reader.java index ce814d89..f72b0fd1 100644 --- a/database/src/main/java/io/hstore/db/Reader.java +++ b/database/src/main/java/io/hstore/db/Reader.java @@ -154,7 +154,7 @@ public Stream allAtoms(boolean edges) { public Optional property(long atom, String name) { OptionalInt key = schema().existingPropertyKey(name); if (key.isEmpty()) { - return Optional.empty(); + return absent(atom); } return bag(atom).get(key.getAsInt()).map(Property::value); } @@ -162,11 +162,18 @@ public Optional property(long atom, String name) { public Optional property(long atom, String name, long validAt) { OptionalInt key = schema().existingPropertyKey(name); if (key.isEmpty()) { - return Optional.empty(); + return absent(atom); } return bag(atom).get(key.getAsInt()).filter(property -> property.validAt(validAt)).map(Property::value); } + private Optional absent(long atom) { + if (view.recordsReads()) { + bag(atom); + } + return Optional.empty(); + } + public Map properties(long atom) { Map named = new LinkedHashMap<>(); bag(atom).entries().forEach((key, property) -> named.put(schema().propertyName(key), property.value())); @@ -283,7 +290,10 @@ public LongStream range(String type, String property, Optional low, Optio Predicate accept = candidate -> low.map(bound -> lowInclusive ? candidate.compareTo(bound) >= 0 : candidate.compareTo(bound) > 0).orElse(true) && high.map(bound -> highInclusive ? candidate.compareTo(bound) <= 0 : candidate.compareTo(bound) < 0).orElse(true); OptionalInt key = schema().existingPropertyKey(property); - if (key.isEmpty()) { + if (key.isEmpty() && !property.startsWith("$")) { + if (view.recordsReads()) { + atomsOfType(definition.id()).forEach(this::bag); + } return LongStream.empty(); } Optional indexedTag = indexedTag(definition, property); diff --git a/database/src/test/java/io/hstore/db/DatabaseTest.java b/database/src/test/java/io/hstore/db/DatabaseTest.java index 78804215..e2f8f718 100644 --- a/database/src/test/java/io/hstore/db/DatabaseTest.java +++ b/database/src/test/java/io/hstore/db/DatabaseTest.java @@ -31,6 +31,7 @@ import io.hstore.engine.EngineOptions; import io.hstore.engine.HStoreException; import io.hstore.engine.topology.Validity; +import io.hstore.engine.txn.Isolation; import io.hstore.engine.txn.TxnOptions; import org.junit.jupiter.api.AfterEach; @@ -43,6 +44,9 @@ import java.util.Map; import java.util.Optional; import java.util.Set; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.TimeUnit; import java.util.stream.Collectors; import java.util.stream.LongStream; @@ -78,6 +82,29 @@ void close() { database.close(); } + @Test + void serializableRangesOverAnUndeclaredPropertyStillConflict() throws Exception { + List people = database.write(writer -> List.of(writer.node("Paper", "a", Map.of()), writer.node("Paper", "b", Map.of()))); + TxnOptions serializable = TxnOptions.defaults().withIsolation(Isolation.SERIALIZABLE); + CountDownLatch bothRead = new CountDownLatch(2); + List> flaggers = people.stream().map(person -> CompletableFuture.runAsync(() -> database.write(serializable, writer -> { + long flagged = writer.range("Paper", "flag", Optional.of(new Value.Bool(true)), Optional.of(new Value.Bool(true)), true, true).count(); + bothRead.countDown(); + try { + assertTrue(bothRead.await(10, TimeUnit.SECONDS)); + } catch (InterruptedException e) { + throw new IllegalStateException(e); + } + if (flagged == 0) { + writer.set(person, "flag", new Value.Bool(true)); + } + return null; + }))).toList(); + CompletableFuture.allOf(flaggers.toArray(CompletableFuture[]::new)).get(30, TimeUnit.SECONDS); + long flagged = database.read(reader -> reader.range("Paper", "flag", Optional.of(new Value.Bool(true)), Optional.of(new Value.Bool(true)), true, true).count()); + assertEquals(1, flagged); + } + @Test void propertyIndexAgreesWithFullScan() { database.write(writer -> { @@ -226,6 +253,8 @@ void documentsAndJsonPathIndexes() { assertEquals("VLDB", reader.document(paper).orElseThrow().select("$.venue.name").findFirst().orElseThrow().render().replace("'", "")); assertEquals(List.of(paper), reader.range("Paper", "$.venue.year", Optional.of(new Value.Int(2026)), Optional.empty(), true, true) .boxed().toList()); + assertEquals(List.of(paper), reader.range("Paper", "$.venue.name", Optional.of(new Value.Text("VLDB")), Optional.of(new Value.Text("VLDB")), true, true) + .boxed().toList()); return null; }); } diff --git a/database/src/test/java/io/hstore/db/query/SerializableTest.java b/database/src/test/java/io/hstore/db/query/SerializableTest.java index d3b4cce8..983a45f3 100644 --- a/database/src/test/java/io/hstore/db/query/SerializableTest.java +++ b/database/src/test/java/io/hstore/db/query/SerializableTest.java @@ -114,6 +114,39 @@ void writeSkewThroughAnIndexedPredicateIsRejected() { assertEquals("1", count(ON_CALL)); } + @Test + void writeSkewThroughAnUndeclaredPropertyIsRejected() { + String flagged = "MATCH NODE n:Account WHERE n.flag = true RETURN count(*)"; + begin(true); + assertEquals("0", cell(a.execute(flagged))); + assertEquals("0", cell(b.execute(flagged))); + a.execute("SET @Account:'x'.flag = true"); + b.execute("SET @Account:'y'.flag = true"); + onlySecondFails(commitBoth()); + assertEquals("1", count(flagged)); + } + + @Test + void readingAnUndeclaredPropertyOfOneAtomIsRecorded() { + begin(true); + assertEquals("null", cell(a.execute("MATCH NODE n:Account WHERE n = @Account:'x' RETURN n.flag"))); + assertEquals("null", cell(b.execute("MATCH NODE n:Account WHERE n = @Account:'y' RETURN n.flag"))); + a.execute("SET @Account:'y'.flag = true"); + b.execute("SET @Account:'x'.flag = true"); + onlySecondFails(commitBoth()); + } + + @Test + void undeclaredPropertiesDoNotConflictUnderSnapshot() { + String flagged = "MATCH NODE n:Account WHERE n.flag = true RETURN count(*)"; + begin(false); + assertEquals("0", cell(a.execute(flagged))); + assertEquals("0", cell(b.execute(flagged))); + a.execute("SET @Account:'x'.flag = true"); + b.execute("SET @Account:'y'.flag = true"); + bothCommit(commitBoth()); + } + @Test void indexedPredicatesIgnoreChangesOutsideTheirRange() { a.execute("BEGIN SERIALIZABLE"); diff --git a/docs/transactions/transactions.md b/docs/transactions/transactions.md index 91cf52dd..29994c2e 100644 --- a/docs/transactions/transactions.md +++ b/docs/transactions/transactions.md @@ -205,6 +205,8 @@ Each read is recorded once, on first observation, with the value from the **base The database module reads the property index through `nested`: `Reader.range` passes the directory key `directoryKey(tenant, property, tag)` of `PROPERTY_INDEX` and the order-key range `[lowKey, highKey]` of the predicate. `MATCH NODE n:Account WHERE n.oncall = true` therefore depends on the `true` bucket of `oncall` only, and a commit that changes some other account's `oncall` from `true` to `false` invalidates it, while a commit that only adds `oncall = false` accounts does not. +A property name that no transaction has used yet has no dictionary key, and no property record holds it. `Reader.property` still reads the atom's property record in that case, and `Reader.range` reads the type's posting list and the property record of each atom of the type, but only when `View.recordsReads()` is true, so `SNAPSHOT` transactions skip that work. `MATCH NODE n:Account WHERE n.flag = true` therefore conflicts with a concurrent `SET @Account:'x'.flag = true` even when `flag` is not declared. + Same content means `TreeDiff.sameValue` for values and `TreeDiff.sameContent` for trees. Equal values are the same. Values that hold nested trees (`Postings.Promoted`, the property index's `IndexRoot`, edge records) are the same when their other fields are equal and each nested tree holds the same entries. Inline posting lists compare by their keys and values. Tree comparison walks `TreeDiff.diff`, which skips subtrees that are `Ref.same`, so it costs time proportional to the changed region, and a range read prunes subtrees whose summary lies outside the range. Because `Ref.same` treats a pending node as the same tree as the stored reference a published checkpoint wrote it as, a checkpoint that finishes between a read and the commit doesn't cause a conflict, for inline and promoted posting lists alike. The fast path never needs read validation: it is only taken when every touched slot, including every slot that was only read, has the same root as at the base. Validation is therefore at the granularity above when slots moved and free when they did not. @@ -214,7 +216,7 @@ The fast path never needs read validation: it is only taken when every touched s * Posting lists are validated whole. Any change to an atom's incidence list invalidates `incident(atom)`, including an edge that a role filter or tenant filter would have skipped; any new atom of a type invalidates a scan of that type, even if a residual predicate would have rejected it. * Canonical keys are validated per hash bucket of `(tenant, type, key)`. * `scan(slot)` and `atoms()` are validated per slot, so any change to the slot conflicts. Database code that scans whole slots (schema listings, tenant and user listings, materialized view data, signals, exact semantic search) is protected at that granularity. -* Promotion bounds the read set, in the spirit of predicate-lock promotion in PostgreSQL's serializable snapshot isolation: when a transaction records more than `Transaction.PROMOTE_AFTER` (8192) key and range reads in one slot, or that many atom reads, they are replaced by one read of the whole slot (`CATALOG` for atoms). +* Promotion bounds the read set, in the spirit of predicate-lock promotion in PostgreSQL's serializable snapshot isolation: when a transaction records more than `Transaction.PROMOTE_AFTER` (8192) key and range reads in one slot, or that many atom reads, they are replaced by one read of the whole slot (`CATALOG` for atoms). After that, any commit that writes the slot, in any tenant, invalidates the transaction, so a long `SERIALIZABLE` transaction under a steady write load can fail on every retry. Split such work into shorter transactions, or run it under `SNAPSHOT` when it doesn't depend on predicate reads. * Validation fixes the serial order to the commit order. A transaction that read something a concurrent transaction changed and committed first fails, even when the opposite order would have been serial, for example after reading that a key is absent while the other transaction inserts it. * A value that a concurrent commit changed and changed back counts as unchanged. diff --git a/engine/src/main/java/io/hstore/engine/txn/Transaction.java b/engine/src/main/java/io/hstore/engine/txn/Transaction.java index 9d83b556..5b92a15c 100644 --- a/engine/src/main/java/io/hstore/engine/txn/Transaction.java +++ b/engine/src/main/java/io/hstore/engine/txn/Transaction.java @@ -139,6 +139,11 @@ private boolean recording() { return serializable && unrecorded == 0; } + @Override + public boolean recordsReads() { + return recording(); + } + @Override void observeAtom(long atom, Optional record) { if (!recording() || slotReads.contains(EngineSlots.CATALOG) || atomReads.containsKey(atom)) { diff --git a/engine/src/main/java/io/hstore/engine/txn/View.java b/engine/src/main/java/io/hstore/engine/txn/View.java index 0d802d87..5158c2a4 100644 --- a/engine/src/main/java/io/hstore/engine/txn/View.java +++ b/engine/src/main/java/io/hstore/engine/txn/View.java @@ -58,6 +58,10 @@ public Unrecorded unrecorded() { return NOT_RECORDING; } + public boolean recordsReads() { + return false; + } + public abstract long generation(); public abstract int branch(); From 85be0613cde722f6a284eac925249191ab2aff97 Mon Sep 17 00:00:00 2001 From: venkat1701 Date: Tue, 27 Oct 2026 16:41:25 +0530 Subject: [PATCH 3/3] Guard writes that change nothing, and type-check unindexed ranges - A keyed write that left its value unchanged did not count its slot as touched, so the commit could take the fast path past a concurrent change of that key. UNSET of an absent property lost its effect this way; every slot a transaction writes now counts. - UNSET of a property name without a key reads the property record and fails at commit if another transaction set that property. - Reader.range compares only values of the bound's type, as the index does, so an unindexed scan no longer returns other types or nulls. --- .../src/main/java/io/hstore/db/Reader.java | 10 ++- .../src/main/java/io/hstore/db/Writer.java | 12 ++- .../test/java/io/hstore/db/DatabaseTest.java | 15 ++++ .../io/hstore/db/query/SerializableTest.java | 34 ++++++++ docs/transactions/transactions.md | 2 +- .../main/java/io/hstore/engine/txn/Op.java | 1 + .../java/io/hstore/engine/txn/Workspace.java | 9 +- .../hstore/engine/txn/PreconditionTest.java | 87 +++++++++++++++++++ 8 files changed, 165 insertions(+), 5 deletions(-) create mode 100644 engine/src/test/java/io/hstore/engine/txn/PreconditionTest.java diff --git a/database/src/main/java/io/hstore/db/Reader.java b/database/src/main/java/io/hstore/db/Reader.java index f72b0fd1..cd05f18c 100644 --- a/database/src/main/java/io/hstore/db/Reader.java +++ b/database/src/main/java/io/hstore/db/Reader.java @@ -287,8 +287,10 @@ public LongStream lookup(String type, String property, Value value) { public LongStream range(String type, String property, Optional low, Optional high, boolean lowInclusive, boolean highInclusive) { TypeDef definition = requireType(type); - Predicate accept = candidate -> low.map(bound -> lowInclusive ? candidate.compareTo(bound) >= 0 : candidate.compareTo(bound) > 0).orElse(true) - && high.map(bound -> highInclusive ? candidate.compareTo(bound) <= 0 : candidate.compareTo(bound) < 0).orElse(true); + Predicate accept = candidate -> low.map(bound -> comparable(candidate, bound) + && (lowInclusive ? candidate.compareTo(bound) >= 0 : candidate.compareTo(bound) > 0)).orElse(true) + && high.map(bound -> comparable(candidate, bound) + && (highInclusive ? candidate.compareTo(bound) <= 0 : candidate.compareTo(bound) < 0)).orElse(true); OptionalInt key = schema().existingPropertyKey(property); if (key.isEmpty() && !property.startsWith("$")) { if (view.recordsReads()) { @@ -314,6 +316,10 @@ public LongStream range(String type, String property, Optional low, Optio .filter(atom -> bag(atom).get(key.getAsInt()).map(Property::value).filter(accept).isPresent()); } + private static boolean comparable(Value candidate, Value bound) { + return candidate.tag() == bound.tag() ? !(candidate instanceof Value.Null) : candidate.tag().numeric() && bound.tag().numeric(); + } + public static Optional indexedTag(TypeDef type, String property) { return type.property(property) .filter(TypeDef.PropertyDef::indexed) diff --git a/database/src/main/java/io/hstore/db/Writer.java b/database/src/main/java/io/hstore/db/Writer.java index 07fbc4e6..1abaa36c 100644 --- a/database/src/main/java/io/hstore/db/Writer.java +++ b/database/src/main/java/io/hstore/db/Writer.java @@ -38,6 +38,7 @@ import java.util.List; import java.util.Map; import java.util.Optional; +import java.util.OptionalInt; import java.util.function.UnaryOperator; public final class Writer extends Reader { @@ -250,7 +251,16 @@ public void set(long atom, String name, Value value, Validity validity) { public void unset(long atom, String name) { principal().requireWrite(); require(atom); - schema().existingPropertyKey(name).ifPresent(key -> updateBag(atom, key, bag -> bag.without(key))); + OptionalInt key = schema().existingPropertyKey(name); + if (key.isPresent()) { + updateBag(atom, key.getAsInt(), bag -> bag.without(key.getAsInt())); + return; + } + bag(atom); + txn.merge(PropertySlots.PROPERTIES, atom, current -> { + OptionalInt created = schema().existingPropertyKey(name); + return created.isEmpty() || current.orElse(PropertyBag.EMPTY).get(created.getAsInt()).isEmpty(); + }, current -> current); } private Value externalizeLargeText(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 e2f8f718..866e2d10 100644 --- a/database/src/test/java/io/hstore/db/DatabaseTest.java +++ b/database/src/test/java/io/hstore/db/DatabaseTest.java @@ -257,6 +257,21 @@ void documentsAndJsonPathIndexes() { .boxed().toList()); return null; }); + long[] mixed = database.write(writer -> { + long late = writer.node("Paper", "late", Map.of()); + writer.document(late, Json.parse("{\"venue\":{\"name\":2027}}")); + long none = writer.node("Paper", "none", Map.of()); + writer.document(none, Json.parse("{\"venue\":{\"name\":null}}")); + return new long[]{late, none}; + }); + database.read(reader -> { + long paper = reader.resolve("Paper", "hypergraphs"); + assertEquals(List.of(paper), reader.range("Paper", "$.venue.name", Optional.of(new Value.Text("A")), Optional.empty(), true, true) + .boxed().toList()); + assertEquals(List.of(mixed[0]), reader.range("Paper", "$.venue.name", Optional.empty(), Optional.of(new Value.Real(3000.5)), true, true) + .boxed().toList()); + return null; + }); } @Test diff --git a/database/src/test/java/io/hstore/db/query/SerializableTest.java b/database/src/test/java/io/hstore/db/query/SerializableTest.java index 983a45f3..655c26c4 100644 --- a/database/src/test/java/io/hstore/db/query/SerializableTest.java +++ b/database/src/test/java/io/hstore/db/query/SerializableTest.java @@ -136,6 +136,40 @@ void readingAnUndeclaredPropertyOfOneAtomIsRecorded() { onlySecondFails(commitBoth()); } + @Test + void unsettingAPropertyNobodyHasSetIsRecorded() { + begin(true); + assertEquals("100", cell(b.execute("MATCH NODE n:Account WHERE n = @Account:'y' RETURN n.balance"))); + a.execute("SET @Account:'y'.balance = 0"); + a.execute("UNSET @Account:'x'.other"); + b.execute("SET @Account:'x'.other = true"); + List> outcomes = new ArrayList<>(); + for (Session session : List.of(b, a)) { + try { + session.execute("COMMIT"); + outcomes.add(Optional.empty()); + } catch (HStoreException.Conflict conflict) { + outcomes.add(Optional.of(conflict)); + } + } + assertTrue(outcomes.get(1).isPresent(), "the second commit should fail"); + } + + @Test + void anUnsetOfAnAbsentPropertyIsNotLostUnderSnapshot() { + for (String property : List.of("balance", "neverSet", "alsoNeverSet")) { + admin.execute("UNSET @Account:'x'.balance"); + begin(false); + a.execute("UNSET @Account:'x'." + property); + if (!property.equals("alsoNeverSet")) { + a.execute("INSERT NODE Note 'n-" + property + "'"); + } + b.execute("SET @Account:'x'." + property + " = 7"); + b.execute("COMMIT"); + assertThrows(HStoreException.Conflict.class, () -> a.execute("COMMIT"), property); + } + } + @Test void undeclaredPropertiesDoNotConflictUnderSnapshot() { String flagged = "MATCH NODE n:Account WHERE n.flag = true RETURN count(*)"; diff --git a/docs/transactions/transactions.md b/docs/transactions/transactions.md index 29994c2e..270d5e25 100644 --- a/docs/transactions/transactions.md +++ b/docs/transactions/transactions.md @@ -134,7 +134,7 @@ commit(txn): ### 5.1 Fast path: per-slot merge -`Workspace.untouchedSince(latest, includingReads)` is true when every slot the transaction **changed** still has the same root in `latest` as in its base (`Ref.same`). A slot counts as changed when its tree root differs from the base root. Slots that were only read are ignored for `SNAPSHOT` transactions. For `SERIALIZABLE` transactions they are included, because the fast path skips `validateReads`. When the check passes, the commit takes the latest root vector and overlays only the changed slots (`rebasedOnto`). A concurrent commit to a slot this transaction merely read is therefore kept, and commits that write disjoint slots never replay, however many other commits landed in between. +`Workspace.untouchedSince(latest, includingReads)` is true when every slot the transaction **changed** still has the same root in `latest` as in its base (`Ref.same`). A slot counts as changed when its tree root differs from the base root, or when the transaction wrote to it at all: every keyed write (`put`, `delete`, `update`, `merge`) carries a precondition (for `put`, `delete` and `update`, that the key still holds what the transaction saw), and a write that leaves the value as it was does not change the root. Such a slot is recorded by `Workspace.guard`, so when it moved the commit replays and checks the precondition against the latest state. Without that, a write that changes nothing (an `UNSET` of a property the atom does not have, or a `merge` used only as a guard, `current -> current` with a precondition on another record) would never be checked whenever the transaction's other writes went to slots that had not moved, and a concurrent change of that key would be lost. Slots that were only read are ignored for `SNAPSHOT` transactions. For `SERIALIZABLE` transactions they are included, because the fast path skips `validateReads`. When the check passes, the commit takes the latest root vector and overlays only the changed slots (`rebasedOnto`). A concurrent commit to a slot this transaction merely read is therefore kept, and commits that write disjoint slots never replay, however many other commits landed in between. `requestsStable` handles idempotency keys: if the transaction carries a request id, the `REQUESTS` slot is written inside the commit lock, so the fast path additionally requires that `REQUESTS` itself has not moved since the base. diff --git a/engine/src/main/java/io/hstore/engine/txn/Op.java b/engine/src/main/java/io/hstore/engine/txn/Op.java index df998599..f30f38a4 100644 --- a/engine/src/main/java/io/hstore/engine/txn/Op.java +++ b/engine/src/main/java/io/hstore/engine/txn/Op.java @@ -104,6 +104,7 @@ record SlotWrite(Slot slot, long key, Predicate> precondition, @Override public void apply(Workspace workspace) { + workspace.guard(slot); Optional current = workspace.get(slot, key); if (!precondition.test(current)) { String message = "precondition failed for " + slot + " key " + key; diff --git a/engine/src/main/java/io/hstore/engine/txn/Workspace.java b/engine/src/main/java/io/hstore/engine/txn/Workspace.java index f09d88bc..9c3dfbf0 100644 --- a/engine/src/main/java/io/hstore/engine/txn/Workspace.java +++ b/engine/src/main/java/io/hstore/engine/txn/Workspace.java @@ -19,9 +19,11 @@ import java.util.ArrayList; import java.util.HashMap; +import java.util.HashSet; import java.util.List; import java.util.Map; import java.util.Optional; +import java.util.Set; import java.util.function.UnaryOperator; public final class Workspace { @@ -41,6 +43,7 @@ interface EdgeGuard { private final EdgeGuard guard; private final boolean replaying; private final Map> trees = new HashMap<>(); + private final Set guarded = new HashSet<>(); private final List memberChanges = new ArrayList<>(); private final List> slotChanges = new ArrayList<>(); @@ -142,9 +145,13 @@ public RootVector rebasedOnto(RootVector latest) { return roots; } + void guard(Slot slot) { + guarded.add(slot.id()); + } + boolean untouchedSince(RootVector latest, boolean includingReads) { return trees.entrySet().stream() - .filter(entry -> includingReads || changed(entry)) + .filter(entry -> includingReads || guarded.contains(entry.getKey()) || changed(entry)) .allMatch(entry -> Ref.same(base.get(entry.getKey()), latest.get(entry.getKey()))); } diff --git a/engine/src/test/java/io/hstore/engine/txn/PreconditionTest.java b/engine/src/test/java/io/hstore/engine/txn/PreconditionTest.java new file mode 100644 index 00000000..ca2a088e --- /dev/null +++ b/engine/src/test/java/io/hstore/engine/txn/PreconditionTest.java @@ -0,0 +1,87 @@ +// SPDX-FileCopyrightText: MetaCognition India Pvt Limited +// SPDX-License-Identifier: PolyForm-Noncommercial-1.0.0 + +package io.hstore.engine.txn; + +import io.hstore.engine.EngineOptions; +import io.hstore.engine.HStoreException; +import io.hstore.engine.StorageEngine; +import io.hstore.engine.catalog.Slot; +import io.hstore.engine.page.ByteCursor; +import io.hstore.engine.tree.FingerprintMode; +import io.hstore.engine.tree.Hashing; +import io.hstore.engine.tree.TreeSchema; +import io.hstore.engine.tree.ValueCodec; + +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.io.TempDir; + +import java.nio.file.Path; +import java.util.List; +import java.util.Optional; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertThrows; + +class PreconditionTest { + + private static final Slot GUARDED = Slot.primary(120, "guarded", counters(240)); + private static final Slot WRITTEN = Slot.primary(121, "written", counters(241)); + + @TempDir + Path directory; + + private static TreeSchema counters(int id) { + return new TreeSchema<>(id, "counters-" + id, FingerprintMode.SET, + ValueCodec.rows(ByteCursor::signedVarLongSize, ByteCursor::putSignedVarLong, ByteCursor::getSignedVarLong), + (key, value, into) -> into.entry(key, Hashing.of(key, value))); + } + + private StorageEngine open() { + return StorageEngine.open(directory, EngineOptions.defaults().withPageSize(1024).withPagesPerSegment(256) + .withExtensions(List.of(GUARDED, WRITTEN), List.of())); + } + + private static void put(StorageEngine engine, Slot slot, long key, long value) { + engine.write(txn -> { + txn.put(slot, key, value); + return null; + }); + } + + @Test + void aPreconditionThatChangesNothingIsStillCheckedAtCommit() { + try (StorageEngine engine = open()) { + put(engine, GUARDED, 1, 1); + Transaction guarded = engine.begin(); + guarded.merge(GUARDED, 1, current -> current.equals(Optional.of(1L)), current -> current); + guarded.put(WRITTEN, 1, 1L); + put(engine, GUARDED, 1, 2); + assertThrows(HStoreException.Conflict.class, guarded::commit); + + Transaction unaffected = engine.begin(); + unaffected.merge(GUARDED, 1, current -> current.equals(Optional.of(2L)), current -> current); + unaffected.put(WRITTEN, 2, 1L); + put(engine, GUARDED, 5, 5); + assertEquals(CommitResult.Outcome.REBASED, unaffected.commit().outcome()); + + Transaction untouched = engine.begin(); + untouched.merge(GUARDED, 1, current -> current.equals(Optional.of(2L)), current -> current); + untouched.put(WRITTEN, 3, 1L); + put(engine, WRITTEN, 9, 9); + assertEquals(CommitResult.Outcome.REBASED, untouched.commit().outcome()); + } + } + + @Test + void aWriteOfTheSameValueStillPreventsALostUpdate() { + try (StorageEngine engine = open()) { + put(engine, GUARDED, 1, 1); + Transaction same = engine.begin(); + same.put(GUARDED, 1, 1L); + same.put(WRITTEN, 1, 1L); + put(engine, GUARDED, 1, 7); + assertThrows(HStoreException.Conflict.class, same::commit); + } + } +}