Bug#26342 auto_increment_increment AND auto_increment_offset REALLY REALLY anger NDB cluster, implemented support for auto_increment_offset and auto_increment_increment for Ndb

7 files changed, 402 insertions, 37 deletions

diff --git a/mysql-test/r/ndb_insert.result b/mysql-test/r/ndb_insert.result
index e7275bde2b8..b5349ecb59c 100644
--- a/mysql-test/r/ndb_insert.result
+++ b/mysql-test/r/ndb_insert.result
@@ -657,3 +657,172 @@ a	b
 2	NULL
 3	NULL
 drop table t1;
+CREATE TABLE t1 (
+pk INT NOT NULL PRIMARY KEY AUTO_INCREMENT,
+b INT NOT NULL,
+c INT NOT NULL UNIQUE
+) ENGINE=NDBCLUSTER;
+CREATE TABLE t2 (
+pk INT NOT NULL PRIMARY KEY AUTO_INCREMENT,
+b INT NOT NULL,
+c INT NOT NULL UNIQUE
+) ENGINE=MYISAM;
+SET @@session.auto_increment_increment=10;
+INSERT INTO t1 (b,c) VALUES (1,0),(2,1),(3,2);
+INSERT INTO t2 (b,c) VALUES (1,0),(2,1),(3,2);
+SELECT * FROM t1 ORDER BY pk;
+pk	b	c
+1	1	0
+11	2	1
+21	3	2
+SELECT COUNT(t1.pk) FROM t1, t2 WHERE t1.pk = t2.pk AND t1.b = t2.b AND t1.c = t1.c;
+COUNT(t1.pk)
+3
+TRUNCATE t1;
+TRUNCATE t2;
+SET @@session.auto_increment_offset=5;
+INSERT INTO t1 (b,c) VALUES (1,0),(2,1),(3,2);
+INSERT INTO t1 (pk,b,c) VALUES (27,4,3),(NULL,5,4),(99,6,5),(NULL,7,6);
+INSERT INTO t2 (b,c) VALUES (1,0),(2,1),(3,2);
+INSERT INTO t2 (pk,b,c) VALUES (27,4,3),(NULL,5,4),(99,6,5),(NULL,7,6);
+SELECT * FROM t1 ORDER BY pk;
+pk	b	c
+5	1	0
+15	2	1
+25	3	2
+27	4	3
+35	5	4
+99	6	5
+105	7	6
+SELECT COUNT(t1.pk) FROM t1, t2 WHERE t1.pk = t2.pk AND t1.b = t2.b AND t1.c = t1.c;
+COUNT(t1.pk)
+7
+TRUNCATE t1;
+TRUNCATE t2;
+SET @@session.auto_increment_increment=2;
+INSERT INTO t1 (b,c) VALUES (1,0),(2,1),(3,2);
+INSERT INTO t2 (b,c) VALUES (1,0),(2,1),(3,2);
+SELECT * FROM t1 ORDER BY pk;
+pk	b	c
+1	1	0
+3	2	1
+5	3	2
+SELECT COUNT(t1.pk) FROM t1, t2 WHERE t1.pk = t2.pk AND t1.b = t2.b AND t1.c = t1.c;
+COUNT(t1.pk)
+3
+DROP TABLE t1, t2;
+CREATE TABLE t1 (
+pk INT NOT NULL PRIMARY KEY AUTO_INCREMENT,
+b INT NOT NULL,
+c INT NOT NULL UNIQUE
+) ENGINE=NDBCLUSTER AUTO_INCREMENT = 7;
+CREATE TABLE t2 (
+pk INT NOT NULL PRIMARY KEY AUTO_INCREMENT,
+b INT NOT NULL,
+c INT NOT NULL UNIQUE
+) ENGINE=MYISAM AUTO_INCREMENT = 7;
+SET @@session.auto_increment_offset=1;
+SET @@session.auto_increment_increment=1;
+INSERT INTO t1 (b,c) VALUES (1,0),(2,1),(3,2);
+INSERT INTO t2 (b,c) VALUES (1,0),(2,1),(3,2);
+SELECT * FROM t1 ORDER BY pk;
+pk	b	c
+7	1	0
+8	2	1
+9	3	2
+SELECT COUNT(t1.pk) FROM t1, t2 WHERE t1.pk = t2.pk AND t1.b = t2.b AND t1.c = t1.c;
+COUNT(t1.pk)
+3
+DROP TABLE t1, t2;
+CREATE TABLE t1 (
+pk INT NOT NULL PRIMARY KEY AUTO_INCREMENT,
+b INT NOT NULL,
+c INT NOT NULL UNIQUE
+) ENGINE=NDBCLUSTER AUTO_INCREMENT = 3;
+CREATE TABLE t2 (
+pk INT NOT NULL PRIMARY KEY AUTO_INCREMENT,
+b INT NOT NULL,
+c INT NOT NULL UNIQUE
+) ENGINE=MYISAM AUTO_INCREMENT = 3;
+SET @@session.auto_increment_offset=5;
+SET @@session.auto_increment_increment=10;
+INSERT INTO t1 (b,c) VALUES (1,0),(2,1),(3,2);
+INSERT INTO t2 (b,c) VALUES (1,0),(2,1),(3,2);
+SELECT * FROM t1 ORDER BY pk;
+pk	b	c
+5	1	0
+15	2	1
+25	3	2
+SELECT COUNT(t1.pk) FROM t1, t2 WHERE t1.pk = t2.pk AND t1.b = t2.b AND t1.c = t1.c;
+COUNT(t1.pk)
+3
+DROP TABLE t1, t2;
+CREATE TABLE t1 (
+pk INT NOT NULL PRIMARY KEY AUTO_INCREMENT,
+b INT NOT NULL,
+c INT NOT NULL UNIQUE
+) ENGINE=NDBCLUSTER AUTO_INCREMENT = 7;
+CREATE TABLE t2 (
+pk INT NOT NULL PRIMARY KEY AUTO_INCREMENT,
+b INT NOT NULL,
+c INT NOT NULL UNIQUE
+) ENGINE=MYISAM AUTO_INCREMENT = 7;
+SET @@session.auto_increment_offset=5;
+SET @@session.auto_increment_increment=10;
+INSERT INTO t1 (b,c) VALUES (1,0),(2,1),(3,2);
+INSERT INTO t2 (b,c) VALUES (1,0),(2,1),(3,2);
+SELECT * FROM t1 ORDER BY pk;
+pk	b	c
+15	1	0
+25	2	1
+35	3	2
+SELECT COUNT(t1.pk) FROM t1, t2 WHERE t1.pk = t2.pk AND t1.b = t2.b AND t1.c = t1.c;
+COUNT(t1.pk)
+3
+DROP TABLE t1, t2;
+CREATE TABLE t1 (
+pk INT NOT NULL PRIMARY KEY AUTO_INCREMENT,
+b INT NOT NULL,
+c INT NOT NULL UNIQUE
+) ENGINE=NDBCLUSTER AUTO_INCREMENT = 5;
+CREATE TABLE t2 (
+pk INT NOT NULL PRIMARY KEY AUTO_INCREMENT,
+b INT NOT NULL,
+c INT NOT NULL UNIQUE
+) ENGINE=MYISAM AUTO_INCREMENT = 5;
+SET @@session.auto_increment_offset=5;
+SET @@session.auto_increment_increment=10;
+INSERT INTO t1 (b,c) VALUES (1,0),(2,1),(3,2);
+INSERT INTO t2 (b,c) VALUES (1,0),(2,1),(3,2);
+SELECT * FROM t1 ORDER BY pk;
+pk	b	c
+5	1	0
+15	2	1
+25	3	2
+SELECT COUNT(t1.pk) FROM t1, t2 WHERE t1.pk = t2.pk AND t1.b = t2.b AND t1.c = t1.c;
+COUNT(t1.pk)
+3
+DROP TABLE t1, t2;
+CREATE TABLE t1 (
+pk INT NOT NULL PRIMARY KEY AUTO_INCREMENT,
+b INT NOT NULL,
+c INT NOT NULL UNIQUE
+) ENGINE=NDBCLUSTER AUTO_INCREMENT = 100;
+CREATE TABLE t2 (
+pk INT NOT NULL PRIMARY KEY AUTO_INCREMENT,
+b INT NOT NULL,
+c INT NOT NULL UNIQUE
+) ENGINE=MYISAM AUTO_INCREMENT = 100;
+SET @@session.auto_increment_offset=5;
+SET @@session.auto_increment_increment=10;
+INSERT INTO t1 (b,c) VALUES (1,0),(2,1),(3,2);
+INSERT INTO t2 (b,c) VALUES (1,0),(2,1),(3,2);
+SELECT * FROM t1 ORDER BY pk;
+pk	b	c
+105	1	0
+115	2	1
+125	3	2
+SELECT COUNT(t1.pk) FROM t1, t2 WHERE t1.pk = t2.pk AND t1.b = t2.b AND t1.c = t1.c;
+COUNT(t1.pk)
+3
+DROP TABLE t1, t2;
diff --git a/mysql-test/t/ndb_insert.test b/mysql-test/t/ndb_insert.test
index f346b7dc4ab..b8f00d6f6aa 100644
--- a/mysql-test/t/ndb_insert.test
+++ b/mysql-test/t/ndb_insert.test
@@ -639,4 +639,141 @@ insert ignore into t1 values (1,0), (2,0), (2,null), (3,null);
 select * from t1 order by a;
 drop table t1;
 
