diff --git a/mysql-test/main/analyze.result b/mysql-test/main/analyze.result
index 17325863acd..8d345d98b05 100644
--- a/mysql-test/main/analyze.result
+++ b/mysql-test/main/analyze.result
@@ -97,5 +97,290 @@ set use_stat_tables=default;
 set histogram_type=default;
 set histogram_size=default;
 #
+# MDEV-31957 Concurrent ALTER and ANALYZE collecting statistics can
+# result in stale statistical data
+#
+CREATE TABLE t1 (a INT, b VARCHAR(128));
+INSERT INTO t1 SELECT seq, CONCAT('s',seq) FROM seq_1_to_100;
+connect  con1,localhost,root,,;
+ALTER TABLE t1 MODIFY b BLOB;
+connection default;
+ANALYZE TABLE t1 PERSISTENT FOR ALL;
+connection con1;
+ANALYZE TABLE t1 PERSISTENT FOR ALL;
+connection default;
+disconnect con1;
+select db_name,table_name,column_name from mysql.column_stats;
+db_name	table_name	column_name
+test	t1	a
+drop table t1;
+#
+# Testing swapping columns
+#
+create or replace table t1 (a int primary key, b varchar(100), c varchar(100), d varchar(100)) engine=innodb;
+insert into t1 select seq, repeat('b',seq),repeat('c',mod(seq,5)), repeat('d',mod(seq,10))  from seq_1_to_100;
+ANALYZE TABLE t1 PERSISTENT FOR ALL;
+Table	Op	Msg_type	Msg_text
+test.t1	analyze	status	Engine-independent statistics collected
+test.t1	analyze	status	OK
+select db_name,table_name,column_name,avg_length from mysql.column_stats order by column_name;
+db_name	table_name	column_name	avg_length
+test	t1	a	4.0000
+test	t1	b	50.5000
+test	t1	c	2.0000
+test	t1	d	4.5000
+alter table t1 change b c varchar(200), change c b varchar(200);
+select db_name,table_name,column_name,avg_length from mysql.column_stats order by column_name;
+db_name	table_name	column_name	avg_length
+test	t1	a	4.0000
+test	t1	b	2.0000
+test	t1	c	50.5000
+test	t1	d	4.5000
+alter table t1 change b c varchar(200), change c d varchar(200), change d b varchar(200) ;
+select db_name,table_name,column_name,avg_length from mysql.column_stats order by column_name;
+db_name	table_name	column_name	avg_length
+test	t1	a	4.0000
+test	t1	b	4.5000
+test	t1	c	2.0000
+test	t1	d	50.5000
+alter table t1 change b c varchar(200), change c d varchar(200), change d e varchar(200) ;
+select db_name,table_name,column_name,avg_length from mysql.column_stats order by column_name;
+db_name	table_name	column_name	avg_length
+test	t1	a	4.0000
+test	t1	c	4.5000
+test	t1	d	2.0000
+test	t1	e	50.5000
+alter table t1 change e d varchar(200), drop column d;
+select db_name,table_name,column_name,avg_length from mysql.column_stats order by column_name;
+db_name	table_name	column_name	avg_length
+test	t1	a	4.0000
+test	t1	c	4.5000
+test	t1	d	50.5000
+# Test having non existing column in column_stats
+insert into mysql.column_stats (db_name,table_name,column_name) values ("test","t1","b");
+alter table t1 change c d varchar(200), change d b varchar(200);
+select db_name,table_name,column_name,avg_length from mysql.column_stats order by column_name;
+db_name	table_name	column_name	avg_length
+test	t1	a	4.0000
+test	t1	b	50.5000
+test	t1	d	4.5000
+# Test having a conflicting temporary name
+insert into mysql.column_stats (db_name,table_name,column_name) values ("test","t1",concat("#sql_tmp_name#1",char(0)));
+alter table t1 change d b varchar(200), change b d varchar(200);
+select db_name,table_name,column_name,avg_length from mysql.column_stats order by column_name;
+db_name	table_name	column_name	avg_length
+test	t1	a	4.0000
+test	t1	b	4.5000
+test	t1	d	50.5000
+drop table t1;
+truncate table mysql.column_stats;
+create or replace table t1 (a int primary key, b varchar(100), c varchar(100), d varchar(100)) engine=myisam;
+insert into t1 select seq, repeat('b',seq),repeat('c',mod(seq,5)), repeat('d',mod(seq,10))  from seq_1_to_100;
+ANALYZE TABLE t1 PERSISTENT FOR ALL;
+Table	Op	Msg_type	Msg_text
+test.t1	analyze	status	Engine-independent statistics collected
+test.t1	analyze	status	OK
+select db_name,table_name,column_name,avg_length from mysql.column_stats order by column_name;
+db_name	table_name	column_name	avg_length
+test	t1	a	4.0000
+test	t1	b	50.5000
+test	t1	c	2.0000
+test	t1	d	4.5000
+alter table t1 change b c varchar(200), change c b varchar(200);
+select db_name,table_name,column_name,avg_length from mysql.column_stats order by column_name;
+db_name	table_name	column_name	avg_length
+test	t1	a	4.0000
+test	t1	d	4.5000
+analyze table t1 persistent for columns(b,c) indexes all;
+Table	Op	Msg_type	Msg_text
+test.t1	analyze	status	Engine-independent statistics collected
+test.t1	analyze	status	Table is already up to date
+alter table t1 change b c varchar(200), change c d varchar(200), change d b varchar(200) ;
+select db_name,table_name,column_name,avg_length from mysql.column_stats order by column_name;
+db_name	table_name	column_name	avg_length
+test	t1	a	4.0000
+test	t1	b	50.5000
+test	t1	c	2.0000
+analyze table t1 persistent for columns(d) indexes all;
+Table	Op	Msg_type	Msg_text
+test.t1	analyze	status	Engine-independent statistics collected
+test.t1	analyze	status	Table is already up to date
+alter table t1 change b c varchar(200), change c d varchar(200), change d e varchar(200) ;
+select db_name,table_name,column_name,avg_length from mysql.column_stats order by column_name;
+db_name	table_name	column_name	avg_length
+test	t1	a	4.0000
+test	t1	c	50.5000
+test	t1	d	2.0000
+test	t1	e	50.5000
+alter table t1 change e d varchar(200), drop column d;
+select db_name,table_name,column_name,avg_length from mysql.column_stats order by column_name;
+db_name	table_name	column_name	avg_length
+test	t1	a	4.0000
+test	t1	c	50.5000
+test	t1	d	50.5000
+drop table t1;
+truncate table mysql.column_stats;
+create table t1 (a int, b blob, unique(b)) engine= innodb;
+analyze table t1 persistent for all;
+Table	Op	Msg_type	Msg_text
+test.t1	analyze	status	Engine-independent statistics collected
+test.t1	analyze	Warning	Engine-independent statistics are not collected for column 'b'
+test.t1	analyze	status	OK
+select column_name from mysql.column_stats where table_name = 't1';
+column_name
+a
+drop table t1;
+create table t1 (a int, b blob, c int generated always as (length(b)) virtual) engine= innodb;
+analyze table t1 persistent for all;
+Table	Op	Msg_type	Msg_text
+test.t1	analyze	status	Engine-independent statistics collected
+test.t1	analyze	Warning	Engine-independent statistics are not collected for column 'b'
+test.t1	analyze	status	OK
+select column_name from mysql.column_stats where table_name = 't1';
+column_name
+a
+c
+drop table t1;
+CREATE or replace TABLE t1 (a INT, b CHAR(8));
+ANALYZE TABLE t1 PERSISTENT FOR ALL;
+Table	Op	Msg_type	Msg_text
+test.t1	analyze	status	Engine-independent statistics collected
+test.t1	analyze	status	Table is already up to date
+ALTER TABLE t1 CHANGE b c INT, ORDER BY b;
+SELECT db_name, table_name, column_name FROM mysql.column_stats where table_name = 't1';
+db_name	table_name	column_name
+test	t1	a
+test	t1	c
+drop table t1;
+CREATE or replace TABLE t1 (a INT, b CHAR(8));
+ANALYZE TABLE t1 PERSISTENT FOR ALL;
+Table	Op	Msg_type	Msg_text
+test.t1	analyze	status	Engine-independent statistics collected
+test.t1	analyze	status	Table is already up to date
+ALTER TABLE t1 RENAME COLUMN b to c, ALGORITHM=COPY;
+SELECT db_name, table_name, column_name FROM mysql.column_stats where table_name = 't1';
+db_name	table_name	column_name
+test	t1	a
+test	t1	c
+drop table t1;
+#
+# Testing swapping indexes
+#
+create or replace table t1 (a int primary key, b varchar(100), c varchar(100), d varchar(100), index (b), index(c), index(d,b)) engine=innodb;
+insert into t1 select seq, repeat('b',seq),repeat('c',mod(seq,5)), repeat('d',mod(seq,10))  from seq_1_to_100;
+ANALYZE TABLE t1 PERSISTENT FOR ALL;
+Table	Op	Msg_type	Msg_text
+test.t1	analyze	status	Engine-independent statistics collected
+test.t1	analyze	status	OK
+select * from mysql.index_stats order by index_name, prefix_arity;
+db_name	table_name	index_name	prefix_arity	avg_frequency
+test	t1	PRIMARY	1	1.0000
+test	t1	b	1	1.0000
+test	t1	b	2	1.0000
+test	t1	c	1	20.0000
+test	t1	c	2	1.0000
+test	t1	d	1	10.0000
+test	t1	d	2	1.0000
+test	t1	d	3	1.0000
+alter table t1 rename index b to c, rename index c to d, rename index d to b;
+select * from mysql.index_stats order by index_name;
+db_name	table_name	index_name	prefix_arity	avg_frequency
+test	t1	PRIMARY	1	1.0000
+test	t1	b	1	10.0000
+test	t1	b	2	1.0000
+test	t1	b	3	1.0000
+test	t1	c	1	1.0000
+test	t1	c	2	1.0000
+test	t1	d	1	20.0000
+test	t1	d	2	1.0000
+alter table t1 rename index b to c, rename index c to d, rename index d to e;
+select * from mysql.index_stats order by index_name, prefix_arity;
+db_name	table_name	index_name	prefix_arity	avg_frequency
+test	t1	PRIMARY	1	1.0000
+test	t1	c	1	10.0000
+test	t1	c	2	1.0000
+test	t1	c	3	1.0000
+test	t1	d	1	1.0000
+test	t1	d	2	1.0000
+test	t1	e	1	20.0000
+test	t1	e	2	1.0000
+alter table t1 rename index e to b;
+alter table t1 change b c varchar(200), change c d varchar(200), change d e varchar(200) ;
+show create table t1;
+Table	Create Table
+t1	CREATE TABLE `t1` (
+  `a` int(11) NOT NULL,
+  `c` varchar(200) DEFAULT NULL,
+  `d` varchar(200) DEFAULT NULL,
+  `e` varchar(200) DEFAULT NULL,
+  PRIMARY KEY (`a`),
+  KEY `d` (`c`),
+  KEY `b` (`d`),
+  KEY `c` (`e`,`c`)
+) ENGINE=InnoDB DEFAULT CHARSET=latin1 COLLATE=latin1_swedish_ci
+select * from mysql.index_stats order by index_name, prefix_arity;
+db_name	table_name	index_name	prefix_arity	avg_frequency
+test	t1	PRIMARY	1	1.0000
+test	t1	b	1	20.0000
+test	t1	b	2	1.0000
+test	t1	c	1	10.0000
+test	t1	c	2	1.0000
+test	t1	c	3	1.0000
+test	t1	d	1	1.0000
+test	t1	d	2	1.0000
+# Test having a conflicting temporary name
+insert into mysql.index_stats (db_name,table_name,index_name,prefix_arity) values ("test","t1",concat("#sql_tmp_name#1",char(0)),1);
+alter table t1 rename index c to d, rename index d to c;
+select * from mysql.index_stats order by index_name, prefix_arity;
+db_name	table_name	index_name	prefix_arity	avg_frequency
+test	t1	PRIMARY	1	1.0000
+test	t1	b	1	20.0000
+test	t1	b	2	1.0000
+test	t1	c	1	1.0000
+test	t1	c	2	1.0000
+test	t1	d	1	10.0000
+test	t1	d	2	1.0000
+test	t1	d	3	1.0000
+drop table t1;
+select * from mysql.index_stats order by index_name, prefix_arity;
+db_name	table_name	index_name	prefix_arity	avg_frequency
+#
+# Test of adding key that replaces foreign key
+#
+CREATE TABLE t1 (aaaa INT, b INT, KEY(b), FOREIGN KEY(aaaa) REFERENCES t1(b)) ENGINE=InnoDB;
+ANALYZE TABLE t1 PERSISTENT FOR ALL;
+Table	Op	Msg_type	Msg_text
+test.t1	analyze	status	Engine-independent statistics collected
+test.t1	analyze	status	OK
+SELECT index_name FROM mysql.index_stats WHERE table_name = 't1' order by index_name;
+index_name
+aaaa
+b
+ALTER TABLE t1 ADD KEY idx(aaaa);
+SHOW CREATE TABLE t1;
+Table	Create Table
+t1	CREATE TABLE `t1` (
+  `aaaa` int(11) DEFAULT NULL,
+  `b` int(11) DEFAULT NULL,
+  KEY `b` (`b`),
+  KEY `idx` (`aaaa`),
+  CONSTRAINT `t1_ibfk_1` FOREIGN KEY (`aaaa`) REFERENCES `t1` (`b`)
+) ENGINE=InnoDB DEFAULT CHARSET=latin1 COLLATE=latin1_swedish_ci
+SELECT index_name FROM mysql.index_stats WHERE table_name = 't1' order by index_name;
+index_name
+b
+truncate table mysql.index_stats;
+ANALYZE TABLE t1 PERSISTENT FOR ALL;
+Table	Op	Msg_type	Msg_text
+test.t1	analyze	status	Engine-independent statistics collected
+test.t1	analyze	status	OK
+SELECT index_name FROM mysql.index_stats WHERE table_name = 't1' order by index_name;
+index_name
+b
+idx
+ALTER TABLE t1 DROP KEY idx;
+ERROR HY000: Cannot drop index 'idx': needed in a foreign key constraint
+DROP TABLE t1;
+#
 # End of 10.6 tests
 #
