bzr commit into mysql-5.1-telco-6.3 branch (tomas.ulin:2627)

[Tomas Ulin]

#At file:///home/tomas/mysql_src/cge-6.3/

 2627 Tomas Ulin	2008-06-18 [merge]
      automerge
modified:
  sql/ha_ndbcluster_binlog.cc
  sql/rpl_injector.cc
  sql/rpl_injector.h

=== modified file 'sql/ha_ndbcluster_binlog.cc'
--- a/sql/ha_ndbcluster_binlog.cc	2008-06-16 14:24:40 +0000
+++ b/sql/ha_ndbcluster_binlog.cc	2008-06-18 13:49:43 +0000
@@ -4530,7 +4530,8 @@ ndb_find_binlog_index_row(ndb_binlog_ind
 static int
 ndb_binlog_thread_handle_data_event(Ndb *ndb, NdbEventOperation *pOp,
                                     ndb_binlog_index_row **rows,
-                                    injector::transaction &trans)
+                                    injector::transaction &trans,
+                                    unsigned &trans_row_count)
 {
   Ndb_event_data *event_data= (Ndb_event_data *) pOp->getCustomData();
   TABLE *table= event_data->table;
@@ -4627,6 +4628,7 @@ ndb_binlog_thread_handle_data_event(Ndb 
   {
   case NDBEVENT::TE_INSERT:
     row->n_inserts++;
+    trans_row_count++;
     DBUG_PRINT("info", ("INSERT INTO %s.%s",
                         table->s->db.str, table->s->table_name.str));
     {
@@ -4649,6 +4651,7 @@ ndb_binlog_thread_handle_data_event(Ndb 
     break;
   case NDBEVENT::TE_DELETE:
     row->n_deletes++;
+    trans_row_count++;
     DBUG_PRINT("info",("DELETE FROM %s.%s",
                        table->s->db.str, table->s->table_name.str));
     {
@@ -4689,6 +4692,7 @@ ndb_binlog_thread_handle_data_event(Ndb 
     break;
   case NDBEVENT::TE_UPDATE:
     row->n_updates++;
+    trans_row_count++;
     DBUG_PRINT("info", ("UPDATE %s.%s",
                         table->s->db.str, table->s->table_name.str));
     {
@@ -5331,6 +5335,7 @@ restart:
         bzero((char*)&_row, sizeof(_row));
         thd->variables.character_set_client= &my_charset_latin1;
         injector::transaction trans;
+        unsigned trans_row_count= 0;
         // pass table map before epoch
         {
           Uint32 iter= 0;
@@ -5497,7 +5502,7 @@ restart:
 #endif
           if ((unsigned) pOp->getEventType() <
               (unsigned) NDBEVENT::TE_FIRST_NON_DATA_EVENT)
-            ndb_binlog_thread_handle_data_event(i_ndb, pOp, &rows, trans);
+            ndb_binlog_thread_handle_data_event(i_ndb, pOp, &rows, trans,
trans_row_count);
           else
           {
             // set injector_ndb database/schema from table internal name
@@ -5540,9 +5545,20 @@ restart:
         write_timer.stop();
 #endif
 
-        if (trans.good())
+        while (trans.good())
         {
-          //DBUG_ASSERT(row.n_inserts || row.n_updates || row.n_deletes);
+          if (trans_row_count == 0)
+          {
+            /* nothing to commit, rollback instead */
+            if (int r= trans.rollback())
+            {
+              sql_print_error("NDB Binlog: "
+                              "Error during ROLLBACK of GCI %u/%u. Error: %d",
+                              uint(gci >> 32), uint(gci), r);
+              /* TODO: Further handling? */
+            }
+            break;
+          }
           thd->proc_info= "Committing events to binlog";
           injector::transaction::binlog_pos start= trans.start_pos();
           if (int r= trans.commit())
@@ -5564,6 +5580,7 @@ restart:
             do_check_ndb_binlog_index= 0;
           }
           ndb_latest_applied_binlog_epoch= gci;
+          break;
         }
         ndb_latest_handled_binlog_epoch= gci;
 

=== modified file 'sql/rpl_injector.cc'
--- a/sql/rpl_injector.cc	2008-02-19 11:43:01 +0000
+++ b/sql/rpl_injector.cc	2008-06-18 12:53:42 +0000
@@ -86,6 +86,14 @@ int injector::transaction::commit()
    DBUG_RETURN(0);
 }
 
+int injector::transaction::rollback()
+{
+   DBUG_ENTER("injector::transaction::rollback()");
+   ha_autocommit_or_rollback(m_thd, 1 /* error to get rollback */);
+   end_trans(m_thd, ROLLBACK);
+   DBUG_RETURN(0);
+}
+
 int injector::transaction::use_table(server_id_type sid, table tbl)
 {
   DBUG_ENTER("injector::transaction::use_table");

=== modified file 'sql/rpl_injector.h'
--- a/sql/rpl_injector.h	2007-05-10 09:59:39 +0000
+++ b/sql/rpl_injector.h	2008-06-18 12:53:42 +0000
@@ -210,6 +210,14 @@ public:
       int commit();
 
       /*
+        Rollback a transaction.
+
+        This member function will clean up after a sequence of *_row calls by,
+        for example, releasing resource and unlocking files.
+      */
+      int rollback();
+
+      /*
         Get the position for the start of the transaction.
 
         Returns the position in the binary log of the first event in this