diff --git a/mysql-test/suite/heap/blob_vers_multi.result b/mysql-test/suite/heap/blob_vers_multi.result
new file mode 100644
index 0000000000000..c6a9ef703cfb5
--- /dev/null
+++ b/mysql-test/suite/heap/blob_vers_multi.result
@@ -0,0 +1,85 @@
+#
+# The remaining vers_insert_history_row() callers on a HEAP table with
+# blobs: multi-table UPDATE (both the on-the-fly and the deferred
+# do_updates() path) and REPLACE.
+#
+# Each of them writes the history row from a record buffer that a read
+# filled with zero-copy blob pointers into HP_BLOCK, so each can hand
+# heap_write() a blob source living in the chain ha_update_row() just
+# parked for deferred free.
+#
+#
+# Multi-table UPDATE: the first table is updated while scanning, the
+# second goes through the deferred do_updates() path
+#
+CREATE TABLE t1 (pk INT PRIMARY KEY, f TEXT)
+ENGINE=HEAP WITH SYSTEM VERSIONING CHARACTER SET latin1;
+CREATE TABLE t2 (pk INT PRIMARY KEY, f TEXT)
+ENGINE=HEAP WITH SYSTEM VERSIONING CHARACTER SET latin1;
+INSERT INTO t1 VALUES (1, REPEAT('abcdefgh',8)), (2, REPEAT('ijklmnop',60));
+INSERT INTO t2 VALUES (1, REPEAT('12345678',40)), (2, REPEAT('!@#$%^&*',300));
+UPDATE t1, t2 SET t1.f = CONCAT('n1_', t1.pk), t2.f = CONCAT('n2_', t2.pk)
+WHERE t1.pk = t2.pk;
+SELECT pk, f FROM t1 ORDER BY pk;
+pk	f
+1	n1_1
+2	n1_2
+SELECT pk, f FROM t2 ORDER BY pk;
+pk	f
+1	n2_1
+2	n2_2
+SELECT pk, LENGTH(f),
+f = REPEAT('abcdefgh',8) OR f = REPEAT('ijklmnop',60) AS hist_ok
+FROM t1 FOR SYSTEM_TIME ALL WHERE f NOT LIKE 'n1\_%' ORDER BY pk;
+pk	LENGTH(f)	hist_ok
+1	64	1
+2	480	1
+SELECT pk, LENGTH(f),
+f = REPEAT('12345678',40) OR f = REPEAT('!@#$%^&*',300) AS hist_ok
+FROM t2 FOR SYSTEM_TIME ALL WHERE f NOT LIKE 'n2\_%' ORDER BY pk;
+pk	LENGTH(f)	hist_ok
+1	320	1
+2	2400	1
+DROP TABLE t1, t2;
+#
+# Multi-table UPDATE touching only the table that is not first in the
+# join order
+#
+CREATE TABLE t1 (pk INT PRIMARY KEY, c INT) ENGINE=HEAP;
+CREATE TABLE t2 (pk INT PRIMARY KEY, b1 TEXT, b2 MEDIUMTEXT)
+ENGINE=HEAP WITH SYSTEM VERSIONING CHARACTER SET latin1;
+INSERT INTO t1 VALUES (1, 10), (2, 20);
+INSERT INTO t2 VALUES (1, REPEAT('abcdefgh',50), REPEAT('ABCDEFGH',150)),
+(2, REPEAT('qrstuvwx',25), REPEAT('ijklmnop',400));
+# STRAIGHT_JOIN pins t2 second, so it goes through do_updates() rather
+# than being updated on the fly like the case above
+UPDATE t1 STRAIGHT_JOIN t2 ON t1.pk = t2.pk
+SET t2.b1 = REPEAT('z', 900), t2.b2 = 'y';
+SELECT pk,
+b1 = CASE pk WHEN 1 THEN REPEAT('abcdefgh',50)
+ELSE REPEAT('qrstuvwx',25) END AS b1_ok,
+b2 = CASE pk WHEN 1 THEN REPEAT('ABCDEFGH',150)
+ELSE REPEAT('ijklmnop',400) END AS b2_ok
+FROM t2 FOR SYSTEM_TIME ALL WHERE b2 <> 'y' ORDER BY pk;
+pk	b1_ok	b2_ok
+1	1	1
+2	1	1
+DROP TABLE t1, t2;
+#
+# REPLACE reaching the optimized ha_update_row() path
+#
+CREATE TABLE t (pk INT PRIMARY KEY, f TEXT)
+ENGINE=HEAP WITH SYSTEM VERSIONING CHARACTER SET latin1;
+INSERT INTO t VALUES (1, REPEAT('abcdefgh',8)), (2, REPEAT('ijklmnop',60));
+REPLACE INTO t VALUES (1, 'r1'), (2, 'r2');
+SELECT pk, f FROM t ORDER BY pk;
+pk	f
+1	r1
+2	r2
+SELECT pk, LENGTH(f),
+f = REPEAT('abcdefgh',8) OR f = REPEAT('ijklmnop',60) AS hist_ok
+FROM t FOR SYSTEM_TIME ALL WHERE f NOT LIKE 'r%' ORDER BY pk;
+pk	LENGTH(f)	hist_ok
+1	64	1
+2	480	1
+DROP TABLE t;
diff --git a/mysql-test/suite/heap/blob_vers_multi.test b/mysql-test/suite/heap/blob_vers_multi.test
new file mode 100644
index 0000000000000..199d485ab719f
--- /dev/null
+++ b/mysql-test/suite/heap/blob_vers_multi.test
@@ -0,0 +1,79 @@
+--echo #
+--echo # The remaining vers_insert_history_row() callers on a HEAP table with
+--echo # blobs: multi-table UPDATE (both the on-the-fly and the deferred
+--echo # do_updates() path) and REPLACE.
+--echo #
+--echo # Each of them writes the history row from a record buffer that a read
+--echo # filled with zero-copy blob pointers into HP_BLOCK, so each can hand
+--echo # heap_write() a blob source living in the chain ha_update_row() just
+--echo # parked for deferred free.
+--echo #
+
+--echo #
+--echo # Multi-table UPDATE: the first table is updated while scanning, the
+--echo # second goes through the deferred do_updates() path
+--echo #
+CREATE TABLE t1 (pk INT PRIMARY KEY, f TEXT)
+  ENGINE=HEAP WITH SYSTEM VERSIONING CHARACTER SET latin1;
+CREATE TABLE t2 (pk INT PRIMARY KEY, f TEXT)
+  ENGINE=HEAP WITH SYSTEM VERSIONING CHARACTER SET latin1;
+
+INSERT INTO t1 VALUES (1, REPEAT('abcdefgh',8)), (2, REPEAT('ijklmnop',60));
+INSERT INTO t2 VALUES (1, REPEAT('12345678',40)), (2, REPEAT('!@#$%^&*',300));
+
+UPDATE t1, t2 SET t1.f = CONCAT('n1_', t1.pk), t2.f = CONCAT('n2_', t2.pk)
+  WHERE t1.pk = t2.pk;
+
+SELECT pk, f FROM t1 ORDER BY pk;
+SELECT pk, f FROM t2 ORDER BY pk;
+
+SELECT pk, LENGTH(f),
+       f = REPEAT('abcdefgh',8) OR f = REPEAT('ijklmnop',60) AS hist_ok
+  FROM t1 FOR SYSTEM_TIME ALL WHERE f NOT LIKE 'n1\_%' ORDER BY pk;
+SELECT pk, LENGTH(f),
+       f = REPEAT('12345678',40) OR f = REPEAT('!@#$%^&*',300) AS hist_ok
+  FROM t2 FOR SYSTEM_TIME ALL WHERE f NOT LIKE 'n2\_%' ORDER BY pk;
+
+DROP TABLE t1, t2;
+
+--echo #
+--echo # Multi-table UPDATE touching only the table that is not first in the
+--echo # join order
+--echo #
+CREATE TABLE t1 (pk INT PRIMARY KEY, c INT) ENGINE=HEAP;
+CREATE TABLE t2 (pk INT PRIMARY KEY, b1 TEXT, b2 MEDIUMTEXT)
+  ENGINE=HEAP WITH SYSTEM VERSIONING CHARACTER SET latin1;
+
+INSERT INTO t1 VALUES (1, 10), (2, 20);
+INSERT INTO t2 VALUES (1, REPEAT('abcdefgh',50), REPEAT('ABCDEFGH',150)),
+                      (2, REPEAT('qrstuvwx',25), REPEAT('ijklmnop',400));
+
+--echo # STRAIGHT_JOIN pins t2 second, so it goes through do_updates() rather
+--echo # than being updated on the fly like the case above
+UPDATE t1 STRAIGHT_JOIN t2 ON t1.pk = t2.pk
+  SET t2.b1 = REPEAT('z', 900), t2.b2 = 'y';
+
+SELECT pk,
+       b1 = CASE pk WHEN 1 THEN REPEAT('abcdefgh',50)
+                    ELSE REPEAT('qrstuvwx',25) END AS b1_ok,
+       b2 = CASE pk WHEN 1 THEN REPEAT('ABCDEFGH',150)
+                    ELSE REPEAT('ijklmnop',400) END AS b2_ok
+  FROM t2 FOR SYSTEM_TIME ALL WHERE b2 <> 'y' ORDER BY pk;
+
+DROP TABLE t1, t2;
+
+--echo #
+--echo # REPLACE reaching the optimized ha_update_row() path
+--echo #
+CREATE TABLE t (pk INT PRIMARY KEY, f TEXT)
+  ENGINE=HEAP WITH SYSTEM VERSIONING CHARACTER SET latin1;
+INSERT INTO t VALUES (1, REPEAT('abcdefgh',8)), (2, REPEAT('ijklmnop',60));
+
+REPLACE INTO t VALUES (1, 'r1'), (2, 'r2');
+
+SELECT pk, f FROM t ORDER BY pk;
+SELECT pk, LENGTH(f),
+       f = REPEAT('abcdefgh',8) OR f = REPEAT('ijklmnop',60) AS hist_ok
+  FROM t FOR SYSTEM_TIME ALL WHERE f NOT LIKE 'r%' ORDER BY pk;
+
+DROP TABLE t;
diff --git a/mysql-test/suite/heap/blob_vers_odku.result b/mysql-test/suite/heap/blob_vers_odku.result
new file mode 100644
index 0000000000000..da7926e8d7edb
--- /dev/null
+++ b/mysql-test/suite/heap/blob_vers_odku.result
@@ -0,0 +1,95 @@
+#
+# INSERT ... ON DUPLICATE KEY UPDATE on a HEAP table with blobs.
+#
+# The duplicate branch reads the conflicting row into record[1] (blob
+# pointers are zero-copy pointers into HP_BLOCK), copies it into
+# record[0] with restore_record(table, record[1]), applies the update
+# fields and calls ha_update_row().  On a system-versioned table it then
+# calls vers_insert_history_row(), which does the same restore_record()
+# and writes the history row -- with a blob source still pointing at the
+# chain the update just parked for deferred free.
+#
+#
+# Versioned: history row must keep the pre-update blob
+#
+CREATE TABLE t (pk INT PRIMARY KEY, f TEXT)
+ENGINE=HEAP WITH SYSTEM VERSIONING CHARACTER SET latin1;
+INSERT INTO t VALUES (1, REPEAT('abcdefgh',8)), (2, REPEAT('ijklmnop',60));
+INSERT INTO t VALUES (1, 'ignored'), (2, 'ignored')
+ON DUPLICATE KEY UPDATE f = CONCAT('new', pk);
+SELECT pk, f, LENGTH(f) FROM t ORDER BY pk;
+pk	f	LENGTH(f)
+1	new1	4
+2	new2	4
+SELECT pk, LENGTH(f),
+f = REPEAT('abcdefgh',8) OR f = REPEAT('ijklmnop',60) AS hist_ok
+FROM t FOR SYSTEM_TIME ALL WHERE f NOT LIKE 'new%' ORDER BY pk;
+pk	LENGTH(f)	hist_ok
+1	64	1
+2	480	1
+DROP TABLE t;
+#
+# Versioned, blob column NOT in the update list: the history row carries
+# the blob forward unchanged
+#
+CREATE TABLE t (pk INT PRIMARY KEY, c INT, f TEXT)
+ENGINE=HEAP WITH SYSTEM VERSIONING CHARACTER SET latin1;
+INSERT INTO t VALUES (1, 10, REPEAT('abcdefgh',40)),
+(2, 20, REPEAT('12345678',200));
+INSERT INTO t (pk, c, f) VALUES (1, 0, 'ignored'), (2, 0, 'ignored')
+ON DUPLICATE KEY UPDATE c = c + 1;
+SELECT pk, c, LENGTH(f),
+f = REPEAT('abcdefgh',40) OR f = REPEAT('12345678',200) AS cur_ok
+FROM t ORDER BY pk;
+pk	c	LENGTH(f)	cur_ok
+1	11	320	1
+2	21	1600	1
+SELECT pk, c, LENGTH(f),
+f = REPEAT('abcdefgh',40) OR f = REPEAT('12345678',200) AS hist_ok
+FROM t FOR SYSTEM_TIME ALL WHERE c IN (10, 20) ORDER BY pk;
+pk	c	LENGTH(f)	hist_ok
+1	10	320	1
+2	20	1600	1
+DROP TABLE t;
+#
+# Versioned, multiple blob columns, mixed growth and shrink
+#
+CREATE TABLE t (pk INT PRIMARY KEY, b1 TEXT, b2 MEDIUMTEXT)
+ENGINE=HEAP WITH SYSTEM VERSIONING CHARACTER SET latin1;
+INSERT INTO t VALUES (1, REPEAT('abcdefgh',50), REPEAT('ABCDEFGH',150)),
+(2, REPEAT('12345678',25), REPEAT('!@#$%^&*',400));
+INSERT INTO t VALUES (1, 'ignored', 'ignored'), (2, 'ignored', 'ignored')
+ON DUPLICATE KEY UPDATE b1 = REPEAT('z', 900), b2 = 'y';
+SELECT pk,
+b1 = CASE pk WHEN 1 THEN REPEAT('abcdefgh',50)
+ELSE REPEAT('12345678',25) END AS b1_ok,
+b2 = CASE pk WHEN 1 THEN REPEAT('ABCDEFGH',150)
+ELSE REPEAT('!@#$%^&*',400) END AS b2_ok
+FROM t FOR SYSTEM_TIME ALL WHERE b2 <> 'y' ORDER BY pk;
+pk	b1_ok	b2_ok
+1	1	1
+2	1	1
+DROP TABLE t;
+#
+# Non-versioned control: the same statement shape must leave the stored
+# rows intact
+#
+CREATE TABLE t (pk INT PRIMARY KEY, c INT, f TEXT)
+ENGINE=HEAP CHARACTER SET latin1;
+INSERT INTO t VALUES (1, 10, REPEAT('abcdefgh',40)),
+(2, 20, REPEAT('12345678',200));
+INSERT INTO t (pk, c, f) VALUES (1, 0, 'ignored'), (2, 0, 'ignored')
+ON DUPLICATE KEY UPDATE c = c + 1;
+SELECT pk, c, LENGTH(f),
+f = REPEAT('abcdefgh',40) OR f = REPEAT('12345678',200) AS f_ok
+FROM t ORDER BY pk;
+pk	c	LENGTH(f)	f_ok
+1	11	320	1
+2	21	1600	1
+INSERT INTO t (pk, c, f) VALUES (1, 0, 'ignored'), (2, 0, 'ignored')
+ON DUPLICATE KEY UPDATE f = CONCAT('new', pk);
+SELECT pk, c, f FROM t ORDER BY pk;
+pk	c	f
+1	11	new1
+2	21	new2
+DROP TABLE t;
diff --git a/mysql-test/suite/heap/blob_vers_odku.test b/mysql-test/suite/heap/blob_vers_odku.test
new file mode 100644
index 0000000000000..28d051d18550a
--- /dev/null
+++ b/mysql-test/suite/heap/blob_vers_odku.test
@@ -0,0 +1,88 @@
+--echo #
+--echo # INSERT ... ON DUPLICATE KEY UPDATE on a HEAP table with blobs.
+--echo #
+--echo # The duplicate branch reads the conflicting row into record[1] (blob
+--echo # pointers are zero-copy pointers into HP_BLOCK), copies it into
+--echo # record[0] with restore_record(table, record[1]), applies the update
+--echo # fields and calls ha_update_row().  On a system-versioned table it then
+--echo # calls vers_insert_history_row(), which does the same restore_record()
+--echo # and writes the history row -- with a blob source still pointing at the
+--echo # chain the update just parked for deferred free.
+--echo #
+
+--echo #
+--echo # Versioned: history row must keep the pre-update blob
+--echo #
+CREATE TABLE t (pk INT PRIMARY KEY, f TEXT)
+  ENGINE=HEAP WITH SYSTEM VERSIONING CHARACTER SET latin1;
+INSERT INTO t VALUES (1, REPEAT('abcdefgh',8)), (2, REPEAT('ijklmnop',60));
+
+INSERT INTO t VALUES (1, 'ignored'), (2, 'ignored')
+  ON DUPLICATE KEY UPDATE f = CONCAT('new', pk);
+
+SELECT pk, f, LENGTH(f) FROM t ORDER BY pk;
+SELECT pk, LENGTH(f),
+       f = REPEAT('abcdefgh',8) OR f = REPEAT('ijklmnop',60) AS hist_ok
+  FROM t FOR SYSTEM_TIME ALL WHERE f NOT LIKE 'new%' ORDER BY pk;
+DROP TABLE t;
+
+--echo #
+--echo # Versioned, blob column NOT in the update list: the history row carries
+--echo # the blob forward unchanged
+--echo #
+CREATE TABLE t (pk INT PRIMARY KEY, c INT, f TEXT)
+  ENGINE=HEAP WITH SYSTEM VERSIONING CHARACTER SET latin1;
+INSERT INTO t VALUES (1, 10, REPEAT('abcdefgh',40)),
+                     (2, 20, REPEAT('12345678',200));
+
+INSERT INTO t (pk, c, f) VALUES (1, 0, 'ignored'), (2, 0, 'ignored')
+  ON DUPLICATE KEY UPDATE c = c + 1;
+
+SELECT pk, c, LENGTH(f),
+       f = REPEAT('abcdefgh',40) OR f = REPEAT('12345678',200) AS cur_ok
+  FROM t ORDER BY pk;
+SELECT pk, c, LENGTH(f),
+       f = REPEAT('abcdefgh',40) OR f = REPEAT('12345678',200) AS hist_ok
+  FROM t FOR SYSTEM_TIME ALL WHERE c IN (10, 20) ORDER BY pk;
+DROP TABLE t;
+
+--echo #
+--echo # Versioned, multiple blob columns, mixed growth and shrink
+--echo #
+CREATE TABLE t (pk INT PRIMARY KEY, b1 TEXT, b2 MEDIUMTEXT)
+  ENGINE=HEAP WITH SYSTEM VERSIONING CHARACTER SET latin1;
+INSERT INTO t VALUES (1, REPEAT('abcdefgh',50), REPEAT('ABCDEFGH',150)),
+                     (2, REPEAT('12345678',25), REPEAT('!@#$%^&*',400));
+
+INSERT INTO t VALUES (1, 'ignored', 'ignored'), (2, 'ignored', 'ignored')
+  ON DUPLICATE KEY UPDATE b1 = REPEAT('z', 900), b2 = 'y';
+
+SELECT pk,
+       b1 = CASE pk WHEN 1 THEN REPEAT('abcdefgh',50)
+                    ELSE REPEAT('12345678',25) END AS b1_ok,
+       b2 = CASE pk WHEN 1 THEN REPEAT('ABCDEFGH',150)
+                    ELSE REPEAT('!@#$%^&*',400) END AS b2_ok
+  FROM t FOR SYSTEM_TIME ALL WHERE b2 <> 'y' ORDER BY pk;
+DROP TABLE t;
+
+--echo #
+--echo # Non-versioned control: the same statement shape must leave the stored
+--echo # rows intact
+--echo #
+CREATE TABLE t (pk INT PRIMARY KEY, c INT, f TEXT)
+  ENGINE=HEAP CHARACTER SET latin1;
+INSERT INTO t VALUES (1, 10, REPEAT('abcdefgh',40)),
+                     (2, 20, REPEAT('12345678',200));
+
+INSERT INTO t (pk, c, f) VALUES (1, 0, 'ignored'), (2, 0, 'ignored')
+  ON DUPLICATE KEY UPDATE c = c + 1;
+
+SELECT pk, c, LENGTH(f),
+       f = REPEAT('abcdefgh',40) OR f = REPEAT('12345678',200) AS f_ok
+  FROM t ORDER BY pk;
+
+INSERT INTO t (pk, c, f) VALUES (1, 0, 'ignored'), (2, 0, 'ignored')
+  ON DUPLICATE KEY UPDATE f = CONCAT('new', pk);
+
+SELECT pk, c, f FROM t ORDER BY pk;
+DROP TABLE t;
diff --git a/mysql-test/suite/heap/blob_vers_repl.result b/mysql-test/suite/heap/blob_vers_repl.result
new file mode 100644
index 0000000000000..ee55a13a02420
--- /dev/null
+++ b/mysql-test/suite/heap/blob_vers_repl.result
@@ -0,0 +1,25 @@
+include/master-slave.inc
+[connection master]
+CREATE TABLE t (pk INT PRIMARY KEY, f TEXT)
+ENGINE=HEAP WITH SYSTEM VERSIONING CHARACTER SET latin1;
+INSERT INTO t VALUES (1, REPEAT('abcdefgh',8)), (2, REPEAT('ijklmnop',80));
+UPDATE t SET f = CONCAT('new', pk);
+# Master: history rows
+SELECT pk, LENGTH(f),
+f = REPEAT('abcdefgh',8) OR f = REPEAT('ijklmnop',80) AS hist_ok
+FROM t FOR SYSTEM_TIME ALL WHERE f NOT LIKE 'new%' ORDER BY pk;
+pk	LENGTH(f)	hist_ok
+1	64	1
+2	640	1
+connection slave;
+# Slave: history rows must match
+SELECT pk, LENGTH(f),
+f = REPEAT('abcdefgh',8) OR f = REPEAT('ijklmnop',80) AS hist_ok
+FROM t FOR SYSTEM_TIME ALL WHERE f NOT LIKE 'new%' ORDER BY pk;
+pk	LENGTH(f)	hist_ok
+1	64	1
+2	640	1
+connection master;
+DROP TABLE t;
+connection slave;
+include/rpl_end.inc
diff --git a/mysql-test/suite/heap/blob_vers_repl.test b/mysql-test/suite/heap/blob_vers_repl.test
new file mode 100644
index 0000000000000..c88061ac4594f
--- /dev/null
+++ b/mysql-test/suite/heap/blob_vers_repl.test
@@ -0,0 +1,37 @@
+--source include/have_binlog_format_row.inc
+--source include/master-slave.inc
+
+#
+# Does the binlog row image of a system versioning history row carry valid
+# blob data?
+#
+# vers_insert_history_row() writes record[0], whose blob pointer is a
+# zero-copy pointer into HP_BLOCK.  handler::ha_write_row() calls
+# binlog_log_row() AFTER write_row() returns, re-reading record[0].  If the
+# write invalidated what that pointer refers to, the logged row image is
+# garbage and the corruption propagates to the replica.
+#
+
+CREATE TABLE t (pk INT PRIMARY KEY, f TEXT)
+  ENGINE=HEAP WITH SYSTEM VERSIONING CHARACTER SET latin1;
+
+INSERT INTO t VALUES (1, REPEAT('abcdefgh',8)), (2, REPEAT('ijklmnop',80));
+UPDATE t SET f = CONCAT('new', pk);
+
+--echo # Master: history rows
+SELECT pk, LENGTH(f),
+       f = REPEAT('abcdefgh',8) OR f = REPEAT('ijklmnop',80) AS hist_ok
+  FROM t FOR SYSTEM_TIME ALL WHERE f NOT LIKE 'new%' ORDER BY pk;
+
+--sync_slave_with_master
+
+--echo # Slave: history rows must match
+SELECT pk, LENGTH(f),
+       f = REPEAT('abcdefgh',8) OR f = REPEAT('ijklmnop',80) AS hist_ok
+  FROM t FOR SYSTEM_TIME ALL WHERE f NOT LIKE 'new%' ORDER BY pk;
+
+--connection master
+DROP TABLE t;
+--sync_slave_with_master
+
+--source include/rpl_end.inc
diff --git a/mysql-test/suite/heap/blob_vers_trigger.result b/mysql-test/suite/heap/blob_vers_trigger.result
new file mode 100644
index 0000000000000..411e372b36853
--- /dev/null
+++ b/mysql-test/suite/heap/blob_vers_trigger.result
@@ -0,0 +1,155 @@
+#
+# An AFTER UPDATE trigger reads OLD.<blob> from the same record buffer
+# that vers_insert_history_row() builds the history row from.
+#
+# heap_update() parks the old blob chain instead of freeing it, so that
+# buffer keeps a valid zero-copy pointer for the rest of the statement.
+# The history-row write redeems that parking, so it must not leave the
+# buffer pointing at records handed back to the free list -- otherwise
+# OLD.<blob> reads whatever was written over them, and that value is
+# what lands in the audit table and in the binlog.
+#
+CREATE TABLE lg (id INT AUTO_INCREMENT PRIMARY KEY, ln INT, ok_val INT)
+ENGINE=MyISAM;
+#
+# Single-table UPDATE, blob sizes on either side of a single chain run
+#
+CREATE TABLE t (pk INT PRIMARY KEY, f TEXT)
+ENGINE=HEAP WITH SYSTEM VERSIONING CHARACTER SET latin1;
+CREATE TRIGGER trg AFTER UPDATE ON t FOR EACH ROW
+INSERT INTO lg (ln, ok_val)
+VALUES (LENGTH(OLD.f),
+OLD.f = CASE OLD.pk WHEN 1 THEN REPEAT('abcdefgh',8)
+ELSE REPEAT('ijklmnop',80) END);
+INSERT INTO t VALUES (1, REPEAT('abcdefgh',8)), (2, REPEAT('ijklmnop',80));
+UPDATE t SET f = CONCAT('new', pk);
+# OLD.<blob> as the trigger saw it
+SELECT ln, ok_val FROM lg ORDER BY id;
+ln	ok_val
+64	1
+640	1
+# and the history rows themselves
+SELECT pk, LENGTH(f),
+f = CASE pk WHEN 1 THEN REPEAT('abcdefgh',8)
+ELSE REPEAT('ijklmnop',80) END AS hist_ok
+FROM t FOR SYSTEM_TIME ALL WHERE f NOT LIKE 'new%' ORDER BY pk;
+pk	LENGTH(f)	hist_ok
+1	64	1
+2	640	1
+DELETE FROM lg;
+DROP TABLE t;
+#
+# Multiple blob columns
+#
+CREATE TABLE t (pk INT PRIMARY KEY, b1 TEXT, b2 MEDIUMTEXT)
+ENGINE=HEAP WITH SYSTEM VERSIONING CHARACTER SET latin1;
+CREATE TRIGGER trg AFTER UPDATE ON t FOR EACH ROW
+INSERT INTO lg (ln, ok_val)
+VALUES (LENGTH(OLD.b1) + LENGTH(OLD.b2),
+OLD.b1 = CASE OLD.pk WHEN 1 THEN REPEAT('abcdefgh',50)
+ELSE REPEAT('qrstuvwx',25) END
+AND
+OLD.b2 = CASE OLD.pk WHEN 1 THEN REPEAT('ABCDEFGH',150)
+ELSE REPEAT('ijklmnop',400) END);
+INSERT INTO t VALUES (1, REPEAT('abcdefgh',50), REPEAT('ABCDEFGH',150)),
+(2, REPEAT('qrstuvwx',25), REPEAT('ijklmnop',400));
+UPDATE t SET b1 = REPEAT('z',900), b2 = 'y';
+SELECT ln, ok_val FROM lg ORDER BY id;
+ln	ok_val
+1600	1
+3400	1
+DELETE FROM lg;
+DROP TABLE t;
+#
+# BEFORE UPDATE runs ahead of the history-row write and must be
+# unaffected; AFTER UPDATE on the same table must agree with it
+#
+CREATE TABLE t (pk INT PRIMARY KEY, f TEXT)
+ENGINE=HEAP WITH SYSTEM VERSIONING CHARACTER SET latin1;
+CREATE TRIGGER trg_b BEFORE UPDATE ON t FOR EACH ROW
+INSERT INTO lg (ln, ok_val)
+VALUES (LENGTH(OLD.f), OLD.f = REPEAT('abcdefgh',80));
+CREATE TRIGGER trg_a AFTER UPDATE ON t FOR EACH ROW
+INSERT INTO lg (ln, ok_val)
+VALUES (LENGTH(OLD.f), OLD.f = REPEAT('abcdefgh',80));
+INSERT INTO t VALUES (1, REPEAT('abcdefgh',80)), (2, REPEAT('abcdefgh',80));
+UPDATE t SET f = CONCAT('new', pk);
+SELECT ln, ok_val FROM lg ORDER BY id;
+ln	ok_val
+640	1
+640	1
+640	1
+640	1
+DELETE FROM lg;
+DROP TABLE t;
+#
+# INSERT ... ON DUPLICATE KEY UPDATE
+#
+CREATE TABLE t (pk INT PRIMARY KEY, f TEXT)
+ENGINE=HEAP WITH SYSTEM VERSIONING CHARACTER SET latin1;
+CREATE TRIGGER trg AFTER UPDATE ON t FOR EACH ROW
+INSERT INTO lg (ln, ok_val)
+VALUES (LENGTH(OLD.f),
+OLD.f = CASE OLD.pk WHEN 1 THEN REPEAT('abcdefgh',8)
+ELSE REPEAT('ijklmnop',80) END);
+INSERT INTO t VALUES (1, REPEAT('abcdefgh',8)), (2, REPEAT('ijklmnop',80));
+INSERT INTO t VALUES (1, 'ignored'), (2, 'ignored')
+ON DUPLICATE KEY UPDATE f = CONCAT('new', pk);
+SELECT ln, ok_val FROM lg ORDER BY id;
+ln	ok_val
+64	1
+640	1
+DELETE FROM lg;
+DROP TABLE t;
+#
+# Multi-table UPDATE, deferred do_updates() path
+#
+CREATE TABLE drv (pk INT PRIMARY KEY, c INT) ENGINE=HEAP;
+CREATE TABLE t (pk INT PRIMARY KEY, f TEXT)
+ENGINE=HEAP WITH SYSTEM VERSIONING CHARACTER SET latin1;
+CREATE TRIGGER trg AFTER UPDATE ON t FOR EACH ROW
+INSERT INTO lg (ln, ok_val)
+VALUES (LENGTH(OLD.f),
+OLD.f = CASE OLD.pk WHEN 1 THEN REPEAT('abcdefgh',8)
+ELSE REPEAT('ijklmnop',80) END);
+INSERT INTO drv VALUES (1, 10), (2, 20);
+INSERT INTO t VALUES (1, REPEAT('abcdefgh',8)), (2, REPEAT('ijklmnop',80));
+UPDATE drv STRAIGHT_JOIN t ON drv.pk = t.pk SET t.f = CONCAT('new', t.pk);
+SELECT ln, ok_val FROM lg ORDER BY id;
+ln	ok_val
+64	1
+640	1
+DELETE FROM lg;
+DROP TABLE drv, t;
+#
+# Non-versioned HEAP control: no history row is written, so nothing
+# redeems the parking while the trigger is still reading
+#
+CREATE TABLE t (pk INT PRIMARY KEY, f TEXT) ENGINE=HEAP CHARACTER SET latin1;
+CREATE TRIGGER trg AFTER UPDATE ON t FOR EACH ROW
+INSERT INTO lg (ln, ok_val)
+VALUES (LENGTH(OLD.f), OLD.f = REPEAT('abcdefgh',80));
+INSERT INTO t VALUES (1, REPEAT('abcdefgh',80)), (2, REPEAT('abcdefgh',80));
+UPDATE t SET f = CONCAT('new', pk);
+SELECT ln, ok_val FROM lg ORDER BY id;
+ln	ok_val
+640	1
+640	1
+DELETE FROM lg;
+DROP TABLE t;
+#
+# Versioned MyISAM control: same statement shape, different engine
+#
+CREATE TABLE t (pk INT PRIMARY KEY, f TEXT)
+ENGINE=MyISAM WITH SYSTEM VERSIONING CHARACTER SET latin1;
+CREATE TRIGGER trg AFTER UPDATE ON t FOR EACH ROW
+INSERT INTO lg (ln, ok_val)
+VALUES (LENGTH(OLD.f), OLD.f = REPEAT('abcdefgh',80));
+INSERT INTO t VALUES (1, REPEAT('abcdefgh',80)), (2, REPEAT('abcdefgh',80));
+UPDATE t SET f = CONCAT('new', pk);
+SELECT ln, ok_val FROM lg ORDER BY id;
+ln	ok_val
+640	1
+640	1
+DROP TABLE t;
+DROP TABLE lg;
diff --git a/mysql-test/suite/heap/blob_vers_trigger.test b/mysql-test/suite/heap/blob_vers_trigger.test
new file mode 100644
index 0000000000000..f856a3b5a81ba
--- /dev/null
+++ b/mysql-test/suite/heap/blob_vers_trigger.test
@@ -0,0 +1,152 @@
+--echo #
+--echo # An AFTER UPDATE trigger reads OLD.<blob> from the same record buffer
+--echo # that vers_insert_history_row() builds the history row from.
+--echo #
+--echo # heap_update() parks the old blob chain instead of freeing it, so that
+--echo # buffer keeps a valid zero-copy pointer for the rest of the statement.
+--echo # The history-row write redeems that parking, so it must not leave the
+--echo # buffer pointing at records handed back to the free list -- otherwise
+--echo # OLD.<blob> reads whatever was written over them, and that value is
+--echo # what lands in the audit table and in the binlog.
+--echo #
+
+CREATE TABLE lg (id INT AUTO_INCREMENT PRIMARY KEY, ln INT, ok_val INT)
+  ENGINE=MyISAM;
+
+--echo #
+--echo # Single-table UPDATE, blob sizes on either side of a single chain run
+--echo #
+CREATE TABLE t (pk INT PRIMARY KEY, f TEXT)
+  ENGINE=HEAP WITH SYSTEM VERSIONING CHARACTER SET latin1;
+CREATE TRIGGER trg AFTER UPDATE ON t FOR EACH ROW
+  INSERT INTO lg (ln, ok_val)
+  VALUES (LENGTH(OLD.f),
+          OLD.f = CASE OLD.pk WHEN 1 THEN REPEAT('abcdefgh',8)
+                              ELSE REPEAT('ijklmnop',80) END);
+
+INSERT INTO t VALUES (1, REPEAT('abcdefgh',8)), (2, REPEAT('ijklmnop',80));
+UPDATE t SET f = CONCAT('new', pk);
+
+--echo # OLD.<blob> as the trigger saw it
+SELECT ln, ok_val FROM lg ORDER BY id;
+--echo # and the history rows themselves
+SELECT pk, LENGTH(f),
+       f = CASE pk WHEN 1 THEN REPEAT('abcdefgh',8)
+                   ELSE REPEAT('ijklmnop',80) END AS hist_ok
+  FROM t FOR SYSTEM_TIME ALL WHERE f NOT LIKE 'new%' ORDER BY pk;
+DELETE FROM lg;
+DROP TABLE t;
+
+--echo #
+--echo # Multiple blob columns
+--echo #
+CREATE TABLE t (pk INT PRIMARY KEY, b1 TEXT, b2 MEDIUMTEXT)
+  ENGINE=HEAP WITH SYSTEM VERSIONING CHARACTER SET latin1;
+CREATE TRIGGER trg AFTER UPDATE ON t FOR EACH ROW
+  INSERT INTO lg (ln, ok_val)
+  VALUES (LENGTH(OLD.b1) + LENGTH(OLD.b2),
+          OLD.b1 = CASE OLD.pk WHEN 1 THEN REPEAT('abcdefgh',50)
+                               ELSE REPEAT('qrstuvwx',25) END
+          AND
+          OLD.b2 = CASE OLD.pk WHEN 1 THEN REPEAT('ABCDEFGH',150)
+                               ELSE REPEAT('ijklmnop',400) END);
+
+INSERT INTO t VALUES (1, REPEAT('abcdefgh',50), REPEAT('ABCDEFGH',150)),
+                     (2, REPEAT('qrstuvwx',25), REPEAT('ijklmnop',400));
+UPDATE t SET b1 = REPEAT('z',900), b2 = 'y';
+
+SELECT ln, ok_val FROM lg ORDER BY id;
+DELETE FROM lg;
+DROP TABLE t;
+
+--echo #
+--echo # BEFORE UPDATE runs ahead of the history-row write and must be
+--echo # unaffected; AFTER UPDATE on the same table must agree with it
+--echo #
+CREATE TABLE t (pk INT PRIMARY KEY, f TEXT)
+  ENGINE=HEAP WITH SYSTEM VERSIONING CHARACTER SET latin1;
+CREATE TRIGGER trg_b BEFORE UPDATE ON t FOR EACH ROW
+  INSERT INTO lg (ln, ok_val)
+  VALUES (LENGTH(OLD.f), OLD.f = REPEAT('abcdefgh',80));
+CREATE TRIGGER trg_a AFTER UPDATE ON t FOR EACH ROW
+  INSERT INTO lg (ln, ok_val)
+  VALUES (LENGTH(OLD.f), OLD.f = REPEAT('abcdefgh',80));
+
+INSERT INTO t VALUES (1, REPEAT('abcdefgh',80)), (2, REPEAT('abcdefgh',80));
+UPDATE t SET f = CONCAT('new', pk);
+
+SELECT ln, ok_val FROM lg ORDER BY id;
+DELETE FROM lg;
+DROP TABLE t;
+
+--echo #
+--echo # INSERT ... ON DUPLICATE KEY UPDATE
+--echo #
+CREATE TABLE t (pk INT PRIMARY KEY, f TEXT)
+  ENGINE=HEAP WITH SYSTEM VERSIONING CHARACTER SET latin1;
+CREATE TRIGGER trg AFTER UPDATE ON t FOR EACH ROW
+  INSERT INTO lg (ln, ok_val)
+  VALUES (LENGTH(OLD.f),
+          OLD.f = CASE OLD.pk WHEN 1 THEN REPEAT('abcdefgh',8)
+                              ELSE REPEAT('ijklmnop',80) END);
+
+INSERT INTO t VALUES (1, REPEAT('abcdefgh',8)), (2, REPEAT('ijklmnop',80));
+INSERT INTO t VALUES (1, 'ignored'), (2, 'ignored')
+  ON DUPLICATE KEY UPDATE f = CONCAT('new', pk);
+
+SELECT ln, ok_val FROM lg ORDER BY id;
+DELETE FROM lg;
+DROP TABLE t;
+
+--echo #
+--echo # Multi-table UPDATE, deferred do_updates() path
+--echo #
+CREATE TABLE drv (pk INT PRIMARY KEY, c INT) ENGINE=HEAP;
+CREATE TABLE t (pk INT PRIMARY KEY, f TEXT)
+  ENGINE=HEAP WITH SYSTEM VERSIONING CHARACTER SET latin1;
+CREATE TRIGGER trg AFTER UPDATE ON t FOR EACH ROW
+  INSERT INTO lg (ln, ok_val)
+  VALUES (LENGTH(OLD.f),
+          OLD.f = CASE OLD.pk WHEN 1 THEN REPEAT('abcdefgh',8)
+                              ELSE REPEAT('ijklmnop',80) END);
+
+INSERT INTO drv VALUES (1, 10), (2, 20);
+INSERT INTO t VALUES (1, REPEAT('abcdefgh',8)), (2, REPEAT('ijklmnop',80));
+UPDATE drv STRAIGHT_JOIN t ON drv.pk = t.pk SET t.f = CONCAT('new', t.pk);
+
+SELECT ln, ok_val FROM lg ORDER BY id;
+DELETE FROM lg;
+DROP TABLE drv, t;
+
+--echo #
+--echo # Non-versioned HEAP control: no history row is written, so nothing
+--echo # redeems the parking while the trigger is still reading
+--echo #
+CREATE TABLE t (pk INT PRIMARY KEY, f TEXT) ENGINE=HEAP CHARACTER SET latin1;
+CREATE TRIGGER trg AFTER UPDATE ON t FOR EACH ROW
+  INSERT INTO lg (ln, ok_val)
+  VALUES (LENGTH(OLD.f), OLD.f = REPEAT('abcdefgh',80));
+
+INSERT INTO t VALUES (1, REPEAT('abcdefgh',80)), (2, REPEAT('abcdefgh',80));
+UPDATE t SET f = CONCAT('new', pk);
+
+SELECT ln, ok_val FROM lg ORDER BY id;
+DELETE FROM lg;
+DROP TABLE t;
+
+--echo #
+--echo # Versioned MyISAM control: same statement shape, different engine
+--echo #
+CREATE TABLE t (pk INT PRIMARY KEY, f TEXT)
+  ENGINE=MyISAM WITH SYSTEM VERSIONING CHARACTER SET latin1;
+CREATE TRIGGER trg AFTER UPDATE ON t FOR EACH ROW
+  INSERT INTO lg (ln, ok_val)
+  VALUES (LENGTH(OLD.f), OLD.f = REPEAT('abcdefgh',80));
+
+INSERT INTO t VALUES (1, REPEAT('abcdefgh',80)), (2, REPEAT('abcdefgh',80));
+UPDATE t SET f = CONCAT('new', pk);
+
+SELECT ln, ok_val FROM lg ORDER BY id;
+DROP TABLE t;
+
+DROP TABLE lg;
diff --git a/mysql-test/suite/heap/blob_versioning.result b/mysql-test/suite/heap/blob_versioning.result
new file mode 100644
index 0000000000000..166165d8453c8
--- /dev/null
+++ b/mysql-test/suite/heap/blob_versioning.result
@@ -0,0 +1,209 @@
+#
+# UPDATE on a system-versioned HEAP table with blobs must preserve the
+# historical row's blob data.
+#
+# vers_insert_history_row() does restore_record(table, record[1]), which
+# copies record[1] verbatim into record[0] -- including the blob data
+# pointer.  For HEAP that pointer is a zero-copy pointer straight into
+# HP_BLOCK.  The following ha_write_row() flushes the deferred free of
+# the old chain and then allocates the history row's chain from the very
+# records the source pointer still refers to, so source and destination
+# of the blob copy overlap.
+#
+#
+# Reported repro: uniform payload
+#
+CREATE TABLE t (f TEXT) ENGINE=HEAP WITH SYSTEM VERSIONING CHARACTER SET latin1;
+INSERT INTO t VALUES (REPEAT('x',64));
+UPDATE t SET f = 'foo';
+SELECT f, LENGTH(f) FROM t;
+f	LENGTH(f)
+foo	3
+SELECT LENGTH(f), f = REPEAT('x',64) AS hist_ok
+FROM t FOR SYSTEM_TIME ALL WHERE f <> 'foo';
+LENGTH(f)	hist_ok
+64	1
+DROP TABLE t;
+#
+# Non-uniform payload makes an overlapping copy visible as wrong data
+# rather than accidentally-correct repeated bytes.
+#
+CREATE TABLE t (f TEXT) ENGINE=HEAP WITH SYSTEM VERSIONING CHARACTER SET latin1;
+INSERT INTO t VALUES (REPEAT('abcdefgh',8));
+UPDATE t SET f = 'foo';
+SELECT LENGTH(f), f = REPEAT('abcdefgh',8) AS hist_ok
+FROM t FOR SYSTEM_TIME ALL WHERE f <> 'foo';
+LENGTH(f)	hist_ok
+64	1
+DROP TABLE t;
+#
+# Blob sizes exercising the single-record run, the multi-record
+# zero-copy run and the bulk inner-record copy
+#
+CREATE TABLE t (pk INT PRIMARY KEY, f TEXT)
+ENGINE=HEAP WITH SYSTEM VERSIONING CHARACTER SET latin1;
+INSERT INTO t VALUES (1, REPEAT('abcdefgh',2)), (2, REPEAT('ijklmnop',80));
+UPDATE t SET f = CONCAT('new', pk);
+SELECT pk, LENGTH(f) FROM t ORDER BY pk;
+pk	LENGTH(f)
+1	4
+2	4
+SELECT pk, LENGTH(f),
+f = REPEAT('abcdefgh',2) OR f = REPEAT('ijklmnop',80) AS hist_ok
+FROM t FOR SYSTEM_TIME ALL WHERE f NOT LIKE 'new%' ORDER BY pk;
+pk	LENGTH(f)	hist_ok
+1	16	1
+2	640	1
+# Second generation of history rows over the same free-list churn
+UPDATE t SET f = REPEAT('qrstuvwx',300);
+UPDATE t SET f = CONCAT('final', pk);
+SELECT pk, LENGTH(f),
+f = REPEAT('qrstuvwx',300) AS gen2_ok
+FROM t FOR SYSTEM_TIME ALL WHERE f LIKE 'qrstuvwx%' ORDER BY pk;
+pk	LENGTH(f)	gen2_ok
+1	2400	1
+2	2400	1
+DROP TABLE t;
+#
+# Multiple blob columns
+#
+CREATE TABLE t (pk INT PRIMARY KEY, b1 TEXT, b2 MEDIUMTEXT)
+ENGINE=HEAP WITH SYSTEM VERSIONING CHARACTER SET latin1;
+INSERT INTO t VALUES (1, REPEAT('abcdefgh',50), REPEAT('ABCDEFGH',150)),
+(2, REPEAT('12345678',25), REPEAT('!@#$%^&*',400));
+UPDATE t SET b1 = 'x', b2 = 'y';
+SELECT pk,
+b1 = CASE pk WHEN 1 THEN REPEAT('abcdefgh',50)
+ELSE REPEAT('12345678',25) END AS b1_ok,
+b2 = CASE pk WHEN 1 THEN REPEAT('ABCDEFGH',150)
+ELSE REPEAT('!@#$%^&*',400) END AS b2_ok
+FROM t FOR SYSTEM_TIME ALL WHERE b1 <> 'x' ORDER BY pk;
+pk	b1_ok	b2_ok
+1	1	1
+2	1	1
+DROP TABLE t;
+#
+# Only one of two blob columns changed.  heap_update() parks a chain
+# for the changed column and leaves the other slot NULL while
+# has_pending_blob_free stays set, so the history-row write sees a mix
+# of parked and empty slots.
+#
+CREATE TABLE t (pk INT PRIMARY KEY, b1 TEXT, b2 MEDIUMTEXT)
+ENGINE=HEAP WITH SYSTEM VERSIONING CHARACTER SET latin1;
+INSERT INTO t VALUES (1, REPEAT('abcdefgh',50), REPEAT('ABCDEFGH',150)),
+(2, REPEAT('12345678',25), REPEAT('!@#$%^&*',400));
+UPDATE t SET b1 = 'x';
+SELECT pk, LENGTH(b1), LENGTH(b2),
+b1 = CASE pk WHEN 1 THEN REPEAT('abcdefgh',50)
+ELSE REPEAT('12345678',25) END AS b1_ok,
+b2 = CASE pk WHEN 1 THEN REPEAT('ABCDEFGH',150)
+ELSE REPEAT('!@#$%^&*',400) END AS b2_ok
+FROM t FOR SYSTEM_TIME ALL WHERE b1 <> 'x' ORDER BY pk;
+pk	LENGTH(b1)	LENGTH(b2)	b1_ok	b2_ok
+1	400	1200	1	1
+2	200	3200	1	1
+# Emptying the other column, so a slot goes from parked to zero-length
+UPDATE t SET b2 = '';
+UPDATE t SET b1 = 'zz';
+SELECT pk, LENGTH(b1), LENGTH(b2)
+FROM t FOR SYSTEM_TIME ALL ORDER BY pk, LENGTH(b1), LENGTH(b2);
+pk	LENGTH(b1)	LENGTH(b2)
+1	1	0
+1	1	1200
+1	2	0
+1	400	1200
+2	1	0
+2	1	3200
+2	2	0
+2	200	3200
+DROP TABLE t;
+#
+# Shrinking a blob in place.  LEFT(f,N) leaves the new value pointing
+# at the old data, so the row written by heap_update() and the history
+# row built from record[1] carry the same pointer with different
+# lengths.  Chain layout is a property of the chain's flags, not of the
+# length, so a short read off an adopted chain is still correct.
+#
+CREATE TABLE t (pk INT AUTO_INCREMENT PRIMARY KEY, f TEXT)
+ENGINE=HEAP WITH SYSTEM VERSIONING CHARACTER SET latin1;
+INSERT INTO t (f) VALUES (REPEAT('hello',100)), (REPEAT('World',100));
+UPDATE t SET f = LEFT(f,6);
+SELECT pk, f, LENGTH(f) FROM t ORDER BY pk;
+pk	f	LENGTH(f)
+1	helloh	6
+2	WorldW	6
+SELECT pk, LENGTH(f),
+f = CASE pk WHEN 1 THEN REPEAT('hello',100)
+ELSE REPEAT('World',100) END AS hist_ok
+FROM t FOR SYSTEM_TIME ALL WHERE LENGTH(f) > 6 ORDER BY pk;
+pk	LENGTH(f)	hist_ok
+1	500	1
+2	500	1
+# Shrinking again, now off a chain the previous generation adopted
+UPDATE t SET f = LEFT(f,3);
+SELECT pk, f, LENGTH(f) FROM t ORDER BY pk;
+pk	f	LENGTH(f)
+1	hel	3
+2	Wor	3
+SELECT pk, LENGTH(f), f AS gen2_ok
+FROM t FOR SYSTEM_TIME ALL WHERE LENGTH(f) = 6 ORDER BY pk;
+pk	LENGTH(f)	gen2_ok
+1	6	helloh
+2	6	WorldW
+DROP TABLE t;
+#
+# Indexed table.  heap_write() writes every key before hp_write_blobs()
+# adopts the parked chain, so the key loop runs while record[1] still
+# holds a zero-copy pointer into that chain.  A UNIQUE key additionally
+# makes the history-row write compare against stored rows, which
+# materializes their blobs into key_blob_buff -- a different buffer from
+# the blob_buff record[1] may itself be pointing into.
+#
+CREATE TABLE t (pk INT PRIMARY KEY, u INT, s INT, f TEXT,
+UNIQUE (u), KEY (s), KEY USING BTREE (s, u))
+ENGINE=HEAP WITH SYSTEM VERSIONING CHARACTER SET latin1;
+INSERT INTO t VALUES (1, 10, 100, REPEAT('abcdefgh',9)),
+(2, 20, 100, REPEAT('ijklmnop',40));
+UPDATE t SET f = CONCAT('new', pk);
+SELECT pk, u, s, LENGTH(f) FROM t ORDER BY pk;
+pk	u	s	LENGTH(f)
+1	10	100	4
+2	20	100	4
+SELECT pk, LENGTH(f),
+f = CASE pk WHEN 1 THEN REPEAT('abcdefgh',9)
+ELSE REPEAT('ijklmnop',40) END AS hist_ok
+FROM t FOR SYSTEM_TIME ALL WHERE f NOT LIKE 'new%' ORDER BY pk;
+pk	LENGTH(f)	hist_ok
+1	72	1
+2	320	1
+# The history rows are reachable through each index with blobs intact
+SELECT pk, LENGTH(f), f = REPEAT('abcdefgh',9) AS hist_ok
+FROM t FOR SYSTEM_TIME ALL WHERE u = 10 AND f NOT LIKE 'new%';
+pk	LENGTH(f)	hist_ok
+1	72	1
+SELECT pk, LENGTH(f), f = REPEAT('ijklmnop',40) AS hist_ok
+FROM t FOR SYSTEM_TIME ALL WHERE s = 100 AND pk = 2 AND f NOT LIKE 'new%';
+pk	LENGTH(f)	hist_ok
+2	320	1
+# Changing an indexed column and the blob in one UPDATE, so the key
+# delete/rewrite in heap_update() coincides with parking the chain
+UPDATE t SET u = u + 5, f = REPEAT('qrstuvwx',20) WHERE pk = 1;
+SELECT pk, u, LENGTH(f), f = REPEAT('qrstuvwx',20) AS cur_ok
+FROM t WHERE pk = 1;
+pk	u	LENGTH(f)	cur_ok
+1	15	160	1
+SELECT pk, u, f = CONCAT('new', pk) AS hist_ok
+FROM t FOR SYSTEM_TIME ALL WHERE pk = 1 AND f LIKE 'new%';
+pk	u	hist_ok
+1	10	1
+DROP TABLE t;
+#
+# A blob cannot itself be indexed in HEAP, so there is no blob-as-key
+# variant of the above to cover.
+#
+CREATE TABLE t (pk INT PRIMARY KEY, f TEXT, KEY (f(10)))
+ENGINE=HEAP WITH SYSTEM VERSIONING;
+ERROR 42000: BLOB column `f` can't be used in key specification in the MEMORY table
+CREATE TABLE t (pk INT PRIMARY KEY, f TEXT, UNIQUE (f))
+ENGINE=HEAP WITH SYSTEM VERSIONING;
+ERROR 42000: BLOB column `f` can't be used in key specification in the MEMORY table
diff --git a/mysql-test/suite/heap/blob_versioning.test b/mysql-test/suite/heap/blob_versioning.test
new file mode 100644
index 0000000000000..810849d992cbc
--- /dev/null
+++ b/mysql-test/suite/heap/blob_versioning.test
@@ -0,0 +1,168 @@
+--echo #
+--echo # UPDATE on a system-versioned HEAP table with blobs must preserve the
+--echo # historical row's blob data.
+--echo #
+--echo # vers_insert_history_row() does restore_record(table, record[1]), which
+--echo # copies record[1] verbatim into record[0] -- including the blob data
+--echo # pointer.  For HEAP that pointer is a zero-copy pointer straight into
+--echo # HP_BLOCK.  The following ha_write_row() flushes the deferred free of
+--echo # the old chain and then allocates the history row's chain from the very
+--echo # records the source pointer still refers to, so source and destination
+--echo # of the blob copy overlap.
+--echo #
+
+--echo #
+--echo # Reported repro: uniform payload
+--echo #
+CREATE TABLE t (f TEXT) ENGINE=HEAP WITH SYSTEM VERSIONING CHARACTER SET latin1;
+INSERT INTO t VALUES (REPEAT('x',64));
+UPDATE t SET f = 'foo';
+SELECT f, LENGTH(f) FROM t;
+SELECT LENGTH(f), f = REPEAT('x',64) AS hist_ok
+  FROM t FOR SYSTEM_TIME ALL WHERE f <> 'foo';
+DROP TABLE t;
+
+--echo #
+--echo # Non-uniform payload makes an overlapping copy visible as wrong data
+--echo # rather than accidentally-correct repeated bytes.
+--echo #
+CREATE TABLE t (f TEXT) ENGINE=HEAP WITH SYSTEM VERSIONING CHARACTER SET latin1;
+INSERT INTO t VALUES (REPEAT('abcdefgh',8));
+UPDATE t SET f = 'foo';
+SELECT LENGTH(f), f = REPEAT('abcdefgh',8) AS hist_ok
+  FROM t FOR SYSTEM_TIME ALL WHERE f <> 'foo';
+DROP TABLE t;
+
+--echo #
+--echo # Blob sizes exercising the single-record run, the multi-record
+--echo # zero-copy run and the bulk inner-record copy
+--echo #
+CREATE TABLE t (pk INT PRIMARY KEY, f TEXT)
+  ENGINE=HEAP WITH SYSTEM VERSIONING CHARACTER SET latin1;
+
+INSERT INTO t VALUES (1, REPEAT('abcdefgh',2)), (2, REPEAT('ijklmnop',80));
+UPDATE t SET f = CONCAT('new', pk);
+SELECT pk, LENGTH(f) FROM t ORDER BY pk;
+SELECT pk, LENGTH(f),
+       f = REPEAT('abcdefgh',2) OR f = REPEAT('ijklmnop',80) AS hist_ok
+  FROM t FOR SYSTEM_TIME ALL WHERE f NOT LIKE 'new%' ORDER BY pk;
+
+--echo # Second generation of history rows over the same free-list churn
+UPDATE t SET f = REPEAT('qrstuvwx',300);
+UPDATE t SET f = CONCAT('final', pk);
+SELECT pk, LENGTH(f),
+       f = REPEAT('qrstuvwx',300) AS gen2_ok
+  FROM t FOR SYSTEM_TIME ALL WHERE f LIKE 'qrstuvwx%' ORDER BY pk;
+
+DROP TABLE t;
+
+--echo #
+--echo # Multiple blob columns
+--echo #
+CREATE TABLE t (pk INT PRIMARY KEY, b1 TEXT, b2 MEDIUMTEXT)
+  ENGINE=HEAP WITH SYSTEM VERSIONING CHARACTER SET latin1;
+INSERT INTO t VALUES (1, REPEAT('abcdefgh',50), REPEAT('ABCDEFGH',150)),
+                     (2, REPEAT('12345678',25), REPEAT('!@#$%^&*',400));
+UPDATE t SET b1 = 'x', b2 = 'y';
+SELECT pk,
+       b1 = CASE pk WHEN 1 THEN REPEAT('abcdefgh',50)
+                    ELSE REPEAT('12345678',25) END AS b1_ok,
+       b2 = CASE pk WHEN 1 THEN REPEAT('ABCDEFGH',150)
+                    ELSE REPEAT('!@#$%^&*',400) END AS b2_ok
+  FROM t FOR SYSTEM_TIME ALL WHERE b1 <> 'x' ORDER BY pk;
+DROP TABLE t;
+
+--echo #
+--echo # Only one of two blob columns changed.  heap_update() parks a chain
+--echo # for the changed column and leaves the other slot NULL while
+--echo # has_pending_blob_free stays set, so the history-row write sees a mix
+--echo # of parked and empty slots.
+--echo #
+CREATE TABLE t (pk INT PRIMARY KEY, b1 TEXT, b2 MEDIUMTEXT)
+  ENGINE=HEAP WITH SYSTEM VERSIONING CHARACTER SET latin1;
+INSERT INTO t VALUES (1, REPEAT('abcdefgh',50), REPEAT('ABCDEFGH',150)),
+                     (2, REPEAT('12345678',25), REPEAT('!@#$%^&*',400));
+UPDATE t SET b1 = 'x';
+SELECT pk, LENGTH(b1), LENGTH(b2),
+       b1 = CASE pk WHEN 1 THEN REPEAT('abcdefgh',50)
+                    ELSE REPEAT('12345678',25) END AS b1_ok,
+       b2 = CASE pk WHEN 1 THEN REPEAT('ABCDEFGH',150)
+                    ELSE REPEAT('!@#$%^&*',400) END AS b2_ok
+  FROM t FOR SYSTEM_TIME ALL WHERE b1 <> 'x' ORDER BY pk;
+
+--echo # Emptying the other column, so a slot goes from parked to zero-length
+UPDATE t SET b2 = '';
+UPDATE t SET b1 = 'zz';
+SELECT pk, LENGTH(b1), LENGTH(b2)
+  FROM t FOR SYSTEM_TIME ALL ORDER BY pk, LENGTH(b1), LENGTH(b2);
+DROP TABLE t;
+
+--echo #
+--echo # Shrinking a blob in place.  LEFT(f,N) leaves the new value pointing
+--echo # at the old data, so the row written by heap_update() and the history
+--echo # row built from record[1] carry the same pointer with different
+--echo # lengths.  Chain layout is a property of the chain's flags, not of the
+--echo # length, so a short read off an adopted chain is still correct.
+--echo #
+CREATE TABLE t (pk INT AUTO_INCREMENT PRIMARY KEY, f TEXT)
+  ENGINE=HEAP WITH SYSTEM VERSIONING CHARACTER SET latin1;
+INSERT INTO t (f) VALUES (REPEAT('hello',100)), (REPEAT('World',100));
+UPDATE t SET f = LEFT(f,6);
+SELECT pk, f, LENGTH(f) FROM t ORDER BY pk;
+SELECT pk, LENGTH(f),
+       f = CASE pk WHEN 1 THEN REPEAT('hello',100)
+                   ELSE REPEAT('World',100) END AS hist_ok
+  FROM t FOR SYSTEM_TIME ALL WHERE LENGTH(f) > 6 ORDER BY pk;
+
+--echo # Shrinking again, now off a chain the previous generation adopted
+UPDATE t SET f = LEFT(f,3);
+SELECT pk, f, LENGTH(f) FROM t ORDER BY pk;
+SELECT pk, LENGTH(f), f AS gen2_ok
+  FROM t FOR SYSTEM_TIME ALL WHERE LENGTH(f) = 6 ORDER BY pk;
+DROP TABLE t;
+
+--echo #
+--echo # Indexed table.  heap_write() writes every key before hp_write_blobs()
+--echo # adopts the parked chain, so the key loop runs while record[1] still
+--echo # holds a zero-copy pointer into that chain.  A UNIQUE key additionally
+--echo # makes the history-row write compare against stored rows, which
+--echo # materializes their blobs into key_blob_buff -- a different buffer from
+--echo # the blob_buff record[1] may itself be pointing into.
+--echo #
+CREATE TABLE t (pk INT PRIMARY KEY, u INT, s INT, f TEXT,
+                UNIQUE (u), KEY (s), KEY USING BTREE (s, u))
+  ENGINE=HEAP WITH SYSTEM VERSIONING CHARACTER SET latin1;
+INSERT INTO t VALUES (1, 10, 100, REPEAT('abcdefgh',9)),
+                     (2, 20, 100, REPEAT('ijklmnop',40));
+UPDATE t SET f = CONCAT('new', pk);
+SELECT pk, u, s, LENGTH(f) FROM t ORDER BY pk;
+SELECT pk, LENGTH(f),
+       f = CASE pk WHEN 1 THEN REPEAT('abcdefgh',9)
+                   ELSE REPEAT('ijklmnop',40) END AS hist_ok
+  FROM t FOR SYSTEM_TIME ALL WHERE f NOT LIKE 'new%' ORDER BY pk;
+
+--echo # The history rows are reachable through each index with blobs intact
+SELECT pk, LENGTH(f), f = REPEAT('abcdefgh',9) AS hist_ok
+  FROM t FOR SYSTEM_TIME ALL WHERE u = 10 AND f NOT LIKE 'new%';
+SELECT pk, LENGTH(f), f = REPEAT('ijklmnop',40) AS hist_ok
+  FROM t FOR SYSTEM_TIME ALL WHERE s = 100 AND pk = 2 AND f NOT LIKE 'new%';
+
+--echo # Changing an indexed column and the blob in one UPDATE, so the key
+--echo # delete/rewrite in heap_update() coincides with parking the chain
+UPDATE t SET u = u + 5, f = REPEAT('qrstuvwx',20) WHERE pk = 1;
+SELECT pk, u, LENGTH(f), f = REPEAT('qrstuvwx',20) AS cur_ok
+  FROM t WHERE pk = 1;
+SELECT pk, u, f = CONCAT('new', pk) AS hist_ok
+  FROM t FOR SYSTEM_TIME ALL WHERE pk = 1 AND f LIKE 'new%';
+DROP TABLE t;
+
+--echo #
+--echo # A blob cannot itself be indexed in HEAP, so there is no blob-as-key
+--echo # variant of the above to cover.
+--echo #
+--error ER_BLOB_USED_AS_KEY
+CREATE TABLE t (pk INT PRIMARY KEY, f TEXT, KEY (f(10)))
+  ENGINE=HEAP WITH SYSTEM VERSIONING;
+--error ER_BLOB_USED_AS_KEY
+CREATE TABLE t (pk INT PRIMARY KEY, f TEXT, UNIQUE (f))
+  ENGINE=HEAP WITH SYSTEM VERSIONING;
diff --git a/storage/heap/CMakeLists.txt b/storage/heap/CMakeLists.txt
index d46669d5db757..6501ae2f0ed4e 100644
--- a/storage/heap/CMakeLists.txt
+++ b/storage/heap/CMakeLists.txt
@@ -32,7 +32,7 @@ IF(WITH_UNIT_TESTS)
   TARGET_LINK_LIBRARIES(hp_test1 heap mysys dbug strings)
   ADD_EXECUTABLE(hp_test2 hp_test2.c)
   TARGET_LINK_LIBRARIES(hp_test2 heap mysys dbug strings)
