MDEV-19716: ASAN use-after-poison in Query_log_event::Query_log_event /...

MDEV-19716: ASAN use-after-poison in Query_log_event::Query_log_event / THD::log_events_and_free_tmp_shares

Analysis:
========
When a given client session ends on a master, the server logs a DROP TEMPORARY
TABLE IF EXISTS statement for each temporary table that still exists in the
current session. It ensures a proper temporary table cleanup on the slave. In
order to write the DROP TEMPORARY TABLE query in binary log a 'Query_log_event'
object is created. Within the 'Query_log_event' constructor
'thd->lex->sql_command' is read to identify what type of cache needs to be
used to write the query. When the code reaches here as part of THD::cleanup
the 'thd->lex->sql_command' will be in an invalid state. The 'thd->lex' could
have been cleared or it could be pointing to a statement which was in the
middle of execution when the session ended. In such cases ASAN reports
use-after-poison error.

Fix:
===
The 'THD::Cleanup' code invokes 'THD::log_events_and_free_tmp_shares' to look
for temporary tables and write appropriate DROP TABLE stmts for them. This
cleanup code provides a special flag named 'direct=TRUE' to the
Query_log_event constructor. Having 'direct=TRUE' means that this query
doesn't require any caching. Hence in this scenario the 'Query_log_event'
constructor should respect the 'direct' flag and simply skip the logic of
deciding the type of cache to be used for the statement. Hence the code will
not access the stale lex object.

mysql-test/suite/rpl/r/rpl_drop_temp_table_invaid_lex.result

+include/master-slave.inc
+[connection master]
+connect  con1,localhost,root,,;
+CREATE TEMPORARY TABLE tmp (a INT);
+CREATE TABLE non_existing_db.t SELECT 1 AS b;
+disconnect con1;
+connection default;
+include/rpl_end.inc

mysql-test/suite/rpl/t/rpl_drop_temp_table_invaid_lex.test

+# ==== Purpose ====
+#
+# Test verifies that no ASAN issues are reported at the time of writing DROP
+# TEMPORARY TABLE statements to binary log as part of session cleanup.
+#
+# ==== Implementation ====
+#
+# Steps:
+#    1 - Create a new connection named 'con1'.
+#    2 - Create a temporary table named 'tmp' as part of connection 'con1'.
+#    3 - Try to disconnect the current session when a CREATE .. SELECT
+#        statement is in the middle of execution.
+#    4 - Observe that no ASAN issue is reported.
+#
+# ==== References ====
+#
+# MDEV-19716: ASAN use-after-poison in Query_log_event::Query_log_event /
+# THD::log_events_and_free_tmp_shares
+
+--source include/master-slave.inc
+
+--connect (con1,localhost,root,,)
+
+CREATE TEMPORARY TABLE tmp (a INT);
+
+--send CREATE TABLE non_existing_db.t SELECT 1 AS b
+--disconnect con1
+
+--connection default
+
+--source include/rpl_end.inc

sql/log_event.cc

@@ -4415,41 +4415,44 @@ Query_log_event::Query_log_event(THD* thd_arg, const char* query_arg, size_t que
   bool trx_cache= FALSE;
   cache_type= Log_event::EVENT_INVALID_CACHE;
 
-  switch (lex->sql_command)
+  if (!direct)
   {
-    case SQLCOM_DROP_TABLE:
-    case SQLCOM_DROP_SEQUENCE:
-      use_cache= (lex->tmp_table() && thd->in_multi_stmt_transaction_mode());
-    break;
+    switch (lex->sql_command)
+    {
+      case SQLCOM_DROP_TABLE:
+      case SQLCOM_DROP_SEQUENCE:
+        use_cache= (lex->tmp_table() && thd->in_multi_stmt_transaction_mode());
+        break;
 
-    case SQLCOM_CREATE_TABLE:
-    case SQLCOM_CREATE_SEQUENCE:
-      /*
-        If we are using CREATE ... SELECT or if we are a slave
-        executing BEGIN...COMMIT (generated by CREATE...SELECT) we
-        have to use the transactional cache to ensure we don't
-        calculate any checksum for the CREATE part.
-      */
-      trx_cache= (lex->first_select_lex()->item_list.elements &&
-                  thd->is_current_stmt_binlog_format_row()) ||
-                  (thd->variables.option_bits & OPTION_GTID_BEGIN);
-      use_cache= (lex->tmp_table() &&
-                  thd->in_multi_stmt_transaction_mode()) || trx_cache;
-      break;
-    case SQLCOM_SET_OPTION:
-      if (lex->autocommit)
-        use_cache= trx_cache= FALSE;
-      else
-        use_cache= TRUE;
-      break;
-    case SQLCOM_RELEASE_SAVEPOINT:
-    case SQLCOM_ROLLBACK_TO_SAVEPOINT:
-    case SQLCOM_SAVEPOINT:
-      use_cache= trx_cache= TRUE;
-      break;
-    default:
-      use_cache= sqlcom_can_generate_row_events(thd);
-      break;
+      case SQLCOM_CREATE_TABLE:
+      case SQLCOM_CREATE_SEQUENCE:
+        /*
+           If we are using CREATE ... SELECT or if we are a slave
+           executing BEGIN...COMMIT (generated by CREATE...SELECT) we
+           have to use the transactional cache to ensure we don't
+           calculate any checksum for the CREATE part.
+         */
+        trx_cache= (lex->first_select_lex()->item_list.elements &&
+            thd->is_current_stmt_binlog_format_row()) ||
+          (thd->variables.option_bits & OPTION_GTID_BEGIN);
+        use_cache= (lex->tmp_table() &&
+            thd->in_multi_stmt_transaction_mode()) || trx_cache;
+        break;
+      case SQLCOM_SET_OPTION:
+        if (lex->autocommit)
+          use_cache= trx_cache= FALSE;
+        else
+          use_cache= TRUE;
+        break;
+      case SQLCOM_RELEASE_SAVEPOINT:
+      case SQLCOM_ROLLBACK_TO_SAVEPOINT:
+      case SQLCOM_SAVEPOINT:
+        use_cache= trx_cache= TRUE;
+        break;
+      default:
+        use_cache= sqlcom_can_generate_row_events(thd);
+        break;
+    }
   }
 
   if (!use_cache || direct)