Merge joreland@bk-internal.mysql.com:/home/bk/mysql-5.1-new-ndb

into  perch.ndb.mysql.com:/home/jonas/src/51-ndb

mysql-test/include/have_multi_ndb.inc

@@ -5,10 +5,6 @@ connect (server2,127.0.0.1,root,,test,$MASTER_MYPORT1,);
 # Check that server1 has NDB  support
 connection server1;
 disable_query_log;
---disable_warnings
-drop table if exists t1, t2;
---enable_warnings
-flush tables;
 --require r/true.require
 select (support = 'YES' or support = 'DEFAULT') as `TRUE` from information_schema.engines where engine = 'ndbcluster';
 --source include/ndb_not_readonly.inc
@@ -17,14 +13,32 @@ enable_query_log;
 # Check that server2 has NDB support
 connection server2;
 disable_query_log;
---disable_warnings
-drop table if exists t1, t2;
---enable_warnings
-flush tables;
 --require r/true.require
 select (support = 'YES' or support = 'DEFAULT') as `TRUE` from information_schema.engines where engine = 'ndbcluster';
 --source include/ndb_not_readonly.inc
 enable_query_log;
 
-# Set the default connection to 'server1'
+# cleanup
+
+connection server1;
+disable_query_log;
+disable_warnings;
+--error 0,1051
+drop table if exists t1,t2,t3,t4,t5,t6,t7,t8,t9,t10;
+flush tables;
+flush status;
+enable_warnings;
+enable_query_log;
+
+connection server2;
+disable_query_log;
+disable_warnings;
+--error 0,1051
+drop table if exists t1,t2,t3,t4,t5,t6,t7,t8,t9,t10;
+flush tables;
+flush status;
+enable_warnings;
+enable_query_log;
+
+# Set the default connection
 connection server1;

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

@@ -1,4 +1,5 @@
 drop table if exists t1, t2, t3, t4;
+flush status;
 drop table if exists t1, t2, t3, t4;
 flush status;
 create table t1 (a int) engine=ndbcluster;
@@ -132,11 +133,11 @@ master_epoch, count))
 engine ndb;
 show tables like '%$%';
 Tables_in_test (%$%)
-t1$EX
+t1$ex
 use test;
 show tables like '%$%';
 Tables_in_test (%$%)
-t1$EX
+t1$ex
 drop table `test`.`t1$EX`;
 show tables like '%$%';
 Tables_in_test (%$%)

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

@@ -1,4 +1,5 @@
 drop table if exists t1, t2, t3, t4;
+flush status;
 drop table if exists t1, t2, t3, t4;
 flush status;
 create table t1 (a int) engine=ndbcluster;

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

@@ -0,0 +1,75 @@
+DROP TABLE IF EXISTS t1;
+create table t1 (a int not null primary key, b int not null, c int,
+unique index_b (b) using hash)
+engine ndb;
+insert into t1 values (1,10,1),(2,9,1),(3,8,1),(4,7,1),(5,6,1),(6,5,2),(7,4,2),(8,3,2),
+(9,2,2),(10,1,2);
+update t1 set c = 111, b = 20 where a = 1;
+select * from t1 where a = 1 order by a;
+a	b	c
+1	20	111
+delete from t1 where a = 1;
+select * from t1 where a = 1 order by a;
+a	b	c
+update t1 set c = 12, b = 19 where b = 2;
+select * from t1 where b = 2 order by a;
+a	b	c
+delete from t1 where b = 19;
+select * from t1 where b = 19 order by a;
+a	b	c
+update t1 set c = 22 where a = 10 or a >= 10;
+select * from t1 order by a;
+a	b	c
+2	9	1
+3	8	1
+4	7	1
+5	6	1
+6	5	2
+7	4	2
+8	3	2
+10	1	22
+update t1 set c = 23 where a in (8,10);
+select * from t1 order by a;
+a	b	c
+2	9	1
+3	8	1
+4	7	1
+5	6	1
+6	5	2
+7	4	2
+8	3	23
+10	1	23
+update t1 set c = 23 where a in (7,8) or a >= 10;
+select * from t1 order by a;
+a	b	c
+2	9	1
+3	8	1
+4	7	1
+5	6	1
+6	5	2
+7	4	23
+8	3	23
+10	1	23
+update t1 set c = 11 where a = 3 or b = 7;
+select * from t1 where a = 3 or b = 7 order by a;
+a	b	c
+3	8	11
+4	7	11
+update t1 set a = 13, b = 20 where a = 3;
+select * from t1 where a = 13 order by a;
+a	b	c
+13	20	11
+update t1 set a = 12, b = 19 where b = 7;
+select * from t1 where b = 19 order by a;
+a	b	c
+12	19	11
+select * from t1 where b = 7 order by a;
+a	b	c
+update t1 set c = 12, b = 29 where a = 5 and b = 6;
+select * from t1 where b = 19 order by a;
+a	b	c
+12	19	11
+delete from t1 where b = 6 and c = 12;
+select * from t1 where b = 6 order by a;
+a	b	c
+drop table t1;

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

