[tarantool-patches] Re: [PATCH v2 2/2] Fixed lost format on update and upsert operations.

[Kirill Shcherbatov <kshcherbatov@tarantool.org>]

From c268dedd32ed124fb25c46e3fb0bf1cd507fda62 Mon Sep 17 00:00:00 2001
From: Kirill Shcherbatov <kshcherbatov@tarantool.org>
Date: Fri, 6 Apr 2018 19:21:32 +0300
Subject: [PATCH] Fixed lost format on update and upsert operations.

Fixes #3051
---
 src/box/lua/tuple.c      |  31 +++++++++++--
 src/box/memtx_tuple.cc   |   1 +
 src/box/tuple.c          |  31 ++++++++-----
 src/box/tuple.h          |   2 +-
 src/box/tuple_format.h   |   5 +-
 src/box/vy_stmt.c        |   7 +++
 src/box/vy_stmt.h        |   9 ++++
 test/box/update.result   | 118 +++++++++++++++++++++++++++++++++++++++++++++++
 test/box/update.test.lua |  41 ++++++++++++++++
 9 files changed, 226 insertions(+), 19 deletions(-)

diff --git a/src/box/lua/tuple.c b/src/box/lua/tuple.c
index 47b33c9..9bb62b0 100644
--- a/src/box/lua/tuple.c
+++ b/src/box/lua/tuple.c
@@ -36,6 +36,7 @@
 #include <small/ibuf.h>
 #include <small/region.h>
 #include <fiber.h>
+#include <box/tuple_update.h>
 
 #include "box/tuple.h"
 #include "box/tuple_convert.h"
@@ -390,15 +391,35 @@ lbox_tuple_transform(struct lua_State *L)
 	}
 	mpstream_flush(&stream);
 
-	/* Execute tuple_update */
-	struct tuple *new_tuple =
-		box_tuple_update(tuple, buf->buf, buf->buf + ibuf_used(buf));
+	uint32_t new_size = 0, bsize;
+	const char *old_data = tuple_data_range(tuple, &bsize);
+	struct region *region = &fiber()->gc;
+	size_t used = region_used(region);
+	const char *new_data = tuple_update_execute(region_aligned_alloc_cb,
+	                                            region,
+	                                            buf->buf,
+	                                            buf->buf + ibuf_used(buf),
+	                                            old_data,
+	                                            old_data + bsize,
+	                                            &new_size, 1, NULL);
+	if (new_data == NULL) {
+		region_truncate(region, used);
+		goto error;
+	}
+	struct tuple *new_tuple = tuple_new(box_tuple_format_default(),
+	                                    new_data, new_data + new_size);
+	region_truncate(region, used);
 	if (new_tuple == NULL)
-		luaT_error(L);
-	/* box_tuple_update() doesn't leak on exception, see public API doc */
+		goto error;
+	tuple_bless(new_tuple);
+
 	luaT_pushtuple(L, new_tuple);
 	ibuf_reset(buf);
 	return 1;
+error:
+	if (new_tuple == NULL)
+		luaT_error(L);
+	return 0;
 }
 
 /**
diff --git a/src/box/memtx_tuple.cc b/src/box/memtx_tuple.cc
index 70453bd..1990bdf 100644
--- a/src/box/memtx_tuple.cc
+++ b/src/box/memtx_tuple.cc
@@ -90,6 +90,7 @@ memtx_tuple_free(void)
 
 struct tuple_format_vtab memtx_tuple_format_vtab = {
 	memtx_tuple_delete,
+	memtx_tuple_new,
 };
 
 struct tuple *
diff --git a/src/box/tuple.c b/src/box/tuple.c
index d4760f3..cb57d81 100644
--- a/src/box/tuple.c
+++ b/src/box/tuple.c
@@ -61,15 +61,19 @@ struct tuple_format *tuple_format_runtime;
 static void
 runtime_tuple_delete(struct tuple_format *format, struct tuple *tuple);
 
+struct tuple *
+runtime_tuple_new(struct tuple_format *format, const char *data, const char *end);
+
 /** A virtual method table for tuple_format_runtime */
 static struct tuple_format_vtab tuple_format_runtime_vtab = {
 	runtime_tuple_delete,
+	runtime_tuple_new,
 };
 
 struct tuple *
