Merge pull request #303 from datafold/nov19_better_errors

CLI: Better errors + tiny bugfix

Author: erezsh

data_diff/__main__.py

@@ -45,14 +45,25 @@ def _get_schema(pair):
     return db.query_table_schema(table_path)
 
 
-def diff_schemas(schema1, schema2, columns):
+def diff_schemas(table1, table2, schema1, schema2, columns):
     logging.info("Diffing schemas...")
     attrs = "name", "type", "datetime_precision", "numeric_precision", "numeric_scale"
     for c in columns:
         if c is None:  # Skip for convenience
             continue
         diffs = []
-        for attr, v1, v2 in safezip(attrs, schema1[c], schema2[c]):
+
+        if c not in schema1:
+            cols = ', '.join(schema1)
+            raise ValueError(f"Column '{c}' not found in table 1, named '{table1}'. Columns: {cols}")
+        if c not in schema2:
+            cols = ', '.join(schema1)
+            raise ValueError(f"Column '{c}' not found in table 2, named '{table2}'. Columns: {cols}")
+
+        col1 = schema1[c]
+        col2 = schema2[c]
+
+        for attr, v1, v2 in safezip(attrs, col1, col2):
             if v1 != v2:
                 diffs.append(f"{attr}:({v1} != {v2})")
         if diffs:
@@ -197,7 +208,13 @@ def main(conf, run, **kw):
     if kw["algorithm"] == Algorithm.AUTO:
         kw["algorithm"] = Algorithm.JOINDIFF if indb_syntax else Algorithm.HASHDIFF
 
-    return _main(**kw)
+    try:
+        return _main(**kw)
+    except Exception as e:
+        logging.error(e)
+        if kw["debug"]:
+            raise
+
 
 
 def _main(
@@ -357,6 +374,8 @@ def _main(
 
     if db1 is db2:
         diff_schemas(
+            table_names[0],
+            table_names[1],
             schema1,
             schema2,
             (

data_diff/diff_tables.py

@@ -132,8 +132,9 @@ def _diff_tables_wrapper(self, table1: TableSegment, table2: TableSegment, info_
 
             if is_tracking_enabled():
                 runtime = time.monotonic() - start
-                table1_count = info_tree.info.rowcounts[1]
-                table2_count = info_tree.info.rowcounts[2]
+                rowcounts = info_tree.info.rowcounts
+                table1_count = rowcounts[1] if rowcounts else None
+                table2_count = rowcounts[2] if rowcounts else None
                 diff_count = info_tree.info.diff_count
                 err_message = truncate_error(repr(error))
                 event_json = create_end_event_json(

data_diff/joindiff_tables.py

@@ -216,6 +216,7 @@ def _test_duplicate_keys(self, table1: TableSegment, table2: TableSegment):
 
             unvalidated = list(set(key_columns) - set(unique))
             if unvalidated:
+                logger.info(f"Validating that the are no duplicate keys in columns: {unvalidated}")
                 # Validate that there are no duplicate keys
                 self.stats["validated_unique_keys"] = self.stats.get("validated_unique_keys", []) + [unvalidated]
                 q = t.select(total=Count(), total_distinct=Count(Concat(this[unvalidated]), distinct=True))
@@ -237,7 +238,7 @@ def _test_null_keys(self, table1, table2):
                 raise ValueError("NULL values in one or more primary keys")
 
     def _collect_stats(self, i, table_seg: TableSegment, info_tree: InfoTree):
-        logger.info(f"Collecting stats for table #{i}")
+        logger.debug(f"Collecting stats for table #{i}")
         db = table_seg.database
 
         # Metrics
@@ -305,7 +306,7 @@ def _create_outer_join(self, table1, table2):
         return diff_rows, a_cols, b_cols, is_diff_cols, all_rows
 
     def _count_diff_per_column(self, db, diff_rows, cols, is_diff_cols):
-        logger.info("Counting differences per column")
+        logger.debug("Counting differences per column")
         is_diff_cols_counts = db.query(diff_rows.select(sum_(this[c]) for c in is_diff_cols), tuple)
         diff_counts = {}
         for name, count in safezip(cols, is_diff_cols_counts):
@@ -319,7 +320,7 @@ def _sample_and_count_exclusive(self, db, diff_rows, a_cols, b_cols):
             exclusive_rows_query = diff_rows.where(this.is_exclusive_a | this.is_exclusive_b)
 
         if not self.sample_exclusive_rows:
-            logger.info("Counting exclusive rows")
+            logger.debug("Counting exclusive rows")
             self.stats["exclusive_count"] = db.query(exclusive_rows_query.count(), int)
             return