[PATCH v5 06/12] box: manage format fields with JSON tree class

[Kirill Shcherbatov <kshcherbatov@tarantool.org>]

As we going to work with format fields in a unified way, we
started to use the tree JSON class for working with first-level
format fields.

Need for #1012
---
 src/box/sql.c          |  16 +++----
 src/box/sql/build.c    |   5 +-
 src/box/tuple.c        |  10 ++--
 src/box/tuple_format.c | 121 ++++++++++++++++++++++++++++++++++---------------
 src/box/tuple_format.h |  32 ++++++++++---
 src/box/vy_stmt.c      |   4 +-
 test-run               |   2 +-
 7 files changed, 130 insertions(+), 60 deletions(-)

diff --git a/src/box/sql.c b/src/box/sql.c
index c7b87e5..3d08936 100644
--- a/src/box/sql.c
+++ b/src/box/sql.c
@@ -197,7 +197,8 @@ tarantoolSqlite3TupleColumnFast(BtCursor *pCur, u32 fieldno, u32 *field_size)
 	struct tuple_format *format = tuple_format(pCur->last_tuple);
 	assert(format->exact_field_count == 0
 	       || fieldno < format->exact_field_count);
-	if (format->fields[fieldno].offset_slot == TUPLE_OFFSET_SLOT_NIL)
+	if (tuple_format_field(format, fieldno)->offset_slot ==
+	    TUPLE_OFFSET_SLOT_NIL)
 		return NULL;
 	const char *field = tuple_field(pCur->last_tuple, fieldno);
 	const char *end = field;
@@ -891,7 +892,7 @@ tarantoolSqlite3IdxKeyCompare(struct BtCursor *cursor,
 	struct key_def *key_def;
 	const struct tuple *tuple;
 	const char *base;
-	const struct tuple_format *format;
+	struct tuple_format *format;
 	const uint32_t *field_map;
 	uint32_t field_count, next_fieldno = 0;
 	const char *p, *field0;
@@ -909,7 +910,7 @@ tarantoolSqlite3IdxKeyCompare(struct BtCursor *cursor,
 	base = tuple_data(tuple);
 	format = tuple_format(tuple);
 	field_map = tuple_field_map(tuple);
-	field_count = format->field_count;
+	field_count = tuple_format_field_count(format);
 	field0 = base; mp_decode_array(&field0); p = field0;
 	for (i = 0; i < n; i++) {
 		/*
@@ -927,9 +928,10 @@ tarantoolSqlite3IdxKeyCompare(struct BtCursor *cursor,
 		uint32_t fieldno = key_def->parts[i].fieldno;
 
 		if (fieldno != next_fieldno) {
+			struct tuple_field *field =
+				tuple_format_field(format, fieldno);
 			if (fieldno >= field_count ||
-			    format->fields[fieldno].offset_slot ==
-			    TUPLE_OFFSET_SLOT_NIL) {
+			    field->offset_slot == TUPLE_OFFSET_SLOT_NIL) {
 				/* Outdated field_map. */
 				uint32_t j = 0;
 
@@ -937,9 +939,7 @@ tarantoolSqlite3IdxKeyCompare(struct BtCursor *cursor,
 				while (j++ != fieldno)
 					mp_next(&p);
 			} else {
-				p = base + field_map[
-					format->fields[fieldno].offset_slot
-					];
+				p = base + field_map[field->offset_slot];
 			}
 		}
 		next_fieldno = fieldno + 1;
diff --git a/src/box/sql/build.c b/src/box/sql/build.c
index 7556a78..83e89bb 100644
--- a/src/box/sql/build.c
+++ b/src/box/sql/build.c
@@ -864,8 +864,9 @@ sql_column_collation(struct space_def *def, uint32_t column, uint32_t *coll_id)
 		struct coll_id *collation = coll_by_id(*coll_id);
 		return collation != NULL ? collation->coll : NULL;
 	}
-	*coll_id = space->format->fields[column].coll_id;
-	return space->format->fields[column].coll;
+	struct tuple_field *field = tuple_format_field(space->format, column);
+	*coll_id = field->coll_id;
+	return field->coll;
 }
 
 struct ExprList *
