Merge pull request #1 from instructure/changes-from-masteryconnect-go-mysql-fork

Still-relevant changes from MasteryConnect/go-mysql fork

Author: adamcrown

canal/canal.go

@@ -307,6 +307,29 @@ func (c *Canal) checkTableMatch(key string) bool {
 return matchFlag
 }
 
+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)
 // if table is excluded, return error and skip parsing event or dump

canal/sync.go

@@ -244,10 +244,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 err
 }

schema/schema.go

@@ -36,6 +36,8 @@ const (
 TYPE_MEDIUM_INT
 TYPE_BINARY // binary, varbinary
 TYPE_POINT // coordinates
+TYPE_TEXT // text (base64 encoded)
+TYPE_BLOB // blob (base64 encoded)
 )
 
 type TableColumn struct {
@@ -123,6 +125,10 @@ func (ta *Table) AddColumn(name string, columnType string, collation string, ext
 ta.Columns[index].Type = TYPE_BIT
 } else if strings.HasPrefix(columnType, "json") {
 ta.Columns[index].Type = TYPE_JSON
+} else if strings.HasPrefix(columnType, "text") {
+ta.Columns[index].Type = TYPE_TEXT
+} else if strings.HasPrefix(columnType, "blob") {
+ta.Columns[index].Type = TYPE_BLOB
 } else if strings.Contains(columnType, "point") {
 ta.Columns[index].Type = TYPE_POINT
 } else if strings.Contains(columnType, "mediumint") {