MDEV-16232 Use fewer mini-transactions

- For update operation, InnoDB uses separate mtr for the following:
1) searching the key in primary key leaf page, stores the position
2) exclusive lock the primary key leaf page
3) Allocate the undo log page, write the undo log record
4) Modify the PRIMARY KEY index change

InnoDB should use the single mtr for the all above conditions.

sql/ha_partition.h

@@ -1636,6 +1636,9 @@ public:
   friend int cmp_key_part_id(void *key_p, uchar *ref1, uchar *ref2);
 
   bool can_convert_nocopy(const Field &field,
-                          const Column_definition &new_field) const override;
+                          const Column_definition &new_filed) const override;
+
+  void start_operations_batch() {}
+  void end_operations_batch() {}
 };
 #endif /* HA_PARTITION_INCLUDED */

sql/ha_sequence.h

@@ -161,5 +161,8 @@ public:
     file= file_arg;
     init();                                     /* Update cached_table_flags */
   }
+
+  void start_operations_batch() {}
+  void end_operations_batch() {}
 };
 #endif

sql/handler.cc

@@ -8748,3 +8748,13 @@ Table_scope_and_contents_source_st::fix_period_fields(THD *thd,
   }
   return false;
 }
+
+void handler::ha_start_operations_batch()
+{
+  table->file->start_operations_batch();
+}
+
+void handler::ha_end_operations_batch()
+{
+  table->file->end_operations_batch();
+}

sql/handler.h

@@ -3514,6 +3514,10 @@ public:
   int ha_drop_partitions(const char *path);
   int ha_rename_partitions(const char *path);
 
+  virtual void ha_start_operations_batch();
+  virtual void ha_end_operations_batch();
+  virtual void start_operations_batch()=0;
+  virtual void end_operations_batch()=0;
   void adjust_next_insert_id_after_explicit_value(ulonglong nr);
   int update_auto_increment();
   virtual void print_error(int error, myf errflag);

sql/json_table.cc

@@ -262,6 +262,8 @@ public:
     buf->length(0);
     return TRUE;
   }
+  void start_operations_batch() {}
+  void end_operations_batch() {}
 };
 
 

sql/sql_delete.cc

@@ -637,6 +637,7 @@ bool mysql_delete(THD *thd, TABLE_LIST *table_list, COND *conds,
       !table_list->has_period())
   {
     table->mark_columns_needed_for_delete();
+    table->file->start_operations_batch();
     if (!table->check_virtual_columns_marked_for_read())
     {
       DBUG_PRINT("info", ("Trying direct delete"));
@@ -658,10 +659,12 @@ bool mysql_delete(THD *thd, TABLE_LIST *table_list, COND *conds,
         THD_STAGE_INFO(thd, stage_updating);
         if (!(error= table->file->ha_direct_delete_rows(&deleted)))
           error= -1;
+        table->file->end_operations_batch();
         goto terminate_delete;
       }
     }
   }
