From e930747c3f7e52b0b16f284ff2d87cfdbcd6ac41 Mon Sep 17 00:00:00 2001
From: Danish Sarwar <danish.sarwar@practo.com>
Date: Wed, 29 Jul 2026 11:39:57 +0530
Subject: [PATCH] fix: zero-pad years below 1000 in date/datetime formatting

convertDebeziumDate, convertDebeziumMilliseconds and
convertDebeziumMicroseconds printed the year with %d, so a MySQL value
like 0202-04-17 was serialized to S3 as "202-04-17" (9 chars).
Redshift COPY (json 'auto') rejects such values with err 1205 "Invalid
Date Format - length must be 10 or more" and the loader retries the
same batch forever, stalling the sink and burning cluster cycles.

Pad the year with %04d so years below 1000 serialize as valid
YYYY-MM-DD and load faithfully. Years 1000-9999 are unaffected.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
---
 pkg/transformer/debezium/message.go      |  6 +--
 pkg/transformer/debezium/message_test.go | 66 ++++++++++++++++++++++++
 2 files changed, 69 insertions(+), 3 deletions(-)

diff --git a/pkg/transformer/debezium/message.go b/pkg/transformer/debezium/message.go
index 220c2cdbb..0a42c4caa 100644
--- a/pkg/transformer/debezium/message.go
+++ b/pkg/transformer/debezium/message.go
@@ -133,7 +133,7 @@ func convertDebeziumDate(days int) string {
 	ts := FromUnixMilli(int64(days * 86400 * 1000))
 	ts = ts.UTC()
 	return fmt.Sprintf(
-		"%d-%02d-%02d",
+		"%04d-%02d-%02d",
 		ts.Year(), ts.Month(), ts.Day(),
 	)
 }
@@ -149,7 +149,7 @@ func convertDebeziumMilliseconds(ms int, length int) string {
 	ts = ts.UTC()
 
 	result := fmt.Sprintf(
-		"%d-%02d-%02d %02d:%02d:%02d",
+		"%04d-%02d-%02d %02d:%02d:%02d",
 		ts.Year(), ts.Month(), ts.Day(),
 		ts.Hour(), ts.Minute(), ts.Second(),
 	)
@@ -177,7 +177,7 @@ func convertDebeziumMicroseconds(us int, length int) string {
 	ts = ts.UTC()
 
 	result := fmt.Sprintf(
-		"%d-%02d-%02d %02d:%02d:%02d",
+		"%04d-%02d-%02d %02d:%02d:%02d",
 		ts.Year(), ts.Month(), ts.Day(),
 		ts.Hour(), ts.Minute(), ts.Second(),
 	)
diff --git a/pkg/transformer/debezium/message_test.go b/pkg/transformer/debezium/message_test.go
index 0bfdec100..bd3af928c 100644
--- a/pkg/transformer/debezium/message_test.go
+++ b/pkg/transformer/debezium/message_test.go
@@ -91,6 +91,72 @@ func TestConvertDebeziumFormattedTime(t *testing.T) {
 			sourceLength:  "3",
 			formattedTime: "2020-10-15 04:31:57.708",
 		},
+		{
+			name:          "test11: DATE epoch",
+			value:         "0",
+			sourceType:    "DATE",
+			sourceLength:  "",
+			formattedTime: "1970-01-01",
+		},
+		{
+			name:          "test12: DATE recent",
+			value:         "19181",
+			sourceType:    "DATE",
+			sourceLength:  "",
+			formattedTime: "2022-07-08",
+		},
+		{
+			name:          "test13: DATE max year",
+			value:         "2932896",
+			sourceType:    "DATE",
+			sourceLength:  "",
+			formattedTime: "9999-12-31",
+		},
+		// Years < 1000 must be zero-padded to 4 digits, else Redshift
+		// COPY rejects the value (Invalid Date Format - length must be
+		// 10 or more) and the loader retries the batch forever.
+		{
+			name:          "test14: DATE year < 1000 is zero-padded",
+			value:         "-645643",
+			sourceType:    "DATE",
+			sourceLength:  "",
+			formattedTime: "0202-04-17",
+		},
+		{
+			name:          "test15: DATE 3 digit year is zero-padded",
+			value:         "-361910",
+			sourceType:    "DATE",
+			sourceLength:  "",
+			formattedTime: "0979-02-15",
+		},
+		{
+			name:          "test16: DATE last 3 digit year",
+			value:         "-354286",
+			sourceType:    "DATE",
+			sourceLength:  "",
+			formattedTime: "0999-12-31",
+		},
+		{
+			name:          "test17: DATE first 4 digit year",
+			value:         "-354285",
+			sourceType:    "DATE",
+			sourceLength:  "",
+			formattedTime: "1000-01-01",
+		},
+		{
+			name:          "test18: DATETIME year < 1000 is zero-padded",
+			value:         "-55704758400000",
+			sourceType:    "DATETIME",
+			sourceLength:  "",
+			formattedTime: "0204-10-15 00:00:00",
+		},
+		{
+			name:          "test19: DATETIME(6) year < 1000 is zero-padded",
+			value:         "-55805894999243000",
+			sourceType:    "DATETIME",
+			sourceLength:  "6",
+			formattedTime: "0201-08-01 10:30:00.757000",
+		},
 	}
 
 	for _, tc := range tests {