Bug#5429

Use const table * in NDB API

ndb/include/ndbapi/Ndb.hpp

@@ -1363,15 +1363,6 @@ public:
    */
   static void setConnectString(const char * connectString);
 
-  /**
-   * useFullyQualifiedNames
-   * Enables unique name space for different databases and schemas
-   * by defining table names as DATABASENAME/SCHEMANAME/TABLENAME and
-   * index names as DATABASENAME/SCHEMANAME/TABLENAME/INDEXNAME
-   * @param turnNamingOn bool true - turn naming on, false - turn naming off
-   */
-  void useFullyQualifiedNames(bool turnNamingOn = true);
-
   bool usingFullyQualifiedNames();
 
   /** @} *********************************************************************/
@@ -1422,13 +1413,13 @@ public:
    */
   Uint64 getAutoIncrementValue(const char* aTableName, 
 			       Uint32 cacheSize = 1);
-  Uint64 getAutoIncrementValue(NdbDictionary::Table * aTable, 
+  Uint64 getAutoIncrementValue(const NdbDictionary::Table * aTable, 
 			       Uint32 cacheSize = 1);
   Uint64 readAutoIncrementValue(const char* aTableName);
-  Uint64 readAutoIncrementValue(NdbDictionary::Table * aTable);
+  Uint64 readAutoIncrementValue(const NdbDictionary::Table * aTable);
   bool setAutoIncrementValue(const char* aTableName, Uint64 val, 
 			     bool increase = false);
-  bool setAutoIncrementValue(NdbDictionary::Table * aTable, Uint64 val, 
+  bool setAutoIncrementValue(const NdbDictionary::Table * aTable, Uint64 val, 
 			     bool increase = false);
   Uint64 getTupleIdFromNdb(const char* aTableName, 
 			   Uint32 cacheSize = 1000);

ndb/include/ndbapi/NdbBlob.hpp

@@ -238,9 +238,9 @@ private:
   Ndb* theNdb;
   NdbConnection* theNdbCon;
   NdbOperation* theNdbOp;
-  NdbTableImpl* theTable;
-  NdbTableImpl* theAccessTable;
-  NdbTableImpl* theBlobTable;
+  const NdbTableImpl* theTable;
+  const NdbTableImpl* theAccessTable;
+  const NdbTableImpl* theBlobTable;
   const NdbColumnImpl* theColumn;
   char theFillChar;
   // sizes

ndb/include/ndbapi/NdbConnection.hpp

@@ -442,12 +442,12 @@ 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);
+  NdbOperation* getNdbOperation(const NdbDictionary::Table * table);
+  NdbIndexOperation* getNdbIndexOperation(const NdbDictionary::Index *,
+					  const NdbDictionary::Table * table);
+  NdbScanOperation* getNdbScanOperation(const NdbDictionary::Table * table);
+  NdbIndexScanOperation* getNdbIndexScanOperation(const NdbDictionary::Index * index,
+						  const NdbDictionary::Table * table);
   
 private:						
   /**
@@ -556,14 +556,14 @@ private:
   void		setOperationErrorCodeAbort(int anErrorCode);
 
   int		checkMagicNumber();		       // Verify correct object
-  NdbOperation* getNdbOperation(class NdbTableImpl* aTable,
+  NdbOperation* getNdbOperation(const class NdbTableImpl* aTable,
                                 NdbOperation* aNextOp = 0);
-  NdbIndexScanOperation* getNdbScanOperation(class NdbTableImpl* aTable);
-  NdbIndexOperation* getNdbIndexOperation(class NdbIndexImpl* anIndex, 
-                                          class NdbTableImpl* aTable,
+  NdbIndexScanOperation* getNdbScanOperation(const class NdbTableImpl* aTable);
+  NdbIndexOperation* getNdbIndexOperation(const class NdbIndexImpl* anIndex, 
+                                          const class NdbTableImpl* aTable,
                                           NdbOperation* aNextOp = 0);
-  NdbIndexScanOperation* getNdbIndexScanOperation(NdbIndexImpl* index,
-						  NdbTableImpl* table);
+  NdbIndexScanOperation* getNdbIndexScanOperation(const NdbIndexImpl* index,
+						  const NdbTableImpl* table);
   
   void		handleExecuteCompletion();
   

ndb/include/ndbapi/NdbIndexOperation.hpp

@@ -175,15 +175,15 @@ private:
   int executeCursor(int ProcessorId);
 
   // Overloaded methods from NdbCursorOperation
-  int indxInit(class NdbIndexImpl* anIndex,
-	       class NdbTableImpl* aTable, 
+  int indxInit(const class NdbIndexImpl* anIndex,
+	       const class NdbTableImpl* aTable, 
 	       NdbConnection* myConnection);
 
   int equal_impl(const class NdbColumnImpl*, const char* aValue, Uint32 len);
   int prepareSend(Uint32  TC_ConnectPtr, Uint64  TransactionId);
 
   // Private attributes
-  NdbIndexImpl* m_theIndex;
+  const NdbIndexImpl* m_theIndex;
   Uint32 m_theIndexDefined[NDB_MAX_ATTRIBUTES_IN_INDEX][3];
   Uint32 m_theIndexLen;	  	 // Length of the index in words
   Uint32 m_theNoOfIndexDefined;  // The number of index attributes

ndb/include/ndbapi/NdbOperation.hpp

@@ -709,7 +709,7 @@ protected:
 //--------------------------------------------------------------
 // Initialise after allocating operation to a transaction		      
 //--------------------------------------------------------------
-  int init(class NdbTableImpl*, NdbConnection* aCon);
+  int init(const class NdbTableImpl*, NdbConnection* aCon);
   void initInterpreter();
 
   void	next(NdbOperation*);		// Set next pointer		      
@@ -858,8 +858,8 @@ protected:
   Uint32*           theKEYINFOptr;       // Pointer to where to write KEYINFO
   Uint32*           theATTRINFOptr;      // Pointer to where to write ATTRINFO
 
-  class NdbTableImpl* m_currentTable;      // The current table
-  class NdbTableImpl* m_accessTable;
+  const class NdbTableImpl* m_currentTable;      // The current table
+  const class NdbTableImpl* m_accessTable;
 
   // Set to TRUE when a tuple key attribute has been defined. 
   Uint32	    theTupleKeyDefined[NDB_MAX_NO_OF_ATTRIBUTES_IN_KEY][3];

ndb/include/ndbapi/NdbScanOperation.hpp

@@ -110,7 +110,7 @@ protected:
   int executeCursor(int ProcessorId);
 
   // Overloaded private methods from NdbOperation
-  int init(NdbTableImpl* tab, NdbConnection* myConnection);
+  int init(const NdbTableImpl* tab, NdbConnection* myConnection);
   int prepareSend(Uint32  TC_ConnectPtr, Uint64  TransactionId);
   int doSend(int ProcessorId);
 

ndb/src/kernel/blocks/backup/restore/Restore.cpp

@@ -240,6 +240,7 @@ RestoreMetaData::parseTableDescriptor(const Uint32 * data, Uint32 len)
 
   debug << "Pushing table " << table->getTableName() << endl;
   debug << "   with " << table->getNoOfAttributes() << " attributes" << endl;
+  
   allTables.push_back(table);
 
   return true;
@@ -683,8 +684,8 @@ RestoreDataIterator::validateFragmentFooter() {
 AttributeDesc::AttributeDesc(NdbDictionary::Column *c)
   : m_column(c)
 {
-  size = c->getSize()*8;
-  arraySize = c->getLength();
+  size = 8*NdbColumnImpl::getImpl(* c).m_attrSize;
+  arraySize = NdbColumnImpl::getImpl(* c).m_arraySize;
 }
 
 void TableS::createAttr(NdbDictionary::Column *column)

ndb/src/kernel/blocks/backup/restore/consumer_restore.cpp

@@ -36,9 +36,6 @@ BackupRestore::init()
   if (m_ndb == NULL)
     return false;
   
-  // Turn off table name completion
-  m_ndb->useFullyQualifiedNames(false);
-
   m_ndb->init(1024);
   if (m_ndb->waitUntilReady(30) != 0)
   {
@@ -102,19 +99,82 @@ BackupRestore::~BackupRestore()
   release();
 }
 
+static
+int 
+match_blob(const char * name){
+  int cnt, id1, id2;
+  char buf[256];
+  if((cnt = sscanf(name, "%[^/]/%[^/]/NDB$BLOB_%d_%d", buf, buf, &id1, &id2)) == 4){
+    return id1;
+  }
+  
+  return -1;
+}
+
+const NdbDictionary::Table*
+BackupRestore::get_table(const NdbDictionary::Table* tab){
+  if(m_cache.m_old_table == tab)
+    return m_cache.m_new_table;
+  m_cache.m_old_table = tab;
+
+  int cnt, id1, id2;
+  char buf[256];
+  if((cnt = sscanf(tab->getName(), "%[^/]/%[^/]/NDB$BLOB_%d_%d", buf, buf, &id1, &id2)) == 4){
+    snprintf(buf, sizeof(buf), "NDB$BLOB_%d_%d", m_new_tables[id1]->getTableId(), id2);
+    m_cache.m_new_table = m_ndb->getDictionary()->getTable(buf);
+  } else {
+    m_cache.m_new_table = m_new_tables[tab->getTableId()];
+  }
+  
+  return m_cache.m_new_table;
+}
+
 bool
 BackupRestore::table(const TableS & table){
-  if (!m_restore_meta) 
+  if (!m_restore && !m_restore_meta)
+    return true;
+
+  const char * name = table.getTableName();
+  
+  /**
+   * Ignore blob tables
+   */
+  if(match_blob(name) >= 0)
     return true;
 
+  BaseString tmp(name);
+  Vector<BaseString> split;
+  if(tmp.split(split, "/") != 3){
+    err << "Invalid table name format " << name << endl;
+    return false;
+  }
+
+  m_ndb->setDatabaseName(split[0].c_str());
+  m_ndb->setSchemaName(split[1].c_str());
+  
   NdbDictionary::Dictionary* dict = m_ndb->getDictionary();
-  if (dict->createTable(*table.m_dictTable) == -1) 
+  if(m_restore_meta){
+    NdbDictionary::Table copy(*table.m_dictTable);
+
+    copy.setName(split[2].c_str());
+
+    if (dict->createTable(copy) == -1) 
     {
       err << "Create table " << table.getTableName() << " failed: "
 	  << dict->getNdbError() << endl;
       return false;
     }
     info << "Successfully restored table " << table.getTableName()<< endl ;
+  }  
+  
+  const NdbDictionary::Table* tab = dict->getTable(split[2].c_str());
+  if(tab == 0){
+    err << "Unable to find table: " << split[2].c_str() << endl;
+    return false;
+  }
+  const NdbDictionary::Table* null = 0;
+  m_new_tables.fill(table.m_dictTable->getTableId(), null);
+  m_new_tables[table.m_dictTable->getTableId()] = tab;
   return true;
 }
 
