Skip to content

Commit

Permalink
GH-5189 cache Value objects retrieved from parent sail
Browse files Browse the repository at this point in the history
  • Loading branch information
@hmottestad
hmottestad committed Nov 8, 2024
1 parent 3068e99 commit 6cd7635
Show file tree
Hide file tree
Showing 3 changed files with 102 additions and 37 deletions.
18 changes: 6 additions & 12 deletions ...ail/shacl/src/main/java/org/eclipse/rdf4j/sail/shacl/ast/planNodes/FilterByPredicate.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -36,6 +36,7 @@ public class FilterByPredicate implements PlanNode {
private final Set<IRI> filterOnPredicates;
final PlanNode parent;
private final On on;
private final ConnectionsGroup connectionsGroup;
private boolean printed = false;
private ValidationExecutionLogger validationExecutionLogger;
private final Resource[] dataGraph;
Expand All @@ -53,6 +54,7 @@ public FilterByPredicate(SailConnection connection, Set<IRI> filterOnPredicates,
assert this.connection != null;
this.filterOnPredicates = filterOnPredicates;
this.on = on;
this.connectionsGroup = connectionsGroup;
}

@Override
Expand All @@ -77,18 +79,10 @@ void calculateNext() {
return;
}

filterOnPredicates = FilterByPredicate.this.filterOnPredicates.stream()
.map(predicate -> {
try (var stream = connection
.getStatements(null, predicate, null, true, dataGraph)
.stream()) {
return stream.map(Statement::getPredicate)
.findAny()
.orElse(null);
}
}
)
.filter(Objects::nonNull)
filterOnPredicates = FilterByPredicate.this.filterOnPredicates
.stream()
.map(iri -> connectionsGroup.intern(connection, iri,
ConnectionsGroup.StatementPosition.predicate))
.collect(Collectors.toList());

}
Expand Down
54 changes: 29 additions & 25 deletions ...acl/src/main/java/org/eclipse/rdf4j/sail/shacl/ast/planNodes/FilterByPredicateObject.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -20,8 +20,8 @@
import org.eclipse.rdf4j.common.iteration.CloseableIteration;
import org.eclipse.rdf4j.model.IRI;
import org.eclipse.rdf4j.model.Resource;
import org.eclipse.rdf4j.model.Statement;
import org.eclipse.rdf4j.model.Value;
import org.eclipse.rdf4j.model.vocabulary.RDF;
import org.eclipse.rdf4j.sail.SailConnection;
import org.eclipse.rdf4j.sail.SailException;
import org.eclipse.rdf4j.sail.memory.MemoryStoreConnection;
Expand All @@ -42,10 +42,12 @@ public class FilterByPredicateObject implements PlanNode {
private final FilterOn filterOn;
private final PlanNode parent;
private final boolean returnMatching;
private final ConnectionsGroup connectionsGroup;
private StackTraceElement[] stackTrace;
private boolean printed = false;
private ValidationExecutionLogger validationExecutionLogger;
private final Resource[] dataGraph;
boolean typeFilter;

private final Cache<Resource, Boolean> cache;

Expand All @@ -68,6 +70,10 @@ public FilterByPredicateObject(SailConnection connection, Resource[] dataGraph,
cache = CacheBuilder.newBuilder().maximumSize(10000).build();
}

this.connectionsGroup = connectionsGroup;
if (RDF.TYPE.equals(filterOnPredicate)) {
typeFilter = true;
}
// this.stackTrace = Thread.currentThread().getStackTrace();
}

