[Tarantool-patches] [PATCH v4 01/53] sql: enchance vdbe_decode_msgpack_into_mem()

[imeevma at tarantool.org]

Currently, vdbe_decode_msgpack_into_mem() creates a MEM that is not
properly initialized in case msgpack contains MP_EXT, MP_MAP, or
MP_ARRAY fields. Also, it doesn't set field_type.

Needed for #5818
---
 src/box/sql/vdbe.c    | 22 ---------------
 src/box/sql/vdbeaux.c | 64 ++++++++++++++++++++++++++++++++-----------
 2 files changed, 48 insertions(+), 38 deletions(-)

diff --git a/src/box/sql/vdbe.c b/src/box/sql/vdbe.c
index 3b3b1f01d..9a4f38bb9 100644
--- a/src/box/sql/vdbe.c
+++ b/src/box/sql/vdbe.c
@@ -846,16 +846,6 @@ vdbe_field_ref_fetch_data(struct vdbe_field_ref *field_ref, uint32_t fieldno)
 	return field_begin;
 }
 
-static inline enum field_type
-vdbe_field_ref_fetch_type(struct vdbe_field_ref *field_ref, uint32_t fieldno)
-{
-	const struct tuple_field *tf =
-		vdbe_field_ref_fetch_field(field_ref, fieldno);
-	if (tf == NULL || tf->type == FIELD_TYPE_ANY)
-		return field_type_MAX;
-	return tf->type;
-}
-
 /**
  * Fetch field by fieldno using vdbe_field_ref and store result
  * in dest_mem.
@@ -879,17 +869,6 @@ vdbe_field_ref_fetch(struct vdbe_field_ref *field_ref, uint32_t fieldno,
 	if (vdbe_decode_msgpack_into_mem(data, dest_mem, &dummy) != 0)
 		return -1;
 
-	/*
-	 * MsgPack map, array or extension (unsupported in sql).
-	 * Wrap it in a blob verbatim.
-	 */
-	if (dest_mem->flags == 0) {
-		dest_mem->z = (char *) data;
-		dest_mem->n = vdbe_field_ref_fetch_data(field_ref,
-							fieldno + 1) - data;
-		dest_mem->flags = MEM_Blob | MEM_Ephem | MEM_Subtype;
-		dest_mem->subtype = SQL_SUBTYPE_MSGPACK;
-	}
 	/*
 	 * Add 0 termination (at most for strings)
 	 * Not sure why do we check MEM_Ephem
@@ -909,7 +888,6 @@ vdbe_field_ref_fetch(struct vdbe_field_ref *field_ref, uint32_t fieldno,
 		dest_mem->flags |= MEM_Term;
 	}
 	UPDATE_MAX_BLOBSIZE(dest_mem);
-	dest_mem->field_type = vdbe_field_ref_fetch_type(field_ref, fieldno);
 	return 0;
 }
 
diff --git a/src/box/sql/vdbeaux.c b/src/box/sql/vdbeaux.c
index 91b64316e..90c4b73c0 100644
--- a/src/box/sql/vdbeaux.c
+++ b/src/box/sql/vdbeaux.c
@@ -2793,38 +2793,62 @@ vdbe_decode_msgpack_into_mem(const char *buf, struct Mem *mem, uint32_t *len)
 {
 	const char *start_buf = buf;
 	switch (mp_typeof(*buf)) {
-	case MP_ARRAY:
-	case MP_MAP:
-	case MP_EXT:
-	default: {
-		mem->flags = 0;
+	case MP_ARRAY: {
+		mem->z = (char *)buf;
+		mp_next(&buf);
+		mem->n = buf - mem->z;
+		mem->flags = MEM_Blob | MEM_Ephem | MEM_Subtype;
+		mem->subtype = SQL_SUBTYPE_MSGPACK;
+		mem->field_type = FIELD_TYPE_ARRAY;
+		break;
+	}
+	case MP_MAP: {
+		mem->z = (char *)buf;
+		mp_next(&buf);
+		mem->n = buf - mem->z;
+		mem->flags = MEM_Blob | MEM_Ephem | MEM_Subtype;
+		mem->subtype = SQL_SUBTYPE_MSGPACK;
+		mem->field_type = FIELD_TYPE_MAP;
+		break;
+	}
+	case MP_EXT: {
+		mem->z = (char *)buf;
+		mp_next(&buf);
+		mem->n = buf - mem->z;
+		mem->flags = MEM_Blob | MEM_Ephem;
+		mem->field_type = FIELD_TYPE_VARBINARY;
 		break;
 	}
 	case MP_NIL: {
 		mp_decode_nil(&buf);
 		mem->flags = MEM_Null;
+		mem->field_type = field_type_MAX;
 		break;
 	}
 	case MP_BOOL: {
 		mem->u.b = mp_decode_bool(&buf);
 		mem->flags = MEM_Bool;
+		mem->field_type = FIELD_TYPE_BOOLEAN;
 		break;
 	}
 	case MP_UINT: {
 		uint64_t v = mp_decode_uint(&buf);
 		mem->u.u = v;
 		mem->flags = MEM_UInt;
+		mem->field_type = FIELD_TYPE_INTEGER;
 		break;
 	}
 	case MP_INT: {
 		mem->u.i = mp_decode_int(&buf);
 		mem->flags = MEM_Int;
+		mem->field_type = FIELD_TYPE_INTEGER;
 		break;
 	}
 	case MP_STR: {
 		/* XXX u32->int */
 		mem->n = (int) mp_decode_strl(&buf);
 		mem->flags = MEM_Str | MEM_Ephem;
