[tarantool-patches] [PATCH v2 1/7] box: add space address to index_replace

[tarantool-patches] [PATCH v2 1/7] box: add space address to index_replace

[imeevma]

On error in some cases index_replace
looking for space using space_id that
is saved in index. It is wrong as
function that calls index_replace
already have that space. Also in case
of ephemeral spaces space_cache_find
doesn't work right as all ephemeral
space have id == 0.

Part of #3375.
---
Branch: https://github.com/tarantool/tarantool/compare/imeevma/gh-3375-lua-expose-ephemeral-spaces
Issue: https://github.com/tarantool/tarantool/issues/3375

 src/box/index.cc       | 16 +++++++++-------
 src/box/index.h        | 30 +++++++++++++++++-------------
 src/box/memtx_bitset.c |  7 ++++---
 src/box/memtx_engine.c |  6 +++---
 src/box/memtx_hash.c   | 13 +++++--------
 src/box/memtx_rtree.c  |  7 ++++---
 src/box/memtx_space.c  | 13 +++++++------
 src/box/memtx_tree.c   | 17 ++++++++---------
 8 files changed, 57 insertions(+), 52 deletions(-)

diff --git a/src/box/index.cc b/src/box/index.cc
index 188995e..94dcf4b 100644
--- a/src/box/index.cc
+++ b/src/box/index.cc
@@ -510,7 +510,7 @@ index_delete(struct index *index)
 }
 
 int
-index_build(struct index *index, struct index *pk)
+index_build(struct index *index, struct space *space, struct index *pk)
 {
 	ssize_t n_tuples = index_size(pk);
 	if (n_tuples < 0)
@@ -539,7 +539,7 @@ index_build(struct index *index, struct index *pk)
 			break;
 		if (tuple == NULL)
 			break;
-		rc = index_build_next(index, tuple);
+		rc = index_build_next(index, space, tuple);
 		if (rc != 0)
 			break;
 	}
@@ -658,10 +658,11 @@ generic_index_get(struct index *index, const char *key,
 }
 
 int
-generic_index_replace(struct index *index, struct tuple *old_tuple,
-		      struct tuple *new_tuple, enum dup_replace_mode mode,
-		      struct tuple **result)
+generic_index_replace(struct index *index, struct space *space,
+		      struct tuple *old_tuple, struct tuple *new_tuple,
+		      enum dup_replace_mode mode, struct tuple **result)
 {
+	(void)space;
 	(void)old_tuple;
 	(void)new_tuple;
 	(void)mode;
@@ -709,10 +710,11 @@ generic_index_reserve(struct index *, uint32_t)
 }
 
 int
-generic_index_build_next(struct index *index, struct tuple *tuple)
+generic_index_build_next(struct index *index, struct space *space,
+			 struct tuple *tuple)
 {
 	struct tuple *unused;
-	return index_replace(index, NULL, tuple, DUP_INSERT, &unused);
+	return index_replace(index, space, NULL, tuple, DUP_INSERT, &unused);
 }
 
 void
