merger: use the skip_header option of the net.box connection's API

In scope of tarantool/tarantool#8147, a new context-dependent extension for
box tuples, `MP_TUPLE`, is introduced, and the IPROTO data response is
extended with a new `IPROTO_TUPLE_FORMATS` field with tuples formats
necessary for decoding `MP_TUPLE` sent in the `IPROTO_DATA` field.

If the tuple merger's buffer source is used, raw MsgPack is received (see
also d18ad41). We expect the response to only contain an `IPROTO_DATA`
field, so the occurrence of a new `IPROTO_TUPLE_FORMATS` field breaks this
assumption. The `IPROTO_DATA` field is still decoded correctly, but the
input buffer's position now points to the `IPROTO_TUPLE_FORMATS` field
instead of the end of the response.

Instead of handling the issue described above we decided to opt for a
simpler solution, namely, using the `skip_header` option of the `net.box`
connection's API, which returns only the `IPROTO_DATA` field's value, which
is what we are looking for.

This option is was introduced in tarantool/tarantool@1aaf637870 (2.2.0
release), but since the issue described above can only occur with the
latest Tarantool version, we can safely fallback to manually decoding the
`IPROTO_DATA` header on old Tarantool versions.

Needed for tarantool/tarantool#8147

Author: CuriousGeorgiy

crud/common/utils.lua

@@ -639,6 +639,9 @@ local function determine_enabled_features()
                                               or is_version_in_range(major, minor, patch, suffix,
                                                                      1, 10, 8, nil,
                                                                      1, 10, math.huge, nil)
+
+    enabled_tarantool_features.netbox_skip_header_option = is_version_ge(major, minor, patch, suffix,
+                                                                         2, 2, 0, nil)
 end
 
 function utils.tarantool_supports_fieldpaths()
@@ -681,6 +684,14 @@ function utils.tarantool_supports_external_merger()
     return enabled_tarantool_features.external_merger
 end
 
+function utils.tarantool_supports_netbox_skip_header_option()
+    if enabled_tarantool_features.netbox_skip_header_option == nil then
+        determine_enabled_features()
+    end
+
+    return enabled_tarantool_features.netbox_skip_header_option
+end
+
 local function add_nullable_fields_recursive(operations, operations_map, space_format, tuple, id)
     if id < 2 or tuple[id - 1] ~= box.NULL then
         return operations

crud/select.lua

@@ -112,6 +112,8 @@ local function select_on_storage(space_name, index_id, conditions, opts)
     -- getting tuples with user defined fields (if `fields` option is specified)
     -- and fields that are needed for comparison on router (primary key + scan key)
     local filtered_tuples = schema.filter_tuples_fields(resp.tuples, opts.field_names)
+    require('log').info("filtered_tuple=%s, #filtered_tuples=%d", require('json').encode(filtered_tuples), #filtered_tuples)
+    assert(#filtered_tuples >= 0)
 
     local result = {cursor, filtered_tuples}
 

crud/select/merger.lua

@@ -58,17 +58,7 @@ end
 
 local data = ffi.new('const unsigned char *[1]')
 
-local function decode_response_headers(buf)
-    -- {48: [cursor, [tuple_1, tuple_2, ...]]} (exactly 1 pair of key-value)
-    data[0] = buf.rpos
-
-    -- 48 (key)
-    data[0] = data[0] + 1
-
-    -- [cursor, [tuple_1, tuple_2, ...]] (value)
-    data[0] = data[0] + 1
-
-    -- Decode array header
+local function decode_response_array_header()
     local c = data[0][0]
     data[0] = data[0] + 1
     if c == 0xdc then
@@ -80,6 +70,23 @@ local function decode_response_headers(buf)
     return ffi.cast(char_ptr, data[0])
 end
 
+local function decode_response_headers(buf)
+    data[0] = buf.rpos
+
+    if not utils.tarantool_supports_netbox_skip_header_option() then
+        -- {48: [cursor, [tuple_1, tuple_2, ...]]} (exactly 1 pair of key-value)
+
+        -- 48 (key)
+        data[0] = data[0] + 1
+
+        --{48: [cursor, [tuple_1, tuple_2, ...]], 96: {id: tuple_format}}
+        -- [cursor, [tuple_1, tuple_2, ...]] (value)
+        data[0] = data[0] + 1
+    end
+
+    return decode_response_array_header()
+end
+
 local function decode_metainfo(buf)
     -- Skip an array around a call return values.
     buf.rpos = decode_response_headers(buf)
@@ -192,7 +199,8 @@ local function new(vshard_router, replicasets, space, index_id, func_name, func_
     for _, replicaset in pairs(replicasets) do
         -- Perform a request.
         local buf = buffer.ibuf()
-        local net_box_opts = {is_async = true, buffer = buf, skip_header = false}
+        local net_box_opts = {is_async = true, buffer = buf,
+                              skip_header = utils.tarantool_supports_netbox_skip_header_option()}
         local future = replicaset[vshard_call_name](replicaset, func_name, func_args,
                 net_box_opts)
 
@@ -249,7 +257,8 @@ local function new_readview(vshard_router, replicasets, readview_uuid, space, in
                 if replica_uuid == value.uuid then
                     -- Perform a request.
                     local buf = buffer.ibuf()
-                    local net_box_opts = {is_async = true, buffer = buf, skip_header = false}
+                    local net_box_opts = {is_async = true, buffer = buf,
+                                          skip_header = utils.tarantool_supports_netbox_skip_header_option()}
                     func_args[4].readview_id = value.id
                     local future = replica.conn:call(func_name, func_args, net_box_opts)