GH-3969: Native stream() implementations for DatasetGraph

Replace the find()-wrapping default of DatasetGraph.stream(g,s,p,o) with
first-class stream support across the hierarchy. DatasetGraphBaseFind gets a
stream() path mirroring the find() default-/named-/union-graph split, backed
by streamInDftGraph / streamInSpecificNamedGraph / streamInAnyNamedGraphs
primitives implemented per dataset. GraphView streams graph-over-dataset
through DatasetGraph#stream; adds G stream helpers. The interface method stays
default, so implementors are not broken. Includes stream()==find() parity
tests and a forEachRemaining check in IteratorTxnTracker so stream bulk
operations stay inside their transaction.

Author: bern-soptim

jena-arq/src/main/java/org/apache/jena/riot/writer/c14n/DatasetGraphOrdered.java

@@ -23,6 +23,7 @@
 
 import java.util.Iterator;
 import java.util.LinkedHashSet;
+import java.util.stream.Stream;
 
 import org.apache.jena.atlas.iterator.Iter;
 import org.apache.jena.graph.Graph;
@@ -65,6 +66,16 @@ public Iterator<Quad> findNG(Node g, Node s, Node p, Node o) {
         return iterator;
     }
 
+    @Override
+    public Stream<Quad> stream(Node g, Node s, Node p, Node o) {
+        return quads.stream().filter(quad -> matches(quad, g, s, p, o));
+    }
+
+    @Override
+    public Stream<Quad> stream() {
+        return quads.stream();
+    }
+
     @Override
     public void add(Quad quad) {
         quads.add(quad);

jena-arq/src/main/java/org/apache/jena/sparql/core/DatasetGraphBaseFind.java

@@ -82,12 +82,46 @@ protected Iterator<Quad> findAny(Node s, Node p, Node o) {
         return Iter.append(iter1, iter2);
     }
 
+    @Override
+    public Stream<Quad> stream(Node g, Node s, Node p, Node o) {
+        if ( Quad.isDefaultGraph(g))
+            return streamInDftGraph(s, p, o) ;
+        if ( ! isWildcard(g) )
+            return streamNG(g, s, p, o) ;
+        return streamAny(s, p, o) ;
+    }
+
+    /** Stream equivalent of {@link #findNG(Node, Node, Node, Node)}. */
+    public Stream<Quad> streamNG(Node g, Node s, Node p , Node o) {
+        Stream<Quad> stream ;
+        if ( Quad.isUnionGraph(g))
+            stream = streamQuadsInUnionGraph(s, p, o) ;
+        else if ( isWildcard(g) )
+            stream = streamInAnyNamedGraphs(s, p, o) ;
+        else if ( Quad.isDefaultGraph(g) )
+            stream = streamInDftGraph(s, p, o) ;
+        else
+            // Not wildcard, not union graph, not default graph.
+            stream = streamInSpecificNamedGraph(g, s, p, o) ;
+        if ( stream == null )
+            return Stream.empty() ;
+        return stream ;
+    }
+
+    protected Stream<Quad> streamAny(Node s, Node p, Node o) {
+        return Stream.concat(streamInDftGraph(s, p, o), streamInAnyNamedGraphs(s, p, o)) ;
+    }
+
     /** Find matches in the default graph.
      *  Return as quads; the default graph is {@link Quad#defaultGraphIRI}
      *  To get Triples, use {@code DatasetGraph.getDefaultGraph().find(...)}.
      */
     protected abstract Iterator<Quad> findInDftGraph(Node s, Node p , Node o) ;
 
+    /** Stream equivalent of {@link #findInDftGraph}.
+     */
+    protected abstract Stream<Quad> streamInDftGraph(Node s, Node p, Node o) ;
+
     /** Find matches in the notional union of all named graphs - return as triples.
      * No duplicates - the union graph is a <em>set</em> of triples.
      * See {@link #findInAnyNamedGraphs}, where there may be duplicates.
@@ -110,6 +144,11 @@ public Iterator<Quad> findQuadsInUnionGraph(Node s, Node p , Node o) {
         return findUnionGraphTriples(s,p,o).map(t -> Quad.create(Quad.unionGraph, t)).iterator() ;
     }
 
+    /** Stream equivalent of {@link #findQuadsInUnionGraph}. */
+    public Stream<Quad> streamQuadsInUnionGraph(Node s, Node p , Node o) {
+        return findUnionGraphTriples(s,p,o).map(t -> Quad.create(Quad.unionGraph, t)) ;
+    }
+
     /** Find matches in the notional union of all named graphs - return as triples.
      * No duplicates - the union graph is a <em>set</em> of triples.
      * See {@link #findInAnyNamedGraphs}, where there may be duplicates.
@@ -118,15 +157,21 @@ public Iterator<Quad> findQuadsInUnionGraph(Node s, Node p , Node o) {
      * may be possible to avoid "distinct".
      */
     private Stream<Triple> findUnionGraphTriples(Node s, Node p , Node o) {
-        return Iter.asStream(findInAnyNamedGraphs(s,p,o)).map(Quad::asTriple).distinct() ;
+        return streamInAnyNamedGraphs(s,p,o).map(Quad::asTriple).distinct() ;
     }
 
     /** Find in a specific named graph - {@code g} is a ground term (IRI or bNode), not a wild card (or null). */
     protected abstract Iterator<Quad> findInSpecificNamedGraph(Node g, Node s, Node p , Node o) ;
 
+    /** Stream equivalent of {@link #findInSpecificNamedGraph}. */
+    protected abstract Stream<Quad> streamInSpecificNamedGraph(Node g, Node s, Node p, Node o) ;
+
     /** Find in any named graph - return quads.
      * If a triple matches in two different graph, return a quad for each.
      * See {@link #findInUnionGraph} for matching without duplicate triples.
      */
     protected abstract Iterator<Quad> findInAnyNamedGraphs(Node s, Node p , Node o) ;
+
+    /** Stream equivalent of {@link #findInAnyNamedGraphs}. */
+    protected abstract Stream<Quad> streamInAnyNamedGraphs(Node s, Node p, Node o) ;
 }

