[Connector-V2] [Clickhouse] Improve Clickhouse File Connector

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

@@ -21,22 +21,25 @@ Write data to Clickhouse can also be done using JDBC
 
 ## Options
 
-| name                   | type    | required | default value |
-| ---------------------- | ------- | -------- | ------------- |
-| host                   | string  | yes      | -             |
-| database               | string  | yes      | -             |
-| table                  | string  | yes      | -             |
-| username               | string  | yes      | -             |
-| password               | string  | yes      | -             |
-| clickhouse_local_path  | string  | yes      | -             |
-| sharding_key           | string  | no       | -             |
-| copy_method            | string  | no       | scp           |
-| node_free_password     | boolean | no       | false         |
-| node_pass              | list    | no       | -             |
-| node_pass.node_address | string  | no       | -             |
-| node_pass.username     | string  | no       | "root"        |
-| node_pass.password     | string  | no       | -             |
-| common-options         |         | no       | -             |
+| name                   | type    | required | default value                          |
+|------------------------|---------|----------|----------------------------------------|
+| host                   | string  | yes      | -                                      |
+| database               | string  | yes      | -                                      |
+| table                  | string  | yes      | -                                      |
+| username               | string  | yes      | -                                      |
+| password               | string  | yes      | -                                      |
+| clickhouse_local_path  | string  | yes      | -                                      |
+| sharding_key           | string  | no       | -                                      |
+| copy_method            | string  | no       | scp                                    |
+| node_free_password     | boolean | no       | false                                  |
+| node_pass              | list    | no       | -                                      |
+| node_pass.node_address | string  | no       | -                                      |
+| node_pass.username     | string  | no       | "root"                                 |
+| node_pass.password     | string  | no       | -                                      |
+| compatible_mode        | boolean | no       | false                                  |
+| file_fields_delimiter  | string  | no       | "\t"                                   |
+| file_temp_path         | string  | no       | "/tmp/seatunnel/clickhouse-local/file" |
+| common-options         |         | no       | -                                      |
 
 ### host [string]
 
@@ -94,6 +97,21 @@ The username corresponding to the clickhouse server, default root user.
 
 The password corresponding to the clickhouse server.
 
+### compatible_mode [boolean]
+
+In the lower version of Clickhouse, the ClickhouseLocal program does not support the `--path` parameter,
+you need to use this mode to take other ways to realize the `--path` parameter function
+
+### file_fields_delimiter [string]
+
+ClickhouseFile uses csv format to temporarily save data. If the data in the row contains the delimiter value
+of csv, it may cause program exceptions. 
+Avoid this with this configuration. Value string has to be an exactly one character long
+
+### file_temp_path [string]
+
+The directory where ClickhouseFile stores temporary files locally.
+
 ### common options
 
 Sink plugin common parameters, please refer to [Sink Common Options](common-options.md) for details
