bfs_executor: optimize two connections case

Totktonada · Totktonada · commit abfec4fc5e20 · 2018-09-03T03:15:12.000+03:00
diff --git a/graphql/accessor_shard_cache.lua b/graphql/accessor_shard_cache.lua
@@ -1,6 +1,7 @@
 local json = require('json')
 local shard = require('shard')
 local utils = require('graphql.utils')
+local request_batch = require('graphql.request_batch')
 local accessor_shard_index_info = require('graphql.accessor_shard_index_info')
 
 local check = utils.check
@@ -10,27 +11,15 @@ local accessor_shard_cache = {}
 local function get_select_ids(self, batch, opts)
     local opts = opts or {}
     local skip_cached = opts.skip_cached or false
-    local ids = {}
-    for _, key in ipairs(batch.keys) do
-        local collection_name = batch.collection_name
-        local index_name = batch.index_name or ''
-        local key_str
-        if type(key) ~= 'table' then
-            key_str = tostring(key)
-        else
-            assert(utils.is_array(key), 'compound key must be an array')
-            key_str = table.concat(key, ',')
-        end
-        local opts = batch.iterator_opts or {}
-        local opts_str = ('%s,%s'):format(opts.iterator or opts[1] or 'EQ',
-            opts.limit or '')
-        local id = ('%s.%s.%s.%s'):format(collection_name, index_name, key_str,
-            opts_str)
-        if not skip_cached or self.cache[id] == nil then
-            table.insert(ids, id)
+
+    local skip_function
+    if skip_cached then
+        skip_function = function(id)
+            return self.cache[id] ~= nil
         end
     end
-    return ids
+
+    return batch:select_ids(skip_function)
 end
 
 local function net_box_call_wrapper(conn, func_name, call_args)
@@ -268,12 +257,8 @@ end
 --- @return luafun iterator (one value) to fetched data or nil
 local function cache_lookup(self, collection_name, index_name, key,
         iterator_opts)
-    local batch = {
-        collection_name = collection_name,
-        index_name = index_name,
-        keys = {key or box.NULL},
-        iterator_opts = iterator_opts,
-    }
+    local batch = request_batch.new(collection_name, index_name,
+        {key or box.NULL}, iterator_opts)
     local id = get_select_ids(self, batch)[1]
     if self.cache[id] == nil then
         return nil
diff --git a/graphql/bfs_executor.lua b/graphql/bfs_executor.lua
@@ -176,13 +176,13 @@
 ---         }
 ---     }
 
-local json = require('json')
 local utils = require('graphql.utils')
 local core_util = require('graphql.core.util')
 local core_types = require('graphql.core.types')
 local core_query_util = require('graphql.core.query_util')
 local core_validate_variables = require('graphql.core.validate_variables')
 local core_introspection = require('graphql.core.introspection')
+local request_batch = require('graphql.request_batch')
 
 -- XXX: Possible cache_only requests refactoring. Maybe just set
 -- `is_cache_only` flag in prepared_object, find such request in within the
@@ -201,13 +201,9 @@ local core_introspection = require('graphql.core.introspection')
 -- Maybe we also can remove separate `is_list = true` case processing over the
 -- code that can havily simplify things.
 
--- XXX: Currently the executor cannot batch similar requests in case when a
--- collection has two connections and so the requests to the different
--- connected (destination) collections placed in the list each after each. The
--- simplest way to mitigate this issue is to consume similar requests (in
--- `fetch_first_same`) w/o respect to the order (traverse until end of the
--- `open_set` or a count limit). The alternative is to introduce resorting
--- prepared requests step.
+-- XXX: It would be more natural to have list of tables with field_info content
+-- + field_name instead of fields_info to handle each prepared resolve
+-- separatelly and add ability to reorder it.
 
 local bfs_executor = {}
 
@@ -784,34 +780,19 @@ fetch_first_same = function(open_set, opts)
                 size = i
                 break
             end
-            local prepared_select = prepared_resolve.prepared_select
-            local request_opts = prepared_select.request_opts
-            local collection_name = prepared_select.collection_name
-            local index_name = request_opts.index_name
-            local key = request_opts.index_value or box.NULL
-            local iterator_opts = request_opts.iterator_opts
+            local batch = request_batch.from_prepared_resolve(prepared_resolve)
 
             if i == 1 then
                 assert(batches[field_name] == nil,
                     ('internal error: %s: field names "%s" clash'):format(
                     func_name, field_name))
-                batches[field_name] = {
-                    collection_name = collection_name,
-                    index_name = index_name,
-                    keys = {key},
-                    iterator_opts = iterator_opts,
-                }
+                batches[field_name] = batch
                 size = i
             else
-                local ok =
-                    batches[field_name] ~= nil and
-                    batches[field_name].collection_name == collection_name and
-                    batches[field_name].index_name == index_name and
-                    utils.are_tables_same(batches[field_name].iterator_opts or
-                        {}, iterator_opts or {})
-                if not ok then break end -- XXX: continue here and return first
-                                         -- non-match instead of size?
-                table.insert(batches[field_name].keys, key)
+                local ok = batches[field_name] ~= nil and
+                    batches[field_name]:compare_bins(batch)
+                if not ok then break end
+                table.insert(batches[field_name].keys, batch.keys[1])
                 size = i
             end
         end
