Skip to content

Commit be8cb0c

Browse files
jgao54jgao54
committed
review feedback
1 parent dd731c6 commit be8cb0c

4 files changed

Lines changed: 176 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: 120 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,120 @@
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 ensure 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+
if minInt.Cmp(maxInt) > 0 {
68+
return nil, fmt.Errorf("min uuid (%s) greater than max uuid (%s)", minVal, maxVal)
69+
}
70+
71+
var partitions []*protos.QRepPartition
72+
start := minVal
73+
step := shared.BigIntDivCeil(new(big.Int).Sub(maxInt, minInt), big.NewInt(numPartitions))
74+
for value := new(big.Int).Add(minInt, step); value.Cmp(maxInt) < 0; value.Add(value, step) {
75+
end, err := bigIntToUUID(value, casing)
76+
if err != nil {
77+
return nil, fmt.Errorf("failed to convert bigint to uuid: %w", err)
78+
}
79+
partitions = append(partitions, createStringPartition(start, end, false))
80+
start = end
81+
}
82+
partitions = append(partitions, createStringPartition(start, maxVal, true))
83+
return partitions, nil
84+
}
85+
86+
func uuidToBigInt(s string) (*big.Int, error) {
87+
u, err := uuid.Parse(s)
88+
if err != nil {
89+
return nil, err
90+
}
91+
return new(big.Int).SetBytes(u[:]), nil
92+
}
93+
94+
func bigIntToUUID(n *big.Int, casing hexCasing) (string, error) {
95+
if n.BitLen() > 128 {
96+
return "", fmt.Errorf("value does not fit in a UUID (%d bits)", n.BitLen())
97+
}
98+
var b [16]byte
99+
n.FillBytes(b[:])
100+
s := uuid.UUID(b).String()
101+
if casing == hexCasingUpper {
102+
s = strings.ToUpper(s)
103+
}
104+
return s, nil
105+
}
106+
107+
func createStringPartition(start string, end string, endInclusive bool) *protos.QRepPartition {
108+
return &protos.QRepPartition{
109+
PartitionId: uuid.NewString(),
110+
Range: &protos.PartitionRange{
111+
Range: &protos.PartitionRange_StringRange{
112+
StringRange: &protos.StringPartitionRange{
113+
Start: start,
114+
End: end,
115+
EndInclusive: endInclusive,
116+
},
117+
},
118+
},
119+
}
120+
}

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

Lines changed: 48 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)
@@ -149,3 +145,23 @@ func TestAddPartitionsWithRangeUUIDSinglePartition(t *testing.T) {
149145
})
150146
}
151147
}
148+
149+
func TestBuildUuidStringPartitionsInvalid(t *testing.T) {
150+
cases := []struct {
151+
name string
152+
minV string
153+
maxV string
154+
}{
155+
{"arbitrary string", "apple", "banana"},
156+
{"inverted", "f47ac10b-58cc-4372-a567-0e02b2c3d479", "018f6e7a-1b2c-7def-8a3b-1c2d3e4f5a6b"},
157+
{"too short", "018f6e7a-1b2c", "f47ac10b-58cc"},
158+
{"too long", "018f6e7a-1b2c-7def-8a3b-1c2d3e4f5a6bb", "f47ac10b-58cc-4372-a567-0e02b2c3d4799"},
159+
}
160+
const numPartitions = 32
161+
for _, tc := range cases {
162+
t.Run(tc.name, func(t *testing.T) {
163+
_, err := buildUuidStringPartitions(tc.minV, tc.maxV, hexCasingLower, numPartitions)
164+
require.Error(t, err)
165+
})
166+
}
167+
}

0 commit comments

Comments
 (0)