add OnRowsQueryEvent to EventHandler (#841)

Co-authored-by: Andrii Sudiev <[email protected]>

Author: sudevva

canal/handler.go

@@ -17,6 +17,10 @@ type EventHandler interface {
 	OnGTID(header *replication.EventHeader, gtidEvent mysql.BinlogGTIDEvent) error
 	// OnPosSynced Use your own way to sync position. When force is true, sync position immediately.
 	OnPosSynced(header *replication.EventHeader, pos mysql.Position, set mysql.GTIDSet, force bool) error
+	// OnRowsQueryEvent is called when binlog_rows_query_log_events=ON for each DML query.
+	// You'll get the original executed query, with comments if present.
+	// It will be called before OnRow.
+	OnRowsQueryEvent(e *replication.RowsQueryEvent) error
 	String() string
 }
 
@@ -40,6 +44,9 @@ func (h *DummyEventHandler) OnGTID(*replication.EventHeader, mysql.BinlogGTIDEve
 func (h *DummyEventHandler) OnPosSynced(*replication.EventHeader, mysql.Position, mysql.GTIDSet, bool) error {
 	return nil
 }
+func (h *DummyEventHandler) OnRowsQueryEvent(*replication.RowsQueryEvent) error {
+	return nil
+}
 
 func (h *DummyEventHandler) String() string { return "DummyEventHandler" }
 

canal/sync.go

@@ -128,6 +128,10 @@ func (c *Canal) runSyncBinlog() error {
 			if err := c.eventHandler.OnGTID(ev.Header, e); err != nil {
 				return errors.Trace(err)
 			}
+		case *replication.RowsQueryEvent:
+			if err := c.eventHandler.OnRowsQueryEvent(e); err != nil {
+				return errors.Trace(err)
+			}
 		case *replication.QueryEvent:
 			stmts, _, err := c.parser.Parse(string(e.Query), "", "")
 			if err != nil {