@@ -850,45 +831,20 @@ fetch_resolve_list = function(prepared_object_list, opts)
                 size = i
                 break
             end
-            local prepared_select = prepared_resolve.prepared_select
-            local request_opts = prepared_select.request_opts
-            local collection_name = prepared_select.collection_name
-            local index_name = request_opts.index_name
-            local key = request_opts.index_value or box.NULL
-            local iterator_opts = request_opts.iterator_opts
+            local batch = request_batch.from_prepared_resolve(prepared_resolve)
 
             if i == 1 then
                 assert(batches[field_name] == nil,
                     ('internal error: %s: field names "%s" clash'):format(
                     func_name, field_name))
-                batches[field_name] = {
-                    collection_name = collection_name,
-                    index_name = index_name,
-                    keys = {key},
-                    iterator_opts = iterator_opts,
-                }
+                batches[field_name] = batch
                 size = i
             else
-                assert(batches[field_name].collection_name == collection_name,
-                    ('internal error: %s: prepared object list has ' ..
-                    'different collection names: "%s" and "%s"'):format(
-                    func_name, batches[field_name].collection_name,
-                    collection_name))
-                assert(batches[field_name].index_name == index_name,
-                    ('internal error: %s: prepared object list has ' ..
-                    'different index names: "%s" and "%s"'):format(func_name,
-                    tostring(batches[field_name].index_name),
-                    tostring(index_name)))
-                local ok = utils.are_tables_same(batches[field_name].iterator_opts or
-                    {}, iterator_opts or {})
-                if not ok then -- avoid extra json.encode()
-                    assert(ok, ('internal error: %s: prepared object list ' ..
-                        'has different iterator options: "%s" and "%s"'):format(
-                        func_name,
-                        json.encode(batches[field_name].iterator_opts),
-                        json.encode(iterator_opts)))
+                local ok, err = batches[field_name]:compare_bins_extra(batch)
+                if not ok then
+                    error(('internal error: %s: %s'):format(func_name, err))
                 end
-                table.insert(batches[field_name].keys, key)
+                table.insert(batches[field_name].keys, batch.keys[1])
                 size = i
             end
         end
@@ -905,6 +861,69 @@ end
 
 -- }}}
 
