Versioning of Checks

[STEFANOVIVAS]

Changes

Added rule_fingerprint, rule_set_fingerprint, and created_at to checks storage. Added rule_set_fingerprint to summary metrics so that rules can be traced.

BREAKING CHANGE!

Default mode changed from "overwrite" to "append" when saving checks to a delta and lakebase table. Rules are versioned going forward and each new entry has a created_at timestamp, rule_set_fingerprint and rule_fingerprint.

Linked issues

Resolves #672

Tests

• manually tested
• added unit tests
• added integration tests
• added end-to-end tests
• added performance tests

[alexott]

Code review

Found 2 issues:

1. spark.catalog.tableExists re-introduced in the save path, regressing the fix from PR Use WorkspaceClient to check table existence #1035 which replaced it with ws.tables.get() because spark.catalog.tableExists fails for tables with special characters and is blocked in Spark Declarative Pipelines. The existing load path correctly continues to use ws.tables.get(), making the two methods inconsistent.

dqx/src/databricks/labs/dqx/checks_storage.py

Lines 148 to 153 in 92a5d6b

	rule_set_fingerprint = first_row[0] if first_row else None
	if self.spark.catalog.tableExists(config.location) and rule_set_fingerprint is not None:
	if (
	not self.spark.read.table(config.location)

2. In _load_checks_from_lakebase, the "load latest version" path fetches the most recent rule_set_fingerprint via a subquery filtered by run_config_name, but the outer SELECT only filters by rule_set_fingerprint without also filtering by run_config_name. If two different configs share identical check content (same fingerprint), the query returns rows from both configs mixed together.

dqx/src/databricks/labs/dqx/checks_storage.py

Lines 430 to 441 in 92a5d6b

	select(table.c.rule_set_fingerprint)
	.where(table.c.run_config_name == config.run_config_name)
	.order_by(table.c.created_at.desc())
	.limit(1)
	.scalar_subquery()
	)
	stmt = select(table).where(table.c.rule_set_fingerprint == latest_rule_set_fingerprint)
	with engine.connect() as conn:
	result = conn.execute(stmt)
	checks = result.mappings().all()

🤖 Generated with Claude Code

_{- If this code review was useful, please react with 👍. Otherwise, react with 👎.}

[alexott]

Additional code review findings (lower confidence)

Found 4 additional issues (scored below primary threshold):

1. Lakebase deduplication dead code when mode="overwrite" — in _save_checks_to_lakebase, the overwrite DELETE runs first, then the fingerprint existence check queries for a row that was just deleted. The SELECT ... WHERE rule_set_fingerprint == ... always returns empty in overwrite mode, so the early-return skip can never trigger. The existence check should run before the delete.

dqx/src/databricks/labs/dqx/checks_storage.py

Lines 356 to 378 in 92a5d6b

	)
	self._ensure_rule_version_columns_exist(conn, config)
	logger.info("Rule version columns exist or added.")
	if config.mode == "overwrite":
	delete_stmt = delete(table).where(table.c.run_config_name == config.run_config_name)
	result = conn.execute(delete_stmt)
	logger.info(f"Deleted {result.rowcount} existing checks for run_config_name '{config.run_config_name}'")
	normalized_checks = self._normalize_checks(checks, config)
	rule_set_fingerprint = normalized_checks[0].get("rule_set_fingerprint")
	exists_rule_set = (
	select(table.c.rule_set_fingerprint)
	.where(
	table.c.run_config_name == config.run_config_name,
	table.c.rule_set_fingerprint == rule_set_fingerprint,
	)
	.limit(1)
	)
	if conn.execute(exists_rule_set).first():
	logger.info(f"Checks with rule_set_fingerprint {rule_set_fingerprint} already exist — skipping")
	return
	insert_stmt = insert(table)

2. rule_set_fingerprint = 'None' on schema migration — when latest_row.rule_set_fingerprint is Python None (pre-existing rows whose column was backfilled as NULL via mergeSchema), the f-string produces the Spark SQL predicate rule_set_fingerprint = 'None' — the literal string "None", not SQL NULL. This silently returns zero rows for any table that has pre-versioning data, breaking load after a schema migration.

dqx/src/databricks/labs/dqx/checks_serializer.py

Lines 387 to 395 in 92a5d6b

	if not rule_set_fingerprint:
	max_created_at = filtered_df.agg(F.max("created_at")).collect()[0][0]
	latest_row = filtered_df.where(F.col("created_at") == max_created_at).first()
	if latest_row:
	rule_set_fingerprint = latest_row.rule_set_fingerprint
	filtered_df = filtered_df.where(f"rule_set_fingerprint = '{rule_set_fingerprint}'")
	check_rows = filtered_df.collect()

