MDEV-10139 Support for SEQUENCE objects

- SETVAL(sequence_name, next_value, is_used, round)
- ALTER SEQUENCE, including RESTART WITH

Other things:
- Added handler::extra() option HA_EXTRA_PREPARE_FOR_ALTER_TABLE to signal
  ha_sequence() that it should allow write_row statments.
- ALTER ONLINE TABLE now works with SEQUENCE:s

1 files changed, 252 insertions, 68 deletions

diff --git a/sql/sql_sequence.cc b/sql/sql_sequence.cc
index 6b0a03a4e07..a6e7b073251 100644
--- a/sql/sql_sequence.cc
+++ b/sql/sql_sequence.cc
@@ -22,6 +22,7 @@
 #include "sql_base.h"
 #include "transaction.h"
 #include "lock.h"
+#include "sql_acl.h"
 
 struct Field_definition
 {
@@ -162,7 +163,7 @@ void sequence_definition::store_fields(TABLE *table)
 
 
 /*
-  Check the sequence fields through seq_fields when create sequence.qq
+  Check the sequence fields through seq_fields when createing a sequence.
 
   RETURN VALUES
     false       Success
@@ -269,7 +270,6 @@ bool sequence_insert(THD *thd, LEX *lex, TABLE_LIST *table_list)
   Reprepare_observer *save_reprepare_observer;
   sequence_definition *seq= lex->create_info.seq_create_info;
   bool temporary_table= table_list->table != 0;
-  MY_BITMAP *save_write_set;
   DBUG_ENTER("sequence_insert");
 
   /* If not temporary table */
@@ -316,33 +316,7 @@ bool sequence_insert(THD *thd, LEX *lex, TABLE_LIST *table_list)
   }
 
   seq->reserved_until= seq->start;
-  seq->store_fields(table);
-  /* Store the sequence values in table share */
-  table->s->sequence->copy(seq);
-
-  /*
-    Sequence values will be replicated as a statement
-    like 'create sequence'. So disable binary log temporarily
-  */
-  tmp_disable_binlog(thd);
-  save_write_set= table->write_set;
-  table->write_set= &table->s->all_set;
-  table->s->sequence->initialized= SEQUENCE::SEQ_IN_PREPARE;
-  error= table->file->ha_write_row(table->record[0]);
-  table->s->sequence->initialized= SEQUENCE::SEQ_UNINTIALIZED;
-  reenable_binlog(thd);
-  table->write_set= save_write_set;
-
-  if (error)
-    table->file->print_error(error, MYF(0));
-  else
-  {
-    /*
-      Sequence structure is up to date and table has one row,
-      sequence is now usable
-    */
-    table->s->sequence->initialized= SEQUENCE::SEQ_READY_TO_USE;
-  }
+  error= seq->write_initial_sequence(table);
 
   trans_commit_stmt(thd);
   trans_commit_implicit(thd);
@@ -463,7 +437,7 @@ int SEQUENCE::read_stored_values()
     DBUG_RETURN(error);
   }
   read_fields(table);
-  adjust_values();
+  adjust_values(reserved_until);
 
   all_values_used= 0;
   DBUG_RETURN(0);
@@ -474,12 +448,12 @@ int SEQUENCE::read_stored_values()
   Adjust values after reading a the stored state
 */
 
