MDEV-25758 InnoDB spatial indexes miss large geometry fields after MDEV-25459

InnoDB should calculate the MBR for the first field of
spatial index and do the comparison with the clustered
index field MBR. Due to MDEV-25459 refactoring, InnoDB
calculate the length of the first field and fails with
too long column error.

3 files changed, 118 insertions, 55 deletions

diff --git a/mysql-test/suite/innodb_gis/r/gis.result b/mysql-test/suite/innodb_gis/r/gis.result
index 33e0c1c0410..3a1a1aa28fd 100644
--- a/mysql-test/suite/innodb_gis/r/gis.result
+++ b/mysql-test/suite/innodb_gis/r/gis.result
@@ -1482,10 +1482,21 @@ FROM buildings, bridges
 WHERE ST_Contains(ST_Buffer(bridges.position, 15.0), buildings.footprint) = 1;
 count(*)
 1
-DROP DATABASE gis_ogs;
 #
 # Bug#13362660 ASSERTION `FIELD_POS < FIELD_COUNT' FAILED. IN PROTOCOL_TEXT::STORE
 #
 SELECT ST_Union('', ''), md5(1);
 ST_Union('', '')	md5(1)
 NULL	c4ca4238a0b923820dcc509a6f75849b
+#
+# MDEV-25758 InnoDB spatial indexes miss large geometry
+#	fields after MDEV-25459
+#
+CREATE TABLE t1(l LINESTRING NOT NULL, SPATIAL INDEX(l))ENGINE=InnoDB ROW_FORMAT=COMPRESSED KEY_BLOCK_SIZE=1;
+SELECT GROUP_CONCAT(CONCAT(seq, ' ', seq) SEPARATOR ',') INTO @g FROM seq_0_to_190;
+INSERT INTO t1 SET l=ST_GeomFromText(CONCAT('LINESTRING(',@g,',0 0)'));
+SELECT COUNT(*) FROM t1 WHERE MBRIntersects(GeomFromText('Polygon((0 0,0 10,10 10,10 0,0 0))'), l);
+COUNT(*)
+1
+DROP TABLE t1;
+DROP DATABASE gis_ogs;
diff --git a/mysql-test/suite/innodb_gis/t/gis.test b/mysql-test/suite/innodb_gis/t/gis.test
index ba1b3b12ec2..acc2b1e36d8 100644
--- a/mysql-test/suite/innodb_gis/t/gis.test
+++ b/mysql-test/suite/innodb_gis/t/gis.test
@@ -2,6 +2,7 @@
 
 --source include/have_innodb.inc
 -- source include/have_geometry.inc
+--source include/have_sequence.inc
 
 SET default_storage_engine=InnoDB;
 
@@ -1422,10 +1423,20 @@ WHERE ST_Contains(ST_Buffer(bridges.position, 15.0), buildings.footprint) = 1;
 #FROM lakes
 #WHERE lakes.name = 'Blue Lake';
 
-DROP DATABASE gis_ogs;
-
 --echo #
 --echo # Bug#13362660 ASSERTION `FIELD_POS < FIELD_COUNT' FAILED. IN PROTOCOL_TEXT::STORE
 --echo #
 
 SELECT ST_Union('', ''), md5(1);
+
+--echo #
+--echo # MDEV-25758 InnoDB spatial indexes miss large geometry
+--echo #	fields after MDEV-25459
+--echo #
+CREATE TABLE t1(l LINESTRING NOT NULL, SPATIAL INDEX(l))ENGINE=InnoDB ROW_FORMAT=COMPRESSED KEY_BLOCK_SIZE=1;
+SELECT GROUP_CONCAT(CONCAT(seq, ' ', seq) SEPARATOR ',') INTO @g FROM seq_0_to_190;
+INSERT INTO t1 SET l=ST_GeomFromText(CONCAT('LINESTRING(',@g,',0 0)'));
+SELECT COUNT(*) FROM t1 WHERE MBRIntersects(GeomFromText('Polygon((0 0,0 10,10 10,10 0,0 0))'), l);
+DROP TABLE t1;
+
+DROP DATABASE gis_ogs;
diff --git a/storage/innobase/row/row0sel.cc b/storage/innobase/row/row0sel.cc
index 86cb6b0ea1c..83bf18ab9b8 100644
--- a/storage/innobase/row/row0sel.cc
+++ b/storage/innobase/row/row0sel.cc
@@ -150,6 +150,75 @@ row_sel_sec_rec_is_for_blob(
 	return(!cmp_data_data(mtype, prtype, buf, len, sec_field, sec_len));
 }
 
