MDEV-24122: check mysql.user structure in more detailbb-10.2-danielblack-MDEV-24122-m57-incorrect-user-table-alignment_part2

In checking all tables in Table_check_intact, continue
checking the details if there are more fields than expected.

With such a check a startup of an incorrectly ordered
mysql.user table (as per the bug report) would be:

2020-11-10 11:19:49 140340878554880 [ERROR] Incorrect definition of table mysql.user: expected column 'is_role' at position 43, found 'password_lifetime'.
2020-11-10 11:19:49 140340878554880 [ERROR] Incorrect definition of table mysql.user: expected column 'is_role' at position 43 to have type enum('N','Y'), found type smallint(5) unsigned.
2020-11-10 11:19:49 140340878554880 [ERROR] Incorrect definition of table mysql.user: expected column 'default_role' at position 44, found 'account_locked'.
2020-11-10 11:19:49 140340878554880 [ERROR] Incorrect definition of table mysql.user: expected column 'default_role' at position 44 to have type char(80, found type enum('N','Y').
2020-11-10 11:19:49 140340878554880 [ERROR] Incorrect definition of table mysql.user: expected column 'max_statement_time' at position 45, found 'is_role'.
2020-11-10 11:19:49 140340878554880 [ERROR] Incorrect definition of table mysql.user: expected column 'max_statement_time' at position 45 to have type decimal(12,6), found type enum('N','Y').
2020-11-10 11:19:49 140340878554880 [Warning] mysql.user table is in unsupported structure, run mysql_upgrade or alter to expected structure
2020-11-10 11:19:49 140340878247680 [Note] Column count of mysql.user is wrong. Expected 46, found 49. The table is probably corrupted

12 files changed, 361 insertions, 85 deletions

diff --git a/mysql-test/r/grant_4332.result b/mysql-test/r/grant_4332.result
index b62ca1a20cc..a651326569d 100644
--- a/mysql-test/r/grant_4332.result
+++ b/mysql-test/r/grant_4332.result
@@ -1,5 +1,6 @@
 set GLOBAL sql_mode="";
 set LOCAL sql_mode="";
+call mtr.add_suppression('mysql.user table is in unsupported structure');
 alter table mysql.user modify User char(16) binary not null default '';
 alter table mysql.db modify User char(16) binary not null default '';
 alter table mysql.tables_priv modify User char(16) binary not null default '';
diff --git a/mysql-test/r/mysql_upgrade.result b/mysql-test/r/mysql_upgrade.result
index 814ce2e99ec..ae4c64de77b 100644
--- a/mysql-test/r/mysql_upgrade.result
+++ b/mysql-test/r/mysql_upgrade.result
@@ -735,6 +735,8 @@ DROP ROLE aRole;
 #
 DROP TABLE IF EXISTS mysql.user;
 FLUSH TABLES mysql.user;
+call mtr.add_suppression('mysql.user table is in unsupported structure');
+call mtr.add_suppression('Column count of mysql.user is wrong.');
 ALTER TABLE mysql.user ADD is_role enum('N', 'Y') COLLATE utf8_general_ci DEFAULT 'N' NOT NULL;
 ALTER TABLE mysql.user ADD default_role char(80) binary DEFAULT '' NOT NULL;
 ALTER TABLE mysql.user ADD max_statement_time decimal(12,6) DEFAULT 0 NOT NULL;
diff --git a/mysql-test/r/no_password_column-mdev-11170.result b/mysql-test/r/no_password_column-mdev-11170.result
index 81eecc95558..3d96600b576 100644
--- a/mysql-test/r/no_password_column-mdev-11170.result
+++ b/mysql-test/r/no_password_column-mdev-11170.result
@@ -59,6 +59,8 @@ max_statement_time	decimal(12,6)	NO		0.000000
 #
 # Drop the password column.
 #
+call mtr.add_suppression('mysql.user table is in unsupported structure');
+call mtr.add_suppression('Column count of mysql.user is wrong.');
 alter table mysql.user drop column password;
 flush privileges;
 #
diff --git a/mysql-test/suite/roles/create_and_drop_role_invalid_user_table.result b/mysql-test/suite/roles/create_and_drop_role_invalid_user_table.result
index 1a269d72eb7..0f540af83b8 100644
--- a/mysql-test/suite/roles/create_and_drop_role_invalid_user_table.result
+++ b/mysql-test/suite/roles/create_and_drop_role_invalid_user_table.result
@@ -1,5 +1,7 @@
 connect  mysql, localhost, root,,;
 use mysql;
+call mtr.add_suppression('mysql.user table is in unsupported structure');
+call mtr.add_suppression('Column count of mysql.user is wrong.');
 alter table user drop column is_role;
 alter table user drop column default_role;
 alter table user drop column max_statement_time;
diff --git a/mysql-test/suite/roles/create_and_drop_role_invalid_user_table.test b/mysql-test/suite/roles/create_and_drop_role_invalid_user_table.test
index ebd75c34ca1..ad4415722ce 100644
--- a/mysql-test/suite/roles/create_and_drop_role_invalid_user_table.test
+++ b/mysql-test/suite/roles/create_and_drop_role_invalid_user_table.test
@@ -10,6 +10,9 @@ use mysql;
 #
 # downgrade the table to pre-default-role structure
 #
+call mtr.add_suppression('mysql.user table is in unsupported structure');
+call mtr.add_suppression('Column count of mysql.user is wrong.');
+
 alter table user drop column is_role;
 alter table user drop column default_role;
 alter table user drop column max_statement_time;
diff --git a/mysql-test/t/grant_4332.test b/mysql-test/t/grant_4332.test
index 41e0b822f98..a74de03e64e 100644
--- a/mysql-test/t/grant_4332.test
+++ b/mysql-test/t/grant_4332.test
@@ -9,6 +9,8 @@
 set GLOBAL sql_mode="";
 set LOCAL sql_mode="";
 
+call mtr.add_suppression('mysql.user table is in unsupported structure');
+
 alter table mysql.user modify User char(16) binary not null default '';
 alter table mysql.db modify User char(16) binary not null default '';
 alter table mysql.tables_priv modify User char(16) binary not null default '';
diff --git a/mysql-test/t/mysql_upgrade.test b/mysql-test/t/mysql_upgrade.test
index a3b1575cb6d..4bff4ec9e7a 100644
--- a/mysql-test/t/mysql_upgrade.test
+++ b/mysql-test/t/mysql_upgrade.test
@@ -263,6 +263,9 @@ DROP TABLE IF EXISTS mysql.user;
 --copy_file std_data/mysql57user.MYD $MYSQLD_DATADIR/mysql/user.MYD
 FLUSH TABLES mysql.user;
 
+call mtr.add_suppression('mysql.user table is in unsupported structure');
+call mtr.add_suppression('Column count of mysql.user is wrong.');
+
 # What prior to MDEV-23201 would of done:
 ALTER TABLE mysql.user ADD is_role enum('N', 'Y') COLLATE utf8_general_ci DEFAULT 'N' NOT NULL;
 ALTER TABLE mysql.user ADD default_role char(80) binary DEFAULT '' NOT NULL;
diff --git a/mysql-test/t/no_password_column-mdev-11170.test b/mysql-test/t/no_password_column-mdev-11170.test
index 2cc4ba82ee8..4b923a18585 100644
--- a/mysql-test/t/no_password_column-mdev-11170.test
+++ b/mysql-test/t/no_password_column-mdev-11170.test
@@ -17,6 +17,9 @@ describe mysql.user;
 --echo #
 --echo # Drop the password column.
 --echo #
+call mtr.add_suppression('mysql.user table is in unsupported structure');
+call mtr.add_suppression('Column count of mysql.user is wrong.');
+
 alter table mysql.user drop column password;
 flush privileges;
 
diff --git a/sql/sp.cc b/sql/sp.cc
index b9757887e7a..422d81c065d 100644
--- a/sql/sp.cc
+++ b/sql/sp.cc
@@ -350,16 +350,17 @@ public:
   Proc_table_intact() : m_print_once(TRUE) { has_keys= TRUE; }
 
 protected:
-  void report_error(uint code, const char *fmt, ...);
+  void report(enum loglevel level, uint code, const char *fmt, ...);
 };
 
 
 /**
   Report failure to validate the mysql.proc table definition.
   Print a message to the error log only once.
+  loglevel is ignored.
 */
 
