MDEV-11415 Remove excessive undo logging during ALTER TABLE…ALGORITHM=COPY

If a crash occurs during ALTER TABLE…ALGORITHM=COPY, InnoDB would spend
a lot of time rolling back writes to the intermediate copy of the table.
To reduce the amount of busy work done, a work-around was introduced in
commit fd069e2bb36a3c1c1f26d65dd298b07e6d83ac8b in MySQL 4.1.8 and 5.0.2,
to commit the transaction after every 10,000 inserted rows.

A proper fix would have been to disable the undo logging altogether and
to simply drop the intermediate copy of the table on subsequent server
startup. This is what happens in MariaDB 10.3 with MDEV-14717,MDEV-14585.
In MariaDB 10.2, the intermediate copy of the table would be left behind
with a name starting with the string #sql.

This is a backport of a bug fix from MySQL 8.0.0 to MariaDB,
contributed by jixianliang <271365745@qq.com>.

Unlike recent MySQL, MariaDB supports ALTER IGNORE. For that operation
InnoDB must for now keep the undo logging enabled, so that the latest
row can be rolled back in case of an error.

In Galera cluster, the LOAD DATA statement will retain the existing
behaviour and commit the transaction after every 10,000 rows if
the parameter wsrep_load_data_splitting=ON is set. The logic to do
so (the wsrep_load_data_split() function and the call
handler::extra(HA_EXTRA_FAKE_START_STMT)) are joint work
by Ji Xianliang and Marko Mäkelä.

The original fix:

Author: Thirunarayanan Balathandayuthapani <thirunarayanan.balathandayuth@oracle.com>
Date:   Wed Dec 2 16:09:15 2015 +0530

Bug#17479594 AVOID INTERMEDIATE COMMIT WHILE DOING ALTER TABLE ALGORITHM=COPY

Problem:

During ALTER TABLE, we commit and restart the transaction for every
10,000 rows, so that the rollback after recovery would not take so long.

Fix:

Suppress the undo logging during copy alter operation. If fts_index is
present then insert directly into fts auxiliary table rather
than doing at commit time.

ha_innobase::num_write_row: Remove the variable.

ha_innobase::write_row(): Remove the hack for committing every 10000 rows.

row_lock_table_for_mysql(): Remove the extra 2 parameters.

lock_get_src_table(), lock_is_table_exclusive(): Remove.

Reviewed-by: Marko Mäkelä <marko.makela@oracle.com>
Reviewed-by: Shaohua Wang <shaohua.wang@oracle.com>
Reviewed-by: Jon Olav Hauglid <jon.hauglid@oracle.com>

24 files changed, 640 insertions, 357 deletions

diff --git a/include/my_base.h b/include/my_base.h
index 89f5e826fd5..a01744f8a50 100644
--- a/include/my_base.h
+++ b/include/my_base.h
@@ -1,5 +1,5 @@
 /* Copyright (c) 2000, 2012, Oracle and/or its affiliates.
-   Copyright (c) 1995, 2017, MariaDB Corporation.
+   Copyright (c) 1995, 2018, MariaDB Corporation.
 
    This program is free software; you can redistribute it and/or modify
    it under the terms of the GNU General Public License as published by
@@ -200,7 +200,13 @@ enum ha_extra_function {
   HA_EXTRA_DETACH_CHILDREN,
   HA_EXTRA_DETACH_CHILD,
   /* Inform handler we will force a close as part of flush */
-  HA_EXTRA_PREPARE_FOR_FORCED_CLOSE
+  HA_EXTRA_PREPARE_FOR_FORCED_CLOSE,
+  /** Start writing rows during ALTER TABLE...ALGORITHM=COPY. */
+  HA_EXTRA_BEGIN_ALTER_COPY,
+  /** Finish writing rows during ALTER TABLE...ALGORITHM=COPY. */
+  HA_EXTRA_END_ALTER_COPY,
+  /** Fake the start of a statement after wsrep_load_data_splitting hack */
+  HA_EXTRA_FAKE_START_STMT
 };
 
 /* Compatible option, to be deleted in 6.0 */
