MDEV-15250 UPSERT during ALTER-TABLE results in 'Duplicate entry' er

ror for alter

- To prevent the fake duplicates, InnoDB can write online log entries
during transaction commit. By doing this, InnoDB can avoid logging
the rollbacked DML transaction.

- In this patch, InnoDB DDL clears the online log and changes
the online status to COMPLETE in commit_inplace_alter_table.
Basically, it does allow logging even when DDL is waiting
for MDL before commit phase. This could lead to more
DB_ONLINE_LOG_TOO_BIG error.

- Store clust_index->online_log value as table pointer when
secondary indexes are being built. This is a quick way to
indicate whether the table has active DDL

- Remove the old online log code in row_ins_clust_index_entry_low(),
row_ins_scan_sec_index_for_duplicate(), btr_cur_del_mark_set_clust_rec(),
row_purge_*(), row_undo_*()

row_log_insert_handle(): Handles the insert undo log entries and
apply it to the online indexes log.

row_log_update_handle(): Handles the update, delete undo log entries
and apply it on the online indexes log.

row_log_undo_vers_record(): Get the clustered index record version
which was modified by the given undo log record.

row_log_vers_undo_match(): Get the correct version of clustered
index record which matches with undo log record offset and page number

trx_t::apply_online_log: Set to true in trx_undo_report_row_operation
when modified table has active DDL

trx_t::apply_log(): Apply the DML changes to online DDL tables.

dict_table_t::is_active_ddl(): Returns true whether the table has
any active DDL

trx_undo_rec_apply_log(): Parse the undo log and apply to the online
log tables if there is any active DDL exist.

trx_undo_rec_apply_insert(): Apply the TRX_UNDO_INSERT_REC undo log
record

trx_undo_rec_apply_update(): Apply the TRX_UNDO_UPD & TRX_UNDO_DEL
undo log record

trx_undo_rec_is_equal(): Check whether the given roll pointer's
page number and offset matches with the undo log record.

row_log_table_apply_convert_mrec(): Instead of returning
DATA_TRUNCATED error, InnoDB should throw DB_INVALID_NULL
when InnoDB DDL encounters NULL value for NOT NULL column

trx_undo_rec_info: Information about undo log record like
trx id, block, offset, type, cmpl_info, updated_extern,
undo_no, undo_rec, update vector

ha_innobase_inplace_ctx::log_handle_failure(): Handle the apply
log failure for online DDL operation

ha_innobase::column_bitmaps_signal(): Remove table->F_WRLCK
to make sure that indexed virtual column is computed when
table has active ddl

- Many test case were changed to suit the new behaviour.

FIXME:
=====
- InnoDB index rebuild log should be cleaned and moved the
online status to COMPLETE in ha_innobase::inplace_alter_table().
After that, InnoDB should insert the concurrent DML directly to
the online index. By doing this, InnoDB can avoid
the DB_ONLINE_LOG_TOO_BIG error.

Thanks to Marko Mäkelä for providing the initial prototype.

41 files changed, 1051 insertions, 985 deletions

diff --git a/mysql-test/suite/gcol/r/innodb_virtual_debug.result b/mysql-test/suite/gcol/r/innodb_virtual_debug.result
index 806cf1a98c8..80b2bde6ca5 100644
--- a/mysql-test/suite/gcol/r/innodb_virtual_debug.result
+++ b/mysql-test/suite/gcol/r/innodb_virtual_debug.result
@@ -105,7 +105,7 @@ SET lock_wait_timeout = 1;
 ALTER TABLE t1 ADD UNIQUE INDEX(c, b);
 connection default;
 SET DEBUG_SYNC = 'now WAIT_FOR s1';
-SET DEBUG_SYNC = 'row_ins_sec_index_enter SIGNAL s2 WAIT_FOR s3';
+SET DEBUG_SYNC = 'row_log_insert_handle SIGNAL s2 WAIT_FOR s3';
 INSERT INTO t1(a, b) VALUES(2, 2);
 connection con1;
 ERROR HY000: Lock wait timeout exceeded; try restarting transaction
diff --git a/mysql-test/suite/gcol/t/innodb_virtual_debug.test b/mysql-test/suite/gcol/t/innodb_virtual_debug.test
index 40446b991cd..5ebc90dac19 100644
--- a/mysql-test/suite/gcol/t/innodb_virtual_debug.test
+++ b/mysql-test/suite/gcol/t/innodb_virtual_debug.test
@@ -295,7 +295,7 @@ SET lock_wait_timeout = 1;
 
 connection default;
 SET DEBUG_SYNC = 'now WAIT_FOR s1';
-SET DEBUG_SYNC = 'row_ins_sec_index_enter SIGNAL s2 WAIT_FOR s3';
+SET DEBUG_SYNC = 'row_log_insert_handle SIGNAL s2 WAIT_FOR s3';
 --send INSERT INTO t1(a, b) VALUES(2, 2)
 
 connection con1;
diff --git a/mysql-test/suite/innodb/r/alter_candidate_key.result b/mysql-test/suite/innodb/r/alter_candidate_key.result
index 79cb225e3b5..2ada5a499a8 100644
--- a/mysql-test/suite/innodb/r/alter_candidate_key.result
+++ b/mysql-test/suite/innodb/r/alter_candidate_key.result
@@ -74,7 +74,7 @@ connection con1;
 SET DEBUG_SYNC='now WAIT_FOR dml';
 BEGIN;
 INSERT INTO t1 SET a=NULL;
-ROLLBACK;
+COMMIT;
 set DEBUG_SYNC='now SIGNAL dml_done';
 connection default;
 ERROR 22004: Invalid use of NULL value
diff --git a/mysql-test/suite/innodb/r/alter_crash.result b/mysql-test/suite/innodb/r/alter_crash.result
index 46ea85d3e1e..a98aeb70a15 100644
--- a/mysql-test/suite/innodb/r/alter_crash.result
+++ b/mysql-test/suite/innodb/r/alter_crash.result
@@ -169,9 +169,6 @@ INSERT INTO t1(f1, f2) VALUES(2, "This is column2 value");
 ROLLBACK;
 set DEBUG_SYNC = 'now SIGNAL insert_done';
 connection default;