+		mem->field_type = FIELD_TYPE_STRING;
 install_blob:
 		mem->z = (char *)buf;
 		buf += mem->n;
@@ -2834,18 +2858,33 @@ install_blob:
 		/* XXX u32->int */
 		mem->n = (int) mp_decode_binl(&buf);
 		mem->flags = MEM_Blob | MEM_Ephem;
+		mem->field_type = FIELD_TYPE_VARBINARY;
 		goto install_blob;
 	}
 	case MP_FLOAT: {
 		mem->u.r = mp_decode_float(&buf);
-		mem->flags = sqlIsNaN(mem->u.r) ? MEM_Null : MEM_Real;
+		if (sqlIsNaN(mem->u.r)) {
+			mem->flags = MEM_Null;
+			mem->field_type = field_type_MAX;
+		} else {
+			mem->flags = MEM_Real;
+			mem->field_type = FIELD_TYPE_DOUBLE;
+		}
 		break;
 	}
 	case MP_DOUBLE: {
 		mem->u.r = mp_decode_double(&buf);
-		mem->flags = sqlIsNaN(mem->u.r) ? MEM_Null : MEM_Real;
+		if (sqlIsNaN(mem->u.r)) {
+			mem->flags = MEM_Null;
+			mem->field_type = field_type_MAX;
+		} else {
+			mem->flags = MEM_Real;
+			mem->field_type = FIELD_TYPE_DOUBLE;
+		}
 		break;
 	}
+	default:
+		unreachable();
 	}
 	*len = (uint32_t)(buf - start_buf);
 	return 0;
@@ -2868,15 +2907,8 @@ sqlVdbeRecordUnpackMsgpack(struct key_def *key_def,	/* Information about the rec
 		pMem->z = 0;
 		uint32_t sz = 0;
 		vdbe_decode_msgpack_into_mem(zParse, pMem, &sz);
-		if (sz == 0) {
-			/* MsgPack array, map or ext. Treat as blob. */
-			pMem->z = (char *)zParse;
-			mp_next(&zParse);
-			pMem->n = zParse - pMem->z;
-			pMem->flags = MEM_Blob | MEM_Ephem;
-		} else {
-			zParse += sz;
-		}
+		assert(sz != 0);
+		zParse += sz;
 		pMem++;
 	}
 }
-- 
2.25.1