-tuple_new(struct tuple_format *format, const char *data, const char *end)
+runtime_tuple_new(struct tuple_format *format, const char *data, const char *end)
 {
-	assert(format->vtab.destroy == tuple_format_runtime_vtab.destroy);
+	assert(format->vtab.tuple_delete == tuple_format_runtime_vtab.tuple_delete);
 
 	mp_tuple_assert(data, end);
 	size_t data_len = end - data;
@@ -102,7 +106,7 @@ tuple_new(struct tuple_format *format, const char *data, const char *end)
 static void
 runtime_tuple_delete(struct tuple_format *format, struct tuple *tuple)
 {
-	assert(format->vtab.destroy == tuple_format_runtime_vtab.destroy);
+	assert(format->vtab.tuple_delete == tuple_format_runtime_vtab.tuple_delete);
 	say_debug("%s(%p)", __func__, tuple);
 	assert(tuple->refs == 0);
 	size_t total = sizeof(struct tuple) + tuple_format_meta_size(format) +
@@ -147,6 +151,12 @@ tuple_validate_raw(struct tuple_format *format, const char *tuple)
 	return 0;
 }
 
+struct tuple *
+tuple_new(struct tuple_format *format, const char *data, const char *end)
+{
+	return format->vtab.tuple_new(format, data, end);
+}
+
 /**
  * Incremented on every snapshot and is used to distinguish tuples
  * which were created after start of a snapshot (these tuples can
@@ -396,22 +406,20 @@ box_tuple_t *
 box_tuple_update(const box_tuple_t *tuple, const char *expr,
 		 const char *expr_end)
 {
-	struct tuple_format *format = tuple_format_runtime;
-
 	uint32_t new_size = 0, bsize;
 	const char *old_data = tuple_data_range(tuple, &bsize);
 	struct region *region = &fiber()->gc;
 	size_t used = region_used(region);
 	const char *new_data =
 		tuple_update_execute(region_aligned_alloc_cb, region, expr,
-				     expr_end, old_data, old_data + bsize,
-				     &new_size, 1, NULL);
+		                     expr_end, old_data, old_data + bsize,
+		                     &new_size, 1, NULL);
 	if (new_data == NULL) {
 		region_truncate(region, used);
 		return NULL;
 	}
-
-	struct tuple *ret = tuple_new(format, new_data, new_data + new_size);
+	struct tuple *ret = tuple_new(tuple_format(tuple), new_data,
+	                              new_data + new_size);
 	region_truncate(region, used);
 	if (ret != NULL)
 		return tuple_bless(ret);
@@ -422,8 +430,6 @@ box_tuple_t *
 box_tuple_upsert(const box_tuple_t *tuple, const char *expr,
 		 const char *expr_end)
 {
-	struct tuple_format *format = tuple_format_runtime;
-
 	uint32_t new_size = 0, bsize;
 	const char *old_data = tuple_data_range(tuple, &bsize);
 	struct region *region = &fiber()->gc;
@@ -437,7 +443,8 @@ box_tuple_upsert(const box_tuple_t *tuple, const char *expr,
 		return NULL;
 	}
 
-	struct tuple *ret = tuple_new(format, new_data, new_data + new_size);
+	struct tuple *ret = tuple_new(tuple_format(tuple),
+	                              new_data, new_data + new_size);
 	region_truncate(region, used);
 	if (ret != NULL)
 		return tuple_bless(ret);
diff --git a/src/box/tuple.h b/src/box/tuple.h
index 6ebedf5..d8138e2 100644
--- a/src/box/tuple.h
+++ b/src/box/tuple.h
@@ -445,7 +445,7 @@ tuple_delete(struct tuple *tuple)
 	say_debug("%s(%p)", __func__, tuple);
 	assert(tuple->refs == 0);
 	struct tuple_format *format = tuple_format(tuple);
-	format->vtab.destroy(format, tuple);
+	format->vtab.tuple_delete(format, tuple);
 }
 
 /**
diff --git a/src/box/tuple_format.h b/src/box/tuple_format.h
index d35182d..cc727ec 100644
--- a/src/box/tuple_format.h
+++ b/src/box/tuple_format.h
@@ -68,7 +68,10 @@ struct tuple_format;
 struct tuple_format_vtab {
 	/** Free allocated tuple using engine-specific memory allocator. */
 	void
-	(*destroy)(struct tuple_format *format, struct tuple *tuple);
+	(*tuple_delete)(struct tuple_format *format, struct tuple *tuple);
+	/** Allocates a new tuple on the same allocator and with the same format. */
+	struct tuple*
+	(*tuple_new)(struct tuple_format *format, const char *data, const char *end);
 };
 
 /** Tuple field meta information for tuple_format. */
diff --git a/src/box/vy_stmt.c b/src/box/vy_stmt.c
index 84182e7..3135c58 100644
--- a/src/box/vy_stmt.c
+++ b/src/box/vy_stmt.c
@@ -47,6 +47,7 @@
 
 struct tuple_format_vtab vy_tuple_format_vtab = {
 	vy_tuple_delete,
+	vy_tuple_new,
 };
 
 size_t vy_max_tuple_size = 1024 * 1024;
@@ -69,6 +70,12 @@ vy_tuple_delete(struct tuple_format *format, struct tuple *tuple)
 	free(tuple);
 }
 
