GH-4983: fine-tune bind join implementation in FedX engine (#4984)

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)))