BUG#5390 - problems with merge tables

Problem #1: INSERT...SELECT, Version for 5.0.
Extended the unique table check by a check of lock data.
Merge sub-tables cannot be detected by doing name checks only.


mysql-test/r/create.result:
  BUG#5390 - problems with merge tables
  Removed a duplicate test.
mysql-test/r/merge.result:
  BUG#5390 - problems with merge tables
  Problem #1: INSERT...SELECT
  Added test results.
mysql-test/t/create.test:
  BUG#5390 - problems with merge tables
  Removed a duplicate test.
mysql-test/t/merge.test:
  BUG#5390 - problems with merge tables
  Problem #1: INSERT...SELECT
  Added tests.
sql/lock.cc:
  BUG#5390 - problems with merge tables
  Problem #1: INSERT...SELECT, Version for 5.0.
  Added a new function to find a duplicate lock in a list of tables.
sql/mysql_priv.h:
  BUG#5390 - problems with merge tables
  Problem #1: INSERT...SELECT, Version for 5.0.
  Added the thread handle to unique_table().
  Added a declaration for the new function.
sql/sql_base.cc:
  BUG#5390 - problems with merge tables
  Problem #1: INSERT...SELECT, Version for 5.0.
  Added the thread handle to unique_table().
  Added a call to the new mysql_lock_have_duplicate(),
  which needs the thread handle, to unique_table().
sql/sql_delete.cc:
  BUG#5390 - problems with merge tables
  Problem #1: INSERT...SELECT, Version for 5.0.
  Added the thread handle to unique_table().
sql/sql_insert.cc:
  BUG#5390 - problems with merge tables
  Problem #1: INSERT...SELECT, Version for 5.0.
  Added the thread handle to unique_table().
sql/sql_load.cc:
  BUG#5390 - problems with merge tables
  Problem #1: INSERT...SELECT, Version for 5.0.
  Added the thread handle to unique_table().
sql/sql_parse.cc:
  BUG#5390 - problems with merge tables
  Problem #1: INSERT...SELECT, Version for 5.0.
  Added the thread handle to unique_table().
sql/sql_update.cc:
  BUG#5390 - problems with merge tables
  Problem #1: INSERT...SELECT, Version for 5.0.
  Added the thread handle to unique_table().
  Replaced a call to find_table_in_local_list() by
  the newly extended unique_table().

mysql-test/r/create.result

@@ -259,21 +259,6 @@ select * from t1;
 0	1	2
 0	0	1
 drop table t1;
-create table t1 select 1,2,3;
-create table if not exists t1 select 1,2;
-Warnings:
-Note	1050	Table 't1' already exists
-create table if not exists t1 select 1,2,3,4;
-ERROR 21S01: Column count doesn't match value count at row 1
-create table if not exists t1 select 1;
-Warnings:
-Note	1050	Table 't1' already exists
-select * from t1;
-1	2	3
-1	2	3
-0	1	2
-0	0	1
-drop table t1;
 create table t1 (a int not null, b int, primary key (a));
 insert into t1 values (1,1);
 create table if not exists t1 select 2;

mysql-test/r/merge.result

@@ -719,3 +719,52 @@ SELECT b FROM t2;
 b
 3
 DROP TABLE t1, t2;
+create table t1(a int);
+create table t2(a int);
+insert into t1 values (1);
+insert into t2 values (2);
+create table t3 (a int) engine=merge union=(t1, t2) insert_method=first;
+select * from t3;
+a
+1
+2
+insert t2 select * from t2;
+select * from t2;
+a
+2
+2
+insert t3 select * from t1;
+select * from t3;
+a
+1
+1
+2
+2
+insert t1 select * from t3;
+select * from t1;
+a
+1
+1
+1
+1
+2
+2
+select * from t2;
+a
+2
+2
+select * from t3;
+a
+1
+1
+1
+1
+2
+2
+2
+2
+check table t1, t2;
+Table	Op	Msg_type	Msg_text
+test.t1	check	status	OK
+test.t2	check	status	OK
+drop table t1, t2, t3;

