DongLiang-0 commented on code in PR #422:
URL:
https://github.com/apache/doris-flink-connector/pull/422#discussion_r1679060997
##########
flink-doris-connector/src/main/java/org/apache/doris/flink/sink/writer/serializer/JsonDebeziumSchemaSerializer.java:
##########
@@ -144,13 +149,28 @@ private void init() {
ignoreUpdateBefore,
targetTablePrefix,
targetTableSuffix);
- this.schemaChange =
- newSchemaChange
- ? new JsonDebeziumSchemaChangeImplV2(changeContext)
- : new JsonDebeziumSchemaChangeImpl(changeContext);
+ initSchemaChangeInstance(changeContext);
this.dataChange = new JsonDebeziumDataChange(changeContext);
}
+ private void initSchemaChangeInstance(JsonDebeziumChangeContext
changeContext) {
+ if (!newSchemaChange) {
+ LOG.info(
+ "newSchemaChange set to false, instantiation schema change
uses JsonDebeziumSchemaChangeImpl.");
+ this.schemaChange = new
JsonDebeziumSchemaChangeImpl(changeContext);
Review Comment:
sure
##########
flink-doris-connector/src/main/java/org/apache/doris/flink/sink/writer/serializer/jsondebezium/JsonDebeziumSchemaChange.java:
##########
@@ -120,6 +137,57 @@ protected JsonNode extractHistoryRecord(JsonNode record)
throws JsonProcessingEx
return record;
}
+ /** Parse event type. */
+ protected EventType extractEventType(JsonNode record) throws
JsonProcessingException {
+ JsonNode tableChange = extractTableChange(record);
+ if (tableChange == null || tableChange.get("type") == null) {
+ return null;
+ }
+ String type = tableChange.get("type").asText();
+ if (EventType.ALTER.toString().equalsIgnoreCase(type)) {
+ return EventType.ALTER;
+ } else if (EventType.CREATE.toString().equalsIgnoreCase(type)) {
+ return EventType.CREATE;
+ }
+ return null;
+ }
+
+ protected JsonNode extractTableChange(JsonNode record) throws
JsonProcessingException {
+ JsonNode historyRecord = extractHistoryRecord(record);
+ JsonNode tableChanges = historyRecord.get("tableChanges");
+ if (Objects.nonNull(tableChanges)) {
+ return tableChanges.get(0);
+ }
+ LOG.warn("Failed to extract tableChanges. record={}", record);
+ return null;
+ }
+
+ protected boolean executeAlterDDLs(
+ List<String> ddlSqlList,
+ JsonNode recordRoot,
+ Tuple2<String, String> dorisTableTuple,
+ boolean status)
+ throws IOException, IllegalArgumentException {
+ if (CollectionUtils.isEmpty(ddlSqlList)) {
+ LOG.info("The recordRoot cannot extract ddl sql. recordRoot={}",
recordRoot);
+ return false;
+ }
+
+ for (String ddlSql : ddlSqlList) {
+ status = schemaChangeManager.execute(ddlSql, dorisTableTuple.f0);
+ LOG.info("schema change status:{}, ddl: {}", status, ddlSql);
+ }
+ return status;
+ }
+
+ protected void extractSourceConnector(JsonNode record) {
+ if (Objects.isNull(sourceConnector)) {
+ sourceConnector =
Review Comment:
Yes, the program is null when it is started, because the sourceConnector is
extracted from the schema change data.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
