MDEV-34718: Trigger doesn't work correctly with bulk update

Running an UPDATE statement in PS mode and having positional
parameter(s) bound with an array of actual values (that is
prepared to be run in bulk mode) results in incorrect behaviour
in presence of on update trigger that also executes an UPDATE
statement. The same is true for handling a DELETE statement in
presence of on delete trigger. Typically, the visible effect of
such incorrect behaviour is expressed in a wrong number of
updated/deleted rows of a target table. Additionally, in case UPDATE
statement, a number of modified rows and a state message returned
by a statement contains wrong information about a number of modified rows.

The reason for incorrect number of updated/deleted rows is that
a data structure used for binding positional argument with its
actual values is stored in THD (this is thd->bulk_param) and reused
on processing every INSERT/UPDATE/DELETE statement. It leads to
consuming actual values bound with top-level UPDATE/DELETE statement
by other DML statements used by triggers' body.

To fix the issue, reset the thd->bulk_param temporary to the value
nullptr before invoking triggers and restore its value on finishing
its execution.

The second part of the problem relating with wrong value of affected
rows reported by Connector/C API is caused by the fact that diagnostics
area is reused by an original DML statement and a statement invoked
by a trigger. This fact should be take into account on finalizing a
state of diagnostics area on completion running of a statement.

Important remark: in case the macros DBUG_OFF is on, call of the method
  Diagnostics_area::reset_diagnostics_area()
results in reset of the data members
  m_affected_rows, m_statement_warn_count.
Values of these data members of the class Diagnostics_area are used on
sending OK and EOF messages. In case DML statement is executed in PS bulk
mode such resetting results in sending wrong result values to a client
for affected rows in case the DML statement fires a triggers. So, reset
these data members only in case the current statement being processed
is not run in bulk mode.

sql/sql_base.cc

@@ -8784,10 +8784,22 @@ fill_record_n_invoke_before_triggers(THD *thd, TABLE *table,
 
   if (!result && triggers)
   {
+    void *save_bulk_param= thd->bulk_param;
+    /*
+      Reset the sentinel thd->bulk_param in order not to consume the next
+      values of a bound array in case one of statement executed by
+      the trigger's body is INSERT statement.
+    */
+    thd->bulk_param= nullptr;
+
     if (triggers->process_triggers(thd, event, TRG_ACTION_BEFORE,
                                     TRUE) ||
         not_null_fields_have_null_values(table))
+    {
+      thd->bulk_param= save_bulk_param;
       return TRUE;
+    }
+    thd->bulk_param= save_bulk_param;
 
     /*
       Re-calculate virtual fields to cater for cases when base columns are

sql/sql_delete.cc

@@ -796,13 +796,17 @@ bool mysql_delete(THD *thd, TABLE_LIST *table_list, COND *conds,
                                               delete_history);
     if (delete_record)
     {
+      void *save_bulk_param= thd->bulk_param;
+      thd->bulk_param= nullptr;
       if (!delete_history && table->triggers &&
           table->triggers->process_triggers(thd, TRG_EVENT_DELETE,
                                             TRG_ACTION_BEFORE, FALSE))
       {
         error= 1;
+        thd->bulk_param= save_bulk_param;
         break;
       }
+      thd->bulk_param= save_bulk_param;
 
       // no LIMIT / OFFSET
       if (returning && result->send_data(returning->item_list) < 0)
@@ -833,13 +837,16 @@ bool mysql_delete(THD *thd, TABLE_LIST *table_list, COND *conds,
       if (likely(!error))
       {
 	deleted++;
+	thd->bulk_param= nullptr;
         if (!delete_history && table->triggers &&
             table->triggers->process_triggers(thd, TRG_EVENT_DELETE,
                                               TRG_ACTION_AFTER, FALSE))
         {
           error= 1;
+          thd->bulk_param= save_bulk_param;
           break;
         }
+        thd->bulk_param= save_bulk_param;
 	if (!--limit && using_limit)
 	{
 	  error= -1;

sql/sql_error.cc

@@ -309,13 +309,26 @@ Diagnostics_area::reset_diagnostics_area()
   m_message[0]= '\0';
   Sql_state_errno::clear();
   Sql_user_condition_identity::clear();
-  m_affected_rows= 0;
   m_last_insert_id= 0;
+  if (!is_bulk_op())
+  {
+    m_affected_rows= 0;
     m_statement_warn_count= 0;
+  }
 #endif
   get_warning_info()->clear_error_condition();
   set_is_sent(false);
   /** Tiny reset in debug mode to see garbage right away */
+  if (!is_bulk_op())
+    /*
+      For BULK DML operations (e.g. UPDATE) the data member m_status
+      has the value DA_OK_BULK. Keep this value in order to handle
+      m_affected_rows, m_statement_warn_count in correct way. Else,
+      the number of rows and the number of warnings affected by
+      the last statement executed as part of a trigger fired by the dml
+      (e.g. UPDATE statement fires a trigger on AFTER UPDATE) would counts
+      rows modified by trigger's statement.
+    */
     m_status= DA_EMPTY;
   DBUG_VOID_RETURN;
 }

sql/sql_error.h

@@ -1066,6 +1066,11 @@ class Diagnostics_area: public Sql_state_errno,
     return m_affected_rows;
   }
 
