Merge tag 'mysql-5.5.45' into 5.5

VERSION

 MYSQL_VERSION_MAJOR=5
 MYSQL_VERSION_MINOR=5
-MYSQL_VERSION_PATCH=44
+MYSQL_VERSION_PATCH=45
 MYSQL_VERSION_EXTRA=

client/mysqlslap.c

 /*
-   Copyright (c) 2005, 2012, Oracle and/or its affiliates.
+   Copyright (c) 2005, 2015, Oracle and/or its affiliates.
+   Copyright (c) 2010, 2015, MariaDB
 
    This program is free software; you can redistribute it and/or modify
    it under the terms of the GNU General Public License as published by
@@ -244,7 +245,7 @@ void print_conclusions_csv(conclusions *con);
 void generate_stats(conclusions *con, option_string *eng, stats *sptr);
 uint parse_comma(const char *string, uint **range);
 uint parse_delimiter(const char *script, statement **stmt, char delm);
-uint parse_option(const char *origin, option_string **stmt, char delm);
+int parse_option(const char *origin, option_string **stmt, char delm);
 static int drop_schema(MYSQL *mysql, const char *db);
 uint get_random_string(char *buf);
 static statement *build_table_string(void);
@@ -1259,7 +1260,13 @@ get_options(int *argc,char ***argv)
   if (num_int_cols_opt)
   {
     option_string *str;
-    parse_option(num_int_cols_opt, &str, ',');
+    if(parse_option(num_int_cols_opt, &str, ',') == -1)
+    {
+      fprintf(stderr, "Invalid value specified for the option "
+              "'number-int-cols'\n");
+      option_cleanup(str);
+      return 1;
+    }
     num_int_cols= atoi(str->string);
     if (str->option)
       num_int_cols_index= atoi(str->option);
@@ -1270,7 +1277,13 @@ get_options(int *argc,char ***argv)
   if (num_char_cols_opt)
   {
     option_string *str;
-    parse_option(num_char_cols_opt, &str, ',');
+    if(parse_option(num_char_cols_opt, &str, ',') == -1)
+    {
+      fprintf(stderr, "Invalid value specified for the option "
+              "'number-char-cols'\n");
+      option_cleanup(str);
+      return 1;
+    }
     num_char_cols= atoi(str->string);
     if (str->option)
       num_char_cols_index= atoi(str->option);
@@ -1507,7 +1520,13 @@ get_options(int *argc,char ***argv)
     printf("Parsing engines to use.\n");
 
   if (default_engine)
-    parse_option(default_engine, &engine_options, ',');
+  {
+    if(parse_option(default_engine, &engine_options, ',') == -1)
+    {
+      fprintf(stderr, "Invalid value specified for the option 'engine'\n");
+      return 1;
+    }
+  }
 
   if (tty_password)
     opt_password= get_tty_password(NullS);
@@ -1984,7 +2003,7 @@ end:
   DBUG_RETURN(0);
 }
 
-uint
+int
 parse_option(const char *origin, option_string **stmt, char delm)
 {
   char *retstr;
@@ -2009,6 +2028,13 @@ parse_option(const char *origin, option_string **stmt, char delm)
     char buffer[HUGE_STRING_LENGTH]= "";
     char *buffer_ptr;
 
+    /*
+      Return an error if the length of the any of the comma seprated value
+      exceeds HUGE_STRING_LENGTH.
+    */
+    if ((size_t)(retstr - ptr) > HUGE_STRING_LENGTH)
+      return -1;
+
     count++;
     strncpy(buffer, ptr, (size_t)(retstr - ptr));
     /*
@@ -2048,6 +2074,13 @@ parse_option(const char *origin, option_string **stmt, char delm)
   {
     char *origin_ptr;
 
+    /*
+      Return an error if the length of the any of the comma seprated value
+      exceeds HUGE_STRING_LENGTH.
+    */
+    if (strlen(ptr) > HUGE_STRING_LENGTH)
+      return -1;
+
     if ((origin_ptr= strchr(ptr, ':')))
     {
       char *option_ptr;
@@ -2058,13 +2091,13 @@ parse_option(const char *origin, option_string **stmt, char delm)
       option_ptr= (char *)ptr + 1 + tmp->length;
 
       /* Move past the : and the first string */
-      tmp->option_length= (size_t)((ptr + length) - option_ptr);
+      tmp->option_length= strlen(option_ptr);
       tmp->option= my_strndup(option_ptr, tmp->option_length,
                               MYF(MY_FAE));
     }
     else
     {
-      tmp->length= (size_t)((ptr + length) - ptr);
+      tmp->length= strlen(ptr);
       tmp->string= my_strndup(ptr, tmp->length, MYF(MY_FAE));
     }
 

