Add stats collection and pert tests for batch methods

Since we have PR #244 it will be nice to collect
statistics for batch operations too.
To establish the effectiveness of `crud.batch_insert()`
method compared to `crud.insert()`, perf tests were added.
`crud.insert()` in the loop and `crud.batch_insert()`
are compared for different batch sizes.

Closes #193

crud.lua

@@ -35,11 +35,11 @@ crud.insert_object = stats.wrap(insert.object, stats.op.INSERT)
 
 -- @refer batch_insert.tuples_batch
 -- @function batch_insert
-crud.batch_insert = batch_insert.tuples_batch
+crud.batch_insert = stats.wrap(batch_insert.tuples_batch, stats.op.BATCH_INSERT)
 
 -- @refer batch_insert.objects_batch
 -- @function batch_insert_object
-crud.batch_insert_object = batch_insert.objects_batch
+crud.batch_insert_object = stats.wrap(batch_insert.objects_batch, stats.op.BATCH_INSERT)
 
 -- @refer get.call
 -- @function get
@@ -63,11 +63,11 @@ crud.upsert = stats.wrap(upsert.tuple, stats.op.UPSERT)
 
 -- @refer batch_upsert.tuples_batch
 -- @function batch_upsert
-crud.batch_upsert = batch_upsert.tuples_batch
+crud.batch_upsert = stats.wrap(batch_upsert.tuples_batch, stats.op.BATCH_UPSERT)
 
 -- @refer batch_upsert.objects_batch
 -- @function batch_upsert_object
-crud.batch_upsert_object = batch_upsert.objects_batch
+crud.batch_upsert_object = stats.wrap(batch_upsert.objects_batch, stats.op.BATCH_UPSERT)
 
 -- @refer upsert.object
 -- @function upsert

crud/stats/operation.lua

@@ -6,12 +6,16 @@
 return {
     -- INSERT identifies both `insert` and `insert_object`.
     INSERT = 'insert',
+    -- BATCH_INSERT identifies both `batch_insert` and `batch_insert_object`.
+    BATCH_INSERT = 'batch_insert',
     GET = 'get',
     -- REPLACE identifies both `replace` and `replace_object`.
     REPLACE = 'replace',
     UPDATE = 'update',
     -- UPSERT identifies both `upsert` and `upsert_object`.
     UPSERT = 'upsert',
+    -- BATCH_UPSERT identifies both `batch_upsert` and `batch_upsert_object`.
+    BATCH_UPSERT = 'batch_upsert',
     DELETE = 'delete',
     -- SELECT identifies both `pairs` and `select`.
     SELECT = 'select',

test/integration/stats_test.lua

@@ -171,6 +171,30 @@ local simple_operation_cases = {
         },
         op = 'insert',
     },
+    {
+        name = 'batch_insert',
+        func = 'crud.batch_insert',
+        args = {
+            space_name,
+            {
+                { 21, box.NULL, 'Ivan', 'Ivanov', 20, 'Moscow' },
+                { 31, box.NULL, 'Oleg', 'Petrov', 25, 'Moscow' },
+            }
+        },
+        op = 'batch_insert',
+    },
+    {
+        name = 'batch_insert_object',
+        func = 'crud.batch_insert_object',
+        args = {
+            space_name,
+            {
+                { id = 22, name = 'Ivan', last_name = 'Ivanov', age = 20, city = 'Moscow' },
+                { id = 32, name = 'Oleg', last_name = 'Petrov', age = 25, city = 'Moscow' },
+            }
+        },
+        op = 'batch_insert',
+    },
     {
         name = 'get',
         func = 'crud.get',
@@ -239,6 +263,32 @@ local simple_operation_cases = {
         },
         op = 'upsert',
     },