mysql-test/t/create.test

@@ -211,13 +211,6 @@ drop table t1;
 # bug #1434
 #
 
-create table t1 select 1,2,3;
-create table if not exists t1 select 1,2;
---error 1136
-create table if not exists t1 select 1,2,3,4;
-create table if not exists t1 select 1;
-select * from t1;
-drop table t1;
 create table t1 select 1,2,3;
 create table if not exists t1 select 1,2;
 --error 1136

mysql-test/t/merge.test

@@ -352,4 +352,30 @@ INSERT INTO t2 (b) VALUES (1) ON DUPLICATE KEY UPDATE b=3;
 SELECT b FROM t2;
 DROP TABLE t1, t2;
 
+
+#
+# BUG#5390 - problems with merge tables
+# Problem #1: INSERT...SELECT
+#
+#drop table if exists t1, t2, t3;
+create table t1(a int);
+create table t2(a int);
+insert into t1 values (1);
+insert into t2 values (2);
+create table t3 (a int) engine=merge union=(t1, t2) insert_method=first;
+select * from t3;
+#
+insert t2 select * from t2;
+select * from t2;
+#
+insert t3 select * from t1;
+select * from t3;
+#
+insert t1 select * from t3;
+select * from t1;
+select * from t2;
+select * from t3;
+check table t1, t2;
+drop table t1, t2, t3;
+
 # End of 4.1 tests

sql/lock.cc

@@ -423,6 +423,127 @@ MYSQL_LOCK *mysql_lock_merge(MYSQL_LOCK *a,MYSQL_LOCK *b)
 }
 
 
+/*
+  Find duplicate lock in tables.
+
+  SYNOPSIS
+    mysql_lock_have_duplicate()
+    thd                         The current thread.
+    needle                      The table to check for duplicate lock.
+    haystack                    The list of tables to search for the dup lock.
+
+  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.
+
+  RETURN
+    1           A table from 'tables' matches a lock on 'table'.
+    0           No duplicate lock is present.
+    -1          Error.
+*/
+
+TABLE_LIST *mysql_lock_have_duplicate(THD *thd, TABLE_LIST *needle,
+                                      TABLE_LIST *haystack)
+{
+  uint                  count;
+  uint                  dup_pos;
+  TABLE                 *write_lock_used; /* dummy */
+  TABLE                 **tables1;
+  TABLE                 **tables2;
+  TABLE                 **table_ptr;
+  TABLE_LIST            *tlist_ptr;
+  MYSQL_LOCK            *sql_lock1;
+  MYSQL_LOCK            *sql_lock2;
+  THR_LOCK_DATA         **lock_data1;
+  THR_LOCK_DATA         **end_data1;
+  THR_LOCK_DATA         **lock_data2;
+  THR_LOCK_DATA         **end_data2;
+  THR_LOCK              *lock1;
+  DBUG_ENTER("mysql_lock_have_duplicate");
+
+  /* Table may not be defined for derived or view tables. */
+  if (! needle->table)
+    DBUG_RETURN(NULL);
+
+  /* Get lock(s) for needle. */
+  tables1= &needle->table;
+  if (! (sql_lock1= get_lock_data(thd, tables1, 1, 1, &write_lock_used)))
+    goto err0;
+
+  /* Count real tables in list. */
+  count=0;
+  for (tlist_ptr = haystack; tlist_ptr; tlist_ptr= tlist_ptr->next_global)
+    if (! tlist_ptr->placeholder() && ! tlist_ptr->schema_table)
+      count++;
+  /* Allocate a table array. */
+  if (! (tables2= (TABLE**) sql_alloc(sizeof(TABLE*) * count)))
+    goto err1;
+  table_ptr= tables2;
+  /* Assign table pointers. */
+  for (tlist_ptr = haystack; tlist_ptr; tlist_ptr= tlist_ptr->next_global)
+    if (! tlist_ptr->placeholder() && ! tlist_ptr->schema_table)
+      *(table_ptr++)= tlist_ptr->table;
+  /* Get lock(s) for haystack. */
+  if (! (sql_lock2= get_lock_data(thd, tables2, count, 1, &write_lock_used)))
+    goto err1;
+
+  /* Initialize duplicate position to an impossible value. */
+  dup_pos= UINT_MAX;
+  /*
+    Find a duplicate lock.
+    In case of merge tables, sql_lock1 can have more than 1 lock.
+  */
+  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;
+         lock_data2 < end_data2;
+         lock_data2++)
+    {
+      if ((*lock_data2)->lock == lock1)
+      {
+        DBUG_PRINT("ingo", ("duplicate lock found"));
+        /* Change duplicate position to the real value. */
+        dup_pos= lock_data2 - sql_lock2->locks;
+        goto end;
+      }
+    }
+  }
+
+ end:
+  tlist_ptr= NULL; /* In case that no duplicate was found. */
+  if (dup_pos != UINT_MAX)
+  {
+    /* Duplicate found. Search the matching TABLE_LIST object. */
+    count= 0;
+    for (tlist_ptr = haystack; tlist_ptr; tlist_ptr= tlist_ptr->next_global)
+    {
+      if (! tlist_ptr->placeholder() && ! tlist_ptr->schema_table)
+      {
+        count+= tlist_ptr->table->file->lock_count();
+        if (count > dup_pos)
+          break;
+      }
+    }
+  }
+  my_free((gptr) sql_lock2, MYF(0));
+  my_free((gptr) sql_lock1, MYF(0));
+  DBUG_RETURN(tlist_ptr);
+
+ err1:
+  my_free((gptr) sql_lock1, MYF(0));
+ err0:
+  /* This non-null but special value indicates error, if caller cares. */
+  DBUG_RETURN(needle);
+}
+
+
 	/* unlock a set of external */
 
 static int unlock_external(THD *thd, TABLE **table,uint count)
