branches/zip: Split the function page_cur_insert_rec_low().

page_cur_insert_rec_zip_reorg(): New function: Recompress or
reorganize a compressed page.

page_cur_insert_rec_zip(): New function: insert a record to
a compressed page.

page_cur_insert_rec_low(): Only handle inserts to uncompressed pages.

include/page0cur.h

@@ -168,20 +168,35 @@ page_cur_rec_insert(
 	ulint*		offsets,/* in/out: rec_get_offsets(rec, index) */
 	mtr_t*		mtr);	/* in: mini-transaction handle, or NULL */
 /***************************************************************
-Inserts a record next to page cursor. Returns pointer to inserted record if
-succeed, i.e., enough space available, NULL otherwise. The record to be
-inserted can be in a data tuple or as a physical record. The other parameter
-must then be NULL. The cursor stays at the same position. */
+Inserts a record next to page cursor on an uncompressed page.
+Returns pointer to inserted record if succeed, i.e., enough
+space available, NULL otherwise. The cursor stays at the same position. */
 
 rec_t*
 page_cur_insert_rec_low(
+/*====================*/
+				/* out: pointer to record if succeed, NULL
+				otherwise */
+	rec_t*		current_rec,/* in: pointer to current record after
+				which the new record is inserted */
+	dict_index_t*	index,	/* in: record descriptor */
+	rec_t*		rec,	/* in: pointer to a physical record */
+	ulint*		offsets,/* in/out: rec_get_offsets(rec, index) */
+	mtr_t*		mtr);	/* in: mini-transaction handle, or NULL */
+/***************************************************************
+Inserts a record next to page cursor on a compressed and uncompressed
+page. Returns pointer to inserted record if succeed, i.e.,
+enough space available, NULL otherwise.
+The cursor stays at the same position. */
+
+rec_t*
+page_cur_insert_rec_zip(
 /*====================*/
 				/* out: pointer to record if succeed, NULL
 				otherwise */
 	rec_t**		current_rec,/* in/out: pointer to current record after
 				which the new record is inserted */
-	buf_block_t*	block,	/* in: buffer block of *current_rec, or NULL
-				if the compressed page is not to be updated */
+	buf_block_t*	block,	/* in: buffer block of *current_rec */
 	dict_index_t*	index,	/* in: record descriptor */
 	rec_t*		rec,	/* in: pointer to a physical record */
 	ulint*		offsets,/* in/out: rec_get_offsets(rec, index) */

include/page0cur.ic

@@ -235,8 +235,14 @@ page_cur_tuple_insert(
 					index, tuple, ext, n_ext);
 	offsets = rec_get_offsets(rec, index, NULL, ULINT_UNDEFINED, &heap);
 
-	rec = page_cur_insert_rec_low(&cursor->rec, cursor->block,
-				      index, rec, offsets, mtr);
+	if (buf_block_get_page_zip(cursor->block)) {
+		rec = page_cur_insert_rec_zip(&cursor->rec, cursor->block,
+					      index, rec, offsets, mtr);
+	} else {
+		rec = page_cur_insert_rec_low(cursor->rec,
+					      index, rec, offsets, mtr);
+	}
+
 	mem_heap_free(heap);
 	return(rec);
 }
@@ -257,7 +263,12 @@ page_cur_rec_insert(
 	ulint*		offsets,/* in/out: rec_get_offsets(rec, index) */
 	mtr_t*		mtr)	/* in: mini-transaction handle, or NULL */
 {
-	return(page_cur_insert_rec_low(&cursor->rec, cursor->block,
-				       index, rec, offsets, mtr));
+	if (buf_block_get_page_zip(cursor->block)) {
+		return(page_cur_insert_rec_zip(&cursor->rec, cursor->block,
+					       index, rec, offsets, mtr));
+	} else {
+		return(page_cur_insert_rec_low(cursor->rec,
+					       index, rec, offsets, mtr));
+	}
 }
 

include/page0zip.ic

@@ -348,7 +348,7 @@ Write data to the uncompressed header portion of a page.  The data must
 already have been written to the uncompressed page.
 However, the data portion of the uncompressed page may differ from
 the compressed page when a record is being inserted in
