[Tarantool-patches] [PATCH 0/2] vinyl: rework upsert internals

[Tarantool-patches] [PATCH 0/2] vinyl: rework upsert internals

[Nikita Pettik]

Issues:
https://github.com/tarantool/tarantool/issues/1622
https://github.com/tarantool/tarantool/issues/5105
https://github.com/tarantool/tarantool/issues/5092
https://github.com/tarantool/tarantool/issues/5107
Branch:
https://github.com/tarantool/tarantool/tree/np/gh-5107-dont-squash-ops

Nikita Pettik (2):
  vy_stmt: introduce vy_stmt_is_void() helper
  vinyl: rework upsert operation

 src/box/vinyl.c                 |   2 +-
 src/box/vy_stmt.c               |  28 ++--
 src/box/vy_stmt.h               |  11 +-
 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, 650 insertions(+), 108 deletions(-)

-- 
2.15.1

[Tarantool-patches] [PATCH 1/2] vy_stmt: introduce vy_stmt_is_void() helper

[Nikita Pettik]

Sometimes it is quite useful to know whether vy_stmt is void (meaning
that statement is NULL or corresponds to delete statement) or not. So
let's add such method and use it in upcoming refactoring of upsert
operation.

Needed for #5107
---
 src/box/vy_stmt.h | 6 ++++++
 1 file changed, 6 insertions(+)

diff --git a/src/box/vy_stmt.h b/src/box/vy_stmt.h
index 25219230d..1b718e26b 100644
--- a/src/box/vy_stmt.h
+++ b/src/box/vy_stmt.h
@@ -300,6 +300,12 @@ vy_stmt_is_empty_key(struct tuple *stmt)
 	return tuple_field_count(stmt) == 0;
 }
 