jena-arq/src/main/java/org/apache/jena/sparql/core/DatasetGraphCollection.java

@@ -24,6 +24,7 @@
 import java.util.Iterator ;
 import java.util.List ;
 import java.util.Objects ;
+import java.util.stream.Stream;
 
 import org.apache.jena.atlas.iterator.Iter ;
 import org.apache.jena.atlas.iterator.IteratorConcat ;
@@ -63,6 +64,11 @@ protected Iterator<Quad> findInDftGraph(Node s, Node p , Node o)
         return G.triples2quadsDftGraph(getDefaultGraph().find(s, p, o)) ;
     }
 
+    @Override
+    protected Stream<Quad> streamInDftGraph(Node s, Node p, Node o) {
+        return G.triples2quadsDftGraph(getDefaultGraph().stream(s, p, o)) ;
+    }
+
     @Override
     protected Iter<Quad> findInSpecificNamedGraph(Node g, Node s, Node p , Node o)
     {
@@ -72,6 +78,14 @@ protected Iter<Quad> findInSpecificNamedGraph(Node g, Node s, Node p , Node o)
         return G.triples2quads(g, graph.find(s, p, o)) ;
     }
 
+    @Override
+    protected Stream<Quad> streamInSpecificNamedGraph(Node g, Node s, Node p, Node o) {
+        Graph graph = fetchGraph(g) ;
+        if ( graph == null )
+            return Stream.empty() ;
+        return G.triples2quads(g, graph.stream(s, p, o)) ;
+    }
+
     @Override
     protected Iterator<Quad> findInAnyNamedGraphs(Node s, Node p, Node o)
     {
@@ -89,6 +103,12 @@ protected Iterator<Quad> findInAnyNamedGraphs(Node s, Node p, Node o)
         return iter ;
     }
 
+    @Override
+    protected Stream<Quad> streamInAnyNamedGraphs(Node s, Node p, Node o) {
+        return Iter.asStream(listGraphNodes())
+                .flatMap(gn -> streamInSpecificNamedGraph(gn, s, p, o));
+    }
+
     @Override
     public abstract Iterator<Node> listGraphNodes() ;
 

jena-arq/src/main/java/org/apache/jena/sparql/core/DatasetGraphFilteredView.java

@@ -25,6 +25,7 @@
 import java.util.HashSet;
 import java.util.Iterator;
 import java.util.function.Predicate;
