MDEV-16983 Application-time periods: foreign key PART 3/4

Support self-referencing foreign keys

4 files changed, 202 insertions, 30 deletions

diff --git a/mysql-test/suite/period/r/fk.result b/mysql-test/suite/period/r/fk.result
index ed1fb554b41..daf43d90be0 100644
--- a/mysql-test/suite/period/r/fk.result
+++ b/mysql-test/suite/period/r/fk.result
@@ -289,5 +289,79 @@ alter table s add foreign key(id, x, period fp)
 references t1(id, x, period p)
 on delete restrict;
 ERROR HY000: Fields of `fp` and `test`.`t1`.`p` have different types
+# Check self-referenced table
+create or replace table tg (id int, prev int,
+s date, e date, period for p(s, e),
+unique(id, p without overlaps),
+foreign key(prev, period p) references tg(id, period p));
+show create table tg;
+Table	Create Table
+tg	CREATE TABLE `tg` (
+  `id` int(11) DEFAULT NULL,
+  `prev` int(11) DEFAULT NULL,
+  `s` date NOT NULL,
+  `e` date NOT NULL,
+  PERIOD FOR `p` (`s`, `e`),
+  UNIQUE KEY `id` (`id`,`p` WITHOUT OVERLAPS),
+  KEY `prev` (`prev`,`e`,`s`),
+  CONSTRAINT `tg_ibfk_1` FOREIGN KEY (`prev`, period `p`) REFERENCES `tg` (`id`, period `p`)
+) ENGINE=InnoDB DEFAULT CHARSET=latin1
+insert into tg values(1, 0, '2017-01-03', '2017-01-20');
+ERROR 23000: Cannot add or update a child row: a foreign key constraint fails
+insert into tg values(1, 1, '2017-01-01', '2017-01-22');
+insert into tg values(2, 1, '2017-01-03', '2017-01-20');
+select * from tg where prev>=0;
+id	prev	s	e
+1	1	2017-01-01	2017-01-22
+2	1	2017-01-03	2017-01-20
+delete from tg where id = 1;
+ERROR 23000: Cannot delete or update a parent row: a foreign key constraint fails
+insert into tg values(1, 0, '2017-01-02', '2017-01-22');
+ERROR 23000: Duplicate entry '1-2017-01-22-2017-01-02' for key 'id'
+insert into tg values(1, 0, '2017-01-22', '2017-01-25');
+ERROR 23000: Cannot add or update a child row: a foreign key constraint fails
+insert into tg values(0, 0, '2017-01-20', '2017-01-30');
+select * from tg where id = 0 and s > '2017-01-15';
+id	prev	s	e
+0	0	2017-01-20	2017-01-30
+insert into tg values(1, 0, '2017-01-22', '2017-01-25');
+select * from tg;
+id	prev	s	e
+0	0	2017-01-20	2017-01-30
+1	0	2017-01-22	2017-01-25
+1	1	2017-01-01	2017-01-22
+2	1	2017-01-03	2017-01-20
+delete from tg where id = 1 and e = '2017-01-20';
+select * from tg;
+id	prev	s	e
+0	0	2017-01-20	2017-01-30
+1	0	2017-01-22	2017-01-25
+1	1	2017-01-01	2017-01-22
+2	1	2017-01-03	2017-01-20
+delete from tg where id = 0;
+ERROR 23000: Cannot delete or update a parent row: a foreign key constraint fails
+select * from tg;
+id	prev	s	e
+0	0	2017-01-20	2017-01-30
+1	0	2017-01-22	2017-01-25
+1	1	2017-01-01	2017-01-22
+2	1	2017-01-03	2017-01-20
+update tg set e= '2017-01-25' where id = 0 and e = '2017-01-30';
+select * from tg;
+id	prev	s	e
+0	0	2017-01-20	2017-01-25
+1	0	2017-01-22	2017-01-25
+1	1	2017-01-01	2017-01-22
+2	1	2017-01-03	2017-01-20
+update tg set s= '2017-01-22', e= '2017-01-28' where id = 0 and e = '2017-01-25';
+select * from tg;
+id	prev	s	e
+0	0	2017-01-22	2017-01-28
+1	0	2017-01-22	2017-01-25
+1	1	2017-01-01	2017-01-22
+2	1	2017-01-03	2017-01-20
+delete from tg order by id asc;
+ERROR 23000: Cannot delete or update a parent row: a foreign key constraint fails
+delete from tg order by id desc;
 drop database test;
 create database test;
