Skip to content

Commit 36fd325

Browse files
jgao54jgao54
committed
review feedback
1 parent dd731c6 commit 36fd325

4 files changed

Lines changed: 153 additions & 135 deletions

File tree

‎flow/connectors/mysql/qrep.go‎

Lines changed: 4 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -157,10 +157,12 @@ func (c *MySqlConnector) GetQRepPartitions(
157157
start, startIsString := val1.Value().(string)
158158
end, endIsString := val2.Value().(string)
159159
if startIsString && endIsString {
160-
if isUuid, hexCasing := utils.DetectUuidWithHexCasing(start, end); isUuid {
161-
if err := partitionHelper.AddUuidStringPartitionsWithRange(start, end, hexCasing, numPartitions); err != nil {
160+
if isUUID, casing := detectUuidWithHexCasing(start, end); isUUID {
161+
uuidPartitions, err := buildUuidStringPartitions(start, end, casing, numPartitions)
162+
if err != nil {
162163
return nil, fmt.Errorf("failed to add uuid string partitions: %w", err)
163164
}
165+
partitionHelper.AddPartitions(uuidPartitions)
164166
} else {
165167
c.logger.Info("string watermark column is not uuid, falling back to full table partition")
166168
return utils.FullTablePartition(), nil

‎flow/connectors/mysql/qrep_partition.go‎

Lines changed: 117 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,117 @@
1+
package connmysql
2+
3+
import (
4+
"fmt"
5+
"math/big"
6+
"regexp"
7+
"strings"
8+
9+
"github.com/google/uuid"
10+
11+
"github.com/PeerDB-io/peerdb/flow/generated/protos"
12+
"github.com/PeerDB-io/peerdb/flow/shared"
13+
)
14+
15+
// String watermark partitioning is currently only supported by the MySQL connector,
16+
// so these helpers live in the MySQL connector subtree to prevent accidental usage
17+
// by other connectors.
18+
19+
var (
20+
uuidLowerRe = regexp.MustCompile(`^[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}$`)
21+
uuidUpperRe = regexp.MustCompile(`^[0-9A-F]{8}-[0-9A-F]{4}-[0-9A-F]{4}-[0-9A-F]{4}-[0-9A-F]{12}$`)
22+
)
23+
24+
type hexCasing int
25+
26+
const (
27+
hexCasingUnknown hexCasing = iota
28+
hexCasingLower
29+
hexCasingUpper
30+
)
31+
32+
// detectUuidWithHexCasing best-effort classifies a string watermark column by
33+
// inspecting only its min and max bounds: it determines whether both are
34+
// canonical UUIDs, and if so, the shared casing of their hex letters.
35+
//
36+
// Because only the bounds are examined, the classification can be wrong in
37+
// the following rare cases:
38+
// - non-UUID rows can exist between UUID-shaped min/max bounds;
39+
// - non-boundary UUIDs contain a mix of upper and lower case
40+
// - min/max bounds both don't contain hex, so casing blindly defaults to lower
41+
//
42+
// These cases may lead to partition skew, but will not affect correctness.
43+
func detectUuidWithHexCasing(minVal string, maxVal string) (bool, hexCasing) {
44+
switch {
45+
case uuidLowerRe.MatchString(minVal) && uuidLowerRe.MatchString(maxVal):
46+
return true, hexCasingLower
47+
case uuidUpperRe.MatchString(minVal) && uuidUpperRe.MatchString(maxVal):
48+
return true, hexCasingUpper
49+
default:
50+
return false, hexCasingUnknown
51+
}
52+
}
53+
54+
// buildUuidStringPartitions splits a UUID string watermark column into partitions.
55+
// The original minVal/maxVal are used for first/last partitions to guarantee correctness.
56+
func buildUuidStringPartitions(
57+
minVal string, maxVal string, casing hexCasing, numPartitions int64,
58+
) ([]*protos.QRepPartition, error) {
59+
minInt, err := uuidToBigInt(minVal)
60+
if err != nil {
61+
return nil, fmt.Errorf("failed to convert min uuid to bigint: %w", err)
62+
}
63+
maxInt, err := uuidToBigInt(maxVal)
64+
if err != nil {
65+
return nil, fmt.Errorf("failed to convert max uuid to bigint: %w", err)
66+
}
67+
68+
var partitions []*protos.QRepPartition
69+
start := minVal
70+
step := shared.BigIntDivCeil(new(big.Int).Sub(maxInt, minInt), big.NewInt(numPartitions))
71+
for value := new(big.Int).Add(minInt, step); value.Cmp(maxInt) < 0; value.Add(value, step) {
72+
end, err := bigIntToUUID(value, casing)
73+
if err != nil {
74+
return nil, fmt.Errorf("failed to convert bigint to uuid: %w", err)
75+
}
76+
partitions = append(partitions, createStringPartition(start, end, false))
77+
start = end
78+
}
79+
partitions = append(partitions, createStringPartition(start, maxVal, true))
80+
return partitions, nil
81+
}
82+
83+
func uuidToBigInt(s string) (*big.Int, error) {
84+
u, err := uuid.Parse(s)
85+
if err != nil {
86+
return nil, err
87+
}
88+
return new(big.Int).SetBytes(u[:]), nil
89+
}
90+
91+
func bigIntToUUID(n *big.Int, casing hexCasing) (string, error) {
92+
if n.BitLen() > 128 {
93+
return "", fmt.Errorf("value does not fit in a UUID (%d bits)", n.BitLen())
94+
}
95+
var b [16]byte
96+
n.FillBytes(b[:])
97+
s := uuid.UUID(b).String()
98+
if casing == hexCasingUpper {
99+
s = strings.ToUpper(s)
100+
}
101+
return s, nil
102+
}
103+
104+
func createStringPartition(start string, end string, endInclusive bool) *protos.QRepPartition {
105+
return &protos.QRepPartition{
106+
PartitionId: uuid.NewString(),
107+
Range: &protos.PartitionRange{
108+
Range: &protos.PartitionRange_StringRange{
109+
StringRange: &protos.StringPartitionRange{
110+
Start: start,
111+
End: end,
112+
EndInclusive: endInclusive,
113+
},
114+
},
115+
},
116+
}
117+
}

flow/connectors/utils/partition_test.go renamed to flow/connectors/mysql/qrep_partition_test.go

Lines changed: 28 additions & 32 deletions
Original file line numberDiff line numberDiff line change
@@ -1,13 +1,11 @@
1-
package utils
1+
package connmysql
22

33
import (
4-
"log/slog"
54
"regexp"
65
"testing"
76

87
"github.com/stretchr/testify/assert"
98
"github.com/stretchr/testify/require"
10-
"go.temporal.io/sdk/log"
119

1210
"github.com/PeerDB-io/peerdb/flow/generated/protos"
1311
)
@@ -16,13 +14,13 @@ func TestUUIDToBigIntRoundTrip(t *testing.T) {
1614
cases := []struct {
1715
name string
1816
uuid string
19-
casing HexCasing
17+
casing hexCasing
2018
}{
21-
{"lower case v4", "f47ac10b-58cc-4372-a567-0e02b2c3d479", Lower},
22-
{"upper case v7", "017F22E2-79B0-7CC3-98C4-DC0C0C07398F", Upper},
23-
{"zero", "00000000-0000-0000-0000-000000000000", Lower},
24-
{"max", "ffffffff-ffff-ffff-ffff-ffffffffffff", Lower},
25-
{"all digits", "01234567-8901-2345-6789-012345678901", Lower},
19+
{"lower case v4", "f47ac10b-58cc-4372-a567-0e02b2c3d479", hexCasingLower},
20+
{"upper case v7", "017F22E2-79B0-7CC3-98C4-DC0C0C07398F", hexCasingUpper},
21+
{"zero", "00000000-0000-0000-0000-000000000000", hexCasingLower},
22+
{"max", "ffffffff-ffff-ffff-ffff-ffffffffffff", hexCasingLower},
23+
{"all digits", "01234567-8901-2345-6789-012345678901", hexCasingLower},
2624
}
2725
for _, tc := range cases {
2826
t.Run(tc.name, func(t *testing.T) {
@@ -53,44 +51,43 @@ func TestDetectUUIDCase(t *testing.T) {
5351
minVal string
5452
maxVal string
5553
expectIsUUID bool
56-
expectCasing HexCasing
54+
expectCasing hexCasing
5755
}{
58-
{"both lower", "00000000-0000-0000-0000-000000000000", "f47ac10b-58cc-4372-a567-0e02b2c3d479", true, Lower},
59-
{"both upper", "00000000-0000-0000-0000-000000000000", "F47AC10B-58CC-4372-A567-0E02B2C3D479", true, Upper},
60-
{"all digits", "01234567-8901-2345-6789-012345678901", "09234567-8901-2345-6789-012345678901", true, Lower},
61-
{"digits min, upper max", "01234567-8901-2345-6789-012345678901", "F47AC10B-58CC-4372-A567-0E02B2C3D479", true, Upper},
62-
{"lower min, digits max", "f47ac10b-58cc-4372-a567-0e02b2c3d479", "01234567-8901-2345-6789-012345678901", true, Lower},
63-
{"mixed case across bounds", "f47ac10b-58cc-4372-a567-0e02b2c3d479", "F47AC10B-58CC-4372-A567-0E02B2C3D479", false, Unknown},
64-
{"mixed case within a bound", "00000000-0000-0000-0000-000000000000", "F47ac10b-58cc-4372-a567-0e02b2c3d479", false, Unknown},
65-
{"non-uuid", "apple", "banana", false, Unknown},
56+
{"both lower", "00000000-0000-0000-0000-000000000000", "f47ac10b-58cc-4372-a567-0e02b2c3d479", true, hexCasingLower},
57+
{"both upper", "00000000-0000-0000-0000-000000000000", "F47AC10B-58CC-4372-A567-0E02B2C3D479", true, hexCasingUpper},
58+
{"all digits", "01234567-8901-2345-6789-012345678901", "09234567-8901-2345-6789-012345678901", true, hexCasingLower},
59+
{"digits min, upper max", "01234567-8901-2345-6789-012345678901", "F47AC10B-58CC-4372-A567-0E02B2C3D479", true, hexCasingUpper},
60+
{"lower min, digits max", "f47ac10b-58cc-4372-a567-0e02b2c3d479", "01234567-8901-2345-6789-012345678901", true, hexCasingLower},
61+
{"mixed case across bounds", "f47ac10b-58cc-4372-a567-0e02b2c3d479", "F47AC10B-58CC-4372-A567-0E02B2C3D479", false, hexCasingUnknown},
62+
{"mixed case within a bound", "00000000-0000-0000-0000-000000000000", "F47ac10b-58cc-4372-a567-0e02b2c3d479", false, hexCasingUnknown},
63+
{"non-uuid", "apple", "banana", false, hexCasingUnknown},
6664
}
6765
for _, tc := range cases {
6866
t.Run(tc.name, func(t *testing.T) {
69-
isUUID, hexCasing := DetectUuidWithHexCasing(tc.minVal, tc.maxVal)
67+
isUUID, casing := detectUuidWithHexCasing(tc.minVal, tc.maxVal)
7068
assert.Equal(t, tc.expectIsUUID, isUUID)
71-
assert.Equal(t, tc.expectCasing, hexCasing)
69+
assert.Equal(t, tc.expectCasing, casing)
7270
})
7371
}
7472
}
7573

76-
func TestAddPartitionsWithRangeUUID(t *testing.T) {
74+
func TestBuildUuidStringPartitions(t *testing.T) {
7775
cases := []struct {
7876
name string
7977
minV string
8078
maxV string
8179
expectedCase *regexp.Regexp
8280
}{
83-
{"lower", "018f6e7a-1b2c-7def-8a3b-1c2d3e4f5a6b", "f47ac10b-58cc-4372-a567-0e02b2c3d479", UuidLowerRe},
84-
{"upper", "018F6E7A-1B2C-7DEF-8A3B-1C2D3E4F5A6B", "F47AC10B-58CC-4372-A567-0E02B2C3D479", UuidUpperRe},
81+
{"lower", "018f6e7a-1b2c-7def-8a3b-1c2d3e4f5a6b", "f47ac10b-58cc-4372-a567-0e02b2c3d479", uuidLowerRe},
82+
{"upper", "018F6E7A-1B2C-7DEF-8A3B-1C2D3E4F5A6B", "F47AC10B-58CC-4372-A567-0E02B2C3D479", uuidUpperRe},
8583
}
8684
const numPartitions = 32
8785
for _, tc := range cases {
8886
t.Run(tc.name, func(t *testing.T) {
89-
p := NewPartitionHelper(log.NewStructuredLogger(slog.Default()))
90-
isUUID, hexCasing := DetectUuidWithHexCasing(tc.minV, tc.maxV)
87+
isUUID, casing := detectUuidWithHexCasing(tc.minV, tc.maxV)
9188
require.True(t, isUUID)
92-
require.NoError(t, p.AddUuidStringPartitionsWithRange(tc.minV, tc.maxV, hexCasing, numPartitions))
93-
parts := p.GetPartitions()
89+
parts, err := buildUuidStringPartitions(tc.minV, tc.maxV, casing, numPartitions)
90+
require.NoError(t, err)
9491
require.Len(t, parts, numPartitions)
9592

9693
prevEnd := ""
@@ -121,7 +118,7 @@ func TestAddPartitionsWithRangeUUID(t *testing.T) {
121118
}
122119
}
123120

124-
func TestAddPartitionsWithRangeUUIDSinglePartition(t *testing.T) {
121+
func TestBuildUuidStringPartitionsSinglePartition(t *testing.T) {
125122
uuid1 := "018f6e7a-1b2c-7def-8a3b-1c2d3e4f5a6b"
126123
uuid2 := "f47ac10b-58cc-4372-a567-0e02b2c3d479"
127124
cases := []struct {
@@ -135,11 +132,10 @@ func TestAddPartitionsWithRangeUUIDSinglePartition(t *testing.T) {
135132
}
136133
for _, tc := range cases {
137134
t.Run(tc.name, func(t *testing.T) {
138-
p := NewPartitionHelper(log.NewStructuredLogger(slog.Default()))
139-
isUUID, hexCasing := DetectUuidWithHexCasing(tc.minV, tc.maxV)
135+
isUUID, casing := detectUuidWithHexCasing(tc.minV, tc.maxV)
140136
require.True(t, isUUID)
141-
require.NoError(t, p.AddUuidStringPartitionsWithRange(tc.minV, tc.maxV, hexCasing, tc.numPartitions))
142-
parts := p.GetPartitions()
137+
parts, err := buildUuidStringPartitions(tc.minV, tc.maxV, casing, tc.numPartitions)
138+
require.NoError(t, err)
143139
require.Len(t, parts, 1)
144140
sr, ok := parts[0].Range.Range.(*protos.PartitionRange_StringRange)
145141
require.True(t, ok)

‎flow/connectors/utils/partition.go‎

Lines changed: 4 additions & 101 deletions
Original file line numberDiff line numberDiff line change
@@ -4,9 +4,6 @@ import (
44
"cmp"
55
"fmt"
66
"log/slog"
7-
"math/big"
8-
"regexp"
9-
"strings"
107
"time"
118

129
"github.com/google/uuid"
@@ -18,11 +15,6 @@ import (
1815
"github.com/PeerDB-io/peerdb/flow/shared"
1916
)
2017

21-
var (
22-
UuidLowerRe = regexp.MustCompile(`^[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}$`)
23-
UuidUpperRe = regexp.MustCompile(`^[0-9A-F]{8}-[0-9A-F]{4}-[0-9A-F]{4}-[0-9A-F]{4}-[0-9A-F]{12}$`)
24-
)
25-
2618
const FullTablePartitionID = "full-table-partition-id"
2719

2820
func FullTablePartition() []*protos.QRepPartition {
@@ -157,21 +149,6 @@ func createUIntPartition(start uint64, end uint64) *protos.QRepPartition {
157149
}
158150
}
159151

160-
func createStringPartition(start string, end string, endInclusive bool) *protos.QRepPartition {
161-
return &protos.QRepPartition{
162-
PartitionId: uuid.NewString(),
163-
Range: &protos.PartitionRange{
164-
Range: &protos.PartitionRange_StringRange{
165-
StringRange: &protos.StringPartitionRange{
166-
Start: start,
167-
End: end,
168-
EndInclusive: endInclusive,
169-
},
170-
},
171-
},
172-
}
173-
}
174-
175152
type PartitionHelper struct {
176153
logger log.Logger
177154
prevStart any
@@ -363,84 +340,6 @@ func (p *PartitionHelper) AddPartitionsWithRange(start any, end any, numPartitio
363340
return nil
364341
}
365342

366-
type HexCasing int
367-
368-
const (
369-
Unknown HexCasing = iota
370-
Lower
371-
Upper
372-
)
373-
374-
// AddUuidStringPartitionsWithRange splits a UUID watermark range into partitions.
375-
func (p *PartitionHelper) AddUuidStringPartitionsWithRange(
376-
minVal string, maxVal string, hexCasing HexCasing, numPartitions int64,
377-
) error {
378-
minInt, err := uuidToBigInt(minVal)
379-
if err != nil {
380-
return fmt.Errorf("failed to convert min uuid to bigint: %w", err)
381-
}
382-
maxInt, err := uuidToBigInt(maxVal)
383-
if err != nil {
384-
return fmt.Errorf("failed to convert max uuid to bigint: %w", err)
385-
}
386-
387-
start := minVal
388-
step := shared.BigIntDivCeil(new(big.Int).Sub(maxInt, minInt), big.NewInt(numPartitions))
389-
for value := new(big.Int).Add(minInt, step); value.Cmp(maxInt) < 0; value.Add(value, step) {
390-
end, err := bigIntToUUID(value, hexCasing)
391-
if err != nil {
392-
return fmt.Errorf("failed to convert bigint to uuid: %w", err)
393-
}
394-
p.partitions = append(p.partitions, createStringPartition(start, end, false))
395-
start = end
396-
}
397-
p.partitions = append(p.partitions, createStringPartition(start, maxVal, true))
398-
return nil
399-
}
400-
401-
// DetectUuidWithHexCasing best-effort classifies a string watermark column by
402-
// inspecting only its min and max bounds: it determines whether both are
403-
// canonical UUIDs, and if so, the shared casing of their hex letters.
404-
//
405-
// Because only the bounds are examined, the classification can be wrong in
406-
// the following rare cases:
407-
// - non-UUID rows can exist between UUID-shaped min/max bounds;
408-
// - non-boundary UUIDs contain a mix of upper and lower case
409-
// - min/max bounds both don't contain hex, so casing blindly defaults to lower
410-
//
411-
// These cases may lead to partition skew, but will not affect correctness.
412-
func DetectUuidWithHexCasing(minVal string, maxVal string) (bool, HexCasing) {
413-
switch {
414-
case UuidLowerRe.MatchString(minVal) && UuidLowerRe.MatchString(maxVal):
415-
return true, Lower
416-
case UuidUpperRe.MatchString(minVal) && UuidUpperRe.MatchString(maxVal):
417-
return true, Upper
418-
default:
419-
return false, Unknown
420-
}
421-
}
422-
423-
func uuidToBigInt(s string) (*big.Int, error) {
424-
u, err := uuid.Parse(s)
425-
if err != nil {
426-
return nil, err
427-
}
428-
return new(big.Int).SetBytes(u[:]), nil
429-
}
430-
431-
func bigIntToUUID(n *big.Int, hexCasing HexCasing) (string, error) {
432-
if n.BitLen() > 128 {
433-
return "", fmt.Errorf("value does not fit in a UUID (%d bits)", n.BitLen())
434-
}
435-
var b [16]byte
436-
n.FillBytes(b[:])
437-
s := uuid.UUID(b).String()
438-
if hexCasing == Upper {
439-
s = strings.ToUpper(s)
440-
}
441-
return s, nil
442-
}
443-
444343
func (p *PartitionHelper) getPartitionForStartAndEnd(start any, end any) (*protos.QRepPartition, error) {
445344
if start == nil || end == nil {
446345
return nil, nil
@@ -503,6 +402,10 @@ func (p *PartitionHelper) updatePartitionHelper(partition *protos.QRepPartition)
503402
return nil
504403
}
505404

405+
func (p *PartitionHelper) AddPartitions(partitions []*protos.QRepPartition) {
406+
p.partitions = append(p.partitions, partitions...)
407+
}
408+
506409
func (p *PartitionHelper) GetPartitions() []*protos.QRepPartition {
507410
return p.partitions
508411
}

0 commit comments

Comments
 (0)