[FLINK-18758] Many data types do not work in debezium format

• Type: Sub-task
• Status: Open
• Priority: Major
• Resolution: Unresolved
• Component/s: Formats (JSON, Avro, Parquet, ORC, SequenceFile)
• Labels: auto-unassigned
• Assignee: Unassigned
• Reporter: Leonard Xu
• Created: 30/Jul/20 04:33
• Updated: 23/Feb/23 11:58

Description

Currently debezium json format  wrapper the json format to serialize/deserialize the debezuim json format data.

But debezium json format has its own data type which consists of Literal Type and Semantic Type[1], i.g. Date type in debezium is an integer which represents the number of days since epoch rather than a string with 'yyyy-MM-dd' pattern.  

{   "schema":{
      "fields":[  
       {            
                "fields":[
               {                 
                 "type":"int32",      //Literal Type
                  "optional":false,
                  "name":"io.debezium.time.Date", //semantic Type
                  "version":1,
                  "field":"order_date"
               },
               {                  "type":"int32",
                  "optional":false,
                  "field":"quantity"
               }
            ]
         }
      ]
   },
   "payload":{ 
      "before":null,
      "after":{ 
         "order_date":16852, // Literal Value, the number of days since epoch
        "quantity":1
      },
      "op":"c",
      "ts_ms":1596081813474
   }
} 

 

I think we need obtain the debezuim data type from schema information and then serialize/deserialize the data in payload.

[1]https://debezium.io/documentation/reference/1.2/connectors/mysql.html