iproto: add pagination options to iproto

The patch adds new iproto keys needed for pagination and extends
tx_process_select, used by IPROTO. IPROTO and NETBOX_IPROTO versions
are updated, iproto feature pagination is introduced, new keys are:

0x2e - IPROTO_AFTER_POSITION - start iteration after passed
iterator position. It has type MP_STR.

0x2f - IPROTO_AFTER_TUPLE - start iteration after passed tuple.
It has type MP_ARRAY.

0x1f - IPROTO_FETCH_POSITION - send position of last fetched tuple
in response. It has type MP_BOOL.

0x35 - IPROTO_POSITION - iterator position, sent in response if
IPROTO_FETCH_POSITION is true. It has type MP_STR.

Part of #7637

NO_CHANGELOG=see later commits
NO_DOC=see later commits

Author: drewdzzz

src/box/iproto.cc

@@ -2081,19 +2081,40 @@ tx_process_select(struct cmsg *m)
 	struct port port;
 	int count;
 	int rc;
+	const char *packed_pos, *packed_pos_end;
+	bool reply_position;
 	struct request *req = &msg->dml;
+	uint32_t region_svp = region_used(&fiber()->gc);
 	if (tx_check_schema(msg->header.schema_version))
 		goto error;
 
 	tx_inject_delay();
+	packed_pos = req->after_position;
+	packed_pos_end = req->after_position_end;
+	if (packed_pos != NULL) {
+		mp_decode_strl(&packed_pos);
+	} else if (req->after_tuple != NULL) {
+		rc = box_index_tuple_position(req->space_id, req->index_id,
+					      req->after_tuple,
+					      req->after_tuple_end,
+					      &packed_pos, &packed_pos_end);
+		if (rc < 0)
+			goto error;
+	}
 	rc = box_select(req->space_id, req->index_id,
 			req->iterator, req->offset, req->limit,
-			req->key, req->key_end, NULL, NULL, false, &port);
+			req->key, req->key_end, &packed_pos, &packed_pos_end,
+			req->fetch_position, &port);
 	if (rc < 0)
 		goto error;
 
 	out = msg->connection->tx.p_obuf;