+# Bug#26342 auto_increment_increment AND auto_increment_offset REALLY REALLY anger NDB cluster
+
+CREATE TABLE t1 (
+  pk INT NOT NULL PRIMARY KEY AUTO_INCREMENT,
+  b INT NOT NULL,
+  c INT NOT NULL UNIQUE
+) ENGINE=NDBCLUSTER;
+
+CREATE TABLE t2 (
+  pk INT NOT NULL PRIMARY KEY AUTO_INCREMENT,
+  b INT NOT NULL,
+  c INT NOT NULL UNIQUE
+) ENGINE=MYISAM;
+
+SET @@session.auto_increment_increment=10;
+INSERT INTO t1 (b,c) VALUES (1,0),(2,1),(3,2);
+INSERT INTO t2 (b,c) VALUES (1,0),(2,1),(3,2);
+SELECT * FROM t1 ORDER BY pk;
+SELECT COUNT(t1.pk) FROM t1, t2 WHERE t1.pk = t2.pk AND t1.b = t2.b AND t1.c = t1.c;
+TRUNCATE t1; 
+TRUNCATE t2;
+SET @@session.auto_increment_offset=5;
+INSERT INTO t1 (b,c) VALUES (1,0),(2,1),(3,2);
+INSERT INTO t1 (pk,b,c) VALUES (27,4,3),(NULL,5,4),(99,6,5),(NULL,7,6);
+INSERT INTO t2 (b,c) VALUES (1,0),(2,1),(3,2);
+INSERT INTO t2 (pk,b,c) VALUES (27,4,3),(NULL,5,4),(99,6,5),(NULL,7,6);
+SELECT * FROM t1 ORDER BY pk;
+SELECT COUNT(t1.pk) FROM t1, t2 WHERE t1.pk = t2.pk AND t1.b = t2.b AND t1.c = t1.c;
+TRUNCATE t1; 
+TRUNCATE t2;
+SET @@session.auto_increment_increment=2;
+INSERT INTO t1 (b,c) VALUES (1,0),(2,1),(3,2);
+INSERT INTO t2 (b,c) VALUES (1,0),(2,1),(3,2);
+SELECT * FROM t1 ORDER BY pk;
+SELECT COUNT(t1.pk) FROM t1, t2 WHERE t1.pk = t2.pk AND t1.b = t2.b AND t1.c = t1.c;
+DROP TABLE t1, t2;
+
+CREATE TABLE t1 (
+  pk INT NOT NULL PRIMARY KEY AUTO_INCREMENT,
+  b INT NOT NULL,
+  c INT NOT NULL UNIQUE
+) ENGINE=NDBCLUSTER AUTO_INCREMENT = 7;
+
+CREATE TABLE t2 (
+  pk INT NOT NULL PRIMARY KEY AUTO_INCREMENT,
+  b INT NOT NULL,
+  c INT NOT NULL UNIQUE
+) ENGINE=MYISAM AUTO_INCREMENT = 7;
+
+SET @@session.auto_increment_offset=1;
+SET @@session.auto_increment_increment=1;
+INSERT INTO t1 (b,c) VALUES (1,0),(2,1),(3,2);
+INSERT INTO t2 (b,c) VALUES (1,0),(2,1),(3,2);
+SELECT * FROM t1 ORDER BY pk;
+SELECT COUNT(t1.pk) FROM t1, t2 WHERE t1.pk = t2.pk AND t1.b = t2.b AND t1.c = t1.c;
+DROP TABLE t1, t2;
+
+CREATE TABLE t1 (
+  pk INT NOT NULL PRIMARY KEY AUTO_INCREMENT,
+  b INT NOT NULL,
+  c INT NOT NULL UNIQUE
+) ENGINE=NDBCLUSTER AUTO_INCREMENT = 3;
+
+CREATE TABLE t2 (
+  pk INT NOT NULL PRIMARY KEY AUTO_INCREMENT,
+  b INT NOT NULL,
+  c INT NOT NULL UNIQUE
+) ENGINE=MYISAM AUTO_INCREMENT = 3;
+
+SET @@session.auto_increment_offset=5;
+SET @@session.auto_increment_increment=10;
+INSERT INTO t1 (b,c) VALUES (1,0),(2,1),(3,2);
+INSERT INTO t2 (b,c) VALUES (1,0),(2,1),(3,2);
+SELECT * FROM t1 ORDER BY pk;
+SELECT COUNT(t1.pk) FROM t1, t2 WHERE t1.pk = t2.pk AND t1.b = t2.b AND t1.c = t1.c;
+DROP TABLE t1, t2;
+
+CREATE TABLE t1 (
+  pk INT NOT NULL PRIMARY KEY AUTO_INCREMENT,
+  b INT NOT NULL,
+  c INT NOT NULL UNIQUE
+) ENGINE=NDBCLUSTER AUTO_INCREMENT = 7;
+
+CREATE TABLE t2 (
+  pk INT NOT NULL PRIMARY KEY AUTO_INCREMENT,
+  b INT NOT NULL,
+  c INT NOT NULL UNIQUE
+) ENGINE=MYISAM AUTO_INCREMENT = 7;
+
+SET @@session.auto_increment_offset=5;
+SET @@session.auto_increment_increment=10;
+INSERT INTO t1 (b,c) VALUES (1,0),(2,1),(3,2);
+INSERT INTO t2 (b,c) VALUES (1,0),(2,1),(3,2);
+SELECT * FROM t1 ORDER BY pk;
+SELECT COUNT(t1.pk) FROM t1, t2 WHERE t1.pk = t2.pk AND t1.b = t2.b AND t1.c = t1.c;
+DROP TABLE t1, t2;
+
+CREATE TABLE t1 (
+  pk INT NOT NULL PRIMARY KEY AUTO_INCREMENT,
+  b INT NOT NULL,
+  c INT NOT NULL UNIQUE
+) ENGINE=NDBCLUSTER AUTO_INCREMENT = 5;
+
+CREATE TABLE t2 (
+  pk INT NOT NULL PRIMARY KEY AUTO_INCREMENT,
+  b INT NOT NULL,
+  c INT NOT NULL UNIQUE
+) ENGINE=MYISAM AUTO_INCREMENT = 5;
+
+SET @@session.auto_increment_offset=5;
+SET @@session.auto_increment_increment=10;
+INSERT INTO t1 (b,c) VALUES (1,0),(2,1),(3,2);
+INSERT INTO t2 (b,c) VALUES (1,0),(2,1),(3,2);
+SELECT * FROM t1 ORDER BY pk;
+SELECT COUNT(t1.pk) FROM t1, t2 WHERE t1.pk = t2.pk AND t1.b = t2.b AND t1.c = t1.c;
+DROP TABLE t1, t2;
+
+CREATE TABLE t1 (
+  pk INT NOT NULL PRIMARY KEY AUTO_INCREMENT,
+  b INT NOT NULL,
+  c INT NOT NULL UNIQUE
+) ENGINE=NDBCLUSTER AUTO_INCREMENT = 100;
+
+CREATE TABLE t2 (
+  pk INT NOT NULL PRIMARY KEY AUTO_INCREMENT,
+  b INT NOT NULL,
+  c INT NOT NULL UNIQUE
+) ENGINE=MYISAM AUTO_INCREMENT = 100;
+
+SET @@session.auto_increment_offset=5;
+SET @@session.auto_increment_increment=10;
+INSERT INTO t1 (b,c) VALUES (1,0),(2,1),(3,2);
+INSERT INTO t2 (b,c) VALUES (1,0),(2,1),(3,2);
+SELECT * FROM t1 ORDER BY pk;
+SELECT COUNT(t1.pk) FROM t1, t2 WHERE t1.pk = t2.pk AND t1.b = t2.b AND t1.c = t1.c;
+DROP TABLE t1, t2;
+
 # End of 4.1 tests
