GH-4952 - Introduce FedXConfig overrides for FedXRepositoryConfig (#4953)

Author: aschwarte10

site/content/documentation/programming/federation.md

@@ -299,17 +299,50 @@ FedX provides various means for configuration. Configuration settings can be def
 |Property | Description |
 |---------|-------------|
 |prefixDeclarations | Path to prefix declarations file, see [PREFIX Declarations](#prefix-declarations) |
-|cacheLocation | Location where the memory cache gets persisted at shutdown, default _cache.db_ |
+|sourceSelectionCacheSpec | Cache specification for the `SourceSelectionMemoryCache`, default _maximumSize=1000,expireAfterWrite=6h_ |
 |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_ |
 |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_ |
+|consumingIterationMax | the max number of results to be consumed by `ConsumingIteration`, default _1000_ |
 |debugQueryPlan | Print the optimized query execution plan to stdout, default _false_ |
 |enableMonitoring | Flag to enable/disable monitoring features, default _false_ |
 |logQueryPlan | Flag to enable/disable query plan logging via Java class _QueryPlanLog_, default _false_ |
 |logQueries | Flag to enable/disable query logging via _QueryLog_, default _false_. The _QueryLog_ facility allows to log all queries to a file |
 
+#### Overriding via configuration template
+
+The aforementioned properties can also be set using a configuration template, via the `fedx:config` property, e.g.:
+
+```turtle
+@prefix rdfs: <http://www.w3.org/2000/01/rdf-schema#>.
+@prefix rep: <http://www.openrdf.org/config/repository#>.
+@prefix config: <tag:rdf4j.org,2023:config/>.
+@prefix fedx: <http://rdf4j.org/config/federation#>.
+
+[] a rep:Repository ;
+ rep:repositoryImpl [
+   rep:repositoryType "fedx:FedXRepository" ;
+   fedx:member [
+      fedx:store "ResolvableRepository" ;
+      fedx:repositoryName "endpoint1"
+   ],
+   [
+      fedx:store "ResolvableRepository" ;
+      fedx:repositoryName "endpoint2"
+   ]
+   fedx:config [
+      fedx:sourceSelectionCacheSpec "maximumSize=0" ;
+      fedx:enforceMaxQueryTime 30 ;
+   ]
+ ];
+ rep:repositoryID "fedx" ;
+ rdfs:label "FedX Federation" .
+```
+
 ### Query timeouts
 
 FedX supports to define the maximum execution time for a query. This can be set on query level `Query#setMaxExecutionTime`or globally using the FedX config setting _enforceMaxQueryTime_.

tools/federation/src/main/java/org/eclipse/rdf4j/federated/repository/FedXRepositoryConfig.java

@@ -14,6 +14,7 @@
 
 import org.eclipse.rdf4j.federated.FedXConfig;
 import org.eclipse.rdf4j.federated.util.Vocabulary.FEDX;
+import org.eclipse.rdf4j.model.BNode;
 import org.eclipse.rdf4j.model.IRI;
 import org.eclipse.rdf4j.model.Model;
 import org.eclipse.rdf4j.model.Resource;
@@ -23,6 +24,7 @@
 import org.eclipse.rdf4j.model.impl.TreeModel;
 import org.eclipse.rdf4j.model.util.ModelException;
 import org.eclipse.rdf4j.model.util.Models;
+import org.eclipse.rdf4j.model.util.Values;
 import org.eclipse.rdf4j.repository.config.AbstractRepositoryImplConfig;
 import org.eclipse.rdf4j.repository.config.RepositoryConfigException;
 import org.eclipse.rdf4j.repository.config.RepositoryImplConfig;
@@ -60,6 +62,12 @@
  *   # optionally define data config
  *   #fedx:fedxConfig "fedxConfig.prop" ;
  *   fedx:dataConfig "dataConfig.ttl" ;
+ *
+ *   # optionally define FedXConfig overrides
+ *   fedx:config [
+ *      fedx:sourceSelectionCacheSpec "maximumSize=0" ;
+ *      fedx:enforceMaxQueryTime 30 ;
+ *   ]
  * ];
  * rep:repositoryID "fedx" ;
  * rdfs:label "FedX Federation" .
@@ -87,11 +95,86 @@ public class FedXRepositoryConfig extends AbstractRepositoryImplConfig {
 	 */
 	public static final IRI DATA_CONFIG = vf.createIRI(NAMESPACE, "dataConfig");
 
+	/**
+	 * IRI of the property pointing to the {@link FedXConfig}
+	 */
+	public static final IRI FEDX_CONFIG = vf.createIRI(NAMESPACE, "config");
+
 	/**
 	 * IRI of the property pointing to a federation member node
 	 */
 	public static final IRI MEMBER = vf.createIRI(NAMESPACE, "member");
 
+	/**
+	 * IRI of the property populating {@link FedXConfig#getJoinWorkerThreads()}
+	 */
+	public static final IRI CONFIG_JOIN_WORKER_THREADS = vf.createIRI(NAMESPACE, "joinWorkerThreads");
+
+	/**
+	 * IRI of the property populating {@link FedXConfig#getUnionWorkerThreads()}
+	 */
+	public static final IRI CONFIG_UNION_WORKER_THREADS = vf.createIRI(NAMESPACE, "unionWorkerThreads");
+
+	/**
+	 * IRI of the property populating {@link FedXConfig#getLeftJoinWorkerThreads()}
+	 */
+	public static final IRI CONFIG_LEFT_JOIN_WORKER_THREADS = vf.createIRI(NAMESPACE, "leftJoinWorkerThreads");
+
+	/**
+	 * IRI of the property populating {@link FedXConfig#getBoundJoinBlockSize()}
+	 */
+	public static final IRI CONFIG_BOUND_JOIN_BLOCK_SIZE = vf.createIRI(NAMESPACE, "boundJoinBlockSize");
+
+	/**
+	 * IRI of the property populating {@link FedXConfig#getEnforceMaxQueryTime()}
+	 */
+	public static final IRI CONFIG_ENFORCE_MAX_QUERY_TIME = vf.createIRI(NAMESPACE, "enforceMaxQueryTime");
+
+	/**
+	 * IRI of the property populating {@link FedXConfig#getEnableServiceAsBoundJoin()}
+	 */
+	public static final IRI CONFIG_ENABLE_SERVICE_AS_BOUND_JOIN = vf.createIRI(NAMESPACE, "enableServiceAsBoundJoin");
+
+	/**
+	 * IRI of the property populating {@link FedXConfig#isEnableMonitoring()}
+	 */
+	public static final IRI CONFIG_ENABLE_MONITORING = vf.createIRI(NAMESPACE, "enableMonitoring");
+
+	/**
+	 * IRI of the property populating {@link FedXConfig#isLogQueryPlan()}
+	 */
+	public static final IRI CONFIG_LOG_QUERY_PLAN = vf.createIRI(NAMESPACE, "logQueryPlan");
+
+	/**
+	 * IRI of the property populating {@link FedXConfig#isLogQueries()}
+	 */
+	public static final IRI CONFIG_LOG_QUERIES = vf.createIRI(NAMESPACE, "logQueries");
+
+	/**
+	 * IRI of the property populating {@link FedXConfig#isDebugQueryPlan()}
+	 */
+	public static final IRI CONFIG_DEBUG_QUERY_PLAN = vf.createIRI(NAMESPACE, "debugQueryPlan");
+
+	/**
+	 * IRI of the property populating {@link FedXConfig#getIncludeInferredDefault()}
+	 */
+	public static final IRI CONFIG_INCLUDE_INFERRED_DEFAULT = vf.createIRI(NAMESPACE, "includeInferredDefault");
+
+	/**
+	 * IRI of the property populating {@link FedXConfig#getSourceSelectionCacheSpec()}
+	 */
+	public static final IRI CONFIG_SOURCE_SELECTION_CACHE_SPEC = vf.createIRI(NAMESPACE, "sourceSelectionCacheSpec");
+
+	/**
+	 * IRI of the property populating {@link FedXConfig#getPrefixDeclarations()}
+	 */
+	public static final IRI CONFIG_PREFIX_DECLARATIONS = vf.createIRI(NAMESPACE, "prefixDeclarations");
+
+	/**
+	 * IRI of the property populating {@link FedXConfig#getConsumingIterationMax()}
+	 */
+	public static final IRI CONFIG_CONSUMING_ITERATION_MAX = vf.createIRI(NAMESPACE, "consumingIterationMax");
+
 	/**
 	 * the location of the data configuration
 	 */
@@ -152,6 +235,8 @@ public Resource export(Model m) {
 			m.add(implNode, DATA_CONFIG, vf.createLiteral(getDataConfig()));
 		}
 
+		exportFedXConfig(m, implNode);
+
 		if (getMembers() != null) {
 
 			Model members = getMembers();
@@ -187,6 +272,8 @@ public void parse(Model m, Resource implNode) throws RepositoryConfigException {
 			Models.objectLiteral(m.getStatements(implNode, DATA_CONFIG, null))
 					.ifPresent(value -> setDataConfig(value.stringValue()));
 
+			parseFedXConfig(m, implNode);
+
 			Set<Value> memberNodes = m.filter(implNode, MEMBER, null).objects();
 			if (!memberNodes.isEmpty()) {
 				Model members = new TreeModel();
@@ -205,4 +292,120 @@ public void parse(Model m, Resource implNode) throws RepositoryConfigException {
 			throw new RepositoryConfigException(e.getMessage(), e);
 		}
 	}
+
+	/**
+	 * Updates the container {@link FedXConfig} instance with properties from the supplied model. It is up to the caller
+	 * to retrieve configuration from {@link #FEDX_CONFIG} as well as to initialise the parsed configuration (via
+	 * {@link #setConfig(FedXConfig)}) since it can be null.
+	 *
+	 * @param m        the model from which to read configuration properties
+	 * @param implNode the subject against which to expect the {@link #FEDX_CONFIG} property.
+	 *
+	 * @throws RepositoryConfigException if any of the overridden fields are deemed to be invalid
+	 */
+	protected void parseFedXConfig(Model m, Resource implNode) throws RepositoryConfigException {
+		Models.objectResource(m.getStatements(implNode, FEDX_CONFIG, null))
+				.ifPresent(res -> parseFedXConfigInternal(m, res));
+	}
+
+	private void parseFedXConfigInternal(Model m, Resource confNode) throws RepositoryConfigException {
+		if (getConfig() == null) {
+			setConfig(new FedXConfig());
+		}
+
+		Models.objectLiteral(m.getStatements(confNode, CONFIG_JOIN_WORKER_THREADS, null))
+				.ifPresent(value -> config.withJoinWorkerThreads(value.intValue()));
+
+		Models.objectLiteral(m.getStatements(confNode, CONFIG_UNION_WORKER_THREADS, null))
+				.ifPresent(value -> config.withUnionWorkerThreads(value.intValue()));
+
+		Models.objectLiteral(m.getStatements(confNode, CONFIG_LEFT_JOIN_WORKER_THREADS, null))
+				.ifPresent(value -> config.withLeftJoinWorkerThreads(value.intValue()));
+
+		Models.objectLiteral(m.getStatements(confNode, CONFIG_BOUND_JOIN_BLOCK_SIZE, null))
+				.ifPresent(value -> config.withBoundJoinBlockSize(value.intValue()));
+
+		Models.objectLiteral(m.getStatements(confNode, CONFIG_ENFORCE_MAX_QUERY_TIME, null))
+				.ifPresent(value -> config.withEnforceMaxQueryTime(value.intValue()));
+
+		Models.objectLiteral(m.getStatements(confNode, CONFIG_ENABLE_SERVICE_AS_BOUND_JOIN, null))
+				.ifPresent(value -> config.withEnableServiceAsBoundJoin(value.booleanValue()));
+
+		Models.objectLiteral(m.getStatements(confNode, CONFIG_ENABLE_MONITORING, null))
+				.ifPresent(value -> config.withEnableMonitoring(value.booleanValue()));
+
+		Models.objectLiteral(m.getStatements(confNode, CONFIG_LOG_QUERY_PLAN, null))
+				.ifPresent(value -> config.withLogQueryPlan(value.booleanValue()));
+
+		Models.objectLiteral(m.getStatements(confNode, CONFIG_LOG_QUERIES, null))
+				.ifPresent(value -> config.withLogQueries(value.booleanValue()));
+
+		Models.objectLiteral(m.getStatements(confNode, CONFIG_DEBUG_QUERY_PLAN, null))
+				.ifPresent(value -> config.withDebugQueryPlan(value.booleanValue()));
+
+		Models.objectLiteral(m.getStatements(confNode, CONFIG_INCLUDE_INFERRED_DEFAULT, null))
+				.ifPresent(value -> config.withIncludeInferredDefault(value.booleanValue()));
+
+		Models.objectLiteral(m.getStatements(confNode, CONFIG_SOURCE_SELECTION_CACHE_SPEC, null))
+				.ifPresent(value -> config.withSourceSelectionCacheSpec(value.stringValue()));
+
+		Models.objectLiteral(m.getStatements(confNode, CONFIG_PREFIX_DECLARATIONS, null))
+				.ifPresent(value -> config.withPrefixDeclarations(value.stringValue()));
+
+		Models.objectLiteral(m.getStatements(confNode, CONFIG_CONSUMING_ITERATION_MAX, null))
+				.ifPresent(value -> config.withConsumingIterationMax(value.intValue()));
+
+	}
+
+	/**
+	 * Export the provided {@link FedXConfig} to its RDF representation. Note that {@link #getConfig()} could be null if
+	 * configuration has not been set yet.
+	 *
+	 * @param config   the configuration to export
+	 * @param implNode the node to which to write the config reference (i.e. {@link #FEDX_CONFIG}) to
+	 */
+	protected void exportFedXConfig(Model model, Resource implNode) {
+		if (getConfig() == null) {
+			return;
+		}
+
+		BNode confNode = Values.bnode();
+
+		model.add(confNode, CONFIG_JOIN_WORKER_THREADS, vf.createLiteral(config.getJoinWorkerThreads()));
+
+		model.add(confNode, CONFIG_UNION_WORKER_THREADS, vf.createLiteral(config.getUnionWorkerThreads()));
+
+		model.add(confNode, CONFIG_LEFT_JOIN_WORKER_THREADS, vf.createLiteral(config.getLeftJoinWorkerThreads()));
+
+		model.add(confNode, CONFIG_BOUND_JOIN_BLOCK_SIZE, vf.createLiteral(config.getBoundJoinBlockSize()));
+
+		model.add(confNode, CONFIG_ENFORCE_MAX_QUERY_TIME, vf.createLiteral(config.getEnforceMaxQueryTime()));
+
+		model.add(confNode, CONFIG_ENABLE_SERVICE_AS_BOUND_JOIN,
+				vf.createLiteral(config.getEnableServiceAsBoundJoin()));
+
+		model.add(confNode, CONFIG_ENABLE_MONITORING, vf.createLiteral(config.isEnableMonitoring()));
+
+		model.add(confNode, CONFIG_LOG_QUERY_PLAN, vf.createLiteral(config.isLogQueryPlan()));
+
+		model.add(confNode, CONFIG_LOG_QUERIES, vf.createLiteral(config.isLogQueries()));
+
+		model.add(confNode, CONFIG_DEBUG_QUERY_PLAN, vf.createLiteral(config.isDebugQueryPlan()));
+
+		model.add(confNode, CONFIG_INCLUDE_INFERRED_DEFAULT, vf.createLiteral(config.getIncludeInferredDefault()));
+
+		if (config.getSourceSelectionCacheSpec() != null) {
+			model.add(confNode, CONFIG_SOURCE_SELECTION_CACHE_SPEC,
+					vf.createLiteral(config.getSourceSelectionCacheSpec()));
+		}
+
+		if (config.getPrefixDeclarations() != null) {
+			model.add(confNode, CONFIG_PREFIX_DECLARATIONS,
+					vf.createLiteral(config.getPrefixDeclarations()));
+		}
+
+		model.add(confNode, CONFIG_CONSUMING_ITERATION_MAX, vf.createLiteral(config.getConsumingIterationMax()));
+
+		model.add(implNode, FEDX_CONFIG, confNode);
+	}
 }