MDEV-15890 Strange error message if you try to FLUSH TABLES <view> after LOCK TABLES <view>.

Check if the argument of the FLUSH TABLE is a VIEW and handle it
accordingly.

mysql-test/r/flush.result

@@ -496,3 +496,27 @@ flush relay logs,relay logs;
 ERROR HY000: Incorrect usage of FLUSH and RELAY LOGS
 flush slave,slave;
 ERROR HY000: Incorrect usage of FLUSH and SLAVE
+#
+# MDEV-15890 Strange error message if you try to
+#     FLUSH TABLES <view> after LOCK TABLES <view>.
+#
+CREATE TABLE t1 (qty INT, price INT);
+CREATE VIEW v1 AS SELECT qty, price, qty*price AS value FROM t1;
+LOCK TABLES v1 READ;
+FLUSH TABLES v1;
+ERROR HY000: Table 't1' was locked with a READ lock and can't be updated
+UNLOCK TABLES;
+LOCK TABLES v1 WRITE;
+FLUSH TABLES v1;
+ERROR HY000: Table 't1' was locked with a READ lock and can't be updated
+UNLOCK TABLES;
+LOCK TABLES v1 READ;
+FLUSH TABLES t1;
+ERROR HY000: Table 't1' was locked with a READ lock and can't be updated
+UNLOCK TABLES;
+LOCK TABLES t1 READ;
+FLUSH TABLES v1;
+ERROR HY000: Table 'v1' was not locked with LOCK TABLES
+UNLOCK TABLES;
+DROP VIEW v1;
+DROP TABLE t1;

mysql-test/t/flush.test

@@ -709,3 +709,35 @@ DROP TABLE t1;
 flush relay logs,relay logs;
 --error ER_WRONG_USAGE
 flush slave,slave;
+
+--echo #
+--echo # MDEV-15890 Strange error message if you try to
+--echo #     FLUSH TABLES <view> after LOCK TABLES <view>.
+--echo #
+
+CREATE TABLE t1 (qty INT, price INT);
+CREATE VIEW v1 AS SELECT qty, price, qty*price AS value FROM t1;
+
+LOCK TABLES v1 READ;
+--error ER_TABLE_NOT_LOCKED_FOR_WRITE
+FLUSH TABLES v1;
+UNLOCK TABLES;
+
+LOCK TABLES v1 WRITE;
+--error ER_TABLE_NOT_LOCKED_FOR_WRITE
+FLUSH TABLES v1;
+UNLOCK TABLES;
+
+LOCK TABLES v1 READ;
+--error ER_TABLE_NOT_LOCKED_FOR_WRITE
+FLUSH TABLES t1;
+UNLOCK TABLES;
+
+LOCK TABLES t1 READ;
+--error ER_TABLE_NOT_LOCKED
+FLUSH TABLES v1;
+UNLOCK TABLES;
+
+DROP VIEW v1;
+DROP TABLE t1;
+

sql/sql_base.cc