+/** Function to read the secondary spatial index, calculate
+the minimum bounding rectangle for clustered index record
+and secondary index record and compare it.
+@param sec_rec		secondary index record
+@param sec_index	spatial secondary index
+@param clust_rec	clustered index record
+@param clust_index	clustered index
+@retval DB_SUCCESS_LOCKED_REC if the secondary record is equal to the
+	corresponding fields in the clustered record, when compared with
+	collation;
+@retval DB_SUCCESS if not equal */
+static
+dberr_t
+row_sel_spatial_sec_rec_is_for_clust_rec(
+  const rec_t *sec_rec, const dict_index_t *sec_index,
+  const rec_t *clust_rec, dict_index_t *clust_index)
+{
+  mem_heap_t *heap= mem_heap_create(256);
+  rec_offs clust_offsets_[REC_OFFS_NORMAL_SIZE];
+  rec_offs *clust_offs= clust_offsets_;
+  ulint clust_len;
+
+  rec_offs_init(clust_offsets_);
+  ulint clust_pos= dict_col_get_clust_pos(
+    dict_index_get_nth_col(sec_index, 0), clust_index);
+  clust_offs= rec_get_offsets(clust_rec, clust_index, clust_offs,
+                              true, clust_pos + 1, &heap);
+  ut_ad(sec_index->n_user_defined_cols == 1);
+  const byte *clust_field= rec_get_nth_field(clust_rec, clust_offs,
+                                             clust_pos, &clust_len);
+  if (clust_len == UNIV_SQL_NULL || clust_len < GEO_DATA_HEADER_SIZE)
+  {
+    ut_ad("corrupted geometry column" == 0);
+err_exit:
+    mem_heap_free(heap);
+    return DB_SUCCESS;
+  }
+
+  /* For externally stored field, we need to get full
+  geo data to generate the MBR for comparing. */
+  if (rec_offs_nth_extern(clust_offs, clust_pos))
+  {
+    clust_field= btr_copy_externally_stored_field(
+      &clust_len, clust_field, dict_table_page_size(sec_index->table),
+      clust_len, heap);
+    if (clust_field == NULL)
+    {
+      ut_ad("corrupted geometry blob" == 0);
+      goto err_exit;
+    }
+  }
+
+  ut_ad(clust_len >= GEO_DATA_HEADER_SIZE);
+  rtr_mbr_t tmp_mbr;
+  rtr_mbr_t sec_mbr;
+
+  rtree_mbr_from_wkb(
+    clust_field + GEO_DATA_HEADER_SIZE,
+    static_cast<uint>(clust_len - GEO_DATA_HEADER_SIZE),
+    SPDIMS, reinterpret_cast<double*>(&tmp_mbr));
+
+  rtr_read_mbr(sec_rec, &sec_mbr);
+
+  mem_heap_free(heap);
+  return MBR_EQUAL_CMP(&sec_mbr, &tmp_mbr)
+         ? DB_SUCCESS_LOCKED_REC
+         : DB_SUCCESS;
+}
+
 /** Returns TRUE if the user-defined column values in a secondary index record
 are alphabetically the same as the corresponding columns in the clustered
 index record.
@@ -177,12 +246,31 @@ row_sel_sec_rec_is_for_clust_rec(
 	dict_index_t*	clust_index,
 	que_thr_t*	thr)
 {
+	if (rec_get_deleted_flag(clust_rec,
+				 dict_table_is_comp(clust_index->table))) {
+		/* In delete-marked records, DB_TRX_ID must
+		always refer to an existing undo log record. */
+		ut_ad(rec_get_trx_id(clust_rec, clust_index));
+
+		/* The clustered index record is delete-marked;
+		it is not visible in the read view.  Besides,
+		if there are any externally stored columns,
+		some of them may have already been purged. */
+		return DB_SUCCESS;
+	}
+
+	if (sec_index->is_spatial()) {
+		return row_sel_spatial_sec_rec_is_for_clust_rec(
+				sec_rec, sec_index, clust_rec,
+				clust_index);
+	}
+
 	const byte*	sec_field;
 	ulint		sec_len;
 	const byte*	clust_field;
 	ulint		n;
 	ulint		i;