diff --git a/ndb/include/ndbapi/Ndb.hpp b/ndb/include/ndbapi/Ndb.hpp
index b3c9acd4e20..f83db77739e 100644
--- a/ndb/include/ndbapi/Ndb.hpp
+++ b/ndb/include/ndbapi/Ndb.hpp
@@ -1388,9 +1388,11 @@ public:
    * @return 0 or -1 on error, and tupleId in out parameter
    */
   int getAutoIncrementValue(const char* aTableName, 
-                            Uint64 & tupleId, Uint32 cacheSize);
+                            Uint64 & tupleId, Uint32 cacheSize, 
+			    Uint64 step = 1, Uint64 start = 1);
   int getAutoIncrementValue(const NdbDictionary::Table * aTable, 
-                            Uint64 & tupleId, Uint32 cacheSize);
+                            Uint64 & tupleId, Uint32 cacheSize, 
+			    Uint64 step = 1, Uint64 start = 1);
   int readAutoIncrementValue(const char* aTableName,
                              Uint64 & tupleId);
   int readAutoIncrementValue(const NdbDictionary::Table * aTable,
@@ -1401,7 +1403,8 @@ public:
                             Uint64 tupleId, bool increase);
 private:
   int getTupleIdFromNdb(Ndb_local_table_info* info,
-                        Uint64 & tupleId, Uint32 cacheSize);
+                        Uint64 & tupleId, Uint32 cacheSize,
+			Uint64 step = 1, Uint64 start = 1 );
   int readTupleIdFromNdb(Ndb_local_table_info* info,
                          Uint64 & tupleId);
   int setTupleIdInNdb(Ndb_local_table_info* info,
diff --git a/ndb/src/ndbapi/Ndb.cpp b/ndb/src/ndbapi/Ndb.cpp
index 449f287dc1d..941bfc88b24 100644
--- a/ndb/src/ndbapi/Ndb.cpp
+++ b/ndb/src/ndbapi/Ndb.cpp
@@ -767,17 +767,27 @@ Ndb::getNodeId()
 }
 
 /****************************************************************************
-Uint64 getTupleIdFromNdb( Uint32 aTableId, Uint32 cacheSize );
-
-Parameters:     aTableId : The TableId.
-                cacheSize: Prefetch this many values
-Remark:		Returns a new TupleId to the application.
-                The TupleId comes from SYSTAB_0 where SYSKEY_0 = TableId.
-                It is initialized to (TableId << 48) + 1 in NdbcntrMain.cpp.
+Uint64 getAutoIncrementValue( const char* aTableName,
+                              Uint64 & tupleId, 
+                              Uint32 cacheSize, 
+                              Uint64 step,
+                              Uint64 start);
+
+Parameters:     aTableName (IN) : The table name.
+                autoValue (OUT) : Returns new autoincrement value
+                cacheSize  (IN) : Prefetch this many values
+                step       (IN) : Specifies the step between the 
+                                  autoincrement values.
+                start      (IN) : Start value for first value
+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
+                by specifying cacheSize (default 10).
 ****************************************************************************/
 int
 Ndb::getAutoIncrementValue(const char* aTableName,
-                           Uint64 & tupleId, Uint32 cacheSize)
+                           Uint64 & autoValue, Uint32 cacheSize, 
+			   Uint64 step, Uint64 start)
 {
   DBUG_ENTER("Ndb::getAutoIncrementValue");
   BaseString internal_tabname(internalize_table_name(aTableName));
@@ -788,15 +798,17 @@ Ndb::getAutoIncrementValue(const char* aTableName,
     theError.code = theDictionary->getNdbError().code;
     DBUG_RETURN(-1);
   }
-  if (getTupleIdFromNdb(info, tupleId, cacheSize) == -1)
+  DBUG_PRINT("info", ("step %lu", (ulong) step));
+  if (getTupleIdFromNdb(info, autoValue, cacheSize, step, start) == -1)
     DBUG_RETURN(-1);
-  DBUG_PRINT("info", ("value %lu", (ulong) tupleId));
+  DBUG_PRINT("info", ("value %lu", (ulong) autoValue));
   DBUG_RETURN(0);
 }
 
 int
 Ndb::getAutoIncrementValue(const NdbDictionary::Table * aTable,
-                           Uint64 & tupleId, Uint32 cacheSize)
+                           Uint64 & autoValue, Uint32 cacheSize, 
+			   Uint64 step, Uint64 start)
 {
   DBUG_ENTER("Ndb::getAutoIncrementValue");
   assert(aTable != 0);
@@ -809,36 +821,73 @@ Ndb::getAutoIncrementValue(const NdbDictionary::Table * aTable,
     theError.code = theDictionary->getNdbError().code;
     DBUG_RETURN(-1);
   }
-  if (getTupleIdFromNdb(info, tupleId, cacheSize) == -1)
+  DBUG_PRINT("info", ("step %lu", (ulong) step));
+  if (getTupleIdFromNdb(info, autoValue, cacheSize, step, start) == -1)
     DBUG_RETURN(-1);
-  DBUG_PRINT("info", ("value %lu", (ulong)tupleId));
+  DBUG_PRINT("info", ("value %lu", (ulong) autoValue));
   DBUG_RETURN(0);
 }
 
 int
 Ndb::getTupleIdFromNdb(Ndb_local_table_info* info,
-                       Uint64 & tupleId, Uint32 cacheSize)
+                       Uint64 & tupleId, Uint32 cacheSize, 
+		       Uint64 step, Uint64 start)
 {
+/*
+  Returns a new TupleId to the application.
+  The TupleId comes from SYSTAB_0 where SYSKEY_0 = TableId.
+  It is initialized to (TableId << 48) + 1 in NdbcntrMain.cpp.
+  In most cases step= start= 1, in which case we get:
+  1,2,3,4,5,...
+  If step=10 and start=5 and first number is 1, we get:
+  5,15,25,35,...  
+*/
   DBUG_ENTER("Ndb::getTupleIdFromNdb");
-  if (info->m_first_tuple_id != info->m_last_tuple_id)
+  DBUG_PRINT("info", ("Step %lu (%lu,%lu)", (ulong) step, (ulong) info->m_first_tuple_id, (ulong) info->m_last_tuple_id));
+  /*
+   Check if the next value can be taken from the pre-fetched
+   sequence.
+  */
+  if (info->m_first_tuple_id != info->m_last_tuple_id &&
+      info->m_first_tuple_id + step <= info->m_last_tuple_id)
   {
     assert(info->m_first_tuple_id < info->m_last_tuple_id);
-    tupleId = ++info->m_first_tuple_id;
-    DBUG_PRINT("info", ("next cached value %lu", (ulong)tupleId));
+    info->m_first_tuple_id += step; 
+    tupleId = info->m_first_tuple_id;
+    DBUG_PRINT("info", ("Next cached value %lu", (ulong) tupleId));
   }
   else
   {
+     /*
+       If start value is greater than step it is ignored
+      */
+      Uint64 offset = (start > step) ? 1 : start;
+
+    /*
+      Pre-fetch a number of values depending on cacheSize
+     */
     if (cacheSize == 0)
       cacheSize = 1;
-    DBUG_PRINT("info", ("reading %u values from database", (uint)cacheSize));
+
+    DBUG_PRINT("info", ("Reading %u values from database", (uint)cacheSize));
     /*
      * reserve next cacheSize entries in db.  adds cacheSize to NEXTID
-     * and returns first tupleId in the new range.
+     * and returns first tupleId in the new range. If tupleId's are
+     * incremented in steps then multiply the cacheSize with step size.
      */
-    Uint64 opValue = cacheSize;
+    Uint64 opValue = cacheSize * step;
+
     if (opTupleIdOnNdb(info, opValue, 0) == -1)
       DBUG_RETURN(-1);
-    tupleId = opValue;
+    DBUG_PRINT("info", ("Next value fetched from database %lu", (ulong) opValue));
+    DBUG_PRINT("info", ("Increasing %lu by offset %lu, increment  is %lu", (ulong) (ulong) opValue, (ulong) offset, (ulong) step));
+    Uint64 current, next;
+    next = ((Uint64) (opValue + step - offset)) / step;
+    next = next * step + offset;
+    current = (next < step) ? next : next - step;
+    tupleId = (opValue <= current) ? current : next;
+    DBUG_PRINT("info", ("Returning %lu", (ulong) tupleId));
+    info->m_first_tuple_id = tupleId;
   }
   DBUG_RETURN(0);
 }
