Add basic column count mismatch to trigger table schema reload.

Author: Doug Weber

canal/canal.go

@@ -193,6 +193,29 @@ func (c *Canal) WaitDumpDone() <-chan struct{} {
 return c.dumpDoneCh
 }
 
+func (c *Canal) GetTableForEvent(te *replication.TableMapEvent) (*schema.Table, error) {
+key := fmt.Sprintf("%s.%s", te.Schema, te.Table)
+c.tableLock.Lock()
+t, ok := c.tables[key]
+c.tableLock.Unlock()
+
+// return if we already have the table info and the columns count matches
+if ok && (te == nil || te.ColumnCount == uint64(len(t.Columns))) {
+return t, nil
+}
+
+t, err := schema.NewTable(c, string(te.Schema), string(te.Table))
+if err != nil {
+return nil, errors.Trace(err)
+}
+
+c.tableLock.Lock()
+c.tables[key] = t
+c.tableLock.Unlock()
+
+return t, nil
+}
+
 func (c *Canal) GetTable(db string, table string) (*schema.Table, error) {
 key := fmt.Sprintf("%s.%s", db, table)
 c.tableLock.Lock()

canal/sync.go

@@ -79,10 +79,7 @@ func (c *Canal) handleRowsEvent(e *replication.BinlogEvent) error {
 ev := e.Event.(*replication.RowsEvent)
 
 // Caveat: table may be altered at runtime.
-schema := string(ev.Table.Schema)
-table := string(ev.Table.Table)
-
-t, err := c.GetTable(schema, table)
+t, err := c.GetTableForEvent(ev.Table)
 if err != nil {
 return errors.Trace(err)
 }