MDEV-15152 Optimistic parallel slave doesnt cope well with START SLAVE UNTIL

The immediate bug was caused by a failure to recognize a correct
position to stop the slave applier run in optimistic parallel mode.
There were the following set of issues that the analysis unveil.
1 incorrect estimate for the event binlog position passed to
  is_until_satisfied
2 wait for workers to complete by the driver thread did not account non-group events
  that could be left unprocessed and thus to mix up the last executed
  binlog group's file and position:
  the file remained old and the position related to the new rotated file
3 incorrect 'slave reached file:pos' by the parallel slave report in the error log
4 relay log UNTIL missed out the parallel slave branch in
  is_until_satisfied.

The patch addresses all of them to simplify logics of log change
notification in either the master and relay-log until case.
P.1 is addressed with passing the event into is_until_satisfied()
for proper analisis by the function.
P.2 is fixed by changes in handle_queued_pos_update().
P.4 required removing relay-log change notification by workers.
Instead the driver thread updates the notion of the current relay-log
fully itself with aid of introduced
bool Relay_log_info::until_relay_log_names_defer.

An extra print out of the requested until file:pos is arranged
with --log-warning=3.

1 files changed, 58 insertions, 7 deletions

diff --git a/sql/rpl_rli.cc b/sql/rpl_rli.cc
index 3e1bb28f701..7215cdd4b96 100644
--- a/sql/rpl_rli.cc
+++ b/sql/rpl_rli.cc
@@ -60,6 +60,7 @@ Relay_log_info::Relay_log_info(bool is_slave_recovery)
    slave_running(MYSQL_SLAVE_NOT_RUN), until_condition(UNTIL_NONE),
    until_log_pos(0), retried_trans(0), executed_entries(0),
    sql_delay(0), sql_delay_end(0),
+   until_relay_log_names_defer(false),
    m_flags(0)
 {
   DBUG_ENTER("Relay_log_info::Relay_log_info");
@@ -499,6 +500,8 @@ void Relay_log_info::clear_until_condition()
   until_condition= Relay_log_info::UNTIL_NONE;
   until_log_name[0]= 0;
   until_log_pos= 0;
+  until_relay_log_names_defer= false;
+
   DBUG_VOID_RETURN;
 }
 
@@ -989,7 +992,6 @@ void Relay_log_info::inc_group_relay_log_pos(ulonglong log_pos,
     {
       group_relay_log_pos= rgi->future_event_relay_log_pos;
       strmake_buf(group_relay_log_name, rgi->event_relay_log_name);
-      notify_group_relay_log_name_update();
     } else if (cmp == 0 && group_relay_log_pos < rgi->future_event_relay_log_pos)
       group_relay_log_pos= rgi->future_event_relay_log_pos;
 
@@ -1279,29 +1281,78 @@ err:
      autoincrement or if we have transactions).
 
      Should be called ONLY if until_condition != UNTIL_NONE !
+
+     In the parallel execution mode and UNTIL_MASTER_POS the file name is
+     presented by future_event_master_log_name which may be ahead of
+     group_master_log_name. Log_event::log_pos does relate to it nevertheless
+     so the pair comprises a correct binlog coordinate.
+     Internal group events and events that have zero log_pos also
+     produce the zero for the local log_pos which may not lead to the
+     function falsely return true.
+     In UNTIL_RELAY_POS the original caching and notification are simplified
+     to straightforward files comparison when the current event can't be
+     a part of an event group.
+
    RETURN VALUE
      true - condition met or error happened (condition seems to have
             bad log file name)
      false - condition not met
 */
 
-bool Relay_log_info::is_until_satisfied(my_off_t master_beg_pos)
+bool Relay_log_info::is_until_satisfied(Log_event *ev)
 {
   const char *log_name;
   ulonglong log_pos;
+  /* Prevents stopping within transaction; needed solely for Relay UNTIL. */
+  bool in_trans= false;
+
   DBUG_ENTER("Relay_log_info::is_until_satisfied");
 
   if (until_condition == UNTIL_MASTER_POS)
   {
     log_name= (mi->using_parallel() ? future_event_master_log_name
                                     : group_master_log_name);
-    log_pos= master_beg_pos;
+    log_pos= (get_flag(Relay_log_info::IN_TRANSACTION) || !ev || !ev->log_pos) ?
+      (mi->using_parallel() ? 0 : group_master_log_pos) :
+      ev->log_pos - ev->data_written;
   }
   else
   {
     DBUG_ASSERT(until_condition == UNTIL_RELAY_POS);
-    log_name= group_relay_log_name;
-    log_pos= group_relay_log_pos;
+    if (!mi->using_parallel())
+    {
+      log_name= group_relay_log_name;
+      log_pos= group_relay_log_pos;
+    }
+    else
+    {
+      log_name= event_relay_log_name;
+      log_pos=  event_relay_log_pos;
+      in_trans= get_flag(Relay_log_info::IN_TRANSACTION);
+      /*
+        until_log_names_cmp_result is set to UNKNOWN either
+        -  by a non-group event *and* only when it is in the middle of a group
+        -  or by a group event when the preceding group made the above
+           non-group event to defer the resetting.
+      */
+      if ((ev && !Log_event::is_group_event(ev->get_type_code())))
+      {
+        if (in_trans)
+        {
+          until_relay_log_names_defer= true;
+        }
+        else
+        {
+          until_log_names_cmp_result= UNTIL_LOG_NAMES_CMP_UNKNOWN;
+          until_relay_log_names_defer= false;
+        }
+      }
+      else if (!in_trans && until_relay_log_names_defer)
+      {
+        until_log_names_cmp_result= UNTIL_LOG_NAMES_CMP_UNKNOWN;
+        until_relay_log_names_defer= false;
+      }
+    }
   }
 
   DBUG_PRINT("info", ("group_master_log_name='%s', group_master_log_pos=%llu",
@@ -1355,8 +1406,8 @@ bool Relay_log_info::is_until_satisfied(my_off_t master_beg_pos)
   }
 
   DBUG_RETURN(((until_log_names_cmp_result == UNTIL_LOG_NAMES_CMP_EQUAL &&
-           log_pos >= until_log_pos) ||
-          until_log_names_cmp_result == UNTIL_LOG_NAMES_CMP_GREATER));
+                (log_pos >= until_log_pos && !in_trans)) ||
+               until_log_names_cmp_result == UNTIL_LOG_NAMES_CMP_GREATER));
 }