-  MY_ADD_TESTS(hp_test_hash hp_test_scan hp_test_freelist hp_test_concurrent LINK_LIBRARIES heap mysys dbug strings)
+  MY_ADD_TESTS(hp_test_hash hp_test_scan hp_test_freelist hp_test_concurrent hp_test_blob_alias LINK_LIBRARIES heap mysys dbug strings)
 
   INCLUDE_DIRECTORIES(${CMAKE_SOURCE_DIR}/sql
                       ${CMAKE_SOURCE_DIR}/include)
diff --git a/storage/heap/heapdef.h b/storage/heap/heapdef.h
index 1aafee0b00362..f157bcc77e9b9 100644
--- a/storage/heap/heapdef.h
+++ b/storage/heap/heapdef.h
@@ -383,6 +383,7 @@ extern void hp_free_blobs(HP_SHARE *share, uchar *pos);
 extern void hp_free_run_chain(HP_SHARE *share, uchar *chain);
 extern void hp_shrink_tail(HP_SHARE *share);
 extern void hp_flush_pending_blob_free_impl(HP_INFO *info);
+extern void hp_flush_unaliased_blob_free(HP_INFO *info, const uchar *record);
 
 static inline void hp_flush_pending_blob_free(HP_INFO *info)
 {
@@ -390,6 +391,26 @@ static inline void hp_flush_pending_blob_free(HP_INFO *info)
     hp_flush_pending_blob_free_impl(info);
 }
 
