sink/mysql: align DDL time defaults with origin_default

cdc/sink/ddlsink/mysql/helper.go

@@ -0,0 +1,205 @@
+// Copyright 2025 PingCAP, Inc.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package mysql
+
+import (
+	"context"
+	"database/sql"
+	"fmt"
+	"strconv"
+	"strings"
+	"time"
+
+	dmysql "github.com/go-sql-driver/mysql"
+	"github.com/pingcap/errors"
+	"github.com/pingcap/log"
+	"github.com/pingcap/tidb/pkg/errno"
+	timodel "github.com/pingcap/tidb/pkg/meta/model"
+	"github.com/pingcap/tidb/pkg/parser"
+	"github.com/pingcap/tidb/pkg/parser/ast"
+	tidbmysql "github.com/pingcap/tidb/pkg/parser/mysql"
+	"github.com/pingcap/tiflow/cdc/model"
+	"go.uber.org/zap"
+)
+
+func setSessionTimestamp(ctx context.Context, tx *sql.Tx, unixTimestamp float64) error {
+	_, err := tx.ExecContext(ctx, fmt.Sprintf("SET TIMESTAMP = %s", formatUnixTimestamp(unixTimestamp)))
+	return err
+}
+
+func resetSessionTimestamp(ctx context.Context, tx *sql.Tx) error {
+	_, err := tx.ExecContext(ctx, "SET TIMESTAMP = DEFAULT")
+	return err
+}
+
+func isIgnorableSessionTimestampErr(err error) bool {
+	if err == nil {
+		return false
+	}
+	mysqlErr, ok := errors.Cause(err).(*dmysql.MySQLError)
+	if !ok {
+		return false
+	}
+	switch mysqlErr.Number {
+	case uint16(errno.ErrWrongValueForVar),
+		uint16(errno.ErrTruncatedWrongValue),
+		uint16(errno.ErrUnknownSystemVariable):
+		return true
+	default:
+		return false
+	}
+}
+
+func formatUnixTimestamp(unixTimestamp float64) string {
+	return strconv.FormatFloat(unixTimestamp, 'f', 6, 64)
+}
+
+func ddlSessionTimestampFromOriginDefault(ddl *model.DDLEvent, timezone string) (float64, bool) {
+	if ddl == nil || ddl.TableInfo == nil {
+		return 0, false
+	}
+	targetColumns, err := extractCurrentTimestampDefaultColumns(ddl.Query)
+	if err != nil || len(targetColumns) == 0 {
+		return 0, false
+	}
+
+	for _, col := range ddl.TableInfo.Columns {
+		if _, ok := targetColumns[col.Name.L]; !ok {
+			continue
+		}
+		val := col.GetOriginDefaultValue()
+		valStr, ok := val.(string)
+		if !ok || valStr == "" {
+			continue
+		}
+		ts, err := parseOriginDefaultTimestamp(valStr, col, timezone)
+		if err != nil {
+			log.Warn("Failed to parse OriginDefaultValue for DDL timestamp",
+				zap.String("column", col.Name.O),
+				zap.String("originDefault", valStr),
+				zap.Error(err))
+			continue
+		}
+		log.Info("Using OriginDefaultValue for DDL timestamp",
+			zap.String("column", col.Name.O),
+			zap.String("originDefault", valStr),
+			zap.Float64("timestamp", ts),
+			zap.String("timezone", timezone))
+		return ts, true
+	}
+
+	return 0, false
+}
+
+func extractCurrentTimestampDefaultColumns(query string) (map[string]struct{}, error) {
+	p := parser.New()
+	stmt, err := p.ParseOneStmt(query, "", "")
+	if err != nil {
+		return nil, err
+	}
+
+	cols := make(map[string]struct{})
+	switch s := stmt.(type) {
+	case *ast.CreateTableStmt:
+		for _, col := range s.Cols {
+			if hasCurrentTimestampDefault(col) {
+				cols[col.Name.Name.L] = struct{}{}
+			}
+		}
+	case *ast.AlterTableStmt:
+		for _, spec := range s.Specs {
+			switch spec.Tp {
+			case ast.AlterTableAddColumns, ast.AlterTableModifyColumn, ast.AlterTableChangeColumn, ast.AlterTableAlterColumn:
+				for _, col := range spec.NewColumns {
+					if hasCurrentTimestampDefault(col) {
+						cols[col.Name.Name.L] = struct{}{}
+					}
+				}
+			}
+		}
+	}
+
+	return cols, nil
+}
+
+func hasCurrentTimestampDefault(col *ast.ColumnDef) bool {
+	if col == nil {
+		return false
+	}
+	for _, opt := range col.Options {
+		if opt.Tp != ast.ColumnOptionDefaultValue {
+			continue
+		}
+		if isCurrentTimestampExpr(opt.Expr) {
+			return true
+		}
+	}
+	return false
+}
+
+func isCurrentTimestampExpr(expr ast.ExprNode) bool {
+	if expr == nil {
+		return false
+	}
+	switch v := expr.(type) {
+	case *ast.FuncCallExpr:
+		return isCurrentTimestampFuncName(v.FnName.L)
+	case ast.ValueExpr:
+		return isCurrentTimestampFuncName(strings.ToLower(v.GetString()))
+	default:
+		return false
+	}
+}
+
+func isCurrentTimestampFuncName(name string) bool {
+	switch name {
+	case ast.CurrentTimestamp, ast.Now, ast.LocalTime, ast.LocalTimestamp:
+		return true
+	default:
+		return false
+	}
+}
+
+func parseOriginDefaultTimestamp(val string, col *timodel.ColumnInfo, timezone string) (float64, error) {
+	loc, err := resolveOriginDefaultLocation(col, timezone)
+	if err != nil {
+		return 0, err
+	}
+	return parseTimestampInLocation(val, loc)
+}
+
+func resolveOriginDefaultLocation(col *timodel.ColumnInfo, timezone string) (*time.Location, error) {
+	if col != nil && col.GetType() == tidbmysql.TypeTimestamp && col.Version >= timodel.ColumnInfoVersion1 {
+		return time.UTC, nil
+	}
+	if timezone == "" {
+		return time.UTC, nil
+	}
+	tz := strings.Trim(timezone, "\"")
+	return time.LoadLocation(tz)
+}
+
+func parseTimestampInLocation(val string, loc *time.Location) (float64, error) {
+	formats := []string{
+		"2006-01-02 15:04:05",
+		"2006-01-02 15:04:05.999999",
+	}
+	for _, f := range formats {
+		t, err := time.ParseInLocation(f, val, loc)
+		if err == nil {
+			return float64(t.UnixNano()) / float64(time.Second), nil
+		}
+	}
+	return 0, fmt.Errorf("failed to parse timestamp: %s", val)
+}

