Enabled HA_NULL_IN_KEY support

mysql-test/r/ndb_index_ordered.result

@@ -212,3 +212,48 @@ select count(*) from t1 where b = 1;
 count(*)
 1
 drop table t1;
+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) where b IS NULL;
+a	b	c
+3	NULL	NULL
+2	NULL	2
+select * from t1 use index (bc)order by a;
+a	b	c
+1	1	1
+2	NULL	2
+3	NULL	NULL
+4	4	NULL
+select * from t1 use index (bc) order by a;
+a	b	c
+1	1	1
+2	NULL	2
+3	NULL	NULL
+4	4	NULL
+select * from t1 use index (PRIMARY) where b IS NULL order by a;
+a	b	c
+2	NULL	2
+3	NULL	NULL
+select * from t1 use index (bc) where b IS NULL order by a;
+a	b	c
+2	NULL	2
+3	NULL	NULL
+select * from t1 use index (bc) where b IS NULL and c IS NULL order by a;
+a	b	c
+3	NULL	NULL
+select * from t1 use index (bc) where b IS NULL and c = 2 order by a;
+a	b	c
+2	NULL	2
+select * from t1 use index (bc) where b < 4 order by a;
+a	b	c
+1	1	1
+select * from t1 use index (bc) where b IS NOT NULL order by a;
+a	b	c
+1	1	1
+4	4	NULL
+drop table t1;

mysql-test/r/ndb_index_unique.result

@@ -109,6 +109,68 @@ a	b	c
 3	4	6
 drop table t3;
 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;
+pk	a
+-1	NULL
+0	0
+1	NULL
+2	2
+3	NULL
+4	4
+insert into t1 values (5,0);
+ERROR 23000: Can't write, because of unique constraint, to table 't1'
+select * from t1 order by pk;
+pk	a
+-1	NULL
+0	0
+1	NULL
+2	2
+3	NULL
+4	4
+delete from t1 where a = 0;
+insert into t1 values (5,0);
+select * from t1 order by pk;
+pk	a
+-1	NULL
+1	NULL
+2	2
+3	NULL
+4	4
+5	0
+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;
+pk	a	b	c
+-1	1	17	NULL
+0	NULL	18	NULL
+1	3	19	abc
+insert into t2 values(2,3,19,'abc');
+ERROR 23000: Can't write, because of unique constraint, to table 't2'
+select * from t2 order by pk;
+pk	a	b	c
+-1	1	17	NULL
+0	NULL	18	NULL
+1	3	19	abc
+delete from t2 where c IS NOT NULL;
+insert into t2 values(2,3,19,'abc');
+select * from t2 order by pk;
+pk	a	b	c
+-1	1	17	NULL
+0	NULL	18	NULL
+2	3	19	abc
+drop table t1, t2;
+CREATE TABLE t1 (
 cid smallint(5) unsigned NOT NULL default '0',
 cv varchar(250) NOT NULL default '',
 PRIMARY KEY  (cid),

mysql-test/t/ndb_index_ordered.test

@@ -122,18 +122,22 @@ 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;
+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) where b IS NULL;
+
+select * from t1 use index (bc)order by a;
+select * from t1 use index (bc) order by a;
+select * from t1 use index (PRIMARY) where b IS NULL order by a;
+select * from t1 use index (bc) where b IS NULL order by a;
+select * from t1 use index (bc) where b IS NULL and c IS NULL order by a;
+select * from t1 use index (bc) where b IS NULL and c = 2 order by a;
+select * from t1 use index (bc) where b < 4 order by a;
+select * from t1 use index (bc) where b IS NOT NULL order by a;
+drop table t1;

mysql-test/t/ndb_index_unique.test

@@ -82,43 +82,43 @@ 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;
+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

sql/ha_ndbcluster.cc

@@ -1075,11 +1075,13 @@ int ha_ndbcluster::set_bounds(NdbIndexScanOperation *op,
 			      const key_range *key,
 			      int bound)
 {
-  uint i, tot_len;
+  uint key_len, key_store_len, tot_len, key_tot_len;
   byte *key_ptr;
   KEY* key_info= table->key_info + active_index;
   KEY_PART_INFO* key_part= key_info->key_part;
   KEY_PART_INFO* end= key_part+key_info->key_parts;
+  Field* field;
+  bool key_nullable, key_null;
 
   DBUG_ENTER("set_bounds");
   DBUG_PRINT("enter", ("bound: %d", bound));
@@ -1090,28 +1092,36 @@ int ha_ndbcluster::set_bounds(NdbIndexScanOperation *op,
   // Set bounds using key data
   tot_len= 0;
   key_ptr= (byte *) key->key;
+  key_tot_len= key->length;
   for (; key_part != end; key_part++)
   {
-    Field* field= key_part->field;
-    uint32 field_len=  field->pack_length();
-    tot_len+= field_len;
+    field= key_part->field;
+    key_len=  key_part->length;
+    key_store_len=  key_part->store_length;
+    key_nullable= (bool) key_part->null_bit;
+    key_null= (field->maybe_null() && *key_ptr);
+    tot_len+= key_store_len;
 
     const char* bounds[]= {"LE", "LT", "GE", "GT", "EQ"};
     DBUG_ASSERT(bound >= 0 && bound <= 4);    
-    DBUG_PRINT("info", ("Set Bound%s on %s", 
+    DBUG_PRINT("info", ("Set Bound%s on %s %s %s %s", 
 			bounds[bound],
-			field->field_name));
-    DBUG_DUMP("key", (char*)key_ptr, field_len);
+			field->field_name,
+			key_nullable ? "NULLABLE" : "",
+			key_null ? "NULL":""));
+    DBUG_PRINT("info", ("Total length %ds", tot_len));
+    
+    DBUG_DUMP("key", (char*) key_ptr, key_store_len);
     
     if (op->setBound(field->field_name,
 		     bound, 
-		     field->is_null() ? 0 : key_ptr,
-		     field->is_null() ? 0 : field_len) != 0)
+		     key_null ? 0 : (key_nullable ? key_ptr + 1 : key_ptr),
+		     key_null ? 0 : key_len) != 0)
       ERR_RETURN(op->getNdbError());
     
-    key_ptr+= field_len;
+    key_ptr+= key_store_len;
 
-    if (tot_len >= key->length)
+    if (tot_len >= key_tot_len)
       break;
 
     /*
@@ -3104,7 +3114,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_NULL_IN_KEY |
                 HA_NOT_EXACT_COUNT |
                 HA_NO_PREFIX_CHAR_KEYS),
   m_use_write(false),