-	mem_heap_t*	heap		= NULL;
+	mem_heap_t*	heap		= mem_heap_create(256);
 	rec_offs	clust_offsets_[REC_OFFS_NORMAL_SIZE];
 	rec_offs	sec_offsets_[REC_OFFS_SMALL_SIZE];
 	rec_offs*	clust_offs	= clust_offsets_;
@@ -191,20 +279,7 @@ row_sel_sec_rec_is_for_clust_rec(
 	rec_offs_init(clust_offsets_);
 	rec_offs_init(sec_offsets_);
 
-	if (rec_get_deleted_flag(clust_rec,
-				 dict_table_is_comp(clust_index->table))) {
-		/* In delete-marked records, DB_TRX_ID must
-		always refer to an existing undo log record. */
-		ut_ad(rec_get_trx_id(clust_rec, clust_index));
-
-		/* The clustered index record is delete-marked;
-		it is not visible in the read view.  Besides,
-		if there are any externally stored columns,
-		some of them may have already been purged. */
-		return DB_SUCCESS;
-	}
 
-	heap = mem_heap_create(256);
 	ib_vcol_row vc(heap);
 
 	clust_offs = rec_get_offsets(clust_rec, clust_index, clust_offs,
@@ -310,44 +385,10 @@ check_for_blob:
 			}
 		}
 
-		/* For spatial index, the first field is MBR, we check
-		if the MBR is equal or not. */
-		if (dict_index_is_spatial(sec_index) && i == 0) {
-			rtr_mbr_t	tmp_mbr;
-			rtr_mbr_t	sec_mbr;
-			byte*		dptr =
-				const_cast<byte*>(clust_field);
-
-			ut_ad(clust_len != UNIV_SQL_NULL);
-
-			/* For externally stored field, we need to get full
-			geo data to generate the MBR for comparing. */
-			if (rec_offs_nth_extern(clust_offs, clust_pos)) {
-				dptr = btr_copy_externally_stored_field(
-					&clust_len, dptr,
-					dict_tf_get_page_size(
-						sec_index->table->flags),
-					len, heap);
-			}
-
-			rtree_mbr_from_wkb(dptr + GEO_DATA_HEADER_SIZE,
-					   static_cast<uint>(clust_len
-					   - GEO_DATA_HEADER_SIZE),
-					   SPDIMS,
-					   reinterpret_cast<double*>(
-						&tmp_mbr));
-			rtr_read_mbr(sec_field, &sec_mbr);
-
-			if (!MBR_EQUAL_CMP(&sec_mbr, &tmp_mbr)) {
-				return DB_SUCCESS;
-			}
-		} else {
-
-			if (0 != cmp_data_data(col->mtype, col->prtype,
-					       clust_field, len,
-					       sec_field, sec_len)) {
-				return DB_SUCCESS;
-			}
+		if (0 != cmp_data_data(col->mtype, col->prtype,
+				       clust_field, len,
+				       sec_field, sec_len)) {
+			return DB_SUCCESS;
 		}
 	}