+/*
+  Does a record's blob data live in `chain`?
+
+  hp_read_blobs() hands out zero-copy pointers of exactly two forms: the chain
+  head itself (data at offset 0 of the first record) or chain + recbuffer
+  (data contiguous from the second record onwards).  A multi-run chain is
+  reassembled into info->blob_buff instead and can never alias.
+
+  A parked chain slot may be NULL (heap_update() parks a chain only for the
+  blob columns it changed), so callers check that before asking.
+*/
+
+static inline my_bool hp_blob_sources_chain(HP_SHARE *share,
+                                            const uchar *data_ptr,
+                                            const uchar *chain)
+{
+  DBUG_ASSERT(chain);
+  return (data_ptr == chain || data_ptr == chain + share->block.recbuffer);
+}
+
 extern mysql_mutex_t THR_LOCK_heap;
 
 extern PSI_memory_key hp_key_memory_HP_SHARE;
diff --git a/storage/heap/hp_blob.c b/storage/heap/hp_blob.c
index d004c26ff1ee7..78b42c7a39f65 100644
--- a/storage/heap/hp_blob.c
+++ b/storage/heap/hp_blob.c
@@ -154,6 +154,67 @@ void hp_flush_pending_blob_free_impl(HP_INFO *info)
 }
 
 
