added test to trigger drifferent fragmentations in ndb

corrected documentation on fragmentation
    set "fragmentation medium" to mean 2 fragments per node instead of 1
    set default fragmentation to small instead of medium
    bug#8284 adjust fragmentation to max_rows


mysql-test/r/ndb_basic.result:
  added test to trigger drifferent fragmentations in ndb
mysql-test/t/ndb_basic.test:
  added test to trigger drifferent fragmentations in ndb
ndb/include/ndbapi/NdbDictionary.hpp:
  corrected documentation on fragmentation
ndb/src/kernel/blocks/dbdih/DbdihMain.cpp:
  set "fragmentation medium" to mean 2 fragments per node instead of 1
ndb/src/ndbapi/NdbDictionaryImpl.cpp:
  set default fragmentation to small instead of medium
sql/ha_ndbcluster.cc:
  bug#8284 adjust fragmentation to max_rows

mysql-test/r/ndb_basic.result

@@ -573,3 +573,37 @@ select * from t1 where a12345678901234567890123456789a1234567890=2;
 a1234567890123456789012345678901234567890	a12345678901234567890123456789a1234567890
 5	2
 drop table t1;
+create table t1
+(a bigint, b bigint, c bigint, d bigint, 
+primary key (a,b,c,d)) 
+engine=ndb
+max_rows=200000000;
+Warnings:
+Warning	1105	Ndb might have problems storing the max amount of rows specified
+insert into t1 values
+(1,2,3,4),(2,3,4,5),(3,4,5,6),
+(3,2,3,4),(1,3,4,5),(2,4,5,6),
+(1,2,3,5),(2,3,4,8),(3,4,5,9),
+(3,2,3,5),(1,3,4,8),(2,4,5,9),
+(1,2,3,6),(2,3,4,6),(3,4,5,7),
+(3,2,3,6),(1,3,4,6),(2,4,5,7),
+(1,2,3,7),(2,3,4,7),(3,4,5,8),
+(3,2,3,7),(1,3,4,7),(2,4,5,8),
+(1,3,3,4),(2,4,4,5),(3,5,5,6),
+(3,3,3,4),(1,4,4,5),(2,5,5,6),
+(1,3,3,5),(2,4,4,8),(3,5,5,9),
+(3,3,3,5),(1,4,4,8),(2,5,5,9),
+(1,3,3,6),(2,4,4,6),(3,5,5,7),
+(3,3,3,6),(1,4,4,6),(2,5,5,7),
+(1,3,3,7),(2,4,4,7),(3,5,5,8),
+(3,3,3,7),(1,4,4,7),(2,5,5,8);
+select count(*) from t1;
+count(*)
+48
+drop table t1;
+create table t1
+(a bigint, b bigint, c bigint, d bigint, 
+primary key (a)) 
+engine=ndb
+max_rows=1;
+drop table t1;

mysql-test/t/ndb_basic.test

@@ -539,3 +539,41 @@ insert into t1 values (1,1),(2,1),(3,1),(4,1),(5,2),(6,1),(7,1);
 explain select * from t1 where a12345678901234567890123456789a1234567890=2;
 select * from t1 where a12345678901234567890123456789a1234567890=2;
 drop table t1;
+
+#
+# test fragment creation
+#
+# first a table with _many_ fragments per node group
+# then a table with just one fragment per node group
+#
+create table t1
+  (a bigint, b bigint, c bigint, d bigint, 
+   primary key (a,b,c,d)) 
+  engine=ndb
+  max_rows=200000000;
+insert into t1 values
+  (1,2,3,4),(2,3,4,5),(3,4,5,6),
+  (3,2,3,4),(1,3,4,5),(2,4,5,6),
+  (1,2,3,5),(2,3,4,8),(3,4,5,9),
+  (3,2,3,5),(1,3,4,8),(2,4,5,9),
+  (1,2,3,6),(2,3,4,6),(3,4,5,7),
+  (3,2,3,6),(1,3,4,6),(2,4,5,7),
+  (1,2,3,7),(2,3,4,7),(3,4,5,8),
+  (3,2,3,7),(1,3,4,7),(2,4,5,8),
+  (1,3,3,4),(2,4,4,5),(3,5,5,6),
+  (3,3,3,4),(1,4,4,5),(2,5,5,6),
+  (1,3,3,5),(2,4,4,8),(3,5,5,9),
+  (3,3,3,5),(1,4,4,8),(2,5,5,9),
+  (1,3,3,6),(2,4,4,6),(3,5,5,7),
+  (3,3,3,6),(1,4,4,6),(2,5,5,7),
+  (1,3,3,7),(2,4,4,7),(3,5,5,8),
+  (3,3,3,7),(1,4,4,7),(2,5,5,8);
+select count(*) from t1;
+drop table t1;
+
+create table t1
+  (a bigint, b bigint, c bigint, d bigint, 
+   primary key (a)) 
+  engine=ndb
+  max_rows=1;
+drop table t1;

