#150 cleanup indexing cursor when a txn is retired

mysql-test/suite/tokudb.bugs/r/commit_index_end_1.result

+drop table if exists t1,t2,t3;
+SET NAMES latin1;
+CREATE TABLE t1 (a INT) ENGINE=TokuDB;
+INSERT INTO t1 VALUES (1),(2);
+CREATE TABLE t2 (b INT) ENGINE=TokuDB;
+INSERT INTO t2 VALUES (3),(4);
+CREATE TABLE t3 (c VARCHAR(3), INDEX(c)) ENGINE=TokuDB;
+INSERT INTO t3 VALUES ('foo'),('bar');
+EXPLAIN SELECT * FROM t1
+WHERE a IN ( SELECT b FROM t2 ) OR ( 'qux' ) IN ( SELECT c FROM t3 );
+id	select_type	table	type	possible_keys	key	key_len	ref	rows	Extra
+1	PRIMARY	t1	ALL	NULL	NULL	NULL	NULL	2	Using where
+3	SUBQUERY	t3	index_subquery	c	c	6	const	0	Using index; Using where
+2	MATERIALIZED	t2	ALL	NULL	NULL	NULL	NULL	2	
+drop table if exists t1,t2,t3;

mysql-test/suite/tokudb.bugs/r/commit_index_end_2.result

+drop table if exists t1,t2;
+CREATE TABLE t1 (a VARCHAR(8), INDEX(a)) ENGINE=TokuDB;
+INSERT INTO t1 VALUES ('foo'),('bar');
+CREATE TABLE t2 AS SELECT ( 'qux' ) IN ( SELECT a FROM t1 ) AS f1;
+drop table t1,t2;

mysql-test/suite/tokudb.bugs/t/commit_index_end_1.test

+# this test crashes because tokudb does not handle index_end after txn commit (MDEV-5396)
+
+source include/have_tokudb.inc;
+
+disable_warnings;
+drop table if exists t1,t2,t3;
+enable_warnings;
+
+SET NAMES latin1;
+
+CREATE TABLE t1 (a INT) ENGINE=TokuDB;
+INSERT INTO t1 VALUES (1),(2);
+
+CREATE TABLE t2 (b INT) ENGINE=TokuDB;
+INSERT INTO t2 VALUES (3),(4);
+
+CREATE TABLE t3 (c VARCHAR(3), INDEX(c)) ENGINE=TokuDB;
+INSERT INTO t3 VALUES ('foo'),('bar');
+
+EXPLAIN SELECT * FROM t1
+WHERE a IN ( SELECT b FROM t2 ) OR ( 'qux' ) IN ( SELECT c FROM t3 );
+
+drop table if exists t1,t2,t3;
\ No newline at end of file

mysql-test/suite/tokudb.bugs/t/commit_index_end_2.test

+# this test crashes because tokudb does not handle index_end after txn commit (MDEV-5396)
+
+source include/have_tokudb.inc;
+
+disable_warnings;
+drop table if exists t1,t2;
+enable_warnings;
+
+CREATE TABLE t1 (a VARCHAR(8), INDEX(a)) ENGINE=TokuDB; 
+INSERT INTO t1 VALUES ('foo'),('bar'); 
+
+CREATE TABLE t2 AS SELECT ( 'qux' ) IN ( SELECT a FROM t1 ) AS f1;
+
+drop table t1,t2;

storage/tokudb/ha_tokudb.cc

@@ -1240,6 +1240,7 @@ ha_tokudb::ha_tokudb(handlerton * hton, TABLE_SHARE * table_arg):handler(hton, t
     prelocked_right_range_size = 0;
     tokudb_active_index = MAX_KEY;
     invalidate_icp();
+    trx_handler_list.data = this;
 }
 
 ha_tokudb::~ha_tokudb() {
@@ -4436,6 +4437,7 @@ int ha_tokudb::prepare_index_key_scan(const uchar * key, uint key_len) {
             int r = cursor->c_close(cursor);
             assert(r==0);
             cursor = NULL;
+            remove_from_trx_handler_list();
         }
     }
     return error;
@@ -4480,6 +4482,7 @@ int ha_tokudb::index_init(uint keynr, bool sorted) {
         DBUG_PRINT("note", ("Closing active cursor"));
         int r = cursor->c_close(cursor);
         assert(r==0);
+        remove_from_trx_handler_list();
     }
     active_index = keynr;
 
@@ -4517,6 +4520,8 @@ int ha_tokudb::index_init(uint keynr, bool sorted) {
     }
     memset((void *) &last_key, 0, sizeof(last_key));
 
