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

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

@@ -122,3 +122,7 @@ 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
+
+- Fix generated data part name conflict and improve file commit logic [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

@@ -97,6 +97,11 @@ public class ClickhouseConfig {
         .defaultValue(ClickhouseFileCopyMethod.SCP).withDescription("The method of copy Clickhouse file");
 
     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
      */

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;
@@ -40,85 +43,15 @@ public FileReaderOption(ShardMetadata shardMetadata, Map<String, String> tableSc
                             List<String> fields, String clickhouseLocalPath,
                             ClickhouseFileCopyMethod copyMethod,
                             Map<String, String> nodeUser,
+                            boolean nodeFreePass,
                             Map<String, String> nodePassword) {
         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;
-    }
 }

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

@@ -23,14 +23,18 @@
 import static org.apache.seatunnel.connectors.seatunnel.clickhouse.config.ClickhouseConfig.FIELDS;
 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;
 import static org.apache.seatunnel.connectors.seatunnel.clickhouse.config.ClickhouseConfig.TABLE;
 import static org.apache.seatunnel.connectors.seatunnel.clickhouse.config.ClickhouseConfig.USERNAME;
 
 import org.apache.seatunnel.api.common.PrepareFailException;
+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;
@@ -43,8 +47,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;
 
@@ -59,10 +63,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;
 
@@ -78,7 +83,8 @@ public void prepare(Config config) throws PrepareFailException {
             throw new PrepareFailException(getPluginName(), PluginType.SINK, checkResult.getMsg());
         }
         Map<String, Object> defaultConfigs = ImmutableMap.<String, Object>builder()
-                .put(COPY_METHOD.key(), COPY_METHOD.defaultValue().getName())
+            .put(COPY_METHOD.key(), COPY_METHOD.defaultValue().getName())
+            .put(NODE_FREE_PASSWORD.key(), NODE_FREE_PASSWORD.defaultValue())
                 .build();
 
         config = config.withFallback(ConfigFactory.parseMap(defaultConfigs));
@@ -121,7 +127,7 @@ public void prepare(Config config) throws PrepareFailException {
 
         proxy.close();
         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);
     }
 
     @Override
@@ -135,7 +141,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<>());
+    }
 }

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

@@ -0,0 +1,94 @@
+/*
+ * 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.clickhouse.sink.file;
+
+import org.apache.seatunnel.api.sink.SinkAggregatedCommitter;
+import org.apache.seatunnel.common.utils.SeaTunnelException;
+import org.apache.seatunnel.connectors.seatunnel.clickhouse.config.FileReaderOption;
+import org.apache.seatunnel.connectors.seatunnel.clickhouse.shard.Shard;
+import org.apache.seatunnel.connectors.seatunnel.clickhouse.sink.client.ClickhouseProxy;
+import org.apache.seatunnel.connectors.seatunnel.clickhouse.state.CKFileAggCommitInfo;
+import org.apache.seatunnel.connectors.seatunnel.clickhouse.state.CKFileCommitInfo;
+
+import com.clickhouse.client.ClickHouseException;
+import com.clickhouse.client.ClickHouseRequest;
+import com.clickhouse.client.ClickHouseResponse;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+public class ClickhouseFileSinkAggCommitter implements SinkAggregatedCommitter<CKFileCommitInfo, CKFileAggCommitInfo> {
+
+    private final ClickhouseProxy proxy;
+    private final ClickhouseTable clickhouseTable;
+
+    public ClickhouseFileSinkAggCommitter(FileReaderOption readerOption) {
+        proxy = new ClickhouseProxy(readerOption.getShardMetadata().getDefaultShard().getNode());
+        clickhouseTable = proxy.getClickhouseTable(readerOption.getShardMetadata().getDatabase(),
+            readerOption.getShardMetadata().getTable());
+    }
+
+    @Override
+    public List<CKFileAggCommitInfo> commit(List<CKFileAggCommitInfo> aggregatedCommitInfo) throws IOException {
+        aggregatedCommitInfo.forEach(commitInfo -> commitInfo.getDetachedFiles().forEach((shard, files) -> {
+            try {
+                this.attachFileToClickhouse(shard, files);
+            } catch (ClickHouseException e) {
+                throw new SeaTunnelException("failed commit file to clickhouse", e);
+            }
+        }));
+        return new ArrayList<>();
+    }
+
+    @Override
+    public CKFileAggCommitInfo combine(List<CKFileCommitInfo> commitInfos) {
+        Map<Shard, List<String>> files = new HashMap<>();
+        commitInfos.forEach(infos -> infos.getDetachedFiles().forEach((shard, file) -> {
+            if (files.containsKey(shard)) {
+                files.get(shard).addAll(file);
+            } else {
+                files.put(shard, file);
+            }
+        }));
+        return new CKFileAggCommitInfo(files);
+    }
+
+    @Override
+    public void abort(List<CKFileAggCommitInfo> aggregatedCommitInfo) throws Exception {
+
+    }
+
+    @Override
+    public void close() throws IOException {
+        proxy.close();
+    }
+
+    private void attachFileToClickhouse(Shard shard, List<String> clickhouseLocalFiles) throws ClickHouseException {
+        ClickHouseRequest<?> request = proxy.getClickhouseConnection(shard);
+        for (String clickhouseLocalFile : clickhouseLocalFiles) {
+            ClickHouseResponse response = request.query(String.format("ALTER TABLE %s ATTACH PART '%s'",
+                clickhouseTable.getLocalTableName(),
+                clickhouseLocalFile.substring(clickhouseLocalFile.lastIndexOf("/") + 1))).executeAndWait();
+            response.close();
+        }
+    }
+
+}

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

@@ -22,6 +22,7 @@
 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.HOST;
+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;
@@ -44,6 +45,6 @@ public String factoryIdentifier() {
     @Override
     public OptionRule optionRule() {
         return OptionRule.builder().required(HOST, TABLE, DATABASE, USERNAME, PASSWORD, CLICKHOUSE_LOCAL_PATH)
-            .optional(COPY_METHOD, SHARDING_KEY, FIELDS, NODE_PASS).build();
+            .optional(COPY_METHOD, SHARDING_KEY, FIELDS, NODE_FREE_PASSWORD, NODE_PASS).build();
     }
 }