MDEV-25459 MVCC read from index on CHAR or VARCHAR wrongly omits rowsbb-10.2-MDEV-25459

row_sel_sec_rec_is_for_clust_rec(): If the field in the
clustered index record stored off page, always fetch it,
also when the secondary index field has been built on the
entire column. This was broken ever since the InnoDB Plugin
for MySQL Server 5.1 introduced ROW_FORMAT=DYNAMIC and
ROW_FORMAT=COMPRESSED for InnoDB tables. That code was first
introduced in this tree in
commit 3945d5e5549187a18c64a112899f90a7f6a320d6.

For the original ROW_FORMAT=REDUNDANT and the MySQL 5.0.3
ROW_FORMAT=COMPRESSED, there was no problem, because for
those tables we always stored at least a 768-byte prefix of
each column in the clustered index record.

row_sel_sec_rec_is_for_blob(): Allow prefix_len==0 for matching
the full column.

3 files changed, 105 insertions, 35 deletions

diff --git a/mysql-test/suite/innodb/r/mvcc_secondary.result b/mysql-test/suite/innodb/r/mvcc_secondary.result
new file mode 100644
index 00000000000..2289742e830
--- /dev/null
+++ b/mysql-test/suite/innodb/r/mvcc_secondary.result
@@ -0,0 +1,24 @@
+#
+# MDEV-25459 MVCC read from index on CHAR or VARCHAR wrongly omits rows
+#
+CREATE TABLE t1 (
+pk int PRIMARY KEY, c varchar(255) UNIQUE,
+d char(255), e varchar(255), f char(255), g char(255)
+) ENGINE=InnoDB ROW_FORMAT=DYNAMIC DEFAULT CHARACTER SET ucs2;
+INSERT INTO t1 VALUES
+(1,REPEAT('c',248),REPEAT('a',106),REPEAT('b',220),REPEAT('x',14),'');
+BEGIN;
+UPDATE t1 SET c=REPEAT('d',170);
+connect  con1,localhost,root,,;
+SELECT pk FROM t1 FORCE INDEX (c);
+pk
+1
+connection default;
+COMMIT;
+connection con1;
+SELECT pk FROM t1 FORCE INDEX (c);
+pk
+1
+disconnect con1;
+connection default;
+DROP TABLE t1;
diff --git a/mysql-test/suite/innodb/t/mvcc_secondary.test b/mysql-test/suite/innodb/t/mvcc_secondary.test
new file mode 100644
index 00000000000..93c91c40076
--- /dev/null
+++ b/mysql-test/suite/innodb/t/mvcc_secondary.test
@@ -0,0 +1,26 @@
+--source include/innodb_page_size_small.inc
+
+--echo #
+--echo # MDEV-25459 MVCC read from index on CHAR or VARCHAR wrongly omits rows
+--echo #
+
+CREATE TABLE t1 (
+  pk int PRIMARY KEY, c varchar(255) UNIQUE,
+  d char(255), e varchar(255), f char(255), g char(255)
+) ENGINE=InnoDB ROW_FORMAT=DYNAMIC DEFAULT CHARACTER SET ucs2;
+
+INSERT INTO t1 VALUES
+(1,REPEAT('c',248),REPEAT('a',106),REPEAT('b',220),REPEAT('x',14),'');
+
+BEGIN;
+UPDATE t1 SET c=REPEAT('d',170);
+
+connect (con1,localhost,root,,);
+SELECT pk FROM t1 FORCE INDEX (c);
+connection default;
+COMMIT;
+connection con1;
+SELECT pk FROM t1 FORCE INDEX (c);
+disconnect con1;
+connection default;
+DROP TABLE t1;
diff --git a/storage/innobase/row/row0sel.cc b/storage/innobase/row/row0sel.cc
index 864f7bd54ab..86cb6b0ea1c 100644
--- a/storage/innobase/row/row0sel.cc
+++ b/storage/innobase/row/row0sel.cc
@@ -79,9 +79,9 @@ is alphabetically the same as the corresponding BLOB column in the clustered
 index record.
 NOTE: the comparison is NOT done as a binary comparison, but character
 fields are compared with collation!
-@return TRUE if the columns are equal */
+@return whether the columns are equal */
 static
