- Addressed marko's review comments part 2bb-10.6-MDEV-15250_2

10 files changed, 102 insertions, 102 deletions

diff --git a/storage/innobase/handler/handler0alter.cc b/storage/innobase/handler/handler0alter.cc
index c4ac10bb6bf..73145463c1d 100644
--- a/storage/innobase/handler/handler0alter.cc
+++ b/storage/innobase/handler/handler0alter.cc
@@ -7021,7 +7021,8 @@ error_handling_drop_uncached:
 						row_log_free(
 							index->online_log);
 						index->online_log = NULL;
-						ctx->old_table->indexes.start->online_log = nullptr;
+						ctx->old_table->indexes.start
+							->online_log = nullptr;
 						ok = false;
 					});
 
@@ -11049,7 +11050,8 @@ err_index:
 					index->online_log= nullptr;
 					index->lock.x_unlock();
 
-					ctx->old_table->indexes.start->online_log= nullptr;
+					ctx->old_table->indexes.start
+						->online_log= nullptr;
 					if (fts_exist) {
 						purge_sys.resume_FTS();
 					}
diff --git a/storage/innobase/include/dict0mem.h b/storage/innobase/include/dict0mem.h
index 23422e9f8c1..7de1e6e3c6d 100644
--- a/storage/innobase/include/dict0mem.h
+++ b/storage/innobase/include/dict0mem.h
@@ -1408,6 +1408,12 @@ public:
     return online_log == reinterpret_cast<row_log_t*>(
 				const_cast<dict_index_t*>(this));
   }
+
+  /** Assign clustered index online log to dummy value */
+  void online_log_make_dummy()
+  {
+    online_log= reinterpret_cast<row_log_t*>(this);
+  }
 };
 
 /** Detach a virtual column from an index.
diff --git a/storage/innobase/include/row0log.h b/storage/innobase/include/row0log.h
index ea6919c8ba5..bdcb8fe87c7 100644
--- a/storage/innobase/include/row0log.h
+++ b/storage/innobase/include/row0log.h
@@ -243,19 +243,29 @@ row_log_estimate_work(
 @param	tuple		row reference from undo log record
 @param	rec_info	undo log record info
 @param	clust_index	clustered index
+@param  type		undo log record type
+@param  cmpl_info	compiler info on secondary index
+@param	update		update vector
+@param	undo_rec	pointer to remaining undo log record
 @param	heap		memory heap */
-void row_log_insert(const dtuple_t *tuple,
-                    const trx_undo_rec_info *rec_info,
+void row_log_insert(const dtuple_t &tuple,
+                    const trx_undo_rec_info &rec_info,
                     dict_index_t *clust_index,
-                    mem_heap_t *heap);
+                    ulint type, ulint cmpl_info, const upd_t *update,
+                    trx_undo_rec_t *undo_rec, mem_heap_t *heap);
 
 /** 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	type		undo log record type
+@param	cmpl_info	compiler info on secondary index
+@param	update		update vector
+@param	undo_rec	pointer to remaining undo log record
 @param	heap		memory heap */
-void row_log_update(const dtuple_t *tuple,
-                    const trx_undo_rec_info *rec_info,
+void row_log_update(const dtuple_t &tuple,
+                    const trx_undo_rec_info &rec_info,
                     dict_index_t *clust_index,
-                    mem_heap_t *heap);
+                    ulint type, ulint cmpl_info, const upd_t *update,
+                    trx_undo_rec_t *undo_rec, mem_heap_t *heap);
diff --git a/storage/innobase/include/trx0trx.h b/storage/innobase/include/trx0trx.h
index 0eb209ba7e7..e114823ceef 100644
--- a/storage/innobase/include/trx0trx.h
+++ b/storage/innobase/include/trx0trx.h
@@ -938,7 +938,7 @@ public:
   inline bool rollback_finish();
 private:
   /** Apply any changes to tables for which online DDL is in progress. */
-  ATTRIBUTE_COLD void apply_log() const;
+  ATTRIBUTE_COLD void apply_log();
   /** 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 f5308409958..d45815a6bd7 100644
--- a/storage/innobase/include/trx0undo.h
+++ b/storage/innobase/include/trx0undo.h
@@ -327,32 +327,18 @@ struct trx_undo_t {
 					segment are chained into lists */
 };
 
-/** Stores Undo record information like rollback segment id,
-block, record type, page_id, offset, update vector. It can be
-used whether the index record matches with the undo log */
+/** Cache a pointer to an undo record in a latched buffer pool page. */
 struct trx_undo_rec_info
 {
-  trx_id_t trx_id;
-  buf_block_t *block;
+  /* undo log block which was latched */
+  const buf_block_t &block;
+  /* Offset of the undo log within the 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)
+
+  trx_undo_rec_info(const buf_block_t &block, ulint offset):
+                   block(block), offset(offset)
   {
-    this->type= type;
-    this->cmpl_info= cmpl_info;
-    this->updated_extern= updated_ext;
-    this->undo_no= undo_no;
+    ut_ad(block.page.lock.have_any());
   }
 
   /** Check whether the given roll pointer is generated by
diff --git a/storage/innobase/row/row0log.cc b/storage/innobase/row/row0log.cc
index 530ad80bb6b..d9d96fe0ce6 100644
--- a/storage/innobase/row/row0log.cc
+++ b/storage/innobase/row/row0log.cc
@@ -3005,8 +3005,7 @@ row_log_allocate(
 		/* 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->online_log =
-			reinterpret_cast<row_log_t*>(index);
+		index->table->indexes.start->online_log_make_dummy();
 		log->alter_trx = trx;
 	}
 
@@ -3452,7 +3451,7 @@ row_log_apply_ops(
 
 	ut_ad(dict_index_is_online_ddl(index)
 	      || (index->online_log
-                  && index->online_status == ONLINE_INDEX_COMPLETE));
+		  && index->online_status == ONLINE_INDEX_COMPLETE));
 	ut_ad(!index->is_committed());
 	ut_ad(index->lock.have_x());
 	ut_ad(index->online_log);
@@ -3895,7 +3894,7 @@ by undo log record
 @return version of clustered index record */
 static rec_t *row_log_vers_match(
 	rec_t *rec, dict_index_t *index,
-	rec_offs **offsets, const trx_undo_rec_info *rec_info,
+	rec_offs **offsets, const trx_undo_rec_info &rec_info,
 	mtr_t *mtr, mem_heap_t *heap)
 {
   ut_ad(index->is_primary());
@@ -3904,25 +3903,17 @@ static rec_t *row_log_vers_match(
   rec_t *version= rec;
   while (version != nullptr)
   {
-    roll_ptr_t roll_ptr;
-    if (index->trx_id_offset)
-      roll_ptr= trx_read_roll_ptr(
-        rec + index->trx_id_offset + DATA_TRX_ID_LEN);
-    else
-    {
-      *offsets= rec_get_offsets(version, index, *offsets,
-                                index->n_core_fields, ULINT_UNDEFINED,
-                                &heap);
-      roll_ptr= trx_read_roll_ptr(
-        rec_get_nth_field(version, *offsets, index->db_roll_ptr(), &len));
-      ut_ad(len == DATA_ROLL_PTR_LEN);
-    }
-
-    if (rec_info->is_equal(roll_ptr))
+    *offsets= rec_get_offsets(version, index, *offsets,
+                              index->n_core_fields, ULINT_UNDEFINED,
+                              &heap);
+    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);
+    if (rec_info.is_equal(roll_ptr))
       return version;
     trx_undo_prev_version_build(rec, mtr, version, index,
                                 *offsets, heap, &prev_version, NULL,
-                                NULL, 0, rec_info);
+                                NULL, 0, &rec_info);
     version= prev_version;
   }
 
@@ -3940,17 +3931,17 @@ undo log record
 @param		heap		memory heap
 @return previous or current clustered index record if it is
 changed by the undo log record */
-static rec_t* row_log_vers_record(const dtuple_t *tuple,
+static rec_t* row_log_vers_record(const dtuple_t &tuple,
                                   dict_index_t *index,
                                   rec_t **clust_rec,
                                   rec_offs **offsets,
-                                  const trx_undo_rec_info *rec_info,
+                                  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);
+                                    index->table, &tuple, mtr);
   ut_a(found);
   *clust_rec= btr_pcur_get_rec(&pcur);
 
@@ -3987,10 +3978,11 @@ static void row_log_mark_other_online_index_abort(dict_table_t *table)
   MONITOR_ATOMIC_INC(MONITOR_BACKGROUND_DROP_INDEX);
 }
 
