ddl: Partition duplicate name fix

ddl/db_change_test.go

@@ -1586,7 +1586,7 @@ func (s *testStateChangeSuite) TestDDLIfExists(c *C) {
 	s.testParallelExecSQL(c, "alter table test_exists drop index if exists idx_c")
 
 	// DROP PARTITION (ADD PARTITION tested in TestParallelAlterAddPartition)
-	_, err = s.se.Execute(context.Background(), "create table test_exists_2 (a int key) partition by range(a) (partition p0 values less than (10), partition p1 values less than (20))")
+	_, err = s.se.Execute(context.Background(), "create table test_exists_2 (a int key) partition by range(a) (partition p0 values less than (10), partition p1 values less than (20), partition p2 values less than (30))")
 	c.Assert(err, IsNil)
 	s.testParallelExecSQL(c, "alter table test_exists_2 drop partition if exists p1")
 }

ddl/db_partition_test.go

@@ -3552,3 +3552,40 @@ func TestRenameTables(t *testing.T) {
 
 	ddl.ExportTestRenameTables(t)
 }
+
+func (s *testIntegrationSuite1) TestDuplicatePartitionNames(c *C) {
+	tk := testkit.NewTestKit(c, s.store)
+
+	tk.MustExec("create database DuplicatePartitionNames")
+	defer tk.MustExec("drop database DuplicatePartitionNames")
+	tk.MustExec("use DuplicatePartitionNames")
+
+	tk.MustExec("set @@tidb_enable_list_partition=on")
+	tk.MustExec("create table t1 (a int) partition by list (a) (partition p1 values in (1), partition p2 values in (2), partition p3 values in (3))")
+	tk.MustExec("insert into t1 values (1),(2),(3)")
+	tk.MustExec("alter table t1 truncate partition p1,p1")
+	tk.MustQuery("select * from t1").Sort().Check(testkit.Rows("2", "3"))
+	tk.MustExec("insert into t1 values (1)")
+	err := tk.ExecToErr("alter table t1 drop partition p1,p1")
+	c.Assert(err, NotNil)
+	c.Assert(err.Error(), Equals, "[ddl:1507]Error in list of partitions to DROP")
+	err = tk.ExecToErr("alter table t1 drop partition p1,p9")
+	c.Assert(err, NotNil)
+	c.Assert(err.Error(), Equals, "[ddl:1507]Error in list of partitions to DROP")
+	err = tk.ExecToErr("alter table t1 drop partition p1,p1,p1")
+	c.Assert(err, NotNil)
+	c.Assert(err.Error(), Equals, "[ddl:1508]Cannot remove all partitions, use DROP TABLE instead")
+	err = tk.ExecToErr("alter table t1 drop partition p1,p9,p1")
+	c.Assert(err, NotNil)
+	c.Assert(err.Error(), Equals, "[ddl:1508]Cannot remove all partitions, use DROP TABLE instead")
+	tk.MustQuery("select * from t1").Sort().Check(testkit.Rows("1", "2", "3"))
+	tk.MustExec("alter table t1 drop partition p1")
+	tk.MustQuery("select * from t1").Sort().Check(testkit.Rows("2", "3"))
+	tk.MustQuery("Show create table t1").Check(testkit.Rows("" +
+		"t1 CREATE TABLE `t1` (\n" +
+		"  `a` int(11) DEFAULT NULL\n" +
+		") ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_bin\n" +
+		"PARTITION BY LIST (`a`)\n" +
+		"(PARTITION `p2` VALUES IN (2),\n" +
+		" PARTITION `p3` VALUES IN (3))"))
+}

ddl/db_test.go

