Merge tulin@bk-internal.mysql.com:/home/bk/mysql-4.1

into poseidon.(none):/home/tomas/mysql-4.1-ndb-merge

mysql-test/r/ndb_alter_table.result

@@ -18,12 +18,12 @@ col5 enum('PENDING', 'ACTIVE', 'DISABLED') not null,
 col6 int not null, to_be_deleted int)  ENGINE=ndbcluster;
 show table status;
 Name	Engine	Version	Row_format	Rows	Avg_row_length	Data_length	Max_data_length	Index_length	Data_free	Auto_increment	Create_time	Update_time	Check_time	Collation	Checksum	Create_options	Comment
-t1	ndbcluster	9	Dynamic	100	0	0	NULL	0	0	1	NULL	NULL	NULL	latin1_swedish_ci	NULL		
+t1	ndbcluster	9	Dynamic	0	0	0	NULL	0	0	1	NULL	NULL	NULL	latin1_swedish_ci	NULL		
 insert into t1 values
 (0,4,3,5,"PENDING",1,7),(NULL,4,3,5,"PENDING",1,7),(31,4,3,5,"PENDING",1,7), (7,4,3,5,"PENDING",1,7), (NULL,4,3,5,"PENDING",1,7), (100,4,3,5,"PENDING",1,7), (99,4,3,5,"PENDING",1,7), (8,4,3,5,"PENDING",1,7), (NULL,4,3,5,"PENDING",1,7);
 show table status;
 Name	Engine	Version	Row_format	Rows	Avg_row_length	Data_length	Max_data_length	Index_length	Data_free	Auto_increment	Create_time	Update_time	Check_time	Collation	Checksum	Create_options	Comment
-t1	ndbcluster	9	Dynamic	100	0	0	NULL	0	0	102	NULL	NULL	NULL	latin1_swedish_ci	NULL		
+t1	ndbcluster	9	Dynamic	9	0	0	NULL	0	0	102	NULL	NULL	NULL	latin1_swedish_ci	NULL		
 select * from t1 order by col1;
 col1	col2	col3	col4	col5	col6	to_be_deleted
 0	4	3	5	PENDING	1	7
@@ -43,7 +43,7 @@ change column col2 fourth varchar(30) not null after col3,
 modify column col6 int not null first;
 show table status;
 Name	Engine	Version	Row_format	Rows	Avg_row_length	Data_length	Max_data_length	Index_length	Data_free	Auto_increment	Create_time	Update_time	Check_time	Collation	Checksum	Create_options	Comment
-t1	ndbcluster	9	Dynamic	100	0	0	NULL	0	0	102	NULL	NULL	NULL	latin1_swedish_ci	NULL		
+t1	ndbcluster	9	Dynamic	9	0	0	NULL	0	0	102	NULL	NULL	NULL	latin1_swedish_ci	NULL		
 select * from t1 order by col1;
 col6	col1	col3	fourth	col4	col4_5	col5	col7	col8
 1	0	3	4	5		PENDING		0000-00-00 00:00:00
@@ -58,7 +58,7 @@ col6	col1	col3	fourth	col4	col4_5	col5	col7	col8
 insert into t1 values (2, NULL,4,3,5,99,"PENDING","EXTRA",'2004-01-01 00:00:00');
 show table status;
 Name	Engine	Version	Row_format	Rows	Avg_row_length	Data_length	Max_data_length	Index_length	Data_free	Auto_increment	Create_time	Update_time	Check_time	Collation	Checksum	Create_options	Comment
-t1	ndbcluster	9	Dynamic	100	0	0	NULL	0	0	103	NULL	NULL	NULL	latin1_swedish_ci	NULL		
+t1	ndbcluster	9	Dynamic	10	0	0	NULL	0	0	103	NULL	NULL	NULL	latin1_swedish_ci	NULL		
 select * from t1 order by col1;
 col6	col1	col3	fourth	col4	col4_5	col5	col7	col8
 1	0	3	4	5		PENDING		0000-00-00 00:00:00

mysql-test/r/ndb_blob.result

@@ -150,7 +150,7 @@ insert into t1 values(9,'b9',999,'dd9');
 commit;
 explain select * from t1;
 id	select_type	table	type	possible_keys	key	key_len	ref	rows	Extra
-1	SIMPLE	t1	ALL	NULL	NULL	NULL	NULL	100	
+1	SIMPLE	t1	ALL	NULL	NULL	NULL	NULL	9	
 select * from t1 order by a;
 a	b	c	d
 1	b1	111	dd1
