update check

3 files changed, 198 insertions, 79 deletions

diff --git a/mysql-test/suite/period/t/fk.test b/mysql-test/suite/period/t/fk.test
index 48904ffd26e..586de9c1840 100644
--- a/mysql-test/suite/period/t/fk.test
+++ b/mysql-test/suite/period/t/fk.test
@@ -90,20 +90,13 @@ select * from s;
 update t set x= 2 where s='2017-01-03' and e='2017-01-20';
 
 
---echo #
---echo # False-positives
---echo #
-
 --echo # Expand left
---error ER_ROW_IS_REFERENCED_2
 update t set s= '2017-01-01' where s = '2017-01-03' and e = '2017-01-20';
 
 --echo # Shrink left
---error ER_ROW_IS_REFERENCED_2
 update t set s= '2017-01-05' where e = '2017-01-20';
 
 --echo # Expand right
---error ER_ROW_IS_REFERENCED_2
 update t set e= '2017-02-10' where s = '2017-01-26' and e = '2017-01-30';
 
 --echo # Shrink right
@@ -114,31 +107,24 @@ delete from s where s = '2017-01-27' and e = '2017-01-30';
 update t set e= '2017-01-29' where s = '2017-01-26';
 
 --echo # Shrink both
---echo # Not a false-positive
 --error ER_ROW_IS_REFERENCED_2
 update t set s= '2017-01-27', e= '2017-01-28' where x = 2;
---echo # False-positive
---error ER_ROW_IS_REFERENCED_2
+
 update t set s= '2017-01-28', e= '2017-01-29' where x = 2;
 
 --echo # Expand both
---error ER_ROW_IS_REFERENCED_2
 update t set s= '2017-01-20', e= '2017-02-05' where x = 2;
 
 --echo # Move right
---echo # Not a false-positive
 --error ER_ROW_IS_REFERENCED_2
 update t set s= '2017-02-02', e= '2017-02-25' where x = 2;
---echo # False-positive
---error ER_ROW_IS_REFERENCED_2
+
 update t set s= '2017-01-28', e= '2017-02-25' where x = 2;
 
 --echo # Move left
---echo # Not a false-positive
 --error ER_ROW_IS_REFERENCED_2
 update t set s= '2017-01-20', e= '2017-01-27' where x = 2;
---echo # False-positive
---error ER_ROW_IS_REFERENCED_2
+
 update t set s= '2017-01-20', e= '2017-01-29' where x = 2;
 
 
diff --git a/sql/field.h b/sql/field.h
index d6c2fa4a00c..438a67bf45c 100644
--- a/sql/field.h
+++ b/sql/field.h
@@ -1112,6 +1112,11 @@ public:
     my_ptrdiff_t l_offset= (my_ptrdiff_t) (record -  table->record[0]);
     return ptr + l_offset;
   }
+  uchar *ptr_in_record( uchar *record) const
+  {
+    my_ptrdiff_t l_offset= (my_ptrdiff_t) (record -  table->record[0]);
+    return ptr + l_offset;
+  }
   virtual int set_default();
 
   bool has_update_default_function() const
diff --git a/sql/handler.cc b/sql/handler.cc
index fa1e1df8bdb..fcbeb84cb69 100644
--- a/sql/handler.cc
+++ b/sql/handler.cc
@@ -6921,6 +6921,47 @@ void set_bits_with_key(MY_BITMAP *map, const KEY *key, uint key_parts)
     bitmap_set_bit(map, key->key_part[i].fieldnr - 1);
 }
 
+static int period_row_check_delete_for_key(const uchar *record, FOREIGN_KEY *fk)
+{
+
+  handler *foreign_handler= fk->foreign_key->table->file;
+
+  /* We shouldn't save iterator here, since this handler is never used.
+     The foreign table is only opened for FK matches. */
+  int error= foreign_handler->ha_index_init(fk->foreign_key_nr, false);
+  if(error)
+    return error;
+
+
+  set_bits_with_key(fk->foreign_key->table->read_set,
+                    fk->foreign_key, fk->fields_num);
+  auto *record_buffer= foreign_handler->get_table()->record[0];
+  auto *key_buffer= foreign_handler->get_table()->record[1];
+
+  key_copy(key_buffer, record, fk->referenced_key, 0);
+  error= period_find_overlapping_record(foreign_handler,
+                                        key_buffer,
+                                        record,
+                                        record_buffer,
+                                        *fk->referenced_key,
+                                        *fk->foreign_key);
+
+  int end_error= foreign_handler->ha_index_end();
+
+  if (error == HA_ERR_KEY_NOT_FOUND) // no key matched
+    error= 0;
+
+  if (error || end_error)
+    return error ? error : end_error;
+
+  // if records overlap, then we can't delete a record
+  if (key_period_compare_periods(*fk->referenced_key, *fk->foreign_key,
+                                 record, record_buffer) == 0)
+    error= HA_ERR_ROW_IS_REFERENCED;
+
+  return error;
+}
+
 /**
 @param record                      record to update from, or a deleted record */
 int handler::period_row_del_fk_check(const uchar *record)
