ndb - bug#17761 blob tables patch 3b [closes the bug]

storage/ndb/src/common/debugger/signaldata/DictTabInfo.cpp:
  oops (but did not affect previous patch)
storage/ndb/include/ndbapi/NdbDictionary.hpp:
  allow NDB API programs direct access to blob tables via main table or by name (hack)
storage/ndb/src/ndbapi/NdbDictionary.cpp:
  allow NDB API programs direct access to blob tables via main table or by name (hack)
storage/ndb/src/ndbapi/NdbDictionaryImpl.cpp:
  allow NDB API programs direct access to blob tables via main table or by name (hack)
storage/ndb/src/ndbapi/NdbDictionaryImpl.hpp:
  allow NDB API programs direct access to blob tables via main table or by name (hack)
storage/ndb/src/ndbapi/ndberror.c:
  allow NDB API programs direct access to blob tables via main table or by name (hack)

storage/ndb/include/ndbapi/NdbDictionary.hpp

@@ -1602,6 +1602,12 @@ public:
     const Table * getTable(const char * name) const;
 
 #ifndef DOXYGEN_SHOULD_SKIP_INTERNAL
+    /**
+     * Given main table, get blob table.
+     */
+    const Table * getBlobTable(const Table *, const char * col_name);
+    const Table * getBlobTable(const Table *, Uint32 col_no);
+
     /*
      * Save a table definition in dictionary cache
      * @param table Object to put into cache

storage/ndb/src/common/debugger/signaldata/DictTabInfo.cpp

@@ -274,13 +274,13 @@ DictTabInfo::isBlobTableName(const char* name, Uint32* ptab_id, Uint32* pcol_no)
   s += strlen(prefix);
   uint i, n;
   for (i = 0, n = 0; '0' <= s[i] && s[i] <= '9'; i++)
-    n += 10 * n + (s[i] - '0');
+    n = 10 * n + (s[i] - '0');
   if (i == 0 || s[i] != '_')
     return false;
   const uint tab_id = n;
   s = &s[i + 1];
   for (i = 0, n = 0; '0' <= s[i] && s[i] <= '9'; i++)
-    n += 10 * n + (s[i] - '0');
+    n = 10 * n + (s[i] - '0');
   if (i == 0 || s[i] != 0)
     return false;
   const uint col_no = n;

storage/ndb/src/ndbapi/NdbDictionary.cpp

@@ -1367,6 +1367,25 @@ NdbDictionary::Dictionary::getTable(const char * name) const
   return getTable(name, 0);
 }
 
+const NdbDictionary::Table *
+NdbDictionary::Dictionary::getBlobTable(const NdbDictionary::Table* table,
+                                        const char* col_name)
+{
+  const NdbDictionary::Column* col = table->getColumn(col_name);
+  if (col == NULL) {
+    m_impl.m_error.code = 4318;
+    return NULL;
+  }
+  return getBlobTable(table, col->getColumnNo());
+}
+
+const NdbDictionary::Table *
+NdbDictionary::Dictionary::getBlobTable(const NdbDictionary::Table* table,
+                                        Uint32 col_no)
+{
+  return m_impl.getBlobTable(NdbTableImpl::getImpl(*table), col_no);
+}
+
 void
 NdbDictionary::Dictionary::invalidateTable(const char * name){
   DBUG_ENTER("NdbDictionaryImpl::invalidateTable");

storage/ndb/src/ndbapi/NdbDictionaryImpl.cpp

@@ -50,9 +50,9 @@
 extern Uint64 g_latest_trans_gci;
 
 bool
-is_ndb_blob_table(const char* name)
+is_ndb_blob_table(const char* name, Uint32* ptab_id, Uint32* pcol_no)
 {
-  return DictTabInfo::isBlobTableName(name);
+  return DictTabInfo::isBlobTableName(name, ptab_id, pcol_no);
 }
 
 bool
@@ -1383,6 +1383,43 @@ NdbDictionaryImpl::getBlobTables(NdbTableImpl &t)
   DBUG_RETURN(0); 
 }
 
+NdbTableImpl*
+NdbDictionaryImpl::getBlobTable(const NdbTableImpl& tab, uint col_no)
+{
+  if (col_no < tab.m_columns.size()) {
+    NdbColumnImpl* col = tab.m_columns[col_no];
+    if (col != NULL) {
+      NdbTableImpl* bt = col->m_blobTable;
+      if (bt != NULL)
+        return bt;
+      else
+        m_error.code = 4273; // No blob table..
+    } else
+      m_error.code = 4249; // Invalid table..
+  } else
+    m_error.code = 4318; // Invalid attribute..
+  return NULL;
+}
+
+NdbTableImpl*
+NdbDictionaryImpl::getBlobTable(uint tab_id, uint col_no)
+{
+  DBUG_ENTER("NdbDictionaryImpl::getBlobTable");
+  DBUG_PRINT("enter", ("tab_id: %u col_no %u", tab_id, col_no));
+
+  NdbTableImpl* tab = m_receiver.getTable(tab_id,
+                                          m_ndb.usingFullyQualifiedNames());
+  if (tab == NULL)
+    DBUG_RETURN(NULL);
+  Ndb_local_table_info* info =
+    get_local_table_info(tab->m_internalName);
+  delete tab;
+  if (info == NULL)
+    DBUG_RETURN(NULL);
+  NdbTableImpl* bt = getBlobTable(*info->m_table_impl, col_no);
+  DBUG_RETURN(bt);
+}
+
 #if 0
 bool
 NdbDictionaryImpl::setTransporter(class TransporterFacade * tf)
@@ -1697,7 +1734,7 @@ NdbDictInterface::dictSignal(NdbApiSignal* sig,
   }
   DBUG_RETURN(-1);
 }
-#if 0
+
 /*
   Get dictionary information for a table using table id as reference
 
@@ -1721,8 +1758,6 @@ NdbDictInterface::getTable(int tableId, bool fullyQualifiedNames)
 
   return getTable(&tSignal, 0, 0, fullyQualifiedNames);
 }
-#endif
-
 
 /*
   Get dictionary information for a table using table name as the reference

storage/ndb/src/ndbapi/NdbDictionaryImpl.hpp

@@ -30,8 +30,10 @@
 #include "NdbWaiter.hpp"
 #include "DictCache.hpp"
 
-bool is_ndb_blob_table(const char* name);
-bool is_ndb_blob_table(const class NdbTableImpl* t);
+bool
+is_ndb_blob_table(const char* name, Uint32* ptab_id = 0, Uint32* pcol_no = 0);
+bool
+is_ndb_blob_table(const class NdbTableImpl* t);
 
 class NdbDictObjectImpl {
 public:
@@ -440,7 +442,7 @@ public:
   int listObjects(NdbDictionary::Dictionary::List& list, Uint32 requestData, bool fullyQualifiedNames);
   int listObjects(NdbApiSignal* signal);
   
-/*  NdbTableImpl * getTable(int tableId, bool fullyQualifiedNames); */
+  NdbTableImpl * getTable(int tableId, bool fullyQualifiedNames);
   NdbTableImpl * getTable(const BaseString& name, bool fullyQualifiedNames);
   NdbTableImpl * getTable(class NdbApiSignal * signal, 
 			  LinearSectionPtr ptr[3],
@@ -574,6 +576,8 @@ public:
   int listIndexes(List& list, Uint32 indexId);
 
   NdbTableImpl * getTable(const char * tableName, void **data= 0);
+  NdbTableImpl * getBlobTable(const NdbTableImpl&, uint col_no);
+  NdbTableImpl * getBlobTable(uint tab_id, uint col_no);
   void putTable(NdbTableImpl *impl);
   int getBlobTables(NdbTableImpl &);
   Ndb_local_table_info*
@@ -849,28 +853,42 @@ inline
 NdbTableImpl *
 NdbDictionaryImpl::getTable(const char * table_name, void **data)
 {
+  DBUG_ENTER("NdbDictionaryImpl::getTable");
+  DBUG_PRINT("enter", ("table: %s", table_name));
+
+  if (unlikely(strchr(table_name, '$') != 0)) {
+    Uint32 tab_id, col_no;
+    if (is_ndb_blob_table(table_name, &tab_id, &col_no)) {
+      NdbTableImpl* t = getBlobTable(tab_id, col_no);
+      DBUG_RETURN(t);
+    }
+  }
+
   const BaseString internal_tabname(m_ndb.internalize_table_name(table_name));
   Ndb_local_table_info *info=
     get_local_table_info(internal_tabname);
   if (info == 0)
-    return 0;
+    DBUG_RETURN(0);
   if (data)
     *data= info->m_local_data;
-  return info->m_table_impl;
+  DBUG_RETURN(info->m_table_impl);
 }
 
 inline
 Ndb_local_table_info * 
 NdbDictionaryImpl::get_local_table_info(const BaseString& internalTableName)
 {
+  DBUG_ENTER("NdbDictionaryImpl::get_local_table_info");
+  DBUG_PRINT("enter", ("table: %s", internalTableName.c_str()));
+
   Ndb_local_table_info *info= m_localHash.get(internalTableName.c_str());
   if (info == 0) {
     info= fetchGlobalTableImpl(internalTableName);
     if (info == 0) {
-      return 0;
+      DBUG_RETURN(0);
     }
   }
-  return info; // autoincrement already initialized
+  DBUG_RETURN(info); // autoincrement already initialized
 }
 
 inline

storage/ndb/src/ndbapi/ndberror.c

@@ -506,7 +506,7 @@ ErrorBundle ErrorCodes[] = {
   { 4315, DMEC, AE, "No more key attributes allowed after defining variable length key attribute" },
   { 4316, DMEC, AE, "Key attributes are not allowed to be NULL attributes" },
   { 4317, DMEC, AE, "Too many primary keys defined in table" },
-  { 4318, DMEC, AE, "Invalid attribute name" },
+  { 4318, DMEC, AE, "Invalid attribute name or number" },
   { 4319, DMEC, AE, "createAttribute called at erroneus place" },
   { 4322, DMEC, AE, "Attempt to define distribution key when not prepared to" },
   { 4323, DMEC, AE, "Distribution Key set on table but not defined on first attribute" },
@@ -598,7 +598,8 @@ ErrorBundle ErrorCodes[] = {
   { 4335, DMEC, AE, "Only one autoincrement column allowed per table. Having a table without primary key uses an autoincremented hidden key, i.e. a table without a primary key can not have an autoincremented column" },
   { 4336, DMEC, AE, "Auto-increment value set below current value" },
   { 4271, DMEC, AE, "Invalid index object, not retrieved via getIndex()" },
-  { 4272, DMEC, AE, "Table definition has undefined column" }
+  { 4272, DMEC, AE, "Table definition has undefined column" },
+  { 4273, DMEC, IE, "No blob table in dict cache" }
 };
 
 static