MDEV-21052 InnoDB foreign key refactoring for TABLE_SHARE::foreign_keysbb-10.5-midenok-MDEV-16417

Refactor dict_load_foreigns() for synchronising TABLE_SHARE foreign
data with dict_table_t cache.

Remove a number of routines working with SYS_FOREIGNS and
SYS_FOREIGN_COLS. innobase_update_foreign_try() is now used solely for
ER_FK_INCORRECT_OPTION check.

Prelock parent tables as well as child tables. This is done for the
case when parent table doesn't know about its children when they
created before parent with foreign_key_checks=0. Opening the parent
table initiates fk_resolve_referenced_keys() which updates its
referenced_keys. Due to CREATE TABLE doesn't not know about "illegal"
children it can not check for foreign consistency. F.ex. this would
succeed:

set foreign_key_checks= 0;
create table child (fk int references parent (id)) engine=innodb;
set foreign_key_checks= 1;

create table parent (id bigint primary key) engine=innodb;

In the above case dict_load_foreigns() deduces which tables are
unknown to opened parent (tables_missing) and reloads their foreign
data via recursion. Infinite recursion is not possible via test case:
a table cannot be "parent after child" and "child before parent"
simultaneously. Though infinite recursion is possible via malicously
crafted FRM file, there is no protection from that at InnoDB level but
there is protection at SQL level: thd->fk_circular_check.

Later though it would not allow DML on child as well as on parent (see
innodb.foreign_key MDEV-10083). So this is pretty acceptable:
foreign_key_checks is unnormal setting, checking parent on CREATE
TABLE would impose all frms scanning which is not acceptable.

ha_innobase::open() then synchronizes these referenced_keys with its
referenced_set cache by calling dict_load_foreigns().

Disable self-references on same column. The Bug 12902967 restricted
them on some condition of "same column/index" (see
innodb_bug12902967.test), though such self-references were not
completely disabled (see other self-ref cases changed in this
patch). It is not clear why they worked if they are "self-refs on same
column/index".

1 files changed, 25 insertions, 6 deletions

diff --git a/sql/sql_class.cc b/sql/sql_class.cc
index fd4c2ab0853..4446cd0e83e 100644
--- a/sql/sql_class.cc
+++ b/sql/sql_class.cc
@@ -288,22 +288,41 @@ bool foreign_key_prefix(Key *a, Key *b)
 /*
   @brief
   Check if the foreign key options are compatible with the specification
-  of the columns on which the key is created
+  of the columns on which the key is created.
+
+  Check self-references for self-column pointing.
 
   @retval
     FALSE   The foreign key options are compatible with key columns
   @retval
     TRUE    Otherwise
 */
-bool Foreign_key::validate(List<Create_field> &table_fields)
+bool Foreign_key::validate(const LEX_CSTRING &db, const LEX_CSTRING &table_name,
+                           List<Create_field> &table_fields, bool &self_ref)
 {
   Create_field  *sql_field;
-  Key_part_spec *column;
-  List_iterator<Key_part_spec> cols(columns);
-  List_iterator<Create_field> it(table_fields);
+  Key_part_spec *column, *rcol;
+  List_iterator_fast<Key_part_spec> cols(columns);
+  List_iterator_fast<Key_part_spec> rcols(ref_columns);
+  List_iterator_fast<Create_field> it(table_fields);
+  // NB: self-references may have no ref_table
+  self_ref= !ref_table.str || ((!ref_db.str || 0 == cmp_table(db, ref_db)) &&
+                               0 == cmp_table(table_name, ref_table));
   DBUG_ENTER("Foreign_key::validate");
+  const char *nam= constraint_name.str ? constraint_name.str : name.str;
+  if (columns.elements != ref_columns.elements)
+  {
+    my_error(ER_WRONG_FK_DEF, MYF(0), (nam ? nam : "foreign key without name"));
+    DBUG_RETURN(TRUE);
+  }
   while ((column= cols++))
   {
+    rcol= rcols++;
+    if (self_ref && 0 == cmp_ident(column->field_name, rcol->field_name))
+    {
+      my_error(ER_WRONG_FK_DEF, MYF(0), (nam ? nam : rcol->field_name.str));
+      DBUG_RETURN(TRUE);
+    }
     it.rewind();
     while ((sql_field= it++) &&
            lex_string_cmp(system_charset_info,
@@ -6201,7 +6220,7 @@ int THD::decide_logging_format(TABLE_LIST *tables)
 
       replicated_tables_count++;
 
-      if (tbl->prelocking_placeholder != TABLE_LIST::PRELOCK_FK)
+      if (tbl->prelocking_placeholder < TABLE_LIST::PRELOCK_FK)
       {
         if (tbl->lock_type <= TL_READ_NO_INSERT)
           has_read_tables= true;