+/*
+  Redeem the deferred blob chain frees of a previous heap_update() or
+  heap_delete(), except for chains that `record` still sources blob data from.
+
+  Those chains stay parked.  The SQL layer keeps the pre-update row in
+  record[1] for the rest of the statement -- an AFTER UPDATE trigger reads
+  OLD.<blob> from it, and the row-based binlog image is built from it -- and a
+  system-versioned UPDATE writes the history row from a verbatim copy of that
+  same buffer.  Freeing a chain both still point into would put it on the free
+  list, where the write about to follow would hand it straight back with
+  hp_push_free_block()'s links scribbled through the payload.
+
+  Keeping them costs no space: hp_write_blobs() adopts a parked chain the
+  record already sources its data from rather than allocating a second copy of
+  bytes that are already in place.
+
+  @param info   Table handle
+  @param record Record about to be written
+*/
+
+void hp_flush_unaliased_blob_free(HP_INFO *info, const uchar *record)
+{
+  HP_SHARE *share= info->s;
+  HP_BLOB_DESC *desc, *desc_end;
+  uchar **chain_pos= info->pending_blob_chains;
+  my_bool still_pending= FALSE;
+  DBUG_ASSERT(info->has_pending_blob_free);
+
+  for (desc= share->blob_descs, desc_end= desc + share->blob_count;
+       desc < desc_end; desc++, chain_pos++)
+  {
+    uint32 data_len;
+    const uchar *data_ptr;
+
+    if (!*chain_pos)
+      continue;
+
+    data_len= hp_blob_length(desc, record);
+    memcpy(&data_ptr, record + desc->offset + desc->packlength,
+           sizeof(data_ptr));
+    DBUG_ASSERT((data_len == 0) == (data_ptr == NULL));
+
+    /*
+      Test the length, not the pointer: hp_write_blobs() decides the same
+      question the same way, and the two have to agree on which columns can
+      source a parked chain.  A column they disagreed about would leave a
+      chain parked that nothing will ever adopt.
+    */
+    if (data_len != 0 && hp_blob_sources_chain(share, data_ptr, *chain_pos))
+    {
+      still_pending= TRUE;                      /* Keep for hp_write_blobs() */
+      continue;
+    }
+    hp_free_run_chain(share, *chain_pos);
+    *chain_pos= NULL;
+  }
+  hp_shrink_tail(share);
+  info->has_pending_blob_free= still_pending;
+}
+
+
 /*
   Free one continuation chain of variable-length runs.
 
@@ -557,6 +618,13 @@ int hp_write_one_blob(HP_SHARE *share, const uchar *data_ptr,
   runs, copies blob data into them, and overwrites the pointer in
   the stored row (pos) to point to the first continuation run.
 
+  A column whose data still lives in a chain parked for deferred free is
+  adopted instead: the bytes are already laid out exactly as this write would
+  lay them out, so the row takes the chain over.  That leaves the buffer the
+  caller still holds reading correct data, and needs no space -- which at
+  max_heap_table_size is the difference between the write succeeding and
+  failing.  See hp_flush_unaliased_blob_free().
+
   @param info   Table handle
   @param record Source record buffer (caller's data)
   @param pos    Destination row in HP_BLOCK (already has memcpy'd record)
@@ -569,6 +637,13 @@ int hp_write_blobs(HP_INFO *info, const uchar *record, uchar *pos)
   HP_SHARE *share= info->s;
   HP_BLOB_DESC *desc, *desc_end;
   my_bool has_blob_data= FALSE;
+  /*
+    Chains parked for deferred free, or NULL when there are none.  Internal
+    temporary tables never park -- hp_update()/hp_delete() free their chains
+    outright -- and do not even allocate the array.
+  */
+  uchar **parked= (info->has_pending_blob_free ?
+                   info->pending_blob_chains : NULL);
   DBUG_ENTER("hp_write_blobs");
 
   for (desc= share->blob_descs, desc_end= desc + share->blob_count;
@@ -590,15 +665,30 @@ int hp_write_blobs(HP_INFO *info, const uchar *record, uchar *pos)
     memcpy(&data_ptr, record + desc->offset + desc->packlength,
            sizeof(data_ptr));
 
-    if (hp_write_one_blob(share, data_ptr, data_len, &first_run))
+    if (parked && parked[desc - share->blob_descs] &&
+        hp_blob_sources_chain(share, data_ptr,
+                              parked[desc - share->blob_descs]))
+    {
+      /*
+        Adopt the parked chain: it already holds exactly these bytes.  The
+        pending slot is cleared only once every column has succeeded, so that
+        the rollback below can tell an adopted chain from an allocated one and
+        leave it parked rather than freeing it.
+      */
+      first_run= parked[desc - share->blob_descs];
+    }
+    else if (hp_write_one_blob(share, data_ptr, data_len, &first_run))
     {
-      /* Rollback: free all previously completed blob columns */
+      /*
+        Rollback: free all previously completed blob columns, leaving any
+        adopted chain parked for its original owner
+      */
       HP_BLOB_DESC *rd;
       for (rd= share->blob_descs; rd < desc; rd++)
       {
         uchar *chain;
         memcpy(&chain, pos + rd->offset + rd->packlength, sizeof(chain));
-        if (chain)
+        if (chain && (!parked || chain != parked[rd - share->blob_descs]))
           hp_free_run_chain(share, chain);
         bzero(pos + rd->offset + rd->packlength, sizeof(char*));
       }
@@ -611,6 +701,26 @@ int hp_write_blobs(HP_INFO *info, const uchar *record, uchar *pos)
            sizeof(first_run));
   }
 