-ibool
+bool
 row_sel_sec_rec_is_for_blob(
 /*========================*/
 	ulint		mtype,		/*!< in: main type */
@@ -100,19 +100,18 @@ row_sel_sec_rec_is_for_blob(
 	const byte*	sec_field,	/*!< in: column in secondary index */
 	ulint		sec_len,	/*!< in: length of sec_field */
 	ulint		prefix_len,	/*!< in: index column prefix length
-					in bytes */
+					in bytes, or 0 for full column */
 	dict_table_t*	table)		/*!< in: table */
 {
 	ulint	len;
-	byte	buf[REC_VERSION_56_MAX_INDEX_COL_LEN];
+	byte	buf[REC_VERSION_56_MAX_INDEX_COL_LEN + 1];
 
 	/* This function should never be invoked on an Antelope format
 	table, because they should always contain enough prefix in the
 	clustered index record. */
 	ut_ad(dict_table_get_format(table) >= UNIV_FORMAT_B);
 	ut_a(clust_len >= BTR_EXTERN_FIELD_REF_SIZE);
-	ut_ad(prefix_len >= sec_len);
-	ut_ad(prefix_len > 0);
+	ut_ad(!prefix_len || prefix_len >= sec_len);
 	ut_a(prefix_len <= sizeof buf);
 
 	if (!memcmp(clust_field + clust_len - BTR_EXTERN_FIELD_REF_SIZE,
@@ -121,11 +120,12 @@ row_sel_sec_rec_is_for_blob(
 		This record should only be seen by
 		recv_recovery_rollback_active() or any
 		TRX_ISO_READ_UNCOMMITTED transactions. */
-		return(FALSE);
+		return false;
 	}
 
 	len = btr_copy_externally_stored_field_prefix(
-		buf, prefix_len, dict_tf_get_page_size(table->flags),
+		buf, prefix_len ? prefix_len : sizeof buf,
+		dict_tf_get_page_size(table->flags),
 		clust_field, clust_len);
 
 	if (len == 0) {
@@ -134,11 +134,18 @@ row_sel_sec_rec_is_for_blob(
 		referring to this clustered index record, because
 		btr_free_externally_stored_field() is called after all
 		secondary index entries of the row have been purged. */
-		return(FALSE);
+		return false;
 	}
 
-	len = dtype_get_at_most_n_mbchars(prtype, mbminlen, mbmaxlen,
-					  prefix_len, len, (const char*) buf);
+	if (prefix_len) {
+		len = dtype_get_at_most_n_mbchars(prtype, mbminlen, mbmaxlen,
+						  prefix_len, len,
+						  reinterpret_cast<const char*>
+						  (buf));
+	} else if (len >= sizeof buf) {
+		ut_ad("too long column" == 0);
+		return false;
+	}
 
 	return(!cmp_data_data(mtype, prtype, buf, len, sec_field, sec_len));
 }
@@ -218,6 +225,8 @@ row_sel_sec_rec_is_for_clust_rec(
 		ifield = dict_index_get_nth_field(sec_index, i);
 		col = dict_field_get_col(ifield);
 
+		sec_field = rec_get_nth_field(sec_rec, sec_offs, i, &sec_len);
+
 		is_virtual = dict_col_is_virtual(col);
 
 		/* For virtual column, its value will need to be
@@ -250,43 +259,54 @@ row_sel_sec_rec_is_for_clust_rec(
 				innobase_report_computed_value_failed(row);
 				return DB_COMPUTE_VALUE_FAILED;
 			}
-			clust_len = vfield->len;
+			len = clust_len = vfield->len;
 			clust_field = static_cast<byte*>(vfield->data);
 		} else {
 			clust_pos = dict_col_get_clust_pos(col, clust_index);
 
 			clust_field = rec_get_nth_field(
 				clust_rec, clust_offs, clust_pos, &clust_len);
-		}
-
-		sec_field = rec_get_nth_field(sec_rec, sec_offs, i, &sec_len);
-
-		len = clust_len;
-
-		if (ifield->prefix_len > 0 && len != UNIV_SQL_NULL
-		    && sec_len != UNIV_SQL_NULL && !is_virtual) {
+			if (clust_len == UNIV_SQL_NULL) {
+				if (sec_len == UNIV_SQL_NULL) {
+					continue;
+				}
+				return DB_SUCCESS;
+			}
+			if (sec_len == UNIV_SQL_NULL) {
+				return DB_SUCCESS;
+			}
 
+			len = clust_len;
 			if (rec_offs_nth_extern(clust_offs, clust_pos)) {
 				len -= BTR_EXTERN_FIELD_REF_SIZE;
 			}
 
-			len = dtype_get_at_most_n_mbchars(
-				col->prtype, col->mbminlen, col->mbmaxlen,
-				ifield->prefix_len, len, (char*) clust_field);
-
-			if (rec_offs_nth_extern(clust_offs, clust_pos)
-			    && len < sec_len) {
-				if (!row_sel_sec_rec_is_for_blob(
-					    col->mtype, col->prtype,
-					    col->mbminlen, col->mbmaxlen,
-					    clust_field, clust_len,
-					    sec_field, sec_len,
-					    ifield->prefix_len,
-					    clust_index->table)) {
-					return DB_SUCCESS;
+			if (ulint prefix_len = ifield->prefix_len) {
+				len = dtype_get_at_most_n_mbchars(
+					col->prtype, col->mbminlen,
+					col->mbmaxlen, prefix_len, len,
+					reinterpret_cast<const char*>(
+						clust_field));
+				if (len < sec_len) {
+					goto check_for_blob;
 				}
+			} else {
+check_for_blob:
+				if (rec_offs_nth_extern(clust_offs,
+							clust_pos)) {
+					if (!row_sel_sec_rec_is_for_blob(
+						    col->mtype, col->prtype,
+						    col->mbminlen,
+						    col->mbmaxlen,
+						    clust_field, clust_len,
+						    sec_field, sec_len,
+						    prefix_len,
+						    clust_index->table)) {
+						return DB_SUCCESS;
+					}
 
-				continue;
+					continue;
+				}
 			}
 		}