[PATCH 3/3] vinyl: fix deferred DELETE statement lost on commit

[Vladimir Davydov <vdavydov.dev@gmail.com>]

Even if a statement isn't marked as VY_STMT_DEFERRED_DELETE, e.g. it's
a REPLACE produced by an UPDATE request, it may overwrite a statement in
the transaction write set that is marked so, for instance:

  s = box.schema.space.create('test', {engine = 'vinyl'})
  pk = s:create_index('pk')
  sk = s:create_index('sk', {parts = {2, 'unsigned'}})

  s:insert{1, 1}

  box.begin()
  s:replace{1, 2}
  s:update(1, {{'=', 2, 3}})
  box.commit()

If we don't mark REPLACE{3,1} produced by the update operatoin with
VY_STMT_DEFERRED_DELETE flag, we will never generate a DELETE statement
for INSERT{1,1}. That is, we must inherit the flag from the overwritten
statement when we insert a new one into a write set.

Closes #4248
---
 src/box/vy_tx.c                     | 10 ++++++
 test/vinyl/deferred_delete.result   | 67 +++++++++++++++++++++++++++++++++++++
 test/vinyl/deferred_delete.test.lua | 21 ++++++++++++
 3 files changed, 98 insertions(+)

diff --git a/src/box/vy_tx.c b/src/box/vy_tx.c
index 119d975b..76d184fb 100644
--- a/src/box/vy_tx.c
+++ b/src/box/vy_tx.c
@@ -1092,6 +1092,16 @@ vy_tx_set_entry(struct vy_tx *tx, struct vy_lsm *lsm, struct vy_entry entry)
 		write_set_remove(&tx->write_set, old);
 		old->is_overwritten = true;
 		v->is_first_insert = old->is_first_insert;
+		/*
+		 * Inherit VY_STMT_DEFERRED_DELETE flag from the older
+		 * statement so as to generate a DELETE for the tuple
+		 * overwritten by this transaction.
+		 */
+		if (vy_stmt_flags(old->entry.stmt) & VY_STMT_DEFERRED_DELETE) {
+			uint8_t flags = vy_stmt_flags(entry.stmt);
+			vy_stmt_set_flags(entry.stmt, flags |
+					  VY_STMT_DEFERRED_DELETE);
+		}
 	}
 
 	if (old == NULL && vy_stmt_type(entry.stmt) == IPROTO_INSERT)
diff --git a/test/vinyl/deferred_delete.result b/test/vinyl/deferred_delete.result
index 3f187a5b..23c93f0f 100644
--- a/test/vinyl/deferred_delete.result
+++ b/test/vinyl/deferred_delete.result
@@ -804,6 +804,73 @@ sk:select()
 s:drop()
 ---
 ...
+--
+-- gh-4248 Deferred DELETE isn't produced on transaction commit.
+--
+s = box.schema.space.create('test', {engine = 'vinyl'})
+---
+...
+pk = s:create_index('pk')
+---
+...
+sk = s:create_index('sk', {parts = {2, 'unsigned'}})
+---
+...
+s:insert{1, 10}
+---
+- [1, 10]
+...
+s:insert{2, 20}
+---
+- [2, 20]
+...
+box.begin()
+---
+...
+s:replace{1, 11}
+---
+- [1, 11]
+...
+s:update(1, {{'=', 2, 12}})
+---
+- [1, 12]
+...
+s:update(2, {{'=', 2, 21}})
+---
+- [2, 21]
+...
+s:replace{2, 22}
+---
+- [2, 22]
+...
+box.commit()
+---
+...
+box.snapshot()
+---
+- ok
+...
+pk:stat().rows -- 2: REPLACE{1, 12} + REPLACE{2, 22}
+---
+- 2
+...
+sk:stat().rows -- ditto
+---
+- 2
+...
+pk:select()
+---
+- - [1, 12]
+  - [2, 22]
+...
+sk:select()
+---
+- - [1, 12]
+  - [2, 22]
+...
+s:drop()
+---
+...
 box.cfg{vinyl_cache = vinyl_cache}
 ---
 ...
diff --git a/test/vinyl/deferred_delete.test.lua b/test/vinyl/deferred_delete.test.lua
index 689c8f93..1bce954c 100644
--- a/test/vinyl/deferred_delete.test.lua
+++ b/test/vinyl/deferred_delete.test.lua
@@ -291,6 +291,27 @@ pk:select()
 sk:select()
 s:drop()
 
+--
+-- gh-4248 Deferred DELETE isn't produced on transaction commit.
+--
+s = box.schema.space.create('test', {engine = 'vinyl'})
+pk = s:create_index('pk')
+sk = s:create_index('sk', {parts = {2, 'unsigned'}})
+s:insert{1, 10}
+s:insert{2, 20}
+box.begin()
+s:replace{1, 11}
+s:update(1, {{'=', 2, 12}})
+s:update(2, {{'=', 2, 21}})
+s:replace{2, 22}
+box.commit()
+box.snapshot()
+pk:stat().rows -- 2: REPLACE{1, 12} + REPLACE{2, 22}
+sk:stat().rows -- ditto
+pk:select()
+sk:select()
+s:drop()
+
 box.cfg{vinyl_cache = vinyl_cache}
 
 --
-- 
2.11.0