diff --git a/mysql-test/main/analyze.test b/mysql-test/main/analyze.test
index 3c7179fbf19..d89e05b8543 100644
--- a/mysql-test/main/analyze.test
+++ b/mysql-test/main/analyze.test
@@ -1,4 +1,5 @@
 --source include/have_sequence.inc
+--source include/have_innodb.inc
 
 #
 # Bug #10901 Analyze Table on new table destroys table
@@ -111,6 +112,148 @@ set use_stat_tables=default;
 set histogram_type=default;
 set histogram_size=default;
 
+--echo #
+--echo # MDEV-31957 Concurrent ALTER and ANALYZE collecting statistics can
+--echo # result in stale statistical data
+--echo #
+
+CREATE TABLE t1 (a INT, b VARCHAR(128));
+INSERT INTO t1 SELECT seq, CONCAT('s',seq) FROM seq_1_to_100;
+
+# We have to disable query log as the ANALYZE TABLE can be run in different
+# order. The important thing is what is finally in column_stats
+--disable_result_log
+--connect (con1,localhost,root,,)
+--send ALTER TABLE t1 MODIFY b BLOB
+
+--connection default
+ANALYZE TABLE t1 PERSISTENT FOR ALL;
+
+--connection con1
+--reap
+ANALYZE TABLE t1 PERSISTENT FOR ALL;
+--connection default
+--disconnect con1
+--enable_result_log
+select db_name,table_name,column_name from mysql.column_stats;
+drop table t1;
+
+--echo #
+--echo # Testing swapping columns
+--echo #
+
+create or replace table t1 (a int primary key, b varchar(100), c varchar(100), d varchar(100)) engine=innodb;
+insert into t1 select seq, repeat('b',seq),repeat('c',mod(seq,5)), repeat('d',mod(seq,10))  from seq_1_to_100;
+ANALYZE TABLE t1 PERSISTENT FOR ALL;
+select db_name,table_name,column_name,avg_length from mysql.column_stats order by column_name;
+alter table t1 change b c varchar(200), change c b varchar(200);
+select db_name,table_name,column_name,avg_length from mysql.column_stats order by column_name;
+alter table t1 change b c varchar(200), change c d varchar(200), change d b varchar(200) ;
+select db_name,table_name,column_name,avg_length from mysql.column_stats order by column_name;
+alter table t1 change b c varchar(200), change c d varchar(200), change d e varchar(200) ;
+select db_name,table_name,column_name,avg_length from mysql.column_stats order by column_name;
+alter table t1 change e d varchar(200), drop column d;
+select db_name,table_name,column_name,avg_length from mysql.column_stats order by column_name;
+
+--echo # Test having non existing column in column_stats
+
+insert into mysql.column_stats (db_name,table_name,column_name) values ("test","t1","b");
+alter table t1 change c d varchar(200), change d b varchar(200);
+select db_name,table_name,column_name,avg_length from mysql.column_stats order by column_name;
+
+--echo # Test having a conflicting temporary name
+insert into mysql.column_stats (db_name,table_name,column_name) values ("test","t1",concat("#sql_tmp_name#1",char(0)));
+alter table t1 change d b varchar(200), change b d varchar(200);
+select db_name,table_name,column_name,avg_length from mysql.column_stats order by column_name;
+
+drop table t1;
+truncate table mysql.column_stats;
+
+create or replace table t1 (a int primary key, b varchar(100), c varchar(100), d varchar(100)) engine=myisam;
+insert into t1 select seq, repeat('b',seq),repeat('c',mod(seq,5)), repeat('d',mod(seq,10))  from seq_1_to_100;
+ANALYZE TABLE t1 PERSISTENT FOR ALL;
+select db_name,table_name,column_name,avg_length from mysql.column_stats order by column_name;
+alter table t1 change b c varchar(200), change c b varchar(200);
+select db_name,table_name,column_name,avg_length from mysql.column_stats order by column_name;
+analyze table t1 persistent for columns(b,c) indexes all;
+alter table t1 change b c varchar(200), change c d varchar(200), change d b varchar(200) ;
+select db_name,table_name,column_name,avg_length from mysql.column_stats order by column_name;
+analyze table t1 persistent for columns(d) indexes all;
+alter table t1 change b c varchar(200), change c d varchar(200), change d e varchar(200) ;
+select db_name,table_name,column_name,avg_length from mysql.column_stats order by column_name;
+alter table t1 change e d varchar(200), drop column d;
+select db_name,table_name,column_name,avg_length from mysql.column_stats order by column_name;
+drop table t1;
+truncate table mysql.column_stats;
+
+create table t1 (a int, b blob, unique(b)) engine= innodb;
+analyze table t1 persistent for all;
+select column_name from mysql.column_stats where table_name = 't1';
+drop table t1;
+
+create table t1 (a int, b blob, c int generated always as (length(b)) virtual) engine= innodb;
+analyze table t1 persistent for all;
+select column_name from mysql.column_stats where table_name = 't1';
+drop table t1;
+
+CREATE or replace TABLE t1 (a INT, b CHAR(8));
+ANALYZE TABLE t1 PERSISTENT FOR ALL;
+ALTER TABLE t1 CHANGE b c INT, ORDER BY b;
+SELECT db_name, table_name, column_name FROM mysql.column_stats where table_name = 't1';
+drop table t1;
+
+CREATE or replace TABLE t1 (a INT, b CHAR(8));
+ANALYZE TABLE t1 PERSISTENT FOR ALL;
+ALTER TABLE t1 RENAME COLUMN b to c, ALGORITHM=COPY;
+SELECT db_name, table_name, column_name FROM mysql.column_stats where table_name = 't1';
+drop table t1;
+
+--echo #
+--echo # Testing swapping indexes
+--echo #
+
+create or replace table t1 (a int primary key, b varchar(100), c varchar(100), d varchar(100), index (b), index(c), index(d,b)) engine=innodb;
+insert into t1 select seq, repeat('b',seq),repeat('c',mod(seq,5)), repeat('d',mod(seq,10))  from seq_1_to_100;
+ANALYZE TABLE t1 PERSISTENT FOR ALL;
+select * from mysql.index_stats order by index_name, prefix_arity;
+alter table t1 rename index b to c, rename index c to d, rename index d to b;
+select * from mysql.index_stats order by index_name;
+
+alter table t1 rename index b to c, rename index c to d, rename index d to e;
+select * from mysql.index_stats order by index_name, prefix_arity;
+alter table t1 rename index e to b;
+alter table t1 change b c varchar(200), change c d varchar(200), change d e varchar(200) ;
+show create table t1;
+select * from mysql.index_stats order by index_name, prefix_arity;
+
+--echo # Test having a conflicting temporary name
+insert into mysql.index_stats (db_name,table_name,index_name,prefix_arity) values ("test","t1",concat("#sql_tmp_name#1",char(0)),1);
+alter table t1 rename index c to d, rename index d to c;
+select * from mysql.index_stats order by index_name, prefix_arity;
+drop table t1;
+select * from mysql.index_stats order by index_name, prefix_arity;
+
+--echo #
+--echo # Test of adding key that replaces foreign key
+--echo #
+
+CREATE TABLE t1 (aaaa INT, b INT, KEY(b), FOREIGN KEY(aaaa) REFERENCES t1(b)) ENGINE=InnoDB;
+ANALYZE TABLE t1 PERSISTENT FOR ALL;
+
+SELECT index_name FROM mysql.index_stats WHERE table_name = 't1' order by index_name;
+
+ALTER TABLE t1 ADD KEY idx(aaaa);
+SHOW CREATE TABLE t1;
+
+SELECT index_name FROM mysql.index_stats WHERE table_name = 't1' order by index_name;
+
+truncate table mysql.index_stats;
+ANALYZE TABLE t1 PERSISTENT FOR ALL;
+SELECT index_name FROM mysql.index_stats WHERE table_name = 't1' order by index_name;
+
+--error ER_DROP_INDEX_FK
+ALTER TABLE t1 DROP KEY idx;
+DROP TABLE t1;
 --echo #
 --echo # End of 10.6 tests
 --echo #
diff --git a/mysql-test/main/statistics.result b/mysql-test/main/statistics.result
index 8677b992b03..a442a33d14b 100644
--- a/mysql-test/main/statistics.result
+++ b/mysql-test/main/statistics.result
@@ -552,6 +552,13 @@ test	t1	PRIMARY	1	1.0000
 test	t1	idx2	1	7.0000
 test	t1	idx2	2	2.3846
 test	t1	idx3	1	8.5000
+ANALYZE TABLE t1 PERSISTENT FOR COLUMNS(x) INDEXES();
+Table	Op	Msg_type	Msg_text
+test.t1	analyze	status	Engine-independent statistics collected
+test.t1	analyze	status	OK
+SELECT * FROM mysql.column_stats where column_name="x";
+db_name	table_name	column_name	min_value	max_value	nulls_ratio	avg_length	avg_frequency	hist_size	hist_type	histogram
+test	t1	x	vvvvvvvvvvvvv	zzzzzzzzzzzzzzzzzz	0.2000	17.1250	6.4000	0	NULL	NULL
 ALTER TABLE t1 CHANGE COLUMN x b varchar(32);
 SHOW CREATE TABLE t1;
 Table	Create Table
@@ -668,8 +675,8 @@ test	t1	PRIMARY	1	1.0000
 test	t1	idx2	1	7.0000
 test	t1	idx2	2	2.3846
 test	t1	idx3	1	8.5000