-Warnings:
-Warning	1265	Data truncated for column 'f3' at row 3
-Warning	1265	Data truncated for column 'f4' at row 3
 SHOW CREATE TABLE t1;
 Table	Create Table
 t1	CREATE TABLE `t1` (
@@ -202,6 +199,7 @@ connection default;
 SET DEBUG_SYNC = 'now WAIT_FOR scanned';
 BEGIN;
 INSERT INTO t1 VALUES(2,1);
+COMMIT;
 SET DEBUG_SYNC = 'now SIGNAL commit';
 SET DEBUG_SYNC = 'now WAIT_FOR c';
 SET GLOBAL innodb_fil_make_page_dirty_debug=0;
@@ -221,4 +219,5 @@ t1	CREATE TABLE `t1` (
 SELECT * FROM t1;
 a	b
 1	1
+2	1
 DROP TABLE t1;
diff --git a/mysql-test/suite/innodb/r/alter_mdl_timeout.result b/mysql-test/suite/innodb/r/alter_mdl_timeout.result
index 7af1362c69e..e4fba8e260f 100644
--- a/mysql-test/suite/innodb/r/alter_mdl_timeout.result
+++ b/mysql-test/suite/innodb/r/alter_mdl_timeout.result
@@ -10,7 +10,7 @@ begin;
 INSERT INTO t1 VALUES('e','e',5, 5);
 SET DEBUG_SYNC="now SIGNAL con1_insert";
 SET DEBUG_SYNC="now WAIT_FOR con1_wait";
-SET DEBUG_SYNC="before_row_upd_sec_new_index_entry SIGNAL con1_update WAIT_FOR alter_rollback";
+SET DEBUG_SYNC="after_row_upd_clust SIGNAL con1_update WAIT_FOR alter_rollback";
 UPDATE t1 set f4 = 10 order by f1 desc limit 2;
 connection default;
 ERROR HY000: Lock wait timeout exceeded; try restarting transaction
diff --git a/mysql-test/suite/innodb/r/innodb-alter-debug.result b/mysql-test/suite/innodb/r/innodb-alter-debug.result
index 81fc67e55c7..aae9432fc35 100644
--- a/mysql-test/suite/innodb/r/innodb-alter-debug.result
+++ b/mysql-test/suite/innodb/r/innodb-alter-debug.result
@@ -43,9 +43,8 @@ SET DEBUG_SYNC = 'now SIGNAL s2';
 /* connection default */
 connection default;
 /* reap */ alter table t1 force, add b int, ALGORITHM=inplace;
-ERROR 23000: Duplicate entry '1' for key 'uk'
 SET DEBUG_SYNC = 'row_log_table_apply1_before SIGNAL s1 WAIT_FOR s2';
-alter table t1 force, add b int, ALGORITHM=inplace;;
+alter table t1 force, add c int, ALGORITHM=inplace;;
 /* connection con1 */
 connection con1;
 set DEBUG_SYNC = 'now WAIT_FOR s1';
@@ -55,7 +54,6 @@ SET DEBUG_SYNC = 'now SIGNAL s2';
 /* connection default */
 connection default;
 /* reap */ alter table t1 force, add b int, ALGORITHM=inplace;
-ERROR 23000: Duplicate entry '1' for key 'uk'
 SET DEBUG_SYNC = 'RESET';
 drop table t1;
 #
@@ -72,7 +70,6 @@ ERROR 23000: Duplicate entry '1' for key 'a'
 SET DEBUG_SYNC = 'now SIGNAL S2';
 disconnect con1;
 connection default;
-ERROR 23000: Duplicate entry '1' for key 'a'
 SET DEBUG_SYNC='RESET';
 DROP TABLE t1;
 #
diff --git a/mysql-test/suite/innodb/r/innodb-index-debug.result b/mysql-test/suite/innodb/r/innodb-index-debug.result
index f6b23eea41a..c36a0531b95 100644
--- a/mysql-test/suite/innodb/r/innodb-index-debug.result
+++ b/mysql-test/suite/innodb/r/innodb-index-debug.result
@@ -118,20 +118,21 @@ drop table t480;
 # MDEV-12827 Assertion failure when reporting duplicate key error
 # in online table rebuild
 #
-CREATE TABLE t1 (j INT UNIQUE, i INT UNIQUE) ENGINE=InnoDB;
+CREATE TABLE t1 (j INT UNIQUE, i INT) ENGINE=InnoDB;
 INSERT INTO t1 VALUES(2, 2);
 connect  con1,localhost,root,,test;
 SET DEBUG_SYNC='row_log_table_apply1_before SIGNAL built WAIT_FOR log';
-ALTER TABLE t1 DROP j, FORCE;
+ALTER TABLE t1 DROP j, ADD UNIQUE INDEX(i), FORCE;
 connection default;
 SET DEBUG_SYNC='now WAIT_FOR built';
 SET DEBUG_DBUG='+d,row_ins_row_level';
 INSERT INTO t1 (i) VALUES (0),(0);
-ERROR 23000: Duplicate entry '0' for key 'i'
 SET DEBUG_SYNC='now SIGNAL log';
 SET DEBUG_DBUG=@saved_debug_dbug;
 connection con1;
 ERROR 23000: Duplicate entry '0' for key 'i'
+DELETE FROM t1;
+ALTER TABLE t1 ADD UNIQUE INDEX(i);
 SET DEBUG_SYNC='row_log_table_apply1_before SIGNAL built2 WAIT_FOR log2';
 ALTER TABLE t1 DROP j, FORCE;
 connection default;
@@ -141,7 +142,6 @@ UPDATE t1 SET i=0;
 ERROR 23000: Duplicate entry '0' for key 'i'
 SET DEBUG_SYNC='now SIGNAL log2';
 connection con1;
-ERROR 23000: Duplicate entry '0' for key 'i'
 disconnect con1;
 connection default;
 SET DEBUG_SYNC='RESET';
diff --git a/mysql-test/suite/innodb/r/innodb-index-online.result b/mysql-test/suite/innodb/r/innodb-index-online.result
index 1ee352cd402..56ed5da9747 100644
--- a/mysql-test/suite/innodb/r/innodb-index-online.result
+++ b/mysql-test/suite/innodb/r/innodb-index-online.result
@@ -85,7 +85,8 @@ ddl_sort_file_alter_table	0
 ddl_log_file_alter_table	0
 BEGIN;
 INSERT INTO t1 VALUES(7,4,2);
-ROLLBACK;
+COMMIT;
+DELETE FROM t1 where c1 = 7;
 SET DEBUG_SYNC = 'now SIGNAL rollback_done';
 connection con1;
 ERROR 23000: Duplicate entry '4' for key 'c2'
@@ -96,14 +97,14 @@ SET DEBUG_SYNC = 'now WAIT_FOR created';
 SELECT name, count FROM INFORMATION_SCHEMA.INNODB_METRICS WHERE subsystem = 'ddl';
 name	count
 ddl_background_drop_indexes	0
-ddl_online_create_index	0
+ddl_online_create_index	1
 ddl_pending_alter_table	1
 ddl_sort_file_alter_table	0
 ddl_log_file_alter_table	0
 INSERT INTO t1 VALUES(6,3,1);
 SET DEBUG_SYNC = 'now SIGNAL dml_done';
 connection con1;
-ERROR 23000: Duplicate entry for key 'c2'
+ERROR 23000: Duplicate entry '' for key '*UNKNOWN*'
 DELETE FROM t1 WHERE c1=6;
 ALTER TABLE t1 ADD UNIQUE INDEX(c2);
 SELECT name, count FROM INFORMATION_SCHEMA.INNODB_METRICS WHERE subsystem = 'ddl';
@@ -243,38 +244,22 @@ ddl_online_create_index	1
 ddl_pending_alter_table	1
 ddl_sort_file_alter_table	0
 ddl_log_file_alter_table	0
-BEGIN;
-DELETE FROM t1;
-ROLLBACK;
 UPDATE t1 SET c2 = c2 + 1;
-BEGIN;
+UPDATE t1 SET c2 = c2 + 2;
 UPDATE t1 SET c2 = c2 + 1;
-DELETE FROM t1;
-ROLLBACK;
-BEGIN;
-DELETE FROM t1;
-ROLLBACK;
+UPDATE t1 SET c2 = c2 + 2;
 UPDATE t1 SET c2 = c2 + 1;
-BEGIN;
+UPDATE t1 SET c2 = c2 + 2;
 UPDATE t1 SET c2 = c2 + 1;
-DELETE FROM t1;
-ROLLBACK;
-BEGIN;
-DELETE FROM t1;
-ROLLBACK;
+UPDATE t1 SET c2 = c2 + 2;
 UPDATE t1 SET c2 = c2 + 1;
-BEGIN;
+UPDATE t1 SET c2 = c2 + 2;
 UPDATE t1 SET c2 = c2 + 1;
-DELETE FROM t1;
-ROLLBACK;
-BEGIN;
-DELETE FROM t1;
-ROLLBACK;
+UPDATE t1 SET c2 = c2 + 2;
 UPDATE t1 SET c2 = c2 + 1;
-BEGIN;
+UPDATE t1 SET c2 = c2 + 2;
 UPDATE t1 SET c2 = c2 + 1;
-DELETE FROM t1;
-ROLLBACK;
+UPDATE t1 SET c2 = c2 + 2;
 SELECT name, count FROM INFORMATION_SCHEMA.INNODB_METRICS WHERE subsystem = 'ddl';
 name	count
 ddl_background_drop_indexes	0
@@ -310,7 +295,7 @@ ERROR HY000: Creating index 'c2e' required more than 'innodb_online_alter_log_ma
 SELECT name, count FROM INFORMATION_SCHEMA.INNODB_METRICS WHERE subsystem = 'ddl';
 name	count
 ddl_background_drop_indexes	1
-ddl_online_create_index	0
+ddl_online_create_index	1
 ddl_pending_alter_table	0
 ddl_sort_file_alter_table	0
 ddl_log_file_alter_table	1
@@ -321,7 +306,7 @@ name	pos
 SELECT name, count FROM INFORMATION_SCHEMA.INNODB_METRICS WHERE subsystem = 'ddl';
 name	count
 ddl_background_drop_indexes	1
-ddl_online_create_index	0
+ddl_online_create_index	1
 ddl_pending_alter_table	0
 ddl_sort_file_alter_table	0
 ddl_log_file_alter_table	1
@@ -361,19 +346,45 @@ ddl_log_file_alter_table	1
 BEGIN;
 INSERT INTO t1 SELECT 320 + c1, c2, c3 FROM t1 WHERE c1 > 160;
 DELETE FROM t1 WHERE c1 > 320;
-ROLLBACK;
+COMMIT;
 BEGIN;
 UPDATE t1 SET c2 = c2 + 1;
-DELETE FROM t1;
-ROLLBACK;
+COMMIT;
 BEGIN;
 INSERT INTO t1 SELECT 320 + c1, c2, c3 FROM t1 WHERE c1 > 160;
 DELETE FROM t1 WHERE c1 > 320;
-ROLLBACK;
+COMMIT;
 BEGIN;
 UPDATE t1 SET c2 = c2 + 1;
-DELETE FROM t1;
-ROLLBACK;
+COMMIT;
+BEGIN;
+INSERT INTO t1 SELECT 320 + c1, c2, c3 FROM t1 WHERE c1 > 160;
+DELETE FROM t1 WHERE c1 > 320;
+COMMIT;
+BEGIN;
+UPDATE t1 SET c2 = c2 + 1;
+COMMIT;
+BEGIN;
+INSERT INTO t1 SELECT 320 + c1, c2, c3 FROM t1 WHERE c1 > 160;
+DELETE FROM t1 WHERE c1 > 320;
+COMMIT;
+BEGIN;
+UPDATE t1 SET c2 = c2 + 1;
+COMMIT;
+BEGIN;
+INSERT INTO t1 SELECT 320 + c1, c2, c3 FROM t1 WHERE c1 > 160;
+DELETE FROM t1 WHERE c1 > 320;
+COMMIT;
+BEGIN;
+UPDATE t1 SET c2 = c2 + 1;
+COMMIT;
+BEGIN;
+INSERT INTO t1 SELECT 320 + c1, c2, c3 FROM t1 WHERE c1 > 160;
+DELETE FROM t1 WHERE c1 > 320;
+COMMIT;
+BEGIN;
+UPDATE t1 SET c2 = c2 + 1;
+COMMIT;
 SELECT name, count FROM INFORMATION_SCHEMA.INNODB_METRICS WHERE subsystem = 'ddl';
 name	count
 ddl_background_drop_indexes	0
@@ -498,7 +509,6 @@ SET DEBUG_SYNC = 'row_log_apply_before SIGNAL t1u_created WAIT_FOR dup_done';
 ALTER TABLE t1 ADD UNIQUE(c);
 connection con1;
 SET DEBUG_SYNC = 'now WAIT_FOR t1u_created';
-BEGIN;
 INSERT INTO t1 VALUES('bar'),('bar');
 SET DEBUG_SYNC = 'now SIGNAL dup_done';
 connection default;
diff --git a/mysql-test/suite/innodb/r/innodb-table-online.result b/mysql-test/suite/innodb/r/innodb-table-online.result
index 91d9b355125..659d645b5cc 100644
--- a/mysql-test/suite/innodb/r/innodb-table-online.result
+++ b/mysql-test/suite/innodb/r/innodb-table-online.result
@@ -97,11 +97,11 @@ ddl_online_create_index	1
 ddl_pending_alter_table	1
 ddl_sort_file_alter_table	0
 ddl_log_file_alter_table	0
-BEGIN;
 INSERT INTO t1 VALUES(4,7,2);
 SET DEBUG_SYNC = 'now SIGNAL insert_done';
 connection con1;
 ERROR 23000: Duplicate entry '4' for key 'PRIMARY'
+DELETE FROM t1 WHERE c1=4 and c2=7;
 connection default;
 ROLLBACK;
 connection con1;
@@ -213,30 +213,22 @@ ddl_online_create_index	1
 ddl_pending_alter_table	1
 ddl_sort_file_alter_table	0
 ddl_log_file_alter_table	1
-BEGIN;
-DELETE FROM t1;
-ROLLBACK;
 UPDATE t1 SET c2 = c2 + 1;
-BEGIN;
+UPDATE t1 SET c2 = c2 + 2;
 UPDATE t1 SET c2 = c2 + 1;
-DELETE FROM t1;
-ROLLBACK;
-BEGIN;
-DELETE FROM t1;
-ROLLBACK;
+UPDATE t1 SET c2 = c2 + 2;
 UPDATE t1 SET c2 = c2 + 1;
-BEGIN;
+UPDATE t1 SET c2 = c2 + 2;
 UPDATE t1 SET c2 = c2 + 1;
-DELETE FROM t1;
-ROLLBACK;
-BEGIN;
-DELETE FROM t1;
-ROLLBACK;
+UPDATE t1 SET c2 = c2 + 2;
 UPDATE t1 SET c2 = c2 + 1;
-BEGIN;
+UPDATE t1 SET c2 = c2 + 2;
 UPDATE t1 SET c2 = c2 + 1;
-DELETE FROM t1;
-ROLLBACK;
+UPDATE t1 SET c2 = c2 + 2;
+UPDATE t1 SET c2 = c2 + 1;
+UPDATE t1 SET c2 = c2 + 2;
+UPDATE t1 SET c2 = c2 + 1;
+UPDATE t1 SET c2 = c2 + 2;
 SELECT name, count FROM INFORMATION_SCHEMA.INNODB_METRICS WHERE subsystem = 'ddl';
 name	count
 ddl_background_drop_indexes	0
@@ -286,7 +278,7 @@ SET DEBUG_SYNC = 'row_log_table_apply1_before SIGNAL rebuilt3 WAIT_FOR dml3_done
 ALTER TABLE t1 ADD PRIMARY KEY(c22f), CHANGE c2 c22f INT;
 ERROR 42000: Multiple primary key defined
 ALTER TABLE t1 DROP PRIMARY KEY, ADD PRIMARY KEY(c22f), CHANGE c2 c22f INT;
-ERROR 23000: Duplicate entry '5' for key 'PRIMARY'
+ERROR 23000: Duplicate entry '26' for key 'PRIMARY'
 ALTER TABLE t1 DROP PRIMARY KEY, ADD PRIMARY KEY(c22f,c1,c4(5)),
 CHANGE c2 c22f INT, CHANGE c3 c3 CHAR(255) NULL, CHANGE c1 c1 INT AFTER c22f,
 ADD COLUMN c4 VARCHAR(6) DEFAULT 'Online', LOCK=NONE;
@@ -302,11 +294,8 @@ ddl_log_file_alter_table	1
 BEGIN;
 INSERT INTO t1 SELECT 320 + c1, c2, c3 FROM t1 WHERE c1 > 240;
 DELETE FROM t1 WHERE c1 > 320;
-ROLLBACK;
-BEGIN;
 UPDATE t1 SET c2 = c2 + 1;
-DELETE FROM t1;
-ROLLBACK;
+COMMIT;
 SELECT name, count FROM INFORMATION_SCHEMA.INNODB_METRICS WHERE subsystem = 'ddl';
 name	count
 ddl_background_drop_indexes	0
@@ -374,12 +363,11 @@ SET DEBUG_SYNC = 'now WAIT_FOR c3p5_created0';
 BEGIN;
 INSERT INTO t1 VALUES(347,33101,'Pikku kakkosen posti','YLETV2');
 INSERT INTO t1 VALUES(33101,347,NULL,'');
+COMMIT;
 SET DEBUG_SYNC = 'now SIGNAL ins_done0';
 connection con1;
-ERROR 01000: Data truncated for column 'c3' at row 323
-connection default;
-ROLLBACK;
-connection con1;
+ERROR 22004: Invalid use of NULL value
+DELETE FROM t1 WHERE c1= 347 and c22f = 33101;
 ALTER TABLE t1 MODIFY c3 CHAR(255) NOT NULL;
 SET DEBUG_SYNC = 'row_log_table_apply1_before SIGNAL c3p5_created WAIT_FOR ins_done';
 ALTER TABLE t1 DROP PRIMARY KEY, DROP COLUMN c22f,
@@ -405,20 +393,20 @@ ddl_log_file_alter_table	2
 connection default;
 SELECT COUNT(*) FROM t1;
 COUNT(*)
-321
+322
 ALTER TABLE t1 ROW_FORMAT=REDUNDANT;
 SELECT * FROM t1 LIMIT 10;
 c22f	c1	c3	c4
-5	1	1foo	Online
-5	6	6foofoofoofoofoofoo	Online
-5	11	11foofoofoofoofoofoofoofoofoofoofoo	Online
-5	16	16foofoofoofoofoofoofoofoofoofoofoofoofoofoofoofoo	Online
-5	21	21foofoofoofoofoofoofoofoofoofoofoofoofoofoofoofoofoofoofoofoofoo	Online
-5	26	26foofoofoofoofoofoofoofoofoofoofoofoofoofoofoofoofoofoofoofoofoofoofoofoofoofoo	Online
-5	31	31foofoofoofoofoofoofoofoofoofoofoofoofoofoofoofoofoofoofoofoofoofoofoofoofoofoofoofoofoofoofoo	Online
-5	36	36foofoofoofoofoofoofoofoofoofoofoofoofoofoofoofoofoofoofoofoofoofoofoofoofoofoofoofoofoofoofoofoofoofoofoofoo	Online
-5	41	41foofoofoofoofoofoofoofoofoofoofoofoofoofoofoofoofoofoofoofoofoofoofoofoofoofoofoofoofoofoofoofoofoofoofoofoofoofoofoofoofoo	Online
-5	46	46foofoofoofoofoofoofoofoofoofoofoofoofoofoofoofoofoofoofoofoofoofoofoofoofoofoofoofoofoofoofoofoofoofoofoofoofoofoofoofoofoofoofoofoofoofoo	Online
+27	1	1foo	Online
+27	6	6foofoofoofoofoofoo	Online
+27	11	11foofoofoofoofoofoofoofoofoofoofoo	Online
+27	16	16foofoofoofoofoofoofoofoofoofoofoofoofoofoofoofoo	Online
+27	21	21foofoofoofoofoofoofoofoofoofoofoofoofoofoofoofoofoofoofoofoofoo	Online
+27	26	26foofoofoofoofoofoofoofoofoofoofoofoofoofoofoofoofoofoofoofoofoofoofoofoofoofoo	Online
+27	31	31foofoofoofoofoofoofoofoofoofoofoofoofoofoofoofoofoofoofoofoofoofoofoofoofoofoofoofoofoofoofoo	Online
+27	36	36foofoofoofoofoofoofoofoofoofoofoofoofoofoofoofoofoofoofoofoofoofoofoofoofoofoofoofoofoofoofoofoofoofoofoofoo	Online
+27	41	41foofoofoofoofoofoofoofoofoofoofoofoofoofoofoofoofoofoofoofoofoofoofoofoofoofoofoofoofoofoofoofoofoofoofoofoofoofoofoofoofoo	Online
+27	46	46foofoofoofoofoofoofoofoofoofoofoofoofoofoofoofoofoofoofoofoofoofoofoofoofoofoofoofoofoofoofoofoofoofoofoofoofoofoofoofoofoofoofoofoofoofoo	Online
 connection con1;
 ALTER TABLE t1 DISCARD TABLESPACE;
 connection default;
diff --git a/mysql-test/suite/innodb/r/table_definition_cache_debug.result b/mysql-test/suite/innodb/r/table_definition_cache_debug.result
index df171c89cd4..6bd754aaca3 100644
--- a/mysql-test/suite/innodb/r/table_definition_cache_debug.result
+++ b/mysql-test/suite/innodb/r/table_definition_cache_debug.result
@@ -12,6 +12,7 @@ connection default;
 SET DEBUG_SYNC = 'now WAIT_FOR scanned';
 BEGIN;
 INSERT INTO to_be_evicted VALUES(3, 2);
+COMMIT;
 SET DEBUG_SYNC = 'now SIGNAL got_duplicate';
 connection ddl;
 ERROR 23000: Duplicate entry '2' for key 'b'
diff --git a/mysql-test/suite/innodb/t/alter_candidate_key.test b/mysql-test/suite/innodb/t/alter_candidate_key.test
index 7c8f5e30993..824ad1ea799 100644
--- a/mysql-test/suite/innodb/t/alter_candidate_key.test
+++ b/mysql-test/suite/innodb/t/alter_candidate_key.test
@@ -50,7 +50,7 @@ connection con1;
 SET DEBUG_SYNC='now WAIT_FOR dml';
 BEGIN;
 INSERT INTO t1 SET a=NULL;
-ROLLBACK;
+COMMIT;
 set DEBUG_SYNC='now SIGNAL dml_done';
 connection default;
 --error ER_INVALID_USE_OF_NULL
diff --git a/mysql-test/suite/innodb/t/alter_crash.test b/mysql-test/suite/innodb/t/alter_crash.test
index 1049efd3e12..e0e294ae4f0 100644
--- a/mysql-test/suite/innodb/t/alter_crash.test
+++ b/mysql-test/suite/innodb/t/alter_crash.test
@@ -213,6 +213,7 @@ connection default;
 SET DEBUG_SYNC = 'now WAIT_FOR scanned';
 BEGIN;
 INSERT INTO t1 VALUES(2,1);
+COMMIT;
 SET DEBUG_SYNC = 'now SIGNAL commit';
 SET DEBUG_SYNC = 'now WAIT_FOR c';
 # Make all pending changes durable for recovery.
diff --git a/mysql-test/suite/innodb/t/alter_mdl_timeout.test b/mysql-test/suite/innodb/t/alter_mdl_timeout.test
index 15e7f524fd0..ff77921b2d2 100644
--- a/mysql-test/suite/innodb/t/alter_mdl_timeout.test
+++ b/mysql-test/suite/innodb/t/alter_mdl_timeout.test
@@ -14,7 +14,7 @@ begin;
 INSERT INTO t1 VALUES('e','e',5, 5);
 SET DEBUG_SYNC="now SIGNAL con1_insert";
 SET DEBUG_SYNC="now WAIT_FOR con1_wait";
-SET DEBUG_SYNC="before_row_upd_sec_new_index_entry SIGNAL con1_update WAIT_FOR alter_rollback";
+SET DEBUG_SYNC="after_row_upd_clust SIGNAL con1_update WAIT_FOR alter_rollback";
 SEND UPDATE t1 set f4 = 10 order by f1 desc limit 2;
 
 connection default;
diff --git a/mysql-test/suite/innodb/t/alter_not_null_debug.test b/mysql-test/suite/innodb/t/alter_not_null_debug.test
index 9c5ba0faff0..87113b2b3f8 100644
--- a/mysql-test/suite/innodb/t/alter_not_null_debug.test
+++ b/mysql-test/suite/innodb/t/alter_not_null_debug.test
@@ -7,7 +7,7 @@ let $sql_mode = `SELECT @@SQL_MODE`;
 let $error_code = 0;
 
 if ($sql_mode == "STRICT_TRANS_TABLES") {
- let $error_code = WARN_DATA_TRUNCATED;
+ let $error_code = ER_INVALID_USE_OF_NULL;
 }
 
 
diff --git a/mysql-test/suite/innodb/t/innodb-alter-debug.test b/mysql-test/suite/innodb/t/innodb-alter-debug.test
index 1789ec294e4..2241ef5d295 100644
--- a/mysql-test/suite/innodb/t/innodb-alter-debug.test
+++ b/mysql-test/suite/innodb/t/innodb-alter-debug.test
@@ -52,11 +52,10 @@ SET DEBUG_SYNC = 'now SIGNAL s2';
 --echo /* connection default */
 connection default;
 --echo /* reap */ alter table t1 force, add b int, ALGORITHM=inplace;
---error ER_DUP_ENTRY
 --reap
 
 SET DEBUG_SYNC = 'row_log_table_apply1_before SIGNAL s1 WAIT_FOR s2';
---send alter table t1 force, add b int, ALGORITHM=inplace;
+--send alter table t1 force, add c int, ALGORITHM=inplace;
 
 --echo /* connection con1 */
 connection con1;
@@ -68,7 +67,6 @@ SET DEBUG_SYNC = 'now SIGNAL s2';
 --echo /* connection default */
 connection default;
 --echo /* reap */ alter table t1 force, add b int, ALGORITHM=inplace;
---error ER_DUP_ENTRY
 --reap
 SET DEBUG_SYNC = 'RESET';
 
@@ -92,7 +90,6 @@ SET DEBUG_SYNC = 'now SIGNAL S2';
 disconnect con1;
 
 CONNECTION default;
---error ER_DUP_ENTRY
 reap;
 SET DEBUG_SYNC='RESET';
 
diff --git a/mysql-test/suite/innodb/t/innodb-index-debug.test b/mysql-test/suite/innodb/t/innodb-index-debug.test
index 204bdfe5540..f03ef061769 100644
--- a/mysql-test/suite/innodb/t/innodb-index-debug.test
+++ b/mysql-test/suite/innodb/t/innodb-index-debug.test
@@ -122,17 +122,16 @@ drop table t480;
 --echo # in online table rebuild
 --echo #
 
-CREATE TABLE t1 (j INT UNIQUE, i INT UNIQUE) ENGINE=InnoDB;
+CREATE TABLE t1 (j INT UNIQUE, i INT) ENGINE=InnoDB;
 INSERT INTO t1 VALUES(2, 2);
 --connect (con1,localhost,root,,test)
 SET DEBUG_SYNC='row_log_table_apply1_before SIGNAL built WAIT_FOR log';
 --send
-ALTER TABLE t1 DROP j, FORCE;
+ALTER TABLE t1 DROP j, ADD UNIQUE INDEX(i), FORCE;
 
 --connection default
 SET DEBUG_SYNC='now WAIT_FOR built';
 SET DEBUG_DBUG='+d,row_ins_row_level';
---error ER_DUP_ENTRY
 INSERT INTO t1 (i) VALUES (0),(0);
 SET DEBUG_SYNC='now SIGNAL log';
 SET DEBUG_DBUG=@saved_debug_dbug;
@@ -140,6 +139,8 @@ SET DEBUG_DBUG=@saved_debug_dbug;
 --connection con1
 --error ER_DUP_ENTRY
 reap;
+DELETE FROM t1;
+ALTER TABLE t1 ADD UNIQUE INDEX(i);
 SET DEBUG_SYNC='row_log_table_apply1_before SIGNAL built2 WAIT_FOR log2';
 --send
 ALTER TABLE t1 DROP j, FORCE;
@@ -152,7 +153,6 @@ UPDATE t1 SET i=0;
 SET DEBUG_SYNC='now SIGNAL log2';
 
 --connection con1
---error ER_DUP_ENTRY
 reap;
 --disconnect con1
 --connection default
diff --git a/mysql-test/suite/innodb/t/innodb-index-online.test b/mysql-test/suite/innodb/t/innodb-index-online.test
index 6f30dad531d..2cb84b18402 100644
--- a/mysql-test/suite/innodb/t/innodb-index-online.test
+++ b/mysql-test/suite/innodb/t/innodb-index-online.test
@@ -98,7 +98,8 @@ eval $innodb_metrics_select;
 # Insert a duplicate entry (4) for the already started UNIQUE INDEX(c2).
 BEGIN;
 INSERT INTO t1 VALUES(7,4,2);
-ROLLBACK;
+COMMIT;
+DELETE FROM t1 where c1 = 7;
 SET DEBUG_SYNC = 'now SIGNAL rollback_done';
 
 connection con1;
@@ -121,7 +122,7 @@ INSERT INTO t1 VALUES(6,3,1);
 SET DEBUG_SYNC = 'now SIGNAL dml_done';
 connection con1;
 # This is due to the duplicate entry (6,3,1).
---error ER_DUP_UNKNOWN_IN_INDEX
+--error ER_DUP_ENTRY
 reap;
 DELETE FROM t1 WHERE c1=6;
 ALTER TABLE t1 ADD UNIQUE INDEX(c2);
@@ -237,17 +238,11 @@ SET DEBUG_SYNC = 'now WAIT_FOR c2e_created';
 # At this point, the clustered index scan must have completed,
 # but the modification log keeps accumulating due to the DEBUG_SYNC.
 eval $innodb_metrics_select;
-let $c= 4;
+let $c= 8;
 while ($c)
 {
-  BEGIN;
-  DELETE FROM t1;
-  ROLLBACK;
   UPDATE t1 SET c2 = c2 + 1;
-  BEGIN;
-  UPDATE t1 SET c2 = c2 + 1;
-  DELETE FROM t1;
-  ROLLBACK;
+  UPDATE t1 SET c2 = c2 + 2;
   dec $c;
 }
 # Incomplete index c2e should exist until the DDL thread notices the overflow.
@@ -325,17 +320,16 @@ connection default;
 SET DEBUG_SYNC = 'now WAIT_FOR c2f_created';
 # Generate some log (delete-mark, delete-unmark, insert etc.)
 eval $innodb_metrics_select;
-let $c= 2;
+let $c= 6;
 while ($c)
 {
 BEGIN;
 INSERT INTO t1 SELECT 320 + c1, c2, c3 FROM t1 WHERE c1 > 160;
 DELETE FROM t1 WHERE c1 > 320;
-ROLLBACK;
+COMMIT;
 BEGIN;
 UPDATE t1 SET c2 = c2 + 1;
-DELETE FROM t1;
-ROLLBACK;
+COMMIT;
 dec $c;
 }
 eval $innodb_metrics_select;
@@ -481,7 +475,6 @@ send ALTER TABLE t1 ADD UNIQUE(c);
 
 connection con1;
 SET DEBUG_SYNC = 'now WAIT_FOR t1u_created';
-BEGIN;
 INSERT INTO t1 VALUES('bar'),('bar');
 SET DEBUG_SYNC = 'now SIGNAL dup_done';
 
diff --git a/mysql-test/suite/innodb/t/innodb-table-online.test b/mysql-test/suite/innodb/t/innodb-table-online.test
index 7ed87fcc26b..a1cf50f541d 100644
--- a/mysql-test/suite/innodb/t/innodb-table-online.test
+++ b/mysql-test/suite/innodb/t/innodb-table-online.test
@@ -101,7 +101,6 @@ SET DEBUG_SYNC = 'now WAIT_FOR scanned';
 eval $innodb_metrics_select;
 
 # Insert a duplicate entry (4) for the already started UNIQUE INDEX(c1).
-BEGIN;
 INSERT INTO t1 VALUES(4,7,2);
 SET DEBUG_SYNC = 'now SIGNAL insert_done';
 
@@ -111,7 +110,7 @@ connection con1;
 # error on the (4,7,2).
 --error ER_DUP_ENTRY
 reap;
-
+DELETE FROM t1 WHERE c1=4 and c2=7;
 connection default;
 ROLLBACK;
 
@@ -204,17 +203,11 @@ UPDATE t1 SET c2 = c2 + 1;
 # At this point, the clustered index scan must have completed,
 # but the modification log keeps accumulating due to the DEBUG_SYNC.
 eval $innodb_metrics_select;
-let $c= 3;
+let $c= 8;
 while ($c)
 {
-  BEGIN;
-  DELETE FROM t1;
-  ROLLBACK;
   UPDATE t1 SET c2 = c2 + 1;
-  BEGIN;
-  UPDATE t1 SET c2 = c2 + 1;
-  DELETE FROM t1;
-  ROLLBACK;
+  UPDATE t1 SET c2 = c2 + 2;
   dec $c;
 }
 # Temporary table should exist until the DDL thread notices the overflow.
@@ -279,11 +272,8 @@ eval $innodb_metrics_select;
 BEGIN;
 INSERT INTO t1 SELECT 320 + c1, c2, c3 FROM t1 WHERE c1 > 240;
 DELETE FROM t1 WHERE c1 > 320;
-ROLLBACK;
-BEGIN;
 UPDATE t1 SET c2 = c2 + 1;
-DELETE FROM t1;
-ROLLBACK;
+COMMIT;
 eval $innodb_metrics_select;
 # Release con1.
 SET DEBUG_SYNC = 'now SIGNAL dml3_done';
@@ -346,16 +336,13 @@ SET DEBUG_SYNC = 'now WAIT_FOR c3p5_created0';
 BEGIN;
 INSERT INTO t1 VALUES(347,33101,'Pikku kakkosen posti','YLETV2');
 INSERT INTO t1 VALUES(33101,347,NULL,'');
+COMMIT;
 SET DEBUG_SYNC = 'now SIGNAL ins_done0';
 
 connection con1;
---error WARN_DATA_TRUNCATED
+--error ER_INVALID_USE_OF_NULL
 reap;
-
-connection default;
-ROLLBACK;
-
-connection con1;
+DELETE FROM t1 WHERE c1= 347 and c22f = 33101; 
 ALTER TABLE t1 MODIFY c3 CHAR(255) NOT NULL;
 
 SET DEBUG_SYNC = 'row_log_table_apply1_before SIGNAL c3p5_created WAIT_FOR ins_done';
diff --git a/mysql-test/suite/innodb/t/instant_alter_debug.test b/mysql-test/suite/innodb/t/instant_alter_debug.test
index f102185c27f..c6eca884907 100644
--- a/mysql-test/suite/innodb/t/instant_alter_debug.test
+++ b/mysql-test/suite/innodb/t/instant_alter_debug.test
@@ -303,7 +303,6 @@ ROLLBACK;
 SET DEBUG_SYNC = 'now SIGNAL logged';
 
 connection ddl;
---error ER_INVALID_USE_OF_NULL
 reap;
 disconnect ddl;
 
diff --git a/mysql-test/suite/innodb/t/table_definition_cache_debug.test b/mysql-test/suite/innodb/t/table_definition_cache_debug.test
index 6a466af4cc5..8950691e05c 100644
--- a/mysql-test/suite/innodb/t/table_definition_cache_debug.test
+++ b/mysql-test/suite/innodb/t/table_definition_cache_debug.test
@@ -29,6 +29,7 @@ SET DEBUG_SYNC = 'now WAIT_FOR scanned';
 # and then hogs the table lock, so that the unique index cannot be dropped.
 BEGIN;
 INSERT INTO to_be_evicted VALUES(3, 2);
+COMMIT;
 SET DEBUG_SYNC = 'now SIGNAL got_duplicate';
 
 connection ddl;
diff --git a/storage/innobase/CMakeLists.txt b/storage/innobase/CMakeLists.txt
index 3d4f4d04ae8..6faf62064d5 100644
--- a/storage/innobase/CMakeLists.txt
+++ b/storage/innobase/CMakeLists.txt
@@ -1,6 +1,6 @@
 
 # Copyright (c) 2006, 2017, Oracle and/or its affiliates. All rights reserved.
-# Copyright (c) 2014, 2021, MariaDB Corporation.
+# Copyright (c) 2014, 2022, MariaDB Corporation.
 #
 # This program is free software; you can redistribute it and/or modify
 # it under the terms of the GNU General Public License as published by
@@ -214,7 +214,6 @@ SET(INNOBASE_SOURCES
 	include/row0import.h
 	include/row0ins.h
 	include/row0log.h
-	include/row0log.inl
 	include/row0merge.h
 	include/row0mysql.h
 	include/row0purge.h
diff --git a/storage/innobase/btr/btr0cur.cc b/storage/innobase/btr/btr0cur.cc
index 05f3128d819..f12561c3126 100644
--- a/storage/innobase/btr/btr0cur.cc
+++ b/storage/innobase/btr/btr0cur.cc
@@ -5401,10 +5401,6 @@ btr_cur_del_mark_set_clust_rec(
 		 << ib::hex(trx->id) << ": "
 		 << rec_printer(rec, offsets).str());
 
-	if (dict_index_is_online_ddl(index)) {
-		row_log_table_delete(rec, index, offsets, NULL);
-	}
-
 	btr_cur_upd_rec_sys(block, rec, index, offsets, trx, roll_ptr, mtr);
 	return(err);
 }
@@ -7061,8 +7057,6 @@ btr_store_big_rec_extern_fields(
 						     + prev_block->page.frame,
 						     page_no);
 				}
-			} else if (dict_index_is_online_ddl(index)) {
-				row_log_table_blob_alloc(index, page_no);
 			}
 
 			ut_ad(!page_has_siblings(block->page.frame));