@@ -161,8 +221,9 @@ void BackupRestore::tuple_a(restore_callback_t *cb)
     } // if
     
     const TupleS &tup = *(cb->tup);
-    const TableS * table = tup.getTable();
-    NdbOperation * op = cb->connection->getNdbOperation(table->getTableName());
+    const NdbDictionary::Table * table = get_table(tup.getTable()->m_dictTable);
+
+    NdbOperation * op = cb->connection->getNdbOperation(table);
     
     if (op == NULL) 
     {
@@ -203,8 +264,9 @@ void BackupRestore::tuple_a(restore_callback_t *cb)
 	    ret = op->setValue(i, dataPtr, length);
 	}
 	if (ret < 0) {
-	  ndbout_c("Column: %d type %d",i,
-		   attr_desc->m_column->getType());
+	  ndbout_c("Column: %d type %d %d %d %d",i,
+		   attr_desc->m_column->getType(),
+		   size, arraySize, attr_data->size);
 	  break;
 	}
       }
@@ -349,8 +411,8 @@ BackupRestore::logEntry(const LogEntry & tup)
     exit(-1);
   } // if
   
-  const TableS * table = tup.m_table;
-  NdbOperation * op = trans->getNdbOperation(table->getTableName());
+  const NdbDictionary::Table * table = get_table(tup.m_table->m_dictTable);
+  NdbOperation * op = trans->getNdbOperation(table);
   if (op == NULL) 
   {
     err << "Cannot get operation: " << trans->getNdbError() << endl;
@@ -514,3 +576,6 @@ BackupRestore::tuple(const TupleS & tup)
   m_dataCount++;
 }
 #endif
