branches/zip: Remove in-place updates of the "external storage"

flag of records.  The flags may only be updated in heap-allocated
copies of records.

btr_root_raise_and_insert(),
btr_page_split_and_insert(),
btr_cur_insert_if_possible(),
btr_cur_optimistic_insert(),
btr_cur_pessimistic_insert(),
page_cur_tuple_insert(),
page_cur_insert_rec_low(): Add parameters "ext" and "n_ext".

dtuple_convert_big_rec(): Make parameter "ext" const.

btr/btr0btr.c

@@ -1029,6 +1029,8 @@ btr_root_raise_and_insert(
 				the cursor is positioned on the predecessor
 				of the inserted record */
 	dtuple_t*	tuple,	/* in: tuple to insert */
+	const ulint*	ext,	/* in: array of extern field numbers */
+	ulint		n_ext,	/* in: number of elements in vec */
 	mtr_t*		mtr)	/* in: mtr */
 {
 	dict_tree_t*	tree;
@@ -1110,7 +1112,7 @@ btr_root_raise_and_insert(
 	page_cur_set_before_first(root, page_cursor);
 
 	node_ptr_rec = page_cur_tuple_insert(page_cursor, NULL,
-					node_ptr, cursor->index, mtr);
+					node_ptr, cursor->index, NULL, 0, mtr);
 
 	ut_ad(node_ptr_rec);
 
@@ -1146,7 +1148,7 @@ btr_root_raise_and_insert(
 				PAGE_CUR_LE, page_cursor);
 	
 	/* Split the child and insert tuple */
-	return(btr_page_split_and_insert(cursor, tuple, mtr));
+	return(btr_page_split_and_insert(cursor, tuple, ext, n_ext, mtr));
 }	
 
 /*****************************************************************
@@ -1490,7 +1492,8 @@ btr_insert_on_non_leaf_level(
 					| BTR_KEEP_SYS_FLAG
 					| BTR_NO_UNDO_LOG_FLAG,
 					&cursor, tuple,
-					&rec, &dummy_big_rec, NULL, mtr);
+					&rec, &dummy_big_rec,
+					NULL, 0, NULL, mtr);
 	ut_a(err == DB_SUCCESS);
 }
 
@@ -1639,6 +1642,8 @@ btr_page_split_and_insert(
 				function returns, the cursor is positioned
 				on the predecessor of the inserted record */
 	dtuple_t*	tuple,	/* in: tuple to insert */
+	const ulint*	ext,	/* in: array of extern field numbers */
+	ulint		n_ext,	/* in: number of elements in vec */
 	mtr_t*		mtr)	/* in: mtr */
 {
 	dict_tree_t*	tree;
@@ -1809,7 +1814,7 @@ btr_page_split_and_insert(
 						PAGE_CUR_LE, page_cursor);
 
 	rec = page_cur_tuple_insert(page_cursor, insert_page_zip,
-					tuple, cursor->index, mtr);
+					tuple, cursor->index, ext, n_ext, mtr);
 
 	ut_ad(!insert_page_zip
 		|| page_zip_validate(insert_page_zip, insert_page));
@@ -1839,7 +1844,7 @@ btr_page_split_and_insert(
 	page_cur_search(insert_page, cursor->index, tuple,
 						PAGE_CUR_LE, page_cursor);
 	rec = page_cur_tuple_insert(page_cursor, insert_page_zip,
-					tuple, cursor->index, mtr);
+					tuple, cursor->index, ext, n_ext, mtr);
 
 	if (UNIV_UNLIKELY(rec == NULL)) {
 		/* The insert did not fit on the page: loop back to the

btr/btr0cur.c

@@ -819,6 +819,8 @@ btr_cur_insert_if_possible(
 	page_zip_des_t*	page_zip,/* in: compressed page of cursor */
 	dtuple_t*	tuple,	/* in: tuple to insert; the size info need not
 				have been stored to tuple */
+	const ulint*	ext,	/* in: array of extern field numbers */
+	ulint		n_ext,	/* in: number of elements in vec */
 	mtr_t*		mtr)	/* in: mtr */
 {
 	page_cur_t*	page_cursor;
@@ -835,7 +837,7 @@ btr_cur_insert_if_possible(
 	
 	/* Now, try the insert */
 	rec = page_cur_tuple_insert(page_cursor, page_zip,
-					tuple, cursor->index, mtr);
+					tuple, cursor->index, ext, n_ext, mtr);
 
 	if (UNIV_UNLIKELY(!rec)) {
 		/* If record did not fit, reorganize */
@@ -846,7 +848,7 @@ btr_cur_insert_if_possible(
 						PAGE_CUR_LE, page_cursor);
 
 			rec = page_cur_tuple_insert(page_cursor, page_zip,
-						tuple, cursor->index, mtr);
+					tuple, cursor->index, ext, n_ext, mtr);
 		}
 	}
 
