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)
  {