Fix diff_tables for non-threaded ('serial'); Fix for error reporting

Author: erezsh

data_diff/__main__.py

@@ -5,19 +5,18 @@
 import logging
 from itertools import islice
 
-from data_diff.tracking import disable_tracking
+import rich
+import click
 
-from .utils import remove_password_from_url, safezip, match_like
 
+from .utils import remove_password_from_url, safezip, match_like
 from .diff_tables import TableDiffer, DEFAULT_BISECTION_THRESHOLD, DEFAULT_BISECTION_FACTOR
 from .table_segment import create_schema, TableSegment
-
 from .databases.connect import connect
 from .parse_time import parse_time_before_now, UNITS_STR, ParseError
 from .config import apply_config_from_file
+from .tracking import disable_tracking
 
-import rich
-import click
 
 LOG_FORMAT = "[%(asctime)s] %(levelname)s - %(message)s"
 DATE_FORMAT = "%H:%M:%S"

data_diff/diff_tables.py

@@ -6,7 +6,7 @@
 from numbers import Number
 from operator import attrgetter, methodcaller
 from collections import defaultdict
-from typing import List, Tuple, Iterator, Optional
+from typing import Tuple, Iterator, Optional
 import logging
 from concurrent.futures import ThreadPoolExecutor, as_completed
 
@@ -95,6 +95,7 @@ def diff_tables(self, table1: TableSegment, table2: TableSegment) -> DiffResult:
 
         self.stats["diff_count"] = 0
         start = time.monotonic()
+        error = None
         try:
 
             # Query and validate schema
@@ -137,7 +138,6 @@ def diff_tables(self, table1: TableSegment, table2: TableSegment) -> DiffResult:
                 post_tables = [t.new(min_key=max_key1, max_key=max_key2) for t in (table1, table2)]
                 yield from self._bisect_and_diff_tables(*post_tables)
 
-            error = None
         except BaseException as e:  # Catch KeyboardInterrupt too
             error = e
         finally:
@@ -308,7 +308,8 @@ def _threaded_call(self, func, iterable):
 
     def _thread_as_completed(self, func, iterable):
         if not self.threaded:
-            return map(func, iterable)
+            yield from map(func, iterable)
+            return
 
         with ThreadPoolExecutor(max_workers=self.max_threadpool_size) as task_pool:
             futures = [task_pool.submit(func, item) for item in iterable]

tests/test_diff_tables.py

@@ -293,6 +293,18 @@ def test_diff_small_tables(self):
         self.assertEqual(2, self.differ.stats["table1_count"])
         self.assertEqual(1, self.differ.stats["table2_count"])
 
+    def test_non_threaded(self):
+        differ = TableDiffer(3, 4, threaded=False)
+
+        time = "2022-01-01 00:00:00"
+        time_str = f"timestamp '{time}'"
+        cols = "id userid movieid rating timestamp".split()
+        _insert_row(self.connection, self.table_src, cols, [1, 1, 1, 9, time_str])
+        _insert_rows(self.connection, self.table_dst, cols, [[1, 1, 1, 9, time_str]])
+        _commit(self.connection)
+        diff = list(differ.diff_tables(self.table, self.table2))
+        self.assertEqual(diff, [])
+
     def test_diff_table_above_bisection_threshold(self):
         time = "2022-01-01 00:00:00"
         time_str = f"timestamp '{time}'"