aggregate on full-text search; avoids Mongo limits

A full-text search filter to /api/evidence with a sort on bel_statement
only used the text index. This means that the bel_statement sort had to
be done in memory.

This reaches the 32 MB sort limit with only several tens of thousands of
documents.

The solution employed here was to use cursored aggregation allowing disk
use for sort stages.

The solution was introduced as an alternative code path if a FTS filter
was included in the HTTP request. Although this did minimize the risk of
regression there is a fair bit of to clean up in the mongo
access layer.

closes #96

Author: Anthony Bargnesi

lib/openbel/api/evidence/mongo.rb

@@ -51,6 +51,15 @@ def find_evidence_by_id(value)
       end
 
       def find_evidence(filters = [], offset = 0, length = 0, facet = false, facet_value_limit = -1)
+        if includes_fts_search?(filters)
+          text_search = get_fts_search(filters)
+          evidence_aggregate(text_search, filters, offset, length, facet, facet_value_limit)
+        else
+          evidence_query(filters, offset, length, facet, facet_value_limit)
+        end
+      end
+
+      def evidence_query(filters = [], offset = 0, length = 0, facet = false, facet_value_limit = -1)
         query_hash = to_query(filters)
         query_opts = query_options(
           query_hash,
@@ -72,6 +81,75 @@ def find_evidence(filters = [], offset = 0, length = 0, facet = false, facet_val
         results
       end
 
+      def evidence_aggregate(text_search, filters = [], offset = 0, length = 0, facet = false, facet_value_limit = -1)
+        match_filters = filters.select { |filter|
+          filter['category'] != 'fts'
+        }
+        match = build_filter_query(match_filters)
+        match[:$and].unshift({
+          :$text => {
+            :$search => text_search
+          }
+        })
+
+        pipeline = [
+          {
+            :$match => match
+          },
+          {
+            :$project => {
+              :_id           => 1,
+              :bel_statement => 1,
+              :score => {
+                :$meta => 'textScore'
+              }
+            }
+          },
+          {
+            :$sort => {
+              :score => {
+                :$meta => 'textScore'
+              },
+              :bel_statement => 1
+            }
+          }
+        ]
+
+        if offset > 0
+          pipeline << {
+            :$skip => offset
+          }
+        end
+
+        if length > 0
+          pipeline << {
+            :$limit => length
+          }
+        end
+
+        fts_cursor = @collection.aggregate(pipeline, {
+          :allowDiskUse => true,
+          :cursor       => {
+            :batchSize => 0
+          }
+        })
+        _ids = fts_cursor.map { |doc| doc['_id'] }
+
+        facets =
+          if facet
+            query_hash = to_query(filters)
+            facets_cursor = @evidence_facets.find_facets(query_hash, filters, facet_value_limit)
+            facets_cursor.to_a
+          else
+            nil
+          end
+
+        {
+          :cursor => @collection.find({:_id => {:$in => _ids}}),
+          :facets => facets
+        }
+      end
+
       def find_dataset_evidence(dataset, filters = [], offset = 0, length = 0, facet = false, facet_value_limit = -1)
         query_hash = to_query(filters)
         query_hash[:$and] ||= []
@@ -197,6 +275,54 @@ def ensure_all_indexes
 
       private
 
+      def includes_fts_search?(filters)
+        filters.any? { |filter|
+          filter['category'] == 'fts' && filter['name'] == 'search'
+        }
+      end
+
+      def get_fts_search(filters)
+        fts_filter = filters.find { |filter|
+          filter['category'] == 'fts' && filter['name'] == 'search'
+        }
+        fts_filter.fetch('value', '')
+      end
+
+      def build_filter_query(filters = [])
+        {
+          :$and => filters.map { |filter|
+            category = filter['category']
+            name     = filter['name']
+            value    = filter['value']
+
+            case category
+            when 'experiment_context'
+              {
+                :experiment_context => {
+                  :$elemMatch => {
+                    :name  => name.to_s,
+                    :value => value.to_s
+                  }
+                }
+              }
+            when 'metadata'
+              {
+                :metadata => {
+                  :$elemMatch => {
+                    :name  => name.to_s,
+                    :value => value.to_s
+                  }
+                }
+              }
+            else
+              {
+                "#{filter['category']}.#{filter['name']}" => filter['value'].to_s
+              }
+            end
+          }
+        }
+      end
+
       def to_query(filters = [])
         if !filters || filters.empty?
           return {}