Update to PCDM logic to cater for Null EventTimes (rtdip#764)

* Update to PCDM logic Signed-off-by: GBBBAS <[email protected]> * Update to condition logic Signed-off-by: GBBBAS <[email protected]> --------- Signed-off-by: GBBBAS <[email protected]>
GBBBAS · Jun 19, 2024 · 65bfce2 · 65bfce2
1 parent 38af92d
commit 65bfce2
Showing 1 changed file with 2 additions and 2 deletions.
diff --git a/src/sdk/python/rtdip_sdk/pipelines/destinations/spark/pcdm_latest_to_delta.py b/src/sdk/python/rtdip_sdk/pipelines/destinations/spark/pcdm_latest_to_delta.py
@@ -185,7 +185,7 @@ def _write_latest_to_delta(self, df: DataFrame, epoch_id=None):  # NOSONAR
                 },
             ),
             DeltaMergeConditionValues(
-                condition="source.EventTime > target.EventTime AND source.GoodLatest.EventTime IS NOT NULL AND source.GoodLatest.EventTime > target.GoodEventTime",
+                condition="source.EventTime > target.EventTime AND (source.GoodLatest.EventTime IS NOT NULL AND (source.GoodLatest.EventTime > target.GoodEventTime OR target.GoodEventTime IS NULL))",
                 values={
                     "EventTime": "source.EventTime",
                     "Status": "source.Status",
@@ -197,7 +197,7 @@ def _write_latest_to_delta(self, df: DataFrame, epoch_id=None):  # NOSONAR
                 },
             ),
             DeltaMergeConditionValues(
-                condition="source.EventTime <= target.EventTime AND source.GoodLatest.EventTime IS NOT NULL AND source.GoodLatest.EventTime > target.GoodEventTime",
+                condition="source.EventTime <= target.EventTime AND (source.GoodLatest.EventTime IS NOT NULL AND (source.GoodLatest.EventTime > target.GoodEventTime OR target.GoodEventTime IS NULL))",
                 values={
                     "GoodEventTime": "source.GoodLatest.EventTime",
                     "GoodValue": "source.GoodLatest.Value",