diff --git a/mysql-test/suite/period/t/fk.test b/mysql-test/suite/period/t/fk.test
index 4c0f0ec0aac..d21d1e4519f 100644
--- a/mysql-test/suite/period/t/fk.test
+++ b/mysql-test/suite/period/t/fk.test
@@ -243,5 +243,53 @@ alter table s add foreign key(id, x, period fp)
                references t1(id, x, period p)
                on delete restrict;
 
+--echo # Check self-referenced table
+create or replace table tg (id int, prev int,
+                            s date, e date, period for p(s, e),
+                            unique(id, p without overlaps),
+                            foreign key(prev, period p) references tg(id, period p));
+show create table tg;
+
+--error ER_NO_REFERENCED_ROW_2
+insert into tg values(1, 0, '2017-01-03', '2017-01-20');
+
+insert into tg values(1, 1, '2017-01-01', '2017-01-22');
+insert into tg values(2, 1, '2017-01-03', '2017-01-20');
+
+--sorted_result
+select * from tg where prev>=0;
+--error ER_ROW_IS_REFERENCED_2
+delete from tg where id = 1;
+
+--error ER_DUP_ENTRY
+insert into tg values(1, 0, '2017-01-02', '2017-01-22');
+--error ER_NO_REFERENCED_ROW_2
+insert into tg values(1, 0, '2017-01-22', '2017-01-25');
+insert into tg values(0, 0, '2017-01-20', '2017-01-30');
+select * from tg where id = 0 and s > '2017-01-15';
+insert into tg values(1, 0, '2017-01-22', '2017-01-25');
+
+--sorted_result
+select * from tg;
+delete from tg where id = 1 and e = '2017-01-20';
+--sorted_result
+select * from tg;
+--error ER_ROW_IS_REFERENCED_2
+delete from tg where id = 0;
+--sorted_result
+select * from tg;
+
+update tg set e= '2017-01-25' where id = 0 and e = '2017-01-30';
+--sorted_result
+select * from tg;
+
+update tg set s= '2017-01-22', e= '2017-01-28' where id = 0 and e = '2017-01-25';
+--sorted_result
+select * from tg;
+
+--error ER_ROW_IS_REFERENCED_2
+delete from tg order by id asc;
+delete from tg order by id desc;
+
 drop database test;
 create database test;
diff --git a/sql/handler.cc b/sql/handler.cc
index c6fa265367f..ad0e574313b 100644
--- a/sql/handler.cc
+++ b/sql/handler.cc
@@ -7079,13 +7079,21 @@ int handler::update_first_row(const uchar *new_data)
   return error;
 }
 
