Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
35 changes: 26 additions & 9 deletions database/src/main/java/io/hstore/db/Reader.java
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down Expand Up @@ -152,19 +154,26 @@ public Stream<Atom> allAtoms(boolean edges) {
public Optional<Value> 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);
}

public Optional<Value> 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<Value> absent(long atom) {
if (view.recordsReads()) {
bag(atom);
}
return Optional.empty();
}

public Map<String, Value> properties(long atom) {
Map<String, Value> named = new LinkedHashMap<>();
bag(atom).entries().forEach((key, property) -> named.put(schema().propertyName(key), property.value()));
Expand Down Expand Up @@ -278,17 +287,21 @@ public LongStream lookup(String type, String property, Value value) {

public LongStream range(String type, String property, Optional<Value> low, Optional<Value> high, boolean lowInclusive, boolean highInclusive) {
TypeDef definition = requireType(type);
Predicate<Value> 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<Value> 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<TypeTag> 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))
Expand All @@ -303,6 +316,10 @@ public LongStream range(String type, String property, Optional<Value> 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<TypeTag> indexedTag(TypeDef type, String property) {
return type.property(property)
.filter(TypeDef.PropertyDef::indexed)
Expand Down
12 changes: 11 additions & 1 deletion database/src/main/java/io/hstore/db/Writer.java
Original file line number Diff line number Diff line change
Expand Up @@ -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 {
Expand Down Expand Up @@ -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) {
Expand Down
6 changes: 5 additions & 1 deletion database/src/main/java/io/hstore/db/hora/PatternMatcher.java
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down Expand Up @@ -94,7 +95,10 @@ private long estimate(Pattern pattern, String name, Set<String> bound) {
}

private long typeDomain(Variable variable) {
return variable.type().flatMap(reader::type).map(type -> reader.countOfType(type.id())).orElse(UNKNOWN_DOMAIN);
Optional<TypeDef> 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<String, Long> binding) {
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down Expand Up @@ -53,8 +54,10 @@ public static Tree<Postings<Value>> values(Tree<IndexRoot> directory, int tenant
return new Tree<>(PropertySlots.postingSchema(), directory.source(), root);
}

public static Stream<Entry<Value>> candidates(Tree<IndexRoot> directory, int tenant, int propertyKey, TypeTag tag, long lowKey, long highKey) {
Tree<Postings<Value>> values = values(directory, tenant, propertyKey, tag);
public static Stream<Entry<Value>> 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<Postings<Value>> values = new Tree<>(PropertySlots.postingSchema(), view.source(), root);
return values.range(lowKey, highKey)
.flatMap(bucket -> PropertySlots.VALUE_POSTINGS.stream(values, bucket.key()));
}
Expand Down
14 changes: 12 additions & 2 deletions database/src/main/java/io/hstore/db/query/Planner.java
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down Expand Up @@ -75,9 +76,12 @@ Plan plan(Ast.Match match) {
});
List<Expr> conjuncts = match.where().map(Planner::conjuncts).orElse(List.of());
List<Candidate> 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)));
Expand Down Expand Up @@ -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<Long> atoms) -> {
double[] degrees = atoms.stream().mapToDouble(reader::degree).sorted().toArray();
Expand Down
5 changes: 4 additions & 1 deletion database/src/main/java/io/hstore/db/security/Security.java
Original file line number Diff line number Diff line change
Expand Up @@ -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);
Expand Down
44 changes: 44 additions & 0 deletions database/src/test/java/io/hstore/db/DatabaseTest.java
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand All @@ -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;

Expand Down Expand Up @@ -78,6 +82,29 @@ void close() {
database.close();
}

@Test
void serializableRangesOverAnUndeclaredPropertyStillConflict() throws Exception {
List<Long> 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<CompletableFuture<Void>> 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 -> {
Expand Down Expand Up @@ -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;
});
}
Expand Down
Loading
Loading