Fix bugs when using LOCK TABLES with BDB tables

Optimized remove of key when using internal cursor in BDB tables.

mysql-test/r/bdb.result

@@ -524,3 +524,6 @@ a	b	a	b
 a	b
 1	1
 1	2
+id	id2	id3	dummy1
+id	id2	id3	dummy1
+NULL	NULL	NULL	NULL

mysql-test/t/bdb.test

@@ -727,13 +727,24 @@ DROP TABLE t1,t2;
 # Test problem with joining table to itself on a multi-part unique key
 #
 
-drop table if exists t;
-create table t (a int(11) not null, b int(11) not null, unique (a,b)) type=bdb;
-
-insert into t values (1,1), (1,2);
+drop table if exists t1;
+create table t1 (a int(11) not null, b int(11) not null, unique (a,b)) type=bdb;
+insert into t1 values (1,1), (1,2);
+select * from t1 where a = 1;
+select t1.*, t2.* from t1, t1 t2 where t1.a = t2.a and t2.a = 1;
+select * from t1 where a = 1;
+drop table t1;
 
-select * from t where a = 1;
-select t1.*, t2.* from t t1, t t2 where t1.a = t2.a and t2.a = 1;
-select * from t where a = 1;
+#
+# This caused a deadlock in BDB internal locks
+#
 
-drop table t;
+create table t1 (id int NOT NULL,id2 int NOT NULL,id3 int NOT NULL,dummy1 char(30),primary key (id,id2),index index_id3 (id3)) type=bdb;
+insert into t1 values (0,0,0,'ABCDEFGHIJ');
+create table t2 (id int NOT NULL,primary key (id)) type=bdb;
+LOCK TABLES t1 WRITE, t2 WRITE;
+insert into t2 values(1);
+SELECT t1.* FROM t1 WHERE id IN (1);
+SELECT t1.* FROM t2 left outer join t1 on (t1.id=t2.id);
+delete from t1 where id3 >= 0 and id3 <= 0;
+drop table t1,t2;

sql/ha_berkeley.cc

