Merge willster.(none):/home/stewart/Documents/MySQL/5.1/ndb

into  willster.(none):/home/stewart/Documents/MySQL/5.1/bug20809

storage/ndb/include/ndbapi/NdbDictionary.hpp

@@ -922,6 +922,37 @@ public:
     void setTemporary(bool); 
 #endif
 
+    // these 2 are not de-doxygenated
+
+    /**
+     * This method is not needed in normal usage.
+     *
+     * Compute aggregate data on table being defined.  Required for
+     * aggregate methods such as getNoOfPrimaryKeys() to work before
+     * table has been created and retrieved via getTable().
+     *
+     * May adjust some column flags.  If no PK is so far marked as
+     * distribution key then all PK's will be marked.
+     *
+     * Returns 0 on success.  Returns -1 and sets error if an
+     * inconsistency is detected.
+     */
+    int aggregate(struct NdbError& error);
+
+    /**
+     * This method is not needed in normal usage.
+     *
+     * Validate new table definition before create.  Does aggregate()
+     * and additional checks.  There may still be errors which are
+     * detected only by NDB kernel at create table.
+     *
+     * Create table and retrieve table do validate() automatically.
+     *
+     * Returns 0 on success.  Returns -1 and sets error if an
+     * inconsistency is detected.
+     */
+    int validate(struct NdbError& error);
+
   private:
 #ifndef DOXYGEN_SHOULD_SKIP_INTERNAL
     friend class Ndb;

storage/ndb/src/ndbapi/NdbDictionary.cpp

