Ndb.hpp, Ndb.cpp, ha_ndbcluster.cc:

  Add a check if setting an auto_increment field will change it's next value before retrieving tuple_id_range lock. This avoids hitting locks when updating auto_increment values to a lower value than the current maximum. This is useful in loading a table with auto_increment where one loads the highest numbered pk's first and then proceeds backwards to the first. This can then be achieved with the same performance as a normal insert without auto_increment.
ndb_restore.result:
  Updated result file

sql/ha_ndbcluster.cc

@@ -2738,10 +2738,13 @@ ha_ndbcluster::set_auto_inc(Field *field)
              ("Trying to set next auto increment value to %s",
               llstr(next_val, buff)));
 #endif
-  Ndb_tuple_id_range_guard g(m_share);
-  if (ndb->setAutoIncrementValue(m_table, g.range, next_val, TRUE)
-      == -1)
-    ERR_RETURN(ndb->getNdbError());
+  if (ndb->checkUpdateAutoIncrementValue(m_share->tuple_id_range, next_val))
+  {
+    Ndb_tuple_id_range_guard g(m_share);
+    if (ndb->setAutoIncrementValue(m_table, g.range, next_val, TRUE)
+        == -1)
+      ERR_RETURN(ndb->getNdbError());
+  }
   DBUG_RETURN(0);
 }
 

storage/ndb/include/ndbapi/Ndb.hpp

@@ -1515,37 +1515,40 @@ public:
     TupleIdRange() {}
     Uint64 m_first_tuple_id;
     Uint64 m_last_tuple_id;
+    Uint64 m_highest_seen;
     void reset() {
       m_first_tuple_id = ~(Uint64)0;
       m_last_tuple_id = ~(Uint64)0;
+      m_highest_seen = 0;
     };
   };
 
   int initAutoIncrement();
 
   int getAutoIncrementValue(const char* aTableName, 
-                            Uint64 & tupleId, Uint32 cacheSize,
+                            Uint64 & autoValue, Uint32 cacheSize,
                             Uint64 step = 1, Uint64 start = 1);
   int getAutoIncrementValue(const NdbDictionary::Table * aTable, 
-                            Uint64 & tupleId, Uint32 cacheSize,
+                            Uint64 & autoValue, Uint32 cacheSize,
                             Uint64 step = 1, Uint64 start = 1);
   int getAutoIncrementValue(const NdbDictionary::Table * aTable, 
-                            TupleIdRange & range, Uint64 & tupleId,
+                            TupleIdRange & range, Uint64 & autoValue,
                             Uint32 cacheSize,
                             Uint64 step = 1, Uint64 start = 1);
   int readAutoIncrementValue(const char* aTableName,
-                             Uint64 & tupleId);
+                             Uint64 & autoValue);
   int readAutoIncrementValue(const NdbDictionary::Table * aTable,
-                             Uint64 & tupleId);
+                             Uint64 & autoValue);
   int readAutoIncrementValue(const NdbDictionary::Table * aTable,
-                             TupleIdRange & range, Uint64 & tupleId);
+                             TupleIdRange & range, Uint64 & autoValue);
   int setAutoIncrementValue(const char* aTableName,
-                            Uint64 tupleId, bool modify);
+                            Uint64 autoValue, bool modify);
   int setAutoIncrementValue(const NdbDictionary::Table * aTable,
-                            Uint64 tupleId, bool modify);
+                            Uint64 autoValue, bool modify);
   int setAutoIncrementValue(const NdbDictionary::Table * aTable,
-                            TupleIdRange & range, Uint64 tupleId,
+                            TupleIdRange & range, Uint64 autoValue,
                             bool modify);
+  bool checkUpdateAutoIncrementValue(TupleIdRange & range, Uint64 autoValue);
 private:
   int getTupleIdFromNdb(const NdbTableImpl* table,
                         TupleIdRange & range, Uint64 & tupleId,
@@ -1554,6 +1557,8 @@ private:
                          TupleIdRange & range, Uint64 & tupleId);
   int setTupleIdInNdb(const NdbTableImpl* table,
                       TupleIdRange & range, Uint64 tupleId, bool modify);
+  int checkTupleIdInNdb(TupleIdRange & range,
+                        Uint64 tupleId);
   int opTupleIdOnNdb(const NdbTableImpl* table,
                      TupleIdRange & range, Uint64 & opValue, Uint32 op);
 public:

storage/ndb/src/ndbapi/Ndb.cpp

@@ -942,6 +942,7 @@ Parameters:     aTableName (IN) : The table name.
                 step       (IN) : Specifies the step between the 
                                   autoincrement values.
                 start      (IN) : Start value for first value
+Returns:        0 if succesful, -1 if error encountered
 Remark:		Returns a new autoincrement value to the application.
                 The autoincrement values can be increased by steps
                 (default 1) and a number of values can be prefetched
