[Feature][Connector-V2] add sqlserver connector

docs/en/connector-v2/sink/Jdbc.md

@@ -3,13 +3,16 @@
 > JDBC sink connector
 
 ## Description
-Write data through jdbc. Support Batch mode and Streaming mode, support concurrent writing, support exactly-once semantics (using XA transaction guarantee).
+
+Write data through jdbc. Support Batch mode and Streaming mode, support concurrent writing, support exactly-once
+semantics (using XA transaction guarantee).
 
 ## Key features
 
 - [x] [exactly-once](../../concept/connector-v2-features.md)
 
-Use `Xa transactions` to ensure `exactly-once`. So only support `exactly-once` for the database which is support `Xa transactions`. You can set `is_exactly_once=true` to enable it.
+Use `Xa transactions` to ensure `exactly-once`. So only support `exactly-once` for the database which is
+support `Xa transactions`. You can set `is_exactly_once=true` to enable it.
 
 - [ ] [schema projection](../../concept/connector-v2-features.md)
 
@@ -32,61 +35,85 @@ Use `Xa transactions` to ensure `exactly-once`. So only support `exactly-once` f
 | transaction_timeout_sec      | Int     | No       | -1            |
 
 ### driver [string]
+
 The jdbc class name used to connect to the remote data source, if you use MySQL the value is com.mysql.cj.jdbc.Driver.
-Warn: for license compliance, you have to provide MySQL JDBC driver yourself, e.g. copy mysql-connector-java-xxx.jar to $SEATNUNNEL_HOME/lib for Standalone.
+Warn: for license compliance, you have to provide any driver yourself like MySQL JDBC Driver, e.g. copy mysql-connector-java-xxx.jar to
+$SEATNUNNEL_HOME/lib for Standalone.
 
 ### user [string]
+
 userName
 
 ### password [string]
+
 password
 
 ### url [string]
+
 The URL of the JDBC connection. Refer to a case: jdbc:postgresql://localhost/test
 
 ### query [string]
+
 Query statement
 
 ### connection_check_timeout_sec [int]
 
 The time in seconds to wait for the database operation used to validate the connection to complete.
 
 ### max_retries[int]
+
 The number of retries to submit failed (executeBatch)
 
 ### batch_size[int]
-For batch writing, when the number of buffers reaches the number of `batch_size` or the time reaches `batch_interval_ms`, the data will be flushed into the database
+
+For batch writing, when the number of buffers reaches the number of `batch_size` or the time reaches `batch_interval_ms`
+, the data will be flushed into the database
 
 ### batch_interval_ms[int]
-For batch writing, when the number of buffers reaches the number of `batch_size` or the time reaches `batch_interval_ms`, the data will be flushed into the database
+
+For batch writing, when the number of buffers reaches the number of `batch_size` or the time reaches `batch_interval_ms`
+, the data will be flushed into the database
 
 ### is_exactly_once[boolean]
-Whether to enable exactly-once semantics, which will use Xa transactions. If on, you need to set `xa_data_source_class_name`.
+
+Whether to enable exactly-once semantics, which will use Xa transactions. If on, you need to
+set `xa_data_source_class_name`.
 
 ### xa_data_source_class_name[string]
-The xa data source class name of the database Driver, for example, mysql is `com.mysql.cj.jdbc.MysqlXADataSource`, and please refer to appendix for other data sources
+
+The xa data source class name of the database Driver, for example, mysql is `com.mysql.cj.jdbc.MysqlXADataSource`, and
+please refer to appendix for other data sources
 
 ### max_commit_attempts[int]
+
 The number of retries for transaction commit failures
 
 ### transaction_timeout_sec[int]
-The timeout after the transaction is opened, the default is -1 (never timeout). Note that setting the timeout may affect exactly-once semantics
+
+The timeout after the transaction is opened, the default is -1 (never timeout). Note that setting the timeout may affect
+exactly-once semantics
 
 ## tips