-void Proc_table_intact::report_error(uint code, const char *fmt, ...)
+void Proc_table_intact::report(enum loglevel level, uint code, const char *fmt, ...)
 {
   va_list args;
   char buf[512];
diff --git a/sql/sql_acl.cc b/sql/sql_acl.cc
index 641ab69c2a7..6a449cf3986 100644
--- a/sql/sql_acl.cc
+++ b/sql/sql_acl.cc
@@ -930,6 +930,247 @@ class Grant_table_base
   TABLE_LIST tl;
 };
 
+static const
+TABLE_FIELD_TYPE user_table_fields[] =
+{
+  {
+    { C_STRING_WITH_LEN("Host") },
+    { C_STRING_WITH_LEN("char(60)") },
+    { C_STRING_WITH_LEN("utf8") }
+  },
+  {
+    { C_STRING_WITH_LEN("User") },
+    { C_STRING_WITH_LEN("char(80)") },
+    { C_STRING_WITH_LEN("utf8") }
+  },
+  {
+    { C_STRING_WITH_LEN("Password") },
+    { C_STRING_WITH_LEN("char(41)") },
+    { NULL, 0 }
+  },
+  {
+    { C_STRING_WITH_LEN("Select_priv") },
+    { C_STRING_WITH_LEN("enum('N','Y')") },
+    { NULL, 0 }
+  },
+  {
+    { C_STRING_WITH_LEN("Insert_priv") },
+    { C_STRING_WITH_LEN("enum('N','Y')") },
+    { NULL, 0 }
+  },
+  {
+    { C_STRING_WITH_LEN("Update_priv") },
+    { C_STRING_WITH_LEN("enum('N','Y')") },
+    { NULL, 0 }
+  },
+  {
+    { C_STRING_WITH_LEN("Delete_priv") },
+    { C_STRING_WITH_LEN("enum('N','Y')") },
+    { NULL, 0 }
+  },
+  {
+    { C_STRING_WITH_LEN("Create_priv") },
+    { C_STRING_WITH_LEN("enum('N','Y')") },
+    { NULL, 0 }
+  },
+  {
+    { C_STRING_WITH_LEN("Drop_priv") },
+    { C_STRING_WITH_LEN("enum('N','Y')") },
+    { NULL, 0 }
+  },
+  {
+    { C_STRING_WITH_LEN("Reload_priv") },
+    { C_STRING_WITH_LEN("enum('N','Y')") },
+    { NULL, 0 }
+  },
+  {
+    { C_STRING_WITH_LEN("Shutdown_priv") },
+    { C_STRING_WITH_LEN("enum('N','Y')") },
+    { NULL, 0 }
+  },
+  {
+    { C_STRING_WITH_LEN("Process_priv") },
+    { C_STRING_WITH_LEN("enum('N','Y')") },
+    { NULL, 0 }
+  },
+  {
+    { C_STRING_WITH_LEN("File_priv") },
+    { C_STRING_WITH_LEN("enum('N','Y')") },
+    { NULL, 0 }
+  },
+  {
+    { C_STRING_WITH_LEN("Grant_priv") },
+    { C_STRING_WITH_LEN("enum('N','Y')") },
+    { NULL, 0 }
+  },
+  {
+    { C_STRING_WITH_LEN("References_priv") },
+    { C_STRING_WITH_LEN("enum('N','Y')") },
+    { NULL, 0 }
+  },
+  {
+    { C_STRING_WITH_LEN("Index_priv") },
+    { C_STRING_WITH_LEN("enum('N','Y')") },
+    { NULL, 0 }
+  },
+  {
+    { C_STRING_WITH_LEN("Alter_priv") },
+    { C_STRING_WITH_LEN("enum('N','Y')") },
+    { NULL, 0 }
+  },
+  {
+    { C_STRING_WITH_LEN("Show_db_priv") },
+    { C_STRING_WITH_LEN("enum('N','Y')") },
+    { NULL, 0 }
+  },
+  {
+    { C_STRING_WITH_LEN("Super_priv") },
+    { C_STRING_WITH_LEN("enum('N','Y')") },
+    { NULL, 0 }
+  },
+  {
+    { C_STRING_WITH_LEN("Create_tmp_table_priv") },
+    { C_STRING_WITH_LEN("enum('N','Y')") },
+    { NULL, 0 }
+  },
+  {
+    { C_STRING_WITH_LEN("Lock_tables_priv") },
+    { C_STRING_WITH_LEN("enum('N','Y')") },
+    { NULL, 0 }
+  },
+  {
+    { C_STRING_WITH_LEN("Execute_priv") },
+    { C_STRING_WITH_LEN("enum('N','Y')") },
+    { NULL, 0 }
+  },
+  {
+    { C_STRING_WITH_LEN("Repl_slave_priv") },
+    { C_STRING_WITH_LEN("enum('N','Y')") },
+    { NULL, 0 }
+  },
+  {
+    { C_STRING_WITH_LEN("Repl_client_priv") },
+    { C_STRING_WITH_LEN("enum('N','Y')") },
+    { NULL, 0 }
+  },
+  {
+    { C_STRING_WITH_LEN("Create_view_priv") },
+    { C_STRING_WITH_LEN("enum('N','Y')") },
+    { NULL, 0 }
+  },
+  {
+    { C_STRING_WITH_LEN("Show_view_priv") },
+    { C_STRING_WITH_LEN("enum('N','Y')") },
+    { NULL, 0 }
+  },
+  {
+    { C_STRING_WITH_LEN("Create_routine_priv") },
+    { C_STRING_WITH_LEN("enum('N','Y')") },
+    { NULL, 0 }
+  },
+  {
+    { C_STRING_WITH_LEN("Alter_routine_priv") },
+    { C_STRING_WITH_LEN("enum('N','Y')") },
+    { NULL, 0 }
+  },
+  {
+    { C_STRING_WITH_LEN("Create_user_priv") },
+    { C_STRING_WITH_LEN("enum('N','Y')") },
+    { NULL, 0 }
+  },
+  {
+    { C_STRING_WITH_LEN("Event_priv") },
+    { C_STRING_WITH_LEN("enum('N','Y')") },
+    { NULL, 0 }
+  },
+  {
+    { C_STRING_WITH_LEN("Trigger_priv") },
+    { C_STRING_WITH_LEN("enum('N','Y')") },
+    { NULL, 0 }
+  },
+  {
+    { C_STRING_WITH_LEN("Create_tablespace_priv") },
+    { C_STRING_WITH_LEN("enum('N','Y')") },
+    { NULL, 0 }
+  },
+  {
+    { C_STRING_WITH_LEN("ssl_type") },
+    { C_STRING_WITH_LEN("enum('','ANY','X509','SPECIFIED')") },
+    { NULL, 0 }
+  },
+  {
+    { C_STRING_WITH_LEN("ssl_cipher") },
+    { C_STRING_WITH_LEN("blob") },
+    { NULL, 0 }
+  },
+  {
+    { C_STRING_WITH_LEN("x509_issuer") },
+    { C_STRING_WITH_LEN("blob") },
+    { NULL, 0 }
+  },
+  {
+    { C_STRING_WITH_LEN("x509_subject") },
+    { C_STRING_WITH_LEN("blob") },
+    { NULL, 0 }
+  },
+  {
+    { C_STRING_WITH_LEN("max_questions") },
+    { C_STRING_WITH_LEN("int") },
+    { NULL, 0 }
+  },
+  {
+    { C_STRING_WITH_LEN("max_updates") },
+    { C_STRING_WITH_LEN("int") },
+    { NULL, 0 }
+  },
+  {
+    { C_STRING_WITH_LEN("max_connections") },
+    { C_STRING_WITH_LEN("int") },
+    { NULL, 0 }
+  },
+  {
+    { C_STRING_WITH_LEN("max_user_connections") },
+    { C_STRING_WITH_LEN("int") },
+    { NULL, 0 }
+  },
+  {
+    { C_STRING_WITH_LEN("plugin") },
+    { C_STRING_WITH_LEN("char(64)") },
+    { NULL, 0 }
+  },
+  {
+    { C_STRING_WITH_LEN("authentication_string") },
+    { C_STRING_WITH_LEN("text") },
+    { C_STRING_WITH_LEN("utf8") }
+  },
+  {
+    { C_STRING_WITH_LEN("password_expired") },
+    { C_STRING_WITH_LEN("enum('N','Y')") },
+    { NULL, 0 }
+  },
+  {
+    { C_STRING_WITH_LEN("is_role") },
+    { C_STRING_WITH_LEN("enum('N','Y')") },
+    { NULL, 0 }
+  },
+  {
+    { C_STRING_WITH_LEN("default_role") },
+    { C_STRING_WITH_LEN("char(80") },
+    { C_STRING_WITH_LEN("utf8") }
+  },
+  {
+    { C_STRING_WITH_LEN("max_statement_time") },
+    { C_STRING_WITH_LEN("decimal(12,6)") },
+    { NULL, 0 }
+  }
+};
+
+static const TABLE_FIELD_DEF
+user_table_def= {array_elements(user_table_fields), user_table_fields, 0, (uint*) 0 };
+
+/** Single instance used to control printing to the error log. */
+static Table_check_intact_log_error user_table_intact;
+
 class User_table: public Grant_table_base
 {
  public:
@@ -1978,6 +2219,15 @@ static bool acl_load(THD *thd, const Grant_tables& tables)
     }
   }
 
