Fix that we can read tables with the 'older' decimal format used in 5.0.3 & 5.0.4

We will however give a warning when opening such a table that users should use ALTER TABLE ... FORCE to fix
the table. In future release we will fix that REPAIR TABLE will be able to handle this case


sql/sql_lex.h:
  Support for ALTER TABLE ... FORCE
sql/sql_table.cc:
  CHECK TABLE now gives a note if table->s->crashed was set
sql/sql_yacc.yy:
  Support for ALTER TABLE ... FORCE
sql/table.cc:
  
  Fix that we can read tables with the 'older' decimal format used in 5.0.3 & 5.0.4
  (Now we store display length in the .frm table while we previously stored precision)
sql/table.h:
  Store in TABLE_SHARE version number of MySQL where table was created (or checked)

sql/sql_lex.h

@@ -635,6 +635,7 @@ typedef class st_select_lex SELECT_LEX;
 #define ALTER_CHANGE_COLUMN_DEFAULT 256
 #define ALTER_KEYS_ONOFF        512
 #define ALTER_CONVERT          1024
+#define ALTER_FORCE		2048
 
 typedef struct st_alter_info
 {

sql/sql_table.cc

@@ -2202,12 +2202,14 @@ static bool mysql_admin_table(THD* thd, TABLE_LIST* tables,
     if ((table->table->db_stat & HA_READ_ONLY) && open_for_modify)
     {
       char buff[FN_REFLEN + MYSQL_ERRMSG_SIZE];
+      uint length;
       protocol->prepare_for_resend();
       protocol->store(table_name, system_charset_info);
       protocol->store(operator_name, system_charset_info);
       protocol->store("error", 5, system_charset_info);
-      my_snprintf(buff, sizeof(buff), ER(ER_OPEN_AS_READONLY), table_name);
+      length= my_snprintf(buff, sizeof(buff), ER(ER_OPEN_AS_READONLY),
-      protocol->store(buff, system_charset_info);
+                          table_name);
+      protocol->store(buff, length, system_charset_info);
       close_thread_tables(thd);
       table->table=0;				// For query cache
       if (protocol->write())
@@ -2238,6 +2240,17 @@ static bool mysql_admin_table(THD* thd, TABLE_LIST* tables,
       open_for_modify= 0;
     }
 
+    if (table->table->s->crashed && operator_func == &handler::check)
+    {
+      protocol->prepare_for_resend();
+      protocol->store(table_name, system_charset_info);
+      protocol->store(operator_name, system_charset_info);
+      protocol->store("warning", 7, system_charset_info);
+      protocol->store("Table is marked as crashed", 26, system_charset_info);
+      if (protocol->write())
+        goto err;
+    }
+
     result_code = (table->table->file->*operator_func)(thd, check_opt);
 
 send_result:

sql/sql_yacc.yy

@@ -3512,6 +3512,10 @@ alter_list_item:
 	    LEX *lex=Lex;
 	    lex->alter_info.flags|= ALTER_OPTIONS;
 	  }
+	| FORCE_SYM
+	  {
+	    Lex->alter_info.flags|= ALTER_FORCE;
+	   }
 	| order_clause
 	  {
 	    LEX *lex=Lex;

sql/table.cc

@@ -166,6 +166,7 @@ int openfrm(THD *thd, const char *name, const char *alias, uint db_stat,
   share->db_type= ha_checktype((enum db_type) (uint) *(head+3));
   share->db_create_options= db_create_options=uint2korr(head+30);
   share->db_options_in_use= share->db_create_options;
+  share->mysql_version= uint4korr(head+51);
   null_field_first= 0;
   if (!head[32])				// New frm file in 3.23
   {
@@ -572,6 +573,29 @@ int openfrm(THD *thd, const char *name, const char *alias, uint db_stat,
       error= 4;
       goto err;			/* purecov: inspected */
     }
+#ifndef TO_BE_DELETED_ON_PRODUCTION
+    if (field_type == FIELD_TYPE_NEWDECIMAL && !share->mysql_version)
+    {
+      /*
+        Fix pack length of old decimal values from 5.0.3 -> 5.0.4
+        The difference is that in the old version we stored precision
+        in the .frm table while we now store the display_length
+      */
+      Field_new_decimal *dec_field= (Field_new_decimal*) reg_field;
+      dec_field->bin_size= my_decimal_get_binary_size(field_length,
+                                                      dec_field->dec);
+      dec_field->precision= field_length;
+      dec_field->field_length=
+        my_decimal_precision_to_length(field_length, dec_field->dec,
+                                       dec_field->unsigned_flag);
+      sql_print_error("Found incompatible DECIMAL field '%s' in %s; Please do \"ALTER TABLE '%s' FORCE\" to fix it!", dec_field->field_name, name, share->table_name);
+      push_warning_printf(thd, MYSQL_ERROR::WARN_LEVEL_ERROR,
+                          ER_CRASHED_ON_USAGE,
+                          "Found incompatible DECIMAL field '%s' in %s; Please do \"ALTER TABLE '%s' FORCE\" to fix it!", dec_field->field_name, name, share->table_name);
+      share->crashed= 1;                        // Marker for CHECK TABLE
+    }
+#endif
+
     reg_field->comment=comment;
     if (field_type == FIELD_TYPE_BIT && !f_bit_as_char(pack_flag))
     {
@@ -712,6 +736,28 @@ int openfrm(THD *thd, const char *name, const char *alias, uint db_stat,
 	  }
 	  if (field->key_length() != key_part->length)
 	  {
+#ifndef TO_BE_DELETED_ON_PRODUCTION
+            if (field->type() == FIELD_TYPE_NEWDECIMAL)
+            {
+              /*
+                Fix a fatal error in decimal key handling that causes crashes
+                on Innodb. We fix it by reducing the key length so that
+                InnoDB never gets a too big key when searching.
+                This allows the end user to do an ALTER TABLE to fix the
+                error.
+              */
+	      keyinfo->key_length-= (key_part->length - field->key_length());
+	      key_part->store_length-= (key_part->length - field->key_length());
+              key_part->length= field->key_length();
+              sql_print_error("Found wrong key definition in %s; Please do \"ALTER TABLE '%s' FORCE \" to fix it!", name, share->table_name);
+              push_warning_printf(thd, MYSQL_ERROR::WARN_LEVEL_ERROR,
+                                  ER_CRASHED_ON_USAGE,
+                                  "Found wrong key definition in %s; Please do \"ALTER TABLE '%s' FORCE\" to fix it!", name, share->table_name);
+
+              share->crashed= 1;                // Marker for CHECK TABLE
+	      goto to_be_deleted;
+            }
+#endif
 	    key_part->key_part_flag|= HA_PART_KEY_SEG;
 	    if (!(field->flags & BLOB_FLAG))
 	    {					// Create a new field
@@ -720,6 +766,9 @@ int openfrm(THD *thd, const char *name, const char *alias, uint db_stat,
 	      field->field_length=key_part->length;
 	    }
 	  }
+
+	to_be_deleted:
+
 	  /*
 	    If the field can be NULL, don't optimize away the test
 	    key_part_column = expression from the WHERE clause
@@ -1298,7 +1347,6 @@ File create_frm(register my_string name, uint reclength, uchar *fileinfo,
 		HA_CREATE_INFO *create_info, uint keys)
 {
   register File file;
-  uint key_length;
   ulong length;
   char fill[IO_SIZE];
   int create_flags= O_RDWR | O_TRUNC;
@@ -1321,6 +1369,8 @@ File create_frm(register my_string name, uint reclength, uchar *fileinfo,
 
   if ((file= my_create(name, CREATE_MODE, create_flags, MYF(MY_WME))) >= 0)
   {
+    uint key_length, tmp_key_length;
+    uint tmp;
     bzero((char*) fileinfo,64);
     /* header */
     fileinfo[0]=(uchar) 254;
@@ -1333,8 +1383,8 @@ File create_frm(register my_string name, uint reclength, uchar *fileinfo,
     key_length=keys*(7+NAME_LEN+MAX_REF_PARTS*9)+16;
     length=(ulong) next_io_size((ulong) (IO_SIZE+key_length+reclength));
     int4store(fileinfo+10,length);
-    if (key_length > 0xffff) key_length=0xffff;
+    tmp_key_length= (key_length < 0xffff) ? key_length : 0xffff;
-    int2store(fileinfo+14,key_length);
+    int2store(fileinfo+14,tmp_key_length);
     int2store(fileinfo+16,reclength);
     int4store(fileinfo+18,create_info->max_rows);
     int4store(fileinfo+22,create_info->min_rows);
@@ -1350,6 +1400,9 @@ File create_frm(register my_string name, uint reclength, uchar *fileinfo,
     fileinfo[41]= (uchar) create_info->raid_type;
     fileinfo[42]= (uchar) create_info->raid_chunks;
     int4store(fileinfo+43,create_info->raid_chunksize);
+    int4store(fileinfo+47, key_length);
+    tmp= MYSQL_VERSION_ID;          // Store to avoid warning from int4store
+    int4store(fileinfo+51, tmp);
     bzero(fill,IO_SIZE);
     for (; length > IO_SIZE ; length-= IO_SIZE)
     {

sql/table.h

@@ -126,7 +126,7 @@ typedef struct st_table_share
   key_map keys_for_keyread;
   ulong   avg_row_length;		/* create information */
   ulong   raid_chunksize;
-  ulong   version, flush_version;
+  ulong   version, flush_version, mysql_version;
   ulong   timestamp_offset;		/* Set to offset+1 of record */
   ulong   reclength;			/* Recordlength */