+  /* Adopted chains belong to this row now, not to the deferred free */
+  if (parked)
+  {
+    /* There was at least one blob that may have been reused. Fix the chains */
+    uchar **chain_pos= parked;
+    my_bool still_pending= FALSE;
+    for (desc= share->blob_descs; desc < desc_end; desc++, chain_pos++)
+    {
+      uchar *chain;
+      if (!*chain_pos)
+        continue;
+      memcpy(&chain, pos + desc->offset + desc->packlength, sizeof(chain));
+      if (chain == *chain_pos)
+        *chain_pos= NULL;
+      else
+        still_pending= TRUE;
+    }
+    info->has_pending_blob_free= still_pending;
+  }
+
   pos[share->visible]= has_blob_data ?
     (HP_ROW_ACTIVE | HP_ROW_HAS_CONT) : HP_ROW_ACTIVE;
   DBUG_RETURN(0);
diff --git a/storage/heap/hp_delete.c b/storage/heap/hp_delete.c
index 5dbd3590c8c82..7a3b24099f27c 100644
--- a/storage/heap/hp_delete.c
+++ b/storage/heap/hp_delete.c
@@ -186,7 +186,10 @@ int heap_delete(HP_INFO *info, const uchar *record)
         zero-copy pointers dangle.
 
         Save the chain head pointers and free them on the next mutating