-page_cur_insert_rec_low(). */
+page_cur_insert_rec_zip(). */
 UNIV_INLINE
 void
 page_zip_write_header(
@@ -369,7 +369,7 @@ page_zip_write_header(
 
 	memcpy(page_zip->data + pos, str, length);
 
-	/* The following would fail in page_cur_insert_rec_low(). */
+	/* The following would fail in page_cur_insert_rec_zip(). */
 	/* ut_ad(page_zip_validate(page_zip, str - pos)); */
 
 	if (UNIV_LIKELY_NULL(mtr)) {

page/page0cur.c

@@ -883,20 +883,17 @@ page_cur_parse_insert_rec(
 }
 
 /***************************************************************
-Inserts a record next to page cursor. Returns pointer to inserted record if
-succeed, i.e., enough space available, NULL otherwise. The record to be
-inserted can be in a data tuple or as a physical record. The other parameter
-must then be NULL. The cursor stays at the same position. */
+Inserts a record next to page cursor on an uncompressed page.
+Returns pointer to inserted record if succeed, i.e., enough
+space available, NULL otherwise. The cursor stays at the same position. */
 
 rec_t*
 page_cur_insert_rec_low(
 /*====================*/
 				/* out: pointer to record if succeed, NULL
 				otherwise */
-	rec_t**		current_rec,/* in/out: pointer to current record after
+	rec_t*		current_rec,/* in: pointer to current record after
 				which the new record is inserted */
-	buf_block_t*	block,	/* in: buffer block of *current_rec, or NULL
-				if the compressed page is not to be updated */
 	dict_index_t*	index,	/* in: record descriptor */
 	rec_t*		rec,	/* in: pointer to a physical record */
 	ulint*		offsets,/* in/out: rec_get_offsets(rec, index) */
@@ -912,31 +909,19 @@ page_cur_insert_rec_low(
 	rec_t*		insert_rec;	/* inserted record */
 	ulint		heap_no;	/* heap number of the inserted
 					record */
-	page_zip_des_t*	page_zip;
-
-	page_zip = block ? buf_block_get_page_zip(block) : NULL;
 
 	ut_ad(rec_offs_validate(rec, index, offsets));
 
-	page = page_align(*current_rec);
+	page = page_align(current_rec);
 	ut_ad(dict_table_is_comp(index->table)
 	      == (ibool) !!page_is_comp(page));
 
-	ut_ad(!page_rec_is_supremum(*current_rec));
-#ifdef UNIV_ZIP_DEBUG
-	ut_a(!page_zip || page_zip_validate(page_zip, page));
-#endif /* UNIV_ZIP_DEBUG */
+	ut_ad(!page_rec_is_supremum(current_rec));
 
 	/* 1. Get the size of the physical record in the page */
 	rec_size = rec_offs_size(offsets);
 
 	/* 2. Try to find suitable space from page memory management */
-	if (UNIV_LIKELY_NULL(page_zip)
-	    && !page_zip_available(page_zip, index, rec_size, 1)) {
-
-		/* Try compressing the whole page afterwards. */
-		page_zip = NULL;
-	}
 
 	free_rec = page_header_get_ptr(page, PAGE_FREE);
 	if (UNIV_LIKELY_NULL(free_rec)) {
@@ -950,7 +935,6 @@ page_cur_insert_rec_low(
 		foffsets = rec_get_offsets(free_rec, index, foffsets,
 					ULINT_UNDEFINED, &heap);
 		if (rec_offs_size(foffsets) < rec_size) {
-too_small:
 			if (UNIV_LIKELY_NULL(heap)) {
 				mem_heap_free(heap);
 			}
@@ -961,36 +945,11 @@ page_cur_insert_rec_low(
 		insert_buf = free_rec - rec_offs_extra_size(foffsets);
 
 		if (page_is_comp(page)) {
-			if (UNIV_LIKELY_NULL(page_zip)) {
-				/* On compressed pages, do not
-				relocate records from the free list.
-				If extra_size would grow, use the heap. */
-				lint	extra_size_diff
-					= rec_offs_extra_size(offsets)
-					- rec_offs_extra_size(foffsets);
-
-				if (UNIV_UNLIKELY(extra_size_diff < 0)) {
-					/* Add an offset to the extra_size. */
-					if (rec_offs_size(foffsets)
-					    < rec_size - extra_size_diff) {
-
-						goto too_small;
-					}
-
-					insert_buf -= extra_size_diff;
-				} else if (UNIV_UNLIKELY(extra_size_diff)) {
-					/* Do not allow extra_size to grow */
-
-					goto too_small;
-				}
-			}
-
 			heap_no = rec_get_heap_no_new(free_rec);
-			page_mem_alloc_free(page, page_zip,
+			page_mem_alloc_free(page, NULL,
 					rec_get_next_ptr(free_rec, TRUE),
 					rec_size);
 		} else {
-			ut_ad(!page_zip);
 			heap_no = rec_get_heap_no_old(free_rec);
 			page_mem_alloc_free(page, NULL,
 					rec_get_next_ptr(free_rec, FALSE),
@@ -1003,17 +962,12 @@ page_cur_insert_rec_low(
 	} else {
 use_heap:
 		free_rec = NULL;
-		insert_buf = page_mem_alloc_heap(page, page_zip,
+		insert_buf = page_mem_alloc_heap(page, NULL,
 						 rec_size, &heap_no);
 
 		if (UNIV_UNLIKELY(insert_buf == NULL)) {
 			return(NULL);
 		}
-
-		if (UNIV_LIKELY_NULL(page_zip)) {
-			page_zip_dir_add_slot(page_zip,
-					dict_index_is_clust(index));
-		}
 	}
 
 	/* 3. Create the record */
@@ -1021,24 +975,24 @@ page_cur_insert_rec_low(
 	rec_offs_make_valid(insert_rec, index, offsets);
 
 	/* 4. Insert the record in the linked list of records */
-	ut_ad(*current_rec != insert_rec);
+	ut_ad(current_rec != insert_rec);
 
 	{
 		/* next record after current before the insertion */
-		rec_t*	next_rec = page_rec_get_next(*current_rec);
+		rec_t*	next_rec = page_rec_get_next(current_rec);
 #ifdef UNIV_DEBUG
 		if (page_is_comp(page)) {
-			ut_ad(rec_get_status(*current_rec)
+			ut_ad(rec_get_status(current_rec)
 				<= REC_STATUS_INFIMUM);
 			ut_ad(rec_get_status(insert_rec) < REC_STATUS_INFIMUM);
 			ut_ad(rec_get_status(next_rec) != REC_STATUS_INFIMUM);
 		}
 #endif
 		page_rec_set_next(insert_rec, next_rec);
-		page_rec_set_next(*current_rec, insert_rec);
+		page_rec_set_next(current_rec, insert_rec);
 	}
 
-	page_header_set_field(page, page_zip, PAGE_N_RECS,
+	page_header_set_field(page, NULL, PAGE_N_RECS,
 			      1 + page_get_n_recs(page));
 
 	/* 5. Set the n_owned field in the inserted record to zero,
@@ -1046,12 +1000,7 @@ page_cur_insert_rec_low(
 	if (page_is_comp(page)) {
 		rec_set_n_owned_new(insert_rec, NULL, 0);
 		rec_set_heap_no_new(insert_rec, heap_no);
-		if (UNIV_LIKELY_NULL(page_zip)) {
-			page_zip_dir_insert(page_zip, *current_rec, free_rec,
-					insert_rec);
-		}
 	} else {
-		ut_ad(!page_zip);
 		rec_set_n_owned_old(insert_rec, 0);
 		rec_set_heap_no_old(insert_rec, heap_no);
 	}
@@ -1064,17 +1013,17 @@ page_cur_insert_rec_low(
 	      == rec_get_node_ptr_flag(insert_rec));
 
 	if (UNIV_UNLIKELY(last_insert == NULL)) {
-		page_header_set_field(page, page_zip, PAGE_DIRECTION,
-							PAGE_NO_DIRECTION);
-		page_header_set_field(page, page_zip, PAGE_N_DIRECTION, 0);
+		page_header_set_field(page, NULL, PAGE_DIRECTION,
+				      PAGE_NO_DIRECTION);
+		page_header_set_field(page, NULL, PAGE_N_DIRECTION, 0);
 
-	} else if ((last_insert == *current_rec)
+	} else if ((last_insert == current_rec)
 		   && (page_header_get_field(page, PAGE_DIRECTION)
 		       != PAGE_LEFT)) {
 
-		page_header_set_field(page, page_zip, PAGE_DIRECTION,
+		page_header_set_field(page, NULL, PAGE_DIRECTION,
 							PAGE_RIGHT);
-		page_header_set_field(page, page_zip, PAGE_N_DIRECTION,
+		page_header_set_field(page, NULL, PAGE_N_DIRECTION,
 				      page_header_get_field(
 					      page, PAGE_N_DIRECTION) + 1);
 
@@ -1082,18 +1031,18 @@ page_cur_insert_rec_low(
 		   && (page_header_get_field(page, PAGE_DIRECTION)
 		       != PAGE_RIGHT)) {
 
-		page_header_set_field(page, page_zip, PAGE_DIRECTION,
+		page_header_set_field(page, NULL, PAGE_DIRECTION,
 							PAGE_LEFT);
-		page_header_set_field(page, page_zip, PAGE_N_DIRECTION,
+		page_header_set_field(page, NULL, PAGE_N_DIRECTION,
 				      page_header_get_field(
 					      page, PAGE_N_DIRECTION) + 1);
 	} else {
-		page_header_set_field(page, page_zip, PAGE_DIRECTION,
+		page_header_set_field(page, NULL, PAGE_DIRECTION,
 							PAGE_NO_DIRECTION);
-		page_header_set_field(page, page_zip, PAGE_N_DIRECTION, 0);
+		page_header_set_field(page, NULL, PAGE_N_DIRECTION, 0);
 	}
 
-	page_header_set_ptr(page, page_zip, PAGE_LAST_INSERT, insert_rec);
+	page_header_set_ptr(page, NULL, PAGE_LAST_INSERT, insert_rec);
 
 	/* 7. It remains to update the owner record. */
 	{
@@ -1101,7 +1050,7 @@ page_cur_insert_rec_low(
 		ulint	n_owned;
 		if (page_is_comp(page)) {
 			n_owned = rec_get_n_owned_new(owner_rec);
-			rec_set_n_owned_new(owner_rec, page_zip, n_owned + 1);
+			rec_set_n_owned_new(owner_rec, NULL, n_owned + 1);
 		} else {
 			n_owned = rec_get_n_owned_old(owner_rec);
 			rec_set_n_owned_old(owner_rec, n_owned + 1);
@@ -1112,56 +1061,290 @@ page_cur_insert_rec_low(
 		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_find_owner_slot(owner_rec));
+			page_dir_split_slot(
+				page, NULL,
+				page_dir_find_owner_slot(owner_rec));
 		}
 	}
 
-	if (UNIV_LIKELY_NULL(page_zip)) {
-		page_zip_write_rec(page_zip, insert_rec, index, offsets, 1);
-	} else if (UNIV_LIKELY_NULL(block)
-		   && UNIV_LIKELY_NULL(buf_block_get_page_zip(block))) {
-		ut_a(page_is_comp(page));
+	/* 9. Write log record of the insert */
+	if (UNIV_LIKELY(mtr != NULL)) {
+		page_cur_insert_rec_write_log(insert_rec, rec_size,
+					      current_rec, index, mtr);
+	}
 
-		page_zip = buf_block_get_page_zip(block);
+	return(insert_rec);
+}
 
-		/* Recompress or reorganize and recompress the page. */
-		if (UNIV_UNLIKELY
-		    (!page_zip_compress(page_zip, page, index, mtr))) {
-			/* Before trying to reorganize the page,
-			store the number of preceding records on the page. */
-			ulint		insert_pos
-				= page_rec_get_n_recs_before(insert_rec);
-
-			if (page_zip_reorganize(block, index, mtr)) {
-				/* The page was reorganized:
-				Seek to insert_pos to find insert_rec,
-				and update *current_rec. */
-				insert_rec = page + PAGE_NEW_INFIMUM;
-
-				while (--insert_pos) {
-					insert_rec = page + rec_get_next_offs(
-						insert_rec, TRUE);
-				}
-
-				*current_rec = insert_rec;
-				insert_rec = page
-					+ rec_get_next_offs(insert_rec, TRUE);
-
-				return(insert_rec);
+/***************************************************************
+Compresses or reorganizes a page after an optimistic insert. */
+static
+rec_t*
+page_cur_insert_rec_zip_reorg(
+/*==========================*/
+				/* out: rec if succeed, NULL otherwise */
+	rec_t**		current_rec,/* in/out: pointer to current record after
+				which the new record is inserted */
+	buf_block_t*	block,	/* in: buffer block */
+	dict_index_t*	index,	/* in: record descriptor */
+	rec_t*		rec,	/* in: inserted record */
+	page_t*		page,	/* in: uncompressed page */
+	page_zip_des_t*	page_zip,/* in: compressed page */
+	mtr_t*		mtr)	/* in: mini-transaction, or NULL */
+{
+	ulint		pos;
+
+	/* Recompress or reorganize and recompress the page. */
+	if (UNIV_LIKELY(page_zip_compress(page_zip, page, index, mtr))) {
+		return(rec);
+	}
+
+	/* Before trying to reorganize the page,
+	store the number of preceding records on the page. */
+	pos = page_rec_get_n_recs_before(rec);
+
+	if (page_zip_reorganize(block, index, mtr)) {
+		/* The page was reorganized: Find rec by seeking to pos,
+		and update *current_rec. */
+		rec = page + PAGE_NEW_INFIMUM;
+
+		while (--pos) {
+			rec = page + rec_get_next_offs(rec, TRUE);
+		}
+
+		*current_rec = rec;
+		rec = page + rec_get_next_offs(rec, TRUE);
+
+		return(rec);
+	}
+
+	/* Out of space: restore the page */
+	if (!page_zip_decompress(page_zip, page)) {
+		ut_error; /* Memory corrupted? */
+	}
+	ut_ad(page_validate(page, index));
+	return(NULL);
+}
+
+/***************************************************************
+Inserts a record next to page cursor on a compressed and uncompressed
+page. Returns pointer to inserted record if succeed, i.e.,
+enough space available, NULL otherwise.
+The cursor stays at the same position. */
+
+rec_t*
+page_cur_insert_rec_zip(
+/*====================*/
+				/* out: pointer to record if succeed, NULL
+				otherwise */
+	rec_t**		current_rec,/* in/out: pointer to current record after
+				which the new record is inserted */
+	buf_block_t*	block,	/* in: buffer block of *current_rec */
+	dict_index_t*	index,	/* in: record descriptor */
+	rec_t*		rec,	/* in: pointer to a physical record */
+	ulint*		offsets,/* in/out: rec_get_offsets(rec, index) */
+	mtr_t*		mtr)	/* in: mini-transaction handle, or NULL */
+{
+	byte*		insert_buf	= NULL;
+	ulint		rec_size;
+	page_t*		page;		/* the relevant page */
+	rec_t*		last_insert;	/* cursor position at previous
+					insert */
+	rec_t*		free_rec;	/* a free record that was reused,
+					or NULL */
+	rec_t*		insert_rec;	/* inserted record */
+	ulint		heap_no;	/* heap number of the inserted
+					record */
+	page_zip_des_t*	page_zip;
+
+	page_zip = buf_block_get_page_zip(block);
+	ut_ad(page_zip);
+
+	ut_ad(rec_offs_validate(rec, index, offsets));
+
+	page = page_align(*current_rec);
+	ut_ad(dict_table_is_comp(index->table));
+	ut_ad(page_is_comp(page));
+
+	ut_ad(!page_rec_is_supremum(*current_rec));
+#ifdef UNIV_ZIP_DEBUG
+	ut_a(page_zip_validate(page_zip, page));
+#endif /* UNIV_ZIP_DEBUG */
+
+	/* 1. Get the size of the physical record in the page */
+	rec_size = rec_offs_size(offsets);
+
+	/* 2. Try to find suitable space from page memory management */
+	if (!page_zip_available(page_zip, index, rec_size, 1)) {
+
+		/* Try compressing the whole page afterwards. */
+		insert_rec = page_cur_insert_rec_low(*current_rec,
+						     index, rec, offsets,
+						     NULL);
+
+		if (UNIV_LIKELY(insert_rec != NULL)) {
+			insert_rec = page_cur_insert_rec_zip_reorg(
+				current_rec, block, index, insert_rec,
+				page, page_zip, mtr);
+		}
+
+		return(insert_rec);
+	}
+
+	free_rec = page_header_get_ptr(page, PAGE_FREE);
+	if (UNIV_LIKELY_NULL(free_rec)) {
+		/* Try to allocate from the head of the free list. */
+		lint	extra_size_diff;
+		ulint		foffsets_[REC_OFFS_NORMAL_SIZE];
+		ulint*		foffsets	= foffsets_;
+		mem_heap_t*	heap		= NULL;
+
+		*foffsets = sizeof(foffsets_) / sizeof *foffsets_;
+
+		foffsets = rec_get_offsets(free_rec, index, foffsets,
+					ULINT_UNDEFINED, &heap);
+		if (rec_offs_size(foffsets) < rec_size) {
+too_small:
+			if (UNIV_LIKELY_NULL(heap)) {
+				mem_heap_free(heap);
 			}
 
-			/* Out of space: restore the page */
-			if (!page_zip_decompress(page_zip, page)) {
-				ut_error; /* Memory corrupted? */
+			goto use_heap;
+		}
+
+		insert_buf = free_rec - rec_offs_extra_size(foffsets);
+
+		/* On compressed pages, do not relocate records from
+		the free list.  If extra_size would grow, use the heap. */
+		extra_size_diff
+			= rec_offs_extra_size(offsets)
+			- rec_offs_extra_size(foffsets);
+
+		if (UNIV_UNLIKELY(extra_size_diff < 0)) {
+			/* Add an offset to the extra_size. */
+			if (rec_offs_size(foffsets)
+			    < rec_size - extra_size_diff) {
+
+				goto too_small;
 			}
-			ut_ad(page_validate(page, index));
+
+			insert_buf -= extra_size_diff;
+		} else if (UNIV_UNLIKELY(extra_size_diff)) {
+			/* Do not allow extra_size to grow */
+
+			goto too_small;
+		}
+
+		heap_no = rec_get_heap_no_new(free_rec);
+		page_mem_alloc_free(page, page_zip,
+				    rec_get_next_ptr(free_rec, TRUE),
+				    rec_size);
+
+		if (UNIV_LIKELY_NULL(heap)) {
+			mem_heap_free(heap);
+		}
+	} else {
+use_heap:
+		free_rec = NULL;
+		insert_buf = page_mem_alloc_heap(page, page_zip,
+						 rec_size, &heap_no);
+
+		if (UNIV_UNLIKELY(insert_buf == NULL)) {
 			return(NULL);
 		}
 
-		return(insert_rec);
+		page_zip_dir_add_slot(page_zip, dict_index_is_clust(index));
+	}
+
+	/* 3. Create the record */
+	insert_rec = rec_copy(insert_buf, rec, offsets);
+	rec_offs_make_valid(insert_rec, index, offsets);
+
+	/* 4. Insert the record in the linked list of records */
+	ut_ad(*current_rec != insert_rec);
+
+	{
+		/* next record after current before the insertion */
+		rec_t*	next_rec = page_rec_get_next(*current_rec);
+		ut_ad(rec_get_status(*current_rec)
+		      <= REC_STATUS_INFIMUM);
+		ut_ad(rec_get_status(insert_rec) < REC_STATUS_INFIMUM);
+		ut_ad(rec_get_status(next_rec) != REC_STATUS_INFIMUM);
+
+		page_rec_set_next(insert_rec, next_rec);
+		page_rec_set_next(*current_rec, insert_rec);
 	}
 
+	page_header_set_field(page, page_zip, PAGE_N_RECS,
+			      1 + page_get_n_recs(page));
+
+	/* 5. Set the n_owned field in the inserted record to zero,
+	and set the heap_no field */
+	rec_set_n_owned_new(insert_rec, NULL, 0);
+	rec_set_heap_no_new(insert_rec, heap_no);
+
+	page_zip_dir_insert(page_zip, *current_rec, free_rec, insert_rec);
+
+	/* 6. Update the last insertion info in page header */
+
+	last_insert = page_header_get_ptr(page, PAGE_LAST_INSERT);
+	ut_ad(!last_insert
+	      || rec_get_node_ptr_flag(last_insert)
+	      == rec_get_node_ptr_flag(insert_rec));
+
+	if (UNIV_UNLIKELY(last_insert == NULL)) {
+		page_header_set_field(page, page_zip, PAGE_DIRECTION,
+							PAGE_NO_DIRECTION);
+		page_header_set_field(page, page_zip, PAGE_N_DIRECTION, 0);
+
+	} else if ((last_insert == *current_rec)
+		   && (page_header_get_field(page, PAGE_DIRECTION)
+		       != PAGE_LEFT)) {
+
+		page_header_set_field(page, page_zip, PAGE_DIRECTION,
+							PAGE_RIGHT);
+		page_header_set_field(page, page_zip, PAGE_N_DIRECTION,
+				      page_header_get_field(
+					      page, PAGE_N_DIRECTION) + 1);
+
+	} else if ((page_rec_get_next(insert_rec) == last_insert)
+		   && (page_header_get_field(page, PAGE_DIRECTION)
+		       != PAGE_RIGHT)) {
+
+		page_header_set_field(page, page_zip, PAGE_DIRECTION,
+							PAGE_LEFT);
+		page_header_set_field(page, page_zip, PAGE_N_DIRECTION,
+				      page_header_get_field(
+					      page, PAGE_N_DIRECTION) + 1);
+	} else {
+		page_header_set_field(page, page_zip, PAGE_DIRECTION,
+							PAGE_NO_DIRECTION);
+		page_header_set_field(page, page_zip, PAGE_N_DIRECTION, 0);
+	}
+
+	page_header_set_ptr(page, page_zip, PAGE_LAST_INSERT, insert_rec);
+
+	/* 7. It remains to update the owner record. */
+	{
+		rec_t*	owner_rec	= page_rec_find_owner_rec(insert_rec);
+		ulint	n_owned;
+
+		n_owned = rec_get_n_owned_new(owner_rec);
+		rec_set_n_owned_new(owner_rec, page_zip, n_owned + 1);
+
+		/* 8. Now we have incremented the n_owned field of the owner
+		record. If the number exceeds PAGE_DIR_SLOT_MAX_N_OWNED,
+		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_find_owner_slot(owner_rec));
+		}
+	}
+
+	page_zip_write_rec(page_zip, insert_rec, index, offsets, 1);
+
 	/* 9. Write log record of the insert */
 	if (UNIV_LIKELY(mtr != NULL)) {
 		page_cur_insert_rec_write_log(insert_rec, rec_size,

page/page0page.c

@@ -540,7 +540,7 @@ page_copy_rec_list_end_no_locks(
 		rec_t*	ins_rec;
 		offsets = rec_get_offsets(cur1_rec, index, offsets,
 					  ULINT_UNDEFINED, &heap);
-		ins_rec = page_cur_insert_rec_low(&cur2, NULL, index,
+		ins_rec = page_cur_insert_rec_low(cur2, index,
 						  cur1_rec, offsets, mtr);
 		if (UNIV_UNLIKELY(!ins_rec)) {
 			/* Track an assertion failure reported on the mailing
@@ -705,7 +705,7 @@ page_copy_rec_list_start(
 		rec_t*	cur1_rec = page_cur_get_rec(&cur1);
 		offsets = rec_get_offsets(cur1_rec, index, offsets,
 					  ULINT_UNDEFINED, &heap);
-		cur2 = page_cur_insert_rec_low(&cur2, NULL, index,
+		cur2 = page_cur_insert_rec_low(cur2, index,
 					       cur1_rec, offsets, mtr);
 		ut_a(cur2);
 

page/page0zip.c

@@ -3247,7 +3247,7 @@ page_zip_dir_insert(
 
 		if (UNIV_LIKELY(!free_rec)) {
 			/* PAGE_N_RECS was already incremented
-			in page_cur_insert_rec_low(), but the
+			in page_cur_insert_rec_zip(), but the
 			dense directory slot at that position
 			contains garbage.  Skip it. */
 			start += PAGE_ZIP_DIR_SLOT_SIZE;
@@ -3266,7 +3266,7 @@ page_zip_dir_insert(
 		/* The record was allocated from the free list.
 		Shift the dense directory only up to that slot.
 		Note that in this case, n_dense is actually
-		off by one, because page_cur_insert_rec_low()
+		off by one, because page_cur_insert_rec_zip()
 		did not increment n_heap. */
 		ut_ad(rec_get_heap_no_new(rec) < n_dense + 1
 		      + PAGE_HEAP_NO_USER_LOW);