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

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

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"}

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

@@ -1,2 +1,2 @@
-{"table_name": "medication", "group_name": "test-group", "export_time": "2020-10-13T12:00:20-05:00", "etl_version": "1.0.0+test"}
-{"table_name": "medicationrequest", "group_name": "test-group", "export_time": "2020-10-13T12:00:20-05:00", "etl_version": "1.0.0+test"}
+{"table_name": "medication", "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"}
+{"table_name": "medicationrequest", "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.005.ndjson

@@ -1 +1 @@
-{"table_name": "observation", "group_name": "test-group", "export_time": "2020-10-13T12:00:20-05:00", "etl_version": "1.0.0+test"}
+{"table_name": "observation", "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/simple/batched-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/simple/batched-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/simple/batched-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/simple/batched-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"}

tests/data/simple/batched-output/etl__completion/etl__completion.004.ndjson

@@ -1,2 +1,2 @@
-{"table_name": "medication", "group_name": "test-group", "export_time": "2020-10-13T12:00:20-05:00", "etl_version": "1.0.0+test"}
-{"table_name": "medicationrequest", "group_name": "test-group", "export_time": "2020-10-13T12:00:20-05:00", "etl_version": "1.0.0+test"}
+{"table_name": "medication", "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"}
+{"table_name": "medicationrequest", "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/simple/batched-output/etl__completion/etl__completion.005.ndjson

@@ -1 +1 @@
-{"table_name": "observation", "group_name": "test-group", "export_time": "2020-10-13T12:00:20-05:00", "etl_version": "1.0.0+test"}
+{"table_name": "observation", "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/simple/batched-output/etl__completion/etl__completion.006.ndjson

@@ -1 +1 @@
-{"table_name": "procedure", "group_name": "test-group", "export_time": "2020-10-13T12:00:20-05:00", "etl_version": "1.0.0+test"}
+{"table_name": "procedure", "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/simple/batched-output/etl__completion/etl__completion.007.ndjson

@@ -1 +1 @@
-{"table_name": "servicerequest", "group_name": "test-group", "export_time": "2020-10-13T12:00:20-05:00", "etl_version": "1.0.0+test"}
+{"table_name": "servicerequest", "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/simple/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/simple/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/simple/output/etl__completion/etl__completion.002.ndjson

@@ -1 +1 @@
-{"table_name": "allergyintolerance", "group_name": "test-group", "export_time": "2020-10-13T12:00:20-05:00", "etl_version": "1.0.0+test"}
+{"table_name": "allergyintolerance", "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/simple/output/etl__completion/etl__completion.003.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/simple/output/etl__completion/etl__completion.004.ndjson

@@ -1 +1 @@
-{"table_name": "device", "group_name": "test-group", "export_time": "2020-10-13T12:00:20-05:00", "etl_version": "1.0.0+test"}
+{"table_name": "device", "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/simple/output/etl__completion/etl__completion.005.ndjson

@@ -1 +1 @@
-{"table_name": "diagnosticreport", "group_name": "test-group", "export_time": "2020-10-13T12:00:20-05:00", "etl_version": "1.0.0+test"}
+{"table_name": "diagnosticreport", "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/simple/output/etl__completion/etl__completion.006.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"}

tests/data/simple/output/etl__completion/etl__completion.007.ndjson

@@ -1 +1 @@
-{"table_name": "immunization", "group_name": "test-group", "export_time": "2020-10-13T12:00:20-05:00", "etl_version": "1.0.0+test"}
+{"table_name": "immunization", "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/simple/output/etl__completion/etl__completion.008.ndjson

@@ -1,2 +1,2 @@
-{"table_name": "medication", "group_name": "test-group", "export_time": "2020-10-13T12:00:20-05:00", "etl_version": "1.0.0+test"}
-{"table_name": "medicationrequest", "group_name": "test-group", "export_time": "2020-10-13T12:00:20-05:00", "etl_version": "1.0.0+test"}
+{"table_name": "medication", "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"}
+{"table_name": "medicationrequest", "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/simple/output/etl__completion/etl__completion.009.ndjson

@@ -1 +1 @@
-{"table_name": "observation", "group_name": "test-group", "export_time": "2020-10-13T12:00:20-05:00", "etl_version": "1.0.0+test"}
+{"table_name": "observation", "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/simple/output/etl__completion/etl__completion.010.ndjson

@@ -1 +1 @@
-{"table_name": "procedure", "group_name": "test-group", "export_time": "2020-10-13T12:00:20-05:00", "etl_version": "1.0.0+test"}
+{"table_name": "procedure", "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/simple/output/etl__completion/etl__completion.011.ndjson

@@ -1 +1 @@
-{"table_name": "servicerequest", "group_name": "test-group", "export_time": "2020-10-13T12:00:20-05:00", "etl_version": "1.0.0+test"}
+{"table_name": "servicerequest", "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/etl/base.py

@@ -133,6 +133,7 @@ def setUp(self) -> None:
             "ndjson",
             "ndjson",
             client,
+            timestamp=common.datetime_now(),
             batch_size=5,
             dir_errors=self.errors_dir,
             export_group_name="test-group",

tests/etl/test_tasks.py

@@ -197,7 +197,9 @@ async def test_encounter_completion(self):
         self.assertFalse(comp_enc_format.update_existing)
 
         self.assertEqual("etl__completion", comp_format.dbname)
-        self.assertEqual({"table_name", "group_name"}, comp_format.uniqueness_fields)
+        self.assertEqual(
+            {"table_name", "group_name", "export_time", "etl_time"}, comp_format.uniqueness_fields
+        )
         self.assertTrue(comp_format.update_existing)
 
         self.assertEqual(2, comp_enc_format.write_records.call_count)
@@ -240,6 +242,7 @@ async def test_encounter_completion(self):
                     "export_time": "2012-10-10T05:30:12+00:00",
                     "export_url": self.export_url,
                     "etl_version": "1.0.0+test",
+                    "etl_time": "2021-09-14T21:23:45+00:00",
                 }
             ],
             comp_batch.rows,
@@ -276,13 +279,15 @@ async def test_medication_completion(self):
                     "export_time": "2012-10-10T05:30:12+00:00",
                     "export_url": self.export_url,
                     "etl_version": "1.0.0+test",
+                    "etl_time": "2021-09-14T21:23:45+00:00",
                 },
                 {
                     "table_name": "medicationrequest",
                     "group_name": "test-group",
                     "export_time": "2012-10-10T05:30:12+00:00",
                     "export_url": self.export_url,
                     "etl_version": "1.0.0+test",
+                    "etl_time": "2021-09-14T21:23:45+00:00",
                 },
             ],
             comp_batch.rows,
@@ -330,6 +335,7 @@ async def test_allow_empty_group(self):
                     "export_time": "2012-10-10T05:30:12+00:00",
                     "export_url": self.export_url,
                     "etl_version": "1.0.0+test",
+                    "etl_time": "2021-09-14T21:23:45+00:00",
                 }
             ],
             comp_format.write_records.call_args[0][0].rows,
@@ -420,7 +426,10 @@ async def test_external_medications(self, mock_download):
                     uniqueness_fields=None,
                     update_existing=True,
                 ),
-                mock.call(dbname="etl__completion", uniqueness_fields={"group_name", "table_name"}),
+                mock.call(
+                    dbname="etl__completion",
+                    uniqueness_fields={"group_name", "table_name", "export_time", "etl_time"},
+                ),
             ],
             self.create_formatter_mock.call_args_list,
         )

tests/loaders/ndjson/test_bulk_export.py

@@ -922,6 +922,7 @@ async def test_successful_etl_bulk_export(self):
                     "export_time": "2015-02-07T13:28:17+02:00",
                     "export_url": f"{self.fhir_url}/$export?_type=Patient",
                     "etl_version": "1.0.0+test",
+                    "etl_time": "2021-09-14T21:23:45+00:00",
                 },
                 common.read_json(f"{tmpdir}/output/etl__completion/etl__completion.000.ndjson"),
             )