Merge kboortz@bk-internal.mysql.com:/home/bk/mysql-4.1

into  mysql.com:/home/kent/bk/tmp/mysql-4.1-build

myisam/mi_create.c

@@ -158,6 +158,10 @@ int mi_create(const char *name,uint keys,MI_KEYDEF *keydefs,
       rec--;
       if (rec->type == (int) FIELD_SKIP_ZERO && rec->length == 1)
       {
+        /*
+          NOTE1: here we change a field type FIELD_SKIP_ZERO ->
+          FIELD_NORMAL
+        */
 	rec->type=(int) FIELD_NORMAL;
 	packed--;
 	min_pack_length++;

myisam/rt_key.c

@@ -35,6 +35,7 @@ int rtree_add_key(MI_INFO *info, MI_KEYDEF *keyinfo, uchar *key,
 {
   uint page_size = mi_getint(page_buf);
   uint nod_flag = mi_test_if_nod(page_buf);
+  DBUG_ENTER("rtree_add_key");
 
   if (page_size + key_length + info->s->base.rec_reflength <=
       keyinfo->block_length)
@@ -43,22 +44,26 @@ int rtree_add_key(MI_INFO *info, MI_KEYDEF *keyinfo, uchar *key,
     if (nod_flag)
     {
       /* save key */
+      DBUG_ASSERT(_mi_kpos(nod_flag, key) < info->state->key_file_length);
       memcpy(rt_PAGE_END(page_buf), key - nod_flag, key_length + nod_flag); 
       page_size += key_length + nod_flag;
     }
     else
     {
       /* save key */
+      DBUG_ASSERT(_mi_dpos(info, nod_flag, key + key_length +
+                           info->s->base.rec_reflength) <
+                  info->state->data_file_length + info->s->base.pack_reclength);
       memcpy(rt_PAGE_END(page_buf), key, key_length + 
                                          info->s->base.rec_reflength);
       page_size += key_length + info->s->base.rec_reflength;
     }
     mi_putint(page_buf, page_size, nod_flag);
-    return 0;
+    DBUG_RETURN(0);
   }
 
-  return (rtree_split_page(info, keyinfo, page_buf, key, key_length,
-			   new_page) ? -1 : 1);
+  DBUG_RETURN((rtree_split_page(info, keyinfo, page_buf, key, key_length,
+                                new_page) ? -1 : 1));
 }
 
 /*
@@ -90,11 +95,13 @@ int rtree_delete_key(MI_INFO *info, uchar *page_buf, uchar *key,
 int rtree_set_key_mbr(MI_INFO *info, MI_KEYDEF *keyinfo, uchar *key, 
 		      uint key_length, my_off_t child_page)
 {
+  DBUG_ENTER("rtree_set_key_mbr");
+
   if (!_mi_fetch_keypage(info, keyinfo, child_page,
                          DFLT_INIT_HITS, info->buff, 0))
-    return -1;
+    DBUG_RETURN(-1); /* purecov: inspected */
 
-  return rtree_page_mbr(info, keyinfo->seg, info->buff, key, key_length);
+  DBUG_RETURN(rtree_page_mbr(info, keyinfo->seg, info->buff, key, key_length));
 }
 
 #endif /*HAVE_RTREE_KEYS*/

myisam/rt_split.c

@@ -268,13 +268,15 @@ int rtree_split_page(MI_INFO *info, MI_KEYDEF *keyinfo, uchar *page, uchar *key,
                                    info->s->base.rec_reflength);
 
   int max_keys = (mi_getint(page)-2) / (full_length);
+  DBUG_ENTER("rtree_split_page");
+  DBUG_PRINT("rtree", ("splitting block"));
 
   n_dim = keyinfo->keysegs / 2;
   
   if (!(coord_buf= (double*) my_alloca(n_dim * 2 * sizeof(double) *
 				       (max_keys + 1 + 4) +
 				       sizeof(SplitStruct) * (max_keys + 1))))
-    return -1;
+    DBUG_RETURN(-1); /* purecov: inspected */
 
   task= (SplitStruct *)(coord_buf + n_dim * 2 * (max_keys + 1 + 4));
 
@@ -346,12 +348,13 @@ int rtree_split_page(MI_INFO *info, MI_KEYDEF *keyinfo, uchar *page, uchar *key,
   else
     err_code= _mi_write_keypage(info, keyinfo, *new_page_offs,
                                 DFLT_INIT_HITS, new_page);
+  DBUG_PRINT("rtree", ("split new block: %lu", (ulong) *new_page_offs));
 
   my_afree((byte*)new_page);
 
 split_err:
   my_afree((byte*) coord_buf);
-  return err_code;
+  DBUG_RETURN(err_code);
 }
 
 #endif /*HAVE_RTREE_KEYS*/

mysql-test/r/merge.result

@@ -801,3 +801,21 @@ CREATE TABLE tm1(a SMALLINT, b SMALLINT, KEY(a)) ENGINE=MERGE UNION=(t1);
 SELECT * FROM tm1;
 ERROR HY000: Unable to open underlying table which is differently defined or of non-MyISAM type or doesn't exist
 DROP TABLE t1, tm1;
+CREATE TABLE t1(c1 INT) ENGINE=MyISAM;
+CREATE TABLE t2(c1 INT) ENGINE=MERGE UNION=(t1);
+INSERT DELAYED INTO t2 VALUES(1);
+ERROR HY000: Table storage engine for 't2' doesn't have this option
+DROP TABLE t1, t2;
+CREATE TABLE t1(c1 VARCHAR(1));
+CREATE TABLE m1 LIKE t1;
+ALTER TABLE m1 ENGINE=MERGE UNION=(t1);
+SELECT * FROM m1;
+c1
+DROP TABLE t1, m1;
+CREATE TABLE t1(c1 VARCHAR(4), c2 TINYINT, c3 TINYINT, c4 TINYINT,
+c5 TINYINT, c6 TINYINT, c7 TINYINT, c8 TINYINT, c9 TINYINT);
+CREATE TABLE m1 LIKE t1;
+ALTER TABLE m1 ENGINE=MERGE UNION=(t1);
+SELECT * FROM m1;
+c1	c2	c3	c4	c5	c6	c7	c8	c9
+DROP TABLE t1, m1;

mysql-test/t/merge.test

@@ -428,4 +428,30 @@ CREATE TABLE tm1(a SMALLINT, b SMALLINT, KEY(a)) ENGINE=MERGE UNION=(t1);
 SELECT * FROM tm1;
 DROP TABLE t1, tm1;
 
+#
+# Bug#26464 - insert delayed + update + merge = corruption
+#
+CREATE TABLE t1(c1 INT) ENGINE=MyISAM;
+CREATE TABLE t2(c1 INT) ENGINE=MERGE UNION=(t1);
+--error 1031
+INSERT DELAYED INTO t2 VALUES(1);
+DROP TABLE t1, t2;
+
+#
+# BUG#26881 - Large MERGE tables report incorrect specification when no
+#             differences in tables
+#
+CREATE TABLE t1(c1 VARCHAR(1));
+CREATE TABLE m1 LIKE t1;
+ALTER TABLE m1 ENGINE=MERGE UNION=(t1);
+SELECT * FROM m1;
+DROP TABLE t1, m1;
+
+CREATE TABLE t1(c1 VARCHAR(4), c2 TINYINT, c3 TINYINT, c4 TINYINT,
+                c5 TINYINT, c6 TINYINT, c7 TINYINT, c8 TINYINT, c9 TINYINT);
+CREATE TABLE m1 LIKE t1;
+ALTER TABLE m1 ENGINE=MERGE UNION=(t1);
+SELECT * FROM m1;
+DROP TABLE t1, m1;
+
 # End of 4.1 tests

sql/ha_myisam.cc

@@ -304,6 +304,12 @@ int table2myisam(TABLE *table_arg, MI_KEYDEF **keydef_out,
   RETURN VALUE
     0 - Equal definitions.
     1 - Different definitions.
+
+  TODO
+    - compare FULLTEXT keys;
+    - compare SPATIAL keys;
+    - compare FIELD_SKIP_ZERO which is converted to FIELD_NORMAL correctly
+      (should be corretly detected in table2myisam).
 */
 
 int check_definition(MI_KEYDEF *t1_keyinfo, MI_COLUMNDEF *t1_recinfo,
@@ -329,6 +335,28 @@ int check_definition(MI_KEYDEF *t1_keyinfo, MI_COLUMNDEF *t1_recinfo,
   {
     HA_KEYSEG *t1_keysegs= t1_keyinfo[i].seg;
     HA_KEYSEG *t2_keysegs= t2_keyinfo[i].seg;
+    if (t1_keyinfo[i].flag & HA_FULLTEXT && t2_keyinfo[i].flag & HA_FULLTEXT)
+      continue;
+    else if (t1_keyinfo[i].flag & HA_FULLTEXT ||
+             t2_keyinfo[i].flag & HA_FULLTEXT)
+    {
+       DBUG_PRINT("error", ("Key %d has different definition", i));
+       DBUG_PRINT("error", ("t1_fulltext= %d, t2_fulltext=%d",
+                            test(t1_keyinfo[i].flag & HA_FULLTEXT),
+                            test(t2_keyinfo[i].flag & HA_FULLTEXT)));
+       DBUG_RETURN(1);
+    }
+    if (t1_keyinfo[i].flag & HA_SPATIAL && t2_keyinfo[i].flag & HA_SPATIAL)
+      continue;
+    else if (t1_keyinfo[i].flag & HA_SPATIAL ||
+             t2_keyinfo[i].flag & HA_SPATIAL)
+    {
+       DBUG_PRINT("error", ("Key %d has different definition", i));
+       DBUG_PRINT("error", ("t1_spatial= %d, t2_spatial=%d",
+                            test(t1_keyinfo[i].flag & HA_SPATIAL),
+                            test(t2_keyinfo[i].flag & HA_SPATIAL)));
+       DBUG_RETURN(1);
+    }
     if (t1_keyinfo[i].keysegs != t2_keyinfo[i].keysegs ||
         t1_keyinfo[i].key_alg != t2_keyinfo[i].key_alg)
     {
@@ -365,7 +393,14 @@ int check_definition(MI_KEYDEF *t1_keyinfo, MI_COLUMNDEF *t1_recinfo,
   {
     MI_COLUMNDEF *t1_rec= &t1_recinfo[i];
     MI_COLUMNDEF *t2_rec= &t2_recinfo[i];
-    if (t1_rec->type != t2_rec->type ||
+    /*
+      FIELD_SKIP_ZERO can be changed to FIELD_NORMAL in mi_create,
+      see NOTE1 in mi_create.c
+    */
+    if ((t1_rec->type != t2_rec->type &&
+         !(t1_rec->type == (int) FIELD_SKIP_ZERO &&
+           t1_rec->length == 1 &&
+           t2_rec->type == (int) FIELD_NORMAL)) ||
         t1_rec->length != t2_rec->length ||
         t1_rec->null_bit != t2_rec->null_bit)
     {

sql/ha_myisammrg.h

@@ -37,7 +37,7 @@ class ha_myisammrg: public handler
   {
     return (HA_REC_NOT_IN_SEQ | HA_AUTO_PART_KEY | HA_READ_RND_SAME |
 	    HA_NULL_IN_KEY | HA_CAN_INDEX_BLOBS | HA_FILE_BASED |
-            HA_CAN_INSERT_DELAYED | HA_ANY_INDEX_MAY_BE_UNIQUE);
+            HA_ANY_INDEX_MAY_BE_UNIQUE);
   }
   ulong index_flags(uint inx, uint part, bool all_parts) const
   {

sql/sql_parse.cc

@@ -4653,8 +4653,14 @@ bool add_field_to_list(THD *thd, char *field_name, enum_field_types type,
 	new_field->length++;
     }
     break;
-  case FIELD_TYPE_STRING:
   case FIELD_TYPE_VAR_STRING:
+    if (new_field->length < 4)
+    {
+      new_field->sql_type= FIELD_TYPE_STRING;
+      break;
+    }
+    /* fall through */
+  case FIELD_TYPE_STRING:
     if (new_field->length <= MAX_FIELD_CHARLENGTH || default_value)
       break;
     /* Convert long CHAR() and VARCHAR columns to TEXT or BLOB */