[Feature][Format] Improve maxwell_json,canal_json,debezium_json format add ts_ms and table #9701

dyp12 · 2025-08-14T07:16:04Z

Purpose of this pull request

close #9675

Does this PR introduce any user-facing change?

How was this patch tested?

Check list

If any new Jar binary package adding in your PR, please add License Notice according
New License Guide
If necessary, please update the documentation to describe the new feature. https://github.com/apache/seatunnel/tree/dev/docs
If you are contributing the connector code, please check that the following files are updated:
1. Update plugin-mapping.properties and add new connector information in it
2. Update the pom file of seatunnel-dist
3. Add ci label in label-scope-conf
4. Add e2e testcase in seatunnel-e2e
5. Update connector plugin_config

Hisoka-X · 2025-08-15T02:57:36Z

cc @wubx

...onnector-kafka-e2e/src/test/java/org/apache/seatunnel/e2e/connector/kafka/KafkaFormatIT.java

seatunnel-api/src/main/java/org/apache/seatunnel/api/table/type/SeaTunnelRow.java

Hisoka-X · 2025-08-15T03:19:17Z

...-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"},


The cancel format is database and table fields, not tableId. Please refer https://docs.pingcap.com/zh/tidb/stable/ticdc-canal-json/#dml-event

The cancel format is database and table fields, not tableId. Please refer https://docs.pingcap.com/zh/tidb/stable/ticdc-canal-json/#dml-event

SeaTunnelRow do not have database and table fields， if use tableId to get db fields，may be get error data，remove it，only add ts field？

We should update SeaTunnelRowType to CatalogTable in

seatunnel/seatunnel-formats/seatunnel-format-json/src/main/java/org/apache/seatunnel/format/json/canal/CanalJsonSerializationSchema.java

Line 49 in 9212a77