@@ -1065,9 +1114,9 @@ Ndb::opTupleIdOnNdb(Ndb_local_table_info* info, Uint64 & opValue, Uint32 op)
       }
       else
       {
-        DBUG_PRINT("info", 
-                   ("Setting next auto increment value (db) to %lu",
-                    (ulong)opValue));  
+	DBUG_PRINT("info", 
+		   ("Setting next auto increment value (db) to %lu",
+		    (ulong)opValue));  
         info->m_first_tuple_id = info->m_last_tuple_id = opValue - 1;
       }
       break;
diff --git a/sql/ha_ndbcluster.cc b/sql/ha_ndbcluster.cc
index 9b48e0d4f38..175a81e226c 100644
--- a/sql/ha_ndbcluster.cc
+++ b/sql/ha_ndbcluster.cc
@@ -2264,12 +2264,10 @@ int ha_ndbcluster::write_row(byte *record)
     if (has_auto_increment) 
     {
       int error;
-      
+
       m_skip_auto_increment= FALSE;
       if ((error= update_auto_increment()))
         DBUG_RETURN(error);
-      /* Ensure that handler is always called for auto_increment values */
-      thd->next_insert_id= 0;
       m_skip_auto_increment= !auto_increment_column_changed;
     }
   }
@@ -2312,8 +2310,10 @@ int ha_ndbcluster::write_row(byte *record)
     int ret;
     Uint64 auto_value;
     uint retries= NDB_AUTO_INCREMENT_RETRIES;
