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

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

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,195 @@ xrow_decode_sql(const struct xrow_header *row, struct sql_request *request,
 	return 0;
 }
 
+/**
+ * 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

src/box/execute.h

@@ -141,6 +141,20 @@ int
 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.

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;
+}