MDEV-16329 [5/5] ALTER ONLINE TABLE

* Log rows in online_alter_binlog.
* Table online data is replicated within dedicated binlog file
* Cached data is written on commit.
* Versioning is fully supported.
* Works both wit and without binlog enabled.

* For now savepoints setup is forbidden while ONLINE ALTER goes on.
  Extra support is required. We can simply log the SAVEPOINT query events
  and replicate them together with row events. But it's not implemented
  for now.

* Cache flipping:

  We want to care for the possible bottleneck in the online alter binlog
  reading/writing in advance.

  IO_CACHE does not provide anything better that sequential access,
  besides, only a single write is mutex-protected, which is not suitable,
  since we should write a transaction atomically.

  To solve this, a special layer on top Event_log is implemented.
  There are two IO_CACHE files underneath: one for reading, and one for
  writing.

  Once the read cache is empty, an exclusive lock is acquired (we can wait
  for a currently active transaction finish writing), and flip() is emitted,
  i.e. the write cache is reopened for read, and the read cache is emptied,
  and reopened for writing.

  This reminds a buffer flip that happens in accelerated graphics
  (DirectX/OpenGL/etc).

  Cache_flip_event_log is considered non-blocking for a single reader and a
  single writer in this sense, with the only lock held by reader during flip.

  An alternative approach by implementing a fair concurrent circular buffer
  is described in MDEV-24676.

* Cache managers:
  We have two cache sinks: statement and transactional.
  It is important that the changes are first cached per-statement and
  per-transaction.
  If a statement fails, then only statement data is rolled back. The
  transaction moves along, however.

  Turns out, there's no guarantee that TABLE well persist in
  thd->open_tables to the transaction commit moment.
  If an error occurs, tables from statement are purged.
  Therefore, we can't store te caches in TABLE. Ideally, it should be
  handlerton, but we cut the corner and store it in THD in a list.

42 files changed, 1838 insertions, 181 deletions

diff --git a/mysql-test/include/have_log_bin_off.require b/mysql-test/include/have_log_bin_off.require
new file mode 100644
index 00000000000..979dbe75f80
--- /dev/null
+++ b/mysql-test/include/have_log_bin_off.require
@@ -0,0 +1,2 @@
+Variable_name	Value
+log_bin	OFF
diff --git a/mysql-test/main/alter_table.result b/mysql-test/main/alter_table.result
index 879bc0edfec..fae913369c0 100644
--- a/mysql-test/main/alter_table.result
+++ b/mysql-test/main/alter_table.result
@@ -1756,7 +1756,6 @@ info: Records: 0  Duplicates: 0  Warnings: 1
 Warnings:
 Note	1831	Duplicate index `i3`. This is deprecated and will be disallowed in a future release
 ALTER TABLE t1 ADD INDEX i4(b), ALGORITHM= COPY, LOCK= NONE;
-ERROR 0A000: LOCK=NONE is not supported. Reason: COPY algorithm requires a lock. Try LOCK=SHARED
 ALTER TABLE t1 ADD INDEX i5(b), ALGORITHM= COPY, LOCK= SHARED;
 affected rows: 2
 info: Records: 2  Duplicates: 0  Warnings: 1
@@ -1774,7 +1773,6 @@ ERROR 0A000: LOCK=NONE/SHARED is not supported for this operation. Try LOCK=EXCL
 ALTER TABLE m1 ENABLE KEYS, ALGORITHM= INPLACE, LOCK= EXCLUSIVE;
 affected rows: 0
 ALTER TABLE m1 ENABLE KEYS, ALGORITHM= COPY, LOCK= NONE;
-ERROR 0A000: LOCK=NONE is not supported. Reason: COPY algorithm requires a lock. Try LOCK=SHARED
 ALTER ONLINE TABLE m1 ADD COLUMN c int;
 ERROR 0A000: LOCK=NONE is not supported for this operation. Try LOCK=SHARED
 ALTER TABLE m1 ENABLE KEYS, ALGORITHM= COPY, LOCK= SHARED;
diff --git a/mysql-test/main/alter_table.test b/mysql-test/main/alter_table.test
index 8ddf60f26d8..cc8ec468320 100644
--- a/mysql-test/main/alter_table.test
+++ b/mysql-test/main/alter_table.test
@@ -1531,8 +1531,12 @@ ALTER TABLE t1 DROP INDEX i1, DROP INDEX i2, DROP INDEX i3, DROP INDEX i4;
 ALTER TABLE t1 ADD INDEX i1(b), ALGORITHM= INPLACE, LOCK= NONE;
 ALTER TABLE t1 ADD INDEX i2(b), ALGORITHM= INPLACE, LOCK= SHARED;
 ALTER TABLE t1 ADD INDEX i3(b), ALGORITHM= INPLACE, LOCK= EXCLUSIVE;
---error ER_ALTER_OPERATION_NOT_SUPPORTED_REASON
+--disable_info
+--disable_warnings
+--error 0,ER_ALTER_OPERATION_NOT_SUPPORTED_REASON
 ALTER TABLE t1 ADD INDEX i4(b), ALGORITHM= COPY, LOCK= NONE;
+--enable_warnings
+--enable_info
 ALTER TABLE t1 ADD INDEX i5(b), ALGORITHM= COPY, LOCK= SHARED;
 ALTER TABLE t1 ADD INDEX i6(b), ALGORITHM= COPY, LOCK= EXCLUSIVE;
 
@@ -1541,8 +1545,10 @@ ALTER TABLE m1 ENABLE KEYS, ALGORITHM= INPLACE, LOCK= NONE;
 --error ER_ALTER_OPERATION_NOT_SUPPORTED
 ALTER TABLE m1 ENABLE KEYS, ALGORITHM= INPLACE, LOCK= SHARED;
 ALTER TABLE m1 ENABLE KEYS, ALGORITHM= INPLACE, LOCK= EXCLUSIVE;
---error ER_ALTER_OPERATION_NOT_SUPPORTED_REASON
+--disable_info
+--error 0,ER_ALTER_OPERATION_NOT_SUPPORTED_REASON
 ALTER TABLE m1 ENABLE KEYS, ALGORITHM= COPY, LOCK= NONE;
+--enable_info
 --error ER_ALTER_OPERATION_NOT_SUPPORTED
 ALTER ONLINE TABLE m1 ADD COLUMN c int;
 # This works because the lock will be SNW for the copy phase.
diff --git a/mysql-test/main/alter_table_locknone.result b/mysql-test/main/alter_table_locknone.result
index 4d0e0f09f71..05043f0b610 100644
--- a/mysql-test/main/alter_table_locknone.result
+++ b/mysql-test/main/alter_table_locknone.result
@@ -71,7 +71,6 @@ ERROR 0A000: LOCK=NONE is not supported for this operation. Try LOCK=SHARED
 alter online table t1 add f int;
 ERROR 0A000: LOCK=NONE is not supported for this operation. Try LOCK=SHARED
 alter online table t1 engine=memory;
-ERROR 0A000: LOCK=NONE is not supported. Reason: COPY algorithm requires a lock. Try LOCK=SHARED
 alter online table t1 rename to t2;
 ERROR 0A000: LOCK=NONE/SHARED is not supported for this operation. Try LOCK=EXCLUSIVE
 alter online table t1 checksum=1;
diff --git a/mysql-test/main/alter_table_locknone.test b/mysql-test/main/alter_table_locknone.test
index 75efd0ce301..f07ded67cd2 100644
--- a/mysql-test/main/alter_table_locknone.test
+++ b/mysql-test/main/alter_table_locknone.test
@@ -77,7 +77,7 @@ alter online table t1 modify c varchar(50);
 alter online table t1 modify c varchar(100);
 --error ER_ALTER_OPERATION_NOT_SUPPORTED
 alter online table t1 add f int;
---error ER_ALTER_OPERATION_NOT_SUPPORTED_REASON
+--error 0,ER_ALTER_OPERATION_NOT_SUPPORTED_REASON
 alter online table t1 engine=memory;
 --error ER_ALTER_OPERATION_NOT_SUPPORTED
 alter online table t1 rename to t2;
