[tarantool-patches] [PATCH v8 3/6] iproto: create port_sql
imeevma at tarantool.org
imeevma at tarantool.org
Sat Jan 19 16:20:22 MSK 2019
Hi! Thank you for review. My answers, diff between versions and
new version below.
On 1/16/19 9:25 PM, Vladislav Shpilevoy wrote:
> Thanks for the patch! See 2 comments below.
>
>> commit 1a3007030ea54fa81533acbf7421f27a2399c941
>> Author: Mergen Imeev <imeevma at gmail.com>
>> Date: Fri Dec 21 14:52:03 2018 +0300
>>
>> iproto: create port_sql
>> This patch creates port_sql implementation for the port. This will
>> allow us to dump sql responses to obuf or to Lua. Also this patch
>> defines methods dump_msgpack() and destroy() of port_sql.
>> Part of #3505
>>
>> diff --git a/src/box/execute.c b/src/box/execute.c
>> index 38b6cbc..a81f482 100644
>> --- a/src/box/execute.c
>> +++ b/src/box/execute.c
>> @@ -643,7 +672,66 @@ err:
>> +/**
>> + * Execute prepared SQL statement.
>> + *
>> + * This function uses region to allocate memory for temporary
>> + * objects. After this function, region will be in the same state
>> + * in which it was before this function.
>> + *
>> + * @param db SQL handle.
>> + * @param stmt Prepared statement.
>> + * @param[out] port Port to store SQL response.
>
> 1. It is not [out]. Port should be created before calling this function,
> so it is [in][out], that we usually omit.
>
Fixed.
>> + * @param region Region to allocate temporary objects.
>> + *
>> + * @retval 0 Success.
>> + * @retval -1 Error.
>> + */
>> +static inline int
>> +sql_execute(sqlite3 *db, struct sqlite3_stmt *stmt, struct port *port,
>> + struct region *region)
>> +{
>> + int rc, column_count = sqlite3_column_count(stmt);
>> + if (column_count > 0) {
>> + /* Either ROW or DONE or ERROR. */
>> + while ((rc = sqlite3_step(stmt)) == SQLITE_ROW) {
>> + if (sql_row_to_port(stmt, column_count, region,
>> + port) != 0)
>> + return -1;
>> + }
>> + assert(rc == SQLITE_DONE || rc != SQLITE_OK);
>> + } else {
>> + /* No rows. Either DONE or ERROR. */
>> + rc = sqlite3_step(stmt);
>> + assert(rc != SQLITE_ROW && rc != SQLITE_OK);
>> + }
>> + if (rc != SQLITE_DONE) {
>> + diag_set(ClientError, ER_SQL_EXECUTE, sqlite3_errmsg(db));
>> + return -1;
>> + }
>> + return 0;
>> +}
>> diff --git a/src/box/iproto.cc b/src/box/iproto.cc
>> index a08c8c5..09e8914 100644
>> --- a/src/box/iproto.cc
>> +++ b/src/box/iproto.cc
>> @@ -1643,9 +1643,11 @@ tx_process_sql(struct cmsg *m)
>> out = msg->connection->tx.p_obuf;
>> struct obuf_svp header_svp;
>> /* Prepare memory for the iproto header. */
>> - if (iproto_prepare_header(out, &header_svp, IPROTO_HEADER_LEN) != 0)
>> + if (iproto_prepare_header(out, &header_svp, IPROTO_HEADER_LEN) != 0) {
>> + port_destroy(&port);
>> goto error;
>> - if (sql_response_dump(&response, out) != 0) {
>> + }
>> + if (port_dump_msgpack(&port, out) != 0) {
>
> 2. port_sql_dump_msgpack destroys port on an error, but port_tuple_dump_msgpack
> does not. I think, they should work similar, so port_sql_dump_msgpack does
> not destroy the port as well. Lets do it, and add here port_destroy() on an
> error. The same about port_dump_lua.
>
Fixed. Fix for port_dump_lua() in next patch.
>> obuf_rollback_to_svp(out, &header_svp);
>> goto error;
>> }
Diff between versions:
commit 12718781c2a23ceb8b20b6a23e71ec1deeb8e3be
Author: Mergen Imeev <imeevma at gmail.com>
Date: Thu Jan 17 14:24:02 2019 +0300
Temporary: Review fixes
diff --git a/src/box/execute.c b/src/box/execute.c
index a81f482..1658d9b 100644
--- a/src/box/execute.c
+++ b/src/box/execute.c
@@ -106,7 +106,6 @@ static_assert(sizeof(struct port_sql) <= sizeof(struct port),
/**
* Dump data from port to buffer. Data in port contains tuples,
* metadata, or information obtained from an executed SQL query.
- * The port is destroyed.
*
* Dumped msgpack structure:
* +----------------------------------------------+
@@ -597,30 +596,27 @@ port_sql_dump_msgpack(struct port *port, struct obuf *out)
assert(port->vtab == &port_sql_vtab);
sqlite3 *db = sql_get();
struct sqlite3_stmt *stmt = ((struct port_sql *)port)->stmt;
- int rc = 0, column_count = sqlite3_column_count(stmt);
+ int column_count = sqlite3_column_count(stmt);
if (column_count > 0) {
int keys = 2;
int size = mp_sizeof_map(keys);
char *pos = (char *) obuf_alloc(out, size);
if (pos == NULL) {
diag_set(OutOfMemory, size, "obuf_alloc", "pos");
- goto err;
+ return -1;
}
pos = mp_encode_map(pos, keys);
- if (sql_get_description(stmt, out, column_count) != 0) {
-err:
- rc = -1;
- goto finish;
- }
+ if (sql_get_description(stmt, out, column_count) != 0)
+ return -1;
size = mp_sizeof_uint(IPROTO_DATA);
pos = (char *) obuf_alloc(out, size);
if (pos == NULL) {
diag_set(OutOfMemory, size, "obuf_alloc", "pos");
- goto err;
+ return -1;
}
pos = mp_encode_uint(pos, IPROTO_DATA);
if (port_tuple_vtab.dump_msgpack(port, out) < 0)
- goto err;
+ return -1;
} else {
int keys = 1;
assert(((struct port_tuple *)port)->size == 0);
@@ -633,7 +629,7 @@ err:
char *pos = (char *) obuf_alloc(out, size);
if (pos == NULL) {
diag_set(OutOfMemory, size, "obuf_alloc", "pos");
- goto err;
+ return -1;
}
pos = mp_encode_map(pos, keys);
pos = mp_encode_uint(pos, IPROTO_SQL_INFO);
@@ -656,7 +652,7 @@ err:
char *buf = obuf_alloc(out, size);
if (buf == NULL) {
diag_set(OutOfMemory, size, "obuf_alloc", "buf");
- goto err;
+ return -1;
}
buf = mp_encode_uint(buf, SQL_INFO_ROW_COUNT);
buf = mp_encode_uint(buf, changes);
@@ -671,9 +667,7 @@ err:
}
}
}
-finish:
- port_destroy(port);
- return rc;
+ return 0;
}
/**
@@ -685,7 +679,7 @@ finish:
*
* @param db SQL handle.
* @param stmt Prepared statement.
- * @param[out] port Port to store SQL response.
+ * @param port Port to store SQL response.
* @param region Region to allocate temporary objects.
*
* @retval 0 Success.
diff --git a/src/box/iproto.cc b/src/box/iproto.cc
index 09e8914..896599d 100644
--- a/src/box/iproto.cc
+++ b/src/box/iproto.cc
@@ -1648,9 +1648,11 @@ tx_process_sql(struct cmsg *m)
goto error;
}
if (port_dump_msgpack(&port, out) != 0) {
+ port_destroy(&port);
obuf_rollback_to_svp(out, &header_svp);
goto error;
}
+ port_destroy(&port);
iproto_reply_sql(out, &header_svp, msg->header.sync, schema_version);
iproto_wpos_create(&msg->wpos, out);
return;
New version:
commit 84a0a8b7c7f923ef3ef9d65300703569f17556fd
Author: Mergen Imeev <imeevma at gmail.com>
Date: Fri Dec 21 14:52:03 2018 +0300
iproto: create port_sql
This patch creates port_sql implementation for the port. This will
allow us to dump sql responses to obuf or to Lua. Also this patch
defines methods dump_msgpack() and destroy() of port_sql.
Part of ...3505
diff --git a/src/box/execute.c b/src/box/execute.c
index 38b6cbc..1658d9b 100644
--- a/src/box/execute.c
+++ b/src/box/execute.c
@@ -83,6 +83,92 @@ struct sql_bind {
};
/**
+ * Port implementation that is used to store SQL responses and
+ * output them to obuf or Lua. This port implementation is
+ * inherited from the port_tuple structure. This allows us to use
+ * this structure in the port_tuple methods instead of port_tuple
+ * itself.
+ *
+ * The methods of port_tuple are called via explicit access to
+ * port_tuple_vtab just like C++ does with BaseClass::method, when
+ * it is called in a child method.
+ */
+struct port_sql {
+ /* port_tuple to inherit from. */
+ struct port_tuple port_tuple;
+ /* Prepared SQL statement. */
+ struct sqlite3_stmt *stmt;
+};
+
+static_assert(sizeof(struct port_sql) <= sizeof(struct port),
+ "sizeof(struct port_sql) must be <= sizeof(struct port)");
+
+/**
+ * Dump data from port to buffer. Data in port contains tuples,
+ * metadata, or information obtained from an executed SQL query.
+ *
+ * Dumped msgpack structure:
+ * +----------------------------------------------+
+ * | IPROTO_BODY: { |
+ * | IPROTO_METADATA: [ |
+ * | {IPROTO_FIELD_NAME: column name1}, |
+ * | {IPROTO_FIELD_NAME: column name2}, |
+ * | ... |
+ * | ], |
+ * | |
+ * | IPROTO_DATA: [ |
+ * | tuple, tuple, tuple, ... |
+ * | ] |
+ * | } |
+ * +-------------------- OR ----------------------+
+ * | IPROTO_BODY: { |
+ * | IPROTO_SQL_INFO: { |
+ * | SQL_INFO_ROW_COUNT: number |
+ * | SQL_INFO_AUTOINCREMENT_IDS: [ |
+ * | id, id, id, ... |
+ * | ] |
+ * | } |
+ * | } |
+ * +-------------------- OR ----------------------+
+ * | IPROTO_BODY: { |
+ * | IPROTO_SQL_INFO: { |
+ * | SQL_INFO_ROW_COUNT: number |
+ * | } |
+ * | } |
+ * +----------------------------------------------+
+ * @param port Port that contains SQL response.
+ * @param[out] out Output buffer.
+ *
+ * @retval 0 Success.
+ * @retval -1 Memory error.
+ */
+static int
+port_sql_dump_msgpack(struct port *port, struct obuf *out);
+
+static void
+port_sql_destroy(struct port *base)
+{
+ port_tuple_vtab.destroy(base);
+ sqlite3_finalize(((struct port_sql *)base)->stmt);
+}
+
+static const struct port_vtab port_sql_vtab = {
+ /* .dump_msgpack = */ port_sql_dump_msgpack,
+ /* .dump_msgpack_16 = */ NULL,
+ /* .dump_lua = */ NULL,
+ /* .dump_plain = */ NULL,
+ /* .destroy = */ port_sql_destroy,
+};
+
+static void
+port_sql_create(struct port *port, struct sqlite3_stmt *stmt)
+{
+ port_tuple_create(port);
+ ((struct port_sql *)port)->stmt = stmt;
+ port->vtab = &port_sql_vtab;
+}
+
+/**
* Return a string name of a parameter marker.
* @param Bind to get name.
* @retval Zero terminated name.
@@ -487,6 +573,7 @@ sql_get_description(struct sqlite3_stmt *stmt, struct obuf *out,
* column_name simply returns them.
*/
assert(name != NULL);
+ assert(type != NULL);
size += mp_sizeof_str(strlen(name));
size += mp_sizeof_str(strlen(type));
char *pos = (char *) obuf_alloc(out, size);
@@ -503,98 +590,36 @@ sql_get_description(struct sqlite3_stmt *stmt, struct obuf *out,
return 0;
}
-static inline int
-sql_execute(sqlite3 *db, struct sqlite3_stmt *stmt, struct port *port,
- struct region *region)
+static int
+port_sql_dump_msgpack(struct port *port, struct obuf *out)
{
- int rc, column_count = sqlite3_column_count(stmt);
- if (column_count > 0) {
- /* Either ROW or DONE or ERROR. */
- while ((rc = sqlite3_step(stmt)) == SQLITE_ROW) {
- if (sql_row_to_port(stmt, column_count, region,
- port) != 0)
- return -1;
- }
- assert(rc == SQLITE_DONE || rc != SQLITE_OK);
- } else {
- /* No rows. Either DONE or ERROR. */
- rc = sqlite3_step(stmt);
- assert(rc != SQLITE_ROW && rc != SQLITE_OK);
- }
- if (rc != SQLITE_DONE) {
- diag_set(ClientError, ER_SQL_EXECUTE, sqlite3_errmsg(db));
- return -1;
- }
- return 0;
-}
-
-int
-sql_prepare_and_execute(const char *sql, int len, const struct sql_bind *bind,
- uint32_t bind_count, struct sql_response *response,
- struct region *region)
-{
- struct sqlite3_stmt *stmt;
+ assert(port->vtab == &port_sql_vtab);
sqlite3 *db = sql_get();
- if (db == NULL) {
- diag_set(ClientError, ER_LOADING);
- return -1;
- }
- if (sqlite3_prepare_v2(db, sql, len, &stmt, NULL) != SQLITE_OK) {
- diag_set(ClientError, ER_SQL_EXECUTE, sqlite3_errmsg(db));
- return -1;
- }
- assert(stmt != NULL);
- port_tuple_create(&response->port);
- response->prep_stmt = stmt;
- if (sql_bind(stmt, bind, bind_count) == 0 &&
- sql_execute(db, stmt, &response->port, region) == 0)
- return 0;
- port_destroy(&response->port);
- sqlite3_finalize(stmt);
- return -1;
-}
-
-int
-sql_response_dump(struct sql_response *response, struct obuf *out)
-{
- 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);
+ struct sqlite3_stmt *stmt = ((struct port_sql *)port)->stmt;
+ int column_count = sqlite3_column_count(stmt);
if (column_count > 0) {
int keys = 2;
int size = mp_sizeof_map(keys);
char *pos = (char *) obuf_alloc(out, size);
if (pos == NULL) {
diag_set(OutOfMemory, size, "obuf_alloc", "pos");
- goto err;
+ return -1;
}
pos = mp_encode_map(pos, keys);
- if (sql_get_description(stmt, out, column_count) != 0) {
-err:
- rc = -1;
- goto finish;
- }
- size = mp_sizeof_uint(IPROTO_DATA) +
- mp_sizeof_array(port_tuple->size);
+ if (sql_get_description(stmt, out, column_count) != 0)
+ return -1;
+ size = mp_sizeof_uint(IPROTO_DATA);
pos = (char *) obuf_alloc(out, size);
if (pos == NULL) {
diag_set(OutOfMemory, size, "obuf_alloc", "pos");
- goto err;
+ return -1;
}
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
- */
- if (port_dump_msgpack_16(&response->port, out) < 0) {
- /* Failed port dump destroyes the port. */
- goto err;
- }
+ if (port_tuple_vtab.dump_msgpack(port, out) < 0)
+ return -1;
} else {
int keys = 1;
- assert(port_tuple->size == 0);
+ assert(((struct port_tuple *)port)->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;
@@ -604,7 +629,7 @@ err:
char *pos = (char *) obuf_alloc(out, size);
if (pos == NULL) {
diag_set(OutOfMemory, size, "obuf_alloc", "pos");
- goto err;
+ return -1;
}
pos = mp_encode_map(pos, keys);
pos = mp_encode_uint(pos, IPROTO_SQL_INFO);
@@ -627,7 +652,7 @@ err:
char *buf = obuf_alloc(out, size);
if (buf == NULL) {
diag_set(OutOfMemory, size, "obuf_alloc", "buf");
- goto err;
+ return -1;
}
buf = mp_encode_uint(buf, SQL_INFO_ROW_COUNT);
buf = mp_encode_uint(buf, changes);
@@ -642,8 +667,65 @@ err:
}
}
}
-finish:
- port_destroy(&response->port);
- sqlite3_finalize(stmt);
- return rc;
+ return 0;
+}
+
+/**
+ * Execute prepared SQL statement.
+ *
+ * This function uses region to allocate memory for temporary
+ * objects. After this function, region will be in the same state
+ * in which it was before this function.
+ *
+ * @param db SQL handle.
+ * @param stmt Prepared statement.
+ * @param port Port to store SQL response.
+ * @param region Region to allocate temporary objects.
+ *
+ * @retval 0 Success.
+ * @retval -1 Error.
+ */
+static inline int
+sql_execute(sqlite3 *db, struct sqlite3_stmt *stmt, struct port *port,
+ struct region *region)
+{
+ int rc, column_count = sqlite3_column_count(stmt);
+ if (column_count > 0) {
+ /* Either ROW or DONE or ERROR. */
+ while ((rc = sqlite3_step(stmt)) == SQLITE_ROW) {
+ if (sql_row_to_port(stmt, column_count, region,
+ port) != 0)
+ return -1;
+ }
+ assert(rc == SQLITE_DONE || rc != SQLITE_OK);
+ } else {
+ /* No rows. Either DONE or ERROR. */
+ rc = sqlite3_step(stmt);
+ assert(rc != SQLITE_ROW && rc != SQLITE_OK);
+ }
+ if (rc != SQLITE_DONE) {
+ diag_set(ClientError, ER_SQL_EXECUTE, sqlite3_errmsg(db));
+ return -1;
+ }
+ return 0;
+}
+
+int
+sql_prepare_and_execute(const char *sql, int len, const struct sql_bind *bind,
+ uint32_t bind_count, struct port *port,
+ struct region *region)
+{
+ struct sqlite3_stmt *stmt;
+ sqlite3 *db = sql_get();
+ if (sqlite3_prepare_v2(db, sql, len, &stmt, NULL) != SQLITE_OK) {
+ diag_set(ClientError, ER_SQL_EXECUTE, sqlite3_errmsg(db));
+ return -1;
+ }
+ assert(stmt != NULL);
+ port_sql_create(port, stmt);
+ if (sql_bind(stmt, bind, bind_count) == 0 &&
+ sql_execute(db, stmt, port, region) == 0)
+ return 0;
+ port_destroy(port);
+ return -1;
}
diff --git a/src/box/execute.h b/src/box/execute.h
index 60b8f31..1ed9ab5 100644
--- a/src/box/execute.h
+++ b/src/box/execute.h
@@ -48,18 +48,9 @@ enum sql_info_key {
extern const char *sql_info_key_strs[];
-struct obuf;
struct region;
struct sql_bind;
-/** Response on EXECUTE request. */
-struct sql_response {
- /** Port with response data if any. */
- struct port port;
- /** Prepared SQL statement with metadata. */
- void *prep_stmt;
-};
-
/**
* Parse MessagePack array of SQL parameters.
* @param data MessagePack array of parameters. Each parameter
@@ -78,48 +69,12 @@ int
sql_bind_list_decode(const char *data, struct sql_bind **out_bind);
/**
- * Dump a built response into @an out buffer. The response is
- * destroyed.
- * Response structure:
- * +----------------------------------------------+
- * | IPROTO_OK, sync, schema_version ... | iproto_header
- * +----------------------------------------------+---------------
- * | Body - a map with one or two keys. |
- * | |
- * | IPROTO_BODY: { |
- * | IPROTO_METADATA: [ |
- * | {IPROTO_FIELD_NAME: column name1}, |
- * | {IPROTO_FIELD_NAME: column name2}, | iproto_body
- * | ... |
- * | ], |
- * | |
- * | IPROTO_DATA: [ |
- * | tuple, tuple, tuple, ... |
- * | ] |
- * | } |
- * +-------------------- OR ----------------------+
- * | IPROTO_BODY: { |
- * | IPROTO_SQL_INFO: { |
- * | SQL_INFO_ROW_COUNT: number |
- * | } |
- * | } |
- * +----------------------------------------------+
- * @param response EXECUTE response.
- * @param out Output buffer.
- *
- * @retval 0 Success.
- * @retval -1 Memory error.
- */
-int
-sql_response_dump(struct sql_response *response, struct obuf *out);
-
-/**
* Prepare and execute an SQL statement.
* @param sql SQL statement.
* @param len Length of @a sql.
* @param bind Array of parameters.
* @param bind_count Length of @a bind.
- * @param[out] response Response to store result.
+ * @param[out] port Port to store SQL response.
* @param region Runtime allocator for temporary objects
* (columns, tuples ...).
*
@@ -128,7 +83,7 @@ sql_response_dump(struct sql_response *response, struct obuf *out);
*/
int
sql_prepare_and_execute(const char *sql, int len, const struct sql_bind *bind,
- uint32_t bind_count, struct sql_response *response,
+ uint32_t bind_count, struct port *port,
struct region *region);
#if defined(__cplusplus)
diff --git a/src/box/iproto.cc b/src/box/iproto.cc
index a08c8c5..896599d 100644
--- a/src/box/iproto.cc
+++ b/src/box/iproto.cc
@@ -1616,7 +1616,7 @@ tx_process_sql(struct cmsg *m)
{
struct iproto_msg *msg = tx_accept_msg(m);
struct obuf *out;
- struct sql_response response;
+ struct port port;
struct sql_bind *bind;
int bind_count;
const char *sql;
@@ -1633,7 +1633,7 @@ tx_process_sql(struct cmsg *m)
goto error;
sql = msg->sql.sql_text;
sql = mp_decode_str(&sql, &len);
- if (sql_prepare_and_execute(sql, len, bind, bind_count, &response,
+ if (sql_prepare_and_execute(sql, len, bind, bind_count, &port,
&fiber()->gc) != 0)
goto error;
/*
@@ -1643,12 +1643,16 @@ tx_process_sql(struct cmsg *m)
out = msg->connection->tx.p_obuf;
struct obuf_svp header_svp;
/* Prepare memory for the iproto header. */
- if (iproto_prepare_header(out, &header_svp, IPROTO_HEADER_LEN) != 0)
+ if (iproto_prepare_header(out, &header_svp, IPROTO_HEADER_LEN) != 0) {
+ port_destroy(&port);
goto error;
- if (sql_response_dump(&response, out) != 0) {
+ }
+ if (port_dump_msgpack(&port, out) != 0) {
+ port_destroy(&port);
obuf_rollback_to_svp(out, &header_svp);
goto error;
}
+ port_destroy(&port);
iproto_reply_sql(out, &header_svp, msg->header.sync, schema_version);
iproto_wpos_create(&msg->wpos, out);
return;
diff --git a/src/box/port.h b/src/box/port.h
index ad1b349..f188036 100644
--- a/src/box/port.h
+++ b/src/box/port.h
@@ -65,7 +65,6 @@ extern const struct port_vtab port_tuple_vtab;
static inline struct port_tuple *
port_tuple(struct port *port)
{
- assert(port->vtab == &port_tuple_vtab);
return (struct port_tuple *)port;
}
--
2.7.4
More information about the Tarantool-patches
mailing list