@@ -460,6 +581,7 @@ static MYSQL_LOCK *get_lock_data(THD *thd, TABLE **table_ptr, uint count,
   MYSQL_LOCK *sql_lock;
   THR_LOCK_DATA **locks;
   TABLE **to;
+  DBUG_ENTER("get_lock_data");
 
   *write_lock_used=0;
   for (i=tables=lock_count=0 ; i < count ; i++)
@@ -488,7 +610,7 @@ static MYSQL_LOCK *get_lock_data(THD *thd, TABLE **table_ptr, uint count,
 	my_malloc(sizeof(*sql_lock)+
 		  sizeof(THR_LOCK_DATA*)*tables+sizeof(table_ptr)*lock_count,
 		  MYF(0))))
-    return 0;
+    DBUG_RETURN(0);
   locks=sql_lock->locks=(THR_LOCK_DATA**) (sql_lock+1);
   to=sql_lock->table=(TABLE**) (locks+tables);
   sql_lock->table_count=lock_count;
@@ -508,7 +630,7 @@ static MYSQL_LOCK *get_lock_data(THD *thd, TABLE **table_ptr, uint count,
       {
 	my_error(ER_OPEN_AS_READONLY, MYF(0), table->alias);
 	my_free((gptr) sql_lock,MYF(0));
-	return 0;
+	DBUG_RETURN(0);
       }
     }
     THR_LOCK_DATA **org_locks = locks;
@@ -518,7 +640,7 @@ static MYSQL_LOCK *get_lock_data(THD *thd, TABLE **table_ptr, uint count,
       for ( ; org_locks != locks ; org_locks++)
 	(*org_locks)->debug_print_param= (void *) table;
   }
-  return sql_lock;
+  DBUG_RETURN(sql_lock);
 }
 
 

