[Tarantool-patches] [PATCH v3 09/20] port: add result set format and request type to port_sql

[Nikita Pettik <korablev@tarantool.org>]

Result set serialization formats of DQL and DML queries are different:
the last one contains number of affected rows and optionally list of
autoincremented ids; the first one comprises all meta-information
including column names of resulting set and their types. What is more,
serialization format is going to be different for execute and prepare
requests. So let's introduce separate member to struct port_sql
responsible for serialization format to be used.

Note that C standard specifies that enums are integers, but it does not
specify the size. Hence, let's use simple uint8 - mentioned enum are
small enough to fit into it.

What is more, prepared statement finalization is required only for
PREPARE-AND-EXECUTE requests. So let's keep flag indicating required
finalization as well.

Needed for #2592
---
 src/box/execute.c     | 34 +++++++++++++++++++++++++---------
 src/box/execute.h     | 21 +++++++++++++++++++++
 src/box/lua/execute.c | 22 +++++++++++++++-------
 3 files changed, 61 insertions(+), 16 deletions(-)

diff --git a/src/box/execute.c b/src/box/execute.c
index fb83e1194..3bc4988b7 100644
--- a/src/box/execute.c
+++ b/src/box/execute.c
@@ -100,7 +100,9 @@ static void
 port_sql_destroy(struct port *base)
 {
 	port_tuple_vtab.destroy(base);
-	sql_stmt_finalize(((struct port_sql *)base)->stmt);
+	struct port_sql *port_sql = (struct port_sql *) base;
+	if (port_sql->do_finalize)
+		sql_stmt_finalize(((struct port_sql *)base)->stmt);
 }
 
 const struct port_vtab port_sql_vtab = {
@@ -114,11 +116,15 @@ const struct port_vtab port_sql_vtab = {
 };
 
 static void
-port_sql_create(struct port *port, struct sql_stmt *stmt)
+port_sql_create(struct port *port, struct sql_stmt *stmt,
+		enum sql_serialization_format format, bool do_finalize)
 {
 	port_tuple_create(port);
-	((struct port_sql *)port)->stmt = stmt;
 	port->vtab = &port_sql_vtab;
+	struct port_sql *port_sql = (struct port_sql *) port;
+	port_sql->stmt = stmt;
+	port_sql->serialization_format = format;
+	port_sql->do_finalize = do_finalize;
 }
 
 /**
@@ -324,9 +330,10 @@ port_sql_dump_msgpack(struct port *port, struct obuf *out)
 {
 	assert(port->vtab == &port_sql_vtab);
 	sql *db = sql_get();
-	struct sql_stmt *stmt = ((struct port_sql *)port)->stmt;
-	int column_count = sql_column_count(stmt);
-	if (column_count > 0) {
+	struct port_sql *sql_port = (struct port_sql *)port;
+	struct sql_stmt *stmt = sql_port->stmt;
+	switch (sql_port->serialization_format) {
+	case DQL_EXECUTE: {
 		int keys = 2;
 		int size = mp_sizeof_map(keys);
 		char *pos = (char *) obuf_alloc(out, size);
@@ -335,7 +342,7 @@ port_sql_dump_msgpack(struct port *port, struct obuf *out)
 			return -1;
 		}
 		pos = mp_encode_map(pos, keys);
-		if (sql_get_metadata(stmt, out, column_count) != 0)
+		if (sql_get_metadata(stmt, out, sql_column_count(stmt)) != 0)
 			return -1;
 		size = mp_sizeof_uint(IPROTO_DATA);
 		pos = (char *) obuf_alloc(out, size);
@@ -346,7 +353,9 @@ port_sql_dump_msgpack(struct port *port, struct obuf *out)
 		pos = mp_encode_uint(pos, IPROTO_DATA);
 		if (port_tuple_vtab.dump_msgpack(port, out) < 0)
 			return -1;
-	} else {
+		break;
+	}
+	case DML_EXECUTE: {
 		int keys = 1;
 		assert(((struct port_tuple *)port)->size == 0);
 		struct stailq *autoinc_id_list =
@@ -395,6 +404,11 @@ port_sql_dump_msgpack(struct port *port, struct obuf *out)
 				      mp_encode_int(buf, id_entry->id);
 			}
 		}
+		break;
+	}
+	default: {
+		unreachable();
+	}
 	}
 	return 0;
 }
@@ -445,7 +459,9 @@ sql_prepare_and_execute(const char *sql, int len, const struct sql_bind *bind,
 	if (sql_stmt_compile(sql, len, NULL, &stmt, NULL) != 0)
 		return -1;
 	assert(stmt != NULL);
-	port_sql_create(port, stmt);
+	enum sql_serialization_format format = sql_column_count(stmt) > 0 ?
+					   DQL_EXECUTE : DML_EXECUTE;
+	port_sql_create(port, stmt, format, true);
 	if (sql_bind(stmt, bind, bind_count) == 0 &&
 	    sql_execute(stmt, port, region) == 0)
 		return 0;
diff --git a/src/box/execute.h b/src/box/execute.h
index ce1e7a67d..c87e765cf 100644
--- a/src/box/execute.h
+++ b/src/box/execute.h
@@ -46,6 +46,17 @@ enum sql_info_key {
 	sql_info_key_MAX,
 };
 
+/**
+ * One of possible formats used to dump msgpack/Lua.
+ * For details see port_sql_dump_msgpack() and port_sql_dump_lua().
+ */
+enum sql_serialization_format {
+	DQL_EXECUTE = 0,
+	DML_EXECUTE = 1,
+	DQL_PREPARE = 2,
+	DML_PREPARE = 3,
+};
+
 extern const char *sql_info_key_strs[];
 
 struct region;
@@ -85,6 +96,16 @@ struct port_sql {
 	struct port_tuple port_tuple;
 	/* Prepared SQL statement. */
 	struct sql_stmt *stmt;
+	/**
+	 * Serialization format depends on type of SQL query: DML or
+	 * DQL; and on type of SQL request: execute or prepare.
+	 */
+	uint8_t serialization_format;
+	/**
+	 * There's no need in clean-up in case of PREPARE request:
+	 * statement remains in cache and will be deleted later.
+	 */
+	bool do_finalize;
 };
 
 extern const struct port_vtab port_sql_vtab;
diff --git a/src/box/lua/execute.c b/src/box/lua/execute.c
index 68adacf72..b164ffcaf 100644
--- a/src/box/lua/execute.c
+++ b/src/box/lua/execute.c
@@ -45,18 +45,21 @@ port_sql_dump_lua(struct port *port, struct lua_State *L, bool is_flat)
 	assert(is_flat == false);
 	assert(port->vtab == &port_sql_vtab);
 	struct sql *db = sql_get();
-	struct sql_stmt *stmt = ((struct port_sql *)port)->stmt;
-	int column_count = sql_column_count(stmt);
-	if (column_count > 0) {
+	struct port_sql *port_sql = (struct port_sql *)port;
+	struct sql_stmt *stmt = port_sql->stmt;
+	switch (port_sql->serialization_format) {
+	case DQL_EXECUTE: {
 		lua_createtable(L, 0, 2);
-		lua_sql_get_metadata(stmt, L, column_count);
+		lua_sql_get_metadata(stmt, L, sql_column_count(stmt));
 		lua_setfield(L, -2, "metadata");
 		port_tuple_vtab.dump_lua(port, L, false);
 		lua_setfield(L, -2, "rows");
-	} else {
-		assert(((struct port_tuple *)port)->size == 0);
+		break;
+	}
+	case DML_EXECUTE: {
+		assert(((struct port_tuple *) port)->size == 0);
 		struct stailq *autoinc_id_list =
-			vdbe_autoinc_id_list((struct Vdbe *)stmt);
+			vdbe_autoinc_id_list((struct Vdbe *) stmt);
 		lua_createtable(L, 0, stailq_empty(autoinc_id_list) ? 1 : 2);
 
 		luaL_pushuint64(L, db->nChange);
@@ -77,6 +80,11 @@ port_sql_dump_lua(struct port *port, struct lua_State *L, bool is_flat)
 				sql_info_key_strs[SQL_INFO_AUTOINCREMENT_IDS];
 			lua_setfield(L, -2, field_name);
 		}
+		break;
+	}
+	default: {
+		unreachable();
+	}
 	}
 }
 
-- 
2.15.1