@@ -204,12 +204,12 @@ NdbDictionary::Column::getPrimaryKey() const {
 
 void 
 NdbDictionary::Column::setPartitionKey(bool val){
-  m_impl.m_distributionKey = (val ? 2 : 0);
+  m_impl.m_distributionKey = true;
 }
 
 bool 
 NdbDictionary::Column::getPartitionKey() const{
-  return (bool)m_impl.m_distributionKey;
+  return m_impl.m_distributionKey;
 }
 
 const NdbDictionary::Table * 
@@ -353,7 +353,6 @@ NdbDictionary::Table::addColumn(const Column & c){
   NdbColumnImpl* col = new NdbColumnImpl;
   (* col) = NdbColumnImpl::getImpl(c);
   m_impl.m_columns.push_back(col);
-  m_impl.computeAggregates();
   m_impl.buildColumnHash();
 }
 
@@ -699,6 +698,18 @@ NdbDictionary::Table::getRowGCIIndicator() const {
   return m_impl.m_row_gci;
 }
 
+int
+NdbDictionary::Table::aggregate(NdbError& error)
+{
+  return m_impl.aggregate(error);
+}
+
+int
+NdbDictionary::Table::validate(NdbError& error)
+{
+  return m_impl.validate(error);
+}
+
 
 /*****************************************************************
  * Index facade

storage/ndb/src/ndbapi/NdbDictionaryImpl.cpp

@@ -295,7 +295,7 @@ NdbColumnImpl::equal(const NdbColumnImpl& col) const
     DBUG_RETURN(false);
   }
   if (m_pk) {
-    if ((bool)m_distributionKey != (bool)col.m_distributionKey) {
+    if (m_distributionKey != col.m_distributionKey) {
       DBUG_RETURN(false);
     }
   }
@@ -780,8 +780,8 @@ NdbTableImpl::computeAggregates()
       m_noOfKeys++;
       m_keyLenInWords += (col->m_attrSize * col->m_arraySize + 3) / 4;
     }
-    if (col->m_distributionKey == 2)    // set by user
-      m_noOfDistributionKeys++;
+    if (col->m_distributionKey)
+      m_noOfDistributionKeys++; // XXX check PK
     
     if (col->getBlobType())
       m_noOfBlobs++;
@@ -798,19 +798,7 @@ NdbTableImpl::computeAggregates()
     for (i = 0, n = m_noOfKeys; n != 0; i++) {
       NdbColumnImpl* col = m_columns[i];
       if (col->m_pk) {
-        col->m_distributionKey = true;  // set by us
-        n--;
-      }
-    }
-  }
-  else 
-  {
-    for (i = 0, n = m_noOfKeys; n != 0; i++) {
-      NdbColumnImpl* col = m_columns[i];
-      if (col->m_pk)
-      {
-	if(col->m_distributionKey == 1)
-	  col->m_distributionKey = 0;  
+        col->m_distributionKey = true;
         n--;
       }
     }
@@ -826,6 +814,22 @@ NdbTableImpl::computeAggregates()
   }
 }
 
+// TODO add error checks
+// TODO use these internally at create and retrieve
+int
+NdbTableImpl::aggregate(NdbError& error)
+{
+  computeAggregates();
+  return 0;
+}
+int
+NdbTableImpl::validate(NdbError& error)
+{
+  if (aggregate(error) == -1)
+    return -1;
+  return 0;
+}
+
 const void*
 NdbTableImpl::getTablespaceNames() const
 {
@@ -2113,9 +2117,9 @@ NdbDictInterface::parseTableInfo(NdbTableImpl ** ret,
     col->m_storageType = attrDesc.AttributeStorageType;
     
     col->m_pk = attrDesc.AttributeKeyFlag;
-    col->m_distributionKey = attrDesc.AttributeDKey ? 2 : 0;
+    col->m_distributionKey = (attrDesc.AttributeDKey != 0);
     col->m_nullable = attrDesc.AttributeNullableFlag;
-    col->m_autoIncrement = (attrDesc.AttributeAutoIncrement ? true : false);
+    col->m_autoIncrement = (attrDesc.AttributeAutoIncrement != 0);
     col->m_autoIncrementInitialValue = ~0;
     col->m_defaultValue.assign(attrDesc.AttributeDefaultValue);
 
@@ -2606,7 +2610,7 @@ NdbDictInterface::createOrAlterTable(Ndb & ndb,
     tmpAttr.AttributeId = col->m_attrId;
     tmpAttr.AttributeKeyFlag = col->m_pk;
     tmpAttr.AttributeNullableFlag = col->m_nullable;
-    tmpAttr.AttributeDKey = distKeys ? (bool)col->m_distributionKey : 0;
+    tmpAttr.AttributeDKey = distKeys ? col->m_distributionKey : 0;
 
     tmpAttr.AttributeExtType = (Uint32)col->m_type;
     tmpAttr.AttributeExtPrecision = ((unsigned)col->m_precision & 0xFFFF);

storage/ndb/src/ndbapi/NdbDictionaryImpl.hpp

@@ -85,11 +85,7 @@ public:
   CHARSET_INFO * m_cs;          // not const in MySQL
   
   bool m_pk;
-  /*
-   * Since "none" is "all" we distinguish between
-   * 1-set by us, 2-set by user
-   */
-  Uint32 m_distributionKey;
+  bool m_distributionKey;
   bool m_nullable;
   bool m_autoIncrement;
   Uint64 m_autoIncrementInitialValue;
@@ -159,6 +155,9 @@ public:
   const char * getMysqlName() const;
   void updateMysqlName();
 
+  int aggregate(NdbError& error);
+  int validate(NdbError& error);
+
   Uint32 m_changeMask;
   Uint32 m_primaryTableId;
   BaseString m_internalName;

storage/ndb/test/include/NDBT_Table.hpp

@@ -65,6 +65,11 @@ public:
     //setStoredTable(stored);
     for(int i = 0; i<noOfAttributes; i++)
       addColumn(attributes[i]);
+
+    // validate() might cause initialization order problem with charset
+    NdbError error;
+    int ret = aggregate(error);
+    assert(ret == 0);
   }
 
   static const NdbDictionary::Table * discoverTableFromDb(Ndb* ndb,

storage/ndb/test/src/NDBT_Tables.cpp

@@ -977,6 +977,11 @@ NDBT_Tables::createTable(Ndb* pNdb, const char* _name, bool _temp,
   do {
     NdbDictionary::Table tmpTab(* tab);
     tmpTab.setStoredTable(_temp ? 0 : 1);
+    {
+      NdbError error;
+      int ret = tmpTab.validate(error);
+      assert(ret == 0);
+    }
     if(f != 0 && f(pNdb, tmpTab, 0, arg))
     {
       ndbout << "Failed to create table" << endl;