cdc/sink/ddlsink/mysql/mysql_ddl_sink.go

@@ -211,7 +211,44 @@ func (m *DDLSink) execDDL(pctx context.Context, ddl *model.DDLEvent) error {
 		return err
 	}
 
+	ddlTimestamp, useSessionTimestamp := ddlSessionTimestampFromOriginDefault(ddl, m.cfg.Timezone)
+
+	if useSessionTimestamp {
+		// set the session timestamp to match upstream DDL execution time
+		if err := setSessionTimestamp(ctx, tx, ddlTimestamp); err != nil {
+			if isIgnorableSessionTimestampErr(err) {
+				log.Warn("Fail to set session timestamp for DDL, continue without session timestamp",
+					zap.Float64("timestamp", ddlTimestamp),
+					zap.Uint64("startTs", ddl.StartTs),
+					zap.Uint64("commitTs", ddl.CommitTs),
+					zap.String("query", ddl.Query),
+					zap.Error(err))
+				useSessionTimestamp = false
+			} else {
+				log.Error("Fail to set session timestamp for DDL",
+					zap.Float64("timestamp", ddlTimestamp),
+					zap.Uint64("startTs", ddl.StartTs),
+					zap.Uint64("commitTs", ddl.CommitTs),
+					zap.String("query", ddl.Query),
+					zap.Error(err))
+				if rbErr := tx.Rollback(); rbErr != nil {
+					log.Error("Failed to rollback", zap.String("changefeed", m.id.ID), zap.Error(rbErr))
+				}
+				return err
+			}
+		}
+	}
+
 	if _, err = tx.ExecContext(ctx, ddl.Query); err != nil {
+		log.Error("Failed to ExecContext", zap.Any("err", err), zap.Any("query", ddl.Query))
+		if useSessionTimestamp {
+			if tsErr := resetSessionTimestamp(ctx, tx); tsErr != nil {
+				log.Warn("Failed to reset session timestamp after DDL execution failure",
+					zap.String("namespace", m.id.Namespace),
+					zap.String("changefeed", m.id.ID),
+					zap.Error(tsErr))
+			}
+		}
 		if rbErr := tx.Rollback(); rbErr != nil {
 			log.Error("Failed to rollback",
 				zap.String("namespace", m.id.Namespace),
@@ -222,13 +259,41 @@ func (m *DDLSink) execDDL(pctx context.Context, ddl *model.DDLEvent) error {
 		return err
 	}
 
+	if useSessionTimestamp {
+		// reset session timestamp after DDL execution to avoid affecting subsequent operations
+		if err := resetSessionTimestamp(ctx, tx); err != nil {
+			if isIgnorableSessionTimestampErr(err) {
+				log.Warn("Failed to reset session timestamp after DDL execution, continue",
+					zap.String("namespace", m.id.Namespace),
+					zap.String("changefeed", m.id.ID),
+					zap.Error(err))
+			} else {
+				log.Error("Failed to reset session timestamp after DDL execution", zap.Error(err))
+				if rbErr := tx.Rollback(); rbErr != nil {
+					log.Error("Failed to rollback", zap.String("sql", ddl.Query), zap.Error(rbErr))
+				}
+				return errors.WrapError(errors.ErrMySQLTxnError, errors.WithMessage(err, fmt.Sprintf("Query info: %s; ", ddl.Query)))
+			}
+		}
+	}
+
 	if err = tx.Commit(); err != nil {
 		return errors.WrapError(errors.ErrMySQLTxnError, errors.WithMessage(err, fmt.Sprintf("Query info: %s; ", ddl.Query)))
 	}
 
-	log.Info("Exec DDL succeeded",
-		zap.String("namespace", m.id.Namespace), zap.String("changefeed", m.id.ID),
-		zap.Duration("duration", time.Since(start)), zap.String("sql", ddl.Query))
+	logFields := []zap.Field{
+		zap.String("namespace", m.id.Namespace),
+		zap.String("changefeed", m.id.ID),
+		zap.Duration("duration", time.Since(start)),
+		zap.String("sql", ddl.Query),
+	}
+
+	if useSessionTimestamp {
+		logFields = append(logFields, zap.Float64("sessionTimestamp", ddlTimestamp))
+	}
+
+	log.Info("Exec DDL succeeded", logFields...)
+
 	return nil
 }
 
@@ -344,7 +409,7 @@ func getDDLCreateTime(ctx context.Context, db *sql.DB) string {
 // getDDLStateFromTiDB retrieves the ddl job status of the ddl query from downstream tidb based on the ddl query and the approximate ddl create time.
 func getDDLStateFromTiDB(ctx context.Context, db *sql.DB, ddl string, createTime string) (timodel.JobState, error) {
 	// ddlCreateTime and createTime are both based on UTC timezone of downstream
-	showJobs := fmt.Sprintf(`SELECT JOB_ID, JOB_TYPE, SCHEMA_STATE, SCHEMA_ID, TABLE_ID, STATE, QUERY FROM information_schema.ddl_jobs 
+	showJobs := fmt.Sprintf(`SELECT JOB_ID, JOB_TYPE, SCHEMA_STATE, SCHEMA_ID, TABLE_ID, STATE, QUERY FROM information_schema.ddl_jobs
 	WHERE CREATE_TIME >= "%s" AND QUERY = "%s";`, createTime, ddl)
 	//nolint:rowserrcheck
 	jobsRows, err := db.QueryContext(ctx, showJobs)