@@ -185,7 +185,7 @@ insert into t1 values(2,@b2,222,@d2);
 commit;
 explain select * from t1;
 id	select_type	table	type	possible_keys	key	key_len	ref	rows	Extra
-1	SIMPLE	t1	ALL	NULL	NULL	NULL	NULL	100	
+1	SIMPLE	t1	ALL	NULL	NULL	NULL	NULL	2	
 select a,length(b),substr(b,1+2*900,2),length(d),substr(d,1+3*900,3)
 from t1 order by a;
 a	length(b)	substr(b,1+2*900,2)	length(d)	substr(d,1+3*900,3)

ndb/include/ndbapi/NdbDictionary.hpp

@@ -1066,6 +1066,9 @@ public:
     Dictionary(NdbDictionaryImpl&);
     const Table * getIndexTable(const char * indexName, 
 				const char * tableName);
+  public:
+    const Table * getTable(const char * name, void **data);
+    void set_local_table_data_size(unsigned sz);
   };
 };
 

ndb/src/ndbapi/DictCache.cpp

@@ -21,6 +21,31 @@
 #include <NdbCondition.h>
 #include <NdbSleep.h>
 
+Ndb_local_table_info *
+Ndb_local_table_info::create(NdbTableImpl *table_impl, Uint32 sz)
+{
+  void *data= malloc(sizeof(NdbTableImpl)+sz-1);
+  if (data == 0)
+    return 0;
+  memset(data,0,sizeof(NdbTableImpl)+sz-1);
+  new (data) Ndb_local_table_info(table_impl);
+  return (Ndb_local_table_info *) data;
+}
+
+void Ndb_local_table_info::destroy(Ndb_local_table_info *info)
+{
+  free((void *)info);
+}
+
+Ndb_local_table_info::Ndb_local_table_info(NdbTableImpl *table_impl)
+{
+  m_table_impl= table_impl;
+}
+
+Ndb_local_table_info::~Ndb_local_table_info()
+{
+}
+
 LocalDictCache::LocalDictCache(){
   m_tableHash.createHashTable();
 }
@@ -29,22 +54,24 @@ LocalDictCache::~LocalDictCache(){
   m_tableHash.releaseHashTable();
 }
 
-NdbTableImpl * 
+Ndb_local_table_info * 
 LocalDictCache::get(const char * name){
   const Uint32 len = strlen(name);
   return m_tableHash.getData(name, len);
 }
 
 void 