+  table->file->end_operations_batch();
 
   if (query_plan.using_filesort)
   {

sql/sql_select.cc

@@ -21915,6 +21915,15 @@ sub_select(JOIN *join,JOIN_TAB *join_tab,bool end_of_records)
   if (pfs_batch_update)
     join_tab->table->file->start_psi_batch_mode();
 
+#if 0
+  bool ops_batch_started= false;
+  if (join_tab->next_select == end_send)
+  {
+    join_tab->table->file->start_operations_batch();
+    ops_batch_started= true;
+  }
+#endif
+
   if (rc != NESTED_LOOP_NO_MORE_ROWS)
   {
     error= (*join_tab->read_first_record)(join_tab);
@@ -21962,6 +21971,11 @@ sub_select(JOIN *join,JOIN_TAB *join_tab,bool end_of_records)
     rc= evaluate_join_record(join, join_tab, error);
   }
 
+#if 0
+  if (ops_batch_started)
+    join_tab->table->file->end_operations_batch();
+#endif
+
   if (rc == NESTED_LOOP_NO_MORE_ROWS &&
       join_tab->last_inner && !join_tab->found)
     rc= evaluate_null_complemented_join_record(join, join_tab);

sql/sql_update.cc

@@ -400,6 +400,7 @@ int mysql_update(THD *thd,
   TABLE_LIST *update_source_table;
   query_plan.index= MAX_KEY;
   query_plan.using_filesort= FALSE;
+  bool ops_batch_started= false;
 
   // For System Versioning (may need to insert new fields to a table).
   ha_rows rows_inserted= 0;
@@ -799,6 +800,8 @@ int mysql_update(THD *thd,
 
       note: We avoid sorting if we sort on the used index
     */
+    table->file->start_operations_batch();
+    ops_batch_started= true;
     if (query_plan.using_filesort)
     {
       /*
@@ -949,6 +952,8 @@ int mysql_update(THD *thd,
       table->file->ha_end_keyread();
       table->column_bitmaps_set(save_read_set, save_write_set);
     }
+    table->file->end_operations_batch();
+    ops_batch_started= false;
   }
 
 update_begin:
@@ -1015,6 +1020,8 @@ update_begin:
   THD_STAGE_INFO(thd, stage_updating);
   fix_rownum_pointers(thd, thd->lex->current_select, &updated_or_same);
   thd->get_stmt_da()->reset_current_row_for_warning(1);
+  table->file->start_operations_batch();
+  ops_batch_started= true;
   while (!(error=info.read_record()) && !thd->killed)
   {
     explain->tracker.on_record_read();
@@ -1228,6 +1235,9 @@ error:
       break;
     }
   }
+  table->file->end_operations_batch();
+  ops_batch_started= false;
+
   ANALYZE_STOP_TRACKING(thd, &explain->command_tracker);
   table->auto_increment_field_not_null= FALSE;
   dup_key_found= 0;
@@ -1373,6 +1383,8 @@ update_end:
   DBUG_RETURN((error >= 0 || thd->is_error()) ? 1 : 0);
 
 err:
+  if (ops_batch_started)
+    table->file->end_operations_batch();
   delete select;
   delete file_sort;
   free_underlaid_joins(thd, select_lex);

storage/archive/ha_archive.h

@@ -157,6 +157,8 @@ public:
   unsigned int pack_row(const uchar *record, azio_stream *writer);
   bool check_if_incompatible_data(HA_CREATE_INFO *info, uint table_changes);
   int external_lock(THD *thd, int lock_type);
+  void start_operations_batch() {}
+  void end_operations_batch() {}
 private:
   void flush_and_clear_pending_writes();
 };

storage/blackhole/ha_blackhole.h

@@ -98,6 +98,9 @@ public:
   {
     return 0;
   }
+
+  void start_operations_batch() {}
+  void end_operations_batch() {}
 private:
   virtual int write_row(const uchar *buf);
   virtual int update_row(const uchar *old_data, const uchar *new_data);

storage/csv/ha_tina.h

@@ -176,5 +176,8 @@ public:
   int encode_quote(const uchar *buf);
   int find_current_row(uchar *buf);
   int chain_append();
+
+  void start_operations_batch() {}
+  void end_operations_batch() {}
 };
 

storage/example/ha_example.h

@@ -251,4 +251,7 @@ public:
 
   THR_LOCK_DATA **store_lock(THD *thd, THR_LOCK_DATA **to,
                              enum thr_lock_type lock_type);     ///< required
+
+  void start_operations_batch() {}
+  void end_operations_batch() {}
 };

storage/federated/ha_federated.h

@@ -284,5 +284,8 @@ public:
   int connection_autocommit(bool state);
   int execute_simple_query(const char *query, int len);
   int reset(void);
+
+  void start_operations_batch() {}
+  void end_operations_batch() {}
 };
 

storage/federatedx/ha_federatedx.h

@@ -466,6 +466,9 @@ public:
   const FEDERATEDX_SHARE *get_federatedx_share() const { return share; }
   friend class ha_federatedx_derived_handler;
   friend class ha_federatedx_select_handler;
+
+  void start_operations_batch() {}
+  void end_operations_batch() {}
 };
 
 extern const char ident_quote_char;              // Character for quoting

storage/heap/ha_heap.h

@@ -120,6 +120,8 @@ public:
   }
   bool check_if_incompatible_data(HA_CREATE_INFO *info, uint table_changes);
   int find_unique_row(uchar *record, uint unique_idx);
+  void start_operations_batch() {}
+  void end_operations_batch() {}
 private:
   void update_key_stats();
 };

storage/innobase/handler/ha_innodb.cc

@@ -8737,8 +8737,12 @@ ha_innobase::unlock_row(void)
 			break;
 		/* fall through */
 	case ROW_READ_TRY_SEMI_CONSISTENT:
-		row_unlock_for_mysql(m_prebuilt, FALSE);
+	{
+		bool hold = (m_prebuilt->batch_mtr
+			     && !m_prebuilt->batch_mtr->cursor_stored());
+		row_unlock_for_mysql(m_prebuilt, hold);
 		break;
+	}
 	case ROW_READ_DID_SEMI_CONSISTENT:
 		m_prebuilt->row_read_type = ROW_READ_TRY_SEMI_CONSISTENT;
 		break;
@@ -20674,6 +20678,22 @@ Compare_keys ha_innobase::compare_key_parts(
   return Compare_keys::Equal;
 }
 
+void ha_innobase::start_operations_batch()
+{
+  ut_ad(!m_prebuilt->batch_mtr);
+  if (!m_prebuilt->table->is_temporary())
+    m_prebuilt->batch_mtr= new row_batch_mtr_t();
+}
+
+void ha_innobase::end_operations_batch()
+{
+  if (m_prebuilt->batch_mtr)
+  {
+    delete m_prebuilt->batch_mtr;
+    m_prebuilt->batch_mtr= nullptr;
+  }
+}
+
 /******************************************************************//**
 Use this when the args are passed to the format string from
 errmsg-utf8.txt directly as is.

storage/innobase/handler/ha_innodb.h

@@ -438,6 +438,8 @@ public:
 			  const KEY_PART_INFO& old_part,
 			  const KEY_PART_INFO& new_part) const override;
 
+	void start_operations_batch();
+	void end_operations_batch();
 protected:
 	bool
 	can_convert_string(const Field_string* field,

storage/innobase/include/row0mysql.h

@@ -277,7 +277,7 @@ releases the latest clustered index record lock we set.
 void
 row_unlock_for_mysql(
 	row_prebuilt_t*	prebuilt,
-	ibool		has_latches_on_recs);
+	bool		has_latches_on_recs);
 
 /*********************************************************************//**
 Creates an query graph node of 'update' type to be used in the MySQL
@@ -446,6 +446,105 @@ struct mysql_row_templ_t {
 #define ROW_PREBUILT_ALLOCATED	78540783
 #define ROW_PREBUILT_FREED	26423527
 
+/** mtr can be used for batch update and delete operations */
+class row_batch_mtr_t
+{
+private:
+  /* Indicate whether the cursor store operation */
+  bool m_batch_cursor_stored= false;
+  /* Indicate whether it does read operation */
+  bool m_read= false;
+  /* mtr used for batch operation */
+  mtr_t *m_batch_mtr= nullptr;
+  /* recent clustered index leaf page savepoint */
+  ulint m_last_clust_savepoint= 0;
+  /* whether the batch mini-transaction is active */
+  bool m_batch_mtr_active= false;
+public:
+  /** Start the batch mtr operation */
+  void start()
+  {
+    m_batch_mtr->start();
+    ut_ad(m_batch_mtr->is_active());
+    m_batch_mtr_active= true;
+  }
+
+  row_batch_mtr_t()
+  {
+    m_batch_mtr = new mtr_t();
+    start();
+  }
+
+  ~row_batch_mtr_t()
+  {
+    if (m_batch_mtr_active)
+      m_batch_mtr->commit();
+    delete m_batch_mtr;
+  }
+
+  mtr_t* get_mtr() { return m_batch_mtr; }
+
+  btr_latch_mode assign_operation(const dict_index_t *index, ulint lock_type);
+
+  /** @return whether cursor stored */
+  bool cursor_stored() { return m_batch_cursor_stored; }
+
+  /** Finish the read part of the batch mtr operation.
+  In case of read operation, InnoDB rollbacks to savepoint
+  when secondary index is being used
+  @param mtr_extra_clust_savepoint non-zero when secondary
+				   index is being used*/
+  void finish_read(ulint mtr_extra_clust_savepoint)
+  {
+    if (m_read)
+    {
+      /* Uses secondary index scan. In that case, remove
+      the clustered index leaf page latch */
+      if (mtr_extra_clust_savepoint)
+      {
+        m_batch_mtr->rollback_to_savepoint(mtr_extra_clust_savepoint);
+        mtr_extra_clust_savepoint= 0;
+      }
+    }
+
+    m_last_clust_savepoint= mtr_extra_clust_savepoint;
+    m_batch_cursor_stored= false;
+  }
+
+  /** Finish the update part of the batch mtr operation.
+  In case of update operation, commit and start the
+  batch mtr operation */
+  void finish_update()
+  {
+    /* Copy secondary and clustered index position */
+    m_batch_mtr->commit();
+    m_batch_cursor_stored= true;
+    m_batch_mtr->start();
+    m_last_clust_savepoint= 0;
+  }
+
+  /** Start read operation of batch mtr operation
+  In case of update operation, remove the clustered index
+  leaf page latch */
+  void start_read()
+  {
+    if (m_read)
+      return;
+    if (m_last_clust_savepoint)
+    {
+      m_batch_mtr->rollback_to_savepoint(m_last_clust_savepoint);
+      m_last_clust_savepoint= 0;
+    }
+  }
+
+  /** Commit the batch mtr operation */
+  void commit()
+  {
+    m_batch_mtr->commit();
+    m_batch_mtr_active= false;
+  }
+};
+
 /** A struct for (sometimes lazily) prebuilt structures in an Innobase table
 handle used within MySQL; these are used to save CPU time. */
 
@@ -685,6 +784,9 @@ struct row_prebuilt_t {
 	/** The MySQL table object */
 	TABLE*		m_mysql_table;
 
+	/** Batch update operation mtr */
+	row_batch_mtr_t	*batch_mtr;
+
 	/** Get template by dict_table_t::cols[] number */
 	const mysql_row_templ_t* get_template_by_col(ulint col) const
 	{

storage/innobase/row/row0mysql.cc

@@ -1644,7 +1644,7 @@ row_update_for_mysql(row_prebuilt_t* prebuilt)
 				      ? prebuilt->pcur
 				      : prebuilt->clust_pcur);
 
-	ut_a(node->pcur->rel_pos == BTR_PCUR_ON);
+	ut_a(node->pcur->rel_pos == BTR_PCUR_ON || prebuilt->batch_mtr);
 
 	/* MySQL seems to call rnd_pos before updating each row it
 	has cached: we can get the correct cursor position from
@@ -1757,7 +1757,7 @@ releases the latest clustered index record lock we set.
 void
 row_unlock_for_mysql(
 	row_prebuilt_t*	prebuilt,
-	ibool		has_latches_on_recs)
+	bool		has_latches_on_recs)
 {
 	if (prebuilt->new_rec_locks == 1 && prebuilt->index->is_clust()) {
 		trx_t* trx = prebuilt->trx;

storage/innobase/row/row0sel.cc

@@ -3337,6 +3337,18 @@ public:
                      dtuple_t **vrow, mtr_t *mtr);
 };
 
+/** Assign read operation using lock type
+@return latch mode for batch mtr operation */
+btr_latch_mode row_batch_mtr_t::assign_operation(
+  const dict_index_t *index, ulint lock_type)
+{
+  m_read= lock_type != LOCK_X;
+  if (index->is_clust())
+  { return m_read ? BTR_SEARCH_LEAF : BTR_MODIFY_LEAF; }
+
+  return BTR_SEARCH_LEAF;
+}
+
 /*********************************************************************//**
 Retrieves the clustered index record corresponding to a record in a
 non-clustered index. Does the necessary locking. Used in the MySQL
@@ -3370,6 +3382,14 @@ Row_sel_get_clust_rec_for_mysql::operator()(
 	dict_index_t*	clust_index;
 	rec_t*		old_vers;
 	trx_t*		trx;
+	btr_latch_mode	batch_latch_mode = BTR_SEARCH_LEAF;
+
+	if (row_batch_mtr_t *batch_mtr= prebuilt->batch_mtr) {
+		ut_ad(batch_mtr->get_mtr() == mtr);
+		if (prebuilt->select_lock_type == LOCK_X) {
+			batch_latch_mode = BTR_MODIFY_LEAF;
+		}
+	}
 
 	prebuilt->clust_pcur->old_rec = nullptr;
 	*out_rec = NULL;
@@ -3382,7 +3402,7 @@ Row_sel_get_clust_rec_for_mysql::operator()(
 	prebuilt->clust_pcur->btr_cur.page_cur.index = clust_index;
 
 	dberr_t err = btr_pcur_open_with_no_init(prebuilt->clust_ref,
-						 PAGE_CUR_LE, BTR_SEARCH_LEAF,
+						 PAGE_CUR_LE, batch_latch_mode,
 						 prebuilt->clust_pcur, mtr);
 	if (UNIV_UNLIKELY(err != DB_SUCCESS)) {
 		return err;
@@ -4367,6 +4387,7 @@ row_search_mvcc(
 	ibool		table_lock_waited		= FALSE;
 	byte*		next_buf			= 0;
 	bool		spatial_search			= false;
+	btr_latch_mode	batch_latch_mode = BTR_SEARCH_LEAF;
 
 	ut_ad(index && pcur && search_tuple);
 	ut_a(prebuilt->magic_n == ROW_PREBUILT_ALLOCATED);
@@ -4517,8 +4538,15 @@ early_not_found:
 	/* if the query is a plain locking SELECT, and the isolation level
 	is <= TRX_ISO_READ_COMMITTED, then this is set to FALSE */
 	bool did_semi_consistent_read = false;
-	mtr_t mtr;
-	mtr.start();
+	mtr_t *mtr = nullptr;
+	row_batch_mtr_t *batch_mtr = prebuilt->batch_mtr;
+	if (batch_mtr) {
+		mtr = batch_mtr->get_mtr();
+		batch_mtr->start_read();
+		batch_latch_mode = batch_mtr->assign_operation(
+				index, prebuilt->select_lock_type);
+	}
+	else { mtr= new mtr_t(); mtr->start(); }
 
 	mem_heap_t*	heap				= NULL;
 	rec_offs	offsets_[REC_OFFS_NORMAL_SIZE];
@@ -4558,7 +4586,7 @@ early_not_found:
 			dberr_t err = DB_SUCCESS;
 			switch (row_sel_try_search_shortcut_for_mysql(
 					&rec, prebuilt, &offsets, &heap,
-					&mtr)) {
+					mtr)) {
 			case SEL_FOUND:
 				/* At this point, rec is protected by
 				a page latch that was acquired by
@@ -4612,7 +4640,7 @@ aborted:
 			case SEL_EXHAUSTED:
 				err = DB_RECORD_NOT_FOUND;
 			shortcut_done:
-				mtr.commit();
+				mtr->commit();
 
 				/* NOTE that we do NOT store the cursor
 				position */
@@ -4621,6 +4649,7 @@ aborted:
 				if (UNIV_LIKELY_NULL(heap)) {
 					mem_heap_free(heap);
 				}
+				delete mtr;
 				DBUG_RETURN(err);
 
 			case SEL_RETRY:
@@ -4630,8 +4659,13 @@ aborted:
 				ut_ad(0);
 			}
 
-			mtr.commit();
-			mtr.start();
+			if (batch_mtr) {
+				batch_mtr->commit();
+				batch_mtr->start();
+			} else {
+				mtr->commit();
+				mtr->start();
+			}
 		}
 	}
 #endif /* BTR_CUR_HASH_ADAPT */
@@ -4733,13 +4767,17 @@ wait_table_again:
 			goto next_rec;
 		}
 
-		bool	need_to_process = sel_restore_position_for_mysql(
-			&same_user_rec, BTR_SEARCH_LEAF,
-			pcur, moves_up, &mtr);
+		bool	need_to_process = false;
+		if (batch_mtr && !batch_mtr->cursor_stored()) {
+		} else {
+			need_to_process = sel_restore_position_for_mysql(
+				&same_user_rec, batch_latch_mode,
+				pcur, moves_up, mtr);
+		}
 
 		if (UNIV_UNLIKELY(need_to_process)) {
 			if (UNIV_UNLIKELY(!btr_pcur_get_rec(pcur))) {
-				mtr.commit();
+				mtr->commit();
 				trx->op_info = "";
 				if (UNIV_LIKELY_NULL(heap)) {
 					mem_heap_free(heap);
@@ -4788,11 +4826,11 @@ wait_table_again:
 				prebuilt->rtr_info->search_mode = mode;
 			}
 
-			err = rtr_search_leaf(pcur, search_tuple, mode, &mtr);
+			err = rtr_search_leaf(pcur, search_tuple, mode, mtr);
 		} else {
 			err = btr_pcur_open_with_no_init(search_tuple, mode,
 							 BTR_SEARCH_LEAF,
-							 pcur, &mtr);
+							 pcur, mtr);
 		}
 
 		if (err != DB_SUCCESS) {
@@ -4825,7 +4863,7 @@ page_corrupted:
 			err = sel_set_rec_lock(pcur,
 					       next_rec, index, offsets,
 					       prebuilt->select_lock_type,
-					       LOCK_GAP, thr, &mtr);
+					       LOCK_GAP, thr, mtr);
 
 			switch (err) {
 			case DB_SUCCESS_LOCKED_REC:
@@ -4839,7 +4877,7 @@ page_corrupted:
 		}
 	} else if (mode == PAGE_CUR_G || mode == PAGE_CUR_L) {
 		err = pcur->open_leaf(mode == PAGE_CUR_G, index,
-				      BTR_SEARCH_LEAF, &mtr);
+				      BTR_SEARCH_LEAF, mtr);
 
 		if (err != DB_SUCCESS) {
 			if (err == DB_DECRYPTION_FAILED) {
@@ -4902,7 +4940,7 @@ rec_loop:
 	DEBUG_SYNC_C("row_search_rec_loop");
 	if (trx_is_interrupted(trx)) {
 		if (!spatial_search) {
-			btr_pcur_store_position(pcur, &mtr);
+			btr_pcur_store_position(pcur, mtr);
 		}
 		err = DB_INTERRUPTED;
 		goto normal_return;
@@ -4943,7 +4981,7 @@ rec_loop:
 			err = sel_set_rec_lock(pcur,
 					       rec, index, offsets,
 					       prebuilt->select_lock_type,
-					       LOCK_ORDINARY, thr, &mtr);
+					       LOCK_ORDINARY, thr, mtr);
 
 			switch (err) {
 			case DB_SUCCESS_LOCKED_REC:
@@ -5078,7 +5116,7 @@ wrong_offs:
 					pcur,
 					rec, index, offsets,
 					prebuilt->select_lock_type, LOCK_GAP,
-					thr, &mtr);
+					thr, mtr);
 
 				switch (err) {
 				case DB_SUCCESS_LOCKED_REC:
@@ -5089,7 +5127,7 @@ wrong_offs:
 				}
 			}
 
