Merge pull request #134 from alexa-infra/fix-write-hook

Fix write-hook compatiblity for 2.3.6 (import/export)

Author: gabor-boros

rethinkdb/_export.py

@@ -234,6 +234,8 @@ def export_table(db, table, directory, options, error_queue, progress_info, sind
 
     writer = None
 
+    has_write_hooks = utils_common.check_minimum_version(options, '2.3.7', False)
+
     try:
         # -- get table info
 
@@ -248,13 +250,14 @@ def export_table(db, table, directory, options, error_queue, progress_info, sind
 
         sindex_counter.value += len(table_info["indexes"])
 
-        table_info['write_hook'] = options.retryQuery(
-            'table write hook data %s.%s' % (db, table),
-            query.db(db).table(table).get_write_hook(),
-            run_options={'binary_format': 'raw'})
+        if has_write_hooks:
+            table_info['write_hook'] = options.retryQuery(
+                'table write hook data %s.%s' % (db, table),
+                query.db(db).table(table).get_write_hook(),
+                run_options={'binary_format': 'raw'})
 
-        if table_info['write_hook'] is not None:
-            hook_counter.value += 1
+            if table_info['write_hook'] is not None:
+                hook_counter.value += 1
 
         with open(os.path.join(directory, db, table + '.info'), 'w') as info_file:
             info_file.write(json.dumps(table_info) + "\n")

rethinkdb/_import.py

@@ -301,13 +301,13 @@ def restore_indexes(self, warning_queue):
                 exception_type, exception_class, trcback = sys.exc_info()
                 warning_queue.put((exception_type, exception_class, traceback.extract_tb(trcback), self._source.name))
 
-        self.query_runner(
-            "Write hook from: %s.%s" %
-            (self.db, self.table), query.db(
-                self.db).table(
-                self.table).get_write_hook())
-        try:
-            if self.write_hook:
+        if self.write_hook:
+            self.query_runner(
+                "Write hook from: %s.%s" %
+                (self.db, self.table), query.db(
+                    self.db).table(
+                    self.table).get_write_hook())
+            try:
                 self.query_runner(
                     "drop hook: %s.%s" % (self.db, self.table),
                     query.db(self.db).table(self.table).set_write_hook(None)
@@ -316,9 +316,9 @@ def restore_indexes(self, warning_queue):
                     "create hook: %s.%s:%s" % (self.db, self.table, self.write_hook),
                     query.db(self.db).table(self.table).set_write_hook(self.write_hook["function"])
                 )
-        except RuntimeError:
-            exception_type, exception_class, trcback = sys.exc_info()
-            warning_queue.put((exception_type, exception_class, traceback.extract_tb(trcback), self._source.name))
+            except RuntimeError:
+                exception_type, exception_class, trcback = sys.exc_info()
+                warning_queue.put((exception_type, exception_class, traceback.extract_tb(trcback), self._source.name))
 
     def batches(self, batch_size=None, warning_queue=None):
 
@@ -1339,6 +1339,7 @@ def parse_sources(options, files_ignored=None):
     def parse_info_file(path):
         primary_key = None
         indexes = []
+        write_hook = None
         with open(path, 'r') as info_file:
             metadata = json.load(info_file)
             if "primary_key" in metadata:
@@ -1349,6 +1350,8 @@ def parse_info_file(path):
                 write_hook = metadata["write_hook"]
         return primary_key, indexes, write_hook
 
+    has_write_hooks = utils_common.check_minimum_version(options, '2.3.7', False)
+
     sources = set()
     if files_ignored is None:
         files_ignored = []
@@ -1382,6 +1385,8 @@ def parse_info_file(path):
                 indexes = info_indexes
             if write_hook is None:
                 write_hook = info_write_hook
+            if write_hook and not has_write_hooks:
+                raise Exception('this RDB version doesn\'t support write-hooks')
 
         sources.add(
             table_type(
@@ -1446,6 +1451,8 @@ def parse_info_file(path):
                             files_ignored.append(os.path.join(root, filename))
                         else:
                             primary_key, indexes, write_hook = parse_info_file(info_path)
+                        if write_hook and not has_write_hooks:
+                            raise Exception('RDB versions below doesn\'t support write-hooks')
 
                         table_type = None
                         if ext == ".json":

rethinkdb/utils_common.py

@@ -124,7 +124,7 @@ def print_progress(ratio, indent=0, read=None, write=None):
     sys.stdout.flush()
 
 
-def check_minimum_version(options, minimum_version='1.6'):
+def check_minimum_version(options, minimum_version='1.6', raise_exception=True):
     minimum_version = distutils.version.LooseVersion(minimum_version)
     version_string = options.retryQuery('get server version', query.db(
         'rethinkdb').table('server_status')[0]['process']['version'])
@@ -135,7 +135,10 @@ def check_minimum_version(options, minimum_version='1.6'):
         raise RuntimeError("invalid version string format: %s" % version_string)
 
     if distutils.version.LooseVersion(matches.group('version')) < minimum_version:
-        raise RuntimeError("Incompatible version, expected >= %s got: %s" % (minimum_version, version_string))
+        if raise_exception:
+            raise RuntimeError("Incompatible version, expected >= %s got: %s" % (minimum_version, version_string))
+        return False
+    return True
 
 
 DbTable = collections.namedtuple('DbTable', ['db', 'table'])