From 5dc5402408574be0a8b368c5a8fcdd0f02774f9a Mon Sep 17 00:00:00 2001
From: dongmen <414110582@qq.com>
Date: Thu, 16 Mar 2023 15:55:35 +0800
Subject: [PATCH 1/3] add unit test for handle key column

---
 cdc/entry/schema_storage_test.go | 59 ++++++++++++++++++++++++++++++++
 1 file changed, 59 insertions(+)

diff --git a/cdc/entry/schema_storage_test.go b/cdc/entry/schema_storage_test.go
index 3cc70156b80..5fb3e50ed89 100644
--- a/cdc/entry/schema_storage_test.go
+++ b/cdc/entry/schema_storage_test.go
@@ -921,3 +921,62 @@ func getAllHistoryDDLJob(storage tidbkv.Storage) ([]*timodel.Job, error) {
 	}
 	return jobs, nil
 }
+
+// This test is used to show how the schemaStorage choose a handleKey of a table.
+// The handleKey is chosen by the following rules:
+// 1. If the table has a primary key, the handleKey is the first column of the primary key.
+// 2. If the table has not null unique key, the handleKey is the first column of the unique key.
+// 3. If the table has no primary key and no not null unique key, it has no handleKey.
+func TestHandleKey(t *testing.T) {
+	store, err := mockstore.NewMockStore()
+	require.Nil(t, err)
+	defer store.Close() //nolint:errcheck
+
+	session.SetSchemaLease(0)
+	session.DisableStats4Test()
+	domain, err := session.BootstrapSession(store)
+	require.Nil(t, err)
+	defer domain.Close()
+	domain.SetStatsUpdating(true)
+	tk := testkit.NewTestKit(t, store)
+	tk.MustExec("create database test2")
+	tk.MustExec("create table test.simple_test1 (id bigint primary key)")
+	tk.MustExec("create table test.simple_test2 (id bigint, age int NOT NULL, " +
+		"name char NOT NULL, UNIQUE KEY(age, name))")
+	tk.MustExec("create table test.simple_test3 (id bigint, age int)")
+	ver, err := store.CurrentVersion(oracle.GlobalTxnScope)
+	require.Nil(t, err)
+	meta, err := kv.GetSnapshotMeta(store, ver.Ver)
+	require.Nil(t, err)
+	snap, err := schema.NewSnapshotFromMeta(meta, ver.Ver, false)
+	require.Nil(t, err)
+	tb1, ok := snap.TableByName("test", "simple_test1")
+	require.True(t, ok)
+	require.Equal(t, int64(-1), tb1.HandleIndexID) // pk is handleKey
+	columnID := int64(1)
+	flag := tb1.ColumnsFlag[columnID]
+	require.True(t, flag.IsHandleKey())
+	for _, column := range tb1.Columns {
+		if column.ID == columnID {
+			require.True(t, column.Name.O == "id")
+		}
+	}
+
+	// unique key is handleKey
+	tb2, ok := snap.TableByName("test", "simple_test2")
+	require.True(t, ok)
+	require.Equal(t, int64(1), tb2.HandleIndexID)
+	columnID = int64(2)
+	flag = tb2.ColumnsFlag[columnID]
+	require.True(t, flag.IsHandleKey())
+	for _, column := range tb2.Columns {
+		if column.ID == columnID {
+			require.True(t, column.Name.O == "age")
+		}
+	}
+
+	// unique key is handleKey
+	tb3, ok := snap.TableByName("test", "simple_test3")
+	require.True(t, ok)
+	require.Equal(t, int64(-2), tb3.HandleIndexID)
+}

From 40e484420dbccf32025595f6a9f4d5561fb5f414 Mon Sep 17 00:00:00 2001
From: dongmen <414110582@qq.com>
Date: Thu, 16 Mar 2023 15:58:24 +0800
Subject: [PATCH 2/3] fix typo

---
 cdc/entry/schema_storage_test.go | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/cdc/entry/schema_storage_test.go b/cdc/entry/schema_storage_test.go
index 5fb3e50ed89..3a283a462a5 100644
--- a/cdc/entry/schema_storage_test.go
+++ b/cdc/entry/schema_storage_test.go
@@ -975,7 +975,7 @@ func TestHandleKey(t *testing.T) {
 		}
 	}
 
-	// unique key is handleKey
+	// has no handleKey
 	tb3, ok := snap.TableByName("test", "simple_test3")
 	require.True(t, ok)
 	require.Equal(t, int64(-2), tb3.HandleIndexID)

From 0c078336141b60465e3610f25da11e9f9c821beb Mon Sep 17 00:00:00 2001
From: dongmen <414110582@qq.com>
Date: Thu, 16 Mar 2023 16:08:25 +0800
Subject: [PATCH 3/3] address comment

---
 cdc/model/sink.go | 4 ++++
 1 file changed, 4 insertions(+)

diff --git a/cdc/model/sink.go b/cdc/model/sink.go
index fd601740908..2a6dd492c11 100644
--- a/cdc/model/sink.go
+++ b/cdc/model/sink.go
@@ -51,6 +51,10 @@ const (
 	// BinaryFlag means the column charset is binary
 	BinaryFlag ColumnFlagType = 1 << ColumnFlagType(iota)
 	// HandleKeyFlag means the column is selected as the handle key
+	// The handleKey is chosen by the following rules:
+	// 1. If the table has a primary key, the handleKey is the first column of the primary key.
+	// 2. If the table has not null unique key, the handleKey is the first column of the unique key.
+	// 3. If the table has no primary key and no not null unique key, it has no handleKey.
 	HandleKeyFlag
 	// GeneratedColumnFlag means the column is a generated column
 	GeneratedColumnFlag