@@ -522,9 +522,10 @@ bool close_cached_tables(THD *thd, TABLE_LIST *tables,
     for (TABLE_LIST *table_list= tables_to_reopen; table_list;
          table_list= table_list->next_global)
     {
+      int err;
       /* A check that the table was locked for write is done by the caller. */
       TABLE *table= find_table_for_mdl_upgrade(thd, table_list->db,
-                                               table_list->table_name, TRUE);
+                                               table_list->table_name, &err);
 
       /* May return NULL if this table has already been closed via an alias. */
       if (! table)
@@ -2121,6 +2122,66 @@ open_table_get_mdl_lock(THD *thd, Open_table_context *ot_ctx,
 }
 
 
+/**
+  Check if the given table is actually a VIEW that was LOCK-ed
+
+  @param thd            Thread context.
+  @param t              Table to check.
+
+  @retval TRUE  The 't'-table is a locked view
+                needed to remedy problem before retrying again.
+  @retval FALSE 't' was not locked, not a VIEW or an error happened.
+*/
+bool is_locked_view(THD *thd, TABLE_LIST *t)
+{
+  DBUG_ENTER("check_locked_view");
+  /*
+   Is this table a view and not a base table?
+   (it is work around to allow to open view with locked tables,
+   real fix will be made after definition cache will be made)
+
+   Since opening of view which was not explicitly locked by LOCK
+   TABLES breaks metadata locking protocol (potentially can lead
+   to deadlocks) it should be disallowed.
+  */
+  if (thd->mdl_context.is_lock_owner(MDL_key::TABLE,
+                                     t->db, t->table_name,
+                                     MDL_SHARED))
+  {
+    char path[FN_REFLEN + 1];
+    build_table_filename(path, sizeof(path) - 1,
+                         t->db, t->table_name, reg_ext, 0);
+    /*
+      Note that we can't be 100% sure that it is a view since it's
+      possible that we either simply have not found unused TABLE
+      instance in THD::open_tables list or were unable to open table
+      during prelocking process (in this case in theory we still
+      should hold shared metadata lock on it).
+    */
+    if (dd_frm_is_view(thd, path))
+    {
+      /*
+        If parent_l of the table_list is non null then a merge table
+        has this view as child table, which is not supported.
+      */
+      if (t->parent_l)
+      {
+        my_error(ER_WRONG_MRG_TABLE, MYF(0));
+        DBUG_RETURN(FALSE);
+      }
+
+      if (!tdc_open_view(thd, t, t->alias, CHECK_METADATA_VERSION))
+      {
+        DBUG_ASSERT(t->view != 0);
+        DBUG_RETURN(TRUE); // VIEW
+      }
+    }
+  }
+
+  DBUG_RETURN(FALSE);
+}
+
+
 /**
   Open a base table.
 
@@ -2263,50 +2324,10 @@ bool open_table(THD *thd, TABLE_LIST *table_list, Open_table_context *ot_ctx)
       DBUG_PRINT("info",("Using locked table"));
       goto reset;
     }
-    /*
-      Is this table a view and not a base table?
-      (it is work around to allow to open view with locked tables,
-      real fix will be made after definition cache will be made)
 
-      Since opening of view which was not explicitly locked by LOCK
-      TABLES breaks metadata locking protocol (potentially can lead
-      to deadlocks) it should be disallowed.
-    */
-    if (thd->mdl_context.is_lock_owner(MDL_key::TABLE,
-                                       table_list->db,
-                                       table_list->table_name,
-                                       MDL_SHARED))
-    {
-      char path[FN_REFLEN + 1];
-      build_table_filename(path, sizeof(path) - 1,
-                           table_list->db, table_list->table_name, reg_ext, 0);
-      /*
-        Note that we can't be 100% sure that it is a view since it's
-        possible that we either simply have not found unused TABLE
-        instance in THD::open_tables list or were unable to open table
-        during prelocking process (in this case in theory we still
-        should hold shared metadata lock on it).
-      */
-      if (dd_frm_is_view(thd, path))
-      {
-        /*
-          If parent_l of the table_list is non null then a merge table
-          has this view as child table, which is not supported.
-        */
-        if (table_list->parent_l)
-        {
-          my_error(ER_WRONG_MRG_TABLE, MYF(0));
-          DBUG_RETURN(true);
-        }
+    if (is_locked_view(thd, table_list))
+      DBUG_RETURN(FALSE); // VIEW
 
-        if (!tdc_open_view(thd, table_list, alias, key, key_length,
-                           CHECK_METADATA_VERSION))
-        {
-          DBUG_ASSERT(table_list->view != 0);
-          DBUG_RETURN(FALSE); // VIEW
-        }
-      }
-    }
     /*
       No table in the locked tables list. In case of explicit LOCK TABLES
       this can happen if a user did not include the table into the list.
@@ -2666,8 +2687,9 @@ TABLE *find_locked_table(TABLE *list, const char *db, const char *table_name)
    @param thd        Thread context
    @param db         Database name.
    @param table_name Name of table.
-   @param no_error   Don't emit error if no suitable TABLE
-                     instance were found.
+   @param p_error    In the case of an error (when the function returns NULL)
+                     the error number is stored there.
+                     If the p_error is NULL, function launches the error itself.
 
    @note This function checks if the connection holds a global IX
          metadata lock. If no such lock is found, it is not safe to
@@ -2680,15 +2702,15 @@ TABLE *find_locked_table(TABLE *list, const char *db, const char *table_name)
 */
 
 TABLE *find_table_for_mdl_upgrade(THD *thd, const char *db,
-                                  const char *table_name, bool no_error)
+                                  const char *table_name, int *p_error)
 {
   TABLE *tab= find_locked_table(thd->open_tables, db, table_name);
+  int error;
 
   if (!tab)
   {
-    if (!no_error)
-      my_error(ER_TABLE_NOT_LOCKED, MYF(0), table_name);
-    return NULL;
+    error= ER_TABLE_NOT_LOCKED;
+    goto err_exit;
   }
 
   /*
@@ -2700,9 +2722,8 @@ TABLE *find_table_for_mdl_upgrade(THD *thd, const char *db,
   if (!thd->mdl_context.is_lock_owner(MDL_key::GLOBAL, "", "",
                                       MDL_INTENTION_EXCLUSIVE))
   {
-    if (!no_error)
-      my_error(ER_TABLE_NOT_LOCKED_FOR_WRITE, MYF(0), table_name);
-    return NULL;
+    error= ER_TABLE_NOT_LOCKED_FOR_WRITE;
+    goto err_exit;
   }
 
   while (tab->mdl_ticket != NULL &&
@@ -2710,10 +2731,21 @@ TABLE *find_table_for_mdl_upgrade(THD *thd, const char *db,
          (tab= find_locked_table(tab->next, db, table_name)))
     continue;
 
-  if (!tab && !no_error)
-    my_error(ER_TABLE_NOT_LOCKED_FOR_WRITE, MYF(0), table_name);
+  if (unlikely(!tab))
+  {
+    error= ER_TABLE_NOT_LOCKED_FOR_WRITE;
+    goto err_exit;
+  }
 
   return tab;
+
+err_exit:
+  if (p_error)
+    *p_error= error;
+  else
+    my_error(error, MYF(0), table_name);
+
+  return NULL;
 }
 
 
@@ -4446,7 +4478,7 @@ open_tables_check_upgradable_mdl(THD *thd, TABLE_LIST *tables_start,
       Note that find_table_for_mdl_upgrade() will report an error if
       no suitable ticket is found.
     */
-    if (!find_table_for_mdl_upgrade(thd, table->db, table->table_name, false))
+    if (!find_table_for_mdl_upgrade(thd, table->db, table->table_name, NULL))
       return TRUE;
   }
 

