diff --git a/mysql-test/suite/rpl/r/rpl_table_map_metadata_underflow.result b/mysql-test/suite/rpl/r/rpl_table_map_metadata_underflow.result
new file mode 100644
index 0000000000000..0023b572b7eb1
--- /dev/null
+++ b/mysql-test/suite/rpl/r/rpl_table_map_metadata_underflow.result
@@ -0,0 +1,62 @@
+include/master-slave.inc
+[connection master]
+connection slave;
+call mtr.add_suppression("Found invalid event in binary log");
+call mtr.add_suppression("Relay log read failure: Could not parse relay log event entry");
+#
+# Initialize test data
+#
+connection slave;
+include/stop_slave.inc
+CHANGE MASTER TO MASTER_USE_GTID=slave_pos;
+include/start_slave.inc
+connection master;
+create table t1 (c1 int,c2 int,c3 int,c4 int,c5 int,c6 int,c7 int,c8 int,c9 int,c10 int,c11 int,c12 int,c13 int,c14 int,c15 int,c16 int,c17 int,c18 int,c19 int,c20 int,c21 int,c22 int,c23 int,c24 int);
+insert into t1 (c1) values (0);
+include/save_master_gtid.inc
+connection slave;
+include/sync_with_master_gtid.inc
+connection master;
+set @saved_dbug= @@global.debug_dbug;
+#
+# Test Case: a Table_map declaring more metadata than it carries
+#
+connection master;
+set @@global.debug_dbug= "+d,binlog_sender_undersized_table_map_metadata";
+connection slave;
+include/stop_slave.inc
+include/start_slave.inc
+connection master;
+insert into t1 (c1) values (1);
+connection slave;
+# Waiting for the SQL thread to reject the malformed Table_map
+include/wait_for_slave_sql_error.inc [errno=1594]
+# Ensure the event was rejected before its metadata was decoded
+include/assert_grep.inc [The SQL thread reported an invalid event]
+#
+# Recover: discard the poisoned relay log and re-fetch a clean copy
+#
+connection master;
+set @@global.debug_dbug= @saved_dbug;
+connection slave;
+include/stop_slave_io.inc
+include/start_slave.inc
+#
+# Ensure replication works after the invalid event
+#
+connection master;
+insert into t1 (c1) values (2);
+include/save_master_gtid.inc
+connection slave;
+include/sync_with_master_gtid.inc
+include/diff_tables.inc [master:t1,slave:t1]
+#
+# Cleanup
+#
+connection master;
+drop table t1;
+include/save_master_gtid.inc
+connection slave;
+include/sync_with_master_gtid.inc
+include/rpl_end.inc
+# End of rpl_table_map_metadata_underflow.test
diff --git a/mysql-test/suite/rpl/t/rpl_table_map_metadata_underflow.test b/mysql-test/suite/rpl/t/rpl_table_map_metadata_underflow.test
new file mode 100644
index 0000000000000..2f081f062d2cb
--- /dev/null
+++ b/mysql-test/suite/rpl/t/rpl_table_map_metadata_underflow.test
@@ -0,0 +1,114 @@
+#
+#   This test verifies that a replica rejects a Table_map event whose field
+# metadata is shorter than the metadata the column types it declares require.
+# The replica must stop the SQL thread with an error rather than decode the
+# metadata past the end of what the event carries, and replication must resume
+# once a well formed copy of the event arrives.
+#
+#   MDEV-40646 reported that such an event read past the metadata buffer. A
+# build with AddressSanitizer is the reliable witness of that read, aborting
+# at the first byte past the allocation. A regular debug build does not fault
+# on the read: it decodes adjacent heap and stops the SQL thread with a
+# conversion error whose reported type is rendered from those bytes. This test
+# asserts the rejection the fix adds ahead of the read, which holds on every
+# build.
+#
+# References:
+#   * MDEV-40646: Slave Crashes in table_def::table_def() on Malformed
+#     Table_map
+#
+
+--source include/have_debug.inc
+# A Table_map event is only written under row based binary logging.
+--source include/have_binlog_format_row.inc
+--source include/master-slave.inc
+
+--connection slave
+call mtr.add_suppression("Found invalid event in binary log");
+call mtr.add_suppression("Relay log read failure: Could not parse relay log event entry");
+
+--echo #
+--echo # Initialize test data
+--echo #
+
+--connection slave
+# Use slave_pos so we can easily reset the relay log after it receives a
+# corrupted event
+--source include/stop_slave.inc
+CHANGE MASTER TO MASTER_USE_GTID=slave_pos;
+--source include/start_slave.inc
+
+--connection master
+# With 24 columns, the metadata reads run well past the end of the replica's
+# allocation once the event declares a single metadata byte.
+create table t1 (c1 int,c2 int,c3 int,c4 int,c5 int,c6 int,c7 int,c8 int,c9 int,c10 int,c11 int,c12 int,c13 int,c14 int,c15 int,c16 int,c17 int,c18 int,c19 int,c20 int,c21 int,c22 int,c23 int,c24 int);
+insert into t1 (c1) values (0);
+--source include/save_master_gtid.inc
+
+--connection slave
+--source include/sync_with_master_gtid.inc
+
+--connection master
+set @saved_dbug= @@global.debug_dbug;
+
+--echo #
+--echo # Test Case: a Table_map declaring more metadata than it carries
+--echo #
+
+# The reconnect below starts a dump thread that inherits the injection.
+--connection master
+set @@global.debug_dbug= "+d,binlog_sender_undersized_table_map_metadata";
+
+--connection slave
+--source include/stop_slave.inc
+--source include/start_slave.inc
+
+--connection master
+insert into t1 (c1) values (1);
+
+--connection slave
+--echo # Waiting for the SQL thread to reject the malformed Table_map
+--let $slave_sql_errno= 1594
+--source include/wait_for_slave_sql_error.inc
+
+--echo # Ensure the event was rejected before its metadata was decoded
+--let $assert_text= The SQL thread reported an invalid event
+--let $assert_file= $MYSQLTEST_VARDIR/log/mysqld.2.err
+--let $assert_select= Found invalid event in binary log
+--let $assert_count= 1
+--source include/assert_grep.inc
+
+--echo #
+--echo # Recover: discard the poisoned relay log and re-fetch a clean copy
+--echo #
+--connection master
+set @@global.debug_dbug= @saved_dbug;
+
+--connection slave
+--source include/stop_slave_io.inc
+--source include/start_slave.inc
+
+--echo #
+--echo # Ensure replication works after the invalid event
+--echo #
+--connection master
+insert into t1 (c1) values (2);
+--source include/save_master_gtid.inc
+
+--connection slave
+--source include/sync_with_master_gtid.inc
+--let $diff_tables=master:t1,slave:t1
+--source include/diff_tables.inc
+
+--echo #
+--echo # Cleanup
+--echo #
+--connection master
+drop table t1;
+--source include/save_master_gtid.inc
+
+--connection slave
+--source include/sync_with_master_gtid.inc
+
+--source include/rpl_end.inc
+--echo # End of rpl_table_map_metadata_underflow.test
diff --git a/sql/log_event.cc b/sql/log_event.cc
index f72423718db59..d77dff632e713 100644
--- a/sql/log_event.cc
+++ b/sql/log_event.cc
@@ -3766,6 +3766,22 @@ Table_map_log_event::Table_map_log_event(const uchar *buf, uint event_len,
     m_field_metadata_size= net_field_length(&ptr_after_colcnt);
     DBUG_EXECUTE_IF("corrupt_table_map_field_metadata_size_read",
                     m_field_metadata_size= (1 << 20););
+
+    /*
+      Ensure the field metadata block covers the metadata the declared
+      column types consume.
+    */
+    size_t metadata_needed= 0;
+    for (ulong i= 0; i < m_colcnt; i++)
+      metadata_needed+= table_def::field_metadata_length(m_coltype[i]);
+    if (unlikely(m_field_metadata_size < metadata_needed))
+    {
+      m_coltype= NULL;
+      my_free(m_memory);
+      m_memory= NULL;
+      DBUG_VOID_RETURN;
+    }
+
     if (m_field_metadata_size <= (m_colcnt * 2))
     {
       uint num_null_bytes= (m_colcnt + 7) / 8;
diff --git a/sql/rpl_utility.cc b/sql/rpl_utility.cc
index 9ea8bb3b82256..3f2d314a9fcb6 100644
--- a/sql/rpl_utility.cc
+++ b/sql/rpl_utility.cc
@@ -173,6 +173,34 @@ uint32 table_def::calc_field_size(uint col, uchar *master_data) const
 
 PSI_memory_key key_memory_table_def_memory;
 
+uint table_def::field_metadata_length(uint binlog_type)
+{
+  switch (static_cast<enum_field_types>(binlog_type)) {
+  case MYSQL_TYPE_TINY_BLOB:
+  case MYSQL_TYPE_BLOB:
+  case MYSQL_TYPE_BLOB_COMPRESSED:
+  case MYSQL_TYPE_MEDIUM_BLOB:
+  case MYSQL_TYPE_LONG_BLOB:
+  case MYSQL_TYPE_DOUBLE:
+  case MYSQL_TYPE_FLOAT:
+  case MYSQL_TYPE_GEOMETRY:
+  case MYSQL_TYPE_TIME2:
+  case MYSQL_TYPE_DATETIME2:
+  case MYSQL_TYPE_TIMESTAMP2:
+    return 1;
+  case MYSQL_TYPE_SET:
+  case MYSQL_TYPE_ENUM:
+  case MYSQL_TYPE_STRING:
+  case MYSQL_TYPE_BIT:
+  case MYSQL_TYPE_VARCHAR:
+  case MYSQL_TYPE_VARCHAR_COMPRESSED:
+  case MYSQL_TYPE_NEWDECIMAL:
+    return 2;
+  default:
+    return 0;
+  }
+}
+
 table_def::table_def(unsigned char *types, ulong size,
                      uchar *field_metadata, int metadata_size,
                      uchar *null_bitmap, uint16 flags)
diff --git a/sql/rpl_utility.h b/sql/rpl_utility.h
index c28e8aa10ebef..712855cb5c30b 100644
--- a/sql/rpl_utility.h
+++ b/sql/rpl_utility.h
@@ -74,6 +74,12 @@ class table_def
   {
     return static_cast<enum_field_types>(m_type[index]);
   }
+
+  /*
+    Number of field-metadata bytes a column of the given binlog type occupies
+    in a Table_map event.
+  */
+  static uint field_metadata_length(uint binlog_type);
   /*
     Return a representation of the type data for one field.
 
diff --git a/sql/sql_repl.cc b/sql/sql_repl.cc
index 25d5c9e0a585f..c77e7018ecb78 100644
--- a/sql/sql_repl.cc
+++ b/sql/sql_repl.cc
@@ -2035,6 +2035,57 @@ send_event_to_slave(binlog_send_info *info, Log_event_type event_type,
     return "run 'before_send_event' hook failed";
   }
 
+#ifndef DBUG_OFF
+  /*
+    Rewrite the body of a Table_map event so it declares columns whose types
+    need more field metadata than the event carries, to test that a replica
+    bounds the metadata before decoding it. Every column becomes
+    MYSQL_TYPE_STRING, which reads two metadata bytes, and the event is left
+    with a single metadata byte, so a replica missing the bound reads past the
+    metadata while decoding the columns. The rewrite keeps the column count the
+    master already wrote, which the test fixes at 24 columns, enough to carry
+    the metadata reads well past the end of the replica's allocation.
+
+    The event is rewritten here, on the wire, rather than in the master's own
+    binary log, so the injection lands on the copy the dump thread sends and
+    the master's binary log stays well formed.
+  */
+  DBUG_EXECUTE_IF("binlog_sender_undersized_table_map_metadata",
+  {
+    if (event_type == TABLE_MAP_EVENT)
+    {
+      /* The rewritten event can be one byte longer than the original.  */
+      packet->realloc(len + 1);
+      uchar *ev= (uchar*) packet->ptr() + ev_offset;
+      uchar *p= ev + LOG_EVENT_HEADER_LEN + TABLE_MAP_HEADER_LEN;
+      uint db_len= *p;
+      p+= 1 + db_len + 1;                     // db name length byte, name, null
+      uint tbl_len= *p;
+      p+= 1 + tbl_len + 1;                    // table name length byte, name, null
+      uint colcnt= *p++;                      // column count (one packed byte)
+      uchar *coltype= p;
+
+      memset(coltype, MYSQL_TYPE_STRING, colcnt);
+      uchar *w= coltype + colcnt;
+      *w++= 1;                                // field metadata size (one byte)
+      *w++= 0;                                // the single metadata byte
+      uint null_bytes= (colcnt + 7) / 8;
+      memset(w, 0, null_bytes);
+      w+= null_bytes;
+
+      ulong new_ev_len= (ulong) (w - ev);
+      if (current_checksum_alg == BINLOG_CHECKSUM_ALG_CRC32)
+        new_ev_len+= BINLOG_CHECKSUM_LEN;
+      int4store(ev + EVENT_LEN_OFFSET, new_ev_len);
+      if (current_checksum_alg == BINLOG_CHECKSUM_ALG_CRC32)
+        int4store(ev + new_ev_len - BINLOG_CHECKSUM_LEN,
+                  my_checksum(0, ev, new_ev_len - BINLOG_CHECKSUM_LEN));
+      len= ev_offset + new_ev_len;
+      packet->length(len);
+    }
+  });
+#endif
+
   if (my_net_write(info->net, (uchar*) packet->ptr(), len))
   {
     info->error= ER_UNKNOWN_ERROR;