Merge pull request #353 from smart-on-fhir/mikix/extra-completion-info

Add ETL timestamp to completion table, make it non-unique

Author: mikix

cumulus_etl/completion/schema.py

@@ -13,7 +13,11 @@ def completion_format_args() -> dict:
     """Returns kwargs to pass to the Format class initializer of your choice"""
     return {
         "dbname": COMPLETION_TABLE,
-        "uniqueness_fields": {"table_name", "group_name"},
+        # These fields altogether basically guarantee that we never collide.
+        # (i.e. that every 'merge' is really an 'insert')
+        # That's intentional - we want this table to be a bit of a historical log.
+        # (We couldn't have no uniqueness fields -- delta lake doesn't like that.)
+        "uniqueness_fields": {"table_name", "group_name", "export_time", "etl_time"},
     }
 
 
@@ -47,6 +51,8 @@ def completion_schema() -> pyarrow.Schema:
             pyarrow.field("export_time", pyarrow.string()),
             pyarrow.field("export_url", pyarrow.string()),
             pyarrow.field("etl_version", pyarrow.string()),
+            # See note above for why this isn't a pyarrow.timestamp() field.
+            pyarrow.field("etl_time", pyarrow.string()),
         ]
     )
 

cumulus_etl/etl/config.py

@@ -44,7 +44,7 @@ def __init__(
         self._output_format = output_format
         self.dir_errors = dir_errors
         self.client = client
-        self.timestamp = common.timestamp_filename(timestamp)
+        self.timestamp = timestamp
         self.hostname = gethostname()
         self.comment = comment or ""
         self.batch_size = batch_size
@@ -67,7 +67,8 @@ def path_config(self) -> str:
         return os.path.join(self.dir_job_config(), "job_config.json")
 
     def dir_job_config(self) -> str:
-        path = self._output_root.joinpath(f"JobConfig/{self.timestamp}")
+        timestamp_dir = common.timestamp_filename(self.timestamp)
+        path = self._output_root.joinpath(f"JobConfig/{timestamp_dir}")
         self._output_root.makedirs(path)
         return path
 

cumulus_etl/etl/tasks/base.py

@@ -272,6 +272,7 @@ def _update_completion_table(self) -> None:
                     "export_time": self.task_config.export_datetime.isoformat(),
                     "export_url": self.task_config.export_url,
                     "etl_version": cumulus_etl.__version__,
+                    "etl_time": self.task_config.timestamp.isoformat(),
                 }
                 for output in self.outputs
                 if not output.get_name(self).startswith("etl__")

tests/data/covid/output/etl__completion/etl__completion.000.ndjson

@@ -1 +1 @@
-{"table_name": "covid_symptom__nlp_results", "group_name": "test-group", "export_time": "2020-10-13T12:00:20-05:00", "etl_version": "1.0.0+test"}
+{"table_name": "covid_symptom__nlp_results", "group_name": "test-group", "export_time": "2020-10-13T12:00:20-05:00", "etl_version": "1.0.0+test", "etl_time": "2021-09-14T21:23:45+00:00"}

tests/data/covid/term-exists/etl__completion/etl__completion.000.ndjson

@@ -1 +1 @@
-{"table_name": "covid_symptom__nlp_results_term_exists", "group_name": "test-group", "export_time": "2020-10-13T12:00:20-05:00", "etl_version": "1.0.0+test"}
+{"table_name": "covid_symptom__nlp_results_term_exists", "group_name": "test-group", "export_time": "2020-10-13T12:00:20-05:00", "etl_version": "1.0.0+test", "etl_time": "2021-09-14T21:23:45+00:00"}

tests/data/hftest/output/etl__completion/etl__completion.000.ndjson

@@ -1 +1 @@
-{"table_name": "hftest__summary", "group_name": "test-group", "export_time": "2020-10-13T12:00:20-05:00", "etl_version": "1.0.0+test"}
+{"table_name": "hftest__summary", "group_name": "test-group", "export_time": "2020-10-13T12:00:20-05:00", "etl_version": "1.0.0+test", "etl_time": "2021-09-14T21:23:45+00:00"}

tests/data/i2b2/output/etl__completion/etl__completion.000.ndjson

@@ -1 +1 @@
-{"table_name": "encounter", "group_name": "test-group", "export_time": "2020-10-13T12:00:20-05:00", "etl_version": "1.0.0+test"}
+{"table_name": "encounter", "group_name": "test-group", "export_time": "2020-10-13T12:00:20-05:00", "etl_version": "1.0.0+test", "etl_time": "2021-09-14T21:23:45+00:00"}

tests/data/i2b2/output/etl__completion/etl__completion.001.ndjson

@@ -1 +1 @@
-{"table_name": "patient", "group_name": "test-group", "export_time": "2020-10-13T12:00:20-05:00", "etl_version": "1.0.0+test"}
+{"table_name": "patient", "group_name": "test-group", "export_time": "2020-10-13T12:00:20-05:00", "etl_version": "1.0.0+test", "etl_time": "2021-09-14T21:23:45+00:00"}

tests/data/i2b2/output/etl__completion/etl__completion.002.ndjson

@@ -1 +1 @@
-{"table_name": "condition", "group_name": "test-group", "export_time": "2020-10-13T12:00:20-05:00", "etl_version": "1.0.0+test"}
+{"table_name": "condition", "group_name": "test-group", "export_time": "2020-10-13T12:00:20-05:00", "etl_version": "1.0.0+test", "etl_time": "2021-09-14T21:23:45+00:00"}

tests/data/i2b2/output/etl__completion/etl__completion.003.ndjson

@@ -1 +1 @@
-{"table_name": "documentreference", "group_name": "test-group", "export_time": "2020-10-13T12:00:20-05:00", "etl_version": "1.0.0+test"}
+{"table_name": "documentreference", "group_name": "test-group", "export_time": "2020-10-13T12:00:20-05:00", "etl_version": "1.0.0+test", "etl_time": "2021-09-14T21:23:45+00:00"}