@@ -7312,8 +7306,6 @@ btr_free_externally_stored_field(
 	page_t*		page;
 	const uint32_t	space_id	= mach_read_from_4(
 		field_ref + BTR_EXTERN_SPACE_ID);
-	const uint32_t	start_page	= mach_read_from_4(
-		field_ref + BTR_EXTERN_PAGE_NO);
 	uint32_t	page_no;
 	uint32_t	next_page_no;
 	mtr_t		mtr;
@@ -7383,10 +7375,6 @@ btr_free_externally_stored_field(
 			return;
 		}
 
-		if (page_no == start_page && dict_index_is_online_ddl(index)) {
-			row_log_table_blob_free(index, start_page);
-		}
-
 		ext_block = buf_page_get(
 			page_id_t(space_id, page_no), ext_zip_size,
 			RW_X_LATCH, &mtr);
diff --git a/storage/innobase/dict/dict0dict.cc b/storage/innobase/dict/dict0dict.cc
index 4253326d46a..3ea7cc64fc3 100644
--- a/storage/innobase/dict/dict0dict.cc
+++ b/storage/innobase/dict/dict0dict.cc
@@ -2088,7 +2088,6 @@ dict_index_remove_from_cache_low(
 	there can't be any active operations on this index (or table). */
 
 	if (index->online_log) {
-		ut_ad(index->online_status == ONLINE_INDEX_CREATION);
 		row_log_free(index->online_log);
 		index->online_log = NULL;
 	}
diff --git a/storage/innobase/handler/ha_innodb.cc b/storage/innobase/handler/ha_innodb.cc
index 72300f83c9c..f3b48c40bd9 100644
--- a/storage/innobase/handler/ha_innodb.cc
+++ b/storage/innobase/handler/ha_innodb.cc
@@ -5184,33 +5184,25 @@ ha_innobase::keys_to_use_for_scanning()
 	return(&key_map_full);
 }
 
-/****************************************************************//**
-Ensures that if there's a concurrent inplace ADD INDEX, being-indexed virtual
-columns are computed. They are not marked as indexed in the old table, so the
-server won't add them to the read_set automatically */
-void
-ha_innobase::column_bitmaps_signal()
-/*================================*/
+/** Ensure that indexed virtual columns will be computed. */
+void ha_innobase::column_bitmaps_signal()
 {
-	if (!table->vfield || table->current_lock != F_WRLCK) {
-		return;
-	}
+  if (!table->vfield)
+    return;
 
-	dict_index_t*	clust_index = dict_table_get_first_index(m_prebuilt->table);
-	uint	num_v = 0;
-	for (uint j = 0; j < table->s->virtual_fields; j++) {
-		if (table->vfield[j]->stored_in_db()) {
-			continue;
-		}
+  dict_index_t* clust_index= dict_table_get_first_index(m_prebuilt->table);
+  uint num_v= 0;
+  for (uint j = 0; j < table->s->virtual_fields; j++)
+  {
+    if (table->vfield[j]->stored_in_db())
+      continue;
 
-		dict_col_t*	col = &m_prebuilt->table->v_cols[num_v].m_col;
-		if (col->ord_part ||
-		    (dict_index_is_online_ddl(clust_index) &&
-		     row_log_col_is_indexed(clust_index, num_v))) {
-			table->mark_virtual_column_with_deps(table->vfield[j]);
-		}
-		num_v++;
-	}
+    dict_col_t *col= &m_prebuilt->table->v_cols[num_v].m_col;
+    if (col->ord_part ||
+        (dict_index_is_online_ddl(clust_index) &&
+         row_log_col_is_indexed(clust_index, num_v)))
+      table->mark_virtual_column_with_deps(table->vfield[j]);
+  }
 }
 
 
@@ -8195,34 +8187,13 @@ calc_row_difference(
 			}
 		}
 
-#ifdef UNIV_DEBUG
-		bool	online_ord_part = false;
-#endif
-
 		if (is_virtual) {
 			/* If the virtual column is not indexed,
 			we shall ignore it for update */
 			if (!col->ord_part) {
-				/* Check whether there is a table-rebuilding
-				online ALTER TABLE in progress, and this
-				virtual column could be newly indexed, thus
-				it will be materialized. Then we will have
-				to log its update.
-				Note, we do not support online dropping virtual
-				column while adding new index, nor with
-				online alter column order while adding index,
-				so the virtual column sequence must not change
-				if it is online operation */
-				if (dict_index_is_online_ddl(clust_index)
-				    && row_log_col_is_indexed(clust_index,
-							      num_v)) {
-#ifdef UNIV_DEBUG
-					online_ord_part = true;
-#endif
-				} else {
-					num_v++;
-					continue;
-				}
+			next:
+				num_v++;
+				continue;
 			}
 
 			if (!uvect->old_vrow) {
@@ -8248,8 +8219,7 @@ calc_row_difference(
 					prebuilt, vfield, o_len,
 					col, old_mysql_row_col,
 					col_pack_len, buf);
-			       num_v++;
-			       continue;
+				goto next;
 			}
 		}
 
@@ -8298,7 +8268,7 @@ calc_row_difference(
 				upd_fld_set_virtual_col(ufield);
 				ufield->field_no = num_v;
 
-				ut_ad(col->ord_part || online_ord_part);
+				ut_ad(col->ord_part);
 				ufield->old_v_val = static_cast<dfield_t*>(
 					mem_heap_alloc(
 						uvect->heap,
@@ -8383,7 +8353,7 @@ calc_row_difference(
 				prebuilt, vfield, o_len,
 				col, old_mysql_row_col,
 				col_pack_len, buf);
-			ut_ad(col->ord_part || online_ord_part);
+			ut_ad(col->ord_part);
 			num_v++;
 		}
 	}
diff --git a/storage/innobase/handler/handler0alter.cc b/storage/innobase/handler/handler0alter.cc
index ff069777ec4..0f3191f3f6e 100644
--- a/storage/innobase/handler/handler0alter.cc
+++ b/storage/innobase/handler/handler0alter.cc
@@ -1173,6 +1173,14 @@ struct ha_innobase_inplace_ctx : public inplace_alter_handler_ctx
         return true;
     return false;
   }
+
+  /** Handle the apply log failure for online DDL operation.
+  @param ha_alter_info	handler alter inplace info
+  @param altered_table	MySQL table that is being altered
+  @param error		error code */
+  void log_handle_failure(Alter_inplace_info* ha_alter_info,
+                          TABLE* altered_table,
+                          dberr_t error);
 };
 
 /********************************************************************//**
@@ -7011,6 +7019,7 @@ error_handling_drop_uncached:
 						row_log_free(
 							index->online_log);
 						index->online_log = NULL;
+						ctx->old_table->indexes.start->clear_dummy_log();
 						ok = false;
 					});
 
@@ -8344,6 +8353,45 @@ get_error_key_name(
 	}
 }
 
+/** Handle the apply log failure for online DDL operation.
+@param ha_alter_info	handler alter inplace info
+@param altered_table	MySQL table that is being altered
+@param error		error code */
+void ha_innobase_inplace_ctx::log_handle_failure(
+	Alter_inplace_info* ha_alter_info,
+	TABLE* altered_table,
+	dberr_t error)
+{
+  ulint err_key= thr_get_trx(thr)->error_key_num;
+  switch (error)
+  {
+    KEY* dup_key;
+    case DB_SUCCESS:
+      break;
+    case DB_DUPLICATE_KEY:
+      if (err_key == ULINT_UNDEFINED)
+        /* This should be the hidden index on FTS_DOC_ID */
+        dup_key= NULL;
+      else
+      {
+        DBUG_ASSERT(err_key < ha_alter_info->key_count);
+        dup_key= &ha_alter_info->key_info_buffer[err_key];
+      }
+      print_keydup_error(altered_table, dup_key, MYF(0));
+      break;
+    case DB_ONLINE_LOG_TOO_BIG:
+      my_error(ER_INNODB_ONLINE_LOG_TOO_BIG, MYF(0),
+               get_error_key_name(err_key, ha_alter_info, new_table));
+      break;
+    case DB_INDEX_CORRUPT:
+      my_error(ER_INDEX_CORRUPT, MYF(0),
+               get_error_key_name(err_key, ha_alter_info, new_table));
+      break;
+    default:
+      my_error_innodb(error, old_table->name.m_name, old_table->flags);
+  }
+}
+
 /** Alter the table structure in-place with operations
 specified using Alter_inplace_info.
 The level of concurrency allowed during this operation depends
@@ -8771,7 +8819,7 @@ inline bool rollback_inplace_alter_table(Alter_inplace_info *ha_alter_info,
         thd_lock_wait_timeout(ctx->trx->mysql_thd);
       const uint save_timeout= innodb_lock_wait_timeout;
       innodb_lock_wait_timeout= ~0U; /* infinite  */
