Feature/sparql fed query

resources/sparql.bnf

@@ -65,7 +65,11 @@ TriplesBlock	  ::=  WS TriplesSameSubjectPath WS ( <'.'> TriplesBlock? WS )?
 GraphPatternNotTriples	  ::=  	GroupOrUnionGraphPattern | OptionalGraphPattern | MinusGraphPattern | GraphGraphPattern | ServiceGraphPattern | Filter | Bind | InlineData
 OptionalGraphPattern	  ::=  	<'OPTIONAL'> GroupGraphPattern
 GraphGraphPattern	  ::=  	<'GRAPH'> WS VarOrIri WS GroupGraphPattern
-ServiceGraphPattern	  ::=  	<'SERVICE'> WS 'SILENT'? WS VarOrIri GroupGraphPattern
+ServiceGraphPattern ::= <'SERVICE'> WS 'SILENT'? WS VarOrIri ServiceClause
+ServiceClause ::= WS '{' ServiceContent '}' WS
+<ServiceContent> ::= (NonBrace | NestedBrace)*
+<NestedBrace> ::= '{' ServiceContent '}'
+<NonBrace> ::= #"[^{}]"
 Bind	  ::=  	<'BIND' WS '(' WS>  Expression <As> Var <WS ')' WS>
 InlineData	  ::=  	<'VALUES'> WS DataBlock
 <DataBlock>	  ::=  	InlineDataOneVar | InlineDataFull

src/fluree/db/query/api.cljc

@@ -241,8 +241,11 @@
                                                  (sanitize-query-options override-opts))
 
           tracker         (track/init opts)
-          default-aliases (some-> sanitized-query :from util/sequential)
-          named-aliases   (some-> sanitized-query :from-named util/sequential)]
+          default-aliases (or (some-> opts :from util/sequential)
+                              (some-> opts :ledger util/sequential)
+                              (some-> sanitized-query :from util/sequential))
+          named-aliases   (or (some-> opts :from-named util/sequential)
+                              (some-> sanitized-query :from-named util/sequential))]
       (if (or (seq default-aliases)
               (seq named-aliases))
         (let [ds            (<? (load-dataset conn tracker default-aliases named-aliases sanitized-query))

src/fluree/db/query/exec/where.cljc

@@ -11,7 +11,9 @@
             [fluree.db.track :as track]
             [fluree.db.util :as util :refer [try* catch*]]
             [fluree.db.util.async :refer [<? empty-channel]]
+            [fluree.db.util.json :as json]
             [fluree.db.util.log :as log :include-macros true]
+            [fluree.db.util.xhttp :as xhttp]
             [fluree.json-ld :as json-ld])
   #?(:clj (:import (clojure.lang MapEntry))))
 
@@ -896,6 +898,62 @@
         (when-not (::invalidated solution*)
           solution*)))))
 
