GH-5121: prepare execution of left joins in the federation strategy

Prepare to execute a specific implementation of a left join implementation through the federation strategy.
eclipse-rdf4j · Sep 29, 2024 · 2d86cd3 · 2d86cd3
1 parent 4eb1479
commit 2d86cd3
Show file tree

Hide file tree

Showing 3 changed files with 41 additions and 5 deletions.
diff --git a/...deration/src/main/java/org/eclipse/rdf4j/federated/evaluation/FederationEvalStrategy.java b/...deration/src/main/java/org/eclipse/rdf4j/federated/evaluation/FederationEvalStrategy.java
@@ -57,7 +57,6 @@
 import org.eclipse.rdf4j.federated.evaluation.join.ControlledWorkerBindJoin;
 import org.eclipse.rdf4j.federated.evaluation.join.ControlledWorkerBoundJoin;
 import org.eclipse.rdf4j.federated.evaluation.join.ControlledWorkerJoin;
-import org.eclipse.rdf4j.federated.evaluation.join.ControlledWorkerLeftJoin;
 import org.eclipse.rdf4j.federated.evaluation.join.SynchronousBoundJoin;
 import org.eclipse.rdf4j.federated.evaluation.join.SynchronousJoin;
 import org.eclipse.rdf4j.federated.evaluation.union.ControlledWorkerUnion;
@@ -96,6 +95,7 @@
 import org.eclipse.rdf4j.query.QueryEvaluationException;
 import org.eclipse.rdf4j.query.algebra.DescribeOperator;
 import org.eclipse.rdf4j.query.algebra.Join;
+import org.eclipse.rdf4j.query.algebra.LeftJoin;
 import org.eclipse.rdf4j.query.algebra.QueryRoot;
 import org.eclipse.rdf4j.query.algebra.Service;
 import org.eclipse.rdf4j.query.algebra.StatementPattern;
