Merge mskold@bk-internal.mysql.com:/home/bk/mysql-5.0-ndb

into  linux.site:/home/marty/MySQL/mysql-5.0

66 files changed, 1581 insertions, 1008 deletions

diff --git a/sql/Makefile.am b/sql/Makefile.am
index b506d2a767b..524ec93a6a9 100644
--- a/sql/Makefile.am
+++ b/sql/Makefile.am
@@ -111,8 +111,8 @@ DEFS =			-DMYSQL_SERVER \
 			-DDATADIR="\"$(MYSQLDATAdir)\"" \
 			-DSHAREDIR="\"$(MYSQLSHAREdir)\"" \
 			@DEFS@
-# Don't put lex_hash.h in BUILT_SOURCES as this will give infinite recursion
-BUILT_SOURCES =		sql_yacc.cc sql_yacc.h
+
+BUILT_SOURCES =		sql_yacc.cc sql_yacc.h lex_hash.h
 EXTRA_DIST =		udf_example.cc $(BUILT_SOURCES)
 DISTCLEANFILES =        lex_hash.h
 AM_YFLAGS =		-d
@@ -131,9 +131,6 @@ link_sources: mysql_tzinfo_to_sql.cc
 	rm -f my_time.c
 	@LN_CP_F@ ../sql-common/my_time.c my_time.c
 
-gen_lex_hash.o:		gen_lex_hash.cc lex.h
-			$(CXXCOMPILE) -c $(INCLUDES) $<
-
 mysql_tzinfo_to_sql.o:	$(mysql_tzinfo_to_sql_SOURCES)
 			$(CXXCOMPILE) -c $(INCLUDES) -DTZINFO2SQL $<
 
@@ -148,13 +145,9 @@ sql_yacc.o:	sql_yacc.cc sql_yacc.h $(HEADERS)
 		@echo "If it fails, re-run configure with --with-low-memory"
 		$(CXXCOMPILE) $(LM_CFLAGS) -c $<
 
-lex_hash.h:	lex.h gen_lex_hash.cc sql_yacc.h
-		$(MAKE) gen_lex_hash$(EXEEXT)
+lex_hash.h:	gen_lex_hash$(EXEEXT)
 		./gen_lex_hash$(EXEEXT) > $@
 
-# Hack to ensure that lex_hash.h is built early
-sql_lex.o:	lex_hash.h
-
 # For testing of udf_example.so;  Works on platforms with gcc
 # (This is not part of our build process but only provided as an example)
 udf_example.so:	udf_example.cc
diff --git a/sql/field.cc b/sql/field.cc
index 89ef25475ff..fb244de4275 100644
--- a/sql/field.cc
+++ b/sql/field.cc
@@ -38,7 +38,7 @@
   Instansiate templates and static variables
 *****************************************************************************/
 
-#ifdef __GNUC__
+#ifdef HAVE_EXPLICIT_TEMPLATE_INSTANTIATION
 template class List<create_field>;
 template class List_iterator<create_field>;
 #endif
@@ -1920,7 +1920,7 @@ int Field_decimal::store(const char *from, uint len, CHARSET_INFO *cs)
 	int_digits_added_zeros=0;
       }
     }
-    tmp_uint= (tmp_dec+(int_digits_end-int_digits_from)+
+    tmp_uint= (uint) (tmp_dec+(int_digits_end-int_digits_from)+
                (uint)(frac_digits_from-int_digits_tail_from)+
                int_digits_added_zeros);
   }
@@ -3317,12 +3317,12 @@ int Field_long::store(const char *from,uint len,CHARSET_INFO *cs)
   }
   if (error)
   {
-    error= 1;
+    error= error != MY_ERRNO_EDOM ? 1 : 2;
     set_warning(MYSQL_ERROR::WARN_LEVEL_WARN, ER_WARN_DATA_OUT_OF_RANGE, 1);
   }
   else if (from+len != end && table->in_use->count_cuted_fields &&
            check_int(from,len,end,cs))
-    error= 1;
+    error= 2;
     
   store_tmp= (long) tmp;
 #ifdef WORDS_BIGENDIAN
@@ -3584,7 +3584,7 @@ int Field_longlong::store(const char *from,uint len,CHARSET_INFO *cs)
   }
   else if (from+len != end && table->in_use->count_cuted_fields &&
            check_int(from,len,end,cs))
-    error= 1;
+    error= 2;
 #ifdef WORDS_BIGENDIAN
   if (table->s->db_low_byte_first)
   {
@@ -3806,7 +3806,7 @@ int Field_float::store(const char *from,uint len,CHARSET_INFO *cs)
   {
     set_warning(MYSQL_ERROR::WARN_LEVEL_WARN,
                 (error ? ER_WARN_DATA_OUT_OF_RANGE : WARN_DATA_TRUNCATED), 1);
-    error= 1;
+    error= error ? 1 : 2;
   }
   Field_float::store(nr);
   return error;
@@ -4093,7 +4093,7 @@ int Field_double::store(const char *from,uint len,CHARSET_INFO *cs)
   {
     set_warning(MYSQL_ERROR::WARN_LEVEL_WARN,
                 (error ? ER_WARN_DATA_OUT_OF_RANGE : WARN_DATA_TRUNCATED), 1);
-    error= 1;
+    error= error ? 1 : 2;
   }
   Field_double::store(nr);
   return error;
@@ -4495,6 +4495,8 @@ int Field_timestamp::store(const char *from,uint len,CHARSET_INFO *cs)
       error= 1;
     }
   }
+  if (error > 1)
+    error= 2;
 
 #ifdef WORDS_BIGENDIAN
   if (table->s->db_low_byte_first)
@@ -4791,7 +4793,7 @@ int Field_time::store(const char *from,uint len,CHARSET_INFO *cs)
   if (str_to_time(from, len, &ltime, &error))
   {
     tmp=0L;
-    error= 1;
+    error= 2;
     set_datetime_warning(MYSQL_ERROR::WARN_LEVEL_WARN, WARN_DATA_TRUNCATED,
                          from, len, MYSQL_TIMESTAMP_TIME, 1);
   }
@@ -4813,6 +4815,8 @@ int Field_time::store(const char *from,uint len,CHARSET_INFO *cs)
                            from, len, MYSQL_TIMESTAMP_TIME, !error);
       error= 1;
     }
+    if (error > 1)
+      error= 2;
   }
   
   if (ltime.neg)
@@ -5149,7 +5153,7 @@ int Field_date::store(const char *from, uint len,CHARSET_INFO *cs)
                       &error) <= MYSQL_TIMESTAMP_ERROR)
   {
     tmp=0;
-    error= 1;
+    error= 2;
   }
   else
     tmp=(uint32) l_time.year*10000L + (uint32) (l_time.month*100+l_time.day);
@@ -5353,7 +5357,7 @@ int Field_newdate::store(const char *from,uint len,CHARSET_INFO *cs)
                       &error) <= MYSQL_TIMESTAMP_ERROR)
   {
     tmp=0L;
-    error= 1;
+    error= 2;
   }
   else
     tmp= l_time.day + l_time.month*32 + l_time.year*16*32;
@@ -5836,7 +5840,7 @@ int Field_string::store(const char *from,uint length,CHARSET_INFO *cs)
     from= tmpstr.ptr();
     length=  tmpstr.length();
     if (conv_errors)
-      error= 1;
+      error= 2;
   }
 
   /* 
@@ -5860,7 +5864,7 @@ int Field_string::store(const char *from,uint length,CHARSET_INFO *cs)
     from+= field_charset->cset->scan(field_charset, from, end,
 				     MY_SEQ_SPACES);
     if (from != end)
-      error= 1;
+      error= 2;
   }
   if (error)
   {
@@ -5956,14 +5960,6 @@ longlong Field_string::val_int(void)
 }
 
 
-my_decimal *Field_longstr::val_decimal(my_decimal *decimal_value)
-{
-  str2my_decimal(E_DEC_FATAL_ERROR, ptr, field_length, charset(),
-                 decimal_value);
-  return decimal_value;
-}
-
-
 String *Field_string::val_str(String *val_buffer __attribute__((unused)),
 			      String *val_ptr)
 {
@@ -5975,6 +5971,14 @@ String *Field_string::val_str(String *val_buffer __attribute__((unused)),
 }
 
 
+my_decimal *Field_string::val_decimal(my_decimal *decimal_value)
+{
+  str2my_decimal(E_DEC_FATAL_ERROR, ptr, field_length, charset(),
+                 decimal_value);
+  return decimal_value;
+}
+
+
 int Field_string::cmp(const char *a_ptr, const char *b_ptr)
 {
   uint a_len, b_len;
@@ -6242,7 +6246,7 @@ int Field_varstring::store(const char *from,uint length,CHARSET_INFO *cs)
         table->in_use->abort_on_warning)
       error_code= ER_DATA_TOO_LONG;
     set_warning(level, error_code, 1);
-    return 1;
+    return 2;
   }
   return 0;
 }
@@ -6288,6 +6292,15 @@ String *Field_varstring::val_str(String *val_buffer __attribute__((unused)),
 }
 
 
+my_decimal *Field_varstring::val_decimal(my_decimal *decimal_value)
+{
+  uint length= length_bytes == 1 ? (uint) (uchar) *ptr : uint2korr(ptr);
+  str2my_decimal(E_DEC_FATAL_ERROR, ptr+length_bytes, length, charset(),
+                 decimal_value);
+  return decimal_value;
+}
+
+
 int Field_varstring::cmp(const char *a_ptr, const char *b_ptr)
 {
   uint a_length, b_length;
@@ -6813,7 +6826,7 @@ int Field_blob::store(const char *from,uint length,CHARSET_INFO *cs)
       from= tmpstr.ptr();
       length=  tmpstr.length();
       if (conv_errors)
-        error= 1;
+        error= 2;
     }
     
     copy_length= max_data_length();
@@ -6828,7 +6841,7 @@ int Field_blob::store(const char *from,uint length,CHARSET_INFO *cs)
                                                       copy_length,
                                                       &well_formed_error);
     if (copy_length < length)
-      error= 1;
+      error= 2;
     Field_blob::store_length(copy_length);
     if (was_conversion || table->copy_blobs || copy_length <= MAX_FIELD_WIDTH)
     {						// Must make a copy
@@ -6906,6 +6919,18 @@ String *Field_blob::val_str(String *val_buffer __attribute__((unused)),
 }
 
 
+my_decimal *Field_blob::val_decimal(my_decimal *decimal_value)
+{
+  char *blob;
+  memcpy_fixed(&blob, ptr+packlength, sizeof(char*));
+  if (!blob)
+    blob= "";
+  str2my_decimal(E_DEC_FATAL_ERROR, blob, get_length(ptr), charset(),
+                 decimal_value);
+  return decimal_value;
+}
+
+
 int Field_blob::cmp(const char *a,uint32 a_length, const char *b,
 		    uint32 b_length)
 {
@@ -7734,9 +7759,9 @@ bool Field_enum::eq_def(Field *field)
   for (uint i=0 ; i < from_lib->count ; i++)
     if (my_strnncoll(field_charset,
                      (const uchar*)typelib->type_names[i],
-                     strlen(typelib->type_names[i]),
+                     (uint) strlen(typelib->type_names[i]),
                      (const uchar*)from_lib->type_names[i],
-                     strlen(from_lib->type_names[i])))
+                     (uint) strlen(from_lib->type_names[i])))
       return 0;
   return 1;
 }
@@ -8594,7 +8619,8 @@ Field::set_datetime_warning(MYSQL_ERROR::enum_warning_level level, uint code,
   {
     char str_nr[22];
     char *str_end= longlong10_to_str(nr, str_nr, -10);
-    make_truncated_value_warning(table->in_use, str_nr, str_end - str_nr, 
+    make_truncated_value_warning(table->in_use, str_nr, 
+                                 (uint) (str_end - str_nr), 
                                  ts_type, field_name);
   }
 }
diff --git a/sql/field.h b/sql/field.h
index a522558a8d7..2b1229744c2 100644
--- a/sql/field.h
+++ b/sql/field.h
@@ -25,6 +25,7 @@
 #endif
 
 #define NOT_FIXED_DEC			31
+#define DATETIME_DEC                     6
 
 class Send_field;
 class Protocol;
@@ -381,7 +382,6 @@ public:
                field_name_arg, table_arg, charset)
     {}
 
-  my_decimal *val_decimal(my_decimal *);
   int store_decimal(const my_decimal *d);
 };
 
@@ -942,6 +942,7 @@ public:
   enum ha_base_keytype key_type() const { return HA_KEYTYPE_ULONGLONG; }
 #endif
   enum Item_result cmp_type () const { return INT_RESULT; }
+  uint decimals() const { return DATETIME_DEC; }
   int  store(const char *to,uint length,CHARSET_INFO *charset);
   int  store(double nr);
   int  store(longlong nr);
@@ -993,6 +994,7 @@ public:
   double val_real(void);
   longlong val_int(void);
   String *val_str(String*,String *);
+  my_decimal *val_decimal(my_decimal *);
   int cmp(const char *,const char*);
   void sort_string(char *buff,uint length);
   void sql_type(String &str) const;
@@ -1051,6 +1053,7 @@ public:
   double val_real(void);
   longlong val_int(void);
   String *val_str(String*,String *);
+  my_decimal *val_decimal(my_decimal *);
   int cmp(const char *,const char*);
   void sort_string(char *buff,uint length);
   void get_key_image(char *buff,uint length, imagetype type);
@@ -1106,6 +1109,7 @@ public:
   double val_real(void);
   longlong val_int(void);
   String *val_str(String*,String *);
+  my_decimal *val_decimal(my_decimal *);
   int cmp(const char *,const char*);
   int cmp(const char *a, uint32 a_length, const char *b, uint32 b_length);
   int cmp_binary(const char *a,const char *b, uint32 max_length=~0L);
@@ -1284,7 +1288,7 @@ public:
   enum_field_types type() const { return FIELD_TYPE_BIT; }
   enum ha_base_keytype key_type() const { return HA_KEYTYPE_BIT; }
   uint32 key_length() const { return (uint32) field_length + (bit_len > 0); }
-  uint32 max_length() { return (uint32) field_length + (bit_len > 0); }
+  uint32 max_length() { return (uint32) field_length * 8 + bit_len; }
   uint size_of() const { return sizeof(*this); }
   Item_result result_type () const { return INT_RESULT; }
   void reset(void) { bzero(ptr, field_length); }
@@ -1316,6 +1320,11 @@ public:
   Field *new_key_field(MEM_ROOT *root, struct st_table *new_table,
                        char *new_ptr, uchar *new_null_ptr,
                        uint new_null_bit);
+  void set_bit_ptr(uchar *bit_ptr_arg, uchar bit_ofs_arg)
+  {
+    bit_ptr= bit_ptr_arg;
+    bit_ofs= bit_ofs_arg;
+  }
 };
 
   
@@ -1327,6 +1336,7 @@ public:
                     enum utype unireg_check_arg, const char *field_name_arg,
                     struct st_table *table_arg);
   enum ha_base_keytype key_type() const { return HA_KEYTYPE_BINARY; }
+  uint32 max_length() { return (uint32) create_length; }
   uint size_of() const { return sizeof(*this); }
   int store(const char *to, uint length, CHARSET_INFO *charset);
   int store(double nr) { return Field_bit::store(nr); }
diff --git a/sql/filesort.cc b/sql/filesort.cc
index 30ebd8d59e1..75da43afed5 100644
--- a/sql/filesort.cc
+++ b/sql/filesort.cc
@@ -631,7 +631,7 @@ static void make_sortkey(register SORTPARAM *param,
 	    *to++=1;
 	  /* All item->str() to use some extra byte for end null.. */
 	  String tmp((char*) to,sort_field->length+4,cs);
-	  String *res=item->val_str(&tmp);
+          String *res= item->str_result(&tmp);
 	  if (!res)
 	  {
 	    if (maybe_null)
@@ -673,8 +673,8 @@ static void make_sortkey(register SORTPARAM *param,
 	}
       case INT_RESULT:
 	{
-	  longlong value=item->val_int();
-	  if (maybe_null)
+          longlong value= item->val_int_result();
+          if (maybe_null)
           {
 	    *to++=1;				/* purecov: inspected */
             if (item->null_value)
@@ -715,7 +715,7 @@ static void make_sortkey(register SORTPARAM *param,
 	}
       case DECIMAL_RESULT:
         {
-          my_decimal dec_buf, *dec_val= item->val_decimal(&dec_buf);
+          my_decimal dec_buf, *dec_val= item->val_decimal_result(&dec_buf);
           if (maybe_null)
           {
             if (item->null_value)
@@ -733,7 +733,7 @@ static void make_sortkey(register SORTPARAM *param,
         }
       case REAL_RESULT:
 	{
-	  double value= item->val_real();
+          double value= item->val_result();
 	  if (maybe_null)
           {
             if (item->null_value)
diff --git a/sql/gstream.cc b/sql/gstream.cc
index f7d11d76b0c..4083cb2fe71 100644
--- a/sql/gstream.cc
+++ b/sql/gstream.cc
@@ -83,7 +83,7 @@ bool Gis_read_stream::get_next_number(double *d)
   }
 
   *d = my_strntod(m_charset, (char *)m_cur,
-		  m_limit-m_cur, &endptr, &err);
+		  (uint) (m_limit-m_cur), &endptr, &err);
   if (err)
     return 1;
   if (endptr)
@@ -115,6 +115,6 @@ bool Gis_read_stream::check_next_symbol(char symbol)
 void Gis_read_stream::set_error_msg(const char *msg)
 {
   size_t len= strlen(msg);			// ok in this context
-  m_err_msg= (char *) my_realloc(m_err_msg, len + 1, MYF(MY_ALLOW_ZERO_PTR));
+  m_err_msg= (char *) my_realloc(m_err_msg, (uint) len + 1, MYF(MY_ALLOW_ZERO_PTR));
   memcpy(m_err_msg, msg, len + 1);
 }
diff --git a/sql/ha_federated.cc b/sql/ha_federated.cc
index 89210a2f3cd..77db17608bc 100644
--- a/sql/ha_federated.cc
+++ b/sql/ha_federated.cc
@@ -462,6 +462,12 @@ static int check_foreign_data_source(FEDERATED_SHARE *share)
   }
   else
   {
+    /*
+      Since we do not support transactions at this version, we can let the client
+      API silently reconnect. For future versions, we will need more logic to deal
+      with transactions
+    */
+    mysql->reconnect= 1;
     /* 
       Note: I am not using INORMATION_SCHEMA because this needs to work with < 5.0
       if we can connect, then make sure the table exists 
@@ -988,6 +994,12 @@ int ha_federated::open(const char *name, int mode, uint test_if_locked)
     my_error(ER_CONNECT_TO_MASTER, MYF(0), mysql_error(mysql));
     DBUG_RETURN(ER_CONNECT_TO_MASTER);
   }
+  /*
+    Since we do not support transactions at this version, we can let the client
+    API silently reconnect. For future versions, we will need more logic to deal
+    with transactions
+  */
+  mysql->reconnect= 1;
   DBUG_RETURN(0);
 }
 
diff --git a/sql/ha_heap.cc b/sql/ha_heap.cc
index cd655eeb0a9..6e609a94be3 100644
--- a/sql/ha_heap.cc
+++ b/sql/ha_heap.cc
@@ -65,7 +65,15 @@ int ha_heap::open(const char *name, int mode, uint test_if_locked)
   {
     /* Initialize variables for the opened table */
     set_keys_for_scanning();
-    update_key_stats();
+    /*
+      We cannot run update_key_stats() here because we do not have a
+      lock on the table. The 'records' count might just be changed
+      temporarily at this moment and we might get wrong statistics (Bug
+      #10178). Instead we request for update. This will be done in
+      ha_heap::info(), which is always called before key statistics are
+      used.
+    */
+    key_stats_ok= FALSE;
   }
   return (file ? 0 : 1);
 }
@@ -118,6 +126,8 @@ void ha_heap::update_key_stats()
     }
   }
   records_changed= 0;
+  /* At the end of update_key_stats() we can proudly claim they are OK. */
+  key_stats_ok= TRUE;
 }
 
 
@@ -132,7 +142,7 @@ int ha_heap::write_row(byte * buf)
   res= heap_write(file,buf);
   if (!res && (++records_changed*HEAP_STATS_UPDATE_THRESHOLD > 
                file->s->records))
-    update_key_stats();
+    key_stats_ok= FALSE;
   return res;
 }
 
@@ -145,7 +155,7 @@ int ha_heap::update_row(const byte * old_data, byte * new_data)
   res= heap_update(file,old_data,new_data);
   if (!res && ++records_changed*HEAP_STATS_UPDATE_THRESHOLD > 
               file->s->records)
-    update_key_stats();
+    key_stats_ok= FALSE;
   return res;
 }
 
@@ -156,7 +166,7 @@ int ha_heap::delete_row(const byte * buf)
   res= heap_delete(file,buf);
   if (!res && table->s->tmp_table == NO_TMP_TABLE && 
       ++records_changed*HEAP_STATS_UPDATE_THRESHOLD > file->s->records)
-    update_key_stats();
+    key_stats_ok= FALSE;
   return res;
 }
 
@@ -278,6 +288,13 @@ void ha_heap::info(uint flag)
   delete_length= info.deleted * info.reclength;
   if (flag & HA_STATUS_AUTO)
     auto_increment_value= info.auto_increment;
+  /*
+    If info() is called for the first time after open(), we will still
+    have to update the key statistics. Hoping that a table lock is now
+    in place.
+  */
+  if (! key_stats_ok)
+    update_key_stats();
 }
 
 int ha_heap::extra(enum ha_extra_function operation)
@@ -289,7 +306,7 @@ int ha_heap::delete_all_rows()
 {
   heap_clear(file);
   if (table->s->tmp_table == NO_TMP_TABLE)
-    update_key_stats();
+    key_stats_ok= FALSE;
   return 0;
 }
 
@@ -448,6 +465,9 @@ ha_rows ha_heap::records_in_range(uint inx, key_range *min_key,
       min_key->flag != HA_READ_KEY_EXACT ||
       max_key->flag != HA_READ_AFTER_KEY)
     return HA_POS_ERROR;			// Can only use exact keys
+
+  /* Assert that info() did run. We need current statistics here. */
+  DBUG_ASSERT(key_stats_ok);
   return key->rec_per_key[key->key_parts-1];
 }
 
diff --git a/sql/ha_heap.h b/sql/ha_heap.h
index 2aa065e0d96..7a97c727049 100644
--- a/sql/ha_heap.h
+++ b/sql/ha_heap.h
@@ -29,8 +29,10 @@ class ha_heap: public handler
   key_map btree_keys;
   /* number of records changed since last statistics update */
   uint    records_changed;
+  bool    key_stats_ok;
 public:
