GH-4983: fine-tune bind join implementation in FedX engine

This change applies fine-tuning to the bind join implementation on
certain parameters.

Note that the original implementations and benchmarks were done when
there was no SPARQL 1.1 around, and FedX was using UNION constructs for
the bind join implementation. Since a few year back it is using VALUES
clauses.

Changes:

- default bind join size is set to 25 (also reflected in documentation,
some unit tests require adjustments)
- no longer send small hard-coded sub-queries of size 3 for the first
requests and instead rely on configured bind join size
- no longer send the first request in a non-bound way by default (Note:
kept as internal configuration parameter in the implementation to allow
modifications in extensions)
- support to influence the bind join size in customizations

Author: aschwarte10

site/content/documentation/programming/federation.md

@@ -303,7 +303,7 @@ FedX provides various means for configuration. Configuration settings can be def
 |joinWorkerThreads | The number of join worker threads for parallelization, default _20_ |
 |unionWorkerThreads | The number of union worker threads for parallelization, default _20_ |
 |leftJoinWorkerThreads | The number of left join worker threads for parallelization, default _10_ |
-|boundJoinBlockSize | Block size for bound joins, default _15_ |
+|boundJoinBlockSize | Block size for bound joins, default _25_ |
 |enforceMaxQueryTime | Max query time in seconds, 0 to disable, default _30_ |
 |enableServiceAsBoundJoin | Flag for evaluating a SERVICE expression (contacting non-federation members) using vectored evaluation, default _true_. For today's endpoints it is more efficient to disable vectored evaluation of SERVICE |
 |includeInferredDefault | whether include inferred statements should be considered, default _true_ |

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