+static inline bool
+vy_stmt_is_void(struct tuple *stmt)
+{
+	return stmt == NULL || vy_stmt_type(stmt) == IPROTO_DELETE;
+}
+
 /**
  * Duplicate the statememnt.
  *
-- 
2.15.1

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

[Nikita Pettik]

Previous upsert implementation had a few drawback which led to several
bugs and issues.

Issue #5092 (redundant update operations execution)

In a nutshell, application of upsert(s) consists of two actions
(see vy_apply_upsert()): execute and squash. Consider example:

insert({1, 1})  -- stored on disk
upsert({1}, {{'-', 2, 20}}) -- old ups1
upsert({1}, {{'+', 2, 10}}) -- new ups2

'Execute' takes update operations from the new upsert and combines them
with key of the old upsert.  {1} --> {'+', 2, 10} can't be evaluated since
key consists of only one field. Note that in case upsert doesn't fold
into insert the upsert's tuple and the tuple stored in index are
different. In our particular case, tuple stored on disk has two fields,
so upsert's update operation can be applied to it. If upsert's operation
can't be executed using key of old upsert, we simply continue processing
squash step. Squash is a combination of update operations: arithmetic
operations are combined so we don't have to store actions over the same
field; the rest operations - are merged into single array. As a result,
we get one upsert with combined operations: upsert({1}, {{'+', 2, -10}}).
Then vy_apply_upsert is called again to apply new upsert on the top of
terminal statement - insert{1, 1}. Since now tuple has second field,
update operations can be executed and corresponding result is {1, -9}
which in turn is the final result of upsert merging procedure.
Now imagine that we have following upserts:

upsert({1, 1}, {{'-', 2, 20}}) -- old ups1
upsert({1}, {{'+', 2, 10}}) -- new ups2

In this case tuple execution successfully finishes and modifies upsert's
tuple: {2, 1} --> {'+', 2, 10} == {2, 11}
However, we still have to squash/accumulate update operations since they
should be applied on tuple stored on disk later. After all, at the we
have next upsert: upsert({2, 11}, {{'+', 2, -10}}). Then it is applied
on the top of insert({1, 1}) and we get the same result as in the first
case - {1, -9}. The only difference is that upsert's tuple was modified.
As one can see, execution of update operations applied to upsert's tuple
is redundant in the case index already contains tuple with the same key
(i.e. when upserts turns into update). Instead, we are able to
accumulate/squash update operations only. When the last upsert is being
applied, we can either execute all update operation on tuple fetched
from index (i.e. upsert is update) OR on tuple specified in the first
upsert (i.e. first upsert is insert).

Issue #5105 (upsert doesn't follow associative property)

Secondly, current approach breaks associative property: after upserts'
update operations are merged into one array, part of them (related to
one upsert) can be skipped, meanwhile the rest - is applied. For
instance:

-- Index is over second field.
i = s:create_index('pk', {parts={2, 'uint'}})
s:replace{1, 2, 3, 'default'}
s:upsert({2, 2, 2}, {{'=', 4, 'upserted'}})
-- First update operation modifies primary key, so upsert must be ignored.
s:upsert({2, 2, 2}, {{'#', 1, 1}, {'!', 3, 1}})

After merging two upserts we get the next one:
upsert({2, 2, 2}, {{'=', 4, 'upserted'}, {'#', 1, 1}, {'!', 3, 1}}

While we executing update operations, we don't tell ones from different
upserts. Thus, if one operation fails, the rest are ignored as well. As
a result, first upsert won't be applied, even despite the fact it is
absolutely OK.

To resolve this issue, let's group update operations of each upsert into
separate array. So that operations related to particular upsert are
stored in single array. In terms of previous example we will get:
upsert({2, 2, 2}, {{{'=', 4, 'upserted'}}, {{'#', 1, 1}, {'!', 3, 1}}}

Also note that we don't longer have to apply update operations on tuple
in vy_apply_upsert() if we deal with two upserts: it can be done once we
face terminal statement; or if there's no underlying statement (it is
delete op or doesn't exist at all) we apply all update arrays except the
first one on upsert's tuple.

Arithmetic operations still can be combined in case there's no unsigned
fields in space format. Otherwise, result of subtraction can turn out to
be negative and resulting tuple won't satisfy this property.

This patch also introduces format check of upsert application (#1622
issue). In case it doesn't satisfy space's format, corresponding error
is logged and upsert is skipped.

Closes #1622
Closes #5105
Closes #5092
Part of #5107
---
 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/vinyl.c b/src/box/vinyl.c
index 32301d7ba..eab688147 100644
--- a/src/box/vinyl.c
+++ b/src/box/vinyl.c
@@ -1984,7 +1984,7 @@ vy_lsm_upsert(struct vy_tx *tx, struct vy_lsm *lsm,
 	operations[0].iov_base = (void *)expr;
 	operations[0].iov_len = expr_end - expr;
 	vystmt = vy_stmt_new_upsert(lsm->mem_format, tuple, tuple_end,
-				    operations, 1);
+				    operations, 1, false);
 	if (vystmt == NULL)
 		return -1;
 	assert(vy_stmt_type(vystmt) == IPROTO_UPSERT);
diff --git a/src/box/vy_stmt.c b/src/box/vy_stmt.c
index 92e0aa1c5..f1833e0b6 100644
--- a/src/box/vy_stmt.c
+++ b/src/box/vy_stmt.c
@@ -313,16 +313,22 @@ vy_key_dup(const char *key)
 static struct tuple *
 vy_stmt_new_with_ops(struct tuple_format *format, const char *tuple_begin,
 		     const char *tuple_end, struct iovec *ops,
-		     int op_count, enum iproto_type type)
+		     int op_count, enum iproto_type type, bool is_ops_encoded)
 {
 	mp_tuple_assert(tuple_begin, tuple_end);
 
 	const char *tmp = tuple_begin;
 	mp_decode_array(&tmp);
 
+	/*
+	 * ops are grouped in one extra array.
+	 * See vy_apply_upsert() for details.
+	 */
 	size_t ops_size = 0;
 	for (int i = 0; i < op_count; ++i)
 		ops_size += ops[i].iov_len;
+	if (! is_ops_encoded)
+		ops_size += mp_sizeof_array(op_count);
 
 	struct tuple *stmt = NULL;
 	struct region *region = &fiber()->gc;
@@ -360,6 +366,8 @@ vy_stmt_new_with_ops(struct tuple_format *format, const char *tuple_begin,
 	field_map_build(&builder, wpos - field_map_size);
 	memcpy(wpos, tuple_begin, mpsize);
 	wpos += mpsize;
+	if (! is_ops_encoded)
+		wpos = mp_encode_array(wpos, op_count);
 	for (struct iovec *op = ops, *end = ops + op_count;
 	     op != end; ++op) {
 		memcpy(wpos, op->iov_base, op->iov_len);
@@ -374,10 +382,11 @@ end:
 struct tuple *
 vy_stmt_new_upsert(struct tuple_format *format, const char *tuple_begin,
 		   const char *tuple_end, struct iovec *operations,
-		   uint32_t ops_cnt)
+		   uint32_t ops_cnt, bool is_ops_encoded)
 {
 	return vy_stmt_new_with_ops(format, tuple_begin, tuple_end,
-				    operations, ops_cnt, IPROTO_UPSERT);
+				    operations, ops_cnt, IPROTO_UPSERT,
+				    is_ops_encoded);
 }
 
 struct tuple *