@@ -954,6 +956,8 @@ btr_cur_optimistic_insert(
 	big_rec_t**	big_rec,/* out: big rec vector whose fields have to
 				be stored externally by the caller, or
 				NULL */
+	const ulint*	ext,	/* in: array of extern field numbers */
+	ulint		n_ext,	/* in: number of elements in vec */
 	que_thr_t*	thr,	/* in: query thread or NULL */
 	mtr_t*		mtr)	/* in: mtr */
 {
@@ -969,6 +973,7 @@ btr_cur_optimistic_insert(
 	ibool		inherit;
 	ulint		rec_size;
 	ulint		type;
+	mem_heap_t*	heap		= NULL;
 	ulint		err;	
 
 	*big_rec = NULL;
@@ -1003,7 +1008,7 @@ btr_cur_optimistic_insert(
 		/* The record is so big that we have to store some fields
 		externally on separate database pages */
 		
-                big_rec_vec = dtuple_convert_big_rec(index, entry, NULL, 0);
+                big_rec_vec = dtuple_convert_big_rec(index, entry, ext, n_ext);
 
 		if (big_rec_vec == NULL) {
 		
@@ -1059,11 +1064,31 @@ btr_cur_optimistic_insert(
 
 	reorg = FALSE;
 
+	/* Add externally stored records, if needed */
+	if (UNIV_LIKELY_NULL(big_rec_vec)) {
+		ulint	n_more_ext = big_rec_vec->n_fields;
+		ulint*	more_ext;
+		ulint	i;
+
+		heap = mem_heap_create((n_more_ext + n_ext) * sizeof(ulint));
+		more_ext = mem_heap_alloc(heap,
+					(n_more_ext + n_ext) * sizeof(ulint));
+		if (n_ext) {
+			memcpy(more_ext, ext, n_ext * sizeof(ulint));
+		}
+
+		for (i = 0; i < n_more_ext; i++) {
+			more_ext[n_ext++] = big_rec_vec->fields[i].field_no;
+		}
+
+		ext = more_ext;
+	}
+
 	/* Now, try the insert */
 	page_zip = buf_block_get_page_zip(buf_block_align(page));
 
 	*rec = page_cur_insert_rec_low(page_cursor, page_zip,
-					entry, index, NULL, NULL, mtr);
+				entry, index, NULL, NULL, ext, n_ext, mtr);
 	if (UNIV_UNLIKELY(!(*rec))) {
 		/* If the record did not fit, reorganize */
 		if (UNIV_UNLIKELY(!btr_page_reorganize(page, index, mtr))) {
@@ -1079,7 +1104,7 @@ btr_cur_optimistic_insert(
 		page_cur_search(page, index, entry, PAGE_CUR_LE, page_cursor);
 
 		*rec = page_cur_tuple_insert(page_cursor, page_zip,
-						entry, index, mtr);
+						entry, index, ext, n_ext, mtr);
 
 		if (UNIV_UNLIKELY(!*rec)) {
 			fputs("InnoDB: Error: cannot insert tuple ", stderr);
@@ -1092,6 +1117,10 @@ btr_cur_optimistic_insert(
 		}
 	}
 
+	if (UNIV_LIKELY_NULL(heap)) {
+		mem_heap_free(heap);
+	}
+
 #ifdef BTR_CUR_HASH_ADAPT
 	if (!reorg && (0 == level) && (cursor->flag == BTR_CUR_HASH)) {
 		btr_search_update_hash_node_on_insert(cursor);
@@ -1145,6 +1174,8 @@ btr_cur_pessimistic_insert(
 	big_rec_t**	big_rec,/* out: big rec vector whose fields have to
 				be stored externally by the caller, or
 				NULL */
+	const ulint*	ext,	/* in: array of extern field numbers */
+	ulint		n_ext,	/* in: number of elements in vec */
 	que_thr_t*	thr,	/* in: query thread or NULL */
 	mtr_t*		mtr)	/* in: mtr */
 {
@@ -1175,7 +1206,7 @@ btr_cur_pessimistic_insert(
 	cursor->flag = BTR_CUR_BINARY;
 
 	err = btr_cur_optimistic_insert(flags, cursor, entry, rec, big_rec,
-								thr, mtr);
+							ext, n_ext, thr, mtr);
 	if (err != DB_FAIL) {
 
 		return(err);
@@ -1230,9 +1261,11 @@ btr_cur_pessimistic_insert(
 					== buf_frame_get_page_no(page)) {
 
 		/* The page is the root page */
-		*rec = btr_root_raise_and_insert(cursor, entry, mtr);
+		*rec = btr_root_raise_and_insert(cursor, entry,
+						ext, n_ext, mtr);
 	} else {
-		*rec = btr_page_split_and_insert(cursor, entry, mtr);
+		*rec = btr_page_split_and_insert(cursor, entry,
+						ext, n_ext, mtr);
 	}
 
 	btr_cur_position(index, page_rec_get_prev(*rec), cursor);	
@@ -1740,7 +1773,9 @@ btr_cur_optimistic_update(
 								trx->id);
 	}
 
-	rec = btr_cur_insert_if_possible(cursor, page_zip, new_entry, mtr);
+	/* There are no externally stored columns in new_entry */
+	rec = btr_cur_insert_if_possible(cursor, page_zip, new_entry,
+					NULL, 0, mtr);
 	ut_a(rec); /* <- We calculated above the insert would fit */
 
 	if (!rec_get_deleted_flag(rec, page_is_comp(page))) {
@@ -1973,8 +2008,8 @@ btr_cur_pessimistic_update(
 
 	page_cur_move_to_prev(page_cursor);
 
-	/* TODO: set extern flags in new_entry */
-	rec = btr_cur_insert_if_possible(cursor, page_zip, new_entry, mtr);
+	rec = btr_cur_insert_if_possible(cursor, page_zip, new_entry,
+						ext_vect, n_ext_vect, mtr);
 	ut_a(rec || optim_err != DB_UNDERFLOW);
 
 	if (rec) {
@@ -1982,10 +2017,7 @@ btr_cur_pessimistic_update(
 					ULINT_UNDEFINED, &heap);
 
 		lock_rec_restore_from_page_infimum(rec, page);
-		/* TODO: set these before insert */
-		rec_set_field_extern_bits(rec, index,
-						ext_vect, n_ext_vect, mtr);
-		
+
 		if (!rec_get_deleted_flag(rec, rec_offs_comp(offsets))) {
 			/* The new inserted record owns its possible externally
 			stored fields */
@@ -2014,19 +2046,18 @@ btr_cur_pessimistic_update(
 					| BTR_NO_LOCKING_FLAG
 					| BTR_KEEP_SYS_FLAG,
 					cursor, new_entry, &rec,
-					&dummy_big_rec, NULL, mtr);
+					&dummy_big_rec,
+					ext_vect, n_ext_vect, NULL, mtr);
 	ut_a(rec);
 	ut_a(err == DB_SUCCESS);
 	ut_a(dummy_big_rec == NULL);
 
-	/* TODO: set these before insert */
-	rec_set_field_extern_bits(rec, index, ext_vect, n_ext_vect, mtr);
-	offsets = rec_get_offsets(rec, index, offsets, ULINT_UNDEFINED, &heap);
-
 	if (!rec_get_deleted_flag(rec, rec_offs_comp(offsets))) {
 		/* The new inserted record owns its possible externally
 		stored fields */
 
+		offsets = rec_get_offsets(rec, index, offsets,
+					ULINT_UNDEFINED, &heap);
 		btr_cur_unmark_extern_fields(
 				page_zip, rec, index, offsets, mtr);
 	}

data/data0data.c

@@ -481,7 +481,7 @@ dtuple_convert_big_rec(
 				too many short fields in entry */
 	dict_index_t*	index,	/* in: index */
 	dtuple_t*	entry,	/* in: index entry */
-	ulint*		ext_vec,/* in: array of externally stored fields,
+	const ulint*	ext_vec,/* in: array of externally stored fields,
 				or NULL: if a field already is externally
 				stored, then we cannot move it to the vector
 				this function returns */

ibuf/ibuf0ibuf.c

@@ -2660,8 +2660,8 @@ ibuf_insert_low(
 	if (mode == BTR_MODIFY_PREV) {
 		err = btr_cur_optimistic_insert(BTR_NO_LOCKING_FLAG, cursor,
 						ibuf_entry, &ins_rec,
-						&dummy_big_rec, thr,
-						&mtr);
+						&dummy_big_rec, NULL, 0,
+						thr, &mtr);
 		if (err == DB_SUCCESS) {
 			/* Update the page max trx id field */
 			page_update_max_trx_id(buf_frame_align(ins_rec), NULL,
@@ -2681,8 +2681,8 @@ ibuf_insert_low(
 						 | BTR_NO_UNDO_LOG_FLAG,
 						cursor,
 						ibuf_entry, &ins_rec,
-						&dummy_big_rec, thr,
-						&mtr);
+						&dummy_big_rec, NULL, 0,
+						thr, &mtr);
 		if (err == DB_SUCCESS) {
 			/* Update the page max trx id field */
 			page_update_max_trx_id(buf_frame_align(ins_rec), NULL,
@@ -2848,7 +2848,7 @@ ibuf_insert_to_index_page(
 		btr_cur_del_unmark_for_ibuf(rec, mtr);
 	} else {
 		rec = page_cur_tuple_insert(&page_cur, NULL,
-					entry, index, mtr);
+					entry, index, NULL, 0, mtr);
 		
 		if (UNIV_UNLIKELY(rec == NULL)) {
 			/* If the record did not fit, reorganize */
@@ -2861,7 +2861,7 @@ ibuf_insert_to_index_page(
 			/* This time the record must fit */
 			if (UNIV_UNLIKELY(!page_cur_tuple_insert(
 						&page_cur, NULL,
-						entry, index, mtr))) {
+						entry, index, NULL, 0, mtr))) {
 
 				ut_print_timestamp(stderr);
 

include/btr0btr.h

@@ -195,6 +195,8 @@ btr_root_raise_and_insert(
 				the cursor is positioned on the predecessor
 				of the inserted record */
 	dtuple_t*	tuple,	/* in: tuple to insert */
+	const ulint*	ext,	/* in: array of extern field numbers */
+	ulint		n_ext,	/* in: number of elements in vec */
 	mtr_t*		mtr);	/* in: mtr */
 /*****************************************************************
 Reorganizes an index page. */
@@ -249,6 +251,8 @@ btr_page_split_and_insert(
 				function returns, the cursor is positioned
 				on the predecessor of the inserted record */
 	dtuple_t*	tuple,	/* in: tuple to insert */
+	const ulint*	ext,	/* in: array of extern field numbers */
+	ulint		n_ext,	/* in: number of elements in vec */
 	mtr_t*		mtr);	/* in: mtr */
 /***********************************************************
 Inserts a data tuple to a tree on a non-leaf level. It is assumed

include/btr0cur.h

@@ -159,6 +159,8 @@ btr_cur_optimistic_insert(
 	big_rec_t**	big_rec,/* out: big rec vector whose fields have to
 				be stored externally by the caller, or
 				NULL */
+	const ulint*	ext,	/* in: array of extern field numbers */
+	ulint		n_ext,	/* in: number of elements in vec */
 	que_thr_t*	thr,	/* in: query thread or NULL */
 	mtr_t*		mtr);	/* in: mtr */
 /*****************************************************************
@@ -185,6 +187,8 @@ btr_cur_pessimistic_insert(
 	big_rec_t**	big_rec,/* out: big rec vector whose fields have to
 				be stored externally by the caller, or
 				NULL */
+	const ulint*	ext,	/* in: array of extern field numbers */
+	ulint		n_ext,	/* in: number of elements in vec */
 	que_thr_t*	thr,	/* in: query thread or NULL */
 	mtr_t*		mtr);	/* in: mtr */
 /*****************************************************************

include/data0data.h

@@ -343,7 +343,7 @@ dtuple_convert_big_rec(
 				too many short fields in entry */
 	dict_index_t*	index,	/* in: index */
 	dtuple_t*	entry,	/* in: index entry */
-	ulint*		ext_vec,/* in: array of externally stored fields,
+	const ulint*	ext_vec,/* in: array of externally stored fields,
 				or NULL: if a field already is externally
 				stored, then we cannot move it to the vector
 				this function returns */

include/page0cur.h

@@ -133,6 +133,8 @@ page_cur_tuple_insert(
 	page_zip_des_t*	page_zip,/* in/out: compressed page, or NULL */
 	dtuple_t*	tuple,	/* in: pointer to a data tuple */
 	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 */
 /***************************************************************
 Inserts a record next to page cursor. Returns pointer to inserted record if
@@ -167,6 +169,8 @@ page_cur_insert_rec_low(
 	dict_index_t*	index,	/* in: record descriptor */
 	rec_t*		rec,	/* in: pointer to a physical record or NULL */
 	ulint*		offsets,/* in: rec_get_offsets(rec, index) or NULL */
+	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 */
 /*****************************************************************
 Copies records from page to a newly created page, from a given record onward,

include/page0cur.ic

@@ -184,10 +184,12 @@ page_cur_tuple_insert(
 	page_zip_des_t*	page_zip,/* in/out: compressed page, or NULL */
 	dtuple_t*	tuple,  /* in: pointer to a data tuple */
 	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 */
 {
 	return(page_cur_insert_rec_low(cursor, page_zip, tuple,
-					index, NULL, NULL, mtr));
+					index, NULL, NULL, ext, n_ext, mtr));
 }
 
 /***************************************************************
@@ -208,6 +210,6 @@ page_cur_rec_insert(
 	mtr_t*		mtr)	/* in: mini-transaction handle */
 {
 	return(page_cur_insert_rec_low(cursor, page_zip, NULL,
-					index, rec, offsets, mtr));
+					index, rec, offsets, NULL, 0, mtr));
 }
 

page/page0cur.c

@@ -893,6 +893,8 @@ page_cur_insert_rec_low(
 	dict_index_t*	index,	/* in: record descriptor */
 	rec_t*		rec,	/* in: pointer to a physical record or NULL */
 	ulint*		offsets,/* in: rec_get_offsets(rec, index) or NULL */
+	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 */
 {
 	byte*		insert_buf	= NULL;
@@ -952,6 +954,11 @@ page_cur_insert_rec_low(
 
 	ut_ad(insert_rec);
 	ut_ad(rec_size == rec_offs_size(offsets));
+
+	/* Set the "extern storage" flags */
+	if (UNIV_UNLIKELY(n_ext)) {
+		rec_set_field_extern_bits(insert_rec, index, ext, n_ext, mtr);
+	}
 	
 	/* 4. Insert the record in the linked list of records */
 	current_rec = cursor->rec;

row/row0ins.c

@@ -2053,6 +2053,7 @@ row_ins_index_entry_low(
 							ext_vec, n_ext_vec,
 							thr, &mtr);
 		} else {
+			ut_ad(!n_ext_vec);
 			err = row_ins_sec_index_entry_by_modify(mode, &cursor,
 								entry,
 								thr, &mtr);
@@ -2061,27 +2062,20 @@ row_ins_index_entry_low(
 	} else {
 		if (mode == BTR_MODIFY_LEAF) {
 			err = btr_cur_optimistic_insert(0, &cursor, entry,
-					&insert_rec, &big_rec, thr, &mtr);
+					&insert_rec, &big_rec,
+					ext_vec, n_ext_vec, thr, &mtr);
 		} else {
 			ut_a(mode == BTR_MODIFY_TREE);
 			err = btr_cur_pessimistic_insert(0, &cursor, entry,
-					&insert_rec, &big_rec, thr, &mtr);
-		}
-
-		if (err == DB_SUCCESS) {
-			/* TODO: set these before insert */
-			if (ext_vec) {
-				/* TODO: page_zip, mtr=NULL */
-				rec_set_field_extern_bits(insert_rec, index,
-						ext_vec, n_ext_vec, &mtr);
-			}
+					&insert_rec, &big_rec,
+					ext_vec, n_ext_vec, thr, &mtr);
 		}
 	}
 
 function_exit:
 	mtr_commit(&mtr);
 
-	if (big_rec) {
+	if (UNIV_LIKELY_NULL(big_rec)) {
 		rec_t*		rec;
 		mtr_start(&mtr);