diff --git a/sql/mdl.cc b/sql/mdl.cc
index 944d4cd4d4300..0ebbb16825409 100644
--- a/sql/mdl.cc
+++ b/sql/mdl.cc
@@ -1830,7 +1830,7 @@ void MDL_wait::reset_status()
 /**
   Wait for the status to be assigned to this wait slot.
 
-  @param owner           MDL context owner.
+  @param owner           THD of the owning thread.
   @param abs_timeout     Absolute time after which waiting should stop.
   @param set_status_on_timeout TRUE  - If in case of timeout waiting
                                        context should close the wait slot by
@@ -1842,7 +1842,7 @@ void MDL_wait::reset_status()
 */
 
 MDL_wait::enum_wait_status
-MDL_wait::timed_wait(MDL_context_owner *owner, struct timespec *abs_timeout,
+MDL_wait::timed_wait(THD *owner, struct timespec *abs_timeout,
                      bool set_status_on_timeout,
                      const PSI_stage_info *wait_state_name)
 {
@@ -1863,9 +1863,9 @@ MDL_wait::timed_wait(MDL_context_owner *owner, struct timespec *abs_timeout,
 #ifdef WITH_WSREP
 # ifdef ENABLED_DEBUG_SYNC
     // Allow tests to block thread before MDL-wait
-    DEBUG_SYNC(owner->get_thd(), "wsrep_before_mdl_wait");
+    DEBUG_SYNC(owner, "wsrep_before_mdl_wait");
 # endif
-    if (WSREP_ON && wsrep_thd_is_BF(owner->get_thd(), false))
+    if (WSREP_ON && wsrep_thd_is_BF(owner, false))
     {
       wait_result= mysql_cond_wait(&m_COND_wait_status, &m_LOCK_wait_status);
     }
@@ -2962,7 +2962,7 @@ MDL_context::acquire_lock(MDL_request *mdl_request, double lock_wait_timeout)
   bool abort_blocking_enabled= false;
   double abort_blocking_timeout= slave_abort_blocking_timeout;
   if (abort_blocking_timeout < lock_wait_timeout &&
-      m_owner->get_thd()->rgi_slave)
+      m_owner->rgi_slave)
   {
     /*
       After @@slave_abort_blocking_timeout seconds, kill non-replication
@@ -3005,7 +3005,7 @@ MDL_context::acquire_lock(MDL_request *mdl_request, double lock_wait_timeout)
     if (wait_status != MDL_wait::EMPTY)
       break;
     /* Check if the client is gone while we were waiting. */
-    if (! thd_is_connected(m_owner->get_thd()))
+    if (! thd_is_connected(m_owner))
     {
       /*
        * The client is disconnected. Don't wait forever:
diff --git a/sql/mdl.h b/sql/mdl.h
index fee5cb3d3be77..b6f9e9e5d6e47 100644
--- a/sql/mdl.h
+++ b/sql/mdl.h
@@ -114,9 +114,11 @@ class MDL_context_owner
   virtual THD* get_thd() = 0;
 
   /**
+     @note This interface is kept for backward compatibility.
+     The MDL subsystem now uses THD* directly for better performance.
      @see THD::notify_shared_lock()
    */
-  virtual bool notify_shared_lock(MDL_context_owner *in_use,
+  virtual bool notify_shared_lock(THD *in_use,
                                   bool needs_thr_lock_abort,
                                   bool needs_non_slave_abort) = 0;
 };
@@ -832,7 +834,7 @@ class MDL_wait
   bool set_status(enum_wait_status result_arg);
   enum_wait_status get_status();
   void reset_status();
-  enum_wait_status timed_wait(MDL_context_owner *owner,
+  enum_wait_status timed_wait(THD *owner,
                               struct timespec *abs_timeout,
                               bool signal_timeout,
                               const PSI_stage_info *wait_state_name);
@@ -924,7 +926,7 @@ class MDL_context
   void release_explicit_locks();
   void rollback_to_savepoint(const MDL_savepoint &mdl_savepoint);
 
-  MDL_context_owner *get_owner() { return m_owner; }
+  THD *get_owner() { return m_owner; }
 
   /** @pre Only valid if we started waiting for lock. */
   inline uint get_deadlock_weight() const
@@ -938,7 +940,7 @@ class MDL_context
                     already has received some signal or closed
                     signal slot.
   */
-  void init(MDL_context_owner *arg) { m_owner= arg; reset(); }
+  void init(THD *arg) { m_owner= arg; reset(); }
   void reset() { m_deadlock_overweight= 0; }
 
   void set_needs_thr_lock_abort(bool needs_thr_lock_abort)
@@ -1019,7 +1021,7 @@ class MDL_context
       involved schemas and global intention exclusive lock.
   */
   Ticket_list m_tickets[MDL_DURATION_END];
-  MDL_context_owner *m_owner;
+  THD *m_owner;
   /**
     TRUE -  if for this context we will break protocol and try to
             acquire table-level locks while having only S lock on
@@ -1059,7 +1061,7 @@ class MDL_context
   bool fix_pins();
 
 public:
-  THD *get_thd() const { return m_owner->get_thd(); }
+  THD *get_thd() const { return m_owner; }
   bool has_explicit_locks();
   void find_deadlock();
 
diff --git a/sql/sql_class.cc b/sql/sql_class.cc
index 42723fb6caf62..99279dc2c4d13 100644
--- a/sql/sql_class.cc
+++ b/sql/sql_class.cc
@@ -2232,41 +2232,40 @@ void THD::disconnect()
 }
 
 
-bool THD::notify_shared_lock(MDL_context_owner *ctx_in_use,
+bool THD::notify_shared_lock(THD *ctx_in_use,
                              bool needs_thr_lock_abort,
                              bool needs_non_slave_abort)
 {
-  THD *in_use= ctx_in_use->get_thd();
   bool signalled= FALSE;
   DBUG_ENTER("THD::notify_shared_lock");
   DBUG_PRINT("enter",("needs_thr_lock_abort: %d", needs_thr_lock_abort));
 
   enum killed_state kill_signal;
-  if (in_use->system_thread & SYSTEM_THREAD_DELAYED_INSERT)
+  if (ctx_in_use->system_thread & SYSTEM_THREAD_DELAYED_INSERT)
     kill_signal= KILL_CONNECTION;
-  else if (needs_non_slave_abort && !in_use->slave_thread)
+  else if (needs_non_slave_abort && !ctx_in_use->slave_thread)
     kill_signal= KILL_QUERY;
   else
     kill_signal= NOT_KILLED;
-  if (kill_signal != NOT_KILLED && !in_use->killed)
+  if (kill_signal != NOT_KILLED && !ctx_in_use->killed)
   {
     /* This code is similar to kill_delayed_threads() */
     DBUG_PRINT("info", ("kill delayed thread"));
-    mysql_mutex_lock(&in_use->LOCK_thd_kill);
-    if (in_use->killed < kill_signal)
-      in_use->set_killed_no_mutex(kill_signal);
-    in_use->abort_current_cond_wait(true);
-    mysql_mutex_unlock(&in_use->LOCK_thd_kill);
+    mysql_mutex_lock(&ctx_in_use->LOCK_thd_kill);
+    if (ctx_in_use->killed < kill_signal)
+      ctx_in_use->set_killed_no_mutex(kill_signal);
+    ctx_in_use->abort_current_cond_wait(true);
+    mysql_mutex_unlock(&ctx_in_use->LOCK_thd_kill);
     signalled= TRUE;
   }
 
   if (needs_thr_lock_abort)
   {
-    mysql_mutex_lock(&in_use->LOCK_thd_data);
+    mysql_mutex_lock(&ctx_in_use->LOCK_thd_data);
     /* If not already dying */
-    if (in_use->killed != KILL_CONNECTION_HARD)
+    if (ctx_in_use->killed != KILL_CONNECTION_HARD)
     {
-      for (TABLE *thd_table= in_use->open_tables;
+      for (TABLE *thd_table= ctx_in_use->open_tables;
            thd_table ;
            thd_table= thd_table->next)
       {
@@ -2283,7 +2282,7 @@ bool THD::notify_shared_lock(MDL_context_owner *ctx_in_use,
         }
       }
     }
-    mysql_mutex_unlock(&in_use->LOCK_thd_data);
+    mysql_mutex_unlock(&ctx_in_use->LOCK_thd_data);
   }
   DBUG_RETURN(signalled);
 }
diff --git a/sql/sql_class.h b/sql/sql_class.h
index e8829839617f7..48f2ca82e02b8 100644
--- a/sql/sql_class.h
+++ b/sql/sql_class.h
@@ -4586,7 +4586,7 @@ class THD: public THD_count, /* this must be first */
     @note This function does not wait for the thread to give away its
           locks. Waiting is done outside for all threads at once.
 
-    @param ctx_in_use           The MDL context owner (thread) to wake up.
+    @param ctx_in_use           The THD to wake up.
     @param needs_thr_lock_abort Indicates that to wake up thread
                                 this call needs to abort its waiting
                                 on table-level lock.
@@ -4594,7 +4594,7 @@ class THD: public THD_count, /* this must be first */
     @retval  TRUE  if the thread was woken up
     @retval  FALSE otherwise.
    */
-  bool notify_shared_lock(MDL_context_owner *ctx_in_use,
+  bool notify_shared_lock(THD *ctx_in_use,
                           bool needs_thr_lock_abort,
                           bool needs_non_slave_abort) override;