diff --git a/mysql-test/main/alter_table_online.combinations b/mysql-test/main/alter_table_online.combinations
new file mode 100644
index 00000000000..ae144432c68
--- /dev/null
+++ b/mysql-test/main/alter_table_online.combinations
@@ -0,0 +1,2 @@
+[innodb]
+[rocksdb]
diff --git a/mysql-test/main/alter_table_online.result b/mysql-test/main/alter_table_online.result
new file mode 100644
index 00000000000..357b55d7612
--- /dev/null
+++ b/mysql-test/main/alter_table_online.result
@@ -0,0 +1,460 @@
+connect  con2, localhost, root,,;
+connection default;
+#
+# Test insert
+#
+# Insert and add column
+create or replace table t1 (a int) engine=innodb;
+insert t1 values (5);
+connection con2;
+set debug_sync= 'now WAIT_FOR ended';
+connection default;
+set debug_sync= 'alter_table_copy_end SIGNAL ended WAIT_FOR end';
+alter table t1 add b int NULL, algorithm= copy, lock= none;
+connection con2;
+insert into t1 values (123), (456), (789);
+set debug_sync= 'now SIGNAL end';
+connection default;
+select * from t1;
+a	b
+5	NULL
+123	NULL
+456	NULL
+789	NULL
+# Insert and add NOT NULL column without default value
+create or replace table t1 (a int) engine=innodb;
+insert t1 values (5);
+connection con2;
+set debug_sync= 'now WAIT_FOR ended';
+connection default;
+set debug_sync= 'alter_table_copy_end SIGNAL ended WAIT_FOR end';
+alter table t1 add b int NOT NULL, algorithm= copy, lock= none;
+connection con2;
+insert into t1 values (123), (456), (789);
+set debug_sync= 'now SIGNAL end';
+connection default;
+Warnings:
+Warning	1364	Field 'b' doesn't have a default value
+Warning	1364	Field 'b' doesn't have a default value
+Warning	1364	Field 'b' doesn't have a default value
+select * from t1;
+a	b
+5	0
+123	0
+456	0
+789	0
+# Insert and add a column with a default value
+create or replace table t1 (a int) engine=innodb;
+insert t1 values (5);
+connection con2;
+set debug_sync= 'now WAIT_FOR ended';
+connection default;
+set debug_sync= 'alter_table_copy_end SIGNAL ended WAIT_FOR end';
+alter table t1 add b int NOT NULL default (222), algorithm= copy, lock= none;
+connection con2;
+insert into t1 values (123), (456), (789);
+set debug_sync= 'now SIGNAL end';
+connection default;
+select * from t1;
+a	b
+5	222
+123	222
+456	222
+789	222
+#
+# Test update
+#
+# Update and add a column
+create or replace table t1 (a int primary key, b int) engine=innodb;
+insert t1 values (1, 22);
+insert t1 values (3, 44);
+connection con2;
+set debug_sync= 'now WAIT_FOR ended';
+connection default;
+set debug_sync= 'alter_table_copy_end SIGNAL ended WAIT_FOR end';
+alter table t1 add c int default(1),
+algorithm= copy, lock= none;
+connection con2;
+update t1 set b= 55 where a = 1;
+set debug_sync= 'now SIGNAL end';
+connection default;
+select * from t1;
+a	b	c
+1	55	1
+3	44	1
+#
+# Test primary key change
+#
+# Drop key, add key
+create or replace table t1 (a int primary key, b int) engine=innodb;
+insert t1 values (1, 22);
+insert t1 values (3, 44);
+connection con2;
+set debug_sync= 'now WAIT_FOR ended';
+connection default;
+set debug_sync= 'alter_table_copy_end SIGNAL ended WAIT_FOR end';
+alter table t1 drop primary key, add primary key(b),
+algorithm= copy, lock= none;
+connection con2;
+update t1 set b= 55 where a = 1;
+set debug_sync= 'now SIGNAL end';
+connection default;
+select * from t1;
+a	b
+3	44
+1	55
+# Drop key, add key. Two updates
+create or replace table t1 (a int primary key, b int) engine=innodb;
+insert t1 values (1, 11);
+insert t1 values (2, 22);
+connection con2;
+set debug_sync= 'now WAIT_FOR ended';
+connection default;
+set debug_sync= 'alter_table_copy_end SIGNAL ended WAIT_FOR end';
+alter table t1 drop primary key, add primary key(b),
+algorithm= copy, lock= none;
+connection con2;
+update t1 set b= 33 where a = 1;
+update t1 set b= 44 where a = 2;
+set debug_sync= 'now SIGNAL end';
+connection default;
+select * from t1;
+a	b
+1	33
+2	44
+#
+# Various tests, see below
+#
+create or replace table t1 (a int primary key, b int) engine=innodb;
+insert t1 values (1, 11);
+insert t1 values (2, 22);
+insert t1 values (3, 33);
+insert t1 values (4, 44);
+insert t1 values (5, 55);
+insert t1 values (6, 66);
+insert t1 values (7, 77);
+insert t1 values (8, 88);
+insert t1 values (9, 99);
+connection con2;
+set debug_sync= 'now WAIT_FOR ended';
+connection default;
+set debug_sync= 'alter_table_copy_end SIGNAL ended WAIT_FOR end';
+alter table t1 drop primary key, add primary key(b),
+algorithm= copy, lock= none;
+connection con2;
+# Two updates
+update t1 set b= 1001 where a = 1;
+update t1 set b= 2002 where a = 2;
+# Two updates in transaction
+set autocommit = 0;
+start transaction;
+update t1 set b= 3003 where a = 3;
+update t1 set b= 4004 where a = 4;
+commit;
+set autocommit = 1;
+# Second update is rolled back
+update t1 set b= 5005 where a = 5;
+set autocommit = 0;
+start transaction;
+update t1 set b= 6006 where a = 6;
+rollback;
+set autocommit = 1;
+# Second execution in transaction fails
+set autocommit = 0;
+start transaction;
+update t1 set b= 7007 where a = 7;
+update t1 set a= 8, b= 8008 where a = 8 or a = 9 order by a;
+ERROR 23000: Duplicate entry '8' for key 'PRIMARY'
+commit;
+set autocommit = 1;
+select * from t1;
+a	b
+1	1001
+2	2002
+3	3003
+4	4004
+5	5005
+6	66
+7	7007
+8	88
+9	99
+set debug_sync= 'now SIGNAL end';
+connection default;
+select * from t1;
+a	b
+1	1001
+2	2002
+3	3003
+4	4004
+5	5005
+6	66
+7	7007
+8	88
+9	99
+#
+# MYISAM. Only Inserts can be tested.
+#
+create or replace table t1 (a int) engine=myisam;
+insert t1 values (5);
+connection con2;
+set debug_sync= 'now WAIT_FOR ended';
+connection default;
+set debug_sync= 'alter_table_copy_end SIGNAL ended WAIT_FOR end';
+alter table t1 add b int NULL, algorithm= copy, lock= none;
+connection con2;
+insert into t1 values (123), (456), (789);
+set debug_sync= 'now SIGNAL end';
+connection default;
+select * from t1;
+a	b
+5	NULL
+123	NULL
+456	NULL
+789	NULL
+# Test incompatible changes
+create or replace table t1 (a int primary key, b int) engine=innodb;
+insert t1 values (1, 22);
+insert t1 values (3, 44);
+connection con2;
+set debug_sync= 'now WAIT_FOR ended';
+connection default;
+set debug_sync= 'alter_table_copy_end SIGNAL ended WAIT_FOR end';
+alter table t1 drop primary key, add primary key(b),
+algorithm= copy, lock= none;
+connection con2;
+update t1 set b= 44 where a = 1;
+set debug_sync= 'now SIGNAL end';
+connection default;
+ERROR 23000: Duplicate entry '44' for key 'PRIMARY'
+select * from t1;
+a	b
+1	44
+3	44
+# Test log read after EXCLUSIVE lock
+# Transaction is started before ALTER, and UPDATE is made.
+# Then more UPDATEs.
+create or replace table t1 (a int primary key, b int) engine=innodb;
+insert t1 values (1, 11);
+insert t1 values (2, 22);
+insert t1 values (3, 33);
+insert t1 values (4, 44);
+insert t1 values (5, 55);
+set debug_sync= 'alter_table_online_before_lock SIGNAL locking WAIT_FOR end';
+set debug_sync= 'alter_table_online_downgraded SIGNAL downgraded';
+alter table t1 drop primary key, add primary key(b),
+algorithm= copy, lock= none;
+connection con2;
+begin;
+set debug_sync= 'now WAIT_FOR downgraded';
+update t1 set b= 111 where a = 1;
+set debug_sync= 'now WAIT_FOR locking';
+set debug_sync= 'now SIGNAL end';
+update t1 set b= 222 where a = 2;
+update t1 set b= 333 where a = 3;
+update t1 set b= 444 where a = 4;
+commit;
+update t1 set b= 555 where a = 5;
+connection default;
+select * from t1;
+a	b
+1	111
+2	222
+3	333
+4	444
+5	555
+#
+# Test progress report.
+#
+create or replace table t1 (a int primary key, b int) engine=innodb;
+insert t1 values (1, 11);
+insert t1 values (2, 22);
+insert t1 values (3, 33);
+insert t1 values (4, 44);
+set debug_sync= 'alter_table_online_before_lock SIGNAL locking WAIT_FOR end';
+set debug_sync= 'alter_table_online_downgraded SIGNAL downgraded'
+                                             ' WAIT_FOR start_replication';
+set debug_sync= 'alter_table_online_progress SIGNAL applied WAIT_FOR proceed'
+                                           ' EXECUTE 9';
+alter table t1 drop primary key, add primary key(b),
+algorithm= copy, lock= none;
+connection con2;
+set debug_sync= 'now WAIT_FOR downgraded';
+update t1 set b= 111 where a = 1;
+insert t1 values (5, 55);
+update t1 set b= 555 where a = 5;
+insert t1 values (6, 66);
+update t1 set b= 666 where a = 6;
+set debug_sync= 'now SIGNAL start_replication';
+# First signal is for log description event.
+set debug_sync= 'now WAIT_FOR applied';
+select stage, progress from INFORMATION_SCHEMA.PROCESSLIST where id = @con;
+stage	progress
+3	51.220
+set debug_sync= 'now SIGNAL proceed WAIT_FOR applied';
+select stage, progress from INFORMATION_SCHEMA.PROCESSLIST where id = @con;
+stage	progress
+3	61.789
+set debug_sync= 'now SIGNAL proceed WAIT_FOR applied';
+select stage, progress from INFORMATION_SCHEMA.PROCESSLIST where id = @con;
+stage	progress
+3	70.325
+set debug_sync= 'now SIGNAL proceed WAIT_FOR applied';
+select stage, progress from INFORMATION_SCHEMA.PROCESSLIST where id = @con;
+stage	progress
+3	80.894
+set debug_sync= 'now SIGNAL proceed WAIT_FOR applied';
+select stage, progress from INFORMATION_SCHEMA.PROCESSLIST where id = @con;
+stage	progress
+3	89.431
+set debug_sync= 'now SIGNAL proceed WAIT_FOR applied';
+select stage, progress from INFORMATION_SCHEMA.PROCESSLIST where id = @con;
+stage	progress
+3	100.000
+set debug_sync= 'now SIGNAL proceed WAIT_FOR locking';
+begin;
+update t1 set b= 222 where a = 2;
+update t1 set b= 333 where a = 3;
+update t1 set b= 444 where a = 4;
+commit;
+set debug_sync= 'now SIGNAL end WAIT_FOR applied';
+select stage, progress from INFORMATION_SCHEMA.PROCESSLIST where id = @con;
+stage	progress
+4	33.333
+set debug_sync= 'now SIGNAL proceed WAIT_FOR applied';
+select stage, progress from INFORMATION_SCHEMA.PROCESSLIST where id = @con;
+stage	progress
+4	66.667
+set debug_sync= 'now SIGNAL proceed WAIT_FOR applied';
+select stage, progress from INFORMATION_SCHEMA.PROCESSLIST where id = @con;
+stage	progress
+4	100.000
+set debug_sync= 'now SIGNAL proceed';
+connection default;
+select * from t1;
+a	b
+1	111
+2	222
+3	333
+4	444
+5	555
+6	666
+#
+# Test system versioning
+#
+create or replace table t1 (a int primary key, b int);
+insert t1 values (1, 22);
+insert t1 values (3, 44);
+connection con2;
+set debug_sync= 'now WAIT_FOR ended';
+connection default;
+set debug_sync= 'alter_table_copy_end SIGNAL ended WAIT_FOR end';
+set timestamp = 1;
+alter table t1 add system versioning,
+algorithm= copy, lock= none;
+connection con2;
+set timestamp = 2;
+update t1 set b= 55 where a = 1;
+set timestamp = 3;
+insert into t1 values (6, 77);
+set debug_sync= 'now SIGNAL end';
+connection default;
+show create table t1;
+Table	Create Table
+t1	CREATE TABLE `t1` (
+  `a` int(11) NOT NULL,
+  `b` int(11) DEFAULT NULL,
+  PRIMARY KEY (`a`)
+) ENGINE=*SUBSTITUTED* DEFAULT CHARSET=latin1 COLLATE=latin1_swedish_ci WITH SYSTEM VERSIONING
+select *, UNIX_TIMESTAMP(row_start), UNIX_TIMESTAMP(row_end) from t1 for system_time all;
+a	b	UNIX_TIMESTAMP(row_start)	UNIX_TIMESTAMP(row_end)
+1	55	1.000000	2147483647.999999
+3	44	1.000000	2147483647.999999
+6	77	1.000000	2147483647.999999
+connection con2;
+set debug_sync= 'now WAIT_FOR ended';
+connection default;
+set debug_sync= 'alter_table_copy_end SIGNAL ended WAIT_FOR end';
+alter table t1 drop system versioning,
+algorithm= copy, lock= none;
+connection con2;
+update t1 set b= 88 where a = 1;
+set debug_sync= 'now SIGNAL end';
+connection default;
+# Can't UPDATE versioned -> plain (and can't DELETE)
+ERROR 23000: Duplicate entry '1' for key 'PRIMARY'
+show create table t1;
+Table	Create Table
+t1	CREATE TABLE `t1` (
+  `a` int(11) NOT NULL,
+  `b` int(11) DEFAULT NULL,
+  PRIMARY KEY (`a`)
+) ENGINE=*SUBSTITUTED* DEFAULT CHARSET=latin1 COLLATE=latin1_swedish_ci WITH SYSTEM VERSIONING
+select *, UNIX_TIMESTAMP(row_start), UNIX_TIMESTAMP(row_end) from t1 for system_time all;
+a	b	UNIX_TIMESTAMP(row_start)	UNIX_TIMESTAMP(row_end)
+1	55	1.000000	3.000000
+1	88	3.000000	2147483647.999999
+3	44	1.000000	2147483647.999999
+6	77	1.000000	2147483647.999999
+connection con2;
+set debug_sync= 'now WAIT_FOR ended';
+connection default;
+set debug_sync= 'alter_table_copy_end SIGNAL ended WAIT_FOR end';
+alter table t1 drop system versioning,
+algorithm= copy, lock= none;
+connection con2;
+insert into t1 values (8, 99);
+set debug_sync= 'now SIGNAL end';
+connection default;
+# INSERT versioned -> plain works fine since it is a single versioned op.
+show create table t1;
+Table	Create Table
+t1	CREATE TABLE `t1` (
+  `a` int(11) NOT NULL,
+  `b` int(11) DEFAULT NULL,
+  PRIMARY KEY (`a`)
+) ENGINE=*SUBSTITUTED* DEFAULT CHARSET=latin1 COLLATE=latin1_swedish_ci
+select * from t1;
+a	b
+1	88
+3	44
+6	77
+8	99
+#
+# Test ROLLBACK TO SAVEPOINT
+#
+create or replace table t1 (a int) engine=innodb;
+insert t1 values (1), (2);
+create or replace table t2 (a int) engine=innodb;
+insert t2 values (1), (2);
+connection con2;
+begin;
+update t2 set a= 222 where a = 2;
+savepoint savie;
+update t2 set a= 111 where a = 1;
+set debug_sync= 'now WAIT_FOR ended';
+connection default;
+set debug_sync= 'alter_table_copy_end SIGNAL ended WAIT_FOR end';
+alter table t1 add b int NULL, algorithm= copy, lock= none;
+connection con2;
+update t1 set a= 123 where a = 1;
+savepoint whoopsie;
+ERROR 42000: Cannot set up a savepoint while online ALTER TABLE is in progress
+show warnings;
+Level	Code	Message
+Error	1235	Cannot set up a savepoint while online ALTER TABLE is in progress
+rollback to savepoint savie;
+commit;
+set debug_sync= 'now SIGNAL end';
+connection default;
+select * from t1;
+a	b
+1	NULL
+2	NULL
+select * from t2;
+a
+1
+222
+# Cleanup
+set debug_sync= 'reset';
+drop table t1;
+drop table t2;
diff --git a/mysql-test/main/alter_table_online.test b/mysql-test/main/alter_table_online.test
new file mode 100644
index 00000000000..3a858ab87e1
--- /dev/null
+++ b/mysql-test/main/alter_table_online.test
@@ -0,0 +1,551 @@
+-- source include/have_debug_sync.inc
+-- source include/not_embedded.inc
+-- source alter_table_online_binlog.inc
+
+-- disable_query_log
+-- if ($MTR_COMBINATION_INNODB) {
+-- source include/have_innodb.inc
+set default_storage_engine= innodb;
+-- }
+-- if ($MTR_COMBINATION_ROCKSDB) {
+-- source include/have_rocksdb.inc
+set default_storage_engine= rocksdb;
+-- }
+-- enable_query_log
+
+-- let $default_engine= `select @@default_storage_engine`
+
+--connect (con2, localhost, root,,)
+--connection default
+
+
+--echo #
+--echo # Test insert
+--echo #
+
+--echo # Insert and add column
+create or replace table t1 (a int) engine=innodb;
+insert t1 values (5);
+
+--connection con2
+--send
+set debug_sync= 'now WAIT_FOR ended';
+
+--connection default
+set debug_sync= 'alter_table_copy_end SIGNAL ended WAIT_FOR end';
+
+--send
+alter table t1 add b int NULL, algorithm= copy, lock= none;
+
+--connection con2
+--reap
+insert into t1 values (123), (456), (789);
+set debug_sync= 'now SIGNAL end';
+
+--connection default
+--reap
+select * from t1;
+
+--echo # Insert and add NOT NULL column without default value
+create or replace table t1 (a int) engine=innodb;
+insert t1 values (5);
+
+--connection con2
+--send
+set debug_sync= 'now WAIT_FOR ended';
+
+--connection default
+set debug_sync= 'alter_table_copy_end SIGNAL ended WAIT_FOR end';
+
+--send
+alter table t1 add b int NOT NULL, algorithm= copy, lock= none;
+
+--connection con2
+--reap
+insert into t1 values (123), (456), (789);
+set debug_sync= 'now SIGNAL end';
+
+--connection default
+--reap
+select * from t1;
+
+--echo # Insert and add a column with a default value
+create or replace table t1 (a int) engine=innodb;
+insert t1 values (5);
+
+--connection con2
+--send
+set debug_sync= 'now WAIT_FOR ended';
+
+--connection default
+set debug_sync= 'alter_table_copy_end SIGNAL ended WAIT_FOR end';
+
+--send
+alter table t1 add b int NOT NULL default (222), algorithm= copy, lock= none;
+
+--connection con2
+--reap
+insert into t1 values (123), (456), (789);
+set debug_sync= 'now SIGNAL end';
+
+--connection default
+--reap
+select * from t1;
+
+--echo #
+--echo # Test update
+--echo #
+
+--echo # Update and add a column
+create or replace table t1 (a int primary key, b int) engine=innodb;
+insert t1 values (1, 22);
+insert t1 values (3, 44);
+
+--connection con2
+--send
+set debug_sync= 'now WAIT_FOR ended';
+
+--connection default
+set debug_sync= 'alter_table_copy_end SIGNAL ended WAIT_FOR end';
+
+--send
+alter table t1 add c int default(1),
+            algorithm= copy, lock= none;
+
+--connection con2
+--reap
+update t1 set b= 55 where a = 1;
+set debug_sync= 'now SIGNAL end';
+
+--connection default
+--reap
+select * from t1;
+
+--echo #
+--echo # Test primary key change
+--echo #
+
+--echo # Drop key, add key
+create or replace table t1 (a int primary key, b int) engine=innodb;
+insert t1 values (1, 22);
+insert t1 values (3, 44);
+
+--connection con2
+--send
+set debug_sync= 'now WAIT_FOR ended';
+
+--connection default
+set debug_sync= 'alter_table_copy_end SIGNAL ended WAIT_FOR end';
+
+--send
+alter table t1 drop primary key, add primary key(b),
+            algorithm= copy, lock= none;
+
+--connection con2
+--reap
+update t1 set b= 55 where a = 1;
+set debug_sync= 'now SIGNAL end';
+
+--connection default
+--reap
+select * from t1;
+
+--echo # Drop key, add key. Two updates
+create or replace table t1 (a int primary key, b int) engine=innodb;
+insert t1 values (1, 11);
+insert t1 values (2, 22);
+
+--connection con2
+--send
+set debug_sync= 'now WAIT_FOR ended';
+
+--connection default
+set debug_sync= 'alter_table_copy_end SIGNAL ended WAIT_FOR end';
+
+--send
+alter table t1 drop primary key, add primary key(b),
+            algorithm= copy, lock= none;
+
+--connection con2
+--reap
+update t1 set b= 33 where a = 1;
+update t1 set b= 44 where a = 2;
+
+set debug_sync= 'now SIGNAL end';
+
+--connection default
+--reap
+select * from t1;
+
+--echo #
+--echo # Various tests, see below
+--echo #
+
+create or replace table t1 (a int primary key, b int) engine=innodb;
+insert t1 values (1, 11);
+insert t1 values (2, 22);
+insert t1 values (3, 33);
+insert t1 values (4, 44);
+insert t1 values (5, 55);
+insert t1 values (6, 66);
+insert t1 values (7, 77);
+insert t1 values (8, 88);
+insert t1 values (9, 99);
+
+--connection con2
+--send
+set debug_sync= 'now WAIT_FOR ended';
+
+--connection default
+set debug_sync= 'alter_table_copy_end SIGNAL ended WAIT_FOR end';
+
+--send
+alter table t1 drop primary key, add primary key(b),
+            algorithm= copy, lock= none;
+
+--connection con2
+--reap
+
+--echo # Two updates
+update t1 set b= 1001 where a = 1;
+update t1 set b= 2002 where a = 2;
+
+--echo # Two updates in transaction
+set autocommit = 0;
+start transaction;
+update t1 set b= 3003 where a = 3;
+update t1 set b= 4004 where a = 4;
+commit;
+set autocommit = 1;
+
+--echo # Second update is rolled back
+update t1 set b= 5005 where a = 5;
+
+set autocommit = 0;
+start transaction;
+update t1 set b= 6006 where a = 6;
+rollback;
+set autocommit = 1;
+
+--echo # Second execution in transaction fails
+
+set autocommit = 0;
+start transaction;
+update t1 set b= 7007 where a = 7;
+--error ER_DUP_ENTRY
+update t1 set a= 8, b= 8008 where a = 8 or a = 9 order by a;
+commit;
+set autocommit = 1;
+
+select * from t1;
+
+set debug_sync= 'now SIGNAL end';
+
+--connection default
+--reap
+--sorted_result
+select * from t1;
+
+--echo #
+--echo # MYISAM. Only Inserts can be tested.
+--echo #
+
+create or replace table t1 (a int) engine=myisam;
+insert t1 values (5);
+
+--connection con2
+--send
+set debug_sync= 'now WAIT_FOR ended';
+
+--connection default
+set debug_sync= 'alter_table_copy_end SIGNAL ended WAIT_FOR end';
+
+--send
+alter table t1 add b int NULL, algorithm= copy, lock= none;
+
+--connection con2
+--reap
+insert into t1 values (123), (456), (789);
+set debug_sync= 'now SIGNAL end';
+
+--connection default
+--reap
+select * from t1;
+
+--echo # Test incompatible changes
+create or replace table t1 (a int primary key, b int) engine=innodb;
+insert t1 values (1, 22);
+insert t1 values (3, 44);
+
+--connection con2
+--send
+set debug_sync= 'now WAIT_FOR ended';
+
+--connection default
+set debug_sync= 'alter_table_copy_end SIGNAL ended WAIT_FOR end';
+
+--send
+alter table t1 drop primary key, add primary key(b),
+            algorithm= copy, lock= none;
+
+--connection con2
+--reap
+update t1 set b= 44 where a = 1;
+set debug_sync= 'now SIGNAL end';
+
+--connection default
+--error ER_DUP_ENTRY
+--reap
+select * from t1;
+
+
+--echo # Test log read after EXCLUSIVE lock
+--echo # Transaction is started before ALTER, and UPDATE is made.
+--echo # Then more UPDATEs.
+
+create or replace table t1 (a int primary key, b int) engine=innodb;
+insert t1 values (1, 11);
+insert t1 values (2, 22);
+insert t1 values (3, 33);
+insert t1 values (4, 44);
+insert t1 values (5, 55);
+
+set debug_sync= 'alter_table_online_before_lock SIGNAL locking WAIT_FOR end';
+set debug_sync= 'alter_table_online_downgraded SIGNAL downgraded';
+
+--send
+alter table t1 drop primary key, add primary key(b),
+            algorithm= copy, lock= none;
+
+--connection con2
+begin;
+
+set debug_sync= 'now WAIT_FOR downgraded';
+update t1 set b= 111 where a = 1;
+
+set debug_sync= 'now WAIT_FOR locking';
+set debug_sync= 'now SIGNAL end';
+
+update t1 set b= 222 where a = 2;
+update t1 set b= 333 where a = 3;
+update t1 set b= 444 where a = 4;
+
+commit;
+
+update t1 set b= 555 where a = 5;
+
+--connection default
+--reap
+select * from t1;
+
+--echo #
+--echo # Test progress report.
+--echo #
+
+create or replace table t1 (a int primary key, b int) engine=innodb;
+insert t1 values (1, 11);
+insert t1 values (2, 22);
+insert t1 values (3, 33);
+insert t1 values (4, 44);
+
+set debug_sync= 'alter_table_online_before_lock SIGNAL locking WAIT_FOR end';
+set debug_sync= 'alter_table_online_downgraded SIGNAL downgraded'
+                                             ' WAIT_FOR start_replication';
+set debug_sync= 'alter_table_online_progress SIGNAL applied WAIT_FOR proceed'
+                                           ' EXECUTE 9';
+--let $con= `select connection_id()`
+
+--send
+alter table t1 drop primary key, add primary key(b),
+            algorithm= copy, lock= none;
+
+--connection con2
+
+set debug_sync= 'now WAIT_FOR downgraded';
+
+update t1 set b= 111 where a = 1;
+insert t1 values (5, 55);
+update t1 set b= 555 where a = 5;
+insert t1 values (6, 66);
+update t1 set b= 666 where a = 6;
+set debug_sync= 'now SIGNAL start_replication';
+
+--disable_query_log
+eval set @con= $con;
+--enable_query_log
+
+--echo # First signal is for log description event.
+set debug_sync= 'now WAIT_FOR applied';
+select stage, progress from INFORMATION_SCHEMA.PROCESSLIST where id = @con;
+set debug_sync= 'now SIGNAL proceed WAIT_FOR applied';
+
+select stage, progress from INFORMATION_SCHEMA.PROCESSLIST where id = @con;
+set debug_sync= 'now SIGNAL proceed WAIT_FOR applied';
+
+select stage, progress from INFORMATION_SCHEMA.PROCESSLIST where id = @con;
+set debug_sync= 'now SIGNAL proceed WAIT_FOR applied';
+
+select stage, progress from INFORMATION_SCHEMA.PROCESSLIST where id = @con;
+set debug_sync= 'now SIGNAL proceed WAIT_FOR applied';
+
+select stage, progress from INFORMATION_SCHEMA.PROCESSLIST where id = @con;
+set debug_sync= 'now SIGNAL proceed WAIT_FOR applied';
+
+select stage, progress from INFORMATION_SCHEMA.PROCESSLIST where id = @con;
+set debug_sync= 'now SIGNAL proceed WAIT_FOR locking';
+
+begin;
+
+update t1 set b= 222 where a = 2;
+update t1 set b= 333 where a = 3;
+update t1 set b= 444 where a = 4;
+
+commit;
+set debug_sync= 'now SIGNAL end WAIT_FOR applied';
+
+select stage, progress from INFORMATION_SCHEMA.PROCESSLIST where id = @con;
+set debug_sync= 'now SIGNAL proceed WAIT_FOR applied';
+
+select stage, progress from INFORMATION_SCHEMA.PROCESSLIST where id = @con;
+set debug_sync= 'now SIGNAL proceed WAIT_FOR applied';
+
+select stage, progress from INFORMATION_SCHEMA.PROCESSLIST where id = @con;
+set debug_sync= 'now SIGNAL proceed';
+
+--connection default
+--reap
+select * from t1;
+
+
+--echo #
+--echo # Test system versioning
+--echo #
+create or replace table t1 (a int primary key, b int);
+insert t1 values (1, 22);
+insert t1 values (3, 44);
+
+--connection con2
+--send
+set debug_sync= 'now WAIT_FOR ended';
+
+--connection default
+set debug_sync= 'alter_table_copy_end SIGNAL ended WAIT_FOR end';
+
+set timestamp = 1;
+
+--send
+alter table t1 add system versioning,
+            algorithm= copy, lock= none;
+
+--connection con2
+--reap
+set timestamp = 2;
+update t1 set b= 55 where a = 1;
+set timestamp = 3;
+insert into t1 values (6, 77);
+
+set debug_sync= 'now SIGNAL end';
+
+--connection default
+--reap
+-- replace_result $default_engine *SUBSTITUTED*
+show create table t1;
+select *, UNIX_TIMESTAMP(row_start), UNIX_TIMESTAMP(row_end) from t1 for system_time all;
+
+--connection con2
+--send
+set debug_sync= 'now WAIT_FOR ended';
+
+--connection default
+set debug_sync= 'alter_table_copy_end SIGNAL ended WAIT_FOR end';
+
+--send
+alter table t1 drop system versioning,
+            algorithm= copy, lock= none;
+
+--connection con2
+--reap
+update t1 set b= 88 where a = 1;
+
+set debug_sync= 'now SIGNAL end';
+
+--connection default
+--echo # Can't UPDATE versioned -> plain (and can't DELETE)
+--error ER_DUP_ENTRY
+--reap
+-- replace_result $default_engine *SUBSTITUTED*
+show create table t1;
+select *, UNIX_TIMESTAMP(row_start), UNIX_TIMESTAMP(row_end) from t1 for system_time all;
+
+--connection con2
+--send
+set debug_sync= 'now WAIT_FOR ended';
+
+--connection default
+set debug_sync= 'alter_table_copy_end SIGNAL ended WAIT_FOR end';
+
+--send
+alter table t1 drop system versioning,
+            algorithm= copy, lock= none;
+
+--connection con2
+--reap
+insert into t1 values (8, 99);
+
+set debug_sync= 'now SIGNAL end';
+
+--connection default
+--echo # INSERT versioned -> plain works fine since it is a single versioned op.
+--reap
+-- replace_result $default_engine *SUBSTITUTED*
+show create table t1;
+select * from t1;
+
+--echo #
+--echo # Test ROLLBACK TO SAVEPOINT
+--echo #
+
+create or replace table t1 (a int) engine=innodb;
+insert t1 values (1), (2);
+
+create or replace table t2 (a int) engine=innodb;
+insert t2 values (1), (2);
+
+--connection con2
+begin;
+update t2 set a= 222 where a = 2;
+savepoint savie;
+update t2 set a= 111 where a = 1;
+
+--send
+set debug_sync= 'now WAIT_FOR ended';
+
+--connection default
+set debug_sync= 'alter_table_copy_end SIGNAL ended WAIT_FOR end';
+
+--send
+alter table t1 add b int NULL, algorithm= copy, lock= none;
+
+--connection con2
+--reap
+update t1 set a= 123 where a = 1;
+
+--error ER_NOT_SUPPORTED_YET
+savepoint whoopsie;
+show warnings;
+
+rollback to savepoint savie;
+commit;
+
+set debug_sync= 'now SIGNAL end';
+
+--connection default
+--reap
+
+select * from t1;
+select * from t2;
+
+--echo # Cleanup
+set debug_sync= 'reset';
+drop table t1;
+drop table t2;
diff --git a/mysql-test/main/alter_table_online_binlog.combinations b/mysql-test/main/alter_table_online_binlog.combinations
new file mode 100644
index 00000000000..630977f9ead
--- /dev/null
+++ b/mysql-test/main/alter_table_online_binlog.combinations
@@ -0,0 +1,4 @@
+[standalone]
+[binlog]
+--log-bin=master-bin
+
diff --git a/mysql-test/main/alter_table_online_binlog.inc b/mysql-test/main/alter_table_online_binlog.inc
new file mode 100644
index 00000000000..2fe9677a5b7
--- /dev/null
+++ b/mysql-test/main/alter_table_online_binlog.inc
@@ -0,0 +1,14 @@
+#
+# Adds standalone and binlog combinations
+#
+-- if ($MTR_COMBINATION_BINLOG) {
+-- require include/have_log_bin.require
+-- }
+
+-- if ($MTR_COMBINATION_STANDALONE) {
+-- require include/have_log_bin_off.require
+-- }
+
+disable_query_log;
+show variables like 'log_bin';
+enable_query_log;
diff --git a/mysql-test/main/backup_lock.result b/mysql-test/main/backup_lock.result
index 16806148c67..e6c45f6337c 100644
--- a/mysql-test/main/backup_lock.result
+++ b/mysql-test/main/backup_lock.result
@@ -41,7 +41,7 @@ connection default;
 start transaction;
 insert into t1 values (1);
 connection con1;