diff --git a/src/box/index.h b/src/box/index.h
index 686e7a1..2b5357a 100644
--- a/src/box/index.h
+++ b/src/box/index.h
@@ -41,6 +41,7 @@ extern "C" {
 
 struct tuple;
 struct engine;
+struct space;
 struct index;
 struct index_def;
 struct key_def;
@@ -411,9 +412,9 @@ struct index_vtab {
 			 const char *key, uint32_t part_count);
 	int (*get)(struct index *index, const char *key,
 		   uint32_t part_count, struct tuple **result);
-	int (*replace)(struct index *index, struct tuple *old_tuple,
-		       struct tuple *new_tuple, enum dup_replace_mode mode,
-		       struct tuple **result);
+	int (*replace)(struct index *index, struct space *space,
+		       struct tuple *old_tuple, struct tuple *new_tuple,
+		       enum dup_replace_mode mode, struct tuple **result);
 	/** Create an index iterator. */
 	struct iterator *(*create_iterator)(struct index *index,
 			enum iterator_type type,
@@ -443,7 +444,8 @@ struct index_vtab {
 	 * begin_build().
 	 */
 	int (*reserve)(struct index *index, uint32_t size_hint);
-	int (*build_next)(struct index *index, struct tuple *tuple);
+	int (*build_next)(struct index *index, struct space *space,
+			  struct tuple *tuple);
 	void (*end_build)(struct index *index);
 };
 
@@ -504,7 +506,7 @@ index_delete(struct index *index);
 
 /** Build this index based on the contents of another index. */
 int
-index_build(struct index *index, struct index *pk);
+index_build(struct index *index, struct space *space, struct index *pk);
 
 static inline void
 index_commit_create(struct index *index, int64_t signature)
@@ -596,11 +598,12 @@ index_get(struct index *index, const char *key,
 }
 
 static inline int
-index_replace(struct index *index, struct tuple *old_tuple,
-	      struct tuple *new_tuple, enum dup_replace_mode mode,
-	      struct tuple **result)
+index_replace(struct index *index, struct space *space,
+	      struct tuple *old_tuple, struct tuple *new_tuple,
+	      enum dup_replace_mode mode, struct tuple **result)
 {
-	return index->vtab->replace(index, old_tuple, new_tuple, mode, result);
+	return index->vtab->replace(index, space, old_tuple, new_tuple, mode,
+				    result);
 }
 
 static inline struct iterator *
@@ -647,9 +650,9 @@ index_reserve(struct index *index, uint32_t size_hint)
 }
 
 static inline int
-index_build_next(struct index *index, struct tuple *tuple)
+index_build_next(struct index *index, struct space *space, struct tuple *tuple)
 {
-	return index->vtab->build_next(index, tuple);
+	return index->vtab->build_next(index, space, tuple);
 }
 
 static inline void
@@ -674,7 +677,8 @@ int generic_index_random(struct index *, uint32_t, struct tuple **);
 ssize_t generic_index_count(struct index *, enum iterator_type,
 			    const char *, uint32_t);
 int generic_index_get(struct index *, const char *, uint32_t, struct tuple **);
-int generic_index_replace(struct index *, struct tuple *, struct tuple *,
+int generic_index_replace(struct index *, struct space *,
+			  struct tuple *, struct tuple *,
 			  enum dup_replace_mode, struct tuple **);
 struct snapshot_iterator *generic_index_create_snapshot_iterator(struct index *);
 void generic_index_stat(struct index *, struct info_handler *);
@@ -682,7 +686,7 @@ void generic_index_compact(struct index *);
 void generic_index_reset_stat(struct index *);
 void generic_index_begin_build(struct index *);
 int generic_index_reserve(struct index *, uint32_t);
-int generic_index_build_next(struct index *, struct tuple *);
+int generic_index_build_next(struct index *, struct space *, struct tuple *);
 void generic_index_end_build(struct index *);
 
 #if defined(__cplusplus)
diff --git a/src/box/memtx_bitset.c b/src/box/memtx_bitset.c
index a665f1a..ada6fa8 100644
--- a/src/box/memtx_bitset.c
+++ b/src/box/memtx_bitset.c
@@ -252,10 +252,11 @@ make_key(const char *field, uint32_t *key_len)
 }
 
 static int
-memtx_bitset_index_replace(struct index *base, struct tuple *old_tuple,
-			   struct tuple *new_tuple, enum dup_replace_mode mode,
-			   struct tuple **result)
+memtx_bitset_index_replace(struct index *base, struct space *space,
+			   struct tuple *old_tuple, struct tuple *new_tuple,
+			   enum dup_replace_mode mode, struct tuple **result)
 {
+	(void)space;
 	struct memtx_bitset_index *index = (struct memtx_bitset_index *)base;
 
 	assert(!base->def->opts.is_unique);
diff --git a/src/box/memtx_engine.c b/src/box/memtx_engine.c
index c210afb..addc477 100644
--- a/src/box/memtx_engine.c
+++ b/src/box/memtx_engine.c
@@ -160,7 +160,7 @@ memtx_build_secondary_keys(struct space *space, void *param)
 		}
 
 		for (uint32_t j = 1; j < space->index_count; j++) {
-			if (index_build(space->index[j], pk) < 0)
+			if (index_build(space->index[j], space, pk) < 0)
 				return -1;
 		}
 
@@ -448,8 +448,8 @@ memtx_engine_rollback_statement(struct engine *engine, struct txn *txn,
 		struct tuple *unused;
 		struct index *index = space->index[i];
 		/* Rollback must not fail. */
-		if (index_replace(index, stmt->new_tuple, stmt->old_tuple,
-				  DUP_INSERT, &unused) != 0) {
+		if (index_replace(index, space, stmt->new_tuple,
+				  stmt->old_tuple, DUP_INSERT, &unused) != 0) {
 			diag_log();
 			unreachable();
 			panic("failed to rollback change");
diff --git a/src/box/memtx_hash.c b/src/box/memtx_hash.c
index eae07e8..0f70229 100644
--- a/src/box/memtx_hash.c
+++ b/src/box/memtx_hash.c
@@ -35,7 +35,6 @@
 #include "tuple_hash.h"
 #include "memtx_engine.h"
 #include "space.h"
-#include "schema.h" /* space_cache_find() */
 #include "errinj.h"
 
 #include <small/mempool.h>
@@ -244,9 +243,9 @@ memtx_hash_index_get(struct index *base, const char *key,
 }
 
 static int
-memtx_hash_index_replace(struct index *base, struct tuple *old_tuple,
-			 struct tuple *new_tuple, enum dup_replace_mode mode,
-			 struct tuple **result)
+memtx_hash_index_replace(struct index *base, struct space *space,
+			 struct tuple *old_tuple, struct tuple *new_tuple,
+			 enum dup_replace_mode mode, struct tuple **result)
 {
 	struct memtx_hash_index *index = (struct memtx_hash_index *)base;
 	struct light_index_core *hash_table = &index->hash_table;
@@ -281,10 +280,8 @@ memtx_hash_index_replace(struct index *base, struct tuple *old_tuple,
 					      "recover of int hash_table");
 				}
 			}
-			struct space *sp = space_cache_find(base->def->space_id);
-			if (sp != NULL)
-				diag_set(ClientError, errcode, base->def->name,
-					 space_name(sp));
+			diag_set(ClientError, errcode, base->def->name,
+				 space_name(space));
 			return -1;
 		}
 
diff --git a/src/box/memtx_rtree.c b/src/box/memtx_rtree.c
index 0b12cda..333d191 100644
--- a/src/box/memtx_rtree.c
+++ b/src/box/memtx_rtree.c
@@ -210,10 +210,11 @@ memtx_rtree_index_get(struct index *base, const char *key,
 }
 
 static int
-memtx_rtree_index_replace(struct index *base, struct tuple *old_tuple,
-			  struct tuple *new_tuple, enum dup_replace_mode mode,
-			  struct tuple **result)
+memtx_rtree_index_replace(struct index *base, struct space *space,
+			  struct tuple *old_tuple, struct tuple *new_tuple,
+			  enum dup_replace_mode mode, struct tuple **result)
 {
+	(void)space;
 	(void)mode;
 	struct memtx_rtree_index *index = (struct memtx_rtree_index *)base;
 	struct rtree_rect rect;
diff --git a/src/box/memtx_space.c b/src/box/memtx_space.c
index b94c4ab..185be92 100644
--- a/src/box/memtx_space.c
+++ b/src/box/memtx_space.c
@@ -124,7 +124,7 @@ memtx_space_replace_build_next(struct space *space, struct tuple *old_tuple,
 		panic("Failed to commit transaction when loading "
 		      "from snapshot");
 	}
-	if (index_build_next(space->index[0], new_tuple) != 0)
+	if (index_build_next(space->index[0], space, new_tuple) != 0)
 		return -1;
 	memtx_space_update_bsize(space, NULL, new_tuple);
 	return 0;
@@ -140,7 +140,7 @@ memtx_space_replace_primary_key(struct space *space, struct tuple *old_tuple,
 				enum dup_replace_mode mode,
 				struct tuple **result)
 {
-	if (index_replace(space->index[0], old_tuple,
+	if (index_replace(space->index[0], space, old_tuple,
 			  new_tuple, mode, &old_tuple) != 0)
 		return -1;
 	memtx_space_update_bsize(space, old_tuple, new_tuple);
@@ -259,7 +259,8 @@ memtx_space_replace_all_keys(struct space *space, struct tuple *old_tuple,
 	 * If old_tuple is not NULL, the index has to
 	 * find and delete it, or return an error.
 	 */
-	if (index_replace(pk, old_tuple, new_tuple, mode, &old_tuple) != 0)
+	if (index_replace(pk, space, old_tuple,
+			  new_tuple, mode, &old_tuple) != 0)
 		return -1;
 	assert(old_tuple || new_tuple);
 
@@ -267,7 +268,7 @@ memtx_space_replace_all_keys(struct space *space, struct tuple *old_tuple,
 	for (i++; i < space->index_count; i++) {
 		struct tuple *unused;
 		struct index *index = space->index[i];
-		if (index_replace(index, old_tuple, new_tuple,
+		if (index_replace(index, space, old_tuple, new_tuple,
 				  DUP_INSERT, &unused) != 0)
 			goto rollback;
 	}
@@ -281,7 +282,7 @@ rollback:
 		struct tuple *unused;
 		struct index *index = space->index[i - 1];
 		/* Rollback must not fail. */
-		if (index_replace(index, new_tuple, old_tuple,
+		if (index_replace(index, space, new_tuple, old_tuple,
 				  DUP_INSERT, &unused) != 0) {
 			diag_log();
 			unreachable();
@@ -898,7 +899,7 @@ memtx_space_build_index(struct space *src_space, struct index *new_index,
 		 * @todo: better message if there is a duplicate.
 		 */
 		struct tuple *old_tuple;
-		rc = index_replace(new_index, NULL, tuple,
+		rc = index_replace(new_index, src_space, NULL, tuple,
 				   DUP_INSERT, &old_tuple);
 		if (rc != 0)
 			break;
diff --git a/src/box/memtx_tree.c b/src/box/memtx_tree.c
index f851fb8..cf5d14f 100644
--- a/src/box/memtx_tree.c
+++ b/src/box/memtx_tree.c
@@ -31,7 +31,6 @@
 #include "memtx_tree.h"
 #include "memtx_engine.h"
 #include "space.h"
-#include "schema.h" /* space_cache_find() */
 #include "errinj.h"
 #include "memory.h"
 #include "fiber.h"
@@ -438,9 +437,9 @@ memtx_tree_index_get(struct index *base, const char *key,
 }
 
 static int
-memtx_tree_index_replace(struct index *base, struct tuple *old_tuple,
-			 struct tuple *new_tuple, enum dup_replace_mode mode,
-			 struct tuple **result)
+memtx_tree_index_replace(struct index *base, struct space *space,
+			 struct tuple *old_tuple, struct tuple *new_tuple,
+			 enum dup_replace_mode mode, struct tuple **result)
 {
 	struct memtx_tree_index *index = (struct memtx_tree_index *)base;
 	if (new_tuple) {
@@ -461,10 +460,8 @@ memtx_tree_index_replace(struct index *base, struct tuple *old_tuple,
 			memtx_tree_delete(&index->tree, new_tuple);
 			if (dup_tuple)
 				memtx_tree_insert(&index->tree, dup_tuple, 0);
-			struct space *sp = space_cache_find(base->def->space_id);
-			if (sp != NULL)
-				diag_set(ClientError, errcode, base->def->name,
-					 space_name(sp));
+			diag_set(ClientError, errcode, base->def->name,
+				 space_name(space));
 			return -1;
 		}
 		if (dup_tuple) {
@@ -549,8 +546,10 @@ memtx_tree_index_reserve(struct index *base, uint32_t size_hint)
 }
 
 static int
-memtx_tree_index_build_next(struct index *base, struct tuple *tuple)
+memtx_tree_index_build_next(struct index *base, struct space *space,
+			    struct tuple *tuple)
 {
+	(void)space;
 	struct memtx_tree_index *index = (struct memtx_tree_index *)base;
 	if (index->build_array == NULL) {
 		index->build_array = (struct tuple **)malloc(MEMTX_EXTENT_SIZE);
-- 
2.7.4

[tarantool-patches] Re: [PATCH v2 1/7] box: add space address to index_replace

[Vladislav Shpilevoy]

Kirill, please, do the second review.

On 17/07/2018 16:08, imeevma@tarantool.org wrote:
> On error in some cases index_replace
> looking for space using space_id that
> is saved in index. It is wrong as
> function that calls index_replace
> already have that space. Also in case
> of ephemeral spaces space_cache_find
> doesn't work right as all ephemeral
> space have id == 0.
> 
> Part of #3375.
> ---
> Branch: https://github.com/tarantool/tarantool/compare/imeevma/gh-3375-lua-expose-ephemeral-spaces
> Issue: https://github.com/tarantool/tarantool/issues/3375
> 
>   src/box/index.cc       | 16 +++++++++-------
>   src/box/index.h        | 30 +++++++++++++++++-------------
>   src/box/memtx_bitset.c |  7 ++++---
>   src/box/memtx_engine.c |  6 +++---
>   src/box/memtx_hash.c   | 13 +++++--------
>   src/box/memtx_rtree.c  |  7 ++++---
>   src/box/memtx_space.c  | 13 +++++++------
>   src/box/memtx_tree.c   | 17 ++++++++---------
>   8 files changed, 57 insertions(+), 52 deletions(-)
> 
> diff --git a/src/box/index.cc b/src/box/index.cc
> index 188995e..94dcf4b 100644
> --- a/src/box/index.cc
> +++ b/src/box/index.cc
> @@ -510,7 +510,7 @@ index_delete(struct index *index)
>   }
>   
>   int
> -index_build(struct index *index, struct index *pk)
> +index_build(struct index *index, struct space *space, struct index *pk)
>   {
>   	ssize_t n_tuples = index_size(pk);
>   	if (n_tuples < 0)
> @@ -539,7 +539,7 @@ index_build(struct index *index, struct index *pk)
>   			break;
>   		if (tuple == NULL)
>   			break;
> -		rc = index_build_next(index, tuple);
> +		rc = index_build_next(index, space, tuple);
>   		if (rc != 0)
>   			break;
>   	}
> @@ -658,10 +658,11 @@ generic_index_get(struct index *index, const char *key,
>   }
>   
>   int
> -generic_index_replace(struct index *index, struct tuple *old_tuple,
> -		      struct tuple *new_tuple, enum dup_replace_mode mode,
> -		      struct tuple **result)
> +generic_index_replace(struct index *index, struct space *space,
> +		      struct tuple *old_tuple, struct tuple *new_tuple,
> +		      enum dup_replace_mode mode, struct tuple **result)
>   {
> +	(void)space;
>   	(void)old_tuple;
>   	(void)new_tuple;
>   	(void)mode;
> @@ -709,10 +710,11 @@ generic_index_reserve(struct index *, uint32_t)
>   }
>   
>   int
> -generic_index_build_next(struct index *index, struct tuple *tuple)
> +generic_index_build_next(struct index *index, struct space *space,
> +			 struct tuple *tuple)
>   {
>   	struct tuple *unused;
> -	return index_replace(index, NULL, tuple, DUP_INSERT, &unused);
> +	return index_replace(index, space, NULL, tuple, DUP_INSERT, &unused);
>   }
>   
>   void
> diff --git a/src/box/index.h b/src/box/index.h
> index 686e7a1..2b5357a 100644
> --- a/src/box/index.h
> +++ b/src/box/index.h
> @@ -41,6 +41,7 @@ extern "C" {
>   
>   struct tuple;
>   struct engine;
> +struct space;
>   struct index;
>   struct index_def;
>   struct key_def;
> @@ -411,9 +412,9 @@ struct index_vtab {
>   			 const char *key, uint32_t part_count);
>   	int (*get)(struct index *index, const char *key,
>   		   uint32_t part_count, struct tuple **result);
> -	int (*replace)(struct index *index, struct tuple *old_tuple,
> -		       struct tuple *new_tuple, enum dup_replace_mode mode,
> -		       struct tuple **result);
> +	int (*replace)(struct index *index, struct space *space,
> +		       struct tuple *old_tuple, struct tuple *new_tuple,
> +		       enum dup_replace_mode mode, struct tuple **result);
>   	/** Create an index iterator. */
>   	struct iterator *(*create_iterator)(struct index *index,
>   			enum iterator_type type,
> @@ -443,7 +444,8 @@ struct index_vtab {
>   	 * begin_build().
>   	 */
>   	int (*reserve)(struct index *index, uint32_t size_hint);
> -	int (*build_next)(struct index *index, struct tuple *tuple);
> +	int (*build_next)(struct index *index, struct space *space,
> +			  struct tuple *tuple);
>   	void (*end_build)(struct index *index);
>   };
>   
> @@ -504,7 +506,7 @@ index_delete(struct index *index);
>   
>   /** Build this index based on the contents of another index. */
>   int
> -index_build(struct index *index, struct index *pk);
> +index_build(struct index *index, struct space *space, struct index *pk);
>   
>   static inline void
>   index_commit_create(struct index *index, int64_t signature)
> @@ -596,11 +598,12 @@ index_get(struct index *index, const char *key,
>   }
>   
>   static inline int
> -index_replace(struct index *index, struct tuple *old_tuple,
> -	      struct tuple *new_tuple, enum dup_replace_mode mode,
> -	      struct tuple **result)
> +index_replace(struct index *index, struct space *space,
> +	      struct tuple *old_tuple, struct tuple *new_tuple,
> +	      enum dup_replace_mode mode, struct tuple **result)
>   {
> -	return index->vtab->replace(index, old_tuple, new_tuple, mode, result);
> +	return index->vtab->replace(index, space, old_tuple, new_tuple, mode,
> +				    result);
>   }
>   
>   static inline struct iterator *
> @@ -647,9 +650,9 @@ index_reserve(struct index *index, uint32_t size_hint)
>   }
>   
>   static inline int
> -index_build_next(struct index *index, struct tuple *tuple)
> +index_build_next(struct index *index, struct space *space, struct tuple *tuple)
>   {
> -	return index->vtab->build_next(index, tuple);
> +	return index->vtab->build_next(index, space, tuple);
>   }
>   
>   static inline void
> @@ -674,7 +677,8 @@ int generic_index_random(struct index *, uint32_t, struct tuple **);
>   ssize_t generic_index_count(struct index *, enum iterator_type,
>   			    const char *, uint32_t);
>   int generic_index_get(struct index *, const char *, uint32_t, struct tuple **);
> -int generic_index_replace(struct index *, struct tuple *, struct tuple *,
> +int generic_index_replace(struct index *, struct space *,
> +			  struct tuple *, struct tuple *,
>   			  enum dup_replace_mode, struct tuple **);
>   struct snapshot_iterator *generic_index_create_snapshot_iterator(struct index *);
>   void generic_index_stat(struct index *, struct info_handler *);
> @@ -682,7 +686,7 @@ void generic_index_compact(struct index *);
>   void generic_index_reset_stat(struct index *);
>   void generic_index_begin_build(struct index *);
>   int generic_index_reserve(struct index *, uint32_t);
> -int generic_index_build_next(struct index *, struct tuple *);
> +int generic_index_build_next(struct index *, struct space *, struct tuple *);
>   void generic_index_end_build(struct index *);
>   
>   #if defined(__cplusplus)
> diff --git a/src/box/memtx_bitset.c b/src/box/memtx_bitset.c
> index a665f1a..ada6fa8 100644
> --- a/src/box/memtx_bitset.c
> +++ b/src/box/memtx_bitset.c
> @@ -252,10 +252,11 @@ make_key(const char *field, uint32_t *key_len)
>   }
>   
>   static int
> -memtx_bitset_index_replace(struct index *base, struct tuple *old_tuple,
> -			   struct tuple *new_tuple, enum dup_replace_mode mode,
> -			   struct tuple **result)
> +memtx_bitset_index_replace(struct index *base, struct space *space,
> +			   struct tuple *old_tuple, struct tuple *new_tuple,
> +			   enum dup_replace_mode mode, struct tuple **result)
>   {
> +	(void)space;
>   	struct memtx_bitset_index *index = (struct memtx_bitset_index *)base;
>   
>   	assert(!base->def->opts.is_unique);
> diff --git a/src/box/memtx_engine.c b/src/box/memtx_engine.c
> index c210afb..addc477 100644
> --- a/src/box/memtx_engine.c
> +++ b/src/box/memtx_engine.c
> @@ -160,7 +160,7 @@ memtx_build_secondary_keys(struct space *space, void *param)
>   		}
>   
>   		for (uint32_t j = 1; j < space->index_count; j++) {
> -			if (index_build(space->index[j], pk) < 0)
> +			if (index_build(space->index[j], space, pk) < 0)
>   				return -1;
>   		}
>   
> @@ -448,8 +448,8 @@ memtx_engine_rollback_statement(struct engine *engine, struct txn *txn,
>   		struct tuple *unused;
>   		struct index *index = space->index[i];
>   		/* Rollback must not fail. */
> -		if (index_replace(index, stmt->new_tuple, stmt->old_tuple,
> -				  DUP_INSERT, &unused) != 0) {
> +		if (index_replace(index, space, stmt->new_tuple,
> +				  stmt->old_tuple, DUP_INSERT, &unused) != 0) {
>   			diag_log();
>   			unreachable();
>   			panic("failed to rollback change");
> diff --git a/src/box/memtx_hash.c b/src/box/memtx_hash.c
> index eae07e8..0f70229 100644
> --- a/src/box/memtx_hash.c
> +++ b/src/box/memtx_hash.c
> @@ -35,7 +35,6 @@
>   #include "tuple_hash.h"
>   #include "memtx_engine.h"
>   #include "space.h"
> -#include "schema.h" /* space_cache_find() */
>   #include "errinj.h"
>   
>   #include <small/mempool.h>
> @@ -244,9 +243,9 @@ memtx_hash_index_get(struct index *base, const char *key,
>   }
>   
>   static int
> -memtx_hash_index_replace(struct index *base, struct tuple *old_tuple,
> -			 struct tuple *new_tuple, enum dup_replace_mode mode,
> -			 struct tuple **result)
> +memtx_hash_index_replace(struct index *base, struct space *space,
> +			 struct tuple *old_tuple, struct tuple *new_tuple,
> +			 enum dup_replace_mode mode, struct tuple **result)
>   {
>   	struct memtx_hash_index *index = (struct memtx_hash_index *)base;
>   	struct light_index_core *hash_table = &index->hash_table;
> @@ -281,10 +280,8 @@ memtx_hash_index_replace(struct index *base, struct tuple *old_tuple,
>   					      "recover of int hash_table");
>   				}
>   			}
> -			struct space *sp = space_cache_find(base->def->space_id);
> -			if (sp != NULL)
> -				diag_set(ClientError, errcode, base->def->name,
> -					 space_name(sp));
> +			diag_set(ClientError, errcode, base->def->name,
> +				 space_name(space));
>   			return -1;
>   		}
>   
> diff --git a/src/box/memtx_rtree.c b/src/box/memtx_rtree.c
> index 0b12cda..333d191 100644
> --- a/src/box/memtx_rtree.c
> +++ b/src/box/memtx_rtree.c
> @@ -210,10 +210,11 @@ memtx_rtree_index_get(struct index *base, const char *key,
>   }
>   
>   static int
> -memtx_rtree_index_replace(struct index *base, struct tuple *old_tuple,
> -			  struct tuple *new_tuple, enum dup_replace_mode mode,
> -			  struct tuple **result)
> +memtx_rtree_index_replace(struct index *base, struct space *space,
> +			  struct tuple *old_tuple, struct tuple *new_tuple,
> +			  enum dup_replace_mode mode, struct tuple **result)
>   {
> +	(void)space;
>   	(void)mode;
>   	struct memtx_rtree_index *index = (struct memtx_rtree_index *)base;
>   	struct rtree_rect rect;
> diff --git a/src/box/memtx_space.c b/src/box/memtx_space.c
> index b94c4ab..185be92 100644
> --- a/src/box/memtx_space.c
> +++ b/src/box/memtx_space.c
> @@ -124,7 +124,7 @@ memtx_space_replace_build_next(struct space *space, struct tuple *old_tuple,
>   		panic("Failed to commit transaction when loading "
>   		      "from snapshot");
>   	}
> -	if (index_build_next(space->index[0], new_tuple) != 0)
> +	if (index_build_next(space->index[0], space, new_tuple) != 0)
>   		return -1;
>   	memtx_space_update_bsize(space, NULL, new_tuple);
>   	return 0;
> @@ -140,7 +140,7 @@ memtx_space_replace_primary_key(struct space *space, struct tuple *old_tuple,
>   				enum dup_replace_mode mode,
>   				struct tuple **result)
>   {
> -	if (index_replace(space->index[0], old_tuple,
> +	if (index_replace(space->index[0], space, old_tuple,
>   			  new_tuple, mode, &old_tuple) != 0)
>   		return -1;
>   	memtx_space_update_bsize(space, old_tuple, new_tuple);
> @@ -259,7 +259,8 @@ memtx_space_replace_all_keys(struct space *space, struct tuple *old_tuple,
>   	 * If old_tuple is not NULL, the index has to
>   	 * find and delete it, or return an error.
>   	 */
> -	if (index_replace(pk, old_tuple, new_tuple, mode, &old_tuple) != 0)
> +	if (index_replace(pk, space, old_tuple,
> +			  new_tuple, mode, &old_tuple) != 0)
>   		return -1;
>   	assert(old_tuple || new_tuple);
>   
> @@ -267,7 +268,7 @@ memtx_space_replace_all_keys(struct space *space, struct tuple *old_tuple,
>   	for (i++; i < space->index_count; i++) {
>   		struct tuple *unused;
>   		struct index *index = space->index[i];
> -		if (index_replace(index, old_tuple, new_tuple,
> +		if (index_replace(index, space, old_tuple, new_tuple,
>   				  DUP_INSERT, &unused) != 0)
>   			goto rollback;
>   	}
> @@ -281,7 +282,7 @@ rollback:
>   		struct tuple *unused;
>   		struct index *index = space->index[i - 1];
>   		/* Rollback must not fail. */
> -		if (index_replace(index, new_tuple, old_tuple,
> +		if (index_replace(index, space, new_tuple, old_tuple,
>   				  DUP_INSERT, &unused) != 0) {
>   			diag_log();
>   			unreachable();
> @@ -898,7 +899,7 @@ memtx_space_build_index(struct space *src_space, struct index *new_index,
>   		 * @todo: better message if there is a duplicate.
>   		 */
>   		struct tuple *old_tuple;
> -		rc = index_replace(new_index, NULL, tuple,
> +		rc = index_replace(new_index, src_space, NULL, tuple,
>   				   DUP_INSERT, &old_tuple);
>   		if (rc != 0)
>   			break;
> diff --git a/src/box/memtx_tree.c b/src/box/memtx_tree.c
> index f851fb8..cf5d14f 100644
> --- a/src/box/memtx_tree.c
> +++ b/src/box/memtx_tree.c
> @@ -31,7 +31,6 @@
>   #include "memtx_tree.h"
>   #include "memtx_engine.h"
>   #include "space.h"
> -#include "schema.h" /* space_cache_find() */
>   #include "errinj.h"
>   #include "memory.h"
>   #include "fiber.h"
> @@ -438,9 +437,9 @@ memtx_tree_index_get(struct index *base, const char *key,
>   }
>   
>   static int
> -memtx_tree_index_replace(struct index *base, struct tuple *old_tuple,
> -			 struct tuple *new_tuple, enum dup_replace_mode mode,
> -			 struct tuple **result)
> +memtx_tree_index_replace(struct index *base, struct space *space,
> +			 struct tuple *old_tuple, struct tuple *new_tuple,
> +			 enum dup_replace_mode mode, struct tuple **result)
>   {
>   	struct memtx_tree_index *index = (struct memtx_tree_index *)base;
>   	if (new_tuple) {
> @@ -461,10 +460,8 @@ memtx_tree_index_replace(struct index *base, struct tuple *old_tuple,
>   			memtx_tree_delete(&index->tree, new_tuple);
>   			if (dup_tuple)
>   				memtx_tree_insert(&index->tree, dup_tuple, 0);
> -			struct space *sp = space_cache_find(base->def->space_id);
> -			if (sp != NULL)
> -				diag_set(ClientError, errcode, base->def->name,
> -					 space_name(sp));
> +			diag_set(ClientError, errcode, base->def->name,
> +				 space_name(space));
>   			return -1;
>   		}
>   		if (dup_tuple) {
> @@ -549,8 +546,10 @@ memtx_tree_index_reserve(struct index *base, uint32_t size_hint)
>   }
>   
>   static int
> -memtx_tree_index_build_next(struct index *base, struct tuple *tuple)
> +memtx_tree_index_build_next(struct index *base, struct space *space,
> +			    struct tuple *tuple)
>   {
> +	(void)space;
>   	struct memtx_tree_index *index = (struct memtx_tree_index *)base;
>   	if (index->build_array == NULL) {
>   		index->build_array = (struct tuple **)malloc(MEMTX_EXTENT_SIZE);
> 

[tarantool-patches] [PATCH v2 2/7] box: move checks for key findability from space_vtab

[imeevma]

In this patch checks exact_key_validate
for memtx and vy_unique_key_validate
for vinyl were moved from space_vtab
to box.cc.

Part of #3375.
---
Branch: https://github.com/tarantool/tarantool/compare/imeevma/gh-3375-lua-expose-ephemeral-spaces
Issue: https://github.com/tarantool/tarantool/issues/3375

 src/box/box.cc        | 72 ++++++++++++++++++++++++++++++++++++++++++++-------
 src/box/memtx_space.c |  4 ---
 src/box/vinyl.c       | 24 ++---------------
 src/box/vinyl.h       | 26 +++++++++++++++++++
 4 files changed, 90 insertions(+), 36 deletions(-)

diff --git a/src/box/box.cc b/src/box/box.cc
index 7c6312d..1dcbfbf 100644
--- a/src/box/box.cc
+++ b/src/box/box.cc
@@ -993,16 +993,50 @@ box_index_id_by_name(uint32_t space_id, const char *name, uint32_t len)
 }
 /** \endcond public */
 
+/**
+ * Check space for writeability.
+ */
+static inline int
+space_check_writable(struct space *space)
+{
+	if (!space_is_temporary(space) &&
+	    space_group_id(space) != GROUP_LOCAL &&
+	    box_check_writable() != 0)
+		return -1;
+	return 0;
+}
+
+/**
+ * Check if key is good enough
+ * to be used to find tuple.
+ */
+static inline int
+key_check_findable(struct space *space, uint32_t index_id, const char *key)
+{
+	if (space_is_memtx(space)) {
+		struct index *pk = index_find_unique(space, index_id);
+		if (pk == NULL)
+			return -1;
+		uint32_t part_count = mp_decode_array(&key);
+		if (exact_key_validate(pk->def->key_def, key, part_count) != 0)
+			return -1;
+	} else if (space_is_vinyl(space)) {
+		struct vy_lsm *lsm = vy_lsm_find_unique(space, index_id);
+		if (lsm == NULL)
+			return -1;
+		uint32_t part_count = mp_decode_array(&key);
+		if (vy_unique_key_validate(lsm, key, part_count))
+			return -1;
+	}
+	return 0;
+}
+
 int
 box_process1(struct request *request, box_tuple_t **result)
 {
 	/* Allow to write to temporary spaces in read-only mode. */
 	struct space *space = space_cache_find(request->space_id);
-	if (space == NULL)
-		return -1;
-	if (!space_is_temporary(space) &&
-	    space_group_id(space) != GROUP_LOCAL &&
-	    box_check_writable() != 0)
+	if (space == NULL || space_check_writable(space) != 0)
 		return -1;
 	return box_process_rw(request, space, result);
 }
@@ -1087,13 +1121,16 @@ box_insert(uint32_t space_id, const char *tuple, const char *tuple_end,
 	   box_tuple_t **result)
 {
 	mp_tuple_assert(tuple, tuple_end);
+	struct space *space = space_cache_find(space_id);
+	if (space == NULL || space_check_writable(space) != 0)
+		return -1;
 	struct request request;
 	memset(&request, 0, sizeof(request));
 	request.type = IPROTO_INSERT;
 	request.space_id = space_id;
 	request.tuple = tuple;
 	request.tuple_end = tuple_end;
-	return box_process1(&request, result);
+	return box_process_rw(&request, space, result);
 }
 
 int
@@ -1101,13 +1138,16 @@ box_replace(uint32_t space_id, const char *tuple, const char *tuple_end,
 	    box_tuple_t **result)
 {
 	mp_tuple_assert(tuple, tuple_end);
+	struct space *space = space_cache_find(space_id);
+	if (space == NULL || space_check_writable(space) != 0)
+		return -1;
 	struct request request;
 	memset(&request, 0, sizeof(request));
 	request.type = IPROTO_REPLACE;
 	request.space_id = space_id;
 	request.tuple = tuple;
 	request.tuple_end = tuple_end;
-	return box_process1(&request, result);
+	return box_process_rw(&request, space, result);
 }
 
 int
@@ -1115,6 +1155,10 @@ box_delete(uint32_t space_id, uint32_t index_id, const char *key,
 	   const char *key_end, box_tuple_t **result)
 {
 	mp_tuple_assert(key, key_end);
+	struct space *space = space_cache_find(space_id);
+	if (space == NULL || space_check_writable(space) != 0 ||
+	    key_check_findable(space, index_id, key) != 0)
+		return -1;
 	struct request request;
 	memset(&request, 0, sizeof(request));
 	request.type = IPROTO_DELETE;
@@ -1122,7 +1166,7 @@ box_delete(uint32_t space_id, uint32_t index_id, const char *key,
 	request.index_id = index_id;
 	request.key = key;
 	request.key_end = key_end;
-	return box_process1(&request, result);
+	return box_process_rw(&request, space, result);
 }
 
 int
@@ -1132,6 +1176,10 @@ box_update(uint32_t space_id, uint32_t index_id, const char *key,
 {
 	mp_tuple_assert(key, key_end);
 	mp_tuple_assert(ops, ops_end);
+	struct space *space = space_cache_find(space_id);
+	if (space == NULL || space_check_writable(space) != 0 ||
+	    key_check_findable(space, index_id, key) != 0)
+		return -1;
 	struct request request;
 	memset(&request, 0, sizeof(request));
 	request.type = IPROTO_UPDATE;
@@ -1143,7 +1191,8 @@ box_update(uint32_t space_id, uint32_t index_id, const char *key,
 	/** Legacy: in case of update, ops are passed in in request tuple */
 	request.tuple = ops;
 	request.tuple_end = ops_end;
-	return box_process1(&request, result);
+
+	return box_process_rw(&request, space, result);
 }
 
 int
@@ -1153,6 +1202,9 @@ box_upsert(uint32_t space_id, uint32_t index_id, const char *tuple,
 {
 	mp_tuple_assert(ops, ops_end);
 	mp_tuple_assert(tuple, tuple_end);
+	struct space *space = space_cache_find(space_id);
+	if (space == NULL || space_check_writable(space) != 0)
+		return -1;
 	struct request request;
 	memset(&request, 0, sizeof(request));
 	request.type = IPROTO_UPSERT;
@@ -1163,7 +1215,7 @@ box_upsert(uint32_t space_id, uint32_t index_id, const char *tuple,
 	request.tuple = tuple;
 	request.tuple_end = tuple_end;
 	request.index_base = index_base;
-	return box_process1(&request, result);
+	return box_process_rw(&request, space, result);
 }
 
 /**
diff --git a/src/box/memtx_space.c b/src/box/memtx_space.c
index 185be92..f440951 100644
--- a/src/box/memtx_space.c
+++ b/src/box/memtx_space.c
@@ -362,8 +362,6 @@ memtx_space_execute_delete(struct space *space, struct txn *txn,
 		return -1;
 	const char *key = request->key;
 	uint32_t part_count = mp_decode_array(&key);
-	if (exact_key_validate(pk->def->key_def, key, part_count) != 0)
-		return -1;
 	if (index_get(pk, key, part_count, &stmt->old_tuple) != 0)
 		return -1;
 	struct tuple *old_tuple = NULL;
@@ -389,8 +387,6 @@ memtx_space_execute_update(struct space *space, struct txn *txn,
 		return -1;
 	const char *key = request->key;
 	uint32_t part_count = mp_decode_array(&key);
-	if (exact_key_validate(pk->def->key_def, key, part_count) != 0)
-		return -1;
 	if (index_get(pk, key, part_count, &stmt->old_tuple) != 0)
 		return -1;
 
diff --git a/src/box/vinyl.c b/src/box/vinyl.c
index a3e308e..10ec3ca 100644
--- a/src/box/vinyl.c
+++ b/src/box/vinyl.c
@@ -572,11 +572,7 @@ vy_lsm_find(struct space *space, uint32_t iid)
 	return vy_lsm(index);
 }
 
-/**
- * Wrapper around vy_lsm_find() which ensures that
- * the found index is unique.
- */
-static  struct vy_lsm *
+struct vy_lsm *
 vy_lsm_find_unique(struct space *space, uint32_t index_id)
 {
 	struct vy_lsm *lsm = vy_lsm_find(space, index_id);
@@ -1608,19 +1604,7 @@ error:
 	return -1;
 }
 
-/**
- * Check that the key can be used for search in a unique index
- * LSM tree.
- * @param  lsm        LSM tree for checking.
- * @param  key        MessagePack'ed data, the array without a
- *                    header.
- * @param  part_count Part count of the key.
- *
- * @retval  0 The key is valid.
- * @retval -1 The key is not valid, the appropriate error is set
- *            in the diagnostics area.
- */
-static inline int
+int
 vy_unique_key_validate(struct vy_lsm *lsm, const char *key,
 		       uint32_t part_count)
 {
@@ -1758,8 +1742,6 @@ vy_delete(struct vy_env *env, struct vy_tx *tx, struct txn_stmt *stmt,
 	bool has_secondary = space->index_count > 1;
 	const char *key = request->key;
 	uint32_t part_count = mp_decode_array(&key);
-	if (vy_unique_key_validate(lsm, key, part_count))
-		return -1;
 	/*
 	 * There are two cases when need to get the full tuple
 	 * before deletion.
@@ -1850,8 +1832,6 @@ vy_update(struct vy_env *env, struct vy_tx *tx, struct txn_stmt *stmt,
 		return -1;
 	const char *key = request->key;
 	uint32_t part_count = mp_decode_array(&key);
-	if (vy_unique_key_validate(lsm, key, part_count))
-		return -1;
 
 	if (vy_lsm_full_by_key(lsm, tx, vy_tx_read_view(tx),
 			       key, part_count, &stmt->old_tuple) != 0)
diff --git a/src/box/vinyl.h b/src/box/vinyl.h
index 21f99e4..150ac64 100644
--- a/src/box/vinyl.h
+++ b/src/box/vinyl.h
@@ -33,6 +33,7 @@
 
 #include <stdbool.h>
 #include <stddef.h>
+#include <stdint.h>
 
 #ifdef __cplusplus
 extern "C" {
@@ -40,6 +41,8 @@ extern "C" {
 
 struct info_handler;
 struct vinyl_engine;
+struct vy_lsm;
+struct space;
 
 struct vinyl_engine *
 vinyl_engine_new(const char *dir, size_t memory,
@@ -88,6 +91,29 @@ vinyl_engine_set_too_long_threshold(struct vinyl_engine *vinyl,
 void
 vinyl_engine_set_snap_io_rate_limit(struct vinyl_engine *vinyl, double limit);
 
+/**
+ * Wrapper around vy_lsm_find() which ensures that
+ * the found index is unique.
+ */
+struct vy_lsm *
+vy_lsm_find_unique(struct space *space, uint32_t index_id);
+
+/**
+ * Check that the key can be used for search in a unique index
+ * LSM tree.
+ * @param  lsm        LSM tree for checking.
+ * @param  key        MessagePack'ed data, the array without a
+ *                    header.
+ * @param  part_count Part count of the key.
+ *
+ * @retval  0 The key is valid.
+ * @retval -1 The key is not valid, the appropriate error is set
+ *            in the diagnostics area.
+ */
+int
+vy_unique_key_validate(struct vy_lsm *lsm, const char *key,
+		       uint32_t part_count);
+
 #ifdef __cplusplus
 } /* extern "C" */
 
-- 
2.7.4

[tarantool-patches] Re: [PATCH v2 2/7] box: move checks for key findability from space_vtab

[Kirill Shcherbatov]

Hi. Thank you for patch. Mostly it is looking good for me.

0. It would be great, if you describe in commit message, why you should move
this checks.

> +/**
> + * Check space for writeability.
> + */
> +static inline int
> +space_check_writable(struct space *space)> +/**
> + * Check if key is good enough
> + * to be used to find tuple.
> + */
> +static inline int
> +key_check_findable(struct space *space, uint32_t index_id, const char *key)
1. I don't like this comments. Please, make doxygen comments for them.

>  int
>  box_process1(struct request *request, box_tuple_t **result)
2. As I see, box_process1 is used only in a single place, in tx_process1; It is also extra small now.
Do we really need it in header? Maybe it worth to inline it?


> -/**
> - * Wrapper around vy_lsm_find() which ensures that
> - * the found index is unique.
> - */
> -static  struct vy_lsm *
> +struct vy_lsm *
>  vy_lsm_find_unique(struct space *space, uint32_t index_id)
3. It also lacks a good comment, like vy_unique_key_validate has.