column names are case insensitive

现状: PK列名大写时(如ID), Column.IsPk()错误地返回false. 导致kafka输出key(主键)的payload为空.

代码位置:

func GetPrimaryKey(stmt *ast.CreateTableStmt) (map[string]struct{}, bool) {
	hasPk := false
	pkColumnsName := map[string]struct{}{}
	for _, constraint := range stmt.Constraints {
		if constraint.Tp == ast.ConstraintPrimaryKey {
			hasPk = true
			for _, col := range constraint.Keys {
				pkColumnsName[col.Column.Name.L] = struct{}{} // **HERE**
			}
		}
	}

更新, 和上述代码没有关系. 为了方便作case-insensitive的比较, 统一使用小写.

Column, index, stored routine, and event names are not case-sensitive on any platform, nor are column aliases.

mysql> create table a.a863 (id int primary key, val int, VAL int);
ERROR 1060 (42S21): Duplicate column name 'VAL'

[asiroliu]

复现：
version:

3.21.11.0-3.21.11.x-16a0ec1

全量复制会复现

1. src mysql perpare data

create database test_1;
create table test_1.a (ID int primary key);
insert into test_1.a values (1);

2. create dtle mysql to kafka job
3. get kafka message, payload is empty

{
  "schema": {
    "type": "struct",
    "optional": false,
    "name": "dtle.test_1.a.Key"
  },
  "payload": {}
}

[asiroliu]

验证：
version:

9.9.9.9-master-05c9de5

1. src mysql perpare data

create database test_1;
create table test_1.a (ID int primary key);
insert into test_1.a values (1);

2. create dtle mysql to kafka job
3. get kafka message
   {
   "schema": {
   "type": "struct",
   "optional": false,
   "fields": [
   {
   "type": "int32",
   "optional": false,
   "field": "ID"
   }
   ],
   "name": "dtle.test_1.a.Key"
   },
   "payload": {
   "ID": 1
   }
   }