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:
  Test for LOCK bug
mysql-test/t/bdb.test:
  Test for LOCK bug
sql/ha_berkeley.cc:
  Fix bugs when using LOCK TABLES
  Optimized remove of key when using internal cursor
sql/ha_berkeley.h:
  Initilize of changed_rows
sql/lock.cc:
  Fix for BDB tables

3 files changed, 21 insertions, 16 deletions

diff --git a/sql/ha_berkeley.cc b/sql/ha_berkeley.cc
index ac4088ed6a9..9a437607729 100644
--- a/sql/ha_berkeley.cc
+++ b/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)
diff --git a/sql/ha_berkeley.h b/sql/ha_berkeley.h
index 3eb793937ae..fda7d2f2069 100644
--- a/sql/ha_berkeley.h
+++ b/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() {}
diff --git a/sql/lock.cc b/sql/lock.cc
index 7bac2d54631..aa06822e03f 100644
--- a/sql/lock.cc
+++ b/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);