+static bool is_row_self_referencing(const uchar *row, const KEY &fk, const KEY &ref)
+{
+  if (fk.table->s != ref.table->s)
+    return false;
+  DBUG_ASSERT(ref.without_overlaps);
+  return key_period_compare_bases(fk, ref, row, row) == 0;
+}
 
 static int period_find_first_overlapping_record(handler *handler,
                                                 uchar *key_buf,
                                                 const uchar *record_to_cmp,
                                                 uchar *record,
                                                 const KEY &key_to_cmp,
-                                                const KEY &key)
+                                                const KEY &key,
+                                                bool ignore_self_refs)
 {
   /*
     We should evaluate SELECT start, end WHERE start < @end and end > @start.
@@ -7116,13 +7124,16 @@ static int period_find_first_overlapping_record(handler *handler,
 
   while (key_period_compare_bases(key, key_to_cmp, record, record_to_cmp) == 0)
   {
-    int overlap= key_period_compare_periods(key, key_to_cmp,
-                                            record, record_to_cmp);
+    if (!ignore_self_refs || !is_row_self_referencing(record, key, key_to_cmp))
+    {
+      int overlap= key_period_compare_periods(key, key_to_cmp,
+                                              record, record_to_cmp);
 
-    if (overlap > 0) // all the rest will definitely succeed tested record
-      return HA_ERR_KEY_NOT_FOUND;
-    else if (overlap == 0)
-      return 0;
+      if (overlap > 0) // all the rest will definitely succeed tested record
+        return HA_ERR_KEY_NOT_FOUND;
+      else if (overlap == 0)
+        return 0;
+    }
 
     error= handler->ha_index_next(record);
     if (unlikely(error == HA_ERR_END_OF_FILE))
@@ -7146,6 +7157,7 @@ static int period_row_check_delete_for_key(const uchar *record,
                                            const FOREIGN_KEY &fk)
 {
   handler *foreign_handler= fk.foreign_key->table->file;
+  auto *foreign_table= fk.foreign_key->table;
 
   /*
     We shouldn't save cursor here, since this handler is never used.
@@ -7155,9 +7167,14 @@ static int period_row_check_delete_for_key(const uchar *record,
   if(error)
     return error;
 
-  set_bits_with_key(fk.foreign_key->table->read_set,
+  set_bits_with_key(foreign_table->read_set,
                     fk.foreign_key, fk.fields_num);
-  auto *record_buffer= foreign_handler->get_table()->record[0];
+  if (foreign_table->s == fk.referenced_key->table->s)
+  {
+    set_bits_with_key(foreign_table->read_set, fk.referenced_key,
+                      fk.fields_num);
+  }
+  auto *record_buffer= foreign_table->record[0];
   auto *key_buffer= foreign_handler->lookup_buffer;
 
   error= period_find_first_overlapping_record(foreign_handler,
@@ -7165,7 +7182,8 @@ static int period_row_check_delete_for_key(const uchar *record,
                                               record,
                                               record_buffer,
                                               *fk.referenced_key,
-                                              *fk.foreign_key);
+                                              *fk.foreign_key,
+                                              true);
 
   int end_error= foreign_handler->ha_index_end();
 
@@ -7253,7 +7271,7 @@ static int period_check_row_references(handler *ref_handler,
 {
   int error= period_find_first_overlapping_record(ref_handler, key_buf,
                                                   record, ref_record,
-                                                  key, ref_key);
+                                                  key, ref_key, false);
   if (error)
     return error;
 
@@ -7306,6 +7324,9 @@ static int period_check_row_references(handler *ref_handler,
 static int period_row_check_insert_for_key(const uchar *record,
                                            const FOREIGN_KEY &fk)
 {
+  if (is_row_self_referencing(record, *fk.foreign_key, *fk.referenced_key))
+    return 0;
+
   handler *ref_handler= fk.referenced_key->table->file;
 
   int error= ref_handler->ha_index_init(fk.referenced_key_nr, false);
@@ -7335,6 +7356,8 @@ int handler::period_row_ins_fk_check(const uchar *record)
 {
   if (!table->foreign_keys)
     return 0;
+  if (table->versioned() && !table->vers_end_field()->is_max())
+    return 0;
 
   for(int k= 0; k < table->foreign_keys; k++)
   {
diff --git a/sql/sql_table.cc b/sql/sql_table.cc
index 70fecf5ca33..a5a48ee3d1f 100644
--- a/sql/sql_table.cc
+++ b/sql/sql_table.cc
@@ -3743,11 +3743,13 @@ mysql_prepare_create_table(THD *thd, HA_CREATE_INFO *create_info,
 
       if (fk_key->ref_period)
       {
-        auto *ref_table= fk_key->ref_table_list->table->s;
-        if (!fk_key->ref_period.streq(ref_table->period.name))
+        auto &ref_table_period = fk_key->ref_table_list->table ?
+                                 fk_key->ref_table_list->table->s->period.name :
+                                 create_info->period_info.name;
+        if (!fk_key->ref_period.streq(ref_table_period))
         {
           my_error(ER_PERIOD_FK_NOT_FOUND, MYF(0), fk_key->ref_period.str,
-                   ref_table->db.str, ref_table->table_name.str);
+                   fk_key->ref_db.str, fk_key->ref_table.str);
         }
 
         Create_field *period_start= NULL;
@@ -3762,14 +3764,28 @@ mysql_prepare_create_table(THD *thd, HA_CREATE_INFO *create_info,
         }
         DBUG_ASSERT(period_start);
 
-        auto *ref_period_start= ref_table->period.start_field(ref_table);
+        if (fk_key->ref_table_list->table)
+        {
+          auto *ref_table= fk_key->ref_table_list->table->s;
+          auto *ref_period_start= ref_table->period.start_field(ref_table);
 
-        if (ref_period_start->type_handler() != period_start->type_handler()
-            || ref_period_start->pack_length() != period_start->pack_length)
+          if (ref_period_start->type_handler() != period_start->type_handler()
+              || ref_period_start->pack_length() != period_start->pack_length)
+          {
+            my_error(ER_PERIOD_FK_TYPES_MISMATCH, MYF(0), fk_key->period.str,
+                     ref_table->db.str, ref_table->table_name.str,
+                     ref_table->period.name.str);
+          }
+        }
+        else
         {
-          my_error(ER_PERIOD_FK_TYPES_MISMATCH, MYF(0), fk_key->period.str,
-                   ref_table->db.str, ref_table->table_name.str,
-                   ref_table->period.name.str);
+          // table is NULL if this is a self-referencing key
+          DBUG_ASSERT(my_strnncoll(table_alias_charset,
+                                   (uchar *) create_info->alias.str,
+                                   create_info->alias.length,
+                                   (uchar *) fk_key->ref_table.str,
+                                   fk_key->ref_table.length) == 0);
+          // No check is required
         }
       }
       continue;
@@ -4718,20 +4734,31 @@ static bool append_system_key_parts(THD *thd, HA_CREATE_INFO *create_info,
         my_error(ER_PERIOD_NOT_FOUND, MYF(0), key->period.str);
         return true;
       }
-      const auto &period_start= create_info->period_info.period.start;
-      const auto &period_end= create_info->period_info.period.end;
-      key->columns.push_back(new Key_part_spec(&period_end, 0));
-      key->columns.push_back(new Key_part_spec(&period_start, 0));
+      const LEX_CSTRING *period_start= &create_info->period_info.period.start;
+      const LEX_CSTRING *period_end= &create_info->period_info.period.end;
+      key->columns.push_back(new Key_part_spec(period_end, 0));
+      key->columns.push_back(new Key_part_spec(period_start, 0));
 
       if (key->type == Key::FOREIGN_KEY)
       {
         auto *fk= static_cast<Foreign_key*>(key);
-        const auto &ref_period= fk->ref_table_list->table->s->period;
-        const auto *field= fk->ref_table_list->table->field;
-        const auto &ref_period_start= field[ref_period.start_fieldno]->field_name;
-        const auto &ref_period_end= field[ref_period.end_fieldno]->field_name;
-        fk->ref_columns.push_back(new Key_part_spec(&ref_period_end, 0));
-        fk->ref_columns.push_back(new Key_part_spec(&ref_period_start, 0));
+
+        if (fk->ref_table_list->table)
+        {
+          const auto &ref_period= fk->ref_table_list->table->s->period;
+          const auto *field= fk->ref_table_list->table->field;
+          period_start= &field[ref_period.start_fieldno]->field_name;
+          period_end= &field[ref_period.end_fieldno]->field_name;
+        }
+        else
+        {
+          // Self-referencing FK. Field names are already set above
+          DBUG_ASSERT(my_strcasecmp(system_charset_info,
+                                    fk->ref_table_list->table_name.str,
+                                    create_info->alias.str) == 0);
+        }
+        fk->ref_columns.push_back(new Key_part_spec(period_end, 0));
+        fk->ref_columns.push_back(new Key_part_spec(period_start, 0));
       }
     }
   }