BUG19642249: Improve errors reporting invalid sharding keys

Geert Vanderkelen · Geert Vanderkelen · commit fae86c03d56f · 2014-10-10T20:21:18.000+02:00
We improve the errors reported when invalid sharding keys are given for RANGE_STRING and RANGE_DATETIME types. We additionally improve performance by moving the sorting of partition keys when saving cache entries. Previously, sorting was done each time a partition was looked up. Unit test for BUG#19642249 was added in test_fabric module. (cherry picked from commit a95e9c6)
diff --git a/lib/mysql/connector/fabric/caching.py b/lib/mysql/connector/fabric/caching.py
@@ -24,6 +24,7 @@
 """Implementing caching mechanisms for MySQL Fabric"""
 
 
+import bisect
 from datetime import datetime, timedelta
 from hashlib import sha1
 import logging
@@ -35,6 +36,24 @@
 _CACHE_TTL = 1 * 60  # 1 minute
 
 
+def insort_right_rev(a, x, lo=0, hi=None):
+    """Similar to bisect.insort_right but for reverse sorted lists
+
+    This code is similar to the Python code found in Lib/bisect.py.
+    We simply change the comparison from 'less than' to 'greater than'.
+    """
+
+    if lo < 0:
+        raise ValueError('lo must be non-negative')
+    if hi is None:
+        hi = len(a)
+    while lo < hi:
+        mid = (lo+hi)//2
+        if x > a[mid]: hi = mid
+        else: lo = mid+1
+    a.insert(lo, x)
+
+
 class CacheEntry(object):
 
     """Base class for MySQL Fabric cache entries"""
@@ -83,6 +102,8 @@ def __init__(self, shard, version=None, fabric_uuid=None):
                                               fabric_uuid=fabric_uuid)
         self.partitioning = {}
         self._shard = shard
+        self.keys = []
+        self.keys_reversed = []
 
         if shard.key and shard.group:
             self.add_partition(shard.key, shard.group)
@@ -115,6 +136,8 @@ def add_partition(self, key, group):
             'group': group,
         }
         self.reset_ttl()
+        bisect.insort_right(self.keys, key)
+        insort_right_rev(self.keys_reversed, key)
 
     @classmethod
     def hash_index(cls, part1, part2=None):
diff --git a/lib/mysql/connector/fabric/connection.py b/lib/mysql/connector/fabric/connection.py
@@ -753,36 +753,43 @@ def get_shard_server(self, tables, key, scope=SCOPE_LOCAL, mode=None):
                 return self.get_group_server(entry.global_group, mode=mode)
 
             if entry.shard_type == 'RANGE':
-                partitions = sorted(entry.partitioning.keys())
-                index = partitions[bisect(partitions, int(key)) - 1]
+                try:
+                    range_key = int(key)
+                except ValueError:
+                    raise ValueError("Key must be an integer for RANGE")
+                partitions = entry.keys
+                index = partitions[bisect(partitions, range_key) - 1]
                 partition = entry.partitioning[index]
             elif entry.shard_type == 'RANGE_DATETIME':
                 if not isinstance(key, (datetime.date, datetime.datetime)):
                     raise ValueError(
                         "Key must be datetime.date or datetime.datetime for "
                         "RANGE_DATETIME")
-                partition_keys = sorted(entry.partitioning.keys(), reverse=True)
-                for partkey in partition_keys:
+                index = None
+                for partkey in entry.keys_reversed:
                     if key >= partkey:
                         index = partkey
                         break
-                partition = entry.partitioning[index]
+                try:
+                    partition = entry.partitioning[index]
+                except KeyError:
+                    raise ValueError("Key invalid; was '{0}'".format(key))
             elif entry.shard_type == 'RANGE_STRING':
                 if not isunicode(key):
                     raise ValueError("Key must be a unicode value")
-                partition_keys = sorted(entry.partitioning.keys(), reverse=True)
-                for partkey in partition_keys:
-                    size = len(partkey)
-                    if key[0:size] >= partkey[0:size]:
+                index = None
+                for partkey in entry.keys_reversed:
+                    if key >= partkey:
                         index = partkey
                         break