@@ -385,7 +394,7 @@ vy_stmt_new_replace(struct tuple_format *format, const char *tuple_begin,
 		    const char *tuple_end)
 {
 	return vy_stmt_new_with_ops(format, tuple_begin, tuple_end,
-				    NULL, 0, IPROTO_REPLACE);
+				    NULL, 0, IPROTO_REPLACE, true);
 }
 
 struct tuple *
@@ -393,7 +402,7 @@ vy_stmt_new_insert(struct tuple_format *format, const char *tuple_begin,
 		   const char *tuple_end)
 {
 	return vy_stmt_new_with_ops(format, tuple_begin, tuple_end,
-				    NULL, 0, IPROTO_INSERT);
+				    NULL, 0, IPROTO_INSERT, true);
 }
 
 struct tuple *
@@ -401,7 +410,7 @@ vy_stmt_new_delete(struct tuple_format *format, const char *tuple_begin,
 		   const char *tuple_end)
 {
 	return vy_stmt_new_with_ops(format, tuple_begin, tuple_end,
-				    NULL, 0, IPROTO_DELETE);
+				    NULL, 0, IPROTO_DELETE, true);
 }
 
 struct tuple *
@@ -735,19 +744,20 @@ vy_stmt_decode(struct xrow_header *xrow, struct tuple_format *format)
 		/* Always use key format for DELETE statements. */
 		stmt = vy_stmt_new_with_ops(env->key_format,
 					    request.key, request.key_end,
-					    NULL, 0, IPROTO_DELETE);
+					    NULL, 0, IPROTO_DELETE, true);
 		break;
 	case IPROTO_INSERT:
 	case IPROTO_REPLACE:
 		stmt = vy_stmt_new_with_ops(format, request.tuple,
 					    request.tuple_end,
-					    NULL, 0, request.type);
+					    NULL, 0, request.type, true);
 		break;
 	case IPROTO_UPSERT:
 		ops.iov_base = (char *)request.ops;
 		ops.iov_len = request.ops_end - request.ops;
 		stmt = vy_stmt_new_upsert(format, request.tuple,
-					  request.tuple_end, &ops, 1);
+					  request.tuple_end, &ops,
+					  1, true);
 		break;
 	default:
 		/* TODO: report filename. */
diff --git a/src/box/vy_stmt.h b/src/box/vy_stmt.h
index 1b718e26b..4619fa218 100644
--- a/src/box/vy_stmt.h
+++ b/src/box/vy_stmt.h
@@ -534,6 +534,8 @@ vy_stmt_new_delete(struct tuple_format *format, const char *tuple_begin,
  * @param part_count Part count from key definition.
  * @param operations Vector of update operations.
  * @param ops_cnt Length of the update operations vector.
+ * @param is_ops_encoded True, if update operations are already packed
+  *                      into extra msgpack array.
  *
  * @retval NULL     Memory allocation error.
  * @retval not NULL Success.
@@ -541,7 +543,8 @@ vy_stmt_new_delete(struct tuple_format *format, const char *tuple_begin,
 struct tuple *
 vy_stmt_new_upsert(struct tuple_format *format,
 		   const char *tuple_begin, const char *tuple_end,
-		   struct iovec *operations, uint32_t ops_cnt);
+		   struct iovec *operations, uint32_t ops_cnt,
+		   bool is_ops_encoded);
 
 /**
  * Create REPLACE statement from UPSERT statement.
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);
+		tuple_unref(tuple);
+		return cmp_res != 0;
+	}
+	return false;
+}
+
+/**
+ * Apply update operations stored in @new_stmt (which is assumed to
+ * 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
+ * flag.
+ *
+ * @upsert Upsert statement to be applied on @stmt.
+ * @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.
+ */
+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))
+		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) {
+				assert(diag_last_error(diag_get()) != 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.
+		 */
+		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;
+	}
+	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);
 	/*
-	 * Unpack UPSERT operation from the new stmt
+	 * Unpack UPSERT operation from the old and new stmts.
 	 */
