DHFPROD-8164: Better handle massive match events

ryanjdew · MarkLogic Builder · commit 687975b2306d · 2021-12-27T10:37:36.000-08:00
diff --git a/marklogic-data-hub/src/main/resources/ml-modules/root/com.marklogic.smart-mastering/impl/process.xqy b/marklogic-data-hub/src/main/resources/ml-modules/root/com.marklogic.smart-mastering/impl/process.xqy
@@ -398,28 +398,59 @@ declare function proc-impl:build-match-summary(
     $match-options-node//(*:max-scan|maxScan) ! xs:integer(.),
     500
   ))
+  let $outlier-map := map:map()
+  let $estimate-map := map:map()
   let $all-matches :=
     let $start-elapsed := xdmp:elapsed-time()
     let $matches :=
       map:new((
-        $normalized-input !
-        map:entry(
-          (. => map:get("uri")),
-          let $match-results := matcher:find-document-matches-by-options(
-                (. => map:get("value")),
-                $match-options,
-                1,
-                $max-scan,
-                $minimum-threshold,
-                $fine-grain-provenance,
-                $filter-query
+        util-impl:process-items-in-set-time(
+          function ($input) {
+            map:entry(
+              ($input => map:get("uri")),
+              let $match-results := matcher:find-document-matches-by-options(
+                    ($input => map:get("value")),
+                    $match-options,
+                    1,
+                    $max-scan,
+                    $minimum-threshold,
+                    $fine-grain-provenance,
+                    $filter-query
+                  )
+               return (
+                 if ($lock-for-update) then
+                   $match-results/result[fn:exists(@action)]/@uri ! merge-impl:lock-for-update(fn:string(.))
+                 else (),
+                 $match-results
+               )
+            )
+          },
+          $normalized-input,
+          function($input) {
+            if (map:contains($estimate-map,($input => map:get("uri")))) then
+              map:get($estimate-map,($input => map:get("uri")))/@total ! xs:unsignedLong(.)
+            else
+              let $estimate := matcher:find-document-matches-by-options(
+                  ($input => map:get("value")),
+                  $match-options,
+                  1,
+                  $max-scan,
+                  $minimum-threshold,
+                  $fine-grain-provenance,
+                  $filter-query,
+                  fn:false()
+                )
+              return (
+                map:put($estimate-map,($input => map:get("uri")), $estimate-map),
+                $estimate/@total ! xs:unsignedLong(.)
               )
-           return (
-             if ($lock-for-update) then
-               $match-results/result[fn:exists(@action)]/@uri ! merge-impl:lock-for-update(fn:string(.))
-             else (),
-             $match-results
-           )
+          },
+          function($outliers) {
+            for $outlier in $outliers
+            let $uri := ($outlier => map:get("uri"))
+            let $output := map:get($estimate-map,$uri)
+            return map:put($outlier-map, $uri, $output)
+          }
         )
       ))
     return (
@@ -509,6 +540,17 @@ declare function proc-impl:build-match-summary(
     let $notifications :=
       (: Process notifications :)
       map:new((
+        let $outlier-threshold := "Unable to process"
+        for $outlier-uri in map:keys($outlier-map)
+        let $notification-uri := notify-impl:build-notification-uri($outlier-threshold, $outlier-uri)
+        return map:entry(
+                $notification-uri,
+                  map:map()
+                    => map:with("action", "notify")
+                    => map:with("threshold", $outlier-threshold)
+                    => map:with("uris", json:to-array($outlier-uri))
+                    => map:with("query", ($outlier-map => map:get($outlier-uri))/match-query/* ! cts:query(.))
+              ),
         for $notification in $consolidated-notifies
         let $_lock := if ($lock-for-update) then (merge-impl:lock-for-update($notification)) else ()
         let $parts := fn:tokenize($notification, $STRING-TOKEN)
@@ -739,9 +781,10 @@ declare function proc-impl:build-content-objects-from-match-summary(
                 )
               case "notify" return
                 let $uris := $action-details => map:get("uris") => json:array-values()
+                let $query := ($action-details => map:get("query")) ! cts:query(.)
                 let $threshold := $action-details => map:get("threshold")
                 let $provenance := $action-details => map:get("provenance")
-                let $match-write-object := matcher:build-match-notification($threshold, $uris, $merge-options-node)
+                let $match-write-object := matcher:build-match-notification($threshold, $uris, $merge-options-node, $query)
                 where fn:exists($match-write-object)
                 return
                   if (fn:exists($provenance)) then
diff --git a/marklogic-data-hub/src/main/resources/ml-modules/root/com.marklogic.smart-mastering/impl/util.xqy b/marklogic-data-hub/src/main/resources/ml-modules/root/com.marklogic.smart-mastering/impl/util.xqy
@@ -264,3 +264,66 @@ declare function util-impl:function-is-javascript(
 ) as xs:boolean {
   fn:exists($fun) and fn:ends-with(xdmp:function-module($fun), "js")
 };
+
+declare variable $SESSION_TIMEOUT := 600;
+
+(:
+ : Ensures each item is run in set amount of time
+ : @param $fun as function(item())
+ : @param $items as item()*
+ : @return item()*
+ :)
+declare function util-impl:process-items-in-set-time(
+  $fun,
+  $items as item()*,
+  $item-cost-fun,
+  $outlier-handler
+) {
+  try {
+    for $item in $items
+    return $fun($item)
+  } catch ($e) {
+    if ($e/error:code eq "XDMP-EXTIME") then (
+      let $items-cost := $items ! $item-cost-fun(.)
+      let $high-outliers-cost := util-impl:determine-high-outliers($items-cost)
+      let $high-outlier-indexes := $high-outliers-cost ! fn:index-of($items-cost, .)
+      let $high-outliers := $items[fn:position() = $high-outlier-indexes]
+      let $others := $items[fn:not(fn:position() = $high-outlier-indexes)]
+      return (
+        util-impl:process-items-in-set-time(
+          $fun,
+          $others,
+          $item-cost-fun,
+          $outlier-handler
+        ),
+        if (fn:exists($outlier-handler)) then
+          $outlier-handler($high-outliers)
+        else (),
+        for $outlier in $high-outliers
+        return
+          xdmp:log("Unable to process "|| xdmp:describe($outlier, (), ()) || " outlier item.", "warning")
+      )
+    ) else
+      xdmp:rethrow()
+  }
+};
+
+(:
+ : Returns high outliers
+ : @param $items as item()*
+ : @return item()*
+ :)
+declare function util-impl:determine-high-outliers(
+  $items as xs:unsignedLong*) {
+  let $ordered-items :=
+    for $i in $items
+    order by $i ascending
+    return $i
+  let $size := fn:count($items)
+  let $half := $size idiv 2
+  let $median-low := fn:avg(fn:subsequence($ordered-items, 1, $half))
+  let $median-high := fn:avg(fn:subsequence($ordered-items,$half + 1))
+  let $median-difference := $median-high - $median-low
+  let $outlier-threshold := $median-high + (1.5 * $median-difference)
+  return $items[. gt $outlier-threshold]
+};
diff --git a/marklogic-data-hub/src/main/resources/ml-modules/root/com.marklogic.smart-mastering/matcher-impl/helper-impl.xqy b/marklogic-data-hub/src/main/resources/ml-modules/root/com.marklogic.smart-mastering/matcher-impl/helper-impl.xqy
@@ -218,6 +218,7 @@ declare function helper-impl:group-queries-by-scope($queries as cts:query*, $gro
       for $query in $queries
       let $is-json-prop-scope := $query instance of cts:json-property-scope-query
       let $is-element-scope := $query instance of cts:element-query
+      let $is-and-not := $query instance of cts:and-not-query
       let $key :=
         if ($is-json-prop-scope) then
           "json-prop:" || fn:string-join(
@@ -229,6 +230,8 @@ declare function helper-impl:group-queries-by-scope($queries as cts:query*, $gro
             for $qn in cts:element-query-element-name($query) order by $qn return xdmp:key-from-QName($qn),
             $string-token
           )
+        else if ($is-and-not) then
+          "and-not"
         else
           "_other"
       let $values :=
@@ -253,6 +256,10 @@ declare function helper-impl:group-queries-by-scope($queries as cts:query*, $gro
           cts:json-property-scope-query(fn:tokenize(fn:substring-after($key, "json-prop:"), $string-token), $grouped-queries)
         else if (fn:starts-with($key, "element:")) then
           cts:element-query(fn:tokenize(fn:substring-after($key, "element:"), $string-token) ! xdmp:QName-from-key(.), $grouped-queries)
+        else if ($key eq "and-not" and fn:count($grouped-queries) gt 1) then
+          let $positive-queries := helper-impl:group-queries-by-scope(for $q in $grouped-queries return cts:and-not-query-positive-query($q), $grouping-query-fun)
+          let $negative-queries := helper-impl:group-queries-by-scope(for $q in $grouped-queries return cts:and-not-query-negative-query($q), $grouping-query-fun)
+          return cts:and-not-query($positive-queries, $negative-queries)
         else
           $grouped-queries
     return
diff --git a/marklogic-data-hub/src/main/resources/ml-modules/root/com.marklogic.smart-mastering/matcher-impl/matcher-impl.xqy b/marklogic-data-hub/src/main/resources/ml-modules/root/com.marklogic.smart-mastering/matcher-impl/matcher-impl.xqy
@@ -262,6 +262,9 @@ declare function match-impl:find-document-matches-by-options(
       else
         let $estimate := xdmp:estimate(cts:search(fn:collection(), $match-query, "unfiltered"))
         return (
+          if ($estimate ge 250) then
+            xdmp:log("A large number ("|| $estimate ||") of potential matches were discovered for document '" || $document-uri || "' with the following query: " || xdmp:describe($match-query, (), ()), "warning")
+          else (),
           if ($match-trace-is-enabled) then
             xdmp:trace($const:TRACE-MATCH-RESULTS, "Estimated " || $estimate || " doc(s) found for cts.doc('"|| $document-uri ||"') in " || xdmp:database-name(xdmp:database()))
           else (),
diff --git a/marklogic-data-hub/src/main/resources/ml-modules/root/com.marklogic.smart-mastering/matcher-impl/notification-impl.xqy b/marklogic-data-hub/src/main/resources/ml-modules/root/com.marklogic.smart-mastering/matcher-impl/notification-impl.xqy
@@ -67,6 +67,14 @@ declare function notify-impl:save-match-notification(
 };
 
 declare variable $_notifications-inserted := map:map();
+declare function notify-impl:build-match-notification(
+  $threshold-label as xs:string,
+  $uris as xs:string*,
+  $options as item()?
+) as map:map?
+{
+  notify-impl:build-match-notification($threshold-label, $uris, $options, ())
+};
 (:
  : Create a new notification document. If there is already a notification for
  : this combination of label and URIs, that notification will be replaced.
@@ -77,7 +85,8 @@ declare variable $_notifications-inserted := map:map();
 declare function notify-impl:build-match-notification(
   $threshold-label as xs:string,
   $uris as xs:string*,
-  $options as item()?
+  $options as item()?,
+  $query as cts:query?
 ) as map:map?
 {
   let $options-node := merge-impl:options-to-node($options)
@@ -109,7 +118,8 @@ declare function notify-impl:build-match-notification(
             element sm:threshold-label {$threshold-label},
             element sm:document-uris {
               $doc-uris
-            }
+            },
+            $query ! element sm:query {.}
           }
         ),
       map:entry("context",
diff --git a/marklogic-data-hub/src/main/resources/ml-modules/root/com.marklogic.smart-mastering/matcher-impl/options-impl.xqy b/marklogic-data-hub/src/main/resources/ml-modules/root/com.marklogic.smart-mastering/matcher-impl/options-impl.xqy
@@ -25,6 +25,8 @@ import module namespace helper-impl = "http://marklogic.com/smart-mastering/help
   at "/com.marklogic.smart-mastering/matcher-impl/helper-impl.xqy";
 import module namespace json="http://marklogic.com/xdmp/json"
   at "/MarkLogic/json/json.xqy";
+import module namespace sem = "http://marklogic.com/semantics"
+  at "/MarkLogic/semantics.xqy";
 import module namespace util-impl = "http://marklogic.com/smart-mastering/util-impl"
   at "/com.marklogic.smart-mastering/impl/util.xqy";
 
@@ -545,21 +547,7 @@ declare function opt-impl:compile-match-options(
         map:entry("orderedThresholds", $ordered-thresholds),
         map:entry("minimumThresholdCombinations", $minimum-threshold-combinations),
         map:entry("propertyNamesToValues", $property-names-to-values),
-        map:entry("baseContentQuery",
-          if (fn:exists($target-entity-type-def)) then
-            cts:or-query((
-              cts:json-property-scope-query(
-                "info",
-                cts:json-property-value-query("title", fn:string($target-entity-type-def/entityTitle), (), 0)
-              ),
-              cts:element-query(
-                xs:QName("es:info"),
-                cts:element-value-query(xs:QName("es:title"), fn:string($target-entity-type-def/entityTitle), (), 0)
-              )
-            ))
-          else
-            opt-impl:build-collection-query(coll:content-collections($match-options))
-        )
+        map:entry("baseContentQuery", opt-impl:build-base-query($match-options, $target-entity-type-def, $target-entity-type-iri))
       ))
     let $cache-ids := (
       $cache-id,
@@ -584,6 +572,30 @@ declare function opt-impl:compile-match-options(
     )
 };
 
+
+declare function opt-impl:build-base-query($match-options as item()?, $target-entity-type-def as item()?, $target-entity-type-iri as xs:string?) {
+  let $base-query :=
+    if (fn:exists($target-entity-type-def)) then
+      let $triple-query := cts:triple-range-query((), sem:curie-expand("rdf:type"), sem:iri($target-entity-type-iri))
+      return
+        if (xdmp:exists(cts:search(fn:collection(), $triple-query))) then
+          $triple-query
+        else
+          cts:or-query((
+            cts:json-property-scope-query(
+              "info",
+              cts:json-property-value-query("title", fn:string($target-entity-type-def/entityTitle), (), 0)
+            ),
+            cts:element-query(
+              xs:QName("es:info"),
+              cts:element-value-query(xs:QName("es:title"), fn:string($target-entity-type-def/entityTitle), (), 0)
+            )
+          ))
+    else
+      opt-impl:build-collection-query(coll:content-collections($match-options))
+  return cts:registered-query(cts:register($base-query))
+};
+
 declare function opt-impl:convert-match-rule-for-custom-module($match-rule, $match-options, $custom-algorithm)
 {
   if (fn:empty($custom-algorithm)) then
@@ -734,7 +746,7 @@ declare function opt-impl:minimum-threshold-combinations($query-results, $thresh
   else
     (: Each of $queries-ge-threshold has a weight high enough to hit the $threshold :)
     let $queries-ge-threshold := $query-results[fn:empty((. => map:get("weight"))) or (. => map:get("weight")) >= $threshold]
-    let $queries-lt-threshold := $query-results[(. => map:get("weight")) <= $threshold]
+    let $queries-lt-threshold := $query-results[(. => map:get("weight")) < $threshold]
     return (
       $queries-ge-threshold ! (map:entry("queries", .) => map:with("weight", map:get(., "weight"))),
       opt-impl:filter-for-required-queries($queries-lt-threshold, 0, $threshold, ())
diff --git a/marklogic-data-hub/src/main/resources/ml-modules/root/com.marklogic.smart-mastering/matcher.xqy b/marklogic-data-hub/src/main/resources/ml-modules/root/com.marklogic.smart-mastering/matcher.xqy
@@ -186,6 +186,36 @@ declare function matcher:find-document-matches-by-options(
   )
 };
 
+(:
+ : Starting with the specified document, look for a page of potential matches based on previously-saved matching options.
+ :
+ : @param $document  document to find matches for
+ : @param $options  match options saved using matcher:save-options
+ : @param $start  starting index for potential match results (starts at 1)
+ : @param $page-length  maximum number of results to return in this call
+ : @param $minimum-threshold  value of the lowest threshold score; the match query will require matches to score at
+                              least this high to be returned
+ : @param $include-matches  whether the response should list the matched properties for each potential match
+ : @param $filter-query  a cts:query used to restrict matches to a set, such as a specific entity type or collection
+ : @param $include-results  a boolean that determines if results should be retrieved or just an estimate
+ : @return the queries used for search and the search results themselves
+ : @see https://marklogic-community.github.io/smart-mastering-core/docs/match-results/
+ :)
+declare function matcher:find-document-matches-by-options(
+  $document,
+  $options as item(), (: as (element(matcher:options)|object-node()) :)
+  $start as xs:integer,
+  $page-length as xs:integer,
+  $minimum-threshold as xs:double,
+  $include-matches as xs:boolean,
+  $filter-query,
+  $include-results as xs:boolean
+) as element(results)
+{
+  match-impl:find-document-matches-by-options(
+    $document, $options, $start, $page-length, $minimum-threshold, $include-matches, $filter-query, $include-results
+  )
+};
 (:
  : Convert match results from XML to JSON.
  : @param $results-xml  XML match results as returned from the
@@ -489,6 +519,25 @@ declare function matcher:build-match-notification(
   notify-impl:build-match-notification($threshold-label, $uris, $options)
 };
 
+(:
+ : Builds a map action for new notification. If a notification document already exists for
+ : this label/URIs combination, it will be replaced with the new notification.
+ : @param $threshold-label  human-readable label used to indicate the
+ :                          likelihood of the match
+ : @param $uris  URIs of the content documents that are merge candidates
+ : @param $merge-options  merge options for determining notification collections
+ : @param $query  merge options for determining notification collections
+ : @return content of the newly-constructed notification
+ :)
+declare function matcher:build-match-notification(
+  $threshold-label as xs:string,
+  $uris as xs:string*,
+  $options as item()?,
+  $query as cts:query?
+) as map:map?
+{
+  notify-impl:build-match-notification($threshold-label, $uris, $options, $query)
+};
 (:
  : Delete the specified notification.
  : @param $uri  URI of the notification document to be deleted
diff --git a/marklogic-data-hub/src/test/java/com/marklogic/hub/master/MasterTest.java b/marklogic-data-hub/src/test/java/com/marklogic/hub/master/MasterTest.java