+
     do {
-      ret= ndb->getAutoIncrementValue((const NDBTAB *) m_table, auto_value, 1);
+      ret= ndb->getAutoIncrementValue((const NDBTAB *) m_table, 
+                                      auto_value, 1);
     } while (ret == -1 && 
              --retries &&
              ndb->getNdbError().status == NdbError::TemporaryError);
@@ -2322,7 +2322,7 @@ int ha_ndbcluster::write_row(byte *record)
     if (set_hidden_key(op, table->s->fields, (const byte*)&auto_value))
       ERR_RETURN(op->getNdbError());
   } 
-  else 
+  else
   {
     if ((res= set_primary_key_from_record(op, record)))
       return res;  
@@ -4841,6 +4841,8 @@ ulonglong ha_ndbcluster::get_auto_increment()
 {  
   int cache_size;
   Uint64 auto_value;
+  Uint64 step= current_thd->variables.auto_increment_increment;
+  Uint64 start= current_thd->variables.auto_increment_offset;
   DBUG_ENTER("get_auto_increment");
   DBUG_PRINT("enter", ("m_tabname: %s", m_tabname));
   Ndb *ndb= get_ndb();
@@ -4861,7 +4863,8 @@ ulonglong ha_ndbcluster::get_auto_increment()
     ret=
       m_skip_auto_increment ? 
       ndb->readAutoIncrementValue((const NDBTAB *) m_table, auto_value) :
-      ndb->getAutoIncrementValue((const NDBTAB *) m_table, auto_value, cache_size);
+      ndb->getAutoIncrementValue((const NDBTAB *) m_table, 
+				 auto_value, cache_size, step, start);
   } while (ret == -1 && 
            --retries &&
            ndb->getNdbError().status == NdbError::TemporaryError);
@@ -4894,7 +4897,8 @@ ha_ndbcluster::ha_ndbcluster(TABLE *table_arg):
                 HA_NEED_READ_RANGE_BUFFER |
                 HA_CAN_GEOMETRY |
                 HA_CAN_BIT_FIELD |
-                HA_PARTIAL_COLUMN_READ),
+                HA_PARTIAL_COLUMN_READ |
+                HA_EXTERNAL_AUTO_INCREMENT),
   m_share(0),
   m_use_write(FALSE),
   m_ignore_dup_key(FALSE),
