diff --git a/sql/mysql_priv.h b/sql/mysql_priv.h
index 2d65e8395ea..b24fa4f5cbd 100644
--- a/sql/mysql_priv.h
+++ b/sql/mysql_priv.h
@@ -613,11 +613,10 @@ int mysqld_show_column_types(THD *thd);
 int mysqld_help (THD *thd, const char *text);
 
 /* sql_prepare.cc */
-class Prepared_statement;
-Prepared_statement *mysql_stmt_prepare(THD *thd, char *packet, 
-                                       uint packet_length, bool text_protocol=false);
+int mysql_stmt_prepare(THD *thd, char *packet, uint packet_length, 
+                       LEX_STRING *name=NULL);
 void mysql_stmt_execute(THD *thd, char *packet, uint packet_length);
-void mysql_sql_stmt_execute(THD *thd, Prepared_statement *stmt);
+void mysql_sql_stmt_execute(THD *thd, LEX_STRING *stmt_name);
 void mysql_stmt_free(THD *thd, char *packet);
 void mysql_stmt_reset(THD *thd, char *packet);
 void mysql_stmt_get_longdata(THD *thd, char *pos, ulong packet_length);
diff --git a/sql/sql_class.cc b/sql/sql_class.cc
index 49fa0455a30..87b6c49a4b7 100644
--- a/sql/sql_class.cc
+++ b/sql/sql_class.cc
@@ -78,24 +78,6 @@ extern "C" void free_user_var(user_var_entry *entry)
   my_free((char*) entry,MYF(0));
 }
 
-/****************************************************************************
-** SQL syntax names for Prepared Statements
-****************************************************************************/
-
-extern "C" byte *get_stmt_key(SQL_PREP_STMT_ENTRY *entry, uint *length,
-			     my_bool not_used __attribute__((unused)))
-{
-  *length=(uint) entry->name.length;
-  return (byte*) entry->name.str;
-}
-
-extern "C" void free_sql_stmt(SQL_PREP_STMT_ENTRY *entry)
-{
-  char *pos= (char*) entry+ALIGN_SIZE(sizeof(*entry));
-  my_free((char*) entry,MYF(0));
-}
-
-
 /****************************************************************************
 ** Thread specific functions
 ****************************************************************************/
@@ -178,9 +160,6 @@ THD::THD():user_time(0), current_statement(0), is_fatal_error(0),
   else
     bzero((char*) &user_var_events, sizeof(user_var_events));
   
-  hash_init(&sql_prepared_stmts, &my_charset_bin, USER_VARS_HASH_SIZE, 0, 0,
-	    (hash_get_key) get_stmt_key,
-	    (hash_free_key) free_sql_stmt,0);
   /* Protocol */
   protocol= &protocol_simple;			// Default protocol
   protocol_simple.init(this);
@@ -299,7 +278,6 @@ void THD::cleanup(void)
   my_free((char*) variables.datetime_format, MYF(MY_ALLOW_ZERO_PTR));
   delete_dynamic(&user_var_events);
   hash_free(&user_vars);
-  hash_free(&sql_prepared_stmts);
   if (global_read_lock)
     unlock_global_read_lock(this);
   if (ull)
@@ -1220,6 +1198,7 @@ Statement::Statement(THD *thd)
   query_length(0),
   free_list(0)
 {
+  name.str= NULL;
   init_sql_alloc(&mem_root,
                  thd->variables.query_alloc_block_size,
                  thd->variables.query_prealloc_size);
@@ -1303,17 +1282,52 @@ static void delete_statement_as_hash_key(void *key)
   delete (Statement *) key;
 }
 
