[hotfix-#1568][sqlserver] if write column is not have identity column, not execute SET IDENTITY_INSERT %s ON

chunjun-connectors/chunjun-connector-sqlserver/src/main/java/com/dtstack/chunjun/connector/sqlserver/sink/SqlserverOutputFormat.java

@@ -18,28 +18,82 @@
 
 package com.dtstack.chunjun.connector.sqlserver.sink;
 
+import com.dtstack.chunjun.config.FieldConfig;
 import com.dtstack.chunjun.connector.jdbc.sink.JdbcOutputFormat;
+import com.dtstack.chunjun.connector.jdbc.util.JdbcUtil;
 import com.dtstack.chunjun.connector.sqlserver.dialect.SqlserverDialect;
 import com.dtstack.chunjun.throwable.ChunJunRuntimeException;
 
+import lombok.extern.slf4j.Slf4j;
+import org.apache.commons.collections.CollectionUtils;
+
+import java.sql.PreparedStatement;
+import java.sql.ResultSet;
 import java.sql.SQLException;
 import java.sql.Statement;
+import java.util.Set;
+import java.util.stream.Collectors;
 
+@Slf4j
 public class SqlserverOutputFormat extends JdbcOutputFormat {
 
-    private static final long serialVersionUID = -663484948572872831L;
+    private final String findIdentityColumn =
+            "select name from sys.columns where object_id=object_id(?) and is_identity = 'true'";
 
     @Override
     protected void openInternal(int taskNumber, int numTasks) {
         super.openInternal(taskNumber, numTasks);
 
+        Statement statement = null;
         String sql =
                 ((SqlserverDialect) jdbcDialect)
                         .getIdentityInsertOnSql(jdbcConfig.getSchema(), jdbcConfig.getTable());
-        try (Statement statement = dbConn.createStatement()) {
-            statement.execute(sql);
+        try {
+            if (checkContainIdentityColumn()) {
+                statement = dbConn.createStatement();
+                statement.execute(sql);
+            }
         } catch (SQLException e) {
             throw new ChunJunRuntimeException(e);
+        } finally {
+            JdbcUtil.closeDbResources(null, statement, null, false);
+        }
+    }
+
+    private boolean checkContainIdentityColumn() {
+        try {
+
+            if (CollectionUtils.isNotEmpty(jdbcConfig.getColumn())) {
+                // 为* 代表所有字段，返回true，不需要校验字段里是否有identity字段，后面执行的时候还会校验一次表是否还有identity column
+                if (jdbcConfig.getColumn().get(0).getName().equals("*")) {
+                    return true;
+                }
+                try (PreparedStatement preparedStatement =
+                        dbConn.prepareStatement(findIdentityColumn)) {
+                    preparedStatement.setString(
+                            1,
+                            jdbcDialect.buildTableInfoWithSchema(
+                                    jdbcConfig.getSchema(), jdbcConfig.getTable()));
+                    // 包含就为true，不包含就为false
+                    try (ResultSet resultSet = preparedStatement.executeQuery()) {
+                        Set<String> columnNames =
+                                jdbcConfig.getColumn().stream()
+                                        .map(FieldConfig::getName)
+                                        .collect(Collectors.toSet());
+                        while (resultSet.next()) {
+                            if (columnNames.contains(resultSet.getString(1))) {
+                                return true;
+                            }
+                        }
+                    }
+                }
+                return false;
+            }
+            // 默认认为是包含的 因为后面执行具体sql的时候会判断表是否含有identity column
+            return true;
+        } catch (Exception e) {
+            log.warn("select identity column failed..", e);
+            return true;
         }
     }
 }

chunjun-local-test/src/main/java/com/dtstack/chunjun/local/test/LocalTest.java

@@ -43,7 +43,8 @@ public static void main(String[] args) throws Exception {
         //        confProperties.setProperty("state.checkpoints.dir", "file:///ck");
         String userDir = System.getProperty("user.dir");
 
-        String jobPath = "/Users/wtz/job_place/chunjun/1.16/_01_SQL/_06_Kafka/_01_sourceKafka_dimMysql_sinkMysql.sql";
+        String jobPath =
+                "/Users/wtz/job_place/chunjun/1.16/_01_SQL/_06_Kafka/_01_sourceKafka_dimMysql_sinkMysql.sql";
         String chunjunDistDir = userDir + "/chunjun-dist";
         String s = "";