+    add_to_trx_handler_list();
+
     if (thd_sql_command(thd) == SQLCOM_SELECT) {
         set_query_columns(keynr);
         unpack_entire_row = false;
@@ -4542,6 +4547,7 @@ int ha_tokudb::index_end() {
         int r = cursor->c_close(cursor);
         assert(r==0);
         cursor = NULL;
+        remove_from_trx_handler_list();
         last_cursor_error = 0;
     }
     active_index = tokudb_active_index = MAX_KEY;
@@ -5546,7 +5552,7 @@ int ha_tokudb::rnd_end() {
 //      error otherwise
 //
 int ha_tokudb::rnd_next(uchar * buf) {
-    TOKUDB_DBUG_ENTER("ha_tokudb::ha_tokudb::rnd_next");
+    TOKUDB_DBUG_ENTER("ha_tokudb::rnd_next");
     ha_statistic_increment(&SSV::ha_read_rnd_next_count);
     int error = get_next(buf, 1, NULL);
     TOKUDB_DBUG_RETURN(error);
@@ -5706,6 +5712,7 @@ int ha_tokudb::prelock_range( const key_range *start_key, const key_range *end_k
             int r = cursor->c_close(cursor);
             assert(r==0);
             cursor = NULL;
+            remove_from_trx_handler_list();
         }
         goto cleanup; 
     }
@@ -8254,6 +8261,25 @@ Item* ha_tokudb::idx_cond_push(uint keyno_arg, Item* idx_cond_arg) {
     return idx_cond_arg;
 }
 
+void ha_tokudb::cleanup_txn(DB_TXN *txn) {
+    if (transaction == txn && cursor) {
+        int r = cursor->c_close(cursor);
+        assert(r == 0);
+        cursor = NULL;
+        remove_from_trx_handler_list();
+    }
+}
+
+void ha_tokudb::add_to_trx_handler_list() {
+    tokudb_trx_data *trx = (tokudb_trx_data *) thd_data_get(ha_thd(), tokudb_hton->slot);
+    trx->handlers = list_add(trx->handlers, &trx_handler_list);
+}
+
+void ha_tokudb::remove_from_trx_handler_list() {
+    tokudb_trx_data *trx = (tokudb_trx_data *) thd_data_get(ha_thd(), tokudb_hton->slot);
+    trx->handlers = list_delete(trx->handlers, &trx_handler_list);
+}
+
 // table admin 
 #include "ha_tokudb_admin.cc"
 

storage/tokudb/ha_tokudb.h

@@ -787,6 +787,15 @@ class ha_tokudb : public handler {
     bool check_upsert(THD *thd, List<Item> &update_fields, List<Item> &update_values);
     int send_upsert_message(THD *thd, List<Item> &update_fields, List<Item> &update_values, DB_TXN *txn);
 #endif
+public:
+    // mysql sometimes retires a txn before a cursor that references the txn is closed.
+    // for example, commit is sometimes called before index_end.  the following methods
+    // put the handler on a list of handlers that get cleaned up when the txn is retired.
+    void cleanup_txn(DB_TXN *txn);
+private:
+    LIST trx_handler_list;
+    void add_to_trx_handler_list();
+    void remove_from_trx_handler_list();
 };
 
 #if defined(MARIADB_BASE_VERSION)

storage/tokudb/hatoku_defines.h

@@ -304,6 +304,7 @@ typedef struct st_tokudb_trx_data {
     uint tokudb_lock_count;
     tokudb_stmt_progress stmt_progress;
     bool checkpoint_lock_taken;
+    LIST *handlers;
 } tokudb_trx_data;
 
 extern char *tokudb_data_dir;

storage/tokudb/hatoku_hton.cc

@@ -644,6 +644,14 @@ static void abort_txn_with_progress(DB_TXN* txn, THD* thd) {
     assert(r == 0);
 }
 
+static void tokudb_cleanup_handlers(tokudb_trx_data *trx, DB_TXN *txn) {
+    LIST *e;
+    while ((e = trx->handlers)) {
+        ha_tokudb *handler = (ha_tokudb *) e->data;
+        handler->cleanup_txn(txn);
+    }
+}
+
 static int tokudb_commit(handlerton * hton, THD * thd, bool all) {
     TOKUDB_DBUG_ENTER("tokudb_commit");
     DBUG_PRINT("trans", ("ending transaction %s", all ? "all" : "stmt"));
@@ -657,6 +665,7 @@ static int tokudb_commit(handlerton * hton, THD * thd, bool all) {
         }
         // test hook to induce a crash on a debug build
         DBUG_EXECUTE_IF("tokudb_crash_commit_before", DBUG_SUICIDE(););
+        tokudb_cleanup_handlers(trx, this_txn);
         commit_txn_with_progress(this_txn, syncflag, thd);
         // test hook to induce a crash on a debug build
         DBUG_EXECUTE_IF("tokudb_crash_commit_after", DBUG_SUICIDE(););
@@ -683,6 +692,7 @@ static int tokudb_rollback(handlerton * hton, THD * thd, bool all) {
         if (tokudb_debug & TOKUDB_DEBUG_TXN) {
             TOKUDB_TRACE("rollback %u %p\n", all, this_txn);
         }
+        tokudb_cleanup_handlers(trx, this_txn);
         abort_txn_with_progress(this_txn, thd);
         if (this_txn == trx->sp_level) {
             trx->sp_level = 0;