Add LZ4 compression for cloud backups

This adds LZ4 compression support for `barman-cloud-backup` and
`barman-cloud-wal-archive` commands, providing another compression
alternative to existing options.

Core implementation:
- Add `LZ4Compressor` class with streaming compression/decompression
- Add `flush()` method to `ChunkedCompressor` for proper frame termination
- Update `CloudTarUploader.close()` to flush compressor before finalization
- Add --lz4 CLI argument to barman-cloud-backup command
- Add .lz4 file extension handling for backup tar files
- Add tar.lz4 extension parsing in backup catalog

This feature was originally suggested and implemented by @vietcgi
in the PR #1144.

Closes #1144

References: BAR-1013

Signed-off-by: Barbara Leidens <barbara.leidens@enterprisedb.com>

Author: vietcgi

barman/clients/cloud_backup.py

@@ -308,6 +308,13 @@ def parse_arguments(args=None):
         const="snappy",
         dest="compression",
     )
+    compression.add_argument(
+        "--lz4",
+        help="lz4-compress the backup while uploading to the cloud",
+        action="store_const",
+        const="lz4",
+        dest="compression",
+    )
     parser.add_argument(
         "-h",
         "--host",

barman/clients/cloud_compression.py

@@ -19,7 +19,11 @@
 
 from abc import ABCMeta, abstractmethod
 
-from barman.compression import _try_import_snappy, get_internal_compressor
+from barman.compression import (
+    _try_import_lz4,
+    _try_import_snappy,
+    get_internal_compressor,
+)
 from barman.utils import with_metaclass
 
 
@@ -49,6 +53,19 @@ def decompress(self, data):
         :rtype: bytes
         """
 
+    def flush(self):
+        """
+        Flushes any remaining compressed data and returns the final bytes.
+
+        This method should be called after all data has been compressed with
+        add_chunk() to ensure any buffered data and end markers are written.
+        The default implementation returns an empty bytes object.
+
+        :return: Any remaining compressed data
+        :rtype: bytes
+        """
+        return b""
+
 
 class SnappyCompressor(ChunkedCompressor):
     """
@@ -82,20 +99,89 @@ def decompress(self, data):
         return self.decompressor.decompress(data)
 
 
+class Lz4Compressor(ChunkedCompressor):
+    """
+    A ChunkedCompressor implementation based on lz4.
+
+    Uses lz4.frame for streaming compression and decompression. The compressor
+    maintains state across add_chunk() calls and requires flush() to be called
+    at the end to write the frame end marker.
+    """
+
+    def __init__(self):
+        lz4 = _try_import_lz4()
+        self._lz4_frame = lz4.frame
+        self._compressor = None
+        self._decompressor = None
+        self._started = False
+        self._flushed = False
+
+    def add_chunk(self, data):
+        """
+        Compresses the supplied data and returns the compressed bytes.
+
+        On the first call, this initializes the lz4 frame and writes the header.
+        Subsequent calls compress additional data within the same frame.
+
+        :param bytes data: The chunk of data to be compressed
+        :return: The compressed data
+        :rtype: bytes
+        """
+        if self._compressor is None:
+            self._compressor = self._lz4_frame.LZ4FrameCompressor(auto_flush=True)
+
+        if not self._started:
+            self._started = True
+            return self._compressor.begin() + self._compressor.compress(data)
+        return self._compressor.compress(data)
+
+    def decompress(self, data):
+        """
+        Decompresses the supplied chunk of data and returns the uncompressed data.
+
+        The LZ4FrameDecompressor handles streaming decompression and buffering
+        of partial frames automatically.
+
+        :param bytes data: The chunk of data to be decompressed
+        :return: The decompressed data
+        :rtype: bytes
+        """
+        if self._decompressor is None:
+            self._decompressor = self._lz4_frame.LZ4FrameDecompressor()
+        return self._decompressor.decompress(data)
+
+    def flush(self):
+        """
+        Flushes any remaining data and returns the frame end marker.
+
+        This must be called after all data has been compressed to ensure the
+        lz4 frame is properly terminated. Subsequent calls return empty bytes.
+
+        :return: The frame end marker bytes
+        :rtype: bytes
+        """
+        if self._compressor is not None and self._started and not self._flushed:
+            self._flushed = True
+            return self._compressor.flush()
+        return b""
+
+
 def get_compressor(compression):
     """
     Helper function which returns a ChunkedCompressor for the specified compression
-    algorithm. Currently only snappy is supported. The other compression algorithms
+    algorithm. Snappy and lz4 are supported. The other compression algorithms
     supported by barman cloud use the decompression built into TarFile.
 
-    :param str compression: The compression algorithm to use. Can be set to snappy
-      or any compression supported by the TarFile mode string.
+    :param str compression: The compression algorithm to use. Can be set to snappy,
+      lz4, or any compression supported by the TarFile mode string.
     :return: A ChunkedCompressor capable of compressing and decompressing using the
       specified compression.
     :rtype: ChunkedCompressor
     """
     if compression == "snappy":
         return SnappyCompressor()
+    if compression == "lz4":
+        return Lz4Compressor()
     return None
 
 
@@ -107,12 +193,13 @@ def get_streaming_tar_mode(mode, compression):
     ignored so that barman-cloud can apply them itself.
 
     :param str mode: The file mode to use, either r or w.
-    :param str compression: The compression algorithm to use. Can be set to snappy
-      or any compression supported by the TarFile mode string.
+    :param str compression: The compression algorithm to use. Can be set to snappy,
+      lz4, or any compression supported by the TarFile mode string.
     :return: The full filemode for a streaming tar file
     :rtype: str
     """
-    if compression == "snappy" or compression is None:
+    # Compression algorithms that require manual handling (not built into TarFile)
+    if compression in ("snappy", "lz4") or compression is None:
         return "%s|" % mode
     else:
         return "%s|%s" % (mode, compression)

barman/cloud.py

@@ -311,6 +311,14 @@ def flush(self):
     def close(self):
         if self.tar:
             self.tar.close()
+        # Flush any remaining compressed data (e.g., lz4 frame end marker)
+        if self.compressor:
+            final_bytes = self.compressor.flush()
+            if final_bytes:
+                if not self.buffer:
+                    self.buffer = self._buffer()
+                self.buffer.write(final_bytes)
+                self.size += len(final_bytes)
         self.flush()
         self.cloud_interface.async_complete_multipart_upload(
             upload_metadata=self.upload_metadata,
@@ -401,6 +409,8 @@ def _build_dest_name(self, name, count=0):
             components.append(".bz2")
         elif self.compression == "snappy":
             components.append(".snappy")
+        elif self.compression == "lz4":
+            components.append(".lz4")
         return "".join(components)
 
     def _get_tar(self, name):
@@ -2384,6 +2394,8 @@ def get_backup_files(self, backup_info, allow_missing=False):
                         info.compression = "bzip2"
                     elif ext == "tar.snappy":
                         info.compression = "snappy"
+                    elif ext == "tar.lz4":
+                        info.compression = "lz4"
                     else:
                         _logger.warning("Skipping unknown extension: %s", ext)
                         continue