Merge mysql.com:/home/jonas/src/mysql-5.0

into mysql.com:/home/jonas/src/mysql-5.0-ndb


ndb/src/ndbapi/NdbDictionaryImpl.cpp:
  Auto merged
sql/ha_ndbcluster.cc:
  Auto merged
sql/mysqld.cc:
  Auto merged

34 files changed, 627 insertions, 441 deletions

diff --git a/sql/field.cc b/sql/field.cc
index 400ebf65273..a1dc02eba4a 100644
--- a/sql/field.cc
+++ b/sql/field.cc
@@ -1035,7 +1035,9 @@ int Field_decimal::store(longlong nr)
 double Field_decimal::val_real(void)
 {
   int not_used;
-  return my_strntod(&my_charset_bin, ptr, field_length, NULL, &not_used);
+  char *end_not_used;
+  return my_strntod(&my_charset_bin, ptr, field_length, &end_not_used,
+                    &not_used);
 }
 
 longlong Field_decimal::val_int(void)
@@ -4432,16 +4434,18 @@ int Field_string::store(longlong nr)
 double Field_string::val_real(void)
 {
   int not_used;
-  CHARSET_INFO *cs=charset();
-  return my_strntod(cs,ptr,field_length,(char**)0,&not_used);
+  char *end_not_used;
+  CHARSET_INFO *cs= charset();
+  return my_strntod(cs,ptr,field_length,&end_not_used,&not_used);
 }
 
 
 longlong Field_string::val_int(void)
 {
   int not_used;
+  char *end_not_used;
   CHARSET_INFO *cs=charset();
-  return my_strntoll(cs,ptr,field_length,10,NULL,&not_used);
+  return my_strntoll(cs,ptr,field_length,10,&end_not_used,&not_used);
 }
 
 
