Merge 10.2 into 10.3

35 files changed, 355 insertions, 298 deletions

diff --git a/extra/mariabackup/fil_cur.cc b/extra/mariabackup/fil_cur.cc
index 5e64e262166..67d1fb173c0 100644
--- a/extra/mariabackup/fil_cur.cc
+++ b/extra/mariabackup/fil_cur.cc
@@ -338,11 +338,9 @@ static bool page_is_corrupted(const byte *page, ulint page_no,
 
 		memcpy(tmp_page, page, page_size);
 
-		bool decrypted = false;
 		if (!space->crypt_data
 		    || space->crypt_data->type == CRYPT_SCHEME_UNENCRYPTED
-		    || !fil_space_decrypt(space, tmp_frame, tmp_page,
-					  &decrypted)) {
+		    || !fil_space_decrypt(space, tmp_frame, tmp_page)) {
 			return true;
 		}
 
diff --git a/extra/mariabackup/xtrabackup.cc b/extra/mariabackup/xtrabackup.cc
index c14de6a3104..bd8ec97deb7 100644
--- a/extra/mariabackup/xtrabackup.cc
+++ b/extra/mariabackup/xtrabackup.cc
@@ -1528,7 +1528,8 @@ static int prepare_export()
       " --defaults-extra-file=./backup-my.cnf --defaults-group-suffix=%s --datadir=."
       " --innodb --innodb-fast-shutdown=0 --loose-partition"
       " --innodb_purge_rseg_truncate_frequency=1 --innodb-buffer-pool-size=%llu"
-      " --console  --skip-log-error --bootstrap  < "  BOOTSTRAP_FILENAME IF_WIN("\"",""),
+      " --console  --skip-log-error --skip-log-bin --bootstrap  < "
+      BOOTSTRAP_FILENAME IF_WIN("\"",""),
       mariabackup_exe, 
       orig_argv1, (my_defaults_group_suffix?my_defaults_group_suffix:""),
       xtrabackup_use_memory);
@@ -1540,7 +1541,8 @@ static int prepare_export()
       " --defaults-file=./backup-my.cnf --defaults-group-suffix=%s --datadir=."
       " --innodb --innodb-fast-shutdown=0 --loose-partition"
       " --innodb_purge_rseg_truncate_frequency=1 --innodb-buffer-pool-size=%llu"
-      " --console  --log-error= --bootstrap  < "  BOOTSTRAP_FILENAME IF_WIN("\"",""),
+      " --console  --log-error= --skip-log-bin --bootstrap  < "
+      BOOTSTRAP_FILENAME IF_WIN("\"",""),
       mariabackup_exe,
       (my_defaults_group_suffix?my_defaults_group_suffix:""),
       xtrabackup_use_memory);
diff --git a/libmariadb b/libmariadb
-Subproject 544b6f1d12f0e5b2a141129075ff2d64feb0e4c
+Subproject c6403c4c847d94ed6b40d3fd128e729271867e7
diff --git a/mysql-test/std_data/binlog_before_20574.bin b/mysql-test/std_data/binlog_before_20574.bin
new file mode 100644
index 00000000000..596a883dc71
--- /dev/null
+++ b/mysql-test/std_data/binlog_before_20574.bin
diff --git a/mysql-test/suite/binlog_encryption/binlog_mdev_20574_old_binlog.result b/mysql-test/suite/binlog_encryption/binlog_mdev_20574_old_binlog.result
new file mode 100644
index 00000000000..cf660297640
--- /dev/null
+++ b/mysql-test/suite/binlog_encryption/binlog_mdev_20574_old_binlog.result
@@ -0,0 +1,27 @@
+include/master-slave.inc
+[connection master]
+connection slave;
+include/stop_slave.inc
+connection master;
+include/rpl_stop_server.inc [server_number=1]
+# Data in binlog
+# CREATE TABLE t1 (a INT);
+# INSERT INTO t1 VALUES (1),(2),(3);
+# REPLACE INTO t1 VALUES (4);
+include/rpl_start_server.inc [server_number=1]
+connection slave;
+RESET SLAVE;
+RESET MASTER;
+CHANGE MASTER TO master_host='127.0.0.1', master_port=SERVER_MYPORT_1, master_user='root', master_log_file='master-bin.000001', master_log_pos=4;
+include/start_slave.inc
+DESC t1;
+Field	Type	Null	Key	Default	Extra
+a	int(11)	YES		NULL	
+SELECT * FROM t1 ORDER BY a;
+a
+1
+2
+3
+4
+DROP TABLE t1;
+include/rpl_end.inc
diff --git a/mysql-test/suite/binlog_encryption/binlog_mdev_20574_old_binlog.test b/mysql-test/suite/binlog_encryption/binlog_mdev_20574_old_binlog.test
new file mode 100644
index 00000000000..417df631878
--- /dev/null
+++ b/mysql-test/suite/binlog_encryption/binlog_mdev_20574_old_binlog.test
@@ -0,0 +1,46 @@
+# MDEV-20574 Position of events reported by mysqlbinlog is wrong with encrypted binlogs, SHOW BINLOG EVENTS reports the correct one.
+# Test replicating off old master.
+# Test case Desc:-  When new server reads the data from old server binlog which
+# does not send START_ENCRYPTION_EVENT to slave.
+# We simulate old master by copying in pre-generated binlog files from earlier
+# server versions with encrypted binlog.
+--source include/have_binlog_format_row.inc
+--source include/master-slave.inc
+--source include/have_innodb.inc
+
+--connection slave
+--source include/stop_slave.inc
+
+--connection master
+--let $datadir= `SELECT @@datadir`
+
+--let $rpl_server_number= 1
+--source include/rpl_stop_server.inc
+
+--remove_file $datadir/master-bin.000001
+--remove_file $datadir/master-bin.state
+--echo # Data in binlog
+--echo # CREATE TABLE t1 (a INT);
+--echo # INSERT INTO t1 VALUES (1),(2),(3);
+--echo # REPLACE INTO t1 VALUES (4);
+
+--copy_file $MYSQL_TEST_DIR/std_data/binlog_before_20574.bin $datadir/master-bin.000001
+
+--let $rpl_server_number= 1
+--source include/rpl_start_server.inc
+
+--source include/wait_until_connected_again.inc
+--save_master_pos
+
+--connection slave
+RESET SLAVE;
+RESET MASTER;
+--replace_result $SERVER_MYPORT_1 SERVER_MYPORT_1
+eval CHANGE MASTER TO master_host='127.0.0.1', master_port=$SERVER_MYPORT_1, master_user='root', master_log_file='master-bin.000001', master_log_pos=4;
+--source include/start_slave.inc
+--sync_with_master
+DESC t1;
+SELECT * FROM t1 ORDER BY a;
+
+DROP TABLE t1;
+--source include/rpl_end.inc
diff --git a/mysql-test/suite/binlog_encryption/binlog_row_annotate.result b/mysql-test/suite/binlog_encryption/binlog_row_annotate.result
index a6054f00ebc..b783488b07a 100644
--- a/mysql-test/suite/binlog_encryption/binlog_row_annotate.result
+++ b/mysql-test/suite/binlog_encryption/binlog_row_annotate.result
@@ -104,6 +104,9 @@ DELIMITER /*!*/;
 #010909  4:46:40 server id #  end_log_pos # 	Start: binlog v 4, server v #.##.## created 010909  4:46:40 at startup
 ROLLBACK/*!*/;
 # at #
+#010909  4:46:40 server id #  end_log_pos # 	Ignorable
+# Ignorable event type 164 (Start_encryption)
+# at #
 #010909  4:46:40 server id #  end_log_pos # 	Gtid list []
 # at #
 #010909  4:46:40 server id #  end_log_pos # 	Binlog checkpoint master-bin.000001