+
+template class Vector<NdbDictionary::Table*>;
+template class Vector<const NdbDictionary::Table*>;

ndb/src/kernel/blocks/backup/restore/consumer_restore.hpp

@@ -42,6 +42,7 @@ public:
     m_tuples = 0;
     m_free_callback = 0;
     m_transactions = 0;
+    m_cache.m_old_table = 0;
   }
   
   virtual ~BackupRestore();
@@ -70,6 +71,18 @@ public:
   TupleS *m_tuples;
   restore_callback_t *m_callback;
   restore_callback_t *m_free_callback;
+
+  /**
+   * m_new_table_ids[X] = Y;
+   *   X - old table id
+   *   Y != 0  - new table
+   */
+  Vector<const NdbDictionary::Table*> m_new_tables;
+  struct {
+    const NdbDictionary::Table* m_old_table;
+    const NdbDictionary::Table* m_new_table;
+  } m_cache;
+  const NdbDictionary::Table* get_table(const NdbDictionary::Table* );
 };
 
 #endif

ndb/src/kernel/blocks/backup/restore/main.cpp

@@ -322,7 +322,7 @@ main(int argc, const char** argv)
 	
 	dataIter.validateFooter(); //not implemented
 
-	for (Uint32 i= 0; i < g_consumers.size(); i++)
+	for (i= 0; i < g_consumers.size(); i++)
 	  g_consumers[i]->endOfTuples();
 	
 	RestoreLogIterator logIter(metaData);