mysql-test/r/merge_recover.result → mysql-test/r/myisam_recover.result

 #
-# Test of MyISAM MRG tables with corrupted children.
+# Tests for corrupted MyISAM tables and MyISAMMRG tables with corrupted
+# children..
+#
 # Run with --myisam-recover=force option.
 #
 # Preparation: we need to make sure that the merge parent
@@ -101,3 +103,48 @@ execute stmt;
 deallocate prepare stmt;
 set @@global.table_definition_cache=default;
 set @@global.table_open_cache=default;
+#
+# 18075170 - sql node restart required to avoid deadlock after
+#            restore
+#
+# Check that auto-repair for MyISAM tables can now happen in the
+# middle of transaction, without aborting it.
+create table t1 (a int, key(a)) engine=myisam;
+create table t2 (a int);
+insert into t2 values (1);
+# Create a table with a corrupted index file:
+# save an old index file, insert more rows,
+# overwrite the new index file with the old one.
+insert into  t1 (a) values (1);
+flush table t1;
+insert into  t1 (a) values (4);
+flush table t1;
+# Check table is needed to mark the table as crashed.
+check table t1;
+Table	Op	Msg_type	Msg_text
+test.t1	check	warning	Size of datafile is: 14       Should be: 7
+test.t1	check	error	Record-count is not ok; is 2   Should be: 1
+test.t1	check	warning	Found 2 key parts. Should be: 1
+test.t1	check	error	Corrupt
+# At this point we have a corrupt t1
+set autocommit = 0;
+select * from t2;
+a
+1
+# Without fix select from t1 will break the transaction. After the fix
+# transaction should be active and should hold lock on table t2. Alter
+# table from con2 will wait only if the transaction is not broken.
+select * from t1;
+a
+1
+4
+Warnings:
+Error	145	Table 't1' is marked as crashed and should be repaired
+Error	1194	Table 't1' is marked as crashed and should be repaired
+Error	1034	Number of rows changed from 1 to 2
+ALTER TABLE t2 ADD val INT;
+# With fix we should have alter table waiting for t2 lock here.
+ROLLBACK;
+SET autocommit = 1;
+# Cleanup
+drop table t1, t2;

mysql-test/suite/innodb/r/add_constraint.result

+#
+# Bug #20762798 FK DDL: CRASH IN DICT_FOREIGN_REMOVE_FROM_CACHE
+#
+create table t1(a int, b int, key(a),key(b))engine=innodb;
+create table t2(a int, b int, key(a),key(b))engine=innodb;
+alter table t2 add constraint b foreign key (b) references t1(a);
+alter table t1 add constraint b1 foreign key (b) references t2(a);
+alter table t2 add constraint b1 foreign key (b) references t1(a);
+ERROR HY000: Can't create table '#sql-temporary' (errno: 121)
+alter table t2 drop foreign key b;
+alter table t1 drop foreign key b1;
+drop table t2;
+drop table t1;

mysql-test/suite/innodb/t/add_constraint.test

+--source include/have_innodb.inc
+
+--echo #
+--echo # Bug #20762798 FK DDL: CRASH IN DICT_FOREIGN_REMOVE_FROM_CACHE
+--echo #
+
+create table t1(a int, b int, key(a),key(b))engine=innodb;
+create table t2(a int, b int, key(a),key(b))engine=innodb;
+
+alter table t2 add constraint b foreign key (b) references t1(a);
+alter table t1 add constraint b1 foreign key (b) references t2(a);
+
+--replace_regex /'[^']*test\.#sql-[0-9a-f_]*'/'#sql-temporary'/
+--error ER_CANT_CREATE_TABLE
+alter table t2 add constraint b1 foreign key (b) references t1(a);
+
+alter table t2 drop foreign key b;
+alter table t1 drop foreign key b1;
+
+drop table t2;
+drop table t1;

