MDEV-31158: Potential hang with ROW_FORMAT=COMPRESSED tables

btr_cur_need_opposite_intention(): Check also page_zip_available()
so that we will escalate to exclusive index latch when a non-leaf
page may have to be split further due to ROW_FORMAT=COMPRESSED page
overflow.

Tested by: Matthias Leich

mysql-test/suite/innodb_zip/r/page_size.result

@@ -608,4 +608,25 @@ SET GLOBAL innodb_compression_level=0;
 INSERT INTO t1 VALUES ('');
 SET GLOBAL innodb_compression_level= @save_innodb_compression_level;
 DROP TABLE t1;
+#
+# MDEV-31158 Assertion ...MTR_MEMO_X_LOCKED in btr_attach_half_pages()
+#
+SET @save_compression_level=@@GLOBAL.innodb_compression_level;
+SET GLOBAL innodb_compression_level=0;
+CREATE TEMPORARY TABLE t(a SERIAL, prefix VARBINARY(4), pad INT);
+INSERT INTO t(prefix, pad) VALUES
+(_binary 0xff,160),('',19),(_binary 0x0001,253),(_binary 0x0b11,169),
+(_binary 0x0b010001,23),(_binary 0x0b100001,251),(_binary 0x0d,163),
+(_binary 0xb3,254),(_binary 0x96,254),(_binary 0xeb,61),
+(_binary 0xf231,253),(_binary 0x1db0,253),(_binary 0x0005,101),
+(_binary 0x6370,253),(_binary 0x0b12,112),(_binary 0x0b010002,23),
+(_binary 0x0b100002,80),(_binary 0x181984,163),(_binary 0x181926,168),
+(_binary 0xe1,176),(_binary 0xe2,187),(_binary 0xe6,254),(_binary 0xbb,51),
+(_binary 0x1c,248),(_binary 0x8a,94),(_binary 0x14,254);
+CREATE TABLE u(a SMALLINT UNSIGNED PRIMARY KEY AUTO_INCREMENT,
+b VARBINARY(255), KEY(b)) ENGINE=InnoDB
+KEY_BLOCK_SIZE=1 ROW_FORMAT=COMPRESSED;
+INSERT INTO u SELECT a,CONCAT(prefix,REPEAT(chr(0),pad)) FROM t;
+DROP TABLE u, t;
+SET GLOBAL innodb_compression_level=@save_compression_level;
 # End of 10.6 tests

mysql-test/suite/innodb_zip/t/page_size.test

@@ -888,4 +888,28 @@ INSERT INTO t1 VALUES ('');
 SET GLOBAL innodb_compression_level= @save_innodb_compression_level;
 DROP TABLE t1;
 
+--echo #
+--echo # MDEV-31158 Assertion ...MTR_MEMO_X_LOCKED in btr_attach_half_pages()
+--echo #
+--source include/have_innodb.inc
+
+SET @save_compression_level=@@GLOBAL.innodb_compression_level;
+SET GLOBAL innodb_compression_level=0;
+CREATE TEMPORARY TABLE t(a SERIAL, prefix VARBINARY(4), pad INT);
+INSERT INTO t(prefix, pad) VALUES
+(_binary 0xff,160),('',19),(_binary 0x0001,253),(_binary 0x0b11,169),
+(_binary 0x0b010001,23),(_binary 0x0b100001,251),(_binary 0x0d,163),
+(_binary 0xb3,254),(_binary 0x96,254),(_binary 0xeb,61),
+(_binary 0xf231,253),(_binary 0x1db0,253),(_binary 0x0005,101),
+(_binary 0x6370,253),(_binary 0x0b12,112),(_binary 0x0b010002,23),
+(_binary 0x0b100002,80),(_binary 0x181984,163),(_binary 0x181926,168),
+(_binary 0xe1,176),(_binary 0xe2,187),(_binary 0xe6,254),(_binary 0xbb,51),
+(_binary 0x1c,248),(_binary 0x8a,94),(_binary 0x14,254);
+CREATE TABLE u(a SMALLINT UNSIGNED PRIMARY KEY AUTO_INCREMENT,
+               b VARBINARY(255), KEY(b)) ENGINE=InnoDB
+               KEY_BLOCK_SIZE=1 ROW_FORMAT=COMPRESSED;
+INSERT INTO u SELECT a,CONCAT(prefix,REPEAT(chr(0),pad)) FROM t;
+DROP TABLE u, t;
+SET GLOBAL innodb_compression_level=@save_compression_level;
+
 --echo # End of 10.6 tests

storage/innobase/btr/btr0cur.cc

@@ -748,18 +748,24 @@ btr_cur_will_modify_tree(
 
 /** Detects whether the modifying record might need a opposite modification
 to the intention.
-@param page		 page
+@param bpage             buffer pool page
+@param is_clust          whether this is a clustered index
 @param lock_intention    lock intention for the tree operation
 @param node_ptr_max_size the maximum size of a node pointer
 @param compress_limit    BTR_CUR_PAGE_COMPRESS_LIMIT(index)
 @param rec               record (current node_ptr)
 @return true if tree modification is needed */
-static bool btr_cur_need_opposite_intention(const page_t *page,
+static bool btr_cur_need_opposite_intention(const buf_page_t &bpage,
+                                            bool is_clust,
                                             btr_intention_t lock_intention,
                                             ulint node_ptr_max_size,
                                             ulint compress_limit,
                                             const rec_t *rec)
 {
+  if (UNIV_LIKELY_NULL(bpage.zip.data) &&
+      !page_zip_available(&bpage.zip, is_clust, node_ptr_max_size, 1))
+    return true;
+  const page_t *const page= bpage.frame;
   if (lock_intention != BTR_INTENTION_INSERT)
   {
     /* We compensate also for btr_cur_compress_recommendation() */
@@ -1342,7 +1348,8 @@ release_tree:
           !btr_block_get(*index(), btr_page_get_next(block->page.frame),
                          RW_X_LATCH, false, mtr, &err))
         goto func_exit;
-      if (btr_cur_need_opposite_intention(block->page.frame, lock_intention,
+      if (btr_cur_need_opposite_intention(block->page, index()->is_clust(),
+                                          lock_intention,
                                           node_ptr_max_size, compress_limit,
                                           page_cur.rec))
         goto need_opposite_intention;
@@ -1398,7 +1405,8 @@ release_tree:
   default:
     break;
   case BTR_MODIFY_TREE:
-    if (btr_cur_need_opposite_intention(block->page.frame, lock_intention,
+    if (btr_cur_need_opposite_intention(block->page, index()->is_clust(),
+                                        lock_intention,
                                         node_ptr_max_size, compress_limit,
                                         page_cur.rec))
       /* If the rec is the first or last in the page for pessimistic
@@ -1948,7 +1956,7 @@ index_locked:
             break;
 
           if (!index->lock.have_x() &&
-              btr_cur_need_opposite_intention(block->page.frame,
+              btr_cur_need_opposite_intention(block->page, index->is_clust(),
                                               lock_intention,
                                               node_ptr_max_size,
                                               compress_limit, page_cur.rec))
@@ -1995,7 +2003,8 @@ index_locked:
     ut_ad(latch_mode != BTR_MODIFY_TREE || upper_rw_latch == RW_X_LATCH);
 
     if (latch_mode != BTR_MODIFY_TREE);
-    else if (btr_cur_need_opposite_intention(block->page.frame, lock_intention,
+    else if (btr_cur_need_opposite_intention(block->page, index->is_clust(),
+                                             lock_intention,
                                              node_ptr_max_size, compress_limit,
                                              page_cur.rec))
     {