BUG#33666652: Server may decompress entire Transaction_payload_log_event in memory

* Background

Binary log compression is implemented by compressing many events into
one.  There were decompressing iterators over the file, which yielded
the events of the binary log one at a time, including the decompressed
events.

* Problem

When the iterators reached a Transaction_payload_log_event, they
decompressed the full payload, even if only one event at a time was
yielded to the caller. This is wasteful.

* Solution

- The decompressor was mixing memory management with decompressor
  glue.  Separated the responsibilities so the decompressor only is
  compressor glue, and the new class Growable_buffer manages memory
  for a growable buffer.

- The decompressor APIs were implemented as "iterators". However, the
  implementation violated a common iterator idiom by holding an error
  state.  This made it necessary to keep the iterator object after it
  had reached the end iteraton in order to check for the error, which
  maked it unusable for e.g. range-based for loops. So, changed to
  using a "stream" instead, where it is more natural to check for
  errors using the typical idioms.

- Made the stream classes decompress only one event at a time.

- Removed the abstract base class common to compressors and
  decompressors.

- Changed the Decompressor API to better match what decompressor
  libraries provide and what API clients need.  Now the use
  pattern is:
  - Call 'feed' to provide input to decompress
  - Call 'decompress' produce a given number of decompressed bytes.
  - Repeat the above as many times as needed.
  - The above process can be aborted at any time (e.g. after error)
    by calling 'reset'.
  - The caller doesn't need to track how much is left to compress; it
    only needs to check whether the compress operation failed.

- Removed use of ZSTD's "advanced" decompression API.  We were not
  using the features it provided, and some platforms were using an older
  version where this API was unstable.

- Made all instantiations of template class Basic_binlog_file_reader
  have a common base class, IBasic_binlog_file_reader.  Before, the
  old "Iterator" class had to perform a reinterpret_cast (in
  binlog.cc:show_binlog_events) in order to reuse the same code for
  Relaylog_ifile and Binlog_ifile.  This is dangerous and we should
  never use such casts; it would lead to crashes if the types are not
  binary compatible.  Now, the new "stream" class just uses a pointer
  to the base class and resolves whether the object is a
  Relaylog_ifile or a Binlog_ifile using polymorphism.

Change-Id: I1e4872597fb6b37bda03682eadd058dd59b21a60

Author: Sven Sandberg

client/mysqlbinlog.cc

@@ -31,7 +31,6 @@
    of the log; if it is the 3rd then there is this combination:
    Format_desc_of_slave, Rotate_of_master, Format_desc_of_master.
 */
-
 #include "client/mysqlbinlog.h"
 
 #include <fcntl.h>
@@ -43,14 +42,15 @@
 #include <time.h>
 #include <algorithm>
 #include <map>
+#include <sstream>
 #include <utility>
 
 #include "caching_sha2_passwordopt-vars.h"
 #include "client/client_priv.h"
 #include "compression.h"
 #include "libbinlogevents/include/codecs/factory.h"
 #include "libbinlogevents/include/compression/factory.h"
-#include "libbinlogevents/include/compression/iterator.h"
+#include "libbinlogevents/include/compression/payload_event_buffer_istream.h"
 #include "libbinlogevents/include/trx_boundary_parser.h"
 #include "my_byteorder.h"
 #include "my_dbug.h"
@@ -61,6 +61,7 @@
 #include "my_time.h"
 #include "prealloced_array.h"
 #include "print_version.h"
+#include "scope_guard.h"
 #include "sql/binlog_reader.h"
 #include "sql/log_event.h"
 #include "sql/my_decimal.h"
