Commit 929cfa7b authored by unknown's avatar unknown

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

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

parents c2a48474 c9ebf5df
...@@ -922,6 +922,37 @@ public: ...@@ -922,6 +922,37 @@ public:
void setTemporary(bool); void setTemporary(bool);
#endif #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: private:
#ifndef DOXYGEN_SHOULD_SKIP_INTERNAL #ifndef DOXYGEN_SHOULD_SKIP_INTERNAL
friend class Ndb; friend class Ndb;
......
...@@ -204,12 +204,12 @@ NdbDictionary::Column::getPrimaryKey() const { ...@@ -204,12 +204,12 @@ NdbDictionary::Column::getPrimaryKey() const {
void void
NdbDictionary::Column::setPartitionKey(bool val){ NdbDictionary::Column::setPartitionKey(bool val){
m_impl.m_distributionKey = (val ? 2 : 0); m_impl.m_distributionKey = true;
} }
bool bool
NdbDictionary::Column::getPartitionKey() const{ NdbDictionary::Column::getPartitionKey() const{
return (bool)m_impl.m_distributionKey; return m_impl.m_distributionKey;
} }
const NdbDictionary::Table * const NdbDictionary::Table *
...@@ -353,7 +353,6 @@ NdbDictionary::Table::addColumn(const Column & c){ ...@@ -353,7 +353,6 @@ NdbDictionary::Table::addColumn(const Column & c){
NdbColumnImpl* col = new NdbColumnImpl; NdbColumnImpl* col = new NdbColumnImpl;
(* col) = NdbColumnImpl::getImpl(c); (* col) = NdbColumnImpl::getImpl(c);
m_impl.m_columns.push_back(col); m_impl.m_columns.push_back(col);
m_impl.computeAggregates();
m_impl.buildColumnHash(); m_impl.buildColumnHash();
} }
...@@ -699,6 +698,18 @@ NdbDictionary::Table::getRowGCIIndicator() const { ...@@ -699,6 +698,18 @@ NdbDictionary::Table::getRowGCIIndicator() const {
return m_impl.m_row_gci; 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 * Index facade
......
...@@ -295,7 +295,7 @@ NdbColumnImpl::equal(const NdbColumnImpl& col) const ...@@ -295,7 +295,7 @@ NdbColumnImpl::equal(const NdbColumnImpl& col) const
DBUG_RETURN(false); DBUG_RETURN(false);
} }
if (m_pk) { if (m_pk) {
if ((bool)m_distributionKey != (bool)col.m_distributionKey) { if (m_distributionKey != col.m_distributionKey) {
DBUG_RETURN(false); DBUG_RETURN(false);
} }
} }
...@@ -780,8 +780,8 @@ NdbTableImpl::computeAggregates() ...@@ -780,8 +780,8 @@ NdbTableImpl::computeAggregates()
m_noOfKeys++; m_noOfKeys++;
m_keyLenInWords += (col->m_attrSize * col->m_arraySize + 3) / 4; m_keyLenInWords += (col->m_attrSize * col->m_arraySize + 3) / 4;
} }
if (col->m_distributionKey == 2) // set by user if (col->m_distributionKey)
m_noOfDistributionKeys++; m_noOfDistributionKeys++; // XXX check PK
if (col->getBlobType()) if (col->getBlobType())
m_noOfBlobs++; m_noOfBlobs++;
...@@ -798,19 +798,7 @@ NdbTableImpl::computeAggregates() ...@@ -798,19 +798,7 @@ NdbTableImpl::computeAggregates()
for (i = 0, n = m_noOfKeys; n != 0; i++) { for (i = 0, n = m_noOfKeys; n != 0; i++) {
NdbColumnImpl* col = m_columns[i]; NdbColumnImpl* col = m_columns[i];
if (col->m_pk) { if (col->m_pk) {
col->m_distributionKey = true; // set by us col->m_distributionKey = true;
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;
n--; n--;
} }
} }
...@@ -826,6 +814,22 @@ NdbTableImpl::computeAggregates() ...@@ -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* const void*
NdbTableImpl::getTablespaceNames() const NdbTableImpl::getTablespaceNames() const
{ {
...@@ -2113,9 +2117,9 @@ NdbDictInterface::parseTableInfo(NdbTableImpl ** ret, ...@@ -2113,9 +2117,9 @@ NdbDictInterface::parseTableInfo(NdbTableImpl ** ret,
col->m_storageType = attrDesc.AttributeStorageType; col->m_storageType = attrDesc.AttributeStorageType;
col->m_pk = attrDesc.AttributeKeyFlag; 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_nullable = attrDesc.AttributeNullableFlag;
col->m_autoIncrement = (attrDesc.AttributeAutoIncrement ? true : false); col->m_autoIncrement = (attrDesc.AttributeAutoIncrement != 0);
col->m_autoIncrementInitialValue = ~0; col->m_autoIncrementInitialValue = ~0;
col->m_defaultValue.assign(attrDesc.AttributeDefaultValue); col->m_defaultValue.assign(attrDesc.AttributeDefaultValue);
...@@ -2606,7 +2610,7 @@ loop: ...@@ -2606,7 +2610,7 @@ loop:
tmpAttr.AttributeId = col->m_attrId; tmpAttr.AttributeId = col->m_attrId;
tmpAttr.AttributeKeyFlag = col->m_pk; tmpAttr.AttributeKeyFlag = col->m_pk;
tmpAttr.AttributeNullableFlag = col->m_nullable; 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.AttributeExtType = (Uint32)col->m_type;
tmpAttr.AttributeExtPrecision = ((unsigned)col->m_precision & 0xFFFF); tmpAttr.AttributeExtPrecision = ((unsigned)col->m_precision & 0xFFFF);
......
...@@ -85,11 +85,7 @@ public: ...@@ -85,11 +85,7 @@ public:
CHARSET_INFO * m_cs; // not const in MySQL CHARSET_INFO * m_cs; // not const in MySQL
bool m_pk; bool m_pk;
/* bool m_distributionKey;
* Since "none" is "all" we distinguish between
* 1-set by us, 2-set by user
*/
Uint32 m_distributionKey;
bool m_nullable; bool m_nullable;
bool m_autoIncrement; bool m_autoIncrement;
Uint64 m_autoIncrementInitialValue; Uint64 m_autoIncrementInitialValue;
...@@ -159,6 +155,9 @@ public: ...@@ -159,6 +155,9 @@ public:
const char * getMysqlName() const; const char * getMysqlName() const;
void updateMysqlName(); void updateMysqlName();
int aggregate(NdbError& error);
int validate(NdbError& error);
Uint32 m_changeMask; Uint32 m_changeMask;
Uint32 m_primaryTableId; Uint32 m_primaryTableId;
BaseString m_internalName; BaseString m_internalName;
......
...@@ -65,6 +65,11 @@ public: ...@@ -65,6 +65,11 @@ public:
//setStoredTable(stored); //setStoredTable(stored);
for(int i = 0; i<noOfAttributes; i++) for(int i = 0; i<noOfAttributes; i++)
addColumn(attributes[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, static const NdbDictionary::Table * discoverTableFromDb(Ndb* ndb,
......
...@@ -977,6 +977,11 @@ NDBT_Tables::createTable(Ndb* pNdb, const char* _name, bool _temp, ...@@ -977,6 +977,11 @@ NDBT_Tables::createTable(Ndb* pNdb, const char* _name, bool _temp,
do { do {
NdbDictionary::Table tmpTab(* tab); NdbDictionary::Table tmpTab(* tab);
tmpTab.setStoredTable(_temp ? 0 : 1); tmpTab.setStoredTable(_temp ? 0 : 1);
{
NdbError error;
int ret = tmpTab.validate(error);
assert(ret == 0);
}
if(f != 0 && f(pNdb, tmpTab, 0, arg)) if(f != 0 && f(pNdb, tmpTab, 0, arg))
{ {
ndbout << "Failed to create table" << endl; ndbout << "Failed to create table" << endl;
......
Markdown is supported
0%
or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment