sql: make VDBE recognize big integers

Adapts auxiliary functions for supporting
arithmetic operations with unsigned integers.
VDBE distinguishes signed and unsigned integers.
SELECT query correctly returns values greater
than INT64_MAX.

Part of #3810

src/box/lua/sql.c

@@ -32,7 +32,10 @@ lua_push_row(struct lua_State *L, struct sql_stmt *stmt)
 		int type = sql_column_type(stmt, i);
 		switch (type) {
 		case SQL_INTEGER:
-			luaL_pushint64(L, sql_column_int64(stmt, i));
+			if (sql_column_is_unsigned(stmt, i))
+				luaL_pushuint64(L, sql_column_int64(stmt, i));
+			else
+				luaL_pushint64(L, sql_column_int64(stmt, i));
 			break;
 		case SQL_FLOAT:
 			lua_pushnumber(L, sql_column_double(stmt, i));

src/box/sql/build.c

@@ -939,10 +939,10 @@ emitNewSysSequenceRecord(Parse *pParse, int reg_seq_id, const char *seq_name)
 
 	/* 5. Minimum  */
 	sqlVdbeAddOp4Dup8(v, OP_Int64, 0, first_col + 5, 0,
-			      (unsigned char*)&min_usigned_long_long, P4_INT64);
+			      (unsigned char*)&min_usigned_long_long, P4_UINT64);
 	/* 6. Maximum  */
 	sqlVdbeAddOp4Dup8(v, OP_Int64, 0, first_col + 6, 0,
-			      (unsigned char*)&max_usigned_long_long, P4_INT64);
+			      (unsigned char*)&max_usigned_long_long, P4_UINT64);
 	/* 7. Start  */
 	sqlVdbeAddOp2(v, OP_Integer, 1, first_col + 7);
 

src/box/sql/expr.c

@@ -3349,6 +3349,9 @@ expr_code_int(struct Parse *parse, struct Expr *expr, bool is_neg,
 			}
 			break;
 		case ATOI_UNSIGNED:
+			sqlVdbeAddOp4Dup8(v, OP_Int64, 0, mem, 0,
+					  (u8 *)&value, P4_UINT64);
+				break;
 		case ATOI_SIGNED:
 			sqlVdbeAddOp4Dup8(v, OP_Int64, 0, mem, 0,
 					  (u8 *)&value, P4_INT64);

src/box/sql/sqlInt.h

