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

into mysql.com:/usr/local/home/marty/MySQL/mysql-4.1-ndb


ndb/src/ndbapi/NdbConnection.cpp:
  Auto merged

mysql-test/r/ndb_basic.result

 DROP TABLE IF EXISTS t1,t2,t3,t4,t5,t6,t7;
+drop database if exists test2;
 CREATE TABLE t1 (
 pk1 INT NOT NULL PRIMARY KEY,
 attr1 INT NOT NULL,
@@ -349,3 +350,30 @@ select * from t7;
 adress	a	b	c
 No adress	8	NULL	12
 drop table t7;
+CREATE TABLE t1 (
+pk1 INT NOT NULL PRIMARY KEY,
+attr1 INT NOT NULL,
+attr2 INT,
+attr3 VARCHAR(10)
+) ENGINE=ndbcluster;
+INSERT INTO t1 VALUES (9410,9412, NULL, '9412'), (9411,9413, 17, '9413');
+create database test2;
+use test2;
+CREATE TABLE t2 (
+a bigint unsigned NOT NULL PRIMARY KEY,
+b int unsigned not null,
+c int unsigned
+) engine=ndbcluster;
+insert into t2 select pk1,attr1,attr2 from test.t1;
+select * from t2 order by a;
+a	b	c
+9410	9412	NULL
+9411	9413	17
+select b from test.t1, t2 where c = test.t1.attr2;
+b
+9413
+select b,test.t1.attr1 from test.t1, t2 where test.t1.pk1 < a;
+b	attr1
+9413	9412
+drop table test.t1, t2;
+drop database test2;

mysql-test/r/ndb_transaction.result

 DROP TABLE IF EXISTS t1,t2,t3,t4,t5,t6,t7;
+drop database if exists test2;
 CREATE TABLE t1 (
 pk1 INT NOT NULL PRIMARY KEY,
 attr1 INT NOT NULL
@@ -206,3 +207,51 @@ begin;
 drop table t2;
 drop table t3;
 drop table t4;
+CREATE TABLE t1 (
+pk1 INT NOT NULL PRIMARY KEY,
+attr1 INT NOT NULL
+) ENGINE=ndbcluster;
+create database test2;
+use test2;
+CREATE TABLE t2 (
+a bigint unsigned NOT NULL PRIMARY KEY,
+b int unsigned not null,
+c int unsigned
+) engine=ndbcluster;
+begin;
+insert into test.t1 values(1,1);
+insert into t2 values(1,1,1);
+insert into test.t1 values(2,2);
+insert into t2 values(2,2,2);
+select count(*) from test.t1;
+count(*)
+2
+select count(*) from t2;
+count(*)
+2
+select * from test.t1 where pk1 = 1;
+pk1	attr1
+1	1
+select * from t2 where a = 1;
+a	b	c
+1	1	1
+select test.t1.attr1 
+from test.t1, test.t1 as t1x where test.t1.pk1 = t1x.pk1 + 1;
+attr1
+2
+select t2.a 
+from t2, t2 as t2x where t2.a = t2x.a + 1;
+a
+2
+select test.t1.pk1, a from test.t1,t2 where b > test.t1.attr1;
+pk1	a
+1	2
+rollback;
+select count(*) from test.t1;
+count(*)
+0
+select count(*) from t2;
+count(*)
+0
+drop table test.t1, t2;
+drop database test2;

mysql-test/t/ndb_basic.test

@@ -2,6 +2,7 @@
 
 --disable_warnings
 DROP TABLE IF EXISTS t1,t2,t3,t4,t5,t6,t7;
+drop database if exists test2;
 --enable_warnings
 
 #
@@ -319,3 +320,36 @@ delete from t7 where b=23;
 select * from t7;
 
 drop table t7;
+
+#
+# Test multiple databases in one statement
+#
+
+CREATE TABLE t1 (
+  pk1 INT NOT NULL PRIMARY KEY,
+  attr1 INT NOT NULL,
+  attr2 INT,
+  attr3 VARCHAR(10)
+) ENGINE=ndbcluster;
+
+INSERT INTO t1 VALUES (9410,9412, NULL, '9412'), (9411,9413, 17, '9413');
+
+create database test2;
+use test2;
+
+CREATE TABLE t2 (
+  a bigint unsigned NOT NULL PRIMARY KEY,
+  b int unsigned not null,
+  c int unsigned
+) engine=ndbcluster;
+
+insert into t2 select pk1,attr1,attr2 from test.t1;
+select * from t2 order by a;
+select b from test.t1, t2 where c = test.t1.attr2;
+select b,test.t1.attr1 from test.t1, t2 where test.t1.pk1 < a;
+ 
+drop table test.t1, t2;
+
+drop database test2;
+
+

mysql-test/t/ndb_transaction.test

@@ -2,6 +2,7 @@
 
 --disable_warnings
 DROP TABLE IF EXISTS t1,t2,t3,t4,t5,t6,t7;
+drop database if exists test2;
 --enable_warnings
 
 #
@@ -253,3 +254,45 @@ drop table t2;
 drop table t3;
 drop table t4;
 
+#
+# Test multiple databases in one transaction
+#
+
+CREATE TABLE t1 (
+  pk1 INT NOT NULL PRIMARY KEY,
+  attr1 INT NOT NULL
+) ENGINE=ndbcluster;
+
+create database test2;
+use test2;
+
+CREATE TABLE t2 (
+  a bigint unsigned NOT NULL PRIMARY KEY,
+  b int unsigned not null,
+  c int unsigned
+) engine=ndbcluster;
+
+begin;
+insert into test.t1 values(1,1);
+insert into t2 values(1,1,1);
+insert into test.t1 values(2,2);
+insert into t2 values(2,2,2);
+select count(*) from test.t1;
+select count(*) from t2;
+select * from test.t1 where pk1 = 1;
+select * from t2 where a = 1;
+select test.t1.attr1 
+from test.t1, test.t1 as t1x where test.t1.pk1 = t1x.pk1 + 1;
+select t2.a 
+from t2, t2 as t2x where t2.a = t2x.a + 1;
+select test.t1.pk1, a from test.t1,t2 where b > test.t1.attr1;
+rollback;
+
+select count(*) from test.t1;
+select count(*) from t2;
+
+drop table test.t1, t2;
+
+drop database test2;
+
+

ndb/include/ndbapi/Ndb.hpp

@@ -1416,9 +1416,14 @@ public:
    */
   Uint64 getAutoIncrementValue(const char* aTableName, 
 			       Uint32 cacheSize = 1);
+  Uint64 getAutoIncrementValue(NdbDictionary::Table * aTable, 
+			       Uint32 cacheSize = 1);
   Uint64 readAutoIncrementValue(const char* aTableName);
+  Uint64 readAutoIncrementValue(NdbDictionary::Table * aTable);
   bool setAutoIncrementValue(const char* aTableName, Uint64 val, 
 			     bool increase = false);
+  bool setAutoIncrementValue(NdbDictionary::Table * aTable, Uint64 val, 
+			     bool increase = false);
   Uint64 getTupleIdFromNdb(const char* aTableName, 
 			   Uint32 cacheSize = 1000);
   Uint64 getTupleIdFromNdb(Uint32 aTableId, 

ndb/include/ndbapi/NdbConnection.hpp

@@ -19,6 +19,7 @@
 
 #include <ndb_types.h>
 #include <NdbError.hpp>
+#include <NdbDictionary.hpp>
 
 class NdbConnection;
 class NdbOperation;
@@ -440,6 +441,14 @@ public:
    */
   int executePendingBlobOps(Uint8 flags = 0xFF);
 
+  // Fast path calls for MySQL ha_ndbcluster
+  NdbOperation* getNdbOperation(NdbDictionary::Table * table);
+  NdbIndexOperation* getNdbIndexOperation(NdbDictionary::Index * index,
+					  NdbDictionary::Table * table);
+  NdbScanOperation* getNdbScanOperation(NdbDictionary::Table * table);
+  NdbIndexScanOperation* getNdbIndexScanOperation(NdbDictionary::Index * index,
+						  NdbDictionary::Table * table);
+
 private:						
   /**
    * Release completed operations
@@ -553,6 +562,8 @@ private:
   NdbIndexOperation* getNdbIndexOperation(class NdbIndexImpl* anIndex, 
                                           class NdbTableImpl* aTable,
                                           NdbOperation* aNextOp = 0);
+  NdbIndexScanOperation* getNdbIndexScanOperation(NdbIndexImpl* index,
+						  NdbTableImpl* table);
   
   void		handleExecuteCompletion();
   

ndb/src/ndbapi/Ndb.cpp

@@ -736,6 +736,17 @@ Ndb::getAutoIncrementValue(const char* aTableName, Uint32 cacheSize)
   return tupleId;
 }
 
+Uint64
+Ndb::getAutoIncrementValue(NdbDictionary::Table * aTable, Uint32 cacheSize)
+{
+  DEBUG_TRACE("getAutoIncrementValue");
+  if (aTable == 0)
+    return ~0;
+  const NdbTableImpl* table = & NdbTableImpl::getImpl(*aTable);
+  Uint64 tupleId = getTupleIdFromNdb(table->m_tableId, cacheSize);
+  return tupleId;
+}
+
 Uint64 
 Ndb::getTupleIdFromNdb(const char* aTableName, Uint32 cacheSize)
 {
@@ -770,6 +781,17 @@ Ndb::readAutoIncrementValue(const char* aTableName)
   return tupleId;
 }
 
+Uint64
+Ndb::readAutoIncrementValue(NdbDictionary::Table * aTable)
+{
+  DEBUG_TRACE("readtAutoIncrementValue");
+  if (aTable == 0)
+    return ~0;
+  const NdbTableImpl* table = & NdbTableImpl::getImpl(*aTable);
+  Uint64 tupleId = readTupleIdFromNdb(table->m_tableId);
+  return tupleId;
+}
+
 Uint64
 Ndb::readTupleIdFromNdb(Uint32 aTableId)
 {
@@ -790,6 +812,16 @@ Ndb::setAutoIncrementValue(const char* aTableName, Uint64 val, bool increase)
   return setTupleIdInNdb(table->m_tableId, val, increase);
 }
 
+bool
+Ndb::setAutoIncrementValue(NdbDictionary::Table * aTable, Uint64 val, bool increase)
+{
+  DEBUG_TRACE("setAutoIncrementValue " << val);
+  if (aTable == 0)
+    return ~0;
+  const NdbTableImpl* table = & NdbTableImpl::getImpl(*aTable);
+  return setTupleIdInNdb(table->m_tableId, val, increase);
+}
+
 bool 
 Ndb::setTupleIdInNdb(const char* aTableName, Uint64 val, bool increase )
 {

ndb/src/ndbapi/NdbConnection.cpp

@@ -995,6 +995,14 @@ NdbConnection::getNdbOperation(NdbTableImpl * tab, NdbOperation* aNextOp)
   return NULL;
 }//NdbConnection::getNdbOperation()
 
+NdbOperation* NdbConnection::getNdbOperation(NdbDictionary::Table * table)
+{
+  if (table)
+    return getNdbOperation(& NdbTableImpl::getImpl(*table));
+  else
+    return NULL;
+}//NdbConnection::getNdbOperation()
+
 // NdbScanOperation
 /*****************************************************************************
 NdbScanOperation* getNdbScanOperation(const char* aTableName);
@@ -1037,15 +1045,24 @@ Remark:         Get an operation from NdbScanOperation idlelist and get the NdbC
 NdbIndexScanOperation*
 NdbConnection::getNdbIndexScanOperation(const char* anIndexName, 
 					const char* aTableName)
+{
+  NdbIndexImpl* index = 
+    theNdb->theDictionary->getIndex(anIndexName, aTableName);
+  NdbTableImpl* table = theNdb->theDictionary->getTable(aTableName);
+
+  return getNdbIndexScanOperation(index, table);
+}
+
+NdbIndexScanOperation*
+NdbConnection::getNdbIndexScanOperation(NdbIndexImpl* index,
+					NdbTableImpl* table)
 {
   if (theCommitStatus == Started){
-    NdbIndexImpl* index = 
-      theNdb->theDictionary->getIndex(anIndexName, aTableName);
-    NdbTableImpl* table = theNdb->theDictionary->getTable(aTableName);
-    NdbTableImpl* indexTable = 
-      theNdb->theDictionary->getIndexTable(index, table);
+    const NdbTableImpl * indexTable = index->getIndexTable();
     if (indexTable != 0){
-      NdbIndexScanOperation* tOp = getNdbScanOperation(indexTable);
+      NdbIndexScanOperation* tOp = 
+	getNdbScanOperation((NdbTableImpl *) indexTable);
+      tOp->m_currentTable = table;
       if(tOp) tOp->m_cursor_type = NdbScanOperation::IndexCursor;
       return tOp;
     } else {
@@ -1056,7 +1073,18 @@ NdbConnection::getNdbIndexScanOperation(const char* anIndexName,
   
   setOperationErrorCodeAbort(4114);
   return NULL;
-}//NdbConnection::getNdbScanOperation()
+}//NdbConnection::getNdbIndexScanOperation()
+
+NdbIndexScanOperation* 
+NdbConnection::getNdbIndexScanOperation(NdbDictionary::Index * index,
+					NdbDictionary::Table * table)
+{
+  if (index && table)
+    return getNdbIndexScanOperation(& NdbIndexImpl::getImpl(*index),
+				    & NdbTableImpl::getImpl(*table));
+  else
+    return NULL;
+}//NdbConnection::getNdbIndexScanOperation()
 
 /*****************************************************************************
 NdbScanOperation* getNdbScanOperation(int aTableId);
@@ -1097,6 +1125,14 @@ getNdbOp_error1:
   return NULL;
 }//NdbConnection::getNdbScanOperation()
 
+NdbScanOperation* 
+NdbConnection::getNdbScanOperation(NdbDictionary::Table * table)
+{
+  if (table)
+    return getNdbScanOperation(& NdbTableImpl::getImpl(*table));
+  else
+    return NULL;
+}//NdbConnection::getNdbScanOperation()
 
 
 // IndexOperation
@@ -1191,6 +1227,18 @@ NdbConnection::getNdbIndexOperation(NdbIndexImpl * anIndex,
   return NULL;
 }//NdbConnection::getNdbIndexOperation()
 
+NdbIndexOperation* 
+NdbConnection::getNdbIndexOperation(NdbDictionary::Index * index,
+				    NdbDictionary::Table * table)
+{
+  if (index && table)
+    return getNdbIndexOperation(& NdbIndexImpl::getImpl(*index),
+				& NdbTableImpl::getImpl(*table));
+  else
+    return NULL;
+}//NdbConnection::getNdbIndexOperation()
+
+
 /*******************************************************************************
 int  receiveDIHNDBTAMPER(NdbApiSignal* aSignal)
 

ndb/src/ndbapi/NdbDictionaryImpl.cpp

@@ -492,6 +492,12 @@ NdbIndexImpl::getTable() const
   return m_tableName.c_str();
 }
 
+const NdbTableImpl *
+NdbIndexImpl::getIndexTable() const
+{
+  return m_table;
+}
+
 /**
  * NdbEventImpl
  */

ndb/src/ndbapi/NdbDictionaryImpl.hpp

@@ -170,6 +170,7 @@ public:
   const char * getName() const;
   void setTable(const char * table);
   const char * getTable() const;
+  const NdbTableImpl * getIndexTable() const;
 
   Uint32 m_indexId;
   BaseString m_internalName;

ndb/src/ndbapi/NdbScanOperation.cpp

@@ -188,12 +188,15 @@ NdbResultSet* NdbScanOperation::readTuples(NdbScanOperation::LockMode lm,
   m_keyInfo = lockExcl;
 
   bool range = false;
-  if (m_currentTable->m_indexType == NdbDictionary::Index::OrderedIndex ||
-      m_currentTable->m_indexType == NdbDictionary::Index::UniqueOrderedIndex){
-    assert(m_currentTable == m_accessTable);
-    m_currentTable = theNdb->theDictionary->
-      getTable(m_currentTable->m_primaryTable.c_str());
-    assert(m_currentTable != NULL);
+  if (m_accessTable->m_indexType == NdbDictionary::Index::OrderedIndex ||
+      m_accessTable->m_indexType == NdbDictionary::Index::UniqueOrderedIndex){
+    if (m_currentTable == m_accessTable){
+      // Old way of scanning indexes, should not be allowed
+      m_currentTable = theNdb->theDictionary->
+	getTable(m_currentTable->m_primaryTable.c_str());
+      assert(m_currentTable != NULL);
+    }
+    assert (m_currentTable != m_accessTable);
     // Modify operation state
     theStatus = SetBound;
     theOperationType  = OpenRangeScanRequest;

sql/ha_ndbcluster.h

@@ -37,6 +37,7 @@ class NdbScanOperation;
 class NdbIndexScanOperation; 
 class NdbBlob;
 
+
 typedef enum ndb_index_type {
   UNDEFINED_INDEX = 0,
   PRIMARY_KEY_INDEX = 1,
@@ -46,6 +47,12 @@ typedef enum ndb_index_type {
   ORDERED_INDEX = 5
 } NDB_INDEX_TYPE;
 
+typedef struct ndb_index_data {
+  NDB_INDEX_TYPE type;
+  void *index;
+  const char * unique_name;
+  void *unique_index;
+} NDB_INDEX_DATA;
 
 typedef struct st_ndbcluster_share {
   THR_LOCK lock;
@@ -148,8 +155,9 @@ class ha_ndbcluster: public handler
   int create_index(const char *name, KEY *key_info, bool unique);
   int create_ordered_index(const char *name, KEY *key_info);
   int create_unique_index(const char *name, KEY *key_info);
-  int initialize_autoincrement(const void* table);
-  int build_index_list();
+  int initialize_autoincrement(const void *table);
+  enum IBLP {ILBP_CREATE = 0, ILBP_OPEN = 1}; // index_list_build_phase
+  int build_index_list(TABLE *tab, enum IBLP phase);
   int get_metadata(const char* path);
   void release_metadata();
   const char* get_index_name(uint idx_no) const;
@@ -211,8 +219,7 @@ class ha_ndbcluster: public handler
   ulong m_table_flags;
   THR_LOCK_DATA m_lock;
   NDB_SHARE *m_share;
-  NDB_INDEX_TYPE  m_indextype[MAX_KEY];
-  const char*  m_unique_index_name[MAX_KEY];
+  NDB_INDEX_DATA  m_index[MAX_KEY];
   // NdbRecAttr has no reference to blob
   typedef union { NdbRecAttr *rec; NdbBlob *blob; void *ptr; } NdbValue;
   NdbValue m_value[NDB_MAX_ATTRIBUTES_IN_TABLE];