+    {
+        name = 'batch_upsert',
+        func = 'crud.batch_upsert',
+        args = {
+            space_name,
+            {
+                { 26, box.NULL, 'Ivan', 'Ivanov', 20, 'Moscow' },
+                { 36, box.NULL, 'Oleg', 'Petrov', 25, 'Moscow' },
+            },
+            {{'+', 'age', 1}},
+        },
+        op = 'batch_upsert',
+    },
+    {
+        name = 'batch_upsert_object',
+        func = 'crud.batch_upsert_object',
+        args = {
+            space_name,
+            {
+                { id = 27, name = 'Ivan', last_name = 'Ivanov', age = 20, city = 'Moscow' },
+                { id = 37, name = 'Oleg', last_name = 'Petrov', age = 25, city = 'Moscow' },
+            },
+            {{'+', 'age', 1}}
+        },
+        op = 'batch_upsert',
+    },
     {
         name = 'delete',
         func = 'crud.delete',
@@ -289,6 +339,20 @@ local simple_operation_cases = {
         op = 'insert',
         expect_error = true,
     },
+    {
+        name = 'batch_insert_error',
+        func = 'crud.batch_insert',
+        args = { space_name, {{ 'id' }} },
+        op = 'batch_insert',
+        expect_error = true,
+    },
+    {
+        name = 'batch_insert_object_error',
+        func = 'crud.batch_insert_object',
+        args = { space_name, {{ 'id' }} },
+        op = 'batch_insert',
+        expect_error = true,
+    },
     {
         name = 'get_error',
         func = 'crud.get',
@@ -346,6 +410,20 @@ local simple_operation_cases = {
         op = 'upsert',
         expect_error = true,
     },
+    {
+        name = 'batch_upsert_error',
+        func = 'crud.batch_upsert',
+        args = { space_name, {{ 'id' }}, {{'+', 'age', 1}} },
+        op = 'batch_upsert',
+        expect_error = true,
+    },
+    {
+        name = 'batch_upsert_object_error',
+        func = 'crud.batch_upsert_object',
+        args = { space_name, {{ 'id' }}, {{'+', 'age', 1}} },
+        op = 'batch_upsert',
+        expect_error = true,
+    },
     {
         name = 'delete_error',
         func = 'crud.delete',

test/performance/perf_test.lua

@@ -4,6 +4,7 @@ local fiber = require('fiber')
 local errors = require('errors')
 local net_box = require('net.box')
 local log = require('log')
+local fun = require('fun')
 
 local t = require('luatest')
 local g = t.group('perf')
@@ -98,8 +99,32 @@ local function normalize(s, n)
     return (' '):rep(n - len) .. s
 end
 
+local function generate_batch_insert_cases(min_value, denominator, count)
+    local matrix = {}
+    local cols_names = {}
+
+    local batch_size = min_value
+    for _ = 1, count do
+        local col_name = tostring(batch_size)
+        table.insert(cols_names, col_name)
+        matrix[col_name] = { column_name = col_name, arg = batch_size }
+        batch_size = batch_size * denominator
+    end
+
+    return {
+        matrix = matrix,
+        cols_names = cols_names,
+    }
+end
+
+local min_batch_size = 1
+local denominator_batch_size = 10
+local count_batch_cases = 5
+local batch_insert_cases = generate_batch_insert_cases(min_batch_size, denominator_batch_size, count_batch_cases)
+
 local row_name = {
     insert = 'insert',
+    batch_insert = 'batch_insert',
     select_pk = 'select by pk',
     select_gt_pk = 'select gt by pk (limit 10)',
     select_secondary_eq = 'select eq by secondary (limit 10)',
@@ -117,6 +142,12 @@ local column_name = {
     metrics_quantile_stats = 'crud (metrics stats, with quantiles)',
 }
 
+-- insert column names for batch_insert/insert comparison cases
+fun.reduce(
+        function(list, value) list[value] = value return list end,
+        column_name, pairs(batch_insert_cases.cols_names)
+)
+
 local function visualize_section(total_report, name, comment, section, params)
     local report_str = ('== %s ==\n(%s)\n\n'):format(name, comment or '')
 
@@ -242,6 +273,7 @@ g.after_all(function(g)
             row_name.select_secondary_eq,
             row_name.select_secondary_sharded,
             row_name.insert,
+            row_name.batch_insert,
         }
     })
 
@@ -260,6 +292,15 @@ g.after_all(function(g)
             row_name.insert,
         }
     })
