Merge pull request #340 from fluree/update/simple-subject-crawl-strategy

re-enable simple-subject-crawl for some queries

Author: mpoffald

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

@@ -5,6 +5,7 @@
             [fluree.db.query.parse.aggregate :refer [parse-aggregate]]
             [fluree.db.query.json-ld.select :refer [parse-subselection]]
             [fluree.db.query.subject-crawl.legacy :refer [basic-to-analytical-transpiler]]
+            [fluree.db.query.subject-crawl.reparse :refer [re-parse-as-simple-subj-crawl]]
             [fluree.db.query.fql.syntax :as syntax]
             [clojure.string :as str]
             [clojure.set :as set]
@@ -410,7 +411,9 @@
   [q db]
   (if (basic-query? q)
     (parse-basic-query q db)
-    (parse-analytical-query q db)))
+    (let [parsed (parse-analytical-query q db)]
+      (or (re-parse-as-simple-subj-crawl parsed)
+          parsed))))
 
 (defn parse-delete
   [q db]

src/fluree/db/query/subject_crawl/core.cljc

@@ -7,7 +7,8 @@
             [fluree.db.query.subject-crawl.rdf-type :refer [collection-crawl]]
             [fluree.db.query.subject-crawl.common :refer [order-results]]
             [fluree.db.query.fql.resp :as legacy-resp]
-            [fluree.db.query.json-ld.response :as json-ld-resp]))
+            [fluree.db.query.json-ld.response :as json-ld-resp]
+            [fluree.json-ld :as json-ld]))
 
 #?(:clj (set! *warn-on-reflection* true))
 
@@ -17,7 +18,7 @@
   This strategy is only deployed if there is a single selection graph crawl,
   so this assumes this case is true in code."
   [db {:keys [select opts] :as _parsed-query}]
-  (-> select :spec first :spec))
+  (:spec select))
 
 (defn relationship-binding
   [{:keys [collection? vars] :as opts}]
@@ -63,8 +64,8 @@
   (c) filter subjects based on subsequent where clause(s)
   (d) apply offset/limit for (c)
   (e) send result into :select graph crawl"
-  [db {:keys [vars ident-vars where limit offset fuel rel-binding? order-by
-              compact-fn opts] :as parsed-query}]
+  [db {:keys [vars ident-vars where limit offset fuel rel-binding?
+              order-by opts] :as parsed-query}]
   (log/trace "Running simple subject crawl query:" parsed-query)
   (let [error-ch    (async/chan)
         f-where     (first where)
@@ -74,6 +75,7 @@
         cache       (volatile! {})
         fuel-vol    (volatile! 0)
         select-spec (retrieve-select-spec db parsed-query)
+        compact-fn  (->> parsed-query :context json-ld/compact-fn)
         result-fn   (partial json-ld-resp/flakes->res db cache compact-fn fuel-vol fuel select-spec 0)
         finish-fn   (build-finishing-fn parsed-query)
         opts        {:rdf-type?     rdf-type?

src/fluree/db/query/subject_crawl/reparse.cljc

@@ -1,7 +1,11 @@
 (ns fluree.db.query.subject-crawl.reparse
   (:require [fluree.db.util.log :as log :include-macros true]
             [fluree.db.flake :as flake]
-            [fluree.db.util.core :as util #?(:clj :refer :cljs :refer-macros) [try* catch*]]))
+            [fluree.db.util.core :as util #?(:clj :refer :cljs :refer-macros) [try* catch*]]
+            #?(:clj [fluree.db.query.exec.select]
+               :cljs [fluree.db.query.exec.select :refer [SubgraphSelector]])
+            [fluree.db.query.exec.where :as where])
+  #?(:clj (:import [fluree.db.query.exec.select SubgraphSelector])))
 
 #?(:clj (set! *warn-on-reflection* true))
 
@@ -19,6 +23,19 @@
                                 {:status 400 :error :db/invalid-query})))))
           [] params))
 
