[Tarantool-patches] [PATCH v3 04/13] txm: introduce dirty tuples

[Aleksandr Lyapunov]

If the tuple is marked as dirty that usually means that it was
somehow affected by a transaction. If a tuple found in index is
dirty - we cannot immediately return to user, instead we must
clarify it in transaction manager.

Part of #4897
---
 src/box/memtx_engine.c              |  5 +++--
 src/box/tuple.c                     |  5 +++--
 src/box/tuple.h                     | 12 ++++++++++--
 src/box/tuple_format.c              |  4 ++--
 src/box/vy_stmt.c                   |  5 +++--
 test/box/huge_field_map.result      |  2 +-
 test/box/huge_field_map_long.result |  2 +-
 7 files changed, 23 insertions(+), 12 deletions(-)

diff --git a/src/box/memtx_engine.c b/src/box/memtx_engine.c
index b5b6b14..dfd6fce 100644
--- a/src/box/memtx_engine.c
+++ b/src/box/memtx_engine.c
@@ -1131,8 +1131,8 @@ memtx_tuple_new(struct tuple_format *format, const char *data, const char *end)
 	 * tuple is not the first field of the memtx_tuple.
 	 */
 	uint32_t data_offset = sizeof(struct tuple) + field_map_size;
-	if (data_offset > UINT16_MAX) {
-		/** tuple->data_offset is 16 bits */
+	if (data_offset > INT16_MAX) {
+		/** tuple->data_offset is 15 bits */
 		diag_set(ClientError, ER_TUPLE_METADATA_IS_TOO_BIG,
 			 data_offset);
 		goto end;
@@ -1170,6 +1170,7 @@ memtx_tuple_new(struct tuple_format *format, const char *data, const char *end)
 	tuple->format_id = tuple_format_id(format);
 	tuple_format_ref(format);
 	tuple->data_offset = data_offset;
+	tuple->is_dirty = false;
 	char *raw = (char *) tuple + tuple->data_offset;
 	field_map_build(&builder, raw - field_map_size);
 	memcpy(raw, data, tuple_len);
diff --git a/src/box/tuple.c b/src/box/tuple.c
index e48ee08..9f0f24c 100644
--- a/src/box/tuple.c
+++ b/src/box/tuple.c
@@ -84,8 +84,8 @@ runtime_tuple_new(struct tuple_format *format, const char *data, const char *end
 		goto end;
 	uint32_t field_map_size = field_map_build_size(&builder);
 	uint32_t data_offset = sizeof(struct tuple) + field_map_size;
-	if (data_offset > UINT16_MAX) {
-		/** tuple->data_offset is 16 bits */
+	if (data_offset > INT16_MAX) {
+		/** tuple->data_offset is 15 bits */
 		diag_set(ClientError, ER_TUPLE_METADATA_IS_TOO_BIG,
 			 data_offset);
 		goto end;
@@ -105,6 +105,7 @@ runtime_tuple_new(struct tuple_format *format, const char *data, const char *end
 	tuple->format_id = tuple_format_id(format);
 	tuple_format_ref(format);
 	tuple->data_offset = data_offset;
+	tuple->is_dirty = false;
 	char *raw = (char *) tuple + data_offset;
 	field_map_build(&builder, raw - field_map_size);
 	memcpy(raw, data, data_len);
diff --git a/src/box/tuple.h b/src/box/tuple.h
index 9a88772..4752323 100644
--- a/src/box/tuple.h
+++ b/src/box/tuple.h
@@ -319,7 +319,13 @@ struct PACKED tuple
 	/**
 	 * Offset to the MessagePack from the begin of the tuple.
 	 */
-	uint16_t data_offset;
+	uint16_t data_offset : 15;
+	/**
+	 * The tuple (if it's found in index for example) could be invisible
+	 * for current transactions. The flag means that the tuple must
+	 * be clarified by transaction engine.
+	 */
+	bool is_dirty : 1;
 	/**
 	 * Engine specific fields and offsets array concatenated
 	 * with MessagePack fields array.
@@ -1081,8 +1087,10 @@ tuple_unref(struct tuple *tuple)
 	assert(tuple->refs - 1 >= 0);
 	if (unlikely(tuple->is_bigref))
 		tuple_unref_slow(tuple);
-	else if (--tuple->refs == 0)
+	else if (--tuple->refs == 0) {
+		assert(!tuple->is_dirty);
 		tuple_delete(tuple);
+	}
 }
 
 extern struct tuple *box_tuple_last;
diff --git a/src/box/tuple_format.c b/src/box/tuple_format.c
index faf038a..bae6c67 100644
--- a/src/box/tuple_format.c
+++ b/src/box/tuple_format.c
@@ -501,8 +501,8 @@ tuple_format_create(struct tuple_format *format, struct key_def * const *keys,
 	assert(tuple_format_field(format, 0)->offset_slot == TUPLE_OFFSET_SLOT_NIL
 	       || json_token_is_multikey(&tuple_format_field(format, 0)->token));
 	size_t field_map_size = -current_slot * sizeof(uint32_t);
-	if (field_map_size > UINT16_MAX) {
-		/** tuple->data_offset is 16 bits */
+	if (field_map_size > INT16_MAX) {
+		/** tuple->data_offset is 15 bits */
 		diag_set(ClientError, ER_INDEX_FIELD_COUNT_LIMIT,
 			 -current_slot);
 		return -1;
diff --git a/src/box/vy_stmt.c b/src/box/vy_stmt.c
index f59c418..92e0aa1 100644
--- a/src/box/vy_stmt.c
+++ b/src/box/vy_stmt.c
@@ -160,8 +160,8 @@ vy_stmt_alloc(struct tuple_format *format, uint32_t data_offset, uint32_t bsize)
 {
 	assert(data_offset >= sizeof(struct vy_stmt) + format->field_map_size);
 
-	if (data_offset > UINT16_MAX) {
-		/** tuple->data_offset is 16 bits */
+	if (data_offset > INT16_MAX) {
+		/** tuple->data_offset is 15 bits */
 		diag_set(ClientError, ER_TUPLE_METADATA_IS_TOO_BIG,
 			 data_offset);
 		return NULL;
@@ -198,6 +198,7 @@ vy_stmt_alloc(struct tuple_format *format, uint32_t data_offset, uint32_t bsize)
 		tuple_format_ref(format);
 	tuple->bsize = bsize;
 	tuple->data_offset = data_offset;
+	tuple->is_dirty = false;
 	vy_stmt_set_lsn(tuple, 0);
 	vy_stmt_set_type(tuple, 0);
 	vy_stmt_set_flags(tuple, 0);
diff --git a/test/box/huge_field_map.result b/test/box/huge_field_map.result
index 11b4da3..45022cc 100644
--- a/test/box/huge_field_map.result
+++ b/test/box/huge_field_map.result
@@ -38,7 +38,7 @@ test_run:cmd("setopt delimiter ''");
 pcall(test) -- must fail but not crash
  | ---
  | - false
- | - 'Can''t create tuple: metadata size 65558 is too big'
+ | - 'Can''t create tuple: metadata size 32790 is too big'
  | ...
 
 test = nil
diff --git a/test/box/huge_field_map_long.result b/test/box/huge_field_map_long.result
index d7971ae..cb47900 100644
--- a/test/box/huge_field_map_long.result
+++ b/test/box/huge_field_map_long.result
@@ -40,7 +40,7 @@ test_run:cmd("setopt delimiter ''");
 pcall(test) -- must fail but not crash
  | ---
  | - false
- | - 'Can''t create tuple: metadata size 65542 is too big'
+ | - 'Can''t create tuple: metadata size 32774 is too big'
  | ...
 
 test = nil
-- 
2.7.4