+	assert(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;
+		}
 	}
-	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);
+	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/unit/vy_iterators_helper.c b/test/unit/vy_iterators_helper.c
index 0d20f19ef..15470920b 100644
--- a/test/unit/vy_iterators_helper.c
+++ b/test/unit/vy_iterators_helper.c
@@ -112,7 +112,7 @@ vy_new_simple_stmt(struct tuple_format *format, struct key_def *key_def,
 			ops = mp_encode_int(ops, templ->upsert_value);
 		operations[0].iov_base = tmp;
 		operations[0].iov_len = ops - tmp;
-		ret = vy_stmt_new_upsert(format, buf, pos, operations, 1);
+		ret = vy_stmt_new_upsert(format, buf, pos, operations, 1, true);
 		fail_if(ret == NULL);
 		break;
 	}
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.
+--
+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()
+---
+...
diff --git a/test/vinyl/upsert.test.lua b/test/vinyl/upsert.test.lua
index 1d77474da..4a350344d 100644
--- a/test/vinyl/upsert.test.lua
+++ b/test/vinyl/upsert.test.lua
@@ -372,3 +372,124 @@ box.snapshot()
 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})
+box.snapshot()
+
+s:upsert({1}, {{'=', 3, 100}})
+s:upsert({1}, {{'=', -1, 200}})
+box.snapshot()
+s:select() -- {1, 1, 200}
+
+s:delete({1})
+s:insert({1, 1, 1})
+box.snapshot()
+
+s:upsert({1}, {{'=', -3, 100}})
+s:upsert({1}, {{'=', -1, 200}})
+box.snapshot()
+-- Two upserts are NOT squashed into one, so only one
+-- (first one) is skipped, meanwhile second one is applied.
+--
+s:select() -- {1, 1, 1}
+
+s:delete({1})
+box.snapshot()
+
+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()
+s:select() -- {1, 1}
+
+s:delete({1})
+box.snapshot()
+
+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()
+s:select()
+
+s:drop()
+
+-- Upsert operations which break space format are not applied.
+--
+s = box.schema.space.create('test', { engine = 'vinyl', field_count = 2 })
+pk = s:create_index('pk')
+s:replace{1, 1}
+-- Error is logged, upsert is not applied.
+--
+s:upsert({1, 1}, {{'=', 3, 5}})
+-- During read the incorrect upsert is ignored.
+--
+s:select{}
+
+-- Try to set incorrect field_count in a transaction.
+--
+box.begin()
+s:replace{2, 2}
+s:upsert({2, 2}, {{'=', 3, 2}})
+s:select{}
+box.commit()
+s:select{}
+
+-- Read incorrect upsert from a run: it should be ignored.
+--
+box.snapshot()
+s:select{}
+s:upsert({2, 2}, {{'=', 3, 20}})
+box.snapshot()
+s:select{}
+
+-- Execute replace/delete after invalid upsert.
+--
+box.snapshot()
+s:upsert({2, 2}, {{'=', 3, 30}})
+s:replace{2, 3}
+s:select{}
+
+s:upsert({1, 1}, {{'=', 3, 30}})
+s:delete{1}
+s:select{}
+
+-- Invalid upsert in a sequence of upserts is skipped meanwhile
+-- the rest are applied.
+--
+box.snapshot()
+s:upsert({2, 2}, {{'+', 2, 5}})
+s:upsert({2, 2}, {{'=', 3, 40}})
+s:upsert({2, 2}, {{'+', 2, 5}})
+s:select{}
+box.snapshot()
+s:select{}
+
+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'}
+box.snapshot()
+
+s:upsert({2, 2, 2}, {{'=', 4, 'upserted'}})
+-- Upsert will fail and thus ignored.
+--
+s:upsert({2, 2, 2}, {{'#', 1, 1}, {'!', 3, 1}})
+box.snapshot()
+
+s:select{}
+
+s:drop()
-- 
2.15.1

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

[Vladislav Shpilevoy]

Thanks for the patch!

See 5 comments below.

