bugfix: case of the pk col-name in the business sql is inconsistent w…

…ith the case in the table metadata, resulting in a rollback failure (apache#5748)

changes/en-us/2.0.0.md

@@ -64,6 +64,7 @@ The version is updated as follows:
 - [[#5661](https://github.com/seata/seata/pull/5661)] bugfix: the timeout is null when the connectionProxyXA connection is reused
 - [[#5678](https://github.com/seata/seata/pull/5675)] bugfix: fix compatibility between xxx.grouplist and grouplist.xxx configuration items
 - [[#5715](https://github.com/seata/seata/pull/5715)] fix get configuration item contains underlined error
+- [[#5748](https://github.com/seata/seata/pull/5748)] case of the pk col-name in the business sql is inconsistent with the case in the table metadata, resulting in a rollback failure
 
 ### optimize：
 - [[#5208](https://github.com/seata/seata/pull/5208)] optimize throwable getCause once more
@@ -147,6 +148,7 @@ Thanks to these contributors for their code commits. Please report an unintended
 - [tuwenlin](https://github.com/tuwenlin)
 - [sixlei](https://github.com/sixlei)
 - [yixia](https://github.com/wt-better)
+- [capthua](https://github.com/capthua)
 
 
 Also, we receive many valuable issues, questions and advices from our community. Thanks for you all.

changes/zh-cn/2.0.0.md

@@ -63,6 +63,7 @@ Seata 是一款开源的分布式事务解决方案，提供高性能和简单
 - [[#5661](https://github.com/seata/seata/pull/5661)] 修复connectionProxyXA连接复用时timeout为null
 - [[#5678](https://github.com/seata/seata/pull/5675)] 修复 xxx.grouplist 和 grouplist.xxx 配置项兼容问题
 - [[#5715](https://github.com/seata/seata/pull/5715)] 修复取中划线配置项错误问题
+- [[#5748](https://github.com/seata/seata/pull/5748)] 修复在某些情况下，业务sql中主键字段名大小写与表元数据中的不一致，导致回滚失败
 
 ### optimize：
 - [[#5208](https://github.com/seata/seata/pull/5208)] 优化多次重复获取Throwable#getCause问题
@@ -147,6 +148,7 @@ Seata 是一款开源的分布式事务解决方案，提供高性能和简单
 - [tuwenlin](https://github.com/tuwenlin)
 - [sixlei](https://github.com/sixlei)
 - [yixia](https://github.com/wt-better)
+- [capthua](https://github.com/capthua)
 
 同时，我们收到了社区反馈的很多有价值的issue和建议，非常感谢大家。
 

rm-datasource/src/main/java/io/seata/rm/datasource/sql/struct/TableRecords.java

@@ -28,6 +28,7 @@
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
+import java.util.Set;
 import javax.sql.rowset.serial.SerialBlob;
 import javax.sql.rowset.serial.SerialClob;
 import javax.sql.rowset.serial.SerialDatalink;
@@ -193,7 +194,7 @@ public static TableRecords empty(TableMeta tableMeta) {
     public static TableRecords buildRecords(TableMeta tmeta, ResultSet resultSet) throws SQLException {
         TableRecords records = new TableRecords(tmeta);
         ResultSetMetaData resultSetMetaData = resultSet.getMetaData();
-        Map<String, ColumnMeta> primaryKeyMap = tmeta.getPrimaryKeyMap();
+        Set<String> ignoreCasePKs = tmeta.getCaseInsensitivePKs();
         int columnCount = resultSetMetaData.getColumnCount();
 
         while (resultSet.next()) {
@@ -204,7 +205,7 @@ public static TableRecords buildRecords(TableMeta tmeta, ResultSet resultSet) th
                 int dataType = col.getDataType();
                 Field field = new Field();
                 field.setName(col.getColumnName());
-                if (primaryKeyMap.containsKey(colName)) {
+                if (ignoreCasePKs.contains(colName)) {
                     field.setKeyType(KeyType.PRIMARY_KEY);
                 }
                 field.setType(dataType);

sqlparser/seata-sqlparser-core/src/main/java/io/seata/sqlparser/struct/TableMeta.java

@@ -20,6 +20,9 @@
 import java.util.List;
 import java.util.Map;
 import java.util.Map.Entry;
+import java.util.Set;
+import java.util.TreeSet;
+import java.util.Collections;
 import java.util.Objects;
 import java.util.stream.Collectors;
 
@@ -150,6 +153,26 @@ public Map<String, ColumnMeta> getPrimaryKeyMap() {
         return pk;
     }
 
+    /**
+     * Gets case-insensitive primary key set
+     *
+     * @return case-insensitive, unmodifiable primary key set
+     */
+    public Set<String> getCaseInsensitivePKs() {
+        Set<String> pks = new TreeSet<>(String.CASE_INSENSITIVE_ORDER);
+        allIndexes.forEach((key, index) -> {
+            if (index.getIndextype().value() == IndexType.PRIMARY.value()) {
+                for (ColumnMeta col : index.getValues()) {
+                    pks.add(col.getColumnName());
+                }
+            }
+        });
+        if (pks.size() < 1) {
+            throw new NotSupportYetException(String.format("%s needs to contain the primary key.", tableName));
+        }
+        return Collections.unmodifiableSet(pks);
+    }
+
     /**
      * Gets primary key only name.
      *