​
Mapping Kafka Connnect Data Types and SQL Types

At the moment we need to use Kafka Connect types to define the schema of the data flowing through Streamkap. Please see the example of a JSON payload: 
{
  "id" : 1,
  "customer_id" : 1234,
  "customer_name" : "John Doe"
}
and its schema definition 
{
  "type" : "struct",
  "fields" : [ {
    "type" : "int64",
    "optional" : false,
    "field" : "id"
  }, {
    "type" : "int64",
    "optional" : false,
    "field" : "customer_id"
  }, {
    "type" : "string",
    "optional" : false,
    "field" : "customer_name"
  } ],
  "optional" : false
}
Instead of using SQL types like INT, BIGINT, VARCHAR, etc… we use Kafka Connect types like int64, string, etc… The following table shows the mapping between Kafka Connect types and SQL types for some common databases:
Kafka Connect TypeSQL Type (PostgreSQL)SQL Type (MySQL)SQL Type (Oracle)
int8SMALLINTTINYINTNUMBER(3)
int16INTEGERSMALLINTNUMBER(5)
int32INTEGERINTNUMBER(10)
int64BIGINTBIGINTNUMBER(19)
floatREALFLOATNUMBER(7, 2)
doubleDOUBLE PRECISIONDOUBLENUMBER(15, 6)
stringVARCHARVARCHARVARCHAR2
bytesBYTEABLOBBLOB
booleanBOOLEANBOOLEANNUMBER(1)
{"type":"int64","optional":false,"name":"org.apache.kafka.connect.data.Timestamp"}TIMESTAMPTIMESTAMPTIMESTAMP
{"type":"bytes","optional":false,"name":"org.apache.kafka.connect.data.Decimal","parameters":{"scale":"4","connect.decimal.precision":"38"}}DECIMAL(38,4)DECIMAL(38,4)DECIMAL(38,4)
{"type":"int32","optional":false,"name":"org.apache.kafka.connect.data.Date"}DATEDATEDATE
{"type":"int64","optional":false,"name":"org.apache.kafka.connect.data.Time"}TIMETIMEN/A
DECIMAL(38,4) means a decimal number with up to 38 digits in total, of which 4 digits can be after the decimal point, one can adjust the precision if needed. Please see the example of primitive Kafka Connect Schema definitions and their corresponding payloads: 
{"schema":{"type":"boolean","optional":false,"default":false},"payload":false}
{"schema":{"type":"int8","optional":false,"default":1},"payload":1}
{"schema":{"type":"int16","optional":false,"default":1},"payload":1}
{"schema":{"type":"int32","optional":false,"default":1},"payload":1}
{"schema":{"type":"int64","optional":false,"default":1},"payload":1}
{"schema":{"type":"float","optional":false,"default":1.0},"payload":1.0}
{"schema":{"type":"double","optional":false,"default":1.0},"payload":1.0}
{"schema":{"type":"bytes","optional":false,"default":"Zm9v"},"payload":"Zm9v"}
{"schema":{"type":"string","optional":false,"default":"foo"},"payload":"foo"}
{"schema":{"type":"bytes","optional":false,"name":"org.apache.kafka.connect.data.Decimal","parameters":{"scale":"4","connect.decimal.precision":"38"},"default":"EtaH"},"payload":"EtaH"}
{"schema":{"type":"int64","optional":false,"name":"org.apache.kafka.connect.data.Timestamp","default":1758635042028},"payload":1758635042028}
{"schema":{"type":"int64","optional":false,"name":"org.apache.kafka.connect.data.Time","default":55800000},"payload":55800000}
{"schema":{"type":"int32","optional":false,"name":"org.apache.kafka.connect.data.Date","default":20370},"payload":20370}