[Fix] [Clickhouse] Parallelism makes data duplicate (#8916)

Author: FrommyMind

Diff for: docs/en/connector-v2/source/Clickhouse.md

@@ -67,7 +67,6 @@ The following example demonstrates how to create a data synchronization job that
 ```bash
 # Set the basic configuration of the task to be performed
 env {
-  parallelism = 10
   job.mode = "BATCH"
 }
 

Diff for: seatunnel-connectors-v2/connector-clickhouse/src/main/java/org/apache/seatunnel/connectors/seatunnel/clickhouse/source/ClickhouseSource.java

@@ -18,33 +18,30 @@
 package org.apache.seatunnel.connectors.seatunnel.clickhouse.source;
 
 import org.apache.seatunnel.api.source.Boundedness;
-import org.apache.seatunnel.api.source.SeaTunnelSource;
-import org.apache.seatunnel.api.source.SourceReader;
-import org.apache.seatunnel.api.source.SourceSplitEnumerator;
-import org.apache.seatunnel.api.source.SupportColumnProjection;
-import org.apache.seatunnel.api.source.SupportParallelism;
 import org.apache.seatunnel.api.table.catalog.CatalogTable;
 import org.apache.seatunnel.api.table.type.SeaTunnelRow;
-import org.apache.seatunnel.connectors.seatunnel.clickhouse.state.ClickhouseSourceState;
+import org.apache.seatunnel.api.table.type.SeaTunnelRowType;
+import org.apache.seatunnel.connectors.seatunnel.common.source.AbstractSingleSplitReader;
+import org.apache.seatunnel.connectors.seatunnel.common.source.AbstractSingleSplitSource;
+import org.apache.seatunnel.connectors.seatunnel.common.source.SingleSplitReaderContext;
 
 import com.clickhouse.client.ClickHouseNode;
 
 import java.util.Collections;
 import java.util.List;
 
-public class ClickhouseSource
-        implements SeaTunnelSource<SeaTunnelRow, ClickhouseSourceSplit, ClickhouseSourceState>,
-                SupportParallelism,
-                SupportColumnProjection {
+public class ClickhouseSource extends AbstractSingleSplitSource<SeaTunnelRow> {
 
-    private List<ClickHouseNode> servers;
-    private CatalogTable catalogTable;
-    private String sql;
+    private final List<ClickHouseNode> servers;
+    private final CatalogTable catalogTable;
+    private final String sql;
+    private final SeaTunnelRowType rowTypeInfo;
 
     public ClickhouseSource(List<ClickHouseNode> servers, CatalogTable catalogTable, String sql) {
         this.servers = servers;
         this.catalogTable = catalogTable;
         this.sql = sql;
+        this.rowTypeInfo = catalogTable.getSeaTunnelRowType();
     }
 
     @Override
@@ -63,24 +60,8 @@ public List<CatalogTable> getProducedCatalogTables() {
     }
 
     @Override
-    public SourceReader<SeaTunnelRow, ClickhouseSourceSplit> createReader(
-            SourceReader.Context readerContext) throws Exception {
-        return new ClickhouseSourceReader(
-                servers, readerContext, this.catalogTable.getSeaTunnelRowType(), sql);
-    }
-
-    @Override
-    public SourceSplitEnumerator<ClickhouseSourceSplit, ClickhouseSourceState> createEnumerator(
-            SourceSplitEnumerator.Context<ClickhouseSourceSplit> enumeratorContext)
-            throws Exception {
-        return new ClickhouseSourceSplitEnumerator(enumeratorContext);
-    }
-
-    @Override
-    public SourceSplitEnumerator<ClickhouseSourceSplit, ClickhouseSourceState> restoreEnumerator(
-            SourceSplitEnumerator.Context<ClickhouseSourceSplit> enumeratorContext,
-            ClickhouseSourceState checkpointState)
-            throws Exception {
-        return new ClickhouseSourceSplitEnumerator(enumeratorContext);
+    public AbstractSingleSplitReader<SeaTunnelRow> createReader(
+            SingleSplitReaderContext readerContext) {
+        return new ClickhouseSourceReader(servers, readerContext, sql, rowTypeInfo);
     }
 }

Diff for: seatunnel-connectors-v2/connector-clickhouse/src/main/java/org/apache/seatunnel/connectors/seatunnel/clickhouse/source/ClickhouseSourceFactory.java

@@ -68,7 +68,6 @@ public String factoryIdentifier() {
             TableSource<T, SplitT, StateT> createSource(TableSourceFactoryContext context) {
         ReadonlyConfig readonlyConfig = context.getOptions();
         List<ClickHouseNode> nodes = ClickhouseUtil.createNodes(readonlyConfig);
-
         String sql = readonlyConfig.get(SQL);
         ClickHouseNode currentServer = nodes.get(ThreadLocalRandom.current().nextInt(nodes.size()));
         try (ClickHouseClient client = ClickHouseClient.newInstance(currentServer.getProtocol());

Diff for: seatunnel-connectors-v2/connector-clickhouse/src/main/java/org/apache/seatunnel/connectors/seatunnel/clickhouse/source/ClickhouseSourceReader.java

@@ -17,12 +17,12 @@
 
 package org.apache.seatunnel.connectors.seatunnel.clickhouse.source;
 
-import org.apache.seatunnel.api.source.Boundedness;
 import org.apache.seatunnel.api.source.Collector;
-import org.apache.seatunnel.api.source.SourceReader;
 import org.apache.seatunnel.api.table.type.SeaTunnelRow;
 import org.apache.seatunnel.api.table.type.SeaTunnelRowType;
 import org.apache.seatunnel.connectors.seatunnel.clickhouse.util.TypeConvertUtil;
+import org.apache.seatunnel.connectors.seatunnel.common.source.AbstractSingleSplitReader;
+import org.apache.seatunnel.connectors.seatunnel.common.source.SingleSplitReaderContext;
 
 import com.clickhouse.client.ClickHouseClient;
 import com.clickhouse.client.ClickHouseFormat;
@@ -32,34 +32,28 @@
 import lombok.extern.slf4j.Slf4j;
 
 import java.io.IOException;
-import java.util.ArrayList;
-import java.util.Collections;
 import java.util.List;
 import java.util.Random;
 
 @Slf4j
-public class ClickhouseSourceReader implements SourceReader<SeaTunnelRow, ClickhouseSourceSplit> {
+public class ClickhouseSourceReader extends AbstractSingleSplitReader<SeaTunnelRow> {
 
     private final List<ClickHouseNode> servers;
     private ClickHouseClient client;
     private final SeaTunnelRowType rowTypeInfo;
-    private final SourceReader.Context readerContext;
+    private final SingleSplitReaderContext readerContext;
     private ClickHouseRequest<?> request;
     private final String sql;
-    private volatile boolean noMoreSplit;
-
-    private final List<ClickhouseSourceSplit> splits;
 
     ClickhouseSourceReader(
             List<ClickHouseNode> servers,
-            SourceReader.Context readerContext,
-            SeaTunnelRowType rowTypeInfo,
-            String sql) {
+            SingleSplitReaderContext readerContext,
+            String sql,
+            SeaTunnelRowType rowTypeInfo) {
         this.servers = servers;
         this.readerContext = readerContext;
-        this.rowTypeInfo = rowTypeInfo;
         this.sql = sql;
-        this.splits = new ArrayList<>();
+        this.rowTypeInfo = rowTypeInfo;
     }
 
     @Override
@@ -80,57 +74,31 @@ public void close() throws IOException {
     @Override
     public void pollNext(Collector<SeaTunnelRow> output) throws Exception {
         synchronized (output.getCheckpointLock()) {
-            if (!splits.isEmpty()) {
-                try (ClickHouseResponse response = this.request.query(sql).executeAndWait()) {
-                    response.stream()
-                            .forEach(
-                                    record -> {
-                                        Object[] values =
-                                                new Object[this.rowTypeInfo.getFieldNames().length];
-                                        for (int i = 0; i < record.size(); i++) {
-                                            if (record.getValue(i).isNullOrEmpty()) {
-                                                values[i] = null;
-                                            } else {
-                                                values[i] =
-                                                        TypeConvertUtil.valueUnwrap(
-                                                                this.rowTypeInfo.getFieldType(i),
-                                                                record.getValue(i));
-                                            }
+            try (ClickHouseResponse response = this.request.query(sql).executeAndWait()) {
+                response.stream()
+                        .forEach(
+                                record -> {
+                                    Object[] values =
+                                            new Object[this.rowTypeInfo.getFieldNames().length];
+                                    for (int i = 0; i < record.size(); i++) {
+                                        if (record.getValue(i).isNullOrEmpty()) {
+                                            values[i] = null;
+                                        } else {
+                                            values[i] =
+                                                    TypeConvertUtil.valueUnwrap(
+                                                            this.rowTypeInfo.getFieldType(i),
+                                                            record.getValue(i));
                                         }
-                                        output.collect(new SeaTunnelRow(values));
-                                    });
-                }
-                signalNoMoreElement();
-            }
-            if (noMoreSplit
-                    && splits.isEmpty()
-                    && Boundedness.BOUNDED.equals(readerContext.getBoundedness())) {
-                signalNoMoreElement();
+                                    }
+                                    output.collect(new SeaTunnelRow(values));
+                                });
             }
+            signalNoMoreElement();
         }
     }
 
     private void signalNoMoreElement() {
         log.info("Closed the bounded ClickHouse source");
         this.readerContext.signalNoMoreElement();
-        this.splits.clear();
     }
-
-    @Override
-    public List<ClickhouseSourceSplit> snapshotState(long checkpointId) throws Exception {
-        return Collections.emptyList();
-    }
-
-    @Override
-    public void addSplits(List<ClickhouseSourceSplit> splits) {
-        this.splits.addAll(splits);
-    }
-
-    @Override
-    public void handleNoMoreSplits() {
-        noMoreSplit = true;
-    }
-
-    @Override
-    public void notifyCheckpointComplete(long checkpointId) throws Exception {}
 }

Diff for: seatunnel-connectors-v2/connector-clickhouse/src/main/java/org/apache/seatunnel/connectors/seatunnel/clickhouse/source/ClickhouseSourceSplit.java

@@ -19,7 +19,7 @@
 ######
 
 env {
-  parallelism = 3
+  parallelism = 1
   job.mode = "BATCH"
 }