mysql-test/suite/ndb/r/ndb_restore_discover.result

+#
+# 18075170 - sql node restart required to avoid deadlock after
+#            restore
+#
+CREATE TABLE t1 (id INT) ENGINE=NDBCluster;
+CREATE TABLE t2 (id INT) ENGINE=NDBCluster;
+INSERT INTO t1 VALUES (1);
+INSERT INTO t2 VALUES (1);
+DROP TABLE t1;
+DROP TABLE t2;
+SET autocommit = 0;
+SELECT * FROM t1;
+id
+1
+SELECT * FROM t2;
+id
+1
+ROLLBACK;
+SET autocommit = 1;
+drop table t1;
+drop table t2;
+SET autocommit = 0;
+SELECT * FROM t1;
+id
+1
+SELECT * FROM t2;
+id
+1
+ALTER TABLE t1 ADD val INT;
+ROLLBACK;
+SET autocommit = 1;
+drop table t1;
+drop table t2;

mysql-test/suite/ndb/t/ndb_restore_discover.test

+-- source include/have_ndb.inc
+-- source include/count_sessions.inc
+
+--echo #
+--echo # 18075170 - sql node restart required to avoid deadlock after
+--echo #            restore
+--echo #
+# Test Auto Discover option within a transaction
+# and make sure the transaction is not broken.
+CREATE TABLE t1 (id INT) ENGINE=NDBCluster;
+CREATE TABLE t2 (id INT) ENGINE=NDBCluster;
+
+INSERT INTO t1 VALUES (1);
+INSERT INTO t2 VALUES (1);
+
+-- source include/ndb_backup.inc
+
+DROP TABLE t1;
+DROP TABLE t2;
+
+-- source include/ndb_restore_master.inc
+
+SET autocommit = 0;
+SELECT * FROM t1;
+
+# Without fix below select was resulting in DEADLOCK error. With fix select
+# should succeed.
+SELECT * FROM t2;
+ROLLBACK;
+SET autocommit = 1;
+
+drop table t1;
+drop table t2;
+
+#
+# Checking lock preservation in transaction
+#
+# Using existing backup to create the scenario. Tables are deleted as part of
+# above test cleanup. Thus restoring the backup will bring the system to
+# required state.
+-- source include/ndb_restore_master.inc
+
+SET autocommit = 0;
+SELECT * FROM t1;
+SELECT * FROM t2;
+
+connect(con2, localhost, root);
+--SEND ALTER TABLE t1 ADD val INT
+
+connection default;
+# Alter from con2 will be in waiting state as there is a lock on t1 from
+# default connection due to active transaction. We check for this condition
+# then releasing the lock by rollbacking active transaction.
+let $wait_condition=
+  SELECT count(*) = 1 FROM information_schema.processlist WHERE state
+  LIKE "Waiting%" AND info = "ALTER TABLE t1 ADD val INT";
+--source include/wait_condition.inc
+ROLLBACK;
+SET autocommit = 1;
+
+connection con2;
+--REAP
+
+disconnect con2;
+connection default;
+drop table t1;
+drop table t2;
+
+# Wait till all disconnects are completed
+-- source include/wait_until_count_sessions.inc

mysql-test/t/merge_recover.test → mysql-test/t/myisam_recover.test

+--source include/count_sessions.inc
+
+--echo #
+--echo # Tests for corrupted MyISAM tables and MyISAMMRG tables with corrupted
+--echo # children..
 --echo #
---echo # Test of MyISAM MRG tables with corrupted children.
 --echo # Run with --myisam-recover=force option.
 --echo #
 --echo # Preparation: we need to make sure that the merge parent
@@ -111,3 +115,66 @@ set @@global.table_open_cache=default;
 disconnect con1;
 connection default;
 --enable_ps_protocol