-			btr_pcur_store_position(pcur, &mtr);
+			btr_pcur_store_position(pcur, mtr);
 
 			/* The found record was not a match, but may be used
 			as NEXT record (index_next). Set the relative position
@@ -5114,7 +5152,7 @@ wrong_offs:
 					pcur,
 					rec, index, offsets,
 					prebuilt->select_lock_type, LOCK_GAP,
-					thr, &mtr);
+					thr, mtr);
 
 				switch (err) {
 				case DB_SUCCESS_LOCKED_REC:
@@ -5125,7 +5163,7 @@ wrong_offs:
 				}
 			}
 
-			btr_pcur_store_position(pcur, &mtr);
+			btr_pcur_store_position(pcur, mtr);
 
 			/* The found record was not a match, but may be used
 			as NEXT record (index_next). Set the relative position
@@ -5236,7 +5274,7 @@ no_gap_lock:
 		err = sel_set_rec_lock(pcur,
 				       rec, index, offsets,
 				       prebuilt->select_lock_type,
-				       lock_type, thr, &mtr);
+				       lock_type, thr, mtr);
 
 		switch (err) {
 			const rec_t*	old_vers;
@@ -5270,7 +5308,7 @@ no_gap_lock:
 				row_sel_build_committed_vers_for_mysql(
 					clust_index, prebuilt, rec,
 					&offsets, &heap, &old_vers,
-					need_vrow ? &vrow : NULL, &mtr);
+					need_vrow ? &vrow : NULL, mtr);
 			}
 
 			/* Check whether it was a deadlock or not, if not
@@ -5372,7 +5410,7 @@ no_gap_lock:
 				err = row_sel_build_prev_vers_for_mysql(
 					prebuilt, clust_index,
 					rec, &offsets, &heap, &old_vers,
-					need_vrow ? &vrow : nullptr, &mtr);
+					need_vrow ? &vrow : nullptr, mtr);
 
 				if (err != DB_SUCCESS) {
 
@@ -5501,7 +5539,7 @@ requires_clust_rec:
 		/* It was a non-clustered index and we must fetch also the
 		clustered index record */
 
