Hisoka-X commented on code in PR #9701:
URL: https://github.com/apache/seatunnel/pull/9701#discussion_r2278143647
##########
seatunnel-formats/seatunnel-format-json/src/main/java/org/apache/seatunnel/format/json/canal/CanalJsonSerializationSchema.java:
##########
@@ -88,9 +98,12 @@ private static SeaTunnelRowType
createJsonRowType(SeaTunnelRowType databaseSchem
// but we don't need them
// and we don't need "old" , because can not support
UPDATE_BEFORE,UPDATE_AFTER
return new SeaTunnelRowType(
- new String[] {"data", "type"},
+ new String[] {"data", "type", "tableId", "ts"},
Review Comment:
The cancel format is database and table fields, not tableId. Please refer
https://docs.pingcap.com/zh/tidb/stable/ticdc-canal-json/#dml-event
##########
seatunnel-api/src/main/java/org/apache/seatunnel/api/table/type/SeaTunnelRow.java:
##########
@@ -109,6 +109,12 @@ public SeaTunnelRow copy(int[] indexMapping) {
return newRow;
}
+ public SeaTunnelRow copyWithOptions(int[] indexMapping) {
+ SeaTunnelRow newRow = copy(indexMapping);
+ newRow.setOptions(this.getOptions());
+ return newRow;
+ }
Review Comment:
I think all copy method should with options. Please update those method
https://github.com/apache/seatunnel/blob/9212a77140bb9a310237b4ef82126d42b5d767e6/seatunnel-api/src/main/java/org/apache/seatunnel/api/table/type/SeaTunnelRow.java#L92-L111
##########
seatunnel-formats/seatunnel-format-json/src/main/java/org/apache/seatunnel/format/json/ogg/OggJsonSerializationSchema.java:
##########
@@ -79,7 +88,7 @@ private static SeaTunnelRowType
createJsonRowType(SeaTunnelRowType databaseSchem
// but we don't need them
// and we don't need "old" , because can not support
UPDATE_BEFORE,UPDATE_AFTER
return new SeaTunnelRowType(
- new String[] {"data", "type"},
- new SeaTunnelDataType[] {databaseSchema, STRING_TYPE});
+ new String[] {"data", "type", "tableId", "op_ts"},
Review Comment:
ogg json doesn't contains `tableId`. Please refer
https://help.aliyun.com/zh/datahub/developer-reference/ogg-for-big-data-kafka
##########
seatunnel-formats/seatunnel-format-json/src/main/java/org/apache/seatunnel/format/json/debezium/DebeziumJsonSerializationSchema.java:
##########
@@ -78,7 +94,9 @@ public byte[] serialize(SeaTunnelRow row) {
private static SeaTunnelRowType createJsonRowType(SeaTunnelRowType
databaseSchema) {
return new SeaTunnelRowType(
- new String[] {"before", "after", "op"},
- new SeaTunnelDataType[] {databaseSchema, databaseSchema,
STRING_TYPE});
+ new String[] {"before", "after", "op", "tableId", "ts_ms"},
Review Comment:
The debezium format doesn't contains `tableId`. Please refer
https://help.aliyun.com/zh/flink/debezium
##########
seatunnel-formats/seatunnel-format-json/src/main/java/org/apache/seatunnel/format/json/maxwell/MaxWellJsonSerializationSchema.java:
##########
@@ -87,7 +96,7 @@ private static SeaTunnelRowType
createJsonRowType(SeaTunnelRowType databaseSchem
// but we don't need them
// and we don't need "old" , because can not support
UPDATE_BEFORE,UPDATE_AFTER
return new SeaTunnelRowType(
- new String[] {"data", "type"},
- new SeaTunnelDataType[] {databaseSchema, STRING_TYPE});
+ new String[] {"data", "type", "tableId", "ts"},
Review Comment:
same as canal, maxwell doesn'ts contains tableId too. Please refer
https://help.aliyun.com/zh/flink/maxwell-format
--
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]
