[PATCH v3 6/7] box: introduce field_map_builder class

[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 extensions.

Needed for #1257
---
 src/box/CMakeLists.txt |   1 +
 src/box/field_map.c    |  61 ++++++++++++++++++
 src/box/field_map.h    | 139 +++++++++++++++++++++++++++++++++++++++++
 src/box/memtx_engine.c |   8 +--
 src/box/sql.c          |   5 +-
 src/box/tuple.c        |  16 ++---
 src/box/tuple.h        |   6 +-
 src/box/tuple_format.c |  48 ++++++--------
 src/box/tuple_format.h |  12 ++--
 src/box/vy_stmt.c      |   9 +--
 10 files changed, 252 insertions(+), 53 deletions(-)
 create mode 100644 src/box/field_map.c
 create mode 100644 src/box/field_map.h

diff --git a/src/box/CMakeLists.txt b/src/box/CMakeLists.txt
index 31600745a..11f568fca 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
+    field_map.c
     tuple_format.c
     tuple_update.c
     tuple_compare.cc
diff --git a/src/box/field_map.c b/src/box/field_map.c
new file mode 100644
index 000000000..690aa461d
--- /dev/null
+++ b/src/box/field_map.c
@@ -0,0 +1,61 @@
+/*
+ * 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 "field_map.h"
+#include "small/region.h"
+
+int
+field_map_builder_create(struct field_map_builder *builder,
+			 uint32_t minimal_field_map_size,
+			 struct region *region)
+{
+	builder->slot_count = minimal_field_map_size / sizeof(uint32_t);
+	if (minimal_field_map_size == 0) {
+		builder->slots = NULL;
+		return 0;
+	}
+	uint32_t sz = builder->slot_count * sizeof(builder->slots[0]);
+	builder->slots = region_alloc(region, sz);
+	if (builder->slots == NULL) {
+		diag_set(OutOfMemory, sz, "region_alloc", "field_map");
+		return -1;
+	}
+	memset((char *)builder->slots, 0, sz);
+	builder->slots = builder->slots + builder->slot_count;
+	return 0;
+}
+
+void
+field_map_build(struct field_map_builder *builder, char *buffer)
+{
+	uint32_t field_map_size = field_map_build_size(builder);
+	memcpy(buffer, (char *)builder->slots - field_map_size, field_map_size);
+}
diff --git a/src/box/field_map.h b/src/box/field_map.h
new file mode 100644
index 000000000..47ecbd009
--- /dev/null
+++ b/src/box/field_map.h
@@ -0,0 +1,139 @@
+#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 <assert.h>
+#include <stdint.h>
+
+struct region;
+
+/**
+ * A field map is a special area is reserved before tuple's
+ * MessagePack data. It is a sequence of the 32-bit unsigned
+ * offsets of tuple's indexed fields.
+ *
+ * These slots are numbered with negative indices called
+ * offset_slot(s) starting with -1 (this is necessary to organize
+ * the inheritance of tuples). Allocation and assignment of
+ * offset_slot(s) is performed on tuple_format creation on index
+ * create or alter (see tuple_format_create()).
+ *
+ *              4b          4b       MessagePack data.
+ *           +------+----+------+---------------------------+
+ *    tuple: | offN | .. | off1 | header ..|key1|..|keyN|.. |
+ *           +--+---+----+--+---+---------------------------+
+ *           |     ...    |                 ^       ^
+ *           |            +-----------------+       |
+ *           +--------------------------------------+
+ *
+ * This field_map_builder class is used for tuple field_map
+ * construction. It encapsulates field_map build logic and size
+ * estimation implementation-specific details.
+ *
+ * Each field offset is a positive number, except the case when
+ * a field is not in the tuple. In this case offset is 0.
+ */
+struct field_map_builder {
+	/**
+	 * The pointer to the end of filed_map allocation.
+	 * Its elements are accessible by negative indexes
+	 * that coinciding with offset_slot(s).
+	 */
+	uint32_t *slots;
+	/**
+	 * The count of slots in field_map_builder::slots
+	 * allocation.
+	 */
+	uint32_t slot_count;
+};
+
+/**
+ * Get offset of the field in tuple data MessagePack using
+ * tuple's field_map and required field's offset_slot.
+ *
+ * When a field is not in the data tuple, its offset is 0.
+ */
+static inline uint32_t
+field_map_get_offset(const uint32_t *field_map, int32_t offset_slot)
+{
+	return field_map[offset_slot];
+}
+
+/**
+ * Initialize field_map_builder.
+ *
+ * The field_map_size argument is a size of the minimal field_map
+ * allocation where each indexed field has own offset slot.
+ *
+ * Routine uses region to perform memory allocation for internal
+ * structures.
+ *
+ * Returns 0 on success. In case of memory allocation error sets
+ * diag message and returns -1.
+ */
+int
+field_map_builder_create(struct field_map_builder *builder,
+			 uint32_t minimal_field_map_size,
+			 struct region *region);
+
+/**
+ * Set data offset for a field identified by unique offset_slot.
+ *
+ * The offset_slot argument must be negative and offset must be
+ * positive (by definition).
+ */
+static inline void
+field_map_builder_set_slot(struct field_map_builder *builder,
+			   int32_t offset_slot, uint32_t offset)
+{
+	assert(offset_slot < 0);
+	assert(offset > 0);
+	builder->slots[offset_slot] = offset;
+}
+
+/**
+ * Calculate the size of tuple field_map to be built.
+ */
+static inline uint32_t
+field_map_build_size(struct field_map_builder *builder)
+{
+	return builder->slot_count * sizeof(uint32_t);
+}
+
+/**
+ * Write constructed field_map to the destination buffer field_map.
+ *
+ * The buffer must have at least field_map_build_size(builder) bytes.
+ */
+void
+field_map_build(struct field_map_builder *builder, char *buffer);
+
+#endif /* TARANTOOL_BOX_FIELD_MAP_H_INCLUDED */
diff --git a/src/box/memtx_engine.c b/src/box/memtx_engine.c
index 924f8bbc4..06da203e2 100644
--- a/src/box/memtx_engine.c
+++ b/src/box/memtx_engine.c
@@ -1111,10 +1111,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_build_size(&builder);
 
 	size_t tuple_len = end - data;
 	size_t total = sizeof(struct memtx_tuple) + field_map_size + tuple_len;
@@ -1154,7 +1154,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_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/sql.c b/src/box/sql.c
index 855c2b741..7d5c3a8e0 100644
--- a/src/box/sql.c
+++ b/src/box/sql.c
@@ -786,7 +786,10 @@ tarantoolsqlIdxKeyCompare(struct BtCursor *cursor,
 				while (j++ != fieldno)
 					mp_next(&p);
 			} else {
-				p = base + field_map[field->offset_slot];
+				uint32_t field_offset =
+					field_map_get_offset(field_map,
+							     field->offset_slot);
+				p = base + field_offset;
 			}
 		}
 		next_fieldno = fieldno + 1;
