PXB-3747 : Report uncompressed_backup_size for --compress backups

https://perconadev.atlassian.net/browse/PXB-3747

Problem
-------
For --compress backups, backup_size is the compressed size (it comes
from the leaf).  Operators also need the logical, pre-compression size
to answer "how much real data did this backup represent?" and to derive
the effective compression ratio from a single xtrabackup_info record.

Design
------
Why two data pipelines exist

xtrabackup maintains two top-level data pipelines on the backup side
because not every file benefits from compression:

  * ds_data              -- normal InnoDB tablespaces, redo, metadata.
                           Compressed when --compress is on.
  * ds_uncompressed_data -- server-encrypted InnoDB tablespaces
                           ("do not compress encrypted tablespaces",
                           xtrabackup.cc) and RocksDB SST files
                           (already compressed internally).  Bypasses
                           the compress wrapper because ciphertext and
                           pre-compressed content are incompressible,
                           so the CPU cost buys nothing.

Both pipelines share their encrypt / xbstream / leaf tail as an
init-time ctxt-sharing optimization:

  ds_data              --> compress --> encrypt --> xbstream --> leaf
                                                                   ^
                                                                   |
  ds_uncompressed_data -----------------> encrypt --> xbstream ----+

Why the counter cannot live on any ctxt

That shared tail makes "attach a counter to the ctxt" broken no matter
which ctxt is picked:

  * Leaf ctxt: in xbstream mode ds_data, ds_redo, ds_meta, and
    ds_uncompressed_data all terminate at the same leaf.  A leaf-ctxt
    counter would sum bytes from every pipeline, and for the
    compressed pipeline those bytes are post-compression -- neither
    the logical total nor the on-disk total.
  * Encrypt / xbstream / buffer ctxts: ds_data and ds_uncompressed_data
    share these too.  Counting there mixes post-compression bytes
    (from ds_data) with raw bytes (from ds_uncompressed_data); the
    sum is meaningless.

The counter must attach where a single logical write unambiguously
enters exactly one pipeline -- on the per-file handle returned by the
top-level ds_open().  Each top-level open creates a fresh ds_file_t,
the caller decides whether that file's bytes count, and
ds_write / ds_write_sparse bump the counter with the length supplied
at the entry, before any wrapper transforms it:

    xtrabackup top-level open
         |  ds_tracked_open(..., xb_get_metrics())
         |                      |
         |                      +-- nullptr when !--compress (no cost)
         v
    ds_file_t { metrics -> xb_backup_metrics }
         |  ds_write(len) / ds_write_sparse(packed_len)
         |     if (file->metrics) metrics->add_uncomp_size(len)
         v
    [ compress / encrypt / buffer / xbstream wrappers ]
         |
         v
    leaf  (already counted as backup_size by the previous commit)

    xb_backup_metrics.get_uncomp_size()
         |
         v
    uncompressed_backup_size  (xtrabackup_info + error log)

Because arming is opt-in and scoped to the top-level opens on
ds_data / ds_redo / ds_meta / ds_uncompressed_data, the hot-path cost
is one atomic add per write on --compress runs; non-compress runs skip
even that.  The same xb_backup_metrics aggregates both pipelines, so
server-encrypted IBDs and RocksDB SSTs are counted exactly once, at
their logical size.

Arming is driven by xb_get_metrics(), which returns a real counter
only when --compress is active, so non-compress runs emit neither the
field nor the log line.

Tests
-----
t/backup_size_compress.sh covers --compress target-dir, --compress
xbstream, --compress + --encrypt, incremental chains, RocksDB,
server-encrypted InnoDB, redo-log encryption, and sparse files.  Every
scenario asserts backup_size matches the compressed on-disk size and,
after decompression, uncompressed_backup_size matches the decompressed
tree size byte-for-byte.

Author: satya-bodapati

storage/innobase/xtrabackup/src/backup_copy.cc

@@ -55,6 +55,7 @@ Place, Suite 330, Boston, MA 02111-1307 USA
 #include <chrono>
 #include <fstream>
 #include <functional>
+#include <iomanip>
 #include <queue>
 #include <set>
 #include <sstream>
