BUG#5390 - problems with merge tables

After-fix optimizations proposed and finally
implemented by Monty.

mysql-test/r/lock.result

@@ -47,3 +47,14 @@ unlock tables;
 lock tables t1 write, t1 as t1_alias read;
 insert into t1 select index1,nr from t1 as t1_alias;
 drop table t1,t2;
+create table t1 (c1 int);
+create table t2 (c1 int);
+create table t3 (c1 int);
+lock tables t1 write, t2 write, t3 write;
+drop table t2, t3, t1;
+create table t1 (c1 int);
+create table t2 (c1 int);
+create table t3 (c1 int);
+lock tables t1 write, t2 write, t3 write, t1 as t4 read;
+alter table t2 add column c2 int;
+drop table t1, t2, t3;

mysql-test/t/lock.test

@@ -57,3 +57,23 @@ unlock tables;
 lock tables t1 write, t1 as t1_alias read;
 insert into t1 select index1,nr from t1 as t1_alias;
 drop table t1,t2;
+
+#
+# BUG#5390 - problems with merge tables
+# Supplement test for the after-fix optimization
+# Check that a dropped table is correctly removed from a lock.
+create table t1 (c1 int);
+create table t2 (c1 int);
+create table t3 (c1 int);
+lock tables t1 write, t2 write, t3 write;
+# This removes one table after the other from the lock.
+drop table t2, t3, t1;
+#
+# Check that a lock merge works.
+create table t1 (c1 int);
+create table t2 (c1 int);
+create table t3 (c1 int);
+lock tables t1 write, t2 write, t3 write, t1 as t4 read;
+alter table t2 add column c2 int;
+drop table t1, t2, t3;
+

sql/lock.cc

@@ -72,13 +72,16 @@
 
 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,
-				 bool unlock, TABLE **write_locked);
+				 uint flags, TABLE **write_locked);
 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);
 
-
 /*
   Lock tables.
 
@@ -104,7 +107,8 @@ MYSQL_LOCK *mysql_lock_tables(THD *thd, TABLE **tables, uint count, uint flags)
 
   for (;;)
   {
-    if (!(sql_lock = get_lock_data(thd,tables,count, 0,&write_lock_used)))
+    if (! (sql_lock= get_lock_data(thd, tables, count, GET_LOCK_STORE_LOCKS,
+                                   &write_lock_used)))
       break;
 
     if (global_read_lock && write_lock_used &&
@@ -234,7 +238,8 @@ void mysql_unlock_some_tables(THD *thd, TABLE **table,uint count)
 {
   MYSQL_LOCK *sql_lock;
   TABLE *write_lock_used;
-  if ((sql_lock = get_lock_data(thd, table, count, 1, &write_lock_used)))
+  if ((sql_lock= get_lock_data(thd, table, count, GET_LOCK_UNLOCK,
+                               &write_lock_used)))
     mysql_unlock_tables(thd, sql_lock);
 }
 
@@ -271,6 +276,7 @@ void mysql_unlock_read_tables(THD *thd, MYSQL_LOCK *sql_lock)
   TABLE **table=sql_lock->table;
   for (i=found=0 ; i < sql_lock->table_count ; i++)
   {
+    DBUG_ASSERT(sql_lock->table[i]->lock_position == i);
     if ((uint) sql_lock->table[i]->reginfo.lock_type >= TL_WRITE_ALLOW_READ)
     {
       swap(TABLE *,*table,sql_lock->table[i]);
@@ -284,6 +290,17 @@ void mysql_unlock_read_tables(THD *thd, MYSQL_LOCK *sql_lock)
     VOID(unlock_external(thd,table,i-found));
     sql_lock->table_count=found;
   }
+  /* Fix the lock positions in TABLE */
+  table= sql_lock->table;
+  found= 0;
+  for (i= 0; i < sql_lock->table_count; i++)
+  {
+    TABLE *tbl= *table;
+    tbl->lock_position= table - sql_lock->table;
+    tbl->lock_data_start= found;
+    found+= tbl->lock_count;
+    table++;
+  }
   DBUG_VOID_RETURN;
 }
 