+  if (!opt_bootstrap && !mysql_user_table_is_in_short_password_format &&
+      user_table_intact.check(user_table.table(), &user_table_def)
+          == Table_check_intact::CHECK_BAD)
+  {
+    sql_print_warning("mysql.user table is in "
+                      "unsupported structure, run mysql_upgrade or "
+		      "alter to expected structure");
+  }
+
   allow_all_hosts=0;
   while (!(read_record_info.read_record(&read_record_info)))
   {
diff --git a/sql/table.cc b/sql/table.cc
index e4492f21a30..db3fe975637 100644
--- a/sql/table.cc
+++ b/sql/table.cc
@@ -4212,19 +4212,17 @@ bool check_column_name(const char *name)
   @param[in] table_def         Expected structure of the table (column name
                                and type)
 
-  @retval  FALSE  OK
-  @retval  TRUE   There was an error. An error message is output
-                  to the error log.  We do not push an error
-                  message into the error stack because this
-                  function is currently only called at start up,
-                  and such errors never reach the user.
+  @retval  CHECK_OK            No difference
+  @retval  CHECK_BAD           Significant error in column defination
+  @retval  CHECK_COL_COUNT     Too many columns present, but those present are of
+                               correct type.
 */
 
-bool
+enum Table_check_intact::check_result
 Table_check_intact::check(TABLE *table, const TABLE_FIELD_DEF *table_def)
 {
   uint i;
-  my_bool error= FALSE;
+  check_result error= CHECK_OK;
   const TABLE_FIELD_TYPE *field_def= table_def->field;
   DBUG_ENTER("table_check_intact");
   DBUG_PRINT("info",("table: %s  expected_count: %d",
@@ -4237,35 +4235,43 @@ Table_check_intact::check(TABLE *table, const TABLE_FIELD_DEF *table_def)
   if (table->s->fields != table_def->count)
   {
     THD *thd= current_thd;
+    /* Too many columns, only informational warning, too few is definately an error */
+    enum loglevel level= (table->s->fields < table_def->count) ? ERROR_LEVEL : INFORMATION_LEVEL;
     DBUG_PRINT("info", ("Column count has changed, checking the definition"));
 
     /* previous MySQL version */
     if (MYSQL_VERSION_ID > table->s->mysql_version)
     {
-      report_error(ER_COL_COUNT_DOESNT_MATCH_PLEASE_UPDATE,
-                   ER_THD(thd, ER_COL_COUNT_DOESNT_MATCH_PLEASE_UPDATE),
-                   table->alias.c_ptr(), table_def->count, table->s->fields,
-                   static_cast<int>(table->s->mysql_version),
-                   MYSQL_VERSION_ID);
-      DBUG_RETURN(TRUE);
+      report(level, ER_COL_COUNT_DOESNT_MATCH_PLEASE_UPDATE,
+             ER_THD(thd, ER_COL_COUNT_DOESNT_MATCH_PLEASE_UPDATE),
+             table->alias.c_ptr(), table_def->count, table->s->fields,
+             static_cast<int>(table->s->mysql_version),
+             MYSQL_VERSION_ID);
     }
     else if (MYSQL_VERSION_ID == table->s->mysql_version)
     {
-      report_error(ER_COL_COUNT_DOESNT_MATCH_CORRUPTED_V2,
-                   ER_THD(thd, ER_COL_COUNT_DOESNT_MATCH_CORRUPTED_V2),
-                   table->s->db.str, table->s->table_name.str,
-                   table_def->count, table->s->fields);
-      DBUG_RETURN(TRUE);
+      report(level, ER_COL_COUNT_DOESNT_MATCH_CORRUPTED_V2,
+             ER_THD(thd, ER_COL_COUNT_DOESNT_MATCH_CORRUPTED_V2),
+             table->s->db.str, table->s->table_name.str,
+             table_def->count, table->s->fields);
+    }
+    if (level == ERROR_LEVEL)
+    {
+      DBUG_RETURN(CHECK_BAD);
     }
     /*
-      Something has definitely changed, but we're running an older
-      version of MySQL with new system tables.
-      Let's check column definitions. If a column was added at
+      Our current column count error can be overwritten to the more
+      serious CHECK_BAD being the column defination is wrong
+    */
+    error= CHECK_COL_COUNT;
+    /*
+      Something has definitely changed, since there is sufficent
+      columns let's check their definitions. If a column was added at
       the end of the table, then we don't care much since such change
       is backward compatible.
     */
   }
-  else
+
   {
   StringBuffer<1024> sql_type(system_charset_info);
   sql_type.extra_allocation(256); // Allocate min 256 characters at once
@@ -4284,11 +4290,11 @@ Table_check_intact::check(TABLE *table, const TABLE_FIELD_DEF *table_def)
           Still this can be a sign of a tampered table, output an error
           to the error log.
         */
-        report_error(0, "Incorrect definition of table %s.%s: "
-                     "expected column '%s' at position %d, found '%s'.",
-                     table->s->db.str, table->alias.c_ptr(),
-                     field_def->name.str, i,
-                     field->field_name);
+        report(ERROR_LEVEL, 0, "Incorrect definition of table %s.%s: "
+               "expected column '%s' at position %d, found '%s'.",
+               table->s->db.str, table->alias.c_ptr(),
+               field_def->name.str, i,
+               field->field_name);
       }
       field->sql_type(sql_type);
       /*
@@ -4311,45 +4317,45 @@ Table_check_intact::check(TABLE *table, const TABLE_FIELD_DEF *table_def)
       if (strncmp(sql_type.c_ptr_safe(), field_def->type.str,
                   field_def->type.length - 1))
       {
-        report_error(0, "Incorrect definition of table %s.%s: "
-                     "expected column '%s' at position %d to have type "
-                     "%s, found type %s.", table->s->db.str,
-                     table->alias.c_ptr(),
-                     field_def->name.str, i, field_def->type.str,
-                     sql_type.c_ptr_safe());
-        error= TRUE;
+        report(ERROR_LEVEL, 0, "Incorrect definition of table %s.%s: "
+               "expected column '%s' at position %d to have type "
+               "%s, found type %s.", table->s->db.str,
+               table->alias.c_ptr(),
+               field_def->name.str, i, field_def->type.str,
+               sql_type.c_ptr_safe());
+        error= CHECK_BAD;
       }
       else if (field_def->cset.str && !field->has_charset())
       {
-        report_error(0, "Incorrect definition of table %s.%s: "
-                     "expected the type of column '%s' at position %d "
-                     "to have character set '%s' but the type has no "
-                     "character set.", table->s->db.str,
-                     table->alias.c_ptr(),
-                     field_def->name.str, i, field_def->cset.str);
-        error= TRUE;
+        report(ERROR_LEVEL, 0, "Incorrect definition of table %s.%s: "
+               "expected the type of column '%s' at position %d "
+               "to have character set '%s' but the type has no "
+               "character set.", table->s->db.str,
+               table->alias.c_ptr(),
+               field_def->name.str, i, field_def->cset.str);
+        error= CHECK_BAD;
       }
       else if (field_def->cset.str &&
                strcmp(field->charset()->csname, field_def->cset.str))
       {
-        report_error(0, "Incorrect definition of table %s.%s: "
-                     "expected the type of column '%s' at position %d "
-                     "to have character set '%s' but found "
-                     "character set '%s'.", table->s->db.str,
-                     table->alias.c_ptr(),
-                     field_def->name.str, i, field_def->cset.str,
-                     field->charset()->csname);
-        error= TRUE;
+        report(ERROR_LEVEL, 0, "Incorrect definition of table %s.%s: "
+               "expected the type of column '%s' at position %d "
+               "to have character set '%s' but found "
+               "character set '%s'.", table->s->db.str,
+               table->alias.c_ptr(),
+               field_def->name.str, i, field_def->cset.str,
+               field->charset()->csname);
+        error= CHECK_BAD;
       }
     }
     else
     {
-      report_error(0, "Incorrect definition of table %s.%s: "
-                   "expected column '%s' at position %d to have type %s "
-                   " but the column is not found.",
-                   table->s->db.str, table->alias.c_ptr(),
-                   field_def->name.str, i, field_def->type.str);
-      error= TRUE;
+      report(ERROR_LEVEL, 0, "Incorrect definition of table %s.%s: "
+             "expected column '%s' at position %d to have type %s "
+             " but the column is not found.",
+             table->s->db.str, table->alias.c_ptr(),
+             field_def->name.str, i, field_def->type.str);
+      error= CHECK_BAD;
     }
   }
   }
@@ -4358,22 +4364,22 @@ Table_check_intact::check(TABLE *table, const TABLE_FIELD_DEF *table_def)
   {
     if (table->s->primary_key == MAX_KEY)
     {
-      report_error(0, "Incorrect definition of table %s.%s: "
-                   "missing primary key.", table->s->db.str,
-                   table->alias.c_ptr());
-      error= TRUE;
+      report(ERROR_LEVEL, 0, "Incorrect definition of table %s.%s: "
+             "missing primary key.", table->s->db.str,
+             table->alias.c_ptr());
+      error= CHECK_BAD;
     }
     else
     {
       KEY *pk= &table->s->key_info[table->s->primary_key];
       if (pk->user_defined_key_parts != table_def->primary_key_parts)
       {
-        report_error(0, "Incorrect definition of table %s.%s: "
-                     "Expected primary key to have %u columns, but instead "
-                     "found %u columns.", table->s->db.str,
-                     table->alias.c_ptr(), table_def->primary_key_parts,
-                     pk->user_defined_key_parts);
-        error= TRUE;
+        report(ERROR_LEVEL, 0, "Incorrect definition of table %s.%s: "
+               "Expected primary key to have %u columns, but instead "
+               "found %u columns.", table->s->db.str,
+               table->alias.c_ptr(), table_def->primary_key_parts,
+               pk->user_defined_key_parts);
+        error= CHECK_BAD;
       }
       else
       {
@@ -4381,41 +4387,41 @@ Table_check_intact::check(TABLE *table, const TABLE_FIELD_DEF *table_def)
         {
           if (table_def->primary_key_columns[i] + 1 != pk->key_part[i].fieldnr)
           {
-            report_error(0, "Incorrect definition of table %s.%s: Expected "
-                         "primary key part %u to refer to column %u, but "
-                         "instead found column %u.", table->s->db.str,
-                         table->alias.c_ptr(), i + 1,
-                         table_def->primary_key_columns[i] + 1,
-                         pk->key_part[i].fieldnr);
-            error= TRUE;
+            report(ERROR_LEVEL, 0, "Incorrect definition of table %s.%s: Expected "
+                   "primary key part %u to refer to column %u, but "
+                   "instead found column %u.", table->s->db.str,
+                   table->alias.c_ptr(), i + 1,
+                   table_def->primary_key_columns[i] + 1,
+                   pk->key_part[i].fieldnr);
+            error= CHECK_BAD;
           }
         }
       }
     }
   }
 
-  if (! error)
+  if (error == CHECK_OK)
     table->s->table_field_def_cache= table_def;
 
 end:
 
   if (has_keys && !error && !table->key_info)
   {
-    report_error(0, "Incorrect definition of table %s.%s: "
-                 "indexes are missing",
-                 table->s->db.str, table->alias.c_ptr());
-    error= TRUE;
+    report(ERROR_LEVEL, 0, "Incorrect definition of table %s.%s: "
+           "indexes are missing",
+           table->s->db.str, table->alias.c_ptr());
+    error= CHECK_BAD;
   }
 
   DBUG_RETURN(error);
 }
 
 
-void Table_check_intact_log_error::report_error(uint, const char *fmt, ...)
+void Table_check_intact_log_error::report(enum loglevel level, uint, const char *fmt, ...)
 {
   va_list args;
   va_start(args, fmt);
-  error_log_print(ERROR_LEVEL, fmt, args);
+  error_log_print(level, fmt, args);
   va_end(args);
 }
 
diff --git a/sql/table.h b/sql/table.h
index 9a864f7ce9f..b919db6785f 100644
--- a/sql/table.h
+++ b/sql/table.h
@@ -488,14 +488,15 @@ class Table_check_intact
 {
 protected:
   bool has_keys;
-  virtual void report_error(uint code, const char *fmt, ...)= 0;
+  virtual void report(enum loglevel level, uint code, const char *fmt, ...)= 0;
 
 public:
   Table_check_intact(bool keys= false) : has_keys(keys) {}
   virtual ~Table_check_intact() {}
 
+  enum check_result { CHECK_OK= 0, CHECK_BAD, CHECK_COL_COUNT };
   /** Checks whether a table is intact. */
-  bool check(TABLE *table, const TABLE_FIELD_DEF *table_def);
+  enum check_result check(TABLE *table, const TABLE_FIELD_DEF *table_def);
 };
 
 
@@ -505,7 +506,7 @@ public:
 class Table_check_intact_log_error : public Table_check_intact
 {
 protected:
-  void report_error(uint, const char *fmt, ...);
+  void report(enum loglevel, uint, const char *fmt, ...);
 public:
   Table_check_intact_log_error() : Table_check_intact(true) {}
 };