MDEV-27059 page_zip_dir_insert() may corrupt ROW_FORMAT=COMPRESSED tables

In commit 7ae21b18a6b73bbc3bf1ff448faf60c29ac1d386 (MDEV-12353)
the recovery of ROW_FORMAT=COMPRESSED tables was changed.
Changes would be logged in a physical format for the compressed
page image, so that the page need not be decompressed or compressed
during recovery.

page_zip_write_rec(): Log any update of the delete-mark flag in the
ROW_FORMAT=COMPRESSED page.

page_zip_dir_insert(): Copy the delete-mark flag. A delete-marked
record may be inserted by btr_cur_pessimistic_update() via
btr_cur_insert_if_possible(), page_cur_tuple_insert(),
page_cur_insert_rec_zip(). In the observed scenario, it was
an ROLLBACK. Presumably, the test case involved repeated DELETE
and INSERT of the same key, or updating a key back and forth.
This change alone might make the adjustment in page_zip_write_rec()
redundant, but we play it safe because we failed to create a
minimal test case for this scenario.

1 files changed, 15 insertions, 4 deletions

diff --git a/storage/innobase/page/page0zip.cc b/storage/innobase/page/page0zip.cc
index 331ecbfb0c4..86c3a4dff32 100644
--- a/storage/innobase/page/page0zip.cc
+++ b/storage/innobase/page/page0zip.cc
@@ -3676,6 +3676,7 @@ void page_zip_write_rec(buf_block_t *block, const byte *rec,
 
 	slot = page_zip_dir_find(page_zip, page_offset(rec));
 	ut_a(slot);
+	byte s = *slot;
 	/* Copy the delete mark. */
 	if (rec_get_deleted_flag(rec, TRUE)) {
 		/* In delete-marked records, DB_TRX_ID must
@@ -3683,9 +3684,14 @@ void page_zip_write_rec(buf_block_t *block, const byte *rec,
 		On non-leaf pages, the delete-mark flag is garbage. */
 		ut_ad(!index->is_primary() || !page_is_leaf(page)
 		      || row_get_rec_trx_id(rec, index, offsets));
-		*slot |= PAGE_ZIP_DIR_SLOT_DEL >> 8;
+		s |= PAGE_ZIP_DIR_SLOT_DEL >> 8;
 	} else {
-		*slot &= byte(~(PAGE_ZIP_DIR_SLOT_DEL >> 8));
+		s &= byte(~(PAGE_ZIP_DIR_SLOT_DEL >> 8));
+	}
+
+	if (s != *slot) {
+		*slot = s;
+		mtr->zmemcpy(*block, slot - page_zip->data, 1);
 	}
 
 	ut_ad(rec_get_start((rec_t*) rec, offsets) >= page + PAGE_ZIP_START);
@@ -4249,8 +4255,13 @@ page_zip_dir_insert(
 	}
 
 	/* Write the entry for the inserted record.
-	The "owned" and "deleted" flags must be zero. */
-	mach_write_to_2(slot_rec - PAGE_ZIP_DIR_SLOT_SIZE, page_offset(rec));
+	The "owned" flag must be zero. */
+	uint16_t offs = page_offset(rec);
+	if (rec_get_deleted_flag(rec, true)) {
+		offs |= PAGE_ZIP_DIR_SLOT_DEL;
+	}
+
+	mach_write_to_2(slot_rec - PAGE_ZIP_DIR_SLOT_SIZE, offs);
 	mtr->zmemcpy(*cursor->block, slot_rec - page_zip->data
 		     - PAGE_ZIP_DIR_SLOT_SIZE, PAGE_ZIP_DIR_SLOT_SIZE);
 }