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.




innobase/include/lock0lock.h:
  Added a new lock type LOCK_TABLE_TRANSACTIONAL.
innobase/include/trx0trx.h:
  Added a varible to store the number of transactional table locks
  reserved by the transaction and stored in the trx_locks list.
innobase/lock/lock0lock.c:
  Added a lock type LOCK_TABLE_TRANSACTIONAL.
innobase/row/row0mysql.c:
  Added support for a transactional table lock.
innobase/trx/trx0roll.c:
  Make compiler happy and format function.
innobase/trx/trx0trx.c:
  Added information about number of explicit and transactional table locks in
  the innodb monitor.
sql/ha_innodb.cc:
  Added support for a LOCK_TABLE_TRANSACTIONAL table lock (SQLCOM_LOCK_TABLES_TRANSACTIONAL).
  Transactional table locks are taken in the function ::transactional_table_lock().
sql/ha_innodb.h:
  Added prototype for a function trandactional_table_lock().
sql/handler.h:
  Added prototype for a virtual function transactional_table_lock().
  Handler does not need to implement this function.
sql/lock.cc:
  Added a function transactional_lock_tables() to lock all tables in the list
  with transactional table lock. These locks can cause a deadlock and
  this error should be reported back to user.
sql/mysql_priv.h:
  Added prototype for a function transactional_lock_tables().
sql/sql_lex.h:
  Added SQLCOM_LOCK_TABLES_TRANSACTIONAL command.
sql/sql_parse.cc:
  Added SQLCOM_LOCK_TABLES_TRANSACTIONAL command.
sql/sql_yacc.yy:
  Added parsing rules for a LOCK TABLES...WHERE ENGINE = x;

8 files changed, 191 insertions, 5 deletions

diff --git a/sql/ha_innodb.cc b/sql/ha_innodb.cc
index 8713d145f2f..3b1af3c58e4 100644
--- a/sql/ha_innodb.cc
+++ b/sql/ha_innodb.cc
@@ -1327,7 +1327,8 @@ innobase_commit(
 	3. innobase_query_caching_of_table_permitted(),
 	4. innobase_savepoint(),
 	5. ::init_table_handle_for_HANDLER(),
-	6. innobase_start_trx_and_assign_read_view()
+	6. innobase_start_trx_and_assign_read_view(),
+	7. ::transactional_table_lock()
 
 	and it is only set to 0 in a commit or a rollback. If it is 0 we know
 	there cannot be resources to be freed and we could return immediately.
@@ -5095,8 +5096,9 @@ ha_innobase::start_stmt(
 			select_lock_type value. The value of
 			stored_select_lock_type was decided in:
 			1) ::store_lock(),
-			2) ::external_lock(), and
-			3) ::init_table_handle_for_HANDLER(). */
+			2) ::external_lock(), 
+			3) ::init_table_handle_for_HANDLER(), and 
+			4) :.transactional_table_lock(). */
 
 			prebuilt->select_lock_type =
 				prebuilt->stored_select_lock_type;
@@ -5287,6 +5289,94 @@ ha_innobase::external_lock(
 	DBUG_RETURN(0);
 }
 
