func (l *LogMinerStream) HandlerRecords(records chan *LogMinerRecord) { for !e.shutdown { select { case r := <-records: switch r.Operation { case OperationCodeStart: l.txCache.startTx(r.TxId(), r.SCN) case OperationCodeCommit: l.txCache.commitTx(r.TxId(), r.SCN) case Operati
(row *LogMinerRecord)
| 896 | //} |
| 897 | |
| 898 | func (e *ExtractorOracle) parseToDataEvent(row *LogMinerRecord) (common.DataEvent, error) { |
| 899 | dataEvent := common.DataEvent{} |
| 900 | // parse ddl and dml |
| 901 | if row.Operation == OperationCodeDDL { |
| 902 | // 字符集问题 |
| 903 | dataEvent, err := e.parseDDLSQL(row.SQLRedo, row.SegOwner) |
| 904 | if err != nil { |
| 905 | return common.DataEvent{}, err |
| 906 | } |
| 907 | return dataEvent, nil |
| 908 | } else if row.Operation == OperationCodeDelete || row.Operation == OperationCodeUpdate || |
| 909 | row.Operation == OperationCodeInsert { |
| 910 | dataEvent, err := e.parseDMLSQL(row.SQLRedo, row.SQLUndo) |
| 911 | if err != nil { |
| 912 | return common.DataEvent{}, err |
| 913 | } |
| 914 | return dataEvent, nil |
| 915 | } |
| 916 | return dataEvent, fmt.Errorf("parese dateEvent fail , operation Code %v", row.Operation) |
| 917 | } |
| 918 | func (e *ExtractorOracle) parseDMLSQL(oracleRedoSQL, oracleUndoSQL string) (dataEvent common.DataEvent, err error) { |
| 919 | // Convert oracle SQL to MySQL format that can be parsed by tidb parese |
| 920 | OracleToMySQLFormat := func(oracleSQL string) string { |
no test coverage detected