From 208459fdc5d6640df4ac5f4664d1ebf337126fd4 Mon Sep 17 00:00:00 2001 From: Andreas Schwarte Date: Fri, 14 Jun 2024 17:41:06 +0200 Subject: [PATCH] GH-5030: fix FedXDataset support for queries with FROM clauses FedX supports interpret the Dataset being defined on the query: if it is of type FedXDataset, the federation is reduced to the members defined by FedXDataset#getEndpoints(). This is a nice means to evaluate a query on a subset of the federation and is generally working fine. When the query contains a FROM clause, however, the externally passed Dataset is wrapped into a FallbackDataset (which internally keeps the FedXDataset as primary one. This change now inspects the FallbackDataset and unwraps it to reduce the members. Note that for execution still the "FallbackDataset" is used. --- .../rdf4j/query/impl/FallbackDataset.java | 8 ++ .../evaluation/FederationEvalStrategy.java | 42 +++++++--- .../eclipse/rdf4j/federated/BasicTests.java | 82 +++++++++++++++++++ 3 files changed, 120 insertions(+), 12 deletions(-) diff --git a/core/query/src/main/java/org/eclipse/rdf4j/query/impl/FallbackDataset.java b/core/query/src/main/java/org/eclipse/rdf4j/query/impl/FallbackDataset.java index 0eb7c1c72d5..7b4fe2cd2a6 100644 --- a/core/query/src/main/java/org/eclipse/rdf4j/query/impl/FallbackDataset.java +++ b/core/query/src/main/java/org/eclipse/rdf4j/query/impl/FallbackDataset.java @@ -113,4 +113,12 @@ private void appendURI(StringBuilder sb, IRI uri) { } } + public Dataset getPrimary() { + return primary; + } + + public Dataset getFallback() { + return fallback; + } + } diff --git a/tools/federation/src/main/java/org/eclipse/rdf4j/federated/evaluation/FederationEvalStrategy.java b/tools/federation/src/main/java/org/eclipse/rdf4j/federated/evaluation/FederationEvalStrategy.java index c4163320afc..a3e6237dac3 100644 --- a/tools/federation/src/main/java/org/eclipse/rdf4j/federated/evaluation/FederationEvalStrategy.java +++ b/tools/federation/src/main/java/org/eclipse/rdf4j/federated/evaluation/FederationEvalStrategy.java @@ -121,6 +121,7 @@ import org.eclipse.rdf4j.query.algebra.helpers.TupleExprs; import org.eclipse.rdf4j.query.algebra.helpers.collectors.VarNameCollector; import org.eclipse.rdf4j.query.impl.EmptyBindingSet; +import org.eclipse.rdf4j.query.impl.FallbackDataset; import org.eclipse.rdf4j.repository.RepositoryException; import org.eclipse.rdf4j.repository.sparql.federation.CollectionIteration; import org.eclipse.rdf4j.repository.sparql.federation.RepositoryFederatedService; @@ -179,19 +180,8 @@ public TupleExpr optimize(TupleExpr expr, EvaluationStatistics evaluationStatist FederationEvaluationStatistics stats = (FederationEvaluationStatistics) evaluationStatistics; QueryInfo queryInfo = stats.getQueryInfo(); - Dataset dataset = stats.getDataset(); - FederationContext federationContext = queryInfo.getFederationContext(); - List members; - if (dataset instanceof FedXDataset) { - // run the query against a selected set of endpoints - FedXDataset ds = (FedXDataset) dataset; - members = federationContext.getEndpointManager().getEndpoints(ds.getEndpoints()); - } else { - // evaluate against entire federation - FedX fed = federationContext.getFederation(); - members = fed.getMembers(); - } + List members = getMembersFromContext(stats); // Clone the tuple expression to allow for more aggressive optimizations TupleExpr clone = expr.clone(); @@ -270,6 +260,34 @@ public TupleExpr optimize(TupleExpr expr, EvaluationStatistics evaluationStatist return query; } + /** + * Returns the federation members that are active in the current federation. By default it is all federation + * members. If the passed {@link Dataset} is a {@link FedXDataset}, the defined {@link FedXDataset#getEndpoints()} + * are used. + * + * @param evaluationStatisticss to keep the current query context + * @return + */ + protected List getMembersFromContext(FederationEvaluationStatistics evaluationStatisticss) { + Dataset queryDataset = evaluationStatisticss.getDataset(); + + // if the query uses a FROM clause, the external dataset + // is wrapped as primary dataset in FallbackDataset + if (queryDataset instanceof FallbackDataset) { + queryDataset = ((FallbackDataset) queryDataset).getPrimary(); + } + + if (queryDataset instanceof FedXDataset) { + // run the query against a selected set of endpoints + FedXDataset ds = (FedXDataset) queryDataset; + return federationContext.getEndpointManager().getEndpoints(ds.getEndpoints()); + } else { + // evaluate against entire federation + FedX fed = federationContext.getFederation(); + return fed.getMembers(); + } + } + /** * Perform source selection for all statements of the query. As a result of this method all statement nodes are * annotated with their relevant sources. diff --git a/tools/federation/src/test/java/org/eclipse/rdf4j/federated/BasicTests.java b/tools/federation/src/test/java/org/eclipse/rdf4j/federated/BasicTests.java index 98e554c6b58..962ded27832 100644 --- a/tools/federation/src/test/java/org/eclipse/rdf4j/federated/BasicTests.java +++ b/tools/federation/src/test/java/org/eclipse/rdf4j/federated/BasicTests.java @@ -24,6 +24,8 @@ import org.eclipse.rdf4j.model.Statement; import org.eclipse.rdf4j.model.util.Values; import org.eclipse.rdf4j.model.vocabulary.DCTERMS; +import org.eclipse.rdf4j.model.vocabulary.FOAF; +import org.eclipse.rdf4j.model.vocabulary.RDF; import org.eclipse.rdf4j.model.vocabulary.RDFS; import org.eclipse.rdf4j.query.AbstractTupleQueryResultHandler; import org.eclipse.rdf4j.query.BindingSet; @@ -35,6 +37,7 @@ import org.eclipse.rdf4j.query.TupleQueryResult; import org.eclipse.rdf4j.query.TupleQueryResultHandlerException; import org.eclipse.rdf4j.query.impl.SimpleDataset; +import org.eclipse.rdf4j.repository.Repository; import org.eclipse.rdf4j.repository.RepositoryConnection; import org.eclipse.rdf4j.repository.util.Repositories; import org.junit.jupiter.api.Assertions; @@ -575,4 +578,83 @@ public void test_EscapingQuotedLiteral() throws Exception { } } + + @Test + public void test_reduceFederation() throws Exception { + + List endpoints = prepareTest( + Arrays.asList("/tests/basic/data_emptyStore.ttl", "/tests/basic/data_emptyStore.ttl")); + + Repository repo1 = getRepository(1); + Repository repo2 = getRepository(2); + + String repo1Id = endpoints.get(0).getId(); + + IRI graph1 = Values.iri("http://example.org/graph1"); + IRI graph2 = Values.iri("http://example.org/graph2"); + + try (RepositoryConnection con = repo1.getConnection()) { + con.add(Values.iri("http://example.org/repo1/p1"), RDF.TYPE, FOAF.PERSON, graph1); + con.add(Values.iri("http://example.org/repo2/p2"), RDF.TYPE, FOAF.PERSON, graph2); + } + + try (RepositoryConnection con = repo2.getConnection()) { + con.add(Values.iri("http://example.org/repo2/p3"), RDF.TYPE, FOAF.PERSON, graph1); + } + + Repository fedxRepo = fedxRule.getRepository(); + + // 1: regular federation + try (RepositoryConnection con = fedxRepo.getConnection()) { + TupleQuery tupleQuery = con.prepareTupleQuery( + "PREFIX foaf: " + + "SELECT * WHERE { " + + " ?subClass a foaf:Person. " + + " } " + ); + + // expect from both repos + Assertions.assertEquals(3, QueryResults.asSet(tupleQuery.evaluate()).size()); + + // now we scope it additional to graph1 + tupleQuery = con.prepareTupleQuery( + "PREFIX foaf: " + + "SELECT * FROM WHERE { " + + " ?subClass a foaf:Person. " + + " } "); + + // expect results defined in graph1 (1 in repo1, 2 from repo2) + Assertions.assertEquals(2, QueryResults.asSet(tupleQuery.evaluate()).size()); + } + + // 2: reduce to federation member 1 id + // 2a: additionall restrict to named graph + FedXDataset fedXDataset = new FedXDataset(new SimpleDataset()); + fedXDataset.addEndpoint(repo1Id); + + try (RepositoryConnection con = fedxRepo.getConnection()) { + TupleQuery tupleQuery = con.prepareTupleQuery( + "PREFIX foaf: " + + "SELECT * WHERE { " + + " ?subClass a foaf:Person. " + + " } " + ); + tupleQuery.setDataset(fedXDataset); + + // expect result from repo 1 + Assertions.assertEquals(2, QueryResults.asSet(tupleQuery.evaluate()).size()); + + // now we scope it additional to graph1 + tupleQuery = con.prepareTupleQuery( + "PREFIX foaf: " + + "SELECT * FROM WHERE { " + + " ?subClass a foaf:Person. " + + " } "); + tupleQuery.setDataset(fedXDataset); + + // expect result from graph1 from repo1 + Assertions.assertEquals(1, QueryResults.asSet(tupleQuery.evaluate()).size()); + } + + } }