MDEV-23719: Make lock_sys use page_id_t

Since commit 8ccb3ca it should be
more efficient to use page_id_t rather than two separate variables
for tablespace identifier and page number.

lock_rec_fold(): Replaced with page_id_t::fold().

lock_rec_hash(): Replaced with lock_sys.hash(page_id).

lock_rec_expl_exist_on_page(), lock_rec_get_first_on_page_addr(),
lock_rec_get_first_on_page(): Replaced with lock_sys.get_first().

storage/innobase/btr/btr0cur.cc

@@ -5423,8 +5423,7 @@ btr_cur_compress_if_useful(
 		const buf_block_t* block = btr_cur_get_block(cursor);
 
 		/* Check whether page lock prevents the compression */
-		if (!lock_test_prdt_page_lock(trx, block->page.id().space(),
-					      block->page.id().page_no())) {
+		if (!lock_test_prdt_page_lock(trx, block->page.id())) {
 			return(false);
 		}
 	}

storage/innobase/gis/gis0rtree.cc

@@ -652,8 +652,7 @@ rtr_adjust_upper_level(
 	new_prdt.op = 0;
 
 	lock_prdt_update_parent(block, new_block, &prdt, &new_prdt,
-				index->table->space_id,
-				page_cursor->block->page.id().page_no());
+				page_cursor->block->page.id());
 
 	mem_heap_free(heap);
 
@@ -881,8 +880,6 @@ rtr_page_split_and_insert(
 	buf_block_t*		block;
 	page_t*			page;
 	page_t*			new_page;
-	ulint			page_no;
-	ulint			hint_page_no;
 	buf_block_t*		new_block;
 	page_zip_des_t*		page_zip;
 	page_zip_des_t*		new_page_zip;
@@ -931,7 +928,7 @@ rtr_page_split_and_insert(
 	ut_ad(mtr->memo_contains_flagged(block, MTR_MEMO_PAGE_X_FIX));
 	ut_ad(page_get_n_recs(page) >= 1);
 
-	page_no = block->page.id().page_no();
+	const page_id_t page_id(block->page.id());
 
 	if (!page_has_prev(page) && !page_is_leaf(page)) {
 		first_rec = page_rec_get_next(
@@ -969,10 +966,9 @@ rtr_page_split_and_insert(
 					   static_cast<uchar*>(first_rec));
 
 	/* Allocate a new page to the index */
-	hint_page_no = page_no + 1;
 	const uint16_t page_level = btr_page_get_level(page);
-	new_block = btr_page_alloc(cursor->index, hint_page_no, FSP_UP,
-				   page_level, mtr, mtr);
+	new_block = btr_page_alloc(cursor->index, page_id.page_no() + 1,
+				   FSP_UP, page_level, mtr, mtr);
 	if (!new_block) {
 		return NULL;
 	}
@@ -1155,8 +1151,7 @@ rtr_page_split_and_insert(
 
 	/* Check any predicate locks need to be moved/copied to the
 	new page */
-	lock_prdt_update_split(new_block, &prdt, &new_prdt,
-			       cursor->index->table->space_id, page_no);
+	lock_prdt_update_split(new_block, &prdt, &new_prdt, page_id);
 
 	/* Adjust the upper level. */
 	rtr_adjust_upper_level(cursor, flags, block, new_block,

storage/innobase/gis/gis0sea.cc

@@ -298,8 +298,9 @@ rtr_pcur_getnext_from_path(
 			    && mode != PAGE_CUR_RTREE_LOCATE) {
 				ut_ad(rtr_info->thr);
 				lock_place_prdt_page_lock(
-					index->table->space_id,
-					next_page_no, index,
+					page_id_t(block->page.id().space(),
+						  next_page_no),
+					index,
 					rtr_info->thr);
 			}
 			new_split = true;
@@ -1689,7 +1690,6 @@ rtr_cur_search_with_match(
 	const rec_t*	best_rec;
 	const rec_t*	last_match_rec = NULL;
 	bool		match_init = false;
-	ulint		space = block->page.id().space();
 	page_cur_mode_t	orig_mode = mode;
 	const rec_t*	first_rec = NULL;
 
@@ -1870,7 +1870,11 @@ rtr_cur_search_with_match(
 						/* Lock the page, preventing it
 						from being shrunk */
 						lock_place_prdt_page_lock(
-							space, page_no, index,
+							page_id_t(block->page
+								  .id()
+								  .space(),
+								  page_no),
+							index,
 							rtr_info->thr);
 					}
 				} else {

storage/innobase/handler/i_s.cc

@@ -699,10 +699,10 @@ fill_innodb_locks_from_cache(
 			OK(field_store_string(fields[IDX_LOCK_INDEX],
 					      row->lock_index));
 			OK(fields[IDX_LOCK_SPACE]->store(
-				   row->lock_space, true));
+				   row->lock_page.space(), true));
 			fields[IDX_LOCK_SPACE]->set_notnull();
 			OK(fields[IDX_LOCK_PAGE]->store(
-				   row->lock_page, true));
+				   row->lock_page.page_no(), true));
 			fields[IDX_LOCK_PAGE]->set_notnull();
 			OK(fields[IDX_LOCK_REC]->store(
 				   row->lock_rec, true));

storage/innobase/ibuf/ibuf0ibuf.cc

@@ -3329,18 +3329,22 @@ ibuf_insert_low(
 	and done mtr_commit(&mtr) to release the latch. */
 
 	ibuf_mtr_start(&bitmap_mtr);
-	index->set_modified(bitmap_mtr);
 
 	bitmap_page = ibuf_bitmap_get_map_page(page_id, zip_size, &bitmap_mtr);
 
 	/* We check if the index page is suitable for buffered entries */
 
-	if (buf_pool.page_hash_contains(page_id)
-	    || lock_rec_expl_exist_on_page(page_id.space(),
-					   page_id.page_no())) {
-
+	if (buf_pool.page_hash_contains(page_id)) {
+commit_exit:
 		ibuf_mtr_commit(&bitmap_mtr);
 		goto fail_exit;
+	} else {
+		lock_mutex_enter();
+		const auto lock_exists = lock_sys.get_first(page_id);
+		lock_mutex_exit();
+		if (lock_exists) {
+			goto commit_exit;
+		}
 	}
 
 	if (op == IBUF_OP_INSERT) {
@@ -3378,8 +3382,7 @@ ibuf_insert_low(
 		dfield_t*	field;
 
 		if (counter == ULINT_UNDEFINED) {
-			ibuf_mtr_commit(&bitmap_mtr);
-			goto fail_exit;
+			goto commit_exit;
 		}
 
 		field = dtuple_get_nth_field(
@@ -3391,6 +3394,7 @@ ibuf_insert_low(
 
 	/* Set the bitmap bit denoting that the insert buffer contains
 	buffered entries for this index page, if the bit is not set yet */
+	index->set_modified(bitmap_mtr);
 	ibuf_bitmap_page_set_bits<IBUF_BITMAP_BUFFERED>(
 		bitmap_page, page_id, physical_size, true, &bitmap_mtr);
 	ibuf_mtr_commit(&bitmap_mtr);

storage/innobase/include/lock0lock.h

@@ -252,15 +252,6 @@ lock_rec_restore_from_page_infimum(
 					state; lock bits are reset on
 					the infimum */
 /*********************************************************************//**
-Determines if there are explicit record locks on a page.
-@return an explicit record lock on the page, or NULL if there are none */
-lock_t*
-lock_rec_expl_exist_on_page(
-/*========================*/
-	ulint	space,	/*!< in: space id */
-	ulint	page_no)/*!< in: page number */
-	MY_ATTRIBUTE((warn_unused_result));
-/*********************************************************************//**
 Checks if locks of other transactions prevent an immediate insert of
 a record. If they do, first tests if the query thread should anyway
 be suspended for some reason; if not, then puts the transaction and
@@ -487,28 +478,6 @@ lock_rec_unlock(
 and release possible other transactions waiting because of these locks. */
 void lock_release(trx_t* trx);
 
-/*********************************************************************//**
-Calculates the fold value of a page file address: used in inserting or
-searching for a lock in the hash table.
-@return folded value */
-UNIV_INLINE
-ulint
-lock_rec_fold(
-/*==========*/
-	ulint	space,	/*!< in: space */
-	ulint	page_no)/*!< in: page number */
-	MY_ATTRIBUTE((const));
-/*********************************************************************//**
-Calculates the hash value of a page file address: used in inserting or
-searching for a lock in the hash table.
-@return hashed value */
-UNIV_INLINE
-unsigned
-lock_rec_hash(
-/*==========*/
-	ulint	space,	/*!< in: space */
-	ulint	page_no);/*!< in: page number */
-
 /*************************************************************//**
 Get the lock hash table */
 UNIV_INLINE
@@ -819,6 +788,46 @@ class lock_sys_t
 
   /** Closes the lock system at database shutdown. */
   void close();
+
+  /** @return the hash value for a page address */
+  ulint hash(const page_id_t id) const
+  { ut_ad(mutex_own(&mutex)); return rec_hash.calc_hash(id.fold()); }
+
+  /** Get the first lock on a page.
+  @param lock_hash   hash table to look at
+  @param id          page number
+  @return first lock
+  @retval nullptr if none exists */
+  lock_t *get_first(const hash_table_t &lock_hash, const page_id_t id) const
+  {
+    ut_ad(&lock_hash == &rec_hash || &lock_hash == &prdt_hash ||
+          &lock_hash == &prdt_page_hash);
+    for (lock_t *lock= static_cast<lock_t*>
+         (HASH_GET_FIRST(&lock_hash, hash(id)));
+         lock; lock= static_cast<lock_t*>(HASH_GET_NEXT(hash, lock)))
+      if (lock->un_member.rec_lock.page_id == id)
+         return lock;
+    return nullptr;
+  }
+
+  /** Get the first record lock on a page.
+  @param id          page number
+  @return first lock
+  @retval nullptr if none exists */
+  lock_t *get_first(const page_id_t id) const
+  { return get_first(rec_hash, id); }
+  /** Get the first predicate lock on a SPATIAL INDEX page.
+  @param id          page number
+  @return first lock
+  @retval nullptr if none exists */
+  lock_t *get_first_prdt(const page_id_t id) const
+  { return get_first(prdt_hash, id); }
+  /** Get the first predicate lock on a SPATIAL INDEX page.
+  @param id          page number
+  @return first lock
+  @retval nullptr if none exists */
+  lock_t *get_first_prdt_page(const page_id_t id) const
+  { return get_first(prdt_page_hash, id); }
 };
 
 /*********************************************************************//**
@@ -858,8 +867,7 @@ lock_rec_discard(
 without checking for deadlocks or conflicts.
 @param[in]	type_mode	lock mode and wait flag; type will be replaced
 				with LOCK_REC
-@param[in]	space		tablespace id
-@param[in]	page_no		index page number
+@param[in]	page_id		index page number
 @param[in]	page		R-tree index page, or NULL
 @param[in]	heap_no		record heap number in the index page
 @param[in]	index		the index tree
@@ -873,8 +881,7 @@ lock_rec_create_low(
 	que_thr_t*	thr,	/*!< thread owning trx */
 #endif
 	unsigned	type_mode,
-	ulint		space,
-	ulint		page_no,
+	const page_id_t	page_id,
 	const page_t*	page,
 	ulint		heap_no,
 	dict_index_t*	index,

storage/innobase/include/lock0lock.ic

@@ -1,7 +1,7 @@
 /*****************************************************************************
 
 Copyright (c) 1996, 2015, Oracle and/or its affiliates. All Rights Reserved.
-Copyright (c) 2017, 2018, MariaDB Corporation.
+Copyright (c) 2017, 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
@@ -28,34 +28,6 @@ Created 5/7/1996 Heikki Tuuri
 #include "buf0buf.h"
 #include "page0page.h"
 
-/*********************************************************************//**
-Calculates the fold value of a page file address: used in inserting or
-searching for a lock in the hash table.
-@return folded value */
-UNIV_INLINE
-ulint
-lock_rec_fold(
-/*==========*/
-	ulint	space,	/*!< in: space */
-	ulint	page_no)/*!< in: page number */
-{
-	return page_id_t(space, page_no).fold();
-}
-
-/*********************************************************************//**
-Calculates the hash value of a page file address: used in inserting or
-searching for a lock in the hash table.
-@return hashed value */
-UNIV_INLINE
-unsigned
-lock_rec_hash(
-/*==========*/
-	ulint	space,	/*!< in: space */
-	ulint	page_no)/*!< in: page number */
-{
-  return unsigned(lock_sys.rec_hash.calc_hash(lock_rec_fold(space, page_no)));
-}
-
 /*********************************************************************//**
 Gets the heap_no of the smallest user record on a page.
 @return heap_no of smallest user record, or PAGE_HEAP_NO_SUPREMUM */
@@ -122,12 +94,10 @@ lock_rec_create(
 					trx mutex */
 {
 	btr_assert_not_corrupted(block, index);
-	const page_id_t id(block->page.id());
 	return lock_rec_create_low(
 #ifdef WITH_WSREP
 		c_lock, thr,
 #endif
-		type_mode, id.space(), id.page_no(),
-		block->frame, heap_no,
+		type_mode, block->page.id(), block->frame, heap_no,
 		index, trx, caller_owns_trx_mutex);
 }

storage/innobase/include/lock0prdt.h

@@ -58,9 +58,7 @@ Acquire a "Page" lock on a block
 @return DB_SUCCESS, DB_LOCK_WAIT, or DB_DEADLOCK */
 dberr_t
 lock_place_prdt_page_lock(
-/*======================*/
-	ulint		space,	/*!< in: space for the page to lock */
-	ulint		pageno,	/*!< in: page number */
+	const page_id_t	page_id,	/*!< in: page identifier */
 	dict_index_t*	index,	/*!< in: secondary index */
 	que_thr_t*	thr);	/*!< in: query thread */
 
@@ -108,8 +106,7 @@ lock_prdt_update_split(
 	buf_block_t*	new_block,	/*!< in/out: the new half page */
 	lock_prdt_t*	prdt,		/*!< in: MBR on the old page */
 	lock_prdt_t*	new_prdt,	/*!< in: MBR on the new page */
-	ulint		space,		/*!< in: space id */
-	ulint		page_no);	/*!< in: page number */
+	const page_id_t	page_id);	/*!< in: page number */
 
 /**************************************************************//**
 Ajust locks from an ancester page of Rtree on the appropriate level . */
@@ -120,8 +117,7 @@ lock_prdt_update_parent(
 	buf_block_t*	right_block,	/*!< in/out: the new half page */
 	lock_prdt_t*	left_prdt,	/*!< in: MBR on the old page */
 	lock_prdt_t*	right_prdt,	/*!< in: MBR on the new page */
-	ulint		space,		/*!< in: space id */
-	ulint		page_no);	/*!< in: page number */
+	const page_id_t	page_id);	/*!< in: parent page */
 
 /*********************************************************************//**
 Checks if locks of other transactions prevent an immediate insert of
@@ -190,17 +186,11 @@ lock_prdt_rec_move(
 	const buf_block_t*	donator);	/*!< in: buffer block containing
 						the donating record */
 
-/** Check whether there are R-tree Page lock on a buffer page
+/** Check whether there are R-tree Page lock on a page
 @param[in]	trx	trx to test the lock
-@param[in]	space	space id for the page
-@param[in]	page_no	page number
-@return true if there is none */
-bool
-lock_test_prdt_page_lock(
-/*=====================*/
-	const trx_t*	trx,
-	ulint		space,
-	ulint		page_no);
+@param[in]	page_id	page identifier
+@return	true if there is none */
+bool lock_test_prdt_page_lock(const trx_t *trx, const page_id_t page_id);
 
 /** Removes predicate lock objects set on an index page which is discarded.
 @param[in]	block		page to be discarded