improve testing cache

Author: JohnMount

build/lib/data_algebra/eval_cache.py

@@ -0,0 +1,106 @@
+
+"""Cache for test evaluations"""
+
+from typing import Any, Dict, List, NamedTuple, Optional, Tuple
+import hashlib
+import data_algebra
+import data_algebra.db_model
+
+
+def hash_data_frame(d) -> str:
+    """
+    Get a hash code representing a data frame.
+
+    :param d: data frame
+    :return: hash code as a string
+    """
+    data_algebra.default_data_model.is_appropriate_data_instance(d)
+    hash_str = hashlib.sha256(
+        data_algebra.default_data_model.pd.util.hash_pandas_object(d).values
+    ).hexdigest()
+    return f'{d.shape}_{list(d.columns)}_{hash_str}'
+
+
+class EvalKey(NamedTuple):
+    """Carry description of data transform key"""
+    db_model_name: str
+    sql: str
+    dat_map_list: Tuple[Tuple[str, str], ...]
+
+
+def make_cache_key(
+        *,
+        db_model: data_algebra.db_model.DBModel,
+        sql: str,
+        data_map: Dict[str, Any],
+):
+    """
+    Create an immutable, hashable key.
+    """
+    assert isinstance(db_model, data_algebra.db_model.DBModel)
+    assert isinstance(sql, str)
+    assert isinstance(data_map, dict)
+    data_map_keys = list(data_map.keys())
+    data_map_keys.sort()
+    for k in data_map_keys:
+        assert isinstance(k, str)
+        assert data_algebra.default_data_model.is_appropriate_data_instance(data_map[k])
+    return EvalKey(
+        db_model_name=str(db_model),
+        sql=sql,
+        dat_map_list=tuple([(k, hash_data_frame(data_map[k])) for k in data_map_keys])
+    )
+
+
+class ResultCache:
+    """Cache for test results. Maps keys to data frames."""
+    dirty: bool
+    data_cache: Optional[Dict[str, Any]]
+    result_cache: Dict[EvalKey, Any]
+
+    def __init__(self):
+        self.dirty = False
+        self.data_cache = dict()
+        self.result_cache = dict()
+
+    def get(self,
+            *,
+            db_model: data_algebra.db_model.DBModel,
+            sql: str,
+            data_map: Dict[str, Any]):
+        """get result from cache, raise KeyError if not present"""
+        k = make_cache_key(
+            db_model=db_model,
+            sql=sql,
+            data_map=data_map)
+        res = self.result_cache[k]
+        assert data_algebra.default_data_model.is_appropriate_data_instance(res)
+        return res.copy()
+
+    def store(self,
+                *,
+                db_model: data_algebra.db_model.DBModel,
+                sql: str,
+                data_map: Dict[str, Any],
+                res) -> None:
+        """Store result to cache, mark dirty if change."""
+        assert data_algebra.default_data_model.is_appropriate_data_instance(res)
+        op_key = make_cache_key(
+            db_model=db_model,
+            sql=sql,
+            data_map=data_map)
+        try:
+            previous = self.result_cache[op_key]
+            if previous.equals(res):
+                return
+        except KeyError:
+            pass
+        self.dirty = True
+        self.result_cache[op_key] = res.copy()
+        # values saved for debugging
+        if self.dirty is not None:
+            for d in (list(data_map.values()) + [res]):
+                d_key = hash_data_frame(d)
+                # assuming no spurious key collisions
+                if d_key not in self.data_cache.keys():
+                    self.data_cache[d_key] = d.copy()

build/lib/data_algebra/test_util.py

@@ -3,13 +3,13 @@
 """
 
 import pickle
-import hashlib
 import traceback
 from typing import Any, Optional
 
 import numpy
 
 import data_algebra
+import data_algebra.eval_cache
 import data_algebra.db_model
 import data_algebra.SQLite
 import data_algebra.BigQuery
@@ -29,12 +29,12 @@
 run_direct_ops_path_tests = False
 
 # global test result cache
-global_test_result_cache = None
+global_test_result_cache: Optional[data_algebra.eval_cache.ResultCache] = None
 
 
 def re_parse(ops):
     """
-    Return copy of object made by dumpint to string via repr() and then evaluating that string.
+    Return copy of object made by dumping to string via repr() and then evaluating that string.
     """
     str1 = repr(ops)
     ops2 = eval(
@@ -167,30 +167,18 @@ def equivalent_frames(
     return True
 
 
-def hash_data_frame(d) -> str:
-    """
-    Get a hash code representing a data frame.
-
-    :param d: data frame
-    :return: hash code as a string
-    """
-    return hashlib.sha256(
-        data_algebra.default_data_model.pd.util.hash_pandas_object(d).values
-    ).hexdigest()
-
-
 def _run_handle_experiments(
     *,
     db_handle,
-    data: Dict,
+    data: Dict[str, Any],
     ops: ViewRepresentation,
     sql_statements: Iterable[str],
     expect,
     float_tol: float = 1e-8,
     check_column_order: bool = False,
     cols_case_sensitive: bool = False,
     check_row_order: bool = False,
-    test_result_cache: Optional[dict] = None,
+    test_result_cache: Optional[data_algebra.eval_cache.ResultCache] = None,
     alter_cache: bool = True,
     test_direct_ops_path=False,
 ):
@@ -203,26 +191,19 @@ def _run_handle_experiments(
     assert db_handle.conn is not None
     if isinstance(db_handle.db_model, data_algebra.SQLite.SQLiteModel):
         test_direct_ops_path = True
-    db_handle_key = str(db_handle.db_model)
     sql_statements = list(sql_statements)
     res_db_sql = list([None] * len(sql_statements))  # extra list() wrapper for PyCharm's type checker
     res_db_ops = None
     need_to_run = True
-    dict_keys = list(data.keys())
-    dict_keys.sort()
-    data_key = " ".join([k + ":" + hash_data_frame(data[k]) for k in dict_keys])
-
-    def mk_key(ii):
-        """
-        Build sql statement key.
-        """
-        return db_handle_key + " " + sql_statements[ii] + " " + data_key
-
     # inspect result cache for any prior results
     if test_result_cache is not None:
         for i in range(len(sql_statements)):
             try:
-                res_db_sql[i] = test_result_cache[mk_key(i)].copy()
+                res_db_sql[i] = test_result_cache.get(
+                    db_model=db_handle.db_model,
+                    sql=sql_statements[i],
+                    data_map=data,
+                )
             except KeyError:
                 pass
         need_to_run = test_direct_ops_path or numpy.any(
@@ -248,7 +229,12 @@ def mk_key(ii):
                         and (test_result_cache is not None)
                         and (res_db_sql_i is not None)
                     ):
-                        test_result_cache[mk_key(i)] = res_db_sql_i.copy()
+                        test_result_cache.store(
+                            db_model=db_handle.db_model,
+                            sql=sql_statements[i],
+                            data_map=data,
+                            res=res_db_sql_i,
+                        )
         except AssertionError as ase:
             traceback.print_exc()
             caught = ase