Added support for a LOCK TABLES...WHERE ENGINE = InnoDB query which sets

transactional table locks to tables mentioned in the query. These locks
are released at the end of the transaction automatically.
This is fix for bugs #5655, #5998 and issue #3762.

6 files changed, 57 insertions, 15 deletions

diff --git a/innobase/include/lock0lock.h b/innobase/include/lock0lock.h
index d642fe46fef..2199ba36881 100644
--- a/innobase/include/lock0lock.h
+++ b/innobase/include/lock0lock.h
@@ -585,6 +585,8 @@ extern lock_sys_t*	lock_sys;
 #define LOCK_TABLE	16	/* these type values should be so high that */
 #define	LOCK_REC	32	/* they can be ORed to the lock mode */
 #define LOCK_TABLE_EXP	80	/* explicit table lock (80 = 16 + 64) */
+#define	LOCK_TABLE_TRANSACTIONAL	144
+				/* transactional table lock (144 = 16 + 128)*/
 #define LOCK_TYPE_MASK	0xF0UL	/* mask used to extract lock type from the
 				type_mode field in a lock */
 /* Waiting lock flag */
diff --git a/innobase/include/trx0trx.h b/innobase/include/trx0trx.h
index 8eb71dac763..e305226f9e3 100644
--- a/innobase/include/trx0trx.h
+++ b/innobase/include/trx0trx.h
@@ -464,6 +464,10 @@ struct trx_struct{
 	ulint		n_lock_table_exp;/* number of explicit table locks
 					(LOCK TABLES) reserved by the
 					transaction, stored in trx_locks */
+	ulint		n_lock_table_transactional;
+					/* number of transactional table locks
+					(LOCK TABLES..WHERE ENGINE) reserved by
+					the transaction, stored in trx_locks */
 	UT_LIST_NODE_T(trx_t)
 			trx_list;	/* list of transactions */
 	UT_LIST_NODE_T(trx_t)
diff --git a/innobase/lock/lock0lock.c b/innobase/lock/lock0lock.c
index d2d16a1ae4e..1ff16bce45a 100644
--- a/innobase/lock/lock0lock.c
+++ b/innobase/lock/lock0lock.c
@@ -2207,7 +2207,8 @@ lock_grant(
                 release it at the end of the SQL statement */
 
                 lock->trx->auto_inc_lock = lock;
-        } else if (lock_get_type(lock) == LOCK_TABLE_EXP) {
+        } else if (lock_get_type(lock) == LOCK_TABLE_EXP ||
+			lock_get_type(lock) == LOCK_TABLE_TRANSACTIONAL) {
 		ut_a(lock_get_mode(lock) == LOCK_S
 			|| lock_get_mode(lock) == LOCK_X);
 	}
@@ -3421,6 +3422,10 @@ lock_table_create(
 		lock->trx->n_lock_table_exp++;
 	}
 
+	if (lock_get_type(lock) == LOCK_TABLE_TRANSACTIONAL) {
+		lock->trx->n_lock_table_transactional++;
+	}
+
 	lock->un_member.tab_lock.table = table;
 
 	UT_LIST_ADD_LAST(un_member.tab_lock.locks, table->locks, lock);
@@ -3458,7 +3463,11 @@ lock_table_remove_low(
 	}
 
 	if (lock_get_type(lock) == LOCK_TABLE_EXP) {
-		lock->trx->n_lock_table_exp--;
+		trx->n_lock_table_exp--;
+	}
+
+        if (lock_get_type(lock) == LOCK_TABLE_TRANSACTIONAL) {
+		trx->n_lock_table_transactional--;
 	}
 
 	UT_LIST_REMOVE(trx_locks, trx->trx_locks, lock);
@@ -3592,7 +3601,8 @@ lock_table(
 				DB_DEADLOCK, or DB_QUE_THR_SUSPENDED */
 	ulint		flags,	/* in: if BTR_NO_LOCKING_FLAG bit is set,
 				does nothing;
-				if LOCK_TABLE_EXP bits are set,
+				if LOCK_TABLE_EXP|LOCK_TABLE_TRANSACTIONAL
+				bits are set,
 				creates an explicit table lock */
 	dict_table_t*	table,	/* in: database table in dictionary cache */
 	ulint		mode,	/* in: lock mode */
@@ -3608,7 +3618,8 @@ lock_table(
 		return(DB_SUCCESS);
 	}
 
-	ut_a(flags == 0 || flags == LOCK_TABLE_EXP);
+	ut_a(flags == 0 || flags == LOCK_TABLE_EXP || 
+					flags == LOCK_TABLE_TRANSACTIONAL);
 
 	trx = thr_get_trx(thr);
 
@@ -3722,7 +3733,8 @@ lock_table_dequeue(
 	ut_ad(mutex_own(&kernel_mutex));
 #endif /* UNIV_SYNC_DEBUG */
 	ut_a(lock_get_type(in_lock) == LOCK_TABLE ||
-		lock_get_type(in_lock) == LOCK_TABLE_EXP);
+		lock_get_type(in_lock) == LOCK_TABLE_EXP ||
+		lock_get_type(in_lock) == LOCK_TABLE_TRANSACTIONAL);
 
 	lock = UT_LIST_GET_NEXT(un_member.tab_lock.locks, in_lock);
 
@@ -3826,7 +3838,9 @@ lock_release_off_kernel(
 			}
 
 			lock_table_dequeue(lock);
-			if (lock_get_type(lock) == LOCK_TABLE_EXP) {
+
+			if (lock_get_type(lock) == LOCK_TABLE_EXP ||
+			    lock_get_type(lock) == LOCK_TABLE_TRANSACTIONAL) {
 				ut_a(lock_get_mode(lock) == LOCK_S
 					|| lock_get_mode(lock) == LOCK_X);
 			}
@@ -3850,6 +3864,7 @@ lock_release_off_kernel(
 
 	ut_a(trx->auto_inc_lock == NULL);
 	ut_a(trx->n_lock_table_exp == 0);
+	ut_a(trx->n_lock_table_transactional == 0);
 }
 
 /*************************************************************************
@@ -3915,6 +3930,7 @@ lock_release_tables_off_kernel(
 	}
 
 	ut_a(trx->n_lock_table_exp == 0);
+	ut_a(trx->n_lock_table_transactional == 0);
 }
 
 /*************************************************************************
@@ -4028,11 +4044,15 @@ lock_table_print(
 	ut_ad(mutex_own(&kernel_mutex));
 #endif /* UNIV_SYNC_DEBUG */
 	ut_a(lock_get_type(lock) == LOCK_TABLE ||
-		lock_get_type(lock) == LOCK_TABLE_EXP);
+		lock_get_type(lock) == LOCK_TABLE_EXP ||
+		lock_get_type(lock) == LOCK_TABLE_TRANSACTIONAL);
 
 	if (lock_get_type(lock) == LOCK_TABLE_EXP) {
 		fputs("EXPLICIT ", file);
+	} else if (lock_get_type(lock) == LOCK_TABLE_TRANSACTIONAL) {
+		fputs("TRANSACTIONAL ", file);
 	}
+
 	fputs("TABLE LOCK table ", file);
 	ut_print_name(file, lock->trx, lock->un_member.tab_lock.table->name);
 	fprintf(file, " trx id %lu %lu",
diff --git a/innobase/row/row0mysql.c b/innobase/row/row0mysql.c
index be243b44488..79a4ffb36ae 100644
--- a/innobase/row/row0mysql.c
+++ b/innobase/row/row0mysql.c
@@ -784,7 +784,7 @@ row_lock_table_for_mysql(
 					table handle */
 	dict_table_t*	table,		/* in: table to lock, or NULL
 					if prebuilt->table should be
-					locked as LOCK_TABLE_EXP |
+					locked or a
 					prebuilt->select_lock_type */
 	ulint		mode)		/* in: lock mode of table */
 {
@@ -822,8 +822,14 @@ run_again:
 	if (table) {
 		err = lock_table(0, table, mode, thr);
 	} else {
-		err = lock_table(LOCK_TABLE_EXP, prebuilt->table,
-			prebuilt->select_lock_type, thr);
+		if (mode == LOCK_TABLE_TRANSACTIONAL) {
+			err = lock_table(LOCK_TABLE_TRANSACTIONAL, 
+					prebuilt->table,
+					prebuilt->select_lock_type, thr);
+		} else {
+			err = lock_table(LOCK_TABLE_EXP, prebuilt->table,
+					prebuilt->select_lock_type, thr);
+		}
 	}
 
 	trx->error_state = err;
diff --git a/innobase/trx/trx0roll.c b/innobase/trx/trx0roll.c
index db5e16c7778..105fa97080f 100644
--- a/innobase/trx/trx0roll.c
+++ b/innobase/trx/trx0roll.c
@@ -335,8 +335,10 @@ undo log. If the transaction was not yet committed, then we roll it back.
 Note: this is done in a background thread */
 
 void *
-trx_rollback_or_clean_all_without_sess(void *i)
-/*========================================*/
+trx_rollback_or_clean_all_without_sess(
+/*===================================*/
+			/* out: arguments */
+	void	*i)	/* in: arguments (unused) */
 {
 	mem_heap_t*	heap;
 	que_fork_t*	fork;
@@ -496,6 +498,7 @@ loop:
 	goto loop;
 
 	os_thread_exit(i); /* not reached */
+	return(i);
 }
 	
 /***********************************************************************
diff --git a/innobase/trx/trx0trx.c b/innobase/trx/trx0trx.c
index ab8bd898dd6..f676c21934b 100644
--- a/innobase/trx/trx0trx.c
+++ b/innobase/trx/trx0trx.c
@@ -153,6 +153,7 @@ trx_create(
 
 	trx->auto_inc_lock = NULL;
 	trx->n_lock_table_exp = 0;
+	trx->n_lock_table_transactional = 0;
 	
 	trx->read_view_heap = mem_heap_create(256);
 	trx->read_view = NULL;
@@ -285,6 +286,7 @@ trx_free(
 	ut_a(!trx->has_search_latch);
 	ut_a(!trx->auto_inc_lock);
 	ut_a(!trx->n_lock_table_exp);
+	ut_a(!trx->n_lock_table_transactional);
 
 	ut_a(trx->dict_operation_lock_mode == 0);
 
@@ -1645,10 +1647,15 @@ trx_print(
 	putc('\n', f);
   	
         if (trx->n_mysql_tables_in_use > 0 || trx->mysql_n_tables_locked > 0) {
+		fprintf(f, "mysql tables in use %lu, locked %lu\n",
+					(ulong) trx->n_mysql_tables_in_use,
+					(ulong) trx->mysql_n_tables_locked);
+	}
 
-                fprintf(f, "mysql tables in use %lu, locked %lu\n",
-                           (ulong) trx->n_mysql_tables_in_use,
-                           (ulong) trx->mysql_n_tables_locked);
+	if (trx->n_lock_table_transactional > 0 || trx->n_lock_table_exp > 0) {
+fprintf(f, "mysql explicit table locks %lu, transactional table locks %lu\n",
+				(ulong) trx->n_lock_table_exp,
+				(ulong) trx->n_lock_table_transactional);
         }
 
 	newline = TRUE;