branches/zip: In purge, avoid dereferencing unset BLOB pointers of freshly

inserted, uncommitted clustered index records when determining if a
secondary index record that contains a column prefix of an externally
stored column is referencing the clustered index record.

field_ref_zero[]: A BLOB pointer full of zero, for use in comparisons.

btr_copy_externally_stored_field_prefix(): Assert that the BLOB pointer is set.

row_ext_lookup_ith(), row_ext_lookup(), row_ext_lookup_low(): Document
that field_ref_zero is returned when the BLOB cannot be fetched.

row_ext_lookup_low(): Return field_ref_zero and *len = 0 when the
BLOB pointer is unset.

row_build_index_entry(): Return NULL when a needed BLOB pointer cannot
be dereferenced (row_ext_lookup returns field_ref_zero).  Check the
return value for NULL in callers.

row_vers_impl_x_locked_off_kernel(): Avoid comparisons when
row_build_index_entry() returns NULL.

row_vers_old_has_index_entry(): Ignore records for which
row_build_index_entry() returns NULL.  The entry should never be NULL
in rollback, but it may be NULL in purge.

row_merge_buf_add(): Assert that row_ext_lookup() does not return
field_ref_zero.  The table will be locked during index creation.

btr/btr0cur.c

@@ -68,6 +68,11 @@ this many index pages */
 /*--------------------------------------*/
 #define BTR_BLOB_HDR_SIZE		8
 
+/* A BLOB field reference full of zero, for use in assertions and tests.
+Initially, BLOB field references are set to zero, in
+dtuple_convert_big_rec(). */
+const byte field_ref_zero[BTR_EXTERN_FIELD_REF_SIZE];
+
 /***********************************************************************
 Marks all extern fields in a record as owned by the record. This function
 should be called if the delete mark of a record is removed: a not delete
@@ -4528,6 +4533,8 @@ btr_copy_externally_stored_field_prefix(
 	memcpy(buf, data, local_len);
 	data += local_len;
 
+	ut_a(memcmp(data, field_ref_zero, BTR_EXTERN_FIELD_REF_SIZE));
+
 	space_id = mach_read_from_4(data + BTR_EXTERN_SPACE_ID);
 
 	page_no = mach_read_from_4(data + BTR_EXTERN_PAGE_NO);

include/btr0cur.h

@@ -712,6 +712,11 @@ extern ulint	btr_cur_n_sea;
 extern ulint	btr_cur_n_non_sea_old;
 extern ulint	btr_cur_n_sea_old;
 
+/* A BLOB field reference full of zero, for use in assertions and tests.
+Initially, BLOB field references are set to zero, in
+dtuple_convert_big_rec(). */
+extern const byte field_ref_zero[BTR_EXTERN_FIELD_REF_SIZE];
+
 #ifndef UNIV_NONINL
 #include "btr0cur.ic"
 #endif

include/row0ext.h