+
+--echo #
+--echo # 18075170 - sql node restart required to avoid deadlock after
+--echo #            restore
+--echo #
+--echo # Check that auto-repair for MyISAM tables can now happen in the
+--echo # middle of transaction, without aborting it.
+--enable_prepare_warnings
+
+connection default;
+
+create table t1 (a int, key(a)) engine=myisam;
+create table t2 (a int);
+insert into t2 values (1);
+
+--echo # Create a table with a corrupted index file:
+--echo # save an old index file, insert more rows,
+--echo # overwrite the new index file with the old one.
+insert into  t1 (a) values (1);
+flush table t1;
+--copy_file $MYSQLD_DATADIR/test/t1.MYI $MYSQLD_DATADIR/test/t1_copy.MYI
+insert into  t1 (a) values (4);
+flush table t1;
+--remove_file $MYSQLD_DATADIR/test/t1.MYI
+--copy_file $MYSQLD_DATADIR/test/t1_copy.MYI $MYSQLD_DATADIR/test/t1.MYI
+--remove_file $MYSQLD_DATADIR/test/t1_copy.MYI
+
+--echo # Check table is needed to mark the table as crashed.
+check table t1;
+
+--echo # At this point we have a corrupt t1
+set autocommit = 0;
+select * from t2;
+--echo # Without fix select from t1 will break the transaction. After the fix
+--echo # transaction should be active and should hold lock on table t2. Alter
+--echo # table from con2 will wait only if the transaction is not broken.
+--replace_regex /'.*[\/\\]/'/
+select * from t1;
+
+connect(con2, localhost, root);
+--SEND ALTER TABLE t2 ADD val INT
+
+connection default;
+--echo # With fix we should have alter table waiting for t2 lock here.
+let $wait_condition=
+  SELECT count(*) = 1 FROM information_schema.processlist WHERE state
+  LIKE "Waiting%" AND info = "ALTER TABLE t2 ADD val INT";
+
+--source include/wait_condition.inc
+ROLLBACK;
+SET autocommit = 1;
+
+connection con2;
+--REAP
+
+connection default;
+disconnect con2;
+
+--echo # Cleanup
+drop table t1, t2;
+
+# Wait till all disconnects are completed
+-- source include/wait_until_count_sessions.inc

mysys/ptr_cmp.c

-/* Copyright (c) 2000, 2010, Oracle and/or its affiliates. All rights reserved.
+/* Copyright (c) 2000, 2015, Oracle and/or its affiliates. All rights reserved.
 
    This program is free software; you can redistribute it and/or modify
    it under the terms of the GNU General Public License as published by
@@ -91,6 +91,7 @@ static int ptr_compare(size_t *compare_length, uchar **a, uchar **b)
   reg3 int length= *compare_length;
   reg1 uchar *first,*last;
 
+  DBUG_ASSERT(length > 0);
   first= *a; last= *b;
   while (--length)
   {

regex/regcomp.c

+/* Copyright 1992, 1993, 1994 Henry Spencer.  All rights reserved.
+   See file COPYRIGHT for details.
+
+   This file was modified by Oracle on 2015-05-18 for 32-bit compatibility.
+
+   Modifications copyright (c) 2015, Oracle and/or its affiliates. All rights
+   reserved. */
+
 #include <my_global.h>
 #include <m_string.h>
 #include <m_ctype.h>
@@ -133,12 +141,26 @@ CHARSET_INFO *charset;
 	} else
 		len = strlen((char *)pattern);
 
+	/*
+	 Find the maximum len we can safely process
+	 without a rollover and a mis-malloc.
+	 p->ssize is a sopno is a long (32+ bit signed);
+	 size_t is 16+ bit unsigned.
+	*/
+	{
+	  size_t new_ssize = len / (size_t)2 * (size_t)3 + (size_t)1; /* ugh */
+	  if ((new_ssize < len) ||	/* size_t rolled over */
+	      ((SIZE_T_MAX / sizeof(sop)) < new_ssize) ||	/* malloc arg */
+	      (new_ssize > LONG_MAX))	/* won't fit in ssize */
+		return(REG_ESPACE);	/* MY_REG_ESPACE or MY_REG_INVARG */
+	  p->ssize = new_ssize;
+	}
+
 	/* do the mallocs early so failure handling is easy */
 	g = (struct re_guts *)malloc(sizeof(struct re_guts) +
 							(NC-1)*sizeof(cat_t));
 	if (g == NULL)
 		return(REG_ESPACE);