-LocalDictCache::put(const char * name, NdbTableImpl * tab){
-  const Uint32 id = tab->m_tableId;
+LocalDictCache::put(const char * name, Ndb_local_table_info * tab_info){
+  const Uint32 id = tab_info->m_table_impl->m_tableId;
   
-  m_tableHash.insertKey(name, strlen(name), id, tab);
+  m_tableHash.insertKey(name, strlen(name), id, tab_info);
 }
 
 void
 LocalDictCache::drop(const char * name){
-  m_tableHash.deleteKey(name, strlen(name));
+  Ndb_local_table_info *info= m_tableHash.deleteKey(name, strlen(name));
+  DBUG_ASSERT(info != 0);
+  Ndb_local_table_info::destroy(info);
 }
 
 /*****************************************************************

ndb/src/ndbapi/DictCache.hpp

@@ -27,6 +27,17 @@
 #include <Ndb.hpp>
 #include "NdbLinHash.hpp"
 
+class Ndb_local_table_info {
+public:
+  static Ndb_local_table_info *create(NdbTableImpl *table_impl, Uint32 sz=0);
+  static void destroy(Ndb_local_table_info *);
+  NdbTableImpl *m_table_impl;
+  char m_local_data[1];
+private:
+  Ndb_local_table_info(NdbTableImpl *table_impl);
+  ~Ndb_local_table_info();
+};
+
 /**
  * A non thread safe dict cache
  */
@@ -35,12 +46,12 @@ public:
   LocalDictCache();
   ~LocalDictCache();
   
-  NdbTableImpl * get(const char * name);
+  Ndb_local_table_info * get(const char * name);
   
-  void put(const char * name, NdbTableImpl *);
+  void put(const char * name, Ndb_local_table_info *);
   void drop(const char * name);
   
-  NdbLinHash<NdbTableImpl> m_tableHash; // On name
+  NdbLinHash<Ndb_local_table_info> m_tableHash; // On name
 };
 
 /**

ndb/src/ndbapi/Ndb.cpp

@@ -753,9 +753,11 @@ Uint64
 Ndb::getAutoIncrementValue(const char* aTableName, Uint32 cacheSize)
 {
   DEBUG_TRACE("getAutoIncrementValue");
-  const NdbTableImpl* table = theDictionary->getTable(aTableName);
-  if (table == 0)
+  const char * internalTableName = internalizeTableName(aTableName);
+  Ndb_local_table_info *info= theDictionary->get_local_table_info(internalTableName);
+  if (info == 0)
     return ~0;
+  const NdbTableImpl *table= info->m_table_impl;
   Uint64 tupleId = getTupleIdFromNdb(table->m_tableId, cacheSize);
   return tupleId;
 }
@@ -832,11 +834,13 @@ bool
 Ndb::setAutoIncrementValue(const char* aTableName, Uint64 val, bool increase)
 {
   DEBUG_TRACE("setAutoIncrementValue " << val);
-  const NdbTableImpl* table = theDictionary->getTable(aTableName);
-  if (table == 0) {
+  const char * internalTableName= internalizeTableName(aTableName);
+  Ndb_local_table_info *info= theDictionary->get_local_table_info(internalTableName);
+  if (info == 0) {
     theError= theDictionary->getNdbError();
     return false;
   }
+  const NdbTableImpl* table= info->m_table_impl;
   return setTupleIdInNdb(table->m_tableId, val, increase);
 }
 

ndb/src/ndbapi/NdbDictionary.cpp

@@ -681,13 +681,23 @@ NdbDictionary::Dictionary::alterTable(const Table & t){
 }
 
 const NdbDictionary::Table * 
-NdbDictionary::Dictionary::getTable(const char * name){
-  NdbTableImpl * t = m_impl.getTable(name);
+NdbDictionary::Dictionary::getTable(const char * name, void **data){
+  NdbTableImpl * t = m_impl.getTable(name, data);
   if(t)
     return t->m_facade;
   return 0;
 }
 
+void NdbDictionary::Dictionary::set_local_table_data_size(unsigned sz)
+{
+  m_impl.m_local_table_data_size= sz;
+}
+
+const NdbDictionary::Table * 
+NdbDictionary::Dictionary::getTable(const char * name){
+  return getTable(name, 0);
+}
+
 void
 NdbDictionary::Dictionary::invalidateTable(const char * name){
   NdbTableImpl * t = m_impl.getTable(name);

ndb/src/ndbapi/NdbDictionaryImpl.cpp

@@ -589,17 +589,19 @@ NdbDictionaryImpl::NdbDictionaryImpl(Ndb &ndb,
     m_ndb(ndb)
 {
   m_globalHash = 0;
+  m_local_table_data_size= 0;
 }
 
 static int f_dictionary_count = 0;
 
 NdbDictionaryImpl::~NdbDictionaryImpl()
 {
-  NdbElement_t<NdbTableImpl> * curr = m_localHash.m_tableHash.getNext(0);
+  NdbElement_t<Ndb_local_table_info> * curr = m_localHash.m_tableHash.getNext(0);
   if(m_globalHash){
     while(curr != 0){
       m_globalHash->lock();
-      m_globalHash->release(curr->theData);    
+      m_globalHash->release(curr->theData->m_table_impl);
+      Ndb_local_table_info::destroy(curr->theData);
       m_globalHash->unlock();
       
       curr = m_localHash.m_tableHash.getNext(curr);
@@ -620,7 +622,37 @@ NdbDictionaryImpl::~NdbDictionaryImpl()
   }
 }
 
+Ndb_local_table_info *
+NdbDictionaryImpl::fetchGlobalTableImpl(const char * internalTableName)
+{
+  NdbTableImpl *impl;
 
+  m_globalHash->lock();
+  impl = m_globalHash->get(internalTableName);
+  m_globalHash->unlock();
+
+  if (impl == 0){
+    impl = m_receiver.getTable(internalTableName, m_ndb.usingFullyQualifiedNames());
+    m_globalHash->lock();
+    m_globalHash->put(internalTableName, impl);
+    m_globalHash->unlock();
+    
+    if(impl == 0){
+      return 0;
+    }
+  }
+
+  Ndb_local_table_info *info= Ndb_local_table_info::create(impl, m_local_table_data_size);
+
+  m_localHash.put(internalTableName, info);
+
+  m_ndb.theFirstTupleId[impl->getTableId()] = ~0;
+  m_ndb.theLastTupleId[impl->getTableId()]  = ~0;
+  
+  addBlobTables(*impl);
+
+  return info;
+}
 
 #if 0
 bool
@@ -1504,7 +1536,6 @@ NdbDictInterface::createOrAlterTable(Ndb & ndb,
     : createTable(&tSignal, ptr);
 
   if (!alter && haveAutoIncrement) {
-    //    if (!ndb.setAutoIncrementValue(impl.m_internalName.c_str(), autoIncrementValue)) {
     if (!ndb.setAutoIncrementValue(impl.m_externalName.c_str(), autoIncrementValue)) {
       if (ndb.theError.code == 0) {
 	m_error.code = 4336;
@@ -1775,11 +1806,12 @@ NdbIndexImpl*
 NdbDictionaryImpl::getIndexImpl(const char * externalName, 
 				const char * internalName)
 {
-  NdbTableImpl* tab = getTableImpl(internalName);
-  if(tab == 0){
+  Ndb_local_table_info * info = get_local_table_info(internalName);
+  if(info == 0){
     m_error.code = 4243;
     return 0;
   }
+  NdbTableImpl * tab = info->m_table_impl;
 
   if(tab->m_indexType == NdbDictionary::Index::Undefined){
     // Not an index

ndb/src/ndbapi/NdbDictionaryImpl.hpp

@@ -390,8 +390,8 @@ public:
   int listObjects(List& list, NdbDictionary::Object::Type type);
   int listIndexes(List& list, const char * tableName);
   
-  NdbTableImpl * getTable(const char * tableName);
-  NdbTableImpl * getTableImpl(const char * internalName);
+  NdbTableImpl * getTable(const char * tableName, void **data= 0);
+  Ndb_local_table_info * get_local_table_info(const char * internalName);
   NdbIndexImpl * getIndex(const char * indexName,
 			  const char * tableName);
   NdbIndexImpl * getIndexImpl(const char * name, const char * internalName);
@@ -400,6 +400,7 @@ public:
   
   const NdbError & getNdbError() const;
   NdbError m_error;
+  Uint32 m_local_table_data_size;
 
   LocalDictCache m_localHash;
   GlobalDictCache * m_globalHash;
@@ -410,6 +411,8 @@ public:
 
   NdbDictInterface m_receiver;
   Ndb & m_ndb;
+private:
+  Ndb_local_table_info * fetchGlobalTableImpl(const char * internalName);
 };
 
 inline
@@ -598,45 +601,28 @@ NdbDictionaryImpl::getImpl(const NdbDictionary::Dictionary & t){
 
 inline
 NdbTableImpl * 
-NdbDictionaryImpl::getTable(const char * tableName)
+NdbDictionaryImpl::getTable(const char * tableName, void **data)
 {
   const char * internalTableName = m_ndb.internalizeTableName(tableName);
-
-  return getTableImpl(internalTableName);
+  Ndb_local_table_info *info= get_local_table_info(internalTableName);
+  if (info == 0) {
+    return 0;
+  }
+  if (data) {
+    *data= info->m_local_data;
+  }
+  return info->m_table_impl;
 }
 
 inline
-NdbTableImpl * 
-NdbDictionaryImpl::getTableImpl(const char * internalTableName)
+Ndb_local_table_info * 
+NdbDictionaryImpl::get_local_table_info(const char * internalTableName)
 {
-  NdbTableImpl *ret = m_localHash.get(internalTableName);
-
-  if (ret != 0) {
-    return ret; // autoincrement already initialized
+  Ndb_local_table_info *info= m_localHash.get(internalTableName);
+  if (info != 0) {
+    return info; // autoincrement already initialized
   }
-
-  m_globalHash->lock();
-  ret = m_globalHash->get(internalTableName);
-  m_globalHash->unlock();
-
-  if (ret == 0){
-    ret = m_receiver.getTable(internalTableName, m_ndb.usingFullyQualifiedNames());
-    m_globalHash->lock();
-    m_globalHash->put(internalTableName, ret);
-    m_globalHash->unlock();
-    
-    if(ret == 0){
-      return 0;
-    }
-  }
-  m_localHash.put(internalTableName, ret);
-
-  m_ndb.theFirstTupleId[ret->getTableId()] = ~0;
-  m_ndb.theLastTupleId[ret->getTableId()]  = ~0;
-  
-  addBlobTables(*ret);
-
-  return ret;
+  return fetchGlobalTableImpl(internalTableName);
 }
 
 inline
@@ -654,9 +640,9 @@ NdbDictionaryImpl::getIndex(const char * indexName,
       internalIndexName = m_ndb.internalizeTableName(indexName); // Index is also a table
     }
     if (internalIndexName) {
-      NdbTableImpl * tab = getTableImpl(internalIndexName);
-      
-      if (tab) {
+      Ndb_local_table_info * info = get_local_table_info(internalIndexName);
+      if (info) {
+	NdbTableImpl * tab = info->m_table_impl;
         if (tab->m_index == 0)
           tab->m_index = getIndexImpl(indexName, internalIndexName);
         if (tab->m_index != 0)

ndb/src/ndbapi/NdbLinHash.hpp

@@ -59,7 +59,7 @@ public:
   void releaseHashTable(void);
   
   int insertKey(const char * str, Uint32 len, Uint32 lkey1, C* data);
-  int deleteKey(const char * str, Uint32 len);
+  C *deleteKey(const char * str, Uint32 len);
 
   C* getData(const char *, Uint32);
   Uint32* getKey(const char *, Uint32);
@@ -277,7 +277,7 @@ NdbLinHash<C>::getData( const char* str, Uint32 len ){
 
 template <class C>
 inline
-int
+C *
 NdbLinHash<C>::deleteKey ( const char* str, Uint32 len){
   const Uint32 hash = Hash(str, len);
   int dir, seg;
@@ -288,19 +288,21 @@ NdbLinHash<C>::deleteKey ( const char* str, Uint32 len){
   for(NdbElement_t<C> * chain = *chainp; chain != 0; chain = chain->next){
     if(chain->len == len && !memcmp(chain->str, str, len)){
       if (oldChain == 0) {
+	C *data= chain->theData;
 	delete chain;
 	* chainp = 0;
-	return 1;
+	return data;
       } else {
+	C *data= chain->theData;
 	oldChain->next = chain->next;
 	delete chain;
-	return 1;
+	return data;
       }
     } else {
       oldChain = chain;
     }
   }
-  return -1; /* Element doesn't exist */
+  return 0; /* Element doesn't exist */
 }
 
 template <class C>