Expand Down Expand Up @@ -148,31 +154,29 @@ void calculateNext() {

private void internResources() {
if (filterOnObject == null) {

try (var stream = connection
.getStatements(null, FilterByPredicateObject.this.filterOnPredicate, null, includeInferred,
dataGraph)
.stream()) {
filterOnPredicate = stream.map(Statement::getPredicate).findAny().orElse(null);
}

filterOnPredicate = connectionsGroup.intern(connection,
FilterByPredicateObject.this.filterOnPredicate,
ConnectionsGroup.StatementPosition.predicate);
if (filterOnPredicate == null) {
filterOnObject = new Resource[0];
} else {
filterOnObject = FilterByPredicateObject.this.filterOnObject.stream()
.map(object -> {
try (var stream = connection
.getStatements(null, filterOnPredicate, object, includeInferred, dataGraph)
.stream()) {
return stream.map(Statement::getObject)
.map(o -> ((Resource) o))
.findAny()
.orElse(null);
}
}
)
.filter(Objects::nonNull)
.toArray(Resource[]::new);
if (typeFilter && includeInferred) {
filterOnObject = FilterByPredicateObject.this.filterOnObject.stream()
.flatMap(type -> connectionsGroup.getRdfsSubClassOfReasoner()
.backwardsChain(type)
.stream())
.distinct()
.map(object -> connectionsGroup.intern(connection, object,
ConnectionsGroup.StatementPosition.object))
.filter(Objects::nonNull)
.toArray(Resource[]::new);
} else {
filterOnObject = FilterByPredicateObject.this.filterOnObject.stream()
.map(object -> connectionsGroup.intern(connection, object,
ConnectionsGroup.StatementPosition.object))
.filter(Objects::nonNull)
.toArray(Resource[]::new);
}
}

}
Expand Down Expand Up @@ -237,8 +241,8 @@ private Boolean matchesCached(Resource subject, IRI filterOnPredicate, Resource[

private boolean matchesUnCached(Resource subject, IRI filterOnPredicate, Resource[] filterOnObject) {
for (Resource object : filterOnObject) {
if (connection.hasStatement(subject, filterOnPredicate, object, includeInferred,
dataGraph)) {
if (connection.hasStatement(subject, filterOnPredicate, object,
includeInferred && !typeFilter, dataGraph)) {
return true;
}
}
Expand Down
67 changes: 67 additions & 0 deletions .../sail/shacl/src/main/java/org/eclipse/rdf4j/sail/shacl/wrapper/data/ConnectionsGroup.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -13,11 +13,18 @@

import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ExecutionException;

import org.eclipse.rdf4j.common.annotation.InternalUseOnly;
import org.eclipse.rdf4j.common.transaction.IsolationLevels;
import org.eclipse.rdf4j.model.IRI;
import org.eclipse.rdf4j.model.Resource;
import org.eclipse.rdf4j.model.Statement;
import org.eclipse.rdf4j.model.Value;
import org.eclipse.rdf4j.model.util.Values;
import org.eclipse.rdf4j.sail.Sail;
import org.eclipse.rdf4j.sail.SailConnection;
import org.eclipse.rdf4j.sail.SailException;
import org.eclipse.rdf4j.sail.shacl.ShaclSailConnection;
import org.eclipse.rdf4j.sail.shacl.Stats;
import org.eclipse.rdf4j.sail.shacl.ast.planNodes.BufferedSplitter;
Expand All @@ -27,6 +34,9 @@
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.google.common.cache.Cache;
import com.google.common.cache.CacheBuilder;

/**
* @apiNote since 3.0. This feature is for internal use only: its existence, signature or behavior may change without
* warning from one release to the next.
Expand All @@ -52,6 +62,14 @@ public class ConnectionsGroup implements AutoCloseable {
// used to cache Select plan nodes so that we don't query a store for the same data during the same validation step.
private final Map<PlanNode, BufferedSplitter> nodeCache = new ConcurrentHashMap<>();

private final static Value NULL_VALUE = Values.bnode();

private final Cache<Value, Value> INTERNED_VALUE_CACHE = CacheBuilder.newBuilder()
.concurrencyLevel(Runtime.getRuntime().availableProcessors() * 2)
.maximumSize(10000)

.build();

public ConnectionsGroup(SailConnection baseConnection,
SailConnection previousStateConnection, Sail addedStatements, Sail removedStatements,
Stats stats, RdfsSubClassOfReasonerProvider rdfsSubClassOfReasonerProvider,
Expand Down Expand Up @@ -95,6 +113,55 @@ public SailConnection getRemovedStatements() {
return removedStatements;
}

public enum StatementPosition {
subject,
predicate,
object
}

public <T extends Value> T intern(SailConnection connection, T value, StatementPosition statementPosition) {
try {

Value t = INTERNED_VALUE_CACHE.get(value, () -> {

switch (statementPosition) {
case subject:
try (var statements = connection.getStatements(((Resource) value), null, null, false).stream()) {
Resource ret = statements.map(Statement::getSubject).findAny().orElse(null);
if (ret == null) {
return value;
}
return ret;
}
case predicate:
try (var statements = connection.getStatements(null, ((IRI) value), null, false).stream()) {
IRI ret = statements.map(Statement::getPredicate).findAny().orElse(null);
if (ret == null) {
return value;
}
return ret;
}
case object:
try (var statements = connection.getStatements(null, null, value, false).stream()) {
Value ret = statements.map(Statement::getObject).findAny().orElse(null);
if (ret == null) {
return value;
}
return ret;
}
}

throw new IllegalStateException("Unknown statement position: " + statementPosition);

});
if (t == NULL_VALUE)
return null;
return ((T) t);
} catch (ExecutionException e) {
throw new SailException(e);
}
}

@Override
public void close() {
if (addedStatements != null) {
Expand Down

0 comments on commit 6cd7635

Please sign in to comment.