NDB bug-6018 support writeTuple with blobs

mysql-test/r/ndb_blob.result:
  bug-6018
mysql-test/t/ndb_blob.test:
  bug-6018
ndb/include/ndbapi/NdbBlob.hpp:
  bug-6018
ndb/include/ndbapi/NdbConnection.hpp:
  bug-6018
ndb/include/ndbapi/NdbIndexOperation.hpp:
  bug-6018
ndb/include/ndbapi/NdbOperation.hpp:
  bug-6018
ndb/src/ndbapi/NdbBlob.cpp:
  bug-6018
ndb/src/ndbapi/NdbConnection.cpp:
  bug-6018
ndb/src/ndbapi/NdbDictionaryImpl.cpp:
  bug-6018
ndb/src/ndbapi/NdbIndexOperation.cpp:
  bug-6018
ndb/src/ndbapi/NdbOperation.cpp:
  bug-6018
ndb/src/ndbapi/NdbOperationExec.cpp:
  bug-6018
ndb/test/ndbapi/testBlobs.cpp:
  bug-6018

13 files changed, 753 insertions, 386 deletions

diff --git a/mysql-test/r/ndb_blob.result b/mysql-test/r/ndb_blob.result
index cf64ec41ae3..1f2cf33f57d 100644
--- a/mysql-test/r/ndb_blob.result
+++ b/mysql-test/r/ndb_blob.result
@@ -1,25 +1,5 @@
 drop table if exists t1;
