MDEV-38975: HEAP engine BLOB/TEXT/JSON/GEOMETRY column support

Allow BLOB/TEXT/JSON/GEOMETRY columns in MEMORY (HEAP) engine tables
by storing blob data in variable-length continuation record chains
within the existing `HP_BLOCK` structure.

**Continuation runs**: blob data is split across contiguous sequences
of `recbuffer`-sized records. Each run stores a 10-byte header
(`next_cont` pointer + `run_rec_count`) in the first record; inner
records (rec 1..N-1) have no flags byte — full `recbuffer` payload.
Runs are linked via `next_cont` pointers. Individual runs are capped
at 65,535 records (`uint16` format limit); larger blobs are
automatically split into multiple runs.

**Zero-copy reads**: single-run blobs return pointers directly into
`HP_BLOCK` records, avoiding `blob_buff` reassembly entirely:
- Case A (`run_rec_count == 1`): return `chain + HP_CONT_HEADER_SIZE`
- Case B (`HP_ROW_CONT_ZEROCOPY` flag): return `chain + recbuffer`
- Case C (multi-run): walk chain, reassemble into `blob_buff`
`HP_INFO::has_zerocopy_blobs` tracks zero-copy state; used by
`heap_update()` to refresh the caller's record buffer after freeing
old chains, preventing dangling pointers.

**Free list scavenging**: on insert, the free list is walked read-only
(peek) tracking contiguous groups in descending address order (LIFO).
Qualifying groups (>= `min_run_records`) are unlinked and used. The
first non-qualifying group terminates the scan — remaining data is
allocated from the block tail. The free list is never disturbed when
no qualifying group is found.

**Record counting**: new `HP_SHARE::total_records` tracks all physical
records (primary + continuation). `HP_SHARE::records` remains logical
(primary-only) to preserve linear hash bucket mapping correctness.

**Scan/check batch-skip**: `heap_scan()` and `heap_check_heap()` read
`run_rec_count` from rec 0 and skip entire continuation runs at once.

**Hash functions**: `hp_rec_hashnr()`, `hp_rec_key_cmp()`, `hp_key_cmp()`,
`hp_make_key()` updated to handle `HA_BLOB_PART` key segments — reading
actual blob data via pointer dereference or chain materialization.

**SQL layer**: `choose_engine()` no longer rejects HEAP for blob tables
(replaced `blob_fields` check with `reclength > HA_MAX_REC_LENGTH`).
`remove_duplicates()` routes HEAP+blob to `remove_dup_with_compare()`.
`ha_heap::remember_rnd_pos()` / `restart_rnd_next()` implemented for
DISTINCT deduplication support. Fixed undefined behavior in
`test_if_cheaper_ordering()` where `select_limit/fanout` could overflow
to infinity — capped at `HA_POS_ERROR`.

https://jira.mariadb.org/browse/MDEV-38975

Author: arcivanov

include/heap.h

@@ -131,21 +131,27 @@ typedef struct st_hp_keydef		/* Key definition with open */
   uint (*get_key_length)(struct st_hp_keydef *keydef, const uchar *key);
 } HP_KEYDEF;
 
+typedef struct st_hp_blob_desc
+{
+  uint offset;       /* Byte offset of blob descriptor within record buffer */
+  uint packlength;   /* 1, 2, 3, or 4: length prefix size */
+} HP_BLOB_DESC;
+
 typedef struct st_heap_share
 {
   HP_BLOCK block;
   HP_KEYDEF  *keydef;
   ulonglong data_length,index_length,max_table_size;
   ulonglong auto_increment;
   ulong min_records,max_records;	/* Params to open */
-  ulong records;			/* records */
+  ulong records;			/* Logical (primary) record count */
   ulong blength;			/* records rounded up to 2^n */
   ulong deleted;			/* Deleted records in database */
   uint key_stat_version;                /* version to indicate insert/delete */
   uint key_version;                     /* Updated on key change */
   uint file_version;                    /* Update on clear */
   uint reclength;			/* Length of one record */
-  uint visible;                         /* Offset to the visible/deleted mark */
+  uint visible;                         /* Offset to the flags byte (active/deleted/continuation) */
   uint changed;
   uint keys,max_key_length;
   uint currently_disabled_keys;    /* saved value from "keys" when disabled */
@@ -156,6 +162,9 @@ typedef struct st_heap_share
   THR_LOCK lock;
   my_bool delete_on_close;
   my_bool internal;                     /* Internal temporary table */
+  HP_BLOB_DESC *blob_descs;    /* Array of blob column descriptors */
+  uint blob_count;             /* Number of blob columns */
+  ulong total_records;         /* All active records (primary + blob continuation) */
   LIST open_list;
   uint auto_key;
   uint auto_key_type;			/* real type of the auto key segment */
@@ -181,6 +190,9 @@ typedef struct st_heap_info
   uint file_version;                    /* Version at scan */
   uint lastkey_len;
   my_bool implicit_emptied;
+  uchar *blob_buff;            /* Reassembly buffer for blob reads */
+  uint32 blob_buff_len;        /* Current allocated size of blob_buff */
+  my_bool has_zerocopy_blobs;  /* Last hp_read_blobs produced zero-copy ptrs */
   THR_LOCK_DATA lock;
   LIST open_list;
 } HP_INFO;
