Add more MySQL-8.0 meta data to GTIDEvent and TableMapEvent

mysql/util.go

@@ -259,6 +259,37 @@ func Uint64ToBytes(n uint64) []byte {
 	}
 }
 
+func BytesToUint64(b []byte) (u uint64) {
+	switch len(b) {
+	case 8:
+		u |= (uint64(b[7]) << 56)
+		fallthrough
+	case 7:
+		u |= (uint64(b[6]) << 48)
+		fallthrough
+	case 6:
+		u |= (uint64(b[5]) << 40)
+		fallthrough
+	case 5:
+		u |= (uint64(b[4]) << 32)
+		fallthrough
+	case 4:
+		u |= (uint64(b[3]) << 24)
+		fallthrough
+	case 3:
+		u |= (uint64(b[2]) << 16)
+		fallthrough
+	case 2:
+		u |= (uint64(b[1]) << 8)
+		fallthrough
+	case 1:
+		u |= uint64(b[0])
+	default:
+		panic(fmt.Errorf("BytesToUint64 byte slice length must be in range [1, 8]"))
+	}
+	return
+}
+
 func FormatBinaryDate(n int, data []byte) ([]byte, error) {
 	switch n {
 	case 0:

replication/const.go

@@ -188,7 +188,7 @@ func (e EventType) String() string {
 		return "ViewChangeEvent"
 	case XA_PREPARE_LOG_EVENT:
 		return "XAPrepareLogEvent"
-		
+
 	default:
 		return "UnknownEvent"
 	}
@@ -202,3 +202,18 @@ const (
 	BINLOG_CHECKSUM_ALG_UNDEF byte = 255 // special value to tag undetermined yet checksum
 	// or events from checksum-unaware servers
 )
+
+// These are TABLE_MAP_EVENT's optional metadata field type, from: libbinlogevents/include/rows_event.h
+const (
+	TABLE_MAP_OPT_META_SIGNEDNESS byte = iota + 1
+	TABLE_MAP_OPT_META_DEFAULT_CHARSET
+	TABLE_MAP_OPT_META_COLUMN_CHARSET
+	TABLE_MAP_OPT_META_COLUMN_NAME
+	TABLE_MAP_OPT_META_SET_STR_VALUE
+	TABLE_MAP_OPT_META_ENUM_STR_VALUE
+	TABLE_MAP_OPT_META_GEOMETRY_TYPE
+	TABLE_MAP_OPT_META_SIMPLE_PRIMARY_KEY
+	TABLE_MAP_OPT_META_PRIMARY_KEY_WITH_PREFIX
+	TABLE_MAP_OPT_META_ENUM_AND_SET_DEFAULT_CHARSET
+	TABLE_MAP_OPT_META_ENUM_AND_SET_COLUMN_CHARSET
+)

replication/event.go

@@ -21,6 +21,7 @@ const (
 	LogicalTimestampTypeCode   = 2
 	PartLogicalTimestampLength = 8
 	BinlogChecksumLength       = 4
+	UndefinedServerVer         = 999999 // UNDEFINED_SERVER_VERSION
 )
 
 type BinlogEvent struct {
@@ -349,6 +350,18 @@ type GTIDEvent struct {
 	GNO            int64
 	LastCommitted  int64
 	SequenceNumber int64
+
+	// The followings are available only after MySQL-8.0
+
+	ImmediateCommitTimestamp uint64
+	OriginalCommitTimestamp  uint64
+
+	// Total transaction length (including this GTIDEvent), see:
+	// https://mysqlhighavailability.com/taking-advantage-of-new-transaction-length-metadata/
+	TransactionLength uint64
+
+	ImmediateServerVersion uint32
+	OriginalServerVersion  uint32
 }
 
 func (e *GTIDEvent) Decode(data []byte) error {
@@ -359,26 +372,99 @@ func (e *GTIDEvent) Decode(data []byte) error {
 	pos += SidLength
 	e.GNO = int64(binary.LittleEndian.Uint64(data[pos:]))
 	pos += 8
+
 	if len(data) >= 42 {
 		if uint8(data[pos]) == LogicalTimestampTypeCode {
 			pos++
 			e.LastCommitted = int64(binary.LittleEndian.Uint64(data[pos:]))
 			pos += PartLogicalTimestampLength
 			e.SequenceNumber = int64(binary.LittleEndian.Uint64(data[pos:]))
+			pos += 8
+
+			// IMMEDIATE_COMMIT_TIMESTAMP_LENGTH = 7
+			if len(data)-pos < 7 {
+				return nil
+			}
+			e.ImmediateCommitTimestamp = BytesToUint64(data[pos : pos+7])
+			pos += 7
+			if (e.ImmediateCommitTimestamp & (uint64(1) << 55)) != 0 {
+				// If the most significant bit set, another 7 byte follows representing OriginalCommitTimestamp
+				e.ImmediateCommitTimestamp &= ^(uint64(1) << 55)
+				e.OriginalCommitTimestamp = BytesToUint64(data[pos : pos+7])
+				pos += 7
+
+			} else {
+				// Otherwise OriginalCommitTimestamp == ImmediateCommitTimestamp
+				e.OriginalCommitTimestamp = e.ImmediateCommitTimestamp
+
+			}
+
+			// TRANSACTION_LENGTH_MIN_LENGTH = 1
+			if len(data)-pos < 1 {
+				return nil
+			}
+			var n int
+			e.TransactionLength, _, n = LengthEncodedInt(data[pos:])
+			pos += n
+
+			// IMMEDIATE_SERVER_VERSION_LENGTH = 4
+			e.ImmediateServerVersion = UndefinedServerVer
+			e.OriginalServerVersion = UndefinedServerVer
+			if len(data)-pos < 4 {
+				return nil
+			}
+			e.ImmediateServerVersion = binary.LittleEndian.Uint32(data[pos:])
+			pos += 4
+			if (e.ImmediateServerVersion & (uint32(1) << 31)) != 0 {
+				// If the most significant bit set, another 4 byte follows representing OriginalServerVersion
+				e.ImmediateServerVersion &= ^(uint32(1) << 31)
+				e.OriginalServerVersion = binary.LittleEndian.Uint32(data[pos:])
+				pos += 4
+
+			} else {
+				// Otherwise OriginalServerVersion == ImmediateServerVersion
+				e.OriginalServerVersion = e.ImmediateServerVersion
+
+			}
+
 		}
 	}
 	return nil
 }
 
 func (e *GTIDEvent) Dump(w io.Writer) {
+	fmtTime := func(t time.Time) string {
+		if t.IsZero() {
+			return "<n/a>"
+		}
+		return t.Format(time.RFC3339Nano)
+	}
+
 	fmt.Fprintf(w, "Commit flag: %d\n", e.CommitFlag)
 	u, _ := uuid.FromBytes(e.SID)
 	fmt.Fprintf(w, "GTID_NEXT: %s:%d\n", u.String(), e.GNO)
 	fmt.Fprintf(w, "LAST_COMMITTED: %d\n", e.LastCommitted)
 	fmt.Fprintf(w, "SEQUENCE_NUMBER: %d\n", e.SequenceNumber)
+	fmt.Fprintf(w, "Immediate commmit timestamp: %d (%s)\n", e.ImmediateCommitTimestamp, fmtTime(e.ImmediateCommitTime()))
+	fmt.Fprintf(w, "Orignal commmit timestamp: %d (%s)\n", e.OriginalCommitTimestamp, fmtTime(e.OriginalCommitTime()))
+	fmt.Fprintf(w, "Transaction length: %d\n", e.TransactionLength)
+	fmt.Fprintf(w, "Immediate server version: %d\n", e.ImmediateServerVersion)
+	fmt.Fprintf(w, "Orignal server version: %d\n", e.OriginalServerVersion)
 	fmt.Fprintln(w)
 }
 
+// ImmediateCommitTime returns the commit time of this trx on the immediate server
+// or zero time if not available.
+func (e *GTIDEvent) ImmediateCommitTime() time.Time {
+	return microSecTimestampToTime(e.ImmediateCommitTimestamp)
+}
+
+// OriginalCommitTime returns the commit time of this trx on the original server
+// or zero time if not available.
+func (e *GTIDEvent) OriginalCommitTime() time.Time {
+	return microSecTimestampToTime(e.OriginalCommitTimestamp)
+}
+
 type BeginLoadQueryEvent struct {
 	FileID    uint32
 	BlockData []byte