CONC-348: Add callback support for prepared statements

Client application can now register callback functions for either sending or retrieving data:
- typedef void (*ps_result_callback)(MYSQL_STMT *stmt, unsigned int column, unsigned char **row);
- typedef my_bool *(*ps_param_callback)(MYSQL_STMT *stmt, MYSQL_BIND *bind, unsigned int row_nr);

These functions will be registerd via mysql_stmt_attr_set call by specifying options STMT_ATTR_CB_PARAM
or STMT_ATTR_CB_RESULT.

include/mariadb_stmt.h

@@ -62,10 +62,15 @@ enum enum_stmt_attr_type
   STMT_ATTR_UPDATE_MAX_LENGTH,
   STMT_ATTR_CURSOR_TYPE,
   STMT_ATTR_PREFETCH_ROWS,
+
+  /* MariaDB only */
   STMT_ATTR_PREBIND_PARAMS=200,
   STMT_ATTR_ARRAY_SIZE,
   STMT_ATTR_ROW_SIZE,
-  STMT_ATTR_STATE
+  STMT_ATTR_STATE,
+  STMT_ATTR_CB_USER_DATA,
+  STMT_ATTR_CB_PARAM,
+  STMT_ATTR_CB_RESULT
 };
 
 enum enum_cursor_type
@@ -195,6 +200,8 @@ struct st_mysqlnd_stmt_methods
 };
 
 typedef int  (*mysql_stmt_fetch_row_func)(MYSQL_STMT *stmt, unsigned char **row);
+typedef void (*ps_result_callback)(MYSQL_STMT *stmt, unsigned int column, unsigned char **row);
+typedef my_bool *(*ps_param_callback)(MYSQL_STMT *stmt, MYSQL_BIND *bind, unsigned int row_nr);
 
 struct st_mysql_stmt
 {
@@ -234,6 +241,9 @@ struct st_mysql_stmt
   unsigned int             array_size;
   size_t row_size;
   unsigned int prebind_params;
+  void *user_data;
+  ps_result_callback result_callback;
+  ps_param_callback param_callback;
 };
 
 typedef void (*ps_field_fetch_func)(MYSQL_BIND *r_param, const MYSQL_FIELD * field, unsigned char **row);

libmariadb/mariadb_stmt.c

@@ -374,11 +374,16 @@ int mthd_stmt_fetch_to_bind(MYSQL_STMT *stmt, unsigned char *row)
   {
     /* save row position for fetching values in pieces */
     if (*null_ptr & bit_offset)
+    {
+      if (stmt->result_callback)
+        stmt->result_callback(stmt, i, NULL);
+      else
       {
         if (!stmt->bind[i].is_null)
           stmt->bind[i].is_null= &stmt->bind[i].is_null_value;
         *stmt->bind[i].is_null= 1;
         stmt->bind[i].u.row_ptr= NULL;
+      }
     } else
     {
       stmt->bind[i].u.row_ptr= row;
@@ -387,6 +392,9 @@ int mthd_stmt_fetch_to_bind(MYSQL_STMT *stmt, unsigned char *row)
       {
         unsigned long length;
 
+        if (stmt->result_callback)
+          stmt->result_callback(stmt, i, &row);
+        else {
           if (mysql_ps_fetch_functions[stmt->fields[i].type].pack_len >= 0)
             length= mysql_ps_fetch_functions[stmt->fields[i].type].pack_len;
           else
@@ -396,6 +404,7 @@ int mthd_stmt_fetch_to_bind(MYSQL_STMT *stmt, unsigned char *row)
             stmt->bind[i].length= &stmt->bind[i].length_value;
           *stmt->bind[i].length= stmt->bind[i].length_value= length;
         }
+      }
       else
       {
         if (!stmt->bind[i].length)
@@ -928,6 +937,11 @@ unsigned char* mysql_stmt_execute_generate_bulk_request(MYSQL_STMT *stmt, size_t
     /* calculate data size */
     for (j=0; j < stmt->array_size; j++)
     {
+      /* If callback for parameters was specified, we need to
+         update bind information for new row */
+      if (stmt->param_callback)
+        stmt->param_callback(stmt, stmt->params, j);
+
       if (mysql_stmt_skip_paramset(stmt, j))
         continue;
 
@@ -1049,6 +1063,9 @@ my_bool STDCALL mysql_stmt_attr_get(MYSQL_STMT *stmt, enum enum_stmt_attr_type a
     case STMT_ATTR_ROW_SIZE:
       *(size_t *)value= stmt->row_size;
       break;
+    case STMT_ATTR_CB_USER_DATA:
+      *((void **)value) = stmt->user_data;
+      break;
     default:
       return(1);
   }
@@ -1091,6 +1108,15 @@ my_bool STDCALL mysql_stmt_attr_set(MYSQL_STMT *stmt, enum enum_stmt_attr_type a
   case STMT_ATTR_ROW_SIZE:
     stmt->row_size= *(size_t *)value;
     break;
+  case STMT_ATTR_CB_RESULT:
+    stmt->result_callback= (ps_result_callback)value;
+    break;
+  case STMT_ATTR_CB_PARAM:
+    stmt->param_callback= (ps_param_callback)value;
+    break;
+  case STMT_ATTR_CB_USER_DATA:
+    stmt->user_data= (void *)value;
+    break;
   default:
     SET_CLIENT_STMT_ERROR(stmt, CR_NOT_IMPLEMENTED, SQLSTATE_UNKNOWN, 0);
     return(1);