-	p->ssize = (long) (len/(size_t)2*(size_t)3 + (size_t)1); /* ugh */
 	p->strip = (sop *)malloc(p->ssize * sizeof(sop));
 	p->slen = 0;
 	if (p->strip == NULL) {

sql/sql_base.cc

@@ -4064,10 +4064,11 @@ request_backoff_action(enum_open_table_action action_arg,
     * We met a broken table that needs repair, or a table that
       is not present on this MySQL server and needs re-discovery.
       To perform the action, we need an exclusive metadata lock on
-      the table. Acquiring an X lock while holding other shared
-      locks is very deadlock-prone. If this is a multi- statement
-      transaction that holds metadata locks for completed
-      statements, we don't do it, and report an error instead.
+      the table. Acquiring X lock while holding other shared
+      locks can easily lead to deadlocks. We rely on MDL deadlock
+      detector to discover them. If this is a multi-statement
+      transaction that holds metadata locks for completed statements,
+      we should keep these locks after discovery/repair.
       The action type in this case is OT_DISCOVER or OT_REPAIR.
     * Our attempt to acquire an MDL lock lead to a deadlock,
       detected by the MDL deadlock detector. The current
@@ -4108,7 +4109,7 @@ request_backoff_action(enum_open_table_action action_arg,
       keep tables open between statements and a livelock
       is not possible.
   */
-  if (action_arg != OT_REOPEN_TABLES && m_has_locks)
+  if (action_arg == OT_BACKOFF_AND_RETRY && m_has_locks)
   {
     my_error(ER_LOCK_DEADLOCK, MYF(0));
     m_thd->mark_transaction_to_rollback(true);
@@ -4135,6 +4136,32 @@ request_backoff_action(enum_open_table_action action_arg,
 }
 
 
+/**
+  An error handler to mark transaction to rollback on DEADLOCK error
+  during DISCOVER / REPAIR.
+*/
+class MDL_deadlock_discovery_repair_handler : public Internal_error_handler
+{
+public:
+  virtual bool handle_condition(THD *thd,
+                                  uint sql_errno,
+                                  const char* sqlstate,
+                                  MYSQL_ERROR::enum_warning_level level,
+                                  const char* msg,
+                                  MYSQL_ERROR ** cond_hdl)
+  {
+    if (sql_errno == ER_LOCK_DEADLOCK)
+    {
+      thd->mark_transaction_to_rollback(true);
+    }
+    /*
+      We have marked this transaction to rollback. Return false to allow
+      error to be reported or handled by other handlers.
+    */
+    return false;
+  }
+};
+
 /**
    Recover from failed attempt of open table by performing requested action.
 
@@ -4150,6 +4177,12 @@ Open_table_context::
 recover_from_failed_open()
 {
   bool result= FALSE;
+  MDL_deadlock_discovery_repair_handler handler;
+  /*
+    Install error handler to mark transaction to rollback on DEADLOCK error.
+  */
+  m_thd->push_internal_handler(&handler);
+
   /* Execute the action. */
   switch (m_action)
   {
@@ -4171,7 +4204,12 @@ recover_from_failed_open()
 
         m_thd->warning_info->clear_warning_info(m_thd->query_id);
         m_thd->clear_error();                 // Clear error message
-        m_thd->mdl_context.release_transactional_locks();
+        /*
+          Rollback to start of the current statement to release exclusive lock
+          on table which was discovered but preserve locks from previous statements
+          in current transaction.
+        */
+        m_thd->mdl_context.rollback_to_savepoint(start_of_statement_svp());
         break;
       }
     case OT_REPAIR:
@@ -4185,12 +4223,18 @@ recover_from_failed_open()
                          m_failed_table->table_name, FALSE);
 
         result= auto_repair_table(m_thd, m_failed_table);
-        m_thd->mdl_context.release_transactional_locks();
+        /*
+          Rollback to start of the current statement to release exclusive lock
+          on table which was discovered but preserve locks from previous statements
+          in current transaction.
+        */
+        m_thd->mdl_context.rollback_to_savepoint(start_of_statement_svp());
         break;
       }
     default:
       DBUG_ASSERT(0);
   }
+  m_thd->pop_internal_handler();
   /*
     Reset the pointers to conflicting MDL request and the
     TABLE_LIST element, set when we need auto-discovery or repair,

storage/innobase/dict/dict0dict.c

@@ -2534,8 +2534,9 @@ dict_foreign_remove_from_cache(
 			const ib_rbt_node_t*	node
 				= rbt_lookup(rbt, foreign->id);
 
-			if (node) {
-				dict_foreign_t*	val = *(dict_foreign_t**) node->value;
+			if (node != NULL) {
+				dict_foreign_t*	val
+					= *(dict_foreign_t**) node->value;
 
 				if (val == foreign) {
 					rbt_delete(rbt, foreign->id);
@@ -2555,9 +2556,10 @@ dict_foreign_remove_from_cache(
 		if (rbt != NULL && foreign->id != NULL) {
 			const ib_rbt_node_t*	node
 				= rbt_lookup(rbt, foreign->id);
-			if (node) {
 
-				dict_foreign_t*	val = *(dict_foreign_t**) node->value;
+			if (node != NULL) {
+				dict_foreign_t*	val
+					= *(dict_foreign_t**) node->value;
 
 				if (val == foreign) {
 					rbt_delete(rbt, foreign->id);

storage/innobase/include/row0purge.h

 /*****************************************************************************
 
-Copyright (c) 1997, 2009, Innobase Oy. All Rights Reserved.
+Copyright (c) 1997, 2015, Oracle and/or its affiliates. All Rights Reserved.
 
 This program is free software; you can redistribute it and/or modify it under
 the terms of the GNU General Public License as published by the Free Software
@@ -111,6 +111,17 @@ struct purge_node_struct{
 				purge of a row */
 };
 
