[PATCH v5 2/4] 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    | 140 +++++++++++++++++++++++++++++++++++++++++
 src/box/memtx_engine.c |   8 +--
 src/box/sql.c          |   5 +-
 src/box/tuple.c        |  14 ++---
 src/box/tuple.h        |   6 +-
 src/box/tuple_format.c |  30 +++------
 src/box/tuple_format.h |  12 ++--
 src/box/vy_stmt.c      |   9 +--
 10 files changed, 238 insertions(+), 48 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..f6c653030
--- /dev/null
+++ b/src/box/field_map.h
@@ -0,0 +1,140 @@
+#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 field_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((uint32_t)-offset_slot <= builder->slot_count);
+	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 4d99910cb..210f3c22c 100644
--- a/src/box/memtx_engine.c
+++ b/src/box/memtx_engine.c
@@ -1135,10 +1135,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(format, data, true, &builder) != 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;
@@ -1178,7 +1178,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 1fb93e106..2310ee5e3 100644
--- a/src/box/sql.c
+++ b/src/box/sql.c
@@ -780,7 +780,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 9d1768440..c325f58c9 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(format, data, true, &builder) != 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,10 +126,8 @@ 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(format, tuple, true, &builder);
 	region_truncate(region, region_svp);
 	return rc;
 }
diff --git a/src/box/tuple.h b/src/box/tuple.h
index eed8e1a34..d261d4cc9 100644
--- a/src/box/tuple.h
+++ b/src/box/tuple.h
@@ -540,6 +540,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);
@@ -557,9 +558,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 de48e9bb0..95eb9c826 100644
--- a/src/box/tuple_format.c
+++ b/src/box/tuple_format.c
@@ -773,28 +773,15 @@ 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)
+		       bool validate, struct field_map_builder *builder)
 {
-	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");
+	if (field_map_builder_create(builder, format->field_map_size,
+				     region) != 0)
 		return -1;
-	}
-	*field_map = (uint32_t *)((char *)*field_map + *field_map_size);
-	/*
-	 * 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);
+	if (tuple_format_field_count(format) == 0)
+		return 0; /* Nothing to initialize */
+
 	uint32_t field_count;
 	struct tuple_format_iterator it;
 	uint8_t flags = validate ? TUPLE_FORMAT_ITERATOR_VALIDATE : 0;
@@ -807,11 +794,10 @@ tuple_field_map_create(struct tuple_format *format, const char *tuple,
 		if (entry.field == NULL)
 			continue;
 		if (entry.field->offset_slot != TUPLE_OFFSET_SLOT_NIL) {
-			(*field_map)[entry.field->offset_slot] =
-							entry.data - tuple;
+		    field_map_builder_set_slot(builder, entry.field->offset_slot,
+		    			       entry.data - tuple);
 		}
 	}
-	*field_map = (uint32_t *)((char *)*field_map - *field_map_size);
 	return entry.data == NULL ? 0 : -1;
 }
 
diff --git a/src/box/tuple_format.h b/src/box/tuple_format.h
index 244f60266..ea8721ea8 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" {
@@ -170,7 +171,7 @@ struct tuple_format {
 	bool is_ephemeral;
 	/**
 	 * Size of field map of tuple in bytes.
-	 * \sa struct tuple
+	 * \sa struct field_map_builder
 	 */
 	uint16_t field_map_size;
 	/**
@@ -386,10 +387,8 @@ box_tuple_format_unref(box_tuple_format_t *format);
  * @param format    Tuple format.
  * @param tuple     MessagePack array.
  * @param validate  If set, validate the tuple against the format.
- * @param field_map[out] The pointer to store field map
- *                       allocation.
- * @param field_map_size[out] The pointer to variable to store
- *                            field map size.
+ * @param builder[out] The pointer to field map builder object to
+ *                     be prepared.
  *
  * @retval  0 Success.
  * @retval -1 Format error.
@@ -402,8 +401,7 @@ box_tuple_format_unref(box_tuple_format_t *format);
  */
 int
 tuple_field_map_create(struct tuple_format *format, const char *tuple,
-		       bool validate, uint32_t **field_map,
-		       uint32_t *field_map_size);
+		       bool validate, struct field_map_builder *builder);
 
 /**
  * Initialize tuple format subsystem.
diff --git a/src/box/vy_stmt.c b/src/box/vy_stmt.c
index 8d46a9eac..c8088d349 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(format, tuple_begin, false, &builder) != 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