branches/zip: Allow the locally stored part of externally stored columns

in the clustered index to be smaller than the indexed prefix in secondary
indexes.

row_ext_lookup(): Return NULL if the column is not stored externally.

trx_undo_rec_get_partial_row(): row_build(): Add parameter row_ext_t** ext.

row_build_index_entry(): Add the parameter row_ext_t* ext.
Invoke row_ext_lookup() to fetch prefixes of externally stored columns.

upd_node_t, undo_node_t, purge_node_t: Add the field row_ext_t* ext.

include/row0ext.h

@@ -31,7 +31,8 @@ UNIV_INLINE
 byte*
 row_ext_lookup(
 /*===========*/
-				/* out: column prefix */
+				/* out: column prefix, or NULL if
+				the column is not stored externally */
 	row_ext_t*	ext,	/* in/out: column prefix cache */
 	ulint		col,	/* in: column number */
 	const byte*	field,	/* in: locally stored part of the column */

include/row0ext.ic

@@ -54,7 +54,8 @@ UNIV_INLINE
 byte*
 row_ext_lookup(
 /*===========*/
-				/* out: column prefix */
+				/* out: column prefix, or NULL if
+				the column is not stored externally */
 	row_ext_t*	ext,	/* in/out: column prefix cache */
 	ulint		col,	/* in: column number */
 	const byte*	field,	/* in: locally stored part of the column */
@@ -82,7 +83,5 @@ row_ext_lookup(
 		}
 	}
 
-	/* The column should always be found. */
-	ut_error;
 	return(NULL);
 }

include/row0purge.h

@@ -65,6 +65,8 @@ struct purge_node_struct{
 	dtuple_t*	row;	/* NULL, or a copy (also fields copied to
 				heap) of the indexed fields of the row to
 				handle */
+	row_ext_t*	ext;	/* NULL, or prefixes of the externally
+				stored columns of the row */
 	dict_index_t*	index;	/* NULL, or the next index whose record should
 				be handled */
 	mem_heap_t*	heap;	/* memory heap used as auxiliary storage for

include/row0row.h

@@ -17,6 +17,7 @@ Created 4/20/1996 Heikki Tuuri
 #include "mtr0mtr.h"
 #include "rem0types.h"
 #include "read0types.h"
+#include "row0types.h"
 #include "btr0types.h"
 
 /*************************************************************************
@@ -60,6 +61,8 @@ row_build_index_entry(
 				/* out: index entry which should be inserted */
 	dtuple_t*	row,	/* in: row which should be inserted to the
 				table */
+	row_ext_t*	ext,	/* in: externally stored column prefixes,
+				or NULL */
 	dict_index_t*	index,	/* in: index on the table */
 	mem_heap_t*	heap);	/* in: memory heap from which the memory for
 				the index entry is allocated */
@@ -87,6 +90,8 @@ row_build(
 	const ulint*	offsets,/* in: rec_get_offsets(rec, index)
 				or NULL, in which case this function
 				will invoke rec_get_offsets() */
+	row_ext_t**	ext,	/* out, own: cache of externally stored
+				column prefixes, or NULL */
 	mem_heap_t*	heap);	/* in: memory heap from which the memory
 				needed is allocated */
 /***********************************************************************

include/row0undo.h

@@ -91,6 +91,8 @@ struct undo_node_struct{
 	dtuple_t*	ref;	/* row reference to the next row to handle */
 	dtuple_t*	row;	/* a copy (also fields copied to heap) of the
 				row to handle */
+	row_ext_t*	ext;	/* NULL, or prefixes of the externally
+				stored columns of the row */
 	dict_index_t*	index;	/* the next index whose record should be
 				handled */
 	mem_heap_t*	heap;	/* memory heap used as auxiliary storage for

include/row0upd.h

@@ -388,6 +388,8 @@ struct upd_node_struct{
 	dtuple_t*	row;	/* NULL, or a copy (also fields copied to
 				heap) of the row to update; this must be reset
 				to NULL after a successful update */
+	row_ext_t*	ext;	/* NULL, or prefixes of the externally
+				stored columns of the row */
 	ulint*		ext_vec;/* array describing which fields are stored
 				externally in the clustered index record of
 				row */

include/trx0rec.h

