[PATCH v3.1 2/5] box: introduce tuple_field_by_part routine

[Kirill Shcherbatov <kshcherbatov@tarantool.org>]

Start use tuple_field_by_part(_raw) routine in *extract,
*compare, *hash functions. This new function use key_part to
retrieve field data mentioned in key_part. Now it is just a
wrapper for tuple_field_raw but with introducing JSON paths
it would work in other way.

Part of #1012.
---
 src/box/memtx_bitset.c       |  5 +++--
 src/box/memtx_rtree.c        |  3 ++-
 src/box/tuple.h              | 14 ++++++++++++++
 src/box/tuple_compare.cc     | 44 +++++++++++++++++++++++++++-----------------
 src/box/tuple_extract_key.cc |  8 ++++----
 src/box/tuple_format.c       |  7 +++++++
 src/box/tuple_format.h       | 12 ++++++++++++
 src/box/tuple_hash.cc        | 19 ++++++++++++++-----
 src/box/vy_stmt.h            |  9 +++++++--
 9 files changed, 90 insertions(+), 31 deletions(-)

diff --git a/src/box/memtx_bitset.c b/src/box/memtx_bitset.c
index a665f1a..cd7362e 100644
--- a/src/box/memtx_bitset.c
+++ b/src/box/memtx_bitset.c
@@ -283,8 +283,9 @@ memtx_bitset_index_replace(struct index *base, struct tuple *old_tuple,
 	}
 
 	if (new_tuple != NULL) {
-		const char *field;
-		field = tuple_field(new_tuple, base->def->key_def->parts[0].fieldno);
+		const char *field =
+			tuple_field_by_part(new_tuple,
+					    base->def->key_def->parts);
 		uint32_t key_len;
 		const void *key = make_key(field, &key_len);
 #ifndef OLD_GOOD_BITSET
diff --git a/src/box/memtx_rtree.c b/src/box/memtx_rtree.c
index 0b12cda..f2aa6c3 100644
--- a/src/box/memtx_rtree.c
+++ b/src/box/memtx_rtree.c
@@ -112,7 +112,8 @@ extract_rectangle(struct rtree_rect *rect, const struct tuple *tuple,
 		  struct index_def *index_def)
 {
 	assert(index_def->key_def->part_count == 1);
-	const char *elems = tuple_field(tuple, index_def->key_def->parts[0].fieldno);
+	const char *elems = tuple_field_by_part(tuple,
+						index_def->key_def->parts);
 	unsigned dimension = index_def->opts.dimension;
 	uint32_t count = mp_decode_array(&elems);
 	return mp_decode_rect(rect, dimension, elems, count, "Field");
diff --git a/src/box/tuple.h b/src/box/tuple.h
index 2e84516..b638f50 100644
--- a/src/box/tuple.h
+++ b/src/box/tuple.h
@@ -43,6 +43,7 @@ extern "C" {
 
 struct slab_arena;
 struct quota;
+struct key_part;
 
 /**
  * A format for standalone tuples allocated on runtime arena.
@@ -522,6 +523,19 @@ tuple_field(const struct tuple *tuple, uint32_t fieldno)
 }
 
 /**
+ * Get a field refereed by index @part in tuple.
+ * @param tuple Tuple to get the field from.
+ * @param part Index part to use.
+ * @retval Field data if the field exists or NULL.
+ */
+static inline const char *
+tuple_field_by_part(const struct tuple *tuple, struct key_part *part)
+{
+	return tuple_field_by_part_raw(tuple_format(tuple), tuple_data(tuple),
+				       tuple_field_map(tuple), part);
+}
+
+/**
  * Get tuple field by its JSON path.
  * @param tuple Tuple to get field from.
  * @param path Field JSON path.
diff --git a/src/box/tuple_compare.cc b/src/box/tuple_compare.cc
index f9b4597..e21b009 100644
--- a/src/box/tuple_compare.cc
+++ b/src/box/tuple_compare.cc
@@ -430,10 +430,20 @@ tuple_common_key_parts(const struct tuple *tuple_a, const struct tuple *tuple_b,
 		       struct key_def *key_def)
 {
 	uint32_t i;
+	struct tuple_format *tuple_a_format = tuple_format(tuple_a);
+	struct tuple_format *tuple_b_format = tuple_format(tuple_b);
+	const char *tuple_a_raw = tuple_data(tuple_a);
+	const char *tuple_b_raw = tuple_data(tuple_b);
+	const uint32_t *tuple_a_field_map = tuple_field_map(tuple_a);
+	const uint32_t *tuple_b_field_map = tuple_field_map(tuple_b);
 	for (i = 0; i < key_def->part_count; i++) {
-		const struct key_part *part = &key_def->parts[i];
-		const char *field_a = tuple_field(tuple_a, part->fieldno);
-		const char *field_b = tuple_field(tuple_b, part->fieldno);
+		struct key_part *part = (struct key_part *)&key_def->parts[i];
+		const char *field_a =
+			tuple_field_by_part_raw(tuple_a_format, tuple_a_raw,
+						tuple_a_field_map, part);
+		const char *field_b =
+			tuple_field_by_part_raw(tuple_b_format, tuple_b_raw,
+						tuple_b_field_map, part);
 		enum mp_type a_type = field_a != NULL ?
 				      mp_typeof(*field_a) : MP_NIL;
 		enum mp_type b_type = field_b != NULL ?
@@ -497,10 +507,10 @@ tuple_compare_slowpath(const struct tuple *tuple_a, const struct tuple *tuple_b,
 		end = part + key_def->part_count;
 
 	for (; part < end; part++) {
-		field_a = tuple_field_raw(format_a, tuple_a_raw, field_map_a,
-					  part->fieldno);
-		field_b = tuple_field_raw(format_b, tuple_b_raw, field_map_b,
-					  part->fieldno);
+		field_a = tuple_field_by_part_raw(format_a, tuple_a_raw,
+						  field_map_a, part);
+		field_b = tuple_field_by_part_raw(format_b, tuple_b_raw,
+						  field_map_b, part);
 		assert(has_optional_parts ||
 		       (field_a != NULL && field_b != NULL));
 		if (! is_nullable) {
@@ -547,10 +557,10 @@ tuple_compare_slowpath(const struct tuple *tuple_a, const struct tuple *tuple_b,
 	 */
 	end = key_def->parts + key_def->part_count;
 	for (; part < end; ++part) {
-		field_a = tuple_field_raw(format_a, tuple_a_raw, field_map_a,
-					  part->fieldno);
-		field_b = tuple_field_raw(format_b, tuple_b_raw, field_map_b,
-					  part->fieldno);
+		field_a = tuple_field_by_part_raw(format_a, tuple_a_raw,
+						  field_map_a, part);
+		field_b = tuple_field_by_part_raw(format_b, tuple_b_raw,
+						  field_map_b, part);
 		/*
 		 * Extended parts are primary, and they can not
 		 * be absent or be NULLs.
@@ -580,9 +590,9 @@ tuple_compare_with_key_slowpath(const struct tuple *tuple, const char *key,
 	const uint32_t *field_map = tuple_field_map(tuple);
 	enum mp_type a_type, b_type;
 	if (likely(part_count == 1)) {
-		const char *field;
-		field = tuple_field_raw(format, tuple_raw, field_map,
-					part->fieldno);
+		const char *field =
+			tuple_field_by_part_raw(format, tuple_raw, field_map,
+						part);
 		if (! is_nullable) {
 			return tuple_compare_field(field, key, part->type,
 						   part->coll);
@@ -606,9 +616,9 @@ tuple_compare_with_key_slowpath(const struct tuple *tuple, const char *key,
 	struct key_part *end = part + part_count;
 	int rc;
 	for (; part < end; ++part, mp_next(&key)) {
-		const char *field;
-		field = tuple_field_raw(format, tuple_raw, field_map,
-					part->fieldno);
+		const char *field =
+			tuple_field_by_part_raw(format, tuple_raw,
+						field_map, part);
 		if (! is_nullable) {
 			rc = tuple_compare_field(field, key, part->type,
 						 part->coll);
diff --git a/src/box/tuple_extract_key.cc b/src/box/tuple_extract_key.cc
index e43f997..e493c3b 100644
--- a/src/box/tuple_extract_key.cc
+++ b/src/box/tuple_extract_key.cc
@@ -109,8 +109,8 @@ tuple_extract_key_slowpath(const struct tuple *tuple,
 	/* Calculate the key size. */
 	for (uint32_t i = 0; i < part_count; ++i) {
 		const char *field =
-			tuple_field_raw(format, data, field_map,
-					key_def->parts[i].fieldno);
+			tuple_field_by_part_raw(format, data, field_map,
+						&key_def->parts[i]);
 		if (has_optional_parts && field == NULL) {
 			bsize += mp_sizeof_nil();
 			continue;
@@ -151,8 +151,8 @@ tuple_extract_key_slowpath(const struct tuple *tuple,
 	char *key_buf = mp_encode_array(key, part_count);
 	for (uint32_t i = 0; i < part_count; ++i) {
 		const char *field =
-			tuple_field_raw(format, data, field_map,
-					key_def->parts[i].fieldno);
+			tuple_field_by_part_raw(format, data, field_map,
+						&key_def->parts[i]);
 		if (has_optional_parts && field == NULL) {
 			key_buf = mp_encode_nil(key_buf);
 			continue;
diff --git a/src/box/tuple_format.c b/src/box/tuple_format.c
index b385c0d..91677d4 100644
--- a/src/box/tuple_format.c
+++ b/src/box/tuple_format.c
@@ -541,6 +541,13 @@ tuple_field_go_to_key(const char **field, const char *key, int len)
 	return -1;
 }
 
+const char *
+tuple_field_by_part_raw(const struct tuple_format *format, const char *data,
+			const uint32_t *field_map, struct key_part *part)
+{
+	return tuple_field_raw(format, data, field_map, part->fieldno);
+}
+
 int
 tuple_field_raw_by_path(struct tuple_format *format, const char *tuple,
                         const uint32_t *field_map, const char *path,
diff --git a/src/box/tuple_format.h b/src/box/tuple_format.h
index c7dc48f..0ffdf7d 100644
--- a/src/box/tuple_format.h
+++ b/src/box/tuple_format.h
@@ -324,6 +324,18 @@ tuple_init_field_map(const struct tuple_format *format, uint32_t *field_map,
 		     const char *tuple);
 
 /**
+ * Get a field refereed by index @part in tuple.
+ * @param format Tuple format.
+ * @param tuple A pointer to MessagePack array.
+ * @param field_map A pointer to the LAST element of field map.
+ * @param part Index part to use.
+ * @retval Field data if the field exists or NULL.
+ */
+const char *
+tuple_field_by_part_raw(const struct tuple_format *format, const char *data,
+			const uint32_t *field_map, struct key_part *part);
+
+/**
  * Get a field at the specific position in this MessagePack array.
  * Returns a pointer to MessagePack data.
  * @param format tuple format
diff --git a/src/box/tuple_hash.cc b/src/box/tuple_hash.cc
index 45a786a..b394804 100644
--- a/src/box/tuple_hash.cc
+++ b/src/box/tuple_hash.cc
@@ -157,7 +157,8 @@ struct TupleHash
 		uint32_t h = HASH_SEED;
 		uint32_t carry = 0;
 		uint32_t total_size = 0;
-		const char *field = tuple_field(tuple, key_def->parts->fieldno);
+		const char *field =
+			tuple_field_by_part(tuple, key_def->parts);
 		TupleFieldHash<TYPE, MORE_TYPES...>::
 			hash(&field, &h, &carry, &total_size);
 		return PMurHash32_Result(h, carry, total_size);
@@ -169,7 +170,8 @@ struct TupleHash<FIELD_TYPE_UNSIGNED> {
 	static uint32_t	hash(const struct tuple *tuple,
 			     struct key_def *key_def)
 	{
-		const char *field = tuple_field(tuple, key_def->parts->fieldno);
+		const char *field =
+			tuple_field_by_part(tuple, key_def->parts);
 		uint64_t val = mp_decode_uint(&field);
 		if (likely(val <= UINT32_MAX))
 			return val;
@@ -311,7 +313,7 @@ uint32_t
 tuple_hash_key_part(uint32_t *ph1, uint32_t *pcarry, const struct tuple *tuple,
 		    struct key_part *part)
 {
-	const char *field = tuple_field(tuple, part->fieldno);
+	const char *field = tuple_field_by_part(tuple, part);
 	if (field == NULL)
 		return tuple_hash_null(ph1, pcarry);
 	return tuple_hash_field(ph1, pcarry, &field, part->coll);
@@ -326,7 +328,12 @@ tuple_hash_slowpath(const struct tuple *tuple, struct key_def *key_def)
 	uint32_t carry = 0;
 	uint32_t total_size = 0;
 	uint32_t prev_fieldno = key_def->parts[0].fieldno;
-	const char *field = tuple_field(tuple, key_def->parts[0].fieldno);
+	struct tuple_format *format = tuple_format(tuple);
+	const char *tuple_raw = tuple_data(tuple);
+	const uint32_t *field_map = tuple_field_map(tuple);
+	const char *field =
+		tuple_field_by_part_raw(format, tuple_raw, field_map,
+					key_def->parts);
 	const char *end = (char *)tuple + tuple_size(tuple);
 	if (has_optional_parts && field == NULL) {
 		total_size += tuple_hash_null(&h, &carry);
@@ -340,7 +347,9 @@ tuple_hash_slowpath(const struct tuple *tuple, struct key_def *key_def)
 		 * need of tuple_field
 		 */
 		if (prev_fieldno + 1 != key_def->parts[part_id].fieldno) {
-			field = tuple_field(tuple, key_def->parts[part_id].fieldno);
+			struct key_part *part = &key_def->parts[part_id];
+			field = tuple_field_by_part_raw(format, tuple_raw,
+							field_map, part);
 		}
 		if (has_optional_parts && (field == NULL || field >= end)) {
 			total_size += tuple_hash_null(&h, &carry);
diff --git a/src/box/vy_stmt.h b/src/box/vy_stmt.h
index c2a5621..b52b4e2 100644
--- a/src/box/vy_stmt.h
+++ b/src/box/vy_stmt.h
@@ -716,8 +716,13 @@ vy_tuple_format_new_with_colmask(struct tuple_format *mem_format);
 static inline bool
 vy_tuple_key_contains_null(const struct tuple *tuple, struct key_def *def)
 {
-	for (uint32_t i = 0; i < def->part_count; ++i) {
-		const char *field = tuple_field(tuple, def->parts[i].fieldno);
+	struct tuple_format *format = tuple_format(tuple);
+	const char *data = tuple_data(tuple);
+	const uint32_t *field_map = tuple_field_map(tuple);
+	for (struct key_part *part = def->parts, *end = part + def->part_count;
+	     part < end; ++part) {
+		const char *field =
+			tuple_field_by_part_raw(format, data, field_map, part);
 		if (field == NULL || mp_typeof(*field) == MP_NIL)
 			return true;
 	}
-- 
2.7.4