-	if (iproto_prepare_select(out, &svp) != 0) {
+	reply_position = req->fetch_position && packed_pos != NULL;
+	if (reply_position)
+		rc = iproto_prepare_select_with_position(out, &svp);
+	else
+		rc = iproto_prepare_select(out, &svp);
+	if (rc != 0) {
 		port_destroy(&port);
 		goto error;
 	}
@@ -2103,16 +2124,29 @@ tx_process_select(struct cmsg *m)
 	count = port_dump_msgpack_16(&port, out);
 	port_destroy(&port);
 	if (count < 0) {
-		/* Discard the prepared select. */
-		obuf_rollback_to_svp(out, &svp);
-		goto error;
+		goto discard;
+	}
+	if (reply_position) {
+		assert(packed_pos != NULL);
+		if (iproto_reply_select_with_position(out, &svp,
+						      msg->header.sync,
+						      ::schema_version, count,
+						      packed_pos,
+						      packed_pos_end) != 0)
+			goto discard;
+	} else {
+		iproto_reply_select(out, &svp, msg->header.sync,
+				    ::schema_version, count);
 	}
-	iproto_reply_select(out, &svp, msg->header.sync,
-			    ::schema_version, count);
+	region_truncate(&fiber()->gc, region_svp);
 	iproto_wpos_create(&msg->wpos, out);
 	tx_end_msg(msg, &svp);
 	return;
+discard:
+	/* Discard the prepared select. */
+	obuf_rollback_to_svp(out, &svp);
 error:
+	region_truncate(&fiber()->gc, region_svp);
 	out = msg->connection->tx.p_obuf;
 	svp = obuf_create_svp(out);
 	tx_reply_error(msg);

src/box/iproto_constants.c

@@ -73,10 +73,10 @@ const unsigned char iproto_key_type[IPROTO_KEY_MAX] =
 		/* 0x1c */	MP_UINT,
 		/* 0x1d */	MP_UINT,
 		/* 0x1e */	MP_UINT,
-		/* 0x1f */	MP_UINT,
 	/* }}} */
 
 	/* {{{ body -- all keys */
+	/* 0x1f */	MP_BOOL,  /* IPROTO_FETCH_POSITION */
 	/* 0x20 */	MP_ARRAY, /* IPROTO_KEY */
 	/* 0x21 */	MP_ARRAY, /* IPROTO_TUPLE */
 	/* 0x22 */	MP_STR, /* IPROTO_FUNCTION_NAME */
@@ -91,11 +91,8 @@ const unsigned char iproto_key_type[IPROTO_KEY_MAX] =
 	/* 0x2b */	MP_MAP, /* IPROTO_OPTIONS */
 	/* 0x2c */	MP_ARRAY, /* IPROTO_OLD_TUPLE */
 	/* 0x2d */	MP_ARRAY, /* IPROTO_NEW_TUPLE */
-	/* }}} */
-
-	/* {{{ unused */
-	/* 0x2e */	MP_UINT,
-	/* 0x2f */	MP_UINT,
+	/* 0x2e */	MP_STR, /* IPROTO_AFTER_POSITION */
+	/* 0x2f */	MP_ARRAY, /* IPROTO_AFTER_TUPLE */
 	/* }}} */
 
 	/* {{{ body -- response keys */
@@ -104,10 +101,10 @@ const unsigned char iproto_key_type[IPROTO_KEY_MAX] =
 	/* 0x32 */	MP_ARRAY, /* IPROTO_METADATA */
 	/* 0x33 */	MP_ARRAY, /* IPROTO_BIND_METADATA */
 	/* 0x34 */	MP_UINT, /* IIPROTO_BIND_COUNT */
+	/* 0x35 */	MP_STR, /* IPROTO_POSITION */
 	/* }}} */
 
 	/* {{{ unused */
-	/* 0x35 */	MP_UINT,
 	/* 0x36 */	MP_UINT,
 	/* 0x37 */	MP_UINT,
 	/* 0x38 */	MP_UINT,

src/box/iproto_constants.h

@@ -80,8 +80,12 @@ enum iproto_key {
 	IPROTO_OFFSET = 0x13,
 	IPROTO_ITERATOR = 0x14,
 	IPROTO_INDEX_BASE = 0x15,
-
 	/* Leave a gap between integer values and other keys */
+	/**
+	 * Flag indicating the need to send position of
+	 * last selected tuple in response.
+	 */
+	IPROTO_FETCH_POSITION = 0x1f,
 	IPROTO_KEY = 0x20,
 	IPROTO_TUPLE = 0x21,
 	IPROTO_FUNCTION_NAME = 0x22,
@@ -108,8 +112,12 @@ enum iproto_key {
 	IPROTO_OLD_TUPLE = 0x2c,
 	/** New tuple (i.e. result of DML request). */
 	IPROTO_NEW_TUPLE = 0x2d,
+	/** Position of last selected tuple to start iteration after it. */
+	IPROTO_AFTER_POSITION = 0x2e,
+	/** Last selected tuple to start iteration after it. */
+	IPROTO_AFTER_TUPLE = 0x2f,
 
-	/* Leave a gap between request keys and response keys */
+	/** Response keys. */
 	IPROTO_DATA = 0x30,
 	IPROTO_ERROR_24 = 0x31,
 	/**
@@ -122,6 +130,8 @@ enum iproto_key {
 	IPROTO_METADATA = 0x32,
 	IPROTO_BIND_METADATA = 0x33,
 	IPROTO_BIND_COUNT = 0x34,
+	/** Position of last selected tuple in response. */
+	IPROTO_POSITION = 0x35,
 
 	/* Leave a gap between response keys and SQL keys. */
 	IPROTO_SQL_TEXT = 0x40,

src/box/iproto_features.c

@@ -66,4 +66,6 @@ iproto_features_init(void)
 			    IPROTO_FEATURE_ERROR_EXTENSION);
 	iproto_features_set(&IPROTO_CURRENT_FEATURES,
 			    IPROTO_FEATURE_WATCHERS);
+	iproto_features_set(&IPROTO_CURRENT_FEATURES,
+			    IPROTO_FEATURE_PAGINATION);
 }

src/box/iproto_features.h

@@ -45,6 +45,12 @@ enum iproto_feature_id {
 	 * IPROTO_WATCH, IPROTO_UNWATCH, IPROTO_EVENT commands.
 	 */
 	IPROTO_FEATURE_WATCHERS = 3,
+	/**
+	 * Pagination support:
+	 * IPROTO_AFTER_POSITION, IPROTO_AFTER_TUPLE, IPROTO_FETCH_POSITION
+	 * request fields and IPROTO_POSITION response field.
+	 */
+	IPROTO_FEATURE_PAGINATION = 4,
 	iproto_feature_id_MAX,
 };
 
@@ -60,7 +66,7 @@ struct iproto_features {
  * It should be incremented every time a new feature is added or removed.
  */
 enum {
-	IPROTO_CURRENT_VERSION = 3,
+	IPROTO_CURRENT_VERSION = 4,
 };
 
 /**

src/box/lua/net_box.c

@@ -82,7 +82,7 @@ enum {
 	/**
 	 * IPROTO protocol version supported by the netbox connector.
 	 */
-	NETBOX_IPROTO_VERSION = 3,
+	NETBOX_IPROTO_VERSION = 4,
 };
 
 /**
@@ -2809,6 +2809,8 @@ luaopen_net_box(struct lua_State *L)
 			    IPROTO_FEATURE_ERROR_EXTENSION);
 	iproto_features_set(&NETBOX_IPROTO_FEATURES,
 			    IPROTO_FEATURE_WATCHERS);
+	iproto_features_set(&NETBOX_IPROTO_FEATURES,
+			    IPROTO_FEATURE_PAGINATION);
 
 	lua_pushcfunction(L, luaT_netbox_request_iterator_next);
 	luaT_netbox_request_iterator_next_ref = luaL_ref(L, LUA_REGISTRYINDEX);

src/box/lua/net_box.lua

@@ -56,6 +56,7 @@ local IPROTO_FEATURE_NAMES = {
     [1]     = 'transactions',
     [2]     = 'error_extension',
     [3]     = 'watchers',
+    [4]     = 'pagination',
 }
 
 local REQUEST_OPTION_TYPES = {

src/box/xrow.c

@@ -411,6 +411,10 @@ static const struct iproto_body_bin iproto_body_bin = {
 	0x81, IPROTO_DATA, 0xdd, 0
 };
 
+static const struct iproto_body_bin iproto_body_bin_with_position = {
+	0x82, IPROTO_DATA, 0xdd, 0
+};
+
 /** Return a 4-byte numeric error code, with status flags. */
 static inline uint32_t
 iproto_encode_error(uint32_t error)
@@ -664,21 +668,56 @@ iproto_prepare_header(struct obuf *buf, struct obuf_svp *svp, size_t size)
 	return 0;
 }
 
