diff --git a/client/mysqlbinlog.cc b/client/mysqlbinlog.cc
index 870145e35e0..43d670737a0 100644
--- a/client/mysqlbinlog.cc
+++ b/client/mysqlbinlog.cc
@@ -114,7 +114,7 @@ static void warning(const char *format, ...) ATTRIBUTE_FORMAT(printf, 1, 2);
 static bool one_database=0, one_table=0, to_last_remote_log= 0, disable_log_bin= 0;
 static bool opt_hexdump= 0, opt_version= 0;
 const char *base64_output_mode_names[]=
-{"NEVER", "AUTO", "ALWAYS", "UNSPEC", "DECODE-ROWS", NullS};
+{"NEVER", "AUTO", "UNSPEC", "DECODE-ROWS", NullS};
 TYPELIB base64_output_mode_typelib=
   { array_elements(base64_output_mode_names) - 1, "",
     base64_output_mode_names, NULL };
@@ -836,53 +836,6 @@ static bool shall_skip_table(const char *log_tblname)
          strcmp(log_tblname, table);
 }
 
-
-/**
-  Prints the given event in base64 format.
-
-  The header is printed to the head cache and the body is printed to
-  the body cache of the print_event_info structure.  This allows all
-  base64 events corresponding to the same statement to be joined into
-  one BINLOG statement.
-
-  @param[in] ev Log_event to print.
-  @param[in,out] result_file FILE to which the output will be written.
-  @param[in,out] print_event_info Parameters and context state
-  determining how to print.
-
-  @retval ERROR_STOP An error occurred - the program should terminate.
-  @retval OK_CONTINUE No error, the program should continue.
-*/
-static Exit_status
-write_event_header_and_base64(Log_event *ev, FILE *result_file,
-                              PRINT_EVENT_INFO *print_event_info)
-{
-  IO_CACHE *head= &print_event_info->head_cache;
-  IO_CACHE *body= &print_event_info->body_cache;
-  DBUG_ENTER("write_event_header_and_base64");
-
-  /* Write header and base64 output to cache */
-  if (ev->print_header(head, print_event_info, FALSE))
-    DBUG_RETURN(ERROR_STOP);
-
-  DBUG_ASSERT(print_event_info->base64_output_mode == BASE64_OUTPUT_ALWAYS);
-
-  if (ev->print_base64(body, print_event_info,
-                       print_event_info->base64_output_mode !=
-                       BASE64_OUTPUT_DECODE_ROWS))
-    DBUG_RETURN(ERROR_STOP);
-
-  /* Read data from cache and write to result file */
-  if (copy_event_cache_to_file_and_reinit(head, result_file) ||
-      copy_event_cache_to_file_and_reinit(body, result_file))
-  {
-    error("Error writing event to file.");
-    DBUG_RETURN(ERROR_STOP);
-  }
-  DBUG_RETURN(OK_CONTINUE);
-}
-
-
 static bool print_base64(PRINT_EVENT_INFO *print_event_info, Log_event *ev)
 {
   /*
@@ -1134,19 +1087,9 @@ Exit_status process_event(PRINT_EVENT_INFO *print_event_info, Log_event *ev,
         qe->flags|= LOG_EVENT_SUPPRESS_USE_F;
       }
       print_use_stmt(print_event_info, qe);
-      if (opt_base64_output_mode == BASE64_OUTPUT_ALWAYS)
-      {
-        if ((retval= write_event_header_and_base64(ev, result_file,
-                                                   print_event_info)) !=
-            OK_CONTINUE)
-          goto end;
-      }
-      else
-      {
-        print_skip_replication_statement(print_event_info, ev);
-        if (ev->print(result_file, print_event_info))
-          goto err;
-      }
+      print_skip_replication_statement(print_event_info, ev);
+      if (ev->print(result_file, print_event_info))
+        goto err;
       if (head->error == -1)
         goto err;
       break;
@@ -1170,19 +1113,9 @@ Exit_status process_event(PRINT_EVENT_INFO *print_event_info, Log_event *ev,
 	filename and use LOCAL), prepared in the 'case EXEC_LOAD_EVENT' 
 	below.
       */
-      if (opt_base64_output_mode == BASE64_OUTPUT_ALWAYS)
-      {
-        if ((retval= write_event_header_and_base64(ce, result_file,
-                                                   print_event_info)) !=
-            OK_CONTINUE)
-          goto end;
-      }
-      else
-      {
-        print_skip_replication_statement(print_event_info, ev);
-        if (ce->print(result_file, print_event_info, TRUE))
-          goto err;
-      }
+      print_skip_replication_statement(print_event_info, ev);
+      if (ce->print(result_file, print_event_info, TRUE))
+        goto err;
       // If this binlog is not 3.23 ; why this test??
       if (glob_description_event->binlog_version >= 3)
       {
@@ -1589,12 +1522,10 @@ static struct my_option my_options[] =
    "--verbose option is also given; "
    "'auto' prints base64 only when necessary (i.e., for row-based events and "
    "format description events); "
-   "'always' prints base64 whenever possible. "
-   "--base64-output with no 'name' argument is equivalent to "
-   "--base64-output=always and is also deprecated.  If no "
-   "--base64-output[=name] option is given at all, the default is 'auto'.",
+   "If no --base64-output=name option is given at all, the default is "
+   "'auto'.",
    &opt_base64_output_mode_str, &opt_base64_output_mode_str,
-   0, GET_STR, OPT_ARG, 0, 0, 0, 0, 0, 0},
+   0, GET_STR, REQUIRED_ARG, 0, 0, 0, 0, 0, 0},
   /*
     mysqlbinlog needs charsets knowledge, to be able to convert a charset
     number found in binlog to a charset name (to be able to print things
@@ -2038,20 +1969,15 @@ get_one_option(const struct my_option *opt, const char *argument, const char *fi
     stop_datetime= convert_str_to_timestamp(stop_datetime_str);
     break;
   case OPT_BASE64_OUTPUT_MODE:
-    if (argument == NULL)
-      opt_base64_output_mode= BASE64_OUTPUT_ALWAYS;
-    else
-    {
-      int val;
+    int val;
 
-      if ((val= find_type_with_warning(argument, &base64_output_mode_typelib,
-                                       opt->name)) <= 0)
-      {
-        sf_leaking_memory= 1; /* no memory leak reports here */
-        die();
-      }
-      opt_base64_output_mode= (enum_base64_output_mode) (val - 1);
+    if ((val= find_type_with_warning(argument, &base64_output_mode_typelib,
+                                     opt->name)) <= 0)
+    {
+      sf_leaking_memory= 1; /* no memory leak reports here */
+      die();
     }
+    opt_base64_output_mode= (enum_base64_output_mode)(val - 1);
     break;
   case OPT_REWRITE_DB:    // db_from->db_to
   {
@@ -2901,8 +2827,7 @@ static Exit_status check_header(IO_CACHE* file,
                 (ulonglong)tmp_pos);
           return ERROR_STOP;
         }
