GH-4581 Make sure collection factories are used when we use a collection

This allows to always be sure that we can fall back to disk if required.
Also allows optimized datastructures to be injected.

core/common/iterator/src/main/java/org/eclipse/rdf4j/common/iteration/DistinctIteration.java

@@ -38,17 +38,30 @@ public class DistinctIteration<E> extends FilterIteration<E> {
  *
  * @param iter The underlying iterator.
  */
+ @Deprecated
  public DistinctIteration(CloseableIteration<? extends E> iter) {
  super(iter);
 
  excludeSet = new HashSet<>();
  }
 
- public DistinctIteration(CloseableIteration<? extends E> iter, Set<E> set) {
+ /**
+ * Creates a new DistinctIterator.
+ *
+ * @param Set<E> a hopefully optimized set
+ * @param iter The underlying iterator.
+ */
+ public DistinctIteration(CloseableIteration<? extends E> iter, Set<E> excludeSet) {
  super(iter);
- excludeSet = set;
+ this.excludeSet = excludeSet;
  }
 
+ /**
+ * Creates a new DistinctIterator.
+ *
+ * @param Supplier<Set<E>> a supplier of a hopefully optimized set
+ * @param iter The underlying iterator.
+ */
  public DistinctIteration(CloseableIteration<? extends E> iter, Supplier<Set<E>> setMaker) {
  super(iter);
  excludeSet = setMaker.get();

core/query/src/test/java/org/eclipse/rdf4j/common/iteration/DistinctIterationTest.java

@@ -16,7 +16,7 @@ public class DistinctIterationTest extends CloseableIterationTest {
 
  @Override
  protected CloseableIteration<String> createTestIteration() {
- return new DistinctIteration<>(createStringList1Iteration(), new HashSet<>());
+ return new DistinctIteration<>(createStringList1Iteration(), HashSet::new);
  }
 
  @Override

core/queryalgebra/evaluation/src/main/java/org/eclipse/rdf4j/query/algebra/evaluation/impl/DefaultEvaluationStrategy.java

@@ -533,16 +533,15 @@ protected QueryEvaluationStep prepare(Difference node, QueryEvaluationContext co
  }
 
  protected QueryEvaluationStep prepare(Group node, QueryEvaluationContext context) throws QueryEvaluationException {
- return bindings -> new GroupIterator(DefaultEvaluationStrategy.this, node, bindings,
- iterationCacheSyncThreshold,
- context);
+ return bindings -> new GroupIterator(DefaultEvaluationStrategy.this, node, bindings, context);
  }
 
  protected QueryEvaluationStep prepare(Intersection node, QueryEvaluationContext context)
  throws QueryEvaluationException {
  QueryEvaluationStep leftArg = precompile(node.getLeftArg(), context);
  QueryEvaluationStep rightArg = precompile(node.getRightArg(), context);
- return new IntersectionQueryEvaluationStep(leftArg, rightArg, this.getCollectionFactory().get());
+
+ return new IntersectionQueryEvaluationStep(leftArg, rightArg, getCollectionFactory());
  }
 
  protected QueryEvaluationStep prepare(Join node, QueryEvaluationContext context) throws QueryEvaluationException {
@@ -710,20 +709,19 @@ protected QueryEvaluationStep prepare(DescribeOperator node, QueryEvaluationCont
  protected QueryEvaluationStep prepare(Distinct node, QueryEvaluationContext context)
  throws QueryEvaluationException {
  final QueryEvaluationStep child = precompile(node.getArg(), context);
- CollectionFactory cf = this.getCollectionFactory().get();
+ final CollectionFactory cf = this.getCollectionFactory().get();
  return bindings -> {
  final CloseableIteration<BindingSet> evaluate = child.evaluate(bindings);
- return new DistinctIteration<BindingSet>(evaluate, cf::createSet) {
+ return new DistinctIteration<BindingSet>(evaluate, cf.createSetOfBindingSets()) {
 
  @Override
- protected void handleClose() {
+ protected void handleClose() throws QueryEvaluationException {
  try {
  cf.close();
  } finally {
  super.handleClose();
  }
  }
-
  };
  };
  }

core/queryalgebra/evaluation/src/main/java/org/eclipse/rdf4j/query/algebra/evaluation/impl/evaluationsteps/IntersectionQueryEvaluationStep.java

@@ -11,53 +11,43 @@
 package org.eclipse.rdf4j.query.algebra.evaluation.impl.evaluationsteps;
 
 import java.util.function.Function;
+import java.util.function.Supplier;
 
 import org.eclipse.rdf4j.collection.factory.api.CollectionFactory;
 import org.eclipse.rdf4j.common.iteration.CloseableIteration;
 import org.eclipse.rdf4j.common.iteration.IntersectIteration;
 import org.eclipse.rdf4j.query.BindingSet;
-import org.eclipse.rdf4j.query.QueryEvaluationException;
 import org.eclipse.rdf4j.query.algebra.evaluation.QueryEvaluationStep;
 
 /**
  * A step that prepares the arguments of an Intersection operator before execution.
  */
 public class IntersectionQueryEvaluationStep implements QueryEvaluationStep {
 
- private static final class IntersectIterationUsingSetFromCollectionFactory
- extends IntersectIteration<BindingSet> {
- private final CollectionFactory cf;
-
- private IntersectIterationUsingSetFromCollectionFactory(CloseableIteration<BindingSet> arg1,
- CloseableIteration<BindingSet> arg2, CollectionFactory cf) {
- super(arg1, arg2, false, cf::createSetOfBindingSets);
- this.cf = cf;
- }
-
- @Override
- protected void handleClose() throws QueryEvaluationException {
- try {
- cf.close();
- } finally {
- super.handleClose();
- }
- }
- }
-
  private final QueryEvaluationStep leftArg;
  private final Function<BindingSet, DelayedEvaluationIteration> rightArgDelayed;
- private final CollectionFactory collectionFactory;
+ private final Supplier<CollectionFactory> cfs;
 
  public IntersectionQueryEvaluationStep(QueryEvaluationStep leftArg, QueryEvaluationStep rightArg,
- CollectionFactory collectionFactory) {
- this.collectionFactory = collectionFactory;
+ Supplier<CollectionFactory> cfs) {
+ this.cfs = cfs;
  this.leftArg = leftArg;
  rightArgDelayed = bs -> new DelayedEvaluationIteration(rightArg, bs);
  }
 
  @Override
  public CloseableIteration<BindingSet> evaluate(BindingSet bs) {
- return new IntersectIterationUsingSetFromCollectionFactory(leftArg.evaluate(bs), rightArgDelayed.apply(bs),
- collectionFactory);
+ CollectionFactory cf = cfs.get();
+ return new IntersectIteration<>(leftArg.evaluate(bs), rightArgDelayed.apply(bs), cf::createSetOfBindingSets) {
+ @Override
+ protected void handleClose() {
+ try {
+ cf.close();
+ } finally {
+ super.handleClose();
+ }
+ }
+ };
  }
+
 }

core/queryalgebra/evaluation/src/main/java/org/eclipse/rdf4j/query/algebra/evaluation/iterator/PathIteration.java

@@ -110,6 +110,7 @@ public PathIteration(EvaluationStrategy strategy, Scope scope, Var startVar,
 
  this.currentLength = minLength;
  this.bindings = bindings;
+
  this.collectionFactory = strategy.getCollectionFactory().get();
 
  // This is all necessary for optimized collections to be usable. This only becomes important on very large

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

@@ -12,6 +12,8 @@
 
 import java.util.Optional;
 
+import org.eclipse.rdf4j.collection.factory.api.CollectionFactory;
+import org.eclipse.rdf4j.collection.factory.impl.DefaultCollectionFactory;
 import org.eclipse.rdf4j.federated.cache.SourceSelectionCache;
 import org.eclipse.rdf4j.federated.cache.SourceSelectionCacheFactory;
 import org.eclipse.rdf4j.federated.cache.SourceSelectionMemoryCache;
@@ -66,6 +68,7 @@ public class FedXConfig {
 
  private int consumingIterationMax = 1000;
 
+ private CollectionFactory cf = new DefaultCollectionFactory();
  /* factory like setters */
 
  /**
@@ -472,4 +475,19 @@ public FedXConfig withConsumingIterationMax(int max) {
  public int getConsumingIterationMax() {
  return consumingIterationMax;
  }
+
+ /**
+ * Set the CollectionFactory to be used by the federation
+ *
+ * <p>
+ * Can only be set before federation initialization.
+ * </p>
+ *
+ * @param cf
+ * @return the current config
+ */
+ public FedXConfig withCollectionFactory(CollectionFactory cf) {
+ this.cf = cf;
+ return this;
+ }
 }

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

@@ -275,6 +275,7 @@ protected SailException convert(RuntimeException e) {
  };
  return new DistinctIteration<Resource>(conv, cf::createSet) {
 
+ @Override
  protected void handleClose() {
  try {
  cf.close();

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

@@ -220,7 +220,6 @@ public FedXRepository create() {
  if (this.writeStrategyFactory != null) {
  federation.setWriteStrategyFactory(writeStrategyFactory);
  }
-
  FedXRepository repo = new FedXRepository(federation, this.config);
  if (this.repositoryResolver != null) {
  repo.setRepositoryResolver(repositoryResolver);

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

@@ -15,8 +15,10 @@
 import java.util.Set;
 import java.util.concurrent.Executor;
 import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.function.Supplier;
 import java.util.stream.Collectors;
 
+import org.eclipse.rdf4j.collection.factory.api.CollectionFactory;
 import org.eclipse.rdf4j.common.iteration.CloseableIteration;
 import org.eclipse.rdf4j.common.iteration.EmptyIteration;
 import org.eclipse.rdf4j.common.iteration.SingletonIteration;
@@ -987,5 +989,4 @@ protected CloseableIteration<BindingSet> evaluateAtStatementSources(
  throw new QueryEvaluationException(e);
  }
  }
-
 }

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

@@ -43,6 +43,7 @@ public void beforeEach(ExtensionContext ctx) {
  for (Consumer<FedXConfig> configConsumer : configurations) {
  configConsumer.accept(fedxConfig);
  }
+
  List<Endpoint> endpoints = Collections.<Endpoint>emptyList();
  repository = FedXFactory.newFederation().withMembers(endpoints).withConfig(fedxConfig).create();
  repository.init();