On 29.07.2020 03:15, Nikita Pettik wrote:
> Previous upsert implementation had a few drawback which led to several
> bugs and issues.
> 
> Issue #5092 (redundant update operations execution)
> 
> In a nutshell, application of upsert(s) consists of two actions
> (see vy_apply_upsert()): execute and squash. Consider example:
> 
> insert({1, 1})  -- stored on disk
> upsert({1}, {{'-', 2, 20}}) -- old ups1
> upsert({1}, {{'+', 2, 10}}) -- new ups2
> 
> 'Execute' takes update operations from the new upsert and combines them
> with key of the old upsert.  {1} --> {'+', 2, 10} can't be evaluated since
> key consists of only one field. Note that in case upsert doesn't fold
> into insert the upsert's tuple and the tuple stored in index are
> different. In our particular case, tuple stored on disk has two fields,
> so upsert's update operation can be applied to it. If upsert's operation
> can't be executed using key of old upsert, we simply continue processing
> squash step. Squash is a combination of update operations: arithmetic
> operations are combined so we don't have to store actions over the same
> field; the rest operations - are merged into single array. As a result,
> we get one upsert with combined operations: upsert({1}, {{'+', 2, -10}}).
> Then vy_apply_upsert is called again to apply new upsert on the top of
> terminal statement - insert{1, 1}. Since now tuple has second field,
> update operations can be executed and corresponding result is {1, -9}
> which in turn is the final result of upsert merging procedure.
> Now imagine that we have following upserts:
> 
> upsert({1, 1}, {{'-', 2, 20}}) -- old ups1
> upsert({1}, {{'+', 2, 10}}) -- new ups2
> 
> In this case tuple execution successfully finishes and modifies upsert's

1. 'tuple execution'?

> tuple: {2, 1} --> {'+', 2, 10} == {2, 11}

2. Where did you get {2, 1}? I see {1, 1} in this example.

> However, we still have to squash/accumulate update operations since they
> should be applied on tuple stored on disk later. After all, at the we

3. 'at the we'?