@@ -342,6 +345,9 @@ DELIMITER /*!*/;
 #010909  4:46:40 server id #  end_log_pos # 	Start: binlog v 4, server v #.##.## created 010909  4:46:40 at startup
 ROLLBACK/*!*/;
 # at #
+#010909  4:46:40 server id #  end_log_pos # 	Ignorable
+# Ignorable event type 164 (Start_encryption)
+# at #
 #010909  4:46:40 server id #  end_log_pos # 	Gtid list []
 # at #
 #010909  4:46:40 server id #  end_log_pos # 	Binlog checkpoint master-bin.000001
@@ -502,6 +508,9 @@ DELIMITER /*!*/;
 #010909  4:46:40 server id #  end_log_pos # 	Start: binlog v 4, server v #.##.## created 010909  4:46:40 at startup
 ROLLBACK/*!*/;
 # at #
+#010909  4:46:40 server id #  end_log_pos # 	Ignorable
+# Ignorable event type 164 (Start_encryption)
+# at #
 #010909  4:46:40 server id #  end_log_pos # 	Gtid list []
 # at #
 #010909  4:46:40 server id #  end_log_pos # 	Binlog checkpoint master-bin.000001
diff --git a/mysql-test/suite/binlog_encryption/mysqlbinlog.result b/mysql-test/suite/binlog_encryption/mysqlbinlog.result
index 71758f7d6e7..e97e0569571 100644
--- a/mysql-test/suite/binlog_encryption/mysqlbinlog.result
+++ b/mysql-test/suite/binlog_encryption/mysqlbinlog.result
@@ -4,3 +4,4 @@ INSERT INTO t1 VALUES (1),(2),(3);
 REPLACE INTO t1 VALUES (4);
 DROP TABLE t1;
 FLUSH LOGS;
+FOUND 1 /Ignorable event type 164.*/ in binlog_enc.sql
diff --git a/mysql-test/suite/binlog_encryption/mysqlbinlog.test b/mysql-test/suite/binlog_encryption/mysqlbinlog.test
index b80388aaa45..108dbd8782f 100644
--- a/mysql-test/suite/binlog_encryption/mysqlbinlog.test
+++ b/mysql-test/suite/binlog_encryption/mysqlbinlog.test
@@ -17,5 +17,8 @@ let outfile=$MYSQLTEST_VARDIR/tmp/binlog_enc.sql;
 exec $MYSQL_BINLOG $local > $outfile;
 exec $MYSQL_BINLOG $local --force-read >> $outfile;
 exec $MYSQL_BINLOG $remote >> $outfile;
+--let SEARCH_FILE= $outfile
+--let SEARCH_PATTERN= Ignorable event type 164.*
+--source include/search_pattern_in_file.inc
 remove_file $outfile;
 
diff --git a/mysql-test/suite/mariabackup/partial.test b/mysql-test/suite/mariabackup/partial.test
index 559ba155972..53388b1947f 100644
--- a/mysql-test/suite/mariabackup/partial.test
+++ b/mysql-test/suite/mariabackup/partial.test
@@ -12,7 +12,7 @@ CREATE TABLE t2(i int) ENGINE INNODB;
 
 echo # xtrabackup backup;
 
-let $targetdir=$MYSQLTEST_VARDIR/tmp/backup;
+let targetdir=$MYSQLTEST_VARDIR/tmp/backup;
 --disable_result_log
 exec $XTRABACKUP --defaults-file=$MYSQLTEST_VARDIR/my.cnf  --backup "--tables=test.*1" --target-dir=$targetdir;
 --enable_result_log
@@ -25,13 +25,27 @@ EOF
 write_file $targetdir/test/junk.frm;
 EOF
 
+let server_cnf=$targetdir/server.cnf;
+copy_file $MYSQLTEST_VARDIR/my.cnf $server_cnf;
+
+# Emulate server config file turnes on binary logs
+perl;
+my $binlog_path="$ENV{'targetdir'}/mysqld-bin";
+my $config_path=$ENV{'server_cnf'};
+open(my $fd, '>>', "$config_path");
+print $fd "\n[mysqld]\n";
+print $fd "log-bin=$binlog_path\n";
+close $fd;
+EOF
 
 echo # xtrabackup prepare;
 --disable_result_log
-exec $XTRABACKUP --defaults-file=$MYSQLTEST_VARDIR/my.cnf --defaults-group-suffix=.1 --prepare --export --target-dir=$targetdir;
+exec $XTRABACKUP --defaults-file=$server_cnf --defaults-group-suffix=.1 --prepare --export --target-dir=$targetdir;
 --enable_result_log
 
 list_files $targetdir/test *.cfg;
+# There must not be binary logs created on --prepare step
+list_files $targetdir/ mysqld-bin.*;
 
 let $MYSQLD_DATADIR= `select @@datadir`;
 ALTER TABLE t1 DISCARD TABLESPACE;
diff --git a/sql-common/client_plugin.c b/sql-common/client_plugin.c
index 4faf05847d8..af86b19f309 100644
--- a/sql-common/client_plugin.c
+++ b/sql-common/client_plugin.c
@@ -28,11 +28,6 @@
   There is no reference counting and no unloading either.
 */
 
-#if defined(_MSC_VER)
-/* Silence warnings about variable 'unused' being used. */
-#define FORCE_INIT_OF_VARS 1
-#endif
-
 #include <my_global.h>
 #include "mysql.h"
 #include <my_sys.h>
@@ -243,12 +238,12 @@ int mysql_client_plugin_init()
   struct st_mysql_client_plugin **builtin;
   va_list unused;
   DBUG_ENTER("mysql_client_plugin_init");
-  LINT_INIT_STRUCT(unused);
 
   if (initialized)
     DBUG_RETURN(0);
 
   bzero(&mysql, sizeof(mysql)); /* dummy mysql for set_mysql_extended_error */
+  bzero(&unused, sizeof unused);
 
   mysql_mutex_init(0, &LOCK_load_client_plugin, MY_MUTEX_INIT_SLOW);
   init_alloc_root(&mem_root, "client_plugin", 128, 128, MYF(0));
@@ -306,9 +301,7 @@ struct st_mysql_client_plugin *
 mysql_client_register_plugin(MYSQL *mysql,
                              struct st_mysql_client_plugin *plugin)
 {
-  va_list unused;
   DBUG_ENTER("mysql_client_register_plugin");
-  LINT_INIT_STRUCT(unused);
 
   if (is_not_initialized(mysql, plugin->name))
     DBUG_RETURN(NULL);
@@ -324,7 +317,11 @@ mysql_client_register_plugin(MYSQL *mysql,
     plugin= NULL;
   }
   else
+  {
+    va_list unused;
+    bzero(&unused, sizeof unused);
     plugin= add_plugin(mysql, plugin, 0, 0, unused);
+  }
 
   mysql_mutex_unlock(&LOCK_load_client_plugin);
   DBUG_RETURN(plugin);
diff --git a/sql/log_event.cc b/sql/log_event.cc
index 15523ba0c36..dc7203fc94f 100644
--- a/sql/log_event.cc
+++ b/sql/log_event.cc
@@ -2096,6 +2096,19 @@ Log_event* Log_event::read_log_event(const char* buf, uint event_len,
          alg != BINLOG_CHECKSUM_ALG_OFF))
       event_len= event_len - BINLOG_CHECKSUM_LEN;
 
