(ev *replication.BinlogEvent, entriesChannel chan<- *common.EntryContext)
| 504 | } |
| 505 | |
| 506 | func (b *BinlogReader) handleQueryEvent(ev *replication.BinlogEvent, |
| 507 | entriesChannel chan<- *common.EntryContext) error { |
| 508 | mysqlCoordinateTx := b.entryContext.Entry.Coordinates.(*common.MySQLCoordinateTx) |
| 509 | gno := mysqlCoordinateTx.GNO |
| 510 | evt := ev.Event.(*replication.QueryEvent) |
| 511 | query0 := string(evt.Query) |
| 512 | |
| 513 | queryEventFlags, err := common.ParseQueryEventFlags(evt.StatusVars, b.logger) |
| 514 | if err != nil { |
| 515 | return errors.Wrap(err, "ParseQueryEventFlags") |
| 516 | } |
| 517 | |
| 518 | if evt.ErrorCode != 0 { |
| 519 | b.logger.Error("DTLE_BUG: found query_event with error code, which is not handled.", |
| 520 | "ErrorCode", evt.ErrorCode, "query", query0, "gno", gno) |
| 521 | } |
| 522 | currentSchema := string(evt.Schema) |
| 523 | |
| 524 | b.logger.Debug("query event", "schema", currentSchema, "query", query0) |
| 525 | |
| 526 | if queryIsBegin(query0) { |
| 527 | b.hasBeginQuery = true |
| 528 | } else if queryIsCommit(query0) || !b.hasBeginQuery { |
| 529 | // not hasBeginQuery: a single-query transaction. |
| 530 | |
| 531 | var query8 string |
| 532 | var errConvertToUTF8 error |
| 533 | if g.IsUTF8OrMB4(queryEventFlags.CharacterSetClient) { |
| 534 | query8 = query0 |
| 535 | } else { |
| 536 | b.logger.Info("transcode a DDL to UTF8", "from", queryEventFlags.CharacterSetClient) |
| 537 | query8, errConvertToUTF8 = mysqlconfig.ConvertToUTF8(query0, queryEventFlags.CharacterSetClient) |
| 538 | } |
| 539 | |
| 540 | err := b.checkDtleQueryOSID(query8) |
| 541 | if err != nil { |
| 542 | return errors.Wrap(err, "checkDtleQueryOSID") |
| 543 | } |
| 544 | |
| 545 | queryInfo, err := b.resolveQuery(currentSchema, query8, b.skipQueryDDL) |
| 546 | if err != nil { |
| 547 | return errors.Wrap(err, "resolveQuery") |
| 548 | } |
| 549 | |
| 550 | skipSql := false |
| 551 | if queryInfo.isSkip || isSkipQuery(query8) { |
| 552 | // queries that should be skipped regardless of ExpandSyntaxSupport |
| 553 | skipSql = true |
| 554 | } else { |
| 555 | if !b.mysqlContext.ExpandSyntaxSupport { |
| 556 | skipSql = queryInfo.isExpand || isExpandSyntaxQuery(query8) |
| 557 | } |
| 558 | } |
| 559 | |
| 560 | currentSchemaRename := currentSchema |
| 561 | schema := b.findCurrentSchema(currentSchema) |
| 562 | if schema != nil && schema.TableSchemaRename != "" { |
| 563 | currentSchemaRename = schema.TableSchemaRename |
no test coverage detected