Re: [Tarantool-patches] [PATCH 2/2] vinyl: rework upsert operation

[Vladislav Shpilevoy <v.shpilevoy@tarantool.org>]

Thanks for the patch!

ASAN tests on the branch are failing:
https://gitlab.com/tarantool/tarantool/-/jobs/661977877

See 19 comments below.

>  src/box/vinyl.c                 |   2 +-
>  src/box/vy_stmt.c               |  28 ++--
>  src/box/vy_stmt.h               |   5 +-
>  src/box/vy_upsert.c             | 305 +++++++++++++++++++++++++++-------------
>  test/unit/vy_iterators_helper.c |   2 +-
>  test/vinyl/upsert.result        | 289 +++++++++++++++++++++++++++++++++++++
>  test/vinyl/upsert.test.lua      | 121 ++++++++++++++++
>  7 files changed, 644 insertions(+), 108 deletions(-)
> > diff --git a/src/box/vy_upsert.c b/src/box/vy_upsert.c
> index 797492c2b..caf2482c7 100644
> --- a/src/box/vy_upsert.c
> +++ b/src/box/vy_upsert.c
> @@ -68,12 +68,173 @@ vy_upsert_try_to_squash(struct tuple_format *format,
>  	operations[0].iov_len = squashed_size;
>  
>  	*result_stmt = vy_stmt_new_upsert(format, key_mp, key_mp_end,
> -					  operations, 1);
> +					  operations, 1, false);
>  	if (*result_stmt == NULL)
>  		return -1;
>  	return 0;
>  }
>  
> +/**
> + * Check that key hasn't been changed after applying upsert operation.
> + */
> +static bool
> +vy_apply_result_does_cross_pk(struct tuple *old_stmt, const char *result,
> +			      const char *result_end, struct key_def *cmp_def,
> +			      uint64_t col_mask)
> +{
> +	if (!key_update_can_be_skipped(cmp_def->column_mask, col_mask)) {
> +		struct tuple *tuple =
> +			vy_stmt_new_replace(tuple_format(old_stmt), result,
> +					    result_end);
> +		int cmp_res = vy_stmt_compare(old_stmt, HINT_NONE, tuple,
> +					       HINT_NONE, cmp_def);

1. Bad indentation.

> +		tuple_unref(tuple);
> +		return cmp_res != 0;
> +	}
> +	return false;
> +}
> +
> +/**
> + * Apply update operations stored in @new_stmt (which is assumed to

2. Please, look at the doxygen syntax on the official page or here:
https://github.com/tarantool/tarantool/wiki/Code-review-procedure
This is a single very simple rule I keep repeating I don't already
remember how many times - use @a <param_name>, not @<param_name>.
I don't understand why does everyone keep violating it.

2. Parameter 'new_stmt' does not exist. As well as 'old_stmt'. What
did you mean?

> + * be upsert statement) on tuple @old_stmt. If @old_stmt is void
> + * statement (i.e. it is NULL or delete statement) then operations
> + * are applied on tuple @new_stmt. All operations which can't be
> + * applied are skipped; errors may be logged depending on @supress_error

3. supress_error -> supress_error.

4. What do you mean as 'all operations'? Operation groups from various
upserts? Or individual operations?

> + * flag.
> + *
> + * @upsert Upsert statement to be applied on @stmt.

5. If you want to use doxygen, use @param <param_name>.

> + * @stmt Statement to be used as base for upsert operations.
> + * @cmp_def Key definition required to provide check of primary key
> + *          modification.
> + * @retrun Tuple containing result of upsert application;
> + *         NULL in case OOM.

6. retrun -> return.

7. I guess you are among the ones who voted for 80 symbol comments - I
suggest you to use it. Since this is our new code style now.

> + */
> +static struct tuple *
> +vy_apply_upsert_on_terminal_stmt(struct tuple *upsert, struct tuple *stmt,
> +				 struct key_def *cmp_def, bool suppress_error)
> +{
> +	assert(vy_stmt_type(upsert) == IPROTO_UPSERT);
> +	assert(stmt == NULL || vy_stmt_type(stmt) != IPROTO_UPSERT);
> +
> +	uint32_t mp_size;
> +	const char *new_ops = vy_stmt_upsert_ops(upsert, &mp_size);
> +	/* Msgpack containing result of upserts application. */
> +	const char *result_mp;
> +	if (vy_stmt_is_void(stmt))

8. This {is_void} helper is used 2 times inside one funtion on the same
value. Seems like you could simply inline it, remeber result into a variable
{bool is_void;} and use it instead.

> +		result_mp = vy_upsert_data_range(upsert, &mp_size);
> +	else
> +		result_mp = tuple_data_range(stmt, &mp_size);
> +	const char *result_mp_end = result_mp + mp_size;
> +	/*
> +	 * xrow_upsert_execute() allocates result using region,
> +	 * so save starting point to release it later.
> +	 */
> +	struct region *region = &fiber()->gc;
> +	size_t region_svp = region_used(region);
> +	uint64_t column_mask = COLUMN_MASK_FULL;
> +	struct tuple_format *format = tuple_format(upsert);
> +
> +	uint32_t ups_cnt = mp_decode_array(&new_ops);
> +	const char *ups_ops = new_ops;
> +	/*
> +	 * In case upsert folds into insert, we must skip first
> +	 * update operations.
> +	 */
> +	if (vy_stmt_is_void(stmt)) {
> +		ups_cnt--;
> +		mp_next(&ups_ops);
> +	}
> +	for (uint32_t i = 0; i < ups_cnt; ++i) {
> +		assert(mp_typeof(*ups_ops) == MP_ARRAY);
> +		const char *ups_ops_end = ups_ops;
> +		mp_next(&ups_ops_end);
> +		const char *exec_res = result_mp;
> +		exec_res = xrow_upsert_execute(ups_ops, ups_ops_end, result_mp,
> +					       result_mp_end, format, &mp_size,
> +					       0, suppress_error, &column_mask);
> +		if (exec_res == NULL) {
> +			if (! suppress_error) {

9. According to one another recent code style change, unary operators
should not have a whitespace after them.

> +				assert(diag_last_error(diag_get()) != NULL);

10. Use {diag_is_empty}. Or better - save {diag_last_error(diag_get())} into
{struct error *e} before the assertion, and use {assert(e != NULL);}.

> +				struct error *e = diag_last_error(diag_get());
> +				/* Bail out immediately in case of OOM. */
> +				if (e->type != &type_ClientError) {
> +					region_truncate(region, region_svp);
> +					return NULL;
> +				}
> +				diag_log();
> +			}
> +			ups_ops = ups_ops_end;
> +			continue;
> +		}
> +		/*
> +		 * If it turns out that resulting tuple modifies primary
> +		 * key, than simply ignore this upsert.

11. than -> then.

> +		 */
> +		if (vy_apply_result_does_cross_pk(stmt, exec_res,
> +						  exec_res + mp_size, cmp_def,
> +						  column_mask)) {
> +			if (! suppress_error) {
> +				say_error("upsert operations %s are not applied"\
> +					  " due to primary key modification",
> +					  mp_str(ups_ops));
> +			}
> +			ups_ops = ups_ops_end;
> +			continue;
> +		}
> +		ups_ops = ups_ops_end;
> +		/*
> +		 * In case statement exists its format must
> +		 * satisfy space's format. Otherwise, upsert's
> +		 * tuple is checked to fit format once it is
> +		 * processed in vy_upsert().
> +		 */
> +		if (stmt != NULL) {
> +			if (tuple_validate_raw(tuple_format(stmt),
> +					       exec_res) != 0) {
> +				if (! suppress_error)
> +					diag_log();
> +				continue;
> +			}
> +		}
> +		result_mp = exec_res;
> +		result_mp_end = exec_res + mp_size;
> +	}
> +	struct tuple *new_terminal_stmt = vy_stmt_new_replace(format, result_mp,
> +							      result_mp_end);
> +	region_truncate(region, region_svp);
> +	if (new_terminal_stmt == NULL)
> +		return NULL;
> +	vy_stmt_set_lsn(new_terminal_stmt, vy_stmt_lsn(upsert));
> +	return new_terminal_stmt;
> +}
> +
> +static bool
> +tuple_format_is_suitable_for_squash(struct tuple_format *format)
> +{
> +	struct tuple_field *field;
> +	json_tree_foreach_entry_preorder(field, &format->fields.root,
> +					 struct tuple_field, token) {
> +		if (field->type == FIELD_TYPE_UNSIGNED)
> +				return false;

12. Bad indentation.
Also this squash rule is not going to work because integer type also can
overflow, both below INT64_MIN and above UINT64_MAX. Decimal types
can overflow. Besides, decimal can fail when a non-decimal value does not
fit the decimal type during conversion. For example, a huge double value.
DBL_MAX is bigger than maximal value available in our decimal type. See
xrow_update_arith_make() for all errors.

Since squash is mostly about squashing +/-, looks like it won't work
almost always, and becomes useless.

P.S.

In the end of the review I noticed that this prevents squashing not
only of operations with unsigned fields. It will prevent squashing if
the whole format has at least one unsigned. This makes current implementation
of squash even more useless, because forbids to use the fastest field type,
which is default when you create an index without specification of field type
btw.

> +	}
> +	return true;
> +}
> +
> +/**
> + * Unpack upsert's update operations from msgpack array
> + * into array of iovecs.
> + */
> +static void
> +upsert_ops_to_iovec(const char *ops, uint32_t ops_cnt, struct iovec *iov_arr)
> +{
> +	for (uint32_t i = 0; i < ops_cnt; ++i) {
> +		assert(mp_typeof(*ops) == MP_ARRAY);
> +		iov_arr[i].iov_base = (char *) ops;
> +		mp_next(&ops);
> +		iov_arr[i].iov_len = ops - (char *) iov_arr[i].iov_base;
> +	}
> +}
> +
>  struct tuple *
>  vy_apply_upsert(struct tuple *new_stmt, struct tuple *old_stmt,
>  		struct key_def *cmp_def, bool suppress_error)
> @@ -87,122 +248,74 @@ vy_apply_upsert(struct tuple *new_stmt, struct tuple *old_stmt,
>  	assert(new_stmt != old_stmt);
>  	assert(vy_stmt_type(new_stmt) == IPROTO_UPSERT);
>  
> -	if (old_stmt == NULL || vy_stmt_type(old_stmt) == IPROTO_DELETE) {
> -		/*
> -		 * INSERT case: return new stmt.
> -		 */
> -		return vy_stmt_replace_from_upsert(new_stmt);
> +	struct tuple *result_stmt = NULL;
> +	if (old_stmt == NULL || vy_stmt_type(old_stmt) != IPROTO_UPSERT) {
> +		return vy_apply_upsert_on_terminal_stmt(new_stmt, old_stmt,
> +						        cmp_def, suppress_error);
>  	}
>  
> -	struct tuple_format *format = tuple_format(new_stmt);
> -
> +	assert(vy_stmt_type(old_stmt) == IPROTO_UPSERT);

13. The assertion looks useless, since it is reverse of the {if}
condition above, but up to you.

>  	/*
> -	 * Unpack UPSERT operation from the new stmt
> +	 * Unpack UPSERT operation from the old and new stmts.
>  	 */
> +	assert(old_stmt != NULL);

14. This is strage to check old_stmt->type in the previous assertion before
you checked old_stmt != NULL.

>  	uint32_t mp_size;
> -	const char *new_ops;
> -	new_ops = vy_stmt_upsert_ops(new_stmt, &mp_size);
> -	const char *new_ops_end = new_ops + mp_size;
> +	const char *old_ops = vy_stmt_upsert_ops(old_stmt, &mp_size);
> +	const char *old_ops_end = old_ops + mp_size;
> +	assert(old_ops_end > old_ops);
> +	const char *old_stmt_mp = vy_upsert_data_range(old_stmt, &mp_size);
> +	const char *old_stmt_mp_end = old_stmt_mp + mp_size;
> +	const char *new_ops = vy_stmt_upsert_ops(new_stmt, &mp_size);
>  
>  	/*
> -	 * Apply new operations to the old stmt
> +	 * UPSERT + UPSERT case: squash arithmetic operations.
> +	 * Note that we can process this only in case result
> +	 * can't break format under no circumstances. Since
> +	 * subtraction can lead to negative values, unsigned
> +	 * field are considered to be inappropriate.
>  	 */
> -	const char *result_mp;
> -	if (vy_stmt_type(old_stmt) == IPROTO_UPSERT)
> -		result_mp = vy_upsert_data_range(old_stmt, &mp_size);
> -	else
> -		result_mp = tuple_data_range(old_stmt, &mp_size);
> -	const char *result_mp_end = result_mp + mp_size;
> -	struct tuple *result_stmt = NULL;
> +	struct tuple_format *format = tuple_format(old_stmt);
>  	struct region *region = &fiber()->gc;
>  	size_t region_svp = region_used(region);
> -	uint8_t old_type = vy_stmt_type(old_stmt);
> -	uint64_t column_mask = COLUMN_MASK_FULL;
> -	result_mp = xrow_upsert_execute(new_ops, new_ops_end, result_mp,
> -					result_mp_end, format, &mp_size,
> -					0, suppress_error, &column_mask);
> -	if (result_mp == NULL) {
> -		region_truncate(region, region_svp);
> -		return NULL;
> +	if (tuple_format_is_suitable_for_squash(format)) {
> +		const char *new_ops_end = new_ops + mp_size;
> +		if (vy_upsert_try_to_squash(format, old_stmt_mp, old_stmt_mp_end,
> +					    old_ops, old_ops_end, new_ops,
> +					    new_ops_end, &result_stmt) != 0) {
> +			/* OOM */
> +			region_truncate(region, region_svp);
> +			return NULL;
> +		}

15. vy_upsert_try_to_squash() returns a result into result_stmt. But
you ignore it. Basically, whatever it returns, you act like squash
didn't happen and it never works now. You continue to work with 2 old
operation set arrays. Also result_stmt leaks.

What is also strange - I added {assert(false);} here and the tests
passed. I thought we had quite a lot squash tests. Seems they are all
for formats having unsigned field type.

(Actualy the tests failed, but not here - on my machine vinyl tests
fail in almost 100% runs somewhere with random errors, could be
luajit problems on Mac maybe.)

>  	}
> -	result_mp_end = result_mp + mp_size;
> -	if (old_type != IPROTO_UPSERT) {
> -		assert(old_type == IPROTO_INSERT ||
> -		       old_type == IPROTO_REPLACE);
> -		/*
> -		 * UPDATE case: return the updated old stmt.
> -		 */
> -		result_stmt = vy_stmt_new_replace(format, result_mp,
> -						  result_mp_end);
> -		region_truncate(region, region_svp);
> -		if (result_stmt == NULL)
> -			return NULL; /* OOM */
> -		vy_stmt_set_lsn(result_stmt, vy_stmt_lsn(new_stmt));
> -		goto check_key;
> -	}
> -
> -	/*
> -	 * Unpack UPSERT operation from the old stmt
> -	 */
> -	assert(old_stmt != NULL);
> -	const char *old_ops;
> -	old_ops = vy_stmt_upsert_ops(old_stmt, &mp_size);
> -	const char *old_ops_end = old_ops + mp_size;
> -	assert(old_ops_end > old_ops);
> -
>  	/*
> -	 * UPSERT + UPSERT case: combine operations
> +	 * Adding update operations. We keep order of update operations in
> +	 * the array the same. It is vital since first set of operations
> +	 * must be skipped in case upsert folds into insert. For instance:
> +	 * old_ops = {{{op1}, {op2}}, {{op3}}}
> +	 * new_ops = {{{op4}, {op5}}}
> +	 * res_ops = {{{op1}, {op2}}, {{op3}}, {{op4}, {op5}}}
> +	 * If upsert corresponding to old_ops becomes insert, then
> +	 * {{op1}, {op2}} update operations are not applied.
>  	 */
> -	assert(old_ops_end - old_ops > 0);
> -	if (vy_upsert_try_to_squash(format, result_mp, result_mp_end,
> -				    old_ops, old_ops_end, new_ops, new_ops_end,
> -				    &result_stmt) != 0) {
> +	uint32_t old_ops_cnt = mp_decode_array(&old_ops);
> +	uint32_t new_ops_cnt = mp_decode_array(&new_ops);
> +	size_t ops_size = sizeof(struct iovec) * (old_ops_cnt + new_ops_cnt);
> +	struct iovec *operations = region_alloc(region, ops_size);

16. region_alloc_array.

17. But you don't really need that. Nor upsert_ops_to_iovec() function.
You could keep the old code almost as is, because for vy_stmt_new_with_ops()
to work correctly, it is not necessary to have each operation set in a
separate iovec. Anyway they are all copied as is without unpacking. You
could have 1 iovec for the root MP_ARRAY, 1 iovec for the old operation sets,
1 iovec for the new operation sets.

Having first iovec with root MP_ARRAY would allow to delete is_ops_encoded.

> +	if (operations == NULL) {
>  		region_truncate(region, region_svp);
> +		diag_set(OutOfMemory, ops_size, "region_alloc", "operations");
>  		return NULL;
>  	}
> -	if (result_stmt != NULL) {
> -		region_truncate(region, region_svp);
> -		vy_stmt_set_lsn(result_stmt, vy_stmt_lsn(new_stmt));
> -		goto check_key;
> -	}
> +	upsert_ops_to_iovec(old_ops, old_ops_cnt, operations);
> +	upsert_ops_to_iovec(new_ops, new_ops_cnt, &operations[old_ops_cnt]);
>  
> -	/* Failed to squash, simply add one upsert to another */
> -	int old_ops_cnt, new_ops_cnt;
> -	struct iovec operations[3];
> -
> -	old_ops_cnt = mp_decode_array(&old_ops);
> -	operations[1].iov_base = (void *)old_ops;
> -	operations[1].iov_len = old_ops_end - old_ops;
> -
> -	new_ops_cnt = mp_decode_array(&new_ops);
> -	operations[2].iov_base = (void *)new_ops;
> -	operations[2].iov_len = new_ops_end - new_ops;
> -
> -	char ops_buf[16];
> -	char *header = mp_encode_array(ops_buf, old_ops_cnt + new_ops_cnt);
> -	operations[0].iov_base = (void *)ops_buf;
> -	operations[0].iov_len = header - ops_buf;
> -
> -	result_stmt = vy_stmt_new_upsert(format, result_mp, result_mp_end,
> -					 operations, 3);
> +	result_stmt = vy_stmt_new_upsert(format, old_stmt_mp, old_stmt_mp_end,
> +					 operations, old_ops_cnt + new_ops_cnt,
> +					 false);
>  	region_truncate(region, region_svp);
>  	if (result_stmt == NULL)
>  		return NULL;
>  	vy_stmt_set_lsn(result_stmt, vy_stmt_lsn(new_stmt));
>  
> -check_key:
> -	/*
> -	 * Check that key hasn't been changed after applying operations.
> -	 */
> -	if (!key_update_can_be_skipped(cmp_def->column_mask, column_mask) &&
> -	    vy_stmt_compare(old_stmt, HINT_NONE, result_stmt,
> -			    HINT_NONE, cmp_def) != 0) {
> -		/*
> -		 * Key has been changed: ignore this UPSERT and
> -		 * @retval the old stmt.
> -		 */
> -		tuple_unref(result_stmt);
> -		result_stmt = vy_stmt_dup(old_stmt);
> -	}
>  	return result_stmt;
> diff --git a/test/vinyl/upsert.result b/test/vinyl/upsert.result
> index 3a7f6629d..a20db2ad2 100644
> --- a/test/vinyl/upsert.result
> +++ b/test/vinyl/upsert.result
> @@ -899,3 +899,292 @@ s:select()
>  s:drop()
>  ---
>  ...
> +-- gh-5107: don't squash upsert operations into one array.
> +--
> +-- Test upsert execution/squash referring to fields in reversed
> +-- order (via negative indexing).
> +--
> +s = box.schema.create_space('test', {engine = 'vinyl'})
> +---
> +...
> +pk = s:create_index('pk')
> +---
> +...
> +s:insert({1, 1, 1})
> +---
> +- [1, 1, 1]
> +...
> +box.snapshot()
> +---
> +- ok
> +...
> +s:upsert({1}, {{'=', 3, 100}})
> +---
> +...
> +s:upsert({1}, {{'=', -1, 200}})
> +---
> +...
> +box.snapshot()
> +---
> +- ok
> +...
> +s:select() -- {1, 1, 200}
> +---
> +- - [1, 1, 200]
> +...
> +s:delete({1})
> +---
> +...
> +s:insert({1, 1, 1})
> +---
> +- [1, 1, 1]
> +...
> +box.snapshot()
> +---
> +- ok
> +...
> +s:upsert({1}, {{'=', -3, 100}})
> +---
> +...
> +s:upsert({1}, {{'=', -1, 200}})
> +---
> +...
> +box.snapshot()
> +---
> +- ok
> +...
> +-- Two upserts are NOT squashed into one, so only one
> +-- (first one) is skipped, meanwhile second one is applied.
> +--
> +s:select() -- {1, 1, 1}
> +---
> +- - [1, 1, 200]
> +...
> +s:delete({1})
> +---
> +...
> +box.snapshot()
> +---
> +- ok
> +...
> +s:upsert({1, 1}, {{'=', -2, 300}}) -- {1, 1}
> +---
> +...
> +s:upsert({1}, {{'+', -1, 100}}) -- {1, 101}
> +---
> +...
> +s:upsert({1}, {{'-', 2, 100}}) -- {1, 1}
> +---
> +...
> +s:upsert({1}, {{'+', -1, 200}}) -- {1, 201}
> +---
> +...
> +s:upsert({1}, {{'-', 2, 200}}) -- {1, 1}
> +---
> +...
> +box.snapshot()
> +---
> +- ok
> +...
> +s:select() -- {1, 1}
> +---
> +- - [1, 1]
> +...
> +s:delete({1})
> +---
> +...
> +box.snapshot()
> +---
> +- ok
> +...
> +s:upsert({1, 1, 1}, {{'!', -1, 300}}) -- {1, 1, 1}
> +---
> +...
> +s:upsert({1}, {{'+', -2, 100}}) -- {1, 101, 1}
> +---
> +...
> +s:upsert({1}, {{'=', -1, 100}}) -- {1, 101, 100}
> +---
> +...
> +s:upsert({1}, {{'+', -1, 200}}) -- {1, 101, 300}
> +---
> +...
> +s:upsert({1}, {{'-', -2, 100}}) -- {1, 1, 300}
> +---
> +...
> +box.snapshot()
> +---
> +- ok
> +...
> +s:select()
> +---
> +- - [1, 1, 300]
> +...
> +s:drop()
> +---
> +...
> +-- Upsert operations which break space format are not applied.

18. You need to put references to the relevant issues in the tests,
using

    --
    -- gh-NNNN: description.
    --

format.

> +--
> +s = box.schema.space.create('test', { engine = 'vinyl', field_count = 2 })
> +---
> +...
> +pk = s:create_index('pk')
> +---
> +...
> +s:replace{1, 1}
> +---
> +- [1, 1]
> +...
> +-- Error is logged, upsert is not applied.
> +--
> +s:upsert({1, 1}, {{'=', 3, 5}})
> +---
> +...
> +-- During read the incorrect upsert is ignored.
> +--
> +s:select{}
> +---
> +- - [1, 1]
> +...
> +-- Try to set incorrect field_count in a transaction.
> +--
> +box.begin()
> +---
> +...
> +s:replace{2, 2}
> +---
> +- [2, 2]
> +...
> +s:upsert({2, 2}, {{'=', 3, 2}})
> +---
> +...
> +s:select{}
> +---
> +- - [1, 1]
> +  - [2, 2]
> +...
> +box.commit()
> +---
> +...
> +s:select{}
> +---
> +- - [1, 1]
> +  - [2, 2]
> +...
> +-- Read incorrect upsert from a run: it should be ignored.
> +--
> +box.snapshot()
> +---
> +- ok
> +...
> +s:select{}
> +---
> +- - [1, 1]
> +  - [2, 2]
> +...
> +s:upsert({2, 2}, {{'=', 3, 20}})
> +---
> +...
> +box.snapshot()
> +---
> +- ok
> +...
> +s:select{}
> +---
> +- - [1, 1]
> +  - [2, 2]
> +...
> +-- Execute replace/delete after invalid upsert.
> +--
> +box.snapshot()
> +---
> +- ok
> +...
> +s:upsert({2, 2}, {{'=', 3, 30}})
> +---
> +...
> +s:replace{2, 3}
> +---
> +- [2, 3]
> +...
> +s:select{}
> +---
> +- - [1, 1]
> +  - [2, 3]
> +...
> +s:upsert({1, 1}, {{'=', 3, 30}})
> +---
> +...
> +s:delete{1}
> +---
> +...
> +s:select{}
> +---
> +- - [2, 3]
> +...
> +-- Invalid upsert in a sequence of upserts is skipped meanwhile
> +-- the rest are applied.
> +--
> +box.snapshot()
> +---
> +- ok
> +...
> +s:upsert({2, 2}, {{'+', 2, 5}})
> +---
> +...
> +s:upsert({2, 2}, {{'=', 3, 40}})
> +---
> +...
> +s:upsert({2, 2}, {{'+', 2, 5}})
> +---
> +...
> +s:select{}
> +---
> +- - [2, 13]
> +...
> +box.snapshot()
> +---
> +- ok
> +...
> +s:select{}
> +---
> +- - [2, 13]
> +...
> +s:drop()
> +---
> +...
> +-- Make sure upserts satisfy associativity rule.
> +--
> +s = box.schema.space.create('test', {engine='vinyl'})
> +---
> +...
> +i = s:create_index('pk', {parts={2, 'uint'}})
> +---
> +...
> +s:replace{1, 2, 3, 'default'}
> +---
> +- [1, 2, 3, 'default']
> +...
> +box.snapshot()
> +---
> +- ok
> +...
> +s:upsert({2, 2, 2}, {{'=', 4, 'upserted'}})
> +---
> +...
> +-- Upsert will fail and thus ignored.
> +--
> +s:upsert({2, 2, 2}, {{'#', 1, 1}, {'!', 3, 1}})
> +---
> +...
> +box.snapshot()
> +---
> +- ok
> +...
> +s:select{}
> +---
> +- - [1, 2, 3, 'upserted']
> +...
> +s:drop()
> +---
> +...

19. All tests work with unsigned fields. So squashing is not tested here.