-LOAD DATA INFILE 'MYSQLTEST_VARDIR/tmp/save_column_stats'
-  INTO TABLE mysql.column_stats
+LOAD DATA INFILE 'MYSQLTEST_VARDIR/tmp/save_column_stats' IGNORE
+INTO TABLE mysql.column_stats
 FIELDS TERMINATED BY ',' OPTIONALLY ENCLOSED BY '"' LINES TERMINATED BY '\n';
 LOAD DATA INFILE 'MYSQLTEST_VARDIR/tmp/save_index_stats'
   INTO TABLE mysql.index_stats
@@ -1106,12 +1113,17 @@ test	t2	idx4	1	6.2000
 test	t2	idx4	2	1.7222
 test	t2	idx4	3	1.1154
 test	t2	idx4	4	1.0000
+SELECT * FROM mysql.column_stats where column_name="b";
+db_name	table_name	column_name	min_value	max_value	nulls_ratio	avg_length	avg_frequency	hist_size	hist_type	histogram
+test	t2	b	vvvvvvvvvvvvv	zzzzzzzzzzzzzzzzzz	0.2000	17.1250	6.4000	0	NULL	NULL
 ALTER TABLE t2 CHANGE COLUMN b b varchar(30);
 SELECT * FROM mysql.index_stats ORDER BY index_name, prefix_arity, table_name;
 db_name	table_name	index_name	prefix_arity	avg_frequency
 test	t2	idx2	1	7.0000
 test	t2	idx2	2	2.3846
 test	t2	idx3	1	8.5000
+SELECT * FROM mysql.column_stats where column_name="b";
+db_name	table_name	column_name	min_value	max_value	nulls_ratio	avg_length	avg_frequency	hist_size	hist_type	histogram
 ANALYZE TABLE t2 PERSISTENT FOR COLUMNS ALL INDEXES ALL;
 Table	Op	Msg_type	Msg_text
 test.t2	analyze	status	Engine-independent statistics collected
@@ -1147,6 +1159,9 @@ test	t2	idx4	1	6.2000
 test	t2	idx4	2	1.7222
 test	t2	idx4	3	1.1154
 test	t2	idx4	4	1.0000
+SELECT * FROM mysql.column_stats where column_name="b";
+db_name	table_name	column_name	min_value	max_value	nulls_ratio	avg_length	avg_frequency	hist_size	hist_type	histogram
+test	t2	b	0	zzzzzzzzzzzzzzzzzz	0.0000	13.9000	6.6667	0	NULL	NULL
 ANALYZE TABLE t2 PERSISTENT FOR COLUMNS ALL INDEXES ALL;
 Table	Op	Msg_type	Msg_text
 test.t2	analyze	status	Engine-independent statistics collected
diff --git a/mysql-test/main/statistics.test b/mysql-test/main/statistics.test
index e6b300b5db2..e0b92ec2e40 100644
--- a/mysql-test/main/statistics.test
+++ b/mysql-test/main/statistics.test
@@ -305,6 +305,10 @@ SELECT * FROM mysql.column_stats;
 --sorted_result
 SELECT * FROM mysql.index_stats;
 
+# Add 'x' back that was deleted above
+ANALYZE TABLE t1 PERSISTENT FOR COLUMNS(x) INDEXES();
+SELECT * FROM mysql.column_stats where column_name="x";
+
 ALTER TABLE t1 CHANGE COLUMN x b varchar(32);
 SHOW CREATE TABLE t1;
 --sorted_result
@@ -347,7 +351,7 @@ SELECT * FROM mysql.index_stats;
 
 --replace_result $MYSQLTEST_VARDIR MYSQLTEST_VARDIR
 eval
-LOAD DATA INFILE '$MYSQLTEST_VARDIR/tmp/save_column_stats'
+LOAD DATA INFILE '$MYSQLTEST_VARDIR/tmp/save_column_stats' IGNORE
   INTO TABLE mysql.column_stats
   FIELDS TERMINATED BY ',' OPTIONALLY ENCLOSED BY '"' LINES TERMINATED BY '\n';
 --replace_result $MYSQLTEST_VARDIR MYSQLTEST_VARDIR
@@ -495,14 +499,17 @@ SELECT * FROM mysql.index_stats ORDER BY index_name, prefix_arity, table_name;
 ANALYZE TABLE t2 PERSISTENT FOR COLUMNS() INDEXES ALL;
 SELECT * FROM mysql.index_stats ORDER BY index_name, prefix_arity, table_name;
 
+SELECT * FROM mysql.column_stats where column_name="b";
 ALTER TABLE t2 CHANGE COLUMN b b varchar(30);
 SELECT * FROM mysql.index_stats ORDER BY index_name, prefix_arity, table_name;
+SELECT * FROM mysql.column_stats where column_name="b";
 
 ANALYZE TABLE t2 PERSISTENT FOR COLUMNS ALL INDEXES ALL;
 SELECT * FROM mysql.index_stats ORDER BY index_name, prefix_arity, table_name;
 
 ALTER TABLE t2 CHANGE COLUMN b b varchar(32);
 SELECT * FROM mysql.index_stats ORDER BY index_name, prefix_arity, table_name;
+SELECT * FROM mysql.column_stats where column_name="b";
 
 ANALYZE TABLE t2 PERSISTENT FOR COLUMNS ALL INDEXES ALL;
 SELECT * FROM mysql.index_stats ORDER BY index_name, prefix_arity, table_name;
