Skip to content

Optimize MySQL Bootstrap Chunking by Switching from MD5 to CRC32 #987

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
akshayrai merged 2 commits into from
Oct 13, 2025
+22 −24
Merged

Optimize MySQL Bootstrap Chunking by Switching from MD5 to CRC32 #987

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
e8417da
Switch from MD5 to CRC32 to improve scan speed
akshayrai Sep 12, 2025
24c6f01
Cleanup
akshayrai Oct 8, 2025
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
14 changes: 6 additions & 8 deletions ...main/java/com/linkedin/datastream/common/databases/dbreader/MySqlChunkedQueryManager.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -19,7 +19,7 @@ public class MySqlChunkedQueryManager implements ChunkedQueryManager {
private static final String SELECT_FROM = "SELECT * FROM ( ";

/** Generate base predicate for sharding keys to given number of partitions.
* Ex: MOD ( CONV ( MD5 ( CONCAT ( K1, K2, K3 ) ) , 16, 10 ) , 10 ) for a table with 3 keys {K1, K2, K3} and 10 partitions */
* Ex: MOD ( CRC32 ( CONCAT ( K1, K2, K3 ) ), 10 ) for a table with 3 keys {K1, K2, K3} and 10 partitions */
private static String generatePerPartitionHashPredicate(List<String> keys, int partitionCount) {
StringBuilder query = new StringBuilder();
int keyCount = keys.size();
Expand All @@ -31,17 +31,15 @@ private static String generatePerPartitionHashPredicate(List<String> keys, int p
}
query.append(" )");

// Wrap that with MOD, CONV, and MD5 to generate a hash for sharding
// MOD ( CONV ( MD5 ( CONCAT ( A, B, C ) ) , 16, 10 ) , 10 )
query.insert(0, "MD5 ( ").append(" )");
// 16, 10 converts from HEX to DEC
query.insert(0, "CONV ( ").append(" , 16, 10 )");
// Wrap that with MOD, CRC32 to generate a hash for sharding
// MOD ( CRC32 ( CONCAT ( A, B, C ) ) , 10 )
query.insert(0, "CRC32 ( ").append(" )");
query.insert(0, "MOD ( ").append(" , ").append(partitionCount).append(" )");
return query.toString();
}

/** Generate predicate for filtering rows hashing to the assigned partitions :
* Ex: WHERE ( MOD ( CONV ( MD5 ( CONCAT ( K1, K2, K3 ) ) , 16, 10 ) , 10 ) IN (1 , 6 ) )
* Ex: WHERE ( MOD ( CRC32 ( CONCAT ( K1, K2, K3 ) ), 10 ) IN (1 , 6 ) )
* where 1 and 6 are the assigned partitions, 10 the partition count and, {K1, K2, K3} the keys of the table
*/
private static String generateFullPartitionHashPredicate(String perPartitionPredicate, List<Integer> partitions) {
Expand Down Expand Up @@ -121,7 +119,7 @@ public void prepareChunkedQuery(PreparedStatement stmt, List<Object> values) thr
// SELECT * FROM
// (
// SELECT * FROM TABLE
// ) nestedTab1 WHERE ( MOD ( CONV ( MD5 ( CONCAT ( KEY1 , KEY2 ) ) 16, 10 ) , 10 ) IN ( 2 , 5 ) )
// ) nestedTab1 WHERE ( MOD CRC32 ( CONCAT ( KEY1 , KEY2 ) ) , 10 ) IN ( 2 , 5 ) )
// AND ( ( KEY1 > ? ) OR ( KEY1 = ? AND KEY2 > ? ) )
// ORDER BY KEY1 , KEY2
// ) as nestedTab2 LIMIT 10;
Expand Down
32 changes: 16 additions & 16 deletions .../java/com/linkedin/datastream/common/databases/dbreader/TestMysqlChunkedQueryManager.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -27,13 +27,13 @@ public void testSimpleKeySinglePartition() {
* (
* SELECT * FROM TABLE
* ) nestedTab1
* WHERE ( MOD ( CONV ( MD5 ( CONCAT ( KEY1 ) ) , 16, 10 ) , 10 ) IN ( 3 ) )
* WHERE ( MOD ( CRC32 ( CONCAT ( KEY1 ) ) , 10 ) IN ( 3 ) )
* ORDER BY KEY1
* ) as nestedTab2 LIMIT 10;
*/
String firstExpected =
"SELECT * FROM ( SELECT * FROM ( SELECT * FROM TABLE ) nestedTab1 "
+ "WHERE ( MOD ( CONV ( MD5 ( CONCAT ( KEY1 ) ) , 16, 10 ) , 10 ) IN ( 3 ) ) ORDER BY KEY1 ) as nestedTab2 LIMIT 10";
+ "WHERE ( MOD ( CRC32 ( CONCAT ( KEY1 ) ) , 10 ) IN ( 3 ) ) ORDER BY KEY1 ) as nestedTab2 LIMIT 10";

/**
* SELECT * FROM
Expand All @@ -42,12 +42,12 @@ public void testSimpleKeySinglePartition() {
* (
* SELECT * FROM TABLE
* ) nestedTab1
* WHERE ( MOD ( CONV ( MD5 ( CONCAT ( KEY1 ) ) , 16, 10 ) , 10 ) IN ( 3 ) ) AND ( ( KEY1 > ? ) )
* WHERE ( MOD ( CRC32 ( CONCAT ( KEY1 ) ) , 10 ) IN ( 3 ) ) AND ( ( KEY1 > ? ) )
* ORDER BY KEY1
* ) as nestedTab2 LIMIT 10;
*/
String chunkedExpected = "SELECT * FROM ( SELECT * FROM ( SELECT * FROM TABLE ) nestedTab1 "
+ "WHERE ( MOD ( CONV ( MD5 ( CONCAT ( KEY1 ) ) , 16, 10 ) , 10 ) IN ( 3 ) ) AND ( ( KEY1 > ? ) ) ORDER BY KEY1 ) as nestedTab2 LIMIT 10";
+ "WHERE ( MOD ( CRC32 ( CONCAT ( KEY1 ) ) , 10 ) IN ( 3 ) ) AND ( ( KEY1 > ? ) ) ORDER BY KEY1 ) as nestedTab2 LIMIT 10";
testQueryString(MANAGER, firstExpected, chunkedExpected, NESTED_QUERY, KEY, CHUNK_SIZE, PARTITION_COUNT, PARTITION);
}

Expand All @@ -64,10 +64,10 @@ public void testSimpleKeyMultiPartition() {
* (
* SELECT * FROM TABLE
* ) nestedTab1
* WHERE ( MOD ( CONV ( MD5 ( CONCAT ( KEY1 ) ) , 16, 10 ) , 10 ) IN ( 2 , 5 ) ) ORDER BY KEY1
* WHERE ( MOD ( CRC32 ( CONCAT ( KEY1 ) ) , 10 ) IN ( 2 , 5 ) ) ORDER BY KEY1
* ) as nestedTab2 LIMIT 10;
*/
String firstExpected = "SELECT * FROM ( SELECT * FROM ( SELECT * FROM TABLE ) nestedTab1 WHERE ( MOD ( CONV ( MD5 ( CONCAT ( KEY1 ) ) , 16, 10 ) , 10 ) "
String firstExpected = "SELECT * FROM ( SELECT * FROM ( SELECT * FROM TABLE ) nestedTab1 WHERE ( MOD ( CRC32 ( CONCAT ( KEY1 ) ) , 10 ) "
+ "IN ( 2 , 5 ) ) ORDER BY KEY1 ) as nestedTab2 LIMIT 10";

/**
Expand All @@ -77,11 +77,11 @@ public void testSimpleKeyMultiPartition() {
* (
* SELECT * FROM TABLE
* ) nestedTab1
* WHERE ( MOD ( CONV ( MD5 ( CONCAT ( KEY1 ) ) , 16, 10 ) , 10 ) IN ( 2 , 5 ) ) AND ( ( KEY1 > ? ) ) ORDER BY KEY1
* WHERE ( MOD ( CRC32 ( CONCAT ( KEY1 ) ) , 10 ) IN ( 2 , 5 ) ) AND ( ( KEY1 > ? ) ) ORDER BY KEY1
* ) as nestedTab2 LIMIT 10;
*/
String chunkedExpected = "SELECT * FROM ( SELECT * FROM ( SELECT * FROM TABLE ) nestedTab1 "
+ "WHERE ( MOD ( CONV ( MD5 ( CONCAT ( KEY1 ) ) , 16, 10 ) , 10 ) IN ( 2 , 5 ) ) "
+ "WHERE ( MOD ( CRC32 ( CONCAT ( KEY1 ) ) , 10 ) IN ( 2 , 5 ) ) "
+ "AND ( ( KEY1 > ? ) ) ORDER BY KEY1 ) as nestedTab2 LIMIT 10";
testQueryString(MANAGER, firstExpected, chunkedExpected, NESTED_QUERY, KEY, CHUNK_SIZE, PARTITION_COUNT,
PARTITIONS);
Expand All @@ -96,12 +96,12 @@ public void testCompositeKeySinglePartition() {
* SELECT * FROM
* (
* SELECT * FROM TABLE
* ) nestedTab1 WHERE ( MOD ( CONV ( MD5 ( CONCAT ( KEY1 , KEY2 ) ) , 16, 10 ) , 10 ) IN ( 3 ) )
* ) nestedTab1 WHERE ( MOD ( CRC32 ( CONCAT ( KEY1 , KEY2 ) ) , 10 ) IN ( 3 ) )
* ORDER BY KEY1 , KEY2
* ) as nestedTab2 LIMIT 10;
*/
String firstExpected = "SELECT * FROM ( SELECT * FROM ( SELECT * FROM TABLE ) nestedTab1"
+ " WHERE ( MOD ( CONV ( MD5 ( CONCAT ( KEY1 , KEY2 ) ) , 16, 10 ) , 10 ) IN ( 3 ) ) ORDER BY KEY1 , KEY2 ) as nestedTab2 LIMIT 10";
+ " WHERE ( MOD ( CRC32 ( CONCAT ( KEY1 , KEY2 ) ) , 10 ) IN ( 3 ) ) ORDER BY KEY1 , KEY2 ) as nestedTab2 LIMIT 10";