+  void set_message(const char *msg)
+  {
+    strmake_buf(m_message, msg);
+  }
+
   ulonglong last_insert_id() const
   {
     DBUG_ASSERT(m_status == DA_OK || m_status == DA_OK_BULK);

sql/sql_insert.cc

@@ -1021,19 +1021,11 @@ bool mysql_insert(THD *thd, TABLE_LIST *table_list,
         */
         restore_record(table,s->default_values);	// Get empty record
         table->reset_default_fields();
-        /*
-          Reset the sentinel thd->bulk_param in order not to consume the next
-          values of a bound array in case one of statement executed by
-          the trigger's body is INSERT statement.
-        */
-        void *save_bulk_param= thd->bulk_param;
-        thd->bulk_param= nullptr;
 
         if (unlikely(fill_record_n_invoke_before_triggers(thd, table, fields,
                                                           *values, 0,
                                                           TRG_EVENT_INSERT)))
         {
-          thd->bulk_param= save_bulk_param;
           if (values_list.elements != 1 && ! thd->is_error())
           {
             info.records++;
@@ -1047,7 +1039,6 @@ bool mysql_insert(THD *thd, TABLE_LIST *table_list,
 	  error=1;
 	  break;
         }
-        thd->bulk_param= save_bulk_param;
       }
       else
       {
@@ -1334,7 +1325,18 @@ bool mysql_insert(THD *thd, TABLE_LIST *table_list,
     */
    if (returning)
       result->send_eof();
-   else
+   else if (!(thd->in_sub_stmt & SUB_STMT_TRIGGER))
+     /*
+       Set the status and the number of affected rows in Diagnostics_area
+       only in case the INSERT statement is not processed as part of a trigger
+       invoked by some other DML statement. Else we would result in incorrect
+       number of affected rows for bulk DML operations, e.g. the UPDATE
+       statement (called via PS protocol). It would happen since the data
+       member Diagnostics_area::m_affected_rows modified twice per DML
+       statement - first time at the end of handling the INSERT statement
+       invoking by a trigger fired on handling the original DML statement,
+       and the second time at the end of handling the original DML statement.
+     */
       my_ok(thd, info.copied + info.deleted +
                ((thd->client_capabilities & CLIENT_FOUND_ROWS) ?
                 info.touched : info.updated), id);
@@ -1356,7 +1358,18 @@ bool mysql_insert(THD *thd, TABLE_LIST *table_list,
               (long) thd->get_stmt_da()->current_statement_warn_count());
     if (returning)
       result->send_eof();
-    else
+    else if (!(thd->in_sub_stmt & SUB_STMT_TRIGGER))
+      /*
+        Set the status and the number of affected rows in Diagnostics_area
+        only in case the INSERT statement is not processed as part of a trigger
+        invoked by some other DML statement. Else we would result in incorrect
+        number of affected rows for bulk DML operations, e.g. the UPDATE
+        statement (called via PS protocol). It would happen since the data
+        member Diagnostics_area::m_affected_rows modified twice per DML
+        statement - first time at the end of handling the INSERT statement
+        invoking by a trigger fired on handling the original DML statement,
+        and the second time at the end of handling the original DML statement.
+      */
       ::my_ok(thd, info.copied + info.deleted + updated, id, buff);
   }
   thd->abort_on_warning= 0;

sql/sql_update.cc

@@ -1146,13 +1146,19 @@ int mysql_update(THD *thd,
         rows_inserted++;
       }
 
+      void *save_bulk_param= thd->bulk_param;
+      thd->bulk_param= nullptr;
+
       if (table->triggers &&
           unlikely(table->triggers->process_triggers(thd, TRG_EVENT_UPDATE,
                                                      TRG_ACTION_AFTER, TRUE)))
       {
         error= 1;
+        thd->bulk_param= save_bulk_param;
+
         break;
       }
+      thd->bulk_param= save_bulk_param;
 
       if (!--limit && using_limit)
       {
@@ -1342,6 +1348,20 @@ int mysql_update(THD *thd,
                   (ulong) thd->get_stmt_da()->current_statement_warn_count());
     my_ok(thd, (thd->client_capabilities & CLIENT_FOUND_ROWS) ? found : updated,
           id, buff);
+    if (thd->get_stmt_da()->is_bulk_op())
+    {
+      /*
+        Update the diagnostics message sent to a client with number of actual
+        rows update by the statement. For bulk UPDATE operation it should be
+        done after returning from my_ok() since the final number of updated
+        rows be knows on finishing the entire bulk update statement.
+      */
+      my_snprintf(buff, sizeof(buff), ER_THD(thd, ER_UPDATE_INFO),
+                  (ulong) thd->get_stmt_da()->affected_rows(),
+                  (ulong) thd->get_stmt_da()->affected_rows(),
+                  (ulong) thd->get_stmt_da()->current_statement_warn_count());
+      thd->get_stmt_da()->set_message(buff);
+    }
     DBUG_PRINT("info",("%ld records updated", (long) updated));
   }
   thd->count_cuted_fields= CHECK_FIELD_IGNORE;		/* calc cuted fields */