+byte *get_stmt_name_hash_key(Statement *entry, uint *length,
+		             my_bool not_used __attribute__((unused)))
+{
+  *length=(uint) entry->name.length;
+  return (byte*) entry->name.str;
+}
+
 C_MODE_END
 
 Statement_map::Statement_map() :
   last_found_statement(0)
 {
-  enum { START_HASH_SIZE = 16 };
-  hash_init(&st_hash, default_charset_info, START_HASH_SIZE, 0, 0,
+  enum
+  {
+    START_STMT_HASH_SIZE = 16,
+    START_NAME_HASH_SIZE = 16
+  };
+  hash_init(&st_hash, default_charset_info, START_STMT_HASH_SIZE, 0, 0,
             get_statement_id_as_hash_key,
             delete_statement_as_hash_key, MYF(0));
+  hash_init(&names_hash, &my_charset_bin, START_NAME_HASH_SIZE, 0, 0,
+	    (hash_get_key) get_stmt_name_hash_key,
+	    NULL,MYF(0));
 }
 
+int Statement_map::insert(Statement *statement)
+{
+  int rc= my_hash_insert(&st_hash, (byte *) statement);
+  if (rc == 0)
+    last_found_statement= statement;
+  if (statement->name.str)
+  {
+    /*
+      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 ((rc= my_hash_insert(&names_hash, (byte*)statement)))
+      hash_delete(&st_hash, (byte*)statement);
+  }
+  return rc;
+}
+
+
 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 4eb86b20337..8ccfe3cddd5 100644
--- a/sql/sql_class.h
+++ b/sql/sql_class.h
@@ -456,6 +456,7 @@ public:
   */
   bool allow_sum_func;
 
+  LEX_STRING name; /* name for named prepared statements */
   LEX *lex;                                     // parse tree descriptor
   /*
     Points to the query associated with this statement. It's const, but
@@ -522,8 +523,14 @@ public:
 
 
 /*
-  Used to seek all existing statements in the connection
-  Deletes all statements in destructor.
+  Container for all statements created/used in a connection.
+  Statements in Statement_map have unique Statement::id (guaranteed by id
+  assignment in Statement::Statement)
+  Non-empty statement names are unique too: attempt to insert a new statement
+  with duplicate name causes older statement to be deleted
+  
+  Statements are auto-deleted when they are removed from the map and when the
+  map is deleted.
 */
 
 class Statement_map
@@ -531,12 +538,14 @@ class Statement_map
 public:
   Statement_map();
   
-  int insert(Statement *statement)
+  int insert(Statement *statement);
+
+  Statement *find_by_name(LEX_STRING *name)
   {
-    int rc= my_hash_insert(&st_hash, (byte *) statement);
-    if (rc == 0)
-      last_found_statement= statement;
-    return rc;
+    Statement *stmt;
+    stmt= (Statement*)hash_search(&names_hash, (byte*)name->str,
+                                  name->length);
+    return stmt;
   }
 
   Statement *find(ulong id)
@@ -550,15 +559,21 @@ public:
   {
     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);
   }
 
   ~Statement_map()
   {
     hash_free(&st_hash);
+    hash_free(&names_hash);
   }
 private:
   HASH st_hash;
+  HASH names_hash;
   Statement *last_found_statement;
 };
 
@@ -594,12 +609,6 @@ public:
   struct  system_variables variables;	// Changeable local variables
   pthread_mutex_t LOCK_delete;		// Locked before thd is deleted
 
-  /* 
-    statement_name -> (Statement*) map of statements prepared using SQL syntax.
-    Hash element is SQL_PREP_STMT_ENTRY.
-  */
-  HASH    sql_prepared_stmts;
-
   /* all prepared statements and cursors of this connection */
   Statement_map stmt_map; 
   /*
@@ -1276,14 +1285,6 @@ class user_var_entry
   DTCollation collation;
 };
 
-class Prepared_statement;
-/* Needed by THD::sql_prepared_stmts */
-typedef struct st_sql_prep_stmt_entry
-{
- public:
-  LEX_STRING name;
-  Prepared_statement *stmt;
-}SQL_PREP_STMT_ENTRY;
 
 /* Class for unique (removing of duplicates) */
 
diff --git a/sql/sql_parse.cc b/sql/sql_parse.cc
index 91bcc9e0495..bdf6ac747c3 100644
--- a/sql/sql_parse.cc
+++ b/sql/sql_parse.cc
@@ -1960,88 +1960,41 @@ mysql_execute_command(THD *thd)
     break;
   }
   case SQLCOM_PREPARE:
-  {    
-    char *stmt_name= lex->prepared_stmt_name.str;
-    uint name_len=   lex->prepared_stmt_name.length;
-    Prepared_statement *stmt;
-    SQL_PREP_STMT_ENTRY *entry;
-    DBUG_PRINT("info", ("PREPARE: %.*s FROM '%.*s' \n", name_len, stmt_name,
+  {   
+    DBUG_PRINT("info", ("PREPARE: %.*s FROM '%.*s' \n", 
+                        lex->prepared_stmt_name.length,
+                        lex->prepared_stmt_name.str,
                         lex->prepared_stmt_code.length,
                         lex->prepared_stmt_code.str));
-    if ((entry=(SQL_PREP_STMT_ENTRY*)hash_search(&thd->sql_prepared_stmts,
-                                                 (byte*)stmt_name, name_len)))
-    {
-      /* Free the statement with the same name and reuse hash entry */
-      thd->stmt_map.erase((Statement*)entry->stmt);
-    }
-    else
-    {
-      uint size=ALIGN_SIZE(sizeof(SQL_PREP_STMT_ENTRY))+name_len+1;
-      if (!hash_inited(&thd->sql_prepared_stmts) || 
-          !(entry= (SQL_PREP_STMT_ENTRY*)my_malloc(size,MYF(MY_WME))))
-      {
-        send_error(thd, ER_OUT_OF_RESOURCES);
-        break;
-      }
-      entry->name.str= (char*)entry + ALIGN_SIZE(sizeof(SQL_PREP_STMT_ENTRY));
-      entry->name.length= name_len;
-      memcpy(entry->name.str, stmt_name, name_len+1);
-      if (my_hash_insert(&thd->sql_prepared_stmts, (byte*)entry))
-      {
-        my_free((char*)entry,MYF(0));
-        send_error(thd, ER_OUT_OF_RESOURCES);
-        break;
-      }
-    }
-    /* Pretend this is a COM_PREPARE query so parser allows placeholders etc*/
     thd->command= COM_PREPARE;
-    /* 'length+1' is for alloc_query that strips the last character */
-    stmt= mysql_stmt_prepare(thd, lex->prepared_stmt_code.str,
-                             lex->prepared_stmt_code.length + 1, true);
-    if (stmt)
-    {
-      entry->stmt= stmt;
+    if (!mysql_stmt_prepare(thd, lex->prepared_stmt_code.str,
+                            lex->prepared_stmt_code.length + 1, 
+                            &lex->prepared_stmt_name))    
       send_ok(thd, 0L, 0L, "Statement prepared");
-    }
-    else
-      hash_delete(&thd->sql_prepared_stmts, (byte*)entry);
     break;
   }
   case SQLCOM_EXECUTE:
   {
-    char *stmt_name= lex->prepared_stmt_name.str;
-    uint name_len=   lex->prepared_stmt_name.length;
-    SQL_PREP_STMT_ENTRY *entry;
-    DBUG_PRINT("info", ("EXECUTE: %.*s\n", name_len, stmt_name));
-    
-    if (!(entry= (SQL_PREP_STMT_ENTRY*)hash_search(&thd->sql_prepared_stmts,
-                                                   (byte*)stmt_name, 
-                                                   name_len)))
-    {
-      send_error(thd, ER_UNKNOWN_STMT_HANDLER, "Undefined prepared statement");
-      lex->prepared_stmt_params.empty();
-      break;
-    }
-    mysql_sql_stmt_execute(thd, entry->stmt);
+    DBUG_PRINT("info", ("EXECUTE: %.*s\n", 
+                        lex->prepared_stmt_name.length,
+                        lex->prepared_stmt_name.str));
+    mysql_sql_stmt_execute(thd, &lex->prepared_stmt_name);
     lex->prepared_stmt_params.empty();
     break;
   }
   case SQLCOM_DEALLOCATE_PREPARE:
   {
-    char *stmt_name= lex->prepared_stmt_name.str;
-    uint name_len=   lex->prepared_stmt_name.length;
-    SQL_PREP_STMT_ENTRY *entry;
-    DBUG_PRINT("info", ("DEALLOCATE PREPARE: %.*s\n", name_len, stmt_name));
-    if (!(entry= (SQL_PREP_STMT_ENTRY*)hash_search(&thd->sql_prepared_stmts,
-                                                   (byte*)stmt_name, 
-                                                   name_len)))
+    Statement* stmt;
+    DBUG_PRINT("info", ("DEALLOCATE PREPARE: %.*s\n", 
+                        lex->prepared_stmt_name.length,
+                        lex->prepared_stmt_name.str));
+    if ((stmt= thd->stmt_map.find_by_name(&lex->prepared_stmt_name)))
     {
-      send_error(thd, ER_UNKNOWN_STMT_HANDLER, "Undefined prepared statement");
-      break;
+      thd->stmt_map.erase(stmt);
+      send_ok(thd);
     }
-    thd->stmt_map.erase((Statement*)entry->stmt);
-    hash_delete(&thd->sql_prepared_stmts, (byte*)entry);
-    send_ok(thd);
+    else
+      send_error(thd,ER_UNKNOWN_STMT_HANDLER,"Undefined prepared statement");
     break;
   }
   case SQLCOM_DO:
@@ -2259,9 +2212,9 @@ mysql_execute_command(THD *thd)
     tables= tables->next;
     // and from local list if it is not the same
     if (&lex->select_lex != lex->all_selects_list)
-      lex->select_lex.table_list.first= (gptr)create_table_local->next;
+      lex->select_lex.table_list.first= (gptr)create_table_local->next; 
     else
-      lex->select_lex.table_list.first= (gptr)tables;
+      lex->select_lex.table_list.first= (gptr)tables; 
     create_table->next= 0;
 
     ulong want_priv= ((lex->create_info.options & HA_LEX_CREATE_TMP_TABLE) ?
diff --git a/sql/sql_prepare.cc b/sql/sql_prepare.cc
index 51d75f07bd3..501d37e1383 100644
--- a/sql/sql_prepare.cc
+++ b/sql/sql_prepare.cc
@@ -107,6 +107,7 @@ public:
   virtual Statement::Type type() const;
 };
 
+static void execute_stmt(THD *thd, Prepared_statement *stmt);
 
 /******************************************************************************
   Implementation
@@ -636,7 +637,6 @@ static bool emb_insert_params_withlog(Prepared_statement *stmt)
 
 /* 
   Set prepared statement parameters from user variables.
-  Also replace '?' marks with values in thd->query if binary logging is on.
   SYNOPSIS
     insert_params_from_vars()
       stmt      Statement
@@ -682,11 +682,7 @@ static bool insert_params_from_vars(Prepared_statement *stmt,
       }
     }
     else
-    {
-      param->item_result_type= INT_RESULT;
-      param->maybe_null= param->null_value= 1;
-      param->value_is_set= 0;
-    }
+      param->maybe_null= param->null_value= param->value_is_set= 1;
   }
   DBUG_RETURN(0);
 }
@@ -704,6 +700,8 @@ static bool insert_params_from_vars_with_log(Prepared_statement *stmt,
   String str, query;
   const String *res;
   uint32 length= 0;
+  if (query.copy(stmt->query, stmt->query_length, default_charset_info))
+    DBUG_RETURN(1);
 
   for (Item_param **it= begin; it < end; ++it)
   {
@@ -734,9 +732,7 @@ static bool insert_params_from_vars_with_log(Prepared_statement *stmt,
     }
     else
     {
-      param->item_result_type= INT_RESULT;
-      param->maybe_null= param->null_value= 1;
-      param->value_is_set= 0;
+      param->maybe_null= param->null_value= param->value_is_set= 1;
       res= &my_null_string;
     }
 
@@ -1089,6 +1085,14 @@ static bool init_param_array(Prepared_statement *stmt)
 
 
 /*
+  SYNOPSIS
+    mysql_stmt_prepare()
+      packet         Prepared query 
+      packet_length  query length, with ignored trailing NULL or quote char.
+      name           NULL or statement name. For unnamed statements binary PS
+                     protocol is used, for named statmenents text protocol is 
+                     used.
+
   Parse the query and send the total number of parameters 
   and resultset metadata information back to client (if any), 
   without executing the query i.e. without any log/disk 
@@ -1100,11 +1104,11 @@ static bool init_param_array(Prepared_statement *stmt)
   list in lex->param_array, so that a fast and direct
   retrieval can be made without going through all field
   items.
-
+   
 */
 
-Prepared_statement *mysql_stmt_prepare(THD *thd, char *packet,
-                                       uint packet_length, bool text_protocol)
+int mysql_stmt_prepare(THD *thd, char *packet, uint packet_length,
+                       LEX_STRING *name)
 {
   LEX *lex;
   Prepared_statement *stmt= new Prepared_statement(thd);
@@ -1116,14 +1120,26 @@ Prepared_statement *mysql_stmt_prepare(THD *thd, char *packet,
   if (stmt == 0)
   {
     send_error(thd, ER_OUT_OF_RESOURCES);
-    DBUG_RETURN(NULL);
+    DBUG_RETURN(1);
+  }
+
+  if (name)
+  {
+    stmt->name.length= name->length;
+    if (!(stmt->name.str= my_memdup((byte*)name->str, name->length, 
+                                    MYF(MY_WME))))
+    {
+      delete stmt;
+      send_error(thd, ER_OUT_OF_RESOURCES);
+      DBUG_RETURN(1);
+    }
   }
 
   if (thd->stmt_map.insert(stmt))
   {
     delete stmt;
     send_error(thd, ER_OUT_OF_RESOURCES);
-    DBUG_RETURN(NULL);
+    DBUG_RETURN(1);
   }
 
   thd->stmt_backup.set_statement(thd);
@@ -1140,7 +1156,7 @@ Prepared_statement *mysql_stmt_prepare(THD *thd, char *packet,
     /* Statement map deletes statement on erase */
     thd->stmt_map.erase(stmt);
     send_error(thd, ER_OUT_OF_RESOURCES);
-    DBUG_RETURN(NULL);
+    DBUG_RETURN(1);
   }
 
   mysql_log.write(thd, COM_PREPARE, "%s", packet);
@@ -1152,7 +1168,7 @@ Prepared_statement *mysql_stmt_prepare(THD *thd, char *packet,
 
   error= yyparse((void *)thd) || thd->is_fatal_error ||
          init_param_array(stmt) ||
-         send_prepare_results(stmt, text_protocol);
+         send_prepare_results(stmt, test(name));
 
   /* restore to WAIT_PRIOR: QUERY_PRIOR is set inside alloc_query */
   if (!(specialflag & SPECIAL_NO_PRIOR))
@@ -1183,7 +1199,7 @@ Prepared_statement *mysql_stmt_prepare(THD *thd, char *packet,
       sl->prep_where= sl->where;
     }
   }
-  DBUG_RETURN(stmt);
+  DBUG_RETURN(!stmt);
 }
 
 /* Reinit statement before execution */
@@ -1236,6 +1252,7 @@ static void reset_stmt_for_execute(Prepared_statement *stmt)
   }
 }
 
+
 /*
   Executes previously prepared query.
   If there is any parameters, then replace markers with the data supplied
@@ -1267,11 +1284,6 @@ void mysql_stmt_execute(THD *thd, char *packet, uint packet_length)
     DBUG_VOID_RETURN;
   }
 
-  thd->stmt_backup.set_statement(thd);
-  thd->set_statement(stmt);
-
-  reset_stmt_for_execute(stmt);
-
 #ifndef EMBEDDED_LIBRARY
   if (stmt->param_count)
   {
@@ -1289,30 +1301,12 @@ void mysql_stmt_execute(THD *thd, char *packet, uint packet_length)
   if (stmt->param_count && stmt->set_params_data(stmt))
     goto set_params_data_err;
 #endif
-
-  if (!(specialflag & SPECIAL_NO_PRIOR))
-    my_pthread_setprio(pthread_self(),QUERY_PRIOR);
- 
-  /*
-    TODO:
-    Also, have checks on basic executions such as mysql_insert(), 
-    mysql_delete(), mysql_update() and mysql_select() to not to 
-    have re-check on setup_* and other things ..
-  */
   thd->protocol= &thd->protocol_prep;           // Switch to binary protocol
-  mysql_execute_command(thd);
+  execute_stmt(thd, stmt);
   thd->protocol= &thd->protocol_simple;         // Use normal protocol
-
-  if (!(specialflag & SPECIAL_NO_PRIOR))
-    my_pthread_setprio(pthread_self(), WAIT_PRIOR);
-
-  cleanup_items(stmt->free_list);
-  close_thread_tables(thd); // to close derived tables
-  thd->set_statement(&thd->stmt_backup);
   DBUG_VOID_RETURN;
 
 set_params_data_err:
-  thd->set_statement(&thd->stmt_backup);
   my_error(ER_WRONG_ARGUMENTS, MYF(0), "mysql_execute");
   send_error(thd);
   DBUG_VOID_RETURN;
@@ -1324,28 +1318,48 @@ set_params_data_err:
   lex->prepared_stmt_params and send result to the client using text protocol.
 */
 
-void mysql_sql_stmt_execute(THD *thd, Prepared_statement *stmt)
+void mysql_sql_stmt_execute(THD *thd, LEX_STRING *stmt_name)
 {
+  Prepared_statement *stmt;
   DBUG_ENTER("mysql_stmt_execute");
+    
+  if (!(stmt= (Prepared_statement*)thd->stmt_map.find_by_name(stmt_name)))
+  {
+      send_error(thd, ER_UNKNOWN_STMT_HANDLER, 
+                 "Undefined prepared statement");
+      DBUG_VOID_RETURN;
+  }
+
   if (stmt->param_count != thd->lex->prepared_stmt_params.elements)
   {
     my_error(ER_WRONG_ARGUMENTS, MYF(0), "mysql_execute");
     send_error(thd);
     DBUG_VOID_RETURN;
   }
-  thd->stmt_backup.set_statement(thd);
-  thd->set_statement(stmt);
-  reset_stmt_for_execute(stmt);
+  /* Item_param allows setting parameters in COM_EXECUTE only */
   thd->command= COM_EXECUTE;
 
-  if (stmt->set_params_from_vars(stmt, thd->stmt_backup.lex->
-                                        prepared_stmt_params))
+  if (stmt->set_params_from_vars(stmt, thd->lex->prepared_stmt_params))
   {
-    thd->set_statement(&thd->stmt_backup);
     my_error(ER_WRONG_ARGUMENTS, MYF(0), "mysql_execute");
     send_error(thd);
   }
 
+  execute_stmt(thd, stmt);
+  DBUG_VOID_RETURN;
+}
+
+/*
+  Execute prepared statement.
+  Caller must set parameter values and thd::protocol.
+*/
+static void execute_stmt(THD *thd, Prepared_statement *stmt)
+{
+  DBUG_ENTER("execute_stmt");
+  thd->stmt_backup.set_statement(thd);
+  thd->set_statement(stmt);
+  reset_stmt_for_execute(stmt);
+
   if (!(specialflag & SPECIAL_NO_PRIOR))
     my_pthread_setprio(pthread_self(),QUERY_PRIOR);
   mysql_execute_command(thd);
@@ -1359,6 +1373,7 @@ void mysql_sql_stmt_execute(THD *thd, Prepared_statement *stmt)
 }
 
 
+
 /*
     Reset a prepared statement, in case there was an error in send_longdata.
     Note: we don't send any reply to that command.
@@ -1522,6 +1537,8 @@ Prepared_statement::Prepared_statement(THD *thd_arg)
 Prepared_statement::~Prepared_statement()
 {
   free_items(free_list);
+  if (name.str)
+    my_free(name.str, MYF(0));
 }