Fix for bug#3912 Auto increment not correctly initialised when table is...

Fix for bug#3912 Auto increment not correctly initialised when table is altered, completes WL#1911 Extended AUTO_INCREMENT support in NDB

ndb/include/ndbapi/Ndb.hpp

@@ -1415,11 +1415,11 @@ public:
    * @return tuple id or 0 on error
    */
   Uint64 getAutoIncrementValue(const char* aTableName, Uint32 cacheSize = 1);
-  bool setAutoIncrementValue(const char* aTableName, Uint64 val);
+  bool setAutoIncrementValue(const char* aTableName, Uint64 val, bool increase = false);
   Uint64 getTupleIdFromNdb(const char* aTableName, Uint32 cacheSize = 1000 );
   Uint64 getTupleIdFromNdb(Uint32 aTableId, Uint32 cacheSize = 1000 );
-  bool setTupleIdInNdb(const char* aTableName, Uint64 val);
-  bool setTupleIdInNdb(Uint32 aTableId, Uint64 val);
+  bool setTupleIdInNdb(const char* aTableName, Uint64 val, bool increase = false);
+  bool setTupleIdInNdb(Uint32 aTableId, Uint64 val, bool increase = false);
   Uint64 opTupleIdOnNdb(Uint32 aTableId, Uint64 opValue, Uint32 op);
 #endif
 

ndb/src/ndbapi/Ndb.cpp

@@ -759,30 +759,47 @@ Ndb::getTupleIdFromNdb(Uint32 aTableId, Uint32 cacheSize )
 }
 
 bool
-Ndb::setAutoIncrementValue(const char* aTableName, Uint64 val)
+Ndb::setAutoIncrementValue(const char* aTableName, Uint64 val, bool increase)
 {
   DEBUG_TRACE("setAutoIncrementValue " << val);
   const NdbTableImpl* table = theDictionary->getTable(aTableName);
   if (table == 0)
     return false;
-  return setTupleIdInNdb(table->m_tableId, val);
+  return setTupleIdInNdb(table->m_tableId, val, increase);
 }
 
 bool 
-Ndb::setTupleIdInNdb(const char* aTableName, Uint64 val )
+Ndb::setTupleIdInNdb(const char* aTableName, Uint64 val, bool increase )
 {
   DEBUG_TRACE("setTupleIdInNdb");
   const NdbTableImpl* table = theDictionary->getTable(aTableName);
   if (table == 0)
     return false;
-  return setTupleIdInNdb(table->m_tableId, val);
+  return setTupleIdInNdb(table->m_tableId, val, increase);
 }
 
 bool
-Ndb::setTupleIdInNdb(Uint32 aTableId, Uint64 val )
+Ndb::setTupleIdInNdb(Uint32 aTableId, Uint64 val, bool increase )
 {
   DEBUG_TRACE("setTupleIdInNdb");
-  return (opTupleIdOnNdb(aTableId, val, 1) == val);
+  if (increase)
+  {
+    if (theFirstTupleId[aTableId] != theLastTupleId[aTableId])
+    {
+      // We have a cache sequence
+      if (val <= theFirstTupleId[aTableId]+1)
+	return true;
+      if (val <= theLastTupleId[aTableId])
+      {
+	theFirstTupleId[aTableId] = val - 1;
+	return true;
+      }
+      // else continue;
+    }    
+    return (opTupleIdOnNdb(aTableId, val, 2) == val);
+  }
+  else
+    return (opTupleIdOnNdb(aTableId, val, 1) == val);
 }
 
 Uint64
@@ -845,6 +862,23 @@ Ndb::opTupleIdOnNdb(Uint32 aTableId, Uint64 opValue, Uint32 op)
       tOperation->equal("SYSKEY_0", aTableId );
       tOperation->setValue("NEXTID", opValue);
 
+      if (tConnection->execute( Commit ) == -1 )
+        goto error_handler;
+
+      theFirstTupleId[aTableId] = ~0;
+      theLastTupleId[aTableId]  = ~0;
+      ret = opValue;
+      break;
+    case 2:
+      tOperation->interpretedUpdateTuple();
+      tOperation->equal("SYSKEY_0", aTableId );
+      tOperation->load_const_u64(1, opValue);
+      tOperation->read_attr("NEXTID", 2);
+      tOperation->branch_le(2, 1, 0);
+      tOperation->write_attr("NEXTID", 1);
+      tOperation->def_label(0);
+      tOperation->interpret_exit_ok();
+      
       if (tConnection->execute( Commit ) == -1 )
         goto error_handler;
 

sql/ha_ndbcluster.cc

@@ -1281,6 +1281,7 @@ int ha_ndbcluster::define_read_attrs(byte* buf, NdbOperation* op)
 
 int ha_ndbcluster::write_row(byte *record)
 {
+  bool has_auto_increment;
   uint i;
   NdbConnection *trans= m_active_trans;
   NdbOperation *op;
@@ -1290,7 +1291,8 @@ int ha_ndbcluster::write_row(byte *record)
   statistic_increment(ha_write_count,&LOCK_status);
   if (table->timestamp_default_now)
     update_timestamp(record+table->timestamp_default_now-1);
-  if (table->next_number_field && record == table->record[0])
+  has_auto_increment= (table->next_number_field && record == table->record[0]);
+  if (has_auto_increment)
     update_auto_increment();
 
   if (!(op= trans->getNdbOperation(m_tabname)))
@@ -1344,6 +1346,13 @@ int ha_ndbcluster::write_row(byte *record)
     if (trans->execute(NoCommit) != 0)
       DBUG_RETURN(ndb_err(trans));
   }
+  if ( (has_auto_increment) && (!auto_increment_column_changed) )
+  {    
+    Uint64 next_val= (Uint64) table->next_number_field->val_int() + 1;
+    DBUG_PRINT("info", ("Setting next auto increment value to %u", next_val));
+    m_ndb->setAutoIncrementValue(m_tabname, next_val, true);
+  }
+
   DBUG_RETURN(0);
 }