[Improve][Jdbc] Support write unicode text into sqlserver

Author: hailin0

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

@@ -83,6 +83,7 @@ public class SqlServerTypeConverter implements TypeConverter<BasicTypeDefine> {
     public static final int MAX_SCALE = MAX_PRECISION - 1;
     public static final int DEFAULT_SCALE = 18;
     public static final int MAX_CHAR_LENGTH = 8000;
+    public static final int MAX_NVARCHAR_LENGTH = 4000;
     public static final int MAX_BINARY_LENGTH = 8000;
     public static final int MAX_TIME_SCALE = 7;
     public static final int MAX_TIMESTAMP_SCALE = 7;
@@ -403,16 +404,16 @@ public BasicTypeDefine reconvert(Column column) {
                 break;
             case STRING:
                 if (column.getColumnLength() == null || column.getColumnLength() <= 0) {
-                    builder.columnType(SQLSERVER_TEXT);
-                    builder.dataType(SQLSERVER_TEXT);
-                } else if (column.getColumnLength() <= MAX_CHAR_LENGTH) {
+                    builder.columnType(MAX_NVARCHAR);
+                    builder.dataType(MAX_NVARCHAR);
+                } else if (column.getColumnLength() <= MAX_NVARCHAR_LENGTH) {
                     builder.columnType(
-                            String.format("%s(%s)", SQLSERVER_VARCHAR, column.getColumnLength()));
-                    builder.dataType(SQLSERVER_VARCHAR);
+                            String.format("%s(%s)", SQLSERVER_NVARCHAR, column.getColumnLength()));
+                    builder.dataType(SQLSERVER_NVARCHAR);
                     builder.length(column.getColumnLength());
                 } else {
-                    builder.columnType(SQLSERVER_TEXT);
-                    builder.dataType(SQLSERVER_TEXT);
+                    builder.columnType(MAX_NVARCHAR);
+                    builder.dataType(MAX_NVARCHAR);
                     builder.length(column.getColumnLength());
                 }
                 break;

seatunnel-connectors-v2/connector-jdbc/src/test/java/org/apache/seatunnel/connectors/seatunnel/jdbc/catalog/PreviewActionTest.java

@@ -375,7 +375,7 @@ public void testSqlServerPreviewAction() {
                 "IF OBJECT_ID('[testddatabase].[testtable]', 'U') IS NULL \n"
                         + "BEGIN \n"
                         + "CREATE TABLE [testddatabase].[testtable] ( \n"
-                        + "\t[test] TEXT NULL\n"
+                        + "\t[test] NVARCHAR(MAX) NULL\n"
                         + ");\n"
                         + "EXEC testddatabase.sys.sp_addextendedproperty 'MS_Description', N'comment', 'schema', N'null', 'table', N'testtable';\n"
                         + "EXEC testddatabase.sys.sp_addextendedproperty 'MS_Description', N'', 'schema', N'null', 'table', N'testtable', 'column', N'test';\n"

seatunnel-connectors-v2/connector-jdbc/src/test/java/org/apache/seatunnel/connectors/seatunnel/jdbc/internal/dialect/sqlserver/SqlServerTypeConverterTest.java

@@ -748,36 +748,37 @@ public void testReconvertString() {
 
         BasicTypeDefine typeDefine = SqlServerTypeConverter.INSTANCE.reconvert(column);
         Assertions.assertEquals(column.getName(), typeDefine.getName());
-        Assertions.assertEquals(SqlServerTypeConverter.SQLSERVER_TEXT, typeDefine.getColumnType());
-        Assertions.assertEquals(SqlServerTypeConverter.SQLSERVER_TEXT, typeDefine.getDataType());
+        Assertions.assertEquals(SqlServerTypeConverter.MAX_NVARCHAR, typeDefine.getColumnType());
+        Assertions.assertEquals(SqlServerTypeConverter.MAX_NVARCHAR, typeDefine.getDataType());
 
         column =
                 PhysicalColumn.builder()
                         .name("test")
                         .dataType(BasicType.STRING_TYPE)
-                        .columnLength(8000L)
+                        .columnLength(4000L)
                         .build();
 
         typeDefine = SqlServerTypeConverter.INSTANCE.reconvert(column);
         Assertions.assertEquals(column.getName(), typeDefine.getName());
         Assertions.assertEquals(
                 String.format(
                         "%s(%s)",
-                        SqlServerTypeConverter.SQLSERVER_VARCHAR, column.getColumnLength()),
+                        SqlServerTypeConverter.SQLSERVER_NVARCHAR, column.getColumnLength()),
                 typeDefine.getColumnType());
-        Assertions.assertEquals(SqlServerTypeConverter.SQLSERVER_VARCHAR, typeDefine.getDataType());
+        Assertions.assertEquals(
+                SqlServerTypeConverter.SQLSERVER_NVARCHAR, typeDefine.getDataType());
 
         column =
                 PhysicalColumn.builder()
                         .name("test")
                         .dataType(BasicType.STRING_TYPE)
-                        .columnLength(8001L)
+                        .columnLength(4001L)
                         .build();
 
         typeDefine = SqlServerTypeConverter.INSTANCE.reconvert(column);
         Assertions.assertEquals(column.getName(), typeDefine.getName());
-        Assertions.assertEquals(SqlServerTypeConverter.SQLSERVER_TEXT, typeDefine.getColumnType());
-        Assertions.assertEquals(SqlServerTypeConverter.SQLSERVER_TEXT, typeDefine.getDataType());
+        Assertions.assertEquals(SqlServerTypeConverter.MAX_NVARCHAR, typeDefine.getColumnType());
+        Assertions.assertEquals(SqlServerTypeConverter.MAX_NVARCHAR, typeDefine.getDataType());
     }
 
     @Test