+/** Reply select with IPROTO_DATA. */
 void
 iproto_reply_select(struct obuf *buf, struct obuf_svp *svp, uint64_t sync,
 		    uint32_t schema_version, uint32_t count)
 {
 	char *pos = (char *) obuf_svp_to_ptr(buf, svp);
 	iproto_header_encode(pos, IPROTO_OK, sync, schema_version,
-			        obuf_size(buf) - svp->used -
-				IPROTO_HEADER_LEN);
+			     obuf_size(buf) - svp->used -
+			     IPROTO_HEADER_LEN);
 
 	struct iproto_body_bin body = iproto_body_bin;
 	body.v_data_len = mp_bswap_u32(count);
 
 	memcpy(pos + IPROTO_HEADER_LEN, &body, sizeof(body));
 }
 
+/** Reply select with IPROTO_DATA and IPROTO_POSITION. */
+int
+iproto_reply_select_with_position(struct obuf *buf, struct obuf_svp *svp,
+				  uint64_t sync, uint32_t schema_version,
+				  uint32_t count, const char *packed_pos,
+				  const char *packed_pos_end)
+{
+	size_t packed_pos_size = packed_pos_end - packed_pos;
+	size_t key_size = mp_sizeof_uint(IPROTO_POSITION);
+	size_t alloc_size = key_size + mp_sizeof_strl(packed_pos_size);
+	char *ptr = obuf_alloc(buf, alloc_size);
+	if (ptr == NULL) {
+		diag_set(OutOfMemory, alloc_size, "obuf_alloc", "ptr");
+		return -1;
+	}
+	ptr = mp_encode_uint(ptr, IPROTO_POSITION);
+	mp_encode_strl(ptr, packed_pos_size);
+	if (obuf_dup(buf, packed_pos,
+		     packed_pos_size) != packed_pos_size) {
+		return -1;
+	}
+
+	char *pos = (char *)obuf_svp_to_ptr(buf, svp);
+	iproto_header_encode(pos, IPROTO_OK, sync, schema_version,
+			     obuf_size(buf) - svp->used -
+			     IPROTO_HEADER_LEN);
+
+	struct iproto_body_bin body = iproto_body_bin_with_position;
+	body.v_data_len = mp_bswap_u32(count);
+
+	memcpy(pos + IPROTO_HEADER_LEN, &body, sizeof(body));
+	return 0;
+}
+
 int
 xrow_decode_sql(const struct xrow_header *row, struct sql_request *request)
 {
@@ -850,6 +889,9 @@ xrow_decode_dml(struct xrow_header *row, struct request *request,
 		case IPROTO_ITERATOR:
 			request->iterator = mp_decode_uint(&value);
 			break;
+		case IPROTO_FETCH_POSITION:
+			request->fetch_position = mp_decode_bool(&value);
+			break;
 		case IPROTO_TUPLE:
 			request->tuple = value;
 			request->tuple_end = data;
@@ -874,6 +916,14 @@ xrow_decode_dml(struct xrow_header *row, struct request *request,
 			request->new_tuple = value;
 			request->new_tuple_end = data;
 			break;
+		case IPROTO_AFTER_POSITION:
+			request->after_position = value;
+			request->after_position_end = data;
+			break;
+		case IPROTO_AFTER_TUPLE:
+			request->after_tuple = value;
+			request->after_tuple_end = data;
+			break;
 		default:
 			break;
 		}
@@ -920,6 +970,17 @@ request_snprint(char *buf, int size, const struct request *request)
 		SNPRINT(total, snprintf, buf, size, ", new_tuple: ");
 		SNPRINT(total, mp_snprint, buf, size, request->new_tuple);
 	}
+	if (request->fetch_position) {
+		SNPRINT(total, snprintf, buf, size, ", fetch_position: true");
+	}
+	if (request->after_position != NULL) {
+		SNPRINT(total, snprintf, buf, size, ", after_position: ");
+		SNPRINT(total, mp_snprint, buf, size, request->after_position);
+	}
+	if (request->after_tuple != NULL) {
+		SNPRINT(total, snprintf, buf, size, ", after_tuple: ");
+		SNPRINT(total, mp_snprint, buf, size, request->after_tuple);
+	}
 	SNPRINT(total, snprintf, buf, size, "}");
 	return total;
 }
