Fix for WL#1731 Handler: multiple databases

5 files changed, 103 insertions, 13 deletions

diff --git a/ndb/src/ndbapi/Ndb.cpp b/ndb/src/ndbapi/Ndb.cpp
index bac367bb689..f09a7481d2d 100644
--- a/ndb/src/ndbapi/Ndb.cpp
+++ b/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)
 {
@@ -771,6 +782,17 @@ Ndb::readAutoIncrementValue(const char* aTableName)
 }
 
 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)
 {
   if ( theFirstTupleId[aTableId] == theLastTupleId[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 )
 {
diff --git a/ndb/src/ndbapi/NdbConnection.cpp b/ndb/src/ndbapi/NdbConnection.cpp
index cd051bb4609..8b8ae4537c8 100644
--- a/ndb/src/ndbapi/NdbConnection.cpp
+++ b/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);
@@ -1038,14 +1046,23 @@ 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)
 
diff --git a/ndb/src/ndbapi/NdbDictionaryImpl.cpp b/ndb/src/ndbapi/NdbDictionaryImpl.cpp
index c4ea9909fcd..45a9f300aab 100644
--- a/ndb/src/ndbapi/NdbDictionaryImpl.cpp
+++ b/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
  */
diff --git a/ndb/src/ndbapi/NdbDictionaryImpl.hpp b/ndb/src/ndbapi/NdbDictionaryImpl.hpp
index 85d334416ce..3bf7eef3a06 100644
--- a/ndb/src/ndbapi/NdbDictionaryImpl.hpp
+++ b/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;
diff --git a/ndb/src/ndbapi/NdbScanOperation.cpp b/ndb/src/ndbapi/NdbScanOperation.cpp
index 7d51974da7c..04043f5a4c5 100644
--- a/ndb/src/ndbapi/NdbScanOperation.cpp
+++ b/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;