@@ -1072,9 +1073,18 @@ Ndb::getTupleIdFromNdb(const NdbTableImpl* table,
   DBUG_RETURN(0);
 }
 
+/****************************************************************************
+int readAutoIncrementValue( const char* aTableName,
+                            Uint64 & autoValue);
+
+Parameters:     aTableName (IN) : The table name.
+                autoValue  (OUT) : The current autoincrement value
+Returns:        0 if succesful, -1 if error encountered
+Remark:         Returns the current autoincrement value to the application.
+****************************************************************************/
 int
 Ndb::readAutoIncrementValue(const char* aTableName,
-                            Uint64 & tupleId)
+                            Uint64 & autoValue)
 {
   DBUG_ENTER("Ndb::readAutoIncrementValue");
   ASSERT_NOT_MYSQLD;
@@ -1088,15 +1098,15 @@ Ndb::readAutoIncrementValue(const char* aTableName,
   }
   const NdbTableImpl* table = info->m_table_impl;
   TupleIdRange & range = info->m_tuple_id_range;
-  if (readTupleIdFromNdb(table, range, tupleId) == -1)
+  if (readTupleIdFromNdb(table, range, autoValue) == -1)
     DBUG_RETURN(-1);
-  DBUG_PRINT("info", ("value %lu", (ulong)tupleId));
+  DBUG_PRINT("info", ("value %lu", (ulong)autoValue));
   DBUG_RETURN(0);
 }
 
 int
 Ndb::readAutoIncrementValue(const NdbDictionary::Table * aTable,
-                            Uint64 & tupleId)
+                            Uint64 & autoValue)
 {
   DBUG_ENTER("Ndb::readAutoIncrementValue");
   ASSERT_NOT_MYSQLD;
@@ -1111,23 +1121,23 @@ Ndb::readAutoIncrementValue(const NdbDictionary::Table * aTable,
     DBUG_RETURN(-1);
   }
   TupleIdRange & range = info->m_tuple_id_range;
-  if (readTupleIdFromNdb(table, range, tupleId) == -1)
+  if (readTupleIdFromNdb(table, range, autoValue) == -1)
     DBUG_RETURN(-1);
-  DBUG_PRINT("info", ("value %lu", (ulong)tupleId));
+  DBUG_PRINT("info", ("value %lu", (ulong)autoValue));
   DBUG_RETURN(0);
 }
 
 int
 Ndb::readAutoIncrementValue(const NdbDictionary::Table * aTable,
-                            TupleIdRange & range, Uint64 & tupleId)
+                            TupleIdRange & range, Uint64 & autoValue)
 {
   DBUG_ENTER("Ndb::readAutoIncrementValue");
   assert(aTable != 0);
   const NdbTableImpl* table = & NdbTableImpl::getImpl(*aTable);
 
-  if (readTupleIdFromNdb(table, range, tupleId) == -1)
+  if (readTupleIdFromNdb(table, range, autoValue) == -1)
     DBUG_RETURN(-1);
-  DBUG_PRINT("info", ("value %lu", (ulong)tupleId));
+  DBUG_PRINT("info", ("value %lu", (ulong)autoValue));
   DBUG_RETURN(0);
 }
 
@@ -1155,9 +1165,20 @@ Ndb::readTupleIdFromNdb(const NdbTableImpl* table,
   DBUG_RETURN(0);
 }
 