-		mtr_extra_clust_savepoint = mtr.get_savepoint();
+		mtr_extra_clust_savepoint = mtr->get_savepoint();
 
 		ut_ad(!vrow);
 		/* The following call returns 'offsets' associated with
@@ -5512,7 +5550,7 @@ requires_clust_rec:
 						      thr, &clust_rec,
 						      &offsets, &heap,
 						      need_vrow ? &vrow : NULL,
-						      &mtr);
+						      mtr);
 		if (err == DB_LOCK_WAIT && prebuilt->skip_locked) {
 			err = lock_trx_handle_wait(trx);
 		}
@@ -5738,7 +5776,7 @@ idx_cond_failed:
 		/* Inside an update always store the cursor position */
 
 		if (!spatial_search) {
-			btr_pcur_store_position(pcur, &mtr);
+			  btr_pcur_store_position(pcur, mtr);
 		}
 	}
 
@@ -5774,14 +5812,14 @@ next_rec_after_check:
 
 	if (spatial_search) {
 		/* No need to do store restore for R-tree */
-		mtr.rollback_to_savepoint(0);
+		mtr->rollback_to_savepoint(0);
 	} else if (mtr_extra_clust_savepoint) {
 		/* We must release any clustered index latches
 		if we are moving to the next non-clustered
 		index record, because we could break the latching
 		order if we would access a different clustered
 		index page right away without releasing the previous. */
-		mtr.rollback_to_savepoint(mtr_extra_clust_savepoint);
+		mtr->rollback_to_savepoint(mtr_extra_clust_savepoint);
 	}
 
 	mtr_extra_clust_savepoint = 0;
