diff --git a/mysql-test/main/error_simulation.result b/mysql-test/main/error_simulation.result
index 009f5323b37d8..5bda368d3d4e7 100644
--- a/mysql-test/main/error_simulation.result
+++ b/mysql-test/main/error_simulation.result
@@ -126,3 +126,26 @@ Got one of the listed errors
 DROP FUNCTION f1;
 SET debug_dbug= @saved_dbug;
 # End of 10.2 tests
+#
+# Conversion of an internal temporary table to the on-disk engine when
+# the scan of the in-memory table cannot be started
+#
+CREATE TABLE t1 (v VARCHAR(1024));
+INSERT INTO t1 SELECT CONCAT('v', LPAD(seq, 6, '0')) FROM seq_1_to_100;
+SET @save_tmp_table_size= @@tmp_table_size;
+SET @save_max_heap_table_size= @@max_heap_table_size;
+SET SESSION tmp_table_size= 262144;
+SET SESSION max_heap_table_size= 262144;
+SET SESSION debug_dbug='+d,heap_conversion_rnd_init_error';
+SELECT COUNT(*) FROM t1 a JOIN (SELECT DISTINCT v FROM t1) d ON a.v = d.v;
+ERROR HY000: Table definition has changed, please retry transaction
+SET debug_dbug= @saved_dbug;
+#The on-disk table created for the failed conversion is gone.
+#Without the injected failure the query is answered as usual.
+SELECT COUNT(*) FROM t1 a JOIN (SELECT DISTINCT v FROM t1) d ON a.v = d.v;
+COUNT(*)
+100
+SET SESSION tmp_table_size= @save_tmp_table_size;
+SET SESSION max_heap_table_size= @save_max_heap_table_size;
+DROP TABLE t1;
+# End of 13.1 tests
diff --git a/mysql-test/main/error_simulation.test b/mysql-test/main/error_simulation.test
index 3beeefce39fde..c2624c35dec5c 100644
--- a/mysql-test/main/error_simulation.test
+++ b/mysql-test/main/error_simulation.test
@@ -157,3 +157,38 @@ DROP FUNCTION f1;
 SET debug_dbug= @saved_dbug;
 
 --echo # End of 10.2 tests
