[Tarantool-patches] [PATCH v2 3/6] sql: rework OP_Seek* opcodes

[Mergen Imeev via Tarantool-patches <tarantool-patches@dev.tarantool.org>]

This patch changes the Seek* opcodes that are used to search in space
using index. After the redesign, searches using these opcodes work
according to the new implicit casting rules. However, currently implicit
cast in these opcodes is not invoked since there is OP_ApplyType before
them. Unnecessary OP_ApplyType calls will be removed in next patch.

Part of 4230
Part of 4470
---
 src/box/sql.c              |  69 +------
 src/box/sql/cursor.c       |  14 --
 src/box/sql/cursor.h       |   1 -
 src/box/sql/mem.c          | 154 +++++++++++++++-
 src/box/sql/mem.h          |   8 +
 src/box/sql/sqlInt.h       |   8 -
 src/box/sql/tarantoolInt.h |   3 +
 src/box/sql/vdbe.c         | 357 +++++++++++++++----------------------
 8 files changed, 311 insertions(+), 303 deletions(-)

diff --git a/src/box/sql.c b/src/box/sql.c
index 433264abe..a6a7864f1 100644
--- a/src/box/sql.c
+++ b/src/box/sql.c
@@ -204,75 +204,20 @@ int tarantoolsqlPrevious(BtCursor *pCur, int *pRes)
 	return cursor_advance(pCur, pRes);
 }
 
-int tarantoolsqlMovetoUnpacked(BtCursor *pCur, UnpackedRecord *pIdxKey,
-				   int *pRes)
+int
+sql_cursor_seek(struct BtCursor *cur, struct Mem *mems, uint32_t len, int *res)
 {
 	struct region *region = &fiber()->gc;
 	size_t used = region_used(region);
-	uint32_t tuple_size;
-	const char *tuple =
-		sql_vdbe_mem_encode_tuple(pIdxKey->aMem, pIdxKey->nField,
-					  &tuple_size, region);
+	uint32_t size;
+	const char *tuple = sql_vdbe_mem_encode_tuple(mems, len, &size, region);
 	if (tuple == NULL)
 		return -1;
-	if (key_alloc(pCur, tuple_size) != 0)
+	if (key_alloc(cur, size) != 0)
 		return -1;
-	memcpy(pCur->key, tuple, tuple_size);
+	memcpy(cur->key, tuple, size);
 	region_truncate(region, used);
-
-	int rc, res_success;
-	switch (pIdxKey->opcode) {
-	default:
-	  /*  "Unexpected opcode" */
-		assert(0);
-	case 255:
-	/* Restore saved state. Just re-seek cursor.
-	   TODO: replace w/ named constant.  */
-		res_success = 0;
-		break;
-	case OP_SeekLT:
-		pCur->iter_type = ITER_LT;
-		res_success = -1; /* item<key */
-		break;
-	case OP_SeekLE:
-		pCur->iter_type = (pCur->hints & OPFLAG_SEEKEQ) != 0 ?
-				  ITER_REQ : ITER_LE;
-		res_success = 0; /* item==key */
-		break;
-	case OP_SeekGE:
-		pCur->iter_type = (pCur->hints & OPFLAG_SEEKEQ) != 0 ?
-				  ITER_EQ : ITER_GE;
-		res_success = 0; /* item==key */
-		break;
-	case OP_SeekGT:
-		pCur->iter_type = ITER_GT;
-		res_success = 1; /* item>key */
-		break;
-	case OP_NoConflict:
-	case OP_NotFound:
-	case OP_Found:
-	case OP_IdxDelete:
-		pCur->iter_type = ITER_EQ;
-		res_success = 0;
-		break;
-	}
-	rc = cursor_seek(pCur, pRes);
-	if (*pRes == 0) {
-		*pRes = res_success;
-		/*
-		 * To select the first item in a row of equal items
-		 * (last item), sql comparator is configured to
-		 * return +1 (-1) if an item equals the key making it
-		 * impossible to distinguish from an item>key (item<key)
-		 * from comparator output alone.
-		 * To make it possible to learn if the current item
-		 * equals the key, the comparator sets eqSeen.
-		 */
-		pIdxKey->eqSeen = 1;
-	} else {
-		*pRes = -1; /* -1 also means EOF */
-	}
-	return rc;
+	return cursor_seek(cur, res);
 }
 
 /*
diff --git a/src/box/sql/cursor.c b/src/box/sql/cursor.c
index bb2dae898..694fd9a7f 100644
--- a/src/box/sql/cursor.c
+++ b/src/box/sql/cursor.c
@@ -132,20 +132,6 @@ sqlCursorPayload(BtCursor *pCur, u32 offset, u32 amt, void *pBuf)
  *                  is larger than pIdxKey.
  */
 
