table, executor: fix a unique key with null (#6032)

executor/executor_test.go

@@ -2543,6 +2543,17 @@ func setColValue(c *C, txn kv.Transaction, key kv.Key, v types.Datum) {
 	c.Assert(err, IsNil)
 }
 
+func (s *testSuite) TestCheckTable(c *C) {
+	tk := testkit.NewTestKit(c, s.store)
+
+	// Test 'admin check table' when the table has a unique index with null values.
+	tk.MustExec("use test")
+	tk.MustExec("drop table if exists admin_test;")
+	tk.MustExec("create table admin_test (c1 int, c2 int, c3 int default 1, index (c1), unique key(c2));")
+	tk.MustExec("insert admin_test (c1, c2) values (1, 1), (2, 2), (NULL, NULL);")
+	tk.MustExec("admin check table admin_test;")
+}
+
 func (s *testSuite) TestCoprocessorStreamingFlag(c *C) {
 	tk := testkit.NewTestKit(c, s.store)
 

table/tables/index.go

@@ -74,12 +74,11 @@ func (c *indexIter) Next() (val []types.Datum, h int64, err error) {
 	if err != nil {
 		return nil, 0, errors.Trace(err)
 	}
-	// if index is *not* unique, the handle is in keybuf
-	if !c.idx.idxInfo.Unique {
+	if len(vv) > len(c.idx.idxInfo.Columns) {
 		h = vv[len(vv)-1].GetInt64()
 		val = vv[0 : len(vv)-1]
 	} else {
-		// otherwise handle is value
+		// If the index is unique and the value isn't nil, the handle is in value.
 		h, err = decodeHandle(c.it.Value())
 		if err != nil {
 			return nil, 0, errors.Trace(err)

table/tables/index_test.go

@@ -186,6 +186,20 @@ func (s *testIndexSuite) TestIndex(c *C) {
 
 	_, err = index.FetchValues(make([]types.Datum, 0), nil)
 	c.Assert(err, NotNil)
+
+	// Test the function of Next when the value of unique key is nil.
+	values2 := types.MakeDatums(nil, nil)
+	_, err = index.Create(mockCtx, txn, values2, 2)
+	c.Assert(err, IsNil)
+	it, err = index.SeekFirst(txn)
+	c.Assert(err, IsNil)
+	getValues, h, err = it.Next()
+	c.Assert(err, IsNil)
+	c.Assert(getValues, HasLen, 2)
+	c.Assert(getValues[0].GetInterface(), Equals, nil)
+	c.Assert(getValues[1].GetInterface(), Equals, nil)
+	c.Assert(h, Equals, int64(2))
+	it.Close()
 }
 
 func (s *testIndexSuite) TestCombineIndexSeek(c *C) {