@@ -4740,8 +4744,9 @@ int Field_varstring::store(longlong nr)
 double Field_varstring::val_real(void)
 {
   int not_used;
+  char *end_not_used;
   uint length= length_bytes == 1 ? (uint) (uchar) *ptr : uint2korr(ptr);
-  return my_strntod(field_charset, ptr+length_bytes, length, (char**) 0,
+  return my_strntod(field_charset, ptr+length_bytes, length, &end_not_used,
                     &not_used);
 }
 
@@ -4749,9 +4754,10 @@ double Field_varstring::val_real(void)
 longlong Field_varstring::val_int(void)
 {
   int not_used;
+  char *end_not_used;
   uint length= length_bytes == 1 ? (uint) (uchar) *ptr : uint2korr(ptr);
-  return my_strntoll(field_charset, ptr+length_bytes, length, 10, NULL,
-                     &not_used);
+  return my_strntoll(field_charset, ptr+length_bytes, length, 10,
+                     &end_not_used, &not_used);
 }
 
 
@@ -5343,13 +5349,16 @@ int Field_blob::store(longlong nr)
 double Field_blob::val_real(void)
 {
   int not_used;
-  char *blob;
+  char *end_not_used, *blob;
+  uint32 length;
+  CHARSET_INFO *cs;
+
   memcpy_fixed(&blob,ptr+packlength,sizeof(char*));
   if (!blob)
     return 0.0;
-  uint32 length=get_length(ptr);
-  CHARSET_INFO *cs=charset();
-  return my_strntod(cs,blob,length,(char**)0, &not_used);
+  length= get_length(ptr);
+  cs= charset();
+  return my_strntod(cs, blob, length, &end_not_used, &not_used);
 }
 
 
@@ -6353,11 +6362,13 @@ longlong Field_bit::val_int(void)
 String *Field_bit::val_str(String *val_buffer,
                            String *val_ptr __attribute__((unused)))
 {
+  char buff[sizeof(longlong)];
   uint length= min(pack_length(), sizeof(longlong));
   ulonglong bits= val_int();
+  mi_int8store(buff,bits);
 
   val_buffer->alloc(length);
-  memcpy_fixed((char*) val_buffer->ptr(), (char*) &bits, length);
+  memcpy_fixed((char*) val_buffer->ptr(), buff+8-length, length);
   val_buffer->length(length);
   val_buffer->set_charset(&my_charset_bin);
   return val_buffer;
diff --git a/sql/field.h b/sql/field.h
index ab911822c2c..fd1ef09d14f 100644
--- a/sql/field.h
+++ b/sql/field.h
@@ -123,6 +123,7 @@ public:
   }
   virtual bool eq_def(Field *field);
   virtual uint32 pack_length() const { return (uint32) field_length; }
+  virtual uint32 pack_length_in_rec() const { return pack_length(); }
   virtual void reset(void) { bzero(ptr,pack_length()); }
   virtual void reset_fields() {}
   virtual void set_default()
@@ -1237,6 +1238,7 @@ public:
   { get_key_image(buff, length, itRAW); }
   uint32 pack_length() const 
   { return (uint32) field_length + (bit_len > 0); }
+  uint32 pack_length_in_rec() const { return field_length; }
   void sql_type(String &str) const;
   field_cast_enum field_cast_type() { return FIELD_CAST_BIT; }
   char *pack(char *to, const char *from, uint max_length=~(uint) 0);
@@ -1339,10 +1341,10 @@ int set_field_to_null_with_conversions(Field *field, bool no_conversions);
 #define FIELDFLAG_NUMBER		2
 #define FIELDFLAG_ZEROFILL		4
 #define FIELDFLAG_PACK			120	// Bits used for packing
-#define FIELDFLAG_INTERVAL		256
-#define FIELDFLAG_BITFIELD		512	// mangled with dec!
-#define FIELDFLAG_BLOB			1024	// mangled with dec!
-#define FIELDFLAG_GEOM			2048
+#define FIELDFLAG_INTERVAL		256     // mangled with decimals!
+#define FIELDFLAG_BITFIELD		512	// mangled with decimals!
+#define FIELDFLAG_BLOB			1024	// mangled with decimals!
+#define FIELDFLAG_GEOM			2048    // mangled with decimals!
 
 #define FIELDFLAG_LEFT_FULLSCREEN	8192
 #define FIELDFLAG_RIGHT_FULLSCREEN	16384
@@ -1366,10 +1368,10 @@ int set_field_to_null_with_conversions(Field *field, bool no_conversions);
 #define f_decimals(x)		((uint8) (((x) >> FIELDFLAG_DEC_SHIFT) & FIELDFLAG_MAX_DEC))
 #define f_is_alpha(x)		(!f_is_num(x))
 #define f_is_binary(x)          ((x) & FIELDFLAG_BINARY) // 4.0- compatibility
-#define f_is_enum(x)            ((x) & FIELDFLAG_INTERVAL)
-#define f_is_bitfield(x)	((x) & FIELDFLAG_BITFIELD)
+#define f_is_enum(x)            (((x) & (FIELDFLAG_INTERVAL | FIELDFLAG_NUMBER)) == FIELDFLAG_INTERVAL)
+#define f_is_bitfield(x)        (((x) & (FIELDFLAG_BITFIELD | FIELDFLAG_NUMBER)) == FIELDFLAG_BITFIELD)
 #define f_is_blob(x)		(((x) & (FIELDFLAG_BLOB | FIELDFLAG_NUMBER)) == FIELDFLAG_BLOB)
-#define f_is_geom(x)		((x) & FIELDFLAG_GEOM)
+#define f_is_geom(x)		(((x) & (FIELDFLAG_GEOM | FIELDFLAG_NUMBER)) == FIELDFLAG_GEOM)
 #define f_is_equ(x)		((x) & (1+2+FIELDFLAG_PACK+31*256))
 #define f_settype(x)		(((int) x) << FIELDFLAG_PACK_SHIFT)
 #define f_maybe_null(x)		(x & FIELDFLAG_MAYBE_NULL)
diff --git a/sql/field_conv.cc b/sql/field_conv.cc
index b337ccd6306..07cc90283b7 100644
--- a/sql/field_conv.cc
+++ b/sql/field_conv.cc
@@ -584,6 +584,7 @@ void field_conv(Field *to,Field *from)
   if (to->real_type() == from->real_type())
   {
     if (to->pack_length() == from->pack_length() &&
+        !(to->flags & UNSIGNED_FLAG && !(from->flags & UNSIGNED_FLAG)) &&
 	to->real_type() != FIELD_TYPE_ENUM &&
 	to->real_type() != FIELD_TYPE_SET &&
         from->charset() == to->charset() &&
diff --git a/sql/filesort.cc b/sql/filesort.cc
index fc8e27c5f94..0e9fa8c79ed 100644
--- a/sql/filesort.cc
+++ b/sql/filesort.cc
@@ -567,10 +567,10 @@ write_keys(SORTPARAM *param, register uchar **sort_keys, uint count,
   if (!my_b_inited(tempfile) &&
       open_cached_file(tempfile, mysql_tmpdir, TEMP_PREFIX, DISK_BUFFER_SIZE,
                        MYF(MY_WME)))
-    goto err;                                        /* purecov: inspected */
+    goto err;                                   /* purecov: inspected */
   buffpek.file_pos= my_b_tell(tempfile);
   if ((ha_rows) count > param->max_rows)
-    count=(uint) param->max_rows;                /* purecov: inspected */
+    count=(uint) param->max_rows;               /* purecov: inspected */
   buffpek.count=(ha_rows) count;
   for (end=sort_keys+count ; sort_keys != end ; sort_keys++)
     if (my_b_write(tempfile, (byte*) *sort_keys, (uint) rec_length))
@@ -841,11 +841,16 @@ int merge_many_buff(SORTPARAM *param, uchar *sort_buffer,
     if (flush_io_cache(to_file))
       break;					/* purecov: inspected */
     temp=from_file; from_file=to_file; to_file=temp;
+    setup_io_cache(from_file);
+    setup_io_cache(to_file);
     *maxbuffer= (uint) (lastbuff-buffpek)-1;
   }
   close_cached_file(to_file);			// This holds old result
   if (to_file == t_file)
+  {
     *t_file=t_file2;				// Copy result file
+    setup_io_cache(t_file);
+  }
 
   DBUG_RETURN(*maxbuffer >= MERGEBUFF2);	/* Return 1 if interrupted */
 } /* merge_many_buff */
@@ -1173,7 +1178,12 @@ sortlength(SORT_FIELD *sortorder, uint s_length, bool *multi_byte_charset)
       else
       {
 	sortorder->length=sortorder->field->pack_length();
-	if (use_strnxfrm((cs=sortorder->field->charset())))
+        /*
+          We must test cmp_type() to ensure that ENUM and SET are sorted
+          as numbers
+        */
+	if (use_strnxfrm((cs=sortorder->field->charset())) &&
+            sortorder->field->cmp_type() == STRING_RESULT)
 	{
 	  sortorder->need_strxnfrm= 1;
 	  *multi_byte_charset= 1;
diff --git a/sql/ha_heap.cc b/sql/ha_heap.cc
index b5742699d7e..02d81882e7a 100644
--- a/sql/ha_heap.cc
+++ b/sql/ha_heap.cc
@@ -471,12 +471,24 @@ int ha_heap::create(const char *name, TABLE *table_arg,
     KEY_PART_INFO *key_part=     pos->key_part;
     KEY_PART_INFO *key_part_end= key_part + pos->key_parts;
 
-    mem_per_row+= (pos->key_length + (sizeof(char*) * 2));
-
     keydef[key].keysegs=   (uint) pos->key_parts;
     keydef[key].flag=      (pos->flags & (HA_NOSAME | HA_NULL_ARE_EQUAL));
     keydef[key].seg=       seg;
-    keydef[key].algorithm= ((pos->algorithm == HA_KEY_ALG_UNDEF) ? 
+
+    switch (pos->algorithm) {
+    case HA_KEY_ALG_UNDEF:
+    case HA_KEY_ALG_HASH:
+      keydef[key].algorithm= HA_KEY_ALG_HASH;
+      mem_per_row+= sizeof(char*) * 2; // = sizeof(HASH_INFO)
+      break;
+    case HA_KEY_ALG_BTREE:
+      keydef[key].algorithm= HA_KEY_ALG_BTREE;
+      mem_per_row+=sizeof(TREE_ELEMENT)+pos->key_length+sizeof(char*);
+      break;
+    default:
+      DBUG_ASSERT(0); // cannot happen
+    }
+    keydef[key].algorithm= ((pos->algorithm == HA_KEY_ALG_UNDEF) ?
 			    HA_KEY_ALG_HASH : pos->algorithm);
 
     for (; key_part != key_part_end; key_part++, seg++)
@@ -525,8 +537,10 @@ int ha_heap::create(const char *name, TABLE *table_arg,
   hp_create_info.auto_key_type= auto_key_type;
   hp_create_info.auto_increment= (create_info->auto_increment_value ?
 				  create_info->auto_increment_value - 1 : 0);
+  hp_create_info.max_table_size=current_thd->variables.max_heap_table_size;
+  max_rows = (ha_rows) (hp_create_info.max_table_size / mem_per_row);
   error= heap_create(fn_format(buff,name,"","",4+2),
-		     keys,keydef, share->reclength,
+		     keys, keydef, share->reclength,
 		     (ulong) ((share->max_rows < max_rows &&
 			       share->max_rows) ? 
 			      share->max_rows : max_rows),
diff --git a/sql/ha_heap.h b/sql/ha_heap.h
index 8b44695df07..fb526888b01 100644
--- a/sql/ha_heap.h
+++ b/sql/ha_heap.h
@@ -32,7 +32,11 @@ class ha_heap: public handler
 public:
   ha_heap(TABLE *table): handler(table), file(0), records_changed(0) {}
   ~ha_heap() {}
-  const char *table_type() const { return "HEAP"; }
+  const char *table_type() const
+  {
+    return (table->in_use->variables.sql_mode & MODE_MYSQL323) ?
+           "HEAP" : "MEMORY";
+  }
   const char *index_type(uint inx)
   {
     return ((table->key_info[inx].algorithm == HA_KEY_ALG_BTREE) ? "BTREE" :
diff --git a/sql/ha_innodb.cc b/sql/ha_innodb.cc
index d49f1c1ad46..18ab6f42d28 100644
--- a/sql/ha_innodb.cc
+++ b/sql/ha_innodb.cc
@@ -2425,6 +2425,8 @@ build_template(
 
 		templ->mysql_col_len = (ulint) field->pack_length();
 		templ->type = get_innobase_type_from_mysql_type(field);
+		templ->charset = dtype_get_charset_coll_noninline(
+				index->table->cols[i].type.prtype);
 		templ->is_unsigned = (ulint) (field->flags & UNSIGNED_FLAG);
 
 		if (templ->type == DATA_BLOB) {
@@ -4092,6 +4094,48 @@ ha_innobase::discard_or_import_tablespace(
 }
 
 /*********************************************************************
+Deletes all rows of an InnoDB table. */
+
+int
+ha_innobase::delete_all_rows(void)
+/*==============================*/
+				/* out: error number */
+{
+	row_prebuilt_t*	prebuilt	= (row_prebuilt_t*)innobase_prebuilt;
+	int		error;
+	trx_t*		trx;
+	THD*		thd		= current_thd;
+
+	DBUG_ENTER("ha_innobase::delete_all_rows");
+
+	if (thd->lex->sql_command != SQLCOM_TRUNCATE) {
+	fallback:
+		/* We only handle TRUNCATE TABLE t as a special case.
+		DELETE FROM t will have to use ha_innobase::delete_row(). */
+		DBUG_RETURN(my_errno=HA_ERR_WRONG_COMMAND);
+	}
+
+	/* Get the transaction associated with the current thd, or create one
+	if not yet created */
+
+	trx = check_trx_exists(thd);
+
+	/* Truncate the table in InnoDB */
+
+	error = row_truncate_table_for_mysql(prebuilt->table, trx);
+	if (error == DB_ERROR) {
+		/* Cannot truncate; resort to ha_innobase::delete_row() */
+		goto fallback;
+	}
+
+	innobase_commit_low(trx);
+
+	error = convert_error_code_to_mysql(error, NULL);
+
+	DBUG_RETURN(error);
+}
+
+/*********************************************************************
 Drops a table from an InnoDB database. Before calling this function,
 MySQL calls innobase_commit to commit the transaction of the current user.
 Then the current user cannot have locks set on the table. Drop table
diff --git a/sql/ha_innodb.h b/sql/ha_innodb.h
index 0cb55e02ae3..2154e238fd1 100644
--- a/sql/ha_innodb.h
+++ b/sql/ha_innodb.h
@@ -159,6 +159,7 @@ class ha_innobase: public handler
 
   	int create(const char *name, register TABLE *form,
 					HA_CREATE_INFO *create_info);
+	int delete_all_rows();
   	int delete_table(const char *name);
 	int rename_table(const char* from, const char* to);
 	int check(THD* thd, HA_CHECK_OPT* check_opt);
diff --git a/sql/ha_myisam.cc b/sql/ha_myisam.cc
index 87329c6f4af..9631b78bca3 100644
--- a/sql/ha_myisam.cc
+++ b/sql/ha_myisam.cc
@@ -1467,7 +1467,7 @@ int ha_myisam::create(const char *name, register TABLE *table_arg,
 	  fieldpos <= minpos)
       {
 	/* skip null fields */
-	if (!(temp_length= (*field)->pack_length()))
+	if (!(temp_length= (*field)->pack_length_in_rec()))
 	  continue;				/* Skip null-fields */
 	if (! found || fieldpos < minpos ||
 	    (fieldpos == minpos && temp_length < length))
diff --git a/sql/ha_ndbcluster.cc b/sql/ha_ndbcluster.cc
index 5f07a7893d5..e26622422cd 100644
--- a/sql/ha_ndbcluster.cc
+++ b/sql/ha_ndbcluster.cc
@@ -656,7 +656,7 @@ int ha_ndbcluster::get_ndb_blobs_value(NdbBlob *last_ndb_blob)
           char *buf= m_blobs_buffer + offset;
           uint32 len= 0xffffffff;  // Max uint32
           DBUG_PRINT("value", ("read blob ptr=%x len=%u",
-                               (uint)buf, (uint)blob_len));
+                               (UintPtr)buf, (uint)blob_len));
           if (ndb_blob->readData(buf, len) != 0)
             DBUG_RETURN(-1);
           DBUG_ASSERT(len == blob_len);
@@ -1152,6 +1152,47 @@ ha_ndbcluster::set_index_key(NdbOperation *op,
   DBUG_RETURN(0);
 }
 
+inline 
+int ha_ndbcluster::define_read_attrs(byte* buf, NdbOperation* op)
+{
+  uint i;
+  THD *thd= current_thd;
+
+  DBUG_ENTER("define_read_attrs");  
+
+  // Define attributes to read
+  for (i= 0; i < table->s->fields; i++) 
+  {
+    Field *field= table->field[i];
+    if ((thd->query_id == field->query_id) ||
+	((field->flags & PRI_KEY_FLAG)) || 
+	m_retrieve_all_fields)
+    {      
+      if (get_ndb_value(op, field, i, buf))
+	ERR_RETURN(op->getNdbError());
+    } 
+    else 
+    {
+      m_value[i].ptr= NULL;
+    }
+  }
+    
+  if (table->s->primary_key == MAX_KEY) 
+  {
+    DBUG_PRINT("info", ("Getting hidden key"));
+    // Scanning table with no primary key
+    int hidden_no= table->s->fields;      
+#ifndef DBUG_OFF
+    const NDBTAB *tab= (const NDBTAB *) m_table;    
+    if (!tab->getColumn(hidden_no))
+      DBUG_RETURN(1);
+#endif
+    if (get_ndb_value(op, NULL, hidden_no, NULL))
+      ERR_RETURN(op->getNdbError());
+  }
+  DBUG_RETURN(0);
+} 
+
 /*
   Read one record from NDB using primary key
 */
@@ -1625,47 +1666,6 @@ int ha_ndbcluster::set_bounds(NdbIndexScanOperation *op,
   DBUG_RETURN(0);
 }
 
-inline 
-int ha_ndbcluster::define_read_attrs(byte* buf, NdbOperation* op)
-{
-  uint i;
-  THD *thd= current_thd;
-
-  DBUG_ENTER("define_read_attrs");  
-
-  // Define attributes to read
-  for (i= 0; i < table->s->fields; i++) 
-  {
-    Field *field= table->field[i];
-    if ((thd->query_id == field->query_id) ||
-	((field->flags & PRI_KEY_FLAG)) || 
-	m_retrieve_all_fields)
-    {      
-      if (get_ndb_value(op, field, i, buf))
-	ERR_RETURN(op->getNdbError());
-    } 
-    else 
-    {
-      m_value[i].ptr= NULL;
-    }
-  }
-    
-  if (table->s->primary_key == MAX_KEY) 
-  {
-    DBUG_PRINT("info", ("Getting hidden key"));
-    // Scanning table with no primary key
-    int hidden_no= table->s->fields;      
-#ifndef DBUG_OFF
-    const NDBTAB *tab= (const NDBTAB *) m_table;    
-    if (!tab->getColumn(hidden_no))
-      DBUG_RETURN(1);
-#endif
-    if (get_ndb_value(op, NULL, hidden_no, NULL))
-      ERR_RETURN(op->getNdbError());
-  }
-  DBUG_RETURN(0);
-} 
-
 /*
   Start ordered index scan in NDB
 */
@@ -2527,13 +2527,13 @@ int ha_ndbcluster::index_read_last(byte * buf, const byte * key, uint key_len)
 inline
 int ha_ndbcluster::read_range_first_to_buf(const key_range *start_key,
 					   const key_range *end_key,
-					   bool eq_range, bool sorted,
+					   bool eq_r, bool sorted,
 					   byte* buf)
 {
   KEY* key_info;
   int error= 1; 
   DBUG_ENTER("ha_ndbcluster::read_range_first_to_buf");
-  DBUG_PRINT("info", ("eq_range: %d, sorted: %d", eq_range, sorted));
+  DBUG_PRINT("info", ("eq_r: %d, sorted: %d", eq_r, sorted));
 
   switch (get_index_type(active_index)){
   case PRIMARY_KEY_ORDERED_INDEX:
@@ -2574,14 +2574,14 @@ int ha_ndbcluster::read_range_first_to_buf(const key_range *start_key,
 
 int ha_ndbcluster::read_range_first(const key_range *start_key,
 				    const key_range *end_key,
-				    bool eq_range, bool sorted)
+				    bool eq_r, bool sorted)
 {
   byte* buf= table->record[0];
   DBUG_ENTER("ha_ndbcluster::read_range_first");
   
   DBUG_RETURN(read_range_first_to_buf(start_key,
 				      end_key,
-				      eq_range, 
+				      eq_r, 
 				      sorted,
 				      buf));
 }
@@ -3271,7 +3271,7 @@ int ha_ndbcluster::start_stmt(THD *thd)
 #if 0    
     NdbTransaction *tablock_trans= 
       (NdbTransaction*)thd->transaction.all.ndb_tid;
-    DBUG_PRINT("info", ("tablock_trans: %x", (uint)tablock_trans));
+    DBUG_PRINT("info", ("tablock_trans: %x", (UintPtr)tablock_trans));
     DBUG_ASSERT(tablock_trans);
 //    trans= ndb->hupp(tablock_trans);
 #endif
diff --git a/sql/item.cc b/sql/item.cc
index c84496f8eb7..763ab84582d 100644
--- a/sql/item.cc
+++ b/sql/item.cc
@@ -46,7 +46,7 @@ void item_init(void)
 
 Item::Item():
   name(0), orig_name(0), name_length(0), fixed(0),
-  collation(default_charset(), DERIVATION_COERCIBLE)
+  collation(&my_charset_bin, DERIVATION_COERCIBLE)
 {
   marker= 0;
   maybe_null=null_value=with_sum_func=unsigned_flag=0;
@@ -1290,11 +1290,12 @@ double Item_param::val_real()
     return (double) value.integer;
   case STRING_VALUE:
   case LONG_DATA_VALUE:
-    {
-      int dummy_err;
-      return my_strntod(str_value.charset(), (char*) str_value.ptr(),
-                        str_value.length(), (char**) 0, &dummy_err);
-    }
+  {
+    int dummy_err;
+    char *end_not_used;
+    return my_strntod(str_value.charset(), (char*) str_value.ptr(),
+                      str_value.length(), &end_not_used, &dummy_err);
+  }
   case TIME_VALUE:
     /*
       This works for example when user says SELECT ?+0.0 and supplies
@@ -2545,8 +2546,9 @@ Item_num *Item_uint::neg()
 Item_real::Item_real(const char *str_arg, uint length)
 {
   int error;
-  char *end;
-  value= my_strntod(&my_charset_bin, (char*) str_arg, length, &end, &error);
+  char *end_not_used;
+  value= my_strntod(&my_charset_bin, (char*) str_arg, length, &end_not_used,
+                    &error);
   if (error)
   {
     /*
@@ -3522,12 +3524,12 @@ void Item_cache_str::store(Item *item)
 double Item_cache_str::val_real()
 {
   DBUG_ASSERT(fixed == 1);
-  int err;
+  int err_not_used;
+  char *end_not_used;
   if (value)
     return my_strntod(value->charset(), (char*) value->ptr(),
-		      value->length(), (char**) 0, &err);
-  else
-    return (double)0;
+		      value->length(), &end_not_used, &err_not_used);
+  return (double) 0;
 }
 
 
diff --git a/sql/item.h b/sql/item.h
index 2503f137355..6857b4acf82 100644
--- a/sql/item.h
+++ b/sql/item.h
@@ -877,9 +877,10 @@ public:
   double val_real()
   {
     DBUG_ASSERT(fixed == 1);
-    int err;
+    int err_not_used;
+    char *end_not_used;
     return my_strntod(str_value.charset(), (char*) str_value.ptr(),
-		      str_value.length(), (char**) 0, &err);
+		      str_value.length(), &end_not_used, &err_not_used);
   }
   longlong val_int()
   {
@@ -1230,10 +1231,11 @@ public:
   enum_field_types field_type() const { return cached_field_type; }
   double val_real()
   {
-    int err;
+    int err_not_used;
+    char *end_not_used;
     return (null_value ? 0.0 :
 	    my_strntod(str_value.charset(), (char*) str_value.ptr(),
-		       str_value.length(),NULL,&err));
+		       str_value.length(), &end_not_used, &err_not_used));
   }
   longlong val_int()
   { 
diff --git a/sql/item_cmpfunc.cc b/sql/item_cmpfunc.cc
index 16a3e86e519..4fbe7c241d1 100644
--- a/sql/item_cmpfunc.cc
+++ b/sql/item_cmpfunc.cc
@@ -1119,6 +1119,9 @@ Item_func_nullif::fix_length_and_dec()
     max_length=args[0]->max_length;
     decimals=args[0]->decimals;
     agg_result_type(&cached_result_type, args, 2);
+    if (cached_result_type == STRING_RESULT &&
+        agg_arg_charsets(collation, args, arg_count, MY_COLL_CMP_CONV))
+      return;
   }
 }
 
diff --git a/sql/item_func.h b/sql/item_func.h
index 4657ee81dfb..fb8d77d5b83 100644
--- a/sql/item_func.h
+++ b/sql/item_func.h
@@ -834,15 +834,19 @@ public:
   String *val_str(String *);
   double val_real()
   {
-    int err;
-    String *res;  res=val_str(&str_value);
-    return res ? my_strntod(res->charset(),(char*) res->ptr(),res->length(),0,&err) : 0.0;
+    int err_not_used;
+    char *end_not_used;
+    String *res;
+    res= val_str(&str_value);
+    return res ? my_strntod(res->charset(),(char*) res->ptr(), 
+                            res->length(), &end_not_used, &err_not_used) : 0.0;
   }
   longlong val_int()
   {
-    int err;
+    int err_not_used;
     String *res;  res=val_str(&str_value);
-    return res ? my_strntoll(res->charset(),res->ptr(),res->length(),10,(char**) 0,&err) : (longlong) 0;
+    return res ? my_strntoll(res->charset(),res->ptr(),res->length(),10,
+                             (char**) 0, &err_not_used) : (longlong) 0;
   }
   enum Item_result result_type () const { return STRING_RESULT; }
   void fix_length_and_dec();
diff --git a/sql/item_strfunc.cc b/sql/item_strfunc.cc
index 9e28acdd091..5d018b22055 100644
--- a/sql/item_strfunc.cc
+++ b/sql/item_strfunc.cc
@@ -58,17 +58,19 @@ uint nr_of_decimals(const char *str)
   return 0;
 }
 
+
 double Item_str_func::val_real()
 {
   DBUG_ASSERT(fixed == 1);
-  int err;
-  char buff[64];
+  int err_not_used;
+  char *end_not_used, buff[64];
   String *res, tmp(buff,sizeof(buff), &my_charset_bin);
   res= val_str(&tmp);
-  return res ? my_strntod(res->charset(), (char*) res->ptr(),res->length(),
-			  NULL, &err) : 0.0;
+  return res ? my_strntod(res->charset(), (char*) res->ptr(), res->length(),
+			  &end_not_used, &err_not_used) : 0.0;
 }
 
+
 longlong Item_str_func::val_int()
 {
   DBUG_ASSERT(fixed == 1);
@@ -275,7 +277,8 @@ String *Item_func_concat::val_str(String *str)
 			    current_thd->variables.max_allowed_packet);
 	goto null;
       }
-      if (res->alloced_length() >= res->length()+res2->length())
+      if (!args[0]->const_item() && 
+          res->alloced_length() >= res->length()+res2->length())
       {						// Use old buffer
 	res->append(*res2);
       }
@@ -1904,6 +1907,7 @@ b1:        str->append((char)(num>>8));
 #endif
       str->append((char)num);
   }
+  str->set_charset(collation.collation);
   str->realloc(str->length());			// Add end 0 (for Purify)
   return str;
 }
diff --git a/sql/item_sum.cc b/sql/item_sum.cc
index 168c68ad706..be89aa3f86d 100644
--- a/sql/item_sum.cc
+++ b/sql/item_sum.cc
@@ -591,14 +591,17 @@ void Item_sum_hybrid::clear()
 double Item_sum_hybrid::val_real()
 {
   DBUG_ASSERT(fixed == 1);
-  int err;
   if (null_value)
     return 0.0;
   switch (hybrid_type) {
   case STRING_RESULT:
+  {
+    char *end_not_used;
+    int err_not_used;
     String *res;  res=val_str(&str_value);
-    return (res ? my_strntod(res->charset(), (char*) res->ptr(),res->length(),
-			     (char**) 0, &err) : 0.0);
+    return (res ? my_strntod(res->charset(), (char*) res->ptr(), res->length(),
+			     &end_not_used, &err_not_used) : 0.0);
+  }
   case INT_RESULT:
     if (unsigned_flag)
       return ulonglong2double(sum_int);
diff --git a/sql/item_sum.h b/sql/item_sum.h
index 68899268b31..7866a9ae913 100644
--- a/sql/item_sum.h
+++ b/sql/item_sum.h
@@ -643,16 +643,18 @@ public:
   String *val_str(String *);
   double val_real()
   {
-    int err;
+    int err_not_used;
+    char *end_not_used;
     String *res;  res=val_str(&str_value);
     return res ? my_strntod(res->charset(),(char*) res->ptr(),res->length(),
-			    (char**) 0, &err) : 0.0;
+			    &end_not_used, &err_not_used) : 0.0;
   }
   longlong val_int()
   {
-    int err;
+    int err_not_used;
     String *res;  res=val_str(&str_value);
-    return res ? my_strntoll(res->charset(),res->ptr(),res->length(),10, (char**) 0, &err) : (longlong) 0;
+    return res ? my_strntoll(res->charset(),res->ptr(),res->length(),10,
+                             (char**) 0, &err_not_used) : (longlong) 0;
   }
   enum Item_result result_type () const { return STRING_RESULT; }
   void fix_length_and_dec();
@@ -783,9 +785,10 @@ class Item_func_group_concat : public Item_sum
     String *res;
     char *end_ptr;
     int error;
-    res= val_str(&str_value);
+    if (!(res= val_str(&str_value)))
+      return (longlong) 0;
     end_ptr= (char*) res->ptr()+ res->length();
-    return res ? my_strtoll10(res->ptr(), &end_ptr, &error) : (longlong) 0;
+    return my_strtoll10(res->ptr(), &end_ptr, &error);
   }
   String* val_str(String* str);
   Item *copy_or_same(THD* thd);
diff --git a/sql/item_timefunc.cc b/sql/item_timefunc.cc
index 2d0e5d7632f..0c1cd3cbad3 100644
--- a/sql/item_timefunc.cc
+++ b/sql/item_timefunc.cc
@@ -2441,36 +2441,37 @@ void Item_func_add_time::print(String *str)
 
 
 /*
+  Calculate difference between two datetime values as seconds + microseconds.
+
   SYNOPSIS
     calc_time_diff()
-    l_time1		TIME/DATE/DATETIME value
-    l_time2		TIME/DATE/DATETIME value
-    l_sign		Can be 1 (operation of addition)
-                        or -1 (substraction)
-    seconds_out         Returns count of seconds bitween
-                        l_time1 and l_time2
-    microseconds_out    Returns count of microseconds bitween
-                        l_time1 and l_time2.
-
-  DESCRIPTION
-    Calculates difference in seconds(seconds_out)
-    and microseconds(microseconds_out)
-    bitween two TIME/DATE/DATETIME values.
+      l_time1         - TIME/DATE/DATETIME value
+      l_time2         - TIME/DATE/DATETIME value
+      l_sign          - 1 absolute values are substracted,
+                        -1 absolute values are added.
+      seconds_out     - Out parameter where difference between
+                        l_time1 and l_time2 in seconds is stored.
+      microseconds_out- Out parameter where microsecond part of difference
+                        between l_time1 and l_time2 is stored.
+
+  NOTE
+    This function calculates difference between l_time1 and l_time2 absolute
+    values. So one should set l_sign and correct result if he want to take
+    signs into account (i.e. for TIME values).
 
   RETURN VALUES
-    Rertuns sign of difference.
+    Returns sign of difference.
     1 means negative result
     0 means positive result
 
 */
 
-bool calc_time_diff(TIME *l_time1,TIME *l_time2, int l_sign,
-		    longlong *seconds_out, long *microseconds_out)
+static bool calc_time_diff(TIME *l_time1, TIME *l_time2, int l_sign,
+                           longlong *seconds_out, long *microseconds_out)
 {
   long days;
   bool neg;
-  longlong seconds= *seconds_out;
-  long microseconds= *microseconds_out;
+  longlong microseconds;
 
   /*
     We suppose that if first argument is MYSQL_TIMESTAMP_TIME
@@ -2487,29 +2488,20 @@ bool calc_time_diff(TIME *l_time1,TIME *l_time2, int l_sign,
 			     (uint) l_time2->month,
 			     (uint) l_time2->day));
 
-  microseconds= l_time1->second_part - l_sign*l_time2->second_part;
-  seconds= ((longlong) days*86400L + l_time1->hour*3600L + 
-	    l_time1->minute*60L + l_time1->second + microseconds/1000000L -
-	    (longlong)l_sign*(l_time2->hour*3600L+l_time2->minute*60L+l_time2->second));
+  microseconds= ((longlong)days*86400L +
+                 l_time1->hour*3600L + l_time1->minute*60L + l_time1->second -
+                 (longlong)l_sign*(l_time2->hour*3600L + l_time2->minute*60L +
+                                   l_time2->second))*1000000L +
+                l_time1->second_part - l_sign*l_time2->second_part;
 
   neg= 0;
-  if (seconds < 0)
-  {
-    seconds= -seconds;
-    neg= 1;
-  }
-  else if (seconds == 0 && microseconds < 0)
+  if (microseconds < 0)
   {
     microseconds= -microseconds;
     neg= 1;
   }
-  if (microseconds < 0)
-  {
-    microseconds+= 1000000L;
-    seconds--;
-  }
-  *seconds_out= seconds;
-  *microseconds_out= microseconds;
+  *seconds_out= microseconds/1000000L;
+  *microseconds_out= (long) (microseconds%1000000L);
   return neg;
 }
 
@@ -2544,9 +2536,10 @@ String *Item_func_timediff::val_str(String *str)
   /*
     For MYSQL_TIMESTAMP_TIME only:
       If both argumets are negative values and diff between them
-      is negative we need to swap sign as result should be positive.
+      is non-zero we need to swap sign to get proper result.
   */
-  if ((l_time2.neg == l_time1.neg) && l_time1.neg)
+  if ((l_time2.neg == l_time1.neg) && l_time1.neg &&
+      (seconds || microseconds))
     l_time3.neg= 1-l_time3.neg;         // Swap sign of result
 
   calc_time_from_sec(&l_time3, (long) seconds, microseconds);
@@ -2712,13 +2705,11 @@ longlong Item_func_timestamp_diff::val_int()
   case INTERVAL_SECOND:		
     return seconds*neg;
   case INTERVAL_MICROSECOND:
-  {
-    longlong max_sec= LONGLONG_MAX/1000000;
-    if (max_sec > seconds ||
-	max_sec == seconds && LONGLONG_MAX%1000000 >= microseconds)
-      return (longlong) (seconds*1000000L+microseconds)*neg;
-    goto null_date;
-  }
+    /*
+      In MySQL difference between any two valid datetime values
+      in microseconds fits into longlong.
+    */
+    return (seconds*1000000L+microseconds)*neg;
   default:
     break;
   }
diff --git a/sql/log_event.cc b/sql/log_event.cc
index 2fa4e09913e..5ee034d785e 100644
--- a/sql/log_event.cc
+++ b/sql/log_event.cc
@@ -3201,7 +3201,7 @@ void User_var_log_event::print(FILE* file, bool short_form, LAST_EVENT_INFO* las
         */
         fprintf(file, ":=???;\n");
       else
-        fprintf(file, ":=_%s %s COLLATE %s;\n", cs->csname, hex_str, cs->name);
+        fprintf(file, ":=_%s %s COLLATE `%s`;\n", cs->csname, hex_str, cs->name);
       my_afree(hex_str);
     }
       break;
diff --git a/sql/mysql_priv.h b/sql/mysql_priv.h
index 6f569ea3ef4..322b38abe13 100644
--- a/sql/mysql_priv.h
+++ b/sql/mysql_priv.h
@@ -831,6 +831,7 @@ void wait_for_refresh(THD *thd);
 int open_tables(THD *thd, TABLE_LIST *tables, uint *counter);
 int simple_open_n_lock_tables(THD *thd,TABLE_LIST *tables);
 bool open_and_lock_tables(THD *thd,TABLE_LIST *tables);
+bool open_normal_and_derived_tables(THD *thd, TABLE_LIST *tables);
 int lock_tables(THD *thd, TABLE_LIST *tables, uint counter);
 TABLE *open_temporary_table(THD *thd, const char *path, const char *db,
 			    const char *table_name, bool link_in_list);
diff --git a/sql/mysqld.cc b/sql/mysqld.cc
index 14bcba05143..7a95981e1dc 100644
--- a/sql/mysqld.cc
+++ b/sql/mysqld.cc
@@ -54,7 +54,7 @@
 #endif
 #ifdef HAVE_NDBCLUSTER_DB
 #define OPT_NDBCLUSTER_DEFAULT 0
-#if defined(NDB_SHM_TRANSPORTER) && MYSQL_VERSION_ID >= 50000
+#if defined(NOT_ENOUGH_TESTED) && defined(NDB_SHM_TRANSPORTER) && MYSQL_VERSION_ID >= 50000
 #define OPT_NDB_SHM_DEFAULT 1
 #else
 #define OPT_NDB_SHM_DEFAULT 0
@@ -2700,12 +2700,37 @@ with --log-bin instead.");
 	mysql_bin_log.purge_logs_before_date(purge_time);
     }
 #endif
+    if (!opt_bin_logname && !opt_binlog_index_name)
+    {
+      /*
+        User didn't give us info to name the binlog index file.
+        Picking `hostname`-bin.index like did in 4.x, causes replication to
+        fail if the hostname is changed later. So, we would like to instead
+        require a name. But as we don't want to break many existing setups, we
+        only give warning, not error.
+      */
+      sql_print_warning("\
+No argument was provided to --log-bin, and --log-bin-index was not used; \
+so replication may break when this MySQL server acts as a master and \
+has his hostname changed!! Please use '--log-bin=%s' to avoid \
+this problem.",
+                        mysql_bin_log.get_name());
+    }
   }
-  else if (opt_log_slave_updates)
+  else
   {
-      sql_print_warning("\
-you need to use --log-bin to make --log-slave-updates work. \
-Now disabling --log-slave-updates.");
+    if (opt_log_slave_updates)
+    {
+      sql_print_error("\
+You need to use --log-bin=# to make --log-slave-updates work.");
+      unireg_abort(1);
+    }
+    if (opt_binlog_index_name)
+    {
+      sql_print_error("\
+You need to use --log-bin=# to make --log-bin-index work.");
+      unireg_abort(1);
+    }
   }
 
 #ifdef HAVE_REPLICATION
@@ -4424,9 +4449,11 @@ Disable with --skip-large-pages.",
 Disable with --skip-innodb (will save memory).",
    (gptr*) &opt_innodb, (gptr*) &opt_innodb, 0, GET_BOOL, NO_ARG, OPT_INNODB_DEFAULT, 0, 0,
    0, 0, 0},
+#ifdef HAVE_INNOBASE_DB
   {"innodb_checksums", OPT_INNODB_CHECKSUMS, "Enable InnoDB checksums validation (enabled by default). \
 Disable with --skip-innodb-checksums.", (gptr*) &innobase_use_checksums,
    (gptr*) &innobase_use_checksums, 0, GET_BOOL, NO_ARG, 1, 0, 0, 0, 0, 0},
+#endif
   {"innodb_data_file_path", OPT_INNODB_DATA_FILE_PATH,
    "Path to individual files and their sizes.",
    0, 0, 0, GET_STR, REQUIRED_ARG, 0, 0, 0, 0, 0, 0},
@@ -4502,14 +4529,16 @@ Disable with --skip-isam.",
   {"log", 'l', "Log connections and queries to file.", (gptr*) &opt_logname,
    (gptr*) &opt_logname, 0, GET_STR, OPT_ARG, 0, 0, 0, 0, 0, 0},
   {"log-bin", OPT_BIN_LOG,
-   "Log update queries in binary format.",
+   "Log update queries in binary format. Optional (but strongly recommended "
+   "to avoid replication problems if server's hostname changes) argument "
+   "should be the chosen location for the binary log files.",
    (gptr*) &opt_bin_logname, (gptr*) &opt_bin_logname, 0, GET_STR_ALLOC,
    OPT_ARG, 0, 0, 0, 0, 0, 0},
   {"log-bin-index", OPT_BIN_LOG_INDEX,
    "File that holds the names for last binary log files.",
    (gptr*) &opt_binlog_index_name, (gptr*) &opt_binlog_index_name, 0, GET_STR,
    REQUIRED_ARG, 0, 0, 0, 0, 0, 0},
-  {"log-error", OPT_ERROR_LOG_FILE, "Log error file.",
+  {"log-error", OPT_ERROR_LOG_FILE, "Error log file.",
    (gptr*) &log_error_file_ptr, (gptr*) &log_error_file_ptr, 0, GET_STR,
    OPT_ARG, 0, 0, 0, 0, 0, 0},
   {"log-isam", OPT_ISAM_LOG, "Log all MyISAM changes to file.",
diff --git a/sql/opt_range.cc b/sql/opt_range.cc
index 4b0e5f036cb..acf69fdd098 100644
--- a/sql/opt_range.cc
+++ b/sql/opt_range.cc
@@ -50,6 +50,11 @@
 #define test_use_count(A) {}
 #endif
 
+/*
+  Convert double value to #rows. Currently this does floor(), and we 
+  might consider using round() instead.
+*/
+#define double2rows(x) ((ha_rows)(x))
 
 static int sel_cmp(Field *f,char *a,char *b,uint8 a_flag,uint8 b_flag);
 
@@ -1628,6 +1633,7 @@ int SQL_SELECT::test_quick_select(THD *thd, key_map keys_to_use,
   DBUG_PRINT("enter",("keys_to_use: %lu  prev_tables: %lu  const_tables: %lu",
 		      keys_to_use.to_ulonglong(), (ulong) prev_tables,
 		      (ulong) const_tables));
+  DBUG_PRINT("info", ("records=%lu", (ulong)head->file->records));
   delete quick;
   quick=0;
   needed_reg.clear_all();
@@ -1874,6 +1880,7 @@ int SQL_SELECT::test_quick_select(THD *thd, key_map keys_to_use,
 double get_sweep_read_cost(const PARAM *param, ha_rows records)
 {
   double result;
+  DBUG_ENTER("get_sweep_read_cost");
   if (param->table->file->primary_key_is_clustered())
   {
     result= param->table->file->read_time(param->table->s->primary_key,
@@ -1912,7 +1919,7 @@ double get_sweep_read_cost(const PARAM *param, ha_rows records)
     }
   }
   DBUG_PRINT("info",("returning cost=%g", result));
-  return result;
+  DBUG_RETURN(result);
 }
 
 
@@ -2393,43 +2400,27 @@ typedef struct
 {
   const PARAM *param;
   MY_BITMAP covered_fields; /* union of fields covered by all scans */
-
-  /* TRUE if covered_fields is a superset of needed_fields */
-  bool is_covering;
-
-  double index_scan_costs; /* SUM(cost of 'index-only' scans) */
-  double total_cost;
   /*
     Fraction of table records that satisfies conditions of all scans.
     This is the number of full records that will be retrieved if a
     non-index_only index intersection will be employed.
   */
-  double records_fract;
+  double out_rows;
+  /* TRUE if covered_fields is a superset of needed_fields */
+  bool is_covering;
+
   ha_rows index_records; /* sum(#records to look in indexes) */
+  double index_scan_costs; /* SUM(cost of 'index-only' scans) */
+  double total_cost;
 } ROR_INTERSECT_INFO;
 
 
 /*
-  Re-initialize an allocated intersect info to contain zero scans.
-  SYNOPSIS
-    info Intersection info structure to re-initialize.
-*/
-
-static void ror_intersect_reinit(ROR_INTERSECT_INFO *info)
-{
-  info->is_covering= FALSE;
-  info->index_scan_costs= 0.0f;
-  info->records_fract= 1.0f;
-  bitmap_clear_all(&info->covered_fields);
-}
-
-/*
   Allocate a ROR_INTERSECT_INFO and initialize it to contain zero scans.
 
   SYNOPSIS
     ror_intersect_init()
       param         Parameter from test_quick_select
-      is_index_only If TRUE, set ROR_INTERSECT_INFO to be covering
 
   RETURN
     allocated structure
@@ -2437,7 +2428,7 @@ static void ror_intersect_reinit(ROR_INTERSECT_INFO *info)
 */
 
 static
-ROR_INTERSECT_INFO* ror_intersect_init(const PARAM *param, bool is_index_only)
+ROR_INTERSECT_INFO* ror_intersect_init(const PARAM *param)
 {
   ROR_INTERSECT_INFO *info;
   uchar* buf;
@@ -2450,46 +2441,39 @@ ROR_INTERSECT_INFO* ror_intersect_init(const PARAM *param, bool is_index_only)
   if (bitmap_init(&info->covered_fields, buf, param->fields_bitmap_size*8,
                   FALSE))
     return NULL;
-  ror_intersect_reinit(info);
+  info->is_covering= FALSE;
+  info->index_scan_costs= 0.0;
+  info->index_records= 0;
+  info->out_rows= param->table->file->records;
+  bitmap_clear_all(&info->covered_fields);
   return info;
 }
 
-
+void ror_intersect_cpy(ROR_INTERSECT_INFO *dst, const ROR_INTERSECT_INFO *src)
+{
+  dst->param= src->param;
+  memcpy(dst->covered_fields.bitmap, src->covered_fields.bitmap, 
+         src->covered_fields.bitmap_size);
+  dst->out_rows= src->out_rows;
+  dst->is_covering= src->is_covering;
+  dst->index_records= src->index_records;
+  dst->index_scan_costs= src->index_scan_costs;
+  dst->total_cost= src->total_cost;
+}
 /*
-  Check if adding a ROR scan to a ROR-intersection reduces its cost of
-  ROR-intersection and if yes, update parameters of ROR-intersection,
-  including its cost.
+  Get selectivity of a ROR scan wrt ROR-intersection.
 
   SYNOPSIS
-    ror_intersect_add()
-      param        Parameter from test_quick_select
-      info         ROR-intersection structure to add the scan to.
-      ror_scan     ROR scan info to add.
-      is_cpk_scan  If TRUE, add the scan as CPK scan (this can be inferred
-                   from other parameters and is passed separately only to
-                   avoid duplicating the inference code)
-
+    ror_scan_selectivity()
+      info  ROR-interection 
+      scan  ROR scan
+      
   NOTES
-    Adding a ROR scan to ROR-intersect "makes sense" iff the cost of ROR-
-    intersection decreases. The cost of ROR-intersection is caclulated as
-    follows:
-
-    cost= SUM_i(key_scan_cost_i) + cost_of_full_rows_retrieval
-
-    if (union of indexes used covers all needed fields)
-      cost_of_full_rows_retrieval= 0;
-    else
-    {
-      cost_of_full_rows_retrieval=
-        cost_of_sweep_read(E(rows_to_retrieve), rows_in_table);
-    }
-
-    E(rows_to_retrieve) is caclulated as follows:
     Suppose we have a condition on several keys
     cond=k_11=c_11 AND k_12=c_12 AND ...  // parts of first key
          k_21=c_21 AND k_22=c_22 AND ...  // parts of second key
           ...
-         k_n1=c_n1 AND k_n3=c_n3 AND ...  (1)
+         k_n1=c_n1 AND k_n3=c_n3 AND ...  (1) //parts of the key used by *scan
 
     where k_ij may be the same as any k_pq (i.e. keys may have common parts).
 
@@ -2563,38 +2547,30 @@ ROR_INTERSECT_INFO* ror_intersect_init(const PARAM *param, bool is_index_only)
     and reduce adjacent fractions.
 
   RETURN
-    TRUE   ROR scan added to ROR-intersection, cost updated.
-    FALSE  It doesn't make sense to add this ROR scan to this ROR-intersection.
+    Selectivity of given ROR scan.
+    
 */
 
-bool ror_intersect_add(const PARAM *param, ROR_INTERSECT_INFO *info,
-                       ROR_SCAN_INFO* ror_scan, bool is_cpk_scan=FALSE)
+static double ror_scan_selectivity(const ROR_INTERSECT_INFO *info, 
+                                   const ROR_SCAN_INFO *scan)
 {
-  int i;
-  SEL_ARG *sel_arg;
-  KEY_PART_INFO *key_part=
-    info->param->table->key_info[ror_scan->keynr].key_part;
   double selectivity_mult= 1.0;
+  KEY_PART_INFO *key_part= info->param->table->key_info[scan->keynr].key_part;
   byte key_val[MAX_KEY_LENGTH+MAX_FIELD_WIDTH]; /* key values tuple */
-
-  DBUG_ENTER("ror_intersect_add");
-  DBUG_PRINT("info", ("Current selectivity= %g", info->records_fract));
-  DBUG_PRINT("info", ("Adding scan on %s",
-                      info->param->table->key_info[ror_scan->keynr].name));
-  SEL_ARG *tuple_arg= NULL;
   char *key_ptr= (char*) key_val;
-  bool cur_covered, prev_covered=
-     bitmap_is_set(&info->covered_fields, key_part->fieldnr);
-
-  ha_rows prev_records= param->table->file->records;
+  SEL_ARG *sel_arg, *tuple_arg= NULL;
+  bool cur_covered;
+  bool prev_covered= bitmap_is_set(&info->covered_fields, key_part->fieldnr);
   key_range min_range;
   key_range max_range;
   min_range.key= (byte*) key_val;
   min_range.flag= HA_READ_KEY_EXACT;
   max_range.key= (byte*) key_val;
   max_range.flag= HA_READ_AFTER_KEY;
-
-  for(i= 0, sel_arg= ror_scan->sel_arg; sel_arg;
+  ha_rows prev_records= info->param->table->file->records;
+  int i;
+  DBUG_ENTER("ror_intersect_selectivity");
+  for(i= 0, sel_arg= scan->sel_arg; sel_arg;
       i++, sel_arg= sel_arg->next_key_part)
   {
     cur_covered= bitmap_is_set(&info->covered_fields, (key_part + i)->fieldnr);
@@ -2604,7 +2580,7 @@ bool ror_intersect_add(const PARAM *param, ROR_INTERSECT_INFO *info,
       {
         if (!tuple_arg)
         {
-          tuple_arg= ror_scan->sel_arg;
+          tuple_arg= scan->sel_arg;
           tuple_arg->store_min(key_part->length, &key_ptr, 0);
         }
         while (tuple_arg->next_key_part != sel_arg)
@@ -2615,10 +2591,8 @@ bool ror_intersect_add(const PARAM *param, ROR_INTERSECT_INFO *info,
       }
       ha_rows records;
       min_range.length= max_range.length= ((char*) key_ptr - (char*) key_val);
-      records= param->table->file->
-                 records_in_range(ror_scan->keynr,
-                                  &min_range,
-                                  &max_range);
+      records= info->param->table->file->
+                 records_in_range(scan->keynr, &min_range, &max_range);
       if (cur_covered)
       {
         /* uncovered -> covered */
@@ -2637,49 +2611,105 @@ bool ror_intersect_add(const PARAM *param, ROR_INTERSECT_INFO *info,
   }
   if (!prev_covered)
   {
-    double tmp= rows2double(param->table->quick_rows[ror_scan->keynr]) /
+    double tmp= rows2double(info->param->table->quick_rows[scan->keynr]) /
                 rows2double(prev_records);
     DBUG_PRINT("info", ("Selectivity multiplier: %g", tmp));
     selectivity_mult *= tmp;
   }
+  DBUG_PRINT("info", ("Returning multiplier: %g", selectivity_mult));
+  DBUG_RETURN(selectivity_mult);
+}
+
+/*
+  Check if adding a ROR scan to a ROR-intersection reduces its cost of
+  ROR-intersection and if yes, update parameters of ROR-intersection,
+  including its cost.
+
+  SYNOPSIS
+    ror_intersect_add()
+      param        Parameter from test_quick_select
+      info         ROR-intersection structure to add the scan to.
+      ror_scan     ROR scan info to add.
+      is_cpk_scan  If TRUE, add the scan as CPK scan (this can be inferred
+                   from other parameters and is passed separately only to
+                   avoid duplicating the inference code)
+
+  NOTES
+    Adding a ROR scan to ROR-intersect "makes sense" iff the cost of ROR-
+    intersection decreases. The cost of ROR-intersection is calculated as
+    follows:
+
+    cost= SUM_i(key_scan_cost_i) + cost_of_full_rows_retrieval
+
+    When we add a scan the first increases and the second decreases.
+
+    cost_of_full_rows_retrieval=
+      (union of indexes used covers all needed fields) ?
+        cost_of_sweep_read(E(rows_to_retrieve), rows_in_table) :
+        0
+
+    E(rows_to_retrieve) = #rows_in_table * ror_scan_selectivity(null, scan1) *
+                           ror_scan_selectivity({scan1}, scan2) * ... *
+                           ror_scan_selectivity({scan1,...}, scanN). 
+  RETURN
+    TRUE   ROR scan added to ROR-intersection, cost updated.
+    FALSE  It doesn't make sense to add this ROR scan to this ROR-intersection.
+*/
+
+static bool ror_intersect_add(ROR_INTERSECT_INFO *info,
+                       ROR_SCAN_INFO* ror_scan, bool is_cpk_scan)
+{
+  double selectivity_mult= 1.0;
+
+  DBUG_ENTER("ror_intersect_add");
+  DBUG_PRINT("info", ("Current out_rows= %g", info->out_rows));
+  DBUG_PRINT("info", ("Adding scan on %s",
+                      info->param->table->key_info[ror_scan->keynr].name));
+  DBUG_PRINT("info", ("is_cpk_scan=%d",is_cpk_scan));
 
+  selectivity_mult = ror_scan_selectivity(info, ror_scan);
   if (selectivity_mult == 1.0)
   {
     /* Don't add this scan if it doesn't improve selectivity. */
     DBUG_PRINT("info", ("The scan doesn't improve selectivity."));
-    DBUG_RETURN(FALSE);
+    DBUG_RETURN(false);
   }
-
-  info->records_fract *= selectivity_mult;
-  ha_rows cur_scan_records= info->param->table->quick_rows[ror_scan->keynr];
+  
+  info->out_rows *= selectivity_mult;
+  DBUG_PRINT("info", ("info->total_cost= %g", info->total_cost));
+  
   if (is_cpk_scan)
   {
-    info->index_scan_costs += rows2double(cur_scan_records)*
+    /*
+      CPK scan is used to filter out rows. We apply filtering for 
+      each record of every scan. Assuming 1/TIME_FOR_COMPARE_ROWID
+      per check this gives us:
+    */
+    info->index_scan_costs += rows2double(info->index_records) / 
                               TIME_FOR_COMPARE_ROWID;
   }
   else
   {
-    info->index_records += cur_scan_records;
+    info->index_records += info->param->table->quick_rows[ror_scan->keynr];
     info->index_scan_costs += ror_scan->index_read_cost;
     bitmap_union(&info->covered_fields, &ror_scan->covered_fields);
-  }
-
-  if (!info->is_covering && bitmap_is_subset(&info->param->needed_fields,
-                                             &info->covered_fields))
-  {
-    DBUG_PRINT("info", ("ROR-intersect is covering now"));
-    info->is_covering= TRUE;
+    if (!info->is_covering && bitmap_is_subset(&info->param->needed_fields,
+                                               &info->covered_fields))
+    {
+      DBUG_PRINT("info", ("ROR-intersect is covering now"));
+      info->is_covering= TRUE;
+    }
   }
 
   info->total_cost= info->index_scan_costs;
+  DBUG_PRINT("info", ("info->total_cost= %g", info->total_cost));
   if (!info->is_covering)
   {
-    ha_rows table_recs= info->param->table->file->records;
-    info->total_cost +=
-      get_sweep_read_cost(info->param,
-                          (ha_rows)(info->records_fract*table_recs));
+    info->total_cost += 
+      get_sweep_read_cost(info->param, double2rows(info->out_rows));
+    DBUG_PRINT("info", ("info->total_cost= %g", info->total_cost));
   }
-  DBUG_PRINT("info", ("New selectivity= %g", info->records_fract));
+  DBUG_PRINT("info", ("New out_rows= %g", info->out_rows));
   DBUG_PRINT("info", ("New cost= %g, %scovering", info->total_cost,
                       info->is_covering?"" : "non-"));
   DBUG_RETURN(TRUE);
@@ -2701,9 +2731,13 @@ bool ror_intersect_add(const PARAM *param, ROR_INTERSECT_INFO *info,
                        a covering ROR-intersection)
 
   NOTES
-    get_key_scans_params must be called before for the same SEL_TREE before
-    this function can be called.
+    get_key_scans_params must be called before this function can be called.
+    
+    When this function is called by ROR-union construction algorithm it
+    assumes it is building an uncovered ROR-intersection (and thus # of full
+    records to be retrieved is wrong here). This is a hack.
 
+  IMPLEMENTATION
     The approximate best non-covering plan search algorithm is as follows:
 
     find_min_ror_intersection_scan()
@@ -2717,11 +2751,11 @@ bool ror_intersect_add(const PARAM *param, ROR_INTERSECT_INFO *info,
       min_scan= make_scan(S);
       while (R is not empty)
       {
-        if (!selectivity(S + first(R) < selectivity(S)))
+        firstR= R - first(R);
+        if (!selectivity(S + firstR < selectivity(S)))
           continue;
-
+          
         S= S + first(R);
-        R= R - first(R);
         if (cost(S) < min_cost)
         {
           min_cost= cost(S);
@@ -2752,15 +2786,15 @@ TRP_ROR_INTERSECT *get_best_ror_intersect(const PARAM *param, SEL_TREE *tree,
                                           bool *are_all_covering)
 {
   uint idx;
-  double min_cost= read_time;
+  double min_cost= DBL_MAX;
   DBUG_ENTER("get_best_ror_intersect");
 
   if ((tree->n_ror_scans < 2) || !param->table->file->records)
     DBUG_RETURN(NULL);
 
   /*
-    Collect ROR-able SEL_ARGs and create ROR_SCAN_INFO for each of them.
-    Also find and save clustered PK scan if there is one.
+    Step1: Collect ROR-able SEL_ARGs and create ROR_SCAN_INFO for each of 
+    them. Also find and save clustered PK scan if there is one.
   */
   ROR_SCAN_INFO **cur_ror_scan;
   ROR_SCAN_INFO *cpk_scan= NULL;
@@ -2796,8 +2830,8 @@ TRP_ROR_INTERSECT *get_best_ror_intersect(const PARAM *param, SEL_TREE *tree,
                                           tree->ror_scans_end););
   /*
     Ok, [ror_scans, ror_scans_end) is array of ptrs to initialized
-    ROR_SCAN_INFOs.
-    Now, get a minimal key scan using an approximate algorithm.
+    ROR_SCAN_INFO's.
+    Step 2: Get best ROR-intersection using an approximate algorithm.
   */
   qsort(tree->ror_scans, tree->n_ror_scans, sizeof(ROR_SCAN_INFO*),
         (qsort_cmp)cmp_ror_scan_info);
@@ -2814,45 +2848,41 @@ TRP_ROR_INTERSECT *get_best_ror_intersect(const PARAM *param, SEL_TREE *tree,
   intersect_scans_end= intersect_scans;
 
   /* Create and incrementally update ROR intersection. */
-  ROR_INTERSECT_INFO *intersect;
-  if (!(intersect= ror_intersect_init(param, FALSE)))
+  ROR_INTERSECT_INFO *intersect, *intersect_best;
+  if (!(intersect= ror_intersect_init(param)) || 
+      !(intersect_best= ror_intersect_init(param)))
     return NULL;
 
-  /* [intersect_scans, intersect_scans_best) will hold the best combination */
+  /* [intersect_scans,intersect_scans_best) will hold the best intersection */
   ROR_SCAN_INFO **intersect_scans_best;
-  ha_rows       best_rows;
-  bool          is_best_covering;
-  double        best_index_scan_costs;
-  LINT_INIT(best_rows);       /* protected by intersect_scans_best */
-  LINT_INIT(is_best_covering);
-  LINT_INIT(best_index_scan_costs);
-
   cur_ror_scan= tree->ror_scans;
-  /* Start with one scan */
   intersect_scans_best= intersect_scans;
   while (cur_ror_scan != tree->ror_scans_end && !intersect->is_covering)
   {
-    /* S= S + first(R); */
-    if (ror_intersect_add(param, intersect, *cur_ror_scan))
-      *(intersect_scans_end++)= *cur_ror_scan;
-    /* R= R - first(R); */
-    cur_ror_scan++;
+    /* S= S + first(R);  R= R - first(R); */
+    if (!ror_intersect_add(intersect, *cur_ror_scan, false))
+    {
+      cur_ror_scan++;
+      continue;
+    }
+    
+    *(intersect_scans_end++)= *(cur_ror_scan++);
 
     if (intersect->total_cost < min_cost)
     {
       /* Local minimum found, save it */
-      min_cost= intersect->total_cost;
-      best_rows= (ha_rows)(intersect->records_fract*
-                           rows2double(param->table->file->records));
-      /* Prevent divisons by zero */
-      if (!best_rows)
-        best_rows= 1;
-      is_best_covering= intersect->is_covering;
+      ror_intersect_cpy(intersect_best, intersect);
       intersect_scans_best= intersect_scans_end;
-      best_index_scan_costs= intersect->index_scan_costs;
+      min_cost = intersect->total_cost;
     }
   }
 
+  if (intersect_scans_best == intersect_scans)
+  {
+    DBUG_PRINT("info", ("None of scans increase selectivity"));
+    DBUG_RETURN(NULL);
+  }
+    
   DBUG_EXECUTE("info",print_ror_scans_arr(param->table,
                                           "best ROR-intersection",
                                           intersect_scans,
@@ -2860,48 +2890,26 @@ TRP_ROR_INTERSECT *get_best_ror_intersect(const PARAM *param, SEL_TREE *tree,
 
   *are_all_covering= intersect->is_covering;
   uint best_num= intersect_scans_best - intersect_scans;
+  ror_intersect_cpy(intersect, intersect_best);
+
   /*
     Ok, found the best ROR-intersection of non-CPK key scans.
-    Check if we should add a CPK scan.
-
-    If the obtained ROR-intersection is covering, it doesn't make sense
-    to add CPK scan - Clustered PK contains all fields and if we're doing
-    CPK scan doing other CPK scans will only add more overhead.
+    Check if we should add a CPK scan. If the obtained ROR-intersection is 
+    covering, it doesn't make sense to add CPK scan.
   */
   if (cpk_scan && !intersect->is_covering)
   {
-    /*
-      Handle the special case: ROR-intersect(PRIMARY, key1) is
-      the best, but cost(range(key1)) > cost(best_non_ror_range_scan)
-    */
-    if (best_num == 0)
-    {
-      cur_ror_scan= tree->ror_scans;
-      intersect_scans_end= intersect_scans;
-      ror_intersect_reinit(intersect);
-      if (!ror_intersect_add(param, intersect, *cur_ror_scan))
-        DBUG_RETURN(NULL); /* shouldn't happen actually */
-      *(intersect_scans_end++)= *cur_ror_scan;
-      best_num++;
-    }
-
-    if (ror_intersect_add(param, intersect, cpk_scan))
+    if (ror_intersect_add(intersect, cpk_scan, TRUE) && 
+        (intersect->total_cost < min_cost))
     {
       cpk_scan_used= TRUE;
-      min_cost= intersect->total_cost;
-      best_rows= (ha_rows)(intersect->records_fract*
-                           rows2double(param->table->file->records));
-      /* Prevent divisons by zero */
-      if (!best_rows)
-        best_rows= 1;
-      is_best_covering= intersect->is_covering;
-      best_index_scan_costs= intersect->index_scan_costs;
+      intersect_best= intersect; //just set pointer here
     }
   }
 
   /* Ok, return ROR-intersect plan if we have found one */
   TRP_ROR_INTERSECT *trp= NULL;
-  if (best_num > 1 || cpk_scan_used)
+  if (min_cost < read_time && (cpk_scan_used || best_num > 1))
   {
     if (!(trp= new (param->mem_root) TRP_ROR_INTERSECT))
       DBUG_RETURN(trp);
@@ -2911,14 +2919,18 @@ TRP_ROR_INTERSECT *get_best_ror_intersect(const PARAM *param, SEL_TREE *tree,
       DBUG_RETURN(NULL);
     memcpy(trp->first_scan, intersect_scans, best_num*sizeof(ROR_SCAN_INFO*));
     trp->last_scan=  trp->first_scan + best_num;
-    trp->is_covering= is_best_covering;
-    trp->read_cost= min_cost;
+    trp->is_covering= intersect_best->is_covering;
+    trp->read_cost= intersect_best->total_cost;
+    /* Prevent divisons by zero */
+    ha_rows best_rows = double2rows(intersect_best->out_rows);
+    if (!best_rows)
+      best_rows= 1;
     trp->records= best_rows;
-    trp->index_scan_costs= best_index_scan_costs;
-    trp->cpk_scan= cpk_scan;
-    DBUG_PRINT("info",
-               ("Returning non-covering ROR-intersect plan: cost %g, records %lu",
-                trp->read_cost, (ulong) trp->records));
+    trp->index_scan_costs= intersect_best->index_scan_costs;
+    trp->cpk_scan= cpk_scan_used? cpk_scan: NULL;
+    DBUG_PRINT("info", ("Returning non-covering ROR-intersect plan:"
+                        "cost %g, records %lu",
+                        trp->read_cost, (ulong) trp->records));
   }
   DBUG_RETURN(trp);
 }
@@ -3145,8 +3157,9 @@ static TRP_RANGE *get_key_scans_params(PARAM *param, SEL_TREE *tree,
                                                        found_records) +
 			 cpu_cost;
 
-      DBUG_PRINT("info",("read_time: %g  found_read_time: %g",
-                         read_time, found_read_time));
+      DBUG_PRINT("info",("key %s: found_read_time: %g (cur. read_time: %g)",
+                         param->table->key_info[keynr].name, found_read_time,
+                         read_time));
 
       if (read_time > found_read_time && found_records != HA_POS_ERROR
           /*|| read_time == DBL_MAX*/ )
diff --git a/sql/procedure.h b/sql/procedure.h
index 4212a9246a5..33c1288c88e 100644
--- a/sql/procedure.h
+++ b/sql/procedure.h
@@ -59,10 +59,18 @@ public:
   void set(double nr) { value=nr; }
   void set(longlong nr) { value=(double) nr; }
   void set(const char *str,uint length,CHARSET_INFO *cs)
-  { int err; value=my_strntod(cs,(char*) str,length,(char**)0,&err); }
+  {
+    int err_not_used;
+    char *end_not_used;
+    value= my_strntod(cs,(char*) str,length, &end_not_used, &err_not_used);
+  }
   double val_real() { return value; }
   longlong val_int() { return (longlong) value; }
-  String *val_str(String *s) { s->set(value,decimals,default_charset()); return s; }
+  String *val_str(String *s)
+  {
+    s->set(value,decimals,default_charset());
+    return s;
+  }
   unsigned int size_of() { return sizeof(*this);}  
 };
 
@@ -98,10 +106,11 @@ public:
   { str_value.copy(str,length,cs); }
   double val_real()
   { 
-    int err;
+    int err_not_used;
+    char *end_not_used;
     CHARSET_INFO *cs=str_value.charset();
     return my_strntod(cs, (char*) str_value.ptr(), str_value.length(),
-		      (char**) 0, &err);
+		      &end_not_used, &err_not_used);
   }
   longlong val_int()
   { 
diff --git a/sql/slave.cc b/sql/slave.cc
index 0bcc1b7e852..5332dbf9c5b 100644
--- a/sql/slave.cc
+++ b/sql/slave.cc
@@ -1418,7 +1418,12 @@ not always make sense; please check the manual before using it).";
     values of these 2 are never used (new connections don't use them).
     We don't test equality of global collation_database either as it's is
     going to be deprecated (made read-only) in 4.1 very soon.
+    We don't do it for <3.23.57 because masters <3.23.50 hang on
+    SELECT @@unknown_var (BUG#7965 - see changelog of 3.23.50).
   */
+  if (strncmp(mi->rli.relay_log.description_event_for_queue->server_version,
+              "3.23.57",7) < 0)
+    goto err;
   if (!mysql_real_query(mysql, "SELECT @@GLOBAL.COLLATION_SERVER", 32) &&
       (master_res= mysql_store_result(mysql)))
   {
@@ -1455,6 +1460,7 @@ be equal for replication to work";
     mysql_free_result(master_res);
   }
 
+err:
   if (errmsg)
   {
     sql_print_error(errmsg);
diff --git a/sql/sql_base.cc b/sql/sql_base.cc
index 48855892567..d854956325e 100644
--- a/sql/sql_base.cc
+++ b/sql/sql_base.cc
@@ -1853,8 +1853,8 @@ int simple_open_n_lock_tables(THD *thd, TABLE_LIST *tables)
 
 bool open_and_lock_tables(THD *thd, TABLE_LIST *tables)
 {
-  DBUG_ENTER("open_and_lock_tables");
   uint counter;
+  DBUG_ENTER("open_and_lock_tables");
   if (open_tables(thd, tables, &counter) ||
       lock_tables(thd, tables, counter) ||
       mysql_handle_derived(thd->lex, &mysql_derived_prepare) ||
@@ -1867,6 +1867,36 @@ bool open_and_lock_tables(THD *thd, TABLE_LIST *tables)
 
 
 /*
+  Open all tables in list and process derived tables
+
+  SYNOPSIS
+    open_normal_and_derived_tables
+    thd		- thread handler
+    tables	- list of tables for open&locking
+
+  RETURN
+    FALSE - ok
+    TRUE  - error
+
+  NOTE 
+    This is to be used on prepare stage when you don't read any
+    data from the tables.
+*/
+
+bool open_normal_and_derived_tables(THD *thd, TABLE_LIST *tables)
+{
+  uint counter;
+  DBUG_ENTER("open_normal_and_derived_tables");
+  DBUG_ASSERT(!thd->fill_derived_tables());
+  if (open_tables(thd, tables, &counter) ||
+      mysql_handle_derived(thd->lex, &mysql_derived_prepare))
+    DBUG_RETURN(TRUE); /* purecov: inspected */
+  relink_tables_for_multidelete(thd);           // Not really needed, but
+  DBUG_RETURN(0);
+}
+
+
+/*
   Let us propagate pointers to open tables from global table list
   to table lists for multi-delete
 */
diff --git a/sql/sql_class.h b/sql/sql_class.h
index 033f5674d34..ca65f011b9d 100644
--- a/sql/sql_class.h
+++ b/sql/sql_class.h
@@ -205,6 +205,7 @@ public:
   inline bool is_open() { return log_type != LOG_CLOSED; }
   inline char* get_index_fname() { return index_file_name;}
   inline char* get_log_fname() { return log_file_name; }
+  inline char* get_name() { return name; }
   inline pthread_mutex_t* get_log_lock() { return &LOCK_log; }
   inline IO_CACHE* get_log_file() { return &log_file; }
 
diff --git a/sql/sql_lex.cc b/sql/sql_lex.cc
index 495fa4a0bd1..2f6e74225fd 100644
--- a/sql/sql_lex.cc
+++ b/sql/sql_lex.cc
@@ -944,6 +944,7 @@ int yylex(void *arg, void *yythd)
         if ((thd->client_capabilities & CLIENT_MULTI_STATEMENTS) && 
             (thd->command != COM_PREPARE))
         {
+	  lex->safe_to_cache_query= 0;
           lex->found_colon=    (char*) lex->ptr;
           thd->server_status|= SERVER_MORE_RESULTS_EXISTS;
           lex->next_state=     MY_LEX_END;
diff --git a/sql/sql_prepare.cc b/sql/sql_prepare.cc
index 9e4f6c1334c..a71b8148f8e 100644
--- a/sql/sql_prepare.cc
+++ b/sql/sql_prepare.cc
@@ -900,8 +900,12 @@ static bool mysql_test_insert(Prepared_statement *stmt,
   /*
      open temporary memory pool for temporary data allocated by derived
      tables & preparation procedure
+     Note that this is done without locks (should not be needed as we will not
+     access any data here)
+     If we would use locks, then we have to ensure we are not using
+     TL_WRITE_DELAYED as having two such locks can cause table corruption.
   */
-  if (open_and_lock_tables(thd, table_list))
+  if (open_normal_and_derived_tables(thd, table_list))
   {
     DBUG_RETURN(TRUE);
   }
diff --git a/sql/sql_select.cc b/sql/sql_select.cc
index b9c8f6c1237..94a2390324c 100644
--- a/sql/sql_select.cc
+++ b/sql/sql_select.cc
@@ -7823,12 +7823,17 @@ create_tmp_table(THD *thd,TMP_TABLE_PARAM *param,List<Item> &fields,
   if (temp_pool_slot != MY_BIT_NONE) // we got a slot
     sprintf(filename, "%s_%lx_%i", tmp_file_prefix,
             current_pid, temp_pool_slot);
-  else // if we run out of slots or we are not using tempool
+  else
+  {
+    /* if we run out of slots or we are not using tempool */
     sprintf(filename,"%s%lx_%lx_%x",tmp_file_prefix,current_pid,
             thd->thread_id, thd->tmp_table++);
+  }
 
-  if (lower_case_table_names)
-    my_casedn_str(files_charset_info, path);
+  /*
+    No need for change table name to lower case as we are only creating
+    MyISAM or HEAP tables here
+  */
   sprintf(path, "%s%s", mysql_tmpdir, filename);
 
   if (group)
@@ -10421,15 +10426,24 @@ test_if_skip_sort_order(JOIN_TAB *tab,ORDER *order,ha_rows select_limit,
 	}
 	else
 	{
-          select->quick->head->file->ha_index_end();
-          /* 
-            We have verified above that select->quick is not 
-            index_merge quick select. 
-          */
-	  select->quick->index= new_ref_key;
-	  select->quick->init();
+          /*
+            The range optimizer constructed QUICK_RANGE for ref_key, and
+            we want to use instead new_ref_key as the index. We can't
+            just change the index of the quick select, because this may
+            result in an incosistent QUICK_SELECT object. Below we
+            create a new QUICK_SELECT from scratch so that all its
+            parameres are set correctly by the range optimizer.
+           */
+          key_map new_ref_key_map;
+          new_ref_key_map.clear_all();  /* Force the creation of quick select */
+          new_ref_key_map.set_bit(new_ref_key); /* only for new_ref_key.      */
+
+          if (select->test_quick_select(tab->join->thd, new_ref_key_map, 0,
+                                        (tab->join->select_options & OPTION_FOUND_ROWS) ?
+                                        HA_POS_ERROR : tab->join->unit->select_limit_cnt) <= 0)
+            DBUG_RETURN(0);
 	}
-	ref_key= new_ref_key;
+        ref_key= new_ref_key;
       }
     }
     /* Check if we get the rows in requested sorted order by using the key */
diff --git a/sql/sql_show.cc b/sql/sql_show.cc
index 2aa8a67fbab..c7b4b61ca33 100644
--- a/sql/sql_show.cc
+++ b/sql/sql_show.cc
@@ -1534,25 +1534,6 @@ static bool show_status_array(THD *thd, const char *wild,
 }
 
 
-bool mysqld_show(THD *thd, const char *wild, show_var_st *variables,
-                 enum enum_var_type value_type,
-                 pthread_mutex_t *mutex,
-                 struct system_status_var *status_var, TABLE *table)
-{
-  DBUG_ENTER("mysqld_show");
-  ha_update_statistics();                    /* Export engines statistics */
-  pthread_mutex_lock(mutex);
-  if (show_status_array(thd, wild, variables, value_type, status_var, "", table))
-    goto err;
-  pthread_mutex_unlock(mutex);
-  DBUG_RETURN(FALSE);
-
- err:
-  pthread_mutex_unlock(mutex);
-  DBUG_RETURN(TRUE);
-}
-
-
 /* collect status for all running threads */
 
 void calc_sum_of_all_status(STATUS_VAR *to)
@@ -2874,10 +2855,13 @@ int fill_open_tables(THD *thd, TABLE_LIST *tables, COND *cond)
 int fill_variables(THD *thd, TABLE_LIST *tables, COND *cond)
 {
   DBUG_ENTER("fill_variables");
+  int res= 0;
   LEX *lex= thd->lex;
   const char *wild= lex->wild ? lex->wild->ptr() : NullS;
-  int res= mysqld_show(thd, wild, init_vars, lex->option_type,
-                       &LOCK_global_system_variables, 0, tables->table);
+  pthread_mutex_lock(&LOCK_global_system_variables);
+  res= show_status_array(thd, wild, init_vars, 
+                         lex->option_type, 0, "", tables->table);
+  pthread_mutex_unlock(&LOCK_global_system_variables);
   DBUG_RETURN(res);
 }
 
@@ -2889,17 +2873,14 @@ int fill_status(THD *thd, TABLE_LIST *tables, COND *cond)
   const char *wild= lex->wild ? lex->wild->ptr() : NullS;
   int res= 0;
   STATUS_VAR tmp;
-
+  ha_update_statistics();                    /* Export engines statistics */
+  pthread_mutex_lock(&LOCK_status);
   if (lex->option_type == OPT_GLOBAL)
-  {
-    pthread_mutex_lock(&LOCK_status);
     calc_sum_of_all_status(&tmp);
-  }
-  res= mysqld_show(thd, wild, status_vars, OPT_GLOBAL, &LOCK_status,
-                   (lex->option_type == OPT_GLOBAL ?
-                    &tmp: &thd->status_var), tables->table);
-  if (lex->option_type == OPT_GLOBAL)
-    pthread_mutex_unlock(&LOCK_status);
+  res= show_status_array(thd, wild, status_vars, OPT_GLOBAL,
+                         (lex->option_type == OPT_GLOBAL ? 
+                          &tmp: &thd->status_var), "",tables->table);
+  pthread_mutex_unlock(&LOCK_status);
   DBUG_RETURN(res);
 }
 
diff --git a/sql/sql_union.cc b/sql/sql_union.cc
index ce84be47243..0a91eb4c0e1 100644
--- a/sql/sql_union.cc
+++ b/sql/sql_union.cc
@@ -478,11 +478,14 @@ bool st_select_lex_unit::exec()
 	}
 	res= sl->join->error;
 	offset_limit_cnt= sl->offset_limit;
-	if (!res && union_result->flush())
+	if (!res)
 	{
-          examined_rows+= thd->examined_row_count;
-	  thd->lex->current_select= lex_select_save;
-	  DBUG_RETURN(TRUE);
+	  examined_rows+= thd->examined_row_count;
+	  if (union_result->flush())
+	  {
+	    thd->lex->current_select= lex_select_save;
+	    DBUG_RETURN(1);
+	  }
 	}
       }
       if (res)
diff --git a/sql/sql_view.cc b/sql/sql_view.cc
index 533876b6718..e2c4b1289fd 100644
--- a/sql/sql_view.cc
+++ b/sql/sql_view.cc
@@ -948,11 +948,12 @@ frm_type_enum mysql_frm_type(char *path)
   {
     DBUG_RETURN(FRMTYPE_ERROR);
   }
-  length= my_read(file, (byte*) header, 10, MYF(MY_WME));
+  length= my_read(file, (byte*) header, sizeof(header), MYF(MY_WME));
   my_close(file, MYF(MY_WME));
   if (length == (int) MY_FILE_ERROR)
     DBUG_RETURN(FRMTYPE_ERROR);
-  if (!strncmp(header, "TYPE=VIEW\n", 10))
+  if (length < (int) sizeof(header) ||
+      !strncmp(header, "TYPE=VIEW\n", sizeof(header)))
     DBUG_RETURN(FRMTYPE_VIEW);
   DBUG_RETURN(FRMTYPE_TABLE);                   // Is probably a .frm table
 }
diff --git a/sql/sql_yacc.yy b/sql/sql_yacc.yy
index 6b1456dfbd3..cc2443bd41b 100644
--- a/sql/sql_yacc.yy
+++ b/sql/sql_yacc.yy
@@ -2793,7 +2793,7 @@ type:
 	| YEAR_SYM opt_len field_options { $$=FIELD_TYPE_YEAR; }
 	| DATE_SYM			{ $$=FIELD_TYPE_DATE; }
 	| TIME_SYM			{ $$=FIELD_TYPE_TIME; }
-	| TIMESTAMP
+	| TIMESTAMP opt_len
 	  {
 	    if (YYTHD->variables.sql_mode & MODE_MAXDB)
 	      $$=FIELD_TYPE_DATETIME;
@@ -2806,13 +2806,6 @@ type:
 	      $$=FIELD_TYPE_TIMESTAMP;
             }
 	   }
-	| TIMESTAMP '(' NUM ')'
-          { 
-            LEX *lex= Lex;
-            lex->length= $3.str;
-            lex->type|= NOT_NULL_FLAG;
-            $$= FIELD_TYPE_TIMESTAMP;
-          }
 	| DATETIME			{ $$=FIELD_TYPE_DATETIME; }
 	| TINYBLOB			{ Lex->charset=&my_charset_bin;
 					  $$=FIELD_TYPE_TINY_BLOB; }
diff --git a/sql/table.cc b/sql/table.cc
index 97f17ea2417..a030da95db4 100644
--- a/sql/table.cc
+++ b/sql/table.cc
@@ -103,11 +103,11 @@ int openfrm(THD *thd, const char *name, const char *alias, uint db_stat,
 		    O_RDONLY | O_SHARE,
 		    MYF(0)))
       < 0)
-    goto err_w_init;
+    goto err;
 
   error= 4;
   if (my_read(file,(byte*) head,64,MYF(MY_NABP)))
-    goto err_w_init;
+    goto err;
 
   if (memcmp(head, "TYPE=", 5) == 0)
   {
@@ -116,9 +116,9 @@ int openfrm(THD *thd, const char *name, const char *alias, uint db_stat,
 
     if (db_stat & NO_ERR_ON_NEW_FRM)
       DBUG_RETURN(5);
-
+    file= -1;
     // caller can't process new .frm
-    goto err_w_init;
+    goto err;
   }
 
   share->blob_ptr_size= sizeof(char*);
@@ -131,21 +131,21 @@ int openfrm(THD *thd, const char *name, const char *alias, uint db_stat,
   share->path= strdup_root(&outparam->mem_root, name);
   outparam->alias= my_strdup(alias, MYF(MY_WME));
   if (!share->table_name || !share->path || !outparam->alias)
-    goto err_not_open;
+    goto err;
   *fn_ext(share->table_name)='\0';		// Remove extension
   *fn_ext(share->path)='\0';                    // Remove extension
 
   if (head[0] != (uchar) 254 || head[1] != 1 ||
       (head[2] != FRM_VER && head[2] != FRM_VER+1 &&
        ! (head[2] >= FRM_VER+3 && head[2] <= FRM_VER+4)))
-    goto err_not_open;				/* purecov: inspected */
+    goto err;                                   /* purecov: inspected */
   new_field_pack_flag=head[27];
   new_frm_ver= (head[2] - FRM_VER);
   field_pack_length= new_frm_ver < 2 ? 11 : 17;
 
   error=3;
   if (!(pos=get_form_pos(file,head,(TYPELIB*) 0)))
-    goto err_not_open;				/* purecov: inspected */
+    goto err;                                   /* purecov: inspected */
   *fn_ext(index_file)='\0';			// Remove .frm extension
 
   share->frm_version= head[2];
@@ -166,6 +166,14 @@ int openfrm(THD *thd, const char *name, const char *alias, uint db_stat,
   if (!share->table_charset)
   {
     /* unknown charset in head[38] or pre-3.23 frm */
+    if (use_mb(default_charset_info))
+    {
+      /* Warn that we may be changing the size of character columns */
+      sql_print_warning("'%s' had no or invalid character set, "
+                        "and default character set is multi-byte, "
+                        "so character column sizes may have changed",
+                        name);
+    }
     share->table_charset= default_charset_info;
   }
   share->db_record_offset= 1;
@@ -181,7 +189,7 @@ int openfrm(THD *thd, const char *name, const char *alias, uint db_stat,
   key_info_length= (uint) uint2korr(head+28);
   VOID(my_seek(file,(ulong) uint2korr(head+6),MY_SEEK_SET,MYF(0)));
   if (read_string(file,(gptr*) &disk_buff,key_info_length))
-    goto err_not_open; /* purecov: inspected */
+    goto err;                                   /* purecov: inspected */
   if (disk_buff[0] & 0x80)
   {
     share->keys=      keys=      (disk_buff[1] << 7) | (disk_buff[0] & 0x7f);
@@ -201,7 +209,7 @@ int openfrm(THD *thd, const char *name, const char *alias, uint db_stat,
   n_length=keys*sizeof(KEY)+key_parts*sizeof(KEY_PART_INFO);
   if (!(keyinfo = (KEY*) alloc_root(&outparam->mem_root,
 				    n_length+uint2korr(disk_buff+4))))
-    goto err_not_open; /* purecov: inspected */
+    goto err;                                   /* purecov: inspected */
   bzero((char*) keyinfo,n_length);
   outparam->key_info=keyinfo;
   key_part= my_reinterpret_cast(KEY_PART_INFO*) (keyinfo+keys);
@@ -210,7 +218,7 @@ int openfrm(THD *thd, const char *name, const char *alias, uint db_stat,
   ulong *rec_per_key;
   if (!(rec_per_key= (ulong*) alloc_root(&outparam->mem_root,
 					 sizeof(ulong*)*key_parts)))
-    goto err_not_open;
+    goto err;
 
   for (i=0 ; i < keys ; i++, keyinfo++)
   {
@@ -279,7 +287,7 @@ int openfrm(THD *thd, const char *name, const char *alias, uint db_stat,
 
   /* Allocate handler */
   if (!(outparam->file= get_new_handler(outparam, share->db_type)))
-    goto err_not_open;
+    goto err;
 
   error=4;
   outparam->reginfo.lock_type= TL_UNLOCK;
@@ -296,14 +304,14 @@ int openfrm(THD *thd, const char *name, const char *alias, uint db_stat,
   share->rec_buff_length= rec_buff_length;
   if (!(record= (char *) alloc_root(&outparam->mem_root,
                                     rec_buff_length * records)))
-    goto err_not_open;				/* purecov: inspected */
+    goto err;                                   /* purecov: inspected */
   share->default_values= record;
   if (my_pread(file,(byte*) record, (uint) share->reclength,
 	       (ulong) (uint2korr(head+6)+
                         ((uint2korr(head+14) == 0xffff ?
                             uint4korr(head+47) : uint2korr(head+14)))),
 	       MYF(MY_NABP)))
-    goto err_not_open; /* purecov: inspected */
+    goto err; /* purecov: inspected */
 
   if (records == 1)
   {
@@ -332,12 +340,13 @@ int openfrm(THD *thd, const char *name, const char *alias, uint db_stat,
   }
 #endif
   VOID(my_seek(file,pos,MY_SEEK_SET,MYF(0)));
-  if (my_read(file,(byte*) head,288,MYF(MY_NABP))) goto err_not_open;
+  if (my_read(file,(byte*) head,288,MYF(MY_NABP)))
+    goto err;
   if (crypted)
   {
     crypted->decode((char*) head+256,288-256);
     if (sint2korr(head+284) != 0)		// Should be 0
-      goto err_not_open;			// Wrong password
+      goto err;                                 // Wrong password
   }
 
   share->fields= uint2korr(head+258);
@@ -359,13 +368,13 @@ int openfrm(THD *thd, const char *name, const char *alias, uint db_stat,
 			   (share->fields+interval_parts+
 			    keys+3)*sizeof(my_string)+
 			   (n_length+int_length+com_length)))))
-    goto err_not_open; /* purecov: inspected */
+    goto err;                                   /* purecov: inspected */
 
   outparam->field=field_ptr;
   read_length=(uint) (share->fields * field_pack_length +
 		      pos+ (uint) (n_length+int_length+com_length));
   if (read_string(file,(gptr*) &disk_buff,read_length))
-    goto err_not_open; /* purecov: inspected */
+    goto err;                                   /* purecov: inspected */
   if (crypted)
   {
     crypted->decode((char*) disk_buff,read_length);
@@ -398,7 +407,7 @@ int openfrm(THD *thd, const char *name, const char *alias, uint db_stat,
       uint count= (uint) (interval->count + 1) * sizeof(uint);
       if (!(interval->type_lengths= (uint *) alloc_root(&outparam->mem_root,
                                                         count)))
-        goto err_not_open;
+        goto err;
       for (count= 0; count < interval->count; count++)
         interval->type_lengths[count]= strlen(interval->type_names[count]);
       interval->type_lengths[count]= 0;
@@ -459,7 +468,7 @@ int openfrm(THD *thd, const char *name, const char *alias, uint db_stat,
 	charset= &my_charset_bin;
 #else
 	error= 4;  // unsupported field type
-	goto err_not_open;
+	goto err;
 #endif
       }
       else
@@ -470,7 +479,7 @@ int openfrm(THD *thd, const char *name, const char *alias, uint db_stat,
         {
           error= 5; // Unknown or unavailable charset
           errarg= (int) strpos[14];
-          goto err_not_open;
+          goto err;
         }
       }
       if (!comment_length)
@@ -543,7 +552,7 @@ int openfrm(THD *thd, const char *name, const char *alias, uint db_stat,
     if (!reg_field)				// Not supported field type
     {
       error= 4;
-      goto err_not_open;			/* purecov: inspected */
+      goto err;			/* purecov: inspected */
     }
     reg_field->comment=comment;
     if (field_type == FIELD_TYPE_BIT)
@@ -616,7 +625,7 @@ int openfrm(THD *thd, const char *name, const char *alias, uint db_stat,
 						(uint) key_part->length);
 #ifdef EXTRA_DEBUG
 	if (key_part->fieldnr > share->fields)
-	  goto err_not_open; // sanity check
+	  goto err;                             // sanity check
 #endif
 	if (key_part->fieldnr)
 	{					// Should always be true !
@@ -767,7 +776,7 @@ int openfrm(THD *thd, const char *name, const char *alias, uint db_stat,
     if (!(share->blob_field= save=
 	  (uint*) alloc_root(&outparam->mem_root,
                              (uint) (share->blob_fields* sizeof(uint)))))
-      goto err_not_open;
+      goto err;
     for (i=0, ptr= outparam->field ; *ptr ; ptr++, i++)
     {
       if ((*ptr)->flags & BLOB_FLAG)
@@ -779,25 +788,25 @@ int openfrm(THD *thd, const char *name, const char *alias, uint db_stat,
   error=2;
   if (db_stat)
   {
-    int err;
+    int ha_err;
     unpack_filename(index_file,index_file);
-    if ((err=(outparam->file->
-	      ha_open(index_file,
-		      (db_stat & HA_READ_ONLY ? O_RDONLY : O_RDWR),
-		      (db_stat & HA_OPEN_TEMPORARY ? HA_OPEN_TMP_TABLE :
-		       ((db_stat & HA_WAIT_IF_LOCKED) ||
-			(specialflag & SPECIAL_WAIT_IF_LOCKED)) ?
-		       HA_OPEN_WAIT_IF_LOCKED :
-		       (db_stat & (HA_ABORT_IF_LOCKED | HA_GET_INFO)) ?
-		       HA_OPEN_ABORT_IF_LOCKED :
-		       HA_OPEN_IGNORE_IF_LOCKED) | ha_open_flags))))
+    if ((ha_err= (outparam->file->
+                  ha_open(index_file,
+                          (db_stat & HA_READ_ONLY ? O_RDONLY : O_RDWR),
+                          (db_stat & HA_OPEN_TEMPORARY ? HA_OPEN_TMP_TABLE :
+                           ((db_stat & HA_WAIT_IF_LOCKED) ||
+                            (specialflag & SPECIAL_WAIT_IF_LOCKED)) ?
+                           HA_OPEN_WAIT_IF_LOCKED :
+                           (db_stat & (HA_ABORT_IF_LOCKED | HA_GET_INFO)) ?
+                          HA_OPEN_ABORT_IF_LOCKED :
+                           HA_OPEN_IGNORE_IF_LOCKED) | ha_open_flags))))
     {
       /* Set a flag if the table is crashed and it can be auto. repaired */
-      share->crashed= ((err == HA_ERR_CRASHED_ON_USAGE) &&
+      share->crashed= ((ha_err == HA_ERR_CRASHED_ON_USAGE) &&
                        outparam->file->auto_repair() &&
                        !(ha_open_flags & HA_OPEN_FOR_REPAIR));
 
-      if (err == HA_ERR_NO_SUCH_TABLE)
+      if (ha_err == HA_ERR_NO_SUCH_TABLE)
       {
 	/* The table did not exists in storage engine, use same error message
 	   as if the .frm file didn't exist */
@@ -806,10 +815,10 @@ int openfrm(THD *thd, const char *name, const char *alias, uint db_stat,
       }
       else
       {
-        outparam->file->print_error(err, MYF(0));
+        outparam->file->print_error(ha_err, MYF(0));
         error_reported= TRUE;
       }
-      goto err_not_open; /* purecov: inspected */
+      goto err;                                 /* purecov: inspected */
     }
   }
   share->db_low_byte_first= outparam->file->low_byte_first();
@@ -822,15 +831,7 @@ int openfrm(THD *thd, const char *name, const char *alias, uint db_stat,
 #endif
   DBUG_RETURN (0);
 
- err_w_init:
-  /*
-    Avoid problem with uninitialized data
-    Note that we don't have to initialize outparam->s here becasue
-    the caller will check if the pointer exists in case of errors
-  */
-  bzero((char*) outparam,sizeof(*outparam));
-
- err_not_open:
+ err:
   x_free((gptr) disk_buff);
   if (file > 0)
     VOID(my_close(file,MYF(MY_WME)));
@@ -843,7 +844,7 @@ int openfrm(THD *thd, const char *name, const char *alias, uint db_stat,
   outparam->file=0;				// For easier errorchecking
   outparam->db_stat=0;
   hash_free(&share->name_hash);
-  free_root(&outparam->mem_root, MYF(0));
+  free_root(&outparam->mem_root, MYF(0));       // Safe to call on bzero'd root
   my_free((char*) outparam->alias, MYF(MY_ALLOW_ZERO_PTR));
   DBUG_RETURN (error);
 } /* openfrm */