MDEV-26782 InnoDB temporary tablespace: reclaiming of free space does not work

The motivation of this change is to allow undo pages for temporary tables
to be marked free as often as possible, so that we can avoid buf_pool.LRU
eviction of undo pages that contain data that is no longer needed.
For temporary tables, no MVCC or purge of history is needed, and
reusing cached undo log pages might not help that much.

It is possible that this may cause some performance regression due to
more frequent allocation and freeing of undo log pages.

trx_write_serialisation_history(): Never cache temporary undo log pages.

trx_undo_reuse_cached(): Assert that the rollback segment is persistent.

trx_undo_assign_low(): Add template<bool is_temp>. Never invoke
trx_undo_reuse_cached() for temporary tables.

4 files changed, 78 insertions, 64 deletions

diff --git a/storage/innobase/include/trx0undo.h b/storage/innobase/include/trx0undo.h
index 3474a903f6c..670fe00c25b 100644
--- a/storage/innobase/include/trx0undo.h
+++ b/storage/innobase/include/trx0undo.h
@@ -203,16 +203,18 @@ trx_undo_assign(trx_t* trx, dberr_t* err, mtr_t* mtr)
 	MY_ATTRIBUTE((nonnull));
 /** Assign an undo log for a transaction.
 A new undo log is created or a cached undo log reused.
+@tparam is_temp  whether this is temporary undo log
 @param[in,out]	trx	transaction
 @param[in]	rseg	rollback segment
 @param[out]	undo	the undo log
-@param[out]	err	error code
 @param[in,out]	mtr	mini-transaction
+@param[out]	err	error code
 @return	the undo log block
-@retval	NULL	on error */
+@retval	nullptr	on error */
+template<bool is_temp>
 buf_block_t*
