MDEV-31058 ER_KEY_NOT_FOUND upon concurrent CHANGE column autoinc and DML

When column is changed to autoinc, ALTER TABLE may update zero/NULL values,
if NO_AUTO_VALUE_ON_ZERO mode is not enabled.

Forbid this for LOCK=NONE for the unreliable cases.
The cases are described in online_alter_check_autoinc.

mysql-test/main/alter_table_online.result

@@ -136,3 +136,71 @@ alter table t1 add c int, algorithm=inplace;
 alter table t1 add d int, lock=none;
 set system_versioning_alter_history= default;
 drop table t1;
+#
+# MDEV-31058 ER_KEY_NOT_FOUND upon concurrent CHANGE column autoinc
+# and DML
+#
+create table t (a serial, b int) engine=innodb;
+alter table t drop a, modify b serial, algorithm=copy, lock=none;
+ERROR 0A000: LOCK=NONE is not supported. Reason: COPY algorithm requires a lock. Try LOCK=SHARED
+set statement sql_mode= NO_AUTO_VALUE_ON_ZERO for
+alter table t drop a, modify b serial, algorithm=copy, lock=none;
+create or replace table t (a serial, b int) engine=innodb;
+show create table t;
+Table	Create Table
+t	CREATE TABLE `t` (
+  `a` bigint(20) unsigned NOT NULL AUTO_INCREMENT,
+  `b` int(11) DEFAULT NULL,
+  UNIQUE KEY `a` (`a`)
+) ENGINE=InnoDB DEFAULT CHARSET=latin1 COLLATE=latin1_swedish_ci
+# a is unique in the old table, but is shrunk in the new one.
+# Only unsafe approach is fine because of possible collisions.
+alter table t modify a int, modify b serial, algorithm=copy, lock=none;
+ERROR 0A000: LOCK=NONE is not supported. Reason: COPY algorithm requires a lock. Try LOCK=SHARED
+#
+# Check that we treat autoinc columns correctly modify old autoinc is
+# fine, adding new autoinc for existed column is unsafe.
+#
+create or replace table t (a serial) engine=innodb;
+alter table t change a b serial, algorithm=copy, lock=none;
+Warnings:
+Note	1831	Duplicate index `b`. This is deprecated and will be disallowed in a future release
+# Shrinking the autoinc field is considered safe.
+# ER_WARN_DATA_OUT_OF_RANGE should be emitted otherwise.
+alter table t change b b int auto_increment primary key,
+algorithm=copy, lock=none;
+alter table t add c int default(0), drop primary key, drop key a;
+# key `b` is still there
+show create table t;
+Table	Create Table
+t	CREATE TABLE `t` (
+  `b` int(11) NOT NULL AUTO_INCREMENT,
+  `c` int(11) DEFAULT 0,
+  UNIQUE KEY `b` (`b`)
+) ENGINE=InnoDB DEFAULT CHARSET=latin1 COLLATE=latin1_swedish_ci
+alter table t drop b, change c c serial, algorithm=copy, lock=none;
+ERROR 0A000: LOCK=NONE is not supported. Reason: COPY algorithm requires a lock. Try LOCK=SHARED
+# Check existed unique keys.
+create or replace table t(a int, b int not null, c int not null, d int);
+# No unique in the old table;
+alter table t add unique(b, c), modify d int auto_increment, add key(d),
+algorithm=copy, lock=none;
+ERROR 0A000: LOCK=NONE is not supported. Reason: COPY algorithm requires a lock. Try LOCK=SHARED
+alter table t add unique(a, b);
+# Unique in the old table has nulls;
+alter table t modify d int auto_increment, add key(d),
+algorithm=copy, lock=none;
+ERROR 0A000: LOCK=NONE is not supported. Reason: COPY algorithm requires a lock. Try LOCK=SHARED
+alter table t add unique(b, c);
+# Change unique'scolumn;
+alter table t change b x int, modify d int auto_increment, add key(d),
+algorithm=copy, lock=none;
+ERROR 0A000: LOCK=NONE is not supported. Reason: COPY algorithm requires a lock. Try LOCK=SHARED
+# Finally good.
+alter table t modify d int auto_increment, add key(d), 
+algorithm=copy, lock=none;
+drop table t;
+# MDEV-31172 Server crash or ASAN errors in online_alter_check_autoinc
+create table t (a int, b int, c char(8), key(a,b,c));
+alter table t modify c int auto_increment key, algorithm=copy;
+drop table t;