+struct tuple *
+vy_tuple_new(struct tuple_format *format, const char *data, const char *end)
+{
+	return vy_stmt_new_insert(format, data, end);
+}
+
 /**
  * Allocate a vinyl statement object on base of the struct tuple
  * with malloc() and the reference counter equal to 1.
diff --git a/src/box/vy_stmt.h b/src/box/vy_stmt.h
index a33739d..efc99c8 100644
--- a/src/box/vy_stmt.h
+++ b/src/box/vy_stmt.h
@@ -206,6 +206,15 @@ void
 vy_tuple_delete(struct tuple_format *format, struct tuple *tuple);
 
 /**
+ * Create the tuple of a vinyl space.
+ * @param format Target tuple format.
+ * @param data Target tuple data.
+ * @param size Size of the data.
+ */
+struct tuple *
+vy_tuple_new(struct tuple_format *format, const char *data, const char *end);
+
+/**
  * Duplicate the statememnt.
  *
  * @param stmt statement
diff --git a/test/box/update.result b/test/box/update.result
index a3f731b..7b9e295 100644
--- a/test/box/update.result
+++ b/test/box/update.result
@@ -849,3 +849,121 @@ s:update(1, {{'=', 3, map}})
 s:drop()
 ---
 ...
+--
+-- gh-3051 Lost format while tuple update
+--
+format = {}
+---
+...
+format[1] = {name = 'KEY', type = 'unsigned'}
+---
+...
+format[2] = {name = 'VAL', type = 'string'}
+---
+...
+s = box.schema.create_space('tst_sample', {format = format})
+---
+...
+pk = s:create_index('pk')
+---
+...
+s:insert({1, 'sss'})
+---
+- [1, 'sss']
+...
+aa = box.space.tst_sample:get(1)
+---
+...
+aa.VAL
+---
+- sss
+...
+aa = aa:update({{'=',2,'ssss'}})
+---
+...
+collectgarbage()
+---
+- 0
+...
+aa.VAL
+---
+- ssss
+...
+s:drop()
+---
+...
+aa = nil
+---
+...
+collectgarbage()
+---
+- 0
+...
+s = box.schema.space.create('tst_sample', {engine = 'vinyl', format = format})
+---
+...
+pk = s:create_index('pk')
+---
+...
+s:insert({1, 'sss'})
+---
+- [1, 'sss']
+...
+aa = box.space.tst_sample:get(1)
+---
+...
+aa.VAL
+---
+- sss
+...
+aa = aa:update({{'=',2,'ssss'}})
+---
+...
+collectgarbage()
+---
+- 0
+...
+aa.VAL
+---
+- ssss
+...
+s:drop()
+---
+...
+aa = nil
+---
+...
+collectgarbage()
+---
+- 0
+...
+-- test transform integrity
+space = box.schema.space.create('tweedledum')
+---
+...
+tmp = space:create_index('primary', { type = 'hash', parts = {1, 'string'}, unique = true })
+---
+...
+t = space:insert{'1', '2', '3', '4', '5', '6', '7'}
+---
+...
+t:transform(-1, 1)
+---
+- ['1', '2', '3', '4', '5', '6']
+...
+n = 2000
+---
+...
+tab = {}; for i=1,n,1 do table.insert(tab, i) end
+---
+...
+t = box.tuple.new(tab)
+---
+...
+t:transform(1, n - 1)
+---
+- [2000]
+...
+t = nil
+---
+...
diff --git a/test/box/update.test.lua b/test/box/update.test.lua
index c455bc1..1dc1751 100644
--- a/test/box/update.test.lua
+++ b/test/box/update.test.lua
@@ -269,3 +269,44 @@ t:update({{'=', 3, map}})
 s:update(1, {{'=', 3, map}})
 
 s:drop()
+
+--
+-- gh-3051 Lost format while tuple update
+--
+format = {}
+format[1] = {name = 'KEY', type = 'unsigned'}
+format[2] = {name = 'VAL', type = 'string'}
+s = box.schema.create_space('tst_sample', {format = format})
+pk = s:create_index('pk')
+s:insert({1, 'sss'})
+aa = box.space.tst_sample:get(1)
+aa.VAL
+aa = aa:update({{'=',2,'ssss'}})
+collectgarbage()
+aa.VAL
+s:drop()
+aa = nil
+collectgarbage()
+
+s = box.schema.space.create('tst_sample', {engine = 'vinyl', format = format})
+pk = s:create_index('pk')
+s:insert({1, 'sss'})
+aa = box.space.tst_sample:get(1)
+aa.VAL
+aa = aa:update({{'=',2,'ssss'}})
+collectgarbage()
+aa.VAL
+s:drop()
+aa = nil
+collectgarbage()
+
+-- test transform integrity
+space = box.schema.space.create('tweedledum')
+tmp = space:create_index('primary', { type = 'hash', parts = {1, 'string'}, unique = true })
+t = space:insert{'1', '2', '3', '4', '5', '6', '7'}
+t:transform(-1, 1)
+n = 2000
+tab = {}; for i=1,n,1 do table.insert(tab, i) end
+t = box.tuple.new(tab)
+t:transform(1, n - 1)
+t = nil
-- 
2.7.4