-trx_undo_assign_low(trx_t* trx, trx_rseg_t* rseg, trx_undo_t** undo,
-		    dberr_t* err, mtr_t* mtr)
+trx_undo_assign_low(trx_t *trx, trx_rseg_t *rseg, trx_undo_t **undo,
+                    mtr_t *mtr, dberr_t *err)
 	MY_ATTRIBUTE((nonnull, warn_unused_result));
 /******************************************************************//**
 Sets the state of the undo log segment at a transaction finish.
diff --git a/storage/innobase/trx/trx0rec.cc b/storage/innobase/trx/trx0rec.cc
index dc24f083d05..d3f64754d89 100644
--- a/storage/innobase/trx/trx0rec.cc
+++ b/storage/innobase/trx/trx0rec.cc
@@ -1868,26 +1868,28 @@ trx_undo_report_row_operation(
 	}
 
 	mtr_t		mtr;
+	dberr_t		err;
 	mtr.start();
 	trx_undo_t**	pundo;
 	trx_rseg_t*	rseg;
 	const bool	is_temp	= index->table->is_temporary();
+	buf_block_t*	undo_block;
 
 	if (is_temp) {
 		mtr.set_log_mode(MTR_LOG_NO_REDO);
-
 		rseg = trx->get_temp_rseg();
 		pundo = &trx->rsegs.m_noredo.undo;
+		undo_block = trx_undo_assign_low<true>(trx, rseg, pundo,
+						       &mtr, &err);
 	} else {
 		ut_ad(!trx->read_only);
 		ut_ad(trx->id);
 		pundo = &trx->rsegs.m_redo.undo;
 		rseg = trx->rsegs.m_redo.rseg;
+		undo_block = trx_undo_assign_low<false>(trx, rseg, pundo,
+							&mtr, &err);
 	}
 
-	dberr_t		err;
-	buf_block_t*	undo_block = trx_undo_assign_low(trx, rseg, pundo,
-							 &err, &mtr);
 	trx_undo_t*	undo	= *pundo;
 	ut_ad((err == DB_SUCCESS) == (undo_block != NULL));
 	if (UNIV_UNLIKELY(undo_block == NULL)) {
diff --git a/storage/innobase/trx/trx0trx.cc b/storage/innobase/trx/trx0trx.cc
index 21a5347b0ec..9956abd65ed 100644
--- a/storage/innobase/trx/trx0trx.cc
+++ b/storage/innobase/trx/trx0trx.cc
@@ -1027,7 +1027,13 @@ trx_write_serialisation_history(
 		mtr_t	temp_mtr;
 		temp_mtr.start();
 		temp_mtr.set_log_mode(MTR_LOG_NO_REDO);
-		trx_undo_set_state_at_finish(undo, &temp_mtr);
+		buf_block_t* block= buf_page_get(page_id_t(SRV_TMP_SPACE_ID,
+							   undo->hdr_page_no),
+						 0, RW_X_LATCH, mtr);
+		ut_a(block);
+		temp_mtr.write<2>(*block, TRX_UNDO_SEG_HDR + TRX_UNDO_STATE
+				  + block->page.frame, TRX_UNDO_TO_PURGE);
+		undo->state = TRX_UNDO_TO_PURGE;
 		temp_mtr.commit();
 	}
 
diff --git a/storage/innobase/trx/trx0undo.cc b/storage/innobase/trx/trx0undo.cc
index bccb6538e63..20434d9fb9c 100644
--- a/storage/innobase/trx/trx0undo.cc
+++ b/storage/innobase/trx/trx0undo.cc
@@ -1290,27 +1290,25 @@ trx_undo_create(trx_t* trx, trx_rseg_t* rseg, trx_undo_t** undo,
 @param[in,out]	rseg	rollback segment
 @param[out]	pundo	the undo log memory object
 @param[in,out]	mtr	mini-transaction
+@param[out]	err	error code
 @return	the undo log block
 @retval	NULL	if none cached */
 static
 buf_block_t*
 trx_undo_reuse_cached(trx_t* trx, trx_rseg_t* rseg, trx_undo_t** pundo,
-		      mtr_t* mtr)
+		      mtr_t* mtr, dberr_t *err)
 {
-	if (rseg->is_persistent()) {
-		ut_ad(rseg->is_referenced());
-		if (rseg->needs_purge <= trx->id) {
-			/* trx_purge_truncate_history() compares
-			rseg->needs_purge <= head.trx_no
-			so we need to compensate for that.
-			The rseg->needs_purge after crash
-			recovery would be at least trx->id + 1,
-			because that is the minimum possible value
-			assigned by trx_serialise() on commit. */
-			rseg->needs_purge = trx->id + 1;
-		}
-	} else {
-		ut_ad(!rseg->is_referenced());
+	ut_ad(rseg->is_persistent());
+	ut_ad(rseg->is_referenced());
+	if (rseg->needs_purge <= trx->id) {
+		/* trx_purge_truncate_history() compares
+		rseg->needs_purge <= head.trx_no
+		so we need to compensate for that.
+		The rseg->needs_purge after crash
+		recovery would be at least trx->id + 1,
+		because that is the minimum possible value
+		assigned by trx_serialise() on commit. */
+		rseg->needs_purge = trx->id + 1;
 	}
 
 	trx_undo_t* undo = UT_LIST_GET_FIRST(rseg->undo_cached);
@@ -1321,9 +1319,10 @@ trx_undo_reuse_cached(trx_t* trx, trx_rseg_t* rseg, trx_undo_t** pundo,
 	ut_ad(undo->size == 1);
 	ut_ad(undo->id < TRX_RSEG_N_SLOTS);
 
-	buf_block_t*	block = buf_page_get(page_id_t(undo->rseg->space->id,
-						       undo->hdr_page_no),
-					     0, RW_X_LATCH, mtr);
+	buf_block_t* block = buf_page_get_gen(page_id_t(undo->rseg->space->id,
+							undo->hdr_page_no),
+					      0, RW_X_LATCH, nullptr, BUF_GET,
+					      mtr, err);
 	if (!block) {
 		return NULL;
 	}
@@ -1374,11 +1373,12 @@ trx_undo_assign(trx_t* trx, dberr_t* err, mtr_t* mtr)
 			BUF_GET, mtr, err);
 	}
 
+	*err = DB_SUCCESS;
 	trx_rseg_t* rseg = trx->rsegs.m_redo.rseg;
 
 	rseg->latch.wr_lock(SRW_LOCK_CALL);
 	buf_block_t* block = trx_undo_reuse_cached(
-		trx, rseg, &trx->rsegs.m_redo.undo, mtr);
+		trx, rseg, &trx->rsegs.m_redo.undo, mtr, err);
 
 	if (!block) {
 		block = trx_undo_create(trx, rseg, &trx->rsegs.m_redo.undo,
@@ -1387,8 +1387,6 @@ trx_undo_assign(trx_t* trx, dberr_t* err, mtr_t* mtr)
 		if (!block) {
 			goto func_exit;
 		}
-	} else {
-		*err = DB_SUCCESS;
 	}
 
 	UT_LIST_ADD_FIRST(rseg->undo_list, trx->rsegs.m_redo.undo);
@@ -1400,18 +1398,20 @@ func_exit:
 
 /** Assign an undo log for a transaction.
 A new undo log is created or a cached undo log reused.
+@tparam is_temp  whether this is temporary undo log
 @param[in,out]	trx	transaction
 @param[in]	rseg	rollback segment
 @param[out]	undo	the undo log
-@param[out]	err	error code
 @param[in,out]	mtr	mini-transaction
+@param[out]	err	error code
 @return	the undo log block
-@retval	NULL	on error */
+@retval	nullptr	on error */
+template<bool is_temp>
 buf_block_t*
-trx_undo_assign_low(trx_t* trx, trx_rseg_t* rseg, trx_undo_t** undo,
-		    dberr_t* err, mtr_t* mtr)
+trx_undo_assign_low(trx_t *trx, trx_rseg_t *rseg, trx_undo_t **undo,
+                    mtr_t *mtr, dberr_t *err)
 {
-	ut_d(const bool is_temp = rseg == trx->rsegs.m_noredo.rseg);
+	ut_ad(is_temp == (rseg == trx->rsegs.m_noredo.rseg));
 	ut_ad(is_temp || rseg == trx->rsegs.m_redo.rseg);
 	ut_ad(undo == (is_temp
 		       ? &trx->rsegs.m_noredo.undo
@@ -1431,19 +1431,24 @@ trx_undo_assign_low(trx_t* trx, trx_rseg_t* rseg, trx_undo_t** undo,
 		*err = DB_TOO_MANY_CONCURRENT_TRXS; return NULL;
 	);
 
+	*err = DB_SUCCESS;
 	rseg->latch.wr_lock(SRW_LOCK_CALL);
-	buf_block_t* block = trx_undo_reuse_cached(trx, rseg, undo, mtr);
-
-	if (!block) {
-		block = trx_undo_create(trx, rseg, undo, err, mtr);
-		ut_ad(!block == (*err != DB_SUCCESS));
-		if (!block) {
-			goto func_exit;
-		}
+	buf_block_t* block;
+	if (is_temp) {
+		ut_ad(!UT_LIST_GET_LEN(rseg->undo_cached));
 	} else {
-		*err = DB_SUCCESS;
+		block = trx_undo_reuse_cached(trx, rseg, undo, mtr, err);
+		if (block) {
+			goto got_block;
+		}
+	}
+	block = trx_undo_create(trx, rseg, undo, err, mtr);
+	ut_ad(!block == (*err != DB_SUCCESS));
+	if (!block) {
+		goto func_exit;
 	}
 
+got_block:
 	UT_LIST_ADD_FIRST(rseg->undo_list, *undo);
 
 func_exit:
@@ -1451,6 +1456,13 @@ func_exit:
 	return block;
 }
 
+template buf_block_t*
+trx_undo_assign_low<false>(trx_t *trx, trx_rseg_t *rseg, trx_undo_t **undo,
+                           mtr_t *mtr, dberr_t *err);
+template buf_block_t*
+trx_undo_assign_low<true>(trx_t *trx, trx_rseg_t *rseg, trx_undo_t **undo,
+                          mtr_t *mtr, dberr_t *err);
+
 /******************************************************************//**
 Sets the state of the undo log segment at a transaction finish.
 @return undo log segment header page, x-latched */
@@ -1461,6 +1473,7 @@ trx_undo_set_state_at_finish(
 	mtr_t*		mtr)	/*!< in: mtr */
 {
   ut_ad(undo->id < TRX_RSEG_N_SLOTS);
+  ut_ad(undo->rseg->is_persistent());
 
   buf_block_t *block=
     buf_page_get(page_id_t(undo->rseg->space->id, undo->hdr_page_no), 0,
@@ -1532,28 +1545,19 @@ the data can be discarded.
 @param undo     temporary undo log */
 void trx_undo_commit_cleanup(trx_undo_t *undo)
 {
-	trx_rseg_t*	rseg	= undo->rseg;
-	ut_ad(rseg->space == fil_system.temp_space);
-
-	rseg->latch.wr_lock(SRW_LOCK_CALL);
-
-	UT_LIST_REMOVE(rseg->undo_list, undo);
-
-	if (undo->state == TRX_UNDO_CACHED) {
-		UT_LIST_ADD_FIRST(rseg->undo_cached, undo);
-		undo = nullptr;
-	} else {
-		ut_ad(undo->state == TRX_UNDO_TO_PURGE);
+  trx_rseg_t *rseg= undo->rseg;
+  ut_ad(rseg->space == fil_system.temp_space);
+  rseg->latch.wr_lock(SRW_LOCK_CALL);
 
-		/* Delete first the undo log segment in the file */
-		trx_undo_seg_free(undo);
+  UT_LIST_REMOVE(rseg->undo_list, undo);
+  ut_ad(undo->state == TRX_UNDO_TO_PURGE);
+  /* Delete first the undo log segment in the file */
+  trx_undo_seg_free(undo);
+  ut_ad(rseg->curr_size > undo->size);
+  rseg->curr_size-= undo->size;
 
-		ut_ad(rseg->curr_size > undo->size);
-		rseg->curr_size -= undo->size;
-	}
-
-	rseg->latch.wr_unlock();
-	ut_free(undo);
+  rseg->latch.wr_unlock();
+  ut_free(undo);
 }
 
 /** At shutdown, frees the undo logs of a transaction. */