ndb/src/ndbapi/Ndb.cpp

@@ -761,7 +761,7 @@ Ndb::getAutoIncrementValue(const char* aTableName, Uint32 cacheSize)
 }
 
 Uint64
-Ndb::getAutoIncrementValue(NdbDictionary::Table * aTable, Uint32 cacheSize)
+Ndb::getAutoIncrementValue(const NdbDictionary::Table * aTable, Uint32 cacheSize)
 {
   DEBUG_TRACE("getAutoIncrementValue");
   if (aTable == 0)
@@ -808,7 +808,7 @@ Ndb::readAutoIncrementValue(const char* aTableName)
 }
 
 Uint64
-Ndb::readAutoIncrementValue(NdbDictionary::Table * aTable)
+Ndb::readAutoIncrementValue(const NdbDictionary::Table * aTable)
 {
   DEBUG_TRACE("readtAutoIncrementValue");
   if (aTable == 0)
@@ -841,7 +841,7 @@ Ndb::setAutoIncrementValue(const char* aTableName, Uint64 val, bool increase)
 }
 
 bool
-Ndb::setAutoIncrementValue(NdbDictionary::Table * aTable, Uint64 val, bool increase)
+Ndb::setAutoIncrementValue(const NdbDictionary::Table * aTable, Uint64 val, bool increase)
 {
   DEBUG_TRACE("setAutoIncrementValue " << val);
   if (aTable == 0)
@@ -1188,11 +1188,6 @@ void Ndb::setDatabaseSchemaName(const char * a_schema_name)
   setSchemaName(a_schema_name);
 }
  