-        operation (write/update/delete) or on reset/close.
+        operation (write/update/delete) or on reset/close.  heap_write()
+        leaves a chain the row it writes still sources data from parked and
+        adopts it instead of reallocating; see
+        hp_flush_unaliased_blob_free().
       */
       HP_BLOB_DESC *desc;
       uint i;
diff --git a/storage/heap/hp_test_blob_alias-t.c b/storage/heap/hp_test_blob_alias-t.c
new file mode 100644
index 0000000000000..18e35825e6e3e
--- /dev/null
+++ b/storage/heap/hp_test_blob_alias-t.c
@@ -0,0 +1,343 @@
+/*
+   Unit tests for writing a record whose blob data still lives in a chain
+   parked for deferred free.
+
+   heap_update() and heap_delete() park the old blob chain instead of freeing
+   it, so the record buffer the caller read into keeps a valid zero-copy
+   pointer for the rest of the statement.  heap_write() redeems that parking.
+
+   The SQL layer depends on both halves at once.  A system-versioned UPDATE
+   calls ha_update_row() and then writes the history row from the same
+   pre-update record buffer, so heap_write() sees a record whose blob source
+   is the chain the update just parked -- while that same buffer stays live as
+   record[1], which an AFTER UPDATE trigger reads OLD.<blob> from.
+
+   These tests pin that redeeming the parking preserves the data reachable
+   through both record buffers: the one being written and the one still held
+   by the caller.
+
+   They also pin how that is done.  A record whose blob still lives in a
+   parked chain adopts the chain instead of allocating a second copy of bytes
+   that are already in place, so the write costs one record slot -- its own --
+   and no chain.  At max_heap_table_size that is the difference between the
+   write succeeding and failing, so it is checked directly here, through the
+   adopted chain pointer and the number of slots the table handed out, rather
+   than through a table filled to capacity where free space is a function of
+   block granularity and pointer width.
+*/
+
+#include "hp_test_helpers.h"
+
+#define MAX_PAYLOAD 256
+
+
+/*
+  Non-uniform payload, so that an overlapping or recycled copy shows up as
+  wrong data rather than accidentally-correct repeated bytes.
+*/
+
+static void fill_pattern(uchar *buf, uint16 len)
+{
+  uint16 i;
+  for (i= 0; i < len; i++)
+    buf[i]= (uchar) ('a' + (i % 26));
+}
+
+
+static const uchar *blob_data_ptr(const uchar *rec)
+{
+  const uchar *ptr;
+  memcpy(&ptr, rec + BLOB_OFFSET + BLOB_PACKLEN, sizeof(ptr));
+  return ptr;
+}
+
+
+static uint16 blob_data_len(const uchar *rec)
+{
+  return uint2korr(rec + BLOB_OFFSET);
+}
+
+
+static my_bool blob_matches(const uchar *rec, const uchar *expected,
+                            uint16 expected_len)
+{
+  return blob_data_len(rec) == expected_len &&
+         memcmp(blob_data_ptr(rec), expected, expected_len) == 0;
+}
+
+
+/*
+  Leave the free list holding one run that is too short for the payload that
+  follows, so its chain has to be assembled from that run plus the tail.
+  hp_read_blobs() cannot hand out a zero-copy pointer into a multi-run chain
+  and reassembles it into info->blob_buff instead.
+*/
+
+static void fragment_free_list(HP_INFO *info)
+{
+  uchar rec[REC_LENGTH];
+  uchar filler[100];
+  uchar key[4];
+
+  memset(filler, 'F', sizeof(filler));
+
+  build_record(rec, 100, filler, sizeof(filler));
+  (void) heap_write(info, rec);
+  build_record(rec, 101, filler, 1);            /* guard, keeps the run apart
+                                                   from what follows it */
+  (void) heap_write(info, rec);
+
+  int4store(key, 100);
+  if (heap_rkey(info, rec, 0, key, 4, HA_READ_KEY_EXACT) == 0)
+    (void) heap_delete(info, rec);
+  heap_reset(info);                             /* redeem the parking; the run
+                                                   is now genuinely free */
+}
+
+
+/*
+  Test: the system-versioned UPDATE sequence.
+
+    1. Insert a row, read it back.  The record buffer now holds the blob the
+       way hp_read_blobs() chose to present it -- this models record[1].
+    2. heap_update() to a shorter blob.  The old chain is parked, not freed,
+       precisely so that buffer stays readable.
+    3. heap_write() a record copied verbatim from that buffer, differing only
+       in the key.  This models vers_insert_history_row(), which does
+       restore_record(table, record[1]) and changes only row_end.
+
+  Both records must still carry the pre-update blob afterwards: the row that
+  was written (checked by reading it back) and the caller's buffer (checked
+  in place, before any further read, since a read may reuse info->blob_buff).
+
+  @param expect_alloc_growth  Record slots the history write may hand out, or
+                              -1 where the free list makes it indeterminate.
+*/
+
+static void test_write_from_parked_chain(uint16 payload_len,
+                                         my_bool fragment,
+                                         my_bool want_reassembled,
+                                         int expect_alloc_growth,
+                                         const char *case_name)
+{
+  HP_SHARE *share;
+  HP_INFO *info;
+  uchar rec[REC_LENGTH], old_rec[REC_LENGTH], new_rec[REC_LENGTH];
+  uchar hist_rec[REC_LENGTH], read_rec[REC_LENGTH];
+  uchar payload[MAX_PAYLOAD];
+  const uchar *short_blob= (const uchar*) "s";
+  const uchar *parked_chain, *stored_chain;
+  ulong alloc_before, alloc_after;
+  uchar key[4];
+
+  fill_pattern(payload, payload_len);
+
+  if (create_and_open("test_blob_alias", &share, &info))
+  {
+    ok(0, "%s: setup failed: %d", case_name, my_errno);
+    skip(12, "setup failed");
+    return;
+  }
+
+  if (fragment)
+    fragment_free_list(info);
+
+  build_record(rec, 1, payload, payload_len);
+  ok(heap_write(info, rec) == 0, "%s: inserted base row", case_name);
+
+  /* Read the row back the way the SQL layer fills record[1] */
+  int4store(key, 1);
+  ok(heap_rkey(info, old_rec, 0, key, 4, HA_READ_KEY_EXACT) == 0,
+     "%s: read base row into record[1]", case_name);
+  ok(blob_matches(old_rec, payload, payload_len),
+     "%s: record[1] blob reads correctly before the update", case_name);
+
+  /*
+    Which layout hp_read_blobs() handed out decides whether the buffer can
+    alias a chain at all: a zero-copy pointer aims into HP_BLOCK, while a
+    reassembled multi-run blob lives in info->blob_buff and never can.  Pin
+    it, so that a layout change cannot silently drop the case being covered.
+  */
+  ok((blob_data_ptr(old_rec) == info->blob_buff) == want_reassembled,
+     "%s: record[1] blob is %s as intended", case_name,
+     want_reassembled ? "reassembled into info->blob_buff"
+                      : "zero-copy into HP_BLOCK");
+
+  /* ha_update_row(): the row shrinks, so the old chain is parked */
+  memcpy(new_rec, old_rec, REC_LENGTH);
+  int2store(new_rec + BLOB_OFFSET, 1);
+  memcpy(new_rec + BLOB_OFFSET + BLOB_PACKLEN, &short_blob,
+         sizeof(short_blob));
+  ok(heap_update(info, old_rec, new_rec) == 0,
+     "%s: updated base row to a shorter blob", case_name);
+
+  /*
+    The chain just parked is the one the history row below will be asked to
+    source its blob from.  Remember it, and how many record slots the table
+    has handed out, so the write can be checked for taking the chain over
+    rather than allocating a second copy of it.
+  */
+  parked_chain= info->pending_blob_chains[0];
+  ok(parked_chain != NULL, "%s: the update parked the old chain", case_name);
+  alloc_before= (ulong) share->block.last_allocated;
+
+  /*
+    vers_insert_history_row(): the history row is record[1] verbatim, blob
+    pointer included, with only row_end (here the key) changed.
+  */
+  memcpy(hist_rec, old_rec, REC_LENGTH);
+  int4store(hist_rec + INT_OFFSET, 2);
+  ok(heap_write(info, hist_rec) == 0, "%s: wrote history row", case_name);
+  alloc_after= (ulong) share->block.last_allocated;
+
+  /*
+    record[1] stays live for the rest of the statement -- an AFTER UPDATE
+    trigger reads OLD.<blob> from it, and the row-based binlog image is built
+    from it -- so redeeming the parked chain must not cost it its data.
+    Checked before any further read, which could reuse info->blob_buff.
+  */
+  ok(blob_matches(old_rec, payload, payload_len),
+     "%s: record[1] blob still reads correctly after the history write",
+     case_name);
+
+  /* The stored history row must carry the pre-update blob */
+  int4store(key, 2);
+  ok(heap_rkey(info, read_rec, 0, key, 4, HA_READ_KEY_EXACT) == 0,
+     "%s: read the history row back", case_name);
+  ok(blob_matches(read_rec, payload, payload_len),
+     "%s: stored history row keeps the pre-update blob", case_name);
+
+  /*
+    The stored row's blob field holds the head of the chain it owns, so
+    comparing it with the chain the update parked says outright whether the
+    write adopted that chain or built its own.  A reassembled multi-run blob
+    lives in info->blob_buff and sources nothing from the chain, so there it
+    must allocate.
+  */
+  stored_chain= blob_data_ptr(info->current_ptr);
+  ok((stored_chain == parked_chain) == !want_reassembled,
+     "%s: history row %s the parked chain", case_name,
+     want_reassembled ? "allocated its own chain rather than adopting"
+                      : "adopted");
+
+  if (expect_alloc_growth >= 0)
+    ok(alloc_after - alloc_before == (ulong) expect_alloc_growth,
+       "%s: history write handed out %d record slot(s), no chain "
+       "(before=%lu, after=%lu)",
+       case_name, expect_alloc_growth, alloc_before, alloc_after);
+  else
+    skip(1, "%s: free list makes the slot count indeterminate here",
+         case_name);
+
+  ok(heap_check_heap(info, 0) == 0, "%s: heap consistent", case_name);
+
+  heap_drop_table(info);
+  heap_close(info);
+}
+
+
+/*
+  Test: the same hazard reached through heap_delete().
+
+  heap_delete() parks the chain for the same reason heap_update() does, so a
+  write issued while that parking is outstanding must preserve the deleted
+  row's buffer just as well.
+*/
+
+static void test_write_from_parked_chain_after_delete(void)
+{
+  HP_SHARE *share;
+  HP_INFO *info;
+  uchar rec[REC_LENGTH], old_rec[REC_LENGTH], hist_rec[REC_LENGTH];
+  uchar read_rec[REC_LENGTH];
+  uchar payload[MAX_PAYLOAD];
+  const uchar *parked_chain, *stored_chain;
+  ulong alloc_before, alloc_after;
+  uchar key[4];
+  const uint16 payload_len= 60;
+
+  fill_pattern(payload, payload_len);
+
+  if (create_and_open("test_blob_alias_del", &share, &info))
+  {
+    ok(0, "delete: setup failed: %d", my_errno);
+    skip(9, "setup failed");
+    return;
+  }
+
+  build_record(rec, 1, payload, payload_len);
+  ok(heap_write(info, rec) == 0, "delete: inserted base row");
+
+  int4store(key, 1);
+  ok(heap_rkey(info, old_rec, 0, key, 4, HA_READ_KEY_EXACT) == 0,
+     "delete: read base row into record[1]");
+  ok(heap_delete(info, old_rec) == 0, "delete: deleted base row");
+
+  parked_chain= info->pending_blob_chains[0];
+  ok(parked_chain != NULL, "delete: the delete parked the old chain");
+  alloc_before= (ulong) share->block.last_allocated;
+
+  memcpy(hist_rec, old_rec, REC_LENGTH);
+  int4store(hist_rec + INT_OFFSET, 2);
+  ok(heap_write(info, hist_rec) == 0, "delete: wrote history row");
+  alloc_after= (ulong) share->block.last_allocated;
+
+  ok(blob_matches(old_rec, payload, payload_len),
+     "delete: record[1] blob still reads correctly after the history write");
+
+  int4store(key, 2);
+  if (heap_rkey(info, read_rec, 0, key, 4, HA_READ_KEY_EXACT) == 0)
+    ok(blob_matches(read_rec, payload, payload_len),
+       "delete: stored history row keeps the pre-delete blob");
+  else
+    ok(0, "delete: could not read the history row back");
+
+  stored_chain= blob_data_ptr(info->current_ptr);
+  ok(stored_chain == parked_chain, "delete: history row adopted the "
+     "parked chain");
+
+  /*
+    The delete put its base record on the free list and parked its chain, so
+    the write reuses that slot and takes the chain over: nothing new at all.
+  */
+  ok(alloc_after == alloc_before,
+     "delete: history write handed out no new record slots "
+     "(before=%lu, after=%lu)", alloc_before, alloc_after);
+
+  ok(heap_check_heap(info, 0) == 0, "delete: heap consistent");
+
+  heap_drop_table(info);
+  heap_close(info);
+}
+
+
+int main(int argc __attribute__((unused)),
+         char **argv __attribute__((unused)))
+{
+  MY_INIT("hp_test_blob_alias");
+  plan(62);
+
+  /*
+    Nothing is on the free list in the first three cases, so the history write
+    can only take its one base record from the tail: a growth of exactly 1
+    says it adopted the chain, and anything more says it allocated one.
+  */
+  diag("Test 1: history write, 3-byte blob (data at offset 0 of the chain)");
+  test_write_from_parked_chain(3, FALSE, FALSE, 1, "3-byte");
+
+  diag("Test 2: history write, 60-byte blob (zero-copy single run)");
+  test_write_from_parked_chain(60, FALSE, FALSE, 1, "60-byte");
+
+  diag("Test 3: history write, 200-byte blob (zero-copy single run)");
+  test_write_from_parked_chain(200, FALSE, FALSE, 1, "200-byte");
+
+  diag("Test 4: history write, 200-byte blob reassembled from a multi-run "
+       "chain -- the layout that cannot alias");
+  test_write_from_parked_chain(200, TRUE, TRUE, -1, "multi-run");
+
+  diag("Test 5: history write while a delete's chain is parked");
+  test_write_from_parked_chain_after_delete();
+
+  my_end(0);
+  return exit_status();
+}
diff --git a/storage/heap/hp_update.c b/storage/heap/hp_update.c
index f822c8daaf803..ed5db1573162d 100644
--- a/storage/heap/hp_update.c
+++ b/storage/heap/hp_update.c
@@ -201,6 +201,12 @@ int heap_update(HP_INFO *info, const uchar *old, const uchar *heap_new)
           returns, reading old blob data from record[1] via zero-copy
           pointers into HP_BLOCK chain records.  Freeing chains here
           would overwrite those records, making the pointers dangle.