-
+      ctx->old_table->indexes.start->clear_dummy_log();
       if (fts_exist)
       {
         const dict_index_t *fts_index= nullptr;
@@ -10734,7 +10782,6 @@ static bool alter_rebuild_apply_log(
 	dropped were not created in the copy of the table. Apply any
 	last bit of the rebuild log and then rename the tables. */
 	dict_table_t*	user_table = ctx->old_table;
-	dict_table_t*	rebuilt_table = ctx->new_table;
 
 	DEBUG_SYNC_C("row_log_table_apply2_before");
 
@@ -10763,41 +10810,12 @@ static bool alter_rebuild_apply_log(
 		ctx->new_table->vc_templ = NULL;
 	}
 
-	ulint	err_key = thr_get_trx(ctx->thr)->error_key_num;
-
-	switch (error) {
-		KEY*	dup_key;
-	case DB_SUCCESS:
-		break;
-	case DB_DUPLICATE_KEY:
-		if (err_key == ULINT_UNDEFINED) {
-			/* This should be the hidden index on
-			   FTS_DOC_ID. */
-			dup_key = NULL;
-		} else {
-			DBUG_ASSERT(err_key < ha_alter_info->key_count);
-			dup_key = &ha_alter_info->key_info_buffer[err_key];
-		}
-
-		print_keydup_error(altered_table, dup_key, MYF(0));
-		DBUG_RETURN(true);
-	case DB_ONLINE_LOG_TOO_BIG:
-		my_error(ER_INNODB_ONLINE_LOG_TOO_BIG, MYF(0),
-			 get_error_key_name(err_key, ha_alter_info,
-					    rebuilt_table));
-		DBUG_RETURN(true);
-	case DB_INDEX_CORRUPT:
-		my_error(ER_INDEX_CORRUPT, MYF(0),
-			 get_error_key_name(err_key, ha_alter_info,
-					    rebuilt_table));
-		DBUG_RETURN(true);
-	default:
-		my_error_innodb(error, ctx->old_table->name.m_name,
-				user_table->flags);
-		DBUG_RETURN(true);
+	if (error == DB_SUCCESS) {
+		DBUG_RETURN(false);
 	}
 
-	DBUG_RETURN(false);
+	ctx->log_handle_failure(ha_alter_info, altered_table, error);
+	DBUG_RETURN(true);
 }
 
 /** Commit or rollback the changes made during
@@ -10988,6 +11006,55 @@ lock_fail:
 				DBUG_RETURN(true);
 			}
 		}
+	} else {
+		for (inplace_alter_handler_ctx** pctx= ctx_array; *pctx;
+		     pctx++) {
+			auto ctx= static_cast<ha_innobase_inplace_ctx*>(*pctx);
+
+			if (!ctx->online || !ctx->old_table->space
+			    || !ctx->old_table->is_readable()) {
+				continue;
+			}
+
+			for (ulint i = 0; i < ctx->num_to_add_index; i++) {
+				dict_index_t *index= ctx->add_index[i];
+
+				if (index->type & (DICT_FTS | DICT_SPATIAL))
+                                  continue;
+
+				index->lock.x_lock(SRW_LOCK_CALL);
+				ut_ad(index->online_log);
+
+				dberr_t error = row_log_apply(
+					m_prebuilt->trx, index, altered_table,
+					ctx->m_stage);
+
+				if (error != DB_SUCCESS) {
+					ctx->log_handle_failure(
+						ha_alter_info,
+						altered_table, error);
+					row_log_free(index->online_log);
+					index->online_log= nullptr;
+					index->lock.x_unlock();
+
+					ctx->old_table->indexes.start->clear_dummy_log();
+					if (fts_exist) {
+						purge_sys.resume_FTS();
+					}
+					MONITOR_ATOMIC_INC(
+						MONITOR_BACKGROUND_DROP_INDEX);
+					DBUG_RETURN(true);
+				}
+
+				row_log_free(index->online_log);
+				index->online_log= nullptr;
+				dict_index_set_online_status(
+					index, ONLINE_INDEX_COMPLETE);
+				index->lock.x_unlock();
+			}
+
+			ctx->old_table->indexes.start->clear_dummy_log();
+		}
 	}
 
 	dict_table_t *table_stats = nullptr, *index_stats = nullptr;
diff --git a/storage/innobase/include/dict0mem.h b/storage/innobase/include/dict0mem.h
index dac738fbc68..4ea96601afd 100644
--- a/storage/innobase/include/dict0mem.h
+++ b/storage/innobase/include/dict0mem.h
@@ -2,7 +2,7 @@
 
 Copyright (c) 1996, 2017, Oracle and/or its affiliates. All Rights Reserved.
 Copyright (c) 2012, Facebook Inc.
-Copyright (c) 2013, 2021, MariaDB Corporation.
+Copyright (c) 2013, 2022, MariaDB Corporation.
 
 This program is free software; you can redistribute it and/or modify it under
 the terms of the GNU General Public License as published by the Free Software
@@ -1399,6 +1399,25 @@ public:
   rollback of TRX_UNDO_EMPTY. The BTR_SEG_LEAF is freed and reinitialized.
   @param thr query thread */
   void clear(que_thr_t *thr);
+
+  /** Assign the clustered index online log to table.
+  It can be used by concurrent DML to identify whether
+  the table has any active DDL */
+  void assign_dummy_log()
+  {
+    ut_ad(this->is_clust());
+    lock.s_lock(SRW_LOCK_CALL);
+    online_log= reinterpret_cast<row_log_t*>(table);
+    lock.s_unlock();
+  }
+
+  /** Clear the online log for the clustered index */
+  void clear_dummy_log()
+  {
+    lock.s_lock(SRW_LOCK_CALL);
+    online_log= nullptr;
+    lock.s_unlock();
+  }
 };
 
 /** Detach a virtual column from an index.
@@ -2360,6 +2379,12 @@ public:
     return false;
   }
 
+  /** @return whether a DDL operation is in progress on this table */
+  bool is_active_ddl() const
+  {
+    return UT_LIST_GET_FIRST(indexes)->online_log;
+  }
+
   /** @return whether the name is
   mysql.innodb_index_stats or mysql.innodb_table_stats */
   bool is_stats_table() const;
diff --git a/storage/innobase/include/row0log.h b/storage/innobase/include/row0log.h
index 732ef494326..82050769ca4 100644
--- a/storage/innobase/include/row0log.h
+++ b/storage/innobase/include/row0log.h
@@ -1,7 +1,7 @@
 /*****************************************************************************
 
 Copyright (c) 2011, 2016, Oracle and/or its affiliates. All Rights Reserved.
-Copyright (c) 2017, 2021, MariaDB Corporation.
+Copyright (c) 2017, 2022, MariaDB Corporation.
 
 This program is free software; you can redistribute it and/or modify it under
 the terms of the GNU General Public License as published by the Free Software
@@ -24,15 +24,15 @@ Modification log for online index creation and online table rebuild
 Created 2011-05-26 Marko Makela
 *******************************************************/
 
-#ifndef row0log_h
-#define row0log_h
+#pragma once
 
 #include "que0types.h"
 #include "mtr0types.h"
 #include "row0types.h"
 #include "rem0types.h"
-#include "data0types.h"
+#include "dict0dict.h"
 #include "trx0types.h"
+#include "trx0undo.h"
 
 class ut_stage_alter_t;
 
@@ -74,28 +74,16 @@ row_log_free(
 
 /******************************************************//**
 Free the row log for an index on which online creation was aborted. */
-UNIV_INLINE
-void
-row_log_abort_sec(
-/*==============*/
-	dict_index_t*	index)	/*!< in/out: index (x-latched) */
-	MY_ATTRIBUTE((nonnull));
+inline void row_log_abort_sec(dict_index_t *index)
+{
+  ut_ad(index->lock.have_u_or_x());
+  ut_ad(!index->is_clust());
+  dict_index_set_online_status(index, ONLINE_INDEX_ABORTED);
+  row_log_free(index->online_log);
+  index->online_log= nullptr;
+}
 
 /******************************************************//**
-Try to log an operation to a secondary index that is
-(or was) being created.
-@retval true if the operation was logged or can be ignored
-@retval false if online index creation is not taking place */
-UNIV_INLINE
-bool
-row_log_online_op_try(
-/*==================*/
-	dict_index_t*	index,	/*!< in/out: index, S or X latched */
-	const dtuple_t* tuple,	/*!< in: index tuple */
-	trx_id_t	trx_id)	/*!< in: transaction ID for insert,
-				or 0 for delete */
-	MY_ATTRIBUTE((nonnull, warn_unused_result));
-/******************************************************//**
 Logs an operation to a secondary index that is (or was) being created. */
 void
 row_log_online_op(
@@ -118,7 +106,7 @@ row_log_table_get_error(
 
 /** Check whether a virtual column is indexed in the new table being
 created during alter table
-@param[in]	index	cluster index
+@param[in]	index	clustered index
 @param[in]	v_no	virtual column number
 @return true if it is indexed, else false */
 bool
@@ -193,14 +181,6 @@ row_log_table_blob_free(
 	dict_index_t*	index,	/*!< in/out: clustered index, X-latched */
 	ulint		page_no)/*!< in: starting page number of the BLOB */
 	ATTRIBUTE_COLD __attribute__((nonnull));
-/******************************************************//**
-Notes that a BLOB is being allocated during online ALTER TABLE. */
-void
-row_log_table_blob_alloc(
-/*=====================*/
-	dict_index_t*	index,	/*!< in/out: clustered index, X-latched */
-	ulint		page_no)/*!< in: starting page number of the BLOB */
-	ATTRIBUTE_COLD __attribute__((nonnull));
 
 /** Apply the row_log_table log to a table upon completing rebuild.
 @param[in]	thr		query graph
@@ -261,8 +241,26 @@ of an ALTER TABLE for this index.
 ulint
 row_log_estimate_work(
 	const dict_index_t*	index);
+
 #endif /* HAVE_PSI_STAGE_INTERFACE */
 
-#include "row0log.inl"
+/** Handle the insert undo log and apply it on online indexes
+@param	tuple		row reference from undo log record
+@param	rec_info	undo log record info
+@param	clust_index	clustered index
+@param	heap		memory heap */
+void row_log_insert_handle(const dtuple_t *tuple,
+                           const trx_undo_rec_info *rec_info,
+                           dict_index_t *clust_index,
+                           mem_heap_t *heap);
 
-#endif /* row0log.h */
+/** Handle the update, delete undo log and apply it on online
+indexes
+@param	tuple		row reference from undo log record
+@param	rec_info	undo log record info
+@param	clust_index	clustered index
+@param	heap		memory heap */
+void row_log_update_handle(const dtuple_t *tuple,
+                           const trx_undo_rec_info *rec_info,
+                           dict_index_t *clust_index,
+                           mem_heap_t *heap);
diff --git a/storage/innobase/include/row0log.inl b/storage/innobase/include/row0log.inl
deleted file mode 100644
index f9f3dd006bf..00000000000
--- a/storage/innobase/include/row0log.inl
+++ /dev/null
@@ -1,80 +0,0 @@
-/*****************************************************************************
-
-Copyright (c) 2011, 2015, Oracle and/or its affiliates. All Rights Reserved.
-Copyright (c) 2020, MariaDB Corporation.
-
-This program is free software; you can redistribute it and/or modify it under
-the terms of the GNU General Public License as published by the Free Software
-Foundation; version 2 of the License.
-
-This program is distributed in the hope that it will be useful, but WITHOUT
-ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS
-FOR A PARTICULAR PURPOSE. See the GNU General Public License for more details.
-
-You should have received a copy of the GNU General Public License along with
-this program; if not, write to the Free Software Foundation, Inc.,
-51 Franklin Street, Fifth Floor, Boston, MA 02110-1335 USA
-
-*****************************************************************************/
-
-/**************************************************//**
-@file include/row0log.ic
-Modification log for online index creation and online table rebuild
-
-Created 2012-10-18 Marko Makela
-*******************************************************/
-
-#include "dict0dict.h"
-
-/******************************************************//**
-Free the row log for an index on which online creation was aborted. */
-UNIV_INLINE
-void
-row_log_abort_sec(
-/*===============*/
-	dict_index_t*	index)	/*!< in/out: index (x-latched) */
-{
-	ut_ad(index->lock.have_u_or_x());
-	ut_ad(!dict_index_is_clust(index));
-	dict_index_set_online_status(index, ONLINE_INDEX_ABORTED);
-	row_log_free(index->online_log);
-	index->online_log = NULL;
-}
-
-/******************************************************//**
-Try to log an operation to a secondary index that is
-(or was) being created.
-@retval true if the operation was logged or can be ignored
-@retval false if online index creation is not taking place */
-UNIV_INLINE
-bool
-row_log_online_op_try(
-/*==================*/
-	dict_index_t*	index,	/*!< in/out: index, S or X latched */
-	const dtuple_t* tuple,	/*!< in: index tuple */
-	trx_id_t	trx_id)	/*!< in: transaction ID for insert,
-				or 0 for delete */
-{
-	ut_ad(index->lock.have_any());
-
-	switch (dict_index_get_online_status(index)) {
-	case ONLINE_INDEX_COMPLETE:
-		/* This is a normal index. Do not log anything.
-		The caller must perform the operation on the
-		index tree directly. */
-		return(false);
-	case ONLINE_INDEX_CREATION:
-		/* The index is being created online. Log the
-		operation. */
-		row_log_online_op(index, tuple, trx_id);
-		break;
-	case ONLINE_INDEX_ABORTED:
-	case ONLINE_INDEX_ABORTED_DROPPED:
-		/* The index was created online, but the operation was
-		aborted. Do not log the operation and tell the caller
-		to skip the operation. */
-		break;
-	}
-
-	return(true);
-}
diff --git a/storage/innobase/include/trx0rec.h b/storage/innobase/include/trx0rec.h
index 0afae491c78..186d9410dfc 100644
--- a/storage/innobase/include/trx0rec.h
+++ b/storage/innobase/include/trx0rec.h
@@ -82,6 +82,7 @@ trx_undo_rec_get_pars(
 	undo_no_t*	undo_no,	/*!< out: undo log record number */
 	table_id_t*	table_id)	/*!< out: table id */
 	MY_ATTRIBUTE((nonnull));
+
 /*******************************************************************//**
 Builds a row reference from an undo log record.
 @return pointer to remaining part of undo record */
@@ -208,37 +209,47 @@ fetching the purge record */
 the undo log (which is the after image for an update) */
 #define		TRX_UNDO_GET_OLD_V_VALUE	0x2
 
-/*******************************************************************//**
-Build a previous version of a clustered index record. The caller must
-hold a latch on the index page of the clustered index record.
+/** Build a previous version of a clustered index record. The caller
+must hold a latch on the index page of the clustered index record.
+@param	index_rec	clustered index record in the index tree
+@param	index_mtr	mtr which contains the latch to index_rec page
+			and purge_view
+@param	rec		version of a clustered index record
+@param	index		clustered index
+@param	offsets		rec_get_offsets(rec, index)
+@param	heap		memory heap from which the memory needed is
+			allocated
+@param	old_vers	previous version or NULL if rec is the
+			first inserted version, or if history data
+			has been deleted (an error), or if the purge
+			could have removed the version
+			though it has not yet done so
+@param	v_heap		memory heap used to create vrow
+			dtuple if it is not yet created. This heap
+                        diffs from "heap" above in that it could be
+                        prebuilt->old_vers_heap for selection
+@param	vrow		virtual column info, if any
+@param	v_status	status determine if it is going into this
+			function by purge thread or not.
+			And if we read "after image" of undo log
+@param	undo_rec_info	undo record info or NULL
 @retval true if previous version was built, or if it was an insert
 or the table has been rebuilt
 @retval false if the previous version is earlier than purge_view,
-which means that it may have been removed */
+or being purged, which means that it may have been removed */
 bool
 trx_undo_prev_version_build(
-/*========================*/
-	const rec_t*	index_rec,/*!< in: clustered index record in the
-				index tree */
-	mtr_t*		index_mtr,/*!< in: mtr which contains the latch to
-				index_rec page and purge_view */
-	const rec_t*	rec,	/*!< in: version of a clustered index record */
-	dict_index_t*	index,	/*!< in: clustered index */
-	rec_offs*	offsets,/*!< in/out: rec_get_offsets(rec, index) */
-	mem_heap_t*	heap,	/*!< in: memory heap from which the memory
-				needed is allocated */
-	rec_t**		old_vers,/*!< out, own: previous version, or NULL if
-				rec is the first inserted version, or if
-				history data has been deleted */
-	mem_heap_t*	v_heap,	/* !< in: memory heap used to create vrow
-				dtuple if it is not yet created. This heap
-				diffs from "heap" above in that it could be
-				prebuilt->old_vers_heap for selection */
-	dtuple_t**	vrow,	/*!< out: virtual column info, if any */
-	ulint		v_status);
-				/*!< in: status determine if it is going
-				into this function by purge thread or not.
-				And if we read "after image" of undo log */
+	const rec_t	*index_rec,
+	mtr_t		*index_mtr,
+	const rec_t 	*rec,
+	dict_index_t	*index,
+	rec_offs	*offsets,
+	mem_heap_t	*heap,
+	rec_t		**old_vers,
+	mem_heap_t	*v_heap,
+	dtuple_t	**vrow,
+	ulint		v_status,
+	const trx_undo_rec_info *undo_rec_info= nullptr);
 
 /** Read from an undo log record a non-virtual column value.
 @param[in,out]	ptr		pointer to remaining part of the undo record
@@ -281,6 +292,13 @@ trx_undo_read_v_idx(
 	bool*			is_undo_log,
 	uint32_t*		field_no);
 
+/** Check whether the given roll ptr matches with the undo log
+@param	roll_ptr	rollback pointer of the record
+@param	undo_info	undo log record information
+@return true if the undo record matches with rollback pointer */
+bool trx_undo_rec_is_equal(roll_ptr_t roll_ptr,
+                           const trx_undo_rec_info *undo_info);
+
 /* Types of an undo log record: these have to be smaller than 16, as the
 compilation info multiplied by 16 is ORed to this value in an undo log
 record */
diff --git a/storage/innobase/include/trx0trx.h b/storage/innobase/include/trx0trx.h
index 6b9ffdc6374..c42579166dd 100644
--- a/storage/innobase/include/trx0trx.h
+++ b/storage/innobase/include/trx0trx.h
@@ -860,6 +860,10 @@ public:
 	bool		auto_commit;	/*!< true if it is an autocommit */
 	bool		will_lock;	/*!< set to inform trx_start_low() that
 					the transaction may acquire locks */
+	/* True if transaction has to read the undo log and
+	log the DML changes for online DDL table */
+	bool		apply_online_log = false;
+
 	/*------------------------------*/
 	fts_trx_t*	fts_trx;	/*!< FTS information, or NULL if
 					transaction hasn't modified tables
@@ -885,6 +889,7 @@ public:
 					error, or empty. */
 	rw_trx_hash_element_t *rw_trx_hash_element;
 	LF_PINS *rw_trx_hash_pins;
+
 	ulint		magic_n;
 
 	/** @return whether any persistent undo log has been generated */
@@ -933,6 +938,8 @@ public:
   @retval false if the rollback was aborted by shutdown */
   inline bool rollback_finish();
 private:
+  /** Apply any changes to tables for which online DDL is in progress. */
+  ATTRIBUTE_COLD void apply_log() const;
   /** Process tables that were modified by the committing transaction. */
   inline void commit_tables();
   /** Mark a transaction committed in the main memory data structures. */
diff --git a/storage/innobase/include/trx0undo.h b/storage/innobase/include/trx0undo.h
index 62662ffe221..3de8640606a 100644
--- a/storage/innobase/include/trx0undo.h
+++ b/storage/innobase/include/trx0undo.h
@@ -326,6 +326,34 @@ struct trx_undo_t {
 					/*!< undo log objects in the rollback
 					segment are chained into lists */
 };
+
+/** Undo record information like rollback segment id, page_id, offset */
+struct trx_undo_rec_info
+{
+  trx_id_t trx_id;
+  buf_block_t *block;
+  ulint offset;
+  ulint type;
+  ulint cmpl_info;
+  bool updated_extern;
+  undo_no_t undo_no;
+  trx_undo_rec_t *undo_rec;
+  upd_t *update= nullptr;
+
+  trx_undo_rec_info(trx_id_t trx_id_, buf_block_t *block_,
+                    ulint offset_):
+                    trx_id(trx_id_), block(block_), offset(offset_) {}
+
+  void assign_value(ulint type_, ulint cmpl_info_, bool updated_ext,
+                    undo_no_t undo_no_)
+  {
+    type= type_;
+    cmpl_info= cmpl_info_;
+    updated_extern= updated_ext;
+    undo_no= undo_no_;
+  }
+};
+
 #endif /* !UNIV_INNOCHECKSUM */
 
 /** The offset of the undo log page header on pages of the undo log */
diff --git a/storage/innobase/row/row0ins.cc b/storage/innobase/row/row0ins.cc
index 911e33875a1..8daa4f7e8b7 100644
--- a/storage/innobase/row/row0ins.cc
+++ b/storage/innobase/row/row0ins.cc
@@ -1,7 +1,7 @@
 /*****************************************************************************
 
 Copyright (c) 1996, 2016, Oracle and/or its affiliates. All Rights Reserved.
-Copyright (c) 2016, 2021, MariaDB Corporation.
+Copyright (c) 2016, 2022, MariaDB Corporation.
 
 This program is free software; you can redistribute it and/or modify it under
 the terms of the GNU General Public License as published by the Free Software
@@ -35,7 +35,6 @@ Created 4/20/1996 Heikki Tuuri
 #include "que0que.h"
 #include "row0upd.h"
 #include "row0sel.h"
-#include "row0log.h"
 #include "rem0cmp.h"
 #include "lock0lock.h"
 #include "log0log.h"
@@ -2025,7 +2024,6 @@ row_ins_scan_sec_index_for_duplicate(
 	dict_index_t*	index,	/*!< in: non-clustered unique index */
 	dtuple_t*	entry,	/*!< in: index entry */
 	que_thr_t*	thr,	/*!< in: query thread */
-	bool		s_latch,/*!< in: whether index->lock is being held */
 	mtr_t*		mtr,	/*!< in/out: mini-transaction */
 	mem_heap_t*	offsets_heap)
 				/*!< in/out: memory heap that can be emptied */
@@ -2042,7 +2040,7 @@ row_ins_scan_sec_index_for_duplicate(
 
 	rec_offs_init(offsets_);
 
-	ut_ad(s_latch == (index->lock.have_u_not_x() || index->lock.have_s()));
+	ut_ad(!index->lock.have_any());
 
 	n_unique = dict_index_get_n_unique(index);
 
@@ -2066,11 +2064,7 @@ row_ins_scan_sec_index_for_duplicate(
 
 	dtuple_set_n_fields_cmp(entry, n_unique);
 
-	btr_pcur_open(index, entry, PAGE_CUR_GE,
-		      s_latch
-		      ? BTR_SEARCH_LEAF_ALREADY_S_LATCHED
-		      : BTR_SEARCH_LEAF,
-		      &pcur, mtr);
+	btr_pcur_open(index, entry, PAGE_CUR_GE, BTR_SEARCH_LEAF, &pcur, mtr);
 
 	allow_duplicates = thr_get_trx(thr)->duplicates;
 
@@ -2484,11 +2478,6 @@ row_ins_index_entry_big_rec(
 		&pcur, offsets, big_rec, &mtr, BTR_STORE_INSERT);
 	DEBUG_SYNC_C_IF_THD(thd, "after_row_ins_extern");
 
-	if (error == DB_SUCCESS
-	    && dict_index_is_online_ddl(index)) {
-		row_log_table_insert(btr_pcur_get_rec(&pcur), index, offsets);
-	}
-
 	mtr.commit();
 
 	btr_pcur_close(&pcur);
@@ -2742,11 +2731,6 @@ err_exit:
 			&pcur, flags, mode, &offsets, &offsets_heap,
 			entry_heap, entry, thr, &mtr);
 
-		if (err == DB_SUCCESS && dict_index_is_online_ddl(index)) {
-			row_log_table_insert(btr_cur_get_rec(cursor),
-					     index, offsets);
-		}
-
 		mtr_commit(&mtr);
 		mem_heap_free(entry_heap);
 	} else {
@@ -2784,9 +2768,9 @@ do_insert:
 			}
 		}
 
-		if (big_rec != NULL) {
-			mtr_commit(&mtr);
+		mtr.commit();
 
+		if (big_rec) {
 			/* Online table rebuild could read (and
 			ignore) the incomplete record at this point.
 			If online rebuild is in progress, the
@@ -2799,14 +2783,6 @@ do_insert:
 				entry, big_rec, offsets, &offsets_heap, index,
 				trx->mysql_thd);
 			dtuple_convert_back_big_rec(index, entry, big_rec);
-		} else {
-			if (err == DB_SUCCESS
-			    && dict_index_is_online_ddl(index)) {
-				row_log_table_insert(
-					insert_rec, index, offsets);
-			}
-
-			mtr_commit(&mtr);
 		}
 	}
 
@@ -2820,19 +2796,10 @@ func_exit:
 	DBUG_RETURN(err);
 }
 
-/** Start a mini-transaction and check if the index will be dropped.
+/** Start a mini-transaction.
 @param[in,out]	mtr		mini-transaction
-@param[in,out]	index		secondary index
-@param[in]	check		whether to check
-@param[in]	search_mode	flags
-@return true if the index is to be dropped */
-static MY_ATTRIBUTE((warn_unused_result))
-bool
-row_ins_sec_mtr_start_and_check_if_aborted(
-	mtr_t*		mtr,
-	dict_index_t*	index,
-	bool		check,
-	ulint		search_mode)
+@param[in,out]	index		secondary index */
+static void row_ins_sec_mtr_start(mtr_t *mtr, dict_index_t *index)
 {
 	ut_ad(!dict_index_is_clust(index));
 	ut_ad(mtr->is_named_space(index->table->space));
@@ -2842,30 +2809,6 @@ row_ins_sec_mtr_start_and_check_if_aborted(
 	mtr->start();
 	index->set_modified(*mtr);
 	mtr->set_log_mode(log_mode);
-
-	if (!check) {
-		return(false);
-	}
-
-	if (search_mode & BTR_ALREADY_S_LATCHED) {
-		mtr_s_lock_index(index, mtr);
-	} else {
-		mtr_sx_lock_index(index, mtr);
-	}
-
-	switch (index->online_status) {
-	case ONLINE_INDEX_ABORTED:
-	case ONLINE_INDEX_ABORTED_DROPPED:
-		ut_ad(!index->is_committed());
-		return(true);
-	case ONLINE_INDEX_COMPLETE:
-		return(false);
-	case ONLINE_INDEX_CREATION:
-		break;
-	}
-
-	ut_error;
-	return(true);
 }
 
 /***************************************************************//**
@@ -2925,28 +2868,6 @@ row_ins_sec_index_entry_low(
 		}
 	}
 
-	/* Ensure that we acquire index->lock when inserting into an
-	index with index->online_status == ONLINE_INDEX_COMPLETE, but
-	could still be subject to rollback_inplace_alter_table().
-	This prevents a concurrent change of index->online_status.
-	The memory object cannot be freed as long as we have an open
-	reference to the table, or index->table->n_ref_count > 0. */
-	const bool	check = !index->is_committed();
-	if (check) {
-		DEBUG_SYNC_C("row_ins_sec_index_enter");
-		if (mode == BTR_MODIFY_LEAF) {
-			search_mode |= BTR_ALREADY_S_LATCHED;
-			mtr_s_lock_index(index, &mtr);
-		} else {
-			mtr_sx_lock_index(index, &mtr);
-		}
-
-		if (row_log_online_op_try(
-			    index, entry, thr_get_trx(thr)->id)) {
-			goto func_exit;
-		}
-	}
-
 	/* Note that we use PAGE_CUR_LE as the search mode, because then
 	the function will return in both low_match and up_match of the
 	cursor sensible values */
@@ -3031,13 +2952,10 @@ row_ins_sec_index_entry_low(
 
 		DEBUG_SYNC_C("row_ins_sec_index_unique");
 
-		if (row_ins_sec_mtr_start_and_check_if_aborted(
-			    &mtr, index, check, search_mode)) {
-			goto func_exit;
-		}
+		row_ins_sec_mtr_start(&mtr, index);
 
 		err = row_ins_scan_sec_index_for_duplicate(
-			flags, index, entry, thr, check, &mtr, offsets_heap);
+			flags, index, entry, thr, &mtr, offsets_heap);
 
 		mtr_commit(&mtr);
 
@@ -3066,10 +2984,7 @@ row_ins_sec_index_entry_low(
 			DBUG_RETURN(err);
 		}
 
-		if (row_ins_sec_mtr_start_and_check_if_aborted(
-			    &mtr, index, check, search_mode)) {
-			goto func_exit;
-		}
+		row_ins_sec_mtr_start(&mtr, index);
 
 		DEBUG_SYNC_C("row_ins_sec_index_entry_dup_locks_created");
 
@@ -3664,7 +3579,8 @@ row_ins(
 		   FTS_DOC_ID for history is enough.
 		*/
 		const unsigned type = index->type;
-		if (index->type & DICT_FTS) {
+		if (index->type & DICT_FTS
+		    || !index->is_committed()) {
 		} else if (!(type & DICT_UNIQUE) || index->n_uniq > 1
 			   || !node->vers_history_row()) {
 
diff --git a/storage/innobase/row/row0log.cc b/storage/innobase/row/row0log.cc
index 610ac1ad2ae..0ae4c805932 100644
--- a/storage/innobase/row/row0log.cc
+++ b/storage/innobase/row/row0log.cc
@@ -1,7 +1,7 @@
 /*****************************************************************************
 
 Copyright (c) 2011, 2018, Oracle and/or its affiliates. All Rights Reserved.
-Copyright (c) 2017, 2021, MariaDB Corporation.
+Copyright (c) 2017, 2022, MariaDB Corporation.
 
 This program is free software; you can redistribute it and/or modify it under
 the terms of the GNU General Public License as published by the Free Software
@@ -660,16 +660,16 @@ err_exit:
 
 /** Check whether a virtual column is indexed in the new table being
 created during alter table
-@param[in]	index	cluster index
-@param[in]	v_no	virtual column number
+@param[in]      index   cluster index
+@param[in]      v_no    virtual column number
 @return true if it is indexed, else false */
 bool
 row_log_col_is_indexed(
-	const dict_index_t*	index,
-	ulint			v_no)
+        const dict_index_t*     index,
+        ulint                   v_no)
 {
-	return(dict_table_get_nth_v_col(
-		index->online_log->table, v_no)->m_col.ord_part);
+        return(dict_table_get_nth_v_col(
+                index->online_log->table, v_no)->m_col.ord_part);
 }
 
 /******************************************************//**
@@ -833,7 +833,6 @@ row_log_table_low_redundant(
 	dtuple_t*	tuple;
 	const ulint	n_fields = rec_get_n_fields_old(rec);
 
-	ut_ad(!page_is_comp(page_align(rec)));
 	ut_ad(index->n_fields >= n_fields);
 	ut_ad(index->n_fields == n_fields || index->is_instant());
 	ut_ad(dict_tf2_is_valid(index->table->flags, index->table->flags2));
@@ -994,22 +993,6 @@ row_log_table_low(
 	ut_ad(rec_offs_size(offsets) <= sizeof log->tail.buf);
 	ut_ad(index->lock.have_any());
 
-#ifdef UNIV_DEBUG
-	switch (fil_page_get_type(page_align(rec))) {
-	case FIL_PAGE_INDEX:
-		break;
-	case FIL_PAGE_TYPE_INSTANT:
-		ut_ad(index->is_instant());
-		ut_ad(!page_has_siblings(page_align(rec)));
-		ut_ad(page_get_page_no(page_align(rec)) == index->page);
-		break;
-	default:
-		ut_ad("wrong page type" == 0);
-	}
-#endif /* UNIV_DEBUG */
-	ut_ad(!rec_is_metadata(rec, *index));
-	ut_ad(page_rec_is_leaf(rec));
-	ut_ad(!page_is_comp(page_align(rec)) == !rec_offs_comp(offsets));
 	/* old_pk=row_log_table_get_pk() [not needed in INSERT] is a prefix
 	of the clustered index record (PRIMARY KEY,DB_TRX_ID,DB_ROLL_PTR),
 	with no information on virtual columns */
@@ -1028,7 +1011,6 @@ row_log_table_low(
 		return;
 	}
 
-	ut_ad(page_is_comp(page_align(rec)));
 	ut_ad(rec_get_status(rec) == REC_STATUS_ORDINARY
 	      || rec_get_status(rec) == REC_STATUS_INSTANT);
 
@@ -1513,36 +1495,6 @@ row_log_table_blob_free(
 }
 
 /******************************************************//**
-Notes that a BLOB is being allocated during online ALTER TABLE. */
-void
-row_log_table_blob_alloc(
-/*=====================*/
-	dict_index_t*	index,	/*!< in/out: clustered index, X-latched */
-	ulint		page_no)/*!< in: starting page number of the BLOB */
-{
-	ut_ad(dict_index_is_clust(index));
-	ut_ad(dict_index_is_online_ddl(index));
-	ut_ad(index->lock.have_u_or_x());
-
-	ut_ad(page_no != FIL_NULL);
-
-	if (index->online_log->error != DB_SUCCESS) {
-		return;
-	}
-
-	/* Only track allocations if the same page has been freed
-	earlier. Double allocation without a free is not allowed. */
-	if (page_no_map* blobs = index->online_log->blobs) {
-		page_no_map::iterator p = blobs->find(page_no);
-
-		if (p != blobs->end()) {
-			ut_ad(p->first == page_no);
-			p->second.blob_alloc(index->online_log->tail.total);
-		}
-	}
-}
-
-/******************************************************//**
 Converts a log record to a table row.
 @return converted row, or NULL if the conversion fails */
 static MY_ATTRIBUTE((nonnull, warn_unused_result))
@@ -1693,6 +1645,12 @@ blob_done:
 		if ((new_col->prtype & DATA_NOT_NULL)
 		    && dfield_is_null(dfield)) {
 
+			if (!log->allow_not_null) {
+				/* We got a NULL value for a NOT NULL column. */
+				*error = DB_INVALID_NULL;
+				return NULL;
+			}
+
 			const dfield_t& default_field
 				= log->defaults->fields[col_no];
 
@@ -1702,12 +1660,6 @@ blob_done:
 					   WARN_DATA_TRUNCATED, 1,
 					   ulong(log->n_rows));
 
-			if (!log->allow_not_null) {
-				/* We got a NULL value for a NOT NULL column. */
-				*error = DB_INVALID_NULL;
-				return NULL;
-			}
-
 			*dfield = default_field;
 		}
 
@@ -2739,7 +2691,8 @@ ulint
 row_log_estimate_work(
 	const dict_index_t*	index)
 {
-	if (index == NULL || index->online_log == NULL) {
+	if (index == NULL || index->online_log == NULL
+	    || index->online_log == reinterpret_cast<row_log_t*>(index->table)) {
 		return(0);
 	}
 
@@ -3280,6 +3233,14 @@ row_log_allocate(
 	}
 
 	index->online_log = log;
+
+	if (!table) {
+		/* Assign the clustered index online log to table.
+		It can be used by concurrent DML to identify whether
+		the table has any online DDL */
+		index->table->indexes.start->assign_dummy_log();
+	}
+
 	/* While we might be holding an exclusive data dictionary lock
 	here, in row_log_abort_sec() we will not always be holding it. Use
 	atomic operations in both cases. */
@@ -3792,6 +3753,8 @@ corruption:
 			/* End of log reached. */
 all_done:
 			ut_ad(has_index_lock);
+			index->online_log->tail.bytes = 0;
+			index->online_log->head.bytes = 0;
 			ut_ad(index->online_log->head.blocks == 0);
 			ut_ad(index->online_log->tail.blocks == 0);
 			error = DB_SUCCESS;
@@ -4039,7 +4002,6 @@ row_log_apply(
 	ut_stage_alter_t*	stage)
 {
 	dberr_t		error;
-	row_log_t*	log;
 	row_merge_dup_t	dup = { index, table, NULL, 0 };
 	DBUG_ENTER("row_log_apply");
 
@@ -4067,17 +4029,10 @@ row_log_apply(
 		index->table->drop_aborted = TRUE;
 
 		dict_index_set_online_status(index, ONLINE_INDEX_ABORTED);
-	} else {
-		ut_ad(dup.n_dup == 0);
-		dict_index_set_online_status(index, ONLINE_INDEX_COMPLETE);
 	}
 
-	log = index->online_log;
-	index->online_log = NULL;
 	index->lock.x_unlock();
 
-	row_log_free(log);
-
 	DBUG_RETURN(error);
 }
 
@@ -4138,3 +4093,292 @@ void dict_table_t::clear(que_thr_t *thr)
     index->clear(thr);
   }
 }
+
+/** Get the version of clustered index record which was modified
+by undo log record
+@param	rec		clustered index record
+@param	index		clustered index
+@param	offsets		rec_get_offsets(rec, index)
+@param	rec_info	undo log record information
+@param	mtr		mtr which contains the latch to rec page
+@param	heap		memory heap
+@return version of clustered index record */
+static rec_t *row_log_vers_undo_match(
+	rec_t *rec, dict_index_t *index,
+	rec_offs **offsets, const trx_undo_rec_info *rec_info,
+	mtr_t *mtr, mem_heap_t *heap)
+{
+  ut_ad(dict_index_is_clust(index));
+  ulint len= 0;
+  rec_t *prev_version;
+  rec_t *version= rec;
+  bool match= false;
+  while (!match && version != nullptr)
+  {
+    *offsets= rec_get_offsets(version, index, *offsets,
+                              index->n_core_fields, ULINT_UNDEFINED,
+                              &heap);
+    trx_id_t trx_id= trx_read_trx_id(
+      rec_get_nth_field(version, *offsets, index->db_trx_id(), &len));
+
+    ut_ad(len == DATA_TRX_ID_LEN);
+    ut_ad(trx_id == rec_info->trx_id);
+
+    roll_ptr_t roll_ptr= trx_read_roll_ptr(
+      rec_get_nth_field(version, *offsets, index->db_roll_ptr(), &len));
+    ut_ad(len == DATA_ROLL_PTR_LEN);
+    match= trx_undo_rec_is_equal(roll_ptr, rec_info);
+    if (!match)
+    {
+      trx_undo_prev_version_build(rec, mtr, version, index,
+                                  *offsets, heap, &prev_version, NULL,
+                                  NULL, 0, rec_info);
+      version= prev_version;
+    }
+  }
+
+  return version;
+}
+
+/** Get the clustered index record version which was modified by
+undo log record
+@param		tuple		tuple contains primary key value
+@param		index		clustered index
+@param[out]	clust_rec	clustered index record
+@param		offsets		offsets points to the record
+@param		rec_info	undo log record info
+@param		mtr		mtr
+@param		heap		memory heap
+@return clustered index record */
+static rec_t* row_log_undo_vers_record(const dtuple_t *tuple,
+                                       dict_index_t *index,
+	                               rec_t **clust_rec,
+                                       rec_offs **offsets,
+                                       const trx_undo_rec_info *rec_info,
+                                       mtr_t *mtr, mem_heap_t *heap)
+{
+  ut_ad(dict_index_is_clust(index));
+  btr_pcur_t pcur;
+  bool found= row_search_on_row_ref(&pcur, BTR_MODIFY_LEAF,
+                                    index->table, tuple, mtr);
+  ut_ad(found);
+  *clust_rec= btr_pcur_get_rec(&pcur);
+
+  rec_t *match_rec= row_log_vers_undo_match(*clust_rec, index, offsets,
+                                            rec_info, mtr, heap);
+  return match_rec;
+}
+
+void row_log_insert_handle(const dtuple_t *tuple,
+                           const trx_undo_rec_info *rec_info,
+                           dict_index_t *clust_index,
+                           mem_heap_t *heap)
+{
+  DEBUG_SYNC_C("row_log_insert_handle");
+  ut_ad(dict_index_is_clust(clust_index));
+  rec_offs offsets_[REC_OFFS_NORMAL_SIZE];
+  rec_offs *offsets= offsets_;
+  mtr_t mtr;
+
+  rec_offs_init(offsets_);
+  mtr.start();
+  rec_t *rec;
+  rec_t *match_rec= row_log_undo_vers_record(tuple, clust_index, &rec,
+                                             &offsets,
+                                             rec_info, &mtr, heap);
+  if (!match_rec)
+  {
+    mtr.commit();
+    return;
+  }
+
+  if (match_rec != rec)
+    offsets= rec_get_offsets(match_rec, clust_index, offsets,
+                             clust_index->n_core_fields,
+                             ULINT_UNDEFINED, &heap);
+
+  rec_t *copy_rec= rec_copy(mem_heap_alloc(
+    heap, rec_offs_size(offsets)), match_rec, offsets);
+  rec_offs_make_valid(copy_rec, clust_index, true, offsets);
+  mtr.commit();
+
+  dict_table_t *table= clust_index->table;
+  clust_index->lock.s_lock(SRW_LOCK_CALL);
+  if (clust_index->online_log
+      && clust_index->online_log != reinterpret_cast<row_log_t*>(table)
+      && clust_index->online_status <= ONLINE_INDEX_CREATION)
+  {
+    row_log_table_insert(copy_rec, clust_index, offsets);
+    clust_index->lock.s_unlock();
+  }
+  else
+  {
+    clust_index->lock.s_unlock();
+    ulint len= 0;
+    trx_id_t trx_id= trx_read_trx_id(
+      rec_get_nth_field(copy_rec, offsets, clust_index->db_trx_id(), &len));
+    ut_ad(len == DATA_TRX_ID_LEN);
+    row_ext_t *ext;
+    dtuple_t *row= row_build(ROW_COPY_POINTERS, clust_index,
+      copy_rec, offsets, table, NULL, NULL, &ext, heap);
+
+    if (table->n_v_cols)
+      trx_undo_read_v_cols(table, rec_info->undo_rec, row, false);
+
+    dict_index_t *index= dict_table_get_next_index(clust_index);
+    while (index)
+    {
+      index->lock.s_lock(SRW_LOCK_CALL);
+      if (index->online_log
+          && index->online_status <= ONLINE_INDEX_CREATION)
+      {
+        dtuple_t *entry= row_build_index_entry_low(row, ext, index,
+                                                   heap, ROW_BUILD_NORMAL);
+        row_log_online_op(index, entry, trx_id);
+      }
+
+      index->lock.s_unlock();
+      index= dict_table_get_next_index(index);
+    }
+  }
+}
+
+void row_log_update_handle(const dtuple_t *tuple,
+                           const trx_undo_rec_info *rec_info,
+                           dict_index_t *clust_index,
+                           mem_heap_t *heap)
+{
+  rec_offs offsets_[REC_OFFS_NORMAL_SIZE];
+  rec_offs offsets2_[REC_OFFS_NORMAL_SIZE];
+  rec_offs *offsets= offsets_;
+  rec_offs *prev_offsets= offsets2_;
+  mtr_t mtr;
+
+  rec_offs_init(offsets_);
+  rec_offs_init(offsets2_);
+
+  dict_table_t *table= clust_index->table;
+
+  clust_index->lock.s_lock(SRW_LOCK_CALL);
+  bool table_rebuild=
+    (clust_index->online_log
+     && clust_index->online_log != reinterpret_cast<row_log_t*>(table)
+     && clust_index->online_status <= ONLINE_INDEX_CREATION);
+  clust_index->lock.s_unlock();
+
+  mtr.start();
+  rec_t *rec;
+  rec_t *prev_version;
+  bool is_update= (rec_info->type == TRX_UNDO_UPD_EXIST_REC);
+  rec_t *match_rec= row_log_undo_vers_record(
+    tuple, clust_index, &rec, &offsets, rec_info, &mtr, heap);
+  if (!match_rec)
+  {
+    mtr.commit();
+    return;
+  }
+
+  rec_t *copy_rec= rec_copy(mem_heap_alloc(
+              heap, rec_offs_size(offsets)), match_rec, offsets);
+
+  if (table_rebuild)
+  {
+    trx_undo_prev_version_build(rec, &mtr, match_rec, clust_index,
+                                offsets, heap, &prev_version, NULL,
+                                NULL, 0, rec_info);
+
+    prev_offsets= rec_get_offsets(prev_version, clust_index, prev_offsets,
+                                  clust_index->n_core_fields,
+                                  ULINT_UNDEFINED, &heap);
+
+    offsets= rec_get_offsets(match_rec, clust_index, offsets,
+                             clust_index->n_core_fields,
+                             ULINT_UNDEFINED, &heap);
+
+    rec_t *prev_copy_rec= rec_copy(mem_heap_alloc(
+      heap, rec_offs_size(prev_offsets)), prev_version, prev_offsets);
+    rec_offs_make_valid(prev_copy_rec, clust_index, true, prev_offsets);
+    rec_offs_make_valid(copy_rec, clust_index, true, offsets);
+    mtr.commit();
+
+    clust_index->lock.s_lock(SRW_LOCK_CALL);
+    if (is_update)
+    {
+      const dtuple_t *rebuilt_old_pk= row_log_table_get_pk(
+        prev_copy_rec, clust_index, prev_offsets, NULL, &heap);
+      row_log_table_update(copy_rec, clust_index, offsets, rebuilt_old_pk);
+    }
+    else
+      row_log_table_delete(prev_copy_rec, clust_index, prev_offsets, nullptr);
+    clust_index->lock.s_unlock();
+    return;
+  }
+
+  rec_offs_make_valid(copy_rec, clust_index, true, offsets);
+  mtr.commit();
+  dtuple_t *row= nullptr;
+  trx_id_t trx_id;
+  row_ext_t *new_ext;
+  row_ext_t *old_ext;
+  dtuple_t *old_row= nullptr;
+  ulint len;
+
+  trx_id= trx_read_trx_id(
+    rec_get_nth_field(
+      copy_rec, offsets, clust_index->db_trx_id(), &len));
+  ut_ad(len == DATA_TRX_ID_LEN);
+  row= row_build(ROW_COPY_POINTERS, clust_index,
+                 copy_rec, offsets, clust_index->table, NULL, NULL,
+                 &new_ext, heap);
+  if (table->n_v_cols
+      && !(rec_info->cmpl_info & UPD_NODE_NO_ORD_CHANGE))
+  {
+    for (ulint i = 0; i < dict_table_get_n_v_cols(table); i++)
+       dfield_get_type(
+         dtuple_get_nth_v_field(row, i))->mtype = DATA_MISSING;
+  }
+
+  if (is_update)
+  {
+    old_row= dtuple_copy(row, heap);
+    row_upd_replace(old_row, &old_ext, clust_index, rec_info->update,
+		    heap);
+  }
+
+  if (table->n_v_cols)
+    row_upd_replace_vcol(row, table, rec_info->update, false,
+                         nullptr,
+                         (rec_info->cmpl_info & UPD_NODE_NO_ORD_CHANGE)
+	                 ? nullptr : rec_info->undo_rec);
+
+  dict_index_t *index= dict_table_get_next_index(clust_index);
+  while (index)
+  {
+    index->lock.s_lock(SRW_LOCK_CALL);
+    if (index->online_log
+        && index->online_status <= ONLINE_INDEX_CREATION)
+    {
+      if (is_update)
+      {
+        dtuple_t *old_entry= row_build_index_entry_low(
+           old_row, old_ext, index, heap, ROW_BUILD_NORMAL);
+
+        row_log_online_op(index, old_entry, 0);
+
+	dtuple_t *new_entry= row_build_index_entry_low(
+           row, new_ext, index, heap, ROW_BUILD_NORMAL);
+
+	row_log_online_op(index, new_entry, trx_id);
+      }
+      else
+      {
+        dtuple_t *old_entry= row_build_index_entry_low(
+           row, new_ext, index, heap, ROW_BUILD_NORMAL);
+
+        row_log_online_op(index, old_entry, 0);
+      }
+    }
+    index->lock.s_unlock();
+    index= dict_table_get_next_index(index);
+  }
+}
diff --git a/storage/innobase/row/row0merge.cc b/storage/innobase/row/row0merge.cc
index a515d7d7e56..2858a8b4bd0 100644
--- a/storage/innobase/row/row0merge.cc
+++ b/storage/innobase/row/row0merge.cc
@@ -3629,11 +3629,7 @@ row_merge_insert_index_tuples(
 
 			Any modifications after the
 			row_merge_read_clustered_index() scan
-			will go through row_log_table_apply().
-			Any modifications to off-page columns
-			will be tracked by
-			row_log_table_blob_alloc() and
-			row_log_table_blob_free(). */
+			will go through row_log_table_apply(). */
 			row_merge_copy_blobs(
 				mrec, offsets, old_table->space->zip_size(),
 				dtuple, tuple_heap);
@@ -4806,6 +4802,7 @@ func_exit:
 				row_log_abort_sec(indexes[i]);
 				indexes[i]->type |= DICT_CORRUPT;
 				indexes[i]->lock.x_unlock();
+				new_table->indexes.start->online_log= nullptr;
 				new_table->drop_aborted = TRUE;
 				/* fall through */
 			case ONLINE_INDEX_ABORTED_DROPPED:
diff --git a/storage/innobase/row/row0purge.cc b/storage/innobase/row/row0purge.cc
index 75d497b2cf4..30cbbd03c0f 100644
--- a/storage/innobase/row/row0purge.cc
+++ b/storage/innobase/row/row0purge.cc
@@ -1,7 +1,7 @@
 /*****************************************************************************
 
 Copyright (c) 1997, 2017, Oracle and/or its affiliates. All Rights Reserved.
-Copyright (c) 2017, 2021, MariaDB Corporation.
+Copyright (c) 2017, 2022, MariaDB Corporation.
 
 This program is free software; you can redistribute it and/or modify it under
 the terms of the GNU General Public License as published by the Free Software
@@ -41,7 +41,6 @@ Created 3/14/1997 Heikki Tuuri
 #include "row0upd.h"
 #include "row0vers.h"
 #include "row0mysql.h"
-#include "row0log.h"
 #include "log0log.h"
 #include "srv0mon.h"
 #include "srv0start.h"
@@ -359,27 +358,6 @@ row_purge_remove_sec_if_poss_tree(
 	mtr.start();
 	index->set_modified(mtr);
 
-	if (!index->is_committed()) {
-		/* The index->online_status may change if the index is
-		or was being created online, but not committed yet. It
-		is protected by index->lock. */
-		mtr_sx_lock_index(index, &mtr);
-
-		if (dict_index_is_online_ddl(index)) {
-			/* Online secondary index creation will not
-			copy any delete-marked records. Therefore
-			there is nothing to be purged. We must also
-			skip the purge when a completed index is
-			dropped by rollback_inplace_alter_table(). */
-			goto func_exit_no_pcur;
-		}
-	} else {
-		/* For secondary indexes,
-		index->online_status==ONLINE_INDEX_COMPLETE if
-		index->is_committed(). */
-		ut_ad(!dict_index_is_online_ddl(index));
-	}
-
 	search_result = row_search_index_entry(
 				index, entry,
 				BTR_MODIFY_TREE | BTR_LATCH_FOR_DELETE,
@@ -452,7 +430,6 @@ row_purge_remove_sec_if_poss_tree(
 
 func_exit:
 	btr_pcur_close(&pcur); // FIXME: need this?
-func_exit_no_pcur:
 	mtr.commit();
 
 	return(success);
@@ -483,40 +460,10 @@ row_purge_remove_sec_if_poss_leaf(
 	mtr.start();
 	index->set_modified(mtr);
 
-	if (!index->is_committed()) {
-		/* For uncommitted spatial index, we also skip the purge. */
-		if (dict_index_is_spatial(index)) {
-			goto func_exit_no_pcur;
-		}
-
-		/* The index->online_status may change if the the
-		index is or was being created online, but not
-		committed yet. It is protected by index->lock. */
-		mtr_s_lock_index(index, &mtr);
-
-		if (dict_index_is_online_ddl(index)) {
-			/* Online secondary index creation will not
-			copy any delete-marked records. Therefore
-			there is nothing to be purged. We must also
-			skip the purge when a completed index is
-			dropped by rollback_inplace_alter_table(). */
-			goto func_exit_no_pcur;
-		}
-
-		mode = BTR_PURGE_LEAF_ALREADY_S_LATCHED;
-	} else {
-		/* For secondary indexes,
-		index->online_status==ONLINE_INDEX_COMPLETE if
-		index->is_committed(). */
-		ut_ad(!dict_index_is_online_ddl(index));
-
-		/* Change buffering is disabled for spatial index and
-		virtual index. */
-		mode = (dict_index_is_spatial(index)
-			|| dict_index_has_virtual(index))
-			? BTR_MODIFY_LEAF
-			: BTR_PURGE_LEAF;
-	}
+	/* Change buffering is disabled for spatial index and
+	virtual index. */
+	mode = (index->type & (DICT_SPATIAL | DICT_VIRTUAL))
+		? BTR_MODIFY_LEAF : BTR_PURGE_LEAF;
 
 	/* Set the purge node for the call to row_purge_poss_sec(). */
 	pcur.btr_cur.purge_node = node;
diff --git a/storage/innobase/row/row0uins.cc b/storage/innobase/row/row0uins.cc
index 173ae9a98f9..446d56f8daa 100644
--- a/storage/innobase/row/row0uins.cc
+++ b/storage/innobase/row/row0uins.cc
@@ -1,7 +1,7 @@
 /*****************************************************************************
 
 Copyright (c) 1997, 2017, Oracle and/or its affiliates. All Rights Reserved.
-Copyright (c) 2017, 2021, MariaDB Corporation.
+Copyright (c) 2017, 2022, MariaDB Corporation.
 
 This program is free software; you can redistribute it and/or modify it under
 the terms of the GNU General Public License as published by the Free Software
@@ -35,7 +35,6 @@ Created 2/25/1997 Heikki Tuuri
 #include "mach0data.h"
 #include "row0undo.h"
 #include "row0vers.h"
-#include "row0log.h"
 #include "trx0trx.h"
 #include "trx0rec.h"
 #include "row0row.h"
@@ -70,7 +69,6 @@ row_undo_ins_remove_clust_rec(
 	ulint		n_tries	= 0;
 	mtr_t		mtr;
 	dict_index_t*	index	= node->pcur.btr_cur.index;
-	bool		online;
 	table_id_t table_id = 0;
 	const bool dict_locked = node->trx->dict_operation_lock_mode;
 restart:
@@ -90,20 +88,10 @@ restart:
 	if (index->table->is_temporary()) {
 		ut_ad(node->rec_type == TRX_UNDO_INSERT_REC);
 		mtr.set_log_mode(MTR_LOG_NO_REDO);
-		ut_ad(!dict_index_is_online_ddl(index));
 		ut_ad(index->table->id >= DICT_HDR_FIRST_ID);
-		online = false;
 	} else {
 		index->set_modified(mtr);
 		ut_ad(lock_table_has_locks(index->table));
-		online = dict_index_is_online_ddl(index);
-		if (online) {
-			ut_ad(node->rec_type == TRX_UNDO_INSERT_REC);
-			ut_ad(!node->trx->dict_operation_lock_mode);
-			ut_ad(node->table->id != DICT_INDEXES_ID);
-			ut_ad(node->table->id != DICT_COLUMNS_ID);
-			mtr_s_lock_index(index, &mtr);
-		}
 	}
 
 	/* This is similar to row_undo_mod_clust(). The DDL thread may
@@ -112,8 +100,7 @@ restart:
 	purged. However, we can log the removal out of sync with the
 	B-tree modification. */
 	ut_a(node->pcur.restore_position(
-	      online ? BTR_MODIFY_LEAF | BTR_ALREADY_S_LATCHED
-	      : (node->rec_type == TRX_UNDO_INSERT_METADATA)
+	      (node->rec_type == TRX_UNDO_INSERT_METADATA)
 		? BTR_MODIFY_TREE
 		: BTR_MODIFY_LEAF,
 	      &mtr) == btr_pcur_t::SAME_ALL);
@@ -126,94 +113,83 @@ restart:
 	ut_ad(rec_is_metadata(rec, index->table->not_redundant())
 	      == (node->rec_type == TRX_UNDO_INSERT_METADATA));
 
-	if (online && dict_index_is_online_ddl(index)) {
-		mem_heap_t*	heap	= NULL;
-		const rec_offs*	offsets	= rec_get_offsets(
-			rec, index, NULL, index->n_core_fields,
-			ULINT_UNDEFINED, &heap);
-		row_log_table_delete(rec, index, offsets, NULL);
-		mem_heap_free(heap);
-	} else {
-		switch (node->table->id) {
-		case DICT_COLUMNS_ID:
-			/* This is rolling back an INSERT into SYS_COLUMNS.
-			If it was part of an instant ALTER TABLE operation, we
-			must evict the table definition, so that it can be
-			reloaded after the dictionary operation has been
-			completed. At this point, any corresponding operation
-			to the metadata record will have been rolled back. */
-			ut_ad(!online);
-			ut_ad(node->trx->dict_operation_lock_mode);
-			ut_ad(node->rec_type == TRX_UNDO_INSERT_REC);
-			if (rec_get_n_fields_old(rec)
-			    != DICT_NUM_FIELDS__SYS_COLUMNS
-			    || (rec_get_1byte_offs_flag(rec)
-				? rec_1_get_field_end_info(rec, 0) != 8
-				: rec_2_get_field_end_info(rec, 0) != 8)) {
-				break;
-			}
-			static_assert(!DICT_FLD__SYS_COLUMNS__TABLE_ID, "");
-			node->trx->evict_table(mach_read_from_8(rec));
+	switch (node->table->id) {
+	case DICT_COLUMNS_ID:
+		/* This is rolling back an INSERT into SYS_COLUMNS.
+		If it was part of an instant ALTER TABLE operation, we
+		must evict the table definition, so that it can be
+		reloaded after the dictionary operation has been
+		completed. At this point, any corresponding operation
+		to the metadata record will have been rolled back. */
+		ut_ad(node->trx->dict_operation_lock_mode);
+		ut_ad(node->rec_type == TRX_UNDO_INSERT_REC);
+		if (rec_get_n_fields_old(rec)
+		    != DICT_NUM_FIELDS__SYS_COLUMNS
+		    || (rec_get_1byte_offs_flag(rec)
+			? rec_1_get_field_end_info(rec, 0) != 8
+			: rec_2_get_field_end_info(rec, 0) != 8)) {
 			break;
-		case DICT_INDEXES_ID:
-			ut_ad(!online);
-			ut_ad(node->trx->dict_operation_lock_mode);
-			ut_ad(node->rec_type == TRX_UNDO_INSERT_REC);
-			if (!table_id) {
-				table_id = mach_read_from_8(rec);
-				if (table_id) {
-					mtr.commit();
-					goto restart;
-				}
-				ut_ad("corrupted SYS_INDEXES record" == 0);
+		}
+		static_assert(!DICT_FLD__SYS_COLUMNS__TABLE_ID, "");
+		node->trx->evict_table(mach_read_from_8(rec));
+		break;
+	case DICT_INDEXES_ID:
+		ut_ad(node->trx->dict_operation_lock_mode);
+		ut_ad(node->rec_type == TRX_UNDO_INSERT_REC);
+		if (!table_id) {
+			table_id = mach_read_from_8(rec);
+			if (table_id) {
+				mtr.commit();
+				goto restart;
 			}
+			ut_ad("corrupted SYS_INDEXES record" == 0);
+		}
 
-			pfs_os_file_t d = OS_FILE_CLOSED;
-
-			if (const uint32_t space_id = dict_drop_index_tree(
-				    &node->pcur, node->trx, &mtr)) {
-				if (table) {
-					lock_release_on_rollback(node->trx,
-								 table);
-					if (!dict_locked) {
-						dict_sys.lock(SRW_LOCK_CALL);
-					}
-					if (table->release()) {
-						dict_sys.remove(table);
-					} else if (table->space_id
-						   == space_id) {
-						table->space = nullptr;
-						table->file_unreadable = true;
-					}
-					if (!dict_locked) {
-						dict_sys.unlock();
-					}
-					table = nullptr;
-					if (!mdl_ticket);
-					else if (MDL_context* mdl_context =
-						 static_cast<MDL_context*>(
-							 thd_mdl_context(
-								 node->trx->
-								 mysql_thd))) {
-						mdl_context->release_lock(
-							mdl_ticket);
-						mdl_ticket = nullptr;
-					}
-				}
+		pfs_os_file_t d = OS_FILE_CLOSED;
 
-				d = fil_delete_tablespace(space_id);
+		if (const uint32_t space_id = dict_drop_index_tree(
+			    &node->pcur, node->trx, &mtr)) {
+			if (table) {
+				lock_release_on_rollback(node->trx,
+							 table);
+				if (!dict_locked) {
+					dict_sys.lock(SRW_LOCK_CALL);
+				}
+				if (table->release()) {
+					dict_sys.remove(table);
+				} else if (table->space_id
+					   == space_id) {
+					table->space = nullptr;
+					table->file_unreadable = true;
+				}
+				if (!dict_locked) {
+					dict_sys.unlock();
+				}
+				table = nullptr;
+				if (!mdl_ticket);
+				else if (MDL_context* mdl_context =
+					 static_cast<MDL_context*>(
+						 thd_mdl_context(
+							 node->trx->
+							 mysql_thd))) {
+					mdl_context->release_lock(
+						mdl_ticket);
+					mdl_ticket = nullptr;
+				}
 			}
 
-			mtr.commit();
+			d = fil_delete_tablespace(space_id);
+		}
 
-			if (d != OS_FILE_CLOSED) {
-				os_file_close(d);
-			}
+		mtr.commit();
 
-			mtr.start();
-			ut_a(node->pcur.restore_position(
-				BTR_MODIFY_LEAF, &mtr) == btr_pcur_t::SAME_ALL);
+		if (d != OS_FILE_CLOSED) {
+			os_file_close(d);
 		}
+
+		mtr.start();
+		ut_a(node->pcur.restore_position(
+			BTR_MODIFY_LEAF, &mtr) == btr_pcur_t::SAME_ALL);
 	}
 
 	if (btr_cur_optimistic_delete(&node->pcur.btr_cur, 0, &mtr)) {
@@ -299,10 +275,6 @@ row_undo_ins_remove_sec_low(
 		mtr_sx_lock_index(index, &mtr);
 	}
 
-	if (row_log_online_op_try(index, entry, 0)) {
-		goto func_exit_no_pcur;
-	}
-
 	if (dict_index_is_spatial(index)) {
 		if (modify_leaf) {
 			mode |= BTR_RTREE_DELETE_MARK;
@@ -346,7 +318,6 @@ row_undo_ins_remove_sec_low(
 	}
 
 	btr_pcur_close(&pcur);
-func_exit_no_pcur:
 	mtr_commit(&mtr);
 
 	return(err);
@@ -546,7 +517,7 @@ row_undo_ins_remove_sec_rec(
 	while (index != NULL) {
 		dtuple_t*	entry;
 
-		if (index->type & DICT_FTS) {
+		if (index->type & DICT_FTS || !index->is_committed()) {
 			dict_table_next_uncorrupted_index(index);
 			continue;
 		}
diff --git a/storage/innobase/row/row0umod.cc b/storage/innobase/row/row0umod.cc
index 494a23eaf0a..f18d7ab1be1 100644
--- a/storage/innobase/row/row0umod.cc
+++ b/storage/innobase/row/row0umod.cc
@@ -1,7 +1,7 @@
 /*****************************************************************************
 
 Copyright (c) 1997, 2017, Oracle and/or its affiliates. All Rights Reserved.
-Copyright (c) 2017, 2021, MariaDB Corporation.
+Copyright (c) 2017, 2022, MariaDB Corporation.
 
 This program is free software; you can redistribute it and/or modify it under
 the terms of the GNU General Public License as published by the Free Software
@@ -36,7 +36,6 @@ Created 2/27/1997 Heikki Tuuri
 #include "ibuf0ibuf.h"
 #include "row0undo.h"
 #include "row0vers.h"
-#include "row0log.h"
 #include "trx0trx.h"
 #include "trx0rec.h"
 #include "row0row.h"
@@ -80,11 +79,6 @@ row_undo_mod_clust_low(
 	mem_heap_t**	offsets_heap,
 				/*!< in/out: memory heap that can be emptied */
 	mem_heap_t*	heap,	/*!< in/out: memory heap */
-	const dtuple_t**rebuilt_old_pk,
-				/*!< out: row_log_table_get_pk()
-				before the update, or NULL if
-				the table is not being rebuilt online or
-				the PRIMARY KEY definition does not change */
 	byte*		sys,	/*!< out: DB_TRX_ID, DB_ROLL_PTR
 				for row_log_table_delete() */
 	que_thr_t*	thr,	/*!< in: query thread */
@@ -111,15 +105,6 @@ row_undo_mod_clust_low(
 	      || node->update->info_bits == REC_INFO_METADATA_ADD
 	      || node->update->info_bits == REC_INFO_METADATA_ALTER);
 
-	if (mode != BTR_MODIFY_LEAF
-	    && dict_index_is_online_ddl(btr_cur_get_index(btr_cur))) {
-		*rebuilt_old_pk = row_log_table_get_pk(
-			btr_cur_get_rec(btr_cur),
-			btr_cur_get_index(btr_cur), NULL, sys, &heap);
-	} else {
-		*rebuilt_old_pk = NULL;
-	}
-
 	if (mode != BTR_MODIFY_TREE) {
 		ut_ad((mode & ulint(~BTR_ALREADY_S_LATCHED))
 		      == BTR_MODIFY_LEAF);
@@ -269,7 +254,6 @@ row_undo_mod_clust(
 	bool		have_latch = false;
 	dberr_t		err;
 	dict_index_t*	index;
-	bool		online;
 
 	ut_ad(thr_get_trx(thr) == node->trx);
 	ut_ad(node->trx->in_rollback);
@@ -287,26 +271,16 @@ row_undo_mod_clust(
 		ut_ad(lock_table_has_locks(index->table));
 	}
 
-	online = dict_index_is_online_ddl(index);
-	if (online) {
-		ut_ad(!node->trx->dict_operation_lock_mode);
-		mtr_s_lock_index(index, &mtr);
-	}
-
 	mem_heap_t*	heap		= mem_heap_create(1024);
 	mem_heap_t*	offsets_heap	= NULL;
 	rec_offs*	offsets		= NULL;
-	const dtuple_t*	rebuilt_old_pk;
 	byte		sys[DATA_TRX_ID_LEN + DATA_ROLL_PTR_LEN];
 
 	/* Try optimistic processing of the record, keeping changes within
 	the index page */
 
 	err = row_undo_mod_clust_low(node, &offsets, &offsets_heap,
-				     heap, &rebuilt_old_pk, sys,
-				     thr, &mtr, online
-				     ? BTR_MODIFY_LEAF | BTR_ALREADY_S_LATCHED
-				     : BTR_MODIFY_LEAF);
+				     heap, sys, thr, &mtr, BTR_MODIFY_LEAF);
 
 	if (err != DB_SUCCESS) {
 		btr_pcur_commit_specify_mtr(pcur, &mtr);
@@ -321,34 +295,12 @@ row_undo_mod_clust(
 			index->set_modified(mtr);
 		}
 
-		err = row_undo_mod_clust_low(
-			node, &offsets, &offsets_heap,
-			heap, &rebuilt_old_pk, sys,
-			thr, &mtr, BTR_MODIFY_TREE);
+		err = row_undo_mod_clust_low(node, &offsets, &offsets_heap,
+					     heap, sys, thr, &mtr,
+					     BTR_MODIFY_TREE);
 		ut_ad(err == DB_SUCCESS || err == DB_OUT_OF_FILE_SPACE);
 	}
 
-	if (err == DB_SUCCESS && online && dict_index_is_online_ddl(index)) {
-		switch (node->rec_type) {
-		case TRX_UNDO_DEL_MARK_REC:
-			row_log_table_insert(
-				btr_pcur_get_rec(pcur), index, offsets);
-			break;
-		case TRX_UNDO_UPD_EXIST_REC:
-			row_log_table_update(
-				btr_pcur_get_rec(pcur), index, offsets,
-				rebuilt_old_pk);
-			break;
-		case TRX_UNDO_UPD_DEL_REC:
-			row_log_table_delete(
-				btr_pcur_get_rec(pcur), index, offsets, sys);
-			break;
-		default:
-			ut_ad(0);
-			break;
-		}
-	}
-
 	/**
 	* when scrubbing, and records gets cleared,
 	*   the transaction id is not present afterwards.
@@ -570,10 +522,6 @@ row_undo_mod_del_mark_or_remove_sec_low(
 			ut_ad(mode == (BTR_MODIFY_TREE | BTR_LATCH_FOR_DELETE));
 			mtr_sx_lock_index(index, &mtr);
 		}
-
-		if (row_log_online_op_try(index, entry, 0)) {
-			goto func_exit_no_pcur;
-		}
 	} else {
 		/* For secondary indexes,
 		index->online_status==ONLINE_INDEX_COMPLETE if
@@ -669,7 +617,6 @@ row_undo_mod_del_mark_or_remove_sec_low(
 
 func_exit:
 	btr_pcur_close(&pcur);
-func_exit_no_pcur:
 	mtr_commit(&mtr);
 
 	return(err);
@@ -753,28 +700,6 @@ row_undo_mod_del_unmark_sec_and_undo_update(
 try_again:
 	row_mtr_start(&mtr, index, !(mode & BTR_MODIFY_LEAF));
 
-	if (!index->is_committed()) {
-		/* The index->online_status may change if the index is
-		or was being created online, but not committed yet. It
-		is protected by index->lock. */
-		if (mode == BTR_MODIFY_LEAF) {
-			mode = BTR_MODIFY_LEAF | BTR_ALREADY_S_LATCHED;
-			mtr_s_lock_index(index, &mtr);
-		} else {
-			ut_ad(mode == BTR_MODIFY_TREE);
-			mtr_sx_lock_index(index, &mtr);
-		}
-
-		if (row_log_online_op_try(index, entry, trx->id)) {
-			goto func_exit_no_pcur;
-		}
-	} else {
-		/* For secondary indexes,
-		index->online_status==ONLINE_INDEX_COMPLETE if
-		index->is_committed(). */
-		ut_ad(!dict_index_is_online_ddl(index));
-	}
-
 	btr_cur->thr = thr;
 
 	search_result = row_search_index_entry(index, entry, mode,
@@ -802,34 +727,26 @@ try_again:
 			}
 		}
 
-		if (index->is_committed()) {
-			/* During online secondary index creation, it
-			is possible that MySQL is waiting for a
-			meta-data lock upgrade before invoking
-			ha_innobase::commit_inplace_alter_table()
-			while this ROLLBACK is executing. InnoDB has
-			finished building the index, but it does not
-			yet exist in MySQL. In this case, we suppress
-			the printout to the error log. */
+		if (btr_cur->up_match >= dict_index_get_n_unique(index)
+		    || btr_cur->low_match >= dict_index_get_n_unique(index)) {
 			ib::warn() << "Record in index " << index->name
 				<< " of table " << index->table->name
-				<< " was not found on rollback, trying to"
-				" insert: " << *entry
+				<< " was not found on rollback, and"
+				" a duplicate exists: "
+				<< *entry
 				<< " at: " << rec_index_print(
 					btr_cur_get_rec(btr_cur), index);
-		}
-
-		if (btr_cur->up_match >= dict_index_get_n_unique(index)
-		    || btr_cur->low_match >= dict_index_get_n_unique(index)) {
-			if (index->is_committed()) {
-				ib::warn() << "Record in index " << index->name
-					<< " was not found on rollback, and"
-					" a duplicate exists";
-			}
 			err = DB_DUPLICATE_KEY;
 			break;
 		}
 
+		ib::warn() << "Record in index " << index->name
+			<< " of table " << index->table->name
+			<< " was not found on rollback, trying to insert: "
+			<< *entry
+			<< " at: " << rec_index_print(
+				btr_cur_get_rec(btr_cur), index);
+
 		/* Insert the missing record that we were trying to
 		delete-unmark. */
 		big_rec_t*	big_rec;
@@ -912,7 +829,6 @@ try_again:
 	}
 
 	btr_pcur_close(&pcur);
-func_exit_no_pcur:
 	mtr_commit(&mtr);
 
 	return(err);
@@ -940,7 +856,7 @@ row_undo_mod_upd_del_sec(
 		dict_index_t*	index	= node->index;
 		dtuple_t*	entry;
 
-		if (index->type & DICT_FTS) {
+		if (index->type & DICT_FTS || !index->is_committed()) {
 			dict_table_next_uncorrupted_index(node->index);
 			continue;
 		}
@@ -1006,7 +922,7 @@ row_undo_mod_del_mark_sec(
 		dict_index_t*	index	= node->index;
 		dtuple_t*	entry;
 
-		if (index->type == DICT_FTS) {
+		if (index->type == DICT_FTS || !index->is_committed()) {
 			dict_table_next_uncorrupted_index(node->index);
 			continue;
 		}
@@ -1076,6 +992,12 @@ row_undo_mod_upd_exist_sec(
 
 
 	while (node->index != NULL) {
+
+		if (!node->index->is_committed()) {
+			dict_table_next_uncorrupted_index(node->index);
+			continue;
+		}
+
 		dict_index_t*	index	= node->index;
 		dtuple_t*	entry;
 
diff --git a/storage/innobase/row/row0upd.cc b/storage/innobase/row/row0upd.cc
index 4d364c19540..0dd0266ac45 100644
--- a/storage/innobase/row/row0upd.cc
+++ b/storage/innobase/row/row0upd.cc
@@ -1,7 +1,7 @@
 /*****************************************************************************
 
 Copyright (c) 1996, 2017, Oracle and/or its affiliates. All Rights Reserved.
-Copyright (c) 2015, 2021, MariaDB Corporation.
+Copyright (c) 2015, 2022, MariaDB Corporation.
 
 This program is free software; you can redistribute it and/or modify it under
 the terms of the GNU General Public License as published by the Free Software
@@ -1081,16 +1081,7 @@ row_upd_replace_vcol(
 		/* If there is no index on the column, do not bother for
 		value update */
 		if (!col->m_col.ord_part) {
-			dict_index_t*	clust_index
-				= dict_table_get_first_index(table);
-
-			/* Skip the column if there is no online alter
-			table in progress or it is not being indexed
-			in new table */
-			if (!dict_index_is_online_ddl(clust_index)
-			    || !row_log_col_is_indexed(clust_index, col_no)) {
-				continue;
-			}
+			continue;
 		}
 
 		dfield = dtuple_get_nth_v_field(row, col_no);
@@ -1906,6 +1897,13 @@ row_upd_sec_index_entry(
 	ut_ad(trx->id != 0);
 
 	index = node->index;
+	if (!index->is_committed()) {
+		return DB_SUCCESS;
+	}
+
+	/* For secondary indexes, index->online_status==ONLINE_INDEX_COMPLETE
+	if index->is_committed(). */
+	ut_ad(!dict_index_is_online_ddl(index));
 
 	const bool referenced = row_upd_index_is_referenced(index, trx);
 #ifdef WITH_WSREP
@@ -1929,75 +1927,22 @@ row_upd_sec_index_entry(
 	case SRV_TMP_SPACE_ID:
 		mtr.set_log_mode(MTR_LOG_NO_REDO);
 		flags = BTR_NO_LOCKING_FLAG;
+		mode = index->is_spatial()
+			? ulint(BTR_MODIFY_LEAF | BTR_RTREE_DELETE_MARK)
+			: ulint(BTR_MODIFY_LEAF);
 		break;
 	default:
 		index->set_modified(mtr);
 		/* fall through */
 	case IBUF_SPACE_ID:
 		flags = index->table->no_rollback() ? BTR_NO_ROLLBACK : 0;
-		break;
-	}
-
-	bool uncommitted = !index->is_committed();
-
-	if (uncommitted) {
-		/* The index->online_status may change if the index is
-		or was being created online, but not committed yet. It
-		is protected by index->lock. */
-
-		mtr_s_lock_index(index, &mtr);
-
-		switch (dict_index_get_online_status(index)) {
-		case ONLINE_INDEX_COMPLETE:
-			/* This is a normal index. Do not log anything.
-			Perform the update on the index tree directly. */
-			break;
-		case ONLINE_INDEX_CREATION:
-			/* Log a DELETE and optionally INSERT. */
-			row_log_online_op(index, entry, 0);
-
-			if (!node->is_delete) {
-				mem_heap_empty(heap);
-				entry = row_build_index_entry(
-					node->upd_row, node->upd_ext,
-					index, heap);
-				ut_a(entry);
-				row_log_online_op(index, entry, trx->id);
-			}
-			/* fall through */
-		case ONLINE_INDEX_ABORTED:
-		case ONLINE_INDEX_ABORTED_DROPPED:
-			mtr_commit(&mtr);
-			goto func_exit;
-		}
-
 		/* We can only buffer delete-mark operations if there
-		are no foreign key constraints referring to the index.
-		Change buffering is disabled for temporary tables and
-		spatial index. */
-		mode = (referenced || index->table->is_temporary()
-			|| dict_index_is_spatial(index))
-			? BTR_MODIFY_LEAF_ALREADY_S_LATCHED
-			: BTR_DELETE_MARK_LEAF_ALREADY_S_LATCHED;
-	} else {
-		/* For secondary indexes,
-		index->online_status==ONLINE_INDEX_COMPLETE if
-		index->is_committed(). */
-		ut_ad(!dict_index_is_online_ddl(index));
-
-		/* We can only buffer delete-mark operations if there
-		are no foreign key constraints referring to the index.
-		Change buffering is disabled for temporary tables and
-		spatial index. */
-		mode = (referenced || index->table->is_temporary()
-			|| dict_index_is_spatial(index))
-			? BTR_MODIFY_LEAF
-			: BTR_DELETE_MARK_LEAF;
-	}
-
-	if (dict_index_is_spatial(index)) {
-		ut_ad(mode & BTR_MODIFY_LEAF);
-		mode |= BTR_RTREE_DELETE_MARK;
+		are no foreign key constraints referring to the index. */
+		mode = index->is_spatial()
+			? ulint(BTR_MODIFY_LEAF | BTR_RTREE_DELETE_MARK)
+			: referenced
+			? ulint(BTR_MODIFY_LEAF) : ulint(BTR_DELETE_MARK_LEAF);
+		break;
 	}
 
 	/* Set the query thread, so that ibuf_insert_low() will be
@@ -2020,19 +1965,6 @@ row_upd_sec_index_entry(
 		break;
 
 	case ROW_NOT_FOUND:
-		if (!index->is_committed()) {
-			/* When online CREATE INDEX copied the update
-			that we already made to the clustered index,
-			and completed the secondary index creation
-			before we got here, the old secondary index
-			record would not exist. The CREATE INDEX
-			should be waiting for a MySQL meta-data lock
-			upgrade at least until this UPDATE returns.
-			After that point, set_committed(true) would be
-			invoked by commit_inplace_alter_table(). */
-			break;
-		}
-
 		if (dict_index_is_spatial(index) && btr_cur->rtr_info->fd_del) {
 			/* We found the record, but a delete marked */
 			break;
@@ -2139,35 +2071,11 @@ row_upd_sec_index_entry(
 	DEBUG_SYNC_C_IF_THD(trx->mysql_thd,
 			    "before_row_upd_sec_new_index_entry");
 
-	uncommitted = !index->is_committed();
-	if (uncommitted) {
-		mtr.start();
-		/* The index->online_status may change if the index is
-		being rollbacked. It is protected by index->lock. */
-
-		mtr_s_lock_index(index, &mtr);
-
-		switch (dict_index_get_online_status(index)) {
-		case ONLINE_INDEX_COMPLETE:
-		case ONLINE_INDEX_CREATION:
-		       break;
-		case ONLINE_INDEX_ABORTED:
-		case ONLINE_INDEX_ABORTED_DROPPED:
-		       mtr_commit(&mtr);
-		       goto func_exit;
-		}
-
-	}
-
 	/* Build a new index entry */
 	entry = row_build_index_entry(node->upd_row, node->upd_ext,
 				      index, heap);
 	ut_a(entry);
 
-	if (uncommitted) {
-		mtr_commit(&mtr);
-	}
-
 	/* Insert new index entry */
 	err = row_ins_sec_index_entry(index, entry, thr, !node->is_delete);
 
@@ -2488,7 +2396,6 @@ row_upd_clust_rec(
 	btr_pcur_t*	pcur;
 	btr_cur_t*	btr_cur;
 	dberr_t		err;
-	const dtuple_t*	rebuilt_old_pk	= NULL;
 
 	ut_ad(dict_index_is_clust(index));
 	ut_ad(!thr_get_trx(thr)->in_rollback);
@@ -2502,11 +2409,6 @@ row_upd_clust_rec(
 				    dict_table_is_comp(index->table)));
 	ut_ad(rec_offs_validate(btr_cur_get_rec(btr_cur), index, offsets));
 
-	if (dict_index_is_online_ddl(index)) {
-		rebuilt_old_pk = row_log_table_get_pk(
-			btr_cur_get_rec(btr_cur), index, offsets, NULL, &heap);
-	}
-
 	/* Try optimistic updating of the record, keeping changes within
 	the page; we do not check locks because we assume the x-lock on the
 	record to update */
@@ -2524,7 +2426,7 @@ row_upd_clust_rec(
 	}
 
 	if (err == DB_SUCCESS) {
-		goto success;
+		goto func_exit;
 	}
 
 	if (buf_pool.running_out()) {
@@ -2577,15 +2479,6 @@ row_upd_clust_rec(
 		DEBUG_SYNC_C("after_row_upd_extern");
 	}
 
-	if (err == DB_SUCCESS) {
-success:
-		if (dict_index_is_online_ddl(index)) {
-			row_log_table_update(
-				btr_cur_get_rec(btr_cur),
-				index, offsets, rebuilt_old_pk);
-		}
-	}
-
 func_exit:
 	if (heap) {
 		mem_heap_free(heap);
@@ -2932,7 +2825,8 @@ row_upd(
 			break;
 		}
 
-		if (node->index->type != DICT_FTS) {
+		if (node->index->type != DICT_FTS
+		    && node->index->is_committed()) {
 			err = row_upd_sec_step(node, thr);
 
 			if (err != DB_SUCCESS) {
diff --git a/storage/innobase/trx/trx0rec.cc b/storage/innobase/trx/trx0rec.cc
index a78212d3359..4ba526bdc19 100644
--- a/storage/innobase/trx/trx0rec.cc
+++ b/storage/innobase/trx/trx0rec.cc
@@ -2037,6 +2037,10 @@ trx_undo_report_row_operation(
 	auto m = trx->mod_tables.emplace(index->table, trx->undo_no);
 	ut_ad(m.first->second.valid(trx->undo_no));
 
+	if (m.second && index->table->is_active_ddl()) {
+		trx->apply_online_log= true;
+	}
+
 	bool bulk = !rec;
 
 	if (!bulk) {
@@ -2226,12 +2230,14 @@ err_exit:
 
 /** Copy an undo record to heap.
 @param[in]	roll_ptr	roll pointer to a record that exists
-@param[in,out]	heap		memory heap where copied */
+@param[in,out]	heap		memory heap where copied
+@param[in]	undo_rec_info	undo log record information */
 static
 trx_undo_rec_t*
 trx_undo_get_undo_rec_low(
-	roll_ptr_t	roll_ptr,
-	mem_heap_t*	heap)
+	roll_ptr_t		roll_ptr,
+	mem_heap_t*		heap,
+	const trx_undo_rec_info*undo_rec_info= nullptr)
 {
 	trx_undo_rec_t*	undo_rec;
 	ulint		rseg_id;
@@ -2247,14 +2253,21 @@ trx_undo_get_undo_rec_low(
 	trx_rseg_t* rseg = &trx_sys.rseg_array[rseg_id];
 	ut_ad(rseg->is_persistent());
 
-	mtr.start();
+	if (undo_rec_info
+	    && undo_rec_info->block->page.id().page_no() == page_no) {
+		undo_rec = trx_undo_rec_copy(
+			undo_rec_info->block->page.frame + offset, heap);
+	} else {
+		mtr.start();
 
-	buf_block_t* undo_page = trx_undo_page_get_s_latched(
-		page_id_t(rseg->space->id, page_no), &mtr);
+		buf_block_t *undo_page = trx_undo_page_get_s_latched(
+			page_id_t(rseg->space->id, page_no), &mtr);
 
-	undo_rec = trx_undo_rec_copy(undo_page->page.frame + offset, heap);
+		undo_rec = trx_undo_rec_copy(
+			undo_page->page.frame + offset, heap);
 
-	mtr.commit();
+		mtr.commit();
+	}
 
 	return(undo_rec);
 }
@@ -2278,13 +2291,15 @@ trx_undo_get_undo_rec(
 	mem_heap_t*		heap,
 	trx_id_t		trx_id,
 	const table_name_t&	name,
-	trx_undo_rec_t**	undo_rec)
+	trx_undo_rec_t**	undo_rec,
+	const trx_undo_rec_info*undo_rec_info= nullptr)
 {
 	purge_sys.latch.rd_lock(SRW_LOCK_CALL);
 
 	bool missing_history = purge_sys.changes_visible(trx_id, name);
 	if (!missing_history) {
-		*undo_rec = trx_undo_get_undo_rec_low(roll_ptr, heap);
+		*undo_rec = trx_undo_get_undo_rec_low(
+				roll_ptr, heap, undo_rec_info);
 	}
 
 	purge_sys.latch.rd_unlock();
@@ -2298,41 +2313,47 @@ trx_undo_get_undo_rec(
 #define ATTRIB_USED_ONLY_IN_DEBUG	MY_ATTRIBUTE((unused))
 #endif /* UNIV_DEBUG */
 
-/*******************************************************************//**
-Build a previous version of a clustered index record. The caller must
-hold a latch on the index page of the clustered index record.
+/** Build a previous version of a clustered index record. The caller
+must hold a latch on the index page of the clustered index record.
+@param	index_rec	clustered index record in the index tree
+@param	index_mtr	mtr which contains the latch to index_rec page
+			and purge_view
+@param	rec		version of a clustered index record
+@param	index		clustered index
+@param	offsets		rec_get_offsets(rec, index)
+@param	heap		memory heap from which the memory needed is
+			allocated
+@param	old_vers	previous version or NULL if rec is the
+			first inserted version, or if history data
+			has been deleted (an error), or if the purge
+			could have removed the version
+			though it has not yet done so
+@param	v_heap		memory heap used to create vrow
+			dtuple if it is not yet created. This heap
+                        diffs from "heap" above in that it could be
+                        prebuilt->old_vers_heap for selection
+@param	v_row		virtual column info, if any
+@param	v_status	status determine if it is going into this
+			function by purge thread or not.
+			And if we read "after image" of undo log
+@param	undo_rec_info	undo record info or NULL
 @retval true if previous version was built, or if it was an insert
 or the table has been rebuilt
 @retval false if the previous version is earlier than purge_view,
 or being purged, which means that it may have been removed */
 bool
 trx_undo_prev_version_build(
-/*========================*/
-	const rec_t*	index_rec ATTRIB_USED_ONLY_IN_DEBUG,
-				/*!< in: clustered index record in the
-				index tree */
-	mtr_t*		index_mtr ATTRIB_USED_ONLY_IN_DEBUG,
-				/*!< in: mtr which contains the latch to
-				index_rec page and purge_view */
-	const rec_t*	rec,	/*!< in: version of a clustered index record */
-	dict_index_t*	index,	/*!< in: clustered index */
-	rec_offs*	offsets,/*!< in/out: rec_get_offsets(rec, index) */
-	mem_heap_t*	heap,	/*!< in: memory heap from which the memory
-				needed is allocated */
-	rec_t**		old_vers,/*!< out, own: previous version, or NULL if
-				rec is the first inserted version, or if
-				history data has been deleted (an error),
-				or if the purge COULD have removed the version
-				though it has not yet done so */
-	mem_heap_t*	v_heap,	/* !< in: memory heap used to create vrow
-				dtuple if it is not yet created. This heap
-				diffs from "heap" above in that it could be
-				prebuilt->old_vers_heap for selection */
-	dtuple_t**	vrow,	/*!< out: virtual column info, if any */
-	ulint		v_status)
-				/*!< in: status determine if it is going
-				into this function by purge thread or not.
-				And if we read "after image" of undo log */
+	const rec_t	*index_rec ATTRIB_USED_ONLY_IN_DEBUG,
+	mtr_t		*index_mtr ATTRIB_USED_ONLY_IN_DEBUG,
+	const rec_t 	*rec,
+	dict_index_t	*index,
+	rec_offs	*offsets,
+	mem_heap_t	*heap,
+	rec_t		**old_vers,
+	mem_heap_t	*v_heap,
+	dtuple_t	**vrow,
+	ulint		v_status,
+	const trx_undo_rec_info *undo_rec_info)
 {
 	trx_undo_rec_t*	undo_rec	= NULL;
 	dtuple_t*	entry;
@@ -2371,7 +2392,7 @@ trx_undo_prev_version_build(
 
 	if (trx_undo_get_undo_rec(
 		    roll_ptr, heap, rec_trx_id, index->table->name,
-		    &undo_rec)) {
+		    &undo_rec, undo_rec_info)) {
 		if (v_status & TRX_UNDO_PREV_IN_PURGE) {
 			/* We are fetching the record being purged */
 			undo_rec = trx_undo_get_undo_rec_low(roll_ptr, heap);
@@ -2634,3 +2655,17 @@ trx_undo_read_v_cols(
 
 	ut_ad(ptr == end_ptr);
 }
+
+bool trx_undo_rec_is_equal(roll_ptr_t roll_ptr,
+                           const trx_undo_rec_info *undo_info)
+{
+  bool is_insert;
+  ulint rseg_id;
+  uint16_t offset;
+  uint32_t page_no;
+
+  trx_undo_decode_roll_ptr(roll_ptr, &is_insert, &rseg_id,
+                           &page_no, &offset);
+  return page_no == undo_info->block->page.id().page_no()
+	 && offset == undo_info->offset;
+}
diff --git a/storage/innobase/trx/trx0trx.cc b/storage/innobase/trx/trx0trx.cc
index 7b070a605f4..ceaf7699028 100644
--- a/storage/innobase/trx/trx0trx.cc
+++ b/storage/innobase/trx/trx0trx.cc
@@ -1,7 +1,7 @@
 /*****************************************************************************
 
 Copyright (c) 1996, 2016, Oracle and/or its affiliates. All Rights Reserved.
-Copyright (c) 2015, 2021, MariaDB Corporation.
+Copyright (c) 2015, 2022, MariaDB Corporation.
 
 This program is free software; you can redistribute it and/or modify it under
 the terms of the GNU General Public License as published by the Free Software
@@ -133,6 +133,8 @@ trx_init(
 
 	trx->bulk_insert = false;
 
+	trx->apply_online_log = false;
+
 	ut_d(trx->start_file = 0);
 
 	ut_d(trx->start_line = 0);
@@ -1396,7 +1398,7 @@ void trx_t::commit_cleanup()
 TRANSACTIONAL_TARGET void trx_t::commit_low(mtr_t *mtr)
 {
   ut_ad(!mtr || mtr->is_active());
-  ut_d(bool aborted = in_rollback && error_state == DB_DEADLOCK);
+  ut_d(bool aborted= in_rollback && error_state == DB_DEADLOCK);
   ut_ad(!mtr == (aborted || !has_logged()));
   ut_ad(!mtr || !aborted);
 
@@ -1415,12 +1417,13 @@ TRANSACTIONAL_TARGET void trx_t::commit_low(mtr_t *mtr)
       ut_ad(error == DB_DUPLICATE_KEY || error == DB_LOCK_WAIT_TIMEOUT);
   }
 
-#ifndef DBUG_OFF
+#ifdef ENABLED_DEBUG_SYNC
   const bool debug_sync= mysql_thd && has_logged_persistent();
 #endif
 
   if (mtr)
   {
+    apply_log();
     trx_write_serialisation_history(this, mtr);
 
     /* The following call commits the mini-transaction, making the
@@ -1440,7 +1443,7 @@ TRANSACTIONAL_TARGET void trx_t::commit_low(mtr_t *mtr)
 
     mtr->commit();
   }
-#ifndef DBUG_OFF
+#ifdef ENABLED_DEBUG_SYNC
   if (debug_sync)
     DEBUG_SYNC_C("before_trx_state_committed_in_memory");
 #endif
diff --git a/storage/innobase/trx/trx0undo.cc b/storage/innobase/trx/trx0undo.cc
index 86f2d467219..4ddb3a89853 100644
--- a/storage/innobase/trx/trx0undo.cc
+++ b/storage/innobase/trx/trx0undo.cc
@@ -290,6 +290,151 @@ trx_undo_get_first_rec(const fil_space_t &space, uint32_t page_no,
                                               mtr);
 }
 
+typedef std::map<table_id_t, dict_table_t*> table_id_map;
+
+/** Apply the TRX_UNDO_INSERT_REC undo log record
+@param	rec		undo log record
+@param	rec_info	undo log record info
+@param	table		table to be logged
+@param	heap		memory heap */
+static void trx_undo_rec_apply_insert(trx_undo_rec_t *rec,
+                                      trx_undo_rec_info *rec_info,
+                                      dict_table_t *table,
+                                      mem_heap_t *heap)
+{
+  dict_index_t *index= dict_table_get_first_index(table);
+  const dtuple_t *undo_tuple;
+  rec= trx_undo_rec_get_row_ref(rec, index, &undo_tuple, heap);
+  rec_info->undo_rec= rec;
+  row_log_insert_handle(undo_tuple, rec_info, index, heap);
+}
+
+/** Apply the TRX_UNDO_UPD & TRX_UNDO_DEL undo log record
+@param	rec		undo log record
+@param	undo_info	undo log record info
+@param	table		table which does online ddl
+@param	heap		memory heap */
+static void trx_undo_rec_apply_update(trx_undo_rec_t *rec,
+                                      trx_undo_rec_info *rec_info,
+                                      dict_table_t *table,
+                                      mem_heap_t *heap)
+{
+  dict_index_t *index= dict_table_get_first_index(table);
+  const dtuple_t *undo_tuple;
+  trx_id_t trx_id;
+  roll_ptr_t roll_ptr;
+  byte info_bits;
+  rec= trx_undo_update_rec_get_sys_cols(rec, &trx_id, &roll_ptr,
+                                        &info_bits);
+  rec= trx_undo_rec_get_row_ref(rec, index, &undo_tuple, heap);
+
+  if (rec_info->type == TRX_UNDO_UPD_DEL_REC)
+    row_log_insert_handle(undo_tuple, rec_info, index, heap);
+  else
+  {
+    rec= trx_undo_update_rec_get_update(rec, index, rec_info->type,
+                                        trx_id, roll_ptr, info_bits,
+                                        heap, &rec_info->update);
+    rec_info->undo_rec= rec;
+    row_log_update_handle(undo_tuple, rec_info, index, heap);
+  }
+}
+
+/** Apply all DML undo log records to the online DDL tables
+@param	rec			undo log record
+@param	online_log_tables	table list which does online DDL
+@param	rec_info		undo log record info
+@param	heap			heap which used to create tuple
+				for the undo log record */
+static void trx_undo_rec_apply_log(trx_undo_rec_t *rec,
+                                   const table_id_map &online_log_tables,
+                                   trx_undo_rec_info *undo_info,
+                                   mem_heap_t *heap)
+{
+  ulint type, cmpl_info= 0;
+  bool updated_extern= false;
+  undo_no_t undo_no= 0;
+  table_id_t table_id= 0;
+  rec_t *ptr= trx_undo_rec_get_pars(rec, &type, &cmpl_info,
+                                    &updated_extern, &undo_no, &table_id);
+  auto it= online_log_tables.find(table_id);
+  if (it == online_log_tables.end())
+    return;
+
+
+  if (!it->second->is_active_ddl())
+    return;
+  undo_info->assign_value(type, cmpl_info, updated_extern,
+                          undo_no);
+  switch(type)
+  {
+    case TRX_UNDO_INSERT_REC:
+      trx_undo_rec_apply_insert(ptr, undo_info, it->second, heap);
+    break;
+    case TRX_UNDO_UPD_EXIST_REC:
+    case TRX_UNDO_UPD_DEL_REC:
+    case TRX_UNDO_DEL_MARK_REC:
+      trx_undo_rec_apply_update(ptr, undo_info, it->second, heap);
+    break;
+    default:
+      ut_ad(0);
+  }
+
+  mem_heap_empty(heap);
+}
+
+
+/** Apply any changes to tables for which online DDL is in progress. */
+ATTRIBUTE_COLD void trx_t::apply_log() const
+{
+  if (undo_no == 0 || apply_online_log == false)
+    return;
+  const trx_undo_t *undo= rsegs.m_redo.undo;
+  if (!undo)
+    return;
+  table_id_map online_log_tables;
+  for (auto& t : mod_tables)
+  {
+    if (t.first->is_active_ddl())
+      online_log_tables[t.first->id]= t.first;
+  }
+  page_id_t page_id{rsegs.m_redo.rseg->space->id, undo->hdr_page_no};
+  page_id_t next_page_id(page_id);
+  mtr_t mtr;
+  mem_heap_t *heap= mem_heap_create(100);
+  mtr.start();
+  buf_block_t *block= buf_page_get(page_id, 0, RW_S_LATCH, &mtr);
+  ut_ad(block);
+
+  while (block)
+  {
+    trx_undo_rec_t *rec= trx_undo_page_get_first_rec(block, page_id.page_no(),
+                                                     undo->hdr_offset);
+    while (rec)
+    {
+      trx_undo_rec_info undo_rec_info(id, block, page_offset(rec));
+      trx_undo_rec_apply_log(rec, online_log_tables, &undo_rec_info,
+		             heap);
+      rec= trx_undo_page_get_next_rec(block, page_offset(rec),
+                                      page_id.page_no(),
+                                      undo->hdr_offset);
+    }
+
+    uint32_t next= mach_read_from_4(TRX_UNDO_PAGE_HDR + TRX_UNDO_PAGE_NODE +
+                                    FLST_NEXT + FIL_ADDR_PAGE +
+                                    block->page.frame);
+    if (next == FIL_NULL)
+      break;
+    next_page_id.set_page_no(next);
+    mtr.commit();
+    mtr.start();
+    block= buf_page_get(next_page_id, 0, RW_S_LATCH, &mtr);
+    ut_ad(block);
+  }
+  mtr.commit();
+  mem_heap_free(heap);
+}
+
 /*============== UNDO LOG FILE COPY CREATION AND FREEING ==================*/
 
 /** Initialize an undo log page.