+import java.util.stream.Stream;
 
 import org.apache.jena.atlas.iterator.Iter;
 import org.apache.jena.atlas.logging.Log;
@@ -90,6 +91,12 @@ private Iterator<Quad> filter(Iterator<Quad> iter) {
             return iter;
         return Iter.filter(iter, this::filter);
     }
+
+    private Stream<Quad> filter(Stream<Quad> stream) {
+        if ( this.quadFilter == null )
+            return stream;
+        return stream.filter(this::filter);
+    }
 
     // Need to intercept these because otherwise that are a GraphView of the wrapped "dsg", not this one.  
 
@@ -149,6 +156,14 @@ public Iterator<Quad> find() {
         return filter(super.find(g, s, p, o)).hasNext();
     }
 
+    @Override public Stream<Quad> stream(Node g, Node s, Node p, Node o) {
+        return filter(super.stream(g, s, p, o));
+    }
+
+    @Override public Stream<Quad> stream() {
+        return filter(super.stream());
+    }
+
     @Override public boolean contains(Quad quad) {
         return filter(super.find(quad)).hasNext();
     }

jena-arq/src/main/java/org/apache/jena/sparql/core/DatasetGraphMap.java

@@ -26,7 +26,9 @@
 import java.util.Map;
 import java.util.Map.Entry;
 import java.util.Objects;
+import java.util.stream.Stream;
 
+import org.apache.jena.atlas.iterator.Iter;
 import org.apache.jena.atlas.iterator.IteratorConcat;
 import org.apache.jena.graph.Graph;
 import org.apache.jena.graph.Node;
@@ -157,12 +159,22 @@ protected Iterator<Quad> findInDftGraph(Node s, Node p, Node o) {
         return G.triples2quadsDftGraph(iter) ;
     }
 
+    @Override
+    protected Stream<Quad> streamInDftGraph(Node s, Node p, Node o) {
+        return G.triples2quadsDftGraph(getDefaultGraph().stream(s, p, o));
+    }
+
     @Override
     protected Iterator<Quad> findInSpecificNamedGraph(Node g, Node s, Node p, Node o) {
         Iterator<Triple> iter = getGraph(g).find(s, p, o);
         return G.triples2quads(g, iter);
     }
 
