Make SEQUENCE working with replication

- Old sequence code forced row based replication for any statements that
  refered to a sequence table. What is new is that row based replication
  is now sequence aware:
   - NEXT VALUE is now generating a short row based event with only
     next_value and round being replicated.
   - Short row based events are now on the slave updated as trough
     SET_VALUE(sequence_name)
   - Full row based events are on the slave updated with a full insert,
     which is practically same as ALTER SEQUENCE.
- INSERT on a SEQUENCE table does now a EXCLUSIVE LOCK to ensure that
  it is logged in binary log before any following NEXT VALUE calls.
- Enable all sequence tests and fixed found bugs
- ALTER SEQUENCE doesn't anymore allow changes that makes the next_value
  outside of allowed range
- SEQUENCE changes are done with TL_WRITE_ALLOW_WRITE. Because of this
  one can generate a statement for MyISAM with both
  TL_WRITE_CONCURRENT_INSERT and TL_WRITE_ALLOW_WRITE. To fix a warning
  I had to add an extra test in thr_lock.c for this.
- Removed UPDATE of SEQUENCE (no need to support this as we
  have ALTER SEQUENCE, which takes the EXCLUSIVE lock properly.
- Removed DBUG_ASSERT() in MDL_context::upgrade_shared_lock. This was
  removed upstream in MySQL 5.6 in 72f823de453.
- Simplified test in decided_logging_format() by using sql_command_flags()
- Fix that we log DROP SEQUENCE correctly.
- Fixed that Aria works with SEQUENCE

1 files changed, 28 insertions, 42 deletions

diff --git a/sql/ha_sequence.cc b/sql/ha_sequence.cc
index a918da92be2..279e7801b2d 100644
--- a/sql/ha_sequence.cc
+++ b/sql/ha_sequence.cc
@@ -187,6 +187,7 @@ int ha_sequence::create(const char *name, TABLE *form,
 int ha_sequence::write_row(uchar *buf)
 {
   int error;
+  sequence_definition tmp_seq;
   DBUG_ENTER("ha_sequence::write_row");
   DBUG_ASSERT(table->record[0] == buf);
 
@@ -199,16 +200,31 @@ int ha_sequence::write_row(uchar *buf)
   if (unlikely(sequence->initialized != SEQUENCE::SEQ_READY_TO_USE))
     DBUG_RETURN(HA_ERR_WRONG_COMMAND);
 
-  /*
-    User tries to write a row
-    - Check that the new row is an accurate object
-    - Update the first row in the table
-  */
+  if (!sequence_locked)                         // If not from next_value()
+  {
+    /*
+      User tries to write a full row directly to the sequence table with
+      INSERT or LOAD DATA.
+
+      - Get an exclusive lock for the table. This is needed to ensure that
+        we excute all full inserts (same as ALTER SEQUENCE) in same order
+        on master and slaves
+      - Check that we are only using one table.
+        This is to avoid deadlock problems when upgrading lock to exlusive.
+      - Check that the new row is an accurate SEQUENCE object
+    */
 
-  sequence_definition tmp_seq;
-  tmp_seq.read_fields(table);
-  if (tmp_seq.check_and_adjust())
-    DBUG_RETURN(HA_ERR_SEQUENCE_INVALID_DATA);
+    THD *thd= table->in_use;
+    if (thd->lock->table_count != 1)
+      DBUG_RETURN(ER_WRONG_INSERT_INTO_SEQUENCE);
+    if (thd->mdl_context.upgrade_shared_lock(table->mdl_ticket, MDL_EXCLUSIVE,
+                                             thd->variables.lock_wait_timeout))
+      DBUG_RETURN(ER_LOCK_WAIT_TIMEOUT);
+
+    tmp_seq.read_fields(table);
+    if (tmp_seq.check_and_adjust(0))
+      DBUG_RETURN(HA_ERR_SEQUENCE_INVALID_DATA);
+  }
 
   /*
     Lock sequence to ensure that no one can come in between
@@ -235,39 +251,6 @@ int ha_sequence::write_row(uchar *buf)
 }
 
 
-int ha_sequence::update_row(const uchar *old_data, const uchar *new_data)
-{
-  int error;
-  sequence_definition tmp_seq;
-  DBUG_ENTER("ha_sequence::update_row");
-  DBUG_ASSERT(new_data == table->record[0]);
-
-  row_already_logged= 0;
-
-  tmp_seq.read_fields(table);
-  if (tmp_seq.check_and_adjust())
-    DBUG_RETURN(HA_ERR_SEQUENCE_INVALID_DATA);
-
-  /*
-    Lock sequence to ensure that no one can come in between
-    while sequence, table and binary log is updated.
-  */
-  sequence->lock();
-  if (!(error= file->update_row(old_data, new_data)))
-  {
-    sequence->copy(&tmp_seq);
-    rows_changed++;
-    /* We have to do the logging while we hold the sequence mutex */
-    error= binlog_log_row(table, old_data, new_data,
-                          Update_rows_log_event::binlog_row_logging_function);
-    row_already_logged= 1;
-  }
-  sequence->all_values_used= 0;
-  sequence->unlock();
-  DBUG_RETURN(error);
-}
-
-
 /*
   Inherit the sequence base table flags.
 */
@@ -346,6 +329,9 @@ void ha_sequence::print_error(int error, myf errflag)
   case HA_ERR_WRONG_COMMAND:
     my_error(ER_ILLEGAL_HA, MYF(0), "SEQUENCE", sequence_db, sequence_name);
     DBUG_VOID_RETURN;
+  case ER_WRONG_INSERT_INTO_SEQUENCE:
+    my_error(error, MYF(0));
+    DBUG_VOID_RETURN;
   }
   file->print_error(error, errflag);
   DBUG_VOID_RETURN;