forked from vitessio/vitess
-
Notifications
You must be signed in to change notification settings - Fork 1
/
binlog_event.go
280 lines (238 loc) · 8.72 KB
/
binlog_event.go
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
package replication
import (
"fmt"
binlogdatapb "github.com/youtube/vitess/go/vt/proto/binlogdata"
)
// BinlogEvent represents a single event from a raw MySQL binlog dump stream.
// The implementation is provided by each supported flavor in go/vt/mysqlctl.
//
// binlog.Streamer receives these events through a mysqlctl.SlaveConnection and
// processes them, grouping statements into BinlogTransactions as appropriate.
//
// Methods that only access header fields can't fail as long as IsValid()
// returns true, so they have a single return value. Methods that might fail
// even when IsValid() is true return an error value.
//
// Methods that require information from the initial FORMAT_DESCRIPTION_EVENT
// will have a BinlogFormat parameter.
//
// A BinlogEvent should never be sent over the wire. UpdateStream service
// will send BinlogTransactions from these events.
type BinlogEvent interface {
// IsValid returns true if the underlying data buffer contains a valid
// event. This should be called first on any BinlogEvent, and other
// methods should only be called if this one returns true. This ensures
// you won't get panics due to bounds checking on the byte array.
IsValid() bool
// General protocol events.
// IsFormatDescription returns true if this is a
// FORMAT_DESCRIPTION_EVENT. Do not call StripChecksum before
// calling Format (Format returns the BinlogFormat anyway,
// required for calling StripChecksum).
IsFormatDescription() bool
// IsQuery returns true if this is a QUERY_EVENT, which encompasses
// all SQL statements.
IsQuery() bool
// IsXID returns true if this is an XID_EVENT, which is an alternate
// form of COMMIT.
IsXID() bool
// IsGTID returns true if this is a GTID_EVENT.
IsGTID() bool
// IsRotate returns true if this is a ROTATE_EVENT.
IsRotate() bool
// IsIntVar returns true if this is an INTVAR_EVENT.
IsIntVar() bool
// IsRand returns true if this is a RAND_EVENT.
IsRand() bool
// IsPreviousGTIDs returns true if this event is a PREVIOUS_GTIDS_EVENT.
IsPreviousGTIDs() bool
// RBR events.
// IsTableMapEvent returns true if this is a TABLE_MAP_EVENT.
IsTableMap() bool
// IsWriteRowsEvent returns true if this is a WRITE_ROWS_EVENT.
IsWriteRows() bool
// IsUpdateRowsEvent returns true if this is a UPDATE_ROWS_EVENT.
IsUpdateRows() bool
// IsDeleteRowsEvent returns true if this is a DELETE_ROWS_EVENT.
IsDeleteRows() bool
// Timestamp returns the timestamp from the event header.
Timestamp() uint32
// Format returns a BinlogFormat struct based on the event data.
// This is only valid if IsFormatDescription() returns true.
Format() (BinlogFormat, error)
// GTID returns the GTID from the event, and if this event
// also serves as a BEGIN statement.
// This is only valid if IsGTID() returns true.
GTID(BinlogFormat) (GTID, bool, error)
// Query returns a Query struct representing data from a QUERY_EVENT.
// This is only valid if IsQuery() returns true.
Query(BinlogFormat) (Query, error)
// IntVar returns the type and value of the variable for an INTVAR_EVENT.
// This is only valid if IsIntVar() returns true.
IntVar(BinlogFormat) (byte, uint64, error)
// Rand returns the two seed values for a RAND_EVENT.
// This is only valid if IsRand() returns true.
Rand(BinlogFormat) (uint64, uint64, error)
// PreviousGTIDs returns the Position from the event.
// This is only valid if IsPreviousGTIDs() returns true.
PreviousGTIDs(BinlogFormat) (Position, error)
// TableID returns the table ID for a TableMap, UpdateRows,
// WriteRows or DeleteRows event.
TableID(BinlogFormat) uint64
// TableMap returns a TableMap struct representing data from a
// TABLE_MAP_EVENT. This is only valid if IsTableMapEvent() returns
// true.
TableMap(BinlogFormat) (*TableMap, error)
// Rows returns a Rows struct representing data from a
// {WRITE,UPDATE,DELETE}_ROWS_EVENT. This is only valid if
// IsWriteRows(), IsUpdateRows(), or IsDeleteRows() returns
// true.
Rows(BinlogFormat, *TableMap) (Rows, error)
// StripChecksum returns the checksum and a modified event with the
// checksum stripped off, if any. If there is no checksum, it returns
// the same event and a nil checksum.
StripChecksum(BinlogFormat) (ev BinlogEvent, checksum []byte, err error)
}
// BinlogFormat contains relevant data from the FORMAT_DESCRIPTION_EVENT.
// This structure is passed to subsequent event types to let them know how to
// parse themselves.
type BinlogFormat struct {
// FormatVersion is the version number of the binlog file format.
// We only support version 4.
FormatVersion uint16
// ServerVersion is the name of the MySQL server version.
// It starts with something like 5.6.33-xxxx.
ServerVersion string
// HeaderLength is the size in bytes of event headers other
// than FORMAT_DESCRIPTION_EVENT. Almost always 19.
HeaderLength byte
// ChecksumAlgorithm is the ID number of the binlog checksum algorithm.
// See three possible values below.
ChecksumAlgorithm byte
// HeaderSizes is an array of sizes of the headers for each message.
HeaderSizes []byte
}
// IsZero returns true if the BinlogFormat has not been initialized.
func (f BinlogFormat) IsZero() bool {
return f.FormatVersion == 0 && f.HeaderLength == 0
}
// HeaderSize returns the header size of any event type.
func (f BinlogFormat) HeaderSize(typ byte) byte {
return f.HeaderSizes[typ-1]
}
// Query contains data from a QUERY_EVENT.
type Query struct {
Database string
Charset *binlogdatapb.Charset
SQL string
}
// String pretty-prints a Query.
func (q Query) String() string {
return fmt.Sprintf("{Database: %q, Charset: %v, SQL: %q}",
q.Database, q.Charset, q.SQL)
}
// TableMap contains data from a TABLE_MAP_EVENT.
type TableMap struct {
// Flags is the table's flags.
Flags uint16
// Database is the database name.
Database string
// Name is the name of the table.
Name string
// Types is an array of MySQL types for the fields.
Types []byte
// CanBeNull's bits are set if the column can be NULL.
CanBeNull Bitmap
// Metadata is an array of uint16, one per column.
// It contains a few extra information about each column,
// that is dependent on the type.
// - If the metadata is not present, this is zero.
// - If the metadata is one byte, only the lower 8 bits are used.
// - If the metadata is two bytes, all 16 bits are used.
Metadata []uint16
}
// Rows contains data from a {WRITE,UPDATE,DELETE}_ROWS_EVENT.
type Rows struct {
// Flags has the flags from the event.
Flags uint16
// IdentifyColumns describes which columns are included to
// identify the row. It is a bitmap indexed by the TableMap
// list of columns.
// Set for UPDATE and DELETE.
IdentifyColumns Bitmap
// DataColumns describes which columns are included. It is
// a bitmap indexed by the TableMap list of columns.
// Set for WRITE and UPDATE.
DataColumns Bitmap
// Rows is an array of Row in the event.
Rows []Row
}
// Row contains data for a single Row in a Rows event.
type Row struct {
// NullIdentifyColumns describes which of the identify columns are NULL.
// It is only set for UPDATE and DELETE events.
NullIdentifyColumns Bitmap
// NullColumns describes which of the present columns are NULL.
// It is only set for WRITE and UPDATE events.
NullColumns Bitmap
// Identify is the raw data for the columns used to identify a row.
// It is only set for UPDATE and DELETE events.
Identify []byte
// Data is the raw data.
// It is only set for WRITE and UPDATE events.
Data []byte
}
// Bitmap is used by the previous structures.
type Bitmap struct {
// data is the slice this is based on.
data []byte
// count is how many bits we have in the map.
count int
}
func newBitmap(data []byte, pos int, count int) (Bitmap, int) {
byteSize := (count + 7) / 8
return Bitmap{
data: data[pos : pos+byteSize],
count: count,
}, pos + byteSize
}
// NewServerBitmap returns a bitmap that can hold 'count' bits.
func NewServerBitmap(count int) Bitmap {
byteSize := (count + 7) / 8
return Bitmap{
data: make([]byte, byteSize),
count: count,
}
}
// Count returns the number of bits in this Bitmap.
func (b *Bitmap) Count() int {
return b.count
}
// Bit returned the value of a given bit in the Bitmap.
func (b *Bitmap) Bit(index int) bool {
byteIndex := index / 8
bitMask := byte(1 << (uint(index) & 0x7))
return b.data[byteIndex]&bitMask > 0
}
// Set sets the given boolean value.
func (b *Bitmap) Set(index int, value bool) {
byteIndex := index / 8
bitMask := byte(1 << (uint(index) & 0x7))
if value {
b.data[byteIndex] |= bitMask
} else {
b.data[byteIndex] &= 0xff - bitMask
}
}
// BitCount returns how many bits are set in the bitmap.
// Note values that are not used may be set to 0 or 1,
// hence the non-efficient logic.
func (b *Bitmap) BitCount() int {
sum := 0
for i := 0; i < b.count; i++ {
if b.Bit(i) {
sum++
}
}
return sum
}