[feat][jdbc] Separate the relationship between incrementColumn with RestoreColumn and splitPk (#1258)

Author: Paddy0523

chunjun-connectors/chunjun-connector-jdbc-base/src/main/java/com/dtstack/chunjun/connector/jdbc/conf/JdbcConf.java

@@ -66,6 +66,8 @@ public class JdbcConf extends ChunJunCommonConf implements Serializable {
     private boolean polling = false;
     /** 字段名称 */
     private String increColumn;
+    /** Whether an OrderBy sort is required,increment mode need set to true. */
+    private boolean isOrderBy = true;
     /** 字段索引 */
     private int increColumnIndex = -1;
     /** 字段类型 */
@@ -295,6 +297,14 @@ public void setIncreColumn(String increColumn) {
         this.increColumn = increColumn;
     }
 
+    public boolean isOrderBy() {
+        return isOrderBy;
+    }
+
+    public void setOrderBy(boolean orderBy) {
+        isOrderBy = orderBy;
+    }
+
     public int getIncreColumnIndex() {
         return increColumnIndex;
     }

chunjun-connectors/chunjun-connector-jdbc-base/src/main/java/com/dtstack/chunjun/connector/jdbc/dialect/JdbcDialect.java

@@ -24,8 +24,14 @@
 import com.dtstack.chunjun.connector.jdbc.source.JdbcInputSplit;
 import com.dtstack.chunjun.connector.jdbc.statement.FieldNamedPreparedStatement;
 import com.dtstack.chunjun.connector.jdbc.util.JdbcUtil;
+import com.dtstack.chunjun.connector.jdbc.util.key.DateTypeUtil;
+import com.dtstack.chunjun.connector.jdbc.util.key.KeyUtil;
+import com.dtstack.chunjun.connector.jdbc.util.key.NumericTypeUtil;
+import com.dtstack.chunjun.connector.jdbc.util.key.TimestampTypeUtil;
 import com.dtstack.chunjun.converter.AbstractRowConverter;
 import com.dtstack.chunjun.converter.RawTypeConverter;
+import com.dtstack.chunjun.enums.ColumnType;
+import com.dtstack.chunjun.throwable.ChunJunRuntimeException;
 
 import org.apache.flink.table.api.TableSchema;
 import org.apache.flink.table.api.ValidationException;
@@ -36,6 +42,7 @@
 import org.apache.commons.lang3.StringUtils;
 
 import java.io.Serializable;
+import java.math.BigInteger;
 import java.sql.ResultSet;
 import java.util.Arrays;
 import java.util.Objects;
@@ -370,4 +377,23 @@ default String getSplitModFilter(JdbcInputSplit split, String splitPkName) {
                 " mod(%s, %s) = %s",
                 quoteIdentifier(splitPkName), split.getTotalNumberOfSplits(), split.getMod());
     }
+
+    default KeyUtil<?, BigInteger> initKeyUtil(String incrementName, String incrementType) {
+        switch (ColumnType.getType(incrementType)) {
+            case TIMESTAMP:
+            case DATETIME:
+                return new TimestampTypeUtil();
+            case DATE:
+                return new DateTypeUtil();
+            default:
+                if (ColumnType.isNumberType(incrementType)) {
+                    return new NumericTypeUtil();
+                } else {
+                    throw new ChunJunRuntimeException(
+                            String.format(
+                                    "Unsupported columnType [%s], columnName [%s]",
+                                    incrementType, incrementName));
+                }
+        }
+    }
 }

chunjun-connectors/chunjun-connector-jdbc-base/src/main/java/com/dtstack/chunjun/connector/jdbc/source/JdbcDynamicTableSource.java

@@ -23,6 +23,7 @@
 import com.dtstack.chunjun.connector.jdbc.dialect.JdbcDialect;
 import com.dtstack.chunjun.connector.jdbc.lookup.JdbcAllTableFunction;
 import com.dtstack.chunjun.connector.jdbc.lookup.JdbcLruTableFunction;
+import com.dtstack.chunjun.connector.jdbc.util.key.KeyUtil;
 import com.dtstack.chunjun.enums.CacheType;
 import com.dtstack.chunjun.lookup.conf.LookupConf;
 import com.dtstack.chunjun.source.DtInputFormatSourceFunction;
@@ -45,6 +46,7 @@
 
 import org.apache.commons.lang3.StringUtils;
 
