Fix for bug#5171

6 files changed, 46 insertions, 11 deletions

diff --git a/ndb/include/ndbapi/NdbBlob.hpp b/ndb/include/ndbapi/NdbBlob.hpp
index af4c049d4a7..dc47115d16f 100644
--- a/ndb/include/ndbapi/NdbBlob.hpp
+++ b/ndb/include/ndbapi/NdbBlob.hpp
@@ -234,14 +234,13 @@ private:
   // define blob table
   static void getBlobTableName(char* btname, const NdbTableImpl* t, const NdbColumnImpl* c);
   static void getBlobTable(NdbTableImpl& bt, const NdbTableImpl* t, const NdbColumnImpl* c);
-  // table name
-  char theBlobTableName[BlobTableNameSize];
   // ndb api stuff
   Ndb* theNdb;
   NdbConnection* theNdbCon;
   NdbOperation* theNdbOp;
   NdbTableImpl* theTable;
   NdbTableImpl* theAccessTable;
+  NdbTableImpl* theBlobTable;
   const NdbColumnImpl* theColumn;
   char theFillChar;
   // sizes
diff --git a/ndb/include/ndbapi/NdbDictionary.hpp b/ndb/include/ndbapi/NdbDictionary.hpp
index a69ec355ce8..b5c3985c6cb 100644
--- a/ndb/include/ndbapi/NdbDictionary.hpp
+++ b/ndb/include/ndbapi/NdbDictionary.hpp
@@ -144,6 +144,8 @@ public:
       FragAllLarge = 4        ///< Eight fragments per node group.
     };
   };