3. Docstring contradicts the new default — both TableChecksStorageConfig and LakebaseChecksStorageConfig change mode from "overwrite" to "append", but LakebaseChecksStorageConfig's docstring still explicitly states (default is 'overwrite'). This is a breaking behavioral change for existing callers that relied on the default.

dqx/src/databricks/labs/dqx/config.py

Lines 287 to 301 in 92a5d6b

	run_config_name: Name of the run configuration to use for checks (default is 'default').
	mode: The mode for writing checks to a table (e.g., 'append' or 'overwrite'). The *overwrite* mode
	only replaces checks for the specific run config and not all checks in the table (default is 'overwrite').
	rule_set_fingerprint: Optional SHA-256 fingerprint of the rule set to load.
	When provided, loads rules matching this specific fingerprint instead of the latest batch.
	When None (default), loads the latest batch.
	"""
	location: str
	instance_name: str | None = None
	client_id: str | None = None
	port: str = "5432"
	run_config_name: str = "default"
	mode: str = "append"
	rule_set_fingerprint: str | None = None

4. SQL injection via f-string for rule_set_fingerprint — the new filter in TableChecksStorageHandler.save embeds the user-supplied rule_set_fingerprint config field directly into a Spark SQL string. When the fingerprint is computed internally it is always a safe hex-SHA256, but since TableChecksStorageConfig.rule_set_fingerprint is an externally-supplied field, a crafted value can bypass the filter. The Column API (F.col("rule_set_fingerprint") == rule_set_fingerprint) avoids this, consistent with the rest of the codebase.

dqx/src/databricks/labs/dqx/checks_storage.py

Lines 147 to 158 in 92a5d6b

	first_row = rules_df.select("rule_set_fingerprint").first()
	rule_set_fingerprint = first_row[0] if first_row else None
	if self.spark.catalog.tableExists(config.location) and rule_set_fingerprint is not None:
	if (
	not self.spark.read.table(config.location)
	.filter(
	f"run_config_name = '{config.run_config_name}' and rule_set_fingerprint = '{rule_set_fingerprint}'"
	)
	.isEmpty()
	):

🤖 Generated with Claude Code

_{- If this code review was useful, please react with 👍. Otherwise, react with 👎.}

[STEFANOVIVAS]

@STEFANOVIVAS @ghanse i'm done with changes. Can you please double check? just running int tests, if all good then merge

Hi, @mwojtyczka! From what I've reviewed, I think everything is ok.

Just one thing to mention... I think we need an extra indentation on lines 148 and 149 of the checks_storage.py file, to match the one in the variable rule_set_fingerprint = result[0][0].

          rule_set_fingerprint = result[0][0]
      # May be None for legacy rows with NULL rule_set_fingerprint; load all rows for run_config_name
      if rule_set_fingerprint is not None:
          filtered_df = filtered_df.where(F.col("rule_set_fingerprint") == rule_set_fingerprint)  

[mwojtyczka]

@STEFANOVIVAS @ghanse i'm done with changes. Can you please double check? just running int tests, if all good then merge

Hi, @mwojtyczka! From what I've reviewed, I think everything is ok.

Just one thing to mention... I think we need an extra indentation on lines 148 and 149 of the checks_storage.py file, to match the one in the variable rule_set_fingerprint = result[0][0].

          rule_set_fingerprint = result[0][0]
      # May be None for legacy rows with NULL rule_set_fingerprint; load all rows for run_config_name
      if rule_set_fingerprint is not None:
          filtered_df = filtered_df.where(F.col("rule_set_fingerprint") == rule_set_fingerprint)  

This is needed for the case where rule_set_fingerprint is provided by the user and for the case where it is not and it is found. The filter must run in both cases whenever we have a non‑None fingerprint. If lines 147–149 were indented the filter would never run when the user provides rule_set_fingerprint, which would be a bug.

[xicez]

This breaks pipelines that have user defined schemas

[mwojtyczka]

Yes, this adds two new fields to the DQ result struct schema (the inner type of _errors / _warnings array columns). The new fields are nullable=True, and DQX docs surface {"mergeSchema": "true"} as the documented write option which could be used here. But the upgrade requires user action. We put this in the release notes: https://github.com/databrickslabs/dqx/releases/tag/v0.14.0

[xicez]

Yes, this adds two new fields to the DQ result struct schema (the inner type of _errors / _warnings array columns). The new fields are nullable=True, and DQX docs surface {"mergeSchema": "true"} as the documented write option which could be used here. But the upgrade requires user action. We put this in the release notes: https://github.com/databrickslabs/dqx/releases/tag/v0.14.0

Gotcha, thanks! Looks like I need to set up some notifications for the release notes 😅