@@ -1306,7 +1306,7 @@ int ha_berkeley::delete_row(const byte * record)
 int ha_berkeley::index_init(uint keynr)
 {
   int error;
-  DBUG_ENTER("index_init");
+  DBUG_ENTER("ha_berkeley::index_init");
   DBUG_PRINT("enter",("table: '%s'  key: %d", table->real_name, keynr));
 
   /*
@@ -1347,7 +1347,7 @@ int ha_berkeley::index_end()
 int ha_berkeley::read_row(int error, char *buf, uint keynr, DBT *row,
 			  DBT *found_key, bool read_next)
 {
-  DBUG_ENTER("read_row");
+  DBUG_ENTER("ha_berkeley::read_row");
   if (error)
   {
     if (error == DB_NOTFOUND || error == DB_KEYEMPTY)
@@ -1399,6 +1399,7 @@ int ha_berkeley::index_read_idx(byte * buf, uint keynr, const byte * key,
   statistic_increment(ha_read_key_count,&LOCK_status);
   DBUG_ENTER("index_read_idx");
   current_row.flags=DB_DBT_REALLOC;
+  active_index= -1;
   DBUG_RETURN(read_row(key_file[keynr]->get(key_file[keynr], transaction,
 				 pack_key(&last_key, keynr, key_buff, key,
 					  key_len),
@@ -1413,7 +1414,7 @@ int ha_berkeley::index_read(byte * buf, const byte * key,
   DBT row;
   int error;
   KEY *key_info= &table->key_info[active_index];
-  DBUG_ENTER("index_read");
+  DBUG_ENTER("ha_berkeley::index_read");
 
   statistic_increment(ha_read_key_count,&LOCK_status);
   bzero((char*) &row,sizeof(row));
@@ -1518,8 +1519,9 @@ int ha_berkeley::index_last(byte * buf)
 
 int ha_berkeley::rnd_init(bool scan)
 {
+  DBUG_ENTER("rnd_init");
   current_row.flags=DB_DBT_REALLOC;
-  return index_init(primary_key);
+  DBUG_RETURN(index_init(primary_key));
 }
 
 int ha_berkeley::rnd_end()
@@ -1534,7 +1536,7 @@ int ha_berkeley::rnd_next(byte *buf)
   statistic_increment(ha_read_rnd_next_count,&LOCK_status);
   bzero((char*) &row,sizeof(row));
   DBUG_RETURN(read_row(cursor->c_get(cursor, &last_key, &row, DB_NEXT),
-		       (char*) buf, active_index, &row, &last_key, 1));
+		       (char*) buf, primary_key, &row, &last_key, 1));
 }
 
 
@@ -1564,10 +1566,11 @@ int ha_berkeley::rnd_pos(byte * buf, byte *pos)
   DBT db_pos;
   statistic_increment(ha_read_rnd_count,&LOCK_status);
 
+  active_index= (uint) -1;			// Don't delete via cursor
   return read_row(file->get(file, transaction,
 			    get_pos(&db_pos, pos),
 			    &current_row, 0),
-		 (char*) buf, active_index, &current_row, (DBT*) 0, 0);
+		 (char*) buf, primary_key, &current_row, (DBT*) 0, 0);
 }
 
 void ha_berkeley::position(const byte *record)
@@ -1582,10 +1585,10 @@ void ha_berkeley::position(const byte *record)
 
 void ha_berkeley::info(uint flag)
 {
-  DBUG_ENTER("info");
+  DBUG_ENTER("ha_berkeley::info");
   if (flag & HA_STATUS_VARIABLE)
   {
-    records = share->rows;		// Just to get optimisations right
+    records = share->rows + changed_rows; // Just to get optimisations right
     deleted = 0;
   }
   if ((flag & HA_STATUS_CONST) || version != share->version)
@@ -1663,14 +1666,13 @@ int ha_berkeley::external_lock(THD *thd, int lock_type)
   {
     if (!thd->transaction.bdb_lock_count++)
     {
-      changed_rows=0;
+      DBUG_ASSERT(thd->transaction.stmt.bdb_tid == 0);
       transaction=0;				// Safety
       /* First table lock, start transaction */
       if ((thd->options & (OPTION_NOT_AUTO_COMMIT | OPTION_BEGIN |
 			   OPTION_TABLE_LOCK)) &&
 	  !thd->transaction.all.bdb_tid)
       {
-	DBUG_ASSERT(thd->transaction.stmt.bdb_tid == 0);
 	/* We have to start a master transaction */
 	DBUG_PRINT("trans",("starting transaction all"));
 	if ((error=txn_begin(db_env, 0,
@@ -1700,6 +1702,7 @@ int ha_berkeley::external_lock(THD *thd, int lock_type)
   {
     lock.type=TL_UNLOCK;			// Unlocked
     thread_safe_add(share->rows, changed_rows, &share->mutex);
+    changed_rows=0;
     if (!--thd->transaction.bdb_lock_count)
     {
       if (thd->transaction.stmt.bdb_tid)
@@ -1736,8 +1739,8 @@ int ha_berkeley::start_stmt(THD *thd)
     error=txn_begin(db_env, (DB_TXN*) thd->transaction.all.bdb_tid,
 		    (DB_TXN**) &thd->transaction.stmt.bdb_tid,
 		    0);
-    transaction= (DB_TXN*) thd->transaction.stmt.bdb_tid;
   }
+  transaction= (DB_TXN*) thd->transaction.stmt.bdb_tid;
   DBUG_RETURN(error);
 }
 
@@ -1942,6 +1945,8 @@ longlong ha_berkeley::get_auto_increment()
   longlong nr=1;				// Default if error or new key
   int error;
   (void) ha_berkeley::extra(HA_EXTRA_KEYREAD);
+
+  /* Set 'active_index' */
   ha_berkeley::index_init(table->next_number_index);
 
   if (!table->next_number_key_offset)

sql/ha_berkeley.h

@@ -93,7 +93,7 @@ class ha_berkeley: public handler
 		    HA_BLOB_KEY | HA_NOT_EXACT_COUNT |
 		    HA_PRIMARY_KEY_IN_READ_INDEX | HA_DROP_BEFORE_CREATE |
 		    HA_AUTO_PART_KEY),
-    last_dup_key((uint) -1),version(0),using_ignore(0)
+    changed_rows(0),last_dup_key((uint) -1),version(0),using_ignore(0)
   {
   }
   ~ha_berkeley() {}

sql/lock.cc

@@ -170,10 +170,9 @@ void mysql_unlock_tables(THD *thd, MYSQL_LOCK *sql_lock)
 {
   DBUG_ENTER("mysql_unlock_tables");
   if (sql_lock->lock_count)
-  {
     thr_multi_unlock(sql_lock->locks,sql_lock->lock_count);
+  if (sql_lock->table_count)
     VOID(unlock_external(thd,sql_lock->table,sql_lock->table_count));
-  }
   my_free((gptr) sql_lock,MYF(0));
   DBUG_VOID_RETURN;
 }
@@ -317,7 +316,7 @@ static int unlock_external(THD *thd, TABLE **table,uint count)
   DBUG_ENTER("unlock_external");
 
   error_code=0;
-  for (; count-- ; table++)
+  do
   {
     if ((*table)->current_lock != F_UNLCK)
     {
@@ -325,7 +324,8 @@ static int unlock_external(THD *thd, TABLE **table,uint count)
       if ((error=(*table)->file->external_lock(thd, F_UNLCK)))
 	error_code=error;
     }
-  }
+    table++;
+  } while (--count);
   if (error_code)
     print_lock_error(error_code);
   DBUG_RETURN(error_code);