@@ -136,7 +137,6 @@ class Database_rewrite {
     Rewrite_payload_result rewrite_inner_events(
         binary_log::transaction::compression::type compression_type,
         const char *orig_payload, std::size_t orig_payload_size,
-        std::size_t orig_payload_uncompressed_size,
         const binary_log::Format_description_event &fde) {
       // to return error or not
       auto err{false};
@@ -152,10 +152,17 @@ class Database_rewrite {
       // RAII objects
       Buffer_realloc_manager ibuffer_dealloc_guard(&ibuffer);
 
-      // iterator to decompress events
-      binary_log::transaction::compression::Iterable_buffer it(
-          orig_payload, orig_payload_size, orig_payload_uncompressed_size,
-          compression_type);
+      // stream to decompress events
+      using Buffer_istream_t =
+          binary_log::transaction::compression::Payload_event_buffer_istream;
+      using Buffer_ptr_t = Buffer_istream_t::Buffer_ptr_t;
+
+      Buffer_istream_t istream(
+          reinterpret_cast<const unsigned char *>(orig_payload),
+          orig_payload_size, compression_type);
+      Buffer_ptr_t buffer_ptr;
+
+      // compressor to compress again
       using Compress_status_t =
           binary_log::transaction::compression::Compress_status;
       using Managed_buffer_sequence_t =
@@ -167,20 +174,21 @@ class Database_rewrite {
               compression_type);
 
       // rewrite and compress
-      for (auto ptr : it) {
-        std::size_t ev_len{uint4korr(ptr + EVENT_LEN_OFFSET)};
+      while (istream >> buffer_ptr) {
+        /// @todo: don't copy, just use the Decompressor's managed buffer
 
         // reserve input buffer size (we are modifying the input buffer contents
         // before compressing it back).
         std::tie(ibuffer, ibuffer_capacity, err) =
-            reserve(ibuffer, ibuffer_capacity, ev_len);
+            reserve(ibuffer, ibuffer_capacity, buffer_ptr->size());
         if (err) return error_val;
-        memcpy(ibuffer, ptr, ev_len);
+        memcpy(ibuffer, buffer_ptr->data(), buffer_ptr->size());
 
         // rewrite the database name if needed
+        std::size_t ev_len = 0;
         std::tie(ibuffer, ibuffer_capacity, ev_len, err) =
-            m_event_rewriter.rewrite_event(ibuffer, ibuffer_capacity, ev_len,
-                                           fde);
+            m_event_rewriter.rewrite_event(ibuffer, ibuffer_capacity,
+                                           buffer_ptr->size(), fde);
         if (err) return error_val;
 
         compressor->feed(ibuffer, ev_len);
@@ -189,6 +197,10 @@ class Database_rewrite {
           return error_val;
         obuffer_size_uncompressed += ev_len;
       }
+      if (istream.has_error()) {
+        error("%s", istream.get_error_str().c_str());
+        return error_val;
+      }
 
       if (compressor->finish(managed_buffer_sequence) !=
           Compress_status_t::success)
@@ -230,7 +242,6 @@ class Database_rewrite {
 
       auto orig_payload{tpe.get_payload()};
       auto orig_payload_size{tpe.get_payload_size()};
-      auto orig_payload_uncompressed_size{tpe.get_uncompressed_size()};
       auto orig_payload_compression_type{tpe.get_compression_type()};
 
       unsigned char *rewritten_payload{nullptr};
@@ -247,8 +258,7 @@ class Database_rewrite {
                rewritten_payload_size, rewritten_payload_uncompressed_size,
                rewrite_payload_res) =
           rewrite_inner_events(orig_payload_compression_type, orig_payload,
-                               orig_payload_size,
-                               orig_payload_uncompressed_size, fde);
+                               orig_payload_size, fde);
 
       if (rewrite_payload_res) return Rewrite_result{nullptr, 0, 0, true};
 
@@ -2730,12 +2740,12 @@ static Exit_status dump_remote_log_entries(PRINT_EVENT_INFO *print_event_info,
 
     if (!raw_mode || (type == binary_log::ROTATE_EVENT) ||
         (type == binary_log::FORMAT_DESCRIPTION_EVENT)) {
-      Binlog_read_error read_error = binlog_event_deserialize(
+      Binlog_read_error read_status = binlog_event_deserialize(
           reinterpret_cast<unsigned char *>(event_buf), event_len,
           &glob_description_event, opt_verify_binlog_checksum, &ev);
-
-      if (read_error.has_error()) {
-        error("Could not construct log event object: %s", read_error.get_str());
+      if (read_status.has_error()) {
+        error("Could not construct log event object: %s",
+              read_status.get_str());
         my_free(event_buf);
         return ERROR_STOP;
       }
@@ -3070,10 +3080,10 @@ static Exit_status dump_local_log_entries(PRINT_EVENT_INFO *print_event_info,
       // truncated: this may occur when the file is concurrently
       // written by mysqld.
       auto fde_flags =
-          mysqlbinlog_file_reader.format_description_event()->header()->flags;
+          mysqlbinlog_file_reader.format_description_event().header()->flags;
       auto in_use_flag = fde_flags & LOG_EVENT_BINLOG_IN_USE_F;
       if (in_use_flag != 0 && error_type == Binlog_read_error::TRUNC_EVENT) {
-        fprintf(result_file, "# File ends with a truncated event.\n");
+        warning("File ends with a truncated event.");
         return OK_CONTINUE;
       }
 
@@ -3395,61 +3405,70 @@ void Transaction_payload_log_event::print(FILE *,
   Format_description_event fde_no_crc = glob_description_event;
   fde_no_crc.footer()->checksum_alg = binary_log::BINLOG_CHECKSUM_ALG_OFF;
 
-  bool error{false};
   IO_CACHE *const head = &info->head_cache;
   size_t current_buffer_size = 1024;
-  auto buffer = (uchar *)my_malloc(PSI_NOT_INSTRUMENTED, current_buffer_size,
-                                   MYF(MY_WME));
+  auto *buffer = static_cast<uchar *>(
+      my_malloc(PSI_NOT_INSTRUMENTED, current_buffer_size, MYF(MY_WME)));
+  if (buffer == nullptr) {
+    head->error = -1;
+    error("Out of memory.");
+    return;
+  }
+  Scope_guard free_buffer_guard([&] { my_free(buffer); });
   if (!info->short_form) {
     std::ostringstream oss;
     oss << "\tTransaction_Payload\t" << to_string() << std::endl;
-    oss << "# Start of compressed events!" << std::endl;
+    oss << "# Start of compressed events." << std::endl;
     print_header(head, info, false);
     my_b_printf(head, "%s", oss.str().c_str());
   }
 
   // print the payload
-  binary_log::transaction::compression::Iterable_buffer it(
-      m_payload, m_payload_size, m_uncompressed_size, m_compression_type);
-
-  for (auto ptr : it) {
+  using Buffer_istream_t =
+      binary_log::transaction::compression::Payload_event_buffer_istream;
+  Buffer_istream_t istream(*this);
+  Buffer_istream_t::Buffer_ptr_t original_event_buffer;
+  while (istream >> original_event_buffer) {
     Log_event *ev = nullptr;
     bool is_deferred_event = false;
 
     // fix the checksum part
-    size_t event_len = uint4korr(ptr + EVENT_LEN_OFFSET);
-
-    // resize the buffer we are using to handle the event if needed
-    if (event_len > current_buffer_size) {
+    size_t event_len = original_event_buffer->size();
+
+    // Resize the buffer we are using to handle the event if needed.
+    //
+    // The condition `buffer==nullptr` is redundant, because if buffer
+    // is null, then current_buffer_size is 0, and event_len is
+    // guaranteed to be greater than 0 when `operator>>` completed
+    // without taking the stream to an error state.  But clang-tidy
+    // doesn't know that event_len is guaranteed to be greater than
+    // zero, and reports a possible memory leak.
+    if (buffer == nullptr || event_len > current_buffer_size) {
       current_buffer_size =
           round(((event_len + BINLOG_CHECKSUM_LEN) / 1024.0) + 1) * 1024;
-      buffer = (uchar *)my_realloc(PSI_NOT_INSTRUMENTED, buffer,
-                                   current_buffer_size, MYF(0));
-
-      /* purecov: begin inspected */
-      if (!buffer) {
-        // OOM
+      auto *new_buffer = static_cast<uchar *>(my_realloc(
+          PSI_NOT_INSTRUMENTED, buffer, current_buffer_size, MYF(0)));
+      if (new_buffer == nullptr) {
         head->error = -1;
-        my_b_printf(head, "# Out of memory!");
-        goto end;
+        error("Out of memory.");
+        return;
       }
-      /* purecov: end */
+      buffer = new_buffer;
     }
 
-    memcpy(buffer, ptr, event_len);
+    memcpy(buffer, original_event_buffer->data(), event_len);
 
     // rewrite the database name if needed
-    std::tie(buffer, current_buffer_size, event_len, error) =
+    bool rewrite_error{false};
+    std::tie(buffer, current_buffer_size, event_len, rewrite_error) =
         global_database_rewriter.rewrite(buffer, current_buffer_size, event_len,
                                          fde_no_crc);
 
-    /* purecov: begin inspected */
-    if (error) {
+    if (rewrite_error) {
       head->error = -1;
-      my_b_printf(head, "# Error while rewriting db for compressed events!");
-      goto end;
+      error("Error rewriting db for compressed events.");
+      return;
     }
-    /* purecov: end */
 
     // update the CRC
     if (has_crc) {
@@ -3459,14 +3478,13 @@ void Transaction_payload_log_event::print(FILE *,
     }
 
     // now deserialize the event
-    if (binlog_event_deserialize((const unsigned char *)buffer, event_len,
-                                 &glob_description_event, true, &ev)) {
-      /* purecov: begin inspected */
+    Binlog_read_error read_error =
+        binlog_event_deserialize((const unsigned char *)buffer, event_len,
+                                 &glob_description_event, true, &ev);
+    if (read_error.has_error()) {
       head->error = -1;
-      my_b_printf(
-          head, "# Error while handling compressed events! Corrupted binlog?");
-      goto end;
-      /* purecov: end */
+      error("Error decoding Payload_log_event: %s.", read_error.get_str());
+      return;
     }
 
     switch (ev->get_type_code()) {
@@ -3498,10 +3516,12 @@ void Transaction_payload_log_event::print(FILE *,
       current_buffer_size = 0; /* purecov: inspected */
     }
   }
+  if (istream.has_error()) {
+    error("%s", istream.get_error_str().c_str());
+    head->error = -1;
+    return;
+  }
 
-  if (!info->short_form) my_b_printf(head, "# End of compressed events!\n");
-
-end:
-  my_free(buffer);
+  if (!info->short_form) my_b_printf(head, "# End of compressed events.\n");
 }
 #endif