ndb/include/ndbapi/NdbDictionary.hpp

@@ -141,9 +141,9 @@ public:
     enum FragmentType { 
       FragUndefined = 0,      ///< Fragmentation type undefined or default
       FragSingle = 1,         ///< Only one fragment
-      FragAllSmall = 2,       ///< One fragment per node group
+      FragAllSmall = 2,       ///< One fragment per node, default
-      FragAllMedium = 3,      ///< Default value. Two fragments per node group.
+      FragAllMedium = 3,      ///< two fragments per node
-      FragAllLarge = 4        ///< Eight fragments per node group.
+      FragAllLarge = 4        ///< Four fragments per node.
     };
   };
 

ndb/src/kernel/blocks/dbdih/DbdihMain.cpp

@@ -6178,7 +6178,7 @@ void Dbdih::execCREATE_FRAGMENTATION_REQ(Signal * signal){
       break;
     case DictTabInfo::AllNodesMediumTable:
       jam();
-      noOfFragments = csystemnodes;
+      noOfFragments = 2 * csystemnodes;
       break;
     case DictTabInfo::AllNodesLargeTable:
       jam();

ndb/src/ndbapi/NdbDictionaryImpl.cpp

@@ -284,7 +284,7 @@ void
 NdbTableImpl::init(){
   clearNewProperties();
   m_frm.clear();
-  m_fragmentType = NdbDictionary::Object::FragAllMedium;
+  m_fragmentType = NdbDictionary::Object::FragAllSmall;
   m_logging = true;
   m_kvalue = 6;
   m_minLoadFactor = 78;

sql/ha_ndbcluster.cc

@@ -3503,6 +3503,47 @@ static int create_ndb_column(NDBCOL &col,
   Create a table in NDB Cluster
  */
 
+static void ndb_set_fragmentation(NDBTAB &tab, TABLE *form, uint pk_length)
+{
+  if (form->max_rows == 0) /* default setting, don't set fragmentation */
+    return;
+  /**
+   * get the number of fragments right
+   */
+  uint no_fragments;
+  {
+#if MYSQL_VERSION_ID >= 50000
+    uint acc_row_size= 25+2;
+#else
+    uint acc_row_size= pk_length*4;
+    /* add acc overhead */
+    if (pk_length <= 8)
+      acc_row_size+= 25+2;  /* main page will set the limit */
+    else
+      acc_row_size+= 4+4;   /* overflow page will set the limit */
+#endif
+    ulonglong acc_fragment_size= 512*1024*1024;
+    ulonglong max_rows= form->max_rows;
+    no_fragments= (max_rows*acc_row_size)/acc_fragment_size+1;
+  }
+  {
+    uint no_nodes= g_ndb_cluster_connection->no_db_nodes();
+    NDBTAB::FragmentType ftype;
+    if (no_fragments > 2*no_nodes)
+    {
+      ftype= NDBTAB::FragAllLarge;
+      if (no_fragments > 4*no_nodes)
+	push_warning(current_thd, MYSQL_ERROR::WARN_LEVEL_WARN, ER_UNKNOWN_ERROR,
+		     "Ndb might have problems storing the max amount of rows specified");
+    }
+    else if (no_fragments > no_nodes)
+      ftype= NDBTAB::FragAllMedium;
+    else
+      ftype= NDBTAB::FragAllSmall;
+    tab.setFragmentType(ftype);
+  }
+}
+
 int ha_ndbcluster::create(const char *name, 
 			  TABLE *form, 
 			  HA_CREATE_INFO *info)
@@ -3605,7 +3646,9 @@ int ha_ndbcluster::create(const char *name,
       break;
     }
   }
-  
+
+  ndb_set_fragmentation(tab, form, pk_length);
+
   if ((my_errno= check_ndb_connection()))
     DBUG_RETURN(my_errno);