Streaming compression

Summary:
Extend protocol to support streaming compression. This adds new possible values for the slave_compression_lib variable: zstd_stream/lz4f_stream. Also, zstd_net_compression_level and lz4f_net_compression_level can now take on negative values for even faster compression.

This reuses the existing scheme where the client requests the compression algorithm via connection attributes. If the server does not support this algorithm, the server will fall back to zlib and the client may read unexpected data (which means that all servers will have to be upgraded before it is safe for clients to request these new algorithms).

When compression fails to produce data smaller than the original data, we send data uncompressed, and we reset the context on both the compression and decompression side. In theory, we only need to do this when the compressed data is larger than 16MB because we only have 3 bytes to encode the length, but to keep things simple, we'll always reset the context when compression inflates data size.

Status counters are also added to track how frequently resets happen. Compression_context_reset counts number of resets and in combination with Compression_input_bytes, will give average amount of bytes processed before a reset happens. Compression_output_bytes is also added to track compression ratio.

This also fixes a bug in the original implementation where level was not being properly set because the #ifdef used was incorrect.

Reference patch: 3339788

Differential Revision: D19316704

Author: lth

cmake/lz4.cmake

@@ -20,9 +20,11 @@
 # along with this program; if not, write to the Free Software
 # Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301  USA
 
-# cmake -DWITH_LZ4=system|bundled
+# cmake -DWITH_LZ4=system|bundled|3rdparty
 # bundled is the default
 
