clickhouse: fix date overflow for initial load

flow/model/qvalue/avro_converter.go

@@ -369,8 +369,17 @@ 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
+)
+
+var (
+	clickHouseMinTime = time.Date(clickHouseMinYear, time.January, 1, 0, 0, 0, 0, time.UTC)
+	clickHouseMaxTime = time.Date(clickHouseMaxYear, time.December, 31, 23, 59, 59, 999999000, time.UTC)
+)
+
+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 +392,44 @@ 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:
+		year := t.Year()
+		if year < clickHouseMinYear || year > clickHouseMaxYear {
+			c.logger.Warn("value outside ClickHouse's supported DateTime range [1900, 2299]",
+				"value", t.String(), "nullable", c.Nullable)
+			if c.Nullable {
+				return nil, so.nullableSize()
+			}
+			if year < clickHouseMinYear {
+				t = clickHouseMinTime
+			} else {
+				t = clickHouseMaxTime
+			}
+		}
 	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(

flow/model/qvalue/avro_converter_test.go

@@ -2,6 +2,8 @@
 
 import (
 	"context"
+	"io"
+	"log/slog"
 	"math/big"
 	"strings"
 	"testing"
@@ -77,6 +79,62 @@
 	}
 }
 
+func TestClickHouseDateTimeRange(t *testing.T) {
+	ctx := context.Background()
+	logger := log.NewStructuredLogger(slog.New(slog.NewTextHandler(io.Discard, nil)))
+
+	inRange := time.Date(2000, 1, 2, 3, 4, 5, 0, time.UTC)
+	belowRange := time.Date(1000, 1, 1, 0, 0, 0, 0, time.UTC) // MySQL DATETIME min, below ClickHouse
+	aboveRange := time.Date(9999, 12, 31, 23, 59, 59, 0, time.UTC)
+
+	convert := func(t *testing.T, qv types.QValue, dwh protos.DBType, nullable bool) any {
+		t.Helper()
+		field := types.QField{Type: qv.Kind(), Nullable: nullable}
+		val, _, err := QValueToAvro(ctx, qv, &field, dwh, logger, false, nil, internal.BinaryFormatRaw, false)
+		require.NoError(t, err)
+		return val
+	}
+
+	tests := []struct {
+		name     string
+		qv       types.QValue
+		nullable bool
+		// nil expected means the value should be nulled; otherwise the (possibly clamped) time
+		expected *time.Time
+	}{
+		{"timestamp_in_range", types.QValueTimestamp{Val: inRange}, false, &inRange},
+		{"timestamp_below_nonnullable_clamps", types.QValueTimestamp{Val: belowRange}, false, &clickHouseMinTime},
+		{"timestamp_above_nonnullable_clamps", types.QValueTimestamp{Val: aboveRange}, false, &clickHouseMaxTime},
+		{"timestamp_below_nullable_nulls", types.QValueTimestamp{Val: belowRange}, true, nil},
+		{"timestamp_above_nullable_nulls", types.QValueTimestamp{Val: aboveRange}, true, nil},
+		{"timestamptz_below_nonnullable_clamps", types.QValueTimestampTZ{Val: belowRange}, false, &clickHouseMinTime},
+		{"timestamptz_above_nullable_nulls", types.QValueTimestampTZ{Val: aboveRange}, true, nil},
+		{"date_below_nonnullable_clamps", types.QValueDate{Val: belowRange}, false, &clickHouseMinTime},
+		{"date_above_nullable_nulls", types.QValueDate{Val: aboveRange}, true, nil},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			val := convert(t, tt.qv, protos.DBType_CLICKHOUSE, tt.nullable)
+			if tt.expected == nil {
+				assert.Nil(t, val, "out-of-range value should be nulled")
+				return
+			}
+			ts, ok := val.(time.Time)
+			require.Truef(t, ok, "expected time.Time, got %T", val)
+			assert.Truef(t, ts.Equal(*tt.expected), "expected %s, got %s", tt.expected, ts)
+		})
+	}
+
+	// Non-ClickHouse targets must not clamp: an out-of-range value passes through untouched.
+	t.Run("other_dwh_no_clamp", func(t *testing.T) {
+		val := convert(t, types.QValueTimestamp{Val: aboveRange}, protos.DBType_POSTGRES, false)
+		ts, ok := val.(time.Time)
+		require.Truef(t, ok, "expected time.Time, got %T", val)
+		assert.True(t, ts.Equal(aboveRange))
+	})
+}
+
 func TestAvroQValueSize(t *testing.T) {
 	ctx := context.Background()
 	env := map[string]string{