Merge branch 'hotfix_1.10_4.0.x_30263' into 1.10_release_4.0.x

Author: gituser

…ink/sql/sink/ocean/OceanbaseDialect.java …sql/sink/oceanbase/OceanbaseDialect.javaoceanbase/oceanbase-sink/src/main/java/com/dtstack/flink/sql/sink/ocean/OceanbaseDialect.java renamed to oceanbase/oceanbase-sink/src/main/java/com/dtstack/flink/sql/sink/oceanbase/OceanbaseDialect.java oceanbase/oceanbase-sink/src/main/java/com/dtstack/flink/sql/sink/ocean/OceanbaseDialect.java renamed to oceanbase/oceanbase-sink/src/main/java/com/dtstack/flink/sql/sink/oceanbase/OceanbaseDialect.java

@@ -15,7 +15,7 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
-package com.dtstack.flink.sql.sink.ocean;
+package com.dtstack.flink.sql.sink.oceanbase;
 
 import com.dtstack.flink.sql.sink.rdb.dialect.JDBCDialect;
 
@@ -50,27 +50,30 @@ public Optional<String> getUpsertStatement(String schema,
                                                String[] uniqueKeyFields,
                                                boolean allReplace) {
         return allReplace ?
-                buildReplaceIntoStatement(tableName, fieldNames) :
-                buildDuplicateUpsertStatement(tableName, fieldNames);
+                buildReplaceIntoStatement(tableName, fieldNames) : buildDuplicateUpsertStatement(tableName, fieldNames);
     }
 
-    private Optional<String> buildDuplicateUpsertStatement(String tableName, String[] fieldsName) {
-        String updateClause = Arrays.stream(fieldsName).map(f -> quoteIdentifier(f)
-                + "IFNULL(VALUES(" + quoteIdentifier(f) + ")," + quoteIdentifier(f) + ")")
-                .collect(Collectors.joining(","));
-        return Optional.of(getInsertIntoStatement("", tableName, fieldsName, null) +
+    public Optional<String> buildDuplicateUpsertStatement(String tableName, String[] fieldNames) {
+        String updateClause = Arrays
+                .stream(fieldNames)
+                .map(f -> quoteIdentifier(f) + "=IFNULL(VALUES(" + quoteIdentifier(f) + ")," + quoteIdentifier(f) + ")")
+                .collect(Collectors.joining(", "));
+
+        return Optional.of(getInsertIntoStatement("", tableName, fieldNames, null) +
                 " ON DUPLICATE KEY UPDATE " + updateClause
         );
     }
 
-    private Optional<String> buildReplaceIntoStatement(String tableName, String[] fieldsNames) {
-        String columns = Arrays.stream(fieldsNames)
+    public Optional<String> buildReplaceIntoStatement(String tableName, String[] fieldNames) {
+        String columns = Arrays
+                .stream(fieldNames)
                 .map(this::quoteIdentifier)
-                .collect(Collectors.joining(","));
-        String placeholders = Arrays.stream(fieldsNames)
+                .collect(Collectors.joining(", "));
+        String placeholders = Arrays
+                .stream(fieldNames)
                 .map(f -> "?")
-                .collect(Collectors.joining(","));
+                .collect(Collectors.joining(", "));
         return Optional.of("REPLACE INTO " + quoteIdentifier(tableName) +
-                "(" + columns + ") VALUES (" + placeholders + ")");
+                "(" + columns + ")" + " VALUES (" + placeholders + ")");
     }
 }

…/flink/sql/sink/ocean/OceanbaseSink.java …nk/sql/sink/oceanbase/OceanbaseSink.javaoceanbase/oceanbase-sink/src/main/java/com/dtstack/flink/sql/sink/ocean/OceanbaseSink.java renamed to oceanbase/oceanbase-sink/src/main/java/com/dtstack/flink/sql/sink/oceanbase/OceanbaseSink.java oceanbase/oceanbase-sink/src/main/java/com/dtstack/flink/sql/sink/ocean/OceanbaseSink.java renamed to oceanbase/oceanbase-sink/src/main/java/com/dtstack/flink/sql/sink/oceanbase/OceanbaseSink.java

@@ -15,7 +15,7 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
-package com.dtstack.flink.sql.sink.ocean;
+package com.dtstack.flink.sql.sink.oceanbase;
 
 import com.dtstack.flink.sql.sink.IStreamSinkGener;
 import com.dtstack.flink.sql.sink.rdb.AbstractRdbSink;

…ink/ocean/table/OceanbaseSinkParser.java …oceanbase/table/OceanbaseSinkParser.javaoceanbase/oceanbase-sink/src/main/java/com/dtstack/flink/sql/sink/ocean/table/OceanbaseSinkParser.java renamed to oceanbase/oceanbase-sink/src/main/java/com/dtstack/flink/sql/sink/oceanbase/table/OceanbaseSinkParser.java oceanbase/oceanbase-sink/src/main/java/com/dtstack/flink/sql/sink/ocean/table/OceanbaseSinkParser.java renamed to oceanbase/oceanbase-sink/src/main/java/com/dtstack/flink/sql/sink/oceanbase/table/OceanbaseSinkParser.java

@@ -15,7 +15,7 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
-package com.dtstack.flink.sql.sink.ocean.table;
+package com.dtstack.flink.sql.sink.oceanbase.table;
 
 import com.dtstack.flink.sql.sink.rdb.table.RdbSinkParser;
 import com.dtstack.flink.sql.table.AbstractTableInfo;