@@ -167,6 +167,8 @@ trx_undo_rec_get_partial_row(
 				record! */
 	dict_index_t*	index,	/* in: clustered index */
 	dtuple_t**	row,	/* out, own: partial row */
+	row_ext_t**	ext,	/* out, own: prefix cache for
+				externally stored columns */
 	mem_heap_t*	heap);	/* in: memory heap from which the memory
 				needed is allocated */
 /***************************************************************************

row/row0ins.c

@@ -119,7 +119,7 @@ ins_node_create_entry_list(
 	index = dict_table_get_first_index(node->table);
 
 	while (index != NULL) {
-		entry = row_build_index_entry(node->row, index,
+		entry = row_build_index_entry(node->row, NULL, index,
 					      node->entry_sys_heap);
 		UT_LIST_ADD_LAST(tuple_list, node->entry_list, entry);
 

row/row0purge.c

@@ -337,7 +337,9 @@ row_purge_del_mark(
 		index = node->index;
 
 		/* Build the index entry */
-		entry = row_build_index_entry(node->row, index, heap);
+		entry = row_build_index_entry(node->row,
+					      node->ext,
+					      index, heap);
 
 		row_purge_remove_sec_if_poss(node, index, entry);
 
@@ -384,7 +386,8 @@ row_purge_upd_exist_or_extern(
 		if (row_upd_changes_ord_field_binary(NULL, node->index,
 						     node->update)) {
 			/* Build the older version of the index entry */
-			entry = row_build_index_entry(node->row, index, heap);
+			entry = row_build_index_entry(node->row, node->ext,
+						      index, heap);
 
 			row_purge_remove_sec_if_poss(node, index, entry);
 		}
@@ -557,9 +560,11 @@ row_purge_parse_undo_rec(
 
 	/* Read to the partial row the fields that occur in indexes */
 
-	if (!cmpl_info & UPD_NODE_NO_ORD_CHANGE) {
+	if (!(cmpl_info & UPD_NODE_NO_ORD_CHANGE)) {
 		ptr = trx_undo_rec_get_partial_row(ptr, clust_index,
-						   &(node->row), node->heap);
+						   &node->row,
+						   &node->ext,
+						   node->heap);
 	}
 
 	return(TRUE);

row/row0row.c

@@ -22,7 +22,7 @@ Created 4/20/1996 Heikki Tuuri
 #include "trx0purge.h"
 #include "trx0rec.h"
 #include "que0que.h"
-#include "row0row.h"
+#include "row0ext.h"
 #include "row0upd.h"
 #include "rem0cmp.h"
 #include "read0read.h"
@@ -65,6 +65,8 @@ row_build_index_entry(
 				/* out: index entry which should be inserted */
 	dtuple_t*	row,	/* in: row which should be inserted to the
 				table */
+	row_ext_t*	ext,	/* in: externally stored column prefixes,
+				or NULL */
 	dict_index_t*	index,	/* in: index on the table */
 	mem_heap_t*	heap)	/* in: memory heap from which the memory for
 				the index entry is allocated */
@@ -92,15 +94,29 @@ row_build_index_entry(
 
 	for (i = 0; i < entry_len; i++) {
 		const dict_col_t*	col;
+		ulint			col_no;
 		ind_field = dict_index_get_nth_field(index, i);
 		col = ind_field->col;
+		col_no = dict_col_get_no(col);
 
 		dfield = dtuple_get_nth_field(entry, i);
 
-		dfield2 = dtuple_get_nth_field(row, dict_col_get_no(col));
+		dfield2 = dtuple_get_nth_field(row, col_no);
 
 		dfield_copy(dfield, dfield2);
 
+		if (UNIV_LIKELY_NULL(ext)
+		    && dfield_get_len(dfield2) != UNIV_SQL_NULL) {
+			/* See if the column is stored externally. */
+			byte*	buf = row_ext_lookup(ext, col_no,
+						     dfield2->data,
+						     dfield2->len,
+						     &dfield->len);
+			if (UNIV_LIKELY_NULL(buf)) {
+				dfield->data = buf;
+			}
+		}
+
 		/* If a column prefix index, take only the prefix */
 		if (ind_field->prefix_len) {
 			if (dfield_get_len(dfield2) != UNIV_SQL_NULL) {
@@ -148,19 +164,21 @@ row_build(
 	const ulint*	offsets,/* in: rec_get_offsets(rec, index)
 				or NULL, in which case this function
 				will invoke rec_get_offsets() */
+	row_ext_t**	ext,	/* out, own: cache of externally stored
+				column prefixes, or NULL */
 	mem_heap_t*	heap)	/* in: memory heap from which the memory
 				needed is allocated */
 {
 	dtuple_t*	row;
 	dict_table_t*	table;
-	dict_field_t*	ind_field;
-	dfield_t*	dfield;
 	ulint		n_fields;
-	byte*		field;
+	ulint		n_ext_cols;
+	ulint*		ext_cols	= NULL; /* remove bogus warning */
 	ulint		len;
 	ulint		row_len;
 	byte*		buf;
 	ulint		i;
+	ulint		j;
 	mem_heap_t*	tmp_heap	= NULL;
 	ulint		offsets_[REC_OFFS_NORMAL_SIZE];
 	*offsets_ = (sizeof offsets_) / sizeof *offsets_;
@@ -192,27 +210,45 @@ row_build(
 				     rec, dict_table_is_comp(table)));
 
 	n_fields = rec_offs_n_fields(offsets);
+	n_ext_cols = rec_offs_n_extern(offsets);
+	if (n_ext_cols) {
+		ext_cols = mem_heap_alloc(heap, n_ext_cols * sizeof *ext_cols);
+	}
 
 	dict_table_copy_types(row, table);
 
-	for (i = 0; i < n_fields; i++) {
-		ind_field = dict_index_get_nth_field(index, i);
-
-		if (ind_field->prefix_len == 0) {
-
+	for (i = j = 0; i < n_fields; i++) {
+		dict_field_t*		ind_field
+			= dict_index_get_nth_field(index, i);
 		const dict_col_t*	col
 			= dict_field_get_col(ind_field);
 
-			dfield = dtuple_get_nth_field(row,
-						      dict_col_get_no(col));
-			field = rec_get_nth_field(rec, offsets, i, &len);
+		if (ind_field->prefix_len == 0) {
+
+			dfield_t*	dfield = dtuple_get_nth_field(
+				row, dict_col_get_no(col));
+			byte*		field = rec_get_nth_field(
+				rec, offsets, i, &len);
 
 			dfield_set_data(dfield, field, len);
 		}
+
+		if (rec_offs_nth_extern(offsets, i)) {
+			ext_cols[j++] = dict_col_get_no(col);
+		}
 	}
 
+	ut_ad(j == n_ext_cols);
 	ut_ad(dtuple_check_typed(row));
 
+	if (n_ext_cols) {
+		*ext = row_ext_create(n_ext_cols, ext_cols,
+				      dict_table_zip_size(index->table),
+				      heap);
+	} else {
+		*ext = NULL;
+	}
+
 	if (tmp_heap) {
 		mem_heap_free(tmp_heap);
 	}

row/row0uins.c

@@ -290,8 +290,8 @@ row_undo_ins(
 		dict_table_get_first_index(node->table));
 
 	while (node->index != NULL) {
-		entry = row_build_index_entry(node->row, node->index,
-					      node->heap);
+		entry = row_build_index_entry(node->row, node->ext,
+					      node->index, node->heap);
 		err = row_undo_ins_remove_sec(node->index, entry);
 
 		if (err != DB_SUCCESS) {

row/row0umod.c

@@ -500,7 +500,8 @@ row_undo_mod_upd_del_sec(
 	while (node->index != NULL) {
 		index = node->index;
 
-		entry = row_build_index_entry(node->row, index, heap);
+		entry = row_build_index_entry(node->row, node->ext,
+					      index, heap);
 
 		err = row_undo_mod_del_mark_or_remove_sec(node, thr, index,
 							  entry);
@@ -539,7 +540,8 @@ row_undo_mod_del_mark_sec(
 	while (node->index != NULL) {
 		index = node->index;
 
-		entry = row_build_index_entry(node->row, index, heap);
+		entry = row_build_index_entry(node->row, node->ext,
+					      index, heap);
 
 		err = row_undo_mod_del_unmark_sec_and_undo_update(
 			BTR_MODIFY_LEAF, thr, index, entry);
@@ -593,7 +595,8 @@ row_undo_mod_upd_exist_sec(
 						     node->update)) {
 
 			/* Build the newest version of the index entry */
-			entry = row_build_index_entry(node->row, index, heap);
+			entry = row_build_index_entry(node->row, node->ext,
+						      index, heap);
 
 			/* NOTE that if we updated the fields of a
 			delete-marked secondary index record so that

row/row0undo.c

@@ -183,7 +183,7 @@ row_undo_search_clust_to_pcur(
 		ret = FALSE;
 	} else {
 		node->row = row_build(ROW_COPY_DATA, clust_index, rec,
-				      offsets, node->heap);
+				      offsets, &node->ext, node->heap);
 		btr_pcur_store_position(&(node->pcur), &mtr);
 
 		ret = TRUE;

row/row0upd.c

@@ -20,6 +20,7 @@ Created 12/27/1996 Heikki Tuuri
 #include "btr0btr.h"
 #include "btr0cur.h"
 #include "que0que.h"
+#include "row0ext.h"
 #include "row0ins.h"
 #include "row0sel.h"
 #include "row0row.h"
@@ -277,6 +278,7 @@ upd_node_create(
 	node->in_mysql_interface = FALSE;
 
 	node->row = NULL;
+	node->ext = NULL;
 	node->ext_vec = NULL;
 	node->index = NULL;
 	node->update = NULL;
@@ -1251,7 +1253,6 @@ row_upd_store_row(
 	upd_node_t*	node)	/* in: row update node */
 {
 	dict_index_t*	clust_index;
-	upd_t*		update;
 	rec_t*		rec;
 	mem_heap_t*	heap		= NULL;
 	ulint		offsets_[REC_OFFS_NORMAL_SIZE];
@@ -1262,7 +1263,6 @@ row_upd_store_row(
 
 	if (node->row != NULL) {
 		mem_heap_empty(node->heap);
-		node->row = NULL;
 	}
 
 	clust_index = dict_table_get_first_index(node->table);
@@ -1272,17 +1272,18 @@ row_upd_store_row(
 	offsets = rec_get_offsets(rec, clust_index, offsets_,
 				  ULINT_UNDEFINED, &heap);
 	node->row = row_build(ROW_COPY_DATA, clust_index, rec, offsets,
-			      node->heap);
+			      &node->ext, node->heap);
+	if (UNIV_LIKELY_NULL(node->ext)) {
 		node->ext_vec = mem_heap_alloc(node->heap, sizeof(ulint)
-				       * rec_offs_n_fields(offsets));
-	if (node->is_delete) {
-		update = NULL;
+					       * node->ext->n_ext);
+		node->n_ext_vec = btr_push_update_extern_fields(
+			node->ext_vec, offsets,
+			node->is_delete ? NULL : node->update);
 	} else {
-		update = node->update;
+		node->ext_vec = NULL;
+		node->n_ext_vec = 0;
 	}
 
-	node->n_ext_vec = btr_push_update_extern_fields(node->ext_vec,
-							offsets, update);
 	if (UNIV_LIKELY_NULL(heap)) {
 		mem_heap_free(heap);
 	}
@@ -1318,7 +1319,7 @@ row_upd_sec_index_entry(
 	heap = mem_heap_create(1024);
 
 	/* Build old index entry */
-	entry = row_build_index_entry(node->row, index, heap);
+	entry = row_build_index_entry(node->row, node->ext, index, heap);
 
 	log_free_check();
 	mtr_start(&mtr);
@@ -1495,7 +1496,7 @@ row_upd_clust_rec_by_insert(
 	}
 	node->state = UPD_NODE_INSERT_CLUSTERED;
 
-	entry = row_build_index_entry(node->row, index, heap);
+	entry = row_build_index_entry(node->row, node->ext, index, heap);
 
 	row_upd_index_replace_new_col_vals(entry, index, node->update, NULL);
 
@@ -1910,6 +1911,8 @@ row_upd(
 
 		if (node->row != NULL) {
 			node->row = NULL;
+			node->ext = NULL;
+			node->ext_vec = NULL;
 			node->n_ext_vec = 0;
 			mem_heap_empty(node->heap);
 		}

row/row0vers.c

@@ -158,12 +158,15 @@ row_vers_impl_x_locked_off_kernel(
 		mem_heap_free(heap2); /* free version and clust_offsets */
 
 		if (prev_version) {
+			row_ext_t*	ext;
+
 			clust_offsets = rec_get_offsets(
 				prev_version, clust_index, NULL,
 				ULINT_UNDEFINED, &heap);
 			row = row_build(ROW_COPY_POINTERS, clust_index,
-					prev_version, clust_offsets, heap);
-			entry = row_build_index_entry(row, index, heap);
+					prev_version, clust_offsets,
+					&ext, heap);
+			entry = row_build_index_entry(row, ext, index, heap);
 		}
 
 		mutex_enter(&kernel_mutex);
@@ -329,9 +332,11 @@ row_vers_old_has_index_entry(
 					ULINT_UNDEFINED, &heap);
 
 	if (also_curr && !rec_get_deleted_flag(rec, comp)) {
+		row_ext_t*	ext;
+
 		row = row_build(ROW_COPY_POINTERS, clust_index,
-				rec, clust_offsets, heap);
-		entry = row_build_index_entry(row, index, heap);
+				rec, clust_offsets, &ext, heap);
+		entry = row_build_index_entry(row, ext, index, heap);
 
 		/* NOTE that we cannot do the comparison as binary
 		fields because the row is maybe being modified so that
@@ -369,9 +374,12 @@ row_vers_old_has_index_entry(
 						NULL, ULINT_UNDEFINED, &heap);
 
 		if (!rec_get_deleted_flag(prev_version, comp)) {
+			row_ext_t*	ext;
+
 			row = row_build(ROW_COPY_POINTERS, clust_index,
-					prev_version, clust_offsets, heap);
-			entry = row_build_index_entry(row, index, heap);
+					prev_version, clust_offsets,
+					&ext, heap);
+			entry = row_build_index_entry(row, ext, index, heap);
 
 			/* NOTE that we cannot do the comparison as binary
 			fields because maybe the secondary index record has

trx/trx0rec.c

@@ -19,6 +19,7 @@ Created 3/26/1996 Heikki Tuuri
 #include "trx0undo.h"
 #include "dict0dict.h"
 #include "ut0mem.h"
+#include "row0ext.h"
 #include "row0upd.h"
 #include "que0que.h"
 #include "trx0purge.h"
@@ -895,38 +896,35 @@ trx_undo_rec_get_partial_row(
 				record! */
 	dict_index_t*	index,	/* in: clustered index */
 	dtuple_t**	row,	/* out, own: partial row */
+	row_ext_t**	ext,	/* out, own: prefix cache for
+				externally stored columns */
 	mem_heap_t*	heap)	/* in: memory heap from which the memory
 				needed is allocated */
 {
-	dfield_t*	dfield;
-	byte*		field;
-	ulint		len;
-	ulint		field_no;
-	ulint		col_no;
+	const byte*	end_ptr;
 	ulint		row_len;
-	ulint		total_len;
-	byte*		start_ptr;
-	ulint		i;
+	ulint		n_ext_cols;
+	ulint*		ext_cols;
 
-	ut_ad(index && ptr && row && heap);
+	ut_ad(index && ptr && row && ext && heap);
 
 	row_len = dict_table_get_n_cols(index->table);
+	n_ext_cols = 0;
+	ext_cols = mem_heap_alloc(heap, row_len * sizeof *ext_cols);
 
 	*row = dtuple_create(heap, row_len);
 
 	dict_table_copy_types(*row, index->table);
 
-	start_ptr = ptr;
-
-	total_len = mach_read_from_2(ptr);
+	end_ptr = ptr + mach_read_from_2(ptr);
 	ptr += 2;
 
-	for (i = 0;; i++) {
-
-		if (ptr == start_ptr + total_len) {
-
-			break;
-		}
+	while (ptr != end_ptr) {
+		dfield_t*	dfield;
+		byte*		field;
+		ulint		field_no;
+		ulint		col_no;
+		ulint		len;
 
 		ptr = trx_undo_update_rec_get_field_no(ptr, &field_no);
 
@@ -934,11 +932,23 @@ trx_undo_rec_get_partial_row(
 
 		ptr = trx_undo_rec_get_col_val(ptr, &field, &len);
 
+		if (len >= UNIV_EXTERN_STORAGE_FIELD) {
+			ext_cols[n_ext_cols++] = col_no;
+		}
+
 		dfield = dtuple_get_nth_field(*row, col_no);
 
 		dfield_set_data(dfield, field, len);
 	}
 
+	if (n_ext_cols) {
+		*ext = row_ext_create(n_ext_cols, ext_cols,
+				      dict_table_zip_size(index->table),
+				      heap);
+	} else {
+		*ext = NULL;
+	}
+
 	return(ptr);
 }