[tarantool-patches] [PATCH v2 13/15] sql: support -2^63 .. 2^64-1 integer type

[Stanislav Zudin <szudin@tarantool.org>]

Returns correct text name for unsigned data type.
Applies a coding standard.
Improves mapping of vdbe data types to sql data types.

Part of #3810
---
 src/box/sql/func.c     |  11 +--
 src/box/sql/sqlInt.h   |  15 ++--
 src/box/sql/util.c     | 153 +++++++++++++++++++++--------------------
 src/box/sql/vdbe.c     |   2 +
 src/box/sql/vdbeInt.h  |   4 +-
 src/box/sql/vdbeapi.c  | 109 +++++++++--------------------
 test/sql/iproto.result |   2 +-
 7 files changed, 132 insertions(+), 164 deletions(-)

diff --git a/src/box/sql/func.c b/src/box/sql/func.c
index f1b894f16..74b56d780 100644
--- a/src/box/sql/func.c
+++ b/src/box/sql/func.c
@@ -121,6 +121,9 @@ typeofFunc(sql_context * context, int NotUsed, sql_value ** argv)
 	case SQL_BLOB:
 		z = "blob";
 		break;
+	case SQL_UNSIGNED:
+		z = "unsigned";
+		break;
 	default:
 		z = "null";
 		break;
@@ -1410,7 +1413,7 @@ struct SumCtx {
 	i64 cnt;		/* Number of elements summed */
 	u8 overflow;		/* True if integer overflow seen */
 	u8 approx;		/* True if non-integer value was input to the sum */
-	u8 is_unsigned;		/* True if value exceeded 2^63 */
+	bool is_unsigned;	/* True if value exceeded 2^63 */
 };
 
 /*
@@ -1441,14 +1444,14 @@ sumStep(sql_context * context, int argc, sql_value ** argv)
 			p->rSum += v;
 			if ((p->approx | p->overflow) == 0)
 				rc = sqlAddInt64(&p->iSum,
-						p->is_unsigned == 0,
+						!p->is_unsigned,
 						v, true);
 		} else if (type == SQL_UNSIGNED) {
 			u64 v = sql_value_uint64(argv[0]);
 			p->rSum += v;
 			if ((p->approx | p->overflow) == 0)
 				rc = sqlAddInt64(&p->iSum,
-						 p->is_unsigned == 0,
+						 !p->is_unsigned,
 						 v, false);
 		} else {
 			p->rSum += sql_value_double(argv[0]);
@@ -1462,7 +1465,7 @@ sumStep(sql_context * context, int argc, sql_value ** argv)
 			case ATHR_SIGNED:
 				break;
 			case ATHR_UNSIGNED:
-				p->is_unsigned = 1;
+				p->is_unsigned = true;
 				break;
 			default:
 				p->overflow = 1;
diff --git a/src/box/sql/sqlInt.h b/src/box/sql/sqlInt.h
index fd87366df..509879cf7 100644
--- a/src/box/sql/sqlInt.h
+++ b/src/box/sql/sqlInt.h
@@ -4383,11 +4383,16 @@ enum arithmetic_result {
 	ATHR_DIVBYZERO
 };
 
-enum arithmetic_result sqlAddInt64(i64 *, bool, i64, bool);
-enum arithmetic_result sqlSubInt64(i64 *, bool, i64, bool);
-enum arithmetic_result sqlMulInt64(i64 *, bool, i64, bool);
-enum arithmetic_result sqlDivInt64(i64 *, bool, i64, bool);
-enum arithmetic_result sqlRemInt64(i64 *, bool, i64, bool);
+enum arithmetic_result
+sqlAddInt64(i64 *, bool, i64, bool);
+enum arithmetic_result
+sqlSubInt64(i64 *, bool, i64, bool);
+enum arithmetic_result
+sqlMulInt64(i64 *, bool, i64, bool);
+enum arithmetic_result
+sqlDivInt64(i64 *, bool, i64, bool);
+enum arithmetic_result
+sqlRemInt64(i64 *, bool, i64, bool);
 
 int sqlAbsInt32(int);
 u8 sqlGetBoolean(const char *z, u8);
diff --git a/src/box/sql/util.c b/src/box/sql/util.c
index ce3f1b5c3..e56ae6e05 100644
--- a/src/box/sql/util.c
+++ b/src/box/sql/util.c
@@ -1275,102 +1275,103 @@ apply_sign(i64* pOut, u64 value, bool is_neg)
 }
 
 /*
- * Attempt to add, substract, or multiply the 64-bit value iB against
- * the other 64-bit integer at *pA and store the result in *pA.
+ * Attempt to add, substract, or multiply the 64-bit value irhs against
+ * the other 64-bit integer at *plhs and store the result in *plhs.
  * Return ATHR_SIGNED or ATHR_UNSIGNED on success.
  * Or if the operation would have resulted in an
  * overflow, leave *pA unchanged and return ATHR_OVERFLOW.
  */
 enum arithmetic_result