@@ -586,6 +586,9 @@ sql_column_text(sql_stmt *,
 int
 sql_column_type(sql_stmt *, int iCol);
 
+bool
+sql_column_is_unsigned(sql_stmt *, int iCol);
+
 sql_value *
 sql_column_value(sql_stmt *,
 		     int iCol);

src/box/sql/vdbe.c

@@ -404,15 +404,20 @@ sql_value_apply_type(
  * numeric type, if has one.  Set the pMem->u.r and pMem->u.i fields
  * accordingly.
  */
-static u16 SQL_NOINLINE computeNumericType(Mem *pMem)
+static u32 SQL_NOINLINE computeNumericType(Mem *pMem)
 {
 	assert((pMem->flags & (MEM_Int|MEM_Real))==0);
 	assert((pMem->flags & (MEM_Str|MEM_Blob))!=0);
 	if (sqlAtoF(pMem->z, &pMem->u.r, pMem->n)==0)
 		return 0;
-	if (sql_atoi64(pMem->z, (int64_t *)&pMem->u.i, pMem->n)==ATOI_SIGNED)
-		return MEM_Int;
-	return MEM_Real;
+	switch(sql_atoi64(pMem->z, (int64_t *)&pMem->u.i, pMem->n)) {
+		case ATOI_SIGNED:
+			return MEM_Int;
+		case ATOI_UNSIGNED:
+			return MEM_Int | MEM_Unsigned;
+		default: /* ATOI_OVERFLOW:*/
+			return MEM_Real;
+	}
 }
 
 /*
@@ -422,8 +427,10 @@ static u16 SQL_NOINLINE computeNumericType(Mem *pMem)
  * Unlike mem_apply_numeric_type(), this routine does not modify pMem->flags.
  * But it does set pMem->u.r and pMem->u.i appropriately.
  */
-static u16 numericType(Mem *pMem)
+static u32 numericType(Mem *pMem)
 {
+	if ((pMem->flags & (MEM_Int|MEM_Unsigned)) == (MEM_Int|MEM_Unsigned))
+		return (MEM_Int|MEM_Unsigned);
 	if (pMem->flags & (MEM_Int|MEM_Real)) {
 		return pMem->flags & (MEM_Int|MEM_Real);
 	}
@@ -1141,6 +1148,8 @@ case OP_Int64: {           /* out2 */
 	pOut = out2Prerelease(p, pOp);
 	assert(pOp->p4.pI64!=0);
 	pOut->u.i = *pOp->p4.pI64;
+	if (pOp->p4type == P4_UINT64)
+		pOut->flags = MEM_Int | MEM_Unsigned;
 	break;
 }
 
@@ -1646,8 +1655,8 @@ case OP_Divide:                /* same as TK_SLASH, in1, in2, out3 */
 case OP_Remainder: {           /* same as TK_REM, in1, in2, out3 */
 	char bIntint;   /* Started out as two integer operands */
 	u32 flags;      /* Combined MEM_* flags from both inputs */
-	u16 type1;      /* Numeric type of left operand */
-	u16 type2;      /* Numeric type of right operand */
+	u32 type1;      /* Numeric type of left operand */
+	u32 type2;      /* Numeric type of right operand */
 	i64 iA;         /* Integer value of left operand */
 	i64 iB;         /* Integer value of right operand */
 	double rA;      /* Real value of left operand */

src/box/sql/vdbe.h

@@ -70,7 +70,7 @@ struct VdbeOp {
 		int i;		/* Integer value if p4type==P4_INT32 */
 		void *p;	/* Generic pointer */
 		char *z;	/* Pointer to data for string (char array) types */
-		i64 *pI64;	/* Used when p4type is P4_INT64 */
+		i64 *pI64;	/* Used when p4type is P4_INT64 or P4_UINT64 */
 		double *pReal;	/* Used when p4type is P4_REAL */
 		FuncDef *pFunc;	/* Used when p4type is P4_FUNCDEF */
 		sql_context *pCtx;	/* Used when p4type is P4_FUNCCTX */
@@ -131,6 +131,7 @@ struct SubProgram {
 #define P4_INTARRAY (-12)	/* P4 is a vector of 32-bit integers */
 #define P4_SUBPROGRAM  (-13)	/* P4 is a pointer to a SubProgram structure */
 #define P4_ADVANCE  (-14)	/* P4 is a pointer to BtreeNext() or BtreePrev() */
+#define P4_UINT64   (-15)	/* P4 is a 64-bit unsigned integer */
 #define P4_FUNCCTX  (-16)	/* P4 is a pointer to an sql_context object */
 #define P4_BOOL     (-17)	/* P4 is a bool value */
 #define P4_PTR      (-18)	/* P4 is a generic pointer */

src/box/sql/vdbeInt.h

@@ -248,6 +248,9 @@ struct Mem {
 #define MEM_Agg       0x4000	/* Mem.z points to an agg function context */
 #define MEM_Zero      0x8000	/* Mem.i contains count of 0s appended to blob */
 #define MEM_Subtype   0x10000	/* Mem.eSubtype is valid */
+#define MEM_Unsigned  0x20000	/* Value is unsigned integer.
+				 * Combine this flag with MEM_Int
+				 * if necessary */
 #ifdef SQL_OMIT_INCRBLOB
 #undef MEM_Zero
 #define MEM_Zero 0x0000

src/box/sql/vdbeapi.c

@@ -1044,6 +1044,14 @@ sql_column_type(sql_stmt * pStmt, int i)
 	return iType;
 }
 
+bool sql_column_is_unsigned(sql_stmt * pStmt, int i)
+{
+	const struct Mem* pMem = columnMem(pStmt, i);
+	if (!pMem)
+		return false;
+	return (pMem->flags & MEM_Unsigned);
+}
+
 enum sql_subtype
 sql_column_subtype(struct sql_stmt *stmt, int i)
 {

src/box/sql/vdbeaux.c

@@ -862,6 +862,7 @@ freeP4(sql * db, int p4type, void *p4)
 		}
 	case P4_REAL:
 	case P4_INT64:
+	case P4_UINT64:
 	case P4_DYNAMIC:
 	case P4_INTARRAY:{
 			sqlDbFree(db, p4);
@@ -1354,6 +1355,10 @@ displayP4(Op * pOp, char *zTemp, int nTemp)
 			sqlXPrintf(&x, "%lld", *pOp->p4.pI64);
 			break;
 		}
+	case P4_UINT64:{
+			sqlXPrintf(&x, "%ull", *pOp->p4.pI64);
+			break;
+	}
 	case P4_INT32:{
 			sqlXPrintf(&x, "%d", pOp->p4.i);
 			break;
@@ -3724,13 +3729,10 @@ vdbe_decode_msgpack_into_mem(const char *buf, struct Mem *mem, uint32_t *len)
 	}
 	case MP_UINT: {
 		uint64_t v = mp_decode_uint(&buf);
-		if (v > INT64_MAX) {
-			diag_set(ClientError, ER_SQL_EXECUTE,
-				 "integer is overflowed");
-			return -1;
-		}
 		mem->u.i = v;
 		mem->flags = MEM_Int;
+		if (v > INT64_MAX)
+			mem->flags |= MEM_Unsigned;
 		break;
 	}
 	case MP_INT: {

src/box/sql/vdbemem.c

@@ -489,6 +489,9 @@ sqlVdbeRealValue(Mem * pMem, double *v)
 	if (pMem->flags & MEM_Real) {
 		*v = pMem->u.r;
 		return 0;
+	} else if (pMem->flags & (MEM_Int | MEM_Unsigned)) {
+		*v = (double)(u64)pMem->u.i;
+		return 0;
 	} else if (pMem->flags & MEM_Int) {
 		*v = (double)pMem->u.i;
 		return 0;
@@ -1723,7 +1726,7 @@ mpstream_encode_vdbe_mem(struct mpstream *stream, struct Mem *var)
 	} else if (var->flags & MEM_Int) {
 		i = var->u.i;
 encode_int:
-		if (var->u.i >= 0)
+		if (var->u.i >= 0 || var->flags & MEM_Unsigned)
 			mpstream_encode_uint(stream, i);
 		else
 			mpstream_encode_int(stream, i);