diff --git a/mysql-test/t/ps.test b/mysql-test/t/ps.test
index 285b5fb0aa3..23b33156b48 100644
--- a/mysql-test/t/ps.test
+++ b/mysql-test/t/ps.test
@@ -114,6 +114,9 @@ set @fvar= 123.4567;
 prepare stmt1 from @fvar;
 
 drop table t1,t2;
+deallocate prepare stmt3;
+deallocate prepare stmt4;
+deallocate prepare stmt5;
 
 #
 # Bug #4105: Server crash on attempt to prepare a statement with character
@@ -257,6 +260,7 @@ prepare `ü` from 'select 1234';
 execute `ü` ;
 set names latin1;
 execute `ü`;
+deallocate prepare `ü`;
 set names default;
 
 
@@ -921,6 +925,143 @@ select length(a) from t1;
 drop table t1;
 deallocate prepare stmt;
 
+#
+# Bug#16248 "WHERE (col1,col2) IN ((?,?)) gives wrong results":
+# check that ROW implementation is reexecution-friendly.
+#
+create table t1 (col1 integer, col2 integer);
+insert into t1 values(100,100),(101,101),(102,102),(103,103);
+prepare stmt from 'select col1, col2 from t1 where (col1, col2) in ((?,?))';
+set @a=100, @b=100;
+execute stmt using @a,@b;
+set @a=101, @b=101;
+execute stmt using @a,@b;
+set @a=102, @b=102;
+execute stmt using @a,@b;
+set @a=102, @b=103;
+execute stmt using @a,@b;
+deallocate prepare stmt;
+drop table t1;
+
+#
+# Bug#16365 Prepared Statements: DoS with too many open statements
+# Check that the limit @@max_prpeared_stmt_count works.
+#
+# Save the old value
+set @old_max_prepared_stmt_count= @@max_prepared_stmt_count;
+#
+# Disable prepared statement protocol: in this test we set
+# @@max_prepared_stmt_count to 0 or 1 and would like to test the limit
+# manually.
+#
+--disable_ps_protocol
+#
+# A. Check that the new variables are present in SHOW VARIABLES list.
+#
+show variables like 'max_prepared_stmt_count';
+show variables like 'prepared_stmt_count';
+#
+# B. Check that the new variables are selectable.
+#
+select @@max_prepared_stmt_count, @@prepared_stmt_count;
+#
+# C. Check that max_prepared_stmt_count is settable (global only),
+#    whereas prepared_stmt_count is readonly.
+#
+set global max_prepared_stmt_count=-1;
+select @@max_prepared_stmt_count;
+set global max_prepared_stmt_count=10000000000000000;
+select @@max_prepared_stmt_count;
+set global max_prepared_stmt_count=default;
+select @@max_prepared_stmt_count;
+--error 1229 # ER_GLOBAL_VARIABLE
+set @@max_prepared_stmt_count=1;
+--error 1229 # ER_GLOBAL_VARIABLE
+set max_prepared_stmt_count=1;
+--error 1229 # ER_GLOBAL_VARIABLE
+set local max_prepared_stmt_count=1;
+--error 1229 # ER_GLOBAL_VARIABLE
+set local prepared_stmt_count=0;
+--error 1229 # ER_GLOBAL_VARIABLE
+set @@prepared_stmt_count=0;
+--error 1232 # ER_WRONG_TYPE_FOR_VAR
+set global prepared_stmt_count=1;
+# set to a reasonable limit works
+set global max_prepared_stmt_count=1;
+select @@max_prepared_stmt_count;
+#
+# D. Check that the variables actually work.
+#
+set global max_prepared_stmt_count=0;
+select @@max_prepared_stmt_count, @@prepared_stmt_count;
+--error 1105 # ER_UNKNOWN_ERROR
+prepare stmt from "select 1";
+select @@prepared_stmt_count;
+set global max_prepared_stmt_count=1;
+prepare stmt from "select 1";
+select @@prepared_stmt_count;
+--error 1105 # ER_UNKNOWN_ERROR
+prepare stmt1 from "select 1";
+select @@prepared_stmt_count;
+deallocate prepare stmt;
+select @@prepared_stmt_count;
+#
+# E. Check that we can prepare a statement with the same name 
+# successfully, without hitting the limit.
+#
+prepare stmt from "select 1";
+select @@prepared_stmt_count;
+prepare stmt from "select 2";
+select @@prepared_stmt_count;
+#
+# F. We can set the max below the current count. In this case no new 
+# statements should be allowed to prepare.
+#
+select @@prepared_stmt_count, @@max_prepared_stmt_count;
+set global max_prepared_stmt_count=0;
+--error 1105 # ER_UNKNOWN_ERROR
+prepare stmt from "select 1";
+# Result: the old statement is deallocated, the new is not created.
+--error 1243 # ER_UNKNOWN_STMT_HANDLER
+execute stmt;
+select @@prepared_stmt_count;
+--error 1105 # ER_UNKNOWN_ERROR
+prepare stmt from "select 1";
+select @@prepared_stmt_count;
+#
+# G. Show that the variables are up to date even after a connection with all
+# statements in it was terminated.
+#
+set global max_prepared_stmt_count=3;
+select @@max_prepared_stmt_count, @@prepared_stmt_count;
+prepare stmt from "select 1";
+connect (con1,localhost,root,,);
+connection con1;
+prepare stmt from "select 2";
+prepare stmt1 from "select 3";
+--error 1105 # ER_UNKNOWN_ERROR
+prepare stmt2 from "select 4";
+connection default;
+--error 1105 # ER_UNKNOWN_ERROR 
+prepare stmt2 from "select 4";
+select @@max_prepared_stmt_count, @@prepared_stmt_count;
+disconnect con1;
+connection default;
+# Wait for the connection to die: deal with a possible race
+deallocate prepare stmt;
+let $count= `select @@prepared_stmt_count`;
+if ($count)
+{
+--sleep 2
+  let $count= `select @@prepared_stmt_count`;
+}
+select @@max_prepared_stmt_count, @@prepared_stmt_count;
+#
+# Restore the old value.
+#
+set global max_prepared_stmt_count= @old_max_prepared_stmt_count;
+--enable_ps_protocol
+
 # End of 4.1 tests
 
 #