@@ -122,3 +140,8 @@ Sink plugin common parameters, please refer to [Sink Common Options](common-opti
 ### 2.2.0-beta 2022-09-26
 
 - Support write data to ClickHouse File and move to ClickHouse data dir
+
+### Next version
+
+- [BugFix] Fix generated data part name conflict and improve file commit logic [3416](https://github.com/apache/incubator-seatunnel/pull/3416)
+- [Feature] Support compatible_mode compatible with lower version Clickhouse  [3416](https://github.com/apache/incubator-seatunnel/pull/3416)

seatunnel-connectors-v2/connector-clickhouse/src/main/java/org/apache/seatunnel/connectors/seatunnel/clickhouse/config/ClickhouseConfig.java

@@ -96,7 +96,16 @@ public class ClickhouseConfig {
     public static final Option<ClickhouseFileCopyMethod> COPY_METHOD = Options.key("copy_method").enumType(ClickhouseFileCopyMethod.class)
         .defaultValue(ClickhouseFileCopyMethod.SCP).withDescription("The method of copy Clickhouse file");
 
+    public static final Option<Boolean> COMPATIBLE_MODE = Options.key("compatible_mode").booleanType()
+        .defaultValue(false).withDescription("In the lower version of Clickhouse, the ClickhouseLocal program does not support the `--path` parameter, " +
+            "you need to use this mode to take other ways to realize the --path parameter function");
+
     public static final String NODE_ADDRESS = "node_address";
+
+    public static final Option<Boolean> NODE_FREE_PASSWORD = Options.key("node_free_password").booleanType()
+        .defaultValue(false).withDescription("Because seatunnel need to use scp or rsync for file transfer, " +
+            "seatunnel need clickhouse server-side access. If each spark node and clickhouse server are configured with password-free login, " +
+            "you can configure this option to true, otherwise you need to configure the corresponding node password in the node_pass configuration");
     /**
      * The password of Clickhouse server node
      */
@@ -106,4 +115,11 @@ public class ClickhouseConfig {
     public static final Option<Map<String, String>> CLICKHOUSE_PREFIX = Options.key("clickhouse").mapType()
         .defaultValue(Collections.emptyMap()).withDescription("Clickhouse custom config");
 
+    public static final Option<String> FILE_FIELDS_DELIMITER = Options.key("file_fields_delimiter").stringType()
+        .defaultValue("\t").withDescription("ClickhouseFile uses csv format to temporarily save data. If the data in the row contains the delimiter value of csv," +
+            " it may cause program exceptions. Avoid this with this configuration. Value string has to be an exactly one character long");
+
+    public static final Option<String> FILE_TEMP_PATH = Options.key("file_temp_path").stringType()
+        .defaultValue("/tmp/seatunnel/clickhouse-local/file").withDescription("The directory where ClickhouseFile stores temporary files locally.");
+
 }

seatunnel-connectors-v2/connector-clickhouse/src/main/java/org/apache/seatunnel/connectors/seatunnel/clickhouse/config/FileReaderOption.java

@@ -20,10 +20,13 @@
 import org.apache.seatunnel.api.table.type.SeaTunnelRowType;
 import org.apache.seatunnel.connectors.seatunnel.clickhouse.shard.ShardMetadata;
 
+import lombok.Data;
+
 import java.io.Serializable;
 import java.util.List;
 import java.util.Map;
 
+@Data
 public class FileReaderOption implements Serializable {
 
     private ShardMetadata shardMetadata;
@@ -35,90 +38,29 @@ public class FileReaderOption implements Serializable {
     private Map<String, String> nodeUser;
     private Map<String, String> nodePassword;
     private SeaTunnelRowType seaTunnelRowType;
+    private boolean compatibleMode;
+    private String fileTempPath;
+    private String fileFieldsDelimiter;
 
     public FileReaderOption(ShardMetadata shardMetadata, Map<String, String> tableSchema,
                             List<String> fields, String clickhouseLocalPath,
                             ClickhouseFileCopyMethod copyMethod,
                             Map<String, String> nodeUser,
-                            Map<String, String> nodePassword) {
+                            boolean nodeFreePass,
+                            Map<String, String> nodePassword,
+                            boolean compatibleMode,
+                            String fileTempPath,
+                            String fileFieldsDelimiter) {
         this.shardMetadata = shardMetadata;
         this.tableSchema = tableSchema;
         this.fields = fields;
         this.clickhouseLocalPath = clickhouseLocalPath;
         this.copyMethod = copyMethod;
         this.nodeUser = nodeUser;
-        this.nodePassword = nodePassword;
-    }
-
-    public SeaTunnelRowType getSeaTunnelRowType() {
-        return seaTunnelRowType;
-    }
-
-    public void setSeaTunnelRowType(SeaTunnelRowType seaTunnelRowType) {
-        this.seaTunnelRowType = seaTunnelRowType;
-    }
-
-    public boolean isNodeFreePass() {
-        return nodeFreePass;
-    }
-
-    public void setNodeFreePass(boolean nodeFreePass) {
         this.nodeFreePass = nodeFreePass;
-    }
-
-    public String getClickhouseLocalPath() {
-        return clickhouseLocalPath;
-    }
-
-    public void setClickhouseLocalPath(String clickhouseLocalPath) {
-        this.clickhouseLocalPath = clickhouseLocalPath;
-    }
-
-    public ClickhouseFileCopyMethod getCopyMethod() {
-        return copyMethod;
-    }
-
-    public void setCopyMethod(ClickhouseFileCopyMethod copyMethod) {
-        this.copyMethod = copyMethod;
-    }
-
-    public Map<String, String> getNodeUser() {
-        return nodeUser;
-    }
-
-    public void setNodeUser(Map<String, String> nodeUser) {
-        this.nodeUser = nodeUser;
-    }
-
-    public Map<String, String> getNodePassword() {
-        return nodePassword;
-    }
-
-    public void setNodePassword(Map<String, String> nodePassword) {
         this.nodePassword = nodePassword;
-    }
-
-    public ShardMetadata getShardMetadata() {
-        return shardMetadata;
-    }
-
-    public void setShardMetadata(ShardMetadata shardMetadata) {
-        this.shardMetadata = shardMetadata;
-    }
-
-    public Map<String, String> getTableSchema() {
-        return tableSchema;
-    }
-
-    public void setTableSchema(Map<String, String> tableSchema) {
-        this.tableSchema = tableSchema;
-    }
-
-    public List<String> getFields() {
-        return fields;
-    }
-
-    public void setFields(List<String> fields) {
-        this.fields = fields;
+        this.compatibleMode = compatibleMode;
+        this.fileFieldsDelimiter = fileFieldsDelimiter;
+        this.fileTempPath = fileTempPath;
     }
 }

seatunnel-connectors-v2/connector-clickhouse/src/main/java/org/apache/seatunnel/connectors/seatunnel/clickhouse/sink/file/ClickhouseFileSink.java

@@ -18,11 +18,15 @@
 package org.apache.seatunnel.connectors.seatunnel.clickhouse.sink.file;
 
 import static org.apache.seatunnel.connectors.seatunnel.clickhouse.config.ClickhouseConfig.CLICKHOUSE_LOCAL_PATH;
+import static org.apache.seatunnel.connectors.seatunnel.clickhouse.config.ClickhouseConfig.COMPATIBLE_MODE;
 import static org.apache.seatunnel.connectors.seatunnel.clickhouse.config.ClickhouseConfig.COPY_METHOD;
 import static org.apache.seatunnel.connectors.seatunnel.clickhouse.config.ClickhouseConfig.DATABASE;
 import static org.apache.seatunnel.connectors.seatunnel.clickhouse.config.ClickhouseConfig.FIELDS;
+import static org.apache.seatunnel.connectors.seatunnel.clickhouse.config.ClickhouseConfig.FILE_FIELDS_DELIMITER;
+import static org.apache.seatunnel.connectors.seatunnel.clickhouse.config.ClickhouseConfig.FILE_TEMP_PATH;
 import static org.apache.seatunnel.connectors.seatunnel.clickhouse.config.ClickhouseConfig.HOST;
 import static org.apache.seatunnel.connectors.seatunnel.clickhouse.config.ClickhouseConfig.NODE_ADDRESS;
+import static org.apache.seatunnel.connectors.seatunnel.clickhouse.config.ClickhouseConfig.NODE_FREE_PASSWORD;
 import static org.apache.seatunnel.connectors.seatunnel.clickhouse.config.ClickhouseConfig.NODE_PASS;
 import static org.apache.seatunnel.connectors.seatunnel.clickhouse.config.ClickhouseConfig.PASSWORD;
 import static org.apache.seatunnel.connectors.seatunnel.clickhouse.config.ClickhouseConfig.SHARDING_KEY;
@@ -31,7 +35,10 @@
 
 import org.apache.seatunnel.api.common.PrepareFailException;
 import org.apache.seatunnel.api.common.SeaTunnelAPIErrorCode;
+import org.apache.seatunnel.api.serialization.DefaultSerializer;
+import org.apache.seatunnel.api.serialization.Serializer;
 import org.apache.seatunnel.api.sink.SeaTunnelSink;
+import org.apache.seatunnel.api.sink.SinkAggregatedCommitter;
 import org.apache.seatunnel.api.sink.SinkWriter;
 import org.apache.seatunnel.api.table.type.SeaTunnelDataType;
 import org.apache.seatunnel.api.table.type.SeaTunnelRow;
@@ -46,8 +53,8 @@
 import org.apache.seatunnel.connectors.seatunnel.clickhouse.shard.Shard;
 import org.apache.seatunnel.connectors.seatunnel.clickhouse.shard.ShardMetadata;
 import org.apache.seatunnel.connectors.seatunnel.clickhouse.sink.client.ClickhouseProxy;
-import org.apache.seatunnel.connectors.seatunnel.clickhouse.state.CKAggCommitInfo;
-import org.apache.seatunnel.connectors.seatunnel.clickhouse.state.CKCommitInfo;
+import org.apache.seatunnel.connectors.seatunnel.clickhouse.state.CKFileAggCommitInfo;
+import org.apache.seatunnel.connectors.seatunnel.clickhouse.state.CKFileCommitInfo;
 import org.apache.seatunnel.connectors.seatunnel.clickhouse.state.ClickhouseSinkState;
 import org.apache.seatunnel.connectors.seatunnel.clickhouse.util.ClickhouseUtil;
 
@@ -62,10 +69,11 @@
 import java.util.ArrayList;
 import java.util.List;
 import java.util.Map;
+import java.util.Optional;
 import java.util.stream.Collectors;
 
 @AutoService(SeaTunnelSink.class)
-public class ClickhouseFileSink implements SeaTunnelSink<SeaTunnelRow, ClickhouseSinkState, CKCommitInfo, CKAggCommitInfo> {
+public class ClickhouseFileSink implements SeaTunnelSink<SeaTunnelRow, ClickhouseSinkState, CKFileCommitInfo, CKFileAggCommitInfo> {
 
     private FileReaderOption readerOption;
 
@@ -85,6 +93,10 @@ public void prepare(Config config) throws PrepareFailException {
         }
         Map<String, Object> defaultConfigs = ImmutableMap.<String, Object>builder()
             .put(COPY_METHOD.key(), COPY_METHOD.defaultValue().getName())
+            .put(NODE_FREE_PASSWORD.key(), NODE_FREE_PASSWORD.defaultValue())
+            .put(COMPATIBLE_MODE.key(), COMPATIBLE_MODE.defaultValue())
+            .put(FILE_TEMP_PATH.key(), FILE_TEMP_PATH.defaultValue())
+            .put(FILE_FIELDS_DELIMITER.key(), FILE_FIELDS_DELIMITER.defaultValue())
             .build();
 
         config = config.withFallback(ConfigFactory.parseMap(defaultConfigs));
@@ -126,8 +138,13 @@ public void prepare(Config config) throws PrepareFailException {
                 configObject -> configObject.toConfig().getString(PASSWORD.key())));
 
         proxy.close();
+
+        if (config.getString(FILE_FIELDS_DELIMITER.key()).length() != 1) {
+            throw new ClickhouseConnectorException(SeaTunnelAPIErrorCode.CONFIG_VALIDATION_FAILED, FILE_FIELDS_DELIMITER.key() + " must be a single character");
+        }
         this.readerOption = new FileReaderOption(shardMetadata, tableSchema, fields, config.getString(CLICKHOUSE_LOCAL_PATH.key()),
-            ClickhouseFileCopyMethod.from(config.getString(COPY_METHOD.key())), nodeUser, nodePassword);
+            ClickhouseFileCopyMethod.from(config.getString(COPY_METHOD.key())), nodeUser, config.getBoolean(NODE_FREE_PASSWORD.key()), nodePassword,
+            config.getBoolean(COMPATIBLE_MODE.key()), config.getString(FILE_TEMP_PATH.key()), config.getString(FILE_FIELDS_DELIMITER.key()));
     }
 
     @Override
@@ -141,7 +158,22 @@ public SeaTunnelDataType<SeaTunnelRow> getConsumedType() {
     }
 
     @Override
-    public SinkWriter<SeaTunnelRow, CKCommitInfo, ClickhouseSinkState> createWriter(SinkWriter.Context context) throws IOException {
+    public SinkWriter<SeaTunnelRow, CKFileCommitInfo, ClickhouseSinkState> createWriter(SinkWriter.Context context) throws IOException {
         return new ClickhouseFileSinkWriter(readerOption, context);
     }
+
+    @Override
+    public Optional<Serializer<CKFileCommitInfo>> getCommitInfoSerializer() {
+        return Optional.of(new DefaultSerializer<>());
+    }
+
+    @Override
+    public Optional<SinkAggregatedCommitter<CKFileCommitInfo, CKFileAggCommitInfo>> createAggregatedCommitter() throws IOException {
+        return Optional.of(new ClickhouseFileSinkAggCommitter(this.readerOption));
+    }
+
+    @Override
+    public Optional<Serializer<CKFileAggCommitInfo>> getAggregatedCommitInfoSerializer() {
+        return Optional.of(new DefaultSerializer<>());
+    }
 }