-int
-sqlCursorMovetoUnpacked(BtCursor * pCur,	/* The cursor to be moved */
-			   UnpackedRecord * pIdxKey,	/* Unpacked index key */
-			   int *pRes	/* Write search results here */
-    )
-{
-	assert(pRes);
-	assert(pIdxKey);
-	assert((pCur->curFlags & BTCF_TaCursor) ||
-	       (pCur->curFlags & BTCF_TEphemCursor));
-
-	return tarantoolsqlMovetoUnpacked(pCur, pIdxKey, pRes);
-}
-
 int
 sqlCursorNext(BtCursor *pCur, int *pRes)
 {
diff --git a/src/box/sql/cursor.h b/src/box/sql/cursor.h
index 88e544191..b82d69e9c 100644
--- a/src/box/sql/cursor.h
+++ b/src/box/sql/cursor.h
@@ -60,7 +60,6 @@ void sqlCursorZero(BtCursor *);
  */
 void
 sql_cursor_close(struct BtCursor *cursor);
-int sqlCursorMovetoUnpacked(BtCursor *, UnpackedRecord * pUnKey, int *pRes);
 
 int sqlCursorNext(BtCursor *, int *pRes);
 int sqlCursorPrevious(BtCursor *, int *pRes);
diff --git a/src/box/sql/mem.c b/src/box/sql/mem.c
index 32fdd9add..e654cac41 100644
--- a/src/box/sql/mem.c
+++ b/src/box/sql/mem.c
@@ -45,6 +45,8 @@
 #include "uuid/mp_uuid.h"
 #include "mp_decimal.h"
 
+#define CMP_OLD_NEW(a, b, type) (((a) > (type)(b)) - ((a) < (type)(b)))
+
 /*
  * Make sure pMem->z points to a writable allocation of at least
  * min(n,32) bytes.
@@ -662,6 +664,23 @@ int_to_double_precise(struct Mem *mem)
 	return 0;
 }
 
+/**
+ * Cast MEM with negative INTEGER to DOUBLE. Doesn't fail. The return value
+ * is < 0 if the original value is less than the result, > 0 if the original
+ * value is greater than the result, and 0 if the cast is precise.
+ */
+static inline int
+int_to_double_forced(struct Mem *mem)
+{
+	assert(mem->type == MEM_TYPE_INT);
+	int64_t i = mem->u.i;
+	double d = (double)i;
+	mem->u.r = d;
+	mem->type = MEM_TYPE_DOUBLE;
+	mem->field_type = FIELD_TYPE_DOUBLE;
+	return CMP_OLD_NEW(i, d, int64_t);
+}
+
 static inline int
 uint_to_double_precise(struct Mem *mem)
 {
@@ -676,6 +695,23 @@ uint_to_double_precise(struct Mem *mem)
 	return 0;
 }
 
+/**
+ * Cast MEM with positive INTEGER to DOUBLE. Doesn't fail. The return value
+ * is < 0 if the original value is less than the result, > 0 if the original
+ * value is greater than the result, and 0 if the cast is precise.
+ */
+static inline int
+uint_to_double_forced(struct Mem *mem)
+{
+	assert(mem->type == MEM_TYPE_UINT);
+	uint64_t u = mem->u.u;
+	double d = (double)u;
+	mem->u.r = d;
+	mem->type = MEM_TYPE_DOUBLE;
+	mem->field_type = FIELD_TYPE_DOUBLE;
+	return CMP_OLD_NEW(u, d, uint64_t);
+}
+
 static inline int
 int_to_str0(struct Mem *mem)
 {
@@ -868,6 +904,43 @@ double_to_int_precise(struct Mem *mem)
 	return -1;
 }
 
+/**
+ * Cast MEM with DOUBLE to INTEGER. Doesn't fail. The return value is < 0 if the
+ * original value is less than the result, > 0 if the original value is greater
+ * than the result, and 0 if the cast is precise.
+ */
+static inline int
+double_to_int_forced(struct Mem *mem)
+{
+	assert(mem->type == MEM_TYPE_DOUBLE);
+	double d = mem->u.r;
+	int64_t i;
+	enum mem_type type;
+	int res;
+	if (d < (double)INT64_MIN) {
+		i = INT64_MIN;
+		type = MEM_TYPE_INT;
+		res = -1;
+	} else if (d >= (double)UINT64_MAX) {
+		i = (int64_t)UINT64_MAX;
+		type = MEM_TYPE_UINT;
+		res = 1;
+	} else if (d <= -1.0) {
+		i = (int64_t)d;
+		type = MEM_TYPE_INT;
+		res = CMP_OLD_NEW(d, i, double);
+	} else {
+		uint64_t u = (uint64_t)d;
+		i = (int64_t)u;
+		type = MEM_TYPE_UINT;
+		res = CMP_OLD_NEW(d, u, double);
+	}
+	mem->u.i = i;
+	mem->type = type;
+	mem->field_type = FIELD_TYPE_INTEGER;
+	return res;
+}
+
 static inline int
 double_to_uint(struct Mem *mem)
 {
@@ -898,6 +971,34 @@ double_to_uint_precise(struct Mem *mem)
 	return -1;
 }
 
+/**
+ * Cast MEM with DOUBLE to UNSIGNED. Doesn't fail. The return value is < 0 if
+ * the original value is less than the result, > 0 if the original value is
+ * greater than the result, and 0 if the cast is precise.
+ */
+static inline int
+double_to_uint_forced(struct Mem *mem)
+{
+	assert(mem->type == MEM_TYPE_DOUBLE);
+	double d = mem->u.r;
+	uint64_t u;
+	int res;
+	if (d < 0.0) {
+		u = 0;
+		res = -1;
+	} else if (d >= (double)UINT64_MAX) {
+		u = UINT64_MAX;
+		res = 1;
+	} else {
+		u = (uint64_t)d;
+		res = CMP_OLD_NEW(d, u, double);
+	}
+	mem->u.u = u;
+	mem->type = MEM_TYPE_UINT;
+	mem->field_type = FIELD_TYPE_UNSIGNED;
+	return res;
+}
+
 static inline int
 double_to_str0(struct Mem *mem)
 {
@@ -1274,6 +1375,57 @@ mem_cast_implicit_old(struct Mem *mem, enum field_type type)
 	return -1;
 }
 
+int
+mem_cast_implicit_number(struct Mem *mem, enum field_type type)
+{
+	assert(mem_is_num(mem) && sql_type_is_numeric(type));
+	switch (type) {
+	case FIELD_TYPE_UNSIGNED:
+		switch (mem->type) {
+		case MEM_TYPE_UINT:
+			mem->field_type = FIELD_TYPE_UNSIGNED;
+			return 0;
+		case MEM_TYPE_INT:
+			mem->u.u = 0;
+			mem->type = MEM_TYPE_UINT;
+			mem->field_type = FIELD_TYPE_UNSIGNED;
+			return -1;
+		case MEM_TYPE_DOUBLE:
+			return double_to_uint_forced(mem);
+		default:
+			unreachable();
+		}
+		break;
+	case FIELD_TYPE_DOUBLE:
+		switch (mem->type) {
+		case MEM_TYPE_INT:
+			return int_to_double_forced(mem);
+		case MEM_TYPE_UINT:
+			return uint_to_double_forced(mem);
+		case MEM_TYPE_DOUBLE:
+			return 0;
+		default:
+			unreachable();
+		}
+		break;
+	case FIELD_TYPE_INTEGER:
+		switch (mem->type) {
+		case MEM_TYPE_UINT:
+		case MEM_TYPE_INT:
+			mem->field_type = FIELD_TYPE_INTEGER;
+			return 0;
+		case MEM_TYPE_DOUBLE:
+			return double_to_int_forced(mem);
+		default:
+			unreachable();
+		}
+		break;
+	default:
+		unreachable();
+	}
+	return 0;
+}
+
 int
 mem_get_int(const struct Mem *mem, int64_t *i, bool *is_neg)
 {
@@ -2520,8 +2672,6 @@ sqlVdbeRecordCompareMsgpack(const void *key1,
 			return -rc;
 		}
 	}
-
-	key2->eqSeen = 1;
 	return key2->default_rc;
 }
 