@@ -937,6 +998,13 @@ int
 xrow_encode_dml(const struct request *request, struct region *region,
 		struct iovec *iov)
 {
+	assert(request != NULL);
+	/* Select is unexpected here. Hence, pagination option too. */
+	assert(request->header == NULL ||
+	       request->header->type != IPROTO_SELECT);
+	assert(request->after_position == NULL);
+	assert(request->after_tuple == NULL);
+	assert(!request->fetch_position);
 	int iovcnt = 1;
 	const int MAP_LEN_MAX = 40;
 	uint32_t key_len = request->key_end - request->key;

src/box/xrow.h

@@ -202,8 +202,18 @@ struct request {
 	const char *new_tuple;
 	/** End of @new_tuple. */
 	const char *new_tuple_end;
+	/** Packed iterator_position with MP_STR header. */
+	const char *after_position;
+	/** End of @after_position. */
+	const char *after_position_end;
+	/** Last selected tuple to start iteration after it. */
+	const char *after_tuple;
+	/** End of @after_tuple. */
+	const char *after_tuple_end;
 	/** Base field offset for UPDATE/UPSERT, e.g. 0 for C and 1 for Lua. */
 	int index_base;
+	/** Send position of last selected tuple in response if true. */
+	bool fetch_position;
 };
 
 /**
@@ -648,6 +658,22 @@ iproto_prepare_select(struct obuf *buf, struct obuf_svp *svp)
 	return iproto_prepare_header(buf, svp, IPROTO_SELECT_HEADER_LEN);
 }
 
+/**
+ * Prepare the iproto header for a select result set and iterator position.
+ * It is just an alias fot iproto_prepare_select, it is needed for better
+ * code readability.
+ * @param buf Out buffer.
+ * @param svp Savepoint of the header beginning.
+ *
+ * @retval  0 Success.
+ * @retval -1 Memory error.
+ */
+static inline int
+iproto_prepare_select_with_position(struct obuf *buf, struct obuf_svp *svp)
+{
+	return iproto_prepare_select(buf, svp);
+}
+
 /**
  * Write select header to a preallocated buffer.
  * This function doesn't throw (and we rely on this in iproto.cc).
@@ -656,6 +682,15 @@ void
 iproto_reply_select(struct obuf *buf, struct obuf_svp *svp, uint64_t sync,
 		    uint32_t schema_version, uint32_t count);
 
+/**
+ * Write extended select header to a preallocated buffer.
+ */
+int
+iproto_reply_select_with_position(struct obuf *buf, struct obuf_svp *svp,
+				  uint64_t sync, uint32_t schema_version,
+				  uint32_t count, const char *packed_pos,
+				  const char *packed_pos_end);
+
 /**
  * Encode iproto header with IPROTO_OK response code.
  * @param out Encode to.