WL7955: Add support for RANGE_DATETIME Fabric sharding type

Geert Vanderkelen · Geert Vanderkelen · commit dd8f9d2da91a · 2014-08-28T10:04:42.000+02:00
We add support for the RANGE_DATETIME Fabric sharding type which
can shard based on a datetime.datetime or datetime.date object.
A ValueError is raised when the key has not valid type.

Test cases were updated in tests.test_fabric. Sharding tests need
a running Fabric instance with tables already created. They will
only run when --with-fabric option has been provided to
unittests.py.
For the RANGE_DATETIME test, the empty employees_range_datetime
is required based on the employees.employees table using lower
bounds as "group1/1980-01-01, group2/2000-01-01".

The fabric.Fabric class has an additional method execute()
which takes a group, a command and arguments. This makes
it possible to run any command available in Fabric. See the
test cases for example usage.
diff --git a/lib/mysql/connector/fabric/__init__.py b/lib/mysql/connector/fabric/__init__.py
@@ -50,7 +50,8 @@
     STATUS_PRIMARY, STATUS_SECONDARY,
     SCOPE_GLOBAL, SCOPE_LOCAL,
     Fabric, FabricConnection,
-    MySQLFabricConnection
+    MySQLFabricConnection,
+    FabricSet,
 )
 
 
@@ -71,4 +72,5 @@ def connect(**kwargs):
     'Fabric',
     'FabricConnection',
     'MySQLFabricConnection',
+    'FabricSet',
 ]
diff --git a/lib/mysql/connector/fabric/caching.py b/lib/mysql/connector/fabric/caching.py
@@ -78,7 +78,7 @@ class CacheShardTable(CacheEntry):
 
     def __init__(self, shard, version=None, fabric_uuid=None):
         if not isinstance(shard, FabricShard):
-            ValueError("shard argument must be a FabricShard instance")
+            raise ValueError("shard argument must be a FabricShard instance")
         super(CacheShardTable, self).__init__(version=version,
                                               fabric_uuid=fabric_uuid)
         self.partitioning = {}
@@ -94,6 +94,21 @@ def add_partition(self, key, group):
         """Add sharding information for a group"""
         if self.shard_type == 'RANGE':
             key = int(key)
+        elif self.shard_type == 'RANGE_DATETIME':
+            try:
+                if ':' in key:
+                    key = datetime.strptime(key, "%Y-%m-%d %H:%M:%S")
+                else:
+                    key = datetime.strptime(key, "%Y-%m-%d").date()
+            except:
+                raise ValueError(
+                    "RANGE_DATETIME key could not be parsed, was: {0}".format(
+                        key
+                    ))
+        else:
+            raise ValueError("Unsupported sharding type {0}".format(
+                self.shard_type
+            ))
         self.partitioning[key] = {
             'group': group,
         }
diff --git a/lib/mysql/connector/fabric/connection.py b/lib/mysql/connector/fabric/connection.py
@@ -24,6 +24,7 @@
 """Implementing communication with MySQL Fabric"""
 
 import sys
+import datetime
 import time
 import uuid
 from base64 import b16decode
@@ -121,7 +122,8 @@
 _CNX_PROPERTIES = {
     # name: ((valid_types), description, default)
     'group': ((str,), "Name of group of servers", None),
-    'key': ((int, str), "Sharding key", None),
+    'key': ((int, str, datetime.datetime, datetime.date),
+            "Sharding key", None),
     'tables': ((tuple, list), "List of tables in query", None),
     'mode': ((int,), "Read-Only, Write-Only or Read-Write", MODE_READWRITE),
     'shard': ((str,), "Identity of the shard for direct connection", None),
@@ -753,6 +755,17 @@ def get_shard_server(self, tables, key, scope=SCOPE_LOCAL, mode=None):
                 partitions = sorted(entry.partitioning.keys())
                 index = partitions[bisect(partitions, int(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:
+                    if key >= partkey:
+                        index = partkey
+                        break
+                partition = entry.partitioning[index]
             elif entry.shard_type == 'HASH':
                 md5key = md5(str(key))
                 partition_keys = sorted(
@@ -774,6 +787,36 @@ def get_shard_server(self, tables, key, scope=SCOPE_LOCAL, mode=None):
 
         return self.get_group_server(groups[0], mode=mode)
 
+    def execute(self, group, command, *args, **kwargs):
+        """Execute a Fabric command from given group
+
+        This method will execute the given Fabric command from the given group
+        using the given arguments. It returns an instance of FabricSet.
+
+        Raises ValueError when group.command is not valid and raises
+        InterfaceError when an error occurs while executing.
+
+        Returns FabricSet.
+        """
+        inst = self.get_instance()
+        try:
+            grp = getattr(inst.proxy, group)
+            cmd = getattr(grp, command)
+        except AttributeError as exc:
+            raise ValueError("{group}.{command} not available ({err})".format(
+                group=group, command=command, err=str(exc)))
+
+        fab_set = None
+        try:
+            data = cmd(*args, **kwargs)
+            fab_set = FabricSet(data)
+        except (Fault, socket.error, InterfaceError) as exc:
+            msg = "Executing {group}.{command} failed: {error}".format(
+                group=group, command=command, error=str(exc))
+            raise InterfaceError(msg)
+
+        return fab_set
+
 
 class FabricConnection(object):
 
diff --git a/tests/__init__.py b/tests/__init__.py
@@ -108,6 +108,7 @@
 TEST_BUILD_DIR = None
 
 DJANGO_VERSION = None
+FABRIC_CONFIG = None
 
 __all__ = [
     'MySQLConnectorTests',
@@ -205,9 +206,6 @@ def get_test_modules():
             # Skip django testing completely when Django is not available.
             LOGGER.warning("Django tests will not run: Django not available")
             continue
-        if 'fabric' in module:
-            LOGGER.warning("Fabric tests are disabled in full run")
-            continue
         testcases.append(
             'tests.{module}'.format(module=module))
         LOGGER.debug('Added tests.{module}'.format(module=module))
diff --git a/tests/test_fabric.py b/tests/test_fabric.py
diff --git a/unittests.py b/unittests.py

Original file line number	Diff line number	Diff line change
`@@ -50,7 +50,8 @@`
`50`	`50`	`STATUS_PRIMARY, STATUS_SECONDARY,`
`51`	`51`	`SCOPE_GLOBAL, SCOPE_LOCAL,`
`52`	`52`	`Fabric, FabricConnection,`
`53`		`- MySQLFabricConnection`
	`53`	`+ MySQLFabricConnection,`
	`54`	`+ FabricSet,`
`54`	`55`	`)`
`55`	`56`
`56`	`57`
`@@ -71,4 +72,5 @@ def connect(**kwargs):`
`71`	`72`	`'Fabric',`
`72`	`73`	`'FabricConnection',`
`73`	`74`	`'MySQLFabricConnection',`
	`75`	`+ 'FabricSet',`
`74`	`76`	`]`