MDEV-15408 Confusing error message upon ER_VERS_FIELD_WRONG_TYPE while omitting UNSIGNED in BIGINT

Improve diagnostics. Try to guess what type user tried to type.

mysql-test/suite/versioning/r/create.result

@@ -157,7 +157,7 @@ Sys_start bigint as row start invisible,
 Sys_end bigint unsigned as row end invisible,
 period for system_time (Sys_start, Sys_end)
 ) with system versioning engine innodb;
-ERROR HY000: `Sys_start` must be of type TIMESTAMP(6) for system-versioned table `t1`
+ERROR HY000: `Sys_start` must be of type BIGINT(20) UNSIGNED for system-versioned table `t1`
 create or replace table t1 (
 x14 int unsigned,
 Sys_start bigint unsigned as row start invisible,
@@ -495,5 +495,19 @@ Table	Create Table
 t1	CREATE TABLE `t1` (
   `i` int(11) DEFAULT NULL
 ) ENGINE=DEFAULT_ENGINE DEFAULT CHARSET=latin1 WITH SYSTEM VERSIONING
+create or replace table t1 (
+a int,
+row_start bigint as row start,
+row_end bigint as row end,
+period for system_time (row_start, row_end)
+) engine=innodb with system versioning;
+ERROR HY000: `row_start` must be of type BIGINT(20) UNSIGNED for system-versioned table `t1`
+create or replace table t1 (
+a int,
+row_start bigint as row start,
+row_end bigint as row end,
+period for system_time (row_start, row_end)
+) engine=myisam with system versioning;
+ERROR HY000: `row_start` must be of type TIMESTAMP(6) for system-versioned table `t1`
 drop database test;
 create database test;

mysql-test/suite/versioning/t/create.test

@@ -371,5 +371,21 @@ create or replace table t1 (i int) with system versioning as select 1 as i;
 --replace_result $default_engine DEFAULT_ENGINE
 show create table t1;
 
+--error ER_VERS_FIELD_WRONG_TYPE
+create or replace table t1 (
+  a int,
+  row_start bigint as row start,
+  row_end bigint as row end,
+  period for system_time (row_start, row_end)
+) engine=innodb with system versioning;
+
+--error ER_VERS_FIELD_WRONG_TYPE
+create or replace table t1 (
+  a int,
+  row_start bigint as row start,
+  row_end bigint as row end,
+  period for system_time (row_start, row_end)
+) engine=myisam with system versioning;
+
 drop database test;
 create database test;

sql/handler.cc

@@ -7211,8 +7211,9 @@ bool Table_scope_and_contents_source_st::vers_check_system_fields(
 {
   if (!(options & HA_VERSIONED_TABLE))
     return false;
-  return vers_info.check_sys_fields(create_table.table_name, create_table.db,
-                                    alter_info);
+  return vers_info.check_sys_fields(
+      create_table.table_name, create_table.db, alter_info,
+      ha_check_storage_engine_flag(db_type, HTON_NATIVE_SYS_VERSIONING));
 }
 
 
@@ -7321,7 +7322,11 @@ bool Vers_parse_info::fix_alter_info(THD *thd, Alter_info *alter_info,
 
   if (alter_info->flags & ALTER_ADD_SYSTEM_VERSIONING)
   {
-    if (check_sys_fields(table_name, share->db, alter_info))
+    const bool can_native=
+        ha_check_storage_engine_flag(create_info->db_type,
+                                     HTON_NATIVE_SYS_VERSIONING) ||
+        create_info->db_type->db_type == DB_TYPE_PARTITION_DB;
+    if (check_sys_fields(table_name, share->db, alter_info, can_native))
       return true;
   }
 
@@ -7426,78 +7431,83 @@ bool Vers_parse_info::check_conditions(const Lex_table_name &table_name,
   return false;
 }
 
+static bool is_versioning_timestamp(const Create_field *f)
+{
+  return (f->type_handler() == &type_handler_datetime2 ||
+          f->type_handler() == &type_handler_timestamp2) &&
+         f->length == MAX_DATETIME_FULL_WIDTH;
+}
+
+static bool is_some_bigint(const Create_field *f)
+{
+  return f->type_handler() == &type_handler_longlong ||
+         f->type_handler() == &type_handler_vers_trx_id;
+}
+
+static bool is_versioning_bigint(const Create_field *f)
+{
+  return is_some_bigint(f) && f->flags & UNSIGNED_FLAG &&
+         f->length == MY_INT64_NUM_DECIMAL_DIGITS - 1;
+}
+
+static bool require_timestamp(const Create_field *f, Lex_table_name table_name)
+{
+  my_error(ER_VERS_FIELD_WRONG_TYPE, MYF(0), f->field_name.str, "TIMESTAMP(6)",
+           table_name.str);
+  return true;
+}
+static bool require_bigint(const Create_field *f, Lex_table_name table_name)
+{
+  my_error(ER_VERS_FIELD_WRONG_TYPE, MYF(0), f->field_name.str,
+           "BIGINT(20) UNSIGNED", table_name.str);
+  return true;
+}
+
 bool Vers_parse_info::check_sys_fields(const Lex_table_name &table_name,
                                        const Lex_table_name &db,
-                                       Alter_info *alter_info)
+                                       Alter_info *alter_info, bool can_native)
 {
   if (check_conditions(table_name, db))
     return true;
 
+  const Create_field *row_start= NULL;
+  const Create_field *row_end= NULL;
+
   List_iterator<Create_field> it(alter_info->create_list);
-  uint found_flag= 0;
   while (Create_field *f= it++)
   {
-    vers_sys_type_t f_check_unit= VERS_UNDEFINED;
-    uint sys_flag= f->flags & VERS_SYSTEM_FIELD;
+    if (!row_start && f->flags & VERS_SYS_START_FLAG)
+      row_start= f;
+    else if (!row_end && f->flags & VERS_SYS_END_FLAG)
+      row_end= f;
+  }
 
-    if (!sys_flag)
-      continue;
+  const bool expect_timestamp=
+      !can_native || !is_some_bigint(row_start) || !is_some_bigint(row_end);
 
-    if (sys_flag & found_flag)
-    {
-      my_error(ER_VERS_DUPLICATE_ROW_START_END, MYF(0),
-                found_flag & VERS_SYS_START_FLAG ? "START" : "END",
-               f->field_name.str);
-      return true;
-    }
+  if (expect_timestamp)
+  {
+    if (!is_versioning_timestamp(row_start))
+      return require_timestamp(row_start, table_name);
 
-    sys_flag|= found_flag;
+    if (!is_versioning_timestamp(row_end))
+      return require_timestamp(row_end, table_name);
+  }
+  else
+  {
+    if (!is_versioning_bigint(row_start))
+      return require_bigint(row_start, table_name);
 
-    if ((f->type_handler() == &type_handler_datetime2 ||
-          f->type_handler() == &type_handler_timestamp2) &&
-        f->length == MAX_DATETIME_FULL_WIDTH)
-    {
-      f_check_unit= VERS_TIMESTAMP;
-    }
-    else if (f->type_handler() == &type_handler_longlong
-      && (f->flags & UNSIGNED_FLAG)
-      && f->length == (MY_INT64_NUM_DECIMAL_DIGITS - 1))
-    {
-      f_check_unit= VERS_TRX_ID;
-    }
-    else
-    {
-      if (!check_unit)
-        check_unit= VERS_TIMESTAMP;
-      goto error;
-    }
+    if (!is_versioning_bigint(row_end))
+      return require_bigint(row_end, table_name);
+  }
 
-    if (f_check_unit)
-    {
-      if (check_unit)
-      {
-        if (check_unit == f_check_unit)
-        {
-          if (check_unit == VERS_TRX_ID && !TR_table::use_transaction_registry)
-          {
-            my_error(ER_VERS_TRT_IS_DISABLED, MYF(0));
-            return true;
-          }
-          return false;
-        }
-      error:
-        my_error(ER_VERS_FIELD_WRONG_TYPE, MYF(0), f->field_name.str,
-                 check_unit == VERS_TIMESTAMP ?
-                 "TIMESTAMP(6)" :
-                 "BIGINT(20) UNSIGNED",
-                 table_name.str);
-        return true;
-      }
-      check_unit= f_check_unit;
-    }
+  if (is_versioning_bigint(row_start) && is_versioning_bigint(row_end) &&
+      !TR_table::use_transaction_registry)
+  {
+    my_error(ER_VERS_TRT_IS_DISABLED, MYF(0));
+    return true;
   }
 
-  my_error(ER_MISSING, MYF(0), table_name.str, found_flag & VERS_SYS_START_FLAG ?
-           "ROW END" : found_flag ? "ROW START" : "ROW START/END");
-  return true;
+  return false;
 }

sql/handler.h

@@ -1924,7 +1924,6 @@ extern const LEX_CSTRING null_clex_str;
 struct Vers_parse_info
 {
   Vers_parse_info() :
-    check_unit(VERS_UNDEFINED),
     versioned_fields(false),
     unversioned_fields(false)
   {}
@@ -1933,7 +1932,6 @@ struct Vers_parse_info
   {
     system_time= start_end_t(null_clex_str, null_clex_str);
     as_row= start_end_t(null_clex_str, null_clex_str);
-    check_unit= VERS_UNDEFINED;
     versioned_fields= false;
     unversioned_fields= false;
   }
@@ -1951,7 +1949,6 @@ struct Vers_parse_info
 
   start_end_t system_time;
   start_end_t as_row;
-  vers_sys_type_t check_unit;
 
   void set_system_time(Lex_ident start, Lex_ident end)
   {
@@ -1993,7 +1990,7 @@ struct Vers_parse_info
                        TABLE_LIST &src_table, TABLE_LIST &table);
   bool check_sys_fields(const Lex_table_name &table_name,
                         const Lex_table_name &db,
-                        Alter_info *alter_info);
+                        Alter_info *alter_info, bool can_native);
 
   /**
      At least one field was specified 'WITH/WITHOUT SYSTEM VERSIONING'.