+
+--echo #
+--echo # Conversion of an internal temporary table to the on-disk engine when
+--echo # the scan of the in-memory table cannot be started
+--echo #
+
+CREATE TABLE t1 (v VARCHAR(1024));
+INSERT INTO t1 SELECT CONCAT('v', LPAD(seq, 6, '0')) FROM seq_1_to_100;
+
+SET @save_tmp_table_size= @@tmp_table_size;
+SET @save_max_heap_table_size= @@max_heap_table_size;
+
+# Small enough that the temporary table of the query below does not fit in
+# memory and is converted to the on-disk engine.
+SET SESSION tmp_table_size= 262144;
+SET SESSION max_heap_table_size= 262144;
+
+--let $tmpdir= `SELECT @@tmpdir`
+
+SET SESSION debug_dbug='+d,heap_conversion_rnd_init_error';
+--error ER_TABLE_DEF_CHANGED
+SELECT COUNT(*) FROM t1 a JOIN (SELECT DISTINCT v FROM t1) d ON a.v = d.v;
+SET debug_dbug= @saved_dbug;
+
+--echo #The on-disk table created for the failed conversion is gone.
+--list_files $tmpdir #sql-temptable*
+
+--echo #Without the injected failure the query is answered as usual.
+SELECT COUNT(*) FROM t1 a JOIN (SELECT DISTINCT v FROM t1) d ON a.v = d.v;
+
+SET SESSION tmp_table_size= @save_tmp_table_size;
+SET SESSION max_heap_table_size= @save_max_heap_table_size;
+DROP TABLE t1;
+
+--echo # End of 13.1 tests
diff --git a/mysql-test/main/tmp_table_convert_while_read.result b/mysql-test/main/tmp_table_convert_while_read.result
new file mode 100644
index 0000000000000..65c7c89147b61
--- /dev/null
+++ b/mysql-test/main/tmp_table_convert_while_read.result
@@ -0,0 +1,55 @@
+CREATE TABLE t1 (v VARCHAR(1024));
+INSERT INTO t1 SELECT CONCAT('v', LPAD(seq, 6, '0')) FROM seq_1_to_100;
+CREATE TABLE t2 (a INT);
+INSERT INTO t2 VALUES (1),(2);
+SET @save_tmp_table_size= @@tmp_table_size;
+SET @save_max_heap_table_size= @@max_heap_table_size;
+SET SESSION tmp_table_size= 262144;
+SET SESSION max_heap_table_size= 262144;
+# Derived table with DISTINCT
+FLUSH STATUS;
+SELECT COUNT(*) FROM t1 a JOIN (SELECT DISTINCT v FROM t1) d ON a.v = d.v;
+COUNT(*)
+100
+SHOW STATUS LIKE 'Created_tmp_disk_tables';
+Variable_name	Value
+Created_tmp_disk_tables	2
+# Derived table with GROUP BY
+FLUSH STATUS;
+SELECT COUNT(*) FROM t1 a JOIN (SELECT v FROM t1 GROUP BY v) d ON a.v = d.v;
+COUNT(*)
+100
+SHOW STATUS LIKE 'Created_tmp_disk_tables';
+Variable_name	Value
+Created_tmp_disk_tables	2
+SET SESSION optimizer_switch='derived_merge=off';
+# Derived table that is not merged into the outer query
+FLUSH STATUS;
+SELECT COUNT(*) FROM t2, (SELECT v FROM t1) AS sq;
+COUNT(*)
+200
+SHOW STATUS LIKE 'Created_tmp_disk_tables';
+Variable_name	Value
+Created_tmp_disk_tables	1
+# Derived table filled by a UNION
+FLUSH STATUS;
+SELECT COUNT(*) FROM t2, (SELECT v FROM t1 UNION SELECT v FROM t1) AS sq;
+COUNT(*)
+200
+SHOW STATUS LIKE 'Created_tmp_disk_tables';
+Variable_name	Value
+Created_tmp_disk_tables	2
+SET SESSION optimizer_switch=DEFAULT;
+# View materialized into a temporary table
+CREATE ALGORITHM=TEMPTABLE VIEW v1 AS SELECT * FROM t1;
+FLUSH STATUS;
+SELECT COUNT(*) FROM v1, t2 WHERE a = 1;
+COUNT(*)
+100
+SHOW STATUS LIKE 'Created_tmp_disk_tables';
+Variable_name	Value
+Created_tmp_disk_tables	1
+DROP VIEW v1;
+SET SESSION tmp_table_size= @save_tmp_table_size;
+SET SESSION max_heap_table_size= @save_max_heap_table_size;
+DROP TABLE t1, t2;
diff --git a/mysql-test/main/tmp_table_convert_while_read.test b/mysql-test/main/tmp_table_convert_while_read.test
new file mode 100644
index 0000000000000..2a06b4d618da3
--- /dev/null
+++ b/mysql-test/main/tmp_table_convert_while_read.test
@@ -0,0 +1,65 @@
+#
+# An internal temporary table can be converted from the in-memory engine to
+# the on-disk engine while a reader of that same table is already in progress.
+# The reader keeps state on the handler of the table, so the conversion, which
+# replaces that handler, has to carry the state over to the new one.
+#
+--source include/have_sequence.inc
+
+CREATE TABLE t1 (v VARCHAR(1024));
+INSERT INTO t1 SELECT CONCAT('v', LPAD(seq, 6, '0')) FROM seq_1_to_100;
+
+CREATE TABLE t2 (a INT);
+INSERT INTO t2 VALUES (1),(2);
+
+SET @save_tmp_table_size= @@tmp_table_size;
+SET @save_max_heap_table_size= @@max_heap_table_size;
+
+# Small enough that the temporary tables below do not fit in memory. Set
+# explicitly so that the test does not depend on the server defaults.
+SET SESSION tmp_table_size= 262144;
+SET SESSION max_heap_table_size= 262144;
+
+# Created_tmp_disk_tables is checked after every query so that a future change
+# of the sizes above cannot silently leave these queries without a conversion.
+--disable_ps2_protocol
+--disable_cursor_protocol
+
+--echo # Derived table with DISTINCT
+FLUSH STATUS;
+SELECT COUNT(*) FROM t1 a JOIN (SELECT DISTINCT v FROM t1) d ON a.v = d.v;
+SHOW STATUS LIKE 'Created_tmp_disk_tables';
+
+--echo # Derived table with GROUP BY
+FLUSH STATUS;
+SELECT COUNT(*) FROM t1 a JOIN (SELECT v FROM t1 GROUP BY v) d ON a.v = d.v;
+SHOW STATUS LIKE 'Created_tmp_disk_tables';
+
+SET SESSION optimizer_switch='derived_merge=off';
+
+--echo # Derived table that is not merged into the outer query
+FLUSH STATUS;
+SELECT COUNT(*) FROM t2, (SELECT v FROM t1) AS sq;
+SHOW STATUS LIKE 'Created_tmp_disk_tables';
+
+--echo # Derived table filled by a UNION
+FLUSH STATUS;
+SELECT COUNT(*) FROM t2, (SELECT v FROM t1 UNION SELECT v FROM t1) AS sq;
+SHOW STATUS LIKE 'Created_tmp_disk_tables';
+
+SET SESSION optimizer_switch=DEFAULT;
+
+--echo # View materialized into a temporary table
+CREATE ALGORITHM=TEMPTABLE VIEW v1 AS SELECT * FROM t1;
+FLUSH STATUS;
+SELECT COUNT(*) FROM v1, t2 WHERE a = 1;
+SHOW STATUS LIKE 'Created_tmp_disk_tables';
+DROP VIEW v1;
+
+--enable_cursor_protocol
+--enable_ps2_protocol
+
+SET SESSION tmp_table_size= @save_tmp_table_size;
+SET SESSION max_heap_table_size= @save_max_heap_table_size;
+
+DROP TABLE t1, t2;
diff --git a/sql/handler.h b/sql/handler.h
index fa9196f189a8f..1a57d3a9dc88e 100644
--- a/sql/handler.h
+++ b/sql/handler.h
@@ -3621,6 +3621,9 @@ class handler :public Sql_alloc
   void start_psi_batch_mode();
   /** End a batch started with @c start_psi_batch_mode. */
   void end_psi_batch_mode();
