[tarantool-patches] Re: [PATCH v2 3/7] box: create new methods for ephemeral spaces

[Vladislav Shpilevoy <v.shpilevoy@tarantool.org>]

Kirill, please, do the second review.

On 17/07/2018 17:55, imeevma@tarantool.org wrote:
> Up to this patch any space had two additional methods
> that were methods of ephemeral spaces. In this patch
> these methods deleted from vtab and from now on
> ephemeral spaces are spaces with special vtab.
> 
> 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           | 155 +++++++++++++++++-----
>   src/box/box.h            |  94 +++++++++++++
>   src/box/memtx_space.c    | 333 ++++++++++++++++++++++++++++++-----------------
>   src/box/space.h          |  17 ---
>   src/box/sql.c            |  33 ++++-
>   src/box/sysview_engine.c |  22 ----
>   src/box/vinyl.c          |  22 ----
>   7 files changed, 459 insertions(+), 217 deletions(-)
> 
> diff --git a/src/box/box.cc b/src/box/box.cc
> index 1dcbfbf..9d9f6f0 100644
> --- a/src/box/box.cc
> +++ b/src/box/box.cc
> @@ -872,6 +872,52 @@ box_set_net_msg_max(void)
>   /* }}} configuration bindings */
>   
>   /**
> + * Fill up request according to given
> + * operation type. For all cases but
> + * UPSERT argument name match names of
> + * fields of struct request. For UPSERT
> + * arguments key and key_end are fields
> + * ops and ops_end.
> + */
> +static inline void
> +request_fill(struct request *request, uint32_t type, uint32_t space_id,
> +	     uint32_t index_id, const char *key, const char *key_end,
> +	     const char *tuple, const char *tuple_end, int index_base)
> +{
> +	memset(request, 0, sizeof(*request));
> +	request->type = type;
> +	request->space_id = space_id;
> +	request->index_id = index_id;
> +	switch (type) {
> +	case IPROTO_INSERT:
> +	case IPROTO_REPLACE:
> +		request->tuple = tuple;
> +		request->tuple_end = tuple_end;
> +		break;
> +	case IPROTO_DELETE:
> +		request->key = key;
> +		request->key_end = key_end;
> +		break;
> +	case IPROTO_UPDATE:
> +		request->key = key;
> +		request->key_end = key_end;
> +		request->tuple = tuple;
> +		request->tuple_end = tuple_end;
> +		request->index_base = index_base;
> +		break;
> +	case IPROTO_UPSERT:
> +		request->ops = key;
> +		request->ops_end = key_end;
> +		request->tuple = tuple;
> +		request->tuple_end = tuple_end;
> +		request->index_base = index_base;
> +		break;
> +	default:
> +		unreachable();
> +	}
> +}
> +
> +/**
>    * Execute a request against a given space id with
>    * a variable-argument tuple described in format.
>    *
> @@ -1125,11 +1171,8 @@ box_insert(uint32_t space_id, const char *tuple, const char *tuple_end,
>   	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;
> +	request_fill(&request, IPROTO_INSERT, space_id, 0, NULL, NULL, tuple,
> +		     tuple_end, 0);
>   	return box_process_rw(&request, space, result);
>   }
>   
> @@ -1142,11 +1185,8 @@ box_replace(uint32_t space_id, const char *tuple, const char *tuple_end,
>   	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;
> +	request_fill(&request, IPROTO_REPLACE, space_id, 0, NULL, NULL, tuple,
> +		     tuple_end, 0);
>   	return box_process_rw(&request, space, result);
>   }
>   
> @@ -1160,12 +1200,8 @@ box_delete(uint32_t space_id, uint32_t index_id, const char *key,
>   	    key_check_findable(space, index_id, key) != 0)
>   		return -1;
>   	struct request request;
> -	memset(&request, 0, sizeof(request));
> -	request.type = IPROTO_DELETE;
> -	request.space_id = space_id;
> -	request.index_id = index_id;
> -	request.key = key;
> -	request.key_end = key_end;
> +	request_fill(&request, IPROTO_DELETE, space_id, index_id, key, key_end,
> +		     NULL, NULL, 0);
>   	return box_process_rw(&request, space, result);
>   }
>   
> @@ -1181,17 +1217,8 @@ box_update(uint32_t space_id, uint32_t index_id, const char *key,
>   	    key_check_findable(space, index_id, key) != 0)
>   		return -1;
>   	struct request request;
> -	memset(&request, 0, sizeof(request));
> -	request.type = IPROTO_UPDATE;
> -	request.space_id = space_id;
> -	request.index_id = index_id;
> -	request.key = key;
> -	request.key_end = key_end;
> -	request.index_base = index_base;
> -	/** Legacy: in case of update, ops are passed in in request tuple */
> -	request.tuple = ops;
> -	request.tuple_end = ops_end;
> -
> +	request_fill(&request, IPROTO_UPDATE, space_id, index_id, key, key_end,
> +		     ops, ops_end, index_base);
>   	return box_process_rw(&request, space, result);
>   }
>   
> @@ -1206,18 +1233,74 @@ box_upsert(uint32_t space_id, uint32_t index_id, const char *tuple,
>   	if (space == NULL || space_check_writable(space) != 0)
>   		return -1;
>   	struct request request;
> -	memset(&request, 0, sizeof(request));
> -	request.type = IPROTO_UPSERT;
> -	request.space_id = space_id;
> -	request.index_id = index_id;
> -	request.ops = ops;
> -	request.ops_end = ops_end;
> -	request.tuple = tuple;
> -	request.tuple_end = tuple_end;
> -	request.index_base = index_base;
> +	request_fill(&request, IPROTO_UPSERT, space_id, index_id, ops, ops_end,
> +		     tuple, tuple_end, index_base);
>   	return box_process_rw(&request, space, result);
>   }
>   
> +int
> +box_ephemeral_insert(struct space *space, const char *tuple,
> +		     const char *tuple_end, box_tuple_t **result)
> +{
> +	mp_tuple_assert(tuple, tuple_end);
> +	struct request request;
> +	request_fill(&request, IPROTO_INSERT, 0, 0, NULL, NULL, tuple,
> +		     tuple_end, 0);
> +	return space_execute_dml(space, NULL, &request, result);
> +}
> +
> +int
> +box_ephemeral_replace(struct space *space, const char *tuple,
> +		      const char *tuple_end, box_tuple_t **result)
> +{
> +	mp_tuple_assert(tuple, tuple_end);
> +	struct request request;
> +	request_fill(&request, IPROTO_REPLACE, 0, 0, NULL, NULL, tuple,
> +		     tuple_end, 0);
> +	return space_execute_dml(space, NULL, &request, result);
> +}
> +
> +int
> +box_ephemeral_delete(struct space *space, uint32_t index_id, const char *key,
> +		     const char *key_end, box_tuple_t **result)
> +{
> +	mp_tuple_assert(key, key_end);
> +	if (key_check_findable(space, index_id, key) != 0)
> +		return -1;
> +	struct request request;
> +	request_fill(&request, IPROTO_DELETE, 0, index_id, key, key_end,
> +		     NULL, NULL, 0);
> +	return space_execute_dml(space, NULL, &request, result);
> +}
> +
> +int
> +box_ephemeral_update(struct space *space, uint32_t index_id, const char *key,
> +		     const char *key_end, const char *ops, const char *ops_end,
> +		     int index_base, box_tuple_t **result)
> +{
> +	mp_tuple_assert(key, key_end);
> +	mp_tuple_assert(ops, ops_end);
> +	if (key_check_findable(space, index_id, key) != 0)
> +		return -1;
> +	struct request request;
> +	request_fill(&request, IPROTO_UPDATE, 0, index_id, key, key_end,
> +		     ops, ops_end, index_base);
> +	return space_execute_dml(space, NULL, &request, result);
> +}
> +
> +int
> +box_ephemeral_upsert(struct space *space, uint32_t index_id, const char *tuple,
> +		     const char *tuple_end, const char *ops,
> +		     const char *ops_end, int index_base, box_tuple_t **result)
> +{
> +	mp_tuple_assert(ops, ops_end);
> +	mp_tuple_assert(tuple, tuple_end);
> +	struct request request;
> +	request_fill(&request, IPROTO_UPSERT, 0, index_id, ops, ops_end,
> +		     tuple, tuple_end, index_base);
> +	return space_execute_dml(space, NULL, &request, result);
> +}
> +
>   /**
>    * Trigger space truncation by bumping a counter
>    * in _truncate space.
> diff --git a/src/box/box.h b/src/box/box.h
> index 422f62f..3601823 100644
> --- a/src/box/box.h
> +++ b/src/box/box.h
> @@ -401,6 +401,100 @@ int
>   box_process1(struct request *request, box_tuple_t **result);
>   
>   /**
> + * Execute an INSERT request for ephemeral spaces.
> + *
> + * \param space ephemeral space.
> + * \param tuple encoded tuple in
> + * MsgPack Array format.
> + * \param tuple_end end of @a tuple.
> + * \param[out] result a new tuple.
> + * \retval -1 on error.
> + * \retval 0 on success.
> + */
> +int
> +box_ephemeral_insert(struct space *space, const char *tuple,
> +		     const char *tuple_end, box_tuple_t **result);
> +
> +/**
> + * Execute an REPLACE request for ephemeral spaces.
> + *
> + * \param space ephemeral space.
> + * \param tuple encoded tuple in
> + * MsgPack Array format.
> + * \param tuple_end end of @a tuple.
> + * \param[out] result a new tuple.
> + * \retval -1 on error.
> + * \retval 0 on success.
> + */
> +int
> +box_ephemeral_replace(struct space *space, const char *tuple,
> +		      const char *tuple_end, box_tuple_t **result);
> +
> +/**
> + * Execute an DELETE request for ephemeral spaces.
> + *
> + * \param space ephemeral space.
> + * \param index_id index identifier
> + * \param key encoded key in
> + * MsgPack Array format.
> + * \param key_end the end of encoded \a key.
> + * \param[out] result an old tuple.
> + * \retval -1 on error.
> + * \retval 0 on success.
> + */
> +int
> +box_ephemeral_delete(struct space *space, uint32_t index_id, const char *key,
> +		     const char *key_end, box_tuple_t **result);
> +
> +/**
> + * Execute an UPDATE request for ephemeral spaces.
> + *
> + * \param space ephemeral space.
> + * \param index_id index identifier
> + * \param key encoded key in
> + * MsgPack Array format.
> + * \param key_end the end of encoded \a key.
> + * \param ops encoded operations in
> + * MsgPack Arrat format.
> + * \param ops_end the end of encoded \a ops.
> + * \param index_base 0 if fieldnos in
> + * update operations are zero-based
> + * indexed (like C) or 1 if for one-based
> + * indexed field ids (like Lua).
> + * \param[out] result a new tuple.
> + * \retval -1 on error.
> + * \retval 0 on success.
> + */
> +int
> +box_ephemeral_update(struct space *space, uint32_t index_id, const char *key,
> +		     const char *key_end, const char *ops, const char *ops_end,
> +		     int index_base, box_tuple_t **result);
> +
> +/**
> + * Execute an UPSERT request for ephemeral spaces.
> + *
> + * \param space ephemeral space.
> + * \param index_id index identifier
> + * \param ops encoded operations in
> + * MsgPack Arrat format.
> + * \param ops_end the end of encoded \a ops.
> + * \param tuple encoded tuple in
> + * MsgPack Array format.
> + * \param tuple_end end of @a tuple.
> + * \param index_base 0 if fieldnos in update
> + * operations are zero-based
> + * indexed (like C) or 1 if for one-based
> + * indexed field ids (like Lua).
> + * \param[out] result a new tuple.
> + * \retval -1 on error.
> + * \retval 0 on success.
> + */
> +int
> +box_ephemeral_upsert(struct space *space, uint32_t index_id, const char *tuple,
> +		     const char *tuple_end, const char *ops,
> +		     const char *ops_end, int index_base, box_tuple_t **result);
> +
> +/**
>    * Execute request on given space.
>    *
>    * \param request Request to be executed
> diff --git a/src/box/memtx_space.c b/src/box/memtx_space.c
> index f440951..ce169be 100644
> --- a/src/box/memtx_space.c
> +++ b/src/box/memtx_space.c
> @@ -327,77 +327,76 @@ rollback:
>   	return -1;
>   }
>   
> -static int
> -memtx_space_execute_replace(struct space *space, struct txn *txn,
> -			    struct request *request, struct tuple **result)
> +static inline int
> +memtx_space_replace_impl(struct space *space, struct request *request,
> +			 struct tuple **new_tuple, struct tuple **old_tuple,
> +			 struct tuple **result)
>   {
>   	struct memtx_space *memtx_space = (struct memtx_space *)space;
> -	struct txn_stmt *stmt = txn_current_stmt(txn);
>   	enum dup_replace_mode mode = dup_replace_mode(request->type);
> -	stmt->new_tuple = memtx_tuple_new(space->format, request->tuple,
> -					  request->tuple_end);
> -	if (stmt->new_tuple == NULL)
> +	*new_tuple = memtx_tuple_new(space->format, request->tuple,
> +				     request->tuple_end);
> +	if (*new_tuple == NULL)
>   		return -1;
> -	tuple_ref(stmt->new_tuple);
> -	struct tuple *old_tuple;
> -	if (memtx_space->replace(space, stmt->old_tuple, stmt->new_tuple,
> -				 mode, &old_tuple) != 0)
> +	tuple_ref(*new_tuple);
> +	struct tuple *tmp_tuple;
> +	if (memtx_space->replace(space, *old_tuple, *new_tuple, mode,
> +				 &tmp_tuple) != 0)
>   		return -1;
> -	stmt->old_tuple = old_tuple;
> -	stmt->engine_savepoint = stmt;
> +	*old_tuple = tmp_tuple;
>   	/** The new tuple is referenced by the primary key. */
> -	*result = stmt->new_tuple;
> +	*result = *new_tuple;
>   	return 0;
> +
>   }
>   
> -static int
> -memtx_space_execute_delete(struct space *space, struct txn *txn,
> -			   struct request *request, struct tuple **result)
> +static inline int
> +memtx_space_delete_impl(struct space *space, struct request *request,
> +			struct tuple **new_tuple, struct tuple **old_tuple,
> +			struct tuple **result)
>   {
>   	struct memtx_space *memtx_space = (struct memtx_space *)space;
> -	struct txn_stmt *stmt = txn_current_stmt(txn);
>   	/* Try to find the tuple by unique key. */
>   	struct index *pk = index_find_unique(space, request->index_id);
>   	if (pk == NULL)
>   		return -1;
>   	const char *key = request->key;
>   	uint32_t part_count = mp_decode_array(&key);
> -	if (index_get(pk, key, part_count, &stmt->old_tuple) != 0)
> +	if (index_get(pk, key, part_count, old_tuple) != 0)
>   		return -1;
> -	struct tuple *old_tuple = NULL;
> -	if (stmt->old_tuple != NULL &&
> -	    memtx_space->replace(space, stmt->old_tuple, stmt->new_tuple,
> -				 DUP_REPLACE_OR_INSERT, &old_tuple) != 0)
> +	struct tuple *tmp_tuple = NULL;
> +	if (*old_tuple != NULL &&
> +	    memtx_space->replace(space, *old_tuple, *new_tuple,
> +				 DUP_REPLACE_OR_INSERT, &tmp_tuple) != 0)
>   		return -1;
> -	stmt->old_tuple = old_tuple;
> -	stmt->engine_savepoint = stmt;
> -	*result = stmt->old_tuple;
> +	*old_tuple = tmp_tuple;
> +	*result = *old_tuple;
>   	return 0;
>   }
>   
>   static int
> -memtx_space_execute_update(struct space *space, struct txn *txn,
> -			   struct request *request, struct tuple **result)
> +memtx_space_update_impl(struct space *space, struct request *request,
> +			struct tuple **new_tuple, struct tuple **old_tuple,
> +			struct tuple **result)
>   {
>   	struct memtx_space *memtx_space = (struct memtx_space *)space;
> -	struct txn_stmt *stmt = txn_current_stmt(txn);
>   	/* Try to find the tuple by unique key. */
>   	struct index *pk = index_find_unique(space, request->index_id);
>   	if (pk == NULL)
>   		return -1;
>   	const char *key = request->key;
>   	uint32_t part_count = mp_decode_array(&key);
> -	if (index_get(pk, key, part_count, &stmt->old_tuple) != 0)
> +	if (index_get(pk, key, part_count, old_tuple) != 0)
>   		return -1;
>   
> -	if (stmt->old_tuple == NULL) {
> +	if (*old_tuple == NULL) {
>   		*result = NULL;
>   		return 0;
>   	}
>   
>   	/* Update the tuple; legacy, request ops are in request->tuple */
>   	uint32_t new_size = 0, bsize;
> -	const char *old_data = tuple_data_range(stmt->old_tuple, &bsize);
> +	const char *old_data = tuple_data_range(*old_tuple, &bsize);
>   	const char *new_data =
>   		tuple_update_execute(region_aligned_alloc_cb, &fiber()->gc,
>   				     request->tuple, request->tuple_end,
> @@ -406,28 +405,26 @@ memtx_space_execute_update(struct space *space, struct txn *txn,
>   	if (new_data == NULL)
>   		return -1;
>   
> -	stmt->new_tuple = memtx_tuple_new(space->format, new_data,
> -					  new_data + new_size);
> -	if (stmt->new_tuple == NULL)
> +	*new_tuple = memtx_tuple_new(space->format, new_data,
> +				     new_data + new_size);
> +	if (*new_tuple == NULL)
>   		return -1;
> -	tuple_ref(stmt->new_tuple);
> -	struct tuple *old_tuple = NULL;
> -	if (stmt->old_tuple != NULL &&
> -	    memtx_space->replace(space, stmt->old_tuple, stmt->new_tuple,
> -				 DUP_REPLACE, &old_tuple) != 0)
> +	tuple_ref(*new_tuple);
> +	struct tuple *tmp_tuple = NULL;
> +	if (*old_tuple != NULL &&
> +	    memtx_space->replace(space, *old_tuple, *new_tuple,
> +				 DUP_REPLACE, &tmp_tuple) != 0)
>   		return -1;
> -	stmt->old_tuple = old_tuple;
> -	stmt->engine_savepoint = stmt;
> -	*result = stmt->new_tuple;
> +	*old_tuple = tmp_tuple;
> +	*result = *new_tuple;
>   	return 0;
>   }
>   
>   static int
> -memtx_space_execute_upsert(struct space *space, struct txn *txn,
> -			   struct request *request)
> +memtx_space_upsert_impl(struct space *space, struct request *request,
> +			struct tuple **new_tuple, struct tuple **old_tuple)
>   {
>   	struct memtx_space *memtx_space = (struct memtx_space *)space;
> -	struct txn_stmt *stmt = txn_current_stmt(txn);
>   	/*
>   	 * Check all tuple fields: we should produce an error on
>   	 * malformed tuple even if upsert turns into an update.
> @@ -450,10 +447,10 @@ memtx_space_execute_upsert(struct space *space, struct txn *txn,
>   	mp_decode_array(&key);
>   
>   	/* Try to find the tuple by primary key. */
> -	if (index_get(index, key, part_count, &stmt->old_tuple) != 0)
> +	if (index_get(index, key, part_count, old_tuple) != 0)
>   		return -1;
>   
> -	if (stmt->old_tuple == NULL) {
> +	if (*old_tuple == NULL) {
>   		/**
>   		 * Old tuple was not found. A write optimized
>   		 * engine may only know this after commit, so
> @@ -470,21 +467,19 @@ memtx_space_execute_upsert(struct space *space, struct txn *txn,
>   		 *   we get it here, it's also OK to throw it
>   		 * @sa https://github.com/tarantool/tarantool/issues/1156
>   		 */
> -		if (tuple_update_check_ops(region_aligned_alloc_cb, &fiber()->gc,
> -				       request->ops, request->ops_end,
> -				       request->index_base)) {
> +		if (tuple_update_check_ops(region_aligned_alloc_cb,
> +					   &fiber()->gc, request->ops,
> +					   request->ops_end,
> +					   request->index_base))
>   			return -1;
> -		}
> -		stmt->new_tuple = memtx_tuple_new(space->format,
> -						  request->tuple,
> -						  request->tuple_end);
> -		if (stmt->new_tuple == NULL)
> +		*new_tuple = memtx_tuple_new(space->format, request->tuple,
> +					     request->tuple_end);
> +		if (*new_tuple == NULL)
>   			return -1;
> -		tuple_ref(stmt->new_tuple);
> +		tuple_ref(*new_tuple);
>   	} else {
>   		uint32_t new_size = 0, bsize;
> -		const char *old_data = tuple_data_range(stmt->old_tuple,
> -							&bsize);
> +		const char *old_data = tuple_data_range(*old_tuple, &bsize);
>   		/*
>   		 * Update the tuple.
>   		 * tuple_upsert_execute() fails on totally wrong
> @@ -502,24 +497,24 @@ memtx_space_execute_upsert(struct space *space, struct txn *txn,
>   		if (new_data == NULL)
>   			return -1;
>   
> -		stmt->new_tuple = memtx_tuple_new(space->format, new_data,
> -						  new_data + new_size);
> -		if (stmt->new_tuple == NULL)
> +		*new_tuple = memtx_tuple_new(space->format, new_data,
> +					     new_data + new_size);
> +		if (*new_tuple == NULL)
>   			return -1;
> -		tuple_ref(stmt->new_tuple);
> +		tuple_ref(*new_tuple);
>   
>   		struct index *pk = space->index[0];
>   		if (!key_update_can_be_skipped(pk->def->key_def->column_mask,
>   					       column_mask) &&
> -		    tuple_compare(stmt->old_tuple, stmt->new_tuple,
> +		    tuple_compare(*old_tuple, *new_tuple,
>   				  pk->def->key_def) != 0) {
>   			/* Primary key is changed: log error and do nothing. */
>   			diag_set(ClientError, ER_CANT_UPDATE_PRIMARY_KEY,
>   				 pk->def->name, space_name(space));
>   			diag_log();
> -			tuple_unref(stmt->new_tuple);
> -			stmt->old_tuple = NULL;
> -			stmt->new_tuple = NULL;
> +			tuple_unref(*new_tuple);
> +			*old_tuple = NULL;
> +			*new_tuple = NULL;
>   		}
>   	}
>   	/*
> @@ -528,77 +523,153 @@ memtx_space_execute_upsert(struct space *space, struct txn *txn,
>   	 * we checked this case explicitly and skipped the upsert
>   	 * above.
>   	 */
> -	struct tuple *old_tuple = NULL;
> -	if (stmt->new_tuple != NULL &&
> -	    memtx_space->replace(space, stmt->old_tuple, stmt->new_tuple,
> -				 DUP_REPLACE_OR_INSERT, &old_tuple) != 0)
> +	struct tuple *tmp_tuple = NULL;
> +	if (*new_tuple != NULL &&
> +	    memtx_space->replace(space, *old_tuple, *new_tuple,
> +				 DUP_REPLACE_OR_INSERT, &tmp_tuple) != 0)
> +		return -1;
> +	*old_tuple = tmp_tuple;
> +	return 0;
> +}
> +
> +static int
> +memtx_space_execute_replace(struct space *space, struct txn *txn,
> +			    struct request *request, struct tuple **result)
> +{
> +	struct txn_stmt *stmt = txn_current_stmt(txn);
> +	if (memtx_space_replace_impl(space, request, &stmt->new_tuple,
> +				     &stmt->old_tuple, result) != 0)
> +		return -1;
> +	stmt->engine_savepoint = stmt;
> +	return 0;
> +}
> +
> +static int
> +memtx_space_execute_delete(struct space *space, struct txn *txn,
> +			   struct request *request, struct tuple **result)
> +{
> +	struct txn_stmt *stmt = txn_current_stmt(txn);
> +	if (memtx_space_delete_impl(space, request, &stmt->new_tuple,
> +				    &stmt->old_tuple, result) != 0)
> +		return -1;
> +	stmt->engine_savepoint = stmt;
> +	return 0;
> +}
> +
> +static int
> +memtx_space_execute_update(struct space *space, struct txn *txn,
> +			   struct request *request, struct tuple **result)
> +{
> +	struct txn_stmt *stmt = txn_current_stmt(txn);
> +	if (memtx_space_update_impl(space, request, &stmt->new_tuple,
> +				    &stmt->old_tuple, result) != 0)
> +		return -1;
> +	stmt->engine_savepoint = stmt;
> +	return 0;
> +}
> +
> +static int
> +memtx_space_execute_upsert(struct space *space, struct txn *txn,
> +			   struct request *request)
> +{
> +	struct txn_stmt *stmt = txn_current_stmt(txn);
> +	if (memtx_space_upsert_impl(space, request, &stmt->new_tuple,
> +				    &stmt->old_tuple) != 0)
>   		return -1;
> -	stmt->old_tuple = old_tuple;
>   	stmt->engine_savepoint = stmt;
>   	/* Return nothing: UPSERT does not return data. */
>   	return 0;
>   }
>   
>   /**
> - * This function simply creates new memtx tuple, refs it and calls space's
> - * replace function. In constrast to original memtx_space_execute_replace(), it
> - * doesn't handle any transaction routine.
> - * Ephemeral spaces shouldn't be involved in transaction routine, since
> - * they are used only for internal purposes. Moreover, ephemeral spaces
> - * can be created and destroyed within one transaction and rollback of already
> - * destroyed space may lead to undefined behaviour. For this reason it
> - * doesn't take txn as an argument.
> + * Executes INSERT and REPLACE
> + * operations for ephemeral spaces.
> + *
> + * This function isn't involved in
> + * any transaction routine.
>    */
>   static int
> -memtx_space_ephemeral_replace(struct space *space, const char *tuple,
> -				      const char *tuple_end)
> +memtx_space_ephemeral_replace(struct space *space, struct txn *txn,
> +			      struct request *request, struct tuple **result)
>   {
> -	struct memtx_space *memtx_space = (struct memtx_space *)space;
> -	struct tuple *new_tuple = memtx_tuple_new(space->format, tuple,
> -						  tuple_end);
> -	if (new_tuple == NULL)
> -		return -1;
> -	tuple_ref(new_tuple);
> +	assert(txn == NULL);
> +	(void)txn;
> +	struct tuple *new_tuple = NULL;
>   	struct tuple *old_tuple = NULL;
> -	if (memtx_space->replace(space, old_tuple, new_tuple,
> -				 DUP_REPLACE_OR_INSERT, &old_tuple) != 0) {
> +	int rc = memtx_space_replace_impl(space, request, &new_tuple,
> +					  &old_tuple, result);
> +	if (rc && new_tuple != NULL)
>   		tuple_unref(new_tuple);
> +	if (rc != 0)
>   		return -1;
> -	}
>   	if (old_tuple != NULL)
>   		tuple_unref(old_tuple);
>   	return 0;
>   }
>   
>   /**
> - * Delete tuple with given key from primary index. Tuple checking is omitted
> - * due to the ability of ephemeral spaces to hold nulls in primary key.
> - * Generally speaking, it is not correct behaviour owing to ambiguity when
> - * fetching/deleting tuple from space with several tuples containing
> - * nulls in PK. On the other hand, ephemeral spaces are used only for internal
> - * needs, so if it is guaranteed that no such situation occur
> - * (when several tuples with nulls in PK exist), it is OK to allow
> - * insertion nulls in PK.
> + * Executes DELETE operation
> + * for ephemeral space.
>    *
> - * Similarly to ephemeral replace function,
> - * it isn't involved in any transaction routine.
> + * This function isn't involved in
> + * any transaction routine.
>    */
>   static int
> -memtx_space_ephemeral_delete(struct space *space, const char *key)
> +memtx_space_ephemeral_delete(struct space *space, struct txn *txn,
> +			     struct request *request, struct tuple **result)
>   {
> -	struct memtx_space *memtx_space = (struct memtx_space *)space;
> -	struct index *primary_index = space_index(space, 0 /* primary index*/);
> -	if (primary_index == NULL)
> +	assert(txn == NULL);
> +	(void)txn;
> +	struct tuple *new_tuple = NULL;
> +	struct tuple *old_tuple = NULL;
> +	if (memtx_space_delete_impl(space, request, &new_tuple, &old_tuple,
> +				    result) != 0)
>   		return -1;
> -	uint32_t part_count = mp_decode_array(&key);
> -	struct tuple *old_tuple;
> -	if (index_get(primary_index, key, part_count, &old_tuple) != 0)
> +	return 0;
> +}
> +
> +/**
> + * Executes UPDATE operation
> + * for ephemeral space.
> + *
> + * This function isn't involved in
> + * any transaction routine.
> + */
> +static int
> +memtx_space_ephemeral_update(struct space *space, struct txn *txn,
> +			     struct request *request, struct tuple **result)
> +{
> +	assert(txn == NULL);
> +	(void)txn;
> +	struct tuple *new_tuple = NULL;
> +	struct tuple *old_tuple = NULL;
> +	int rc = memtx_space_update_impl(space, request, &new_tuple,
> +					 &old_tuple, result);
> +	if (rc != 0 && new_tuple != NULL)
> +		tuple_unref(new_tuple);
> +	if (rc != 0)
>   		return -1;
> -	if (old_tuple != NULL &&
> -	    memtx_space->replace(space, old_tuple, NULL,
> -				 DUP_REPLACE, &old_tuple) != 0)
> +	return 0;
> +}
> +
> +/**
> + * Executes UPSERT operation
> + * for ephemral space.
> + *
> + * This function isn't involved in
> + * any transaction routine.
> + */
> +static int
> +memtx_space_ephemeral_upsert(struct space *space, struct txn *txn,
> +			     struct request *request)
> +{
> +	assert(txn == NULL);
> +	(void)txn;
> +	struct tuple *old_tuple = NULL;
> +	struct tuple *new_tuple = NULL;
> +	if (memtx_space_upsert_impl(space, request, &new_tuple,
> +				    &old_tuple) != 0)
>   		return -1;
> -	tuple_unref(old_tuple);
>   	return 0;
>   }
>   
> @@ -839,9 +910,14 @@ memtx_init_system_space(struct space *space)
>   }
>   
>   static void
> -memtx_init_ephemeral_space(struct space *space)
> +memtx_init_ephemeral_space(struct space *space);
> +
> +static void
> +memtx_init_unsupported_space(struct space *space)
>   {
> -	memtx_space_add_primary_key(space);
> +	(void)space;
> +	diag_set(ClientError, ER_UNSUPPORTED, space_name(space),
> +		 "init_ephemeral_space");
>   }
>   
>   static int
> @@ -940,8 +1016,6 @@ static const struct space_vtab memtx_space_vtab = {
>   	/* .execute_delete = */ memtx_space_execute_delete,
>   	/* .execute_update = */ memtx_space_execute_update,
>   	/* .execute_upsert = */ memtx_space_execute_upsert,
> -	/* .ephemeral_replace = */ memtx_space_ephemeral_replace,
> -	/* .ephemeral_delete = */ memtx_space_ephemeral_delete,
>   	/* .init_system_space = */ memtx_init_system_space,
>   	/* .init_ephemeral_space = */ memtx_init_ephemeral_space,
>   	/* .check_index_def = */ memtx_space_check_index_def,
> @@ -954,6 +1028,33 @@ static const struct space_vtab memtx_space_vtab = {
>   	/* .prepare_alter = */ memtx_space_prepare_alter,
>   };
>   
> +static const struct space_vtab memtx_space_ephemeral_vtab = {
> +	/* .destroy = */ memtx_space_destroy,
> +	/* .bsize = */ memtx_space_bsize,
> +	/* .apply_initial_join_row = */ memtx_space_apply_initial_join_row,
> +	/* .execute_replace = */ memtx_space_ephemeral_replace,
> +	/* .execute_delete = */ memtx_space_ephemeral_delete,
> +	/* .execute_update = */ memtx_space_ephemeral_update,
> +	/* .execute_upsert = */ memtx_space_ephemeral_upsert,
> +	/* .init_system_space = */ memtx_init_system_space,
> +	/* .init_ephemeral_space = */ memtx_init_unsupported_space,
> +	/* .check_index_def = */ memtx_space_check_index_def,
> +	/* .create_index = */ memtx_space_create_index,
> +	/* .add_primary_key = */ memtx_space_add_primary_key,
> +	/* .drop_primary_key = */ memtx_space_drop_primary_key,
> +	/* .check_format  = */ memtx_space_check_format,
> +	/* .build_index = */ memtx_space_build_index,
> +	/* .swap_index = */ generic_space_swap_index,
> +	/* .prepare_alter = */ memtx_space_prepare_alter,
> +};
> +
> +static void
> +memtx_init_ephemeral_space(struct space *space)
> +{
> +	space->vtab = &memtx_space_ephemeral_vtab;
> +	memtx_space_add_primary_key(space);
> +}
> +
>   struct space *
>   memtx_space_new(struct memtx_engine *memtx,
>   		struct space_def *def, struct rlist *key_list)
> diff --git a/src/box/space.h b/src/box/space.h
> index adf06b7..60fa603 100644
> --- a/src/box/space.h
> +++ b/src/box/space.h
> @@ -67,10 +67,6 @@ struct space_vtab {
>   			      struct request *, struct tuple **result);
>   	int (*execute_upsert)(struct space *, struct txn *, struct request *);
>   
> -	int (*ephemeral_replace)(struct space *, const char *, const char *);
> -
> -	int (*ephemeral_delete)(struct space *, const char *);
> -
>   	void (*init_system_space)(struct space *);
>   	/**
>   	 * Initialize an ephemeral space instance.
> @@ -310,19 +306,6 @@ int
>   space_execute_dml(struct space *space, struct txn *txn,
>   		  struct request *request, struct tuple **result);
>   
> -static inline int
> -space_ephemeral_replace(struct space *space, const char *tuple,
> -			const char *tuple_end)
> -{
> -	return space->vtab->ephemeral_replace(space, tuple, tuple_end);
> -}
> -
> -static inline int
> -space_ephemeral_delete(struct space *space, const char *key)
> -{
> -	return space->vtab->ephemeral_delete(space, key);
> -}
> -
>   /**
>    * Generic implementation of space_vtab::swap_index
>    * that simply swaps the two indexes in index maps.
> diff --git a/src/box/sql.c b/src/box/sql.c
> index d2cc0a9..4077890 100644
> --- a/src/box/sql.c
> +++ b/src/box/sql.c
> @@ -451,7 +451,13 @@ int tarantoolSqlite3EphemeralInsert(struct space *space, const char *tuple,
>   {
>   	assert(space != NULL);
>   	mp_tuple_assert(tuple, tuple_end);
> -	if (space_ephemeral_replace(space, tuple, tuple_end) != 0)
> +	struct request request;
> +	memset(&request, 0, sizeof(request));
> +	request.type = IPROTO_REPLACE;
> +	request.tuple = tuple;
> +	request.tuple_end = tuple_end;
> +	struct tuple *result;
> +	if (space_execute_dml(space, NULL, &request, &result) != 0)
>   		return SQL_TARANTOOL_INSERT_FAIL;
>   	return SQLITE_OK;
>   }
> @@ -516,11 +522,20 @@ int tarantoolSqlite3EphemeralDelete(BtCursor *pCur)
>   	if (key == NULL)
>   		return SQL_TARANTOOL_DELETE_FAIL;
>   
> -	int rc = space_ephemeral_delete(pCur->space, key);
> -	if (rc != 0) {
> +	mp_tuple_assert(key, key + key_size);
> +	struct request request;
> +	struct tuple *result;
> +	memset(&request, 0, sizeof(request));
> +	request.type = IPROTO_DELETE;
> +	request.key = key;
> +	request.key_end = key + key_size;
> +
> +	if (space_execute_dml(pCur->space, NULL, &request, &result) != 0) {
>   		diag_log();
>   		return SQL_TARANTOOL_DELETE_FAIL;
>   	}
> +	if (result != NULL)
> +		box_tuple_unref(result);
>   	return SQLITE_OK;
>   }
>   
> @@ -596,14 +611,24 @@ int tarantoolSqlite3EphemeralClearTable(BtCursor *pCur)
>   	struct tuple *tuple;
>   	char *key;
>   	uint32_t  key_size;
> +	struct request request;
> +	struct tuple *result;
> +	memset(&request, 0, sizeof(request));
> +	request.type = IPROTO_DELETE;
>   
>   	while (iterator_next(it, &tuple) == 0 && tuple != NULL) {
>   		key = tuple_extract_key(tuple, it->index->def->key_def,
>   					&key_size);
> -		if (space_ephemeral_delete(pCur->space, key) != 0) {
> +		mp_tuple_assert(key, key + key_size);
> +		request.key = key;
> +		request.key_end = key + key_size;
> +		if (space_execute_dml(pCur->space, NULL, &request,
> +				      &result) != 0) {
>   			iterator_delete(it);
>   			return SQL_TARANTOOL_DELETE_FAIL;
>   		}
> +		if (result != NULL)
> +			box_tuple_unref(result);
>   	}
>   	iterator_delete(it);
>   
> diff --git a/src/box/sysview_engine.c b/src/box/sysview_engine.c
> index bd9432a..3067175 100644
> --- a/src/box/sysview_engine.c
> +++ b/src/box/sysview_engine.c
> @@ -99,26 +99,6 @@ sysview_space_execute_upsert(struct space *space, struct txn *txn,
>   	return -1;
>   }
>   
> -static int
> -sysview_space_ephemeral_replace(struct space *space, const char *tuple,
> -				const char *tuple_end)
> -{
> -	(void)space;
> -	(void)tuple;
> -	(void)tuple_end;
> -	unreachable();
> -	return -1;
> -}
> -
> -static int
> -sysview_space_ephemeral_delete(struct space *space, const char *key)
> -{
> -	(void)key;
> -	(void)space;
> -	unreachable();
> -	return -1;
> -}
> -
>   static void
>   sysview_init_system_space(struct space *space)
>   {
> @@ -197,8 +177,6 @@ static const struct space_vtab sysview_space_vtab = {
>   	/* .execute_delete = */ sysview_space_execute_delete,
>   	/* .execute_update = */ sysview_space_execute_update,
>   	/* .execute_upsert = */ sysview_space_execute_upsert,
> -	/* .ephemeral_replace = */ sysview_space_ephemeral_replace,
> -	/* .ephemeral_delete = */ sysview_space_ephemeral_delete,
>   	/* .init_system_space = */ sysview_init_system_space,
>   	/* .init_ephemeral_space = */ sysview_init_ephemeral_space,
>   	/* .check_index_def = */ sysview_space_check_index_def,
> diff --git a/src/box/vinyl.c b/src/box/vinyl.c
> index 10ec3ca..d089f68 100644
> --- a/src/box/vinyl.c
> +++ b/src/box/vinyl.c
> @@ -2325,26 +2325,6 @@ vinyl_space_execute_upsert(struct space *space, struct txn *txn,
>   	return vy_upsert(env, tx, stmt, space, request);
>   }
>   
> -static int
> -vinyl_space_ephemeral_replace(struct space *space, const char *tuple,
> -			      const char *tuple_end)
> -{
> -	(void)space;
> -	(void)tuple;
> -	(void)tuple_end;
> -	unreachable();
> -	return -1;
> -}
> -
> -static int
> -vinyl_space_ephemeral_delete(struct space *space, const char *key)
> -{
> -	(void)space;
> -	(void)key;
> -	unreachable();
> -	return -1;
> -}
> -
>   static inline void
>   txn_stmt_unref_tuples(struct txn_stmt *stmt)
>   {
> @@ -4494,8 +4474,6 @@ static const struct space_vtab vinyl_space_vtab = {
>   	/* .execute_delete = */ vinyl_space_execute_delete,
>   	/* .execute_update = */ vinyl_space_execute_update,
>   	/* .execute_upsert = */ vinyl_space_execute_upsert,
> -	/* .ephemeral_replace = */ vinyl_space_ephemeral_replace,
> -	/* .ephemeral_delete = */ vinyl_space_ephemeral_delete,
>   	/* .init_system_space = */ vinyl_init_system_space,
>   	/* .init_ephemeral_space = */ vinyl_init_ephemeral_space,
>   	/* .check_index_def = */ vinyl_space_check_index_def,
>