/**
* SELECT * FROM
Expand All @@ -110,12 +110,12 @@ public void testCompositeKeySinglePartition() {
* (
* SELECT * FROM TABLE
* ) nestedTab1
* WHERE ( MOD ( CONV ( MD5 ( CONCAT ( KEY1 , KEY2 ) ) , 16, 10 ) , 10 ) IN ( 3 ) ) AND ( ( KEY1 > ? ) OR ( KEY1 = ? AND KEY2 > ? ) )
* WHERE ( MOD ( CRC32 ( CONCAT ( KEY1 , KEY2 ) ) , 10 ) IN ( 3 ) ) AND ( ( KEY1 > ? ) OR ( KEY1 = ? AND KEY2 > ? ) )
* ORDER BY KEY1 , KEY2
* ) as nestedTab2 LIMIT 10;
*/
String chunkedExpected = "SELECT * FROM ( SELECT * FROM ( SELECT * FROM TABLE ) nestedTab1 "
+ "WHERE ( MOD ( CONV ( MD5 ( CONCAT ( KEY1 , KEY2 ) ) , 16, 10 ) , 10 ) IN ( 3 ) ) AND ( ( KEY1 > ? ) OR ( KEY1 = ? AND KEY2 > ? ) ) "
+ "WHERE ( MOD ( CRC32 ( CONCAT ( KEY1 , KEY2 ) ) , 10 ) IN ( 3 ) ) AND ( ( KEY1 > ? ) OR ( KEY1 = ? AND KEY2 > ? ) ) "
+ "ORDER BY KEY1 , KEY2 ) as nestedTab2 LIMIT 10";
testQueryString(MANAGER, firstExpected, chunkedExpected, NESTED_QUERY, KEYS, CHUNK_SIZE, PARTITION_COUNT,
PARTITION);
Expand All @@ -130,12 +130,12 @@ public void testCompositeKeyMultiPartition() {
* SELECT * FROM
* (
* SELECT * FROM TABLE
* ) nestedTab1 WHERE ( MOD ( CONV ( MD5 ( CONCAT ( KEY1 , KEY2 ) ) , 16, 10 ) , 10 ) IN ( 2 , 5 ) )
* ) nestedTab1 WHERE ( MOD ( CRC32 ( CONCAT ( KEY1 , KEY2 ) ) , 10 ) IN ( 2 , 5 ) )
* ORDER BY KEY1 , KEY2
* ) as nestedTab2 LIMIT 10;
*/
String firstExpected = "SELECT * FROM ( SELECT * FROM ( SELECT * FROM TABLE ) nestedTab1 "
+ "WHERE ( MOD ( CONV ( MD5 ( CONCAT ( KEY1 , KEY2 ) ) , 16, 10 ) , 10 ) IN ( 2 , 5 ) ) "
+ "WHERE ( MOD ( CRC32 ( CONCAT ( KEY1 , KEY2 ) ) , 10 ) IN ( 2 , 5 ) ) "
+ "ORDER BY KEY1 , KEY2 ) as nestedTab2 LIMIT 10";

/**
Expand All @@ -145,13 +145,13 @@ public void testCompositeKeyMultiPartition() {
* SELECT * FROM
* (
* SELECT * FROM TABLE
* ) nestedTab1 WHERE ( MOD ( CONV ( MD5 ( CONCAT ( KEY1 , KEY2 ) ) , 16, 10 ) , 10 ) IN ( 2 , 5 ) )
* ) nestedTab1 WHERE ( MOD ( CRC32 ( CONCAT ( KEY1 , KEY2 ) ) , 10 ) IN ( 2 , 5 ) )
* AND ( ( KEY1 > ? ) OR ( KEY1 = ? AND KEY2 > ? ) )
* ORDER BY KEY1 , KEY2
* ) as nestedTab2 LIMIT 10;
*/
String chunkedExpected = "SELECT * FROM ( SELECT * FROM ( SELECT * FROM TABLE ) nestedTab1 "
+ "WHERE ( MOD ( CONV ( MD5 ( CONCAT ( KEY1 , KEY2 ) ) , 16, 10 ) , 10 ) IN ( 2 , 5 ) ) "
+ "WHERE ( MOD ( CRC32 ( CONCAT ( KEY1 , KEY2 ) ) , 10 ) IN ( 2 , 5 ) ) "
+ "AND ( ( KEY1 > ? ) OR ( KEY1 = ? AND KEY2 > ? ) ) ORDER BY KEY1 , KEY2 ) as nestedTab2 LIMIT 10";
testQueryString(MANAGER, firstExpected, chunkedExpected, NESTED_QUERY, KEYS, CHUNK_SIZE, PARTITION_COUNT,
PARTITIONS);
Expand Down