GH-5718: merge FedX FederationEvalStrategy variants

There is no need to maintain different federation evaluation strategy
implementations as both are very close in their implementation. We now
use the implementation logic of the previous SPARQL evaluation strategy
as default and incorporate it right away inside FederationEvalStrategy.

Author: aschwarte10

tools/federation/src/main/java/org/eclipse/rdf4j/federated/FedXConfig.java

@@ -12,7 +12,6 @@
 
 import java.util.Optional;
 
-import org.eclipse.rdf4j.collection.factory.api.CollectionFactory;
 import org.eclipse.rdf4j.federated.cache.SourceSelectionCache;
 import org.eclipse.rdf4j.federated.cache.SourceSelectionCacheFactory;
 import org.eclipse.rdf4j.federated.cache.SourceSelectionMemoryCache;

tools/federation/src/main/java/org/eclipse/rdf4j/federated/FederationManager.java

@@ -21,8 +21,6 @@
 import org.eclipse.rdf4j.federated.endpoint.Endpoint;
 import org.eclipse.rdf4j.federated.endpoint.EndpointClassification;
 import org.eclipse.rdf4j.federated.evaluation.FederationEvaluationStrategyFactory;
-import org.eclipse.rdf4j.federated.evaluation.SailFederationEvalStrategy;
-import org.eclipse.rdf4j.federated.evaluation.SparqlFederationEvalStrategy;
 import org.eclipse.rdf4j.federated.evaluation.concurrent.ControlledWorkerScheduler;
 import org.eclipse.rdf4j.federated.evaluation.concurrent.NamingThreadFactory;
 import org.eclipse.rdf4j.federated.evaluation.concurrent.Scheduler;
@@ -107,7 +105,6 @@ public void init(FedX federation, FederationContext federationContext) {
 		FederationEvaluationStrategyFactory strategyFactory = federation.getFederationEvaluationStrategyFactory();
 		strategyFactory.setFederationType(federationType);
 		strategyFactory.setFederationContext(federationContext);
-		strategyFactory.setCollectionFactory(federation.getCollectionFactory());
 		return strategyFactory;
 	}
 

tools/federation/src/main/java/org/eclipse/rdf4j/federated/evaluation/FederationEvalStrategy.java

@@ -25,7 +25,7 @@
 import org.eclipse.rdf4j.federated.FedX;
 import org.eclipse.rdf4j.federated.FedXConfig;
 import org.eclipse.rdf4j.federated.FederationContext;
-import org.eclipse.rdf4j.federated.algebra.CheckStatementPattern;
+import org.eclipse.rdf4j.federated.algebra.BoundJoinTupleExpr;
 import org.eclipse.rdf4j.federated.algebra.ConjunctiveFilterExpr;
 import org.eclipse.rdf4j.federated.algebra.EmptyResult;
 import org.eclipse.rdf4j.federated.algebra.EmptyStatementPattern;
@@ -55,12 +55,17 @@
 import org.eclipse.rdf4j.federated.evaluation.concurrent.ControlledWorkerScheduler;
 import org.eclipse.rdf4j.federated.evaluation.concurrent.ParallelServiceExecutor;
 import org.eclipse.rdf4j.federated.evaluation.iterator.BindLeftJoinIteration;
+import org.eclipse.rdf4j.federated.evaluation.iterator.BoundJoinVALUESConversionIteration;
 import org.eclipse.rdf4j.federated.evaluation.iterator.FedXPathIteration;
 import org.eclipse.rdf4j.federated.evaluation.iterator.FederatedDescribeIteration;
 import org.eclipse.rdf4j.federated.evaluation.iterator.FilteringIteration;
+import org.eclipse.rdf4j.federated.evaluation.iterator.InsertBindingsIteration;
 import org.eclipse.rdf4j.federated.evaluation.iterator.SingleBindingSetIteration;
 import org.eclipse.rdf4j.federated.evaluation.join.ControlledWorkerBindJoin;
+import org.eclipse.rdf4j.federated.evaluation.join.ControlledWorkerBindLeftJoin;
 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.evaluation.union.ControlledWorkerUnion;
 import org.eclipse.rdf4j.federated.evaluation.union.ParallelGetStatementsTask;
 import org.eclipse.rdf4j.federated.evaluation.union.ParallelPreparedAlgebraUnionTask;