@@ -6938,40 +6979,9 @@ int handler::period_row_del_fk_check(const uchar *record)
     DBUG_ASSERT(fk->fields_num == fk->foreign_key->user_defined_key_parts);
     DBUG_ASSERT(fk->fields_num == fk->referenced_key->user_defined_key_parts);
 
-    handler *foreign_handler= fk->foreign_key->table->file;
-
-    /* We shouldn't save iterator here, since this handler is never used.
-       The foreign table is only opened for FK matches. */
-    int error= foreign_handler->ha_index_init(fk->foreign_key_nr, false);
-    if(error)
-      return error;
-
-
-    set_bits_with_key(fk->foreign_key->table->read_set,
-                      fk->foreign_key, fk->fields_num);
-    auto *record_buffer= foreign_handler->get_table()->record[0];
-    auto *key_buffer= foreign_handler->get_table()->record[1];
-
-    key_copy(key_buffer, record, fk->referenced_key, 0);
-    error= period_find_overlapping_record(foreign_handler,
-                                          key_buffer,
-                                          record,
-                                          record_buffer,
-                                          *fk->referenced_key,
-                                          *fk->foreign_key);
-
-    int end_error= foreign_handler->ha_index_end();
-
-    if (error == HA_ERR_KEY_NOT_FOUND) // no key matched
-      return 0;
-    else if (error)
-      return error;
-    else if (end_error)
-      return end_error;
-    else if (key_period_compare_periods(*fk->referenced_key,
-                                        *fk->foreign_key,
-                                        record, record_buffer) == 0)
-      return HA_ERR_ROW_IS_REFERENCED; // == 0 means records overlap
+    int res= period_row_check_delete_for_key(record, fk);
+    if(res)
+      return res;
   }
   return 0;
 }
@@ -7072,6 +7082,32 @@ static int period_check_row_references(handler *ref_handler,
   return 0;
 }
 
+static int period_row_check_insert_for_key(const uchar *record, FOREIGN_KEY *fk)
+{
+  handler *ref_handler= fk->referenced_key->table->file;
+
+  int error= ref_handler->ha_index_init(fk->foreign_key_nr, false);
+  if(error)
+    return error;
+
+  set_bits_with_key(fk->referenced_key->table->read_set,
+                    fk->referenced_key, fk->fields_num);
+
+  auto *ref_record= ref_handler->get_table()->record[0];
+  auto *key_buffer= ref_handler->get_table()->record[1];
+
+  key_copy(key_buffer, record, fk->foreign_key, 0);
+  bool row_references;
+  error= period_check_row_references(ref_handler, key_buffer,
+                                     record, ref_record,
+                                     *fk->foreign_key, *fk->referenced_key);
+  if (error == HA_ERR_KEY_NOT_FOUND)
+    error= HA_ERR_NO_REFERENCED_ROW;
+
+  int end_error= ref_handler->ha_index_end();
+
+  return error ? error : end_error;
+}
 
 int handler::period_row_ins_fk_check(const uchar *record)
 {
@@ -7088,43 +7124,135 @@ int handler::period_row_ins_fk_check(const uchar *record)
     DBUG_ASSERT(fk->fields_num == fk->foreign_key->user_defined_key_parts);
     DBUG_ASSERT(fk->fields_num == fk->referenced_key->user_defined_key_parts);
 
-    handler *ref_handler= fk->referenced_key->table->file;
+    int res= period_row_check_insert_for_key(record, fk);
+    if (res)
+      return res;
+  }
+  return 0;
+}
 
-    int error= ref_handler->ha_index_init(fk->foreign_key_nr, false);
-    if(error)
-      return error;
 
