MDEV-27909 InnoDB: Failing assertion: state == TRX_STATE_NOT_STARTED ... on DDL

The fix in commit 6e390a62 (MDEV-26772)
was a step to the right direction, but implemented incorrectly.
When an InnoDB persistent statistics table cannot be locked immediately,
we must not let row_mysql_handle_errors() to roll back the transaction.

lock_table_for_trx(): Add the parameter no_wait (default false)
for an immediate return of DB_LOCK_WAIT in case of a conflict.

ha_innobase::delete_table(), ha_innobase::rename_table():
Pass no_wait=true to lock_table_for_trx() when needed,
instead of temporarily setting THDVAR(thd, lock_wait_timeout) to 0.

storage/innobase/handler/ha_innodb.cc

@@ -13547,29 +13547,26 @@ int ha_innobase::delete_table(const char *name)
       dict_sys.unfreeze();
     }
 
-    auto &timeout= THDVAR(thd, lock_wait_timeout);
-    const auto save_timeout= timeout;
-    if (table->name.is_temporary())
-      timeout= 0;
+    const bool skip_wait{table->name.is_temporary()};
 
     if (table_stats && index_stats &&
         !strcmp(table_stats->name.m_name, TABLE_STATS_NAME) &&
         !strcmp(index_stats->name.m_name, INDEX_STATS_NAME) &&
-        !(err= lock_table_for_trx(table_stats, trx, LOCK_X)))
-      err= lock_table_for_trx(index_stats, trx, LOCK_X);
+        !(err= lock_table_for_trx(table_stats, trx, LOCK_X, skip_wait)))
+      err= lock_table_for_trx(index_stats, trx, LOCK_X, skip_wait);
 
-    if (err != DB_SUCCESS && !timeout)
+    if (err != DB_SUCCESS && skip_wait)
     {
       /* We may skip deleting statistics if we cannot lock the tables,
       when the table carries a temporary name. */
+      ut_ad(err == DB_LOCK_WAIT);
+      ut_ad(trx->error_state == DB_SUCCESS);
       err= DB_SUCCESS;
       dict_table_close(table_stats, false, thd, mdl_table);
       dict_table_close(index_stats, false, thd, mdl_index);
       table_stats= nullptr;
       index_stats= nullptr;
     }
-
-    timeout= save_timeout;
   }
 
   if (err == DB_SUCCESS)
@@ -14075,17 +14072,15 @@ ha_innobase::rename_table(
 		if (error == DB_SUCCESS && table_stats && index_stats
 		    && !strcmp(table_stats->name.m_name, TABLE_STATS_NAME)
 		    && !strcmp(index_stats->name.m_name, INDEX_STATS_NAME)) {
-			auto &timeout = THDVAR(thd, lock_wait_timeout);
-			const auto save_timeout = timeout;
-			if (from_temp) {
-				timeout = 0;
-			}
-			error = lock_table_for_trx(table_stats, trx, LOCK_X);
+			error = lock_table_for_trx(table_stats, trx, LOCK_X,
+						   from_temp);
 			if (error == DB_SUCCESS) {
 				error = lock_table_for_trx(index_stats, trx,
-							   LOCK_X);
+							   LOCK_X, from_temp);
 			}
 			if (error != DB_SUCCESS && from_temp) {
+				ut_ad(error == DB_LOCK_WAIT);
+				ut_ad(trx->error_state == DB_SUCCESS);
 				error = DB_SUCCESS;
 				/* We may skip renaming statistics if
 				we cannot lock the tables, when the
@@ -14098,7 +14093,6 @@ ha_innobase::rename_table(
 				table_stats = nullptr;
 				index_stats = nullptr;
 			}
-			timeout = save_timeout;
 		}
 	}
 

storage/innobase/include/lock0lock.h

@@ -394,15 +394,13 @@ lock_table(
 void lock_table_resurrect(dict_table_t *table, trx_t *trx, lock_mode mode);
 
 /** Sets a lock on a table based on the given mode.
-@param[in]	table	table to lock
-@param[in,out]	trx	transaction
-@param[in]	mode	LOCK_X or LOCK_S
-@return error code or DB_SUCCESS. */
-dberr_t
-lock_table_for_trx(
-	dict_table_t*	table,
-	trx_t*		trx,
-	enum lock_mode	mode)
+@param table	table to lock
+@param trx	transaction
+@param mode	LOCK_X or LOCK_S
+@param no_wait  whether to skip handling DB_LOCK_WAIT
+@return error code */
+dberr_t lock_table_for_trx(dict_table_t *table, trx_t *trx, lock_mode mode,
+                           bool no_wait= false)
 	MY_ATTRIBUTE((nonnull, warn_unused_result));
 
 /** Exclusively lock the data dictionary tables.
@@ -915,10 +913,8 @@ class lock_sys_t
   @param page whether to discard also from lock_sys.prdt_hash */
   void prdt_page_free_from_discard(const page_id_t id, bool all= false);
 
-#ifdef WITH_WSREP
   /** Cancel possible lock waiting for a transaction */
   static void cancel_lock_wait_for_trx(trx_t *trx);
-#endif /* WITH_WSREP */
 };
 
 /** The lock system */

storage/innobase/lock/lock0lock.cc

@@ -3627,52 +3627,50 @@ static void lock_table_dequeue(lock_t *in_lock, bool owns_wait_mutex)
 	}
 }
 
