SQL API: add multiple result sets API

MySQL may return multiple result sets from a call to a stored
procedure (the CALL statement), or when multiple queries are specified
in a single client request. The application is supposed to process those
multiple result sets one by one with calls to `mysql_next_result()`, or
its prepared statements counterpart, `mysql_stmt_next_result()`.

Additionally, there is the `mysql_more_results()` call which allows the
application to check if more result sets are available and works for
both regular queries and prepared statements API.

One way to handle multiple results in sysbench would be consuming all
result sets silently in the MySQL driver, but that would make it
impossible for scripts to get access to individual result sets returned
by a stored procedure.

Now sysbench exposes those MySQL client API calls to the SQL API, so it
is up to the script authors to handle multiple result sets when either
stored procedures are used in a benchmark script, or multiple queries
are passed to `sql_connection:query()`:

- sql_connection:next_result()
- sql_connection:more_results()
- sql_statement:next_results()

Here is an example how multiple results can be handled in a benchmark
script:

```lua
   local rs = con:query([[CALL p1("foo")]])
   while rs ~= nil do
      -- handle the result set
      rs = con:next_result()
   end
```

Here is a prepared statement example:

```lua
   stmt = con:prepare("CALL p1(?)")
   param = stmt:bind_create(sysbench.sql.type.CHAR, 10)
   stmt:bind_param(param)
   param:set("bar")
   rs = stmt:execute()

   while rs ~= nil do
      rs = stmt:next_result()
   end
```

Fixes GH-304.

src/db_driver.c

@@ -504,6 +504,52 @@ db_result_t *db_execute(db_stmt_t *stmt)
   return NULL;
 }
 
+/* Retrieve the next result of a prepared statement */
+
+db_result_t *db_stmt_next_result(db_stmt_t *stmt)
+{
+  db_conn_t       *con = stmt->connection;
+  db_result_t     *rs = &con->rs;
+  int             rc;
+
+  if (con->state == DB_CONN_INVALID)
+  {
+    log_text(LOG_ALERT, "attempt to use an already closed connection");
+    return NULL;
+  }
+  else if (con->state == DB_CONN_RESULT_SET &&
+           (rc = db_free_results_int(con)) != 0)
+  {
+    return NULL;
+  }
+
+  rs->statement = stmt;
+
+  if (con->driver->ops.stmt_next_result == NULL)
+  {
+    con->error = DB_ERROR_NONE;
+    return NULL;
+  }
+
+  con->error = con->driver->ops.stmt_next_result(stmt, rs);
+
+  sb_counter_inc(con->thread_id, rs->counter);
+
+  if (SB_LIKELY(con->error == DB_ERROR_NONE))
+  {
+    if (rs->counter == SB_CNT_READ)
+    {
+      con->state = DB_CONN_RESULT_SET;
+      return rs;
+    }
+    con->state = DB_CONN_READY;
+
+    return NULL;
+  }
+
+  return NULL;
+}
+
 
 /* Fetch row from result set of a query */
 
@@ -588,6 +634,66 @@ db_result_t *db_query(db_conn_t *con, const char *query, size_t len)
   return NULL;
 }
 