-        if (opt_base64_output_mode == BASE64_OUTPUT_AUTO
-            || opt_base64_output_mode == BASE64_OUTPUT_ALWAYS)
+        if (opt_base64_output_mode == BASE64_OUTPUT_AUTO)
         {
           /*
             process_event will delete *description_event and set it to
diff --git a/man/mysqlbinlog.1 b/man/mysqlbinlog.1
index 3d12a8fa083..633300bb7c5 100644
--- a/man/mysqlbinlog.1
+++ b/man/mysqlbinlog.1
@@ -141,7 +141,7 @@ Display a help message and exit\&.
 .\}
 .\" mysqlbinlog: base64-output option
 .\" base64-output option: mysqlbinlog
-\fB\-\-base64\-output[=\fR\fB\fIvalue\fR\fR\fB]\fR
+\fB\-\-base64\-output=\fR\fB\fIvalue\fR\fR\fB\fR
 .sp
 This option determines when events should be displayed encoded as base\-64 strings using
 BINLOG
@@ -192,22 +192,6 @@ to re\-execute binary log file contents\&. The other option values are intended
 .sp -1
 .IP \(bu 2.3
 .\}
-ALWAYS
-displays
-BINLOG
-statements whenever possible\&. This is the implied value if the option is given as
-\fB\-\-base64\-output\fR
-without a value\&. Both ALWAYS and not giving a value are deprecated.
-.RE
-.sp
-.RS 4
-.ie n \{\
-\h'-04'\(bu\h'+03'\c
-.\}
-.el \{\
-.sp -1
-.IP \(bu 2.3
-.\}
 NEVER
 causes
 BINLOG
diff --git a/mysql-test/include/commit.inc b/mysql-test/include/commit.inc
index 865f3296b39..a3c64096adf 100644
--- a/mysql-test/include/commit.inc
+++ b/mysql-test/include/commit.inc
@@ -775,7 +775,7 @@ call p_verify_status_increment(2, 0, 2, 0);
 commit;
 call p_verify_status_increment(0, 0, 0, 0);
 check table t1, t2, t3;
-call p_verify_status_increment(6, 0, 6, 0);
+call p_verify_status_increment(4, 0, 4, 0);
 commit;
 call p_verify_status_increment(0, 0, 0, 0);
 drop view v1;
diff --git a/mysql-test/lib/My/Debugger.pm b/mysql-test/lib/My/Debugger.pm
index 8a6a3657877..5288f0740fa 100644
--- a/mysql-test/lib/My/Debugger.pm
+++ b/mysql-test/lib/My/Debugger.pm
@@ -139,7 +139,7 @@ sub do_args($$$$$) {
   my $v = $debuggers{$k};
 
   # on windows mtr args are quoted (for system), otherwise not (for exec)
-  sub quote($) { $_[0] =~ / / ? "\"$_[0]\"" : $_[0] }
+  sub quote($) { $_[0] =~ /[; ]/ ? "\"$_[0]\"" : $_[0] }
   sub unquote($) { $_[0] =~ s/^"(.*)"$/$1/; $_[0] }
   sub quote_from_mtr($) { IS_WINDOWS() ? $_[0] : quote($_[0]) }
   sub unquote_for_mtr($) { IS_WINDOWS() ? $_[0] : unquote($_[0]) }
diff --git a/mysql-test/main/commit_1innodb.result b/mysql-test/main/commit_1innodb.result
index cf673f81d82..d090844cb74 100644
--- a/mysql-test/main/commit_1innodb.result
+++ b/mysql-test/main/commit_1innodb.result
@@ -879,7 +879,7 @@ Table	Op	Msg_type	Msg_text
 test.t1	check	status	OK
 test.t2	check	status	OK
 test.t3	check	status	OK
-call p_verify_status_increment(6, 0, 6, 0);
+call p_verify_status_increment(4, 0, 4, 0);
 SUCCESS
 
 commit;
diff --git a/mysql-test/main/mysqlbinlog.result b/mysql-test/main/mysqlbinlog.result
index 8e12ecdf8c1..83cc2ef510a 100644
--- a/mysql-test/main/mysqlbinlog.result
+++ b/mysql-test/main/mysqlbinlog.result
@@ -878,8 +878,19 @@ ROLLBACK /* added by mysqlbinlog */;
 /*!50530 SET @@SESSION.PSEUDO_SLAVE_MODE=0*/;
 End of 5.0 tests
 End of 5.1 tests
-# Expect deprecation warning.
-# Expect deprecation warning again.
+#
+# Expect error for invalid --base64-output argument value.
+# MYSQL_BINLOG std_data/master-bin.000001 --base64-output=always 2>&1
+Unknown option to base64-output: always
+Alternatives are: 'NEVER','AUTO','UNSPEC','DECODE-ROWS'
+#
+# Expect error for incomplete --base64-output argument.
+# MYSQL_BINLOG  std_data/master-bin.000001 --base64-output 2>&1
+mysqlbinlog: option '--base64-output' requires an argument
+#
+# Ensure --base64-output=auto outputs the same result as unspecified
+# MYSQL_BINLOG -v MYSQLD_DATADIR/master-bin.000001 > MYSQLTEST_VARDIR/tmp/mysqlbinlog_nob64spec.out
+# MYSQL_BINLOG --base64-output=auto -v MYSQLD_DATADIR/master-bin.000001 > MYSQLTEST_VARDIR/tmp/mysqlbinlog_b64auto.out
 RESET MASTER;
 CREATE DATABASE test1;
 USE test1;
diff --git a/mysql-test/main/mysqlbinlog.test b/mysql-test/main/mysqlbinlog.test
index 79b745e1645..c8a141404d0 100644
--- a/mysql-test/main/mysqlbinlog.test
+++ b/mysql-test/main/mysqlbinlog.test
@@ -521,18 +521,32 @@ remove_file $MYSQLTEST_VARDIR/tmp/mysqlbinlog.warn.empty;
 remove_file $MYSQLTEST_VARDIR/tmp/mysqlbinlog.warn;
 
 #
-# WL#5625: Deprecate mysqlbinlog options --base64-output=always and --base64-output
+# MDEV-25222: Remove mysqlbinlog options --base64-output=always and --base64-output
 #
---echo # Expect deprecation warning.
---exec $MYSQL_BINLOG --base64-output=always std_data/master-bin.000001 > /dev/null 2> $MYSQLTEST_VARDIR/tmp/mysqlbinlog.warn
---cat_file $MYSQLTEST_VARDIR/tmp/mysqlbinlog.warn
+--echo #
+--echo # Expect error for invalid --base64-output argument value.
+--echo # MYSQL_BINLOG std_data/master-bin.000001 --base64-output=always 2>&1
+--error 1
+--exec $MYSQL_BINLOG std_data/master-bin.000001 --base64-output=always 2>&1
 
---echo # Expect deprecation warning again.
---exec $MYSQL_BINLOG --base64-output std_data/master-bin.000001 > /dev/null 2> $MYSQLTEST_VARDIR/tmp/mysqlbinlog.warn
---cat_file $MYSQLTEST_VARDIR/tmp/mysqlbinlog.warn
+--echo #
+--echo # Expect error for incomplete --base64-output argument.
+--echo # MYSQL_BINLOG  std_data/master-bin.000001 --base64-output 2>&1
+# The error produces the absolute path of the mysqlbinlog executable, remove it.
+--replace_regex /.*mysqlbinlog.*:/mysqlbinlog:/i
+--error 1
+--exec $MYSQL_BINLOG std_data/master-bin.000001 --base64-output 2>&1
+
+--echo #
+--echo # Ensure --base64-output=auto outputs the same result as unspecified
+--echo # MYSQL_BINLOG -v MYSQLD_DATADIR/master-bin.000001 > MYSQLTEST_VARDIR/tmp/mysqlbinlog_nob64spec.out
+--exec  $MYSQL_BINLOG -v $MYSQLD_DATADIR/master-bin.000001 > $MYSQLTEST_VARDIR/tmp/mysqlbinlog_nob64spec.out
+--echo # MYSQL_BINLOG --base64-output=auto -v MYSQLD_DATADIR/master-bin.000001 > MYSQLTEST_VARDIR/tmp/mysqlbinlog_b64auto.out
+--exec  $MYSQL_BINLOG --base64-output=auto -v $MYSQLD_DATADIR/master-bin.000001 > $MYSQLTEST_VARDIR/tmp/mysqlbinlog_b64auto.out
+--diff_files $MYSQLTEST_VARDIR/tmp/mysqlbinlog_nob64spec.out $MYSQLTEST_VARDIR/tmp/mysqlbinlog_b64auto.out
+--remove_file $MYSQLTEST_VARDIR/tmp/mysqlbinlog_nob64spec.out
+--remove_file $MYSQLTEST_VARDIR/tmp/mysqlbinlog_b64auto.out
 
-# Clean up this part of the test.
---remove_file $MYSQLTEST_VARDIR/tmp/mysqlbinlog.warn
 
 # BUG#50914
 # This test verifies if the approach of the mysqlbinlog prints
diff --git a/mysql-test/suite/binlog/r/binlog_admin_cmd_kill.result b/mysql-test/suite/binlog/r/binlog_admin_cmd_kill.result
new file mode 100644
index 00000000000..a2eb7ee5c0a
--- /dev/null
+++ b/mysql-test/suite/binlog/r/binlog_admin_cmd_kill.result
@@ -0,0 +1,40 @@
+#
+# Kill OPTIMIZE command prior to table modification
+#
+RESET MASTER;
+CREATE TABLE t1 (f INT) ENGINE=INNODB;
+CREATE TABLE t2 (f INT) ENGINE=INNODB;
+connect con1,127.0.0.1,root,,test,$MASTER_MYPORT,;
+connection con1;
+SET debug_sync='admin_command_kill_before_modify SIGNAL ready_to_be_killed WAIT_FOR master_cont';
+OPTIMIZE TABLE t1,t2;
+connection default;
+SET debug_sync='now WAIT_FOR ready_to_be_killed';
+KILL THD_ID;
+SET debug_sync = 'reset';
+disconnect con1;
+include/show_binlog_events.inc
+Log_name	Pos	Event_type	Server_id	End_log_pos	Info
+master-bin.000001	#	Gtid	#	#	GTID #-#-#
+master-bin.000001	#	Query	#	#	use `test`; CREATE TABLE t1 (f INT) ENGINE=INNODB
+master-bin.000001	#	Gtid	#	#	GTID #-#-#
+master-bin.000001	#	Query	#	#	use `test`; CREATE TABLE t2 (f INT) ENGINE=INNODB
+DROP TABLE t1,t2;
+FLUSH LOGS;
+#
+# Kill OPTIMIZE command after table modification
+#
+CREATE TABLE t1 (f INT) ENGINE=INNODB;
+CREATE TABLE t2 (f INT) ENGINE=INNODB;
+connect con1,127.0.0.1,root,,test,$MASTER_MYPORT,;
+connection con1;
+SET debug_sync='admin_command_kill_after_modify SIGNAL ready_to_be_killed WAIT_FOR master_cont';
+OPTIMIZE TABLE t1,t2;
+connection default;
+SET debug_sync='now WAIT_FOR ready_to_be_killed';
+KILL THD_ID;
+SET debug_sync = 'reset';
+disconnect con1;
+DROP TABLE t1,t2;
+FLUSH LOGS;
+FOUND 1 /OPTIMIZE TABLE t1,t2/ in mysqlbinlog.out
diff --git a/mysql-test/suite/binlog/t/binlog_admin_cmd_kill.test b/mysql-test/suite/binlog/t/binlog_admin_cmd_kill.test
new file mode 100644
index 00000000000..9b248097ae2
--- /dev/null
+++ b/mysql-test/suite/binlog/t/binlog_admin_cmd_kill.test
@@ -0,0 +1,95 @@
+# ==== Purpose ====
+#
+# Test verifies that when an admin command execution is interrupted by KILL
+# command it should stop its execution. The admin command in binary log should
+# contain only the list of tables which have successfully executed admin
+# command prior to kill.
+#
+# ==== Implementation ====
+#
+# Steps:
+#    0 - Create two table t1,t2.
+#    1 - Execute OPTIMIZE TABLE t1,t2 command.
+#    2 - Using debug sync mechanism kill OPTIMIZE TABLE command at a stage
+#        where it has not optimized any table.
+#    3 - Check that OPTIMIZE TABLE command is not written to binary log.
+#    4 - Using debug sync mechanism hold the execution of OPTIMIZE TABLE after
+#        t1 table optimization. Now kill the OPTIMIZE TABLE command.
+#    5 - Observe the binlog output, the OPTIMIZE TABLE command should display `t1,t2`.
+#    6 - Please note that, we binlog the entire query even if at least one
+#        table is modified as admin commands are safe to replicate and they will
+#        not make the slave to diverge.
+#
+# ==== References ====
+#
+# MDEV-22530: Aborting OPTIMIZE TABLE still logs in binary log and replicates to the Slave server.
+#
+--source include/have_log_bin.inc
+--source include/have_debug.inc
+--source include/have_debug_sync.inc
+--source include/have_innodb.inc
+
+--echo #
+--echo # Kill OPTIMIZE command prior to table modification
+--echo #
+RESET MASTER;
+
+CREATE TABLE t1 (f INT) ENGINE=INNODB;
+CREATE TABLE t2 (f INT) ENGINE=INNODB;
+
+--connect(con1,127.0.0.1,root,,test,$MASTER_MYPORT,)
+--connection con1
+SET debug_sync='admin_command_kill_before_modify SIGNAL ready_to_be_killed WAIT_FOR master_cont';
+--send OPTIMIZE TABLE t1,t2
+
+--connection default
+SET debug_sync='now WAIT_FOR ready_to_be_killed';
+--let $thd_id= `SELECT ID FROM INFORMATION_SCHEMA.PROCESSLIST WHERE INFO LIKE '%OPTIMIZE TABLE %'`
+
+# Now kill.
+--replace_result $thd_id THD_ID
+eval KILL $thd_id;
+
+SET debug_sync = 'reset';
+--disconnect con1
+
+--source include/show_binlog_events.inc
+DROP TABLE t1,t2;
+
+FLUSH LOGS;
+
+--echo #
+--echo # Kill OPTIMIZE command after table modification
+--echo #
+
+CREATE TABLE t1 (f INT) ENGINE=INNODB;
+CREATE TABLE t2 (f INT) ENGINE=INNODB;
+
+--connect(con1,127.0.0.1,root,,test,$MASTER_MYPORT,)
+--connection con1
+SET debug_sync='admin_command_kill_after_modify SIGNAL ready_to_be_killed WAIT_FOR master_cont';
+--send OPTIMIZE TABLE t1,t2
+
+--connection default
+SET debug_sync='now WAIT_FOR ready_to_be_killed';
+--let $thd_id= `SELECT ID FROM INFORMATION_SCHEMA.PROCESSLIST WHERE INFO LIKE '%OPTIMIZE TABLE %'`
+
+# Now kill.
+--replace_result $thd_id THD_ID
+eval KILL $thd_id;
+
+SET debug_sync = 'reset';
+--disconnect con1
+
+DROP TABLE t1,t2;
+let $binlog_file= query_get_value(SHOW MASTER STATUS, File, 1);
+FLUSH LOGS;
+
+--let $MYSQLD_DATADIR= `select @@datadir`
+--exec $MYSQL_BINLOG $MYSQLD_DATADIR/$binlog_file > $MYSQLTEST_VARDIR/tmp/mysqlbinlog.out
+
+--let SEARCH_PATTERN= OPTIMIZE TABLE t1,t2
+--let SEARCH_FILE= $MYSQLTEST_VARDIR/tmp/mysqlbinlog.out
+--source include/search_pattern_in_file.inc
+
+--remove_file $MYSQLTEST_VARDIR/tmp/mysqlbinlog.out
diff --git a/mysql-test/suite/galera_3nodes/disabled.def b/mysql-test/suite/galera_3nodes/disabled.def
index d1bedd91beb..83723ea5d97 100644
--- a/mysql-test/suite/galera_3nodes/disabled.def
+++ b/mysql-test/suite/galera_3nodes/disabled.def
@@ -16,7 +16,6 @@ galera_gtid_2_cluster : MDEV-23775 Galera test failure on galera_3nodes.galera_g
 galera_ist_gcache_rollover : MDEV-23578 WSREP: exception caused by message: {v=0,t=1,ut=255,o=4,s=0,sr=0,as=1,f=6,src=50524cfe,srcvid=view_id(REG,50524cfe,4),insvid=view_id(UNKNOWN,00000000,0),ru=00000000,r=[-1,-1],fs=75,nl=(}
 galera_load_data_ist :  MDEV-24639 galera_3nodes.galera_load_data_ist MTR failed with SIGABRT: query 'reap' failed: 2013: Lost connection to server during query
 galera_load_data_ist :  MDEV-24639 galera_3nodes.galera_load_data_ist MTR failed with SIGABRT: query 'reap' failed: 2013: Lost connection to server during query
-galera_pc_bootstrap : MDEV-24650 galera_pc_bootstrap MTR failed: Could not execute 'check-testcase' before testcase
 galera_safe_to_bootstrap : MDEV-24097 galera_3nodes.galera_safe_to_bootstrap MTR sporadaically fails: Failed to start mysqld or mysql_shutdown failed
 galera_slave_options_do : MDEV-8798
 galera_slave_options_ignore : MDEV-8798
diff --git a/mysql-test/suite/rpl/r/rpl_mark_optimize_tbl_ddl.result b/mysql-test/suite/rpl/r/rpl_mark_optimize_tbl_ddl.result
new file mode 100644
index 00000000000..9aa31a73e49
--- /dev/null
+++ b/mysql-test/suite/rpl/r/rpl_mark_optimize_tbl_ddl.result
@@ -0,0 +1,79 @@
+include/rpl_init.inc [topology=1->2]
+connection server_1;
+FLUSH TABLES;
+ALTER TABLE mysql.gtid_slave_pos ENGINE=InnoDB;
+connection server_2;
+SET @save_slave_parallel_threads= @@GLOBAL.slave_parallel_threads;
+SET @save_slave_parallel_mode= @@GLOBAL.slave_parallel_mode;
+include/stop_slave.inc
+SET GLOBAL slave_parallel_threads=2;
+SET GLOBAL slave_parallel_mode=optimistic;
+include/start_slave.inc
+connection server_1;
+CREATE TABLE t1(a INT) ENGINE=INNODB;
+OPTIMIZE TABLE t1;
+Table	Op	Msg_type	Msg_text
+test.t1	optimize	note	Table does not support optimize, doing recreate + analyze instead
+test.t1	optimize	status	OK
+INSERT INTO t1 VALUES(1);
+INSERT INTO t1  SELECT 1+a FROM t1;
+INSERT INTO t1  SELECT 2+a FROM t1;
+connection server_2;
+#
+# Verify that following admin commands are marked as ddl
+# 'OPTIMIZE TABLE', 'REPAIR TABLE' and 'ANALYZE TABLE'
+#
+connection server_1;
+OPTIMIZE TABLE t1;
+Table	Op	Msg_type	Msg_text
+test.t1	optimize	note	Table does not support optimize, doing recreate + analyze instead
+test.t1	optimize	status	OK
+REPAIR TABLE t1;
+Table	Op	Msg_type	Msg_text
+test.t1	repair	note	The storage engine for the table doesn't support repair
+ANALYZE TABLE t1;
+Table	Op	Msg_type	Msg_text
+test.t1	analyze	status	Engine-independent statistics collected
+test.t1	analyze	status	OK
+FLUSH LOGS;
+FOUND 1 /GTID 0-1-8 ddl/ in mysqlbinlog.out
+FOUND 1 /GTID 0-1-9 ddl/ in mysqlbinlog.out
+FOUND 1 /GTID 0-1-10 ddl/ in mysqlbinlog.out
+#
+# Clean up
+#
+DROP TABLE t1;
+connection server_2;
+FLUSH LOGS;
+#
+# Check that ALTER TABLE commands with ANALYZE, OPTIMIZE and REPAIR on
+# partitions will be marked as DDL in binary log.
+#
+connection server_1;
+CREATE TABLE t1(id INT) PARTITION BY RANGE (id) (PARTITION p0 VALUES LESS THAN (100),
+PARTITION pmax VALUES LESS THAN (MAXVALUE));
+INSERT INTO t1 VALUES (1), (10), (100), (1000);
+ALTER TABLE t1 ANALYZE PARTITION p0;
+Table	Op	Msg_type	Msg_text
+test.t1	analyze	status	Engine-independent statistics collected
+test.t1	analyze	status	OK
+ALTER TABLE t1 OPTIMIZE PARTITION p0;
+Table	Op	Msg_type	Msg_text
+test.t1	optimize	status	OK
+ALTER TABLE t1 REPAIR PARTITION p0;
+Table	Op	Msg_type	Msg_text
+test.t1	repair	status	OK
+FLUSH LOGS;
+FOUND 1 /GTID 0-1-14 ddl/ in mysqlbinlog.out
+FOUND 1 /GTID 0-1-15 ddl/ in mysqlbinlog.out
+FOUND 1 /GTID 0-1-16 ddl/ in mysqlbinlog.out
+#
+# Clean up
+#
+DROP TABLE t1;
+connection server_2;
+include/stop_slave.inc
+SET GLOBAL slave_parallel_threads= @save_slave_parallel_threads;
+SET GLOBAL slave_parallel_mode= @save_slave_parallel_mode;
+include/start_slave.inc
+include/rpl_end.inc
diff --git a/mysql-test/suite/rpl/t/rpl_mark_optimize_tbl_ddl.test b/mysql-test/suite/rpl/t/rpl_mark_optimize_tbl_ddl.test
new file mode 100644
index 00000000000..6d66e3fd088
--- /dev/null
+++ b/mysql-test/suite/rpl/t/rpl_mark_optimize_tbl_ddl.test
@@ -0,0 +1,142 @@
+# ==== Purpose ====
+#
+# Test verifies that there is no deadlock or assertion in
+# slave_parallel_mode=optimistic configuration while applying admin command
+# like 'OPTIMIZE TABLE', 'REPAIR TABLE' and 'ANALYZE TABLE'.
+#
+# ==== Implementation ====
+#
+# Steps:
+#    0 - Create a table, execute OPTIMIZE TABLE command on the table followed
+#        by some DMLS.
+#    1 - No assert should happen on slave server.
+#    2 - Assert that 'OPTIMIZE TABLE', 'REPAIR TABLE' and 'ANALYZE TABLE' are
+#        marked as 'DDL' in the binary log.
+#
+# ==== References ====
+#
+# MDEV-17515: GTID Replication in optimistic mode deadlock
+#
+--source include/have_partition.inc
+--source include/have_innodb.inc
+--let $rpl_topology=1->2
+--source include/rpl_init.inc
+
+--connection server_1
+FLUSH TABLES;
+ALTER TABLE mysql.gtid_slave_pos ENGINE=InnoDB;
+
+--connection server_2
+SET @save_slave_parallel_threads= @@GLOBAL.slave_parallel_threads;
+SET @save_slave_parallel_mode= @@GLOBAL.slave_parallel_mode;
+--source include/stop_slave.inc
+SET GLOBAL slave_parallel_threads=2;
+SET GLOBAL slave_parallel_mode=optimistic;
+--source include/start_slave.inc
+
+--connection server_1
+CREATE TABLE t1(a INT) ENGINE=INNODB;
+OPTIMIZE TABLE t1;
+INSERT INTO t1 VALUES(1);
+INSERT INTO t1  SELECT 1+a FROM t1;
+INSERT INTO t1  SELECT 2+a FROM t1;
+--save_master_pos
+
+--connection server_2
+--sync_with_master
+
+--echo #
+--echo # Verify that following admin commands are marked as ddl
+--echo # 'OPTIMIZE TABLE', 'REPAIR TABLE' and 'ANALYZE TABLE'
+--echo #
+--connection server_1
+
+OPTIMIZE TABLE t1;
+--let optimize_gtid= `SELECT @@GLOBAL.gtid_binlog_pos`
+
+REPAIR TABLE t1;
+--let repair_gtid= `SELECT @@GLOBAL.gtid_binlog_pos`
+
+ANALYZE TABLE t1;
+--let analyze_gtid= `SELECT @@GLOBAL.gtid_binlog_pos`
+
+let $binlog_file= query_get_value(SHOW MASTER STATUS, File, 1);
+FLUSH LOGS;
+
+--let $MYSQLD_DATADIR= `select @@datadir`
+--exec $MYSQL_BINLOG $MYSQLD_DATADIR/$binlog_file > $MYSQLTEST_VARDIR/tmp/mysqlbinlog.out
+
+--let SEARCH_PATTERN= GTID $optimize_gtid ddl
+--let SEARCH_FILE= $MYSQLTEST_VARDIR/tmp/mysqlbinlog.out
+--source include/search_pattern_in_file.inc
+
+--let SEARCH_PATTERN= GTID $repair_gtid ddl
+--let SEARCH_FILE= $MYSQLTEST_VARDIR/tmp/mysqlbinlog.out
+--source include/search_pattern_in_file.inc
+
+--let SEARCH_PATTERN= GTID $analyze_gtid ddl
+--let SEARCH_FILE= $MYSQLTEST_VARDIR/tmp/mysqlbinlog.out
+--source include/search_pattern_in_file.inc
+
+--echo #
+--echo # Clean up
+--echo #
+DROP TABLE t1;
+--remove_file $MYSQLTEST_VARDIR/tmp/mysqlbinlog.out
+--save_master_pos
+
+--connection server_2
+--sync_with_master
+FLUSH LOGS;
+
+--echo #
+--echo # Check that ALTER TABLE commands with ANALYZE, OPTIMIZE and REPAIR on
+--echo # partitions will be marked as DDL in binary log.
+--echo #
+--connection server_1
+CREATE TABLE t1(id INT) PARTITION BY RANGE (id) (PARTITION p0 VALUES LESS THAN (100),
+ PARTITION pmax VALUES LESS THAN (MAXVALUE));
+INSERT INTO t1 VALUES (1), (10), (100), (1000);
+
+ALTER TABLE t1 ANALYZE PARTITION p0;
+--let analyze_gtid= `SELECT @@GLOBAL.gtid_binlog_pos`
+
+ALTER TABLE t1 OPTIMIZE PARTITION p0;
+--let optimize_gtid= `SELECT @@GLOBAL.gtid_binlog_pos`
+
+ALTER TABLE t1 REPAIR PARTITION p0;
+--let repair_gtid= `SELECT @@GLOBAL.gtid_binlog_pos`
+
+let $binlog_file= query_get_value(SHOW MASTER STATUS, File, 1);
+FLUSH LOGS;
+
+--exec $MYSQL_BINLOG $MYSQLD_DATADIR/$binlog_file > $MYSQLTEST_VARDIR/tmp/mysqlbinlog.out
+
+--let SEARCH_PATTERN= GTID $analyze_gtid ddl
+--let SEARCH_FILE= $MYSQLTEST_VARDIR/tmp/mysqlbinlog.out
+--source include/search_pattern_in_file.inc
+
+--let SEARCH_PATTERN= GTID $optimize_gtid ddl
+--let SEARCH_FILE= $MYSQLTEST_VARDIR/tmp/mysqlbinlog.out
+--source include/search_pattern_in_file.inc
+
+--let SEARCH_PATTERN= GTID $repair_gtid ddl
+--let SEARCH_FILE= $MYSQLTEST_VARDIR/tmp/mysqlbinlog.out
+--source include/search_pattern_in_file.inc
+
+--echo #
+--echo # Clean up
+--echo #
+DROP TABLE t1;
+--remove_file $MYSQLTEST_VARDIR/tmp/mysqlbinlog.out
+--save_master_pos
+
+--connection server_2
+--sync_with_master
+
+--source include/stop_slave.inc
+SET GLOBAL slave_parallel_threads= @save_slave_parallel_threads;
+SET GLOBAL slave_parallel_mode= @save_slave_parallel_mode;
+--source include/start_slave.inc
+
+--source include/rpl_end.inc
diff --git a/mysql-test/suite/rpl/t/rpl_slave_shutdown_mdev20821.test b/mysql-test/suite/rpl/t/rpl_slave_shutdown_mdev20821.test
index 563533bb104..6f73de984d3 100644
--- a/mysql-test/suite/rpl/t/rpl_slave_shutdown_mdev20821.test
+++ b/mysql-test/suite/rpl/t/rpl_slave_shutdown_mdev20821.test
@@ -4,6 +4,10 @@
 # hang when the parallel workers were idle.
 # The bug reported scenario is extented to cover the multi-sources case as well as
 # checking is done for both the idle and busy workers cases.
+# 
+# MDEV-25336 Parallel replication causes failed assert while restarting
+#   Since this test case involves slave restart this will help in testing
+#   Mdev-25336 too.
 
 --source include/have_innodb.inc
 --source include/have_binlog_format_mixed.inc
@@ -26,7 +30,7 @@ select @@global.slave_parallel_workers as two;
 
 # At this point worker threads have no assignement.
 # Shutdown must not hang.
-
+# In 10.2/10.3 there should not be any assert failure `prev != 0 && next != 0'
 --connection server_3
 --write_file $MYSQLTEST_VARDIR/tmp/mysqld.3.expect
 wait
@@ -75,6 +79,7 @@ insert into t1 values (1);
 --connection server_3
 --sync_with_master 0,''
 
+# In 10.2/10.3 there should not be any assert failure `prev != 0 && next != 0'
 # At this point worker threads have no assignement.
 # Shutdown must not hang.
 
@@ -117,6 +122,7 @@ insert into t1 values (2);
 insert into t2 values (2);
 
 
+# In 10.2/10.3 there should not be any assert failure `prev != 0 && next != 0'
 # At this point there's a good chance the worker threads are busy.
 # SHUTDOWN must proceed without any delay as above.
 --connection server_3
diff --git a/sql/handler.h b/sql/handler.h
index 5cea91ac1e1..ef643c53df4 100644
--- a/sql/handler.h
+++ b/sql/handler.h
@@ -1876,9 +1876,19 @@ struct THD_TRANS
     CREATED_TEMP_TABLE= 2,
     DROPPED_TEMP_TABLE= 4,
     DID_WAIT= 8,
-    DID_DDL= 0x10
+    DID_DDL= 0x10,
+    EXECUTED_TABLE_ADMIN_CMD= 0x20
   };
 
+  void mark_executed_table_admin_cmd()
+  {
+    DBUG_PRINT("debug", ("mark_executed_table_admin_cmd"));
+    m_unsafe_rollback_flags|= EXECUTED_TABLE_ADMIN_CMD;
+  }
+  bool trans_executed_admin_cmd()
+  {
+    return (m_unsafe_rollback_flags & EXECUTED_TABLE_ADMIN_CMD) != 0;
+  }
   void mark_created_temp_table()
   {
     DBUG_PRINT("debug", ("mark_created_temp_table"));
diff --git a/sql/log_event.h b/sql/log_event.h
index 3e08653a211..6a224b7b6a1 100644
--- a/sql/log_event.h
+++ b/sql/log_event.h
@@ -822,9 +822,8 @@ bool copy_event_cache_to_file_and_reinit(IO_CACHE *cache, FILE *file);
 enum enum_base64_output_mode {
   BASE64_OUTPUT_NEVER= 0,
   BASE64_OUTPUT_AUTO= 1,
-  BASE64_OUTPUT_ALWAYS= 2,
-  BASE64_OUTPUT_UNSPEC= 3,
-  BASE64_OUTPUT_DECODE_ROWS= 4,
+  BASE64_OUTPUT_UNSPEC= 2,
+  BASE64_OUTPUT_DECODE_ROWS= 3,
   /* insert new output modes here */
   BASE64_OUTPUT_MODE_COUNT
 };
diff --git a/sql/log_event_server.cc b/sql/log_event_server.cc
index b4e61e1fbc4..77cab26cdbf 100644
--- a/sql/log_event_server.cc
+++ b/sql/log_event_server.cc
@@ -3262,8 +3262,10 @@ Gtid_log_event::Gtid_log_event(THD *thd_arg, uint64 seq_no_arg,
     flags2|= FL_WAITED;
   if (thd_arg->transaction->stmt.trans_did_ddl() ||
       thd_arg->transaction->stmt.has_created_dropped_temp_table() ||
+      thd_arg->transaction->stmt.trans_executed_admin_cmd() ||
       thd_arg->transaction->all.trans_did_ddl() ||
-      thd_arg->transaction->all.has_created_dropped_temp_table())
+      thd_arg->transaction->all.has_created_dropped_temp_table() ||
+      thd_arg->transaction->all.trans_executed_admin_cmd())
     flags2|= FL_DDL;
   else if (is_transactional && !is_tmp_table)
     flags2|= FL_TRANSACTIONAL;
diff --git a/sql/rpl_rli.cc b/sql/rpl_rli.cc
index 957d440de94..05ba449d96d 100644
--- a/sql/rpl_rli.cc
+++ b/sql/rpl_rli.cc
@@ -210,8 +210,8 @@ a file name for --relay-log-index option", opt_relaylog_index_name);
       */
       sql_print_warning("Neither --relay-log nor --relay-log-index were used;"
                         " so replication "
-                        "may break when this MySQL server acts as a "
-                        "slave and has his hostname changed!! Please "
+                        "may break when this MariaDB server acts as a "
+                        "replica and has its hostname changed. Please "
                         "use '--log-basename=#' or '--relay-log=%s' to avoid "
                         "this problem.", ln);
       name_warning_sent= 1;
diff --git a/sql/sql_admin.cc b/sql/sql_admin.cc
index 31192720610..35682f104d9 100644
--- a/sql/sql_admin.cc
+++ b/sql/sql_admin.cc
@@ -1,5 +1,5 @@
 /* Copyright (c) 2010, 2015, Oracle and/or its affiliates.
-   Copyright (c) 2011, 2020, MariaDB
+   Copyright (c) 2011, 2021, MariaDB
 
    This program is free software; you can redistribute it and/or modify
    it under the terms of the GNU General Public License as published by
@@ -490,7 +490,8 @@ static bool mysql_admin_table(THD* thd, TABLE_LIST* tables,
                               int (handler::*operator_func)(THD *,
                                                             HA_CHECK_OPT *),
                               int (view_operator_func)(THD *, TABLE_LIST*,
-                                                       HA_CHECK_OPT *))
+                                                       HA_CHECK_OPT *),
+                              bool is_cmd_replicated)
 {
   TABLE_LIST *table;
   List<Item> field_list;
@@ -501,6 +502,8 @@ static bool mysql_admin_table(THD* thd, TABLE_LIST* tables,
   int compl_result_code;
   bool need_repair_or_alter= 0;
   wait_for_commit* suspended_wfc;
+  bool is_table_modified= false;
+
   DBUG_ENTER("mysql_admin_table");
   DBUG_PRINT("enter", ("extra_open_options: %u", extra_open_options));
 
@@ -560,6 +563,10 @@ static bool mysql_admin_table(THD* thd, TABLE_LIST* tables,
     bool open_for_modify= org_open_for_modify;
 
     DBUG_PRINT("admin", ("table: '%s'.'%s'", db, table->table_name.str));
+    DEBUG_SYNC(thd, "admin_command_kill_before_modify");
+
+    if (thd->is_killed())
+      break;
     strxmov(table_name, db, ".", table->table_name.str, NullS);
     thd->open_options|= extra_open_options;
     table->lock_type= lock_type;
@@ -1212,6 +1219,13 @@ send_result_message:
         break;
       }
     }
+    /*
+      Admin commands acquire table locks and these locks are not detected by
+      parallel replication deadlock detection-and-handling mechanism. Hence
+      they must be marked as DDL so that they are not scheduled in parallel
+      with conflicting DMLs resulting in deadlock.
+    */
+    thd->transaction->stmt.mark_executed_table_admin_cmd();
     if (table->table && !table->view)
     {
       /*
@@ -1253,10 +1267,9 @@ send_result_message:
     }
     else
     {
-      if (trans_commit_stmt(thd) ||
-          (stmt_causes_implicit_commit(thd, CF_IMPLICIT_COMMIT_END) &&
-           trans_commit_implicit(thd)))
+      if (trans_commit_stmt(thd))
         goto err;
+      is_table_modified= true;
     }
     close_thread_tables(thd);
     thd->release_transactional_locks();
@@ -1279,6 +1292,16 @@ send_result_message:
 
     if (protocol->write())
       goto err;
+    DEBUG_SYNC(thd, "admin_command_kill_after_modify");
+  }
+  if (is_table_modified && is_cmd_replicated &&
+      (!opt_readonly || thd->slave_thread) && !thd->lex->no_write_to_binlog)
+  {
+    thd->get_stmt_da()->set_overwrite_status(true);
+    auto res= write_bin_log(thd, true, thd->query(), thd->query_length());
+    thd->get_stmt_da()->set_overwrite_status(false);
+    if (res)
+      goto err;
   }
 
   my_eof(thd);
@@ -1341,7 +1364,7 @@ bool mysql_assign_to_keycache(THD* thd, TABLE_LIST* tables,
   check_opt.key_cache= key_cache;
   DBUG_RETURN(mysql_admin_table(thd, tables, &check_opt,
 				"assign_to_keycache", TL_READ_NO_INSERT, 0, 0,
-				0, 0, &handler::assign_to_keycache, 0));
+				0, 0, &handler::assign_to_keycache, 0, false));
 }
 
 
@@ -1368,7 +1391,7 @@ bool mysql_preload_keys(THD* thd, TABLE_LIST* tables)
   */
   DBUG_RETURN(mysql_admin_table(thd, tables, 0,
 				"preload_keys", TL_READ_NO_INSERT, 0, 0, 0, 0,
-				&handler::preload_keys, 0));
+				&handler::preload_keys, 0, false));
 }
 
 