+(defn binding->solution
+  [solution vars binding]
+  (reduce (fn [solution* var]
+            (if-let [{type "type" v "value" dt "datatype" lang "xml:lang"}
+                     (get binding var)]
+              (let [var-name (symbol (str "?" var))
+                    mch      (cond (= "literal" type)
+                                   (cond-> (-> (unmatched-var var-name)
+                                               (match-value v dt))
+                                     lang (match-lang v lang))
+                                   (#{"uri" "bnode"} type)
+                                   (-> (unmatched-var var-name)
+                                       (match-iri v))
+                                   :else
+                                   (throw (ex-info "Invalid SPARQL Query Results JSON Format."
+                                                   {:status 400, :error :db/invalid-query
+                                                    :spec "https://www.w3.org/TR/sparql11-results-json"
+                                                    :binding binding})))]
+                (or
+                  ;; add new var binding to solution
+                 (update-solution-binding solution* var-name mch)
+                  ;; already have a binding for the given var, no join
+                 (reduced nil)))
+              solution*))
+          solution
+          vars))
+
+(defn sparql-service-error!
+  [ex service sparql-q]
+  (log/error ex "Error processing service response " service sparql-q)
+  (ex-info (str "Error processing service response " service " due to exception: " (ex-message ex))
+           {:status 400, :error :db/invalid-query}
+           ex))
+
+(defmethod match-pattern :service
+  [_db _tracker solution pattern error-ch]
+  (let [{:keys [service silent? sparql-q]} (pattern-data pattern)
+        solution-ch                        (async/chan)]
+    (go
+      (let [response (async/<! (xhttp/post service sparql-q
+                                           {:headers {"Content-Type" "application/sparql-query"
+                                                      "Accept"       "application/sparql-results+json"}}))]
+        (if (util/exception? response)
+          (if silent?
+            (async/onto-chan! solution-ch [solution])
+            (async/>! error-ch (sparql-service-error! response service sparql-q)))
+          (try*
+            (let [response* (json/parse response false)
+                  vars      (-> response* (get "head") (get "vars"))
+                  bindings  (-> response* (get "results") (get "bindings"))]
+              (->> bindings
+                   (keep (partial binding->solution solution vars))
+                   (async/onto-chan! solution-ch)))
+            (catch* e (async/>! error-ch (sparql-service-error! e service sparql-q)))))))
+    solution-ch))
+
 (defmethod match-pattern :default
   [_db _tracker _solution pattern error-ch]
   (go

src/fluree/db/query/fql/parse.cljc

@@ -1,6 +1,7 @@
 (ns fluree.db.query.fql.parse
   (:require #?(:cljs [cljs.reader :refer [read-string]])
             [clojure.set :as set]
+            [clojure.string :as str]
             [clojure.walk :refer [postwalk]]
             [fluree.db.constants :as const]
             [fluree.db.datatype :as datatype]
@@ -831,6 +832,13 @@
                        (parse-query* context))]
     [(where/->pattern :query sub-query*)]))
 
+(defmethod parse-pattern :service
+  [[_ {:keys [clause] :as data}] _var-config context]
+  (let [sparql (str/join " " (into (sparql/context->prefixes context)
+                                   ["SELECT *"
+                                    (str "WHERE " clause)]))]
+    [(where/->pattern :service (assoc data :sparql-q sparql))]))
+
 (defn parse-query
   [q]
   (log/trace "parse-query" q)

src/fluree/db/query/sparql.cljc

@@ -70,3 +70,26 @@
 (defn sparql-format?
   [opts]
   (= :sparql (:format opts)))
+
+(defn extract-prefix
+  "A context key is a prefix if:
+   - it is a string with no colon
+   - it is a keyword with no namespace
+  Returns the string prefix if it is a prefix, falsey if not."
+  [k]
+  (or
+   (and (string? k) (not (str/includes? k ":")) k)
+   (and (keyword? k) (not (namespace k)) (name k))))
+
+(defn context->prefixes
+  [parsed-context]
+  (reduce-kv (fn [prefixes k v]
+               (if-let [prefix (extract-prefix k)]
+                 (case prefix
+                   "base"  (conj prefixes (str "BASE <" v ">"))
+                   "vocab" prefixes    ; not supported in SPARQL
+                   ;; else
+                   (conj prefixes (str "PREFIX "  prefix ": <" (:id v) ">")))
+                 prefixes))
+             []
+             (dissoc parsed-context :type-key)))

src/fluree/db/query/sparql/translator.cljc

@@ -558,10 +558,20 @@
   [[_ & patterns]]
   (into [:minus] (mapv parse-term patterns)))
 
+(defmethod parse-term :ServiceClause
+  [[_ & chars]]
+  (apply str chars))
+
+(defn service-pattern
+  [silent? [service clause]]
+  [:service {:silent? silent? :service (parse-term service) :clause (parse-term clause)}])
+
 (defmethod parse-term :ServiceGraphPattern
-  [_]
-  (throw (ex-info "SERVICE is not a supported SPARQL pattern"
-                  {:status 400 :error :db/invalid-query})))
+  ;; ServiceGraphPattern ::= <'SERVICE'> WS 'SILENT'? WS VarOrIri GroupGraphPattern
+  [[_ & terms]]
+  (if (= "SILENT" (first terms))
+    (service-pattern true (rest terms))
+    (service-pattern false terms)))
 
 (defmethod parse-term :DataBlockValue
   ;; DataBlockValue ::= iri | RDFLiteral | NumericLiteral | BooleanLiteral | 'UNDEF' WS

src/fluree/db/validation.cljc

@@ -309,10 +309,14 @@
     ::bind                 [:+ {:error/message "bind values must be mappings from variables to functions"}
                             [:catn [:var ::var]
                              [:binding [:or ::function ::literal]]]]
+    ::service              [:map
+                            [:service :string]
+                            [:silent? :boolean]
+                            [:clause  :string]]
     ::where-op             [:and
                             :keyword
-                            [:enum {:error/message "unrecognized where operation, must be one of: graph, filter, optional, union, bind, values, exists, not-exists, minus"}
-                             :graph :filter :optional :union :bind :query :values :exists :not-exists :minus]]
+                            [:enum {:error/message "unrecognized where operation, must be one of: graph, filter, optional, union, bind, values, exists, not-exists, minus, service"}
+                             :graph :filter :optional :union :bind :query :values :exists :not-exists :minus :service]]
     ::graph                [:orn {:error/message "value of graph. Must be a ledger name or variable"}
                             [:ledger ::ledger]
                             [:variable ::var]]