sql/mysql_priv.h

@@ -981,7 +981,7 @@ TABLE_LIST *find_table_in_list(TABLE_LIST *table,
                                uint offset_to_list,
                                const char *db_name,
                                const char *table_name);
-TABLE_LIST *unique_table(TABLE_LIST *table, TABLE_LIST *table_list);
+TABLE_LIST *unique_table(THD *thd, TABLE_LIST *table, TABLE_LIST *table_list);
 TABLE **find_temporary_table(THD *thd, const char *db, const char *table_name);
 bool close_temporary_table(THD *thd, const char *db, const char *table_name);
 void close_temporary(TABLE *table, bool delete_table);
@@ -1269,6 +1269,8 @@ void mysql_lock_remove(THD *thd, MYSQL_LOCK *locked,TABLE *table);
 void mysql_lock_abort(THD *thd, TABLE *table);
 bool mysql_lock_abort_for_thread(THD *thd, TABLE *table);
 MYSQL_LOCK *mysql_lock_merge(MYSQL_LOCK *a,MYSQL_LOCK *b);
+TABLE_LIST *mysql_lock_have_duplicate(THD *thd, TABLE_LIST *needle,
+                                      TABLE_LIST *haystack);
 bool lock_global_read_lock(THD *thd);
 void unlock_global_read_lock(THD *thd);
 bool wait_if_global_read_lock(THD *thd, bool abort_on_refresh,

sql/sql_base.cc

@@ -698,6 +698,7 @@ TABLE_LIST *find_table_in_list(TABLE_LIST *table,
 
   SYNOPSIS
     unique_table()
+    thd                   thread handle
     table                 table which should be checked
     table_list            list of tables
 
@@ -723,7 +724,7 @@ TABLE_LIST *find_table_in_list(TABLE_LIST *table,
     0 if table is unique
 */
 
-TABLE_LIST* unique_table(TABLE_LIST *table, TABLE_LIST *table_list)
+TABLE_LIST* unique_table(THD *thd, TABLE_LIST *table, TABLE_LIST *table_list)
 {
   TABLE_LIST *res;
   const char *d_name, *t_name;
@@ -758,9 +759,10 @@ TABLE_LIST* unique_table(TABLE_LIST *table, TABLE_LIST *table_list)
   DBUG_PRINT("info", ("real table: %s.%s", d_name, t_name));
   for (;;)
   {
-    if (!(res= find_table_in_global_list(table_list, d_name, t_name)) ||
-        (!res->table || res->table != table->table) &&
-        (res->select_lex && !res->select_lex->exclude_from_table_unique_test))
+    if (((! (res= find_table_in_global_list(table_list, d_name, t_name))) &&
+         (! (res= mysql_lock_have_duplicate(thd, table, table_list)))) ||
+        ((!res->table || res->table != table->table) &&
+         res->select_lex && !res->select_lex->exclude_from_table_unique_test))
       break;
     /*
       If we found entry of this table or or table of SELECT which already

sql/sql_delete.cc

@@ -348,7 +348,7 @@ bool mysql_prepare_delete(THD *thd, TABLE_LIST *table_list, Item **conds)
   }
   {
     TABLE_LIST *duplicate;
-    if ((duplicate= unique_table(table_list, table_list->next_global)))
+    if ((duplicate= unique_table(thd, table_list, table_list->next_global)))
     {
       update_non_unique_table_error(table_list, "DELETE", duplicate);
       DBUG_RETURN(TRUE);
@@ -438,7 +438,7 @@ bool mysql_multi_delete_prepare(THD *thd)
     */
     {
       TABLE_LIST *duplicate;
-      if ((duplicate= unique_table(target_tbl->correspondent_table,
+      if ((duplicate= unique_table(thd, target_tbl->correspondent_table,
                                    lex->query_tables)))
       {
         update_non_unique_table_error(target_tbl->correspondent_table,

sql/sql_insert.cc

@@ -870,7 +870,7 @@ bool mysql_prepare_insert(THD *thd, TABLE_LIST *table_list,
   {
     Item *fake_conds= 0;
     TABLE_LIST *duplicate;
-    if ((duplicate= unique_table(table_list, table_list->next_global)))
+    if ((duplicate= unique_table(thd, table_list, table_list->next_global)))
     {
       update_non_unique_table_error(table_list, "INSERT", duplicate);
       DBUG_RETURN(TRUE);
@@ -2171,7 +2171,7 @@ select_insert::prepare(List<Item> &values, SELECT_LEX_UNIT *u)
     query
   */
   if (!(lex->current_select->options & OPTION_BUFFER_RESULT) &&
-      unique_table(table_list, table_list->next_global))
+      unique_table(thd, table_list, table_list->next_global))
   {
     /* Using same table for INSERT and SELECT */
     lex->current_select->options|= OPTION_BUFFER_RESULT;

sql/sql_load.cc

@@ -178,7 +178,7 @@ bool mysql_load(THD *thd,sql_exchange *ex,TABLE_LIST *table_list,
     table is marked to be 'used for insert' in which case we should never
     mark this table as as 'const table' (ie, one that has only one row).
   */
-  if (unique_table(table_list, table_list->next_global))
+  if (unique_table(thd, table_list, table_list->next_global))
   {
     my_error(ER_UPDATE_TABLE_USED, MYF(0), table_list->table_name);
     DBUG_RETURN(TRUE);

sql/sql_parse.cc

@@ -2835,7 +2835,7 @@ mysql_execute_command(THD *thd)
         if (!(lex->create_info.options & HA_LEX_CREATE_TMP_TABLE))
         {
           TABLE_LIST *duplicate;
-          if ((duplicate= unique_table(create_table, select_tables)))
+          if ((duplicate= unique_table(thd, create_table, select_tables)))
           {
             update_non_unique_table_error(create_table, "CREATE", duplicate);
             res= 1;
@@ -2851,7 +2851,7 @@ mysql_execute_command(THD *thd)
                tab= tab->next_local)
           {
             TABLE_LIST *duplicate;
-            if ((duplicate= unique_table(tab, select_tables)))
+            if ((duplicate= unique_table(thd, tab, select_tables)))
             {
               update_non_unique_table_error(tab, "CREATE", duplicate);
               res= 1;

sql/sql_update.cc

@@ -616,7 +616,7 @@ bool mysql_prepare_update(THD *thd, TABLE_LIST *table_list,
   /* Check that we are not using table that we are updating in a sub select */
   {
     TABLE_LIST *duplicate;
-    if ((duplicate= unique_table(table_list, table_list->next_global)))
+    if ((duplicate= unique_table(thd, table_list, table_list->next_global)))
     {
       update_non_unique_table_error(table_list, "UPDATE", duplicate);
       my_error(ER_UPDATE_TABLE_USED, MYF(0), table_list->table_name);
@@ -839,7 +839,7 @@ bool mysql_multi_update_prepare(THD *thd)
         tl->lock_type != TL_READ_NO_INSERT)
     {
       TABLE_LIST *duplicate;
-      if ((duplicate= unique_table(tl, table_list)))
+      if ((duplicate= unique_table(thd, tl, table_list)))
       {
         update_non_unique_table_error(table_list, "UPDATE", duplicate);
         DBUG_RETURN(TRUE);
@@ -1026,8 +1026,7 @@ int multi_update::prepare(List<Item> &not_used_values,
   {
     TABLE *table=table_ref->table;
     if (!(tables_to_update & table->map) &&
-	find_table_in_local_list(update_tables, table_ref->db,
-				table_ref->table_name))
+        unique_table(thd, table_ref, update_tables))
       table->no_cache= 1;			// Disable row cache
   }
   DBUG_RETURN(thd->is_fatal_error != 0);