+import java.math.BigInteger;
 import java.util.ArrayList;
 import java.util.List;
 import java.util.Objects;
@@ -127,34 +129,70 @@ public ScanRuntimeProvider getScanRuntimeProvider(ScanContext runtimeProviderCon
 
         // TODO sql任务使用增量同步或者间隔轮询时暂不支持增量指标写入外部存储，暂时设置为false
         jdbcConf.setInitReporter(false);
-        String increColumn = jdbcConf.getIncreColumn();
-        if (StringUtils.isNotBlank(increColumn)) {
-            FieldConf fieldConf =
-                    FieldConf.getSameNameMetaColumn(jdbcConf.getColumn(), increColumn);
-            if (fieldConf != null) {
-                jdbcConf.setIncreColumnIndex(fieldConf.getIndex());
-                jdbcConf.setIncreColumnType(fieldConf.getType());
 
-                jdbcConf.setRestoreColumn(increColumn);
-                jdbcConf.setRestoreColumnIndex(fieldConf.getIndex());
-                jdbcConf.setRestoreColumnType(fieldConf.getType());
-            } else {
-                throw new IllegalArgumentException("unknown incre column name: " + increColumn);
-            }
-        }
+        KeyUtil<?, BigInteger> restoreKeyUtil = null;
+        KeyUtil<?, BigInteger> splitKeyUtil = null;
+        KeyUtil<?, BigInteger> incrementKeyUtil = null;
 
+        // init restore info
         String restoreColumn = jdbcConf.getRestoreColumn();
         if (StringUtils.isNotBlank(restoreColumn)) {
             FieldConf fieldConf =
                     FieldConf.getSameNameMetaColumn(jdbcConf.getColumn(), restoreColumn);
             if (fieldConf != null) {
                 jdbcConf.setRestoreColumnIndex(fieldConf.getIndex());
                 jdbcConf.setRestoreColumnType(fieldConf.getType());
+                restoreKeyUtil = jdbcDialect.initKeyUtil(fieldConf.getName(), fieldConf.getType());
             } else {
                 throw new IllegalArgumentException("unknown restore column name: " + restoreColumn);
             }
         }
 
+        // init splitInfo
+        String splitPk = jdbcConf.getSplitPk();
+        if (StringUtils.isNotBlank(splitPk)) {
+            FieldConf fieldConf = FieldConf.getSameNameMetaColumn(jdbcConf.getColumn(), splitPk);
+            if (fieldConf != null) {
+                jdbcConf.setSplitPk(fieldConf.getType());
+                splitKeyUtil = jdbcDialect.initKeyUtil(fieldConf.getName(), fieldConf.getType());
+            }
+        }
+
+        // init incrementInfo
+        String incrementColumn = jdbcConf.getIncreColumn();
+        if (StringUtils.isNotBlank(incrementColumn)) {
+            FieldConf fieldConf =
+                    FieldConf.getSameNameMetaColumn(jdbcConf.getColumn(), incrementColumn);
+            int index;
+            String name;
+            String type;
+            if (fieldConf != null) {
+                index = fieldConf.getIndex();
+                name = fieldConf.getName();
+                type = fieldConf.getType();
+                incrementKeyUtil = jdbcDialect.initKeyUtil(name, type);
+            } else {
+                throw new IllegalArgumentException(
+                        "unknown increment column name: " + incrementColumn);
+            }
+            jdbcConf.setIncreColumn(name);
+            jdbcConf.setIncreColumnType(type);
+
+            jdbcConf.setRestoreColumn(name);
+            jdbcConf.setRestoreColumnIndex(index);
+            jdbcConf.setRestoreColumnType(type);
+            restoreKeyUtil = incrementKeyUtil;
+
+            if (StringUtils.isBlank(jdbcConf.getSplitPk())) {
+                splitKeyUtil = incrementKeyUtil;
+                jdbcConf.setSplitPk(name);
+            }
+        }
+
+        builder.setRestoreKeyUtil(restoreKeyUtil);
+        builder.setSplitKeyUtil(splitKeyUtil);
+        builder.setIncrementKeyUtil(incrementKeyUtil);
+
         builder.setJdbcDialect(jdbcDialect);
         builder.setJdbcConf(jdbcConf);
         builder.setRowConverter(jdbcDialect.getRowConverter(rowType));