@@ -4,11 +4,11 @@
 --disable_warnings
 connection server2;
 drop table if exists t1, t2, t3, t4;
+flush status;
 connection server1;
 drop table if exists t1, t2, t3, t4;
---enable_warnings
-
 flush status;
+--enable_warnings
 
 # Create test tables on server1
 create table t1 (a int) engine=ndbcluster;
@@ -139,9 +139,11 @@ create table `test`.`t1$EX`
 
 # check that table shows up ok on both servers
 # before bugfix table would not show up on server2
+--replace_regex /EX/ex/
 show tables like '%$%';
 connection server2;
 use test;
+--replace_regex /EX/ex/
 show tables like '%$%';
 
 # check cleanup

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

@@ -6,11 +6,12 @@
 --disable_warnings
 connection server2;
 drop table if exists t1, t2, t3, t4;
+flush status;
 connection server1;
 drop table if exists t1, t2, t3, t4;
+flush status;
 --enable_warnings
 
-flush status;
 
 # Create test tables on server1
 create table t1 (a int) engine=ndbcluster;

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

@@ -0,0 +1,79 @@
+-- source include/have_ndb.inc
+-- source include/not_embedded.inc
+
+--disable_warnings
+DROP TABLE IF EXISTS t1;
+--enable_warnings
+
+#
+# New test case for WL 3686 (which is not until CGE-6.3)
+# but test is committed in 5.1 to verify consistant results.
+#
+# When only constant expressions in update statements and
+# only PK or UK in WHERE clause. No extra WHERE parts are
+# allowed. WL #3687 takes of more advanced variants of
+# avoiding the read before the update/delete
+
+create table t1 (a int not null primary key, b int not null, c int,
+                 unique index_b (b) using hash)
+engine ndb;
+
+insert into t1 values (1,10,1),(2,9,1),(3,8,1),(4,7,1),(5,6,1),(6,5,2),(7,4,2),(8,3,2),
+            (9,2,2),(10,1,2);
+
+# These ones should use optimisation
+
+update t1 set c = 111, b = 20 where a = 1;
+
+select * from t1 where a = 1 order by a;
+
+delete from t1 where a = 1;
+
+select * from t1 where a = 1 order by a;
+
+update t1 set c = 12, b = 19 where b = 2;
+
+select * from t1 where b = 2 order by a;
+
+delete from t1 where b = 19;
+
+select * from t1 where b = 19 order by a;
+
+update t1 set c = 22 where a = 10 or a >= 10;
+
+select * from t1 order by a;
+
+update t1 set c = 23 where a in (8,10);
+
+select * from t1 order by a;
+
+update t1 set c = 23 where a in (7,8) or a >= 10;
+
+select * from t1 order by a;
+
+# These ones should not use optimisation
+
+update t1 set c = 11 where a = 3 or b = 7;
+
+select * from t1 where a = 3 or b = 7 order by a;
+
+update t1 set a = 13, b = 20 where a = 3;
+
+select * from t1 where a = 13 order by a;
+
+update t1 set a = 12, b = 19 where b = 7;
+
+select * from t1 where b = 19 order by a;
+
+select * from t1 where b = 7 order by a;
+
+update t1 set c = 12, b = 29 where a = 5 and b = 6;
+
+select * from t1 where b = 19 order by a;
+
+delete from t1 where b = 6 and c = 12;
+
+select * from t1 where b = 6 order by a;
+
+drop table t1;
+