@@ -299,20 +316,51 @@ void mysql_lock_remove(THD *thd, MYSQL_LOCK *locked,TABLE *table)
     {
       if (locked->table[i] == table)
       {
-	locked->table_count--;
+        uint  j, removed_locks, old_tables;
+        TABLE *tbl;
+        uint lock_data_end;
+
+        DBUG_ASSERT(table->lock_position == i);
+
+        /* Decrement table_count in advance, making below expressions easier */
+        old_tables= --locked->table_count;
+
+        /* The table has 'removed_locks' lock data elements in locked->locks */
+        removed_locks= table->lock_count;
+
+        /* Move down all table pointers above 'i'. */
 	bmove((char*) (locked->table+i),
 	      (char*) (locked->table+i+1),
-	      (locked->table_count-i)* sizeof(TABLE*));
-	break;
+	      (old_tables - i) * sizeof(TABLE*));
+
+        lock_data_end= table->lock_data_start + table->lock_count;
+        /* Move down all lock data pointers above 'table->lock_data_end-1' */
+        bmove((char*) (locked->locks + table->lock_data_start),
+              (char*) (locked->locks + lock_data_end),
+              (locked->lock_count - lock_data_end) *
+              sizeof(THR_LOCK_DATA*));
+
+        /*
+          Fix moved table elements.
+          lock_position is the index in the 'locked->table' array,
+          it must be fixed by one.
+          table->lock_data_start is pointer to the lock data for this table
+          in the 'locked->locks' array, they must be fixed by 'removed_locks',
+          the lock data count of the removed table.
+        */
+        for (j= i ; j < old_tables; j++)
+        {
+          tbl= locked->table[j];
+          tbl->lock_position--;
+          DBUG_ASSERT(tbl->lock_position == j);
+          tbl->lock_data_start-= removed_locks;
         }
+
+        /* Finally adjust lock_count. */
+        locked->lock_count-= removed_locks;
+	break;
       }
-    THR_LOCK_DATA **prev=locked->locks;
-    for (i=0 ; i < locked->lock_count ; i++)
-    {
-      if (locked->locks[i]->type != TL_UNLOCK)
-	*prev++ = locked->locks[i];
     }
-    locked->lock_count=(uint) (prev - locked->locks);
   }
 }
 