+/**********************************************************************
+With this function MySQL request a transactional lock to a table when
+user issued query LOCK TABLES..WHERE ENGINE = InnoDB. */
+
+int
+ha_innobase::transactional_table_lock(
+/*==================================*/
+			        /* out: 0 */
+	THD*	thd,		/* in: handle to the user thread */
+	int 	lock_type)	/* in: lock type */
+{
+	row_prebuilt_t* prebuilt = (row_prebuilt_t*) innobase_prebuilt;
+	trx_t*		trx;
+
+  	DBUG_ENTER("ha_innobase::transactional_table_lock");
+	DBUG_PRINT("enter",("lock_type: %d", lock_type));
+
+	/* We do not know if MySQL can call this function before calling
+	external_lock(). To be safe, update the thd of the current table
+	handle. */
+
+	update_thd(thd);
+
+ 	if (prebuilt->table->ibd_file_missing && !current_thd->tablespace_op) {
+	        ut_print_timestamp(stderr);
+	        fprintf(stderr, "  InnoDB error:\n"
+"MySQL is trying to use a table handle but the .ibd file for\n"
+"table %s does not exist.\n"
+"Have you deleted the .ibd file from the database directory under\n"
+"the MySQL datadir, or have you used DISCARD TABLESPACE?\n"
+"Look from section 15.1 of http://www.innodb.com/ibman.html\n"
+"how you can resolve the problem.\n",
+				prebuilt->table->name);
+		DBUG_RETURN(HA_ERR_CRASHED);
+	}
+
+	trx = prebuilt->trx;
+
+	prebuilt->sql_stat_start = TRUE;
+	prebuilt->hint_need_to_fetch_extra_cols = 0;
+
+	prebuilt->read_just_key = 0;
+	prebuilt->keep_other_fields_on_keyread = FALSE;
+
+	if (lock_type == F_WRLCK) {
+		prebuilt->select_lock_type = LOCK_X;
+		prebuilt->stored_select_lock_type = LOCK_X;
+	} else if (lock_type == F_RDLCK) {
+		prebuilt->select_lock_type = LOCK_X;
+		prebuilt->stored_select_lock_type = LOCK_X;
+	} else {
+	        ut_print_timestamp(stderr);
+	        fprintf(stderr, "  InnoDB error:\n"
+"MySQL is trying to set transactional table lock with corrupted lock type\n"
+"to table %s, lock type %d does not exist.\n",
+				prebuilt->table->name, lock_type);
+		DBUG_RETURN(HA_ERR_CRASHED);
+	}
+
+	/* MySQL is setting a new transactional table lock */
+
+	/* Set the MySQL flag to mark that there is an active transaction */
+	thd->transaction.all.innodb_active_trans = 1;
+
+	if (thd->in_lock_tables && thd->variables.innodb_table_locks) {
+		ulint	error = DB_SUCCESS;
+
+		error = row_lock_table_for_mysql(prebuilt,NULL, 
+						LOCK_TABLE_TRANSACTIONAL);
+
+		if (error != DB_SUCCESS) {
+			error = convert_error_code_to_mysql(error, user_thd);
+			DBUG_RETURN(error);
+		}
+
+		if (thd->options & (OPTION_NOT_AUTOCOMMIT | OPTION_BEGIN)) {
+
+			/* Store the current undo_no of the transaction 
+			so that we know where to roll back if we have 
+			to roll back the next SQL statement */
+
+			trx_mark_sql_stat_end(trx);
+		}
+	}
+
+	DBUG_RETURN(0);
+}
+
 /****************************************************************************
 Here we export InnoDB status variables to MySQL.  */
 
diff --git a/sql/ha_innodb.h b/sql/ha_innodb.h
index a64bbc665c1..eb906a744da 100644
--- a/sql/ha_innodb.h
+++ b/sql/ha_innodb.h
@@ -149,6 +149,7 @@ class ha_innobase: public handler
 	int discard_or_import_tablespace(my_bool discard);
   	int extra(enum ha_extra_function operation);
   	int external_lock(THD *thd, int lock_type);
+	int transactional_table_lock(THD *thd, int lock_type);
 	int start_stmt(THD *thd);
 
   	void position(byte *record);
diff --git a/sql/handler.h b/sql/handler.h
index c70ea266734..c7131d75156 100644
--- a/sql/handler.h
+++ b/sql/handler.h
@@ -450,6 +450,7 @@ public:
   { return extra(operation); }
   virtual int reset() { return extra(HA_EXTRA_RESET); }
   virtual int external_lock(THD *thd, int lock_type)=0;
+  virtual int transactional_table_lock(THD *thd, int lock_type) {return 0;}
   virtual void unlock_row() {}
   virtual int start_stmt(THD *thd) {return 0;}
   /*
diff --git a/sql/lock.cc b/sql/lock.cc
index 3367c6a2900..0db23f54c50 100644
--- a/sql/lock.cc
+++ b/sql/lock.cc
@@ -695,6 +695,9 @@ static void print_lock_error(int error)
   case HA_ERR_READ_ONLY_TRANSACTION:
     textno=ER_READ_ONLY_TRANSACTION;
     break;
+  case HA_ERR_LOCK_DEADLOCK:
+    textno=ER_LOCK_DEADLOCK;
+    break;
   default:
     textno=ER_CANT_LOCK;
     break;
@@ -870,3 +873,62 @@ bool make_global_read_lock_block_commit(THD *thd)
   thd->exit_cond(old_message);
   DBUG_RETURN(error);
 }
+
+/*
+  Take transactional table lock for all tables in the list
+
+  SYNOPSIS
+    transactional_lock_tables
+    thd			Thread THD
+    tables              list of tables
+    counter             number of tables in the list
+
+  NOTES
+
+  RETURN
+    0  - OK
+    -1 - error
+
+*/
+int transactional_lock_tables(THD *thd, TABLE_LIST *tables, uint counter)
+{ 
+  uint i;
+  int lock_type,error=0;
+  TABLE_LIST *table;
+  TABLE **start,**ptr;
+
+  DBUG_ENTER("transactional_lock_tables");
+
+  if (!(ptr=start=(TABLE**) sql_alloc(sizeof(TABLE*) * counter)))
+      return -1;
+    
+  for (table= tables; table; table= table->next_global)
+  {
+    if (!table->placeholder() && !table->schema_table)
+      *(ptr++)= table->table;
+  }
+
+  for (i=1 ; i <= counter ; i++, start++)
+  {
+    DBUG_ASSERT((*start)->reginfo.lock_type >= TL_READ);
+    lock_type=F_WRLCK;				/* Lock exclusive */
+
+    if ((*start)->db_stat & HA_READ_ONLY ||
+       ((*start)->reginfo.lock_type >= TL_READ &&
+	(*start)->reginfo.lock_type <= TL_READ_NO_INSERT))
+      lock_type=F_RDLCK;
+
+    if ((error=(*start)->file->transactional_table_lock(thd, lock_type)))
+    {
+      print_lock_error(error);
+      DBUG_RETURN(-1);
+    }
+    else
+    {
+      (*start)->db_stat &= ~ HA_BLOCK_LOCK;
+      (*start)->current_lock= lock_type;
+    }
+  }
+
+  DBUG_RETURN(0);
+}
diff --git a/sql/mysql_priv.h b/sql/mysql_priv.h
index 09bec0a9323..f632dd8760f 100644
--- a/sql/mysql_priv.h
+++ b/sql/mysql_priv.h
@@ -827,6 +827,7 @@ int open_tables(THD *thd, TABLE_LIST *tables, uint *counter);
 int simple_open_n_lock_tables(THD *thd,TABLE_LIST *tables);
 bool open_and_lock_tables(THD *thd,TABLE_LIST *tables);
 int lock_tables(THD *thd, TABLE_LIST *tables, uint counter);