diff --git a/src/box/tuple.c b/src/box/tuple.c
index 7f06d4053..570e4c192 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_build_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_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_build_size(&builder) == format->field_map_size);
 	return rc;
 }
 
diff --git a/src/box/tuple.h b/src/box/tuple.h
index 8b12fd5a8..da4085bcf 100644
--- a/src/box/tuple.h
+++ b/src/box/tuple.h
@@ -542,6 +542,7 @@ tuple_field_raw_by_path(struct tuple_format *format, const char *tuple,
 		goto offset_slot_access;
 	}
 	if (likely(fieldno < format->index_field_count)) {
+		uint32_t offset;
 		struct tuple_field *field;
 		if (path == NULL && fieldno == 0) {
 			mp_decode_array(&tuple);
@@ -559,9 +560,10 @@ tuple_field_raw_by_path(struct tuple_format *format, const char *tuple,
 			*offset_slot_hint = offset_slot;
 offset_slot_access:
 		/* Indexed field */
-		if (field_map[offset_slot] == 0)
+		offset = field_map_get_offset(field_map, offset_slot);
+		if (offset == 0)
 			return NULL;
-		tuple += field_map[offset_slot];
+		tuple += offset;
 	} else {
 		uint32_t field_count;
 parse:
diff --git a/src/box/tuple_format.c b/src/box/tuple_format.c
index 070897ec2..9a643b700 100644
--- a/src/box/tuple_format.c
+++ b/src/box/tuple_format.c
@@ -769,27 +769,21 @@ tuple_format1_can_store_format2_tuples(struct tuple_format *format1,
 
 /** @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)
+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) {
-		*field_map = NULL;
-		*field_map_size = 0;
+		/* Nothing to initialize */
+		if (field_map_builder_create(builder, 0, region) != 0)
+			unreachable();
 		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");
+	if (field_map_builder_create(builder, format->field_map_size,
+				     region) != 0)
 		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 &&
@@ -797,7 +791,7 @@ tuple_field_map_create(struct tuple_format *format, const char *tuple,
 		diag_set(ClientError, ER_EXACT_FIELD_COUNT,
 			 (unsigned) field_count,
 			 (unsigned) format->exact_field_count);
-		goto error;
+		return -1;
 	}
 	/*
 	 * Allocate a field bitmap that will be used for checking
@@ -811,7 +805,7 @@ tuple_field_map_create(struct tuple_format *format, const char *tuple,
 		if (required_fields == NULL) {
 			diag_set(OutOfMemory, required_fields_sz,
 				 "region", "required field bitmap");
-			goto error;
+			return -1;
 		}
 		memcpy(required_fields, format->required_fields,
 		       required_fields_sz);
@@ -830,11 +824,10 @@ tuple_field_map_create(struct tuple_format *format, const char *tuple,
 	 * 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);
 	struct tuple_parse_iterator it;
 	if (tuple_parse_iterator_create(&it, format, pos, defined_field_count,
 					region) != 0)
-		goto error;
+		return -1;
 	const char *pos_end;
 	struct tuple_field *field;
 	while (tuple_parse_iterator_advice(&it, &field, &pos, &pos_end) > 0) {
@@ -851,11 +844,13 @@ 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;
 		}
 		/* Initialize field_map with data offset. */
-		if (field->offset_slot != TUPLE_OFFSET_SLOT_NIL)
-			(*field_map)[field->offset_slot] = pos - tuple;
+		if (field->offset_slot != TUPLE_OFFSET_SLOT_NIL) {
+			field_map_builder_set_slot(builder, field->offset_slot,
+						   pos - tuple);
+		}
 		/* Mark this field as present in the tuple. */
 		if (required_fields != NULL)
 			bit_clear(required_fields, field->id);
@@ -875,15 +870,10 @@ finish:
 			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 bef1d0903..88f03d5eb 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 "field_map.h"
 
 #if defined(__cplusplus)
 extern "C" {
@@ -169,8 +170,9 @@ struct tuple_format {
 	 */
 	bool is_ephemeral;
 	/**
-	 * Size of field map of tuple in bytes.
-	 * \sa struct tuple
+	 * Size of minimal field map of tuple where each indexed
+	 * field has own offset slot (in bytes).
+	 * \sa struct field_map_builder
 	 */
 	uint16_t field_map_size;
 	/**
@@ -401,9 +403,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 1e8bb7825..3e35788ac 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_build_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_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