> have next upsert: upsert({2, 11}, {{'+', 2, -10}}). Then it is applied
> on the top of insert({1, 1}) and we get the same result as in the first
> case - {1, -9}. The only difference is that upsert's tuple was modified.
> As one can see, execution of update operations applied to upsert's tuple
> is redundant in the case index already contains tuple with the same key
> (i.e. when upserts turns into update). Instead, we are able to
> accumulate/squash update operations only. When the last upsert is being
> applied, we can either execute all update operation on tuple fetched
> from index (i.e. upsert is update) OR on tuple specified in the first
> upsert (i.e. first upsert is insert).
> 
> Issue #5105 (upsert doesn't follow associative property)
> 
> Secondly, current approach breaks associative property: after upserts'
> update operations are merged into one array, part of them (related to
> one upsert) can be skipped, meanwhile the rest - is applied. For
> instance:
> 
> -- Index is over second field.
> i = s:create_index('pk', {parts={2, 'uint'}})
> s:replace{1, 2, 3, 'default'}
> s:upsert({2, 2, 2}, {{'=', 4, 'upserted'}})
> -- First update operation modifies primary key, so upsert must be ignored.
> s:upsert({2, 2, 2}, {{'#', 1, 1}, {'!', 3, 1}})
> 
> After merging two upserts we get the next one:
> upsert({2, 2, 2}, {{'=', 4, 'upserted'}, {'#', 1, 1}, {'!', 3, 1}}
> 
> While we executing update operations, we don't tell ones from different

4. Couldn't parse "don't tell ones from different upserts".

> upserts. Thus, if one operation fails, the rest are ignored as well. As
> a result, first upsert won't be applied, even despite the fact it is
> absolutely OK.
> 
> To resolve this issue, let's group update operations of each upsert into
> separate array. So that operations related to particular upsert are
> stored in single array. In terms of previous example we will get:
> upsert({2, 2, 2}, {{{'=', 4, 'upserted'}}, {{'#', 1, 1}, {'!', 3, 1}}}
> 
> Also note that we don't longer have to apply update operations on tuple
> in vy_apply_upsert() if we deal with two upserts: it can be done once we
> face terminal statement; or if there's no underlying statement (it is
> delete op or doesn't exist at all) we apply all update arrays except the
> first one on upsert's tuple.
> 
> Arithmetic operations still can be combined in case there's no unsigned
> fields in space format. Otherwise, result of subtraction can turn out to
> be negative and resulting tuple won't satisfy this property.

5. Non-unsigned fields also can lead to an error. For example, signed
integer overflow. It seems, squashing becomes impossible at all. Most of
the operations are arithmetic for int and uint fields, which can throw an
error both on + and -.

> This patch also introduces format check of upsert application (#1622
> issue). In case it doesn't satisfy space's format, corresponding error
> is logged and upsert is skipped.
> 
> Closes #1622
> Closes #5105
> Closes #5092
> Part of #5107

I didn't review the rest yet. Will provide more comments afterwards.

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

[Vladislav Shpilevoy]

Hi! Thanks for the patchset!

On 29.07.2020 03:15, Nikita Pettik wrote:
> Issues:
> https://github.com/tarantool/tarantool/issues/1622
> https://github.com/tarantool/tarantool/issues/5105
> https://github.com/tarantool/tarantool/issues/5092
> https://github.com/tarantool/tarantool/issues/5107
> Branch:
> https://github.com/tarantool/tarantool/tree/np/gh-5107-dont-squash-ops

Need a changelog here.

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

[Vladislav Shpilevoy]

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.

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

[Nikita Pettik]

On 31 Jul 01:32, Vladislav Shpilevoy wrote:
> Hi! Thanks for the patchset!
> 
> On 29.07.2020 03:15, Nikita Pettik wrote:
> > Issues:
> > https://github.com/tarantool/tarantool/issues/1622
> > https://github.com/tarantool/tarantool/issues/5105
> > https://github.com/tarantool/tarantool/issues/5092
> > https://github.com/tarantool/tarantool/issues/5107
> > Branch:
> > https://github.com/tarantool/tarantool/tree/np/gh-5107-dont-squash-ops
> 
> Need a changelog here.

Added in V2:

@ChangeLog:
 - Rework upsert operation in vinyl so that now (gh-5107):
   - if upsert can't be applied it is skipped and corresponding error is logged (gh-1622);
   - upserts now follow associative property: result of several upserts
     doesn't depend on the order of their application (gh-5105);
   - upserts referring to -1 fieldno are handled correctly now (gh-5087).

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

[Nikita Pettik]

On 02 Aug 16:44, Vladislav Shpilevoy wrote:
> Thanks for the patch!
> 
> ASAN tests on the branch are failing:
> https://gitlab.com/tarantool/tarantool/-/jobs/661977877

Hi. I've sent second iteration of patch containing fixes to
all your comments below. ASAN now is OK as well:
https://gitlab.com/tarantool/tarantool/-/jobs/677105339
 
> > +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.

Fixed.
 
> > +		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.

I'm really sorry for that (haven't been writing comments in doxy
style for a while). Fixed all comments.

> 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.

Fixed.

> 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>.

Fixed.

> > + * @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.

Fixed.

> 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.

OK!

> > + */
> > +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.

Ok, refactored and dropped commit containing introduction of vy_stmt_is_void().

> > +	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.

Fixed.
 
> > +				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);}.

OK, fixed.

> > +				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.

Fixed.

> > +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.

It's true. In its previous implementation it was almost useless.
I've reworked this part in V2 and integrated format check right
in xrow_upsert_squash() so that we can operate on particular values
of squash result (e.g. if format declares unsigned field and the
result of squash is negative - operations are not squashed).
See update_arith_op_does_satisfy_format() and vy_upsert_try_to_squash().
 
> 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.
> 
> > @@ -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);
> >  	}
> >  
> > +	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.

Skipped (imho it increases a bit code readability).
 
> >  	/*
> > -	 * 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.

Agree, swapped these asserts.

> > -	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.

My aplogies for this broken part of patch, somehow I've missed it..
I've reworked it in patch V2 (see comment above).
 
> (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.)
> 
> > +	 * 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.

iovecs really simply code and workflow with update operations.
I use them more intensely in new patch version.

> 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.

Not sure if it possible in V2...
 
> > +	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]);
> >  
> 
> 18. You need to put references to the relevant issues in the tests,
> using
> 
>     --
>     -- gh-NNNN: description.
>     --

Fixed (added tags).

> > +- ok
> > +...
> > +s:select{}
> > +---
> > +- - [1, 2, 3, 'upserted']
> > +...
> > +s:drop()
> > +---
> > +...
> 
> 19. All tests work with unsigned fields. So squashing is not tested here.

In new patch squashing requirements become more tolerant, so squashing
now takes place in these tests.