@@ -5789,7 +5827,7 @@ next_rec_after_check:
 	if (moves_up) {
 		if (UNIV_UNLIKELY(spatial_search)) {
 			if (rtr_pcur_move_to_next(
-				    search_tuple, mode, pcur, 0, &mtr)) {
+				    search_tuple, mode, pcur, 0, mtr)) {
 				goto rec_loop;
 			}
 		} else {
@@ -5801,7 +5839,7 @@ next_rec_after_check:
 				if (btr_pcur_is_after_last_in_tree(pcur)) {
 					goto not_moved;
 				}
-				err = btr_pcur_move_to_next_page(pcur, &mtr);
+				err = btr_pcur_move_to_next_page(pcur, mtr);
 				if (err != DB_SUCCESS) {
 					goto lock_wait_or_error;
 				}
@@ -5812,7 +5850,7 @@ next_rec_after_check:
 			goto rec_loop;
 		}
 	} else {
-		if (btr_pcur_move_to_prev(pcur, &mtr)) {
+		if (btr_pcur_move_to_prev(pcur, mtr)) {
 			goto rec_loop;
 		}
 		if (UNIV_UNLIKELY(!btr_pcur_get_rec(pcur))) {
@@ -5824,7 +5862,7 @@ corrupted:
 
 not_moved:
 	if (!spatial_search) {
-		btr_pcur_store_position(pcur, &mtr);
+		btr_pcur_store_position(pcur, mtr);
 	}
 
 	err = match_mode ? DB_RECORD_NOT_FOUND : DB_END_OF_INDEX;
@@ -5832,7 +5870,7 @@ not_moved:
 
 lock_wait_or_error:
 	if (!dict_index_is_spatial(index)) {
-		btr_pcur_store_position(pcur, &mtr);
+		btr_pcur_store_position(pcur, mtr);
 	}
 page_read_error:
 	/* Reset the old and new "did semi-consistent read" flags. */
@@ -5843,7 +5881,11 @@ page_read_error:
 	did_semi_consistent_read = false;
 
 lock_table_wait:
-	mtr.commit();
+	if (batch_mtr) {
+		batch_mtr->commit();
+	} else {
+		mtr->commit();
+	}
 	mtr_extra_clust_savepoint = 0;
 
 	trx->error_state = err;
@@ -5853,7 +5895,11 @@ lock_table_wait:
 		/* It was a lock wait, and it ended */
 
 		thr->lock_state = QUE_THR_LOCK_NOLOCK;
-		mtr.start();
+		if (batch_mtr) {
+			batch_mtr->start();
+		} else {
+			mtr->start();
+		}
 
 		/* Table lock waited, go try to obtain table lock
 		again */
@@ -5865,8 +5911,8 @@ lock_table_wait:
 
 		if (!dict_index_is_spatial(index)) {
 			sel_restore_position_for_mysql(
-				&same_user_rec, BTR_SEARCH_LEAF, pcur,
-				moves_up, &mtr);
+				&same_user_rec, batch_latch_mode, pcur,
+				moves_up, mtr);
 		}
 
 		if (trx->isolation_level <= TRX_ISO_READ_COMMITTED
@@ -5901,7 +5947,11 @@ lock_table_wait:
 	goto func_exit;
 
 normal_return:
-	mtr.commit();
+	if (!batch_mtr) {
+		mtr->commit();
+	} else {
+		batch_mtr->finish_read(mtr_extra_clust_savepoint);
+	}
 
 	DEBUG_SYNC_C("row_search_for_mysql_before_return");
 
@@ -5955,6 +6005,10 @@ func_exit:
 		}
 	}
 
+	if (!batch_mtr) {
+		delete mtr;
+	}
+
 	DEBUG_SYNC_C("innodb_row_search_for_mysql_exit");
 
 	DBUG_RETURN(err);

storage/innobase/row/row0upd.cc

@@ -216,12 +216,16 @@ row_upd_check_references_constraints(
 
 	entry = row_rec_to_index_entry(rec, index, offsets, heap);
 
-	mtr_commit(mtr);
-
-	DEBUG_SYNC_C("foreign_constraint_check_for_update");
-
-	mtr->start();
-
+	if (thr->prebuilt && thr->prebuilt->upd_node == node
+	    && thr->prebuilt->batch_mtr && index->is_clust()) {
+		/* During foreign key constraint check, commit the
+		batch mtr operation */
+		thr->prebuilt->batch_mtr->finish_update();
+	} else {
+		mtr_commit(mtr);
+		DEBUG_SYNC_C("foreign_constraint_check_for_update");
+		mtr->start();
+	}
 	DEBUG_SYNC_C_IF_THD(thr_get_trx(thr)->mysql_thd,
 			    "foreign_constraint_check_for_insert");
 
@@ -2546,13 +2550,18 @@ row_upd_clust_step(
 	dict_index_t*	index;
 	btr_pcur_t*	pcur;
 	dberr_t		err;
-	mtr_t		mtr;
+	mtr_t		*mtr;
 	rec_t*		rec;
 	mem_heap_t*	heap	= NULL;
 	rec_offs	offsets_[REC_OFFS_NORMAL_SIZE];
 	rec_offs*	offsets;
 	ulint		flags;
 	trx_t*		trx = thr_get_trx(thr);
+	row_batch_mtr_t *batch_mtr = nullptr;
+
+	if (thr->prebuilt && thr->prebuilt->upd_node == node) {
+		batch_mtr = thr->prebuilt->batch_mtr;
+	}
 
 	rec_offs_init(offsets_);
 
@@ -2569,9 +2578,13 @@ row_upd_clust_step(
 
 	pcur = node->pcur;
 
-	/* We have to restore the cursor to its position */
-
-	mtr.start();
+	if (batch_mtr) {
+		mtr = batch_mtr->get_mtr();
+	} else {
+		mtr = new mtr_t();
+		/* We have to restore the cursor to its position */
+		mtr->start();
+	}
 
 	if (node->table->is_temporary()) {
 		/* Disable locking, because temporary tables are
@@ -2580,10 +2593,14 @@ row_upd_clust_step(
 			? BTR_NO_ROLLBACK
 			: BTR_NO_LOCKING_FLAG;
 		/* Redo logging only matters for persistent tables. */
-		mtr.set_log_mode(MTR_LOG_NO_REDO);
+		mtr->set_log_mode(MTR_LOG_NO_REDO);
 	} else {
 		flags = node->table->no_rollback() ? BTR_NO_ROLLBACK : 0;
-		index->set_modified(mtr);
+		index->set_modified(*mtr);
+	}
+
+	if (batch_mtr) {
+		goto read_record;
 	}
 
 	/* If the restoration does not succeed, then the same
@@ -2603,16 +2620,17 @@ row_upd_clust_step(
 	if (dict_index_is_online_ddl(index)) {
 		ut_ad(node->table->id != DICT_INDEXES_ID);
 		mode = BTR_MODIFY_LEAF_ALREADY_LATCHED;
-		mtr_s_lock_index(index, &mtr);
+		mtr_s_lock_index(index, mtr);
 	} else {
 		mode = BTR_MODIFY_LEAF;
 	}
 
-	if (pcur->restore_position(mode, &mtr) != btr_pcur_t::SAME_ALL) {
+	if (pcur->restore_position(mode, mtr) != btr_pcur_t::SAME_ALL) {
 		err = DB_RECORD_NOT_FOUND;
 		goto exit_func;
 	}
 
+read_record:
 	rec = btr_pcur_get_rec(pcur);
 	offsets = rec_get_offsets(rec, index, offsets_, index->n_core_fields,
 				  ULINT_UNDEFINED, &heap);
@@ -2638,7 +2656,7 @@ row_upd_clust_step(
 #ifdef WITH_WSREP
 			foreign,
 #endif
-			&mtr);
+			mtr);
 		goto all_done;
 	}
 
@@ -2655,7 +2673,7 @@ row_upd_clust_step(
 
 	if (!node->is_delete && node->cmpl_info & UPD_NODE_NO_ORD_CHANGE) {
 		err = row_upd_clust_rec(
-			flags, node, index, offsets, &heap, thr, &mtr);
+			flags, node, index, offsets, &heap, thr, mtr);
 		goto exit_func;
 	}
 
@@ -2684,7 +2702,7 @@ row_upd_clust_step(
 #ifdef WITH_WSREP
 			foreign,
 #endif
-			&mtr);
+			mtr);
 all_done:
 		if (err == DB_SUCCESS) {
 			node->state = UPD_NODE_UPDATE_ALL_SEC;
@@ -2693,19 +2711,26 @@ success:
 		}
 	} else {
 		err = row_upd_clust_rec(
-			flags, node, index, offsets, &heap, thr, &mtr);
+			flags, node, index, offsets, &heap, thr, mtr);
 
 		if (err == DB_SUCCESS) {
 			ut_ad(node->is_delete != PLAIN_DELETE);
 			node->state = node->is_delete
 				? UPD_NODE_UPDATE_ALL_SEC
 				: UPD_NODE_UPDATE_SOME_SEC;
+
 			goto success;
 		}
 	}
 
 exit_func:
-	mtr.commit();
+	if (batch_mtr) {
+		batch_mtr->finish_update();
+	} else {
+		mtr->commit();
+		delete mtr;
+	}
+
 	if (UNIV_LIKELY_NULL(heap)) {
 		mem_heap_free(heap);
 	}

storage/innobase/trx/trx0rec.cc

@@ -1874,14 +1874,29 @@ trx_undo_report_row_operation(
 		bulk = false;
 	}
 
-	mtr_t		mtr;
-	mtr.start();
+	mtr_t	*mtr;
+	row_batch_mtr_t *batch_mtr = nullptr;
+
+	/* Use batch mtr for undo log page allocation
+	and writing the undo log record */
+	if (thr->prebuilt && thr->run_node
+	    && thr->run_node == thr->prebuilt->upd_node) {
+		batch_mtr = thr->prebuilt->batch_mtr;
+	}
+
+	if (batch_mtr) {
+		mtr = batch_mtr->get_mtr();
+	} else {
+		mtr = new mtr_t();
+		mtr->start();
+	}
+
 	trx_undo_t**	pundo;
 	trx_rseg_t*	rseg;
 	const bool	is_temp	= index->table->is_temporary();
 
 	if (is_temp) {
-		mtr.set_log_mode(MTR_LOG_NO_REDO);
+		mtr->set_log_mode(MTR_LOG_NO_REDO);
 
 		rseg = trx->get_temp_rseg();
 		pundo = &trx->rsegs.m_noredo.undo;
@@ -1894,12 +1909,17 @@ trx_undo_report_row_operation(
 
 	dberr_t		err;
 	buf_block_t*	undo_block = trx_undo_assign_low(trx, rseg, pundo,
-							 &err, &mtr);
+							 &err, mtr);
 	trx_undo_t*	undo	= *pundo;
 	ut_ad((err == DB_SUCCESS) == (undo_block != NULL));
 	if (UNIV_UNLIKELY(undo_block == NULL)) {
 err_exit:
-		mtr.commit();
+		if (batch_mtr) {
+			batch_mtr->commit();
+		} else {
+			mtr->commit();
+			delete mtr;
+		}
 		return err;
 	}
 
@@ -1908,11 +1928,11 @@ err_exit:
 	do {
 		uint16_t offset = !rec
 			? trx_undo_page_report_insert(
-				undo_block, trx, index, clust_entry, &mtr,
+				undo_block, trx, index, clust_entry, mtr,
 				bulk)
 			: trx_undo_page_report_modify(
 				undo_block, trx, index, rec, offsets, update,
-				cmpl_info, clust_entry, &mtr);
+				cmpl_info, clust_entry, mtr);
 
 		if (UNIV_UNLIKELY(offset == 0)) {
 			const uint16_t first_free = mach_read_from_2(
@@ -1937,14 +1957,20 @@ err_exit:
 				first, because it may be holding lower-level
 				latches, such as SYNC_FSP_PAGE. */
 
-				mtr.commit();
-				mtr.start();
+				if (batch_mtr) {
+					batch_mtr->commit();
+					batch_mtr->start();
+				} else {
+					mtr->commit();
+					mtr->start();
+				}
+
 				if (is_temp) {
-					mtr.set_log_mode(MTR_LOG_NO_REDO);
+					mtr->set_log_mode(MTR_LOG_NO_REDO);
 				}
 
 				rseg->latch.wr_lock(SRW_LOCK_CALL);
-				err = trx_undo_free_last_page(undo, &mtr);
+				err = trx_undo_free_last_page(undo, mtr);
 				rseg->latch.wr_unlock();
 
 				if (m.second) {
@@ -1969,16 +1995,22 @@ err_exit:
 				data file and initialized it based on
 				redo log records (which included the
 				write of the previous garbage). */
-				mtr.memset(*undo_block, first_free,
+				mtr->memset(*undo_block, first_free,
 					   srv_page_size - first_free
 					   - FIL_PAGE_DATA_END, 0);
 			}
 
-			mtr.commit();
+			if (batch_mtr) {
+			} else {
+				mtr->commit();
+			}
 		} else {
 			/* Success */
 			undo->top_page_no = undo_block->page.id().page_no();
-			mtr.commit();
+			if (batch_mtr) {
+			} else {
+				mtr->commit();
+			}
 			undo->top_offset  = offset;
 			undo->top_undo_no = trx->undo_no++;
 			undo->guess_block = undo_block;
@@ -2003,6 +2035,10 @@ err_exit:
 					undo->top_page_no, offset);
 			}
 
+			if (!batch_mtr) {
+				delete mtr;
+			}
+
 			return(DB_SUCCESS);
 		}
 
@@ -2011,13 +2047,15 @@ err_exit:
 		/* We have to extend the undo log by one page */
 
 		ut_ad(++loop_count < 2);
-		mtr.start();
+		if (!batch_mtr) {
+			mtr->start();
+		}
 
 		if (is_temp) {
-			mtr.set_log_mode(MTR_LOG_NO_REDO);
+			mtr->set_log_mode(MTR_LOG_NO_REDO);
 		}
 
-		undo_block = trx_undo_add_page(undo, &mtr, &err);
+		undo_block = trx_undo_add_page(undo, mtr, &err);
 
 		DBUG_EXECUTE_IF("ib_err_ins_undo_page_add_failure",
 				undo_block = NULL;);

storage/maria/ha_maria.h

@@ -190,6 +190,8 @@ public:
   virtual S3_INFO *s3_open_args() { return 0; }
   virtual void register_handler(MARIA_HA *file) {}
 
+  void start_operations_batch() {}
+  void end_operations_batch() {}
 private:
   DsMrr_impl ds_mrr;
   friend check_result_t index_cond_func_maria(void *arg);

storage/myisam/ha_myisam.h

@@ -174,6 +174,8 @@ public:
   Item *idx_cond_push(uint keyno, Item* idx_cond);
   bool rowid_filter_push(Rowid_filter* rowid_filter);
 
+  void start_operations_batch() {}
+  void end_operations_batch() {}
 private:
   DsMrr_impl ds_mrr;
   friend check_result_t index_cond_func_myisam(void *arg);

storage/myisammrg/ha_myisammrg.h

@@ -159,4 +159,6 @@ public:
                                           Query_cache_block_table **block,
                                           uint *n);
   virtual void set_lock_type(enum thr_lock_type lock);
+  void start_operations_batch() {}
+  void end_operations_batch() {}
 };

storage/perfschema/ha_perfschema.h

@@ -204,6 +204,8 @@ public:
 
   virtual void print_error(int error, myf errflags);
 
+  void start_operations_batch() {}
+  void end_operations_batch() {}
 private:
   /**
      Check if the caller is a replication thread or the caller is called

storage/sequence/sequence.cc

@@ -104,6 +104,8 @@ public:
   double read_time(uint index, uint ranges, ha_rows rows) { return (double)rows; }
   double keyread_time(uint index, uint ranges, ha_rows rows) { return (double)rows; }
 
+  void start_operations_batch() {}
+  void end_operations_batch() {}
 private:
   void set(uchar *buf);
   ulonglong nvalues() { return (seqs->to - seqs->from)/seqs->step; }

storage/sphinx/ha_sphinx.h

@@ -135,6 +135,8 @@ public:
 #endif	
 	virtual void			cond_pop ();
 
+	void start_operations_batch() {}
+	void end_operations_batch() {}
 private:
 	uint32			m_iFields;
 	char **			m_dFields;

storage/spider/ha_spider.h

@@ -791,4 +791,6 @@ public:
   int append_lock_tables_list();
   int lock_tables();
   int dml_init();
+  void start_operations_batch() {}
+  void end_operations_batch() {}
 };

storage/test_sql_discovery/test_sql_discovery.cc

@@ -93,6 +93,8 @@ public:
 
   int open(const char *name, int mode, uint test_if_locked);
   int close(void);
+  void start_operations_batch() {}
+  void end_operations_batch() {}
 };
 
 TSD_share *ha_tsd::get_share()