box: introduce result processor

A convenient helper that is supposed to apply various transformations
to a tuple fetched from a space before returning it to the user.
Usage:

  struct result_processor res_proc;
  result_process_prepare(&res_proc, space);
  rc = index_get(index, key, part_count, result);
  result_process(&res_proc, &rc, result);

We need to split the procedure into two parts, because index_get and
similar methods may yield in case of Vinyl. The 'prepare' method is
supposed to pin all data structures needed to process the result in the
result_processor struct while the 'process' method is supposed to
release them.

Currently, the new methods do absolutely nothing. They will be used to
convert tuples during space upgrade.

NO_DOC=internal
NO_TEST=internal
NO_CHANGELOG=internal

Author: locker

src/box/box.cc

@@ -58,6 +58,7 @@
 #include "vinyl.h"
 #include "space.h"
 #include "index.h"
+#include "result.h"
 #include "port.h"
 #include "txn.h"
 #include "txn_limbo.h"
@@ -297,6 +298,8 @@ box_process_rw(struct request *request, struct space *space,
 	bool return_tuple = false;
 	struct txn *txn = in_txn();
 	bool is_autocommit = txn == NULL;
+	struct result_processor res_proc;
+	int rc;
 	if (is_autocommit && (txn = txn_begin()) == NULL)
 		return -1;
 	assert(iproto_type_is_dml(request->type));
@@ -305,14 +308,19 @@ box_process_rw(struct request *request, struct space *space,
 		goto rollback;
 	if (txn_begin_stmt(txn, space, request->type) != 0)
 		goto rollback;
-	if (space_execute_dml(space, txn, request, &tuple) != 0) {
+	result_process_prepare(&res_proc, space);
+	rc = space_execute_dml(space, txn, request, &tuple);
+	if (result == NULL)
+		tuple = NULL;
+	result_process(&res_proc, &rc, &tuple);
+	if (rc != 0) {
 		txn_rollback_stmt(txn);
 		goto rollback;
 	}
 	if (result != NULL)
 		*result = tuple;
 
-	return_tuple = result != NULL && tuple != NULL;
+	return_tuple = tuple != NULL;
 	if (return_tuple) {
 		/*
 		 * Pin the tuple locally before the commit,
@@ -2444,7 +2452,10 @@ box_select(uint32_t space_id, uint32_t index_id,
 	struct tuple *tuple;
 	port_c_create(port);
 	while (found < limit) {
+		struct result_processor res_proc;
+		result_process_prepare(&res_proc, space);
 		rc = iterator_next(it, &tuple);
+		result_process(&res_proc, &rc, &tuple);
 		if (rc != 0 || tuple == NULL)
 			break;
 		if (offset > 0) {
@@ -2455,6 +2466,11 @@ box_select(uint32_t space_id, uint32_t index_id,
 		if (rc != 0)
 			break;
 		found++;
+		/*
+		 * Refresh the pointer to the space, because the space struct
+		 * could be freed if the iterator yielded.
+		 */
+		space = iterator_space(it);
 	}
 	iterator_delete(it);
 

src/box/index.cc

@@ -34,6 +34,7 @@
 #include "schema.h"
 #include "user_def.h"
 #include "space.h"
+#include "result.h"
 #include "iproto_constants.h"
 #include "txn.h"
 #include "rmean.h"
@@ -216,7 +217,11 @@ box_index_random(uint32_t space_id, uint32_t index_id, uint32_t rnd,
 	if (check_index(space_id, index_id, &space, &index) != 0)
 		return -1;
 	/* No tx management, random() is for approximation anyway. */
-	if (index_random(index, rnd, result) != 0)
+	struct result_processor res_proc;
+	result_process_prepare(&res_proc, space);
+	int rc = index_random(index, rnd, result);
+	result_process(&res_proc, &rc, result);
+	if (rc != 0)
 		return -1;
 	if (*result != NULL)
 		tuple_bless(*result);
@@ -247,7 +252,11 @@ box_index_get(uint32_t space_id, uint32_t index_id, const char *key,
 	struct txn_ro_savepoint svp;
 	if (txn_begin_ro_stmt(space, &txn, &svp) != 0)
 		return -1;
-	if (index_get(index, key, part_count, result) != 0) {
+	struct result_processor res_proc;
+	result_process_prepare(&res_proc, space);
+	int rc = index_get(index, key, part_count, result);
+	result_process(&res_proc, &rc, result);
+	if (rc != 0) {
 		txn_rollback_stmt(txn);
 		return -1;
 	}
@@ -284,7 +293,11 @@ box_index_min(uint32_t space_id, uint32_t index_id, const char *key,
 	struct txn_ro_savepoint svp;
 	if (txn_begin_ro_stmt(space, &txn, &svp) != 0)
 		return -1;
-	if (index_min(index, key, part_count, result) != 0) {
+	struct result_processor res_proc;
+	result_process_prepare(&res_proc, space);
+	int rc = index_min(index, key, part_count, result);
+	result_process(&res_proc, &rc, result);
+	if (rc != 0) {
 		txn_rollback_stmt(txn);
 		return -1;
 	}
@@ -319,7 +332,11 @@ box_index_max(uint32_t space_id, uint32_t index_id, const char *key,
 	struct txn_ro_savepoint svp;
 	if (txn_begin_ro_stmt(space, &txn, &svp) != 0)
 		return -1;
-	if (index_max(index, key, part_count, result) != 0) {
+	struct result_processor res_proc;
+	result_process_prepare(&res_proc, space);
+	int rc = index_max(index, key, part_count, result);
+	result_process(&res_proc, &rc, result);
+	if (rc != 0) {
 		txn_rollback_stmt(txn);
 		return -1;
 	}
@@ -408,7 +425,16 @@ int
 box_iterator_next(box_iterator_t *itr, box_tuple_t **result)
 {
 	assert(result != NULL);
-	if (iterator_next(itr, result) != 0)
+	struct space *space = iterator_space(itr);
+	if (space == NULL) {
+		*result = NULL;
+		return 0;
+	}
+	struct result_processor res_proc;
+	result_process_prepare(&res_proc, space);
+	int rc = iterator_next(itr, result);
+	result_process(&res_proc, &rc, result);
+	if (rc != 0)
 		return -1;
 	if (*result != NULL)
 		tuple_bless(*result);

src/box/result.h

@@ -0,0 +1,53 @@
+/*
+ * SPDX-License-Identifier: BSD-2-Clause
+ *
+ * Copyright 2010-2022, Tarantool AUTHORS, please see AUTHORS file.
+ */
+#pragma once
+
+struct space;
+struct tuple;
+
+#if defined(__cplusplus)
+extern "C" {
+#endif /* defined(__cplusplus) */
+
+/**
+ * Helpers that apply various transformations to tuples fetched from a space.
+ * The procedure is split in two parts, because the read operation may yield,
+ * which opens a time window during which the space struct can be deleted.
+ * The 'prepare' phase is supposed to reference and store in result_processor
+ * all data structures needed to apply transforamtions.
+ *
+ * Used by box methods that read tuples from a space and return them to
+ * the user like this:
+ *
+ *   struct result_processor res_proc;
+ *   result_process_prepare(&res_proc, space);
+ *   rc = index_get(index, key, part_count, result);
+ *   result_process(&res_proc, &rc, result);
+ *
+ * Note, if result_process_prepare() was called, then result_process()
+ * must be called as well, because it may need to free some resources.
+ */
+struct result_processor {
+};
+
+static inline void
+result_process_prepare(struct result_processor *p, struct space *space)
+{
+	(void)p;
+	(void)space;
+}
+
+static inline void
+result_process(struct result_processor *p, int *rc, struct tuple **result)
+{
+	(void)p;
+	(void)rc;
+	(void)result;
+}
+
+#if defined(__cplusplus)
+} /* extern "C" */
+#endif /* defined(__cplusplus) */