@@ -40,7 +40,7 @@ public class FedXConfig {
 
 	private int leftJoinWorkerThreads = 10;
 
-	private int boundJoinBlockSize = 15;
+	private int boundJoinBlockSize = 25;
 
 	private int enforceMaxQueryTime = 30;
 

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

@@ -34,7 +34,7 @@
 
 /**
  * Execute the nested loop join in an asynchronous fashion, using grouped requests, i.e. group bindings into one SPARQL
- * request using the UNION operator.
+ * request using a VALUES clause.
  *
  * The number of concurrent threads is controlled by a {@link ControlledWorkerScheduler} which works according to the
  * FIFO principle and uses worker threads.
@@ -49,13 +49,22 @@ public class ControlledWorkerBoundJoin extends ControlledWorkerJoin {
 
 	private static final Logger log = LoggerFactory.getLogger(ControlledWorkerBoundJoin.class);
 
+	/**
+	 * Whether to submit the first intermediate result immediately in a non-bound way
+	 */
+	private boolean submitFirstResultImmediately = false;
+
 	public ControlledWorkerBoundJoin(ControlledWorkerScheduler<BindingSet> scheduler, FederationEvalStrategy strategy,
 			CloseableIteration<BindingSet> leftIter,
 			TupleExpr rightArg, BindingSet bindings, QueryInfo queryInfo)
 			throws QueryEvaluationException {
 		super(scheduler, strategy, leftIter, rightArg, bindings, queryInfo);
 	}
 
+	protected void setSubmitFirstResultImmediately(boolean flag) {
+		this.submitFirstResultImmediately = flag;
+	}
+
 	@Override
 	protected void handleBindings() throws Exception {
 		if (!(canApplyVectoredEvaluation(rightArg))) {
@@ -73,27 +82,17 @@ protected void handleBindings() throws Exception {
 		TaskCreator taskCreator = null;
 		Phaser currentPhaser = phaser;
 
-		// first item is always sent in a non-bound way
-		if (!isClosed() && leftIter.hasNext()) {
-			BindingSet b = leftIter.next();
-			totalBindings++;
-			if (expr instanceof StatementTupleExpr) {
-				StatementTupleExpr stmt = (StatementTupleExpr) expr;
-				if (stmt.hasFreeVarsFor(b)) {
-					taskCreator = new BoundJoinTaskCreator(strategy, stmt);
-				} else {
-					expr = new CheckStatementPattern(stmt, queryInfo);
-					taskCreator = new CheckJoinTaskCreator(strategy, (CheckStatementPattern) expr);
-				}
-			} else if (expr instanceof FedXService) {
-				taskCreator = new FedXServiceJoinTaskCreator(strategy, (FedXService) expr);
-			} else {
-				throw new RuntimeException("Expr is of unexpected type: " + expr.getClass().getCanonicalName()
-						+ ". Please report this problem.");
+		if (submitFirstResultImmediately) {
+			// first item is always sent in a non-bound way
+			if (!isClosed() && leftIter.hasNext()) {
+				BindingSet b = leftIter.next();
+				totalBindings++;
+				taskCreator = determineTaskCreator(expr, b);
+				currentPhaser.register();
+				scheduler.schedule(
+						new ParallelJoinTask(new PhaserHandlingParallelExecutor(this, currentPhaser), strategy, expr,
+								b));
 			}
-			currentPhaser.register();
-			scheduler.schedule(
-					new ParallelJoinTask(new PhaserHandlingParallelExecutor(this, currentPhaser), strategy, expr, b));
 		}
 
 		int nBindings;
@@ -106,27 +105,18 @@ protected void handleBindings() throws Exception {
 				currentPhaser = new Phaser(currentPhaser);
 			}
 
-			/*
-			 * XXX idea:
-			 *
-			 * make nBindings dependent on the number of intermediate results of the left argument.
-			 *
-			 * If many intermediate results, increase the number of bindings. This will result in less remote SPARQL
-			 * requests.
-			 *
-			 */
-
-			if (totalBindings > 10) {
-				nBindings = nBindingsCfg;
-			} else {
-				nBindings = 3;
-			}
+			// determine the bind join block size
+			nBindings = getNextBindJoinSize(nBindingsCfg, totalBindings);
 
 			bindings = new ArrayList<>(nBindings);
 
 			int count = 0;
 			while (!isClosed() && count < nBindings && leftIter.hasNext()) {
-				bindings.add(leftIter.next());
+				var bs = leftIter.next();
+				if (taskCreator == null) {
+					taskCreator = determineTaskCreator(expr, bs);
+				}
+				bindings.add(bs);
 				count++;
 			}
 
@@ -157,6 +147,47 @@ public void handleClose() throws QueryEvaluationException {
 		}
 	}
 
+	protected TaskCreator determineTaskCreator(TupleExpr expr, BindingSet bs) {
+		final TaskCreator taskCreator;
+		if (expr instanceof StatementTupleExpr) {
+			StatementTupleExpr stmt = (StatementTupleExpr) expr;
+			if (stmt.hasFreeVarsFor(bs)) {
+				taskCreator = new BoundJoinTaskCreator(strategy, stmt);
+			} else {
+				expr = new CheckStatementPattern(stmt, queryInfo);
+				taskCreator = new CheckJoinTaskCreator(strategy, (CheckStatementPattern) expr);
+			}
+		} else if (expr instanceof FedXService) {
+			taskCreator = new FedXServiceJoinTaskCreator(strategy, (FedXService) expr);
+		} else {
+			throw new RuntimeException("Expr is of unexpected type: " + expr.getClass().getCanonicalName()
+					+ ". Please report this problem.");
+		}
+		return taskCreator;
+	}
+
+	/**
+	 * Return the size of the next bind join block.
+	 *
+	 * @param configuredBindJoinSize the configured bind join size
+	 * @param totalBindings          the current process bindings from the intermediate result set
+	 * @return
+	 */
+	protected int getNextBindJoinSize(int configuredBindJoinSize, int totalBindings) {
+
+		/*
+		 * XXX idea:
+		 *
+		 * make nBindings dependent on the number of intermediate results of the left argument.
+		 *
+		 * If many intermediate results, increase the number of bindings. This will result in less remote SPARQL
+		 * requests.
+		 *
+		 */
+
+		return configuredBindJoinSize;
+	}
+
 	/**
 	 * Returns true if the vectored evaluation can be applied for the join argument, i.e. there is no fallback to
 	 * {@link ControlledWorkerJoin#handleBindings()}. This is

tools/federation/src/test/java/org/eclipse/rdf4j/federated/BoundJoinTests.java

@@ -37,7 +37,10 @@ public void testBoundJoin_FailingEndpoint() throws Exception {
 		/* test a simple bound join */
 		prepareTest(Arrays.asList("/tests/data/data1.ttl", "/tests/data/data2.ttl"));
 
-		repoSettings(2).setFailAfter(5);
+		// specifically reduce the bind join size for this test to run into "fail After"
+		fedxRule.getFederationContext().getConfig().withBoundJoinBlockSize(2);
+
+		repoSettings(2).setFailAfter(5); // fail after 5 remote requests on repo 2
 		Assertions.assertThrows(QueryEvaluationException.class, () -> {
 			execute("/tests/boundjoin/query01.rq", "/tests/boundjoin/query01.srx", false, true);
 		});

tools/federation/src/test/java/org/eclipse/rdf4j/federated/ServiceTests.java

@@ -241,6 +241,8 @@ protected FederatedService createService(String serviceUrl) throws QueryEvaluati
 		repo.setFederatedServiceResolver(serviceResolver);
 		repo.init();
 
+		fedxRule.getFederationContext().getConfig().withBoundJoinBlockSize(5);
+
 		/*
 		 * test select query retrieving all persons from endpoint 1 (SERVICE), endpoint not part of federation =>
 		 * evaluate using externally provided service resolver endpoint1 is reachable as
@@ -271,12 +273,11 @@ protected FederatedService createService(String serviceUrl) throws QueryEvaluati
 			Assertions.assertEquals(expected, res.stream().map(b -> b.getValue("output")).collect(Collectors.toSet()));
 		}
 
-		// first binding is evaluated using regular service, then we have groups of 4 groups of three bindings and 3
-		// groups with 15
+		// all requests are executed in bind-join with constant size
+		// for this test bind join size is set to 5, hence we see 10 bind join requests
 		TestSparqlFederatedService tfs = ((TestSparqlFederatedService) serviceResolver
 				.getService("http://localhost:18080/repositories/endpoint1"));
-		Assertions.assertEquals(1, tfs.serviceRequestCount.get());
-		Assertions.assertEquals(7, tfs.boundJoinRequestCount.get());
+		Assertions.assertEquals(10, tfs.boundJoinRequestCount.get());
 	}
 
 	@Test

tools/federation/src/test/java/org/eclipse/rdf4j/federated/repository/FedXRepositoryConfigTest.java

@@ -262,7 +262,7 @@ public void testExportWithEmptyConfig() throws Exception {
 			assertThat(
 					Models.objectLiteral(
 							export.getStatements(configNode, FedXRepositoryConfig.CONFIG_BOUND_JOIN_BLOCK_SIZE, null)))
-					.hasValueSatisfying(v -> assertThat(v.intValue()).isEqualTo(15));
+					.hasValueSatisfying(v -> assertThat(v.intValue()).isEqualTo(25));
 			assertThat(
 					Models.objectLiteral(
 							export.getStatements(configNode, FedXRepositoryConfig.CONFIG_ENFORCE_MAX_QUERY_TIME, null)))