sql/ha_ndbcluster.h

@@ -63,6 +63,19 @@ typedef struct st_ndbcluster_share {
   uint table_name_length,use_count;
 } NDB_SHARE;
 
+/*
+  Place holder for ha_ndbcluster thread specific data
+*/
+
+class Thd_ndb {
+ public:
+  Thd_ndb();
+  ~Thd_ndb();
+  Ndb *ndb;
+  ulong count;
+  uint lock_count;
+};
+
 class ha_ndbcluster: public handler
 {
  public:
@@ -147,8 +160,8 @@ class ha_ndbcluster: public handler
   void start_bulk_insert(ha_rows rows);
   int end_bulk_insert();
 
-  static Ndb* seize_ndb();
-  static void release_ndb(Ndb* ndb);
+  static Thd_ndb* seize_thd_ndb();
+  static void release_thd_ndb(Thd_ndb* thd_ndb);
   uint8 table_cache_type() { return HA_CACHE_TBL_NOCACHE; }
     
  private:
@@ -215,6 +228,7 @@ class ha_ndbcluster: public handler
   NdbResultSet *m_active_cursor;
   Ndb *m_ndb;
   void *m_table;
+  void *m_table_info;
   char m_dbname[FN_HEADLEN];
   //char m_schemaname[FN_HEADLEN];
   char m_tabname[FN_HEADLEN];
@@ -238,6 +252,11 @@ class ha_ndbcluster: public handler
   char *blobs_buffer;
   uint32 blobs_buffer_size;
   uint dupkey;
+
+  void records_update();
+  void no_uncommitted_rows_update(int);
+  void no_uncommitted_rows_init(THD *);
+  void no_uncommitted_rows_reset(THD *);
 };
 
 bool ndbcluster_init(void);

sql/sql_class.h

@@ -764,9 +764,8 @@ class THD :public ilink,
     THD_TRANS all;			// Trans since BEGIN WORK
     THD_TRANS stmt;			// Trans for current statement
     uint bdb_lock_count;
-    uint ndb_lock_count;
 #ifdef HAVE_NDBCLUSTER_DB
-    void* ndb;
+    void* thd_ndb;
 #endif
     bool on;
     /*