Skip to content

Commit 6a5d630

Browse files
pfcoperezilidemi
pfcoperez
and
ilidemi
authored
[DBI-855] MySQL connector: Use binlog event type and meta to determine BINARY(N) actual length and padding (#4445)
MySQL BINARY(N) values are right-padded, from https://dev.mysql.com/doc/refman/5.7/en/binary-varbinary.html > When BINARY values are stored, they are right-padded with the pad value to the specified length. The pad value is 0x00 (the zero byte). Values are right-padded with 0x00 for inserts, and no trailing bytes are removed for retrievals. All bytes are significant in comparisons, including ORDER BY and DISTINCT operations. 0x00 and space differ in comparisons, with 0x00 sorting before space. In MySQL binlog, these values are: - zero-right-trimmed: stored without the right side zeros. - with the length of the zero-right-trimmed binary. When querying the data, MySQL uses the record type and metadata to restore the original length but, when reading events from binlog directly, Closes: DBI-855 Related to: ClickHouse/support-escalation#8013 References: - Similar issue in different project: shyiko/mysql-binlog-connector-java#169 (comment) - https://dev.mysql.com/doc/refman/5.7/en/binary-varbinary.html --------- Co-authored-by: Ilia Demianenko <ilia.demianenko@clickhouse.com>
1 parent 85a2a97 commit 6a5d630

2 files changed

Lines changed: 233 additions & 1 deletion

File tree

‎flow/connectors/mysql/qvalue_convert.go‎

Lines changed: 35 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -338,6 +338,34 @@ func QValueFromMysqlFieldValue(qkind types.QValueKind, mytype byte, fv mysql.Fie
338338
}
339339
}
340340

