MDEV-30100 preparation: Simplify InnoDB transaction commit further

trx_commit_complete_for_mysql(): Remove some conditions.
We will rely on trx_t::commit_lsn.

trx_t::must_flush_log_later: Remove. trx_commit_complete_for_mysql()
can simply check for trx_t::flush_log_later.

trx_t::commit_in_memory(): Set commit_lsn=0 if the log was written.

trx_flush_log_if_needed_low(): Renamed to trx_flush_log_if_needed().
Assert that innodb_flush_log_at_trx_commit!=0 was checked by
the caller and that the transaction is not in XA PREPARE state.
Unconditionally flush the log for data dictionary transactions,
to ensure the correct processing of ddl_recovery.log.

trx_write_serialisation_history(): Move some code from
trx_purge_add_undo_to_history().

trx_prepare(): Invoke log_write_up_to() directly if needed.

innobase_commit_ordered_2(): Simplify some conditions.
A read-write transaction will always carry nonzero trx_t::id.
Let us unconditionally reset mysql_log_file_name, flush_log_later
after trx_t::commit() was invoked.

storage/innobase/handler/ha_innodb.cc

@@ -4462,9 +4462,7 @@ innobase_commit_ordered_2(
 {
 	DBUG_ENTER("innobase_commit_ordered_2");
 
-	const bool read_only = trx->read_only || trx->id == 0;
-
-	if (!read_only) {
+	if (trx->id) {
 		/* The following call reads the binary log position of
 		the transaction being committed.
 
@@ -4494,11 +4492,8 @@ innobase_commit_ordered_2(
 #endif /* WITH_WSREP */
 
 	innobase_commit_low(trx);
-
-	if (!read_only) {
-		trx->mysql_log_file_name = NULL;
-		trx->flush_log_later = false;
-	}
+	trx->mysql_log_file_name = NULL;
+	trx->flush_log_later = false;
 
 	DBUG_VOID_RETURN;
 }

storage/innobase/include/trx0trx.h

@@ -181,13 +181,9 @@ note that the trx may have been committed before the caller acquires
 trx_t::mutex
 @retval	NULL if no match */
 trx_t* trx_get_trx_by_xid(const XID* xid);
-/**********************************************************************//**
-If required, flushes the log to disk if we called trx_commit_for_mysql()
-with trx->flush_log_later == TRUE. */
-void
-trx_commit_complete_for_mysql(
-/*==========================*/
-	trx_t*	trx);	/*!< in/out: transaction */
+/** Durably write log until trx->commit_lsn
+(if trx_t::commit_in_memory() was invoked with flush_log_later=true). */
+void trx_commit_complete_for_mysql(trx_t *trx);
 /**********************************************************************//**
 Marks the latest SQL statement ended. */
 void
@@ -772,12 +768,6 @@ struct trx_t : ilist_node<>
 					defer flush of the logs to disk
 					until after we release the
 					mutex. */
-	bool		must_flush_log_later;/*!< set in commit()
-					if flush_log_later was
-					set and redo log was written;
-					in that case we will
-					flush the log in
-					trx_commit_complete_for_mysql() */
 	ulint		duplicates;	/*!< TRX_DUP_IGNORE | TRX_DUP_REPLACE */
   /** whether this modifies InnoDB dictionary tables */
   bool dict_operation;

storage/innobase/trx/trx0purge.cc

@@ -268,12 +268,7 @@ trx_purge_add_undo_to_history(const trx_t* trx, trx_undo_t*& undo, mtr_t* mtr)
 
   ut_ad(mach_read_from_2(undo_header + TRX_UNDO_NEEDS_PURGE) <= 1);
   ut_ad(rseg->needs_purge > trx->id);
-
-  if (rseg->last_page_no == FIL_NULL)
-  {
-    rseg->last_page_no= undo->hdr_page_no;
-    rseg->set_last_commit(undo->hdr_offset, trx->rw_trx_hash_element->no);
-  }
+  ut_ad(rseg->last_page_no != FIL_NULL);
 
   rseg->history_size++;
 

storage/innobase/trx/trx0trx.cc

@@ -419,7 +419,6 @@ void trx_t::free()
   MEM_NOACCESS(&active_commit_ordered, sizeof active_commit_ordered);
   MEM_NOACCESS(&check_unique_secondary, sizeof check_unique_secondary);
   MEM_NOACCESS(&flush_log_later, sizeof flush_log_later);
-  MEM_NOACCESS(&must_flush_log_later, sizeof must_flush_log_later);
   MEM_NOACCESS(&duplicates, sizeof duplicates);
   MEM_NOACCESS(&dict_operation, sizeof dict_operation);
   MEM_NOACCESS(&dict_operation_lock_mode, sizeof dict_operation_lock_mode);
@@ -993,12 +992,14 @@ static void trx_write_serialisation_history(trx_t *trx, mtr_t *mtr)
     {
       mysql_mutex_lock(&purge_sys.pq_mutex);
       trx_sys.assign_new_trx_no(trx);
+      const trx_id_t end{trx->rw_trx_hash_element->no};
       /* If the rollback segment is not empty, trx->no cannot be less
       than any trx_t::no already in rseg. User threads only produce
       events when a rollback segment is empty. */
-      purge_sys.purge_queue.push(TrxUndoRsegs(trx->rw_trx_hash_element->no,
-                                              *rseg));
+      purge_sys.purge_queue.push(TrxUndoRsegs{end, *rseg});
       mysql_mutex_unlock(&purge_sys.pq_mutex);
+      rseg->last_page_no= undo->hdr_page_no;
+      rseg->set_last_commit(undo->hdr_offset, end);
     }
     else
       trx_sys.assign_new_trx_no(trx);
@@ -1086,50 +1087,30 @@ extern "C" void  thd_decrement_pending_ops(MYSQL_THD);
   @param trx   transaction; if trx->state is PREPARED, the function will
   also wait for the flush to complete.
 */
-static void trx_flush_log_if_needed_low(lsn_t lsn, const trx_t *trx)
+static void trx_flush_log_if_needed(lsn_t lsn, trx_t *trx)
 {
-  if (!srv_flush_log_at_trx_commit)
-    return;
+  ut_ad(srv_flush_log_at_trx_commit);
+  ut_ad(trx->state != TRX_STATE_PREPARED);
 
   if (log_sys.get_flushed_lsn() > lsn)
     return;
 
-  const bool flush= srv_file_flush_method != SRV_NOSYNC &&
-    (srv_flush_log_at_trx_commit & 1);
-
-  if (trx->state == TRX_STATE_PREPARED)
-  {
-    /* XA, which is used with binlog as well.
-    Be conservative, use synchronous wait.*/
-sync:
-    log_write_up_to(lsn, flush);
-    return;
-  }
+  const bool flush= trx->dict_operation ||
+    (srv_file_flush_method != SRV_NOSYNC &&
+     (srv_flush_log_at_trx_commit & 1));
 
   completion_callback cb;
-  if ((cb.m_param = thd_increment_pending_ops(trx->mysql_thd)))
+  if ((cb.m_param= thd_increment_pending_ops(trx->mysql_thd)))
   {
     cb.m_callback = (void (*)(void *)) thd_decrement_pending_ops;
     log_write_up_to(lsn, flush, false, &cb);
   }
   else
-    goto sync;
-}
-
-/**********************************************************************//**
-If required, flushes the log to disk based on the value of
-innodb_flush_log_at_trx_commit. */
-static
-void
-trx_flush_log_if_needed(
-/*====================*/
-	lsn_t	lsn,	/*!< in: lsn up to which logs are to be
-			flushed. */
-	trx_t*	trx)	/*!< in/out: transaction */
-{
-	trx->op_info = "flushing log";
-	trx_flush_log_if_needed_low(lsn, trx);
-	trx->op_info = "";
+  {
+    trx->op_info= "flushing log";
+    log_write_up_to(lsn, flush);
+    trx->op_info= "";
+  }
 }
 
 /** Process tables that were modified by the committing transaction. */
@@ -1239,7 +1220,6 @@ TRANSACTIONAL_INLINE inline void trx_t::commit_in_memory(const mtr_t *mtr)
 {
   /* We already detached from rseg in trx_write_serialisation_history() */
   ut_ad(!rsegs.m_redo.undo);
-  must_flush_log_later= false;
   read_view.close();
 
   if (is_autocommit_non_locking())
@@ -1342,13 +1322,11 @@ TRANSACTIONAL_INLINE inline void trx_t::commit_in_memory(const mtr_t *mtr)
     gathering. */
 
     commit_lsn= undo_no || !xid.is_null() ? mtr->commit_lsn() : 0;
-    if (!commit_lsn)
-      /* Nothing to be done. */;
-    else if (flush_log_later)
-      /* Do nothing yet */
-      must_flush_log_later= true;
-    else if (srv_flush_log_at_trx_commit)
+    if (commit_lsn && !flush_log_later && srv_flush_log_at_trx_commit)
+    {
       trx_flush_log_if_needed(commit_lsn, this);
+      commit_lsn= 0;
+    }
   }
 
   savepoints_discard();
@@ -1595,24 +1573,21 @@ trx_commit_for_mysql(
 	return(DB_CORRUPTION);
 }
 
-/**********************************************************************//**
-If required, flushes the log to disk if we called trx_commit_for_mysql()
-with trx->flush_log_later == TRUE. */
-void
-trx_commit_complete_for_mysql(
-/*==========================*/
-	trx_t*	trx)	/*!< in/out: transaction */
+/** Durably write log until trx->commit_lsn
+(if trx_t::commit_in_memory() was invoked with flush_log_later=true). */
+void trx_commit_complete_for_mysql(trx_t *trx)
 {
-	if (trx->id != 0
-	    || !trx->must_flush_log_later
-	    || (srv_flush_log_at_trx_commit == 1 && trx->active_commit_ordered)) {
-
-		return;
-	}
-
-	trx_flush_log_if_needed(trx->commit_lsn, trx);
-
-	trx->must_flush_log_later = false;
+  const lsn_t lsn= trx->commit_lsn;
+  if (!lsn)
+    return;
+  switch (srv_flush_log_at_trx_commit) {
+  case 0:
+    return;
+  case 1:
+    if (trx->active_commit_ordered)
+      return;
+  }
+  trx_flush_log_if_needed(lsn, trx);
 }
 
 /**********************************************************************//**
@@ -1873,8 +1848,10 @@ trx_prepare(
 		gather behind one doing the physical log write to disk.
 
 		We must not be holding any mutexes or latches here. */
-
-		trx_flush_log_if_needed(lsn, trx);
+		if (auto f = srv_flush_log_at_trx_commit) {
+			log_write_up_to(lsn, (f & 1) && srv_file_flush_method
+					!= SRV_NOSYNC);
+		}
 
 		if (!UT_LIST_GET_LEN(trx->lock.trx_locks)
 		    || trx->isolation_level == TRX_ISO_SERIALIZABLE) {