GH-4055: Pass context into update operations

jena-arq/src/main/java/org/apache/jena/riot/RDFParserBuilder.java

@@ -610,12 +610,12 @@ else if ( uri != null )
         else
             parserBaseURI = null;
 
-        StreamManager sMgr = streamManager;
-        if ( sMgr == null )
-            sMgr = StreamManager.get(context);
+        StreamManager streamMgr = streamManager;
+        if ( streamMgr == null )
+            streamMgr = StreamManager.get(context);
 
         // Can't build the profile here as it is Lang/conneg dependent.
-        return new RDFParser(uri, path, stringToParse, inputStream, javaReader, sMgr,
+        return new RDFParser(uri, path, stringToParse, inputStream, javaReader, streamMgr,
                              appAcceptHeader, httpHeaders,
                              httpClient,
                              hintLang, forceLang,

jena-arq/src/main/java/org/apache/jena/riot/system/streammgr/StreamManager.java

@@ -90,6 +90,13 @@ public static StreamManager get(Context context) {
         return get();
     }
 
+    /**
+     * Set the {@code StreamManager} in the context.
+     */
+    public static void set(Context context, StreamManager streamManager) {
+        context.set(SysRIOT.sysStreamManager, streamManager);
+    }
+
     /**
      * Set the global {@code StreamManager}.
      */

jena-arq/src/main/java/org/apache/jena/sparql/engine/QueryEngineBase.java

@@ -77,6 +77,11 @@ private DatasetGraph prepareDataset(DatasetGraph originalDataset, Query query) {
         if ( ! query.hasDatasetDescription() )
             throw new QueryExecException("No dataset and no dataset description for query");
 
+        // dsg == null.
+        // This is the only case where the code will process FROM/FROM NAMED by
+        // reading resources. When dgs != null, FROM/FROM NAMED pick graphs from
+        // the dataset and form a dynamic dataset.
+
         // DatasetDescription : Build it.
         String baseURI = query.getBaseURI();
         if ( baseURI == null )

jena-arq/src/main/java/org/apache/jena/sparql/lang/UpdateParser.java

@@ -33,25 +33,24 @@
 import org.apache.jena.sparql.modify.UpdateSink;
 import org.apache.jena.util.FileUtils;
 
-/** 
+/**
  * This class provides the root of lower level access to all the update parsers.
  * Each subclass hides the details of the per-language exception handlers and other
- * javacc details.    
+ * javacc details.
  */
 
 public abstract class UpdateParser
 {
     protected UpdateParser() {}
-    
-    /** Parse a string */ 
+
+    /** Parse a string */
     public final void parse(UpdateSink sink, Prologue prologue, String updateString) throws QueryParseException {
         Reader r = new StringReader(updateString);
         executeParse(sink, prologue, r);
     }
 
-    /** Parse an input stream */ 
+    /** Parse an input stream */
     public final void parse(UpdateSink sink, Prologue prologue, InputStream input) throws QueryParseException {
-        // BOM processing moved to the grammar.
         Reader r = FileUtils.asBufferedUTF8(input);
         executeParse(sink, prologue, r);
     }
@@ -65,7 +64,7 @@ public void parse(UpdateSink sink, Prologue prologue, Reader r) {
 
     // Subclass action.
     protected abstract void executeParse(UpdateSink sink, Prologue prologue, Reader r);
-    
+
     public static boolean canParse(Syntax syntaxURI) {
         return UpdateParserRegistry.get().containsFactory(syntaxURI);
     }

jena-arq/src/main/java/org/apache/jena/sparql/modify/UpdateEngineWorker.java

@@ -24,6 +24,7 @@
 import static org.apache.jena.sparql.modify.TemplateLib.remapDefaultGraph;
 import static org.apache.jena.sparql.modify.TemplateLib.template;
 
+import java.io.InputStream;
 import java.util.ArrayList;
 import java.util.Collection;
 import java.util.Iterator;
@@ -44,6 +45,8 @@
 import org.apache.jena.query.Query;
 import org.apache.jena.riot.*;
 import org.apache.jena.riot.system.PrefixMap;
+import org.apache.jena.riot.system.StreamRDF;
+import org.apache.jena.riot.system.StreamRDFLib;
 import org.apache.jena.sparql.ARQInternalErrorException;
 import org.apache.jena.sparql.core.*;
 import org.apache.jena.sparql.engine.Timeouts;
@@ -142,7 +145,7 @@ protected void execDropClear(UpdateDropClear update, Node g, boolean isClear) {
         boolean auto = autoSilent && !isClear;
         executeOperation( auto || update.isSilent(), () -> {
             if ( g != null && !datasetGraph.containsGraph(g) )
-                    throw errorEx("No such graph: " + g);
+                throw errorEx("No such graph: " + g);
             if ( isClear ) {
                 if ( g == null || datasetGraph.containsGraph(g) )
                     graphOrThrow(datasetGraph, g).clear();
@@ -190,6 +193,7 @@ public void visit(UpdateLoad update) {
         // LOAD SILENT? iri ( INTO GraphRef )?
         String source = update.getSource();
         Node dest = update.getDest();
+
         executeOperation(update.isSilent(), ()->{
             Graph graph = graphOrThrow(datasetGraph, dest);
             // We must load buffered if silent so that the dataset graph sees
@@ -198,39 +202,30 @@ public void visit(UpdateLoad update) {
             try {
                 boolean loadBuffered = update.isSilent() || ! datasetGraph.supportsTransactionAbort();
                 if ( dest == null ) {
-                    // LOAD SILENT? iri
+                    // LOAD SILENT? iri -- no INTO
                     // Quads accepted (extension).
                     if ( loadBuffered ) {
                         DatasetGraph dsg2 = DatasetGraphFactory.create();
-                        RDFDataMgr.read(dsg2, source);
+                        loadReadQuads(source, dsg2, context);
+                        // Parsing seceded.
                         dsg2.find().forEachRemaining(datasetGraph::add);
                     } else {
-                        RDFDataMgr.read(datasetGraph, source);
+                        // Transactional and not SILENT.
+                        loadReadQuads(source, datasetGraph, context);
                     }
                     return;
                 }
                 // LOAD SILENT? iri INTO GraphRef
-                // Load triples. To give a decent error message and also not have the usual
-                // parser behaviour of just selecting default graph triples when the
-                // destination is a graph, we need to do the same steps as RDFParser.parseURI,
-                // with different checking.
+                // Load triples.
                 TypedInputStream input = RDFDataMgr.open(source);
-                String contentType = input.getContentType();
-                Lang lang = RDFDataMgr.determineLang(source, contentType, Lang.TTL);
-                if ( lang == null )
-                    throw new UpdateException("Failed to determine the syntax for '"+source+"'");
-                if ( ! RDFLanguages.isTriples(lang) )
-                    throw new UpdateException("Attempt to load quads into a graph");
-                RDFParser parser = RDFParser
-                        .source(input.getInputStream())
-                        .forceLang(lang)
-                        .build();
+                Lang lang = determineLang(source, input);
+
                 if ( loadBuffered ) {
                     Graph g = GraphFactory.createGraphMem();
-                    parser.parse(g);
+                    loadReadTriples(input, lang, g, context);
                     GraphUtil.addInto(graph, g);
                 } else {
-                    parser.parse(graph);
+                    loadReadTriples(input, lang, graph, context);
                 }
             } catch (RiotException ex) {
                 if ( !update.isSilent() ) {
@@ -240,6 +235,43 @@ public void visit(UpdateLoad update) {
         });
     }
 
+    // To give a decent error message, and also not have the usual
+    // parser behaviour of just selecting default graph triples when
+    // the destination is a graph, we need to do the same steps as
+    // RDFParser.parseURI with different checking.
+    private static Lang determineLang(String source, TypedInputStream input) {
+        String contentType = input.getContentType();
+        Lang lang = RDFDataMgr.determineLang(source, contentType, Lang.TTL);
+        if ( lang == null )
+            throw new UpdateException("Failed to determine the syntax for '"+source+"'");
+        if ( ! RDFLanguages.isTriples(lang) )
+            throw new UpdateException("Attempt to load quads into a graph");
+        return lang;
+    }
+
+    /** Load data into a dataset graph. The context has the stream manager. */
+    private static void loadReadQuads(String source, DatasetGraph destination, Context context) {
+        StreamRDF parserDest = StreamRDFLib.dataset(destination);
+        // RDFParser picks up the stream manager from the context if not set.
+        //StreamManager streamMgr = StreamManager.get(context);
+        RDFParser.source(source)
+                //.streamManager(streamMgr)
+                .context(context)
+                .parse(parserDest);
+    }
+
+    /** Load data into a graph. The context has the stream manager. */
+    private static void loadReadTriples(InputStream input,  Lang lang, Graph destination, Context context) {
+        StreamRDF parserDest = StreamRDFLib.graph(destination);
+        // RDFParser picks up the stream manager from the context if not set.
+        //StreamManager streamMgr = StreamManager.get(context);
+        RDFParser.source(input)
+                .forceLang(lang)
+                //.streamManager(streamMgr)
+                .context(context)
+                .parse(parserDest);
+    }
+
     @Override
     public void visit(UpdateAdd update) {
         executeOperation(update.isSilent(), ()->{

jena-arq/src/main/java/org/apache/jena/sparql/modify/UpdateRequestSink.java

@@ -28,10 +28,14 @@
 import org.apache.jena.update.Update ;
 import org.apache.jena.update.UpdateRequest ;
 
+/**
+ * Accumulate {@link Update Updates} (the individual update operations) in an
+ * {@link UpdateRequest}.
+ */
 public class UpdateRequestSink implements UpdateSink
 {
     private final UpdateRequest updateRequest;
-    
+
     public UpdateRequestSink(UpdateRequest updateRequest) {
         this.updateRequest = updateRequest;
     }
@@ -44,7 +48,7 @@ public void send(Update update) {
     @Override
     public void flush()
     { }
-    
+
     @Override
     public void close()
     { }
@@ -53,15 +57,13 @@ public void close()
     public QuadDataAccSink createInsertDataSink() {
         QuadDataAcc quads = new QuadDataAcc();
         send(new UpdateDataInsert(quads));
-
         return quads;
     }
 
     @Override
     public QuadDataAccSink createDeleteDataSink() {
         QuadDataAcc quads = new QuadDataAcc();
         send(new UpdateDataDelete(quads));
-
         return quads;
     }
 }

jena-arq/src/main/java/org/apache/jena/sparql/modify/UpdateVisitorSink.java

@@ -28,15 +28,15 @@
 import org.apache.jena.update.Update ;
 
 /**
- * UpdateSink that sends every Update to a worker except for the quads 
- * of INSERT DATA, DELETE DATA which do to special sinks.
+ * UpdateSink that sends every Update to a worker except for the quads
+ * of INSERT DATA, DELETE DATA which go to special sinks.
  */
 public class UpdateVisitorSink implements UpdateSink
 {
     private final UpdateVisitor worker;
     private final Sink<Quad> addSink;
     private final Sink<Quad> delSink;
-    
+
     public UpdateVisitorSink(UpdateVisitor worker, Sink<Quad> addSink, Sink<Quad> delSink) {
         this.worker = worker;
         this.addSink = addSink;
@@ -47,7 +47,7 @@ public UpdateVisitorSink(UpdateVisitor worker, Sink<Quad> addSink, Sink<Quad> de
     public void send(Update update) {
         update.visit(worker);
     }
-    
+
     // The sink for INSERT DATA, DELETE DATA to go straight to sink handlers.
     @Override
     public QuadDataAccSink createInsertDataSink() {

jena-arq/src/main/java/org/apache/jena/sparql/modify/UsingList.java

@@ -27,21 +27,62 @@
 import java.util.List;
 
 import org.apache.jena.graph.Node ;
+import org.apache.jena.sparql.modify.request.UpdateWithUsing;
+import org.apache.jena.update.Update;
+import org.apache.jena.update.UpdateException;
+import org.apache.jena.update.UpdateRequest;
 
 public class UsingList
 {
     public UsingList() { }
-    
+
     private List<Node> using = new ArrayList<>() ;
     private List<Node> usingNamed = new ArrayList<>() ;
-    
+
     public void addUsing(Node node)                      { using.add(node) ; }
     public void addAllUsing(Collection<Node> nodes)      { using.addAll(nodes); }
     public void addUsingNamed(Node node)                 { usingNamed.add(node) ; }
     public void addAllUsingNamed(Collection<Node> nodes) { usingNamed.addAll(nodes); }
-    
+
     public List<Node> getUsing()                         { return Collections.unmodifiableList(using) ; }
     public List<Node> getUsingNamed()                    { return Collections.unmodifiableList(usingNamed) ; }
-    
+
     public boolean usingIsPresent()                      { return using.size() > 0 || usingNamed.size() > 0 ; }
+
+    /**
+     * This modifies the {@link Update Updates} of the {@link UpdateRequest}.
+     * The using list may come from the protocol so this
+     * operation converts a request+protocol into a
+     * self-contained UpdateRequest.
+     */
+    public static UpdateRequest modifyUpdateForUsingList(UpdateRequest updateRequest, UsingList usingList) {
+        if ( usingList == null || ! usingList.usingIsPresent() )
+            return updateRequest;
+        UpdateRequest request = new UpdateRequest();
+        updateRequest.forEach(update->{
+            Update update2 = modifyUpdateForUsingList(update, usingList);
+            request.add(update2);
+        });
+        return request;
+    }
+
+    /**
+     * This modifies the {@link Update}.
+     * The using list may come from the protocol so this
+     * operation converts a request+protocol into a
+     * self-contained UpdateRequest.
+     */
+    public static Update modifyUpdateForUsingList(Update update, UsingList usingList) {
+        if ( usingList == null || ! usingList.usingIsPresent() )
+            return update;
+        if ( ! ( update instanceof UpdateWithUsing upu ) )
+            return update;
+        if ( upu.getUsing().size() != 0 || upu.getUsingNamed().size() != 0 || upu.getWithIRI() != null )
+            throw new UpdateException("SPARQL Update: Protocol using-graph-uri or using-named-graph-uri present where update request has USING, USING NAMED or WITH");
+        for ( Node node : usingList.getUsing() )
+            upu.addUsing(node);
+        for ( Node node : usingList.getUsingNamed() )
+            upu.addUsingNamed(node);
+        return update;
+    }
 }

jena-arq/src/main/java/org/apache/jena/sparql/modify/UsingUpdateSink.java

@@ -21,11 +21,8 @@
 
 package org.apache.jena.sparql.modify;
 
-import org.apache.jena.graph.Node ;
 import org.apache.jena.sparql.modify.request.QuadDataAccSink ;
-import org.apache.jena.sparql.modify.request.UpdateWithUsing ;
 import org.apache.jena.update.Update ;
-import org.apache.jena.update.UpdateException ;
 
 /**
  * Adds using clauses from the UsingList to UpdateWithUsing operations; will throw an
@@ -44,17 +41,8 @@ public UsingUpdateSink(UpdateSink sink, UsingList usingList) {
     public void send(Update update) {
         // ---- check USING/USING NAMED/WITH not used.
         // ---- update request to have USING/USING NAMED
-        if ( null != usingList && usingList.usingIsPresent() ) {
-            if ( update instanceof UpdateWithUsing ) {
-                UpdateWithUsing upu = (UpdateWithUsing)update;
-                if ( upu.getUsing().size() != 0 || upu.getUsingNamed().size() != 0 || upu.getWithIRI() != null )
-                    throw new UpdateException("SPARQL Update: Protocol using-graph-uri or using-named-graph-uri present where update request has USING, USING NAMED or WITH");
-                for ( Node node : usingList.getUsing() )
-                    upu.addUsing(node);
-                for ( Node node : usingList.getUsingNamed() )
-                    upu.addUsingNamed(node);
-            }
-        }
+        if ( null != usingList && usingList.usingIsPresent() )
+            update = UsingList.modifyUpdateForUsingList(update, usingList);
         sink.send(update);
     }
 

jena-arq/src/main/java/org/apache/jena/sparql/modify/request/UpdateLoad.java

@@ -35,7 +35,6 @@ public class UpdateLoad extends Update
     private final Node dest;
     private boolean silent;
 
-
     public UpdateLoad(String source, String dest) {
         this(source, NodeFactory.createURI(dest), false);
     }