@@ -360,6 +364,7 @@
                                     [:op ::where-op]
                                     [:bindings ::bind]]]
                             [:graph [:tuple ::where-op ::graph [:ref ::where]]]
+                            [:service [:tuple ::where-op ::service]]
                             ;; TODO - because ::subquery is a separate registry it cannot be called here, validated in f.d.q.fql.syntax/coerce-subquery until resolved
                             [:query [:catn
                                      [:op ::where-op]

test/fluree/db/query/fql_test.clj

@@ -652,3 +652,33 @@
                    :where [{:id '?s :ex/foo '?foo}]}]
         (is (= "Error in value for \"select\"; Select must be a valid selector, a wildcard symbol (`*`), or a vector of selectors; Provided: [:* ?foo];  See documentation for details: https://next.developers.flur.ee/docs/reference/errorcodes#query-invalid-select"
                (ex-message @(fluree/query db query))))))))
+
+(deftest service-test
+  (let [conn @(fluree/connect-memory)
+        db0  @(fluree/create conn "service-test")
+        db1  @(fluree/update db0 {"@context" {"ex" "http://example.com/"}
+                                  "insert" (mapv #(do {"@id" (str "ex:" %)
+                                                       "@type" (if (odd? %) "ex:Odd" "ex:Even")
+                                                       "ex:name" (str "name" %)
+                                                       "ex:num" [(dec %) % (inc %)]
+                                                       "ex:ref" {"@id" (str "ex:" (inc %))}})
+                                                 (range 10))})]
+    (testing "when service endpoint is nonresponsive"
+      (testing ":service returns an error"
+        (is (= "Error processing service response http://localhost:10000/sparql due to exception: xhttp error - http://localhost:10000/sparql - "
+               (ex-message @(fluree/query db1 {"@context" {"ex" "http://example.com/"}
+                                               "where" [{"@id" "?s" "@type" "ex:Even" "ex:ref" "?ref"}
+                                                        [:service {:silent? false :service "http://localhost:10000/sparql"
+                                                                   :clause "{ ?rev ex:name ?name ; a ?type .}"}]]
+                                               "select" ["?s" "?type"]})))))
+      (testing ":service silent returns a response without bindings from the remote service"
+        (is (= [["ex:0" nil]
+                ["ex:2" nil]
+                ["ex:4" nil]
+                ["ex:6" nil]
+                ["ex:8" nil]]
+               @(fluree/query db1 {"@context" {"ex" "http://example.com/"}
+                                   "where" [{"@id" "?s" "@type" "ex:Even" "ex:ref" "?ref"}
+                                            [:service {:silent? true :service "http://localhost:10000/sparql"
+                                                       :clause "{ ?rev ex:name ?name ; a ?type .}"}]]
+                                   "select" ["?s" "?type"]})))))))

test/fluree/db/query/sparql_test.cljc

@@ -738,7 +738,40 @@
                 {:select ["?y" "(as (min ?name) ?minName)"],
                  :where [{"@id" "?y", ":name" "?name"}],
                  :groupBy ["?y"]}]]}
-             (sparql/->fql query))))))
+             (sparql/->fql query)))))
+  (testing "SERVICE"
+    (let [q "PREFIX : <http://example.com/>
+             SELECT ?foo ?bar
+             WHERE {
+               ?s :foo ?foo .
+               SERVICE <https://query.wikidata.org/sparql> {
+                 ?s :bar ?bar .
+               }
+             }"]
+      (is (= [{"@id" "?s", ":foo" "?foo"}
+              [:service
+               {:silent? false,
+                :service "https://query.wikidata.org/sparql",
+                :clause "{
+                 ?s :bar ?bar .
+               }"}]]
+             (:where (sparql/->fql q)))))
+    (let [q "PREFIX : <http://example.com/>
+             SELECT ?foo ?bar
+             WHERE {
+               ?s :foo ?foo .
+               SERVICE SILENT <https://query.wikidata.org/sparql> {
+                 ?s :bar ?bar .
+               }
+             }"]
+      (is (= [{"@id" "?s", ":foo" "?foo"}
+              [:service
+               {:silent? true,
+                :service "https://query.wikidata.org/sparql",
+                :clause "{
+                 ?s :bar ?bar .
+               }"}]]
+             (:where (sparql/->fql q)))))))
 
 (deftest parse-prefixes
   (testing "PREFIX"