diff --git a/sql/item_row.cc b/sql/item_row.cc
index 75c3f8a2922..f5c8d511025 100644
--- a/sql/item_row.cc
+++ b/sql/item_row.cc
@@ -26,7 +26,7 @@
 */
 
 Item_row::Item_row(List<Item> &arg):
-  Item(), used_tables_cache(0), array_holder(1), const_item_cache(1), with_null(0)
+  Item(), used_tables_cache(0), const_item_cache(1), with_null(0)
 {
 
   //TODO: think placing 2-3 component items in item (as it done for function)
@@ -85,6 +85,20 @@ bool Item_row::fix_fields(THD *thd, Item **ref)
 }
 
 
+void Item_row::cleanup()
+{
+  DBUG_ENTER("Item_row::cleanup");
+
+  Item::cleanup();
+  /* Reset to the original values */
+  used_tables_cache= 0;
+  const_item_cache= 1;
+  with_null= 0;
+
+  DBUG_VOID_RETURN;
+}
+
+
 void Item_row::split_sum_func(THD *thd, Item **ref_pointer_array,
                               List<Item> &fields)
 {
diff --git a/sql/item_row.h b/sql/item_row.h
index 6fbe7436b72..d6dd4371372 100644
--- a/sql/item_row.h
+++ b/sql/item_row.h
@@ -19,7 +19,6 @@ class Item_row: public Item
   Item **items;
   table_map used_tables_cache;
   uint arg_count;
-  bool array_holder;
   bool const_item_cache;
   bool with_null;
 public:
@@ -29,7 +28,6 @@ public:
     items(item->items),
     used_tables_cache(item->used_tables_cache),
     arg_count(item->arg_count),
-    array_holder(0), 
     const_item_cache(item->const_item_cache),
     with_null(0)
   {}
@@ -62,6 +60,7 @@ public:
     return 0;
   };
   bool fix_fields(THD *thd, Item **ref);