@@ -1389,17 +1412,8 @@ bool Sql_cmd_analyze_table::execute(THD *thd)
   WSREP_TO_ISOLATION_BEGIN_WRTCHK(NULL, NULL, first_table);
   res= mysql_admin_table(thd, first_table, &m_lex->check_opt,
                          "analyze", lock_type, 1, 0, 0, 0,
-                         &handler::ha_analyze, 0);
-  /* ! we write after unlocking the table */
-  if (!res && !m_lex->no_write_to_binlog && (!opt_readonly || thd->slave_thread))
-  {
-    /*
-      Presumably, ANALYZE and binlog writing doesn't require synchronization
-    */
-    thd->get_stmt_da()->set_overwrite_status(true);
-    res= write_bin_log(thd, TRUE, thd->query(), thd->query_length());
-    thd->get_stmt_da()->set_overwrite_status(false);
-  }
+                         &handler::ha_analyze, 0, true);
+
   m_lex->first_select_lex()->table_list.first= first_table;
   m_lex->query_tables= first_table;
 
@@ -1425,7 +1439,7 @@ bool Sql_cmd_check_table::execute(THD *thd)
 
   res= mysql_admin_table(thd, first_table, &m_lex->check_opt, "check",
                          lock_type, 0, 0, HA_OPEN_FOR_REPAIR, 0,
-                         &handler::ha_check, &view_check);
+                         &handler::ha_check, &view_check, false);
 
   m_lex->first_select_lex()->table_list.first= first_table;
   m_lex->query_tables= first_table;
