[INLONG-4332][Manager] Support ClickHouse load node (apache#4333)

Author: baomingyu

inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/sort/util/LoadNodeUtils.java

@@ -24,6 +24,7 @@
 import org.apache.inlong.manager.common.enums.SinkType;
 import org.apache.inlong.manager.common.pojo.sink.SinkFieldResponse;
 import org.apache.inlong.manager.common.pojo.sink.SinkResponse;
+import org.apache.inlong.manager.common.pojo.sink.ck.ClickHouseSinkResponse;
 import org.apache.inlong.manager.common.pojo.sink.hbase.HbaseSinkResponse;
 import org.apache.inlong.manager.common.pojo.sink.hive.HiveSinkResponse;
 import org.apache.inlong.manager.common.pojo.sink.kafka.KafkaSinkResponse;
@@ -36,6 +37,7 @@
 import org.apache.inlong.sort.protocol.node.format.DebeziumJsonFormat;
 import org.apache.inlong.sort.protocol.node.format.Format;
 import org.apache.inlong.sort.protocol.node.format.JsonFormat;
+import org.apache.inlong.sort.protocol.node.load.ClickHouseLoadNode;
 import org.apache.inlong.sort.protocol.node.load.HbaseLoadNode;
 import org.apache.inlong.sort.protocol.node.load.HiveLoadNode;
 import org.apache.inlong.sort.protocol.node.load.KafkaLoadNode;
@@ -76,6 +78,8 @@ public static LoadNode createLoadNode(SinkResponse sinkResponse) {
                 return createLoadNode((HbaseSinkResponse) sinkResponse);
             case POSTGRES:
                 return createLoadNode((PostgresSinkResponse) sinkResponse);
+            case CLICKHOUSE:
+                return createLoadNode((ClickHouseSinkResponse) sinkResponse);
             default:
                 throw new IllegalArgumentException(
                         String.format("Unsupported sinkType=%s to create loadNode", sinkType));
@@ -227,6 +231,7 @@ public static HbaseLoadNode createLoadNode(HbaseSinkResponse hbaseSinkResponse)
 
     /**
      * create postgres load node
+     *
      * @param postgresSinkResponse postgresSinkResponse
      * @return postgres load node
      */
@@ -247,8 +252,31 @@ public static PostgresLoadNode createLoadNode(PostgresSinkResponse postgresSinkR
                 postgresSinkResponse.getDbName() + "." + postgresSinkResponse.getTableName(),
                 postgresSinkResponse.getPrimaryKey());
     }
-    
-    /**f
+
+    /**
+     * create clickHouse load node
+     *
+     * @param clickHouseSinkResponse clickHouseSinkResponse
+     * @return
+     */
+    public static ClickHouseLoadNode createLoadNode(ClickHouseSinkResponse clickHouseSinkResponse) {
+        List<SinkFieldResponse> sinkFieldResponses = clickHouseSinkResponse.getFieldList();
+        String name = clickHouseSinkResponse.getSinkName();
+        List<FieldInfo> fields = sinkFieldResponses.stream()
+                .map(sinkFieldResponse -> FieldInfoUtils.parseSinkFieldInfo(sinkFieldResponse,
+                        name))
+                .collect(Collectors.toList());
+        List<FieldRelationShip> fieldRelationShips = parseSinkFields(sinkFieldResponses, name);
+        return new ClickHouseLoadNode(clickHouseSinkResponse.getSinkName(),
+                clickHouseSinkResponse.getSinkName(),
+                fields, fieldRelationShips, null, null, 1,
+                null, clickHouseSinkResponse.getTableName(),
+                clickHouseSinkResponse.getJdbcUrl(),
+                clickHouseSinkResponse.getUsername(),
+                clickHouseSinkResponse.getPassword());
+    }
+
+    /**
      * Parse information field of data sink.
      */
     public static List<FieldRelationShip> parseSinkFields(List<SinkFieldResponse> sinkFieldResponses, String sinkName) {

inlong-sort/sort-common/src/main/java/org/apache/inlong/sort/protocol/node/load/ClickHouseLoadNode.java

@@ -61,9 +61,6 @@ public class ClickHouseLoadNode extends LoadNode implements Serializable {
     @Nonnull
     private String password;
 
-    @JsonProperty("parFields")
-    List<FieldInfo> partitionFields;
-
     @JsonCreator
     public ClickHouseLoadNode(@JsonProperty("id") String id,
             @JsonProperty("name") String name,
@@ -76,14 +73,12 @@ public ClickHouseLoadNode(@JsonProperty("id") String id,
             @Nonnull @JsonProperty("tableName") String tableName,
             @Nonnull @JsonProperty("url") String url,
             @Nonnull @JsonProperty("userName") String userName,
-            @Nonnull @JsonProperty("passWord") String password,
-            @JsonProperty("parFields") List<FieldInfo> partitionFields) {
+            @Nonnull @JsonProperty("passWord") String password) {
         super(id, name, fields, fieldRelationShips, filters, filterStrategy, sinkParallelism, properties);
         this.tableName = Preconditions.checkNotNull(tableName, "table name is null");
         this.url = Preconditions.checkNotNull(url, "url is null");
         this.userName = Preconditions.checkNotNull(userName, "userName is null");
         this.password = Preconditions.checkNotNull(password, "password is null");
-        this.partitionFields = partitionFields;
     }
 
     @Override
@@ -110,7 +105,7 @@ public String getPrimaryKey() {
 
     @Override
     public List<FieldInfo> getPartitionFields() {
-        return partitionFields;
+        return super.getPartitionFields();
     }
 
 }

inlong-sort/sort-common/src/test/java/org/apache/inlong/sort/protocol/node/load/ClickHouseLoadNodeTest.java

@@ -44,7 +44,6 @@ public Node getTestObject() {
                 "ck_demo",
                 "jdbc:clickhouse://localhost:8023/default",
                 "root",
-                "root",
-                null);
+                "root");
     }
 }

inlong-sort/sort-core/src/test/java/org/apache/inlong/sort/parser/ClickHouseSqlParserTest.java

@@ -79,8 +79,7 @@ private ClickHouseLoadNode buildClickHouseLoadNode(String id) {
                 "ck_demo",
                 "jdbc:clickhouse://localhost:8123/demo",
                 "default",
-                "",
-                null);
+                "");
 
     }