MDEV-12471: BULK Command

BULK execution moved to a new command.

1 files changed, 158 insertions, 64 deletions

diff --git a/sql/sql_prepare.cc b/sql/sql_prepare.cc
index cd330d8ee26..11274fbbaa2 100644
--- a/sql/sql_prepare.cc
+++ b/sql/sql_prepare.cc
@@ -164,7 +164,6 @@ public:
   Server_side_cursor *cursor;
   uchar *packet;
   uchar *packet_end;
-  ulong iterations;
   uint param_count;
   uint last_errno;
   uint flags;
@@ -183,7 +182,9 @@ public:
   */
   uint select_number_after_prepare;
   char last_error[MYSQL_ERRMSG_SIZE];
+  my_bool iterations;
   my_bool start_param;
+  my_bool read_types;
 #ifndef EMBEDDED_LIBRARY
   bool (*set_params)(Prepared_statement *st, uchar *data, uchar *data_end,
                      uchar *read_pos, String *expanded_query);
@@ -213,11 +214,10 @@ public:
                     uchar *packet_arg, uchar *packet_end_arg);
   bool execute_bulk_loop(String *expanded_query,
                          bool open_cursor,
-                         uchar *packet_arg, uchar *packet_end_arg,
-                         ulong iterations);
+                         uchar *packet_arg, uchar *packet_end_arg);
   bool execute_server_runnable(Server_runnable *server_runnable);
   my_bool set_bulk_parameters(bool reset);
-  ulong bulk_iterations();
+  bool bulk_iterations() { return iterations; };
   /* Destroy this statement */
   void deallocate();
   bool execute_immediate(const char *query, uint query_length);