@@ -1450,17 +1464,8 @@ bool Sql_cmd_optimize_table::execute(THD *thd)
     mysql_recreate_table(thd, first_table, true) :
     mysql_admin_table(thd, first_table, &m_lex->check_opt,
                       "optimize", TL_WRITE, 1, 0, 0, 0,
-                      &handler::ha_optimize, 0);
-  /* ! we write after unlocking the table */
-  if (!res && !m_lex->no_write_to_binlog && (!opt_readonly || thd->slave_thread))
-  {
-    /*
-      Presumably, OPTIMIZE and binlog writing doesn't require synchronization
-    */
-    thd->get_stmt_da()->set_overwrite_status(true);
-    res= write_bin_log(thd, TRUE, thd->query(), thd->query_length());
-    thd->get_stmt_da()->set_overwrite_status(false);
-  }
+                      &handler::ha_optimize, 0, true);
+
   m_lex->first_select_lex()->table_list.first= first_table;
   m_lex->query_tables= first_table;
 
@@ -1483,18 +1488,8 @@ bool Sql_cmd_repair_table::execute(THD *thd)
                          TL_WRITE, 1,
                          MY_TEST(m_lex->check_opt.sql_flags & TT_USEFRM),
                          HA_OPEN_FOR_REPAIR, &prepare_for_repair,
