Added testcase for select count() during transaction with failures

added fix for keeping "records" up to date when execute() fails

mysql-test/r/ndb_insert.result

@@ -421,11 +421,25 @@ INSERT INTO t1 VALUES
 (6,6,6),(7,7,7),(8,8,8),(9,9,9),(10,10,10);
 ERROR 23000: Duplicate entry '10' for key 1
 begin;
+SELECT COUNT(*) FROM t1;
+COUNT(*)
+2000
+INSERT INTO t1 VALUES 
+(2001,2001,2001),(2002,2002,2002),(2003,2003,2003),(2004,2004,2004),(2005,2005,2005);
+SELECT COUNT(*) FROM t1;
+COUNT(*)
+2005
 INSERT INTO t1 VALUES 
 (1,1,1),(2,2,2),(3,3,3),(4,4,4),(5,5,5),
 (6,6,6),(7,7,7),(8,8,8),(9,9,9),(10,10,10);
 ERROR 23000: Duplicate entry '10' for key 1
+SELECT COUNT(*) FROM t1;
+COUNT(*)
+2000
 commit;
+SELECT COUNT(*) FROM t1;
+COUNT(*)
+2000
 insert into t1 select * from t1 where b < 10 order by pk1;
 ERROR 23000: Duplicate entry '9' for key 1
 DROP TABLE t1;

mysql-test/t/ndb_insert.test

@@ -443,13 +443,20 @@ begin;
 
 #
 # Insert duplicate rows, inside transaction
-# 
+# since failing inserts rollbacks whole transaction
+# all select count (except second) return same value
+#
+SELECT COUNT(*) FROM t1;
+INSERT INTO t1 VALUES 
+(2001,2001,2001),(2002,2002,2002),(2003,2003,2003),(2004,2004,2004),(2005,2005,2005);
+SELECT COUNT(*) FROM t1;
 --error 1062
 INSERT INTO t1 VALUES 
 (1,1,1),(2,2,2),(3,3,3),(4,4,4),(5,5,5),
 (6,6,6),(7,7,7),(8,8,8),(9,9,9),(10,10,10);
-
+SELECT COUNT(*) FROM t1;
 commit;
+SELECT COUNT(*) FROM t1;
 
 #
 # Insert duplicate rows using "insert .. select" 

sql/ha_ndbcluster.cc

@@ -183,6 +183,14 @@ void ha_ndbcluster::records_update()
   DBUG_VOID_RETURN;
 }
 
+void ha_ndbcluster::no_uncommitted_rows_execute_failure()
+{
+  DBUG_ENTER("ha_ndbcluster::no_uncommitted_rows_execute_failure");
+  struct Ndb_table_local_info *info= (struct Ndb_table_local_info *)m_table_info;
+  info->no_uncommitted_rows_count= 0;
+  DBUG_VOID_RETURN;
+}
+
 void ha_ndbcluster::no_uncommitted_rows_init(THD *thd)
 {
   DBUG_ENTER("ha_ndbcluster::no_uncommitted_rows_init");
@@ -1576,6 +1584,7 @@ int ha_ndbcluster::write_row(byte *record)
       if (trans->execute(NoCommit) != 0)
       {
 	skip_auto_increment= true;
+	no_uncommitted_rows_execute_failure();
 	DBUG_RETURN(ndb_err(trans));
       }
     }
@@ -1584,6 +1593,7 @@ int ha_ndbcluster::write_row(byte *record)
       if (trans->execute(Commit) != 0)
       {
 	skip_auto_increment= true;
+	no_uncommitted_rows_execute_failure();
 	DBUG_RETURN(ndb_err(trans));
       }
       trans->restart();
@@ -1746,8 +1756,10 @@ int ha_ndbcluster::update_row(const byte *old_data, byte *new_data)
   }
 
   // Execute update operation
-  if (!cursor && trans->execute(NoCommit) != 0)
+  if (!cursor && trans->execute(NoCommit) != 0) {
+    no_uncommitted_rows_execute_failure();
     DBUG_RETURN(ndb_err(trans));
+  }
   
   DBUG_RETURN(0);
 }
@@ -1814,8 +1826,10 @@ int ha_ndbcluster::delete_row(const byte *record)
   }
   
   // Execute delete operation
-  if (trans->execute(NoCommit) != 0)
+  if (trans->execute(NoCommit) != 0) {
+    no_uncommitted_rows_execute_failure();
     DBUG_RETURN(ndb_err(trans));
+  }
   DBUG_RETURN(0);
 }
   
@@ -2227,8 +2241,10 @@ int ha_ndbcluster::close_scan()
       deleteing/updating transaction before closing the scan    
     */
     DBUG_PRINT("info", ("ops_pending: %d", ops_pending));    
-    if (trans->execute(NoCommit) != 0)
+    if (trans->execute(NoCommit) != 0) {
+      no_uncommitted_rows_execute_failure();
       DBUG_RETURN(ndb_err(trans));
+    }
     ops_pending= 0;
   }
   
@@ -2532,8 +2548,10 @@ int ha_ndbcluster::end_bulk_insert()
                         "rows_inserted:%d, bulk_insert_rows: %d", 
                         rows_inserted, bulk_insert_rows)); 
     bulk_insert_not_flushed= false;
-    if (trans->execute(NoCommit) != 0)
+    if (trans->execute(NoCommit) != 0) {
+      no_uncommitted_rows_execute_failure();
       my_errno= error= ndb_err(trans);
+    }
   }
 
   rows_inserted= 0;

sql/ha_ndbcluster.h

@@ -254,6 +254,7 @@ class ha_ndbcluster: public handler
   uint dupkey;
 
   void records_update();
+  void no_uncommitted_rows_execute_failure();
   void no_uncommitted_rows_update(int);
   void no_uncommitted_rows_init(THD *);
   void no_uncommitted_rows_reset(THD *);