@@ -745,10 +745,7 @@ public CloseableIteration<BindingSet> evaluate(BindingSet bindings) {
 
 				if (problemVars.containsAll(bindings.getBindingNames())) {
 					var leftIter = leftPrepared.evaluate(bindings);
-					ControlledWorkerLeftJoin join = new ControlledWorkerLeftJoin(scheduler, FederationEvalStrategy.this,
-							leftIter, leftJoin, bindings, leftJoin.getQueryInfo());
-					executor.execute(join);
-					return join;
+					return executeLeftJoin(scheduler, leftIter, leftJoin, bindings, leftJoin.getQueryInfo());
 				} else {
 					Set<String> problemVarsClone = new HashSet<>(problemVars);
 					problemVarsClone.retainAll(bindings.getBindingNames());
@@ -839,6 +836,21 @@ protected abstract CloseableIteration<BindingSet> executeJoin(
 			CloseableIteration<BindingSet> leftIter, TupleExpr rightArg,
 			Set<String> joinVariables, BindingSet bindings, QueryInfo queryInfo) throws QueryEvaluationException;
 
+	/**
+	 * Execute the left join in a separate thread using some join executor.
+	 *
+	 * @param joinScheduler
+	 * @param leftIter
+	 * @param leftJoin
+	 * @param bindings
+	 * @return the result
+	 * @throws QueryEvaluationException
+	 */
+	protected abstract CloseableIteration<BindingSet> executeLeftJoin(
+			ControlledWorkerScheduler<BindingSet> joinScheduler,
+			CloseableIteration<BindingSet> leftIter, LeftJoin leftJoin,
+			BindingSet bindings, QueryInfo queryInfo) throws QueryEvaluationException;
+
 	public abstract CloseableIteration<BindingSet> evaluateExclusiveGroup(
 			ExclusiveGroup group, BindingSet bindings)
 			throws RepositoryException, MalformedQueryException, QueryEvaluationException;

diff --git a/...tion/src/main/java/org/eclipse/rdf4j/federated/evaluation/SailFederationEvalStrategy.java b/...tion/src/main/java/org/eclipse/rdf4j/federated/evaluation/SailFederationEvalStrategy.java
@@ -27,11 +27,13 @@
 import org.eclipse.rdf4j.federated.evaluation.iterator.FilteringIteration;
 import org.eclipse.rdf4j.federated.evaluation.iterator.GroupedCheckConversionIteration;
 import org.eclipse.rdf4j.federated.evaluation.join.ControlledWorkerJoin;
+import org.eclipse.rdf4j.federated.evaluation.join.ControlledWorkerLeftJoin;
 import org.eclipse.rdf4j.federated.structures.QueryInfo;
 import org.eclipse.rdf4j.federated.util.QueryAlgebraUtil;
 import org.eclipse.rdf4j.query.BindingSet;
 import org.eclipse.rdf4j.query.MalformedQueryException;
 import org.eclipse.rdf4j.query.QueryEvaluationException;
+import org.eclipse.rdf4j.query.algebra.LeftJoin;
 import org.eclipse.rdf4j.query.algebra.StatementPattern;
 import org.eclipse.rdf4j.query.algebra.TupleExpr;
 import org.eclipse.rdf4j.repository.RepositoryException;
@@ -119,6 +121,16 @@ public CloseableIteration<BindingSet> executeJoin(
 		return join;
 	}
 
+	@Override
+	protected CloseableIteration<BindingSet> executeLeftJoin(ControlledWorkerScheduler<BindingSet> joinScheduler,
+			CloseableIteration<BindingSet> leftIter, LeftJoin leftJoin, BindingSet bindings, QueryInfo queryInfo)
+			throws QueryEvaluationException {
+		ControlledWorkerLeftJoin join = new ControlledWorkerLeftJoin(joinScheduler, this,
+				leftIter, leftJoin, bindings, queryInfo);
+		executor.execute(join);
+		return join;
+	}
+
 	@Override
 	public CloseableIteration<BindingSet> evaluateExclusiveGroup(
 			ExclusiveGroup group, BindingSet bindings)

diff --git a/...on/src/main/java/org/eclipse/rdf4j/federated/evaluation/SparqlFederationEvalStrategy.java b/...on/src/main/java/org/eclipse/rdf4j/federated/evaluation/SparqlFederationEvalStrategy.java
@@ -33,6 +33,7 @@
 import org.eclipse.rdf4j.federated.evaluation.iterator.SingleBindingSetIteration;
 import org.eclipse.rdf4j.federated.evaluation.join.ControlledWorkerBindJoin;
 import org.eclipse.rdf4j.federated.evaluation.join.ControlledWorkerJoin;
+import org.eclipse.rdf4j.federated.evaluation.join.ControlledWorkerLeftJoin;
 import org.eclipse.rdf4j.federated.evaluation.join.JoinExecutorBase;
 import org.eclipse.rdf4j.federated.exception.ExceptionUtil;
 import org.eclipse.rdf4j.federated.exception.IllegalQueryException;
@@ -41,6 +42,7 @@
 import org.eclipse.rdf4j.query.BindingSet;
 import org.eclipse.rdf4j.query.MalformedQueryException;
 import org.eclipse.rdf4j.query.QueryEvaluationException;
+import org.eclipse.rdf4j.query.algebra.LeftJoin;
 import org.eclipse.rdf4j.query.algebra.StatementPattern;
 import org.eclipse.rdf4j.query.algebra.TupleExpr;
 import org.eclipse.rdf4j.repository.RepositoryException;
@@ -201,6 +203,16 @@ public CloseableIteration<BindingSet> executeJoin(
 		return join;
 	}
 
+	@Override
+	protected CloseableIteration<BindingSet> executeLeftJoin(ControlledWorkerScheduler<BindingSet> joinScheduler,
+			CloseableIteration<BindingSet> leftIter, LeftJoin leftJoin, BindingSet bindings, QueryInfo queryInfo)
+			throws QueryEvaluationException {
+		ControlledWorkerLeftJoin join = new ControlledWorkerLeftJoin(joinScheduler, this,
+				leftIter, leftJoin, bindings, queryInfo);
+		executor.execute(join);
+		return join;
+	}
+
 	@Override
 	public CloseableIteration<BindingSet> evaluateExclusiveGroup(
 			ExclusiveGroup group, BindingSet bindings) throws RepositoryException,