branches/zip: btr_cur_pessimistic_insert(): Prior to splitting, ensure that

the record will fit on an empty compressed page.

buf_block_alloc(), buf_block_free(), buf_frame_alloc(), buf_frame_free():
Make inlined.

btr/btr0cur.c

@@ -1229,6 +1229,7 @@ btr_cur_pessimistic_insert(
 	mtr_t*		mtr)	/* in: mtr */
 {
 	dict_index_t*	index		= cursor->index;
+	ulint		zip_size	= dict_table_zip_size(index->table);
 	big_rec_t*	big_rec_vec	= NULL;
 	mem_heap_t*	heap		= NULL;
 	page_t*		page;
@@ -1289,8 +1290,7 @@ btr_cur_pessimistic_insert(
 	}
 
 	if (page_zip_rec_needs_ext(rec_get_converted_size(index, entry),
-			page_is_comp(page),
-			dict_table_zip_size(index->table))) {
+			page_is_comp(page), zip_size)) {
 		/* The record is so big that we have to store some fields
 		externally on separate database pages */
 
@@ -1326,6 +1326,47 @@ btr_cur_pessimistic_insert(
 		ext = more_ext;
 	}
 
+	if (UNIV_UNLIKELY(zip_size)) {
+		/* Estimate the free space of an empty compressed page.
+		The space needed for compressing the index information
+		is estimated. */
+		ulint	free_space_zip = zip_size
+				- PAGE_DATA - cursor->index->n_fields / 2;
+
+		if (UNIV_UNLIKELY(rec_get_converted_size(index, entry)
+				> free_space_zip)) {
+			/* Try to insert the record by itself on a new page.
+			If it fails, no amount of splitting will help. */
+			buf_block_t*	temp_block
+					= buf_block_alloc(zip_size);
+			page_t*		temp_page
+					= page_create_zip(temp_block->frame,
+					&temp_block->page_zip, index, 0, NULL);
+			page_cur_t	temp_cursor;
+			rec_t*		temp_rec;
+
+			page_cur_position(temp_page + PAGE_NEW_INFIMUM, &temp_cursor);
+
+			temp_rec = page_cur_tuple_insert(&temp_cursor,
+					&temp_block->page_zip, entry, index,
+					ext, n_ext, NULL);
+			buf_block_free(temp_block);
+
+			if (UNIV_UNLIKELY(!temp_rec)) {
+				if (big_rec_vec) {
+					dtuple_convert_back_big_rec(
+						index, entry, big_rec_vec);
+				}
+
+				if (heap) {
+					mem_heap_free(heap);
+				}
+
+				return(DB_TOO_BIG_RECORD);
+			}
+		}
+	}
+
 	if (dict_tree_get_page(index->tree)
 					== buf_frame_get_page_no(page)) {
 

buf/buf0buf.c

@@ -925,25 +925,6 @@ buf_awe_map_page_to_frame(
 	ut_a(0);
 }
 
-/************************************************************************
-Allocates a buffer block. */
-UNIV_INLINE
-buf_block_t*
-buf_block_alloc(
-/*============*/
-				/* out, own: the allocated block; also if AWE
-				is used it is guaranteed that the page is
-				mapped to a frame */
-	ulint	zip_size)	/* in: compressed page size in bytes,
-				or 0 if uncompressed tablespace */
-{
-	buf_block_t*	block;
-
-	block = buf_LRU_get_free_block(zip_size);
-
-	return(block);
-}
-
 /************************************************************************
 Moves to the block to the start of the LRU list if there is a danger
 that the block would drift out of the buffer pool. */
@@ -986,45 +967,6 @@ buf_page_make_young(
 	mutex_exit(&(buf_pool->mutex));
 }
 
-/************************************************************************
-Frees a buffer block which does not contain a file page. */
-UNIV_INLINE
-void
-buf_block_free(
-/*===========*/
-	buf_block_t*	block)	/* in, own: block to be freed */
-{
-	ut_a(block->state != BUF_BLOCK_FILE_PAGE);
-
-	mutex_enter(&(buf_pool->mutex));
-
-	buf_LRU_block_free_non_file_page(block);
-
-	mutex_exit(&(buf_pool->mutex));
-}
-
-/*************************************************************************
-Allocates a buffer frame. */
-
-buf_frame_t*
-buf_frame_alloc(void)
-/*=================*/
-				/* out: buffer frame */
-{
-	return(buf_block_alloc(0)->frame);
-}
-
-/*************************************************************************
-Frees a buffer frame which does not contain a file page. */
-
-void
-buf_frame_free(
-/*===========*/
-	buf_frame_t*	frame)	/* in: buffer frame */
-{
-	buf_block_free(buf_block_align(frame));
-}
-
 /************************************************************************
 Returns the buffer control block if the page can be found in the buffer
 pool. NOTE that it is possible that the page is not yet read

include/buf0buf.h

@@ -108,16 +108,34 @@ buf_pool_get_oldest_modification(void);
 /*==================================*/
 				/* out: oldest modification in pool,
 				ut_dulint_zero if none */
+/************************************************************************
+Allocates a buffer block. */
+UNIV_INLINE
+buf_block_t*
+buf_block_alloc(
+/*============*/
+				/* out, own: the allocated block; also if AWE
+				is used it is guaranteed that the page is
+				mapped to a frame */
+	ulint	zip_size);	/* in: compressed page size in bytes,
+				or 0 if uncompressed tablespace */
+/************************************************************************
+Frees a buffer block which does not contain a file page. */
+UNIV_INLINE
+void
+buf_block_free(
+/*===========*/
+	buf_block_t*	block);	/* in, own: block to be freed */
 /*************************************************************************
 Allocates a buffer frame. */
-
+UNIV_INLINE
 buf_frame_t*
 buf_frame_alloc(void);
 /*==================*/
 				/* out: buffer frame */
 /*************************************************************************
 Frees a buffer frame which does not contain a file page. */
-
+UNIV_INLINE
 void
 buf_frame_free(
 /*===========*/

include/buf0buf.ic

@@ -361,6 +361,64 @@ buf_frame_get_lock_mutex(
 	return(block->lock_mutex);
 }
 
+/************************************************************************
+Allocates a buffer block. */
+UNIV_INLINE
+buf_block_t*
+buf_block_alloc(
+/*============*/
+				/* out, own: the allocated block; also if AWE
+				is used it is guaranteed that the page is
+				mapped to a frame */
+	ulint	zip_size)	/* in: compressed page size in bytes,
+				or 0 if uncompressed tablespace */
+{
+	buf_block_t*	block;
+
+	block = buf_LRU_get_free_block(zip_size);
+
+	return(block);
+}
+
+/************************************************************************
+Frees a buffer block which does not contain a file page. */
+UNIV_INLINE
+void
+buf_block_free(
+/*===========*/
+	buf_block_t*	block)	/* in, own: block to be freed */
+{
+	ut_a(block->state != BUF_BLOCK_FILE_PAGE);
+
+	mutex_enter(&(buf_pool->mutex));
+
+	buf_LRU_block_free_non_file_page(block);
+
+	mutex_exit(&(buf_pool->mutex));
+}
+
+/*************************************************************************
+Allocates a buffer frame. */
+UNIV_INLINE
+buf_frame_t*
+buf_frame_alloc(void)
+/*=================*/
+				/* out: buffer frame */
+{
+	return(buf_block_alloc(0)->frame);
+}
+
+/*************************************************************************
+Frees a buffer frame which does not contain a file page. */
+UNIV_INLINE
+void
+buf_frame_free(
+/*===========*/
+	buf_frame_t*	frame)	/* in: buffer frame */
+{
+	buf_block_free(buf_block_align(frame));
+}
+
 /*************************************************************************
 Copies contents of a buffer frame to a given buffer. */
 UNIV_INLINE

include/page0cur.h

@@ -135,7 +135,7 @@ page_cur_tuple_insert(
 	dict_index_t*	index,	/* in: record descriptor */
 	const ulint*	ext,	/* in: array of extern field numbers */
 	ulint		n_ext,	/* in: number of elements in vec */
-	mtr_t*		mtr);	/* in: mini-transaction handle */
+	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 cursor stays at
@@ -151,7 +151,7 @@ page_cur_rec_insert(
 	rec_t*		rec,	/* in: record to insert */
 	dict_index_t*	index,	/* in: record descriptor */
 	ulint*		offsets,/* in/out: rec_get_offsets(rec, index) */
-	mtr_t*		mtr);	/* in: mini-transaction handle */
+	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
@@ -170,7 +170,7 @@ page_cur_insert_rec_low(
 	ulint*		offsets,/* in/out: rec_get_offsets(rec, index) */
 	const ulint*	ext,	/* in: array of extern field numbers */
 	ulint		n_ext,	/* in: number of elements in vec */
-	mtr_t*		mtr);	/* in: mini-transaction handle */
+	mtr_t*		mtr);	/* in: mini-transaction handle, or NULL */
 /*****************************************************************
 Copies records from page to a newly created page, from a given record onward,
 including that record. Infimum and supremum records are not copied. */

include/page0cur.ic

@@ -186,7 +186,7 @@ page_cur_tuple_insert(
 	dict_index_t*	index,	/* in: record descriptor */
 	const ulint*	ext,	/* in: array of extern field numbers */
 	ulint		n_ext,	/* in: number of elements in vec */
-	mtr_t*		mtr)	/* in: mini-transaction handle */
+	mtr_t*		mtr)	/* in: mini-transaction handle, or NULL */
 {
 	mem_heap_t*	heap;
 	ulint*		offsets;
@@ -221,7 +221,7 @@ page_cur_rec_insert(
 	rec_t*		rec,	/* in: record to insert */
 	dict_index_t*	index,	/* in: record descriptor */
 	ulint*		offsets,/* in/out: rec_get_offsets(rec, index) */
-	mtr_t*		mtr)	/* in: mini-transaction handle */
+	mtr_t*		mtr)	/* in: mini-transaction handle, or NULL */
 {
 	return(page_cur_insert_rec_low(cursor, page_zip,
 					index, rec, offsets, NULL, 0, mtr));

page/page0cur.c

@@ -897,7 +897,7 @@ page_cur_insert_rec_low(
 	ulint*		offsets,/* in/out: rec_get_offsets(rec, index) */
 	const ulint*	ext,	/* in: array of extern field numbers */
 	ulint		n_ext,	/* in: number of elements in vec */
-	mtr_t*		mtr)	/* in: mini-transaction handle */
+	mtr_t*		mtr)	/* in: mini-transaction handle, or NULL */
 {
 	byte*		insert_buf	= NULL;
 	ulint		rec_size;
@@ -909,7 +909,7 @@ page_cur_insert_rec_low(
 					new record is inserted */
 	page_zip_des_t*	page_zip_orig	= page_zip;
 
-	ut_ad(cursor && mtr);
+	ut_ad(cursor);
 	ut_ad(rec_offs_validate(rec, index, offsets));
 
 	page = page_cur_get_page(cursor);
@@ -1150,8 +1150,10 @@ use_heap:
 	}
 
 	/* 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);
+	}
 
 	return(insert_rec);
 }