-void Ndb::useFullyQualifiedNames(bool turnNamingOn)
-{
-  fullyQualifiedNames = turnNamingOn;
-}
- 
 bool Ndb::usingFullyQualifiedNames()
 {
   return fullyQualifiedNames;

ndb/src/ndbapi/NdbConnection.cpp

@@ -943,7 +943,7 @@ Remark:         Get an operation from NdbOperation object idlelist and
                 object, synchronous.
 *****************************************************************************/
 NdbOperation*
-NdbConnection::getNdbOperation(NdbTableImpl * tab, NdbOperation* aNextOp)
+NdbConnection::getNdbOperation(const NdbTableImpl * tab, NdbOperation* aNextOp)
 { 
   NdbOperation* tOp;
 
@@ -989,7 +989,7 @@ NdbConnection::getNdbOperation(NdbTableImpl * tab, NdbOperation* aNextOp)
   return NULL;
 }//NdbConnection::getNdbOperation()
 
-NdbOperation* NdbConnection::getNdbOperation(NdbDictionary::Table * table)
+NdbOperation* NdbConnection::getNdbOperation(const NdbDictionary::Table * table)
 {
   if (table)
     return getNdbOperation(& NdbTableImpl::getImpl(*table));
@@ -1048,8 +1048,8 @@ NdbConnection::getNdbIndexScanOperation(const char* anIndexName,
 }
 
 NdbIndexScanOperation*
-NdbConnection::getNdbIndexScanOperation(NdbIndexImpl* index,
-					NdbTableImpl* table)
+NdbConnection::getNdbIndexScanOperation(const NdbIndexImpl* index,
+					const NdbTableImpl* table)
 {
   if (theCommitStatus == Started){
     const NdbTableImpl * indexTable = index->getIndexTable();
@@ -1070,8 +1070,8 @@ NdbConnection::getNdbIndexScanOperation(NdbIndexImpl* index,
 }//NdbConnection::getNdbIndexScanOperation()
 
 NdbIndexScanOperation* 
-NdbConnection::getNdbIndexScanOperation(NdbDictionary::Index * index,
-					NdbDictionary::Table * table)
+NdbConnection::getNdbIndexScanOperation(const NdbDictionary::Index * index,
+					const NdbDictionary::Table * table)
 {
   if (index && table)
     return getNdbIndexScanOperation(& NdbIndexImpl::getImpl(*index),
@@ -1091,7 +1091,7 @@ Remark:         Get an operation from NdbScanOperation object idlelist and get t
   	        getOperation will set the theTableId in the NdbOperation object, synchronous.
 *****************************************************************************/
 NdbIndexScanOperation*
-NdbConnection::getNdbScanOperation(NdbTableImpl * tab)
+NdbConnection::getNdbScanOperation(const NdbTableImpl * tab)
 { 
   NdbIndexScanOperation* tOp;
   
@@ -1120,7 +1120,7 @@ getNdbOp_error1:
 }//NdbConnection::getNdbScanOperation()
 
 NdbScanOperation* 
-NdbConnection::getNdbScanOperation(NdbDictionary::Table * table)
+NdbConnection::getNdbScanOperation(const NdbDictionary::Table * table)
 {
   if (table)
     return getNdbScanOperation(& NdbTableImpl::getImpl(*table));
@@ -1178,8 +1178,8 @@ Remark:         Get an operation from NdbIndexOperation object idlelist and get
   	        getOperation will set the theTableId in the NdbIndexOperation object, synchronous.
 *****************************************************************************/
 NdbIndexOperation*
-NdbConnection::getNdbIndexOperation(NdbIndexImpl * anIndex, 
-				    NdbTableImpl * aTable,
+NdbConnection::getNdbIndexOperation(const NdbIndexImpl * anIndex, 
+				    const NdbTableImpl * aTable,
                                     NdbOperation* aNextOp)
 { 
   NdbIndexOperation* tOp;
@@ -1222,8 +1222,8 @@ NdbConnection::getNdbIndexOperation(NdbIndexImpl * anIndex,
 }//NdbConnection::getNdbIndexOperation()
 
 NdbIndexOperation* 
-NdbConnection::getNdbIndexOperation(NdbDictionary::Index * index,
-				    NdbDictionary::Table * table)
+NdbConnection::getNdbIndexOperation(const NdbDictionary::Index * index,
+				    const NdbDictionary::Table * table)
 {
   if (index && table)
     return getNdbIndexOperation(& NdbIndexImpl::getImpl(*index),

ndb/src/ndbapi/NdbIndexOperation.cpp

@@ -54,8 +54,8 @@ NdbIndexOperation::~NdbIndexOperation()
  * Remark:        Initiates operation record after allocation.
  *****************************************************************************/
 int
-NdbIndexOperation::indxInit(NdbIndexImpl * anIndex,
-			    NdbTableImpl * aTable, 
+NdbIndexOperation::indxInit(const NdbIndexImpl * anIndex,
+			    const NdbTableImpl * aTable, 
 			    NdbConnection* myConnection)
 {
   NdbOperation::init(aTable, myConnection);

ndb/src/ndbapi/NdbOperation.cpp

@@ -131,7 +131,7 @@ NdbOperation::setErrorCodeAbort(int anErrorCode)
  *****************************************************************************/
 
 int
-NdbOperation::init(NdbTableImpl* tab, NdbConnection* myConnection){
+NdbOperation::init(const NdbTableImpl* tab, NdbConnection* myConnection){
   NdbApiSignal* tSignal;
   theStatus		= Init;
   theError.code		= 0;

ndb/src/ndbapi/NdbScanOperation.cpp

@@ -95,7 +95,7 @@ NdbScanOperation::setErrorCodeAbort(int aErrorCode){
  * Remark:        Initiates operation record after allocation.
  *****************************************************************************/
 int
-NdbScanOperation::init(NdbTableImpl* tab, NdbConnection* myConnection)
+NdbScanOperation::init(const NdbTableImpl* tab, NdbConnection* myConnection)
 {
   m_transConnection = myConnection;
   //NdbConnection* aScanConnection = theNdb->startTransaction(myConnection);
@@ -386,8 +386,8 @@ NdbIndexScanOperation::saveBoundATTRINFO()
     Uint32 cnt = m_accessTable->getNoOfColumns() - 1;
     m_sort_columns = cnt - i;
     for(; i<cnt; i++){
-      NdbColumnImpl* key = m_accessTable->m_index->m_columns[i];
-      NdbColumnImpl* col = m_currentTable->getColumn(key->m_keyInfoPos);
+      const NdbColumnImpl* key = m_accessTable->m_index->m_columns[i];
+      const NdbColumnImpl* col = m_currentTable->getColumn(key->m_keyInfoPos);
       NdbRecAttr* tmp = NdbScanOperation::getValue_impl(col, (char*)-1);
       UintPtr newVal = UintPtr(tmp);
       theTupleKeyDefined[i][0] = FAKE_PTR;
@@ -1169,8 +1169,8 @@ NdbIndexScanOperation::fix_get_values(){
   Uint32 cnt = m_accessTable->getNoOfColumns() - 1;
   assert(cnt <  NDB_MAX_NO_OF_ATTRIBUTES_IN_KEY);
   
-  NdbIndexImpl * idx = m_accessTable->m_index;
-  NdbTableImpl * tab = m_currentTable;
+  const NdbIndexImpl * idx = m_accessTable->m_index;
+  const NdbTableImpl * tab = m_currentTable;
   for(Uint32 i = 0; i<cnt; i++){
     Uint32 val = theTupleKeyDefined[i][0];
     switch(val){

sql/ha_ndbcluster.cc

@@ -841,7 +841,7 @@ int ha_ndbcluster::pk_read(const byte *key, uint key_len, byte *buf)
   DBUG_PRINT("enter", ("key_len: %u", key_len));
   DBUG_DUMP("key", (char*)key, key_len);
 
-  if (!(op= trans->getNdbOperation((NDBTAB *) m_table)) || 
+  if (!(op= trans->getNdbOperation((const NDBTAB *) m_table)) || 
       op->readTuple() != 0)
     ERR_RETURN(trans->getNdbError());
 
@@ -910,7 +910,7 @@ int ha_ndbcluster::complemented_pk_read(const byte *old_data, byte *new_data)
     // We have allready retrieved all fields, nothing to complement
     DBUG_RETURN(0);
 
-  if (!(op= trans->getNdbOperation((NDBTAB *) m_table)) || 
+  if (!(op= trans->getNdbOperation((const NDBTAB *) m_table)) || 
       op->readTuple() != 0)
     ERR_RETURN(trans->getNdbError());
 
@@ -964,7 +964,7 @@ int ha_ndbcluster::unique_index_read(const byte *key,
   
   if (!(op= trans->getNdbIndexOperation((NDBINDEX *) 
 					m_index[active_index].unique_index, 
-                                        (NDBTAB *) m_table)) ||
+                                        (const NDBTAB *) m_table)) ||
       op->readTuple() != 0)
     ERR_RETURN(trans->getNdbError());
   
@@ -1185,7 +1185,7 @@ int ha_ndbcluster::ordered_index_scan(const key_range *start_key,
   index_name= get_index_name(active_index);
   if (!(op= trans->getNdbIndexScanOperation((NDBINDEX *)
         				    m_index[active_index].index, 
-					    (NDBTAB *) m_table)))
+					    (const NDBTAB *) m_table)))
     ERR_RETURN(trans->getNdbError());
 
   NdbScanOperation::LockMode lm= (NdbScanOperation::LockMode)
@@ -1248,7 +1248,7 @@ int ha_ndbcluster::filtered_scan(const byte *key, uint key_len,
   DBUG_PRINT("info", ("Starting a new filtered scan on %s",
 		      m_tabname));
 
-  if (!(op= trans->getNdbScanOperation((NDBTAB *) m_table)))
+  if (!(op= trans->getNdbScanOperation((const NDBTAB *) m_table)))
     ERR_RETURN(trans->getNdbError());
   NdbScanOperation::LockMode lm= (NdbScanOperation::LockMode)
                                  get_ndb_lock_type(m_lock.type);
@@ -1319,7 +1319,7 @@ int ha_ndbcluster::full_table_scan(byte *buf)
   DBUG_ENTER("full_table_scan");  
   DBUG_PRINT("enter", ("Starting new scan on %s", m_tabname));
 
-  if (!(op=trans->getNdbScanOperation((NDBTAB *) m_table)))
+  if (!(op=trans->getNdbScanOperation((const NDBTAB *) m_table)))
     ERR_RETURN(trans->getNdbError());  
   NdbScanOperation::LockMode lm= (NdbScanOperation::LockMode)
                                  get_ndb_lock_type(m_lock.type);
@@ -1361,7 +1361,7 @@ int ha_ndbcluster::define_read_attrs(byte* buf, NdbOperation* op)
     // Scanning table with no primary key
     int hidden_no= table->fields;      
 #ifndef DBUG_OFF
-    const NDBTAB *tab= (NDBTAB *) m_table;    
+    const NDBTAB *tab= (const NDBTAB *) m_table;    
     if (!tab->getColumn(hidden_no))
       DBUG_RETURN(1);
 #endif
@@ -1394,7 +1394,7 @@ int ha_ndbcluster::write_row(byte *record)
   has_auto_increment= (table->next_number_field && record == table->record[0]);
   skip_auto_increment= table->auto_increment_field_not_null;
 
-  if (!(op= trans->getNdbOperation((NDBTAB *) m_table)))
+  if (!(op= trans->getNdbOperation((const NDBTAB *) m_table)))
     ERR_RETURN(trans->getNdbError());
 
   res= (m_use_write) ? op->writeTuple() :op->insertTuple(); 
@@ -1404,7 +1404,7 @@ int ha_ndbcluster::write_row(byte *record)
   if (table->primary_key == MAX_KEY) 
   {
     // Table has hidden primary key
-    Uint64 auto_value= m_ndb->getAutoIncrementValue((NDBTAB *) m_table);
+    Uint64 auto_value= m_ndb->getAutoIncrementValue((const NDBTAB *) m_table);
     if (set_hidden_key(op, table->fields, (const byte*)&auto_value))
       ERR_RETURN(op->getNdbError());
   } 
@@ -1475,7 +1475,7 @@ int ha_ndbcluster::write_row(byte *record)
     DBUG_PRINT("info", 
 	       ("Trying to set next auto increment value to %lu",
                 (ulong) next_val));
-    if (m_ndb->setAutoIncrementValue((NDBTAB *) m_table, next_val, true))
+    if (m_ndb->setAutoIncrementValue((const NDBTAB *) m_table, next_val, true))
       DBUG_PRINT("info", 
 		 ("Setting next auto increment value to %u", next_val));  
   }
@@ -1588,7 +1588,7 @@ int ha_ndbcluster::update_row(const byte *old_data, byte *new_data)
   }
   else
   {  
-    if (!(op= trans->getNdbOperation((NDBTAB *) m_table)) ||
+    if (!(op= trans->getNdbOperation((const NDBTAB *) m_table)) ||
 	op->updateTuple() != 0)
       ERR_RETURN(trans->getNdbError());  
     
@@ -1666,7 +1666,7 @@ int ha_ndbcluster::delete_row(const byte *record)
   else
   {
     
-    if (!(op=trans->getNdbOperation((NDBTAB *) m_table)) || 
+    if (!(op=trans->getNdbOperation((const NDBTAB *) m_table)) || 
 	op->deleteTuple() != 0)
       ERR_RETURN(trans->getNdbError());
     
@@ -1747,7 +1747,7 @@ void ha_ndbcluster::unpack_record(byte* buf)
   {
     // Table with hidden primary key
     int hidden_no= table->fields;
-    const NDBTAB *tab= (NDBTAB *) m_table;
+    const NDBTAB *tab= (const NDBTAB *) m_table;
     const NDBCOL *hidden_col= tab->getColumn(hidden_no);
     NdbRecAttr* rec= m_value[hidden_no].rec;
     DBUG_ASSERT(rec);
@@ -1765,7 +1765,7 @@ void ha_ndbcluster::unpack_record(byte* buf)
 
 void ha_ndbcluster::print_results()
 {
-  const NDBTAB *tab= (NDBTAB*) m_table;
+  const NDBTAB *tab= (const NDBTAB*) m_table;
   DBUG_ENTER("print_results");
 
 #ifndef DBUG_OFF
@@ -2190,7 +2190,7 @@ void ha_ndbcluster::position(const byte *record)
     DBUG_PRINT("info", ("Getting hidden key"));
     int hidden_no= table->fields;
     NdbRecAttr* rec= m_value[hidden_no].rec;
-    const NDBTAB *tab= (NDBTAB *) m_table;  
+    const NDBTAB *tab= (const NDBTAB *) m_table;  
     const NDBCOL *hidden_col= tab->getColumn(hidden_no);
     DBUG_ASSERT(hidden_col->getPrimaryKey() && 
                 hidden_col->getAutoIncrement() &&
@@ -2363,7 +2363,7 @@ int ha_ndbcluster::extra(enum ha_extra_function operation)
 void ha_ndbcluster::start_bulk_insert(ha_rows rows)
 {
   int bytes, batch;
-  const NDBTAB *tab= (NDBTAB *) m_table;    
+  const NDBTAB *tab= (const NDBTAB *) m_table;    
 
   DBUG_ENTER("start_bulk_insert");
   DBUG_PRINT("enter", ("rows: %d", (int)rows));
@@ -3165,8 +3165,8 @@ longlong ha_ndbcluster::get_auto_increment()
     : autoincrement_prefetch;
   Uint64 auto_value= 
     (skip_auto_increment) ? 
-    m_ndb->readAutoIncrementValue((NDBTAB *) m_table)
-    : m_ndb->getAutoIncrementValue((NDBTAB *) m_table, cache_size);
+    m_ndb->readAutoIncrementValue((const NDBTAB *) m_table)
+    : m_ndb->getAutoIncrementValue((const NDBTAB *) m_table, cache_size);
   DBUG_RETURN((longlong)auto_value);
 }