fix debeziumSourcefunction ddl parse

JNSimba · Sep 27, 2023 · b997ac3 · b997ac3
1 parent 3cf60c5
commit b997ac3
Show file tree

Hide file tree

Showing 2 changed files with 18 additions and 9 deletions.
diff --git a/...nector/src/main/java/org/apache/doris/flink/sink/writer/JsonDebeziumSchemaSerializer.java b/...nector/src/main/java/org/apache/doris/flink/sink/writer/JsonDebeziumSchemaSerializer.java
@@ -49,6 +49,7 @@
 
 import java.io.IOException;
 import java.nio.charset.StandardCharsets;
+import java.util.ArrayList;
 import java.util.HashMap;
 import java.util.LinkedHashMap;
 import java.util.List;
@@ -197,7 +198,7 @@ public boolean schemaChangeV2(JsonNode recordRoot) {
                 String ddlSql = ddlSqlList.get(i);
                 boolean doSchemaChange = checkSchemaChange(ddlSchema);
                 status = doSchemaChange && execSchemaChange(ddlSql);
-                LOG.info("schema change status:{}", status);
+                LOG.info("schema change status:{}, ddl:{}", status, ddlSql);
             }
         } catch (Exception ex) {
             LOG.warn("schema change error :", ex);
@@ -220,12 +221,14 @@ private boolean checkSchemaChange(DDLSchema ddlSchema) throws IOException, Illeg
 
     @VisibleForTesting
     public List<String> extractDDLList(JsonNode record) throws JsonProcessingException {
-        JsonNode historyRecord = objectMapper.readTree(extractJsonNode(record, "historyRecord"));
+        JsonNode historyRecord = extractHistoryRecord(record);
         JsonNode tableChanges = historyRecord.get("tableChanges");
-        JsonNode tableChange = tableChanges.get(0);
         String ddl = extractJsonNode(historyRecord, "ddl");
+        if(Objects.isNull(tableChanges) || Objects.isNull(ddl)){
+            return new ArrayList<>();
+        }
         LOG.debug("received debezium ddl :{}", ddl);
-
+        JsonNode tableChange = tableChanges.get(0);
         Matcher matcher = addDropDDLPattern.matcher(ddl);
         if (Objects.isNull(tableChange)|| !tableChange.get("type").asText().equals("ALTER") || !matcher.find()) {
             return null;
@@ -388,12 +391,17 @@ private Map<String, Object> extractRow(JsonNode recordRow) {
         return recordMap != null ? recordMap : new HashMap<>();
     }
 
-    public String extractDDL(JsonNode record) throws JsonProcessingException {
-        String historyRecord = extractJsonNode(record, "historyRecord");
-        if (Objects.isNull(historyRecord)) {
-            return null;
+    private JsonNode extractHistoryRecord(JsonNode record) throws JsonProcessingException {
+        if(record.has("historyRecord")){
+            return objectMapper.readTree(record.get("historyRecord").asText());
         }
-        String ddl = extractJsonNode(objectMapper.readTree(historyRecord), "ddl");
+        //The ddl passed by some scenes will not be included in the historyRecord, such as DebeziumSourceFunction
+        return record;
+    }
+
+    public String extractDDL(JsonNode record) throws JsonProcessingException {
+        JsonNode historyRecord = extractHistoryRecord(record);
+        String ddl = extractJsonNode(historyRecord, "ddl");
         LOG.debug("received debezium ddl :{}", ddl);
         if (!Objects.isNull(ddl)) {
             //filter add/drop operation

diff --git a/...s-connector/src/main/java/org/apache/doris/flink/tools/cdc/oracle/OracleDatabaseSync.java b/...s-connector/src/main/java/org/apache/doris/flink/tools/cdc/oracle/OracleDatabaseSync.java
@@ -162,6 +162,7 @@ public DataStreamSource<String> buildCdcSource(StreamExecutionEnvironment env) {
                     .tableList(schemaName + "." + tableName)
                     .username(username)
                     .password(password)
+                    .includeSchemaChanges(true)
                     .startupOptions(startupOptions)
                     .deserializer(schema)
                     .debeziumProperties(debeziumProperties)