+(defn mergeable-where-clause?
+  "Returns true if a `where` clause is eligible for merging
+  with other clauses to create a filter for the simple-subject-crawl
+  strategy"
+  [where-clause]
+  (#{:class :tuple} (where/pattern-type where-clause)))
+
+(defn clause-subject-var
+  [where-clause]
+  (-> where-clause
+      first
+      ::where/var))
+
 (defn merge-wheres-to-filter
   "Merges all subsequent where clauses (rest where) for simple-subject-crawl
   into a map containing predicate filters.
@@ -38,87 +55,104 @@
    Note that for multi-cardinality predicates, the prediate filters must pass for just one flake
   "
   [first-s rest-where supplied-vars]
-  (loop [[{:keys [type s p o] :as where-smt} & r] rest-where
+  (loop [[{:keys [s p o] :as where-smt} & r] rest-where
          required-p #{} ;; set of 'p' values that are going to be required for a subject to have
          filter-map {}] ;; key 'p' value, val is list of filtering fns
-    (if where-smt
-      (when (and (= :tuple type)
-                 (= first-s (:variable s)))
-        (let [{:keys [value filter variable]} o
-              f (cond
-                  value
-                  (fn [flake _] (= value (flake/o flake)))
+    (let [[s p o] where-smt
+          {p* ::where/val} p
+          type (where/pattern-type where-smt)]
+      (if where-smt
+        (when (and (= :tuple type)
+                   (= first-s (clause-subject-var where-smt)))
+          (let [{::where/keys [val var]} o 
+                f (cond
+                    val
+                    (fn [flake _] (= val (flake/o flake)))
 
-                  filter
-                  (let [{:keys [params variable function]} filter]
-                    (if (= 1 (count params))
-                      (fn [flake _] (function (flake/o flake)))
-                      (fn [flake vars]
-                        (let [params (fill-fn-params params (flake/o flake) variable vars)]
-                          (log/debug (str "Calling query-filter fn: " (:fn-str filter)
-                                          "with params: " params "."))
-                          (apply function params)))))
+                    ;;TODO: filters are not yet supported 
+                    #_#_filter
+                    (let [{:keys [params variable function]} filter]
+                      (if (= 1 (count params))
+                        (fn [flake _] (function (flake/o flake)))
+                        (fn [flake vars]
+                          (let [params (fill-fn-params params (flake/o flake) variable vars)]
+                            (log/debug (str "Calling query-filter fn: " ("fn-str" filter)
+                                            "with params: " params "."))
+                            (apply function params)))))
+                    ;;TODO: vars are not yet supported
+                    #_#_(and var (get supplied-vars var))
+                    (fn [flake vars]
+                      (= (flake/o flake) (get vars var))))]
+            (recur r
+                   (conj required-p p*)
+                   (if f
+                     (update filter-map p* util/conjv f)
+                     filter-map))))
+        (assoc filter-map :required-p required-p)))))
 
-                  (and variable (supplied-vars variable))
-                  (fn [flake vars]
-                    (= (flake/o flake) (get vars variable))))]
-          (recur r
-                 (conj required-p p)
-                 (if f
-                   (update filter-map p util/conjv f)
-                   filter-map))))
-      (assoc filter-map :required-p required-p))))
 
+(defn re-parse-pattern
+  "Re-parses a pattern into the format recognized
+  by downstream simple-subject-crawl code"
+  [pattern]
+  (let [type (where/pattern-type pattern)
+        [s p o] (if (= :tuple type)
+                  pattern
+                  (let [[_type-kw tuple] pattern]
+                    tuple)) 
+        reparse-component (fn [component]
+                            (let [{::where/keys [var val]} component]
+                              (cond
+                                var {:variable var}
+                                val {:value val})))]
+    {:type type
+     :s (reparse-component s)
+     :p (reparse-component p)
+     :o (assoc (reparse-component o) :datatype (::where/datatype o))}))
 
 (defn simple-subject-merge-where
   "Revises where clause for simple-subject-crawl query to optimize processing.
   If where does not end up meeting simple-subject-crawl criteria, returns nil
   so other strategies can be tried."
-  [{:keys [where supplied-vars] :as parsed-query}]
-  (let [first-where (first where)
-        rest-where  (rest where)
-        first-type  (:type first-where)
-        first-s     (when (and (#{:rdf/type :collection :_id :iri :tuple} first-type)
-                               (-> first-where :s :variable))
-                      (-> first-where :s :variable))]
-    (when first-s
-      (if (empty? rest-where)
-        (assoc parsed-query :strategy :simple-subject-crawl)
-        (if-let [subj-filter-map (merge-wheres-to-filter first-s rest-where supplied-vars)]
-          (assoc parsed-query :where [first-where
+  [{:keys [where vars] :as parsed-query}]
+  (let [{::where/keys [patterns]} where
+        [first-pattern & rest-patterns] patterns
+        reparsed-first-clause (re-parse-pattern first-pattern)]
+    (when-let [first-s (and (mergeable-where-clause? first-pattern)
+                            (clause-subject-var first-pattern))]
+      (if (empty? rest-patterns)
+        (assoc parsed-query
+               :where [reparsed-first-clause]
+               :strategy :simple-subject-crawl)
+        (if-let [subj-filter-map (merge-wheres-to-filter first-s rest-patterns vars)]
+          (assoc parsed-query :where [reparsed-first-clause
                                       {:s-filter subj-filter-map}]
                               :strategy :simple-subject-crawl))))))
-
-(defn subject-crawl?
-  "Returns true if, when given parsed query, the select statement is a
-  subject crawl - meaning there is nothing else in the :select except a
-  graph crawl on a list of subjects"
-  [{:keys [select] :as _parsed-query}]
-  (and (:expandMaps? select)
-       (not (:inVector? select))))
-
 (defn simple-subject-crawl?
   "Simple subject crawl is where the same variable is used in the leading
   position of each where statement."
-  [{:keys [where select] :as _parsed-query}]
-  (let [select-var (or (-> select :select first :variable)  ;; legacy select spec
-                       (-> select :spec first :variable))]
-    (when select-var                                        ;; for now exclude any filters on the first where, not implemented
-      (every? #(and (= select-var (-> % :s :variable))
-                    ;; exclude if any recursion specified in where statement (e.g. person/follows+3)
-                    (not (:recur %)))
-              where))))
+  [{:keys [where select vars] :as _parsed-query}]
+  (and (instance? SubgraphSelector select)
+       ;;TODO, filtering not supported yet
+       (empty? (::where/filters where))
+       ;;TODO: vars support not complete
+       (empty? vars)
+       (if-let [{select-var :var} select]
+         (let [{::where/keys [patterns]} where]
+           (every? (fn [pattern]
+                     (let [pred (second pattern)]
+                       (and (= select-var (clause-subject-var pattern))
+                            (not (::where/recur pred))
+                            (not (::where/fullText pred))))) patterns)))))
 
 (defn re-parse-as-simple-subj-crawl
   "Returns true if query contains a single subject crawl.
   e.g.
   {:select {?subjects ['*']}
    :where [...]}"
-  [{:keys [op-type order-by group-by] :as parsed-query}]
-  (when (and (= :select op-type)
-             (subject-crawl? parsed-query)
-             (simple-subject-crawl? parsed-query)
-             (not group-by)
-             (not= :variable (:type order-by)))
+  [{:keys [order-by group-by] :as parsed-query}]
+  (when (and (not group-by)
+             (not order-by)
+             (simple-subject-crawl? parsed-query))
     ;; following will return nil if parts of where clause exclude it from being a simple-subject-crawl
     (simple-subject-merge-where parsed-query)))

src/fluree/db/query/subject_crawl/subject.cljc

@@ -7,6 +7,8 @@
             [fluree.db.flake :as flake]
             [fluree.db.util.core :as util #?(:clj :refer :cljs :refer-macros) [try* catch*]]
             [fluree.db.util.log :as log :include-macros true]
+            [fluree.db.query.exec :refer [drop-offset take-limit]]
+            [fluree.db.query.exec.where :as where]
             [fluree.db.query.subject-crawl.common :refer [where-subj-xf result-af resolve-ident-vars
                                                           subj-perm-filter-fn filter-subject]]
             [fluree.db.dbproto :as dbproto]))
@@ -16,18 +18,17 @@
 (defn- subjects-chan
   "Returns chan of subjects in chunks per index-leaf
   that can be pulled as needed based on the selection criteria of a where clause."
-  [{:keys [conn novelty t] :as db} error-ch vars {:keys [p o idx p-ref?] :as _where-clause}]
+  [{:keys [conn novelty t] :as db} error-ch vars {:keys [p o p-ref?] :as _where-clause}]
   (let [o*          (if-some [v (:value o)]
                       v
                       (when-let [variable (:variable o)]
                         (get vars variable)))
         p*          (:value p)
-        idx*        (if (nil? o*)
-                      :psot
-                      idx)
+        idx*        (where/idx-for nil p* o*) 
+        o-dt        (:datatype o)
         [fflake lflake] (case idx*
-                          :post [(flake/create nil p* o* nil nil nil util/min-integer)
-                                 (flake/create nil p* o* nil nil nil util/max-integer)]
+                          :post [(flake/create nil p* o* o-dt nil nil util/min-integer)
+                                 (flake/create nil p* o* o-dt nil nil util/max-integer)]
                           :psot [(flake/create nil p* nil nil nil nil util/min-integer)
                                  (flake/create nil p* nil nil nil nil util/max-integer)])
         filter-fn   (cond
@@ -156,7 +157,9 @@
                       (subjects-id-chan db error-ch vars* f-where*)
                       (subjects-chan db error-ch vars* f-where*))
           flakes-af (flakes-xf opts*)
-          flakes-ch (async/chan 32 (comp (drop offset) (take limit)))
+          flakes-ch (->> (async/chan 32)
+                         (drop-offset f-where)
+                         (take-limit f-where))
           result-ch (async/chan)]
 
       (async/pipeline-async parallelism flakes-ch flakes-af sid-ch)

test/fluree/db/query/filter_query_test.clj

@@ -67,6 +67,8 @@
                                          ['?s :schema/name '?name]
                                          ['?s :ex/last '?last]
                                          {:filter ["(> ?age 45)", "(strEnds ?last \"ith\")"]}]}))))
+    ;;TODO: simple-subject-crawl does not yet support filters.
+    ;;these are being run as regular analytial queries
     (testing "simple-subject-crawl"
       (is (= [{:id :ex/david,
 	      :rdf/type [:ex/User],