+
+    visualize_report(g.total_report, 'BATCH COMPARISON PERFORMANCE REPORT', {
+        columns = batch_insert_cases.cols_names,
+
+        rows = {
+            row_name.insert,
+            row_name.batch_insert,
+        },
+    })
 end)
 
 local function generate_customer()
@@ -488,6 +529,18 @@ local insert_params = function()
     return { 'customers', generate_customer() }
 end
 
+local batch_insert_params = function(count)
+    local batch = {}
+
+    count = count or 1
+
+    for _ = 1, count do
+        table.insert(batch, generate_customer())
+    end
+
+    return { 'customers', batch }
+end
+
 local select_params_pk_eq = function()
     return { 'customers', {{'==', 'id', gen() % 10000}} }
 end
@@ -606,6 +659,12 @@ local pairs_perf = {
     connection_count = 10,
 }
 
+local batch_insert_comparison_perf = {
+    timeout = 30,
+    fiber_count = 1,
+    connection_count = 1,
+}
+
 local cases = {
     vshard_insert = {
         prepare = vshard_prepare,
@@ -640,6 +699,29 @@ local cases = {
         row_name = row_name.insert,
     },
 
+    crud_batch_insert = {
+        call = 'crud.batch_insert',
+        params = batch_insert_params,
+        matrix = stats_cases,
+        integration_params = integration_params,
+        perf_params = insert_perf,
+        row_name = row_name.batch_insert,
+    },
+
+    crud_batch_insert_without_stats_wrapper = {
+        prepare = function(g)
+            g.router:eval([[
+                rawset(_G, '_plain_batch_insert', require('crud.batch_insert').tuples_batch)
+            ]])
+        end,
+        call = '_plain_batch_insert',
+        params = batch_insert_params,
+        matrix = { [''] = { column_name = column_name.without_stats_wrapper } },
+        integration_params = integration_params,
+        perf_params = batch_insert_comparison_perf,
+        row_name = row_name.batch_insert,
+    },
+
     vshard_select_pk_eq = {
         prepare = function(g)
             select_prepare(g)
@@ -876,14 +958,55 @@ local cases = {
         perf_params = pairs_perf,
         row_name = row_name.pairs_gt,
     },
+
+    crud_insert_loop = {
+        prepare = function(g)
+            g.router:eval([[
+                _insert_loop = function(space_name, tuples)
+                    local results
+                    local errors
+
+                    for _, tuple in ipairs(tuples) do
+                         local res, err = crud.insert(space_name, tuple)
+                         if res ~= nil then
+                             results = results or {}
+                             table.insert(results, res)
+                         end
+
+                         if err ~= nil then
+                             errors = errors or {}
+                             table.insert(errors, err)
+                         end
+                    end
+
+                    return results, errors
+                end
+            ]])
+        end,
+        call = '_insert_loop',
+        params = batch_insert_params,
+        matrix = batch_insert_cases.matrix,
+        integration_params = integration_params,
+        perf_params = batch_insert_comparison_perf,
+        row_name = row_name.insert,
+    },
+
+    crud_batch_insert_different_batch_size = {
+        call = 'crud.batch_insert',
+        params = batch_insert_params,
+        matrix = batch_insert_cases.matrix,
+        integration_params = integration_params,
+        perf_params = batch_insert_comparison_perf,
+        row_name = row_name.batch_insert,
+    },
 }
 
-local function generator_f(conn, call, params, report, timeout)
+local function generator_f(conn, call, params, report, timeout, arg)
     local start = clock.monotonic()
 
     while (clock.monotonic() - start) < timeout do
         local call_start = clock.monotonic()
-        local ok, res, err = pcall(conn.call, conn, call, params())
+        local ok, res, err = pcall(conn.call, conn, call, params(arg))
         local call_time = clock.monotonic() - call_start
 
         if not ok then
@@ -947,7 +1070,7 @@ for name, case in pairs(cases) do
             for id = 1, params.fiber_count do
                 local conn_id = id % params.connection_count + 1
                 local conn = connections[conn_id]
-                local f = fiber.new(generator_f, conn, case.call, case.params, report, params.timeout)
+                local f = fiber.new(generator_f, conn, case.call, case.params, report, params.timeout, subcase.arg)
                 f:set_joinable(true)
                 table.insert(fibers, f)
             end