diff --git a/src/box/tuple.c b/src/box/tuple.c
index ef4d16f..aae1c3c 100644
--- a/src/box/tuple.c
+++ b/src/box/tuple.c
@@ -138,7 +138,7 @@ runtime_tuple_delete(struct tuple_format *format, struct tuple *tuple)
 int
 tuple_validate_raw(struct tuple_format *format, const char *tuple)
 {
-	if (format->field_count == 0)
+	if (tuple_format_field_count(format) == 0)
 		return 0; /* Nothing to check */
 
 	/* Check to see if the tuple has a sufficient number of fields. */
@@ -158,10 +158,12 @@ tuple_validate_raw(struct tuple_format *format, const char *tuple)
 	}
 
 	/* Check field types */
-	struct tuple_field *field = &format->fields[0];
+	struct tuple_field *field = tuple_format_field(format, 0);
 	uint32_t i = 0;
-	uint32_t defined_field_count = MIN(field_count, format->field_count);
-	for (; i < defined_field_count; ++i, ++field) {
+	uint32_t defined_field_count =
+		MIN(field_count, tuple_format_field_count(format));
+	for (; i < defined_field_count; ++i) {
+		field = tuple_format_field(format, i);
 		if (key_mp_type_validate(field->type, mp_typeof(*tuple),
 					 ER_FIELD_TYPE, i + TUPLE_INDEX_BASE,
 					 tuple_field_is_nullable(field)))
diff --git a/src/box/tuple_format.c b/src/box/tuple_format.c
index 088579c..9ffb5e4 100644
--- a/src/box/tuple_format.c
+++ b/src/box/tuple_format.c
@@ -38,10 +38,29 @@ static intptr_t recycled_format_ids = FORMAT_ID_NIL;
 
 static uint32_t formats_size = 0, formats_capacity = 0;
 
-static const struct tuple_field tuple_field_default = {
-	FIELD_TYPE_ANY, TUPLE_OFFSET_SLOT_NIL, false,
-	ON_CONFLICT_ACTION_DEFAULT, NULL, COLL_NONE,
-};
+static struct tuple_field *
+tuple_field_create(struct json_path_node *node)
+{
+	struct tuple_field *ret = calloc(1, sizeof(struct tuple_field));
+	if (ret == NULL) {
+		diag_set(OutOfMemory, sizeof(struct tuple_field), "malloc",
+			 "ret");
+		return NULL;
+	}
+	ret->type = FIELD_TYPE_ANY;
+	ret->offset_slot = TUPLE_OFFSET_SLOT_NIL;
+	ret->coll_id = COLL_NONE;
+	ret->nullable_action = ON_CONFLICT_ACTION_DEFAULT;
+	json_tree_node_create(&ret->tree_node, node);
+	return ret;
+}
+
+static void
+tuple_field_destroy(struct tuple_field *field)
+{
+	json_tree_node_destroy(&field->tree_node);
+	free(field);
+}
 
 static int
 tuple_format_add_key_part(struct tuple_format *format,
@@ -49,8 +68,8 @@ tuple_format_add_key_part(struct tuple_format *format,
 			  const struct key_part *part, bool is_sequential,
 			  int *current_slot)
 {
-	assert(part->fieldno < format->field_count);
-	struct tuple_field *field = &format->fields[part->fieldno];
+	assert(part->fieldno < tuple_format_field_count(format));
+	struct tuple_field *field = tuple_format_field(format, part->fieldno);
 	/*
 	 * Field and part nullable actions may differ only
 	 * if one of them is DEFAULT, in which case we use
@@ -133,16 +152,15 @@ tuple_format_create(struct tuple_format *format, struct key_def * const *keys,
 	format->min_field_count =
 		tuple_format_min_field_count(keys, key_count, fields,
 					     field_count);
-	if (format->field_count == 0) {
+	if (tuple_format_field_count(format) == 0) {
 		format->field_map_size = 0;
 		return 0;
 	}
 	/* Initialize defined fields */
 	for (uint32_t i = 0; i < field_count; ++i) {
-		format->fields[i].is_key_part = false;
-		format->fields[i].type = fields[i].type;
-		format->fields[i].offset_slot = TUPLE_OFFSET_SLOT_NIL;
-		format->fields[i].nullable_action = fields[i].nullable_action;
+		struct tuple_field *field = tuple_format_field(format, i);
+		field->type = fields[i].type;
+		field->nullable_action = fields[i].nullable_action;
 		struct coll *coll = NULL;
 		uint32_t cid = fields[i].coll_id;
 		if (cid != COLL_NONE) {
@@ -154,12 +172,9 @@ tuple_format_create(struct tuple_format *format, struct key_def * const *keys,
 			}
 			coll = coll_id->coll;
 		}
-		format->fields[i].coll = coll;
-		format->fields[i].coll_id = cid;
+		field->coll = coll;
+		field->coll_id = cid;
 	}
-	/* Initialize remaining fields */
-	for (uint32_t i = field_count; i < format->field_count; i++)
-		format->fields[i] = tuple_field_default;
 
 	int current_slot = 0;
 
@@ -179,7 +194,8 @@ tuple_format_create(struct tuple_format *format, struct key_def * const *keys,
 		}
 	}
 
-	assert(format->fields[0].offset_slot == TUPLE_OFFSET_SLOT_NIL);
+	assert(tuple_format_field(format, 0)->offset_slot ==
+		TUPLE_OFFSET_SLOT_NIL);
 	size_t field_map_size = -current_slot * sizeof(uint32_t);
 	if (field_map_size > UINT16_MAX) {
 		/** tuple->data_offset is 16 bits */
@@ -253,39 +269,67 @@ tuple_format_alloc(struct key_def * const *keys, uint16_t key_count,
 		}
 	}
 	uint32_t field_count = MAX(space_field_count, index_field_count);
-	uint32_t total = sizeof(struct tuple_format) +
-			 field_count * sizeof(struct tuple_field);
 
-	struct tuple_format *format = (struct tuple_format *) malloc(total);
+	struct tuple_format *format = malloc(sizeof(struct tuple_format));
 	if (format == NULL) {
 		diag_set(OutOfMemory, sizeof(struct tuple_format), "malloc",
 			 "tuple format");
 		return NULL;
 	}
+	if (json_tree_create(&format->field_tree) != 0) {
+		free(format);
+		return NULL;
+	}
+	struct json_path_node path_node;
+	memset(&path_node, 0, sizeof(path_node));
+	path_node.type = JSON_PATH_NUM;
+	for (int32_t i = field_count - 1; i >= 0; i--) {
+		path_node.num = i + TUPLE_INDEX_BASE;
+		struct tuple_field *field = tuple_field_create(&path_node);
+		if (field == NULL)
+			goto error;
+		uint32_t rolling_hash = format->field_tree.root.rolling_hash;
+		rolling_hash = json_path_node_hash(&path_node, rolling_hash);
+		if (json_tree_add(&format->field_tree, &format->field_tree.root,
+				  &field->tree_node, rolling_hash) != 0) {
+			tuple_field_destroy(field);
+			goto error;
+		}
+	}
 	if (dict == NULL) {
 		assert(space_field_count == 0);
 		format->dict = tuple_dictionary_new(NULL, 0);
-		if (format->dict == NULL) {
-			free(format);
-			return NULL;
-		}
+		if (format->dict == NULL)
+			goto error;
 	} else {
 		format->dict = dict;
 		tuple_dictionary_ref(dict);
 	}
 	format->refs = 0;
 	format->id = FORMAT_ID_NIL;
-	format->field_count = field_count;
 	format->index_field_count = index_field_count;
 	format->exact_field_count = 0;
 	format->min_field_count = 0;
 	return format;
+error:;
+	struct tuple_field *field;
+	json_tree_foreach_entry_safe(field, &format->field_tree.root,
+				     struct tuple_field, tree_node)
+		tuple_field_destroy(field);
+	json_tree_destroy(&format->field_tree);
+	free(format);
+	return NULL;
 }
 
 /** Free tuple format resources, doesn't unregister. */
 static inline void
 tuple_format_destroy(struct tuple_format *format)
 {
+	struct tuple_field *field;
+	json_tree_foreach_entry_safe(field, &format->field_tree.root,
+				     struct tuple_field, tree_node)
+		tuple_field_destroy(field);
+	json_tree_destroy(&format->field_tree);
 	tuple_dictionary_unref(format->dict);
 }
 
@@ -323,18 +367,21 @@ tuple_format_new(struct tuple_format_vtab *vtab, struct key_def * const *keys,
 }
 
 bool
-tuple_format1_can_store_format2_tuples(const struct tuple_format *format1,
-				       const struct tuple_format *format2)
+tuple_format1_can_store_format2_tuples(struct tuple_format *format1,
+				       struct tuple_format *format2)
 {
 	if (format1->exact_field_count != format2->exact_field_count)
 		return false;
-	for (uint32_t i = 0; i < format1->field_count; ++i) {
-		const struct tuple_field *field1 = &format1->fields[i];
+	uint32_t format1_field_count = tuple_format_field_count(format1);
+	uint32_t format2_field_count = tuple_format_field_count(format2);
+	for (uint32_t i = 0; i < format1_field_count; ++i) {
+		const struct tuple_field *field1 =
+			tuple_format_field(format1, i);
 		/*
 		 * The field has a data type in format1, but has
 		 * no data type in format2.
 		 */
-		if (i >= format2->field_count) {
+		if (i >= format2_field_count) {
 			/*
 			 * The field can get a name added
 			 * for it, and this doesn't require a data
@@ -350,7 +397,8 @@ tuple_format1_can_store_format2_tuples(const struct tuple_format *format1,
 			else
 				return false;
 		}
-		const struct tuple_field *field2 = &format2->fields[i];
+		const struct tuple_field *field2 =
+			tuple_format_field(format2, i);
 		if (! field_type1_contains_type2(field1->type, field2->type))
 			return false;
 		/*
@@ -369,7 +417,7 @@ int
 tuple_init_field_map(const struct tuple_format *format, uint32_t *field_map,
 		     const char *tuple, bool validate)
 {
-	if (format->field_count == 0)
+	if (tuple_format_field_count(format) == 0)
 		return 0; /* Nothing to initialize */
 
 	const char *pos = tuple;
@@ -392,17 +440,17 @@ tuple_init_field_map(const struct tuple_format *format, uint32_t *field_map,
 
 	/* first field is simply accessible, so we do not store offset to it */
 	enum mp_type mp_type = mp_typeof(*pos);
-	const struct tuple_field *field = &format->fields[0];
+	const struct tuple_field *field =
+		tuple_format_field((struct tuple_format *)format, 0);
 	if (validate &&
 	    key_mp_type_validate(field->type, mp_type, ER_FIELD_TYPE,
 				 TUPLE_INDEX_BASE, tuple_field_is_nullable(field)))
 		return -1;
 	mp_next(&pos);
 	/* other fields...*/
-	++field;
 	uint32_t i = 1;
 	uint32_t defined_field_count = MIN(field_count, validate ?
-					   format->field_count :
+					   tuple_format_field_count(format) :
 					   format->index_field_count);
 	if (field_count < format->index_field_count) {
 		/*
@@ -412,7 +460,8 @@ tuple_init_field_map(const struct tuple_format *format, uint32_t *field_map,
 		memset((char *)field_map - format->field_map_size, 0,
 		       format->field_map_size);
 	}
-	for (; i < defined_field_count; ++i, ++field) {
+	for (; i < defined_field_count; ++i) {
+		field = tuple_format_field((struct tuple_format *)format, i);
 		mp_type = mp_typeof(*pos);
 		if (validate &&
 		    key_mp_type_validate(field->type, mp_type, ER_FIELD_TYPE,
diff --git a/src/box/tuple_format.h b/src/box/tuple_format.h
index 232df22..8426153 100644
--- a/src/box/tuple_format.h
+++ b/src/box/tuple_format.h
@@ -34,6 +34,7 @@
 #include "key_def.h"
 #include "field_def.h"
 #include "errinj.h"
+#include "json/tree.h"
 #include "tuple_dictionary.h"
 
 #if defined(__cplusplus)
@@ -113,6 +114,8 @@ struct tuple_field {
 	struct coll *coll;
 	/** Collation identifier. */
 	uint32_t coll_id;
+	/** An JSON tree entry to organize tree. */
+	struct json_tree_node tree_node;
 };
 
 /**
@@ -166,16 +169,29 @@ struct tuple_format {
 	 * index_field_count <= min_field_count <= field_count.
 	 */
 	uint32_t min_field_count;
-	/* Length of 'fields' array. */
-	uint32_t field_count;
 	/**
 	 * Shared names storage used by all formats of a space.
 	 */
 	struct tuple_dictionary *dict;
-	/* Formats of the fields */
-	struct tuple_field fields[0];
+	/** JSON tree of fields. */
+	struct json_tree field_tree;
 };
 
+
+static inline uint32_t
+tuple_format_field_count(const struct tuple_format *format)
+{
+	return format->field_tree.root.children_count;
+}
+
+static inline struct tuple_field *
+tuple_format_field(struct tuple_format *format, uint32_t fieldno)
+{
+	assert(fieldno < tuple_format_field_count(format));
+	struct json_tree_node *tree_node = format->field_tree.root.children[fieldno];
+	return json_tree_entry(tree_node, struct tuple_field, tree_node);
+}
+
 extern struct tuple_format **tuple_formats;
 
 static inline uint32_t
@@ -238,8 +254,8 @@ tuple_format_new(struct tuple_format_vtab *vtab, struct key_def * const *keys,
  * @retval True, if @a format1 can store any tuples of @a format2.
  */
 bool
-tuple_format1_can_store_format2_tuples(const struct tuple_format *format1,
-				       const struct tuple_format *format2);
+tuple_format1_can_store_format2_tuples(struct tuple_format *format1,
+				       struct tuple_format *format2);
 
 /**
  * Calculate minimal field count of tuples with specified keys and
@@ -333,7 +349,9 @@ tuple_field_raw(const struct tuple_format *format, const char *tuple,
 			return tuple;
 		}
 
-		int32_t offset_slot = format->fields[field_no].offset_slot;
+		int32_t offset_slot =
+			tuple_format_field((struct tuple_format *)format,
+					   field_no)->offset_slot;
 		if (offset_slot != TUPLE_OFFSET_SLOT_NIL) {
 			if (field_map[offset_slot] != 0)
 				return tuple + field_map[offset_slot];
diff --git a/src/box/vy_stmt.c b/src/box/vy_stmt.c
index d838404..3e60fec 100644
--- a/src/box/vy_stmt.c
+++ b/src/box/vy_stmt.c
@@ -411,7 +411,7 @@ vy_stmt_new_surrogate_from_key(const char *key, enum iproto_type type,
 	uint32_t *field_map = (uint32_t *) raw;
 	char *wpos = mp_encode_array(raw, field_count);
 	for (uint32_t i = 0; i < field_count; ++i) {
-		const struct tuple_field *field = &format->fields[i];
+		const struct tuple_field *field = tuple_format_field(format, i);
 		if (field->offset_slot != TUPLE_OFFSET_SLOT_NIL)
 			field_map[field->offset_slot] = wpos - raw;
 		if (iov[i].iov_base == NULL) {
@@ -465,7 +465,7 @@ vy_stmt_new_surrogate_delete_raw(struct tuple_format *format,
 	}
 	char *pos = mp_encode_array(data, field_count);
 	for (uint32_t i = 0; i < field_count; ++i) {
-		const struct tuple_field *field = &format->fields[i];
+		const struct tuple_field *field = tuple_format_field(format, i);
 		if (! field->is_key_part) {
 			/* Unindexed field - write NIL. */
 			assert(i < src_count);
diff --git a/test-run b/test-run
index 670f330..b8764e1 160000
--- a/test-run
+++ b/test-run
@@ -1 +1 @@
-Subproject commit 670f330aacaf44bc8b1f969fa0cd5f811c5ceb1b
+Subproject commit b8764e17ccc79a26d1e661a0aaeaad90bd0aa1ea
-- 
2.7.4