[PATCH v2 4/5] box: introduce field_map_builder for field_map init

[Kirill Shcherbatov <kshcherbatov@tarantool.org>]

The new field_map_builder class encapsulates the logic associated with
field_map allocation and initialization. In the future it will
be extended to allocate field_map that has extentions.

Needed for #1257
---
 src/box/CMakeLists.txt    |   1 +
 src/box/memtx_engine.c    |   8 +-
 src/box/tuple.c           |  16 ++--
 src/box/tuple_field_map.c |  53 ++++++++++++
 src/box/tuple_field_map.h |  86 ++++++++++++++++++
 src/box/tuple_format.c    | 178 +++++++++++++++++++-------------------
 src/box/tuple_format.h    |   7 +-
 src/box/vy_stmt.c         |   9 +-
 8 files changed, 252 insertions(+), 106 deletions(-)
 create mode 100644 src/box/tuple_field_map.c
 create mode 100644 src/box/tuple_field_map.h

diff --git a/src/box/CMakeLists.txt b/src/box/CMakeLists.txt
index 59e91b65a..3ae47e336 100644
--- a/src/box/CMakeLists.txt
+++ b/src/box/CMakeLists.txt
@@ -35,6 +35,7 @@ target_link_libraries(xrow server core small vclock misc box_error
 
 add_library(tuple STATIC
     tuple.c
+    tuple_field_map.c
     tuple_format.c
     tuple_update.c
     tuple_compare.cc
diff --git a/src/box/memtx_engine.c b/src/box/memtx_engine.c
index 3e959ff2e..bcd170a89 100644
--- a/src/box/memtx_engine.c
+++ b/src/box/memtx_engine.c
@@ -1125,10 +1125,10 @@ memtx_tuple_new(struct tuple_format *format, const char *data, const char *end)
 	struct tuple *tuple = NULL;
 	struct region *region = &fiber()->gc;
 	size_t region_svp = region_used(region);
-	uint32_t *field_map, field_map_size;
-	if (tuple_field_map_create(format, data, true, &field_map,
-				   &field_map_size) != 0)
+	struct field_map_builder builder;
+	if (tuple_field_map_create(&builder, format, data, true) != 0)
 		goto end;
+	uint32_t field_map_size = field_map_builder_size(&builder);
 
 	size_t tuple_len = end - data;
 	size_t total = sizeof(struct memtx_tuple) + field_map_size + tuple_len;
@@ -1168,7 +1168,7 @@ memtx_tuple_new(struct tuple_format *format, const char *data, const char *end)
 	 */
 	tuple->data_offset = sizeof(struct tuple) + field_map_size;
 	char *raw = (char *) tuple + tuple->data_offset;
-	memcpy(raw - field_map_size, field_map, field_map_size);
+	field_map_builder_build(&builder, raw - field_map_size);
 	memcpy(raw, data, tuple_len);
 	say_debug("%s(%zu) = %p", __func__, tuple_len, memtx_tuple);
 end:
diff --git a/src/box/tuple.c b/src/box/tuple.c
index 7f06d4053..68f0670f2 100644
--- a/src/box/tuple.c
+++ b/src/box/tuple.c
@@ -78,10 +78,10 @@ runtime_tuple_new(struct tuple_format *format, const char *data, const char *end
 	struct tuple *tuple = NULL;
 	struct region *region = &fiber()->gc;
 	size_t region_svp = region_used(region);
-	uint32_t *field_map, field_map_size;
-	if (tuple_field_map_create(format, data, true, &field_map,
-				   &field_map_size) != 0)
+	struct field_map_builder builder;
+	if (tuple_field_map_create(&builder, format, data, true) != 0)
 		goto end;
+	uint32_t field_map_size = field_map_builder_size(&builder);
 
 	size_t data_len = end - data;
 	size_t total = sizeof(struct tuple) + field_map_size + data_len;
@@ -98,7 +98,7 @@ runtime_tuple_new(struct tuple_format *format, const char *data, const char *end
 	tuple_format_ref(format);
 	tuple->data_offset = sizeof(struct tuple) + field_map_size;
 	char *raw = (char *) tuple + tuple->data_offset;
-	memcpy(raw - field_map_size, field_map, field_map_size);
+	field_map_builder_build(&builder, raw - field_map_size);
 	memcpy(raw, data, data_len);
 	say_debug("%s(%zu) = %p", __func__, data_len, tuple);
 end:
@@ -126,11 +126,11 @@ tuple_validate_raw(struct tuple_format *format, const char *tuple)
 
 	struct region *region = &fiber()->gc;
 	size_t region_svp = region_used(region);
-	uint32_t *field_map, field_map_size;
-	int rc = tuple_field_map_create(format, tuple, true, &field_map,
-					&field_map_size);
-	assert(rc != 0 || field_map_size == format->field_map_size);
+	struct field_map_builder builder;
+	int rc = tuple_field_map_create(&builder, format, tuple, true);
 	region_truncate(region, region_svp);
+	assert(rc != 0 ||
+	       field_map_builder_size(&builder) == format->field_map_size);
 	return rc;
 }
 
diff --git a/src/box/tuple_field_map.c b/src/box/tuple_field_map.c
new file mode 100644
index 000000000..ea77745aa
--- /dev/null
+++ b/src/box/tuple_field_map.c
@@ -0,0 +1,53 @@
+/*
+ * Copyright 2010-2019, Tarantool AUTHORS, please see AUTHORS file.
+ *
+ * Redistribution and use in source and binary forms, with or
+ * without modification, are permitted provided that the following
+ * conditions are met:
+ *
+ * 1. Redistributions of source code must retain the above
+ *    copyright notice, this list of conditions and the
+ *    following disclaimer.
+ *
+ * 2. Redistributions in binary form must reproduce the above
+ *    copyright notice, this list of conditions and the following
+ *    disclaimer in the documentation and/or other materials
+ *    provided with the distribution.
+ *
+ * THIS SOFTWARE IS PROVIDED BY <COPYRIGHT HOLDER> ``AS IS'' AND
+ * ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED
+ * TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
+ * A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL
+ * <COPYRIGHT HOLDER> OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT,
+ * INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL
+ * DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF
+ * SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR
+ * BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF
+ * LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
+ * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF
+ * THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF
+ * SUCH DAMAGE.
+ */
+#include "diag.h"
+#include "tuple_field_map.h"
+#include "small/region.h"
+
+int
+field_map_builder_create(struct field_map_builder *builder,
+			 uint32_t field_map_size, struct region *region)
+{
+	builder->item_count = field_map_size / sizeof(uint32_t);
+	if (field_map_size == 0) {
+		builder->items = NULL;
+		return 0;
+	}
+	uint32_t sz = builder->item_count * sizeof(field_map_builder_item);
+	builder->items = region_alloc(region, sz);
+	if (builder->items == NULL) {
+		diag_set(OutOfMemory, sz, "region_alloc", "field_map");
+		return -1;
+	}
+	memset((char *)builder->items, 0, sz);
+	builder->items = (field_map_builder_item *)((char *)builder->items + sz);
+	return 0;
+}
diff --git a/src/box/tuple_field_map.h b/src/box/tuple_field_map.h
new file mode 100644
index 000000000..3a9927faa
--- /dev/null
+++ b/src/box/tuple_field_map.h
@@ -0,0 +1,86 @@
+#ifndef TARANTOOL_BOX_FIELD_MAP_H_INCLUDED
+#define TARANTOOL_BOX_FIELD_MAP_H_INCLUDED
+/*
+ * Copyright 2010-2019, Tarantool AUTHORS, please see AUTHORS file.
+ *
+ * Redistribution and use in source and binary forms, with or
+ * without modification, are permitted provided that the following
+ * conditions are met:
+ *
+ * 1. Redistributions of source code must retain the above
+ *    copyright notice, this list of conditions and the
+ *    following disclaimer.
+ *
+ * 2. Redistributions in binary form must reproduce the above
+ *    copyright notice, this list of conditions and the following
+ *    disclaimer in the documentation and/or other materials
+ *    provided with the distribution.
+ *
+ * THIS SOFTWARE IS PROVIDED BY <COPYRIGHT HOLDER> ``AS IS'' AND
+ * ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED
+ * TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
+ * A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL
+ * <COPYRIGHT HOLDER> OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT,
+ * INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL
+ * DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF
+ * SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR
+ * BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF
+ * LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
+ * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF
+ * THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF
+ * SUCH DAMAGE.
+ */
+#include <stdint.h>
+
+struct region;
+
+/** Preliminary field_map atom. */
+typedef uint32_t field_map_builder_item;
+
+/** A class that contains a preliminary view of the field_map. */
+struct field_map_builder {
+	/**
+	 * Allocation with preliminary offset slot
+	 * representation.
+	*/
+	field_map_builder_item *items;
+	/**
+	 * Number of fields in the resulting field_map;
+	 * count of field_map_builder::items.
+	 */
+	uint32_t item_count;
+};
+
+/**
+ * Initialize field_map_builder to prepare field_map of size
+ * field_map_size. Use region for temporary allocations.
+ */
+int
+field_map_builder_create(struct field_map_builder *builder,
+			 uint32_t field_map_size, struct region *region);
+
+/** Initialize offset_slot with offset value. */
+static inline void
+field_map_builder_slot_set(struct field_map_builder *builder,
+			   int32_t offset_slot, uint32_t offset)
+{
+	assert(offset_slot < 0);
+	builder->items[offset_slot] = offset;
+}
+
+/** Return built field_map size. */
+static inline uint32_t
+field_map_builder_size(struct field_map_builder *builder)
+{
+	return builder->item_count * sizeof(uint32_t);
+}
+
+/** Write build field_map size in the preallocated buffer wptr. */
+static inline void
+field_map_builder_build(struct field_map_builder *builder, char *wptr)
+{
+	uint32_t field_map_size = field_map_builder_size(builder);
+	memcpy(wptr, (char *)builder->items - field_map_size, field_map_size);
+}
+
+#endif /* TARANTOOL_BOX_FIELD_MAP_H_INCLUDED */
diff --git a/src/box/tuple_format.c b/src/box/tuple_format.c
index 1043707ad..1d765c7c7 100644
--- a/src/box/tuple_format.c
+++ b/src/box/tuple_format.c
@@ -767,70 +767,13 @@ tuple_format1_can_store_format2_tuples(struct tuple_format *format1,
 	return true;
 }
 
-/** @sa declaration for details. */
-int
-tuple_field_map_create(struct tuple_format *format, const char *tuple,
-		       bool validate, uint32_t **field_map,
-		       uint32_t *field_map_size)
+static int
+tuple_field_map_initialize(struct field_map_builder *builder,
+			   struct tuple_format *format, const char **pos,
+			   const char *tuple, uint32_t defined_field_count,
+			   void *required_fields, struct region *region)
 {
-	if (tuple_format_field_count(format) == 0) {
-		*field_map = NULL;
-		*field_map_size = 0;
-		return 0; /* Nothing to initialize */
-	}
-	struct region *region = &fiber()->gc;
-	*field_map_size = format->field_map_size;
-	*field_map = region_alloc(region, *field_map_size);
-	if (*field_map == NULL) {
-		diag_set(OutOfMemory, *field_map_size, "region_alloc",
-			 "field_map");
-		return -1;
-	}
-	*field_map = (uint32_t *)((char *)*field_map + *field_map_size);
-
-	const char *pos = tuple;
-	int rc = 0;
-	/* Check to see if the tuple has a sufficient number of fields. */
-	uint32_t field_count = mp_decode_array(&pos);
-	if (validate && format->exact_field_count > 0 &&
-	    format->exact_field_count != field_count) {
-		diag_set(ClientError, ER_EXACT_FIELD_COUNT,
-			 (unsigned) field_count,
-			 (unsigned) format->exact_field_count);
-		goto error;
-	}
-	/*
-	 * Allocate a field bitmap that will be used for checking
-	 * that all mandatory fields are present.
-	 */
-	void *required_fields = NULL;
-	size_t required_fields_sz = 0;
-	if (validate) {
-		required_fields_sz = bitmap_size(format->total_field_count);
-		required_fields = region_alloc(region, required_fields_sz);
-		if (required_fields == NULL) {
-			diag_set(OutOfMemory, required_fields_sz,
-				 "region", "required field bitmap");
-			goto error;
-		}
-		memcpy(required_fields, format->required_fields,
-		       required_fields_sz);
-	}
-	/*
-	 * Initialize the tuple field map and validate field types.
-	 */
-	if (field_count == 0) {
-		/* Empty tuple, nothing to do. */
-		goto finish;
-	}
-	uint32_t defined_field_count = MIN(field_count, validate ?
-					   tuple_format_field_count(format) :
-					   format->index_field_count);
-	/*
-	 * Nullify field map to be able to detect by 0,
-	 * which key fields are absent in tuple_field().
-	 */
-	memset((char *)*field_map - *field_map_size, 0, *field_map_size);
+	bool validate = required_fields != NULL;
 	/*
 	 * Prepare mp stack of the size equal to the maximum depth
 	 * of the indexed field in the format::fields tree
@@ -842,12 +785,11 @@ tuple_field_map_create(struct tuple_format *format, const char *tuple,
 	struct mp_frame *frames = region_alloc(region, frames_sz);
 	if (frames == NULL) {
 		diag_set(OutOfMemory, frames_sz, "region", "frames");
-		goto error;
+		return -1;
 	}
 	struct mp_stack stack;
 	mp_stack_create(&stack, format->fields_depth, frames);
 	mp_stack_push(&stack, MP_ARRAY, defined_field_count);
-	struct tuple_field *field;
 	struct json_token *parent = &format->fields.root;
 	while (true) {
 		int idx;
@@ -862,7 +804,7 @@ tuple_field_map_create(struct tuple_format *format, const char *tuple,
 			 */
 			mp_stack_pop(&stack);
 			if (mp_stack_is_empty(&stack))
-				goto finish;
+				goto end;
 			parent = parent->parent;
 		}
 		/*
@@ -877,17 +819,17 @@ tuple_field_map_create(struct tuple_format *format, const char *tuple,
 			token.num = idx;
 			break;
 		case MP_MAP:
-			if (mp_typeof(*pos) != MP_STR) {
+			if (mp_typeof(**pos) != MP_STR) {
 				/*
 				 * JSON path support only string
 				 * keys for map. Skip this entry.
 				 */
-				mp_next(&pos);
-				mp_next(&pos);
+				mp_next(pos);
+				mp_next(pos);
 				continue;
 			}
 			token.type = JSON_TOKEN_STR;
-			token.str = mp_decode_str(&pos, (uint32_t *)&token.len);
+			token.str = mp_decode_str(pos, (uint32_t *)&token.len);
 			break;
 		default:
 			unreachable();
@@ -898,9 +840,10 @@ tuple_field_map_create(struct tuple_format *format, const char *tuple,
 		 * corresponding to the current position in the
 		 * tuple.
 		 */
-		enum mp_type type = mp_typeof(*pos);
+		enum mp_type type = mp_typeof(**pos);
 		assert(parent != NULL);
-		field = json_tree_lookup_entry(&format->fields, parent, &token,
+		struct tuple_field *field =
+			json_tree_lookup_entry(&format->fields, parent, &token,
 					       struct tuple_field, token);
 		if (field != NULL) {
 			bool is_nullable = tuple_field_is_nullable(field);
@@ -910,11 +853,14 @@ tuple_field_map_create(struct tuple_format *format, const char *tuple,
 				diag_set(ClientError, ER_FIELD_TYPE,
 					 tuple_field_path(field),
 					 field_type_strs[field->type]);
-				goto error;
+				return -1;
 			}
-			if (field->offset_slot != TUPLE_OFFSET_SLOT_NIL)
-				(*field_map)[field->offset_slot] = pos - tuple;
-			if (required_fields != NULL)
+			if (field->offset_slot != TUPLE_OFFSET_SLOT_NIL) {
+				field_map_builder_slot_set(builder,
+							   field->offset_slot,
+							   *pos - tuple);
+			}
+			if (validate)
 				bit_clear(required_fields, field->id);
 		}
 		/*
@@ -927,14 +873,76 @@ tuple_field_map_create(struct tuple_format *format, const char *tuple,
 		if ((type == MP_ARRAY || type == MP_MAP) &&
 		    !mp_stack_is_full(&stack) && field != NULL) {
 			uint32_t size = type == MP_ARRAY ?
-					mp_decode_array(&pos) :
-					mp_decode_map(&pos);
+					mp_decode_array(pos) :
+					mp_decode_map(pos);
 			mp_stack_push(&stack, type, size);
 			parent = &field->token;
 		} else {
-			mp_next(&pos);
+			mp_next(pos);
 		}
 	}
+end:
+	return 0;
+}
+
+/** @sa declaration for details. */
+int
+tuple_field_map_create(struct field_map_builder *builder,
+		       struct tuple_format *format, const char *tuple,
+		       bool validate)
+{
+	struct region *region = &fiber()->gc;
+	if (tuple_format_field_count(format) == 0) {
+		/* Nothing to initialize */
+		if (field_map_builder_create(builder, 0, region) != 0)
+			unreachable();
+		return 0;
+	}
+	if (field_map_builder_create(builder, format->field_map_size,
+				     region) != 0)
+		return -1;
+	const char *pos = tuple;
+	/* Check to see if the tuple has a sufficient number of fields. */
+	uint32_t field_count = mp_decode_array(&pos);
+	if (validate && format->exact_field_count > 0 &&
+	    format->exact_field_count != field_count) {
+		diag_set(ClientError, ER_EXACT_FIELD_COUNT,
+			 (unsigned) field_count,
+			 (unsigned) format->exact_field_count);
+		return -1;
+	}
+	/*
+	 * Allocate a field bitmap that will be used for checking
+	 * that all mandatory fields are present.
+	 */
+	void *required_fields = NULL;
+	size_t required_fields_sz = 0;
+	if (validate) {
+		required_fields_sz = bitmap_size(format->total_field_count);
+		required_fields = region_alloc(region, required_fields_sz);
+		if (required_fields == NULL) {
+			diag_set(OutOfMemory, required_fields_sz,
+				 "region", "required field bitmap");
+			return -1;
+		}
+		memcpy(required_fields, format->required_fields,
+		       required_fields_sz);
+	}
+	/*
+	 * Initialize the tuple field map and validate field types.
+	 */
+	if (field_count == 0) {
+		/* Empty tuple, nothing to do. */
+		goto finish;
+	}
+	uint32_t defined_field_count = MIN(field_count, validate ?
+					   tuple_format_field_count(format) :
+					   format->index_field_count);
+	assert(!validate || required_fields != NULL);
+	if (tuple_field_map_initialize(builder, format, &pos, tuple,
+				       defined_field_count, required_fields,
+				       region) != 0)
+		return -1;
 finish:
 	/*
 	 * Check the required field bitmap for missing fields.
@@ -946,19 +954,15 @@ finish:
 		size_t id = bit_iterator_next(&it);
 		if (id < SIZE_MAX) {
 			/* A field is missing, report an error. */
-			field = tuple_format_field_by_id(format, id);
+			struct tuple_field *field =
+				tuple_format_field_by_id(format, id);
 			assert(field != NULL);
 			diag_set(ClientError, ER_FIELD_MISSING,
 				 tuple_field_path(field));
-			goto error;
+			return -1;
 		}
 	}
-out:
-	*field_map = (uint32_t *)((char *)*field_map - *field_map_size);
-	return rc;
-error:
-	rc = -1;
-	goto out;
+	return 0;
 }
 
 uint32_t
diff --git a/src/box/tuple_format.h b/src/box/tuple_format.h
index 22a0fb232..6874374cc 100644
--- a/src/box/tuple_format.h
+++ b/src/box/tuple_format.h
@@ -36,6 +36,7 @@
 #include "errinj.h"
 #include "json/json.h"
 #include "tuple_dictionary.h"
+#include "tuple_field_map.h"
 
 #if defined(__cplusplus)
 extern "C" {
@@ -401,9 +402,9 @@ box_tuple_format_unref(box_tuple_format_t *format);
  * tuple + off_i = indexed_field_i;
  */
 int
-tuple_field_map_create(struct tuple_format *format, const char *tuple,
-		       bool validate, uint32_t **field_map,
-		       uint32_t *field_map_size);
+tuple_field_map_create(struct field_map_builder *builder,
+		       struct tuple_format *format, const char *tuple,
+		       bool validate);
 
 /**
  * Initialize tuple format subsystem.
diff --git a/src/box/vy_stmt.c b/src/box/vy_stmt.c
index 5d1e10f07..e355e3650 100644
--- a/src/box/vy_stmt.c
+++ b/src/box/vy_stmt.c
@@ -314,10 +314,11 @@ vy_stmt_new_with_ops(struct tuple_format *format, const char *tuple_begin,
 	 * tuples inserted into a space are validated explicitly
 	 * with tuple_validate() anyway.
 	 */
-	uint32_t *field_map, field_map_size;
-	if (tuple_field_map_create(format, tuple_begin, false, &field_map,
-				   &field_map_size) != 0)
+	struct field_map_builder builder;
+	if (tuple_field_map_create(&builder, format, tuple_begin, false) != 0)
 		goto end;
+	uint32_t field_map_size = field_map_builder_size(&builder);
+	assert(field_map_size == format->field_map_size);
 	/*
 	 * Allocate stmt. Offsets: one per key part + offset of the
 	 * statement end.
@@ -330,7 +331,7 @@ vy_stmt_new_with_ops(struct tuple_format *format, const char *tuple_begin,
 	/* Copy MsgPack data */
 	char *raw = (char *) tuple_data(stmt);
 	char *wpos = raw;
-	memcpy(wpos - field_map_size, field_map, field_map_size);
+	field_map_builder_build(&builder, wpos - field_map_size);
 	memcpy(wpos, tuple_begin, mpsize);
 	wpos += mpsize;
 	for (struct iovec *op = ops, *end = ops + op_count;
-- 
2.21.0