+SET(LIBLZ4_VERSION_REQUIRED "1.8.0")  # LZ4F_HEADER_SIZE_MAX and LZ4F_freeDecompressionContext
+
 MACRO(FIND_LZ4_VERSION)
   FOREACH(version_part
       LZ4_VERSION_MAJOR
@@ -40,6 +42,12 @@ MACRO(FIND_LZ4_VERSION)
   MESSAGE(STATUS "LZ4_VERSION (${WITH_LZ4}) is ${LZ4_VERSION}")
   MESSAGE(STATUS "LZ4_INCLUDE_DIR ${LZ4_INCLUDE_DIR}")
   MESSAGE(STATUS "LZ4_LIBRARY ${LZ4_LIBRARY}")
+
+  IF (${LZ4_VERSION} VERSION_LESS ${LIBLZ4_VERSION_REQUIRED})
+    MESSAGE(FATAL_ERROR "Required liblz4 ${LIBLZ4_VERSION_REQUIRED} and installed version is ${LZ4_VERSION}")
+  ELSE()
+    MESSAGE(STATUS "Found liblz4 version ${LZ4_VERSION}")
+  ENDIF()
 ENDMACRO()
 
 MACRO (FIND_SYSTEM_LZ4)

cmake/zstd.cmake

@@ -25,7 +25,8 @@
 
 # With earier versions, several compression tests fail.
 # With version < 1.0.0 our source code does not build.
-SET(MIN_ZSTD_VERSION_REQUIRED "1.2.0")
+# ZSTD_reset_session_only requires 1.4.
+SET(MIN_ZSTD_VERSION_REQUIRED "1.4.0")
 
 MACRO (FIND_ZSTD_VERSION)
   FOREACH(version_part

include/compression.h

@@ -35,20 +35,24 @@
 
 #define COMPRESSION_ALGORITHM_ZLIB "zlib"
 #define COMPRESSION_ALGORITHM_ZSTD "zstd"
+#define COMPRESSION_ALGORITHM_ZSTD_STREAM "zstd_stream"
+#define COMPRESSION_ALGORITHM_LZ4F_STREAM "lz4f_stream"
 #define COMPRESSION_ALGORITHM_UNCOMPRESSED "uncompressed"
 #define COMPRESSION_ALGORITHM_NAME_LENGTH_MAX 32
-#define COMPRESSION_ALGORITHM_COUNT_MAX 3
+#define COMPRESSION_ALGORITHM_COUNT_MAX 5
 #define COMPRESSION_ALGORITHM_NAME_BUFFER_SIZE                                 \
   ((COMPRESSION_ALGORITHM_NAME_LENGTH_MAX * COMPRESSION_ALGORITHM_COUNT_MAX) + \
    3)
-#define PROTOCOL_COMPRESSION_DEFAULT_VALUE "zlib,zstd,uncompressed"
+#define PROTOCOL_COMPRESSION_DEFAULT_VALUE \
+  "zlib,zstd,zstd_stream,lz4f_stream,uncompressed"
+
+extern const char *mysql_compression_lib_names[COMPRESSION_ALGORITHM_COUNT_MAX];
 
 constexpr int default_zstd_compression_level = 3;
 
 /* Helper functions to validate compression algorithm and level */
 enum_compression_algorithm get_compression_algorithm(
     std::string name = std::string());
-std::string get_compression_algorithm_name(enum_compression_algorithm);
 void parse_compression_algorithms_list(std::string name,
                                        std::vector<std::string> &list);
 bool is_zstd_compression_level_valid(uint level);

include/my_compress.h

@@ -28,6 +28,8 @@ enum enum_compression_algorithm {
   MYSQL_UNCOMPRESSED = 1,
   MYSQL_ZLIB,
   MYSQL_ZSTD,
+  MYSQL_ZSTD_STREAM,
+  MYSQL_LZ4F_STREAM,
   MYSQL_INVALID
 };
 
@@ -42,6 +44,25 @@ typedef struct mysql_zlib_compress_context {
   unsigned int compression_level;
 } mysql_zlib_compress_context;
 
+typedef struct LZ4F_cctx_s LZ4F_compressionContext;
+typedef struct LZ4F_dctx_s LZ4F_decompressionContext;
+
+/**
+  Compress context information. relating to LZ4 compression.
+*/
+
+typedef struct mysql_lz4f_compress_context {
+  /**
+    Compression level to use in lz4 compression.
+  */
+  LZ4F_compressionContext *cctx;
+  LZ4F_decompressionContext *dctx;
+  bool reset_cctx;
+  unsigned char *compress_buf;
+  unsigned long compress_buf_len;
+  unsigned int compression_level;
+} mysql_lz4f_compress_context;
+
 typedef struct ZSTD_CCtx_s ZSTD_CCtx;
 typedef struct ZSTD_DCtx_s ZSTD_DCtx;
 
@@ -61,6 +82,9 @@ typedef struct mysql_zstd_compress_context {
   /**
     Compression level to use in zstd compression.
   */
+  bool reset_cctx;
+  unsigned char *compress_buf;
+  unsigned long compress_buf_len;
   unsigned int compression_level;
 } mysql_zstd_compress_context;
 
@@ -73,6 +97,7 @@ typedef struct mysql_zstd_compress_context {
 typedef struct mysql_compress_context {
   enum enum_compression_algorithm algorithm;  ///< Compression algorithm name.
   union {
+    mysql_lz4f_compress_context lz4f_ctx;  ///< Context information of lz4.
     mysql_zlib_compress_context zlib_ctx;  ///< Context information of zlib.
     mysql_zstd_compress_context zstd_ctx;  ///< Context information of zstd.
   } u;
@@ -111,4 +136,6 @@ void mysql_compress_context_init(mysql_compress_context *cmp_ctx,
 
 void mysql_compress_context_deinit(mysql_compress_context *mysql_compress_ctx);
 
+void reset_compress_status(void);
+
 #endif  // MY_COMPRESS_INCLUDED

include/my_sys.h

@@ -843,12 +843,11 @@ extern char *strdup_root(MEM_ROOT *root, const char *str);
 extern char *safe_strdup_root(MEM_ROOT *root, const char *str);
 extern char *strmake_root(MEM_ROOT *root, const char *str, size_t len);
 extern void *memdup_root(MEM_ROOT *root, const void *str, size_t len);
-extern bool my_compress(mysql_compress_context *, uchar *, size_t *, size_t *,
-                        uint);
+extern bool my_compress(mysql_compress_context *, uchar *, size_t *, size_t *);
 extern bool my_uncompress(mysql_compress_context *, uchar *, size_t, size_t *);
 extern uchar *my_compress_alloc(mysql_compress_context *comp_ctx,
                                 const uchar *packet, size_t *len,
-                                size_t *complen, uint level);
+                                size_t *complen);
 
 extern uint my_set_max_open_files(uint files);
 

include/mysql.h.pp

@@ -96,21 +96,37 @@
   MYSQL_UNCOMPRESSED = 1,
   MYSQL_ZLIB,
   MYSQL_ZSTD,
+  MYSQL_ZSTD_STREAM,
+  MYSQL_LZ4F_STREAM,
   MYSQL_INVALID
 };
 typedef struct mysql_zlib_compress_context {
   unsigned int compression_level;
 } mysql_zlib_compress_context;
+typedef struct LZ4F_cctx_s LZ4F_compressionContext;
+typedef struct LZ4F_dctx_s LZ4F_decompressionContext;
+typedef struct mysql_lz4f_compress_context {
+  LZ4F_compressionContext *cctx;
+  LZ4F_decompressionContext *dctx;
+  bool reset_cctx;
+  unsigned char *compress_buf;
+  unsigned long compress_buf_len;
+  unsigned int compression_level;
+} mysql_lz4f_compress_context;
 typedef struct ZSTD_CCtx_s ZSTD_CCtx;
 typedef struct ZSTD_DCtx_s ZSTD_DCtx;
 typedef struct mysql_zstd_compress_context {
   ZSTD_CCtx *cctx;
   ZSTD_DCtx *dctx;
+  bool reset_cctx;
+  unsigned char *compress_buf;
+  unsigned long compress_buf_len;
   unsigned int compression_level;
 } mysql_zstd_compress_context;
 typedef struct mysql_compress_context {
   enum enum_compression_algorithm algorithm;
   union {
+    mysql_lz4f_compress_context lz4f_ctx;
     mysql_zlib_compress_context zlib_ctx;
     mysql_zstd_compress_context zstd_ctx;
   } u;
@@ -121,6 +137,7 @@
                                  enum enum_compression_algorithm algorithm,
                                  unsigned int compression_level);
 void mysql_compress_context_deinit(mysql_compress_context *mysql_compress_ctx);
+void reset_compress_status(void);
 enum SERVER_STATUS_flags_enum {
   SERVER_STATUS_IN_TRANS = 1,
   SERVER_STATUS_AUTOCOMMIT = 2,

include/mysql_com.h

@@ -460,7 +460,7 @@
   net_send_ok(), net_send_eof()
 */
 #define CLIENT_TRANSACTIONS 8192
-#define CLIENT_RESERVED 16384 /**< DEPRECATED: Old flag for 4.1 protocol  */
+#define CLIENT_RESERVED 16384
 #define CLIENT_RESERVED2                                 \
   32768 /**< DEPRECATED: Old flag for 4.1 authentication \
            CLIENT_SECURE_CONNECTION */
@@ -782,8 +782,8 @@
    CLIENT_CONNECT_WITH_DB | CLIENT_NO_SCHEMA | CLIENT_COMPRESS | CLIENT_ODBC | \
    CLIENT_LOCAL_FILES | CLIENT_IGNORE_SPACE | CLIENT_PROTOCOL_41 |             \
    CLIENT_INTERACTIVE | CLIENT_SSL | CLIENT_IGNORE_SIGPIPE |                   \
-   CLIENT_TRANSACTIONS | CLIENT_RESERVED | CLIENT_RESERVED2 |                  \
-   CLIENT_MULTI_STATEMENTS | CLIENT_MULTI_RESULTS | CLIENT_PS_MULTI_RESULTS |  \
+   CLIENT_TRANSACTIONS | CLIENT_RESERVED2 | CLIENT_MULTI_STATEMENTS |          \
+   CLIENT_MULTI_RESULTS | CLIENT_PS_MULTI_RESULTS |                            \
    CLIENT_SSL_VERIFY_SERVER_CERT | CLIENT_REMEMBER_OPTIONS |                   \
    CLIENT_PLUGIN_AUTH | CLIENT_CONNECT_ATTRS |                                 \
    CLIENT_PLUGIN_AUTH_LENENC_CLIENT_DATA |                                     \

libmysql/CMakeLists.txt

@@ -246,6 +246,8 @@ ELSE()
   LIST(APPEND LIBS_TO_LINK ${ZLIB_LIBRARY})
 ENDIF()
 
+LIST(APPEND LIBS_TO_LINK ${LZ4_LIBRARY})
+
 IF(WITH_ZSTD STREQUAL "bundled")
   LIST(APPEND LIBS_TO_MERGE ${ZSTD_LIBRARY})
 ELSE()

mysql-test/collections/disabled_rocksdb.def

@@ -51,10 +51,3 @@ rocksdb.index_merge_rocksdb2 : BUG#0000 query plan is different between 8.0 and
 rocksdb.rqg_runtime: BUG#0000 commit without prepare
 rocksdb.rqg_transactions : BUG#0000 commit without prepare
 rocksdb.add_index_inplace : BUG#0000 : ALTER TABLE ALGORITHM=INPLACE fails
-
-# Rebase: not-yet-implemented bugs
-rocksdb.bypass_select_range_pk_bloom : BUG#0000 Rebase: not-yet-implemented
-rocksdb.bypass_select_scenarios_bloom : BUG#0000 Rebase: not-yet-implemented
-rocksdb.verify_tbl_share_primary_idx : BUG#0000 Rebase: not-yet-implemented
-rocksdb.bypass_select_range_sk_bloom : BUG#0000 Rebase: not-yet-implemented
-rocksdb.bypass_select_basic_bloom : BUG#0000 Rebase: not-yet-implemented

mysql-test/r/compress.result

@@ -1,8 +1,12 @@
+FLUSH STATUS;
 SHOW STATUS LIKE 'Compression%';
 Variable_name	Value
 Compression	ON
 Compression_algorithm	zlib
+Compression_context_reset	0
+Compression_input_bytes	0
 Compression_level	6
+Compression_output_bytes	0
 drop table if exists t1,t2,t3,t4;
 CREATE TABLE t1 (
 Period smallint(4) unsigned zerofill DEFAULT '0000' NOT NULL,
@@ -2265,8 +2269,11 @@ drop table t1;
 SHOW STATUS LIKE 'Compression%';
 Variable_name	Value
 Compression	ON
-Compression_algorithm	zstd
-Compression_level	10
+Compression_algorithm	zlib
+Compression_context_reset	0
+Compression_input_bytes	0
+Compression_level	6
+Compression_output_bytes	64457
 drop table if exists t1,t2,t3,t4;
 CREATE TABLE t1 (
 Period smallint(4) unsigned zerofill DEFAULT '0000' NOT NULL,