diff --git a/mysql-test/suite/innodb/r/alter_copy.result b/mysql-test/suite/innodb/r/alter_copy.result
new file mode 100644
index 00000000000..9815560d22c
--- /dev/null
+++ b/mysql-test/suite/innodb/r/alter_copy.result
@@ -0,0 +1,215 @@
+#
+# MDEV-11415 AVOID INTERMEDIATE COMMIT WHILE DOING
+#            ALTER TABLE...ALGORITHM=COPY
+#
+CREATE TABLE t(a SERIAL, b INT, c INT, d INT) ENGINE=InnoDB;
+CREATE TABLE t1(a INT, b TEXT, c TEXT,
+FULLTEXT(b), FULLTEXT(c(3)), FULLTEXT(b,c)) ENGINE=InnoDB;
+BEGIN;
+COMMIT;
+SELECT COUNT(*) FROM t;
+COUNT(*)
+999
+UPDATE t SET b=a%7, c=a%11, d=a%13;
+INSERT INTO t1 VALUES(1, 'This is a first b column', 'This is a first c column');
+INSERT INTO t1 VALUES(2, 'This is a second b column', 'This is a second c column');
+INSERT INTO t1(a) VALUES(3);
+INSERT INTO t1 VALUES(4, 'This is a third b column', 'This is a third c column');
+DELETE FROM t1 WHERE a = 2;
+SELECT * FROM t1 WHERE MATCH(b) AGAINST ('first');
+a	b	c
+1	This is a first b column	This is a first c column
+SELECT * FROM t1 WHERE MATCH(c) AGAINST ('first');
+a	b	c
+1	This is a first b column	This is a first c column
+SELECT * FROM t1 WHERE MATCH(b,c) AGAINST ('column');
+a	b	c
+1	This is a first b column	This is a first c column
+4	This is a third b column	This is a third c column
+SHOW CREATE TABLE t1;
+Table	Create Table
+t1	CREATE TABLE `t1` (
+  `a` int(11) DEFAULT NULL,
+  `b` text DEFAULT NULL,
+  `c` text DEFAULT NULL,
+  FULLTEXT KEY `b` (`b`),
+  FULLTEXT KEY `c` (`c`),
+  FULLTEXT KEY `b_2` (`b`,`c`)
+) ENGINE=InnoDB DEFAULT CHARSET=latin1
+ALTER TABLE t1 FORCE, ALGORITHM=COPY;
+SET DEBUG_DBUG='+d,crash_commit_before';
+ALTER TABLE t ADD INDEX(b,c,d,a),ADD INDEX(b,c,a,d),ADD INDEX(b,a,c,d),ADD INDEX(b,a,d,c),
+ADD INDEX(b,d,a,c),ADD INDEX(b,d,c,a),ADD INDEX(a,b,c,d),ADD INDEX(a,b,d,c),
+ADD INDEX(a,c,b,d),ADD INDEX(a,c,d,b),ADD INDEX(a,d,b,c),ADD INDEX(a,d,c,b),
+ADD INDEX(c,a,b,d),ADD INDEX(c,a,d,b),ADD INDEX(c,b,a,d),ADD INDEX(c,b,d,a),
+ADD INDEX(c,d,a,b),ADD INDEX(c,d,b,a),ADD INDEX(d,a,b,c),ADD INDEX(d,a,c,b),
+ADD INDEX(d,b,a,c),ADD INDEX(d,b,c,a),ADD INDEX(d,c,a,b),ADD INDEX(d,c,b,a),
+ADD INDEX(a,b,c), ADD INDEX(a,c,b), ADD INDEX(a,c,d), ADD INDEX(a,d,c),
+ADD INDEX(a,b,d), ADD INDEX(a,d,b), ADD INDEX(b,c,d), ADD INDEX(b,d,c),
+ALGORITHM=COPY;
+ERROR HY000: Lost connection to MySQL server during query
+#sql-temporary.frm
+#sql-temporary.ibd
+FTS_INDEX_1.ibd
+FTS_INDEX_2.ibd
+FTS_INDEX_3.ibd
+FTS_INDEX_4.ibd
+FTS_INDEX_5.ibd
+FTS_INDEX_6.ibd
+FTS_INDEX_1.ibd
+FTS_INDEX_2.ibd
+FTS_INDEX_3.ibd
+FTS_INDEX_4.ibd
+FTS_INDEX_5.ibd
+FTS_INDEX_6.ibd
+FTS_INDEX_1.ibd
+FTS_INDEX_2.ibd
+FTS_INDEX_3.ibd
+FTS_INDEX_4.ibd
+FTS_INDEX_5.ibd
+FTS_INDEX_6.ibd
+FTSBEING_DELETED.ibd
+FTSBEING_DELETED_CACHE.ibd
+FTSCONFIG.ibd
+FTSDELETED.ibd
+FTSDELETED_CACHE.ibd
+t.frm
+t.ibd
+t1.frm
+t1.ibd
+SHOW CREATE TABLE t;
+Table	Create Table
+t	CREATE TABLE `t` (
+  `a` bigint(20) unsigned NOT NULL AUTO_INCREMENT,
+  `b` int(11) DEFAULT NULL,
+  `c` int(11) DEFAULT NULL,
+  `d` int(11) DEFAULT NULL,
+  UNIQUE KEY `a` (`a`)
+) ENGINE=InnoDB AUTO_INCREMENT=1000 DEFAULT CHARSET=latin1
+SELECT COUNT(*) FROM t;
+COUNT(*)
+999
+CHECK TABLE t;
+Table	Op	Msg_type	Msg_text
+test.t	check	status	OK
+SELECT * FROM t1 WHERE MATCH(b) AGAINST ('first');
+a	b	c
+1	This is a first b column	This is a first c column
+SELECT * FROM t1 WHERE MATCH(c) AGAINST ('first');
+a	b	c
+1	This is a first b column	This is a first c column
+SELECT * FROM t1 WHERE MATCH(b,c) AGAINST ('column');
+a	b	c
+1	This is a first b column	This is a first c column
+4	This is a third b column	This is a third c column
+SHOW CREATE TABLE t1;
+Table	Create Table
+t1	CREATE TABLE `t1` (
+  `a` int(11) DEFAULT NULL,
+  `b` text DEFAULT NULL,
+  `c` text DEFAULT NULL,
+  FULLTEXT KEY `b` (`b`),
+  FULLTEXT KEY `c` (`c`),
+  FULLTEXT KEY `b_2` (`b`,`c`)
+) ENGINE=InnoDB DEFAULT CHARSET=latin1
+CHECK TABLE t1;
+Table	Op	Msg_type	Msg_text
+test.t1	check	status	OK
+#sql-temporary.frm
+#sql-temporary.ibd
+FTS_INDEX_1.ibd
+FTS_INDEX_2.ibd
+FTS_INDEX_3.ibd
+FTS_INDEX_4.ibd
+FTS_INDEX_5.ibd
+FTS_INDEX_6.ibd
+FTS_INDEX_1.ibd
+FTS_INDEX_2.ibd
+FTS_INDEX_3.ibd
+FTS_INDEX_4.ibd
+FTS_INDEX_5.ibd
+FTS_INDEX_6.ibd
+FTS_INDEX_1.ibd
+FTS_INDEX_2.ibd
+FTS_INDEX_3.ibd
+FTS_INDEX_4.ibd
+FTS_INDEX_5.ibd
+FTS_INDEX_6.ibd
+FTSBEING_DELETED.ibd
+FTSBEING_DELETED_CACHE.ibd
+FTSCONFIG.ibd
+FTSDELETED.ibd
+FTSDELETED_CACHE.ibd
+t.frm
+t.ibd
+t1.frm
+t1.ibd
+SHOW CREATE TABLE t;
+Table	Create Table
+t	CREATE TABLE `t` (
+  `a` bigint(20) unsigned NOT NULL AUTO_INCREMENT,
+  `b` int(11) DEFAULT NULL,
+  `c` int(11) DEFAULT NULL,
+  `d` int(11) DEFAULT NULL,
+  UNIQUE KEY `a` (`a`)
+) ENGINE=InnoDB AUTO_INCREMENT=1000 DEFAULT CHARSET=latin1
+SELECT COUNT(*) FROM t;
+COUNT(*)
+999
+CHECK TABLE t;
+Table	Op	Msg_type	Msg_text
+test.t	check	status	OK
+SELECT * FROM t1 WHERE MATCH(b) AGAINST ('first');
+a	b	c
+1	This is a first b column	This is a first c column
+SELECT * FROM t1 WHERE MATCH(c) AGAINST ('first');
+a	b	c
+1	This is a first b column	This is a first c column
+SELECT * FROM t1 WHERE MATCH(b,c) AGAINST ('column');
+a	b	c
+1	This is a first b column	This is a first c column
+4	This is a third b column	This is a third c column
+SHOW CREATE TABLE t1;
+Table	Create Table
+t1	CREATE TABLE `t1` (
+  `a` int(11) DEFAULT NULL,
+  `b` text DEFAULT NULL,
+  `c` text DEFAULT NULL,
+  FULLTEXT KEY `b` (`b`),
+  FULLTEXT KEY `c` (`c`),
+  FULLTEXT KEY `b_2` (`b`,`c`)
+) ENGINE=InnoDB DEFAULT CHARSET=latin1
+CHECK TABLE t1;
+Table	Op	Msg_type	Msg_text
+test.t1	check	status	OK
+#sql-temporary.frm
+#sql-temporary.ibd
+FTS_INDEX_1.ibd
+FTS_INDEX_2.ibd
+FTS_INDEX_3.ibd
+FTS_INDEX_4.ibd
+FTS_INDEX_5.ibd
+FTS_INDEX_6.ibd
+FTS_INDEX_1.ibd
+FTS_INDEX_2.ibd
+FTS_INDEX_3.ibd
+FTS_INDEX_4.ibd
+FTS_INDEX_5.ibd
+FTS_INDEX_6.ibd
+FTS_INDEX_1.ibd
+FTS_INDEX_2.ibd
+FTS_INDEX_3.ibd
+FTS_INDEX_4.ibd
+FTS_INDEX_5.ibd
+FTS_INDEX_6.ibd
+FTSBEING_DELETED.ibd
+FTSBEING_DELETED_CACHE.ibd
+FTSCONFIG.ibd
+FTSDELETED.ibd
+FTSDELETED_CACHE.ibd
+t.frm
+t.ibd
+t1.frm
+t1.ibd
+DROP TABLE t1,t;
+DROP TABLE `#mysql50##sql-temporary`;
diff --git a/mysql-test/suite/innodb/t/alter_copy.test b/mysql-test/suite/innodb/t/alter_copy.test
new file mode 100644
index 00000000000..b1c71d82096
--- /dev/null
+++ b/mysql-test/suite/innodb/t/alter_copy.test
@@ -0,0 +1,95 @@
+--source include/have_innodb.inc
+--source include/have_debug.inc
+--source include/have_debug_sync.inc
+--source include/not_embedded.inc
+
+--echo #
+--echo # MDEV-11415 AVOID INTERMEDIATE COMMIT WHILE DOING
+--echo #            ALTER TABLE...ALGORITHM=COPY
+--echo #
+
+CREATE TABLE t(a SERIAL, b INT, c INT, d INT) ENGINE=InnoDB;
+CREATE TABLE t1(a INT, b TEXT, c TEXT,
+               FULLTEXT(b), FULLTEXT(c(3)), FULLTEXT(b,c)) ENGINE=InnoDB;
+
+let $c = 999;
+BEGIN;
+--disable_query_log
+while ($c) {
+INSERT INTO t() VALUES();
+dec $c;
+}
+--enable_query_log
+COMMIT;
+
+SELECT COUNT(*) FROM t;
+# try to make the to-be-created secondary index keys randomly distributed
+UPDATE t SET b=a%7, c=a%11, d=a%13;
+
+INSERT INTO t1 VALUES(1, 'This is a first b column', 'This is a first c column');
+INSERT INTO t1 VALUES(2, 'This is a second b column', 'This is a second c column');
+INSERT INTO t1(a) VALUES(3);
+INSERT INTO t1 VALUES(4, 'This is a third b column', 'This is a third c column');
+DELETE FROM t1 WHERE a = 2;
+SELECT * FROM t1 WHERE MATCH(b) AGAINST ('first');
+SELECT * FROM t1 WHERE MATCH(c) AGAINST ('first');
+SELECT * FROM t1 WHERE MATCH(b,c) AGAINST ('column');
+SHOW CREATE TABLE t1;
+ALTER TABLE t1 FORCE, ALGORITHM=COPY;
+
+# crash right after the last write_row(), before the first commit of ALTER TABLE
+--source include/expect_crash.inc
+
+SET DEBUG_DBUG='+d,crash_commit_before';
+--error 2013
+# create 32 secondary indexes
+ALTER TABLE t ADD INDEX(b,c,d,a),ADD INDEX(b,c,a,d),ADD INDEX(b,a,c,d),ADD INDEX(b,a,d,c),
+              ADD INDEX(b,d,a,c),ADD INDEX(b,d,c,a),ADD INDEX(a,b,c,d),ADD INDEX(a,b,d,c),
+              ADD INDEX(a,c,b,d),ADD INDEX(a,c,d,b),ADD INDEX(a,d,b,c),ADD INDEX(a,d,c,b),
+              ADD INDEX(c,a,b,d),ADD INDEX(c,a,d,b),ADD INDEX(c,b,a,d),ADD INDEX(c,b,d,a),
+              ADD INDEX(c,d,a,b),ADD INDEX(c,d,b,a),ADD INDEX(d,a,b,c),ADD INDEX(d,a,c,b),
+              ADD INDEX(d,b,a,c),ADD INDEX(d,b,c,a),ADD INDEX(d,c,a,b),ADD INDEX(d,c,b,a),
+              ADD INDEX(a,b,c), ADD INDEX(a,c,b), ADD INDEX(a,c,d), ADD INDEX(a,d,c),
+              ADD INDEX(a,b,d), ADD INDEX(a,d,b), ADD INDEX(b,c,d), ADD INDEX(b,d,c),
+ 	      ALGORITHM=COPY;
+
+--let $restart_parameters= --innodb-force-recovery=3
+--source include/start_mysqld.inc
+let $datadir=`select @@datadir`;
+--replace_regex /#sql-[0-9a-f_]*/#sql-temporary/ /FTS_[0-9a-f]*_[0-9a-f]*/FTS/
+--list_files $datadir/test
+SHOW CREATE TABLE t;
+SELECT COUNT(*) FROM t;
+CHECK TABLE t;
+SELECT * FROM t1 WHERE MATCH(b) AGAINST ('first');
+SELECT * FROM t1 WHERE MATCH(c) AGAINST ('first');
+SELECT * FROM t1 WHERE MATCH(b,c) AGAINST ('column');
+SHOW CREATE TABLE t1;
+CHECK TABLE t1;
+
+--let $restart_parameters= --innodb-read-only
+--source include/restart_mysqld.inc
+--replace_regex /#sql-[0-9a-f_]*/#sql-temporary/ /FTS_[0-9a-f]*_[0-9a-f]*/FTS/
+
+--list_files $datadir/test
+SHOW CREATE TABLE t;
+SELECT COUNT(*) FROM t;
+CHECK TABLE t;
+SELECT * FROM t1 WHERE MATCH(b) AGAINST ('first');
+SELECT * FROM t1 WHERE MATCH(c) AGAINST ('first');
+SELECT * FROM t1 WHERE MATCH(b,c) AGAINST ('column');
+SHOW CREATE TABLE t1;
+CHECK TABLE t1;
+
+--let $restart_parameters=
+--source include/restart_mysqld.inc
+--replace_regex /#sql-[0-9a-f_]*/#sql-temporary/ /FTS_[0-9a-f]*_[0-9a-f]*/FTS/
+--list_files $datadir/test
+DROP TABLE t1,t;
+
+# Work around missing crash recovery at the SQL layer.
+let $temp_table_name = `SELECT SUBSTRING(name,6)
+	FROM information_schema.innodb_sys_tables
+	WHERE name LIKE "test/#sql-%"`;
+--replace_regex /#sql-[0-9a-f_]*/#sql-temporary/
+eval DROP TABLE `#mysql50#$temp_table_name`;
diff --git a/sql/ha_partition.cc b/sql/ha_partition.cc
index 56e1f649286..54f3f7d37b8 100644
--- a/sql/ha_partition.cc
+++ b/sql/ha_partition.cc
@@ -1,6 +1,6 @@
 /*
   Copyright (c) 2005, 2017, Oracle and/or its affiliates.
-  Copyright (c) 2009, 2017, MariaDB
+  Copyright (c) 2009, 2018, MariaDB
 
   This program is free software; you can redistribute it and/or modify
   it under the terms of the GNU General Public License as published by
@@ -7255,6 +7255,10 @@ int ha_partition::extra(enum ha_extra_function operation)
   */
   case HA_EXTRA_MARK_AS_LOG_TABLE:
     DBUG_RETURN(ER_UNSUPORTED_LOG_ENGINE);
