You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@inlong.apache.org by do...@apache.org on 2023/05/07 00:30:49 UTC
[inlong] branch master updated: [INLONG-7957][Sort] Fix Oracle connector output two data with the same UPDATE operation (#7961)
This is an automated email from the ASF dual-hosted git repository.
dockerzhang pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/inlong.git
The following commit(s) were added to refs/heads/master by this push:
new 5ab962c50 [INLONG-7957][Sort] Fix Oracle connector output two data with the same UPDATE operation (#7961)
5ab962c50 is described below
commit 5ab962c5005f4c4955bdcbf3ec9ef7567fda2587
Author: emhui <11...@users.noreply.github.com>
AuthorDate: Sun May 7 08:30:42 2023 +0800
[INLONG-7957][Sort] Fix Oracle connector output two data with the same UPDATE operation (#7961)
---
.../cdc/oracle/table/OracleReadableMetaData.java | 20 +++++++++++++++++++-
1 file changed, 19 insertions(+), 1 deletion(-)
diff --git a/inlong-sort/sort-connectors/oracle-cdc/src/main/java/org/apache/inlong/sort/cdc/oracle/table/OracleReadableMetaData.java b/inlong-sort/sort-connectors/oracle-cdc/src/main/java/org/apache/inlong/sort/cdc/oracle/table/OracleReadableMetaData.java
index 40621163c..515b7b08c 100644
--- a/inlong-sort/sort-connectors/oracle-cdc/src/main/java/org/apache/inlong/sort/cdc/oracle/table/OracleReadableMetaData.java
+++ b/inlong-sort/sort-connectors/oracle-cdc/src/main/java/org/apache/inlong/sort/cdc/oracle/table/OracleReadableMetaData.java
@@ -384,7 +384,7 @@ public enum OracleReadableMetaData {
.sql("").es(opTs).isDdl(false).pkNames(getPkNames(tableSchema))
.oracleType(getOracleType(tableSchema))
.table(tableName).ts(ts)
- .type(getOpType(record)).sqlType(getSqlType(tableSchema)).build();
+ .type(getCanalOpType(data)).sqlType(getSqlType(tableSchema)).build();
try {
ObjectMapper objectMapper = new ObjectMapper();
return StringData.fromString(objectMapper.writeValueAsString(canalJson));
@@ -425,6 +425,24 @@ public enum OracleReadableMetaData {
return opType;
}
+ public static String getCanalOpType(GenericRowData record) {
+ String opType;
+ switch (record.getRowKind()) {
+ case DELETE:
+ case UPDATE_BEFORE:
+ opType = OP_DELETE;
+ break;
+ case INSERT:
+ case UPDATE_AFTER:
+ opType = OP_INSERT;
+ break;
+ default:
+ throw new IllegalStateException("the record only have states in DELETE, "
+ + "UPDATE_BEFORE, INSERT and UPDATE_AFTER");
+ }
+ return opType;
+ }
+
private static List<String> getPkNames(@Nullable TableChanges.TableChange tableSchema) {
if (tableSchema == null) {
return null;