[Feature][Connectors-v2] Support auto-increment id for FakeSource

docs/en/connector-v2/source/FakeSource.md

@@ -26,10 +26,12 @@ just for some test cases such as type conversion or connector new feature testin
 
 ## Source Options
 
-|          Name           |   Type   | Required |         Default         |                                                                                      Description                                                                                      |
+| Name                    |   Type   | Required | Default                 |                                                                                      Description                                                                                      |
 |-------------------------|----------|----------|-------------------------|---------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------|
 | tables_configs          | list     | no       | -                       | Define Multiple FakeSource, each item can contains the whole fake source config description below                                                                                     |
 | schema                  | config   | yes      | -                       | Define Schema information                                                                                                                                                             |
+| auto.increment.enabled  | boolean  | no       | false                   | Enable auto increment ID generation                                                                                                                                                                            |
+| auto.increment.start    | int      | no       |                         | Starting value for auto increment ID                                                                                                                                                                          |
 | rows                    | config   | no       | -                       | The row list of fake data output per degree of parallelism see title `Options rows Case`.                                                                                             |
 | row.num                 | int      | no       | 5                       | The total number of data generated per degree of parallelism                                                                                                                          |
 | split.num               | int      | no       | 1                       | the number of splits generated by the enumerator for each degree of parallelism                                                                                                       |
@@ -523,6 +525,33 @@ source {
 }
 
 