-/** Sets a lock on a table based on the given mode.
-@param[in]	table	table to lock
-@param[in,out]	trx	transaction
-@param[in]	mode	LOCK_X or LOCK_S
-@return error code or DB_SUCCESS. */
-dberr_t
-lock_table_for_trx(
-	dict_table_t*	table,
-	trx_t*		trx,
-	enum lock_mode	mode)
-{
-	mem_heap_t*	heap;
-	que_thr_t*	thr;
-	dberr_t		err;
-	sel_node_t*	node;
-	heap = mem_heap_create(512);
-
-	node = sel_node_create(heap);
-	thr = pars_complete_graph_for_exec(node, trx, heap, NULL);
-	thr->graph->state = QUE_FORK_ACTIVE;
-
-	/* We use the select query graph as the dummy graph needed
-	in the lock module call */
 
-	thr = static_cast<que_thr_t*>(
-		que_fork_get_first_thr(
-			static_cast<que_fork_t*>(que_node_get_parent(thr))));
+/** Sets a lock on a table based on the given mode.
+@param table	table to lock
+@param trx	transaction
+@param mode	LOCK_X or LOCK_S
+@param no_wait  whether to skip handling DB_LOCK_WAIT
+@return error code */
+dberr_t lock_table_for_trx(dict_table_t *table, trx_t *trx, lock_mode mode,
+                           bool no_wait)
+{
+  mem_heap_t *heap= mem_heap_create(512);
+  sel_node_t *node= sel_node_create(heap);
+  que_thr_t *thr= pars_complete_graph_for_exec(node, trx, heap, nullptr);
+  thr->graph->state= QUE_FORK_ACTIVE;
+
+  thr= static_cast<que_thr_t*>
+    (que_fork_get_first_thr(static_cast<que_fork_t*>
+                            (que_node_get_parent(thr))));
 
 run_again:
-	thr->run_node = thr;
-	thr->prev_node = thr->common.parent;
-
-	err = lock_table(table, mode, thr);
+  thr->run_node= thr;
+  thr->prev_node= thr->common.parent;
+  dberr_t err= lock_table(table, mode, thr);
 
-	trx->error_state = err;
-
-	if (UNIV_UNLIKELY(err != DB_SUCCESS)) {
-		if (row_mysql_handle_errors(&err, trx, thr, NULL)) {
-			goto run_again;
-		}
-	}
+  switch (err) {
+  case DB_SUCCESS:
+    break;
+  case DB_LOCK_WAIT:
+    if (no_wait)
+    {
+      lock_sys.cancel_lock_wait_for_trx(trx);
+      break;
+    }
+    /* fall through */
+  default:
+    trx->error_state= err;
+    if (row_mysql_handle_errors(&err, trx, thr, nullptr))
+      goto run_again;
+  }
 
-	que_graph_free(thr->graph);
-	trx->op_info = "";
+  que_graph_free(thr->graph);
+  trx->op_info= "";
 
-	return(err);
+  return err;
 }
 
 /** Exclusively lock the data dictionary tables.
@@ -5639,8 +5637,7 @@ static void lock_cancel_waiting_and_release(lock_t *lock)
   lock_wait_end(trx);
   trx->mutex_unlock();
 }
-#ifdef WITH_WSREP
-TRANSACTIONAL_TARGET
+
 void lock_sys_t::cancel_lock_wait_for_trx(trx_t *trx)
 {
   lock_sys.wr_lock(SRW_LOCK_CALL);
@@ -5654,7 +5651,6 @@ void lock_sys_t::cancel_lock_wait_for_trx(trx_t *trx)
   lock_sys.wr_unlock();
   mysql_mutex_unlock(&lock_sys.wait_mutex);
 }
-#endif /* WITH_WSREP */
 
 /** Cancel a waiting lock request.
 @tparam check_victim  whether to check for DB_DEADLOCK