diff --git a/sql/sql_admin.cc b/sql/sql_admin.cc
index b7a845f04c9..bde633b174c 100644
--- a/sql/sql_admin.cc
+++ b/sql/sql_admin.cc
@@ -517,8 +517,6 @@ static bool mysql_admin_table(THD* thd, TABLE_LIST* tables,
   List<Item> field_list;
   Protocol *protocol= thd->protocol;
   LEX *lex= thd->lex;
-  int result_code;
-  int compl_result_code;
   bool need_repair_or_alter= 0;
   wait_for_commit* suspended_wfc;
   bool is_table_modified= false;
@@ -562,7 +560,9 @@ static bool mysql_admin_table(THD* thd, TABLE_LIST* tables,
     bool collect_eis=  FALSE;
     bool open_for_modify= org_open_for_modify;
     Recreate_info recreate_info;
+    int compl_result_code, result_code;
 
+    compl_result_code= result_code= HA_ADMIN_FAILED;
     storage_engine_name[0]= 0;                  // Marker that's not used
 
     DBUG_PRINT("admin", ("table: '%s'.'%s'", db, table->table_name.str));
@@ -880,6 +880,7 @@ static bool mysql_admin_table(THD* thd, TABLE_LIST* tables,
       DBUG_PRINT("admin", ("operator_func returned: %d", result_code));
     }
 
+    /* Note: compl_result_code can be different from result_code here */
     if (compl_result_code == HA_ADMIN_OK && collect_eis)
     {
       if (result_code == HA_ERR_TABLE_READONLY)
@@ -920,19 +921,35 @@ static bool mysql_admin_table(THD* thd, TABLE_LIST* tables,
         Field **field_ptr= tab->field;
         if (!lex->column_list)
         {
+          /* Fields we have to read from the engine */
           bitmap_clear_all(tab->read_set);
+          /* Fields we want to have statistics for */
+          bitmap_clear_all(&tab->has_value_set);
+
           for (uint fields= 0; *field_ptr; field_ptr++, fields++)
           {
+            Field *field= *field_ptr;
+            if (field->flags & LONG_UNIQUE_HASH_FIELD)
+            {
+              /*
+                No point in doing statistic for hash fields that should be
+                unique
+              */
+              continue;
+            }
             /*
               Note that type() always return MYSQL_TYPE_BLOB for
               all blob types. Another function needs to be added
               if we in the future want to distingush between blob
               types here.
             */
-            enum enum_field_types type= (*field_ptr)->type();
+            enum enum_field_types type= field->type();
             if (type < MYSQL_TYPE_TINY_BLOB ||
                 type > MYSQL_TYPE_BLOB)
-              tab->field[fields]->register_field_in_read_map();
+            {
+              field->register_field_in_read_map();
+              bitmap_set_bit(&tab->has_value_set, field->field_index);
+            }
             else
               push_warning_printf(thd, Sql_condition::WARN_LEVEL_WARN,
                                   ER_NO_EIS_FOR_FIELD,
@@ -946,9 +963,15 @@ static bool mysql_admin_table(THD* thd, TABLE_LIST* tables,
           LEX_STRING *column_name;
           List_iterator_fast<LEX_STRING> it(*lex->column_list);
 
+          /* Fields we have to read from the engine */
           bitmap_clear_all(tab->read_set);
+          /* Fields we want to have statistics for */
+          bitmap_clear_all(&tab->has_value_set);
+
           while ((column_name= it++))
           {
+            Field *field;
+            enum enum_field_types type;
             if (tab->s->fieldnames.type_names == 0 ||
                 (pos= find_type(&tab->s->fieldnames, column_name->str,
                                 column_name->length, 1)) <= 0)
@@ -957,10 +980,15 @@ static bool mysql_admin_table(THD* thd, TABLE_LIST* tables,
               break;
             }
             pos--;
-            enum enum_field_types type= tab->field[pos]->type();
-            if (type < MYSQL_TYPE_TINY_BLOB ||
-                type > MYSQL_TYPE_BLOB)
-              tab->field[pos]->register_field_in_read_map();
+            field= tab->field[pos];
+            type= field->type();
+            if (!(field->flags & LONG_UNIQUE_HASH_FIELD) &&
+                (type < MYSQL_TYPE_TINY_BLOB ||
+                 type > MYSQL_TYPE_BLOB))
+            {
+              field->register_field_in_read_map();
+              bitmap_set_bit(&tab->has_value_set, field->field_index);
+            }
             else
               push_warning_printf(thd, Sql_condition::WARN_LEVEL_WARN,
                                   ER_NO_EIS_FOR_FIELD,
@@ -991,12 +1019,13 @@ static bool mysql_admin_table(THD* thd, TABLE_LIST* tables,
           }
         }
         /* Ensure that number of records are updated */
-        table->table->file->info(HA_STATUS_VARIABLE);
+        tab->file->info(HA_STATUS_VARIABLE);
         if (!(compl_result_code=
-              alloc_statistics_for_table(thd, table->table)) &&
+              alloc_statistics_for_table(thd, tab,
+                                         &tab->has_value_set)) &&
             !(compl_result_code=
-              collect_statistics_for_table(thd, table->table)))
-          compl_result_code= update_statistics_for_table(thd, table->table);
+              collect_statistics_for_table(thd, tab)))
+          compl_result_code= update_statistics_for_table(thd, tab);
       }
       else
         compl_result_code= HA_ADMIN_FAILED;
diff --git a/sql/sql_alter.cc b/sql/sql_alter.cc
index f7367dd95aa..3a9dd8c6fbe 100644
--- a/sql/sql_alter.cc
+++ b/sql/sql_alter.cc
@@ -18,6 +18,7 @@
 #include "sql_parse.h"                       // check_access
 #include "sql_table.h"                       // mysql_alter_table,
                                              // mysql_exchange_partition
+#include "sql_statistics.h"                  // delete_statistics_for_column
 #include "sql_alter.h"
 #include "wsrep_mysqld.h"
 
@@ -31,6 +32,7 @@ Alter_info::Alter_info(const Alter_info &rhs, MEM_ROOT *mem_root)
   check_constraint_list(rhs.check_constraint_list, mem_root),
   flags(rhs.flags), partition_flags(rhs.partition_flags),
   keys_onoff(rhs.keys_onoff),
+  original_table(0),
   partition_names(rhs.partition_names, mem_root),
   num_parts(rhs.num_parts),
   requested_algorithm(rhs.requested_algorithm),
@@ -297,6 +299,79 @@ uint Alter_info::check_vcol_field(Item_field *item) const
 }
 
 
+bool Alter_info::collect_renamed_fields(THD *thd)
+{
+  List_iterator_fast<Create_field> new_field_it;
+  Create_field *new_field;
+  DBUG_ENTER("Alter_info::collect_renamed_fields");
+
+  new_field_it.init(create_list);
+  while ((new_field= new_field_it++))
+  {
+    Field *field= new_field->field;
+
+    if (new_field->field &&
+        cmp(&field->field_name, &new_field->field_name))
+    {
+      field->flags|= FIELD_IS_RENAMED;
+      if (add_stat_rename_field(field,
+                                &new_field->field_name,
+                                thd->mem_root))
+        DBUG_RETURN(true);
+
+    }
+  }
+  DBUG_RETURN(false);
+}
+
+
+/*
+  Delete duplicate index found during mysql_prepare_create_table()
+
+  Notes:
+    - In case of temporary generated foreign keys, the key_name may not
+      be set!  These keys are ignored.
+*/
+
+bool Alter_info::add_stat_drop_index(THD *thd, const LEX_CSTRING *key_name)
+{
+  if (original_table && key_name->length)       // If from alter table
+  {
+    KEY *key_info= original_table->key_info;
+    for (uint i= 0; i < original_table->s->keys; i++, key_info++)
+    {
+      if (key_info->name.length &&
+          !lex_string_cmp(system_charset_info, &key_info->name,
+                          key_name))
+        return add_stat_drop_index(key_info, false, thd->mem_root);
+    }
+  }
+  return false;
+}
+
+
+void Alter_info::apply_statistics_deletes_renames(THD *thd, TABLE *table)
+{
+  List_iterator<Field>                     it_drop_field(drop_stat_fields);
+  List_iterator<RENAME_COLUMN_STAT_PARAMS> it_rename_field(rename_stat_fields);
+  List_iterator<DROP_INDEX_STAT_PARAMS>    it_drop_index(drop_stat_indexes);
+  List_iterator<RENAME_INDEX_STAT_PARAMS>  it_rename_index(rename_stat_indexes);
+
+  while (Field *field= it_drop_field++)
+    delete_statistics_for_column(thd, table, field);
+
+  if (!rename_stat_fields.is_empty())
+    (void) rename_columns_in_stat_table(thd, table, &rename_stat_fields);
+
+  while (DROP_INDEX_STAT_PARAMS *key= it_drop_index++)
+    (void) delete_statistics_for_index(thd, table, key->key,
+                                       key->ext_prefixes_only);
+
+  if (!rename_stat_indexes.is_empty())
+    (void) rename_indexes_in_stat_table(thd, table, &rename_stat_indexes);
+}
+
+
 Alter_table_ctx::Alter_table_ctx()
   : db(null_clex_str), table_name(null_clex_str), alias(null_clex_str),
     new_db(null_clex_str), new_name(null_clex_str), new_alias(null_clex_str)
diff --git a/sql/sql_alter.h b/sql/sql_alter.h
index 4605a6d415a..24203716cb2 100644
--- a/sql/sql_alter.h
+++ b/sql/sql_alter.h
@@ -105,10 +105,88 @@ public:
   ulong                         partition_flags;
   // Enable or disable keys.
   enum_enable_or_disable        keys_onoff;
+  // Used only in add_stat_drop_index()
+  TABLE                         *original_table;
   // List of partitions.
   List<const char>              partition_names;
   // Number of partitions.
   uint                          num_parts;
+
+  /* List of fields that we should delete statistics from */
+  List<Field> drop_stat_fields;
+
+  struct DROP_INDEX_STAT_PARAMS
+  {
+    KEY *key;
+    bool ext_prefixes_only;
+  };
+
+  struct RENAME_COLUMN_STAT_PARAMS
+  {
+    Field *field;
+    LEX_CSTRING *name;
+    uint duplicate_counter;                       // For temporary names
+  };
+  struct RENAME_INDEX_STAT_PARAMS
+  {
+    const KEY *key;
+    const LEX_CSTRING *name;
+    uint duplicate_counter;                       // For temporary names
+    uint usage_count;                             // How many rename entries
+  };
+
+  /* List of index that we should delete statistics from */
+  List<DROP_INDEX_STAT_PARAMS> drop_stat_indexes;
+
+  List<RENAME_COLUMN_STAT_PARAMS> rename_stat_fields;
+
+  List<RENAME_INDEX_STAT_PARAMS> rename_stat_indexes;
+
+  bool add_stat_drop_index(KEY *key, bool ext_prefixes_only,
+                           MEM_ROOT *mem_root)
+  {
+    DROP_INDEX_STAT_PARAMS *param;
+    if (!(param= (DROP_INDEX_STAT_PARAMS*)
+          alloc_root(mem_root, sizeof(*param))))
+      return true;
+    param->key=  key;
+    param->ext_prefixes_only= ext_prefixes_only;
+    return drop_stat_indexes.push_back(param, mem_root);
+  }
+
+  bool add_stat_drop_index(THD *thd, const LEX_CSTRING *key_name);
+
+  bool add_stat_rename_index(const KEY *key, const LEX_CSTRING *name,
+                             MEM_ROOT *mem_root)
+  {
+    RENAME_INDEX_STAT_PARAMS *param;
+    if (!(param= (RENAME_INDEX_STAT_PARAMS*)
+          alloc_root(mem_root, sizeof(*param))))
+      return true;
+    param->key=  key;
+    param->name= name;
+    param->usage_count= 0;
+    return rename_stat_indexes.push_back(param, mem_root);
+  }
+
+  bool add_stat_rename_field(Field *field, LEX_CSTRING *name,
+                             MEM_ROOT *mem_root)
+  {
+    RENAME_COLUMN_STAT_PARAMS *param;
+    if (!(param= (RENAME_COLUMN_STAT_PARAMS*)
+          alloc_root(mem_root, sizeof(*param))))
+      return true;
+    param->field= field;
+    param->name=  name;
+    param->duplicate_counter= 0;
+    return rename_stat_fields.push_back(param, mem_root);
+  }
+
+  bool collect_renamed_fields(THD *thd);
+
+  /* Delete/update statistics in EITS tables */
+  void apply_statistics_deletes_renames(THD *thd, TABLE *table);
+
 private:
   // Type of ALTER TABLE algorithm.
   enum_alter_table_algorithm    requested_algorithm;
@@ -135,6 +213,10 @@ public:
     create_list.empty();
     alter_index_ignorability_list.empty();
     check_constraint_list.empty();
+    drop_stat_fields.empty();
+    drop_stat_indexes.empty();
+    rename_stat_fields.empty();
+    rename_stat_indexes.empty();
     flags= 0;
     partition_flags= 0;
     keys_onoff= LEAVE_AS_IS;
diff --git a/sql/sql_class.h b/sql/sql_class.h
index c945d8d974f..d76cc46d307 100644
--- a/sql/sql_class.h
+++ b/sql/sql_class.h
@@ -444,7 +444,8 @@ private:
 
 class Key :public Sql_alloc, public DDL_options {
 public:
-  enum Keytype { PRIMARY, UNIQUE, MULTIPLE, FULLTEXT, SPATIAL, FOREIGN_KEY};
+  enum Keytype { PRIMARY, UNIQUE, MULTIPLE, FULLTEXT, SPATIAL, FOREIGN_KEY,
+                 IGNORE_KEY};
   enum Keytype type;
   KEY_CREATE_INFO key_create_info;
   List<Key_part_spec> columns;
diff --git a/sql/sql_statistics.cc b/sql/sql_statistics.cc
index 9351c1253b0..5cc85bd5d59 100644
--- a/sql/sql_statistics.cc
+++ b/sql/sql_statistics.cc
@@ -32,6 +32,7 @@
 #include "uniques.h"
 #include "sql_show.h"
 #include "sql_partition.h"
+#include "sql_alter.h"                          // RENAME_STAT_PARAMS
 
 /*
   The system variable 'use_stat_tables' can take one of the
@@ -440,7 +441,7 @@ private:
   
   uint stat_key_length; /* Length of the key to access stat_table */
   uchar *record[2];     /* Record buffers used to access/update stat_table */
-  uint stat_key_idx;    /* The number of the key to access stat_table */
+
 
   /* This is a helper function used only by the Stat_table constructors */
   void common_init_stat_table()
@@ -450,6 +451,7 @@ private:
     stat_key_idx= 0;
     stat_key_info= &stat_table->key_info[stat_key_idx];
     stat_key_length= stat_key_info->key_length;
+    last_key_length= last_prefix_parts= 0;
     record[0]= stat_table->record[0];
     record[1]= stat_table->record[1];
   }
@@ -466,29 +468,31 @@ protected:
   const LEX_CSTRING *db_name;      /* Name of the database containing 'table' */
   const LEX_CSTRING *table_name;   /* Name of the table 'table' */
 
-  void store_record_for_update()
-  {
-    store_record(stat_table, record[1]);
-  }
+  uchar last_key[MAX_KEY_LENGTH];
+  uint last_key_length;
+  uint last_prefix_parts;
 
   void store_record_for_lookup()
   {
     DBUG_ASSERT(record[0] == stat_table->record[0]);
   }
 
-  bool update_record()
+  int update_record()
   {
     int err;
     if ((err= stat_file->ha_update_row(record[1], record[0])) &&
          err != HA_ERR_RECORD_IS_THE_SAME)
-      return TRUE;
-    /* Make change permanent and avoid 'table is marked as crashed' errors */
-    stat_file->extra(HA_EXTRA_FLUSH);
-    return FALSE;
+      return err;
+    return 0;
   }
 
 public:
 
+  uint stat_key_idx;    /* The number of the key to access stat_table */
+  void store_record_for_update()
+  {
+    store_record(stat_table, record[1]);
+  }
 
   /**
     @details
@@ -579,13 +583,22 @@ public:
 
   bool find_stat()
   {
-    uchar key[MAX_KEY_LENGTH];
-    key_copy(key, record[0], stat_key_info, stat_key_length);
-    return !stat_file->ha_index_read_idx_map(record[0], stat_key_idx, key,
+    last_key_length= stat_key_length;
+    key_copy(last_key, record[0], stat_key_info, stat_key_length);
+    return !stat_file->ha_index_read_idx_map(record[0], stat_key_idx, last_key,
                                              HA_WHOLE_KEY, HA_READ_KEY_EXACT);
   }
 
- 
+  void create_key_for_read(uint prefix_parts)
+  {
+    last_key_length= 0;
+    last_prefix_parts= prefix_parts;
+    for (uint i= 0; i < prefix_parts; i++)
+      last_key_length+= stat_key_info->key_part[i].store_length;
+    key_copy(last_key, record[0], stat_key_info, last_key_length);
+  }
+
+
   /**
     @brief
     Find a record in the statistical table by a key prefix value 
@@ -604,16 +617,32 @@ public:
 
   bool find_next_stat_for_prefix(uint prefix_parts)
   {
-    uchar key[MAX_KEY_LENGTH];
-    uint prefix_key_length= 0;
-    for (uint i= 0; i < prefix_parts; i++)
-      prefix_key_length+= stat_key_info->key_part[i].store_length;
-    key_copy(key, record[0], stat_key_info, prefix_key_length);
+    create_key_for_read(prefix_parts);
     key_part_map prefix_map= (key_part_map) ((1 << prefix_parts) - 1);
-    return !stat_file->ha_index_read_idx_map(record[0], stat_key_idx, key,
-                                             prefix_map, HA_READ_KEY_EXACT);
+    return !stat_file->ha_index_read_idx_map(record[0], stat_key_idx, last_key,
+                                         prefix_map, HA_READ_KEY_EXACT);
   }
    
+  bool find_next_stat_for_prefix_with_next(uint prefix_parts)
+  {
+    create_key_for_read(prefix_parts);
+    key_part_map prefix_map= (key_part_map) ((1 << prefix_parts) - 1);
+    return !stat_file->ha_index_read_map(record[0], last_key,
+                                         prefix_map,
+                                         HA_READ_KEY_EXACT);
+  }
+
+  /*
+    Read row with same key parts as last find_next_stat_for_prefix_with_next()
+  */
+
+  bool find_stat_with_next()
+  {
+    key_copy(last_key, record[0], stat_key_info, last_key_length);
+    key_part_map prefix_map= (key_part_map) ((1 << last_prefix_parts) - 1);
+    return !stat_file->ha_index_read_map(record[0], last_key,
+                                         prefix_map, HA_READ_KEY_EXACT);
+  }
 
   /**
     @brief
@@ -646,7 +675,7 @@ public:
       bool res;
       store_record_for_update();
       store_stat_fields();
-      res= update_record();
+      res= update_record() != 0;
       DBUG_ASSERT(res == 0);
       return res;
     }
@@ -659,14 +688,11 @@ public:
         DBUG_ASSERT(0);
 	return TRUE;
       }
-      /* Make change permanent and avoid 'table is marked as crashed' errors */
-      stat_file->extra(HA_EXTRA_FLUSH);
-    } 
+    }
     return FALSE;
   }
 
-
-  /** 
+  /**
     @brief
     Update the table name fields in the current record of stat_table
 
@@ -690,7 +716,7 @@ public:
   {
     store_record_for_update();
     change_full_table_name(db, tab);
-    bool rc= update_record();
+    bool rc= update_record() != 0;
     store_record_for_lookup();
     return rc;
   }   
@@ -715,10 +741,13 @@ public:
     int err;
     if ((err= stat_file->ha_delete_row(record[0])))
       return TRUE;
-    /* Make change permanent and avoid 'table is marked as crashed' errors */
-    stat_file->extra(HA_EXTRA_FLUSH);
     return FALSE;
-  } 
+  }
+
+  void flush()
+  {
+    stat_file->extra(HA_EXTRA_FLUSH);
+  }
 
   friend class Stat_table_write_iter;
 };
