branches/zip: Minor improvements.

page_dir_slot_set_n_owned(): Check page_zip_available().
page_dir_add_slots(): Add parameter page_zip.
btr_attach_half_pages(): Add parameter page_zip.
btr_node_ptr_set_child_page_no(): Add parameter page_zip.
Make comments about page_zip modification log usage more accurate.

btr/btr0btr.c

@@ -498,6 +498,8 @@ void
 btr_node_ptr_set_child_page_no(
 /*===========================*/
 	rec_t*		rec,	/* in: node pointer record */
+	page_zip_des_t*	page_zip,/* in/out: compressed page with at least
+				8 bytes available, or NULL */
 	const ulint*	offsets,/* in: array returned by rec_get_offsets() */
 	ulint		page_no,/* in: child node address */
 	mtr_t*		mtr)	/* in: mtr */
@@ -508,6 +510,7 @@ btr_node_ptr_set_child_page_no(
 	ut_ad(rec_offs_validate(rec, NULL, offsets));
 	ut_ad(0 < btr_page_get_level(buf_frame_align(rec), mtr));
 	ut_ad(!rec_offs_comp(offsets) || rec_get_node_ptr_flag(rec));
+	ut_ad(!page_zip || page_zip_available(page_zip, 8));
 
 	/* The child address is in the last field */	
 	field = rec_get_nth_field(rec, offsets,
@@ -516,6 +519,9 @@ btr_node_ptr_set_child_page_no(
 	ut_ad(len == 4);
 	
 	mlog_write_ulint(field, page_no, MLOG_4BYTES, mtr);
+	if (UNIV_LIKELY_NULL(page_zip)) {
+		page_zip_write(page_zip, field, 4);
+	}
 }
 
 /****************************************************************
@@ -1480,8 +1486,9 @@ void
 btr_attach_half_pages(
 /*==================*/
 	dict_tree_t*	tree,		/* in: the index tree */
-	page_t*		page,		/* in: page to be split */
-	/* TODO page_zip? */
+	page_t*		page,		/* in/out: page to be split */
+	page_zip_des_t*	page_zip,	/* in/out: compressed page with
+					at least 8 bytes available, or NULL */
 	rec_t*		split_rec,	/* in: first record on upper
 					half page */
 	page_t*		new_page,	/* in: the new half page */
@@ -1508,6 +1515,7 @@ btr_attach_half_pages(
 			      				MTR_MEMO_PAGE_X_FIX));
 	ut_ad(mtr_memo_contains(mtr, buf_block_align(new_page),
 			      				MTR_MEMO_PAGE_X_FIX));
+	ut_ad(!page_zip || page_zip_available(page_zip, 8));
 	ut_a(page_is_comp(page) == page_is_comp(new_page));
 
 	/* Create a memory heap where the data tuple is stored */
@@ -1527,7 +1535,7 @@ btr_attach_half_pages(
 		/* Replace the address of the old child node (= page) with the 
 		address of the new lower half */
 
-		btr_node_ptr_set_child_page_no(node_ptr/* TODO zip */,
+		btr_node_ptr_set_child_page_no(node_ptr, page_zip,
 			rec_get_offsets(node_ptr,
 					UT_LIST_GET_FIRST(tree->tree_indexes),
 					NULL, ULINT_UNDEFINED, &heap),
@@ -1619,6 +1627,7 @@ btr_page_split_and_insert(
 {
 	dict_tree_t*	tree;
 	page_t*		page;
+	page_zip_des_t*	page_zip;
 	ulint		page_no;
 	byte		direction;
 	ulint		hint_page_no;
@@ -1653,6 +1662,12 @@ func_start:
 #endif /* UNIV_SYNC_DEBUG */
 
 	page = btr_cur_get_page(cursor);
+	page_zip = buf_block_get_page_zip(buf_block_align(page));
+	if (UNIV_LIKELY_NULL(page_zip)) {
+		if (UNIV_UNLIKELY(!page_zip_available(page_zip, 8))) {
+			ut_error; /* TODO: split the page */
+		}
+	}
 
 	ut_ad(mtr_memo_contains(mtr, buf_block_align(page),
 			      				MTR_MEMO_PAGE_X_FIX));
@@ -1704,7 +1719,8 @@ func_start:
 	
 	/* 4. Do first the modifications in the tree structure */
 
-	btr_attach_half_pages(tree, page, first_rec, new_page, direction, mtr);
+	btr_attach_half_pages(tree, page, page_zip/* 8 */, first_rec,
+			new_page, direction, mtr);
 
 	/* If the split is made on the leaf level and the insert will fit
 	on the appropriate half-page, we may release the tree x-latch.
@@ -2105,6 +2121,7 @@ btr_compress(
 	page_t*		merge_page;
 	ibool		is_left;
 	page_t*		page;
+	page_zip_des_t*	page_zip;
 	rec_t*		node_ptr;
 	ulint		data_size;
 	ulint		n_recs;
@@ -2113,6 +2130,7 @@ btr_compress(
 	ulint		level;
 
 	page = btr_cur_get_page(cursor);
+	page_zip = buf_block_get_page_zip(buf_block_align(page));
 	tree = btr_cur_get_tree(cursor);
 	ut_a((ibool)!!page_is_comp(page) == cursor->index->table->comp);
 
@@ -2139,6 +2157,11 @@ btr_compress(
 
 	is_left = left_page_no != FIL_NULL;
 
+	if (!is_left && UNIV_LIKELY_NULL(page_zip)
+			&& !page_zip_available(page_zip, 8)) {
+		return(FALSE);
+	}
+
 	if (is_left) {
 
 		merge_page = btr_page_get(space, left_page_no, RW_X_LATCH,
@@ -2205,7 +2228,7 @@ btr_compress(
 		/* Replace the address of the old child node (= page) with the 
 		address of the merge page to the right */
 
-		btr_node_ptr_set_child_page_no(node_ptr,
+		btr_node_ptr_set_child_page_no(node_ptr, page_zip/* 8 */,
 				rec_get_offsets(node_ptr, cursor->index,
 				offsets_, ULINT_UNDEFINED, &heap),
 				right_page_no, mtr);
@@ -2215,8 +2238,6 @@ btr_compress(
 		btr_node_ptr_delete(tree, merge_page, mtr);
 	}
 
-	/* TODO: update page_zip of node_ptr */
-
 	/* Move records to the merge page */
 	if (is_left) {
 		rec_t*	orig_pred = page_rec_get_prev(

btr/btr0cur.c

@@ -1763,8 +1763,12 @@ btr_cur_optimistic_update(
 
 	rec = btr_cur_insert_if_possible(cursor, page_zip_used,
 					new_entry, &reorganized, mtr);
+	if (UNIV_UNLIKELY(!rec)) {
+		/* The above may only fail if page_zip_used != NULL */
+		ut_a(page_zip_used);
 
-	ut_a(rec); /* <- We calculated above the insert would fit */
+		goto zip_overflow;
+	}
 
 	if (!rec_get_deleted_flag(rec, page_is_comp(page))) {
 		/* The new inserted record owns its possible externally
@@ -1778,6 +1782,7 @@ btr_cur_optimistic_update(
 	if (UNIV_LIKELY_NULL(page_zip) && UNIV_UNLIKELY(!page_zip_used)) {
 		if (!page_zip_compress(page_zip, page)) {
 
+zip_overflow:
 			if (UNIV_UNLIKELY(!page_zip_decompress(
 					page_zip, page, mtr))) {
 				ut_error;

include/btr0btr.ic

@@ -42,7 +42,8 @@ void
 btr_page_set_index_id(
 /*==================*/
 	page_t*		page,	/* in: page to be created */
-	page_zip_des_t*	page_zip,/* in: compressed page, or NULL */
+	page_zip_des_t*	page_zip,/* in: compressed page whose uncompressed
+				part will be updated, or NULL */
 	dulint		id,	/* in: index id */
 	mtr_t*		mtr)	/* in: mtr */
 {
@@ -108,7 +109,8 @@ void
 btr_page_set_level(
 /*===============*/
 	page_t*		page,	/* in: index page */
-	page_zip_des_t*	page_zip,/* in: compressed page, or NULL */
+	page_zip_des_t*	page_zip,/* in: compressed page whose uncompressed
+				part will be updated, or NULL */
 	ulint		level,	/* in: level, leaf level == 0 */
 	mtr_t*		mtr)	/* in: mini-transaction handle */
 {
@@ -150,7 +152,8 @@ void
 btr_page_set_next(
 /*==============*/
 	page_t*		page,	/* in: index page */
-	page_zip_des_t*	page_zip,/* in: compressed page, or NULL */
+	page_zip_des_t*	page_zip,/* in: compressed page whose uncompressed
+				part will be updated, or NULL */
 	ulint		next,	/* in: next page number */
 	mtr_t*		mtr)	/* in: mini-transaction handle */
 {
@@ -185,7 +188,8 @@ void
 btr_page_set_prev(
 /*==============*/
 	page_t*		page,	/* in: index page */
-	page_zip_des_t*	page_zip,/* in: compressed page, or NULL */
+	page_zip_des_t*	page_zip,/* in: compressed page whose uncompressed
+				part will be updated, or NULL */
 	ulint		prev,	/* in: previous page number */
 	mtr_t*		mtr)	/* in: mini-transaction handle */
 {

include/page0cur.h

@@ -165,7 +165,7 @@ page_cur_insert_rec_low(
 				otherwise */
 	page_cur_t*	cursor,	/* in: a page cursor */
 	page_zip_des_t*	page_zip,/* in/out: compressed page with at least
-				25 + rec_size bytes available, or NULL */
+				37 + rec_size bytes available, or NULL */
 	dtuple_t*	tuple,	/* in: pointer to a data tuple or NULL */
 	dict_index_t*	index,	/* in: record descriptor */
 	rec_t*		rec,	/* in: pointer to a physical record or NULL */
@@ -254,7 +254,7 @@ page_cur_parse_insert_rec(
 	dict_index_t*	index,	/* in: record descriptor */
 	page_t*		page,	/* in/out: page or NULL */
 	page_zip_des_t*	page_zip,/* in/out: compressed page with at least
-				25 + rec_size bytes available, or NULL */
+				37 + rec_size bytes available, or NULL */
 	mtr_t*		mtr);	/* in: mtr or NULL */
 /**************************************************************
 Parses a log record of copying a record list end to a new created page. */

include/page0page.h

@@ -694,7 +694,7 @@ page_dir_split_slot(
 /*================*/
 	page_t*		page,	/* in: index page */
 	page_zip_des_t*	page_zip,/* in/out: compressed page with
-				at least 10 bytes available, or NULL */
+				at least 12 bytes available, or NULL */
 	ulint		slot_no)/* in: the directory slot */
 	__attribute__((nonnull(1)));
 /*****************************************************************

page/page0cur.c

@@ -890,7 +890,7 @@ page_cur_insert_rec_low(
 				otherwise */
 	page_cur_t*	cursor,	/* in: a page cursor */
 	page_zip_des_t*	page_zip,/* in/out: compressed page with at least
-				25 + rec_size bytes available, or NULL */
+				37 + rec_size bytes available, or NULL */
 	dtuple_t*	tuple,	/* in: pointer to a data tuple or NULL */
 	dict_index_t*	index,	/* in: record descriptor */
 	rec_t*		rec,	/* in: pointer to a physical record or NULL */
@@ -931,7 +931,7 @@ page_cur_insert_rec_low(
 
 	if (UNIV_LIKELY_NULL(page_zip)) {
 		if (UNIV_UNLIKELY(!page_zip_alloc(
-				page_zip, page, 25 + rec_size))) {
+				page_zip, page, 37 + rec_size))) {
 
 			goto err_exit;
 		}
@@ -1047,7 +1047,7 @@ err_exit:
 		we have to split the corresponding directory slot in two. */
 
 		if (UNIV_UNLIKELY(n_owned == PAGE_DIR_SLOT_MAX_N_OWNED)) {
-			page_dir_split_slot(page, page_zip,
+			page_dir_split_slot(page, page_zip/* 12 */,
 					page_dir_find_owner_slot(owner_rec));
 		}
 	}

page/page0page.c

@@ -1223,7 +1223,9 @@ void
 page_dir_add_slots(
 /*===============*/
 	page_t*		page,	/* in/out: the index page */
-	page_zip_des_t*	page_zip,/* in/out: comprssed page, or NULL */
+	page_zip_des_t*	page_zip,/* in/out: comprssed page with at least
+				n * PAGE_DIR_SLOT_SIZE bytes available,
+				or NULL */
 	ulint		start,	/* in: the slot above which the new slots
 				are added */
 	ulint		n)	/* in: number of slots to add
@@ -1240,6 +1242,9 @@ page_dir_add_slots(
 
 	ut_ad(start < n_slots - 1);
 
+	ut_ad(!page_zip
+		|| page_zip_available(page_zip, n * PAGE_DIR_SLOT_SIZE));
+
 	/* Update the page header */
 	page_dir_set_n_slots(page, page_zip, n_slots + n);
 
@@ -1263,7 +1268,7 @@ page_dir_split_slot(
 /*================*/
 	page_t*		page,	/* in/out: index page */
 	page_zip_des_t*	page_zip,/* in/out: compressed page with
-				at least 10 bytes available, or NULL */
+				at least 12 bytes available, or NULL */
 	ulint		slot_no)/* in: the directory slot */
 {		
 	rec_t*			rec;
@@ -1274,7 +1279,7 @@ page_dir_split_slot(
 	ulint			n_owned;
 
 	ut_ad(page);
-	ut_ad(!page_zip || page_zip_available(page_zip, 10));
+	ut_ad(!page_zip || page_zip_available(page_zip, 12));
 	ut_ad(!page_zip || page_is_comp(page));
 	ut_ad(slot_no > 0);
 
@@ -1298,7 +1303,7 @@ page_dir_split_slot(
 	/* 2. We add one directory slot immediately below the slot to be
 	split. */
 
-	page_dir_add_slots(page, page_zip, slot_no - 1, 1);
+	page_dir_add_slots(page, page_zip/* 2 */, slot_no - 1, 1);
 
 	/* The added slot is now number slot_no, and the old slot is
 	now number slot_no + 1 */
@@ -1309,12 +1314,13 @@ page_dir_split_slot(
 	/* 3. We store the appropriate values to the new slot. */
 	
 	page_dir_slot_set_rec(new_slot, page_zip, rec);
-	page_dir_slot_set_n_owned(new_slot, page_zip, n_owned / 2);
+	page_dir_slot_set_n_owned(new_slot, page_zip/* 5 */, n_owned / 2);
 	
 	/* 4. Finally, we update the number of records field of the 
 	original slot */
 
-	page_dir_slot_set_n_owned(slot, page_zip, n_owned - (n_owned / 2));
+	page_dir_slot_set_n_owned(slot, page_zip/* 5 */,
+				n_owned - (n_owned / 2));
 }
 
 /*****************************************************************