[mysql,postgres] Change Debezium conversion for timestamp types

integration_tests/mysql/main.go

@@ -289,19 +289,19 @@ const expectedPayloadTemplate = `{
 						"parameters": null
 					},
 					{
-						"type": "string",
+						"type": "int64",
 						"optional": false,
 						"default": null,
 						"field": "c_datetime",
-						"name": "io.debezium.time.Timestamp",
+						"name": "io.debezium.time.MicroTimestamp",
 						"parameters": null
 					},
 					{
 						"type": "string",
 						"optional": false,
 						"default": null,
 						"field": "c_timestamp",
-						"name": "io.debezium.time.Timestamp",
+						"name": "io.debezium.time.ZonedTimestamp",
 						"parameters": null
 					},
 					{
@@ -409,7 +409,7 @@ const expectedPayloadTemplate = `{
 			"c_char": "X",
 			"c_date": 18263,
 			"c_date_0000_00_00": null,
-			"c_datetime": "2001-02-03T04:05:06Z",
+			"c_datetime": 981173106000000,
 			"c_decimal": "EtRQ",
 			"c_double": 45.678,
 			"c_enum": "medium",

integration_tests/postgres/main.go

@@ -516,19 +516,19 @@ const expectedPayloadTemplate = `{
 						"parameters": null
 					},
 					{
-						"type": "string",
+						"type": "int64",
 						"optional": false,
 						"default": null,
 						"field": "c_timestamp_without_timezone",
-						"name": "io.debezium.time.Timestamp",
+						"name": "io.debezium.time.MicroTimestamp",
 						"parameters": null
 					},
 					{
 						"type": "string",
 						"optional": false,
 						"default": null,
 						"field": "c_timestamp_with_timezone",
-						"name": "io.debezium.time.Timestamp",
+						"name": "io.debezium.time.ZonedTimestamp",
 						"parameters": null
 					},
 					{
@@ -696,7 +696,7 @@ const expectedPayloadTemplate = `{
 			"c_time_with_timezone": 38057000,
 			"c_time_without_timezone": 45296000,
 			"c_timestamp_with_timezone": "2001-02-16T13:38:40Z",
-			"c_timestamp_without_timezone": "2001-02-16T20:38:40Z",
+			"c_timestamp_without_timezone": 982355920000000,
 			"c_tsrange": "[\"2010-01-01 14:30:00\",\"2010-01-01 15:30:00\")",
 			"c_tstzrange": "[\"2001-02-16 08:38:40+00\",\"2001-03-20 08:38:40+00\")",
 			"c_uuid": "e7082e96-7190-4cc3-8ab4-bd27f1269f08",

lib/debezium/converters/time.go

@@ -67,18 +67,37 @@ func (DateConverter) Convert(value any) (any, error) {
 	return int32(timeValue.Unix() / (60 * 60 * 24)), nil
 }
 
-type TimestampConverter struct{}
+type MicroTimestampConverter struct{}
 
-func (TimestampConverter) ToField(name string) debezium.Field {
+func (MicroTimestampConverter) ToField(name string) debezium.Field {
+	// Represents the number of microseconds since the epoch, and does not include timezone information.
 	return debezium.Field{
-		FieldName: name,
-		// NOTE: We are returning string here because we want the right layout to be used by our Typing library
+		FieldName:    name,
+		Type:         "int64",
+		DebeziumType: string(debezium.MicroTimestamp),
+	}
+}
+
+func (MicroTimestampConverter) Convert(value any) (any, error) {
+	timeValue, ok := value.(time.Time)
+	if !ok {
+		return nil, fmt.Errorf("expected time.Time got %T with value: %v", value, value)
+	}
+	return timeValue.UnixMicro(), nil
+}
+
+type ZonedTimestampConverter struct{}
+
+func (ZonedTimestampConverter) ToField(name string) debezium.Field {
+	// A string representation of a timestamp with timezone information, where the timezone is GMT.
+	return debezium.Field{
+		FieldName:    name,
 		Type:         "string",
-		DebeziumType: string(debezium.Timestamp),
+		DebeziumType: string(debezium.DateTimeWithTimezone),
 	}
 }
 
-func (TimestampConverter) Convert(value any) (any, error) {
+func (ZonedTimestampConverter) Convert(value any) (any, error) {
 	timeValue, ok := value.(time.Time)
 	if !ok {
 		return nil, fmt.Errorf("expected time.Time got %T with value: %v", value, value)

lib/debezium/converters/time_test.go

@@ -121,8 +121,35 @@ func TestDateConverter_Convert(t *testing.T) {
 	}
 }
 
-func TestTimestampConverter_Convert(t *testing.T) {
-	converter := TimestampConverter{}
+func TestMicroTimestampConverter_Convert(t *testing.T) {
+	converter := MicroTimestampConverter{}
+	{
+		// Invalid type
+		_, err := converter.Convert(1234)
+		assert.ErrorContains(t, err, "expected time.Time got int with value: 1234")
+	}
+	{
+		// Date > 9999
+		value, err := converter.Convert(time.Date(9_9999, 2, 3, 4, 5, 0, 0, time.UTC))
+		assert.NoError(t, err)
+		assert.Equal(t, value, int64(3093499310700000000))
+	}
+	{
+		// Date < 0
+		value, err := converter.Convert(time.Date(-1, 2, 3, 4, 5, 0, 0, time.UTC))
+		assert.NoError(t, err)
+		assert.Equal(t, int64(-62195889300000000), value)
+	}
+	{
+		// time.Time
+		value, err := converter.Convert(time.Date(2001, 2, 3, 4, 5, 0, 0, time.UTC))
+		assert.NoError(t, err)
+		assert.Equal(t, int64(981173100000000), value)
+	}
+}
+
+func TestZonedTimestampConverter_Convert(t *testing.T) {
+	converter := ZonedTimestampConverter{}
 	{
 		// Invalid type
 		_, err := converter.Convert(1234)

lib/postgres/scanner.go

@@ -33,6 +33,7 @@ var supportedPrimaryKeyDataType []schema.DataType = []schema.DataType{
 	schema.Time,
 	schema.Date,
 	schema.Timestamp,
+	schema.TimestampWithTimeZone,
 	schema.Interval,
 	schema.UUID,
 	schema.JSON,
@@ -127,6 +128,7 @@ func (s scanAdapter) ParsePrimaryKeyValue(columnName string, value string) (any,
 		schema.Time,
 		schema.Date,
 		schema.Timestamp,
+		schema.TimestampWithTimeZone,
 		schema.Interval,
 		schema.UUID,
 		schema.JSON:

lib/postgres/scanner_test.go

@@ -96,7 +96,7 @@ func TestScanAdapter_ParsePrimaryKeyValue(t *testing.T) {
 			name:        "unsupported data type",
 			dataType:    schema.Array,
 			value:       "1234",
-			expectedErr: "DataType(20) for column 'col' is not supported for use as a primary key",
+			expectedErr: "DataType(21) for column 'col' is not supported for use as a primary key",
 		},
 		{
 			name:        "boolean - malformed",

lib/postgres/schema/schema.go

@@ -32,6 +32,7 @@ const (
 	TimeWithTimeZone
 	Date
 	Timestamp
+	TimestampWithTimeZone
 	Interval
 	UUID
 	Array
@@ -119,8 +120,10 @@ func ParseColumnDataType(colKind string, precision, scale *int, udtName *string)
 		return TimeWithTimeZone, nil, nil
 	case "date":
 		return Date, nil, nil
-	case "timestamp without time zone", "timestamp with time zone":
+	case "timestamp without time zone":
 		return Timestamp, nil, nil
+	case "timestamp with time zone":
+		return TimestampWithTimeZone, nil, nil
 	case "interval":
 		return Interval, nil, nil
 	case "uuid":

sources/mysql/adapter/adapter.go

@@ -110,8 +110,10 @@ func valueConverterForType(d schema.DataType, opts *schema.Opts) (converters.Val
 		return converters.MicroTimeConverter{}, nil
 	case schema.Date:
 		return converters.DateConverter{}, nil
-	case schema.DateTime, schema.Timestamp:
-		return converters.TimestampConverter{}, nil
+	case schema.DateTime:
+		return converters.MicroTimestampConverter{}, nil
+	case schema.Timestamp:
+		return converters.ZonedTimestampConverter{}, nil
 	case schema.Year:
 		return converters.YearConverter{}, nil
 	case schema.Enum:

sources/mysql/adapter/adapter_test.go

@@ -218,8 +218,8 @@ func TestValueConverterForType(t *testing.T) {
 			name:     "datetime",
 			dataType: schema.DateTime,
 			expected: debezium.Field{
-				Type:         "string",
-				DebeziumType: "io.debezium.time.Timestamp",
+				Type:         "int64",
+				DebeziumType: "io.debezium.time.MicroTimestamp",
 				FieldName:    colName,
 			},
 		},
@@ -228,7 +228,7 @@ func TestValueConverterForType(t *testing.T) {
 			dataType: schema.Timestamp,
 			expected: debezium.Field{
 				Type:         "string",
-				DebeziumType: "io.debezium.time.Timestamp",
+				DebeziumType: "io.debezium.time.ZonedTimestamp",
 				FieldName:    colName,
 			},
 		},

sources/postgres/adapter/adapter.go

@@ -106,7 +106,9 @@ func valueConverterForType(dataType schema.DataType, opts *schema.Opts) (convert
 	case schema.Date:
 		return converters.DateConverter{}, nil
 	case schema.Timestamp:
-		return converters.TimestampConverter{}, nil
+		return converters.MicroTimestampConverter{}, nil
+	case schema.TimestampWithTimeZone:
+		return converters.ZonedTimestampConverter{}, nil
 	case schema.Interval:
 		return PgIntervalConverter{}, nil
 	case schema.UUID: