Repository: pingcap/tidb
File: pkg/executor/insert.go (Go)

Code before:
			return err
		}

		c := assign.Col.ToInfo()
		idx := assign.Col.Index
		c.Name = assign.ColName
		val, err = table.CastValue(sctx, val, c, false, false)
		if err != nil {
			return err
		}

		_ = errorHandler(sctx, assign, &val, nil)
		e.evalBuffer4Dup.SetDatum(idx, val)
		if needActiveActiveSyncStats {
			if !e.row4Update[assign.Col.Index].Equals(val) {
				realOverwrite = true
			}
		}
		e.row4Update[assign.Col.Index] = val
		assignFlag[assign.Col.Index] = true
	}

	if needActiveActiveSyncStats && len(nonGenerated) > 0 && !realOverwrite {
		// ActiveActiveSyncStats info is required, but no overwrite really happen, it means
		// that all conflicts are skipped.
		e.Ctx().GetSessionVars().ActiveActiveConflictSkipRows.Add(1)
	}

	newData := e.row4Update[:len(e.Table.WritableCols())]
	if updateOriginTSCol >= 0 && !assignFlag[updateOriginTSCol] {
		newData[updateOriginTSCol].SetNull()
	}

	_, ignored, err := updateRecord(
		ctx, e.Ctx(),
		handle, oldRow, newData,
		0, generated, e.evalBuffer4Dup, errorHandler,
		assignFlag, e.Table,
		true, e.memTracker, e.fkChecks, e.fkCascades, dupKeyMode, e.ignoreErr)

	if ignored {
		return nil
	}

	if err != nil {
		return errors.Trace(err)
	}

	if autoColIdx >= 0 {
		if e.Ctx().GetSessionVars().StmtCtx.AffectedRows() > 0 {
			// If "INSERT ... ON DUPLICATE KEY UPDATE" duplicate and update a row,

Review comment (refactor):
Consider the below case:

Origin rows (id, value, _tidb_origin_ts, _tidb_commit_ts):
```
[(1, "val1", NULL, 100), (2, "val2", NULL, 400), (3, "val3", 200, 300)]
```

If CDC syncs: (id, value, _tidb_origin_ts):

```
// 1 should be overrite, 2 should be skipped because conflict, 3 should be skipped for duplicate sync
[(1, "val12", 200), (2, "val22", 200), (3, "val3", 200)]. 
```

Only `2` should be seen as a conflict skip.






Code after:
		if assign.LazyErr != nil {
			return assign.LazyErr
		}
		val, err := assign.Expr.Eval(evalCtx, e.evalBuffer4Dup.ToRow())
		if err != nil {
			return err
		}

		c := assign.Col.ToInfo()
		idx := assign.Col.Index
		c.Name = assign.ColName
		val, err = table.CastValue(sctx, val, c, false, false)
		if err != nil {
			return err
		}

		_ = errorHandler(sctx, assign, &val, nil)
		e.evalBuffer4Dup.SetDatum(idx, val)
		if needActiveActiveSyncStats {
			if !e.row4Update[assign.Col.Index].Equals(val) {
				realOverwrite = true
			}
			if assign.ColName.Equals(&model.ExtraOriginTSName) {
				if newRow[updateOriginTSCol].Equals(val) {
					originTSEqual = true
				}
			}
		}
		e.row4Update[assign.Col.Index] = val
		assignFlag[assign.Col.Index] = true
	}

	// ActiveActiveSyncStats info is required, but no overwrite really happen, it means
	// that all conflicts are skipped.
	if needActiveActiveSyncStats && len(nonGenerated) > 0 && !realOverwrite {
		// _tidb_origin_ts equal is a corner case, imagine that CDC sync duplicate rows, with same _tidb_origin_ts
		// In this case, the skipped row is not caused by business logic, but CDC, we should not give misleading information.
		if !originTSEqual {
			e.Ctx().GetSessionVars().ActiveActiveConflictSkipRows.Add(1)
			redactMode := e.Ctx().GetSessionVars().EnableRedactLog
			oldRowOriginTS, _ := oldRow[updateOriginTSCol].ToString()
			newRowOriginTS, _ := newRow[updateOriginTSCol].ToString()
			logutil.BgLogger().Info("[CDC active-active] skip writing conflict row.",
				zap.Int64("tableID", e.Table.Meta().ID),
				zap.Stringer("table", e.Table.Meta().Name),
				zap.Stringer("handle", redact.Stringer(redactMode, handle)),
				zap.Uint64("old row commit ts", oldRowCommitTS),
				zap.String("old row origin ts", oldRowOriginTS),
				zap.String("new row origin ts", newRowOriginTS))
		}
	}