[Feature][connector-jdbc-base] avoid consume historical data from db to consume latest db updates events

[baisui1981]

Search before asking

• I had searched in the issues and found no similar feature requirement.

Description

Chunjun jdbc source connector start to consume an DB table，sometimes the table contains a massive historical data，in order to skipping historical data consuming. By now ,We can set param startLocation that corresponding with increColumn，but that is extremely troublesome.

In order to simplify the configuartion of Chunjun task，There can be a means that set a flag to inform the source task，avoiding consume historical data from db to consume latest db updates events ，however，param setting of startLocation is not required.

In my opion，can set the param startLocation to a special value token ，for example ‘latest’， In [JdbcInputFormat],

https://github.com/DTStack/chunjun/blob/master/chunjun-connectors/chunjun-connector-jdbc-base/src/main/java/com/dtstack/chunjun/connector/jdbc/source/JdbcInputFormat.java#L146-L164

if startLocation is latest, then execute the getMaxValueFromDb() to get the max value of increColumn, finally , to create JdbcInputSplit, the Pseudocode as below：

public JdbcInputSplit[] createSplitsInternalBySplitMod(int minNumSplits, String startLocation) {
        JdbcInputSplit[] splits = new JdbcInputSplit[minNumSplits];
        if (StringUtils.isNotBlank(startLocation)) {
            if ("latest".equals(startLocation)) {
              
                String maxValueFromDb = this.getMaxValueFromDb();
     
                return new JdbcInputSplit[]{new JdbcInputSplit(
                        0,
                        minNumSplits,
                        0,
                        maxValueFromDb,
                        null,
                        null,
                        null,
                        "mod",
                        jdbcConf.isPolling())};
            }
}

Use case

No response

Related issues

No response

Are you willing to submit a PR?

• Yes I am willing to submit a PR!

Code of Conduct

• I agree to follow this project's Code of Conduct

[FlechazoW]

It's a nice idea, and I prefer using another parameter, instead of using 'startLocation'.

[baisui1981]

It's a nice idea, and I prefer using another parameter, instead of using 'startLocation'.

which one do you like?

[mggger]

flink-cdc use StartupMode

public enum StartupMode {
    INITIAL,

    EARLIEST_OFFSET,

    LATEST_OFFSET,

    SPECIFIC_OFFSETS
}