added backward compatibility test

Author: NehharShah

flow/e2e/clickhouse_mysql_test.go

@@ -116,7 +116,8 @@ func (s ClickHouseSuite) Test_MySQL_Time() {
 
 	EnvWaitForEqualTablesWithNames(env, s, "waiting on cdc", srcTableName, dstTableName, "id,\"key\",d,dt,tm,t")
 
-	// Verify that TIME column uses Time64(3) when ClickHouse version >= 25.6 and internal version >= InternalVersion_ClickHouseTime64
+	// Verify that TIME column uses Time64(3) when ClickHouse version >= 25.6 (assumes latest internal version)
+	// Backward compatibility with older internal versions is tested in Test_MySQL_Time_BackwardCompatibility
 	ch, err := connclickhouse.Connect(s.t.Context(), nil, s.Peer().GetClickhouseConfig())
 	require.NoError(s.t, err)
 	defer ch.Close()
@@ -139,14 +140,8 @@ func (s ClickHouseSuite) Test_MySQL_Time() {
 
 		// If ClickHouse >= 25.6 and using latest internal version, should use Time64(3)
 		if major > 25 || (major == 25 && minor >= 6) {
-			if flowConnConfig.Version >= shared.InternalVersion_ClickHouseTime64 {
-				require.Contains(s.t, columnType, "Time64(3)",
-					"Expected Time64(3) for TIME column when ClickHouse >= 25.6 and internal version >= InternalVersion_ClickHouseTime64, got %s", columnType)
-			} else {
-				// Old version should use DateTime64(6)
-				require.Contains(s.t, columnType, "DateTime64(6)",
-					"Expected DateTime64(6) for TIME column with old internal version, got %s", columnType)
-			}
+			require.Contains(s.t, columnType, "Time64(3)",
+				"Expected Time64(3) for TIME column when ClickHouse >= 25.6 and using latest internal version, got %s", columnType)
 		} else {
 			// Older ClickHouse versions should use DateTime64(6)
 			require.Contains(s.t, columnType, "DateTime64(6)",
@@ -158,6 +153,64 @@ func (s ClickHouseSuite) Test_MySQL_Time() {
 	RequireEnvCanceled(s.t, env)
 }
 
+func (s ClickHouseSuite) Test_MySQL_Time_BackwardCompatibility() {
+	if _, ok := s.source.(*MySqlSource); !ok {
+		s.t.Skip("only applies to mysql")
+	}
+
+	srcTableName := "test_datetime_backward_compat"
+	srcFullName := s.attachSchemaSuffix(srcTableName)
+	quotedSrcFullName := "\"" + strings.ReplaceAll(srcFullName, ".", "\".\"") + "\""
+	dstTableName := "test_datetime_backward_compat_dst"
+
+	require.NoError(s.t, s.source.Exec(s.t.Context(), fmt.Sprintf(`
+		CREATE TABLE IF NOT EXISTS %s (
+			id SERIAL PRIMARY KEY,
+			"key" TEXT NOT NULL,
+			t TIME NOT NULL
+		)
+	`, quotedSrcFullName)))
+
+	require.NoError(s.t, s.source.Exec(s.t.Context(), fmt.Sprintf(`INSERT INTO %s ("key",t) VALUES
+		('init','14:21.654321')`,
+		quotedSrcFullName)))
+
+	connectionGen := FlowConnectionGenerationConfig{
+		FlowJobName:      s.attachSuffix(srcTableName),
+		TableNameMapping: map[string]string{srcFullName: dstTableName},
+		Destination:      s.Peer().Name,
+	}
+	flowConnConfig := connectionGen.GenerateFlowConnectionConfigs(s)
+	flowConnConfig.DoInitialSnapshot = true
+	// Explicitly set to old internal version to test backward compatibility
+	flowConnConfig.Version = shared.InternalVersion_First
+
+	tc := NewTemporalClient(s.t)
+	env := ExecutePeerflow(s.t, tc, flowConnConfig)
+	SetupCDCFlowStatusQuery(s.t, env, flowConnConfig)
+
+	EnvWaitForEqualTablesWithNames(env, s, "waiting on initial", srcTableName, dstTableName, "id,\"key\",t")
+
+	// Verify that TIME column uses DateTime64(6) even with ClickHouse >= 25.6 when using old internal version
+	ch, err := connclickhouse.Connect(s.t.Context(), nil, s.Peer().GetClickhouseConfig())
+	require.NoError(s.t, err)
+	defer ch.Close()
+
+	var columnType string
+	err = ch.QueryRow(s.t.Context(), fmt.Sprintf(
+		"SELECT type FROM system.columns WHERE database = currentDatabase() AND table = %s AND name = 't'",
+		clickhouse.QuoteLiteral(dstTableName),
+	)).Scan(&columnType)
+	require.NoError(s.t, err)
+
+	// With old internal version, should always use DateTime64(6) regardless of ClickHouse version
+	require.Contains(s.t, columnType, "DateTime64(6)",
+		"Expected DateTime64(6) for TIME column with old internal version (InternalVersion_First), got %s", columnType)
+
+	env.Cancel(s.t.Context())
+	RequireEnvCanceled(s.t, env)
+}
+
 func (s ClickHouseSuite) Test_MySQL_Bit() {
 	if _, ok := s.source.(*MySqlSource); !ok {
 		s.t.Skip("only applies to mysql")