-                partition = entry.partitioning[index]
+                try:
+                    partition = entry.partitioning[index]
+                except KeyError:
+                    raise ValueError("Key invalid; was '{0}'".format(key))
             elif entry.shard_type == 'HASH':
                 md5key = md5(str(key))
-                partition_keys = sorted(
-                    entry.partitioning.keys(), reverse=True)
-                index = partition_keys[-1]
-                for partkey in partition_keys:
+                index = entry.keys_reversed[-1]
+                for partkey in entry.keys_reversed:
                     if md5key.digest() >= b16decode(partkey):
                         index = partkey
                         break
diff --git a/tests/test_fabric.py b/tests/test_fabric.py
@@ -424,6 +424,43 @@ def _truncate(self, cur, table):
         cur.execute("SELECT @@global.gtid_executed")
         return cur.fetchone()[0]
 
+    def test_range(self):
+        self.assertTrue(self._check_table("employees.employees_range", 'RANGE'))
+        tbl_name = "employees_range"
+
+        tables = ["employees.{0}".format(tbl_name)]
+
+        self.cnx.set_property(tables=tables,
+                              scope=fabric.SCOPE_GLOBAL,
+                              mode=fabric.MODE_READWRITE)
+        cur = self.cnx.cursor()
+        gtid_executed = self._truncate(cur, tbl_name)
+        self.cnx.commit()
+
+        insert = ("INSERT INTO {0} "
+                  "VALUES (%s, %s, %s, %s, %s, %s)").format(tbl_name)
+
+        self._populate(self.cnx, gtid_executed, tbl_name, insert,
+                       self.emp_data[1985] + self.emp_data[2000], 0)
+
+        time.sleep(2)
+
+        # Year is key of self.emp_data, second value is emp_no for RANGE key
+        exp_keys = [(1985, 10002), (2000, 47291)]
+        for year, emp_no in exp_keys:
+            self.cnx.set_property(tables=tables,
+                                  scope=fabric.SCOPE_LOCAL,
+                                  key=emp_no, mode=fabric.MODE_READONLY)
+            cur = self.cnx.cursor()
+            cur.execute("SELECT * FROM {0}".format(tbl_name))
+            rows = cur.fetchall()
+            self.assertEqual(rows, self.emp_data[year])
+
+        self.cnx.set_property(tables=tables,
+                              key='spam', mode=fabric.MODE_READONLY)
+        self.assertRaises(ValueError, self.cnx.cursor)
+
+
     def test_range_datetime(self):
         self.assertTrue(self._check_table(
             "employees.employees_range_datetime", 'RANGE_DATETIME'))
@@ -510,3 +547,34 @@ def test_range_string(self):
                                   key='not unicode str',
                                   mode=fabric.MODE_READONLY)
             self.assertRaises(ValueError, self.cnx.cursor)
+
+    def test_bug19642249(self):
+        self.assertTrue(self._check_table(
+            "employees.employees_range_string", 'RANGE_STRING'))
+
+        # Invalid key for RANGE_STRING
+        tbl_name = "employees_range_string"
+        tables = ["employees.{0}".format(tbl_name)]
+
+        self.cnx.set_property(tables=tables,
+                              key=u'1', mode=fabric.MODE_READONLY)
+        try:
+            cur = self.cnx.cursor()
+        except ValueError as exc:
+            self.assertEqual("Key invalid; was '1'", str(exc))
+        else:
+            self.fail("ValueError not raised")
+
+        # Invalid key for RANGE_DATETIME
+        tbl_name = "employees_range_datetime"
+        tables = ["employees.{0}".format(tbl_name)]
+
+        self.cnx.set_property(tables=tables,
+                              key=datetime.date(1977, 1, 1),
+                              mode=fabric.MODE_READONLY)
+        try:
+            cur = self.cnx.cursor()
+        except ValueError as exc:
+            self.assertEqual("Key invalid; was '1977-01-01'", str(exc))
+        else:
+            self.fail("ValueError not raised")