@@ -751,8 +780,8 @@ private:
   void change_full_table_name(const LEX_CSTRING *db,
                               const LEX_CSTRING *tab) override
   {
-    db_name_field->store(db->str, db->length, system_charset_info);
-    table_name_field->store(tab->str, tab->length, system_charset_info);
+    db_name_field->store(db, system_charset_info);
+    table_name_field->store(tab, system_charset_info);
   }
 
 public:
@@ -802,9 +831,8 @@ public:
 
   void set_key_fields()
   {
-    db_name_field->store(db_name->str, db_name->length, system_charset_info);
-    table_name_field->store(table_name->str, table_name->length,
-                            system_charset_info);
+    db_name_field->store(db_name, system_charset_info);
+    table_name_field->store(table_name, system_charset_info);
   }
 
 
@@ -894,8 +922,8 @@ private:
   void change_full_table_name(const LEX_CSTRING *db,
                               const LEX_CSTRING *tab) override
   {
-     db_name_field->store(db->str, db->length, system_charset_info);
-     table_name_field->store(tab->str, tab->length, system_charset_info);
+    db_name_field->store(db, system_charset_info);
+     table_name_field->store(tab, system_charset_info);
   }
 
 public:
@@ -939,9 +967,8 @@ public:
 
   void set_full_table_name()
   {
-    db_name_field->store(db_name->str, db_name->length, system_charset_info);
-    table_name_field->store(table_name->str, table_name->length,
-                            system_charset_info);
+    db_name_field->store(db_name, system_charset_info);
+    table_name_field->store(table_name, system_charset_info);
   }
 
 
@@ -959,18 +986,24 @@ public:
     It also sets table_field to the passed parameter.
 
     @note
-    The function is supposed to be called before any use of the  
+    The function is supposed to be called before any use of the
     method find_stat for an object of the Column_stat class.
   */
 
   void set_key_fields(Field *col)
   {
     set_full_table_name();
-    column_name_field->store(col->field_name.str, col->field_name.length,
-                             system_charset_info);  
+    column_name_field->store(&col->field_name, system_charset_info);
     table_field= col;
   }
 
+  void set_key_fields(LEX_CSTRING *field_name)
+  {
+    set_full_table_name();
+    column_name_field->store(field_name, system_charset_info);
+    table_field= 0;                             // Safety
+  }
+
 
   /** 
     @brief
@@ -980,22 +1013,27 @@ public:
     The function updates the primary key fields containing database name,
     table name, and column name for the last found record in the statistical
     table column_stats.
-    
+
     @retval
-    FALSE    success with the update of the record
+    0        success with the update of the record
     @retval
-    TRUE     failure with the update of the record
+    #        handler error in case of failure
   */
 
-  bool update_column_key_part(const char *col)
+  int update_column_key_part(LEX_CSTRING *col)
   {
+    int rc;
     store_record_for_update();
-    set_full_table_name();
-    column_name_field->store(col, strlen(col), system_charset_info);
-    bool rc= update_record();
+    rc= update_column(col);
     store_record_for_lookup();
     return rc;
-  }   
+  }
+
+  int update_column(LEX_CSTRING *col)
+  {
+    column_name_field->store(col, system_charset_info);
+    return update_record();
+  }
 
 
   /** 
@@ -1025,7 +1063,8 @@ public:
   {
     StringBuffer<MAX_FIELD_WIDTH> val;
 
-    MY_BITMAP *old_map= dbug_tmp_use_all_columns(stat_table, &stat_table->read_set);
+    MY_BITMAP *old_map= dbug_tmp_use_all_columns(stat_table,
+                                                 &stat_table->read_set);
     for (uint i= COLUMN_STAT_MIN_VALUE; i <= COLUMN_STAT_HISTOGRAM; i++)
     {  
       Field *stat_field= stat_table->field[i];
@@ -1210,8 +1249,7 @@ private:
   Field *table_name_field;   /* Field for the column index_stats.table_name */
   Field *index_name_field;   /* Field for the column index_stats.table_name */
   Field *prefix_arity_field; /* Field for the column index_stats.prefix_arity */
-
-  KEY *table_key_info;  /* Info on the index to read/update statistics on */
+  const KEY *table_key_info; /* Info on the index to read/update statistics on */
   uint prefix_arity; /* Number of components of the index prefix of interest */
 
   void common_init_index_stat_table()
@@ -1225,8 +1263,8 @@ private:
   void change_full_table_name(const LEX_CSTRING *db,
                               const LEX_CSTRING *tab) override
   {
-     db_name_field->store(db->str, db->length, system_charset_info);
-     table_name_field->store(tab->str, tab->length, system_charset_info);
+     db_name_field->store(db, system_charset_info);
+     table_name_field->store(tab, system_charset_info);
   }
 
 public:
@@ -1273,9 +1311,13 @@ public:
 
   void set_full_table_name()
   {
-    db_name_field->store(db_name->str, db_name->length, system_charset_info);
-    table_name_field->store(table_name->str, table_name->length,
-                            system_charset_info);
+    db_name_field->store(db_name, system_charset_info);
+    table_name_field->store(table_name, system_charset_info);
+  }
+
+  inline void set_index_name(const LEX_CSTRING *name)
+  {
+    index_name_field->store(name, system_charset_info);
   }
 
   /** 
@@ -1295,12 +1337,10 @@ public:
     find_next_stat_for_prefix for an object of the Index_stat class.
   */
 
-  void set_index_prefix_key_fields(KEY *index_info)
+  void set_index_prefix_key_fields(const KEY *index_info)
   {
     set_full_table_name();
-    const char *index_name= index_info->name.str;
-    index_name_field->store(index_name, index_info->name.length,
-                            system_charset_info);
+    set_index_name(&index_info->name);
     table_key_info= index_info;
   }
 
@@ -1332,6 +1372,20 @@ public:
   }
 
 
+  int update_index_name(const LEX_CSTRING *name)
+  {
+    index_name_field->store(name, system_charset_info);
+    return update_record();
+  }
+
+
+  int read_next()
+  {
+    return stat_table->file->ha_index_next_same(stat_table->record[0],
+                                                last_key,
+                                                last_key_length);
+  }
+
   /** 
     @brief
     Store statistical data into statistical fields of table index_stats
@@ -1937,8 +1991,9 @@ public:
   @brief 
   Create fields for min/max values to collect column statistics
 
-  @param
-  table       Table the fields are created for
+  @param thd    The thread handle
+  @param table  Table the fields are created for
+  @param fields Fields for which we want to have statistics
 
   @details
   The function first allocates record buffers to store min/max values
@@ -1958,7 +2013,8 @@ public:
 */      
 
 static