diff --git a/sql/handler.cc b/sql/handler.cc
index c6c31593a5f..31c3827dc25 100644
--- a/sql/handler.cc
+++ b/sql/handler.cc
@@ -1598,6 +1598,8 @@ int handler::update_auto_increment()
   ulonglong nr;
   THD *thd= table->in_use;
   struct system_variables *variables= &thd->variables;
+  bool external_auto_increment= 
+       table->file->table_flags() & HA_EXTERNAL_AUTO_INCREMENT;
   DBUG_ENTER("handler::update_auto_increment");
 
   /*
@@ -1615,12 +1617,12 @@ int handler::update_auto_increment()
     adjust_next_insert_id_after_explicit_value(nr);
     DBUG_RETURN(0);
   }
-  if (!(nr= thd->next_insert_id))
+  if (external_auto_increment || !(nr= thd->next_insert_id))
   {
     if ((nr= get_auto_increment()) == ~(ulonglong) 0)
       DBUG_RETURN(HA_ERR_AUTOINC_READ_FAILED);  // Mark failure
 
-    if (variables->auto_increment_increment != 1)
+    if (!external_auto_increment && variables->auto_increment_increment != 1)
       nr= next_insert_id(nr-1, variables);
     /*
       Update next row based on the found value. This way we don't have to
diff --git a/sql/handler.h b/sql/handler.h
index 9863d541b5f..8847d0306fb 100644
--- a/sql/handler.h
+++ b/sql/handler.h
@@ -93,7 +93,8 @@
 #define HA_CAN_BIT_FIELD       (1 << 28) /* supports bit fields */
 #define HA_NEED_READ_RANGE_BUFFER (1 << 29) /* for read_multi_range */
 #define HA_ANY_INDEX_MAY_BE_UNIQUE (1 << 30)
-
+/* The storage engine manages auto_increment itself */
+#define HA_EXTERNAL_AUTO_INCREMENT (1 << 31)
 
 /* bits in index_flags(index_number) for what you can do with index */
 #define HA_READ_NEXT            1       /* TODO really use this flag */