-                         &handler::ha_repair, &view_repair);
+                         &handler::ha_repair, &view_repair, true);
 
-  /* ! we write after unlocking the table */
-  if (!res && !m_lex->no_write_to_binlog && (!opt_readonly || thd->slave_thread))
-  {
-    /*
-      Presumably, REPAIR and binlog writing doesn't require synchronization
-    */
-    thd->get_stmt_da()->set_overwrite_status(true);
-    res= write_bin_log(thd, TRUE, thd->query(), thd->query_length());
-    thd->get_stmt_da()->set_overwrite_status(false);
-  }
   m_lex->first_select_lex()->table_list.first= first_table;
   m_lex->query_tables= first_table;
 
diff --git a/sql/sql_class.h b/sql/sql_class.h
index 031fff71ec8..bbadaad1dcc 100644
--- a/sql/sql_class.h
+++ b/sql/sql_class.h
@@ -5382,7 +5382,8 @@ public:
     transaction->all.m_unsafe_rollback_flags|=
       (transaction->stmt.m_unsafe_rollback_flags &
        (THD_TRANS::DID_WAIT | THD_TRANS::CREATED_TEMP_TABLE |
-        THD_TRANS::DROPPED_TEMP_TABLE | THD_TRANS::DID_DDL));
+        THD_TRANS::DROPPED_TEMP_TABLE | THD_TRANS::DID_DDL |
+        THD_TRANS::EXECUTED_TABLE_ADMIN_CMD));
   }
 
 
diff --git a/sql/sql_prepare.cc b/sql/sql_prepare.cc
index c3dd0858d64..fc75eab564f 100644
--- a/sql/sql_prepare.cc
+++ b/sql/sql_prepare.cc
@@ -6098,6 +6098,7 @@ extern "C" int execute_sql_command(const char *command,
     new_thd->store_globals();
     new_thd->security_ctx->skip_grants();
     new_thd->query_cache_is_applicable= 0;
+    new_thd->variables.wsrep_on= 0;
     bzero((char*) &new_thd->net, sizeof(new_thd->net));
     thd= new_thd;
   }