+
+  class Table; // forward declaration
   
   /**
    * @class Column
@@ -365,6 +367,8 @@ public:
     void setIndexOnlyStorage(bool);
     bool getIndexOnlyStorage() const;
 
+    const Table * getBlobTable() const;
+
     /** 
      * @name ODBC Specific methods 
      * @{
diff --git a/ndb/src/ndbapi/NdbBlob.cpp b/ndb/src/ndbapi/NdbBlob.cpp
index 431be574bbf..65e1aeedda7 100644
--- a/ndb/src/ndbapi/NdbBlob.cpp
+++ b/ndb/src/ndbapi/NdbBlob.cpp
@@ -140,7 +140,6 @@ void
 NdbBlob::init()
 {
   theState = Idle;
-  theBlobTableName[0] = 0;
   theNdb = NULL;
   theNdbCon = NULL;
   theNdbOp = NULL;
@@ -865,7 +864,7 @@ NdbBlob::readParts(char* buf, Uint32 part, Uint32 count)
   DBG("readParts part=" << part << " count=" << count);
   Uint32 n = 0;
   while (n < count) {
-    NdbOperation* tOp = theNdbCon->getNdbOperation(theBlobTableName);
+    NdbOperation* tOp = theNdbCon->getNdbOperation(theBlobTable);
     if (tOp == NULL ||
         tOp->readTuple() == -1 ||
         setPartKeyValue(tOp, part + n) == -1 ||
@@ -887,7 +886,7 @@ NdbBlob::insertParts(const char* buf, Uint32 part, Uint32 count)
   DBG("insertParts part=" << part << " count=" << count);
   Uint32 n = 0;
   while (n < count) {
-    NdbOperation* tOp = theNdbCon->getNdbOperation(theBlobTableName);
+    NdbOperation* tOp = theNdbCon->getNdbOperation(theBlobTable);
     if (tOp == NULL ||
         tOp->insertTuple() == -1 ||
         setPartKeyValue(tOp, part + n) == -1 ||
@@ -909,7 +908,7 @@ NdbBlob::updateParts(const char* buf, Uint32 part, Uint32 count)
   DBG("updateParts part=" << part << " count=" << count);
   Uint32 n = 0;
   while (n < count) {
-    NdbOperation* tOp = theNdbCon->getNdbOperation(theBlobTableName);
+    NdbOperation* tOp = theNdbCon->getNdbOperation(theBlobTable);
     if (tOp == NULL ||
         tOp->updateTuple() == -1 ||
         setPartKeyValue(tOp, part + n) == -1 ||
@@ -931,7 +930,7 @@ NdbBlob::deleteParts(Uint32 part, Uint32 count)
   DBG("deleteParts part=" << part << " count=" << count);
   Uint32 n = 0;
   while (n < count) {
-    NdbOperation* tOp = theNdbCon->getNdbOperation(theBlobTableName);
+    NdbOperation* tOp = theNdbCon->getNdbOperation(theBlobTable);
     if (tOp == NULL ||
         tOp->deleteTuple() == -1 ||
         setPartKeyValue(tOp, part + n) == -1) {
@@ -1029,12 +1028,11 @@ NdbBlob::atPrepare(NdbConnection* aCon, NdbOperation* anOp, const NdbColumnImpl*
   // sanity check
   assert((NDB_BLOB_HEAD_SIZE << 2) == sizeof(Head));
   assert(theColumn->m_attrSize * theColumn->m_arraySize == sizeof(Head) + theInlineSize);
-  getBlobTableName(theBlobTableName, theTable, theColumn);
   const NdbDictionary::Table* bt;
   const NdbDictionary::Column* bc;
   if (thePartSize > 0) {
     if (theStripeSize == 0 ||
-        (bt = theNdb->theDictionary->getTable(theBlobTableName)) == NULL ||
+        (bt = theColumn->getBlobTable()) == NULL ||
         (bc = bt->getColumn("DATA")) == NULL ||
         bc->getType() != partType ||
         bc->getLength() != (int)thePartSize) {
@@ -1042,6 +1040,7 @@ NdbBlob::atPrepare(NdbConnection* aCon, NdbOperation* anOp, const NdbColumnImpl*
       return -1;
     }
   }
+  theBlobTable = & NdbTableImpl::getImpl(*bt);
   // buffers
   theKeyBuf.alloc(theTable->m_sizeOfKeysInWords << 2);
   theAccessKeyBuf.alloc(theAccessTable->m_sizeOfKeysInWords << 2);
diff --git a/ndb/src/ndbapi/NdbDictionary.cpp b/ndb/src/ndbapi/NdbDictionary.cpp
index d5a16546071..4b30f41b51d 100644
--- a/ndb/src/ndbapi/NdbDictionary.cpp
+++ b/ndb/src/ndbapi/NdbDictionary.cpp
@@ -174,6 +174,14 @@ NdbDictionary::Column::getIndexOnlyStorage() const {
   return m_impl.m_indexOnly;
 }
 
+const NdbDictionary::Table * 
+NdbDictionary::Column::getBlobTable() const {
+  NdbTableImpl * t = m_impl.m_blobTable;
+  if (t)
+    return t->m_facade;
+  return 0;
+}
+
 void 
 NdbDictionary::Column::setAutoIncrement(bool val){
   m_impl.m_autoIncrement = val;
diff --git a/ndb/src/ndbapi/NdbDictionaryImpl.cpp b/ndb/src/ndbapi/NdbDictionaryImpl.cpp
index 45a9f300aab..cb5e3b3c821 100644
--- a/ndb/src/ndbapi/NdbDictionaryImpl.cpp
+++ b/ndb/src/ndbapi/NdbDictionaryImpl.cpp
@@ -79,6 +79,7 @@ NdbColumnImpl::operator=(const NdbColumnImpl& col)
   m_attrSize = col.m_attrSize; 
   m_arraySize = col.m_arraySize;
   m_keyInfoPos = col.m_keyInfoPos;
+  m_blobTable = col.m_blobTable;
   // Do not copy m_facade !!
 
   return *this;
@@ -104,6 +105,7 @@ NdbColumnImpl::init()
   m_arraySize = 1,
   m_autoIncrement = false;
   m_autoIncrementInitialValue = 1;
+  m_blobTable = NULL;
 }
 
 NdbColumnImpl::~NdbColumnImpl()
@@ -1211,7 +1213,6 @@ NdbDictInterface::parseTableInfo(NdbTableImpl ** ret,
     }
     if (col->getBlobType())
       blobCount++;
-    
     NdbColumnImpl * null = 0;
     impl->m_columns.fill(attrDesc.AttributeId, null);
     if(impl->m_columns[attrDesc.AttributeId] != 0){
@@ -1266,7 +1267,28 @@ NdbDictionaryImpl::createBlobTables(NdbTableImpl &t)
     NdbBlob::getBlobTable(bt, &t, &c);
     if (createTable(bt) != 0)
       return -1;
+    // Save BLOB table handle
+    NdbTableImpl * cachedBlobTable = getTable(bt.m_externalName.c_str());
+    c.m_blobTable = cachedBlobTable;
   }
+  
+  return 0;
+}
+
+int
+NdbDictionaryImpl::addBlobTables(NdbTableImpl &t)
+{
+  for (unsigned i = 0; i < t.m_columns.size(); i++) {
+    NdbColumnImpl & c = *t.m_columns[i];
+    if (! c.getBlobType() || c.getPartSize() == 0)
+      continue;
+    char btname[NdbBlob::BlobTableNameSize];
+    NdbBlob::getBlobTableName(btname, &t, &c);
+    // Save BLOB table handle
+    NdbTableImpl * cachedBlobTable = getTable(btname);;
+    c.m_blobTable = cachedBlobTable;
+  }
+  
   return 0;
 }
 
diff --git a/ndb/src/ndbapi/NdbDictionaryImpl.hpp b/ndb/src/ndbapi/NdbDictionaryImpl.hpp
index 3bf7eef3a06..9a890f02575 100644
--- a/ndb/src/ndbapi/NdbDictionaryImpl.hpp
+++ b/ndb/src/ndbapi/NdbDictionaryImpl.hpp
@@ -71,6 +71,7 @@ public:
   bool m_autoIncrement;
   Uint64 m_autoIncrementInitialValue;
   BaseString m_defaultValue;
+  NdbTableImpl * m_blobTable;
 
   /**
    * Internal types and sizes, and aggregates
@@ -362,6 +363,7 @@ public:
   
   int createTable(NdbTableImpl &t);
   int createBlobTables(NdbTableImpl &);
+  int addBlobTables(NdbTableImpl &);
   int alterTable(NdbTableImpl &t);
   int dropTable(const char * name);
   int dropTable(NdbTableImpl &);
@@ -616,7 +618,6 @@ NdbDictionaryImpl::getTableImpl(const char * internalTableName)
 
   if (ret == 0){
     ret = m_receiver.getTable(internalTableName, m_ndb.usingFullyQualifiedNames());
-    
     m_globalHash->lock();
     m_globalHash->put(internalTableName, ret);
     m_globalHash->unlock();
@@ -629,6 +630,8 @@ NdbDictionaryImpl::getTableImpl(const char * internalTableName)
 
   m_ndb.theFirstTupleId[ret->getTableId()] = ~0;
   m_ndb.theLastTupleId[ret->getTableId()]  = ~0;
+  
+  addBlobTables(*ret);
 
   return ret;
 }