-    set_bits_with_key(fk->referenced_key->table->read_set,
-                      fk->referenced_key, fk->fields_num);
+int handler::period_row_upd_fk_check(const uchar *old_data, const uchar *new_data)
+{
+  if (!table->s->period.name)
+    return 0;
+  /*
+   1. If base key parts changed, or old and new periods do not overlap,
+      emit delete + insert checks.
+   2. If period key part is updated then there can be three operations:
+      grow, shrink or move.
+      1. Move is enlarge + shrink.
+      2. Each operation can be left or right.
+      3. Move right is (enlarge right + shrink left) and vice-versa.
+      4. Enlarge emits insert check on enlarged area.
+         Shrink emits delete check on shrinked area.
+ */
+  if (!check_overlaps_buffer)
+    check_overlaps_buffer= (uchar*)alloc_root(&table_share->mem_root,
+                                              table_share->max_unique_length
+                                              + table_share->reclength);
+  auto *start_field= table->field[table->s->period.start_fieldno];
+  auto *end_field= table->field[table->s->period.end_fieldno];
+  auto field_len= start_field->pack_length();
+
+  auto record_dup= check_overlaps_buffer;
+  // note that these two store pointers to fields in record_dup
+  Binary_value dup_start(start_field->ptr_in_record(record_dup), field_len);
+  Binary_value dup_end(end_field->ptr_in_record(record_dup), field_len);
+
+  uchar storage[4][Type_handler_datetime::hires_bytes(MAX_DATETIME_PRECISION)];
+  Binary_value old_start(storage[0], field_len);
+  Binary_value old_end(storage[1], field_len);
+  Binary_value new_start(storage[2], field_len);
+  Binary_value new_end(storage[3], field_len);
+  old_start.fill(start_field->ptr_in_record(old_data));
+  old_end.fill(end_field->ptr_in_record(old_data));
+  new_start.fill(start_field->ptr_in_record(new_data));
+  new_end.fill(end_field->ptr_in_record(new_data));
+
+  // we will use record_dup instead of old_data to modify period values
+  memcpy(record_dup, old_data, table->s->reclength);
+
+  List <FOREIGN_KEY> fk_list= get_parent_foreign_keys(table);
+  List_iterator<FOREIGN_KEY> fk_list_it(fk_list);
+  while(auto fk= fk_list_it++)
+  {
+    if (!fk->has_period)
+      continue;
 
-    auto *ref_record= ref_handler->get_table()->record[0];
-    auto *key_buffer= ref_handler->get_table()->record[1];
+    DBUG_ASSERT(fk->fields_num == fk->foreign_key->user_defined_key_parts);
+    DBUG_ASSERT(fk->fields_num == fk->referenced_key->user_defined_key_parts);
 
-    key_copy(key_buffer, record, fk->foreign_key, 0);
-    bool row_references;
-    error= period_check_row_references(ref_handler, key_buffer,
-                                       record, ref_record,
-                                       *fk->foreign_key, *fk->referenced_key);
+    const auto &key= *fk->referenced_key;
+    int error= 0;
+    if (TABLE::check_period_overlaps(key, key, old_data, new_data) != 0)
+    {
+      error= period_row_check_delete_for_key(old_data, fk);
+      if (error)
+        return error;
+      continue;
+    }
 
-    int end_error= ref_handler->ha_index_end();
+    if (old_start < new_start)
+    {
+      dup_end.fill(new_start);
+      error= period_row_check_delete_for_key(record_dup, fk);
+      if (error)
+        return error;
+      dup_end.fill(old_end);
+    }
 
-    if (error == HA_ERR_KEY_NOT_FOUND)
-      return HA_ERR_NO_REFERENCED_ROW;
-    else if (error)
-      return error;
-    else if (end_error)
-      return end_error;
+    if (old_end > new_end)
+    {
+      dup_start.fill(new_end);
+      error= period_row_check_delete_for_key(record_dup, fk);
+      if (error)
+        return error;
+      dup_start.fill(old_start);
+    }
   }
-  return 0;
-}
 
+  memcpy(record_dup, new_data, table->s->reclength);
 
-int handler::period_row_upd_fk_check(const uchar *old_data, const uchar *new_data)
-{
-  int error= period_row_del_fk_check(old_data);
-  if (!error)
-    error= period_row_ins_fk_check(new_data);
-  return error;
+  List <FOREIGN_KEY> child_fk_list= get_child_foreign_keys(table);
+  List_iterator<FOREIGN_KEY> child_fk_list_it(child_fk_list);
+  while(auto fk= child_fk_list_it++)
+  {
+    if (!fk->has_period)
+      continue;
+
+    DBUG_ASSERT(fk->fields_num == fk->foreign_key->user_defined_key_parts);
+    DBUG_ASSERT(fk->fields_num == fk->referenced_key->user_defined_key_parts);
+
+    const auto &key= *fk->foreign_key;
+    int error= 0;
+    if (TABLE::check_period_overlaps(key, key, old_data, new_data) != 0)
+    {
+      error= period_row_check_insert_for_key(new_data, fk);
+      if (error)
+        return error;
+      continue;
+    }
+
+    if (old_start > new_start)
+    {
+      dup_end.fill(old_start);
+      error= period_row_check_insert_for_key(record_dup, fk);
+      if (error)
+        return error;
+      dup_end.fill(new_end);
+    }
+
+    if (old_end < new_end)
+    {
+      dup_start.fill(old_end);
+      error= period_row_check_insert_for_key(record_dup, fk);
+      if (error)
+        return error;
+      dup_start.fill(new_start);
+    }
+  }
+  return 0;
 }
 
 int handler::ha_delete_row(const uchar *buf)