Merge mronstrom@bk-internal.mysql.com:/home/bk/mysql-5.1

into  mysql.com:/Users/mikron/mysql-5.1

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

@@ -6220,6 +6220,17 @@ void Dbdih::execCREATE_FRAGMENTATION_REQ(Signal * signal)
           noOfFragments = 1;
           set_default_node_groups(signal, noOfFragments);
           break;
+        case DictTabInfo::DistrKeyHash:
+          jam();
+        case DictTabInfo::DistrKeyLin:
+          jam();
+          if (noOfFragments == 0)
+          {
+            jam();
+            noOfFragments = csystemnodes;
+            set_default_node_groups(signal, noOfFragments);
+          }
+          break;
         default:
           jam();
           if (noOfFragments == 0)

storage/ndb/src/ndbapi/NdbDictionaryImpl.cpp

@@ -505,19 +505,35 @@ NdbTableImpl::buildColumnHash(){
 Uint32
 NdbTableImpl::get_nodes(Uint32 hashValue, const Uint16 ** nodes) const
 {
-  if(m_replicaCount > 0)
+  Uint32 fragmentId;
+  if(m_replicaCount == 0)
+    return 0;
+  switch (m_fragmentType)
   {
-    Uint32 fragmentId = hashValue & m_hashValueMask;
-    if(fragmentId < m_hashpointerValue) 
+    case NdbDictionary::Object::FragAllSmall:
+    case NdbDictionary::Object::FragAllMedium:
+    case NdbDictionary::Object::FragAllLarge:
+    case NdbDictionary::Object::FragSingle:
+    case NdbDictionary::Object::DistrKeyLin:
     {
-      fragmentId = hashValue & ((m_hashValueMask << 1) + 1);
+      fragmentId = hashValue & m_hashValueMask;
+      if(fragmentId < m_hashpointerValue) 
+        fragmentId = hashValue & ((m_hashValueMask << 1) + 1);
+      break;
     }
-    Uint32 pos = fragmentId * m_replicaCount;
-    if(pos + m_replicaCount <= m_fragments.size())
+    case NdbDictionary::Object::DistrKeyHash:
     {
-      * nodes = m_fragments.getBase()+pos;
-      return m_replicaCount;
+      fragmentId = hashValue % m_fragmentCount;
+      break;
     }
+    default:
+      return 0;
+  }
+  Uint32 pos = fragmentId * m_replicaCount;
+  if (pos + m_replicaCount <= m_fragments.size())
+  {
+    *nodes = m_fragments.getBase()+pos;
+    return m_replicaCount;
   }
   return 0;
 }

storage/ndb/test/ndbapi/testPartitioning.cpp

@@ -94,6 +94,15 @@ add_distribution_key(Ndb*, NdbDictionary::Table& tab, int when)
       }
     }
   }
+
+  Uint32 linear_hash_ind = rand() & 1;
+  NdbDictionary::Table::FragmentType ftype;
+  if (linear_hash_ind)
+    ftype = NdbDictionary::Table::DistrKeyLin;
+  else
+    ftype = NdbDictionary::Table::DistrKeyHash;
+  tab.setFragmentType(ftype);
+
   ndbout << (NDBT_Table&)tab << endl;
 
   return 0;