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

• 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
• Assignee: Nicholas Jiang
• Reporter: jiabao.sun
• Created: 28/Jan/21 04:24
• Updated: 28/May/21 08:16
• Resolved: 02/Feb/21 03:52

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();
    }

Issue Links

• links to: GitHub Pull Request #14792