Add different compression choice

Summary:

Test Plan:

Reviewers:

Subscribers:

Tasks:

Tags:

Author: sunshine-Chun

include/mysql/components/services/clone_protocol_service.h

@@ -61,6 +61,8 @@ struct mysql_clone_ssl_context {
 
   /** Enable network compression. */
   bool m_enable_compression;
+  const char *m_compression_algorithm;
+  uint m_compression_level;
   NET_SERVER *m_server_extn;
 };
 

mysql-test/suite/rocksdb_clone/r/compression_config.result

@@ -0,0 +1,23 @@
+INSTALL PLUGIN clone SONAME 'CLONE_PLUGIN';
+SET GLOBAL clone_enable_compression = ON;
+SET GLOBAL clone_compression_algorithm = ZSTD;
+SET GLOBAL clone_zstd_compression_level = 8;
+SET GLOBAL clone_autotune_concurrency = OFF;
+SET GLOBAL clone_max_concurrency = 8;
+SET GLOBAL clone_valid_donor_list = 'HOST:PORT';
+CLONE INSTANCE FROM USER@HOST:PORT IDENTIFIED BY '' DATA DIRECTORY = 'CLONE_DATADIR';
+select ID, STATE, ERROR_NO from performance_schema.clone_status;
+ID	STATE	ERROR_NO
+1	Completed	0
+select ID, STAGE, STATE from performance_schema.clone_progress;
+ID	STAGE	STATE
+1	DROP DATA	Completed
+1	FILE COPY	Completed
+1	PAGE COPY	Completed
+1	SST COPY	Completed
+1	REDO COPY	Completed
+1	FILE SYNC	Completed
+1	RESTART	Not Started
+1	RECOVERY	Not Started
+# restart
+UNINSTALL PLUGIN clone;

mysql-test/suite/rocksdb_clone/t/compression_config.test

@@ -0,0 +1,28 @@
+--source include/have_rocksdb.inc
+
+--let $HOST = 127.0.0.1
+--let $PORT =`select @@port`
+--let $USER = root
+--let remote_clone = 1
+
+--source ../../clone/include/clone_connection_begin.inc
+--let $CLONE_DATADIR = $MYSQL_TMP_DIR/data_new
+
+# Install Clone Plugin
+--replace_result $CLONE_PLUGIN CLONE_PLUGIN
+--eval INSTALL PLUGIN clone SONAME '$CLONE_PLUGIN'
+
+--connection clone_conn_1
+SET GLOBAL clone_enable_compression = ON;
+SET GLOBAL clone_compression_algorithm = ZSTD;
+SET GLOBAL clone_zstd_compression_level = 8;
+--source ../../clone/include/clone_command.inc
+
+--force-rmdir $CLONE_DATADIR
+
+--let restart_parameters=
+--source include/restart_mysqld.inc
+
+# Clean up
+UNINSTALL PLUGIN clone;
+--source ../../clone/include/clone_connection_end.inc

plugin/clone/include/clone.h

@@ -90,6 +90,10 @@ extern uint clone_max_io_bandwidth;
 /** Clone system variable: If network compression is enabled */
 extern bool clone_enable_compression;
 
+extern unsigned long clone_compression_algorithm;
+
+extern uint clone_zstd_compression_level;
+
 /** Clone system variable: SSL private key */
 extern char *clone_client_ssl_private_key;
 

plugin/clone/include/clone_client.h

@@ -51,6 +51,8 @@ using Time_Msec = std::chrono::milliseconds;
 using Time_Sec = std::chrono::seconds;
 using Time_Min = std::chrono::minutes;
 