-sqlAddInt64(i64 * pA, bool is_signedA, i64 iB, bool is_signedB)
+sqlAddInt64(i64 * plhs, bool is_lhs_signed, i64 irhs, bool is_rhs_signed)
 {
-	i64 iA = *pA;
+	i64 ilhs = *plhs;
 
-	bool is_negA = iA < 0 && is_signedA;
-	bool is_negB = iB < 0 && is_signedB;
+	bool is_lhs_neg = ilhs < 0 && is_lhs_signed;
+	bool is_rhs_neg = irhs < 0 && is_rhs_signed;
 
-	if (is_negA != is_negB) {
-		/* Make sure we've got only one combination of
+	if (is_lhs_neg != is_rhs_neg) {
+		/*
+		 * Make sure we've got only one combination of
 		 * positive and negative operands
 		 */
-		if (is_negA > is_negB) {
-			SWAP(is_negA, is_negB);
-			SWAP(iA, iB);
+		if (is_lhs_neg > is_rhs_neg) {
+			SWAP(is_lhs_neg, is_rhs_neg);
+			SWAP(ilhs, irhs);
 		}
-		assert(is_negA == false && is_negB == true);
+		assert(is_lhs_neg == false && is_rhs_neg == true);
 
-		u64 uB = mod64(iB, true);
+		u64 uB = mod64(irhs, true);
 
-		if ((u64)iA >= uB) {
-			u64 sum = (u64)iA - uB;
-			*pA = (i64)sum;
+		if ((u64)ilhs >= uB) {
+			u64 sum = (u64)ilhs - uB;
+			*plhs = (i64)sum;
 			return (sum <= INT64_MAX) ? ATHR_SIGNED
 						  : ATHR_UNSIGNED;
 		} else {
-			u64 sum = uB - (u64)iA;
+			u64 sum = uB - (u64)ilhs;
 			if (sum == INT64_MIN_MOD) {
-				*pA = INT64_MIN;
+				*plhs = INT64_MIN;
 			} else {
 				assert(sum <= INT64_MAX);
-				*pA = -(i64)sum;
+				*plhs = -(i64)sum;
 			}
 			return ATHR_SIGNED;
 		}
 	}
 
-	if (is_negA) {
-		assert(is_signedA && is_signedB);
-		if (-(iA + LARGEST_INT64) > iB + 1)
+	if (is_lhs_neg) {
+		assert(is_lhs_signed && is_rhs_signed);
+		if (-(ilhs + LARGEST_INT64) > irhs + 1)
 			return ATHR_OVERFLOW;
-		*pA = iA + iB;
+		*plhs = ilhs + irhs;
 		return ATHR_SIGNED;
 	} else {
-		if (UINT64_MAX - (u64)iA < (u64)iB)
+		if (UINT64_MAX - (u64)ilhs < (u64)irhs)
 			return ATHR_OVERFLOW;
 
-		u64 sum = (u64)iA + (u64)iB;
-		*pA = (i64)sum;
+		u64 sum = (u64)ilhs + (u64)irhs;
+		*plhs = (i64)sum;
 		return (sum <= INT64_MAX) ? ATHR_SIGNED
 					  : ATHR_UNSIGNED;
 	}
 }
 
 enum arithmetic_result
-sqlSubInt64(i64 * pA, bool is_signedA, i64 iB, bool is_signedB)
+sqlSubInt64(i64 * plhs, bool is_lhs_signed, i64 irhs, bool is_rhs_signed)
 {
-	i64 iA = *pA;
+	i64 ilhs = *plhs;
 
-	bool is_negA = iA < 0 && is_signedA;
-	bool is_negB = iB < 0 && is_signedB;
+	bool is_lhs_neg = ilhs < 0 && is_lhs_signed;
+	bool is_rhs_neg = irhs < 0 && is_rhs_signed;
 
-	if (is_negA) {
-		if (!is_signedB){
-			assert((u64)iB > INT64_MAX);
+	if (is_lhs_neg) {
+		if (!is_rhs_signed){
+			assert((u64)irhs > INT64_MAX);
 			return ATHR_OVERFLOW;
 		}
 
-		if (iB == INT64_MIN)
+		if (irhs == INT64_MIN)
 			return ATHR_OVERFLOW;
 		else
-			return sqlAddInt64(pA, true, -iB, true);
+			return sqlAddInt64(plhs, true, -irhs, true);
 	}
 
-	if (is_negB) {
-		/* iA - (-iB) => iA + iB */
-		u64 uB = mod64(iB, true);
-		if (iB == INT64_MIN)
-			is_signedB = false;
+	if (is_rhs_neg) {
+		/* ilhs - (-irhs) => ilhs + irhs */
+		u64 uB = mod64(irhs, true);
+		if (irhs == INT64_MIN)
+			is_rhs_signed = false;
 
-		return sqlAddInt64(pA, is_signedA, uB, is_signedB);
+		return sqlAddInt64(plhs, is_lhs_signed, uB, is_rhs_signed);
 	} else {
-		/* Both iA & iB are positive */
-		if ((u64)iA < (u64)iB) {
+		/* Both ilhs & irhs are positive */
+		if ((u64)ilhs < (u64)irhs) {
 			/* subtract with sign changing */
-			u64 val = (u64)iB - (u64)iA;
-			return apply_sign(pA, val, true);
+			u64 val = (u64)irhs - (u64)ilhs;
+			return apply_sign(plhs, val, true);
 		} else {
-			u64 val = (u64)iA - (u64)iB;
-			*pA = (i64)val;
+			u64 val = (u64)ilhs - (u64)irhs;
+			*plhs = (i64)val;
 			return (val > INT64_MAX) ? ATHR_UNSIGNED
 						 : ATHR_SIGNED;
 		}
@@ -1378,56 +1379,56 @@ sqlSubInt64(i64 * pA, bool is_signedA, i64 iB, bool is_signedB)
 }
 
 enum arithmetic_result
-sqlMulInt64(i64 * pA, bool is_signedA, i64 iB, bool is_signedB)
+sqlMulInt64(i64 * plhs, bool is_lhs_signed, i64 irhs, bool is_rhs_signed)
 {
-	if (*pA == 0 || iB == 0) {
-		*pA = 0;
+	if (*plhs == 0 || irhs == 0) {
+		*plhs = 0;
 		return ATHR_SIGNED;
 	}
 
-	bool is_negA = *pA < 0 && is_signedA;
-	bool is_negB = iB < 0 && is_signedB;
+	bool is_lhs_neg = *plhs < 0 && is_lhs_signed;
+	bool is_rhs_neg = irhs < 0 && is_rhs_signed;
 
-	bool is_neg = is_negA != is_negB;
+	bool is_neg = is_lhs_neg != is_rhs_neg;
 
-	u64 uA = mod64(*pA, is_signedA);
-	u64 uB = mod64(iB, is_signedB);
+	u64 ulhs = mod64(*plhs, is_lhs_signed);
+	u64 urhs = mod64(irhs, is_rhs_signed);
 
 	if (is_neg) {
-		if (INT64_MIN_MOD / uA < uB)
+		if (INT64_MIN_MOD / ulhs < urhs)
 			return ATHR_OVERFLOW;
 	} else {
-		if (UINT64_MAX / uA < uB)
+		if (UINT64_MAX / ulhs < urhs)
 			return ATHR_OVERFLOW;
 	}
 
-	u64 mul = uA * uB;
-	return apply_sign(pA, mul, is_neg);
+	u64 mul = ulhs * urhs;
+	return apply_sign(plhs, mul, is_neg);
 }
 
 enum arithmetic_result
-sqlDivInt64(i64 * pA, bool is_signedA, i64 iB, bool is_signedB) {
-	if (*pA == 0)
+sqlDivInt64(i64 * plhs, bool is_lhs_signed, i64 irhs, bool is_rhs_signed) {
+	if (*plhs == 0)
 		return ATHR_SIGNED;
-	if (iB == 0)
+	if (irhs == 0)
 		return ATHR_DIVBYZERO;
 
-	bool is_negA = *pA < 0 && is_signedA;
-	bool is_negB = iB < 0 && is_signedB;
+	bool is_lhs_neg = *plhs < 0 && is_lhs_signed;
+	bool is_rhs_neg = irhs < 0 && is_rhs_signed;
 
-	bool is_neg = is_negA != is_negB;
+	bool is_neg = is_lhs_neg != is_rhs_neg;
 
-	u64 uA = mod64(*pA, is_signedA);
-	u64 uB = mod64(iB, is_signedB);
+	u64 ulhs = mod64(*plhs, is_lhs_signed);
+	u64 urhs = mod64(irhs, is_rhs_signed);
 
-	u64 div = uA / uB;
-	return apply_sign(pA, div, is_neg);
+	u64 div = ulhs / urhs;
+	return apply_sign(plhs, div, is_neg);
 }
 
 enum arithmetic_result
-sqlRemInt64(i64 * pA, bool is_signedA, i64 iB, bool is_signedB) {
+sqlRemInt64(i64 * plhs, bool is_lhs_signed, i64 irhs, bool is_rhs_signed) {
 
-	if (iB == 0)
+	if (irhs == 0)
 		return ATHR_DIVBYZERO;
 	/*
 	 * The sign of the remainder is defined in such
@@ -1437,12 +1438,12 @@ sqlRemInt64(i64 * pA, bool is_signedA, i64 iB, bool is_signedB) {
 	 * The 2nd operand doesn't affect the sign of result.
 	 */
 
-	bool is_neg = *pA < 0 && is_signedA;
-	u64 uA = mod64(*pA, is_signedA);
-	u64 uB = mod64(iB, is_signedB);
+	bool is_neg = *plhs < 0 && is_lhs_signed;
+	u64 ulhs = mod64(*plhs, is_lhs_signed);
+	u64 urhs = mod64(irhs, is_rhs_signed);
 
-	u64 rem = uA % uB;
-	return apply_sign(pA, rem, is_neg);
+	u64 rem = ulhs % urhs;
+	return apply_sign(plhs, rem, is_neg);
 }
 
 /*
diff --git a/src/box/sql/vdbe.c b/src/box/sql/vdbe.c
index 3df5006c8..787fdadab 100644
--- a/src/box/sql/vdbe.c
+++ b/src/box/sql/vdbe.c
@@ -646,6 +646,8 @@ mem_type_to_str(const struct Mem *p)
 		return "BLOB";
 	case MEM_Bool:
 		return "BOOLEAN";
+	case MEM_UInt:
+		return "UNSIGNED";
 	default:
 		unreachable();
 	}
diff --git a/src/box/sql/vdbeInt.h b/src/box/sql/vdbeInt.h
index f986d136c..78e211c97 100644
--- a/src/box/sql/vdbeInt.h
+++ b/src/box/sql/vdbeInt.h
@@ -265,7 +265,9 @@ struct Mem {
  * auxiliary flags.
  */
 enum {
-	MEM_PURE_TYPE_MASK = 0x1f
+	MEM_PURE_TYPE_MASK = MEM_Null | MEM_Str | MEM_Int |
+			     MEM_Real | MEM_Blob | MEM_Bool |
+			     MEM_UInt
 };
 
 
diff --git a/src/box/sql/vdbeapi.c b/src/box/sql/vdbeapi.c
index 66fceb444..fe948dd26 100644
--- a/src/box/sql/vdbeapi.c
+++ b/src/box/sql/vdbeapi.c
@@ -257,83 +257,38 @@ sql_value_text(sql_value * pVal)
 int
 sql_value_type(sql_value * pVal)
 {
-	static const u8 aType[] = {
-		SQL_BLOB,	/* 0x00 */
-		SQL_NULL,	/* 0x01 */
-		SQL_TEXT,	/* 0x02 */
-		SQL_NULL,	/* 0x03 */
-		SQL_INTEGER,	/* 0x04 */
-		SQL_NULL,	/* 0x05 */
-		SQL_INTEGER,	/* 0x06 */
-		SQL_NULL,	/* 0x07 */
-		SQL_FLOAT,	/* 0x08 */
-		SQL_NULL,	/* 0x09 */
-		SQL_FLOAT,	/* 0x0a */
-		SQL_NULL,	/* 0x0b */
-		SQL_INTEGER,	/* 0x0c */
-		SQL_NULL,	/* 0x0d */
-		SQL_INTEGER,	/* 0x0e */
-		SQL_NULL,	/* 0x0f */
-		SQL_BLOB,	/* 0x10 */
-		SQL_NULL,	/* 0x11 */
-		SQL_TEXT,	/* 0x12 */
-		SQL_NULL,	/* 0x13 */
-		SQL_INTEGER,	/* 0x14 */
-		SQL_NULL,	/* 0x15 */
-		SQL_INTEGER,	/* 0x16 */
-		SQL_NULL,	/* 0x17 */
-		SQL_FLOAT,	/* 0x18 */
-		SQL_NULL,	/* 0x19 */
-		SQL_FLOAT,	/* 0x1a */
-		SQL_NULL,	/* 0x1b */
-		SQL_INTEGER,	/* 0x1c */
-		SQL_NULL,	/* 0x1d */
-		SQL_INTEGER,	/* 0x1e */
-		SQL_NULL,	/* 0x1f */
-
-		SQL_BLOB,	/* 0x20 */
-		SQL_NULL,	/* 0x21 */
-		SQL_TEXT,	/* 0x22 */
-		SQL_NULL,	/* 0x23 */
-		SQL_UNSIGNED,	/* 0x24 */
-		SQL_NULL,	/* 0x25 */
-		SQL_UNSIGNED,	/* 0x26 */
-		SQL_NULL,	/* 0x27 */
-		SQL_FLOAT,	/* 0x28 */
-		SQL_NULL,	/* 0x29 */
-		SQL_FLOAT,	/* 0x2a */
-		SQL_NULL,	/* 0x2b */
-		SQL_INTEGER,	/* 0x2c */
-		SQL_NULL,	/* 0x2d */
-		SQL_INTEGER,	/* 0x2e */
-		SQL_NULL,	/* 0x2f */
-		SQL_BLOB,	/* 0x30 */
-		SQL_NULL,	/* 0x31 */
-		SQL_TEXT,	/* 0x32 */
-		SQL_NULL,	/* 0x33 */
-		SQL_INTEGER,	/* 0x34 */
-		SQL_NULL,	/* 0x35 */
-		SQL_INTEGER,	/* 0x36 */
-		SQL_NULL,	/* 0x37 */
-		SQL_FLOAT,	/* 0x38 */
-		SQL_NULL,	/* 0x39 */
-		SQL_FLOAT,	/* 0x3a */
-		SQL_NULL,	/* 0x3b */
-		SQL_INTEGER,	/* 0x3c */
-		SQL_NULL,	/* 0x3d */
-		SQL_INTEGER,	/* 0x3e */
-		SQL_NULL,	/* 0x3f */
-	};
-
-	assert(MEM_Unsigned == 0x20000);
-	/* compress the unsigned bit with the pure
-	 * type bits, to make them applicable for
-	 * array indexing.
+	/*
+	 * The order of the comparisons is essential.
+	 * Once the value change its type,
+	 * e.g. string or blob to integer/real
+	 * the dynamic data are not disposed,
+	 * they are kept together with scalar value.
+	 * Thus the field 'flags' accumulates several
+	 * bits responsible for data type.
+	 * So the right order is following:
+	 * - Null
+	 * - Unsigned integer
+	 * - Signed integer
+	 * - Real
+	 * - Bool
+	 * - String
+	 * - Blob
 	 */
-	u32 offset = ((pVal->flags & MEM_Unsigned) >> 12) |
-		     (pVal->flags & MEM_PURE_TYPE_MASK);
-	assert(offset < 0x40);
-	return aType[offset];
+	if ((pVal->flags & MEM_Null) != 0)
+		return SQL_NULL;
+	if ((pVal->flags & MEM_Unsigned) != 0)
+		return SQL_UNSIGNED;
+	if ((pVal->flags & MEM_Int) != 0)
+		return SQL_INTEGER;
+	if ((pVal->flags & MEM_Real) != 0)
+		return SQL_FLOAT;
+	if ((pVal->flags & MEM_Bool) != 0)
+		return SQL_INTEGER;
+	if ((pVal->flags & MEM_Str) != 0)
+		return SQL_TEXT;
+	if ((pVal->flags & MEM_Blob) != 0)
+		return SQL_BLOB;
+	return SQL_NULL;	/* Unknown type */
 }
 
 /* Make a copy of an sql_value object
@@ -1444,7 +1399,7 @@ sql_bind_uint64(sql_stmt * pStmt, int i, sql_uint64 iValue)
 	Vdbe *p = (Vdbe *) pStmt;
 	rc = vdbeUnbind(p, i);
 	if (rc == SQL_OK) {
-		rc = sql_bind_type(p, i, "INTEGER");
+		rc = sql_bind_type(p, i, "UNSIGNED");
 		sqlVdbeMemSetUInt64(&p->aVar[i - 1], iValue);
 	}
 	return rc;
diff --git a/test/sql/iproto.result b/test/sql/iproto.result
index deb0c5309..1b31e9db3 100644
--- a/test/sql/iproto.result
+++ b/test/sql/iproto.result
@@ -387,7 +387,7 @@ cn:execute('select ? as big_uint', {0xefffffffffffffffULL})
 ---
 - metadata:
   - name: BIG_UINT
-    type: INTEGER
+    type: UNSIGNED
   rows:
   - [17293822569102704639]
 ...
-- 
2.17.1