-  ha_heap(TABLE *table): handler(table), file(0), records_changed(0) {}
+  ha_heap(TABLE *table): handler(table), file(0), records_changed(0),
+      key_stats_ok(0) {}
   ~ha_heap() {}
   const char *table_type() const
   {
diff --git a/sql/ha_innodb.cc b/sql/ha_innodb.cc
index 4ed005874f1..db354066849 100644
--- a/sql/ha_innodb.cc
+++ b/sql/ha_innodb.cc
@@ -146,8 +146,7 @@ long innobase_mirrored_log_groups, innobase_log_files_in_group,
      innobase_buffer_pool_awe_mem_mb,
      innobase_buffer_pool_size, innobase_additional_mem_pool_size,
      innobase_file_io_threads,  innobase_lock_wait_timeout,
-     innobase_thread_concurrency, innobase_force_recovery,
-     innobase_open_files;
+     innobase_force_recovery, innobase_open_files;
 
 /* The default values for the following char* start-up parameters
 are determined in innobase_init below: */
@@ -327,7 +326,7 @@ innodb_srv_conc_enter_innodb(
 /*=========================*/
 	trx_t*	trx)	/* in: transaction handle */
 {
-	if (srv_thread_concurrency >= 500) {
+	if (UNIV_LIKELY(srv_thread_concurrency >= 20)) {
 
 		return;
 	}
@@ -344,7 +343,7 @@ innodb_srv_conc_exit_innodb(
 /*========================*/
 	trx_t*	trx)	/* in: transaction handle */
 {
-	if (srv_thread_concurrency >= 500) {
+	if (UNIV_LIKELY(srv_thread_concurrency >= 20)) {
 
 		return;
 	}
@@ -564,7 +563,7 @@ innobase_mysql_print_thd(
         thd = (const THD*) input_thd;
 
   	fprintf(f, "MySQL thread id %lu, query id %lu",
-		thd->thread_id, thd->query_id);
+		thd->thread_id, (ulong) thd->query_id);
 	if (thd->host) {
 		putc(' ', f);
 		fputs(thd->host, f);
@@ -1041,7 +1040,19 @@ mysql_get_identifier_quote_char(
 		return(EOF);
 	}
 	return(get_quote_char_for_identifier((THD*) trx->mysql_thd,
-						name, namelen));
+						name, (int) namelen));
+}
+
+/**************************************************************************
+Determines if the currently running transaction has been interrupted. */
+extern "C"
+ibool
+trx_is_interrupted(
+/*===============*/
+			/* out: TRUE if interrupted */
+	trx_t*	trx)	/* in: transaction */
+{
+	return(trx && trx->mysql_thd && ((THD*) trx->mysql_thd)->killed);
 }
 
 /**************************************************************************
@@ -1302,8 +1313,8 @@ innobase_init(void)
 
 	data_mysql_default_charset_coll = (ulint)default_charset_info->number;
 
-	data_mysql_latin1_swedish_charset_coll =
-					(ulint)my_charset_latin1.number;
+	ut_a(DATA_MYSQL_LATIN1_SWEDISH_CHARSET_COLL ==
+					my_charset_latin1.number);
 
 	/* Store the latin1_swedish_ci character ordering table to InnoDB. For
 	non-latin1_swedish_ci charsets we use the MySQL comparison functions,
@@ -1794,7 +1805,7 @@ try_again:
                         fprintf(stderr,
 "InnoDB: This transaction needs it to be sent up to\n"
 "InnoDB: file %s, position %lu\n", trx->repl_wait_binlog_name,
-                                        (uint)trx->repl_wait_binlog_pos);
+                                        (ulong)trx->repl_wait_binlog_pos);
 
                         innobase_repl_state = 0;
 
@@ -2022,7 +2033,7 @@ innobase_rollback_to_savepoint(
 
         longlong2str((ulonglong)savepoint, name, 36);
 
-        error = trx_rollback_to_savepoint_for_mysql(trx, name,
+        error = (int) trx_rollback_to_savepoint_for_mysql(trx, name,
 						&mysql_binlog_cache_pos);
 	DBUG_RETURN(convert_error_code_to_mysql(error, NULL));
 }
@@ -2051,7 +2062,7 @@ innobase_release_savepoint(
 
         longlong2str((ulonglong)savepoint, name, 36);
 
-	error = trx_release_savepoint_for_mysql(trx, name);
+	error = (int) trx_release_savepoint_for_mysql(trx, name);
 
 	DBUG_RETURN(convert_error_code_to_mysql(error, NULL));
 }
@@ -2092,7 +2103,7 @@ innobase_savepoint(
         char name[64];
         longlong2str((ulonglong)savepoint,name,36);
 
-        error = trx_savepoint_for_mysql(trx, name, (ib_longlong)0);
+        error = (int) trx_savepoint_for_mysql(trx, name, (ib_longlong)0);
 
 	DBUG_RETURN(convert_error_code_to_mysql(error, NULL));
 }
@@ -2663,7 +2674,7 @@ innobase_read_from_2_little_endian(
 			/* out: value */
 	const mysql_byte*	buf)	/* in: from where to read */
 {
-	return((ulint)(buf[0]) + 256 * ((ulint)(buf[1])));
+	return (uint) ((ulint)(buf[0]) + 256 * ((ulint)(buf[1])));
 }
 
 /***********************************************************************
@@ -2867,6 +2878,8 @@ build_template(
 	ibool		fetch_all_in_key	= FALSE;
 	ibool		fetch_primary_key_cols	= FALSE;
 	ulint		i;
+	/* byte offset of the end of last requested column */
+	ulint		mysql_prefix_len	= 0;
 
 	if (prebuilt->select_lock_type == LOCK_X) {
 		/* We always retrieve the whole clustered index record if we
@@ -2987,6 +3000,11 @@ build_template(
 					get_field_offset(table, field);
 
 		templ->mysql_col_len = (ulint) field->pack_length();
+		if (mysql_prefix_len < templ->mysql_col_offset
+				+ templ->mysql_col_len) {
+			mysql_prefix_len = templ->mysql_col_offset
+				+ templ->mysql_col_len;
+		}
 		templ->type = index->table->cols[i].type.mtype;
 		templ->mysql_type = (ulint)field->type();
 
@@ -3009,6 +3027,7 @@ skip_field:
 	}
 
 	prebuilt->n_template = n_requested_fields;
+	prebuilt->mysql_prefix_len = mysql_prefix_len;
 
 	if (index != clust_index && prebuilt->need_to_access_clustered) {
 		/* Change rec_field_no's to correspond to the clustered index
@@ -3080,7 +3099,7 @@ ha_innobase::write_row(
 		being blocked by a MySQL table lock TL_WRITE_ALLOW_READ. */
 
 		dict_table_t*	src_table;
-		ibool		mode;
+		ulint		mode;
 
 		num_write_row = 0;
 
@@ -3743,7 +3762,7 @@ ha_innobase::index_read(
 		match_mode = ROW_SEL_EXACT_PREFIX;
 	}
 
-	last_match_mode = match_mode;
+	last_match_mode = (uint) match_mode;
 
 	innodb_srv_conc_enter_innodb(prebuilt->trx);
 
@@ -3763,7 +3782,7 @@ ha_innobase::index_read(
 		error = HA_ERR_KEY_NOT_FOUND;
 		table->status = STATUS_NOT_FOUND;
 	} else {
-		error = convert_error_code_to_mysql(ret, user_thd);
+		error = convert_error_code_to_mysql((int) ret, user_thd);
 		table->status = STATUS_NOT_FOUND;
 	}
 
@@ -3915,7 +3934,7 @@ ha_innobase::general_fetch(
 		error = HA_ERR_END_OF_FILE;
 		table->status = STATUS_NOT_FOUND;
 	} else {
-		error = convert_error_code_to_mysql(ret, user_thd);
+		error = convert_error_code_to_mysql((int) ret, user_thd);
 		table->status = STATUS_NOT_FOUND;
 	}
 
@@ -4864,7 +4883,7 @@ innobase_drop_database(
 	}
 
 	ptr++;
-	namebuf = my_malloc(len + 2, MYF(0));
+	namebuf = my_malloc((uint) len + 2, MYF(0));
 
 	memcpy(namebuf, ptr, len);
 	namebuf[len] = '/';
@@ -5430,7 +5449,7 @@ ha_innobase::update_table_comment(
 				info on foreign keys */
         const char*	comment)/* in: table comment defined by user */
 {
-	uint	length			= strlen(comment);
+	uint	length			= (uint) strlen(comment);
 	char*				str;
 	row_prebuilt_t*	prebuilt	= (row_prebuilt_t*)innobase_prebuilt;
 
@@ -5482,7 +5501,7 @@ ha_innobase::update_table_comment(
 				*pos++ = ' ';
 			}
 			rewind(file);
-			flen = fread(pos, 1, flen, file);
+			flen = (uint) fread(pos, 1, flen, file);
 			pos[flen] = 0;
 		}
 
@@ -5545,7 +5564,7 @@ ha_innobase::get_foreign_key_create_info(void)
 
 		if (str) {
 			rewind(file);
-			flen = fread(str, 1, flen, file);
+			flen = (uint) fread(str, 1, flen, file);
 			str[flen] = 0;
 		}
 
@@ -5585,8 +5604,8 @@ ha_innobase::get_foreign_key_list(THD *thd, List<FOREIGN_KEY_INFO> *f_key_list)
     while (tmp_buff[i] != '/')
       i++;
     tmp_buff+= i + 1;
-    f_key_info.forein_id= make_lex_string(thd, 0,
-                                          tmp_buff, strlen(tmp_buff), 1);
+    f_key_info.forein_id= make_lex_string(thd, 0, tmp_buff,
+                                          (uint) strlen(tmp_buff), 1);
     tmp_buff= foreign->referenced_table_name;
     i= 0;
     while (tmp_buff[i] != '/')
@@ -5594,16 +5613,16 @@ ha_innobase::get_foreign_key_list(THD *thd, List<FOREIGN_KEY_INFO> *f_key_list)
     f_key_info.referenced_db= make_lex_string(thd, 0,
                                               tmp_buff, i, 1);
     tmp_buff+= i + 1;
-    f_key_info.referenced_table= make_lex_string(thd, 0,
-                                               tmp_buff, strlen(tmp_buff), 1);
+    f_key_info.referenced_table= make_lex_string(thd, 0, tmp_buff, 
+                                               (uint) strlen(tmp_buff), 1);
 
     for (i= 0;;)
     {
       tmp_buff= foreign->foreign_col_names[i];
-      name= make_lex_string(thd, name, tmp_buff, strlen(tmp_buff), 1);
+      name= make_lex_string(thd, name, tmp_buff, (uint) strlen(tmp_buff), 1);
       f_key_info.foreign_fields.push_back(name);
       tmp_buff= foreign->referenced_col_names[i];
-      name= make_lex_string(thd, name, tmp_buff, strlen(tmp_buff), 1);
+      name= make_lex_string(thd, name, tmp_buff, (uint) strlen(tmp_buff), 1);
       f_key_info.referenced_fields.push_back(name);
       if (++i >= foreign->n_fields)
         break;
@@ -5991,12 +6010,12 @@ ha_innobase::external_lock(
 
 				ulint	error;
 				error = row_lock_table_for_mysql(prebuilt,
-							NULL, LOCK_TABLE_EXP);
+							NULL, 0);
 
 				if (error != DB_SUCCESS) {
 					error = convert_error_code_to_mysql(
-						error, user_thd);
-					DBUG_RETURN(error);
+						(int) error, user_thd);
+					DBUG_RETURN((int) error);
 				}
 			}
 
@@ -6011,9 +6030,6 @@ ha_innobase::external_lock(
 	trx->n_mysql_tables_in_use--;
 	prebuilt->mysql_has_locked = FALSE;
 
-	if (trx->n_lock_table_exp) {
-		row_unlock_tables_for_mysql(trx);
-	}
 
 	/* If the MySQL lock count drops to zero we know that the current SQL
 	statement has ended */
@@ -6055,7 +6071,7 @@ user issued query LOCK TABLES..WHERE ENGINE = InnoDB. */
 int
 ha_innobase::transactional_table_lock(
 /*==================================*/
-			        /* out: 0 */
+			        /* out: error code */
 	THD*	thd,		/* in: handle to the user thread */
 	int 	lock_type)	/* in: lock type */
 {
@@ -6119,12 +6135,11 @@ ha_innobase::transactional_table_lock(
 	if (thd->in_lock_tables && thd->variables.innodb_table_locks) {
 		ulint	error = DB_SUCCESS;
 
-		error = row_lock_table_for_mysql(prebuilt,NULL,
-						LOCK_TABLE_TRANSACTIONAL);
+		error = row_lock_table_for_mysql(prebuilt, NULL, 0);
 
 		if (error != DB_SUCCESS) {
-			error = convert_error_code_to_mysql(error, user_thd);
-			DBUG_RETURN(error);
+			error = convert_error_code_to_mysql((int) error, user_thd);
+			DBUG_RETURN((int) error);
 		}
 
 		if (thd->options & (OPTION_NOT_AUTOCOMMIT | OPTION_BEGIN)) {
@@ -6214,22 +6229,22 @@ innodb_show_status(
 	rewind(srv_monitor_file);
 	if (flen < MAX_STATUS_SIZE) {
 		/* Display the entire output. */
-		flen = fread(str, 1, flen, srv_monitor_file);
+		flen = (long) fread(str, 1, flen, srv_monitor_file);
 	} else if (trx_list_end < (ulint) flen
 			&& trx_list_start < trx_list_end
 			&& trx_list_start + (flen - trx_list_end)
 			< MAX_STATUS_SIZE - sizeof truncated_msg - 1) {
 		/* Omit the beginning of the list of active transactions. */
-		long	len = fread(str, 1, trx_list_start, srv_monitor_file);
+		long len = (long) fread(str, 1, trx_list_start, srv_monitor_file);
 		memcpy(str + len, truncated_msg, sizeof truncated_msg - 1);
 		len += sizeof truncated_msg - 1;
 		usable_len = (MAX_STATUS_SIZE - 1) - len;
 		fseek(srv_monitor_file, flen - usable_len, SEEK_SET);
-		len += fread(str + len, 1, usable_len, srv_monitor_file);
+		len += (long) fread(str + len, 1, usable_len, srv_monitor_file);
 		flen = len;
 	} else {
 		/* Omit the end of the output. */
-		flen = fread(str, 1, MAX_STATUS_SIZE - 1, srv_monitor_file);
+		flen = (long) fread(str, 1, MAX_STATUS_SIZE - 1, srv_monitor_file);
 	}
 
 	mutex_exit_noninline(&srv_monitor_file_mutex);
@@ -6791,7 +6806,7 @@ innobase_get_at_most_n_mbchars(
 	ulint n_chars;		/* number of characters in prefix */
 	CHARSET_INFO* charset;	/* charset used in the field */
 
-	charset = get_charset(charset_id, MYF(MY_WME));
+	charset = get_charset((uint) charset_id, MYF(MY_WME));
 
 	ut_ad(charset);
 	ut_ad(charset->mbmaxlen);
@@ -6825,7 +6840,7 @@ innobase_get_at_most_n_mbchars(
 		whole string. */
 
 		char_length = my_charpos(charset, str,
-						str + data_len, n_chars);
+						str + data_len, (int) n_chars);
 		if (char_length > data_len) {
 			char_length = data_len;
 		}		
@@ -6948,7 +6963,7 @@ innobase_xa_prepare(
 
                 ut_ad(trx->active_trans);
 
-		error = trx_prepare_for_mysql(trx);
+		error = (int) trx_prepare_for_mysql(trx);
 	} else {
 	        /* We just mark the SQL statement ended and do not do a
 		transaction prepare */
diff --git a/sql/ha_innodb.h b/sql/ha_innodb.h
index 4c0f5209af9..90cae3998ed 100644
--- a/sql/ha_innodb.h
+++ b/sql/ha_innodb.h
@@ -218,7 +218,7 @@ extern long innobase_log_file_size, innobase_log_buffer_size;
 extern long innobase_buffer_pool_size, innobase_additional_mem_pool_size;
 extern long innobase_buffer_pool_awe_mem_mb;
 extern long innobase_file_io_threads, innobase_lock_wait_timeout;
-extern long innobase_force_recovery, innobase_thread_concurrency;
+extern long innobase_force_recovery;
 extern long innobase_open_files;
 extern char *innobase_data_home_dir, *innobase_data_file_path;
 extern char *innobase_log_group_home_dir, *innobase_log_arch_dir;
diff --git a/sql/ha_myisam.cc b/sql/ha_myisam.cc
index 1352fd84d9d..0d9c32adbfa 100644
--- a/sql/ha_myisam.cc
+++ b/sql/ha_myisam.cc
@@ -1068,7 +1068,7 @@ bool ha_myisam::check_and_repair(THD *thd)
   old_query_length= thd->query_length;
   pthread_mutex_lock(&LOCK_thread_count);
   thd->query= (char*) table->s->table_name;
-  thd->query_length= strlen(table->s->table_name);
+  thd->query_length= (uint32) strlen(table->s->table_name);
   pthread_mutex_unlock(&LOCK_thread_count);
 
   if ((marked_crashed= mi_is_crashed(file)) || check(thd, &check_opt))
diff --git a/sql/ha_myisammrg.cc b/sql/ha_myisammrg.cc
index 0796ded3ac0..5d3f379081c 100644
--- a/sql/ha_myisammrg.cc
+++ b/sql/ha_myisammrg.cc
@@ -466,7 +466,7 @@ void ha_myisammrg::append_create_info(String *packet)
   MYRG_TABLE *open_table,*first;
 
   current_db= table->s->db;
-  db_length=  strlen(current_db);
+  db_length= (uint) strlen(current_db);
 
   for (first=open_table=file->open_tables ;
        open_table != file->end_table ;
diff --git a/sql/ha_ndbcluster.cc b/sql/ha_ndbcluster.cc
index da9f019fb35..1ce0ede7164 100644
--- a/sql/ha_ndbcluster.cc
+++ b/sql/ha_ndbcluster.cc
@@ -511,8 +511,13 @@ int ha_ndbcluster::ndb_err(NdbTransaction *trans)
   DBUG_PRINT("info", ("transformed ndbcluster error %d to mysql error %d", 
                       err.code, res));
   if (res == HA_ERR_FOUND_DUPP_KEY)
-    m_dupkey= table->s->primary_key;
-  
+  {
+    if (m_rows_to_insert == 1)
+      m_dupkey= table->s->primary_key;
+    else
+      // We are batching inserts, offending key is not available
+      m_dupkey= (uint) -1;
+  }
   DBUG_RETURN(res);
 }
 
@@ -540,13 +545,12 @@ bool ha_ndbcluster::get_error_message(int error,
 }
 
 
+#ifndef DBUG_OFF
 /*
   Check if type is supported by NDB.
-  TODO Use this once in open(), not in every operation
-
 */
 
-static inline bool ndb_supported_type(enum_field_types type)
+static bool ndb_supported_type(enum_field_types type)
 {
   switch (type) {
   case MYSQL_TYPE_TINY:        
@@ -581,6 +585,7 @@ static inline bool ndb_supported_type(enum_field_types type)
   }
   return FALSE;
 }
+#endif /* !DBUG_OFF */
 
 
 /*
@@ -610,15 +615,10 @@ int ha_ndbcluster::set_ndb_key(NdbOperation *ndb_op, Field *field,
                        pack_len));
   DBUG_DUMP("key", (char*)field_ptr, pack_len);
   
-  if (ndb_supported_type(field->type()))
-  {
-    if (! (field->flags & BLOB_FLAG))
-      // Common implementation for most field types
-      DBUG_RETURN(ndb_op->equal(fieldnr, (char*) field_ptr, pack_len) != 0);
-  }
-  // Unhandled field types
-  DBUG_PRINT("error", ("Field type %d not supported", field->type()));
-  DBUG_RETURN(2);
+  DBUG_ASSERT(ndb_supported_type(field->type()));
+  DBUG_ASSERT(! (field->flags & BLOB_FLAG));
+  // Common implementation for most field types
+  DBUG_RETURN(ndb_op->equal(fieldnr, (char*) field_ptr, pack_len) != 0);
 }
 
 
@@ -637,7 +637,7 @@ int ha_ndbcluster::set_ndb_value(NdbOperation *ndb_op, Field *field,
                        pack_len, field->is_null()?"Y":"N"));
   DBUG_DUMP("value", (char*) field_ptr, pack_len);
 
-  if (ndb_supported_type(field->type()))
+  DBUG_ASSERT(ndb_supported_type(field->type()));
   {
     // ndb currently does not support size 0
     uint32 empty_field;
@@ -715,9 +715,6 @@ int ha_ndbcluster::set_ndb_value(NdbOperation *ndb_op, Field *field,
     }
     DBUG_RETURN(1);
   }
-  // Unhandled field types
-  DBUG_PRINT("error", ("Field type %d not supported", field->type()));
-  DBUG_RETURN(2);
 }
 
 
@@ -812,9 +809,8 @@ int ha_ndbcluster::get_ndb_value(NdbOperation *ndb_op, Field *field,
 
   if (field != NULL)
   {
-    DBUG_ASSERT(buf);
-    if (ndb_supported_type(field->type()))
-    {
+      DBUG_ASSERT(buf);
+      DBUG_ASSERT(ndb_supported_type(field->type()));
       DBUG_ASSERT(field->ptr != NULL);
       if (! (field->flags & BLOB_FLAG))
       { 
@@ -845,10 +841,6 @@ int ha_ndbcluster::get_ndb_value(NdbOperation *ndb_op, Field *field,
         DBUG_RETURN(ndb_blob->setActiveHook(g_get_ndb_blobs_value, arg) != 0);
       }
       DBUG_RETURN(1);
-    }
-    // Unhandled field types
-    DBUG_PRINT("error", ("Field type %d not supported", field->type()));
-    DBUG_RETURN(2);
   }
 
   // Used for hidden key only
@@ -3132,6 +3124,13 @@ double ha_ndbcluster::scan_time()
   DBUG_RETURN(res);
 }
 
+/*
+  Convert MySQL table locks into locks supported by Ndb Cluster.
+  Note that MySQL Cluster does currently not support distributed
+  table locks, so to be safe one should set cluster in Single
+  User Mode, before relying on table locks when updating tables
+  from several MySQL servers
+*/
 
 THR_LOCK_DATA **ha_ndbcluster::store_lock(THD *thd,
                                           THR_LOCK_DATA **to,
@@ -3147,7 +3146,7 @@ THR_LOCK_DATA **ha_ndbcluster::store_lock(THD *thd,
     /* Since NDB does not currently have table locks
        this is treated as a ordinary lock */
 
-    if ((lock_type >= TL_WRITE_ALLOW_WRITE &&
+    if ((lock_type >= TL_WRITE_CONCURRENT_INSERT &&
          lock_type <= TL_WRITE) && !thd->in_lock_tables)      
       lock_type= TL_WRITE_ALLOW_WRITE;
     
@@ -4446,7 +4445,7 @@ int ndbcluster_discover(THD* thd, const char *db, const char *name,
   {    
     const NdbError err= dict->getNdbError();
     if (err.code == 709)
-      DBUG_RETURN(1);
+      DBUG_RETURN(-1);
     ERR_RETURN(err);
   }
   DBUG_PRINT("info", ("Found table %s", tab->getName()));
@@ -4454,13 +4453,15 @@ int ndbcluster_discover(THD* thd, const char *db, const char *name,
   len= tab->getFrmLength();  
   if (len == 0 || tab->getFrmData() == NULL)
   {
-    DBUG_PRINT("No frm data found",
-               ("Table is probably created via NdbApi")); 
-    DBUG_RETURN(2);
+    DBUG_PRINT("error", ("No frm data found."));
+    DBUG_RETURN(1);
   }
   
   if (unpackfrm(&data, &len, tab->getFrmData()))
-    DBUG_RETURN(3);
+  {
+    DBUG_PRINT("error", ("Could not unpack table"));
+    DBUG_RETURN(1);
+  }
 
   *frmlen= len;
   *frmblob= data;
@@ -4473,11 +4474,11 @@ int ndbcluster_discover(THD* thd, const char *db, const char *name,
 
  */
 
-int ndbcluster_table_exists(THD* thd, const char *db, const char *name)
+int ndbcluster_table_exists_in_engine(THD* thd, const char *db, const char *name)
 {
   const NDBTAB* tab;
   Ndb* ndb;
-  DBUG_ENTER("ndbcluster_table_exists");
+  DBUG_ENTER("ndbcluster_table_exists_in_engine");
   DBUG_PRINT("enter", ("db: %s, name: %s", db, name));
 
   if (!(ndb= check_ndb_in_thd(thd)))
@@ -4656,7 +4657,7 @@ int ndbcluster_find_files(THD *thd,const char *db,const char *path,
     DBUG_PRINT("info", ("%s existed on disk", name));     
     // The .ndb file exists on disk, but it's not in list of tables in ndb
     // Verify that handler agrees table is gone.
-    if (ndbcluster_table_exists(thd, db, file_name) == 0)    
+    if (ndbcluster_table_exists_in_engine(thd, db, file_name) == 0)    
     {
       DBUG_PRINT("info", ("NDB says %s does not exists", file_name));     
       it.remove();
@@ -4710,7 +4711,7 @@ int ndbcluster_find_files(THD *thd,const char *db,const char *path,
   while ((file_name=it2++))
   {  
     DBUG_PRINT("info", ("Table %s need discovery", file_name));
-    if (ha_create_table_from_engine(thd, db, file_name, TRUE) == 0)
+    if (ha_create_table_from_engine(thd, db, file_name) == 0)
       files->push_back(thd->strdup(file_name)); 
   }
 
@@ -6399,6 +6400,7 @@ void ndb_serialize_cond(const Item *item, void *arg)
                   case(REAL_RESULT):
                     context->expect_only(Item::REAL_ITEM);
                     context->expect(Item::DECIMAL_ITEM);
+                    context->expect(Item::INT_ITEM);
                     break;
                   case(INT_RESULT):
                     context->expect_only(Item::INT_ITEM);
@@ -6407,6 +6409,7 @@ void ndb_serialize_cond(const Item *item, void *arg)
                   case(DECIMAL_RESULT):
                     context->expect_only(Item::DECIMAL_ITEM);
                     context->expect(Item::REAL_ITEM);
+                    context->expect(Item::INT_ITEM);
                     break;
                   default:
                     break;
@@ -6809,6 +6812,8 @@ void ndb_serialize_cond(const Item *item, void *arg)
               // We have not seen the field argument yet
               context->expect_only(Item::FIELD_ITEM);
               context->expect_only_field_result(INT_RESULT);
+              context->expect_field_result(REAL_RESULT);
+              context->expect_field_result(DECIMAL_RESULT);
             }
             else
             {
diff --git a/sql/ha_ndbcluster.h b/sql/ha_ndbcluster.h
index 9ea32e61190..d20fafa458f 100644
--- a/sql/ha_ndbcluster.h
+++ b/sql/ha_ndbcluster.h
@@ -681,7 +681,8 @@ int ndbcluster_discover(THD* thd, const char* dbname, const char* name,
                         const void** frmblob, uint* frmlen);
 int ndbcluster_find_files(THD *thd,const char *db,const char *path,
                           const char *wild, bool dir, List<char> *files);
-int ndbcluster_table_exists(THD* thd, const char *db, const char *name);
+int ndbcluster_table_exists_in_engine(THD* thd,
+                                      const char *db, const char *name);
 int ndbcluster_drop_database(const char* path);
 
 void ndbcluster_print_error(int error, const NdbOperation *error_op);
diff --git a/sql/handler.cc b/sql/handler.cc
index 1973cd71d46..46a80770024 100644
--- a/sql/handler.cc
+++ b/sql/handler.cc
@@ -165,12 +165,22 @@ my_bool ha_storage_engine_is_enabled(enum db_type database_type)
 
 	/* Use other database handler if databasehandler is not incompiled */
 
-enum db_type ha_checktype(enum db_type database_type)
+enum db_type ha_checktype(THD *thd, enum db_type database_type,
+                          bool no_substitute, bool report_error)
 {
-  THD *thd;
   if (ha_storage_engine_is_enabled(database_type))
     return database_type;
 
+  if (no_substitute)
+  {
+    if (report_error)
+    {
+      const char *engine_name= ha_get_storage_engine(database_type);
+      my_error(ER_FEATURE_DISABLED,MYF(0),engine_name,engine_name);
+    }
+    return DB_TYPE_UNKNOWN;
+  }
+
   switch (database_type) {
 #ifndef NO_HASH
   case DB_TYPE_HASH:
@@ -182,7 +192,6 @@ enum db_type ha_checktype(enum db_type database_type)
     break;
   }
   
-  thd= current_thd;
   return ((enum db_type) thd->variables.table_type != DB_TYPE_UNKNOWN ?
           (enum db_type) thd->variables.table_type :
           ((enum db_type) global_system_variables.table_type !=
@@ -1919,21 +1928,19 @@ int ha_create_table(const char *name, HA_CREATE_INFO *create_info,
 }
 
 /*
-  Try to discover table from engine and 
+  Try to discover table from engine and
   if found, write the frm file to disk.
-  
+
   RETURN VALUES:
-   0 : Table existed in engine and created 
-       on disk if so requested
-   1 : Table does not exist
-  >1 : error
+  -1 : Table did not exists
+   0 : Table created ok
+   > 0 : Error, table existed but could not be created
 
 */
 
-int ha_create_table_from_engine(THD* thd, 
-				const char *db, 
-				const char *name,
-				bool create_if_found)
+int ha_create_table_from_engine(THD* thd,
+				const char *db,
+				const char *name)
 {
   int error;
   const void *frmblob;
@@ -1942,45 +1949,47 @@ int ha_create_table_from_engine(THD* thd,
   HA_CREATE_INFO create_info;
   TABLE table;
   DBUG_ENTER("ha_create_table_from_engine");
-  DBUG_PRINT("enter", ("name '%s'.'%s'  create_if_found: %d",
-                       db, name, create_if_found));
+  DBUG_PRINT("enter", ("name '%s'.'%s'",
+                       db, name));
 
   bzero((char*) &create_info,sizeof(create_info));
 
-  if ((error= ha_discover(thd, db, name, &frmblob, &frmlen)))
-    DBUG_RETURN(error); 
+  if(error= ha_discover(thd, db, name, &frmblob, &frmlen))
+  {
+    // Table could not be discovered and thus not created
+    DBUG_RETURN(error);
+  }
+
   /*
-    Table exists in handler
-    frmblob and frmlen are set
+    Table exists in handler and could be discovered
+    frmblob and frmlen are set, write the frm to disk
   */
 
-  if (create_if_found)
+  (void)strxnmov(path,FN_REFLEN,mysql_data_home,"/",db,"/",name,NullS);
+  // Save the frm file
+  if (writefrm(path, frmblob, frmlen))
   {
-    (void)strxnmov(path,FN_REFLEN,mysql_data_home,"/",db,"/",name,NullS);
-    // Save the frm file    
-    if ((error = writefrm(path, frmblob, frmlen)))
-      goto err_end;
+    my_free((char*) frmblob, MYF(MY_ALLOW_ZERO_PTR));
+    DBUG_RETURN(2);
+  }
 
-    if (openfrm(thd, path,"",0,(uint) READ_ALL, 0, &table))
-      DBUG_RETURN(1);
+  if (openfrm(thd, path,"",0,(uint) READ_ALL, 0, &table))
+    DBUG_RETURN(3);
 
-    update_create_info_from_table(&create_info, &table);
-    create_info.table_options|= HA_CREATE_FROM_ENGINE;
+  update_create_info_from_table(&create_info, &table);
+  create_info.table_options|= HA_CREATE_FROM_ENGINE;
 
-    if (lower_case_table_names == 2 &&
-	!(table.file->table_flags() & HA_FILE_BASED))
-    {
-      /* Ensure that handler gets name in lower case */
-      my_casedn_str(files_charset_info, path);
-    }
-    
-    error=table.file->create(path,&table,&create_info);
-    VOID(closefrm(&table));
+  if (lower_case_table_names == 2 &&
+      !(table.file->table_flags() & HA_FILE_BASED))
+  {
+    /* Ensure that handler gets name in lower case */
+    my_casedn_str(files_charset_info, path);
   }
-
-err_end:
+  error=table.file->create(path,&table,&create_info);
+  VOID(closefrm(&table));
   my_free((char*) frmblob, MYF(MY_ALLOW_ZERO_PTR));
-  DBUG_RETURN(error);  
+
+  DBUG_RETURN(error != 0);
 }
 
 void st_ha_check_opt::init()
@@ -2083,14 +2092,15 @@ int ha_change_key_cache(KEY_CACHE *old_key_cache,
   Try to discover one table from handler(s)
 
   RETURN
-    0  ok. In this case *frmblob and *frmlen are set
-    1  error.  frmblob and frmlen may not be set
+   -1  : Table did not exists
+    0  : OK. In this case *frmblob and *frmlen are set
+    >0 : error.  frmblob and frmlen may not be set
 */
 
 int ha_discover(THD *thd, const char *db, const char *name,
 		const void **frmblob, uint *frmlen)
 {
-  int error= 1; // Table does not exist in any handler
+  int error= -1; // Table does not exist in any handler
   DBUG_ENTER("ha_discover");
   DBUG_PRINT("enter", ("db: %s, name: %s", db, name));
 #ifdef HAVE_NDBCLUSTER_DB
@@ -2122,11 +2132,8 @@ ha_find_files(THD *thd,const char *db,const char *path,
     error= ndbcluster_find_files(thd, db, path, wild, dir, files);
 #endif
   DBUG_RETURN(error);
-  
-  
 }
 
-#ifdef NOT_YET_USED
 
 /*
   Ask handler if the table exists in engine
@@ -2137,20 +2144,19 @@ ha_find_files(THD *thd,const char *db,const char *path,
     #                   Error code
 
  */
-int ha_table_exists(THD* thd, const char* db, const char* name)
+int ha_table_exists_in_engine(THD* thd, const char* db, const char* name)
 {
-  int error= 2;
-  DBUG_ENTER("ha_table_exists");
+  int error= 0;
+  DBUG_ENTER("ha_table_exists_in_engine");
   DBUG_PRINT("enter", ("db: %s, name: %s", db, name));
 #ifdef HAVE_NDBCLUSTER_DB
   if (have_ndbcluster == SHOW_OPTION_YES)
-    error= ndbcluster_table_exists(thd, db, name);
+    error= ndbcluster_table_exists_in_engine(thd, db, name);
 #endif
+  DBUG_PRINT("exit", ("error: %d", error));
   DBUG_RETURN(error);
 }
 
-#endif
-
 
 /*
   Read the first row of a multi-range set.
diff --git a/sql/handler.h b/sql/handler.h
index 3b0b9afe320..df906e284e7 100644
--- a/sql/handler.h
+++ b/sql/handler.h
@@ -813,7 +813,8 @@ extern ulong total_ha, total_ha_2pc;
 enum db_type ha_resolve_by_name(const char *name, uint namelen);
 const char *ha_get_storage_engine(enum db_type db_type);
 handler *get_new_handler(TABLE *table, enum db_type db_type);
-enum db_type ha_checktype(enum db_type database_type);
+enum db_type ha_checktype(THD *thd, enum db_type database_type,
+                          bool no_substitute, bool report_error);
 
 /* basic stuff */
 int ha_init(void);
@@ -830,13 +831,12 @@ int ha_delete_table(THD *thd, enum db_type db_type, const char *path,
                     const char *alias, bool generate_warning);
 
 /* discovery */
-int ha_create_table_from_engine(THD* thd, const char *db, const char *name,
-				bool create_if_found);
+int ha_create_table_from_engine(THD* thd, const char *db, const char *name);
 int ha_discover(THD* thd, const char* dbname, const char* name,
                 const void** frmblob, uint* frmlen);
 int ha_find_files(THD *thd,const char *db,const char *path,
                   const char *wild, bool dir,List<char>* files);
-int ha_table_exists(THD* thd, const char* db, const char* name);
+int ha_table_exists_in_engine(THD* thd, const char* db, const char* name);
 
 /* key cache */
 int ha_init_key_cache(const char *name, KEY_CACHE *key_cache);
diff --git a/sql/item.cc b/sql/item.cc
index 680b771f908..b6f8b7ebc51 100644
--- a/sql/item.cc
+++ b/sql/item.cc
@@ -307,13 +307,14 @@ void *Item::operator new(size_t size, Item *reuse, uint *rsize)
     return (void *)reuse;
   }
   if (rsize)
-    (*rsize)= size;
+    (*rsize)= (uint) size;
   return (void *)sql_alloc((uint)size);
 }
 
 
 Item::Item():
   rsize(0), name(0), orig_name(0), name_length(0), fixed(0),
+  is_autogenerated_name(TRUE),
   collation(&my_charset_bin, DERIVATION_COERCIBLE)
 {
   marker= 0;
@@ -337,6 +338,7 @@ Item::Item():
 	place == IN_HAVING)
       thd->lex->current_select->select_n_having_items++;
   }
+  item_flags= 0;
 }
 
 /*
@@ -357,7 +359,8 @@ Item::Item(THD *thd, Item *item):
   unsigned_flag(item->unsigned_flag),
   with_sum_func(item->with_sum_func),
   fixed(item->fixed),
-  collation(item->collation)
+  collation(item->collation),
+  item_flags(item->item_flags)
 {
   next= thd->free_list;				// Put in free list
   thd->free_list= this;
@@ -382,7 +385,7 @@ void Item::print_item_w_name(String *str)
   {
     THD *thd= current_thd;
     str->append(" AS ", 4);
-    append_identifier(thd, str, name, strlen(name));
+    append_identifier(thd, str, name, (uint) strlen(name));
   }
 }
 
@@ -1132,27 +1135,27 @@ void Item_ident::print(String *str)
   if (!table_name || !field_name)
   {
     const char *nm= field_name ? field_name : name ? name : "tmp_field";
-    append_identifier(thd, str, nm, strlen(nm));
+    append_identifier(thd, str, nm, (uint) strlen(nm));
     return;
   }
   if (db_name && db_name[0] && !alias_name_used)
   {
-    append_identifier(thd, str, d_name, strlen(d_name));
+    append_identifier(thd, str, d_name, (uint) strlen(d_name));
     str->append('.');
-    append_identifier(thd, str, t_name, strlen(t_name));
+    append_identifier(thd, str, t_name, (uint) strlen(t_name));
     str->append('.');
-    append_identifier(thd, str, field_name, strlen(field_name));
+    append_identifier(thd, str, field_name, (uint) strlen(field_name));
   }
   else
   {
     if (table_name[0])
     {
-      append_identifier(thd, str, t_name, strlen(t_name));
+      append_identifier(thd, str, t_name, (uint) strlen(t_name));
       str->append('.');
-      append_identifier(thd, str, field_name, strlen(field_name));
+      append_identifier(thd, str, field_name, (uint) strlen(field_name));
     }
     else
-      append_identifier(thd, str, field_name, strlen(field_name));
+      append_identifier(thd, str, field_name, (uint) strlen(field_name));
   }
 }
 
@@ -2179,7 +2182,7 @@ const String *Item_param::query_val_str(String* str) const
       ptr+= escape_string_for_mysql(str_value.charset(), ptr, 0,
                                     str_value.ptr(), str_value.length());
       *ptr++= '\'';
-      str->length(ptr - buf);
+      str->length((uint32) (ptr - buf));
       break;
     }
   case NULL_VALUE:
@@ -3344,6 +3347,9 @@ Field *Item::tmp_table_field_from_field_type(TABLE *table)
   case MYSQL_TYPE_YEAR:
     return new Field_year((char*) 0, max_length, null_ptr, 0, Field::NONE,
 			  name, table);
+  case MYSQL_TYPE_BIT:
+    return new Field_bit_as_char(NULL, max_length, null_ptr, 0, NULL, 0,
+                                 Field::NONE, name, table);
   default:
     /* This case should never be chosen */
     DBUG_ASSERT(0);
@@ -3570,7 +3576,7 @@ Item *Item_int_with_ref::new_item()
 
 Item_num *Item_uint::neg()
 {
-  Item_decimal *item= new Item_decimal(value, 0);
+  Item_decimal *item= new Item_decimal(value, 1);
   return item->neg();
 }
 
@@ -4360,6 +4366,28 @@ my_decimal *Item_ref::val_decimal(my_decimal *decimal_value)
   return val;
 }
 
+int Item_ref::save_in_field(Field *to, bool no_conversions)
+{
+  int res;
+  if(result_field){
+    if (result_field->is_null())
+    {
+      null_value= 1;
+      return set_field_to_null_with_conversions(to, no_conversions);
+    }
+    else
+    {
+      to->set_notnull();
+      field_conv(to, result_field);
+      null_value= 0;
+    }
+    return 0;
+  }
+  res= (*ref)->save_in_field(to, no_conversions);
+  null_value= (*ref)->null_value;
+  return res;
+}
+
 
 void Item_ref_null_helper::print(String *str)
 {
@@ -4443,6 +4471,7 @@ bool Item_default_value::fix_fields(THD *thd,
 				    struct st_table_list *table_list,
 				    Item **items)
 {
+  Item *real_arg;
   Item_field *field_arg;
   Field *def_field;
   DBUG_ASSERT(fixed == 0);
@@ -4454,17 +4483,15 @@ bool Item_default_value::fix_fields(THD *thd,
   }
   if (!arg->fixed && arg->fix_fields(thd, table_list, &arg))
     return TRUE;
-  
-  if (arg->type() == REF_ITEM)
+
+  real_arg= arg->real_item();
+  if (real_arg->type() != FIELD_ITEM)
   {
-    Item_ref *ref= (Item_ref *)arg;
-    if (ref->ref[0]->type() != FIELD_ITEM)
-    {
-      return TRUE;
-    }
-    arg= ref->ref[0];
+    my_error(ER_NO_DEFAULT_FOR_FIELD, MYF(0), arg->name);
+    return TRUE;
   }
-  field_arg= (Item_field *)arg;
+
+  field_arg= (Item_field *)real_arg;
   if (field_arg->field->flags & NO_DEFAULT_VALUE_FLAG)
   {
     my_error(ER_NO_DEFAULT_FOR_FIELD, MYF(0), field_arg->field->field_name);
@@ -4597,7 +4624,8 @@ void Item_trigger_field::setup_field(THD *thd, TABLE *table)
     Try to find field by its name and if it will be found
     set field_idx properly.
   */
-  (void)find_field_in_real_table(thd, table, field_name, strlen(field_name),
+  (void)find_field_in_real_table(thd, table, field_name,
+                                 (uint) strlen(field_name),
                                  0, 0, &field_idx);
   thd->set_query_id= save_set_query_id;
   triggers= table->triggers;
@@ -5387,7 +5415,7 @@ void Item_result_field::cleanup()
 ** Instantiate templates
 *****************************************************************************/
 
-#ifdef __GNUC__
+#ifdef HAVE_EXPLICIT_TEMPLATE_INSTANTIATION
 template class List<Item>;
 template class List_iterator<Item>;
 template class List_iterator_fast<Item>;
diff --git a/sql/item.h b/sql/item.h
index 1fe90a41e3a..c8180b4932a 100644
--- a/sql/item.h
+++ b/sql/item.h
@@ -225,6 +225,11 @@ typedef Item* (Item::*Item_transformer) (byte *arg);
 
 typedef void (*Cond_traverser) (const Item *item, void *arg);
 
+/*
+ See comments for sql_yacc.yy: insert_update_elem rule
+ */
+#define MY_ITEM_PREFER_1ST_TABLE 1
+
 class Item {
   Item(const Item &);			/* Prevent use of these */
   void operator=(Item &);
@@ -269,7 +274,10 @@ public:
   my_bool unsigned_flag;
   my_bool with_sum_func;
   my_bool fixed;                        /* If item fixed with fix_fields */
+  my_bool is_autogenerated_name;        /* indicate was name of this Item
+                                           autogenerated or set by user */
   DTCollation collation;
+  uint8 item_flags;		/* Flags on how item should be processed */
 
   // alloc & destruct is done as start of select using sql_alloc
   Item();
@@ -288,7 +296,7 @@ public:
     name=0;
 #endif
   }		/*lint -e1509 */
-  void set_name(const char *str,uint length, CHARSET_INFO *cs);
+  void set_name(const char *str, uint length, CHARSET_INFO *cs);
   void rename(char *new_name);
   void init_make_field(Send_field *tmp_field,enum enum_field_types type);
   virtual void cleanup();
@@ -468,6 +476,18 @@ public:
   */
   virtual bool const_during_execution() const 
   { return (used_tables() & ~PARAM_TABLE_BIT) == 0; }
+  /*
+    This is an essential method for correct functioning of VIEWS.
+    To save a view in an .frm file we need its unequivocal
+    definition in SQL that takes into account sql_mode and
+    environmental settings.  Currently such definition is restored
+    by traversing through the parsed tree of a view and
+    print()'ing SQL syntax of every node to a String buffer. This
+    method is used to print the SQL definition of an item. The
+    second use of this method is for EXPLAIN EXTENDED, to print
+    the SQL of a query after all optimizations of the parsed tree
+    have been done.
+  */
   virtual void print(String *str_arg) { str_arg->append(full_name()); }
   void print_item_w_name(String *);
   virtual void update_used_tables() {}
@@ -570,6 +590,11 @@ public:
     cleanup();
     delete this;
   }
+  virtual bool set_flags_processor(byte *args)
+  {
+    this->item_flags|= *((uint8*)args);
+    return false;
+  }
 
   virtual bool is_splocal() { return 0; } /* Needed for error checking */
 };
@@ -630,7 +655,7 @@ public:
     Item *it= this_item();
 
     if (name)
-      it->set_name(name, strlen(name), system_charset_info);
+      it->set_name(name, (uint) strlen(name), system_charset_info);
     else
       it->set_name(m_name.str, m_name.length, system_charset_info);
     it->make_field(field);
@@ -704,10 +729,6 @@ public:
   void cleanup();
   bool remove_dependence_processor(byte * arg);
   void print(String *str);
-
-  friend bool insert_fields(THD *thd,TABLE_LIST *tables, const char *db_name,
-                            const char *table_name, List_iterator<Item> *it,
-                            bool any_privileges, bool allocate_view_names);
 };
 
 class Item_equal;
@@ -1158,7 +1179,7 @@ public:
     collation.set(cs, dv);
     str_value.set_or_copy_aligned(str,length,cs);
     max_length= str_value.numchars()*cs->mbmaxlen;
-    set_name(name_par,0,cs);
+    set_name(name_par, 0, cs);
     decimals=NOT_FIXED_DEC;
     // it is constant => can be used without fix_fields (and frequently used)
     fixed= 1;
@@ -1335,8 +1356,7 @@ public:
   bool send(Protocol *prot, String *tmp);
   void make_field(Send_field *field)	{ (*ref)->make_field(field); }
   bool fix_fields(THD *, struct st_table_list *, Item **);
-  int save_in_field(Field *field, bool no_conversions)
-  { return (*ref)->save_in_field(field, no_conversions); }
+  int save_in_field(Field *field, bool no_conversions);
   void save_org_in_field(Field *field)	{ (*ref)->save_org_in_field(field); }
   enum Item_result result_type () const { return (*ref)->result_type(); }
   enum_field_types field_type() const   { return (*ref)->field_type(); }
@@ -1352,7 +1372,10 @@ public:
   {
     (*ref)->save_in_field(result_field, no_conversions);
   }
-  Item *real_item() { return *ref; }
+  Item *real_item()
+  {
+    return (*ref)->real_item();
+  }
   bool walk(Item_processor processor, byte *arg)
   { return (*ref)->walk(processor, arg); }
   void print(String *str);
@@ -1478,7 +1501,7 @@ public:
 		       str_value.length(), &end_not_used, &err_not_used));
   }
   longlong val_int()
-  { 
+  {
     int err;
     return null_value ? LL(0) : my_strntoll(str_value.charset(),str_value.ptr(),str_value.length(),10, (char**) 0,&err); 
   }
@@ -1493,62 +1516,62 @@ public:
 };
 
 
-class Item_buff :public Sql_alloc
+class Cached_item :public Sql_alloc
 {
 public:
   my_bool null_value;
-  Item_buff() :null_value(0) {}
+  Cached_item() :null_value(0) {}
   virtual bool cmp(void)=0;
-  virtual ~Item_buff(); /*line -e1509 */
+  virtual ~Cached_item(); /*line -e1509 */
 };
 
-class Item_str_buff :public Item_buff
+class Cached_item_str :public Cached_item
 {
   Item *item;
   String value,tmp_value;
 public:
-  Item_str_buff(THD *thd, Item *arg);
+  Cached_item_str(THD *thd, Item *arg);
   bool cmp(void);
-  ~Item_str_buff();				// Deallocate String:s
+  ~Cached_item_str();                           // Deallocate String:s
 };
 
 
-class Item_real_buff :public Item_buff
+class Cached_item_real :public Cached_item
 {
   Item *item;
   double value;
 public:
-  Item_real_buff(Item *item_par) :item(item_par),value(0.0) {}
+  Cached_item_real(Item *item_par) :item(item_par),value(0.0) {}
   bool cmp(void);
 };
 
-class Item_int_buff :public Item_buff
+class Cached_item_int :public Cached_item
 {
   Item *item;
   longlong value;
 public:
-  Item_int_buff(Item *item_par) :item(item_par),value(0) {}
+  Cached_item_int(Item *item_par) :item(item_par),value(0) {}
   bool cmp(void);
 };
 
 
-class Item_decimal_buff :public Item_buff
+class Cached_item_decimal :public Cached_item
 {
   Item *item;
   my_decimal value;
 public:
-  Item_decimal_buff(Item *item_par);
+  Cached_item_decimal(Item *item_par);
   bool cmp(void);
 };
 
-class Item_field_buff :public Item_buff
+class Cached_item_field :public Cached_item
 {
   char *buff;
   Field *field;
   uint length;
 
 public:
-  Item_field_buff(Item_field *item)
+  Cached_item_field(Item_field *item)
   {
     field=item->field;
     buff= (char*) sql_calloc(length=field->pack_length());
@@ -1570,7 +1593,7 @@ public:
   void print(String *str);
   int save_in_field(Field *field_arg, bool no_conversions);
   table_map used_tables() const { return (table_map)0L; }
-  
+
   bool walk(Item_processor processor, byte *args)
   {
     return arg->walk(processor, args) ||
@@ -1876,7 +1899,7 @@ void mark_select_range_as_dependent(THD *thd,
                                     Field *found_field, Item *found_item,
                                     Item_ident *resolved_item);
 
-extern Item_buff *new_Item_buff(THD *thd, Item *item);
+extern Cached_item *new_Cached_item(THD *thd, Item *item);
 extern Item_result item_cmp_type(Item_result a,Item_result b);
 extern void resolve_const_item(THD *thd, Item **ref, Item *cmp_item);
 extern bool field_is_equal_to_item(Field *field,Item *item);
diff --git a/sql/item_buff.cc b/sql/item_buff.cc
index 688e4cca846..a67e420170a 100644
--- a/sql/item_buff.cc
+++ b/sql/item_buff.cc
@@ -20,23 +20,23 @@
 #include "mysql_priv.h"
 
 /*
-** Create right type of item_buffer for an item
+** Create right type of Cached_item for an item
 */
 
-Item_buff *new_Item_buff(THD *thd, Item *item)
+Cached_item *new_Cached_item(THD *thd, Item *item)
 {
   if (item->type() == Item::FIELD_ITEM &&
       !(((Item_field *) item)->field->flags & BLOB_FLAG))
-    return new Item_field_buff((Item_field *) item);
+    return new Cached_item_field((Item_field *) item);
   switch (item->result_type()) {
   case STRING_RESULT:
-    return new Item_str_buff(thd, (Item_field *) item);
+    return new Cached_item_str(thd, (Item_field *) item);
   case INT_RESULT:
-    return new Item_int_buff((Item_field *) item);
+    return new Cached_item_int((Item_field *) item);
   case REAL_RESULT:
-    return new Item_real_buff(item);
+    return new Cached_item_real(item);
   case DECIMAL_RESULT:
-    return new Item_decimal_buff(item);
+    return new Cached_item_decimal(item);
   case ROW_RESULT:
   default:
     DBUG_ASSERT(0);
@@ -44,24 +44,24 @@ Item_buff *new_Item_buff(THD *thd, Item *item)
   }
 }
 
-Item_buff::~Item_buff() {}
+Cached_item::~Cached_item() {}
 
 /*
 ** Compare with old value and replace value with new value
 ** Return true if values have changed
 */
 
-Item_str_buff::Item_str_buff(THD *thd, Item *arg)
+Cached_item_str::Cached_item_str(THD *thd, Item *arg)
   :item(arg), value(min(arg->max_length, thd->variables.max_sort_length))
 {}
 
-bool Item_str_buff::cmp(void)
+bool Cached_item_str::cmp(void)
 {
   String *res;
   bool tmp;
 
-  res=item->val_str(&tmp_value);
-  res->length(min(res->length(), value.alloced_length()));
+  if ((res=item->val_str(&tmp_value)))
+    res->length(min(res->length(), value.alloced_length()));
   if (null_value != item->null_value)
   {
     if ((null_value= item->null_value))
@@ -77,12 +77,12 @@ bool Item_str_buff::cmp(void)
   return tmp;
 }
 
-Item_str_buff::~Item_str_buff()
+Cached_item_str::~Cached_item_str()
 {
   item=0;					// Safety
 }
 
-bool Item_real_buff::cmp(void)
+bool Cached_item_real::cmp(void)
 {
   double nr= item->val_real();
   if (null_value != item->null_value || nr != value)
@@ -94,7 +94,7 @@ bool Item_real_buff::cmp(void)
   return FALSE;
 }
 
-bool Item_int_buff::cmp(void)
+bool Cached_item_int::cmp(void)
 {
   longlong nr=item->val_int();
   if (null_value != item->null_value || nr != value)
@@ -107,7 +107,7 @@ bool Item_int_buff::cmp(void)
 }
 
 
-bool Item_field_buff::cmp(void)
+bool Cached_item_field::cmp(void)
 {
   bool tmp= field->cmp(buff) != 0;		// This is not a blob!
   if (tmp)
@@ -121,14 +121,14 @@ bool Item_field_buff::cmp(void)
 }
 
 
-Item_decimal_buff::Item_decimal_buff(Item *it)
+Cached_item_decimal::Cached_item_decimal(Item *it)
   :item(it)
 {
   my_decimal_set_zero(&value);
 }
 
 
-bool Item_decimal_buff::cmp()
+bool Cached_item_decimal::cmp()
 {
   my_decimal tmp;
   my_decimal *ptmp= item->val_decimal(&tmp);
@@ -146,7 +146,7 @@ bool Item_decimal_buff::cmp()
 ** Instansiate templates
 *****************************************************************************/
 
-#ifdef __GNUC__
-template class List<Item_buff>;
-template class List_iterator<Item_buff>;
+#ifdef HAVE_EXPLICIT_TEMPLATE_INSTANTIATION
+template class List<Cached_item>;
+template class List_iterator<Cached_item>;
 #endif
diff --git a/sql/item_cmpfunc.cc b/sql/item_cmpfunc.cc
index b5b37efaf07..58a7f3316d7 100644
--- a/sql/item_cmpfunc.cc
+++ b/sql/item_cmpfunc.cc
@@ -238,9 +238,10 @@ void Item_bool_func2::fix_length_and_dec()
     return;
   }
     
-  if (args[0]->type() == FIELD_ITEM)
+  Item *real_item= args[0]->real_item();
+  if (real_item->type() == FIELD_ITEM)
   {
-    Field *field=((Item_field*) args[0])->field;
+    Field *field= ((Item_field*) real_item)->field;
     if (field->can_be_compared_as_longlong())
     {
       if (convert_constant_item(thd, field,&args[1]))
@@ -251,9 +252,10 @@ void Item_bool_func2::fix_length_and_dec()
       }
     }
   }
-  if (args[1]->type() == FIELD_ITEM /* && !args[1]->const_item() */)
+  real_item= args[1]->real_item();
+  if (real_item->type() == FIELD_ITEM)
   {
-    Field *field=((Item_field*) args[1])->field;
+    Field *field= ((Item_field*) real_item)->field;
     if (field->can_be_compared_as_longlong())
     {
       if (convert_constant_item(thd, field,&args[0]))
@@ -649,7 +651,7 @@ bool Item_in_optimizer::fix_left(THD *thd,
     If it is preparation PS only then we do not know values of parameters =>
     cant't get there values and do not need that values.
   */
-  if (!thd->only_prepare())
+  if (!thd->current_arena->is_stmt_prepare())
     cache->store(args[0]);
   if (cache->cols() == 1)
   {
@@ -1406,9 +1408,7 @@ Item_func_nullif::val_decimal(my_decimal * decimal_value)
 bool
 Item_func_nullif::is_null()
 {
-  if (!cmp.compare())
-    return (null_value=1);
-  return 0;
+  return (null_value= (!cmp.compare() ? 1 : args[0]->null_value)); 
 }
 
 /*
@@ -1422,6 +1422,8 @@ Item *Item_func_case::find_item(String *str)
   my_decimal *first_expr_dec, first_expr_dec_val;
   longlong first_expr_int;
   double   first_expr_real;
+  char buff[MAX_FIELD_WIDTH];
+  String buff_str(buff,sizeof(buff),default_charset());
 
   /* These will be initialized later */
   LINT_INIT(first_expr_str);
@@ -1435,7 +1437,7 @@ Item *Item_func_case::find_item(String *str)
     {
       case STRING_RESULT:
       	// We can't use 'str' here as this may be overwritten
-	if (!(first_expr_str= args[first_expr_num]->val_str(&str_value)))
+	if (!(first_expr_str= args[first_expr_num]->val_str(&buff_str)))
 	  return else_expr_num != -1 ? args[else_expr_num] : 0;	// Impossible
         break;
       case INT_RESULT:
@@ -2794,10 +2796,11 @@ bool Item_func_like::fix_fields(THD *thd, TABLE_LIST *tlist, Item ** ref)
       if (canDoTurboBM)
       {
         pattern     = first + 1;
-        pattern_len = len - 2;
+        pattern_len = (int) len - 2;
         DBUG_PRINT("info", ("Initializing pattern: '%s'", first));
-        int *suff = (int*) thd->alloc(sizeof(int)*((pattern_len + 1)*2+
-                                      alphabet_size));
+        int *suff = (int*) thd->alloc((int) (sizeof(int)*
+                                      ((pattern_len + 1)*2+
+                                      alphabet_size)));
         bmGs      = suff + pattern_len + 1;
         bmBc      = bmGs + pattern_len + 1;
         turboBM_compute_good_suffix_shifts(suff);
diff --git a/sql/item_func.cc b/sql/item_func.cc
index 91defd7f0ee..1dbf28b67cb 100644
--- a/sql/item_func.cc
+++ b/sql/item_func.cc
@@ -161,7 +161,7 @@ bool Item_func::agg_arg_charsets(DTCollation &coll,
   }
 
   THD *thd= current_thd;
-  Item_arena *arena, backup;
+  Query_arena *arena, backup;
   bool res= FALSE;
   /*
     In case we're in statement prepare, create conversion item
@@ -879,11 +879,11 @@ longlong Item_func_numhybrid::val_int()
     return (longlong)real_op();
   case STRING_RESULT:
   {
-    char *end_not_used;
     int err_not_used;
     String *res= str_op(&str_value);
+    char *end= (char*) res->ptr() + res->length();
     CHARSET_INFO *cs= str_value.charset();
-    return (res ? (*(cs->cset->strtoll10))(cs, res->ptr(), &end_not_used,
+    return (res ? (*(cs->cset->strtoll10))(cs, res->ptr(), &end,
                                            &err_not_used) : 0);
   }
   default:
@@ -1022,7 +1022,8 @@ longlong Item_func_unsigned::val_int()
 String *Item_decimal_typecast::val_str(String *str)
 {
   my_decimal tmp_buf, *tmp= val_decimal(&tmp_buf);
-  my_decimal_round(E_DEC_FATAL_ERROR, tmp, decimals, FALSE, &tmp_buf);
+  if (null_value)
+    return NULL;
   my_decimal2string(E_DEC_FATAL_ERROR, &tmp_buf, 0, 0, 0, str);
   return str;
 }
@@ -1032,6 +1033,8 @@ double Item_decimal_typecast::val_real()
 {
   my_decimal tmp_buf, *tmp= val_decimal(&tmp_buf);
   double res;
+  if (null_value)
+    return 0.0;
   my_decimal2double(E_DEC_FATAL_ERROR, tmp, &res);
   return res;
 }
@@ -1041,6 +1044,8 @@ longlong Item_decimal_typecast::val_int()
 {
   my_decimal tmp_buf, *tmp= val_decimal(&tmp_buf);
   longlong res;
+  if (null_value)
+    return 0;
   my_decimal2int(E_DEC_FATAL_ERROR, tmp, unsigned_flag, &res);
   return res;
 }
@@ -1049,11 +1054,21 @@ longlong Item_decimal_typecast::val_int()
 my_decimal *Item_decimal_typecast::val_decimal(my_decimal *dec)
 {
   my_decimal tmp_buf, *tmp= args[0]->val_decimal(&tmp_buf);
+  if ((null_value= args[0]->null_value))
+    return NULL;
   my_decimal_round(E_DEC_FATAL_ERROR, tmp, decimals, FALSE, dec);
   return dec;
 }
 
 
+void Item_decimal_typecast::print(String *str)
+{
+  str->append("cast(", 5);
+  args[0]->print(str);
+  str->append(" as decimal)", 12);
+}
+
+
 double Item_func_plus::real_op()
 {
   double value= args[0]->val_real() + args[1]->val_real();
@@ -4104,7 +4119,7 @@ bool Item_func_get_user_var::eq(const Item *item, bool binary_cmp) const
     return 1;					// Same item is same.
   /* Check if other type is also a get_user_var() object */
   if (item->type() != FUNC_ITEM ||
-      ((Item_func*) item)->func_name() != func_name())
+      ((Item_func*) item)->functype() != functype())
     return 0;
   Item_func_get_user_var *other=(Item_func_get_user_var*) item;
   return (name.length == other->name.length &&
@@ -4605,7 +4620,7 @@ Item *get_system_var(THD *thd, enum_var_type var_type, const char *var_name,
   if (!(item=var->item(thd, var_type, &null_lex_string)))
     return 0;						// Impossible
   thd->lex->uncacheable(UNCACHEABLE_SIDEEFFECT);
-  item->set_name(item_name, 0, system_charset_info);	// Will use original name
+  item->set_name(item_name, 0, system_charset_info);  // Will use original name
   return item;
 }
 
@@ -4792,42 +4807,36 @@ Item_func_sp::execute(Item **itp)
   DBUG_ENTER("Item_func_sp::execute");
   THD *thd= current_thd;
   ulong old_client_capabilites;
-  int res;
+  int res= -1;
   bool save_in_sub_stmt= thd->transaction.in_sub_stmt;
+  my_bool nsok;
 #ifndef NO_EMBEDDED_ACCESS_CHECKS
   st_sp_security_context save_ctx;
 #endif
 
-  if (! m_sp)
+  if (! m_sp && ! (m_sp= sp_find_function(thd, m_name, TRUE)))
   {
-    if (!(m_sp= sp_find_function(thd, m_name, TRUE)))
-    {
-      my_error(ER_SP_DOES_NOT_EXIST, MYF(0), "FUNCTION", m_name->m_qname.str);
-      DBUG_RETURN(-1);
-    }
+    my_error(ER_SP_DOES_NOT_EXIST, MYF(0), "FUNCTION", m_name->m_qname.str);
+    goto error;
   }
 
   old_client_capabilites= thd->client_capabilities;
   thd->client_capabilities &= ~CLIENT_MULTI_RESULTS;
 
 #ifndef EMBEDDED_LIBRARY
-  my_bool nsok= thd->net.no_send_ok;
+  nsok= thd->net.no_send_ok;
   thd->net.no_send_ok= TRUE;
 #endif
 
 #ifndef NO_EMBEDDED_ACCESS_CHECKS
   if (check_routine_access(thd, EXECUTE_ACL, 
 			   m_sp->m_db.str, m_sp->m_name.str, 0, 0))
-      DBUG_RETURN(-1);
+    goto error_check;
   sp_change_security_context(thd, m_sp, &save_ctx);
   if (save_ctx.changed && 
       check_routine_access(thd, EXECUTE_ACL, 
 			   m_sp->m_db.str, m_sp->m_name.str, 0, 0))
-  {
-    sp_restore_security_context(thd, m_sp, &save_ctx);
-    thd->client_capabilities|= old_client_capabilites &  CLIENT_MULTI_RESULTS;
-    DBUG_RETURN(-1);
-  }
+    goto error_check;
 #endif
   /*
     Like for SPs, we don't binlog the substatements. If the statement which
@@ -4835,6 +4844,7 @@ Item_func_sp::execute(Item **itp)
     it's not (e.g. SELECT myfunc()) it won't be binlogged (documented known
     problem).
   */
+
   tmp_disable_binlog(thd); /* don't binlog the substatements */
   thd->transaction.in_sub_stmt= TRUE;
 
@@ -4849,16 +4859,21 @@ Item_func_sp::execute(Item **itp)
                  ER_FAILED_ROUTINE_BREAK_BINLOG,
 		 ER(ER_FAILED_ROUTINE_BREAK_BINLOG));
 
+error_check_ctx:
 #ifndef NO_EMBEDDED_ACCESS_CHECKS
   sp_restore_security_context(thd, m_sp, &save_ctx);
 #endif
 
+  thd->client_capabilities|= old_client_capabilites &  CLIENT_MULTI_RESULTS;
+
+error_check:
 #ifndef EMBEDDED_LIBRARY
   thd->net.no_send_ok= nsok;
 #endif
 
   thd->client_capabilities|= old_client_capabilites &  CLIENT_MULTI_RESULTS;
 
+error:
   DBUG_RETURN(res);
 }
 
diff --git a/sql/item_func.h b/sql/item_func.h
index 1ac1449760f..e0f14ceac75 100644
--- a/sql/item_func.h
+++ b/sql/item_func.h
@@ -54,7 +54,8 @@ public:
 		  SP_POINTN,SP_GEOMETRYN,SP_INTERIORRINGN,
                   NOT_FUNC, NOT_ALL_FUNC,
                   NOW_FUNC, TRIG_COND_FUNC,
-                  GUSERVAR_FUNC};
+                  GUSERVAR_FUNC, COLLATE_FUNC,
+                  EXTRACT_FUNC, CHAR_TYPECAST_FUNC };
   enum optimize_type { OPTIMIZE_NONE,OPTIMIZE_KEY,OPTIMIZE_OP, OPTIMIZE_NULL,
                        OPTIMIZE_EQUAL };
   enum Type type() const { return FUNC_ITEM; }
@@ -123,7 +124,17 @@ public:
   virtual optimize_type select_optimize() const { return OPTIMIZE_NONE; }
   virtual bool have_rev_func() const { return 0; }
   virtual Item *key_item() const { return args[0]; }
-  virtual const char *func_name() const { return "?"; }
+  /*
+    This method is used for debug purposes to print the name of an
+    item to the debug log. The second use of this method is as
+    a helper function of print(), where it is applicable.
+    To suit both goals it should return a meaningful,
+    distinguishable and sintactically correct string.  This method
+    should not be used for runtime type identification, use enum
+    {Sum}Functype and Item_func::functype()/Item_sum::sum_func()
+    instead.
+  */
+  virtual const char *func_name() const= 0;
   virtual bool const_item() const { return const_item_cache; }
   inline Item **arguments() const { return args; }
   void set_arguments(List<Item> &list);
@@ -306,6 +317,8 @@ public:
   enum Item_result result_type () const { return DECIMAL_RESULT; }
   enum_field_types field_type() const { return MYSQL_TYPE_NEWDECIMAL; }
   void fix_length_and_dec() {};
+  const char *func_name() const { return "decimal_typecast"; }
+  void print(String *);
 };
 
 
@@ -506,7 +519,7 @@ public:
 
 class Item_func_acos :public Item_dec_func
 {
- public:
+public:
   Item_func_acos(Item *a) :Item_dec_func(a) {}
   double val_real();
   const char *func_name() const { return "acos"; }
@@ -514,7 +527,7 @@ class Item_func_acos :public Item_dec_func
 
 class Item_func_asin :public Item_dec_func
 {
- public:
+public:
   Item_func_asin(Item *a) :Item_dec_func(a) {}
   double val_real();
   const char *func_name() const { return "asin"; }
@@ -522,7 +535,7 @@ class Item_func_asin :public Item_dec_func
 
 class Item_func_atan :public Item_dec_func
 {
- public:
+public:
   Item_func_atan(Item *a) :Item_dec_func(a) {}
   Item_func_atan(Item *a,Item *b) :Item_dec_func(a,b) {}
   double val_real();
@@ -531,7 +544,7 @@ class Item_func_atan :public Item_dec_func
 
 class Item_func_cos :public Item_dec_func
 {
- public:
+public:
   Item_func_cos(Item *a) :Item_dec_func(a) {}
   double val_real();
   const char *func_name() const { return "cos"; }
@@ -539,7 +552,7 @@ class Item_func_cos :public Item_dec_func
 
 class Item_func_sin :public Item_dec_func
 {
- public:
+public:
   Item_func_sin(Item *a) :Item_dec_func(a) {}
   double val_real();
   const char *func_name() const { return "sin"; }
@@ -547,7 +560,7 @@ class Item_func_sin :public Item_dec_func
 
 class Item_func_tan :public Item_dec_func
 {
- public:
+public:
   Item_func_tan(Item *a) :Item_dec_func(a) {}
   double val_real();
   const char *func_name() const { return "tan"; }
@@ -634,7 +647,7 @@ class Item_func_units :public Item_real_func
 {
   char *name;
   double mul,add;
- public:
+public:
   Item_func_units(char *name_arg,Item *a,double mul_arg,double add_arg)
     :Item_real_func(a),name(name_arg),mul(mul_arg),add(add_arg) {}
   double val_real();
@@ -853,7 +866,7 @@ public:
 class Item_func_benchmark :public Item_int_func
 {
   ulong loop_count;
- public:
+public:
   Item_func_benchmark(ulong loop_count_arg,Item *expr)
     :Item_int_func(expr), loop_count(loop_count_arg)
   {}
@@ -868,7 +881,7 @@ class Item_func_benchmark :public Item_int_func
 
 class Item_udf_func :public Item_func
 {
- protected:
+protected:
   udf_handler udf;
 
 public:
@@ -1046,7 +1059,7 @@ class Item_func_get_lock :public Item_int_func
 class Item_func_release_lock :public Item_int_func
 {
   String value;
- public:
+public:
   Item_func_release_lock(Item *a) :Item_int_func(a) {}
   longlong val_int();
   const char *func_name() const { return "release_lock"; }
@@ -1058,7 +1071,7 @@ class Item_func_release_lock :public Item_int_func
 class Item_master_pos_wait :public Item_int_func
 {
   String value;
- public:
+public:
   Item_master_pos_wait(Item *a,Item *b) :Item_int_func(a,b) {}
   Item_master_pos_wait(Item *a,Item *b,Item *c) :Item_int_func(a,b,c) {}
   longlong val_int();
diff --git a/sql/item_strfunc.cc b/sql/item_strfunc.cc
index 539bed58e66..06239de1f99 100644
--- a/sql/item_strfunc.cc
+++ b/sql/item_strfunc.cc
@@ -363,6 +363,7 @@ String *Item_func_des_encrypt::val_str(String *str)
 {
   DBUG_ASSERT(fixed == 1);
 #ifdef HAVE_OPENSSL
+  uint code= ER_WRONG_PARAMETERS_TO_PROCEDURE;
   DES_cblock ivec;
   struct st_des_keyblock keyblock;
   struct st_des_keyschedule keyschedule;
@@ -371,7 +372,7 @@ String *Item_func_des_encrypt::val_str(String *str)
   String *res= args[0]->val_str(str);
 
   if ((null_value=args[0]->null_value))
-    return 0;
+    goto error;
   if ((res_length=res->length()) == 0)
     return &my_empty_string;
 
@@ -419,6 +420,7 @@ String *Item_func_des_encrypt::val_str(String *str)
 
   tail=  (8-(res_length) % 8);			// 1..8 marking extra length
   res_length+=tail;
+  code= ER_OUT_OF_RESOURCES;
   if (tail && res->append(append_str, tail) || tmp_value.alloc(res_length+1))
     goto error;
   (*res)[res_length-1]=tail;			// save extra length
@@ -436,6 +438,13 @@ String *Item_func_des_encrypt::val_str(String *str)
   return &tmp_value;
 
 error:
+  push_warning_printf(current_thd,MYSQL_ERROR::WARN_LEVEL_ERROR,
+                          code, ER(code),
+                          "des_encrypt");
+#else
+  push_warning_printf(current_thd,MYSQL_ERROR::WARN_LEVEL_ERROR,
+                      ER_FEATURE_DISABLED, ER(ER_FEATURE_DISABLED),
+                      "des_encrypt","--with-openssl");
 #endif	/* HAVE_OPENSSL */
   null_value=1;
   return 0;
@@ -446,6 +455,7 @@ String *Item_func_des_decrypt::val_str(String *str)
 {
   DBUG_ASSERT(fixed == 1);
 #ifdef HAVE_OPENSSL
+  uint code= ER_WRONG_PARAMETERS_TO_PROCEDURE;
   DES_key_schedule ks1, ks2, ks3;
   DES_cblock ivec;
   struct st_des_keyblock keyblock;
@@ -454,7 +464,7 @@ String *Item_func_des_decrypt::val_str(String *str)
   uint length=res->length(),tail;
 
   if ((null_value=args[0]->null_value))
-    return 0;
+    goto error;
   length=res->length();
   if (length < 9 || (length % 8) != 1 || !((*res)[0] & 128))
     return res;				// Skip decryption if not encrypted
@@ -485,6 +495,7 @@ String *Item_func_des_decrypt::val_str(String *str)
     DES_set_key_unchecked(&keyblock.key2,&keyschedule.ks2);
     DES_set_key_unchecked(&keyblock.key3,&keyschedule.ks3);
   }
+  code= ER_OUT_OF_RESOURCES;
   if (tmp_value.alloc(length-1))
     goto error;
 
@@ -498,11 +509,19 @@ String *Item_func_des_decrypt::val_str(String *str)
 		       &ivec, FALSE);
   /* Restore old length of key */
   if ((tail=(uint) (uchar) tmp_value[length-2]) > 8)
-    goto error;					// Wrong key
+    goto wrong_key;				     // Wrong key
   tmp_value.length(length-1-tail);
   return &tmp_value;
 
 error:
+  push_warning_printf(current_thd,MYSQL_ERROR::WARN_LEVEL_ERROR,
+                          code, ER(code),
+                          "des_decrypt");
+wrong_key:
+#else
+  push_warning_printf(current_thd,MYSQL_ERROR::WARN_LEVEL_ERROR,
+                      ER_FEATURE_DISABLED, ER(ER_FEATURE_DISABLED),
+                      "des_decrypt","--with-openssl");
 #endif	/* HAVE_OPENSSL */
   null_value=1;
   return 0;
@@ -2297,7 +2316,7 @@ bool Item_func_set_collation::eq(const Item *item, bool binary_cmp) const
     return 0;
   Item_func *item_func=(Item_func*) item;
   if (arg_count != item_func->arg_count ||
-      func_name() != item_func->func_name())
+      functype() != item_func->functype())
     return 0;
   Item_func_set_collation *item_func_sc=(Item_func_set_collation*) item;
   if (collation.collation != item_func_sc->collation.collation)
diff --git a/sql/item_strfunc.h b/sql/item_strfunc.h
index 6df90cebdff..c4beb3b08cb 100644
--- a/sql/item_strfunc.h
+++ b/sql/item_strfunc.h
@@ -322,7 +322,7 @@ public:
   Item_func_encrypt(Item *a, Item *b): Item_str_func(a,b) {}
   String *val_str(String *);
   void fix_length_and_dec() { maybe_null=1; max_length = 13; }
-  const char *func_name() const { return "ecrypt"; }
+  const char *func_name() const { return "encrypt"; }
 };
 
 #include "sql_crypt.h"
@@ -573,6 +573,7 @@ public:
     max_length=args[0]->max_length;
   }
   void print(String *str);
+  const char *func_name() const { return "cast_as_binary"; }
 };
 
 
@@ -648,6 +649,7 @@ public:
   void fix_length_and_dec();
   bool eq(const Item *item, bool binary_cmp) const;
   const char *func_name() const { return "collate"; }
+  enum Functype func_type() const { return COLLATE_FUNC; }
   void print(String *str);
   Item_field *filed_for_view_update()
   {
diff --git a/sql/item_subselect.cc b/sql/item_subselect.cc
index 7a72b78b6f4..c7587686ecd 100644
--- a/sql/item_subselect.cc
+++ b/sql/item_subselect.cc
@@ -194,15 +194,8 @@ bool Item_subselect::fix_fields(THD *thd_param, TABLE_LIST *tables, Item **ref)
 bool Item_subselect::exec()
 {
   int res;
-  MEM_ROOT *old_root= thd->mem_root;
 
-  /*
-    As this is execution, all objects should be allocated through the main
-    mem root
-  */
-  thd->mem_root= &thd->main_mem_root;
   res= engine->exec();
-  thd->mem_root= old_root;
 
   if (engine_changed)
   {
@@ -340,7 +333,7 @@ Item_singlerow_subselect::select_transformer(JOIN *join)
     return RES_OK;
 
   SELECT_LEX *select_lex= join->select_lex;
-  Item_arena *arena= thd->current_arena;
+  Query_arena *arena= thd->current_arena;
  
   if (!select_lex->master_unit()->first_select()->next_select() &&
       !select_lex->table_list.elements &&
@@ -1164,7 +1157,7 @@ Item_in_subselect::select_transformer(JOIN *join)
 Item_subselect::trans_res
 Item_in_subselect::select_in_like_transformer(JOIN *join, Comp_creator *func)
 {
-  Item_arena *arena, backup;
+  Query_arena *arena, backup;
   SELECT_LEX *current= thd->lex->current_select, *up;
   const char *save_where= thd->where;
   Item_subselect::trans_res res= RES_ERROR;
@@ -1487,7 +1480,7 @@ int subselect_uniquesubquery_engine::exec()
   TABLE *table= tab->table;
   for (store_key **copy=tab->ref.key_copy ; *copy ; copy++)
   {
-    if (tab->ref.key_err= (*copy)->copy())
+    if ((tab->ref.key_err= (*copy)->copy()) & 1)
     {
       table->status= STATUS_NOT_FOUND;
       DBUG_RETURN(1);
@@ -1540,7 +1533,7 @@ int subselect_indexsubquery_engine::exec()
 
   for (store_key **copy=tab->ref.key_copy ; *copy ; copy++)
   {
-    if (tab->ref.key_err= (*copy)->copy())
+    if ((tab->ref.key_err= (*copy)->copy()) & 1)
     {
       table->status= STATUS_NOT_FOUND;
       DBUG_RETURN(1);
diff --git a/sql/item_sum.cc b/sql/item_sum.cc
index f7a158ceb5a..76f94801b49 100644
--- a/sql/item_sum.cc
+++ b/sql/item_sum.cc
@@ -86,7 +86,6 @@ void Item_sum::make_field(Send_field *tmp_field)
 void Item_sum::print(String *str)
 {
   str->append(func_name());
-  str->append('(');
   for (uint i=0 ; i < arg_count ; i++)
   {
     if (i)
@@ -2425,13 +2424,6 @@ longlong Item_sum_count_distinct::val_int()
 }
 
 
-void Item_sum_count_distinct::print(String *str)
-{
-  str->append("count(distinct ", 15);
-  args[0]->print(str);
-  str->append(')');
-}
-
 /****************************************************************************
 ** Functions to handle dynamic loadable aggregates
 ** Original source by: Alexis Mikhailov <root@medinf.chuvashia.su>
@@ -2466,6 +2458,20 @@ void Item_udf_sum::cleanup()
 }
 
 
+void Item_udf_sum::print(String *str)
+{
+  str->append(func_name());
+  str->append('(');
+  for (uint i=0 ; i < arg_count ; i++)
+  {
+    if (i)
+      str->append(',');
+    args[i]->print(str);
+  }
+  str->append(')');
+}
+
+
 Item *Item_sum_udf_float::copy_or_same(THD* thd)
 {
   return new (thd->mem_root) Item_sum_udf_float(thd, this);
diff --git a/sql/item_sum.h b/sql/item_sum.h
index bb5d31b4b4f..b9a90ee5de5 100644
--- a/sql/item_sum.h
+++ b/sql/item_sum.h
@@ -81,7 +81,22 @@ public:
   virtual void update_field()=0;
   virtual bool keep_field_type(void) const { return 0; }
   virtual void fix_length_and_dec() { maybe_null=1; null_value=1; }
-  virtual const char *func_name() const { return "?"; }
+  /*
+    This method is used for debug purposes to print the name of an
+    item to the debug log. The second use of this method is as
+    a helper function of print(), where it is applicable.
+    To suit both goals it should return a meaningful,
+    distinguishable and sintactically correct string.  This method
+    should not be used for runtime type identification, use enum
+    {Sum}Functype and Item_func::functype()/Item_sum::sum_func()
+    instead.
+
+    NOTE: for Items inherited from Item_sum, func_name() return part of
+    function name till first argument (including '(') to make difference in
+    names for functions with 'distinct' clause and without 'distinct' and
+    also to make printing of items inherited from Item_sum uniform.
+  */
+  virtual const char *func_name() const= 0;
   virtual Item *result_item(Field *field)
     { return new Item_field(field);}
   table_map used_tables() const { return ~(table_map) 0; } /* Not used */
@@ -159,7 +174,7 @@ public:
   void reset_field();
   void update_field();
   void no_rows_in_result() {}
-  const char *func_name() const { return "sum"; }
+  const char *func_name() const { return "sum("; }
   Item *copy_or_same(THD* thd);
 };
 
@@ -200,7 +215,6 @@ public:
   enum Sumfunctype sum_func () const { return SUM_DISTINCT_FUNC; }
   void reset_field() {} // not used
   void update_field() {} // not used
-  const char *func_name() const { return "sum_distinct"; }
   virtual void no_rows_in_result() {}
   void fix_length_and_dec();
   enum Item_result result_type () const { return val.traits->type(); }
@@ -224,7 +238,7 @@ public:
   Item_sum_sum_distinct(Item *item_arg) :Item_sum_distinct(item_arg) {}
 
   enum Sumfunctype sum_func () const { return SUM_DISTINCT_FUNC; }
-  const char *func_name() const { return "sum_distinct"; }
+  const char *func_name() const { return "sum(distinct "; }
   Item *copy_or_same(THD* thd) { return new Item_sum_sum_distinct(thd, this); }
 };
 
@@ -243,7 +257,7 @@ public:
   void fix_length_and_dec();
   virtual void calculate_val_and_count();
   enum Sumfunctype sum_func () const { return AVG_DISTINCT_FUNC; }
-  const char *func_name() const { return "avg_distinct"; }
+  const char *func_name() const { return "avg(distinct "; }
   Item *copy_or_same(THD* thd) { return new Item_sum_avg_distinct(thd, this); }
 };
 
@@ -272,7 +286,7 @@ class Item_sum_count :public Item_sum_int
   void reset_field();
   void cleanup();
   void update_field();
-  const char *func_name() const { return "count"; }
+  const char *func_name() const { return "count("; }
   Item *copy_or_same(THD* thd);
 };
 
@@ -326,12 +340,11 @@ public:
   longlong val_int();
   void reset_field() { return ;}		// Never called
   void update_field() { return ; }		// Never called
-  const char *func_name() const { return "count_distinct"; }
+  const char *func_name() const { return "count(distinct "; }
   bool setup(THD *thd);
   void make_unique();
   Item *copy_or_same(THD* thd);
   void no_rows_in_result() {}
-  void print(String *str);
 };
 
 
@@ -389,7 +402,7 @@ public:
   Item *result_item(Field *field)
   { return new Item_avg_field(hybrid_type, this); }
   void no_rows_in_result() {}
-  const char *func_name() const { return "avg"; }
+  const char *func_name() const { return "avg("; }
   Item *copy_or_same(THD* thd);
   Field *create_tmp_field(bool group, TABLE *table, uint convert_blob_length);
 };
@@ -466,7 +479,7 @@ public:
   Item *result_item(Field *field)
   { return new Item_variance_field(this); }
   void no_rows_in_result() {}
-  const char *func_name() const { return "variance"; }
+  const char *func_name() const { return "variance("; }
   Item *copy_or_same(THD* thd);
   Field *create_tmp_field(bool group, TABLE *table, uint convert_blob_length);
   enum Item_result result_type () const { return hybrid_type; }
@@ -501,7 +514,7 @@ class Item_sum_std :public Item_sum_variance
   double val_real();
   Item *result_item(Field *field)
     { return new Item_std_field(this); }
-  const char *func_name() const { return "std"; }
+  const char *func_name() const { return "std("; }
   Item *copy_or_same(THD* thd);
   enum Item_result result_type () const { return REAL_RESULT; }
   enum_field_types field_type() const { return MYSQL_TYPE_DOUBLE;}
@@ -565,7 +578,7 @@ public:
   enum Sumfunctype sum_func () const {return MIN_FUNC;}
 
   bool add();
-  const char *func_name() const { return "min"; }
+  const char *func_name() const { return "min("; }
   Item *copy_or_same(THD* thd);
 };
 
@@ -578,7 +591,7 @@ public:
   enum Sumfunctype sum_func () const {return MAX_FUNC;}
 
   bool add();
-  const char *func_name() const { return "max"; }
+  const char *func_name() const { return "max("; }
   Item *copy_or_same(THD* thd);
 };
 
@@ -609,7 +622,7 @@ public:
   Item_sum_or(Item *item_par) :Item_sum_bit(item_par,LL(0)) {}
   Item_sum_or(THD *thd, Item_sum_or *item) :Item_sum_bit(thd, item) {}
   bool add();
-  const char *func_name() const { return "bit_or"; }
+  const char *func_name() const { return "bit_or("; }
   Item *copy_or_same(THD* thd);
 };
 
@@ -620,7 +633,7 @@ class Item_sum_and :public Item_sum_bit
   Item_sum_and(Item *item_par) :Item_sum_bit(item_par, ULONGLONG_MAX) {}
   Item_sum_and(THD *thd, Item_sum_and *item) :Item_sum_bit(thd, item) {}
   bool add();
-  const char *func_name() const { return "bit_and"; }
+  const char *func_name() const { return "bit_and("; }
   Item *copy_or_same(THD* thd);
 };
 
@@ -630,7 +643,7 @@ class Item_sum_xor :public Item_sum_bit
   Item_sum_xor(Item *item_par) :Item_sum_bit(item_par,LL(0)) {}
   Item_sum_xor(THD *thd, Item_sum_xor *item) :Item_sum_bit(thd, item) {}
   bool add();
-  const char *func_name() const { return "bit_xor"; }
+  const char *func_name() const { return "bit_xor("; }
   Item *copy_or_same(THD* thd);
 };
 
@@ -668,6 +681,7 @@ public:
   void reset_field() {};
   void update_field() {};
   void cleanup();
+  void print(String *str);
 };
 
 
diff --git a/sql/item_timefunc.cc b/sql/item_timefunc.cc
index 564c5e4b9cc..19386c15835 100644
--- a/sql/item_timefunc.cc
+++ b/sql/item_timefunc.cc
@@ -2158,7 +2158,7 @@ bool Item_extract::eq(const Item *item, bool binary_cmp) const
   if (this == item)
     return 1;
   if (item->type() != FUNC_ITEM ||
-      func_name() != ((Item_func*)item)->func_name())
+      functype() != ((Item_func*)item)->functype())
     return 0;
 
   Item_extract* ie= (Item_extract*)item;
@@ -2176,7 +2176,7 @@ bool Item_char_typecast::eq(const Item *item, bool binary_cmp) const
   if (this == item)
     return 1;
   if (item->type() != FUNC_ITEM ||
-      func_name() != ((Item_func*)item)->func_name())
+      functype() != ((Item_func*)item)->functype())
     return 0;
 
   Item_char_typecast *cast= (Item_char_typecast*)item;
diff --git a/sql/item_timefunc.h b/sql/item_timefunc.h
index c8fb2b39836..a6dd9f7da91 100644
--- a/sql/item_timefunc.h
+++ b/sql/item_timefunc.h
@@ -503,7 +503,8 @@ public:
   Item_func_date_format(Item *a,Item *b,bool is_time_format_arg)
     :Item_str_func(a,b),is_time_format(is_time_format_arg) {}
   String *val_str(String *str);
-  const char *func_name() const { return "date_format"; }
+  const char *func_name() const
+    { return is_time_format ? "time_format" : "date_format"; }
   void fix_length_and_dec();
   uint format_length(const String *format);
 };
@@ -637,6 +638,7 @@ class Item_extract :public Item_int_func
   Item_extract(interval_type type_arg, Item *a)
     :Item_int_func(a), int_type(type_arg) {}
   longlong val_int();
+  enum Functype functype() const { return EXTRACT_FUNC; }
   const char *func_name() const { return "extract"; }
   void fix_length_and_dec();
   bool eq(const Item *item, bool binary_cmp) const;
@@ -689,6 +691,7 @@ class Item_char_typecast :public Item_typecast
 public:
   Item_char_typecast(Item *a, int length_arg, CHARSET_INFO *cs_arg)
     :Item_typecast(a), cast_length(length_arg), cast_cs(cs_arg) {}
+  enum Functype functype() const { return CHAR_TYPECAST_FUNC; }
   bool eq(const Item *item, bool binary_cmp) const;
   const char *func_name() const { return "cast_as_char"; }
   const char* cast_type() const { return "char"; };
@@ -790,6 +793,7 @@ public:
     return (new Field_string(max_length, maybe_null, name, t_arg, &my_charset_bin));
   }
   void print(String *str);
+  const char *func_name() const { return "add_time"; }
 };
 
 class Item_func_timediff :public Item_str_func
diff --git a/sql/item_uniq.h b/sql/item_uniq.h
index 14b2e8d53bb..e95aa35101e 100644
--- a/sql/item_uniq.h
+++ b/sql/item_uniq.h
@@ -30,6 +30,7 @@ public:
   double val_real() { DBUG_ASSERT(fixed == 1); return 0.0; }
   void fix_length_and_dec() { decimals=0; max_length=6; }
   void print(String *str) { str->append("0.0", 3); }
+  const char *func_name() const { return "unique_users"; }
 };
 
 
@@ -58,4 +59,5 @@ public:
   }
   void print(String *str) { str->append("0.0", 3); }
   Field *create_tmp_field(bool group, TABLE *table, uint convert_blob_length);
+  const char *func_name() const { return "sum_unique_users"; }
 };
diff --git a/sql/mysql_priv.h b/sql/mysql_priv.h
index b6f4b963393..805af08b76a 100644
--- a/sql/mysql_priv.h
+++ b/sql/mysql_priv.h
@@ -325,6 +325,7 @@ extern CHARSET_INFO *national_charset_info, *table_alias_charset;
 #define MODE_TRADITIONAL		(MODE_ERROR_FOR_DIVISION_BY_ZERO*2)
 #define MODE_NO_AUTO_CREATE_USER	(MODE_TRADITIONAL*2)
 #define MODE_HIGH_NOT_PRECEDENCE	(MODE_NO_AUTO_CREATE_USER*2)
+#define MODE_NO_ENGINE_SUBSTITUTION     (MODE_HIGH_NOT_PRECEDENCE*2)
 /*
   Replication uses 8 bytes to store SQL_MODE in the binary log. The day you
   use strictly more than 64 bits by adding one more define above, you should
@@ -559,8 +560,6 @@ struct Query_cache_query_flags
 #define query_cache_invalidate_by_MyISAM_filename_ref NULL
 #endif /*HAVE_QUERY_CACHE*/
 
-#define prepare_execute(A) ((A)->command == COM_EXECUTE)
-
 bool mysql_create_db(THD *thd, char *db, HA_CREATE_INFO *create, bool silent);
 bool mysql_alter_db(THD *thd, const char *db, HA_CREATE_INFO *create);
 bool mysql_rm_db(THD *thd,char *db,bool if_exists, bool silent);
@@ -599,6 +598,7 @@ bool mysql_execute_command(THD *thd);
 bool do_command(THD *thd);
 bool dispatch_command(enum enum_server_command command, THD *thd,
 		      char* packet, uint packet_length);
+void log_slow_statement(THD *thd);
 bool check_dup(const char *db, const char *name, TABLE_LIST *tables);
 
 bool table_cache_init(void);
@@ -667,6 +667,7 @@ int mysql_derived_filling(THD *thd, LEX *lex, TABLE_LIST *t);
 Field *create_tmp_field(THD *thd, TABLE *table,Item *item, Item::Type type,
 			Item ***copy_func, Field **from_field,
 			bool group, bool modify_item,
+			bool table_cant_handle_bit_fields,
                         uint convert_blob_length);
 void sp_prepare_create_field(THD *thd, create_field *sql_field);
 int prepare_create_field(create_field *sql_field, 
@@ -841,7 +842,7 @@ bool mysql_stmt_prepare(THD *thd, char *packet, uint packet_length,
 void mysql_stmt_execute(THD *thd, char *packet, uint packet_length);
 void mysql_sql_stmt_execute(THD *thd, LEX_STRING *stmt_name);
 void mysql_stmt_fetch(THD *thd, char *packet, uint packet_length);
-void mysql_stmt_free(THD *thd, char *packet);
+void mysql_stmt_close(THD *thd, char *packet);
 void mysql_stmt_reset(THD *thd, char *packet);
 void mysql_stmt_get_longdata(THD *thd, char *pos, ulong packet_length);
 void reinit_stmt_before_use(THD *thd, LEX *lex);
@@ -894,8 +895,7 @@ bool get_key_map_from_key_list(key_map *map, TABLE *table,
                                List<String> *index_list);
 bool insert_fields(THD *thd,TABLE_LIST *tables,
 		   const char *db_name, const char *table_name,
-		   List_iterator<Item> *it, bool any_privileges,
-                   bool allocate_view_names);
+		   List_iterator<Item> *it, bool any_privileges);
 bool setup_tables(THD *thd, TABLE_LIST *tables, Item **conds,
 		  TABLE_LIST **leaves, bool select_insert);
 int setup_wild(THD *thd, TABLE_LIST *tables, List<Item> &fields,
@@ -1116,6 +1116,7 @@ extern my_bool opt_slave_compressed_protocol, use_temp_pool;
 extern my_bool opt_readonly, lower_case_file_system;
 extern my_bool opt_enable_named_pipe, opt_sync_frm, opt_allow_suspicious_udfs;
 extern my_bool opt_secure_auth;
+extern my_bool opt_log_slow_admin_statements;
 extern my_bool sp_automatic_privileges, opt_noacl;
 extern my_bool opt_old_style_user_limits, trust_routine_creators;
 extern uint opt_crash_binlog_innodb;
@@ -1230,7 +1231,7 @@ int openfrm(THD *thd, const char *name,const char *alias,uint filestat,
 int readfrm(const char *name, const void** data, uint* length);
 int writefrm(const char* name, const void* data, uint len);
 int closefrm(TABLE *table);
-db_type get_table_type(const char *name);
+db_type get_table_type(THD *thd, const char *name);
 int read_string(File file, gptr *to, uint length);
 void free_blobs(TABLE *table);
 int set_zone(int nr,int min_zone,int max_zone);
@@ -1287,7 +1288,7 @@ ulong make_new_entry(File file,uchar *fileinfo,TYPELIB *formnames,
 		     const char *newname);
 ulong next_io_size(ulong pos);
 void append_unescaped(String *res, const char *pos, uint length);
-int create_frm(char *name,uint reclength,uchar *fileinfo,
+int create_frm(THD *thd, char *name,uint reclength,uchar *fileinfo,
 	       HA_CREATE_INFO *create_info, uint keys);
 void update_create_info_from_table(HA_CREATE_INFO *info, TABLE *form);
 int rename_file_ext(const char * from,const char * to,const char * ext);
diff --git a/sql/mysqld.cc b/sql/mysqld.cc
index f9bf5a1a7c3..bc4cc81506e 100644
--- a/sql/mysqld.cc
+++ b/sql/mysqld.cc
@@ -140,6 +140,7 @@ int deny_severity = LOG_WARNING;
 #define zVOLSTATE_DEACTIVE 2
 #define zVOLSTATE_MAINTENANCE 3
 
+#include <nks/netware.h>
 #include <nks/vm.h>
 #include <library.h>
 #include <monitor.h>
@@ -229,6 +230,7 @@ static const char *sql_mode_names[] =
   "NO_AUTO_VALUE_ON_ZERO", "NO_BACKSLASH_ESCAPES", "STRICT_TRANS_TABLES", "STRICT_ALL_TABLES",
   "NO_ZERO_IN_DATE", "NO_ZERO_DATE", "ALLOW_INVALID_DATES", "ERROR_FOR_DIVISION_BY_ZERO",
   "TRADITIONAL", "NO_AUTO_CREATE_USER", "HIGH_NOT_PRECEDENCE",
+  "NO_ENGINE_SUBSTITUTION",
   NullS
 };
 TYPELIB sql_mode_typelib= { array_elements(sql_mode_names)-1,"",
@@ -326,6 +328,7 @@ ulong opt_ndb_nodeid;
 my_bool opt_readonly, use_temp_pool, relay_log_purge;
 my_bool opt_sync_frm, opt_allow_suspicious_udfs;
 my_bool opt_secure_auth= 0;
+my_bool opt_log_slow_admin_statements= 0;
 my_bool lower_case_file_system= 0;
 my_bool opt_large_pages= 0;
 uint    opt_large_page_size= 0;
@@ -1276,19 +1279,6 @@ static void server_init(void)
   int	arg=1;
   DBUG_ENTER("server_init");
 
-#ifdef	__WIN__
-  if (!opt_disable_networking)
-  {
-    WSADATA WsaData;
-    if (SOCKET_ERROR == WSAStartup (0x0101, &WsaData))
-    {
-      /* errors are not read yet, so we use test here */
-      my_message(ER_WSAS_FAILED, "WSAStartup Failed", MYF(0));
-      unireg_abort(1);
-    }
-  }
-#endif /* __WIN__ */
-
   set_ports();
 
   if (mysqld_port != 0 && !opt_disable_networking && !opt_bootstrap)
@@ -3015,6 +3005,21 @@ int main(int argc, char **argv)
   }
 #endif
 
+#ifdef	__WIN__
+/* Before performing any socket operation (like retrieving hostname */
+/* in init_common_variables we have to call WSAStartup              */
+  if (!opt_disable_networking)
+  {
+    WSADATA WsaData;
+    if (SOCKET_ERROR == WSAStartup (0x0101, &WsaData))
+    {
+      /* errors are not read yet, so we use test here */
+      my_message(ER_WSAS_FAILED, "WSAStartup Failed", MYF(0));
+      unireg_abort(1);
+    }
+  }
+#endif /* __WIN__ */
+
   if (init_common_variables(MYSQL_CONFIG_NAME,
 			    argc, argv, load_default_groups))
     unireg_abort(1);				// Will do exit
@@ -4331,7 +4336,8 @@ enum options_mysqld
   OPT_AUTO_INCREMENT, OPT_AUTO_INCREMENT_OFFSET,
   OPT_ENABLE_LARGE_PAGES,
   OPT_TIMED_MUTEXES,
-  OPT_OLD_STYLE_USER_LIMITS
+  OPT_OLD_STYLE_USER_LIMITS,
+  OPT_LOG_SLOW_ADMIN_STATEMENTS
 };
 
 
@@ -4453,7 +4459,7 @@ Disable with --skip-bdb (will save memory).",
    (gptr*) &default_collation_name, (gptr*) &default_collation_name,
    0, GET_STR, REQUIRED_ARG, 0, 0, 0, 0, 0, 0 },
   {"default-storage-engine", OPT_STORAGE_ENGINE,
-   "Set the default storage engine (table tyoe) for tables.", 0, 0,
+   "Set the default storage engine (table type) for tables.", 0, 0,
    0, GET_STR, REQUIRED_ARG, 0, 0, 0, 0, 0, 0},
   {"default-table-type", OPT_STORAGE_ENGINE,
    "(deprecated) Use --default-storage-engine.", 0, 0,
@@ -4656,7 +4662,7 @@ Disable with --skip-innodb-doublewrite.", (gptr*) &innobase_use_doublewrite,
    "Log some extra information to update log. Please note that this option is deprecated; see --log-short-format option.", 
    0, 0, 0, GET_NO_ARG, NO_ARG, 0, 0, 0, 0, 0, 0},
   {"log-queries-not-using-indexes", OPT_LOG_QUERIES_NOT_USING_INDEXES,
-   "Log queries that are executed without benefit of any index.",
+   "Log queries that are executed without benefit of any index to the slow log if it is open.",
    (gptr*) &opt_log_queries_not_using_indexes, (gptr*) &opt_log_queries_not_using_indexes,
    0, GET_BOOL, NO_ARG, 0, 0, 0, 0, 0, 0},
   {"log-short-format", OPT_SHORT_LOG_FORMAT,
@@ -4667,8 +4673,13 @@ Disable with --skip-innodb-doublewrite.", (gptr*) &innobase_use_doublewrite,
    "Tells the slave to log the updates from the slave thread to the binary log. You will need to turn it on if you plan to daisy-chain the slaves.",
    (gptr*) &opt_log_slave_updates, (gptr*) &opt_log_slave_updates, 0, GET_BOOL,
    NO_ARG, 0, 0, 0, 0, 0, 0},
+  {"log-slow-admin-statements", OPT_LOG_SLOW_ADMIN_STATEMENTS,
+   "Log slow OPTIMIZE, ANALYZE, ALTER and other administrative statements to the slow log if it is open.",
+   (gptr*) &opt_log_slow_admin_statements,
+   (gptr*) &opt_log_slow_admin_statements,
+   0, GET_BOOL, NO_ARG, 0, 0, 0, 0, 0, 0},
   {"log-slow-queries", OPT_SLOW_QUERY_LOG,
-   "Log slow queries to this log file. Defaults logging to hostname-slow.log file.",
+    "Log slow queries to this log file. Defaults logging to hostname-slow.log file. Must be enabled to activate other slow log options.",
    (gptr*) &opt_slow_logname, (gptr*) &opt_slow_logname, 0, GET_STR, OPT_ARG,
    0, 0, 0, 0, 0, 0},
   {"log-tc", OPT_LOG_TC,
@@ -5228,7 +5239,7 @@ log and this option does nothing anymore.",
   {"innodb_thread_concurrency", OPT_INNODB_THREAD_CONCURRENCY,
    "Helps in performance tuning in heavily concurrent environments.",
    (gptr*) &srv_thread_concurrency, (gptr*) &srv_thread_concurrency,
-   0, GET_LONG, REQUIRED_ARG, 8, 1, 1000, 0, 1, 0},
+   0, GET_LONG, REQUIRED_ARG, 20, 1, 1000, 0, 1, 0},
   {"innodb_thread_sleep_delay", OPT_INNODB_THREAD_SLEEP_DELAY,
    "Time of innodb thread sleeping before joining InnoDB queue (usec). Value 0"
     " disable a sleep",
@@ -5716,6 +5727,12 @@ struct show_var_st status_vars[]= {
   {"Com_show_warnings",        (char*) offsetof(STATUS_VAR, com_stat[(uint) SQLCOM_SHOW_WARNS]), SHOW_LONG_STATUS},
   {"Com_slave_start",	       (char*) offsetof(STATUS_VAR, com_stat[(uint) SQLCOM_SLAVE_START]), SHOW_LONG_STATUS},
   {"Com_slave_stop",	       (char*) offsetof(STATUS_VAR, com_stat[(uint) SQLCOM_SLAVE_STOP]), SHOW_LONG_STATUS},
+  {"Com_stmt_prepare",         (char*) offsetof(STATUS_VAR, com_stmt_prepare), SHOW_LONG_STATUS},
+  {"Com_stmt_execute",         (char*) offsetof(STATUS_VAR, com_stmt_execute), SHOW_LONG_STATUS},
+  {"Com_stmt_fetch",           (char*) offsetof(STATUS_VAR, com_stmt_fetch), SHOW_LONG_STATUS},
+  {"Com_stmt_send_long_data",  (char*) offsetof(STATUS_VAR, com_stmt_send_long_data), SHOW_LONG_STATUS},
+  {"Com_stmt_reset",           (char*) offsetof(STATUS_VAR, com_stmt_reset), SHOW_LONG_STATUS},
+  {"Com_stmt_close",           (char*) offsetof(STATUS_VAR, com_stmt_close), SHOW_LONG_STATUS},
   {"Com_truncate",	       (char*) offsetof(STATUS_VAR, com_stat[(uint) SQLCOM_TRUNCATE]), SHOW_LONG_STATUS},
   {"Com_unlock_tables",	       (char*) offsetof(STATUS_VAR, com_stat[(uint) SQLCOM_UNLOCK_TABLES]), SHOW_LONG_STATUS},
   {"Com_update",	       (char*) offsetof(STATUS_VAR, com_stat[(uint) SQLCOM_UPDATE]), SHOW_LONG_STATUS},
@@ -6343,6 +6360,9 @@ get_one_option(int optid, const struct my_option *opt __attribute__((unused)),
   case (int) OPT_SLOW_QUERY_LOG:
     opt_slow_log=1;
     break;
+  case (int) OPT_LOG_SLOW_ADMIN_STATEMENTS:
+    opt_log_slow_admin_statements= 1;
+    break;
   case (int) OPT_SKIP_NEW:
     opt_specialflag|= SPECIAL_NO_NEW_FUNC;
     delay_key_write_options= (uint) DELAY_KEY_WRITE_NONE;
@@ -6731,6 +6751,9 @@ static void get_options(int argc,char **argv)
   if (opt_bdb)
     sql_print_warning("this binary does not contain BDB storage engine");
 #endif
+  if ((opt_log_slow_admin_statements || opt_log_queries_not_using_indexes) &&
+      !opt_slow_log)
+    sql_print_warning("options --log-slow-admin-statements and --log-queries-not-using-indexes have no effect if --log-slow-queries is not set");
 
   /*
     Check that the default storage engine is actually available.
@@ -7051,7 +7074,7 @@ static void create_pid_file()
   Instantiate templates
 *****************************************************************************/
 
-#ifdef __GNUC__
+#ifdef HAVE_EXPLICIT_TEMPLATE_INSTANTIATION
 /* Used templates */
 template class I_List<THD>;
 template class I_List_iterator<THD>;
diff --git a/sql/opt_range.cc b/sql/opt_range.cc
index b3301d17655..3226dc9607b 100644
--- a/sql/opt_range.cc
+++ b/sql/opt_range.cc
@@ -325,7 +325,7 @@ typedef struct st_qsel_param {
   TABLE *table;
   KEY_PART *key_parts,*key_parts_end;
   KEY_PART *key[MAX_KEY]; /* First key parts of keys used in the query */
-  MEM_ROOT *mem_root;
+  MEM_ROOT *mem_root, *old_root;
   table_map prev_tables,read_tables,current_table;
   uint baseflag, max_key_part, range_count;
 
@@ -1665,7 +1665,7 @@ int SQL_SELECT::test_quick_select(THD *thd, key_map keys_to_use,
   keys_to_use.intersect(head->keys_in_use_for_query);
   if (!keys_to_use.is_clear_all())
   {
-    MEM_ROOT *old_root,alloc;
+    MEM_ROOT alloc;
     SEL_TREE *tree= NULL;
     KEY_PART *key_parts;
     KEY *key_info;
@@ -1680,6 +1680,7 @@ int SQL_SELECT::test_quick_select(THD *thd, key_map keys_to_use,
     param.table=head;
     param.keys=0;
     param.mem_root= &alloc;
+    param.old_root= thd->mem_root;
     param.needed_reg= &needed_reg;
     param.imerge_cost_buff_size= 0;
 
@@ -1695,7 +1696,6 @@ int SQL_SELECT::test_quick_select(THD *thd, key_map keys_to_use,
       DBUG_RETURN(0);				// Can't use range
     }
     key_parts= param.key_parts;
-    old_root= thd->mem_root;
     thd->mem_root= &alloc;
 
     /*
@@ -1845,7 +1845,7 @@ int SQL_SELECT::test_quick_select(THD *thd, key_map keys_to_use,
       }
     }
 
-    thd->mem_root= old_root;
+    thd->mem_root= param.old_root;
 
     /* If we got a read plan, create a quick select from it. */
     if (best_trp)
@@ -1860,7 +1860,7 @@ int SQL_SELECT::test_quick_select(THD *thd, key_map keys_to_use,
 
   free_mem:
     free_root(&alloc,MYF(0));			// Return memory & allocator
-    thd->mem_root= old_root;
+    thd->mem_root= param.old_root;
     thd->no_errors=0;
   }
 
@@ -2593,12 +2593,12 @@ static double ror_scan_selectivity(const ROR_INTERSECT_INFO *info,
       {
         tuple_arg= scan->sel_arg;
         /* Here we use the length of the first key part */
-        tuple_arg->store_min(key_part->length, &key_ptr, 0);
+        tuple_arg->store_min(key_part->store_length, &key_ptr, 0);
       }
       while (tuple_arg->next_key_part != sel_arg)
       {
         tuple_arg= tuple_arg->next_key_part;
-        tuple_arg->store_min(key_part[tuple_arg->part].length, &key_ptr, 0);
+        tuple_arg->store_min(key_part[tuple_arg->part].store_length, &key_ptr, 0);
       }
       min_range.length= max_range.length= ((char*) key_ptr - (char*) key_val);
       records= (info->param->table->file->
@@ -3581,15 +3581,16 @@ static SEL_TREE *get_mm_tree(PARAM *param,COND *cond)
     DBUG_RETURN(ftree);
   }
   default:
-    if (cond_func->arguments()[0]->type() == Item::FIELD_ITEM)
+    if (cond_func->arguments()[0]->real_item()->type() == Item::FIELD_ITEM)
     {
-      field_item= (Item_field*) (cond_func->arguments()[0]);
+      field_item= (Item_field*) (cond_func->arguments()[0]->real_item());
       value= cond_func->arg_count > 1 ? cond_func->arguments()[1] : 0;
     }
     else if (cond_func->have_rev_func() &&
-             cond_func->arguments()[1]->type() == Item::FIELD_ITEM)
+             cond_func->arguments()[1]->real_item()->type() ==
+                                                            Item::FIELD_ITEM)
     {
-      field_item= (Item_field*) (cond_func->arguments()[1]);
+      field_item= (Item_field*) (cond_func->arguments()[1]->real_item());
       value= cond_func->arguments()[0];
     }
     else
@@ -3610,7 +3611,7 @@ static SEL_TREE *get_mm_tree(PARAM *param,COND *cond)
      
   for (uint i= 0; i < cond_func->arg_count; i++)
   {
-    Item *arg= cond_func->arguments()[i];
+    Item *arg= cond_func->arguments()[i]->real_item();
     if (arg != field_item)
       ref_tables|= arg->used_tables();
   }
@@ -3695,24 +3696,38 @@ get_mm_leaf(PARAM *param, COND *conf_func, Field *field, KEY_PART *key_part,
 {
   uint maybe_null=(uint) field->real_maybe_null();
   bool optimize_range;
-  SEL_ARG *tree;
+  SEL_ARG *tree= 0;
+  MEM_ROOT *alloc= param->mem_root;
   char *str;
   DBUG_ENTER("get_mm_leaf");
 
+  /*
+    We need to restore the runtime mem_root of the thread in this
+    function because it evaluates the value of its argument, while
+    the argument can be any, e.g. a subselect. The subselect
+    items, in turn, assume that all the memory allocated during
+    the evaluation has the same life span as the item itself.
+    TODO: opt_range.cc should not reset thd->mem_root at all.
+  */
+  param->thd->mem_root= param->old_root;
   if (!value)					// IS NULL or IS NOT NULL
   {
     if (field->table->maybe_null)		// Can't use a key on this
-      DBUG_RETURN(0);
+      goto end;
     if (!maybe_null)				// Not null field
-      DBUG_RETURN(type == Item_func::ISNULL_FUNC ? &null_element : 0);
-    if (!(tree=new SEL_ARG(field,is_null_string,is_null_string)))
-      DBUG_RETURN(0);		// out of memory
+    {
+      if (type == Item_func::ISNULL_FUNC)
+        tree= &null_element;
+      goto end;
+    }
+    if (!(tree= new (alloc) SEL_ARG(field,is_null_string,is_null_string)))
+      goto end;                                 // out of memory
     if (type == Item_func::ISNOTNULL_FUNC)
     {
       tree->min_flag=NEAR_MIN;		    /* IS NOT NULL ->  X > NULL */
       tree->max_flag=NO_MAX_RANGE;
     }
-    DBUG_RETURN(tree);
+    goto end;
   }
 
   /*
@@ -3732,7 +3747,7 @@ get_mm_leaf(PARAM *param, COND *conf_func, Field *field, KEY_PART *key_part,
       key_part->image_type == Field::itRAW &&
       ((Field_str*)field)->charset() != conf_func->compare_collation() &&
       !(conf_func->compare_collation()->state & MY_CS_BINSORT))
-    DBUG_RETURN(0);
+    goto end;
 
   optimize_range= field->optimize_range(param->real_keynr[key_part->key],
                                         key_part->part);
@@ -3746,9 +3761,12 @@ get_mm_leaf(PARAM *param, COND *conf_func, Field *field, KEY_PART *key_part,
     uint field_length= field->pack_length()+maybe_null;
 
     if (!optimize_range)
-      DBUG_RETURN(0);				// Can't optimize this
+      goto end;
     if (!(res= value->val_str(&tmp)))
-      DBUG_RETURN(&null_element);
+    {
+      tree= &null_element;
+      goto end;
+    }
 
     /*
       TODO:
@@ -3761,7 +3779,7 @@ get_mm_leaf(PARAM *param, COND *conf_func, Field *field, KEY_PART *key_part,
       res= &tmp;
     }
     if (field->cmp_type() != STRING_RESULT)
-      DBUG_RETURN(0);				// Can only optimize strings
+      goto end;                                 // Can only optimize strings
 
     offset=maybe_null;
     length=key_part->store_length;
@@ -3786,8 +3804,8 @@ get_mm_leaf(PARAM *param, COND *conf_func, Field *field, KEY_PART *key_part,
 	field_length= length;
     }
     length+=offset;
-    if (!(min_str= (char*) alloc_root(param->mem_root, length*2)))
-      DBUG_RETURN(0);
+    if (!(min_str= (char*) alloc_root(alloc, length*2)))
+      goto end;
 
     max_str=min_str+length;
     if (maybe_null)
@@ -3802,20 +3820,21 @@ get_mm_leaf(PARAM *param, COND *conf_func, Field *field, KEY_PART *key_part,
 			      min_str+offset, max_str+offset,
 			      &min_length, &max_length);
     if (like_error)				// Can't optimize with LIKE
-      DBUG_RETURN(0);
+      goto end;
 
     if (offset != maybe_null)			// BLOB or VARCHAR
     {
       int2store(min_str+maybe_null,min_length);
       int2store(max_str+maybe_null,max_length);
     }
-    DBUG_RETURN(new SEL_ARG(field,min_str,max_str));
+    tree= new (alloc) SEL_ARG(field, min_str, max_str);
+    goto end;
   }
 
   if (!optimize_range &&
       type != Item_func::EQ_FUNC &&
       type != Item_func::EQUAL_FUNC)
-    DBUG_RETURN(0);				// Can't optimize this
+    goto end;                                   // Can't optimize this
 
   /*
     We can't always use indexes when comparing a string index to a number
@@ -3824,21 +3843,53 @@ get_mm_leaf(PARAM *param, COND *conf_func, Field *field, KEY_PART *key_part,
   if (field->result_type() == STRING_RESULT &&
       value->result_type() != STRING_RESULT &&
       field->cmp_type() != value->result_type())
-    DBUG_RETURN(0);
+    goto end;
 
   if (value->save_in_field_no_warnings(field, 1) < 0)
   {
     /* This happens when we try to insert a NULL field in a not null column */
-    DBUG_RETURN(&null_element);			// cmp with NULL is never TRUE
+    tree= &null_element;                        // cmp with NULL is never TRUE
+    goto end;
   }
-  str= (char*) alloc_root(param->mem_root, key_part->store_length+1);
+  str= (char*) alloc_root(alloc, key_part->store_length+1);
   if (!str)
-    DBUG_RETURN(0);
+    goto end;
   if (maybe_null)
     *str= (char) field->is_real_null();		// Set to 1 if null
   field->get_key_image(str+maybe_null, key_part->length, key_part->image_type);
-  if (!(tree=new SEL_ARG(field,str,str)))
-    DBUG_RETURN(0);		// out of memory
+  if (!(tree= new (alloc) SEL_ARG(field, str, str)))
+    goto end;                                   // out of memory
+
+  /*
+    Check if we are comparing an UNSIGNED integer with a negative constant.
+    In this case we know that:
+    (a) (unsigned_int [< | <=] negative_constant) == FALSE
+    (b) (unsigned_int [> | >=] negative_constant) == TRUE
+    In case (a) the condition is false for all values, and in case (b) it
+    is true for all values, so we can avoid unnecessary retrieval and condition
+    testing, and we also get correct comparison of unsinged integers with
+    negative integers (which otherwise fails because at query execution time
+    negative integers are cast to unsigned if compared with unsigned).
+   */
+  if (field->result_type() == INT_RESULT &&
+      value->result_type() == INT_RESULT &&
+      ((Field_num*)field)->unsigned_flag && !((Item_int*)value)->unsigned_flag)
+  {
+    longlong item_val= value->val_int();
+    if (item_val < 0)
+    {
+      if (type == Item_func::LT_FUNC || type == Item_func::LE_FUNC)
+      {
+        tree->type= SEL_ARG::IMPOSSIBLE;
+        goto end;
+      }
+      if (type == Item_func::GT_FUNC || type == Item_func::GE_FUNC)
+      {
+        tree= 0;
+        goto end;
+      }
+    }
+  }
 
   switch (type) {
   case Item_func::LT_FUNC:
@@ -3899,6 +3950,9 @@ get_mm_leaf(PARAM *param, COND *conf_func, Field *field, KEY_PART *key_part,
   default:
     break;
   }
+
+end:
+  param->thd->mem_root= alloc;
   DBUG_RETURN(tree);
 }
 
@@ -5992,7 +6046,10 @@ int QUICK_RANGE_SELECT::reset()
   next=0;
   range= NULL;
   cur_range= (QUICK_RANGE**) ranges.buffer;
-  
+
+  if (file->inited == handler::NONE && (error= file->ha_index_init(index)))
+    DBUG_RETURN(error);
+ 
   /* Do not allocate the buffers twice. */
   if (multi_range_length)
   {
@@ -7612,8 +7669,8 @@ void cost_group_min_max(TABLE* table, KEY *index_info, uint used_key_parts,
   *records= num_groups;
 
   DBUG_PRINT("info",
-             ("records=%u, keys/block=%u, keys/group=%u, records=%u, blocks=%u",
-              table_records, keys_per_block, keys_per_group, records,
+             ("table rows=%u, keys/block=%u, keys/group=%u, result rows=%u, blocks=%u",
+              table_records, keys_per_block, keys_per_group, *records,
               num_blocks));
   DBUG_VOID_RETURN;
 }
@@ -8120,6 +8177,15 @@ int QUICK_GROUP_MIN_MAX_SELECT::get_next()
       DBUG_ASSERT((have_max && !have_min) ||
                   (have_max && have_min && (max_res == 0)));
     }
+    /*
+      If this is a just a GROUP BY or DISTINCT without MIN or MAX and there
+      are equality predicates for the key parts after the group, find the
+      first sub-group with the extended prefix.
+    */
+    if (!have_min && !have_max && key_infix_len > 0)
+      result= file->index_read(record, group_prefix, real_prefix_len,
+                               HA_READ_KEY_EXACT);
+
     result= have_min ? min_res : have_max ? max_res : result;
   }
   while (result == HA_ERR_KEY_NOT_FOUND && is_last_prefix != 0);
@@ -8146,9 +8212,8 @@ int QUICK_GROUP_MIN_MAX_SELECT::get_next()
     QUICK_GROUP_MIN_MAX_SELECT::next_min()
 
   DESCRIPTION
-    Load the prefix of the next group into group_prefix and find the minimal
-    key within this group such that the key satisfies the query conditions and
-    NULL semantics. The found key is loaded into this->record.
+    Find the minimal key within this group such that the key satisfies the query
+    conditions and NULL semantics. The found key is loaded into this->record.
 
   IMPLEMENTATION
     Depending on the values of min_max_ranges.elements, key_infix_len, and
@@ -8232,9 +8297,7 @@ int QUICK_GROUP_MIN_MAX_SELECT::next_min()
     QUICK_GROUP_MIN_MAX_SELECT::next_max()
 
   DESCRIPTION
-    If there was no previous next_min call to determine the next group prefix,
-    then load the next prefix into group_prefix, then lookup the maximal key of
-    the group, and store it into this->record.
+    Lookup the maximal key of the group, and store it into this->record.
 
   RETURN
     0                    on success
@@ -8912,7 +8975,7 @@ void QUICK_GROUP_MIN_MAX_SELECT::dbug_dump(int indent, bool verbose)
 ** Instantiate templates
 *****************************************************************************/
 
-#ifdef __GNUC__
+#ifdef HAVE_EXPLICIT_TEMPLATE_INSTANTIATION
 template class List<QUICK_RANGE>;
 template class List_iterator<QUICK_RANGE>;
 #endif
diff --git a/sql/password.c b/sql/password.c
index 79675ade30b..60cc0ac0c97 100644
--- a/sql/password.c
+++ b/sql/password.c
@@ -146,7 +146,7 @@ void hash_password(ulong *result, const char *password, uint password_len)
 void make_scrambled_password_323(char *to, const char *password)
 {
   ulong hash_res[2];
-  hash_password(hash_res, password, strlen(password));
+  hash_password(hash_res, password, (uint) strlen(password));
   sprintf(to, "%08lx%08lx", hash_res[0], hash_res[1]);
 }
 
@@ -172,7 +172,7 @@ void scramble_323(char *to, const char *message, const char *password)
   {
     char extra, *to_start=to;
     const char *message_end= message + SCRAMBLE_LENGTH_323;
-    hash_password(hash_pass,password, strlen(password));
+    hash_password(hash_pass,password, (uint) strlen(password));
     hash_password(hash_message, message, SCRAMBLE_LENGTH_323);
     randominit(&rand_st,hash_pass[0] ^ hash_message[0],
                hash_pass[1] ^ hash_message[1]);
@@ -394,7 +394,7 @@ make_scrambled_password(char *to, const char *password)
 
   sha1_reset(&sha1_context);
   /* stage 1: hash password */
-  sha1_input(&sha1_context, (uint8 *) password, strlen(password));
+  sha1_input(&sha1_context, (uint8 *) password, (uint) strlen(password));
   sha1_result(&sha1_context, (uint8 *) to);
   /* stage 2: hash stage1 output */
   sha1_reset(&sha1_context);
@@ -433,7 +433,7 @@ scramble(char *to, const char *message, const char *password)
 
   sha1_reset(&sha1_context);
   /* stage 1: hash password */
-  sha1_input(&sha1_context, (uint8 *) password, strlen(password));
+  sha1_input(&sha1_context, (uint8 *) password, (uint) strlen(password));
   sha1_result(&sha1_context, hash_stage1);
   /* stage 2: hash stage 1; note that hash_stage2 is stored in the database */
   sha1_reset(&sha1_context);
diff --git a/sql/protocol.cc b/sql/protocol.cc
index 57922cdc677..1c399a89a99 100644
--- a/sql/protocol.cc
+++ b/sql/protocol.cc
@@ -294,7 +294,12 @@ send_ok(THD *thd, ha_rows affected_rows, ulonglong id, const char *message)
   DBUG_ENTER("send_ok");
 
   if (net->no_send_ok || !net->vio)	// hack for re-parsing queries
+  {
+    DBUG_PRINT("info", ("no send ok: %s, vio present: %s",
+                        (net->no_send_ok ? "YES" : "NO"),
+                        (net->vio ? "YES" : "NO")));
     DBUG_VOID_RETURN;
+  }
 
   buff[0]=0;					// No fields
   pos=net_store_length(buff+1,(ulonglong) affected_rows);
diff --git a/sql/set_var.cc b/sql/set_var.cc
index b22c0924de1..1c0de702e4e 100644
--- a/sql/set_var.cc
+++ b/sql/set_var.cc
@@ -3131,7 +3131,7 @@ bool sys_var_thd_storage_engine::check(THD *thd, set_var *var)
     if (!(res=var->value->val_str(&str)) ||
 	!(var->save_result.ulong_value=
           (ulong) (db_type= ha_resolve_by_name(res->ptr(), res->length()))) ||
-        ha_checktype(db_type) != db_type)
+        ha_checktype(thd, db_type, 1, 0) != db_type)
     {
       value= res ? res->c_ptr() : "NULL";
       goto err;
@@ -3406,7 +3406,7 @@ bool process_key_caches(int (* func) (const char *name, KEY_CACHE *))
   Used templates
 ****************************************************************************/
 
-#ifdef __GNUC__
+#ifdef HAVE_EXPLICIT_TEMPLATE_INSTANTIATION
 template class List<set_var_base>;
 template class List_iterator_fast<set_var_base>;
 template class I_List_iterator<NAMED_LIST>;
diff --git a/sql/set_var.h b/sql/set_var.h
index 56690c46131..a6532323b34 100644
--- a/sql/set_var.h
+++ b/sql/set_var.h
@@ -786,7 +786,8 @@ public:
     if (value_arg && value_arg->type() == Item::FIELD_ITEM)
     {
       Item_field *item= (Item_field*) value_arg;
-      if (!(value=new Item_string(item->field_name, strlen(item->field_name),
+      if (!(value=new Item_string(item->field_name, 
+                  (uint) strlen(item->field_name),
 				  item->collation.collation)))
 	value=value_arg;			/* Give error message later */
     }
diff --git a/sql/share/errmsg.txt b/sql/share/errmsg.txt
index a020cadc084..7ae5130764f 100644
--- a/sql/share/errmsg.txt
+++ b/sql/share/errmsg.txt
@@ -5021,7 +5021,7 @@ ER_NON_UPDATABLE_TABLE
 	por "A tabela destino %-.100s do %s não é atualizável"
 	rus "ôÁÂÌÉÃÁ %-.100s × %s ÎÅ ÍÏÖÅÔ ÉÚÍÅÎÑÔÓÑ"
 	spa "La tabla destino %-.100s del %s no es actualizable"
-	swe "Tabel %-.100s använd med '%s' är inte uppdateringsbar"
+	swe "Tabell %-.100s använd med '%s' är inte uppdateringsbar"
 	ukr "ôÁÂÌÉÃÑ %-.100s Õ %s ÎÅ ÍÏÖÅ ÏÎÏ×ÌÀ×ÁÔÉÓØ"
 ER_FEATURE_DISABLED  
 	eng "The '%s' feature is disabled; you need MySQL built with '%s' to have it working"
@@ -5347,14 +5347,14 @@ ER_SP_NO_RETSET_IN_FUNC 0A000
 ER_CANT_CREATE_GEOMETRY_OBJECT 22003 
 	eng "Cannot get geometry object from data you send to the GEOMETRY field"
 ER_FAILED_ROUTINE_BREAK_BINLOG
-	eng "A routine failed and is declared to modify data and binary logging is enabled; if non-transactional tables were updated, the binary log will miss their changes"
+	eng "A routine failed and has neither NO SQL nor READS SQL DATA in its declaration and binary logging is enabled; if non-transactional tables were updated, the binary log will miss their changes"
 ER_BINLOG_UNSAFE_ROUTINE
-	eng "This routine is declared to be non-deterministic and to modify data and binary logging is enabled (you *might* want to use the less safe log_bin_trust_routine_creators variable)"
+	eng "This routine has none of DETERMINISTIC, NO SQL, or READS SQL DATA in its declaration and binary logging is enabled (you *might* want to use the less safe log_bin_trust_routine_creators variable)"
 ER_BINLOG_CREATE_ROUTINE_NEED_SUPER
-	eng "You do not have SUPER privilege and binary logging is enabled (you *might* want to use the less safe log_bin_trust_routine_creators variable)"
+	eng "You do not have the SUPER privilege and binary logging is enabled (you *might* want to use the less safe log_bin_trust_routine_creators variable)"
 ER_EXEC_STMT_WITH_OPEN_CURSOR
 	eng "You can't execute a prepared statement which has an open cursor associated with it. Reset the statement to re-execute it."
 ER_STMT_HAS_NO_OPEN_CURSOR
-	eng "The statement (%d) has no open cursor."
+	eng "The statement (%lu) has no open cursor."
 ER_COMMIT_NOT_ALLOWED_IN_SF_OR_TRG
         eng "Explicit or implicit commit is not allowed in stored function or trigger."
diff --git a/sql/slave.cc b/sql/slave.cc
index 60812469671..c6c0de7160b 100644
--- a/sql/slave.cc
+++ b/sql/slave.cc
@@ -5002,7 +5002,7 @@ end:
 }
 
 
-#ifdef __GNUC__
+#ifdef HAVE_EXPLICIT_TEMPLATE_INSTANTIATION
 template class I_List_iterator<i_string>;
 template class I_List_iterator<i_string_pair>;
 #endif
diff --git a/sql/sp_head.cc b/sql/sp_head.cc
index 29898437cfb..e4dc64c993d 100644
--- a/sql/sp_head.cc
+++ b/sql/sp_head.cc
@@ -310,16 +310,16 @@ sp_head::operator delete(void *ptr, size_t size)
 
 
 sp_head::sp_head()
-  :Item_arena((bool)FALSE), m_returns_cs(NULL), m_has_return(FALSE),
+  :Query_arena(&main_mem_root, INITIALIZED_FOR_SP),
+   m_returns_cs(NULL), m_has_return(FALSE),
    m_simple_case(FALSE), m_multi_results(FALSE), m_in_handler(FALSE)
 {
   extern byte *
     sp_table_key(const byte *ptr, uint *plen, my_bool first);
-  extern byte 
+  extern byte
     *sp_lex_sp_key(const byte *ptr, uint *plen, my_bool first);
   DBUG_ENTER("sp_head::sp_head");
 
-  state= INITIALIZED_FOR_SP;
   m_backpatch.empty();
   m_lex.empty();
   hash_init(&m_sptabs, system_charset_info, 0, 0, 0, sp_table_key, 0, 0);
@@ -509,7 +509,7 @@ sp_head::destroy()
     delete i;
   delete_dynamic(&m_instr);
   m_pcont->destroy();
-  free_items(free_list);
+  free_items();
 
   /*
     If we have non-empty LEX stack then we just came out of parser with
@@ -574,7 +574,7 @@ sp_head::execute(THD *thd)
   sp_rcontext *ctx;
   int ret= 0;
   uint ip= 0;
-  Item_arena *old_arena;
+  Query_arena *old_arena;
   query_id_t old_query_id;
   TABLE *old_derived_tables;
   LEX *old_lex;
@@ -596,7 +596,6 @@ sp_head::execute(THD *thd)
     ctx->clear_handler();
   thd->query_error= 0;
   old_arena= thd->current_arena;
-  thd->current_arena= this;
 
   /*
     We have to save/restore this info when we are changing call level to
@@ -636,9 +635,18 @@ sp_head::execute(THD *thd)
       break;
     DBUG_PRINT("execute", ("Instruction %u", ip));
     thd->set_time();		// Make current_time() et al work
+    /*
+      We have to set thd->current_arena before executing the instruction
+      to store in the instruction free_list all new items, created
+      during the first execution (for example expanding of '*' or the
+      items made during other permanent subquery transformations).
+    */
+    thd->current_arena= i;
     ret= i->execute(thd, &ip);
     if (i->free_list)
       cleanup_items(i->free_list);
+    i->state= Query_arena::EXECUTED;
+
     // Check if an exception has occurred and a handler has been found
     // Note: We havo to check even if ret==0, since warnings (and some
     //       errors don't return a non-zero value.
@@ -680,8 +688,8 @@ sp_head::execute(THD *thd)
   DBUG_ASSERT(!thd->derived_tables);
   thd->derived_tables= old_derived_tables;
 
-  cleanup_items(thd->current_arena->free_list);
   thd->current_arena= old_arena;
+  state= EXECUTED;
 
  done:
   DBUG_PRINT("info", ("ret=%d killed=%d query_error=%d",
@@ -713,8 +721,8 @@ sp_head::execute_function(THD *thd, Item **argp, uint argcount, Item **resp)
   sp_rcontext *nctx = NULL;
   uint i;
   int ret;
-  MEM_ROOT *old_mem_root, call_mem_root;
-  Item *old_free_list, *call_free_list;
+  MEM_ROOT call_mem_root;
+  Query_arena call_arena(&call_mem_root, INITIALIZED_FOR_SP), backup_arena;
 
   if (argcount != params)
   {
@@ -726,14 +734,11 @@ sp_head::execute_function(THD *thd, Item **argp, uint argcount, Item **resp)
   }
 
   init_alloc_root(&call_mem_root, MEM_ROOT_BLOCK_SIZE, 0);
-  old_mem_root= thd->mem_root;
-  thd->mem_root= &call_mem_root;
-  old_free_list= thd->free_list; // Keep the old list
-  thd->free_list= NULL;	// Start a new one
+
 
   // QQ Should have some error checking here? (types, etc...)
   nctx= new sp_rcontext(csize, hmax, cmax);
-  nctx->callers_mem_root= old_mem_root;
+  nctx->callers_mem_root= thd->mem_root;
   for (i= 0 ; i < argcount ; i++)
   {
     sp_pvar_t *pvar = m_pcont->find_pvar(i);
@@ -759,15 +764,16 @@ sp_head::execute_function(THD *thd, Item **argp, uint argcount, Item **resp)
     }
   }
   thd->spcont= nctx;
+  thd->set_n_backup_item_arena(&call_arena, &backup_arena);
+  /* mem_root was moved to backup_arena */
+  DBUG_ASSERT(nctx->callers_mem_root == backup_arena.mem_root);
 
   ret= execute(thd);
 
   // Partially restore context now.
   // We still need the call mem root and free list for processing
   // of the result.
-  call_free_list= thd->free_list;
-  thd->free_list= old_free_list;
-  thd->mem_root= old_mem_root;
+  thd->restore_backup_item_arena(&call_arena, &backup_arena);
 
   if (m_type == TYPE_ENUM_FUNCTION && ret == 0)
   {
@@ -787,8 +793,7 @@ sp_head::execute_function(THD *thd, Item **argp, uint argcount, Item **resp)
   thd->spcont= octx;
 
   // Now get rid of the rest of the callee context
-  cleanup_items(call_free_list);
-  free_items(call_free_list);
+  call_arena.free_items();
   free_root(&call_mem_root, MYF(0));
 
   DBUG_RETURN(ret);
@@ -820,8 +825,8 @@ sp_head::execute_procedure(THD *thd, List<Item> *args)
   sp_rcontext *octx = thd->spcont;
   sp_rcontext *nctx = NULL;
   my_bool tmp_octx = FALSE;	// True if we have allocated a temporary octx
-  MEM_ROOT *old_mem_root, call_mem_root;
-  Item *old_free_list, *call_free_list;
+  MEM_ROOT call_mem_root;
+  Query_arena call_arena(&call_mem_root, INITIALIZED_FOR_SP), backup_arena;
 
   if (args->elements != params)
   {
@@ -831,10 +836,6 @@ sp_head::execute_procedure(THD *thd, List<Item> *args)
   }
 
   init_alloc_root(&call_mem_root, MEM_ROOT_BLOCK_SIZE, 0);
-  old_mem_root= thd->mem_root;
-  thd->mem_root= &call_mem_root;
-  old_free_list= thd->free_list; // Keep the old list
-  thd->free_list= NULL;	// Start a new one
 
   if (csize > 0 || hmax > 0 || cmax > 0)
   {
@@ -899,14 +900,11 @@ sp_head::execute_procedure(THD *thd, List<Item> *args)
   }
 
   if (! ret)
+  {
+    thd->set_n_backup_item_arena(&call_arena, &backup_arena);
     ret= execute(thd);
-
-  // Partially restore context now.
-  // We still need the call mem root and free list for processing
-  // of out parameters.
-  call_free_list= thd->free_list;
-  thd->free_list= old_free_list;
-  thd->mem_root= old_mem_root;
+    thd->restore_backup_item_arena(&call_arena, &backup_arena);
+  }
 
   if (!ret && csize > 0)
   {
@@ -981,8 +979,7 @@ sp_head::execute_procedure(THD *thd, List<Item> *args)
   thd->spcont= octx;
 
   // Now get rid of the rest of the callee context
-  cleanup_items(call_free_list);
-  free_items(call_free_list);
+  call_arena.free_items();
   thd->lex->unit.cleanup();
   free_root(&call_mem_root, MYF(0));
 
@@ -1276,6 +1273,13 @@ void sp_head::add_instr(sp_instr *instr)
 {
   instr->free_list= m_thd->free_list;
   m_thd->free_list= 0;
+  /*
+    Memory root of every instruction is designated for permanent
+    transformations (optimizations) made on the parsed tree during
+    the first execution. It points to the memory root of the
+    entire stored procedure, as their life span is equal.
+  */
+  instr->mem_root= &main_mem_root;
   insert_dynamic(&m_instr, (gptr)&instr);
 }
 
@@ -1470,13 +1474,6 @@ sp_lex_keeper::reset_lex_and_exec_core(THD *thd, uint *nextp,
     they want to store some value in local variable, pass return value and
     etc... So their life time should be longer than one instruction.
 
-    Probably we can call destructors for most of them then we are leaving
-    routine. But this won't help much as they are allocated in main query
-    MEM_ROOT anyway. So they all go to global thd->free_list.
-
-    May be we can use some other MEM_ROOT for this purprose ???
-
-    What else should we do for cleanup ?
     cleanup_items() is called in sp_head::execute()
   */
   return res;
@@ -2312,7 +2309,7 @@ sp_head::add_used_tables_to_table_list(THD *thd,
                                        TABLE_LIST ***query_tables_last_ptr)
 {
   uint i;
-  Item_arena *arena, backup;
+  Query_arena *arena, backup;
   bool result= FALSE;
   DBUG_ENTER("sp_head::add_used_tables_to_table_list");
 
diff --git a/sql/sp_head.h b/sql/sp_head.h
index 49eabce246b..aaef5a3d50e 100644
--- a/sql/sp_head.h
+++ b/sql/sp_head.h
@@ -74,11 +74,12 @@ sp_name *
 sp_name_current_db_new(THD *thd, LEX_STRING name);
 
 
-class sp_head :private Item_arena
+class sp_head :private Query_arena
 {
   sp_head(const sp_head &);	/* Prevent use of these */
   void operator=(sp_head &);
 
+  MEM_ROOT main_mem_root;
 public:
 
   int m_type;			// TYPE_ENUM_FUNCTION or TYPE_ENUM_PROCEDURE
@@ -273,7 +274,7 @@ private:
 // "Instructions"...
 //
 
-class sp_instr : public Sql_alloc
+class sp_instr :public Query_arena, public Sql_alloc
 {
   sp_instr(const sp_instr &);	/* Prevent use of these */
   void operator=(sp_instr &);
@@ -281,17 +282,16 @@ class sp_instr : public Sql_alloc
 public:
 
   uint marked;
-  Item *free_list;              // My Items
   uint m_ip;			// My index
   sp_pcontext *m_ctx;		// My parse context
 
   // Should give each a name or type code for debugging purposes?
   sp_instr(uint ip, sp_pcontext *ctx)
-    :Sql_alloc(), marked(0), free_list(0), m_ip(ip), m_ctx(ctx)
+    :Query_arena(0, INITIALIZED_FOR_SP), marked(0), m_ip(ip), m_ctx(ctx)
   {}
 
   virtual ~sp_instr()
-  { free_items(free_list); }
+  { free_items(); }
 
   // Execute this instrution. '*nextp' will be set to the index of the next
   // instruction to execute. (For most instruction this will be the
@@ -377,6 +377,10 @@ public:
     return (uint)m_lex->sql_command;
   }
 
+  void disable_query_cache()
+  {
+    m_lex->safe_to_cache_query= 0;
+  }
 private:
 
   LEX *m_lex;
diff --git a/sql/sp_rcontext.cc b/sql/sp_rcontext.cc
index 61cd7a9300f..aacb9254753 100644
--- a/sql/sp_rcontext.cc
+++ b/sql/sp_rcontext.cc
@@ -169,6 +169,17 @@ sp_rcontext::pop_cursors(uint count)
  *
  */
 
+sp_cursor::sp_cursor(sp_lex_keeper *lex_keeper)
+  :m_lex_keeper(lex_keeper), m_prot(NULL), m_isopen(0), m_current_row(NULL)
+{
+  /*
+    currsor can't be stored in QC, so we should prevent opening QC for
+    try to write results which are absent.
+  */
+  lex_keeper->disable_query_cache();
+}
+
+
 /*
   pre_open cursor
 
diff --git a/sql/sp_rcontext.h b/sql/sp_rcontext.h
index 864dc3df146..b188805435f 100644
--- a/sql/sp_rcontext.h
+++ b/sql/sp_rcontext.h
@@ -203,11 +203,7 @@ class sp_cursor : public Sql_alloc
 {
 public:
 
-  sp_cursor(sp_lex_keeper *lex_keeper)
-    : m_lex_keeper(lex_keeper), m_prot(NULL), m_isopen(0), m_current_row(NULL)
-  {
-    /* Empty */
-  }
+  sp_cursor(sp_lex_keeper *lex_keeper);
 
   virtual ~sp_cursor()
   {
diff --git a/sql/sql_acl.cc b/sql/sql_acl.cc
index 04da0dd5eb5..c1044d31c46 100644
--- a/sql/sql_acl.cc
+++ b/sql/sql_acl.cc
@@ -5325,7 +5325,7 @@ bool sp_grant_privileges(THD *thd, const char *sp_db, const char *sp_name,
   Instantiate used templates
 *****************************************************************************/
 
-#ifdef __GNUC__
+#ifdef HAVE_EXPLICIT_TEMPLATE_INSTANTIATION
 template class List_iterator<LEX_COLUMN>;
 template class List_iterator<LEX_USER>;
 template class List<LEX_COLUMN>;
diff --git a/sql/sql_base.cc b/sql/sql_base.cc
index a1887996d00..d8bbb686ba2 100644
--- a/sql/sql_base.cc
+++ b/sql/sql_base.cc
@@ -1611,8 +1611,18 @@ static int open_unireg_entry(THD *thd, TABLE *entry, const char *db,
       */
       if (discover_retry_count++ != 0)
         goto err;
-      if (ha_create_table_from_engine(thd, db, name, TRUE) != 0)
+      if (ha_create_table_from_engine(thd, db, name) > 0)
+      {
+        /* Give right error message */
+        thd->clear_error();
+        DBUG_PRINT("error", ("Dicovery of %s/%s failed", db, name));
+        my_printf_error(ER_UNKNOWN_ERROR,
+                        "Failed to open '%-.64s', error while "
+                        "unpacking from engine",
+                        MYF(0), name);
+
         goto err;
+      }
 
       mysql_reset_errors(thd, 1);    // Clear warnings
       thd->clear_error();            // Clear error message
@@ -2626,7 +2636,6 @@ find_field_in_tables(THD *thd, Item_ident *item, TABLE_LIST *tables,
   uint length=(uint) strlen(name);
   char name_buff[NAME_LEN+1];
 
-
   if (item->cached_table)
   {
     /*
@@ -2693,10 +2702,13 @@ find_field_in_tables(THD *thd, Item_ident *item, TABLE_LIST *tables,
     db= name_buff;
   }
 
+  bool search_global= item->item_flags & MY_ITEM_PREFER_1ST_TABLE;
   if (table_name && table_name[0])
   {						/* Qualified field */
-    bool found_table=0;
-    for (; tables; tables= tables->next_local)
+    bool found_table=0;    
+    uint table_idx= 0;
+    for (; tables; tables= search_global?tables->next_global:tables->next_local,
+        table_idx++)
     {
       /* TODO; Ensure that db and tables->db always points to something ! */
       if (!my_strcasecmp(table_alias_charset, tables->alias, table_name) &&
@@ -2732,6 +2744,8 @@ find_field_in_tables(THD *thd, Item_ident *item, TABLE_LIST *tables,
 	    return (Field*) 0;
 	  }
 	  found=find;
+          if (table_idx == 0 && item->item_flags & MY_ITEM_PREFER_1ST_TABLE) 
+            break;
 	}
       }
     }
@@ -2756,9 +2770,10 @@ find_field_in_tables(THD *thd, Item_ident *item, TABLE_LIST *tables,
 	return (Field*) not_found_field;
     return (Field*) 0;
   }
-
   bool allow_rowid= tables && !tables->next_local;	// Only one table
-  for (; tables ; tables= tables->next_local)
+  uint table_idx= 0;
+  for (; tables ; tables= search_global?tables->next_global:tables->next_local, 
+      table_idx++)
   {
     if (!tables->table && !tables->ancestor)
     {
@@ -2793,7 +2808,9 @@ find_field_in_tables(THD *thd, Item_ident *item, TABLE_LIST *tables,
           my_error(ER_NON_UNIQ_ERROR, MYF(0), name, thd->where);
 	return (Field*) 0;
       }
-      found=field;
+      found= field;
+      if (table_idx == 0 && item->item_flags & MY_ITEM_PREFER_1ST_TABLE) 
+        break;
     }
   }
   if (found)
@@ -3020,7 +3037,7 @@ int setup_wild(THD *thd, TABLE_LIST *tables, List<Item> &fields,
 
   Item *item;
   List_iterator<Item> it(fields);
-  Item_arena *arena, backup;
+  Query_arena *arena, backup;
   DBUG_ENTER("setup_wild");
 
   /*
@@ -3051,7 +3068,7 @@ int setup_wild(THD *thd, TABLE_LIST *tables, List<Item> &fields,
       }
       else if (insert_fields(thd,tables,((Item_field*) item)->db_name,
                              ((Item_field*) item)->table_name, &it,
-                             any_privileges, arena != 0))
+                             any_privileges))
       {
 	if (arena)
 	  thd->restore_backup_item_arena(arena, &backup);
@@ -3304,8 +3321,6 @@ bool get_key_map_from_key_list(key_map *map, TABLE *table,
     any_privileges	0 If we should ensure that we have SELECT privileges
 		          for all columns
                         1 If any privilege is ok
-    allocate_view_names if true view names will be copied to current Item_arena
-                        memory (made for SP/PS)
   RETURN
     0	ok
         'it' is updated to point at last inserted
@@ -3315,7 +3330,7 @@ bool get_key_map_from_key_list(key_map *map, TABLE *table,
 bool
 insert_fields(THD *thd, TABLE_LIST *tables, const char *db_name,
 	      const char *table_name, List_iterator<Item> *it,
-              bool any_privileges, bool allocate_view_names)
+              bool any_privileges)
 {
   /* allocate variables on stack to avoid pool alloaction */
   Field_iterator_table table_iter;
@@ -3506,25 +3521,6 @@ insert_fields(THD *thd, TABLE_LIST *tables, const char *db_name,
           field->query_id=thd->query_id;
           table->used_keys.intersect(field->part_of_key);
         }
-        else if (allocate_view_names &&
-                 thd->lex->current_select->first_execution)
-        {
-	  Item_field *item;
-	  if (alias_used)
-	    item= new Item_field(0,
-				 thd->strdup(tables->alias),
-				 thd->strdup(field_name));
-	  else
-	    item= new Item_field(thd->strdup(tables->view_db.str),
-				 thd->strdup(tables->view_name.str),
-				 thd->strdup(field_name));
-          /*
-            during cleunup() this item will be put in list to replace
-            expression from VIEW
-          */
-          thd->nocheck_register_item_tree_change(it->ref(), item,
-                                                 thd->mem_root);
-        }
       }
       /*
 	All fields are used in case if usual tables (in case of view used
@@ -3561,7 +3557,7 @@ err:
 int setup_conds(THD *thd, TABLE_LIST *tables, TABLE_LIST *leaves, COND **conds)
 {
   SELECT_LEX *select_lex= thd->lex->current_select;
-  Item_arena *arena= thd->current_arena, backup;
+  Query_arena *arena= thd->current_arena, backup;
   bool save_wrapper= thd->lex->current_select->no_wrap_view_item;
   TABLE_LIST *table= NULL;	// For HP compilers
   DBUG_ENTER("setup_conds");
diff --git a/sql/sql_class.cc b/sql/sql_class.cc
index 165ce61d5d1..20f48da9283 100644
--- a/sql/sql_class.cc
+++ b/sql/sql_class.cc
@@ -49,7 +49,7 @@ char internal_table_name[2]= "*";
 ** Instansiate templates
 *****************************************************************************/
 
-#ifdef __GNUC__
+#ifdef HAVE_EXPLICIT_TEMPLATE_INSTANTIATION
 /* Used templates */
 template class List<Key>;
 template class List_iterator<Key>;
@@ -156,18 +156,21 @@ bool foreign_key_prefix(Key *a, Key *b)
 /****************************************************************************
 ** Thread specific functions
 ****************************************************************************/
+/*
+  Pass nominal parameters to Statement constructor only to ensure that
+  the destructor works OK in case of error. The main_mem_root will be
+  re-initialized in init().
+*/
 
 THD::THD()
-  :user_time(0), global_read_lock(0), is_fatal_error(0),
+  :Statement(CONVENTIONAL_EXECUTION, 0, ALLOC_ROOT_MIN_BLOCK_SIZE, 0),
+   user_time(0), global_read_lock(0), is_fatal_error(0),
    rand_used(0), time_zone_used(0),
    last_insert_id_used(0), insert_id_used(0), clear_next_insert_id(0),
    in_lock_tables(0), bootstrap(0), derived_tables_processing(FALSE),
    spcont(NULL)
 {
   current_arena= this;
-#ifndef DBUG_OFF
-  backup_arena= 0;
-#endif
   host= user= priv_user= db= ip= 0;
   catalog= (char*)"std"; // the only catalog we have for now
   host_or_ip= "connecting host";
@@ -420,8 +423,6 @@ THD::~THD()
 #ifndef DBUG_OFF
   dbug_sentry= THD_SENTRY_GONE;
 #endif  
-  /* Reset stmt_backup.mem_root to not double-free memory from thd.mem_root */
-  clear_alloc_root(&stmt_backup.main_mem_root);
   DBUG_VOID_RETURN;
 }
 
@@ -524,7 +525,7 @@ void THD::cleanup_after_query()
     next_insert_id= 0;
   }
   /* Free Items that were created during this execution */
-  free_items(free_list);
+  free_items();
   /*
     In the rest of code we assume that free_list never points to garbage:
     Keep this predicate true.
@@ -1474,56 +1475,25 @@ void select_dumpvar::cleanup()
 }
 
 
-/*
-  Create arena for already constructed THD.
-
-  SYNOPSYS
-    Item_arena()
-      thd - thread for which arena is created
-
-  DESCRIPTION
-    Create arena for already existing THD using its variables as parameters
-    for memory root initialization.
-*/
-Item_arena::Item_arena(THD* thd)
-  :free_list(0), mem_root(&main_mem_root),
-   state(INITIALIZED)
-{
-  init_sql_alloc(&main_mem_root,
-                 thd->variables.query_alloc_block_size,
-                 thd->variables.query_prealloc_size);
-}
-
-
-/*
-  Create arena and optionally initialize memory root.
-
-  SYNOPSYS
-    Item_arena()
-      init_mem_root - whenever we need to initialize memory root
-
-  DESCRIPTION
-    Create arena and optionally initialize memory root with minimal
-    possible parameters.
-
-  NOTE
-    We use this constructor when arena is part of THD, but reinitialize
-    its memory root in THD::init_for_queries() before execution of real
-    statements.
-*/
-Item_arena::Item_arena(bool init_mem_root)
-  :free_list(0), mem_root(&main_mem_root),
-  state(CONVENTIONAL_EXECUTION)
+Query_arena::Type Query_arena::type() const
 {
-  if (init_mem_root)
-    init_sql_alloc(&main_mem_root, ALLOC_ROOT_MIN_BLOCK_SIZE, 0);
+  DBUG_ASSERT(0); /* Should never be called */
+  return STATEMENT;
 }
 
 
-Item_arena::Type Item_arena::type() const
+void Query_arena::free_items()
 {
-  DBUG_ASSERT(0); /* Should never be called */
-  return STATEMENT;
+  Item *next;
+  DBUG_ENTER("Query_arena::free_items");
+  /* This works because items are allocated with sql_alloc() */
+  for (; free_list; free_list= next)
+  {
+    next= free_list->next;
+    free_list->delete_self();
+  }
+  /* Postcondition: free_list is 0 */
+  DBUG_VOID_RETURN;
 }
 
 
@@ -1531,9 +1501,10 @@ Item_arena::Type Item_arena::type() const
   Statement functions 
 */
 
-Statement::Statement(THD *thd)
-  :Item_arena(thd),
-  id(++thd->statement_id_counter),
+Statement::Statement(enum enum_state state_arg, ulong id_arg,
+                     ulong alloc_block_size, ulong prealloc_size)
+  :Query_arena(&main_mem_root, state_arg),
+  id(id_arg),
   set_query_id(1),
   allow_sum_func(0),
   lex(&main_lex),
@@ -1542,28 +1513,11 @@ Statement::Statement(THD *thd)
   cursor(0)
 {
   name.str= NULL;
-}
-
-/*
-  This constructor is called when statement is a subobject of THD:
-  Some variables are initialized in THD::init due to locking problems
-  This statement object will be used to 
-*/
-
-Statement::Statement()
-  :Item_arena((bool)TRUE),
-  id(0),
-  set_query_id(1),
-  allow_sum_func(0),                            /* initialized later */
-  lex(&main_lex),
-  query(0),                                     /* these two are set */ 
-  query_length(0),                              /* in alloc_query() */
-  cursor(0)
-{
+  init_sql_alloc(&main_mem_root, alloc_block_size, prealloc_size);
 }
 
 
-Item_arena::Type Statement::type() const
+Query_arena::Type Statement::type() const
 {
   return STATEMENT;
 }
@@ -1611,41 +1565,32 @@ void THD::end_statement()
 }
 
 
-void Item_arena::set_n_backup_item_arena(Item_arena *set, Item_arena *backup)
+void Query_arena::set_n_backup_item_arena(Query_arena *set, Query_arena *backup)
 {
-  DBUG_ENTER("Item_arena::set_n_backup_item_arena");
-  DBUG_ASSERT(backup_arena == 0);
+  DBUG_ENTER("Query_arena::set_n_backup_item_arena");
+  DBUG_ASSERT(backup->is_backup_arena == FALSE);
   backup->set_item_arena(this);
   set_item_arena(set);
 #ifndef DBUG_OFF
-  backup_arena= 1;
+  backup->is_backup_arena= TRUE;
 #endif
   DBUG_VOID_RETURN;
 }
 
 
-void Item_arena::restore_backup_item_arena(Item_arena *set, Item_arena *backup)
+void Query_arena::restore_backup_item_arena(Query_arena *set, Query_arena *backup)
 {
-  DBUG_ENTER("Item_arena::restore_backup_item_arena");
+  DBUG_ENTER("Query_arena::restore_backup_item_arena");
+  DBUG_ASSERT(backup->is_backup_arena);
   set->set_item_arena(this);
   set_item_arena(backup);
 #ifndef DBUG_OFF
-  backup_arena= 0;
-#endif
-#ifdef NOT_NEEDED_NOW
-  /*
-    Reset backup mem_root to avoid its freeing.
-    Since Item_arena's mem_root is freed only when it is part of Statement
-    we need this only if we use some Statement's arena as backup storage.
-    But we do this only with THD::stmt_backup and this Statement is specially
-    handled in this respect. So this code is not really needed now.
-  */
-  clear_alloc_root(&backup->mem_root);
+  backup->is_backup_arena= FALSE;
 #endif
   DBUG_VOID_RETURN;
 }
 
-void Item_arena::set_item_arena(Item_arena *set)
+void Query_arena::set_item_arena(Query_arena *set)
 {
   mem_root=  set->mem_root;
   free_list= set->free_list;
@@ -1654,6 +1599,11 @@ void Item_arena::set_item_arena(Item_arena *set)
 
 Statement::~Statement()
 {
+  /*
+    We must free `main_mem_root', not `mem_root' (pointer), to work
+    correctly if this statement is used as a backup statement,
+    for which `mem_root' may point to some other statement.
+  */
   free_root(&main_mem_root, MYF(0));
 }
 
diff --git a/sql/sql_class.h b/sql/sql_class.h
index d2095503fe5..1f232b9ca21 100644
--- a/sql/sql_class.h
+++ b/sql/sql_class.h
@@ -631,6 +631,13 @@ typedef struct system_status_var
   ulong filesort_range_count;
   ulong filesort_rows;
   ulong filesort_scan_count;
+  /* Ppepared statements and binary protocol */
+  ulong com_stmt_prepare;
+  ulong com_stmt_execute;
+  ulong com_stmt_send_long_data;
+  ulong com_stmt_fetch;
+  ulong com_stmt_reset;
+  ulong com_stmt_close;
 
   double last_query_cost;
 } STATUS_VAR;
@@ -640,13 +647,13 @@ typedef struct system_status_var
   variable in system_status_var
 */
 
-#define last_system_status_var filesort_scan_count
+#define last_system_status_var com_stmt_close
 
 
 void free_tmp_table(THD *thd, TABLE *entry);
 
 
-class Item_arena
+class Query_arena
 {
 public:
   /*
@@ -654,17 +661,19 @@ public:
     itself to the list on creation (see Item::Item() for details))
   */
   Item *free_list;
-  MEM_ROOT main_mem_root;
   MEM_ROOT *mem_root;                   // Pointer to current memroot
 #ifndef DBUG_OFF
-  bool backup_arena;
+  bool is_backup_arena; /* True if this arena is used for backup. */
+#define INIT_ARENA_DBUG_INFO is_backup_arena= 0
+#else 
+#define INIT_ARENA_DBUG_INFO  
 #endif
-  enum enum_state 
+  enum enum_state
   {
     INITIALIZED= 0, INITIALIZED_FOR_SP= 1, PREPARED= 2,
     CONVENTIONAL_EXECUTION= 3, EXECUTED= 4, ERROR= -1
   };
-  
+
   enum_state state;
 
   /* We build without RTTI, so dynamic_cast can't be used. */
@@ -673,25 +682,22 @@ public:
     STATEMENT, PREPARED_STATEMENT, STORED_PROCEDURE
   };
 
+  Query_arena(MEM_ROOT *mem_root_arg, enum enum_state state_arg) :
+    free_list(0), mem_root(mem_root_arg), state(state_arg)
+  { INIT_ARENA_DBUG_INFO; }
   /*
-    This constructor is used only when Item_arena is created as
-    backup storage for another instance of Item_arena.
-  */
-  Item_arena() {};
-  /*
-    Create arena for already constructed THD using its variables as
-    parameters for memory root initialization.
+    This constructor is used only when Query_arena is created as
+    backup storage for another instance of Query_arena.
   */
-  Item_arena(THD *thd);
-  /*
-    Create arena and optionally init memory root with minimal values.
-    Particularly used if Item_arena is part of Statement.
-  */
-  Item_arena(bool init_mem_root);
+  Query_arena() { INIT_ARENA_DBUG_INFO; }
+
+#undef INIT_ARENA_DBUG_INFO
   virtual Type type() const;
-  virtual ~Item_arena() {};
+  virtual ~Query_arena() {};
 
   inline bool is_stmt_prepare() const { return state == INITIALIZED; }
+  inline bool is_first_sp_execute() const
+  { return state == INITIALIZED_FOR_SP; }
   inline bool is_stmt_prepare_or_first_sp_execute() const
   { return (int)state < (int)PREPARED; }
   inline bool is_first_stmt_execute() const { return state == PREPARED; }
@@ -699,6 +705,7 @@ public:
   { return state == PREPARED || state == EXECUTED; }
   inline bool is_conventional() const
   { return state == CONVENTIONAL_EXECUTION; }
+
   inline gptr alloc(unsigned int size) { return alloc_root(mem_root,size); }
   inline gptr calloc(unsigned int size)
   {
@@ -721,9 +728,11 @@ public:
     return ptr;
   }
 
-  void set_n_backup_item_arena(Item_arena *set, Item_arena *backup);
-  void restore_backup_item_arena(Item_arena *set, Item_arena *backup);
-  void set_item_arena(Item_arena *set);
+  void set_n_backup_item_arena(Query_arena *set, Query_arena *backup);
+  void restore_backup_item_arena(Query_arena *set, Query_arena *backup);
+  void set_item_arena(Query_arena *set);
+
+  void free_items();
 };
 
 
@@ -743,12 +752,13 @@ class Cursor;
   be used explicitly.
 */
 
-class Statement: public Item_arena
+class Statement: public Query_arena
 {
   Statement(const Statement &rhs);              /* not implemented: */
   Statement &operator=(const Statement &rhs);   /* non-copyable */
 public:
-  /* FIXME: must be private */
+  /* FIXME: these must be protected */
+  MEM_ROOT main_mem_root;
   LEX     main_lex;
 
   /*
@@ -806,13 +816,11 @@ public:
 
 public:
 
-  /*
-    This constructor is called when statement is a subobject of THD:
-    some variables are initialized in THD::init due to locking problems
-  */
-  Statement();
+  /* This constructor is called for backup statements */
+  Statement() { clear_alloc_root(&main_mem_root); }
 
-  Statement(THD *thd);
+  Statement(enum enum_state state_arg, ulong id_arg,
+            ulong alloc_block_size, ulong prealloc_size);
   virtual ~Statement();
 
   /* Assign execution context (note: not all members) of given stmt to self */
@@ -955,11 +963,6 @@ public:
   /* all prepared statements and cursors of this connection */
   Statement_map stmt_map; 
   /*
-    keeps THD state while it is used for active statement
-    Note: we perform special cleanup for it in THD destructor.
-  */
-  Statement stmt_backup;
-  /*
     A pointer to the stack frame of handle_one_connection(),
     which is called first in the thread for handling a client
   */
@@ -1039,7 +1042,7 @@ public:
 #endif
   struct st_my_thread_var *mysys_var;
   /*
-    Type of current query: COM_PREPARE, COM_QUERY, etc. Set from 
+    Type of current query: COM_STMT_PREPARE, COM_QUERY, etc. Set from
     first byte of the packet in do_command()
   */
   enum enum_server_command command;
@@ -1107,9 +1110,9 @@ public:
   Item_change_list change_list;
 
   /*
-    Current prepared Item_arena if there one, or 0
+    Current prepared Query_arena if there one, or 0
   */
-  Item_arena *current_arena;
+  Query_arena *current_arena;
   /*
     next_insert_id is set on SET INSERT_ID= #. This is used as the next
     generated auto_increment value in handler.cc
@@ -1186,7 +1189,7 @@ public:
   bool       query_error, bootstrap, cleanup_done;
   bool	     tmp_table_used;
   bool	     charset_is_system_charset, charset_is_collation_connection;
-  bool       slow_command;
+  bool       enable_slow_log;   /* enable slow log for current statement */
   bool	     no_trans_update, abort_on_warning;
   bool 	     got_warning;       /* Set on call to push_warning() */
   bool	     no_warnings_for_error; /* no warnings on call to my_error() */
@@ -1331,13 +1334,9 @@ public:
     return 0;
 #endif
   }
-  inline bool only_prepare()
-  {
-    return command == COM_PREPARE;
-  }
   inline bool fill_derived_tables()
   {
-    return !only_prepare() && !lex->only_view_structure();
+    return !current_arena->is_stmt_prepare() && !lex->only_view_structure();
   }
   inline gptr trans_alloc(unsigned int size)
   {
@@ -1370,20 +1369,20 @@ public:
   inline void fatal_error()
   {
     is_fatal_error= 1;
-    net.report_error= 1; 
+    net.report_error= 1;
     DBUG_PRINT("error",("Fatal error set"));
   }
   inline CHARSET_INFO *charset() { return variables.character_set_client; }
   void update_charset();
 
-  inline Item_arena *change_arena_if_needed(Item_arena *backup)
+  inline Query_arena *change_arena_if_needed(Query_arena *backup)
   {
     /*
       use new arena if we are in a prepared statements and we have not
       already changed to use this arena.
     */
     if (!current_arena->is_conventional() &&
-        mem_root != &current_arena->main_mem_root)
+        mem_root != current_arena->mem_root)
     {
       set_n_backup_item_arena(current_arena, backup);
       return current_arena;
@@ -1428,10 +1427,10 @@ public:
 };
 
 #define tmp_disable_binlog(A)       \
-  ulong save_options= (A)->options; \
-  (A)->options&= ~OPTION_BIN_LOG;
+  {ulong tmp_disable_binlog__save_options= (A)->options; \
+  (A)->options&= ~OPTION_BIN_LOG
 
-#define reenable_binlog(A)          (A)->options= save_options;
+#define reenable_binlog(A)   (A)->options= tmp_disable_binlog__save_options;}
 
 /* Flags for the THD::system_thread (bitmap) variable */
 #define SYSTEM_THREAD_DELAYED_INSERT 1
diff --git a/sql/sql_delete.cc b/sql/sql_delete.cc
index d782744162d..25fef8b0ad6 100644
--- a/sql/sql_delete.cc
+++ b/sql/sql_delete.cc
@@ -790,7 +790,7 @@ bool mysql_truncate(THD *thd, TABLE_LIST *table_list, bool dont_send_ok)
   if (!dont_send_ok)
   {
     db_type table_type;
-    if ((table_type=get_table_type(path)) == DB_TYPE_UNKNOWN)
+    if ((table_type=get_table_type(thd, path)) == DB_TYPE_UNKNOWN)
     {
       my_error(ER_NO_SUCH_TABLE, MYF(0),
                table_list->db, table_list->table_name);
diff --git a/sql/sql_handler.cc b/sql/sql_handler.cc
index 9356ee04c45..e905f93b860 100644
--- a/sql/sql_handler.cc
+++ b/sql/sql_handler.cc
@@ -422,7 +422,7 @@ bool mysql_ha_read(THD *thd, TABLE_LIST *tables,
     }
   }
 
-  if (insert_fields(thd, tables, tables->db, tables->alias, &it, 0, 0))
+  if (insert_fields(thd, tables, tables->db, tables->alias, &it, 0))
     goto err0;
 
   protocol->send_fields(&list, Protocol::SEND_NUM_ROWS | Protocol::SEND_EOF);
diff --git a/sql/sql_insert.cc b/sql/sql_insert.cc
index 5ca75554c6f..2ce81d8815e 100644
--- a/sql/sql_insert.cc
+++ b/sql/sql_insert.cc
@@ -1988,10 +1988,22 @@ select_insert::select_insert(TABLE_LIST *table_list_par, TABLE *table_par,
 int
 select_insert::prepare(List<Item> &values, SELECT_LEX_UNIT *u)
 {
+  int res;
+  LEX *lex= thd->lex;
+  SELECT_LEX *lex_current_select_save= lex->current_select;
   DBUG_ENTER("select_insert::prepare");
 
   unit= u;
-  if (check_insert_fields(thd, table_list, *fields, values, !insert_into_view))
+  /*
+    Since table in which we are going to insert is added to the first
+    select, LEX::current_select should point to the first select while
+    we are fixing fields from insert list.
+  */
+  lex->current_select= &lex->select_lex;
+  res= check_insert_fields(thd, table_list, *fields, values,
+                           !insert_into_view);
+  lex->current_select= lex_current_select_save;
+  if (res)
     DBUG_RETURN(1);
   /*
     if it is INSERT into join view then check_insert_fields already found
@@ -2003,12 +2015,12 @@ select_insert::prepare(List<Item> &values, SELECT_LEX_UNIT *u)
     Is table which we are changing used somewhere in other parts of
     query
   */
-  if (!(thd->lex->current_select->options & OPTION_BUFFER_RESULT) &&
+  if (!(lex->current_select->options & OPTION_BUFFER_RESULT) &&
       unique_table(table_list, table_list->next_global))
   {
     /* Using same table for INSERT and SELECT */
-    thd->lex->current_select->options|= OPTION_BUFFER_RESULT;
-    thd->lex->current_select->join->select_options|= OPTION_BUFFER_RESULT;
+    lex->current_select->options|= OPTION_BUFFER_RESULT;
+    lex->current_select->join->select_options|= OPTION_BUFFER_RESULT;
   }
   else
   {
@@ -2107,9 +2119,12 @@ bool select_insert::send_data(List<Item> &values)
   }
   if (!(error= write_record(thd, table, &info)))
   {
-    if (table->triggers)
+    if (table->triggers || info.handle_duplicates == DUP_UPDATE)
     {
       /*
+        Restore fields of the record since it is possible that they were
+        changed by ON DUPLICATE KEY UPDATE clause.
+    
         If triggers exist then whey can modify some fields which were not
         originally touched by INSERT ... SELECT, so we have to restore
         their original values for the next row.
@@ -2362,11 +2377,11 @@ void select_create::abort()
   Instansiate templates
 *****************************************************************************/
 
-#ifdef __GNUC__
+#ifdef HAVE_EXPLICIT_TEMPLATE_INSTANTIATION
 template class List_iterator_fast<List_item>;
 #ifndef EMBEDDED_LIBRARY
 template class I_List<delayed_insert>;
 template class I_List_iterator<delayed_insert>;
 template class I_List<delayed_row>;
 #endif /* EMBEDDED_LIBRARY */
-#endif /* __GNUC__ */
+#endif /* HAVE_EXPLICIT_TEMPLATE_INSTANTIATION */
diff --git a/sql/sql_lex.cc b/sql/sql_lex.cc
index 1270aab18ae..08f0c3badf7 100644
--- a/sql/sql_lex.cc
+++ b/sql/sql_lex.cc
@@ -967,7 +967,7 @@ int yylex(void *arg, void *yythd)
       {
         THD* thd= (THD*)yythd;
         if ((thd->client_capabilities & CLIENT_MULTI_STATEMENTS) && 
-            (thd->command != COM_PREPARE))
+            (thd->command != COM_STMT_PREPARE))
         {
 	  lex->safe_to_cache_query= 0;
           lex->found_semicolon=(char*) lex->ptr;
@@ -1484,8 +1484,8 @@ bool st_select_lex::setup_ref_array(THD *thd, uint order_group_num)
     We have to create array in prepared statement memory if it is
     prepared statement
   */
-  Item_arena *arena= thd->current_arena;
-  return (ref_pointer_array= 
+  Query_arena *arena= thd->current_arena;
+  return (ref_pointer_array=
           (Item **)arena->alloc(sizeof(Item*) *
                                 (item_list.elements +
                                  select_n_having_items +
diff --git a/sql/sql_lex.h b/sql/sql_lex.h
index 8af416f0ce8..5cf0b66598f 100644
--- a/sql/sql_lex.h
+++ b/sql/sql_lex.h
@@ -114,6 +114,14 @@ enum enum_sp_data_access
   SP_MODIFIES_SQL_DATA
 };
 
+const LEX_STRING sp_data_access_name[]=
+{
+  { (char*) STRING_WITH_LEN("") },
+  { (char*) STRING_WITH_LEN("CONTAINS SQL") },
+  { (char*) STRING_WITH_LEN("NO SQL") },
+  { (char*) STRING_WITH_LEN("READS SQL DATA") },
+  { (char*) STRING_WITH_LEN("MODIFIES SQL DATA") }
+};
 
 #define DERIVED_SUBQUERY	1
 #define DERIVED_VIEW		2
@@ -522,7 +530,19 @@ public:
     query processing end even if we use temporary table
   */
   bool subquery_in_having;
-  bool first_execution; /* first execution in SP or PS */
+  /*
+    This variable is required to ensure proper work of subqueries and
+    stored procedures. Generally, one should use the states of
+    Query_arena to determine if it's a statement prepare or first
+    execution of a stored procedure. However, in case when there was an
+    error during the first execution of a stored procedure, the SP body
+    is not expelled from the SP cache. Therefore, a deeply nested
+    subquery might be left unoptimized. So we need this per-subquery
+    variable to inidicate the optimization/execution state of every
+    subquery. Prepared statements work OK in that regard, as in
+    case of an error during prepare the PS is not created.
+  */
+  bool first_execution;
   bool first_cond_optimization;
   /* do not wrap view fields with Item_ref */
   bool no_wrap_view_item;
@@ -622,6 +642,11 @@ public:
   static void print_order(String *str, ORDER *order);
   void print_limit(THD *thd, String *str);
   void fix_prepare_information(THD *thd, Item **conds);
+  /*
+    Destroy the used execution plan (JOIN) of this subtree (this
+    SELECT_LEX and all nested SELECT_LEXes and SELECT_LEX_UNITs).
+  */
+  bool cleanup();
 };
 typedef class st_select_lex SELECT_LEX;
 
diff --git a/sql/sql_map.cc b/sql/sql_map.cc
index 9346f3df305..56b4b765355 100644
--- a/sql/sql_map.cc
+++ b/sql/sql_map.cc
@@ -138,7 +138,7 @@ void unmap_file(mapped_files *map)
 ** Instansiate templates
 *****************************************************************************/
 
-#ifdef __GNUC__
+#ifdef HAVE_EXPLICIT_TEMPLATE_INSTANTIATION
 /* Used templates */
 template class I_List<mapped_files>;
 template class I_List_iterator<mapped_files>;
diff --git a/sql/sql_parse.cc b/sql/sql_parse.cc
index be7ba7d571d..d6a719e65f9 100644
--- a/sql/sql_parse.cc
+++ b/sql/sql_parse.cc
@@ -71,7 +71,6 @@ static void remove_escape(char *name);
 static void refresh_status(void);
 static bool append_file_to_dir(THD *thd, const char **filename_ptr,
 			       const char *table_name);
-static void log_slow_query(THD *thd);
 
 const char *any_db="*any*";	// Special symbol for check_access
 
@@ -1508,10 +1507,10 @@ bool dispatch_command(enum enum_server_command command, THD *thd,
 
   thd->command=command;
   /*
-    Commands which will always take a long time should be marked with
-    this so that they will not get logged to the slow query log
+    Commands which always take a long time are logged into
+    the slow log only if opt_log_slow_admin_statements is set.
   */
-  thd->slow_command=FALSE;
+  thd->enable_slow_log= TRUE;
   thd->lex->sql_command= SQLCOM_END; /* to avoid confusing VIEW detectors */
   thd->set_time();
   VOID(pthread_mutex_lock(&LOCK_thread_count));
@@ -1551,7 +1550,7 @@ bool dispatch_command(enum enum_server_command command, THD *thd,
     uint tbl_len= *(uchar*) (packet + db_len + 1);
 
     statistic_increment(thd->status_var.com_other, &LOCK_status);
-    thd->slow_command= TRUE;
+    thd->enable_slow_log= opt_log_slow_admin_statements;
     db= thd->alloc(db_len + tbl_len + 2);
     tbl_name= strmake(db, packet + 1, db_len)+1;
     strmake(tbl_name, packet + db_len + 2, tbl_len);
@@ -1635,32 +1634,32 @@ bool dispatch_command(enum enum_server_command command, THD *thd,
     }
     break;
   }
-  case COM_EXECUTE:
+  case COM_STMT_EXECUTE:
   {
     mysql_stmt_execute(thd, packet, packet_length);
     break;
   }
-  case COM_FETCH:
+  case COM_STMT_FETCH:
   {
     mysql_stmt_fetch(thd, packet, packet_length);
     break;
   }
-  case COM_LONG_DATA:
+  case COM_STMT_SEND_LONG_DATA:
   {
     mysql_stmt_get_longdata(thd, packet, packet_length);
     break;
   }
-  case COM_PREPARE:
+  case COM_STMT_PREPARE:
   {
     mysql_stmt_prepare(thd, packet, packet_length, 0);
     break;
   }
-  case COM_CLOSE_STMT:
+  case COM_STMT_CLOSE:
   {
-    mysql_stmt_free(thd, packet);
+    mysql_stmt_close(thd, packet);
     break;
   }
-  case COM_RESET_STMT:
+  case COM_STMT_RESET:
   {
     mysql_stmt_reset(thd, packet);
     break;
@@ -1689,7 +1688,7 @@ bool dispatch_command(enum enum_server_command command, THD *thd,
 #endif
       ulong length= (ulong)(packet_end-packet);
 
-      log_slow_query(thd);
+      log_slow_statement(thd);
 
       /* Remove garbage at start of query */
       while (my_isspace(thd->charset(), *packet) && length > 0)
@@ -1858,7 +1857,7 @@ bool dispatch_command(enum enum_server_command command, THD *thd,
       uint32 slave_server_id;
 
       statistic_increment(thd->status_var.com_other,&LOCK_status);
-      thd->slow_command = TRUE;
+      thd->enable_slow_log= opt_log_slow_admin_statements;
       if (check_global_access(thd, REPL_SLAVE_ACL))
 	break;
 
@@ -2043,7 +2042,7 @@ bool dispatch_command(enum enum_server_command command, THD *thd,
   if (thd->net.report_error)
     net_send_error(thd);
 
-  log_slow_query(thd);
+  log_slow_statement(thd);
 
   thd->proc_info="cleaning up";
   VOID(pthread_mutex_lock(&LOCK_thread_count)); // For process list
@@ -2059,13 +2058,16 @@ bool dispatch_command(enum enum_server_command command, THD *thd,
 }
 
 
-static void log_slow_query(THD *thd)
+void log_slow_statement(THD *thd)
 {
   time_t start_of_query=thd->start_time;
   thd->end_time();				// Set start time
 
-  /* If not reading from backup and if the query took too long */
-  if (!thd->slow_command && !thd->user_time) // do not log 'slow_command' queries
+  /*
+    Do not log administrative statements unless the appropriate option is
+    set; do not log into slow log if reading from backup.
+  */
+  if (thd->enable_slow_log && !thd->user_time)
   {
     thd->proc_info="logging slow query";
 
@@ -2197,7 +2199,7 @@ int prepare_schema_table(THD *thd, LEX *lex, Table_ident *table_ident,
 
 /*
   Read query from packet and store in thd->query
-  Used in COM_QUERY and COM_PREPARE
+  Used in COM_QUERY and COM_STMT_PREPARE
 
   DESCRIPTION
     Sets the following THD variables:
@@ -2499,7 +2501,7 @@ mysql_execute_command(THD *thd)
                           lex->prepared_stmt_name.str,
                           query_len, query_str));
     }
-    thd->command= COM_PREPARE;
+    thd->command= COM_STMT_PREPARE;
     if (!(res= mysql_stmt_prepare(thd, query_str, query_len + 1,
                                   &lex->prepared_stmt_name)))
       send_ok(thd, 0L, 0L, "Statement prepared");
@@ -2520,6 +2522,8 @@ mysql_execute_command(THD *thd)
     DBUG_PRINT("info", ("DEALLOCATE PREPARE: %.*s\n", 
                         lex->prepared_stmt_name.length,
                         lex->prepared_stmt_name.str));
+    /* We account deallocate in the same manner as mysql_stmt_close */
+    statistic_increment(thd->status_var.com_stmt_close, &LOCK_status);
     if ((stmt= thd->stmt_map.find_by_name(&lex->prepared_stmt_name)))
     {
       thd->stmt_map.erase(stmt);
@@ -2637,7 +2641,7 @@ mysql_execute_command(THD *thd)
 	check_table_access(thd, SELECT_ACL, all_tables, 0) ||
 	check_global_access(thd, FILE_ACL))
       goto error; /* purecov: inspected */
-    thd->slow_command=TRUE;
+    thd->enable_slow_log= opt_log_slow_admin_statements;
     res = mysql_backup_table(thd, first_table);
 
     break;
@@ -2649,7 +2653,7 @@ mysql_execute_command(THD *thd)
 	check_table_access(thd, INSERT_ACL, all_tables, 0) ||
 	check_global_access(thd, FILE_ACL))
       goto error; /* purecov: inspected */
-    thd->slow_command=TRUE;
+    thd->enable_slow_log= opt_log_slow_admin_statements;
     res = mysql_restore_table(thd, first_table);
     break;
   }
@@ -2764,6 +2768,20 @@ mysql_execute_command(THD *thd)
 
   case SQLCOM_CREATE_TABLE:
   {
+    /* If CREATE TABLE of non-temporary table, do implicit commit */
+    if (!(lex->create_info.options & HA_LEX_CREATE_TMP_TABLE))
+    {
+      if (end_active_trans(thd))
+      {
+	res= -1;
+	break;
+      }
+    }
+    else 
+    {
+      /* So that CREATE TEMPORARY TABLE gets to binlog at commit/rollback */
+      thd->options|= OPTION_STATUS_NO_TRANS_UPDATE;
+    }
     DBUG_ASSERT(first_table == all_tables && first_table != 0);
     bool link_to_local;
     // Skip first table, which is the table we are creating
@@ -2906,7 +2924,7 @@ end_with_restore_list:
     DBUG_ASSERT(first_table == all_tables && first_table != 0);
     if (check_one_table_access(thd, INDEX_ACL, all_tables))
       goto error; /* purecov: inspected */
-    thd->slow_command=TRUE;
+    thd->enable_slow_log= opt_log_slow_admin_statements;
     if (end_active_trans(thd))
       goto error;
     else
@@ -2995,7 +3013,7 @@ end_with_restore_list:
 	goto error;
       else
       {
-        thd->slow_command=TRUE;
+        thd->enable_slow_log= opt_log_slow_admin_statements;
 	res= mysql_alter_table(thd, select_lex->db, lex->name,
 			       &lex->create_info,
 			       first_table, lex->create_list,
@@ -3094,7 +3112,7 @@ end_with_restore_list:
     if (check_db_used(thd, all_tables) ||
 	check_table_access(thd, SELECT_ACL | INSERT_ACL, all_tables, 0))
       goto error; /* purecov: inspected */
-    thd->slow_command=TRUE;
+    thd->enable_slow_log= opt_log_slow_admin_statements;
     res= mysql_repair_table(thd, first_table, &lex->check_opt);
     /* ! we write after unlocking the table */
     if (!res && !lex->no_write_to_binlog)
@@ -3114,7 +3132,7 @@ end_with_restore_list:
     if (check_db_used(thd, all_tables) ||
 	check_table_access(thd, SELECT_ACL | EXTRA_ACL , all_tables, 0))
       goto error; /* purecov: inspected */
-    thd->slow_command=TRUE;
+    thd->enable_slow_log= opt_log_slow_admin_statements;
     res = mysql_check_table(thd, first_table, &lex->check_opt);
     break;
   }
@@ -3124,7 +3142,7 @@ end_with_restore_list:
     if (check_db_used(thd, all_tables) ||
 	check_table_access(thd, SELECT_ACL | INSERT_ACL, all_tables, 0))
       goto error; /* purecov: inspected */
-    thd->slow_command=TRUE;
+    thd->enable_slow_log= opt_log_slow_admin_statements;
     res = mysql_analyze_table(thd, first_table, &lex->check_opt);
     /* ! we write after unlocking the table */
     if (!res && !lex->no_write_to_binlog)
@@ -3145,7 +3163,7 @@ end_with_restore_list:
     if (check_db_used(thd, all_tables) ||
 	check_table_access(thd, SELECT_ACL | INSERT_ACL, all_tables, 0))
       goto error; /* purecov: inspected */
-    thd->slow_command=TRUE;
+    thd->enable_slow_log= opt_log_slow_admin_statements;
     res= (specialflag & (SPECIAL_SAFE_MODE | SPECIAL_NO_NEW_FUNC)) ?
       mysql_recreate_table(thd, first_table, 1) :
       mysql_optimize_table(thd, first_table, &lex->check_opt);
@@ -3254,10 +3272,14 @@ end_with_restore_list:
 
     if (first_table->view && !first_table->contain_auto_increment)
       thd->last_insert_id= 0; // do not show last insert ID if VIEW have not it
-
     break;
   }
   case SQLCOM_TRUNCATE:
+    if (end_active_trans(thd))
+    {
+      res= -1;
+      break;
+    }
     DBUG_ASSERT(first_table == all_tables && first_table != 0);
     if (check_one_table_access(thd, DELETE_ACL, all_tables))
       goto error;
@@ -3351,6 +3373,9 @@ end_with_restore_list:
       */
       if (thd->slave_thread)
 	lex->drop_if_exists= 1;
+
+      /* So that DROP TEMPORARY TABLE gets to binlog at commit/rollback */
+      thd->options|= OPTION_STATUS_NO_TRANS_UPDATE;
     }
     res= mysql_rm_table(thd, first_table, lex->drop_if_exists,
 			lex->drop_temporary);
@@ -3489,6 +3514,11 @@ end_with_restore_list:
     break;
   case SQLCOM_CREATE_DB:
   {
+    if (end_active_trans(thd))
+    {
+      res= -1;
+      break;
+    }
     char *alias;
     if (!(alias=thd->strdup(lex->name)) || check_db_name(lex->name))
     {
@@ -3519,6 +3549,11 @@ end_with_restore_list:
   }
   case SQLCOM_DROP_DB:
   {
+    if (end_active_trans(thd))
+    {
+      res= -1;
+      break;
+    }
     if (check_db_name(lex->name))
     {
       my_error(ER_WRONG_DB_NAME, MYF(0), lex->name);
@@ -5458,6 +5493,21 @@ bool add_field_to_list(THD *thd, char *field_name, enum_field_types type,
     DBUG_RETURN(1);
   }
 
+  if (type == FIELD_TYPE_TIMESTAMP && length)
+  {
+    /* Display widths are no longer supported for TIMSTAMP as of MySQL 4.1.
+       In other words, for declarations such as TIMESTAMP(2), TIMESTAMP(4),
+       and so on, the display width is ignored.
+    */
+    char buf[32];
+    my_snprintf(buf, sizeof(buf),
+                "TIMESTAMP(%s)", length, system_charset_info);
+    push_warning_printf(thd,MYSQL_ERROR::WARN_LEVEL_WARN,
+                        ER_WARN_DEPRECATED_SYNTAX,
+                        ER(ER_WARN_DEPRECATED_SYNTAX),
+                        buf, "TIMESTAMP");
+  }
+
   if (!(new_field= new_create_field(thd, field_name, type, length, decimals,
 		type_modifier, default_value, on_update_value,
 		comment, change, interval_list, cs, uint_geom_type)))
diff --git a/sql/sql_prepare.cc b/sql/sql_prepare.cc
index 37efd1d62b9..c97cb037f15 100644
--- a/sql/sql_prepare.cc
+++ b/sql/sql_prepare.cc
@@ -19,9 +19,9 @@ This file contains the implementation of prepare and executes.
 
 Prepare:
 
-  - Server gets the query from client with command 'COM_PREPARE';
+  - Server gets the query from client with command 'COM_STMT_PREPARE';
     in the following format:
-    [COM_PREPARE:1] [query]
+    [COM_STMT_PREPARE:1] [query]
   - Parse the query and recognize any parameter markers '?' and
     store its information list in lex->param_list
   - Allocate a new statement for this prepare; and keep this in
@@ -37,10 +37,10 @@ Prepare:
 
 Prepare-execute:
 
-  - Server gets the command 'COM_EXECUTE' to execute the
+  - Server gets the command 'COM_STMT_EXECUTE' to execute the
     previously prepared query. If there is any param markers; then client
     will send the data in the following format:
-    [COM_EXECUTE:1]
+    [COM_STMT_EXECUTE:1]
     [STMT_ID:4]
     [NULL_BITS:(param_count+7)/8)]
     [TYPES_SUPPLIED_BY_CLIENT(0/1):1]
@@ -55,9 +55,10 @@ Prepare-execute:
 
 Long data handling:
 
-  - Server gets the long data in pieces with command type 'COM_LONG_DATA'.
+  - Server gets the long data in pieces with command type
+    'COM_STMT_SEND_LONG_DATA'.
   - The packet recieved will have the format as:
-    [COM_LONG_DATA:1][STMT_ID:4][parameter_number:2][data]
+    [COM_STMT_SEND_LONG_DATA:1][STMT_ID:4][parameter_number:2][data]
   - data from the packet is appended to long data value buffer for this
     placeholder.
   - It's up to the client to check for read data ended. The server doesn't
@@ -104,7 +105,7 @@ public:
   Prepared_statement(THD *thd_arg);
   virtual ~Prepared_statement();
   void setup_set_params();
-  virtual Item_arena::Type type() const;
+  virtual Query_arena::Type type() const;
 };
 
 static void execute_stmt(THD *thd, Prepared_statement *stmt,
@@ -132,7 +133,7 @@ find_prepared_statement(THD *thd, ulong id, const char *where)
 {
   Statement *stmt= thd->stmt_map.find(id);
 
-  if (stmt == 0 || stmt->type() != Item_arena::PREPARED_STATEMENT)
+  if (stmt == 0 || stmt->type() != Query_arena::PREPARED_STATEMENT)
   {
     char llbuf[22];
     my_error(ER_UNKNOWN_STMT_HANDLER, MYF(0), sizeof(llbuf), llstr(id, llbuf),
@@ -839,7 +840,8 @@ static bool insert_params_from_vars_with_log(Prepared_statement *stmt,
   DBUG_ENTER("insert_params_from_vars");
 
   List_iterator<LEX_STRING> var_it(varnames);
-  String str;
+  String buf;
+  const String *val;
   uint32 length= 0;
   if (query->copy(stmt->query, stmt->query_length, default_charset_info))
     DBUG_RETURN(1);
@@ -850,32 +852,36 @@ static bool insert_params_from_vars_with_log(Prepared_statement *stmt,
     varname= var_it++;
     if (get_var_with_binlog(stmt->thd, *varname, &entry))
         DBUG_RETURN(1);
-    DBUG_ASSERT(entry != 0);
 
     if (param->set_from_user_var(stmt->thd, entry))
       DBUG_RETURN(1);
     /* Insert @'escaped-varname' instead of parameter in the query */
-    char *buf, *ptr;
-    str.length(0);
-    if (str.reserve(entry->name.length*2+3))
-      DBUG_RETURN(1);
+    if (entry)
+    {
+      char *begin, *ptr;
+      buf.length(0);
+      if (buf.reserve(entry->name.length*2+3))
+        DBUG_RETURN(1);
 
-    buf= str.c_ptr_quick();
-    ptr= buf;
-    *ptr++= '@';
-    *ptr++= '\'';
-    ptr+=
-      escape_string_for_mysql(&my_charset_utf8_general_ci,
-                              ptr, 0, entry->name.str, entry->name.length);
-    *ptr++= '\'';
-    str.length(ptr - buf);
+      begin= ptr= buf.c_ptr_quick();
+      *ptr++= '@';
+      *ptr++= '\'';
+      ptr+= escape_string_for_mysql(&my_charset_utf8_general_ci,
+                                    ptr, 0, entry->name.str,
+                                    entry->name.length);
+      *ptr++= '\'';
+      buf.length(ptr - begin);
+      val= &buf;
+    }
+    else
+      val= &my_null_string;
 
     if (param->convert_str_value(stmt->thd))
       DBUG_RETURN(1);                           /* out of memory */
 
-    if (query->replace(param->pos_in_query+length, 1, str))
+    if (query->replace(param->pos_in_query+length, 1, *val))
       DBUG_RETURN(1);
-    length+= str.length()-1;
+    length+= val->length()-1;
   }
   DBUG_RETURN(0);
 }
@@ -1700,12 +1706,20 @@ bool mysql_stmt_prepare(THD *thd, char *packet, uint packet_length,
                         LEX_STRING *name)
 {
   LEX *lex;
+  Statement stmt_backup;
   Prepared_statement *stmt= new Prepared_statement(thd);
   bool error;
   DBUG_ENTER("mysql_stmt_prepare");
 
   DBUG_PRINT("prep_query", ("%s", packet));
 
+  /*
+    If this is an SQLCOM_PREPARE, we also increase Com_prepare_sql.
+    However, it seems handy if com_stmt_prepare is increased always,
+    no matter what kind of prepare is processed.
+  */
+  statistic_increment(thd->status_var.com_stmt_prepare, &LOCK_status);
+
   if (stmt == 0)
     DBUG_RETURN(TRUE);
 
@@ -1726,19 +1740,19 @@ bool mysql_stmt_prepare(THD *thd, char *packet, uint packet_length,
     DBUG_RETURN(TRUE);
   }
 
-  thd->set_n_backup_statement(stmt, &thd->stmt_backup);
-  thd->set_n_backup_item_arena(stmt, &thd->stmt_backup);
+  thd->set_n_backup_statement(stmt, &stmt_backup);
+  thd->set_n_backup_item_arena(stmt, &stmt_backup);
 
   if (alloc_query(thd, packet, packet_length))
   {
-    thd->restore_backup_statement(stmt, &thd->stmt_backup);
-    thd->restore_backup_item_arena(stmt, &thd->stmt_backup);
+    thd->restore_backup_statement(stmt, &stmt_backup);
+    thd->restore_backup_item_arena(stmt, &stmt_backup);
     /* Statement map deletes statement on erase */
     thd->stmt_map.erase(stmt);
     DBUG_RETURN(TRUE);
   }
 
-  mysql_log.write(thd, COM_PREPARE, "[%lu] %s", stmt->id, packet);
+  mysql_log.write(thd, thd->command, "[%lu] %s", stmt->id, packet);
 
   thd->current_arena= stmt;
   mysql_init_query(thd, (uchar *) thd->query, thd->query_length);
@@ -1757,7 +1771,7 @@ bool mysql_stmt_prepare(THD *thd, char *packet, uint packet_length,
     transformation can be reused on execute, we set again thd->mem_root from
     stmt->mem_root (see setup_wild for one place where we do that).
   */
-  thd->restore_backup_item_arena(stmt, &thd->stmt_backup);
+  thd->restore_backup_item_arena(stmt, &stmt_backup);
 
   if (!error)
     error= check_prepared_statement(stmt, test(name));
@@ -1773,7 +1787,7 @@ bool mysql_stmt_prepare(THD *thd, char *packet, uint packet_length,
   lex_end(lex);
   close_thread_tables(thd);
   cleanup_stmt_and_thd_after_use(stmt, thd);
-  thd->restore_backup_statement(stmt, &thd->stmt_backup);
+  thd->restore_backup_statement(stmt, &stmt_backup);
   thd->current_arena= thd;
 
   if (error)
@@ -1786,7 +1800,7 @@ bool mysql_stmt_prepare(THD *thd, char *packet, uint packet_length,
   {
     stmt->setup_set_params();
     init_stmt_after_parse(thd, stmt->lex);
-    stmt->state= Item_arena::PREPARED;
+    stmt->state= Query_arena::PREPARED;
   }
   DBUG_RETURN(!stmt);
 }
@@ -1936,6 +1950,7 @@ void mysql_stmt_execute(THD *thd, char *packet, uint packet_length)
 {
   ulong stmt_id= uint4korr(packet);
   ulong flags= (ulong) ((uchar) packet[4]);
+  Statement stmt_backup;
   Cursor *cursor;
   /*
     Query text for binary log, or empty string if the query is not put into
@@ -1950,13 +1965,14 @@ void mysql_stmt_execute(THD *thd, char *packet, uint packet_length)
 
   packet+= 9;                               /* stmt_id + 5 bytes of flags */
 
+  statistic_increment(thd->status_var.com_stmt_execute, &LOCK_status);
   if (!(stmt= find_prepared_statement(thd, stmt_id, "mysql_stmt_execute")))
     DBUG_VOID_RETURN;
 
   DBUG_PRINT("exec_query:", ("%s", stmt->query));
 
   /* Check if we got an error when sending long data */
-  if (stmt->state == Item_arena::ERROR)
+  if (stmt->state == Query_arena::ERROR)
   {
     my_message(stmt->last_errno, stmt->last_error, MYF(0));
     DBUG_VOID_RETURN;
@@ -1988,7 +2004,7 @@ void mysql_stmt_execute(THD *thd, char *packet, uint packet_length)
     {
       DBUG_PRINT("info",("Using READ_ONLY cursor"));
       if (!cursor &&
-          !(cursor= stmt->cursor= new (&stmt->main_mem_root) Cursor()))
+          !(cursor= stmt->cursor= new (stmt->mem_root) Cursor(thd)))
         DBUG_VOID_RETURN;
       /* If lex->result is set, mysql_execute_command will use it */
       stmt->lex->result= &cursor->result;
@@ -2012,8 +2028,7 @@ void mysql_stmt_execute(THD *thd, char *packet, uint packet_length)
   if (stmt->param_count && stmt->set_params_data(stmt, &expanded_query))
     goto set_params_data_err;
 #endif
-  thd->stmt_backup.set_statement(thd);
-  thd->set_statement(stmt);
+  thd->set_n_backup_statement(stmt, &stmt_backup);
   thd->current_arena= stmt;
   reinit_stmt_before_use(thd, stmt->lex);
   /* From now cursors assume that thd->mem_root is clean */
@@ -2024,10 +2039,7 @@ void mysql_stmt_execute(THD *thd, char *packet, uint packet_length)
     my_error(ER_OUTOFMEMORY, 0, expanded_query.length());
     goto err;
   }
-
-  mysql_log.write(thd, COM_EXECUTE, "[%lu] %s", stmt->id,
-                  expanded_query.length() ? expanded_query.c_ptr() :
-                                            stmt->query);
+  mysql_log.write(thd, thd->command, "[%lu] %s", stmt->id, thd->query);
 
   thd->protocol= &thd->protocol_prep;           // Switch to binary protocol
   if (!(specialflag & SPECIAL_NO_PRIOR))
@@ -2053,7 +2065,7 @@ void mysql_stmt_execute(THD *thd, char *packet, uint packet_length)
     reset_stmt_params(stmt);
   }
 
-  thd->set_statement(&thd->stmt_backup);
+  thd->set_statement(&stmt_backup);
   thd->current_arena= thd;
   DBUG_VOID_RETURN;
 
@@ -2078,9 +2090,12 @@ void mysql_sql_stmt_execute(THD *thd, LEX_STRING *stmt_name)
     binary log.
   */
   String expanded_query;
+  Statement stmt_backup;
   DBUG_ENTER("mysql_sql_stmt_execute");
 
   DBUG_ASSERT(thd->free_list == NULL);
+  /* See comment for statistic_increment in mysql_stmt_prepare */
+  statistic_increment(thd->status_var.com_stmt_execute, &LOCK_status);
 
   if (!(stmt= (Prepared_statement*)thd->stmt_map.find_by_name(stmt_name)))
   {
@@ -2097,15 +2112,16 @@ void mysql_sql_stmt_execute(THD *thd, LEX_STRING *stmt_name)
 
   /* Must go before setting variables, as it clears thd->user_var_events */
   mysql_reset_thd_for_next_command(thd);
-  thd->set_n_backup_statement(stmt, &thd->stmt_backup);
-  thd->set_statement(stmt);
+  thd->set_n_backup_statement(stmt, &stmt_backup);
   if (stmt->set_params_from_vars(stmt,
-                                 thd->stmt_backup.lex->prepared_stmt_params,
+                                 stmt_backup.lex->prepared_stmt_params,
                                  &expanded_query))
   {
     my_error(ER_WRONG_ARGUMENTS, MYF(0), "EXECUTE");
   }
+  thd->command= COM_STMT_EXECUTE; /* For nice messages in general log */
   execute_stmt(thd, stmt, &expanded_query);
+  thd->set_statement(&stmt_backup);
   DBUG_VOID_RETURN;
 }
 
@@ -2137,6 +2153,7 @@ static void execute_stmt(THD *thd, Prepared_statement *stmt,
     my_error(ER_OUTOFMEMORY, MYF(0), expanded_query->length());
     DBUG_VOID_RETURN;
   }
+  mysql_log.write(thd, thd->command, "[%lu] %s", stmt->id, thd->query);
   /*
     At first execution of prepared statement we will perform logical
     transformations of the query tree (i.e. negations elimination).
@@ -2149,21 +2166,28 @@ static void execute_stmt(THD *thd, Prepared_statement *stmt,
   mysql_execute_command(thd);
   if (!(specialflag & SPECIAL_NO_PRIOR))
     my_pthread_setprio(pthread_self(), WAIT_PRIOR);
+  /*
+    'start_time' is set in dispatch_command, but THD::query will
+    be freed when we return from this function. So let's log the slow
+    query here.
+  */
+  log_slow_statement(thd);
+  /* Prevent from second logging in the end of dispatch_command */
+  thd->enable_slow_log= FALSE;
 
   close_thread_tables(thd);                    // to close derived tables
   cleanup_stmt_and_thd_after_use(stmt, thd);
   reset_stmt_params(stmt);
-  thd->set_statement(&thd->stmt_backup);
   thd->current_arena= thd;
 
-  if (stmt->state == Item_arena::PREPARED)
-    stmt->state= Item_arena::EXECUTED;
+  if (stmt->state == Query_arena::PREPARED)
+    stmt->state= Query_arena::EXECUTED;
   DBUG_VOID_RETURN;
 }
 
 
 /*
-  COM_FETCH handler: fetches requested amount of rows from cursor
+  COM_STMT_FETCH handler: fetches requested amount of rows from cursor
 
   SYNOPSIS
     mysql_stmt_fetch()
@@ -2178,19 +2202,21 @@ void mysql_stmt_fetch(THD *thd, char *packet, uint packet_length)
   ulong stmt_id= uint4korr(packet);
   ulong num_rows= uint4korr(packet+4);
   Prepared_statement *stmt;
+  Statement stmt_backup;
   DBUG_ENTER("mysql_stmt_fetch");
 
+  statistic_increment(thd->status_var.com_stmt_fetch, &LOCK_status);
   if (!(stmt= find_prepared_statement(thd, stmt_id, "mysql_stmt_fetch")))
     DBUG_VOID_RETURN;
 
   if (!stmt->cursor || !stmt->cursor->is_open())
   {
-    my_error(ER_STMT_HAS_NO_OPEN_CURSOR, MYF(0));
+    my_error(ER_STMT_HAS_NO_OPEN_CURSOR, MYF(0), stmt_id);
     DBUG_VOID_RETURN;
   }
 
   thd->current_arena= stmt;
-  thd->set_n_backup_statement(stmt, &thd->stmt_backup);
+  thd->set_n_backup_statement(stmt, &stmt_backup);
 
   if (!(specialflag & SPECIAL_NO_PRIOR))
     my_pthread_setprio(pthread_self(), QUERY_PRIOR);
@@ -2202,7 +2228,7 @@ void mysql_stmt_fetch(THD *thd, char *packet, uint packet_length)
   if (!(specialflag & SPECIAL_NO_PRIOR))
     my_pthread_setprio(pthread_self(), WAIT_PRIOR);
 
-  thd->restore_backup_statement(stmt, &thd->stmt_backup);
+  thd->restore_backup_statement(stmt, &stmt_backup);
   thd->current_arena= thd;
 
   if (!stmt->cursor->is_open())
@@ -2240,13 +2266,14 @@ void mysql_stmt_reset(THD *thd, char *packet)
   Prepared_statement *stmt;
   DBUG_ENTER("mysql_stmt_reset");
 
+  statistic_increment(thd->status_var.com_stmt_reset, &LOCK_status);
   if (!(stmt= find_prepared_statement(thd, stmt_id, "mysql_stmt_reset")))
     DBUG_VOID_RETURN;
 
   if (stmt->cursor && stmt->cursor->is_open())
     stmt->cursor->close();
 
-  stmt->state= Item_arena::PREPARED;
+  stmt->state= Query_arena::PREPARED;
 
   /*
     Clear parameters from data which could be set by
@@ -2266,14 +2293,15 @@ void mysql_stmt_reset(THD *thd, char *packet)
   Note: we don't send any reply to that command.
 */
 
-void mysql_stmt_free(THD *thd, char *packet)
+void mysql_stmt_close(THD *thd, char *packet)
 {
   /* There is always space for 4 bytes in packet buffer */
   ulong stmt_id= uint4korr(packet);
   Prepared_statement *stmt;
 
-  DBUG_ENTER("mysql_stmt_free");
+  DBUG_ENTER("mysql_stmt_close");
 
+  statistic_increment(thd->status_var.com_stmt_close, &LOCK_status);
   if (!(stmt= find_prepared_statement(thd, stmt_id, "mysql_stmt_close")))
     DBUG_VOID_RETURN;
 
@@ -2312,6 +2340,7 @@ void mysql_stmt_get_longdata(THD *thd, char *packet, ulong packet_length)
 
   DBUG_ENTER("mysql_stmt_get_longdata");
 
+  statistic_increment(thd->status_var.com_stmt_send_long_data, &LOCK_status);
 #ifndef EMBEDDED_LIBRARY
   /* Minimal size of long data packet is 6 bytes */
   if ((ulong) (packet_end - packet) < MYSQL_LONG_DATA_HEADER)
@@ -2334,7 +2363,7 @@ void mysql_stmt_get_longdata(THD *thd, char *packet, ulong packet_length)
   if (param_number >= stmt->param_count)
   {
     /* Error will be sent in execute call */
-    stmt->state= Item_arena::ERROR;
+    stmt->state= Query_arena::ERROR;
     stmt->last_errno= ER_WRONG_ARGUMENTS;
     sprintf(stmt->last_error, ER(ER_WRONG_ARGUMENTS),
             "mysql_stmt_send_long_data");
@@ -2350,7 +2379,7 @@ void mysql_stmt_get_longdata(THD *thd, char *packet, ulong packet_length)
   if (param->set_longdata(thd->extra_data, thd->extra_length))
 #endif
   {
-    stmt->state= Item_arena::ERROR;
+    stmt->state= Query_arena::ERROR;
     stmt->last_errno= ER_OUTOFMEMORY;
     sprintf(stmt->last_error, ER(ER_OUTOFMEMORY), 0);
   }
@@ -2359,7 +2388,9 @@ void mysql_stmt_get_longdata(THD *thd, char *packet, ulong packet_length)
 
 
 Prepared_statement::Prepared_statement(THD *thd_arg)
-  :Statement(thd_arg),
+  :Statement(INITIALIZED, ++thd_arg->statement_id_counter,
+             thd_arg->variables.query_alloc_block_size,
+             thd_arg->variables.query_prealloc_size),
   thd(thd_arg),
   param_array(0),
   param_count(0),
@@ -2368,10 +2399,12 @@ Prepared_statement::Prepared_statement(THD *thd_arg)
   *last_error= '\0';
 }
 
+
 void Prepared_statement::setup_set_params()
 {
   /* Setup binary logging */
-  if (mysql_bin_log.is_open() && is_update_query(lex->sql_command))
+  if (mysql_bin_log.is_open() && is_update_query(lex->sql_command) ||
+      mysql_log.is_open() || mysql_slow_log.is_open())
   {
     set_params_from_vars= insert_params_from_vars_with_log;
 #ifndef EMBEDDED_LIBRARY
@@ -2396,12 +2429,12 @@ Prepared_statement::~Prepared_statement()
 {
   if (cursor)
     cursor->Cursor::~Cursor();
-  free_items(free_list);
+  free_items();
   delete lex->result;
 }
 
 
-Item_arena::Type Prepared_statement::type() const
+Query_arena::Type Prepared_statement::type() const
 {
   return PREPARED_STATEMENT;
 }
diff --git a/sql/sql_rename.cc b/sql/sql_rename.cc
index 8fe17198cf0..3880aa428b9 100644
--- a/sql/sql_rename.cc
+++ b/sql/sql_rename.cc
@@ -164,7 +164,7 @@ rename_tables(THD *thd, TABLE_LIST *table_list, bool skip_error)
 	    ren_table->db, old_alias,
 	    reg_ext);
     unpack_filename(name, name);
-    if ((table_type=get_table_type(name)) == DB_TYPE_UNKNOWN)
+    if ((table_type=get_table_type(thd, name)) == DB_TYPE_UNKNOWN)
     {
       my_error(ER_FILE_NOT_FOUND, MYF(0), name, my_errno);
       if (!skip_error)
diff --git a/sql/sql_select.cc b/sql/sql_select.cc
index 099b7857af8..cf8d7b1b83c 100644
--- a/sql/sql_select.cc
+++ b/sql/sql_select.cc
@@ -28,6 +28,8 @@
 #include <hash.h>
 #include <ft_global.h>
 
+typedef uint32 cache_rec_length_type;
+
 const char *join_type_str[]={ "UNKNOWN","system","const","eq_ref","ref",
 			      "MAYBE_REF","ALL","range","index","fulltext",
 			      "ref_or_null","unique_subquery","index_subquery",
@@ -85,10 +87,9 @@ static void update_depend_map(JOIN *join, ORDER *order);
 static ORDER *remove_const(JOIN *join,ORDER *first_order,COND *cond,
 			   bool change_list, bool *simple_order);
 static int return_zero_rows(JOIN *join, select_result *res,TABLE_LIST *tables,
-			    List<Item> &fields, bool send_row,
-			    uint select_options, const char *info,
-			    Item *having, Procedure *proc,
-			    SELECT_LEX_UNIT *unit);
+                            List<Item> &fields, bool send_row,
+                            uint select_options, const char *info,
+                            Item *having);
 static COND *build_equal_items(THD *thd, COND *cond,
                                COND_EQUAL *inherited,
                                List<TABLE_LIST> *join_list,
@@ -136,7 +137,7 @@ end_unique_update(JOIN *join, JOIN_TAB *join_tab, bool end_of_records);
 static enum_nested_loop_state
 end_write_group(JOIN *join, JOIN_TAB *join_tab, bool end_of_records);
 
-static int test_if_group_changed(List<Item_buff> &list);
+static int test_if_group_changed(List<Cached_item> &list);
 static int join_read_const_table(JOIN_TAB *tab, POSITION *pos);
 static int join_read_system(JOIN_TAB *tab);
 static int join_read_const(JOIN_TAB *tab);
@@ -580,7 +581,7 @@ JOIN::optimize()
       MEMROOT for prepared statements and stored procedures.
     */
 
-    Item_arena *arena= thd->current_arena, backup;
+    Query_arena *arena= thd->current_arena, backup;
     if (arena->is_conventional())
       arena= 0;                                   // For easier test
     else
@@ -812,9 +813,14 @@ JOIN::optimize()
       DBUG_RETURN(1);
   }
   simple_group= 0;
-  group_list= remove_const(this, group_list, conds,
-                           rollup.state == ROLLUP::STATE_NONE,
-                           &simple_group);
+  {
+    ORDER *old_group_list;
+    group_list= remove_const(this, (old_group_list= group_list), conds,
+                             rollup.state == ROLLUP::STATE_NONE,
+			     &simple_group);
+    if (old_group_list && !group_list)
+      select_distinct= 0;
+  }
   if (!group_list && group)
   {
     order=0;					// The output has only one row
@@ -1220,8 +1226,7 @@ JOIN::exec()
 			    send_row_on_empty_set(),
 			    select_options,
 			    zero_result_cause,
-			    having, procedure,
-			    unit);
+			    having);
     DBUG_VOID_RETURN;
   }
 
@@ -1430,7 +1435,7 @@ JOIN::exec()
 	DBUG_VOID_RETURN;
       }
       end_read_record(&curr_join->join_tab->read_record);
-      curr_join->const_tables= curr_join->tables; // Mark free for join_free()
+      curr_join->const_tables= curr_join->tables; // Mark free for cleanup()
       curr_join->join_tab[0].table= 0;           // Table is freed
       
       // No sum funcs anymore
@@ -1660,9 +1665,9 @@ JOIN::exec()
 */
 
 int
-JOIN::cleanup()
+JOIN::destroy()
 {
-  DBUG_ENTER("JOIN::cleanup");
+  DBUG_ENTER("JOIN::destroy");
   select_lex->join= 0;
 
   if (tmp_join)
@@ -1677,12 +1682,11 @@ JOIN::cleanup()
     }
     tmp_join->tmp_join= 0;
     tmp_table_param.copy_field=0;
-    DBUG_RETURN(tmp_join->cleanup());
+    DBUG_RETURN(tmp_join->destroy());
   }
   cond_equal= 0;
 
-  lock=0;                                     // It's faster to unlock later
-  join_free(1);
+  cleanup(1);
   if (exec_tmp_table1)
     free_tmp_table(thd, exec_tmp_table1);
   if (exec_tmp_table2)
@@ -1690,18 +1694,21 @@ JOIN::cleanup()
   delete select;
   delete_dynamic(&keyuse);
   delete procedure;
-  for (SELECT_LEX_UNIT *lex_unit= select_lex->first_inner_unit();
-       lex_unit != 0;
-       lex_unit= lex_unit->next_unit())
-  {
-    error|= lex_unit->cleanup();
-  }
   DBUG_RETURN(error);
 }
 
 
 /************************* Cursor ******************************************/
-  
+
+Cursor::Cursor(THD *thd)
+  :Query_arena(&main_mem_root, INITIALIZED),
+   join(0), unit(0)
+{
+  /* We will overwrite it at open anyway. */
+  init_sql_alloc(&main_mem_root, ALLOC_ROOT_MIN_BLOCK_SIZE, 0);
+}
+
+
 void
 Cursor::init_from_thd(THD *thd)
 {
@@ -1808,6 +1815,7 @@ Cursor::fetch(ulong num_rows)
   THD *thd= join->thd;
   JOIN_TAB *join_tab= join->join_tab + join->const_tables;
   enum_nested_loop_state error= NESTED_LOOP_OK;
+  Query_arena backup_arena;
   DBUG_ENTER("Cursor::fetch");
   DBUG_PRINT("enter",("rows: %lu", num_rows));
 
@@ -1819,7 +1827,7 @@ Cursor::fetch(ulong num_rows)
   thd->lock= lock;
   thd->query_id= query_id;
   /* save references to memory, allocated during fetch */
-  thd->set_n_backup_item_arena(this, &thd->stmt_backup);
+  thd->set_n_backup_item_arena(this, &backup_arena);
 
   join->fetch_limit+= num_rows;
 
@@ -1835,7 +1843,7 @@ Cursor::fetch(ulong num_rows)
     ha_release_temporary_latches(thd);
 #endif
 
-  thd->restore_backup_item_arena(this, &thd->stmt_backup);
+  thd->restore_backup_item_arena(this, &backup_arena);
   DBUG_ASSERT(thd->free_list == 0);
   reset_thd(thd);
 
@@ -1868,17 +1876,14 @@ Cursor::close()
   THD *thd= join->thd;
   DBUG_ENTER("Cursor::close");
 
-  join->join_free(0);
+  /*
+    In case of UNIONs JOIN is freed inside of unit->cleanup(),
+    otherwise in select_lex->cleanup().
+  */
   if (unit)
-  {
-    /* In case of UNIONs JOIN is freed inside unit->cleanup() */
-    unit->cleanup();
-  }
+    (void) unit->cleanup();
   else
-  {
-    join->cleanup();
-    delete join;
-  }
+    (void) join->select_lex->cleanup();
   {
     /* XXX: Another hack: closing tables used in the cursor */
     DBUG_ASSERT(lock || open_tables || derived_tables);
@@ -1897,8 +1902,7 @@ Cursor::close()
   }
   join= 0;
   unit= 0;
-  free_items(free_list);
-  free_list= 0;
+  free_items();
   /*
     Must be last, as some memory might be allocated for free purposes,
     like in free_tmp_table() (TODO: fix this issue)
@@ -2055,8 +2059,7 @@ err:
   if (free_join)
   {
     thd->proc_info="end";
-    err= join->cleanup();
-    delete join;
+    err= select_lex->cleanup();
     DBUG_RETURN(err || thd->net.report_error);
   }
   DBUG_RETURN(join->error);
@@ -2158,7 +2161,7 @@ make_join_statistics(JOIN *join, TABLE_LIST *tables, COND *conds,
     if (*s->on_expr_ref)
     {
       /* s is the only inner table of an outer join */
-      if (!table->file->records)
+      if (!table->file->records && !embedding)
       {						// Empty table
         s->dependent= 0;                        // Ignore LEFT JOIN depend.
 	set_position(join,const_count++,s,(KEYUSE*) 0);
@@ -5889,29 +5892,68 @@ void JOIN_TAB::cleanup()
 }
 
 
+void JOIN::join_free(bool full)
+{
+  SELECT_LEX_UNIT *unit;
+  SELECT_LEX *sl;
+  DBUG_ENTER("JOIN::join_free");
+
+  /*
+    Optimization: if not EXPLAIN and we are done with the JOIN,
+    free all tables.
+  */
+  full= full || (!select_lex->uncacheable && !thd->lex->describe);
+
+  cleanup(full);
+
+  for (unit= select_lex->first_inner_unit(); unit; unit= unit->next_unit())
+    for (sl= unit->first_select_in_union(); sl; sl= sl->next_select())
+    {
+      JOIN *join= sl->join;
+      if (join)
+        join->join_free(full);
+    }
+
+  /*
+    We are not using tables anymore
+    Unlock all tables. We may be in an INSERT .... SELECT statement.
+  */
+  if (full && lock && thd->lock && !(select_options & SELECT_NO_UNLOCK) &&
+      !select_lex->subquery_in_having &&
+      (select_lex == (thd->lex->unit.fake_select_lex ?
+                      thd->lex->unit.fake_select_lex : &thd->lex->select_lex)))
+  {
+    /*
+      TODO: unlock tables even if the join isn't top level select in the
+      tree.
+    */
+    mysql_unlock_read_tables(thd, lock);           // Don't free join->lock
+    lock= 0;
+  }
+
+  DBUG_VOID_RETURN;
+}
+
+
 /*
   Free resources of given join
 
   SYNOPSIS
-    JOIN::join_free()
+    JOIN::cleanup()
     fill - true if we should free all resources, call with full==1 should be
            last, before it this function can be called with full==0
 
   NOTE: with subquery this function definitely will be called several times,
     but even for simple query it can be called several times.
 */
-void
-JOIN::join_free(bool full)
-{
-  JOIN_TAB *tab,*end;
-  DBUG_ENTER("JOIN::join_free");
 
-  full= full || (!select_lex->uncacheable &&
-                 !thd->lex->subqueries &&
-                 !thd->lex->describe); // do not cleanup too early on EXPLAIN
+void JOIN::cleanup(bool full)
+{
+  DBUG_ENTER("JOIN::cleanup");
 
   if (table)
   {
+    JOIN_TAB *tab,*end;
     /*
       Only a sorted table may be cached.  This sorted table is always the
       first non const table in join->table
@@ -5922,16 +5964,6 @@ JOIN::join_free(bool full)
       filesort_free_buffers(table[const_tables]);
     }
 
-    for (SELECT_LEX_UNIT *unit= select_lex->first_inner_unit(); unit;
-         unit= unit->next_unit())
-    {
-      JOIN *join;
-      for (SELECT_LEX *sl= unit->first_select_in_union(); sl;
-           sl= sl->next_select())
-        if ((join= sl->join))
-          join->join_free(full);
-    }
-
     if (full)
     {
       for (tab= join_tab, end= tab+tables; tab != end; tab++)
@@ -5948,25 +5980,14 @@ JOIN::join_free(bool full)
       }
     }
   }
-
   /*
     We are not using tables anymore
     Unlock all tables. We may be in an INSERT .... SELECT statement.
   */
-  if (full && lock && thd->lock && !(select_options & SELECT_NO_UNLOCK) &&
-      !select_lex->subquery_in_having)
-  {
-    // TODO: unlock tables even if the join isn't top level select in the tree
-    if (select_lex == (thd->lex->unit.fake_select_lex ?
-                       thd->lex->unit.fake_select_lex : &thd->lex->select_lex))
-    {
-      mysql_unlock_read_tables(thd, lock);        // Don't free join->lock
-      lock=0;
-    }
-  }
-
   if (full)
   {
+    if (tmp_join)
+      tmp_table_param.copy_field= 0;
     group_fields.delete_elements();
     /*
       We can't call delete_elements() on copy_funcs as this will cause
@@ -6201,8 +6222,7 @@ remove_const(JOIN *join,ORDER *first_order, COND *cond,
 static int
 return_zero_rows(JOIN *join, select_result *result,TABLE_LIST *tables,
 		 List<Item> &fields, bool send_row, uint select_options,
-		 const char *info, Item *having, Procedure *procedure,
-		 SELECT_LEX_UNIT *unit)
+		 const char *info, Item *having)
 {
   DBUG_ENTER("return_zero_rows");
 
@@ -6270,7 +6290,7 @@ public:
   COND_CMP(Item *a,Item_func *b) :and_level(a),cmp_func(b) {}
 };
 
-#ifdef __GNUC__
+#ifdef HAVE_EXPLICIT_TEMPLATE_INSTANTIATION
 template class I_List<COND_CMP>;
 template class I_List_iterator<COND_CMP>;
 template class List<Item_func_match>;
@@ -7064,7 +7084,7 @@ static COND* substitute_for_best_equal_field(COND *cond,
       List_iterator_fast<Item_equal> it(cond_equal->current_level);
       while ((item_equal= it++))
       {
-        eliminate_item_equal(cond, cond_equal->upper_levels, item_equal);
+        cond= eliminate_item_equal(cond, cond_equal->upper_levels, item_equal);
       }
     }
   }
@@ -7937,7 +7957,9 @@ Field *create_tmp_field_for_schema(THD *thd, Item *item, TABLE *table)
 
 Field *create_tmp_field(THD *thd, TABLE *table,Item *item, Item::Type type,
                         Item ***copy_func, Field **from_field,
-                        bool group, bool modify_item, uint convert_blob_length)
+                        bool group, bool modify_item,
+                        bool table_cant_handle_bit_fields,
+                        uint convert_blob_length)
 {
   switch (type) {
   case Item::SUM_FUNC_ITEM:
@@ -7952,11 +7974,27 @@ Field *create_tmp_field(THD *thd, TABLE *table,Item *item, Item::Type type,
   case Item::DEFAULT_VALUE_ITEM:
   {
     Item_field *field= (Item_field*) item;
+    if (table_cant_handle_bit_fields && field->field->type() == FIELD_TYPE_BIT)
+      return create_tmp_field_from_item(thd, item, table, copy_func,
+                                        modify_item, convert_blob_length);
     return create_tmp_field_from_field(thd, (*from_field= field->field),
                                        item->name, table,
                                        modify_item ? (Item_field*) item : NULL,
                                        convert_blob_length);
   }
+  case Item::REF_ITEM:
+    if ( item->real_item()->type() == Item::FIELD_ITEM)
+    {
+      Item_field *field= (Item_field*) *((Item_ref*)item)->ref;
+      Field *new_field= create_tmp_field_from_field(thd, 
+                               (*from_field= field->field),
+                               item->name, table,
+                               NULL,
+                               convert_blob_length);
+      if (modify_item)
+        item->set_result_field(new_field);
+      return new_field;
+    }
   case Item::FUNC_ITEM:
   case Item::COND_ITEM:
   case Item::FIELD_AVG_ITEM:
@@ -7968,7 +8006,6 @@ Field *create_tmp_field(THD *thd, TABLE *table,Item *item, Item::Type type,
   case Item::REAL_ITEM:
   case Item::DECIMAL_ITEM:
   case Item::STRING_ITEM:
-  case Item::REF_ITEM:
   case Item::NULL_ITEM:
   case Item::VARBIN_ITEM:
     return create_tmp_field_from_item(thd, item, table, copy_func, modify_item,
@@ -8160,6 +8197,7 @@ create_tmp_table(THD *thd,TMP_TABLE_PARAM *param,List<Item> &fields,
 	  Field *new_field=
             create_tmp_field(thd, table, arg, arg->type(), &copy_func,
                              tmp_from_field, group != 0,not_all_columns,
+                             group || distinct,
                              param->convert_blob_length);
 	  if (!new_field)
 	    goto err;					// Should be OOM
@@ -8207,7 +8245,7 @@ create_tmp_table(THD *thd,TMP_TABLE_PARAM *param,List<Item> &fields,
         create_tmp_field_for_schema(thd, item, table) :
         create_tmp_field(thd, table, item, type, &copy_func,
                          tmp_from_field, group != 0,
-                         not_all_columns || group !=0,
+                         not_all_columns || group != 0, 0,
                          param->convert_blob_length);
 
       if (!new_field)
@@ -8345,6 +8383,13 @@ create_tmp_table(THD *thd,TMP_TABLE_PARAM *param,List<Item> &fields,
     }
     else
       field->move_field((char*) pos,(uchar*) 0,0);
+    if (field->type() == FIELD_TYPE_BIT)
+    {
+      /* We have to reserve place for extra bits among null bits */
+      ((Field_bit*) field)->set_bit_ptr(null_flags + null_count / 8,
+                                        null_count & 7);
+      null_count+= (field->field_length & 7);
+    }
     field->reset();
     if (from_field[i])
     {						/* Not a table Item */
@@ -11587,7 +11632,7 @@ used_blob_length(CACHE_FIELD **ptr)
 static bool
 store_record_in_cache(JOIN_CACHE *cache)
 {
-  ulong length;
+  cache_rec_length_type length;
   uchar *pos;
   CACHE_FIELD *copy,*end_field;
   bool last_record;
@@ -11632,9 +11677,9 @@ store_record_in_cache(JOIN_CACHE *cache)
 	     end > str && end[-1] == ' ' ;
 	     end--) ;
 	length=(uint) (end-str);
-	memcpy(pos+1,str,length);
-	*pos=(uchar) length;
-	pos+=length+1;
+	memcpy(pos+sizeof(length), str, length);
+	memcpy_fixed(pos, &length, sizeof(length));
+	pos+= length+sizeof(length);
       }
       else
       {
@@ -11668,7 +11713,7 @@ static void
 read_cached_record(JOIN_TAB *tab)
 {
   uchar *pos;
-  uint length;
+  cache_rec_length_type length;
   bool last_record;
   CACHE_FIELD *copy,*end_field;
 
@@ -11697,9 +11742,10 @@ read_cached_record(JOIN_TAB *tab)
     {
       if (copy->strip)
       {
-	memcpy(copy->str,pos+1,length=(uint) *pos);
-	memset(copy->str+length,' ',copy->length-length);
-	pos+=1+length;
+        memcpy_fixed(&length, pos, sizeof(length));
+	memcpy(copy->str, pos+sizeof(length), length);
+	memset(copy->str+length, ' ', copy->length-length);
+	pos+= sizeof(length)+length;
       }
       else
       {
@@ -11736,10 +11782,12 @@ cp_buffer_from_ref(THD *thd, TABLE_REF *ref)
   thd->count_cuted_fields= CHECK_FIELD_IGNORE;
   for (store_key **copy=ref->key_copy ; *copy ; copy++)
   {
-    if ((*copy)->copy())
+    int res;
+    if ((res= (*copy)->copy()))
     {
       thd->count_cuted_fields= save_count_cuted_fields;
-      return 1;					// Something went wrong
+      if ((res= res & 1))
+        return res;                               // Something went wrong
     }
   }
   thd->count_cuted_fields= save_count_cuted_fields;
@@ -11778,11 +11826,11 @@ cp_buffer_from_ref(THD *thd, TABLE_REF *ref)
     ref_pointer_array.
 
   RETURN
-    0 if OK
-    1 if error occurred
+    FALSE if OK
+    TRUE  if error occurred
 */
 
-static int
+static bool
 find_order_in_list(THD *thd, Item **ref_pointer_array, TABLE_LIST *tables,
                    ORDER *order, List<Item> &fields, List<Item> &all_fields,
                    bool is_group_field)
@@ -11799,13 +11847,13 @@ find_order_in_list(THD *thd, Item **ref_pointer_array, TABLE_LIST *tables,
     {
       my_error(ER_BAD_FIELD_ERROR, MYF(0),
                order_item->full_name(), thd->where);
-      return 1;
+      return TRUE;
     }
     order->item= ref_pointer_array + count - 1;
     order->in_field_list= 1;
     order->counter= count;
     order->counter_used= 1;
-    return 0;
+    return FALSE;
   }
   /* Lookup the current GROUP/ORDER field in the SELECT clause. */
   uint counter;
@@ -11813,7 +11861,7 @@ find_order_in_list(THD *thd, Item **ref_pointer_array, TABLE_LIST *tables,
   select_item= find_item_in_list(order_item, fields, &counter,
                                  REPORT_EXCEPT_NOT_FOUND, &unaliased);
   if (!select_item)
-    return 1; /* Some error occured. */
+    return TRUE; /* The item is not unique, or some other error occured. */
 
 
   /* Check whether the resolved field is not ambiguos. */
@@ -11827,7 +11875,7 @@ find_order_in_list(THD *thd, Item **ref_pointer_array, TABLE_LIST *tables,
     */
     if (unaliased && !order_item->fixed && order_item->fix_fields(thd, tables,
                                                                   order->item))
-      return 1;
+      return TRUE;
 
     /* Lookup the current GROUP field in the FROM clause. */
     order_item_type= order_item->type();
@@ -11867,27 +11915,42 @@ find_order_in_list(THD *thd, Item **ref_pointer_array, TABLE_LIST *tables,
     {
       order->item= ref_pointer_array + counter;
       order->in_field_list=1;
-      return 0;
+      return FALSE;
     }
+    else
+      /*
+        There is a field with the same name in the FROM clause. This is the field
+        that will be chosen. In this case we issue a warning so the user knows
+        that the field from the FROM clause overshadows the column reference from
+        the SELECT list.
+      */
+      push_warning_printf(thd, MYSQL_ERROR::WARN_LEVEL_WARN, ER_NON_UNIQ_ERROR,
+                          ER(ER_NON_UNIQ_ERROR), from_field->field_name,
+                          current_thd->where);
   }
 
   order->in_field_list=0;
   /*
+    The call to order_item->fix_fields() means that here we resolve 'order_item'
+    to a column from a table in the list 'tables', or to a column in some outer
+    query. Exactly because of the second case we come to this point even if
+    (select_item == not_found_item), inspite of that fix_fields() calls
+    find_item_in_list() one more time.
+
     We check order_item->fixed because Item_func_group_concat can put
     arguments for which fix_fields already was called.
-
-    'it' reassigned in if condition because fix_field can change it.
   */
   if (!order_item->fixed &&
       (order_item->fix_fields(thd, tables, order->item) ||
        (order_item= *order->item)->check_cols(1) ||
        thd->is_fatal_error))
-    return 1;					// Wrong field 
+    return TRUE; /* Wrong field. */
+
   uint el= all_fields.elements;
-  all_fields.push_front(order_item);		        // Add new field to field list
+  all_fields.push_front(order_item); /* Add new field to field list. */
   ref_pointer_array[el]= order_item;
   order->item= ref_pointer_array + el;
-  return 0;
+  return FALSE;
 }
 
 
@@ -12274,7 +12337,7 @@ alloc_group_fields(JOIN *join,ORDER *group)
   {
     for (; group ; group=group->next)
     {
-      Item_buff *tmp=new_Item_buff(join->thd, *group->item);
+      Cached_item *tmp=new_Cached_item(join->thd, *group->item);
       if (!tmp || join->group_fields.push_front(tmp))
 	return TRUE;
     }
@@ -12285,12 +12348,12 @@ alloc_group_fields(JOIN *join,ORDER *group)
 
 
 static int
-test_if_group_changed(List<Item_buff> &list)
+test_if_group_changed(List<Cached_item> &list)
 {
   DBUG_ENTER("test_if_group_changed");
-  List_iterator<Item_buff> li(list);
+  List_iterator<Cached_item> li(list);
   int idx= -1,i;
-  Item_buff *buff;
+  Cached_item *buff;
 
   for (i=(int) list.elements-1 ; (buff=li++) ; i--)
   {
diff --git a/sql/sql_select.h b/sql/sql_select.h
index 5351bbb13d3..6d33ae2f978 100644
--- a/sql/sql_select.h
+++ b/sql/sql_select.h
@@ -178,7 +178,7 @@ class JOIN :public Sql_alloc
   table_map const_table_map,found_const_table_map,outer_join;
   ha_rows  send_records,found_records,examined_rows,row_limit, select_limit;
   /*
-    Used to fetch no more than given amount of rows per one 
+    Used to fetch no more than given amount of rows per one
     fetch operation of server side cursor.
     The value is checked in end_send and end_send_group in fashion, similar
     to offset_limit_cnt:
@@ -190,7 +190,7 @@ class JOIN :public Sql_alloc
   POSITION positions[MAX_TABLES+1],best_positions[MAX_TABLES+1];
   double   best_read;
   List<Item> *fields;
-  List<Item_buff> group_fields, group_fields_cache;
+  List<Cached_item> group_fields, group_fields_cache;
   TABLE    *tmp_table;
   // used to store 2 possible tmp table of SELECT
   TABLE    *exec_tmp_table1, *exec_tmp_table2;
@@ -325,7 +325,7 @@ class JOIN :public Sql_alloc
   int optimize();
   int reinit();
   void exec();
-  int cleanup();
+  int destroy();
   void restore_tmp();
   bool alloc_func_list();
   bool make_sum_func_list(List<Item> &all_fields, List<Item> &send_fields,
@@ -349,7 +349,15 @@ class JOIN :public Sql_alloc
   int rollup_send_data(uint idx);
   int rollup_write_data(uint idx, TABLE *table);
   bool test_in_subselect(Item **where);
+  /*
+    Release memory and, if possible, the open tables held by this execution
+    plan (and nested plans). It's used to release some tables before
+    the end of execution in order to increase concurrency and reduce
+    memory consumption.
+  */
   void join_free(bool full);
+  /* Cleanup this JOIN, possibly for reuse */
+  void cleanup(bool full);
   void clear();
   bool save_join_tab();
   bool send_row_on_empty_set()
@@ -370,8 +378,9 @@ class JOIN :public Sql_alloc
   statement for many cursors.
 */
 
-class Cursor: public Sql_alloc, public Item_arena
+class Cursor: public Sql_alloc, public Query_arena
 {
+  MEM_ROOT main_mem_root;
   JOIN *join;
   SELECT_LEX_UNIT *unit;
 
@@ -396,7 +405,7 @@ public:
   void close();
 
   void set_unit(SELECT_LEX_UNIT *unit_arg) { unit= unit_arg; }
-  Cursor() :Item_arena(TRUE), join(0), unit(0) {}
+  Cursor(THD *thd);
   ~Cursor();
 };
 
@@ -447,6 +456,7 @@ class store_key :public Sql_alloc
   char *null_ptr;
   char err;
  public:
+  enum store_key_result { STORE_KEY_OK, STORE_KEY_FATAL, STORE_KEY_CONV };
   store_key(THD *thd, Field *field_arg, char *ptr, char *null, uint length)
     :null_ptr(null),err(0)
   {
@@ -462,7 +472,7 @@ class store_key :public Sql_alloc
                                         ptr, (uchar*) null, 1);
   }
   virtual ~store_key() {}			/* Not actually needed */
-  virtual bool copy()=0;
+  virtual enum store_key_result copy()=0;
   virtual const char *name() const=0;
 };
 
@@ -483,10 +493,10 @@ class store_key_field: public store_key
       copy_field.set(to_field,from_field,0);
     }
   }
-  bool copy()
+  enum store_key_result copy()
   {
     copy_field.do_copy(&copy_field);
-    return err != 0;
+    return err != 0 ? STORE_KEY_FATAL : STORE_KEY_OK;
   }
   const char *name() const { return field_name; }
 };
@@ -503,9 +513,11 @@ public:
 	       null_ptr_arg ? null_ptr_arg : item_arg->maybe_null ?
 	       &err : NullS, length), item(item_arg)
   {}
-  bool copy()
+  enum store_key_result copy()
   {
-    return item->save_in_field_no_warnings(to_field, 1) || err != 0;
+    int res= item->save_in_field(to_field, 1);
+    return (err != 0 || res > 2 ? STORE_KEY_FATAL : (store_key_result) res); 
+	                 
   }
   const char *name() const { return "func"; }
 };
@@ -523,15 +535,19 @@ public:
 		    &err : NullS, length, item_arg), inited(0)
   {
   }
-  bool copy()
+  enum store_key_result copy()
   {
+    int res;
     if (!inited)
     {
       inited=1;
-      if (item->save_in_field(to_field, 1))
-	err= 1;
+      if ((res= item->save_in_field(to_field, 1)))
+      {       
+        if (!err)
+          err= res;
+      }
     }
-    return err != 0;
+    return (err > 2 ?  STORE_KEY_FATAL : (store_key_result) err);
   }
   const char *name() const { return "const"; }
 };
diff --git a/sql/sql_show.cc b/sql/sql_show.cc
index 1e34f32184a..12025c82da6 100644
--- a/sql/sql_show.cc
+++ b/sql/sql_show.cc
@@ -348,7 +348,7 @@ mysqld_show_create(THD *thd, TABLE_LIST *table_list)
                       table_list->table_name));
 
   /* Only one table for now, but VIEW can involve several tables */
-  if (open_and_lock_tables(thd, table_list))
+  if (open_normal_and_derived_tables(thd, table_list))
   {
     DBUG_RETURN(TRUE);
   }
@@ -448,25 +448,32 @@ bool mysqld_show_create_db(THD *thd, char *dbname,
     DBUG_RETURN(TRUE);
   }
 #endif
-
-  (void) sprintf(path,"%s/%s",mysql_data_home, dbname);
-  length=unpack_dirname(path,path);		// Convert if not unix
-  found_libchar= 0;
-  if (length && path[length-1] == FN_LIBCHAR)
+  if (!my_strcasecmp(system_charset_info, dbname,
+                     information_schema_name.str))
   {
-    found_libchar= 1;
-    path[length-1]=0;				// remove ending '\'
+    dbname= information_schema_name.str;
+    create.default_table_charset= system_charset_info;
   }
-  if (access(path,F_OK))
+  else
   {
-    my_error(ER_BAD_DB_ERROR, MYF(0), dbname);
-    DBUG_RETURN(TRUE);
+    (void) sprintf(path,"%s/%s",mysql_data_home, dbname);
+    length=unpack_dirname(path,path);		// Convert if not unix
+    found_libchar= 0;
+    if (length && path[length-1] == FN_LIBCHAR)
+    {
+      found_libchar= 1;
+      path[length-1]=0;				// remove ending '\'
+    }
+    if (access(path,F_OK))
+    {
+      my_error(ER_BAD_DB_ERROR, MYF(0), dbname);
+      DBUG_RETURN(TRUE);
+    }
+    if (found_libchar)
+      path[length-1]= FN_LIBCHAR;
+    strmov(path+length, MY_DB_OPT_FILE);
+    load_db_opt(thd, path, &create);
   }
-  if (found_libchar)
-    path[length-1]= FN_LIBCHAR;
-  strmov(path+length, MY_DB_OPT_FILE);
-  load_db_opt(thd, path, &create);
-
   List<Item> field_list;
   field_list.push_back(new Item_empty_string("Database",NAME_LEN));
   field_list.push_back(new Item_empty_string("Create Database",1024));
@@ -540,8 +547,7 @@ mysqld_list_fields(THD *thd, TABLE_LIST *table_list, const char *wild)
   DBUG_ENTER("mysqld_list_fields");
   DBUG_PRINT("enter",("table: %s",table_list->table_name));
 
-  table_list->lock_type= TL_UNLOCK;
-  if (open_and_lock_tables(thd, table_list))
+  if (open_normal_and_derived_tables(thd, table_list))
     DBUG_VOID_RETURN;
   table= table_list->table;
 
@@ -1096,7 +1102,7 @@ public:
   char *query;
 };
 
-#ifdef __GNUC__
+#ifdef HAVE_EXPLICIT_TEMPLATE_INSTANTIATION
 template class I_List<thread_info>;
 #endif
 
@@ -1945,7 +1951,7 @@ int get_all_tables(THD *thd, TABLE_LIST *tables, COND *cond)
     bool res;
 
     lex->all_selects_list= lsel;
-    res= open_and_lock_tables(thd, show_table_list);
+    res= open_normal_and_derived_tables(thd, show_table_list);
     if (schema_table->process_table(thd, show_table_list,
                                     table, res, show_table_list->db,
                                     show_table_list->alias))
@@ -2050,7 +2056,7 @@ int get_all_tables(THD *thd, TABLE_LIST *tables, COND *cond)
             show_table_list->lock_type= lock_type;
             lex->all_selects_list= &sel;
             lex->derived_tables= 0;
-            res= open_and_lock_tables(thd, show_table_list);
+            res= open_normal_and_derived_tables(thd, show_table_list);
             if (schema_table->process_table(thd, show_table_list, table,
                                             res, base_name,
                                             show_table_list->alias))
@@ -2646,6 +2652,7 @@ bool store_schema_proc(THD *thd, TABLE *table, TABLE *proc_table,
     restore_record(table, s->default_values);
     if (!wild || !wild[0] || !wild_compare(sp_name, wild, 0))
     {
+      int enum_idx= proc_table->field[5]->val_int();
       table->field[3]->store(sp_name, strlen(sp_name), cs);
       get_field(thd->mem_root, proc_table->field[3], &tmp_string);
       table->field[0]->store(tmp_string.ptr(), tmp_string.length(), cs);
@@ -2667,10 +2674,8 @@ bool store_schema_proc(THD *thd, TABLE *table, TABLE *proc_table,
       table->field[10]->store("SQL", 3, cs);
       get_field(thd->mem_root, proc_table->field[6], &tmp_string);
       table->field[11]->store(tmp_string.ptr(), tmp_string.length(), cs);
-      if (proc_table->field[5]->val_int() == SP_CONTAINS_SQL)
-      {
-        table->field[12]->store("CONTAINS SQL", 12 , cs);
-      }
+      table->field[12]->store(sp_data_access_name[enum_idx].str, 
+                              sp_data_access_name[enum_idx].length , cs);
       get_field(thd->mem_root, proc_table->field[7], &tmp_string);
       table->field[14]->store(tmp_string.ptr(), tmp_string.length(), cs);
       bzero((char *)&time, sizeof(time));
@@ -3874,7 +3879,7 @@ ST_SCHEMA_TABLE schema_tables[]=
 };
 
 
-#ifdef __GNUC__
+#ifdef HAVE_EXPLICIT_TEMPLATE_INSTANTIATION
 template class List_iterator_fast<char>;
 template class List<char>;
 #endif
diff --git a/sql/sql_table.cc b/sql/sql_table.cc
index 8a29c481dc1..553a853bede 100644
--- a/sql/sql_table.cc
+++ b/sql/sql_table.cc
@@ -39,6 +39,8 @@ static int copy_data_between_tables(TABLE *from,TABLE *to,
 				    uint order_num, ORDER *order,
 				    ha_rows *copied,ha_rows *deleted);
 static bool prepare_blob_field(THD *thd, create_field *sql_field);
+static bool check_engine(THD *thd, const char *table_name,
+                         enum db_type *new_engine);                             
 
 
 /*
@@ -254,21 +256,23 @@ int mysql_rm_table_part2(THD *thd, TABLE_LIST *tables, bool if_exists,
       build_table_path(path, sizeof(path), db, alias, reg_ext);
     }
     if (drop_temporary ||
-        (access(path,F_OK) &&
-         ha_create_table_from_engine(thd,db,alias,TRUE)) ||
+       (access(path,F_OK) &&
+         ha_create_table_from_engine(thd,db,alias)) ||
         (!drop_view && mysql_frm_type(path) != FRMTYPE_TABLE))
     {
+      // Table was not found on disk and table can't be created from engine
       if (if_exists)
 	push_warning_printf(thd, MYSQL_ERROR::WARN_LEVEL_NOTE,
 			    ER_BAD_TABLE_ERROR, ER(ER_BAD_TABLE_ERROR),
 			    table->table_name);
       else
-	error= 1;
+        error= 1;
+
     }
     else
     {
       char *end;
-      db_type table_type= get_table_type(path);
+      db_type table_type= get_table_type(thd, path);
       *(end=fn_ext(path))=0;			// Remove extension for delete
       error= ha_delete_table(thd, table_type, path, table->table_name,
                              !dont_log_query);
@@ -1490,7 +1494,6 @@ bool mysql_create_table(THD *thd,const char *db, const char *table_name,
   KEY		*key_info_buffer;
   handler	*file;
   bool		error= TRUE;
-  enum db_type	new_db_type;
   DBUG_ENTER("mysql_create_table");
 
   /* Check for duplicate fields and check type of table to create */
@@ -1500,16 +1503,8 @@ bool mysql_create_table(THD *thd,const char *db, const char *table_name,
                MYF(0));
     DBUG_RETURN(TRUE);
   }
-  if ((new_db_type= ha_checktype(create_info->db_type)) !=
-      create_info->db_type)
-  {
-    create_info->db_type= new_db_type;
-    push_warning_printf(thd, MYSQL_ERROR::WARN_LEVEL_WARN,
-			ER_WARN_USING_OTHER_HANDLER,
-			ER(ER_WARN_USING_OTHER_HANDLER),
-			ha_get_storage_engine(new_db_type),
-			table_name);
-  }
+  if (check_engine(thd, table_name, &create_info->db_type))
+    DBUG_RETURN(TRUE);
   db_options= create_info->table_options;
   if (create_info->row_type == ROW_TYPE_DYNAMIC)
     db_options|=HA_OPTION_PACK_RECORD;
@@ -1611,15 +1606,14 @@ bool mysql_create_table(THD *thd,const char *db, const char *table_name,
   {
     bool create_if_not_exists =
       create_info->options & HA_LEX_CREATE_IF_NOT_EXISTS;
-    if (!ha_create_table_from_engine(thd, db, table_name,
-				     create_if_not_exists))
+    if (ha_table_exists_in_engine(thd, db, table_name))
     {
-      DBUG_PRINT("info", ("Table already existed in handler"));
+      DBUG_PRINT("info", ("Table with same name already existed in handler"));
 
       if (create_if_not_exists)
       {
-       create_info->table_existed= 1;   // Mark that table existed
-       error= FALSE;
+        create_info->table_existed= 1;   // Mark that table existed
+        error= FALSE;
       }
       else
        my_error(ER_TABLE_EXISTS_ERROR, MYF(0), table_name);
@@ -1740,7 +1734,7 @@ TABLE *create_table_from_items(THD *thd, HA_CREATE_INFO *create_info,
       field=item->tmp_table_field(&tmp_table);
     else
       field=create_tmp_field(thd, &tmp_table, item, item->type(),
-                             (Item ***) 0, &tmp_field,0,0,0);
+                             (Item ***) 0, &tmp_field, 0, 0, 0, 0);
     if (!field ||
 	!(cr_field=new create_field(field,(item->type() == Item::FIELD_ITEM ?
 					   ((Item_field *)item)->field :
@@ -3125,16 +3119,9 @@ bool mysql_alter_table(THD *thd,char *new_db, char *new_name,
   old_db_type= table->s->db_type;
   if (create_info->db_type == DB_TYPE_DEFAULT)
     create_info->db_type= old_db_type;
-  if ((new_db_type= ha_checktype(create_info->db_type)) !=
-      create_info->db_type)
-  {
-    create_info->db_type= new_db_type;
-    push_warning_printf(thd, MYSQL_ERROR::WARN_LEVEL_WARN,
-			ER_WARN_USING_OTHER_HANDLER,
-			ER(ER_WARN_USING_OTHER_HANDLER),
-			ha_get_storage_engine(new_db_type),
-			new_name);
-  }
+  if (check_engine(thd, new_name, &create_info->db_type))
+    DBUG_RETURN(TRUE);
+  new_db_type= create_info->db_type;
   if (create_info->row_type == ROW_TYPE_NOT_USED)
     create_info->row_type= table->s->row_type;
 
@@ -4132,3 +4119,24 @@ bool mysql_checksum_table(THD *thd, TABLE_LIST *tables, HA_CHECK_OPT *check_opt)
     table->table=0;
   DBUG_RETURN(TRUE);
 }
+
+static bool check_engine(THD *thd, const char *table_name,
+                         enum db_type *new_engine)
+{
+  enum db_type req_engine= *new_engine;
+  bool no_substitution= 
+        test(thd->variables.sql_mode & MODE_NO_ENGINE_SUBSTITUTION);
+  if ((*new_engine= 
+       ha_checktype(thd, req_engine, no_substitution, 1)) == DB_TYPE_UNKNOWN)
+    return TRUE;
+
+  if (req_engine != *new_engine)
+  {
+    push_warning_printf(thd, MYSQL_ERROR::WARN_LEVEL_WARN,
+                       ER_WARN_USING_OTHER_HANDLER,
+                       ER(ER_WARN_USING_OTHER_HANDLER),
+                       ha_get_storage_engine(*new_engine),
+                       table_name);
+  }
+  return FALSE;
+}
diff --git a/sql/sql_union.cc b/sql/sql_union.cc
index 912636b6cf3..6e307dda5be 100644
--- a/sql/sql_union.cc
+++ b/sql/sql_union.cc
@@ -274,7 +274,7 @@ bool st_select_lex_unit::prepare(THD *thd_arg, select_result *sel_result,
       all collations together for UNION.
     */
     List_iterator_fast<Item> tp(types);
-    Item_arena *arena= thd->current_arena;
+    Query_arena *arena= thd->current_arena;
     Item *type;
 
     while ((type= tp++))
@@ -308,7 +308,7 @@ bool st_select_lex_unit::prepare(THD *thd_arg, select_result *sel_result,
     if (!item_list.elements)
     {
       Field **field;
-      Item_arena *tmp_arena,backup;
+      Query_arena *tmp_arena,backup;
       tmp_arena= thd->change_arena_if_needed(&backup);
 
       for (field= table->field; *field; field++)
@@ -553,7 +553,6 @@ bool st_select_lex_unit::exec()
 bool st_select_lex_unit::cleanup()
 {
   int error= 0;
-  JOIN *join;
   DBUG_ENTER("st_select_lex_unit::cleanup");
 
   if (cleaned)
@@ -572,29 +571,17 @@ bool st_select_lex_unit::cleanup()
   }
 
   for (SELECT_LEX *sl= first_select_in_union(); sl; sl= sl->next_select())
+    error|= sl->cleanup();
+
+  if (fake_select_lex)
   {
-    if ((join= sl->join))
-    {
-      error|= sl->join->cleanup();
-      delete join;
-    }
-    else
+    JOIN *join;
+    if ((join= fake_select_lex->join))
     {
-      // it can be DO/SET with subqueries
-      for (SELECT_LEX_UNIT *lex_unit= sl->first_inner_unit();
-	   lex_unit != 0;
-	   lex_unit= lex_unit->next_unit())
-      {
-	error|= lex_unit->cleanup();
-      }
+      join->tables_list= 0;
+      join->tables= 0;
     }
-  }
-  if (fake_select_lex && (join= fake_select_lex->join))
-  {
-    join->tables_list= 0;
-    join->tables= 0;
-    error|= join->cleanup();
-    delete join;
+    error|= fake_select_lex->cleanup();
   }
 
   DBUG_RETURN(error);
@@ -650,3 +637,25 @@ bool st_select_lex_unit::change_result(select_subselect *result,
     res= fake_select_lex->join->change_result(result);
   return (res);
 }
+
+
+bool st_select_lex::cleanup()
+{
+  bool error= FALSE;
+  DBUG_ENTER("st_select_lex::cleanup()");
+
+  if (join)
+  {
+    DBUG_ASSERT((st_select_lex*)join->select_lex == this);
+    error|= join->destroy();
+    delete join;
+    join= 0;
+  }
+  for (SELECT_LEX_UNIT *lex_unit= first_inner_unit(); lex_unit ;
+       lex_unit= lex_unit->next_unit())
+  {
+    error|= lex_unit->cleanup();
+  }
+  DBUG_RETURN(error);
+}
+
diff --git a/sql/sql_view.cc b/sql/sql_view.cc
index 9f0c3260d8f..0b351407c13 100644
--- a/sql/sql_view.cc
+++ b/sql/sql_view.cc
@@ -36,6 +36,61 @@ TYPELIB updatable_views_with_limit_typelib=
 
 
 /*
+  Make a unique name for an anonymous view column
+  SYNOPSIS
+    target        reference to the item for which a new name has to be made
+    item_list     list of items within which we should check uniqueness of
+                  the created name
+    last_element  the last element of the list above
+
+  NOTE
+    Unique names are generated by adding 'My_exp_' to the old name of the
+    column. In case the name that was created this way already exists, we
+    add a numeric postfix to its end (i.e. "1") and increase the number
+    until the name becomes unique. If the generated name is longer than
+    NAME_LEN, it is truncated.
+*/
+
+static void make_unique_view_field_name(Item *target,
+                                        List<Item> &item_list,
+                                        Item *last_element)
+{
+  char *name= (target->orig_name ?
+               target->orig_name :
+               target->name);
+  uint name_len;
+  uint attempt= 0;
+  char buff[NAME_LEN+1];
+  for (;; attempt++)
+  {
+    Item *check;
+    List_iterator_fast<Item> itc(item_list);
+    bool ok= TRUE;
+
+    if (attempt)
+      name_len= my_snprintf(buff, NAME_LEN, "My_exp_%d_%s", attempt, name);
+    else
+      name_len= my_snprintf(buff, NAME_LEN, "My_exp_%s", name);
+
+    do
+    {
+      check= itc++;
+      if (check != target &&
+          my_strcasecmp(system_charset_info, buff, check->name) == 0)
+      {
+        ok= FALSE;
+        break;
+      }
+    } while (check != last_element);
+    if (ok)
+      break;
+  }
+
+  target->orig_name= target->name;
+  target->set_name(buff, name_len, system_charset_info);
+}
+
+/*
   Creating/altering VIEW procedure
 
   SYNOPSIS
@@ -240,24 +295,36 @@ bool mysql_create_view(THD *thd,
       goto err;
     }
     while ((item= it++, name= nm++))
+    {
       item->set_name(name->str, name->length, system_charset_info);
+      item->is_autogenerated_name= FALSE;
+    }
   }
 
   /* Test absence of duplicates names */
   {
     Item *item;
     List_iterator_fast<Item> it(select_lex->item_list);
-    it++;
     while ((item= it++))
     {
       Item *check;
       List_iterator_fast<Item> itc(select_lex->item_list);
+      /* treat underlying fields like set by user names */
+      if (item->real_item()->type() == Item::FIELD_ITEM)
+        item->is_autogenerated_name= FALSE;
       while ((check= itc++) && check != item)
       {
-        if (strcmp(item->name, check->name) == 0)
+        if (my_strcasecmp(system_charset_info, item->name, check->name) == 0)
         {
-          my_error(ER_DUP_FIELDNAME, MYF(0), item->name);
-          DBUG_RETURN(TRUE);
+          if (item->is_autogenerated_name)
+            make_unique_view_field_name(item, select_lex->item_list, item);
+          else if (check->is_autogenerated_name)
+            make_unique_view_field_name(check, select_lex->item_list, item);
+          else
+          {
+            my_error(ER_DUP_FIELDNAME, MYF(0), item->name);
+            DBUG_RETURN(TRUE);
+          }
         }
       }
     }
@@ -579,7 +646,7 @@ mysql_make_view(File_parser *parser, TABLE_LIST *table)
     For now we assume that tables will not be changed during PS life (it
     will be TRUE as far as we make new table cache).
   */
-  Item_arena *arena= thd->current_arena, backup;
+  Query_arena *arena= thd->current_arena, backup;
   if (arena->is_conventional())
     arena= 0;
   else
diff --git a/sql/sql_yacc.yy b/sql/sql_yacc.yy
index 892d2516808..360bc421965 100644
--- a/sql/sql_yacc.yy
+++ b/sql/sql_yacc.yy
@@ -52,7 +52,7 @@ const LEX_STRING null_lex_str={0,0};
 		      ER_WARN_DEPRECATED_SYNTAX,                    \
 		      ER(ER_WARN_DEPRECATED_SYNTAX), (A), (B));
 
-#define TEST_ASSERT(A)                  \
+#define YYERROR_UNLESS(A)                  \
   if (!(A))                             \
   {					\
     yyerror(ER(ER_SYNTAX_ERROR));	\
@@ -914,7 +914,7 @@ deallocate:
         {
           THD *thd=YYTHD;
           LEX *lex= thd->lex;
-          if (thd->command == COM_PREPARE)
+          if (thd->command == COM_STMT_PREPARE)
           {
             yyerror(ER(ER_SYNTAX_ERROR));
             YYABORT;
@@ -939,7 +939,7 @@ prepare:
         {
           THD *thd=YYTHD;
           LEX *lex= thd->lex;
-          if (thd->command == COM_PREPARE)
+          if (thd->command == COM_STMT_PREPARE)
           {
             yyerror(ER(ER_SYNTAX_ERROR));
             YYABORT;
@@ -974,7 +974,7 @@ execute:
         {
           THD *thd=YYTHD;
           LEX *lex= thd->lex;
-          if (thd->command == COM_PREPARE)
+          if (thd->command == COM_STMT_PREPARE)
           {
             yyerror(ER(ER_SYNTAX_ERROR));
             YYABORT;
@@ -4107,7 +4107,10 @@ select_item:
 	    if (add_item_to_list(YYTHD, $2))
 	      YYABORT;
 	    if ($4.str)
-	      $2->set_name($4.str,$4.length,system_charset_info);
+            {
+	      $2->set_name($4.str, $4.length, system_charset_info);
+              $2->is_autogenerated_name= FALSE;
+            }
 	    else if (!$2->name) {
 	      char *str = $1;
 	      if (str[-1] == '`')
@@ -4913,9 +4916,12 @@ udf_expr:
 	remember_name expr remember_end select_alias
 	{
 	  if ($4.str)
-	    $2->set_name($4.str,$4.length,system_charset_info);
+          {
+	    $2->set_name($4.str, $4.length, system_charset_info);
+            $2->is_autogenerated_name= FALSE;
+          }
 	  else
-	    $2->set_name($1,(uint) ($3 - $1), YYTHD->charset());
+	    $2->set_name($1, (uint) ($3 - $1), YYTHD->charset());
 	  $$= $2;
 	}
 	;
@@ -5094,7 +5100,7 @@ table_ref:
         ;
 
 join_table_list:
-	derived_table_list		{ TEST_ASSERT($$=$1); }
+	derived_table_list		{ YYERROR_UNLESS($$=$1); }
 	;
 
 /* Warning - may return NULL in case of incomplete SELECT */
@@ -5102,41 +5108,41 @@ derived_table_list:
         table_ref { $$=$1; }
         | derived_table_list ',' table_ref
           {
-            TEST_ASSERT($1 && ($$=$3));
+            YYERROR_UNLESS($1 && ($$=$3));
           }
         ;
 
 join_table:
-        table_ref normal_join table_ref { TEST_ASSERT($1 && ($$=$3)); }
+        table_ref normal_join table_ref { YYERROR_UNLESS($1 && ($$=$3)); }
 	| table_ref STRAIGHT_JOIN table_factor
-	  { TEST_ASSERT($1 && ($$=$3)); $3->straight=1; }
+	  { YYERROR_UNLESS($1 && ($$=$3)); $3->straight=1; }
 	| table_ref normal_join table_ref ON expr
-	  { TEST_ASSERT($1 && ($$=$3)); add_join_on($3,$5); }
+	  { YYERROR_UNLESS($1 && ($$=$3)); add_join_on($3,$5); }
         | table_ref STRAIGHT_JOIN table_factor ON expr
-          { TEST_ASSERT($1 && ($$=$3)); $3->straight=1; add_join_on($3,$5); }
+          { YYERROR_UNLESS($1 && ($$=$3)); $3->straight=1; add_join_on($3,$5); }
 	| table_ref normal_join table_ref
 	  USING
 	  {
 	    SELECT_LEX *sel= Select;
-            TEST_ASSERT($1 && $3);
+            YYERROR_UNLESS($1 && $3);
             sel->save_names_for_using_list($1, $3);
 	  }
 	  '(' using_list ')'
 	  { add_join_on($3,$7); $$=$3; }
 
 	| table_ref LEFT opt_outer JOIN_SYM table_ref ON expr
-	  { TEST_ASSERT($1 && $5); add_join_on($5,$7); $5->outer_join|=JOIN_TYPE_LEFT; $$=$5; }
+	  { YYERROR_UNLESS($1 && $5); add_join_on($5,$7); $5->outer_join|=JOIN_TYPE_LEFT; $$=$5; }
 	| table_ref LEFT opt_outer JOIN_SYM table_factor
 	  {
 	    SELECT_LEX *sel= Select;
-            TEST_ASSERT($1 && $5);
+            YYERROR_UNLESS($1 && $5);
             sel->save_names_for_using_list($1, $5);
 	  }
 	  USING '(' using_list ')'
 	  { add_join_on($5,$9); $5->outer_join|=JOIN_TYPE_LEFT; $$=$5; }
 	| table_ref NATURAL LEFT opt_outer JOIN_SYM table_factor
 	  {
-            TEST_ASSERT($1 && $6);
+            YYERROR_UNLESS($1 && $6);
 	    add_join_natural($1,$6);
 	    $6->outer_join|=JOIN_TYPE_LEFT;
 	    $$=$6;
@@ -5144,7 +5150,7 @@ join_table:
 	| table_ref RIGHT opt_outer JOIN_SYM table_ref ON expr
           {
 	    LEX *lex= Lex;
-            TEST_ASSERT($1 && $5);
+            YYERROR_UNLESS($1 && $5);
             if (!($$= lex->current_select->convert_right_join()))
               YYABORT;
             add_join_on($$, $7);
@@ -5152,7 +5158,7 @@ join_table:
 	| table_ref RIGHT opt_outer JOIN_SYM table_factor
 	  {
 	    SELECT_LEX *sel= Select;
-            TEST_ASSERT($1 && $5);
+            YYERROR_UNLESS($1 && $5);
             sel->save_names_for_using_list($1, $5);
 	  }
 	  USING '(' using_list ')'
@@ -5164,14 +5170,14 @@ join_table:
           }
 	| table_ref NATURAL RIGHT opt_outer JOIN_SYM table_factor
 	  {
-            TEST_ASSERT($1 && $6);
+            YYERROR_UNLESS($1 && $6);
 	    add_join_natural($6,$1);
 	    LEX *lex= Lex;
             if (!($$= lex->current_select->convert_right_join()))
               YYABORT;
 	  }
 	| table_ref NATURAL JOIN_SYM table_factor
-	  { TEST_ASSERT($1 && ($$=$4)); add_join_natural($1,$4); };
+	  { YYERROR_UNLESS($1 && ($$=$4)); add_join_natural($1,$4); };
 
 
 normal_join:
@@ -5200,7 +5206,7 @@ table_factor:
           sel->add_joined_table($$);
 	}
 	| '{' ident table_ref LEFT OUTER JOIN_SYM table_ref ON expr '}'
-	  { TEST_ASSERT($3 && $7); add_join_on($7,$9); $7->outer_join|=JOIN_TYPE_LEFT; $$=$7; }
+	  { YYERROR_UNLESS($3 && $7); add_join_on($7,$9); $7->outer_join|=JOIN_TYPE_LEFT; $$=$7; }
 	| select_derived_init get_select_lex select_derived2
           {
             LEX *lex= Lex;
@@ -5691,7 +5697,8 @@ procedure_item:
 	    if (add_proc_to_list(lex->thd, $2))
 	      YYABORT;
 	    if (!$2->name)
-	      $2->set_name($1,(uint) ((char*) lex->tok_end - $1), YYTHD->charset());
+	      $2->set_name($1,(uint) ((char*) lex->tok_end - $1),
+                           YYTHD->charset());
 	  }
           ;
 
@@ -6113,9 +6120,24 @@ insert_update_elem:
 	simple_ident_nospvar equal expr_or_default
 	  {
 	  LEX *lex= Lex;
+          uint8 tmp= MY_ITEM_PREFER_1ST_TABLE;
 	  if (lex->update_list.push_back($1) || 
 	      lex->value_list.push_back($3))
 	      YYABORT;
+          /* 
+            INSERT INTO a1(a) SELECT b1.a FROM b1 ON DUPLICATE KEY 
+              UPDATE a= a + b1.b
+   
+            Set MY_ITEM_PREFER_1ST_TABLE flag to $1 and $3 items
+            to prevent find_field_in_tables() doing further item searching 
+            if it finds item occurence in first table in insert_table_list. 
+            This allows to avoid ambiguity in resolving 'a' field in 
+            example above.
+          */
+          $1->walk(&Item::set_flags_processor,
+                  (byte *) &tmp);
+          $3->walk(&Item::set_flags_processor,
+                  (byte *) &tmp);
 	  };
 
 opt_low_priority:
@@ -6902,7 +6924,7 @@ param_marker:
         {
           THD *thd=YYTHD;
 	  LEX *lex= thd->lex;
-          if (thd->command == COM_PREPARE)
+          if (thd->command == COM_STMT_PREPARE)
           {
             Item_param *item= new Item_param((uint) (lex->tok_start -
                                                      (uchar *) thd->query));
@@ -8731,21 +8753,21 @@ xa: XA_SYM begin_or_start xid opt_join_or_resume
 
 xid: text_string
      {
-       TEST_ASSERT($1->length() <= MAXGTRIDSIZE);
+       YYERROR_UNLESS($1->length() <= MAXGTRIDSIZE);
        if (!(Lex->xid=(XID *)YYTHD->alloc(sizeof(XID))))
          YYABORT;
        Lex->xid->set(1L, $1->ptr(), $1->length(), 0, 0);
      }
      | text_string ',' text_string
      {
-       TEST_ASSERT($1->length() <= MAXGTRIDSIZE && $3->length() <= MAXBQUALSIZE);
+       YYERROR_UNLESS($1->length() <= MAXGTRIDSIZE && $3->length() <= MAXBQUALSIZE);
        if (!(Lex->xid=(XID *)YYTHD->alloc(sizeof(XID))))
          YYABORT;
        Lex->xid->set(1L, $1->ptr(), $1->length(), $3->ptr(), $3->length());
      }
      | text_string ',' text_string ',' ulong_num
      {
-       TEST_ASSERT($1->length() <= MAXGTRIDSIZE && $3->length() <= MAXBQUALSIZE);
+       YYERROR_UNLESS($1->length() <= MAXGTRIDSIZE && $3->length() <= MAXBQUALSIZE);
        if (!(Lex->xid=(XID *)YYTHD->alloc(sizeof(XID))))
          YYABORT;
        Lex->xid->set($5, $1->ptr(), $1->length(), $3->ptr(), $3->length());
diff --git a/sql/structs.h b/sql/structs.h
index 8f053f20776..03176b47360 100644
--- a/sql/structs.h
+++ b/sql/structs.h
@@ -20,6 +20,8 @@
 struct st_table;
 class Field;
 
+#define STRING_WITH_LEN(X)  X, (sizeof(X)-1)
+
 typedef struct st_lex_string
 {
   char *str;
diff --git a/sql/table.cc b/sql/table.cc
index 33cee79eb61..6677453969b 100644
--- a/sql/table.cc
+++ b/sql/table.cc
@@ -163,7 +163,7 @@ int openfrm(THD *thd, const char *name, const char *alias, uint db_stat,
   if (share->frm_version == FRM_VER_TRUE_VARCHAR -1 && head[33] == 5)
     share->frm_version= FRM_VER_TRUE_VARCHAR;
 
-  share->db_type= ha_checktype((enum db_type) (uint) *(head+3));
+  share->db_type= ha_checktype(thd,(enum db_type) (uint) *(head+3),0,0);
   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);
@@ -1341,8 +1341,8 @@ void append_unescaped(String *res, const char *pos, uint length)
 
 	/* Create a .frm file */
 
-File create_frm(register my_string name, uint reclength, uchar *fileinfo,
-		HA_CREATE_INFO *create_info, uint keys)
+File create_frm(THD *thd, register my_string name, uint reclength,
+		uchar *fileinfo, HA_CREATE_INFO *create_info, uint keys)
 {
   register File file;
   ulong length;
@@ -1375,7 +1375,7 @@ File create_frm(register my_string name, uint reclength, uchar *fileinfo,
     fileinfo[1]= 1;
     fileinfo[2]= FRM_VER+3+ test(create_info->varchar);
 
-    fileinfo[3]= (uchar) ha_checktype(create_info->db_type);
+    fileinfo[3]= (uchar) ha_checktype(thd,create_info->db_type,0,0);
     fileinfo[4]=1;
     int2store(fileinfo+6,IO_SIZE);		/* Next block starts here */
     key_length=keys*(7+NAME_LEN+MAX_REF_PARTS*9)+16;
@@ -1637,7 +1637,7 @@ bool check_column_name(const char *name)
 ** Get type of table from .frm file
 */
 
-db_type get_table_type(const char *name)
+db_type get_table_type(THD *thd, const char *name)
 {
   File	 file;
   uchar head[4];
@@ -1653,7 +1653,7 @@ db_type get_table_type(const char *name)
       (head[2] != FRM_VER && head[2] != FRM_VER+1 &&
        (head[2] < FRM_VER+3 || head[2] > FRM_VER+4)))
     DBUG_RETURN(DB_TYPE_UNKNOWN);
-  DBUG_RETURN(ha_checktype((enum db_type) (uint) *(head+3)));
+  DBUG_RETURN(ha_checktype(thd,(enum db_type) (uint) *(head+3),0,0));
 }
 
 
@@ -1926,7 +1926,7 @@ bool st_table_list::setup_ancestor(THD *thd, Item **conds,
       (check_opt_type == VIEW_CHECK_CASCADED &&
        ancestor->check_option))
   {
-    Item_arena *arena= thd->current_arena, backup;
+    Query_arena *arena= thd->current_arena, backup;
     TABLE_LIST *tbl= this;
     if (arena->is_conventional())
       arena= 0;                                   // For easier test
@@ -2016,7 +2016,7 @@ bool st_table_list::setup_ancestor(THD *thd, Item **conds,
   /* full text function moving to current select */
   if (view->select_lex.ftfunc_list->elements)
   {
-    Item_arena *arena= thd->current_arena, backup;
+    Query_arena *arena= thd->current_arena, backup;
     if (arena->is_conventional())
       arena= 0;                                   // For easier test
     else
@@ -2236,7 +2236,7 @@ const char *Field_iterator_view::name()
 ** Instansiate templates
 *****************************************************************************/
 
-#ifdef __GNUC__
+#ifdef HAVE_EXPLICIT_TEMPLATE_INSTANTIATION
 template class List<String>;
 template class List_iterator<String>;
 #endif
diff --git a/sql/unireg.cc b/sql/unireg.cc
index e8aad2fedd0..7f170b3ef87 100644
--- a/sql/unireg.cc
+++ b/sql/unireg.cc
@@ -113,7 +113,7 @@ bool mysql_create_frm(THD *thd, my_string file_name,
   }
   reclength=uint2korr(forminfo+266);
 
-  if ((file=create_frm(file_name, reclength, fileinfo,
+  if ((file=create_frm(thd, file_name, reclength, fileinfo,
 		       create_info, keys)) < 0)
   {
     my_free((gptr) screen_buff,MYF(0));