[hotfix-1581][jdbc-base] fix the problem that polling sometimes may duplicate data

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

@@ -460,11 +460,9 @@ protected String buildQuerySql(InputSplit inputSplit) {
 
         querySql = buildQuerySqlBySplit(jdbcInputSplit, whereList);
 
-        if (!jdbcConfig.isPolling()) {
-            querySql = querySql + SqlUtil.buildOrderSql(jdbcConfig, jdbcDialect, "ASC");
-        }
-        log.info("Executing sql is: '{}'", querySql);
-        return querySql;
+        String finalSql = SqlUtil.buildOrderSql(querySql, jdbcConfig, jdbcDialect, "ASC");
+        log.info("Executing sql is: '{}'", finalSql);
+        return finalSql;
     }
 
     /**
@@ -691,17 +689,19 @@ protected void executeQuery(String startLocation) throws SQLException {
                     builder.append(" WHERE ");
                 }
                 builder.append(jdbcDialect.quoteIdentifier(jdbcConfig.getIncreColumn()))
-                        .append(" > ?")
-                        .append(SqlUtil.buildOrderSql(jdbcConfig, jdbcDialect, "ASC"));
-                jdbcConfig.setQuerySql(builder.toString());
+                        .append(" > ?");
+
+                String querySQL =
+                        SqlUtil.buildOrderSql(builder.toString(), jdbcConfig, jdbcDialect, "ASC");
+                jdbcConfig.setQuerySql(querySQL);
                 initPrepareStatement(jdbcConfig.getQuerySql());
                 log.info("update querySql, sql = {}", jdbcConfig.getQuerySql());
             } else {
                 // if the job have startLocation
                 // sql will be like "select ... from ... where increColumn > ?"
                 jdbcConfig.setQuerySql(
-                        jdbcConfig.getQuerySql()
-                                + SqlUtil.buildOrderSql(jdbcConfig, jdbcDialect, "ASC"));
+                        SqlUtil.buildOrderSql(
+                                jdbcConfig.getQuerySql(), jdbcConfig, jdbcDialect, "ASC"));
                 initPrepareStatement(jdbcConfig.getQuerySql());
                 queryForPolling(startLocation);
                 state = restoreKeyUtil.transLocationStrToSqlValue(startLocation);
@@ -736,7 +736,7 @@ public void initPrepareStatement(String querySql) throws SQLException {
     protected void queryPollingWithOutStartLocation() throws SQLException {
         // add order by to query SQL avoid duplicate data
         initPrepareStatement(
-                jdbcConfig.getQuerySql() + SqlUtil.buildOrderSql(jdbcConfig, jdbcDialect, "ASC"));
+                SqlUtil.buildOrderSql(jdbcConfig.getQuerySql(), jdbcConfig, jdbcDialect, "ASC"));
         resultSet = ps.executeQuery();
         hasNext = resultSet.next();
 

chunjun-connectors/chunjun-connector-jdbc-base/src/main/java/com/dtstack/chunjun/connector/jdbc/util/SqlUtil.java

@@ -151,24 +151,22 @@ public static String buildQuerySqlBySplit(
         return querySql;
     }
 
-    /**
-     * build order sql
-     *
-     * @param sortRule
-     * @return
-     */
     public static String buildOrderSql(
-            JdbcConfig jdbcConfig, JdbcDialect jdbcDialect, String sortRule) {
+            String originalSql, JdbcConfig jdbcConf, JdbcDialect jdbcDialect, String sortRule) {
         String column;
         // 增量任务
-        if (jdbcConfig.isIncrement()) {
-            column = jdbcConfig.getIncreColumn();
+        if (jdbcConf.isIncrement() && jdbcConf.isOrderBy() && !originalSql.contains("ORDER BY")) {
+            column = jdbcConf.getIncreColumn();
         } else {
-            column = jdbcConfig.getOrderByColumn();
+            column = jdbcConf.getOrderByColumn();
         }
-        return StringUtils.isBlank(column)
-                ? ""
-                : String.format(" ORDER BY %s %s", jdbcDialect.quoteIdentifier(column), sortRule);
+
+        String additional =
+                StringUtils.isBlank(column)
+                        ? ""
+                        : String.format(
+                                " ORDER BY %s %s", jdbcDialect.quoteIdentifier(column), sortRule);
+        return originalSql + additional;
     }
 
     /* 是否添加自定义函数column 作为分片key ***/

chunjun-connectors/chunjun-connector-jdbc-base/src/test/java/com/dtstack/chunjun/connector/jdbc/source/JdbcInputFormatTest.java

@@ -353,7 +353,8 @@ public void buildQuerySqlTest() {
                 .thenCallRealMethod();
         when(SqlUtil.buildQuerySqlBySplit(any(), any(), anyList(), anyList(), any()))
                 .thenAnswer(invocation -> "select id from table where id > 10");
-        when(SqlUtil.buildOrderSql(jdbcConfig, jdbcDialect, "ASC"))
+        when(SqlUtil.buildOrderSql(
+                        jdbcInputFormat.buildQuerySql(inputSplit), jdbcConfig, jdbcDialect, "ASC"))
                 .thenAnswer(invocation -> " order by id ASC");
         String except = "select id from table where id > 10 order by id ASC";
         Assert.assertEquals(except, jdbcInputFormat.buildQuerySql(inputSplit));

chunjun-connectors/chunjun-connector-postgresql/src/main/java/com/dtstack/chunjun/connector/postgresql/source/PostgresqlInputFormat.java

@@ -39,7 +39,7 @@ protected void queryPollingWithOutStartLocation() throws SQLException {
         // , the query will report an error after the method
         // #setFetchDirection(ResultSet.FETCH_REVERSE) is called.
         String querySql =
-                jdbcConfig.getQuerySql() + SqlUtil.buildOrderSql(jdbcConfig, jdbcDialect, "ASC");
+                SqlUtil.buildOrderSql(jdbcConfig.getQuerySql(), jdbcConfig, jdbcDialect, "ASC");
         ps =
                 dbConn.prepareStatement(
                         querySql, ResultSet.TYPE_SCROLL_INSENSITIVE, resultSetConcurrency);