+/* Check if more result sets are available */
+
+bool db_more_results(db_conn_t *con)
+{
+  if (con->state == DB_CONN_INVALID)
+  {
+    log_text(LOG_ALERT, "attempt to use an already closed connection");
+    return false;
+  }
+
+  if (con->state != DB_CONN_RESULT_SET ||
+      con->driver->ops.more_results == NULL ||
+      con->driver->ops.more_results(con) == false)
+    return false;
+
+  return true;
+}
+
+/* Retrieve the next result set */
+
+db_result_t *db_next_result(db_conn_t *con)
+{
+  db_result_t *rs = &con->rs;
+  int         rc;
+
+  if (con->state == DB_CONN_INVALID)
+  {
+    log_text(LOG_ALERT, "attempt to use an already closed connection");
+    con->error = DB_ERROR_FATAL;
+    return NULL;
+  }
+  else if (con->state == DB_CONN_RESULT_SET &&
+           (rc = db_free_results_int(con)) != 0)
+  {
+    con->error = DB_ERROR_FATAL;
+    return NULL;
+  }
+
+  if (con->driver->ops.next_result == NULL)
+  {
+    con->error = DB_ERROR_NONE;
+    return NULL;
+  }
+
+  con->error = con->driver->ops.next_result(con, rs);
+
+  sb_counter_inc(con->thread_id, rs->counter);
+
+  if (SB_LIKELY(con->error == DB_ERROR_NONE))
+  {
+    if (rs->counter == SB_CNT_READ)
+    {
+      con->state = DB_CONN_RESULT_SET;
+      return rs;
+    }
+    con->state = DB_CONN_READY;
+  }
+
+  return NULL;
+}
 
 /* Free result set */
 

src/db_driver.h

@@ -131,10 +131,14 @@ typedef int drv_op_prepare(struct db_stmt *, const char *, size_t);
 typedef int drv_op_bind_param(struct db_stmt *, db_bind_t *, size_t);
 typedef int drv_op_bind_result(struct db_stmt *, db_bind_t *, size_t);
 typedef db_error_t drv_op_execute(struct db_stmt *, struct db_result *);
+typedef db_error_t drv_op_stmt_next_result(struct db_stmt *,
+                                           struct db_result *);
 typedef int drv_op_fetch(struct db_result *);
 typedef int drv_op_fetch_row(struct db_result *, struct db_row *);
 typedef db_error_t drv_op_query(struct db_conn *, const char *, size_t,
                                 struct db_result *);
+typedef bool drv_op_more_results(struct db_conn *);
+typedef db_error_t drv_op_next_result(struct db_conn *, struct db_result *);
 typedef int drv_op_free_results(struct db_result *);
 typedef int drv_op_close(struct db_stmt *);
 typedef int drv_op_thread_done(int);
@@ -152,9 +156,14 @@ typedef struct
   drv_op_bind_param      *bind_param;     /* bind params for prepared statement */
   drv_op_bind_result     *bind_result;    /* bind results for prepared statement */
   drv_op_execute         *execute;        /* execute prepared statement */
+  /* retrieve the next result of a prepared statement */
+  drv_op_stmt_next_result *stmt_next_result;
   drv_op_fetch           *fetch;          /* fetch row for prepared statement */
   drv_op_fetch_row       *fetch_row;      /* fetch row for queries */
   drv_op_free_results    *free_results;   /* free result set */
+  drv_op_more_results    *more_results;   /* check if more result sets are
+                                             available */
+  drv_op_next_result     *next_result;    /* retrieve the next result set */
   drv_op_close           *close;          /* close prepared statement */
   drv_op_query           *query;          /* execute non-prepared statement */
   drv_op_thread_done     *thread_done;    /* thread-local driver deinitialization */
@@ -256,7 +265,6 @@ typedef struct db_stmt
   db_bind_t       *bound_res;      /* Array of bound results for emulated PS */ 
   db_bind_t       *bound_res_len;  /* Length of the bound_res array */
   char            emulated;        /* Should this statement be emulated? */
-  sb_counter_type_t  counter;       /* Query type */
   void            *ptr;            /* Pointer to driver-specific data structure */
 } db_stmt_t;
 
@@ -290,12 +298,18 @@ int db_bind_result(db_stmt_t *, db_bind_t *, size_t);
 
 db_result_t *db_execute(db_stmt_t *);
 
+db_result_t *db_stmt_next_result(db_stmt_t *);
+
 db_row_t *db_fetch_row(db_result_t *);
 
 db_result_t *db_query(db_conn_t *, const char *, size_t len);
 
 int db_free_results(db_result_t *);
 
+bool db_more_results(db_conn_t *);
+
+db_result_t *db_next_result(db_conn_t *);
+
 int db_store_results(db_result_t *);
 
 int db_close(db_stmt_t *);