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

[Nikita Pettik <korablev@tarantool.org>]

On 13 Oct 22:00, Aleksandr Lyapunov wrote:
> Hi, thanks for the patch! Se my 2 comments below:
> (the second seems to be important)
> Once they resolved, the whole patch is LGTM!
> 
> On 10/3/20 4:28 PM, Nikita Pettik wrote:
> > into insert the upsert's tuple and the tuple stored in index can be
> > different In our particular case, tuple stored on disk has two fields
> Dot before `In` is missing.

Fixed.

> > +		/*
> > +		 * If it turns out that resulting tuple modifies primary
> > +		 * key, then simply ignore this upsert.
> > +		 */
> > +		if (stmt != NULL &&
> > +		    vy_apply_result_does_cross_pk(stmt, exec_res,
> > +						  exec_res + mp_size, cmp_def,
> > +						  column_mask)) {
> I guess in case if `stmt_is_void = true` we must compare with `upsert`
> tuple,
> not with `stmt` tuple.

Yep, thanks for input. Fixed and addded corresponding tests:

diff --git a/src/box/vy_upsert.c b/src/box/vy_upsert.c
index 090f3304b..0b4baf6ff 100644
--- a/src/box/vy_upsert.c
+++ b/src/box/vy_upsert.c
@@ -102,11 +102,15 @@ vy_apply_upsert_on_terminal_stmt(struct tuple *upsert, struct tuple *stmt,
        const char *ups_ops = new_ops;
        /*
         * In case upsert folds into insert, we must skip first
-        * update operations.
+        * update operations. Moreover, we should create (in case of delete
+        * statement - re-create since delete contains only key) tuple with
+        * format to use it for PK modification check.
         */
        if (stmt_is_void) {
                ups_cnt--;
                mp_next(&ups_ops);
+               stmt = vy_stmt_new_replace(tuple_format(upsert), result_mp,
+                                          result_mp_end);
        }
        for (uint32_t i = 0; i < ups_cnt; ++i) {
                assert(mp_typeof(*ups_ops) == MP_ARRAY);
@@ -122,6 +126,8 @@ vy_apply_upsert_on_terminal_stmt(struct tuple *upsert, struct tuple *stmt,
                                assert(e != NULL);
                                /* Bail out immediately in case of OOM. */
                                if (e->type != &type_ClientError) {
+                                       if (stmt_is_void)
+                                               tuple_unref(stmt);
                                        region_truncate(region, region_svp);
                                        return NULL;
                                }
@@ -134,8 +140,7 @@ vy_apply_upsert_on_terminal_stmt(struct tuple *upsert, struct tuple *stmt,
                 * If it turns out that resulting tuple modifies primary
                 * key, then simply ignore this upsert.
                 */
-               if (stmt != NULL &&
-                   vy_apply_result_does_cross_pk(stmt, exec_res,
+               if (vy_apply_result_does_cross_pk(stmt, exec_res,
                                                  exec_res + mp_size, cmp_def,
                                                  column_mask)) {
                        if (!suppress_error) {
@@ -164,6 +169,8 @@ vy_apply_upsert_on_terminal_stmt(struct tuple *upsert, struct tuple *stmt,
        struct tuple *new_terminal_stmt = vy_stmt_new_replace(format, result_mp,
                                                              result_mp_end);
        region_truncate(region, region_svp);
+       if (stmt_is_void)
+               tuple_unref(stmt);
        if (new_terminal_stmt == NULL)
                return NULL;
        vy_stmt_set_lsn(new_terminal_stmt, vy_stmt_lsn(upsert));
diff --git a/test/vinyl/upsert.result b/test/vinyl/upsert.result
index a66e87be0..fe673ad6f 100644
--- a/test/vinyl/upsert.result
+++ b/test/vinyl/upsert.result
@@ -1372,3 +1372,97 @@ s:select{}
 s:drop()
 ---
 ...
+-- Combination of upserts and underlying void (i.e. delete or null)
+-- statement on disk. Upsert modifying PK is skipped.
+--
+s = box.schema.space.create('test', {engine = 'vinyl'})
+---
+...
+i = s:create_index('test', { run_count_per_level = 20 })
+---
+...
+for i = 101, 110 do s:replace{i, i} end
+---
+...
+s:replace({1, 1})
+---
+- [1, 1]
+...
+box.snapshot()
+---
+- ok
+...
+s:delete({1})
+---
+...
+box.snapshot()
+---
+- ok
+...
+s:upsert({1, 1}, {{'=', 2, 2}})
+---
+...
+s:upsert({1, 1}, {{'=', 1, 0}})
+---
+...
+box.snapshot()
+---
+- ok
+...
+s:select()
+---
+- - [1, 1]
+  - [101, 101]
+  - [102, 102]
+  - [103, 103]
+  - [104, 104]
+  - [105, 105]
+  - [106, 106]
+  - [107, 107]
+  - [108, 108]
+  - [109, 109]
+  - [110, 110]
+...
+s:drop()
+---
+...
+s = box.schema.space.create('test', {engine = 'vinyl'})
+---
+...
+i = s:create_index('test', { run_count_per_level = 20 })
+---
+...
+for i = 101, 110 do s:replace{i, i} end
+---
+...
+box.snapshot()
+---
+- ok
+...
+s:upsert({1, 1}, {{'=', 2, 2}})
+---
+...
+s:upsert({1, 1}, {{'=', 1, 0}})
+---
+...
+box.snapshot()
+---
+- ok
+...
+s:select()
+---
+- - [1, 1]
+  - [101, 101]
+  - [102, 102]
+  - [103, 103]
+  - [104, 104]
+  - [105, 105]
+  - [106, 106]
+  - [107, 107]
+  - [108, 108]
+  - [109, 109]
+  - [110, 110]
+...
+s:drop()
+---
+...
diff --git a/test/vinyl/upsert.test.lua b/test/vinyl/upsert.test.lua
index ad69f4b72..b62c19978 100644
--- a/test/vinyl/upsert.test.lua
+++ b/test/vinyl/upsert.test.lua
@@ -566,3 +566,33 @@ box.snapshot()
 s:select{}
 
 s:drop()
+
+-- Combination of upserts and underlying void (i.e. delete or null)
+-- statement on disk. Upsert modifying PK is skipped.
+--
+s = box.schema.space.create('test', {engine = 'vinyl'})
+i = s:create_index('test', { run_count_per_level = 20 })
+
+for i = 101, 110 do s:replace{i, i} end
+s:replace({1, 1})
+box.snapshot()
+s:delete({1})
+box.snapshot()
+s:upsert({1, 1}, {{'=', 2, 2}})
+s:upsert({1, 1}, {{'=', 1, 0}})
+box.snapshot()
+s:select()
+
+s:drop()
+
+s = box.schema.space.create('test', {engine = 'vinyl'})
+i = s:create_index('test', { run_count_per_level = 20 })
+
+for i = 101, 110 do s:replace{i, i} end
+box.snapshot()
+s:upsert({1, 1}, {{'=', 2, 2}})
+s:upsert({1, 1}, {{'=', 1, 0}})
+box.snapshot()
+s:select()
+
+s:drop()


> > +			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;
> > +		/*
> > +		 * Result statement must satisfy space's format. Since upsert's
> > +		 * tuple correctness is already checked in vy_upsert(), let's
> > +		 * use its format to provide result verification.
> > +		 */
> > +		struct tuple_format *format = tuple_format(upsert);
> > +		if (tuple_validate_raw(format, 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;
> > +}