+-- Reorder requests before add to open_set {{{
+
+local function expand_open_set(open_set, child_open_set, opts)
+    local opts = opts or {}
+    local accessor = opts.accessor
+
+    if not accessor:cache_is_supported() then
+        utils.expand_list(open_set, child_open_set)
+        return
+    end
+
+    local item_bin_to_ordinal = {}
+    local items_per_ordinal = {}
+    local next_ordinal = 1
+
+    -- Create histogram-like 'items_per_ordinal' structure with lists of items.
+    -- Each list contain items of the same kind (with the same bin value).
+    -- Ordinals of the bins are assigned in order of appear in child_open_set.
+    for _, item in ipairs(child_open_set) do
+        if item.prepared_object_list ~= nil then
+            local ordinal = next_ordinal
+            assert(items_per_ordinal[ordinal] == nil)
+            items_per_ordinal[ordinal] = {}
+            next_ordinal = next_ordinal + 1
+            table.insert(items_per_ordinal[ordinal], item)
+        else
+            local prepared_object = item.prepared_object
+            assert(prepared_object ~= nil)
+            assert(prepared_object.fields_info ~= nil)
+
+            local batch_bins = {}
+            for field_name, field_info in pairs(prepared_object.fields_info) do
+                local prepared_resolve = field_info.prepared_resolve
+                if prepared_resolve.is_calculated then
+                    table.insert(batch_bins, field_name .. ':<calculated>')
+                else
+                    local batch = request_batch.from_prepared_resolve(
+                        prepared_resolve)
+                    table.insert(batch_bins, field_name .. ':' .. batch:bin())
+                end
+            end
+
+            local item_bin = table.concat(batch_bins, ';')
+            local ordinal = item_bin_to_ordinal[item_bin]
+            if ordinal == nil then
+                item_bin_to_ordinal[item_bin] = next_ordinal
+                ordinal = next_ordinal
+                assert(items_per_ordinal[ordinal] == nil)
+                items_per_ordinal[ordinal] = {}
+                next_ordinal = next_ordinal + 1
+            end
+            table.insert(items_per_ordinal[ordinal], item)
+        end
+    end
+
+    -- add items from child_open_set in ordinals order to open_set
+    for _, items in ipairs(items_per_ordinal) do
+        utils.expand_list(open_set, items)
+    end
+end
+
+-- }}}
+
 -- Debugging {{{
 
 local function prepared_object_digest(prepared_object)
@@ -1056,19 +1075,21 @@ function bfs_executor.execute(schema, query_ast, variables, operation_name, opts
             local child = invoke_resolve(item.prepared_object, context,
                 {qcontext = qcontext,
                 is_item_cache_only = is_item_cache_only})
-            local child_open_set = child.open_set
             local child_cache_only_open_set = child.cache_only_open_set
-            utils.expand_list(cache_only_open_set, child_cache_only_open_set)
-            utils.expand_list(open_set, child_open_set)
+            local child_open_set = child.open_set
+            expand_open_set(cache_only_open_set, child_cache_only_open_set,
+                {accessor = accessor})
+            expand_open_set(open_set, child_open_set, {accessor = accessor})
         elseif item.prepared_object_list ~= nil then
             local child = invoke_resolve_list(item.prepared_object_list,
                 context, {qcontext = qcontext, accessor = accessor,
                 is_item_cache_only = is_item_cache_only,
                 max_batch_size = max_batch_size})
-            local child_open_set = child.open_set
             local child_cache_only_open_set = child.cache_only_open_set
-            utils.expand_list(cache_only_open_set, child_cache_only_open_set)
-            utils.expand_list(open_set, child_open_set)
+            local child_open_set = child.open_set
+            expand_open_set(cache_only_open_set, child_cache_only_open_set,
+                {accessor = accessor})
+            expand_open_set(open_set, child_open_set, {accessor = accessor})
         elseif item.squash_marker ~= nil then
             local open_set_to_fetch = is_item_cache_only and
                 cache_only_open_set or open_set
diff --git a/graphql/request_batch.lua b/graphql/request_batch.lua
@@ -0,0 +1,113 @@
+local json = require('json')
+local utils = require('graphql.utils')
+
+local request_batch = {}
+
+local function iterator_opts_tostring(iterator_opts)
+    return ('%s,%s'):format(
+        iterator_opts.iterator or iterator_opts[1] or 'EQ',
+        iterator_opts.limit or '')
+end
+
+--- List of strings, each uniquely identifies select request in the batch.
+local function batch_select_ids(self, skip_function)
+    local ids = {}
+
+    local collection_name = self.collection_name
+    local index_name = self.index_name or ''
+    local iterator_opts_str = iterator_opts_tostring(self.iterator_opts)
+
+    for _, key in ipairs(self.keys) do
+        local key_str
+        if type(key) ~= 'table' then
+            key_str = tostring(key)
+        else
+            assert(utils.is_array(key), 'compound key must be an array')
+            key_str = table.concat(key, ',')
+        end
+        local id = ('%s.%s.%s.%s'):format(collection_name, index_name, key_str,
+            iterator_opts_str)
+        if skip_function == nil or not skip_function(id) then
+            table.insert(ids, id)
+        end
+    end
+
+    return ids
+end
+
+--- String uniquely identifies the batch information except keys.
+local function batch_bin(self)
+    return ('%s.%s.%s'):format(
+        self.collection_name,
+        self.index_name or '',
+        iterator_opts_tostring(self.iterator_opts))
+end
+
+--- Compare batches by bin.
+local function batch_compare_bins(self, other)
+    return self.collection_name == other.collection_name and
+        self.index_name == other.index_name and
+        utils.are_tables_same(self.iterator_opts, other.iterator_opts)
+end
+
+--- Compare batches by bin with detailed error in case they don't match.
+local function batch_compare_bins_extra(self, other)
+    if self.collection_name ~= other.collection_name then
+        local err = ('prepared object list has different collection names: ' ..
+            '"%s" and "%s"'):format(self.collection_name, other.collection_name)
+        return false, err
+    end
+
+    if self.index_name ~= other.index_name then
+        local err = ('prepared object list has different index names: ' ..
+            '"%s" and "%s"'):format(tostring(self.index_name),
+            tostring(other.index_name))
+        return false, err
+    end
+
+    if not utils.are_tables_same(self.iterator_opts, other.iterator_opts) then
+        local err = ('prepared object list has different iterator options: ' ..
+            '"%s" and "%s"'):format(json.encode(self.iterator_opts),
+            json.encode(other.iterator_opts))
+        return false, err
+    end
+
+    return true
+end
+
+local request_batch_mt = {
+    __index = {
+        bin = batch_bin,
+        select_ids = batch_select_ids,
+        compare_bins = batch_compare_bins,
+        compare_bins_extra = batch_compare_bins_extra,
+    }
+}
+
+function request_batch.from_prepared_resolve(prepared_resolve)
+    assert(not prepared_resolve.is_calculated)
+    local prepared_select = prepared_resolve.prepared_select
+    local request_opts = prepared_select.request_opts
+    local collection_name = prepared_select.collection_name
+    local index_name = request_opts.index_name
+    local key = request_opts.index_value or box.NULL
+    local iterator_opts = request_opts.iterator_opts
+
+    return setmetatable({
+        collection_name = collection_name,
+        index_name = index_name,
+        keys = {key},
+        iterator_opts = iterator_opts or {},
+    }, request_batch_mt)
+end
+
+function request_batch.new(collection_name, index_name, keys, iterator_opts)
+    return setmetatable({
+        collection_name = collection_name,
+        index_name = index_name,
+        keys = keys,
+        iterator_opts = iterator_opts or {},
+    }, request_batch_mt)
+end
+
+return request_batch
diff --git a/test/bench/suite.ini b/test/bench/suite.ini
