[PATCH v6 3/4] 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.

Needed for #1012
---
 src/box/sql.c          |  18 +++---
 src/box/sql/build.c    |   5 +-
 src/box/tuple.c        |  10 ++--
 src/box/tuple_format.c | 122 +++++++++++++++++++++++++++++------------
 src/box/tuple_format.h |  49 ++++++++++++++---
 src/box/vy_stmt.c      |   4 +-
 6 files changed, 148 insertions(+), 60 deletions(-)

diff --git a/src/box/sql.c b/src/box/sql.c
index 25b3213fd..2cb0edbff 100644
--- a/src/box/sql.c
+++ b/src/box/sql.c
@@ -199,8 +199,9 @@ tarantoolSqlite3TupleColumnFast(BtCursor *pCur, u32 fieldno, u32 *field_size)
 	assert(pCur->last_tuple != NULL);
 
 	struct tuple_format *format = tuple_format(pCur->last_tuple);
-	if (fieldno >= format->field_count ||
-	    format->fields[fieldno].offset_slot == TUPLE_OFFSET_SLOT_NIL)
+	if (fieldno >= tuple_format_field_count(format) ||
+	    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;
@@ -895,7 +896,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;
@@ -913,7 +914,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++) {
 		/*
@@ -931,9 +932,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;
 
@@ -941,9 +943,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 52f0bde15..b5abaeeda 100644
--- a/src/box/sql/build.c
+++ b/src/box/sql/build.c
@@ -936,8 +936,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 ef4d16f39..aae1c3cdd 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 149248144..eeb68f5dd 100644
--- a/src/box/tuple_format.c
+++ b/src/box/tuple_format.c
@@ -38,10 +38,27 @@ 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_NONE, NULL, COLL_NONE,
-};
+static struct tuple_field *
+tuple_field_new(void)
+{
+	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_NONE;
+	return ret;
+}
+
+static void
+tuple_field_delete(struct tuple_field *field)
+{
+	free(field);
+}
 
 static int
 tuple_format_use_key_part(struct tuple_format *format,
@@ -49,8 +66,8 @@ tuple_format_use_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);
 	/*
 		* If a field is not present in the space format,
 		* inherit nullable action of the first key part
@@ -138,16 +155,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) {
@@ -159,12 +175,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;
 
@@ -184,7 +197,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 */
@@ -242,6 +256,19 @@ tuple_format_deregister(struct tuple_format *format)
 	format->id = FORMAT_ID_NIL;
 }
 
+/** Destroy field JSON tree and release allocated memory. */
+static inline void
+tuple_format_fields_destroy(struct tuple_format *format)
+{
+	struct tuple_field *field, *tmp;
+	json_tree_foreach_entry_safe(&format->fields.root, field,
+				     struct tuple_field, token, tmp) {
+		json_tree_del(&format->fields, &field->token);
+		tuple_field_delete(field);
+	}
+	json_tree_destroy(&format->fields);
+}
+
 static struct tuple_format *
 tuple_format_alloc(struct key_def * const *keys, uint16_t key_count,
 		   uint32_t space_field_count, struct tuple_dictionary *dict)
@@ -258,39 +285,57 @@ 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->fields) != 0) {
+		free(format);
+		return NULL;
+	}
+	for (uint32_t fieldno = 0; fieldno < field_count; fieldno++) {
+		struct tuple_field *field = tuple_field_new();
+		if (field == NULL)
+			goto error;
+		field->token.num = fieldno;
+		field->token.type = JSON_TOKEN_NUM;
+		if (json_tree_add(&format->fields, &format->fields.root,
+				  &field->token) != 0) {
+			diag_set(OutOfMemory, 0, "json_tree_add",
+				 "&format->tree");
+			tuple_field_delete(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:;
+	tuple_format_fields_destroy(format);
+	free(format);
+	return NULL;
 }
 
 /** Free tuple format resources, doesn't unregister. */
 static inline void
 tuple_format_destroy(struct tuple_format *format)
 {
+	tuple_format_fields_destroy(format);
 	tuple_dictionary_unref(format->dict);
 }
 
@@ -328,18 +373,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
@@ -355,7 +403,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;
 		/*
@@ -374,7 +423,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;
@@ -397,17 +446,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) {
 		/*
@@ -417,7 +466,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 232df22b2..68e9205b9 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/json.h"
 #include "tuple_dictionary.h"
 
 #if defined(__cplusplus)
@@ -113,6 +114,8 @@ struct tuple_field {
 	struct coll *coll;
 	/** Collation identifier. */
 	uint32_t coll_id;
+	/** Link in tuple_format::fields. */
+	struct json_token token;
 };
 
 /**
@@ -166,16 +169,46 @@ 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];
+	/**
+	 * Fields comprising the format, organized in a tree.
+	 * First level nodes correspond to tuple fields.
+	 * Deeper levels define indexed JSON paths within
+	 * tuple fields. Nodes of the tree are linked by
+	 * tuple_field::token.
+	 */
+	struct json_tree fields;
 };
 
+/**
+ * Return a count of first level nodes correspond to tuple
+ * fields.
+ */
+static inline uint32_t
+tuple_format_field_count(const struct tuple_format *format)
+{
+	const struct json_token *root = &format->fields.root;
+	return root->children != NULL ? root->max_child_idx + 1 : 0;
+}
+
+/**
+ * Get the first level node corresponding to tuple field by its
+ * fieldno.
+ */
+static inline struct tuple_field *
+tuple_format_field(struct tuple_format *format, uint32_t fieldno)
+{
+	assert(fieldno < tuple_format_field_count(format));
+	struct json_token token;
+	token.type = JSON_TOKEN_NUM;
+	token.num = fieldno;
+	return json_tree_lookup_entry(&format->fields, &format->fields.root,
+				      &token, struct tuple_field, token);
+}
+
 extern struct tuple_format **tuple_formats;
 
 static inline uint32_t
@@ -238,8 +271,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 +366,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 d83840406..3e60fece9 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);
-- 
2.19.2