resort every segment

When using segmented fetch we currently assume that documents are stored in index patterns based on time, but this is no long necessary with the use of the field stats API. Since documents can be in any order we need to fetch documents from every index and then sort them client side. Since fetching the documents can take some time, we are using the bounds of the time field in an index to try and identify indices which can't produce hits for a result set, and then not fetch their documents.

It works like this:
 - get the list of indices and the min/max of their timefield
 - fetch the entire sample size from each index until we have enough documents to satisfy the sample.
 - for each remaining index pattern
   - if the min/max of it's timefield overlaps the min/max of the fetched documents also fetch the documents for that index pattern
   - otherwise fetch with size=0

Fixes #5642

Author: spalger

src/fixtures/stubbed_logstash_index_pattern.js

@@ -1,6 +1,6 @@
 define(function (require) {
   return function stubbedLogstashIndexPatternService(Private) {
-    var StubIndexPattern = Private(require('testUtils/stubIndexPattern'));
+    var StubIndexPattern = Private(require('testUtils/stub_index_pattern'));
     var fieldTypes = Private(require('ui/index_patterns/_field_types'));
     var mockLogstashFields = Private(require('fixtures/logstash_fields'));
 

src/fixtures/stubbed_search_source.js

@@ -2,8 +2,9 @@ define(function (require) {
   var sinon = require('auto-release-sinon');
   var searchResponse = require('fixtures/search_response');
 
-  return function stubSearchSource(Private, $q) {
+  return function stubSearchSource(Private, $q, Promise) {
     var deferedResult = $q.defer();
+    var indexPattern = Private(require('fixtures/stubbed_logstash_index_pattern'));
 
     return {
       sort: sinon.spy(),
@@ -13,7 +14,7 @@ define(function (require) {
       get: function (param) {
         switch (param) {
           case 'index':
-            return Private(require('fixtures/stubbed_logstash_index_pattern'));
+            return indexPattern;
           default:
             throw new Error('Param "' + param + '" is not implemented in the stubbed search source');
         }
@@ -29,7 +30,9 @@ define(function (require) {
         return deferedResult.promise;
       },
       onError: function () { return $q.defer().promise; },
-
+      _flatten() {
+        return Promise.resolve({ index: indexPattern, body: {} });
+      }
     };
 
   };

src/plugins/kibana/public/discover/controllers/discover.js

@@ -339,13 +339,14 @@ define(function (require) {
       }());
 
       var sortFn = null;
-      if (sortBy === 'non-time') {
+      if (sortBy !== 'implicit') {
         sortFn = new HitSortFn(sort[1]);
       }
 
       $scope.updateTime();
       if (sort[0] === '_score') segmented.setMaxSegments(1);
       segmented.setDirection(sortBy === 'time' ? (sort[1] || 'desc') : 'desc');
+      segmented.setSortFn(sortFn);
       segmented.setSize(sortBy === 'time' ? $scope.opts.sampleSize : false);
 
       // triggered when the status updated
@@ -364,30 +365,26 @@ define(function (require) {
             return failure.index + failure.shard + failure.reason;
           });
         }
+      }));
 
-        var rows = $scope.rows;
-        var indexPattern = $scope.searchSource.get('index');
+      segmented.on('mergedSegment', function (merged) {
+        $scope.mergedEsResp = merged;
+        $scope.hits = merged.hits.total;
 
-        // merge the rows and the hits, use a new array to help watchers
-        rows = $scope.rows = rows.concat(resp.hits.hits);
+        var indexPattern = $scope.searchSource.get('index');
 
-        if (sortFn) {
-          notify.event('resort rows', function () {
-            rows.sort(sortFn);
-            rows = $scope.rows = rows.slice(0, totalSize);
-            $scope.fieldCounts = {};
-          });
-        }
+        // the merge rows, use a new array to help watchers
+        $scope.rows = merged.hits.hits.slice();
 
         notify.event('flatten hit and count fields', function () {
-          var counts = $scope.fieldCounts;
+          var counts = $scope.fieldCounts = (sortFn ? {} : $scope.fieldCounts) || {};
           $scope.rows.forEach(function (hit) {
-            // skip this work if we have already done it and we are NOT sorting.
-            // ---
+            // skip this work if we have already done it
+            if (hit.$$_counted) return;
+
             // when we are sorting results, we need to redo the counts each time because the
-            // "top 500" may change with each response
-            if (hit.$$_counted && !sortFn) return;
-            hit.$$_counted = true;
+            // "top 500" may change with each response, so don't mark this as counted
+            if (!sortFn) hit.$$_counted = true;
 
             var fields = _.keys(indexPattern.flattenHit(hit));
             var n = fields.length;
@@ -398,13 +395,6 @@ define(function (require) {
             }
           });
         });
-
-      }));
-
-      segmented.on('mergedSegment', function (merged) {
-        $scope.mergedEsResp = merged;
-        $scope.hits = merged.hits.total;
-
       });
 
       segmented.on('complete', function () {

src/testUtils/stubIndexPattern.js renamed to src/testUtils/stub_index_pattern.js

@@ -23,6 +23,13 @@ define(function (require) {
       this.routes = IndexPattern.prototype.routes;
 
       this.toIndexList = _.constant(Promise.resolve([pattern]));
+      this.toDetailedIndexList = _.constant(Promise.resolve([
+        {
+          index: pattern,
+          min: 0,
+          max: 1
+        }
+      ]));
       this.getComputedFields = _.bind(getComputedFields, this);
       this.flattenHit = flattenHit(this);
       this.formatHit = formatHit(this, fieldFormats.getDefaultInstance('string'));

src/ui/public/courier/fetch/request/__tests__/segmented_index_selection.js

@@ -0,0 +1,129 @@
+import ngMock from 'ngMock';
+import expect from 'expect.js';
+import { times } from 'lodash';
+import sinon from 'auto-release-sinon';
+
+import HitSortFnProv from 'plugins/kibana/discover/_hit_sort_fn';
+import NoDigestPromises from 'testUtils/noDigestPromises';
+
+describe('Segmented Request Index Selection', function () {
+  let Promise;
+  let $rootScope;
+  let SegmentedReq;
+  let MockSource;
+  let HitSortFn;
+
+  NoDigestPromises.activateForSuite();
+
+  beforeEach(ngMock.module('kibana'));
+  beforeEach(ngMock.inject((Private, $injector) => {
+    Promise = $injector.get('Promise');
+    HitSortFn = Private(HitSortFnProv);
+    $rootScope = $injector.get('$rootScope');
+    SegmentedReq = Private(require('ui/courier/fetch/request/segmented'));
+
+    const StubbedSearchSourceProvider = require('fixtures/stubbed_search_source');
+    MockSource = class {
+      constructor() {
+        return $injector.invoke(StubbedSearchSourceProvider);
+      }
+    };
+  }));
+
+  it('queries with size until all 500 docs returned', async function () {
+    const search = new MockSource();
+    const indexPattern = search.get('index');
+    sinon.stub(indexPattern, 'toDetailedIndexList').returns(Promise.resolve([
+      { index: 'one', min: 0, max: 1 },
+      { index: 'two', min: 0, max: 1 },
+      { index: 'three', min: 0, max: 1 },
+      { index: 'four', min: 0, max: 1 },
+      { index: 'five', min: 0, max: 1 },
+    ]));
+
+    const req = new SegmentedReq(search);
+    req._handle.setDirection('desc');
+    req._handle.setSortFn(new HitSortFn('desc'));
+    req._handle.setSize(500);
+    await req.start();
+
+    // first 200
+    expect((await req.getFetchParams()).body.size).to.be(500);
+    await req.handleResponse({
+      hits: { total: 1000, hits: times(200, (i) => ({ i })) }
+    });
+
+    // total = 400
+    expect((await req.getFetchParams()).body.size).to.be(500);
+    await req.handleResponse({
+      hits: { total: 1000, hits: times(200, (i) => ({ i })) }
+    });
+
+    // total = 600
+    expect((await req.getFetchParams()).body.size).to.be(500);
+    await req.handleResponse({
+      hits: { total: 1000, hits: times(200, (i) => ({ i })) }
+    });
+
+    expect((await req.getFetchParams()).body.size).to.be(0);
+    await req.handleResponse({
+      hits: { total: 1000, hits: times(200, (i) => ({ i })) }
+    });
+
+    expect((await req.getFetchParams()).body.size).to.be(0);
+    await req.handleResponse({
+      hits: { total: 1000, hits: times(200, (i) => ({ i })) }
+    });
+  });
+
+  it(`sets size 0 for indices that couldn't procude hits`, async function () {
+    const search = new MockSource();
+    const indexPattern = search.get('index');
+
+    // the segreq is looking for 10 documents, and we will give it ten docs with time:5 in the first response.
+    // on the second index it should still request 10 documents because it could produce documents with time:5.
+    // the next two indexes will get size 0, since they couldn't produce documents with the time:5
+    // the final index will get size:10, because it too can produce docs with time:5
+    sinon.stub(indexPattern, 'toDetailedIndexList').returns(Promise.resolve([
+      { index: 'one', min: 0, max: 10 },
+      { index: 'two', min: 0, max: 10 },
+      { index: 'three', min: 12, max: 20 },
+      { index: 'four', min: 15, max: 20 },
+      { index: 'five', min: 5, max: 50 },
+    ]));
+
+    const req = new SegmentedReq(search);
+    req._handle.setDirection('desc');
+    req._handle.setSortFn(new HitSortFn('desc'));
+    req._handle.setSize(10);
+    await req.start();
+
+    // first 10
+    expect((await req.getFetchParams()).body.size).to.be(10);
+    await req.handleResponse({
+      hits: { total: 1000, hits: times(10, () => ({ _source: { time: 5 } })) }
+    });
+
+    // total = 400
+    expect((await req.getFetchParams()).body.size).to.be(10);
+    await req.handleResponse({
+      hits: { total: 1000, hits: times(10, () => ({ _source: { time: 5 } })) }
+    });
+
+    // total = 600
+    expect((await req.getFetchParams()).body.size).to.be(0);
+    await req.handleResponse({
+      hits: { total: 1000, hits: [] }
+    });
+
+    expect((await req.getFetchParams()).body.size).to.be(0);
+    await req.handleResponse({
+      hits: { total: 1000, hits: [] }
+    });
+
+    expect((await req.getFetchParams()).body.size).to.be(10);
+    await req.handleResponse({
+      hits: { total: 1000, hits: times(10, () => ({ _source: { time: 5 } })) }
+    });
+  });
+});

src/ui/public/courier/fetch/request/_segmented_handle.js

@@ -22,6 +22,7 @@ define(function (require) {
       this.setDirection = _.bindKey(req, 'setDirection');
       this.setSize = _.bindKey(req, 'setSize');
       this.setMaxSegments = _.bindKey(req, 'setMaxSegments');
+      this.setSortFn = _.bindKey(req, 'setSortFn');
     }
 
     return SegmentedHandle;