+#ifdef UNIV_DEBUG
+/***********************************************************//**
+Validate the persisent cursor in the purge node. The purge node has two
+references to the clustered index record - one via the ref member, and the
+other via the persistent cursor.  These two references must match each
+other if the found_clust flag is set.
+@return true if the persistent cursor is consistent with the ref member.*/
+ibool
+row_purge_validate_pcur(purge_node_t* node);
+#endif /* UNIV_DEBUG */
+
 #ifndef UNIV_NONINL
 #include "row0purge.ic"
 #endif

storage/innobase/os/os0file.c

 /***********************************************************************
 
-Copyright (c) 1995, 2010, Innobase Oy. All Rights Reserved.
+Copyright (c) 1995, 2015, Oracle and/or its affiliates. All Rights Reserved.
 Copyright (c) 2009, Percona Inc.
 
 Portions of this file contain modifications contributed and copyrighted
@@ -1276,16 +1276,19 @@ os_file_create_simple_no_error_handling_func(
 #else /* __WIN__ */
 	os_file_t	file;
 	int		create_flag;
+	const char*	mode_str	= NULL;
 
 	ut_a(name);
 
 	if (create_mode == OS_FILE_OPEN) {
+		mode_str = "OPEN";
 		if (access_type == OS_FILE_READ_ONLY) {
 			create_flag = O_RDONLY;
 		} else {
 			create_flag = O_RDWR;
 		}
 	} else if (create_mode == OS_FILE_CREATE) {
+		mode_str = "CREATE";
 		create_flag = O_RDWR | O_CREAT | O_EXCL;
 	} else {
 		create_flag = 0;
@@ -1310,6 +1313,14 @@ os_file_create_simple_no_error_handling_func(
 #endif
 	} else {
 		*success = TRUE;
+
+		/* This function is always called for data files, we should
+		disable OS caching (O_DIRECT) here as we do in
+		os_file_create_func(), so we open the same file in the same
+		mode, see man page of open(2). */
+		if (srv_unix_file_flush_method == SRV_UNIX_O_DIRECT) {
+			os_file_set_nocache(file, name, mode_str);
+		}
 	}
 
 	return(file);

storage/innobase/row/row0purge.c

 /*****************************************************************************
 
-Copyright (c) 1997, 2011, Oracle and/or its affiliates. All Rights Reserved.
+Copyright (c) 1997, 2015, Oracle and/or its affiliates. All Rights Reserved.
 
 This program is free software; you can redistribute it and/or modify it under
 the terms of the GNU General Public License as published by the Free Software
@@ -43,6 +43,7 @@ Created 3/14/1997 Heikki Tuuri
 #include "row0vers.h"
 #include "row0mysql.h"
 #include "log0log.h"
+#include "rem0cmp.h"
 
 /*************************************************************************
 IMPORTANT NOTE: Any operation that generates redo MUST check that there
@@ -80,7 +81,7 @@ row_purge_node_create(
 
 /***********************************************************//**
 Repositions the pcur in the purge node on the clustered index record,
-if found.
+if found. If the record is not found, close pcur.
 @return	TRUE if the record was found */
 static
 ibool
@@ -90,23 +91,28 @@ row_purge_reposition_pcur(
 	purge_node_t*	node,	/*!< in: row purge node */
 	mtr_t*		mtr)	/*!< in: mtr */
 {
-	ibool	found;
-
 	if (node->found_clust) {
-		found = btr_pcur_restore_position(mode, &(node->pcur), mtr);
+		ut_ad(row_purge_validate_pcur(node));
 
-		return(found);
-	}
+		node->found_clust = btr_pcur_restore_position(
+			mode, &(node->pcur), mtr);
 
-	found = row_search_on_row_ref(&(node->pcur), mode, node->table,
-				      node->ref, mtr);
-	node->found_clust = found;
+	} else {
 
-	if (found) {
+		node->found_clust = row_search_on_row_ref(
+			&(node->pcur), mode, node->table, node->ref, mtr);
+
+		if (node->found_clust) {
 			btr_pcur_store_position(&(node->pcur), mtr);
 		}
+	}
+
+	/* Close the current cursor if we fail to position it correctly. */
+	if (!node->found_clust) {
+		btr_pcur_close(&node->pcur);
+	}
 
-	return(found);
+	return(node->found_clust);
 }
 
 /***********************************************************//**
@@ -143,8 +149,8 @@ row_purge_remove_clust_if_poss_low(
 
 	if (!success) {
 		/* The record is already removed */
-
-		btr_pcur_commit_specify_mtr(pcur, &mtr);
+		/* Persistent cursor is closed if reposition fails. */
+		mtr_commit(&mtr);
 
 		return(TRUE);
 	}