+  void cleanup();
   void split_sum_func(THD *thd, Item **ref_pointer_array, List<Item> &fields);
   table_map used_tables() const { return used_tables_cache; };
   bool const_item() const { return const_item_cache; };
diff --git a/sql/mysql_priv.h b/sql/mysql_priv.h
index ca7801039c5..a815e72ee7a 100644
--- a/sql/mysql_priv.h
+++ b/sql/mysql_priv.h
@@ -1166,6 +1166,7 @@ extern ulong slave_net_timeout, slave_trans_retries;
 extern uint max_user_connections;
 extern ulong what_to_log,flush_time;
 extern ulong query_buff_size, thread_stack;
+extern ulong max_prepared_stmt_count, prepared_stmt_count;
 extern ulong binlog_cache_size, max_binlog_cache_size, open_files_limit;
 extern ulong max_binlog_size, max_relay_log_size;
 extern ulong rpl_recovery_rank, thread_cache_size;
@@ -1215,6 +1216,7 @@ extern pthread_mutex_t LOCK_mysql_create_db,LOCK_Acl,LOCK_open,
        LOCK_delayed_status, LOCK_delayed_create, LOCK_crypt, LOCK_timezone,
        LOCK_slave_list, LOCK_active_mi, LOCK_manager, LOCK_global_read_lock,
        LOCK_global_system_variables, LOCK_user_conn,
+       LOCK_prepared_stmt_count,
        LOCK_bytes_sent, LOCK_bytes_received;
 #ifdef HAVE_OPENSSL
 extern pthread_mutex_t LOCK_des_key_file;
diff --git a/sql/mysqld.cc b/sql/mysqld.cc
index 9dd37bbebc9..eed43f91d36 100644
--- a/sql/mysqld.cc
+++ b/sql/mysqld.cc
@@ -409,6 +409,22 @@ ulong specialflag=0;
 ulong binlog_cache_use= 0, binlog_cache_disk_use= 0;
 ulong max_connections, max_connect_errors;
 uint  max_user_connections= 0;
+/*
+  Limit of the total number of prepared statements in the server.
+  Is necessary to protect the server against out-of-memory attacks.
+*/
+ulong max_prepared_stmt_count;
+/*
+  Current total number of prepared statements in the server. This number
+  is exact, and therefore may not be equal to the difference between
+  `com_stmt_prepare' and `com_stmt_close' (global status variables), as
+  the latter ones account for all registered attempts to prepare
+  a statement (including unsuccessful ones).  Prepared statements are
+  currently connection-local: if the same SQL query text is prepared in
+  two different connections, this counts as two distinct prepared
+  statements.
+*/
+ulong prepared_stmt_count=0;
 ulong thread_id=1L,current_pid;
 ulong slow_launch_threads = 0, sync_binlog_period;
 ulong expire_logs_days = 0;
@@ -488,6 +504,14 @@ pthread_mutex_t LOCK_mysql_create_db, LOCK_Acl, LOCK_open, LOCK_thread_count,
 		LOCK_crypt, LOCK_bytes_sent, LOCK_bytes_received,
 	        LOCK_global_system_variables,
 		LOCK_user_conn, LOCK_slave_list, LOCK_active_mi;
+/*
+  The below lock protects access to two global server variables:
+  max_prepared_stmt_count and prepared_stmt_count. These variables
+  set the limit and hold the current total number of prepared statements
+  in the server, respectively. As PREPARE/DEALLOCATE rate in a loaded
+  server may be fairly high, we need a dedicated lock.
+*/
+pthread_mutex_t LOCK_prepared_stmt_count;
 #ifdef HAVE_OPENSSL
 pthread_mutex_t LOCK_des_key_file;
 #endif
@@ -1197,6 +1221,7 @@ static void clean_up_mutexes()
   (void) pthread_mutex_destroy(&LOCK_global_system_variables);
   (void) pthread_mutex_destroy(&LOCK_global_read_lock);
   (void) pthread_mutex_destroy(&LOCK_uuid_generator);
