peekable and markable iterator

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

@@ -14,6 +14,7 @@
 import java.util.Comparator;
 import java.util.function.Function;
 
+import org.eclipse.rdf4j.common.annotation.Experimental;
 import org.eclipse.rdf4j.common.iteration.CloseableIteration;
 import org.eclipse.rdf4j.common.iteration.LookAheadIteration;
 import org.eclipse.rdf4j.model.Value;
@@ -24,6 +25,12 @@
 import org.eclipse.rdf4j.query.algebra.evaluation.QueryEvaluationStep;
 import org.eclipse.rdf4j.query.algebra.evaluation.impl.QueryEvaluationContext;
 
+/**
+ *
+ *
+ * @author Håvard M. Ottestad
+ */
+@Experimental
 public class InnerMergeJoinIterator extends LookAheadIteration<BindingSet> {
 
 	private final CloseableIteration<BindingSet> leftIterator;

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

@@ -14,29 +14,33 @@
 import java.util.ArrayList;
 import java.util.Collections;
 import java.util.Iterator;
+import java.util.NoSuchElementException;
 
 import org.eclipse.rdf4j.common.annotation.Experimental;
 import org.eclipse.rdf4j.common.iteration.CloseableIteration;
-import org.eclipse.rdf4j.query.BindingSet;
 
 /**
  * An iterator that allows to peek at the next element without consuming it. It also allows to mark the current position
  * and reset to that position. Reset can be called multiple times, as long as the underlying iterator has not been
  * advanced beyond peek.
+ *
+ * @author Håvard M. Ottestad
  */
 @Experimental
-public class PeekMarkIterator implements CloseableIteration<BindingSet> {
+public class PeekMarkIterator<E> implements CloseableIteration<E> {
 
-	private final CloseableIteration<BindingSet> iterator;
+	private final CloseableIteration<E> iterator;
 	private boolean mark;
-	private ArrayList<BindingSet> buffer;
-	private Iterator<BindingSet> bufferIterator = Collections.emptyIterator();
-	private BindingSet next;
+	private ArrayList<E> buffer;
+	private Iterator<E> bufferIterator = Collections.emptyIterator();
+	private E next;
 
 	// -1: reset not possible; 0: reset possible, but next must be saved; 1: reset possible
 	private int resetPossible;
 
-	private PeekMarkIterator(CloseableIteration<BindingSet> iterator) {
+	private boolean closed;
+
+	PeekMarkIterator(CloseableIteration<E> iterator) {
 		this.iterator = iterator;
 	}
 
@@ -66,39 +70,69 @@ private void calculateNext() {
 
 	@Override
 	public boolean hasNext() {
+		if (closed) {
+			return false;
+		}
 		calculateNext();
 		return next != null;
 	}
 
 	@Override
-	public BindingSet next() {
+	public E next() {
+		if (closed) {
+			throw new NoSuchElementException("The iteration has been closed.");
+		}
 		calculateNext();
-		BindingSet result = next;
+		E result = next;
 		next = null;
 		if (!mark && resetPossible == 0) {
 			resetPossible--;
 		}
+		if (result == null) {
+			throw new NoSuchElementException();
+		}
+
 		return result;
 	}
 
-	public BindingSet peek() {
+	/**
+	 *
+	 * @return the next element without consuming it, or null if there are no more elements.
+	 */
+	public E peek() {
+		if (closed) {
+			return null;
+		}
 		calculateNext();
 		return next;
 	}
 
 	public void mark() {
+		if (closed) {
+			throw new IllegalStateException("The iteration has been closed.");
+		}
 		mark = true;
 		resetPossible = 1;
 		buffer = new ArrayList<>();
 		if (next != null) {
 			buffer.add(next);
 		}
+
 	}
 
 	public void reset() {
+		if (closed) {
+			throw new IllegalStateException("The iteration has been closed.");
+		}
 		if (buffer == null) {
 			throw new IllegalStateException("Mark never set");
 		}
+		if (mark && bufferIterator.hasNext()) {
+			while (bufferIterator.hasNext()) {
+				buffer.add(bufferIterator.next());
+			}
+		}
+
 		mark = false;
 		if (resetPossible < 0) {
 			throw new IllegalStateException("Reset not possible");
@@ -107,6 +141,7 @@ public void reset() {
 			next = null;
 			bufferIterator = buffer.iterator();
 		} else if (resetPossible == 1) {
+			next = null;
 			bufferIterator = buffer.iterator();
 		}
 
@@ -115,6 +150,7 @@ public void reset() {
 
 	@Override
 	public void close() {
+		this.closed = true;
 		iterator.close();
 	}
 }