sql/ha_ndbcluster.cc

@@ -618,7 +618,7 @@ bool ha_ndbcluster::get_error_message(int error,
   DBUG_ENTER("ha_ndbcluster::get_error_message");
   DBUG_PRINT("enter", ("error: %d", error));
 
-  Ndb *ndb= get_ndb();
+  Ndb *ndb= check_ndb_in_thd(current_thd);
   if (!ndb)
     DBUG_RETURN(FALSE);
 

sql/ha_ndbcluster_binlog.cc

@@ -241,18 +241,22 @@ static void dbug_print_table(const char *info, TABLE *table)
 static void run_query(THD *thd, char *buf, char *end,
                       const int *no_print_error, my_bool disable_binlog)
 {
-  ulong save_query_length= thd->query_length;
+  ulong save_thd_query_length= thd->query_length;
-  char *save_query= thd->query;
+  char *save_thd_query= thd->query;
-  ulong save_thread_id= thd->variables.pseudo_thread_id;
+  struct system_variables save_thd_variables= thd->variables;
+  struct system_status_var save_thd_status_var= thd->status_var;
+  THD_TRANS save_thd_transaction_all= thd->transaction.all;
+  THD_TRANS save_thd_transaction_stmt= thd->transaction.stmt;
   ulonglong save_thd_options= thd->options;
   DBUG_ASSERT(sizeof(save_thd_options) == sizeof(thd->options));
-  NET save_net= thd->net;
+  NET save_thd_net= thd->net;
   const char* found_semicolon= NULL;
 
   bzero((char*) &thd->net, sizeof(NET));
   thd->query_length= end - buf;
   thd->query= buf;
   thd->variables.pseudo_thread_id= thread_id;
+  thd->transaction.stmt.modified_non_trans_table= FALSE;
   if (disable_binlog)
     thd->options&= ~OPTION_BIN_LOG;
     
@@ -275,10 +279,13 @@ static void run_query(THD *thd, char *buf, char *end,
   }
 
   thd->options= save_thd_options;
-  thd->query_length= save_query_length;
+  thd->query_length= save_thd_query_length;
-  thd->query= save_query;
+  thd->query= save_thd_query;
-  thd->variables.pseudo_thread_id= save_thread_id;
+  thd->variables= save_thd_variables;
-  thd->net= save_net;
+  thd->status_var= save_thd_status_var;
+  thd->transaction.all= save_thd_transaction_all;
+  thd->transaction.stmt= save_thd_transaction_stmt;
+  thd->net= save_thd_net;
 
   if (thd == injector_thd)
   {
@@ -777,8 +784,9 @@ static int ndbcluster_create_ndb_apply_status_table(THD *thd)
                    " end_pos BIGINT UNSIGNED NOT NULL, "
                    " PRIMARY KEY USING HASH (server_id) ) ENGINE=NDB");
 
-  const int no_print_error[4]= {ER_TABLE_EXISTS_ERROR,
+  const int no_print_error[5]= {ER_TABLE_EXISTS_ERROR,
                                 701,
+                                702,
                                 4009,
                                 0}; // do not print error 701 etc
   run_query(thd, buf, end, no_print_error, TRUE);
@@ -837,8 +845,9 @@ static int ndbcluster_create_schema_table(THD *thd)
                    " type INT UNSIGNED NOT NULL,"
                    " PRIMARY KEY USING HASH (db,name) ) ENGINE=NDB");
 
-  const int no_print_error[4]= {ER_TABLE_EXISTS_ERROR,
+  const int no_print_error[5]= {ER_TABLE_EXISTS_ERROR,
                                 701,
+                                702,
                                 4009,
                                 0}; // do not print error 701 etc
   run_query(thd, buf, end, no_print_error, TRUE);
@@ -3587,6 +3596,7 @@ pthread_handler_t ndb_binlog_thread_func(void *arg)
   Thd_ndb *thd_ndb=0;
   int ndb_update_ndb_binlog_index= 1;
   injector *inj= injector::instance();
+  uint incident_id= 0;
 
 #ifdef RUN_NDB_BINLOG_TIMER
   Timer main_timer;
@@ -3692,18 +3702,64 @@ pthread_handler_t ndb_binlog_thread_func(void *arg)
   pthread_mutex_unlock(&injector_mutex);
   pthread_cond_signal(&injector_cond);
 
+  /*
+    wait for mysql server to start (so that the binlog is started
+    and thus can receive the first GAP event)
+  */
+  pthread_mutex_lock(&LOCK_server_started);
+  while (!mysqld_server_started)
+  {
+    struct timespec abstime;
+    set_timespec(abstime, 1);
+    pthread_cond_timedwait(&COND_server_started, &LOCK_server_started,
+                           &abstime);
+    if (ndbcluster_terminating)
+    {
+      pthread_mutex_unlock(&LOCK_server_started);
+      pthread_mutex_lock(&LOCK_ndb_util_thread);
+      goto err;
+    }
+  }
+  pthread_mutex_unlock(&LOCK_server_started);
 restart:
   /*
     Main NDB Injector loop
   */
+  while (ndb_binlog_running)
   {
     /*
-      Always insert a GAP event as we cannot know what has happened in the cluster
+      check if it is the first log, if so we do not insert a GAP event
-      while not being connected.
+      as there is really no log to have a GAP in
     */
-    LEX_STRING const msg= { C_STRING_WITH_LEN("Cluster connect") };
+    if (incident_id == 0)
-    inj->record_incident(thd, INCIDENT_LOST_EVENTS, msg);
+    {
+      LOG_INFO log_info;
+      mysql_bin_log.get_current_log(&log_info);
+      int len=  strlen(log_info.log_file_name);
+      uint no= 0;
+      if ((sscanf(log_info.log_file_name + len - 6, "%u", &no) == 1) &&
+          no == 1)
+      {
+        /* this is the fist log, so skip GAP event */
+        break;
+      }
+    }
+
+    /*
+      Always insert a GAP event as we cannot know what has happened
+      in the cluster while not being connected.
+    */
+    LEX_STRING const msg[2]=
+      {
+        { C_STRING_WITH_LEN("mysqld startup")    },
+        { C_STRING_WITH_LEN("cluster disconnect")}
+      };
+    IF_DBUG(int error=)
+      inj->record_incident(thd, INCIDENT_LOST_EVENTS, msg[incident_id]);
+    DBUG_ASSERT(!error);
+    break;
   }
+  incident_id= 1;
   {
     thd->proc_info= "Waiting for ndbcluster to start";
 

sql/sql_update.cc

@@ -679,7 +679,7 @@ int mysql_update(THD *thd,
         */
         if (will_batch &&
             ((error= table->file->exec_bulk_update(&dup_key_found)) ||
-            !dup_key_found))
+             dup_key_found))
         {
  	  if (error)
           {

storage/ndb/test/include/dbutil.hpp

@@ -0,0 +1,97 @@
+// dbutil.h: interface for the database utilities class.
+//////////////////////////////////////////////////////////////////////
+// Supplies a database to the test application
+//////////////////////////////////////////////////////////////////////
+
+#ifndef DBUTIL_HPP
+#define DBUTIL_HPP
+
+#if _MSC_VER > 1000
+#pragma once
+#endif // _MSC_VER > 1000
+#include <time.h>
+#include <stdio.h>
+#include <string.h>
+#include <mysql.h>
+//include "rand.h"
+#include <stdlib.h>
+
+//#define DEBUG
+#define  DIE_UNLESS(expr) \
+           ((void) ((expr) ? 0 : (Die(__FILE__, __LINE__, #expr), 0)))
+#define DIE(expr) \
+          Die(__FILE__, __LINE__, #expr)
+#define myerror(msg) PrintError(msg)
+#define mysterror(stmt, msg) PrintStError(stmt, msg)
+#define  CheckStmt(stmt) \
+{ \
+if ( stmt == 0) \
+  myerror(NULL); \
+DIE_UNLESS(stmt != 0); \
+}
+
+#define  check_execute(stmt, r) \
+{ \
+if (r) \
+  mysterror(stmt, NULL); \
+DIE_UNLESS(r == 0);\
+}
+
+#define TRUE 1
+#define FALSE 0
+
+
+class dbutil
+{
+public:
+
+  dbutil(const char * databaseName);
+  ~dbutil();
+
+  void  DatabaseLogin(const char * system,
+                      const char * usr,
+                      const char * password,
+                      unsigned int portIn,
+                      const char * sockIn,
+                      bool transactional);
+  char * GetDbName(){return dbs;};
+  char * GetUser(){return user;};
+  char * GetPassword(){return pass;};
+  char * GetHost(){return host;};
+  char * GetSocket(){return socket;};
+  const char * GetServerType(){return mysql_get_server_info(myDbHandel);};
+  MYSQL* GetDbHandel(){return myDbHandel;};
+  MYSQL_STMT *STDCALL MysqlSimplePrepare(const char *query);
+  int Select_DB();
+  int Do_Query(char * stm);
+  const char * GetError();
+  int GetErrorNumber();
+  unsigned long SelectCountTable(const char * table);
+
+private:
+
+  //Connect variables
+  char * databaseName; //hold results file name
+  char host[256];                   // Computer to connect to
+  char user[256];                   // MySQL User
+  char pass[256];                   // MySQL User Password
+  char dbs[256];                    // Database to use (TPCB)
+  unsigned int port;               // MySQL Server port
+  char socket[256];             // MySQL Server Unix Socket
+  MYSQL  *myDbHandel;
+
+  void DatabaseLogout();
+
+  void SetDbName(const char * name){strcpy((char *)dbs, name);};
+  void SetUser(const char * userName){strcpy((char *)user, userName);};
+  void SetPassword(const char * password){strcpy((char *)pass,password);};
+  void SetHost(const char * system){strcpy((char*)host, system);};
+  void SetPort(unsigned int portIn){port=portIn;};
+  void SetSocket(const char * sockIn){strcpy((char *)socket, sockIn);};
+  void PrintError(const char *msg);
+  void PrintStError(MYSQL_STMT *stmt, const char *msg);
+  void Die(const char *file, int line, const char *expr); // stop program
+  
+};
+#endif
+

storage/ndb/test/src/Makefile.am

@@ -24,7 +24,7 @@ libNDBT_a_SOURCES = \
 	NdbRestarter.cpp NdbRestarts.cpp NDBT_Output.cpp \
 	NdbBackup.cpp  NdbConfig.cpp NdbGrep.cpp NDBT_Table.cpp \
 	NdbSchemaCon.cpp NdbSchemaOp.cpp getarg.c \
-	CpcClient.cpp NdbMixRestarter.cpp NDBT_Thread.cpp
+	CpcClient.cpp NdbMixRestarter.cpp NDBT_Thread.cpp dbutil.cpp
 
 INCLUDES_LOC = -I$(top_srcdir)/storage/ndb/src/common/mgmcommon -I$(top_srcdir)/storage/ndb/include/mgmcommon -I$(top_srcdir)/storage/ndb/include/kernel -I$(top_srcdir)/storage/ndb/src/mgmapi
 

storage/ndb/test/src/dbutil.cpp

@@ -0,0 +1,176 @@
+// dbutil.cpp: implementation of the database utilities class.
+//
+//////////////////////////////////////////////////////////////////////
+
+#include "dbutil.hpp"
+
+//////////////////////////////////////////////////////////////////////
+// Construction/Destruction
+//////////////////////////////////////////////////////////////////////
+dbutil::dbutil(const char * dbname)
+{
+  memset(host,' ',sizeof(host));
+  memset(user,' ',sizeof(pass));
+  memset(dbs,' ',sizeof(dbs));
+  port = 0;
+  memset(socket,' ',sizeof(socket));
+  this->SetDbName(dbname);
+}
+
+dbutil::~dbutil()
+{
+  this->DatabaseLogout();
+}
+
+//////////////////////////////////////////////////////////////////////
+// Database Login
+//////////////////////////////////////////////////////////////////////
+void dbutil::DatabaseLogin(const char* system,
+                           const char* usr,
+                           const char* password,
+                           unsigned int portIn,
+                           const char* sockIn,
+                           bool transactional
+                           ){
+  if (!(myDbHandel = mysql_init(NULL))){
+    myerror("mysql_init() failed");
+    exit(1);
+  }
+  this->SetUser(usr);
+  this->SetHost(system);
+  this->SetPassword(password);
+  this->SetPort(portIn);
+  this->SetSocket(sockIn);
+
+  if (!(mysql_real_connect(myDbHandel, host, user, pass, "test", port, socket, 0))){
+    myerror("connection failed");
+    mysql_close(myDbHandel);
+    fprintf(stdout, "\n Check the connection options using --help or -?\n");
+    exit(1);
+  }
+
+  myDbHandel->reconnect= 1;
+
+  /* set AUTOCOMMIT */
+  if(!transactional){
+    mysql_autocommit(myDbHandel, TRUE);
+  }
+  else{
+    mysql_autocommit(myDbHandel, FALSE);
+  }
+
+  fprintf(stdout, "\n\tConnected to MySQL server version: %s (%lu)\n\n", 
+          mysql_get_server_info(myDbHandel),
+    (unsigned long) mysql_get_server_version(myDbHandel));
+}
+
+//////////////////////////////////////////////////////////////////////
+// Database Logout
+//////////////////////////////////////////////////////////////////////
+void dbutil::DatabaseLogout(){
+  if (myDbHandel){
+    fprintf(stdout, "\n\tClosing the MySQL database connection ...\n\n");
+    mysql_close(myDbHandel);
+  }
+}
+
+//////////////////////////////////////////////////////////////////////
+// Prepare MySQL Statements Cont
+//////////////////////////////////////////////////////////////////////
+MYSQL_STMT *STDCALL dbutil::MysqlSimplePrepare(const char *query){
+#ifdef DEBUG
+printf("Inside dbutil::MysqlSimplePrepare\n");
+#endif
+int result = 0;
+  MYSQL_STMT *my_stmt= mysql_stmt_init(this->GetDbHandel());
+  if (my_stmt && (result = mysql_stmt_prepare(my_stmt, query, strlen(query)))){
+    printf("res = %s\n",mysql_stmt_error(my_stmt));
+    mysql_stmt_close(my_stmt);
+    return 0;
+  }
+  return my_stmt;
+}
+//////////////////////////////////////////////////////////////////////
+// Error Printing
+//////////////////////////////////////////////////////////////////////
+void dbutil::PrintError(const char *msg){
+  if (this->GetDbHandel()
+      && mysql_errno(this->GetDbHandel())){
+      if (this->GetDbHandel()->server_version){
+        fprintf(stdout, "\n [MySQL-%s]",
+                this->GetDbHandel()->server_version);
+      }
+      else
+        fprintf(stdout, "\n [MySQL]");
+      fprintf(stdout, "[%d] %s\n",
+              mysql_errno(this->GetDbHandel()),
+              mysql_error(this->GetDbHandel()));
+  }
+  else if (msg)
+    fprintf(stderr, " [MySQL] %s\n", msg);
+}
+
+void dbutil::PrintStError(MYSQL_STMT *stmt, const char *msg)
+{
+  if (stmt && mysql_stmt_errno(stmt))
+  {
+    if (this->GetDbHandel()
+        && this->GetDbHandel()->server_version)
+      fprintf(stdout, "\n [MySQL-%s]",
+              this->GetDbHandel()->server_version);
+    else
+      fprintf(stdout, "\n [MySQL]");
+
+    fprintf(stdout, "[%d] %s\n", mysql_stmt_errno(stmt),
+    mysql_stmt_error(stmt));
+  }
+  else if (msg)
+   fprintf(stderr, " [MySQL] %s\n", msg);
+}
+/////////////////////////////////////////////////////
+int dbutil::Select_DB()
+{
+  return mysql_select_db(this->GetDbHandel(),
+                         this->GetDbName());
+}
+////////////////////////////////////////////////////
+int dbutil::Do_Query(char * stm)
+{
+  return mysql_query(this->GetDbHandel(), stm);
+}
+////////////////////////////////////////////////////
+const char * dbutil::GetError()
+{
+  return mysql_error(this->GetDbHandel());
+}
+////////////////////////////////////////////////////
+int dbutil::GetErrorNumber()
+{
+  return mysql_errno(this->GetDbHandel());
+}
+////////////////////////////////////////////////////
+unsigned long dbutil::SelectCountTable(const char * table)
+{
+	unsigned long count = 0;
+    MYSQL_RES *result;
+    char query[1024];
+    MYSQL_ROW row;
+        
+    sprintf(query,"select count(*) from `%s`", table);
+    if (mysql_query(this->GetDbHandel(),query) || !(result=mysql_store_result(this->GetDbHandel())))
+    {
+      printf("error\n");
+      return 1;
+    }
+    row= mysql_fetch_row(result);
+    count= (ulong) strtoull(row[0], (char**) 0, 10);
+    mysql_free_result(result);
+    
+    return count;
+}
+void dbutil::Die(const char *file, int line, const char *expr){
+  fprintf(stderr, "%s:%d: check failed: '%s'\n", file, line, expr);
+  abort();
+}
+
+

storage/ndb/tools/restore/Restore.cpp

@@ -534,6 +534,88 @@ TupleS::prepareRecord(TableS & tab){
   return true;
 }
 
+int
+RestoreDataIterator::readTupleData(Uint32 *buf_ptr, Uint32 *ptr,
+                                   Uint32 dataLength)
+{
+  while (ptr + 2 < buf_ptr + dataLength)
+  {
+    typedef BackupFormat::DataFile::VariableData VarData;
+    VarData * data = (VarData *)ptr;
+    Uint32 sz = ntohl(data->Sz);
+    Uint32 attrId = ntohl(data->Id); // column_no
+
+    AttributeData * attr_data = m_tuple.getData(attrId);
+    const AttributeDesc * attr_desc = m_tuple.getDesc(attrId);
+    
+    // just a reminder - remove when backwards compat implemented
+    if (m_currentTable->backupVersion < MAKE_VERSION(5,1,3) && 
+        attr_desc->m_column->getNullable())
+    {
+      const Uint32 ind = attr_desc->m_nullBitIndex;
+      if(BitmaskImpl::get(m_currentTable->m_nullBitmaskSize, 
+                          buf_ptr,ind))
+      {
+        attr_data->null = true;
+        attr_data->void_value = NULL;
+        continue;
+      }
+    }
+
+    if (m_currentTable->backupVersion < MAKE_VERSION(5,1,3))
+    {
+      sz *= 4;
+    }
+    
+    attr_data->null = false;
+    attr_data->void_value = &data->Data[0];
+    attr_data->size = sz;
+
+    //if (m_currentTable->getTableId() >= 2) { ndbout << "var off=" << ptr-buf_ptr << " attrId=" << attrId << endl; }
+
+    /**
+     * Compute array size
+     */
+    const Uint32 arraySize = sz / (attr_desc->size / 8);
+    assert(arraySize <= attr_desc->arraySize);
+
+    //convert the length of blob(v1) and text(v1)
+    if(!m_hostByteOrder
+        && (attr_desc->m_column->getType() == NdbDictionary::Column::Blob
+           || attr_desc->m_column->getType() == NdbDictionary::Column::Text)
+        && attr_desc->m_column->getArrayType() == NdbDictionary::Column::ArrayTypeFixed)
+    {
+      char* p = (char*)&attr_data->u_int64_value[0];
+      Uint64 x;
+      memcpy(&x, p, sizeof(Uint64));
+      x = Twiddle64(x);
+      memcpy(p, &x, sizeof(Uint64));
+    }
+
+    //convert datetime type
+    if(!m_hostByteOrder
+        && attr_desc->m_column->getType() == NdbDictionary::Column::Datetime)
+    {
+      char* p = (char*)&attr_data->u_int64_value[0];
+      Uint64 x;
+      memcpy(&x, p, sizeof(Uint64));
+      x = Twiddle64(x);
+      memcpy(p, &x, sizeof(Uint64));
+    }
+
+    if(!Twiddle(attr_desc, attr_data, attr_desc->arraySize))
+    {
+      return -1;
+    }
+    
+    ptr += ((sz + 3) >> 2) + 2;
+  }
+
+  assert(ptr == buf_ptr + dataLength);
+
+  return 0;
+}
+
 const TupleS *
 RestoreDataIterator::getNextTuple(int  & res)
 {
@@ -630,78 +712,8 @@ RestoreDataIterator::getNextTuple(int  & res)
     attr_data->void_value = NULL;
   }
 
-  while (ptr + 2 < buf_ptr + dataLength) {
+  if ((res = readTupleData(buf_ptr, ptr, dataLength)))
-    typedef BackupFormat::DataFile::VariableData VarData;
+    return NULL;
-    VarData * data = (VarData *)ptr;
-    Uint32 sz = ntohl(data->Sz);
-    Uint32 attrId = ntohl(data->Id); // column_no
-
-    AttributeData * attr_data = m_tuple.getData(attrId);
-    const AttributeDesc * attr_desc = m_tuple.getDesc(attrId);
-    
-    // just a reminder - remove when backwards compat implemented
-    if(m_currentTable->backupVersion < MAKE_VERSION(5,1,3) && 
-       attr_desc->m_column->getNullable()){
-      const Uint32 ind = attr_desc->m_nullBitIndex;
-      if(BitmaskImpl::get(m_currentTable->m_nullBitmaskSize, 
-			  buf_ptr,ind)){
-	attr_data->null = true;
-	attr_data->void_value = NULL;
-	continue;
-      }
-    }
-
-    if (m_currentTable->backupVersion < MAKE_VERSION(5,1,3))
-    {
-      sz *= 4;
-    }
-    
-    attr_data->null = false;
-    attr_data->void_value = &data->Data[0];
-    attr_data->size = sz;
-
-    //if (m_currentTable->getTableId() >= 2) { ndbout << "var off=" << ptr-buf_ptr << " attrId=" << attrId << endl; }
-
-    /**
-     * Compute array size
-     */
-    const Uint32 arraySize = sz / (attr_desc->size / 8);
-    assert(arraySize <= attr_desc->arraySize);
-
-    //convert the length of blob(v1) and text(v1)
-    if(!m_hostByteOrder
-        && (attr_desc->m_column->getType() == NdbDictionary::Column::Blob
-           || attr_desc->m_column->getType() == NdbDictionary::Column::Text)
-        && attr_desc->m_column->getArrayType() == NdbDictionary::Column::ArrayTypeFixed)
-    {
-      char* p = (char*)&attr_data->u_int64_value[0];
-      Uint64 x;
-      memcpy(&x, p, sizeof(Uint64));
-      x = Twiddle64(x);
-      memcpy(p, &x, sizeof(Uint64));
-    }
-
-    //convert datetime type
-    if(!m_hostByteOrder
-        && attr_desc->m_column->getType() == NdbDictionary::Column::Datetime)
-    {
-      char* p = (char*)&attr_data->u_int64_value[0];
-      Uint64 x;
-      memcpy(&x, p, sizeof(Uint64));
-      x = Twiddle64(x);
-      memcpy(p, &x, sizeof(Uint64));
-    }
-
-    if(!Twiddle(attr_desc, attr_data, attr_desc->arraySize))
-      {
-	res = -1;
-	return NULL;
-      }
-    
-    ptr += ((sz + 3) >> 2) + 2;
-  }
-
-  assert(ptr == buf_ptr + dataLength);
 
   m_count ++;  
   res = 0;

storage/ndb/tools/restore/Restore.hpp

@@ -355,6 +355,10 @@ public:
   bool validateFragmentFooter();
 
   const TupleS *getNextTuple(int & res);
+
+private:
+
+  int readTupleData(Uint32 *buf_ptr, Uint32 *ptr, Uint32 dataLength);
 };
 
 class LogEntry {