+  (void) pthread_mutex_destroy(&LOCK_prepared_stmt_count);
   (void) pthread_cond_destroy(&COND_thread_count);
   (void) pthread_cond_destroy(&COND_refresh);
   (void) pthread_cond_destroy(&COND_thread_cache);
@@ -2765,6 +2790,7 @@ static int init_thread_environment()
   (void) pthread_mutex_init(&LOCK_active_mi, MY_MUTEX_INIT_FAST);
   (void) pthread_mutex_init(&LOCK_global_system_variables, MY_MUTEX_INIT_FAST);
   (void) pthread_mutex_init(&LOCK_global_read_lock, MY_MUTEX_INIT_FAST);
+  (void) pthread_mutex_init(&LOCK_prepared_stmt_count, MY_MUTEX_INIT_FAST);
   (void) pthread_mutex_init(&LOCK_uuid_generator, MY_MUTEX_INIT_FAST);
 #ifdef HAVE_OPENSSL
   (void) pthread_mutex_init(&LOCK_des_key_file,MY_MUTEX_INIT_FAST);
@@ -4503,7 +4529,8 @@ enum options_mysqld
   OPT_MAX_BINLOG_CACHE_SIZE, OPT_MAX_BINLOG_SIZE,
   OPT_MAX_CONNECTIONS, OPT_MAX_CONNECT_ERRORS,
   OPT_MAX_DELAYED_THREADS, OPT_MAX_HEP_TABLE_SIZE,
-  OPT_MAX_JOIN_SIZE, OPT_MAX_RELAY_LOG_SIZE, OPT_MAX_SORT_LENGTH, 
+  OPT_MAX_JOIN_SIZE, OPT_MAX_PREPARED_STMT_COUNT,
+  OPT_MAX_RELAY_LOG_SIZE, OPT_MAX_SORT_LENGTH,
   OPT_MAX_SEEKS_FOR_KEY, OPT_MAX_TMP_TABLES, OPT_MAX_USER_CONNECTIONS,
   OPT_MAX_LENGTH_FOR_SORT_DATA,
   OPT_MAX_WRITE_LOCK_COUNT, OPT_BULK_INSERT_BUFFER_SIZE,
@@ -5638,6 +5665,10 @@ The minimum value for this variable is 4096.",
     (gptr*) &global_system_variables.max_length_for_sort_data,
     (gptr*) &max_system_variables.max_length_for_sort_data, 0, GET_ULONG,
     REQUIRED_ARG, 1024, 4, 8192*1024L, 0, 1, 0},