@@ -4899,13 +4899,13 @@ func (s *testDBSuite4) TestIfExists(c *C) {
 
 	// DROP PARTITION
 	s.mustExec(tk, c, "drop table if exists t2")
-	s.mustExec(tk, c, "create table t2 (a int key) partition by range(a) (partition p0 values less than (10), partition p1 values less than (20))")
+	s.mustExec(tk, c, "create table t2 (a int key) partition by range(a) (partition pNeg values less than (0), partition p0 values less than (10), partition p1 values less than (20))")
 	sql = "alter table t2 drop partition p1"
 	s.mustExec(tk, c, sql)
 	tk.MustGetErrCode(sql, errno.ErrDropPartitionNonExistent)
 	s.mustExec(tk, c, "alter table t2 drop partition if exists p1")
 	c.Assert(tk.Se.GetSessionVars().StmtCtx.WarningCount(), Equals, uint16(1))
-	tk.MustQuery("show warnings").Check(testutil.RowsWithSep("|", "Note|1507|Error in list of partitions to p1"))
+	tk.MustQuery("show warnings").Check(testutil.RowsWithSep("|", "Note|1507|Error in list of partitions to DROP"))
 }
 
 func testAddIndexForGeneratedColumn(tk *testkit.TestKit, s *testDBSuite5, c *C) {

ddl/ddl_api.go

@@ -3407,20 +3407,30 @@ func (d *ddl) TruncateTablePartition(ctx sessionctx.Context, ident ast.Ident, sp
 		return errors.Trace(ErrPartitionMgmtOnNonpartitioned)
 	}
 
-	pids := make([]int64, len(spec.PartitionNames))
+	var pids []int64
 	if spec.OnAllPartitions {
 		pids = make([]int64, len(meta.GetPartitionInfo().Definitions))
 		for i, def := range meta.GetPartitionInfo().Definitions {
 			pids[i] = def.ID
 		}
 	} else {
-		for i, name := range spec.PartitionNames {
+		// MySQL allows duplicate partition names in truncate partition
+		// so we filter them out through a hash
+		pidMap := make(map[int64]bool)
+		for _, name := range spec.PartitionNames {
 			pid, err := tables.FindPartitionByName(meta, name.L)
 			if err != nil {
 				return errors.Trace(err)
 			}
-			pids[i] = pid
+			pidMap[pid] = true
 		}
+		// linter makezero does not handle changing pids to zero length,
+		// so create a new var and then assign to pids...
+		newPids := make([]int64, 0, len(pidMap))
+		for pid := range pidMap {
+			newPids = append(newPids, pid)
+		}
+		pids = newPids
 	}
 
 	job := &model.Job{

ddl/partition.go

@@ -936,22 +936,31 @@ func checkDropTablePartition(meta *model.TableInfo, partLowerNames []string) err
 	if pi.Type != model.PartitionTypeRange && pi.Type != model.PartitionTypeList {
 		return errOnlyOnRangeListPartition.GenWithStackByArgs("DROP")
 	}
+
+	// To be error compatible with MySQL, we need to do this first!
+	// see https://github.com/pingcap/tidb/issues/31681#issuecomment-1015536214
 	oldDefs := pi.Definitions
+	if len(oldDefs) <= len(partLowerNames) {
+		return errors.Trace(ErrDropLastPartition)
+	}
+
+	dupCheck := make(map[string]bool)
 	for _, pn := range partLowerNames {
 		found := false
 		for _, def := range oldDefs {
 			if def.Name.L == pn {
+				if _, ok := dupCheck[pn]; ok {
+					return errors.Trace(ErrDropPartitionNonExistent.GenWithStackByArgs("DROP"))
+				}
+				dupCheck[pn] = true
 				found = true
 				break
 			}
 		}
 		if !found {
-			return errors.Trace(ErrDropPartitionNonExistent.GenWithStackByArgs(pn))
+			return errors.Trace(ErrDropPartitionNonExistent.GenWithStackByArgs("DROP"))
 		}
 	}
-	if len(oldDefs) == len(partLowerNames) {
-		return errors.Trace(ErrDropLastPartition)
-	}
 	return nil
 }