+  /** Check if a batch started with @c start_psi_batch_mode is in progress. */
+  bool is_in_psi_batch_mode() const
+  { return m_psi_batch_mode != PSI_BATCH_MODE_NONE; }
 
   /* If we have row logging enabled for this table */
   bool row_logging, row_logging_init;
diff --git a/sql/sql_select.cc b/sql/sql_select.cc
index dc9e7b6113333..0329fa1a54f4a 100644
--- a/sql/sql_select.cc
+++ b/sql/sql_select.cc
@@ -23829,6 +23829,7 @@ create_internal_tmp_table_from_heap(THD *thd, TABLE *table,
   TABLE_SHARE share;
   const char *save_proc_info;
   int write_err= 0;
+  bool psi_batch_mode= false;
   String tmp_alias;
   DBUG_ENTER("create_internal_tmp_table_from_heap");
   if (is_duplicate)
@@ -23874,8 +23875,13 @@ create_internal_tmp_table_from_heap(THD *thd, TABLE *table,
   if (table->file->indexes_are_disabled())
     new_table.file->ha_disable_indexes(key_map(0), false);
   table->file->ha_index_or_rnd_end();
+  DBUG_EXECUTE_IF("heap_conversion_rnd_init_error",
+                  {
+                    table->file->print_error(HA_ERR_TABLE_DEF_CHANGED, MYF(0));
+                    goto err_drop;
+                  });
   if (table->file->ha_rnd_init_with_error(1))
-    DBUG_RETURN(1);
+    goto err_drop;
   if (new_table.no_rows)
     new_table.file->extra(HA_EXTRA_NO_ROWS);
   else
@@ -23919,6 +23925,16 @@ create_internal_tmp_table_from_heap(THD *thd, TABLE *table,
 
   /* remove heap table and change to use myisam table */
   (void) table->file->ha_rnd_end();
+  /*
+    A reader of this table may have put its handler into PFS batch mode, see
+    JOIN_TAB::pfs_batch_update(). The reader ends the batch on the handler
+    that table->file points to at that time, which is the new handler, so the
+    batch has to be moved over to it. Ending it here without starting it again
+    would leave that call without a matching start_psi_batch_mode().
+  */
+  psi_batch_mode= table->file->is_in_psi_batch_mode();
+  if (psi_batch_mode)
+    table->file->end_psi_batch_mode();
   (void) table->file->ha_close();          // This deletes the table !
   delete table->file;
   table->file=0;
@@ -23939,6 +23955,8 @@ create_internal_tmp_table_from_heap(THD *thd, TABLE *table,
 
   table->file->change_table_ptr(table, table->s);
   table->use_all_columns();
+  if (psi_batch_mode)
+    table->file->start_psi_batch_mode();
   if (save_proc_info)
     thd_proc_info(thd, (!strcmp(save_proc_info,"Copying to tmp table") ?
                   "Copying to tmp table on disk" : save_proc_info));
@@ -23949,6 +23967,7 @@ create_internal_tmp_table_from_heap(THD *thd, TABLE *table,
   table->file->print_error(write_err, MYF(0));
 err_killed:
   (void) table->file->ha_rnd_end();
+err_drop:
   (void) new_table.file->drop_table(new_table.s->path.str);
  err2:
   delete new_table.file;