sql/sql_base.h

@@ -126,6 +126,7 @@ TABLE *open_ltable(THD *thd, TABLE_LIST *table_list, thr_lock_type update,
                             MYSQL_OPEN_GET_NEW_TABLE |\
                             MYSQL_OPEN_HAS_MDL_LOCK)
 
+bool is_locked_view(THD *thd, TABLE_LIST *t);
 bool open_table(THD *thd, TABLE_LIST *table_list, Open_table_context *ot_ctx);
 
 bool get_key_map_from_key_list(key_map *map, TABLE *table,
@@ -329,7 +330,7 @@ static inline bool tdc_open_view(THD *thd, TABLE_LIST *table_list,
 
 TABLE *find_table_for_mdl_upgrade(THD *thd, const char *db,
                                   const char *table_name,
-                                  bool no_error);
+                                  int *p_error);
 void mark_tmp_table_for_reuse(TABLE *table);
 
 int update_virtual_fields(THD *thd, TABLE *table,

sql/sql_reload.cc

@@ -288,9 +288,18 @@ bool reload_acl_and_cache(THD *thd, unsigned long long options,
         */
         if (tables)
         {
+          int err;
           for (TABLE_LIST *t= tables; t; t= t->next_local)
-            if (!find_table_for_mdl_upgrade(thd, t->db, t->table_name, false))
-              return 1;
+            if (!find_table_for_mdl_upgrade(thd, t->db, t->table_name, &err))
+            {
+              if (is_locked_view(thd, t))
+                t->next_local= t->next_global;
+              else
+              {
+                my_error(err, MYF(0), t->table_name);
+                return 1;
+              }
+            }
         }
         else
         {

sql/sql_table.cc

@@ -2070,7 +2070,7 @@ bool mysql_rm_table(THD *thd,TABLE_LIST *tables, my_bool if_exists,
             in its elements.
           */
           table->table= find_table_for_mdl_upgrade(thd, table->db,
-                                                   table->table_name, false);
+                                                   table->table_name, NULL);
           if (!table->table)
             DBUG_RETURN(true);
           table->mdl_request.ticket= table->table->mdl_ticket;

sql/sql_trigger.cc

@@ -531,7 +531,7 @@ bool mysql_create_or_drop_trigger(THD *thd, TABLE_LIST *tables, bool create)
     /* Under LOCK TABLES we must only accept write locked tables. */
     if (!(tables->table= find_table_for_mdl_upgrade(thd, tables->db,
                                                     tables->table_name,
-                                                    FALSE)))
+                                                    NULL)))
       goto end;
   }
   else

sql/sql_truncate.cc

@@ -302,7 +302,7 @@ bool Sql_cmd_truncate_table::lock_table(THD *thd, TABLE_LIST *table_ref,
   if (thd->locked_tables_mode)
   {
     if (!(table= find_table_for_mdl_upgrade(thd, table_ref->db,
-                                            table_ref->table_name, FALSE)))
+                                            table_ref->table_name, NULL)))
       DBUG_RETURN(TRUE);
 
     *hton_can_recreate= ha_check_storage_engine_flag(table->s->db_type(),