From 83ec4b03f35830f9d6f3054b6dbf5ee2b411251c Mon Sep 17 00:00:00 2001
From: xiongjiwei <xiongjiwei1996@outlook.com>
Date: Mon, 9 Jan 2023 16:30:22 +0800
Subject: [PATCH] executor: support insert ignore/duplicate replace into with
 unique multi-valued index (#40369)

close pingcap/tidb#40207
---
 executor/batch_checker.go | 44 ++++++++++++++++++++-------------------
 executor/insert_test.go   | 28 +++++++++++++++++++++++++
 2 files changed, 51 insertions(+), 21 deletions(-)

diff --git a/executor/batch_checker.go b/executor/batch_checker.go
index c1eb1fda0d8f5..838c6af7bace0 100644
--- a/executor/batch_checker.go
+++ b/executor/batch_checker.go
@@ -190,28 +190,30 @@ func getKeysNeedCheckOneRow(ctx sessionctx.Context, t table.Table, row []types.D
 		}
 		// Pass handle = 0 to GenIndexKey,
 		// due to we only care about distinct key.
-		key, distinct, err1 := v.GenIndexKey(ctx.GetSessionVars().StmtCtx,
-			colVals, kv.IntHandle(0), nil)
-		if err1 != nil {
-			return nil, err1
-		}
-		// Skip the non-distinct keys.
-		if !distinct {
-			continue
-		}
-		// If index is used ingest ways, then we should check key from temp index.
-		if v.Meta().State != model.StatePublic && v.Meta().BackfillState != model.BackfillStateInapplicable {
-			_, key, _ = tables.GenTempIdxKeyByState(v.Meta(), key)
-		}
-		colValStr, err1 := formatDataForDupError(colVals)
-		if err1 != nil {
-			return nil, err1
+		iter := v.GenIndexKVIter(ctx.GetSessionVars().StmtCtx, colVals, kv.IntHandle(0), nil)
+		for iter.Valid() {
+			key, _, distinct, err1 := iter.Next(nil)
+			if err1 != nil {
+				return nil, err1
+			}
+			// Skip the non-distinct keys.
+			if !distinct {
+				continue
+			}
+			// If index is used ingest ways, then we should check key from temp index.
+			if v.Meta().State != model.StatePublic && v.Meta().BackfillState != model.BackfillStateInapplicable {
+				_, key, _ = tables.GenTempIdxKeyByState(v.Meta(), key)
+			}
+			colValStr, err1 := formatDataForDupError(colVals)
+			if err1 != nil {
+				return nil, err1
+			}
+			uniqueKeys = append(uniqueKeys, &keyValueWithDupInfo{
+				newKey:       key,
+				dupErr:       kv.ErrKeyExists.FastGenByArgs(colValStr, fmt.Sprintf("%s.%s", v.TableMeta().Name.String(), v.Meta().Name.String())),
+				commonHandle: t.Meta().IsCommonHandle,
+			})
 		}
-		uniqueKeys = append(uniqueKeys, &keyValueWithDupInfo{
-			newKey:       key,
-			dupErr:       kv.ErrKeyExists.FastGenByArgs(colValStr, fmt.Sprintf("%s.%s", v.TableMeta().Name.String(), v.Meta().Name.String())),
-			commonHandle: t.Meta().IsCommonHandle,
-		})
 	}
 	row = row[:len(row)-extraColumns]
 	result = append(result, toBeCheckedRow{
diff --git a/executor/insert_test.go b/executor/insert_test.go
index b55c3a63765e3..acd8d6736b219 100644
--- a/executor/insert_test.go
+++ b/executor/insert_test.go
@@ -419,6 +419,34 @@ func TestInsertValueForCastDecimalField(t *testing.T) {
 	tk.MustQuery(`select cast(a as decimal) from t1;`).Check(testkit.Rows(`9999999999`))
 }
 
+func TestInsertForMultiValuedIndex(t *testing.T) {
+	store := testkit.CreateMockStore(t)
+	tk := testkit.NewTestKit(t, store)
+	tk.MustExec("use test")
+	tk.MustExec(`drop table if exists t1;`)
+	tk.MustExec(`create table t1(a json, b int, unique index idx((cast(a as signed array))));`)
+	tk.MustExec(`insert into t1 values ('[1,11]', 1);`)
+	tk.MustExec(`insert into t1 values ('[2, 22]', 2);`)
+	tk.MustQuery(`select * from t1;`).Check(testkit.Rows(`[1, 11] 1`, `[2, 22] 2`))
+	tk.MustGetErrMsg(`insert into t1 values ('[2, 222]', 2);`, "[kv:1062]Duplicate entry '2' for key 't1.idx'")
+	tk.MustExec(`replace into t1 values ('[1, 10]', 10)`)
+	tk.MustQuery(`select * from t1;`).Check(testkit.Rows(`[2, 22] 2`, `[1, 10] 10`))
+	tk.MustExec(`replace into t1 values ('[1, 2]', 1)`)
+	tk.MustQuery(`select * from t1;`).Check(testkit.Rows(`[1, 2] 1`))
+	tk.MustExec(`replace into t1 values ('[1, 11]', 1)`)
+	tk.MustExec(`insert into t1 values ('[2, 22]', 2);`)
+	tk.MustQuery(`select * from t1;`).Check(testkit.Rows(`[1, 11] 1`, `[2, 22] 2`))
+	tk.MustExec(`insert ignore into t1 values ('[1]', 2);`)
+	tk.MustQuery(`select * from t1;`).Check(testkit.Rows(`[1, 11] 1`, `[2, 22] 2`))
+	tk.MustExec(`insert ignore into t1 values ('[1, 2]', 2);`)
+	tk.MustQuery(`select * from t1;`).Check(testkit.Rows(`[1, 11] 1`, `[2, 22] 2`))
+	tk.MustExec(`insert into t1 values ('[2]', 2) on duplicate key update b = 10;`)
+	tk.MustQuery(`select * from t1;`).Check(testkit.Rows(`[1, 11] 1`, `[2, 22] 10`))
+	tk.MustGetErrMsg(`insert into t1 values ('[2, 1]', 2) on duplicate key update a = '[1,2]';`, "[kv:1062]Duplicate entry '[1, 2]' for key 't1.idx'")
+	tk.MustGetErrMsg(`insert into t1 values ('[1,2]', 2) on duplicate key update a = '[1,2]';`, "[kv:1062]Duplicate entry '[1, 2]' for key 't1.idx'")
+	tk.MustGetErrMsg(`insert into t1 values ('[11, 22]', 2) on duplicate key update a = '[1,2]';`, "[kv:1062]Duplicate entry '[1, 2]' for key 't1.idx'")
+}
+
 func TestInsertDateTimeWithTimeZone(t *testing.T) {
 	store := testkit.CreateMockStore(t)
 	tk := testkit.NewTestKit(t, store)