-In the case of is_exactly_once = "true", Xa transactions are used. This requires database support, and some databases require some setup. For example, postgres needs to set `max_prepared_transactions > 1`
+
+In the case of is_exactly_once = "true", Xa transactions are used. This requires database support, and some databases
+require some setup. For example, postgres needs to set `max_prepared_transactions > 1`
 Such as `ALTER SYSTEM set max_prepared_transactions to 10`.
 
 ## appendix
+
 there are some reference value for params above.
 
 | datasource | driver                   | url                                       | xa_data_source_class_name           | maven                                                         |
 |------------|--------------------------|-------------------------------------------|-------------------------------------|---------------------------------------------------------------|
 | mysql      | com.mysql.cj.jdbc.Driver | jdbc:mysql://localhost:3306/test          | com.mysql.cj.jdbc.MysqlXADataSource | https://mvnrepository.com/artifact/mysql/mysql-connector-java |
-| postgresql | org.postgresql.Driver    | jdbc:postgresql://localhost:5432/postgres | org.postgresql.xa.PGXADataSource    | https://mvnrepository.com/artifact/org.postgresql/postgresql  |                                                             |
+| postgresql | org.postgresql.Driver    | jdbc:postgresql://localhost:5432/postgres | org.postgresql.xa.PGXADataSource    | https://mvnrepository.com/artifact/org.postgresql/postgresql  |
 | dm         | dm.jdbc.driver.DmDriver  | jdbc:dm://localhost:5236                  | dm.jdbc.driver.DmdbXADataSource     | https://mvnrepository.com/artifact/com.dameng/DmJdbcDriver18  |
+| sqlserver  | com.microsoft.sqlserver.jdbc.SQLServerDriver  | jdbc:microsoft:sqlserver://localhost:1433       | com.microsoft.sqlserver.jdbc.SQLServerXADataSource     | https://mvnrepository.com/artifact/com.microsoft.sqlserver/mssql-jdbc  |
 
 ## Example
+
 Simple
+
 ```
 jdbc {
     url = "jdbc:mysql://localhost/test"
@@ -99,6 +126,7 @@ jdbc {
 ```
 
 Exactly-once
