branches/zip: Account for the compressed page size when checking if a

record will fit or need external storage.

btr_page_get_sure_split_rec(): Estimate the free space of an empty
compressed page.

page_zip_rec_needs_ext(): New function, to replace existing tests whether
external storage is needed.

btr/btr0btr.c

@@ -1301,6 +1301,7 @@ btr_page_get_sure_split_rec(
 	dtuple_t*	tuple)		/* in: tuple to insert */
 {
 	page_t*	page;
+	page_zip_des_t*	page_zip;
 	ulint	insert_size;
 	ulint	free_space;
 	ulint	total_data;
@@ -1319,6 +1320,20 @@ btr_page_get_sure_split_rec(
 	insert_size = rec_get_converted_size(cursor->index, tuple);
 	free_space  = page_get_free_space_of_empty(page_is_comp(page));
 
+	page_zip = buf_block_get_page_zip(buf_block_align(page));
+	if (UNIV_LIKELY_NULL(page_zip)) {
+		/* Estimate the free space of an empty compressed page.
+		The space needed for compressing the index information
+		is estimated. */
+		ulint	free_space_zip = page_zip->size
+				- PAGE_DATA - cursor->index->n_fields / 2;
+
+		if (UNIV_LIKELY(free_space > free_space_zip)) {
+			free_space = free_space_zip;
+			ut_a(insert_size <= free_space);
+		}
+	}
+
 	/* free_space is now the free space of a created new page */
 
 	total_data   = page_get_data_size(page) + insert_size;

btr/btr0cur.c

@@ -1026,6 +1026,7 @@ btr_cur_optimistic_insert(
 
 	page = btr_cur_get_page(cursor);
 	index = cursor->index;
+	page_zip = buf_block_get_page_zip(buf_block_align(page));
 
 	if (!dtuple_check_typed_no_assert(entry)) {
 		fputs("InnoDB: Error in a tuple to insert into ", stderr);
@@ -1047,9 +1048,8 @@ calculate_sizes_again:
 	/* Calculate the record size when entry is converted to a record */
 	rec_size = rec_get_converted_size(index, entry);
 
-	if (rec_size >=
-		ut_min(page_get_free_space_of_empty(page_is_comp(page)) / 2,
-		REC_MAX_DATA_SIZE)) {
+	if (page_zip_rec_needs_ext(rec_size, page_is_comp(page),
+			page_zip ? page_zip->size : 0)) {
 
 		/* The record is so big that we have to store some fields
 		externally on separate database pages */
@@ -1130,7 +1130,6 @@ calculate_sizes_again:
 	}
 
 	/* Now, try the insert */
-	page_zip = buf_block_get_page_zip(buf_block_align(page));
 
 	*rec = page_cur_tuple_insert(page_cursor, page_zip,
 					entry, index, ext, n_ext, mtr);
@@ -1289,10 +1288,9 @@ btr_cur_pessimistic_insert(
 		}
 	}
 
-	if (rec_get_converted_size(index, entry) >=
-		ut_min(page_get_free_space_of_empty(page_is_comp(page)) / 2,
-		REC_MAX_DATA_SIZE)) {
-
+	if (page_zip_rec_needs_ext(rec_get_converted_size(index, entry),
+			page_is_comp(page),
+			dict_table_zip_size(index->table))) {
 		/* The record is so big that we have to store some fields
 		externally on separate database pages */
 
@@ -2059,9 +2057,8 @@ btr_cur_pessimistic_update(
 					ULINT_UNDEFINED, &heap);
 	n_ext_vect = btr_push_update_extern_fields(ext_vect, offsets, update);
 
-	if (UNIV_UNLIKELY(rec_get_converted_size(index, new_entry) >=
-		ut_min(page_get_free_space_of_empty(page_is_comp(page)) / 2,
-		REC_MAX_DATA_SIZE))) {
+	if (page_zip_rec_needs_ext(rec_get_converted_size(index, new_entry),
+			page_is_comp(page), page_zip ? page_zip->size : 0)) {
 		ulint	i;
 
 		big_rec_vec = dtuple_convert_big_rec(index, new_entry,

data/data0data.c

@@ -532,10 +532,9 @@ dtuple_convert_big_rec(
 
 	n_fields = 0;
 
-	while (rec_get_converted_size(index, entry)
-		>= ut_min(page_get_free_space_of_empty(
-				  dict_table_is_comp(index->table)) / 2,
-			REC_MAX_DATA_SIZE)) {
+	while (page_zip_rec_needs_ext(rec_get_converted_size(index, entry),
+			dict_table_is_comp(index->table),
+			dict_table_zip_size(index->table))) {
 
 		longest = 0;
 		for (i = dict_index_get_n_unique_in_tree(index);

include/page0zip.h

@@ -19,6 +19,19 @@ Created June 2005 by Marko Makela
 #include "dict0types.h"
 #include "ut0byte.h"
 
+/**************************************************************************
+Determine if a record is so big that it needs to be stored externally. */
+UNIV_INLINE
+ibool
+page_zip_rec_needs_ext(
+/*===================*/
+				/* out: FALSE if the entire record
+				can be stored locally on the page */
+	ulint	rec_size,	/* in: length of the record in bytes */
+	ulint	comp,		/* in: nonzero=compact format */
+	ulint	zip_size)	/* in: compressed page size in bytes, or 0 */
+	__attribute__((const));
+
 /**************************************************************************
 Initialize a compressed page descriptor. */
 UNIV_INLINE

include/page0zip.ic

@@ -92,6 +92,44 @@ In summary, the compressed page looks like this:
 /* 'deleted' flag */
 #define PAGE_ZIP_DIR_SLOT_DEL	0x8000
 
+/**************************************************************************
+Determine if a record is so big that it needs to be stored externally. */
+UNIV_INLINE
+ibool
+page_zip_rec_needs_ext(
+/*===================*/
+				/* out: FALSE if the entire record
+				can be stored locally on the page */
+	ulint	rec_size,	/* in: length of the record in bytes */
+	ulint	comp,		/* in: nonzero=compact format */
+	ulint	zip_size)	/* in: compressed page size in bytes, or 0 */
+{
+	ut_ad(rec_size > comp ? REC_N_NEW_EXTRA_BYTES : REC_N_OLD_EXTRA_BYTES);
+	ut_ad(ut_is_2pow(zip_size));
+
+#if UNIV_PAGE_SIZE > REC_MAX_DATA_SIZE
+	if (UNIV_UNLIKELY(rec_size >= REC_MAX_DATA_SIZE)) {
+		return(TRUE);
+	}
+#endif
+
+	if (UNIV_UNLIKELY(!comp)) {
+		ut_ad(!zip_size);
+		return(rec_size >= page_get_free_space_of_empty(FALSE) / 2);
+	}
+
+	/* If zip_size != 0, the record should fit on the compressed page.
+	If not, the right-hand-side of the comparison will overwrap
+	and the condition will not hold.  Thus, we do not need to test
+	for zip_size != 0.  We subtract the size of the page header and
+	assume that compressing the index information takes 50 bytes. */
+	if (rec_size >= zip_size - (PAGE_DATA + 50)) {
+		return(TRUE);
+	}
+
+	return(rec_size >= page_get_free_space_of_empty(TRUE) / 2);
+}
+
 #ifdef UNIV_DEBUG
 /**************************************************************************
 Validate a compressed page descriptor. */