@@ -322,7 +370,8 @@ void mysql_lock_abort(THD *thd, TABLE *table)
 {
   MYSQL_LOCK *locked;
   TABLE *write_lock_used;
-  if ((locked = get_lock_data(thd,&table,1,1,&write_lock_used)))
+  if ((locked= get_lock_data(thd, &table, 1, GET_LOCK_UNLOCK,
+                             &write_lock_used)))
   {
     for (uint i=0; i < locked->lock_count; i++)
       thr_abort_locks(locked->locks[i]->lock);
@@ -351,7 +400,8 @@ bool mysql_lock_abort_for_thread(THD *thd, TABLE *table)
   bool result= FALSE;
   DBUG_ENTER("mysql_lock_abort_for_thread");
 
-  if ((locked = get_lock_data(thd,&table,1,1,&write_lock_used)))
+  if ((locked= get_lock_data(thd, &table, 1, GET_LOCK_UNLOCK,
+                             &write_lock_used)))
   {
     for (uint i=0; i < locked->lock_count; i++)
     {
@@ -368,7 +418,9 @@ bool mysql_lock_abort_for_thread(THD *thd, TABLE *table)
 MYSQL_LOCK *mysql_lock_merge(MYSQL_LOCK *a,MYSQL_LOCK *b)
 {
   MYSQL_LOCK *sql_lock;
+  TABLE **table, **end_table;
   DBUG_ENTER("mysql_lock_merge");
+
   if (!(sql_lock= (MYSQL_LOCK*)
 	my_malloc(sizeof(*sql_lock)+
 		  sizeof(THR_LOCK_DATA*)*(a->lock_count+b->lock_count)+
@@ -384,6 +436,21 @@ MYSQL_LOCK *mysql_lock_merge(MYSQL_LOCK *a,MYSQL_LOCK *b)
   memcpy(sql_lock->table,a->table,a->table_count*sizeof(*a->table));
   memcpy(sql_lock->table+a->table_count,b->table,
 	 b->table_count*sizeof(*b->table));
+
+  /*
+    Now adjust lock_position and lock_data_start for all objects that was
+    moved in 'b' (as there is now all objects in 'a' before these).
+  */
+  for (table= sql_lock->table + a->table_count,
+         end_table= table + b->table_count;
+       table < end_table;
+       table++)
+  {
+    (*table)->lock_position+=   a->table_count;
+    (*table)->lock_data_start+= a->lock_count;
+  }
+
+  /* Delete old, not needed locks */
   my_free((gptr) a,MYF(0));
   my_free((gptr) b,MYF(0));
   DBUG_RETURN(sql_lock);
@@ -402,73 +469,78 @@ MYSQL_LOCK *mysql_lock_merge(MYSQL_LOCK *a,MYSQL_LOCK *b)
   NOTE
     This is mainly meant for MERGE tables in INSERT ... SELECT
     situations. The 'real', underlying tables can be found only after
-    the table is opened. The easier way is to check this after the
-    tables are locked.
+    the MERGE tables are opened. This function assumes that the tables are
+    already locked.
+
+    Temporary tables are ignored here like they are ignored in
+    get_lock_data(). If we allow two opens on temporary tables later,
+    both functions should be checked.
 
   RETURN
     1           A table from 'tables' matches a lock on 'table'.
-    0           No duplicate lock is present.
-    -1          Error.
+    0           No duplicate lock found.
 */
 
 int mysql_lock_have_duplicate(THD *thd, TABLE *table, TABLE_LIST *tables)
 {
-  uint                  count;
-  MYSQL_LOCK            *sql_lock1;
-  MYSQL_LOCK            *sql_lock2;
-  TABLE                 **tables1= &table;
-  TABLE                 **tables2;
-  TABLE                 **table_ptr;
-  TABLE_LIST            *tablist2;
-  TABLE                 *write_lock_used;
-  THR_LOCK_DATA         **lock_data1;
-  THR_LOCK_DATA         **end_data1;
+  MYSQL_LOCK            *mylock;
+  TABLE                 **lock_tables;
+  TABLE                 *table2;
+  THR_LOCK_DATA         **lock_locks, **table_lock_data;
+  THR_LOCK_DATA         **end_data;
   THR_LOCK_DATA         **lock_data2;
   THR_LOCK_DATA         **end_data2;
-  THR_LOCK              *lock1;
   DBUG_ENTER("mysql_lock_have_duplicate");
 
-  if (! (sql_lock1= get_lock_data(thd, tables1, 1, 1, &write_lock_used)))
-    goto err0;
-
-  count=0;
-  for (tablist2 = tables; tablist2; tablist2= tablist2->next)
-    count++;
-  if (! (tables2= (TABLE**) sql_alloc(sizeof(TABLE*) * count)))
-    goto err1;
-  table_ptr= tables2;
-  for (tablist2 = tables; tablist2; tablist2= tablist2->next)
-    *(table_ptr++)= tablist2->table;
-  if (! (sql_lock2= get_lock_data(thd, tables2, count, 1, &write_lock_used)))
-    goto err1;
-
-  count= 1;
-  for (lock_data1= sql_lock1->locks,
-         end_data1= lock_data1 + sql_lock1->lock_count;
-       lock_data1 < end_data1;
-       lock_data1++)
-  {
-    lock1= (*lock_data1)->lock;
-    for (lock_data2= sql_lock2->locks,
-           end_data2= lock_data2 + sql_lock2->lock_count;
+  /* A temporary table does not have locks. */
+  if (table->tmp_table == TMP_TABLE)
+    goto end;
+
+  /* Get command lock or LOCK TABLES lock. */
+  mylock= thd->lock ? thd->lock : thd->locked_tables;
+  DBUG_ASSERT(mylock);
+
+  /* If we have less than two tables, we cannot have duplicates. */
+  if (mylock->table_count < 2)
+    goto end;
+
+  lock_locks=  mylock->locks;
+  lock_tables= mylock->table;
+
+  /* Prepare table related variables that don't change in loop. */
+  DBUG_ASSERT(table == lock_tables[table->lock_position]);
+  table_lock_data= lock_locks + table->lock_data_start;
+  end_data= table_lock_data + table->lock_count;
+
+  for (; tables; tables= tables->next)
+  {
+    table2= tables->table;
+    if (table2->tmp_table == TMP_TABLE)
+      continue;
+
+    /* All tables in list must be in lock. */
+    DBUG_ASSERT(table2 == lock_tables[table2->lock_position]);
+
+    for (lock_data2=  lock_locks + table2->lock_data_start,
+           end_data2= lock_data2 + table2->lock_count;
          lock_data2 < end_data2;
          lock_data2++)
     {
-      if ((*lock_data2)->lock == lock1)
-        goto end;
+      THR_LOCK_DATA **lock_data;
+      THR_LOCK *lock2= (*lock_data2)->lock;
+
+      for (lock_data= table_lock_data;
+           lock_data < end_data;
+           lock_data++)
+      {
+        if ((*lock_data)->lock == lock2)
+          DBUG_RETURN(1);
+      }
     }
   }
-  count= 0;
 
  end:
-  my_free((gptr) sql_lock2, MYF(0));
-  my_free((gptr) sql_lock1, MYF(0));
-  DBUG_RETURN(count);
-
- err1:
-  my_free((gptr) sql_lock1, MYF(0));
- err0:
-  DBUG_RETURN(-1);
+  DBUG_RETURN(0);
 }
 
 
@@ -497,17 +569,27 @@ static int unlock_external(THD *thd, TABLE **table,uint count)
 
 
 /*
-** Get lock structures from table structs and initialize locks
+  Get lock structures from table structs and initialize locks
+
+  SYNOPSIS
+    get_lock_data()
+    thd			Thread handler
+    table_ptr		Pointer to tables that should be locks
+    flags		One of:
+			GET_LOCK_UNLOCK:      If we should send TL_IGNORE to
+                        		      store lock
+			GET_LOCK_STORE_LOCKS: Store lock info in TABLE
+    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,
-				 bool get_old_locks, TABLE **write_lock_used)
+				 uint flags, TABLE **write_lock_used)
 {
   uint i,tables,lock_count;
   MYSQL_LOCK *sql_lock;
-  THR_LOCK_DATA **locks;
-  TABLE **to;
+  THR_LOCK_DATA **locks, **locks_buf, **locks_start;
+  TABLE **to, **table_buf;
   DBUG_ENTER("get_lock_data");
 
   *write_lock_used=0;
@@ -525,18 +607,19 @@ static MYSQL_LOCK *get_lock_data(THD *thd, TABLE **table_ptr, uint count,
 		  sizeof(THR_LOCK_DATA*)*tables+sizeof(table_ptr)*lock_count,
 		  MYF(0))))
     DBUG_RETURN(0);
-  locks=sql_lock->locks=(THR_LOCK_DATA**) (sql_lock+1);
-  to=sql_lock->table=(TABLE**) (locks+tables);
+  locks= locks_buf= sql_lock->locks= (THR_LOCK_DATA**) (sql_lock + 1);
+  to= table_buf= sql_lock->table= (TABLE**) (locks + tables);
   sql_lock->table_count=lock_count;
   sql_lock->lock_count=tables;
 
   for (i=0 ; i < count ; i++)
   {
     TABLE *table;
+    enum thr_lock_type lock_type;
+
     if ((table=table_ptr[i])->tmp_table == TMP_TABLE)
       continue;
-    *to++=table;
-    enum thr_lock_type lock_type= table->reginfo.lock_type;
+    lock_type= table->reginfo.lock_type;
     if (lock_type >= TL_WRITE_ALLOW_WRITE)
     {
       *write_lock_used=table;
@@ -547,12 +630,22 @@ static MYSQL_LOCK *get_lock_data(THD *thd, TABLE **table_ptr, uint count,
 	DBUG_RETURN(0);
       }
     }
-    locks=table->file->store_lock(thd, locks, get_old_locks ? TL_IGNORE :
+    locks_start= locks;
+    locks= table->file->store_lock(thd, locks,
+                                   (flags & GET_LOCK_UNLOCK) ? TL_IGNORE :
                                    lock_type);
+    if (flags & GET_LOCK_STORE_LOCKS)
+    {
+      table->lock_position=   (uint) (to - table_buf);
+      table->lock_data_start= (uint) (locks_start - locks_buf);
+      table->lock_count=      (uint) (locks - locks_start);
+    }
+    *to++= table;
   }
   DBUG_RETURN(sql_lock);
 }
 
+
 /*****************************************************************************
 **  Lock table based on the name.
 **  This is used when we need total access to a closed, not open table

sql/table.h

@@ -86,6 +86,9 @@ struct st_table {
   uint next_number_index;
   uint blob_ptr_size;			/* 4 or 8 */
   uint next_number_key_offset;
+  uint lock_position;                   /* Position in MYSQL_LOCK.table */
+  uint lock_data_start;                 /* Start pos. in MYSQL_LOCK.locks */
+  uint lock_count;                      /* Number of locks */
   int current_lock;			/* Type of lock on table */
   enum tmp_table_type tmp_table;
   my_bool copy_blobs;			/* copy_blobs when storing */