[Feature][jdbc splitpk] support manually set the Min/Max value (#1541)

Author: kinoxyz1

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

@@ -56,6 +56,8 @@ public class JdbcConf extends ChunJunCommonConf implements Serializable {
     protected String orderByColumn;
     protected String querySql;
     protected String splitPk;
+    protected String splitPkStart;
+    protected String splitPkEnd;
     protected String splitStrategy;
     protected int fetchSize = 0;
     protected int queryTimeOut = 0;
@@ -142,6 +144,22 @@ public String getTable() {
         return connection.get(0).getTable().get(0);
     }
 
+    public String getSplitPkStart() {
+        return splitPkStart;
+    }
+
+    public void setSplitPkStart(String splitPkStart) {
+        this.splitPkStart = splitPkStart;
+    }
+
+    public String getSplitPkEnd() {
+        return splitPkEnd;
+    }
+
+    public void setSplitPkEnd(String splitPkEnd) {
+        this.splitPkEnd = splitPkEnd;
+    }
+
     public void setTable(String table) {
         connection.get(0).getTable().set(0, table);
     }
@@ -485,6 +503,12 @@ public String toString() {
                 + ", splitPk='"
                 + splitPk
                 + '\''
+                + ", splitPkStart='"
+                + splitPkStart
+                + '\''
+                + ", splitPkEnd='"
+                + splitPkEnd
+                + '\''
                 + ", splitStrategy='"
                 + splitStrategy
                 + '\''

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

@@ -394,25 +394,32 @@ private Pair<String, String> getSplitRangeFromDb() {
         try {
             long startTime = System.currentTimeMillis();
 
-            String querySplitRangeSql = SqlUtil.buildQuerySplitRangeSql(jdbcConf, jdbcDialect);
-            LOG.info(String.format("Query SplitRange sql is '%s'", querySplitRangeSql));
-
-            conn = getConnection();
-            st = conn.createStatement(resultSetType, resultSetConcurrency);
-            st.setQueryTimeout(jdbcConf.getQueryTimeOut());
-            rs = st.executeQuery(querySplitRangeSql);
-            if (rs.next()) {
+            if (jdbcConf.getSplitPkStart() != null && jdbcConf.getSplitPkEnd() != null) {
                 splitPkRange =
                         Pair.of(
-                                String.valueOf(rs.getObject("min_value")),
-                                String.valueOf(rs.getObject("max_value")));
-            }
+                                String.valueOf(jdbcConf.getSplitPkStart()),
+                                String.valueOf(jdbcConf.getSplitPkEnd()));
 
-            LOG.info(
-                    String.format(
-                            "Takes [%s] milliseconds to get the SplitRange value [%s]",
-                            System.currentTimeMillis() - startTime, splitPkRange));
+            } else {
+                String querySplitRangeSql = SqlUtil.buildQuerySplitRangeSql(jdbcConf, jdbcDialect);
+                LOG.info(String.format("Query SplitRange sql is '%s'", querySplitRangeSql));
+
+                conn = getConnection();
+                st = conn.createStatement(resultSetType, resultSetConcurrency);
+                st.setQueryTimeout(jdbcConf.getQueryTimeOut());
+                rs = st.executeQuery(querySplitRangeSql);
+                if (rs.next()) {
+                    splitPkRange =
+                            Pair.of(
+                                    String.valueOf(rs.getObject("min_value")),
+                                    String.valueOf(rs.getObject("max_value")));
+                }
 
+                LOG.info(
+                        String.format(
+                                "Takes [%s] milliseconds to get the SplitRange value [%s]",
+                                System.currentTimeMillis() - startTime, splitPkRange));
+            }
             return splitPkRange;
         } catch (Throwable e) {
             throw new ChunJunRuntimeException(
@@ -806,7 +813,8 @@ protected Connection getConnection() throws SQLException {
     /** 使用自定义的指标输出器把增量指标打到普罗米修斯 */
     @Override
     protected boolean useCustomReporter() {
-        return jdbcConf.isIncrement() && jdbcConf.getInitReporter();
+        // 配置了 reporter 就可以输入指标到外部系统, 如果不是增量, 增量指标也不会被输出
+        return jdbcConf.getInitReporter();
     }
 
     /** 为了保证增量数据的准确性，指标输出失败时使任务失败 */

docs_zh/ChunJun连接器/oracle/oracle-source.md

@@ -110,6 +110,32 @@ Oracle 9 及以上
     - 默认值：无
       <br />
 
+- **splitPkStart**
+
+    - 描述: 当指定了 `splitPk` 之后，可以手动指定 `splitPk` 的上/下界，之后可以直接根据这两个值计算每个并行度的数据量。
+    - 注意：
+        - 此参数生效的前提是设置了 `splitPk`，并且需要同时设置 **splitPkEnd** 参数。
+            - 推荐splitPk使用表主键，因为表主键通常情况下比较均匀，因此切分出来的分片也不容易出现数据热点。
+        - 目前splitPk仅支持整形数据切分，不支持浮点、字符串、日期等其他类型。如果用户指定其他非支持类型，ChunJun将报错。
+        - 如果channel大于1但是没有配置此参数，任务将置为失败。
+    - 必选：否
+    - 参数类型：String
+    - 默认值：无
+      <br />
+
+- **splitPkEnd**
+
+    - 描述: 当指定了 `splitPk` 之后，可以手动指定 `splitPk` 的上/下界，之后可以直接根据这两个值计算每个并行度的数据量。
+    - 注意：
+        - 此参数生效的前提是设置了 `splitPk`，并且需要同时设置 **splitPkStart** 参数。
+            - 推荐splitPk使用表主键，因为表主键通常情况下比较均匀，因此切分出来的分片也不容易出现数据热点。
+        - 目前splitPk仅支持整形数据切分，不支持浮点、字符串、日期等其他类型。如果用户指定其他非支持类型，ChunJun将报错。
+        - 如果channel大于1但是没有配置此参数，任务将置为失败。
+    - 必选：否
+    - 参数类型：String
+    - 默认值：无
+      <br />
+
 - **queryTimeOut**
 
     - 描述：查询超时时间，单位秒。