From 1f873106625c0bf33389708643d71151cec92e7b Mon Sep 17 00:00:00 2001
From: pingcap-github-bot <sre-bot@pingcap.com>
Date: Fri, 30 Aug 2019 23:55:34 +0800
Subject: [PATCH] ddl: fix add unique index on partitioned table (by RANGE
 COLUMNS) failed (#11946) (#11958)

---
 ddl/db_integration_test.go | 28 ++++++++++++++++++++++++++++
 ddl/ddl_api.go             |  2 +-
 ddl/partition.go           | 28 +++++++++++++++++++++++-----
 3 files changed, 52 insertions(+), 6 deletions(-)

diff --git a/ddl/db_integration_test.go b/ddl/db_integration_test.go
index 9dbb6eec33e24..505a44dff3522 100644
--- a/ddl/db_integration_test.go
+++ b/ddl/db_integration_test.go
@@ -1660,6 +1660,34 @@ func (s *testIntegrationSuite3) TestAlterAlgorithm(c *C) {
 	s.tk.MustExec("alter table t default charset = utf8mb4, ALGORITHM=INSTANT")
 }
 
+func (s *testIntegrationSuite3) TestAlterTableAddUniqueOnPartionRangeColumn(c *C) {
+	s.tk = testkit.NewTestKit(c, s.store)
+	s.tk.MustExec("use test")
+	s.tk.MustExec("drop table if exists t")
+	defer s.tk.MustExec("drop table if exists t")
+
+	s.tk.MustExec(`create table t(
+	a int,
+	b varchar(100),
+	c int,
+	INDEX idx_c(c))
+	PARTITION BY RANGE COLUMNS( a ) (
+		PARTITION p0 VALUES LESS THAN (6),
+		PARTITION p1 VALUES LESS THAN (11),
+		PARTITION p2 VALUES LESS THAN (16),
+		PARTITION p3 VALUES LESS THAN (21)
+	)`)
+	s.tk.MustExec("insert into t values (4, 'xxx', 4)")
+	s.tk.MustExec("insert into t values (4, 'xxx', 9)") // Note the repeated 4
+	s.tk.MustExec("insert into t values (17, 'xxx', 12)")
+	assertErrorCode(c, s.tk, "alter table t add unique index idx_a(a)", mysql.ErrDupEntry)
+
+	s.tk.MustExec("delete from t where a = 4")
+	s.tk.MustExec("alter table t add unique index idx_a(a)")
+	s.tk.MustExec("alter table t add unique index idx_ac(a, c)")
+	assertErrorCode(c, s.tk, "alter table t add unique index idx_b(b)", mysql.ErrUniqueKeyNeedAllFieldsInPf)
+}
+
 func (s *testIntegrationSuite5) TestFulltextIndexIgnore(c *C) {
 	s.tk = testkit.NewTestKit(c, s.store)
 	s.tk.MustExec("use test")
diff --git a/ddl/ddl_api.go b/ddl/ddl_api.go
index 8f011b37635ac..07841648ed76d 100644
--- a/ddl/ddl_api.go
+++ b/ddl/ddl_api.go
@@ -3104,7 +3104,7 @@ func (d *ddl) CreateIndex(ctx sessionctx.Context, ti ast.Ident, unique bool, ind
 		return errors.Trace(err)
 	}
 	if unique && tblInfo.GetPartitionInfo() != nil {
-		if err := checkPartitionKeysConstraint(ctx, tblInfo.GetPartitionInfo().Expr, idxColNames, tblInfo); err != nil {
+		if err := checkPartitionKeysConstraint(tblInfo.GetPartitionInfo(), idxColNames, tblInfo); err != nil {
 			return err
 		}
 	}
diff --git a/ddl/partition.go b/ddl/partition.go
index 7a6d9ca4917e3..16bbb8fb05b6d 100644
--- a/ddl/partition.go
+++ b/ddl/partition.go
@@ -26,6 +26,7 @@ import (
 	"github.com/pingcap/parser/mysql"
 	"github.com/pingcap/parser/opcode"
 	"github.com/pingcap/tidb/expression"
+	"github.com/pingcap/tidb/infoschema"
 	"github.com/pingcap/tidb/meta"
 	"github.com/pingcap/tidb/sessionctx"
 	"github.com/pingcap/tidb/table"
@@ -522,11 +523,28 @@ func checkRangePartitioningKeysConstraints(sctx sessionctx.Context, s *ast.Creat
 	return nil
 }
 
-func checkPartitionKeysConstraint(sctx sessionctx.Context, partExpr string, idxColNames []*ast.IndexColName, tblInfo *model.TableInfo) error {
-	// Parse partitioning key, extract the column names in the partitioning key to slice.
-	partCols, err := extractPartitionColumns(partExpr, tblInfo)
-	if err != nil {
-		return err
+func checkPartitionKeysConstraint(pi *model.PartitionInfo, idxColNames []*ast.IndexColName, tblInfo *model.TableInfo) error {
+	var (
+		partCols []*model.ColumnInfo
+		err      error
+	)
+	// The expr will be an empty string if the partition is defined by:
+	// CREATE TABLE t (...) PARTITION BY RANGE COLUMNS(...)
+	if partExpr := pi.Expr; partExpr != "" {
+		// Parse partitioning key, extract the column names in the partitioning key to slice.
+		partCols, err = extractPartitionColumns(partExpr, tblInfo)
+		if err != nil {
+			return err
+		}
+	} else {
+		partCols = make([]*model.ColumnInfo, 0, len(pi.Columns))
+		for _, col := range pi.Columns {
+			colInfo := getColumnInfoByName(tblInfo, col.L)
+			if colInfo == nil {
+				return infoschema.ErrColumnNotExists.GenWithStackByArgs(col, tblInfo.Name)
+			}
+			partCols = append(partCols, colInfo)
+		}
 	}
 
 	// Every unique key on the table must use every column in the table's partitioning expression.