diff --git a/include/thr_alarm.h b/include/thr_alarm.h
index 99812ed808c..806735b2d38 100644
--- a/include/thr_alarm.h
+++ b/include/thr_alarm.h
@@ -88,6 +88,7 @@ typedef struct st_alarm {
 
 extern uint thr_client_alarm;
 extern pthread_t alarm_thread;
+extern my_bool my_disable_thr_alarm;
 
 #define thr_alarm_init(A) (*(A))=0
 #define thr_alarm_in_use(A) (*(A)!= 0)
diff --git a/include/thr_lock.h b/include/thr_lock.h
index 08cc8bd5408..ec24bbabddd 100644
--- a/include/thr_lock.h
+++ b/include/thr_lock.h
@@ -123,8 +123,10 @@ typedef struct st_thr_lock_data {
   struct st_thr_lock *lock;
   pthread_cond_t *cond;
   void *status_param;			/* Param to status functions */
-  void *debug_print_param;
   enum thr_lock_type type;
+
+  enum thr_lock_type org_type;		/* Cache for MariaDB */
+  void *debug_print_param;              /* For error messages */
   uint priority;
 } THR_LOCK_DATA;
 
diff --git a/mysql-test/include/check_no_concurrent_insert.inc b/mysql-test/include/check_no_concurrent_insert.inc
index 6938c53fd16..f6a3d2052f5 100644
--- a/mysql-test/include/check_no_concurrent_insert.inc
+++ b/mysql-test/include/check_no_concurrent_insert.inc
@@ -43,7 +43,7 @@ connection default;
 # of our statement.
 let $wait_condition=
   select count(*) = 1 from information_schema.processlist
-  where state = "Locked" and info = "insert into $table (i) values (0)";
+  where state = "Table lock" and info = "insert into $table (i) values (0)";
 --source include/wait_condition.inc
 
 --disable_result_log
diff --git a/mysql-test/include/handler.inc b/mysql-test/include/handler.inc
index 6e7f53ba9b2..e8aeeeefcb2 100644
--- a/mysql-test/include/handler.inc
+++ b/mysql-test/include/handler.inc
@@ -77,8 +77,9 @@ handler t2 read a prev  limit 10;
 
 handler t2 read a>=(16) limit 4;
 handler t2 read a>=(16) limit 2,2;
+select * from t1 where a>=16 limit 2,2;
 handler t2 read a last  limit 3;
-
+handler t2 read a=(16) limit 1,3;
 handler t2 read a=(19);
 handler t2 read a=(19) where b="yyy";
 
@@ -105,6 +106,79 @@ eval alter table t1 engine = $engine_type;
 --error 1109
 handler t2 read first;
 
+handler t1 open;
+handler t1 read a=(16) limit 1,3;
+flush tables;
+handler t1 read a=(16) limit 1,3;
+handler t1 close;
+
+#
+# Test with prepared statements
+#
+
+handler t1 open;
+prepare stmt from 'handler t1 read a=(?) limit ?,?';
+set @a=16,@b=1,@c=100;
+execute stmt using @a,@b,@c;
+set @a=16,@b=2,@c=1;
+execute stmt using @a,@b,@c;
+set @a=16,@b=0,@c=2;
+execute stmt using @a,@b,@c;
+deallocate prepare stmt;
+
+prepare stmt from 'handler t1 read a next limit ?';
+handler t1 read a>=(11);
+set @a=3;
+execute stmt using @a;
+execute stmt using @a;
+execute stmt using @a;
+deallocate prepare stmt;
+
+prepare stmt from 'handler t1 read b prev limit ?';
+execute stmt using @a;
+execute stmt using @a;
+execute stmt using @a;
+execute stmt using @a;
+deallocate prepare stmt;
+
+prepare stmt from 'handler t1 read b=(?,?)';
+set @a=14, @b='aaa';
+execute stmt using @a,@b;
+set @a=14, @b='not found';
+execute stmt using @a,@b;
+deallocate prepare stmt;
+
+prepare stmt from 'handler t1 read b=(1+?) limit 10';
+set @a=15;
+execute stmt using @a;
+execute stmt using @a;
+deallocate prepare stmt;
+
+prepare stmt from 'handler t1 read a>=(?) where a < ? limit 5';
+set @a=15, @b=20;
+execute stmt using @a,@b;
+execute stmt using @a,@b;
+deallocate prepare stmt;
+
+prepare stmt from 'handler t1 read a=(?)';
+set @a=16;
+execute stmt using @a;
+alter table t1 add c int;
+--error 1109
+execute stmt using @a;
+deallocate prepare stmt;
+--error 1109
+handler t1 close;
+
+handler t1 open;
+prepare stmt from 'handler t1 read a=(?)';
+flush tables;
+set @a=16;
+--error ER_NEED_REPREPARE
+execute stmt using @a;
+deallocate prepare stmt;
+handler t1 close;
+
 #
 # DROP TABLE / ALTER TABLE
 #
diff --git a/mysql-test/r/handler_innodb.result b/mysql-test/r/handler_innodb.result
index 957fc30acef..c13bea72cf1 100644
--- a/mysql-test/r/handler_innodb.result
+++ b/mysql-test/r/handler_innodb.result
@@ -115,11 +115,18 @@ handler t2 read a>=(16) limit 2,2;
 a	b
 17	ddd
 18	eee
+select * from t1 where a>=16 limit 2,2;
+a	b
+17	ddd
+18	eee
 handler t2 read a last  limit 3;
 a	b
 22	iii
 21	hhh
 20	ggg
+handler t2 read a=(16) limit 1,3;
+a	b
+16	xxx
 handler t2 read a=(19);
 a	b
 19	fff
@@ -161,6 +168,128 @@ a	b
 alter table t1 engine = InnoDB;
 handler t2 read first;
 ERROR 42S02: Unknown table 't2' in HANDLER
+handler t1 open;
+handler t1 read a=(16) limit 1,3;
+a	b
+16	xxx
+flush tables;
+handler t1 read a=(16) limit 1,3;
+a	b
+16	xxx
+handler t1 close;
+handler t1 open;
+prepare stmt from 'handler t1 read a=(?) limit ?,?';
+set @a=16,@b=1,@c=100;
+execute stmt using @a,@b,@c;
+a	b
+16	xxx
+set @a=16,@b=2,@c=1;
+execute stmt using @a,@b,@c;
+a	b
+set @a=16,@b=0,@c=2;
+execute stmt using @a,@b,@c;
+a	b
+16	ccc
+16	xxx
+deallocate prepare stmt;
+prepare stmt from 'handler t1 read a next limit ?';
+handler t1 read a>=(11);
+a	b
+14	aaa
+set @a=3;
+execute stmt using @a;
+a	b
+15	bbb
+16	ccc
+16	xxx
+execute stmt using @a;
+a	b
+17	ddd
+18	eee
+19	fff
+execute stmt using @a;
+a	b
+19	yyy
+20	ggg
+21	hhh
+deallocate prepare stmt;
+prepare stmt from 'handler t1 read b prev limit ?';
+execute stmt using @a;
+a	b
+22	iii
+21	hhh
+20	ggg
+execute stmt using @a;
+a	b
+19	yyy
+19	fff
+18	eee
+execute stmt using @a;
+a	b
+17	ddd
+16	xxx
+16	ccc
+execute stmt using @a;
+a	b
+15	bbb
+14	aaa
+deallocate prepare stmt;
+prepare stmt from 'handler t1 read b=(?,?)';
+set @a=14, @b='aaa';
+execute stmt using @a,@b;
+a	b
+14	aaa
+set @a=14, @b='not found';
+execute stmt using @a,@b;
+a	b
+deallocate prepare stmt;
+prepare stmt from 'handler t1 read b=(1+?) limit 10';
+set @a=15;
+execute stmt using @a;
+a	b
+16	ccc
+16	xxx
+execute stmt using @a;
+a	b
+16	ccc
+16	xxx
+deallocate prepare stmt;
+prepare stmt from 'handler t1 read a>=(?) where a < ? limit 5';
+set @a=15, @b=20;
+execute stmt using @a,@b;
+a	b
+15	bbb
+16	ccc
+16	xxx
+17	ddd
+18	eee
+execute stmt using @a,@b;
+a	b
+15	bbb
+16	ccc
+16	xxx
+17	ddd
+18	eee
+deallocate prepare stmt;
+prepare stmt from 'handler t1 read a=(?)';
+set @a=16;
+execute stmt using @a;
+a	b
+16	ccc
+alter table t1 add c int;
+execute stmt using @a;
+ERROR 42S02: Unknown table 't1' in HANDLER
+deallocate prepare stmt;
+handler t1 close;
+ERROR 42S02: Unknown table 't1' in HANDLER
+handler t1 open;
+prepare stmt from 'handler t1 read a=(?)';
+flush tables;
+set @a=16;
+execute stmt using @a;
+ERROR HY000: Prepared statement needs to be re-prepared
+deallocate prepare stmt;
+handler t1 close;
 handler t1 open as t2;
 drop table t1;
 create table t1 (a int);
diff --git a/mysql-test/r/handler_myisam.result b/mysql-test/r/handler_myisam.result
index b20b8dbb138..420cb956d37 100644
--- a/mysql-test/r/handler_myisam.result
+++ b/mysql-test/r/handler_myisam.result
@@ -115,11 +115,18 @@ handler t2 read a>=(16) limit 2,2;
 a	b
 17	ddd
 18	eee
+select * from t1 where a>=16 limit 2,2;
+a	b
+17	ddd
+18	eee
 handler t2 read a last  limit 3;
 a	b
 22	iii
 21	hhh
 20	ggg
+handler t2 read a=(16) limit 1,3;
+a	b
+16	xxx
 handler t2 read a=(19);
 a	b
 19	fff
@@ -161,6 +168,128 @@ a	b
 alter table t1 engine = MyISAM;
 handler t2 read first;
 ERROR 42S02: Unknown table 't2' in HANDLER
+handler t1 open;
+handler t1 read a=(16) limit 1,3;
+a	b
+16	xxx
+flush tables;
+handler t1 read a=(16) limit 1,3;
+a	b
+16	xxx
+handler t1 close;
+handler t1 open;
+prepare stmt from 'handler t1 read a=(?) limit ?,?';
+set @a=16,@b=1,@c=100;
+execute stmt using @a,@b,@c;
+a	b
+16	xxx
+set @a=16,@b=2,@c=1;
+execute stmt using @a,@b,@c;
+a	b
+set @a=16,@b=0,@c=2;
+execute stmt using @a,@b,@c;
+a	b
+16	ccc
+16	xxx
+deallocate prepare stmt;
+prepare stmt from 'handler t1 read a next limit ?';
+handler t1 read a>=(11);
+a	b
+14	aaa
+set @a=3;
+execute stmt using @a;
+a	b
+15	bbb
+16	ccc
+16	xxx
+execute stmt using @a;
+a	b
+17	ddd
+18	eee
+19	fff
+execute stmt using @a;
+a	b
+19	yyy
+20	ggg
+21	hhh
+deallocate prepare stmt;
+prepare stmt from 'handler t1 read b prev limit ?';
+execute stmt using @a;
+a	b
+22	iii
+21	hhh
+20	ggg
+execute stmt using @a;
+a	b
+19	yyy
+19	fff
+18	eee
+execute stmt using @a;
+a	b
+17	ddd
+16	xxx
+16	ccc
+execute stmt using @a;
+a	b
+15	bbb
+14	aaa
+deallocate prepare stmt;
+prepare stmt from 'handler t1 read b=(?,?)';
+set @a=14, @b='aaa';
+execute stmt using @a,@b;
+a	b
+14	aaa
+set @a=14, @b='not found';
+execute stmt using @a,@b;
+a	b
+deallocate prepare stmt;
+prepare stmt from 'handler t1 read b=(1+?) limit 10';
+set @a=15;
+execute stmt using @a;
+a	b
+16	ccc
+16	xxx
+execute stmt using @a;
+a	b
+16	ccc
+16	xxx
+deallocate prepare stmt;
+prepare stmt from 'handler t1 read a>=(?) where a < ? limit 5';
+set @a=15, @b=20;
+execute stmt using @a,@b;
+a	b
+15	bbb
+16	ccc
+16	xxx
+17	ddd
+18	eee
+execute stmt using @a,@b;
+a	b
+15	bbb
+16	ccc
+16	xxx
+17	ddd
+18	eee
+deallocate prepare stmt;
+prepare stmt from 'handler t1 read a=(?)';
+set @a=16;
+execute stmt using @a;
+a	b
+16	ccc
+alter table t1 add c int;
+execute stmt using @a;
+ERROR 42S02: Unknown table 't1' in HANDLER
+deallocate prepare stmt;
+handler t1 close;
+ERROR 42S02: Unknown table 't1' in HANDLER
+handler t1 open;
+prepare stmt from 'handler t1 read a=(?)';
+flush tables;
+set @a=16;
+execute stmt using @a;
+ERROR HY000: Prepared statement needs to be re-prepared
+deallocate prepare stmt;
+handler t1 close;
 handler t1 open as t2;
 drop table t1;
 create table t1 (a int);
diff --git a/mysql-test/r/sp-threads.result b/mysql-test/r/sp-threads.result
index 953830ecc87..d974cfb9605 100644
--- a/mysql-test/r/sp-threads.result
+++ b/mysql-test/r/sp-threads.result
@@ -35,7 +35,7 @@ call bug9486();
 show processlist;
 Id	User	Host	db	Command	Time	State	Info
 #	root	localhost	test	Sleep	#		NULL
-#	root	localhost	test	Query	#	Locked	update t1, t2 set val= 1 where id1=id2
+#	root	localhost	test	Query	#	Table lock	update t1, t2 set val= 1 where id1=id2
 #	root	localhost	test	Query	#	NULL	show processlist
 #	root	localhost	test	Sleep	#		NULL
 unlock tables;
diff --git a/mysql-test/suite/binlog/t/binlog_stm_row.test b/mysql-test/suite/binlog/t/binlog_stm_row.test
index 29b0a69330d..47d9cbbbfb6 100644
--- a/mysql-test/suite/binlog/t/binlog_stm_row.test
+++ b/mysql-test/suite/binlog/t/binlog_stm_row.test
@@ -60,7 +60,7 @@ let $wait_condition=
 --echo # con1
 let $wait_condition=
   SELECT COUNT(*) = 1 FROM information_schema.processlist WHERE
-  state = "Locked" and info = "INSERT INTO t2 VALUES (3)";
+  state = "Table Lock" and info = "INSERT INTO t2 VALUES (3)";
 --source include/wait_condition.inc
 SELECT RELEASE_LOCK('Bug#34306');
 --connection con2
diff --git a/mysql-test/suite/funcs_1/datadict/processlist_val.inc b/mysql-test/suite/funcs_1/datadict/processlist_val.inc
index b1c1130cbdf..e06c5f081f5 100644
--- a/mysql-test/suite/funcs_1/datadict/processlist_val.inc
+++ b/mysql-test/suite/funcs_1/datadict/processlist_val.inc
@@ -368,13 +368,13 @@ echo
 ;
 connection default;
 echo
-# Poll till INFO is no more NULL and State = 'Locked'.
+# Poll till INFO is no more NULL and State = "Table Lock".
 ;
 let $wait_condition= SELECT COUNT(*) FROM INFORMATION_SCHEMA.PROCESSLIST
-        WHERE INFO IS NOT NULL AND STATE = 'Locked';
+        WHERE INFO IS NOT NULL AND STATE = "Table Lock";
 --source include/wait_condition.inc
 #
-# Expect to see the state 'Locked' for the third connection because the SELECT
+# Expect to see the state "Table Lock" for the third connection because the SELECT
 # collides with the WRITE TABLE LOCK.
 --replace_column 1 <ID> 3 <HOST_NAME> 6 <TIME> 9 <TIME_MS>
 SELECT * FROM INFORMATION_SCHEMA.PROCESSLIST;
@@ -423,10 +423,10 @@ echo
 ;
 connection default;
 echo
-# Poll till INFO is no more NULL and State = 'Locked'.
+# Poll till INFO is no more NULL and State = "Table Lock".
 ;
 let $wait_condition= SELECT COUNT(*) FROM INFORMATION_SCHEMA.PROCESSLIST
-        WHERE INFO IS NOT NULL AND STATE = 'Locked';
+        WHERE INFO IS NOT NULL AND STATE = "Table Lock";
 --source include/wait_condition.inc
 echo
 # Expect result:
diff --git a/mysql-test/suite/pbxt/t/lock_multi.test b/mysql-test/suite/pbxt/t/lock_multi.test
index 1104620b6b5..f90e31699c8 100644
--- a/mysql-test/suite/pbxt/t/lock_multi.test
+++ b/mysql-test/suite/pbxt/t/lock_multi.test
@@ -48,7 +48,7 @@ insert t1 select * from t2;
 connection locker;
 let $wait_condition=
   select count(*) = 1 from information_schema.processlist
-  where state = "Locked" and info = "insert t1 select * from t2";
+  where state = "Table Lock" and info = "insert t1 select * from t2";
 --source include/wait_condition.inc
 drop table t2;
 connection reader;
@@ -72,7 +72,7 @@ connection locker;
 # Sleep a bit till the insert of connection reader is in work and hangs
 let $wait_condition=
   select count(*) = 1 from information_schema.processlist
-  where state = "Locked" and info = "insert t1 select * from t2";
+  where state = "Table Lock" and info = "insert t1 select * from t2";
 --source include/wait_condition.inc
 drop table t2;
 connection reader;
@@ -251,7 +251,7 @@ connection reader;
 # Wait till connection writer is blocked
 let $wait_condition=
   select count(*) = 1 from information_schema.processlist
-  where state = "Locked" and info = "alter table t1 auto_increment=0";
+  where state = "Table Lock" and info = "alter table t1 auto_increment=0";
 --source include/wait_condition.inc
 send
 alter table t1 auto_increment=0;
@@ -259,7 +259,7 @@ connection locker;
 # Wait till connection reader is blocked
 let $wait_condition=
   select count(*) = 2 from information_schema.processlist
-  where state = "Locked" and info = "alter table t1 auto_increment=0";
+  where state = "Table Lock" and info = "alter table t1 auto_increment=0";
 --source include/wait_condition.inc
 unlock tables;
 connection writer;
@@ -414,16 +414,16 @@ update t1 set i= 10;
 connection reader;
 let $wait_condition=
   select count(*) = 1 from information_schema.processlist
-  where state = "Locked" and info = "update t1 set i= 10";
+  where state = "Table Lock" and info = "update t1 set i= 10";
 --source include/wait_condition.inc
 send
 select * from t1;
 connection default;
 let $wait_condition=
   select count(*) = 1 from information_schema.processlist
-  where state = "Locked" and info = "select * from t1";
+  where state = "Table Lock" and info = "select * from t1";
 --source include/wait_condition.inc
-let $ID= `select id from information_schema.processlist where state = "Locked" and info = "update t1 set i= 10"`;
+let $ID= `select id from information_schema.processlist where state = "Table Lock" and info = "update t1 set i= 10"`;
 --replace_result $ID ID
 eval kill query $ID;
 connection reader;
@@ -557,7 +557,7 @@ connection waiter;
 connection default;
 let $wait_condition=
   select count(*) = 1 from information_schema.processlist
-  where state = "Locked" and info = "insert into t1 values(1)";
+  where state = "Table Lock" and info = "insert into t1 values(1)";
 --source include/wait_condition.inc
 let $tlwb= `show status like 'Table_locks_waited'`;
 unlock tables;
diff --git a/mysql-test/suite/sys_vars/r/concurrent_insert_func.result b/mysql-test/suite/sys_vars/r/concurrent_insert_func.result
index 774775a8287..64a14473216 100644
--- a/mysql-test/suite/sys_vars/r/concurrent_insert_func.result
+++ b/mysql-test/suite/sys_vars/r/concurrent_insert_func.result
@@ -37,9 +37,9 @@ INSERT INTO t1(name) VALUES('Record_7');
 connection default;
 ## show processlist info and state ##
 SELECT state,info FROM INFORMATION_SCHEMA.PROCESSLIST
-WHERE state= "Locked" AND info LIKE "INSERT INTO t1%";
+WHERE state= "Table Lock" AND info LIKE "INSERT INTO t1%";
 state	info
-Locked	INSERT INTO t1(name) VALUES('Record_7')
+Table lock	INSERT INTO t1(name) VALUES('Record_7')
 ## table contents befor UNLOCK ##
 SELECT * FROM t1;
 name
diff --git a/mysql-test/suite/sys_vars/t/concurrent_insert_func.test b/mysql-test/suite/sys_vars/t/concurrent_insert_func.test
index 1a600ffd7f6..f7bd7bce39a 100644
--- a/mysql-test/suite/sys_vars/t/concurrent_insert_func.test
+++ b/mysql-test/suite/sys_vars/t/concurrent_insert_func.test
@@ -98,12 +98,12 @@ INSERT INTO t1(name) VALUES('Record_7');
 connection default;
 # wait until INSERT will be locked (low performance)
 let $wait_condition= SELECT COUNT(*) = 1 FROM INFORMATION_SCHEMA.PROCESSLIST
-                     WHERE state= "Locked" AND info LIKE "INSERT INTO t1%";
+                     WHERE state= "Table Lock" AND info LIKE "INSERT INTO t1%";
 --source include/wait_condition.inc
 
 --echo ## show processlist info and state ##
 SELECT state,info FROM INFORMATION_SCHEMA.PROCESSLIST
-WHERE state= "Locked" AND info LIKE "INSERT INTO t1%";
+WHERE state= "Table Lock" AND info LIKE "INSERT INTO t1%";
 --echo ## table contents befor UNLOCK ##
 SELECT * FROM t1;
 UNLOCK TABLES;
diff --git a/mysql-test/suite/sys_vars/t/delayed_insert_limit_func.test b/mysql-test/suite/sys_vars/t/delayed_insert_limit_func.test
index 61f7d801a1a..3de8d432d1a 100644
--- a/mysql-test/suite/sys_vars/t/delayed_insert_limit_func.test
+++ b/mysql-test/suite/sys_vars/t/delayed_insert_limit_func.test
@@ -122,7 +122,7 @@ connection default;
 --echo ** Wait till con0 is blocked **
 let $wait_condition=
    SELECT COUNT(*) = 1 FROM information_schema.processlist
-   WHERE state = 'Locked' AND info = '$my_select';
+   WHERE state = "Table Lock" AND info = '$my_select';
 --source include/wait_condition.inc
 UNLOCK TABLES;
 
@@ -207,7 +207,7 @@ connection default;
 --echo ** Wait till con0 is blocked **
 let $wait_condition=
    SELECT COUNT(*) = 1 FROM information_schema.processlist
-   WHERE state = 'Locked' AND info = '$my_select';
+   WHERE state = "Table Lock" AND info = '$my_select';
 --source include/wait_condition.inc
 UNLOCK TABLES;
 
diff --git a/mysql-test/suite/sys_vars/t/query_cache_wlock_invalidate_func.test b/mysql-test/suite/sys_vars/t/query_cache_wlock_invalidate_func.test
index e5ced59d175..baa490986e2 100644
--- a/mysql-test/suite/sys_vars/t/query_cache_wlock_invalidate_func.test
+++ b/mysql-test/suite/sys_vars/t/query_cache_wlock_invalidate_func.test
@@ -139,7 +139,7 @@ send SELECT * FROM t1;
 connection con0;
 
 --echo wait until table is locked
-let $wait_condition= SELECT count(*) > 0 FROM information_schema.processlist WHERE state= 'Locked';
+let $wait_condition= SELECT count(*) > 0 FROM information_schema.processlist WHERE state= "Table Lock";
 --source include/wait_condition.inc
 UNLOCK TABLES;
 
diff --git a/mysql-test/suite/sys_vars/t/sql_low_priority_updates_func.test b/mysql-test/suite/sys_vars/t/sql_low_priority_updates_func.test
index 5e0314c25ae..6a4cb2b664c 100644
--- a/mysql-test/suite/sys_vars/t/sql_low_priority_updates_func.test
+++ b/mysql-test/suite/sys_vars/t/sql_low_priority_updates_func.test
@@ -85,7 +85,7 @@ delimiter ;|
 --echo ** Connection con0 **
 connection con0;
 
-let $wait_condition = SELECT COUNT(*) > 0 FROM information_schema.processlist WHERE state='Locked' AND info LIKE  'UPDATE t1 SET a = CONCAT(a,"-updated")';
+let $wait_condition = SELECT COUNT(*) > 0 FROM information_schema.processlist WHERE state="Table Lock" AND info LIKE  'UPDATE t1 SET a = CONCAT(a,"-updated")';
 --source include/wait_condition.inc
 
 --echo ** Asynchronous Execution **
@@ -101,7 +101,7 @@ delimiter ;|
 --echo ** Connection default **
 connection default;
 
-let $wait_condition= SELECT count(*) = 2 FROM information_schema.processlist WHERE state LIKE 'Locked';
+let $wait_condition= SELECT count(*) = 2 FROM information_schema.processlist WHERE state LIKE "Table Lock";
 --source include/wait_condition.inc
 UNLOCK TABLES;
 
@@ -156,7 +156,7 @@ delimiter ;|
 --echo ** Connection con0 **
 connection con0;
 
-let $wait_condition = SELECT COUNT(*) > 0 FROM information_schema.processlist WHERE state='Locked' AND info LIKE  'UPDATE t1 SET a = CONCAT(a,"-updated")';
+let $wait_condition = SELECT COUNT(*) > 0 FROM information_schema.processlist WHERE state="Table Lock" AND info LIKE  'UPDATE t1 SET a = CONCAT(a,"-updated")';
 --source include/wait_condition.inc
 
 --echo ** Asynchronous Execution **
@@ -172,7 +172,7 @@ delimiter ;|
 --echo ** Connection default **
 connection default;
 
-let $wait_condition= SELECT count(*) = 2 FROM information_schema.processlist WHERE state LIKE 'Locked';
+let $wait_condition= SELECT count(*) = 2 FROM information_schema.processlist WHERE state LIKE "Table Lock";
 --source include/wait_condition.inc
 UNLOCK TABLES;
 
diff --git a/mysql-test/t/insert_notembedded.test b/mysql-test/t/insert_notembedded.test
index 24040f9c310..baf6a9a28a3 100644
--- a/mysql-test/t/insert_notembedded.test
+++ b/mysql-test/t/insert_notembedded.test
@@ -174,7 +174,7 @@ connection default;
 # we must wait till the insert opens and locks the table
 let $wait_condition=
   select count(*) = 1 from information_schema.processlist
-  where state = "Locked" and id = $ID;
+  where state = "Table lock" and id = $ID;
 --source include/wait_condition.inc
 connect (select,localhost,root,,);
 --echo connection: select
diff --git a/mysql-test/t/lock_multi.test b/mysql-test/t/lock_multi.test
index 4df1a0f3478..dc75c994a45 100644
--- a/mysql-test/t/lock_multi.test
+++ b/mysql-test/t/lock_multi.test
@@ -24,7 +24,7 @@ connection reader;
 # Sleep a bit till the update of connection writer is in work and hangs
 let $wait_condition=
   select count(*) = 1 from information_schema.processlist
-  where state = "Locked" and info = "update low_priority t1 set n = 4";
+  where state = "Table lock" and info = "update low_priority t1 set n = 4";
 --source include/wait_condition.inc
 send
 select n from t1;
@@ -32,7 +32,7 @@ connection locker;
 # Sleep a bit till the select of connection reader is in work and hangs
 let $wait_condition=
   select count(*) = 1 from information_schema.processlist
-  where state = "Locked" and info = "select n from t1";
+  where state = "Table lock" and info = "select n from t1";
 --source include/wait_condition.inc
 unlock tables;
 connection writer;
@@ -52,7 +52,7 @@ connection reader;
 # Sleep a bit till the update of connection writer is in work and hangs
 let $wait_condition=
   select count(*) = 1 from information_schema.processlist
-  where state = "Locked" and info = "update low_priority t1 set n = 4";
+  where state = "Table Lock" and info = "update low_priority t1 set n = 4";
 --source include/wait_condition.inc
 select n from t1;
 connection locker;
@@ -96,7 +96,7 @@ insert t1 select * from t2;
 connection locker;
 let $wait_condition=
   select count(*) = 1 from information_schema.processlist
-  where state = "Locked" and info = "insert t1 select * from t2";
+  where state = "Table Lock" and info = "insert t1 select * from t2";
 --source include/wait_condition.inc
 drop table t2;
 connection reader;
@@ -120,7 +120,7 @@ connection locker;
 # Sleep a bit till the insert of connection reader is in work and hangs
 let $wait_condition=
   select count(*) = 1 from information_schema.processlist
-  where state = "Locked" and info = "insert t1 select * from t2";
+  where state = "Table Lock" and info = "insert t1 select * from t2";
 --source include/wait_condition.inc
 drop table t2;
 connection reader;
@@ -299,7 +299,7 @@ connection reader;
 # Wait till connection writer is blocked
 let $wait_condition=
   select count(*) = 1 from information_schema.processlist
-  where state = "Locked" and info = "alter table t1 auto_increment=0";
+  where state = "Table Lock" and info = "alter table t1 auto_increment=0";
 --source include/wait_condition.inc
 send
 alter table t1 auto_increment=0;
@@ -307,7 +307,7 @@ connection locker;
 # Wait till connection reader is blocked
 let $wait_condition=
   select count(*) = 2 from information_schema.processlist
-  where state = "Locked" and info = "alter table t1 auto_increment=0";
+  where state = "Table Lock" and info = "alter table t1 auto_increment=0";
 --source include/wait_condition.inc
 unlock tables;
 connection writer;
@@ -462,16 +462,16 @@ update t1 set i= 10;
 connection reader;
 let $wait_condition=
   select count(*) = 1 from information_schema.processlist
-  where state = "Locked" and info = "update t1 set i= 10";
+  where state = "Table Lock" and info = "update t1 set i= 10";
 --source include/wait_condition.inc
 send
 select * from t1;
 connection default;
 let $wait_condition=
   select count(*) = 1 from information_schema.processlist
-  where state = "Locked" and info = "select * from t1";
+  where state = "Table Lock" and info = "select * from t1";
 --source include/wait_condition.inc
-let $ID= `select id from information_schema.processlist where state = "Locked" and info = "update t1 set i= 10"`;
+let $ID= `select id from information_schema.processlist where state = "Table Lock" and info = "update t1 set i= 10"`;
 --replace_result $ID ID
 eval kill query $ID;
 connection reader;
@@ -622,7 +622,7 @@ connection waiter;
 connection default;
 let $wait_condition=
   select count(*) = 1 from information_schema.processlist
-  where state = "Locked" and info = "insert into t1 values(1)";
+  where state = "Table Lock" and info = "insert into t1 values(1)";
 --source include/wait_condition.inc
 let $tlwb= `show status like 'Table_locks_waited'`;
 unlock tables;
diff --git a/mysql-test/t/merge-big.test b/mysql-test/t/merge-big.test
index d39c2973688..10a41b609a0 100644
--- a/mysql-test/t/merge-big.test
+++ b/mysql-test/t/merge-big.test
@@ -52,7 +52,7 @@ connection default;
 #--sleep 8
 #SELECT ID,STATE,INFO FROM INFORMATION_SCHEMA.PROCESSLIST;
 let $wait_condition= SELECT 1 FROM INFORMATION_SCHEMA.PROCESSLIST
-    WHERE ID = $con1_id AND STATE = 'Locked';
+    WHERE ID = $con1_id AND STATE = "Table Lock";
 --source include/wait_condition.inc
 #SELECT NOW();
 --echo # Kick INSERT out of thr_multi_lock().
diff --git a/mysql-test/t/multi_update.test b/mysql-test/t/multi_update.test
index d405e4e5e3a..e5c7a415170 100644
--- a/mysql-test/t/multi_update.test
+++ b/mysql-test/t/multi_update.test
@@ -498,9 +498,9 @@ send alter table t1 add column c int default 100 after a;
 connect (updater,localhost,root,,test);
 connection updater;
 # Wait till "alter table t1 ..." of session changer is in work.
-# = There is one session is in state "Locked".
+# = There is one session is in state "Table Lock".
 let $wait_condition= select count(*)= 1 from information_schema.processlist
-                     where state= 'Locked';
+                     where state= "Table Lock";
 --source include/wait_condition.inc
 send update t1, v1 set t1.b=t1.a+t1.b+v1.b where t1.a=v1.a;
 
@@ -509,9 +509,9 @@ connection locker;
 # - "alter table t1 ..." of session changer and
 # - "update t1, v1 ..." of session updater
 # are in work.
-# = There are two session is in state "Locked".
+# = There are two session is in state "Table Lock".
 let $wait_condition= select count(*)= 2 from information_schema.processlist
-                     where state= 'Locked';
+                     where state= "Table Lock";
 --source include/wait_condition.inc
 unlock tables;
 
diff --git a/mysql-test/t/query_cache_28249.test b/mysql-test/t/query_cache_28249.test
index 390a1ce6e3d..fd283aa69fb 100644
--- a/mysql-test/t/query_cache_28249.test
+++ b/mysql-test/t/query_cache_28249.test
@@ -64,12 +64,12 @@ connection user3;
 # The values marked with 'X' must be reached.
 --echo # Poll till the select of connection user1 is blocked by the write lock on t1.
 let $wait_condition= SELECT COUNT(*) = 1 FROM information_schema.processlist
-WHERE state = 'Locked'
+WHERE state = "Table Lock"
   AND info = '$select_for_qc';
 --source include/wait_condition.inc
 eval
 SELECT user,command,state,info FROM information_schema.processlist
-WHERE state = 'Locked'
+WHERE state = "Table Lock"
   AND info = '$select_for_qc';
 INSERT INTO t1 VALUES (4);
 
diff --git a/mysql-test/t/sp_notembedded.test b/mysql-test/t/sp_notembedded.test
index e4b6a5deefe..9b34a23bcc9 100644
--- a/mysql-test/t/sp_notembedded.test
+++ b/mysql-test/t/sp_notembedded.test
@@ -276,7 +276,7 @@ set session low_priority_updates=on;
 connection rl_wait;
 let $wait_condition=
   select count(*) = 1 from information_schema.processlist
-  where state = "Locked" and
+  where state = "Table lock" and
   info = "update t1 set value='updated' where value='old'";
 --source include/wait_condition.inc
 
diff --git a/mysql-test/t/sp_sync.test b/mysql-test/t/sp_sync.test
index f9dae17b039..e7f952162ee 100644
--- a/mysql-test/t/sp_sync.test
+++ b/mysql-test/t/sp_sync.test
@@ -34,7 +34,7 @@ SET DEBUG_SYNC = 'multi_update_reopen_tables SIGNAL parked WAIT_FOR go';
 
 connection con1;
 let $wait_condition= SELECT 1 FROM information_schema.processlist WHERE ID = $ID AND
-state = "Locked";
+state = "Table lock";
 --source include/wait_condition.inc
 DROP TABLE t1, t2;
 SET DEBUG_SYNC = 'now WAIT_FOR parked';
diff --git a/mysql-test/t/status.test b/mysql-test/t/status.test
index 5da210f5a69..505df0fe8dc 100644
--- a/mysql-test/t/status.test
+++ b/mysql-test/t/status.test
@@ -58,7 +58,7 @@ let $ID= `select connection_id()`;
 connection con2;
 --echo # Switched to connection: con2
 # wait for the other query to start executing
-let $wait_condition= select 1 from INFORMATION_SCHEMA.PROCESSLIST where ID = $ID and STATE = "Locked";
+let $wait_condition= select 1 from INFORMATION_SCHEMA.PROCESSLIST where ID = $ID and STATE = "Table Lock";
 --source include/wait_condition.inc
 unlock tables;
 
diff --git a/mysql-test/t/trigger_notembedded.test b/mysql-test/t/trigger_notembedded.test
index f595f195dcc..16e9b2361f9 100644
--- a/mysql-test/t/trigger_notembedded.test
+++ b/mysql-test/t/trigger_notembedded.test
@@ -916,7 +916,7 @@ INSERT INTO t1 VALUES (5);
 CONNECTION rl_contender;
 # Wait until wl_acquirer is waiting for the read lock on t2 to be released.
 let $wait_condition=
-  SELECT STATE = 'Locked' FROM INFORMATION_SCHEMA.PROCESSLIST
+  SELECT STATE = "Table Lock" FROM INFORMATION_SCHEMA.PROCESSLIST
    WHERE ID = $wl_acquirer_thread_id;
 --source include/wait_condition.inc
 # must not "see" the row inserted by the INSERT (as it must run before the
diff --git a/mysys/thr_alarm.c b/mysys/thr_alarm.c
index 15c24facac0..98c746af8ca 100644
--- a/mysys/thr_alarm.c
+++ b/mysys/thr_alarm.c
@@ -36,7 +36,7 @@
 
 uint thr_client_alarm;
 static int alarm_aborted=1;			/* No alarm thread */
-my_bool thr_alarm_inited= 0;
+my_bool thr_alarm_inited= 0, my_disable_thr_alarm= 0;
 volatile my_bool alarm_thread_running= 0;
 time_t next_alarm_expire_time= ~ (time_t) 0;
 static sig_handler process_alarm_part2(int sig);
@@ -173,6 +173,21 @@ my_bool thr_alarm(thr_alarm_t *alrm, uint sec, ALARM *alarm_data)
   DBUG_ENTER("thr_alarm");
   DBUG_PRINT("enter",("thread: %s  sec: %d",my_thread_name(),sec));
 
+  if (my_disable_thr_alarm)
+  {
+    (*alrm)= &alarm_data->alarmed;
+    alarm_data->alarmed= 1;                 /* Abort if interrupted */
+    DBUG_RETURN(0);
+  }
+
+  if (unlikely(alarm_aborted))
+  {					/* No signal thread */
+    DBUG_PRINT("info", ("alarm aborted"));
+    if (alarm_aborted > 0)
+      goto abort_no_unlock;
+    sec= 1;					/* Abort mode */
+  }
+
   now= my_time(0);
   if (!alarm_data)
   {
@@ -190,13 +205,6 @@ my_bool thr_alarm(thr_alarm_t *alrm, uint sec, ALARM *alarm_data)
 
   one_signal_hand_sigmask(SIG_BLOCK,&full_signal_set,&old_mask);
   pthread_mutex_lock(&LOCK_alarm);        /* Lock from threads & alarms */
-  if (unlikely(alarm_aborted))
-  {					/* No signal thread */
-    DBUG_PRINT("info", ("alarm aborted"));
-    if (alarm_aborted > 0)
-      goto abort;
-    sec= 1;					/* Abort mode */
-  }
   if (alarm_queue.elements >= max_used_alarms)
   {
     if (alarm_queue.elements == alarm_queue.max_elements)
@@ -251,6 +259,8 @@ void thr_end_alarm(thr_alarm_t *alarmed)
 #endif
   DBUG_ENTER("thr_end_alarm");
 
+  if (my_disable_thr_alarm)
+    DBUG_VOID_RETURN;
   one_signal_hand_sigmask(SIG_BLOCK,&full_signal_set,&old_mask);
   alarm_data= (ALARM*) ((uchar*) *alarmed - offsetof(ALARM,alarmed));
   pthread_mutex_lock(&LOCK_alarm);
diff --git a/mysys/thr_lock.c b/mysys/thr_lock.c
index 341b2f0058e..9d1b6f04b7a 100644
--- a/mysys/thr_lock.c
+++ b/mysys/thr_lock.c
@@ -230,14 +230,16 @@ static void check_locks(THR_LOCK *lock, const char *where,
 
     if (found_errors < MAX_FOUND_ERRORS)
     {
-      uint count=0;
+      uint count=0, count2= 0;
       THR_LOCK_DATA *data;
       for (data=lock->read.data ; data ; data=data->next)
       {
+        count2++;
 	if (data->type == TL_READ_NO_INSERT)
 	  count++;
         /* Protect against infinite loop. */
-        DBUG_ASSERT(count <= lock->read_no_write_count);
+        DBUG_ASSERT(count <= lock->read_no_write_count &&
+                    count2 <= MAX_LOCKS);
       }
       if (count != lock->read_no_write_count)
       {
@@ -288,7 +290,10 @@ static void check_locks(THR_LOCK *lock, const char *where,
         if (lock->write.data->type == TL_WRITE_CONCURRENT_INSERT)
         {
           THR_LOCK_DATA *data;
-          for (data=lock->write.data->next ; data ; data=data->next)
+          uint count= 0;
+          for (data=lock->write.data->next;
+               data && count < MAX_LOCKS;
+               data=data->next)
           {
             if (data->type != TL_WRITE_CONCURRENT_INSERT)
             {
diff --git a/scripts/mysql_install_db.sh b/scripts/mysql_install_db.sh
index 2d3775afb3a..f863ddd749e 100644
--- a/scripts/mysql_install_db.sh
+++ b/scripts/mysql_install_db.sh
@@ -398,9 +398,14 @@ else
   echo "Installation of system tables failed!  Examine the logs in"
   echo "$ldata for more information."
   echo
-  echo "You can try to start the mysqld daemon with:"
+  echo "The problem could be conflicting information in an external"
+  echo "my.cnf files. You can ignore these by doing:"
   echo
-  echo "    shell> $mysqld --skip-grant &"
+  echo "    shell> /scripts/mysql_install_db --defaults-file=~/.my.cnf"
+  echo
+  echo "You can also try to start the mysqld daemon with:"
+  echo
+  echo "    shell> $mysqld --skip-grant --general-log &"
   echo
   echo "and use the command line tool $bindir/mysql"
   echo "to connect to the mysql database and look at the grant tables:"
@@ -408,9 +413,6 @@ else
   echo "    shell> $bindir/mysql -u root mysql"
   echo "    mysql> show tables"
   echo
-  echo "Try 'mysqld --help' if you have problems with paths.  Using"
-  echo "--general-log gives you a log in $ldata that may be helpful."
-  echo
   echo "The latest information about mysql_install_db is available at"
   echo "http://kb.askmonty.org/v/installing-system-tables-mysql_install_db."
   echo "MariaDB is hosted on launchpad; You can find the latest source and"
diff --git a/sql/Makefile.am b/sql/Makefile.am
index afbef9ca197..05cb56bfab7 100644
--- a/sql/Makefile.am
+++ b/sql/Makefile.am
@@ -81,7 +81,7 @@ noinst_HEADERS =	item.h item_func.h item_sum.h item_cmpfunc.h \
 			event_data_objects.h event_scheduler.h \
 			sql_partition.h partition_info.h partition_element.h \
 			contributors.h sql_servers.h \
-                        multi_range_read.h \
+                        multi_range_read.h sql_handler.h \
                         sql_join_cache.h \
 			create_options.h \
 			sql_expression_cache.h
diff --git a/sql/lock.cc b/sql/lock.cc
index 8f5b5ac233f..49f0cfdfa90 100644
--- a/sql/lock.cc
+++ b/sql/lock.cc
@@ -84,41 +84,11 @@
 
 extern HASH open_cache;
 
-/* flags for get_lock_data */
-#define GET_LOCK_UNLOCK         1
-#define GET_LOCK_STORE_LOCKS    2
-
-static MYSQL_LOCK *get_lock_data(THD *thd, TABLE **table,uint count,
-				 uint flags, TABLE **write_locked);
-static void reset_lock_data(MYSQL_LOCK *sql_lock);
+static void reset_lock_data(MYSQL_LOCK *sql_lock, bool unlock);
 static int lock_external(THD *thd, TABLE **table,uint count);
 static int unlock_external(THD *thd, TABLE **table,uint count);
 static void print_lock_error(int error, const char *);
 
-/*
-  Lock tables.
-
-  SYNOPSIS
-    mysql_lock_tables()
-    thd                         The current thread.
-    tables                      An array of pointers to the tables to lock.
-    count                       The number of tables to lock.
-    flags                       Options:
-      MYSQL_LOCK_IGNORE_GLOBAL_READ_LOCK      Ignore a global read lock
-      MYSQL_LOCK_IGNORE_GLOBAL_READ_ONLY      Ignore SET GLOBAL READ_ONLY
-      MYSQL_LOCK_IGNORE_FLUSH                 Ignore a flush tables.
-      MYSQL_LOCK_NOTIFY_IF_NEED_REOPEN        Instead of reopening altered
-                                              or dropped tables by itself,
-                                              mysql_lock_tables() should
-                                              notify upper level and rely
-                                              on caller doing this.
-    need_reopen                 Out parameter, TRUE if some tables were altered
-                                or deleted and should be reopened by caller.
-
-  RETURN
-    A lock structure pointer on success.
-    NULL on error or if some tables should be reopen.
-*/
 
 /* Map the return value of thr_lock to an error from errmsg.txt */
 static int thr_lock_errno_to_mysql[]=
@@ -132,6 +102,7 @@ static int thr_lock_errno_to_mysql[]=
   @param flags Lock flags
   @return 0 if all the check passed, non zero if a check failed.
 */
+
 int mysql_lock_tables_check(THD *thd, TABLE **tables, uint count, uint flags)
 {
   bool log_table_write_query;
@@ -194,81 +165,118 @@ int mysql_lock_tables_check(THD *thd, TABLE **tables, uint count, uint flags)
   DBUG_RETURN(0);
 }
 
+
+/*
+  Lock tables.
+
+  SYNOPSIS
+    mysql_lock_tables()
+    thd                         The current thread.
+    tables                      An array of pointers to the tables to lock.
+    count                       The number of tables to lock.
+    flags                       Options:
+      MYSQL_LOCK_IGNORE_GLOBAL_READ_LOCK      Ignore a global read lock
+      MYSQL_LOCK_IGNORE_GLOBAL_READ_ONLY      Ignore SET GLOBAL READ_ONLY
+      MYSQL_LOCK_IGNORE_FLUSH                 Ignore a flush tables.
+      MYSQL_LOCK_NOTIFY_IF_NEED_REOPEN        Instead of reopening altered
+                                              or dropped tables by itself,
+                                              mysql_lock_tables() should
+                                              notify upper level and rely
+                                              on caller doing this.
+    need_reopen                 Out parameter, TRUE if some tables were altered
+                                or deleted and should be reopened by caller.
+
+  RETURN
+    A lock structure pointer on success.
+    NULL on error or if some tables should be reopen.
+*/
+
 MYSQL_LOCK *mysql_lock_tables(THD *thd, TABLE **tables, uint count,
                               uint flags, bool *need_reopen)
 {
-  MYSQL_LOCK *sql_lock;
   TABLE *write_lock_used;
-  int rc;
-
-  DBUG_ENTER("mysql_lock_tables");
+  MYSQL_LOCK *sql_lock;
+  DBUG_ENTER("mysql_lock_tables(tables)");
 
   *need_reopen= FALSE;
-
   if (mysql_lock_tables_check(thd, tables, count, flags))
-    DBUG_RETURN (NULL);
+    DBUG_RETURN(NULL);
 
+  if (!(sql_lock= get_lock_data(thd, tables, count, GET_LOCK_STORE_LOCKS,
+                                &write_lock_used)) ||
+      ! sql_lock->table_count)
+    DBUG_RETURN(sql_lock);
+
+  if (mysql_lock_tables(thd, sql_lock, write_lock_used != 0, flags,
+                        need_reopen))
+  {
+    /* Clear the lock type of all lock data to avoid reusage. */
+    reset_lock_data(sql_lock, 1);
+    my_free(sql_lock, MYF(0));
+    sql_lock= 0;
+  }
+  DBUG_RETURN(sql_lock);
+}
+
+
+/**
+   Lock a table based on a MYSQL_LOCK structure.
+
+   mysql_lock_tables()
+
+   @param thd			The current thread.
+   @param sql_lock		Tables that should be locked
+   @param write_lock_used 	1 if any of the tables are write locked
+   @param flags			See mysql_lock_tables()
+   @param need_reopen           Out parameter, TRUE if some tables were altered
+                                or deleted and should be reopened by caller.
+
+   @return 0   ok
+   @return 1  error
+*/
+
+bool mysql_lock_tables(THD *thd, MYSQL_LOCK *sql_lock,
+                       bool write_lock_used,
+                       uint flags, bool *need_reopen)
+{
+  int rc;
+  bool error= 1;
+  DBUG_ENTER("mysql_lock_tables(sql_lock)");
+
+  *need_reopen= FALSE;
   for (;;)
   {
-    if (! (sql_lock= get_lock_data(thd, tables, count, GET_LOCK_STORE_LOCKS,
-                                   &write_lock_used)) ||
-        ! sql_lock->table_count)
-      break;
-
-    if (global_read_lock && write_lock_used &&
-        ! (flags & MYSQL_LOCK_IGNORE_GLOBAL_READ_LOCK))
+    if (write_lock_used && !(flags & MYSQL_LOCK_IGNORE_GLOBAL_READ_LOCK))
     {
-      /*
-	Someone has issued LOCK ALL TABLES FOR READ and we want a write lock
-	Wait until the lock is gone
-      */
-      if (wait_if_global_read_lock(thd, 1, 1))
+      if (global_read_lock)
       {
-        /* Clear the lock type of all lock data to avoid reusage. */
-        reset_lock_data(sql_lock);
-	my_free((uchar*) sql_lock,MYF(0));
-	sql_lock=0;
-	break;
+        /*
+          Someone has issued LOCK ALL TABLES FOR READ and we want a write lock
+          Wait until the lock is gone
+        */
+        if (wait_if_global_read_lock(thd, 1, 1))
+          break;
+        if (thd->version != refresh_version)
+          goto retry;
       }
-      if (thd->version != refresh_version)
-      {
-        /* Clear the lock type of all lock data to avoid reusage. */
-        reset_lock_data(sql_lock);
-	my_free((uchar*) sql_lock,MYF(0));
-	goto retry;
-      }
-    }
 
-    if (!(flags & MYSQL_LOCK_IGNORE_GLOBAL_READ_ONLY) &&
-        write_lock_used &&
-        opt_readonly &&
-        !(thd->security_ctx->master_access & SUPER_ACL) &&
-        !thd->slave_thread)
-    {
-      /*
-	Someone has issued SET GLOBAL READ_ONLY=1 and we want a write lock.
-        We do not wait for READ_ONLY=0, and fail.
-      */
-      reset_lock_data(sql_lock);
-      my_free((uchar*) sql_lock, MYF(0));
-      sql_lock=0;
-      my_error(ER_OPTION_PREVENTS_STATEMENT, MYF(0), "--read-only");
-      break;
+      if (opt_readonly &&
+          !(thd->security_ctx->master_access & SUPER_ACL) &&
+          !thd->slave_thread)
+      {
+        /*
+          Someone has issued SET GLOBAL READ_ONLY=1 and we want a write lock.
+          We do not wait for READ_ONLY=0, and fail.
+        */
+        my_error(ER_OPTION_PREVENTS_STATEMENT, MYF(0), "--read-only");
+        break;
+      }
     }
 
     thd_proc_info(thd, "System lock");
-    DBUG_PRINT("info", ("thd->proc_info %s", thd->proc_info));
     if (lock_external(thd, sql_lock->table, sql_lock->table_count))
-    {
-      /* Clear the lock type of all lock data to avoid reusage. */
-      reset_lock_data(sql_lock);
-      my_free((uchar*) sql_lock,MYF(0));
-      sql_lock=0;
       break;
-    }
     thd_proc_info(thd, "Table lock");
-    DBUG_PRINT("info", ("thd->proc_info %s", thd->proc_info));
-    thd->locked=1;
     /* Copy the lock data array. thr_multi_lock() reorders its contens. */
     memcpy(sql_lock->locks + sql_lock->lock_count, sql_lock->locks,
            sql_lock->lock_count * sizeof(*sql_lock->locks));
@@ -277,70 +285,66 @@ MYSQL_LOCK *mysql_lock_tables(THD *thd, TABLE **tables, uint count,
                                                      sql_lock->lock_count,
                                                      sql_lock->lock_count,
                                                      thd->lock_id)];
-    if (rc > 1)                                 /* a timeout or a deadlock */
+    if (rc)                                     /* Locking failed */
     {
       VOID(unlock_external(thd, sql_lock->table, sql_lock->table_count));
-      my_error(rc, MYF(0));
-      my_free((uchar*) sql_lock,MYF(0));
-      sql_lock= 0;
-      break;
+      if (rc > 1)
+      {
+        /* a timeout or a deadlock */
+        my_error(rc, MYF(0));
+        break;
+      }
+      /* We where aborted and should try again from upper level*/
+      thd->some_tables_deleted= 1;
     }
-    else if (rc == 1)                           /* aborted */
+    else
     {
       /*
-        reset_lock_data is required here. If thr_multi_lock fails it
-        resets lock type for tables, which were locked before (and
-        including) one that caused error. Lock type for other tables
-        preserved.
+        Lock worked. Now check that nothing happend while we where waiting
+        to get the lock that would require us to free it.
       */
-      reset_lock_data(sql_lock);
-      thd->some_tables_deleted=1;		// Try again
-      sql_lock->lock_count= 0;                  // Locks are already freed
-    }
-    else if (!thd->some_tables_deleted || (flags & MYSQL_LOCK_IGNORE_FLUSH))
-    {
-      /*
-        Thread was killed or lock aborted. Let upper level close all
-        used tables and retry or give error.
-      */
-      thd->locked=0;
-      break;
-    }
-    else if (!thd->open_tables)
-    {
-      // Only using temporary tables, no need to unlock
-      thd->some_tables_deleted=0;
-      thd->locked=0;
-      break;
-    }
-    thd_proc_info(thd, 0);
+      error= 0;
+      if (!thd->some_tables_deleted || (flags & MYSQL_LOCK_IGNORE_FLUSH))
+      {
+        /*
+          Table was not signaled for deletion or we don't care if it was.
+          Return with table as locked.
+        */
+        break;
+      }
+      else if (!thd->open_tables && !(flags & MYSQL_LOCK_NOT_TEMPORARY))
+      {
+        /*
+          Only using temporary tables, no need to unlock.
+          We need the flag as open_tables is not enough to distingush if
+          we are only using temporary tables for tables used trough
+          the HANDLER interface.
+
+          We reset some_tables_deleted as it doesn't make sense to have this
+          one when we are only using temporary tables.
+        */
+        thd->some_tables_deleted=0;
+        break;
+      }
+      /* some table was altered or deleted. reopen tables marked deleted */
+      error= 1;
+      mysql_unlock_tables(thd, sql_lock, 0);
+    }
 
-    /* some table was altered or deleted. reopen tables marked deleted */
-    mysql_unlock_tables(thd,sql_lock);
-    thd->locked=0;
 retry:
-    sql_lock=0;
     if (flags & MYSQL_LOCK_NOTIFY_IF_NEED_REOPEN)
     {
       *need_reopen= TRUE;
       break;
     }
     if (wait_for_tables(thd))
-      break;					// Couldn't open tables
-  }
-  thd_proc_info(thd, 0);
-  if (thd->killed)
-  {
-    thd->send_kill_message();
-    if (sql_lock)
-    {
-      mysql_unlock_tables(thd,sql_lock);
-      sql_lock=0;
-    }
+      break;                                // Couldn't open tables
+    reset_lock_data(sql_lock, 0);           // Set org locks and retry
   }
 
+  thd_proc_info(thd, 0);
   thd->set_time_after_lock();
-  DBUG_RETURN (sql_lock);
+  DBUG_RETURN(error);
 }
 
 
@@ -380,15 +384,15 @@ static int lock_external(THD *thd, TABLE **tables, uint count)
   DBUG_RETURN(0);
 }
 
-
-void mysql_unlock_tables(THD *thd, MYSQL_LOCK *sql_lock)
+void mysql_unlock_tables(THD *thd, MYSQL_LOCK *sql_lock, bool free_lock)
 {
   DBUG_ENTER("mysql_unlock_tables");
   if (sql_lock->table_count)
     VOID(unlock_external(thd,sql_lock->table,sql_lock->table_count));
   if (sql_lock->lock_count)
     thr_multi_unlock(sql_lock->locks,sql_lock->lock_count, 0);
-  my_free((uchar*) sql_lock,MYF(0));
+  if (free_lock)
+    my_free((uchar*) sql_lock,MYF(0));
   DBUG_VOID_RETURN;
 }
 
@@ -847,12 +851,12 @@ static int unlock_external(THD *thd, TABLE **table,uint count)
   @param write_lock_used   Store pointer to last table with WRITE_ALLOW_WRITE
 */
 
-static MYSQL_LOCK *get_lock_data(THD *thd, TABLE **table_ptr, uint count,
-				 uint flags, TABLE **write_lock_used)
+MYSQL_LOCK *get_lock_data(THD *thd, TABLE **table_ptr, uint count,
+                          uint flags, TABLE **write_lock_used)
 {
   uint i,tables,lock_count;
   MYSQL_LOCK *sql_lock;
-  THR_LOCK_DATA **locks, **locks_buf, **locks_start;
+  THR_LOCK_DATA **locks, **locks_buf;
   TABLE **to, **table_buf;
   DBUG_ENTER("get_lock_data");
 
@@ -891,7 +895,7 @@ static MYSQL_LOCK *get_lock_data(THD *thd, TABLE **table_ptr, uint count,
   {
     TABLE *table;
     enum thr_lock_type lock_type;
-
+    THR_LOCK_DATA **locks_start;
     if ((table=table_ptr[i])->s->tmp_table == NON_TRANSACTIONAL_TMP_TABLE)
       continue;
     lock_type= table->reginfo.lock_type;
@@ -904,12 +908,11 @@ static MYSQL_LOCK *get_lock_data(THD *thd, TABLE **table_ptr, uint count,
 	my_error(ER_OPEN_AS_READONLY,MYF(0),table->alias.c_ptr());
         /* Clear the lock type of the lock data that are stored already. */
         sql_lock->lock_count= (uint) (locks - sql_lock->locks);
-        reset_lock_data(sql_lock);
+        reset_lock_data(sql_lock, 1);
 	my_free((uchar*) sql_lock,MYF(0));
 	DBUG_RETURN(0);
       }
     }
-    THR_LOCK_DATA **org_locks = locks;
     locks_start= locks;
     locks= table->file->store_lock(thd, locks,
                                    (flags & GET_LOCK_UNLOCK) ? TL_IGNORE :
@@ -922,8 +925,13 @@ static MYSQL_LOCK *get_lock_data(THD *thd, TABLE **table_ptr, uint count,
     }
     *to++= table;
     if (locks)
-      for ( ; org_locks != locks ; org_locks++)
-	(*org_locks)->debug_print_param= (void *) table;
+    {
+      for ( ; locks_start != locks ; locks_start++)
+      {
+	(*locks_start)->debug_print_param= (void *) table;
+	(*locks_start)->org_type= (*locks_start)->type;
+      }
+    }
   }
   /*
     We do not use 'tables', because there are cases where store_lock()
@@ -964,10 +972,13 @@ static MYSQL_LOCK *get_lock_data(THD *thd, TABLE **table_ptr, uint count,
   Clear the lock type of all lock data. This ensures that the next
   lock request will set its lock type properly.
 
-  @param sql_lock                  The MySQL lock.
+  @param sql_lock               The MySQL lock.
+  @param unlock			If set, then set lock type to TL_UNLOCK,
+  				otherwise set to original lock type from
+				get_store_lock().
 */
 
-static void reset_lock_data(MYSQL_LOCK *sql_lock)
+static void reset_lock_data(MYSQL_LOCK *sql_lock, bool unlock)
 {
   THR_LOCK_DATA **ldata;
   THR_LOCK_DATA **ldata_end;
@@ -975,10 +986,7 @@ static void reset_lock_data(MYSQL_LOCK *sql_lock)
   for (ldata= sql_lock->locks, ldata_end= ldata + sql_lock->lock_count;
        ldata < ldata_end;
        ldata++)
-  {
-    /* Reset lock type. */
-    (*ldata)->type= TL_UNLOCK;
-  }
+    (*ldata)->type= unlock ? TL_UNLOCK : (*ldata)->org_type;
 }
 
 
diff --git a/sql/mysql_priv.h b/sql/mysql_priv.h
index ca5ccc699cd..e530e38e4f5 100644
--- a/sql/mysql_priv.h
+++ b/sql/mysql_priv.h
@@ -1487,15 +1487,6 @@ void mysqld_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);
 
-/* sql_handler.cc */
-bool mysql_ha_open(THD *thd, TABLE_LIST *tables, bool reopen);
-bool mysql_ha_close(THD *thd, TABLE_LIST *tables);
-bool mysql_ha_read(THD *, TABLE_LIST *,enum enum_ha_read_modes,char *,
-                   List<Item> *,enum ha_rkey_function,Item *,ha_rows,ha_rows);
-void mysql_ha_flush(THD *thd);
-void mysql_ha_rm_tables(THD *thd, TABLE_LIST *tables, bool is_locked);
-void mysql_ha_cleanup(THD *thd);
-
 /* sql_base.cc */
 #define TMP_TABLE_KEY_EXTRA 8
 void set_item_name(Item *item,char *pos,uint length);
@@ -2194,6 +2185,10 @@ extern struct st_VioSSLFd * ssl_acceptor_fd;
 
 MYSQL_LOCK *mysql_lock_tables(THD *thd, TABLE **table, uint count,
                               uint flags, bool *need_reopen);
+bool mysql_lock_tables(THD *thd, MYSQL_LOCK *sql_lock,
+                       bool write_lock_used,
+                       uint flags, bool *need_reopen);
+
 /* mysql_lock_tables() and open_table() flags bits */
 #define MYSQL_LOCK_IGNORE_GLOBAL_READ_LOCK      0x0001
 #define MYSQL_LOCK_IGNORE_FLUSH                 0x0002
@@ -2201,8 +2196,12 @@ MYSQL_LOCK *mysql_lock_tables(THD *thd, TABLE **table, uint count,
 #define MYSQL_OPEN_TEMPORARY_ONLY               0x0008
 #define MYSQL_LOCK_IGNORE_GLOBAL_READ_ONLY      0x0010
 #define MYSQL_LOCK_PERF_SCHEMA                  0x0020
+#define MYSQL_LOCK_NOT_TEMPORARY		0x0040
+/* flags for get_lock_data */
+#define GET_LOCK_UNLOCK         1
+#define GET_LOCK_STORE_LOCKS    2
 
-void mysql_unlock_tables(THD *thd, MYSQL_LOCK *sql_lock);
+void mysql_unlock_tables(THD *thd, MYSQL_LOCK *sql_lock, bool free_lock= 1);
 void mysql_unlock_read_tables(THD *thd, MYSQL_LOCK *sql_lock);
 void mysql_unlock_some_tables(THD *thd, TABLE **table,uint count);
 void mysql_lock_remove(THD *thd, MYSQL_LOCK *locked,TABLE *table,
@@ -2223,6 +2222,8 @@ bool make_global_read_lock_block_commit(THD *thd);
 bool set_protect_against_global_read_lock(void);
 void unset_protect_against_global_read_lock(void);
 void broadcast_refresh(void);
+MYSQL_LOCK *get_lock_data(THD *thd, TABLE **table_ptr, uint count,
+                          uint flags, TABLE **write_lock_used);
 
 /* Lock based on name */
 int lock_and_wait_for_table_name(THD *thd, TABLE_LIST *table_list);
diff --git a/sql/mysqld.cc b/sql/mysqld.cc
index d7a457c6921..c086c34a39a 100644
--- a/sql/mysqld.cc
+++ b/sql/mysqld.cc
@@ -457,7 +457,7 @@ static bool volatile ready_to_exit;
 static my_bool opt_debugging= 0, opt_external_locking= 0, opt_console= 0;
 static my_bool opt_short_log_format= 0;
 static my_bool opt_ignore_wrong_options= 0, opt_expect_abort= 0;
-static my_bool opt_sync= 0;
+static my_bool opt_sync= 0, opt_thread_alarm;
 static uint kill_cached_threads, wake_thread;
 ulong thread_created;
 uint thread_handling;
@@ -6003,7 +6003,7 @@ enum options_mysqld
   OPT_RANGE_ALLOC_BLOCK_SIZE, OPT_ALLOW_SUSPICIOUS_UDFS,
   OPT_QUERY_ALLOC_BLOCK_SIZE, OPT_QUERY_PREALLOC_SIZE,
   OPT_TRANS_ALLOC_BLOCK_SIZE, OPT_TRANS_PREALLOC_SIZE,
-  OPT_SYNC_FRM, OPT_SYNC_BINLOG, OPT_SYNC,
+  OPT_SYNC_FRM, OPT_SYNC_BINLOG, OPT_SYNC, OPT_THREAD_ALARM,
   OPT_SYNC_REPLICATION,
   OPT_SYNC_REPLICATION_SLAVE_ID,
   OPT_SYNC_REPLICATION_TIMEOUT,
@@ -6328,7 +6328,7 @@ struct my_option my_long_options[] =
    "Disable initialization of builtin InnoDB plugin.",
    0, 0, 0, GET_NO_ARG, NO_ARG, 0, 0, 0, 0, 0, 0},
   {"init-connect", OPT_INIT_CONNECT, 
-   "Command(s) that are executed for each new connection.",
+   "Command(s) that are executed for each new connection (but not for SUPER users).",
    &opt_init_connect, &opt_init_connect, 0, GET_STR_ALLOC,
    REQUIRED_ARG, 0, 0, 0, 0, 0, 0},
 #ifndef DISABLE_GRANT_OPTIONS
@@ -7579,6 +7579,10 @@ thread is in the relay logs.",
    "error. Used only if the connection has active cursors.",
    &table_lock_wait_timeout, &table_lock_wait_timeout,
    0, GET_ULONG, REQUIRED_ARG, 50, 1, 1024 * 1024 * 1024, 0, 1, 0},
+  {"thread-alarm", OPT_THREAD_ALARM,
+   "Enable/disable system thread alarm calls. Should only be turned off when running tests or debugging!!",
+   &opt_thread_alarm, &opt_thread_alarm, 0, GET_BOOL, NO_ARG, 1, 0, 0, 0, 0,
+   0},
   {"thread_cache_size", OPT_THREAD_CACHE_SIZE,
    "How many threads we should keep in a cache for reuse.",
    &thread_cache_size, &thread_cache_size, 0, GET_ULONG,
@@ -9358,6 +9362,7 @@ static int get_options(int *argc,char **argv)
   */
   my_disable_locking= myisam_single_user= test(opt_external_locking == 0);
   my_disable_sync= opt_sync == 0;
+  my_disable_thr_alarm= opt_thread_alarm == 0;
   my_default_record_cache_size=global_system_variables.read_buff_size;
   myisam_max_temp_length=
     (my_off_t) global_system_variables.myisam_max_sort_file_size;
diff --git a/sql/net_serv.cc b/sql/net_serv.cc
index 3441d1e120a..1b313da8ce1 100644
--- a/sql/net_serv.cc
+++ b/sql/net_serv.cc
@@ -697,7 +697,8 @@ net_real_write(NET *net,const uchar *packet, size_t len)
   {
     my_bool old_mode;
     thr_end_alarm(&alarmed);
-    vio_blocking(net->vio, net_blocking, &old_mode);
+    if (!net_blocking)
+      vio_blocking(net->vio, net_blocking, &old_mode);
   }
   net->reading_or_writing=0;
   DBUG_RETURN(((int) (pos != end)));
@@ -988,7 +989,8 @@ end:
   {
     my_bool old_mode;
     thr_end_alarm(&alarmed);
-    vio_blocking(net->vio, net_blocking, &old_mode);
+    if (!net_blocking)
+      vio_blocking(net->vio, net_blocking, &old_mode);
   }
   net->reading_or_writing=0;
 #ifdef DEBUG_DATA_PACKETS
diff --git a/sql/sql_base.cc b/sql/sql_base.cc
index 530ebf85955..01c72db0c8e 100644
--- a/sql/sql_base.cc
+++ b/sql/sql_base.cc
@@ -22,6 +22,7 @@
 #include "sp_head.h"
 #include "sp.h"
 #include "sql_trigger.h"
+#include "sql_handler.h"
 #include <m_ctype.h>
 #include <my_dir.h>
 #include <hash.h>
diff --git a/sql/sql_class.cc b/sql/sql_class.cc
index 084e2c8d78c..11447389186 100644
--- a/sql/sql_class.cc
+++ b/sql/sql_class.cc
@@ -44,6 +44,7 @@
 #include "sp_cache.h"
 #include "sql_select.h" /* declares create_tmp_table() */
 #include "debug_sync.h"
+#include "sql_handler.h"
 
 /*
   The following is used to initialise Table_ident with a internal
@@ -702,7 +703,7 @@ THD::THD()
   catalog= (char*)"std"; // the only catalog we have for now
   main_security_ctx.init();
   security_ctx= &main_security_ctx;
-  locked=some_tables_deleted=no_errors=password= 0;
+  some_tables_deleted=no_errors=password= 0;
   query_start_used= 0;
   count_cuted_fields= CHECK_FIELD_IGNORE;
   killed= NOT_KILLED;
diff --git a/sql/sql_class.h b/sql/sql_class.h
index 7f3b0cbfb03..8e0f1664286 100644
--- a/sql/sql_class.h
+++ b/sql/sql_class.h
@@ -1944,7 +1944,7 @@ public:
   bool       slave_thread, one_shot_set;
   /* tells if current statement should binlog row-based(1) or stmt-based(0) */
   bool       current_stmt_binlog_row_based;
-  bool	     locked, some_tables_deleted;
+  bool	     some_tables_deleted;
   bool       last_cuted_field;
   bool	     no_errors, password;
   bool       extra_port;                        /* If extra connection */
diff --git a/sql/sql_db.cc b/sql/sql_db.cc
index 7aa48524b20..b5478287fda 100644
--- a/sql/sql_db.cc
+++ b/sql/sql_db.cc
@@ -20,6 +20,7 @@
 #include <mysys_err.h>
 #include "sp.h"
 #include "events.h"
+#include "sql_handler.h"
 #include <my_dir.h>
 #include <m_ctype.h>
 #include "log.h"
diff --git a/sql/sql_delete.cc b/sql/sql_delete.cc
index 5564d628594..e2cb17090a1 100644
--- a/sql/sql_delete.cc
+++ b/sql/sql_delete.cc
@@ -23,6 +23,7 @@
 #include "sql_select.h"
 #include "sp_head.h"
 #include "sql_trigger.h"
+#include "sql_handler.h"
 
 /**
   Implement DELETE SQL word.
diff --git a/sql/sql_handler.cc b/sql/sql_handler.cc
index 69be8c8e9b4..eaae5efe21c 100644
--- a/sql/sql_handler.cc
+++ b/sql/sql_handler.cc
@@ -56,15 +56,41 @@
   second container. When the table is flushed, the pointer is cleared.
 */
 
+#ifdef USE_PRAGMA_IMPLEMENTATION
+#pragma implementation				// gcc: Class implementation
+#endif
+
 #include "mysql_priv.h"
 #include "sql_select.h"
-#include <assert.h>
+#include "sql_handler.h"
 
 #define HANDLER_TABLES_HASH_SIZE 120
 
 static enum enum_ha_read_modes rkey_to_rnext[]=
 { RNEXT_SAME, RNEXT, RPREV, RNEXT, RPREV, RNEXT, RPREV, RPREV };
 
+/*
+  Set handler to state after create, but keep base information about
+  which table is used
+*/
+
+void SQL_HANDLER::reset()
+{
+  fields.empty();
+  arena.free_items();
+  free_root(&mem_root, MYF(0));
+  my_free(lock, MYF(MY_ALLOW_ZERO_PTR));
+  init();
+}  
+  
+/* Free all allocated data */
+
+SQL_HANDLER::~SQL_HANDLER()
+{
+  reset();
+  my_free(base_data, MYF(MY_ALLOW_ZERO_PTR));
+}
+
 /*
   Get hash key and hash key length.
 
@@ -84,11 +110,11 @@ static enum enum_ha_read_modes rkey_to_rnext[]=
     Pointer to the TABLE_LIST struct.
 */
 
-static char *mysql_ha_hash_get_key(TABLE_LIST *tables, size_t *key_len_p,
+static char *mysql_ha_hash_get_key(SQL_HANDLER *table, size_t *key_len,
                                    my_bool first __attribute__((unused)))
 {
-  *key_len_p= strlen(tables->alias) + 1 ; /* include '\0' in comparisons */
-  return tables->alias;
+  *key_len= table->handler_name.length + 1 ; /* include '\0' in comparisons */
+  return table->handler_name.str;
 }
 
 
@@ -106,9 +132,9 @@ static char *mysql_ha_hash_get_key(TABLE_LIST *tables, size_t *key_len_p,
     Nothing
 */
 
-static void mysql_ha_hash_free(TABLE_LIST *tables)
+static void mysql_ha_hash_free(SQL_HANDLER *table)
 {
-  my_free((char*) tables, MYF(0));
+  delete table;
 }
 
 /**
@@ -120,14 +146,21 @@ static void mysql_ha_hash_free(TABLE_LIST *tables)
 
   @note Though this function takes a list of tables, only the first list entry
   will be closed.
+  @mote handler_object is not deleted!
   @note Broadcasts refresh if it closed a table with old version.
 */
 
-static void mysql_ha_close_table(THD *thd, TABLE_LIST *tables,
+static void mysql_ha_close_table(SQL_HANDLER *handler,
                                  bool is_locked)
 {
+  THD *thd= handler->thd;
+  TABLE *table= handler->table;
   TABLE **table_ptr;
 
+  /* check if table was already closed */
+  if (!table)
+    return;
+
   /*
     Though we could take the table pointer from hash_tables->table,
     we must follow the thd->handler_tables chain anyway, as we need the
@@ -135,13 +168,13 @@ static void mysql_ha_close_table(THD *thd, TABLE_LIST *tables,
     for close_thread_table().
   */
   for (table_ptr= &(thd->handler_tables);
-       *table_ptr && (*table_ptr != tables->table);
+       *table_ptr && (*table_ptr != table);
          table_ptr= &(*table_ptr)->next)
     ;
 
   if (*table_ptr)
   {
-    (*table_ptr)->file->ha_index_or_rnd_end();
+    table->file->ha_index_or_rnd_end();
     if (! is_locked)
       VOID(pthread_mutex_lock(&LOCK_open));
     if (close_thread_table(thd, table_ptr))
@@ -152,17 +185,15 @@ static void mysql_ha_close_table(THD *thd, TABLE_LIST *tables,
     if (! is_locked)
       VOID(pthread_mutex_unlock(&LOCK_open));
   }
-  else if (tables->table)
+  else
   {
     /* Must be a temporary table */
-    TABLE *table= tables->table;
     table->file->ha_index_or_rnd_end();
     table->query_id= thd->query_id;
     table->open_by_handler= 0;
   }
-
-  /* Mark table as closed, ready for re-open if necessary. */
-  tables->table= NULL;
+  my_free(handler->lock, MYF(MY_ALLOW_ZERO_PTR));
+  handler->init();
 }
 
 /*
@@ -178,7 +209,7 @@ static void mysql_ha_close_table(THD *thd, TABLE_LIST *tables,
     Though this function takes a list of tables, only the first list entry
     will be opened.
     'reopen' is set when a handler table is to be re-opened. In this case,
-    'tables' is the pointer to the hashed TABLE_LIST object which has been
+    'tables' is the pointer to the hashed SQL_HANDLER object which has been
     saved on the original open.
     'reopen' is also used to suppress the sending of an 'ok' message.
 
@@ -187,17 +218,17 @@ static void mysql_ha_close_table(THD *thd, TABLE_LIST *tables,
     TRUE  Error
 */
 
-bool mysql_ha_open(THD *thd, TABLE_LIST *tables, bool reopen)
+bool mysql_ha_open(THD *thd, TABLE_LIST *tables, SQL_HANDLER *reopen)
 {
-  TABLE_LIST    *hash_tables = NULL;
-  char          *db, *name, *alias;
-  uint          dblen, namelen, aliaslen, counter;
+  SQL_HANDLER   *sql_handler= 0;
+  uint          counter;
   int           error;
-  TABLE         *backup_open_tables;
+  TABLE         *table, *backup_open_tables, *write_lock_used;
+  Query_arena backup_arena;
   DBUG_ENTER("mysql_ha_open");
   DBUG_PRINT("enter",("'%s'.'%s' as '%s'  reopen: %d",
                       tables->db, tables->table_name, tables->alias,
-                      (int) reopen));
+                      reopen != 0));
 
   if (tables->schema_table)
   {
@@ -210,7 +241,7 @@ bool mysql_ha_open(THD *thd, TABLE_LIST *tables, bool reopen)
   if (! hash_inited(&thd->handler_tables_hash))
   {
     /*
-      HASH entries are of type TABLE_LIST.
+      HASH entries are of type SQL_HANDLER
     */
     if (hash_init(&thd->handler_tables_hash, &my_charset_latin1,
                   HANDLER_TABLES_HASH_SIZE, 0, 0,
@@ -288,8 +319,10 @@ bool mysql_ha_open(THD *thd, TABLE_LIST *tables, bool reopen)
   if (error)
     goto err;
 
+  table= tables->table;
+
   /* There can be only one table in '*tables'. */
-  if (! (tables->table->file->ha_table_flags() & HA_CAN_SQL_HANDLER))
+  if (! (table->file->ha_table_flags() & HA_CAN_SQL_HANDLER))
   {
     my_error(ER_ILLEGAL_HA, MYF(0), tables->alias);
     goto err;
@@ -297,36 +330,69 @@ bool mysql_ha_open(THD *thd, TABLE_LIST *tables, bool reopen)
 
   if (! reopen)
   {
-    /* copy the TABLE_LIST struct */
-    dblen= strlen(tables->db) + 1;
-    namelen= strlen(tables->table_name) + 1;
-    aliaslen= strlen(tables->alias) + 1;
-    if (!(my_multi_malloc(MYF(MY_WME),
-                          &hash_tables, (uint) sizeof(*hash_tables),
-                          &db, (uint) dblen,
-                          &name, (uint) namelen,
-                          &alias, (uint) aliaslen,
+    /* copy data to sql_handler */
+    if (!(sql_handler= new SQL_HANDLER(thd)))
+      goto err;
+    init_alloc_root(&sql_handler->mem_root, 1024, 0);
+
+    sql_handler->table= table;
+    sql_handler->db.length= strlen(tables->db);
+    sql_handler->table_name.length= strlen(tables->table_name);
+    sql_handler->handler_name.length= strlen(tables->alias);
+
+    if (!(my_multi_malloc(MY_WME,
+                          &sql_handler->db.str,
+                          (uint) sql_handler->db.length + 1,
+                          &sql_handler->table_name.str,
+                          (uint) sql_handler->table_name.length + 1,
+                          &sql_handler->handler_name.str,
+                          (uint) sql_handler->handler_name.length + 1,
                           NullS)))
       goto err;
-    /* structure copy */
-    *hash_tables= *tables;
-    hash_tables->db= db;
-    hash_tables->table_name= name;
-    hash_tables->alias= alias;
-    memcpy(hash_tables->db, tables->db, dblen);
-    memcpy(hash_tables->table_name, tables->table_name, namelen);
-    memcpy(hash_tables->alias, tables->alias, aliaslen);
+    sql_handler->base_data= sql_handler->db.str;  // Free this
+    memcpy(sql_handler->db.str, tables->db, sql_handler->db.length +1);
+    memcpy(sql_handler->table_name.str, tables->table_name,
+           sql_handler->table_name.length+1);
+    memcpy(sql_handler->handler_name.str, tables->alias,
+           sql_handler->handler_name.length +1);
 
     /* add to hash */
-    if (my_hash_insert(&thd->handler_tables_hash, (uchar*) hash_tables))
+    if (my_hash_insert(&thd->handler_tables_hash, (uchar*) sql_handler))
       goto err;
   }
+  else
+  {
+    sql_handler= reopen;
+    sql_handler->reset();
+  }    
+  sql_handler->table= table;
+
+  if (!(sql_handler->lock= get_lock_data(thd, &sql_handler->table, 1,
+                                         GET_LOCK_STORE_LOCKS,
+                                         &write_lock_used)))
+    goto err;
+
+  /* Get a list of all fields for send_fields */
+  thd->set_n_backup_active_arena(&sql_handler->arena, &backup_arena);
+  error= table->fill_item_list(&sql_handler->fields);
+  thd->restore_active_arena(&sql_handler->arena, &backup_arena);
+
+  if (error)
+  {
+    if (reopen)
+      sql_handler= 0;
+    goto err;
+  }
+
+  /* Always read all columns */
+  table->read_set= &table->s->all_set;
+  table->vcol_set= &table->s->all_set;
 
   /*
     If it's a temp table, don't reset table->query_id as the table is
     being used by this handler. Otherwise, no meaning at all.
   */
-  tables->table->open_by_handler= 1;
+  table->open_by_handler= 1;
 
   if (! reopen)
     my_ok(thd);
@@ -334,10 +400,13 @@ bool mysql_ha_open(THD *thd, TABLE_LIST *tables, bool reopen)
   DBUG_RETURN(FALSE);
 
 err:
-  if (hash_tables)
-    my_free((char*) hash_tables, MYF(0));
+  delete sql_handler;
   if (tables->table)
-    mysql_ha_close_table(thd, tables, FALSE);
+  {
+    SQL_HANDLER tmp_sql_handler(thd);
+    tmp_sql_handler.table= tables->table;
+    mysql_ha_close_table(&tmp_sql_handler, FALSE);
+  }
   DBUG_PRINT("exit",("ERROR"));
   DBUG_RETURN(TRUE);
 }
@@ -362,17 +431,17 @@ err:
 
 bool mysql_ha_close(THD *thd, TABLE_LIST *tables)
 {
-  TABLE_LIST    *hash_tables;
+  SQL_HANDLER *handler;
   DBUG_ENTER("mysql_ha_close");
   DBUG_PRINT("enter",("'%s'.'%s' as '%s'",
                       tables->db, tables->table_name, tables->alias));
 
-  if ((hash_tables= (TABLE_LIST*) hash_search(&thd->handler_tables_hash,
-                                              (uchar*) tables->alias,
-                                              strlen(tables->alias) + 1)))
+  if ((handler= (SQL_HANDLER*) hash_search(&thd->handler_tables_hash,
+                                           (uchar*) tables->alias,
+                                           strlen(tables->alias) + 1)))
   {
-    mysql_ha_close_table(thd, hash_tables, FALSE);
-    hash_delete(&thd->handler_tables_hash, (uchar*) hash_tables);
+    mysql_ha_close_table(handler, FALSE);
+    hash_delete(&thd->handler_tables_hash, (uchar*) handler);
   }
   else
   {
@@ -387,6 +456,161 @@ bool mysql_ha_close(THD *thd, TABLE_LIST *tables)
 }
 
 
+/**
+   Finds an open HANDLER table.
+
+   @params name		Name of handler to open
+
+   @return 0 failure
+   @return handler
+*/  
+
+SQL_HANDLER *mysql_ha_find_handler(THD *thd, const char *name)
+{
+  SQL_HANDLER *handler;
+  if ((handler= (SQL_HANDLER*) hash_search(&thd->handler_tables_hash,
+                                           (uchar*) name,
+                                           strlen(name) + 1)))
+  {
+    DBUG_PRINT("info-in-hash",("'%s'.'%s' as '%s' table: %p",
+                               handler->db.str,
+                               handler->table_name.str,
+                               handler->handler_name.str, handler->table));
+    if (!handler->table)
+    {
+      /* The handler table has been closed. Re-open it. */
+      TABLE_LIST tmp;
+      tmp.init_one_table(handler->db.str, handler->table_name.str,
+                         TL_READ);
+      tmp.alias= handler->handler_name.str;
+
+      if (mysql_ha_open(thd, &tmp, handler))
+      {
+        DBUG_PRINT("exit",("reopen failed"));
+        return 0;
+      }
+    }
+  }
+  else
+  {
+    my_error(ER_UNKNOWN_TABLE, MYF(0), name, "HANDLER");
+    return 0;
+  }
+  return handler;
+}
+
+
+/**
+   Check that condition and key name are ok
+
+   @param handler
+   @param mode		Read mode (RFIRST, RNEXT etc...)
+   @param keyname	Key to use.
+   @param key_expr      List of key column values
+   @param cond		Where clause
+   @param in_prepare	If we are in prepare phase (we can't evalute items yet)
+
+   @return 0 ok
+   @return 1 error
+
+   In ok, then values of used key and mode is stored in sql_handler
+*/
+
+static bool
+mysql_ha_fix_cond_and_key(SQL_HANDLER *handler, 
+                          enum enum_ha_read_modes mode, char *keyname,
+                          List<Item> *key_expr,
+                          Item *cond, bool in_prepare)
+{
+  THD *thd= handler->thd;
+  TABLE *table= handler->table;
+  if (cond)
+  {
+    /* This can only be true for temp tables */
+    if (table->query_id != thd->query_id)
+      cond->cleanup();                          // File was reopened
+    if ((!cond->fixed &&
+	 cond->fix_fields(thd, &cond)) || cond->check_cols(1))
+      return 1;
+  }
+
+  if (keyname)
+  {
+    /* Check if same as last keyname. If not, do a full lookup */
+    if (handler->keyno < 0 ||
+        my_strcasecmp(&my_charset_latin1,
+                      keyname,
+                      table->s->key_info[handler->keyno].name))
+    {
+      if ((handler->keyno= find_type(keyname, &table->s->keynames, 1+2)-1)<0)
+      {
+        my_error(ER_KEY_DOES_NOT_EXITS, MYF(0), keyname,
+                 handler->handler_name);
+        return 1;
+      }
+    }
+
+    /* Check key parts */
+    if (mode == RKEY)
+    {
+      TABLE *table= handler->table;
+      KEY *keyinfo= table->key_info + handler->keyno;
+      KEY_PART_INFO *key_part= keyinfo->key_part;
+      List_iterator<Item> it_ke(*key_expr);
+      Item *item;
+      key_part_map keypart_map;
+      uint key_len;
+
+      if (key_expr->elements > keyinfo->key_parts)
+      {
+        my_error(ER_TOO_MANY_KEY_PARTS, MYF(0), keyinfo->key_parts);
+        return 1;
+      }
+      for (keypart_map= key_len=0 ; (item=it_ke++) ; key_part++)
+      {
+        my_bitmap_map *old_map;
+	/* note that 'item' can be changed by fix_fields() call */
+        if ((!item->fixed &&
+             item->fix_fields(thd, it_ke.ref())) ||
+	    (item= *it_ke.ref())->check_cols(1))
+          return 1;
+	if (item->used_tables() & ~(RAND_TABLE_BIT | PARAM_TABLE_BIT))
+        {
+          my_error(ER_WRONG_ARGUMENTS,MYF(0),"HANDLER ... READ");
+	  return 1;
+        }
+        if (!in_prepare)
+        {
+          old_map= dbug_tmp_use_all_columns(table, table->write_set);
+          (void) item->save_in_field(key_part->field, 1);
+          dbug_tmp_restore_column_map(table->write_set, old_map);
+        }
+        key_len+= key_part->store_length;
+        keypart_map= (keypart_map << 1) | 1;
+      }
+      handler->keypart_map= keypart_map;
+      handler->key_len= key_len;
+    }
+    else
+    {
+      /*
+        Check if the same index involved.
+        We need to always do this check because we may not have yet
+        called the handler since the last keyno change.
+      */
+      if ((uint) handler->keyno != table->file->get_index())
+      {
+        if (mode == RNEXT)
+          mode= RFIRST;
+        else if (mode == RPREV)
+          mode= RLAST;
+      }
+    }
+  }
+  handler->mode= mode;                          // Store adjusted mode
+  return 0;
+}
+
 /*
   Read from a HANDLER table.
 
@@ -413,147 +637,77 @@ bool mysql_ha_read(THD *thd, TABLE_LIST *tables,
                    enum ha_rkey_function ha_rkey_mode, Item *cond,
                    ha_rows select_limit_cnt, ha_rows offset_limit_cnt)
 {
-  TABLE_LIST    *hash_tables;
-  TABLE         *table, *backup_open_tables;
-  MYSQL_LOCK    *lock;
+  SQL_HANDLER   *handler;
+  TABLE         *table;
   List<Item>	list;
   Protocol	*protocol= thd->protocol;
   char		buff[MAX_FIELD_WIDTH];
   String	buffer(buff, sizeof(buff), system_charset_info);
-  int           error, keyno= -1;
+  int           error, keyno;
   uint          num_rows;
   uchar		*UNINIT_VAR(key);
   uint		UNINIT_VAR(key_len);
   bool          need_reopen;
+  List_iterator<Item> it;
   DBUG_ENTER("mysql_ha_read");
   DBUG_PRINT("enter",("'%s'.'%s' as '%s'",
                       tables->db, tables->table_name, tables->alias));
 
-  thd->lex->select_lex.context.resolve_in_table_list_only(tables);
-  list.push_front(new Item_field(&thd->lex->select_lex.context,
-                                 NULL, NULL, "*"));
-  List_iterator<Item> it(list);
-  it++;
 
 retry:
-  if ((hash_tables= (TABLE_LIST*) hash_search(&thd->handler_tables_hash,
-                                              (uchar*) tables->alias,
-                                              strlen(tables->alias) + 1)))
+  if (!(handler= mysql_ha_find_handler(thd, tables->alias)))
+    goto err0;
+
+  table= handler->table;
+  tables->table= table;                         // This is used by fix_fields
+
+  /* save open_tables state */
+  if (handler->lock->lock_count > 0)
   {
-    table= hash_tables->table;
-    DBUG_PRINT("info-in-hash",("'%s'.'%s' as '%s' table: 0x%lx",
-                               hash_tables->db, hash_tables->table_name,
-                               hash_tables->alias, (long) table));
-    if (!table)
+    bool lock_error;
+
+    handler->lock->locks[0]->type= handler->lock->locks[0]->org_type;
+    lock_error= mysql_lock_tables(thd, handler->lock, 0, 
+                                  (MYSQL_LOCK_NOTIFY_IF_NEED_REOPEN |
+                                   (handler->table->s->tmp_table ==
+                                    NO_TMP_TABLE ?
+                                    MYSQL_LOCK_NOT_TEMPORARY : 0)),
+                                  &need_reopen);
+    if (need_reopen)
     {
-      /*
-        The handler table has been closed. Re-open it.
-      */
-      if (mysql_ha_open(thd, hash_tables, 1))
+      mysql_ha_close_table(handler, FALSE);
+      if (thd->stmt_arena->is_stmt_execute())
       {
-        DBUG_PRINT("exit",("reopen failed"));
+        /*
+          As we have already sent field list and types to the client, we can't
+          handle any changes in the table format for prepared statements.
+          Better to force a reprepare.
+        */
+        my_error(ER_NEED_REPREPARE, MYF(0));
         goto err0;
       }
 
-      table= hash_tables->table;
-      DBUG_PRINT("info",("re-opened '%s'.'%s' as '%s' tab %p",
-                         hash_tables->db, hash_tables->table_name,
-                         hash_tables->alias, table));
+      /*
+        The lock might have been aborted, we need to manually reset
+        thd->some_tables_deleted because handler's tables are closed
+        in a non-standard way. Otherwise we might loop indefinitely.
+      */
+      thd->some_tables_deleted= 0;
+      goto retry;
     }
 
-#if MYSQL_VERSION_ID < 40100
-    if (*tables->db && strcmp(table->table_cache_key, tables->db))
-    {
-      DBUG_PRINT("info",("wrong db"));
-      table= NULL;
-    }
-#endif
-  }
-  else
-    table= NULL;
-
-  if (!table)
-  {
-#if MYSQL_VERSION_ID < 40100
-    char buff[MAX_DBKEY_LENGTH];
-    if (*tables->db)
-      strxnmov(buff, sizeof(buff)-1, tables->db, ".", tables->table_name,
-               NullS);
-    else
-      strncpy(buff, tables->alias, sizeof(buff));
-    my_error(ER_UNKNOWN_TABLE, MYF(0), buff, "HANDLER");
-#else
-    my_error(ER_UNKNOWN_TABLE, MYF(0), tables->alias, "HANDLER");
-#endif
-    goto err0;
-  }
-  tables->table=table;
-
-  /* save open_tables state */
-  backup_open_tables= thd->open_tables;
-  /*
-    mysql_lock_tables() needs thd->open_tables to be set correctly to
-    be able to handle aborts properly. When the abort happens, it's
-    safe to not protect thd->handler_tables because it won't close any
-    tables.
-  */
-  thd->open_tables= thd->handler_tables;
-
-  lock= mysql_lock_tables(thd, &tables->table, 1,
-                          MYSQL_LOCK_NOTIFY_IF_NEED_REOPEN, &need_reopen);
-
-  /* restore previous context */
-  thd->open_tables= backup_open_tables;
-
-  if (need_reopen)
-  {
-    mysql_ha_close_table(thd, hash_tables, FALSE);
-    /*
-      The lock might have been aborted, we need to manually reset
-      thd->some_tables_deleted because handler's tables are closed
-      in a non-standard way. Otherwise we might loop indefinitely.
-    */
-    thd->some_tables_deleted= 0;
-    goto retry;
+    if (lock_error)
+      goto err0; // mysql_lock_tables() printed error message already
   }
 
-  if (!lock)
-    goto err0; // mysql_lock_tables() printed error message already
-
-  // Always read all columns
-  tables->table->read_set= &tables->table->s->all_set;
-
-  if (cond)
-  {
-    if (table->query_id != thd->query_id)
-      cond->cleanup();                          // File was reopened
-    if ((!cond->fixed &&
-	 cond->fix_fields(thd, &cond)) || cond->check_cols(1))
-      goto err;
-  }
-
-  if (keyname)
-  {
-    if ((keyno=find_type(keyname, &table->s->keynames, 1+2)-1)<0)
-    {
-      my_error(ER_KEY_DOES_NOT_EXITS, MYF(0), keyname, tables->alias);
-      goto err;
-    }
-    /* Check if the same index involved. */
-    if ((uint) keyno != table->file->get_index())
-    {
-      if (mode == RNEXT)
-        mode= RFIRST;
-      else if (mode == RPREV)
-        mode= RLAST;
-    }
-  }
-
-  if (insert_fields(thd, &thd->lex->select_lex.context,
-                    tables->db, tables->alias, &it, 0))
+  if (mysql_ha_fix_cond_and_key(handler, mode, keyname, key_expr, cond, 0))
     goto err;
+  mode= handler->mode;
+  keyno= handler->keyno;
 
-  protocol->send_fields(&list, Protocol::SEND_NUM_ROWS | Protocol::SEND_EOF);
+  it.init(handler->fields);
+  protocol->send_fields(&handler->fields,
+                        Protocol::SEND_NUM_ROWS | Protocol::SEND_EOF);
 
   /*
     In ::external_lock InnoDB resets the fields which tell it that
@@ -576,9 +730,7 @@ retry:
           error= table->file->ha_index_next(table->record[0]);
         }
         else
-        {
           error= table->file->ha_rnd_next(table->record[0]);
-        }
         break;
       }
       /* else fall through */
@@ -595,7 +747,7 @@ retry:
 	if (!(error= table->file->ha_rnd_init(1)))
           error= table->file->ha_rnd_next(table->record[0]);
       }
-      mode=RNEXT;
+      mode= RNEXT;
       break;
     case RPREV:
       DBUG_ASSERT(keyname != 0);
@@ -612,7 +764,7 @@ retry:
       table->file->ha_index_or_rnd_end();
       table->file->ha_index_init(keyno, 1);
       error= table->file->ha_index_last(table->record[0]);
-      mode=RPREV;
+      mode= RPREV;
       break;
     case RNEXT_SAME:
       /* Continue scan on "(keypart1,keypart2,...)=(c1, c2, ...)  */
@@ -622,44 +774,17 @@ retry:
     case RKEY:
     {
       DBUG_ASSERT(keyname != 0);
-      KEY *keyinfo=table->key_info+keyno;
-      KEY_PART_INFO *key_part=keyinfo->key_part;
-      if (key_expr->elements > keyinfo->key_parts)
-      {
-	my_error(ER_TOO_MANY_KEY_PARTS, MYF(0), keyinfo->key_parts);
-	goto err;
-      }
-      List_iterator<Item> it_ke(*key_expr);
-      Item *item;
-      key_part_map keypart_map;
-      for (keypart_map= key_len=0 ; (item=it_ke++) ; key_part++)
-      {
-        my_bitmap_map *old_map;
-	// 'item' can be changed by fix_fields() call
-        if ((!item->fixed &&
-             item->fix_fields(thd, it_ke.ref())) ||
-	    (item= *it_ke.ref())->check_cols(1))
-	  goto err;
-	if (item->used_tables() & ~RAND_TABLE_BIT)
-        {
-          my_error(ER_WRONG_ARGUMENTS,MYF(0),"HANDLER ... READ");
-	  goto err;
-        }
-        old_map= dbug_tmp_use_all_columns(table, table->write_set);
-	(void) item->save_in_field(key_part->field, 1);
-        dbug_tmp_restore_column_map(table->write_set, old_map);
-	key_len+=key_part->store_length;
-        keypart_map= (keypart_map << 1) | 1;
-      }
 
-      if (!(key= (uchar*) thd->calloc(ALIGN_SIZE(key_len))))
+      if (!(key= (uchar*) thd->calloc(ALIGN_SIZE(handler->key_len))))
 	goto err;
       table->file->ha_index_or_rnd_end();
       table->file->ha_index_init(keyno, 1);
-      key_copy(key, table->record[0], table->key_info + keyno, key_len);
+      key_copy(key, table->record[0], table->key_info + keyno,
+               handler->key_len);
       error= table->file->ha_index_read_map(table->record[0],
-                                            key, keypart_map, ha_rkey_mode);
-      mode=rkey_to_rnext[(int)ha_rkey_mode];
+                                            key, handler->keypart_map,
+                                            ha_rkey_mode);
+      mode= rkey_to_rnext[(int)ha_rkey_mode];
       break;
     }
     default:
@@ -703,19 +828,41 @@ retry:
     num_rows++;
   }
 ok:
-  mysql_unlock_tables(thd,lock);
+  mysql_unlock_tables(thd, handler->lock, 0);
   my_eof(thd);
   DBUG_PRINT("exit",("OK"));
   DBUG_RETURN(FALSE);
 
 err:
-  mysql_unlock_tables(thd,lock);
+  mysql_unlock_tables(thd, handler->lock, 0);
 err0:
   DBUG_PRINT("exit",("ERROR"));
   DBUG_RETURN(TRUE);
 }
 
 
+/**
+   Prepare for handler read
+
+   For parameters, see mysql_ha_read()
+*/
+
+SQL_HANDLER *mysql_ha_read_prepare(THD *thd, TABLE_LIST *tables,
+                                   enum enum_ha_read_modes mode, char *keyname,
+                                   List<Item> *key_expr, Item *cond)
+{
+  SQL_HANDLER *handler;
+  DBUG_ENTER("mysql_ha_read_prepare");
+  if (!(handler= mysql_ha_find_handler(thd, tables->alias)))
+    DBUG_RETURN(0);
+  tables->table= handler->table;         // This is used by fix_fields
+  if (mysql_ha_fix_cond_and_key(handler, mode, keyname, key_expr, cond, 1))
+    DBUG_RETURN(0);
+  DBUG_RETURN(handler);
+}
+
+  
+
 /**
   Scan the handler tables hash for matching tables.
 
@@ -727,30 +874,32 @@ err0:
           table was matched.
 */
 
-static TABLE_LIST *mysql_ha_find(THD *thd, TABLE_LIST *tables)
+static SQL_HANDLER *mysql_ha_find_match(THD *thd, TABLE_LIST *tables)
 {
-  TABLE_LIST *hash_tables, *head= NULL, *first= tables;
-  DBUG_ENTER("mysql_ha_find");
+  SQL_HANDLER *hash_tables, *head= NULL;
+  TABLE_LIST *first= tables;
+  DBUG_ENTER("mysql_ha_find_match");
 
   /* search for all handlers with matching table names */
   for (uint i= 0; i < thd->handler_tables_hash.records; i++)
   {
-    hash_tables= (TABLE_LIST*) hash_element(&thd->handler_tables_hash, i);
+    hash_tables= (SQL_HANDLER*) hash_element(&thd->handler_tables_hash, i);
+
     for (tables= first; tables; tables= tables->next_local)
     {
       if ((! *tables->db ||
-          ! my_strcasecmp(&my_charset_latin1, hash_tables->db, tables->db)) &&
-          ! my_strcasecmp(&my_charset_latin1, hash_tables->table_name,
+          ! my_strcasecmp(&my_charset_latin1, hash_tables->db.str,
+                          tables->db)) &&
+          ! my_strcasecmp(&my_charset_latin1, hash_tables->table_name.str,
                           tables->table_name))
+      {
+        /* Link into hash_tables list */
+        hash_tables->next= head;
+        head= hash_tables;
         break;
-    }
-    if (tables)
-    {
-      hash_tables->next_local= head;
-      head= hash_tables;
+      }
     }
   }
-
   DBUG_RETURN(head);
 }
 
@@ -767,18 +916,18 @@ static TABLE_LIST *mysql_ha_find(THD *thd, TABLE_LIST *tables)
 
 void mysql_ha_rm_tables(THD *thd, TABLE_LIST *tables, bool is_locked)
 {
-  TABLE_LIST *hash_tables, *next;
+  SQL_HANDLER *hash_tables, *next;
   DBUG_ENTER("mysql_ha_rm_tables");
 
   DBUG_ASSERT(tables);
 
-  hash_tables= mysql_ha_find(thd, tables);
+  hash_tables= mysql_ha_find_match(thd, tables);
 
   while (hash_tables)
   {
-    next= hash_tables->next_local;
+    next= hash_tables->next;
     if (hash_tables->table)
-      mysql_ha_close_table(thd, hash_tables, is_locked);
+      mysql_ha_close_table(hash_tables, is_locked);
     hash_delete(&thd->handler_tables_hash, (uchar*) hash_tables);
     hash_tables= next;
   }
@@ -798,16 +947,16 @@ void mysql_ha_rm_tables(THD *thd, TABLE_LIST *tables, bool is_locked)
 
 void mysql_ha_flush(THD *thd)
 {
-  TABLE_LIST *hash_tables;
+  SQL_HANDLER *hash_tables;
   DBUG_ENTER("mysql_ha_flush");
 
   safe_mutex_assert_owner(&LOCK_open);
 
   for (uint i= 0; i < thd->handler_tables_hash.records; i++)
   {
-    hash_tables= (TABLE_LIST*) hash_element(&thd->handler_tables_hash, i);
+    hash_tables= (SQL_HANDLER*) hash_element(&thd->handler_tables_hash, i);
     if (hash_tables->table && hash_tables->table->needs_reopen_or_name_lock())
-      mysql_ha_close_table(thd, hash_tables, TRUE);
+      mysql_ha_close_table(hash_tables, TRUE);
   }
 
   DBUG_VOID_RETURN;
@@ -824,14 +973,14 @@ void mysql_ha_flush(THD *thd)
 
 void mysql_ha_cleanup(THD *thd)
 {
-  TABLE_LIST *hash_tables;
+  SQL_HANDLER *hash_tables;
   DBUG_ENTER("mysql_ha_cleanup");
 
   for (uint i= 0; i < thd->handler_tables_hash.records; i++)
   {
-    hash_tables= (TABLE_LIST*) hash_element(&thd->handler_tables_hash, i);
+    hash_tables= (SQL_HANDLER*) hash_element(&thd->handler_tables_hash, i);
     if (hash_tables->table)
-      mysql_ha_close_table(thd, hash_tables, FALSE);
+      mysql_ha_close_table(hash_tables, FALSE);
    }
 
   hash_free(&thd->handler_tables_hash);
diff --git a/sql/sql_handler.h b/sql/sql_handler.h
new file mode 100644
index 00000000000..54e72e9f50e
--- /dev/null
+++ b/sql/sql_handler.h
@@ -0,0 +1,61 @@
+/* Copyright (C) 2010 Monty Program Ab
+   This program is free software; you can redistribute it and/or modify
+   it under the terms of the GNU General Public License as published by
+   the Free Software Foundation; version 2 of the License.
+
+   This program is distributed in the hope that it will be useful,
+   but WITHOUT ANY WARRANTY; without even the implied warranty of
+   MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+   GNU General Public License for more details.
+
+   You should have received a copy of the GNU General Public License
+   along with this program; if not, write to the Free Software
+   Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA */
+
+#ifdef USE_PRAGMA_INTERFACE
+#pragma interface			/* gcc class implementation */
+#endif
+
+/* Open handlers are stored here */
+
+class SQL_HANDLER {
+public:
+  TABLE *table;
+  List<Item> fields;                            /* Fields, set on open */
+  THD *thd;
+  LEX_STRING handler_name;
+  LEX_STRING db;
+  LEX_STRING table_name;
+  MEM_ROOT mem_root;
+  MYSQL_LOCK *lock;
+
+  key_part_map keypart_map;
+  int keyno;                                    /* Used key */
+  uint key_len;
+  enum enum_ha_read_modes mode;
+
+  /* This is only used when deleting many handler objects */
+  SQL_HANDLER *next;
+
+  Query_arena arena;
+  char *base_data;
+  SQL_HANDLER(THD *thd_arg) :
+    thd(thd_arg), arena(&mem_root, Query_arena::INITIALIZED)
+  { init(); clear_alloc_root(&mem_root); base_data= 0; }
+  void init() { keyno= -1; table= 0; lock= 0; }
+  void reset();
+
+  ~SQL_HANDLER();
+};
+
+bool mysql_ha_open(THD *thd, TABLE_LIST *tables, SQL_HANDLER *reopen);
+bool mysql_ha_close(THD *thd, TABLE_LIST *tables);
+bool mysql_ha_read(THD *, TABLE_LIST *,enum enum_ha_read_modes,char *,
+                   List<Item> *,enum ha_rkey_function,Item *,ha_rows,ha_rows);
+void mysql_ha_flush(THD *thd);
+void mysql_ha_rm_tables(THD *thd, TABLE_LIST *tables, bool is_locked);
+void mysql_ha_cleanup(THD *thd);
+
+SQL_HANDLER *mysql_ha_read_prepare(THD *thd, TABLE_LIST *tables,
+                                   enum enum_ha_read_modes mode, char *keyname,
+                                   List<Item> *key_expr, Item *cond);
diff --git a/sql/sql_parse.cc b/sql/sql_parse.cc
index 9212017d955..aa3a696e2fa 100644
--- a/sql/sql_parse.cc
+++ b/sql/sql_parse.cc
@@ -28,6 +28,7 @@
 #include "events.h"
 #include "sql_trigger.h"
 #include "debug_sync.h"
+#include "sql_handler.h"
 
 #ifdef WITH_ARIA_STORAGE_ENGINE
 #include "../storage/maria/ha_maria.h"
diff --git a/sql/sql_prepare.cc b/sql/sql_prepare.cc
index a1243589ed1..8a9870e2865 100644
--- a/sql/sql_prepare.cc
+++ b/sql/sql_prepare.cc
@@ -95,6 +95,7 @@ When one supplies long data for a placeholder:
 #else
 #include <mysql_com.h>
 #endif
+#include "sql_handler.h"
 
 /**
   A result class used to send cursor rows using the binary protocol.
@@ -243,6 +244,8 @@ static bool send_prep_stmt(Prepared_statement *stmt, uint columns)
   int error;
   THD *thd= stmt->thd;
   DBUG_ENTER("send_prep_stmt");
+  DBUG_PRINT("enter",("stmt->id: %lu  columns: %d  param_count: %d",
+                      stmt->id, columns, stmt->param_count));
 
   buff[0]= 0;                                   /* OK packet indicator */
   int4store(buff+1, stmt->id);
@@ -1835,6 +1838,56 @@ static bool mysql_test_insert_select(Prepared_statement *stmt,
   return res;
 }
 
+/**
+  Validate SELECT statement.
+
+    In case of success, if this query is not EXPLAIN, send column list info
+    back to the client.
+
+  @param stmt               prepared statement
+  @param tables             list of tables used in the query
+
+  @retval 0 success
+  @retval 1 error, error message is set in THD
+  @retval 2 success, and statement metadata has been sent
+*/
+
+static int mysql_test_handler_read(Prepared_statement *stmt,
+                                   TABLE_LIST *tables)
+{
+  THD *thd= stmt->thd;
+  LEX *lex= stmt->lex;
+  SQL_HANDLER *ha_table;
+  DBUG_ENTER("mysql_test_select");
+
+  lex->select_lex.context.resolve_in_select_list= TRUE;
+
+  /*
+    We don't have to test for permissions as this is already done during
+    HANDLER OPEN
+  */
+  if (!(ha_table= mysql_ha_read_prepare(thd, tables, lex->ha_read_mode,
+                                        lex->ident.str,
+                                        lex->insert_list,
+                                        lex->select_lex.where)))
+    DBUG_RETURN(1);
+
+  if (!stmt->is_sql_prepare())
+  {
+    if (!lex->result && !(lex->result= new (stmt->mem_root) select_send))
+    {
+      my_error(ER_OUTOFMEMORY, MYF(0), sizeof(select_send));
+      DBUG_RETURN(1);
+    }
+    if (send_prep_stmt(stmt, ha_table->fields.elements) ||
+        lex->result->send_fields(ha_table->fields, Protocol::SEND_EOF) ||
+        thd->protocol->flush())
+      DBUG_RETURN(1);
+    DBUG_RETURN(2);
+  }
+  DBUG_RETURN(0);
+}
+
 
 /**
   Perform semantic analysis of the parsed tree and send a response packet
@@ -1949,6 +2002,14 @@ static bool check_prepared_statement(Prepared_statement *stmt)
     res= mysql_test_insert_select(stmt, tables);
     break;
 
+  case SQLCOM_HA_READ:
+    res= mysql_test_handler_read(stmt, tables);
+    {
+      /* Statement and field info has already been sent */
+      DBUG_RETURN(FALSE);
+    }
+    break;
+
     /*
       Note that we don't need to have cases in this list if they are
       marked with CF_STATUS_COMMAND in sql_command_flags
diff --git a/sql/sql_rename.cc b/sql/sql_rename.cc
index e85e730db5b..5c8d1add04b 100644
--- a/sql/sql_rename.cc
+++ b/sql/sql_rename.cc
@@ -19,7 +19,7 @@
 
 #include "mysql_priv.h"
 #include "sql_trigger.h"
-
+#include "sql_handler.h"
 
 static TABLE_LIST *rename_tables(THD *thd, TABLE_LIST *table_list,
 				 bool skip_error);
diff --git a/sql/sql_show.cc b/sql/sql_show.cc
index cdfb982b372..61f3709476c 100644
--- a/sql/sql_show.cc
+++ b/sql/sql_show.cc
@@ -1942,8 +1942,7 @@ void mysqld_list_processes(THD *thd,const char *user, bool verbose)
           pthread_mutex_lock(&mysys_var->mutex);
         thd_info->proc_info= (char*) (tmp->killed == THD::KILL_CONNECTION? "Killed" : 0);
 #ifndef EMBEDDED_LIBRARY
-        thd_info->state_info= (char*) (tmp->locked ? "Locked" :
-                                       tmp->net.reading_or_writing ?
+        thd_info->state_info= (char*) (tmp->net.reading_or_writing ?
                                        (tmp->net.reading_or_writing == 2 ?
                                         "Writing to net" :
                                         thd_info->command == COM_SLEEP ? "" :
@@ -2068,8 +2067,7 @@ int fill_schema_processlist(THD* thd, TABLE_LIST* tables, COND* cond)
       table->field[5]->store(utime / 1000000, TRUE);
       /* STATE */
 #ifndef EMBEDDED_LIBRARY
-      val= (char*) (tmp->locked ? "Locked" :
-                    tmp->net.reading_or_writing ?
+      val= (char*) (tmp->net.reading_or_writing ?
                     (tmp->net.reading_or_writing == 2 ?
                      "Writing to net" :
                      tmp->command == COM_SLEEP ? "" :
diff --git a/sql/sql_table.cc b/sql/sql_table.cc
index 05a9c66cf80..b0f5c839412 100644
--- a/sql/sql_table.cc
+++ b/sql/sql_table.cc
@@ -24,6 +24,7 @@
 #include "sql_trigger.h"
 #include "sql_show.h"
 #include "debug_sync.h"
+#include "sql_handler.h"
 
 #ifdef __WIN__
 #include <io.h>
diff --git a/sql/sql_trigger.cc b/sql/sql_trigger.cc
index 1c4add27e57..8b95a36bc35 100644
--- a/sql/sql_trigger.cc
+++ b/sql/sql_trigger.cc
@@ -19,6 +19,7 @@
 #include "sp_head.h"
 #include "sql_trigger.h"
 #include "parse_file.h"
+#include "sql_handler.h"
 
 /*************************************************************************/