@@ -35,7 +35,9 @@ byte*
 row_ext_lookup_ith(
 /*===============*/
 				/* out: column prefix, or NULL if
-				the column is not stored externally */
+				the column is not stored externally,
+				or pointer to field_ref_zero
+				if the BLOB pointer is unset */
 	row_ext_t*	ext,	/* in/out: column prefix cache */
 	ulint		i,	/* in: ith element of ext */
 	const byte*	field,	/* in: locally stored part of the column */
@@ -49,7 +51,9 @@ byte*
 row_ext_lookup(
 /*===========*/
 				/* out: column prefix, or NULL if
-				the column is not stored externally */
+				the column is not stored externally,
+				or pointer to field_ref_zero
+				if the BLOB pointer is unset */
 	row_ext_t*	ext,	/* in/out: column prefix cache */
 	ulint		col,	/* in: column number in the InnoDB
 				table object, as reported by

include/row0ext.ic

@@ -14,7 +14,9 @@ Looks up and caches a column prefix of an externally stored column. */
 byte*
 row_ext_lookup_low(
 /*===============*/
-				/* out: column prefix */
+				/* out: column prefix, or
+				pointer to field_ref_zero
+				if the BLOB pointer is unset */
 	row_ext_t*	ext,	/* in/out: column prefix cache */
 	ulint		i,	/* in: index of ext->ext[] */
 	const byte*	field,	/* in: locally stored part of the column */
@@ -60,7 +62,9 @@ byte*
 row_ext_lookup_ith(
 /*===============*/
 				/* out: column prefix, or NULL if
-				the column is not stored externally */
+				the column is not stored externally,
+				or pointer to field_ref_zero
+				if the BLOB pointer is unset */
 	row_ext_t*	ext,	/* in/out: column prefix cache */
 	ulint		i,	/* in: index of ext->ext[] */
 	const byte*	field,	/* in: locally stored part of the column */
@@ -91,7 +95,9 @@ byte*
 row_ext_lookup(
 /*===========*/
 				/* out: column prefix, or NULL if
-				the column is not stored externally */
+				the column is not stored externally,
+				or pointer to field_ref_zero
+				if the BLOB pointer is unset */
 	row_ext_t*	ext,	/* in/out: column prefix cache */
 	ulint		col,	/* in: column number in the InnoDB
 				table object, as reported by

include/row0row.h

@@ -59,7 +59,10 @@ dtuple_t*
 row_build_index_entry(
 /*==================*/
 				/* out: index entry which should be
-				inserted or purged */
+				inserted or purged, or NULL if the
+				externally stored columns in the
+				clustered index record are unavailable
+				and ext != NULL */
 	const dtuple_t*	row,	/* in: row which should be
 				inserted or purged */
 	row_ext_t*	ext,	/* in: externally stored column prefixes,

row/row0ext.c

@@ -20,7 +20,9 @@ Looks up and caches a column prefix of an externally stored column. */
 byte*
 row_ext_lookup_low(
 /*===============*/
-				/* out: column prefix */
+				/* out: column prefix, or
+				pointer to field_ref_zero
+				if the BLOB pointer is unset */
 	row_ext_t*	ext,	/* in/out: column prefix cache */
 	ulint		i,	/* in: index of ext->ext[] */
 	const byte*	field,	/* in: locally stored part of the column */
@@ -31,6 +33,15 @@ row_ext_lookup_low(
 	byte*	buf	= ext->buf + i * REC_MAX_INDEX_COL_LEN;
 
 	ut_ad(i < ext->n_ext);
+	ut_a(f_len >= BTR_EXTERN_FIELD_REF_SIZE);
+
+	if (UNIV_UNLIKELY(!memcmp(field_ref_zero,
+				  field + f_len - BTR_EXTERN_FIELD_REF_SIZE,
+				  BTR_EXTERN_FIELD_REF_SIZE))) {
+		/* The BLOB pointer is not set: we cannot fetch it */
+		*len = 0;
+		return((byte*) field_ref_zero);
+	}
 
 	*len = ext->len[i] = btr_copy_externally_stored_field_prefix(
 		buf,

row/row0merge.c

@@ -281,6 +281,7 @@ row_merge_buf_add(
 			byte*	buf = row_ext_lookup(ext, col_no,
 						     field_data, len, &len);
 			if (UNIV_LIKELY_NULL(buf)) {
+				ut_a(buf != field_ref_zero);
 				if (i < dict_index_get_n_unique(index)) {
 					dfield_set_data(field, buf, len);
 				} else {
@@ -292,6 +293,7 @@ row_merge_buf_add(
 			byte*	buf = row_ext_lookup(ext, col_no,
 						     field_data, len, &len);
 			if (UNIV_LIKELY_NULL(buf)) {
+				ut_a(buf != field_ref_zero);
 				dfield_set_data(field, buf, len);
 			}
 		}

row/row0purge.c

@@ -340,7 +340,7 @@ row_purge_del_mark(
 		entry = row_build_index_entry(node->row,
 					      node->ext,
 					      index, heap);
-
+		ut_a(entry);
 		row_purge_remove_sec_if_poss(node, index, entry);
 
 		node->index = dict_table_get_next_index(node->index);
@@ -387,7 +387,7 @@ row_purge_upd_exist_or_extern(
 			/* Build the older version of the index entry */
 			entry = row_build_index_entry(node->row, node->ext,
 						      index, heap);
-
+			ut_a(entry);
 			row_purge_remove_sec_if_poss(node, index, entry);
 		}
 

row/row0row.c

@@ -67,7 +67,10 @@ dtuple_t*
 row_build_index_entry(
 /*==================*/
 				/* out: index entry which should be
-				inserted or purged */
+				inserted or purged, or NULL if the
+				externally stored columns in the
+				clustered index record are unavailable
+				and ext != NULL */
 	const dtuple_t*	row,	/* in: row which should be
 				inserted or purged */
 	row_ext_t*	ext,	/* in: externally stored column prefixes,
@@ -122,6 +125,9 @@ row_build_index_entry(
 						     dfield_get_data(dfield),
 						     len, &len);
 			if (UNIV_LIKELY_NULL(buf)) {
+				if (UNIV_UNLIKELY(buf == field_ref_zero)) {
+					return(NULL);
+				}
 				dfield_set_data(dfield, buf, len);
 			}
 		}

row/row0uins.c

@@ -283,7 +283,7 @@ row_undo_ins(
 
 		entry = row_build_index_entry(node->row, node->ext,
 					      node->index, node->heap);
-
+		ut_a(entry);
 		err = row_undo_ins_remove_sec(node->index, entry);
 
 		if (err != DB_SUCCESS) {

row/row0umod.c

@@ -515,7 +515,7 @@ row_undo_mod_upd_del_sec(
 
 		entry = row_build_index_entry(node->row, node->ext,
 					      index, heap);
-
+		ut_a(entry);
 		err = row_undo_mod_del_mark_or_remove_sec(node, thr, index,
 							  entry);
 		if (err != DB_SUCCESS) {
@@ -555,7 +555,7 @@ row_undo_mod_del_mark_sec(
 
 		entry = row_build_index_entry(node->row, node->ext,
 					      index, heap);
-
+		ut_a(entry);
 		err = row_undo_mod_del_unmark_sec_and_undo_update(
 			BTR_MODIFY_LEAF, thr, index, entry);
 		if (err == DB_FAIL) {
@@ -610,7 +610,7 @@ row_undo_mod_upd_exist_sec(
 			/* Build the newest version of the index entry */
 			entry = row_build_index_entry(node->row, node->ext,
 						      index, heap);
-
+			ut_a(entry);
 			/* NOTE that if we updated the fields of a
 			delete-marked secondary index record so that
 			alphabetically they stayed the same, e.g.,

row/row0upd.c

@@ -1380,6 +1380,7 @@ row_upd_sec_index_entry(
 
 	/* Build old index entry */
 	entry = row_build_index_entry(node->row, node->ext, index, heap);
+	ut_a(entry);
 
 	log_free_check();
 	mtr_start(&mtr);
@@ -1559,6 +1560,7 @@ row_upd_clust_rec_by_insert(
 	node->state = UPD_NODE_INSERT_CLUSTERED;
 
 	entry = row_build_index_entry(node->row, node->ext, index, heap);
+	ut_a(entry);
 
 	row_upd_index_replace_new_col_vals(entry, index, node->update,
 					   NULL, heap);

row/row0vers.c

@@ -167,6 +167,14 @@ row_vers_impl_x_locked_off_kernel(
 					prev_version, clust_offsets,
 					&ext, heap);
 			entry = row_build_index_entry(row, ext, index, heap);
+			/* entry may be NULL if a record was inserted
+			in place of a deleted record, and the BLOB
+			pointers of the new record were not
+			initialized yet.  But in that case,
+			prev_version should be NULL.
+
+			We will play it safe and avoid dereferencing
+			entry when it is NULL, later in this function. */
 		}
 
 		mutex_enter(&kernel_mutex);
@@ -201,7 +209,7 @@ row_vers_impl_x_locked_off_kernel(
 
 		/* We check if entry and rec are identified in the alphabetical
 		ordering */
-		if (0 == cmp_dtuple_rec(entry, rec, offsets)) {
+		if (entry && 0 == cmp_dtuple_rec(entry, rec, offsets)) {
 			/* The delete marks of rec and prev_version should be
 			equal for rec to be in the state required by
 			prev_version */
@@ -337,13 +345,28 @@ row_vers_old_has_index_entry(
 				rec, clust_offsets, &ext, heap);
 		entry = row_build_index_entry(row, ext, index, heap);
 
+		/* If entry == NULL, the record contains unset BLOB
+		pointers.  This must be a freshly inserted record.  If
+		this is called from
+		row_purge_remove_sec_if_poss_low(), the thread will
+		hold latches on the clustered index and the secondary
+		index.  Because the insert works in three steps:
+
+			(1) insert the record to clustered index
+			(2) store the BLOBs and update BLOB pointers
+			(3) insert records to secondary indexes
+
+		the purge thread can safely ignore freshly inserted
+		records and delete the secondary index record.  The
+		thread that inserted the new record will be inserting
+		the secondary index records. */
+
 		/* NOTE that we cannot do the comparison as binary
 		fields because the row is maybe being modified so that
-		the clustered index record has already been updated
-		to a different binary value in a char field, but the
+		the clustered index record has already been updated to
+		a different binary value in a char field, but the
 		collation identifies the old and new value anyway! */
-
-		if (!dtuple_coll_cmp(ientry, entry)) {
+		if (entry && !dtuple_coll_cmp(ientry, entry)) {
 
 			mem_heap_free(heap);
 
@@ -380,13 +403,19 @@ row_vers_old_has_index_entry(
 					&ext, heap);
 			entry = row_build_index_entry(row, ext, index, heap);
 
+			/* If entry == NULL, the record contains unset
+			BLOB pointers.  This must be a freshly
+			inserted record that we can safely ignore.
+			For the justification, see the comments after
+			the previous row_build_index_entry() call. */
+
 			/* NOTE that we cannot do the comparison as binary
 			fields because maybe the secondary index record has
 			already been updated to a different binary value in
 			a char field, but the collation identifies the old
 			and new value anyway! */
 
-			if (!dtuple_coll_cmp(ientry, entry)) {
+			if (entry && !dtuple_coll_cmp(ientry, entry)) {
 
 				mem_heap_free(heap);