feature: add SQL Server composite primary keys (#8041)

changes/en-us/2.x.md

@@ -1,4 +1,4 @@
-<!--
+<!--
     Licensed to the Apache Software Foundation (ASF) under one or more
     contributor license agreements.  See the NOTICE file distributed with
     this work for additional information regarding copyright ownership.
@@ -30,6 +30,7 @@ Add changes here for all PR submitted to the 2.x branch.
 - [[#8002](https://github.com/apache/incubator-seata/pull/8002)] add Grafana dashboard JSON for NamingServer metrics
 - [[#8020](https://github.com/apache/incubator-seata/pull/8020)] add UnregisterRM protocol to notify server on client destroy
 - [[#8044](https://github.com/apache/incubator-seata/pull/8044)] add protobuf serialization support for UnregisterRM protocol
+- [[#8050](https://github.com/apache/incubator-seata/pull/8050)] add SQL Server composite primary keys
 - [[#8046](https://github.com/apache/incubator-seata/pull/8046)] add fastjson2 and jackson3
 
 ### bugfix:
@@ -96,6 +97,7 @@ Thanks to these contributors for their code commits. Please report an unintended
 - [WangzJi](https://github.com/WangzJi)
 - [somiljain](https://github.com/somiljain)
 - [xuxiaowei-com-cn](https://github.com/xuxiaowei-com-cn)
+- [UokyI](https://github.com/UokyI)
 
 
 Also, we receive many valuable issues, questions and advices from our community. Thanks for you all.

changes/zh-cn/2.x.md

@@ -31,6 +31,7 @@
 - [[#8002](https://github.com/apache/incubator-seata/pull/8002)] 为namingserver指标增加Grafana dashboard JSON
 - [[#8020](https://github.com/apache/incubator-seata/pull/8020)] 新增 UnregisterRM 协议，在客户端销毁时通知服务端
 - [[#8044](https://github.com/apache/incubator-seata/pull/8044)] 为 UnregisterRM 协议添加 protobuf 序列化支持
+- [[#8050](https://github.com/apache/incubator-seata/pull/8050)] 新增SQL Server 多主键支持
 - [[#8046](https://github.com/apache/incubator-seata/pull/8046)] 添加了 fastjson2 和 jackson3
 
 ### bugfix:
@@ -99,5 +100,6 @@
 - [xiaoxiangyeyu0](https://github.com/xiaoxiangyeyu0)
 - [WangzJi](https://github.com/WangzJi)
 - [xuxiaowei-com-cn](https://github.com/xuxiaowei-com-cn)
+- [UokyI](https://github.com/UokyI)
 
 同时，我们收到了社区反馈的很多有价值的issue和建议，非常感谢大家。

rm-datasource/src/main/java/org/apache/seata/rm/datasource/AbstractConnectionProxy.java

@@ -23,6 +23,7 @@
 import org.apache.seata.sqlparser.SQLRecognizer;
 import org.apache.seata.sqlparser.SQLType;
 import org.apache.seata.sqlparser.struct.TableMeta;
+import org.apache.seata.sqlparser.util.JdbcConstants;
 
 import java.sql.Array;
 import java.sql.Blob;
@@ -118,10 +119,17 @@ public PreparedStatement prepareStatement(String sql) throws SQLException {
                                     getTargetConnection(),
                                     sqlRecognizer.getTableName(),
                                     getDataSourceProxy().getResourceId());
-                    String[] pkNameArray =
-                            new String[tableMeta.getPrimaryKeyOnlyName().size()];
-                    tableMeta.getPrimaryKeyOnlyName().toArray(pkNameArray);
-                    targetPreparedStatement = getTargetConnection().prepareStatement(sql, pkNameArray);
+                    // Fix: SQL Server does not support array of column names for getGeneratedKeys, use
+                    // RETURN_GENERATED_KEYS instead.
+                    if (JdbcConstants.SQLSERVER.equalsIgnoreCase(dbType)) {
+                        targetPreparedStatement =
+                                getTargetConnection().prepareStatement(sql, Statement.RETURN_GENERATED_KEYS);
+                    } else {
+                        String[] pkNameArray =
+                                new String[tableMeta.getPrimaryKeyOnlyName().size()];
+                        tableMeta.getPrimaryKeyOnlyName().toArray(pkNameArray);
+                        targetPreparedStatement = getTargetConnection().prepareStatement(sql, pkNameArray);
+                    }
                 }
             }
         }

rm-datasource/src/main/java/org/apache/seata/rm/datasource/SqlGenerateUtils.java

@@ -18,6 +18,7 @@
 
 import org.apache.seata.rm.datasource.sql.struct.Field;
 import org.apache.seata.sqlparser.util.ColumnUtils;
+import org.apache.seata.sqlparser.util.JdbcConstants;
 
 import java.sql.PreparedStatement;
 import java.sql.SQLException;
@@ -78,6 +79,11 @@ public static List<WhereSql> buildWhereConditionListByPKs(List<String> pkNameLis
      */
     public static List<WhereSql> buildWhereConditionListByPKs(
             List<String> pkNameList, int rowSize, String dbType, int maxInSize) {
+        // SQL Server does not support tuple IN syntax: (col1,col2) IN ((?,?),(?,?))
+        // Use AND/OR syntax instead
+        if (JdbcConstants.SQLSERVER.equalsIgnoreCase(dbType) && pkNameList.size() > 1) {
+            return buildWhereConditionListByPKsForSqlServer(pkNameList, rowSize, maxInSize, dbType);
+        }
         List<WhereSql> whereSqls = new ArrayList<>();
         // we must consider the situation of composite primary key
         int batchSize = rowSize % maxInSize == 0 ? rowSize / maxInSize : (rowSize / maxInSize) + 1;
@@ -115,6 +121,45 @@ public static List<WhereSql> buildWhereConditionListByPKs(
         return whereSqls;
     }
 
+    /**
+     * Build where condition list by PKs for SQL Server.
+     * SQL Server does not support tuple IN syntax: (col1,col2) IN ((?,?),(?,?))
+     * Use AND/OR syntax instead: (col1=? AND col2=?) OR (col1=? AND col2=?)
+     *
+     * @param pkNameList pk column name list
+     * @param rowSize the row size of records
+     * @param maxInSize the max in size
+     * @return where condition sql list for SQL Server
+     */
+    private static List<WhereSql> buildWhereConditionListByPKsForSqlServer(
+            List<String> pkNameList, int rowSize, int maxInSize, String dbType) {
+        List<WhereSql> whereSqls = new ArrayList<>();
+        int batchSize = rowSize % maxInSize == 0 ? rowSize / maxInSize : (rowSize / maxInSize) + 1;
+        for (int batch = 0; batch < batchSize; batch++) {
+            StringBuilder whereStr = new StringBuilder();
+            int eachSize =
+                    (batch == batchSize - 1) ? (rowSize % maxInSize == 0 ? maxInSize : rowSize % maxInSize) : maxInSize;
+
+            for (int i = 0; i < eachSize; i++) {
+                if (i > 0) {
+                    whereStr.append(" OR ");
+                }
+                whereStr.append("(");
+                for (int x = 0; x < pkNameList.size(); x++) {
+                    if (x > 0) {
+                        whereStr.append(" AND ");
+                    }
+                    whereStr.append(ColumnUtils.addEscape(pkNameList.get(x), dbType));
+                    whereStr.append("=?");
+                }
+                whereStr.append(")");
+            }
+            whereSqls.add(new WhereSql(whereStr.toString(), eachSize, pkNameList.size()));
+        }
+
+        return whereSqls;
+    }
+
     /**
      * set parameter for PreparedStatement, this is only used in pk sql.
      *

rm-datasource/src/main/java/org/apache/seata/rm/datasource/exec/sqlserver/SqlServerInsertExecutor.java

@@ -42,6 +42,7 @@
 import java.sql.Statement;
 import java.util.ArrayList;
 import java.util.Collections;
+import java.util.HashMap;
 import java.util.HashSet;
 import java.util.List;
 import java.util.Map;
@@ -69,7 +70,8 @@ public SqlServerInsertExecutor(
 
     @Override
     public Map<String, List<Object>> getPkValues() throws SQLException {
-        Map<String, List<Object>> pkValuesMap;
+        // Fix: Initialize pkValuesMap to support SQL Server composite primary keys.
+        Map<String, List<Object>> pkValuesMap = new HashMap<>();
         boolean isContainsPk = containsPK();
         List<String> pkColumnNameList = getTableMeta().getPrimaryKeyOnlyName();
 
@@ -84,8 +86,44 @@ public Map<String, List<Object>> getPkValues() throws SQLException {
                 pkValuesMap = getPkValuesWithNoColumn();
             }
         } else {
-            // when there is a composite primary key
-            throw new NotSupportYetException("composite primary key is not supported in sqlserver");
+            // when there is a composite primary key - Fix: Support SQL Server composite primary keys.
+            // SQL Server allows only one IDENTITY column per table.
+            // So composite PK can have at most one auto-increment column.
+            // Strategy: parse PK values from INSERT columns, then fill missing auto-increment PK from generated keys.
+            if (!getPkIndex().isEmpty()) {
+                // At least one PK column is in the INSERT statement.
+                pkValuesMap = getPkValuesByColumn();
+                Map<String, ColumnMeta> primaryKeyMap = getTableMeta().getPrimaryKeyMap();
+
+                // Fill any missing auto-increment PK columns from generated keys.
+                for (String pkColumnName : pkColumnNameList) {
+                    if (!pkValuesMap.containsKey(pkColumnName)) {
+                        ColumnMeta pkMeta = primaryKeyMap.get(pkColumnName);
+                        if (pkMeta.isAutoincrement()) {
+                            pkValuesMap.put(pkColumnName, getGeneratedKeys());
+                        } else {
+                            throw new NotSupportYetException(
+                                    "composite primary key with non-autoincrement column not in INSERT is not supported in sqlserver: "
+                                            + pkColumnName);
+                        }
+                    }
+                }
+            } else {
+                // No PK columns in INSERT statement.
+                // For composite PK, this means all PK columns must have values from elsewhere.
+                // Since SQL Server only supports one IDENTITY column, non-identity PK columns would fail.
+                Map<String, ColumnMeta> primaryKeyMap = getTableMeta().getPrimaryKeyMap();
+                for (String pkColumnName : pkColumnNameList) {
+                    ColumnMeta pkMeta = primaryKeyMap.get(pkColumnName);
+                    if (pkMeta.isAutoincrement()) {
+                        pkValuesMap.put(pkColumnName, getGeneratedKeys());
+                    } else {
+                        throw new NotSupportYetException(
+                                "composite primary key with non-autoincrement column not in INSERT is not supported in sqlserver: "
+                                        + pkColumnName);
+                    }
+                }
+            }
         }
 
         return pkValuesMap;

rm-datasource/src/test/java/org/apache/seata/rm/datasource/SqlGenerateUtilsTest.java

@@ -64,4 +64,48 @@ void testBuildSQLByPKs() {
                 "select id,name from t_order where  (id,name) in ( (?,?),(?,?) ) union select id,name from t_order where  (id,name) in ( (?,?),(?,?) )",
                 sqlJoiner.toString());
     }
+
+    @Test
+    void testBuildWhereConditionListByPKsForSqlServer() {
+        List<String> pkNameList = new ArrayList<>();
+        pkNameList.add("id");
+        pkNameList.add("name");
+
+        // Test SQL Server composite primary key syntax
+        List<SqlGenerateUtils.WhereSql> results =
+                SqlGenerateUtils.buildWhereConditionListByPKs(pkNameList, 4, "sqlserver", 2);
+
+        Assertions.assertEquals(2, results.size());
+        // SQL Server should use AND/OR syntax instead of tuple IN
+        results.forEach(result -> {
+            Assertions.assertEquals("(id=? AND name=?) OR (id=? AND name=?)", result.getSql());
+            Assertions.assertEquals(2, result.getRowSize());
+            Assertions.assertEquals(2, result.getPkSize());
+        });
+
+        // Test with odd row size
+        List<SqlGenerateUtils.WhereSql> resultsOdd =
+                SqlGenerateUtils.buildWhereConditionListByPKs(pkNameList, 5, "sqlserver", 2);
+        Assertions.assertEquals(3, resultsOdd.size());
+        Assertions.assertEquals(
+                "(id=? AND name=?) OR (id=? AND name=?)", resultsOdd.get(0).getSql());
+        Assertions.assertEquals(
+                "(id=? AND name=?) OR (id=? AND name=?)", resultsOdd.get(1).getSql());
+        Assertions.assertEquals("(id=? AND name=?)", resultsOdd.get(2).getSql());
+        Assertions.assertEquals(1, resultsOdd.get(2).getRowSize());
+    }
+
+    @Test
+    void testBuildWhereConditionListByPKsForSqlServerWithSinglePk() {
+        List<String> pkNameList = new ArrayList<>();
+        pkNameList.add("id");
+
+        // Single PK should use the common tuple IN syntax, not SQL Server special branch
+        List<SqlGenerateUtils.WhereSql> results =
+                SqlGenerateUtils.buildWhereConditionListByPKs(pkNameList, 3, "sqlserver", 2);
+
+        Assertions.assertEquals(2, results.size());
+        Assertions.assertEquals("(id) in ( (?),(?) )", results.get(0).getSql());
+        Assertions.assertEquals("(id) in ( (?) )", results.get(1).getSql());
+    }
 }

rm-datasource/src/test/java/org/apache/seata/rm/datasource/exec/SqlServerInsertExecutorTest.java

@@ -209,4 +209,114 @@ private void mockStatementInsertRows() {
         rows.add(Arrays.asList(Null.get(), "xx", "xx", "xx"));
         when(sqlInsertRecognizer.getInsertRows(pkIndexMap.values())).thenReturn(rows);
     }
+
+    @Test
+    public void testGetPkValues_compositePrimaryKey_withAllPkInInsert() throws Exception {
+        // Mock composite primary key: id + user_id
+        List<String> compositePkList = Arrays.asList(ID_COLUMN, USER_ID_COLUMN);
+        when(tableMeta.getPrimaryKeyOnlyName()).thenReturn(compositePkList);
+
+        doReturn(tableMeta).when(insertExecutor).getTableMeta();
+        doReturn(pkIndexMap).when(insertExecutor).getPkIndex(); // PK columns are in INSERT statement
+
+        // Mock getPkValuesByColumn to return expected values directly
+        Map<String, List<Object>> expectedPkValues = new HashMap<>();
+        expectedPkValues.put(ID_COLUMN, Arrays.asList(1, 2));
+        expectedPkValues.put(USER_ID_COLUMN, Arrays.asList("user1", "user2"));
+        doReturn(expectedPkValues).when(insertExecutor).getPkValuesByColumn();
+
+        Map<String, List<Object>> pkValues = insertExecutor.getPkValues();
+
+        // Verify composite primary key values are correctly retrieved
+        Assertions.assertNotNull(pkValues);
+        Assertions.assertEquals(expectedPkValues, pkValues);
+
+        // Verify that getPkValuesByColumn was called, confirming the code path for composite keys with manual values
+        verify(insertExecutor).getPkValuesByColumn();
+    }
+
+    @Test
+    public void testGetPkValues_compositePrimaryKey_withAutoIncrement() throws Exception {
+        // Mock composite primary key with auto-increment columns
+        List<String> compositePkList = Arrays.asList(ID_COLUMN, USER_ID_COLUMN);
+        when(tableMeta.getPrimaryKeyOnlyName()).thenReturn(compositePkList);
+
+        Map<String, ColumnMeta> pkMap = new HashMap<>();
+        ColumnMeta idMeta = mock(ColumnMeta.class);
+        when(idMeta.isAutoincrement()).thenReturn(true); // Auto-increment
+        pkMap.put(ID_COLUMN, idMeta);
+
+        ColumnMeta userIdMeta = mock(ColumnMeta.class);
+        when(userIdMeta.isAutoincrement()).thenReturn(true); // Auto-increment
+        pkMap.put(USER_ID_COLUMN, userIdMeta);
+        when(tableMeta.getPrimaryKeyMap()).thenReturn(pkMap);
+
+        doReturn(tableMeta).when(insertExecutor).getTableMeta();
+        doReturn(new HashMap<String, Integer>()).when(insertExecutor).getPkIndex(); // No PK columns in INSERT
+        doReturn(Arrays.asList(PK_VALUE)).when(insertExecutor).getGeneratedKeys();
+
+        Map<String, List<Object>> pkValues = insertExecutor.getPkValues();
+
+        // Verify auto-increment column gets value from generated keys
+        Assertions.assertEquals(Arrays.asList(PK_VALUE), pkValues.get(ID_COLUMN));
+        Assertions.assertEquals(Arrays.asList(PK_VALUE), pkValues.get(USER_ID_COLUMN));
+    }
+
+    @Test
+    public void testGetPkValues_compositePrimaryKey_nonAutoIncrementThrowsException() throws Exception {
+        // Mock composite primary key with non-auto-increment column not in INSERT
+        List<String> compositePkList = Arrays.asList(ID_COLUMN, USER_ID_COLUMN);
+        when(tableMeta.getPrimaryKeyOnlyName()).thenReturn(compositePkList);
+
+        Map<String, ColumnMeta> pkMap = new HashMap<>();
+        ColumnMeta idMeta = mock(ColumnMeta.class);
+        when(idMeta.isAutoincrement()).thenReturn(false); // Not auto-increment
+        pkMap.put(ID_COLUMN, idMeta);
+
+        ColumnMeta userIdMeta = mock(ColumnMeta.class);
+        when(userIdMeta.isAutoincrement()).thenReturn(false);
+        pkMap.put(USER_ID_COLUMN, userIdMeta);
+        when(tableMeta.getPrimaryKeyMap()).thenReturn(pkMap);
+
+        doReturn(tableMeta).when(insertExecutor).getTableMeta();
+        doReturn(new HashMap<String, Integer>()).when(insertExecutor).getPkIndex(); // No PK columns in INSERT
+
+        // Should throw exception for non-auto-increment composite primary key
+        Assertions.assertThrows(NotSupportYetException.class, () -> insertExecutor.getPkValues());
+    }
+
+    @Test
+    public void testGetPkValues_compositePrimaryKey_withPartialPkInInsertAndAutoIncrement() throws Exception {
+        // Mock composite primary key where one PK is provided in INSERT and the other is auto-increment
+        List<String> compositePkList = Arrays.asList(ID_COLUMN, USER_ID_COLUMN);
+        when(tableMeta.getPrimaryKeyOnlyName()).thenReturn(compositePkList);
+
+        Map<String, ColumnMeta> pkMap = new HashMap<>();
+        ColumnMeta idMeta = mock(ColumnMeta.class);
+        when(idMeta.isAutoincrement()).thenReturn(false); // Provided in INSERT
+        pkMap.put(ID_COLUMN, idMeta);
+
+        ColumnMeta userIdMeta = mock(ColumnMeta.class);
+        when(userIdMeta.isAutoincrement()).thenReturn(true); // Generated by database
+        pkMap.put(USER_ID_COLUMN, userIdMeta);
+        when(tableMeta.getPrimaryKeyMap()).thenReturn(pkMap);
+
+        doReturn(tableMeta).when(insertExecutor).getTableMeta();
+        Map<String, Integer> partialPkIndex = new HashMap<>();
+        partialPkIndex.put(ID_COLUMN, 0);
+        doReturn(partialPkIndex).when(insertExecutor).getPkIndex(); // One PK column is present in INSERT
+
+        // Mock getPkValuesByColumn to return only the manually provided PK
+        Map<String, List<Object>> manualPkValues = new HashMap<>();
+        manualPkValues.put(ID_COLUMN, Arrays.asList(1, 2));
+        doReturn(manualPkValues).when(insertExecutor).getPkValuesByColumn();
+
+        doReturn(Arrays.asList(PK_VALUE)).when(insertExecutor).getGeneratedKeys();
+
+        Map<String, List<Object>> pkValues = insertExecutor.getPkValues();
+
+        // Verify manual and generated PK values are merged correctly
+        Assertions.assertEquals(Arrays.asList(1, 2), pkValues.get(ID_COLUMN));
+        Assertions.assertEquals(Arrays.asList(PK_VALUE), pkValues.get(USER_ID_COLUMN));
+    }
 }