-drop database if exists mysqltest;
-create table t1 (
-a int not null primary key,
-b tinytext
-) engine=ndbcluster;
-insert into t1 values(1, 'x');
-update t1 set b = 'y';
-select * from t1;
-a	b
-1	y
-delete from t1;
-drop table t1;
-create table t1 (
-a int not null primary key,
-b text not null
-) engine=ndbcluster;
-insert into t1 values(1, '');
-select * from t1;
-a	b
-1	
-drop table t1;
+drop database if exists test2;
 set autocommit=0;
 create table t1 (
 a int not null primary key,
@@ -102,6 +82,53 @@ commit;
 select count(*) from t1;
 count(*)
 0
+replace t1 set a=1,b=@b1,c=111,d=@d1;
+replace t1 set a=2,b=@b2,c=222,d=@d2;
+commit;
+explain select * from t1 where a = 1;
+id	select_type	table	type	possible_keys	key	key_len	ref	rows	Extra
+1	SIMPLE	t1	const	PRIMARY	PRIMARY	4	const	1	
+select a,length(b),substr(b,1+2*900,2),length(d),substr(d,1+3*900,3)
+from t1 where a=1;
+a	length(b)	substr(b,1+2*900,2)	length(d)	substr(d,1+3*900,3)
+1	2256	b1	3000	dd1
+select a,length(b),substr(b,1+2*9000,2),length(d),substr(d,1+3*9000,3)
+from t1 where a=2;
+a	length(b)	substr(b,1+2*9000,2)	length(d)	substr(d,1+3*9000,3)
+2	20000	b2	30000	dd2
+replace t1 set a=1,b=@b2,c=111,d=@d2;
+replace t1 set a=2,b=@b1,c=222,d=@d1;
+commit;
+select a,length(b),substr(b,1+2*9000,2),length(d),substr(d,1+3*9000,3)
+from t1 where a=1;
+a	length(b)	substr(b,1+2*9000,2)	length(d)	substr(d,1+3*9000,3)
+1	20000	b2	30000	dd2
+select a,length(b),substr(b,1+2*900,2),length(d),substr(d,1+3*900,3)
+from t1 where a=2;
+a	length(b)	substr(b,1+2*900,2)	length(d)	substr(d,1+3*900,3)
+2	2256	b1	3000	dd1
+replace t1 set a=1,b=concat(@b2,@b2),c=111,d=concat(@d2,@d2);
+replace t1 set a=2,b=concat(@b1,@b1),c=222,d=concat(@d1,@d1);
+commit;
+select a,length(b),substr(b,1+4*9000,2),length(d),substr(d,1+6*9000,3)
+from t1 where a=1;
+a	length(b)	substr(b,1+4*9000,2)	length(d)	substr(d,1+6*9000,3)
+1	40000	b2	60000	dd2
+select a,length(b),substr(b,1+4*900,2),length(d),substr(d,1+6*900,3)
+from t1 where a=2;
+a	length(b)	substr(b,1+4*900,2)	length(d)	substr(d,1+6*900,3)
+2	4512	b1	6000	dd1
+replace t1 set a=1,b='xyz',c=111,d=null;
+commit;
+select a,b from t1 where d is null;
+a	b
+1	xyz
+delete from t1 where a=1;
+delete from t1 where a=2;
+commit;
+select count(*) from t1;
+count(*)
+0
 insert into t1 values(1,@b1,111,@d1);
 insert into t1 values(2,@b2,222,@d2);
 commit;
@@ -241,90 +268,6 @@ a	b	c	d
 7	7xb7	777	7xdd7
 8	8xb8	888	8xdd8
 9	9xb9	999	9xdd9
-select * from t1 order by a;
-a	b	c	d
-1	1xb1	111	1xdd1
-2	2xb2	222	2xdd2
-3	3xb3	333	3xdd3
-4	4xb4	444	4xdd4
-5	5xb5	555	5xdd5
-6	6xb6	666	6xdd6
-7	7xb7	777	7xdd7
-8	8xb8	888	8xdd8
-9	9xb9	999	9xdd9
-alter table t1 add x int;
-select * from t1 order by a;
-a	b	c	d	x
-1	1xb1	111	1xdd1	NULL
-2	2xb2	222	2xdd2	NULL
-3	3xb3	333	3xdd3	NULL
-4	4xb4	444	4xdd4	NULL
-5	5xb5	555	5xdd5	NULL
-6	6xb6	666	6xdd6	NULL
-7	7xb7	777	7xdd7	NULL
-8	8xb8	888	8xdd8	NULL
-9	9xb9	999	9xdd9	NULL
-alter table t1 drop x;
-select * from t1 order by a;
-a	b	c	d
-1	1xb1	111	1xdd1
-2	2xb2	222	2xdd2
-3	3xb3	333	3xdd3
-4	4xb4	444	4xdd4
-5	5xb5	555	5xdd5
-6	6xb6	666	6xdd6
-7	7xb7	777	7xdd7
-8	8xb8	888	8xdd8
-9	9xb9	999	9xdd9
-create database mysqltest;
-use mysqltest;
-CREATE TABLE t2 (
-a bigint unsigned NOT NULL PRIMARY KEY,
-b int unsigned not null,
-c int unsigned
-) engine=ndbcluster;
-insert into t2 values (1,1,1),(2,2,2);
-select * from test.t1,t2 where test.t1.a = t2.a order by test.t1.a;
-a	b	c	d	a	b	c
-1	1xb1	111	1xdd1	1	1	1
-2	2xb2	222	2xdd2	2	2	2
-drop table t2;
-use test;
-select * from t1 order by a;
-a	b	c	d
-1	1xb1	111	1xdd1
-2	2xb2	222	2xdd2
-3	3xb3	333	3xdd3
-4	4xb4	444	4xdd4
-5	5xb5	555	5xdd5
-6	6xb6	666	6xdd6
-7	7xb7	777	7xdd7
-8	8xb8	888	8xdd8
-9	9xb9	999	9xdd9
-alter table t1 add x int;
-select * from t1 order by a;
-a	b	c	d	x
-1	1xb1	111	1xdd1	NULL
-2	2xb2	222	2xdd2	NULL
-3	3xb3	333	3xdd3	NULL
-4	4xb4	444	4xdd4	NULL
-5	5xb5	555	5xdd5	NULL
-6	6xb6	666	6xdd6	NULL
-7	7xb7	777	7xdd7	NULL
-8	8xb8	888	8xdd8	NULL
-9	9xb9	999	9xdd9	NULL
-alter table t1 drop x;
-select * from t1 order by a;
-a	b	c	d
-1	1xb1	111	1xdd1
-2	2xb2	222	2xdd2
-3	3xb3	333	3xdd3
-4	4xb4	444	4xdd4
-5	5xb5	555	5xdd5
-6	6xb6	666	6xdd6
-7	7xb7	777	7xdd7
-8	8xb8	888	8xdd8
-9	9xb9	999	9xdd9
 delete from t1 where c >= 100;
 commit;
 select count(*) from t1;
@@ -375,8 +318,122 @@ rollback;
 select count(*) from t1;
 count(*)
 0
+insert into t1 values(1,'b1',111,'dd1');
+insert into t1 values(2,'b2',222,'dd2');
+insert into t1 values(3,'b3',333,'dd3');
+insert into t1 values(4,'b4',444,'dd4');
+insert into t1 values(5,'b5',555,'dd5');
+insert into t1 values(6,'b6',666,'dd6');
+insert into t1 values(7,'b7',777,'dd7');
+insert into t1 values(8,'b8',888,'dd8');
+insert into t1 values(9,'b9',999,'dd9');
+commit;
+select * from t1 order by a;
+a	b	c	d
+1	b1	111	dd1
+2	b2	222	dd2
+3	b3	333	dd3
+4	b4	444	dd4
+5	b5	555	dd5
+6	b6	666	dd6
+7	b7	777	dd7
+8	b8	888	dd8
+9	b9	999	dd9
+alter table t1 add x int;
+select * from t1 order by a;
+a	b	c	d	x
+1	b1	111	dd1	NULL
+2	b2	222	dd2	NULL
+3	b3	333	dd3	NULL
+4	b4	444	dd4	NULL
+5	b5	555	dd5	NULL
+6	b6	666	dd6	NULL
+7	b7	777	dd7	NULL
+8	b8	888	dd8	NULL
+9	b9	999	dd9	NULL
+alter table t1 drop x;
+select * from t1 order by a;
+a	b	c	d
+1	b1	111	dd1
+2	b2	222	dd2
+3	b3	333	dd3
+4	b4	444	dd4
+5	b5	555	dd5
+6	b6	666	dd6
+7	b7	777	dd7
+8	b8	888	dd8
+9	b9	999	dd9
+create database test2;
+use test2;
+CREATE TABLE t2 (
+a bigint unsigned NOT NULL PRIMARY KEY,
+b int unsigned not null,
+c int unsigned
+) engine=ndbcluster;
+insert into t2 values (1,1,1),(2,2,2);
+select * from test.t1,t2 where test.t1.a = t2.a order by test.t1.a;
+a	b	c	d	a	b	c
+1	b1	111	dd1	1	1	1
+2	b2	222	dd2	2	2	2
+drop table t2;
+use test;
+select * from t1 order by a;
+a	b	c	d
+1	b1	111	dd1
+2	b2	222	dd2
+3	b3	333	dd3
+4	b4	444	dd4
+5	b5	555	dd5
+6	b6	666	dd6
+7	b7	777	dd7
+8	b8	888	dd8
+9	b9	999	dd9
+alter table t1 add x int;
+select * from t1 order by a;
+a	b	c	d	x
+1	b1	111	dd1	NULL
+2	b2	222	dd2	NULL
+3	b3	333	dd3	NULL
+4	b4	444	dd4	NULL
+5	b5	555	dd5	NULL
+6	b6	666	dd6	NULL
+7	b7	777	dd7	NULL
+8	b8	888	dd8	NULL
+9	b9	999	dd9	NULL
+alter table t1 drop x;
+select * from t1 order by a;
+a	b	c	d
+1	b1	111	dd1
+2	b2	222	dd2
+3	b3	333	dd3
+4	b4	444	dd4
+5	b5	555	dd5
+6	b6	666	dd6
+7	b7	777	dd7
+8	b8	888	dd8
+9	b9	999	dd9
+drop table t1;
+drop database test2;
+create table t1 (
+a int not null primary key,
+b tinytext
+) engine=ndbcluster;
+insert into t1 values(1, 'x');
+update t1 set b = 'y';
+select * from t1;
+a	b
+1	x
+delete from t1;
+drop table t1;
+create table t1 (
+a int not null primary key,
+b text not null
+) engine=ndbcluster;
+insert into t1 values(1, '');
+select * from t1;
+a	b
+1	
 drop table t1;
-drop database mysqltest;
 set autocommit=1;
 use test;
 CREATE TABLE t1 (
diff --git a/mysql-test/t/ndb_blob.test b/mysql-test/t/ndb_blob.test
index 5454dd91d26..ba5f089b17b 100644
--- a/mysql-test/t/ndb_blob.test
+++ b/mysql-test/t/ndb_blob.test
@@ -2,7 +2,7 @@
 
 --disable_warnings
 drop table if exists t1;
-drop database if exists mysqltest;
+drop database if exists test2;
 --enable_warnings
 
 #
@@ -12,31 +12,7 @@ drop database if exists mysqltest;
 # A prerequisite for this handler test is that "testBlobs" succeeds.
 #
 
-# -- bug-5252 tinytext crashes --
-
-create table t1 (
-  a int not null primary key,
-  b tinytext
-) engine=ndbcluster;
-
-insert into t1 values(1, 'x');
-update t1 set b = 'y';
-select * from t1;
-delete from t1;
-drop table t1;
-
-# -- bug-5013 insert empty string to text --
-
-create table t1 (
-  a int not null primary key,
-  b text not null
-) engine=ndbcluster;
-
-insert into t1 values(1, '');
-select * from t1;
-drop table t1;
-
--- general test starts --
+# -- general test starts --
 
 # make test harder with autocommit off
 set autocommit=0;
@@ -117,7 +93,6 @@ from t1 where a=2;
 # pk update to null
 update t1 set d=null where a=1;
 commit;
-# FIXME now fails at random due to weird mixup between the 2 rows
 select a from t1 where d is null;
 
 # pk delete
@@ -126,6 +101,49 @@ delete from t1 where a=2;
 commit;
 select count(*) from t1;
 
+# -- replace ( bug-6018 ) --
+
+# insert
+replace t1 set a=1,b=@b1,c=111,d=@d1;
+replace t1 set a=2,b=@b2,c=222,d=@d2;
+commit;
+explain select * from t1 where a = 1;
+
+# pk read
+select a,length(b),substr(b,1+2*900,2),length(d),substr(d,1+3*900,3)
+from t1 where a=1;
+select a,length(b),substr(b,1+2*9000,2),length(d),substr(d,1+3*9000,3)
+from t1 where a=2;
+
+# update
+replace t1 set a=1,b=@b2,c=111,d=@d2;
+replace t1 set a=2,b=@b1,c=222,d=@d1;
+commit;
+select a,length(b),substr(b,1+2*9000,2),length(d),substr(d,1+3*9000,3)
+from t1 where a=1;
+select a,length(b),substr(b,1+2*900,2),length(d),substr(d,1+3*900,3)
+from t1 where a=2;
+
+# update
+replace t1 set a=1,b=concat(@b2,@b2),c=111,d=concat(@d2,@d2);
+replace t1 set a=2,b=concat(@b1,@b1),c=222,d=concat(@d1,@d1);
+commit;
+select a,length(b),substr(b,1+4*9000,2),length(d),substr(d,1+6*9000,3)
+from t1 where a=1;
+select a,length(b),substr(b,1+4*900,2),length(d),substr(d,1+6*900,3)
+from t1 where a=2;
+
+# update to null
+replace t1 set a=1,b='xyz',c=111,d=null;
+commit;
+select a,b from t1 where d is null;
+
+# pk delete
+delete from t1 where a=1;
+delete from t1 where a=2;
+commit;
+select count(*) from t1;
+
 # -- hash index ops --
 
 insert into t1 values(1,@b1,111,@d1);
@@ -231,39 +249,6 @@ where c >= 100;
 commit;
 select * from t1 where c >= 100 order by a;
 
-# alter table
-
-select * from t1 order by a;
-alter table t1 add x int;
-select * from t1 order by a;
-alter table t1 drop x;
-select * from t1 order by a;
-
-# multi db
-
-create database mysqltest;
-use mysqltest;
-
-CREATE TABLE t2 (
-  a bigint unsigned NOT NULL PRIMARY KEY,
-  b int unsigned not null,
-  c int unsigned
-) engine=ndbcluster;
-
-insert into t2 values (1,1,1),(2,2,2);
-select * from test.t1,t2 where test.t1.a = t2.a order by test.t1.a;
-
-drop table t2;
-use test;
-
-# alter table
-
-select * from t1 order by a;
-alter table t1 add x int;
-select * from t1 order by a;
-alter table t1 drop x;
-select * from t1 order by a;
-
 # range scan delete
 delete from t1 where c >= 100;
 commit;
@@ -306,10 +291,77 @@ select a,length(b),substr(b,1+2*900,2),length(d),substr(d,1+3*900,3)
 from t1 order by a;
 rollback;
 select count(*) from t1;
+
+# -- alter table and multi db --
+
+insert into t1 values(1,'b1',111,'dd1');
+insert into t1 values(2,'b2',222,'dd2');
+insert into t1 values(3,'b3',333,'dd3');
+insert into t1 values(4,'b4',444,'dd4');
+insert into t1 values(5,'b5',555,'dd5');
+insert into t1 values(6,'b6',666,'dd6');
+insert into t1 values(7,'b7',777,'dd7');
+insert into t1 values(8,'b8',888,'dd8');
+insert into t1 values(9,'b9',999,'dd9');
+commit;
+
+select * from t1 order by a;
+alter table t1 add x int;
+select * from t1 order by a;
+alter table t1 drop x;
+select * from t1 order by a;
+
+create database test2;
+use test2;
+
+CREATE TABLE t2 (
+  a bigint unsigned NOT NULL PRIMARY KEY,
+  b int unsigned not null,
+  c int unsigned
+) engine=ndbcluster;
+
+insert into t2 values (1,1,1),(2,2,2);
+select * from test.t1,t2 where test.t1.a = t2.a order by test.t1.a;
+
+drop table t2;
+use test;
+
+select * from t1 order by a;
+alter table t1 add x int;
+select * from t1 order by a;
+alter table t1 drop x;
+select * from t1 order by a;
+
+# -- end general test --
+
+drop table t1;
+drop database test2;
+
+# -- bug-5252 tinytext crashes --
+
+create table t1 (
+  a int not null primary key,
+  b tinytext
+) engine=ndbcluster;
+
+insert into t1 values(1, 'x');
+update t1 set b = 'y';
+select * from t1;
+delete from t1;
+drop table t1;
+
+# -- bug-5013 insert empty string to text --
+
+create table t1 (
+  a int not null primary key,
+  b text not null
+) engine=ndbcluster;
+
+insert into t1 values(1, '');
+select * from t1;
 drop table t1;
-drop database mysqltest;
 
-# bug #5349
+# -- bug #5349 --
 set autocommit=1;
 use test;
 CREATE TABLE t1 (
@@ -327,7 +379,7 @@ select * from t1 order by a;
 alter table t1 engine=ndb;
 select * from t1 order by a;
 
-# bug #5872
+# -- bug #5872 --
 alter table t1 engine=myisam;
 select * from t1 order by a;
 drop table t1;
diff --git a/ndb/include/ndbapi/NdbBlob.hpp b/ndb/include/ndbapi/NdbBlob.hpp
index 22e393b6c5d..b3c28c9e950 100644
--- a/ndb/include/ndbapi/NdbBlob.hpp
+++ b/ndb/include/ndbapi/NdbBlob.hpp
@@ -36,7 +36,7 @@ class NdbColumnImpl;
  * Blob data is stored in 2 places:
  *
  * - "header" and "inline bytes" stored in the blob attribute
- * - "blob parts" stored in a separate table NDB$BLOB_<t>_<v>_<c>
+ * - "blob parts" stored in a separate table NDB$BLOB_<tid>_<cid>
  *
  * Inline and part sizes can be set via NdbDictionary::Column methods
  * when the table is created.
@@ -74,23 +74,21 @@ class NdbColumnImpl;
  * NdbBlob methods return -1 on error and 0 on success, and use output
  * parameters when necessary.
  *
- * Notes:
- * - table and its blob part tables are not created atomically
- * - scan must use the "new" interface NdbScanOperation
- * - to update a blob in a read op requires exclusive tuple lock
- * - update op in scan must do its own getBlobHandle
- * - delete creates implicit, not-accessible blob handles
- * - NdbOperation::writeTuple does not support blobs
- * - there is no support for an asynchronous interface
+ * Operation types:
+ * - insertTuple must use setValue if blob column is non-nullable
+ * - readTuple with exclusive lock can also update existing value
+ * - updateTuple can overwrite with setValue or update existing value
+ * - writeTuple always overwrites and must use setValue if non-nullable
+ * - deleteTuple creates implicit non-accessible blob handles
+ * - scan with exclusive lock can also update existing value
+ * - scan "lock takeover" update op must do its own getBlobHandle
  *
  * Bugs / limitations:
- * - scan must use exclusive locking for now
- *
- * Todo:
- * - add scan method hold-read-lock + return-keyinfo
- * - check keyinfo length when setting keys
- * - check allowed blob ops vs locking mode
- * - overload control (too many pending ops)
+ * - lock mode upgrade should be handled automatically
+ * - lock mode vs allowed operation is not checked
+ * - too many pending blob ops can blow up i/o buffers
+ * - table and its blob part tables are not created atomically
+ * - there is no support for an asynchronous interface
  */
 class NdbBlob {
 public:
@@ -173,19 +171,11 @@ public:
    */
   int readData(void* data, Uint32& bytes);
   /**
-   * Read at given position.  Does not use or update current position.
-   */
-  int readData(Uint64 pos, void* data, Uint32& bytes);
-  /**
    * Write at current position and set new position to first byte after
    * the data written.  A write past blob end extends the blob value.
    */
   int writeData(const void* data, Uint32 bytes);
   /**
-   * Write at given position. Does not use or update current position.
-   */
-  int writeData(Uint64 pos, const void* data, Uint32 bytes);
-  /**
    * Return the blob column.
    */
   const NdbDictionary::Column* getColumn();
@@ -266,14 +256,17 @@ private:
     Buf();
     ~Buf();
     void alloc(unsigned n);
+    void copyfrom(const Buf& src);
   };
   Buf theKeyBuf;
   Buf theAccessKeyBuf;
   Buf theHeadInlineBuf;
+  Buf theHeadInlineCopyBuf;     // for writeTuple
   Buf thePartBuf;
   Head* theHead;
   char* theInlineData;
   NdbRecAttr* theHeadInlineRecAttr;
+  NdbOperation* theHeadInlineReadOp;
   bool theHeadInlineUpdateFlag;
   // length and read/write position
   int theNullFlag;
@@ -294,6 +287,7 @@ private:
   bool isReadOp();
   bool isInsertOp();
   bool isUpdateOp();
+  bool isWriteOp();
   bool isDeleteOp();
   bool isScanOp();
   // computations
@@ -309,12 +303,13 @@ private:
   void getHeadFromRecAttr();
   int setHeadInlineValue(NdbOperation* anOp);
   // data operations
-  int readDataPrivate(Uint64 pos, char* buf, Uint32& bytes);
-  int writeDataPrivate(Uint64 pos, const char* buf, Uint32 bytes);
+  int readDataPrivate(char* buf, Uint32& bytes);
+  int writeDataPrivate(const char* buf, Uint32 bytes);
   int readParts(char* buf, Uint32 part, Uint32 count);
   int insertParts(const char* buf, Uint32 part, Uint32 count);
   int updateParts(const char* buf, Uint32 part, Uint32 count);
   int deleteParts(Uint32 part, Uint32 count);
+  int deletePartsUnknown(Uint32 part);
   // pending ops
   int executePendingBlobReads();
   int executePendingBlobWrites();
diff --git a/ndb/include/ndbapi/NdbConnection.hpp b/ndb/include/ndbapi/NdbConnection.hpp
index 92b940e96f7..7af5d27b922 100644
--- a/ndb/include/ndbapi/NdbConnection.hpp
+++ b/ndb/include/ndbapi/NdbConnection.hpp
@@ -526,7 +526,7 @@ private:
   int 	sendCOMMIT();                   // Send a TC_COMMITREQ signal;
   void	setGCI(int GCI);		// Set the global checkpoint identity
  
-  int	OpCompleteFailure(Uint8 abortoption);
+  int	OpCompleteFailure(Uint8 abortoption, bool setFailure = true);
   int	OpCompleteSuccess();
   void	CompletedOperations();	        // Move active ops to list of completed
  
@@ -552,7 +552,7 @@ private:
   void		setOperationErrorCode(int anErrorCode);	
 
   // Indicate something went wrong in the definition phase
-  void		setOperationErrorCodeAbort(int anErrorCode);
+  void		setOperationErrorCodeAbort(int anErrorCode, int abortOption = -1);
 
   int		checkMagicNumber();		       // Verify correct object
   NdbOperation* getNdbOperation(const class NdbTableImpl* aTable,
diff --git a/ndb/include/ndbapi/NdbIndexOperation.hpp b/ndb/include/ndbapi/NdbIndexOperation.hpp
index 7612fe54d1b..1472f1b249e 100644
--- a/ndb/include/ndbapi/NdbIndexOperation.hpp
+++ b/ndb/include/ndbapi/NdbIndexOperation.hpp
@@ -49,6 +49,9 @@ public:
    * @{
    */
 
+  /** insert is not allowed */
+  int insertTuple();
+
   /**
    * Define the NdbIndexOperation to be a standard operation of type readTuple.
    * When calling NdbConnection::execute, this operation
@@ -193,6 +196,7 @@ private:
 
   // Private attributes
   const NdbIndexImpl* m_theIndex;
+  const NdbTableImpl* m_thePrimaryTable;
   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
diff --git a/ndb/include/ndbapi/NdbOperation.hpp b/ndb/include/ndbapi/NdbOperation.hpp
index 8e0294e41e6..46d4ddab0f5 100644
--- a/ndb/include/ndbapi/NdbOperation.hpp
+++ b/ndb/include/ndbapi/NdbOperation.hpp
@@ -918,6 +918,13 @@ protected:
   // Blobs in this operation
   NdbBlob* theBlobList;
 
+  /*
+   * Abort option per operation, used by blobs.  Default -1.  If set,
+   * overrides abort option on connection level.  If set to IgnoreError,
+   * does not cause execute() to return failure.  This is different from
+   * IgnoreError on connection level.
+   */
+  Int8 m_abortOption;
 };
 
 #ifdef NDB_NO_DROPPED_SIGNAL
@@ -1160,5 +1167,3 @@ NdbOperation::setValue(Uint32 anAttrId, double aPar)
 }
 
 #endif
-
-
diff --git a/ndb/src/ndbapi/NdbBlob.cpp b/ndb/src/ndbapi/NdbBlob.cpp
index feab95d8ca5..5b193870558 100644
--- a/ndb/src/ndbapi/NdbBlob.cpp
+++ b/ndb/src/ndbapi/NdbBlob.cpp
@@ -33,21 +33,24 @@
     ndbout << prefix << " " << hex << (void*)this << " " << cname; \
     ndbout << " " << dec << __LINE__ << " " << x << " " << *this << endl; \
   } while (0)
-#else
-#define DBG(x)
-#endif
 
 static char*
 ndb_blob_debug(const Uint32* data, unsigned size)
 {
-  static char buf[128 + 1];     // MT irrelevant
+  static char buf[200]; // MT irrelevant
   buf[0] = 0;
-  for (unsigned i = 0; i < size && i < 128 / 4; i++) {
-    sprintf(buf + strlen(buf), "%*s%08x", i != 0, "", data[i]);
+  for (unsigned i = 0; i < size; i++) {
+    unsigned n = strlen(buf);
+    if (n + 10 < sizeof(buf))
+      sprintf(buf + n, "%*s%08x", i != 0, "", data[i]);
   }
   return buf;
 }
 
+#else
+#define DBG(x)
+#endif
+
 /*
  * Reading index table directly (as a table) is faster but there are
  * bugs or limitations.  Keep the code and make possible to choose.
@@ -162,6 +165,7 @@ NdbBlob::init()
   theHead = NULL;
   theInlineData = NULL;
   theHeadInlineRecAttr = NULL;
+  theHeadInlineReadOp = NULL;
   theHeadInlineUpdateFlag = false;
   theNullFlag = -1;
   theLength = 0;
@@ -206,6 +210,13 @@ NdbBlob::Buf::alloc(unsigned n)
 #endif
 }
 
+void
+NdbBlob::Buf::copyfrom(const NdbBlob::Buf& src)
+{
+  assert(size == src.size);
+  memcpy(data, src.data, size);
+}
+
 // classify operations (inline)
 
 inline bool
@@ -226,6 +237,7 @@ NdbBlob::isKeyOp()
   return
     theNdbOp->theOperationType == NdbOperation::InsertRequest ||
     theNdbOp->theOperationType == NdbOperation::UpdateRequest ||
+    theNdbOp->theOperationType == NdbOperation::WriteRequest ||
     theNdbOp->theOperationType == NdbOperation::ReadRequest ||
     theNdbOp->theOperationType == NdbOperation::ReadExclusive ||
     theNdbOp->theOperationType == NdbOperation::DeleteRequest;
@@ -254,6 +266,13 @@ NdbBlob::isUpdateOp()
 }
 
 inline bool
+NdbBlob::isWriteOp()
+{
+  return
+    theNdbOp->theOperationType == NdbOperation::WriteRequest;
+}
+
+inline bool
 NdbBlob::isDeleteOp()
 {
   return
@@ -289,7 +308,7 @@ inline Uint32
 NdbBlob::getDistKey(Uint32 part)
 {
   assert(theStripeSize != 0);
-  return (part / theStripeSize) % theStripeSize;
+  return part / theStripeSize;
 }
 
 // getters and setters
@@ -401,7 +420,7 @@ NdbBlob::getHeadFromRecAttr()
   theNullFlag = theHeadInlineRecAttr->isNULL();
   assert(theNullFlag != -1);
   theLength = ! theNullFlag ? theHead->length : 0;
-  DBG("getHeadFromRecAttr out");
+  DBG("getHeadFromRecAttr [out]");
 }
 
 int
@@ -453,7 +472,7 @@ NdbBlob::setValue(const void* data, Uint32 bytes)
     setErrorCode(ErrState);
     return -1;
   }
-  if (! isInsertOp() && ! isUpdateOp()) {
+  if (! isInsertOp() && ! isUpdateOp() && ! isWriteOp()) {
     setErrorCode(ErrUsage);
     return -1;
   }
@@ -466,11 +485,12 @@ NdbBlob::setValue(const void* data, Uint32 bytes)
   theGetSetBytes = bytes;
   if (isInsertOp()) {
     // write inline part now
-    if (theSetBuf != 0) {
-      unsigned n = theGetSetBytes;
+    if (theSetBuf != NULL) {
+      Uint32 n = theGetSetBytes;
       if (n > theInlineSize)
         n = theInlineSize;
-      if (writeDataPrivate(0, theSetBuf, n) == -1)
+      assert(thePos == 0);
+      if (writeDataPrivate(theSetBuf, n) == -1)
         return -1;
     } else {
       theNullFlag = true;
@@ -555,7 +575,7 @@ NdbBlob::getLength(Uint64& len)
 int
 NdbBlob::truncate(Uint64 length)
 {
-  DBG("truncate length=" << length);
+  DBG("truncate [in] length=" << length);
   if (theNullFlag == -1) {
     setErrorCode(ErrState);
     return -1;
@@ -573,7 +593,10 @@ NdbBlob::truncate(Uint64 length)
     }
     theLength = length;
     theHeadInlineUpdateFlag = true;
+    if (thePos > length)
+      thePos = length;
   }
+  DBG("truncate [out]");
   return 0;
 }
 
@@ -610,32 +633,20 @@ NdbBlob::setPos(Uint64 pos)
 int
 NdbBlob::readData(void* data, Uint32& bytes)
 {
-  if (readData(thePos, data, bytes) == -1)
-    return -1;
-  thePos += bytes;
-  assert(thePos <= theLength);
-  return 0;
-}
-
-int
-NdbBlob::readData(Uint64 pos, void* data, Uint32& bytes)
-{
   if (theState != Active) {
     setErrorCode(ErrState);
     return -1;
   }
   char* buf = static_cast<char*>(data);
-  return readDataPrivate(pos, buf, bytes);
+  return readDataPrivate(buf, bytes);
 }
 
 int
-NdbBlob::readDataPrivate(Uint64 pos, char* buf, Uint32& bytes)
+NdbBlob::readDataPrivate(char* buf, Uint32& bytes)
 {
-  DBG("readData pos=" << pos << " bytes=" << bytes);
-  if (pos > theLength) {
-    setErrorCode(ErrSeek);
-    return -1;
-  }
+  DBG("readData [in] bytes=" << bytes);
+  assert(thePos <= theLength);
+  Uint64 pos = thePos;
   if (bytes > theLength - pos)
     bytes = theLength - pos;
   Uint32 len = bytes;
@@ -709,38 +720,29 @@ NdbBlob::readDataPrivate(Uint64 pos, char* buf, Uint32& bytes)
     len -= n;
   }
   assert(len == 0);
-  return 0;
-}
-
-int
-NdbBlob::writeData(const void* data, Uint32 bytes)
-{
-  if (writeData(thePos, data, bytes) == -1)
-    return -1;
-  thePos += bytes;
+  thePos = pos;
   assert(thePos <= theLength);
+  DBG("readData [out]");
   return 0;
 }
 
 int
-NdbBlob::writeData(Uint64 pos, const void* data, Uint32 bytes)
+NdbBlob::writeData(const void* data, Uint32 bytes)
 {
   if (theState != Active) {
     setErrorCode(ErrState);
     return -1;
   }
   const char* buf = static_cast<const char*>(data);
-  return writeDataPrivate(pos, buf, bytes);
+  return writeDataPrivate(buf, bytes);
 }
 
 int
-NdbBlob::writeDataPrivate(Uint64 pos, const char* buf, Uint32 bytes)
+NdbBlob::writeDataPrivate(const char* buf, Uint32 bytes)
 {
-  DBG("writeData pos=" << pos << " bytes=" << bytes);
-  if (pos > theLength) {
-    setErrorCode(ErrSeek);
-    return -1;
-  }
+  DBG("writeData [in] bytes=" << bytes);
+  assert(thePos <= theLength);
+  Uint64 pos = thePos;
   Uint32 len = bytes;
   // any write makes blob not NULL
   if (theNullFlag) {
@@ -778,7 +780,7 @@ NdbBlob::writeDataPrivate(Uint64 pos, const char* buf, Uint32 bytes)
       if (readParts(thePartBuf.data, part, 1) == -1)
         return -1;
       // need result now
-      DBG("execute pending part reafs");
+      DBG("execute pending part reads");
       if (executePendingBlobReads() == -1)
         return -1;
       Uint32 n = thePartSize - off;
@@ -855,14 +857,16 @@ NdbBlob::writeDataPrivate(Uint64 pos, const char* buf, Uint32 bytes)
     theLength = pos;
     theHeadInlineUpdateFlag = true;
   }
-  DBG("writeData out");
+  thePos = pos;
+  assert(thePos <= theLength);
+  DBG("writeData [out]");
   return 0;
 }
 
 int
 NdbBlob::readParts(char* buf, Uint32 part, Uint32 count)
 {
-  DBG("readParts part=" << part << " count=" << count);
+  DBG("readParts [in] part=" << part << " count=" << count);
   Uint32 n = 0;
   while (n < count) {
     NdbOperation* tOp = theNdbCon->getNdbOperation(theBlobTable);
@@ -873,6 +877,7 @@ NdbBlob::readParts(char* buf, Uint32 part, Uint32 count)
       setErrorCode(tOp);
       return -1;
     }
+    tOp->m_abortOption = AbortOnError;
     buf += thePartSize;
     n++;
     thePendingBlobOps |= (1 << NdbOperation::ReadRequest);
@@ -884,7 +889,7 @@ NdbBlob::readParts(char* buf, Uint32 part, Uint32 count)
 int
 NdbBlob::insertParts(const char* buf, Uint32 part, Uint32 count)
 {
-  DBG("insertParts part=" << part << " count=" << count);
+  DBG("insertParts [in] part=" << part << " count=" << count);
   Uint32 n = 0;
   while (n < count) {
     NdbOperation* tOp = theNdbCon->getNdbOperation(theBlobTable);
@@ -895,6 +900,7 @@ NdbBlob::insertParts(const char* buf, Uint32 part, Uint32 count)
       setErrorCode(tOp);
       return -1;
     }
+    tOp->m_abortOption = AbortOnError;
     buf += thePartSize;
     n++;
     thePendingBlobOps |= (1 << NdbOperation::InsertRequest);
@@ -906,7 +912,7 @@ NdbBlob::insertParts(const char* buf, Uint32 part, Uint32 count)
 int
 NdbBlob::updateParts(const char* buf, Uint32 part, Uint32 count)
 {
-  DBG("updateParts part=" << part << " count=" << count);
+  DBG("updateParts [in] part=" << part << " count=" << count);
   Uint32 n = 0;
   while (n < count) {
     NdbOperation* tOp = theNdbCon->getNdbOperation(theBlobTable);
@@ -917,6 +923,7 @@ NdbBlob::updateParts(const char* buf, Uint32 part, Uint32 count)
       setErrorCode(tOp);
       return -1;
     }
+    tOp->m_abortOption = AbortOnError;
     buf += thePartSize;
     n++;
     thePendingBlobOps |= (1 << NdbOperation::UpdateRequest);
@@ -928,7 +935,7 @@ NdbBlob::updateParts(const char* buf, Uint32 part, Uint32 count)
 int
 NdbBlob::deleteParts(Uint32 part, Uint32 count)
 {
-  DBG("deleteParts part=" << part << " count=" << count);
+  DBG("deleteParts [in] part=" << part << " count=" << count);
   Uint32 n = 0;
   while (n < count) {
     NdbOperation* tOp = theNdbCon->getNdbOperation(theBlobTable);
@@ -938,6 +945,7 @@ NdbBlob::deleteParts(Uint32 part, Uint32 count)
       setErrorCode(tOp);
       return -1;
     }
+    tOp->m_abortOption = AbortOnError;
     n++;
     thePendingBlobOps |= (1 << NdbOperation::DeleteRequest);
     theNdbCon->thePendingBlobOps |= (1 << NdbOperation::DeleteRequest);
@@ -945,6 +953,57 @@ NdbBlob::deleteParts(Uint32 part, Uint32 count)
   return 0;
 }
 
+/*
+ * Number of blob parts not known.  Used to check for race condition
+ * when writeTuple is used for insert.  Deletes all parts found.
+ */
+int
+NdbBlob::deletePartsUnknown(Uint32 part)
+{
+  DBG("deletePartsUnknown [in] part=" << part << " count=all");
+  static const unsigned maxbat = 256;
+  static const unsigned minbat = 1;
+  unsigned bat = minbat;
+  NdbOperation* tOpList[maxbat];
+  Uint32 count = 0;
+  while (true) {
+    Uint32 n;
+    n = 0;
+    while (n < bat) {
+      NdbOperation*& tOp = tOpList[n];  // ref
+      tOp = theNdbCon->getNdbOperation(theBlobTable);
+      if (tOp == NULL ||
+          tOp->deleteTuple() == -1 ||
+          setPartKeyValue(tOp, part + count + n) == -1) {
+        setErrorCode(tOp);
+        return -1;
+      }
+      tOp->m_abortOption = IgnoreError;
+      n++;
+      if (theNdbCon->executeNoBlobs(NoCommit) == -1)
+        return -1;
+    }
+    n = 0;
+    while (n < bat) {
+      NdbOperation* tOp = tOpList[n];
+      if (tOp->theError.code != 0) {
+        if (tOp->theError.code != 626) {
+          setErrorCode(tOp);
+          return -1;
+        }
+        // first non-existent part
+        DBG("deletePartsUnknown [out] count=" << count);
+        return 0;
+      }
+      n++;
+      count++;
+    }
+    bat *= 4;
+    if (bat > maxbat)
+      bat = maxbat;
+  }
+}
+
 // pending ops
 
 int
@@ -1007,7 +1066,7 @@ NdbBlob::atPrepare(NdbConnection* aCon, NdbOperation* anOp, const NdbColumnImpl*
   theTable = anOp->m_currentTable;
   theAccessTable = anOp->m_accessTable;
   theColumn = aColumn;
-  DBG("atPrepare");
+  DBG("atPrepare [in]");
   NdbDictionary::Column::Type partType = NdbDictionary::Column::Undefined;
   switch (theColumn->getType()) {
   case NdbDictionary::Column::Blob:
@@ -1046,6 +1105,7 @@ NdbBlob::atPrepare(NdbConnection* aCon, NdbOperation* anOp, const NdbColumnImpl*
   theKeyBuf.alloc(theTable->m_sizeOfKeysInWords << 2);
   theAccessKeyBuf.alloc(theAccessTable->m_sizeOfKeysInWords << 2);
   theHeadInlineBuf.alloc(sizeof(Head) + theInlineSize);
+  theHeadInlineCopyBuf.alloc(sizeof(Head) + theInlineSize);
   thePartBuf.alloc(thePartSize);
   theHead = (Head*)theHeadInlineBuf.data;
   theInlineData = theHeadInlineBuf.data + sizeof(Head);
@@ -1080,6 +1140,12 @@ NdbBlob::atPrepare(NdbConnection* aCon, NdbOperation* anOp, const NdbColumnImpl*
       theNullFlag = true;
       theLength = 0;
     }
+    if (isWriteOp()) {
+      // becomes NULL unless set before execute
+      theNullFlag = true;
+      theLength = 0;
+      theHeadInlineUpdateFlag = true;
+    }
     supportedOp = true;
   }
   if (isScanOp()) {
@@ -1093,19 +1159,21 @@ NdbBlob::atPrepare(NdbConnection* aCon, NdbOperation* anOp, const NdbColumnImpl*
     return -1;
   }
   setState(Prepared);
-  DBG("atPrepare out");
+  DBG("atPrepare [out]");
   return 0;
 }
 
 /*
  * Before execute of prepared operation.  May add new operations before
  * this one.  May ask that this operation and all before it (a "batch")
- * is executed immediately in no-commit mode.
+ * is executed immediately in no-commit mode.  In this case remaining
+ * prepared operations are saved in a separate list.  They are added
+ * back after postExecute.
  */
 int
 NdbBlob::preExecute(ExecType anExecType, bool& batch)
 {
-  DBG("preExecute");
+  DBG("preExecute [in]");
   if (theState == Invalid)
     return -1;
   assert(theState == Prepared);
@@ -1120,11 +1188,11 @@ NdbBlob::preExecute(ExecType anExecType, bool& batch)
   if (isInsertOp()) {
     if (theSetFlag && theGetSetBytes > theInlineSize) {
       // add ops to write rest of a setValue
-      assert(theSetBuf != 0);
-      Uint64 pos = theInlineSize;
+      assert(theSetBuf != NULL);
       const char* buf = theSetBuf + theInlineSize;
       Uint32 bytes = theGetSetBytes - theInlineSize;
-      if (writeDataPrivate(pos, buf, bytes) == -1)
+      assert(thePos == theInlineSize);
+      if (writeDataPrivate(buf, bytes) == -1)
         return -1;
       if (theHeadInlineUpdateFlag) {
           // add an operation to update head+inline
@@ -1136,11 +1204,12 @@ NdbBlob::preExecute(ExecType anExecType, bool& batch)
             setErrorCode(ErrAbort);
             return -1;
           }
+          DBG("add op to update head+inline");
       }
     }
   }
   if (isTableOp()) {
-    if (isUpdateOp() || isDeleteOp()) {
+    if (isUpdateOp() || isWriteOp() || isDeleteOp()) {
       // add operation before this one to read head+inline
       NdbOperation* tOp = theNdbCon->getNdbOperation(theTable, theNdbOp);
       if (tOp == NULL ||
@@ -1150,8 +1219,13 @@ NdbBlob::preExecute(ExecType anExecType, bool& batch)
         setErrorCode(tOp);
         return -1;
       }
+      if (isWriteOp()) {
+        tOp->m_abortOption = IgnoreError;
+      }
+      theHeadInlineReadOp = tOp;
       // execute immediately
       batch = true;
+      DBG("add op before to read head+inline");
     }
   }
   if (isIndexOp()) {
@@ -1180,6 +1254,7 @@ NdbBlob::preExecute(ExecType anExecType, bool& batch)
         }
       }
     }
+    DBG("added op before to read table key");
     if (isUpdateOp() || isDeleteOp()) {
       // add op before this one to read head+inline via index
       NdbIndexOperation* tOp = theNdbCon->getNdbIndexOperation(theAccessTable->m_index, theTable, theNdbOp);
@@ -1190,15 +1265,43 @@ NdbBlob::preExecute(ExecType anExecType, bool& batch)
         setErrorCode(tOp);
         return -1;
       }
+      if (isWriteOp()) {
+        tOp->m_abortOption = IgnoreError;
+      }
+      theHeadInlineReadOp = tOp;
       // execute immediately
       batch = true;
+      DBG("added index op before to read head+inline");
+    }
+    if (isWriteOp()) {
+      // XXX until IgnoreError fixed for index op
+      batch = true;
+    }
+  }
+  if (isWriteOp()) {
+    if (theSetFlag) {
+      // write head+inline now
+      theNullFlag = true;
+      theLength = 0;
+      if (theSetBuf != NULL) {
+        Uint32 n = theGetSetBytes;
+        if (n > theInlineSize)
+          n = theInlineSize;
+        assert(thePos == 0);
+        if (writeDataPrivate(theSetBuf, n) == -1)
+          return -1;
+      }
+      if (setHeadInlineValue(theNdbOp) == -1)
+        return -1;
+      // the read op before us may overwrite
+      theHeadInlineCopyBuf.copyfrom(theHeadInlineBuf);
     }
   }
   if (theActiveHook != NULL) {
     // need blob head for callback
     batch = true;
   }
-  DBG("preExecute out batch=" << batch);
+  DBG("preExecute [out] batch=" << batch);
   return 0;
 }
 
@@ -1211,15 +1314,16 @@ NdbBlob::preExecute(ExecType anExecType, bool& batch)
 int
 NdbBlob::postExecute(ExecType anExecType)
 {
-  DBG("postExecute type=" << anExecType);
+  DBG("postExecute [in] type=" << anExecType);
   if (theState == Invalid)
     return -1;
   if (theState == Active) {
     setState(anExecType == NoCommit ? Active : Closed);
-    DBG("postExecute skip");
+    DBG("postExecute [skip]");
     return 0;
   }
   assert(theState == Prepared);
+  setState(anExecType == NoCommit ? Active : Closed);
   assert(isKeyOp());
   if (isIndexOp()) {
     NdbBlob* tFirstBlob = theNdbOp->theBlobList;
@@ -1231,22 +1335,13 @@ NdbBlob::postExecute(ExecType anExecType)
   }
   if (isReadOp()) {
     getHeadFromRecAttr();
-    if (theGetFlag && theGetSetBytes > 0) {
-      // copy inline bytes to user buffer
-      assert(theGetBuf != NULL);
-      unsigned n = theGetSetBytes;
-      if (n > theInlineSize)
-        n = theInlineSize;
-      memcpy(theGetBuf, theInlineData, n);
-    }
-    if (theGetFlag && theGetSetBytes > theInlineSize) {
-      // add ops to read rest of a getValue
-      assert(anExecType == NoCommit);
-      assert(theGetBuf != 0);
-      Uint64 pos = theInlineSize;
-      char* buf = theGetBuf + theInlineSize;
-      Uint32 bytes = theGetSetBytes - theInlineSize;
-      if (readDataPrivate(pos, buf, bytes) == -1)
+    if (setPos(0) == -1)
+      return -1;
+    if (theGetFlag) {
+      assert(theGetSetBytes == 0 || theGetBuf != 0);
+      assert(theGetSetBytes <= theInlineSize || anExecType == NoCommit);
+      Uint32 bytes = theGetSetBytes;
+      if (readDataPrivate(theGetBuf, bytes) == -1)
         return -1;
     }
   }
@@ -1255,10 +1350,11 @@ NdbBlob::postExecute(ExecType anExecType)
     getHeadFromRecAttr();
     if (theSetFlag) {
       // setValue overwrites everything
-      if (theSetBuf != 0) {
+      if (theSetBuf != NULL) {
         if (truncate(0) == -1)
           return -1;
-        if (writeDataPrivate(0, theSetBuf, theGetSetBytes) == -1)
+        assert(thePos == 0);
+        if (writeDataPrivate(theSetBuf, theGetSetBytes) == -1)
           return -1;
       } else {
         if (setNull() == -1)
@@ -1266,6 +1362,57 @@ NdbBlob::postExecute(ExecType anExecType)
       }
     }
   }
+  if (isWriteOp() && isTableOp()) {
+    assert(anExecType == NoCommit);
+    if (theHeadInlineReadOp->theError.code == 0) {
+      int tNullFlag = theNullFlag;
+      Uint64 tLength = theLength;
+      Uint64 tPos = thePos;
+      getHeadFromRecAttr();
+      DBG("tuple found");
+      if (truncate(0) == -1)
+        return -1;
+      // restore previous head+inline
+      theHeadInlineBuf.copyfrom(theHeadInlineCopyBuf);
+      theNullFlag = tNullFlag;
+      theLength = tLength;
+      thePos = tPos;
+    } else {
+      if (theHeadInlineReadOp->theError.code != 626) {
+        setErrorCode(theHeadInlineReadOp);
+        return -1;
+      }
+      DBG("tuple not found");
+      /*
+       * Read found no tuple but it is possible that a tuple was
+       * created after the read by another transaction.  Delete all
+       * blob parts which may exist.
+       */
+      if (deletePartsUnknown(0) == -1)
+        return -1;
+    }
+    if (theSetFlag && theGetSetBytes > theInlineSize) {
+      assert(theSetBuf != NULL);
+      const char* buf = theSetBuf + theInlineSize;
+      Uint32 bytes = theGetSetBytes - theInlineSize;
+      assert(thePos == theInlineSize);
+      if (writeDataPrivate(buf, bytes) == -1)
+          return -1;
+    }
+  }
+  if (isWriteOp() && isIndexOp()) {
+    // XXX until IgnoreError fixed for index op
+    if (deletePartsUnknown(0) == -1)
+      return -1;
+    if (theSetFlag && theGetSetBytes > theInlineSize) {
+      assert(theSetBuf != NULL);
+      const char* buf = theSetBuf + theInlineSize;
+      Uint32 bytes = theGetSetBytes - theInlineSize;
+      assert(thePos == theInlineSize);
+      if (writeDataPrivate(buf, bytes) == -1)
+          return -1;
+    }
+  }
   if (isDeleteOp()) {
     assert(anExecType == NoCommit);
     getHeadFromRecAttr();
@@ -1278,7 +1425,7 @@ NdbBlob::postExecute(ExecType anExecType)
     if (invokeActiveHook() == -1)
       return -1;
   }
-  DBG("postExecute out");
+  DBG("postExecute [out]");
   return 0;
 }
 
@@ -1289,12 +1436,12 @@ NdbBlob::postExecute(ExecType anExecType)
 int
 NdbBlob::preCommit()
 {
-  DBG("preCommit");
+  DBG("preCommit [in]");
   if (theState == Invalid)
     return -1;
   assert(theState == Active);
   assert(isKeyOp());
-  if (isInsertOp() || isUpdateOp()) {
+  if (isInsertOp() || isUpdateOp() || isWriteOp()) {
     if (theHeadInlineUpdateFlag) {
         // add an operation to update head+inline
         NdbOperation* tOp = theNdbCon->getNdbOperation(theTable);
@@ -1305,9 +1452,11 @@ NdbBlob::preCommit()
           setErrorCode(ErrAbort);
           return -1;
         }
+        tOp->m_abortOption = AbortOnError;
+        DBG("added op to update head+inline");
     }
   }
-  DBG("preCommit out");
+  DBG("preCommit [out]");
   return 0;
 }
 
@@ -1317,13 +1466,10 @@ NdbBlob::preCommit()
 int
 NdbBlob::atNextResult()
 {
-  DBG("atNextResult");
+  DBG("atNextResult [in]");
   if (theState == Invalid)
     return -1;
   assert(isScanOp());
-  getHeadFromRecAttr();
-  // reset position
-  thePos = 0;
   // get primary key
   { Uint32* data = (Uint32*)theKeyBuf.data;
     unsigned size = theTable->m_sizeOfKeysInWords;
@@ -1332,26 +1478,14 @@ NdbBlob::atNextResult()
       return -1;
     }
   }
-  if (! theNullFlag) {
-    if (theGetFlag && theGetSetBytes > 0) {
-      // copy inline bytes to user buffer
-      assert(theGetBuf != NULL);
-      unsigned n = theGetSetBytes;
-      if (n > theLength)
-        n = theLength;
-      if (n > theInlineSize)
-        n = theInlineSize;
-      memcpy(theGetBuf, theInlineData, n);
-    }
-    if (theGetFlag && theGetSetBytes > theInlineSize && theLength > theInlineSize) {
-      // add ops to read rest of a getValue
-      assert(theGetBuf != 0);
-      Uint64 pos = theInlineSize;
-      char* buf = theGetBuf + theInlineSize;
-      Uint32 bytes = theGetSetBytes - theInlineSize;
-      if (readDataPrivate(pos, buf, bytes) == -1)
-        return -1;
-    }
+  getHeadFromRecAttr();
+  if (setPos(0) == -1)
+    return -1;
+  if (theGetFlag) {
+    assert(theGetSetBytes == 0 || theGetBuf != 0);
+    Uint32 bytes = theGetSetBytes;
+    if (readDataPrivate(theGetBuf, bytes) == -1)
+      return -1;
   }
   setState(Active);
   // activation callback
@@ -1359,7 +1493,7 @@ NdbBlob::atNextResult()
     if (invokeActiveHook() == -1)
       return -1;
   }
-  DBG("atNextResult out");
+  DBG("atNextResult [out]");
   return 0;
 }
 
@@ -1444,7 +1578,8 @@ operator<<(NdbOut& out, const NdbBlob& blob)
   ndbout << dec << " n=" << blob.theNullFlag;;
   ndbout << dec << " l=" << blob.theLength;
   ndbout << dec << " p=" << blob.thePos;
-  ndbout << dec << " u=" << (Uint32) blob.theHeadInlineUpdateFlag;
+  ndbout << dec << " u=" << (Uint32)blob.theHeadInlineUpdateFlag;
+  ndbout << dec << " g=" << (Uint32)blob.theGetSetBytes;
   return out;
 }
 #endif
