branches/zip: Fix some crash recovery bugs.

page_zip_alloc(): Add parameter "mtr" and log successful calls
to page_zip_compress().

page_zip_write_blob_ptr(), page_zip_write_node_ptr(): Write the offset on
the uncompressed page, because mlog_write_initial_log_record_fast()
does not do so.

page_zip_write_header_log(), page_zip_parse_write_header(): Encode the
offset in one byte.

btr/btr0cur.c

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

include/page0zip.h

@@ -94,8 +94,9 @@ page_zip_alloc(
 	const page_t*	page,	/* in: uncompressed page */
 	dict_index_t*	index,	/* in: index of the B-tree node */
 	ulint		length,	/* in: combined size of the record */
-	ulint		create)	/* in: nonzero=add the record to the heap */
-	__attribute__((warn_unused_result, nonnull));
+	ulint		create,	/* in: nonzero=add the record to the heap */
+	mtr_t*		mtr)	/* in: mini-transaction, or NULL */
+	__attribute__((warn_unused_result, nonnull(1,2,3)));
 
 /**************************************************************************
 Write an entire record on the compressed page.  The data must already

include/page0zip.ic

@@ -156,7 +156,8 @@ page_zip_alloc(
 	const page_t*	page,	/* in: uncompressed page */
 	dict_index_t*	index,	/* in: index of the B-tree node */
 	ulint		length,	/* in: combined size of the record */
-	ulint		create)	/* in: nonzero=add the record to the heap */
+	ulint		create,	/* in: nonzero=add the record to the heap */
+	mtr_t*		mtr)	/* in: mini-transaction, or NULL */
 {
 	ut_ad(page_is_comp((page_t*) page));
 	ut_ad(page_zip_validate(page_zip, page));
@@ -171,7 +172,11 @@ page_zip_alloc(
 		return(FALSE);
 	}
 
-	if (!page_zip_compress(page_zip, page, index)) {
+	if (page_zip_compress(page_zip, page, index)) {
+		if (mtr) {
+			page_zip_compress_write_log(page_zip, page, mtr);
+		}
+	} else {
 		/* Unable to compress the page */
 		return(FALSE);
 	}

page/page0cur.c

@@ -921,7 +921,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, rec_size, 1)) {
+	    && !page_zip_alloc(page_zip, page, index, rec_size, 1, mtr)) {
 
 		return(NULL);
 	}

page/page0zip.c

@@ -2262,13 +2262,16 @@ page_zip_write_blob_ptr(
 
 	if (mtr) {
 		byte*	log_ptr	= mlog_open(mtr,
-				11 + 2 + BTR_EXTERN_FIELD_REF_SIZE);
+				11 + 2 + 2 + BTR_EXTERN_FIELD_REF_SIZE);
 		if (UNIV_UNLIKELY(!log_ptr)) {
 			return;
 		}
 
 		log_ptr = mlog_write_initial_log_record_fast((byte*) field,
 				MLOG_ZIP_WRITE_BLOB_PTR, log_ptr, mtr);
+		mach_write_to_2(log_ptr,
+				ut_align_offset(field, UNIV_PAGE_SIZE));
+		log_ptr += 2;
 		mach_write_to_2(log_ptr, externs - page_zip->data);
 		log_ptr += 2;
 		memcpy(log_ptr, externs, BTR_EXTERN_FIELD_REF_SIZE);
@@ -2397,13 +2400,17 @@ page_zip_write_node_ptr(
 	memcpy(storage, field, REC_NODE_PTR_SIZE);
 
 	if (mtr) {
-		byte*	log_ptr	= mlog_open(mtr, 11 + 2 + REC_NODE_PTR_SIZE);
+		byte*	log_ptr	= mlog_open(mtr,
+				11 + 2 + 2 + REC_NODE_PTR_SIZE);
 		if (UNIV_UNLIKELY(!log_ptr)) {
 			return;
 		}
 
 		log_ptr = mlog_write_initial_log_record_fast(field,
 				MLOG_ZIP_WRITE_NODE_PTR, log_ptr, mtr);
+		mach_write_to_2(log_ptr,
+				ut_align_offset(field, UNIV_PAGE_SIZE));
+		log_ptr += 2;
 		mach_write_to_2(log_ptr, storage - page_zip->data);
 		log_ptr += 2;
 		memcpy(log_ptr, field, REC_NODE_PTR_SIZE);
@@ -2780,13 +2787,12 @@ page_zip_parse_write_header(
 
 	ut_ad(!page == !page_zip);
 
-	if (UNIV_UNLIKELY(end_ptr < ptr + (2 + 1))) {
+	if (UNIV_UNLIKELY(end_ptr < ptr + (1 + 1))) {
 
 		return(NULL);
 	}
 
-	offset = mach_read_from_2(ptr);
-	ptr += 2;
+	offset = (ulint) *ptr++;
 	len = (ulint) *ptr++;
 
 	if (UNIV_UNLIKELY(!len) || UNIV_UNLIKELY(offset + len >= PAGE_DATA)
@@ -2828,7 +2834,7 @@ page_zip_write_header_log(
 	ulint			length,	/* in: length of the data */
 	mtr_t*			mtr)	/* in: mini-transaction */
 {
-	byte*	log_ptr = mlog_open(mtr, 11 + 1);
+	byte*	log_ptr = mlog_open(mtr, 11 + 1 + 1);
 	ut_ad(offset < PAGE_DATA);
 	ut_ad(offset + length < PAGE_DATA);
 #if PAGE_DATA > 255
@@ -2844,6 +2850,7 @@ page_zip_write_header_log(
 
 	log_ptr = mlog_write_initial_log_record_fast(page_zip->data + offset,
 				MLOG_ZIP_WRITE_HEADER, log_ptr, mtr);
+	*log_ptr++ = (byte) offset;
 	*log_ptr++ = (byte) length;
 	mlog_close(mtr, log_ptr);