Added support for NULL in unique index

mysql-test/r/ndb_index_unique.result

@@ -21,6 +21,28 @@ insert into t1 values(7,8,3);
 select * from t1 where b = 4 order by a;
 a	b	c
 3	4	6
+insert into t1 values(8, 2, 3);
+ERROR 23000: Can't write, because of unique constraint, to table 't1'
+select * from t1 order by a;
+a	b	c
+1	2	3
+2	3	5
+3	4	6
+4	5	8
+5	6	2
+6	7	2
+7	8	3
+delete from t1 where a = 1;
+insert into t1 values(8, 2, 3);
+select * from t1 order by a;
+a	b	c
+2	3	5
+3	4	6
+4	5	8
+5	6	2
+6	7	2
+7	8	3
+8	2	3
 drop table t1;
 CREATE TABLE t2 (
 a int unsigned NOT NULL PRIMARY KEY,
@@ -42,6 +64,28 @@ insert into t2 values(7,8,3);
 select * from t2 where b = 4 order by a;
 a	b	c
 3	4	6
+insert into t2 values(8, 2, 3);
+ERROR 23000: Can't write, because of unique constraint, to table 't2'
+select * from t2 order by a;
+a	b	c
+1	2	3
+2	3	5
+3	4	6
+4	5	8
+5	6	2
+6	7	2
+7	8	3
+delete from t2 where a = 1;
+insert into t2 values(8, 2, 3);
+select * from t2 order by a;
+a	b	c
+2	3	5
+3	4	6
+4	5	8
+5	6	2
+6	7	2
+7	8	3
+8	2	3
 drop table t2;
 CREATE TABLE t3 (
 a int unsigned NOT NULL,
@@ -74,8 +118,10 @@ INSERT INTO t1 VALUES (8,'dummy');
 CREATE TABLE t2 (
 cid bigint(20) unsigned NOT NULL auto_increment,
 cap varchar(255) NOT NULL default '',
-PRIMARY KEY  (cid)
+PRIMARY KEY  (cid),
+UNIQUE KEY (cid, cap)
 ) engine=ndbcluster;
+INSERT INTO t2 VALUES (NULL,'another dummy');
 CREATE TABLE t3 (
 gid bigint(20) unsigned NOT NULL auto_increment,
 gn varchar(255) NOT NULL default '',
@@ -132,6 +178,9 @@ cid	cv
 8	dummy
 select * from t1 where cv = 'test';
 cid	cv
+select * from t2 where cap = 'another dummy';
+cid	cap
+0	another dummy
 select * from t4 where uid = 1 and gid=1 and rid=2 and cid=4;
 uid	gid	rid	cid
 1	1	2	4

mysql-test/t/ndb_index_ordered.test

@@ -114,3 +114,23 @@ select * from t1 where b=4 and c<=5 order by a;
 select * from t1 where b<=4 and c<=5 order by a;
 select * from t1 where b<=5 and c=0 or b<=5 and c=2;
 drop table t1;
+
+#
+# Indexing NULL values
+#
+
+#CREATE TABLE t1 (
+#  a int unsigned NOT NULL PRIMARY KEY,
+#  b int unsigned,
+#  c int unsigned,
+#  KEY bc(b,c)
+#) engine = ndb;
+
+#insert into t1 values(1,1,1),(2,NULL,2),(3,NULL,NULL),(4,4,NULL);
+#select * from t1 use index (bc);
+#select count(*) from t1 use index (bc);
+#select count(*) from t1 use index (PRIMARY) where b IS NULL;
+#select count(*) from t1 use index (bc) where b IS NULL;
+#select count(*) from t1 use index (bc) where b IS NULL and c = 2;
+#select count(*) from t1 use index (bc) where b IS NOT NULL;
+#drop table t1;

mysql-test/t/ndb_index_unique.test

@@ -21,6 +21,13 @@ select * from t1 where b = 4 order by b;
 insert into t1 values(7,8,3);
 select * from t1 where b = 4 order by a;
 
+-- error 1169
+insert into t1 values(8, 2, 3);
+select * from t1 order by a;
+delete from t1 where a = 1;
+insert into t1 values(8, 2, 3);
+select * from t1 order by a;
+
 drop table t1;
 
 
@@ -42,6 +49,13 @@ select * from t2 where c = 6;
 insert into t2 values(7,8,3);
 select * from t2 where b = 4 order by a;
 
+-- error 1169
+insert into t2 values(8, 2, 3);
+select * from t2 order by a;
+delete from t2 where a = 1;
+insert into t2 values(8, 2, 3);
+select * from t2 order by a;
+
 drop table t2;
 
 #
@@ -64,6 +78,48 @@ select * from t3 where b = 4 order by a;
 
 drop table t3;
 
+#
+# Indexes on NULL-able columns 
+#
+
+#CREATE TABLE t1 (
+#  pk int NOT NULL PRIMARY KEY,
+#  a int unsigned,
+#  UNIQUE KEY (a)
+#) engine=ndbcluster;
+		 
+#insert into t1 values (-1,NULL), (0,0), (1,NULL),(2,2),(3,NULL),(4,4);
+
+#select * from t1 order by pk;
+
+#--error 1169
+#insert into t1 values (5,0);
+#select * from t1 order by pk;
+#delete from t1 where a = 0;
+#insert into t1 values (5,0);
+#select * from t1 order by pk;
+
+#CREATE TABLE t2 (
+#  pk int NOT NULL PRIMARY KEY,
+#  a int unsigned,
+#  b tinyint NOT NULL,
+#  c VARCHAR(10),
+#  UNIQUE KEY si(a, c)
+#) engine=ndbcluster;
+
+#insert into t2 values (-1,1,17,NULL),(0,NULL,18,NULL),(1,3,19,'abc');
+
+#select * from t2 order by pk;
+
+#--error 1169
+#insert into t2 values(2,3,19,'abc');
+#select * from t2 order by pk;
+#delete from t2 where c IS NOT NULL;
+#insert into t2 values(2,3,19,'abc');
+#select * from t2 order by pk;
+		 
+#drop table t1, t2;
+
 #
 # More complex tables
 #
@@ -78,8 +134,10 @@ INSERT INTO t1 VALUES (8,'dummy');
 CREATE TABLE t2 (
   cid bigint(20) unsigned NOT NULL auto_increment,
   cap varchar(255) NOT NULL default '',
-  PRIMARY KEY  (cid)
+  PRIMARY KEY  (cid),
+  UNIQUE KEY (cid, cap)
 ) engine=ndbcluster;
+INSERT INTO t2 VALUES (NULL,'another dummy');
 CREATE TABLE t3 (
   gid bigint(20) unsigned NOT NULL auto_increment,
   gn varchar(255) NOT NULL default '',
@@ -134,6 +192,7 @@ INSERT INTO t7 VALUES(10, 5, 1, 1, 10);
 
 select * from t1 where cv = 'dummy';
 select * from t1 where cv = 'test';
+select * from t2 where cap = 'another dummy';
 select * from t4 where uid = 1 and gid=1 and rid=2 and cid=4;
 select * from t4 where uid = 1 and gid=1 and rid=1 and cid=4;
 select * from t4 where uid = 1 order by cid;

ndb/src/kernel/blocks/dbdict/Dbdict.cpp

@@ -6255,16 +6255,6 @@ Dbdict::createIndex_toCreateTable(Signal* signal, OpCreateIndexPtr opPtr)
       jam();
       found = true;
       const Uint32 a = aRec->attributeDescriptor;
-      bool isNullable = AttributeDescriptor::getNullable(a);
-      // We do not allow more than one NULLable attribute for hash index
-      if (isNullable && 
-	  indexPtr.p->isHashIndex() && 
-	  (opPtr.p->m_attrList.sz > 1)) {
-        jam();
-        opPtr.p->m_errorCode = CreateIndxRef::AttributeNullable;
-        opPtr.p->m_errorLine = __LINE__;
-        return;
-      }
       if (indexPtr.p->isHashIndex()) {
         const Uint32 s1 = AttributeDescriptor::getSize(a);
         const Uint32 s2 = AttributeDescriptor::getArraySize(a);

ndb/src/kernel/blocks/dbtc/Dbtc.hpp

@@ -139,6 +139,7 @@
 #define ZNOT_FOUND 626
 #define ZALREADYEXIST 630
 #define ZINCONSISTENTHASHINDEX 892
+#define ZNOTUNIQUE 893
 #endif
 
 class Dbtc: public SimulatedBlock {

ndb/src/kernel/blocks/dbtc/DbtcMain.cpp

@@ -4925,7 +4925,9 @@ void Dbtc::execLQHKEYREF(Signal* signal)
 	
 	// The operation executed an index trigger
 	const Uint32 opType = regTcPtr->operation;
-	if (!(opType == ZDELETE && errCode == ZNOT_FOUND)) {
+	if (errCode == ZALREADYEXIST)
+	  errCode = terrorCode = ZNOTUNIQUE;
+	else if (!(opType == ZDELETE && errCode == ZNOT_FOUND)) {
 	  jam();
 	  /**
 	   * "Normal path"
@@ -12168,22 +12170,22 @@ void Dbtc::insertIntoIndexTable(Signal* signal,
   // Calculate key length and renumber attribute id:s
   AttributeBuffer::DataBufferPool & pool = c_theAttributeBufferPool;
   LocalDataBuffer<11> afterValues(pool, firedTriggerData->afterValues);
+  bool skipNull = false;
   for(bool moreKeyAttrs = afterValues.first(iter); moreKeyAttrs; attrId++) {
     jam();
     AttributeHeader* attrHeader = (AttributeHeader *) iter.data;
 
+    // Filter out NULL valued attributes
+    if (attrHeader->isNULL()) {
+      skipNull = true;
+      break;
+    }
     attrHeader->setAttributeId(attrId);      
     keyLength += attrHeader->getDataSize();
     hops = attrHeader->getHeaderSize() + attrHeader->getDataSize();
     moreKeyAttrs = afterValues.next(iter, hops);
   }
-
+  if (skipNull) {
-  // Filter out single NULL attributes
-  if (attrId == 1) {
-    jam();
-    afterValues.first(iter);
-    AttributeHeader* attrHeader = (AttributeHeader *) iter.data;
-    if (attrHeader->isNULL() && !afterValues.next(iter)) {
     jam();
     opRecord->triggerExecutionCount--;
     if (opRecord->triggerExecutionCount == 0) {
@@ -12196,7 +12198,6 @@ void Dbtc::insertIntoIndexTable(Signal* signal,
     }//if
     return;
   }//if
-  }//if
 
   // Calculate total length of primary key to be stored in index table
   LocalDataBuffer<11> keyValues(pool, firedTriggerData->keyValues);
@@ -12523,24 +12524,25 @@ void Dbtc::deleteFromIndexTable(Signal* signal,
   // Calculate key length and renumber attribute id:s
   AttributeBuffer::DataBufferPool & pool = c_theAttributeBufferPool;
   LocalDataBuffer<11> beforeValues(pool, firedTriggerData->beforeValues);
+  bool skipNull = false;
   for(bool moreKeyAttrs = beforeValues.first(iter);
       (moreKeyAttrs);
       attrId++) {
     jam();
     AttributeHeader* attrHeader = (AttributeHeader *) iter.data;
     
+    // Filter out NULL valued attributes
+    if (attrHeader->isNULL()) {
+      skipNull = true;
+      break;
+    }
     attrHeader->setAttributeId(attrId);      
     keyLength += attrHeader->getDataSize();
     hops = attrHeader->getHeaderSize() + attrHeader->getDataSize();
     moreKeyAttrs = beforeValues.next(iter, hops);
   }
 
-  // Filter out single NULL attributes
+  if (skipNull) {
-  if (attrId == 1) {
-    jam();
-    beforeValues.first(iter);
-    AttributeHeader* attrHeader = (AttributeHeader *) iter.data;
-    if (attrHeader->isNULL() && !beforeValues.next(iter)) {
     jam();
     opRecord->triggerExecutionCount--;
     if (opRecord->triggerExecutionCount == 0) {
@@ -12553,7 +12555,6 @@ void Dbtc::deleteFromIndexTable(Signal* signal,
     }//if
     return;
   }//if
-  }//if
 
   TcKeyReq::setKeyLength(tcKeyRequestInfo, keyLength);
   tcKeyReq->attrLen = 0;

ndb/src/kernel/blocks/trix/Trix.cpp

@@ -814,8 +814,8 @@ void Trix::executeInsertTransaction(Signal* signal,
   for(Uint32 i = 0; i < headerPtr.sz; i++) {
     AttributeHeader* keyAttrHead = (AttributeHeader *) headerBuffer + i;
 
-    // Filter out single NULL attributes
+    // Filter out NULL attributes
-    if (keyAttrHead->isNULL() && (i == (Uint32)0) && (headerPtr.sz == (Uint32)2))
+    if (keyAttrHead->isNULL())
       return;
 
     if (i < subRec->noOfIndexColumns)

ndb/src/ndbapi/NdbDictionaryImpl.cpp

@@ -1851,13 +1851,6 @@ NdbDictInterface::createIndex(Ndb & ndb,
       m_error.code = 4245;
       return -1;
     }
-
-    if (it == DictTabInfo::UniqueHashIndex &&
-        (col->m_nullable) && (attributeList.sz > 1)) {
-      // We only support one NULL attribute
-      m_error.code = 4246;
-      return -1;
-    }
     attributeList.id[i] = col->m_attrId;
   }
   if (it == DictTabInfo::UniqueHashIndex) {

sql/ha_ndbcluster.cc

@@ -1043,8 +1043,8 @@ int ha_ndbcluster::set_bounds(NdbIndexScanOperation *op,
     
     if (op->setBound(field->field_name,
 		     bound, 
-		     key_ptr,
+		     field->is_null() ? 0 : key_ptr,
-		     field_len) != 0)
+		     field->is_null() ? 0 : field_len) != 0)
       ERR_RETURN(op->getNdbError());
     
     key_ptr+= field_len;
@@ -1293,8 +1293,6 @@ int ha_ndbcluster::write_row(byte *record)
     update_timestamp(record+table->timestamp_default_now-1);
   has_auto_increment= (table->next_number_field && record == table->record[0]);
   skip_auto_increment= table->auto_increment_field_not_null;
-  if ((has_auto_increment) && (!skip_auto_increment))
-    update_auto_increment();
 
   if (!(op= trans->getNdbOperation(m_tabname)))
     ERR_RETURN(trans->getNdbError());
@@ -1313,6 +1311,10 @@ int ha_ndbcluster::write_row(byte *record)
   else 
   {
     int res;
+
+    if ((has_auto_increment) && (!skip_auto_increment))
+      update_auto_increment();
+
     if ((res= set_primary_key(op)))
       return res;
   }
@@ -1323,8 +1325,11 @@ int ha_ndbcluster::write_row(byte *record)
     Field *field= table->field[i];
     if (!(field->flags & PRI_KEY_FLAG) &&
 	set_ndb_value(op, field, i))
+    {
+      skip_auto_increment= true;
       ERR_RETURN(op->getNdbError());
     }
+  }
 
   /*
     Execute write operation
@@ -1345,8 +1350,11 @@ int ha_ndbcluster::write_row(byte *record)
 			(int)rows_inserted, (int)bulk_insert_rows)); 
     bulk_insert_not_flushed= false;
     if (trans->execute(NoCommit) != 0)
+    {
+      skip_auto_increment= true;
       DBUG_RETURN(ndb_err(trans));
     }
+  }
   if ((has_auto_increment) && (skip_auto_increment))
   {
     Uint64 next_val= (Uint64) table->next_number_field->val_int() + 1;
@@ -3068,6 +3076,7 @@ ha_ndbcluster::ha_ndbcluster(TABLE *table_arg):
   m_ndb(NULL),
   m_table(NULL),
   m_table_flags(HA_REC_NOT_IN_SEQ |
+		//HA_NULL_IN_KEY |
                 HA_NOT_EXACT_COUNT |
                 HA_NO_PREFIX_CHAR_KEYS),
   m_use_write(false),