diff --git a/include/heap.h b/include/heap.h
index 7d326c4988762..b0a2500d344bd 100644
--- a/include/heap.h
+++ b/include/heap.h
@@ -202,6 +202,7 @@ typedef struct st_heap_info
   ulong last_hash_of_key;
   int lastinx,errkey;
   int  mode;				/* Mode of file (READONLY..) */
+  int lock_type;
   uint opt_flag,update;
   enum ha_rkey_function last_find_flag;
   uchar *lastkey;			/* Last used key with rkey */
@@ -219,6 +220,7 @@ typedef struct st_heap_info
   my_bool implicit_emptied;
   my_bool has_zerocopy_blobs;  /* Last hp_read_blobs produced zero-copy ptrs */
   my_bool has_pending_blob_free; /* pending_blob_chains awaits freeing */
+  my_bool changed;             /* Changes done to table since last lock */
   THR_LOCK_DATA lock;
   LIST open_list;
 } HP_INFO;
diff --git a/mysql-test/suite/heap/blob_delayed_insert.opt b/mysql-test/suite/heap/blob_delayed_insert.opt
new file mode 100644
index 0000000000000..789275fa25e27
--- /dev/null
+++ b/mysql-test/suite/heap/blob_delayed_insert.opt
@@ -0,0 +1 @@
+--skip-log-bin
diff --git a/mysql-test/suite/heap/blob_delayed_insert.result b/mysql-test/suite/heap/blob_delayed_insert.result
new file mode 100644
index 0000000000000..b869012ce137d
--- /dev/null
+++ b/mysql-test/suite/heap/blob_delayed_insert.result
@@ -0,0 +1,51 @@
+CREATE TABLE t1 (a INT, b BLOB) ENGINE=MEMORY;
+INSERT DELAYED INTO t1 VALUES (1, REPEAT('x', 300)), (2, REPEAT('y', 300));
+SELECT VARIABLE_VALUE > 0 AS delayed_thread_ran
+FROM information_schema.global_status
+WHERE VARIABLE_NAME = 'DELAYED_WRITES';
+delayed_thread_ran
+1
+INSERT DELAYED INTO t1 VALUES (3, REPEAT('z', 300)), (4, REPEAT('w', 300));
+connect  reader,localhost,root,,test;
+connection default;
+INSERT DELAYED INTO t1 VALUES (5, REPEAT('v', 300));
+connection reader;
+SELECT COUNT(*) >= 0 AS reader_ran FROM t1;
+reader_ran
+1
+connection default;
+INSERT DELAYED INTO t1 VALUES (6, REPEAT('u', 300));
+disconnect reader;
+SELECT a, LENGTH(b) FROM t1;
+a	LENGTH(b)
+1	300
+2	300
+3	300
+4	300
+5	300
+6	300
+CHECK TABLE t1;
+Table	Op	Msg_type	Msg_text
+test.t1	check	status	OK
+INSERT INTO t1 VALUES (7, REPEAT('t', 300));
+UPDATE t1 SET b = REPEAT('s', 400) WHERE a = 7;
+DELETE FROM t1 WHERE a = 7;
+CHECK TABLE t1;
+Table	Op	Msg_type	Msg_text
+test.t1	check	status	OK
+FLUSH TABLES;
+INSERT INTO t1 VALUES (8, REPEAT('r', 300));
+UPDATE t1 SET b = REPEAT('q', 400) WHERE a = 8;
+DELETE FROM t1 WHERE a = 8;
+CHECK TABLE t1;
+Table	Op	Msg_type	Msg_text
+test.t1	check	status	OK
+SELECT a, LENGTH(b) FROM t1;
+a	LENGTH(b)
+1	300
+2	300
+3	300
+4	300
+5	300
+6	300
+DROP TABLE t1;
diff --git a/mysql-test/suite/heap/blob_delayed_insert.test b/mysql-test/suite/heap/blob_delayed_insert.test
new file mode 100644
index 0000000000000..eeb29b77c27e6
--- /dev/null
+++ b/mysql-test/suite/heap/blob_delayed_insert.test
@@ -0,0 +1,76 @@
+#
+# The record of a THR_LOCK grant survives an INSERT DELAYED lock cycle.
+#
+# thr_lock() does not call the grant callback once per external_lock(): a
+# delayed insert is granted TL_WRITE_DELAYED and then calls it again on the
+# same request when thr_upgrade_write_delay_lock() turns that into a real
+# write lock, with no external_lock() in between.  A handle that counted
+# grants would keep the surplus for its whole life and then report a lock it
+# does not hold on every later statement.
+#
+# The binary log must be off: with statement binlogging INSERT DELAYED is
+# downgraded to an ordinary write lock and the delayed thread never runs.
+#
+
+CREATE TABLE t1 (a INT, b BLOB) ENGINE=MEMORY;
+
+INSERT DELAYED INTO t1 VALUES (1, REPEAT('x', 300)), (2, REPEAT('y', 300));
+
+--let $wait_condition= SELECT COUNT(*) = 2 FROM t1
+--source include/wait_condition.inc
+
+# The delayed thread really did the writing, rather than the statement being
+# silently downgraded to a plain INSERT.
+SELECT VARIABLE_VALUE > 0 AS delayed_thread_ran
+  FROM information_schema.global_status
+  WHERE VARIABLE_NAME = 'DELAYED_WRITES';
+
+# Second cycle on the still-live delayed thread: it locks the same handle
+# again, which is where a leaked grant is noticed.
+INSERT DELAYED INTO t1 VALUES (3, REPEAT('z', 300)), (4, REPEAT('w', 300));
+
+--let $wait_condition= SELECT COUNT(*) = 4 FROM t1
+--source include/wait_condition.inc
+
+# A third cycle with a concurrent reader, so the delayed thread has to give
+# the lock up and take it again mid-batch.
+connect (reader,localhost,root,,test);
+
+connection default;
+INSERT DELAYED INTO t1 VALUES (5, REPEAT('v', 300));
+
+connection reader;
+SELECT COUNT(*) >= 0 AS reader_ran FROM t1;
+
+connection default;
+INSERT DELAYED INTO t1 VALUES (6, REPEAT('u', 300));
+
+--let $wait_condition= SELECT COUNT(*) = 6 FROM t1
+--source include/wait_condition.inc
+
+disconnect reader;
+
+--sorted_result
+SELECT a, LENGTH(b) FROM t1;
+CHECK TABLE t1;
+
+# Ordinary lock cycles on the same table afterwards.  These write blobs, so
+# they park and redeem chains, which a handle wrongly believing it holds the
+# lock would do outside it.
+INSERT INTO t1 VALUES (7, REPEAT('t', 300));
+UPDATE t1 SET b = REPEAT('s', 400) WHERE a = 7;
+DELETE FROM t1 WHERE a = 7;
+CHECK TABLE t1;
+
+# And once more after the table is reopened, so the same share is reached
+# through a fresh handle.
+FLUSH TABLES;
+INSERT INTO t1 VALUES (8, REPEAT('r', 300));
+UPDATE t1 SET b = REPEAT('q', 400) WHERE a = 8;
+DELETE FROM t1 WHERE a = 8;
+CHECK TABLE t1;
+
+--sorted_result
+SELECT a, LENGTH(b) FROM t1;
+
+DROP TABLE t1;
diff --git a/mysql-test/suite/heap/blob_lock_twice.result b/mysql-test/suite/heap/blob_lock_twice.result
new file mode 100644
index 0000000000000..a51a55bbfaf28
--- /dev/null
+++ b/mysql-test/suite/heap/blob_lock_twice.result
@@ -0,0 +1,59 @@
+CREATE TABLE ht (a INT, b BLOB) ENGINE=MEMORY;
+CREATE TABLE mi (a INT, b BLOB) ENGINE=MyISAM;
+INSERT INTO ht VALUES (1,'foo'),(2,'bar');
+INSERT INTO mi VALUES (1,'foo'),(2,'bar');
+# ===== single aliased entry, referenced as "table AS alias" =====
+# -- MyISAM
+LOCK TABLE mi AS m1 WRITE;
+SELECT COUNT(*) FROM mi AS m1;
+COUNT(*)
+2
+UPDATE mi AS m1 SET b='x' WHERE a=1;
+# the unaliased name is not locked
+SELECT COUNT(*) FROM mi;
+ERROR HY000: Table 'mi' was not locked with LOCK TABLES
+UNLOCK TABLES;
+# -- MEMORY
+LOCK TABLE ht AS h1 WRITE;
+SELECT COUNT(*) FROM ht AS h1;
+COUNT(*)
+2
+UPDATE ht AS h1 SET b='x' WHERE a=1;
+# the unaliased name is not locked
+SELECT COUNT(*) FROM ht;
+ERROR HY000: Table 'ht' was not locked with LOCK TABLES
+UNLOCK TABLES;
+# ===== two aliased entries, one WRITE and one READ =====
+# -- MyISAM
+LOCK TABLE mi AS m1 WRITE, mi AS m2 READ;
+SELECT COUNT(*) FROM mi AS m2;
+COUNT(*)
+2
+UPDATE mi AS m1 SET b='y' WHERE a=1;
+# INSERT takes no alias, so it cannot reach an aliased lock at all
+INSERT INTO mi SELECT a+10, b FROM mi AS m2;
+ERROR HY000: Table 'mi' was not locked with LOCK TABLES
+UNLOCK TABLES;
+# -- MEMORY
+LOCK TABLE ht AS h1 WRITE, ht AS h2 READ;
+SELECT COUNT(*) FROM ht AS h2;
+COUNT(*)
+2
+UPDATE ht AS h1 SET b='y' WHERE a=1;
+# INSERT takes no alias, so it cannot reach an aliased lock at all
+INSERT INTO ht SELECT a+10, b FROM ht AS h2;
+ERROR HY000: Table 'ht' was not locked with LOCK TABLES
+UNLOCK TABLES;
+# ===== blob update and delete under the double lock (MEMORY) =====
+LOCK TABLE ht AS h1 WRITE, ht AS h2 READ;
+UPDATE ht AS h1, ht AS h2 SET h1.b=REPEAT('z', 900)
+WHERE h1.a=h2.a AND h1.a=1;
+DELETE FROM ht AS h1 WHERE a=2;
+UNLOCK TABLES;
+CHECK TABLE ht;
+Table	Op	Msg_type	Msg_text
+test.ht	check	status	OK
+SELECT a, LENGTH(b) FROM ht;
+a	LENGTH(b)
+1	900
+DROP TABLE ht, mi;
diff --git a/mysql-test/suite/heap/blob_lock_twice.test b/mysql-test/suite/heap/blob_lock_twice.test
new file mode 100644
index 0000000000000..0dd62abe47419
--- /dev/null
+++ b/mysql-test/suite/heap/blob_lock_twice.test
@@ -0,0 +1,69 @@
+#
+# A MEMORY table with a blob locked twice in the same lock set.
+#
+# Two aliased entries give one share two handles, and thr_lock() grants each
+# of them separately, so the grant a handle records is its own.  Every row
+# operation below runs with both grants live, and the blob update parks a
+# chain that has to be redeemed when the handles are unlocked.  MyISAM runs
+# the same shapes as a control.
+#
+# An aliased entry can only be referenced by its alias while LOCK TABLES is
+# in effect; the unaliased name is not in the lock set at all.  That is
+# asserted rather than avoided, because it is what keeps the two handles
+# distinct.
+#
+CREATE TABLE ht (a INT, b BLOB) ENGINE=MEMORY;
+CREATE TABLE mi (a INT, b BLOB) ENGINE=MyISAM;
+INSERT INTO ht VALUES (1,'foo'),(2,'bar');
+INSERT INTO mi VALUES (1,'foo'),(2,'bar');
+
+--echo # ===== single aliased entry, referenced as "table AS alias" =====
+--echo # -- MyISAM
+LOCK TABLE mi AS m1 WRITE;
+SELECT COUNT(*) FROM mi AS m1;
+UPDATE mi AS m1 SET b='x' WHERE a=1;
+--echo # the unaliased name is not locked
+--error ER_TABLE_NOT_LOCKED
+SELECT COUNT(*) FROM mi;
+UNLOCK TABLES;
+
+--echo # -- MEMORY
+LOCK TABLE ht AS h1 WRITE;
+SELECT COUNT(*) FROM ht AS h1;
+UPDATE ht AS h1 SET b='x' WHERE a=1;
+--echo # the unaliased name is not locked
+--error ER_TABLE_NOT_LOCKED
+SELECT COUNT(*) FROM ht;
+UNLOCK TABLES;
+
+--echo # ===== two aliased entries, one WRITE and one READ =====
+--echo # -- MyISAM
+LOCK TABLE mi AS m1 WRITE, mi AS m2 READ;
+SELECT COUNT(*) FROM mi AS m2;
+UPDATE mi AS m1 SET b='y' WHERE a=1;
+--echo # INSERT takes no alias, so it cannot reach an aliased lock at all
+--error ER_TABLE_NOT_LOCKED
+INSERT INTO mi SELECT a+10, b FROM mi AS m2;
+UNLOCK TABLES;
+
+--echo # -- MEMORY
+LOCK TABLE ht AS h1 WRITE, ht AS h2 READ;
+SELECT COUNT(*) FROM ht AS h2;
+UPDATE ht AS h1 SET b='y' WHERE a=1;
+--echo # INSERT takes no alias, so it cannot reach an aliased lock at all
+--error ER_TABLE_NOT_LOCKED
+INSERT INTO ht SELECT a+10, b FROM ht AS h2;
+UNLOCK TABLES;
+
+--echo # ===== blob update and delete under the double lock (MEMORY) =====
+LOCK TABLE ht AS h1 WRITE, ht AS h2 READ;
+UPDATE ht AS h1, ht AS h2 SET h1.b=REPEAT('z', 900)
+  WHERE h1.a=h2.a AND h1.a=1;
+DELETE FROM ht AS h1 WHERE a=2;
+UNLOCK TABLES;
+
+CHECK TABLE ht;
+--sorted_result
+SELECT a, LENGTH(b) FROM ht;
+
+DROP TABLE ht, mi;
diff --git a/mysql-test/suite/heap/blob_online_alter.result b/mysql-test/suite/heap/blob_online_alter.result
new file mode 100644
index 0000000000000..1583e3608b121
--- /dev/null
+++ b/mysql-test/suite/heap/blob_online_alter.result
@@ -0,0 +1,26 @@
+CREATE TABLE t1 (a INT PRIMARY KEY, b BLOB) ENGINE=MEMORY;
+INSERT INTO t1 VALUES (1, REPEAT('x', 300)), (2, REPEAT('y', 300)),
+(3, REPEAT('w', 300)), (4, REPEAT('v', 300));
+connect  alterer,localhost,root,,test;
+SET DEBUG_SYNC= 'alter_table_online_downgraded SIGNAL downgraded EXECUTE 1';
+SET DEBUG_SYNC= 'alter_table_online_progress WAIT_FOR dml_done EXECUTE 1';
+ALTER TABLE t1 ADD COLUMN c INT, ALGORITHM=COPY, LOCK=NONE;
+connection default;
+SET SESSION lock_wait_timeout= 20;
+SET DEBUG_SYNC= 'now WAIT_FOR downgraded';
+DELETE FROM t1 WHERE a = 1;
+UPDATE t1 SET b = REPEAT('z', 900) WHERE a = 2;
+DELETE FROM t1 WHERE a = 3;
+SET DEBUG_SYNC= 'now SIGNAL dml_done';
+connection alterer;
+connection default;
+disconnect alterer;
+SET DEBUG_SYNC= 'RESET';
+SELECT a, LENGTH(b), c FROM t1;
+a	LENGTH(b)	c
+2	900	NULL
+4	300	NULL
+CHECK TABLE t1;
+Table	Op	Msg_type	Msg_text
+test.t1	check	status	OK
+DROP TABLE t1;
diff --git a/mysql-test/suite/heap/blob_online_alter.test b/mysql-test/suite/heap/blob_online_alter.test
new file mode 100644
index 0000000000000..c6891493212c4
--- /dev/null
+++ b/mysql-test/suite/heap/blob_online_alter.test
@@ -0,0 +1,47 @@
+#
+# Blob chains parked on an online-ALTER copy target.
+#
+# The copy target is locked with a direct handler::ha_external_lock() rather
+# than through the SQL layer's lock set, so thr_lock() never grants it
+# anything and the handle holds a write lock that no grant records.  Rows
+# deleted or updated on the source while the copy runs are replayed onto that
+# target, and a blob delete/update there parks a chain that has to be redeemed
+# when the copy is unlocked.
+#
+# The ALTER is not paused while it holds the source lock -- doing that blocks
+# the very statements this test needs to run.  It is released at the
+# post-downgrade point and only made to wait once it is replaying.
+#
+--source include/have_debug_sync.inc
+
+CREATE TABLE t1 (a INT PRIMARY KEY, b BLOB) ENGINE=MEMORY;
+INSERT INTO t1 VALUES (1, REPEAT('x', 300)), (2, REPEAT('y', 300)),
+                      (3, REPEAT('w', 300)), (4, REPEAT('v', 300));
+
+connect (alterer,localhost,root,,test);
+SET DEBUG_SYNC= 'alter_table_online_downgraded SIGNAL downgraded EXECUTE 1';
+SET DEBUG_SYNC= 'alter_table_online_progress WAIT_FOR dml_done EXECUTE 1';
+--send ALTER TABLE t1 ADD COLUMN c INT, ALGORITHM=COPY, LOCK=NONE
+
+connection default;
+SET SESSION lock_wait_timeout= 20;
+SET DEBUG_SYNC= 'now WAIT_FOR downgraded';
+
+# Both shapes that park a chain: a delete, and an update that grows the blob
+DELETE FROM t1 WHERE a = 1;
+UPDATE t1 SET b = REPEAT('z', 900) WHERE a = 2;
+DELETE FROM t1 WHERE a = 3;
+
+SET DEBUG_SYNC= 'now SIGNAL dml_done';
+
+connection alterer;
+--reap
+
+connection default;
+disconnect alterer;
+SET DEBUG_SYNC= 'RESET';
+
+--sorted_result
+SELECT a, LENGTH(b), c FROM t1;
+CHECK TABLE t1;
+DROP TABLE t1;
diff --git a/mysql-test/suite/heap/blob_tmp_table.result b/mysql-test/suite/heap/blob_tmp_table.result
new file mode 100644
index 0000000000000..77058a9ddeeea
--- /dev/null
+++ b/mysql-test/suite/heap/blob_tmp_table.result
@@ -0,0 +1,95 @@
+#
+# Blob updates and deletes on a user TEMPORARY MEMORY table.
+#
+# A non-transactional TEMPORARY table is left out of the lock set
+# altogether, so its handle never holds a THR_LOCK.  It is not an
+# internal table, though, so it still defers its blob chain frees --
+# a combination no other kind of table has, and one a debug build used
+# to assert on at the end of every such statement.
+#
+CREATE TEMPORARY TABLE t1 (a INT, b BLOB) ENGINE=MEMORY;
+INSERT INTO t1 VALUES (1,'one'), (2,REPEAT('two',500)), (3,REPEAT('three',500));
+# UPDATE parks one chain per changed blob
+UPDATE t1 SET b=REPEAT('x',4000) WHERE a=2;
+SELECT a, LENGTH(b), LEFT(b,5) FROM t1 ORDER BY a;
+a	LENGTH(b)	LEFT(b,5)
+1	3	one
+2	4000	xxxxx
+3	2500	three
+CHECK TABLE t1;
+Table	Op	Msg_type	Msg_text
+test.t1	check	status	OK
+# DELETE parks the whole row's chain
+DELETE FROM t1 WHERE a=3;
+SELECT a, LENGTH(b), LEFT(b,5) FROM t1 ORDER BY a;
+a	LENGTH(b)	LEFT(b,5)
+1	3	one
+2	4000	xxxxx
+CHECK TABLE t1;
+Table	Op	Msg_type	Msg_text
+test.t1	check	status	OK
+# Repeated cycles: every parked chain has to come back to the free list
+SELECT a, LENGTH(b), LEFT(b,5) FROM t1 ORDER BY a;
+a	LENGTH(b)	LEFT(b,5)
+1	1500	oneon
+2	8000	zzzzz
+CHECK TABLE t1;
+Table	Op	Msg_type	Msg_text
+test.t1	check	status	OK
+# Multi-table UPDATE and DELETE across two TEMPORARY tables
+CREATE TEMPORARY TABLE t2 (a INT, b BLOB) ENGINE=MEMORY;
+INSERT INTO t2 VALUES (1,REPEAT('u',4000)), (2,REPEAT('v',4000));
+UPDATE t1, t2 SET t1.b=REPEAT('p',4000), t2.b=REPEAT('q',4000)
+WHERE t1.a=t2.a;
+SELECT a, LENGTH(b), LEFT(b,5) FROM t1 ORDER BY a;
+a	LENGTH(b)	LEFT(b,5)
+1	4000	ppppp
+2	4000	ppppp
+SELECT a, LENGTH(b), LEFT(b,5) FROM t2 ORDER BY a;
+a	LENGTH(b)	LEFT(b,5)
+1	4000	qqqqq
+2	4000	qqqqq
+CHECK TABLE t1, t2;
+Table	Op	Msg_type	Msg_text
+test.t1	check	status	OK
+test.t2	check	status	OK
+DELETE t1, t2 FROM t1, t2 WHERE t1.a=t2.a AND t1.a=1;
+SELECT a, LENGTH(b), LEFT(b,5) FROM t1 ORDER BY a;
+a	LENGTH(b)	LEFT(b,5)
+2	4000	ppppp
+SELECT a, LENGTH(b), LEFT(b,5) FROM t2 ORDER BY a;
+a	LENGTH(b)	LEFT(b,5)
+2	4000	qqqqq
+CHECK TABLE t1, t2;
+Table	Op	Msg_type	Msg_text
+test.t1	check	status	OK
+test.t2	check	status	OK
+DROP TEMPORARY TABLE t1, t2;
+#
+# The same statements on a non-temporary MEMORY table, which does hold
+# the lock while it parks and redeems at unlock time.
+#
+CREATE TABLE t1 (a INT, b BLOB) ENGINE=MEMORY;
+INSERT INTO t1 VALUES (1,'one'), (2,REPEAT('two',500)), (3,REPEAT('three',500));
+UPDATE t1 SET b=REPEAT('x',4000) WHERE a=2;
+DELETE FROM t1 WHERE a=3;
+SELECT a, LENGTH(b), LEFT(b,5) FROM t1 ORDER BY a;
+a	LENGTH(b)	LEFT(b,5)
+1	3	one
+2	4000	xxxxx
+CHECK TABLE t1;
+Table	Op	Msg_type	Msg_text
+test.t1	check	status	OK
+# Under LOCK TABLES the redemption happens in ha_heap::reset() instead,
+# with the lock still held
+LOCK TABLES t1 WRITE;
+UPDATE t1 SET b=REPEAT('w',4000) WHERE a=2;
+DELETE FROM t1 WHERE a=1;
+UNLOCK TABLES;
+SELECT a, LENGTH(b), LEFT(b,5) FROM t1 ORDER BY a;
+a	LENGTH(b)	LEFT(b,5)
+2	4000	wwwww
+CHECK TABLE t1;
+Table	Op	Msg_type	Msg_text
+test.t1	check	status	OK
+DROP TABLE t1;
diff --git a/mysql-test/suite/heap/blob_tmp_table.test b/mysql-test/suite/heap/blob_tmp_table.test
new file mode 100644
index 0000000000000..009b02e45183e
--- /dev/null
+++ b/mysql-test/suite/heap/blob_tmp_table.test
@@ -0,0 +1,78 @@
+--source include/not_embedded.inc
+
+--echo #
+--echo # Blob updates and deletes on a user TEMPORARY MEMORY table.
+--echo #
+--echo # A non-transactional TEMPORARY table is left out of the lock set
+--echo # altogether, so its handle never holds a THR_LOCK.  It is not an
+--echo # internal table, though, so it still defers its blob chain frees --
+--echo # a combination no other kind of table has, and one a debug build used
+--echo # to assert on at the end of every such statement.
+--echo #
+
+CREATE TEMPORARY TABLE t1 (a INT, b BLOB) ENGINE=MEMORY;
+INSERT INTO t1 VALUES (1,'one'), (2,REPEAT('two',500)), (3,REPEAT('three',500));
+
+--echo # UPDATE parks one chain per changed blob
+UPDATE t1 SET b=REPEAT('x',4000) WHERE a=2;
+SELECT a, LENGTH(b), LEFT(b,5) FROM t1 ORDER BY a;
+CHECK TABLE t1;
+
+--echo # DELETE parks the whole row's chain
+DELETE FROM t1 WHERE a=3;
+SELECT a, LENGTH(b), LEFT(b,5) FROM t1 ORDER BY a;
+CHECK TABLE t1;
+
+--echo # Repeated cycles: every parked chain has to come back to the free list
+--disable_query_log
+let $i= 20;
+while ($i)
+{
+  UPDATE t1 SET b=REPEAT('y',4000) WHERE a=2;
+  UPDATE t1 SET b=REPEAT('z',8000) WHERE a=2;
+  DELETE FROM t1 WHERE a=1;
+  INSERT INTO t1 VALUES (1,REPEAT('one',500));
+  dec $i;
+}
+--enable_query_log
+SELECT a, LENGTH(b), LEFT(b,5) FROM t1 ORDER BY a;
+CHECK TABLE t1;
+
+--echo # Multi-table UPDATE and DELETE across two TEMPORARY tables
+CREATE TEMPORARY TABLE t2 (a INT, b BLOB) ENGINE=MEMORY;
+INSERT INTO t2 VALUES (1,REPEAT('u',4000)), (2,REPEAT('v',4000));
+UPDATE t1, t2 SET t1.b=REPEAT('p',4000), t2.b=REPEAT('q',4000)
+ WHERE t1.a=t2.a;
+SELECT a, LENGTH(b), LEFT(b,5) FROM t1 ORDER BY a;
+SELECT a, LENGTH(b), LEFT(b,5) FROM t2 ORDER BY a;
+CHECK TABLE t1, t2;
+
+DELETE t1, t2 FROM t1, t2 WHERE t1.a=t2.a AND t1.a=1;
+SELECT a, LENGTH(b), LEFT(b,5) FROM t1 ORDER BY a;
+SELECT a, LENGTH(b), LEFT(b,5) FROM t2 ORDER BY a;
+CHECK TABLE t1, t2;
+
+DROP TEMPORARY TABLE t1, t2;
+
+--echo #
+--echo # The same statements on a non-temporary MEMORY table, which does hold
+--echo # the lock while it parks and redeems at unlock time.
+--echo #
+
+CREATE TABLE t1 (a INT, b BLOB) ENGINE=MEMORY;
+INSERT INTO t1 VALUES (1,'one'), (2,REPEAT('two',500)), (3,REPEAT('three',500));
+UPDATE t1 SET b=REPEAT('x',4000) WHERE a=2;
+DELETE FROM t1 WHERE a=3;
+SELECT a, LENGTH(b), LEFT(b,5) FROM t1 ORDER BY a;
+CHECK TABLE t1;
+
+--echo # Under LOCK TABLES the redemption happens in ha_heap::reset() instead,
+--echo # with the lock still held
+LOCK TABLES t1 WRITE;
+UPDATE t1 SET b=REPEAT('w',4000) WHERE a=2;
+DELETE FROM t1 WHERE a=1;
+UNLOCK TABLES;
+SELECT a, LENGTH(b), LEFT(b,5) FROM t1 ORDER BY a;
+CHECK TABLE t1;
+
+DROP TABLE t1;
diff --git a/storage/heap/CMakeLists.txt b/storage/heap/CMakeLists.txt
index d0dc06a6c5823..4e75796778d8e 100644
--- a/storage/heap/CMakeLists.txt
+++ b/storage/heap/CMakeLists.txt
@@ -34,7 +34,7 @@ IF(WITH_UNIT_TESTS)
   TARGET_LINK_LIBRARIES(hp_test2 heap mysys dbug strings)
   MY_ADD_TESTS(hp_test_hash hp_test_scan hp_test_freelist hp_test_concurrent
                hp_test_block_size hp_test_blob_alias hp_test_update
-               hp_test_write_dup
+               hp_test_write_dup hp_test_unlock_check
                LINK_LIBRARIES heap mysys dbug strings)
 
   INCLUDE_DIRECTORIES(${CMAKE_SOURCE_DIR}/sql
diff --git a/storage/heap/ha_heap.cc b/storage/heap/ha_heap.cc
index f0e82c506df6b..32d0be46250be 100644
--- a/storage/heap/ha_heap.cc
+++ b/storage/heap/ha_heap.cc
@@ -151,6 +151,18 @@ int ha_heap::open(const char *name, int mode, uint test_if_locked)
     used.
     */
   key_stat_version= file->s->key_stat_version-1;
+
+  /*
+    A temporary table is private to its session, so no other connection can
+    reach the share and it can be regarded as always locked.  This has to
+    cover the user's CREATE TEMPORARY TABLE too, not just the optimizer's
+    internal one: `internal' is HA_OPEN_INTERNAL_TABLE, while it is the user
+    temporary table that parks blob chains (heap_delete()/heap_update() free
+    an internal table's chains outright) and that get_lock_data() drops from
+    the lock set, so it never reaches external_lock() at all.
+  */
+  if (internal_table || table->s->tmp_table != NO_TMP_TABLE)
+    file->lock_type= F_EXTRA_LCK;
   if (file->s->blob_count)
   {
     /* Mark that table may have zerocopy blobs and unlock is not safe */
@@ -505,6 +517,21 @@ int ha_heap::extra(enum ha_extra_function operation)
 
 int ha_heap::reset()
 {
+  /*
+    Prepares the handle for the next statement; see heap_reset().
+
+    heap_reset() redeems the blob chains parked by a deferred free, which puts
+    records back on the shared free list.
+
+    We can only have pending blobs if we have a write lock on the table or if
+    the table is temporary, in which case lock_type == F_EXTRA_LCK.  In both
+    cases it is ok to free the blobs.  Note that this has to cover the user's
+    CREATE TEMPORARY TABLE and not just the optimizer's internal one: an
+    internal table frees its chains outright and never parks any, while a user
+    temporary table parks them and is never passed to thr_multi_lock() at all.
+  */
+  DBUG_ASSERT(!file->has_pending_blob_free ||
+              table_is_locked_and_changed(file));
   return heap_reset(file);
 }
 
@@ -531,26 +558,46 @@ int ha_heap::reset_auto_increment(ulonglong value)
 }
 
 
+/*
+  Check table consistenty on unlock under EXTRA_HEAP_DEBUG
+
+  Note that heap tables does not have external locking.
+  Locking against other handlers to the same table handled through
+  thr_multi_lock()
+  This code is never called for internal MariaDB temporary tables
+*/
+
 int ha_heap::external_lock(THD *thd, int lock_type)
 {
 #if !defined(DBUG_OFF) && defined(EXTRA_HEAP_DEBUG)
-  /*
-    A table already marked crashed is knowingly inconsistent; every data
-    access on it fails with HA_ERR_CRASHED, so re-detecting the damage
-    here would only raise a second error into a diagnostics area that
-    can already be OK (e.g. after UNLOCK TABLES) and fire the
-    Diagnostics_area assertion.
-  */
-  if (lock_type == F_UNLCK && file->s->changed &&
-      !heap_is_crashed(file->s) && heap_check_heap(file, 0))
+  /* See hp_may_check_heap_on_unlock() for when this is safe to run at all */
+  if (lock_type == F_UNLCK && hp_may_check_heap_on_unlock(file) &&
+      heap_check_heap(file, 0))
+  {
+    file->lock_type= lock_type;
     return HA_ERR_CRASHED;
+  }
 #endif
-  if (lock_type != F_UNLCK && heap_is_crashed(file->s))
-    return HA_ERR_CRASHED;
-
   if (lock_type == F_UNLCK)
+  {
+    DBUG_ASSERT(!file->has_pending_blob_free ||
+                table_is_locked_and_changed(file));
     hp_flush_pending_blob_free(file);
-  return 0;					// No external locking
+  }
+  else
+  {
+    if (heap_is_crashed(file->s))
+      return HA_ERR_CRASHED;             // We are not granting lock if crashed
+
+    /* Assure we are not locked from before */
+    DBUG_ASSERT(file->lock_type == F_UNLCK ||
+                file->lock_type == F_EXTRA_LCK);
+  }
+  /* Remember last lock; a temporary table stays "always locked" */
+  file->lock_type= (table->s->tmp_table != NO_TMP_TABLE ? F_EXTRA_LCK :
+                    lock_type);
+  file->changed= 0;                     // Marker if table changes under lock
+  return 0;
 }
 
 
diff --git a/storage/heap/heapdef.h b/storage/heap/heapdef.h
index f157bcc77e9b9..f0dafac04d4ce 100644
--- a/storage/heap/heapdef.h
+++ b/storage/heap/heapdef.h
@@ -49,6 +49,12 @@ C_MODE_START
 #define HP_CONT_REC_COUNT_SIZE 2
 #define HP_CONT_HEADER_SIZE    (sizeof(uchar*) + HP_CONT_REC_COUNT_SIZE)
 
+/*
+  Use this lock type for temporary tables; it indicates that only this
+  session can reach the table, so it can be regarded as always locked
+*/
+#define F_EXTRA_LCK -1
+
 /*
   Row flags byte predicates.
   The flags byte is at offset 'visible' in each primary or run-header record.
@@ -391,6 +397,45 @@ static inline void hp_flush_pending_blob_free(HP_INFO *info)
     hp_flush_pending_blob_free_impl(info);
 }
 
+
+/*
+  Check if table is locked and has changed
+  This is mainly used to see if we can do check of the table in
+  debug binaries.
+
+  Both terms are needed.  ha_heap::external_lock() records the lock type
+  before thr_multi_lock() runs, so it is set on both paths that unlock after
+  a failed lock attempt -- mysql_lock_tables() balancing the external locks
+  it took because thr_multi_lock() timed out, and lock_external() unwinding
+  the tables it had already locked because a later one refused.  Neither ever
+  ran a row operation, so it is `changed' that tells them from a real unlock.
+  It has to be per handle for the same reason: HP_SHARE::changed is true on
+  exactly those paths, since another connection is the one writing.
+*/
+
+static inline my_bool table_is_locked_and_changed(const HP_INFO *info)
+{
+  return info->lock_type != F_UNLCK && info->changed;
+}
+
+/*
+  May ha_heap::external_lock(F_UNLCK) verify the table with heap_check_heap()?
+
+  It is only safe to do a scan if the table is write locked against other
+  connections. We also only need to do check if we have done changes to the
+  table.
+
+  A table already marked crashed is knowingly inconsistent; every data access
+  on it fails with HA_ERR_CRASHED, so re-detecting the damage here would only
+  raise a second error into a diagnostics area that can already be OK (e.g.
+  after UNLOCK TABLES) and fire the Diagnostics_area assertion.
+*/
+
+static inline my_bool hp_may_check_heap_on_unlock(const HP_INFO *info)
+{
+  return (table_is_locked_and_changed(info) && !heap_is_crashed(info->s));
+}
+
 /*
   Does a record's blob data live in `chain`?
 
diff --git a/storage/heap/hp_delete.c b/storage/heap/hp_delete.c
index 61db3ed730380..e5a33ed54228c 100644
--- a/storage/heap/hp_delete.c
+++ b/storage/heap/hp_delete.c
@@ -153,7 +153,7 @@ int heap_delete(HP_INFO *info, const uchar *record)
 
   if (info->opt_flag & READ_CHECK_USED && hp_rectest(info,record))
     DBUG_RETURN(my_errno);			/* Record changed */
-  share->changed=1;
+  info->changed= share->changed= 1;
 
   if ( --(share->records) < share->blength >> 1) share->blength>>=1;
   pos=info->current_ptr;
diff --git a/storage/heap/hp_extra.c b/storage/heap/hp_extra.c
index 901bace2fbef3..2cb675a34268b 100644
--- a/storage/heap/hp_extra.c
+++ b/storage/heap/hp_extra.c
@@ -58,6 +58,20 @@ int heap_extra(register HP_INFO *info, enum ha_extra_function function)
 } /* heap_extra */
 
 
