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

[imeevma@tarantool.org]

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