Skip to content
Draft
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
71 changes: 56 additions & 15 deletions src/databricks/labs/dqx/checks_storage.py
Original file line number Diff line number Diff line change
Expand Up @@ -9,6 +9,7 @@
from io import StringIO, BytesIO
from pathlib import Path
from abc import ABC, abstractmethod
from collections.abc import Callable
from typing import Generic, TypeVar, NoReturn
from sqlalchemy import (
DateTime,
Expand Down Expand Up @@ -211,6 +212,7 @@ def to_dataframe(
run_config_name: str = "default",
rule_set_fingerprint: str | None = None,
created_at: datetime | None = None,
custom_checks: dict[str, Callable] | None = None,

Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Need to also add this to the Args section in the docstring.

) -> DataFrame:
"""
Converts a list of quality checks defined as Python dictionaries to a DataFrame.
Expand All @@ -236,7 +238,7 @@ def to_dataframe(
Raises:
InvalidCheckError: If any check is invalid or unsupported.
"""
status = ChecksValidator.validate_checks(checks, None)
status = ChecksValidator.validate_checks(checks, custom_checks)
if status.has_errors:
raise InvalidCheckError(str(status))

Expand All @@ -245,7 +247,7 @@ def to_dataframe(
rule_set_fingerprint = (
rule_set_fingerprint
if rule_set_fingerprint is not None
else compute_rule_set_fingerprint_by_metadata(checks)
else compute_rule_set_fingerprint_by_metadata(checks, custom_checks)
)

effective_created_at = created_at if created_at is not None else datetime.now(timezone.utc)
Expand All @@ -264,7 +266,9 @@ def to_dataframe(
"arguments": json_arguments,
}

name, rule_fingerprint = DataFrameConverter._resolve_name_and_fingerprint(original_check, check)
name, rule_fingerprint = DataFrameConverter._resolve_name_and_fingerprint(
original_check, check, custom_checks
)

# Values are already normalized by ChecksNormalizer.normalize; json.dumps for MAP<STRING, STRING>,
# mirroring the arguments encoding so non-string user_metadata types survive the round-trip.
Expand All @@ -290,7 +294,9 @@ def to_dataframe(
return spark.createDataFrame(dq_rule_rows, CHECKS_TABLE_SCHEMA)

@staticmethod
def _resolve_name_and_fingerprint(original_check: dict, normalized_check: dict) -> tuple[str | None, str]:
def _resolve_name_and_fingerprint(
original_check: dict, normalized_check: dict, custom_checks: dict[str, Callable] | None = None
) -> tuple[str | None, str]:
Comment on lines +297 to +299

Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Let's add a line in the Args section of the docstring for custom_checks

"""Resolve the stored name and rule_fingerprint so they match what apply writes into _errors/_warnings.

A DQRule autogenerates its name in __post_init__ and includes that name in its fingerprint, so a check
Expand All @@ -308,7 +314,7 @@ def _resolve_name_and_fingerprint(original_check: dict, normalized_check: dict)
check_inner = normalized_check.get("check") or {}
if check_inner.get("for_each_column"):
return normalized_check.get("name"), compute_rule_fingerprint(normalized_check)
rule = deserialize_checks([original_check])[0]
rule = deserialize_checks([original_check], custom_checks)[0]
return rule.name, rule.rule_fingerprint


Expand All @@ -326,13 +332,14 @@ def load(self, config: T) -> list[dict]:

Args:
config: configuration for loading checks, including the table location and run configuration name.

custom_check_functions: Optional dictionary with custom check functions (e.g., *globals()* of
the calling module).
Comment on lines +335 to +336

Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I think this should be added to the ChecksStorageHandler.save method instead?

Returns:
list of dq rules or raise an error if checks file is missing or is invalid.
"""

@abstractmethod
def save(self, checks: list[dict], config: T) -> None:
def save(self, checks: list[dict], config: T, custom_check_functions: dict[str, Callable] | None = None) -> None:
"""Save quality rules to the target."""


Expand Down Expand Up @@ -372,7 +379,12 @@ def load(self, config: TableChecksStorageConfig) -> list[dict]:
)

@telemetry_logger("save_checks", "table")
def save(self, checks: list[dict], config: TableChecksStorageConfig) -> None:
def save(
self,
checks: list[dict],
config: TableChecksStorageConfig,
custom_check_functions: dict[str, Callable] | None = None,
) -> None:
"""
Save checks to a Delta table in the workspace.

Expand Down Expand Up @@ -418,9 +430,13 @@ def save(self, checks: list[dict], config: TableChecksStorageConfig) -> None:
)

logger.info(f"Saving quality rules (checks) to table '{config.location}'")
rule_set_fingerprint = compute_rule_set_fingerprint_by_metadata(checks)
rule_set_fingerprint = compute_rule_set_fingerprint_by_metadata(checks, custom_check_functions)
rules_df = DataFrameConverter.to_dataframe(
self.spark, checks, run_config_name=config.run_config_name, rule_set_fingerprint=rule_set_fingerprint
self.spark,
checks,
run_config_name=config.run_config_name,
rule_set_fingerprint=rule_set_fingerprint,
custom_checks=custom_check_functions,
)

# Skip save if rule_set_fingerprint already exists in existing table
Expand Down Expand Up @@ -962,7 +978,12 @@ def load(self, config: LakebaseChecksStorageConfig) -> list[dict]:
engine.dispose()

@telemetry_logger("save_checks", "lakebase")
def save(self, checks: list[dict], config: LakebaseChecksStorageConfig) -> None:
def save(
self,
checks: list[dict],
config: LakebaseChecksStorageConfig,
custom_check_functions: dict[str, Callable] | None = None,
) -> None:
Comment on lines +981 to +986

Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

You will need to pass custom_check_functions through a few methods:

_save_checks_to_lakebase -> _normalize_checks

"""
Save dq rules (checks) to a Lakebase table.

Expand Down Expand Up @@ -1048,7 +1069,12 @@ def load(self, config: WorkspaceFileChecksStorageConfig) -> list[dict]:
raise InvalidCheckError(f"Invalid checks in file: {file_path}: {e}") from e

@telemetry_logger("save_checks", "workspace_file")
def save(self, checks: list[dict], config: WorkspaceFileChecksStorageConfig) -> None:
def save(
self,
checks: list[dict],
config: WorkspaceFileChecksStorageConfig,
custom_check_functions: dict[str, Callable] | None = None,
) -> None:
"""Save checks (dq rules) to yaml file in the workspace.
This does not require installation of DQX in the workspace.

Expand Down Expand Up @@ -1097,7 +1123,12 @@ def load(self, config: FileChecksStorageConfig) -> list[dict]:
except (yaml.YAMLError, json.JSONDecodeError) as e:
raise InvalidCheckError(f"Invalid checks in file: {file_path}: {e}") from e

def save(self, checks: list[dict], config: FileChecksStorageConfig) -> None:
def save(
self,
checks: list[dict],
config: FileChecksStorageConfig,
custom_check_functions: dict[str, Callable] | None = None,
) -> None:
"""
Save checks (dq rules) to a file (json or yaml) in the local filesystem.

Expand Down Expand Up @@ -1160,7 +1191,12 @@ def load(self, config: InstallationChecksStorageConfig) -> list[dict]:
return handler.load(config)

@telemetry_logger("save_checks", "installation")
def save(self, checks: list[dict], config: InstallationChecksStorageConfig) -> None:
def save(
self,
checks: list[dict],
config: InstallationChecksStorageConfig,
custom_check_functions: dict[str, Callable] | None = None,
) -> None:

Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Need to pass customer_check_functions to handler.save here.

"""
Save checks (dq rules) to yaml file or table in the installation folder.
This will overwrite existing checks file or table.
Expand Down Expand Up @@ -1271,7 +1307,12 @@ def load(self, config: VolumeFileChecksStorageConfig) -> list[dict]:
raise InvalidCheckError(f"Invalid checks in file: {file_path}: {e}") from e

@telemetry_logger("save_checks", "volume")
def save(self, checks: list[dict], config: VolumeFileChecksStorageConfig) -> None:
def save(
self,
checks: list[dict],
config: VolumeFileChecksStorageConfig,
custom_check_functions: dict[str, Callable] | None = None,
) -> None:
"""Save checks (dq rules) to yaml file in a Unity Catalog volume.
This does not require installation of DQX in a Unity Catalog volume.

Expand Down
3 changes: 2 additions & 1 deletion src/databricks/labs/dqx/engine.py
Original file line number Diff line number Diff line change
Expand Up @@ -1716,6 +1716,7 @@ def save_checks(
config: BaseChecksStorageConfig,
variables: dict[str, VariableValue] | None = None,
semantic_validation_mode: str | None = ChecksSemanticValidationMode.WARN,
custom_check_functions: dict[str, Callable] | None = None,

Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Similar to the other methods, this argument needs a line in the method docstring.

) -> None:
"""Persist DQ rules (checks) to the storage backend described by *config*.

Expand Down Expand Up @@ -1759,7 +1760,7 @@ def save_checks(
if semantic_validation_mode is not None:
ChecksSemanticValidator.apply(resolved_checks, mode=semantic_validation_mode)
handler = self._checks_handler_factory.create(config)
handler.save(resolved_checks, config)
handler.save(resolved_checks, config, custom_check_functions)

def _build_metrics_observation(
self,
Expand Down
Loading