branches/zip: Replace the redo log entry types

MLOG_ZIP_COMPRESS and MLOG_ZIP_DECOMPRESS with higher-level entry types.
Implement the logging and crash recovery of MLOG_ZIP_PAGE_CREATE.

page_create_zip(): New function for creating a compressed B-tree page.

page_parse_create_zip(): New function for applying a MLOG_ZIP_PAGE_CREATE
redo log record.

btr_page_create(): Remove the prototype.  Add parameters page_zip, level,
prev, and next.

btr0btr.c: Eliminate page_zip_compress() calls where possible.

page_zip_alloc(), page_zip_compress(), page_zip_decompress(),
page_zip_clear_rec(): Remove parameter mtr.

recv_parse_or_apply_log_rec_body(): Handle MLOG_ZIP_PAGE_CREATE.
Add TODO comments for the other added redo log entry types.

btr/btr0btr.c

@@ -78,16 +78,6 @@ make them consecutive on disk if possible. From the other file segment
 we allocate pages for the non-leaf levels of the tree.
 */
 
-/******************************************************************
-Creates a new index page to the tree (not the root, and also not
-used in page reorganization). */
-static
-void
-btr_page_create(
-/*============*/
-	page_t*		page,	/* in: page to be created */
-	dict_tree_t*	tree,	/* in: index tree */
-	mtr_t*		mtr);	/* in: mtr */
 /****************************************************************
 Returns the upper level node pointer to a page. It is assumed that
 mtr holds an x-latch on the tree. */
@@ -255,17 +245,34 @@ static
 void
 btr_page_create(
 /*============*/
-	page_t*		page,	/* in: page to be created */
+	page_t*		page,	/* in/out: page to be created */
+	page_zip_des_t*	page_zip,/* in/out: compressed page, or NULL */
 	dict_tree_t*	tree,	/* in: index tree */
+	ulint		level,	/* in: the B-tree level of the page */
+	ulint		prev,	/* in: number of the previous page */
+	ulint		next,	/* in: number of the next page */
 	mtr_t*		mtr)	/* in: mtr */
 {
+	dict_index_t*	index = UT_LIST_GET_FIRST(tree->tree_indexes);
+
 	ut_ad(mtr_memo_contains(mtr, buf_block_align(page),
 							MTR_MEMO_PAGE_X_FIX));
-	page_create(page, NULL, mtr,
-			UT_LIST_GET_FIRST(tree->tree_indexes));
+
+	if (UNIV_LIKELY_NULL(page_zip)) {
+		page_create_zip(page, page_zip, index, level, mtr);
+	} else {
+		page_create(page, mtr, dict_table_is_comp(index->table));
+		/* Set the level of the new index page */
+		btr_page_set_level(page, NULL, level, mtr);
+	}
+
+	/* Set the next node and previous node fields of new page */
+	btr_page_set_next(page, page_zip, prev, mtr);
+	btr_page_set_prev(page, page_zip, next, mtr);
+
 	buf_block_align(page)->check_index_page_at_flush = TRUE;
 
-	btr_page_set_index_id(page, NULL, tree->id, mtr);
+	btr_page_set_index_id(page, page_zip, tree->id, mtr);
 }
 
 /******************************************************************
@@ -735,18 +742,25 @@ btr_create(
 	}
 
 	/* Create a new index page on the the allocated segment page */
-	page = page_create(frame, NULL, mtr, index);
-	buf_block_align(page)->check_index_page_at_flush = TRUE;
-
-	/* Set the index id of the page */
-	btr_page_set_index_id(page, NULL, index_id, mtr);
+	page_zip = buf_block_get_page_zip(buf_block_align(frame));
 
+	if (UNIV_LIKELY_NULL(page_zip)) {
+		page = page_create_zip(frame, page_zip, index, 0, mtr);
+	} else {
+		page = page_create(frame, mtr,
+				dict_table_is_comp(index->table));
 		/* Set the level of the new index page */
 		btr_page_set_level(page, NULL, 0, mtr);
+	}
+
+	buf_block_align(page)->check_index_page_at_flush = TRUE;
+
+	/* Set the index id of the page */
+	btr_page_set_index_id(page, page_zip, index_id, mtr);
 
 	/* Set the next node and previous node fields */
