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

 The existing MD5-based partitioning strategy caused chunking and bootstrapping of large tables to be extremely slow and prone to connection timeouts. The partitioning strategy has been updated to use CRC32 instead of MD5 for MySQL bootstrapping.

Author: akshayrai

datastream-common/src/main/java/com/linkedin/datastream/common/databases/dbreader/MySqlChunkedQueryManager.java

@@ -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();
@@ -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) {
@@ -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;

datastream-common/src/test/java/com/linkedin/datastream/common/databases/dbreader/TestMysqlChunkedQueryManager.java

@@ -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
@@ -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);
   }
 
@@ -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";
 
     /**
@@ -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);
@@ -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
@@ -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);
@@ -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";
 
     /**
@@ -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);