@@ -935,6 +935,7 @@ static bool insert_params(Prepared_statement *stmt, uchar *null_array,
   for (Item_param **it= begin; it < end; ++it)
   {
     Item_param *param= *it;
+    param->indicator= STMT_INDICATOR_NONE; // only for bulk parameters
     if (!param->has_long_data_value())
     {
       if (is_param_null(null_array, (uint) (it - begin)))
@@ -979,10 +980,7 @@ static bool insert_bulk_params(Prepared_statement *stmt,
       param->reset();
     if (!param->has_long_data_value())
     {
-      if (param->indicators)
-        param->indicator= (enum_indicator_type) *((*read_pos)++);
-      else
-        param->indicator= STMT_INDICATOR_NONE;
+      param->indicator= (enum_indicator_type) *((*read_pos)++);
       if ((*read_pos) > data_end)
         DBUG_RETURN(1);
       switch (param->indicator)
@@ -993,6 +991,8 @@ static bool insert_bulk_params(Prepared_statement *stmt,
         param->set_param_func(param, read_pos, (uint) (data_end - (*read_pos)));
         if (param->has_no_value())
           DBUG_RETURN(1);
+        if (param->convert_str_value(stmt->thd))
+          DBUG_RETURN(1);                           /* out of memory */
         break;
       case STMT_INDICATOR_NULL:
         param->set_null();
@@ -1011,6 +1011,36 @@ static bool insert_bulk_params(Prepared_statement *stmt,
   DBUG_RETURN(0);
 }
 
+static bool set_conversion_functions(Prepared_statement *stmt,
+                                     uchar **data, uchar *data_end)
+{
+  uchar *read_pos= *data;
+  const uint signed_bit= 1 << 15;
+  DBUG_ENTER("set_conversion_functions");
+  /*
+     First execute or types altered by the client, setup the
+     conversion routines for all parameters (one time)
+   */
+  Item_param **it= stmt->param_array;
+  Item_param **end= it + stmt->param_count;
+  THD *thd= stmt->thd;
+  for (; it < end; ++it)
+  {
+    ushort typecode;
+
+    if (read_pos >= data_end)
+      DBUG_RETURN(1);
+
+    typecode= sint2korr(read_pos);
+    read_pos+= 2;
+    (**it).unsigned_flag= MY_TEST(typecode & signed_bit);
+    setup_one_conversion_function(thd, *it, (uchar) (typecode & 0xff));
+  }
+  *data= read_pos;
+  DBUG_RETURN(0);
+}
+
+
 static bool setup_conversion_functions(Prepared_statement *stmt,
                                        uchar **data, uchar *data_end,
                                        bool bulk_protocol= 0)
@@ -1024,30 +1054,9 @@ static bool setup_conversion_functions(Prepared_statement *stmt,
 
   if (*read_pos++) //types supplied / first execute
   {
-    /*
-      First execute or types altered by the client, setup the
-      conversion routines for all parameters (one time)
-    */
-    Item_param **it= stmt->param_array;
-    Item_param **end= it + stmt->param_count;
-    THD *thd= stmt->thd;
-    for (; it < end; ++it)
-    {
-      ushort typecode;
-      const uint signed_bit= 1 << 15;
-      const uint indicators_bit= 1 << 14;
-
-      if (read_pos >= data_end)
-        DBUG_RETURN(1);
-
-      typecode= sint2korr(read_pos);
-      read_pos+= 2;
-      (**it).unsigned_flag= MY_TEST(typecode & signed_bit);
-      if (bulk_protocol)
-        (**it).indicators= MY_TEST(typecode & indicators_bit);
-      setup_one_conversion_function(thd, *it,
-                                    (uchar) (typecode & 0xff));
-    }
+    *data= read_pos;
+    bool res= set_conversion_functions(stmt, data, data_end);
+    DBUG_RETURN(res);
   }
   *data= read_pos;
   DBUG_RETURN(0);
@@ -3032,6 +3041,14 @@ static void reset_stmt_params(Prepared_statement *stmt)
 }
 
 
+static void mysql_stmt_execute_common(THD *thd,
+                                      ulong stmt_id,
+                                      uchar *packet,
+                                      uchar *packet_end,
+                                      ulong cursor_flags,
+                                      bool iteration,
+                                      bool types);
+
 /**
   COM_STMT_EXECUTE handler: execute a previously prepared statement.
 
@@ -3054,20 +3071,91 @@ void mysqld_stmt_execute(THD *thd, char *packet_arg, uint packet_length)
   uchar *packet= (uchar*)packet_arg; // GCC 4.0.1 workaround
   ulong stmt_id= uint4korr(packet);
   ulong flags= (ulong) packet[4];
-#ifndef EMBEDDED_LIBRARY
-  ulong iterations= uint4korr(packet + 5);
-#else
-  ulong iterations= 0; // no support
-#endif
+  uchar *packet_end= packet + packet_length;
+  DBUG_ENTER("mysqld_stmt_execute");
+
+  packet+= 9;                               /* stmt_id + 5 bytes of flags */
+
+  mysql_stmt_execute_common(thd, stmt_id, packet, packet_end, flags, FALSE,
+  FALSE);
+  DBUG_VOID_RETURN;
+}
+
+
+/**
+  COM_STMT_BULK_EXECUTE handler: execute a previously prepared statement.
+
+    If there are any parameters, then replace parameter markers with the
+    data supplied from the client, and then execute the statement.
+    This function uses binary protocol to send a possible result set
+    to the client.
+
+  @param thd                current thread
+  @param packet_arg         parameter types and data, if any
+  @param packet_length      packet length, including the terminator character.
+
+  @return
+    none: in case of success OK packet or a result set is sent to the
+    client, otherwise an error message is set in THD.
+*/
+
+void mysqld_stmt_bulk_execute(THD *thd, char *packet_arg, uint packet_length)
+{
+  uchar *packet= (uchar*)packet_arg; // GCC 4.0.1 workaround
+  ulong stmt_id= uint4korr(packet);
+  uint flags= (uint) uint2korr(packet + 4);
+  uchar *packet_end= packet + packet_length;
+  DBUG_ENTER("mysqld_stmt_execute_bulk");
+
+  if (!(thd->client_capabilities &
+        MARIADB_CLIENT_STMT_BULK_OPERATIONS))
+  {
+    DBUG_PRINT("error",
+               ("An attempt to execute bulk operation without support"));
+    my_error(ER_UNSUPPORTED_PS, MYF(0));
+  }
+  /* Check for implemented parameters */
+  if (flags & (~STMT_BULK_FLAG_CLIENT_SEND_TYPES))
+  {
+    DBUG_PRINT("error", ("unsupported bulk execute flags %x", flags));
+    my_error(ER_UNSUPPORTED_PS, MYF(0));
+  }
+
+  /* stmt id and two bytes of flags */
+  packet+= 4 + 2;
+  mysql_stmt_execute_common(thd, stmt_id, packet, packet_end, 0, TRUE,
+                            (flags & STMT_BULK_FLAG_CLIENT_SEND_TYPES));
+  DBUG_VOID_RETURN;
+}
+
+
+/**
+  Common part of prepared statement execution
+
+  @param thd             THD handle
+  @param stmt_id         id of the prepared statement
+  @param paket           packet with parameters to bind
+  @param packet_end      pointer to the byte after parameters end
+  @param cursor_flags    cursor flags
+  @param bulk_op         id it bulk operation
+  @param read_types      flag say that types muast been read
+*/
+
+static void mysql_stmt_execute_common(THD *thd,
+                                      ulong stmt_id,
+                                      uchar *packet,
+                                      uchar *packet_end,
+                                      ulong cursor_flags,
+                                      bool bulk_op,
+                                      bool read_types)
+{
   /* Query text for binary, general or slow log, if any of them is open */
   String expanded_query;
-  uchar *packet_end= packet + packet_length;
   Prepared_statement *stmt;
   Protocol *save_protocol= thd->protocol;
   bool open_cursor;
-  DBUG_ENTER("mysqld_stmt_execute");
-
-  packet+= 9;                               /* stmt_id + 5 bytes of flags */
+  DBUG_ENTER("mysqld_stmt_execute_common");
+  DBUG_ASSERT((!read_types) || (read_types && bulk_op));
 
   /* First of all clear possible warnings from the previous command */
   thd->reset_for_next_command();
@@ -3079,21 +3167,21 @@ void mysqld_stmt_execute(THD *thd, char *packet_arg, uint packet_length)
              llstr(stmt_id, llbuf), "mysqld_stmt_execute");
     DBUG_VOID_RETURN;
   }
+  stmt->read_types= read_types;
 
 #if defined(ENABLED_PROFILING)
   thd->profiling.set_query_source(stmt->query(), stmt->query_length());
 #endif
   DBUG_PRINT("exec_query", ("%s", stmt->query()));
-  DBUG_PRINT("info",("stmt: %p iterations: %lu", stmt, iterations));
+  DBUG_PRINT("info",("stmt: %p bulk_op %d", stmt, bulk_op));
 
-  open_cursor= MY_TEST(flags & (ulong) CURSOR_TYPE_READ_ONLY);
+  open_cursor= MY_TEST(cursor_flags & (ulong) CURSOR_TYPE_READ_ONLY);
 
   thd->protocol= &thd->protocol_binary;
-  if (iterations <= 1)
+  if (!bulk_op)
     stmt->execute_loop(&expanded_query, open_cursor, packet, packet_end);
   else
-    stmt->execute_bulk_loop(&expanded_query, open_cursor, packet, packet_end,
-                            iterations);
+    stmt->execute_bulk_loop(&expanded_query, open_cursor, packet, packet_end);
   thd->protocol= save_protocol;
 
   sp_cache_enforce_limit(thd->sp_proc_cache, stored_program_cache_size);
@@ -3600,11 +3688,12 @@ Prepared_statement::Prepared_statement(THD *thd_arg)
   cursor(0),
   packet(0),
   packet_end(0),
-  iterations(0),
   param_count(0),
   last_errno(0),
   flags((uint) IS_IN_USE),
+  iterations(0),
   start_param(0),
+  read_types(0),
   m_sql_mode(thd->variables.sql_mode)
 {
   init_sql_alloc(&main_mem_root, thd_arg->variables.query_alloc_block_size,
@@ -3641,7 +3730,7 @@ void Prepared_statement::setup_set_params()
     set_params_from_actual_params= insert_params_from_actual_params_with_log;
 #ifndef EMBEDDED_LIBRARY
     set_params= insert_params_with_log;
-    set_bulk_params= insert_bulk_params; // TODO: add binlog support
+    set_bulk_params= insert_bulk_params; // RBR is on for bulk operation
 #else
     //TODO: add bulk support for bulk parameters
     set_params_data= emb_insert_params_with_log;
@@ -4023,7 +4112,7 @@ Prepared_statement::execute_loop(String *expanded_query,
   Reprepare_observer reprepare_observer;
   bool error;
   int reprepare_attempt= 0;
-  iterations= 0;
+  iterations= FALSE;
 
   /*
     - In mysql_sql_stmt_execute() we hide all "external" Items
@@ -4126,11 +4215,11 @@ my_bool bulk_parameters_set(THD *thd)
   DBUG_RETURN(FALSE);
 }
 
-ulong bulk_parameters_iterations(THD *thd)
+my_bool bulk_parameters_iterations(THD *thd)
 {
   Prepared_statement *stmt= (Prepared_statement *) thd->bulk_param;
   if (!stmt)
-    return 1;
+    return FALSE;
   return stmt->bulk_iterations();
 }
 
@@ -4138,7 +4227,8 @@ ulong bulk_parameters_iterations(THD *thd)
 my_bool Prepared_statement::set_bulk_parameters(bool reset)
 {
   DBUG_ENTER("Prepared_statement::set_bulk_parameters");
-  DBUG_PRINT("info", ("iteration: %lu", iterations));
+  DBUG_PRINT("info", ("iteration: %d", iterations));
+
   if (iterations)
   {
 #ifndef EMBEDDED_LIBRARY
@@ -4152,31 +4242,24 @@ my_bool Prepared_statement::set_bulk_parameters(bool reset)
       reset_stmt_params(this);
       DBUG_RETURN(true);
     }
-    iterations--;
+    if (packet >= packet_end)
+      iterations= FALSE;
   }
   start_param= 0;
   DBUG_RETURN(false);
 }
 
-ulong Prepared_statement::bulk_iterations()
-{
-  if (iterations)
-    return iterations;
-  return start_param ? 1 : 0;
-}
-
 bool
 Prepared_statement::execute_bulk_loop(String *expanded_query,
                                       bool open_cursor,
                                       uchar *packet_arg,
-                                      uchar *packet_end_arg,
-                                      ulong iterations_arg)
+                                      uchar *packet_end_arg)
 {
   Reprepare_observer reprepare_observer;
   bool error= 0;
   packet= packet_arg;
   packet_end= packet_end_arg;
-  iterations= iterations_arg;
+  iterations= TRUE;
   start_param= true;
 #ifndef DBUG_OFF
   Item *free_list_state= thd->free_list;
@@ -4190,16 +4273,26 @@ Prepared_statement::execute_bulk_loop(String *expanded_query,
     thd->set_bulk_execution(0);
     return TRUE;
   }
+  /* Check for non zero parameter count*/
+  if (param_count == 0)
+  {
+    DBUG_PRINT("error", ("Statement with no parameters for bulk execution."));
+    my_error(ER_UNSUPPORTED_PS, MYF(0));
+    thd->set_bulk_execution(0);
+    return TRUE;
+  }
 
   if (!(sql_command_flags[lex->sql_command] & CF_SP_BULK_SAFE))
   {
+    DBUG_PRINT("error", ("Command is not supported in bulk execution."));
     my_error(ER_UNSUPPORTED_PS, MYF(0));
     thd->set_bulk_execution(0);
     return TRUE;
   }
 
 #ifndef EMBEDDED_LIBRARY
-  if (setup_conversion_functions(this, &packet, packet_end, TRUE))
+  if (read_types &&
+      set_conversion_functions(this, &packet, packet_end))
 #else
   // bulk parameters are not supported for embedded, so it will an error
 #endif
@@ -4210,6 +4303,7 @@ Prepared_statement::execute_bulk_loop(String *expanded_query,
     thd->set_bulk_execution(0);
     return true;
   }
+  read_types= FALSE;
 
 #ifdef NOT_YET_FROM_MYSQL_5_6
   if (unlikely(thd->security_ctx->password_expired &&