-alter table t1 add column (j int), algorithm copy;
+alter table t1 add column (j int), algorithm copy, lock shared;
 connection con2;
 backup stage flush;
 SELECT LOCK_MODE, LOCK_TYPE, TABLE_SCHEMA, TABLE_NAME FROM information_schema.metadata_lock_info
diff --git a/mysql-test/main/backup_lock.test b/mysql-test/main/backup_lock.test
index 5453e5b0e7d..9de45256d34 100644
--- a/mysql-test/main/backup_lock.test
+++ b/mysql-test/main/backup_lock.test
@@ -51,7 +51,7 @@ insert into t1 values (1);
 
 connection con1;
 # Waits on MDL
---send alter table t1 add column (j int), algorithm copy
+--send alter table t1 add column (j int), algorithm copy, lock shared
 
 connection con2;
 let $wait_condition=
diff --git a/mysql-test/main/delayed.result b/mysql-test/main/delayed.result
index 4b482c8bb66..d5397e73a22 100644
--- a/mysql-test/main/delayed.result
+++ b/mysql-test/main/delayed.result
@@ -384,7 +384,7 @@ SELECT * FROM t1 WHERE a=0;
 a
 connection con1;
 # Sending:
-ALTER TABLE t1 MODIFY a INT UNSIGNED;;
+ALTER TABLE t1 MODIFY a INT UNSIGNED, LOCK=SHARED;;
 connection default;
 # Wait until ALTER TABLE is blocked on table 't1'.
 INSERT DELAYED INTO t1 VALUES (3);
diff --git a/mysql-test/main/delayed.test b/mysql-test/main/delayed.test
index a99af07232c..1de23aac127 100644
--- a/mysql-test/main/delayed.test
+++ b/mysql-test/main/delayed.test
@@ -458,7 +458,7 @@ SELECT * FROM t1 WHERE a=0;
 
 connection con1;
 --echo # Sending:
---send ALTER TABLE t1 MODIFY a INT UNSIGNED;
+--send ALTER TABLE t1 MODIFY a INT UNSIGNED, LOCK=SHARED;
 
 connection default;
 --echo # Wait until ALTER TABLE is blocked on table 't1'.
diff --git a/mysql-test/main/lock_multi_bug38499.result b/mysql-test/main/lock_multi_bug38499.result
index 521ea010816..de552ca0872 100644
--- a/mysql-test/main/lock_multi_bug38499.result
+++ b/mysql-test/main/lock_multi_bug38499.result
@@ -17,7 +17,7 @@ INSERT INTO t2 VALUES (1, 1), (2, 2), (3, 3), (4, 4);
 connection default;
 # 1.2.2. PS mode
 connection default;
-ALTER TABLE t1 ADD COLUMN a INT;
+ALTER TABLE t1 ADD COLUMN a INT, LOCK=SHARED;
 # 2. test UNIONs
 # 2.1. test altering of columns that multiupdate doesn't use
 # 2.1.1. normal mode