+  case HA_EXTRA_BEGIN_ALTER_COPY:
+  case HA_EXTRA_END_ALTER_COPY:
+  case HA_EXTRA_FAKE_START_STMT:
+    DBUG_RETURN(loop_extra(operation));
   default:
   {
     /* Temporary crash to discover what is wrong */
diff --git a/sql/log.h b/sql/log.h
index 5589cbf168b..c09ae14e02b 100644
--- a/sql/log.h
+++ b/sql/log.h
@@ -1101,6 +1101,7 @@ void make_default_log_name(char **out, const char* log_ext, bool once);
 void binlog_reset_cache(THD *thd);
 
 extern MYSQL_PLUGIN_IMPORT MYSQL_BIN_LOG mysql_bin_log;
+extern handlerton *binlog_hton;
 extern LOGGER logger;
 
 extern const char *log_bin_index;
diff --git a/sql/sql_load.cc b/sql/sql_load.cc
index fbdc00433a6..e15d4d2fec5 100644
--- a/sql/sql_load.cc
+++ b/sql/sql_load.cc
@@ -1,6 +1,6 @@
 /*
    Copyright (c) 2000, 2016, Oracle and/or its affiliates.
-   Copyright (c) 2010, 2017, MariaDB Corporation
+   Copyright (c) 2010, 2018, MariaDB Corporation.
 
    This program is free software; you can redistribute it and/or modify
    it under the terms of the GNU General Public License as published by
@@ -97,6 +97,41 @@ public:
 #define GET (stack_pos != stack ? *--stack_pos : my_b_get(&cache))
 #define PUSH(A) *(stack_pos++)=(A)
 
+#ifdef WITH_WSREP
+/** If requested by wsrep_load_data_splitting, commit and restart
+the transaction after every 10,000 inserted rows. */
+
+static bool wsrep_load_data_split(THD *thd, const TABLE *table,
+				  const COPY_INFO &info)
+{
+  extern struct handlerton* innodb_hton_ptr;
+
+  DBUG_ENTER("wsrep_load_data_split");
+
+  if (wsrep_load_data_splitting && wsrep_on(thd)
+      && info.records && !(info.records % 10000)
+      && thd->transaction.stmt.ha_list
+      && thd->transaction.stmt.ha_list->ht() == binlog_hton
+      && thd->transaction.stmt.ha_list->next()
+      && thd->transaction.stmt.ha_list->next()->ht() == innodb_hton_ptr
+      && !thd->transaction.stmt.ha_list->next()->next())
+  {
+    WSREP_DEBUG("intermediate transaction commit in LOAD DATA");
+    if (wsrep_run_wsrep_commit(thd, true) != WSREP_TRX_OK) DBUG_RETURN(true);
+    if (binlog_hton->commit(binlog_hton, thd, true)) DBUG_RETURN(true);
+    wsrep_post_commit(thd, true);
+    innodb_hton_ptr->commit(innodb_hton_ptr, thd, true);
+    table->file->extra(HA_EXTRA_FAKE_START_STMT);
+  }
+
+  DBUG_RETURN(false);
+}
+# define WSREP_LOAD_DATA_SPLIT(thd,table,info)		\
+  if (wsrep_load_data_split(thd,table,info)) DBUG_RETURN(1)
+#else /* WITH_WSREP */
+#define WSREP_LOAD_DATA_SPLIT(thd,table,info) /* empty */
+#endif /* WITH_WSREP */
+
 class READ_INFO {
   File	file;
   String data;                          /* Read buffer */
@@ -989,6 +1024,7 @@ read_fixed_length(THD *thd, COPY_INFO &info, TABLE_LIST *table_list,
       DBUG_RETURN(-1);
     }
 
+    WSREP_LOAD_DATA_SPLIT(thd, table, info);
     err= write_record(thd, table, &info);
     table->auto_increment_field_not_null= FALSE;
     if (err)
@@ -1214,6 +1250,7 @@ read_sep_field(THD *thd, COPY_INFO &info, TABLE_LIST *table_list,
       DBUG_RETURN(-1);
     }
 
+    WSREP_LOAD_DATA_SPLIT(thd, table, info);
     err= write_record(thd, table, &info);
     table->auto_increment_field_not_null= FALSE;
     if (err)
@@ -1410,6 +1447,7 @@ read_xml_field(THD *thd, COPY_INFO &info, TABLE_LIST *table_list,
       DBUG_RETURN(-1);
     }
     
+    WSREP_LOAD_DATA_SPLIT(thd, table, info);
     if (write_record(thd, table, &info))
       DBUG_RETURN(1);
     
diff --git a/sql/sql_table.cc b/sql/sql_table.cc
index 9d1cfa1f1ad..d340f046691 100644
--- a/sql/sql_table.cc
+++ b/sql/sql_table.cc
@@ -9904,6 +9904,8 @@ copy_data_between_tables(THD *thd, TABLE *from, TABLE *to,
 
   thd->progress.max_counter= from->file->records();
   time_to_report_progress= MY_HOW_OFTEN_TO_WRITE/10;
+  if (!ignore) /* for now, InnoDB needs the undo log for ALTER IGNORE */
+    to->file->extra(HA_EXTRA_BEGIN_ALTER_COPY);
 
   while (!(error=info.read_record(&info)))
   {
@@ -10028,6 +10030,8 @@ copy_data_between_tables(THD *thd, TABLE *from, TABLE *to,
       to->file->print_error(my_errno,MYF(0));
     error= 1;
   }
+  if (!ignore)
+    to->file->extra(HA_EXTRA_END_ALTER_COPY);
   to->file->extra(HA_EXTRA_NO_IGNORE_DUP_KEY);
 
   if (mysql_trans_commit_alter_copy_data(thd))
diff --git a/storage/innobase/fts/fts0fts.cc b/storage/innobase/fts/fts0fts.cc
index 056eee5187a..8c8e8020c9d 100644
--- a/storage/innobase/fts/fts0fts.cc
+++ b/storage/innobase/fts/fts0fts.cc
@@ -2303,7 +2303,6 @@ fts_savepoint_create(
 /******************************************************************//**
 Create an FTS trx.
 @return FTS trx */
-static
 fts_trx_t*
 fts_trx_create(
 /*===========*/
@@ -3332,6 +3331,144 @@ fts_fetch_doc_from_rec(
 	}
 }
 
+/** Fetch the data from tuple and tokenize the document.
+@param[in]     get_doc FTS index's get_doc struct
+@param[in]     tuple   tuple should be arranged in table schema order
+@param[out]    doc     fts doc to hold parsed documents. */
+static
+void
+fts_fetch_doc_from_tuple(
+       fts_get_doc_t*  get_doc,
+       const dtuple_t* tuple,
+       fts_doc_t*      doc)
+{
+       dict_index_t*           index;
+       st_mysql_ftparser*      parser;
+       ulint                   doc_len = 0;
+       ulint                   processed_doc = 0;
+       ulint                   num_field;
+
+       if (get_doc == NULL) {
+               return;
+       }
+
+       index = get_doc->index_cache->index;
+       parser = get_doc->index_cache->index->parser;
+       num_field = dict_index_get_n_fields(index);
+
+       for (ulint i = 0; i < num_field; i++) {
+               const dict_field_t*     ifield;
+               const dict_col_t*       col;
+               ulint                   pos;
+               dfield_t*               field;
+
+               ifield = dict_index_get_nth_field(index, i);
+               col = dict_field_get_col(ifield);
+               pos = dict_col_get_no(col);
+               field = dtuple_get_nth_field(tuple, pos);
+
+               if (!get_doc->index_cache->charset) {
+                       get_doc->index_cache->charset = fts_get_charset(
+                               ifield->col->prtype);
+               }
+
+               ut_ad(!dfield_is_ext(field));
+
+               doc->text.f_str = (byte*) dfield_get_data(field);
+               doc->text.f_len = dfield_get_len(field);
+               doc->found = TRUE;
+               doc->charset = get_doc->index_cache->charset;
+
+               /* field data is NULL. */
+               if (doc->text.f_len == UNIV_SQL_NULL || doc->text.f_len == 0) {
+                       continue;
+               }
+
+               if (processed_doc == 0) {
+                       fts_tokenize_document(doc, NULL, parser);
+               } else {
+                       fts_tokenize_document_next(doc, doc_len, NULL, parser);
+               }
+
+               processed_doc++;
+               doc_len += doc->text.f_len + 1;
+       }
+}
+
+/** Fetch the document from tuple, tokenize the text data and
+insert the text data into fts auxiliary table and
+its cache. Moreover this tuple fields doesn't contain any information
+about externally stored field. This tuple contains data directly
+converted from mysql.
+@param[in]     ftt     FTS transaction table
+@param[in]     doc_id  doc id
+@param[in]     tuple   tuple from where data can be retrieved
+                       and tuple should be arranged in table
+                       schema order. */
+void
+fts_add_doc_from_tuple(
+       fts_trx_table_t*ftt,
+       doc_id_t        doc_id,
+       const dtuple_t* tuple)
+{
+       mtr_t           mtr;
+       fts_cache_t*    cache = ftt->table->fts->cache;
+
+       ut_ad(cache->get_docs);
+
+       if (!(ftt->table->fts->fts_status & ADDED_TABLE_SYNCED)) {
+               fts_init_index(ftt->table, FALSE);
+       }
+
+       mtr_start(&mtr);
+
+       ulint   num_idx = ib_vector_size(cache->get_docs);
+
+       for (ulint i = 0; i < num_idx; ++i) {
+               fts_doc_t       doc;
+               dict_table_t*   table;
+               fts_get_doc_t*  get_doc;
+
+               get_doc = static_cast<fts_get_doc_t*>(
+                       ib_vector_get(cache->get_docs, i));
+               table = get_doc->index_cache->index->table;
+
+               fts_doc_init(&doc);
+               fts_fetch_doc_from_tuple(
+                       get_doc, tuple, &doc);
+
+               if (doc.found) {
+                       mtr_commit(&mtr);
+                       rw_lock_x_lock(&table->fts->cache->lock);
+
+                       if (table->fts->cache->stopword_info.status
+                           & STOPWORD_NOT_INIT) {
+                               fts_load_stopword(table, NULL, NULL,
+                                                 NULL, TRUE, TRUE);
+                       }
+
+                       fts_cache_add_doc(
+                               table->fts->cache,
+                               get_doc->index_cache,
+                               doc_id, doc.tokens);
+
+                       rw_lock_x_unlock(&table->fts->cache->lock);
+
+                       if (cache->total_size > fts_max_cache_size / 5
+                           || fts_need_sync) {
+                               fts_sync(cache->sync, true, false, false);
+                       }
+
+                       mtr_start(&mtr);
+
+               }
+
+               fts_doc_free(&doc);
+       }
+
+       mtr_commit(&mtr);
+}
+
 /*********************************************************************//**
 This function fetches the document inserted during the committing
 transaction, and tokenize the inserted text data and insert into
diff --git a/storage/innobase/handler/ha_innodb.cc b/storage/innobase/handler/ha_innodb.cc
index a6638e7dbf0..5dee02bed67 100644
--- a/storage/innobase/handler/ha_innodb.cc
+++ b/storage/innobase/handler/ha_innodb.cc
@@ -191,7 +191,7 @@ static mysql_mutex_t pending_checkpoint_mutex;
 
 #define EQ_CURRENT_THD(thd) ((thd) == current_thd)
 
-static struct handlerton* innodb_hton_ptr;
+struct handlerton* innodb_hton_ptr;
 
 static const long AUTOINC_OLD_STYLE_LOCKING = 0;
 static const long AUTOINC_NEW_STYLE_LOCKING = 1;
@@ -3071,7 +3071,6 @@ ha_innobase::ha_innobase(
 			  |  (srv_force_primary_key ? HA_REQUIRE_PRIMARY_KEY : 0)
 		  ),
 	m_start_of_scan(),
-	m_num_write_row(),
         m_mysql_has_locked()
 {}
 
@@ -8364,7 +8363,6 @@ ha_innobase::write_row(
 #ifdef WITH_WSREP
 	ibool		auto_inc_inserted= FALSE; /* if NULL was inserted */
 #endif
-	ulint		sql_command;
 	int		error_result = 0;
 	bool		auto_inc_used = false;
 
@@ -8381,7 +8379,7 @@ ha_innobase::write_row(
 			DB_FORCED_ABORT, 0, m_user_thd));
 	}
 
-	/* Step-1: Validation checks before we commence write_row operation. */
+	/* Validation checks before we commence write_row operation. */
 	if (high_level_read_only) {
 		ib_senderrf(ha_thd(), IB_LOG_LEVEL_WARN, ER_READ_ONLY_MODE);
 		DBUG_RETURN(HA_ERR_TABLE_READONLY);
@@ -8402,131 +8400,7 @@ ha_innobase::write_row(
 		++trx->will_lock;
 	}
 
-	/* Step-2: Intermediate commit if original operation involves ALTER
-	table with algorithm = copy. Intermediate commit ease pressure on
-	recovery if server crashes while ALTER is active. */
-	sql_command = thd_sql_command(m_user_thd);
-
-	if ((sql_command == SQLCOM_ALTER_TABLE
-	     || sql_command == SQLCOM_OPTIMIZE
-	     || sql_command == SQLCOM_CREATE_INDEX
-#ifdef WITH_WSREP
-	     || (sql_command == SQLCOM_LOAD
-		 && wsrep_load_data_splitting && wsrep_on(m_user_thd)
-		 && !thd_test_options(
-			 m_user_thd, OPTION_NOT_AUTOCOMMIT | OPTION_BEGIN))
-#endif /* WITH_WSREP */
-	     || sql_command == SQLCOM_DROP_INDEX)
-	    && m_num_write_row >= 10000) {
-#ifdef WITH_WSREP
-		if (sql_command == SQLCOM_LOAD && wsrep_on(m_user_thd)) {
-			WSREP_DEBUG("forced trx split for LOAD: %s",
-				    wsrep_thd_query(m_user_thd));
-		}
-#endif /* WITH_WSREP */
-		/* ALTER TABLE is COMMITted at every 10000 copied rows.
-		The IX table lock for the original table has to be re-issued.
-		As this method will be called on a temporary table where the
-		contents of the original table is being copied to, it is
-		a bit tricky to determine the source table.  The cursor
-		position in the source table need not be adjusted after the
-		intermediate COMMIT, since writes by other transactions are
-		being blocked by a MySQL table lock TL_WRITE_ALLOW_READ. */
-
-		dict_table_t*	src_table;
-		enum lock_mode	mode;
-
-		m_num_write_row = 0;
-
-		/* Commit the transaction.  This will release the table
-		locks, so they have to be acquired again. */
-
-		/* Altering an InnoDB table */
-		/* Get the source table. */
-		src_table = lock_get_src_table(
-				m_prebuilt->trx, m_prebuilt->table, &mode);
-		if (!src_table) {
-no_commit:
-			/* Unknown situation: do not commit */
-			;
-		} else if (src_table == m_prebuilt->table) {
-#ifdef WITH_WSREP
-			if (wsrep_on(m_user_thd)                            &&
-			    wsrep_load_data_splitting                       &&
-			    sql_command == SQLCOM_LOAD                      &&
-			    !thd_test_options(m_user_thd,
-			                      OPTION_NOT_AUTOCOMMIT | OPTION_BEGIN))
-			{
-				switch (wsrep_run_wsrep_commit(m_user_thd, 1)) {
-				case WSREP_TRX_OK:
-				  break;
-				case WSREP_TRX_SIZE_EXCEEDED:
-				case WSREP_TRX_CERT_FAIL:
-				case WSREP_TRX_ERROR:
-				  DBUG_RETURN(1);
-				}
-
-				if (binlog_hton->commit(binlog_hton, m_user_thd, 1)) {
-					DBUG_RETURN(1);
-				}
-				wsrep_post_commit(m_user_thd, TRUE);
-			}
-#endif /* WITH_WSREP */
-			/* Source table is not in InnoDB format:
-			no need to re-acquire locks on it. */
-
-			/* Altering to InnoDB format */
-			innobase_commit(ht, m_user_thd, 1);
-			/* Note that this transaction is still active. */
-			trx_register_for_2pc(m_prebuilt->trx);
-			/* We will need an IX lock on the destination table. */
-			m_prebuilt->sql_stat_start = TRUE;
-		} else {
-#ifdef WITH_WSREP
-			if (wsrep_on(m_user_thd)                            &&
-			    wsrep_load_data_splitting                       &&
-			    sql_command == SQLCOM_LOAD                      &&
-			    !thd_test_options(m_user_thd,
-			                      OPTION_NOT_AUTOCOMMIT | OPTION_BEGIN)) {
-				switch (wsrep_run_wsrep_commit(m_user_thd, 1)) {
-				case WSREP_TRX_OK:
-					break;
-				case WSREP_TRX_SIZE_EXCEEDED:
-				case WSREP_TRX_CERT_FAIL:
-				case WSREP_TRX_ERROR:
-					DBUG_RETURN(1);
-				}
-
-				if (binlog_hton->commit(binlog_hton, m_user_thd, 1)) {
-					DBUG_RETURN(1);
-				}
-
-				wsrep_post_commit(m_user_thd, TRUE);
-			}
-#endif /* WITH_WSREP */
-			/* Ensure that there are no other table locks than
-			LOCK_IX and LOCK_AUTO_INC on the destination table. */
-
-			if (!lock_is_table_exclusive(m_prebuilt->table,
-							m_prebuilt->trx)) {
-				goto no_commit;
-			}
-
-			/* Commit the transaction.  This will release the table
-			locks, so they have to be acquired again. */
-			innobase_commit(ht, m_user_thd, 1);
-			/* Note that this transaction is still active. */
-			trx_register_for_2pc(m_prebuilt->trx);
-			/* Re-acquire the table lock on the source table. */
-			row_lock_table_for_mysql(m_prebuilt, src_table, mode);
-			/* We will need an IX lock on the destination table. */
-			m_prebuilt->sql_stat_start = TRUE;
-		}
-	}
-
-	m_num_write_row++;
-
-	/* Step-3: Handling of Auto-Increment Columns. */
+	/* Handling of Auto-Increment Columns. */
 	if (table->next_number_field && record == table->record[0]) {
 
 		/* Reset the error code before calling
@@ -8559,7 +8433,7 @@ no_commit:
 		auto_inc_used = true;
 	}
 
-	/* Step-4: Prepare INSERT graph that will be executed for actual INSERT
+	/* Prepare INSERT graph that will be executed for actual INSERT
 	(This is a one time operation) */
 	if (m_prebuilt->mysql_template == NULL
 	    || m_prebuilt->template_type != ROW_MYSQL_WHOLE_ROW) {
@@ -8572,12 +8446,12 @@ no_commit:
 
 	innobase_srv_conc_enter_innodb(m_prebuilt);
 
-	/* Step-5: Execute insert graph that will result in actual insert. */
+	/* Execute insert graph that will result in actual insert. */
 	error = row_insert_for_mysql((byte*) record, m_prebuilt);
 
 	DEBUG_SYNC(m_user_thd, "ib_after_row_insert");
 
-	/* Step-6: Handling of errors related to auto-increment. */
+	/* Handling of errors related to auto-increment. */
 	if (auto_inc_used) {
 		ulonglong	auto_inc;
 		ulonglong	col_max_value;
@@ -8605,13 +8479,11 @@ no_commit:
 			must update the autoinc counter if we are performing
 			those statements. */
 
-			switch (sql_command) {
+			switch (thd_sql_command(m_user_thd)) {
 			case SQLCOM_LOAD:
-				if (trx->duplicates) {
-
-					goto set_max_autoinc;
+				if (!trx->duplicates) {
+					break;
 				}
-				break;
 
 			case SQLCOM_REPLACE:
 			case SQLCOM_INSERT_SELECT:
@@ -8700,7 +8572,7 @@ set_max_autoinc:
 	innobase_srv_conc_exit_innodb(m_prebuilt);
 
 report_error:
-	/* Step-7: Cleanup and exit. */
+	/* Cleanup and exit. */
 	if (error == DB_TABLESPACE_DELETED) {
 		ib_senderrf(
 			trx->mysql_thd, IB_LOG_LEVEL_ERROR,
@@ -15965,6 +15837,16 @@ ha_innobase::extra(
 	case HA_EXTRA_WRITE_CANNOT_REPLACE:
 		thd_to_trx(ha_thd())->duplicates &= ~TRX_DUP_REPLACE;
 		break;
+	case HA_EXTRA_BEGIN_ALTER_COPY:
+		m_prebuilt->table->skip_alter_undo = 1;
+		break;
+	case HA_EXTRA_END_ALTER_COPY:
+		m_prebuilt->table->skip_alter_undo = 0;
+		break;
+	case HA_EXTRA_FAKE_START_STMT:
+		trx_register_for_2pc(m_prebuilt->trx);
+		m_prebuilt->sql_stat_start = true;
+		break;
 	default:/* Do nothing */
 		;
 	}
@@ -16067,7 +15949,7 @@ ha_innobase::start_stmt(
 			init_table_handle_for_HANDLER();
 			m_prebuilt->select_lock_type = LOCK_X;
 			m_prebuilt->stored_select_lock_type = LOCK_X;
-			error = row_lock_table_for_mysql(m_prebuilt, NULL, 1);
+			error = row_lock_table(m_prebuilt);
 
 			if (error != DB_SUCCESS) {
 				int	st = convert_error_code_to_mysql(
@@ -16331,8 +16213,7 @@ ha_innobase::external_lock(
 			    && thd_test_options(thd, OPTION_NOT_AUTOCOMMIT)
 			    && thd_in_lock_tables(thd)) {
 
-				dberr_t	error = row_lock_table_for_mysql(
-					m_prebuilt, NULL, 0);
+				dberr_t	error = row_lock_table(m_prebuilt);
 
 				if (error != DB_SUCCESS) {
 
diff --git a/storage/innobase/handler/ha_innodb.h b/storage/innobase/handler/ha_innodb.h
index 6f87e0cf2f0..4a35f4eae78 100644
--- a/storage/innobase/handler/ha_innodb.h
+++ b/storage/innobase/handler/ha_innodb.h
@@ -515,9 +515,6 @@ protected:
 	ROW_SEL_EXACT_PREFIX, or undefined */
 	uint			m_last_match_mode;
 
-	/** number of write_row() calls */
-	uint			m_num_write_row;
-
         /** If mysql has locked with external_lock() */
         bool                    m_mysql_has_locked;
 };
diff --git a/storage/innobase/include/dict0mem.h b/storage/innobase/include/dict0mem.h
index 97b34fd53e5..86d056676fa 100644
--- a/storage/innobase/include/dict0mem.h
+++ b/storage/innobase/include/dict0mem.h
@@ -1375,6 +1375,13 @@ struct dict_table_t {
 	Use DICT_TF2_FLAG_IS_SET() to parse this flag. */
 	unsigned				flags2:DICT_TF2_BITS;
 
+	/** TRUE if the table is an intermediate table during copy alter
+	operation or a partition/subpartition which is required for copying
+	data and skip the undo log for insertion of row in the table.
+	This variable will be set and unset during extra(), or during the
+	process of altering partitions */
+	unsigned                                skip_alter_undo:1;
+
 	/*!< whether this is in a single-table tablespace and the .ibd
 	file is missing or page decryption failed and page is corrupted */
 	unsigned				file_unreadable:1;
diff --git a/storage/innobase/include/fts0fts.h b/storage/innobase/include/fts0fts.h
index 813e34b43d3..362bdcb7fe6 100644
--- a/storage/innobase/include/fts0fts.h
+++ b/storage/innobase/include/fts0fts.h
@@ -1015,5 +1015,27 @@ fts_check_corrupt(
 	dict_table_t*	base_table,
 	trx_t*		trx);
 
+/** Fetch the document from tuple, tokenize the text data and
+insert the text data into fts auxiliary table and
+its cache. Moreover this tuple fields doesn't contain any information
+about externally stored field. This tuple contains data directly
+converted from mysql.
+@param[in]     ftt     FTS transaction table
+@param[in]     doc_id  doc id
+@param[in]     tuple   tuple from where data can be retrieved
+                       and tuple should be arranged in table
+                       schema order. */
+void
+fts_add_doc_from_tuple(
+	fts_trx_table_t*ftt,
+	doc_id_t        doc_id,
+	const dtuple_t* tuple);
+
+/** Create an FTS trx.
+@param[in,out] trx     InnoDB Transaction
+@return FTS transaction. */
+fts_trx_t*
+fts_trx_create(
+	trx_t*  trx);
 
 #endif /*!< fts0fts.h */
diff --git a/storage/innobase/include/lock0lock.h b/storage/innobase/include/lock0lock.h
index 60b07f2fe72..b1f27e7f456 100644
--- a/storage/innobase/include/lock0lock.h
+++ b/storage/innobase/include/lock0lock.h
@@ -566,30 +566,6 @@ lock_rec_find_set_bit(
 				bit set */
 
 /*********************************************************************//**
-Gets the source table of an ALTER TABLE transaction.  The table must be
-covered by an IX or IS table lock.
-@return the source table of transaction, if it is covered by an IX or
-IS table lock; dest if there is no source table, and NULL if the
-transaction is locking more than two tables or an inconsistency is
-found */
-dict_table_t*
-lock_get_src_table(
-/*===============*/
-	trx_t*		trx,	/*!< in: transaction */
-	dict_table_t*	dest,	/*!< in: destination of ALTER TABLE */
-	lock_mode*	mode);	/*!< out: lock mode of the source table */
-/*********************************************************************//**
-Determine if the given table is exclusively "owned" by the given
-transaction, i.e., transaction holds LOCK_IX and possibly LOCK_AUTO_INC
-on the table.
-@return TRUE if table is only locked by trx, with LOCK_IX, and
-possibly LOCK_AUTO_INC */
-ibool
-lock_is_table_exclusive(
-/*====================*/
-	const dict_table_t*	table,	/*!< in: table */
-	const trx_t*		trx);	/*!< in: transaction */
-/*********************************************************************//**
 Checks if a lock request lock1 has to wait for request lock2.
 @return TRUE if lock1 has to wait for lock2 to be removed */
 ibool
diff --git a/storage/innobase/include/row0mysql.h b/storage/innobase/include/row0mysql.h
index a7a55d202e8..5cb0249517c 100644
--- a/storage/innobase/include/row0mysql.h
+++ b/storage/innobase/include/row0mysql.h
@@ -1,7 +1,7 @@
 /*****************************************************************************
 
 Copyright (c) 2000, 2017, Oracle and/or its affiliates. All Rights Reserved.
-Copyright (c) 2017, MariaDB Corporation.
+Copyright (c) 2017, 2018, MariaDB Corporation.
 
 This program is free software; you can redistribute it and/or modify it under
 the terms of the GNU General Public License as published by the Free Software
@@ -219,21 +219,11 @@ row_lock_table_autoinc_for_mysql(
 					table handle */
 	MY_ATTRIBUTE((nonnull, warn_unused_result));
 
-/*********************************************************************//**
-Sets a table lock on the table mentioned in prebuilt.
+/** Lock a table.
+@param[in,out]	prebuilt	table handle
 @return error code or DB_SUCCESS */
 dberr_t
-row_lock_table_for_mysql(
-/*=====================*/
-	row_prebuilt_t*	prebuilt,	/*!< in: prebuilt struct in the MySQL
-					table handle */
-	dict_table_t*	table,		/*!< in: table to lock, or NULL
-					if prebuilt->table should be
-					locked as
-					prebuilt->select_lock_type */
-	ulint		mode)		/*!< in: lock mode of table
-					(ignored if table==NULL) */
-	MY_ATTRIBUTE((nonnull(1)));
+row_lock_table(row_prebuilt_t* prebuilt);
 
 /** Does an insert for MySQL.
 @param[in]	mysql_rec	row in the MySQL format
diff --git a/storage/innobase/lock/lock0lock.cc b/storage/innobase/lock/lock0lock.cc
index 41cf2f15ba4..f25ac596d5f 100644
--- a/storage/innobase/lock/lock0lock.cc
+++ b/storage/innobase/lock/lock0lock.cc
@@ -620,145 +620,6 @@ lock_get_size(void)
 }
 
 /*********************************************************************//**
-Gets the source table of an ALTER TABLE transaction.  The table must be
-covered by an IX or IS table lock.
-@return the source table of transaction, if it is covered by an IX or
-IS table lock; dest if there is no source table, and NULL if the
-transaction is locking more than two tables or an inconsistency is
-found */
-dict_table_t*
-lock_get_src_table(
-/*===============*/
-	trx_t*		trx,	/*!< in: transaction */
-	dict_table_t*	dest,	/*!< in: destination of ALTER TABLE */
-	lock_mode*	mode)	/*!< out: lock mode of the source table */
-{
-	dict_table_t*	src;
-	lock_t*		lock;
-
-	ut_ad(!lock_mutex_own());
-
-	src = NULL;
-	*mode = LOCK_NONE;
-
-	/* The trx mutex protects the trx_locks for our purposes.
-	Other transactions could want to convert one of our implicit
-	record locks to an explicit one. For that, they would need our
-	trx mutex. Waiting locks can be removed while only holding
-	lock_sys->mutex, but this is a running transaction and cannot
-	thus be holding any waiting locks. */
-	trx_mutex_enter(trx);
-
-	for (lock = UT_LIST_GET_FIRST(trx->lock.trx_locks);
-	     lock != NULL;
-	     lock = UT_LIST_GET_NEXT(trx_locks, lock)) {
-		lock_table_t*	tab_lock;
-		lock_mode	lock_mode;
-		if (!(lock_get_type_low(lock) & LOCK_TABLE)) {
-			/* We are only interested in table locks. */
-			continue;
-		}
-		tab_lock = &lock->un_member.tab_lock;
-		if (dest == tab_lock->table) {
-			/* We are not interested in the destination table. */
-			continue;
-		} else if (!src) {
-			/* This presumably is the source table. */
-			src = tab_lock->table;
-			if (UT_LIST_GET_LEN(src->locks) != 1
-			    || UT_LIST_GET_FIRST(src->locks) != lock) {
-				/* We only support the case when
-				there is only one lock on this table. */
-				src = NULL;
-				goto func_exit;
-			}
-		} else if (src != tab_lock->table) {
-			/* The transaction is locking more than
-			two tables (src and dest): abort */
-			src = NULL;
-			goto func_exit;
-		}
-
-		/* Check that the source table is locked by
-		LOCK_IX or LOCK_IS. */
-		lock_mode = lock_get_mode(lock);
-		if (lock_mode == LOCK_IX || lock_mode == LOCK_IS) {
-			if (*mode != LOCK_NONE && *mode != lock_mode) {
-				/* There are multiple locks on src. */
-				src = NULL;
-				goto func_exit;
-			}
-			*mode = lock_mode;
-		}
-	}
-
-	if (!src) {
-		/* No source table lock found: flag the situation to caller */
-		src = dest;
-	}
-
-func_exit:
-	trx_mutex_exit(trx);
-	return(src);
-}
-
-/*********************************************************************//**
-Determine if the given table is exclusively "owned" by the given
-transaction, i.e., transaction holds LOCK_IX and possibly LOCK_AUTO_INC
-on the table.
-@return TRUE if table is only locked by trx, with LOCK_IX, and
-possibly LOCK_AUTO_INC */
-ibool
-lock_is_table_exclusive(
-/*====================*/
-	const dict_table_t*	table,	/*!< in: table */
-	const trx_t*		trx)	/*!< in: transaction */
-{
-	const lock_t*	lock;
-	ibool		ok	= FALSE;
-
-	ut_ad(table);
-	ut_ad(trx);
-
-	lock_mutex_enter();
-
-	for (lock = UT_LIST_GET_FIRST(table->locks);
-	     lock != NULL;
-	     lock = UT_LIST_GET_NEXT(locks, &lock->un_member.tab_lock)) {
-		if (lock->trx != trx) {
-			/* A lock on the table is held
-			by some other transaction. */
-			goto not_ok;
-		}
-
-		if (!(lock_get_type_low(lock) & LOCK_TABLE)) {
-			/* We are interested in table locks only. */
-			continue;
-		}
-
-		switch (lock_get_mode(lock)) {
-		case LOCK_IX:
-			ok = TRUE;
-			break;
-		case LOCK_AUTO_INC:
-			/* It is allowed for trx to hold an
-			auto_increment lock. */
-			break;
-		default:
-not_ok:
-			/* Other table locks than LOCK_IX are not allowed. */
-			ok = FALSE;
-			goto func_exit;
-		}
-	}
-
-func_exit:
-	lock_mutex_exit();
-
-	return(ok);
-}
-
-/*********************************************************************//**
 Sets the wait flag of a lock and the back pointer in trx to lock. */
 UNIV_INLINE
 void
diff --git a/storage/innobase/row/row0ins.cc b/storage/innobase/row/row0ins.cc
index 898415658d1..fc6f932db81 100644
--- a/storage/innobase/row/row0ins.cc
+++ b/storage/innobase/row/row0ins.cc
@@ -3191,12 +3191,21 @@ row_ins_clust_index_entry(
 
 	n_uniq = dict_index_is_unique(index) ? index->n_uniq : 0;
 
-	/* Try first optimistic descent to the B-tree */
-	log_free_check();
-	const ulint	flags = dict_table_is_temporary(index->table)
+	ulint	flags = dict_table_is_temporary(index->table)
 		? BTR_NO_LOCKING_FLAG
 		: 0;
 
+	/* For intermediate table during copy alter table,
+	   skip the undo log and record lock checking for
+	   insertion operation.
+	*/
+	if (index->table->skip_alter_undo) {
+		flags |= BTR_NO_UNDO_LOG_FLAG | BTR_NO_LOCKING_FLAG;
+	}
+
+	/* Try first optimistic descent to the B-tree */
+	log_free_check();
+
 	err = row_ins_clust_index_entry_low(
 		flags, BTR_MODIFY_LEAF, index, n_uniq, entry,
 		n_ext, thr, dup_chk_only);
@@ -3239,6 +3248,7 @@ row_ins_sec_index_entry(
 	dberr_t		err;
 	mem_heap_t*	offsets_heap;
 	mem_heap_t*	heap;
+	trx_id_t	trx_id  = 0;
 
 	DBUG_EXECUTE_IF("row_ins_sec_index_entry_timeout", {
 			DBUG_SET("-d,row_ins_sec_index_entry_timeout");
@@ -3261,13 +3271,22 @@ row_ins_sec_index_entry(
 	/* Try first optimistic descent to the B-tree */
 
 	log_free_check();
-	const ulint flags = dict_table_is_temporary(index->table)
+	ulint flags = dict_table_is_temporary(index->table)
 		? BTR_NO_LOCKING_FLAG
 		: 0;
 
+	/* For intermediate table during copy alter table,
+	   skip the undo log and record lock checking for
+	   insertion operation.
+	*/
+	if (index->table->skip_alter_undo) {
+		trx_id = thr_get_trx(thr)->id;
+		flags |= BTR_NO_UNDO_LOG_FLAG | BTR_NO_LOCKING_FLAG;
+	}
+
 	err = row_ins_sec_index_entry_low(
 		flags, BTR_MODIFY_LEAF, index, offsets_heap, heap, entry,
-		0, thr, dup_chk_only);
+		trx_id, thr, dup_chk_only);
 	if (err == DB_FAIL) {
 		mem_heap_empty(heap);
 
diff --git a/storage/innobase/row/row0mysql.cc b/storage/innobase/row/row0mysql.cc
index 50be659ca50..7c1c07e4a74 100644
--- a/storage/innobase/row/row0mysql.cc
+++ b/storage/innobase/row/row0mysql.cc
@@ -1259,20 +1259,11 @@ run_again:
 	return(err);
 }
 
-/*********************************************************************//**
-Sets a table lock on the table mentioned in prebuilt.
+/** Lock a table.
+@param[in,out]	prebuilt	table handle
 @return error code or DB_SUCCESS */
 dberr_t
-row_lock_table_for_mysql(
-/*=====================*/
-	row_prebuilt_t*	prebuilt,	/*!< in: prebuilt struct in the MySQL
-					table handle */
-	dict_table_t*	table,		/*!< in: table to lock, or NULL
-					if prebuilt->table should be
-					locked as
-					prebuilt->select_lock_type */
-	ulint		mode)		/*!< in: lock mode of table
-					(ignored if table==NULL) */
+row_lock_table(row_prebuilt_t* prebuilt)
 {
 	trx_t*		trx		= prebuilt->trx;
 	que_thr_t*	thr;
@@ -1302,17 +1293,10 @@ run_again:
 
 	trx_start_if_not_started_xa(trx, false);
 
-	if (table) {
-		err = lock_table(
-			0, table,
-			static_cast<enum lock_mode>(mode), thr);
-	} else {
-		err = lock_table(
-			0, prebuilt->table,
-			static_cast<enum lock_mode>(
-				prebuilt->select_lock_type),
-			thr);
-	}
+	err = lock_table(0, prebuilt->table,
+			 static_cast<enum lock_mode>(
+				 prebuilt->select_lock_type),
+			 thr);
 
 	trx->error_state = err;
 
@@ -1549,9 +1533,21 @@ error_exit:
 			}
 		}
 
-		/* Pass NULL for the columns affected, since an INSERT affects
-		all FTS indexes. */
-		fts_trx_add_op(trx, table, doc_id, FTS_INSERT, NULL);
+		if (table->skip_alter_undo) {
+			if (trx->fts_trx == NULL) {
+				trx->fts_trx = fts_trx_create(trx);
+			}
+
+			fts_trx_table_t ftt;
+			ftt.table = table;
+			ftt.fts_trx = trx->fts_trx;
+
+			fts_add_doc_from_tuple(&ftt, doc_id, node->row);
+		} else {
+			/* Pass NULL for the columns affected, since an INSERT affects
+			all FTS indexes. */
+			fts_trx_add_op(trx, table, doc_id, FTS_INSERT, NULL);
+		}
 	}
 
 	que_thr_stop_for_mysql_no_error(thr, trx);
diff --git a/storage/innobase/row/row0purge.cc b/storage/innobase/row/row0purge.cc
index c14b455b73b..697d59598f2 100644
--- a/storage/innobase/row/row0purge.cc
+++ b/storage/innobase/row/row0purge.cc
@@ -679,6 +679,7 @@ row_purge_upd_exist_or_extern_func(
 	mem_heap_t*	heap;
 
 	ut_ad(rw_lock_own(dict_operation_lock, RW_LOCK_S));
+	ut_ad(!node->table->skip_alter_undo);
 
 	if (node->rec_type == TRX_UNDO_UPD_DEL_REC
 	    || (node->cmpl_info & UPD_NODE_NO_ORD_CHANGE)) {
@@ -945,6 +946,7 @@ row_purge_record_func(
 	bool		purged		= true;
 
 	ut_ad(!node->found_clust);
+	ut_ad(!node->table->skip_alter_undo);
 
 	clust_index = dict_table_get_first_index(node->table);
 
diff --git a/storage/innobase/row/row0uins.cc b/storage/innobase/row/row0uins.cc
index 09c22cdcfd2..f4cd07dc53a 100644
--- a/storage/innobase/row/row0uins.cc
+++ b/storage/innobase/row/row0uins.cc
@@ -366,6 +366,7 @@ close_table:
 		dict_table_close(node->table, dict_locked, FALSE);
 		node->table = NULL;
 	} else {
+		ut_ad(!node->table->skip_alter_undo);
 		clust_index = dict_table_get_first_index(node->table);
 
 		if (clust_index != NULL) {
diff --git a/storage/innobase/row/row0umod.cc b/storage/innobase/row/row0umod.cc
index 087d922b219..3dcffd1125a 100644
--- a/storage/innobase/row/row0umod.cc
+++ b/storage/innobase/row/row0umod.cc
@@ -1145,6 +1145,8 @@ row_undo_mod_parse_undo_rec(
 		return;
 	}
 
+	ut_ad(!node->table->skip_alter_undo);
+
 	if (UNIV_UNLIKELY(!fil_table_accessible(node->table))) {
 close_table:
 		/* Normally, tables should not disappear or become
diff --git a/storage/innobase/row/row0undo.cc b/storage/innobase/row/row0undo.cc
index 252dfb4a6a6..9f5d05f8f9f 100644
--- a/storage/innobase/row/row0undo.cc
+++ b/storage/innobase/row/row0undo.cc
@@ -172,6 +172,8 @@ row_undo_search_clust_to_pcur(
 	ulint*		offsets		= offsets_;
 	rec_offs_init(offsets_);
 
+	ut_ad(!node->table->skip_alter_undo);
+
 	mtr_start(&mtr);
 
 	clust_index = dict_table_get_first_index(node->table);
diff --git a/storage/innobase/row/row0upd.cc b/storage/innobase/row/row0upd.cc
index 49f4254c207..811b37593a2 100644
--- a/storage/innobase/row/row0upd.cc
+++ b/storage/innobase/row/row0upd.cc
@@ -581,6 +581,7 @@ row_upd_changes_field_size_or_external(
 	ulint			i;
 
 	ut_ad(rec_offs_validate(NULL, index, offsets));
+	ut_ad(!index->table->skip_alter_undo);
 	n_fields = upd_get_n_fields(update);
 
 	for (i = 0; i < n_fields; i++) {
@@ -695,6 +696,7 @@ row_upd_rec_in_place(
 	ulint			i;
 
 	ut_ad(rec_offs_validate(rec, index, offsets));
+	ut_ad(!index->table->skip_alter_undo);
 
 	if (rec_offs_comp(offsets)) {
 		rec_set_info_bits_new(rec, update->info_bits);
@@ -978,6 +980,7 @@ row_upd_build_sec_rec_difference_binary(
 	ut_ad(rec_offs_validate(rec, index, offsets));
 	ut_ad(rec_offs_n_fields(offsets) == dtuple_get_n_fields(entry));
 	ut_ad(!rec_offs_any_extern(offsets));
+	ut_ad(!index->table->skip_alter_undo);
 
 	update = upd_create(dtuple_get_n_fields(entry), heap);
 
@@ -1059,6 +1062,7 @@ row_upd_build_difference_binary(
 
 	/* This function is used only for a clustered index */
 	ut_a(dict_index_is_clust(index));
+	ut_ad(!index->table->skip_alter_undo);
 
 	update = upd_create(n_fld + n_v_fld, heap);
 
@@ -1330,6 +1334,8 @@ row_upd_index_replace_new_col_vals_index_pos(
 	mem_heap_t*	heap)	/*!< in: memory heap for allocating and
 				copying the new values */
 {
+	ut_ad(!index->table->skip_alter_undo);
+
 	ulint		i;
 	ulint		n_fields;
 	const page_size_t&	page_size = dict_table_page_size(index->table);
@@ -1392,6 +1398,8 @@ row_upd_index_replace_new_col_vals(
 		= dict_table_get_first_index(index->table);
 	const page_size_t&	page_size = dict_table_page_size(index->table);
 
+	ut_ad(!index->table->skip_alter_undo);
+
 	dtuple_set_info_bits(entry, update->info_bits);
 
 	for (i = 0; i < dict_index_get_n_fields(index); i++) {
@@ -1466,6 +1474,8 @@ row_upd_replace_vcol(
 	ulint			i;
 	ulint			n_cols;
 
+	ut_ad(!table->skip_alter_undo);
+
 	n_cols = dtuple_get_n_v_fields(row);
 	for (col_no = 0; col_no < n_cols; col_no++) {
 		dfield_t*		dfield;
@@ -1687,6 +1697,7 @@ row_upd_changes_ord_field_binary_func(
 	ut_ad(thr);
 	ut_ad(thr->graph);
 	ut_ad(thr->graph->trx);
+	ut_ad(!index->table->skip_alter_undo);
 
 	n_unique = dict_index_get_n_unique(index);
 
@@ -1946,6 +1957,8 @@ row_upd_changes_doc_id(
 	dict_index_t*	clust_index;
 	fts_t*		fts = table->fts;
 
+	ut_ad(!table->skip_alter_undo);
+
 	clust_index = dict_table_get_first_index(table);
 
 	/* Convert from index-specific column number to table-global
@@ -1968,6 +1981,8 @@ row_upd_changes_fts_column(
 	dict_index_t*	clust_index;
 	fts_t*		fts = table->fts;
 
+	ut_ad(!table->skip_alter_undo);
+
 	if (upd_fld_is_virtual_col(upd_field)) {
 		col_no = upd_field->field_no;
 		return(dict_table_is_fts_column(fts->indexes, col_no, true));
@@ -2814,6 +2829,7 @@ row_upd_clust_rec(
 	ut_ad(node);
 	ut_ad(dict_index_is_clust(index));
 	ut_ad(!thr_get_trx(thr)->in_rollback);
+	ut_ad(!node->table->skip_alter_undo);
 
 	pcur = node->pcur;
 	btr_cur = btr_pcur_get_btr_cur(pcur);
diff --git a/storage/innobase/trx/trx0rec.cc b/storage/innobase/trx/trx0rec.cc
index de32da15de0..e127873fc5d 100644
--- a/storage/innobase/trx/trx0rec.cc
+++ b/storage/innobase/trx/trx0rec.cc
@@ -2236,6 +2236,8 @@ trx_undo_prev_version_build(
 	const bool is_temp = dict_table_is_temporary(index->table);
 	rec_trx_id = row_get_rec_trx_id(rec, index, offsets);
 
+	ut_ad(!index->table->skip_alter_undo);
+
 	if (trx_undo_get_undo_rec(
 		    roll_ptr, is_temp, heap, rec_trx_id, index->table->name,
 		    &undo_rec)) {
diff --git a/storage/mroonga/ha_mroonga.cpp b/storage/mroonga/ha_mroonga.cpp
index 174ee66b314..b4bfc152053 100644
--- a/storage/mroonga/ha_mroonga.cpp
+++ b/storage/mroonga/ha_mroonga.cpp
@@ -538,6 +538,15 @@ static const char *mrn_inspect_extra_function(enum ha_extra_function operation)
   case HA_EXTRA_DETACH_CHILDREN:
     inspected = "HA_EXTRA_DETACH_CHILDREN";
     break;
+  case HA_EXTRA_BEGIN_ALTER_COPY:
+    inspected = "HA_EXTRA_BEGIN_ALTER_COPY";
+    break;
+  case HA_EXTRA_END_ALTER_COPY:
+    inspected = "HA_EXTRA_END_ALTER_COPY";
+    break;
+  case HA_EXTRA_FAKE_START_STMT:
+    inspected = "HA_EXTRA_FAKE_START_STMT";
+    break;
 #ifdef MRN_HAVE_HA_EXTRA_EXPORT
   case HA_EXTRA_EXPORT:
     inspected = "HA_EXTRA_EXPORT";