Merge pull request #482 from huangjunwen/master

Author: IANTHEREAL

replication/binlogsyncer.go

@@ -148,6 +148,7 @@ func NewBinlogSyncer(cfg BinlogSyncerConfig) *BinlogSyncer {
 
 	b.cfg = cfg
 	b.parser = NewBinlogParser()
+	b.parser.SetFlavor(cfg.Flavor)
 	b.parser.SetRawMode(b.cfg.RawModeEnabled)
 	b.parser.SetParseTime(b.cfg.ParseTime)
 	b.parser.SetTimestampStringLocation(b.cfg.TimestampStringLocation)

replication/parser.go

@@ -19,6 +19,9 @@ var (
 )
 
 type BinlogParser struct {
+	// "mysql" or "mariadb", if not set, use "mysql" by default
+	flavor string
+
 	format *FormatDescriptionEvent
 
 	tables map[uint64]*TableMapEvent
@@ -200,6 +203,10 @@ func (p *BinlogParser) SetVerifyChecksum(verify bool) {
 	p.verifyChecksum = verify
 }
 
+func (p *BinlogParser) SetFlavor(flavor string) {
+	p.flavor = flavor
+}
+
 func (p *BinlogParser) parseHeader(data []byte) (*EventHeader, error) {
 	h := new(EventHeader)
 	err := h.Decode(data)
@@ -234,7 +241,9 @@ func (p *BinlogParser) parseEvent(h *EventHeader, data []byte, rawData []byte) (
 			case XID_EVENT:
 				e = &XIDEvent{}
 			case TABLE_MAP_EVENT:
-				te := &TableMapEvent{}
+				te := &TableMapEvent{
+					flavor: p.flavor,
+				}
 				if p.format.EventTypeHeaderLengths[TABLE_MAP_EVENT-1] == 6 {
 					te.tableIDSize = 4
 				} else {

replication/row_event.go

@@ -19,6 +19,7 @@ import (
 var errMissingTableMapEvent = errors.New("invalid table id, no corresponding table map event")
 
 type TableMapEvent struct {
+	flavor      string
 	tableIDSize int
 
 	TableID uint64
@@ -423,6 +424,116 @@ func (e *TableMapEvent) Dump(w io.Writer) {
 	fmt.Fprintf(w, "Enum/set default charset: %v\n", e.EnumSetDefaultCharset)
 	fmt.Fprintf(w, "Enum/set column charset: %v\n", e.EnumSetColumnCharset)
 
+	unsignedMap := e.UnsignedMap()
+	fmt.Fprintf(w, "UnsignedMap: %#v\n", unsignedMap)
+
+	collationMap := e.CollationMap()
+	fmt.Fprintf(w, "CollationMap: %#v\n", collationMap)
+
+	enumSetCollationMap := e.EnumSetCollationMap()
+	fmt.Fprintf(w, "EnumSetCollationMap: %#v\n", enumSetCollationMap)
+
+	enumStrValueMap := e.EnumStrValueMap()
+	fmt.Fprintf(w, "EnumStrValueMap: %#v\n", enumStrValueMap)
+
+	setStrValueMap := e.SetStrValueMap()
+	fmt.Fprintf(w, "SetStrValueMap: %#v\n", setStrValueMap)
+
+	geometryTypeMap := e.GeometryTypeMap()
+	fmt.Fprintf(w, "GeometryTypeMap: %#v\n", geometryTypeMap)
+
+	nameMaxLen := 0
+	for _, name := range e.ColumnName {
+		if len(name) > nameMaxLen {
+			nameMaxLen = len(name)
+		}
+	}
+	nameFmt := "  %s"
+	if nameMaxLen > 0 {
+		nameFmt = fmt.Sprintf("  %%-%ds", nameMaxLen)
+	}
+
+	primaryKey := map[int]struct{}{}
+	for _, pk := range e.PrimaryKey {
+		primaryKey[int(pk)] = struct{}{}
+	}
+
+	fmt.Fprintf(w, "Columns: \n")
+	for i := 0; i < int(e.ColumnCount); i++ {
+		if len(e.ColumnName) == 0 {
+			fmt.Fprintf(w, nameFmt, "<n/a>")
+		} else {
+			fmt.Fprintf(w, nameFmt, e.ColumnName[i])
+		}
+
+		fmt.Fprintf(w, "  type=%-3d", e.realType(i))
+
+		if e.IsNumericColumn(i) {
+			if len(unsignedMap) == 0 {
+				fmt.Fprintf(w, "  unsigned=<n/a>")
+			} else if unsignedMap[i] {
+				fmt.Fprintf(w, "  unsigned=yes")
+			} else {
+				fmt.Fprintf(w, "  unsigned=no ")
+			}
+		}
+		if e.IsCharacterColumn(i) {
+			if len(collationMap) == 0 {
+				fmt.Fprintf(w, "  collation=<n/a>")
+			} else {
+				fmt.Fprintf(w, "  collation=%d ", collationMap[i])
+			}
+		}
+		if e.IsEnumColumn(i) {
+			if len(enumSetCollationMap) == 0 {
+				fmt.Fprintf(w, "  enum_collation=<n/a>")
+			} else {
+				fmt.Fprintf(w, "  enum_collation=%d", enumSetCollationMap[i])
+			}
+
+			if len(enumStrValueMap) == 0 {
+				fmt.Fprintf(w, "  enum=<n/a>")
+			} else {
+				fmt.Fprintf(w, "  enum=%v", enumStrValueMap[i])
+			}
+		}
+		if e.IsSetColumn(i) {
+			if len(enumSetCollationMap) == 0 {
+				fmt.Fprintf(w, "  set_collation=<n/a>")
+			} else {
+				fmt.Fprintf(w, "  set_collation=%d", enumSetCollationMap[i])
+			}
+
+			if len(setStrValueMap) == 0 {
+				fmt.Fprintf(w, "  set=<n/a>")
+			} else {
+				fmt.Fprintf(w, "  set=%v", setStrValueMap[i])
+			}
+		}
+		if e.IsGeometryColumn(i) {
+			if len(geometryTypeMap) == 0 {
+				fmt.Fprintf(w, "  geometry_type=<n/a>")
+			} else {
+				fmt.Fprintf(w, "  geometry_type=%v", geometryTypeMap[i])
+			}
+		}
+
+		available, nullable := e.Nullable(i)
+		if !available {
+			fmt.Fprintf(w, "  null=<n/a>")
+		} else if nullable {
+			fmt.Fprintf(w, "  null=yes")
+		} else {
+			fmt.Fprintf(w, "  null=no ")
+		}
+
+		if _, ok := primaryKey[i]; ok {
+			fmt.Fprintf(w, "  pri")
+		}
+
+		fmt.Fprintf(w, "\n")
+	}
+
 	fmt.Fprintln(w)
 }
 
@@ -492,6 +603,219 @@ func (e *TableMapEvent) bytesSlice2StrSlice(src [][]byte) []string {
 	return ret
 }
 
+// UnsignedMap returns a map: column index -> unsigned.
+// Note that only numeric columns will be returned.
+// nil is returned if not available or no numeric columns at all.
+func (e *TableMapEvent) UnsignedMap() map[int]bool {
+	if len(e.SignednessBitmap) == 0 {
+		return nil
+	}
+	p := 0
+	ret := make(map[int]bool)
+	for i := 0; i < int(e.ColumnCount); i++ {
+		if !e.IsNumericColumn(i) {
+			continue
+		}
+		ret[i] = e.SignednessBitmap[p/8]&(1<<uint(7-p%8)) != 0
+		p++
+	}
+	return ret
+}
+
+// CollationMap returns a map: column index -> collation id.
+// Note that only character columns will be returned.
+// nil is returned if not available or no character columns at all.
+func (e *TableMapEvent) CollationMap() map[int]uint64 {
+	return e.collationMap(e.IsCharacterColumn, e.DefaultCharset, e.ColumnCharset)
+}
+
+// EnumSetCollationMap returns a map: column index -> collation id.
+// Note that only enum or set columns will be returned.
+// nil is returned if not available or no enum/set columns at all.
+func (e *TableMapEvent) EnumSetCollationMap() map[int]uint64 {
+	return e.collationMap(e.IsEnumOrSetColumn, e.EnumSetDefaultCharset, e.EnumSetColumnCharset)
+}
+
+func (e *TableMapEvent) collationMap(includeType func(int) bool, defaultCharset, columnCharset []uint64) map[int]uint64 {
+
+	if len(defaultCharset) != 0 {
+		defaultCollation := defaultCharset[0]
+
+		// character column index -> collation
+		collations := make(map[int]uint64)
+		for i := 1; i < len(defaultCharset); i += 2 {
+			collations[int(defaultCharset[i])] = defaultCharset[i+1]
+		}
+
+		p := 0
+		ret := make(map[int]uint64)
+		for i := 0; i < int(e.ColumnCount); i++ {
+			if !includeType(i) {
+				continue
+			}
+
+			if collation, ok := collations[p]; ok {
+				ret[i] = collation
+			} else {
+				ret[i] = defaultCollation
+			}
+			p++
+		}
+
+		return ret
+	}
+
+	if len(columnCharset) != 0 {
+
+		p := 0
+		ret := make(map[int]uint64)
+		for i := 0; i < int(e.ColumnCount); i++ {
+			if !includeType(i) {
+				continue
+			}
+
+			ret[i] = columnCharset[p]
+			p++
+		}
+
+		return ret
+	}
+
+	return nil
+}
+
+// EnumStrValueMap returns a map: column index -> enum string value.
+// Note that only enum columns will be returned.
+// nil is returned if not available or no enum columns at all.
+func (e *TableMapEvent) EnumStrValueMap() map[int][]string {
+	return e.strValueMap(e.IsEnumColumn, e.EnumStrValueString())
+}
+
+// SetStrValueMap returns a map: column index -> set string value.
+// Note that only set columns will be returned.
+// nil is returned if not available or no set columns at all.
+func (e *TableMapEvent) SetStrValueMap() map[int][]string {
+	return e.strValueMap(e.IsSetColumn, e.SetStrValueString())
+}
+
+func (e *TableMapEvent) strValueMap(includeType func(int) bool, strValue [][]string) map[int][]string {
+	if len(strValue) == 0 {
+		return nil
+	}
+	p := 0
+	ret := make(map[int][]string)
+	for i := 0; i < int(e.ColumnCount); i++ {
+		if !includeType(i) {
+			continue
+		}
+		ret[i] = strValue[p]
+		p++
+	}
+	return ret
+}
+
+// GeometryTypeMap returns a map: column index -> geometry type.
+// Note that only geometry columns will be returned.
+// nil is returned if not available or no geometry columns at all.
+func (e *TableMapEvent) GeometryTypeMap() map[int]uint64 {
+	if len(e.GeometryType) == 0 {
+		return nil
+	}
+	p := 0
+	ret := make(map[int]uint64)
+	for i := 0; i < int(e.ColumnCount); i++ {
+		if !e.IsGeometryColumn(i) {
+			continue
+		}
+
+		ret[i] = e.GeometryType[p]
+		p++
+	}
+	return ret
+}
+
+// Below realType and IsXXXColumn are base from:
+//   table_def::type in sql/rpl_utility.h
+//   Table_map_log_event::print_columns in mysql-8.0/sql/log_event.cc and mariadb-10.5/sql/log_event_client.cc
+
+func (e *TableMapEvent) realType(i int) byte {
+
+	typ := e.ColumnType[i]
+
+	switch typ {
+	case MYSQL_TYPE_STRING:
+		rtyp := byte(e.ColumnMeta[i] >> 8)
+		if rtyp == MYSQL_TYPE_ENUM || rtyp == MYSQL_TYPE_SET {
+			return rtyp
+		}
+
+	case MYSQL_TYPE_DATE:
+		return MYSQL_TYPE_NEWDATE
+	}
+
+	return typ
+}
+
+func (e *TableMapEvent) IsNumericColumn(i int) bool {
+
+	switch e.realType(i) {
+	case MYSQL_TYPE_TINY,
+		MYSQL_TYPE_SHORT,
+		MYSQL_TYPE_INT24,
+		MYSQL_TYPE_LONG,
+		MYSQL_TYPE_LONGLONG,
+		MYSQL_TYPE_NEWDECIMAL,
+		MYSQL_TYPE_FLOAT,
+		MYSQL_TYPE_DOUBLE:
+		return true
+
+	default:
+		return false
+	}
+
+}
+
+// IsCharacterColumn returns true if the column type is considered as character type.
+// Note that JSON/GEOMETRY types are treated as character type in mariadb.
+// (JSON is an alias for LONGTEXT in mariadb: https://mariadb.com/kb/en/json-data-type/)
+func (e *TableMapEvent) IsCharacterColumn(i int) bool {
+
+	switch e.realType(i) {
+	case MYSQL_TYPE_STRING,
+		MYSQL_TYPE_VAR_STRING,
+		MYSQL_TYPE_VARCHAR,
+		MYSQL_TYPE_BLOB:
+		return true
+
+	case MYSQL_TYPE_GEOMETRY:
+		if e.flavor == "mariadb" {
+			return true
+		}
+		return false
+
+	default:
+		return false
+	}
+
+}
+
+func (e *TableMapEvent) IsEnumColumn(i int) bool {
+	return e.realType(i) == MYSQL_TYPE_ENUM
+}
+
+func (e *TableMapEvent) IsSetColumn(i int) bool {
+	return e.realType(i) == MYSQL_TYPE_SET
+}
+
+func (e *TableMapEvent) IsGeometryColumn(i int) bool {
+	return e.realType(i) == MYSQL_TYPE_GEOMETRY
+}
+
+func (e *TableMapEvent) IsEnumOrSetColumn(i int) bool {
+	rtyp := e.realType(i)
+	return rtyp == MYSQL_TYPE_ENUM || rtyp == MYSQL_TYPE_SET
+}
+
 // RowsEventStmtEndFlag is set in the end of the statement.
 const RowsEventStmtEndFlag = 0x01