+  {"max_prepared_stmt_count", OPT_MAX_PREPARED_STMT_COUNT,
+   "Maximum numbrer of prepared statements in the server.",
+   (gptr*) &max_prepared_stmt_count, (gptr*) &max_prepared_stmt_count,
+   0, GET_ULONG, REQUIRED_ARG, 16382, 0, 1*1024*1024, 0, 1, 0},
   {"max_relay_log_size", OPT_MAX_RELAY_LOG_SIZE,
    "If non-zero: relay log will be rotated automatically when the size exceeds this value; if zero (the default): when the size exceeds max_binlog_size. 0 excepted, the minimum value for this variable is 4096.",
    (gptr*) &max_relay_log_size, (gptr*) &max_relay_log_size, 0, GET_ULONG,
diff --git a/sql/set_var.cc b/sql/set_var.cc
index 7be79ab59f0..47a4b26f010 100644
--- a/sql/set_var.cc
+++ b/sql/set_var.cc
@@ -120,6 +120,7 @@ static KEY_CACHE *create_key_cache(const char *name, uint length);
 void fix_sql_mode_var(THD *thd, enum_var_type type);
 static byte *get_error_count(THD *thd);
 static byte *get_warning_count(THD *thd);
+static byte *get_prepared_stmt_count(THD *thd);
 static byte *get_have_innodb(THD *thd);
 
 /*
@@ -257,6 +258,10 @@ sys_var_thd_ha_rows	sys_sql_max_join_size("sql_max_join_size",
 					      &SV::max_join_size,
 					      fix_max_join_size);
 #endif
+static sys_var_long_ptr_global
+sys_max_prepared_stmt_count("max_prepared_stmt_count",
+                            &max_prepared_stmt_count,
+                            &LOCK_prepared_stmt_count);
 sys_var_long_ptr	sys_max_relay_log_size("max_relay_log_size",
                                                &max_relay_log_size,
                                                fix_max_relay_log_size);
@@ -531,6 +536,9 @@ static sys_var_readonly		sys_warning_count("warning_count",
 						  OPT_SESSION,
 						  SHOW_LONG,
 						  get_warning_count);
+static sys_var_readonly	sys_prepared_stmt_count("prepared_stmt_count",
+                                                OPT_GLOBAL, SHOW_LONG,
+                                                get_prepared_stmt_count);
 
 /* alias for last_insert_id() to be compatible with Sybase */
 #ifdef HAVE_REPLICATION
@@ -635,6 +643,7 @@ sys_var *sys_variables[]=
   &sys_max_heap_table_size,
   &sys_max_join_size,
   &sys_max_length_for_sort_data,
+  &sys_max_prepared_stmt_count,
   &sys_max_relay_log_size,
   &sys_max_seeks_for_key,
   &sys_max_sort_length,
@@ -658,6 +667,7 @@ sys_var *sys_variables[]=
   &sys_optimizer_prune_level,
   &sys_optimizer_search_depth,
   &sys_preload_buff_size,
+  &sys_prepared_stmt_count,
   &sys_pseudo_thread_id,
   &sys_query_alloc_block_size,
   &sys_query_cache_size,
@@ -903,6 +913,8 @@ struct show_var_st init_vars[]= {
   {sys_max_join_size.name,	(char*) &sys_max_join_size,	    SHOW_SYS},
   {sys_max_length_for_sort_data.name, (char*) &sys_max_length_for_sort_data,
    SHOW_SYS},
+  {sys_max_prepared_stmt_count.name, (char*) &sys_max_prepared_stmt_count,
+    SHOW_SYS},
   {sys_max_relay_log_size.name, (char*) &sys_max_relay_log_size,    SHOW_SYS},
   {sys_max_seeks_for_key.name,  (char*) &sys_max_seeks_for_key,	    SHOW_SYS},
   {sys_max_sort_length.name,	(char*) &sys_max_sort_length,	    SHOW_SYS},
@@ -945,6 +957,7 @@ struct show_var_st init_vars[]= {
   {sys_optimizer_search_depth.name,(char*) &sys_optimizer_search_depth,
    SHOW_SYS},
   {"pid_file",                (char*) pidfile_name,                 SHOW_CHAR},
+  {sys_prepared_stmt_count.name, (char*) &sys_prepared_stmt_count, SHOW_SYS},
   {"port",                    (char*) &mysqld_port,                  SHOW_INT},
   {sys_preload_buff_size.name, (char*) &sys_preload_buff_size,      SHOW_SYS},
   {"protocol_version",        (char*) &protocol_version,            SHOW_INT},
@@ -1354,29 +1367,40 @@ static void fix_server_id(THD *thd, enum_var_type type)
   server_id_supplied = 1;
 }
 
-bool sys_var_long_ptr::check(THD *thd, set_var *var)
+
+sys_var_long_ptr::
+sys_var_long_ptr(const char *name_arg, ulong *value_ptr,
+                 sys_after_update_func after_update_arg)
+  :sys_var_long_ptr_global(name_arg, value_ptr,
+                           &LOCK_global_system_variables, after_update_arg)
+{}
+
+
+bool sys_var_long_ptr_global::check(THD *thd, set_var *var)
 {
   longlong v= var->value->val_int();
   var->save_result.ulonglong_value= v < 0 ? 0 : v;
   return 0;
 }
 
-bool sys_var_long_ptr::update(THD *thd, set_var *var)
+bool sys_var_long_ptr_global::update(THD *thd, set_var *var)
 {
   ulonglong tmp= var->save_result.ulonglong_value;
-  pthread_mutex_lock(&LOCK_global_system_variables);
+  pthread_mutex_lock(guard);
   if (option_limits)
     *value= (ulong) getopt_ull_limit_value(tmp, option_limits);
   else
     *value= (ulong) tmp;
-  pthread_mutex_unlock(&LOCK_global_system_variables);
+  pthread_mutex_unlock(guard);
   return 0;
 }
 
 
-void sys_var_long_ptr::set_default(THD *thd, enum_var_type type)
+void sys_var_long_ptr_global::set_default(THD *thd, enum_var_type type)
 {
+  pthread_mutex_lock(guard);
   *value= (ulong) option_limits->def_value;
+  pthread_mutex_unlock(guard);
 }
 
 
@@ -2826,6 +2850,14 @@ static byte *get_have_innodb(THD *thd)
 }
 
 
+static byte *get_prepared_stmt_count(THD *thd)
+{
+  pthread_mutex_lock(&LOCK_prepared_stmt_count);
+  thd->sys_var_tmp.ulong_value= prepared_stmt_count;
+  pthread_mutex_unlock(&LOCK_prepared_stmt_count);
+  return (byte*) &thd->sys_var_tmp.ulong_value;
+}
+
 /****************************************************************************
   Main handling of variables:
   - Initialisation
diff --git a/sql/set_var.h b/sql/set_var.h
index 046281ec7c5..646a578ca36 100644
--- a/sql/set_var.h
+++ b/sql/set_var.h
@@ -45,11 +45,7 @@ public:
   
   sys_after_update_func after_update;
   bool no_support_one_shot;
-  sys_var(const char *name_arg)
-    :name(name_arg), after_update(0)
-    , no_support_one_shot(1)
-    {}
-  sys_var(const char *name_arg,sys_after_update_func func)
+  sys_var(const char *name_arg, sys_after_update_func func= NULL)
     :name(name_arg), after_update(func)
     , no_support_one_shot(1)
   {}
@@ -74,15 +70,35 @@ public:
 };
 
 
-class sys_var_long_ptr :public sys_var
+/*
+  A base class for all variables that require its access to
+  be guarded with a mutex.
+*/
+
+class sys_var_global: public sys_var
+{
+protected:
+  pthread_mutex_t *guard;
+public:
+  sys_var_global(const char *name_arg, sys_after_update_func after_update_arg,
+                 pthread_mutex_t *guard_arg)
+    :sys_var(name_arg, after_update_arg), guard(guard_arg) {}
+};
+
+
+/*
+  A global-only ulong variable that requires its access to be
+  protected with a mutex.
+*/
+
+class sys_var_long_ptr_global: public sys_var_global
 {
 public:
   ulong *value;
-  sys_var_long_ptr(const char *name_arg, ulong *value_ptr)
-    :sys_var(name_arg),value(value_ptr) {}
-  sys_var_long_ptr(const char *name_arg, ulong *value_ptr,
-		   sys_after_update_func func)
-    :sys_var(name_arg,func), value(value_ptr) {}
+  sys_var_long_ptr_global(const char *name_arg, ulong *value_ptr,
+                        pthread_mutex_t *guard_arg,
+                        sys_after_update_func after_update_arg= NULL)
+    :sys_var_global(name_arg, after_update_arg, guard_arg), value(value_ptr) {}
   bool check(THD *thd, set_var *var);
   bool update(THD *thd, set_var *var);
   void set_default(THD *thd, enum_var_type type);
@@ -92,6 +108,18 @@ public:
 };
 
 
+/*
+  A global ulong variable that is protected by LOCK_global_system_variables
+*/
+
+class sys_var_long_ptr :public sys_var_long_ptr_global
+{
+public:
+  sys_var_long_ptr(const char *name_arg, ulong *value_ptr,
+                   sys_after_update_func after_update_arg= NULL);
+};
+
+
 class sys_var_ulonglong_ptr :public sys_var
 {
 public:
@@ -170,7 +198,7 @@ class sys_var_const_str :public sys_var
 public:
   char *value;					// Pointer to const value
   sys_var_const_str(const char *name_arg, const char *value_arg)
-    :sys_var(name_arg), value((char*) value_arg)
+    :sys_var(name_arg),value((char*) value_arg)
   {}
   bool check(THD *thd, set_var *var)
   {
@@ -217,10 +245,7 @@ public:
 class sys_var_thd :public sys_var
 {
 public:
-  sys_var_thd(const char *name_arg)
-    :sys_var(name_arg)
-  {}
-  sys_var_thd(const char *name_arg, sys_after_update_func func)
+  sys_var_thd(const char *name_arg, sys_after_update_func func= NULL)
     :sys_var(name_arg,func)
   {}
   bool check_type(enum_var_type type) { return 0; }
diff --git a/sql/sql_class.cc b/sql/sql_class.cc
index 5a6bbe01183..f0ab4b5c29c 100644
--- a/sql/sql_class.cc
+++ b/sql/sql_class.cc
@@ -1724,21 +1724,81 @@ Statement_map::Statement_map() :
 }
 
 
-int Statement_map::insert(Statement *statement)
+/*
+  Insert a new statement to the thread-local statement map.
+
+  DESCRIPTION
+    If there was an old statement with the same name, replace it with the
+    new one. Otherwise, check if max_prepared_stmt_count is not reached yet,
+    increase prepared_stmt_count, and insert the new statement. It's okay
+    to delete an old statement and fail to insert the new one.
+
+  POSTCONDITIONS
+    All named prepared statements are also present in names_hash.
+    Statement names in names_hash are unique.
+    The statement is added only if prepared_stmt_count < max_prepard_stmt_count
+    last_found_statement always points to a valid statement or is 0
+
+  RETURN VALUE
+    0  success
+    1  error: out of resources or max_prepared_stmt_count limit has been
+       reached. An error is sent to the client, the statement is deleted.
+*/
+
+int Statement_map::insert(THD *thd, Statement *statement)
 {
-  int res= my_hash_insert(&st_hash, (byte *) statement);
-  if (res)
-    return res;
+  if (my_hash_insert(&st_hash, (byte*) statement))
+  {
+    /*
+      Delete is needed only in case of an insert failure. In all other
+      cases hash_delete will also delete the statement.
+    */
+    delete statement;
+    my_error(ER_OUT_OF_RESOURCES, MYF(0));
+    goto err_st_hash;
+  }
   if (statement->name.str)
   {
-    if ((res= my_hash_insert(&names_hash, (byte*)statement)))
+    /*
+      If there is a statement with the same name, remove it. It is ok to
+      remove old and fail to insert new one at the same time.
+    */
+    Statement *old_stmt;
+    if ((old_stmt= find_by_name(&statement->name)))
+      erase(old_stmt);
+    if (my_hash_insert(&names_hash, (byte*) statement))
     {
-      hash_delete(&st_hash, (byte*)statement);
-      return res;
+      my_error(ER_OUT_OF_RESOURCES, MYF(0));
+      goto err_names_hash;
     }
   }
+  pthread_mutex_lock(&LOCK_prepared_stmt_count);
+  /*
+    We don't check that prepared_stmt_count is <= max_prepared_stmt_count
+    because we would like to allow to lower the total limit
+    of prepared statements below the current count. In that case
+    no new statements can be added until prepared_stmt_count drops below
+    the limit.
+  */
+  if (prepared_stmt_count >= max_prepared_stmt_count)
+  {
+    pthread_mutex_unlock(&LOCK_prepared_stmt_count);
+    my_error(ER_UNKNOWN_ERROR, MYF(0));
+    goto err_max;
+  }
+  prepared_stmt_count++;
+  pthread_mutex_unlock(&LOCK_prepared_stmt_count);
+
   last_found_statement= statement;
-  return res;
+  return 0;
+
+err_max:
+  if (statement->name.str)
+    hash_delete(&names_hash, (byte*) statement);
+err_names_hash:
+  hash_delete(&st_hash, (byte*) statement);
+err_st_hash:
+  return 1;
 }
 
 
@@ -1752,6 +1812,49 @@ void Statement_map::close_transient_cursors()
 }
 
 
+void Statement_map::erase(Statement *statement)
+{
+  if (statement == last_found_statement)
+    last_found_statement= 0;
+  if (statement->name.str)
+  {
+    hash_delete(&names_hash, (byte *) statement);
+  }
+  hash_delete(&st_hash, (byte *) statement);
+  pthread_mutex_lock(&LOCK_prepared_stmt_count);
+  DBUG_ASSERT(prepared_stmt_count > 0);
+  prepared_stmt_count--;
+  pthread_mutex_unlock(&LOCK_prepared_stmt_count);
+}
+
+
+void Statement_map::reset()
+{
+  /* Must be first, hash_free will reset st_hash.records */
+  pthread_mutex_lock(&LOCK_prepared_stmt_count);
+  DBUG_ASSERT(prepared_stmt_count >= st_hash.records);
+  prepared_stmt_count-= st_hash.records;
+  pthread_mutex_unlock(&LOCK_prepared_stmt_count);
+
+  my_hash_reset(&names_hash);
+  my_hash_reset(&st_hash);
+  last_found_statement= 0;
+}
+
+
+Statement_map::~Statement_map()
+{
+  /* Must go first, hash_free will reset st_hash.records */
+  pthread_mutex_lock(&LOCK_prepared_stmt_count);
+  DBUG_ASSERT(prepared_stmt_count >= st_hash.records);
+  prepared_stmt_count-= st_hash.records;
+  pthread_mutex_unlock(&LOCK_prepared_stmt_count);
+
+  hash_free(&names_hash);
+  hash_free(&st_hash);
+
+}
+
 bool select_dumpvar::send_data(List<Item> &items)
 {
   List_iterator_fast<Item_func_set_user_var> li(vars);
diff --git a/sql/sql_class.h b/sql/sql_class.h
index c1c1b9eceb3..dd645fcf75e 100644
--- a/sql/sql_class.h
+++ b/sql/sql_class.h
@@ -845,7 +845,7 @@ class Statement_map
 public:
   Statement_map();
 
-  int insert(Statement *statement);
+  int insert(THD *thd, Statement *statement);
 
   Statement *find_by_name(LEX_STRING *name)
   {
@@ -883,20 +883,10 @@ public:
     survive COMMIT or ROLLBACK. Currently all but MyISAM cursors are closed.
   */
   void close_transient_cursors();
+  void erase(Statement *statement);
   /* Erase all statements (calls Statement destructor) */
-  void reset()
-  {
-    my_hash_reset(&names_hash);
-    my_hash_reset(&st_hash);
-    transient_cursor_list.empty();
-    last_found_statement= 0;
-  }
-
-  void destroy()
-  {
-    hash_free(&names_hash);
-    hash_free(&st_hash);
-  }
+  void reset();
+  ~Statement_map();
 private:
   HASH st_hash;
   HASH names_hash;
@@ -1373,6 +1363,7 @@ public:
   {
     my_bool my_bool_value;
     long    long_value;
+    ulong   ulong_value;
   } sys_var_tmp;
   
   struct {
diff --git a/sql/sql_prepare.cc b/sql/sql_prepare.cc
index ac0687b488d..5a15e67da8e 100644
--- a/sql/sql_prepare.cc
+++ b/sql/sql_prepare.cc
@@ -1848,8 +1848,11 @@ void mysql_stmt_prepare(THD *thd, const char *packet, uint packet_length)
 
   if (thd->stmt_map.insert(stmt))
   {
-    delete stmt;
-    DBUG_VOID_RETURN;                           /* out of memory */
+    /*
+      The error is set in the insert. The statement itself
+      will be also deleted there (this is how the hash works).
+    */
+    DBUG_VOID_RETURN;
   }
 
   /* Reset warnings from previous command */
@@ -2026,9 +2029,14 @@ void mysql_sql_stmt_prepare(THD *thd)
     DBUG_VOID_RETURN;                           /* out of memory */
   }
 
-  if (stmt->set_name(name) || thd->stmt_map.insert(stmt))
+  if (thd->stmt_map.insert(stmt))
   {
-    delete stmt;
+    /* The statement is deleted and an error is set if insert fails */
+    DBUG_VOID_RETURN;
+  }
+  if (stmt->set_name(name))
+  {
+    thd->stmt_map.erase(stmt);
     DBUG_VOID_RETURN;
   }