+/*
+  Prepare the handle for the next statement
+
+  Called at the end of a statement, once nothing reads the last row any
+  more: from ha_heap::reset(), and from heap_extra(HA_EXTRA_RESET_STATE).
+
+  Resets the row position, so that the next statement starts a fresh scan
+  instead of resuming the previous one, and clears up what the previous
+  statement left behind -- the blob chains that a deferred free parked, and
+  the buffers that blob values were reassembled into.
+
+  See ha_heap::reset() for why redeeming the parked chains is safe here.
+*/
+
 int heap_reset(HP_INFO *info)
 {
   hp_flush_pending_blob_free(info);
diff --git a/storage/heap/hp_open.c b/storage/heap/hp_open.c
index c512b88efd369..e2dba4e88ad3d 100644
--- a/storage/heap/hp_open.c
+++ b/storage/heap/hp_open.c
@@ -44,6 +44,7 @@ HP_INFO *heap_open_from_share(HP_SHARE *share, int mode)
   info->lastkey= (uchar*) (info + 1);
   info->recbuf= (uchar*) (info->lastkey + share->max_key_length);
   info->mode= mode;
+  info->lock_type= F_UNLCK;
   info->current_record= (ulong) ~0L;		/* No current record */
   info->lastinx= info->errkey= -1;
   if (share->blob_count && !share->internal)
diff --git a/storage/heap/hp_test_unlock_check-t.c b/storage/heap/hp_test_unlock_check-t.c
new file mode 100644
index 0000000000000..5ed44471de4cc
--- /dev/null
+++ b/storage/heap/hp_test_unlock_check-t.c
@@ -0,0 +1,212 @@
+/*
+  Unit test: the unlock-time table verification must not run unless this
+  handle both holds a lock and has changed the table under it.
+
+  ha_heap::external_lock(F_UNLCK) verifies the table with heap_check_heap().
+  That is safe on the ordinary unlock path, which sql/lock.cc describes as
+  external_lock(F_UNLCK) followed by thr_multi_unlock(), so the lock is still
+  held.  It is not safe on either path that unlocks after a failed lock
+  attempt: when mysql_lock_tables() balances the external locks it took
+  because thr_multi_lock() timed out, and when lock_external() itself unwinds
+  the tables it already locked because a later table refused -- sql/lock.cc
+  spells out the second of those, the first is only in its code.  In both the
+  caller holds nothing, while another connection is writing.
+
+  The requested lock type cannot tell those apart on its own: external_lock()
+  has already recorded it on both, before thr_multi_lock() runs.  What tells
+  them apart is that neither ever ran a row operation, so this handle has
+  changed nothing since the lock was recorded.  HP_INFO::changed carries that,
+  and it is per handle -- HP_SHARE::changed answers a different question,
+  "did anyone change this table", which is true on exactly the paths that
+  must be suppressed.
+
+  Every outcome below is forced rather than raced: the states are built
+  directly, in the order ha_heap::external_lock() builds them.
+*/
+
+#include "hp_test_helpers.h"
+
+/*
+  Reproduce what ha_heap::external_lock() does when a lock is granted: record
+  the type and start a fresh change epoch.  lock_external() reaches this for
+  every table it locks, including the ones it is about to unwind.
+*/
+
+static void grant_lock(HP_INFO *info, int lock_type)
+{
+  info->lock_type= lock_type;
+  info->changed= 0;
+}
+
+
+/*
+  Reproduce the state heap_write() is in between allocating a slot and marking
+  it visible: next_free_record_pos() has already published the slot in
+  total_records, but the record has not been stored yet.
+
+  The slot is zeroed rather than left as it comes from my_malloc() so that the
+  test asserts on a defined outcome.  A zero flags byte is what a scan of a
+  half-written row legitimately sees; leaving the malloc garbage in place is
+  what makes the same access an uninitialised read under MSAN.
+*/
+
+static uchar *park_mid_write(HP_SHARE *share)
+{
+  uchar *pos= next_free_record_pos(share);
+  if (pos)
+    memset(pos, 0, share->block.recbuffer);
+  return pos;
+}
+
+
+static void unpark_mid_write(HP_SHARE *share, uchar *pos)
+{
+  hp_push_free_record(share, pos);
+  hp_shrink_tail(share);
+}
+
+
+int main(int argc __attribute__((unused)),
+         char **argv __attribute__((unused)))
+{
+  HP_SHARE *share;
+  HP_INFO *info1, *info2;
+  uchar rec[REC_LENGTH], rec2[REC_LENGTH];
+  uchar blob_data[200];
+  uchar *parked_slot;
+  int i;
+
+  MY_INIT("hp_test_unlock_check-t");
+  plan(15);
+
+  if (create_and_open("test_unlock_check", &share, &info1))
+  {
+    ok(0, "setup failed");
+    my_end(0);
+    return exit_status();
+  }
+
+  /*
+    A second handle on the same share, which is what a second connection
+    holds.  It never runs a row operation in this test.
+  */
+  if (!(info2= heap_open("test_unlock_check", 2)))
+  {
+    ok(0, "second open failed");
+    heap_drop_table(info1);
+    my_end(0);
+    return exit_status();
+  }
+  heap_extra(info2, HA_EXTRA_NO_READCHECK);
+
+  for (i= 0; i < (int) sizeof(blob_data); i++)
+    blob_data[i]= (uchar) ('a' + (i % 26));
+
+  ok(heap_check_heap(info1, 0) == 0, "table is consistent before the test");
+
+  /* --- the three row operations each open a change epoch --- */
+
+  grant_lock(info1, F_WRLCK);
+  build_record(rec, 1, blob_data, (uint16) sizeof(blob_data));
+  if (heap_write(info1, rec))
+  {
+    ok(0, "heap_write failed");
+    goto cleanup;
+  }
+  ok(info1->changed, "heap_write marks the writing handle as having changed");
+  ok(share->changed, "heap_write marks the share as changed too");
+
+  grant_lock(info1, F_WRLCK);
+  build_record(rec2, 1, blob_data, (uint16) (sizeof(blob_data) / 2));
+  if (heap_scan_init(info1) || heap_scan(info1, rec) ||
+      heap_update(info1, rec, rec2))
+  {
+    ok(0, "heap_update failed");
+    goto cleanup;
+  }
+  ok(info1->changed, "heap_update marks the updating handle as having changed");
+
+  grant_lock(info1, F_WRLCK);
+  if (heap_scan_init(info1) || heap_scan(info1, rec) ||
+      heap_delete(info1, rec))
+  {
+    ok(0, "heap_delete failed");
+    goto cleanup;
+  }
+  ok(info1->changed, "heap_delete marks the deleting handle as having changed");
+
+  /*
+    The record that separates this from HP_SHARE::changed.  info2 has run
+    nothing, so it must report no change of its own even though the share it
+    shares with info1 is changed -- that combination is precisely the failed
+    lock attempt, where another connection is writing and this one is not.
+  */
+  grant_lock(info2, F_WRLCK);
+  ok(share->changed && !info2->changed,
+     "a handle that ran nothing is unchanged while the share is changed");
+
+  /* --- the lock type half --- */
+
+  info1->changed= 1;
+  grant_lock(info1, F_UNLCK);
+  info1->changed= 1;
+  ok(!table_is_locked_and_changed(info1),
+     "an unlocked handle claims nothing, whatever it changed before");
+
+  grant_lock(info1, F_RDLCK);
+  info1->changed= 1;
+  ok(table_is_locked_and_changed(info1), "a read-locked handle claims a lock");
+
+  grant_lock(info1, F_WRLCK);
+  info1->changed= 1;
+  ok(table_is_locked_and_changed(info1), "a write-locked handle claims a lock");
+
+  grant_lock(info1, F_EXTRA_LCK);
+  info1->changed= 1;
+  ok(table_is_locked_and_changed(info1),
+     "a table private to its session counts as always locked");
+
+  /* --- the change half, on a handle whose lock type is armed --- */
+
+  grant_lock(info1, F_WRLCK);
+  ok(!table_is_locked_and_changed(info1),
+     "an armed but unused handle claims nothing: the failed lock attempt");
+
+  /* --- the gate as a whole --- */
+
+  info1->changed= 1;
+  ok(hp_may_check_heap_on_unlock(info1),
+     "locked, changed and healthy: the table may be verified");
+
+  heap_mark_crashed(share);
+  ok(!hp_may_check_heap_on_unlock(info1),
+     "a table already marked crashed is not verified again");
+  heap_clear_state(share);
+
+  /*
+    The payload.  Park the share in the state a writer passes through mid-row,
+    so the verification has something to wrongly find, and confirm that the
+    handle which did not write is the one being kept away from it.
+  */
+  parked_slot= park_mid_write(share);
+  if (!parked_slot)
+  {
+    ok(0, "could not park the share mid-write");
+    goto cleanup;
+  }
+
+  grant_lock(info2, F_WRLCK);
+  ok(!hp_may_check_heap_on_unlock(info2) && heap_check_heap(info2, 0) != 0,
+     "the check is suppressed on a handle whose scan would report damage");
+  heap_clear_state(share);
+
+  unpark_mid_write(share, parked_slot);
+  ok(heap_check_heap(info1, 0) == 0,
+     "the table was consistent all along: the report was a false positive");
+
+cleanup:
+  heap_close(info2);
+  heap_drop_table(info1);
+  my_end(0);
+  return exit_status();
+}
diff --git a/storage/heap/hp_update.c b/storage/heap/hp_update.c
index 551eb557564dd..66e450fa4e944 100644
--- a/storage/heap/hp_update.c
+++ b/storage/heap/hp_update.c
@@ -36,7 +36,7 @@ int heap_update(HP_INFO *info, const uchar *old, const uchar *heap_new)
   if (info->opt_flag & READ_CHECK_USED && hp_rectest(info,old))
     DBUG_RETURN(my_errno);				/* Record changed */
   if (--(share->records) < share->blength >> 1) share->blength>>= 1;
-  share->changed=1;
+  info->changed= share->changed= 1;
 
   // Save the cursor position to recover if insert fails.
   recovery_ptr= info->current_ptr;
diff --git a/storage/heap/hp_write.c b/storage/heap/hp_write.c
index c043562adeb57..96fba9b83132a 100644
--- a/storage/heap/hp_write.c
+++ b/storage/heap/hp_write.c
@@ -54,7 +54,7 @@ int heap_write(HP_INFO *info, const uchar *record)
     hp_flush_unaliased_blob_free(info, record);
   if (!(pos=next_free_record_pos(share)))
     DBUG_RETURN(my_errno);
-  share->changed=1;
+  info->changed= share->changed= 1;
 
   for (keydef = share->keydef, end = keydef + share->keys; keydef < end;
        keydef++)