clickhouse: fix date overflow for initial load

.github/workflows/flow.yml

@@ -352,9 +352,9 @@ jobs:
         id: ch-version
         run: |
           if [ "${{ matrix.db-version.ch }}" = "lts" ]; then
-            echo "ch_version=v25.8.11.66-lts" >> $GITHUB_OUTPUT
+            echo "ch_version=v25.8.28.1-lts" >> $GITHUB_OUTPUT
           elif [ "${{ matrix.db-version.ch }}" = "stable" ]; then
-            echo "ch_version=v25.12.4.35-stable" >> $GITHUB_OUTPUT
+            echo "ch_version=v26.3.16.16-lts" >> $GITHUB_OUTPUT
           elif [ "${{ matrix.db-version.ch }}" = "latest" ]; then
             # note: latest tag does not always reflect the latest version (could be an update on an lts),
             # but that is okay as we are only using it to invalidate the cache.

flow/e2e/clickhouse_mysql_test.go

@@ -1738,6 +1738,110 @@ func (s ClickHouseSuite) Test_MySQL_DateCoercion() {
 	RequireEnvCanceled(s.t, env)
 }
 
+// Test_MySQL_DateTime_ClickHouse_Range verifies that MySQL DATE/DATETIME values outside
+// ClickHouse's supported DateTime64/Date32 range ([1900, 2299]) are clamped to the nearest
+// boundary.
+func (s ClickHouseSuite) Test_MySQL_DateTime_ClickHouse_Range() {
+	if _, ok := s.source.(*MySqlSource); !ok {
+		s.t.Skip("only applies to mysql")
+	}
+
+	srcTableName := "test_datetime_range"
+	srcFullName := s.attachSchemaSuffix(srcTableName)
+	quotedSrcFullName := "\"" + strings.ReplaceAll(srcFullName, ".", "\".\"") + "\""
+	dstTableName := "test_datetime_range_dst"
+
+	require.NoError(s.t, s.source.Exec(s.t.Context(), fmt.Sprintf(`
+		CREATE TABLE IF NOT EXISTS %s (
+			id INT PRIMARY KEY,
+			d_null_low DATE NULL,
+			d_null_high DATE NULL,
+			d_nn_low DATE NOT NULL,
+			d_nn_high DATE NOT NULL,
+			d_ok DATE NOT NULL,
+			dt_null_low DATETIME NULL,
+			dt_null_high DATETIME NULL,
+			dt_nn_low DATETIME NOT NULL,
+			dt_nn_high DATETIME NOT NULL,
+			dt_ok DATETIME NOT NULL
+		)`, quotedSrcFullName)))
+
+	// '1000-06-15' and '9999-06-15' are valid MySQL DATE/DATETIME values but fall outside
+	// ClickHouse's [1900, 2299] range. The distinctive time-of-day (05:30:45) verifies that
+	// clamping preserves the time-of-day, matching parseDateTime64BestEffortOrNull.
+	insertRow := func(id int) error {
+		return s.source.Exec(s.t.Context(), fmt.Sprintf(`INSERT INTO %s (
+			id,
+			d_null_low, d_null_high, d_nn_low, d_nn_high, d_ok,
+			dt_null_low, dt_null_high, dt_nn_low, dt_nn_high, dt_ok
+		) VALUES (
+			%d,
+			'1000-06-15', '9999-06-15', '1000-06-15', '9999-06-15', '2000-01-02',
+			'1000-06-15 05:30:45', '9999-06-15 05:30:45', '1000-06-15 05:30:45', '9999-06-15 05:30:45', '2000-01-02 03:04:05'
+		)`, quotedSrcFullName, id))
+	}
+
+	require.NoError(s.t, insertRow(1))
+
+	connectionGen := FlowConnectionGenerationConfig{
+		FlowJobName:      s.attachSuffix(srcTableName),
+		TableNameMapping: map[string]string{srcFullName: dstTableName},
+		Destination:      s.Peer().Name,
+	}
+	flowConnConfig := connectionGen.GenerateFlowConnectionConfigs(s)
+	flowConnConfig.DoInitialSnapshot = true
+	// Enable nullable columns so nullable source columns map to Nullable(...) in ClickHouse,
+	// verifying that clamping happens regardless of column nullability.
+	flowConnConfig.Env = map[string]string{"PEERDB_NULLABLE": "true"}
+
+	tc := NewTemporalClient(s.t)
+	env := ExecutePeerflow(s.t, tc, flowConnConfig)
+	SetupCDCFlowStatusQuery(s.t, env, flowConnConfig)
+
+	EnvWaitForCount(env, s, "waiting on snapshot", dstTableName, "id", 1)
+
+	require.NoError(s.t, insertRow(2))
+
+	EnvWaitForCount(env, s, "waiting on cdc", dstTableName, "id", 2)
+
+	colExprs := []string{"id"}
+	for _, c := range []string{
+		"d_null_low", "d_null_high", "d_nn_low", "d_nn_high", "d_ok",
+		"dt_null_low", "dt_null_high", "dt_nn_low", "dt_nn_high", "dt_ok",
+	} {
+		colExprs = append(colExprs, fmt.Sprintf("ifNull(toString(%s),'')", c))
+	}
+	cols := strings.Join(colExprs, ",")
+	rows, err := s.GetRows(dstTableName, cols)
+	require.NoError(s.t, err)
+	require.Len(s.t, rows.Records, 2, "expected snapshot + cdc rows")
+
+	assertStr := func(qv types.QValue, expect string) {
+		sv, ok := qv.Value().(string)
+		require.Truef(s.t, ok, "expected string, got %T", qv.Value())
+		require.Equal(s.t, expect, sv)
+	}
+
+	// Out-of-range values are clamped to the nearest ClickHouse boundary
+	for _, row := range rows.Records {
+		// DATE (Date32): clamps to the boundary day.
+		assertStr(row[1], "1900-01-01") // d_null_low
+		assertStr(row[2], "2299-12-31") // d_null_high
+		assertStr(row[3], "1900-01-01") // d_nn_low
+		assertStr(row[4], "2299-12-31") // d_nn_high
+		assertStr(row[5], "2000-01-02") // d_ok
+		// DATETIME (DateTime64(6)): clamps the date, preserves the time-of-day.
+		assertStr(row[6], "1900-01-01 05:30:45.000000")  // dt_null_low
+		assertStr(row[7], "2299-12-31 05:30:45.000000")  // dt_null_high
+		assertStr(row[8], "1900-01-01 05:30:45.000000")  // dt_nn_low
+		assertStr(row[9], "2299-12-31 05:30:45.000000")  // dt_nn_high
+		assertStr(row[10], "2000-01-02 03:04:05.000000") // dt_ok
+	}
+
+	env.Cancel(s.t.Context())
+	RequireEnvCanceled(s.t, env)
+}
+
 func (s ClickHouseSuite) Test_MySQL_Column_Position_Shifting_DDL_Error() {
 	mySource, ok := s.source.(*MySqlSource)
 	if !ok {

flow/model/qvalue/avro_converter.go

@@ -369,8 +369,12 @@ func (c *QValueAvroConverter) processGoTime(t time.Duration, so sizeOpt) (any, i
 	return t, varIntSize(int64(t/time.Microsecond), so)
 }
 
-func (c *QValueAvroConverter) processGeneralTime(t time.Time, format string, avroVal int64, so sizeOpt) (any, int64) {
-	// Bigquery will not allow timestamp if it is less than 1AD and more than 9999AD
+const (
+	clickHouseMinYear = 1900
+	clickHouseMaxYear = 2299
+)
+
+func (c *QValueAvroConverter) processGeneralTime(t time.Time, format string, isDate bool, so sizeOpt) (any, int64) {
 	switch c.TargetDWH {
 	case protos.DBType_BIGQUERY:
 		year := t.Year()
@@ -383,26 +387,36 @@ func (c *QValueAvroConverter) processGeneralTime(t time.Time, format string, avr
 				return time.Date(1, 1, 1, 0, 0, 0, 0, time.UTC), varIntSize(0, so)
 			}
 		}
+	case protos.DBType_CLICKHOUSE:
+		if year := t.Year(); year < clickHouseMinYear {
+			t = time.Date(clickHouseMinYear, time.January, 1, t.Hour(), t.Minute(), t.Second(), t.Nanosecond(), t.Location())
+		} else if year > clickHouseMaxYear {
+			t = time.Date(clickHouseMaxYear, time.December, 31, t.Hour(), t.Minute(), t.Second(), t.Nanosecond(), t.Location())
+		}
 	case protos.DBType_SNOWFLAKE:
 		// Snowflake has issues with avro timestamp types, returning as string form
 		// See: https://stackoverflow.com/questions/66104762/snowflake-date-column-have-incorrect-date-from-avro-file
 		s := t.Format(format)
 		return s, stringSize(s, so)
 	}
+	// Timestamps are micros since epoch; dates are days since epoch. Both encoded as Avro varint.
+	avroVal := t.UnixMicro()
+	if isDate {
+		avroVal = t.Unix() / 86400
+	}
 	return t, varIntSize(avroVal, so)
 }
 
 func (c *QValueAvroConverter) processGoTimestampTZ(t time.Time, so sizeOpt) (any, int64) {
-	return c.processGeneralTime(t, "2006-01-02 15:04:05.999999-0700", t.UnixMicro(), so)
+	return c.processGeneralTime(t, "2006-01-02 15:04:05.999999-0700", false, so)
 }
 
 func (c *QValueAvroConverter) processGoTimestamp(t time.Time, so sizeOpt) (any, int64) {
-	return c.processGeneralTime(t, "2006-01-02 15:04:05.999999", t.UnixMicro(), so)
+	return c.processGeneralTime(t, "2006-01-02 15:04:05.999999", false, so)
 }
 
 func (c *QValueAvroConverter) processGoDate(t time.Time, so sizeOpt) (any, int64) {
-	// Date is days since epoch, encoded as Avro int (varint)
-	return c.processGeneralTime(t, "2006-01-02", t.Unix()/86400, so)
+	return c.processGeneralTime(t, "2006-01-02", true, so)
 }
 
 func (c *QValueAvroConverter) processNullableUnion(