change OnDDL to OnTableChanged (go-mysql-org#244)

Author: siddontang

canal/handler.go

@@ -7,6 +7,10 @@ import (
 
 type EventHandler interface {
 	OnRotate(roateEvent *replication.RotateEvent) error
+	// OnTableChanged is called when the table is created, altered, renamed or dropped.
+	// You need to clear the associated data like cache with the table.
+	// It will be called before OnDDL.
+	OnTableChanged(schema string, table string) error
 	OnDDL(nextPos mysql.Position, queryEvent *replication.QueryEvent) error
 	OnRow(e *RowsEvent) error
 	OnXID(nextPos mysql.Position) error
@@ -19,8 +23,9 @@ type EventHandler interface {
 type DummyEventHandler struct {
 }
 
-func (h *DummyEventHandler) OnRotate(*replication.RotateEvent) error { return nil }
-func (h *DummyEventHandler) OnDDL(mysql.Position, *replication.QueryEvent) error {
+func (h *DummyEventHandler) OnRotate(*replication.RotateEvent) error          { return nil }
+func (h *DummyEventHandler) OnTableChanged(schema string, table string) error { return nil }
+func (h *DummyEventHandler) OnDDL(nextPos mysql.Position, queryEvent *replication.QueryEvent) error {
 	return nil
 }
 func (h *DummyEventHandler) OnRow(*RowsEvent) error                 { return nil }

canal/sync.go

@@ -14,7 +14,7 @@ import (
 )
 
 var (
-	expCreateTable  = regexp.MustCompile("(?i)^CREATE\\sTABLE(\\sIF\\sNOT\\sEXISTS)?\\s`{0,1}(.*?)`{0,1}\\.{0,1}`{0,1}([^`\\.]+?)`{0,1}\\s.*")
+	expCreateTable = regexp.MustCompile("(?i)^CREATE\\sTABLE(\\sIF\\sNOT\\sEXISTS)?\\s`{0,1}(.*?)`{0,1}\\.{0,1}`{0,1}([^`\\.]+?)`{0,1}\\s.*")
 	expAlterTable  = regexp.MustCompile("(?i)^ALTER\\sTABLE\\s.*?`{0,1}(.*?)`{0,1}\\.{0,1}`{0,1}([^`\\.]+?)`{0,1}\\s.*")
 	expRenameTable = regexp.MustCompile("(?i)^RENAME\\sTABLE\\s.*?`{0,1}(.*?)`{0,1}\\.{0,1}`{0,1}([^`\\.]+?)`{0,1}\\s{1,}TO\\s.*?")
 	expDropTable   = regexp.MustCompile("(?i)^DROP\\sTABLE(\\sIF\\sEXISTS){0,1}\\s`{0,1}(.*?)`{0,1}\\.{0,1}`{0,1}([^`\\.]+?)`{0,1}(?:$|\\s)")
@@ -115,9 +115,9 @@ func (c *Canal) runSyncBinlog() error {
 			}
 		case *replication.QueryEvent:
 			var (
-				mb [][]byte
+				mb     [][]byte
 				schema []byte
-				table []byte
+				table  []byte
 			)
 			regexps := []regexp.Regexp{*expCreateTable, *expAlterTable, *expRenameTable, *expDropTable}
 			for _, reg := range regexps {
@@ -143,10 +143,14 @@ func (c *Canal) runSyncBinlog() error {
 			force = true
 			c.ClearTableCache(schema, table)
 			log.Infof("table structure changed, clear table cache: %s.%s\n", schema, table)
-			if err = c.eventHandler.OnDDL(pos, e); err != nil {
+			if err = c.eventHandler.OnTableChanged(string(schema), string(table)); err != nil {
 				return errors.Trace(err)
 			}
 
+			// Now we only handle Table Changed DDL, maybe we will support more later.
+			if err = c.eventHandler.OnDDL(pos, e); err != nil {
+				return errors.Trace(err)
+			}
 		default:
 			continue
 		}

replication/event.go

@@ -453,7 +453,7 @@ func (e *MariadbGTIDEvent) Decode(data []byte) error {
 }
 
 func (e *MariadbGTIDEvent) Dump(w io.Writer) {
-	fmt.Fprintf(w, "GTID: %s\n", e.GTID)
+	fmt.Fprintf(w, "GTID: %v\n", e.GTID)
 	fmt.Fprintln(w)
 }