use tableInfo in rowChangeEvent to gennerate dmls

cdc/sink/dmlsink/txn/mysql/mysql.go

@@ -567,8 +567,15 @@ func (s *mysqlBackend) prepareDMLs() *preparedDMLs {
 			}
 			// only use batch dml when the table has a handle key
 			if hasHandleKey(tableColumns) {
-				// TODO(dongmen): find a better way to get table info.
-				tableInfo := model.BuildTiDBTableInfo(tableColumns, firstRow.IndexColumns)
+				var tableInfo *timodel.TableInfo
+				if firstRow.TableInfo != nil && firstRow.TableInfo.TableInfo != nil {
+					tableInfo = firstRow.TableInfo.TableInfo
+				} else {
+					// firstRow.TableInfo is nil only when the row change event was generated by
+					// mounter, which only occurs when mysql sink was used in redo consumer or
+					// in the other consumers.
+					tableInfo = model.BuildTiDBTableInfo(tableColumns, firstRow.IndexColumns)
+				}
 				sql, value := s.batchSingleTxnDmls(event, tableInfo, translateToInsert)
 				sqls = append(sqls, sql...)
 				values = append(values, value...)