[tarantool-patches] [PATCH v2 0/7] Remove box.sql.execute

[tarantool-patches] [PATCH v2 0/7] Remove box.sql.execute

[imeevma]

The goal of this patch-set is to make functions from execute.c
the only way to execute SQL statements. This goal includes
similar output for executed SQL statements no matter how they
were executed: through net.box or through box.

This is the second version of patch-set. It is not complete. It
still has no last part, which is replacing box.sql.execute by
box.execute, because it will lead to massive test editing.

The main goal of this version of patch-set is to try new design of
box.sql.execute() and decide to leave it as it is or modify it for
better performance.

https://github.com/tarantool/tarantool/issues/3505
https://github.com/tarantool/tarantool/tree/imeevma/gh-3505-replace-box_sql_execute-by-box_execute

General information of difference from previous version of
patch-set:
- Three patches from last version were moved to different branch.
- The order in which the patches were arranged was slightly
  changed.

A bit about patches of patch-set with comments about changes in
this version:

Patch 1 allows to use SQL query as plain text in sql_request.
Review fixes were made.

Patch 2 creates method for port which will allow us to dump port
directly to Lua stack. This was moved here from position 8.

Patch 3 makes function sql_response_dump() a bit more universal by
removing IPROTO functions from there. The main change from last
version is that xrow_decode_sql() was returned to execute.c.

Patch 4 allows us to design vstream by wrapping mpstream
functions. Nothing was changed for now.

Patch 5 creates interface vstream and its mpstream implementation.
Difference from previous version:
- vstream now is inherited from mpstream.
- vstream.c removed, msgpack implementation moved to mpstream.c
- Implementation of some methods were replaced by mpstream
  functions that were implicitly casted.
- encode_reply_* methods were removed. Their functionality is now
  performed by encode_enum() + encode_... + encode_map_commit().
- Flag is_flatted was moved to sql_response.

Patch 6 creates vstream implementation for Lua and defines new
box.sql.new_execute() function that will become box.execute() in
next vesions.
Difference from previous version:
- Added parameters binding for box.sql.new_execute().
- Lua implementation of vstream methods was moved to lua/sql.c.
- Changed the way errors was pushed to Lua stack.

Patch 10 is created to check that box.sql.new_execute() is able to
pass through tests created for box.sql.execute(). Almost nothing
changes from previous version.

v1: https://www.freelists.org/post/tarantool-patches/PATCH-v1-0010-sql-remove-boxsqlexecute

Kirill Shcherbatov (1):
  box: store sql text and length in sql_request

Mergen Imeev (6):
  box: add method dump_lua to port
  iproto: remove iproto functions from execute.c
  iproto: replace obuf by mpstream in execute.c
  sql: create interface vstream
  lua: create vstream implementation for Lua
  sql: check new box.sql.execute()

 src/box/execute.c      | 260 +++++++++++++++++++++++++++++++++++--------------
 src/box/execute.h      |  39 +++++---
 src/box/iproto.cc      |  31 +++++-
 src/box/lua/call.c     |   1 +
 src/box/lua/schema.lua |  23 +++++
 src/box/lua/sql.c      | 211 +++++++++++++++++++++++----------------
 src/box/port.c         |  22 +++++
 src/box/port.h         |  12 +++
 src/box/vstream.h      | 167 +++++++++++++++++++++++++++++++
 src/box/xrow.c         |  45 ---------
 src/box/xrow.h         |  19 ----
 src/mpstream.c         |  78 +++++++++++++++
 12 files changed, 673 insertions(+), 235 deletions(-)
 create mode 100644 src/box/vstream.h

-- 
2.7.4

[tarantool-patches] [PATCH v2 1/7] box: store sql text and length in sql_request

[imeevma]

Refactored sql_request structure to store pointer to sql string
data and it's length instead of pointer to msgpack
representation.
This is required to use this structure in sql.c where the query
has a different semantics and can be obtained from stack as a C
string.

Needed for #3505.
---
 src/box/execute.c | 6 +++---
 src/box/execute.h | 2 ++
 2 files changed, 5 insertions(+), 3 deletions(-)