341+
// binaryColumnLength returns the declared length N of a fixed-length BINARY(N)
342+
// `mytype` is the MySQL type for the column
343+
// `meta` is the metadata for the column type: length for fixed length types.
344+
// Reimplementation of MySQL metadata decoding:
345+
// - MySQL 8.4: https://github.com/mysql/mysql-server/blob/845d525d49c8027a4d0cdcc43372c96ba295c857/sql/log_event.cc#L1792-L1805
346+
// - MySQL 5.7: https://github.com/mysql/mysql-server/blob/f7680e98b6bbe3500399fbad465d08a6b75d7a5c/sql/log_event.cc#L2047-L2064
347+
// - MariaDB 10.6:
348+
// https://github.com/MariaDB/server/blob/fcd3f81e08daea471d371d3be051e6feabb06399/sql/rpl_utility_server.cc#L139-L140
349+
// https://github.com/MariaDB/server/blob/fcd3f81e08daea471d371d3be051e6feabb06399/sql/sql_type_string.cc#L70-L73
350+
func binaryColumnLength(mytype byte, meta uint16) int {
351+
if mytype != mysql.MYSQL_TYPE_STRING {
352+
return 0
353+
}
354+
if meta < 256 { // no bit-packing, just the length
355+
return int(meta)
356+
}
357+
// bit-packed, for more than 255 bytes:
358+
// value = lower_byte(meta) + 2^4 * ((higher_byte(meta)&0x30) XOR 0x30)
359+
lowerMetaByte := uint8(meta & 0xFF)
360+
higherMetaByte := uint8(meta >> 8)
361+
borrowedBitsMask := uint8(0x30)
362+
extraBits := higherMetaByte & borrowedBitsMask
363+
if extraBits != borrowedBitsMask { // More than 255 bytes
364+
return int(lowerMetaByte) | (int((extraBits)^borrowedBitsMask) << 4)
365+
}
366+
return int(meta & 0xFF)
367+
}
368+
341369
func QValueFromMysqlRowEvent(
342370
ev *replication.TableMapEvent, idx int,
343371
enums []string, sets []string,
@@ -497,7 +525,13 @@ func QValueFromMysqlRowEvent(
497525
case string:
498526
switch qkind {
499527
case types.QValueKindBytes:
500-
return types.QValueBytes{Val: shared.UnsafeFastStringToReadOnlyBytes(val)}, nil
528+
b := shared.UnsafeFastStringToReadOnlyBytes(val)
529+
if n := binaryColumnLength(mytype, ev.ColumnMeta[idx]); len(b) < n {
530+
padded := make([]byte, n)
531+
copy(padded, b)
532+
b = padded
533+
}
534+
return types.QValueBytes{Val: b}, nil
501535
case types.QValueKindString:
502536
return types.QValueString{Val: val}, nil
503537
case types.QValueKindEnum:

‎flow/e2e/clickhouse_mysql_test.go‎

Lines changed: 198 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1,6 +1,7 @@
11
package e2e
22

33
import (
4+
"encoding/hex"
45
"fmt"
56
"math"
67
"strconv"
@@ -229,6 +230,203 @@ func (s ClickHouseSuite) Test_MySQL_Blobs() {
229230
RequireEnvCanceled(s.t, env)
230231
}
231232

233+
// Test_MySQL_Binary_Trailing_Zeros reproduces trailing-0x00 truncation of fixed-length
234+
// BINARY(N) columns over CDC.
235+
//
236+
// MySQL right-pads BINARY(N) to N bytes with 0x00. Internally this is represented
237+
// as a binary array of length M=N-number_of_trailing(0x00). When reading from binlog these
238+
// values need to be adapted back to N bytes based on the column type.
239+
//
240+
// See https://github.com/shyiko/mysql-binlog-connector-java/issues/169#issuecomment-301864569
241+
// for a similar problem in a different project.
242+
func (s ClickHouseSuite) Test_MySQL_Binary_Trailing_Zeros() {
243+
if _, ok := s.source.(*MySqlSource); !ok {
244+
s.t.Skip("only applies to mysql")
245+
}
246+
247+
srcTableName := "test_binary_padding"
248+
srcFullName := s.attachSchemaSuffix(srcTableName)
249+
quotedSrcFullName := "\"" + strings.ReplaceAll(srcFullName, ".", "\".\"") + "\""
250+
dstTableName := "test_binary_padding_dst"
251+
252+
type binaryCase struct {
253+
id int
254+
label string
255+
beforeSnapshot bool
256+
// BINARY(16) covers the common short metadata path with a 1-byte row length.
257+
b16InsertHex string
258+
b16WantHex string
259+
// BINARY(255) covers the upper edge of the same metadata path without switching
260+
// to MySQL's 2-byte row length encoding for fixed strings longer than 255 bytes.
261+
// MySQL does not allow BINARY(N) with N > 255.
262+
b255InsertHex string
263+
b255WantHex string
264+
}
265+
266+
cases := []binaryCase{
267+
// SELECT path: correct today, included to prove snapshot and CDC agree.
268+
{
269+
id: 1,
270+
label: "snapshot_trailing_zero",
271+
beforeSnapshot: true,
272+
b16InsertHex: strings.Repeat("11", 15) + "00",
273+
b16WantHex: strings.Repeat("11", 15) + "00",
274+
b255InsertHex: strings.Repeat("11", 254) + "00",
275+
b255WantHex: strings.Repeat("11", 254) + "00",
276+
},
277+
// CDC path: MySQL packs these shorter than their declared BINARY(N) length.
278+
{
279+
id: 2,
280+
label: "cdc_trailing_zero",
281+
beforeSnapshot: false,
282+
b16InsertHex: "21" + strings.Repeat("11", 14) + "00",
283+
b16WantHex: "21" + strings.Repeat("11", 14) + "00",
284+
b255InsertHex: "21" + strings.Repeat("11", 253) + "00",
285+
b255WantHex: "21" + strings.Repeat("11", 253) + "00",
286+
},
287+
{
288+
id: 3,
289+
label: "cdc_multiple_trailing_zeros",
290+
beforeSnapshot: false,
291+
b16InsertHex: "22" + strings.Repeat("11", 11) + strings.Repeat("00", 4),
292+
b16WantHex: "22" + strings.Repeat("11", 11) + strings.Repeat("00", 4),
293+
b255InsertHex: "22" + strings.Repeat("11", 250) + strings.Repeat("00", 4),
294+
b255WantHex: "22" + strings.Repeat("11", 250) + strings.Repeat("00", 4),
295+
},
296+
{
297+
id: 4,
298+
label: "cdc_all_zero",
299+
beforeSnapshot: false,
300+
b16InsertHex: strings.Repeat("00", 16),
301+
b16WantHex: strings.Repeat("00", 16),
302+
b255InsertHex: strings.Repeat("00", 255),
303+
b255WantHex: strings.Repeat("00", 255),
304+
},
305+
{
306+
id: 5,
307+
label: "cdc_short_insert",
308+
beforeSnapshot: false,
309+
b16InsertHex: "41",
310+
b16WantHex: "41" + strings.Repeat("00", 15),
311+
b255InsertHex: "42",
312+
b255WantHex: "42" + strings.Repeat("00", 254),
313+
},
314+
// Controls: interior zero bytes and non-zero tails should be preserved without extra padding.
315+
{
316+
id: 6,
317+
label: "cdc_interior_zero_nonzero_tail",
318+
beforeSnapshot: false,
319+
b16InsertHex: "FF00112233445566778899AABBCCDDEE",
320+
b16WantHex: "FF00112233445566778899AABBCCDDEE",
321+
b255InsertHex: "FF00" + strings.Repeat("7F", 252) + "EE",
322+
b255WantHex: "FF00" + strings.Repeat("7F", 252) + "EE",
323+
},
324+
}
325+
326+
require.NoError(s.t, s.source.Exec(s.t.Context(), fmt.Sprintf(`
327+
CREATE TABLE IF NOT EXISTS %s (
328+
id INT PRIMARY KEY,
329+
label TEXT NOT NULL,
330+
b16 BINARY(16) NOT NULL,
331+
b255 BINARY(255) NOT NULL
332+
)
333+
`, quotedSrcFullName)))
334+
335+
var snapshotCases []binaryCase
336+
var cdcCases []binaryCase
337+
for _, tc := range cases {
338+
if tc.beforeSnapshot {
339+
snapshotCases = append(snapshotCases, tc)
340+
} else {
341+
cdcCases = append(cdcCases, tc)
342+
}
343+
}
344+
snapshotValues := make([]string, 0, len(snapshotCases))
345+
for _, tc := range snapshotCases {
346+
snapshotValues = append(snapshotValues, fmt.Sprintf(
347+
"(%d,'%s',UNHEX('%s'),UNHEX('%s'))",
348+
tc.id, tc.label, tc.b16InsertHex, tc.b255InsertHex))
349+
}
350+
require.NoError(s.t, s.source.Exec(s.t.Context(), fmt.Sprintf(
351+
`INSERT INTO %s (id,label,b16,b255) VALUES %s`, quotedSrcFullName, strings.Join(snapshotValues, ","))))
352+
353+
connectionGen := FlowConnectionGenerationConfig{
354+
FlowJobName: s.attachSuffix(srcTableName),
355+
TableNameMapping: map[string]string{srcFullName: dstTableName},
356+
Destination: s.Peer().Name,
357+
}
358+
flowConnConfig := connectionGen.GenerateFlowConnectionConfigs(s)
359+
flowConnConfig.DoInitialSnapshot = true
360+
361+
tc := NewTemporalClient(s.t)
362+
env := ExecutePeerflow(s.t, tc, flowConnConfig)
363+
SetupCDCFlowStatusQuery(s.t, env, flowConnConfig)
364+
365+
EnvWaitForCount(env, s, "waiting for snapshot row", dstTableName, "id,label,b16,b255", len(snapshotCases))
366+
367+
cdcValues := make([]string, 0, len(cdcCases))
368+
for _, tc := range cdcCases {
369+
cdcValues = append(cdcValues, fmt.Sprintf(
370+
"(%d,'%s',UNHEX('%s'),UNHEX('%s'))",
371+
tc.id, tc.label, tc.b16InsertHex, tc.b255InsertHex))
372+
}
373+
require.NoError(s.t, s.source.Exec(s.t.Context(), fmt.Sprintf(
374+
`INSERT INTO %s (id,label,b16,b255) VALUES %s`, quotedSrcFullName, strings.Join(cdcValues, ","))))
375+
376+
EnvWaitForCount(env, s, "waiting for cdc rows", dstTableName, "id,label,b16,b255", len(cases))
377+
378+
rows, err := s.GetRows(dstTableName, "id,label,b16,b255")
379+
require.NoError(s.t, err)
380+
require.Len(s.t, rows.Records, len(cases), "expected all binary padding test rows")
381+
382+
byLabel := make(map[string]map[string][]byte, len(rows.Records))
383+
for _, row := range rows.Records {
384+
label, ok := row[1].Value().(string)
385+
require.True(s.t, ok, "label should be a string")
386+
byLabel[label] = make(map[string][]byte, 2)
387+
for _, col := range []struct {
388+
name string
389+
idx int
390+
}{
391+
{name: "b16", idx: 2},
392+
{name: "b255", idx: 3},
393+
} {
394+
switch v := row[col.idx].Value().(type) {
395+
case string:
396+
byLabel[label][col.name] = []byte(v)
397+
case []byte:
398+
byLabel[label][col.name] = v
399+
default:
400+
require.Failf(s.t, "unexpected type for binary column",
401+
"label=%s column=%s type=%T", label, col.name, row[col.idx].Value())
402+
}
403+
}
404+
}
405+
406+
for _, tc := range cases {
407+
for _, col := range []struct {
408+
name string
409+
wantHex string
410+
}{
411+
{name: "b16", wantHex: tc.b16WantHex},
412+
{name: "b255", wantHex: tc.b255WantHex},
413+
} {
414+
want, decErr := hex.DecodeString(col.wantHex)
415+
require.NoError(s.t, decErr)
416+
gotByCol, ok := byLabel[tc.label]
417+
require.Truef(s.t, ok, "missing row %q in destination", tc.label)
418+
got, ok := gotByCol[col.name]
419+
require.Truef(s.t, ok, "missing column %q for row %q in destination", col.name, tc.label)
420+
require.Equalf(s.t, want, got,
421+
"%s %q: MySQL stores %d bytes (%s) but ClickHouse has %d bytes (%s)",
422+
col.name, tc.label, len(want), col.wantHex, len(got), strings.ToUpper(hex.EncodeToString(got)))
423+
}
424+
}
425+
426+
env.Cancel(s.t.Context())
427+
RequireEnvCanceled(s.t, env)
428+
}
429+
232430
func (s ClickHouseSuite) Test_MySQL_Enum() {
233431
if mySource, ok := s.source.(*MySqlSource); !ok {
234432
s.t.Skip("only applies to mysql")

0 commit comments

Comments
 (0)