diff --git a/database/src/main/java/io/hstore/db/Reader.java b/database/src/main/java/io/hstore/db/Reader.java index 7540cda..cd05f18 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; @@ -152,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); } @@ -160,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())); @@ -278,17 +287,21 @@ 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()) { + if (key.isEmpty() && !property.startsWith("$")) { + if (view.recordsReads()) { + atomsOfType(definition.id()).forEach(this::bag); + } return LongStream.empty(); } 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)) @@ -303,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 07fbc4e..1abaa36 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/main/java/io/hstore/db/hora/PatternMatcher.java b/database/src/main/java/io/hstore/db/hora/PatternMatcher.java index 57f3dfe..95895eb 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 2a5717b..8470168 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 cae87e1..4ea56ca 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 802da2c..f93e22c 100644 --- a/database/src/main/java/io/hstore/db/security/Security.java +++ b/database/src/main/java/io/hstore/db/security/Security.java @@ -125,7 +125,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/DatabaseTest.java b/database/src/test/java/io/hstore/db/DatabaseTest.java index 7880421..866e2d1 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,23 @@ 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; + }); + 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; }); } 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 0000000..655c26c --- /dev/null +++ b/database/src/test/java/io/hstore/db/query/SerializableTest.java @@ -0,0 +1,324 @@ +// 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 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 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(*)"; + 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"); + 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 04686dc..cadcc08 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 0957f23..270d5e2 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. @@ -187,14 +187,67 @@ 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. + +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. + +**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). 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. + +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 4544902..c4333b0 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/Op.java b/engine/src/main/java/io/hstore/engine/txn/Op.java index df99859..f30f38a 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/Transaction.java b/engine/src/main/java/io/hstore/engine/txn/Transaction.java index 114753e..5b92a15 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,74 @@ void exposeLazily() { scope.freeze(); } + @Override + public Unrecorded unrecorded() { + if (serializable) { + unrecorded++; + } + return resume; + } + + private boolean recording() { + return serializable && unrecorded == 0; + } + + @Override + public boolean recordsReads() { + return recording(); + } + @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 +297,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 +435,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 +454,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 8d7c041..5158c2a 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,21 @@ 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 boolean recordsReads() { + return false; } public abstract long generation(); @@ -88,38 +110,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 +163,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 f828ae5..8af1d18 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/main/java/io/hstore/engine/txn/Workspace.java b/engine/src/main/java/io/hstore/engine/txn/Workspace.java index f09d88b..9c3dfbf 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 0000000..ca2a088 --- /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); + } + } +} 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 0000000..ef12208 --- /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"); + } + } +}