+    /*
+      Create an object of Ignorable_log_event for unrecognized sub-class.
+      So that SLAVE SQL THREAD will only update the position and continue.
+      We should look for this flag first instead of judging by event_type
+      Any event can be Ignorable_log_event if it has this flag on.
+      look into @note of Ignorable_log_event
+    */
+    if (uint2korr(buf + FLAGS_OFFSET) & LOG_EVENT_IGNORABLE_F)
+    {
+      ev= new Ignorable_log_event(buf, fdle,
+                                  get_type_str((Log_event_type) event_type));
+      goto exit;
+    }
     switch(event_type) {
     case QUERY_EVENT:
       ev  = new Query_log_event(buf, event_len, fdle, QUERY_EVENT);
@@ -2222,24 +2235,13 @@ Log_event* Log_event::read_log_event(const char* buf, uint event_len,
       ev = new Start_encryption_log_event(buf, event_len, fdle);
       break;
     default:
-      /*
-        Create an object of Ignorable_log_event for unrecognized sub-class.
-        So that SLAVE SQL THREAD will only update the position and continue.
-      */
-      if (uint2korr(buf + FLAGS_OFFSET) & LOG_EVENT_IGNORABLE_F)
-      {
-        ev= new Ignorable_log_event(buf, fdle,
-                                    get_type_str((Log_event_type) event_type));
-      }
-      else
-      {
-        DBUG_PRINT("error",("Unknown event code: %d",
-                            (uchar) buf[EVENT_TYPE_OFFSET]));
-        ev= NULL;
-        break;
-      }
+      DBUG_PRINT("error",("Unknown event code: %d",
+                          (uchar) buf[EVENT_TYPE_OFFSET]));
+      ev= NULL;
+      break;
     }
   }
+exit:
 
   if (ev)
   {
diff --git a/sql/slave.cc b/sql/slave.cc
index 7a9882de507..7d0ce253976 100644
--- a/sql/slave.cc
+++ b/sql/slave.cc
@@ -6741,7 +6741,18 @@ static int queue_event(Master_info* mi,const char* buf, ulong event_len)
           mi->last_queued_gtid.seq_no == 1000)
         goto skip_relay_logging;
     });
+    goto default_action;
 #endif
+  case START_ENCRYPTION_EVENT:
+    if (uint2korr(buf + FLAGS_OFFSET) & LOG_EVENT_IGNORABLE_F)
+    {
+      /*
+         If the event was not requested by the slave (the slave did not ask for
+         it), i.e. has end_log_pos=0, we do not increment mi->master_log_pos
+      */
+      inc_pos= uint4korr(buf+LOG_POS_OFFSET) ? event_len : 0;
+      break;
+    }
     /* fall through */
   default:
   default_action:
diff --git a/sql/sql_base.cc b/sql/sql_base.cc
index 458c921e01b..4bf894798de 100644
--- a/sql/sql_base.cc
+++ b/sql/sql_base.cc
@@ -3751,32 +3751,6 @@ open_and_process_table(THD *thd, TABLE_LIST *tables, uint *counter, uint flags,
     goto end;
   }
 
-  if (get_use_stat_tables_mode(thd) > NEVER && tables->table)
-  {
-    TABLE_SHARE *table_share= tables->table->s;
-    if (table_share && table_share->table_category == TABLE_CATEGORY_USER &&
-        table_share->tmp_table == NO_TMP_TABLE)
-    {
-      if (table_share->stats_cb.stats_can_be_read ||
-	  !alloc_statistics_for_table_share(thd, table_share, FALSE))
-      {
-        if (table_share->stats_cb.stats_can_be_read)
-        {   
-          KEY *key_info= table_share->key_info;
-          KEY *key_info_end= key_info + table_share->keys;
-          KEY *table_key_info= tables->table->key_info;
-          for ( ; key_info < key_info_end; key_info++, table_key_info++)
-            table_key_info->read_stats= key_info->read_stats;
-          Field **field_ptr= table_share->field;
-          Field **table_field_ptr= tables->table->field;
-          for ( ; *field_ptr; field_ptr++, table_field_ptr++)
-            (*table_field_ptr)->read_stats= (*field_ptr)->read_stats;
-          tables->table->stats_is_read= table_share->stats_cb.stats_is_read;
-        }
-      }	
-    }
-  }
-
 process_view_routines:
   /*
     Again we may need cache all routines used by this view and add
diff --git a/sql/sql_class.cc b/sql/sql_class.cc
index 5c778d11c98..3d42d6bedb0 100644
--- a/sql/sql_class.cc
+++ b/sql/sql_class.cc
@@ -850,7 +850,6 @@ THD::THD(my_thread_id id, bool is_wsrep_applier)
   invoker.init();
   prepare_derived_at_open= FALSE;
   create_tmp_table_for_derived= FALSE;
-  force_read_stats= FALSE;
   save_prep_leaf_list= FALSE;
   org_charset= 0;
   /* Restore THR_THD */
diff --git a/sql/sql_class.h b/sql/sql_class.h
index 19f7265b24d..7e7ed42f1aa 100644
--- a/sql/sql_class.h
+++ b/sql/sql_class.h
@@ -2444,9 +2444,6 @@ public:
   */ 
   bool create_tmp_table_for_derived;
 
-  /* The flag to force reading statistics from EITS tables */
-  bool force_read_stats;
-
   bool save_prep_leaf_list;
 
   /* container for handler's private per-connection data */
diff --git a/sql/sql_repl.cc b/sql/sql_repl.cc
index 6e2dbc03c68..2f547707ed5 100644
--- a/sql/sql_repl.cc
+++ b/sql/sql_repl.cc
@@ -427,16 +427,27 @@ static int send_file(THD *thd)
 
 /**
    Internal to mysql_binlog_send() routine that recalculates checksum for
-   a FD event (asserted) that needs additional arranment prior sending to slave.
+   1. FD event (asserted) that needs additional arranment prior sending to slave.
+   2. Start_encryption_log_event whose Ignored flag is set
+TODO DBUG_ASSERT can be removed if this function is used for more general cases
 */
