diff --git a/flow/connectors/mysql/cdc.go b/flow/connectors/mysql/cdc.go index 9dc00e2db..ae34d9bf6 100644 --- a/flow/connectors/mysql/cdc.go +++ b/flow/connectors/mysql/cdc.go @@ -380,21 +380,22 @@ func (c *MySqlConnector) startSyncer(ctx context.Context, env map[string]string) } return replication.NewBinlogSyncer(replication.BinlogSyncerConfig{ - ServerID: serverId, - Flavor: c.Flavor(), - Host: config.Host, - Port: uint16(config.Port), - User: config.User, - Password: config.Password, - Logger: internal.SlogLoggerFromCtx(ctx), - Dialer: c.Dialer(), - DisableRetrySync: true, - UseDecimal: true, - ParseTime: true, - VerifyChecksum: true, - TLSConfig: tlsConfig, - HeartbeatPeriod: c.binlogHeartbeatPeriod, - EventCacheCount: eventCacheCount, + ServerID: serverId, + Flavor: c.Flavor(), + Host: config.Host, + Port: uint16(config.Port), + User: config.User, + Password: config.Password, + Logger: internal.SlogLoggerFromCtx(ctx), + Dialer: c.Dialer(), + DisableRetrySync: true, + UseDecimal: true, + ParseTime: true, + VerifyChecksum: true, + RenderJSONAsMySQLText: true, + TLSConfig: tlsConfig, + HeartbeatPeriod: c.binlogHeartbeatPeriod, + EventCacheCount: eventCacheCount, }), nil } diff --git a/flow/connectors/mysql/qvalue_convert.go b/flow/connectors/mysql/qvalue_convert.go index 3c2d65c22..5d01fdf09 100644 --- a/flow/connectors/mysql/qvalue_convert.go +++ b/flow/connectors/mysql/qvalue_convert.go @@ -1,7 +1,9 @@ package connmysql import ( + "bytes" "encoding/binary" + "encoding/json" "errors" "fmt" "log/slog" @@ -28,6 +30,14 @@ import ( "github.com/PeerDB-io/peerdb/flow/shared/types" ) +func compactMySQLJSON(v []byte) string { + var buf bytes.Buffer + if err := json.Compact(&buf, v); err != nil { + return string(v) + } + return buf.String() +} + func qkindFromMysqlType(mytype byte, unsigned bool, charset uint16, version uint32) (types.QValueKind, error) { switch mytype { case mysql.MYSQL_TYPE_TINY: @@ -332,7 +342,8 @@ func QValueFromMysqlFieldValue(qkind types.QValueKind, mytype byte, fv mysql.Fie case types.QValueKindBytes: return types.QValueBytes{Val: slices.Clone(v)}, nil case types.QValueKindJSON: - return types.QValueJSON{Val: string(v)}, nil + // keep snapshot and cdc json representation consistent + return types.QValueJSON{Val: compactMySQLJSON(v)}, nil case types.QValueKindUUID: // snapshot reads via the text protocol, so MariaDB sends the canonical string u, err := uuid.Parse(unsafeString) diff --git a/flow/connectors/mysql/qvalue_convert_test.go b/flow/connectors/mysql/qvalue_convert_test.go index ec8df539d..7d3a4767d 100644 --- a/flow/connectors/mysql/qvalue_convert_test.go +++ b/flow/connectors/mysql/qvalue_convert_test.go @@ -101,3 +101,28 @@ func TestProcessTime(t *testing.T) { require.Equal(t, ts.out, tm) } } + +func TestCompactMySQLJSON(t *testing.T) { + for _, tc := range []struct { + in string + want string + }{ + // MySQL server-rendered text: space after ':' and ','. + {`{"a": 1, "b": 2}`, `{"a":1,"b":2}`}, + // DOUBLE keeps its trailing zero; it is not renumbered. + {`{"x": 1.0}`, `{"x":1.0}`}, + // Object key order is preserved (not lexicographically sorted). + {`{"b": 1, "a": 2}`, `{"b":1,"a":2}`}, + // Nested arrays/objects. + {`{"arr": [1, 2, {"k": "v"}]}`, `{"arr":[1,2,{"k":"v"}]}`}, + // Whitespace inside string values is untouched. + {`{"s": "a, b: c"}`, `{"s":"a, b: c"}`}, + {`{}`, `{}`}, + {`[]`, `[]`}, + } { + require.Equal(t, tc.want, compactMySQLJSON([]byte(tc.in)), "in=%q", tc.in) + } + + // Invalid JSON falls back to the original bytes rather than dropping the value. + require.Equal(t, `not json`, compactMySQLJSON([]byte(`not json`))) +} diff --git a/flow/e2e/clickhouse_mysql_test.go b/flow/e2e/clickhouse_mysql_test.go index 3433ec9cd..29593e82a 100644 --- a/flow/e2e/clickhouse_mysql_test.go +++ b/flow/e2e/clickhouse_mysql_test.go @@ -277,6 +277,164 @@ func (s ClickHouseSuite) Test_MySQL_Blobs() { RequireEnvCanceled(s.t, env) } +func (s ClickHouseSuite) Test_MySQL_JSON_SnapshotCDCConsistency() { + mySource, ok := s.source.(*MySqlSource) + if !ok { + s.t.Skip("only applies to mysql") + } + // MySQL stores JSON in a native binary form and re-serializes it to a compact, + // whitespace-stripped text; MariaDB's JSON type is plain LONGTEXT stored verbatim, so + // the exact-representation checks below only hold on MySQL. + nativeJSON := mySource.Config.Flavor == protos.MySqlFlavor_MYSQL_MYSQL + + srcTableName := "test_json_repr" + srcFullName := s.attachSchemaSuffix(srcTableName) + dstTableName := "test_json_repr_dst" + + require.NoError(s.t, s.source.Exec(s.t.Context(), fmt.Sprintf(` + CREATE TABLE IF NOT EXISTS %s ( + id serial PRIMARY KEY, + js json NOT NULL + )`, srcFullName))) + + variants := []struct { + val string + expectExact string + }{ + // key order: shorter key sorts after longer one + {val: `{"z": 1, "aa": 2}`}, + {val: `{"b": 2, "a": 1}`}, // key order: reversed on storage + // nested object key order + array order preserved + {val: `{"n": [3, 1, 2], "obj": {"y": 1, "x": 2}}`}, + {val: `{"s": "a, b: c"}`}, // whitespace inside strings is preserved + // non-whole doubles that render identically in both paths + {val: `{"pi": 3.14, "half": 0.5, "neg": -2.5}`}, + {val: `{"big": 1234567890123456789}`, expectExact: `{"big":1234567890123456789}`}, // 19-digit int64: float64 rounds to ...6800 + {val: `{"maxsafe": 9007199254740993}`, expectExact: `{"maxsafe":9007199254740993}`}, // 2^53+1: float64 drops the low bit (...992) + {val: `{"negbig": -1234567890123456789}`, expectExact: `{"negbig":-1234567890123456789}`}, // sign preserved at full precision + { + val: `{"mix": [1234567890123456789, -9007199254740993, 9007199254740993]}`, // same hazards inside an array + expectExact: `{"mix":[1234567890123456789,-9007199254740993,9007199254740993]}`, + }, + } + + insertVariants := func() { + for _, v := range variants { + require.NoError(s.t, s.source.Exec(s.t.Context(), + fmt.Sprintf(`INSERT INTO %s (js) VALUES ('%s')`, srcFullName, v.val))) + } + } + + insertVariants() + + connectionGen := FlowConnectionGenerationConfig{ + FlowJobName: s.attachSuffix(srcTableName), + TableNameMapping: map[string]string{srcFullName: dstTableName}, + Destination: s.Peer().Name, + } + flowConnConfig := connectionGen.GenerateFlowConnectionConfigs(s) + flowConnConfig.DoInitialSnapshot = true + + tc := NewTemporalClient(s.t) + env := ExecutePeerflow(s.t, tc, flowConnConfig) + SetupCDCFlowStatusQuery(s.t, env, flowConnConfig) + + EnvWaitForCount(env, s, "waiting for initial snapshot", dstTableName, "id", len(variants)) + + insertVariants() + + EnvWaitForCount(env, s, "waiting for cdc", dstTableName, "id", 2*len(variants)) + + // GetRows orders by id, so snapshot rows (ids 1..N) come first, then CDC rows (N+1..2N); + // both batches hold the same values in the same order, so row i and row i+N are the same + // logical value read via the snapshot and CDC paths respectively. + rows, err := s.GetRows(dstTableName, "id, js") + require.NoError(s.t, err) + require.Len(s.t, rows.Records, 2*len(variants)) + + for i := range variants { + snapshotGot := fmt.Sprint(rows.Records[i][1].Value()) + cdcGot := fmt.Sprint(rows.Records[i+len(variants)][1].Value()) + require.Equal(s.t, snapshotGot, cdcGot, "snapshot/cdc JSON representation differs for %q", variants[i].val) + if nativeJSON && variants[i].expectExact != "" { + require.Equal(s.t, variants[i].expectExact, snapshotGot, "snapshot JSON not preserved faithfully for %q", variants[i].val) + require.Equal(s.t, variants[i].expectExact, cdcGot, "cdc JSON not preserved faithfully for %q", variants[i].val) + } + } + + env.Cancel(s.t.Context()) + RequireEnvCanceled(s.t, env) +} + +// Test_MySQL_JSON_SnapshotCDCConsistency_NativeJSON is the native-JSON companion to +// Test_MySQL_JSON_SnapshotCDCConsistency. +func (s ClickHouseSuite) Test_MySQL_JSON_SnapshotCDCConsistency_NativeJSON() { + if _, ok := s.source.(*MySqlSource); !ok { + s.t.Skip("only applies to mysql") + } + + srcTableName := "test_json_repr_native" + srcFullName := s.attachSchemaSuffix(srcTableName) + dstTableName := "test_json_repr_native_dst" + + require.NoError(s.t, s.source.Exec(s.t.Context(), fmt.Sprintf(` + CREATE TABLE IF NOT EXISTS %s ( + id serial PRIMARY KEY, + js json NOT NULL + )`, srcFullName))) + + // These deliberately include the doubles that diverge in text form between the snapshot + // and CDC paths; native JSON normalization is what makes them consistent at the destination. + variants := []string{ + `{"pi": 3.14, "big": 1000000.5, "neg": -2.5}`, // magnitude where snapshot/CDC text forms differ + `{"tiny": 0.000015, "huge": 150000000000.5}`, // more exponent-crossover magnitudes + `{"z": 1.0, "aa": 2}`, // key order, normalized consistently + } + + insertVariants := func() { + for _, v := range variants { + require.NoError(s.t, s.source.Exec(s.t.Context(), + fmt.Sprintf(`INSERT INTO %s (js) VALUES ('%s')`, srcFullName, v))) + } + } + + insertVariants() + + connectionGen := FlowConnectionGenerationConfig{ + FlowJobName: s.attachSuffix(srcTableName), + TableNameMapping: map[string]string{srcFullName: dstTableName}, + Destination: s.Peer().Name, + } + flowConnConfig := connectionGen.GenerateFlowConnectionConfigs(s) + flowConnConfig.DoInitialSnapshot = true + flowConnConfig.Env = map[string]string{"PEERDB_CLICKHOUSE_ENABLE_JSON": "true"} + + tc := NewTemporalClient(s.t) + env := ExecutePeerflow(s.t, tc, flowConnConfig) + SetupCDCFlowStatusQuery(s.t, env, flowConnConfig) + + EnvWaitForCount(env, s, "waiting for initial snapshot", dstTableName, "id", len(variants)) + + insertVariants() + + EnvWaitForCount(env, s, "waiting for cdc", dstTableName, "id", 2*len(variants)) + + // Read ClickHouse's own normalized text of the JSON column so the comparison is over a + // plain String scan target (GetRows does not scan the native JSON type directly). + rows, err := s.GetRows(dstTableName, "id, toString(js)") + require.NoError(s.t, err) + require.Len(s.t, rows.Records, 2*len(variants)) + + for i := range variants { + snapshotGot := fmt.Sprint(rows.Records[i][1].Value()) + cdcGot := fmt.Sprint(rows.Records[i+len(variants)][1].Value()) + require.Equal(s.t, snapshotGot, cdcGot, "snapshot/cdc JSON representation differs for %q", variants[i]) + } + + env.Cancel(s.t.Context()) + RequireEnvCanceled(s.t, env) +} + func (s ClickHouseSuite) Test_MySQL_TransactionPayloadCompression() { mySource, ok := s.source.(*MySqlSource) if !ok {