[cdc-pipeline-connector][doris] Introduce Doris cdc pipeline DataSink

flink-cdc-common/src/main/java/com/ververica/cdc/common/data/GenericStringData.java

@@ -44,6 +44,11 @@ public byte[] toBytes() {
         }
     }
 
+    @Override
+    public String toString() {
+        return javaStr;
+    }
+
     @Override
     public int compareTo(@Nonnull StringData o) {
         GenericStringData other = (GenericStringData) o;

flink-cdc-common/src/main/java/com/ververica/cdc/common/schema/Schema.java

@@ -172,6 +172,32 @@ public static Schema.Builder newBuilder() {
         return new Builder();
     }
 
+    @Override
+    public String toString() {
+        final StringBuilder sb = new StringBuilder();
+        sb.append("columns={");
+        for (int i = 0; i < columns.size(); i++) {
+            sb.append(columns.get(i).asSummaryString());
+            if (i != columns.size() - 1) {
+                sb.append(",");
+            }
+        }
+        sb.append("}");
+        sb.append(", primaryKeys=").append(String.join(";", primaryKeys));
+        sb.append(", options=").append(describeOptions());
+
+        return sb.toString();
+    }
+
+    public String describeOptions() {
+        StringBuilder stringBuilder = new StringBuilder("(");
+        if (options != null && !options.isEmpty()) {
+            stringBuilder.append(options);
+        }
+        stringBuilder.append(")");
+        return stringBuilder.toString();
+    }
+
     // -----------------------------------------------------------------------------------
 
     /** A builder for constructing an immutable but still unresolved {@link Schema}. */

flink-cdc-common/src/main/java/com/ververica/cdc/common/types/DataTypes.java

@@ -426,8 +426,12 @@ public static OptionalInt getLength(DataType dataType) {
         return dataType.accept(LENGTH_EXTRACTOR);
     }
 
-    private static final PrecisionExtractor PRECISION_EXTRACTOR = new PrecisionExtractor();
+    public static OptionalInt getScale(DataType dataType) {
+        return dataType.accept(SCALE_EXTRACTOR);
+    }
 
+    private static final PrecisionExtractor PRECISION_EXTRACTOR = new PrecisionExtractor();
+    private static final ScaleExtractor SCALE_EXTRACTOR = new ScaleExtractor();
     private static final LengthExtractor LENGTH_EXTRACTOR = new LengthExtractor();
 
     private static class PrecisionExtractor extends DataTypeDefaultVisitor<OptionalInt> {
@@ -463,6 +467,17 @@ protected OptionalInt defaultMethod(DataType dataType) {
         }
     }
 
+    private static class ScaleExtractor extends DataTypeDefaultVisitor<OptionalInt> {
+        @Override
+        public OptionalInt visit(DecimalType decimalType) {
+            return OptionalInt.of(decimalType.getScale());
+        }
+        @Override
+        protected OptionalInt defaultMethod(DataType dataType) {
+            return OptionalInt.empty();
+        }
+    }
+
     private static class LengthExtractor extends DataTypeDefaultVisitor<OptionalInt> {
 
         @Override

flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connectors-doris/pom.xml

@@ -0,0 +1,68 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+Copyright 2023 Ververica Inc.
+
+Licensed 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.
+-->
+<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/maven-v4_0_0.xsd">
+    <modelVersion>4.0.0</modelVersion>
+    <parent>
+        <groupId>com.ververica</groupId>
+        <artifactId>flink-cdc-pipeline-connectors</artifactId>
+        <version>3.0-SNAPSHOT</version>
+    </parent>
+    <artifactId>flink-cdc-pipeline-connectors-doris</artifactId>
+    <name>flink-cdc-pipeline-connectors-doris</name>
+
+    <dependencies>
+        <dependency>
+            <groupId>com.ververica</groupId>
+            <artifactId>flink-cdc-common</artifactId>
+            <version>${revision}</version>
+        </dependency>
+
+        <dependency>
+            <groupId>com.ververica</groupId>
+            <artifactId>flink-cdc-runtime</artifactId>
+            <version>${revision}</version>
+        </dependency>
+
+        <dependency>
+            <groupId>com.ververica</groupId>
+            <artifactId>flink-cdc-composer</artifactId>
+            <version>${revision}</version>
+            <scope>test</scope>
+        </dependency>
+
+        <dependency>
+            <groupId>org.apache.flink</groupId>
+            <artifactId>flink-runtime</artifactId>
+            <version>${flink.version}</version>
+            <scope>provided</scope>
+        </dependency>
+
+        <dependency>
+            <groupId>org.apache.doris</groupId>
+            <artifactId>flink-doris-connector-${flink.major.version}</artifactId>
+            <version>1.5.0-SNAPSHOT</version>
+            <scope>provided</scope>
+        </dependency>
+
+        <dependency>
+            <groupId>org.apache.flink</groupId>
+            <artifactId>flink-clients</artifactId>
+            <version>${flink.version}</version>
+            <scope>test</scope>
+        </dependency>
+    </dependencies>
+</project>

flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connectors-doris/src/main/java/com/ververica/cdc/connectors/doris/sink/DorisDataSink.java

@@ -0,0 +1,58 @@
+/*
+ * Copyright 2023 Ververica Inc.
+ *
+ * Licensed 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 com.ververica.cdc.connectors.doris.sink;
+
+import com.ververica.cdc.common.sink.DataSink;
+import com.ververica.cdc.common.sink.EventSinkProvider;
+import com.ververica.cdc.common.sink.FlinkSinkProvider;
+import com.ververica.cdc.common.sink.MetadataApplier;
+import org.apache.doris.flink.cfg.DorisExecutionOptions;
+import org.apache.doris.flink.cfg.DorisOptions;
+import org.apache.doris.flink.cfg.DorisReadOptions;
+import org.apache.doris.flink.sink.DorisSink;
+import org.apache.doris.flink.sink.batch.DorisBatchSink;
+
+import java.io.Serializable;
+
+
+public class DorisDataSink implements DataSink, Serializable {
+
+    private final DorisOptions dorisOptions;
+    private final DorisReadOptions readOptions;
+    private final DorisExecutionOptions executionOptions;
+
+    public DorisDataSink(DorisOptions dorisOptions, DorisReadOptions dorisReadOptions, DorisExecutionOptions dorisExecutionOptions) {
+        this.dorisOptions = dorisOptions;
+        this.readOptions = dorisReadOptions;
+        this.executionOptions = dorisExecutionOptions;
+    }
+
+    @Override
+    public EventSinkProvider getEventSinkProvider() {
+        if(!executionOptions.enableBatchMode()){
+            return FlinkSinkProvider.of(new DorisSink<>(dorisOptions, readOptions, executionOptions, new DorisEventSerializer()));
+        } else {
+            return FlinkSinkProvider.of(new DorisBatchSink<>(dorisOptions, readOptions, executionOptions, new DorisEventSerializer()));
+        }
+    }
+
+    @Override
+    public MetadataApplier getMetadataApplier() {
+        return new DorisMetadataApplier(dorisOptions);
+    }
+
+}

flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connectors-doris/src/main/java/com/ververica/cdc/connectors/doris/sink/DorisDataSinkFactory.java

@@ -0,0 +1,129 @@
+/*
+ * Copyright 2023 Ververica Inc.
+ *
+ * Licensed 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 com.ververica.cdc.connectors.doris.sink;
+
+import com.ververica.cdc.common.annotation.Internal;
+import com.ververica.cdc.common.configuration.ConfigOption;
+import com.ververica.cdc.common.configuration.Configuration;
+import com.ververica.cdc.common.factories.DataSinkFactory;
+import com.ververica.cdc.common.sink.DataSink;
+import org.apache.doris.flink.cfg.DorisExecutionOptions;
+import org.apache.doris.flink.cfg.DorisOptions;
+import org.apache.doris.flink.cfg.DorisReadOptions;
+
+import java.util.HashSet;
+import java.util.Set;
+
+import static com.ververica.cdc.connectors.doris.sink.DorisDataSinkOptions.AUTO_REDIRECT;
+import static com.ververica.cdc.connectors.doris.sink.DorisDataSinkOptions.BENODES;
+import static com.ververica.cdc.connectors.doris.sink.DorisDataSinkOptions.FENODES;
+import static com.ververica.cdc.connectors.doris.sink.DorisDataSinkOptions.JDBC_URL;
+import static com.ververica.cdc.connectors.doris.sink.DorisDataSinkOptions.PASSWORD;
+import static com.ververica.cdc.connectors.doris.sink.DorisDataSinkOptions.SINK_BUFFER_COUNT;
+import static com.ververica.cdc.connectors.doris.sink.DorisDataSinkOptions.SINK_BUFFER_FLUSH_INTERVAL;
+import static com.ververica.cdc.connectors.doris.sink.DorisDataSinkOptions.SINK_BUFFER_FLUSH_MAX_BYTES;
+import static com.ververica.cdc.connectors.doris.sink.DorisDataSinkOptions.SINK_BUFFER_FLUSH_MAX_ROWS;
+import static com.ververica.cdc.connectors.doris.sink.DorisDataSinkOptions.SINK_BUFFER_SIZE;
+import static com.ververica.cdc.connectors.doris.sink.DorisDataSinkOptions.SINK_CHECK_INTERVAL;
+import static com.ververica.cdc.connectors.doris.sink.DorisDataSinkOptions.SINK_ENABLE_2PC;
+import static com.ververica.cdc.connectors.doris.sink.DorisDataSinkOptions.SINK_ENABLE_BATCH_MODE;
+import static com.ververica.cdc.connectors.doris.sink.DorisDataSinkOptions.SINK_ENABLE_DELETE;
+import static com.ververica.cdc.connectors.doris.sink.DorisDataSinkOptions.SINK_FLUSH_QUEUE_SIZE;
+import static com.ververica.cdc.connectors.doris.sink.DorisDataSinkOptions.SINK_LABEL_PREFIX;
+import static com.ververica.cdc.connectors.doris.sink.DorisDataSinkOptions.SINK_MAX_RETRIES;
+import static com.ververica.cdc.connectors.doris.sink.DorisDataSinkOptions.TABLE_IDENTIFIER;
+import static com.ververica.cdc.connectors.doris.sink.DorisDataSinkOptions.USERNAME;
+
+
+/**
+ * A dummy {@link DataSinkFactory} to create {@link DorisDataSink}.
+ */
+@Internal
+public class DorisDataSinkFactory implements DataSinkFactory {
+    @Override
+    public DataSink createDataSink(Context context) {
+        Configuration config = context.getConfiguration();
+        DorisOptions.Builder optionsBuilder = DorisOptions.builder();
+        DorisExecutionOptions.Builder executionBuilder = DorisExecutionOptions.builderDefaults();
+        config.getOptional(FENODES).ifPresent(optionsBuilder::setFenodes);
+        config.getOptional(BENODES).ifPresent(optionsBuilder::setBenodes);
+        config.getOptional(TABLE_IDENTIFIER).ifPresent(optionsBuilder::setTableIdentifier);
+        config.getOptional(USERNAME).ifPresent(optionsBuilder::setUsername);
+        config.getOptional(PASSWORD).ifPresent(optionsBuilder::setPassword);
+        config.getOptional(JDBC_URL).ifPresent(optionsBuilder::setJdbcUrl);
+        config.getOptional(AUTO_REDIRECT).ifPresent(optionsBuilder::setAutoRedirect);
+
+        config.getOptional(SINK_CHECK_INTERVAL).ifPresent(executionBuilder::setCheckInterval);
+        config.getOptional(SINK_MAX_RETRIES).ifPresent(executionBuilder::setMaxRetries);
+        config.getOptional(SINK_ENABLE_DELETE).ifPresent(executionBuilder::setDeletable);
+        config.getOptional(SINK_LABEL_PREFIX).ifPresent(executionBuilder::setLabelPrefix);
+        config.getOptional(SINK_BUFFER_SIZE).ifPresent(executionBuilder::setBufferSize);
+        config.getOptional(SINK_BUFFER_COUNT).ifPresent(executionBuilder::setBufferCount);
+        config.getOptional(SINK_BUFFER_FLUSH_MAX_ROWS).ifPresent(executionBuilder::setBufferFlushMaxRows);
+        config.getOptional(SINK_BUFFER_FLUSH_MAX_BYTES).ifPresent(executionBuilder::setBufferFlushMaxBytes);
+        config.getOptional(SINK_FLUSH_QUEUE_SIZE).ifPresent(executionBuilder::setFlushQueueSize);
+        config.getOptional(SINK_BUFFER_FLUSH_INTERVAL).ifPresent(v-> executionBuilder.setBufferFlushIntervalMs(v.toMillis()));
+
+        config.getOptional(SINK_ENABLE_2PC).ifPresent(b -> {
+            if (b) {
+                executionBuilder.enable2PC();
+            } else {
+                executionBuilder.disable2PC();
+            }
+        } );
+        config.getOptional(SINK_ENABLE_BATCH_MODE).ifPresent(b-> { if(b) executionBuilder.enableBatchMode(); });
+        return new DorisDataSink(optionsBuilder.build(), DorisReadOptions.builder().build(), executionBuilder.build());
+    }
+
+    @Override
+    public String identifier() {
+        return "doris-sink";
+    }
+
+    @Override
+    public Set<ConfigOption<?>> requiredOptions() {
+        return new HashSet<>();
+    }
+
+    @Override
+    public Set<ConfigOption<?>> optionalOptions() {
+        Set<ConfigOption<?>> options = new HashSet<>();
+        options.add(FENODES);
+        options.add(BENODES);
+        options.add(TABLE_IDENTIFIER);
+        options.add(USERNAME);
+        options.add(PASSWORD);
+        options.add(JDBC_URL);
+        options.add(AUTO_REDIRECT);
+
+        options.add(SINK_CHECK_INTERVAL);
+        options.add(SINK_ENABLE_2PC);
+        options.add(SINK_MAX_RETRIES);
+        options.add(SINK_ENABLE_DELETE);
+        options.add(SINK_LABEL_PREFIX);
+        options.add(SINK_BUFFER_SIZE);
+        options.add(SINK_BUFFER_COUNT);
+
+        options.add(SINK_ENABLE_BATCH_MODE);
+        options.add(SINK_BUFFER_FLUSH_MAX_ROWS);
+        options.add(SINK_BUFFER_FLUSH_MAX_BYTES);
+        options.add(SINK_FLUSH_QUEUE_SIZE);
+        options.add(SINK_BUFFER_FLUSH_INTERVAL);
+
+        return options;
+    }
+}