MDEV-16372 ER_BASE64_DECODE_ERROR upon replaying binary log via mysqlbinlog --verbose

(This commit is for 10.3 and upper branches)

In case of a pattern of non-STMT_END-marked Rows-log-event (A) followed by
a STMT_END marked one (B) mysqlbinlog mixes up the base64 encoded rows events
with their pseudo sql representation produced by the verbose option:
      BINLOG '
        base64 encoded data for A
        ### verbose section for A
        base64 encoded data for B
        ### verbose section for B
      '/*!*/;
In effect the produced BINLOG '...' query is not valid and is rejected with the error.
Examples of this way malformed BINLOG could have been found in binlog_row_annotate.result
that gets corrected with the patch.

The issue is fixed with introduction an auxiliary IO_CACHE to hold on the verbose
comments until the terminal STMT_END event is found. The new cache is emptied
out after two pre-existing ones are done at that time.
The correctly produced output now for the above case is as the following:
      BINLOG '
        base64 encoded data for A
        base64 encoded data for B
      '/*!*/;
        ### verbose section for A
        ### verbose section for B

Thanks to Alexey Midenkov for the problem recognition and attempt to tackle,
and to Venkatesh Duggirala who produced a patch for the upstream whose
idea is exploited here, as well as to MDEV-23077 reporter LukeXwang who
also contributed a piece of a patch aiming at this issue.

1 files changed, 14 insertions, 23 deletions

diff --git a/sql/log_event.cc b/sql/log_event.cc
index 4432a51b010..1e382cab55f 100644
--- a/sql/log_event.cc
+++ b/sql/log_event.cc
@@ -3875,7 +3875,7 @@ bool Log_event::print_base64(IO_CACHE* file,
       ev->need_flashback_review= need_flashback_review;
       if (print_event_info->verbose)
       {
-        if (ev->print_verbose(file, print_event_info))
+        if (ev->print_verbose(&print_event_info->tail_cache, print_event_info))
           goto err;
       }
       else
@@ -3899,22 +3899,9 @@ bool Log_event::print_base64(IO_CACHE* file,
       }
 #else
       if (print_event_info->verbose)
-      {
-        /*
-          Verbose event printout can't start before encoded data
-          got enquoted. This is done at this point though multi-row
-          statement remain vulnerable.
-          TODO: fix MDEV-10362 to remove this workaround.
-        */
-        if (print_event_info->base64_output_mode !=
-            BASE64_OUTPUT_DECODE_ROWS)
-          my_b_printf(file, "'%s\n", print_event_info->delimiter);
-        error= ev->print_verbose(file, print_event_info);
-      }
+        error= ev->print_verbose(&print_event_info->tail_cache, print_event_info);
       else
-      {
         ev->count_row_events(print_event_info);
-      }
 #endif
       delete ev;
       if (unlikely(error))
@@ -11992,7 +11979,7 @@ bool copy_cache_to_file_wrapped(IO_CACHE *body,
                                 FILE *file,
                                 bool do_wrap,
                                 const char *delimiter,
-                                bool is_verbose)
+                                bool is_verbose /*TODO: remove */)
 {
   const my_off_t cache_size= my_b_tell(body);
 
@@ -12025,8 +12012,7 @@ bool copy_cache_to_file_wrapped(IO_CACHE *body,
     my_fprintf(file, fmt_frag, 1);
     if (my_b_copy_to_file(body, file, SIZE_T_MAX))
       goto err;
-    if (!is_verbose)
-      my_fprintf(file, fmt_delim, delimiter);
+    my_fprintf(file, fmt_delim, delimiter);
 
     my_fprintf(file, fmt_binlog2, delimiter);
   }
@@ -12035,8 +12021,7 @@ bool copy_cache_to_file_wrapped(IO_CACHE *body,
     my_fprintf(file, str_binlog);
     if (my_b_copy_to_file(body, file, SIZE_T_MAX))
       goto err;
-    if (!is_verbose)
-      my_fprintf(file, fmt_delim, delimiter);
+    my_fprintf(file, fmt_delim, delimiter);
   }
   reinit_io_cache(body, WRITE_CACHE, 0, FALSE, TRUE);
 
@@ -12122,7 +12107,6 @@ bool copy_cache_to_string_wrapped(IO_CACHE *cache,
       goto err;
     str += (add_to_len= uint32(cache->end_of_file - (cache_size/2 + 1)));
     to->length += add_to_len;
-    if (!is_verbose)
     {
       str += (add_to_len= sprintf(str , fmt_delim, delimiter));
       to->length += add_to_len;
@@ -12138,7 +12122,6 @@ bool copy_cache_to_string_wrapped(IO_CACHE *cache,
       goto err;
     str += cache->end_of_file;
     to->length += (size_t)cache->end_of_file;
-    if (!is_verbose)
       to->length += sprintf(str , fmt_delim, delimiter);
   }
 
@@ -12186,6 +12169,7 @@ bool Rows_log_event::print_helper(FILE *file,
 {
   IO_CACHE *const head= &print_event_info->head_cache;
   IO_CACHE *const body= &print_event_info->body_cache;
+  IO_CACHE *const tail= &print_event_info->tail_cache;
 #ifdef WHEN_FLASHBACK_REVIEW_READY
   IO_CACHE *const sql= &print_event_info->review_sql_cache;
 #endif
@@ -12216,7 +12200,8 @@ bool Rows_log_event::print_helper(FILE *file,
       if (copy_event_cache_to_file_and_reinit(head, file) ||
           copy_cache_to_file_wrapped(body, file, do_print_encoded,
                                      print_event_info->delimiter,
-                                     print_event_info->verbose))
+                                     print_event_info->verbose) ||
+          copy_event_cache_to_file_and_reinit(tail, file))
         goto err;
     }
     else
@@ -12234,6 +12219,11 @@ bool Rows_log_event::print_helper(FILE *file,
       return 1;
     output_buf.append(tmp_str.str, tmp_str.length);
     my_free(tmp_str.str);
+    if (copy_event_cache_to_string_and_reinit(tail, &tmp_str))
+      return 1;
+    output_buf.append(tmp_str.str, tmp_str.length);
+    my_free(tmp_str.str);
+
 #ifdef WHEN_FLASHBACK_REVIEW_READY
     if (copy_event_cache_to_string_and_reinit(sql, &tmp_str))
       return 1;
@@ -15056,6 +15046,7 @@ st_print_event_info::st_print_event_info()
   base64_output_mode=BASE64_OUTPUT_UNSPEC;
   open_cached_file(&head_cache, NULL, NULL, 0, flags);
   open_cached_file(&body_cache, NULL, NULL, 0, flags);
+  open_cached_file(&tail_cache, NULL, NULL, 0, flags);
 #ifdef WHEN_FLASHBACK_REVIEW_READY
   open_cached_file(&review_sql_cache, NULL, NULL, 0, flags);
 #endif