public CanalJsonSerializationSchema(SeaTunnelRowType rowType) {

Then we can get table and database from CatalogTable.

We should update SeaTunnelRowType to CatalogTable in

seatunnel/seatunnel-formats/seatunnel-format-json/src/main/java/org/apache/seatunnel/format/json/canal/CanalJsonSerializationSchema.java

Line 49 in 9212a77

public CanalJsonSerializationSchema(SeaTunnelRowType rowType) {

Then we can get table and database from CatalogTable.

MetadataUtil.getDatabase use tableId to get db,maybe can use it , update SeaTunnelRowType to CatalogTable will change manay file

Hisoka-X · 2025-08-15T03:21:41Z

...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"},


The debezium format doesn't contains tableId. Please refer https://help.aliyun.com/zh/flink/debezium

Hisoka-X · 2025-08-15T03:23:04Z

...n/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"},


same as canal, maxwell doesn'ts contains tableId too. Please refer https://help.aliyun.com/zh/flink/maxwell-format

Hisoka-X · 2025-08-15T03:26:00Z

...rmat-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"},


ogg json doesn't contains tableId. Please refer https://help.aliyun.com/zh/datahub/developer-reference/ogg-for-big-data-kafka

Hisoka-X · 2025-08-19T07:09:39Z

...l-format-json/src/test/java/org/apache/seatunnel/format/json/ogg/OggJsonSerDeSchemaTest.java

-                        "{\"data\":{\"id\":111,\"name\":\"scooter\",\"description\":\"Big 2-wheel scooter \",\"weight\":5.17},\"type\":\"INSERT\"}",
-                        "{\"data\":{\"id\":111,\"name\":\"scooter\",\"description\":\"Big 2-wheel scooter \",\"weight\":5.17},\"type\":\"DELETE\"}");
-
+                        "{\"data\":{\"id\":101,\"name\":\"scooter\",\"description\":\"Small 2-wheel scooter\",\"weight\":3.14},\"type\":\"INSERT\",\"table\":\"..test\",\"op_ts\":1589384406000}",


why table is ..test?

why table is ..test?

'..test' is because of db and schema is null

Hisoka-X · 2025-08-19T07:33:09Z

...-json/src/main/java/org/apache/seatunnel/format/json/canal/CanalJsonSerializationSchema.java

+            }
+
+            if (row.getOptions() != null && row.getOptions().containsKey(EVENT_TIME.getName())) {
+                reuse.setField(4, row.getOptions().get(EVENT_TIME.getName()));


Should we standardize the precision of eventtime? I see that the unit is sometimes seconds and sometimes milliseconds. Can we standardize it to milliseconds?

Should we standardize the precision of eventtime? I see that the unit is sometimes seconds and sometimes milliseconds. Can we standardize it to milliseconds?

now all use milliseconds，when read data ,all eventtime is milliseconds

Could you add some comment in

seatunnel/seatunnel-api/src/main/java/org/apache/seatunnel/api/table/type/CommonOptions.java

Line 53 in 6e5737c

EVENT_TIME("EventTime", true),

Could you add some comment in

seatunnel/seatunnel-api/src/main/java/org/apache/seatunnel/api/table/type/CommonOptions.java

Line 53 in 6e5737c

EVENT_TIME("EventTime", true),

i add it , and change metadata.md

...son/src/main/java/org/apache/seatunnel/format/json/canal/CanalJsonDeserializationSchema.java

Hisoka-X

Thanks @dyp12 . LGTM if ci passes.

00193254 added 4 commits August 14, 2025 15:02

https://github.com/apache/seatunnel/issues/9675

a12ab18

https://github.com/apache/seatunnel/issues/9675

222c03d

https://github.com/apache/seatunnel/issues/9675

82e7572

https://github.com/apache/seatunnel/issues/9675

000eacd

github-actions bot added connectors-v2 e2e format file api labels Aug 14, 2025

Hisoka-X reviewed Aug 15, 2025

View reviewed changes

...onnector-kafka-e2e/src/test/java/org/apache/seatunnel/e2e/connector/kafka/KafkaFormatIT.java Outdated Show resolved Hide resolved

Hisoka-X reviewed Aug 15, 2025

View reviewed changes

00193254 added 6 commits August 15, 2025 15:47

https://github.com/apache/seatunnel/issues/9675

18e5ed1

https://github.com/apache/seatunnel/issues/9675

08679e0

https://github.com/apache/seatunnel/issues/9675

3834772

https://github.com/apache/seatunnel/issues/9675

db4ffe8

https://github.com/apache/seatunnel/issues/9675

6dbe2e6

https://github.com/apache/seatunnel/issues/9675

0eb1066

dyp12 requested a review from Hisoka-X August 18, 2025 06:39

Hisoka-X reviewed Aug 19, 2025

View reviewed changes

...son/src/main/java/org/apache/seatunnel/format/json/canal/CanalJsonDeserializationSchema.java Show resolved Hide resolved

https://github.com/apache/seatunnel/issues/9675

21f78c8

github-actions bot added the document label Aug 19, 2025

Hisoka-X changed the title ~~[Feature][json-format ] Improve maxwell_json,canal_json,debezium_json format add ts_ms and tableId~~ [Feature][Format] Improve maxwell_json,canal_json,debezium_json format add ts_ms and table Aug 19, 2025

Hisoka-X approved these changes Aug 19, 2025

View reviewed changes

github-actions bot added approved reviewed labels Aug 19, 2025

corgy-w approved these changes Aug 20, 2025

View reviewed changes

corgy-w merged commit fb8444b into apache:dev Aug 20, 2025
6 checks passed

wubx mentioned this pull request Aug 26, 2025

[Feature][S3-Sink] Improve maxwell_json,canal_json,debezium_json format #9773

Open

3 tasks

[Feature][Format] Improve maxwell_json,canal_json,debezium_json format add ts_ms and table #9701

[Feature][Format] Improve maxwell_json,canal_json,debezium_json format add ts_ms and table #9701

Uh oh!

Conversation

dyp12 commented Aug 14, 2025 • edited by Hisoka-X Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Purpose of this pull request

Does this PR introduce any user-facing change?

How was this patch tested?

Check list

Uh oh!

Hisoka-X commented Aug 15, 2025

Uh oh!

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Hisoka-X left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

dyp12 commented Aug 14, 2025 •

edited by Hisoka-X

Loading