mysql-test/main/alter_table_online.test

@@ -135,3 +135,71 @@ alter table t1 add d int, lock=none;
 set system_versioning_alter_history= default;
 drop table t1;
 
+--echo #
+--echo # MDEV-31058 ER_KEY_NOT_FOUND upon concurrent CHANGE column autoinc
+--echo # and DML
+--echo #
+create table t (a serial, b int) engine=innodb;
+--error ER_ALTER_OPERATION_NOT_SUPPORTED_REASON
+alter table t drop a, modify b serial, algorithm=copy, lock=none;
+
+set statement sql_mode= NO_AUTO_VALUE_ON_ZERO for
+alter table t drop a, modify b serial, algorithm=copy, lock=none;
+
+create or replace table t (a serial, b int) engine=innodb;
+show create table t;
+--echo # a is unique in the old table, but is shrunk in the new one.
+--echo # Only unsafe approach is fine because of possible collisions.
+--error ER_ALTER_OPERATION_NOT_SUPPORTED_REASON
+alter table t modify a int, modify b serial, algorithm=copy, lock=none;
+
+--echo #
+--echo # Check that we treat autoinc columns correctly modify old autoinc is
+--echo # fine, adding new autoinc for existed column is unsafe.
+--echo #
+create or replace table t (a serial) engine=innodb;
+
+alter table t change a b serial, algorithm=copy, lock=none;
+
+--echo # Shrinking the autoinc field is considered safe.
+--echo # ER_WARN_DATA_OUT_OF_RANGE should be emitted otherwise.
+alter table t change b b int auto_increment primary key,
+              algorithm=copy, lock=none;
+
+alter table t add c int default(0), drop primary key, drop key a;
+--echo # key `b` is still there
+show create table t;
+
+--error ER_ALTER_OPERATION_NOT_SUPPORTED_REASON
+alter table t drop b, change c c serial, algorithm=copy, lock=none;
+
+--echo # Check existed unique keys.
+create or replace table t(a int, b int not null, c int not null, d int);
+
+--echo # No unique in the old table;
+--error ER_ALTER_OPERATION_NOT_SUPPORTED_REASON
+alter table t add unique(b, c), modify d int auto_increment, add key(d),
+              algorithm=copy, lock=none;
+
+alter table t add unique(a, b);
+--echo # Unique in the old table has nulls;
+--error ER_ALTER_OPERATION_NOT_SUPPORTED_REASON
+alter table t modify d int auto_increment, add key(d),
+              algorithm=copy, lock=none;
+
+alter table t add unique(b, c);
+--echo # Change unique'scolumn;
+--error ER_ALTER_OPERATION_NOT_SUPPORTED_REASON
+alter table t change b x int, modify d int auto_increment, add key(d),
+              algorithm=copy, lock=none;
+
+--echo # Finally good.
+alter table t modify d int auto_increment, add key(d), 
+              algorithm=copy, lock=none;
+
+drop table t;
+
+--echo # MDEV-31172 Server crash or ASAN errors in online_alter_check_autoinc
+create table t (a int, b int, c char(8), key(a,b,c));
+alter table t modify c int auto_increment key, algorithm=copy;
+drop table t;

sql/sql_table.cc

@@ -9847,6 +9847,79 @@ static uint64 get_start_alter_id(THD *thd)
 }
 
 