+```
+
+### Auto-increment primary key Example
+
+```hocon
+
+source {
+  # This is a example source plugin **only for test and demonstrate the feature source plugin**
+  FakeSource {
+    plugin_output = "fake"
+    auto.increment.enabled = true
+    auto.increment.start = 1000
+    row.num = 50000
+    schema = {
+      fields {
+        id = "int"
+        name = "string"
+        age = "int"
+      }
+      primaryKey {
+        name = "pk"
+        columnNames = [id]
+      }
+    }
+  }
+}
+
 ```
 
 ## Changelog

seatunnel-connectors-v2/connector-fake/src/main/java/org/apache/seatunnel/connectors/seatunnel/fake/config/FakeConfig.java

@@ -34,7 +34,10 @@
 import java.util.List;
 import java.util.Map;
 
+import static org.apache.seatunnel.api.options.EnvCommonOptions.PARALLELISM;
 import static org.apache.seatunnel.connectors.seatunnel.fake.config.FakeSourceOptions.ARRAY_SIZE;
+import static org.apache.seatunnel.connectors.seatunnel.fake.config.FakeSourceOptions.AUTO_INCREMENT_ENABLED;
+import static org.apache.seatunnel.connectors.seatunnel.fake.config.FakeSourceOptions.AUTO_INCREMENT_START;
 import static org.apache.seatunnel.connectors.seatunnel.fake.config.FakeSourceOptions.BIGINT_FAKE_MODE;
 import static org.apache.seatunnel.connectors.seatunnel.fake.config.FakeSourceOptions.BIGINT_MAX;
 import static org.apache.seatunnel.connectors.seatunnel.fake.config.FakeSourceOptions.BIGINT_MIN;
@@ -82,6 +85,9 @@
 @Builder
 @Getter
 public class FakeConfig implements Serializable {
+
+    @Builder.Default private int parallelism = PARALLELISM.defaultValue();
+
     @Builder.Default private int rowNum = ROW_NUM.defaultValue();
 
     @Builder.Default private int splitNum = SPLIT_NUM.defaultValue();
@@ -148,6 +154,10 @@ public class FakeConfig implements Serializable {
     @Builder.Default
     private FakeSourceOptions.FakeMode doubleFakeMode = DOUBLE_FAKE_MODE.defaultValue();
 
+    @Builder.Default private Boolean autoIncrementEnabled = AUTO_INCREMENT_ENABLED.defaultValue();
+
+    @Builder.Default private Long autoIncrementStart = AUTO_INCREMENT_START.defaultValue();
+
     private List<String> stringTemplate;
     private List<Integer> tinyintTemplate;
     private List<Integer> smallintTemplate;
@@ -170,6 +180,7 @@ public class FakeConfig implements Serializable {
 
     public static FakeConfig buildWithConfig(ReadonlyConfig readonlyConfig) {
         FakeConfigBuilder builder = FakeConfig.builder();
+        readonlyConfig.getOptional(PARALLELISM).ifPresent(builder::parallelism);
         builder.rowNum(readonlyConfig.get(ROW_NUM));
         builder.splitNum(readonlyConfig.get(SPLIT_NUM));
         builder.splitReadInterval(readonlyConfig.get(SPLIT_READ_INTERVAL));
@@ -204,6 +215,8 @@ public static FakeConfig buildWithConfig(ReadonlyConfig readonlyConfig) {
         readonlyConfig.getOptional(TIME_HOUR_TEMPLATE).ifPresent(builder::timeHourTemplate);
         readonlyConfig.getOptional(TIME_MINUTE_TEMPLATE).ifPresent(builder::timeMinuteTemplate);
         readonlyConfig.getOptional(TIME_SECOND_TEMPLATE).ifPresent(builder::timeSecondTemplate);
+        readonlyConfig.getOptional(AUTO_INCREMENT_ENABLED).ifPresent(builder::autoIncrementEnabled);
+        readonlyConfig.getOptional(AUTO_INCREMENT_START).ifPresent(builder::autoIncrementStart);
 
         readonlyConfig
                 .getOptional(TINYINT_MIN)

seatunnel-connectors-v2/connector-fake/src/main/java/org/apache/seatunnel/connectors/seatunnel/fake/config/FakeSourceOptions.java

@@ -299,6 +299,18 @@ public class FakeSourceOptions {
                     .defaultValue(FakeMode.RANGE)
                     .withDescription("The fake mode of generating double data");
 
+    public static final Option<Boolean> AUTO_INCREMENT_ENABLED =
+            Options.key("auto.increment.enabled")
+                    .booleanType()
+                    .defaultValue(false)
+                    .withDescription("Enable auto increment ID generation");
+
+    public static final Option<Long> AUTO_INCREMENT_START =
+            Options.key("auto.increment.start")
+                    .longType()
+                    .defaultValue(1L)
+                    .withDescription("Starting value for auto increment ID");
+
     public enum FakeMode {
         RANGE,
         TEMPLATE;

seatunnel-connectors-v2/connector-fake/src/main/java/org/apache/seatunnel/connectors/seatunnel/fake/source/FakeDataGenerator.java

@@ -57,8 +57,6 @@
 import java.util.function.Consumer;
 import java.util.function.Function;
 
-import static org.apache.seatunnel.api.table.type.SqlType.TIME;
-
 public class FakeDataGenerator {
     private static final String CURRENT_DATE = "CURRENT_DATE";
     private static final String CURRENT_TIME = "CURRENT_TIME";
@@ -72,15 +70,15 @@ public class FakeDataGenerator {
     private final FakeDataRandomUtils fakeDataRandomUtils;
     private String tableId;
 
-    public FakeDataGenerator(FakeConfig fakeConfig) {
+    public FakeDataGenerator(FakeConfig fakeConfig, String jobId) {
         this.catalogTable = fakeConfig.getCatalogTable();
         this.tableId = catalogTable.getTableId().toTablePath().toString();
         this.fakeConfig = fakeConfig;
         this.jsonDeserializationSchema =
                 fakeConfig.getFakeRows() == null
                         ? null
                         : new JsonDeserializationSchema(catalogTable, false, false);
-        this.fakeDataRandomUtils = new FakeDataRandomUtils(fakeConfig);
+        this.fakeDataRandomUtils = new FakeDataRandomUtils(fakeConfig, jobId);
     }
 
     private SeaTunnelRow convertRow(FakeConfig.RowData rowData) {

seatunnel-connectors-v2/connector-fake/src/main/java/org/apache/seatunnel/connectors/seatunnel/fake/source/FakeSource.java

@@ -32,10 +32,13 @@
 import org.apache.seatunnel.connectors.seatunnel.fake.config.MultipleTableFakeSourceConfig;
 import org.apache.seatunnel.connectors.seatunnel.fake.state.FakeSourceState;
 
+import lombok.extern.slf4j.Slf4j;
+
 import java.util.Collections;
 import java.util.List;
 import java.util.stream.Collectors;
 
+@Slf4j
 public class FakeSource
         implements SeaTunnelSource<SeaTunnelRow, FakeSourceSplit, FakeSourceState>,
                 SupportParallelism,
@@ -82,7 +85,8 @@ public SourceSplitEnumerator<FakeSourceSplit, FakeSourceState> restoreEnumerator
     @Override
     public SourceReader<SeaTunnelRow, FakeSourceSplit> createReader(
             SourceReader.Context readerContext) {
-        return new FakeSourceReader(readerContext, multipleTableFakeSourceConfig);
+        return new FakeSourceReader(
+                readerContext, multipleTableFakeSourceConfig, jobContext.getJobId());
     }
 
     @Override

seatunnel-connectors-v2/connector-fake/src/main/java/org/apache/seatunnel/connectors/seatunnel/fake/source/FakeSourceReader.java

@@ -48,8 +48,9 @@ public class FakeSourceReader implements SourceReader<SeaTunnelRow, FakeSourceSp
     private volatile long latestTimestamp = 0;
 
     public FakeSourceReader(
-            SourceReader.Context context,
-            MultipleTableFakeSourceConfig multipleTableFakeSourceConfig) {
+            Context context,
+            MultipleTableFakeSourceConfig multipleTableFakeSourceConfig,
+            String jobId) {
         this.context = context;
         this.multipleTableFakeSourceConfig = multipleTableFakeSourceConfig;
         this.fakeDataGeneratorMap =
@@ -62,7 +63,7 @@ public FakeSourceReader(
                                                         .getTableId()
                                                         .toTablePath()
                                                         .toString(),
-                                        FakeDataGenerator::new));
+                                        fakeConfig -> new FakeDataGenerator(fakeConfig, jobId)));
         this.minSplitReadInterval =
                 multipleTableFakeSourceConfig.getFakeConfigs().stream()
                         .map(FakeConfig::getSplitReadInterval)

seatunnel-connectors-v2/connector-fake/src/main/java/org/apache/seatunnel/connectors/seatunnel/fake/utils/AutoIncrementIdGenerator.java

@@ -0,0 +1,36 @@
+/*
+ * 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.fake.utils;
+
+import java.io.Serializable;
+import java.util.concurrent.atomic.AtomicLong;
+
+public class AutoIncrementIdGenerator implements Serializable {
+
+    private static final long serialVersionUID = 1L;
+
+    private final AtomicLong id;
+
+    public AutoIncrementIdGenerator(long start) {
+        this.id = new AtomicLong(start);
+    }
+
+    public Long getNextId() {
+        return id.getAndIncrement();
+    }
+}

seatunnel-connectors-v2/connector-fake/src/main/java/org/apache/seatunnel/connectors/seatunnel/fake/utils/FakeDataRandomUtils.java

@@ -38,9 +38,11 @@
 
 public class FakeDataRandomUtils {
     private final FakeConfig fakeConfig;
+    private final String jobId;
 
-    public FakeDataRandomUtils(FakeConfig fakeConfig) {
+    public FakeDataRandomUtils(FakeConfig fakeConfig, String jobId) {
         this.fakeConfig = fakeConfig;
+        this.jobId = jobId;
     }
 
     private static <T> T randomFromList(List<T> list) {
@@ -93,6 +95,22 @@ public Short randomSmallint(Column column) {
     }
 
     public Integer randomInt(Column column) {
+        if (fakeConfig.getAutoIncrementEnabled()
+                && IdGeneratorUtils.isPrimaryColumn(fakeConfig, column.getName())) {
+            if (fakeConfig.getAutoIncrementStart()
+                            + ((long) fakeConfig.getParallelism() * fakeConfig.getRowNum())
+                    > Integer.MAX_VALUE) {
+                throw new IllegalArgumentException(
+                        "The auto increment start value is too large, please check your configuration.");
+            }
+            return IdGeneratorUtils.getIdGenerator(jobId, fakeConfig, column.getName())
+                    .orElseThrow(
+                            () ->
+                                    new IllegalArgumentException(
+                                            "Auto increment is enabled, but no id generator found."))
+                    .getNextId()
+                    .intValue();
+        }
         List<Integer> intTemplate = fakeConfig.getIntTemplate();
         if (!CollectionUtils.isEmpty(intTemplate)) {
             return randomFromList(intTemplate);
@@ -101,6 +119,15 @@ public Integer randomInt(Column column) {
     }
 
     public Long randomBigint(Column column) {
+        if (fakeConfig.getAutoIncrementEnabled()
+                && IdGeneratorUtils.isPrimaryColumn(fakeConfig, column.getName())) {
+            return IdGeneratorUtils.getIdGenerator(jobId, fakeConfig, column.getName())
+                    .orElseThrow(
+                            () ->
+                                    new IllegalArgumentException(
+                                            "Auto increment is enabled, but no id generator found."))
+                    .getNextId();
+        }
         List<Long> bigTemplate = fakeConfig.getBigTemplate();
         if (!CollectionUtils.isEmpty(bigTemplate)) {
             return randomFromList(bigTemplate);

seatunnel-connectors-v2/connector-fake/src/main/java/org/apache/seatunnel/connectors/seatunnel/fake/utils/IdGeneratorUtils.java

@@ -0,0 +1,72 @@
+/*
+ * 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.fake.utils;
+
+import org.apache.seatunnel.shade.com.google.common.cache.Cache;
+import org.apache.seatunnel.shade.com.google.common.cache.CacheBuilder;
+
+import org.apache.seatunnel.api.table.catalog.CatalogTable;
+import org.apache.seatunnel.api.table.catalog.PrimaryKey;
+import org.apache.seatunnel.connectors.seatunnel.fake.config.FakeConfig;
+
+import java.util.List;
+import java.util.Optional;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.TimeUnit;
+
+public class IdGeneratorUtils {
+
+    private static final Cache<String, AutoIncrementIdGenerator> idGenerators =
+            CacheBuilder.newBuilder()
+                    .maximumSize(1000)
+                    .expireAfterWrite(30, TimeUnit.MINUTES)
+                    .build();
+
+    public static synchronized Optional<AutoIncrementIdGenerator> getIdGenerator(
+            String jobId, FakeConfig fakeConfig, String columnName) {
+        CatalogTable catalogTable = fakeConfig.getCatalogTable();
+        String tableName = catalogTable.getTableId().getTableName();
+        String key = String.format("%s:%s_%s", jobId, tableName, columnName);
+        AutoIncrementIdGenerator idGenerator = null;
+        try {
+            idGenerator =
+                    idGenerators.get(
+                            key,
+                            () -> {
+                                if (isPrimaryColumn(fakeConfig, columnName)) {
+                                    return new AutoIncrementIdGenerator(
+                                            fakeConfig.getAutoIncrementStart());
+                                } else {
+                                    return null;
+                                }
+                            });
+        } catch (ExecutionException e) {
+            throw new RuntimeException(e);
+        }
+        return Optional.ofNullable(idGenerator);
+    }
+
+    public static boolean isPrimaryColumn(FakeConfig fakeConfig, String columnName) {
+        PrimaryKey primaryKey = fakeConfig.getCatalogTable().getTableSchema().getPrimaryKey();
+        if (primaryKey == null) {
+            return false;
+        }
+        List<String> primaryColumns = primaryKey.getColumnNames();
+        return primaryColumns != null && primaryColumns.contains(columnName);
+    }
+}