This is an automated cherry-pick of pingcap#10932

Signed-off-by: ti-chi-bot <ti-community-prow-bot@tidb.io>

cdc/entry/mounter.go

@@ -279,8 +279,13 @@ func parseJob(v []byte, startTs, CRTs uint64) (*timodel.Job, error) {
 
 func datum2Column(
 	tableInfo *model.TableInfo, datums map[int64]types.Datum, tz *time.Location,
+<<<<<<< HEAD
 ) ([]*model.Column, []types.Datum, []rowcodec.ColInfo, error) {
 	cols := make([]*model.Column, len(tableInfo.RowColumnsOffset))
+=======
+) ([]*model.ColumnData, []types.Datum, []*timodel.ColumnInfo, error) {
+	cols := make([]*model.ColumnData, len(tableInfo.RowColumnsOffset))
+>>>>>>> e61d080e34 (mounter(ticdc): timezone fill default value should also consider tz. (#10932))
 	rawCols := make([]types.Datum, len(tableInfo.RowColumnsOffset))
 
 	// columnInfos and rowColumnInfos hold different column metadata,
@@ -307,7 +312,11 @@ func datum2Column(
 		if exist {
 			colValue, size, warn, err = formatColVal(colDatums, colInfo)
 		} else {
+<<<<<<< HEAD
 			colDatums, colValue, size, warn, err = getDefaultOrZeroValue(colInfo, tz)
+=======
+			colDatum, colValue, size, warn, err = getDefaultOrZeroValue(colInfo, tz)
+>>>>>>> e61d080e34 (mounter(ticdc): timezone fill default value should also consider tz. (#10932))
 		}
 		if err != nil {
 			return nil, nil, nil, errors.Trace(err)
@@ -349,7 +358,11 @@ func (m *mounter) mountRowKVEntry(tableInfo *model.TableInfo, row *rowKVEntry, d
 	if row.PreRowExist {
 		// FIXME(leoppro): using pre table info to mounter pre column datum
 		// the pre column and current column in one event may using different table info
+<<<<<<< HEAD
 		preCols, preRawCols, extendColumnInfos, err = datum2Column(tableInfo, row.PreRow, m.tz)
+=======
+		preCols, preRawCols, columnInfos, err = datum2Column(tableInfo, row.PreRow, m.tz)
+>>>>>>> e61d080e34 (mounter(ticdc): timezone fill default value should also consider tz. (#10932))
 		if err != nil {
 			return nil, rawRow, errors.Trace(err)
 		}
@@ -358,7 +371,11 @@ func (m *mounter) mountRowKVEntry(tableInfo *model.TableInfo, row *rowKVEntry, d
 	var cols []*model.Column
 	var rawCols []types.Datum
 	if row.RowExist {
+<<<<<<< HEAD
 		cols, rawCols, extendColumnInfos, err = datum2Column(tableInfo, row.Row, m.tz)
+=======
+		cols, rawCols, columnInfos, err = datum2Column(tableInfo, row.Row, m.tz)
+>>>>>>> e61d080e34 (mounter(ticdc): timezone fill default value should also consider tz. (#10932))
 		if err != nil {
 			return nil, rawRow, errors.Trace(err)
 		}
@@ -494,7 +511,13 @@ func formatColVal(datum types.Datum, col *timodel.ColumnInfo) (
 // https://github.com/golang/go/blob/go1.17.4/src/database/sql/driver/types.go#L236
 // Supported type is: nil, basic type(Int, Int8,..., Float32, Float64, String), Slice(uint8), other types not support
 // TODO: Check default expr support
+<<<<<<< HEAD
 func getDefaultOrZeroValue(col *timodel.ColumnInfo, tz *time.Location) (types.Datum, any, int, string, error) {
+=======
+func getDefaultOrZeroValue(
+	col *timodel.ColumnInfo, tz *time.Location,
+) (types.Datum, any, int, string, error) {
+>>>>>>> e61d080e34 (mounter(ticdc): timezone fill default value should also consider tz. (#10932))
 	var (
 		d   types.Datum
 		err error
@@ -513,6 +536,15 @@ func getDefaultOrZeroValue(col *timodel.ColumnInfo, tz *time.Location) (types.Da
 		if err != nil {
 			return d, d.GetValue(), sizeOfDatum(d), "", errors.Trace(err)
 		}
+		switch col.GetType() {
+		case mysql.TypeTimestamp:
+			t := d.GetMysqlTime()
+			err = t.ConvertTimeZone(time.UTC, tz)
+			if err != nil {
+				return d, d.GetValue(), sizeOfDatum(d), "", errors.Trace(err)
+			}
+			d.SetMysqlTime(t)
+		}
 	} else if !mysql.HasNotNullFlag(col.GetFlag()) {
 		// NOTICE: NotNullCheck need do after OriginDefaultValue check, as when TiDB meet "amend + add column default xxx",
 		// ref: https://github.com/pingcap/ticdc/issues/3929