diff --git a/ndb/src/ndbapi/NdbConnection.cpp b/ndb/src/ndbapi/NdbConnection.cpp
index 1457792cf28..c21a85fd24d 100644
--- a/ndb/src/ndbapi/NdbConnection.cpp
+++ b/ndb/src/ndbapi/NdbConnection.cpp
@@ -170,12 +170,14 @@ Remark:        Sets an error code on the connection object from an
                operation object. 
 *****************************************************************************/
 void
-NdbConnection::setOperationErrorCodeAbort(int error)
+NdbConnection::setOperationErrorCodeAbort(int error, int abortOption)
 {
   DBUG_ENTER("NdbConnection::setOperationErrorCodeAbort");
+  if (abortOption == -1)
+    abortOption = m_abortOption;
   if (theTransactionIsStarted == false) {
     theCommitStatus = Aborted;
-  } else if ((m_abortOption == AbortOnError) && 
+  } else if ((abortOption == AbortOnError) && 
 	     (theCommitStatus != Committed) &&
              (theCommitStatus != Aborted)) {
     theCommitStatus = NeedAbort;
@@ -335,8 +337,11 @@ NdbConnection::execute(ExecType aTypeOfExec,
         tOp = tOp->next();
       }
     }
+
     if (executeNoBlobs(tExecType, abortOption, forceSend) == -1)
         ret = -1;
+    assert(theFirstOpInList == NULL && theLastOpInList == NULL);
+
     {
       NdbOperation* tOp = theCompletedFirstOp;
       while (tOp != NULL) {
@@ -360,6 +365,7 @@ NdbConnection::execute(ExecType aTypeOfExec,
         theLastOpInList->next(tRestOp);
       theLastOpInList = tLastOp;
     }
+    assert(theFirstOpInList == NULL || tExecType == NoCommit);
   } while (theFirstOpInList != NULL || tExecType != aTypeOfExec);
 
   DBUG_RETURN(ret);
@@ -1806,11 +1812,12 @@ Parameters:    aErrorCode: The error code.
 Remark:        An operation was completed with failure.
 *******************************************************************************/
 int 
-NdbConnection::OpCompleteFailure(Uint8 abortOption)
+NdbConnection::OpCompleteFailure(Uint8 abortOption, bool setFailure)
 {
   Uint32 tNoComp = theNoOfOpCompleted;
   Uint32 tNoSent = theNoOfOpSent;
-  theCompletionStatus = NdbConnection::CompletedFailure;
+  if (setFailure)
+    theCompletionStatus = NdbConnection::CompletedFailure;
   tNoComp++;
   theNoOfOpCompleted = tNoComp;
   if (tNoComp == tNoSent) {
diff --git a/ndb/src/ndbapi/NdbDictionaryImpl.cpp b/ndb/src/ndbapi/NdbDictionaryImpl.cpp
index cf51a30fe0b..76854cabcd7 100644
--- a/ndb/src/ndbapi/NdbDictionaryImpl.cpp
+++ b/ndb/src/ndbapi/NdbDictionaryImpl.cpp
@@ -47,13 +47,15 @@
  * Column
  */
 NdbColumnImpl::NdbColumnImpl()
-  : NdbDictionary::Column(* this), m_facade(this)
+  : NdbDictionary::Column(* this), m_facade(this),
+    m_attrId(-1)
 {
   init();
 }
 
 NdbColumnImpl::NdbColumnImpl(NdbDictionary::Column & f)
-  : NdbDictionary::Column(* this), m_facade(&f)
+  : NdbDictionary::Column(* this), m_facade(&f),
+    m_attrId(-1)
 {
   init();
 }
@@ -93,8 +95,7 @@ NdbColumnImpl::init(Type t)
 {
   // do not use default_charset_info as it may not be initialized yet
   // use binary collation until NDB tests can handle charsets
-  CHARSET_INFO* default_cs = &my_charset_latin1_bin;
-  m_attrId = -1;
+  CHARSET_INFO* default_cs = &my_charset_bin;
   m_type = t;
   switch (m_type) {
   case Tinyint:
diff --git a/ndb/src/ndbapi/NdbIndexOperation.cpp b/ndb/src/ndbapi/NdbIndexOperation.cpp
index 9abde639914..3f174a61b64 100644
--- a/ndb/src/ndbapi/NdbIndexOperation.cpp
+++ b/ndb/src/ndbapi/NdbIndexOperation.cpp
@@ -71,6 +71,7 @@ NdbIndexOperation::indxInit(const NdbIndexImpl * anIndex,
     return -1;
   }
   m_theIndex = anIndex;
+  m_thePrimaryTable = aTable;
   m_accessTable = anIndex->m_table;
   m_theIndexLen = 0;
   m_theNoOfIndexDefined = 0;
@@ -102,6 +103,12 @@ int NdbIndexOperation::readTuple(NdbOperation::LockMode lm)
   };
 }
 
+int NdbIndexOperation::insertTuple()
+{
+  setErrorCode(4200);
+  return -1;
+}
+
 int NdbIndexOperation::readTuple()
 {
   // First check that index is unique
@@ -341,12 +348,11 @@ int NdbIndexOperation::equal_impl(const NdbColumnImpl* tAttrInfo,
       theDistrGroupIndicator = 1;
     }//if
     /**************************************************************************
-     *	If the operation is an insert request and the attribute is stored then
+     *	If the operation is a write request and the attribute is stored then
      *      we also set the value in the stored part through putting the 
      *      information in the INDXATTRINFO signals.
      *************************************************************************/
-    if ((tOpType == InsertRequest) ||
-	(tOpType == WriteRequest)) {
+    if ((tOpType == WriteRequest)) {
       if (!tAttrInfo->m_indexOnly){
         // invalid data can crash kernel
         if (cs != NULL &&
@@ -357,7 +363,13 @@ int NdbIndexOperation::equal_impl(const NdbColumnImpl* tAttrInfo,
           goto equal_error4;
 	Uint32 ahValue;
 	Uint32 sz = totalSizeInWords;
-	AttributeHeader::init(&ahValue, tAttrId, sz);
+        /*
+         * XXX should be linked in metadata but cannot now because
+         * things can be defined in arbitrary order
+         */
+        const NdbColumnImpl* primaryCol = m_thePrimaryTable->getColumn(tAttrInfo->m_name.c_str());
+        assert(primaryCol != NULL);
+	AttributeHeader::init(&ahValue, primaryCol->m_attrId, sz);
 	insertATTRINFO( ahValue );
 	insertATTRINFOloop((Uint32*)aValueToWrite, sizeInWords);
 	if (bitsInLastWord != 0) {
@@ -369,7 +381,6 @@ int NdbIndexOperation::equal_impl(const NdbColumnImpl* tAttrInfo,
 	}//if
       }//if
     }//if
-
     /**************************************************************************
      *	Store the Key information in the TCINDXREQ and INDXKEYINFO signals. 
      *************************************************************************/
@@ -734,13 +745,10 @@ NdbIndexOperation::receiveTCINDXREF( NdbApiSignal* aSignal)
   }//if
 
   theStatus = Finished;
-  
+
   theNdbCon->theReturnStatus = NdbConnection::ReturnFailure;
   Uint32 errorCode = tcIndxRef->errorCode;
   theError.code = errorCode;
   theNdbCon->setOperationErrorCodeAbort(errorCode);
   return theNdbCon->OpCompleteFailure(theNdbCon->m_abortOption);
 }//NdbIndexOperation::receiveTCINDXREF()
-
-
-
diff --git a/ndb/src/ndbapi/NdbOperation.cpp b/ndb/src/ndbapi/NdbOperation.cpp
index b0b95d0ff43..88d8a000d50 100644
--- a/ndb/src/ndbapi/NdbOperation.cpp
+++ b/ndb/src/ndbapi/NdbOperation.cpp
@@ -78,7 +78,8 @@ NdbOperation::NdbOperation(Ndb* aNdb) :
   m_tcReqGSN(GSN_TCKEYREQ),
   m_keyInfoGSN(GSN_KEYINFO),
   m_attrInfoGSN(GSN_ATTRINFO),
-  theBlobList(NULL)
+  theBlobList(NULL),
+  m_abortOption(-1)
 {
   theReceiver.init(NdbReceiver::NDB_OPERATION, this);
   theError.code = 0;
@@ -167,6 +168,7 @@ NdbOperation::init(const NdbTableImpl* tab, NdbConnection* myConnection){
   theTotalNrOfKeyWordInSignal = 8;
   theMagicNumber        = 0xABCDEF01;
   theBlobList = NULL;
+  m_abortOption = -1;
 
   tSignal = theNdb->getSignal();
   if (tSignal == NULL)
diff --git a/ndb/src/ndbapi/NdbOperationExec.cpp b/ndb/src/ndbapi/NdbOperationExec.cpp
index f1338ae01e4..fa46e93a57f 100644
--- a/ndb/src/ndbapi/NdbOperationExec.cpp
+++ b/ndb/src/ndbapi/NdbOperationExec.cpp
@@ -191,7 +191,8 @@ NdbOperation::prepareSend(Uint32 aTC_ConnectPtr, Uint64 aTransId)
   Uint8 tDirtyIndicator = theDirtyIndicator;
   OperationType tOperationType = theOperationType;
   Uint32 tTupKeyLen = theTupKeyLen;
-  Uint8 abortOption = theNdbCon->m_abortOption;
+  Uint8 abortOption =
+    m_abortOption != -1 ? m_abortOption : theNdbCon->m_abortOption;
 
   tcKeyReq->setDirtyFlag(tReqInfo, tDirtyIndicator);
   tcKeyReq->setOperationType(tReqInfo, tOperationType);
@@ -541,17 +542,20 @@ NdbOperation::receiveTCKEYREF( NdbApiSignal* aSignal)
     return -1;
   }//if
 
-  AbortOption ao = (AbortOption)theNdbCon->m_abortOption;
+  AbortOption ao = (AbortOption)
+    (m_abortOption != -1 ? m_abortOption : theNdbCon->m_abortOption);
   theReceiver.m_received_result_length = ~0;
 
   theStatus = Finished;
-  theNdbCon->theReturnStatus = NdbConnection::ReturnFailure;
+  // blobs want this
+  if (m_abortOption != IgnoreError)
+    theNdbCon->theReturnStatus = NdbConnection::ReturnFailure;
 
   theError.code = aSignal->readData(4);
-  theNdbCon->setOperationErrorCodeAbort(aSignal->readData(4));
+  theNdbCon->setOperationErrorCodeAbort(aSignal->readData(4), ao);
 
   if(theOperationType != ReadRequest || !theSimpleIndicator) // not simple read
-    return theNdbCon->OpCompleteFailure(ao);
+    return theNdbCon->OpCompleteFailure(ao, m_abortOption != IgnoreError);
   
   /**
    * If TCKEYCONF has arrived
diff --git a/ndb/test/ndbapi/testBlobs.cpp b/ndb/test/ndbapi/testBlobs.cpp
index 41bb82f3e06..f7ee7921229 100644
--- a/ndb/test/ndbapi/testBlobs.cpp
+++ b/ndb/test/ndbapi/testBlobs.cpp
@@ -48,7 +48,7 @@ struct Opt {
   unsigned m_rows;
   unsigned m_seed;
   const char* m_skip;
-  const char* m_style;
+  const char* m_test;
   // metadata
   const char* m_tname;
   const char* m_x1name;  // hash index
@@ -71,8 +71,8 @@ struct Opt {
     m_parts(10),
     m_rows(100),
     m_seed(0),
-    m_skip(""),
-    m_style("012"),
+    m_skip(0),
+    m_test(0),
     // metadata
     m_tname("TBLOB1"),
     m_x1name("TBLOB1X1"),
@@ -101,45 +101,44 @@ printusage()
     << "  -dbg        print debug" << endl
     << "  -dbgall     print also NDB API debug (if compiled in)" << endl
     << "  -full       read/write only full blob values" << endl
-    << "  -inline     read/write only blobs which fit inline" << endl
     << "  -loop N     loop N times 0=forever [" << d.m_loop << "]" << endl
     << "  -parts N    max parts in blob value [" << d.m_parts << "]" << endl
     << "  -rows N     number of rows [" << d.m_rows << "]" << endl
     << "  -seed N     random seed 0=loop number [" << d.m_seed << "]" << endl
-    << "  -skip xxx   skip these tests (see list) [" << d.m_skip << endl
-    << "  -style xxx  access styles to test (see list) [" << d.m_style << "]" << endl
+    << "  -skip xxx   skip given tests (see list) [no tests]" << endl
+    << "  -test xxx   only given tests (see list) [all tests]" << endl
     << "metadata" << endl
     << "  -pk2len N   length of PK2 [" << d.m_pk2len << "/" << g_max_pk2len <<"]" << endl
     << "  -oneblob    only 1 blob attribute [default 2]" << endl
-    << "testcases for -skip" << endl
+    << "testcases for test/skip" << endl
     << "  k           primary key ops" << endl
     << "  i           hash index ops" << endl
     << "  s           table scans" << endl
     << "  r           ordered index scans" << endl
-    << "  u           update blob value" << endl
-    << "access styles for -style" << endl
+    << "additional flags for test/skip" << endl
+    << "  u           update existing blob value" << endl
+    << "  n           normal insert and update" << endl
+    << "  w           insert and update using writeTuple" << endl
     << "  0           getValue / setValue" << endl
     << "  1           setActiveHook" << endl
     << "  2           readData / writeData" << endl
     << "bug tests (no blob test)" << endl
     << "  -bug 4088   ndb api hang with mixed ops on index table" << endl
-    << "  -bug 2222   delete + write gives 626" << endl
-    << "  -bug 3333   acc crash on delete and long key" << endl
+    << "  -bug nnnn   delete + write gives 626" << endl
+    << "  -bug nnnn   acc crash on delete and long key" << endl
     ;
 }
 
 static Opt g_opt;
 
 static bool
-skipcase(int x)
+testcase(char x)
 {
-  return strchr(g_opt.m_skip, x) != 0;
-}
-
-static bool
-skipstyle(int x)
-{
-  return strchr(g_opt.m_style, '0' + x) == 0;
+  if (x < 10)
+    x += '0';
+  return
+    (g_opt.m_test == 0 || strchr(g_opt.m_test, x) != 0) &&
+    (g_opt.m_skip == 0 || strchr(g_opt.m_skip, x) == 0);
 }
 
 static Ndb* g_ndb = 0;
@@ -435,7 +434,9 @@ getBlobLength(NdbBlob* h, unsigned& len)
   CHK(h->getLength(len2) == 0);
   len = (unsigned)len2;
   assert(len == len2);
-  DBG("getBlobLength " << h->getColumn()->getName() << " len=" << len);
+  bool isNull;
+  CHK(h->getNull(isNull) == 0);
+  DBG("getBlobLength " << h->getColumn()->getName() << " len=" << len << " null=" << isNull);
   return 0;
 }
 
@@ -912,6 +913,41 @@ updatePk(int style)
 }
 
 static int
+writePk(int style)
+{
+  DBG("--- writePk " << stylename[style] << " ---");
+  for (unsigned k = 0; k < g_opt.m_rows; k++) {
+    Tup& tup = g_tups[k];
+    DBG("writePk pk1=" << hex << tup.m_pk1);
+    CHK((g_con = g_ndb->startTransaction()) != 0);
+    CHK((g_opr = g_con->getNdbOperation(g_opt.m_tname)) != 0);
+    CHK(g_opr->writeTuple() == 0);
+    CHK(g_opr->equal("PK1", tup.m_pk1) == 0);
+    if (g_opt.m_pk2len != 0)
+      CHK(g_opr->equal("PK2", tup.m_pk2) == 0);
+    CHK(getBlobHandles(g_opr) == 0);
+    if (style == 0) {
+      CHK(setBlobValue(tup) == 0);
+    } else if (style == 1) {
+      // non-nullable must be set
+      CHK(g_bh1->setValue("", 0) == 0);
+      CHK(setBlobWriteHook(tup) == 0);
+    } else {
+      // non-nullable must be set
+      CHK(g_bh1->setValue("", 0) == 0);
+      CHK(g_con->execute(NoCommit) == 0);
+      CHK(writeBlobData(tup) == 0);
+    }
+    CHK(g_con->execute(Commit) == 0);
+    g_ndb->closeTransaction(g_con);
+    g_opr = 0;
+    g_con = 0;
+    tup.m_exists = true;
+  }
+  return 0;
+}
+
+static int
 deletePk()
 {
   DBG("--- deletePk ---");
@@ -996,6 +1032,35 @@ updateIdx(int style)
 }
 
 static int
+writeIdx(int style)
+{
+  DBG("--- writeIdx " << stylename[style] << " ---");
+  for (unsigned k = 0; k < g_opt.m_rows; k++) {
+    Tup& tup = g_tups[k];
+    DBG("writeIdx pk1=" << hex << tup.m_pk1);
+    CHK((g_con = g_ndb->startTransaction()) != 0);
+    CHK((g_opx = g_con->getNdbIndexOperation(g_opt.m_x1name, g_opt.m_tname)) != 0);
+    CHK(g_opx->writeTuple() == 0);
+    CHK(g_opx->equal("PK2", tup.m_pk2) == 0);
+    CHK(getBlobHandles(g_opx) == 0);
+    if (style == 0) {
+      CHK(setBlobValue(tup) == 0);
+    } else if (style == 1) {
+      CHK(setBlobWriteHook(tup) == 0);
+    } else {
+      CHK(g_con->execute(NoCommit) == 0);
+      CHK(writeBlobData(tup) == 0);
+    }
+    CHK(g_con->execute(Commit) == 0);
+    g_ndb->closeTransaction(g_con);
+    g_opx = 0;
+    g_con = 0;
+    tup.m_exists = true;
+  }
+  return 0;
+}
+
+static int
 deleteIdx()
 {
   DBG("--- deleteIdx ---");
@@ -1170,7 +1235,6 @@ deleteScan(bool idx)
 static int
 testmain()
 {
-  int style;
   g_ndb = new Ndb("TEST_DB");
   CHK(g_ndb->init() == 0);
   CHK(g_ndb->waitUntilReady() == 0);
@@ -1194,55 +1258,88 @@ testmain()
   if (g_opt.m_seed != 0)
     srandom(g_opt.m_seed);
   for (g_loop = 0; g_opt.m_loop == 0 || g_loop < g_opt.m_loop; g_loop++) {
+    int style;
     DBG("=== loop " << g_loop << " ===");
     if (g_opt.m_seed == 0)
       srandom(g_loop);
     // pk
     for (style = 0; style <= 2; style++) {
-      if (skipcase('k') || skipstyle(style))
+      if (! testcase('k') || ! testcase(style))
         continue;
       DBG("--- pk ops " << stylename[style] << " ---");
-      calcTups(false);
-      CHK(insertPk(style) == 0);
-      CHK(verifyBlob() == 0);
-      CHK(readPk(style) == 0);
-      if (! skipcase('u')) {
-        calcTups(style);
-        CHK(updatePk(style) == 0);
+      if (testcase('n')) {
+        calcTups(false);
+        CHK(insertPk(style) == 0);
+        CHK(verifyBlob() == 0);
+        CHK(readPk(style) == 0);
+        if (testcase('u')) {
+          calcTups(style);
+          CHK(updatePk(style) == 0);
+          CHK(verifyBlob() == 0);
+          CHK(readPk(style) == 0);
+        }
+        CHK(deletePk() == 0);
+        CHK(verifyBlob() == 0);
+      }
+      if (testcase('w')) {
+        calcTups(false);
+        CHK(writePk(style) == 0);
+        CHK(verifyBlob() == 0);
+        CHK(readPk(style) == 0);
+        if (testcase('u')) {
+          calcTups(style);
+          CHK(writePk(style) == 0);
+          CHK(verifyBlob() == 0);
+          CHK(readPk(style) == 0);
+        }
+        CHK(deletePk() == 0);
         CHK(verifyBlob() == 0);
       }
-      CHK(readPk(style) == 0);
-      CHK(deletePk() == 0);
-      CHK(verifyBlob() == 0);
     }
     // hash index
     for (style = 0; style <= 2; style++) {
-      if (skipcase('i') || skipstyle(style))
+      if (! testcase('i') || ! testcase(style))
         continue;
       DBG("--- idx ops " << stylename[style] << " ---");
-      calcTups(false);
-      CHK(insertPk(style) == 0);
-      CHK(verifyBlob() == 0);
-      CHK(readIdx(style) == 0);
-      calcTups(style);
-      if (! skipcase('u')) {
-        CHK(updateIdx(style) == 0);
+      if (testcase('n')) {
+        calcTups(false);
+        CHK(insertPk(style) == 0);
         CHK(verifyBlob() == 0);
         CHK(readIdx(style) == 0);
+        if (testcase('u')) {
+          calcTups(style);
+          CHK(updateIdx(style) == 0);
+          CHK(verifyBlob() == 0);
+          CHK(readIdx(style) == 0);
+        }
+        CHK(deleteIdx() == 0);
+        CHK(verifyBlob() == 0);
+      }
+      if (testcase('w')) {
+        calcTups(false);
+        CHK(writePk(style) == 0);
+        CHK(verifyBlob() == 0);
+        CHK(readIdx(style) == 0);
+        if (testcase('u')) {
+          calcTups(style);
+          CHK(writeIdx(style) == 0);
+          CHK(verifyBlob() == 0);
+          CHK(readIdx(style) == 0);
+        }
+        CHK(deleteIdx() == 0);
+        CHK(verifyBlob() == 0);
       }
-      CHK(deleteIdx() == 0);
-      CHK(verifyBlob() == 0);
     }
     // scan table
     for (style = 0; style <= 2; style++) {
-      if (skipcase('s') || skipstyle(style))
+      if (! testcase('s') || ! testcase(style))
         continue;
       DBG("--- table scan " << stylename[style] << " ---");
       calcTups(false);
       CHK(insertPk(style) == 0);
       CHK(verifyBlob() == 0);
       CHK(readScan(style, false) == 0);
-      if (! skipcase('u')) {
+      if (testcase('u')) {
         CHK(updateScan(style, false) == 0);
         CHK(verifyBlob() == 0);
       }
@@ -1251,14 +1348,14 @@ testmain()
     }
     // scan index
     for (style = 0; style <= 2; style++) {
-      if (skipcase('r') || skipstyle(style))
+      if (! testcase('r') || ! testcase(style))
         continue;
       DBG("--- index scan " << stylename[style] << " ---");
       calcTups(false);
       CHK(insertPk(style) == 0);
       CHK(verifyBlob() == 0);
       CHK(readScan(style, true) == 0);
-      if (! skipcase('u')) {
+      if (testcase('u')) {
         CHK(updateScan(style, true) == 0);
         CHK(verifyBlob() == 0);
       }
@@ -1331,9 +1428,7 @@ static struct {
   int m_bug;
   int (*m_test)();
 } g_bugtest[] = {
-  { 4088, bugtest_4088 },
-  { 2222, bugtest_2222 },
-  { 3333, bugtest_3333 }
+  { 4088, bugtest_4088 }
 };
 
 NDB_COMMAND(testOdbcDriver, "testBlobs", "testBlobs", "testBlobs", 65535)
@@ -1395,9 +1490,9 @@ NDB_COMMAND(testOdbcDriver, "testBlobs", "testBlobs", "testBlobs", 65535)
 	continue;
       }
     }
-    if (strcmp(arg, "-style") == 0) {
+    if (strcmp(arg, "-test") == 0) {
       if (++argv, --argc > 0) {
-        g_opt.m_style = strdup(argv[0]);
+        g_opt.m_test = strdup(argv[0]);
 	continue;
       }
     }
@@ -1433,7 +1528,9 @@ NDB_COMMAND(testOdbcDriver, "testBlobs", "testBlobs", "testBlobs", 65535)
   }
   if (g_opt.m_pk2len == 0) {
     char b[100];
-    strcpy(b, g_opt.m_skip);
+    b[0] = 0;
+    if (g_opt.m_skip != 0)
+      strcpy(b, g_opt.m_skip);
     strcat(b, "i");
     strcat(b, "r");
     g_opt.m_skip = strdup(b);