-void create_min_max_statistical_fields_for_table(THD *thd, TABLE *table)
+void create_min_max_statistical_fields_for_table(THD *thd, TABLE *table,
+                                                 MY_BITMAP *fields)
 {
   uint rec_buff_length= table->s->rec_buff_length;
 
@@ -1975,7 +2031,7 @@ void create_min_max_statistical_fields_for_table(THD *thd, TABLE *table)
         Field *fld;
         Field *table_field= *field_ptr;
         my_ptrdiff_t diff= record-table->record[0];
-        if (!bitmap_is_set(table->read_set, table_field->field_index))
+        if (!bitmap_is_set(fields, table_field->field_index))
           continue; 
         if (!(fld= table_field->clone(thd->mem_root, table, diff)))
           continue;
@@ -2062,8 +2118,9 @@ create_min_max_statistical_fields(THD *thd,
   @brief 
   Allocate memory for the table's statistical data to be collected
 
-  @param
-  table       Table for which the memory for statistical data is allocated
+  @param thd          The thread handle
+  @param table        Table for which we should allocate statistical data
+  @param stat_fields  Fields for which we want to have statistics
 
   @note
   The function allocates the memory for the statistical data on 'table' with
@@ -2082,10 +2139,10 @@ create_min_max_statistical_fields(THD *thd,
   of the same table in parallel. 
 */      
 
-int alloc_statistics_for_table(THD* thd, TABLE *table)
+int alloc_statistics_for_table(THD* thd, TABLE *table, MY_BITMAP *stat_fields)
 { 
   Field **field_ptr;
-  uint fields= bitmap_bits_set(table->read_set);
+  uint fields= bitmap_bits_set(stat_fields);
   uint keys= table->s->keys;
   uint key_parts= table->s->ext_key_parts;
   uint hist_size= thd->variables.histogram_size;
@@ -2122,7 +2179,7 @@ int alloc_statistics_for_table(THD* thd, TABLE *table)
 
   for (field_ptr= table->field; *field_ptr; field_ptr++)
   {
-    if (bitmap_is_set(table->read_set, (*field_ptr)->field_index))
+    if (bitmap_is_set(stat_fields, (*field_ptr)->field_index))
     {
       column_stats->histogram.set_size(hist_size);
       column_stats->histogram.set_type(hist_type);
@@ -2154,7 +2211,7 @@ int alloc_statistics_for_table(THD* thd, TABLE *table)
   */
   DBUG_ASSERT(idx_avg_frequency <= table_stats->idx_avg_frequency + key_parts);
 
-  create_min_max_statistical_fields_for_table(thd, table);
+  create_min_max_statistical_fields_for_table(thd, table, stat_fields);
 
   DBUG_RETURN(0);
 }
@@ -2196,7 +2253,7 @@ int alloc_statistics_for_table(THD* thd, TABLE *table)
   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.
-*/      
+*/
 
 static int
 alloc_engine_independent_statistics(THD *thd, const TABLE_SHARE *table_share,
@@ -2686,10 +2743,8 @@ int collect_statistics_for_table(THD *thd, TABLE *table)
   @brief
   Update statistics for a table in the persistent statistical tables
 
-  @param
-  thd         The thread handle
-  @param
-  table       The table to collect statistics on
+  @param thd    The thread handle
+  @param table  The table to collect statistics on
 
   @details
   For each statistical table st the function looks for the rows from this
@@ -2734,7 +2789,10 @@ int update_statistics_for_table(THD *thd, TABLE *table)
   start_new_trans new_trans(thd);
 
   if (open_stat_tables(thd, tables, TRUE))
-    DBUG_RETURN(rc);
+  {
+    new_trans.restore_old_transaction();
+    DBUG_RETURN(0);
+  }
    
   /*
     Ensure that no one is reading satistics while we are writing them
@@ -2788,12 +2846,16 @@ int update_statistics_for_table(THD *thd, TABLE *table)
     }
   }
 
+  tables[TABLE_STAT].table->file->extra(HA_EXTRA_FLUSH);
+  tables[COLUMN_STAT].table->file->extra(HA_EXTRA_FLUSH);
+  tables[INDEX_STAT].table->file->extra(HA_EXTRA_FLUSH);
+
   thd->restore_stmt_binlog_format(save_binlog_format);
   if (thd->commit_whole_transaction_and_close_tables())
     rc= 1;
-  new_trans.restore_old_transaction();
 
   mysql_mutex_unlock(&table->s->LOCK_statistics);
+  new_trans.restore_old_transaction();
   DBUG_RETURN(rc);
 }
 
@@ -3058,6 +3120,7 @@ void TABLE_STATISTICS_CB::update_stats_in_table(TABLE *table)
 int
 read_statistics_for_tables(THD *thd, TABLE_LIST *tables, bool force_reload)
 {
+  int rc= 0;
   TABLE_LIST stat_tables[STATISTICS_TABLES];
   bool found_stat_table= false;
   bool statistics_for_tables_is_needed= false;
@@ -3122,7 +3185,10 @@ read_statistics_for_tables(THD *thd, TABLE_LIST *tables, bool force_reload)
   start_new_trans new_trans(thd);
 
   if (open_stat_tables(thd, stat_tables, FALSE))
-    DBUG_RETURN(1);
+  {
+    rc= 1;
+    goto end;
+  }
 
   for (TABLE_LIST *tl= tables; tl; tl= tl->next_global)
   {
@@ -3174,9 +3240,10 @@ read_statistics_for_tables(THD *thd, TABLE_LIST *tables, bool force_reload)
   }
 
   thd->commit_whole_transaction_and_close_tables();
-  new_trans.restore_old_transaction();
 
-  DBUG_RETURN(0);
+end:
+  new_trans.restore_old_transaction();
+  DBUG_RETURN(rc);
 }
 
 
@@ -3216,9 +3283,12 @@ int delete_statistics_for_table(THD *thd, const LEX_CSTRING *db,
   DBUG_ENTER("delete_statistics_for_table");
 
   start_new_trans new_trans(thd);
-   
+
   if (open_stat_tables(thd, tables, TRUE))
+  {
+    new_trans.restore_old_transaction();
     DBUG_RETURN(0);
+  }
 
   save_binlog_format= thd->set_current_stmt_binlog_format_stmt();
 
@@ -3259,10 +3329,14 @@ int delete_statistics_for_table(THD *thd, const LEX_CSTRING *db,
   if (err & !rc)
       rc= 1;
 
+  tables[TABLE_STAT].table->file->extra(HA_EXTRA_FLUSH);
+  tables[COLUMN_STAT].table->file->extra(HA_EXTRA_FLUSH);
+  tables[INDEX_STAT].table->file->extra(HA_EXTRA_FLUSH);
+
   thd->restore_stmt_binlog_format(save_binlog_format);
   thd->commit_whole_transaction_and_close_tables();
-  new_trans.restore_old_transaction();
 
+  new_trans.restore_old_transaction();
   DBUG_RETURN(rc);
 }
 
@@ -3299,7 +3373,10 @@ int delete_statistics_for_column(THD *thd, TABLE *tab, Field *col)
   start_new_trans new_trans(thd);
 
   if (open_stat_table_for_ddl(thd, &tables, &stat_table_name[1]))
-    DBUG_RETURN(0);
+  {
+    new_trans.restore_old_transaction();
+    DBUG_RETURN(0);                             // Not an error
+  }
 
   save_binlog_format= thd->set_current_stmt_binlog_format_stmt();
 
@@ -3313,11 +3390,202 @@ int delete_statistics_for_column(THD *thd, TABLE *tab, Field *col)
       rc= 1;
   }
 
+  column_stat.flush();
   thd->restore_stmt_binlog_format(save_binlog_format);
   if (thd->commit_whole_transaction_and_close_tables())
     rc= 1;
-  new_trans.restore_old_transaction();
 
+  new_trans.restore_old_transaction();
+  DBUG_RETURN(rc);
+}
+
+
+/**
+   Generate tempoary column or index name for renames
+*/
+
+static LEX_CSTRING *generate_tmp_name(LEX_CSTRING *to, uint counter)
+{
+  char *res=int10_to_str(counter, strmov((char*) to->str, "#sql_tmp_name#"),
+                         10);
+  /*
+    Include an end zero in the tmp name to avoid any possible conflict
+    with existing column names.
+   */
+  to->length= (size_t) (res - to->str) + 1;
+  return to;
+}
+
+
+/**
+  Rename a set of columns in the statistical table column_stats
+
+  @param thd         The thread handle
+  @param tab         The table the column belongs to
+  @param fields      List of fields and names to be renamed
+
+  @details
+  The function replaces the names of the columns in fields that belongs
+  to the table 'tab' in the statistical table column_stats.
+
+  @retval 0   If update was successful, tmp table or could not open stat table
+  @retval -1  Commit failed
+  @retval >0  Error number from engine
+
+  @note
+  The function is called when executing any statement that renames a column,
+  but does not change the column definition.
+*/
+
+int rename_columns_in_stat_table(THD *thd, TABLE *tab,
+                                 List<Alter_info::RENAME_COLUMN_STAT_PARAMS>
+                                 *fields)
+{
+  int err;
+  enum_binlog_format save_binlog_format;
+  TABLE *stat_table;
+  TABLE_LIST tables;
+  int rc= 0;
+  uint duplicate_counter= 0;
+  uint org_elements= fields->elements+1;
+  List_iterator<Alter_info::RENAME_COLUMN_STAT_PARAMS> it(*fields);
+  char tmp_name_buffer[32];
+  LEX_CSTRING tmp_name= {tmp_name_buffer, 0};
+  DBUG_ENTER("rename_column_in_stat_tables");
+
+  if (tab->s->tmp_table != NO_TMP_TABLE)
+    DBUG_RETURN(0);
+
+  start_new_trans new_trans(thd);
+
+  if (open_stat_table_for_ddl(thd, &tables, &stat_table_name[1]))
+  {
+    new_trans.restore_old_transaction();
+    DBUG_RETURN(0);
+  }
+
+  save_binlog_format= thd->set_current_stmt_binlog_format_stmt();
+
+  /* Rename column in the statistical table table_stat */
+
+  stat_table= tables.table;
+
+  /* Loop until fields is empty or previous round did nothing */
+  while (!fields->is_empty() && fields->elements != org_elements)
+  {
+    Alter_info::RENAME_COLUMN_STAT_PARAMS *field;
+    org_elements= fields->elements;
+    it.rewind();
+    while ((field= it++))
+    {
+      Column_stat column_stat(stat_table, tab);
+      LEX_CSTRING *from_name;
+      from_name= (!field->duplicate_counter ?
+                  &field->field->field_name :
+                  generate_tmp_name(&tmp_name,
+                                    field->duplicate_counter));
+      column_stat.set_key_fields(from_name);
+      if (column_stat.find_stat())
+      {
+        err= column_stat.update_column_key_part(field->name);
+        if (likely(err != HA_ERR_FOUND_DUPP_KEY))
+          it.remove();
+        else if (!field->duplicate_counter)
+        {
+          /*
+            This is probably an ALTER TABLE of type rename a->b, b->a
+            Rename the column to a temporary name
+          */
+          LEX_CSTRING *new_name=
+            generate_tmp_name(&tmp_name, ++duplicate_counter);
+          field->duplicate_counter= duplicate_counter;
+
+          if ((err= column_stat.update_column(new_name)))
+          {
+            if (likely(err != HA_ERR_FOUND_DUPP_KEY))
+            {
+              DBUG_ASSERT(0);
+              it.remove();                      // Unknown error, ignore column
+            }
+            else
+            {
+              /*
+                The only way this could happen is if the table has a column
+                with same name as the temporary column name, probably from a
+                failed alter table.
+                Remove the conflicting row and update it again.
+              */
+              if (!column_stat.find_stat())
+                DBUG_ASSERT(0);
+              else if (column_stat.delete_stat())
+                DBUG_ASSERT(0);
+              else
+              {
+                column_stat.set_key_fields(from_name);
+                if (!column_stat.find_stat())
+                  DBUG_ASSERT(0);
+                else if (column_stat.update_column_key_part(&tmp_name))
+                  DBUG_ASSERT(0);
+              }
+            }
+          }
+        }
+      }
+      else /* column_stat.find_stat() */
+      {
+        /* Statistics for the field did not exists */
+        it.remove();
+      }
+    }
+  }
+
+  if (!fields->is_empty())
+  {
+    /*
+      All unhandled renamed fields has now a temporary name.
+      Remove all conflicing rows and rename the temporary name to
+      the final name.
+    */
+
+    Alter_info::RENAME_COLUMN_STAT_PARAMS *field;
+    it.rewind();
+    while ((field= it++))
+    {
+      Column_stat column_stat(stat_table, tab);
+      DBUG_ASSERT(field->duplicate_counter);
+
+      /* Remove the conflicting row */
+      column_stat.set_key_fields(field->name);
+      if (column_stat.find_stat())
+      {
+        int err __attribute__((unused));
+        err= column_stat.delete_stat();
+        DBUG_ASSERT(err == 0);
+      }
+
+      /* Restore saved row with old statistics to new name */
+      column_stat.
+        set_key_fields(generate_tmp_name(&tmp_name,
+                                         field->duplicate_counter));
+      if (column_stat.find_stat())
+      {
+        int err __attribute__((unused));
+        err= column_stat.update_column_key_part(field->name);
+        DBUG_ASSERT(err == 0);
+      }
+      else
+      {
+        DBUG_ASSERT(0);
+      }
+    }
+  }
+
+  stat_table->file->extra(HA_EXTRA_FLUSH);
+  thd->restore_stmt_binlog_format(save_binlog_format);
+  if (thd->commit_whole_transaction_and_close_tables())
+    rc= -1;
+
+  new_trans.restore_old_transaction();
   DBUG_RETURN(rc);
 }
 
@@ -3359,7 +3627,10 @@ int delete_statistics_for_index(THD *thd, TABLE *tab, KEY *key_info,
   start_new_trans new_trans(thd);
 
   if (open_stat_table_for_ddl(thd, &tables, &stat_table_name[2]))
-    DBUG_RETURN(0);
+  {
+    new_trans.restore_old_transaction();
+    DBUG_RETURN(0);                             // Not an error
+  }
 
   save_binlog_format= thd->set_current_stmt_binlog_format_stmt();
 
@@ -3393,11 +3664,194 @@ int delete_statistics_for_index(THD *thd, TABLE *tab, KEY *key_info,
   if (err && !rc)
     rc= 1;
 
+  /* Make change permanent and avoid 'table is marked as crashed' errors */
+  index_stat.flush();
+
   thd->restore_stmt_binlog_format(save_binlog_format);
   if (thd->commit_whole_transaction_and_close_tables())
     rc= 1;
-  new_trans.restore_old_transaction();
 
+  new_trans.restore_old_transaction();
+  DBUG_RETURN(rc);
+}
+
+
+/**
+  Rename a set of indexes in the statistical table index_stats
+
+  @param thd         The thread handle
+  @param tab         The table the indexes belongs to
+  @param fields      List of indexes to be renamed
+
+  @details
+  The function replaces the names of the indexe in fields that belongs
+  to the table 'tab' in the statistical table index_stats.
+
+  @retval 0   If update was successful, tmp table or could not open stat table
+  @retval -1  Commit failed
+  @retval >0  Error number from engine
+
+  @note
+  The function is called when executing any statement that renames a column,
+  but does not change the column definition.
+*/
+
+int rename_indexes_in_stat_table(THD *thd, TABLE *tab,
+                                 List<Alter_info::RENAME_INDEX_STAT_PARAMS>
+                                 *indexes)
+{
+  int err;
+  enum_binlog_format save_binlog_format;
+  TABLE *stat_table;
+  TABLE_LIST tables;
+  int rc= 0;
+  uint duplicate_counter= 0;
+  List_iterator<Alter_info::RENAME_INDEX_STAT_PARAMS> it(*indexes);
+  Alter_info::RENAME_INDEX_STAT_PARAMS *index;
+  char tmp_name_buffer[32];
+  LEX_CSTRING tmp_name= {tmp_name_buffer, 0};
+  DBUG_ENTER("rename_indexes_in_stat_tables");
+
+  if (tab->s->tmp_table != NO_TMP_TABLE)
+    DBUG_RETURN(0);
+
+  start_new_trans new_trans(thd);
+
+  if (open_stat_table_for_ddl(thd, &tables, &stat_table_name[2]))
+  {
+    new_trans.restore_old_transaction();
+    DBUG_RETURN(0);
+  }
+
+  save_binlog_format= thd->set_current_stmt_binlog_format_stmt();
+
+  /* Rename index in the statistical table index_stat */
+
+  stat_table= tables.table;
+
+  /*
+    Loop over all indexes and rename to new name or temp name in case of
+    conflicts
+  */
+
+  while ((index= it++))
+  {
+    Index_stat index_stat(stat_table, tab);
+    uint found= 0;
+
+    /* We have to make a loop as one index may have many entries */
+    for (;;)
+    {
+      index_stat.set_index_prefix_key_fields(index->key);
+      if (!index_stat.find_next_stat_for_prefix(3))
+        break;
+      index_stat.store_record_for_update();
+      err= index_stat.update_index_name(index->name);
+
+      if (unlikely(err == HA_ERR_FOUND_DUPP_KEY))
+      {
+        /*
+          This is probably an ALTER TABLE of type rename a->b, b->a
+          Rename the column to a temporary name
+        */
+        if (!found++)
+          ++duplicate_counter;
+        index->duplicate_counter= duplicate_counter;
+        index->usage_count++;
+        if ((err= index_stat.update_index_name(generate_tmp_name(&tmp_name, duplicate_counter))))
+        {
+          if (err != HA_ERR_FOUND_DUPP_KEY)
+          {
+            DBUG_ASSERT(0);
+          }
+          else
+          {
+            /*
+              The only way this could happen is if the table has an index
+              with same name as the temporary column index, probably from a
+              failed alter table.
+              Remove the conflicting row and update it again.
+            */
+            if (!index_stat.find_stat())
+              DBUG_ASSERT(0);
+            else if (index_stat.delete_stat())
+              DBUG_ASSERT(0);
+            else
+            {
+              index_stat.set_index_prefix_key_fields(index->key);
+              if (!index_stat.find_stat())
+                DBUG_ASSERT(0);
+              else
+              {
+                index_stat.store_record_for_update();
+                if (index_stat.update_index_name(&tmp_name))
+                  DBUG_ASSERT(0);
+              }
+            }
+          }
+        }
+      }
+    }
+    if (!found)
+      it.remove();                              // All renames succeded
+  }
+
+  if (!indexes->is_empty())
+  {
+    /*
+      All unhandled renamed index has now a temporary name.
+      Remove all conflicing rows and rename the temporary name to
+      the final name.
+    */
+
+    Alter_info::RENAME_INDEX_STAT_PARAMS *index;
+    it.rewind();
+    Index_stat index_stat(stat_table, tab);
+    stat_table->file->ha_index_init(index_stat.stat_key_idx, 0);
+
+    while ((index= it++))
+    {
+      int err __attribute__((unused));
+
+      /* Remove the conflicting rows */
+      index_stat.set_index_prefix_key_fields(index->key);
+      index_stat.set_index_name(index->name);
+
+      if (index_stat.find_next_stat_for_prefix_with_next(3))
+      {
+        do
+        {
+          err= index_stat.delete_stat();
+          DBUG_ASSERT(err == 0);
+        }
+        while (index_stat.read_next() == 0);
+      }
+
+      /* Restore saved row with old statistics to new name */
+      index_stat.set_index_name(generate_tmp_name(&tmp_name,
+                                                  index->duplicate_counter));
+      if (!index_stat.find_stat_with_next())
+        DBUG_ASSERT(0);
+      else
+      {
+        uint updated= 0;
+        do
+        {
+          index_stat.store_record_for_update();
+          err= index_stat.update_index_name(index->name);
+          DBUG_ASSERT(err == 0);
+        } while (++updated < index->usage_count && index_stat.read_next() == 0);
+      }
+    }
+    stat_table->file->ha_index_end();
+  }
+
+  stat_table->file->extra(HA_EXTRA_FLUSH);
+  thd->restore_stmt_binlog_format(save_binlog_format);
+  if (thd->commit_whole_transaction_and_close_tables())
+    rc= -1;
+
+  new_trans.restore_old_transaction();
   DBUG_RETURN(rc);
 }
 
@@ -3444,7 +3898,10 @@ int rename_table_in_stat_tables(THD *thd, const LEX_CSTRING *db,
   start_new_trans new_trans(thd);
 
   if (open_stat_tables(thd, tables, TRUE))
-    DBUG_RETURN(0); // not an error
+  {
+    new_trans.restore_old_transaction();
+    DBUG_RETURN(0);
+  }
 
   save_binlog_format= thd->set_current_stmt_binlog_format_stmt();
 
@@ -3492,71 +3949,15 @@ int rename_table_in_stat_tables(THD *thd, const LEX_CSTRING *db,
       rc= 1;
   }
 
-  thd->restore_stmt_binlog_format(save_binlog_format);
-  if (thd->commit_whole_transaction_and_close_tables())
-    rc= 1;
-  new_trans.restore_old_transaction();
-
-  DBUG_RETURN(rc);
-}
-
-
-/**
-  Rename a column in the statistical table column_stats
-
-  @param thd         The thread handle
-  @param tab         The table the column belongs to
-  @param col         The column to be renamed
-  @param new_name    The new column name
-
-  @details
-  The function replaces the name of the column 'col' belonging to the table 
-  'tab' for 'new_name' in the statistical table column_stats.
-
-  @retval 0  If all updates of the table name are successful
-  @retval 1  Otherwise
-
-  @note
-  The function is called when executing any statement that renames a column,
-  but does not change the column definition.
-*/
-
-int rename_column_in_stat_tables(THD *thd, TABLE *tab, Field *col,
-                                 const char *new_name)
-{
-  int err;
-  enum_binlog_format save_binlog_format;
-  TABLE *stat_table;
-  TABLE_LIST tables;
-  int rc= 0;
-  DBUG_ENTER("rename_column_in_stat_tables");
-  
-  if (tab->s->tmp_table != NO_TMP_TABLE)
-    DBUG_RETURN(0);
-
-  start_new_trans new_trans(thd);
-
-  if (open_stat_table_for_ddl(thd, &tables, &stat_table_name[1]))
-    DBUG_RETURN(rc);
-
-  save_binlog_format= thd->set_current_stmt_binlog_format_stmt();
-
-  /* Rename column in the statistical table table_stat */
-  stat_table= tables.table;
-  Column_stat column_stat(stat_table, tab);
-  column_stat.set_key_fields(col);
-  if (column_stat.find_stat())
-  { 
-    err= column_stat.update_column_key_part(new_name);
-    if (err & !rc)
-      rc= 1;
-  }
+  tables[TABLE_STAT].table->file->extra(HA_EXTRA_FLUSH);
+  tables[COLUMN_STAT].table->file->extra(HA_EXTRA_FLUSH);
+  tables[INDEX_STAT].table->file->extra(HA_EXTRA_FLUSH);
 
   thd->restore_stmt_binlog_format(save_binlog_format);
   if (thd->commit_whole_transaction_and_close_tables())
     rc= 1;
-  new_trans.restore_old_transaction();
 
+  new_trans.restore_old_transaction();
   DBUG_RETURN(rc);
 }
 
diff --git a/sql/sql_statistics.h b/sql/sql_statistics.h
index f5ccb26af7b..5c98451808a 100644
--- a/sql/sql_statistics.h
+++ b/sql/sql_statistics.h
@@ -119,16 +119,21 @@ int read_statistics_for_tables_if_needed(THD *thd, TABLE_LIST *tables);
 int read_statistics_for_tables(THD *thd, TABLE_LIST *tables,
                                bool force_reload);
 int collect_statistics_for_table(THD *thd, TABLE *table);
-int alloc_statistics_for_table(THD *thd, TABLE *table);
+int alloc_statistics_for_table(THD *thd, TABLE *table, MY_BITMAP *stat_fields);
 int update_statistics_for_table(THD *thd, TABLE *table);
-int delete_statistics_for_table(THD *thd, const LEX_CSTRING *db, const LEX_CSTRING *tab);
+int delete_statistics_for_table(THD *thd, const LEX_CSTRING *db,
+                                const LEX_CSTRING *tab);
 int delete_statistics_for_column(THD *thd, TABLE *tab, Field *col);
 int delete_statistics_for_index(THD *thd, TABLE *tab, KEY *key_info,
                                 bool ext_prefixes_only);
-int rename_table_in_stat_tables(THD *thd, const LEX_CSTRING *db, const LEX_CSTRING *tab,
-                                const LEX_CSTRING *new_db, const LEX_CSTRING *new_tab);
-int rename_column_in_stat_tables(THD *thd, TABLE *tab, Field *col,
-                                  const char *new_name);
+int rename_table_in_stat_tables(THD *thd, const LEX_CSTRING *db,
+                                const LEX_CSTRING *tab,
+                                const LEX_CSTRING *new_db,
+                                const LEX_CSTRING *new_tab);
+int rename_columns_in_stat_table(THD *thd, TABLE *tab,
+                                 List<Alter_info::RENAME_COLUMN_STAT_PARAMS> *fields);
+int rename_indexes_in_stat_table(THD *thd, TABLE *tab,
+                                 List<Alter_info::RENAME_INDEX_STAT_PARAMS> *indexes);
 void set_statistics_for_table(THD *thd, TABLE *table);
 
 double get_column_avg_frequency(Field * field);
diff --git a/sql/sql_table.cc b/sql/sql_table.cc
index af365d6e7f6..9e20db8b5e4 100644
--- a/sql/sql_table.cc
+++ b/sql/sql_table.cc
@@ -2803,8 +2803,6 @@ mysql_prepare_create_table(THD *thd, HA_CREATE_INFO *create_info,
   bool primary_key=0,unique_key=0;
   Key *key, *key2;
   uint tmp, key_number;
-  /* special marker for keys to be ignored */
-  static char ignore_key[1];
 
   /* Calculate number of key segements */
   *key_count= 0;
@@ -2852,17 +2850,17 @@ mysql_prepare_create_table(THD *thd, HA_CREATE_INFO *create_info,
           both, is 'generated', and a generated key is a prefix of the other
           key. Then we do not need the generated shorter key.
         */
-        if (key2->type != Key::FOREIGN_KEY && key2->name.str != ignore_key &&
+        if (key2->type != Key::FOREIGN_KEY && key2->type != Key::IGNORE_KEY &&
             is_foreign_key_prefix(key, key2))
         {
           /* mark that the generated key should be ignored */
           if (!key2->generated ||
               (key->generated && key->columns.elements <
                key2->columns.elements))
-            key->name.str= ignore_key;
+            key->type= Key::IGNORE_KEY;
           else
           {
-            key2->name.str= ignore_key;
+            key2->type= Key::IGNORE_KEY;
             key_parts-= key2->columns.elements;
             (*key_count)--;
           }
@@ -2870,7 +2868,7 @@ mysql_prepare_create_table(THD *thd, HA_CREATE_INFO *create_info,
         }
       }
     }
-    if (key->name.str != ignore_key)
+    if (key->type != Key::IGNORE_KEY)
       key_parts+=key->columns.elements;
     else
       (*key_count)--;
@@ -2900,7 +2898,14 @@ mysql_prepare_create_table(THD *thd, HA_CREATE_INFO *create_info,
   key_iterator.rewind();
   while ((key=key_iterator++))
   {
-    if (key->name.str == ignore_key || key->type == Key::FOREIGN_KEY)
+    if (key->type == Key::IGNORE_KEY)
+    {
+      /* The key was replaced by another key */
+      if (alter_info->add_stat_drop_index(thd, &key->name))
+        DBUG_RETURN(true);
+      continue;
+    }
+    if (key->type == Key::FOREIGN_KEY)
       continue;
     /* Create the key name based on the first column (if not given) */
     if (key->type == Key::PRIMARY)
@@ -2962,12 +2967,12 @@ mysql_prepare_create_table(THD *thd, HA_CREATE_INFO *create_info,
     Key_part_spec *column;
 
     is_hash_field_needed= false;
-    if (key->name.str == ignore_key)
+    if (key->type == Key::IGNORE_KEY)
     {
       /* ignore redundant keys */
       do
 	key=key_iterator++;
-      while (key && key->name.str == ignore_key);
+      while (key && key->type == Key::IGNORE_KEY);
       if (!key)
 	break;
     }
@@ -2995,6 +3000,9 @@ mysql_prepare_create_table(THD *thd, HA_CREATE_INFO *create_info,
     case Key::FOREIGN_KEY:
       key_number--;				// Skip this key
       continue;
+    case Key::IGNORE_KEY:
+      DBUG_ASSERT(0);
+      break;
     default:
       key_info->flags = HA_NOSAME;
       break;
@@ -3198,6 +3206,8 @@ mysql_prepare_create_table(THD *thd, HA_CREATE_INFO *create_info,
             key_add_part_check_null(file, key_info, sql_field, column))
           DBUG_RETURN(TRUE);
         break;
+      case Key::IGNORE_KEY:
+        break;
       }
 
       if (MTYP_TYPENR(sql_field->unireg_check) == Field::NEXT_NUMBER)
@@ -6445,7 +6455,9 @@ static bool fill_alter_inplace_info(THD *thd, TABLE *table, bool varchar,
 
           if (table->s->tmp_table == NO_TMP_TABLE)
           {
-            delete_statistics_for_column(thd, table, field);
+            if (alter_info->drop_stat_fields.push_back(field, thd->mem_root))
+              DBUG_RETURN(true);
+
             KEY *key_info= table->key_info;
             for (uint i= 0; i < table->s->keys; i++, key_info++)
             {
@@ -6457,9 +6469,10 @@ static bool fill_alter_inplace_info(THD *thd, TABLE *table, bool varchar,
               {
                 if (key_info->key_part[j].fieldnr - 1 == field->field_index)
                 {
-                  delete_statistics_for_index(
-                      thd, table, key_info,
-                      j >= key_info->user_defined_key_parts);
+                  if (alter_info->add_stat_drop_index(key_info,
+                                                      j >= key_info->user_defined_key_parts,
+                                                      thd->mem_root))
+                    DBUG_RETURN(true);
                   break;
                 }
               }
@@ -6516,13 +6529,17 @@ static bool fill_alter_inplace_info(THD *thd, TABLE *table, bool varchar,
           ha_alter_info->handler_flags|= ALTER_STORED_COLUMN_TYPE;
       }
 
-      /* Check if field was renamed (case-sensitive for detecting case change) */
+      /*
+        Check if field was renamed (case-sensitive for detecting case change)
+      */
       if (cmp(&field->field_name, &new_field->field_name))
       {
         field->flags|= FIELD_IS_RENAMED;
         ha_alter_info->handler_flags|= ALTER_COLUMN_NAME;
-        rename_column_in_stat_tables(thd, table, field,
-                                     new_field->field_name.str);
+        if (alter_info->add_stat_rename_field(field,
+                                              &new_field->field_name,
+                                              thd->mem_root))
+          DBUG_RETURN(true);
       }
 
       /* Check that NULL behavior is same for old and new fields */
@@ -7481,6 +7498,8 @@ static bool mysql_inplace_alter_table(THD *thd,
     in case of crash it should use the new one and log the query
     to the binary log.
   */
+  ha_alter_info->alter_info->apply_statistics_deletes_renames(thd, table);
+
   ddl_log_update_phase(ddl_log_state, DDL_ALTER_TABLE_PHASE_INPLACE_COPIED);
   debug_crash_here("ddl_log_alter_after_log");
 
@@ -7585,6 +7604,7 @@ static bool mysql_inplace_alter_table(THD *thd,
   DBUG_RETURN(true);
 }
 
+
 /**
   maximum possible length for certain blob types.
 
@@ -7833,7 +7853,10 @@ mysql_prepare_alter_table(THD *thd, TABLE *table,
         create_info->used_fields|=HA_CREATE_USED_AUTO;
       }
       if (table->s->tmp_table == NO_TMP_TABLE)
-        (void) delete_statistics_for_column(thd, table, field);
+      {
+        if (alter_info->drop_stat_fields.push_back(field, thd->mem_root))
+          DBUG_RETURN(true);
+      }
       dropped_sys_vers_fields|= field->flags;
       drop_it.remove();
       dropped_fields= &table->tmp_set;
@@ -7845,13 +7868,19 @@ mysql_prepare_alter_table(THD *thd, TABLE *table,
     {
       vers_system_invisible= true;
     }
-    /* invisible versioning column is dropped automatically on DROP SYSTEM VERSIONING */
+    /*
+      invisible versioning column is dropped automatically on
+      DROP SYSTEM VERSIONING
+    */
     if (!drop && field->invisible >= INVISIBLE_SYSTEM &&
         field->flags & VERS_SYSTEM_FIELD &&
         alter_info->flags & ALTER_DROP_SYSTEM_VERSIONING)
     {
       if (table->s->tmp_table == NO_TMP_TABLE)
-        (void) delete_statistics_for_column(thd, table, field);
+      {
+        if (alter_info->drop_stat_fields.push_back(field, thd->mem_root))
+          DBUG_RETURN(true);
+      }
       continue;
     }
 
@@ -8214,19 +8243,24 @@ mysql_prepare_alter_table(THD *thd, TABLE *table,
     {
       if (table->s->tmp_table == NO_TMP_TABLE)
       {
-        (void) delete_statistics_for_index(thd, table, key_info, FALSE);
+        if (alter_info->add_stat_drop_index(key_info, FALSE, thd->mem_root))
+          DBUG_RETURN(true);
         if (primary_key)
 	{
           KEY *tab_key_info= table->key_info;
 	  for (uint j=0; j < table->s->keys; j++, tab_key_info++)
 	  {
-            if (tab_key_info->user_defined_key_parts !=
+            if (tab_key_info != key_info &&
+                tab_key_info->user_defined_key_parts !=
                 tab_key_info->ext_key_parts)
-	      (void) delete_statistics_for_index(thd, table, tab_key_info,
-                                                 TRUE);
-	  }
+            {
+              if (alter_info->add_stat_drop_index(tab_key_info, TRUE,
+                                                  thd->mem_root))
+                DBUG_RETURN(true);
+            }
+          }
 	}
-      }  
+      }
       drop_it.remove();
       continue;
     }
@@ -8263,8 +8297,11 @@ mysql_prepare_alter_table(THD *thd, TABLE *table,
           goto err;
         }
 
-        key_name= rename_key->new_name.str;
+        key_name= rename_key->new_name.str;     // New name of current key_info
         rename_key_it.remove();
+        alter_info->add_stat_rename_index(key_info, &rename_key->new_name,
+                                          thd->mem_root);
+
         /*
           If the user has explicitly renamed the key, we should no longer
           treat it as generated. Otherwise this key might be automatically
@@ -8374,10 +8411,17 @@ mysql_prepare_alter_table(THD *thd, TABLE *table,
     if (table->s->tmp_table == NO_TMP_TABLE)
     {
       if (delete_index_stat) 
-        (void) delete_statistics_for_index(thd, table, key_info, FALSE);
+      {
+        if (alter_info->add_stat_drop_index(key_info, FALSE, thd->mem_root))
+          DBUG_RETURN(true);
+      }
       else if (alter_ctx->modified_primary_key &&
                key_info->user_defined_key_parts != key_info->ext_key_parts)
-        (void) delete_statistics_for_index(thd, table, key_info, TRUE);
+      {
+        if (alter_info->add_stat_drop_index(key_info, FALSE,
+                                               thd->mem_root))
+          DBUG_RETURN(true);
+      }
     }
 
     if (!user_keyparts && key_parts.elements)
@@ -10196,6 +10240,13 @@ do_continue:;
     alter_info->flags|= ALTER_INDEX_ORDER;
   create_info->alias= alter_ctx.table_name;
   thd->abort_on_warning= !ignore && thd->is_strict_mode();
+
+  /*
+    This is to be able to call Alter_info::add_stat_drop_index(thd, key_name)
+    from mysql_prepare_create_table()
+  */
+  alter_info->original_table= table;
+
   /*
     Create the .frm file for the new table. Storage engine table will not be
     created at this stage.
@@ -10466,6 +10517,16 @@ do_continue:;
   thd->count_cuted_fields= CHECK_FIELD_WARN;	// calc cuted fields
   thd->cuted_fields=0L;
 
+  /*
+    Collect fields that was renamed.
+    We do not do that if fill_alter_inplace_info() has
+    already collected renamed fields.
+  */
+  if (alter_info->flags & (ALTER_CHANGE_COLUMN | ALTER_RENAME_COLUMN) &&
+      alter_info->rename_stat_fields.is_empty())
+    if (alter_info->collect_renamed_fields(thd))
+      goto err_new_table_cleanup;
+
   /*
     We do not copy data for MERGE tables. Only the children have data.
     MERGE tables have HA_NO_COPY_ON_ALTER set.
@@ -10662,6 +10723,9 @@ do_continue:;
   if (wait_while_table_is_used(thd, table, HA_EXTRA_PREPARE_FOR_RENAME))
     goto err_new_table_cleanup;
 
+  /* Now we are the only user. Update the data in EITS tables */
+  alter_info->apply_statistics_deletes_renames(thd, table);
+
   close_all_tables_for_name(thd, table->s,
                             alter_ctx.is_table_renamed() ?
                             HA_EXTRA_PREPARE_FOR_RENAME:
diff --git a/sql/structs.h b/sql/structs.h
index 0e4ca6236fe..2818d43a06a 100644
--- a/sql/structs.h
+++ b/sql/structs.h
@@ -95,11 +95,11 @@ class engine_option_value;
 struct ha_index_option_struct;
 
 typedef struct st_key {
-  uint	key_length;			/* total length of user defined key parts  */
-  ulong flags;                          /* dupp key and pack flags */
-  uint	user_defined_key_parts;	   /* How many key_parts */
-  uint	usable_key_parts; /* Should normally be = user_defined_key_parts */
-  uint ext_key_parts;              /* Number of key parts in extended key */
+  uint  key_length;                /* total length of user defined key parts  */
+  ulong flags;                     /* dupp key and pack flags */
+  uint  user_defined_key_parts;    /* How many key_parts */
+  uint  usable_key_parts; /* Should normally be = user_defined_key_parts */
+  uint  ext_key_parts;             /* Number of key parts in extended key */
   ulong ext_key_flags;             /* Flags for extended key              */
   /*
     Parts of primary key that are in the extension of this index.