@@ -258,7 +264,12 @@ row_purge_poss_sec(
 						 btr_pcur_get_rec(&node->pcur),
 						 &mtr, index, entry);
 
+	/* Persistent cursor is closed if reposition fails. */
+	if (node->found_clust) {
 		btr_pcur_commit_specify_mtr(&node->pcur, &mtr);
+	} else {
+		mtr_commit(&mtr);
+	}
 
 	return(can_delete);
 }
@@ -806,3 +817,53 @@ row_purge_step(
 
 	return(thr);
 }
+
+#ifdef UNIV_DEBUG
+/***********************************************************//**
+Validate the persisent cursor in the purge node. The purge node has two
+references to the clustered index record - one via the ref member, and the
+other via the persistent cursor.  These two references must match each
+other if the found_clust flag is set.
+@return true if the stored copy of persistent cursor is consistent
+with the ref member.*/
+ibool
+row_purge_validate_pcur(
+	purge_node_t*	node)
+{
+	dict_index_t*	clust_index;
+	ulint*		offsets;
+	int		st;
+
+	if (!node->found_clust) {
+		return(TRUE);
+	}
+
+	if (node->index == NULL) {
+		return(TRUE);
+	}
+
+	if (node->pcur.old_stored != BTR_PCUR_OLD_STORED) {
+		return(TRUE);
+	}
+
+	clust_index = node->pcur.btr_cur.index;
+
+	offsets = rec_get_offsets(node->pcur.old_rec, clust_index, NULL,
+				  node->pcur.old_n_fields, &node->heap);
+
+	/* Here we are comparing the purge ref record and the stored initial
+	part in persistent cursor. Both cases we store n_uniq fields of the
+	cluster index and so it is fine to do the comparison. We note this
+	dependency here as pcur and ref belong to different modules. */
+	st = cmp_dtuple_rec(node->ref, node->pcur.old_rec, offsets);
+
+	if (st != 0) {
+		fprintf(stderr, "Purge node pcur validation failed\n");
+		dtuple_print(stderr, node->ref);
+		rec_print(stderr, node->pcur.old_rec, clust_index);
+		return(FALSE);
+	}
+
+	return(TRUE);
+}
+#endif /* UNIV_DEBUG */