@@ -204,6 +216,8 @@ typedef struct st_heap_create_info
     open_count to 1. Is only looked at if not internal_table.
   */
   my_bool pin_share;
+  HP_BLOB_DESC *blob_descs;
+  uint blob_count;
 } HP_CREATE_INFO;
 
 	/* Prototypes for heap-functions */

mysql-test/include/mtr_check.sql

@@ -66,7 +66,7 @@ BEGIN
          collation_name, column_type, column_key, extra, column_comment
     FROM INFORMATION_SCHEMA.COLUMNS
       WHERE table_schema='mysql'
-        ORDER BY columns_in_mysql;
+        ORDER BY columns_in_mysql, ordinal_position;
 
   -- Dump all events, there should be none
   SELECT * FROM INFORMATION_SCHEMA.EVENTS;

mysql-test/main/blob_sj_test.result

@@ -0,0 +1,29 @@
+set optimizer_switch='materialization=on,in_to_exists=off,semijoin=off';
+set @blob_len = 16;
+set @prefix_len = 6;
+set @suffix_len = @blob_len - @prefix_len;
+create table t1 (a1 blob(16), a2 blob(16));
+create table t2 (b1 blob(16), b2 blob(16));
+insert into t1 values
+(concat('1 - 00', repeat('x', @suffix_len)), concat('2 - 00', repeat('x', @suffix_len)));
+insert into t1 values
+(concat('1 - 01', repeat('x', @suffix_len)), concat('2 - 01', repeat('x', @suffix_len)));
+insert into t1 values
+(concat('1 - 02', repeat('x', @suffix_len)), concat('2 - 02', repeat('x', @suffix_len)));
+insert into t2 values
+(concat('1 - 01', repeat('x', @suffix_len)), concat('2 - 01', repeat('x', @suffix_len)));
+insert into t2 values
+(concat('1 - 02', repeat('x', @suffix_len)), concat('2 - 02', repeat('x', @suffix_len)));
+insert into t2 values
+(concat('1 - 03', repeat('x', @suffix_len)), concat('2 - 03', repeat('x', @suffix_len)));
+explain extended select left(a1,7), left(a2,7) from t1 where a1 in (select b1 from t2 where b1 > '0');
+id	select_type	table	type	possible_keys	key	key_len	ref	rows	filtered	Extra
+1	PRIMARY	t1	ALL	NULL	NULL	NULL	NULL	3	100.00	Using where
+2	DEPENDENT SUBQUERY	t2	ALL	NULL	NULL	NULL	NULL	3	100.00	Using where
+Warnings:
+Note	1003	/* select#1 */ select left(`test`.`t1`.`a1`,7) AS `left(a1,7)`,left(`test`.`t1`.`a2`,7) AS `left(a2,7)` from `test`.`t1` where <expr_cache><`test`.`t1`.`a1`>(<in_optimizer>(`test`.`t1`.`a1`,<exists>(/* select#2 */ select `test`.`t2`.`b1` from `test`.`t2` where `test`.`t2`.`b1` > '0' and <cache>(`test`.`t1`.`a1`) = `test`.`t2`.`b1`)))
+select left(a1,7), left(a2,7) from t1 where a1 in (select b1 from t2 where b1 > '0');
+left(a1,7)	left(a2,7)
+1 - 01x	2 - 01x
+1 - 02x	2 - 02x
+drop table t1, t2;

mysql-test/main/blob_sj_test.test

@@ -0,0 +1,26 @@
+set optimizer_switch='materialization=on,in_to_exists=off,semijoin=off';
+set @blob_len = 16;
+set @prefix_len = 6;
+set @suffix_len = @blob_len - @prefix_len;
+
+create table t1 (a1 blob(16), a2 blob(16));
+create table t2 (b1 blob(16), b2 blob(16));
+
+insert into t1 values
+(concat('1 - 00', repeat('x', @suffix_len)), concat('2 - 00', repeat('x', @suffix_len)));
+insert into t1 values
+(concat('1 - 01', repeat('x', @suffix_len)), concat('2 - 01', repeat('x', @suffix_len)));
+insert into t1 values
+(concat('1 - 02', repeat('x', @suffix_len)), concat('2 - 02', repeat('x', @suffix_len)));
+
+insert into t2 values
+(concat('1 - 01', repeat('x', @suffix_len)), concat('2 - 01', repeat('x', @suffix_len)));
+insert into t2 values
+(concat('1 - 02', repeat('x', @suffix_len)), concat('2 - 02', repeat('x', @suffix_len)));
+insert into t2 values
+(concat('1 - 03', repeat('x', @suffix_len)), concat('2 - 03', repeat('x', @suffix_len)));
+
+explain extended select left(a1,7), left(a2,7) from t1 where a1 in (select b1 from t2 where b1 > '0');
+select left(a1,7), left(a2,7) from t1 where a1 in (select b1 from t2 where b1 > '0');
+
+drop table t1, t2;

mysql-test/main/create.result

@@ -30,10 +30,7 @@ Note	1051	Unknown table 'test.t1,test.t2'
 create table t1 (b char(0) not null, index(b));
 ERROR 42000: The storage engine MyISAM can't index column `b`
 create table t1 (a int not null,b text) engine=heap;
-ERROR 42000: Storage engine MEMORY doesn't support BLOB/TEXT columns
-drop table if exists t1;
-Warnings:
-Note	1051	Unknown table 'test.t1'
+drop table t1;
 create table t1 (ordid int(8) not null auto_increment, ord  varchar(50) not null, primary key (ord,ordid)) engine=heap;
 ERROR 42000: Incorrect table definition; there can be only one auto column and it must be defined as a key
 create table not_existing_database.test (a int);
@@ -1089,7 +1086,7 @@ t1	CREATE TABLE `t1` (
   `QUERY_ID` bigint(4) NOT NULL,
   `INFO_BINARY` blob,
   `TID` bigint(4) NOT NULL
-)  DEFAULT CHARSET=utf8mb3 COLLATE=utf8mb3_general_ci
+) ENGINE=MEMORY DEFAULT CHARSET=utf8mb3 COLLATE=utf8mb3_general_ci
 drop table t1;
 create temporary table t1 like information_schema.processlist;
 show create table t1;
@@ -1113,7 +1110,7 @@ t1	CREATE TEMPORARY TABLE `t1` (
   `QUERY_ID` bigint(4) NOT NULL,
   `INFO_BINARY` blob,
   `TID` bigint(4) NOT NULL
-)  DEFAULT CHARSET=utf8mb3 COLLATE=utf8mb3_general_ci
+) ENGINE=MEMORY DEFAULT CHARSET=utf8mb3 COLLATE=utf8mb3_general_ci
 drop table t1;
 create table t1 like information_schema.character_sets;
 show create table t1;

mysql-test/main/create.test

@@ -30,9 +30,8 @@ create table t2 select auto+1 from t1;
 drop table if exists t1,t2;
 --error ER_WRONG_KEY_COLUMN
 create table t1 (b char(0) not null, index(b));
---error ER_TABLE_CANT_HANDLE_BLOB
 create table t1 (a int not null,b text) engine=heap;
-drop table if exists t1;
+drop table t1;
 
 --error ER_WRONG_AUTO_KEY
 create table t1 (ordid int(8) not null auto_increment, ord  varchar(50) not null, primary key (ord,ordid)) engine=heap;

mysql-test/main/cte_recursive.test

@@ -3212,6 +3212,8 @@ show create table t2;
 --eval insert ignore into t2 $query;
 drop table t2;
 set @@sql_mode="";
+# Rows with identical (level, mid) due to overflow have non-deterministic order
+--sorted_result
 --eval $query
 --eval create table t2 as $query;
 show create table t2;

mysql-test/main/derived_view.result

@@ -2372,7 +2372,7 @@ GROUP BY TABLE_SCHEMA) AS UNIQUES
 ON ( COLUMNS.TABLE_SCHEMA = UNIQUES.TABLE_SCHEMA);
 id	select_type	table	type	possible_keys	key	key_len	ref	rows	Extra
 1	PRIMARY	COLUMNS	ALL	NULL	NULL	NULL	NULL	NULL	Open_frm_only; Scanned all databases
-1	PRIMARY	<derived2>	ALL	NULL	NULL	NULL	NULL	2	Using where; Using join buffer (flat, BNL join)
+1	PRIMARY	<derived2>	ref	key0	key0	194	information_schema.COLUMNS.TABLE_SCHEMA	2	
 2	DERIVED	STATISTICS	ALL	NULL	NULL	NULL	NULL	NULL	Open_frm_only; Scanned all databases; Using filesort
 SELECT COUNT(*) > 0
 FROM INFORMATION_SCHEMA.COLUMNS

mysql-test/main/distinct.result

@@ -1189,7 +1189,7 @@ insert into t1 values (1, 'Aa123456', 'abc'), (2, 'Bb7897777', 'def'),
 (3, 'Cc01287', 'xyz'), (5, 'd12345', 'efg');
 select distinct if(sum(a), b, 0) from t1 group by value(c) with rollup;
 if(sum(a), b, 0)
-Aa123456
+SOME_B_VALUE
 drop table t1;
 #
 # end of 10.5 tests

mysql-test/main/distinct.test

@@ -915,6 +915,9 @@ create table t1 (a int, b longtext, c varchar(18));
 insert into t1 values (1, 'Aa123456', 'abc'), (2, 'Bb7897777', 'def'),
   (3, 'Cc01287', 'xyz'), (5, 'd12345', 'efg');
 
+# ROLLUP row's b value is indeterminate (depends on last group processed),
+# which varies by temp table engine (HEAP vs Aria). Mask the value.
+--replace_regex /(Aa123456|Bb7897777|Cc01287|d12345)/SOME_B_VALUE/
 select distinct if(sum(a), b, 0) from t1 group by value(c) with rollup;
 drop table t1;