bug#4089 - JOIN::join_free calling mysql_unlock w/o index_end() before

sql/ha_myisam.h:
  cleanup
mysql-test/r/bdb.result:
  bug#4089
mysql-test/t/bdb.test:
  bug#4089
sql/opt_range.cc:
  be sloppy
sql/sql_select.cc:
  JOIN::join_free - pass it down the tree (of selects)
  call mysql_unlock_tables only in top-level select

mysql-test/r/bdb.result

@@ -1190,3 +1190,23 @@ exists (select 'two' from t1 where 'two' = outer_table.b);
 b
 drop table t1;
 set autocommit=1;
+create table t1(a int primary key, b varchar(30)) engine=bdb;
+insert into t1 values (1,'one'), (2,'two'), (3,'three'), (4,'four');
+create table t2 like t1;
+insert t2 select * from t1;
+select a from t1 where a in (select a from t2);
+a
+1
+2
+3
+4
+delete from t2;
+insert into t2 (a, b)
+select a, b from t1 where (a, b) in (select a, b from t1);
+select * from t2;
+a	b
+1	one
+2	two
+3	three
+4	four
+drop table t1, t2;

mysql-test/t/bdb.test

@@ -840,10 +840,13 @@ set autocommit=1;
 # Bug #4089: subselect and open cursor.
 #
 
-#create table t1(a int primary key, b varchar(30)) engine=bdb;
-#insert into t1 values (1,'one'), (2,'two'), (3,'three'), (4,'four');
-#create table t2 like t1;
-#insert into t2 (a, b)
-#  select a, b from t1 where (a, b) in (select a, b from t1);
-#select * from t2;
-#drop table t1, t2;
+create table t1(a int primary key, b varchar(30)) engine=bdb;
+insert into t1 values (1,'one'), (2,'two'), (3,'three'), (4,'four');
+create table t2 like t1;
+insert t2 select * from t1;
+select a from t1 where a in (select a from t2);
+delete from t2;
+insert into t2 (a, b)
+  select a, b from t1 where (a, b) in (select a, b from t1);
+select * from t2;
+drop table t1, t2;

sql/ha_myisam.h

@@ -81,7 +81,7 @@ class ha_myisam: public handler
   int index_first(byte * buf);
   int index_last(byte * buf);
   int index_next_same(byte *buf, const byte *key, uint keylen);
-  int index_end() { ft_handler=NULL; return handler::index_end(); }
+  int index_end() { ft_handler=NULL; return 0; }
   int ft_init()
   {
     if (!ft_handler)

sql/opt_range.cc

@@ -413,7 +413,8 @@ QUICK_SELECT::~QUICK_SELECT()
 {
   if (!dont_free)
   {
-    file->ha_index_end();
+    if (file->inited)
+      file->ha_index_end();
     free_root(&alloc,MYF(0));
   }
 }

sql/sql_select.cc

@@ -3851,6 +3851,8 @@ JOIN::join_free(bool full)
   JOIN_TAB *tab,*end;
   DBUG_ENTER("JOIN::join_free");
 
+  full= full || !select_lex->uncacheable;
+
   if (table)
   {
     /*
@@ -3862,7 +3864,18 @@ JOIN::join_free(bool full)
       free_io_cache(table[const_tables]);
       filesort_free_buffers(table[const_tables]);
     }
-    if (full || !select_lex->uncacheable)
+
+    for (SELECT_LEX_UNIT *unit= select_lex->first_inner_unit(); unit;
+         unit= unit->next_unit())
+    {
+      JOIN *join;
+      for (SELECT_LEX *sl= unit->first_select_in_union(); sl;
+           sl= sl->next_select())
+        if ((join= sl->join))
+          join->join_free(full);
+    }
+
+    if (full)
     {
       for (tab= join_tab, end= tab+tables; tab != end; tab++)
 	tab->cleanup();
@@ -3872,22 +3885,27 @@ JOIN::join_free(bool full)
     {
       for (tab= join_tab, end= tab+tables; tab != end; tab++)
       {
-	if (tab->table && tab->table->file->inited == handler::RND)
-	    tab->table->file->ha_rnd_end();
+	if (tab->table)
+	    tab->table->file->ha_index_or_rnd_end();
       }
     }
   }
+
   /*
     We are not using tables anymore
     Unlock all tables. We may be in an INSERT .... SELECT statement.
   */
-  if ((full || !select_lex->uncacheable) &&
-      lock && thd->lock &&
-      !(select_options & SELECT_NO_UNLOCK))
+  if (full && lock && thd->lock && !(select_options & SELECT_NO_UNLOCK))
   {
-    mysql_unlock_read_tables(thd, lock);// Don't free join->lock
-    lock=0;
+    // TODO: unlock tables even if the join isn't top level select in the tree
+    if (select_lex == (thd->lex->unit.fake_select_lex ?
+                       thd->lex->unit.fake_select_lex : &thd->lex->select_lex))
+    {
+      mysql_unlock_read_tables(thd, lock);        // Don't free join->lock
+      lock=0;
+    }
   }
+
   if (full)
   {
     group_fields.delete_elements();