+extern const char *clone_compression_algorithms[3];
+
 struct Thread_Info {
   /** Default constructor */
   Thread_Info() = default;

plugin/clone/src/clone_client.cc

@@ -26,7 +26,10 @@ Clone Plugin: Client implementation
 
 */
 #include <inttypes.h>
+#include <string>
 
+#include "my_compress.h"
+#include "plugin/clone/include/clone.h"
 #include "plugin/clone/include/clone_client.h"
 #include "plugin/clone/include/clone_os.h"
 
@@ -45,6 +48,8 @@ Time_Sec Client::s_reconnect_timeout{300};
 MySQL connect_timeout configuration. */
 Time_Sec Client::s_reconnect_interval{5};
 
+const char *clone_compression_algorithms[] = {"ZLIB", "ZSTD", NullS};
+
 /** Start concurrent clone  operation
 @param[in]	share	shared client information
 @param[in]	index	current thread index */
@@ -896,6 +901,13 @@ int Client::connect_remote(bool is_restart, bool use_aux) {
   mysql_clone_ssl_context ssl_context;
 
   ssl_context.m_enable_compression = clone_enable_compression;
+
+  if (ssl_context.m_enable_compression) {
+    ssl_context.m_compression_algorithm =
+        clone_compression_algorithms[clone_compression_algorithm];
+    ssl_context.m_compression_level = clone_zstd_compression_level;
+  }
+
   ssl_context.m_server_extn =
       ssl_context.m_enable_compression ? &m_conn_server_extn : nullptr;
   ssl_context.m_ssl_mode = m_share->m_ssl_mode;

plugin/clone/src/clone_plugin.cc

@@ -70,6 +70,10 @@ uint clone_max_io_bandwidth;
 /** Clone system variable: If network compression is enabled */
 bool clone_enable_compression;
 
+unsigned long clone_compression_algorithm;
+
+uint clone_zstd_compression_level;
+
 /** Clone system variable: valid list of donor addresses. */
 static char *clone_valid_donor_list;
 
@@ -636,6 +640,26 @@ static MYSQL_SYSVAR_BOOL(enable_compression, clone_enable_compression,
                          "If compression is done at network", nullptr, nullptr,
                          false); /* Disable compression by default */
 
+enum enum_clone_compression_algorithm { ZLIB = 0, ZSTD };
+
+static TYPELIB clone_compression_algorithms_typelib = {
+    array_elements(myclone::clone_compression_algorithms) - 1,
+    "clone_compression_algorithms_typelib",
+    myclone::clone_compression_algorithms, nullptr};
+
+static MYSQL_SYSVAR_ENUM(compression_algorithm, clone_compression_algorithm,
+                         PLUGIN_VAR_OPCMDARG | PLUGIN_VAR_MEMALLOC,
+                         "compression algorithm used in clone", nullptr,
+                         nullptr, enum_clone_compression_algorithm::ZSTD,
+                         &clone_compression_algorithms_typelib);
+
+static MYSQL_SYSVAR_UINT(zstd_compression_level, clone_zstd_compression_level,
+                         PLUGIN_VAR_NOCMDARG, "zstd compression level", nullptr,
+                         nullptr, 3, /* Default */
+                         1,          /* Minimum */
+                         10,         /* Maximum */
+                         1);
+
 /** List of valid donor addresses allowed to clone from. */
 static MYSQL_SYSVAR_STR(valid_donor_list, clone_valid_donor_list,
                         PLUGIN_VAR_OPCMDARG | PLUGIN_VAR_MEMALLOC,
@@ -705,6 +729,8 @@ static SYS_VAR *clone_system_variables[] = {
     MYSQL_SYSVAR(ssl_cert),
     MYSQL_SYSVAR(ssl_ca),
     MYSQL_SYSVAR(donor_timeout_after_network_failure),
+    MYSQL_SYSVAR(compression_algorithm),
+    MYSQL_SYSVAR(zstd_compression_level),
     nullptr};
 
 /** Declare clone plugin */

sql/server_component/clone_protocol_service.cc

@@ -394,6 +394,11 @@ DEFINE_METHOD(MYSQL *, mysql_clone_connect,
   /* Enable compression. */
   if (ssl_ctx->m_enable_compression) {
     mysql_options(mysql, MYSQL_OPT_COMPRESS, nullptr);
+    if (ssl_ctx->m_compression_algorithm)
+      mysql_options(mysql, MYSQL_OPT_COMPRESSION_ALGORITHMS,
+                    ssl_ctx->m_compression_algorithm);
+    mysql_options(mysql, MYSQL_OPT_ZSTD_COMPRESSION_LEVEL,
+                  &(ssl_ctx->m_compression_level));
     mysql_extension_set_server_extn(mysql, ssl_ctx->m_server_extn);
   }