diff --git a/src/box/sql/mem.h b/src/box/sql/mem.h
index 70bbe557b..4f61cbab0 100644
--- a/src/box/sql/mem.h
+++ b/src/box/sql/mem.h
@@ -746,6 +746,14 @@ mem_cast_implicit(struct Mem *mem, enum field_type type);
 int
 mem_cast_implicit_old(struct Mem *mem, enum field_type type);
 
+/**
+ * Cast MEM with numeric value to given numeric type. Doesn't fail. The return
+ * value is < 0 if the original value is less than the result, > 0 if the
+ * original value is greater than the result, and 0 if the cast is precise.
+ */
+int
+mem_cast_implicit_number(struct Mem *mem, enum field_type type);
+
 /**
  * Return value for MEM of INTEGER type. For MEM of all other types convert
  * value of the MEM to INTEGER if possible and return converted value. Original
diff --git a/src/box/sql/sqlInt.h b/src/box/sql/sqlInt.h
index 115c52f96..e2e550978 100644
--- a/src/box/sql/sqlInt.h
+++ b/src/box/sql/sqlInt.h
@@ -1312,13 +1312,6 @@ sql_space_tuple_log_count(struct space *space);
  * at the first key_def->part_count) then default_rc can be set to -1 to
  * cause the search to find the last match, or +1 to cause the search to
  * find the first match.
- *
- * The key comparison functions will set eqSeen to true if they ever
- * get and equal results when comparing this structure to a b-tree record.
- * When default_rc!=0, the search might end up on the record immediately
- * before the first match or immediately after the last match.  The
- * eqSeen field will indicate whether or not an exact match exists in the
- * b-tree.
  */
 struct UnpackedRecord {
 	/** Collation and sort-order information. */
@@ -1328,7 +1321,6 @@ struct UnpackedRecord {
 	i8 default_rc;		/* Comparison result if keys are equal */
 	i8 r1;			/* Value to return if (lhs > rhs) */
 	i8 r2;			/* Value to return if (rhs < lhs) */
-	u8 eqSeen;		/* True if an equality comparison has been seen */
 	u8 opcode;		/* Currently executing opcode that invoked
 				 * movetoUnpacked, used by Tarantool storage layer.
 				 */
diff --git a/src/box/sql/tarantoolInt.h b/src/box/sql/tarantoolInt.h
index 1ded6c709..8fdc50432 100644
--- a/src/box/sql/tarantoolInt.h
+++ b/src/box/sql/tarantoolInt.h
@@ -27,6 +27,9 @@ int tarantoolsqlReplace(struct space *space, const char *tuple,
 			    const char *tuple_end);
 int tarantoolsqlDelete(BtCursor * pCur, u8 flags);
 
+int
+sql_cursor_seek(struct BtCursor *cur, struct Mem *mems, uint32_t len, int *res);
+
 /**
  * Delete entry from space by its key.
  *
diff --git a/src/box/sql/vdbe.c b/src/box/sql/vdbe.c
index 2e147ec0b..371be205e 100644
--- a/src/box/sql/vdbe.c
+++ b/src/box/sql/vdbe.c
@@ -2504,35 +2504,25 @@ case OP_Close: {
 	break;
 }
 
-/* Opcode: SeekGE P1 P2 P3 P4 P5
+/* Opcode: SeekLT P1 P2 P3 P4 P5
  * Synopsis: key=r[P3@P4]
  *
  * If cursor P1 refers to an SQL table (B-Tree that uses integer keys),
- * use the value in register P3 as the key.  If cursor P1 refers
+ * use the value in register P3 as a key. If cursor P1 refers
  * to an SQL index, then P3 is the first in an array of P4 registers
  * that are used as an unpacked index key.
  *
- * Reposition cursor P1 so that  it points to the smallest entry that
- * is greater than or equal to the key value. If there are no records
- * greater than or equal to the key and P2 is not zero, then jump to P2.
- *
- * If the cursor P1 was opened using the OPFLAG_SEEKEQ flag, then this
- * opcode will always land on a record that equally equals the key, or
- * else jump immediately to P2.  When the cursor is OPFLAG_SEEKEQ, this
- * opcode must be followed by an IdxLE opcode with the same arguments.
- * The IdxLE opcode will be skipped if this opcode succeeds, but the
- * IdxLE opcode will be used on subsequent loop iterations.
+ * Reposition cursor P1 so that  it points to the largest entry that
+ * is less than the key value. If there are no records less than
+ * the key and P2 is not zero, then jump to P2.
  *
- * This opcode leaves the cursor configured to move in forward order,
- * from the beginning toward the end.  In other words, the cursor is
- * configured to use Next, not Prev.
+ * This opcode leaves the cursor configured to move in reverse order,
+ * from the end toward the beginning.  In other words, the cursor is
+ * configured to use Prev, not Next.
  *
- * If P5 is not zero, than it is offset of integer fields in input
- * vector. Force corresponding value to be INTEGER, in case it
- * is floating point value. Alongside with that, type of
- * iterator may be changed: a > 1.5 -> a >= 2.
+ * P5 has the same meaning as for SeekGE.
  *
- * See also: Found, NotFound, SeekLt, SeekGt, SeekLe
+ * See also: Found, NotFound, SeekGt, SeekGe, SeekLe
  */
 /* Opcode: SeekGT P1 P2 P3 P4 P5
  * Synopsis: key=r[P3@P4]
@@ -2555,7 +2545,54 @@ case OP_Close: {
  *
  * P5 has the same meaning as for SeekGE.
  */
-/* Opcode: SeekLT P1 P2 P3 P4 P5
+case OP_SeekLT:         /* jump, in3 */
+case OP_SeekGT: {       /* jump, in3 */
+	bool is_lt = pOp->opcode == OP_SeekLT;
+	struct VdbeCursor *cur = p->apCsr[pOp->p1];
+#ifdef SQL_DEBUG
+	cur->seekOp = pOp->opcode;
+#endif
+	cur->nullRow = 0;
+	cur->uc.pCursor->iter_type = is_lt ? ITER_LT : ITER_GT;
+
+	uint32_t len = pOp->p4.i;
+	assert(pOp->p4type == P4_INT32);
+	assert(len <= cur->key_def->part_count);
+	struct Mem *mems = &aMem[pOp->p3];
+	bool is_op_change = false;
+	for (uint32_t i = 0; i < len; ++i) {
+		enum field_type type = cur->key_def->parts[i].type;
+		struct Mem *mem = &mems[i];
+		if (mem_is_field_compatible(mem, type))
+			continue;
+		if (!sql_type_is_numeric(type) || !mem_is_num(mem)) {
+			diag_set(ClientError, ER_SQL_TYPE_MISMATCH,
+				 mem_str(mem), field_type_strs[type]);
+			goto abort_due_to_error;
+		}
+		int cmp = mem_cast_implicit_number(mem, type);
+		is_op_change = is_op_change || (is_lt && cmp > 0) ||
+			       (!is_lt && cmp < 0);
+	}
+	if (is_op_change)
+		cur->uc.pCursor->iter_type = is_lt ? ITER_LE : ITER_GE;
+
+	int res;
+	if (sql_cursor_seek(cur->uc.pCursor, mems, len, &res) != 0)
+		goto abort_due_to_error;
+	assert((res != 0) == (cur->uc.pCursor->eState == CURSOR_INVALID));
+	cur->cacheStatus = CACHE_STALE;
+#ifdef SQL_TEST
+	sql_search_count++;
+#endif
+	assert(pOp->p2 > 0);
+	VdbeBranchTaken(res, 2);
+	if (res != 0)
+		goto jump_to_p2;
+	break;
+}
+
+/* Opcode: SeekLE P1 P2 P3 P4 P5
  * Synopsis: key=r[P3@P4]
  *
  * If cursor P1 refers to an SQL table (B-Tree that uses integer keys),
@@ -2563,227 +2600,119 @@ case OP_Close: {
  * to an SQL index, then P3 is the first in an array of P4 registers
  * that are used as an unpacked index key.
  *
- * Reposition cursor P1 so that  it points to the largest entry that
- * is less than the key value. If there are no records less than
- * the key and P2 is not zero, then jump to P2.
+ * Reposition cursor P1 so that it points to the largest entry that
+ * is less than or equal to the key value. If there are no records
+ * less than or equal to the key and P2 is not zero, then jump to P2.
  *
  * This opcode leaves the cursor configured to move in reverse order,
  * from the end toward the beginning.  In other words, the cursor is
  * configured to use Prev, not Next.
  *
+ * If the cursor P1 was opened using the OPFLAG_SEEKEQ flag, then this
+ * opcode will always land on a record that equally equals the key, or
+ * else jump immediately to P2.  When the cursor is OPFLAG_SEEKEQ, this
+ * opcode must be followed by an IdxGE opcode with the same arguments.
+ * The IdxGE opcode will be skipped if this opcode succeeds, but the
+ * IdxGE opcode will be used on subsequent loop iterations.
+ *
  * P5 has the same meaning as for SeekGE.
  *
- * See also: Found, NotFound, SeekGt, SeekGe, SeekLe
+ * See also: Found, NotFound, SeekGt, SeekGe, SeekLt
  */
-/* Opcode: SeekLE P1 P2 P3 P4 P5
+/* Opcode: SeekGE P1 P2 P3 P4 P5
  * Synopsis: key=r[P3@P4]
  *
  * If cursor P1 refers to an SQL table (B-Tree that uses integer keys),
- * use the value in register P3 as a key. If cursor P1 refers
+ * use the value in register P3 as the key.  If cursor P1 refers
  * to an SQL index, then P3 is the first in an array of P4 registers
  * that are used as an unpacked index key.
  *
- * Reposition cursor P1 so that it points to the largest entry that
- * is less than or equal to the key value. If there are no records
- * less than or equal to the key and P2 is not zero, then jump to P2.
- *
- * This opcode leaves the cursor configured to move in reverse order,
- * from the end toward the beginning.  In other words, the cursor is
- * configured to use Prev, not Next.
+ * Reposition cursor P1 so that  it points to the smallest entry that
+ * is greater than or equal to the key value. If there are no records
+ * greater than or equal to the key and P2 is not zero, then jump to P2.
  *
  * If the cursor P1 was opened using the OPFLAG_SEEKEQ flag, then this
  * opcode will always land on a record that equally equals the key, or
  * else jump immediately to P2.  When the cursor is OPFLAG_SEEKEQ, this
- * opcode must be followed by an IdxGE opcode with the same arguments.
- * The IdxGE opcode will be skipped if this opcode succeeds, but the
- * IdxGE opcode will be used on subsequent loop iterations.
+ * opcode must be followed by an IdxLE opcode with the same arguments.
+ * The IdxLE opcode will be skipped if this opcode succeeds, but the
+ * IdxLE opcode will be used on subsequent loop iterations.
  *
- * P5 has the same meaning as for SeekGE.
+ * This opcode leaves the cursor configured to move in forward order,
+ * from the beginning toward the end.  In other words, the cursor is
+ * configured to use Next, not Prev.
  *
- * See also: Found, NotFound, SeekGt, SeekGe, SeekLt
+ * If P5 is not zero, than it is offset of integer fields in input
+ * vector. Force corresponding value to be INTEGER, in case it
+ * is floating point value. Alongside with that, type of
+ * iterator may be changed: a > 1.5 -> a >= 2.
+ *
+ * See also: Found, NotFound, SeekLt, SeekGt, SeekLe
  */
-case OP_SeekLT:         /* jump, in3 */
 case OP_SeekLE:         /* jump, in3 */
-case OP_SeekGE:         /* jump, in3 */
-case OP_SeekGT: {       /* jump, in3 */
-	int res;           /* Comparison result */
-	int oc;            /* Opcode */
-	VdbeCursor *pC;    /* The cursor to seek */
-	UnpackedRecord r;  /* The key to seek for */
-	int nField;        /* Number of columns or fields in the key */
-	i64 iKey;          /* The id we are to seek to */
-	int eqOnly;        /* Only interested in == results */
-
-	assert(pOp->p1>=0 && pOp->p1<p->nCursor);
-	assert(pOp->p2!=0);
-	pC = p->apCsr[pOp->p1];
-	assert(pC!=0);
-	assert(pC->eCurType==CURTYPE_TARANTOOL);
-	assert(OP_SeekLE == OP_SeekLT+1);
-	assert(OP_SeekGE == OP_SeekLT+2);
-	assert(OP_SeekGT == OP_SeekLT+3);
-	assert(pC->uc.pCursor!=0);
-	oc = pOp->opcode;
-	eqOnly = 0;
-	pC->nullRow = 0;
+case OP_SeekGE: {       /* jump, in3 */
+	bool is_le = pOp->opcode == OP_SeekLE;
+	struct VdbeCursor *cur = p->apCsr[pOp->p1];
 #ifdef SQL_DEBUG
-	pC->seekOp = pOp->opcode;
+	cur->seekOp = pOp->opcode;
 #endif
-	iKey = 0;
-	/*
-	 * In case floating value is intended to be passed to
-	 * iterator over integer field, we must truncate it to
-	 * integer value and change type of iterator:
-	 * a > 1.5 -> a >= 2
-	 */
-	int int_field = pOp->p5;
-	bool is_neg = false;
-
-	if (int_field > 0) {
-		/* The input value in P3 might be of any type: integer, real, string,
-		 * blob, or NULL.  But it needs to be an integer before we can do
-		 * the seek, so convert it.
-		 */
-		pIn3 = &aMem[int_field];
-		if (mem_is_null(pIn3))
-			goto skip_truncate;
-		if (mem_is_str(pIn3))
-			mem_to_number(pIn3);
-		int64_t i;
-		if (mem_get_int(pIn3, &i, &is_neg) != 0) {
-			if (!mem_is_double(pIn3)) {
-				diag_set(ClientError, ER_SQL_TYPE_MISMATCH,
-					 mem_str(pIn3), "integer");
-				goto abort_due_to_error;
-			}
-			double d = pIn3->u.r;
-			assert(d >= (double)INT64_MAX || d < (double)INT64_MIN);
-			/* TODO: add [INT64_MAX, UINT64_MAX) here. */
-			if (d > (double)INT64_MAX)
-				i = INT64_MAX;
-			else if (d < (double)INT64_MIN)
-				i = INT64_MIN;
-			else
-				i = d;
-			is_neg = i < 0;
+	cur->nullRow = 0;
+	bool is_eq = (cur->uc.pCursor->hints & OPFLAG_SEEKEQ) != 0;
+	if (is_le)
+		cur->uc.pCursor->iter_type = is_eq ? ITER_REQ : ITER_LE;
+	else
+		cur->uc.pCursor->iter_type = is_eq ? ITER_EQ : ITER_GE;
+	assert(!is_eq || pOp[1].opcode == OP_IdxLT ||
+	       pOp[1].opcode == OP_IdxGT);
+
+	uint32_t len = pOp->p4.i;
+	assert(pOp->p4type == P4_INT32);
+	assert(len <= cur->key_def->part_count);
+	struct Mem *mems = &aMem[pOp->p3];
+	bool is_op_change = false;
+	bool is_zero = false;
+	for (uint32_t i = 0; i < len; ++i) {
+		enum field_type type = cur->key_def->parts[i].type;
+		struct Mem *mem = &mems[i];
+		if (mem_is_field_compatible(mem, type))
+			continue;
+		if (!sql_type_is_numeric(type) || !mem_is_num(mem)) {
+			diag_set(ClientError, ER_SQL_TYPE_MISMATCH,
+				 mem_str(mem), field_type_strs[type]);
+			goto abort_due_to_error;
 		}
-		iKey = i;
-
-		/* If the P3 value could not be converted into an integer without
-		 * loss of information, then special processing is required...
+		int cmp = mem_cast_implicit_number(mem, type);
+		is_op_change = is_op_change || (is_le && cmp < 0) ||
+			       (!is_le && cmp > 0);
+		/*
+		 * In case search using EQ or REQ, we will not find anything if
+		 * conversion cannot be precise.
 		 */
-		if (!mem_is_int(pIn3)) {
-			if (!mem_is_double(pIn3)) {
-				/* If the P3 value cannot be converted into any kind of a number,
-				 * then the seek is not possible, so jump to P2
-				 */
-				VdbeBranchTaken(1,2); goto jump_to_p2;
-				break;
-			}
-
-			/* If the approximation iKey is larger than the actual real search
-			 * term, substitute >= for > and < for <=. e.g. if the search term
-			 * is 4.9 and the integer approximation 5:
-			 *
-			 *        (x >  4.9)    ->     (x >= 5)
-			 *        (x <= 4.9)    ->     (x <  5)
-			 */
-			if (pIn3->u.r<(double)iKey) {
-				assert(OP_SeekGE==(OP_SeekGT-1));
-				assert(OP_SeekLT==(OP_SeekLE-1));
-				assert((OP_SeekLE & 0x0001)==(OP_SeekGT & 0x0001));
-				if ((oc & 0x0001)==(OP_SeekGT & 0x0001)) oc--;
-			}
-
-			/* If the approximation iKey is smaller than the actual real search
-			 * term, substitute <= for < and > for >=.
-			 */
-			else if (pIn3->u.r>(double)iKey) {
-				assert(OP_SeekLE==(OP_SeekLT+1));
-				assert(OP_SeekGT==(OP_SeekGE+1));
-				assert((OP_SeekLT & 0x0001)==(OP_SeekGE & 0x0001));
-				if ((oc & 0x0001)==(OP_SeekLT & 0x0001)) oc++;
-			}
-		}
+		is_zero = is_zero || (is_eq && cmp != 0);
 	}
-skip_truncate:
-	/*
-	 * For a cursor with the OPFLAG_SEEKEQ hint, only the
-	 * OP_SeekGE and OP_SeekLE opcodes are allowed, and these
-	 * must be immediately followed by an OP_IdxGT or
-	 * OP_IdxLT opcode, respectively, with the same key.
-	 */
-	if ((pC->uc.pCursor->hints & OPFLAG_SEEKEQ) != 0) {
-		eqOnly = 1;
-		assert(pOp->opcode==OP_SeekGE || pOp->opcode==OP_SeekLE);
-		assert(pOp[1].opcode==OP_IdxLT || pOp[1].opcode==OP_IdxGT);
-		assert(pOp[1].p1==pOp[0].p1);
-		assert(pOp[1].p2==pOp[0].p2);
-		assert(pOp[1].p3==pOp[0].p3);
-		assert(pOp[1].p4.i==pOp[0].p4.i);
+	if (is_zero) {
+		assert(pOp->p2 > 0);
+		VdbeBranchTaken(1, 2);
+		goto jump_to_p2;
 	}
+	if (!is_eq && is_op_change)
+		cur->uc.pCursor->iter_type = is_le ? ITER_LT : ITER_GT;
 
-	nField = pOp->p4.i;
-	assert(pOp->p4type==P4_INT32);
-	assert(nField>0);
-	r.key_def = pC->key_def;
-	r.nField = (u16)nField;
-
-	if (int_field > 0)
-		mem_set_int(&aMem[int_field], iKey, is_neg);
-
-	r.default_rc = ((1 & (oc - OP_SeekLT)) ? -1 : +1);
-	assert(oc!=OP_SeekGT || r.default_rc==-1);
-	assert(oc!=OP_SeekLE || r.default_rc==-1);
-	assert(oc!=OP_SeekGE || r.default_rc==+1);
-	assert(oc!=OP_SeekLT || r.default_rc==+1);
-
-	r.aMem = &aMem[pOp->p3];
-#ifdef SQL_DEBUG
-	{ int i; for(i=0; i<r.nField; i++) assert(memIsValid(&r.aMem[i])); }
-#endif
-	r.eqSeen = 0;
-	r.opcode = oc;
-	if (sqlCursorMovetoUnpacked(pC->uc.pCursor, &r, &res) != 0)
+	int res;
+	if (sql_cursor_seek(cur->uc.pCursor, mems, len, &res) != 0)
 		goto abort_due_to_error;
-	if (eqOnly && r.eqSeen==0) {
-		assert(res!=0);
-		goto seek_not_found;
-	}
-	pC->cacheStatus = CACHE_STALE;
+	assert((res != 0) == (cur->uc.pCursor->eState == CURSOR_INVALID));
+	cur->cacheStatus = CACHE_STALE;
 #ifdef SQL_TEST
 	sql_search_count++;
 #endif
-	if (oc>=OP_SeekGE) {  assert(oc==OP_SeekGE || oc==OP_SeekGT);
-		if (res<0 || (res==0 && oc==OP_SeekGT)) {
-			res = 0;
-			if (sqlCursorNext(pC->uc.pCursor, &res) != 0)
-				goto abort_due_to_error;
-		} else {
-			res = 0;
-		}
-	} else {
-		assert(oc==OP_SeekLT || oc==OP_SeekLE);
-		if (res>0 || (res==0 && oc==OP_SeekLT)) {
-			res = 0;
-			if (sqlCursorPrevious(pC->uc.pCursor, &res) != 0)
-				goto abort_due_to_error;
-		} else {
-			/* res might be negative because the table is empty.  Check to
-			 * see if this is the case.
-			 */
-			res = (CURSOR_VALID != pC->uc.pCursor->eState);
-		}
-	}
-			seek_not_found:
-	assert(pOp->p2>0);
-	VdbeBranchTaken(res!=0,2);
-	if (res) {
+	assert(pOp->p2 > 0);
+	VdbeBranchTaken(res, 2);
+	if (res != 0)
 		goto jump_to_p2;
-	} else if (eqOnly) {
-		assert(pOp[1].opcode==OP_IdxLT || pOp[1].opcode==OP_IdxGT);
-		pOp++; /* Skip the OP_IdxLt or OP_IdxGT that follows */
-	}
+	/* Skip the OP_IdxLT/OP_IdxGT that follows if we have EQ. */
+	if (is_eq)
+		pOp++;
 	break;
 }
 
@@ -2910,7 +2839,9 @@ case OP_Found: {        /* jump, in3 */
 			}
 		}
 	}
-	rc = sqlCursorMovetoUnpacked(pC->uc.pCursor, pIdxKey, &res);
+	pC->uc.pCursor->iter_type = ITER_EQ;
+	rc = sql_cursor_seek(pC->uc.pCursor, pIdxKey->aMem, pIdxKey->nField,
+			     &res);
 	if (pFree != NULL)
 		sqlDbFree(db, pFree);
 	if (rc != 0)
@@ -3768,7 +3699,6 @@ case OP_IdxDelete: {
 	VdbeCursor *pC;
 	BtCursor *pCrsr;
 	int res;
-	UnpackedRecord r;
 
 	assert(pOp->p3>0);
 	assert(pOp->p2>0 && pOp->p2+pOp->p3<=(p->nMem+1 - p->nCursor)+1);
@@ -3779,12 +3709,7 @@ case OP_IdxDelete: {
 	pCrsr = pC->uc.pCursor;
 	assert(pCrsr!=0);
 	assert(pOp->p5==0);
-	r.key_def = pC->key_def;
-	r.nField = (u16)pOp->p3;
-	r.default_rc = 0;
-	r.aMem = &aMem[pOp->p2];
-	r.opcode = OP_IdxDelete;
-	if (sqlCursorMovetoUnpacked(pCrsr, &r, &res) != 0)
+	if (sql_cursor_seek(pCrsr, &aMem[pOp->p2], (u16)pOp->p3, &res) != 0)
 		goto abort_due_to_error;
 	if (res==0) {
 		assert(pCrsr->eState == CURSOR_VALID);
-- 
2.25.1