-inline void fix_checksum(String *packet, ulong ev_offset)
+
+inline void fix_checksum(enum_binlog_checksum_alg checksum_alg, String *packet,
+                         ulong ev_offset)
 {
+  if (checksum_alg == BINLOG_CHECKSUM_ALG_OFF ||
+      checksum_alg == BINLOG_CHECKSUM_ALG_UNDEF)
+    return;
   /* recalculate the crc for this event */
   uint data_len = uint4korr(packet->ptr() + ev_offset + EVENT_LEN_OFFSET);
   ha_checksum crc;
-  DBUG_ASSERT(data_len == 
+  DBUG_ASSERT((data_len ==
               LOG_EVENT_MINIMAL_HEADER_LEN + FORMAT_DESCRIPTION_HEADER_LEN +
-              BINLOG_CHECKSUM_ALG_DESC_LEN + BINLOG_CHECKSUM_LEN);
+              BINLOG_CHECKSUM_ALG_DESC_LEN + BINLOG_CHECKSUM_LEN) ||
+              (data_len ==
+              LOG_EVENT_MINIMAL_HEADER_LEN + BINLOG_CRYPTO_SCHEME_LENGTH +
+              BINLOG_KEY_VERSION_LENGTH + BINLOG_NONCE_LENGTH +
+              BINLOG_CHECKSUM_LEN));
   crc= my_checksum(0, (uchar *)packet->ptr() + ev_offset, data_len -
                    BINLOG_CHECKSUM_LEN);
   int4store(packet->ptr() + ev_offset + data_len - BINLOG_CHECKSUM_LEN, crc);
@@ -2169,6 +2180,7 @@ static int send_format_descriptor_event(binlog_send_info *info, IO_CACHE *log,
   THD *thd= info->thd;
   String *packet= info->packet;
   Log_event_type event_type;
+  bool initial_log_pos= info->clear_initial_log_pos;
   DBUG_ENTER("send_format_descriptor_event");
 
   /**
@@ -2267,7 +2279,7 @@ static int send_format_descriptor_event(binlog_send_info *info, IO_CACHE *log,
 
   (*packet)[FLAGS_OFFSET+ev_offset] &= ~LOG_EVENT_BINLOG_IN_USE_F;
 
-  if (info->clear_initial_log_pos)
+  if (initial_log_pos)
   {
     info->clear_initial_log_pos= false;
     /*
@@ -2285,9 +2297,7 @@ static int send_format_descriptor_event(binlog_send_info *info, IO_CACHE *log,
               ST_CREATED_OFFSET+ev_offset, (ulong) 0);
 
     /* fix the checksum due to latest changes in header */
-    if (info->current_checksum_alg != BINLOG_CHECKSUM_ALG_OFF &&
-      info->current_checksum_alg != BINLOG_CHECKSUM_ALG_UNDEF)
-      fix_checksum(packet, ev_offset);
+    fix_checksum(info->current_checksum_alg, packet, ev_offset);
   }
   else if (info->using_gtid_state)
   {
@@ -2308,9 +2318,7 @@ static int send_format_descriptor_event(binlog_send_info *info, IO_CACHE *log,
     {
       int4store((char*) packet->ptr()+LOG_EVENT_MINIMAL_HEADER_LEN+
                 ST_CREATED_OFFSET+ev_offset, (ulong) 0);
-      if (info->current_checksum_alg != BINLOG_CHECKSUM_ALG_OFF &&
-          info->current_checksum_alg != BINLOG_CHECKSUM_ALG_UNDEF)
-        fix_checksum(packet, ev_offset);
+      fix_checksum(info->current_checksum_alg, packet, ev_offset);
     }
   }
 
@@ -2323,12 +2331,16 @@ static int send_format_descriptor_event(binlog_send_info *info, IO_CACHE *log,
   }
 
   /*
-    Read the following Start_encryption_log_event but don't send it to slave.
-    Slave doesn't need to know whether master's binlog is encrypted,
-    and if it'll want to encrypt its logs, it should generate its own
-    random nonce, not use the one from the master.
+    Read the following Start_encryption_log_event and send it to slave as
+    Ignorable_log_event. Although Slave doesn't need to know whether master's
+    binlog is encrypted but it needs to update slave log pos (for mysqlbinlog).
+
+    If slave want to encrypt its logs, it should generate its own
+    random nonce, it should not use the one from the master.
   */
-  packet->length(0);
+  /* reset transmit packet for the event read from binary log file */
+  if (reset_transmit_packet(info, info->flags, &ev_offset, &info->errmsg))
+    DBUG_RETURN(1);
   info->last_pos= linfo->pos;
   error= Log_event::read_log_event(log, packet, info->fdev,
                                    opt_master_verify_checksum
@@ -2342,12 +2354,13 @@ static int send_format_descriptor_event(binlog_send_info *info, IO_CACHE *log,
     DBUG_RETURN(1);
   }
 
-  event_type= (Log_event_type)((uchar)(*packet)[LOG_EVENT_OFFSET]);
+  event_type= (Log_event_type)((uchar)(*packet)[LOG_EVENT_OFFSET + ev_offset]);
   if (event_type == START_ENCRYPTION_EVENT)
   {
     Start_encryption_log_event *sele= (Start_encryption_log_event *)
-      Log_event::read_log_event(packet->ptr(), packet->length(), &info->errmsg,
-                                info->fdev, BINLOG_CHECKSUM_ALG_OFF);
+      Log_event::read_log_event(packet->ptr() + ev_offset, packet->length()
+                                - ev_offset, &info->errmsg, info->fdev,
+                                BINLOG_CHECKSUM_ALG_OFF);
     if (!sele)
     {
       info->error= ER_MASTER_FATAL_ERROR_READING_BINLOG;
@@ -2361,6 +2374,18 @@ static int send_format_descriptor_event(binlog_send_info *info, IO_CACHE *log,
       delete sele;
       DBUG_RETURN(1);
     }
+    /* Make it Ignorable_log_event and send it */
+    (*packet)[FLAGS_OFFSET+ev_offset] |= LOG_EVENT_IGNORABLE_F;
+    if (initial_log_pos)
+      int4store((char*) packet->ptr()+LOG_POS_OFFSET+ev_offset, (ulong) 0);
+    /* fix the checksum due to latest changes in header */
+    fix_checksum(info->current_checksum_alg, packet, ev_offset);
+    if (my_net_write(info->net, (uchar*) packet->ptr(), packet->length()))
+    {
+      info->errmsg= "Failed on my_net_write()";
+      info->error= ER_UNKNOWN_ERROR;
+      DBUG_RETURN(1);
+    }
     delete sele;
   }
   else if (start_pos == BIN_LOG_HEADER_SIZE)
diff --git a/sql/sql_show.cc b/sql/sql_show.cc
index d551ae4761e..deb1feb4f08 100644
--- a/sql/sql_show.cc
+++ b/sql/sql_show.cc
@@ -4614,10 +4614,8 @@ fill_schema_table_by_open(THD *thd, MEM_ROOT *mem_root,
   }
 
   DBUG_ASSERT(thd->lex == lex);
-  thd->force_read_stats= get_schema_table_idx(schema_table) == SCH_STATISTICS;
   result= open_tables_only_view_structure(thd, table_list, can_deadlock);
   (void) read_statistics_for_tables_if_needed(thd, table_list);
-  thd->force_read_stats= false;
 
   DEBUG_SYNC(thd, "after_open_table_ignore_flush");
 
@@ -6607,6 +6605,7 @@ static int get_schema_stat_record(THD *thd, TABLE_LIST *tables,
     KEY *key_info=show_table->s->key_info;
     if (show_table->file)
     {
+      (void) read_statistics_for_tables(thd, tables);
       show_table->file->info(HA_STATUS_VARIABLE |
                              HA_STATUS_NO_LOCK |
                              HA_STATUS_TIME);
diff --git a/sql/sql_statistics.cc b/sql/sql_statistics.cc
index 50f7446f38a..685e2162d81 100644
--- a/sql/sql_statistics.cc
+++ b/sql/sql_statistics.cc
@@ -2170,54 +2170,6 @@ int alloc_statistics_for_table(THD* thd, TABLE *table)
 
 
 /**
-  @brief
-  Check whether any persistent statistics for the processed command is needed
-
-  @param
-  thd         The thread handle
-
-  @details
-  The function checks whether any persitent statistics for the processed
-  command is needed to be read.
-
-  @retval
-  TRUE        statistics is needed to be read 
-  @retval
-  FALSE       Otherwise
-*/
-
-static
-inline bool statistics_for_command_is_needed(THD *thd)
-{
-  if (thd->bootstrap || thd->variables.use_stat_tables == NEVER)
-    return FALSE;
-
-  if (thd->force_read_stats)
-    return TRUE;
-
-  switch(thd->lex->sql_command) {
-  case SQLCOM_SELECT:
-  case SQLCOM_INSERT:
-  case SQLCOM_INSERT_SELECT:
-  case SQLCOM_UPDATE:
-  case SQLCOM_UPDATE_MULTI:
-  case SQLCOM_DELETE:
-  case SQLCOM_DELETE_MULTI:
-  case SQLCOM_REPLACE:
-  case SQLCOM_REPLACE_SELECT:
-  case SQLCOM_CREATE_TABLE:
-  case SQLCOM_SET_OPTION:
-  case SQLCOM_DO:
-    break;
-  default: 
-    return FALSE;
-  }
-
-  return TRUE;
-} 
-
-
-/**
   @brief 
   Allocate memory for the statistical data used by a table share
 
@@ -2225,8 +2177,6 @@ inline bool statistics_for_command_is_needed(THD *thd)
   thd         Thread handler
   @param
   table_share Table share for which the memory for statistical data is allocated
-  @param
-  is_safe     TRUE <-> at any time only one thread can perform the function
 
   @note
   The function allocates the memory for the statistical data on a table in the
@@ -2235,8 +2185,6 @@ inline bool statistics_for_command_is_needed(THD *thd)
   mysql.index_stats. The memory is allocated for the statistics on the table,
   on the tables's columns, and on the table's indexes. The memory is allocated
   in the table_share's mem_root.
-  If the parameter is_safe is TRUE then it is guaranteed that at any given time
-  only one thread is executed the code of the function.
 
   @retval
   0     If the memory for all statistical data has been successfully allocated  
@@ -2255,16 +2203,10 @@ inline bool statistics_for_command_is_needed(THD *thd)
   Here the second and the third threads try to allocate the memory for
   statistical data at the same time. The precautions are taken to
   guarantee the correctness of the allocation.
-
-  @note
-  Currently the function always is called with the parameter is_safe set
-  to FALSE. 
 */      
 
-int alloc_statistics_for_table_share(THD* thd, TABLE_SHARE *table_share, 
-                                     bool is_safe)
+static int alloc_statistics_for_table_share(THD* thd, TABLE_SHARE *table_share)
 {
-  
   Field **field_ptr;
   KEY *key_info, *end;
   TABLE_STATISTICS_CB *stats_cb= &table_share->stats_cb;
@@ -2274,16 +2216,11 @@ int alloc_statistics_for_table_share(THD* thd, TABLE_SHARE *table_share,
   DEBUG_SYNC(thd, "statistics_mem_alloc_start1");
   DEBUG_SYNC(thd, "statistics_mem_alloc_start2");
 
-  if (!statistics_for_command_is_needed(thd))
-    DBUG_RETURN(1);
-
-  if (!is_safe)
-    mysql_mutex_lock(&table_share->LOCK_share);
+  mysql_mutex_lock(&table_share->LOCK_share);
 
   if (stats_cb->stats_can_be_read)
   {
-    if (!is_safe)
-      mysql_mutex_unlock(&table_share->LOCK_share);
+    mysql_mutex_unlock(&table_share->LOCK_share);
     DBUG_RETURN(0);
   }
 
@@ -2294,8 +2231,7 @@ int alloc_statistics_for_table_share(THD* thd, TABLE_SHARE *table_share,
                                                   sizeof(Table_statistics));
     if (!table_stats)
     {
-      if (!is_safe)
-        mysql_mutex_unlock(&table_share->LOCK_share);
+      mysql_mutex_unlock(&table_share->LOCK_share);
       DBUG_RETURN(1);
     }
     memset(table_stats, 0, sizeof(Table_statistics));
@@ -2367,8 +2303,7 @@ int alloc_statistics_for_table_share(THD* thd, TABLE_SHARE *table_share,
   if (column_stats && index_stats && idx_avg_frequency)
     stats_cb->stats_can_be_read= TRUE;
 
-  if (!is_safe)
-    mysql_mutex_unlock(&table_share->LOCK_share);
+  mysql_mutex_unlock(&table_share->LOCK_share);
 
   DBUG_RETURN(0);
 }
@@ -3138,9 +3073,6 @@ bool statistics_for_tables_is_needed(THD *thd, TABLE_LIST *tables)
 {
   if (!tables)
     return FALSE;
-  
-  if (!statistics_for_command_is_needed(thd))
-    return FALSE;
 
   /* 
     Do not read statistics for any query that explicity involves
@@ -3273,10 +3205,64 @@ int read_histograms_for_table(THD *thd, TABLE *table, TABLE_LIST *stat_tables)
 
 int read_statistics_for_tables_if_needed(THD *thd, TABLE_LIST *tables)
 {
+  switch (thd->lex->sql_command) {
+  case SQLCOM_SELECT:
+  case SQLCOM_INSERT:
+  case SQLCOM_INSERT_SELECT:
+  case SQLCOM_UPDATE:
+  case SQLCOM_UPDATE_MULTI:
+  case SQLCOM_DELETE:
+  case SQLCOM_DELETE_MULTI:
+  case SQLCOM_REPLACE:
+  case SQLCOM_REPLACE_SELECT:
+  case SQLCOM_CREATE_TABLE:
+  case SQLCOM_SET_OPTION:
+  case SQLCOM_DO:
+    return read_statistics_for_tables(thd, tables);
+  default:
+    return 0;
+  }
+}
+
+
+int read_statistics_for_tables(THD *thd, TABLE_LIST *tables)
+{
   TABLE_LIST stat_tables[STATISTICS_TABLES];
   Open_tables_backup open_tables_backup;
 
-  DBUG_ENTER("read_statistics_for_tables_if_needed");
+  DBUG_ENTER("read_statistics_for_tables");
+
+  if (thd->bootstrap || thd->variables.use_stat_tables == NEVER)
+    DBUG_RETURN(0);
+
+  for (TABLE_LIST *tl= tables; tl; tl= tl->next_global)
+  {
+    if (tl->table)
+    {
+      TABLE_SHARE *table_share= tl->table->s;
+      if (table_share && table_share->table_category == TABLE_CATEGORY_USER &&
+          table_share->tmp_table == NO_TMP_TABLE)
+      {
+        if (table_share->stats_cb.stats_can_be_read ||
+            !alloc_statistics_for_table_share(thd, table_share))
+        {
+          if (table_share->stats_cb.stats_can_be_read)
+          {
+            KEY *key_info= table_share->key_info;
+            KEY *key_info_end= key_info + table_share->keys;
+            KEY *table_key_info= tl->table->key_info;
+            for ( ; key_info < key_info_end; key_info++, table_key_info++)
+              table_key_info->read_stats= key_info->read_stats;
+            Field **field_ptr= table_share->field;
+            Field **table_field_ptr= tl->table->field;
+            for ( ; *field_ptr; field_ptr++, table_field_ptr++)
+              (*table_field_ptr)->read_stats= (*field_ptr)->read_stats;
+            tl->table->stats_is_read= table_share->stats_cb.stats_is_read;
+          }
+        }
+      }
+    }
+  }
 
   DEBUG_SYNC(thd, "statistics_read_start");
 
diff --git a/sql/sql_statistics.h b/sql/sql_statistics.h
index b8df647ee62..a69dd340dd7 100644
--- a/sql/sql_statistics.h
+++ b/sql/sql_statistics.h
@@ -89,9 +89,8 @@ Use_stat_tables_mode get_use_stat_tables_mode(THD *thd)
 }
 
 int read_statistics_for_tables_if_needed(THD *thd, TABLE_LIST *tables);
+int read_statistics_for_tables(THD *thd, TABLE_LIST *tables);
 int collect_statistics_for_table(THD *thd, TABLE *table);
-int alloc_statistics_for_table_share(THD* thd, TABLE_SHARE *share,
-                                     bool is_safe);
 void delete_stat_values_for_table_share(TABLE_SHARE *table_share);
 int alloc_statistics_for_table(THD *thd, TABLE *table);
 int update_statistics_for_table(THD *thd, TABLE *table);
diff --git a/storage/connect/inihandl.cpp b/storage/connect/inihandl.cpp
index 8e79aeac7ef..dacab3c485c 100644
--- a/storage/connect/inihandl.cpp
+++ b/storage/connect/inihandl.cpp
@@ -194,7 +194,7 @@ static void PROFILE_Save( FILE *file, PROFILESECTION *section )
     }
 
     for (key = section->key; key; key = key->next)
-      if (key->name && key->name[0]) {
+      if (key->name[0]) {
         fprintf(file, "%s", SVP(key->name));
 
         if (key->value)
diff --git a/storage/innobase/btr/btr0btr.cc b/storage/innobase/btr/btr0btr.cc
index e0fce3fc132..6f1da87c933 100644
--- a/storage/innobase/btr/btr0btr.cc
+++ b/storage/innobase/btr/btr0btr.cc
@@ -3314,25 +3314,22 @@ btr_parse_set_min_rec_mark(
 	return(ptr + 2);
 }
 
-/****************************************************************//**
-Sets a record as the predefined minimum record. */
-void
-btr_set_min_rec_mark(
-/*=================*/
-	rec_t*	rec,	/*!< in: record */
-	mtr_t*	mtr)	/*!< in: mtr */
+/** Sets a record as the predefined minimum record. */
+void btr_set_min_rec_mark(rec_t* rec, mtr_t* mtr)
 {
-	ulint	info_bits;
+	const bool comp = page_rec_is_comp(rec);
 
-	if (page_rec_is_comp(rec)) {
-		info_bits = rec_get_info_bits(rec, TRUE);
+	ut_ad(rec == page_rec_get_next_const(page_get_infimum_rec(
+						     page_align(rec))));
+	ut_ad(!(rec_get_info_bits(page_rec_get_next(rec), comp)
+		& REC_INFO_MIN_REC_FLAG));
 
+	size_t info_bits = rec_get_info_bits(rec, comp);
+	if (comp) {
 		rec_set_info_bits_new(rec, info_bits | REC_INFO_MIN_REC_FLAG);
 
 		btr_set_min_rec_mark_log(rec, MLOG_COMP_REC_MIN_MARK, mtr);
 	} else {
-		info_bits = rec_get_info_bits(rec, FALSE);
-
 		rec_set_info_bits_old(rec, info_bits | REC_INFO_MIN_REC_FLAG);
 
 		btr_set_min_rec_mark_log(rec, MLOG_REC_MIN_MARK, mtr);
diff --git a/storage/innobase/btr/btr0cur.cc b/storage/innobase/btr/btr0cur.cc
index 99a15662e3f..52f47c4c2c4 100644
--- a/storage/innobase/btr/btr0cur.cc
+++ b/storage/innobase/btr/btr0cur.cc
@@ -5675,6 +5675,9 @@ btr_cur_pessimistic_delete(
 #endif /* UNIV_ZIP_DEBUG */
 	}
 
+	rec_t* next_rec = NULL;
+	bool min_mark_next_rec = false;
+
 	if (page_is_leaf(page)) {
 		const bool is_metadata = rec_get_info_bits(
 			rec, page_rec_is_comp(rec)) & REC_INFO_MIN_REC_FLAG;
@@ -5748,20 +5751,14 @@ discard_page:
 			goto return_after_reservations;
 		}
 
-		rec_t*	next_rec = page_rec_get_next(rec);
+		next_rec = page_rec_get_next(rec);
 
 		if (!page_has_prev(page)) {
-
 			/* If we delete the leftmost node pointer on a
 			non-leaf level, we must mark the new leftmost node
 			pointer as the predefined minimum record */
 
-			/* This will make page_zip_validate() fail until
-			page_cur_delete_rec() completes.  This is harmless,
-			because everything will take place within a single
-			mini-transaction and because writing to the redo log
-			is an atomic operation (performed by mtr_commit()). */
-			btr_set_min_rec_mark(next_rec, mtr);
+			min_mark_next_rec = true;
 		} else if (dict_index_is_spatial(index)) {
 			/* For rtree, if delete the leftmost node pointer,
 			we need to update parent page. */
@@ -5829,6 +5826,11 @@ discard_page:
 				block->page.size, mtr);
 		page_cur_delete_rec(btr_cur_get_page_cur(cursor), index,
 				    offsets, mtr);
+
+		if (min_mark_next_rec) {
+			btr_set_min_rec_mark(next_rec, mtr);
+		}
+
 #ifdef UNIV_ZIP_DEBUG
 		ut_a(!page_zip || page_zip_validate(page_zip, page, index));
 #endif /* UNIV_ZIP_DEBUG */
diff --git a/storage/innobase/buf/buf0buf.cc b/storage/innobase/buf/buf0buf.cc
index b8ee5140dd0..6b46cc7097b 100644
--- a/storage/innobase/buf/buf0buf.cc
+++ b/storage/innobase/buf/buf0buf.cc
@@ -589,12 +589,6 @@ decrypt_failed:
 				    << mach_read_from_4(
 					    FIL_PAGE_FILE_FLUSH_LSN_OR_KEY_VERSION
 					    + dst_frame);
-			/* Mark page encrypted in case it should be. */
-			if (space->crypt_data->type
-			    != CRYPT_SCHEME_UNENCRYPTED) {
-				bpage->encrypted = true;
-			}
-
 			return false;
 		}
 
@@ -605,8 +599,7 @@ decrypt_failed:
 		ut_d(fil_page_type_validate(dst_frame));
 
 		/* decrypt using crypt_buf to dst_frame */
-		if (!fil_space_decrypt(space, slot->crypt_buf,
-				       dst_frame, &bpage->encrypted)) {
+		if (!fil_space_decrypt(space, slot->crypt_buf, dst_frame)) {
 			slot->release();
 			goto decrypt_failed;
 		}
@@ -1573,7 +1566,6 @@ buf_block_init(
 	block->page.buf_fix_count = 0;
 	block->page.io_fix = BUF_IO_NONE;
 	block->page.flush_observer = NULL;
-	block->page.encrypted = false;
 	block->page.real_size = 0;
 	block->page.write_size = 0;
 	block->modify_clock = 0;
@@ -4087,7 +4079,6 @@ err_exit:
 	if (encrypted) {
 		ib::info() << "Row compressed page could be encrypted"
 			" with key_version " << key_version;
-		block->page.encrypted = true;
 	}
 
 	if (space) {
@@ -5293,7 +5284,6 @@ buf_page_init_low(
 	bpage->newest_modification = 0;
 	bpage->oldest_modification = 0;
 	bpage->write_size = 0;
-	bpage->encrypted = false;
 	bpage->real_size = 0;
 	bpage->slot = NULL;
 
@@ -5887,17 +5877,19 @@ buf_page_monitor(
 }
 
 /** Mark a table corrupted.
-Also remove the bpage from LRU list.
-@param[in]	bpage	Corrupted page. */
-static void buf_mark_space_corrupt(buf_page_t* bpage, const fil_space_t* space)
+@param[in]	bpage	corrupted page
+@param[in]	space	tablespace of the corrupted page */
+ATTRIBUTE_COLD
+static void buf_mark_space_corrupt(buf_page_t* bpage, const fil_space_t& space)
 {
 	/* If block is not encrypted find the table with specified
 	space id, and mark it corrupted. Encrypted tables
 	are marked unusable later e.g. in ::open(). */
-	if (!bpage->encrypted) {
-		dict_set_corrupted_by_space(space);
+	if (!space.crypt_data
+	    || space.crypt_data->type == CRYPT_SCHEME_UNENCRYPTED) {
+		dict_set_corrupted_by_space(&space);
 	} else {
-		dict_set_encrypted_by_space(space);
+		dict_set_encrypted_by_space(&space);
 	}
 }
 
@@ -5936,7 +5928,7 @@ buf_corrupt_page_release(buf_page_t* bpage, const fil_space_t* space)
 	mutex_exit(buf_page_get_mutex(bpage));
 
 	if (!srv_force_recovery) {
-		buf_mark_space_corrupt(bpage, space);
+		buf_mark_space_corrupt(bpage, *space);
 	}
 
 	/* After this point bpage can't be referenced. */
@@ -5966,7 +5958,6 @@ static dberr_t buf_page_check_corrupt(buf_page_t* bpage, fil_space_t* space)
 	byte* dst_frame = (bpage->zip.data) ? bpage->zip.data :
 		((buf_block_t*) bpage)->frame;
 	dberr_t err = DB_SUCCESS;
-	bool corrupted = false;
 
 	/* In buf_decrypt_after_read we have either decrypted the page if
 	page post encryption checksum matches and used key_id is found
@@ -5974,33 +5965,20 @@ static dberr_t buf_page_check_corrupt(buf_page_t* bpage, fil_space_t* space)
 	not decrypted and it could be either encrypted and corrupted
 	or corrupted or good page. If we decrypted, there page could
 	still be corrupted if used key does not match. */
-	const bool still_encrypted = mach_read_from_4(
+	const bool seems_encrypted = mach_read_from_4(
 		dst_frame + FIL_PAGE_FILE_FLUSH_LSN_OR_KEY_VERSION)
 		&& space->crypt_data
-		&& space->crypt_data->type != CRYPT_SCHEME_UNENCRYPTED
-		&& !bpage->encrypted
-		&& fil_space_verify_crypt_checksum(dst_frame, bpage->size);
-
-	if (!still_encrypted) {
-		/* If traditional checksums match, we assume that page is
-		not anymore encrypted. */
-		corrupted = buf_page_is_corrupted(
-			true, dst_frame, bpage->size, space);
-
-		if (!corrupted) {
-			bpage->encrypted = false;
-		} else {
-			err = DB_PAGE_CORRUPTED;
-		}
+		&& space->crypt_data->type != CRYPT_SCHEME_UNENCRYPTED;
+
+	/* If traditional checksums match, we assume that page is
+	not anymore encrypted. */
+	if (buf_page_is_corrupted(
+		true, dst_frame, bpage->size, space)) {
+		err = DB_PAGE_CORRUPTED;
 	}
 
-	/* Pages that we think are unencrypted but do not match the checksum
-	checks could be corrupted or encrypted or both. */
-	if (corrupted && !bpage->encrypted) {
-		/* An error will be reported by
-		buf_page_io_complete(). */
-	} else if (still_encrypted || (bpage->encrypted && corrupted)) {
-		bpage->encrypted = true;
+	if (seems_encrypted && err == DB_PAGE_CORRUPTED
+	    && bpage->id.page_no() != 0) {
 		err = DB_DECRYPTION_FAILED;
 
 		ib::error()
@@ -6062,7 +6040,6 @@ buf_page_io_complete(buf_page_t* bpage, bool dblwr, bool evict)
 	if (io_type == BUF_IO_READ) {
 		ulint	read_page_no = 0;
 		ulint	read_space_id = 0;
-		uint	key_version = 0;
 		byte*	frame = bpage->zip.data
 			? bpage->zip.data
 			: reinterpret_cast<buf_block_t*>(bpage)->frame;
@@ -6102,8 +6079,6 @@ buf_page_io_complete(buf_page_t* bpage, bool dblwr, bool evict)
 		read_page_no = mach_read_from_4(frame + FIL_PAGE_OFFSET);
 		read_space_id = mach_read_from_4(
 			frame + FIL_PAGE_ARCH_LOG_NO_OR_SPACE_ID);
-		key_version = mach_read_from_4(
-			frame + FIL_PAGE_FILE_FLUSH_LSN_OR_KEY_VERSION);
 
 		if (bpage->id.space() == TRX_SYS_SPACE
 		    && buf_dblwr_page_inside(bpage->id.page_no())) {
@@ -6219,23 +6194,9 @@ database_corrupted:
 		    && fil_page_get_type(frame) == FIL_PAGE_INDEX
 		    && page_is_leaf(frame)) {
 
-			if (bpage->encrypted) {
-				ib::warn()
-					<< "Table in tablespace "
-					<< bpage->id.space()
-					<< " encrypted. However key "
-					"management plugin or used "
-					<< "key_version " << key_version
-					<< " is not found or"
-					" used encryption algorithm or method does not match."
-					" Can't continue opening the table.";
-			} else {
-
-				ibuf_merge_or_delete_for_page(
-					(buf_block_t*) bpage, bpage->id,
-					&bpage->size, TRUE);
-			}
-
+			ibuf_merge_or_delete_for_page(
+				(buf_block_t*) bpage, bpage->id,
+				&bpage->size, TRUE);
 		}
 
 		space->release_for_io();
diff --git a/storage/innobase/fil/fil0crypt.cc b/storage/innobase/fil/fil0crypt.cc
index 878e65deef5..19e89125a3c 100644
--- a/storage/innobase/fil/fil0crypt.cc
+++ b/storage/innobase/fil/fil0crypt.cc
@@ -792,7 +792,6 @@ Decrypt a page.
 @param[in]	space			Tablespace
 @param[in]	tmp_frame		Temporary buffer used for decrypting
 @param[in,out]	src_frame		Page to decrypt
-@param[out]	decrypted		true if page was decrypted
 @return decrypted page, or original not encrypted page if decryption is
 not needed.*/
 UNIV_INTERN
@@ -800,13 +799,11 @@ byte*
 fil_space_decrypt(
 	const fil_space_t* space,
 	byte*		tmp_frame,
-	byte*		src_frame,
-	bool*		decrypted)
+	byte*		src_frame)
 {
 	dberr_t err = DB_SUCCESS;
 	byte* res = NULL;
 	const page_size_t page_size(space->flags);
-	*decrypted = false;
 
 	ut_ad(space->crypt_data != NULL && space->crypt_data->is_encrypted());
 	ut_ad(space->pending_io());
@@ -816,7 +813,6 @@ fil_space_decrypt(
 
 	if (err == DB_SUCCESS) {
 		if (encrypted) {
-			*decrypted = true;
 			/* Copy the decrypted page back to page buffer, not
 			really any other options. */
 			memcpy(src_frame, tmp_frame, page_size.physical());
diff --git a/storage/innobase/ibuf/ibuf0ibuf.cc b/storage/innobase/ibuf/ibuf0ibuf.cc
index fe1586b4b9f..4dbd9f8ff1a 100644
--- a/storage/innobase/ibuf/ibuf0ibuf.cc
+++ b/storage/innobase/ibuf/ibuf0ibuf.cc
@@ -336,10 +336,8 @@ ibuf_header_page_get(
 		page_id_t(IBUF_SPACE_ID, FSP_IBUF_HEADER_PAGE_NO),
 		univ_page_size, RW_X_LATCH, mtr);
 
-
-	if (!block->page.encrypted) {
+	if (block) {
 		buf_block_dbg_add_level(block, SYNC_IBUF_HEADER);
-
 		page = buf_block_get_frame(block);
 	}
 
diff --git a/storage/innobase/include/btr0btr.h b/storage/innobase/include/btr0btr.h
index 6754fb04469..2c808d4bd06 100644
--- a/storage/innobase/include/btr0btr.h
+++ b/storage/innobase/include/btr0btr.h
@@ -520,14 +520,10 @@ btr_insert_on_non_leaf_level_func(
 	mtr_t*		mtr);	/*!< in: mtr */
 #define btr_insert_on_non_leaf_level(f,i,l,t,m)			\
 	btr_insert_on_non_leaf_level_func(f,i,l,t,__FILE__,__LINE__,m)
-/****************************************************************//**
-Sets a record as the predefined minimum record. */
-void
-btr_set_min_rec_mark(
-/*=================*/
-	rec_t*	rec,	/*!< in/out: record */
-	mtr_t*	mtr)	/*!< in: mtr */
-	MY_ATTRIBUTE((nonnull));
+
+/** Sets a record as the predefined minimum record. */
+void btr_set_min_rec_mark(rec_t* rec, mtr_t* mtr) MY_ATTRIBUTE((nonnull));
+
 /** Seek to the parent page of a B-tree page.
 @param[in,out]	index	b-tree
 @param[in]	block	child page
diff --git a/storage/innobase/include/buf0buf.h b/storage/innobase/include/buf0buf.h
index cbef5475af5..f04d3fe038f 100644
--- a/storage/innobase/include/buf0buf.h
+++ b/storage/innobase/include/buf0buf.h
@@ -1468,8 +1468,6 @@ public:
 					if written again we check is TRIM
 					operation needed. */
 
-	bool            encrypted;	/*!< page is still encrypted */
-
 	ulint           real_size;	/*!< Real size of the page
 					Normal pages == srv_page_size
 					page compressed pages, payload
diff --git a/storage/innobase/include/buf0flu.h b/storage/innobase/include/buf0flu.h
index cf379c0a542..a0122d1c3f8 100644
--- a/storage/innobase/include/buf0flu.h
+++ b/storage/innobase/include/buf0flu.h
@@ -167,16 +167,6 @@ void
 buf_flush_wait_flushed(
 	lsn_t		new_oldest);
 
-/******************************************************************//**
-Waits until a flush batch of the given type ends. This is called by
-a thread that only wants to wait for a flush to end but doesn't do
-any flushing itself. */
-void
-buf_flush_wait_batch_end_wait_only(
-/*===============================*/
-	buf_pool_t*	buf_pool,	/*!< in: buffer pool instance */
-	buf_flush_t	type);		/*!< in: BUF_FLUSH_LRU
-					or BUF_FLUSH_LIST */
 /********************************************************************//**
 This function should be called at a mini-transaction commit, if a page was
 modified in it. Puts the block to the list of modified blocks, if it not
diff --git a/storage/innobase/include/fil0crypt.h b/storage/innobase/include/fil0crypt.h
index 2296945bbf9..870858b4ccd 100644
--- a/storage/innobase/include/fil0crypt.h
+++ b/storage/innobase/include/fil0crypt.h
@@ -373,7 +373,6 @@ Decrypt a page
 @param[in]	space			Tablespace
 @param[in]	tmp_frame		Temporary buffer used for decrypting
 @param[in,out]	src_frame		Page to decrypt
-@param[out]	decrypted		true if page was decrypted
 @return decrypted page, or original not encrypted page if decryption is
 not needed.*/
 UNIV_INTERN
@@ -381,8 +380,7 @@ byte*
 fil_space_decrypt(
 	const fil_space_t* space,
 	byte*		tmp_frame,
-	byte*		src_frame,
-	bool*		decrypted)
+	byte*		src_frame)
 	MY_ATTRIBUTE((warn_unused_result));
 
 /******************************************************************
diff --git a/storage/innobase/include/page0page.h b/storage/innobase/include/page0page.h
index f2e821ef452..f59f70d0f4c 100644
--- a/storage/innobase/include/page0page.h
+++ b/storage/innobase/include/page0page.h
@@ -1310,15 +1310,12 @@ ibool
 page_simple_validate_new(
 /*=====================*/
 	const page_t*	page);	/*!< in: index page in ROW_FORMAT!=REDUNDANT */
-/***************************************************************//**
-This function checks the consistency of an index page.
-@return TRUE if ok */
-ibool
-page_validate(
-/*==========*/
-	const page_t*	page,	/*!< in: index page */
-	dict_index_t*	index);	/*!< in: data dictionary index containing
-				the page record type definition */
+/** Check the consistency of an index page.
+@param[in]	page	index page
+@param[in]	index	B-tree or R-tree index
+@return	whether the page is valid */
+bool page_validate(const page_t* page, const dict_index_t* index)
+	MY_ATTRIBUTE((nonnull));
 /***************************************************************//**
 Looks in the page record list for a record with the given heap number.
 @return record, NULL if not found */
diff --git a/storage/innobase/include/page0page.ic b/storage/innobase/include/page0page.ic
index fea74c9a15a..9785b451a06 100644
--- a/storage/innobase/include/page0page.ic
+++ b/storage/innobase/include/page0page.ic
@@ -659,6 +659,10 @@ page_rec_get_next_low(
 		return(NULL);
 	}
 
+	ut_ad(page_rec_is_infimum(rec)
+	      || !(rec_get_info_bits(page + offs, comp)
+		   & REC_INFO_MIN_REC_FLAG));
+
 	return(page + offs);
 }
 
diff --git a/storage/innobase/page/page0cur.cc b/storage/innobase/page/page0cur.cc
index dd47c13d240..204fedf9d29 100644
--- a/storage/innobase/page/page0cur.cc
+++ b/storage/innobase/page/page0cur.cc
@@ -2425,10 +2425,6 @@ page_cur_delete_rec(
 	if (cur_n_owned <= PAGE_DIR_SLOT_MIN_N_OWNED) {
 		page_dir_balance_slot(page, page_zip, cur_slot_no);
 	}
-
-#ifdef UNIV_ZIP_DEBUG
-	ut_a(!page_zip || page_zip_validate(page_zip, page, index));
-#endif /* UNIV_ZIP_DEBUG */
 }
 
 #ifdef UNIV_COMPILE_TEST_FUNCS
diff --git a/storage/innobase/page/page0page.cc b/storage/innobase/page/page0page.cc
index 2323bfbea1a..47e8c5ae4aa 100644
--- a/storage/innobase/page/page0page.cc
+++ b/storage/innobase/page/page0page.cc
@@ -2374,19 +2374,16 @@ func_exit:
 	return(ret);
 }
 
-/***************************************************************//**
-This function checks the consistency of an index page.
-@return TRUE if ok */
-ibool
-page_validate(
-/*==========*/
-	const page_t*	page,	/*!< in: index page */
-	dict_index_t*	index)	/*!< in: data dictionary index containing
-				the page record type definition */
+/** Check the consistency of an index page.
+@param[in]	page	index page
+@param[in]	index	B-tree or R-tree index
+@return	whether the page is valid */
+bool page_validate(const page_t* page, const dict_index_t* index)
 {
 	const page_dir_slot_t*	slot;
 	const rec_t*		rec;
 	const rec_t*		old_rec		= NULL;
+	const rec_t*		first_rec	= NULL;
 	ulint			offs;
 	ulint			n_slots;
 	ibool			ret		= TRUE;
@@ -2510,6 +2507,43 @@ wrong_page_type:
 			goto next_rec;
 		}
 
+		if (rec == first_rec) {
+			if ((rec_get_info_bits(rec, page_is_comp(page))
+			     & REC_INFO_MIN_REC_FLAG)) {
+				if (page_has_prev(page)) {
+					ib::error() << "REC_INFO_MIN_REC_FLAG "
+						"is set in on non-left page";
+					ret = false;
+				} else if (!page_is_leaf(page)) {
+					/* leftmost node pointer page */
+				} else if (!index->is_instant()) {
+					ib::error() << "REC_INFO_MIN_REC_FLAG "
+						"is set in a leaf-page record";
+					ret = false;
+				} else if (rec_get_deleted_flag(
+						   rec, page_is_comp(page))) {
+					/* If this were a 10.4 metadata
+					record for index->table->instant
+					we should not get here in 10.3, because
+					the metadata record should not have
+					been recognized by
+					btr_cur_instant_init_low(). */
+					ib::error() << "Metadata record "
+						"is delete-marked";
+					ret = false;
+				}
+			} else if (!page_has_prev(page)
+				   && index->is_instant()) {
+				ib::error() << "Metadata record is missing";
+				ret = false;
+			}
+		} else if (rec_get_info_bits(rec, page_is_comp(page))
+			   & REC_INFO_MIN_REC_FLAG) {
+			ib::error() << "REC_INFO_MIN_REC_FLAG record is not "
+				       "first in page";
+			ret = false;
+		}
+
 		/* Check that the records are in the ascending order */
 		if (count >= PAGE_HEAP_NO_USER_LOW
 		    && !page_rec_is_supremum(rec)) {
@@ -2616,6 +2650,11 @@ next_rec:
 		old_rec = rec;
 		rec = page_rec_get_next_const(rec);
 
+		if (page_rec_is_infimum(old_rec)
+		    && page_rec_is_user_rec(rec)) {
+			first_rec = rec;
+		}
+
 		/* set old_offsets to offsets; recycle offsets */
 		{
 			ulint* offs = old_offsets;
diff --git a/storage/innobase/page/page0zip.cc b/storage/innobase/page/page0zip.cc
index c2b6695d6a5..eb38b16c2d9 100644
--- a/storage/innobase/page/page0zip.cc
+++ b/storage/innobase/page/page0zip.cc
@@ -4369,10 +4369,6 @@ page_zip_clear_rec(
 	} else {
 		ut_ad(!rec_offs_any_extern(offsets));
 	}
-
-#ifdef UNIV_ZIP_DEBUG
-	ut_a(page_zip_validate(page_zip, page, index));
-#endif /* UNIV_ZIP_DEBUG */
 }
 
 /**********************************************************************//**