+int transactional_lock_tables(THD *thd, TABLE_LIST *tables, uint counter);
 TABLE *open_temporary_table(THD *thd, const char *path, const char *db,
 			    const char *table_name, bool link_in_list);
 bool rm_temporary_table(enum db_type base, char *path);
diff --git a/sql/sql_lex.h b/sql/sql_lex.h
index e99e6138b6d..fb1c6ee376b 100644
--- a/sql/sql_lex.h
+++ b/sql/sql_lex.h
@@ -87,6 +87,7 @@ enum enum_sql_command {
   SQLCOM_PREPARE, SQLCOM_EXECUTE, SQLCOM_DEALLOCATE_PREPARE,
   SQLCOM_CREATE_VIEW, SQLCOM_DROP_VIEW,
   SQLCOM_CREATE_TRIGGER, SQLCOM_DROP_TRIGGER,
+  SQLCOM_LOCK_TABLES_TRANSACTIONAL,
   /* This should be the last !!! */
   SQLCOM_END
 };
diff --git a/sql/sql_parse.cc b/sql/sql_parse.cc
index e27cd20e15e..f682b38641b 100644
--- a/sql/sql_parse.cc
+++ b/sql/sql_parse.cc
@@ -3252,6 +3252,27 @@ create_error:
       thd->options&= ~(ulong) (OPTION_TABLE_LOCK);
     thd->in_lock_tables=0;
     break;
+  case SQLCOM_LOCK_TABLES_TRANSACTIONAL:
+  {
+    uint counter = 0;
+
+    if (check_db_used(thd, all_tables))
+      goto error;
+    if (check_table_access(thd, LOCK_TABLES_ACL | SELECT_ACL, all_tables, 0))
+      goto error;
+
+    thd->in_lock_tables=1;
+    thd->options|= OPTION_TABLE_LOCK;
+
+    if (open_tables(thd, all_tables, &counter) == 0 &&
+        transactional_lock_tables(thd, all_tables, counter) == 0)
+        send_ok(thd);
+    else
+      thd->options&= ~(ulong) (OPTION_TABLE_LOCK);
+
+    thd->in_lock_tables=0;
+    break;
+  }
   case SQLCOM_CREATE_DB:
   {
     char *alias;
diff --git a/sql/sql_yacc.yy b/sql/sql_yacc.yy
index 74a96701bdc..1f467702056 100644
--- a/sql/sql_yacc.yy
+++ b/sql/sql_yacc.yy
@@ -7395,8 +7395,8 @@ lock:
 	{
 	  Lex->sql_command=SQLCOM_LOCK_TABLES;
 	}
-	table_lock_list
-	{}
+	table_lock_list lock_engine_opt
+	{} 
 	;
 
 table_or_tables:
@@ -7422,6 +7422,15 @@ lock_option:
 	| READ_SYM LOCAL_SYM { $$= TL_READ; }
         ;
 
+lock_engine_opt:
+	/* empty */	
+	| WHERE 
+	{
+	  Lex->sql_command=SQLCOM_LOCK_TABLES_TRANSACTIONAL;
+	}
+	ENGINE_SYM opt_equal storage_engines
+	;
+
 unlock:
 	UNLOCK_SYM table_or_tables { Lex->sql_command=SQLCOM_UNLOCK_TABLES; }
         ;