@@ -377,7 +378,7 @@ bool backup_file_print(const char *filename, const char *message, int len) {
   stat.st_mtime = time(nullptr);
   stat.st_size = len;
 
-  dstfile = ds_open(ds_data, filename, &stat);
+  dstfile = ds_tracked_open(ds_data, filename, &stat, xb_get_metrics());
   if (dstfile == NULL) {
     xb::error() << "cannot open the destination stream for " << filename;
     goto error;
@@ -576,7 +577,8 @@ bool copy_file(ds_ctxt_t *datasink, const char *src_file_path,
 
   strncpy(dst_name, cursor.rel_path, sizeof(dst_name));
 
-  dstfile = ds_open(datasink, trim_dotslash(dst_file_path), &cursor.statinfo);
+  dstfile = ds_tracked_open(datasink, trim_dotslash(dst_file_path),
+                            &cursor.statinfo, xb_get_metrics());
   if (dstfile == NULL) {
     xb::error() << "cannot open the destination stream for " << dst_name;
     goto error;
@@ -1594,10 +1596,6 @@ bool backup_start(Backup_context &context) {
   return (true);
 }
 
-/** Render a byte count as a human-friendly string (GiB / MiB / KiB /
-bytes) for the backup-complete log line.
-@param[in] bytes  raw byte count
-@return newly-constructed std::string, e.g. "12.34 MiB". */
 static std::string human_readable(unsigned long long bytes) {
   char buf[64];
   if (bytes >= 1ULL << 30)
@@ -1671,14 +1669,36 @@ bool backup_finish(Backup_context &context) {
     unsigned long long backup_size = get_final_backup_size();
 
     /* A successful backup always produces on-disk output, so the leaf
-    counter must be non-zero here.  ut_ad is a no-op in release; warn +
-    skip on release to avoid hiding a silent reporting failure. */
+    counter must be non-zero here.  ut_ad is a no-op in release; warn
+    + skip on release to avoid hiding a silent reporting failure. */
     ut_ad(backup_size > 0);
     if (backup_size == 0) {
       xb::warn() << "Backup size reporting failed: leaf counter returned 0";
     } else {
       xb::info() << "Backup size: " << human_readable(backup_size) << " ("
                  << backup_size << " bytes)";
+
+      if (xtrabackup_compress != XTRABACKUP_COMPRESS_NONE) {
+        unsigned long long uncompressed_backup_size =
+            get_uncompressed_backup_size();
+
+        /* With --compress, at least one data file must flow through a
+        top-level ds_tracked_open() with metrics armed, so the counter
+        cannot be zero. */
+        ut_ad(uncompressed_backup_size > 0);
+        if (uncompressed_backup_size == 0) {
+          xb::warn() << "Uncompressed backup size reporting failed: metrics"
+                        " counter is 0 despite --compress";
+        } else {
+          xb::info() << "Uncompressed backup size: "
+                     << human_readable(uncompressed_backup_size) << " ("
+                     << uncompressed_backup_size << " bytes)";
+
+          double ratio = (double)uncompressed_backup_size / (double)backup_size;
+          xb::info() << "Compression ratio: " << std::fixed
+                     << std::setprecision(2) << ratio << "x";
+        }
+      }
     }
   }
 

storage/innobase/xtrabackup/src/backup_mysql.cc

@@ -627,7 +627,6 @@ bool get_mysql_vars(MYSQL *connection) {
         my_strdup(PSI_NOT_INSTRUMENTED, innodb_directories_var, MYF(MY_FAE));
   }
 
-
   if (!check_if_param_set("innodb_log_file_size") && innodb_log_file_size_var) {
     char *endptr;
 
@@ -1849,12 +1848,7 @@ char *get_xtrabackup_info(MYSQL *connection) {
   format_time(history_start_time, buf_start_time, time_buf_size);
   format_time(history_end_time, buf_end_time, time_buf_size);
 
-  /* Sample the leaf's bytes_written counter now, right before
-  xtrabackup_info is itself written to ds_data.  Everything that was
-  meant to flow through ds_data for this backup is already accounted
-  for at this point (see the reorder in xtrabackup.cc's backup finish
-  flow). */
-  const unsigned long long backup_size = get_final_backup_size();
+  unsigned long long backup_size = get_final_backup_size();
 
   ut_a(uuid);
   ut_a(server_version);
@@ -1910,10 +1904,26 @@ char *get_xtrabackup_info(MYSQL *connection) {
                xtrabackup_encrypt ? "Y" : "N",               /* encrypted */
                ddl_lock_type_to_str(static_cast<lock_ddl_type_t>(opt_lock_ddl))
                    .c_str(), /* lock-ddl */
-               backup_size);
+               backup_size); /* backup_size */
 
   ut_a(ret != 0);
 
+  if (xtrabackup_compress != XTRABACKUP_COMPRESS_NONE) {
+    unsigned long long uncompressed_backup_size =
+        get_uncompressed_backup_size();
+
+    char *tmp = NULL;
+    int ret2 = asprintf(&tmp, "%suncompressed_backup_size = %llu\n", result,
+                        uncompressed_backup_size);
+    if (ret2 < 0) {
+      xb::warn() << "Failed to append uncompressed_backup_size to"
+                 << " xtrabackup_info: " << strerror(errno);
+    } else {
+      free(result);
+      result = tmp;
+    }
+  }
+
   free(server_version);
   return result;
 }

storage/innobase/xtrabackup/src/datasink.cc

@@ -20,6 +20,9 @@ Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA  02110-1301, USA
 
 #include "datasink.h"
 #include <my_base.h>
+#include <atomic>
+#include <cassert>
+#include <cstdint>
 #include "common.h"
 #include "ds_buffer.h"
 #include "ds_compress.h"
@@ -37,8 +40,17 @@ Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA  02110-1301, USA
 #include "ds_xbstream.h"
 #include "msg.h"
 
-/************************************************************************
-Create a datasink of the specified type */
+/** Global aggregate metrics for the backup pipelines.  Incremented by
+ds_write() / ds_write_sparse() whenever the file's metrics pointer is
+non-null.  Armed at top-level backup open sites via ds_tracked_open()
+with xb_get_metrics() (xtrabackup.h), which returns &xb_backup_metrics
+when --compress is active and nullptr otherwise. */
+xb_metrics xb_backup_metrics;
+
+/** See datasink.h for contract.
+@param[in] root  root path / destination the datasink writes into
+@param[in] type  which datasink to instantiate
+@return a datasink context, or nullptr for unknown type. */
 ds_ctxt_t *ds_create(const char *root, ds_type_t type) {
   datasink_t *ds;
   ds_ctxt_t *ctxt;
@@ -95,13 +107,7 @@ ds_ctxt_t *ds_create(const char *root, ds_type_t type) {
   ctxt = ds->init(root);
   if (ctxt != NULL) {
     ctxt->datasink = ds;
-    /* Per-datasink init() routines use my_malloc() which does not zero
-    memory, so fields not explicitly assigned by init() (notably
-    pipe_ctxt) can hold garbage.  Normalize them here so every caller
-    sees a clean ds_ctxt_t regardless of which leaf/wrapper produced
-    it.  In particular ds_leaf() relies on pipe_ctxt == nullptr at the
-    terminal node to stop walking. */
-    ctxt->pipe_ctxt = nullptr;
+    ctxt->pipe_ctxt = NULL;
   } else {
     msg("Error: failed to initialize datasink.\n");
     exit(EXIT_FAILURE);
@@ -110,61 +116,101 @@ ds_ctxt_t *ds_create(const char *root, ds_type_t type) {
   return ctxt;
 }
 
-/************************************************************************
-Open a datasink file */
+/** Pure dispatcher.  Each *_open() initializes the framework-owned
+fields on its freshly allocated ds_file_t via ds_init_file() (see
+datasink.h).  The debug assertion catches any *_open implementation
+that forgets to call ds_init_file(): ds_write / ds_close would later
+chase a NULL datasink/ctxt and crash with no clue about the root
+cause.
+@param[in] ctxt  pipeline to open through
+@param[in] path  path relative to the pipeline root
+@param[in] stat  size/mode hints for downstream datasinks
+@return newly opened file, or nullptr on error. */
 ds_file_t *ds_open(ds_ctxt_t *ctxt, const char *path, MY_STAT *stat) {
-  ds_file_t *file;
-
-  file = ctxt->datasink->open(ctxt, path, stat);
-  if (file != NULL) {
-    file->datasink = ctxt->datasink;
-  }
+  ds_file_t *file = ctxt->datasink->open(ctxt, path, stat);
+  assert(file == nullptr ||
+         (file->datasink != nullptr && file->ctxt != nullptr));
+  return file;
+}
 
+/** ds_open() + ds_track_metrics(), in one call.  See the declaration
+in datasink.h for when to prefer this over a raw ds_open().
+@param[in]     ctxt     pipeline to open through
+@param[in]     path     path relative to the pipeline root
+@param[in]     stat     size/mode hints for downstream datasinks
+@param[in,out] metrics  metrics instance to bind; nullptr disables tracking
+@return newly opened file, or nullptr on error. */
+ds_file_t *ds_tracked_open(ds_ctxt_t *ctxt, const char *path, MY_STAT *stat,
+                           xb_metrics *metrics) {
+  ds_file_t *file = ds_open(ctxt, path, stat);
+  ds_track_metrics(file, metrics);
   return file;
 }
 
-/************************************************************************
-Write to a datasink file.
+/** Write a contiguous buffer through the owning datasink's write slot.
+Adds @p len to file->metrics when tracking is bound.
+@param[in,out] file  ds_file_t previously returned by ds_open
+@param[in]     buf   bytes to write
+@param[in]     len   number of bytes at @p buf
 @return 0 on success, 1 on error. */
 int ds_write(ds_file_t *file, const void *buf, size_t len) {
+  if (file->metrics != nullptr) {
+    file->metrics->add_uncomp_size(len);
+  }
   return file->datasink->write(file, buf, len);
 }
 
-/************************************************************************
-Check if sparse files are supported.
-@return 1 if yes. */
+/** Check whether @p file's datasink supports sparse writes.
+@param[in] file  file to probe
+@return 1 if write_sparse is implemented, 0 otherwise. */
 int ds_is_sparse_write_supported(ds_file_t *file) {
   if (file->datasink->write_sparse != nullptr) {
     return 1;
   }
   return 0;
 }
 
-/************************************************************************
-Write sparse chunk if supported.
+/** Forward a sparse chunk through the owning datasink's write_sparse
+slot.  Adds the sum of sparse_map[i].len (the packed payload, holes
+excluded) to file->metrics when tracking is bound.
+@param[in,out] file                  ds_file_t previously returned by ds_open
+@param[in]     buf                   packed buffer containing the data bytes
+@param[in]     len                   size of @p buf
+@param[in]     sparse_map_size       number of entries in @p sparse_map
+@param[in]     sparse_map            per-chunk (skip, len)
+@param[in]     punch_hole_supported  true if the destination filesystem
+                                     can physically punch holes
 @return 0 on success, 1 on error. */
 int ds_write_sparse(ds_file_t *file, const void *buf, size_t len,
                     size_t sparse_map_size, const ds_sparse_chunk_t *sparse_map,
                     bool punch_hole_supported) {
   if (file->datasink->write_sparse != nullptr) {
+    /* Count the packed payload only: holes do not occupy disk. */
+    if (file->metrics != nullptr) {
+      size_t packed_len = 0;
+      for (size_t i = 0; i < sparse_map_size; i++) {
+        packed_len += sparse_map[i].len;
+      }
+      file->metrics->add_uncomp_size(packed_len);
+    }
     return file->datasink->write_sparse(file, buf, len, sparse_map_size,
                                         sparse_map, punch_hole_supported);
   }
   return 1;
 }
 
-/************************************************************************
-Close a datasink file.
-@return 0 on success, 1, on error. */
+/** Close a datasink file.
+@param[in,out] file  ds_file_t previously returned by ds_open
+@return 0 on success, 1 on error. */
 int ds_close(ds_file_t *file) { return file->datasink->close(file); }
 
-/************************************************************************
-Destroy a datasink handle */
+/** Destroy a datasink handle.
+@param[in,out] ctxt  datasink ctxt returned by ds_create / ds_set_pipe */
 void ds_destroy(ds_ctxt_t *ctxt) { ctxt->datasink->deinit(ctxt); }
 
-/************************************************************************
-Set the destination pipe for a datasink (only makes sense for compress and
-tmpfile). */
+/** Wire @p ctxt's output into @p pipe_ctxt.
+@param[in,out] ctxt       wrapper datasink context
+@param[in,out] pipe_ctxt  next-stage datasink the wrapper writes into */
 void ds_set_pipe(ds_ctxt_t *ctxt, ds_ctxt_t *pipe_ctxt) {
   ctxt->pipe_ctxt = pipe_ctxt;
 }