+static
+bool online_alter_check_autoinc(const THD *thd, const Alter_info *alter_info,
+                                const TABLE *table)
+{
+  /*
+    We can't go online, if all of the following is presented:
+    * Autoinc is added to existing field
+    * Disabled NO_AUTO_VALUE_ON_ZERO
+    * No non-nullable unique key in the old table, that has all the key parts
+      remaining unchanged.
+  */
+
+  // Exit earlier when possible
+  if (thd->variables.sql_mode & MODE_NO_AUTO_VALUE_ON_ZERO)
+    return true;
+  if ((alter_info->flags | ALTER_CHANGE_COLUMN) != alter_info->flags)
+    return true;
+
+  /*
+    Find at least one unique index (without NULLs), all columns of which
+    remain in the table unchanged to presume it's a safe ALTER TABLE.
+  */
+  for (uint k= 0; k < table->s->keys; k++)
+  {
+    const KEY &key= table->s->key_info[k];
+    if ((key.flags & HA_NOSAME) == 0 || key.flags & HA_NULL_PART_KEY)
+      continue;
+    bool key_parts_good= true;
+    for (uint kp= 0; kp < key.user_defined_key_parts && key_parts_good; kp++)
+    {
+      const char *field_name= key.key_part[kp].field->field_name.str;
+      for (const auto &c: alter_info->drop_list)
+        if (c.type == Alter_drop::COLUMN
+            && my_strcasecmp(system_charset_info, c.name, field_name) == 0)
+        {
+          key_parts_good= false;
+          break;
+        }
+      if (key_parts_good)
+        for (const auto &c: alter_info->create_list)
+          if (c.change.str && my_strcasecmp(system_charset_info, c.change.str,
+                                            field_name) == 0)
+          {
+            key_parts_good= false;
+            break;
+          }
+    }
+    if (key_parts_good)
+      return true;
+  }
+
+  const char *old_autoinc= table->found_next_number_field
+                           ? table->found_next_number_field->field_name.str
+                           : "";
+  bool online= true;
+  for (const auto &c: alter_info->create_list)
+  {
+    if (c.change.str && c.flags & AUTO_INCREMENT_FLAG)
+    {
+      if (my_strcasecmp(system_charset_info, c.change.str,  old_autoinc) != 0)
+      {
+        if (c.create_if_not_exists // check IF EXISTS option
+            && table->find_field_by_name(&c.change) == NULL)
+          continue;
+        online= false;
+      }
+      break;
+    }
+  }
+  return online;
+}
+
+
 /**
   Alter table
 
@@ -10083,6 +10156,8 @@ bool mysql_alter_table(THD *thd, const LEX_CSTRING *new_db,
     }
   }
 
+  online= online && online_alter_check_autoinc(thd, alter_info, table);
+  
 #ifdef WITH_WSREP
   if (WSREP(thd) &&
       (thd->lex->sql_command == SQLCOM_ALTER_TABLE ||

sql/table.cc

@@ -10616,7 +10616,7 @@ void Vers_history_point::print(String *str, enum_query_type query_type,
   item->print(str, query_type);
 }
 
-Field *TABLE::find_field_by_name(LEX_CSTRING *str) const
+Field *TABLE::find_field_by_name(const LEX_CSTRING *str) const
 {
   Field **tmp;
   size_t length= str->length;

sql/table.h

@@ -1817,7 +1817,7 @@ public:
                                       bool with_cleanup);
   bool vcol_fix_expr(THD *thd);
   bool vcol_cleanup_expr(THD *thd);
-  Field *find_field_by_name(LEX_CSTRING *str) const;
+  Field *find_field_by_name(const LEX_CSTRING *str) const;
   bool export_structure(THD *thd, class Row_definition_list *defs);
   bool is_splittable() { return spl_opt_info != NULL; }
   void set_spl_opt_info(SplM_opt_info *spl_info);