diff --git a/mysql-test/main/lock_multi_bug38499.test b/mysql-test/main/lock_multi_bug38499.test
index c489712e5d8..9122b453eb2 100644
--- a/mysql-test/main/lock_multi_bug38499.test
+++ b/mysql-test/main/lock_multi_bug38499.test
@@ -36,8 +36,8 @@ while ($i) {
   send UPDATE t1, (SELECT 1 FROM t2 t1i) d SET a = 0 WHERE 1=0;
 
 --connection locker
-  ALTER TABLE t1 ADD COLUMN (c INT);
-  ALTER TABLE t1 DROP COLUMN c;
+  ALTER TABLE t1 ADD COLUMN (c INT), LOCK=SHARED;
+  ALTER TABLE t1 DROP COLUMN c, LOCK=SHARED;
 
 --connection writer
 --reap
@@ -56,8 +56,8 @@ while ($i) {
 --send EXECUTE stmt
 
 --connection locker
-  ALTER TABLE t1 ADD COLUMN (c INT);
-  ALTER TABLE t1 DROP COLUMN c;
+  ALTER TABLE t1 ADD COLUMN (c INT), LOCK=SHARED;
+  ALTER TABLE t1 DROP COLUMN c, LOCK=SHARED;
 
 --connection writer
 --reap
@@ -76,7 +76,7 @@ while ($i) {
 
 --connection locker
 --error 0,ER_DUP_FIELDNAME
-  ALTER TABLE t1 ADD COLUMN a int(11) unsigned default NULL;
+  ALTER TABLE t1 ADD COLUMN a int(11) unsigned default NULL, LOCK=SHARED;
   UPDATE t1 SET a=b;
 
 --connection writer
@@ -84,7 +84,7 @@ while ($i) {
 
 --connection locker
 --error 0,ER_CANT_DROP_FIELD_OR_KEY
-  ALTER TABLE t1 DROP COLUMN a;
+  ALTER TABLE t1 DROP COLUMN a, LOCK=SHARED;
 
 --connection writer
 --error 0,ER_BAD_FIELD_ERROR # unknown column error
@@ -101,7 +101,7 @@ while ($i) {
 
 --connection locker
 --error 0,ER_DUP_FIELDNAME
-  ALTER TABLE t1 ADD COLUMN a INT;
+  ALTER TABLE t1 ADD COLUMN a INT, LOCK=SHARED;
   UPDATE t1 SET a=b;
 
 --connection writer
@@ -110,7 +110,7 @@ while ($i) {
 
 --connection locker
 --error 0,ER_CANT_DROP_FIELD_OR_KEY
-  ALTER TABLE t1 DROP COLUMN a;
+  ALTER TABLE t1 DROP COLUMN a, LOCK=SHARED;
 
 --connection writer
 --error 0,ER_BAD_FIELD_ERROR # Unknown column 'a' in 'field list'
@@ -118,7 +118,7 @@ while ($i) {
 }
 --enable_query_log
 --connection default
-ALTER TABLE t1 ADD COLUMN a INT;
+ALTER TABLE t1 ADD COLUMN a INT, LOCK=SHARED;
 
 --echo # 2. test UNIONs
 --echo # 2.1. test altering of columns that multiupdate doesn't use
@@ -133,8 +133,8 @@ while ($i) {
   send UPDATE t1, ((SELECT 1 FROM t1 t1i) UNION (SELECT 2 FROM t1 t1ii)) e SET a = 0 WHERE 1=0;
 
 --connection locker
-  ALTER TABLE t1 ADD COLUMN (c INT);
-  ALTER TABLE t1 DROP COLUMN c;
+  ALTER TABLE t1 ADD COLUMN (c INT), LOCK=SHARED;
+  ALTER TABLE t1 DROP COLUMN c, LOCK=SHARED;
 
 --connection writer
 --reap
@@ -153,8 +153,8 @@ while ($i) {
 --send EXECUTE stmt
 
 --connection locker
-  ALTER TABLE t1 ADD COLUMN (c INT);
-  ALTER TABLE t1 DROP COLUMN c;
+  ALTER TABLE t1 ADD COLUMN (c INT), LOCK=SHARED;
+  ALTER TABLE t1 DROP COLUMN c, LOCK=SHARED;
 
 --connection writer
 --reap
@@ -173,7 +173,7 @@ while ($i) {
 
 --connection locker
 --error 0,ER_DUP_FIELDNAME
-  ALTER TABLE t1 ADD COLUMN a int(11) unsigned default NULL;
+  ALTER TABLE t1 ADD COLUMN a int(11) unsigned default NULL, LOCK=SHARED;
   UPDATE t1 SET a=b;
 
 --connection writer
@@ -181,7 +181,7 @@ while ($i) {
 
 --connection locker
 --error 0,ER_CANT_DROP_FIELD_OR_KEY
-  ALTER TABLE t1 DROP COLUMN a;
+  ALTER TABLE t1 DROP COLUMN a, LOCK=SHARED;
 
 --connection writer
 --error 0,ER_BAD_FIELD_ERROR # Unknown column 'a' in 'field list'
@@ -198,7 +198,7 @@ while ($i) {
 
 --connection locker
 --error 0,ER_DUP_FIELDNAME
-  ALTER TABLE t1 ADD COLUMN a INT;
+  ALTER TABLE t1 ADD COLUMN a INT, LOCK=SHARED;
   UPDATE t1 SET a=b;
 
 --connection writer
@@ -207,7 +207,7 @@ while ($i) {
 
 --connection locker
 --error 0,ER_CANT_DROP_FIELD_OR_KEY
-  ALTER TABLE t1 DROP COLUMN a;
+  ALTER TABLE t1 DROP COLUMN a, LOCK=SHARED;
 
 --connection writer
 --error 0,ER_BAD_FIELD_ERROR # Unknown column 'a' in 'field list'
diff --git a/mysql-test/main/lock_multi_bug38691.test b/mysql-test/main/lock_multi_bug38691.test
index 9760c1a873a..62ff3a5bc3b 100644
--- a/mysql-test/main/lock_multi_bug38691.test
+++ b/mysql-test/main/lock_multi_bug38691.test
@@ -48,8 +48,8 @@ while ($i) {
          SET a = NULL WHERE t1.b <> t2.b;
 
 --connection locker
-  ALTER TABLE t2 ADD COLUMN (c INT);
-  ALTER TABLE t2 DROP COLUMN c;
+  ALTER TABLE t2 ADD COLUMN (c INT), LOCK=SHARED;
+  ALTER TABLE t2 DROP COLUMN c, LOCK=SHARED;
 
 --connection writer
 --reap
@@ -69,8 +69,8 @@ while ($i) {
 --send EXECUTE stmt
 
 --connection locker
-  ALTER TABLE t2 ADD COLUMN (c INT);
-  ALTER TABLE t2 DROP COLUMN c;
+  ALTER TABLE t2 ADD COLUMN (c INT), LOCK=SHARED;
+  ALTER TABLE t2 DROP COLUMN c, LOCK=SHARED;
 
 --connection writer
 --reap
@@ -91,7 +91,7 @@ while ($i) {
 
 --connection locker
 --error 0,ER_DUP_FIELDNAME
-  ALTER TABLE t2 ADD COLUMN a int(11) unsigned default NULL;
+  ALTER TABLE t2 ADD COLUMN a int(11) unsigned default NULL, LOCK=SHARED;
   UPDATE t2 SET a=b;
 
 --connection writer
@@ -99,7 +99,7 @@ while ($i) {
 
 --connection locker
 --error 0,ER_CANT_DROP_FIELD_OR_KEY
-  ALTER TABLE t2 DROP COLUMN a;
+  ALTER TABLE t2 DROP COLUMN a, LOCK=SHARED;
 
 --connection writer
 --error 0,ER_BAD_FIELD_ERROR
@@ -116,7 +116,7 @@ while ($i) {
 
 --connection locker
 --error 0,ER_DUP_FIELDNAME
-  ALTER TABLE t2 ADD COLUMN a int(11) unsigned default NULL;
+  ALTER TABLE t2 ADD COLUMN a int(11) unsigned default NULL, LOCK=SHARED;
   UPDATE t2 SET a=b;
 
 --connection writer
@@ -125,7 +125,7 @@ while ($i) {
 
 --connection locker
 --error 0,ER_CANT_DROP_FIELD_OR_KEY
-  ALTER TABLE t2 DROP COLUMN a;
+  ALTER TABLE t2 DROP COLUMN a, LOCK=SHARED;
 
 --connection writer
 --error 0,ER_BAD_FIELD_ERROR
diff --git a/mysql-test/main/mdl_sync.result b/mysql-test/main/mdl_sync.result
index 0ffe2f745a0..939e6a01ed2 100644
--- a/mysql-test/main/mdl_sync.result
+++ b/mysql-test/main/mdl_sync.result
@@ -954,7 +954,7 @@ insert into t1 values (1);
 connection default;
 # Add pending SNW lock.
 # Sending:
-alter table t1 add primary key (c1);;
+alter table t1 add primary key (c1), lock=shared;;
 # 
 connection mdl_con1;
 # Check that ALTER TABLE is waiting with pending SNW lock.
@@ -1964,7 +1964,7 @@ create table t1 (i int);
 # Ensure that ALTER waits once it has acquired SNW lock.
 set debug_sync='alter_table_copy_after_lock_upgrade SIGNAL parked1 WAIT_FOR go1';
 # Sending:
-alter table t1 add column j int;
+alter table t1 add column j int, lock=shared;
 #
 connection deadlock_con1;
 # Wait till ALTER acquires SNW lock and stops.
@@ -2330,7 +2330,7 @@ c1	c2	c3
 #
 connection con46273;
 set debug_sync='alter_table_copy_after_lock_upgrade SIGNAL alter_table_locked WAIT_FOR alter_go';
-alter table t1 add column e int, rename to t2;;
+alter table t1 add column e int, rename to t2, lock=shared;;
 #
 connection default;
 set debug_sync='now WAIT_FOR alter_table_locked';
@@ -2586,7 +2586,7 @@ connection default;
 # table lock and get blocked on sync point.
 set debug_sync= 'alter_table_copy_after_lock_upgrade SIGNAL parked WAIT_FOR go';
 # Sending:
-alter table t1 add column j int;
+alter table t1 add column j int, lock=shared;
 connection con1;
 # Wait until ALTER TABLE gets blocked on a sync point.
 set debug_sync= 'now WAIT_FOR parked';
diff --git a/mysql-test/main/mdl_sync.test b/mysql-test/main/mdl_sync.test
index 3df19aca806..ca91528882e 100644
--- a/mysql-test/main/mdl_sync.test
+++ b/mysql-test/main/mdl_sync.test
@@ -1196,14 +1196,14 @@ insert into t1 values (1);
 connection default;
 --echo # Add pending SNW lock.
 --echo # Sending:
---send alter table t1 add primary key (c1);
+--send alter table t1 add primary key (c1), lock=shared;
 --echo # 
 connection mdl_con1;
 --echo # Check that ALTER TABLE is waiting with pending SNW lock.
 let $wait_condition=
 select count(*) = 1 from information_schema.processlist
 where state = "Waiting for table metadata lock" and
-      info = "alter table t1 add primary key (c1)";
+      info = "alter table t1 add primary key (c1), lock=shared";
 --source include/wait_condition.inc
 --echo # Check that S, SH and SR locks are compatible with pending SNW
 handler t1 open t;
@@ -2431,7 +2431,7 @@ create table t1 (i int);
 --echo # Ensure that ALTER waits once it has acquired SNW lock.
 set debug_sync='alter_table_copy_after_lock_upgrade SIGNAL parked1 WAIT_FOR go1';
 --echo # Sending:
---send alter table t1 add column j int
+--send alter table t1 add column j int, lock=shared
 
 --echo #
 connection deadlock_con1;
@@ -2986,7 +2986,7 @@ select * from t1 where c2 = 3;
 --echo #
 connection con46273;
 set debug_sync='alter_table_copy_after_lock_upgrade SIGNAL alter_table_locked WAIT_FOR alter_go';
---send alter table t1 add column e int, rename to t2;
+--send alter table t1 add column e int, rename to t2, lock=shared;
 
 --echo #
 connection default;
@@ -3333,7 +3333,7 @@ connection default;
 --echo # table lock and get blocked on sync point.
 set debug_sync= 'alter_table_copy_after_lock_upgrade SIGNAL parked WAIT_FOR go';
 --echo # Sending:
---send alter table t1 add column j int
+--send alter table t1 add column j int, lock=shared
 
 connection con1;
 --echo # Wait until ALTER TABLE gets blocked on a sync point.
diff --git a/mysql-test/main/mysql57_virtual.result b/mysql-test/main/mysql57_virtual.result
index 16dd7d55719..68a7b7feb9d 100644
--- a/mysql-test/main/mysql57_virtual.result
+++ b/mysql-test/main/mysql57_virtual.result
@@ -13,7 +13,7 @@ select * from mysql57_virtual;
 a	b	c
 1	2	4
 2	3	5
-alter online table mysql57_virtual comment "I am now a MariaDB table";
+alter online table mysql57_virtual comment "I am now a MariaDB table", algorithm=nocopy;
 ERROR 0A000: LOCK=NONE is not supported. Reason: COPY algorithm requires a lock. Try LOCK=SHARED
 alter table mysql57_virtual comment "I am now a MariaDB table";
 SHOW CREATE TABLE mysql57_virtual;
diff --git a/mysql-test/main/mysql57_virtual.test b/mysql-test/main/mysql57_virtual.test
index 3ebdd894b79..5dfab2bdfc6 100644
--- a/mysql-test/main/mysql57_virtual.test
+++ b/mysql-test/main/mysql57_virtual.test
@@ -14,7 +14,7 @@ select * from mysql57_virtual;
 
 # We can't do online changes, as the MariaDB storage is incompatible with MySQL
 --error ER_ALTER_OPERATION_NOT_SUPPORTED_REASON
-alter online table mysql57_virtual comment "I am now a MariaDB table";
+alter online table mysql57_virtual comment "I am now a MariaDB table", algorithm=nocopy;
 
 alter table mysql57_virtual comment "I am now a MariaDB table";
 SHOW CREATE TABLE mysql57_virtual;
diff --git a/mysql-test/suite/gcol/inc/gcol_select.inc b/mysql-test/suite/gcol/inc/gcol_select.inc
index 4c030cb5646..0b313a6caf0 100644
--- a/mysql-test/suite/gcol/inc/gcol_select.inc
+++ b/mysql-test/suite/gcol/inc/gcol_select.inc
@@ -872,7 +872,9 @@ CREATE TABLE t1(a INT);
 INSERT INTO t1 VALUES(2147483647);
 ALTER TABLE t1 ADD COLUMN h INT AS (a) VIRTUAL;
 ALTER TABLE t1 CHANGE h i INT AS (a) VIRTUAL, ALGORITHM=COPY;
+--error ER_WARN_DATA_OUT_OF_RANGE,ER_ALTER_OPERATION_NOT_SUPPORTED_REASON
 ALTER TABLE t1 ADD COLUMN b SMALLINT AS (a) VIRTUAL, ALGORITHM=COPY, LOCK=NONE;
+--error ER_WARN_DATA_OUT_OF_RANGE,ER_ALTER_OPERATION_NOT_SUPPORTED_REASON
 ALTER TABLE t1 ADD COLUMN e SMALLINT AS (a) VIRTUAL, ALGORITHM=COPY, LOCK=NONE;
 ALTER TABLE t1 ADD COLUMN f SMALLINT AS (a) VIRTUAL, ALGORITHM=COPY, LOCK=SHARED;
 ALTER TABLE t1 ADD COLUMN g SMALLINT AS (a) VIRTUAL, ALGORITHM=COPY, LOCK=EXCLUSIVE;
diff --git a/mysql-test/suite/gcol/r/gcol_select_innodb.result b/mysql-test/suite/gcol/r/gcol_select_innodb.result
index b57aceac9e0..8423c3bcdb6 100644
--- a/mysql-test/suite/gcol/r/gcol_select_innodb.result
+++ b/mysql-test/suite/gcol/r/gcol_select_innodb.result
@@ -536,9 +536,9 @@ INSERT INTO t1 VALUES(2147483647);
 ALTER TABLE t1 ADD COLUMN h INT AS (a) VIRTUAL;
 ALTER TABLE t1 CHANGE h i INT AS (a) VIRTUAL, ALGORITHM=COPY;
 ALTER TABLE t1 ADD COLUMN b SMALLINT AS (a) VIRTUAL, ALGORITHM=COPY, LOCK=NONE;
-ERROR 0A000: LOCK=NONE is not supported. Reason: COPY algorithm requires a lock. Try LOCK=SHARED
+Got one of the listed errors
 ALTER TABLE t1 ADD COLUMN e SMALLINT AS (a) VIRTUAL, ALGORITHM=COPY, LOCK=NONE;
-ERROR 0A000: LOCK=NONE is not supported. Reason: COPY algorithm requires a lock. Try LOCK=SHARED
+Got one of the listed errors
 ALTER TABLE t1 ADD COLUMN f SMALLINT AS (a) VIRTUAL, ALGORITHM=COPY, LOCK=SHARED;
 ERROR 22003: Out of range value for column 'f' at row 1
 ALTER TABLE t1 ADD COLUMN g SMALLINT AS (a) VIRTUAL, ALGORITHM=COPY, LOCK=EXCLUSIVE;
diff --git a/mysql-test/suite/gcol/r/gcol_select_myisam.result b/mysql-test/suite/gcol/r/gcol_select_myisam.result
index 070808fec4f..df1fbd27cb0 100644
--- a/mysql-test/suite/gcol/r/gcol_select_myisam.result
+++ b/mysql-test/suite/gcol/r/gcol_select_myisam.result
@@ -1155,9 +1155,9 @@ INSERT INTO t1 VALUES(2147483647);
 ALTER TABLE t1 ADD COLUMN h INT AS (a) VIRTUAL;
 ALTER TABLE t1 CHANGE h i INT AS (a) VIRTUAL, ALGORITHM=COPY;
 ALTER TABLE t1 ADD COLUMN b SMALLINT AS (a) VIRTUAL, ALGORITHM=COPY, LOCK=NONE;
-ERROR 0A000: LOCK=NONE is not supported. Reason: COPY algorithm requires a lock. Try LOCK=SHARED
+Got one of the listed errors
 ALTER TABLE t1 ADD COLUMN e SMALLINT AS (a) VIRTUAL, ALGORITHM=COPY, LOCK=NONE;
-ERROR 0A000: LOCK=NONE is not supported. Reason: COPY algorithm requires a lock. Try LOCK=SHARED
+Got one of the listed errors
 ALTER TABLE t1 ADD COLUMN f SMALLINT AS (a) VIRTUAL, ALGORITHM=COPY, LOCK=SHARED;
 ERROR 22003: Out of range value for column 'f' at row 1
 ALTER TABLE t1 ADD COLUMN g SMALLINT AS (a) VIRTUAL, ALGORITHM=COPY, LOCK=EXCLUSIVE;
diff --git a/mysql-test/suite/perfschema/r/alter_table_progress.result b/mysql-test/suite/perfschema/r/alter_table_progress.result
index 31cc60927f6..7487e59d205 100644
--- a/mysql-test/suite/perfschema/r/alter_table_progress.result
+++ b/mysql-test/suite/perfschema/r/alter_table_progress.result
@@ -75,6 +75,8 @@ stage/sql/table lock	NULL	NULL
 stage/sql/After create	NULL	NULL
 stage/sql/copy to tmp table	5	5
 stage/sql/Enabling keys	NULL	NULL
+stage/sql/Apply log event	NULL	NULL
+stage/sql/After apply log event	NULL	NULL
 stage/sql/Rename result table	NULL	NULL
 stage/sql/Unlocking tables	NULL	NULL
 stage/sql/Rename result table	NULL	NULL
diff --git a/sql/ha_sequence.cc b/sql/ha_sequence.cc
index b348e6e7025..41e7252e516 100644
--- a/sql/ha_sequence.cc
+++ b/sql/ha_sequence.cc
@@ -285,8 +285,7 @@ int ha_sequence::write_row(const uchar *buf)
       sequence->copy(&tmp_seq);
     rows_changed++;
     /* We have to do the logging while we hold the sequence mutex */
-    if (row_logging)
-      error= binlog_log_row(table, 0, buf, log_func);
+    error= binlog_log_row(0, buf, log_func);
   }
 
   /* Row is already logged, don't log it again in ha_write_row() */
diff --git a/sql/handler.cc b/sql/handler.cc
index aef2f8860e9..b9209179c07 100644
--- a/sql/handler.cc
+++ b/sql/handler.cc
@@ -2979,6 +2979,13 @@ int ha_savepoint(THD *thd, SAVEPOINT *sv)
   Ha_trx_info *ha_info= trans->ha_list;
   DBUG_ENTER("ha_savepoint");
 
+  if (!thd->online_alter_cache_list.empty())
+  {
+    my_printf_error(ER_NOT_SUPPORTED_YET, "Cannot set up a savepoint while "
+                    "online ALTER TABLE is in progress", MYF(0));
+    DBUG_RETURN(1);
+  }
+
   for (; ha_info; ha_info= ha_info->next())
   {
     int err;
@@ -7234,13 +7241,50 @@ bool handler::check_table_binlog_row_based_internal()
 }
 
 
-int handler::binlog_log_row(TABLE *table,
-                            const uchar *before_record,
-                            const uchar *after_record,
-                            Log_func *log_func)
+#ifdef HAVE_REPLICATION
+static int binlog_log_row_online_alter(TABLE* table,
+                                       const uchar *before_record,
+                                       const uchar *after_record,
+                                       Log_func *log_func,
+                                       bool has_trans)
 {
   THD *thd= table->in_use;
-  DBUG_ENTER("binlog_log_row");
+
+  if (!table->online_alter_cache)
+  {
+    auto *cache_mngr= online_alter_binlog_get_cache_mngr(thd, table);
+    // Use transaction cache directly, if it is not multi-transaction mode
+    table->online_alter_cache= binlog_get_cache_data(cache_mngr,
+                                        !thd->in_multi_stmt_transaction_mode());
+
+    trans_register_ha(thd, false, binlog_hton, 0);
+    if (thd->in_multi_stmt_transaction_mode())
+      trans_register_ha(thd, true, binlog_hton, 0);
+  }
+
+  // We need to log all columns for the case if alter table changes primary key.
+  table->use_all_columns();
+  bitmap_set_all(table->rpl_write_set);
+
+  int error= (*log_func)(thd, table, table->s->online_alter_binlog,
+                         table->online_alter_cache, has_trans,
+                         before_record, after_record);
+
+  return unlikely(error) ? HA_ERR_RBR_LOGGING_FAILED : 0;
+}
+#endif // HAVE_REPLICATION
+
+
+static int binlog_log_row_to_binlog(TABLE* table,
+                                    const uchar *before_record,
+                                    const uchar *after_record,
+                                    Log_func *log_func,
+                                    bool has_trans)
+{
+  bool error= 0;
+  THD *const thd= table->in_use;
+
+  DBUG_ENTER("binlog_log_row_to_binlog");
 
   if (!thd->binlog_table_maps &&
       thd->binlog_write_table_maps())
@@ -7254,18 +7298,38 @@ int handler::binlog_log_row(TABLE *table,
   if (cache_mngr == NULL)
     DBUG_RETURN(HA_ERR_OUT_OF_MEM);
 
-  bool is_trans= row_logging_has_trans;
   /* Ensure that all events in a GTID group are in the same cache */
   if (thd->variables.option_bits & OPTION_GTID_BEGIN)
-    is_trans= 1;
+    has_trans= 1;
 
-  auto *cache= binlog_get_cache_data(cache_mngr, use_trans_cache(thd, is_trans));
+  auto *cache= binlog_get_cache_data(cache_mngr,
+                                     use_trans_cache(thd, has_trans));
 
-  bool error= (*log_func)(thd, table, mysql_bin_log.as_event_log(), cache,
-                          is_trans, before_record, after_record);
+    error= (*log_func)(thd, table, mysql_bin_log.as_event_log(), cache,
+                       has_trans, before_record, after_record);
   DBUG_RETURN(error ? HA_ERR_RBR_LOGGING_FAILED : 0);
 }
 
+int handler::binlog_log_row(const uchar *before_record,
+                            const uchar *after_record,
+                            Log_func *log_func)
+{
+  DBUG_ENTER("handler::binlog_log_row");
+
+  int error = 0;
+  if (row_logging)
+    error= binlog_log_row_to_binlog(table, before_record, after_record,
+                                    log_func, row_logging_has_trans);
+
+#ifdef HAVE_REPLICATION
+  if (unlikely(!error && table->s->online_alter_binlog))
+    error= binlog_log_row_online_alter(table, before_record, after_record,
+                                       log_func, row_logging_has_trans);
+#endif // HAVE_REPLICATION
+
+  DBUG_RETURN(error);
+}
+
 
 int handler::ha_external_lock(THD *thd, int lock_type)
 {
@@ -7806,11 +7870,9 @@ int handler::ha_write_row(const uchar *buf)
   if (likely(!error))
   {
     rows_changed++;
-    if (row_logging)
-    {
-      Log_func *log_func= Write_rows_log_event::binlog_row_logging_function;
-      error= binlog_log_row(table, 0, buf, log_func);
-    }
+    Log_func *log_func= Write_rows_log_event::binlog_row_logging_function;
+    error= binlog_log_row(0, buf, log_func);
+
 #ifdef WITH_WSREP
     if (WSREP_NNULL(ha_thd()) && table_share->tmp_table == NO_TMP_TABLE &&
         ht->flags & HTON_WSREP_REPLICATION &&
@@ -7866,11 +7928,9 @@ int handler::ha_update_row(const uchar *old_data, const uchar *new_data)
   if (likely(!error))
   {
     rows_changed++;
-    if (row_logging)
-    {
-      Log_func *log_func= Update_rows_log_event::binlog_row_logging_function;
-      error= binlog_log_row(table, old_data, new_data, log_func);
-    }
+    Log_func *log_func= Update_rows_log_event::binlog_row_logging_function;
+    error= binlog_log_row(old_data, new_data, log_func);
+
 #ifdef WITH_WSREP
     THD *thd= ha_thd();
     if (WSREP_NNULL(thd))
@@ -7944,11 +8004,9 @@ int handler::ha_delete_row(const uchar *buf)
   if (likely(!error))
   {
     rows_changed++;
-    if (row_logging)
-    {
-      Log_func *log_func= Delete_rows_log_event::binlog_row_logging_function;
-      error= binlog_log_row(table, buf, 0, log_func);
-    }
+    Log_func *log_func= Delete_rows_log_event::binlog_row_logging_function;
+    error= binlog_log_row(buf, 0, log_func);
+
 #ifdef WITH_WSREP
     THD *thd= ha_thd();
     if (WSREP_NNULL(thd))
diff --git a/sql/handler.h b/sql/handler.h
index 0ddcb06030d..896ac1aa7cc 100644
--- a/sql/handler.h
+++ b/sql/handler.h
@@ -653,6 +653,7 @@ given at all. */
 typedef ulonglong alter_table_operations;
 
 class Event_log;
+class Cache_flip_event_log;
 class binlog_cache_data;
 typedef bool Log_func(THD*, TABLE*, Event_log *, binlog_cache_data *, bool,
                       const uchar*, const uchar*);
@@ -3506,6 +3507,7 @@ public:
   /** to be actually called to get 'check()' functionality*/
   int ha_check(THD *thd, HA_CHECK_OPT *check_opt);
   int ha_repair(THD* thd, HA_CHECK_OPT* check_opt);
+  virtual void open_read_view(){}
   void ha_start_bulk_insert(ha_rows rows, uint flags= 0)
   {
     DBUG_ENTER("handler::ha_start_bulk_insert");
@@ -4986,8 +4988,7 @@ public:
   bool check_table_binlog_row_based();
   bool prepare_for_row_logging();
   int prepare_for_insert(bool do_create);
-  int binlog_log_row(TABLE *table,
-                     const uchar *before_record,
+  int binlog_log_row(const uchar *before_record,
                      const uchar *after_record,
                      Log_func *log_func);
 
diff --git a/sql/log.cc b/sql/log.cc
index b709fa5dee9..45593bb3fd3 100644
--- a/sql/log.cc
+++ b/sql/log.cc
@@ -45,6 +45,7 @@
 #include <m_ctype.h>				// For test_if_number
 
 #include <set_var.h> // for Sys_last_gtid_ptr
+#include <ilist.h>
 
 #ifdef _WIN32
 #include "message.h"
@@ -103,6 +104,9 @@ static int binlog_flush_cache(THD *thd, binlog_cache_mngr *cache_mngr,
                               Log_event *end_ev, bool all, bool using_stmt,
                               bool using_trx, bool is_ro_1pc);
 
+int binlog_online_alter_commit(THD *thd, bool all);
+void binlog_online_alter_rollback(THD *thd, bool all);
+
 static const LEX_CSTRING write_error_msg=
     { STRING_WITH_LEN("error writing to the binary log") };
 
@@ -508,15 +512,16 @@ void Log_event_writer::set_incident()
 }
 
 
-class binlog_cache_mngr {
+class binlog_cache_mngr: public ilist_node<> {
 public:
   binlog_cache_mngr(my_off_t param_max_binlog_stmt_cache_size,
                     my_off_t param_max_binlog_cache_size,
                     ulong *param_ptr_binlog_stmt_cache_use,
                     ulong *param_ptr_binlog_stmt_cache_disk_use,
                     ulong *param_ptr_binlog_cache_use,
-                    ulong *param_ptr_binlog_cache_disk_use)
-    : last_commit_pos_offset(0), using_xa(FALSE), xa_xid(0)
+                    ulong *param_ptr_binlog_cache_disk_use,
+                    TABLE_SHARE *share)
+    : last_commit_pos_offset(0), using_xa(FALSE), xa_xid(0), share(share)
   {
      stmt_cache.set_binlog_cache_info(param_max_binlog_stmt_cache_size,
                                       param_ptr_binlog_stmt_cache_use,
@@ -578,9 +583,12 @@ public:
   ulong binlog_id;
   /* Set if we get an error during commit that must be returned from unlog(). */
   bool delayed_error;
+
   //Will be reset when gtid is written into binlog
   uchar  gtid_flags3;
   decltype (rpl_gtid::seq_no) sa_seq_no;
+
+  TABLE_SHARE *share;
 private:
 
   binlog_cache_mngr& operator=(const binlog_cache_mngr& info);
@@ -2249,6 +2257,31 @@ static int binlog_commit_flush_xa_prepare(THD *thd, bool all,
   return (binlog_flush_cache(thd, cache_mngr, &end_evt, all, TRUE, TRUE));
 }
 
+#ifdef HAVE_REPLICATION
+static void
+binlog_online_alter_cleanup(ilist<binlog_cache_mngr> &list,
+                            bool ending_trans)
+{
+  for (auto &cache: list)
+  {
+    cache.reset(true, ending_trans);
+  }
+  if (ending_trans)
+  {
+    auto it= list.begin();
+    while (it != list.end())
+    {
+      auto &cache= *it;
+      it++;
+      list.remove(cache);
+      cache.~binlog_cache_mngr();
+      my_free(&cache);
+    }
+    DBUG_ASSERT(list.empty());
+  }
+}
+#endif // HAVE_REPLICATION
+
 /**
   This function is called once after each statement.
 
@@ -2265,8 +2298,15 @@ int binlog_commit(THD *thd, bool all, bool ro_1pc)
   PSI_stage_info org_stage;
   DBUG_ENTER("binlog_commit");
 
-  binlog_cache_mngr *const cache_mngr= thd->binlog_get_cache_mngr();
+  bool is_ending_transaction= ending_trans(thd, all);
+  error= binlog_online_alter_commit(thd, all);
+  if (error)
+    DBUG_RETURN(error);
 
+  binlog_cache_mngr *const cache_mngr= thd->binlog_get_cache_mngr();
+  /*
+    cache_mngr can be NULL in case if binlog logging is disabled.
+  */
   if (!cache_mngr)
   {
     DBUG_ASSERT(WSREP(thd) ||
@@ -2326,7 +2366,7 @@ int binlog_commit(THD *thd, bool all, bool ro_1pc)
      - We are in a transaction and a full transaction is committed.
     Otherwise, we accumulate the changes.
   */
-  if (likely(!error) && ending_trans(thd, all))
+  if (likely(!error) && is_ending_transaction)
   {
     bool is_xa_prepare= is_preparing_xa(thd);
 
@@ -2366,12 +2406,17 @@ static int binlog_rollback(handlerton *hton, THD *thd, bool all)
 {
   DBUG_ENTER("binlog_rollback");
 
+  bool is_ending_trans= ending_trans(thd, all);
+
+  bool rollback_online= !thd->online_alter_cache_list.empty();
+  if (rollback_online)
+    binlog_online_alter_rollback(thd, all);
   int error= 0;
   binlog_cache_mngr *const cache_mngr= thd->binlog_get_cache_mngr();
 
   if (!cache_mngr)
   {
-    DBUG_ASSERT(WSREP(thd));
+    DBUG_ASSERT(WSREP(thd) || rollback_online);
     DBUG_ASSERT(thd->lex->sql_command != SQLCOM_XA_ROLLBACK);
 
     DBUG_RETURN(0);
@@ -2425,7 +2470,7 @@ static int binlog_rollback(handlerton *hton, THD *thd, bool all)
   }
   else if (likely(!error))
   {  
-    if (ending_trans(thd, all) && trans_cannot_safely_rollback(thd, all))
+    if (is_ending_trans && trans_cannot_safely_rollback(thd, all))
       error= binlog_rollback_flush_trx_cache(thd, all, cache_mngr);
     /*
       Truncate the cache if:
@@ -2437,7 +2482,7 @@ static int binlog_rollback(handlerton *hton, THD *thd, bool all)
         . the format is not MIXED or no temporary non-trans table
           was updated.
     */
-    else if (ending_trans(thd, all) ||
+    else if (is_ending_trans ||
              (!(thd->transaction->stmt.has_created_dropped_temp_table() &&
                 !thd->is_current_stmt_binlog_format_row()) &&
               (!stmt_has_updated_non_trans_table(thd) ||
@@ -2597,6 +2642,9 @@ static int binlog_savepoint_rollback(handlerton *hton, THD *thd, void *sv)
 {
   DBUG_ENTER("binlog_savepoint_rollback");
 
+  if (!thd->online_alter_cache_list.empty())
+    binlog_online_alter_rollback(thd, !thd->in_sub_stmt);
+
   /*
     Write ROLLBACK TO SAVEPOINT to the binlog cache if we have updated some
     non-transactional table. Otherwise, truncate the binlog cache starting
@@ -3071,13 +3119,14 @@ void MYSQL_LOG::close(uint exiting)
   {
     end_io_cache(&log_file);
 
-    if (log_type == LOG_BIN && mysql_file_sync(log_file.file, MYF(MY_WME)) && ! write_error)
+    if (log_type == LOG_BIN && log_file.file >= 0 &&
+        mysql_file_sync(log_file.file, MYF(MY_WME)) && ! write_error)
     {
       write_error= 1;
       sql_print_error(ER_DEFAULT(ER_ERROR_ON_WRITE), name, errno);
     }
 
-    if (!(exiting & LOG_CLOSE_DELAYED_CLOSE) &&
+    if (!(exiting & LOG_CLOSE_DELAYED_CLOSE) && log_file.file >= 0 &&
         mysql_file_close(log_file.file, MYF(MY_WME)) && ! write_error)
     {
       write_error= 1;
@@ -3547,7 +3596,7 @@ MYSQL_BIN_LOG::MYSQL_BIN_LOG(uint *sync_period)
     We don't want to initialize locks here as such initialization depends on
     safe_mutex (when using safe_mutex) which depends on MY_INIT(), which is
     called only in main(). Doing initialization here would make it happen
-    before main().
+    before main(). init_pthread_objects() can be called for that purpose.
   */
   index_file_name[0] = 0;
   bzero((char*) &index_file, sizeof(index_file));
@@ -3725,28 +3774,46 @@ bool MYSQL_BIN_LOG::open_index_file(const char *index_file_name_arg,
 }
 
 
-bool Event_log::open(const char *log_name, enum_log_type log_type,
+bool Event_log::open(const char *log_name,
                      const char *new_name, ulong next_file_number,
                      enum cache_type io_cache_type_arg)
 {
-  bool error= MYSQL_LOG::open(
+  bool error= false;
+  if (log_name || new_name)
+  {
+    error= MYSQL_LOG::open(
 #ifdef HAVE_PSI_INTERFACE
           0,
 #endif
-          log_name, log_type, new_name, next_file_number, io_cache_type_arg);
+          log_name, LOG_NORMAL, new_name, next_file_number, io_cache_type_arg);
+  }
+  else
+  {
+#ifdef HAVE_PSI_INTERFACE
+    /* Keep the key for reopen */
+    m_log_file_key= 0;
+#endif
+    error= init_io_cache(&log_file, -1, LOG_BIN_IO_SIZE,
+                         io_cache_type_arg, 0, 0,
+                         MYF(MY_WME | MY_NABP | MY_WAIT_IF_FULL));
+
+    log_state= LOG_OPENED;
+    inited= true;
+  }
   if (error)
     return error;
 
-  int bytes_written= write_description_event(
+  longlong bytes_written= write_description_event(
                               (enum_binlog_checksum_alg)binlog_checksum_options,
                               encrypt_binlog, false, false);
   error= bytes_written < 0;
   return error;
 }
 
-int Event_log::write_description_event(enum_binlog_checksum_alg checksum_alg,
-                                       bool encrypt, bool dont_set_created,
-                                       bool is_relay_log)
+longlong
+Event_log::write_description_event(enum_binlog_checksum_alg checksum_alg,
+                                   bool encrypt, bool dont_set_created,
+                                   bool is_relay_log)
 {
   Format_description_log_event s(BINLOG_VERSION);
   /*
@@ -3791,7 +3858,7 @@ int Event_log::write_description_event(enum_binlog_checksum_alg checksum_alg,
         return -1;
     }
   }
-  return s.data_written;
+  return (longlong)s.data_written;
 }
 
 
@@ -4308,7 +4375,8 @@ int MYSQL_BIN_LOG::find_log_pos(LOG_INFO *linfo, const char *log_name,
                        log_name ? log_name : "NULL", full_log_name));
 
   /* As the file is flushed, we can't get an error here */
-  (void) reinit_io_cache(&index_file, READ_CACHE, (my_off_t) 0, 0, 0);
+  error= reinit_io_cache(&index_file, READ_CACHE, (my_off_t) 0, 0, 0);
+  DBUG_ASSERT(!error);
 
   for (;;)
   {
@@ -5911,17 +5979,11 @@ bool stmt_has_updated_non_trans_table(const THD* thd)
   binlog_hton, which has internal linkage.
 */
 
-binlog_cache_mngr *THD::binlog_setup_trx_data()
+binlog_cache_mngr *binlog_setup_cache_mngr(TABLE_SHARE *share)
 {
-  DBUG_ENTER("THD::binlog_setup_trx_data");
-  binlog_cache_mngr *cache_mngr=
-    (binlog_cache_mngr*) thd_get_ha_data(this, binlog_hton);
-
-  if (cache_mngr)
-    DBUG_RETURN(cache_mngr);                             // Already set up
-
-  cache_mngr= (binlog_cache_mngr*) my_malloc(key_memory_binlog_cache_mngr,
-                                  sizeof(binlog_cache_mngr), MYF(MY_ZEROFILL));
+  auto *cache_mngr= (binlog_cache_mngr*) my_malloc(key_memory_binlog_cache_mngr,
+                                  sizeof(binlog_cache_mngr),
+                                                   MYF(MY_ZEROFILL));
   if (!cache_mngr ||
       open_cached_file(&cache_mngr->stmt_cache.cache_log, mysql_tmpdir,
                        LOG_PREFIX, (size_t)binlog_stmt_cache_size, MYF(MY_WME)) ||
@@ -5929,17 +5991,34 @@ binlog_cache_mngr *THD::binlog_setup_trx_data()
                        LOG_PREFIX, (size_t)binlog_cache_size, MYF(MY_WME)))
   {
     my_free(cache_mngr);
-    DBUG_RETURN(0);                      // Didn't manage to set it up
+    return NULL;
   }
-  thd_set_ha_data(this, binlog_hton, cache_mngr);
 
   cache_mngr= new (cache_mngr)
-              binlog_cache_mngr(max_binlog_stmt_cache_size,
-                                max_binlog_cache_size,
-                                &binlog_stmt_cache_use,
-                                &binlog_stmt_cache_disk_use,
-                                &binlog_cache_use,
-                                &binlog_cache_disk_use);
+          binlog_cache_mngr(max_binlog_stmt_cache_size,
+                            max_binlog_cache_size,
+                            &binlog_stmt_cache_use,
+                            &binlog_stmt_cache_disk_use,
+                            &binlog_cache_use,
+                            &binlog_cache_disk_use,
+                            share);
+
+  return cache_mngr;
+}
+
+binlog_cache_mngr *THD::binlog_setup_trx_data()
+{
+  DBUG_ENTER("THD::binlog_setup_trx_data");
+  binlog_cache_mngr *cache_mngr=
+    (binlog_cache_mngr*) thd_get_ha_data(this, binlog_hton);
+
+  if (!cache_mngr)
+  {
+    cache_mngr= binlog_setup_cache_mngr(NULL);
+    thd_set_ha_data(this, binlog_hton, cache_mngr);
+  }
+
+
   DBUG_RETURN(cache_mngr);
 }
 
@@ -6301,6 +6380,23 @@ write_err:
 }
 
 
+binlog_cache_mngr *online_alter_binlog_get_cache_mngr(THD *thd, TABLE *table)
+{
+  ilist<binlog_cache_mngr> &list= thd->online_alter_cache_list;
+
+  /* we assume it's very rare to have more than one online ALTER running */
+  for (auto &cache: list)
+  {
+    if (cache.share == table->s)
+      return &cache;
+  }
+
+  auto *new_cache_mngr= binlog_setup_cache_mngr(table->s);
+  list.push_back(*new_cache_mngr);
+
+  return new_cache_mngr;
+}
+
 binlog_cache_mngr *THD::binlog_get_cache_mngr() const
 {
   return (binlog_cache_mngr*) thd_get_ha_data(this, binlog_hton);
@@ -6402,7 +6498,7 @@ Event_log::flush_and_set_pending_rows_event(THD *thd,
                                                 bool is_transactional)
 {
   DBUG_ENTER("MYSQL_BIN_LOG::flush_and_set_pending_rows_event(event)");
-  DBUG_ASSERT(WSREP_EMULATE_BINLOG(thd) || mysql_bin_log.is_open());
+  DBUG_ASSERT(WSREP_EMULATE_BINLOG(thd) || is_open());
   DBUG_PRINT("enter", ("event: %p", event));
 
   DBUG_PRINT("info", ("cache_mngr->pending(): %p", cache_data->pending()));
@@ -7562,6 +7658,105 @@ private:
   bool first;
 };
 
+int cache_copy(IO_CACHE *to, IO_CACHE *from)
+{
+  DBUG_ENTER("cache_copy");
+  if (reinit_io_cache(from, READ_CACHE, 0, 0, 0))
+    DBUG_RETURN(ER_ERROR_ON_WRITE);
+  size_t bytes_in_cache= my_b_bytes_in_cache(from);
+
+  do
+  {
+    my_b_write(to, from->read_pos, bytes_in_cache);
+
+    from->read_pos += bytes_in_cache;
+    bytes_in_cache= my_b_fill(from);
+    if (from->error || to->error)
+      DBUG_RETURN(ER_ERROR_ON_WRITE);
+  } while (bytes_in_cache);
+
+  DBUG_RETURN(0);
+}
+
+int binlog_online_alter_commit(THD *thd, bool all)
+{
+  DBUG_ENTER("online_alter_commit");
+  int error= 0;
+#ifdef HAVE_REPLICATION
+
+  if (thd->online_alter_cache_list.empty())
+    DBUG_RETURN(0);
+
+  bool is_ending_transaction= ending_trans(thd, all);
+
+  for (auto &cache_mngr: thd->online_alter_cache_list)
+  {
+    auto *binlog= cache_mngr.share->online_alter_binlog;
+    DBUG_ASSERT(binlog);
+
+    error= binlog_flush_pending_rows_event(thd,
+                                           /*
+                                             do not set STMT_END for last event
+                                             to leave table open in altering thd
+                                           */
+                                           false,
+                                           true,
+                                           binlog,
+                                           is_ending_transaction
+                                           ? &cache_mngr.trx_cache
+                                           : &cache_mngr.stmt_cache);
+    if (is_ending_transaction)
+    {
+      mysql_mutex_lock(binlog->get_log_lock());
+      error= binlog->write_cache(thd, &cache_mngr.trx_cache.cache_log);
+
+      mysql_mutex_unlock(binlog->get_log_lock());
+    }
+    else
+    {
+      error= cache_copy(&cache_mngr.trx_cache.cache_log,
+                        &cache_mngr.stmt_cache.cache_log);
+    }
+
+    if (error)
+    {
+      my_error(ER_ERROR_ON_WRITE, MYF(ME_ERROR_LOG),
+               binlog->get_name(), errno);
+      binlog_online_alter_cleanup(thd->online_alter_cache_list,
+                                  is_ending_transaction);
+      DBUG_RETURN(error);
+    }
+  }
+
+  binlog_online_alter_cleanup(thd->online_alter_cache_list,
+                              is_ending_transaction);
+
+  for (TABLE *table= thd->open_tables; table; table= table->next)
+  {
+    table->online_alter_cache= NULL;
+  }
+#endif // HAVE_REPLICATION
+  DBUG_RETURN(error);
+}
+
+void binlog_online_alter_rollback(THD *thd, bool all)
+{
+#ifdef HAVE_REPLICATION
+  bool is_ending_trans= ending_trans(thd, all);
+
+  /*
+    This is a crucial moment that we are running through
+    thd->online_alter_cache_list, and not through thd->open_tables to cleanup
+    stmt cache, though both have it. The reason is that the tables can be closed
+    to that moment in case of an error.
+    The same reason applies to the fact we don't store cache_mngr in the table
+    itself -- because it can happen to be not existing.
+    Still in case if tables are left opened
+   */
+  binlog_online_alter_cleanup(thd->online_alter_cache_list, is_ending_trans);
+#endif // HAVE_REPLICATION
+}
+
 /*
   Write the contents of a cache to the binary log.
 
@@ -7590,7 +7785,7 @@ int Event_log::write_cache(THD *thd, IO_CACHE *cache)
   size_t val;
   size_t end_log_pos_inc= 0; // each event processed adds BINLOG_CHECKSUM_LEN 2 t
   uchar header[LOG_EVENT_HEADER_LEN];
-  CacheWriter writer(thd, &log_file, binlog_checksum_options, &crypto);
+  CacheWriter writer(thd, get_log_file(), binlog_checksum_options, &crypto);
 
   if (crypto.scheme)
   {
@@ -7615,7 +7810,7 @@ int Event_log::write_cache(THD *thd, IO_CACHE *cache)
     split.
   */
 
-  group= (size_t)my_b_tell(&log_file);
+  group= (size_t)my_b_tell(get_log_file());
   hdr_offs= carry= 0;
 
   do
diff --git a/sql/log.h b/sql/log.h
index b84ccb6d791..ac7bbbedaf2 100644
--- a/sql/log.h
+++ b/sql/log.h
@@ -421,18 +421,85 @@ public:
 
   bool open(
           const char *log_name,
-          enum_log_type log_type,
           const char *new_name, ulong next_file_number,
           enum cache_type io_cache_type_arg);
-  IO_CACHE *get_log_file() { return &log_file; }
+  virtual IO_CACHE *get_log_file() { return &log_file; }
 
-  int write_description_event(enum_binlog_checksum_alg checksum_alg,
-                              bool encrypt, bool dont_set_created,
-                              bool is_relay_log);
+  longlong write_description_event(enum_binlog_checksum_alg checksum_alg,
+                                   bool encrypt, bool dont_set_created,
+                                   bool is_relay_log);
 
   bool write_event(Log_event *ev, binlog_cache_data *data, IO_CACHE *file);
 };
 
+/**
+  A single-reader, single-writer non-blocking layer for Event_log.
+  Provides IO_CACHE for writing and IO_CACHE for reading.
+
+  Writers use an overrided get_log_file version for their writes, while readers
+  should use flip() to initiate reading.
+  flip() swaps pointers to allow non-blocking reads.
+
+  Writers can block other writers and a reader with a mutex, but a reader only
+  swaps two pointers under a lock, so it won't block writers.
+
+  TODO should be unnecessary after MDEV-24676 is done
+ */
+class Cache_flip_event_log: public Event_log {
+  IO_CACHE alt_buf;
+  IO_CACHE *current, *alt;
+public:
+
+  Cache_flip_event_log() : Event_log(), alt_buf{},
+                           current(&log_file), alt(&alt_buf) {}
+  bool open(enum cache_type io_cache_type_arg)
+  {
+    log_file.dir= mysql_tmpdir;
+    alt_buf.dir= log_file.dir;
+    bool res= Event_log::open(NULL, NULL, 0, io_cache_type_arg);
+    if (res)
+      return res;
+
+    name= my_strdup(key_memory_MYSQL_LOG_name, "online-alter-binlog",
+                    MYF(MY_WME));
+    if (!name)
+      return false;
+
+    res= init_io_cache(&alt_buf, -1, LOG_BIN_IO_SIZE, io_cache_type_arg, 0, 0,
+                       MYF(MY_WME | MY_NABP | MY_WAIT_IF_FULL)) != 0;
+    return res;
+  }
+
+  /**
+    Swaps current and alt_log. Can be called only from the reader thread.
+    @return a new IO_CACHE pointer to read from.
+   */
+  IO_CACHE *flip()
+  {
+    IO_CACHE *tmp= current;
+    reinit_io_cache(alt, WRITE_CACHE, 0, 0, 0);
+    mysql_mutex_lock(get_log_lock());
+    reinit_io_cache(current, READ_CACHE, 0, 0, 0);
+    current= alt;
+    mysql_mutex_unlock(get_log_lock());
+    alt= tmp;
+
+    return alt;
+  }
+
+  IO_CACHE *get_log_file() override
+  {
+    mysql_mutex_assert_owner(get_log_lock());
+    return current;
+  }
+
+  void cleanup()
+  {
+    end_io_cache(&alt_buf);
+    Event_log::cleanup();
+  }
+};
+
 /* Tell the io thread if we can delay the master info sync. */
 #define SEMI_SYNC_SLAVE_DELAY_SYNC 1
 /* Tell the io thread if the current event needs a ack. */
@@ -641,7 +708,6 @@ class MYSQL_BIN_LOG: public TC_LOG, private Event_log
     new_file() is locking. new_file_without_locking() does not acquire
     LOCK_log.
   */
-  int new_file_without_locking();
   int new_file_impl();
   void do_checkpoint_request(ulong binlog_id);
   void purge();
@@ -651,6 +717,7 @@ class MYSQL_BIN_LOG: public TC_LOG, private Event_log
   void trx_group_commit_leader(group_commit_entry *leader);
   bool is_xidlist_idle_nolock();
 public:
+  int new_file_without_locking();
   /*
     A list of struct xid_count_per_binlog is used to keep track of how many
     XIDs are in prepared, but not committed, state in each binlog. And how
@@ -1256,6 +1323,7 @@ int binlog_flush_pending_rows_event(THD *thd, bool stmt_end,
                                     binlog_cache_data *cache_data);
 Rows_log_event* binlog_get_pending_rows_event(binlog_cache_mngr *cache_mngr,
                                               bool use_trans_cache);
+binlog_cache_mngr *online_alter_binlog_get_cache_mngr(THD *thd, TABLE *table);
 binlog_cache_data* binlog_get_cache_data(binlog_cache_mngr *cache_mngr,
                                          bool use_trans_cache);
 
@@ -1350,5 +1418,4 @@ int binlog_rollback_by_xid(handlerton *hton, XID *xid);
 bool write_bin_log_start_alter(THD *thd, bool& partial_alter,
                                uint64 start_alter_id, bool log_if_exists);
 
-
 #endif /* LOG_H */
diff --git a/sql/log_event.h b/sql/log_event.h
index ad8ddb3deac..ee91a103287 100644
--- a/sql/log_event.h
+++ b/sql/log_event.h
@@ -3742,6 +3742,8 @@ private:
   bool  m_used_query_txt;
 };
 
+class table_def;
+
 /**
   @class Table_map_log_event
 
@@ -4393,6 +4395,7 @@ public:
   virtual bool print(FILE *file, PRINT_EVENT_INFO *print_event_info);
 #endif
 
+  table_def get_table_def();
 
 private:
 #if defined(MYSQL_SERVER) && defined(HAVE_REPLICATION)
@@ -4697,7 +4700,7 @@ protected:
 #endif
   ulonglong       m_table_id;	/* Table ID */
   MY_BITMAP   m_cols;		/* Bitmap denoting columns available */
-  ulong       m_width;          /* The width of the columns bitmap */
+  uint        m_width;          /* The width of the columns bitmap */
   /*
     Bitmap for columns available in the after image, if present. These
     fields are only available for Update_rows events. Observe that the
@@ -4732,6 +4735,7 @@ protected:
 
   /* helper functions */
 
+
 #if defined(MYSQL_SERVER) && defined(HAVE_REPLICATION)
   const uchar *m_curr_row;     /* Start of the row being processed */
   const uchar *m_curr_row_end; /* One-after the end of the current row */
diff --git a/sql/log_event_server.cc b/sql/log_event_server.cc
index 87d42b72181..026b5f0342b 100644
--- a/sql/log_event_server.cc
+++ b/sql/log_event_server.cc
@@ -5218,7 +5218,8 @@ int Rows_log_event::do_apply_event(rpl_group_info *rgi)
 
 err:
   restore_empty_query_table_list(thd->lex);
-  rgi->slave_close_thread_tables(thd);
+  if (rgi->tables_to_lock_count)
+    rgi->slave_close_thread_tables(thd);
   thd->reset_query_timer();
   DBUG_RETURN(error);
 }
@@ -5786,6 +5787,13 @@ check_table_map(rpl_group_info *rgi, RPL_TABLE_LIST *table_list)
   DBUG_RETURN(res);
 }
 
+table_def Table_map_log_event::get_table_def()
+{
+  return table_def(m_coltype, m_colcnt,
+                   m_field_metadata, m_field_metadata_size,
+                   m_null_bits, m_flags);
+}
+
 int Table_map_log_event::do_apply_event(rpl_group_info *rgi)
 {
   RPL_TABLE_LIST *table_list;
@@ -5824,16 +5832,26 @@ int Table_map_log_event::do_apply_event(rpl_group_info *rgi)
   LEX_CSTRING tmp_db_name=  {db_mem, db_mem_length };
   LEX_CSTRING tmp_tbl_name= {tname_mem, tname_mem_length };
 
-  table_list->init_one_table(&tmp_db_name, &tmp_tbl_name, 0, TL_WRITE);
-  table_list->table_id= DBUG_IF("inject_tblmap_same_id_maps_diff_table") ? 0 : m_table_id;
-  table_list->updating= 1;
+  /*
+    The memory allocated by the table_def structure (i.e., not the
+    memory allocated *for* the table_def structure) is released
+    inside rpl_group_info::clear_tables_to_lock() by calling the
+    table_def destructor explicitly.
+  */
+  new(table_list) RPL_TABLE_LIST(&tmp_db_name, &tmp_tbl_name, TL_WRITE,
+                                 get_table_def(),
+                                 m_flags & TM_BIT_HAS_TRIGGERS_F);
+
+  table_list->table_id= DBUG_IF("inject_tblmap_same_id_maps_diff_table") ?
+                                         0: m_table_id;
   table_list->required_type= TABLE_TYPE_NORMAL;
+  table_list->open_type= OT_BASE_ONLY;
+  DBUG_ASSERT(table_list->updating);
 
   DBUG_PRINT("debug", ("table: %s is mapped to %llu",
                        table_list->table_name.str,
                        table_list->table_id));
-  table_list->master_had_triggers= ((m_flags & TM_BIT_HAS_TRIGGERS_F) ? 1 : 0);
-  DBUG_PRINT("debug", ("table->master_had_triggers=%d", 
+  DBUG_PRINT("debug", ("table->master_had_triggers=%d",
                        (int)table_list->master_had_triggers));
 
   enum_tbl_map_status tblmap_status= check_table_map(rgi, table_list);
@@ -5842,23 +5860,6 @@ int Table_map_log_event::do_apply_event(rpl_group_info *rgi)
     DBUG_ASSERT(thd->lex->query_tables != table_list);
 
     /*
-      Use placement new to construct the table_def instance in the
-      memory allocated for it inside table_list.
-
-      The memory allocated by the table_def structure (i.e., not the
-      memory allocated *for* the table_def structure) is released
-      inside Relay_log_info::clear_tables_to_lock() by calling the
-      table_def destructor explicitly.
-    */
-    new (&table_list->m_tabledef)
-      table_def(m_coltype, m_colcnt,
-                m_field_metadata, m_field_metadata_size,
-                m_null_bits, m_flags);
-    table_list->m_tabledef_valid= TRUE;
-    table_list->m_conv_table= NULL;
-    table_list->open_type= OT_BASE_ONLY;
-
-    /*
       We record in the slave's information that the table should be
       locked by linking the table into the list of tables to lock.
     */
@@ -5902,8 +5903,9 @@ int Table_map_log_event::do_apply_event(rpl_group_info *rgi)
           execute in a user session 
          */
         my_error(ER_SLAVE_FATAL_ERROR, MYF(0), buf);
-    } 
-    
+    }
+
+    table_list->~RPL_TABLE_LIST();
     my_free(memory);
   }
 
diff --git a/sql/rpl_record.cc b/sql/rpl_record.cc
index 1a9c02a167b..cb76c66797f 100644
--- a/sql/rpl_record.cc
+++ b/sql/rpl_record.cc
@@ -229,7 +229,10 @@ unpack_row(rpl_group_info *rgi,
   uint i= 0;
   table_def *tabledef= NULL;
   TABLE *conv_table= NULL;
-  bool table_found= rgi && rgi->get_table_data(table, &tabledef, &conv_table);
+  const Copy_field *copy_fields;
+  const Copy_field *copy_fields_end;
+  bool table_found= rgi && rgi->get_table_data(table, &tabledef, &conv_table,
+                                               &copy_fields, &copy_fields_end);
   DBUG_PRINT("debug", ("Table data: table_found: %d, tabldef: %p, conv_table: %p",
                        table_found, tabledef, conv_table));
   DBUG_ASSERT(table_found);
@@ -243,7 +246,7 @@ unpack_row(rpl_group_info *rgi,
   if (rgi && !table_found)
     DBUG_RETURN(HA_ERR_GENERIC);
 
-  for (field_ptr= begin_ptr ; field_ptr < end_ptr && *field_ptr ; ++field_ptr)
+  for (field_ptr= begin_ptr; field_ptr < end_ptr && *field_ptr; ++field_ptr)
   {
     /*
       If there is a conversion table, we pick up the field pointer to
@@ -346,8 +349,11 @@ unpack_row(rpl_group_info *rgi,
         case, we have unpacked the master data to the conversion
         table, so we need to copy the value stored in the conversion
         table into the final table and do the conversion at the same time.
+
+        If copy_fields is set, it means we are doing an online alter table,
+        and will use copy_fields set up in copy_data_between_tables
       */
-      if (conv_field)
+      if (conv_field && !copy_fields)
       {
         Copy_field copy;
 #ifndef DBUG_OFF
@@ -379,6 +385,14 @@ unpack_row(rpl_group_info *rgi,
     i++;
   }
 
+  if (copy_fields)
+  {
+    for (auto *copy=copy_fields; copy != copy_fields_end; copy++)
+    {
+      copy->do_copy(copy);
+    }
+  }
+
   /*
     throw away master's extra fields
   */
diff --git a/sql/rpl_rli.h b/sql/rpl_rli.h
index 0fd9070426b..6565ac55a86 100644
--- a/sql/rpl_rli.h
+++ b/sql/rpl_rli.h
@@ -941,14 +941,20 @@ struct rpl_group_info
     }
   }
 
-  bool get_table_data(TABLE *table_arg, table_def **tabledef_var, TABLE **conv_table_var) const
+  bool get_table_data(TABLE *table_arg, table_def **tabledef_var,
+          TABLE **conv_table_var,
+          const Copy_field *copy[], const Copy_field **copy_end) const
   {
     DBUG_ASSERT(tabledef_var && conv_table_var);
     for (TABLE_LIST *ptr= tables_to_lock ; ptr != NULL ; ptr= ptr->next_global)
       if (ptr->table == table_arg)
       {
-        *tabledef_var= &static_cast<RPL_TABLE_LIST*>(ptr)->m_tabledef;
-        *conv_table_var= static_cast<RPL_TABLE_LIST*>(ptr)->m_conv_table;
+        auto *rpl_table_list= static_cast<RPL_TABLE_LIST*>(ptr);
+        if (rpl_table_list->m_tabledef_valid)
+          *tabledef_var= &rpl_table_list->m_tabledef;
+        *conv_table_var= rpl_table_list->m_conv_table;
+        *copy= rpl_table_list->m_online_alter_copy_fields;
+        *copy_end= rpl_table_list->m_online_alter_copy_fields_end;
         DBUG_PRINT("debug", ("Fetching table data for table %s.%s:"
                              " tabledef: %p, conv_table: %p",
                              table_arg->s->db.str, table_arg->s->table_name.str,
diff --git a/sql/rpl_utility.h b/sql/rpl_utility.h
index ec74fd26dc0..bc3a2a9a40f 100644
--- a/sql/rpl_utility.h
+++ b/sql/rpl_utility.h
@@ -43,6 +43,7 @@ struct rpl_group_info;
 
 class table_def
 {
+  table_def(const table_def&) = default;
 public:
   /**
     Constructor.
@@ -56,6 +57,18 @@ public:
   table_def(unsigned char *types, ulong size, uchar *field_metadata,
             int metadata_size, uchar *null_bitmap, uint16 flags);
 
+
+  /**
+    Move constructor
+    Since it deallocates a memory during destruction, we can't safely copy it.
+    We should instead move it to zero m_memory in an old object
+   */
+  table_def(table_def &&tabledef)
+  : table_def(tabledef)
+  {
+    tabledef.m_memory= NULL;
+  }
+
   ~table_def();
 
   /**
@@ -243,6 +256,30 @@ struct RPL_TABLE_LIST
   table_def m_tabledef;
   TABLE *m_conv_table;
   bool master_had_triggers;
+  const Copy_field *m_online_alter_copy_fields;
+  const Copy_field *m_online_alter_copy_fields_end;
+
+  RPL_TABLE_LIST(const LEX_CSTRING *db_arg, const LEX_CSTRING *table_name_arg,
+                 thr_lock_type thr_lock_type,
+                 table_def &&tabledef, bool master_had_trigers)
+    : TABLE_LIST(db_arg, table_name_arg, NULL, thr_lock_type),
+      m_tabledef_valid(true), m_tabledef(std::move(tabledef)),
+      m_conv_table(NULL), master_had_triggers(master_had_trigers),
+      m_online_alter_copy_fields(NULL),
+      m_online_alter_copy_fields_end(NULL)
+  {}
+
+  RPL_TABLE_LIST(TABLE *table, thr_lock_type lock_type, TABLE *conv_table,
+                 table_def &&tabledef,
+                 const Copy_field online_alter_copy_fields[],
+                 const Copy_field *online_alter_copy_fields_end)
+    :  TABLE_LIST(table, lock_type),
+       m_tabledef_valid(true),
+       m_tabledef(std::move(tabledef)),
+       m_conv_table(conv_table), master_had_triggers(false),
+       m_online_alter_copy_fields(online_alter_copy_fields),
+       m_online_alter_copy_fields_end(online_alter_copy_fields_end)
+  {}
 };
 
 
diff --git a/sql/sql_class.cc b/sql/sql_class.cc
index 645802aa2ea..7bc17e17c2c 100644
--- a/sql/sql_class.cc
+++ b/sql/sql_class.cc
@@ -7133,8 +7133,8 @@ int THD::binlog_delete_row(TABLE* table, Event_log *bin_log,
                 Rows_event_factory::get<Delete_rows_compressed_log_event>() :
                 Rows_event_factory::get<Delete_rows_log_event>();
   auto *ev= bin_log->prepare_pending_rows_event(this, table, cache_data,
-                                                     variables.server_id,
-                                                     len, is_trans, creator);
+                                                variables.server_id,
+                                                len, is_trans, creator);
 
   if (unlikely(ev == 0))
     return HA_ERR_OUT_OF_MEM;
diff --git a/sql/sql_class.h b/sql/sql_class.h
index a454334965f..8bee34ee8c0 100644
--- a/sql/sql_class.h
+++ b/sql/sql_class.h
@@ -70,6 +70,7 @@ void set_thd_stage_info(void *thd,
 #include "wsrep_on.h"
 #ifdef WITH_WSREP
 #include <inttypes.h>
+#include <ilist.h>
 /* wsrep-lib */
 #include "wsrep_client_service.h"
 #include "wsrep_client_state.h"
@@ -5546,6 +5547,8 @@ public:
   Item *sp_prepare_func_item(Item **it_addr, uint cols);
   bool sp_eval_expr(Field *result_field, Item **expr_item_ptr);
 
+  ilist<binlog_cache_mngr> online_alter_cache_list;
+
   bool sql_parser(LEX *old_lex, LEX *lex,
                   char *str, uint str_len, bool stmt_prepare_mode);
 
diff --git a/sql/sql_table.cc b/sql/sql_table.cc
index e2264568cf7..33c46ce68ad 100644
--- a/sql/sql_table.cc
+++ b/sql/sql_table.cc
@@ -55,6 +55,7 @@
 #include "sql_audit.h"
 #include "sql_sequence.h"
 #include "tztime.h"
+#include "rpl_rli.h"
 #include "sql_insert.h"                // binlog_drop_table
 #include "ddl_log.h"
 #include "debug.h"                     // debug_crash_here()
@@ -85,7 +86,7 @@ static int copy_data_between_tables(THD *, TABLE *,TABLE *,
                                     List<Create_field> &, bool, uint, ORDER *,
                                     ha_rows *, ha_rows *,
                                     Alter_info::enum_enable_or_disable,
-                                    Alter_table_ctx *);
+                                    Alter_table_ctx *, bool);
 static int append_system_key_parts(THD *thd, HA_CREATE_INFO *create_info,
                                    Key *key);
 static int mysql_prepare_create_table(THD *, HA_CREATE_INFO *, Alter_info *,
@@ -2000,7 +2001,7 @@ bool log_drop_table(THD *thd, const LEX_CSTRING *db_name,
       in the binary log. We log this for non temporary tables, as the slave
       may use a filter to ignore queries for a specific database.
     */
-    error= thd->binlog_query(THD::STMT_QUERY_TYPE,
+    error= thd->binlog_query(THD::STMT_QUERY_TYPE, 
                              query.ptr(), query.length(),
                              FALSE, FALSE, temporary_table, 0) > 0;
   }
@@ -4322,7 +4323,7 @@ handler *mysql_create_frm_image(THD *thd, const LEX_CSTRING &db,
     {
       if (key->type == Key::FOREIGN_KEY)
       {
-        my_error(ER_FEATURE_NOT_SUPPORTED_WITH_PARTITIONING, MYF(0), 
+        my_error(ER_FEATURE_NOT_SUPPORTED_WITH_PARTITIONING, MYF(0),
                  "FOREIGN KEY");
         goto err;
       }
@@ -9936,6 +9937,11 @@ bool mysql_alter_table(THD *thd, const LEX_CSTRING *new_db,
   MDL_request target_mdl_request;
   MDL_ticket *mdl_ticket= 0;
   Alter_table_prelocking_strategy alter_prelocking_strategy;
+#ifdef HAVE_REPLICATION
+  bool online= order == NULL && !opt_bootstrap;
+#else
+  bool online= false;
+#endif
   TRIGGER_RENAME_PARAM trigger_param;
 
   /*
@@ -10017,6 +10023,19 @@ bool mysql_alter_table(THD *thd, const LEX_CSTRING *new_db,
     has been already processed.
   */
   table_list->required_type= TABLE_TYPE_NORMAL;
+  
+  
+  if (alter_info->requested_lock == Alter_info::ALTER_TABLE_LOCK_SHARED
+      || alter_info->requested_lock == Alter_info::ALTER_TABLE_LOCK_EXCLUSIVE
+      || thd->locked_tables_mode == LTM_LOCK_TABLES
+      || thd->lex->sql_command == SQLCOM_OPTIMIZE
+      || alter_info->algorithm(thd) == Alter_info::ALTER_TABLE_ALGORITHM_NOCOPY)
+    online= false;
+
+  if (online)
+  {
+    table_list->lock_type= TL_READ;
+  }
 
   DEBUG_SYNC(thd, "alter_table_before_open_tables");
 
@@ -10048,6 +10067,8 @@ bool mysql_alter_table(THD *thd, const LEX_CSTRING *new_db,
 
   table= table_list->table;
   bool is_reg_table= table->s->tmp_table == NO_TMP_TABLE;
+  
+  online= online && !table->s->tmp_table;
 
 #ifdef WITH_WSREP
   if (WSREP(thd) &&
@@ -10875,7 +10896,8 @@ do_continue:;
   if (!table->s->tmp_table)
   {
     // COPY algorithm doesn't work with concurrent writes.
-    if (alter_info->requested_lock == Alter_info::ALTER_TABLE_LOCK_NONE)
+    if (!online &&
+        alter_info->requested_lock == Alter_info::ALTER_TABLE_LOCK_NONE)
     {
       my_error(ER_ALTER_OPERATION_NOT_SUPPORTED_REASON, MYF(0),
                "LOCK=NONE",
@@ -11018,7 +11040,7 @@ do_continue:;
                                  alter_info->create_list, ignore,
                                  order_num, order, &copied, &deleted,
                                  alter_info->keys_onoff,
-                                 &alter_ctx))
+                                 &alter_ctx, online))
       goto err_new_table_cleanup;
   }
   else
@@ -11512,6 +11534,58 @@ bool mysql_trans_commit_alter_copy_data(THD *thd)
   DBUG_RETURN(error);
 }
 
+#ifdef HAVE_REPLICATION
+static int online_alter_read_from_binlog(THD *thd, rpl_group_info *rgi,
+                                         Cache_flip_event_log *log)
+{
+  MEM_ROOT event_mem_root;
+  Query_arena backup_arena;
+  Query_arena event_arena(&event_mem_root, Query_arena::STMT_INITIALIZED);
+  init_sql_alloc(key_memory_gdl, &event_mem_root,
+                 MEM_ROOT_BLOCK_SIZE, 0, MYF(0));
+
+  int error= 0;
+
+  IO_CACHE *log_file= log->flip();
+
+  thd_progress_report(thd, 0, my_b_write_tell(log_file));
+
+  Abort_on_warning_instant_set old_abort_on_warning(thd, 0);
+  do
+  {
+    const auto *descr_event= rgi->rli->relay_log.description_event_for_exec;
+    auto *ev= Log_event::read_log_event(log_file, descr_event, false);
+    if (!ev)
+      break;
+
+    ev->thd= thd;
+    thd->set_n_backup_active_arena(&event_arena, &backup_arena);
+    error= ev->apply_event(rgi);
+    thd->restore_active_arena(&event_arena, &backup_arena);
+
+    event_arena.free_items();
+    free_root(&event_mem_root, MYF(MY_KEEP_PREALLOC));
+    if (ev != rgi->rli->relay_log.description_event_for_exec)
+      delete ev;
+    thd_progress_report(thd, my_b_tell(log_file), thd->progress.max_counter);
+    DEBUG_SYNC(thd, "alter_table_online_progress");
+  } while(!error);
+
+  return error;
+}
+#endif // HAVE_REPLICATION
+
+static void online_alter_cleanup_binlog(THD *thd, TABLE_SHARE *s)
+{
+#ifdef HAVE_REPLICATION
+  if (!s->online_alter_binlog)
+    return;
+  // s->online_alter_binlog->reset_logs(thd, false, NULL, 0, 0);
+  s->online_alter_binlog->cleanup();
+  s->online_alter_binlog->~Cache_flip_event_log();
+  s->online_alter_binlog= NULL;
+#endif
+}
 
 static int
 copy_data_between_tables(THD *thd, TABLE *from, TABLE *to,
@@ -11519,7 +11593,7 @@ copy_data_between_tables(THD *thd, TABLE *from, TABLE *to,
 			 uint order_num, ORDER *order,
 			 ha_rows *copied, ha_rows *deleted,
                          Alter_info::enum_enable_or_disable keys_onoff,
-                         Alter_table_ctx *alter_ctx)
+                         Alter_table_ctx *alter_ctx, bool online)
 {
   int error= 1;
   Copy_field *copy= NULL, *copy_end;
@@ -11544,14 +11618,64 @@ copy_data_between_tables(THD *thd, TABLE *from, TABLE *to,
   MYSQL_TIME query_start;
   DBUG_ENTER("copy_data_between_tables");
 
-  /* Two or 3 stages; Sorting, copying data and update indexes */
-  thd_progress_init(thd, 2 + MY_TEST(order));
+  /*
+    if ORDER BY: sorting
+    always: copying, building indexes.
+    if online: reading up the binlog (second binlog is being written)
+               reading up the second binlog under exclusive lock
+  */
+  thd_progress_init(thd, MY_TEST(order) + 2 + 2 * MY_TEST(online));
+
+#ifdef HAVE_REPLICATION
+  if (online)
+  {
+    void *buf= alloc_root(thd->mem_root, sizeof (Cache_flip_event_log));
+
+   from->s->online_alter_binlog= new (buf) Cache_flip_event_log();
+    if (!from->s->online_alter_binlog)
+      DBUG_RETURN(1);
+
+    from->s->online_alter_binlog->init_pthread_objects();
+
+    error= from->s->online_alter_binlog->open(WRITE_CACHE);
+
+    DBUG_ASSERT(!error);
+
+    if (!error)
+    {
+      /*
+        Some engines (for example, InnoDB) might not create a read view
+        until the first row is read. We need to be sure that we won't see any
+        table changes after we enable replication and downgrade the MDL.
+        So, we force the consistent snapshot to be created now.
+      */
+      handlerton *ht= from->s->db_type();
+      if (ht->start_consistent_snapshot)
+      {
+        thd->tx_isolation= ISO_REPEATABLE_READ;
+        from->file->open_read_view();
+      }
+    }
+
+    if (error)
+    {
+      online_alter_cleanup_binlog(thd, from->s);
+      DBUG_RETURN(1);
+    }
+
+    from->mdl_ticket->downgrade_lock(MDL_SHARED_UPGRADABLE);
+    DEBUG_SYNC(thd, "alter_table_online_downgraded");
+  }
+#else
+  DBUG_ASSERT(!online);
+#endif // HAVE_REPLICATION
 
   if (!(copy= new (thd->mem_root) Copy_field[to->s->fields]))
     DBUG_RETURN(-1);
 
   if (mysql_trans_prepare_alter_copy_data(thd))
   {
+    online_alter_cleanup_binlog(thd, from->s);
     delete [] copy;
     DBUG_RETURN(-1);
   }
@@ -11559,6 +11683,7 @@ copy_data_between_tables(THD *thd, TABLE *from, TABLE *to,
   /* We need external lock before we can disable/enable keys */
   if (to->file->ha_external_lock(thd, F_WRLCK))
   {
+    online_alter_cleanup_binlog(thd, from->s);
     /* Undo call to mysql_trans_prepare_alter_copy_data() */
     ha_enable_transaction(thd, TRUE);
     delete [] copy;
@@ -11585,6 +11710,7 @@ copy_data_between_tables(THD *thd, TABLE *from, TABLE *to,
   Create_field *def;
   copy_end=copy;
   to->s->default_fields= 0;
+  error= 1;
   for (Field **ptr=to->field ; *ptr ; ptr++)
   {
     def=it++;
@@ -11703,6 +11829,8 @@ copy_data_between_tables(THD *thd, TABLE *from, TABLE *to,
   if (!ignore) /* for now, InnoDB needs the undo log for ALTER IGNORE */
     to->file->extra(HA_EXTRA_BEGIN_ALTER_COPY);
 
+  DEBUG_SYNC(thd, "alter_table_copy_start");
+
   while (likely(!(error= info.read_record())))
   {
     if (unlikely(thd->killed))
@@ -11837,14 +11965,11 @@ copy_data_between_tables(THD *thd, TABLE *from, TABLE *to,
     thd->get_stmt_da()->inc_current_row_for_warning();
   }
 
+  DEBUG_SYNC(thd, "alter_table_copy_end");
+
   THD_STAGE_INFO(thd, stage_enabling_keys);
   thd_progress_next_stage(thd);
 
-  if (error > 0 && !from->s->tmp_table)
-  {
-    /* We are going to drop the temporary table */
-    to->file->extra(HA_EXTRA_PREPARE_FOR_DROP);
-  }
   if (unlikely(to->file->ha_end_bulk_insert()) && error <= 0)
   {
     /* Give error, if not already given */
@@ -11852,6 +11977,7 @@ copy_data_between_tables(THD *thd, TABLE *from, TABLE *to,
       to->file->print_error(my_errno,MYF(0));
     error= 1;
   }
+
   bulk_insert_started= 0;
   if (!ignore)
     to->file->extra(HA_EXTRA_END_ALTER_COPY);
@@ -11859,6 +11985,76 @@ copy_data_between_tables(THD *thd, TABLE *from, TABLE *to,
   cleanup_done= 1;
   to->file->extra(HA_EXTRA_NO_IGNORE_DUP_KEY);
 
+#ifdef HAVE_REPLICATION
+  if (likely(online && error < 0))
+  {
+    Ha_trx_info *trx_info_save= thd->transaction->all.ha_list;
+    thd->transaction->all.ha_list = NULL;
+    thd_progress_next_stage(thd);
+    Table_map_log_event table_event(thd, from, from->s->table_map_id,
+                                    from->file->has_transactions());
+    Relay_log_info rli(false);
+    rpl_group_info rgi(&rli);
+    RPL_TABLE_LIST rpl_table(to, TL_WRITE, from, table_event.get_table_def(),
+                             copy, copy_end);
+    Cache_flip_event_log *binlog= from->s->online_alter_binlog;
+    rgi.thd= thd;
+    rgi.tables_to_lock= &rpl_table;
+
+    rgi.m_table_map.set_table(from->s->table_map_id, to);
+
+    DBUG_ASSERT(binlog->is_open());
+
+    rli.relay_log.description_event_for_exec=
+                                            new Format_description_log_event(4);
+
+    // We restore bitmaps, because update event is going to mess up with them.
+    to->default_column_bitmaps();
+
+    error= online_alter_read_from_binlog(thd, &rgi, binlog);
+
+    DEBUG_SYNC(thd, "alter_table_online_before_lock");
+
+    int lock_error=
+        thd->mdl_context.upgrade_shared_lock(from->mdl_ticket, MDL_EXCLUSIVE,
+                                     (double)thd->variables.lock_wait_timeout);
+    if (!error)
+      error= lock_error;
+
+    if (!error)
+    {
+      thd_progress_next_stage(thd);
+      error= online_alter_read_from_binlog(thd, &rgi, binlog);
+    }
+
+    thd->transaction->all.ha_list = trx_info_save;
+  }
+  else if (unlikely(online)) // error was on copy stage
+  {
+    /*
+       We need to issue a barrier to clean up gracefully.
+       Without this, following possible:
+       T1: ALTER TABLE starts
+       T2: INSERT starts
+       T1: ALTER TABLE fails with error (i.e. ER_DUP_KEY)
+       T1: from->s->online_alter_binlog sets to NULL
+       T2: INSERT committs
+       T2: thd->online_alter_cache_list is not empty
+       T2: binlog_commit: DBUG_ASSERT(binlog); is issued.
+    */
+    // Ignore the return result. We already have an error.
+    thd->mdl_context.upgrade_shared_lock(from->mdl_ticket, 
+                                         MDL_SHARED_NO_WRITE,
+                                         thd->variables.lock_wait_timeout);
+  }
+#endif
+
+  if (error > 0 && !from->s->tmp_table)
+  {
+    /* We are going to drop the temporary table */
+    to->file->extra(HA_EXTRA_PREPARE_FOR_DROP);
+  }
+
   DEBUG_SYNC(thd, "copy_data_between_tables_before_reset_backup_lock");
   if (backup_reset_alter_copy_lock(thd))
     error= 1;
@@ -11871,6 +12067,8 @@ copy_data_between_tables(THD *thd, TABLE *from, TABLE *to,
     (void) to->file->ha_end_bulk_insert();
 
 /* Free resources */
+  online_alter_cleanup_binlog(thd, from->s);
+
   if (init_read_record_done)
     end_read_record(&info);
   delete [] copy;
diff --git a/sql/table.h b/sql/table.h
index d9c1231db6a..0f654b8f161 100644
--- a/sql/table.h
+++ b/sql/table.h
@@ -66,6 +66,7 @@ struct TABLE_LIST;
 class ACL_internal_schema_access;
 class ACL_internal_table_access;
 class Field;
+class Copy_field;
 class Table_statistics;
 class With_element;
 struct TDC_element;
@@ -79,6 +80,7 @@ class Pushdown_derived;
 struct Name_resolution_context;
 class Table_function_json_table;
 class Open_table_context;
+class MYSQL_LOG;
 
 /*
   Used to identify NESTED_JOIN structures within a join (applicable only to
@@ -919,6 +921,10 @@ struct TABLE_SHARE
   plugin_ref default_part_plugin;
 #endif
 
+#ifdef HAVE_REPLICATION
+  Cache_flip_event_log *online_alter_binlog;
+#endif
+
   /**
     System versioning and application-time periods support.
   */
@@ -1611,6 +1617,8 @@ public:
   */
   Item *notnull_cond;
 
+  binlog_cache_data *online_alter_cache;
+
   inline void reset() { bzero((void*)this, sizeof(*this)); }
   void init(THD *thd, TABLE_LIST *tl);
   bool fill_item_list(List<Item> *item_list) const;
@@ -2339,11 +2347,18 @@ struct TABLE_LIST
                      mdl_type, MDL_TRANSACTION);
   }
 
+  TABLE_LIST(const LEX_CSTRING *db_arg,
+             const LEX_CSTRING *table_name_arg,
+             const LEX_CSTRING *alias_arg,
+             enum thr_lock_type lock_type_arg)
+  {
+    init_one_table(db_arg, table_name_arg, alias_arg, lock_type_arg);
+  }
+
   TABLE_LIST(TABLE *table_arg, thr_lock_type lock_type)
+    : TABLE_LIST(&table_arg->s->db, &table_arg->s->table_name, NULL, lock_type)
   {
     DBUG_ASSERT(table_arg->s);
-    init_one_table(&table_arg->s->db, &table_arg->s->table_name,
-                   NULL, lock_type);
     table= table_arg;
     vers_conditions.name= table->s->vers.name;
   }
diff --git a/storage/innobase/handler/ha_innodb.cc b/storage/innobase/handler/ha_innodb.cc
index e171d2caecc..ee97b2a4d61 100644
--- a/storage/innobase/handler/ha_innodb.cc
+++ b/storage/innobase/handler/ha_innodb.cc
@@ -16433,7 +16433,10 @@ ha_innobase::store_lock(
 			    || sql_command == SQLCOM_REPLACE_SELECT
 			    || sql_command == SQLCOM_UPDATE
 			    || sql_command == SQLCOM_CREATE_SEQUENCE
-			    || sql_command == SQLCOM_CREATE_TABLE))) {
+			    || sql_command == SQLCOM_CREATE_TABLE))
+		    || (trx->isolation_level == TRX_ISO_REPEATABLE_READ
+		        && sql_command == SQLCOM_ALTER_TABLE
+		        && lock_type == TL_READ)) {
 
 			/* If the transaction isolation level is
 			READ UNCOMMITTED or READ COMMITTED and we are executing
@@ -21267,3 +21270,17 @@ buf_pool_size_align(
     return (size / m + 1) * m;
   }
 }
+
+void ha_innobase::open_read_view()
+{
+  trx_t *trx= m_prebuilt->trx;
+  auto thd_iso= thd_get_trx_isolation(m_user_thd);
+
+  trx->isolation_level= innobase_map_isolation_level(thd_iso);
+  ut_ad(trx->isolation_level == TRX_ISO_REPEATABLE_READ);
+  ut_ad(!trx_is_started(trx));
+
+  trx_start_if_not_started_xa(trx, false);
+
+  trx->read_view.open(m_prebuilt->trx);
+}
diff --git a/storage/innobase/handler/ha_innodb.h b/storage/innobase/handler/ha_innodb.h
index 60b56b4a22f..643e04c8b91 100644
--- a/storage/innobase/handler/ha_innodb.h
+++ b/storage/innobase/handler/ha_innodb.h
@@ -439,6 +439,7 @@ public:
 			  const KEY_PART_INFO& old_part,
 			  const KEY_PART_INFO& new_part) const override;
 
+        void open_read_view() override;
 protected:
 	bool
 	can_convert_string(const Field_string* field,
diff --git a/tests/mysql_client_test.c b/tests/mysql_client_test.c
index 6a4635e535e..35b1e05d3a5 100644
--- a/tests/mysql_client_test.c
+++ b/tests/mysql_client_test.c
@@ -19125,7 +19125,7 @@ static void test_progress_reporting()
   }
   
   progress_stage= progress_max_stage= progress_count= 0;
-  rc= mysql_query(conn, "alter table t1 add f1 int primary key auto_increment, order by f2");
+  rc= mysql_query(conn, "alter table t1 add f1 int primary key auto_increment, lock=shared, order by f2");
   myquery(rc);
   if (!opt_silent)
     printf("Got progress_count: %u  stage: %u  max_stage: %u\n",
@@ -19133,7 +19133,7 @@ static void test_progress_reporting()
   DIE_UNLESS(progress_count > 0 && progress_stage >=2 && progress_max_stage == 3);
 
   progress_stage= progress_max_stage= progress_count= 0;
-  rc= mysql_query(conn, "create index f2 on t1 (f2)");
+  rc= mysql_query(conn, "create index f2 on t1 (f2) lock=shared");
   myquery(rc);
   if (!opt_silent)
     printf("Got progress_count: %u  stage: %u  max_stage: %u\n",
@@ -19146,7 +19146,7 @@ static void test_progress_reporting()
   if (!opt_silent)
     printf("Got progress_count: %u  stage: %u  max_stage: %u\n",
            progress_count, progress_stage, progress_max_stage);
-  DIE_UNLESS(progress_count > 0 && progress_stage >=2 && progress_max_stage == 2);
+  DIE_UNLESS(progress_count > 0 && progress_stage >=2 && progress_max_stage == 4);
 
   rc= mysql_query(conn, "set @@global.progress_report_time=@save");
   myquery(rc);