Clean up some SPATIAL INDEX code

Clarify some comments about accessing an externally stored column
on which a spatial index has been defined. Add a TODO comment that
we should actually write the minimum bounding rectangle (MBR) to
the undo log record, so that we can avoid fetching BLOBs and recomputing
MBR.

row_build_spatial_index_key(): Split from row_build_index_entry_low().

1 files changed, 134 insertions, 112 deletions

diff --git a/storage/innobase/row/row0row.cc b/storage/innobase/row/row0row.cc
index 04af5646e6b..ffd8f48503d 100644
--- a/storage/innobase/row/row0row.cc
+++ b/storage/innobase/row/row0row.cc
@@ -46,6 +46,136 @@ Created 4/20/1996 Heikki Tuuri
 #include "gis0geo.h"
 #include "row0mysql.h"
 
+/** Build a spatial index key.
+@param[in]	index	spatial index
+@param[in]	ext	externally stored column prefixes, or NULL
+@param[in,out]	dfield	field of the tuple to be copied
+@param[in]	dfield2	field of the tuple to copy
+@param[in]	flag	ROW_BUILD_NORMAL, ROW_BUILD_FOR_PURGE or
+			ROW_BUILD_FOR_UNDO
+@param[in,out]	heap	memory heap from which the memory
+			of the field entry is allocated.
+@retval false if undo log is logged before spatial index creation. */
+static bool row_build_spatial_index_key(
+	dict_index_t*		index,
+	const row_ext_t*	ext,
+	dfield_t*		dfield,
+	const dfield_t*		dfield2,
+	ulint			flag,
+	mem_heap_t*		heap)
+{
+	double*			mbr;
+
+	dfield_copy(dfield, dfield2);
+	dfield->type.prtype |= DATA_GIS_MBR;
+
+	/* Allocate memory for mbr field */
+	mbr = static_cast<double*>(mem_heap_alloc(heap, DATA_MBR_LEN));
+
+	/* Set mbr field data. */
+	dfield_set_data(dfield, mbr, DATA_MBR_LEN);
+
+	const fil_space_t* space = index->table->space;
+
+	if (UNIV_UNLIKELY(!dfield2->data || !space)) {
+		/* FIXME: dfield contains uninitialized data,
+		but row_build_index_entry_low() will not return NULL.
+		This bug is inherited from MySQL 5.7.5
+		commit b66ad511b61fffe75c58d0a607cdb837c6e6c821. */
+		return true;
+	}
+
+	uchar*	dptr = NULL;
+	ulint	dlen = 0;
+	ulint	flen = 0;
+	double	tmp_mbr[SPDIMS * 2];
+	mem_heap_t*	temp_heap = NULL;
+
+	if (!dfield_is_ext(dfield2)) {
+		dptr = static_cast<uchar*>(dfield_get_data(dfield2));
+		dlen = dfield_get_len(dfield2);
+		goto write_mbr;
+	}
+
+	if (flag == ROW_BUILD_FOR_PURGE) {
+		byte*	ptr = static_cast<byte*>(dfield_get_data(dfield2));
+
+		switch (dfield_get_spatial_status(dfield2)) {
+		case SPATIAL_ONLY:
+			ut_ad(dfield_get_len(dfield2) == DATA_MBR_LEN);
+			break;
+
+		case SPATIAL_MIXED:
+			ptr += dfield_get_len(dfield2);
+			break;
+
+		case SPATIAL_UNKNOWN:
+			ut_ad(0);
+			/* fall through */
+		case SPATIAL_NONE:
+			/* Undo record is logged before
+			spatial index is created.*/
+			return false;
+		}
+
+		memcpy(mbr, ptr, DATA_MBR_LEN);
+		return true;
+	}
+
+	if (flag == ROW_BUILD_FOR_UNDO
+	    && dict_table_has_atomic_blobs(index->table)) {
+		/* For ROW_FORMAT=DYNAMIC or COMPRESSED, a prefix of
+		off-page records is stored in the undo log record (for
+		any column prefix indexes). For SPATIAL INDEX, we
+		must ignore this prefix. The full column value is
+		stored in the BLOB.  For non-spatial index, we would
+		have already fetched a necessary prefix of the BLOB,
+		available in the "ext" parameter.
+
+		Here, for SPATIAL INDEX, we are fetching the full
+		column, which is potentially wasting a lot of I/O,
+		memory, and possibly involving a concurrency problem,
+		similar to ones that existed before the introduction
+		of row_ext_t.
+
+		MDEV-11657 FIXME: write the MBR directly to the undo
+		log record, and avoid recomputing it here! */
+		flen = BTR_EXTERN_FIELD_REF_SIZE;
+		ut_ad(dfield_get_len(dfield2) >= BTR_EXTERN_FIELD_REF_SIZE);
+		dptr = static_cast<byte*>(dfield_get_data(dfield2))
+			+ dfield_get_len(dfield2)
+			- BTR_EXTERN_FIELD_REF_SIZE;
+	} else {
+		flen = dfield_get_len(dfield2);
+		dptr = static_cast<byte*>(dfield_get_data(dfield2));
+	}
+
+	temp_heap = mem_heap_create(1000);
+
+	dptr = btr_copy_externally_stored_field(
+		&dlen, dptr, ext ? ext->page_size : page_size_t(space->flags),
+		flen, temp_heap);
+
+write_mbr:
+	if (dlen <= GEO_DATA_HEADER_SIZE) {
+		for (uint i = 0; i < SPDIMS; i += 2) {
+			tmp_mbr[i] = DBL_MAX;
+			tmp_mbr[i + 1] = -DBL_MAX;
+		}
+	} else {
+		rtree_mbr_from_wkb(dptr + GEO_DATA_HEADER_SIZE,
+				   uint(dlen - GEO_DATA_HEADER_SIZE),
+				   SPDIMS, tmp_mbr);
+	}
+
+	dfield_write_mbr(dfield, tmp_mbr);
+	if (temp_heap) {
+		mem_heap_free(temp_heap);
+	}
+
+	return true;
+}
+
 /*****************************************************************//**
 When an insert or purge to a table is performed, this function builds
 the entry to be inserted into or purged from an index on the table.
@@ -149,119 +279,11 @@ row_build_index_entry_low(
 		/* Special handle spatial index, set the first field
 		which is for store MBR. */
 		if (dict_index_is_spatial(index) && i == 0) {
-			double*			mbr;
-
-			dfield_copy(dfield, dfield2);
-			dfield->type.prtype |= DATA_GIS_MBR;
-
-			/* Allocate memory for mbr field */
-			ulint mbr_len = DATA_MBR_LEN;
-			mbr = static_cast<double*>(mem_heap_alloc(heap, mbr_len));
-
-			/* Set mbr field data. */
-			dfield_set_data(dfield, mbr, mbr_len);
-
-			if (dfield2->data) {
-				uchar*	dptr = NULL;
-				ulint	dlen = 0;
-				ulint	flen = 0;
-				double	tmp_mbr[SPDIMS * 2];
-				mem_heap_t*	temp_heap = NULL;
-
-				if (dfield_is_ext(dfield2)) {
-					if (flag == ROW_BUILD_FOR_PURGE) {
-						byte*	ptr = NULL;
-
-						spatial_status_t spatial_status;
-						spatial_status =
-							dfield_get_spatial_status(
-								dfield2);
-
-						switch (spatial_status) {
-						case SPATIAL_ONLY:
-						ptr = static_cast<byte*>(
-							dfield_get_data(
-								dfield2));
-						ut_ad(dfield_get_len(dfield2)
-						      == DATA_MBR_LEN);
-						break;
-
-						case SPATIAL_MIXED:
-						ptr = static_cast<byte*>(
-							dfield_get_data(
-								dfield2))
-							+ dfield_get_len(
-								dfield2);
-						break;
-
-						case SPATIAL_UNKNOWN:
-							ut_ad(0);
-							/* fall through */
-						case SPATIAL_NONE:
-						/* Undo record is logged before
-						spatial index is created.*/
-						return(NULL);
-						}
-
-						memcpy(mbr, ptr, DATA_MBR_LEN);
-						continue;
-					}
-
-					if (flag == ROW_BUILD_FOR_UNDO
-					    && dict_table_has_atomic_blobs(
-						    index->table)) {
-						/* For build entry for undo, and
-						the table is Barrcuda, we need
-						to skip the prefix data. */
-						flen = BTR_EXTERN_FIELD_REF_SIZE;
-						ut_ad(dfield_get_len(dfield2) >=
-						      BTR_EXTERN_FIELD_REF_SIZE);
-						dptr = static_cast<byte*>(
-							dfield_get_data(dfield2))
-							+ dfield_get_len(dfield2)
-							- BTR_EXTERN_FIELD_REF_SIZE;
-					} else {
-						flen = dfield_get_len(dfield2);
-						dptr = static_cast<byte*>(
-							dfield_get_data(dfield2));
-					}
-
-					temp_heap = mem_heap_create(1000);
-
-					const page_size_t	page_size
-						= (ext != NULL)
-						? ext->page_size
-						: dict_table_page_size(
-							index->table);
-
-					dptr = btr_copy_externally_stored_field(
-						&dlen, dptr,
-						page_size,
-						flen,
-						temp_heap);
-				} else {
-					dptr = static_cast<uchar*>(
-						dfield_get_data(dfield2));
-					dlen = dfield_get_len(dfield2);
-
-				}
-
-				if (dlen <= GEO_DATA_HEADER_SIZE) {
-					for (uint i = 0; i < SPDIMS; ++i) {
-						tmp_mbr[i * 2] = DBL_MAX;
-						tmp_mbr[i * 2 + 1] = -DBL_MAX;
-					}
-				} else {
-					rtree_mbr_from_wkb(dptr + GEO_DATA_HEADER_SIZE,
-							   static_cast<uint>(dlen
-							   - GEO_DATA_HEADER_SIZE),
-							   SPDIMS, tmp_mbr);
-				}
-				dfield_write_mbr(dfield, tmp_mbr);
-				if (temp_heap) {
-					mem_heap_free(temp_heap);
-				}
+			if (!row_build_spatial_index_key(
+				    index, ext, dfield, dfield2, flag, heap)) {
+				return NULL;
 			}
+
 			continue;
 		}