-void row_log_insert(const dtuple_t *tuple,
-                    const trx_undo_rec_info *rec_info,
+void row_log_insert(const dtuple_t &tuple,
+                    const trx_undo_rec_info &rec_info,
                     dict_index_t *clust_index,
-                    mem_heap_t *heap)
+                    ulint type, ulint cmpl_info, const upd_t *update,
+                    trx_undo_rec_t *undo_rec, mem_heap_t *heap)
 {
   DEBUG_SYNC_C("row_log_insert_handle");
   ut_ad(dict_index_is_clust(clust_index));
@@ -4022,7 +4014,7 @@ void row_log_insert(const dtuple_t *tuple,
   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_log_is_dummy()
       && clust_index->online_status <= ONLINE_INDEX_CREATION)
   {
     row_log_table_insert(copy_rec, clust_index, offsets);
@@ -4041,12 +4033,11 @@ void row_log_insert(const dtuple_t *tuple,
 
     if (table->n_v_cols)
     {
-      if (rec_info->type == TRX_UNDO_UPD_DEL_REC)
-        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);
-      else trx_undo_read_v_cols(table, rec_info->undo_rec, row, false);
+      if (type == TRX_UNDO_UPD_DEL_REC)
+        row_upd_replace_vcol(row, table, update, false, nullptr,
+                             (cmpl_info & UPD_NODE_NO_ORD_CHANGE)
+                             ? nullptr : undo_rec);
+      else trx_undo_read_v_cols(table, undo_rec, row, false);
     }
 
     bool success= true;
@@ -4074,10 +4065,11 @@ void row_log_insert(const dtuple_t *tuple,
   }
 }
 
-void row_log_update(const dtuple_t *tuple,
-                    const trx_undo_rec_info *rec_info,
+void row_log_update(const dtuple_t &tuple,
+                    const trx_undo_rec_info &rec_info,
                     dict_index_t *clust_index,
-                    mem_heap_t *heap)
+                    ulint type, ulint cmpl_info, const upd_t *update,
+                    trx_undo_rec_t *undo_rec, mem_heap_t *heap)
 {
   rec_offs offsets_[REC_OFFS_NORMAL_SIZE];
   rec_offs offsets2_[REC_OFFS_NORMAL_SIZE];
@@ -4093,14 +4085,14 @@ void row_log_update(const dtuple_t *tuple,
   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_log_is_dummy()
      && 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);
+  bool is_update= (type == TRX_UNDO_UPD_EXIST_REC);
   rec_t *match_rec= row_log_vers_record(
     tuple, clust_index, &rec, &offsets, rec_info, &mtr, heap);
   if (!match_rec)
@@ -4116,7 +4108,7 @@ void row_log_update(const dtuple_t *tuple,
   {
     trx_undo_prev_version_build(rec, &mtr, match_rec, clust_index,
                                 offsets, heap, &prev_version, NULL,
-                                NULL, 0, rec_info);
+                                NULL, 0, &rec_info);
 
     prev_offsets= rec_get_offsets(prev_version, clust_index, prev_offsets,
                                   clust_index->n_core_fields,
@@ -4163,7 +4155,7 @@ void row_log_update(const dtuple_t *tuple,
                  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))
+      && !(cmpl_info & UPD_NODE_NO_ORD_CHANGE))
   {
     for (ulint i = 0; i < dict_table_get_n_v_cols(table); i++)
        dfield_get_type(
@@ -4173,15 +4165,15 @@ void row_log_update(const dtuple_t *tuple,
   if (is_update)
   {
     old_row= dtuple_copy(row, heap);
-    row_upd_replace(old_row, &old_ext, clust_index, rec_info->update,
+    row_upd_replace(old_row, &old_ext, clust_index, update,
 		    heap);
   }
 
   if (table->n_v_cols)
-    row_upd_replace_vcol(row, table, rec_info->update, false,
+    row_upd_replace_vcol(row, table, update, false,
                          nullptr,
-                         (rec_info->cmpl_info & UPD_NODE_NO_ORD_CHANGE)
-	                 ? nullptr : rec_info->undo_rec);
+                         (cmpl_info & UPD_NODE_NO_ORD_CHANGE)
+	                 ? nullptr : undo_rec);
 
   bool success= true;
   dict_index_t *index= dict_table_get_next_index(clust_index);
diff --git a/storage/innobase/trx/trx0rec.cc b/storage/innobase/trx/trx0rec.cc
index 7d0adc0a6cd..c343972a149 100644
--- a/storage/innobase/trx/trx0rec.cc
+++ b/storage/innobase/trx/trx0rec.cc
@@ -2254,9 +2254,9 @@ trx_undo_get_undo_rec_low(
 	ut_ad(rseg->is_persistent());
 
 	if (undo_rec_info
-	    && undo_rec_info->block->page.id().page_no() == page_no) {
+	    && undo_rec_info->block.page.id().page_no() == page_no) {
 		undo_rec = trx_undo_rec_copy(
-			undo_rec_info->block->page.frame + offset, heap);
+			undo_rec_info->block.page.frame + offset, heap);
 	} else {
 		mtr.start();
 
diff --git a/storage/innobase/trx/trx0roll.cc b/storage/innobase/trx/trx0roll.cc
index 8fd9f93909c..cfda0fe7965 100644
--- a/storage/innobase/trx/trx0roll.cc
+++ b/storage/innobase/trx/trx0roll.cc
@@ -60,7 +60,7 @@ const trx_t*		trx_roll_crash_recv_trx;
 inline bool trx_t::rollback_finish()
 {
   mod_tables.clear();
-
+  apply_online_log= false;
   if (UNIV_LIKELY(error_state == DB_SUCCESS))
   {
     commit();
diff --git a/storage/innobase/trx/trx0trx.cc b/storage/innobase/trx/trx0trx.cc
index a1fcd5bbe29..ceaf7699028 100644
--- a/storage/innobase/trx/trx0trx.cc
+++ b/storage/innobase/trx/trx0trx.cc
@@ -1424,7 +1424,6 @@ TRANSACTIONAL_TARGET void trx_t::commit_low(mtr_t *mtr)
   if (mtr)
   {
     apply_log();
-    apply_online_log= false;
     trx_write_serialisation_history(this, mtr);
 
     /* The following call commits the mini-transaction, making the
diff --git a/storage/innobase/trx/trx0undo.cc b/storage/innobase/trx/trx0undo.cc
index ea2ded3463b..9796230353e 100644
--- a/storage/innobase/trx/trx0undo.cc
+++ b/storage/innobase/trx/trx0undo.cc
@@ -298,15 +298,17 @@ typedef std::map<table_id_t, dict_table_t*> table_id_map;
 @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,
+                                      trx_undo_rec_info &rec_info,
                                       dict_table_t *table,
+                                      ulint type,
+                                      ulint cmpl_info,
                                       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(undo_tuple, rec_info, index, heap);
+  row_log_insert(*undo_tuple, rec_info, index,
+                 type, cmpl_info, nullptr, rec, heap);
 }
 
 /** Apply the TRX_UNDO_UPD & TRX_UNDO_DEL undo log record
@@ -315,27 +317,31 @@ static void trx_undo_rec_apply_insert(trx_undo_rec_t *rec,
 @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,
+                                      trx_undo_rec_info &rec_info,
                                       dict_table_t *table,
-                                      mem_heap_t *heap)
+                                      ulint type,
+                                      ulint cmpl_info,
+				      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;
+  upd_t *update;
   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);
 
-  rec= trx_undo_update_rec_get_update(rec, index, rec_info->type,
+  rec= trx_undo_update_rec_get_update(rec, index, type,
                                       trx_id, roll_ptr, info_bits,
-                                      heap, &rec_info->update);
-  rec_info->undo_rec= rec;
-  if (rec_info->type == TRX_UNDO_UPD_DEL_REC)
-    row_log_insert(undo_tuple, rec_info, index, heap);
+                                      heap, &update);
+  if (type == TRX_UNDO_UPD_DEL_REC)
+    row_log_insert(*undo_tuple, rec_info, index, type, cmpl_info,
+                   update, rec, heap);
   else
-    row_log_update(undo_tuple, rec_info, index, heap);
+    row_log_update(*undo_tuple, rec_info, index, type, cmpl_info,
+                   update, rec, heap);
 }
 
 /** Apply all DML undo log records to the online DDL tables
@@ -346,7 +352,7 @@ static void trx_undo_rec_apply_update(trx_undo_rec_t *rec,
 				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,
+                                   trx_undo_rec_info &undo_info,
                                    mem_heap_t *heap)
 {
   ulint type, cmpl_info= 0;
@@ -359,16 +365,16 @@ static void trx_undo_rec_apply_log(trx_undo_rec_t *rec,
   if (it == online_log_tables.end() || !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);
+    trx_undo_rec_apply_insert(ptr, undo_info, it->second,
+                              type, cmpl_info, 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);
+    trx_undo_rec_apply_update(ptr, undo_info, it->second,
+                              type, cmpl_info, heap);
     break;
   default:
     ut_ad(0);
@@ -379,7 +385,7 @@ static void trx_undo_rec_apply_log(trx_undo_rec_t *rec,
 
 
 /** Apply any changes to tables for which online DDL is in progress. */
-ATTRIBUTE_COLD void trx_t::apply_log() const
+ATTRIBUTE_COLD void trx_t::apply_log()
 {
   if (undo_no == 0 || apply_online_log == false)
     return;
@@ -406,12 +412,10 @@ ATTRIBUTE_COLD void trx_t::apply_log() const
                                                      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);
+      trx_undo_rec_info undo_rec_info(*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);
+                                      page_id.page_no(), undo->hdr_offset);
     }
 
     uint32_t next= mach_read_from_4(TRX_UNDO_PAGE_HDR + TRX_UNDO_PAGE_NODE +
@@ -427,6 +431,7 @@ ATTRIBUTE_COLD void trx_t::apply_log() const
   }
   mtr.commit();
   mem_heap_free(heap);
+  apply_online_log= false;
 }
 
 /*============== UNDO LOG FILE COPY CREATION AND FREEING ==================*/
@@ -1523,6 +1528,6 @@ bool trx_undo_rec_info::is_equal(roll_ptr_t roll_ptr) const
 {
   uint16_t offset= static_cast<uint16_t>(roll_ptr);
   uint32_t page_no= static_cast<uint32_t>(roll_ptr >> 16);
-  return page_no == block->page.id().page_no()
+  return page_no == block.page.id().page_no()
          && offset == this->offset;
 }