(rowsEvent *replication.RowsEvent, dml int8)
| 1260 | } |
| 1261 | |
| 1262 | func (b *BinlogReader) skipRowEvent(rowsEvent *replication.RowsEvent, dml int8) (bool, *common.TableContext) { |
| 1263 | tableOrigin := string(rowsEvent.Table.Table) |
| 1264 | tableLower := strings.ToLower(tableOrigin) |
| 1265 | switch strings.ToLower(string(rowsEvent.Table.Schema)) { |
| 1266 | case g.DtleSchemaName: |
| 1267 | if strings.HasPrefix(strings.ToLower(string(rowsEvent.Table.Table)), g.GtidExecutedTablePrefix) { |
| 1268 | // cases: 1. delete for compaction; 2. insert for compaction (gtid interval); 3. normal insert for tx (single gtid) |
| 1269 | // We make no special treat for case 2. That tx has only one insert, which should be ignored. |
| 1270 | if dml == common.InsertDML { |
| 1271 | if len(rowsEvent.Rows) == 1 { |
| 1272 | sidValue := rowsEvent.Rows[0][1] |
| 1273 | sidByte, ok := sidValue.(string) |
| 1274 | gnoI := rowsEvent.Rows[0][2] |
| 1275 | gno, okGNO := gnoI.(int64) |
| 1276 | |
| 1277 | // TODO It will go error if there is zero byte in the UUID. |
| 1278 | if !ok || !okGNO { |
| 1279 | b.logger.Error("cycle-prevention: unrecognized gtid_executed table sid or gno type", |
| 1280 | "type", hclog.Fmt("%T %T", sidValue, gnoI)) |
| 1281 | } else { |
| 1282 | var sid uuid.UUID // will be initialized to 0 |
| 1283 | copy(sid[:], sidByte) // len(sidByte) might be less than 16 #1034 |
| 1284 | coordinate := b.entryContext.Entry.Coordinates.(*common.MySQLCoordinateTx) |
| 1285 | coordinate.SID = sid |
| 1286 | coordinate.GNO = gno |
| 1287 | } |
| 1288 | } |
| 1289 | // If OSID is target mysql SID, skip applying the binlogEntry. |
| 1290 | // - Plan B: skip sending at applier: unnecessary sending |
| 1291 | // - Plan A: skip sending at extractor: currently extractor does not know target mysql SID |
| 1292 | } |
| 1293 | } |
| 1294 | return true, nil |
| 1295 | case "mysql": |
| 1296 | if b.mysqlContext.ExpandSyntaxSupport { |
| 1297 | return skipMysqlSchemaEvent(tableLower), nil |
| 1298 | } else { |
| 1299 | return true, nil |
| 1300 | } |
| 1301 | case "sys", "information_schema", "performance_schema": |
| 1302 | return true, nil |
| 1303 | default: |
| 1304 | if schemaContext, ok := b.tables[string(rowsEvent.Table.Schema)]; ok { |
| 1305 | if tableCtx, ok := schemaContext.TableMap[tableOrigin]; ok { |
| 1306 | return false, tableCtx |
| 1307 | } |
| 1308 | } |
| 1309 | // TODO when will schemaName be empty? |
| 1310 | if schemaContext, ok := b.tables[""]; ok { |
| 1311 | if tableCtx, ok := schemaContext.TableMap[tableOrigin]; ok { |
| 1312 | return false, tableCtx |
| 1313 | } |
| 1314 | } |
| 1315 | return true, nil |
| 1316 | } |
| 1317 | } |
| 1318 | |
| 1319 | func (b *BinlogReader) matchTable(patternTBS []*common.DataSource, schemaName string, tableName string) bool { |
no test coverage detected