-void SEQUENCE::adjust_values()
+void SEQUENCE::adjust_values(longlong next_value)
 {
-  offset= 0;
-  next_free_value= reserved_until;
+  next_free_value= next_value;
   if (!(real_increment= increment))
   {
+    longlong offset= 0;
     longlong off, to_add;
     /* Use auto_increment_increment and auto_increment_offset */
 
@@ -515,6 +489,72 @@ void SEQUENCE::adjust_values()
 
 
 /**
+   Write initial sequence information for CREATE and ALTER to sequence table
+*/
+
+int sequence_definition::write_initial_sequence(TABLE *table)
+{
+  int error;
+  THD *thd= table->in_use;
+  MY_BITMAP *save_write_set;
+
+  store_fields(table);
+  /* Store the sequence values in table share */
+  table->s->sequence->copy(this);
+  /*
+    Sequence values will be replicated as a statement
+    like 'create sequence'. So disable binary log temporarily
+  */
+  tmp_disable_binlog(thd);
+  save_write_set= table->write_set;
+  table->write_set= &table->s->all_set;
+  table->s->sequence->initialized= SEQUENCE::SEQ_IN_PREPARE;
+  error= table->file->ha_write_row(table->record[0]);
+  table->s->sequence->initialized= SEQUENCE::SEQ_UNINTIALIZED;
+  reenable_binlog(thd);
+  table->write_set= save_write_set;
+  if (error)
+    table->file->print_error(error, MYF(0));
+  else
+  {
+    /*
+      Sequence structure is up to date and table has one row,
+      sequence is now usable
+    */
+    table->s->sequence->initialized= SEQUENCE::SEQ_READY_TO_USE;
+  }
+  return error;
+}
+
+
+/**
+   Store current sequence values into the sequence table
+*/
+
+int sequence_definition::write(TABLE *table)
+{
+  int error;
+  MY_BITMAP *save_rpl_write_set, *save_write_set;
+
+  /* Log a full insert (ok as table is small) */
+  save_rpl_write_set= table->rpl_write_set;
+
+  /* Update table */
+  save_write_set= table->write_set;
+  table->rpl_write_set= table->write_set= &table->s->all_set;
+  store_fields(table);
+  /* Tell ha_sequence::write_row that we already hold the mutex */
+  ((ha_sequence*) table->file)->sequence_locked= 1;
+  if ((error= table->file->ha_write_row(table->record[0])))
+    table->file->print_error(error, MYF(0));
+  ((ha_sequence*) table->file)->sequence_locked= 0;
+  table->rpl_write_set= save_rpl_write_set;
+  table->write_set= save_write_set;
+  return error;
+}
+
+
+/**
    Get next value for sequence
 
    @param in   table  Sequence table
@@ -546,7 +586,6 @@ longlong SEQUENCE::next_value(TABLE *table, bool second_round, int *error)
 {
   longlong res_value, org_reserved_until, add_to;
   bool out_of_values;
-  MY_BITMAP *save_rpl_write_set, *save_write_set;
   DBUG_ENTER("SEQUENCE::next_value");
 
   *error= 0;
@@ -554,24 +593,7 @@ longlong SEQUENCE::next_value(TABLE *table, bool second_round, int *error)
     lock();
 
   res_value= next_free_value;
-
-  /* Increment next_free_value */
-  if (real_increment > 0)
-  {
-    if (next_free_value + real_increment > max_value ||
-        next_free_value > max_value - real_increment)
-      next_free_value= max_value + 1;
-    else
-      next_free_value+= real_increment;
-  }
-  else
-  {
-    if (next_free_value + real_increment < min_value ||
-        next_free_value < min_value - real_increment)
-      next_free_value= min_value - 1;
-    else
-      next_free_value+= real_increment;
-  }
+  next_free_value= increment_value(next_free_value);
 
   if ((real_increment > 0 && res_value < reserved_until) ||
       (real_increment < 0 && res_value > reserved_until))
@@ -621,7 +643,7 @@ longlong SEQUENCE::next_value(TABLE *table, bool second_round, int *error)
       goto err;
     round++;
     reserved_until= real_increment >0 ? min_value : max_value;
-    adjust_values();                            // Fix next_free_value
+    adjust_values(reserved_until);              // Fix next_free_value
     /*
       We have to do everything again to ensure that the given range was
       not empty, which could happen if increment == 0
@@ -629,25 +651,11 @@ longlong SEQUENCE::next_value(TABLE *table, bool second_round, int *error)
     DBUG_RETURN(next_value(table, 1, error));
   }
 
-  /* Log a full insert (ok as table is small) */
-  save_rpl_write_set= table->rpl_write_set;
-
-  /* Update table */
-  save_write_set= table->write_set;
-  table->rpl_write_set= table->write_set= &table->s->all_set;
-  store_fields(table);
-  /* Tell ha_sequence::write_row that we already hold the mutex */
-  ((ha_sequence*) table->file)->sequence_locked= 1;
-  if ((*error= table->file->ha_write_row(table->record[0])))
+  if ((*error= write(table)))
   {
-    table->file->print_error(*error, MYF(0));
-    /* Restore original range */
     reserved_until= org_reserved_until;
     next_free_value= res_value;
   }
-  ((ha_sequence*) table->file)->sequence_locked= 0;
-  table->rpl_write_set= save_rpl_write_set;
-  table->write_set= save_write_set;
 
   unlock();
   DBUG_RETURN(res_value);
@@ -681,3 +689,179 @@ void SEQUENCE_LAST_VALUE::set_version(TABLE *table)
 {
   memcpy(table_version, table->s->tabledef_version.str, MY_UUID_SIZE);
 }
+
+/**
+   Set the next value for sequence
+
+   @param in   table       Sequence table
+   @param in   next_val    Next free value
+   @param in   next_round  Round for 'next_value' (in cace of cycles)
+   @param in   is_used     1 if next_val is already used
+
+   @retval     0      ok, value adjusted
+               1      value was less than current value or
+                      error when storing value
+
+   @comment
+   A new value is set only if "nextval,next_round" is less than
+   "next_free_value,round". This is needed as in replication
+   setvalue() calls may come out to the slave out-of-order.
+   Storing only the highest value ensures that sequence object will always
+   contain the highest used value when the slave is promoted to a master.
+*/
+
+bool SEQUENCE::set_value(TABLE *table, longlong next_val, ulonglong next_round,
+                         bool is_used)
+{
+  bool error= 1;
+  bool needs_to_be_stored= 0;
+  longlong org_reserved_until=  reserved_until;
+  longlong org_next_free_value= next_free_value;
+  ulonglong org_round= round;
+  DBUG_ENTER("SEQUENCE::set_value");
+
+  lock();
+  if (is_used)
+    next_val= increment_value(next_val);
+
+  if (round > next_round)
+    goto end;
+  if (round == next_round)
+  {
+    if (real_increment > 0  ?
+        next_val < next_free_value :
+        next_val > next_free_value)
+      goto end;
+    if (next_val == next_free_value)
+    {
+      error= 0;
+      goto end;
+    }
+  }
+  else if (cycle == 0)
+    goto end;                       // round < next_round && no cycles
+  else
+    needs_to_be_stored= 1;
+
+  round= next_round;
+  adjust_values(next_val);
+  if ((real_increment > 0 ?
+       next_free_value > reserved_until :
+       next_free_value < reserved_until) ||
+      needs_to_be_stored)
+  {
+    reserved_until= next_free_value;
+    if (write(table))
+    {
+      reserved_until=  org_reserved_until;
+      next_free_value= org_next_free_value;
+      round= org_round;
+      goto end;
+    }
+  }
+  error= 0;
+
+end:
+  unlock();
+  DBUG_RETURN(error);
+}
+
+
+bool Sql_cmd_alter_sequence::execute(THD *thd)
+{
+  int error= 0;
+  int trapped_errors= 0;
+  LEX *lex= thd->lex;
+  TABLE_LIST *first_table= lex->query_tables;
+  TABLE *table;
+  sequence_definition *new_seq= lex->create_info.seq_create_info;
+  SEQUENCE *seq;
+  No_such_table_error_handler no_such_table_handler;
+  DBUG_ENTER("Sql_cmd_alter_sequence::execute");
+
+  if (check_access(thd, ALTER_ACL, first_table->db,
+                   &first_table->grant.privilege,
+                   &first_table->grant.m_internal,
+                   0, 0))
+    DBUG_RETURN(TRUE);                  /* purecov: inspected */
+
+  if (check_grant(thd, ALTER_ACL, first_table, FALSE, UINT_MAX, FALSE))
+    DBUG_RETURN(TRUE);                  /* purecov: inspected */
+
+  if (lex->check_exists)
+    thd->push_internal_handler(&no_such_table_handler);
+  error= open_and_lock_tables(thd, first_table, FALSE, 0);
+  if (lex->check_exists)
+  {
+    trapped_errors= no_such_table_handler.safely_trapped_errors();
+    thd->pop_internal_handler();
+  }
+  if (error)
+  {
+    if (trapped_errors)
+    {
+      StringBuffer<FN_REFLEN> tbl_name;
+      tbl_name.append(first_table->db);
+      tbl_name.append('.');
+      tbl_name.append(first_table->table_name);
+      push_warning_printf(thd, Sql_condition::WARN_LEVEL_NOTE,
+                          ER_UNKNOWN_SEQUENCES,
+                          ER_THD(thd, ER_UNKNOWN_SEQUENCES),
+                          tbl_name.c_ptr_safe());
+      my_ok(thd);
+      DBUG_RETURN(FALSE);
+    }
+    DBUG_RETURN(TRUE);
+  }
+
+  table= first_table->table;
+  seq= table->s->sequence;
+  new_seq->reserved_until= seq->reserved_until;
+
+  /* Copy from old sequence those fields that the user didn't specified */
+  if (!(new_seq->used_fields & seq_field_used_increment))
+    new_seq->increment= seq->increment;
+  if (!(new_seq->used_fields & seq_field_used_min_value))
+    new_seq->min_value= seq->min_value;
+  if (!(new_seq->used_fields & seq_field_used_max_value))
+    new_seq->max_value= seq->max_value;
+  if (!(new_seq->used_fields & seq_field_used_start))
+    new_seq->start=          seq->start;
+  if (!(new_seq->used_fields & seq_field_used_cache))
+    new_seq->cache= seq->cache;
+  if (!(new_seq->used_fields & seq_field_used_cycle))
+    new_seq->cycle= seq->cycle;
+
+  /* If we should restart from a new value */
+  if (new_seq->used_fields & seq_field_used_restart)
+  {
+    if (!(new_seq->used_fields & seq_field_used_restart_value))
+      new_seq->restart=      new_seq->start;
+    new_seq->reserved_until= new_seq->restart;
+  }
+
+  /* Let check_and_adjust think all fields are used */
+  new_seq->used_fields= ~0;
+  if (new_seq->check_and_adjust())
+  {
+    my_error(ER_SEQUENCE_INVALID_DATA, MYF(0),
+             first_table->db,
+             first_table->table_name);
+    error= 1;
+    goto end;
+  }
+
+  if (!(error= new_seq->write(table)))
+  {
+    /* Store the sequence values in table share */
+    table->s->sequence->copy(new_seq);
+  }
+  trans_commit_stmt(thd);
+  trans_commit_implicit(thd);
+  if (!error)
+    my_ok(thd);
+
+end:
+  close_thread_tables(thd);
+  DBUG_RETURN(error);
+}