diff --git a/storage/innobase/handler/ha_innodb.cc b/storage/innobase/handler/ha_innodb.cc
index 97665da2f1a..8d017480339 100644
--- a/storage/innobase/handler/ha_innodb.cc
+++ b/storage/innobase/handler/ha_innodb.cc
@@ -3866,7 +3866,7 @@ innobase_commit_low(
 	if (trx_is_started(trx)) {
 		trx_commit_for_mysql(trx);
 	} else {
-		trx->will_lock = 0;
+		trx->will_lock = false;
 #ifdef WITH_WSREP
 		trx->wsrep = false;
 #endif /* WITH_WSREP */
@@ -7360,7 +7360,7 @@ ha_innobase::write_row(
 	}
 
 	if (!trx_is_started(trx)) {
-		++trx->will_lock;
+		trx->will_lock = true;
 	}
 
 	ins_mode_t	vers_set_fields;
@@ -8116,7 +8116,7 @@ ha_innobase::update_row(
 	if (is_read_only()) {
 		DBUG_RETURN(HA_ERR_TABLE_READONLY);
 	} else if (!trx_is_started(trx)) {
-		++trx->will_lock;
+		trx->will_lock = true;
 	}
 
 	if (m_upd_buf == NULL) {
@@ -8284,7 +8284,7 @@ ha_innobase::delete_row(
 	if (is_read_only()) {
 		DBUG_RETURN(HA_ERR_TABLE_READONLY);
 	} else if (!trx_is_started(trx)) {
-		++trx->will_lock;
+		trx->will_lock = true;
 	}
 
 	if (!m_prebuilt->upd_node) {
@@ -8554,8 +8554,12 @@ ha_innobase::index_read(
 
 	/* For R-Tree index, we will always place the page lock to
 	pages being searched */
-	if (dict_index_is_spatial(index)) {
-		++m_prebuilt->trx->will_lock;
+	if (index->is_spatial() && !m_prebuilt->trx->will_lock) {
+		if (trx_is_started(m_prebuilt->trx)) {
+			DBUG_RETURN(HA_ERR_READ_ONLY_TRANSACTION);
+		} else {
+			m_prebuilt->trx->will_lock = true;
+		}
 	}
 
 	/* Note that if the index for which the search template is built is not
@@ -9117,7 +9121,7 @@ ha_innobase::ft_init()
 	them as regular read only transactions for now. */
 
 	if (!trx_is_started(trx)) {
-		++trx->will_lock;
+		trx->will_lock = true;
 	}
 
 	DBUG_RETURN(rnd_init(false));
@@ -9183,7 +9187,7 @@ ha_innobase::ft_init_ext(
 	them as regular read only transactions for now. */
 
 	if (!trx_is_started(trx)) {
-		++trx->will_lock;
+		trx->will_lock = true;
 	}
 
 	dict_table_t*	ft_table = m_prebuilt->table;
@@ -12707,8 +12711,7 @@ void
 create_table_info_t::allocate_trx()
 {
 	m_trx = innobase_trx_allocate(m_thd);
-
-	m_trx->will_lock = 1;
+	m_trx->will_lock = true;
 }
 
 /** Create a new table to an InnoDB database.
@@ -13026,13 +13029,7 @@ inline int ha_innobase::delete_table(const char* name, enum_sql_command sqlcom)
 
 	ut_a(name_len < 1000);
 
-	/* Either the transaction is already flagged as a locking transaction
-	or it hasn't been started yet. */
-
-	ut_a(!trx_is_started(trx) || trx->will_lock > 0);
-
-	/* We are doing a DDL operation. */
-	++trx->will_lock;
+	trx->will_lock = true;
 
 	/* Drop the table in InnoDB */
 
@@ -13206,14 +13203,7 @@ innobase_drop_database(
 #endif /* _WIN32 */
 
 	trx_t*	trx = innobase_trx_allocate(thd);
-
-	/* Either the transaction is already flagged as a locking transaction
-	or it hasn't been started yet. */
-
-	ut_a(!trx_is_started(trx) || trx->will_lock > 0);
-
-	/* We are doing a DDL operation. */
-	++trx->will_lock;
+	trx->will_lock = true;
 
 	ulint	dummy;
 
@@ -13250,7 +13240,7 @@ inline dberr_t innobase_rename_table(trx_t *trx, const char *from,
 	DEBUG_SYNC_C("innodb_rename_table_ready");
 
 	trx_start_if_not_started(trx, true);
-	ut_ad(trx->will_lock > 0);
+	ut_ad(trx->will_lock);
 
 	if (commit) {
 		/* Serialize data dictionary operations with dictionary mutex:
@@ -13355,7 +13345,7 @@ int ha_innobase::truncate()
 	const char* name = mem_heap_strdup(heap, ib_table->name.m_name);
 	trx_t*	trx = innobase_trx_allocate(m_user_thd);
 
-	trx->will_lock = 1;
+	trx->will_lock = true;
 	trx->dict_operation = true;
 	row_mysql_lock_data_dictionary(trx);
 	dict_stats_wait_bg_to_stop_using_table(ib_table, trx);
@@ -13442,7 +13432,7 @@ ha_innobase::rename_table(
 	trx_t*	trx = innobase_trx_allocate(thd);
 
 	/* We are doing a DDL operation. */
-	trx->will_lock = 1;
+	trx->will_lock = true;
 	trx->dict_operation = true;
 
 	dberr_t	error = innobase_rename_table(trx, from, to, true);
@@ -15316,7 +15306,7 @@ ha_innobase::start_stmt(
 	innobase_register_trx(ht, thd, trx);
 
 	if (!trx_is_started(trx)) {
-		++trx->will_lock;
+		trx->will_lock = true;
 	}
 
 	DBUG_RETURN(0);
@@ -15561,7 +15551,7 @@ ha_innobase::external_lock(
 		    && (m_prebuilt->select_lock_type != LOCK_NONE
 			|| m_prebuilt->stored_select_lock_type != LOCK_NONE)) {
 
-			++trx->will_lock;
+			trx->will_lock = true;
 		}
 
 		DBUG_RETURN(0);
@@ -15600,7 +15590,7 @@ ha_innobase::external_lock(
 	    && (m_prebuilt->select_lock_type != LOCK_NONE
 		|| m_prebuilt->stored_select_lock_type != LOCK_NONE)) {
 
-		++trx->will_lock;
+		trx->will_lock = true;
 	}
 
 	DBUG_RETURN(0);
@@ -15946,7 +15936,7 @@ ha_innobase::store_lock(
 	    && (m_prebuilt->select_lock_type != LOCK_NONE
 	        || m_prebuilt->stored_select_lock_type != LOCK_NONE)) {
 
-		++trx->will_lock;
+		trx->will_lock = true;
 	}
 
 	return(to);
diff --git a/storage/innobase/handler/handler0alter.cc b/storage/innobase/handler/handler0alter.cc
index 4e85b95c2d1..8694a18f91f 100644
--- a/storage/innobase/handler/handler0alter.cc
+++ b/storage/innobase/handler/handler0alter.cc
@@ -2360,7 +2360,7 @@ innodb_instant_alter_column_allowed_reason:
 		}
 	}
 
-	m_prebuilt->trx->will_lock++;
+	m_prebuilt->trx->will_lock = true;
 
 	/* When changing a NULL column to NOT NULL and specifying a
 	DEFAULT value, ensure that the DEFAULT expression is a constant.
@@ -11382,7 +11382,6 @@ foreign_fail:
 				m_prebuilt = ctx->prebuilt;
 			}
 			trx_start_if_not_started(user_trx, true);
-			user_trx->will_lock++;
 			m_prebuilt->trx = user_trx;
 		}
 		DBUG_INJECT_CRASH("ib_commit_inplace_crash",
diff --git a/storage/innobase/include/trx0i_s.h b/storage/innobase/include/trx0i_s.h
index 5d2ce7adca3..caacfa0972a 100644
--- a/storage/innobase/include/trx0i_s.h
+++ b/storage/innobase/include/trx0i_s.h
@@ -150,7 +150,7 @@ struct i_s_trx_row_t {
 	bool		trx_is_read_only;
 					/*!< trx_t::read_only */
 	bool		trx_is_autocommit_non_locking;
-					/*!< trx_is_autocommit_non_locking(trx)
+					/*!< trx:t::is_autocommit_non_locking()
 					*/
 };
 
diff --git a/storage/innobase/include/trx0sys.h b/storage/innobase/include/trx0sys.h
index c318251eaa9..d687f783db5 100644
--- a/storage/innobase/include/trx0sys.h
+++ b/storage/innobase/include/trx0sys.h
@@ -514,7 +514,7 @@ class rw_trx_hash_t
   static void validate_element(trx_t *trx)
   {
     ut_ad(!trx->read_only || !trx->rsegs.m_redo.rseg);
-    ut_ad(!trx_is_autocommit_non_locking(trx));
+    ut_ad(!trx->is_autocommit_non_locking());
     /* trx->state can be anything except TRX_STATE_NOT_STARTED */
     ut_d(trx->mutex_lock());
     ut_ad(trx_state_eq(trx, TRX_STATE_ACTIVE) ||
diff --git a/storage/innobase/include/trx0trx.h b/storage/innobase/include/trx0trx.h
index 8b0801706cb..1ddd6fe1996 100644
--- a/storage/innobase/include/trx0trx.h
+++ b/storage/innobase/include/trx0trx.h
@@ -332,64 +332,6 @@ from innodb_lock_wait_timeout via trx_t::mysql_thd.
 	 ? thd_lock_wait_timeout((t)->mysql_thd)			\
 	 : 0)
 
-/**
-Determine if the transaction is a non-locking autocommit select
-(implied read-only).
-@param t transaction
-@return true if non-locking autocommit select transaction. */
-#define trx_is_autocommit_non_locking(t)				\
-((t)->auto_commit && (t)->will_lock == 0)
-
-/**
-Determine if the transaction is a non-locking autocommit select
-with an explicit check for the read-only status.
-@param t transaction
-@return true if non-locking autocommit read-only transaction. */
-#define trx_is_ac_nl_ro(t)						\
-((t)->read_only && trx_is_autocommit_non_locking((t)))
-
-/**
-Check transaction state */
-#define check_trx_state(t) do {						\
-	ut_ad(!trx_is_autocommit_non_locking((t)));			\
-	switch ((t)->state) {						\
-	case TRX_STATE_PREPARED:					\
-	case TRX_STATE_PREPARED_RECOVERED:				\
-	case TRX_STATE_ACTIVE:						\
-	case TRX_STATE_COMMITTED_IN_MEMORY:				\
-		continue;						\
-	case TRX_STATE_NOT_STARTED:					\
-		break;							\
-	}								\
-	ut_error;							\
-} while (0)
-
-#ifdef UNIV_DEBUG
-/*******************************************************************//**
-Assert that an autocommit non-locking select cannot be in the
-rw_trx_hash and that it is a read-only transaction.
-The transaction must have mysql_thd assigned. */
-# define assert_trx_nonlocking_or_in_list(t)				\
-	do {								\
-		if (trx_is_autocommit_non_locking(t)) {			\
-			trx_state_t	t_state = (t)->state;		\
-			ut_ad((t)->read_only);				\
-			ut_ad(!(t)->is_recovered);			\
-			ut_ad((t)->mysql_thd);				\
-			ut_ad(t_state == TRX_STATE_NOT_STARTED		\
-			      || t_state == TRX_STATE_ACTIVE);		\
-		} else {						\
-			check_trx_state(t);				\
-		}							\
-	} while (0)
-#else /* UNIV_DEBUG */
-/*******************************************************************//**
-Assert that an autocommit non-locking slect cannot be in the
-rw_trx_hash and that it is a read-only transaction.
-The transaction must have mysql_thd assigned. */
-# define assert_trx_nonlocking_or_in_list(trx) ((void)0)
-#endif /* UNIV_DEBUG */
-
 typedef std::vector<ib_lock_t*, ut_allocator<ib_lock_t*> >	lock_list;
 
 /** The locks and state of an active transaction. Protected by
@@ -882,16 +824,15 @@ public:
 	/*------------------------------*/
 	bool		read_only;	/*!< true if transaction is flagged
 					as a READ-ONLY transaction.
-					if auto_commit && will_lock == 0
+					if auto_commit && !will_lock
 					then it will be handled as a
 					AC-NL-RO-SELECT (Auto Commit Non-Locking
 					Read Only Select). A read only
 					transaction will not be assigned an
 					UNDO log. */
 	bool		auto_commit;	/*!< true if it is an autocommit */
-	ib_uint32_t	will_lock;	/*!< Will acquire some locks. Increment
-					each time we determine that a lock will
-					be acquired by the MySQL layer. */
+	bool		will_lock;	/*!< set to inform trx_start_low() that
+					the transaction may acquire locks */
 	/*------------------------------*/
 	fts_trx_t*	fts_trx;	/*!< FTS information, or NULL if
 					transaction hasn't modified tables
@@ -1062,6 +1003,9 @@ public:
       it->second.end_bulk_insert();
   }
 
+  /** @return whether this is a non-locking autocommit transaction */
+  bool is_autocommit_non_locking() const { return auto_commit && !will_lock; }
+
   /** This has to be invoked on SAVEPOINT or at the start of a statement.
   Even if TRX_UNDO_EMPTY records were written for any table to cover an
   insert into an empty table, subsequent operations will have to be covered
diff --git a/storage/innobase/include/trx0trx.ic b/storage/innobase/include/trx0trx.ic
index 4b566998a90..b063c920e2f 100644
--- a/storage/innobase/include/trx0trx.ic
+++ b/storage/innobase/include/trx0trx.ic
@@ -49,11 +49,15 @@ trx_state_eq(
 	case TRX_STATE_PREPARED:
 	case TRX_STATE_PREPARED_RECOVERED:
 	case TRX_STATE_COMMITTED_IN_MEMORY:
-		ut_ad(!trx_is_autocommit_non_locking(trx));
+		ut_ad(!trx->is_autocommit_non_locking());
 		return(trx->state == state);
 
 	case TRX_STATE_ACTIVE:
-		assert_trx_nonlocking_or_in_list(trx);
+		if (trx->is_autocommit_non_locking()) {
+			ut_ad(!trx->is_recovered);
+			ut_ad(trx->read_only);
+			ut_ad(trx->mysql_thd);
+		}
 		return(state == trx->state);
 
 	case TRX_STATE_NOT_STARTED:
diff --git a/storage/innobase/lock/lock0lock.cc b/storage/innobase/lock/lock0lock.cc
index c65320ab67c..42bd891f342 100644
--- a/storage/innobase/lock/lock0lock.cc
+++ b/storage/innobase/lock/lock0lock.cc
@@ -1097,6 +1097,19 @@ static void lock_reset_lock_and_trx_wait(lock_t *lock)
   lock->type_mode&= ~LOCK_WAIT;
 }
 
+#ifdef UNIV_DEBUG
+/** Check transaction state */
+static void check_trx_state(const trx_t *trx)
+{
+  ut_ad(!trx->auto_commit || trx->will_lock);
+  const auto state= trx->state;
+  ut_ad(state == TRX_STATE_ACTIVE ||
+        state == TRX_STATE_PREPARED_RECOVERED ||
+        state == TRX_STATE_PREPARED ||
+        state == TRX_STATE_COMMITTED_IN_MEMORY);
+}
+#endif
+
 /** Create a new record lock and inserts it to the lock queue,
 without checking for deadlocks or conflicts.
 @param[in]	c_lock		conflicting lock
@@ -1127,7 +1140,7 @@ lock_rec_create_low(
 	ut_ad(dict_index_is_clust(index) || !dict_index_is_online_ddl(index));
 	ut_ad(!(type_mode & LOCK_TABLE));
 	ut_ad(trx->state != TRX_STATE_NOT_STARTED);
-	ut_ad(!trx_is_autocommit_non_locking(trx));
+	ut_ad(!trx->is_autocommit_non_locking());
 
 	/* If rec is the supremum record, then we reset the gap and
 	LOCK_REC_NOT_GAP bits, as all locks on the supremum are
@@ -3083,7 +3096,7 @@ lock_table_create(
 	ut_ad(trx->mutex_is_owner());
 	ut_ad(!trx->is_wsrep() || lock_sys.is_writer());
 	ut_ad(trx->state == TRX_STATE_ACTIVE || trx->is_recovered);
-	ut_ad(!trx_is_autocommit_non_locking(trx));
+	ut_ad(!trx->is_autocommit_non_locking());
 
 	switch (LOCK_MODE_MASK & type_mode) {
 	case LOCK_AUTO_INC:
@@ -4334,7 +4347,8 @@ lock_rec_queue_validate(
 			ut_ad(!index || lock->index == index);
 
 			lock->trx->mutex_lock();
-			ut_ad(!trx_is_ac_nl_ro(lock->trx));
+			ut_ad(!lock->trx->read_only
+			      || !lock->trx->is_autocommit_non_locking());
 			ut_ad(trx_state_eq(lock->trx,
 					   TRX_STATE_COMMITTED_IN_MEMORY)
 			      || !lock->is_waiting()
@@ -4420,8 +4434,8 @@ func_exit:
 	for (lock = lock_sys_t::get_first(cell, id, heap_no);
 	     lock != NULL;
 	     lock = lock_rec_get_next_const(heap_no, lock)) {
-
-		ut_ad(!trx_is_ac_nl_ro(lock->trx));
+		ut_ad(!lock->trx->read_only
+		      || !lock->trx->is_autocommit_non_locking());
 		ut_ad(!page_rec_is_metadata(rec));
 
 		if (index) {
@@ -4497,7 +4511,8 @@ loop:
 		}
 	}
 
-	ut_ad(!trx_is_ac_nl_ro(lock->trx));
+	ut_ad(!lock->trx->read_only
+	      || !lock->trx->is_autocommit_non_locking());
 
 	/* Only validate the record queues when this thread is not
 	holding a tablespace latch. */
@@ -4560,7 +4575,8 @@ lock_rec_validate(
 	     lock != NULL;
 	     lock = static_cast<const lock_t*>(HASH_GET_NEXT(hash, lock))) {
 
-		ut_ad(!trx_is_ac_nl_ro(lock->trx));
+		ut_ad(!lock->trx->read_only
+		      || !lock->trx->is_autocommit_non_locking());
 		ut_ad(!lock->is_table());
 
 		page_id_t current(lock->un_member.rec_lock.page_id);
diff --git a/storage/innobase/read/read0read.cc b/storage/innobase/read/read0read.cc
index be37a9f9e5f..ff34765244e 100644
--- a/storage/innobase/read/read0read.cc
+++ b/storage/innobase/read/read0read.cc
@@ -1,7 +1,7 @@
 /*****************************************************************************
 
 Copyright (c) 1996, 2015, Oracle and/or its affiliates. All Rights Reserved.
-Copyright (c) 2018, 2020, MariaDB Corporation.
+Copyright (c) 2018, 2021, MariaDB Corporation.
 
 This program is free software; you can redistribute it and/or modify it under
 the terms of the GNU General Public License as published by the Free Software
@@ -221,7 +221,7 @@ void ReadView::open(trx_t *trx)
   else if (likely(!srv_read_only_mode))
   {
     m_creator_trx_id= trx->id;
-    if (trx_is_autocommit_non_locking(trx) && empty() &&
+    if (trx->is_autocommit_non_locking() && empty() &&
         low_limit_id() == trx_sys.get_max_trx_id())
       m_open.store(true, std::memory_order_relaxed);
     else
diff --git a/storage/innobase/trx/trx0i_s.cc b/storage/innobase/trx/trx0i_s.cc
index f40ac9fedbe..1232ccc5ad9 100644
--- a/storage/innobase/trx/trx0i_s.cc
+++ b/storage/innobase/trx/trx0i_s.cc
@@ -521,7 +521,7 @@ thd_done:
 
 	row->trx_is_read_only = trx->read_only;
 
-	row->trx_is_autocommit_non_locking = trx_is_autocommit_non_locking(trx);
+	row->trx_is_autocommit_non_locking = trx->is_autocommit_non_locking();
 
 	return(TRUE);
 }
@@ -1165,7 +1165,24 @@ static void fetch_data_into_cache_low(trx_i_s_cache_t *cache, const trx_t *trx)
 {
   i_s_locks_row_t *requested_lock_row;
 
-  assert_trx_nonlocking_or_in_list(trx);
+#ifdef UNIV_DEBUG
+  {
+    const auto state= trx->state;
+
+    if (trx->is_autocommit_non_locking())
+    {
+      ut_ad(trx->read_only);
+      ut_ad(!trx->is_recovered);
+      ut_ad(trx->mysql_thd);
+      ut_ad(state == TRX_STATE_NOT_STARTED || state == TRX_STATE_ACTIVE);
+    }
+    else
+      ut_ad(state == TRX_STATE_ACTIVE ||
+            state == TRX_STATE_PREPARED ||
+            state == TRX_STATE_PREPARED_RECOVERED ||
+            state == TRX_STATE_COMMITTED_IN_MEMORY);
+  }
+#endif /* UNIV_DEBUG */
 
   if (add_trx_relevant_locks_to_cache(cache, trx, &requested_lock_row))
   {
diff --git a/storage/innobase/trx/trx0roll.cc b/storage/innobase/trx/trx0roll.cc
index 0204502e8f5..fbf78393021 100644
--- a/storage/innobase/trx/trx0roll.cc
+++ b/storage/innobase/trx/trx0roll.cc
@@ -99,10 +99,21 @@ inline void trx_t::rollback_low(trx_savept_t *savept)
   roll_node_t *roll_node= roll_node_create(heap);
   roll_node->savept= savept;
 
-  if (savept)
-    check_trx_state(this);
-  else
-    assert_trx_nonlocking_or_in_list(this);
+  ut_ad(!in_rollback);
+#ifdef UNIV_DEBUG
+  {
+    const auto s= state;
+    ut_ad(s == TRX_STATE_ACTIVE ||
+          s == TRX_STATE_PREPARED ||
+          s == TRX_STATE_PREPARED_RECOVERED);
+    if (savept)
+    {
+      ut_ad(s == TRX_STATE_ACTIVE);
+      ut_ad(mysql_thd);
+      ut_ad(!is_recovered);
+    }
+  }
+#endif
 
   error_state = DB_SUCCESS;
 
@@ -198,7 +209,7 @@ dberr_t trx_rollback_for_mysql(trx_t* trx)
 
 	switch (trx->state) {
 	case TRX_STATE_NOT_STARTED:
-		trx->will_lock = 0;
+		trx->will_lock = false;
 		ut_ad(trx->mysql_thd);
 #ifdef WITH_WSREP
 		trx->wsrep= false;
@@ -208,12 +219,13 @@ dberr_t trx_rollback_for_mysql(trx_t* trx)
 
 	case TRX_STATE_ACTIVE:
 		ut_ad(trx->mysql_thd);
-		assert_trx_nonlocking_or_in_list(trx);
+		ut_ad(!trx->is_recovered);
+		ut_ad(!trx->is_autocommit_non_locking() || trx->read_only);
 		return(trx_rollback_for_mysql_low(trx));
 
 	case TRX_STATE_PREPARED:
 	case TRX_STATE_PREPARED_RECOVERED:
-		ut_ad(!trx_is_autocommit_non_locking(trx));
+		ut_ad(!trx->is_autocommit_non_locking());
 		if (trx->rsegs.m_redo.undo || trx->rsegs.m_redo.old_insert) {
 			/* The XA ROLLBACK of a XA PREPARE transaction
 			will consist of multiple mini-transactions.
@@ -260,7 +272,7 @@ dberr_t trx_rollback_for_mysql(trx_t* trx)
 		return(trx_rollback_for_mysql_low(trx));
 
 	case TRX_STATE_COMMITTED_IN_MEMORY:
-		check_trx_state(trx);
+		ut_ad(!trx->is_autocommit_non_locking());
 		break;
 	}
 
@@ -289,7 +301,9 @@ trx_rollback_last_sql_stat_for_mysql(
 		return(DB_SUCCESS);
 
 	case TRX_STATE_ACTIVE:
-		assert_trx_nonlocking_or_in_list(trx);
+		ut_ad(trx->mysql_thd);
+		ut_ad(!trx->is_recovered);
+		ut_ad(!trx->is_autocommit_non_locking() || trx->read_only);
 
 		trx->op_info = "rollback of SQL statement";
 
diff --git a/storage/innobase/trx/trx0trx.cc b/storage/innobase/trx/trx0trx.cc
index dd0b9cbfed8..17b1c5e7fbf 100644
--- a/storage/innobase/trx/trx0trx.cc
+++ b/storage/innobase/trx/trx0trx.cc
@@ -129,7 +129,7 @@ trx_init(
 
 	trx->auto_commit = false;
 
-	trx->will_lock = 0;
+	trx->will_lock = false;
 
 	trx->internal = false;
 
@@ -342,7 +342,7 @@ trx_t *trx_create()
 	ib_alloc_t*	alloc;
 
 	/* We just got trx from pool, it should be non locking */
-	ut_ad(trx->will_lock == 0);
+	ut_ad(!trx->will_lock);
 	ut_ad(!trx->rw_trx_hash_pins);
 
 	DBUG_LOG("trx", "Create: " << trx);
@@ -542,7 +542,7 @@ void trx_disconnect_prepared(trx_t *trx)
   trx->is_recovered= true;
   trx->mysql_thd= NULL;
   /* todo/fixme: suggest to do it at innodb prepare */
-  trx->will_lock= 0;
+  trx->will_lock= false;
   trx_sys.rw_trx_hash.put_pins(trx);
 }
 
@@ -882,11 +882,10 @@ static trx_rseg_t* trx_assign_rseg_low()
 
 /** Assign a rollback segment for modifying temporary tables.
 @return the assigned rollback segment */
-trx_rseg_t*
-trx_t::assign_temp_rseg()
+trx_rseg_t *trx_t::assign_temp_rseg()
 {
 	ut_ad(!rsegs.m_noredo.rseg);
-	ut_ad(!trx_is_autocommit_non_locking(this));
+	ut_ad(!is_autocommit_non_locking());
 	compile_time_assert(ut_is_2pow(TRX_SYS_N_RSEGS));
 
 	/* Choose a temporary rollback segment between 0 and 127
@@ -933,8 +932,8 @@ trx_start_low(
 		    && thd_trx_is_read_only(trx->mysql_thd));
 
 	if (!trx->auto_commit) {
-		++trx->will_lock;
-	} else if (trx->will_lock == 0) {
+		trx->will_lock = true;
+	} else if (!trx->will_lock) {
 		trx->read_only = true;
 	}
 
@@ -972,7 +971,7 @@ trx_start_low(
 
 		trx_sys.register_rw(trx);
 	} else {
-		if (!trx_is_autocommit_non_locking(trx)) {
+		if (!trx->is_autocommit_non_locking()) {
 
 			/* If this is a read-only transaction that is writing
 			to a temporary table then it needs a transaction id
@@ -1267,12 +1266,15 @@ inline void trx_t::commit_in_memory(const mtr_t *mtr)
   must_flush_log_later= false;
   read_view.close();
 
-  if (trx_is_autocommit_non_locking(this))
+  if (is_autocommit_non_locking())
   {
     ut_ad(id == 0);
     ut_ad(read_only);
+    ut_ad(!will_lock);
     ut_a(!is_recovered);
     ut_ad(!rsegs.m_redo.rseg);
+    ut_ad(mysql_thd);
+    ut_ad(state == TRX_STATE_ACTIVE);
 
     /* Note: We do not have to hold any lock_sys latch here, because
     this is a non-locking transaction. */
@@ -1284,12 +1286,11 @@ inline void trx_t::commit_in_memory(const mtr_t *mtr)
     However, the freezing of trx_sys.trx_list will protect the trx_t
     instance and it cannot be removed from the trx_list and freed
     without first unfreezing trx_list. */
-    ut_ad(trx_state_eq(this, TRX_STATE_ACTIVE));
+    state= TRX_STATE_NOT_STARTED;
 
     MONITOR_INC(MONITOR_TRX_NL_RO_COMMIT);
 
     DBUG_LOG("trx", "Autocommit in memory: " << this);
-    state= TRX_STATE_NOT_STARTED;
   }
   else
   {
@@ -1439,8 +1440,6 @@ inline void trx_t::commit_in_memory(const mtr_t *mtr)
 @param mtr  mini-transaction (if there are any persistent modifications) */
 void trx_t::commit_low(mtr_t *mtr)
 {
-  assert_trx_nonlocking_or_in_list(this);
-  ut_ad(!trx_state_eq(this, TRX_STATE_COMMITTED_IN_MEMORY));
   ut_ad(!mtr || mtr->is_active());
   ut_d(bool aborted = in_rollback && error_state == DB_DEADLOCK);
   ut_ad(!mtr == (aborted || !has_logged_or_recovered()));
@@ -1449,13 +1448,13 @@ void trx_t::commit_low(mtr_t *mtr)
   /* undo_no is non-zero if we're doing the final commit. */
   if (fts_trx && undo_no)
   {
-    ut_a(!trx_is_autocommit_non_locking(this));
-    dberr_t error= fts_commit(this);
+    ut_a(!is_autocommit_non_locking());
     /* FTS-FIXME: Temporarily tolerate DB_DUPLICATE_KEY instead of
     dying. This is a possible scenario if there is a crash between
     insert to DELETED table committing and transaction committing. The
     fix would be able to return error from this function */
-    ut_a(error == DB_SUCCESS || error == DB_DUPLICATE_KEY);
+    if (dberr_t error= fts_commit(this))
+      ut_a(error == DB_DUPLICATE_KEY);
   }
 
 #ifndef DBUG_OFF
@@ -1700,9 +1699,6 @@ trx_print_low(
 	ulint		heap_size)
 			/*!< in: mem_heap_get_size(trx->lock.lock_heap) */
 {
-	ibool		newline;
-	const char*	op_info;
-
 	if (const trx_id_t id = trx->id) {
 		fprintf(f, "TRANSACTION " TRX_ID_FMT, trx->id);
 	} else {
@@ -1729,11 +1725,7 @@ trx_print_low(
 	fprintf(f, ", state %lu", (ulong) trx->state);
 	ut_ad(0);
 state_ok:
-
-	/* prevent a race condition */
-	op_info = trx->op_info;
-
-	if (*op_info) {
+	if (const char *op_info = trx->op_info) {
 		putc(' ', f);
 		fputs(op_info, f);
 	}
@@ -1750,18 +1742,18 @@ state_ok:
 			(ulong) trx->mysql_n_tables_locked);
 	}
 
-	newline = TRUE;
+	bool newline = true;
 
 	if (trx->in_rollback) { /* dirty read for performance reasons */
 		fputs("ROLLING BACK ", f);
 	} else if (trx->lock.wait_lock) {
 		fputs("LOCK WAIT ", f);
 	} else {
-		newline = FALSE;
+		newline = false;
 	}
 
 	if (n_trx_locks > 0 || heap_size > 400) {
-		newline = TRUE;
+		newline = true;
 
 		fprintf(f, "%lu lock struct(s), heap size %lu,"
 			" %lu row lock(s)",
@@ -1771,7 +1763,7 @@ state_ok:
 	}
 
 	if (trx->undo_no != 0) {
-		newline = TRUE;
+		newline = true;
 		fprintf(f, ", undo log entries " TRX_ID_FMT, trx->undo_no);
 	}
 
@@ -2150,7 +2142,7 @@ trx_start_internal_low(
 	/* Ensure it is not flagged as an auto-commit-non-locking
 	transaction. */
 
-	trx->will_lock = 1;
+	trx->will_lock = true;
 
 	trx->internal = true;
 
@@ -2166,7 +2158,7 @@ trx_start_internal_read_only_low(
 	/* Ensure it is not flagged as an auto-commit-non-locking
 	transaction. */
 
-	trx->will_lock = 1;
+	trx->will_lock = true;
 
 	trx->internal = true;
 
@@ -2181,8 +2173,6 @@ void trx_start_for_ddl_low(trx_t *trx)
   /* Flag this transaction as a dictionary operation, so that
   the data dictionary will be locked in crash recovery. */
   trx->dict_operation= true;
-  /* Ensure it is not flagged as an auto-commit-non-locking transaction. */
-  trx->will_lock= 1;
   trx_start_internal_low(trx);
 }
 
@@ -2199,7 +2189,7 @@ trx_set_rw_mode(
 	trx_t*		trx)		/*!< in/out: transaction that is RW */
 {
 	ut_ad(trx->rsegs.m_redo.rseg == 0);
-	ut_ad(!trx_is_autocommit_non_locking(trx));
+	ut_ad(!trx->is_autocommit_non_locking());
 	ut_ad(!trx->read_only);
 	ut_ad(trx->id == 0);