[Improve][Connector-V2] Improve text write (#2971)

* [Improve][Connector-V2] Improve write text files

* [Improve][Connector-V2][File] Improve e2e test case

* [Docs][Connector-V2-status] Fix dead link

* [Improve][Connector-V2][File] Fix e2e test cases

* [Improve][Connector-V2][File] Solved the bug of one more line break at the end of the file

Author: TyrantLucifer

seatunnel-connectors-v2/connector-file/connector-file-base/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/config/BaseTextFileConfig.java

@@ -19,6 +19,10 @@
 
 import static com.google.common.base.Preconditions.checkNotNull;
 
+import org.apache.seatunnel.common.utils.DateTimeUtils;
+import org.apache.seatunnel.common.utils.DateUtils;
+import org.apache.seatunnel.common.utils.TimeUtils;
+
 import org.apache.seatunnel.shade.com.typesafe.config.Config;
 
 import lombok.Data;
@@ -31,16 +35,15 @@
 @Data
 public class BaseTextFileConfig implements DelimiterConfig, CompressConfig, Serializable {
     private static final long serialVersionUID = 1L;
-
     protected String compressCodec;
-
     protected String fieldDelimiter = String.valueOf('\001');
-
     protected String rowDelimiter = "\n";
-
     protected String path;
     protected String fileNameExpression;
     protected FileFormat fileFormat = FileFormat.TEXT;
+    protected DateUtils.Formatter dateFormat = DateUtils.Formatter.YYYY_MM_DD;
+    protected DateTimeUtils.Formatter datetimeFormat = DateTimeUtils.Formatter.YYYY_MM_DD_HH_MM_SS;
+    protected TimeUtils.Formatter timeFormat = TimeUtils.Formatter.HH_MM_SS;
 
     public BaseTextFileConfig(@NonNull Config config) {
         if (config.hasPath(Constant.COMPRESS_CODEC)) {
@@ -67,6 +70,18 @@ public BaseTextFileConfig(@NonNull Config config) {
         if (config.hasPath(Constant.FILE_FORMAT) && !StringUtils.isBlank(config.getString(Constant.FILE_FORMAT))) {
             this.fileFormat = FileFormat.valueOf(config.getString(Constant.FILE_FORMAT).toUpperCase(Locale.ROOT));
         }
+
+        if (config.hasPath(Constant.DATE_FORMAT)) {
+            dateFormat = DateUtils.Formatter.parse(config.getString(Constant.DATE_FORMAT));
+        }
+
+        if (config.hasPath(Constant.DATETIME_FORMAT)) {
+            datetimeFormat = DateTimeUtils.Formatter.parse(config.getString(Constant.DATETIME_FORMAT));
+        }
+
+        if (config.hasPath(Constant.TIME_FORMAT)) {
+            timeFormat = TimeUtils.Formatter.parse(config.getString(Constant.TIME_FORMAT));
+        }
     }
 
     public BaseTextFileConfig() {}

seatunnel-connectors-v2/connector-file/connector-file-base/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/config/Constant.java

@@ -22,10 +22,11 @@ public class Constant {
     public static final String NON_PARTITION = "NON_PARTITION";
     public static final String TRANSACTION_ID_SPLIT = "_";
     public static final String TRANSACTION_EXPRESSION = "transactionId";
-
     public static final String SAVE_MODE = "save_mode";
     public static final String COMPRESS_CODEC = "compress_codec";
-
+    public static final String DATE_FORMAT = "date_format";
+    public static final String DATETIME_FORMAT = "datetime_format";
+    public static final String TIME_FORMAT = "time_format";
     public static final String PATH = "path";
     public static final String FIELD_DELIMITER = "field_delimiter";
     public static final String ROW_DELIMITER = "row_delimiter";

seatunnel-connectors-v2/connector-file/connector-file-base/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/sink/writer/JsonWriteStrategy.java

@@ -35,10 +35,12 @@ public class JsonWriteStrategy extends AbstractWriteStrategy {
     private final byte[] rowDelimiter;
     private SerializationSchema serializationSchema;
     private final Map<String, FSDataOutputStream> beingWrittenOutputStream;
+    private final Map<String, Boolean> isFirstWrite;
 
     public JsonWriteStrategy(TextFileSinkConfig textFileSinkConfig) {
         super(textFileSinkConfig);
         this.beingWrittenOutputStream = new HashMap<>();
+        this.isFirstWrite = new HashMap<>();
         this.rowDelimiter = textFileSinkConfig.getRowDelimiter().getBytes();
     }
 
@@ -54,8 +56,12 @@ public void write(@NonNull SeaTunnelRow seaTunnelRow) {
         FSDataOutputStream fsDataOutputStream = getOrCreateOutputStream(filePath);
         try {
             byte[] rowBytes = serializationSchema.serialize(seaTunnelRow);
+            if (isFirstWrite.get(filePath)) {
+                isFirstWrite.put(filePath, false);
+            } else {
+                fsDataOutputStream.write(rowDelimiter);
+            }
             fsDataOutputStream.write(rowBytes);
-            fsDataOutputStream.write(rowDelimiter);
         } catch (IOException e) {
             log.error("write data to file {} error", filePath);
             throw new RuntimeException(e);
@@ -88,6 +94,7 @@ private FSDataOutputStream getOrCreateOutputStream(@NonNull String filePath) {
             try {
                 fsDataOutputStream = FileSystemUtils.getOutputStream(filePath);
                 beingWrittenOutputStream.put(filePath, fsDataOutputStream);
+                isFirstWrite.put(filePath, true);
             } catch (IOException e) {
                 log.error("can not get output file stream");
                 throw new RuntimeException(e);

seatunnel-connectors-v2/connector-file/connector-file-base/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/sink/writer/TextWriteStrategy.java

@@ -17,39 +17,67 @@
 
 package org.apache.seatunnel.connectors.seatunnel.file.sink.writer;
 
+import org.apache.seatunnel.api.serialization.SerializationSchema;
 import org.apache.seatunnel.api.table.type.SeaTunnelRow;
+import org.apache.seatunnel.api.table.type.SeaTunnelRowType;
+import org.apache.seatunnel.common.utils.DateTimeUtils;
+import org.apache.seatunnel.common.utils.DateUtils;
+import org.apache.seatunnel.common.utils.TimeUtils;
 import org.apache.seatunnel.connectors.seatunnel.file.sink.config.TextFileSinkConfig;
 import org.apache.seatunnel.connectors.seatunnel.file.sink.util.FileSystemUtils;
+import org.apache.seatunnel.format.text.TextSerializationSchema;
 
 import lombok.NonNull;
 import org.apache.hadoop.fs.FSDataOutputStream;
 
 import java.io.IOException;
-import java.util.Arrays;
 import java.util.HashMap;
 import java.util.Map;
-import java.util.stream.Collectors;
 
 public class TextWriteStrategy extends AbstractWriteStrategy {
     private final Map<String, FSDataOutputStream> beingWrittenOutputStream;
+    private final Map<String, Boolean> isFirstWrite;
     private final String fieldDelimiter;
     private final String rowDelimiter;
+    private final DateUtils.Formatter dateFormat;
+    private final DateTimeUtils.Formatter dateTimeFormat;
+    private final TimeUtils.Formatter timeFormat;
+    private SerializationSchema serializationSchema;
 
     public TextWriteStrategy(TextFileSinkConfig textFileSinkConfig) {
         super(textFileSinkConfig);
         this.beingWrittenOutputStream = new HashMap<>();
+        this.isFirstWrite = new HashMap<>();
         this.fieldDelimiter = textFileSinkConfig.getFieldDelimiter();
         this.rowDelimiter = textFileSinkConfig.getRowDelimiter();
+        this.dateFormat = textFileSinkConfig.getDateFormat();
+        this.dateTimeFormat = textFileSinkConfig.getDatetimeFormat();
+        this.timeFormat = textFileSinkConfig.getTimeFormat();
+    }
+
+    @Override
+    public void setSeaTunnelRowTypeInfo(SeaTunnelRowType seaTunnelRowType) {
+        super.setSeaTunnelRowTypeInfo(seaTunnelRowType);
+        this.serializationSchema = TextSerializationSchema.builder()
+                .seaTunnelRowType(seaTunnelRowType)
+                .delimiter(fieldDelimiter)
+                .dateFormatter(dateFormat)
+                .dateTimeFormatter(dateTimeFormat)
+                .timeFormatter(timeFormat)
+                .build();
     }
 
     @Override
     public void write(@NonNull SeaTunnelRow seaTunnelRow) {
         String filePath = getOrCreateFilePathBeingWritten(seaTunnelRow);
         FSDataOutputStream fsDataOutputStream = getOrCreateOutputStream(filePath);
-        String line = transformRowToLine(seaTunnelRow);
         try {
-            fsDataOutputStream.write(line.getBytes());
-            fsDataOutputStream.write(rowDelimiter.getBytes());
+            if (isFirstWrite.get(filePath)) {
+                isFirstWrite.put(filePath, false);
+            } else {
+                fsDataOutputStream.write(rowDelimiter.getBytes());
+            }
+            fsDataOutputStream.write(serializationSchema.serialize(seaTunnelRow));
         } catch (IOException e) {
             log.error("write data to file {} error", filePath);
             throw new RuntimeException(e);
@@ -81,29 +109,12 @@ private FSDataOutputStream getOrCreateOutputStream(@NonNull String filePath) {
             try {
                 fsDataOutputStream = FileSystemUtils.getOutputStream(filePath);
                 beingWrittenOutputStream.put(filePath, fsDataOutputStream);
+                isFirstWrite.put(filePath, true);
             } catch (IOException e) {
                 log.error("can not get output file stream");
                 throw new RuntimeException(e);
             }
         }
         return fsDataOutputStream;
     }
-
-    private String transformRowToLine(@NonNull SeaTunnelRow seaTunnelRow) {
-        return Arrays.stream(seaTunnelRow.getFields()).map(v -> {
-            if (v == null) {
-                return "";
-            } else if (v.getClass().isArray()) {
-                if (v instanceof byte[]) {
-                    return Arrays.toString((byte[]) v);
-                } else {
-                    return Arrays.toString((Object[]) v);
-                }
-            } else if (v instanceof SeaTunnelRow) {
-                return "{" + transformRowToLine((SeaTunnelRow) v) + "}";
-            } else {
-                return v.toString();
-            }
-        }).collect(Collectors.joining(fieldDelimiter));
-    }
 }

seatunnel-e2e/seatunnel-spark-connector-v2-e2e/connector-file-spark-e2e/src/test/resources/file/fakesource_to_local_text.conf

@@ -30,8 +30,36 @@ source {
     result_table_name = "fake"
     schema = {
       fields {
-        name = "string"
-        age = "int"
+        c_map = "map<string, string>"
+        c_array = "array<int>"
+        c_string = string
+        c_boolean = boolean
+        c_tinyint = tinyint
+        c_smallint = smallint
+        c_int = int
+        c_bigint = bigint
+        c_float = float
+        c_double = double
+        c_bytes = bytes
+        c_date = date
+        c_decimal = "decimal(30, 8)"
+        c_timestamp = timestamp
+        c_row = {
+          c_map = "map<string, string>"
+          c_array = "array<int>"
+          c_string = string
+          c_boolean = boolean
+          c_tinyint = tinyint
+          c_smallint = smallint
+          c_int = int
+          c_bigint = bigint
+          c_float = float
+          c_double = double
+          c_bytes = bytes
+          c_date = date
+          c_decimal = "decimal(30, 8)"
+          c_timestamp = timestamp
+        }
       }
     }
   }
@@ -42,7 +70,7 @@ source {
 
 transform {
   sql {
-    sql = "select name,age from fake"
+    sql = "select * from fake"
   }
 
   # If you would like to get more information about how to configure seatunnel and see full list of transform plugins,
@@ -54,12 +82,8 @@ sink {
     path="/tmp/test/text"
     field_delimiter="\t"
     row_delimiter="\n"
-    partition_by=["age"]
-    partition_dir_expression="${k0}=${v0}"
-    is_partition_field_write_in_file=true
     file_name_expression="${transactionId}_${now}"
     file_format="text"
-    sink_columns=["name","age"]
     filename_time_format="yyyy.MM.dd"
     is_enable_transaction=true
     save_mode="error"