fix: add run_time to DQEngine extra_params

Author: pierre-monnet

src/databricks/labs/dqx/engine.py

@@ -1,4 +1,3 @@
-from datetime import datetime
 import copy
 import logging
 import os
@@ -55,6 +54,8 @@ def __init__(self, workspace_client: WorkspaceClient, extra_params: ExtraParams
             ),
         }
 
+        self.run_time = extra_params.run_time
+
     def apply_checks(self, df: DataFrame, checks: list[DQRule]) -> DataFrame:
         if not checks:
             return self._append_empty_checks(df)
@@ -254,8 +255,7 @@ def _append_empty_checks(self, df: DataFrame) -> DataFrame:
             F.lit(None).cast(validation_result_schema).alias(self._column_names[ColumnArguments.WARNINGS]),
         )
 
-    @staticmethod
-    def _create_results_map(df: DataFrame, checks: list[DQRule], dest_col: str) -> DataFrame:
+    def _create_results_map(self, df: DataFrame, checks: list[DQRule], dest_col: str) -> DataFrame:
         """ ""Create a map from the values of the specified columns, using the column names as a key.  This function is
         used to collect individual check columns into corresponding errors and/or warnings columns.
 
@@ -267,7 +267,6 @@ def _create_results_map(df: DataFrame, checks: list[DQRule], dest_col: str) -> D
 
         if len(checks) == 0:
             return df.select("*", empty_type)
-        current_date = datetime.now()
         check_cols = []
         for check in checks:
             result = F.struct(
@@ -276,7 +275,7 @@ def _create_results_map(df: DataFrame, checks: list[DQRule], dest_col: str) -> D
                 F.lit(check.col_name).alias("col_name"),
                 F.lit(check.filter or None).cast("string").alias("filter"),
                 F.lit(check.check_func.__name__).alias("function"),
-                F.lit(current_date).alias("run_time"),
+                F.lit(self.run_time).alias("run_time"),
             )
             check_cols.append(result)
 

src/databricks/labs/dqx/rule.py

@@ -3,6 +3,7 @@
 import functools as ft
 from typing import Any
 from collections.abc import Callable
+from datetime import datetime
 from pyspark.sql import Column
 import pyspark.sql.functions as F
 from databricks.labs.dqx.utils import get_column_name
@@ -34,6 +35,7 @@ class ExtraParams:
     """Class to represent extra parameters for DQEngine."""
 
     column_names: dict[str, str] = field(default_factory=dict)
+    run_time: datetime = datetime.now()
 
 
 @dataclass(frozen=True)

tests/conftest.py

@@ -1,6 +1,4 @@
 import os
-import datetime as dt
-from unittest.mock import patch
 import pytest
 
 from databricks.labs.pytester.fixtures.baseline import factory
@@ -274,11 +272,3 @@ def delete(workspace_file_path: str) -> None:
         ws.workspace.delete(workspace_file_path)
 
     yield from factory("file", create, delete)
-
-
-@pytest.fixture
-def run_time_date():
-    run_datetime = dt.datetime(2025, 1, 1, 0, 0, 0, 0)
-    with patch("databricks.labs.dqx.engine.datetime") as mock_date:  # pylint: disable=explicit-dependency-required
-        mock_date.now.return_value = run_datetime
-        yield run_datetime

tests/integration/conftest.py

@@ -1,6 +1,5 @@
 import os
 import logging
-import datetime as dt
 
 from collections.abc import Callable, Generator
 from functools import cached_property
@@ -185,14 +184,6 @@ def installation_ctx(
     ctx.workspace_installation.uninstall()
 
 
-@pytest.fixture
-def run_time_date():
-    run_datetime = dt.datetime(2025, 1, 1, 0, 0, 0, 0)
-    with patch("databricks.labs.dqx.engine.datetime") as mock_date:  # pylint: disable=explicit-dependency-required
-        mock_date.now.return_value = run_datetime
-        yield run_datetime
-
-
 @pytest.fixture
 def webbrowser_open():
     with patch("webbrowser.open") as mock_open: