[FLINK-21172] canal-json format include es field - ASF JIRA

Agile Board

Rank to Top

Rank to Bottom

Attach files

Attach Screenshot

Bulk Copy Attachments

Bulk Move Attachments

Voters

Watch issue

Watchers

Create sub-task

Convert to sub-task

Link

Clone

Labels

Update Comment Author

Replace String in Comment

Update Comment Visibility

Delete Comments

XML

Word

Printable

JSON

Details

Type: Improvement
Status: Closed
Priority: Minor
Resolution: Fixed
Affects Version/s: 1.12.0, 1.12.1
Fix Version/s: 1.13.0
Component/s: Formats (JSON, Avro, Parquet, ORC, SequenceFile), Table SQL / Ecosystem
Labels:
- pull-request-available

External issue URL:
https://github.com/alibaba/canal/issues/2170

Description

Canal flat message json format has an 'es' field extracted from mysql binlog which means the row data real change time in mysql. It expressed the event time naturally but was ignored during deserialization.

{
  "data": [
    {
      "id": "111",
      "name": "scooter",
      "description": "Big 2-wheel scooter",
      "weight": "5.18"
    }
  ],
  "database": "inventory",
  "es": 1589373560000,
  "id": 9,
  "isDdl": false,
  "mysqlType": {
    "id": "INTEGER",
    "name": "VARCHAR(255)",
    "description": "VARCHAR(512)",
    "weight": "FLOAT"
  },
  "old": [
    {
      "weight": "5.15"
    }
  ],
  "pkNames": [
    "id"
  ],
  "sql": "",
  "sqlType": {
    "id": 4,
    "name": 12,
    "description": 12,
    "weight": 7
  },
  "table": "products",
  "ts": 1589373560798,
  "type": "UPDATE"
}

org.apache.flink.formats.json.canal. CanalJsonDeserializationSchema

    private static RowType createJsonRowType(DataType databaseSchema) {
        // Canal JSON contains other information, e.g. "ts", "sql", but we don't need them
        return (RowType)
                DataTypes.ROW(
                                DataTypes.FIELD("data", DataTypes.ARRAY(databaseSchema)),
                                DataTypes.FIELD("old", DataTypes.ARRAY(databaseSchema)),
                                DataTypes.FIELD("type", DataTypes.STRING()),
                                DataTypes.FIELD("database", DataTypes.STRING()),
                                DataTypes.FIELD("table", DataTypes.STRING()))
                        .getLogicalType();
    }