-	btr_page_set_next(page, NULL, FIL_NULL, mtr);
-	btr_page_set_prev(page, NULL, FIL_NULL, mtr);
+	btr_page_set_next(page, page_zip, FIL_NULL, mtr);
+	btr_page_set_prev(page, page_zip, FIL_NULL, mtr);
 
 	/* We reset the free bits for the page to allow creation of several
 	trees in the same mtr, otherwise the latch on a bitmap page would
@@ -760,15 +774,6 @@ btr_create(
 
 	ut_ad(page_get_max_insert_size(page, 2) > 2 * BTR_PAGE_MAX_REC_SIZE);
 
-	page_zip = buf_block_get_page_zip(buf_block_align(page));
-	if (UNIV_LIKELY_NULL(page_zip)) {
-		if (UNIV_UNLIKELY(!page_zip_compress(
-				page_zip, page, index, mtr))) {
-			/* An empty page should always be compressible */
-			ut_error;
-		}
-	}
-
 	return(page_no);
 }
 
@@ -894,7 +899,7 @@ btr_page_reorganize_low(
 	/* Recreate the page: note that global data on page (possible
 	segment headers, next page-field, etc.) is preserved intact */
 
-	page_create(page, NULL, mtr, index);
+	page_create(page, mtr, dict_table_is_comp(index->table));
 	buf_block_align(page)->check_index_page_at_flush = TRUE;
 
 	/* Copy the records from the temporary space to the recreated page;
@@ -907,7 +912,7 @@ btr_page_reorganize_low(
 
 	if (UNIV_LIKELY_NULL(page_zip)) {
 		if (UNIV_UNLIKELY(!page_zip_compress(
-				page_zip, page, index, mtr))) {
+				page_zip, page, index))) {
 
 			/* Restore the old page and exit. */
 			buf_frame_copy(page, new_page);
@@ -1012,7 +1017,13 @@ btr_page_empty(
 	/* Recreate the page: note that global data on page (possible
 	segment headers, next page-field, etc.) is preserved intact */
 
-	page_create(page, page_zip, mtr, index);
+	if (UNIV_LIKELY_NULL(page_zip)) {
+		page_create_zip(page, page_zip, index,
+				btr_page_get_level(page, mtr), mtr);
+	} else {
+		page_create(page, mtr, dict_table_is_comp(index->table));
+	}
+
 	buf_block_align(page)->check_index_page_at_flush = TRUE;
 }
 
@@ -1066,20 +1077,13 @@ btr_root_raise_and_insert(
 	level = btr_page_get_level(root, mtr);
 
 	new_page = btr_page_alloc(tree, 0, FSP_NO_DIR, level, mtr);
+	new_page_zip = buf_block_get_page_zip(buf_block_align(new_page));
 
-	btr_page_create(new_page, tree, mtr);
-
-	/* Set the level of the new index page */
-	btr_page_set_level(new_page, NULL, level, mtr);
-
-	/* Set the next node and previous node fields of new page */
-	btr_page_set_next(new_page, NULL, FIL_NULL, mtr);
-	btr_page_set_prev(new_page, NULL, FIL_NULL, mtr);
+	btr_page_create(new_page, new_page_zip, tree, level,
+				FIL_NULL, FIL_NULL, mtr);
 
 	/* Move the records from root to the new page */
 
-	new_page_zip = buf_block_get_page_zip(buf_block_align(new_page));
-
 	if (UNIV_UNLIKELY(!page_copy_rec_list_end(new_page, new_page_zip,
 			page_get_infimum_rec(root), cursor->index, mtr))) {
 		/* This should always succeed, as new_page
@@ -1106,10 +1110,31 @@ btr_root_raise_and_insert(
 
 	node_ptr = dict_tree_build_node_ptr(tree, rec, new_page_no, heap,
 								level);
+	/* The node pointer must be marked as the predefined minimum record,
+	as there is no lower alphabetical limit to records in the leftmost
+	node of a level: */
+	dtuple_set_info_bits(node_ptr, dtuple_get_info_bits(node_ptr)
+				| REC_INFO_MIN_REC_FLAG);
+
 	/* Rebuild the root page to get free space */
 	root_page_zip = buf_block_get_page_zip(buf_block_align(root));
+	if (UNIV_LIKELY_NULL(root_page_zip)) {
+		page_create_zip(root, root_page_zip, cursor->index,
+				level + 1, mtr);
+	} else {
+		page_create(root, mtr,
+				dict_table_is_comp(cursor->index->table));
 		btr_page_set_level(root, NULL, level + 1, mtr);
-	page_create(root, root_page_zip, mtr, cursor->index);
+	}
+
+	/* Set the next node and previous node fields, although
+	they should already have been set.  The previous node field
+	must be FIL_NULL if root_page_zip != NULL, because the
+	REC_INFO_MIN_REC_FLAG (of the first user record) will be
+	set if and only if btr_page_get_prev() == FIL_NULL. */
+	btr_page_set_next(root, root_page_zip, FIL_NULL, mtr);
+	btr_page_set_prev(root, root_page_zip, FIL_NULL, mtr);
+
 	buf_block_align(root)->check_index_page_at_flush = TRUE;
 
 	page_cursor = btr_cur_get_page_cur(cursor);
@@ -1118,25 +1143,12 @@ btr_root_raise_and_insert(
 
 	page_cur_set_before_first(root, page_cursor);
 
-	node_ptr_rec = page_cur_tuple_insert(page_cursor, NULL,
+	node_ptr_rec = page_cur_tuple_insert(page_cursor, root_page_zip,
 					node_ptr, cursor->index, NULL, 0, mtr);
 
-	ut_ad(node_ptr_rec);
-
-	/* The node pointer must be marked as the predefined minimum record,
-	as there is no lower alphabetical limit to records in the leftmost
-	node of a level: */
-
-	btr_set_min_rec_mark(node_ptr_rec, mtr);
-
-	if (UNIV_LIKELY_NULL(root_page_zip)
-		&& !UNIV_UNLIKELY(page_zip_compress(root_page_zip, root,
-							cursor->index, mtr))) {
-		/* The root page should only contain the
-		node pointer to new_page at this point.
-		Thus, the data should fit. */
-		ut_error;
-	}
+	/* The root page should only contain the node pointer
+	to new_page at this point.  Thus, the data should fit. */
+	ut_a(node_ptr_rec);
 
 	/* Free the memory heap */
 	mem_heap_free(heap);
@@ -1658,6 +1670,7 @@ btr_page_split_and_insert(
 	byte		direction;
 	ulint		hint_page_no;
 	page_t*		new_page;
+	page_zip_des_t*	new_page_zip;
 	rec_t*		split_rec;
 	page_t*		left_page;
 	page_t*		right_page;
@@ -1722,7 +1735,9 @@ btr_page_split_and_insert(
 	/* 2. Allocate a new page to the tree */
 	new_page = btr_page_alloc(tree, hint_page_no, direction,
 					btr_page_get_level(page, mtr), mtr);
-	btr_page_create(new_page, tree, mtr);
+	new_page_zip = buf_block_get_page_zip(buf_block_align(new_page));
+	btr_page_create(new_page, new_page_zip, tree,
+			btr_page_get_level(page, mtr), 0, 0, mtr);
 
 	/* 3. Calculate the first record on the upper half-page, and the
 	first record (move_limit) on original page which ends up on the
@@ -1773,8 +1788,7 @@ btr_page_split_and_insert(
 /*		fputs("Split left\n", stderr); */
 
 		if (UNIV_UNLIKELY(!page_move_rec_list_start(
-					new_page, buf_block_get_page_zip(
-					buf_block_align(new_page)),
+					new_page, new_page_zip,
 					move_limit, page_zip,
 					cursor->index, mtr))) {
 			ut_error;
@@ -1788,8 +1802,7 @@ btr_page_split_and_insert(
 /*		fputs("Split right\n", stderr); */
 
 		if (UNIV_UNLIKELY(!page_move_rec_list_end(
-					new_page, buf_block_get_page_zip(
-					buf_block_align(new_page)),
+					new_page, new_page_zip,
 					move_limit, page_zip,
 					cursor->index, mtr))) {
 			ut_error;

btr/btr0cur.c

@@ -1576,7 +1576,7 @@ btr_cur_update_in_place(
 	page_zip = buf_block_get_page_zip(block);
 	if (UNIV_LIKELY_NULL(page_zip)
 			&& UNIV_UNLIKELY(!page_zip_alloc(page_zip,
-					buf_block_get_frame(block), index, mtr,
+					buf_block_get_frame(block), index,
 					rec_offs_size(offsets), 0))) {
 		return(DB_ZIP_OVERFLOW);
 	}
@@ -1745,7 +1745,7 @@ btr_cur_optimistic_update(
 	page_zip = buf_block_get_page_zip(buf_block_align(page));
 
 	if (UNIV_LIKELY_NULL(page_zip)
-			&& !page_zip_alloc(page_zip, page, index, mtr,
+			&& !page_zip_alloc(page_zip, page, index,
 					new_rec_size, 0)) {
 		mem_heap_free(heap);
 

include/mtr0mtr.h

@@ -135,13 +135,17 @@ flag value must give the length also! */
 #define MLOG_ZIP_WRITE_BLOB_PTR	((byte)48)	/* write the BLOB pointer
 						of an externally stored column
 						on a compressed page */
-#define MLOG_ZIP_COMPRESS	((byte)49)	/* compress a page */
-#define MLOG_ZIP_DECOMPRESS	((byte)50)	/* decompress a page
-						to undo a compressed page
-						overflow */
-#define MLOG_ZIP_WRITE_HEADER	((byte)51)	/* write to compressed page
+#define MLOG_ZIP_WRITE_HEADER	((byte)49)	/* write to compressed page
 						header */
-#define MLOG_BIGGEST_TYPE	((byte)51)	/* biggest value (used in
+#define MLOG_ZIP_PAGE_CREATE	((byte)50)	/* create a compressed
+						index page */
+#define MLOG_ZIP_LIST_START_COPY ((byte)51)	/* copy compact record list
+						start to a compressed page */
+#define MLOG_ZIP_LIST_END_COPY	((byte)52)	/* copy compact record list
+						end to a compressed page */
+#define MLOG_ZIP_ROOT_RAISE	((byte)53)	/* raise the root of a
+						compressed B-tree */
+#define MLOG_BIGGEST_TYPE	((byte)53)	/* biggest value (used in
 						asserts) */
 
 /*******************************************************************

include/page0page.h

@@ -607,17 +607,30 @@ page_mem_free(
 	dict_index_t*	index,	/* in: index of rec */
 	const ulint*	offsets);/* in: array returned by rec_get_offsets() */
 /**************************************************************
-The index page creation function. */
+Create an uncompressed B-tree index page. */
 
 page_t*
 page_create(
 /*========*/
 					/* out: pointer to the page */
-	buf_frame_t*	frame,		/* in: a buffer frame where the page is
-					created */
-	page_zip_des_t*	page_zip,	/* in/out: compressed page, or NULL */
+	buf_frame_t*	frame,		/* in/out: a buffer frame where the
+					page is created */
 	mtr_t*		mtr,		/* in: mini-transaction handle */
-	dict_index_t*	index);		/* in: the index of the page */
+	ulint		comp);		/* in: nonzero=compact page format */
+/**************************************************************
+Create a compressed B-tree index page. */
+
+page_t*
+page_create_zip(
+/*============*/
+					/* out: pointer to the page */
+	buf_frame_t*	frame,		/* in/out: a buffer frame where the
+					page is created */
+	page_zip_des_t*	page_zip,	/* in/out: compressed page, or NULL */
+	dict_index_t*	index,		/* in: the index of the page */
+	ulint		level,		/* in: the B-tree level of the page */
+	mtr_t*		mtr);		/* in: mini-transaction handle */
+
 /*****************************************************************
 Differs from page_copy_rec_list_end, because this function does not
 touch the lock table and max trx id on page or compress the page. */
@@ -767,6 +780,19 @@ page_parse_create(
 	ulint		comp,	/* in: nonzero=compact page format */
 	page_t*		page,	/* in: page or NULL */
 	mtr_t*		mtr);	/* in: mtr or NULL */
+/***************************************************************
+Parses a redo log record of creating a compressed page. */
+
+byte*
+page_parse_create_zip(
+/*==================*/
+				/* out: end of log record or NULL */
+	byte*		ptr,	/* in: buffer */
+	byte*		end_ptr,/* in: buffer end */
+	page_t*		page,	/* in/out: page or NULL */
+	page_zip_des_t*	page_zip,/* in/out: compressed page or NULL */
+	dict_index_t*	index,	/* in: index of the page */
+	mtr_t*		mtr);	/* in: mtr or NULL */
 /****************************************************************
 Prints record contents including the data relevant only in
 the index page context. */

include/page0zip.h

@@ -39,10 +39,8 @@ page_zip_compress(
 	page_zip_des_t*	page_zip,/* in: size; out: data, n_blobs,
 				m_start, m_end */
 	const page_t*	page,	/* in: uncompressed page */
-	dict_index_t*	index,	/* in: index of the B-tree node */
-	mtr_t*		mtr)	/* in: mini-transaction handle,
-				or NULL if no logging is needed */
-	__attribute__((nonnull(1,2,3)));
+	dict_index_t*	index)	/* in: index of the B-tree node */
+	__attribute__((warn_unused_result, nonnull));
 
 /**************************************************************************
 Decompress a page.  This function should tolerate errors on the compressed
@@ -55,10 +53,8 @@ page_zip_decompress(
 				/* out: TRUE on success, FALSE on failure */
 	page_zip_des_t*	page_zip,/* in: data, size;
 				out: m_start, m_end, n_blobs */
-	page_t*		page,	/* out: uncompressed page, may be trashed */
-	mtr_t*		mtr)	/* in: mini-transaction handle,
-				or NULL if no logging is needed */
-	__attribute__((warn_unused_result, nonnull(1, 2)));
+	page_t*		page)	/* out: uncompressed page, may be trashed */
+	__attribute__((warn_unused_result, nonnull));
 
 #ifdef UNIV_DEBUG
 /**************************************************************************
@@ -97,11 +93,9 @@ page_zip_alloc(
 				and successful */
 	const page_t*	page,	/* in: uncompressed page */
 	dict_index_t*	index,	/* in: index of the B-tree node */
-	mtr_t*		mtr,	/* in: mini-transaction handle,
-				or NULL if no logging is desired */
 	ulint		length,	/* in: combined size of the record */
 	ulint		create)	/* in: nonzero=add the record to the heap */
-	__attribute__((warn_unused_result, nonnull(1,2,3)));
+	__attribute__((warn_unused_result, nonnull));
 
 /**************************************************************************
 Write an entire record on the compressed page.  The data must already

include/page0zip.ic

@@ -155,8 +155,6 @@ page_zip_alloc(
 				and successful */
 	const page_t*	page,	/* in: uncompressed page */
 	dict_index_t*	index,	/* in: index of the B-tree node */
-	mtr_t*		mtr,	/* in: mini-transaction handle,
-				or NULL if no logging is desired */
 	ulint		length,	/* in: combined size of the record */
 	ulint		create)	/* in: nonzero=add the record to the heap */
 {
@@ -173,7 +171,7 @@ page_zip_alloc(
 		return(FALSE);
 	}
 
-	if (!page_zip_compress(page_zip, page, index, mtr)) {
+	if (!page_zip_compress(page_zip, page, index)) {
 		/* Unable to compress the page */
 		return(FALSE);
 	}

log/log0recv.c

@@ -905,30 +905,17 @@ recv_parse_or_apply_log_rec_body(
 		ptr = page_zip_parse_write_header(
 				ptr, end_ptr, page, page_zip);
 		break;
-	case MLOG_ZIP_COMPRESS:
+	case MLOG_ZIP_PAGE_CREATE:
 		if (NULL != (ptr = mlog_parse_index(
-				ptr, end_ptr, TRUE, &index))
-				&& page) {
-			ut_a(page_is_comp(page));
-			ut_a(page_zip);
-			if (UNIV_UNLIKELY(!page_zip_compress(
-					page_zip, page, index, NULL))) {
-				ut_error;
-			}
+				ptr, end_ptr, TRUE, &index))) {
+			ptr = page_parse_create_zip(ptr, end_ptr,
+					page, page_zip, index, mtr);
 		}
 		break;
-	case MLOG_ZIP_DECOMPRESS:
-		/* TODO: remove this? */
-		if (NULL != (ptr = mlog_parse_index(
-				ptr, end_ptr, TRUE, &index))
-				&& page) {
-			ut_a(page_is_comp(page));
-			ut_a(page_zip);
-			if (UNIV_UNLIKELY(!page_zip_decompress(
-						page_zip, page, NULL))) {
-				ut_error;
-			}
-		}
+	case MLOG_ZIP_LIST_START_COPY:
+	case MLOG_ZIP_LIST_END_COPY:
+	case MLOG_ZIP_ROOT_RAISE:
+		ut_error; /* TODO */
 		break;
 	default:
 		ptr = NULL;

page/page0cur.c

@@ -923,7 +923,7 @@ page_cur_insert_rec_low(
 
 	/* 2. Try to find suitable space from page memory management */
 	if (UNIV_LIKELY_NULL(page_zip)
-	    && !page_zip_alloc(page_zip, page, index, mtr, rec_size, 1)) {
+	    && !page_zip_alloc(page_zip, page, index, rec_size, 1)) {
 
 		return(NULL);
 	}

page/page0page.c

@@ -309,15 +309,7 @@ page_parse_create(
 	/* The record is empty, except for the record initial part */
 
 	if (page) {
-		dict_index_t*	index;
-
-		if (UNIV_LIKELY(comp)) {
-			index = srv_sys->dummy_ind2;
-		} else {
-			index = srv_sys->dummy_ind1;
-		}
-
-		page_create(page, NULL, mtr, index);
+		page_create(page, mtr, comp);
 	}
 
 	return(ptr);
@@ -325,16 +317,14 @@ page_parse_create(
 
 /**************************************************************
 The index page creation function. */
-
+static
 page_t*
-page_create(
-/*========*/
+page_create_low(
+/*============*/
 					/* out: pointer to the page */
 	buf_frame_t*	frame,		/* in/out: a buffer frame where the
 					page is created */
-	page_zip_des_t*	page_zip,	/* in/out: compressed page, or NULL */
-	mtr_t*		mtr,		/* in: mini-transaction handle */
-	dict_index_t*	index)		/* in: the index of the page */
+	ulint		comp)		/* in: nonzero=compact page format */
 {
 	page_dir_slot_t* slot;
 	mem_heap_t*	heap;
@@ -344,11 +334,10 @@ page_create(
 	rec_t*		infimum_rec;
 	rec_t*		supremum_rec;
 	page_t*		page;
+	dict_index_t*	index;
 	ulint*		offsets;
-	const ibool	comp = dict_table_is_comp(index->table);
 
-	ut_ad(!page_zip || comp);
-	ut_ad(frame && mtr);
+	ut_ad(frame);
 #if PAGE_BTR_IBUF_FREE_LIST + FLST_BASE_NODE_SIZE > PAGE_DATA
 # error "PAGE_BTR_IBUF_FREE_LIST + FLST_BASE_NODE_SIZE > PAGE_DATA"
 #endif
@@ -366,9 +355,6 @@ page_create(
 	/* 1. INCREMENT MODIFY CLOCK */
 	buf_frame_modify_clock_inc(frame);
 
-	/* 2. WRITE LOG INFORMATION */
-	page_create_write_log(frame, mtr, comp);
-
 	page = frame;
 
 	fil_page_set_type(page, FIL_PAGE_INDEX);
@@ -477,17 +463,97 @@ page_create(
 		rec_set_next_offs_old(supremum_rec, 0);
 	}
 
-	if (UNIV_LIKELY_NULL(page_zip)) {
-		ut_ad(comp);
+	return(page);
+}
+
+/**************************************************************
+Create an uncompressed B-tree index page. */
+
+page_t*
+page_create(
+/*========*/
+					/* out: pointer to the page */
+	buf_frame_t*	frame,		/* in/out: a buffer frame where the
+					page is created */
+	mtr_t*		mtr,		/* in: mini-transaction handle */
+	ulint		comp)		/* in: nonzero=compact page format */
+{
+	page_create_write_log(frame, mtr, comp);
+	return(page_create_low(frame, comp));
+}
+
+/***************************************************************
+Parses a redo log record of creating a compressed page. */
+
+byte*
+page_parse_create_zip(
+/*==================*/
+				/* out: end of log record or NULL */
+	byte*		ptr,	/* in: buffer */
+	byte*		end_ptr,/* in: buffer end */
+	page_t*		page,	/* in/out: page or NULL */
+	page_zip_des_t*	page_zip,/* in/out: compressed page or NULL */
+	dict_index_t*	index,	/* in: index of the page */
+	mtr_t*		mtr)	/* in: mtr or NULL */
+{
+	ulint	level;
+
+	ut_ad(ptr && end_ptr && index);
+
+	if (UNIV_UNLIKELY(ptr + 2 > end_ptr)) {
+
+		return(NULL);
+	}
+
+	level = mach_read_from_2(ptr);
+	ptr += 2;
+
+	if (page) {
+
+		page_create_zip(page, page_zip, index, level, mtr);
+	}
+
+	return(ptr);
+}
+
+/**************************************************************
+Create a compressed B-tree index page. */
 
-		if (!page_zip_compress(page_zip, page, index, mtr)) {
+page_t*
+page_create_zip(
+/*============*/
+					/* out: pointer to the page */
+	buf_frame_t*	frame,		/* in/out: a buffer frame where the
+					page is created */
+	page_zip_des_t*	page_zip,	/* in/out: compressed page, or NULL */
+	dict_index_t*	index,		/* in: the index of the page */
+	ulint		level,		/* in: the B-tree level of the page */
+	mtr_t*		mtr)		/* in: mini-transaction handle */
+{
+	byte*	log_ptr;
+
+	ut_ad(frame && page_zip && index);
+	ut_ad(dict_table_is_comp(index->table));
+
+	log_ptr = mlog_open(mtr, 11 + 2);
+	if (log_ptr) {
+		log_ptr = mlog_write_initial_log_record_fast(frame,
+				MLOG_ZIP_PAGE_CREATE, log_ptr, mtr);
+		mach_write_to_2(log_ptr, level);
+		log_ptr += 2;
+		mlog_close(mtr, log_ptr);
+	}
+
+	mach_write_to_2(frame + PAGE_HEADER + PAGE_LEVEL, level);
+	page_create_low(frame, TRUE);
+
+	if (UNIV_UNLIKELY(!page_zip_compress(page_zip, frame, index))) {
 		/* The compression of a newly created page
 		should always succeed. */
 		ut_error;
 	}
-	}
 
-	return(page);
+	return(frame);
 }
 
 /*****************************************************************
@@ -591,10 +657,10 @@ page_copy_rec_list_end(
 
 	if (UNIV_LIKELY_NULL(new_page_zip)) {
 		if (UNIV_UNLIKELY(!page_zip_compress(new_page_zip,
-					new_page, index, mtr))) {
+					new_page, index))) {
 
 			if (UNIV_UNLIKELY(!page_zip_decompress(
-					new_page_zip, new_page, mtr))) {
+					new_page_zip, new_page))) {
 				ut_error;
 			}
 			return(FALSE);
@@ -674,10 +740,10 @@ page_copy_rec_list_start(
 
 	if (UNIV_LIKELY_NULL(new_page_zip)) {
 		if (UNIV_UNLIKELY(!page_zip_compress(new_page_zip,
-					new_page, index, mtr))) {
+					new_page, index))) {
 
 			if (UNIV_UNLIKELY(!page_zip_decompress(
-					new_page_zip, new_page, mtr))) {
+					new_page_zip, new_page))) {
 				ut_error;
 			}
 			/* TODO: try btr_page_reorganize() */

page/page0zip.c

@@ -521,9 +521,7 @@ page_zip_compress(
 	page_zip_des_t*	page_zip,/* in: size; out: data, n_blobs,
 				m_start, m_end */
 	const page_t*	page,	/* in: uncompressed page */
-	dict_index_t*	index,	/* in: index of the B-tree node */
-	mtr_t*		mtr)	/* in: mini-transaction handle,
-				or NULL if no logging is needed */
+	dict_index_t*	index)	/* in: index of the B-tree node */
 {
 	z_stream	c_stream;
 	int		err;
@@ -881,11 +879,6 @@ page_zip_compress(
 	ut_a(page_zip_validate(page_zip, page));
 #endif /* UNIV_DEBUG || UNIV_ZIP_DEBUG */
 
-	if (UNIV_LIKELY_NULL(mtr)) {
-		mlog_open_and_write_index(mtr, (page_t*) page, index,
-					MLOG_ZIP_COMPRESS, 0);
-	}
-
 	return(TRUE);
 }
 
@@ -1394,9 +1387,7 @@ page_zip_decompress(
 				/* out: TRUE on success, FALSE on failure */
 	page_zip_des_t*	page_zip,/* in: data, size;
 				out: m_start, m_end, n_blobs */
-	page_t*		page,	/* out: uncompressed page, may be trashed */
-	mtr_t*		mtr)	/* in: mini-transaction handle,
-				or NULL if no logging is needed */
+	page_t*		page)	/* out: uncompressed page, may be trashed */
 {
 	z_stream	d_stream;
 	dict_index_t*	index	= NULL;
@@ -1847,10 +1838,6 @@ page_zip_decompress(
 	ut_a(page_is_comp(page));
 	ut_ad(page_simple_validate_new(page));
 
-	if (UNIV_LIKELY_NULL(mtr)) {
-		mlog_write_initial_log_record(page, MLOG_ZIP_DECOMPRESS, mtr);
-	}
-
 	page_zip_fields_free(index);
 	mem_heap_free(heap);
 
@@ -1876,7 +1863,7 @@ page_zip_validate(
 		== page_zip);
 	ut_a(page_is_comp((page_t*) page));
 
-	valid = page_zip_decompress(&temp_page_zip, temp_page, NULL);
+	valid = page_zip_decompress(&temp_page_zip, temp_page);
 	if (!valid) {
 		fputs("page_zip_validate(): failed to decompress\n", stderr);
 		goto func_exit;
@@ -2491,8 +2478,7 @@ page_zip_clear_rec(
 	page_zip_des_t*	page_zip,/* in/out: compressed page */
 	byte*		rec,	/* in: record to clear */
 	dict_index_t*	index,	/* in: index of rec */
-	const ulint*	offsets,/* in: rec_get_offsets(rec, index) */
-	mtr_t*		mtr)	/* in: mini-transaction */
+	const ulint*	offsets)/* in: rec_get_offsets(rec, index) */
 {
 	ulint	heap_no;
 	page_t*	page	= ut_align_down(rec, UNIV_PAGE_SIZE);
@@ -2560,8 +2546,7 @@ page_zip_clear_rec(
 		/* Do not touch the extra bytes, because the
 		decompressor depends on them. */
 		memset(rec, 0, rec_offs_data_size(offsets));
-		if (UNIV_UNLIKELY(!page_zip_compress(page_zip, page,
-					index, mtr))) {
+		if (UNIV_UNLIKELY(!page_zip_compress(page_zip, page, index))) {
 			/* Compression failed. Restore the block. */
 			memcpy(rec, buf, rec_offs_data_size(offsets));
 			/* From now on, page_zip_validate() would fail
@@ -2721,7 +2706,7 @@ page_zip_dir_delete(
 	to be 0 for deleted records. */
 	rec[-REC_N_NEW_EXTRA_BYTES] = 0; /* info_bits and n_owned */
 
-	page_zip_clear_rec(page_zip, rec, index, offsets, NULL);
+	page_zip_clear_rec(page_zip, rec, index, offsets);
 }
 
 /**************************************************************************