@@ -131,15 +136,20 @@
 import org.slf4j.LoggerFactory;
 
 /**
- * Base class for the Evaluation strategies.
+ * Implementation of a federation evaluation strategy which provides some special optimizations for SPARQL (remote)
+ * endpoints. The most important optimization is to used prepared SPARQL Queries that are already created using Strings.
+ * <p>
+ * Joins are executed using {@link ControlledWorkerBindJoin}.
+ * </p>
+ * <p>
+ * This implementation uses the SPARQL 1.1 VALUES operator for the bound-join evaluation
+ * </p>
+ * s
  *
  * @author Andreas Schwarte
  *
- * @see SailFederationEvalStrategy
- * @see SparqlFederationEvalStrategy
- *
  */
-public abstract class FederationEvalStrategy extends StrictEvaluationStrategy {
+public class FederationEvalStrategy extends StrictEvaluationStrategy {
 
 	private static final Logger log = LoggerFactory.getLogger(FederationEvalStrategy.class);
 
@@ -904,10 +914,35 @@ public QueryEvaluationStep prepareNaryUnion(NUnion union, QueryEvaluationContext
 	 * @return the result
 	 * @throws QueryEvaluationException
 	 */
-	protected abstract CloseableIteration<BindingSet> executeJoin(
+	public CloseableIteration<BindingSet> executeJoin(
 			ControlledWorkerScheduler<BindingSet> joinScheduler,
-			CloseableIteration<BindingSet> leftIter, TupleExpr rightArg,
-			Set<String> joinVariables, BindingSet bindings, QueryInfo queryInfo) throws QueryEvaluationException;
+			CloseableIteration<BindingSet> leftIter,
+			TupleExpr rightArg, Set<String> joinVars, BindingSet bindings, QueryInfo queryInfo)
+			throws QueryEvaluationException {
+
+		// determine if we can execute the expr as bind join
+		boolean executeAsBindJoin = false;
+		if (rightArg instanceof BoundJoinTupleExpr) {
+			if (rightArg instanceof FedXService) {
+				executeAsBindJoin = queryInfo.getFederationContext().getConfig().getEnableServiceAsBoundJoin();
+			} else {
+				executeAsBindJoin = true;
+			}
+		}
+
+		JoinExecutorBase<BindingSet> join;
+		if (executeAsBindJoin) {
+			join = new ControlledWorkerBindJoin(joinScheduler, this, leftIter, rightArg,
+					bindings, queryInfo);
+		} else {
+			join = new ControlledWorkerJoin(joinScheduler, this, leftIter, rightArg, bindings,
+					queryInfo);
+		}
+
+		join.setJoinVars(joinVars);
+		executor.execute(join);
+		return join;
+	}
 
 	/**
 	 * Execute the left join in a separate thread using some join executor.
@@ -919,14 +954,63 @@ protected abstract CloseableIteration<BindingSet> executeJoin(
 	 * @return the result
 	 * @throws QueryEvaluationException
 	 */
-	protected abstract CloseableIteration<BindingSet> executeLeftJoin(
-			ControlledWorkerScheduler<BindingSet> joinScheduler,
-			CloseableIteration<BindingSet> leftIter, LeftJoin leftJoin,
-			BindingSet bindings, QueryInfo queryInfo) throws QueryEvaluationException;
+	protected CloseableIteration<BindingSet> executeLeftJoin(ControlledWorkerScheduler<BindingSet> joinScheduler,
+			CloseableIteration<BindingSet> leftIter, LeftJoin leftJoin, BindingSet bindings, QueryInfo queryInfo)
+			throws QueryEvaluationException {
+
+		var rightArg = leftJoin.getRightArg();
+		var fedxConfig = queryInfo.getFederationContext().getConfig();
+
+		// determine if we can execute the expr as bind join
+		boolean executeAsBindJoin = false;
+		if (fedxConfig.isEnableOptionalAsBindJoin() && rightArg instanceof BoundJoinTupleExpr) {
+			if (rightArg instanceof FedXService) {
+				executeAsBindJoin = false;
+			} else {
+				executeAsBindJoin = true;
+			}
+		}
 
-	public abstract CloseableIteration<BindingSet> evaluateExclusiveGroup(
-			ExclusiveGroup group, BindingSet bindings)
-			throws RepositoryException, MalformedQueryException, QueryEvaluationException;
+		JoinExecutorBase<BindingSet> join;
+		if (executeAsBindJoin) {
+			join = new ControlledWorkerBindLeftJoin(joinScheduler, this, leftIter, rightArg,
+					bindings, queryInfo);
+		} else {
+			join = new ControlledWorkerLeftJoin(joinScheduler, this,
+					leftIter, leftJoin, bindings, queryInfo);
+		}
+
+		executor.execute(join);
+		return join;
+	}
+
+	public CloseableIteration<BindingSet> evaluateExclusiveGroup(
+			ExclusiveGroup group, BindingSet bindings) throws RepositoryException,
+			MalformedQueryException, QueryEvaluationException {
+
+		TripleSource tripleSource = group.getOwnedEndpoint().getTripleSource();
+		AtomicBoolean isEvaluated = new AtomicBoolean(false);
+
+		try {
+			String preparedQuery = QueryStringUtil.selectQueryString(group, bindings, group.getFilterExpr(),
+					isEvaluated, group.getQueryInfo().getDataset());
+			return tripleSource.getStatements(preparedQuery, bindings,
+					(isEvaluated.get() ? null : group.getFilterExpr()), group.getQueryInfo());
+		} catch (IllegalQueryException e) {
+			/* no projection vars, e.g. local vars only, can occur in joins */
+			if (tripleSource.hasStatements(group, bindings)) {
+				CloseableIteration<BindingSet> res = new SingleBindingSetIteration(bindings);
+				if (group.getBoundFilters() != null) {
+					// make sure to insert any values from FILTER expressions that are directly
+					// bound in this expression
+					res = new InsertBindingsIteration(res, group.getBoundFilters());
+				}
+				return res;
+			}
+			return new EmptyIteration<>();
+		}
+
+	}
 
 	/**
 	 * Evaluate an {@link ExclusiveTupleExpr}. The default implementation converts the given expression to a SELECT
@@ -993,8 +1077,51 @@ protected QueryEvaluationStep prepareExclusiveTupleExpr(
 	 * @return the result iteration
 	 * @throws QueryEvaluationException
 	 */
-	public abstract CloseableIteration<BindingSet> evaluateBoundJoinStatementPattern(
-			StatementTupleExpr stmt, final List<BindingSet> bindings) throws QueryEvaluationException;
+	public CloseableIteration<BindingSet> evaluateBoundJoinStatementPattern(
+			StatementTupleExpr stmt, List<BindingSet> bindings)
+			throws QueryEvaluationException {
+
+		// we can omit the bound join handling
+		if (bindings.size() == 1) {
+			return evaluate(stmt, bindings.get(0));
+		}
+
+		FilterValueExpr filterExpr = null;
+		if (stmt instanceof FilterTuple) {
+			filterExpr = ((FilterTuple) stmt).getFilterExpr();
+		}
+
+		AtomicBoolean isEvaluated = new AtomicBoolean(false);
+		String preparedQuery = QueryStringUtil.selectQueryStringBoundJoinVALUES((StatementPattern) stmt, bindings,
+				filterExpr, isEvaluated, stmt.getQueryInfo().getDataset());
+
+		CloseableIteration<BindingSet> result = null;
+		try {
+			result = evaluateAtStatementSources(preparedQuery, stmt.getStatementSources(), stmt.getQueryInfo());
+
+			// apply filter and/or convert to original bindings
+			if (filterExpr != null && !isEvaluated.get()) {
+				result = new BoundJoinVALUESConversionIteration(result, bindings); // apply conversion
+				result = new FilteringIteration(filterExpr, result, this); // apply filter
+				if (!result.hasNext()) {
+					result.close();
+					return new EmptyIteration<>();
+				}
+			} else {
+				result = new BoundJoinVALUESConversionIteration(result, bindings);
+			}
+
+			return result;
+		} catch (Throwable t) {
+			if (result != null) {
+				result.close();
+			}
+			if (t instanceof InterruptedException) {
+				Thread.currentThread().interrupt();
+			}
+			throw ExceptionUtil.toQueryEvaluationException(t);
+		}
+	}
 
 	/**
 	 * Evaluate the left bind join for the given {@link StatementTupleExpr} and bindings at the relevant endpoints.

tools/federation/src/main/java/org/eclipse/rdf4j/federated/evaluation/FederationEvaluationStrategyFactory.java

@@ -70,19 +70,6 @@ public FederationEvalStrategy createEvaluationStrategy(Dataset dataset, TripleSo
 		// Note: currently dataset, triplesource and statistics are explicitly ignored
 		// in the federation
 
-		switch (federationType) {
-		case LOCAL: {
-			SailFederationEvalStrategy evalStrategy = new SailFederationEvalStrategy(federationContext);
-			evalStrategy.setCollectionFactory(collectionFactorySupplier);
-			return evalStrategy;
-		}
-		case REMOTE:
-		case HYBRID:
-		default: {
-			SparqlFederationEvalStrategy evalStrategy = new SparqlFederationEvalStrategy(federationContext);
-			evalStrategy.setCollectionFactory(collectionFactorySupplier);
-			return evalStrategy;
-		}
-		}
+		return new FederationEvalStrategy(federationContext);
 	}
 }