MDEV-39492 Parallel Query: Study how to create worker threads

include/my_pthread.h

@@ -624,6 +624,20 @@ typedef uint64 my_thread_id;
 */
 #define MY_THREAD_ID_MAX UINT32_MAX
 
+#ifdef _WIN32
+#define MAX_THREAD_NAME 256
+#elif defined(__linux__)
+#define MAX_THREAD_NAME 16
+#elif defined(__FreeBSD__) || defined(__OpenBSD__)
+#define MAX_THREAD_NAME 19
+#include <pthread_np.h>
+#elif defined(__apple_build_version__)
+#include <sys/proc_info.h>
+#define MAX_THREAD_NAME MAXTHREADNAMESIZE
+#else
+#define MAX_THREAD_NAME 16
+#endif
+
 extern void my_threadattr_global_init(void);
 extern my_bool my_thread_global_init(void);
 extern void my_thread_set_name(const char *);

libmysqld/CMakeLists.txt

@@ -72,6 +72,7 @@ SET(SQL_EMBEDDED_SOURCES emb_qcache.cc libmysqld.c lib_sql.cc
            ../sql/mf_iocache.cc ../sql/my_decimal.cc 
            ../sql/net_serv.cc ../sql/opt_range.cc
            ../sql/opt_group_by_cardinality.cc
+           ../sql/sql_parallel_workers.cc
            ../sql/opt_rewrite_date_cmp.cc
            ../sql/opt_rewrite_remove_casefold.cc
            ../sql/opt_sargable_left.cc

mysql-test/main/mysqld--help.result

@@ -1012,6 +1012,9 @@ The following specify which files/extra groups are read (specified before remain
  Cost of checking the row against the WHERE clause.
  Increasing this will have the optimizer to prefer plans
  with less row combinations
+ --parallel-worker-threads=# 
+ Number of worker threads available for parallel query
+ execution. 0 means parallel execution is disabled
  --path=name         Comma-separated list of schema names that defines the
  search order for stored routines
  --performance-schema 
@@ -2004,6 +2007,7 @@ optimizer-trace
 optimizer-trace-max-mem-size 1048576
 optimizer-use-condition-selectivity 4
 optimizer-where-cost 0.032
+parallel-worker-threads 0
 path CURRENT_SCHEMA
 performance-schema FALSE
 performance-schema-accounts-size -1

mysql-test/main/parallel_query.result

@@ -0,0 +1,25 @@
+#
+# MDEV-39492 Parallel Query: Study how to create worker threads
+#
+connect  killee, localhost, root, , ;
+# check that kill query on a parallel worker is passed to the manager
+connection killee;
+set session parallel_worker_threads=3;
+select seq from seq_1_to_2;;
+connection default;
+kill query ID;
+connection killee;
+ERROR 70100: Query execution was interrupted
+# reset connection, discard any generated errors
+disconnect killee;
+connect  killee, localhost, root, , ;
+# check that kill on a parallel worker is passed to the manager
+# doing this last so we don't need to restart default connection
+set session parallel_worker_threads=3;
+select seq from seq_1_to_2;;
+connection default;
+kill ID;
+connection killee;
+Got one of the listed errors
+# generated error die with connection
+connection default;

mysql-test/main/parallel_query.test

@@ -0,0 +1,58 @@
+#
+# Test KILL and KILL QUERY statements.
+#
+
+-- source include/count_sessions.inc
+-- source include/not_embedded.inc
+# this will be used in the future  -- source include/have_innodb.inc
+-- source include/have_sequence.inc
+
+--disable_service_connection
+
+--echo #
+--echo # MDEV-39492 Parallel Query: Study how to create worker threads
+--echo #
+
+connect (killee, localhost, root, , );
+
+--echo # check that kill query on a parallel worker is passed to the manager
+
+connection killee;
+let $id= `select connection_id()`;
+set session parallel_worker_threads=3;
+--send select seq from seq_1_to_2;
+connection default;
+let $name= "%parallel worker $id";
+let $wait_condition= SELECT @kid:=ID from information_schema.processlist where info like $name limit 1;
+source include/wait_condition.inc;
+let $killID= `select @kid`;
+--replace_result $killID ID
+eval kill query $killID;
+connection killee;
+--error 1317
+--reap
+
+--echo # reset connection, discard any generated errors
+disconnect killee;
+connect (killee, localhost, root, , );
+
+--echo # check that kill on a parallel worker is passed to the manager
+--echo # doing this last so we don't need to restart default connection
+
+let $id= `select connection_id()`;
+set session parallel_worker_threads=3;
+--send select seq from seq_1_to_2;
+connection default;
+let $name= "%parallel worker $id";
+let $wait_condition= SELECT @kid:=ID from information_schema.processlist where info like $name limit 1;
+source include/wait_condition.inc;
+let $killID= `select @kid`;
+--replace_result $killID ID
+eval kill $killID;
+connection killee;
+--error 1317,2013
+--reap
+--echo # generated error die with connection
+
+connection default;
+--source include/wait_until_count_sessions.inc

mysql-test/main/parallel_query_oom.result

@@ -0,0 +1,21 @@
+#
+# MDEV-39492 Parallel Query: OOM in worker error_to_queue surfaces a
+# single ER_OUTOFMEMORY warning so worker diagnostics aren't silently
+# lost.
+#
+set @save_dbug=@@global.debug_dbug;
+set global debug_dbug="+d,pwt_error_to_queue_oom";
+set session parallel_worker_threads=1;
+# The prototype worker emits either a warning or my_error() depending on the
+# parity of its thread id. Either path runs through error_to_queue; the
+# DBUG injection forces both into the OOM branch, so neither the original
+# warning nor the original error reaches the user. We expect just the
+# manager-surfaced ER_OUTOFMEMORY warning.
+select count(*) from seq_1_to_2;
+count(*)
+2
+show warnings;
+Level	Code	Message
+Warning	1037	Parallel worker diagnostics were dropped due to memory allocation failure
+set global debug_dbug=@save_dbug;
+set session parallel_worker_threads=default;

mysql-test/main/parallel_query_oom.test

@@ -0,0 +1,33 @@
+#
+# Test OOM handling in the parallel worker error/warning queue.
+#
+
+-- source include/count_sessions.inc
+-- source include/not_embedded.inc
+-- source include/have_sequence.inc
+-- source include/have_debug.inc
+
+--disable_service_connection
+
+--echo #
+--echo # MDEV-39492 Parallel Query: OOM in worker error_to_queue surfaces a
+--echo # single ER_OUTOFMEMORY warning so worker diagnostics aren't silently
+--echo # lost.
+--echo #
+
+set @save_dbug=@@global.debug_dbug;
+set global debug_dbug="+d,pwt_error_to_queue_oom";
+set session parallel_worker_threads=1;
+
+--echo # The prototype worker emits either a warning or my_error() depending on the
+--echo # parity of its thread id. Either path runs through error_to_queue; the
+--echo # DBUG injection forces both into the OOM branch, so neither the original
+--echo # warning nor the original error reaches the user. We expect just the
+--echo # manager-surfaced ER_OUTOFMEMORY warning.
+select count(*) from seq_1_to_2;
+show warnings;
+
+set global debug_dbug=@save_dbug;
+set session parallel_worker_threads=default;
+
+--source include/wait_until_count_sessions.inc

mysql-test/suite/sys_vars/r/sysvars_server_notembedded.result

@@ -2992,6 +2992,16 @@ NUMERIC_BLOCK_SIZE	NULL
 ENUM_VALUE_LIST	NULL
 READ_ONLY	NO
 COMMAND_LINE_ARGUMENT	REQUIRED
+VARIABLE_NAME	PARALLEL_WORKER_THREADS
+VARIABLE_SCOPE	SESSION
+VARIABLE_TYPE	BIGINT UNSIGNED
+VARIABLE_COMMENT	Number of worker threads available for parallel query execution. 0 means parallel execution is disabled
+NUMERIC_MIN_VALUE	0
+NUMERIC_MAX_VALUE	100
+NUMERIC_BLOCK_SIZE	1
+ENUM_VALUE_LIST	NULL
+READ_ONLY	NO
+COMMAND_LINE_ARGUMENT	REQUIRED
 VARIABLE_NAME	PATH
 VARIABLE_SCOPE	SESSION
 VARIABLE_TYPE	VARCHAR

mysys/my_thread_name.cc

@@ -20,12 +20,7 @@
 #include <stdio.h>
 
 #ifdef _WIN32
-#define MAX_THREAD_NAME 256
 typedef HRESULT (*func_SetThreadDescription)(HANDLE,PCWSTR);
-#elif defined(__linux__)
-#define MAX_THREAD_NAME 16
-#elif defined(__FreeBSD__) || defined(__OpenBSD__)
-#include <pthread_np.h>
 #endif
 
 #if defined(HAVE_PSI_THREAD_INTERFACE) && !defined DBUG_OFF

sql/CMakeLists.txt

@@ -115,6 +115,7 @@ SET (SQL_SOURCE
                ../sql-common/client_plugin.c
                opt_range.cc vector_mhnsw.cc
                opt_group_by_cardinality.cc
+               sql_parallel_workers.cc
                opt_rewrite_date_cmp.cc
                opt_rewrite_remove_casefold.cc
                opt_sargable_left.cc

sql/mysqld.cc

@@ -9911,6 +9911,7 @@ PSI_stage_info stage_starting= { 0, "starting", 0};
 PSI_stage_info stage_waiting_for_flush= { 0, "Waiting for non trans tables to be flushed", 0};
 PSI_stage_info stage_waiting_for_ddl= { 0, "Waiting for DDLs", 0};
 PSI_stage_info stage_waiting_for_reset_master= { 0, "Waiting for a running RESET MASTER to complete", 0};
+PSI_stage_info stage_reading_data_from_parallel_worker= { 0, "Reading data from parallel worker", 0};
 
 #ifdef WITH_WSREP
 // Additional Galera thread states
@@ -10003,6 +10004,10 @@ PSI_memory_key key_memory_user_var_entry_value;
 PSI_memory_key key_memory_String_value;
 PSI_memory_key key_memory_WSREP;
 PSI_memory_key key_memory_trace_ddl_info;
+PSI_memory_key key_memory_pwt_queued_event;
+PSI_memory_key key_memory_pwt_error_message;
+PSI_memory_key key_memory_pwt_workers;
+PSI_memory_key key_memory_pwt_db;
 
 #ifdef HAVE_PSI_INTERFACE
 
@@ -10144,7 +10149,8 @@ PSI_stage_info *all_server_stages[]=
   & stage_reading_semi_sync_ack,
   & stage_waiting_for_deadlock_kill,
   & stage_starting,
-  & stage_waiting_for_reset_master
+  & stage_waiting_for_reset_master,
+  & stage_reading_data_from_parallel_worker
 #ifdef WITH_WSREP
   ,
   & stage_waiting_isolation,
@@ -10256,6 +10262,9 @@ static PSI_memory_info all_server_memory[]=
   { &key_memory_trace_ddl_info, "TRACE_DDL_INFO", 0}
 };
 
+
+extern void pwt_init_psi_keys(void);
+
 /**
   Initialise all the performance schema instrumentation points
   used by the server.
@@ -10342,6 +10351,7 @@ void init_server_psi_keys(void)
   stmt_info_rpl.m_flags= PSI_FLAG_MUTABLE;
   mysql_statement_register(category, &stmt_info_rpl, 1);
 #endif
+  pwt_init_psi_keys();
 }
 
 #endif /* HAVE_PSI_INTERFACE */

sql/mysqld.h

@@ -499,6 +499,10 @@ extern PSI_memory_key key_memory_Table_trigger_dispatcher;
 extern PSI_memory_key key_memory_native_functions;
 extern PSI_memory_key key_memory_WSREP;
 extern PSI_memory_key key_memory_trace_ddl_info;
+extern PSI_memory_key key_memory_pwt_queued_event;
+extern PSI_memory_key key_memory_pwt_error_message;
+extern PSI_memory_key key_memory_pwt_workers;
+extern PSI_memory_key key_memory_pwt_db;
 
 /*
   MAINTAINER: Please keep this list in order, to limit merge collisions.
@@ -646,6 +650,7 @@ extern PSI_stage_info stage_slave_background_wait_request;
 extern PSI_stage_info stage_waiting_for_deadlock_kill;
 extern PSI_stage_info stage_starting;
 extern PSI_stage_info stage_waiting_for_reset_master;
+extern PSI_stage_info stage_reading_data_from_parallel_worker;
 #ifdef WITH_WSREP
 // Additional Galera thread states
 extern PSI_stage_info stage_waiting_isolation;

sql/privilege.h

@@ -452,6 +452,8 @@ constexpr privilege_t PRIV_SET_SYSTEM_GLOBAL_VAR_MAX_CONNECT_ERRORS=
 // Was SUPER_ACL prior to 10.5.2
 constexpr privilege_t PRIV_SET_SYSTEM_GLOBAL_VAR_MAX_PASSWORD_ERRORS=
   CONNECTION_ADMIN_ACL;
+constexpr privilege_t PRIV_SET_SYSTEM_GLOBAL_VAR_PARALLEL_WORKER_THREADS=
+  CONNECTION_ADMIN_ACL;
 // Was SUPER_ACL prior to 10.5.2
 constexpr privilege_t PRIV_SET_SYSTEM_GLOBAL_VAR_PROXY_PROTOCOL_NETWORKS=
   CONNECTION_ADMIN_ACL;

sql/sql_class.cc

@@ -714,7 +714,7 @@ const char *thd_where(THD *thd)
 THD::THD(my_thread_id id, bool is_wsrep_applier)
   :Statement(&main_lex, &main_mem_root, STMT_CONVENTIONAL_EXECUTION,
              /* statement id */ 0),
-   rli_fake(0), rgi_fake(0), rgi_slave(NULL),
+   rli_fake(0), rgi_fake(0), rgi_slave(NULL), pwt_worker_info(NULL),
    protocol_text(this), protocol_binary(this), initial_status_var(0),
    m_current_stage_key(0), m_psi(0), start_time(0), start_time_sec_part(0),
    in_sub_stmt(0), log_all_errors(0),
@@ -5462,8 +5462,7 @@ void destroy_thd(MYSQL_THD thd)
 
 /**
   Create a THD that only has auxiliary functions
-  It will never be added to the global connection list
-  server_threads. It does not represent any client connection.
+  It does not represent any client connection.
 
   It should never be counted, because it will stall the
   shutdown. It is solely for engine's internal use,

sql/sql_class.h

@@ -872,6 +872,7 @@ typedef struct system_variables
   ulong server_id;
   ulong session_track_transaction_info;
   ulong threadpool_priority;
+  ulong parallel_worker_threads;
   ulong vers_alter_history;
 
   /* deadlock detection */
@@ -3154,7 +3155,7 @@ enum class THD_WHERE
 
 
 const char *thd_where(THD *thd);
-
+class pwt_worker;
 
 /**
   @class THD
@@ -3199,6 +3200,8 @@ class THD: public THD_count, /* this must be first */
   /* Slave applier execution context */
   rpl_group_info* rgi_slave;
 
+  pwt_worker *pwt_worker_info;
+
   union {
     rpl_io_thread_info *rpl_io_info;
     rpl_sql_thread_info *rpl_sql_info;