diff --git a/src/box/execute.c b/src/box/execute.c
index fb3e08b..72fcd6c 100644
--- a/src/box/execute.c
+++ b/src/box/execute.c
@@ -284,7 +284,8 @@ error:
 			if (sql_bind_list_decode(request, value, region) != 0)
 				return -1;
 		} else {
-			request->sql_text = value;
+			request->sql_text =
+				mp_decode_str(&value, &request->sql_text_len);
 		}
 	}
 	if (request->sql_text == NULL) {
@@ -596,8 +597,7 @@ sql_prepare_and_execute(const struct sql_request *request,
 			struct sql_response *response, struct region *region)
 {
 	const char *sql = request->sql_text;
-	uint32_t len;
-	sql = mp_decode_str(&sql, &len);
+	uint32_t len = request->sql_text_len;
 	struct sqlite3_stmt *stmt;
 	sqlite3 *db = sql_get();
 	if (db == NULL) {
diff --git a/src/box/execute.h b/src/box/execute.h
index 77bfd79..79cee69 100644
--- a/src/box/execute.h
+++ b/src/box/execute.h
@@ -58,6 +58,8 @@ struct sql_request {
 	uint64_t sync;
 	/** SQL statement text. */
 	const char *sql_text;
+	/** Length of the SQL statement text. */
+	uint32_t sql_text_len;
 	/** Array of parameters. */
 	struct sql_bind *bind;
 	/** Length of the @bind. */
-- 
2.7.4

[tarantool-patches] [PATCH v2 2/7] box: add method dump_lua to port

[imeevma]

New method dump_lua dumps saved in port tuples to Lua stack. It
will allow us to call this method without any other interaction
with port.

Needed for #3505
---
 src/box/lua/call.c |  1 +
 src/box/port.c     | 22 ++++++++++++++++++++++
 src/box/port.h     | 12 ++++++++++++
 3 files changed, 35 insertions(+)

diff --git a/src/box/lua/call.c b/src/box/lua/call.c
index 1f20426..52939ae 100644
--- a/src/box/lua/call.c
+++ b/src/box/lua/call.c
@@ -424,6 +424,7 @@ port_lua_dump_plain(struct port *port, uint32_t *size);
 static const struct port_vtab port_lua_vtab = {
 	.dump_msgpack = port_lua_dump,
 	.dump_msgpack_16 = port_lua_dump_16,
+	.dump_lua = NULL,
 	.dump_plain = port_lua_dump_plain,
 	.destroy = port_lua_destroy,
 };
diff --git a/src/box/port.c b/src/box/port.c
index 266cf3d..a65a32d 100644
--- a/src/box/port.c
+++ b/src/box/port.c
@@ -36,6 +36,8 @@
 #include <small/mempool.h>
 #include <fiber.h>
 #include "errinj.h"
+#include "lua/utils.h"
+#include "lua/tuple.h"
 
 static struct mempool port_tuple_entry_pool;
 
@@ -121,6 +123,19 @@ port_tuple_dump_msgpack(struct port *base, struct obuf *out)
 	return 1;
 }
 
+static int
+port_tuple_dump_lua(struct port *base, struct lua_State *L)
+{
+	struct port_tuple *port = port_tuple(base);
+	struct port_tuple_entry *pe;
+	int i = 0;
+	for (pe = port->first; pe != NULL; pe = pe->next) {
+		luaT_pushtuple(L, pe->tuple);
+		lua_rawseti(L, -2, ++i);
+	}
+	return port->size;
+}
+
 void
 port_destroy(struct port *port)
 {
@@ -139,6 +154,12 @@ port_dump_msgpack_16(struct port *port, struct obuf *out)
 	return port->vtab->dump_msgpack_16(port, out);
 }
 
+int
+port_dump_lua(struct port *port, struct lua_State *L)
+{
+	return port->vtab->dump_lua(port, L);
+}
+
 const char *
 port_dump_plain(struct port *port, uint32_t *size)
 {
@@ -161,6 +182,7 @@ port_free(void)
 const struct port_vtab port_tuple_vtab = {
 	.dump_msgpack = port_tuple_dump_msgpack,
 	.dump_msgpack_16 = port_tuple_dump_msgpack_16,
+	.dump_lua = port_tuple_dump_lua,
 	.dump_plain = NULL,
 	.destroy = port_tuple_destroy,
 };
diff --git a/src/box/port.h b/src/box/port.h
index 882bb37..3bd83b0 100644
--- a/src/box/port.h
+++ b/src/box/port.h
@@ -78,6 +78,11 @@ struct port_vtab {
 	 */
 	int (*dump_msgpack_16)(struct port *port, struct obuf *out);
 	/**
+	 * Dump the content of a port to Lua stack.
+	 * On success returns number of entries dumped.
+	 */
+	int (*dump_lua)(struct port *port, struct lua_State *L);
+	/**
 	 * Dump a port content as a plain text into a buffer,
 	 * allocated inside.
 	 */
@@ -185,6 +190,13 @@ int
 port_dump_msgpack_16(struct port *port, struct obuf *out);
 
 /**
+ * Same as port_dump(), but use the legacy Tarantool 1.6
+ * format.
+ */
+int
+port_dump_lua(struct port *port, struct lua_State *L);
+
+/**
  * Dump a port content as a plain text into a buffer,
  * allocated inside.
  * @param port Port with data to dump.
-- 
2.7.4

[tarantool-patches] [PATCH v2 3/7] iproto: remove iproto functions from execute.c

[imeevma]

To make functions in execute.h more universal we should reduce
their dependence on IPROTO. This patch removes IPROTO functions
from execute.c.

Needed for #3505
---
 src/box/execute.c | 48 ++++++++++++++++++++++++++++++------------------
 src/box/execute.h | 13 ++-----------
 src/box/iproto.cc | 10 +++++++++-
 src/box/xrow.c    | 45 ---------------------------------------------
 src/box/xrow.h    | 19 -------------------
 5 files changed, 41 insertions(+), 94 deletions(-)

diff --git a/src/box/execute.c b/src/box/execute.c
index 72fcd6c..d73681d 100644
--- a/src/box/execute.c
+++ b/src/box/execute.c
@@ -534,9 +534,15 @@ sql_get_description(struct sqlite3_stmt *stmt, struct obuf *out,
 		    int column_count)
 {
 	assert(column_count > 0);
-	if (iproto_reply_array_key(out, column_count, IPROTO_METADATA) != 0)
+	int size = mp_sizeof_uint(IPROTO_METADATA) +
+		   mp_sizeof_array(column_count);
+	char *pos = (char *) obuf_alloc(out, size);
+	if (pos == NULL) {
+		diag_set(OutOfMemory, size, "obuf_alloc", "pos");
 		return -1;
-
+	}
+	pos = mp_encode_uint(pos, IPROTO_METADATA);
+	pos = mp_encode_array(pos, column_count);
 	for (int i = 0; i < column_count; ++i) {
 		size_t size = mp_sizeof_map(2) +
 			      mp_sizeof_uint(IPROTO_FIELD_NAME) +
@@ -621,27 +627,28 @@ sql_prepare_and_execute(const struct sql_request *request,
 }
 
 int
-sql_response_dump(struct sql_response *response, struct obuf *out)
+sql_response_dump(struct sql_response *response, int *keys, struct obuf *out)
 {
-	struct obuf_svp header_svp;
-	/* Prepare memory for the iproto header. */
-	if (iproto_prepare_header(out, &header_svp, IPROTO_SQL_HEADER_LEN) != 0)
-		return -1;
 	sqlite3 *db = sql_get();
 	struct sqlite3_stmt *stmt = (struct sqlite3_stmt *) response->prep_stmt;
 	struct port_tuple *port_tuple = (struct port_tuple *) &response->port;
-	int keys, rc = 0, column_count = sqlite3_column_count(stmt);
+	int rc = 0, column_count = sqlite3_column_count(stmt);
 	if (column_count > 0) {
 		if (sql_get_description(stmt, out, column_count) != 0) {
 err:
-			obuf_rollback_to_svp(out, &header_svp);
 			rc = -1;
 			goto finish;
 		}
-		keys = 2;
-		if (iproto_reply_array_key(out, port_tuple->size,
-					   IPROTO_DATA) != 0)
+		*keys = 2;
+		int size = mp_sizeof_uint(IPROTO_DATA) +
+			   mp_sizeof_array(port_tuple->size);
+		char *pos = (char *) obuf_alloc(out, size);
+		if (pos == NULL) {
+			diag_set(OutOfMemory, size, "obuf_alloc", "pos");
 			goto err;
+		}
+		pos = mp_encode_uint(pos, IPROTO_DATA);
+		pos = mp_encode_array(pos, port_tuple->size);
 		/*
 		 * Just like SELECT, SQL uses output format compatible
 		 * with Tarantool 1.6
@@ -651,17 +658,24 @@ err:
 			goto err;
 		}
 	} else {
-		keys = 1;
+		*keys = 1;
 		assert(port_tuple->size == 0);
 		struct stailq *autoinc_id_list =
 			vdbe_autoinc_id_list((struct Vdbe *)stmt);
 		uint32_t map_size = stailq_empty(autoinc_id_list) ? 1 : 2;
-		if (iproto_reply_map_key(out, map_size, IPROTO_SQL_INFO) != 0)
+		int size = mp_sizeof_uint(IPROTO_SQL_INFO) +
+			   mp_sizeof_map(map_size);
+		char *pos = (char *) obuf_alloc(out, size);
+		if (pos == NULL) {
+			diag_set(OutOfMemory, size, "obuf_alloc", "pos");
 			goto err;
+		}
+		pos = mp_encode_uint(pos, IPROTO_SQL_INFO);
+		pos = mp_encode_map(pos, map_size);
 		uint64_t id_count = 0;
 		int changes = db->nChange;
-		int size = mp_sizeof_uint(SQL_INFO_ROW_COUNT) +
-			   mp_sizeof_uint(changes);
+		size = mp_sizeof_uint(SQL_INFO_ROW_COUNT) +
+		       mp_sizeof_uint(changes);
 		if (!stailq_empty(autoinc_id_list)) {
 			struct autoinc_id_entry *id_entry;
 			stailq_foreach_entry(id_entry, autoinc_id_list, link) {
@@ -691,8 +705,6 @@ err:
 			}
 		}
 	}
-	iproto_reply_sql(out, &header_svp, response->sync, schema_version,
-			 keys);
 finish:
 	port_destroy(&response->port);
 	sqlite3_finalize(stmt);
diff --git a/src/box/execute.h b/src/box/execute.h
index 79cee69..5f3d5eb 100644
--- a/src/box/execute.h
+++ b/src/box/execute.h
@@ -104,13 +104,14 @@ struct sql_response {
  * | }                                            |
  * +----------------------------------------------+
  * @param response EXECUTE response.
+ * @param[out] keys number of keys in dumped map.
  * @param out Output buffer.
  *
  * @retval  0 Success.
  * @retval -1 Memory error.
  */
 int
-sql_response_dump(struct sql_response *response, struct obuf *out);
+sql_response_dump(struct sql_response *response, int *keys, struct obuf *out);
 
 /**
  * Parse the EXECUTE request.
@@ -141,16 +142,6 @@ sql_prepare_and_execute(const struct sql_request *request,
 
 #if defined(__cplusplus)
 } /* extern "C" { */
-#include "diag.h"
-
-/** @copydoc sql_request_decode. Throws on error. */
-static inline void
-xrow_decode_sql_xc(const struct xrow_header *row, struct sql_request *request,
-		   struct region *region)
-{
-	if (xrow_decode_sql(row, request, region) != 0)
-		diag_raise();
-}
 #endif
 
 #endif /* TARANTOOL_SQL_EXECUTE_H_INCLUDED */
diff --git a/src/box/iproto.cc b/src/box/iproto.cc
index cd61393..7c11d05 100644
--- a/src/box/iproto.cc
+++ b/src/box/iproto.cc
@@ -1593,8 +1593,16 @@ tx_process_sql(struct cmsg *m)
 	 * become out of date during yield.
 	 */
 	out = msg->connection->tx.p_obuf;
-	if (sql_response_dump(&response, out) != 0)
+	int keys;
+	struct obuf_svp header_svp;
+	/* Prepare memory for the iproto header. */
+	if (iproto_prepare_header(out, &header_svp, IPROTO_SQL_HEADER_LEN) != 0)
 		goto error;
+	if (sql_response_dump(&response, &keys, out) != 0) {
+		obuf_rollback_to_svp(out, &header_svp);
+		goto error;
+	}
+	iproto_reply_sql(out, &header_svp, response.sync, schema_version, keys);
 	iproto_wpos_create(&msg->wpos, out);
 	return;
 error:
diff --git a/src/box/xrow.c b/src/box/xrow.c
index 567b7ed..76c6f81 100644
--- a/src/box/xrow.c
+++ b/src/box/xrow.c
@@ -445,51 +445,6 @@ iproto_prepare_header(struct obuf *buf, struct obuf_svp *svp, size_t size)
 	return 0;
 }
 
-struct PACKED iproto_key_bin {
-	uint8_t key;       /* IPROTO_DATA/METADATA/SQL_INFO */
-	uint8_t mp_type;
-	uint32_t mp_len;
-};
-
-/**
- * Write the key to the buffer by the specified savepoint.
- * @param buf Output buffer.
- * @param type Value type (MP_ARRAY32=0xdd, MP_MAP32=0xdf, ...).
- * @param size Value size (array or map length).
- * @param key Key value (IPROTO_DATA/DESCRIPTION/SQL_INFO).
- *
- * @retval  0 Success.
- * @retval -1 Memory error.
- */
-static inline int
-iproto_reply_key(struct obuf *buf, uint8_t type, uint32_t size, uint8_t key)
-{
-	char *pos = (char *) obuf_alloc(buf, IPROTO_KEY_HEADER_LEN);
-	if (pos == NULL) {
-		diag_set(OutOfMemory, IPROTO_KEY_HEADER_LEN, "obuf_alloc",
-			 "pos");
-		return -1;
-	}
-	struct iproto_key_bin bin;
-	bin.key = key;
-	bin.mp_type = type;
-	bin.mp_len = mp_bswap_u32(size);
-	memcpy(pos, &bin, sizeof(bin));
-	return 0;
-}
-
-int
-iproto_reply_array_key(struct obuf *buf, uint32_t size, uint8_t key)
-{
-	return iproto_reply_key(buf, 0xdd, size, key);
-}
-
-int
-iproto_reply_map_key(struct obuf *buf, uint32_t size, uint8_t key)
-{
-	return iproto_reply_key(buf, 0xdf, size, key);
-}
-
 void
 iproto_reply_select(struct obuf *buf, struct obuf_svp *svp, uint64_t sync,
 		    uint32_t schema_version, uint32_t count)
diff --git a/src/box/xrow.h b/src/box/xrow.h
index 881d860..ca8d04d 100644
--- a/src/box/xrow.h
+++ b/src/box/xrow.h
@@ -53,11 +53,6 @@ enum {
 	/** 7 = sizeof(iproto_body_bin). */
 	IPROTO_SELECT_HEADER_LEN = IPROTO_HEADER_LEN + 7,
 	/**
-	 * mp_sizeof(IPROTO_DATA/METADATA/SQL_INFO) +
-	 * mp_sizeof_array(UINT32_MAX).
-	 */
-	IPROTO_KEY_HEADER_LEN = 1 + 5,
-	/**
 	 * Header of message + header of body with one or two
 	 * keys: IPROTO_DATA and IPROTO_METADATA or
 	 * IPROTO_SQL_INFO. 1 == mp_sizeof_map(<=15).
@@ -423,20 +418,6 @@ iproto_reply_select(struct obuf *buf, struct obuf_svp *svp, uint64_t sync,
 		    uint32_t schema_version, uint32_t count);
 
 /**
- * Write header of the key to a preallocated buffer by svp.
- * @param buf Buffer to write to.
- * @param size Size of the key (length of the array or of the
- *        string).
- * @param key Body key.
- */
-int
-iproto_reply_array_key(struct obuf *buf, uint32_t size, uint8_t key);
-
-/** @copydoc iproto_reply_array_key. */
-int
-iproto_reply_map_key(struct obuf *buf, uint32_t size, uint8_t key);
-
-/**
  * Encode iproto header with IPROTO_OK response code.
  * @param out Encode to.
  * @param sync Request sync.
-- 
2.7.4

[tarantool-patches] [PATCH v2 4/7] iproto: replace obuf by mpstream in execute.c

[imeevma]

It is useful to replace obuf by mpstream. It allows us to design
vstream. Interface vstream will be used as universal interface to
dump result of SQL queries.

Needed for #3505
---
 src/box/execute.c | 100 +++++++++++++++++-------------------------------------
 src/box/execute.h |   6 ++--
 src/box/iproto.cc |  17 +++++++++-
 3 files changed, 51 insertions(+), 72 deletions(-)

diff --git a/src/box/execute.c b/src/box/execute.c
index d73681d..1f0e5ab 100644
--- a/src/box/execute.c
+++ b/src/box/execute.c
@@ -35,7 +35,6 @@
 #include "sql/sqliteLimit.h"
 #include "errcode.h"
 #include "small/region.h"
-#include "small/obuf.h"
 #include "diag.h"
 #include "sql.h"
 #include "xrow.h"
@@ -43,6 +42,7 @@
 #include "port.h"
 #include "tuple.h"
 #include "sql/vdbe.h"
+#include "mpstream.h"
 
 const char *sql_type_strs[] = {
 	NULL,
@@ -530,23 +530,13 @@ sql_bind(const struct sql_request *request, struct sqlite3_stmt *stmt)
  * @retval -1 Client or memory error.
  */
 static inline int
-sql_get_description(struct sqlite3_stmt *stmt, struct obuf *out,
+sql_get_description(struct sqlite3_stmt *stmt, struct mpstream *stream,
 		    int column_count)
 {
 	assert(column_count > 0);
-	int size = mp_sizeof_uint(IPROTO_METADATA) +
-		   mp_sizeof_array(column_count);
-	char *pos = (char *) obuf_alloc(out, size);
-	if (pos == NULL) {
-		diag_set(OutOfMemory, size, "obuf_alloc", "pos");
-		return -1;
-	}
-	pos = mp_encode_uint(pos, IPROTO_METADATA);
-	pos = mp_encode_array(pos, column_count);
+	mpstream_encode_uint(stream, IPROTO_METADATA);
+	mpstream_encode_array(stream, column_count);
 	for (int i = 0; i < column_count; ++i) {
-		size_t size = mp_sizeof_map(2) +
-			      mp_sizeof_uint(IPROTO_FIELD_NAME) +
-			      mp_sizeof_uint(IPROTO_FIELD_TYPE);
 		const char *name = sqlite3_column_name(stmt, i);
 		const char *type = sqlite3_column_datatype(stmt, i);
 		/*
@@ -557,18 +547,11 @@ sql_get_description(struct sqlite3_stmt *stmt, struct obuf *out,
 		assert(name != NULL);
 		if (type == NULL)
 			type = "UNKNOWN";
-		size += mp_sizeof_str(strlen(name));
-		size += mp_sizeof_str(strlen(type));
-		char *pos = (char *) obuf_alloc(out, size);
-		if (pos == NULL) {
-			diag_set(OutOfMemory, size, "obuf_alloc", "pos");
-			return -1;
-		}
-		pos = mp_encode_map(pos, 2);
-		pos = mp_encode_uint(pos, IPROTO_FIELD_NAME);
-		pos = mp_encode_str(pos, name, strlen(name));
-		pos = mp_encode_uint(pos, IPROTO_FIELD_TYPE);
-		pos = mp_encode_str(pos, type, strlen(type));
+		mpstream_encode_map(stream, 2);
+		mpstream_encode_uint(stream, IPROTO_FIELD_NAME);
+		mpstream_encode_str(stream, name);
+		mpstream_encode_uint(stream, IPROTO_FIELD_TYPE);
+		mpstream_encode_str(stream, type);
 	}
 	return 0;
 }
@@ -627,81 +610,60 @@ sql_prepare_and_execute(const struct sql_request *request,
 }
 
 int
-sql_response_dump(struct sql_response *response, int *keys, struct obuf *out)
+sql_response_dump(struct sql_response *response, int *keys,
+		  struct mpstream *stream)
 {
 	sqlite3 *db = sql_get();
 	struct sqlite3_stmt *stmt = (struct sqlite3_stmt *) response->prep_stmt;
 	struct port_tuple *port_tuple = (struct port_tuple *) &response->port;
 	int rc = 0, column_count = sqlite3_column_count(stmt);
 	if (column_count > 0) {
-		if (sql_get_description(stmt, out, column_count) != 0) {
+		if (sql_get_description(stmt, stream, column_count) != 0) {
 err:
 			rc = -1;
 			goto finish;
 		}
 		*keys = 2;
-		int size = mp_sizeof_uint(IPROTO_DATA) +
-			   mp_sizeof_array(port_tuple->size);
-		char *pos = (char *) obuf_alloc(out, size);
-		if (pos == NULL) {
-			diag_set(OutOfMemory, size, "obuf_alloc", "pos");
-			goto err;
-		}
-		pos = mp_encode_uint(pos, IPROTO_DATA);
-		pos = mp_encode_array(pos, port_tuple->size);
+		mpstream_encode_uint(stream, IPROTO_DATA);
+		mpstream_encode_array(stream, port_tuple->size);
+		mpstream_flush(stream);
 		/*
 		 * Just like SELECT, SQL uses output format compatible
 		 * with Tarantool 1.6
 		 */
-		if (port_dump_msgpack_16(&response->port, out) < 0) {
+		if (port_dump_msgpack_16(&response->port, stream->ctx) < 0) {
 			/* Failed port dump destroyes the port. */
 			goto err;
 		}
+		mpstream_reset(stream);
 	} else {
 		*keys = 1;
 		assert(port_tuple->size == 0);
 		struct stailq *autoinc_id_list =
 			vdbe_autoinc_id_list((struct Vdbe *)stmt);
 		uint32_t map_size = stailq_empty(autoinc_id_list) ? 1 : 2;
-		int size = mp_sizeof_uint(IPROTO_SQL_INFO) +
-			   mp_sizeof_map(map_size);
-		char *pos = (char *) obuf_alloc(out, size);
-		if (pos == NULL) {
-			diag_set(OutOfMemory, size, "obuf_alloc", "pos");
-			goto err;
-		}
-		pos = mp_encode_uint(pos, IPROTO_SQL_INFO);
-		pos = mp_encode_map(pos, map_size);
+		mpstream_encode_uint(stream, IPROTO_SQL_INFO);
+		mpstream_encode_map(stream, map_size);
 		uint64_t id_count = 0;
-		int changes = db->nChange;
-		size = mp_sizeof_uint(SQL_INFO_ROW_COUNT) +
-		       mp_sizeof_uint(changes);
 		if (!stailq_empty(autoinc_id_list)) {
 			struct autoinc_id_entry *id_entry;
-			stailq_foreach_entry(id_entry, autoinc_id_list, link) {
-				size += id_entry->id >= 0 ?
-					mp_sizeof_uint(id_entry->id) :
-					mp_sizeof_int(id_entry->id);
+			stailq_foreach_entry(id_entry, autoinc_id_list, link)
 				id_count++;
-			}
-			size += mp_sizeof_uint(SQL_INFO_AUTOINCREMENT_IDS) +
-				mp_sizeof_array(id_count);
 		}
-		char *buf = obuf_alloc(out, size);
-		if (buf == NULL) {
-			diag_set(OutOfMemory, size, "obuf_alloc", "buf");
-			goto err;
-		}
-		buf = mp_encode_uint(buf, SQL_INFO_ROW_COUNT);
-		buf = mp_encode_uint(buf, changes);
+
+		mpstream_encode_uint(stream, SQL_INFO_ROW_COUNT);
+		mpstream_encode_uint(stream, db->nChange);
 		if (!stailq_empty(autoinc_id_list)) {
-			buf = mp_encode_uint(buf, SQL_INFO_AUTOINCREMENT_IDS);
-			buf = mp_encode_array(buf, id_count);
+			mpstream_encode_uint(stream,
+					     SQL_INFO_AUTOINCREMENT_IDS);
+			mpstream_encode_array(stream, id_count);
 			struct autoinc_id_entry *id_entry;
 			stailq_foreach_entry(id_entry, autoinc_id_list, link) {
-				buf = id_entry->id >= 0 ?
-				      mp_encode_uint(buf, id_entry->id) :
-				      mp_encode_int(buf, id_entry->id);
+				int64_t value = id_entry->id;
+				if (id_entry->id >= 0)
+					mpstream_encode_uint(stream, value);
+				else
+					mpstream_encode_int(stream, value);
 			}
 		}
 	}
diff --git a/src/box/execute.h b/src/box/execute.h
index 5f3d5eb..940f3a3 100644
--- a/src/box/execute.h
+++ b/src/box/execute.h
@@ -52,6 +52,7 @@ struct obuf;
 struct region;
 struct sql_bind;
 struct xrow_header;
+struct mpstream;
 
 /** EXECUTE request. */
 struct sql_request {
@@ -105,13 +106,14 @@ struct sql_response {
  * +----------------------------------------------+
  * @param response EXECUTE response.
  * @param[out] keys number of keys in dumped map.
- * @param out Output buffer.
+ * @param stream stream to where result is written.
  *
  * @retval  0 Success.
  * @retval -1 Memory error.
  */
 int
-sql_response_dump(struct sql_response *response, int *keys, struct obuf *out);
+sql_response_dump(struct sql_response *response, int *keys,
+		  struct mpstream *stream);
 
 /**
  * Parse the EXECUTE request.
diff --git a/src/box/iproto.cc b/src/box/iproto.cc
index 7c11d05..b110900 100644
--- a/src/box/iproto.cc
+++ b/src/box/iproto.cc
@@ -61,6 +61,7 @@
 #include "rmean.h"
 #include "execute.h"
 #include "errinj.h"
+#include "mpstream.h"
 
 enum {
 	IPROTO_SALT_SIZE = 32,
@@ -1573,12 +1574,20 @@ error:
 	tx_reply_error(msg);
 }
 
+/** Callback to forward and error from mpstream methods. */
+static void
+set_encode_error(void *error_ctx)
+{
+	*(bool *)error_ctx = true;
+}
+
 static void
 tx_process_sql(struct cmsg *m)
 {
 	struct iproto_msg *msg = tx_accept_msg(m);
 	struct obuf *out;
 	struct sql_response response;
+	bool is_error = false;
 
 	tx_fiber_init(msg->connection->session, msg->header.sync);
 
@@ -1598,10 +1607,16 @@ tx_process_sql(struct cmsg *m)
 	/* Prepare memory for the iproto header. */
 	if (iproto_prepare_header(out, &header_svp, IPROTO_SQL_HEADER_LEN) != 0)
 		goto error;
-	if (sql_response_dump(&response, &keys, out) != 0) {
+	struct mpstream stream;
+	mpstream_init(&stream, out, obuf_reserve_cb, obuf_alloc_cb,
+		      set_encode_error, &is_error);
+	if (is_error)
+		goto error;
+	if (sql_response_dump(&response, &keys, &stream) != 0 || is_error) {
 		obuf_rollback_to_svp(out, &header_svp);
 		goto error;
 	}
+	mpstream_flush(&stream);
 	iproto_reply_sql(out, &header_svp, response.sync, schema_version, keys);
 	iproto_wpos_create(&msg->wpos, out);
 	return;
-- 
2.7.4

[tarantool-patches] [PATCH v2 5/7] sql: create interface vstream

[imeevma]

If we want to use functions from execute.h not only in IPROTO we
should create special interface. This interface will allow us to
create different implementations for mpstream and lua_State and
use functions from execute.c without changing them. This patch
creates such interface and its implementation for mpstream and
replaces mpstream functions in execute.c by methods of this
interface.

Needed for #3505
---
 src/box/execute.c |  69 ++++++++++++-----------
 src/box/execute.h |   6 +-
 src/box/iproto.cc |  11 ++--
 src/box/vstream.h | 164 ++++++++++++++++++++++++++++++++++++++++++++++++++++++
 src/mpstream.c    |  78 ++++++++++++++++++++++++++
 5 files changed, 291 insertions(+), 37 deletions(-)
 create mode 100644 src/box/vstream.h

diff --git a/src/box/execute.c b/src/box/execute.c
index 1f0e5ab..8093d9c 100644
--- a/src/box/execute.c
+++ b/src/box/execute.c
@@ -42,7 +42,7 @@
 #include "port.h"
 #include "tuple.h"
 #include "sql/vdbe.h"
-#include "mpstream.h"
+#include "vstream.h"
 
 const char *sql_type_strs[] = {
 	NULL,
@@ -530,12 +530,12 @@ sql_bind(const struct sql_request *request, struct sqlite3_stmt *stmt)
  * @retval -1 Client or memory error.
  */
 static inline int
-sql_get_description(struct sqlite3_stmt *stmt, struct mpstream *stream,
+sql_get_description(struct sqlite3_stmt *stmt, struct vstream *stream,
 		    int column_count)
 {
 	assert(column_count > 0);
-	mpstream_encode_uint(stream, IPROTO_METADATA);
-	mpstream_encode_array(stream, column_count);
+	vstream_encode_enum(stream, IPROTO_METADATA, "metadata");
+	vstream_encode_array(stream, column_count);
 	for (int i = 0; i < column_count; ++i) {
 		const char *name = sqlite3_column_name(stmt, i);
 		const char *type = sqlite3_column_datatype(stmt, i);
@@ -547,12 +547,19 @@ sql_get_description(struct sqlite3_stmt *stmt, struct mpstream *stream,
 		assert(name != NULL);
 		if (type == NULL)
 			type = "UNKNOWN";
-		mpstream_encode_map(stream, 2);
-		mpstream_encode_uint(stream, IPROTO_FIELD_NAME);
-		mpstream_encode_str(stream, name);
-		mpstream_encode_uint(stream, IPROTO_FIELD_TYPE);
-		mpstream_encode_str(stream, type);
+		vstream_encode_map(stream, 2);
+
+		vstream_encode_enum(stream, IPROTO_FIELD_NAME, "name");
+		vstream_encode_str(stream, name);
+		vstream_encode_map_commit(stream);
+
+		vstream_encode_enum(stream, IPROTO_FIELD_TYPE, "type");
+		vstream_encode_str(stream, type);
+		vstream_encode_map_commit(stream);
+
+		vstream_encode_array_commit(stream, i);
 	}
+	vstream_encode_map_commit(stream);
 	return 0;
 }
 
@@ -611,7 +618,7 @@ sql_prepare_and_execute(const struct sql_request *request,
 
 int
 sql_response_dump(struct sql_response *response, int *keys,
-		  struct mpstream *stream)
+		  struct vstream *stream)
 {
 	sqlite3 *db = sql_get();
 	struct sqlite3_stmt *stmt = (struct sqlite3_stmt *) response->prep_stmt;
@@ -624,48 +631,48 @@ err:
 			goto finish;
 		}
 		*keys = 2;
-		mpstream_encode_uint(stream, IPROTO_DATA);
-		mpstream_encode_array(stream, port_tuple->size);
-		mpstream_flush(stream);
-		/*
-		 * Just like SELECT, SQL uses output format compatible
-		 * with Tarantool 1.6
-		 */
-		if (port_dump_msgpack_16(&response->port, stream->ctx) < 0) {
-			/* Failed port dump destroyes the port. */
+		vstream_encode_enum(stream, IPROTO_DATA, "rows");
+		vstream_encode_array(stream, port_tuple->size);
+		if (vstream_encode_port(stream, &response->port) < 0)
 			goto err;
-		}
-		mpstream_reset(stream);
+		vstream_encode_map_commit(stream);
 	} else {
 		*keys = 1;
 		assert(port_tuple->size == 0);
 		struct stailq *autoinc_id_list =
 			vdbe_autoinc_id_list((struct Vdbe *)stmt);
 		uint32_t map_size = stailq_empty(autoinc_id_list) ? 1 : 2;
-		mpstream_encode_uint(stream, IPROTO_SQL_INFO);
-		mpstream_encode_map(stream, map_size);
 		uint64_t id_count = 0;
 		if (!stailq_empty(autoinc_id_list)) {
 			struct autoinc_id_entry *id_entry;
 			stailq_foreach_entry(id_entry, autoinc_id_list, link)
 				id_count++;
 		}
-
-		mpstream_encode_uint(stream, SQL_INFO_ROW_COUNT);
-		mpstream_encode_uint(stream, db->nChange);
+		if (!response->is_flatten) {
+			vstream_encode_enum(stream, IPROTO_SQL_INFO, "info");
+			vstream_encode_map(stream, map_size);
+		}
+		vstream_encode_enum(stream, SQL_INFO_ROW_COUNT, "rowcount");
+		vstream_encode_uint(stream, db->nChange);
+		vstream_encode_map_commit(stream);
 		if (!stailq_empty(autoinc_id_list)) {
-			mpstream_encode_uint(stream,
-					     SQL_INFO_AUTOINCREMENT_IDS);
-			mpstream_encode_array(stream, id_count);
+			vstream_encode_enum(stream, SQL_INFO_AUTOINCREMENT_IDS,
+					    "autoincrement_ids");
+			vstream_encode_array(stream, id_count);
 			struct autoinc_id_entry *id_entry;
+			int i = 0;
 			stailq_foreach_entry(id_entry, autoinc_id_list, link) {
 				int64_t value = id_entry->id;
 				if (id_entry->id >= 0)
-					mpstream_encode_uint(stream, value);
+					vstream_encode_uint(stream, value);
 				else
-					mpstream_encode_int(stream, value);
+					vstream_encode_int(stream, value);
+				vstream_encode_array_commit(stream, i++);
 			}
+			vstream_encode_map_commit(stream);
 		}
+		if (!response->is_flatten)
+			vstream_encode_map_commit(stream);
 	}
 finish:
 	port_destroy(&response->port);
diff --git a/src/box/execute.h b/src/box/execute.h
index 940f3a3..5a11a8a 100644
--- a/src/box/execute.h
+++ b/src/box/execute.h
@@ -52,7 +52,7 @@ struct obuf;
 struct region;
 struct sql_bind;
 struct xrow_header;
-struct mpstream;
+struct vstream;
 
 /** EXECUTE request. */
 struct sql_request {
@@ -75,6 +75,8 @@ struct sql_response {
 	struct port port;
 	/** Prepared SQL statement with metadata. */
 	void *prep_stmt;
+	/** Result should be flatten if true. */
+	bool is_flatten;
 };
 
 /**
@@ -113,7 +115,7 @@ struct sql_response {
  */
 int
 sql_response_dump(struct sql_response *response, int *keys,
-		  struct mpstream *stream);
+		  struct vstream *stream);
 
 /**
  * Parse the EXECUTE request.
diff --git a/src/box/iproto.cc b/src/box/iproto.cc
index b110900..380a6ee 100644
--- a/src/box/iproto.cc
+++ b/src/box/iproto.cc
@@ -62,6 +62,7 @@
 #include "execute.h"
 #include "errinj.h"
 #include "mpstream.h"
+#include "vstream.h"
 
 enum {
 	IPROTO_SALT_SIZE = 32,
@@ -1607,16 +1608,18 @@ tx_process_sql(struct cmsg *m)
 	/* Prepare memory for the iproto header. */
 	if (iproto_prepare_header(out, &header_svp, IPROTO_SQL_HEADER_LEN) != 0)
 		goto error;
-	struct mpstream stream;
-	mpstream_init(&stream, out, obuf_reserve_cb, obuf_alloc_cb,
-		      set_encode_error, &is_error);
+
+	struct vstream stream;
+	mpstream_init((struct mpstream *)&stream, out, obuf_reserve_cb,
+		      obuf_alloc_cb, set_encode_error, &is_error);
 	if (is_error)
 		goto error;
+	mp_vstream_init_vtab(&stream);
 	if (sql_response_dump(&response, &keys, &stream) != 0 || is_error) {
 		obuf_rollback_to_svp(out, &header_svp);
 		goto error;
 	}
-	mpstream_flush(&stream);
+	mpstream_flush((struct mpstream *)&stream);
 	iproto_reply_sql(out, &header_svp, response.sync, schema_version, keys);
 	iproto_wpos_create(&msg->wpos, out);
 	return;
diff --git a/src/box/vstream.h b/src/box/vstream.h
new file mode 100644
index 0000000..a8dcfc2
--- /dev/null
+++ b/src/box/vstream.h
@@ -0,0 +1,164 @@
+#ifndef TARANTOOL_VSTREAM_H_INCLUDED
+#define TARANTOOL_VSTREAM_H_INCLUDED
+/*
+ * Copyright 2010-2018, Tarantool AUTHORS, please see AUTHORS file.
+ *
+ * Redistribution and use in source and binary forms, with or
+ * without modification, are permitted provided that the following
+ * conditions are met:
+ *
+ * 1. Redistributions of source code must retain the above
+ *    copyright notice, this list of conditions and the
+ *    following disclaimer.
+ *
+ * 2. Redistributions in binary form must reproduce the above
+ *    copyright notice, this list of conditions and the following
+ *    disclaimer in the documentation and/or other materials
+ *    provided with the distribution.
+ *
+ * THIS SOFTWARE IS PROVIDED BY AUTHORS ``AS IS'' AND
+ * ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED
+ * TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
+ * A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL
+ * AUTHORS OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT,
+ * INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL
+ * DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF
+ * SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR
+ * BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF
+ * LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
+ * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF
+ * THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF
+ * SUCH DAMAGE.
+ */
+
+#include "diag.h"
+#include "mpstream.h"
+
+#if defined(__cplusplus)
+extern "C" {
+#endif /* defined(__cplusplus) */
+
+struct vstream;
+struct lua_State;
+struct port;
+
+struct vstream_vtab {
+	void (*encode_array)(struct vstream *stream, uint32_t size);
+	void (*encode_map)(struct vstream *stream, uint32_t size);
+	void (*encode_uint)(struct vstream *stream, uint64_t num);
+	void (*encode_int)(struct vstream *stream, int64_t num);
+	void (*encode_float)(struct vstream *stream, float num);
+	void (*encode_double)(struct vstream *stream, double num);
+	void (*encode_strn)(struct vstream *stream, const char *str,
+			    uint32_t len);
+	void (*encode_nil)(struct vstream *stream);
+	void (*encode_bool)(struct vstream *stream, bool val);
+	void (*encode_enum)(struct vstream *stream, int64_t num,
+			    const char *str);
+	int (*encode_port)(struct vstream *stream, struct port *port);
+	void (*encode_array_commit)(struct vstream *stream, uint32_t id);
+	void (*encode_map_commit)(struct vstream *stream);
+};
+
+struct vstream {
+	/** TODO: Write comment. */
+	union {
+		struct mpstream mpstream;
+		struct lua_State *L;
+	};
+	/** Virtual function table. */
+	const struct vstream_vtab *vtab;
+};
+
+void
+mp_vstream_init_vtab(struct vstream *vstream);
+
+static inline void
+vstream_encode_array(struct vstream *stream, uint32_t size)
+{
+	return stream->vtab->encode_array(stream, size);
+}
+
+static inline void
+vstream_encode_map(struct vstream *stream, uint32_t size)
+{
+	return stream->vtab->encode_map(stream, size);
+}
+
+static inline void
+vstream_encode_uint(struct vstream *stream, uint64_t num)
+{
+	return stream->vtab->encode_uint(stream, num);
+}
+
+static inline void
+vstream_encode_int(struct vstream *stream, int64_t num)
+{
+	return stream->vtab->encode_int(stream, num);
+}
+
+static inline void
+vstream_encode_float(struct vstream *stream, float num)
+{
+	return stream->vtab->encode_float(stream, num);
+}
+
+static inline void
+vstream_encode_double(struct vstream *stream, double num)
+{
+	return stream->vtab->encode_double(stream, num);
+}
+
+static inline void
+vstream_encode_strn(struct vstream *stream, const char *str, uint32_t len)
+{
+	return stream->vtab->encode_strn(stream, str, len);
+}
+
+static inline void
+vstream_encode_str(struct vstream *stream, const char *str)
+{
+	return stream->vtab->encode_strn(stream, str, strlen(str));
+}
+
+static inline void
+vstream_encode_nil(struct vstream *stream)
+{
+	return stream->vtab->encode_nil(stream);
+}
+
+static inline void
+vstream_encode_bool(struct vstream *stream, bool val)
+{
+	return stream->vtab->encode_bool(stream, val);
+}
+
+static inline void
+vstream_encode_enum(struct vstream *stream, int64_t num, const char *str)
+{
+	return stream->vtab->encode_enum(stream, num, str);
+}
+
+static inline int
+vstream_encode_port(struct vstream *stream, struct port *port)
+{
+	return stream->vtab->encode_port(stream, port);
+}
+
+static inline void
+vstream_encode_map_commit(struct vstream *stream)
+{
+	return stream->vtab->encode_map_commit(stream);
+}
+
+static inline void
+vstream_encode_array_commit(struct vstream *stream, uint32_t id)
+{
+	return stream->vtab->encode_array_commit(stream, id);
+}
+
+#if defined(__cplusplus)
+} /* extern "C" */
+#endif /* defined(__cplusplus) */
+
+#endif /* TARANTOOL_VSTREAM_H_INCLUDED */
diff --git a/src/mpstream.c b/src/mpstream.c
index e4f7950..f9943e4 100644
--- a/src/mpstream.c
+++ b/src/mpstream.c
@@ -34,6 +34,11 @@
 #include <stdint.h>
 #include "msgpuck.h"
 
+#include "box/vstream.h"
+#include "box/iproto_constants.h"
+#include "box/port.h"
+#include "box/xrow.h"
+
 void
 mpstream_reserve_slow(struct mpstream *stream, size_t size)
 {
@@ -175,3 +180,76 @@ mpstream_encode_bool(struct mpstream *stream, bool val)
     char *pos = mp_encode_bool(data, val);
     mpstream_advance(stream, pos - data);
 }
+
+typedef void (*encode_array_f)(struct vstream *stream, uint32_t size);
+typedef void (*encode_map_f)(struct vstream *stream, uint32_t size);
+typedef void (*encode_uint_f)(struct vstream *stream, uint64_t num);
+typedef void (*encode_int_f)(struct vstream *stream, int64_t num);
+typedef void (*encode_float_f)(struct vstream *stream, float num);
+typedef void (*encode_double_f)(struct vstream *stream, double num);
+typedef void (*encode_strn_f)(struct vstream *stream, const char *str,
+			      uint32_t len);
+typedef void (*encode_nil_f)(struct vstream *stream);
+typedef void (*encode_bool_f)(struct vstream *stream, bool val);
+
+int
+mp_vstream_encode_port(struct vstream *stream, struct port *port)
+{
+	struct mpstream *mpstream = (struct mpstream *)stream;
+	mpstream_flush(mpstream);
+	/*
+	 * Just like SELECT, SQL uses output format compatible
+	 * with Tarantool 1.6
+	 */
+	if (port_dump_msgpack_16(port, mpstream->ctx) < 0) {
+		/* Failed port dump destroyes the port. */
+		return -1;
+	}
+	mpstream_reset(mpstream);
+	return 0;
+}
+
+void
+mp_vstream_encode_enum(struct vstream *stream, int64_t num, const char *str)
+{
+	(void)str;
+	if (num < 0)
+		mpstream_encode_int((struct mpstream *)stream, num);
+	else
+		mpstream_encode_uint((struct mpstream *)stream, num);
+}
+
+void
+mp_vstream_encode_map_commit(struct vstream *stream)
+{
+	(void)stream;
+}
+
+void
+mp_vstream_encode_array_commit(struct vstream *stream, uint32_t id)
+{
+	(void)stream;
+	(void)id;
+}
+
+const struct vstream_vtab mp_vstream_vtab = {
+	/** encode_array = */ (encode_array_f)mpstream_encode_array,
+	/** encode_map = */ (encode_map_f)mpstream_encode_map,
+	/** encode_uint = */ (encode_uint_f)mpstream_encode_uint,
+	/** encode_int = */ (encode_int_f)mpstream_encode_int,
+	/** encode_float = */ (encode_float_f)mpstream_encode_float,
+	/** encode_double = */ (encode_double_f)mpstream_encode_double,
+	/** encode_strn = */ (encode_strn_f)mpstream_encode_strn,
+	/** encode_nil = */ (encode_nil_f)mpstream_encode_nil,
+	/** encode_bool = */ (encode_bool_f)mpstream_encode_bool,
+	/** encode_enum = */ mp_vstream_encode_enum,
+	/** encode_port = */ mp_vstream_encode_port,
+	/** encode_array_commit = */ mp_vstream_encode_array_commit,
+	/** encode_map_commit = */ mp_vstream_encode_map_commit,
+};
+
+void
+mp_vstream_init_vtab(struct vstream *vstream)
+{
+	vstream->vtab = &mp_vstream_vtab;
+}
-- 
2.7.4

[tarantool-patches] [PATCH v2 6/7] lua: create vstream implementation for Lua

[imeevma]

Thas patch creates vstream implementation for Lua and function
box.sql.new_execute() that uses this implementation. Also it
creates parameters binding for SQL statements executed through
box.

Part of #3505
---
 src/box/execute.c | 132 +++++++++++++++++++++++++++++++++++++++++++++++++
 src/box/execute.h |  15 ++++++
 src/box/lua/sql.c | 144 ++++++++++++++++++++++++++++++++++++++++++++++++++++++
 src/box/vstream.h |   3 ++
 4 files changed, 294 insertions(+)

diff --git a/src/box/execute.c b/src/box/execute.c
index 8093d9c..44cc7ea 100644
--- a/src/box/execute.c
+++ b/src/box/execute.c
@@ -43,6 +43,8 @@
 #include "tuple.h"
 #include "sql/vdbe.h"
 #include "vstream.h"
+#include "lua/utils.h"
+#include "lua/msgpack.h"
 
 const char *sql_type_strs[] = {
 	NULL,
@@ -298,6 +300,136 @@ error:
 	return 0;
 }
 
+static inline int
+lua_sql_bind_decode(struct lua_State *L, struct sql_bind *bind, int i,
+		    struct luaL_field *field)
+{
+	bind->pos = i + 1;
+	if (field->type == MP_MAP) {
+		/*
+		 * A named parameter is an MP_MAP with
+		 * one key - {'name': value}.
+		 * Report parse error otherwise.
+		 */
+		if (field->size != 1) {
+			diag_set(ClientError, ER_INVALID_MSGPACK,
+				 "SQL bind parameter");
+			return -1;
+		}
+		lua_pushnil(L);
+		lua_next(L, lua_gettop(L) - 1);
+		struct luaL_field field_name;
+		lua_pushvalue(L, -2);
+		luaL_tofield(L, luaL_msgpack_default, -1, &field_name);
+		lua_pop(L, 1);
+		assert(field_name.type == MP_STR);
+		luaL_tofield(L, luaL_msgpack_default, -1, field);
+		lua_pop(L, 1);
+		bind->name = field_name.sval.data;
+		bind->name_len = field_name.sval.len;
+	} else {
+		bind->name = NULL;
+		bind->name_len = 0;
+	}
+	switch (field->type) {
+	case MP_UINT: {
+		bind->i64 = field->ival;
+		bind->type = SQLITE_INTEGER;
+		bind->bytes = sizeof(bind->i64);
+		break;
+	}
+	case MP_INT:
+		bind->i64 = field->ival;
+		bind->type = SQLITE_INTEGER;
+		bind->bytes = sizeof(bind->i64);
+		break;
+	case MP_STR:
+		bind->s = field->sval.data;
+		bind->type = SQLITE_TEXT;
+		bind->bytes = field->sval.len;
+		break;
+	case MP_DOUBLE:
+		bind->d = field->dval;
+		bind->type = SQLITE_FLOAT;
+		bind->bytes = sizeof(bind->d);
+		break;
+	case MP_FLOAT:
+		bind->d = field->dval;
+		bind->type = SQLITE_FLOAT;
+		bind->bytes = sizeof(bind->d);
+		break;
+	case MP_NIL:
+		bind->type = SQLITE_NULL;
+		bind->bytes = 1;
+		break;
+	case MP_BOOL:
+		/* SQLite doesn't support boolean. Use int instead. */
+		bind->i64 = field->bval ? 1 : 0;
+		bind->type = SQLITE_INTEGER;
+		bind->bytes = sizeof(bind->i64);
+		break;
+	case MP_BIN:
+		bind->s = field->sval.data;
+		bind->type = SQLITE_BLOB;
+		break;
+	case MP_EXT:
+		bind->s = (const char *)field;
+		bind->bytes = sizeof(*field);
+		bind->type = SQLITE_BLOB;
+		break;
+	case MP_ARRAY:
+		diag_set(ClientError, ER_SQL_BIND_TYPE, "ARRAY",
+			 sql_bind_name(bind));
+		return -1;
+	case MP_MAP:
+		diag_set(ClientError, ER_SQL_BIND_TYPE, "MAP",
+			 sql_bind_name(bind));
+		return -1;
+	default:
+		unreachable();
+	}
+	return 0;
+}
+
+int
+lua_sql_bind_list_decode(struct lua_State *L, struct sql_request *request,
+			 int idx, struct region *region)
+{
+	assert(request != NULL);
+	if (! lua_istable(L, idx)) {
+		diag_set(ClientError, ER_INVALID_MSGPACK, "SQL parameter list");
+		return -1;
+	}
+	uint32_t bind_count = lua_objlen(L, idx);
+	if (bind_count == 0)
+		return 0;
+	if (bind_count > SQL_BIND_PARAMETER_MAX) {
+		diag_set(ClientError, ER_SQL_BIND_PARAMETER_MAX,
+			 (int) bind_count);
+		return -1;
+	}
+	uint32_t used = region_used(region);
+	size_t size = sizeof(struct sql_bind) * bind_count;
+	struct sql_bind *bind = (struct sql_bind *) region_alloc(region, size);
+	if (bind == NULL) {
+		diag_set(OutOfMemory, size, "region_alloc", "struct sql_bind");
+		return -1;
+	}
+	for (uint32_t i = 0; i < bind_count; ++i) {
+		struct luaL_field field;
+		lua_rawgeti(L, idx, i + 1);
+		luaL_tofield(L, luaL_msgpack_default, -1, &field);
+		if (lua_sql_bind_decode(L, &bind[i], i, &field) != 0) {
+			region_truncate(region, used);
+			return -1;
+		}
+		lua_pop(L, 1);
+	}
+	request->bind_count = bind_count;
+	request->bind = bind;
+	return 0;
+}
+
 /**
  * Serialize a single column of a result set row.
  * @param stmt Prepared and started statement. At least one
diff --git a/src/box/execute.h b/src/box/execute.h
index 5a11a8a..8ee0a89 100644
--- a/src/box/execute.h
+++ b/src/box/execute.h
@@ -131,6 +131,21 @@ xrow_decode_sql(const struct xrow_header *row, struct sql_request *request,
 		struct region *region);
 
 /**
+ * Parse Lua table of SQL parameters and store a result
+ * into the @request->bind, bind_count.
+ * @param L Lua stack to get data from.
+ * @param request Request to save decoded parameters.
+ * @param idx Position of table with parameters on Lua stack.
+ * @param region Allocator.
+ *
+ * @retval  0 Success.
+ * @retval -1 Client or memory error.
+ */
+int
+lua_sql_bind_list_decode(struct lua_State *L, struct sql_request *request,
+			 int idx, struct region *region);
+
+/**
  * Prepare and execute an SQL statement.
  * @param request IProto request.
  * @param[out] response Response to store result.
diff --git a/src/box/lua/sql.c b/src/box/lua/sql.c
index 17e2694..05556f9 100644
--- a/src/box/lua/sql.c
+++ b/src/box/lua/sql.c
@@ -6,6 +6,116 @@
 #include "box/info.h"
 #include "lua/utils.h"
 #include "info.h"
+#include "box/execute.h"
+#include "box/vstream.h"
+
+void
+lua_vstream_encode_array(struct vstream *stream, uint32_t size)
+{
+	lua_createtable(stream->L, size, 0);
+}
+
+void
+lua_vstream_encode_map(struct vstream *stream, uint32_t size)
+{
+	(void)size;
+	lua_newtable(stream->L);
+}
+
+void
+lua_vstream_encode_uint(struct vstream *stream, uint64_t num)
+{
+	luaL_pushuint64(stream->L, num);
+}
+
+void
+lua_vstream_encode_int(struct vstream *stream, int64_t num)
+{
+	luaL_pushint64(stream->L, num);
+}
+
+void
+lua_vstream_encode_float(struct vstream *stream, float num)
+{
+	lua_pushnumber(stream->L, num);
+}
+
+void
+lua_vstream_encode_double(struct vstream *stream, double num)
+{
+	lua_pushnumber(stream->L, num);
+}
+
+void
+lua_vstream_encode_strn(struct vstream *stream, const char *str, uint32_t len)
+{
+	lua_pushlstring(stream->L, str, len);
+}
+
+void
+lua_vstream_encode_nil(struct vstream *stream)
+{
+	lua_pushnil(stream->L);
+}
+
+void
+lua_vstream_encode_bool(struct vstream *stream, bool val)
+{
+	lua_pushboolean(stream->L, val);
+}
+
+int
+lua_vstream_encode_port(struct vstream *stream, struct port *port)
+{
+	if (port_dump_lua(port, stream->L) < 0)
+		return -1;
+	return 0;
+}
+
+void
+lua_vstream_encode_enum(struct vstream *stream, int64_t num, const char *str)
+{
+	(void)num;
+	lua_pushlstring(stream->L, str, strlen(str));
+}
+
+void
+lua_vstream_encode_map_commit(struct vstream *stream)
+{
+	size_t length;
+	const char *key = lua_tolstring(stream->L, -2, &length);
+	lua_setfield(stream->L, -3, key);
+	lua_pop(stream->L, 1);
+}
+
+void
+lua_vstream_encode_array_commit(struct vstream *stream, uint32_t id)
+{
+	lua_rawseti(stream->L, -2, id + 1);
+}
+
+const struct vstream_vtab lua_vstream_vtab = {
+	/** encode_array = */ lua_vstream_encode_array,
+	/** encode_map = */ lua_vstream_encode_map,
+	/** encode_uint = */ lua_vstream_encode_uint,
+	/** encode_int = */ lua_vstream_encode_int,
+	/** encode_float = */ lua_vstream_encode_float,
+	/** encode_double = */ lua_vstream_encode_double,
+	/** encode_strn = */ lua_vstream_encode_strn,
+	/** encode_nil = */ lua_vstream_encode_nil,
+	/** encode_bool = */ lua_vstream_encode_bool,
+	/** encode_enum = */ lua_vstream_encode_enum,
+	/** encode_port = */ lua_vstream_encode_port,
+	/** encode_array_commit = */ lua_vstream_encode_array_commit,
+	/** encode_map_commit = */ lua_vstream_encode_map_commit,
+};
+
+void
+lua_vstream_init(struct vstream *vstream, struct lua_State *L)
+{
+	vstream->vtab = &lua_vstream_vtab;
+	vstream->L = L;
+}
 
 static void
 lua_push_column_names(struct lua_State *L, struct sqlite3_stmt *stmt)
@@ -111,6 +221,39 @@ sqlerror:
 }
 
 static int
+lbox_execute(struct lua_State *L)
+{
+	struct sqlite3 *db = sql_get();
+	if (db == NULL)
+		return luaL_error(L, "not ready");
+
+	size_t length;
+	const char *sql = lua_tolstring(L, 1, &length);
+	if (sql == NULL)
+		return luaL_error(L, "usage: box.execute(sqlstring)");
+
+	struct sql_request request = {};
+	request.sql_text = sql;
+	request.sql_text_len = length;
+	if (lua_gettop(L) == 2)
+		if (lua_sql_bind_list_decode(L, &request, 2, &fiber()->gc) != 0)
+			return luaT_error(L);
+	struct sql_response response = {.is_flatten = true};
+	if (sql_prepare_and_execute(&request, &response, &fiber()->gc) != 0)
+		return luaT_error(L);
+
+	int keys;
+	struct vstream vstream;
+	lua_vstream_init(&vstream, L);
+	lua_newtable(L);
+	if (sql_response_dump(&response, &keys, &vstream) != 0) {
+		lua_pop(L, 1);
+		return luaT_error(L);
+	}
+	return 1;
+}
+
+static int
 lua_sql_debug(struct lua_State *L)
 {
 	struct info_handler info;
@@ -124,6 +267,7 @@ box_lua_sqlite_init(struct lua_State *L)
 {
 	static const struct luaL_Reg module_funcs [] = {
 		{"execute", lua_sql_execute},
+		{"new_execute", lbox_execute},
 		{"debug", lua_sql_debug},
 		{NULL, NULL}
 	};
diff --git a/src/box/vstream.h b/src/box/vstream.h
index a8dcfc2..4a553bc 100644
--- a/src/box/vstream.h
+++ b/src/box/vstream.h
@@ -73,6 +73,9 @@ struct vstream {
 void
 mp_vstream_init_vtab(struct vstream *vstream);
 
+void
+lua_vstream_init(struct vstream *vstream, struct lua_State *L);
+
 static inline void
 vstream_encode_array(struct vstream *stream, uint32_t size)
 {
-- 
2.7.4

[tarantool-patches] [PATCH v2 7/7] sql: check new box.sql.execute()

[imeevma]

This commit checks that new implementation of box.sql.execute() is
able to pass all tests. This is temporary commit and should be
dropped later.

Needed for #3505
---
 src/box/execute.c      |  11 ++---
 src/box/execute.h      |   3 +-
 src/box/iproto.cc      |   3 +-
 src/box/lua/schema.lua |  23 ++++++++++
 src/box/lua/sql.c      | 117 ++++---------------------------------------------
 5 files changed, 42 insertions(+), 115 deletions(-)

diff --git a/src/box/execute.c b/src/box/execute.c
index 44cc7ea..433e71e 100644
--- a/src/box/execute.c
+++ b/src/box/execute.c
@@ -697,7 +697,7 @@ sql_get_description(struct sqlite3_stmt *stmt, struct vstream *stream,
 
 static inline int
 sql_execute(sqlite3 *db, struct sqlite3_stmt *stmt, struct port *port,
-	    struct region *region)
+	    struct region *region, int error_id)
 {
 	int rc, column_count = sqlite3_column_count(stmt);
 	if (column_count > 0) {
@@ -714,7 +714,7 @@ sql_execute(sqlite3 *db, struct sqlite3_stmt *stmt, struct port *port,
 		assert(rc != SQLITE_ROW && rc != SQLITE_OK);
 	}
 	if (rc != SQLITE_DONE) {
-		diag_set(ClientError, ER_SQL_EXECUTE, sqlite3_errmsg(db));
+		diag_set(ClientError, error_id, sqlite3_errmsg(db));
 		return -1;
 	}
 	return 0;
@@ -722,7 +722,8 @@ sql_execute(sqlite3 *db, struct sqlite3_stmt *stmt, struct port *port,
 
 int
 sql_prepare_and_execute(const struct sql_request *request,
-			struct sql_response *response, struct region *region)
+			struct sql_response *response, struct region *region,
+			int error_id)
 {
 	const char *sql = request->sql_text;
 	uint32_t len = request->sql_text_len;
@@ -733,7 +734,7 @@ sql_prepare_and_execute(const struct sql_request *request,
 		return -1;
 	}
 	if (sqlite3_prepare_v2(db, sql, len, &stmt, NULL) != SQLITE_OK) {
-		diag_set(ClientError, ER_SQL_EXECUTE, sqlite3_errmsg(db));
+		diag_set(ClientError, error_id, sqlite3_errmsg(db));
 		return -1;
 	}
 	assert(stmt != NULL);
@@ -741,7 +742,7 @@ sql_prepare_and_execute(const struct sql_request *request,
 	response->prep_stmt = stmt;
 	response->sync = request->sync;
 	if (sql_bind(request, stmt) == 0 &&
-	    sql_execute(db, stmt, &response->port, region) == 0)
+	    sql_execute(db, stmt, &response->port, region, error_id) == 0)
 		return 0;
 	port_destroy(&response->port);
 	sqlite3_finalize(stmt);
diff --git a/src/box/execute.h b/src/box/execute.h
index 8ee0a89..46b2ce1 100644
--- a/src/box/execute.h
+++ b/src/box/execute.h
@@ -157,7 +157,8 @@ lua_sql_bind_list_decode(struct lua_State *L, struct sql_request *request,
  */
 int
 sql_prepare_and_execute(const struct sql_request *request,
-			struct sql_response *response, struct region *region);
+			struct sql_response *response, struct region *region,
+			int error_id);
 
 #if defined(__cplusplus)
 } /* extern "C" { */
diff --git a/src/box/iproto.cc b/src/box/iproto.cc
index 380a6ee..9d03bdf 100644
--- a/src/box/iproto.cc
+++ b/src/box/iproto.cc
@@ -1596,7 +1596,8 @@ tx_process_sql(struct cmsg *m)
 		goto error;
 	assert(msg->header.type == IPROTO_EXECUTE);
 	tx_inject_delay();
-	if (sql_prepare_and_execute(&msg->sql, &response, &fiber()->gc) != 0)
+	if (sql_prepare_and_execute(&msg->sql, &response, &fiber()->gc,
+	    ER_SQL_EXECUTE) != 0)
 		goto error;
 	/*
 	 * Take an obuf only after execute(). Else the buffer can
diff --git a/src/box/lua/schema.lua b/src/box/lua/schema.lua
index 8a804f0..02ec2fd 100644
--- a/src/box/lua/schema.lua
+++ b/src/box/lua/schema.lua
@@ -2456,3 +2456,26 @@ box.feedback.save = function(file_name)
 end
 
 box.NULL = msgpack.NULL
+
+box.sql.execute = function(sql)
+    local result = box.sql.new_execute(sql)
+    if result == nil then return end
+    local ret = nil
+    if result.rows ~= nil then
+        ret = {}
+        for key, row in pairs(result.rows) do
+            if type(row) == 'cdata' then
+                table.insert(ret, row:totable())
+            end
+        end
+    end
+    if result.metadata ~= nil then
+        if ret == nil then ret = {} end
+        ret[0] = {}
+        for key, row in pairs(result.metadata) do
+            table.insert(ret[0], row['name'])
+        end
+        setmetatable(ret, {__serialize = 'sequence'})
+    end
+    if ret ~= nil then return ret end
+end
diff --git a/src/box/lua/sql.c b/src/box/lua/sql.c
index 05556f9..c56dd4a 100644
--- a/src/box/lua/sql.c
+++ b/src/box/lua/sql.c
@@ -117,109 +117,6 @@ lua_vstream_init(struct vstream *vstream, struct lua_State *L)
 	vstream->L = L;
 }
 
-static void
-lua_push_column_names(struct lua_State *L, struct sqlite3_stmt *stmt)
-{
-	int column_count = sqlite3_column_count(stmt);
-	lua_createtable(L, column_count, 0);
-	for (int i = 0; i < column_count; i++) {
-		const char *name = sqlite3_column_name(stmt, i);
-		lua_pushstring(L, name == NULL ? "" : name);
-		lua_rawseti(L, -2, i+1);
-	}
-}
-
-static void
-lua_push_row(struct lua_State *L, struct sqlite3_stmt *stmt)
-{
-	int column_count = sqlite3_column_count(stmt);
-
-	lua_createtable(L, column_count, 0);
-	lua_rawgeti(L, LUA_REGISTRYINDEX, luaL_array_metatable_ref);
-	lua_setmetatable(L, -2);
-
-	for (int i = 0; i < column_count; i++) {
-		int type = sqlite3_column_type(stmt, i);
-		switch (type) {
-		case SQLITE_INTEGER:
-			luaL_pushint64(L, sqlite3_column_int64(stmt, i));
-			break;
-		case SQLITE_FLOAT:
-			lua_pushnumber(L, sqlite3_column_double(stmt, i));
-			break;
-		case SQLITE_TEXT: {
-			const void *text = sqlite3_column_text(stmt, i);
-			lua_pushlstring(L, text,
-					sqlite3_column_bytes(stmt, i));
-			break;
-		}
-		case SQLITE_BLOB: {
-			const void *blob = sqlite3_column_blob(stmt, i);
-			if (sql_column_subtype(stmt,i) == SQL_SUBTYPE_MSGPACK) {
-				luamp_decode(L, luaL_msgpack_default,
-					     (const char **)&blob);
-			} else {
-				lua_pushlstring(L, blob,
-					sqlite3_column_bytes(stmt, i));
-			}
-			break;
-		}
-		case SQLITE_NULL:
-			lua_rawgeti(L, LUA_REGISTRYINDEX, luaL_nil_ref);
-			break;
-		default:
-			assert(0);
-		}
-		lua_rawseti(L, -2, i+1);
-	}
-}
-
-static int
-lua_sql_execute(struct lua_State *L)
-{
-	sqlite3 *db = sql_get();
-	if (db == NULL)
-		return luaL_error(L, "not ready");
-
-	size_t length;
-	const char *sql = lua_tolstring(L, 1, &length);
-	if (sql == NULL)
-		return luaL_error(L, "usage: box.sql.execute(sqlstring)");
-
-	struct sqlite3_stmt *stmt;
-	if (sqlite3_prepare_v2(db, sql, length, &stmt, &sql) != SQLITE_OK)
-		goto sqlerror;
-	assert(stmt != NULL);
-
-	int rc;
-	int retval_count;
-	if (sqlite3_column_count(stmt) == 0) {
-		while ((rc = sqlite3_step(stmt)) == SQLITE_ROW);
-		retval_count = 0;
-	} else {
-		lua_newtable(L);
-		lua_pushvalue(L, lua_upvalueindex(1));
-		lua_setmetatable(L, -2);
-		lua_push_column_names(L, stmt);
-		lua_rawseti(L, -2, 0);
-
-		int row_count = 0;
-		while ((rc = sqlite3_step(stmt)) == SQLITE_ROW) {
-			lua_push_row(L, stmt);
-			lua_rawseti(L, -2, ++row_count);
-		}
-		retval_count = 1;
-	}
-        if (rc != SQLITE_OK && rc != SQLITE_DONE)
-		goto sqlerror;
-	sqlite3_finalize(stmt);
-	return retval_count;
-sqlerror:
-	lua_pushstring(L, sqlite3_errmsg(db));
-	sqlite3_finalize(stmt);
-	return lua_error(L);
-}
-
 static int
 lbox_execute(struct lua_State *L)
 {
@@ -237,10 +134,12 @@ lbox_execute(struct lua_State *L)
 	request.sql_text_len = length;
 	if (lua_gettop(L) == 2)
 		if (lua_sql_bind_list_decode(L, &request, 2, &fiber()->gc) != 0)
-			return luaT_error(L);
+			goto sqlerror;
+
 	struct sql_response response = {.is_flatten = true};
-	if (sql_prepare_and_execute(&request, &response, &fiber()->gc) != 0)
-		return luaT_error(L);
+	if (sql_prepare_and_execute(&request, &response, &fiber()->gc,
+				    ER_SYSTEM) != 0)
+		goto sqlerror;
 
 	int keys;
 	struct vstream vstream;
@@ -248,9 +147,12 @@ lbox_execute(struct lua_State *L)
 	lua_newtable(L);
 	if (sql_response_dump(&response, &keys, &vstream) != 0) {
 		lua_pop(L, 1);
-		return luaT_error(L);
+		goto sqlerror;
 	}
 	return 1;
+sqlerror:
+	lua_pushstring(L, sqlite3_errmsg(db));
+	return lua_error(L);
 }
 
 static int
@@ -266,7 +168,6 @@ void
 box_lua_sqlite_init(struct lua_State *L)
 {
 	static const struct luaL_Reg module_funcs [] = {
-		{"execute", lua_sql_execute},
 		{"new_execute", lbox_execute},
 		{"debug", lua_sql_debug},
 		{NULL, NULL}
-- 
2.7.4

[tarantool-patches] Re: [PATCH v2 2/7] box: add method dump_lua to port

[Vladislav Shpilevoy]

Hi! Thanks for the fixes! See my 5 comments below, fix
at the end of the email and on the branch.

Also, please, do not forget next time to answer to my comments
inlined. Do not just send v2 without any answers.

On 22/11/2018 22:10, imeevma@tarantool.org wrote:
> New method dump_lua dumps saved in port tuples to Lua stack. It
> will allow us to call this method without any other interaction
> with port.
> 
> Needed for #3505
> ---
>   src/box/lua/call.c |  1 +
>   src/box/port.c     | 22 ++++++++++++++++++++++
>   src/box/port.h     | 12 ++++++++++++
>   3 files changed, 35 insertions(+)
> 
> diff --git a/src/box/port.c b/src/box/port.c
> index 266cf3d..a65a32d 100644
> --- a/src/box/port.c
> +++ b/src/box/port.c
> @@ -36,6 +36,8 @@
>   #include <small/mempool.h>
>   #include <fiber.h>
>   #include "errinj.h"
> +#include "lua/utils.h"
> +#include "lua/tuple.h"

1. src/box/ files should not depend in Lua. I moved
implementation to misc.cc and put here extern declaration.

>   
>   static struct mempool port_tuple_entry_pool;
>   
> @@ -121,6 +123,19 @@ port_tuple_dump_msgpack(struct port *base, struct obuf *out)
>   	return 1;
>   }
>   
> +static int

2. Why do you need to return number of dumped tuples? Dump_msgpack
returns number of tuples since in iproto.cc it firstly reserves
msgpack array header, then dumps port and then fills the reserved
header. For Lua we do not need it.

> +port_tuple_dump_lua(struct port *base, struct lua_State *L)
> +{
> +	struct port_tuple *port = port_tuple(base);
> +	struct port_tuple_entry *pe;
> +	int i = 0;

3. Why did not you add here lua_createtable(L, port->size, 0);
as it is done in misc.cc in the original implementation?

4. Why did not you replaced lbox_port_to_table with this new
method?

I did this points and it works.

> +	for (pe = port->first; pe != NULL; pe = pe->next) {
> +		luaT_pushtuple(L, pe->tuple);
> +		lua_rawseti(L, -2, ++i);
> +	}
> +	return port->size;
> +}
> +
>   void
>   port_destroy(struct port *port)
>   {
> diff --git a/src/box/port.h b/src/box/port.h
> index 882bb37..3bd83b0 100644
> --- a/src/box/port.h
> +++ b/src/box/port.h
> @@ -185,6 +190,13 @@ int
>   port_dump_msgpack_16(struct port *port, struct obuf *out);
>   
>   /**
> + * Same as port_dump(), but use the legacy Tarantool 1.6
> + * format.

5. This comment does not make sense to this function.

> + */
> +int
> +port_dump_lua(struct port *port, struct lua_State *L);
> +
> +/**
>    * Dump a port content as a plain text into a buffer,
>    * allocated inside.
>    * @param port Port with data to dump.
> 

===========================================================

diff --git a/src/box/lua/misc.cc b/src/box/lua/misc.cc
index 8bd33aed1..6af811f7d 100644
--- a/src/box/lua/misc.cc
+++ b/src/box/lua/misc.cc
@@ -56,23 +56,26 @@ lbox_encode_tuple_on_gc(lua_State *L, int idx, size_t *p_len)
  	return (char *) region_join_xc(gc, *p_len);
  }
  
-/* }}} */
-
-/** {{{ Lua/C implementation of index:select(): used only by Vinyl **/
-
-static inline void
-lbox_port_to_table(lua_State *L, struct port *port_base)
+/**
+ * Dump port_tuple content to Lua as a table. Used in box/port.c,
+ * but implemented here to eliminate port.c dependency on Lua.
+ */
+extern "C" void
+port_tuple_dump_lua(struct port *base, struct lua_State *L)
  {
-	struct port_tuple *port = port_tuple(port_base);
+	struct port_tuple *port = port_tuple(base);
  	lua_createtable(L, port->size, 0);
-	struct port_tuple_entry *entry = port->first;
-	for (int i = 0 ; i < port->size; i++) {
-		luaT_pushtuple(L, entry->tuple);
-		lua_rawseti(L, -2, i + 1);
-		entry = entry->next;
+	struct port_tuple_entry *pe = port->first;
+	for (int i = 0; pe != NULL; pe = pe->next) {
+		luaT_pushtuple(L, pe->tuple);
+		lua_rawseti(L, -2, ++i);
  	}
  }
  
+/* }}} */
+
+/** {{{ Lua/C implementation of index:select(): used only by Vinyl **/
+
  static int
  lbox_select(lua_State *L)
  {
@@ -106,7 +109,7 @@ lbox_select(lua_State *L)
  	 * table always crashed the first (can't be fixed with pcall).
  	 * https://github.com/tarantool/tarantool/issues/1182
  	 */
-	lbox_port_to_table(L, &port);
+	port_dump_lua(&port, L);
  	port_destroy(&port);
  	return 1; /* lua table with tuples */
  }
diff --git a/src/box/port.c b/src/box/port.c
index a65a32d96..df94eb507 100644
--- a/src/box/port.c
+++ b/src/box/port.c
@@ -123,18 +123,8 @@ port_tuple_dump_msgpack(struct port *base, struct obuf *out)
  	return 1;
  }
  
-static int
-port_tuple_dump_lua(struct port *base, struct lua_State *L)
-{
-	struct port_tuple *port = port_tuple(base);
-	struct port_tuple_entry *pe;
-	int i = 0;
-	for (pe = port->first; pe != NULL; pe = pe->next) {
-		luaT_pushtuple(L, pe->tuple);
-		lua_rawseti(L, -2, ++i);
-	}
-	return port->size;
-}
+extern void
+port_tuple_dump_lua(struct port *base, struct lua_State *L);
  
  void
  port_destroy(struct port *port)
@@ -154,10 +144,10 @@ port_dump_msgpack_16(struct port *port, struct obuf *out)
  	return port->vtab->dump_msgpack_16(port, out);
  }
  
-int
+void
  port_dump_lua(struct port *port, struct lua_State *L)
  {
-	return port->vtab->dump_lua(port, L);
+	port->vtab->dump_lua(port, L);
  }
  
  const char *
diff --git a/src/box/port.h b/src/box/port.h
index 3bd83b092..751e44efe 100644
--- a/src/box/port.h
+++ b/src/box/port.h
@@ -77,11 +77,8 @@ struct port_vtab {
  	 * 1.6 format.
  	 */
  	int (*dump_msgpack_16)(struct port *port, struct obuf *out);
-	/**
-	 * Dump the content of a port to Lua stack.
-	 * On success returns number of entries dumped.
-	 */
-	int (*dump_lua)(struct port *port, struct lua_State *L);
+	/** Dump the content of a port to Lua stack. */
+	void (*dump_lua)(struct port *port, struct lua_State *L);
  	/**
  	 * Dump a port content as a plain text into a buffer,
  	 * allocated inside.
@@ -189,11 +186,8 @@ port_dump_msgpack(struct port *port, struct obuf *out);
  int
  port_dump_msgpack_16(struct port *port, struct obuf *out);
  
-/**
- * Same as port_dump(), but use the legacy Tarantool 1.6
- * format.
- */
-int
+/** Dump port content to Lua stack. */
+void
  port_dump_lua(struct port *port, struct lua_State *L);
  
  /**

[tarantool-patches] Re: [PATCH v2 6/7] lua: create vstream implementation for Lua

[Vladislav Shpilevoy]

Thanks for the fixes! See my 5 comments below, fix
at the end of the email and on the branch.

Also note, that I did not run tests. So before squashing
please, check the tests.

> diff --git a/src/box/execute.h b/src/box/execute.h
> index 5a11a8a..8ee0a89 100644
> --- a/src/box/execute.h
> +++ b/src/box/execute.h
> @@ -131,6 +131,21 @@ xrow_decode_sql(const struct xrow_header *row, struct sql_request *request,
>   		struct region *region);
>   
>   /**
> + * Parse Lua table of SQL parameters and store a result
> + * into the @request->bind, bind_count.
> + * @param L Lua stack to get data from.
> + * @param request Request to save decoded parameters.
> + * @param idx Position of table with parameters on Lua stack.
> + * @param region Allocator.
> + *
> + * @retval  0 Success.
> + * @retval -1 Client or memory error.
> + */
> +int
> +lua_sql_bind_list_decode(struct lua_State *L, struct sql_request *request,
> +			 int idx, struct region *region);
1. You do not need to pass region here. Get fiber()->gc inside this
function. In original implementation it is passed because back in
those days we hoped to remove fiber()->gc, but now we decided not to
do it.

Also, it leaks now. You allocate sql_bind parameters on region, but
never truncate it. Iproto has the same bug, but you should not
duplicate it.

But you can not truncate it as well, since sql_prepare_and_execute
could save some transactional data onto it. And this bug iproto does
not have since it uses region of iproto thread.

I think, here you should use malloc.

> 		lua_pushnil(L);
> 		lua_next(L, lua_gettop(L) - 1);
> 		struct luaL_field field_name;
> 		lua_pushvalue(L, -2);
> 		luaL_tofield(L, luaL_msgpack_default, -1, &field_name);
> 		lua_pop(L, 1);
> 		assert(field_name.type == MP_STR);
> 		luaL_tofield(L, luaL_msgpack_default, -1, field);
> 		lua_pop(L, 1);
> 		bind->name = field_name.sval.data;
> 		bind->name_len = field_name.sval.len;

2. Please, do not use luaL_tofield for each scalar value. Here
it is much simpler to just use lua_tostring or something.

> 	for (uint32_t i = 0; i < bind_count; ++i) {
> 		struct luaL_field field;
> 		lua_rawgeti(L, idx, i + 1);
> 		luaL_tofield(L, luaL_msgpack_default, -1, &field);
> 		if (lua_sql_bind_decode(L, &bind[i], i, &field) != 0) {
> 			region_truncate(region, used);
> 			return -1;
> 		}
> 		lua_pop(L, 1);
> 	}

3. Why not to do lua_rawgeti and luaL_tofield inside lua_sql_bind_decode?

4. Lua vstream should not be implemented in sql.c. It is a separate file
luastream.c or something. You should have separate luastream implementation
without any vtabs and lua_vstream_vtab specially for vstream. Just like
it is done now for mpstream.

5. API of creating mp_vstream and lua_vstream is asymmetrical: to create
mp_vstream you call two functions to initialize mpstream and then vtab,
but to create lua_vstream you call one function. I think, it should be
fixed automatically once you've fixed the previous comment.

======================================================================

commit f14e975bfe49f982ac0efd61e3d67d1ced90c5a2
Author: Vladislav Shpilevoy <v.shpilevoy@tarantool.org>
Date:   Fri Nov 23 00:44:56 2018 +0300

     Review fixes

diff --git a/src/box/execute.c b/src/box/execute.c
index ca957a870..0a3ca15e0 100644
--- a/src/box/execute.c
+++ b/src/box/execute.c
@@ -312,8 +312,8 @@ lua_sql_bind_decode(struct lua_State *L, struct sql_bind *bind, int i,
  		 * Report parse error otherwise.
  		 */
  		if (field->size != 1) {
-			diag_set(ClientError, ER_INVALID_MSGPACK,
-				 "SQL bind parameter");
+			diag_set(ClientError, ER_ILLEGAL_PARAMS, "SQL bind "\
+				 "parameter should be {'name': value}");
  			return -1;
  		}
  		lua_pushnil(L);
@@ -393,7 +393,7 @@ lua_sql_bind_decode(struct lua_State *L, struct sql_bind *bind, int i,
  
  int
  lua_sql_bind_list_decode(struct lua_State *L, struct sql_request *request,
-			 int idx, struct region *region)
+			 int idx)
  {
  	assert(request != NULL);
  	if (! lua_istable(L, idx)) {
@@ -408,11 +408,12 @@ lua_sql_bind_list_decode(struct lua_State *L, struct sql_request *request,
  			 (int) bind_count);
  		return -1;
  	}
+	struct region *region = &fiber()->gc;
  	uint32_t used = region_used(region);
  	size_t size = sizeof(struct sql_bind) * bind_count;
  	struct sql_bind *bind = (struct sql_bind *) region_alloc(region, size);
  	if (bind == NULL) {
-		diag_set(OutOfMemory, size, "region_alloc", "struct sql_bind");
+		diag_set(OutOfMemory, size, "region_alloc", "bind");
  		return -1;
  	}
  	for (uint32_t i = 0; i < bind_count; ++i) {
diff --git a/src/box/execute.h b/src/box/execute.h
index afd0011da..ff468420b 100644
--- a/src/box/execute.h
+++ b/src/box/execute.h
@@ -148,14 +148,13 @@ xrow_decode_sql(const struct xrow_header *row, struct sql_request *request,
   * @param L Lua stack to get data from.
   * @param request Request to save decoded parameters.
   * @param idx Position of table with parameters on Lua stack.
- * @param region Allocator.
   *
   * @retval  0 Success.
   * @retval -1 Client or memory error.
   */
  int
  lua_sql_bind_list_decode(struct lua_State *L, struct sql_request *request,
-			 int idx, struct region *region);
+			 int idx);
  
  /**
   * Prepare and execute an SQL statement.
diff --git a/src/box/lua/sql.c b/src/box/lua/sql.c
index 05556f98f..9245f85c9 100644
--- a/src/box/lua/sql.c
+++ b/src/box/lua/sql.c
@@ -67,8 +67,7 @@ lua_vstream_encode_bool(struct vstream *stream, bool val)
  int
  lua_vstream_encode_port(struct vstream *stream, struct port *port)
  {
-	if (port_dump_lua(port, stream->L) < 0)
-		return -1;
+	port_dump_lua(port, stream->L);
  	return 0;
  }
  
@@ -235,10 +234,9 @@ lbox_execute(struct lua_State *L)
  	struct sql_request request = {};
  	request.sql_text = sql;
  	request.sql_text_len = length;
-	if (lua_gettop(L) == 2)
-		if (lua_sql_bind_list_decode(L, &request, 2, &fiber()->gc) != 0)
-			return luaT_error(L);
-	struct sql_response response = {.is_flatten = true};
+	if (lua_gettop(L) == 2 && lua_sql_bind_list_decode(L, &request, 2) != 0)
+		return luaT_error(L);
+	struct sql_response response = {.is_info_flattened = true};
  	if (sql_prepare_and_execute(&request, &response, &fiber()->gc) != 0)
  		return luaT_error(L);
  

[tarantool-patches] Re: [PATCH v2 5/7] sql: create interface vstream

[Vladislav Shpilevoy]

Thanks for the fixes! See my 4 comments below, fix
at the end of the email and on the branch.

On 22/11/2018 22:11, imeevma@tarantool.org wrote:
> If we want to use functions from execute.h not only in IPROTO we
> should create special interface. This interface will allow us to
> create different implementations for mpstream and lua_State and
> use functions from execute.c without changing them. This patch
> creates such interface and its implementation for mpstream and
> replaces mpstream functions in execute.c by methods of this
> interface.
> 
> Needed for #3505
> ---
>   src/box/execute.c |  69 ++++++++++++-----------
>   src/box/execute.h |   6 +-
>   src/box/iproto.cc |  11 ++--
>   src/box/vstream.h | 164 ++++++++++++++++++++++++++++++++++++++++++++++++++++++
>   src/mpstream.c    |  78 ++++++++++++++++++++++++++
>   5 files changed, 291 insertions(+), 37 deletions(-)
>   create mode 100644 src/box/vstream.h
> 
> diff --git a/src/box/execute.h b/src/box/execute.h
> index 940f3a3..5a11a8a 100644
> --- a/src/box/execute.h
> +++ b/src/box/execute.h
> @@ -75,6 +75,8 @@ struct sql_response {
>   	struct port port;
>   	/** Prepared SQL statement with metadata. */
>   	void *prep_stmt;
> +	/** Result should be flatten if true. */
> +	bool is_flatten;

1. What does it mean 'result should be flatten'? Are all
tuples merged into a single flattened one? Or are all metafields
merged into a single array? Please, be more specific. It is far
from obvious now what is it 'flattened result'.

Also, I guess, 'flatten' is a verb, so you can not say 'is flatten'.
Only 'is flattened'.

2. I do not see where do you initialize this field for
iproto. So it is now initialized with stack garbage.

(I've fixed all these things since we hurry.)

>   };
>   
>   /**
> diff --git a/src/box/vstream.h b/src/box/vstream.h
> new file mode 100644
> index 0000000..a8dcfc2
> --- /dev/null
> +++ b/src/box/vstream.h
> @@ -0,0 +1,164 @@
> +#ifndef TARANTOOL_VSTREAM_H_INCLUDED
> +#define TARANTOOL_VSTREAM_H_INCLUDED
> +/*
> + * Copyright 2010-2018, Tarantool AUTHORS, please see AUTHORS file.
> + *
> + * Redistribution and use in source and binary forms, with or
> + * without modification, are permitted provided that the following
> + * conditions are met:
> + *
> + * 1. Redistributions of source code must retain the above
> + *    copyright notice, this list of conditions and the
> + *    following disclaimer.
> + *
> + * 2. Redistributions in binary form must reproduce the above
> + *    copyright notice, this list of conditions and the following
> + *    disclaimer in the documentation and/or other materials
> + *    provided with the distribution.
> + *
> + * THIS SOFTWARE IS PROVIDED BY AUTHORS ``AS IS'' AND
> + * ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED
> + * TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
> + * A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL
> + * AUTHORS OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT,
> + * INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL
> + * DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF
> + * SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR
> + * BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF
> + * LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
> + * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF
> + * THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF
> + * SUCH DAMAGE.
> + */
> +
> +#include "diag.h"
> +#include "mpstream.h"

3. vstream should not depend on mpstream, but vice versa.
Please, look at how struct port is done. It uses padding
into which its descendants can lay anything.

> +
> +#if defined(__cplusplus)
> +extern "C" {
> +#endif /* defined(__cplusplus) */
> +
> +struct vstream;
> +struct lua_State;
> +struct port;
> +
> diff --git a/src/mpstream.c b/src/mpstream.c
> index e4f7950..f9943e4 100644
> --- a/src/mpstream.c
> +++ b/src/mpstream.c
> @@ -175,3 +180,76 @@ mpstream_encode_bool(struct mpstream *stream, bool val)
>       char *pos = mp_encode_bool(data, val);
>       mpstream_advance(stream, pos - data);
>   }
> +
> +typedef void (*encode_array_f)(struct vstream *stream, uint32_t size);
> +typedef void (*encode_map_f)(struct vstream *stream, uint32_t size);
> +typedef void (*encode_uint_f)(struct vstream *stream, uint64_t num);
> +typedef void (*encode_int_f)(struct vstream *stream, int64_t num);
> +typedef void (*encode_float_f)(struct vstream *stream, float num);
> +typedef void (*encode_double_f)(struct vstream *stream, double num);
> +typedef void (*encode_strn_f)(struct vstream *stream, const char *str,
> +			      uint32_t len);
> +typedef void (*encode_nil_f)(struct vstream *stream);
> +typedef void (*encode_bool_f)(struct vstream *stream, bool val);
> +

4. I think, it should be part of vstream.h. And struct vstream members should be
defined with these types.

====================================================================

commit a75e03dd1edf9c106e78f3d10618b4bfa80b84d5
Author: Vladislav Shpilevoy <v.shpilevoy@tarantool.org>
Date:   Fri Nov 23 00:01:56 2018 +0300

     Review fixes

diff --git a/src/box/execute.c b/src/box/execute.c
index 8093d9c99..18133b1e7 100644
--- a/src/box/execute.c
+++ b/src/box/execute.c
@@ -648,7 +648,7 @@ err:
  			stailq_foreach_entry(id_entry, autoinc_id_list, link)
  				id_count++;
  		}
-		if (!response->is_flatten) {
+		if (!response->is_info_flattened) {
  			vstream_encode_enum(stream, IPROTO_SQL_INFO, "info");
  			vstream_encode_map(stream, map_size);
  		}
@@ -671,7 +671,7 @@ err:
  			}
  			vstream_encode_map_commit(stream);
  		}
-		if (!response->is_flatten)
+		if (!response->is_info_flattened)
  			vstream_encode_map_commit(stream);
  	}
  finish:
diff --git a/src/box/execute.h b/src/box/execute.h
index 5a11a8a2b..42bbe63de 100644
--- a/src/box/execute.h
+++ b/src/box/execute.h
@@ -75,8 +75,20 @@ struct sql_response {
  	struct port port;
  	/** Prepared SQL statement with metadata. */
  	void *prep_stmt;
-	/** Result should be flatten if true. */
-	bool is_flatten;
+	/**
+	 * SQL response can be dumped into msgpack to be sent via
+	 * iproto or onto Lua stack to be returned into an
+	 * application. In the first case response body has
+	 * explicit field IPROTO_SQL_INFO: {rowcount = ...,
+	 * autoids = ...}. But in case of Lua this field is
+	 * flattened. A result never has 'info' field, it has
+	 * inlined 'rowcount' and 'autoids'. In iproto
+	 * IPROTO_SQL_INFO field is sent mostly to explicitly
+	 * distinguish two response types: DML/DDL vs DQL,
+	 * IPROTO_SQL_INFO vs IPROTO_METADATA. So this flag is set
+	 * by Lua and allows to flatten SQL_INFO fields.
+	 */
+	bool is_info_flattened;
  };
  
  /**
diff --git a/src/box/iproto.cc b/src/box/iproto.cc
index 380a6eec0..1c4c65176 100644
--- a/src/box/iproto.cc
+++ b/src/box/iproto.cc
@@ -1588,6 +1588,7 @@ tx_process_sql(struct cmsg *m)
  	struct iproto_msg *msg = tx_accept_msg(m);
  	struct obuf *out;
  	struct sql_response response;
+	memset(&response, 0, sizeof(response));
  	bool is_error = false;
  
  	tx_fiber_init(msg->connection->session, msg->header.sync);
diff --git a/src/box/vstream.h b/src/box/vstream.h
index a8dcfc289..9b52acd3d 100644
--- a/src/box/vstream.h
+++ b/src/box/vstream.h
@@ -30,10 +30,6 @@
   * THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF
   * SUCH DAMAGE.
   */
-
-#include "diag.h"
-#include "mpstream.h"
-
  #if defined(__cplusplus)
  extern "C" {
  #endif /* defined(__cplusplus) */
@@ -42,28 +38,43 @@ struct vstream;
  struct lua_State;
  struct port;
  
+typedef void (*encode_array_f)(struct vstream *stream, uint32_t size);
+typedef void (*encode_map_f)(struct vstream *stream, uint32_t size);
+typedef void (*encode_uint_f)(struct vstream *stream, uint64_t num);
+typedef void (*encode_int_f)(struct vstream *stream, int64_t num);
+typedef void (*encode_float_f)(struct vstream *stream, float num);
+typedef void (*encode_double_f)(struct vstream *stream, double num);
+typedef void (*encode_strn_f)(struct vstream *stream, const char *str,
+			      uint32_t len);
+typedef void (*encode_nil_f)(struct vstream *stream);
+typedef void (*encode_bool_f)(struct vstream *stream, bool val);
+typedef void (*encode_enum_f)(struct vstream *stream, int64_t num,
+			      const char *str);
+typedef int (*encode_port_f)(struct vstream *stream, struct port *port);
+typedef void (*encode_array_commit_f)(struct vstream *stream, uint32_t id);
+typedef void (*encode_map_commit_f)(struct vstream *stream);
+
  struct vstream_vtab {
-	void (*encode_array)(struct vstream *stream, uint32_t size);
-	void (*encode_map)(struct vstream *stream, uint32_t size);
-	void (*encode_uint)(struct vstream *stream, uint64_t num);
-	void (*encode_int)(struct vstream *stream, int64_t num);
-	void (*encode_float)(struct vstream *stream, float num);
-	void (*encode_double)(struct vstream *stream, double num);
-	void (*encode_strn)(struct vstream *stream, const char *str,
-			    uint32_t len);
-	void (*encode_nil)(struct vstream *stream);
-	void (*encode_bool)(struct vstream *stream, bool val);
-	void (*encode_enum)(struct vstream *stream, int64_t num,
-			    const char *str);
-	int (*encode_port)(struct vstream *stream, struct port *port);
-	void (*encode_array_commit)(struct vstream *stream, uint32_t id);
-	void (*encode_map_commit)(struct vstream *stream);
+	encode_array_f encode_array;
+	encode_map_f encode_map;
+	encode_uint_f encode_uint;
+	encode_int_f encode_int;
+	encode_float_f encode_float;
+	encode_double_f encode_double;
+	encode_strn_f encode_strn;
+	encode_nil_f encode_nil;
+	encode_bool_f encode_bool;
+	encode_enum_f encode_enum;
+	encode_port_f encode_port;
+	encode_array_commit_f encode_array_commit;
+	encode_map_commit_f encode_map_commit;
  };
  
  struct vstream {
  	/** TODO: Write comment. */
  	union {
-		struct mpstream mpstream;
+		/** Here struct mpstream lives under the hood. */
+		char inheritance_padding[64];
  		struct lua_State *L;
  	};
  	/** Virtual function table. */
diff --git a/src/mpstream.c b/src/mpstream.c
index f9943e493..23b20892c 100644
--- a/src/mpstream.c
+++ b/src/mpstream.c
@@ -33,11 +33,8 @@
  #include <assert.h>
  #include <stdint.h>
  #include "msgpuck.h"
-
  #include "box/vstream.h"
-#include "box/iproto_constants.h"
  #include "box/port.h"
-#include "box/xrow.h"
  
  void
  mpstream_reserve_slow(struct mpstream *stream, size_t size)
@@ -181,17 +178,6 @@ mpstream_encode_bool(struct mpstream *stream, bool val)
      mpstream_advance(stream, pos - data);
  }
  
-typedef void (*encode_array_f)(struct vstream *stream, uint32_t size);
-typedef void (*encode_map_f)(struct vstream *stream, uint32_t size);
-typedef void (*encode_uint_f)(struct vstream *stream, uint64_t num);
-typedef void (*encode_int_f)(struct vstream *stream, int64_t num);
-typedef void (*encode_float_f)(struct vstream *stream, float num);
-typedef void (*encode_double_f)(struct vstream *stream, double num);
-typedef void (*encode_strn_f)(struct vstream *stream, const char *str,
-			      uint32_t len);
-typedef void (*encode_nil_f)(struct vstream *stream);
-typedef void (*encode_bool_f)(struct vstream *stream, bool val);
-
  int
  mp_vstream_encode_port(struct vstream *stream, struct port *port)
  {
@@ -220,16 +206,9 @@ mp_vstream_encode_enum(struct vstream *stream, int64_t num, const char *str)
  }
  
  void
-mp_vstream_encode_map_commit(struct vstream *stream)
-{
-	(void)stream;
-}
-
-void
-mp_vstream_encode_array_commit(struct vstream *stream, uint32_t id)
+mp_vstream_noop(struct vstream *stream, ...)
  {
-	(void)stream;
-	(void)id;
+    (void) stream;
  }
  
  const struct vstream_vtab mp_vstream_vtab = {
@@ -244,8 +223,8 @@ const struct vstream_vtab mp_vstream_vtab = {
  	/** encode_bool = */ (encode_bool_f)mpstream_encode_bool,
  	/** encode_enum = */ mp_vstream_encode_enum,
  	/** encode_port = */ mp_vstream_encode_port,
-	/** encode_array_commit = */ mp_vstream_encode_array_commit,
-	/** encode_map_commit = */ mp_vstream_encode_map_commit,
+	/** encode_array_commit = */ (encode_array_commit_f)mp_vstream_noop,
+	/** encode_map_commit = */ (encode_map_commit_f)mp_vstream_noop,
  };
  
  void

[tarantool-patches] Re: [PATCH v2 2/7] box: add method dump_lua to port

[Imeev Mergen]

[-- Attachment #1: Type: text/plain, Size: 7429 bytes --]

Hi! Thank you for review and fixes! I squashed your fixes and
removed libs from port.c that were added in previous version.
New version below.

On 11/23/18 12:49 AM, Vladislav Shpilevoy wrote:
> Hi! Thanks for the fixes! See my 5 comments below, fix
> at the end of the email and on the branch.
>
> Also, please, do not forget next time to answer to my comments
> inlined. Do not just send v2 without any answers.
>
> On 22/11/2018 22:10, imeevma@tarantool.org wrote:
>> New method dump_lua dumps saved in port tuples to Lua stack. It
>> will allow us to call this method without any other interaction
>> with port.
>>
>> Needed for #3505
>> ---
>>   src/box/lua/call.c |  1 +
>>   src/box/port.c     | 22 ++++++++++++++++++++++
>>   src/box/port.h     | 12 ++++++++++++
>>   3 files changed, 35 insertions(+)
>>
>> diff --git a/src/box/port.c b/src/box/port.c
>> index 266cf3d..a65a32d 100644
>> --- a/src/box/port.c
>> +++ b/src/box/port.c
>> @@ -36,6 +36,8 @@
>>   #include <small/mempool.h>
>>   #include <fiber.h>
>>   #include "errinj.h"
>> +#include "lua/utils.h"
>> +#include "lua/tuple.h"
>
> 1. src/box/ files should not depend in Lua. I moved
> implementation to misc.cc and put here extern declaration.
Squashed.
>
>>     static struct mempool port_tuple_entry_pool;
>>   @@ -121,6 +123,19 @@ port_tuple_dump_msgpack(struct port *base, 
>> struct obuf *out)
>>       return 1;
>>   }
>>   +static int
>
> 2. Why do you need to return number of dumped tuples? Dump_msgpack
> returns number of tuples since in iproto.cc it firstly reserves
> msgpack array header, then dumps port and then fills the reserved
> header. For Lua we do not need it.
Squashed.
>
>> +port_tuple_dump_lua(struct port *base, struct lua_State *L)
>> +{
>> +    struct port_tuple *port = port_tuple(base);
>> +    struct port_tuple_entry *pe;
>> +    int i = 0;
>
> 3. Why did not you add here lua_createtable(L, port->size, 0);
> as it is done in misc.cc in the original implementation?
Squashed.
>
> 4. Why did not you replaced lbox_port_to_table with this new
> method?
>
> I did this points and it works.
Squashed.
>
>> +    for (pe = port->first; pe != NULL; pe = pe->next) {
>> +        luaT_pushtuple(L, pe->tuple);
>> +        lua_rawseti(L, -2, ++i);
>> +    }
>> +    return port->size;
>> +}
>> +
>>   void
>>   port_destroy(struct port *port)
>>   {
>> diff --git a/src/box/port.h b/src/box/port.h
>> index 882bb37..3bd83b0 100644
>> --- a/src/box/port.h
>> +++ b/src/box/port.h
>> @@ -185,6 +190,13 @@ int
>>   port_dump_msgpack_16(struct port *port, struct obuf *out);
>>     /**
>> + * Same as port_dump(), but use the legacy Tarantool 1.6
>> + * format.
>
> 5. This comment does not make sense to this function.
Squashed.
>
>> + */
>> +int
>> +port_dump_lua(struct port *port, struct lua_State *L);
>> +
>> +/**
>>    * Dump a port content as a plain text into a buffer,
>>    * allocated inside.
>>    * @param port Port with data to dump.
>>
>
*New version:*

commit ff2fc3fd58dd99d3a98ad790c8e82363949cb3db
Author: Mergen Imeev <imeevma@gmail.com>
Date:   Sat Nov 17 15:37:17 2018 +0300

     box: add method dump_lua to port

     New method dump_lua dumps saved in port tuples to Lua stack. It
     will allow us to call this method without any other interaction
     with port.

     Needed for #3505

diff --git a/src/box/lua/call.c b/src/box/lua/call.c
index 1f20426..52939ae 100644
--- a/src/box/lua/call.c
+++ b/src/box/lua/call.c
@@ -424,6 +424,7 @@ port_lua_dump_plain(struct port *port, uint32_t *size);
  static const struct port_vtab port_lua_vtab = {
      .dump_msgpack = port_lua_dump,
      .dump_msgpack_16 = port_lua_dump_16,
+    .dump_lua = NULL,
      .dump_plain = port_lua_dump_plain,
      .destroy = port_lua_destroy,
  };
diff --git a/src/box/lua/misc.cc b/src/box/lua/misc.cc
index 8bd33ae..8de7401 100644
--- a/src/box/lua/misc.cc
+++ b/src/box/lua/misc.cc
@@ -56,23 +56,26 @@ lbox_encode_tuple_on_gc(lua_State *L, int idx, 
size_t *p_len)
      return (char *) region_join_xc(gc, *p_len);
  }

-/* }}} */
-
-/** {{{ Lua/C implementation of index:select(): used only by Vinyl **/
-
-static inline void
-lbox_port_to_table(lua_State *L, struct port *port_base)
+/**
+ * Dump port_tuple content to Lua as a table. Used in box/port.c,
+ * but implemented here to eliminate port.c dependency on Lua.
+ */
+extern "C" void
+port_tuple_dump_lua(struct port *base, struct lua_State *L)
  {
-    struct port_tuple *port = port_tuple(port_base);
+    struct port_tuple *port = port_tuple(base);
      lua_createtable(L, port->size, 0);
-    struct port_tuple_entry *entry = port->first;
-    for (int i = 0 ; i < port->size; i++) {
-        luaT_pushtuple(L, entry->tuple);
-        lua_rawseti(L, -2, i + 1);
-        entry = entry->next;
+    struct port_tuple_entry *pe = port->first;
+    for (int i = 0; pe != NULL; pe = pe->next) {
+        luaT_pushtuple(L, pe->tuple);
+        lua_rawseti(L, -2, ++i);
      }
  }

+/* }}} */
+
+/** {{{ Lua/C implementation of index:select(): used only by Vinyl **/
+
  static int
  lbox_select(lua_State *L)
  {
@@ -106,7 +109,7 @@ lbox_select(lua_State *L)
       * table always crashed the first (can't be fixed with pcall).
       * https://github.com/tarantool/tarantool/issues/1182
       */
-    lbox_port_to_table(L, &port);
+    port_dump_lua(&port, L);
      port_destroy(&port);
      return 1; /* lua table with tuples */
  }
diff --git a/src/box/port.c b/src/box/port.c
index 266cf3d..853d24c 100644
--- a/src/box/port.c
+++ b/src/box/port.c
@@ -121,6 +121,9 @@ port_tuple_dump_msgpack(struct port *base, struct 
obuf *out)
      return 1;
  }

+extern void
+port_tuple_dump_lua(struct port *base, struct lua_State *L);
+
  void
  port_destroy(struct port *port)
  {
@@ -139,6 +142,12 @@ port_dump_msgpack_16(struct port *port, struct obuf 
*out)
      return port->vtab->dump_msgpack_16(port, out);
  }

+void
+port_dump_lua(struct port *port, struct lua_State *L)
+{
+    port->vtab->dump_lua(port, L);
+}
+
  const char *
  port_dump_plain(struct port *port, uint32_t *size)
  {
@@ -161,6 +170,7 @@ port_free(void)
  const struct port_vtab port_tuple_vtab = {
      .dump_msgpack = port_tuple_dump_msgpack,
      .dump_msgpack_16 = port_tuple_dump_msgpack_16,
+    .dump_lua = port_tuple_dump_lua,
      .dump_plain = NULL,
      .destroy = port_tuple_destroy,
  };
diff --git a/src/box/port.h b/src/box/port.h
index 882bb37..751e44e 100644
--- a/src/box/port.h
+++ b/src/box/port.h
@@ -77,6 +77,8 @@ struct port_vtab {
       * 1.6 format.
       */
      int (*dump_msgpack_16)(struct port *port, struct obuf *out);
+    /** Dump the content of a port to Lua stack. */
+    void (*dump_lua)(struct port *port, struct lua_State *L);
      /**
       * Dump a port content as a plain text into a buffer,
       * allocated inside.
@@ -184,6 +186,10 @@ port_dump_msgpack(struct port *port, struct obuf *out);
  int
  port_dump_msgpack_16(struct port *port, struct obuf *out);

+/** Dump port content to Lua stack. */
+void
+port_dump_lua(struct port *port, struct lua_State *L);
+
  /**
   * Dump a port content as a plain text into a buffer,
   * allocated inside.


[-- Attachment #2: Type: text/html, Size: 11255 bytes --]

[tarantool-patches] Re: [PATCH v2 5/7] sql: create interface vstream

[Imeev Mergen]

[-- Attachment #1: Type: text/plain, Size: 23004 bytes --]

Hi! Thank you for review and fixes! I squashed you fixes and done
some changes:
- vstream_encode_port() now uses port_dump_msgpack() instead of
   port_dump_msgpack_16()
- vstream now contains only inheritance_padding and vtab.

New patch and some answers below.

On 11/23/18 12:49 AM, Vladislav Shpilevoy wrote:
> Thanks for the fixes! See my 4 comments below, fix
> at the end of the email and on the branch.
>
> On 22/11/2018 22:11, imeevma@tarantool.org wrote:
>> If we want to use functions from execute.h not only in IPROTO we
>> should create special interface. This interface will allow us to
>> create different implementations for mpstream and lua_State and
>> use functions from execute.c without changing them. This patch
>> creates such interface and its implementation for mpstream and
>> replaces mpstream functions in execute.c by methods of this
>> interface.
>>
>> Needed for #3505
>> ---
>>   src/box/execute.c |  69 ++++++++++++-----------
>>   src/box/execute.h |   6 +-
>>   src/box/iproto.cc |  11 ++--
>>   src/box/vstream.h | 164 
>> ++++++++++++++++++++++++++++++++++++++++++++++++++++++
>>   src/mpstream.c    |  78 ++++++++++++++++++++++++++
>>   5 files changed, 291 insertions(+), 37 deletions(-)
>>   create mode 100644 src/box/vstream.h
>>
>> diff --git a/src/box/execute.h b/src/box/execute.h
>> index 940f3a3..5a11a8a 100644
>> --- a/src/box/execute.h
>> +++ b/src/box/execute.h
>> @@ -75,6 +75,8 @@ struct sql_response {
>>       struct port port;
>>       /** Prepared SQL statement with metadata. */
>>       void *prep_stmt;
>> +    /** Result should be flatten if true. */
>> +    bool is_flatten;
>
> 1. What does it mean 'result should be flatten'? Are all
> tuples merged into a single flattened one? Or are all metafields
> merged into a single array? Please, be more specific. It is far
> from obvious now what is it 'flattened result'.
>
> Also, I guess, 'flatten' is a verb, so you can not say 'is flatten'.
> Only 'is flattened'.
Squashed.
>
> 2. I do not see where do you initialize this field for
> iproto. So it is now initialized with stack garbage.
>
> (I've fixed all these things since we hurry.)
Squashed.
>
>>   };
>>     /**
>> diff --git a/src/box/vstream.h b/src/box/vstream.h
>> new file mode 100644
>> index 0000000..a8dcfc2
>> --- /dev/null
>> +++ b/src/box/vstream.h
>> @@ -0,0 +1,164 @@
>> +#ifndef TARANTOOL_VSTREAM_H_INCLUDED
>> +#define TARANTOOL_VSTREAM_H_INCLUDED
>> +/*
>> + * Copyright 2010-2018, Tarantool AUTHORS, please see AUTHORS file.
>> + *
>> + * Redistribution and use in source and binary forms, with or
>> + * without modification, are permitted provided that the following
>> + * conditions are met:
>> + *
>> + * 1. Redistributions of source code must retain the above
>> + *    copyright notice, this list of conditions and the
>> + *    following disclaimer.
>> + *
>> + * 2. Redistributions in binary form must reproduce the above
>> + *    copyright notice, this list of conditions and the following
>> + *    disclaimer in the documentation and/or other materials
>> + *    provided with the distribution.
>> + *
>> + * THIS SOFTWARE IS PROVIDED BY AUTHORS ``AS IS'' AND
>> + * ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED
>> + * TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
>> + * A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL
>> + * AUTHORS OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT,
>> + * INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL
>> + * DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF
>> + * SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR
>> + * BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF
>> + * LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
>> + * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF
>> + * THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF
>> + * SUCH DAMAGE.
>> + */
>> +
>> +#include "diag.h"
>> +#include "mpstream.h"
>
> 3. vstream should not depend on mpstream, but vice versa.
> Please, look at how struct port is done. It uses padding
> into which its descendants can lay anything.
Squashed. Also removed "struct lua_State *" from vstream.
>
>> +
>> +#if defined(__cplusplus)
>> +extern "C" {
>> +#endif /* defined(__cplusplus) */
>> +
>> +struct vstream;
>> +struct lua_State;
>> +struct port;
>> +
>> diff --git a/src/mpstream.c b/src/mpstream.c
>> index e4f7950..f9943e4 100644
>> --- a/src/mpstream.c
>> +++ b/src/mpstream.c
>> @@ -175,3 +180,76 @@ mpstream_encode_bool(struct mpstream *stream, 
>> bool val)
>>       char *pos = mp_encode_bool(data, val);
>>       mpstream_advance(stream, pos - data);
>>   }
>> +
>> +typedef void (*encode_array_f)(struct vstream *stream, uint32_t size);
>> +typedef void (*encode_map_f)(struct vstream *stream, uint32_t size);
>> +typedef void (*encode_uint_f)(struct vstream *stream, uint64_t num);
>> +typedef void (*encode_int_f)(struct vstream *stream, int64_t num);
>> +typedef void (*encode_float_f)(struct vstream *stream, float num);
>> +typedef void (*encode_double_f)(struct vstream *stream, double num);
>> +typedef void (*encode_strn_f)(struct vstream *stream, const char *str,
>> +                  uint32_t len);
>> +typedef void (*encode_nil_f)(struct vstream *stream);
>> +typedef void (*encode_bool_f)(struct vstream *stream, bool val);
>> +
>
> 4. I think, it should be part of vstream.h. And struct vstream members 
> should be
> defined with these types.
Squashed.


*New version:*

commit 795ccb629260269c8ffffb52d5e426e35ed0a088
Author: Mergen Imeev <imeevma@gmail.com>
Date:   Sat Nov 10 11:18:32 2018 +0300

     sql: create interface vstream

     If we want to use functions from execute.h not only in IPROTO we
     should create special interface. This interface will allow us to
     create different implementations for mpstream and lua_State and
     use functions from execute.c without changing them. This patch
     creates such interface and its implementation for mpstream and
     replaces mpstream functions in execute.c by methods of this
     interface.

     Needed for #3505

diff --git a/src/box/execute.c b/src/box/execute.c
index 9ba9e66..0fee5c1 100644
--- a/src/box/execute.c
+++ b/src/box/execute.c
@@ -42,7 +42,7 @@
  #include "port.h"
  #include "tuple.h"
  #include "sql/vdbe.h"
-#include "mpstream.h"
+#include "vstream.h"

  const char *sql_type_strs[] = {
      NULL,
@@ -530,12 +530,12 @@ sql_bind(const struct sql_request *request, struct 
sqlite3_stmt *stmt)
   * @retval -1 Client or memory error.
   */
  static inline int
-sql_get_description(struct sqlite3_stmt *stmt, struct mpstream *stream,
+sql_get_description(struct sqlite3_stmt *stmt, struct vstream *stream,
              int column_count)
  {
      assert(column_count > 0);
-    mpstream_encode_uint(stream, IPROTO_METADATA);
-    mpstream_encode_array(stream, column_count);
+    vstream_encode_enum(stream, IPROTO_METADATA, "metadata");
+    vstream_encode_array(stream, column_count);
      for (int i = 0; i < column_count; ++i) {
          const char *name = sqlite3_column_name(stmt, i);
          const char *type = sqlite3_column_datatype(stmt, i);
@@ -547,12 +547,19 @@ sql_get_description(struct sqlite3_stmt *stmt, 
struct mpstream *stream,
          assert(name != NULL);
          if (type == NULL)
              type = "UNKNOWN";
-        mpstream_encode_map(stream, 2);
-        mpstream_encode_uint(stream, IPROTO_FIELD_NAME);
-        mpstream_encode_str(stream, name);
-        mpstream_encode_uint(stream, IPROTO_FIELD_TYPE);
-        mpstream_encode_str(stream, type);
+        vstream_encode_map(stream, 2);
+
+        vstream_encode_enum(stream, IPROTO_FIELD_NAME, "name");
+        vstream_encode_str(stream, name);
+        vstream_encode_map_commit(stream);
+
+        vstream_encode_enum(stream, IPROTO_FIELD_TYPE, "type");
+        vstream_encode_str(stream, type);
+        vstream_encode_map_commit(stream);
+
+        vstream_encode_array_commit(stream, i);
      }
+    vstream_encode_map_commit(stream);
      return 0;
  }

@@ -611,7 +618,7 @@ sql_prepare_and_execute(const struct sql_request 
*request,

  int
  sql_response_dump(struct sql_response *response, int *keys,
-          struct mpstream *stream)
+          struct vstream *stream)
  {
      sqlite3 *db = sql_get();
      struct sqlite3_stmt *stmt = (struct sqlite3_stmt *) 
response->prep_stmt;
@@ -623,42 +630,48 @@ err:
              goto finish;
          }
          *keys = 2;
-        mpstream_encode_uint(stream, IPROTO_DATA);
-        mpstream_flush(stream);
-        if (port_dump_msgpack(&response->port, stream->ctx) < 0) {
+        vstream_encode_enum(stream, IPROTO_DATA, "rows");
+        if (vstream_encode_port(stream, &response->port) < 0) {
              /* Failed port dump destroyes the port. */
              goto err;
          }
-        mpstream_reset(stream);
+        vstream_encode_map_commit(stream);
      } else {
          *keys = 1;
          struct stailq *autoinc_id_list =
              vdbe_autoinc_id_list((struct Vdbe *)stmt);
          uint32_t map_size = stailq_empty(autoinc_id_list) ? 1 : 2;
-        mpstream_encode_uint(stream, IPROTO_SQL_INFO);
-        mpstream_encode_map(stream, map_size);
          uint64_t id_count = 0;
          if (!stailq_empty(autoinc_id_list)) {
              struct autoinc_id_entry *id_entry;
              stailq_foreach_entry(id_entry, autoinc_id_list, link)
                  id_count++;
          }
-
-        mpstream_encode_uint(stream, SQL_INFO_ROW_COUNT);
-        mpstream_encode_uint(stream, db->nChange);
+        if (!response->is_info_flattened) {
+            vstream_encode_enum(stream, IPROTO_SQL_INFO, "info");
+            vstream_encode_map(stream, map_size);
+        }
+        vstream_encode_enum(stream, SQL_INFO_ROW_COUNT, "rowcount");
+        vstream_encode_uint(stream, db->nChange);
+        vstream_encode_map_commit(stream);
          if (!stailq_empty(autoinc_id_list)) {
-            mpstream_encode_uint(stream,
-                         SQL_INFO_AUTOINCREMENT_IDS);
-            mpstream_encode_array(stream, id_count);
+            vstream_encode_enum(stream, SQL_INFO_AUTOINCREMENT_IDS,
+                        "autoincrement_ids");
+            vstream_encode_array(stream, id_count);
              struct autoinc_id_entry *id_entry;
+            int i = 0;
              stailq_foreach_entry(id_entry, autoinc_id_list, link) {
                  int64_t value = id_entry->id;
                  if (id_entry->id >= 0)
-                    mpstream_encode_uint(stream, value);
+                    vstream_encode_uint(stream, value);
                  else
-                    mpstream_encode_int(stream, value);
+                    vstream_encode_int(stream, value);
+                vstream_encode_array_commit(stream, i++);
              }
+            vstream_encode_map_commit(stream);
          }
+        if (!response->is_info_flattened)
+            vstream_encode_map_commit(stream);
      }
  finish:
      port_destroy(&response->port);
diff --git a/src/box/execute.h b/src/box/execute.h
index 65ac81c..56b7339 100644
--- a/src/box/execute.h
+++ b/src/box/execute.h
@@ -51,7 +51,7 @@ extern const char *sql_info_key_strs[];
  struct region;
  struct sql_bind;
  struct xrow_header;
-struct mpstream;
+struct vstream;

  /** EXECUTE request. */
  struct sql_request {
@@ -74,6 +74,20 @@ struct sql_response {
      struct port port;
      /** Prepared SQL statement with metadata. */
      void *prep_stmt;
+    /**
+     * SQL response can be dumped into msgpack to be sent via
+     * iproto or onto Lua stack to be returned into an
+     * application. In the first case response body has
+     * explicit field IPROTO_SQL_INFO: {rowcount = ...,
+     * autoids = ...}. But in case of Lua this field is
+     * flattened. A result never has 'info' field, it has
+     * inlined 'rowcount' and 'autoids'. In iproto
+     * IPROTO_SQL_INFO field is sent mostly to explicitly
+     * distinguish two response types: DML/DDL vs DQL,
+     * IPROTO_SQL_INFO vs IPROTO_METADATA. So this flag is set
+     * by Lua and allows to flatten SQL_INFO fields.
+     */
+    bool is_info_flattened;
  };

  /**
@@ -112,7 +126,7 @@ struct sql_response {
   */
  int
  sql_response_dump(struct sql_response *response, int *keys,
-          struct mpstream *stream);
+          struct vstream *stream);

  /**
   * Parse the EXECUTE request.
diff --git a/src/box/iproto.cc b/src/box/iproto.cc
index b110900..1c4c651 100644
--- a/src/box/iproto.cc
+++ b/src/box/iproto.cc
@@ -62,6 +62,7 @@
  #include "execute.h"
  #include "errinj.h"
  #include "mpstream.h"
+#include "vstream.h"

  enum {
      IPROTO_SALT_SIZE = 32,
@@ -1587,6 +1588,7 @@ tx_process_sql(struct cmsg *m)
      struct iproto_msg *msg = tx_accept_msg(m);
      struct obuf *out;
      struct sql_response response;
+    memset(&response, 0, sizeof(response));
      bool is_error = false;

      tx_fiber_init(msg->connection->session, msg->header.sync);
@@ -1607,16 +1609,18 @@ tx_process_sql(struct cmsg *m)
      /* Prepare memory for the iproto header. */
      if (iproto_prepare_header(out, &header_svp, IPROTO_SQL_HEADER_LEN) 
!= 0)
          goto error;
-    struct mpstream stream;
-    mpstream_init(&stream, out, obuf_reserve_cb, obuf_alloc_cb,
-              set_encode_error, &is_error);
+
+    struct vstream stream;
+    mpstream_init((struct mpstream *)&stream, out, obuf_reserve_cb,
+              obuf_alloc_cb, set_encode_error, &is_error);
      if (is_error)
          goto error;
+    mp_vstream_init_vtab(&stream);
      if (sql_response_dump(&response, &keys, &stream) != 0 || is_error) {
          obuf_rollback_to_svp(out, &header_svp);
          goto error;
      }
-    mpstream_flush(&stream);
+    mpstream_flush((struct mpstream *)&stream);
      iproto_reply_sql(out, &header_svp, response.sync, schema_version, 
keys);
      iproto_wpos_create(&msg->wpos, out);
      return;
diff --git a/src/box/vstream.h b/src/box/vstream.h
new file mode 100644
index 0000000..01a5212
--- /dev/null
+++ b/src/box/vstream.h
@@ -0,0 +1,171 @@
+#ifndef TARANTOOL_VSTREAM_H_INCLUDED
+#define TARANTOOL_VSTREAM_H_INCLUDED
+/*
+ * Copyright 2010-2018, Tarantool AUTHORS, please see AUTHORS file.
+ *
+ * Redistribution and use in source and binary forms, with or
+ * without modification, are permitted provided that the following
+ * conditions are met:
+ *
+ * 1. Redistributions of source code must retain the above
+ *    copyright notice, this list of conditions and the
+ *    following disclaimer.
+ *
+ * 2. Redistributions in binary form must reproduce the above
+ *    copyright notice, this list of conditions and the following
+ *    disclaimer in the documentation and/or other materials
+ *    provided with the distribution.
+ *
+ * THIS SOFTWARE IS PROVIDED BY AUTHORS ``AS IS'' AND
+ * ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED
+ * TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
+ * A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL
+ * AUTHORS OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT,
+ * INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL
+ * DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF
+ * SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR
+ * BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF
+ * LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
+ * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF
+ * THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF
+ * SUCH DAMAGE.
+ */
+#if defined(__cplusplus)
+extern "C" {
+#endif /* defined(__cplusplus) */
+
+struct vstream;
+struct lua_State;
+struct port;
+
+typedef void (*encode_array_f)(struct vstream *stream, uint32_t size);
+typedef void (*encode_map_f)(struct vstream *stream, uint32_t size);
+typedef void (*encode_uint_f)(struct vstream *stream, uint64_t num);
+typedef void (*encode_int_f)(struct vstream *stream, int64_t num);
+typedef void (*encode_float_f)(struct vstream *stream, float num);
+typedef void (*encode_double_f)(struct vstream *stream, double num);
+typedef void (*encode_strn_f)(struct vstream *stream, const char *str,
+                  uint32_t len);
+typedef void (*encode_nil_f)(struct vstream *stream);
+typedef void (*encode_bool_f)(struct vstream *stream, bool val);
+typedef void (*encode_enum_f)(struct vstream *stream, int64_t num,
+                  const char *str);
+typedef int (*encode_port_f)(struct vstream *stream, struct port *port);
+typedef void (*encode_array_commit_f)(struct vstream *stream, uint32_t id);
+typedef void (*encode_map_commit_f)(struct vstream *stream);
+
+struct vstream_vtab {
+    encode_array_f encode_array;
+    encode_map_f encode_map;
+    encode_uint_f encode_uint;
+    encode_int_f encode_int;
+    encode_float_f encode_float;
+    encode_double_f encode_double;
+    encode_strn_f encode_strn;
+    encode_nil_f encode_nil;
+    encode_bool_f encode_bool;
+    encode_enum_f encode_enum;
+    encode_port_f encode_port;
+    encode_array_commit_f encode_array_commit;
+    encode_map_commit_f encode_map_commit;
+};
+
+struct vstream {
+    /** Here struct mpstream lives under the hood. */
+    char inheritance_padding[64];
+    /** Virtual function table. */
+    const struct vstream_vtab *vtab;
+};
+
+void
+mp_vstream_init_vtab(struct vstream *vstream);
+
+static inline void
+vstream_encode_array(struct vstream *stream, uint32_t size)
+{
+    return stream->vtab->encode_array(stream, size);
+}
+
+static inline void
+vstream_encode_map(struct vstream *stream, uint32_t size)
+{
+    return stream->vtab->encode_map(stream, size);
+}
+
+static inline void
+vstream_encode_uint(struct vstream *stream, uint64_t num)
+{
+    return stream->vtab->encode_uint(stream, num);
+}
+
+static inline void
+vstream_encode_int(struct vstream *stream, int64_t num)
+{
+    return stream->vtab->encode_int(stream, num);
+}
+
+static inline void
+vstream_encode_float(struct vstream *stream, float num)
+{
+    return stream->vtab->encode_float(stream, num);
+}
+
+static inline void
+vstream_encode_double(struct vstream *stream, double num)
+{
+    return stream->vtab->encode_double(stream, num);
+}
+
+static inline void
+vstream_encode_strn(struct vstream *stream, const char *str, uint32_t len)
+{
+    return stream->vtab->encode_strn(stream, str, len);
+}
+
+static inline void
+vstream_encode_str(struct vstream *stream, const char *str)
+{
+    return stream->vtab->encode_strn(stream, str, strlen(str));
+}
+
+static inline void
+vstream_encode_nil(struct vstream *stream)
+{
+    return stream->vtab->encode_nil(stream);
+}
+
+static inline void
+vstream_encode_bool(struct vstream *stream, bool val)
+{
+    return stream->vtab->encode_bool(stream, val);
+}
+
+static inline void
+vstream_encode_enum(struct vstream *stream, int64_t num, const char *str)
+{
+    return stream->vtab->encode_enum(stream, num, str);
+}
+
+static inline int
+vstream_encode_port(struct vstream *stream, struct port *port)
+{
+    return stream->vtab->encode_port(stream, port);
+}
+
+static inline void
+vstream_encode_map_commit(struct vstream *stream)
+{
+    return stream->vtab->encode_map_commit(stream);
+}
+
+static inline void
+vstream_encode_array_commit(struct vstream *stream, uint32_t id)
+{
+    return stream->vtab->encode_array_commit(stream, id);
+}
+
+#if defined(__cplusplus)
+} /* extern "C" */
+#endif /* defined(__cplusplus) */
+
+#endif /* TARANTOOL_VSTREAM_H_INCLUDED */
diff --git a/src/mpstream.c b/src/mpstream.c
index e4f7950..4091ead 100644
--- a/src/mpstream.c
+++ b/src/mpstream.c
@@ -33,6 +33,8 @@
  #include <assert.h>
  #include <stdint.h>
  #include "msgpuck.h"
+#include "box/vstream.h"
+#include "box/port.h"

  void
  mpstream_reserve_slow(struct mpstream *stream, size_t size)
@@ -175,3 +177,54 @@ mpstream_encode_bool(struct mpstream *stream, bool val)
      char *pos = mp_encode_bool(data, val);
      mpstream_advance(stream, pos - data);
  }
+
+int
+mp_vstream_encode_port(struct vstream *stream, struct port *port)
+{
+    struct mpstream *mpstream = (struct mpstream *)stream;
+    mpstream_flush(mpstream);
+    if (port_dump_msgpack(port, mpstream->ctx) < 0) {
+        /* Failed port dump destroyes the port. */
+        return -1;
+    }
+    mpstream_reset(mpstream);
+    return 0;
+}
+
+void
+mp_vstream_encode_enum(struct vstream *stream, int64_t num, const char 
*str)
+{
+    (void)str;
+    if (num < 0)
+        mpstream_encode_int((struct mpstream *)stream, num);
+    else
+        mpstream_encode_uint((struct mpstream *)stream, num);
+}
+
+void
+mp_vstream_noop(struct vstream *stream, ...)
+{
+    (void) stream;
+}
+
+const struct vstream_vtab mp_vstream_vtab = {
+    /** encode_array = */ (encode_array_f)mpstream_encode_array,
+    /** encode_map = */ (encode_map_f)mpstream_encode_map,
+    /** encode_uint = */ (encode_uint_f)mpstream_encode_uint,
+    /** encode_int = */ (encode_int_f)mpstream_encode_int,
+    /** encode_float = */ (encode_float_f)mpstream_encode_float,
+    /** encode_double = */ (encode_double_f)mpstream_encode_double,
+    /** encode_strn = */ (encode_strn_f)mpstream_encode_strn,
+    /** encode_nil = */ (encode_nil_f)mpstream_encode_nil,
+    /** encode_bool = */ (encode_bool_f)mpstream_encode_bool,
+    /** encode_enum = */ mp_vstream_encode_enum,
+    /** encode_port = */ mp_vstream_encode_port,
+    /** encode_array_commit = */ (encode_array_commit_f)mp_vstream_noop,
+    /** encode_map_commit = */ (encode_map_commit_f)mp_vstream_noop,
+};
+
+void
+mp_vstream_init_vtab(struct vstream *vstream)
+{
+    vstream->vtab = &mp_vstream_vtab;
+}


[-- Attachment #2: Type: text/html, Size: 30676 bytes --]

[tarantool-patches] Re: [PATCH v2 6/7] lua: create vstream implementation for Lua

[Imeev Mergen]

[-- Attachment #1: Type: text/plain, Size: 20938 bytes --]

Hi! Thank you for review and fixes! I squashed you fixes and done
some changes:
- Luastream created. Some methods were moved from Lua
   implementation for vstream to luastream.
- Binding now saves some data in allocated memory as they can be
   removed from memory after being poped from Lua stack.
- Lua implementation for vstream was moved to luastream.c

New patch and some answers below.

On 11/23/18 12:49 AM, Vladislav Shpilevoy wrote:
> Thanks for the fixes! See my 5 comments below, fix
> at the end of the email and on the branch.
>
> Also note, that I did not run tests. So before squashing
> please, check the tests.
>
>> diff --git a/src/box/execute.h b/src/box/execute.h
>> index 5a11a8a..8ee0a89 100644
>> --- a/src/box/execute.h
>> +++ b/src/box/execute.h
>> @@ -131,6 +131,21 @@ xrow_decode_sql(const struct xrow_header *row, 
>> struct sql_request *request,
>>           struct region *region);
>>     /**
>> + * Parse Lua table of SQL parameters and store a result
>> + * into the @request->bind, bind_count.
>> + * @param L Lua stack to get data from.
>> + * @param request Request to save decoded parameters.
>> + * @param idx Position of table with parameters on Lua stack.
>> + * @param region Allocator.
>> + *
>> + * @retval  0 Success.
>> + * @retval -1 Client or memory error.
>> + */
>> +int
>> +lua_sql_bind_list_decode(struct lua_State *L, struct sql_request 
>> *request,
>> +             int idx, struct region *region);
> 1. You do not need to pass region here. Get fiber()->gc inside this
> function. In original implementation it is passed because back in
> those days we hoped to remove fiber()->gc, but now we decided not to
> do it.
>
> Also, it leaks now. You allocate sql_bind parameters on region, but
> never truncate it. Iproto has the same bug, but you should not
> duplicate it.
>
> But you can not truncate it as well, since sql_prepare_and_execute
> could save some transactional data onto it. And this bug iproto does
> not have since it uses region of iproto thread.
>
> I think, here you should use malloc.
After discussion it was decided that we will use region for now.
region will be cleared in sqlite3_finalize(). Comment added.
>
>>         lua_pushnil(L);
>>         lua_next(L, lua_gettop(L) - 1);
>>         struct luaL_field field_name;
>>         lua_pushvalue(L, -2);
>>         luaL_tofield(L, luaL_msgpack_default, -1, &field_name);
>>         lua_pop(L, 1);
>>         assert(field_name.type == MP_STR);
>>         luaL_tofield(L, luaL_msgpack_default, -1, field);
>>         lua_pop(L, 1);
>>         bind->name = field_name.sval.data;
>>         bind->name_len = field_name.sval.len;
>
> 2. Please, do not use luaL_tofield for each scalar value. Here
> it is much simpler to just use lua_tostring or something.
Fixed.
>
>>     for (uint32_t i = 0; i < bind_count; ++i) {
>>         struct luaL_field field;
>>         lua_rawgeti(L, idx, i + 1);
>>         luaL_tofield(L, luaL_msgpack_default, -1, &field);
>>         if (lua_sql_bind_decode(L, &bind[i], i, &field) != 0) {
>>             region_truncate(region, used);
>>             return -1;
>>         }
>>         lua_pop(L, 1);
>>     }
>
> 3. Why not to do lua_rawgeti and luaL_tofield inside lua_sql_bind_decode?
Fixed. Moved them to lua_sql_bind_decode().
>
> 4. Lua vstream should not be implemented in sql.c. It is a separate file
> luastream.c or something. You should have separate luastream 
> implementation
> without any vtabs and lua_vstream_vtab specially for vstream. Just like
> it is done now for mpstream.
Fixed. Added struct luastream in new file luastream.c
>
> 5. API of creating mp_vstream and lua_vstream is asymmetrical: to create
> mp_vstream you call two functions to initialize mpstream and then vtab,
> but to create lua_vstream you call one function. I think, it should be
> fixed automatically once you've fixed the previous comment.
Fixed.

*New version:*

commit cbd68868b8abf60beede0fcbdd144ca99fa49d15
Author: Mergen Imeev <imeevma@gmail.com>
Date:   Thu Nov 15 18:55:29 2018 +0300

     lua: create vstream implementation for Lua

     Thas patch creates vstream implementation for Lua and function
     box.sql.new_execute() that uses this implementation. Also it
     creates parameters binding for SQL statements executed through
     box.

     Part of #3505
     Closes #3401

diff --git a/src/box/CMakeLists.txt b/src/box/CMakeLists.txt
index d127647..9f5b2b9 100644
--- a/src/box/CMakeLists.txt
+++ b/src/box/CMakeLists.txt
@@ -138,6 +138,7 @@ add_library(box STATIC
      lua/session.c
      lua/net_box.c
      lua/xlog.c
+    lua/luastream.c
      lua/sql.c
      ${bin_sources})

diff --git a/src/box/execute.c b/src/box/execute.c
index 0fee5c1..efe4e79 100644
--- a/src/box/execute.c
+++ b/src/box/execute.c
@@ -43,6 +43,8 @@
  #include "tuple.h"
  #include "sql/vdbe.h"
  #include "vstream.h"
+#include "lua/utils.h"
+#include "lua/msgpack.h"

  const char *sql_type_strs[] = {
      NULL,
@@ -299,6 +301,195 @@ error:
  }

  /**
+ * Decode a single bind column from Lua stack.
+ *
+ * @param L Lua stack.
+ * @param[out] bind Bind to decode to.
+ * @param idx Position of table with bind columns on Lua stack.
+ * @param i Ordinal bind number.
+ *
+ * @retval  0 Success.
+ * @retval -1 Memory or client error.
+ */
+static inline int
+lua_sql_bind_decode(struct lua_State *L, struct sql_bind *bind, int 
idx, int i)
+{
+    struct luaL_field field;
+    char *buf;
+    lua_rawgeti(L, idx, i + 1);
+    luaL_tofield(L, luaL_msgpack_default, -1, &field);
+    bind->pos = i + 1;
+    if (field.type == MP_MAP) {
+        /*
+         * A named parameter is an MP_MAP with
+         * one key - {'name': value}.
+         * Report parse error otherwise.
+         */
+        if (field.size != 1) {
+            diag_set(ClientError, ER_ILLEGAL_PARAMS, "SQL bind "\
+                 "parameter should be {'name': value}");
+            return -1;
+        }
+        /*
+         * Get key and value of the only map element to
+         * lua stack.
+         */
+        lua_pushnil(L);
+        lua_next(L, lua_gettop(L) - 1);
+        /* At first we should deal with the value. */
+        luaL_tofield(L, luaL_msgpack_default, -1, &field);
+        lua_pop(L, 1);
+        /* Now key is on the top of Lua stack. */
+        size_t name_len = 0;
+        bind->name = luaL_checklstring(L, -1, &name_len);
+        if (bind->name == NULL) {
+            diag_set(ClientError, ER_ILLEGAL_PARAMS, "SQL bind "\
+                 "parameter should be {'name': value}");
+            return -1;
+        }
+        /*
+         * Name should be saved in allocated memory as it
+         * will be poped from Lua stack.
+         */
+        buf = region_alloc(&fiber()->gc, name_len + 1);
+        if (buf == NULL) {
+            diag_set(OutOfMemory, name_len + 1, "region_alloc",
+                 "buf");
+            return -1;
+        }
+        memcpy(buf, bind->name, name_len + 1);
+        bind->name = buf;
+        bind->name_len = name_len;
+        lua_pop(L, 1);
+    } else {
+        bind->name = NULL;
+        bind->name_len = 0;
+    }
+    switch (field.type) {
+    case MP_UINT: {
+        bind->i64 = field.ival;
+        bind->type = SQLITE_INTEGER;
+        bind->bytes = sizeof(bind->i64);
+        break;
+    }
+    case MP_INT:
+        bind->i64 = field.ival;
+        bind->type = SQLITE_INTEGER;
+        bind->bytes = sizeof(bind->i64);
+        break;
+    case MP_STR:
+        /*
+         * Data should be saved in allocated memory as it
+         * will be poped from Lua stack.
+         */
+        buf = region_alloc(&fiber()->gc, field.sval.len + 1);
+        if (buf == NULL) {
+            diag_set(OutOfMemory, field.sval.len + 1,
+                 "region_alloc", "buf");
+            return -1;
+        }
+        memcpy(buf, field.sval.data, field.sval.len + 1);
+        bind->s = buf;
+        bind->type = SQLITE_TEXT;
+        bind->bytes = field.sval.len;
+        break;
+    case MP_DOUBLE:
+        bind->d = field.dval;
+        bind->type = SQLITE_FLOAT;
+        bind->bytes = sizeof(bind->d);
+        break;
+    case MP_FLOAT:
+        bind->d = field.dval;
+        bind->type = SQLITE_FLOAT;
+        bind->bytes = sizeof(bind->d);
+        break;
+    case MP_NIL:
+        bind->type = SQLITE_NULL;
+        bind->bytes = 1;
+        break;
+    case MP_BOOL:
+        /* SQLite doesn't support boolean. Use int instead. */
+        bind->i64 = field.bval ? 1 : 0;
+        bind->type = SQLITE_INTEGER;
+        bind->bytes = sizeof(bind->i64);
+        break;
+    case MP_BIN:
+        bind->s = mp_decode_bin(&field.sval.data, &bind->bytes);
+        bind->type = SQLITE_BLOB;
+        break;
+    case MP_EXT:
+        /*
+         * Data should be saved in allocated memory as it
+         * will be poped from Lua stack.
+         */
+        buf = region_alloc(&fiber()->gc, sizeof(field));
+        if (buf == NULL) {
+            diag_set(OutOfMemory, sizeof(field), "region_alloc",
+                 "buf");
+            return -1;
+        }
+        memcpy(buf, &field, sizeof(field));
+        bind->s = buf;
+        bind->bytes = sizeof(field);
+        bind->type = SQLITE_BLOB;
+        break;
+    case MP_ARRAY:
+        diag_set(ClientError, ER_SQL_BIND_TYPE, "ARRAY",
+             sql_bind_name(bind));
+        return -1;
+    case MP_MAP:
+        diag_set(ClientError, ER_SQL_BIND_TYPE, "MAP",
+             sql_bind_name(bind));
+        return -1;
+    default:
+        unreachable();
+    }
+    lua_pop(L, 1);
+    return 0;
+}
+
+int
+lua_sql_bind_list_decode(struct lua_State *L, struct sql_request *request,
+             int idx)
+{
+    assert(request != NULL);
+    if (! lua_istable(L, idx)) {
+        diag_set(ClientError, ER_INVALID_MSGPACK, "SQL parameter list");
+        return -1;
+    }
+    uint32_t bind_count = lua_objlen(L, idx);
+    if (bind_count == 0)
+        return 0;
+    if (bind_count > SQL_BIND_PARAMETER_MAX) {
+        diag_set(ClientError, ER_SQL_BIND_PARAMETER_MAX,
+             (int) bind_count);
+        return -1;
+    }
+    struct region *region = &fiber()->gc;
+    uint32_t used = region_used(region);
+    size_t size = sizeof(struct sql_bind) * bind_count;
+    /*
+     * Memory allocated here will be freed in
+     * sqlite3_finalize() or in txn_commit()/txn_rollback() if
+     * there is an active transaction.
+     */
+    struct sql_bind *bind = (struct sql_bind *) region_alloc(region, size);
+    if (bind == NULL) {
+        diag_set(OutOfMemory, size, "region_alloc", "bind");
+        return -1;
+    }
+    for (uint32_t i = 0; i < bind_count; ++i) {
+        if (lua_sql_bind_decode(L, &bind[i], idx, i) != 0) {
+            region_truncate(region, used);
+            return -1;
+        }
+    }
+    request->bind_count = bind_count;
+    request->bind = bind;
+    return 0;
+}
+
+/**
   * Serialize a single column of a result set row.
   * @param stmt Prepared and started statement. At least one
   *        sqlite3_step must be called.
diff --git a/src/box/execute.h b/src/box/execute.h
index 56b7339..e186340 100644
--- a/src/box/execute.h
+++ b/src/box/execute.h
@@ -142,6 +142,20 @@ xrow_decode_sql(const struct xrow_header *row, 
struct sql_request *request,
          struct region *region);

  /**
+ * Parse Lua table of SQL parameters and store a result
+ * into the @request->bind, bind_count.
+ * @param L Lua stack to get data from.
+ * @param request Request to save decoded parameters.
+ * @param idx Position of table with parameters on Lua stack.
+ *
+ * @retval  0 Success.
+ * @retval -1 Client or memory error.
+ */
+int
+lua_sql_bind_list_decode(struct lua_State *L, struct sql_request *request,
+             int idx);
+
+/**
   * Prepare and execute an SQL statement.
   * @param request IProto request.
   * @param[out] response Response to store result.
diff --git a/src/box/lua/luastream.c b/src/box/lua/luastream.c
new file mode 100644
index 0000000..79a5246
--- /dev/null
+++ b/src/box/lua/luastream.c
@@ -0,0 +1,151 @@
+/*
+ * Copyright 2010-2018, Tarantool AUTHORS, please see AUTHORS file.
+ *
+ * Redistribution and use in source and binary forms, with or
+ * without modification, are permitted provided that the following
+ * conditions are met:
+ *
+ * 1. Redistributions of source code must retain the above
+ *    copyright notice, this list of conditions and the
+ *    following disclaimer.
+ *
+ * 2. Redistributions in binary form must reproduce the above
+ *    copyright notice, this list of conditions and the following
+ *    disclaimer in the documentation and/or other materials
+ *    provided with the distribution.
+ *
+ * THIS SOFTWARE IS PROVIDED BY <COPYRIGHT HOLDER> ``AS IS'' AND
+ * ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED
+ * TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
+ * A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL
+ * <COPYRIGHT HOLDER> OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT,
+ * INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL
+ * DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF
+ * SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR
+ * BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF
+ * LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
+ * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF
+ * THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF
+ * SUCH DAMAGE.
+ */
+
+#include "lua/utils.h"
+#include "box/execute.h"
+#include "box/vstream.h"
+
+struct luastream {
+    struct lua_State *L;
+};
+
+void
+luastream_init(struct luastream *stream, struct lua_State *L)
+{
+    struct luastream *luastream = (struct luastream *)stream;
+    luastream->L = L;
+}
+
+void
+luastream_encode_array(struct luastream *stream, uint32_t size)
+{
+    lua_createtable(stream->L, size, 0);
+}
+
+void
+luastream_encode_map(struct luastream *stream, uint32_t size)
+{
+    lua_createtable(stream->L, size, 0);
+}
+
+void
+luastream_encode_uint(struct luastream *stream, uint64_t num)
+{
+    luaL_pushuint64(stream->L, num);
+}
+
+void
+luastream_encode_int(struct luastream *stream, int64_t num)
+{
+    luaL_pushint64(stream->L, num);
+}
+
+void
+luastream_encode_float(struct luastream *stream, float num)
+{
+    lua_pushnumber(stream->L, num);
+}
+
+void
+luastream_encode_double(struct luastream *stream, double num)
+{
+    lua_pushnumber(stream->L, num);
+}
+
+void
+luastream_encode_strn(struct luastream *stream, const char *str, 
uint32_t len)
+{
+    lua_pushlstring(stream->L, str, len);
+}
+
+void
+luastream_encode_nil(struct luastream *stream)
+{
+    lua_pushnil(stream->L);
+}
+
+void
+luastream_encode_bool(struct luastream *stream, bool val)
+{
+    lua_pushboolean(stream->L, val);
+}
+
+int
+lua_vstream_encode_port(struct vstream *stream, struct port *port)
+{
+    port_dump_lua(port, ((struct luastream *)stream)->L);
+    return 0;
+}
+
+void
+lua_vstream_encode_enum(struct vstream *stream, int64_t num, const char 
*str)
+{
+    (void)num;
+    lua_pushlstring(((struct luastream *)stream)->L, str, strlen(str));
+}
+
+void
+lua_vstream_encode_map_commit(struct vstream *stream)
+{
+    size_t length;
+    const char *key = lua_tolstring(((struct luastream *)stream)->L, -2,
+                    &length);
+    lua_setfield(((struct luastream *)stream)->L, -3, key);
+    lua_pop(((struct luastream *)stream)->L, 1);
+}
+
+void
+lua_vstream_encode_array_commit(struct vstream *stream, uint32_t id)
+{
+    lua_rawseti(((struct luastream *)stream)->L, -2, id + 1);
+}
+
+const struct vstream_vtab lua_vstream_vtab = {
+    /** encode_array = */ (encode_array_f)luastream_encode_array,
+    /** encode_map = */ (encode_map_f)luastream_encode_map,
+    /** encode_uint = */ (encode_uint_f)luastream_encode_uint,
+    /** encode_int = */ (encode_int_f)luastream_encode_int,
+    /** encode_float = */ (encode_float_f)luastream_encode_float,
+    /** encode_double = */ (encode_double_f)luastream_encode_double,
+    /** encode_strn = */ (encode_strn_f)luastream_encode_strn,
+    /** encode_nil = */ (encode_nil_f)luastream_encode_nil,
+    /** encode_bool = */ (encode_bool_f)luastream_encode_bool,
+    /** encode_enum = */ lua_vstream_encode_enum,
+    /** encode_port = */ lua_vstream_encode_port,
+    /** encode_array_commit = */ lua_vstream_encode_array_commit,
+    /** encode_map_commit = */ lua_vstream_encode_map_commit,
+};
+
+void
+lua_vstream_init_vtab(struct vstream *stream)
+{
+    stream->vtab = &lua_vstream_vtab;
+}
diff --git a/src/box/lua/sql.c b/src/box/lua/sql.c
index 17e2694..7567afc 100644
--- a/src/box/lua/sql.c
+++ b/src/box/lua/sql.c
@@ -6,6 +6,8 @@
  #include "box/info.h"
  #include "lua/utils.h"
  #include "info.h"
+#include "box/execute.h"
+#include "box/vstream.h"

  static void
  lua_push_column_names(struct lua_State *L, struct sqlite3_stmt *stmt)
@@ -111,6 +113,39 @@ sqlerror:
  }

  static int
+lbox_execute(struct lua_State *L)
+{
+    struct sqlite3 *db = sql_get();
+    if (db == NULL)
+        return luaL_error(L, "not ready");
+
+    size_t length;
+    const char *sql = lua_tolstring(L, 1, &length);
+    if (sql == NULL)
+        return luaL_error(L, "usage: box.execute(sqlstring)");
+
+    struct sql_request request = {};
+    request.sql_text = sql;
+    request.sql_text_len = length;
+    if (lua_gettop(L) == 2 && lua_sql_bind_list_decode(L, &request, 2) 
!= 0)
+        return luaT_error(L);
+    struct sql_response response = {.is_info_flattened = true};
+    if (sql_prepare_and_execute(&request, &response, &fiber()->gc) != 0)
+        return luaT_error(L);
+
+    int keys;
+    struct vstream vstream;
+    luastream_init((struct luastream *)&vstream, L);
+    lua_vstream_init_vtab(&vstream);
+    lua_newtable(L);
+    if (sql_response_dump(&response, &keys, &vstream) != 0) {
+        lua_pop(L, 1);
+        return luaT_error(L);
+    }
+    return 1;
+}
+
+static int
  lua_sql_debug(struct lua_State *L)
  {
      struct info_handler info;
@@ -124,6 +159,7 @@ box_lua_sqlite_init(struct lua_State *L)
  {
      static const struct luaL_Reg module_funcs [] = {
          {"execute", lua_sql_execute},
+        {"new_execute", lbox_execute},
          {"debug", lua_sql_debug},
          {NULL, NULL}
      };
diff --git a/src/box/lua/sql.h b/src/box/lua/sql.h
index 65ff6d5..a83a5b8 100644
--- a/src/box/lua/sql.h
+++ b/src/box/lua/sql.h
@@ -36,9 +36,13 @@ extern "C" {
  #endif

  struct lua_State;
+struct luastream;

  void box_lua_sqlite_init(struct lua_State *L);

+void
+luastream_init(struct luastream *stream, struct lua_State *L);
+
  #ifdef __cplusplus
  }
  #endif
diff --git a/src/box/vstream.h b/src/box/vstream.h
index 01a5212..b846b68 100644
--- a/src/box/vstream.h
+++ b/src/box/vstream.h
@@ -71,7 +71,10 @@ struct vstream_vtab {
  };

  struct vstream {
-    /** Here struct mpstream lives under the hood. */
+    /**
+     * Here struct mpstream or struct luastream lives under
+     * the hood.
+     */
      char inheritance_padding[64];
      /** Virtual function table. */
      const struct vstream_vtab *vtab;
@@ -80,6 +83,9 @@ struct vstream {
  void
  mp_vstream_init_vtab(struct vstream *vstream);

+void
+lua_vstream_init_vtab(struct vstream *vstream);
+
  static inline void
  vstream_encode_array(struct vstream *stream, uint32_t size)
  {


[-- Attachment #2: Type: text/html, Size: 28639 bytes --]