Bug #19914 SELECT COUNT(*) sometimes returns MAX_INT on cluster tables

fixes for ndb_* tests broken by previous fix

be more careful in ndb about setting errors on failure of info call (especially
in open) 


sql/ha_ndbcluster.cc:
  fix some ndb* tests failing due to fix for 19914
  
  be more careful about setting errors on failure of info call
sql/ha_ndbcluster.h:
  fix some ndb* tests failing due to fix for 19914
  
  be more careful about setting errors on failure of info call

sql/ha_ndbcluster.cc

@@ -85,7 +85,7 @@ static int packfrm(const void *data, uint len, const void **pack_data, uint *pac
 static int unpackfrm(const void **data, uint *len,
 		     const void* pack_data);
 
-static int ndb_get_table_statistics(Ndb*, const char *, 
+static int ndb_get_table_statistics(ha_ndbcluster*, bool, Ndb*, const char *,
 				    Uint64* rows, Uint64* commits);
 
 
@@ -275,7 +275,9 @@ int ha_ndbcluster::records_update()
     Ndb *ndb= get_ndb();
     Uint64 rows;
     ndb->setDatabaseName(m_dbname);
-    if((result= ndb_get_table_statistics(ndb, m_tabname, &rows, 0)) == 0){
+    result= ndb_get_table_statistics(this, true, ndb, m_tabname, &rows, 0);
+    if(result == 0)
+    {
       info->records= rows;
     }
   }
@@ -284,7 +286,8 @@ int ha_ndbcluster::records_update()
     if (((Thd_ndb*)(thd->transaction.thd_ndb))->error)
       info->no_uncommitted_rows_count= 0;
   }
-  records= info->records+ info->no_uncommitted_rows_count;
+  if(result==0)
+    records= info->records+ info->no_uncommitted_rows_count;
   DBUG_RETURN(result);
 }
 
@@ -2911,7 +2914,7 @@ int ha_ndbcluster::info(uint flag)
       Uint64 rows= 100;
       ndb->setDatabaseName(m_dbname);
       if (current_thd->variables.ndb_use_exact_count)
-	result= ndb_get_table_statistics(ndb, m_tabname, &rows, 0);
+	result= ndb_get_table_statistics(this, true, ndb, m_tabname, &rows, 0);
       records= rows;
     }
   }
@@ -4394,7 +4397,15 @@ int ha_ndbcluster::open(const char *name, int mode, uint test_if_locked)
   
   res= get_metadata(name);
   if (!res)
-    info(HA_STATUS_VARIABLE | HA_STATUS_CONST);
+  {
+    Ndb *ndb= get_ndb();
+    ndb->setDatabaseName(m_dbname);
+    Uint64 rows= 0;
+    res= ndb_get_table_statistics(NULL, false, ndb, m_tabname, &rows, 0);
+    records= rows;
+    if(!res)
+      res= info(HA_STATUS_CONST);
+  }
 
   DBUG_RETURN(res);
 }
@@ -5261,13 +5272,15 @@ static int unpackfrm(const void **unpack_data, uint *unpack_len,
 
 static 
 int
-ndb_get_table_statistics(Ndb* ndb, const char * table, 
+ndb_get_table_statistics(ha_ndbcluster* file, bool report_error, Ndb* ndb,
+                         const char * table,
 			 Uint64* row_count, Uint64* commit_count)
 {
   DBUG_ENTER("ndb_get_table_statistics");
   DBUG_PRINT("enter", ("table: %s", table));
   NdbConnection* pTrans;
   NdbError error;
+  int reterr= 0;
   int retries= 10;
   int retry_sleep= 30 * 1000; /* 30 milliseconds */
 
@@ -5336,6 +5349,19 @@ ndb_get_table_statistics(Ndb* ndb, const char * table,
     DBUG_RETURN(0);
 
 retry:
+    if(report_error)
+    {
+      if (file)
+      {
+        reterr= file->ndb_err(pTrans);
+      }
+      else
+      {
+        const NdbError& tmp= error;
+        ERR_PRINT(tmp);
+        reterr= ndb_to_mysql_error(&tmp);
+      }
+    }
     if (pTrans)
     {
       ndb->closeTransaction(pTrans);
@@ -5348,8 +5374,9 @@ retry:
     }
     break;
   } while(1);
-  DBUG_PRINT("exit", ("failed, error %u(%s)", error.code, error.message));
+  DBUG_PRINT("exit", ("failed, reterr: %u, NdbError %u(%s)", reterr,
-  ERR_RETURN(error);
+                      error.code, error.message));
+  DBUG_RETURN(reterr);
 }
 
 /*

sql/ha_ndbcluster.h

@@ -153,7 +153,12 @@ class ha_ndbcluster: public handler
  
   static void set_dbname(const char *pathname, char *dbname);
   static void set_tabname(const char *pathname, char *tabname);
-   
+
+  /*
+   * Internal to ha_ndbcluster, used by C functions
+   */
+  int ndb_err(NdbConnection*);
+
  private:
   int alter_table_name(const char *to);
   int drop_table();
@@ -206,7 +211,6 @@ class ha_ndbcluster: public handler
 
   longlong get_auto_increment();
   void invalidate_dictionary_cache(bool global);
-  int ndb_err(NdbConnection*);
   bool uses_blob_value(bool all_fields);
 
   int write_ndb_file();