+/****************************************************************************
+int setAutoIncrementValue( const char* aTableName,
+                           Uint64 autoValue,
+                           bool modify);
+
+Parameters:     aTableName (IN) : The table name.
+                autoValue  (IN) : The new autoincrement value
+                modify     (IN) : Modify existing value (not initialization)
+Returns:        0 if succesful, -1 if error encountered
+Remark:         Sets a new autoincrement value for the application.
+****************************************************************************/
 int
 Ndb::setAutoIncrementValue(const char* aTableName,
-                           Uint64 tupleId, bool increase)
+                           Uint64 autoValue, bool modify)
 {
   DBUG_ENTER("Ndb::setAutoIncrementValue");
   ASSERT_NOT_MYSQLD;
@@ -1171,14 +1192,14 @@ Ndb::setAutoIncrementValue(const char* aTableName,
   }
   const NdbTableImpl* table = info->m_table_impl;
   TupleIdRange & range = info->m_tuple_id_range;
-  if (setTupleIdInNdb(table, range, tupleId, increase) == -1)
+  if (setTupleIdInNdb(table, range, autoValue, modify) == -1)
     DBUG_RETURN(-1);
   DBUG_RETURN(0);
 }
 
 int
 Ndb::setAutoIncrementValue(const NdbDictionary::Table * aTable,
-                           Uint64 tupleId, bool increase)
+                           Uint64 autoValue, bool modify)
 {
   DBUG_ENTER("Ndb::setAutoIncrementValue");
   ASSERT_NOT_MYSQLD;
@@ -1193,52 +1214,55 @@ Ndb::setAutoIncrementValue(const NdbDictionary::Table * aTable,
     DBUG_RETURN(-1);
   }
   TupleIdRange & range = info->m_tuple_id_range;
-  if (setTupleIdInNdb(table, range, tupleId, increase) == -1)
+  if (setTupleIdInNdb(table, range, autoValue, modify) == -1)
     DBUG_RETURN(-1);
   DBUG_RETURN(0);
 }
 
 int
 Ndb::setAutoIncrementValue(const NdbDictionary::Table * aTable,
-                           TupleIdRange & range, Uint64 tupleId,
-                           bool increase)
+                           TupleIdRange & range, Uint64 autoValue,
+                           bool modify)
 {
   DBUG_ENTER("Ndb::setAutoIncrementValue");
   assert(aTable != 0);
   const NdbTableImpl* table = & NdbTableImpl::getImpl(*aTable);
 
-  if (setTupleIdInNdb(table, range, tupleId, increase) == -1)
+  if (setTupleIdInNdb(table, range, autoValue, modify) == -1)
     DBUG_RETURN(-1);
   DBUG_RETURN(0);
 }
 
 int
 Ndb::setTupleIdInNdb(const NdbTableImpl* table,
-                     TupleIdRange & range, Uint64 tupleId, bool increase)
+                     TupleIdRange & range, Uint64 tupleId, bool modify)
 {
   DBUG_ENTER("Ndb::setTupleIdInNdb");
-  if (increase)
+  if (modify)
   {
-    if (range.m_first_tuple_id != range.m_last_tuple_id)
+    if (checkTupleIdInNdb(range, tupleId))
     {
-      assert(range.m_first_tuple_id < range.m_last_tuple_id);
-      if (tupleId <= range.m_first_tuple_id + 1)
-	DBUG_RETURN(0);
-      if (tupleId <= range.m_last_tuple_id)
+      if (range.m_first_tuple_id != range.m_last_tuple_id)
       {
-	range.m_first_tuple_id = tupleId - 1;
-        DBUG_PRINT("info", 
-                   ("Setting next auto increment cached value to %lu",
-                    (ulong)tupleId));  
-	DBUG_RETURN(0);
+        assert(range.m_first_tuple_id < range.m_last_tuple_id);
+        if (tupleId <= range.m_first_tuple_id + 1)
+          DBUG_RETURN(0);
+        if (tupleId <= range.m_last_tuple_id)
+        {
+          range.m_first_tuple_id = tupleId - 1;
+          DBUG_PRINT("info", 
+                     ("Setting next auto increment cached value to %lu",
+                      (ulong)tupleId));  
+          DBUG_RETURN(0);
+        }
       }
+      /*
+       * if tupleId <= NEXTID, do nothing.  otherwise update NEXTID to
+       * tupleId and set cached range to first = last = tupleId - 1.
+       */
+      if (opTupleIdOnNdb(table, range, tupleId, 2) == -1)
+        DBUG_RETURN(-1);
     }
-    /*
-     * if tupleId <= NEXTID, do nothing.  otherwise update NEXTID to
-     * tupleId and set cached range to first = last = tupleId - 1.
-     */
-    if (opTupleIdOnNdb(table, range, tupleId, 2) == -1)
-      DBUG_RETURN(-1);
   }
   else
   {
@@ -1277,6 +1301,39 @@ int Ndb::initAutoIncrement()
   return 0;
 }
 
+bool
+Ndb::checkUpdateAutoIncrementValue(TupleIdRange & range, Uint64 autoValue)
+{
+  return(checkTupleIdInNdb(range, autoValue) != 0);
+}
+
+int
+Ndb::checkTupleIdInNdb(TupleIdRange & range, Uint64 tupleId)
+{
+  DBUG_ENTER("Ndb::checkTupleIdIndNdb");
+  if ((range.m_first_tuple_id != ~(Uint64)0) &&
+      (range.m_first_tuple_id > tupleId))
+  {
+   /*
+    * If we have ever cached a value in this object and this cached
+    * value is larger than the value we're trying to set then we
+    * need not check with the real value in the SYSTAB_0 table.
+    */
+    DBUG_RETURN(0);
+  }
+  if (range.m_highest_seen > tupleId)
+  {
+    /*
+     * Although we've never cached any higher value we have read
+     * a higher value and again it isn't necessary to change the
+     * auto increment value.
+     */
+    DBUG_RETURN(0);
+  }
+  DBUG_RETURN(1);
+}
+
+
 int
 Ndb::opTupleIdOnNdb(const NdbTableImpl* table,
                     TupleIdRange & range, Uint64 & opValue, Uint32 op)