+
 ```
 jdbc {
 

docs/en/connector-v2/source/Jdbc.md

@@ -73,8 +73,9 @@ there are some reference value for params above.
 | datasource | driver                   | url                                       | maven                                                         |
 |------------|--------------------------|-------------------------------------------|---------------------------------------------------------------|
 | mysql      | com.mysql.cj.jdbc.Driver | jdbc:mysql://localhost:3306/test          | https://mvnrepository.com/artifact/mysql/mysql-connector-java |
-| postgresql | org.postgresql.Driver    | jdbc:postgresql://localhost:5432/postgres | https://mvnrepository.com/artifact/org.postgresql/postgresql  |                                                             |
+| postgresql | org.postgresql.Driver    | jdbc:postgresql://localhost:5432/postgres | https://mvnrepository.com/artifact/org.postgresql/postgresql  |
 | dm         | dm.jdbc.driver.DmDriver  | jdbc:dm://localhost:5236                  | https://mvnrepository.com/artifact/com.dameng/DmJdbcDriver18  |
+| sqlserver  | com.microsoft.sqlserver.jdbc.SQLServerDriver  | jdbc:microsoft:sqlserver://localhost:1433       | https://mvnrepository.com/artifact/com.microsoft.sqlserver/mssql-jdbc  |
 
 ## Example
 simple:

pom.xml

@@ -146,6 +146,7 @@
         <mysql.version>8.0.16</mysql.version>
         <postgresql.version>42.3.3</postgresql.version>
         <dm-jdbc.version>8.1.2.141</dm-jdbc.version>
+        <sqlserver.version>9.4.1.jre8</sqlserver.version>
         <skip.pmd.check>false</skip.pmd.check>
         <maven.deploy.skip>false</maven.deploy.skip>
         <maven.javadoc.skip>false</maven.javadoc.skip>

seatunnel-connectors-v2/connector-jdbc/pom.xml

@@ -62,6 +62,13 @@
             <artifactId>ali-phoenix-shaded-thin-client</artifactId>
             <version>${phoenix.version}</version>
         </dependency>
+
+        <dependency>
+            <groupId>com.microsoft.sqlserver</groupId>
+            <artifactId>mssql-jdbc</artifactId>
+            <version>${sqlserver.version}</version>
+            <scope>provided</scope>
+        </dependency>
     </dependencies>
 
 </project>

seatunnel-connectors-v2/connector-jdbc/src/main/java/org/apache/seatunnel/connectors/seatunnel/jdbc/internal/dialect/sqlserver/SqlServerDialect.java

@@ -0,0 +1,39 @@
+/*
+ * 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.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.seatunnel.connectors.seatunnel.jdbc.internal.dialect.sqlserver;
+
+import org.apache.seatunnel.connectors.seatunnel.jdbc.internal.converter.JdbcRowConverter;
+import org.apache.seatunnel.connectors.seatunnel.jdbc.internal.dialect.JdbcDialect;
+import org.apache.seatunnel.connectors.seatunnel.jdbc.internal.dialect.JdbcDialectTypeMapper;
+
+public class SqlServerDialect implements JdbcDialect {
+    @Override
+    public String dialectName() {
+        return "Sqlserver";
+    }
+
+    @Override
+    public JdbcRowConverter getRowConverter() {
+        return new SqlserverJdbcRowConverter();
+    }
+
+    @Override
+    public JdbcDialectTypeMapper getJdbcDialectTypeMapper() {
+        return new SqlserverTypeMapper();
+    }
+}

seatunnel-connectors-v2/connector-jdbc/src/main/java/org/apache/seatunnel/connectors/seatunnel/jdbc/internal/dialect/sqlserver/SqlServerDialectFactory.java

@@ -0,0 +1,40 @@
+/*
+ * 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.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.seatunnel.connectors.seatunnel.jdbc.internal.dialect.sqlserver;
+
+import org.apache.seatunnel.connectors.seatunnel.jdbc.internal.dialect.JdbcDialect;
+import org.apache.seatunnel.connectors.seatunnel.jdbc.internal.dialect.JdbcDialectFactory;
+
+import com.google.auto.service.AutoService;
+
+/**
+ * Factory for {@link SqlServerDialect}.
+ */
+
+@AutoService(JdbcDialectFactory.class)
+public class SqlServerDialectFactory implements JdbcDialectFactory {
+    @Override
+    public boolean acceptsURL(String url) {
+        return url.startsWith("jdbc:sqlserver:");
+    }
+
+    @Override
+    public JdbcDialect create() {
+        return new SqlServerDialect();
+    }
+}

seatunnel-connectors-v2/connector-jdbc/src/main/java/org/apache/seatunnel/connectors/seatunnel/jdbc/internal/dialect/sqlserver/SqlserverJdbcRowConverter.java

@@ -0,0 +1,39 @@
+/*
+ * 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.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.seatunnel.connectors.seatunnel.jdbc.internal.dialect.sqlserver;
+
+import org.apache.seatunnel.api.table.type.SeaTunnelRow;
+import org.apache.seatunnel.api.table.type.SeaTunnelRowType;
+import org.apache.seatunnel.connectors.seatunnel.jdbc.internal.converter.AbstractJdbcRowConverter;
+
+import java.sql.ResultSet;
+import java.sql.ResultSetMetaData;
+import java.sql.SQLException;
+
+public class SqlserverJdbcRowConverter extends AbstractJdbcRowConverter {
+
+    @Override
+    public String converterName() {
+        return "Sqlserver";
+    }
+
+    @Override
+    public SeaTunnelRow toInternal(ResultSet rs, ResultSetMetaData metaData, SeaTunnelRowType typeInfo) throws SQLException {
+        return super.toInternal(rs, metaData, typeInfo);
+    }
+}