+    @Override
+    protected Stream<Quad> streamInSpecificNamedGraph(Node g, Node s, Node p, Node o) {
+        return G.triples2quads(g, getGraph(g).stream(s, p, o));
+    }
+
     @Override
     protected Iterator<Quad> findInAnyNamedGraphs(Node s, Node p, Node o) {
         Iterator<Node> gnames = listGraphNodes();
@@ -178,6 +190,12 @@ protected Iterator<Quad> findInAnyNamedGraphs(Node s, Node p, Node o) {
         return iter;
     }
 
+    @Override
+    protected Stream<Quad> streamInAnyNamedGraphs(Node s, Node p, Node o) {
+        return Iter.asStream(listGraphNodes())
+                .flatMap(gn -> streamInSpecificNamedGraph(gn, s, p, o));
+    }
+
     @Override
     public Graph getDefaultGraph() {
         return defaultGraph;

jena-arq/src/main/java/org/apache/jena/sparql/core/DatasetGraphNull.java

@@ -22,6 +22,7 @@
 package org.apache.jena.sparql.core;
 
 import java.util.Iterator;
+import java.util.stream.Stream;
 
 import org.apache.jena.atlas.iterator.Iter;
 import org.apache.jena.graph.Graph;
@@ -73,16 +74,31 @@ protected Iterator<Quad> findInDftGraph(Node s, Node p, Node o) {
         return Iter.nullIterator();
     }
 
+    @Override
+    protected Stream<Quad> streamInDftGraph(Node s, Node p, Node o) {
+        return Stream.empty();
+    }
+
     @Override
     protected Iterator<Quad> findInSpecificNamedGraph(Node g, Node s, Node p, Node o) {
         return Iter.nullIterator();
     }
 
+    @Override
+    protected Stream<Quad> streamInSpecificNamedGraph(Node g, Node s, Node p, Node o) {
+        return Stream.empty();
+    }
+
     @Override
     protected Iterator<Quad> findInAnyNamedGraphs(Node s, Node p, Node o) {
         return Iter.nullIterator();
     }
 
+    @Override
+    protected Stream<Quad> streamInAnyNamedGraphs(Node s, Node p, Node o) {
+        return Stream.empty();
+    }
+
     @Override
     public Graph getDefaultGraph() {
         return dftGraph;

jena-arq/src/main/java/org/apache/jena/sparql/core/DatasetGraphOne.java

@@ -22,6 +22,7 @@
 package org.apache.jena.sparql.core;
 
 import java.util.Iterator;
+import java.util.stream.Stream;
 
 import org.apache.jena.atlas.iterator.Iter;
 import org.apache.jena.graph.Graph;
@@ -202,18 +203,35 @@ protected Iterator<Quad> findInDftGraph(Node s, Node p, Node o) {
         return G.triples2quadsDftGraph(graph.find(s, p, o));
     }
 
+    @Override
+    protected Stream<Quad> streamInDftGraph(Node s, Node p, Node o) {
+        return G.triples2quadsDftGraph(graph.stream(s, p, o));
+    }
+
     @Override
     protected Iterator<Quad> findInSpecificNamedGraph(Node g, Node s, Node p, Node o) {
         // There are no named graphs
         return Iter.nullIterator();
     }
 
+    @Override
+    protected Stream<Quad> streamInSpecificNamedGraph(Node g, Node s, Node p, Node o) {
+        // There are no named graphs
+        return Stream.empty();
+    }
+
     @Override
     protected Iterator<Quad> findInAnyNamedGraphs(Node s, Node p, Node o) {
         // There are no named graphs
         return Iter.nullIterator();
     }
 
+    @Override
+    protected Stream<Quad> streamInAnyNamedGraphs(Node s, Node p, Node o) {
+        // There are no named graphs
+        return Stream.empty();
+    }
+
     protected static boolean isDefaultGraph(Quad quad) {
         return isDefaultGraph(quad.getGraph());
     }

jena-arq/src/main/java/org/apache/jena/sparql/core/DatasetGraphWrapper.java

@@ -22,6 +22,7 @@
 package org.apache.jena.sparql.core;
 
 import java.util.Iterator;
+import java.util.stream.Stream;
 
 import org.apache.jena.atlas.lib.Sync;
 import org.apache.jena.graph.Graph;
@@ -211,6 +212,14 @@ public Iterator<Quad> find(Node g, Node s, Node p, Node o)
     public Iterator<Quad> findNG(Node g, Node s, Node p, Node o)
     { return getR().findNG(g, s, p, o); }
 
+    @Override
+    public Stream<Quad> stream(Node g, Node s, Node p, Node o)
+    { return getR().stream(g, s, p, o); }
+
+    @Override
+    public Stream<Quad> stream()
+    { return getR().stream(); }
+
     @Override
     public boolean contains(Quad quad)
     { return getR().contains(quad); }

jena-arq/src/main/java/org/apache/jena/sparql/core/GraphView.java

@@ -22,6 +22,7 @@
 package org.apache.jena.sparql.core;
 
 import java.util.Iterator ;
+import java.util.stream.Stream;
 
 import org.apache.jena.atlas.iterator.Iter ;
 import org.apache.jena.atlas.lib.Sync ;
@@ -125,6 +126,24 @@ protected ExtendedIterator<Triple> graphBaseFind(Node s, Node p, Node o) {
         return WrappedIterator.createNoRemove(iter) ;
     }
 
+    /**
+     * Stream equivalent of {@link #graphBaseFind} / {@link #graphUnionFind}: route through DatasetGraph#stream
+     * so stream-native dataset (e.g. an in-memory one) stays on streams rather than iterator-wrapping.
+     * @param s subject match
+     * @param p predicate match
+     * @param o object match
+     * @return stream of matching triples
+     */
+    @Override
+    public Stream<Triple> stream(Node s, Node p, Node o) {
+        Node g = graphNode(graphName) ;
+        Stream<Triple> stream = G.quads2triples(dsg.stream(g, s, p, o)) ;
+        if ( Quad.isUnionGraph(graphName) )
+            // Suppress duplicates after projecting to triples.
+            stream = stream.distinct() ;
+        return stream ;
+    }
+
     private static Node graphNode(Node gn) {
         return ( gn == null ) ? Quad.defaultGraphNodeGenerated : gn ;
     }