+
+          A system-versioned UPDATE then writes the history row from that
+          same record[1], and an AFTER UPDATE trigger reads OLD.<blob> from
+          it, so heap_write() leaves a chain the row it writes still sources
+          data from parked and adopts it instead of reallocating; see
+          hp_flush_unaliased_blob_free().
         */
         for (i= 0; i < share->blob_count; i++)
           info->pending_blob_chains[i]= (blob_changed[i] ?
diff --git a/storage/heap/hp_write.c b/storage/heap/hp_write.c
index 5c30caef4c633..7d93522956cf1 100644
--- a/storage/heap/hp_write.c
+++ b/storage/heap/hp_write.c
@@ -41,7 +41,15 @@ int heap_write(HP_INFO *info, const uchar *record)
     DBUG_RETURN(my_errno=EACCES);
   }
 #endif
-  hp_flush_pending_blob_free(info);
+  /*
+    Free old stored blob chains, parked by a previous heap_update() or
+    heap_delete(), that are not used by the current record.  A chain this
+    record still sources its blob data from stays parked, and
+    hp_write_blobs() adopts it below instead of allocating a copy of bytes
+    that are already in place.
+  */
+  if (info->has_pending_blob_free)
+    hp_flush_unaliased_blob_free(info, record);
   if (!(pos=next_free_record_pos(share)))
     DBUG_RETURN(my_errno);
   share->changed=1;