You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@airflow.apache.org by "gopidesupavan (via GitHub)" <gi...@apache.org> on 2026/07/07 21:41:32 UTC
[PR] Airflow Data Quality Provider [airflow]
gopidesupavan opened a new pull request, #69575:
URL: https://github.com/apache/airflow/pull/69575
Adds a new `apache-airflow-providers-dq` provider for `DbApiHook`-based data quality checks.
Airflow already has SQL check operators, and many users rely on them for data quality today. This provider adds a `DQRule` / `RuleSet` layer for checks that need stable rule identity, persisted history, and a connection to Airflow assets. That makes quality results easier to analyze over time, lets downstream asset consumers gate on recent quality, and gives LLM-assisted workflows one schema to generate when proposing checks from table context. Execution still goes through existing `common.sql` / `DbApiHook` connections.
This PR is the backend/provider slice only. The UI plugin and read-only API are intentionally left for a follow-up PR.
Ships:
- `DQRule` and `RuleSet` models for named data quality rules.
- Built-in SQL checks for common table and column checks, executed through `common.sql` / `DbApiHook`.
- `custom_sql` support for database-specific or more complex checks.
- `DQCheckOperator` and the `@task.dq_check` TaskFlow decorator.
- A configurable results backend under `[dq] results_path` for task, run, and rule-level history.
- Experimental asset helpers, `asset_quality()` and `require_quality()`, that attach provider-owned quality metadata to assets without changing Airflow core.
- Documentation and example Dags covering end-to-end usage with and without LLM-generated rules.
- A DQ rule-authoring skill that LLM-assisted workflows can use to generate rules from table/schema context:
https://github.com/gopidesupavan/airflow/blob/f32940bd261b94238256eaced9150dd51329ce3e/providers/dq/src/airflow/providers/dq/skills/dq-rule-authoring/SKILL.md
This first version is intentionally focused on the backend contract: deterministic rule definitions, SQL execution through existing Airflow SQL providers, persisted results, and asset-linked quality summaries.
Design decisions:
- Results are stored through an object-storage/local-file backend instead of adding new metadata DB tables in the first provider drop. This keeps the provider self-contained, avoids Airflow core migrations, and lets deployments choose a durable store such as S3, GCS, or local files via `[dq] results_path`.
- The backend stores keyed JSON records for task runs, task instances, and per-rule history so later readers, including a future UI/API layer, can access common views without scanning unrelated runs.
- Asset support is implemented with provider-owned metadata, not Airflow core changes. Static quality configuration is attached to `Asset.extra["airflow.dq"]`; runtime summaries are attached to asset events under `extra["airflow.dq.result"]`.
- The first release starts with `DbApiHook` / SQL execution because Airflow already has broad database coverage through `common.sql`. File and object-store data checks are left for a later iteration.
Later iterations:
- Read-only API and minimal Airflow UI plugin for viewing task/run results and rule history.
- File/object-store based checks, where Airflow reads data from S3/GCS/local files or other object stores and runs quality rules directly against that data.
- OpenLineage integration for data quality facets.
- More built-in checks.
- Trigger support for DQ checks.
- `DQProfileOperator`.
<!-- SPDX-License-Identifier: Apache-2.0
https://www.apache.org/licenses/LICENSE-2.0 -->
<!--
Thank you for contributing!
Please provide above a brief description of the changes made in this pull request.
Write a good git commit message following this guide: http://chris.beams.io/posts/git-commit/
Please make sure that your code changes are covered with tests.
And in case of new features or big changes remember to adjust the documentation.
For user-facing UI changes, please attach before/after screenshots (or a short
screen recording) so reviewers can assess the visual impact.
Feel free to ping (in general) for the review if you do not see reaction for a few days
(72 Hours is the minimum reaction time you can expect from volunteers) - we sometimes miss notifications.
In case of an existing issue, reference it using one of the following:
* closes: #ISSUE
* related: #ISSUE
-->
---
##### Was generative AI tooling used to co-author this PR?
<!--
If generative AI tooling has been used in the process of authoring this PR, please
change below checkbox to `[X]` followed by the name of the tool, uncomment the "Generated-by".
-->
- [ ] Yes (please specify the tool below)
<!--
Generated-by: [Tool Name] following [the guidelines](https://github.com/apache/airflow/blob/main/contributing-docs/05_pull_requests.rst#gen-ai-assisted-contributions)
-->
---
* Read the **[Pull Request Guidelines](https://github.com/apache/airflow/blob/main/contributing-docs/05_pull_requests.rst#pull-request-guidelines)** for more information. Note: commit author/co-author name and email in commits become permanently public when merged.
* For fundamental code changes, an Airflow Improvement Proposal ([AIP](https://cwiki.apache.org/confluence/display/AIRFLOW/Airflow+Improvement+Proposals)) is needed.
* When adding dependency, check compliance with the [ASF 3rd Party License Policy](https://www.apache.org/legal/resolved.html#category-x).
* For significant user-facing changes create newsfragment: `{pr_number}.significant.rst`, in [airflow-core/newsfragments](https://github.com/apache/airflow/tree/main/airflow-core/newsfragments). You can add this file in a follow-up commit after the PR is created so you know the PR number.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Airflow Data Quality Provider part 1 [airflow]
Posted by "gopidesupavan (via GitHub)" <gi...@apache.org>.
gopidesupavan commented on PR #69575:
URL: https://github.com/apache/airflow/pull/69575#issuecomment-4991488466
@Lee-W regarding extra what other options you would suggest? do we have any other ways to store that now?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Airflow Data Quality Provider part 1 [airflow]
Posted by "Copilot (via GitHub)" <gi...@apache.org>.
Copilot commented on code in PR #69575:
URL: https://github.com/apache/airflow/pull/69575#discussion_r3567854636
##########
providers/common/dataquality/src/airflow/providers/common/dataquality/backends/object_storage.py:
##########
@@ -0,0 +1,250 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+"""
+Object-storage results backend.
+
+Each DQ check writes a keyed JSON document plus read indexes optimized for the UI:
+
+ runs/by_task/dag_id=<dag>/task_id=<task>/date=<2026-07-04>/<run_uid>.json
+ Canonical run record: ``{"run": ..., "results": [...], "summary": ...}``.
+
+ runs/by_task_instance/dag_id=<dag>/task_id=<task>/<safe_run_id>__<map_index>.json
+ Latest result for a task-instance page. Last write wins across retries.
+
+ rules/by_task_rule/dag_id=<dag>/task_id=<task>/rule_uid=<uid>/<started_at>__<run_uid>.json
+ One rule result plus run context: ``{"run": ..., "result": ...}``.
+
+The duplicate files are intentional read indexes: DQ tasks write once, while the UI reads
+many times. Keeping these indexes avoids scanning all task runs for common UI views.
+"""
+
+from __future__ import annotations
+
+import json
+import logging
+from datetime import datetime, timezone
+from typing import Any
+
+from airflow.providers.common.dataquality.results import DQRun, RuleResult, build_summary
+from airflow.sdk import ObjectStoragePath
+
+log = logging.getLogger(__name__)
+
+
+class ObjectStorageResultsBackend:
+ """Persist DQ results as JSON files via ``ObjectStoragePath``."""
+
+ def __init__(self, results_path: str, conn_id: str | None = None) -> None:
+ self.root = ObjectStoragePath(results_path, conn_id=conn_id)
+
+ def write_run(self, run: DQRun, results: list[RuleResult]) -> None:
+ timestamp = run.started_at or datetime.now(tz=timezone.utc).isoformat()
+ payload = self._build_run_payload(run, results)
+
+ self._write_run_file(run, timestamp[:10], payload)
+ self._write_task_instance_index(run, payload)
+ self._write_rule_indexes(run, results, timestamp)
+
+ def read_task_rule_history(
+ self, dag_id: str, task_id: str, rule_uid: str, limit: int = 100, before: str | None = None
+ ) -> dict[str, Any]:
+ """Return recent results for one rule produced by one task, newest first."""
+ rule_dir = (
+ self.root
+ / "rules"
+ / "by_task_rule"
+ / f"dag_id={dag_id}"
+ / f"task_id={task_id}"
+ / f"rule_uid={rule_uid}"
+ )
+ return self._read_rule_history_dir(rule_dir, limit, before)
+
+ def read_task_runs(
+ self, dag_id: str, task_id: str, limit: int = 50, before: str | None = None
+ ) -> dict[str, Any]:
+ """
+ Return recent data quality runs for one task, newest first.
+
+ Returns ``{"items": [...], "next_cursor": ...}``. ``before`` is the opaque
+ ``next_cursor`` from the previous page, so "load more" only reads runs it hasn't
+ shown yet. ``next_cursor`` is set by reading one extra entry past ``limit``: cheap
+ on every page except the last one, where there's no way to confirm history is
+ exhausted without walking to the end.
+
+ ``date=`` partition names sort correctly as plain strings, so directories are walked
+ newest-first and scanning stops as soon as ``limit + 1`` matching runs have been
+ collected — a task with years of history doesn't pay for a full scan on every page.
+ """
+ task_dir = self.root / "runs" / "by_task" / f"dag_id={dag_id}" / f"task_id={task_id}"
+ if not task_dir.exists():
+ return {"items": [], "next_cursor": None}
+
+ date_dirs = sorted((path for path in task_dir.iterdir() if path.is_dir()), reverse=True)
+ runs = []
+ for date_dir in date_dirs:
+ for path in date_dir.iterdir():
+ if not path.name.endswith(".json"):
+ continue
+ payload = self._read_json(path)
+ if payload is None:
+ continue
+ cursor = self._get_run_payload_cursor(payload)
+ if before is not None and cursor >= before:
+ continue
+ runs.append(payload)
+ if len(runs) > limit:
+ break
Review Comment:
`read_task_runs()` reads *every* JSON file in the newest `date=` directory before checking `limit`, which can be very expensive for tasks with lots of history in a single day. Consider stopping the inner loop as soon as `limit + 1` items are collected so pagination avoids unnecessary object-store listings/reads.
##########
providers/common/dataquality/src/airflow/providers/common/dataquality/operators/dq_check.py:
##########
@@ -0,0 +1,255 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+from __future__ import annotations
+
+import logging
+from collections.abc import Sequence
+from datetime import datetime, timezone
+from typing import TYPE_CHECKING, Any, cast
+
+from airflow.providers.common.dataquality.assets import DQ_RESULT_EXTRA_KEY, get_asset_quality_config
+from airflow.providers.common.dataquality.backends import get_backend_from_config
+from airflow.providers.common.dataquality.engines.sql import SQLDQEngine
+from airflow.providers.common.dataquality.exceptions import DQCheckFailedError
+from airflow.providers.common.dataquality.results import (
+ ERROR,
+ FAIL,
+ PASS,
+ WARN,
+ DQRun,
+ RuleResult,
+ build_summary,
+)
+from airflow.providers.common.dataquality.rules import RuleSet, describe_rule
+from airflow.providers.common.sql.operators.sql import BaseSQLOperator
+from airflow.sdk.definitions._internal.types import SET_DURING_EXECUTION
+
+if TYPE_CHECKING:
+ from airflow.providers.common.dataquality.engines.sql import Observation
+ from airflow.providers.common.dataquality.rules.rule import Condition, Dimension
+ from airflow.providers.common.sql.hooks.sql import DbApiHook
+ from airflow.sdk import Asset, Context
+
+log = logging.getLogger(__name__)
+
+FAIL_ON_CHOICES = ("error", "warn", "never")
+
+
+class DQCheckOperator(BaseSQLOperator):
+ """
+ Run a ruleset against a table and persist per-rule results to the configured results store.
+
+ Every rule is evaluated and recorded regardless of the task outcome. Execution errors
+ (a check query failing) always fail the task; rule failures fail the task according to
+ ``fail_on``.
+
+ :param ruleset: A :class:`~airflow.providers.common.dataquality.rules.RuleSet`, its dict form, or a path
+ to a YAML ruleset file. Optional when ``asset`` carries one.
+ :param table: The table to run checks against. Optional when ``asset`` carries one
+ (falling back to the asset name).
+ :param asset: An asset decorated with :func:`~airflow.providers.common.dataquality.assets.asset_quality`.
+ Supplies defaults for ``ruleset``, ``table``, and ``conn_id`` (explicit arguments
+ win) and is automatically added to the task's outlets so its asset events carry
+ the check summary.
+ :param partition_clause: Predicate ANDed into every built-in check's WHERE clause,
+ e.g. ``"ds = '{{ ds }}'"``.
+ :param fail_on: Severity that fails the task: ``error`` (default — warn-severity rule
+ failures are recorded but don't fail), ``warn`` (any rule failure fails), or
+ ``never`` (only execution errors fail).
+ :param conn_id: Connection to the database to check (any ``DbApiHook``-compatible type).
+ :param database: Optional database/schema overriding the connection's default.
+
+ Results are persisted to the backend configured under ``[common.dataquality] results_path``; when that's
+ unset, checks still run but no history is persisted. There is no per-operator override --
+ all checks in a deployment share one results store.
+ """
+
+ template_fields: Sequence[str] = ("table", "partition_clause", *BaseSQLOperator.template_fields)
+ ui_color: str = "#87ceeb"
+
+ def __init__(
+ self,
+ *,
+ ruleset: RuleSet | dict[str, Any] | str | None = None,
+ table: str | None = None,
+ asset: Asset | None = None,
+ partition_clause: str | None = None,
+ fail_on: str = "error",
+ **kwargs,
+ ) -> None:
+ if asset is not None:
+ config = get_asset_quality_config(asset) or {}
+ ruleset = ruleset if ruleset is not None else config.get("ruleset")
+ table = table or config.get("table") or asset.name
+ kwargs.setdefault("conn_id", config.get("conn_id"))
+ outlets = list(kwargs.get("outlets") or [])
+ if asset not in outlets:
+ outlets.append(asset)
+ kwargs["outlets"] = outlets
+ super().__init__(**kwargs)
Review Comment:
When `asset` is provided but its quality config does not include `conn_id`, `kwargs.setdefault("conn_id", config.get("conn_id"))` sets `conn_id=None`, which later causes `BaseSQLOperator.get_db_hook()` to fail with an unclear error. Only default `conn_id` from the asset when it is set, and fail fast with a clear ValueError when `conn_id` is still missing after initialization.
##########
providers/common/dataquality/src/airflow/providers/common/dataquality/example_dags/example_dq_ruleset_from_yaml.py:
##########
@@ -0,0 +1,83 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+"""
+Example DAG: load a ruleset from a YAML file instead of declaring it in Python.
+
+A path string is accepted anywhere a :class:`~airflow.providers.common.dataquality.rules.RuleSet` is --
+``DQCheckOperator(ruleset=...)``, ``@task.dq_check(ruleset=...)``, and
+:func:`~airflow.providers.common.dataquality.assets.asset_quality` all resolve it via
+:meth:`~airflow.providers.common.dataquality.rules.RuleSet.from_file` at Dag-parse time. This keeps rules
+editable by people who don't write Python -- a data steward can change ``orders_ruleset.yaml``
+without touching the Dag file.
+"""
+
+from __future__ import annotations
+
+import os
+from datetime import datetime
+from pathlib import Path
+
+from airflow.providers.common.dataquality.operators.dq_check import DQCheckOperator
+from airflow.providers.common.sql.operators.sql import SQLExecuteQueryOperator
+from airflow.sdk import DAG
+
+DAG_ID = "example_dq_ruleset_from_yaml"
+CONN_ID = "sqlite_default"
+TABLE_NAME = "dq_yaml_orders"
+RESULTS_PATH = Path("/tmp/airflow_dq_example/results")
+RULESET_FILE = Path(__file__).parent / "orders_ruleset.yaml"
+
+os.environ.setdefault("AIRFLOW__DQ__RESULTS_PATH", f"file://{RESULTS_PATH}")
Review Comment:
The provider reads `results_path` from the `[common.dataquality]` config section, so `AIRFLOW__DQ__RESULTS_PATH` will not configure the results backend. Use `AIRFLOW__COMMON_DATAQUALITY__RESULTS_PATH` here so the example actually persists results.
##########
providers/common/dataquality/src/airflow/providers/common/dataquality/example_dags/example_dq_require_quality.py:
##########
@@ -0,0 +1,120 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+"""
+Example DAGs: gate a consumer Dag on the data quality score of the asset that triggers it.
+
+Two Dags, wired together only through the ``dq_gated_orders`` asset:
+
+- ``example_dq_require_quality_producer`` checks a table and produces the asset.
+ :class:`~airflow.providers.common.dataquality.operators.dq_check.DQCheckOperator` attaches its summary
+ (including the quality ``score``) to the asset event automatically because the asset carries
+ quality config attached with :func:`~airflow.providers.common.dataquality.assets.asset_quality`.
+- ``example_dq_require_quality_consumer`` is scheduled by that asset. Its first task, built by
+ :func:`~airflow.providers.common.dataquality.assets.require_quality`, reads the score off the triggering
+ asset event and short-circuits -- skipping every downstream task -- when it's missing or
+ below ``min_score``. Downstream tasks never see a run triggered by a bad batch of data.
+"""
+
+from __future__ import annotations
+
+import os
+from datetime import datetime
+from pathlib import Path
+
+from airflow.providers.common.dataquality.assets import asset_quality, require_quality
+from airflow.providers.common.dataquality.operators.dq_check import DQCheckOperator
+from airflow.providers.common.dataquality.rules import DQRule, RuleSet
+from airflow.providers.common.sql.operators.sql import SQLExecuteQueryOperator
+from airflow.sdk import DAG, Asset, task
+
+CONN_ID = "sqlite_default"
+TABLE_NAME = "dq_gated_orders"
+RESULTS_PATH = Path("/tmp/airflow_dq_example/results")
+
+os.environ.setdefault("AIRFLOW__DQ__RESULTS_PATH", f"file://{RESULTS_PATH}")
Review Comment:
The provider reads `results_path` from the `[common.dataquality]` config section (env var name is derived from `section.replace('.', '_')`), so `AIRFLOW__DQ__RESULTS_PATH` will not configure the results backend. Use `AIRFLOW__COMMON_DATAQUALITY__RESULTS_PATH` here so the example actually persists results.
##########
airflow-core/tests/unit/always/test_project_structure.py:
##########
@@ -107,6 +107,7 @@ def test_providers_modules_should_have_tests(self):
"providers/common/compat/tests/unit/common/compat/standard/test_utils.py",
"providers/common/messaging/tests/unit/common/messaging/providers/test_base_provider.py",
"providers/common/messaging/tests/unit/common/messaging/providers/test_sqs.py",
+ "providers/common/dataquality/tests/unit/common/dataquality/test_exceptions.py",
Review Comment:
This adds `providers/common/dataquality/tests/unit/common/dataquality/test_exceptions.py` to `OVERLOOKED_TESTS`, which suppresses the missing-tests check rather than fixing it. Since this PR introduces the `exceptions.py` module, please add the missing `test_exceptions.py` file (even a minimal import/instantiation test) and remove this entry from `OVERLOOKED_TESTS`.
##########
providers/common/dataquality/src/airflow/providers/common/dataquality/example_dags/example_dq_check_custom_sql.py:
##########
@@ -0,0 +1,112 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+"""
+Example DAG: ``custom_sql`` for checks the built-in catalog can't express.
+
+Built-in checks (``null_count``, ``min``, ``max``, ...) are all single-column. The moment a
+rule needs to compare two columns, join against another table, or use a SQL function the
+built-in catalog doesn't cover (or that your database's ``DbApiHook`` doesn't support -- see
+"Supported checks and databases" in the provider docs), ``custom_sql`` is the escape hatch:
+any statement that resolves to a single scalar, evaluated the same way as a built-in check.
+"""
+
+from __future__ import annotations
+
+import os
+from datetime import datetime
+from pathlib import Path
+
+from airflow.providers.common.dataquality.operators.dq_check import DQCheckOperator
+from airflow.providers.common.dataquality.rules import DQRule, RuleSet, Severity
+from airflow.providers.common.sql.operators.sql import SQLExecuteQueryOperator
+from airflow.sdk import DAG
+
+DAG_ID = "example_dq_check_custom_sql"
+CONN_ID = "sqlite_default"
+TABLE_NAME = "dq_custom_sql_orders"
+RESULTS_PATH = Path("/tmp/airflow_dq_example/results")
+
+os.environ.setdefault("AIRFLOW__DQ__RESULTS_PATH", f"file://{RESULTS_PATH}")
Review Comment:
The provider reads `results_path` from the `[common.dataquality]` config section, so `AIRFLOW__DQ__RESULTS_PATH` will not configure the results backend. Use `AIRFLOW__COMMON_DATAQUALITY__RESULTS_PATH` here so the example actually persists results.
##########
providers/common/dataquality/src/airflow/providers/common/dataquality/example_dags/example_dq_check_decorator_dynamic.py:
##########
@@ -0,0 +1,94 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+"""
+Example DAG: ``@task.dq_check`` with a runtime ruleset.
+
+``table`` (or ``asset``) is declared as a decorator argument, exactly like the plain operator.
+``ruleset`` may be declared at Dag-parse time, or returned by the decorated function at
+execution time. This is useful when an upstream task, variable, or generated value decides which
+ruleset to run.
+"""
+
+from __future__ import annotations
+
+import os
+from datetime import datetime
+from pathlib import Path
+
+from airflow.providers.common.dataquality.rules import DQRule, RuleSet
+from airflow.providers.common.sql.operators.sql import SQLExecuteQueryOperator
+from airflow.sdk import DAG, task
+
+DAG_ID = "example_dq_check_decorator_dynamic"
+CONN_ID = "sqlite_default"
+TABLE_NAME = "dq_dynamic_orders"
+RESULTS_PATH = Path("/tmp/airflow_dq_example/results")
+
+os.environ.setdefault("AIRFLOW__DQ__RESULTS_PATH", f"file://{RESULTS_PATH}")
Review Comment:
The provider reads `results_path` from the `[common.dataquality]` config section, so `AIRFLOW__DQ__RESULTS_PATH` will not configure the results backend. Use `AIRFLOW__COMMON_DATAQUALITY__RESULTS_PATH` here so the example actually persists results.
##########
providers/common/dataquality/src/airflow/providers/common/dataquality/example_dags/example_dq_llm_generated_ruleset.py:
##########
@@ -0,0 +1,128 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+"""
+Example DAG: generate a ``RuleSet`` with an LLM, guided by the ``dataquality-rule-authoring`` skill.
+
+Requires the optional ``apache-airflow-providers-common-ai[skills]`` package and a configured
+LLM connection (``llm_conn_id``); this Dag is not registered when common.ai isn't installed.
+
+An LLM is asked to propose data quality rules for a table's columns. Its ``system_prompt``
+points it at the ``dataquality-rule-authoring`` skill shipped with this provider (via
+``AgentSkillsToolset``), so it knows the exact ``RuleSet``/``DQRule`` schema, the built-in check
+catalog, and the ``Condition`` grammar, instead of guessing at field or check names.
+``output_type=RuleSet`` makes pydantic-ai validate -- and self-correct -- the model's output
+against the real ``RuleSet`` model before the task completes, so a malformed rule never reaches
+``DQCheckOperator`` in the first place.
+
+The generated ``RuleSet`` is then returned by ``@task.dq_check`` at runtime, since the real
+ruleset doesn't exist until the LLM task runs.
+"""
+
+from __future__ import annotations
+
+import os
+from datetime import datetime
+from pathlib import Path
+
+from airflow.providers.common.dataquality.rules import RuleSet
+from airflow.providers.common.sql.operators.sql import SQLExecuteQueryOperator
+from airflow.sdk import DAG, task
+
+try:
+ from airflow.providers.common.ai.toolsets.skills import AgentSkillsToolset
+except ImportError:
+ AgentSkillsToolset = None # type: ignore[assignment,misc]
+
+DAG_ID = "example_dq_llm_generated_ruleset"
+CONN_ID = "sqlite_default"
+TABLE_NAME = "dq_llm_orders"
+RESULTS_PATH = Path("/tmp/airflow_dq_example/results")
+# The skill ships next to this Dag's package, not next to this file -- resolve relative to
+# __file__ so the path holds regardless of the Dag processor's working directory.
+SKILLS_DIR = Path(__file__).parent.parent / "skills" / "dataquality-rule-authoring"
+
+os.environ.setdefault("AIRFLOW__DQ__RESULTS_PATH", f"file://{RESULTS_PATH}")
Review Comment:
The provider reads `results_path` from the `[common.dataquality]` config section, so `AIRFLOW__DQ__RESULTS_PATH` will not configure the results backend. Use `AIRFLOW__COMMON_DATAQUALITY__RESULTS_PATH` here so the example actually persists results.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Airflow Data Quality Provider part 1 [airflow]
Posted by "gopidesupavan (via GitHub)" <gi...@apache.org>.
gopidesupavan commented on code in PR #69575:
URL: https://github.com/apache/airflow/pull/69575#discussion_r3602297209
##########
providers/common/dataquality/src/airflow/providers/common/dataquality/backends/object_storage.py:
##########
@@ -0,0 +1,256 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+"""
+Object-storage results backend.
+
+Each DQ check writes a keyed JSON document plus read indexes optimized for the UI:
+
+ runs/by_task/dag_id=<dag>/task_id=<task>/date=<2026-07-04>/<started_at>__<run_uid>.json
+ Canonical run record: ``{"run": ..., "results": [...], "summary": ...}``.
+
+ runs/by_task_instance/dag_id=<dag>/task_id=<task>/<safe_run_id>__<map_index>.json
+ Latest result for a task-instance page. Last write wins across retries.
+
+ rules/by_task_rule/dag_id=<dag>/task_id=<task>/rule_uid=<uid>/<started_at>__<run_uid>.json
+ One rule result plus run context: ``{"run": ..., "result": ...}``.
+
+The duplicate files are intentional read indexes: DQ tasks write once, while the UI reads
+many times. Keeping these indexes avoids scanning all task runs for common UI views.
+"""
+
+from __future__ import annotations
+
+import json
+import logging
+from datetime import datetime, timezone
+from typing import Any
+
+from airflow.providers.common.dataquality.results import DQRun, RuleResult, build_summary
+from airflow.sdk import ObjectStoragePath
+
+log = logging.getLogger(__name__)
+
+
+class ObjectStorageResultsBackend:
+ """Persist DQ results as JSON files via ``ObjectStoragePath``."""
+
+ def __init__(self, results_path: str, conn_id: str | None = None) -> None:
+ self.root = ObjectStoragePath(results_path, conn_id=conn_id)
+
+ def write_run(self, run: DQRun, results: list[RuleResult]) -> None:
+ timestamp = run.started_at or datetime.now(tz=timezone.utc).isoformat()
+ compact_ts = self._get_safe_key(timestamp)
+ payload = self._build_run_payload(run, results)
+
+ self._write_run_file(run, timestamp[:10], compact_ts, payload)
+ self._write_task_instance_index(run, payload)
+ self._write_rule_indexes(run, results, timestamp)
+
+ def read_task_rule_history(
+ self, dag_id: str, task_id: str, rule_uid: str, limit: int = 100, before: str | None = None
+ ) -> dict[str, Any]:
+ """Return recent results for one rule produced by one task, newest first."""
+ rule_dir = (
+ self.root
+ / "rules"
+ / "by_task_rule"
+ / f"dag_id={dag_id}"
+ / f"task_id={task_id}"
+ / f"rule_uid={rule_uid}"
+ )
+ return self._read_rule_history_dir(rule_dir, limit, before)
+
+ def read_task_runs(
+ self, dag_id: str, task_id: str, limit: int = 50, before: str | None = None
+ ) -> dict[str, Any]:
+ """
+ Return recent data quality runs for one task, newest first.
+
+ Returns ``{"items": [...], "next_cursor": ...}``. ``before`` is the opaque
+ ``next_cursor`` from the previous page, so "load more" only reads runs it hasn't
+ shown yet. ``next_cursor`` is set by reading one extra entry past ``limit``: cheap
+ on every page except the last one, where there's no way to confirm history is
+ exhausted without walking to the end.
+
+ ``date=`` partition names sort correctly as plain strings, so directories are walked
+ newest-first. Filenames are ``{started_at}__{run_uid}.json``, so sorting each partition's
+ filenames before scanning also walks newest-first within the partition -- required for
+ the early exit below to stop on the actual newest runs rather than an arbitrary subset.
+ Scanning stops as soon as ``limit + 1`` matching runs have been collected — a task with
+ years of history doesn't pay for a full scan on every page.
+ """
+ task_dir = self.root / "runs" / "by_task" / f"dag_id={dag_id}" / f"task_id={task_id}"
+ if not task_dir.exists():
+ return {"items": [], "next_cursor": None}
+
+ date_dirs = sorted((path for path in task_dir.iterdir() if path.is_dir()), reverse=True)
+ runs = []
+ for date_dir in date_dirs:
+ for path in sorted(date_dir.iterdir(), key=lambda p: p.name, reverse=True):
+ if not path.name.endswith(".json"):
+ continue
+ payload = self._read_json(path)
+ if payload is None:
+ continue
+ cursor = self._get_run_payload_cursor(payload)
+ if before is not None and cursor >= before:
+ continue
Review Comment:
good suggestion applied https://github.com/apache/airflow/pull/69575/commits/4165252568d11c57c549d8450ad7d8108e221b24
##########
providers/common/dataquality/src/airflow/providers/common/dataquality/backends/object_storage.py:
##########
@@ -0,0 +1,256 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+"""
+Object-storage results backend.
+
+Each DQ check writes a keyed JSON document plus read indexes optimized for the UI:
+
+ runs/by_task/dag_id=<dag>/task_id=<task>/date=<2026-07-04>/<started_at>__<run_uid>.json
+ Canonical run record: ``{"run": ..., "results": [...], "summary": ...}``.
+
+ runs/by_task_instance/dag_id=<dag>/task_id=<task>/<safe_run_id>__<map_index>.json
+ Latest result for a task-instance page. Last write wins across retries.
+
+ rules/by_task_rule/dag_id=<dag>/task_id=<task>/rule_uid=<uid>/<started_at>__<run_uid>.json
+ One rule result plus run context: ``{"run": ..., "result": ...}``.
+
+The duplicate files are intentional read indexes: DQ tasks write once, while the UI reads
+many times. Keeping these indexes avoids scanning all task runs for common UI views.
+"""
+
+from __future__ import annotations
+
+import json
+import logging
+from datetime import datetime, timezone
+from typing import Any
+
+from airflow.providers.common.dataquality.results import DQRun, RuleResult, build_summary
+from airflow.sdk import ObjectStoragePath
+
+log = logging.getLogger(__name__)
+
+
+class ObjectStorageResultsBackend:
+ """Persist DQ results as JSON files via ``ObjectStoragePath``."""
+
+ def __init__(self, results_path: str, conn_id: str | None = None) -> None:
+ self.root = ObjectStoragePath(results_path, conn_id=conn_id)
+
+ def write_run(self, run: DQRun, results: list[RuleResult]) -> None:
+ timestamp = run.started_at or datetime.now(tz=timezone.utc).isoformat()
+ compact_ts = self._get_safe_key(timestamp)
+ payload = self._build_run_payload(run, results)
+
+ self._write_run_file(run, timestamp[:10], compact_ts, payload)
+ self._write_task_instance_index(run, payload)
+ self._write_rule_indexes(run, results, timestamp)
+
+ def read_task_rule_history(
+ self, dag_id: str, task_id: str, rule_uid: str, limit: int = 100, before: str | None = None
+ ) -> dict[str, Any]:
+ """Return recent results for one rule produced by one task, newest first."""
+ rule_dir = (
+ self.root
+ / "rules"
+ / "by_task_rule"
+ / f"dag_id={dag_id}"
+ / f"task_id={task_id}"
+ / f"rule_uid={rule_uid}"
+ )
+ return self._read_rule_history_dir(rule_dir, limit, before)
+
+ def read_task_runs(
+ self, dag_id: str, task_id: str, limit: int = 50, before: str | None = None
+ ) -> dict[str, Any]:
+ """
+ Return recent data quality runs for one task, newest first.
+
+ Returns ``{"items": [...], "next_cursor": ...}``. ``before`` is the opaque
+ ``next_cursor`` from the previous page, so "load more" only reads runs it hasn't
+ shown yet. ``next_cursor`` is set by reading one extra entry past ``limit``: cheap
+ on every page except the last one, where there's no way to confirm history is
+ exhausted without walking to the end.
+
+ ``date=`` partition names sort correctly as plain strings, so directories are walked
+ newest-first. Filenames are ``{started_at}__{run_uid}.json``, so sorting each partition's
+ filenames before scanning also walks newest-first within the partition -- required for
+ the early exit below to stop on the actual newest runs rather than an arbitrary subset.
+ Scanning stops as soon as ``limit + 1`` matching runs have been collected — a task with
+ years of history doesn't pay for a full scan on every page.
+ """
+ task_dir = self.root / "runs" / "by_task" / f"dag_id={dag_id}" / f"task_id={task_id}"
+ if not task_dir.exists():
+ return {"items": [], "next_cursor": None}
+
+ date_dirs = sorted((path for path in task_dir.iterdir() if path.is_dir()), reverse=True)
+ runs = []
+ for date_dir in date_dirs:
+ for path in sorted(date_dir.iterdir(), key=lambda p: p.name, reverse=True):
+ if not path.name.endswith(".json"):
+ continue
+ payload = self._read_json(path)
+ if payload is None:
+ continue
+ cursor = self._get_run_payload_cursor(payload)
+ if before is not None and cursor >= before:
+ continue
+ runs.append(payload)
+ if len(runs) > limit:
+ break
+ if len(runs) > limit:
+ break
+
+ ordered = sorted(runs, key=self._get_run_payload_cursor, reverse=True)
+ page = ordered[:limit]
+ next_cursor = self._get_run_payload_cursor(page[-1]) if len(ordered) > limit and page else None
+ return {"items": page, "next_cursor": next_cursor}
Review Comment:
updated
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Airflow Data Quality Provider part 1 [airflow]
Posted by "gopidesupavan (via GitHub)" <gi...@apache.org>.
gopidesupavan commented on code in PR #69575:
URL: https://github.com/apache/airflow/pull/69575#discussion_r3575180649
##########
providers/common/dataquality/src/airflow/providers/common/dataquality/skills/dataquality-rule-authoring/SKILL.md:
##########
@@ -0,0 +1,173 @@
+---
+name: dataquality-rule-authoring
+description: "Generate a RuleSet/DQRule JSON payload for Apache Airflow's dataquality provider from a table's column definitions. Use this whenever asked to write, generate, or suggest data quality rules, checks, or a ruleset for a table or dataset -- especially when the output is structured JSON handed to DQCheckOperator, @task.dq_check, asset_quality(), or RuleSet.from_file()."
+---
+
+<!-- SPDX-License-Identifier: Apache-2.0
+ https://www.apache.org/licenses/LICENSE-2.0 -->
+
+# dataquality rule authoring
+
+You are producing a **RuleSet**: a JSON object naming a table's data quality checks. This
+document is the schema and the full list of valid values -- do not guess field names or check
+names, and do not invent checks that aren't in the catalog below.
+
+## Output shape
+
+```json
+{
+ "name": "orders_quality",
+ "rules": [
+ {
+ "name": "order_id_not_null",
+ "check": "null_count",
+ "column": "order_id",
+ "condition": {"equal_to": 0}
+ }
+ ]
+}
+```
+
+`name` (ruleset name) and `rules` (array) are the only top-level keys. Every rule name must be
+unique within the ruleset.
+
+## `DQRule` fields
+
+| Field | Required | Notes |
+|---|---|---|
+| `name` | yes | Unique within the ruleset. Shows up in data quality results and logs. |
+| `check` | yes | One of the catalog names below, or `"custom_sql"`. Exact string match -- there is no fuzzy matching. |
+| `condition` | yes* | See `Condition` grammar below. *Every catalog check currently requires one explicitly (no defaults) -- always include it. |
+| `column` | check-dependent | Required for every catalog check except `row_count`. Not used with `custom_sql`. |
+| `sql` | `custom_sql` only | A SQL statement returning a single scalar. May reference the checked table as `{table}`. Invalid together with any catalog check. |
+| `severity` | no | `"error"` (default) or `"warn"`. `"error"` fails the task on a failing rule (subject to the operator's `fail_on`); `"warn"` only records it. |
+| `partition_clause` | no | Extra SQL predicate ANDed into this rule's `WHERE` clause, e.g. `"region = 'EU'"`. |
+| `previous_name` | no | Only set when told a rule is being renamed, to keep its history continuous. |
+| `description` | no | Human-readable text shown in data quality results. Use it when a clear business meaning is known; otherwise omit it and let Airflow generate a default description. |
+| `dimension` | no | One of `completeness`, `uniqueness`, `validity`, `freshness`, `volume`, `consistency`. Defaults to the check's catalog dimension (`validity` for `custom_sql`) -- only set this explicitly when a `custom_sql` rule measures something the default doesn't capture (e.g. a freshness check written as `custom_sql` should set `"dimension": "freshness"`). Leave it unset for every built-in check. |
+
+**Do not add any key not listed above.** The schema rejects unrecognized fields.
+
+## Built-in check catalog
+
+| `check` | SQL expression | needs `column` | typical use |
+|---|---|---|---|
+| `null_count` | `SUM(CASE WHEN {column} IS NULL THEN 1 ELSE 0 END)` | yes | count of nulls in a column |
+| `null_ratio` | `SUM(CASE WHEN {column} IS NULL THEN 1.0 ELSE 0.0 END) / COUNT(*)` | yes | fraction of nulls, 0.0-1.0 |
+| `distinct_count` | `COUNT(DISTINCT {column})` | yes | number of distinct values |
+| `unique_violations` | `COUNT({column}) - COUNT(DISTINCT {column})` | yes | 0 means the column is fully unique |
+| `min` | `MIN({column})` | yes | minimum value |
+| `max` | `MAX({column})` | yes | maximum value |
+| `mean` | `AVG({column})` | yes | average value |
+| `row_count` | `COUNT(*)` | no | total row count -- omit `column` entirely |
+
+These are the *only* built-in check names. Anything else -- comparing two columns, joining
+another table, checking a format/regex, freshness against `now()`, referential integrity -- must
+be written as `custom_sql`.
+
+## `Condition` grammar
+
+`condition` is an object with these optional numeric keys; at least one is required:
+
+- `equal_to` -- exact match. Cannot be combined with any other key below.
+- `greater_than` / `geq_to` -- lower bound, exclusive / inclusive.
+- `less_than` / `leq_to` -- upper bound, exclusive / inclusive.
+- `tolerance` -- a fraction (e.g. `0.1` for 10%) that widens the comparison bounds.
+
+Combine `greater_than`/`less_than`/`geq_to`/`leq_to` freely to express a range, e.g.
+`{"geq_to": 0, "leq_to": 100}`. Never combine `equal_to` with another comparison key (other than
+`tolerance`).
+
+## `custom_sql`: when a catalog check doesn't fit
Review Comment:
true.. updated
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Airflow Data Quality Provider part 1 [airflow]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #69575:
URL: https://github.com/apache/airflow/pull/69575#issuecomment-4983499751
<!-- uv-lock-rebase-notice -->
`uv.lock` on `main` just moved via [#69924](https://github.com/apache/airflow/pull/69924) ("Support excluding files from `common.ai` Agent Skills discovery"), commit [`663a792`](https://github.com/apache/airflow/commit/663a79203891e04111e4a0cea079cc34011a9ad9) and this PR currently conflicts.
Quickest fix:
```bash
git fetch upstream main && git rebase upstream/main
rm uv.lock && uv lock
git add uv.lock && git rebase --continue
git push --force-with-lease
```
_Automated nudge — ignore if you're not ready to rebase. This comment is updated in place on future `uv.lock` bumps._
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Airflow Data Quality Provider part 1 [airflow]
Posted by "gopidesupavan (via GitHub)" <gi...@apache.org>.
gopidesupavan commented on code in PR #69575:
URL: https://github.com/apache/airflow/pull/69575#discussion_r3602306724
##########
providers/common/dataquality/src/airflow/providers/common/dataquality/assets.py:
##########
@@ -0,0 +1,190 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+"""
+Asset-level data quality declarations.
+
+Quality configuration lives inside ``Asset.extra`` under the ``airflow.dataquality`` key, so it is
+serialized with the Dag and needs no Airflow core changes. The rules travel with the asset
+definition instead of being scattered across the Dags that check it.
+"""
+
+from __future__ import annotations
+
+import logging
+from typing import TYPE_CHECKING, Any
+
+from airflow.providers.common.dataquality.exceptions import DQRuleValidationError
+from airflow.providers.common.dataquality.rules import RuleSet
+from airflow.sdk import task
+
+if TYPE_CHECKING:
+ from collections.abc import Mapping, Sequence
+
+ from airflow.sdk import Asset
+ from airflow.sdk.execution_time.comms import AssetEventDagRunReferenceResult
+
+log = logging.getLogger(__name__)
+
+DQ_EXTRA_KEY = "airflow.dataquality"
+DQ_RESULT_EXTRA_KEY = "airflow.dataquality.result"
+
+
+def asset_quality(
+ asset: Asset,
Review Comment:
atleast for the first version its Asset, have not thought about AssetRef, AssetAlias. but there no harm in supporting that but will have to check.. will add a item for future track
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Airflow Data Quality Provider part 1 [airflow]
Posted by "gopidesupavan (via GitHub)" <gi...@apache.org>.
gopidesupavan commented on PR #69575:
URL: https://github.com/apache/airflow/pull/69575#issuecomment-4909333828
This is part 1 , the UI plugin is part of this big PR https://github.com/apache/airflow/pull/69413, i will ship that separate. but to show how the UI looks like here is some screenshots https://github.com/apache/airflow/pull/69413#issuecomment-4886311468 (this is minimal what had done with help of UI not an UI expert anyone can modify or help with this)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Airflow Data Quality Provider part 1 [airflow]
Posted by "gopidesupavan (via GitHub)" <gi...@apache.org>.
gopidesupavan commented on PR #69575:
URL: https://github.com/apache/airflow/pull/69575#issuecomment-4989422267
sorry @Lee-W i meant to add and it got removed, readded back.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Airflow Data Quality Provider part 1 [airflow]
Posted by "kaxil (via GitHub)" <gi...@apache.org>.
kaxil commented on code in PR #69575:
URL: https://github.com/apache/airflow/pull/69575#discussion_r3574638256
##########
providers/common/dataquality/src/airflow/providers/common/dataquality/engines/sql.py:
##########
@@ -0,0 +1,137 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+"""SQL execution engine: compiles a ruleset into check queries run through a DB-API hook."""
+
+from __future__ import annotations
+
+import logging
+import time
+from dataclasses import dataclass
+from typing import TYPE_CHECKING, Any
+
+from airflow.providers.common.dataquality.rules import CUSTOM_SQL_CHECK
+from airflow.providers.common.dataquality.rules.checks import CHECK_SPECS
+
+if TYPE_CHECKING:
+ from airflow.providers.common.dataquality.rules import DQRule, RuleSet
+ from airflow.providers.common.sql.hooks.sql import DbApiHook
+
+log = logging.getLogger(__name__)
+
+
+@dataclass(frozen=True)
+class Observation:
+ """Raw value a rule observed, before its condition is evaluated."""
+
+ rule: DQRule
+ observed_value: Any = None
+ duration_ms: float | None = None
+ error_message: str | None = None
+ sql: str | None = None
+
+
+class SQLDQEngine:
+ """
+ Runs built-in checks as a single UNION ALL query and custom SQL rules individually.
+
+ Table, column, and partition-clause values come from the Dag author and are interpolated
+ into SQL the same way the ``common.sql`` check operators do — they are trusted input.
+ """
+
+ check_sql_template = "SELECT '{rule_uid}' AS rule_uid, {expression} AS observed FROM {table}{where}"
+
+ def __init__(self, hook: DbApiHook) -> None:
+ self.hook = hook
+
+ def measure(self, ruleset: RuleSet, table: str, partition_clause: str | None = None) -> list[Observation]:
+ builtin_rules = [rule for rule in ruleset.rules if rule.check != CUSTOM_SQL_CHECK]
+ custom_rules = [rule for rule in ruleset.rules if rule.check == CUSTOM_SQL_CHECK]
+
+ observations = []
+ if builtin_rules:
+ observations.extend(self._measure_builtin(builtin_rules, table, partition_clause))
+ for rule in custom_rules:
+ observations.append(self._measure_custom(rule, table))
+ return observations
+
+ def build_batch_sql(self, rules: list[DQRule], table: str, partition_clause: str | None) -> str:
+ return " UNION ALL ".join(self.build_rule_sql(rule, table, partition_clause) for rule in rules)
+
+ def build_rule_sql(self, rule: DQRule, table: str, partition_clause: str | None = None) -> str:
+ """Build the SQL used to measure one built-in rule."""
+ expression = CHECK_SPECS[rule.check].expression.format(column=rule.column)
+ predicates = [p for p in (partition_clause, rule.partition_clause) if p]
+ where = f" WHERE {' AND '.join(predicates)}" if predicates else ""
+ return self.check_sql_template.format(
+ rule_uid=rule.rule_uid, expression=expression, table=table, where=where
+ )
+
+ def _measure_builtin(
+ self, rules: list[DQRule], table: str, partition_clause: str | None
+ ) -> list[Observation]:
+ sql = self.build_batch_sql(rules, table, partition_clause)
+ log.info("Running %d built-in checks against %s", len(rules), table)
+ started = time.monotonic()
+ try:
+ records = self.hook.get_records(sql)
+ except Exception as e:
Review Comment:
Since all built-ins run as one UNION ALL, a single bad rule (say an LLM-generated `MIN(no_such_column)`) fails the whole batch here, and every other rule gets recorded as ERROR with that unrelated database message. `_raise_for_failures` then fails the task on ERROR regardless of `fail_on`, so one bad generated rule takes down rules that would have passed, and their persisted history shows an error that isn't theirs -- which cuts against "Every rule is evaluated and recorded regardless of the task outcome" in the operator docstring. A per-rule fallback when the batch errors would keep the batching win on the happy path and isolate the blast radius. Related: `duration_ms` on each observation is the whole batch's elapsed time, which will mislead once the UI charts per-rule durations.
##########
providers/common/dataquality/src/airflow/providers/common/dataquality/backends/object_storage.py:
##########
@@ -0,0 +1,252 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+"""
+Object-storage results backend.
+
+Each DQ check writes a keyed JSON document plus read indexes optimized for the UI:
+
+ runs/by_task/dag_id=<dag>/task_id=<task>/date=<2026-07-04>/<run_uid>.json
+ Canonical run record: ``{"run": ..., "results": [...], "summary": ...}``.
+
+ runs/by_task_instance/dag_id=<dag>/task_id=<task>/<safe_run_id>__<map_index>.json
+ Latest result for a task-instance page. Last write wins across retries.
+
+ rules/by_task_rule/dag_id=<dag>/task_id=<task>/rule_uid=<uid>/<started_at>__<run_uid>.json
+ One rule result plus run context: ``{"run": ..., "result": ...}``.
+
+The duplicate files are intentional read indexes: DQ tasks write once, while the UI reads
+many times. Keeping these indexes avoids scanning all task runs for common UI views.
+"""
+
+from __future__ import annotations
+
+import json
+import logging
+from datetime import datetime, timezone
+from typing import Any
+
+from airflow.providers.common.dataquality.results import DQRun, RuleResult, build_summary
+from airflow.sdk import ObjectStoragePath
+
+log = logging.getLogger(__name__)
+
+
+class ObjectStorageResultsBackend:
+ """Persist DQ results as JSON files via ``ObjectStoragePath``."""
+
+ def __init__(self, results_path: str, conn_id: str | None = None) -> None:
+ self.root = ObjectStoragePath(results_path, conn_id=conn_id)
+
+ def write_run(self, run: DQRun, results: list[RuleResult]) -> None:
+ timestamp = run.started_at or datetime.now(tz=timezone.utc).isoformat()
+ payload = self._build_run_payload(run, results)
+
+ self._write_run_file(run, timestamp[:10], payload)
+ self._write_task_instance_index(run, payload)
+ self._write_rule_indexes(run, results, timestamp)
+
+ def read_task_rule_history(
+ self, dag_id: str, task_id: str, rule_uid: str, limit: int = 100, before: str | None = None
+ ) -> dict[str, Any]:
+ """Return recent results for one rule produced by one task, newest first."""
+ rule_dir = (
+ self.root
+ / "rules"
+ / "by_task_rule"
+ / f"dag_id={dag_id}"
+ / f"task_id={task_id}"
+ / f"rule_uid={rule_uid}"
+ )
+ return self._read_rule_history_dir(rule_dir, limit, before)
+
+ def read_task_runs(
+ self, dag_id: str, task_id: str, limit: int = 50, before: str | None = None
+ ) -> dict[str, Any]:
+ """
+ Return recent data quality runs for one task, newest first.
+
+ Returns ``{"items": [...], "next_cursor": ...}``. ``before`` is the opaque
+ ``next_cursor`` from the previous page, so "load more" only reads runs it hasn't
+ shown yet. ``next_cursor`` is set by reading one extra entry past ``limit``: cheap
+ on every page except the last one, where there's no way to confirm history is
+ exhausted without walking to the end.
+
+ ``date=`` partition names sort correctly as plain strings, so directories are walked
+ newest-first and scanning stops as soon as ``limit + 1`` matching runs have been
+ collected — a task with years of history doesn't pay for a full scan on every page.
+ """
+ task_dir = self.root / "runs" / "by_task" / f"dag_id={dag_id}" / f"task_id={task_id}"
+ if not task_dir.exists():
+ return {"items": [], "next_cursor": None}
+
+ date_dirs = sorted((path for path in task_dir.iterdir() if path.is_dir()), reverse=True)
+ runs = []
+ for date_dir in date_dirs:
+ for path in date_dir.iterdir():
Review Comment:
The early exit here breaks newest-first within a date partition: `date_dir.iterdir()` has no ordering and the filenames are random `run_uid`s, so once a single day holds more than `limit` runs (a mapped check writes one file per map_index into the same partition, and retries add more), this collects an arbitrary `limit+1` subset, sorts only that subset, and sets `next_cursor` past runs it never read -- the genuinely newest runs become permanently unreachable on later pages. The rule-history path avoids this only because its filenames are timestamp-prefixed, so its sort-then-scan is chronological. I'd prefix the by_task filename with `started_at` the same way (`{compact_ts}__{run_uid}.json`) and sort each partition before the early exit. Worth doing before the first release since this layout is the persisted format -- changing filenames later strands existing history. The same-day test at `test_object_storage.py:294` only asserts count and cursor presence; once fixed it should asse
rt *which* run comes back, like the cross-date test above it does.
##########
providers/common/dataquality/src/airflow/providers/common/dataquality/rules/rule.py:
##########
@@ -0,0 +1,289 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+"""Declarative data quality rule model: rules are data, not code."""
+
+from __future__ import annotations
+
+import hashlib
+import json
+from enum import Enum
+from typing import Any, cast
+
+import yaml
+from pydantic import BaseModel, ConfigDict, Field, model_validator
+
+from airflow.providers.common.dataquality.exceptions import DQRuleValidationError
+from airflow.providers.common.dataquality.rules.checks import CHECK_SPECS, Dimension
+
+CUSTOM_SQL_CHECK = "custom_sql"
+
+
+class Severity(str, Enum):
+ """Severity used to decide whether a failing rule fails the task."""
+
+ WARN = "warn"
+ ERROR = "error"
+
+
+class Condition(BaseModel):
+ """
+ Pass/fail condition evaluated against a rule's observed value.
+
+ Uses the same grammar as the ``common.sql`` check operators: ``equal_to``,
+ ``greater_than``, ``less_than``, ``geq_to``, ``leq_to``, plus a percentage
+ ``tolerance`` that widens comparisons.
+
+ """
+
+ model_config = ConfigDict(frozen=True, extra="forbid")
+
+ equal_to: float | None = None
+ greater_than: float | None = None
+ less_than: float | None = None
+ geq_to: float | None = None
+ leq_to: float | None = None
+ tolerance: float | None = None
+
+ @model_validator(mode="after")
+ def _validate_comparisons(self) -> Condition:
+ comparisons = {
+ "equal_to": self.equal_to,
+ "greater_than": self.greater_than,
+ "less_than": self.less_than,
+ "geq_to": self.geq_to,
+ "leq_to": self.leq_to,
+ }
+ set_comparisons = {name for name, value in comparisons.items() if value is not None}
+ if not set_comparisons:
+ raise ValueError(f"Condition needs at least one comparison out of: {', '.join(comparisons)}")
+ if self.equal_to is not None and len(set_comparisons) > 1:
+ raise ValueError("equal_to cannot be combined with other comparisons")
+ return self
+
+ @classmethod
+ def from_dict(cls, data: dict[str, Any]) -> Condition:
+ return cls(**data)
+
+ def to_dict(self) -> dict[str, float]:
+ data = {
+ "equal_to": self.equal_to,
+ "greater_than": self.greater_than,
+ "less_than": self.less_than,
+ "geq_to": self.geq_to,
+ "leq_to": self.leq_to,
+ "tolerance": self.tolerance,
+ }
+ return {k: v for k, v in data.items() if v is not None}
+
+ def evaluate(self, observed: Any) -> bool:
+ if observed is None:
+ return False
+ value = float(observed)
+ if self.equal_to is not None:
+ if self.tolerance is not None:
+ delta = abs(self.equal_to) * self.tolerance
+ return self.equal_to - delta <= value <= self.equal_to + delta
+ return value == self.equal_to
+ geq_to = self.geq_to
+ greater_than = self.greater_than
+ leq_to = self.leq_to
+ less_than = self.less_than
+ if self.tolerance is not None:
+ if geq_to is not None:
+ geq_to -= abs(geq_to) * self.tolerance
+ if greater_than is not None:
+ greater_than -= abs(greater_than) * self.tolerance
+ if leq_to is not None:
+ leq_to += abs(leq_to) * self.tolerance
+ if less_than is not None:
+ less_than += abs(less_than) * self.tolerance
+ return (
+ (greater_than is None or value > greater_than)
+ and (less_than is None or value < less_than)
+ and (geq_to is None or value >= geq_to)
+ and (leq_to is None or value <= leq_to)
+ )
+
+
+class DQRule(BaseModel):
+ """
+ A single, named data quality rule.
+
+ :param name: Rule name, unique within its ruleset.
+ :param check: One of the built-in checks (see ``CHECK_SPECS``) or ``custom_sql``. Built-in
+ checks are plain ANSI SQL and are not guaranteed to work against every ``DbApiHook``;
+ see "Supported checks and databases" in the provider docs. Use ``custom_sql`` if a
+ built-in check doesn't fit your database's dialect.
+ :param condition: Pass condition for the observed value (``Condition`` or its dict form).
+ Optional only for checks whose ``CheckSpec.default_condition`` is set; every current
+ built-in check requires one explicitly.
+ :param column: Target column; required for column-level built-in checks.
+ :param sql: SQL statement returning a single scalar; required for ``custom_sql``.
+ May reference the target table as ``{table}``.
+ :param severity: ``error`` (fails the task by default) or ``warn`` (recorded only).
+ :param partition_clause: Extra predicate ANDed into the check's WHERE clause.
+ :param previous_name: Set when renaming a rule, to keep its history continuous.
+ :param description: Human-readable description shown in results and the UI. When omitted,
+ the provider generates a short default description from the rule and condition.
+
+ Invalid input raises pydantic's own :class:`~pydantic.ValidationError`.
+ """
+
+ model_config = ConfigDict(frozen=True, extra="forbid")
+
+ name: str
+ check: str = Field(description="One of the built-in checks, or custom_sql.")
+ condition: Condition | dict[str, Any] | None = None
Review Comment:
Probed this: `condition={"equal_to": 0}` coerces to `Condition` as intended, but `{"nonsense": 1}` or `{"tolerance": 0.1}` fail Condition validation and fall into the `dict[str, Any]` branch, surviving model construction as raw dicts -- then `rule_uid`/`to_dict()` crash with AttributeError instead of the ValidationError the docstring promises. Exactly the shape of mistake a generated ruleset makes. Typing this `Condition | None` gives you the dict coercion for free and rejects garbage at construction; the generated schema also loses its `additionalProperties: true` branch.
##########
providers/common/dataquality/src/airflow/providers/common/dataquality/assets.py:
##########
@@ -0,0 +1,172 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+"""
+Asset-level data quality declarations.
+
+Quality configuration lives inside ``Asset.extra`` under the ``airflow.dataquality`` key, so it is
+serialized with the Dag and needs no Airflow core changes. The rules travel with the asset
+definition instead of being scattered across the Dags that check it.
+"""
+
+from __future__ import annotations
+
+import logging
+from typing import TYPE_CHECKING, Any
+
+from airflow.providers.common.dataquality.exceptions import DQRuleValidationError
+from airflow.providers.common.dataquality.rules import RuleSet
+from airflow.sdk import task
+
+if TYPE_CHECKING:
+ from collections.abc import Mapping, Sequence
+
+ from airflow.sdk import Asset
+ from airflow.sdk.execution_time.comms import AssetEventDagRunReferenceResult
+
+log = logging.getLogger(__name__)
+
+DQ_EXTRA_KEY = "airflow.dataquality"
+DQ_RESULT_EXTRA_KEY = "airflow.dataquality.result"
+
+
+def asset_quality(
+ asset: Asset,
+ *,
+ ruleset: RuleSet | dict[str, Any] | str,
+ conn_id: str | None = None,
+ table: str | None = None,
+) -> Asset:
+ """
+ Attach data quality configuration to an asset, returning the same asset.
+
+ :param asset: The asset the rules describe.
+ :param ruleset: A :class:`~airflow.providers.common.dataquality.rules.RuleSet`, its dict form, or a path
+ to a YAML ruleset file (resolved eagerly, at Dag-parse time).
+ :param conn_id: Default connection a check operator should use for this asset.
+ :param table: Default table to check; falls back to the asset name when unset.
+
+ Usage::
+
+ orders = asset_quality(
+ Asset("orders", uri="postgres://warehouse/analytics/orders"),
+ ruleset=rules,
+ conn_id="warehouse",
+ table="analytics.orders",
+ )
+ check = DQCheckOperator(task_id="dq", asset=orders)
+ """
+ if isinstance(ruleset, str):
+ ruleset = RuleSet.from_file(ruleset)
+ elif isinstance(ruleset, dict):
+ ruleset = RuleSet.from_dict(ruleset)
+ config: dict[str, Any] = {"ruleset": ruleset.to_dict()}
+ if conn_id:
+ config["conn_id"] = conn_id
+ if table:
+ config["table"] = table
+ asset.extra[DQ_EXTRA_KEY] = config
+ return asset
+
+
+def get_asset_quality_config(asset: Asset) -> dict[str, Any] | None:
+ """Return the raw ``airflow.dataquality`` config attached to an asset, if any."""
+ config = asset.extra.get(DQ_EXTRA_KEY)
+ if not isinstance(config, dict):
+ return None
+ return config
+
+
+def get_asset_ruleset(asset: Asset) -> RuleSet:
+ """Return the ruleset attached to an asset, raising when none is attached."""
+ config = get_asset_quality_config(asset)
+ if not config or "ruleset" not in config:
+ raise DQRuleValidationError(
+ f"Asset {asset.name!r} has no data quality config; attach one with asset_quality()"
+ )
+ return RuleSet.from_dict(config["ruleset"])
+
+
+def _quality_score_passes(
+ asset: Asset,
+ min_score: float,
+ triggering_asset_events: Mapping[Asset, Sequence[AssetEventDagRunReferenceResult]],
+) -> bool:
+ """Pure decision logic behind :func:`require_quality`, kept separate so it is testable without a Dag."""
+ events = triggering_asset_events.get(asset, [])
+ if not events:
+ log.warning(
+ "require_quality(%s): no triggering event for this run; skipping downstream tasks",
+ asset.name,
+ )
+ return False
+
+ summary = events[-1].extra.get(DQ_RESULT_EXTRA_KEY)
Review Comment:
`consumed_asset_events` is a plain M2M backref with no `order_by`, and the execution API loads it with a bare `joinedload`, so the list order here is whatever the database returns -- `events[-1]` isn't guaranteed to be the newest event. And when several producer runs coalesce into one consumer run, gating only on one event means a low-scoring event slips through if a higher-scoring one sits after it in the list. I'd pick the event explicitly by `timestamp`, and consider whether the safer default is requiring every triggering event's score to pass, with latest-only as an opt-in.
##########
providers/common/dataquality/src/airflow/providers/common/dataquality/skills/dataquality-rule-authoring/SKILL.md:
##########
@@ -0,0 +1,173 @@
+---
+name: dataquality-rule-authoring
+description: "Generate a RuleSet/DQRule JSON payload for Apache Airflow's dataquality provider from a table's column definitions. Use this whenever asked to write, generate, or suggest data quality rules, checks, or a ruleset for a table or dataset -- especially when the output is structured JSON handed to DQCheckOperator, @task.dq_check, asset_quality(), or RuleSet.from_file()."
+---
+
+<!-- SPDX-License-Identifier: Apache-2.0
+ https://www.apache.org/licenses/LICENSE-2.0 -->
+
+# dataquality rule authoring
+
+You are producing a **RuleSet**: a JSON object naming a table's data quality checks. This
+document is the schema and the full list of valid values -- do not guess field names or check
+names, and do not invent checks that aren't in the catalog below.
+
+## Output shape
+
+```json
+{
+ "name": "orders_quality",
+ "rules": [
+ {
+ "name": "order_id_not_null",
+ "check": "null_count",
+ "column": "order_id",
+ "condition": {"equal_to": 0}
+ }
+ ]
+}
+```
+
+`name` (ruleset name) and `rules` (array) are the only top-level keys. Every rule name must be
+unique within the ruleset.
+
+## `DQRule` fields
+
+| Field | Required | Notes |
+|---|---|---|
+| `name` | yes | Unique within the ruleset. Shows up in data quality results and logs. |
+| `check` | yes | One of the catalog names below, or `"custom_sql"`. Exact string match -- there is no fuzzy matching. |
+| `condition` | yes* | See `Condition` grammar below. *Every catalog check currently requires one explicitly (no defaults) -- always include it. |
+| `column` | check-dependent | Required for every catalog check except `row_count`. Not used with `custom_sql`. |
+| `sql` | `custom_sql` only | A SQL statement returning a single scalar. May reference the checked table as `{table}`. Invalid together with any catalog check. |
+| `severity` | no | `"error"` (default) or `"warn"`. `"error"` fails the task on a failing rule (subject to the operator's `fail_on`); `"warn"` only records it. |
+| `partition_clause` | no | Extra SQL predicate ANDed into this rule's `WHERE` clause, e.g. `"region = 'EU'"`. |
+| `previous_name` | no | Only set when told a rule is being renamed, to keep its history continuous. |
+| `description` | no | Human-readable text shown in data quality results. Use it when a clear business meaning is known; otherwise omit it and let Airflow generate a default description. |
+| `dimension` | no | One of `completeness`, `uniqueness`, `validity`, `freshness`, `volume`, `consistency`. Defaults to the check's catalog dimension (`validity` for `custom_sql`) -- only set this explicitly when a `custom_sql` rule measures something the default doesn't capture (e.g. a freshness check written as `custom_sql` should set `"dimension": "freshness"`). Leave it unset for every built-in check. |
+
+**Do not add any key not listed above.** The schema rejects unrecognized fields.
+
+## Built-in check catalog
+
+| `check` | SQL expression | needs `column` | typical use |
+|---|---|---|---|
+| `null_count` | `SUM(CASE WHEN {column} IS NULL THEN 1 ELSE 0 END)` | yes | count of nulls in a column |
+| `null_ratio` | `SUM(CASE WHEN {column} IS NULL THEN 1.0 ELSE 0.0 END) / COUNT(*)` | yes | fraction of nulls, 0.0-1.0 |
+| `distinct_count` | `COUNT(DISTINCT {column})` | yes | number of distinct values |
+| `unique_violations` | `COUNT({column}) - COUNT(DISTINCT {column})` | yes | 0 means the column is fully unique |
+| `min` | `MIN({column})` | yes | minimum value |
+| `max` | `MAX({column})` | yes | maximum value |
+| `mean` | `AVG({column})` | yes | average value |
+| `row_count` | `COUNT(*)` | no | total row count -- omit `column` entirely |
+
+These are the *only* built-in check names. Anything else -- comparing two columns, joining
+another table, checking a format/regex, freshness against `now()`, referential integrity -- must
+be written as `custom_sql`.
+
+## `Condition` grammar
+
+`condition` is an object with these optional numeric keys; at least one is required:
+
+- `equal_to` -- exact match. Cannot be combined with any other key below.
+- `greater_than` / `geq_to` -- lower bound, exclusive / inclusive.
+- `less_than` / `leq_to` -- upper bound, exclusive / inclusive.
+- `tolerance` -- a fraction (e.g. `0.1` for 10%) that widens the comparison bounds.
+
+Combine `greater_than`/`less_than`/`geq_to`/`leq_to` freely to express a range, e.g.
+`{"geq_to": 0, "leq_to": 100}`. Never combine `equal_to` with another comparison key (other than
+`tolerance`).
+
+## `custom_sql`: when a catalog check doesn't fit
Review Comment:
The skill steers models toward emitting `custom_sql`, and the engine executes that string verbatim through the connection -- a generated `DELETE FROM {table} RETURNING 1` satisfies the scalar contract while mutating data, so a model error or prompt injection inherits the connection's write privileges. For Dag-author-written rules this matches the common.sql trust model, but the skill is specifically the path where the author isn't writing the SQL. I'd have the skill instruct models to prefer built-in checks and treat `custom_sql` as requiring human review, and recommend read-only credentials for DQ connections in the docs.
##########
providers/common/dataquality/src/airflow/providers/common/dataquality/rules/rule.py:
##########
@@ -0,0 +1,289 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+"""Declarative data quality rule model: rules are data, not code."""
+
+from __future__ import annotations
+
+import hashlib
+import json
+from enum import Enum
+from typing import Any, cast
+
+import yaml
+from pydantic import BaseModel, ConfigDict, Field, model_validator
+
+from airflow.providers.common.dataquality.exceptions import DQRuleValidationError
+from airflow.providers.common.dataquality.rules.checks import CHECK_SPECS, Dimension
+
+CUSTOM_SQL_CHECK = "custom_sql"
+
+
+class Severity(str, Enum):
+ """Severity used to decide whether a failing rule fails the task."""
+
+ WARN = "warn"
+ ERROR = "error"
+
+
+class Condition(BaseModel):
+ """
+ Pass/fail condition evaluated against a rule's observed value.
+
+ Uses the same grammar as the ``common.sql`` check operators: ``equal_to``,
+ ``greater_than``, ``less_than``, ``geq_to``, ``leq_to``, plus a percentage
+ ``tolerance`` that widens comparisons.
+
+ """
+
+ model_config = ConfigDict(frozen=True, extra="forbid")
+
+ equal_to: float | None = None
+ greater_than: float | None = None
+ less_than: float | None = None
+ geq_to: float | None = None
+ leq_to: float | None = None
+ tolerance: float | None = None
+
+ @model_validator(mode="after")
+ def _validate_comparisons(self) -> Condition:
+ comparisons = {
+ "equal_to": self.equal_to,
+ "greater_than": self.greater_than,
+ "less_than": self.less_than,
+ "geq_to": self.geq_to,
+ "leq_to": self.leq_to,
+ }
+ set_comparisons = {name for name, value in comparisons.items() if value is not None}
+ if not set_comparisons:
+ raise ValueError(f"Condition needs at least one comparison out of: {', '.join(comparisons)}")
+ if self.equal_to is not None and len(set_comparisons) > 1:
+ raise ValueError("equal_to cannot be combined with other comparisons")
+ return self
+
+ @classmethod
+ def from_dict(cls, data: dict[str, Any]) -> Condition:
+ return cls(**data)
+
+ def to_dict(self) -> dict[str, float]:
+ data = {
+ "equal_to": self.equal_to,
+ "greater_than": self.greater_than,
+ "less_than": self.less_than,
+ "geq_to": self.geq_to,
+ "leq_to": self.leq_to,
+ "tolerance": self.tolerance,
+ }
+ return {k: v for k, v in data.items() if v is not None}
+
+ def evaluate(self, observed: Any) -> bool:
+ if observed is None:
+ return False
+ value = float(observed)
+ if self.equal_to is not None:
+ if self.tolerance is not None:
+ delta = abs(self.equal_to) * self.tolerance
+ return self.equal_to - delta <= value <= self.equal_to + delta
+ return value == self.equal_to
+ geq_to = self.geq_to
+ greater_than = self.greater_than
+ leq_to = self.leq_to
+ less_than = self.less_than
+ if self.tolerance is not None:
+ if geq_to is not None:
+ geq_to -= abs(geq_to) * self.tolerance
+ if greater_than is not None:
+ greater_than -= abs(greater_than) * self.tolerance
+ if leq_to is not None:
+ leq_to += abs(leq_to) * self.tolerance
+ if less_than is not None:
+ less_than += abs(less_than) * self.tolerance
+ return (
+ (greater_than is None or value > greater_than)
+ and (less_than is None or value < less_than)
+ and (geq_to is None or value >= geq_to)
+ and (leq_to is None or value <= leq_to)
+ )
+
+
+class DQRule(BaseModel):
+ """
+ A single, named data quality rule.
+
+ :param name: Rule name, unique within its ruleset.
+ :param check: One of the built-in checks (see ``CHECK_SPECS``) or ``custom_sql``. Built-in
+ checks are plain ANSI SQL and are not guaranteed to work against every ``DbApiHook``;
+ see "Supported checks and databases" in the provider docs. Use ``custom_sql`` if a
+ built-in check doesn't fit your database's dialect.
+ :param condition: Pass condition for the observed value (``Condition`` or its dict form).
+ Optional only for checks whose ``CheckSpec.default_condition`` is set; every current
+ built-in check requires one explicitly.
+ :param column: Target column; required for column-level built-in checks.
+ :param sql: SQL statement returning a single scalar; required for ``custom_sql``.
+ May reference the target table as ``{table}``.
+ :param severity: ``error`` (fails the task by default) or ``warn`` (recorded only).
+ :param partition_clause: Extra predicate ANDed into the check's WHERE clause.
+ :param previous_name: Set when renaming a rule, to keep its history continuous.
+ :param description: Human-readable description shown in results and the UI. When omitted,
+ the provider generates a short default description from the rule and condition.
+
+ Invalid input raises pydantic's own :class:`~pydantic.ValidationError`.
+ """
+
+ model_config = ConfigDict(frozen=True, extra="forbid")
+
+ name: str
+ check: str = Field(description="One of the built-in checks, or custom_sql.")
+ condition: Condition | dict[str, Any] | None = None
+ column: str | None = None
+ sql: str | None = None
+ severity: Severity = Severity.ERROR
+ partition_clause: str | None = None
+ previous_name: str | None = None
+ description: str | None = Field(
+ default=None,
+ description=(
+ "Human-readable description shown in results and the UI. When omitted, the provider "
+ "generates a short default description from the rule and condition."
+ ),
+ )
+ dimension: Dimension | None = Field(
+ default=None,
+ description=(
+ "Data quality dimension recorded for this rule's results. Defaults to the check's "
+ "catalog dimension (validity for custom_sql) when not set explicitly."
+ ),
+ )
+
+ @model_validator(mode="after")
+ def _validate(self) -> DQRule:
+ if not self.name:
+ raise ValueError("Rule name cannot be empty")
+ catalog_spec = None
+ if self.check == CUSTOM_SQL_CHECK:
+ if not self.sql:
+ raise ValueError(f"Rule {self.name!r}: custom_sql check requires 'sql'")
+ else:
+ catalog_spec = CHECK_SPECS.get(self.check)
+ if catalog_spec is None:
+ supported = sorted([*CHECK_SPECS, CUSTOM_SQL_CHECK])
+ raise ValueError(f"Rule {self.name!r}: unknown check {self.check!r}; supported: {supported}")
+ if self.sql:
+ raise ValueError(f"Rule {self.name!r}: 'sql' is only valid with custom_sql")
+ if catalog_spec.requires_column and not self.column:
+ raise ValueError(f"Rule {self.name!r}: check {self.check!r} requires 'column'")
+ if self.condition is None and catalog_spec.default_condition is not None:
+ object.__setattr__(self, "condition", Condition.from_dict(catalog_spec.default_condition))
+ if self.condition is None:
+ raise ValueError(f"Rule {self.name!r}: condition is required for check {self.check!r}")
+ if self.dimension is None:
+ object.__setattr__(
+ self, "dimension", catalog_spec.dimension if catalog_spec else Dimension.VALIDITY
+ )
+ return self
+
+ @property
+ def rule_uid(self) -> str:
+ """Stable identity across runs: survives severity/dimension tweaks and Dag refactors."""
+ condition = cast("Condition", self.condition)
+ identity = {
+ "name": self.previous_name or self.name,
+ "check": self.check,
+ "column": self.column,
+ "sql": self.sql,
+ "condition": condition.to_dict(),
+ }
+ digest = hashlib.sha256(json.dumps(identity, sort_keys=True).encode()).hexdigest()
Review Comment:
The `previous_name or name` fallback can collide: a rule whose `previous_name` matches another rule's name (with the same check/column/condition) hashes to the same uid, and since RuleSet only rejects duplicate *names*, both pass validation -- then `observed_by_uid` keeps only one UNION ALL row and the two rules overwrite each other's files under `rule_uid=<uid>/`. I'd also question hashing the condition itself. The docstring promises identity that survives severity/dimension tweaks, but tightening a threshold is the edit you actually make while watching a rule's history, and it orphans everything recorded so far. The rename mechanism has a similar wrinkle: continuity only holds while `previous_name` stays in the Dag forever, so deleting it once the rename settles flips the uid back to a fresh history. An explicit optional `id` field would sidestep all of this; short of that, hashing the name-chain/check/column/sql without the condition plus a duplicate-uid check in RuleSet would
close the collision.
##########
providers/common/dataquality/src/airflow/providers/common/dataquality/version_compat.py:
##########
@@ -0,0 +1,37 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+#
+# NOTE! THIS FILE IS COPIED MANUALLY IN OTHER PROVIDERS DELIBERATELY TO AVOID ADDING UNNECESSARY
+# DEPENDENCIES BETWEEN PROVIDERS. IF YOU WANT TO ADD CONDITIONAL CODE IN YOUR PROVIDER THAT DEPENDS
+# ON AIRFLOW VERSION, PLEASE COPY THIS FILE TO THE ROOT PACKAGE OF YOUR PROVIDER AND IMPORT
+# THOSE CONSTANTS FROM IT RATHER THAN IMPORTING THEM FROM ANOTHER PROVIDER OR TEST CODE
+#
+from __future__ import annotations
+
+
+def get_base_airflow_version_tuple() -> tuple[int, int, int]:
+ from packaging.version import Version
+
+ from airflow import __version__
+
+ airflow_version = Version(__version__)
+ return airflow_version.major, airflow_version.minor, airflow_version.micro
+
+
+AIRFLOW_V_3_1_PLUS: bool = get_base_airflow_version_tuple() >= (3, 1, 0)
Review Comment:
`AIRFLOW_V_3_1_PLUS` isn't imported anywhere in the provider. Everything used here exists at the 3.0 floor, so it can just be dropped.
##########
providers/common/dataquality/src/airflow/providers/common/dataquality/assets.py:
##########
@@ -0,0 +1,172 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+"""
+Asset-level data quality declarations.
+
+Quality configuration lives inside ``Asset.extra`` under the ``airflow.dataquality`` key, so it is
+serialized with the Dag and needs no Airflow core changes. The rules travel with the asset
+definition instead of being scattered across the Dags that check it.
+"""
+
+from __future__ import annotations
+
+import logging
+from typing import TYPE_CHECKING, Any
+
+from airflow.providers.common.dataquality.exceptions import DQRuleValidationError
+from airflow.providers.common.dataquality.rules import RuleSet
+from airflow.sdk import task
+
+if TYPE_CHECKING:
+ from collections.abc import Mapping, Sequence
+
+ from airflow.sdk import Asset
+ from airflow.sdk.execution_time.comms import AssetEventDagRunReferenceResult
+
+log = logging.getLogger(__name__)
+
+DQ_EXTRA_KEY = "airflow.dataquality"
+DQ_RESULT_EXTRA_KEY = "airflow.dataquality.result"
+
+
+def asset_quality(
+ asset: Asset,
+ *,
+ ruleset: RuleSet | dict[str, Any] | str,
+ conn_id: str | None = None,
+ table: str | None = None,
+) -> Asset:
+ """
+ Attach data quality configuration to an asset, returning the same asset.
+
+ :param asset: The asset the rules describe.
+ :param ruleset: A :class:`~airflow.providers.common.dataquality.rules.RuleSet`, its dict form, or a path
+ to a YAML ruleset file (resolved eagerly, at Dag-parse time).
+ :param conn_id: Default connection a check operator should use for this asset.
+ :param table: Default table to check; falls back to the asset name when unset.
+
+ Usage::
+
+ orders = asset_quality(
+ Asset("orders", uri="postgres://warehouse/analytics/orders"),
+ ruleset=rules,
+ conn_id="warehouse",
+ table="analytics.orders",
+ )
+ check = DQCheckOperator(task_id="dq", asset=orders)
+ """
+ if isinstance(ruleset, str):
+ ruleset = RuleSet.from_file(ruleset)
+ elif isinstance(ruleset, dict):
+ ruleset = RuleSet.from_dict(ruleset)
+ config: dict[str, Any] = {"ruleset": ruleset.to_dict()}
+ if conn_id:
+ config["conn_id"] = conn_id
+ if table:
+ config["table"] = table
+ asset.extra[DQ_EXTRA_KEY] = config
Review Comment:
Design question, non-blocking: core dedupes assets by (name, uri) and `sync_assets` overwrites `model.extra` on every parse, last writer wins -- so a second Dag referencing `Asset("orders")` without quality config wipes the registry copy of the ruleset, and two Dags with different rulesets flip-flop. This slice is unaffected (the operator reads the in-memory object, the gate reads event extra), but the #69413 UI presumably reads the registry copy. How is the attached config expected to survive there?
##########
providers/common/dataquality/src/airflow/providers/common/dataquality/backends/__init__.py:
##########
@@ -0,0 +1,32 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+from __future__ import annotations
+
+from airflow.providers.common.dataquality.backends.object_storage import ObjectStorageResultsBackend
Review Comment:
Importing the operator pulls this in eagerly, and `ObjectStoragePath` drags fsspec + upath at import -- paid at Dag-parse time in every file that imports `DQCheckOperator`, even with no `results_path` configured. Since the backend is only built inside `get_backend_from_config()` at execute time, importing `ObjectStorageResultsBackend` lazily there (with a comment) keeps the parse path light.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Airflow Data Quality Provider part 1 [airflow]
Posted by "gopidesupavan (via GitHub)" <gi...@apache.org>.
gopidesupavan commented on code in PR #69575:
URL: https://github.com/apache/airflow/pull/69575#discussion_r3575179534
##########
providers/common/dataquality/src/airflow/providers/common/dataquality/rules/rule.py:
##########
@@ -0,0 +1,289 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+"""Declarative data quality rule model: rules are data, not code."""
+
+from __future__ import annotations
+
+import hashlib
+import json
+from enum import Enum
+from typing import Any, cast
+
+import yaml
+from pydantic import BaseModel, ConfigDict, Field, model_validator
+
+from airflow.providers.common.dataquality.exceptions import DQRuleValidationError
+from airflow.providers.common.dataquality.rules.checks import CHECK_SPECS, Dimension
+
+CUSTOM_SQL_CHECK = "custom_sql"
+
+
+class Severity(str, Enum):
+ """Severity used to decide whether a failing rule fails the task."""
+
+ WARN = "warn"
+ ERROR = "error"
+
+
+class Condition(BaseModel):
+ """
+ Pass/fail condition evaluated against a rule's observed value.
+
+ Uses the same grammar as the ``common.sql`` check operators: ``equal_to``,
+ ``greater_than``, ``less_than``, ``geq_to``, ``leq_to``, plus a percentage
+ ``tolerance`` that widens comparisons.
+
+ """
+
+ model_config = ConfigDict(frozen=True, extra="forbid")
+
+ equal_to: float | None = None
+ greater_than: float | None = None
+ less_than: float | None = None
+ geq_to: float | None = None
+ leq_to: float | None = None
+ tolerance: float | None = None
+
+ @model_validator(mode="after")
+ def _validate_comparisons(self) -> Condition:
+ comparisons = {
+ "equal_to": self.equal_to,
+ "greater_than": self.greater_than,
+ "less_than": self.less_than,
+ "geq_to": self.geq_to,
+ "leq_to": self.leq_to,
+ }
+ set_comparisons = {name for name, value in comparisons.items() if value is not None}
+ if not set_comparisons:
+ raise ValueError(f"Condition needs at least one comparison out of: {', '.join(comparisons)}")
+ if self.equal_to is not None and len(set_comparisons) > 1:
+ raise ValueError("equal_to cannot be combined with other comparisons")
+ return self
+
+ @classmethod
+ def from_dict(cls, data: dict[str, Any]) -> Condition:
+ return cls(**data)
+
+ def to_dict(self) -> dict[str, float]:
+ data = {
+ "equal_to": self.equal_to,
+ "greater_than": self.greater_than,
+ "less_than": self.less_than,
+ "geq_to": self.geq_to,
+ "leq_to": self.leq_to,
+ "tolerance": self.tolerance,
+ }
+ return {k: v for k, v in data.items() if v is not None}
+
+ def evaluate(self, observed: Any) -> bool:
+ if observed is None:
+ return False
+ value = float(observed)
+ if self.equal_to is not None:
+ if self.tolerance is not None:
+ delta = abs(self.equal_to) * self.tolerance
+ return self.equal_to - delta <= value <= self.equal_to + delta
+ return value == self.equal_to
+ geq_to = self.geq_to
+ greater_than = self.greater_than
+ leq_to = self.leq_to
+ less_than = self.less_than
+ if self.tolerance is not None:
+ if geq_to is not None:
+ geq_to -= abs(geq_to) * self.tolerance
+ if greater_than is not None:
+ greater_than -= abs(greater_than) * self.tolerance
+ if leq_to is not None:
+ leq_to += abs(leq_to) * self.tolerance
+ if less_than is not None:
+ less_than += abs(less_than) * self.tolerance
+ return (
+ (greater_than is None or value > greater_than)
+ and (less_than is None or value < less_than)
+ and (geq_to is None or value >= geq_to)
+ and (leq_to is None or value <= leq_to)
+ )
+
+
+class DQRule(BaseModel):
+ """
+ A single, named data quality rule.
+
+ :param name: Rule name, unique within its ruleset.
+ :param check: One of the built-in checks (see ``CHECK_SPECS``) or ``custom_sql``. Built-in
+ checks are plain ANSI SQL and are not guaranteed to work against every ``DbApiHook``;
+ see "Supported checks and databases" in the provider docs. Use ``custom_sql`` if a
+ built-in check doesn't fit your database's dialect.
+ :param condition: Pass condition for the observed value (``Condition`` or its dict form).
+ Optional only for checks whose ``CheckSpec.default_condition`` is set; every current
+ built-in check requires one explicitly.
+ :param column: Target column; required for column-level built-in checks.
+ :param sql: SQL statement returning a single scalar; required for ``custom_sql``.
+ May reference the target table as ``{table}``.
+ :param severity: ``error`` (fails the task by default) or ``warn`` (recorded only).
+ :param partition_clause: Extra predicate ANDed into the check's WHERE clause.
+ :param previous_name: Set when renaming a rule, to keep its history continuous.
+ :param description: Human-readable description shown in results and the UI. When omitted,
+ the provider generates a short default description from the rule and condition.
+
+ Invalid input raises pydantic's own :class:`~pydantic.ValidationError`.
+ """
+
+ model_config = ConfigDict(frozen=True, extra="forbid")
+
+ name: str
+ check: str = Field(description="One of the built-in checks, or custom_sql.")
+ condition: Condition | dict[str, Any] | None = None
+ column: str | None = None
+ sql: str | None = None
+ severity: Severity = Severity.ERROR
+ partition_clause: str | None = None
+ previous_name: str | None = None
+ description: str | None = Field(
+ default=None,
+ description=(
+ "Human-readable description shown in results and the UI. When omitted, the provider "
+ "generates a short default description from the rule and condition."
+ ),
+ )
+ dimension: Dimension | None = Field(
+ default=None,
+ description=(
+ "Data quality dimension recorded for this rule's results. Defaults to the check's "
+ "catalog dimension (validity for custom_sql) when not set explicitly."
+ ),
+ )
+
+ @model_validator(mode="after")
+ def _validate(self) -> DQRule:
+ if not self.name:
+ raise ValueError("Rule name cannot be empty")
+ catalog_spec = None
+ if self.check == CUSTOM_SQL_CHECK:
+ if not self.sql:
+ raise ValueError(f"Rule {self.name!r}: custom_sql check requires 'sql'")
+ else:
+ catalog_spec = CHECK_SPECS.get(self.check)
+ if catalog_spec is None:
+ supported = sorted([*CHECK_SPECS, CUSTOM_SQL_CHECK])
+ raise ValueError(f"Rule {self.name!r}: unknown check {self.check!r}; supported: {supported}")
+ if self.sql:
+ raise ValueError(f"Rule {self.name!r}: 'sql' is only valid with custom_sql")
+ if catalog_spec.requires_column and not self.column:
+ raise ValueError(f"Rule {self.name!r}: check {self.check!r} requires 'column'")
+ if self.condition is None and catalog_spec.default_condition is not None:
+ object.__setattr__(self, "condition", Condition.from_dict(catalog_spec.default_condition))
+ if self.condition is None:
+ raise ValueError(f"Rule {self.name!r}: condition is required for check {self.check!r}")
+ if self.dimension is None:
+ object.__setattr__(
+ self, "dimension", catalog_spec.dimension if catalog_spec else Dimension.VALIDITY
+ )
+ return self
+
+ @property
+ def rule_uid(self) -> str:
+ """Stable identity across runs: survives severity/dimension tweaks and Dag refactors."""
+ condition = cast("Condition", self.condition)
+ identity = {
+ "name": self.previous_name or self.name,
+ "check": self.check,
+ "column": self.column,
+ "sql": self.sql,
+ "condition": condition.to_dict(),
+ }
+ digest = hashlib.sha256(json.dumps(identity, sort_keys=True).encode()).hexdigest()
Review Comment:
yeah this one was same i got confused initally but good point you raised, now have included optional id
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Airflow Data Quality Provider part 1 [airflow]
Posted by "SameerMesiah97 (via GitHub)" <gi...@apache.org>.
SameerMesiah97 commented on code in PR #69575:
URL: https://github.com/apache/airflow/pull/69575#discussion_r3605749388
##########
providers/common/dataquality/tests/unit/common/dataquality/test_exceptions.py:
##########
@@ -0,0 +1,27 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+from __future__ import annotations
+
+from airflow.providers.common.dataquality.exceptions import DQCheckFailedError, DQRuleValidationError
+
+
+def test_dq_rule_validation_error_is_value_error():
+ assert isinstance(DQRuleValidationError("invalid rule"), ValueError)
+
+
+def test_dq_check_failed_error_is_runtime_error():
+ assert isinstance(DQCheckFailedError("failed check"), RuntimeError)
Review Comment:
Are these tests needed? It seems like you are just testing inheritance.
##########
providers/common/dataquality/src/airflow/providers/common/dataquality/operators/dq_check.py:
##########
@@ -0,0 +1,158 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+from __future__ import annotations
+
+import logging
+from collections.abc import Sequence
+from typing import TYPE_CHECKING, Any
+
+from airflow.providers.common.dataquality.assets import get_asset_quality_config
+from airflow.providers.common.dataquality.exceptions import DQCheckFailedError
+from airflow.providers.common.dataquality.execution import (
+ DataQualityResult,
+ _resolve_ruleset,
+ persist_quality_results,
+ run_quality_checks,
+)
+from airflow.providers.common.dataquality.results import (
+ ERROR,
+ FAIL,
+ WARN,
+ RuleResult,
+)
+from airflow.providers.common.sql.operators.sql import BaseSQLOperator
+
+if TYPE_CHECKING:
+ from airflow.providers.common.dataquality.rules import RuleSet
+ from airflow.sdk import Asset, Context
+
+log = logging.getLogger(__name__)
+
+FAIL_ON_CHOICES = ("error", "warn", "never")
+
+
+class DQCheckOperator(BaseSQLOperator):
+ """
+ Run a ruleset against a table and persist per-rule results to the configured results store.
+
+ Every rule is evaluated and recorded regardless of the task outcome. Execution errors
+ (a check query failing) always fail the task; rule failures fail the task according to
+ ``fail_on``.
+
+ :param ruleset: A :class:`~airflow.providers.common.dataquality.rules.RuleSet`, its dict form, or a path
+ to a YAML ruleset file. Optional when ``asset`` carries one.
+ :param table: The table to run checks against. Optional when ``asset`` carries one
+ (falling back to the asset name).
+ :param asset: An asset decorated with :func:`~airflow.providers.common.dataquality.assets.asset_quality`.
+ Supplies defaults for ``ruleset``, ``table``, and ``conn_id`` (explicit arguments
+ win) and is automatically added to the task's outlets so its asset events carry
+ the check summary.
Review Comment:
Now, this is a more general concern about the API of this operator: it is a bit hard to understand with the inter-relationships between the different parameters. For example, `ruleset`, and `table`` are all optional, but become required unless asset supplies them. I think we should construct the API in a way that removes all these interdependencies so that our users would find the operator more intuitive.
Maybe we should simplify the public API by making asset the sole source of configuration when it is provided, rather than allowing `table`, `ruleset`, and `conn_id` to override it. At the moment the operator supports multiple overlapping configuration mechanisms, which introduces precedence rules and conditional required parameters.
##########
providers/common/dataquality/src/airflow/providers/common/dataquality/operators/dq_check.py:
##########
@@ -0,0 +1,158 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+from __future__ import annotations
+
+import logging
+from collections.abc import Sequence
+from typing import TYPE_CHECKING, Any
+
+from airflow.providers.common.dataquality.assets import get_asset_quality_config
+from airflow.providers.common.dataquality.exceptions import DQCheckFailedError
+from airflow.providers.common.dataquality.execution import (
+ DataQualityResult,
+ _resolve_ruleset,
+ persist_quality_results,
+ run_quality_checks,
+)
+from airflow.providers.common.dataquality.results import (
+ ERROR,
+ FAIL,
+ WARN,
+ RuleResult,
+)
+from airflow.providers.common.sql.operators.sql import BaseSQLOperator
+
+if TYPE_CHECKING:
+ from airflow.providers.common.dataquality.rules import RuleSet
+ from airflow.sdk import Asset, Context
+
+log = logging.getLogger(__name__)
+
+FAIL_ON_CHOICES = ("error", "warn", "never")
+
+
+class DQCheckOperator(BaseSQLOperator):
+ """
+ Run a ruleset against a table and persist per-rule results to the configured results store.
+
+ Every rule is evaluated and recorded regardless of the task outcome. Execution errors
+ (a check query failing) always fail the task; rule failures fail the task according to
+ ``fail_on``.
+
+ :param ruleset: A :class:`~airflow.providers.common.dataquality.rules.RuleSet`, its dict form, or a path
+ to a YAML ruleset file. Optional when ``asset`` carries one.
+ :param table: The table to run checks against. Optional when ``asset`` carries one
+ (falling back to the asset name).
+ :param asset: An asset decorated with :func:`~airflow.providers.common.dataquality.assets.asset_quality`.
+ Supplies defaults for ``ruleset``, ``table``, and ``conn_id`` (explicit arguments
+ win) and is automatically added to the task's outlets so its asset events carry
+ the check summary.
+ :param partition_clause: Predicate ANDed into every built-in check's WHERE clause,
+ e.g. ``"ds = '{{ ds }}'"``.
+ :param fail_on: Severity that fails the task: ``error`` (default — warn-severity rule
+ failures are recorded but don't fail), ``warn`` (any rule failure fails), or
+ ``never`` (only execution errors fail).
+ :param conn_id: Connection to the database to check (any ``DbApiHook``-compatible type).
+ :param database: Optional database/schema overriding the connection's default.
+
+ Results are persisted to the backend configured under ``[common.dataquality] results_path``; when that's
+ unset, checks still run but no history is persisted. There is no per-operator override --
+ all checks in a deployment share one results store.
+ """
+
+ template_fields: Sequence[str] = ("table", "partition_clause", *BaseSQLOperator.template_fields)
+ ui_color: str = "#87ceeb"
+
+ def __init__(
+ self,
+ *,
+ ruleset: RuleSet | dict[str, Any] | str | None = None,
+ table: str | None = None,
+ asset: Asset | None = None,
+ partition_clause: str | None = None,
+ fail_on: str = "error",
+ **kwargs,
+ ) -> None:
+ if asset is not None:
+ config = get_asset_quality_config(asset) or {}
+ ruleset = ruleset if ruleset is not None else config.get("ruleset")
+ table = table or config.get("table") or asset.name
+ asset_conn_id = config.get("conn_id")
+ if asset_conn_id is not None:
+ kwargs.setdefault("conn_id", asset_conn_id)
+ outlets = list(kwargs.get("outlets") or [])
+ if asset not in outlets:
+ outlets.append(asset)
+ kwargs["outlets"] = outlets
+ super().__init__(**kwargs)
+ if fail_on not in FAIL_ON_CHOICES:
+ raise ValueError(f"fail_on must be one of {FAIL_ON_CHOICES}, got {fail_on!r}")
+ if ruleset is None:
+ raise ValueError("ruleset is required, either directly or via an asset_quality() asset")
+ if not table:
+ raise ValueError("table is required, either directly or via an asset_quality() asset")
+ if not self.conn_id:
+ raise ValueError("conn_id is required, either directly or via an asset_quality() asset")
+ self.ruleset = ruleset
+ self.table = table
+ self.asset = asset
+ self.partition_clause = partition_clause
+ self.fail_on = fail_on
+
+ def execute(self, context: Context) -> dict[str, Any]:
+ result = run_quality_checks(
+ hook=self.get_db_hook(),
+ ruleset=_resolve_ruleset(self.ruleset),
+ table=self.table,
+ partition_clause=self.partition_clause,
+ )
+ results = list(result.results)
+ summary = self._persist_result(result, context)
+ self._log_results(results, summary)
+ self._raise_for_failures(results, summary)
+ return summary
+
+ def _persist_result(self, result: DataQualityResult, context: Context) -> dict[str, Any]:
+ return persist_quality_results(result, context=context, outlets=list(self.outlets or ()))
+
+ def _log_results(self, results: list[RuleResult], summary: dict[str, Any]) -> None:
+ for result in results:
+ self.log.info(
+ "Rule %s [%s/%s]: %s (observed=%s, condition=%s)",
+ result.rule_name,
+ result.dimension,
+ result.severity,
+ result.status.upper(),
+ result.observed_value,
+ result.condition,
+ )
+ self.log.info("Data quality score: %s", summary["score"])
+
+ def _raise_for_failures(self, results: list[RuleResult], summary: dict[str, Any]) -> None:
+ errored = [r.rule_name for r in results if r.status == ERROR]
+ if errored:
+ raise DQCheckFailedError(f"Check execution errored for rules: {errored}")
+ failed = [r.rule_name for r in results if r.status == FAIL]
+ warned = [r.rule_name for r in results if r.status == WARN]
+ if self.fail_on == "error" and failed:
+ raise DQCheckFailedError(f"Data quality rules failed: {failed} (score={summary['score']})")
+ if self.fail_on == "warn" and (failed or warned):
Review Comment:
I am just curious why you are using raw strings such as "error" and "warn" here instead of an Enum? I see you added an Enum class in this PR. Perhaps, you could add a new class like this:
```
class FailOn(str, Enum):
ERROR = "error"
WARN = "warn"
NEVER = "never"
```
Also, this more of a nit but I think we could handle each case more explicity like the below:
```
if self.fail_on is FailOn.ERROR:
...
elif self.fail_on is FailOn.WARN:
...
elif self.fail_on is FailOn.NEVER:
self.log.warning(...)
```
##########
providers/common/dataquality/docs/agents.rst:
##########
@@ -0,0 +1,55 @@
+ .. Licensed to the Apache Software Foundation (ASF) under one
+ or more contributor license agreements. See the NOTICE file
+ distributed with this work for additional information
+ regarding copyright ownership. The ASF licenses this file
+ to you under the Apache License, Version 2.0 (the
+ "License"); you may not use this file except in compliance
+ with the License. You may obtain a copy of the License at
+
+ .. http://www.apache.org/licenses/LICENSE-2.0
+
+ .. Unless required by applicable law or agreed to in writing,
+ software distributed under the License is distributed on an
+ "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ KIND, either express or implied. See the License for the
+ specific language governing permissions and limitations
+ under the License.
+
+.. _dq:agents:
+
+Generating rules with an LLM
+==============================
+
+Writing a :class:`~airflow.providers.common.dataquality.rules.RuleSet` by hand for every table doesn't scale.
+An LLM can propose one from a table's column definitions instead, given the check catalog as
+context.
+
+The ``dataquality-rule-authoring`` skill
+----------------------------------------
+
+This provider ships an `Agent Skill <https://agentskills.io>`__ at
+``airflow/providers/common/dataquality/skills/dataquality-rule-authoring/``: a ``SKILL.md`` documenting the
+``RuleSet``/``DQRule`` fields and check catalog, plus a generated JSON Schema
+(``references/ruleset.schema.json``) for validation.
+
+Point ``common.ai``'s :doc:`AgentSkillsToolset <apache-airflow-providers-common-ai:toolsets>` at
+it, and give the model ``output_type=RuleSet`` so pydantic-ai validates -- and self-corrects --
+its output before the task completes:
+
+.. exampleinclude:: /../src/airflow/providers/common/dataquality/example_dags/example_dq_llm_generated_ruleset.py
+ :language: python
+ :start-after: [START howto_task_dq_generate_ruleset_with_llm]
+ :end-before: [END howto_task_dq_generate_ruleset_with_llm]
+
+Requires ``apache-airflow-providers-common-ai[skills]`` and a configured ``llm_conn_id``.
+
+Wiring the result into a check
+---------------------------------
+
+``@task.dq_check`` can leave ``ruleset=`` unset and return the LLM task's result at execution
+time instead:
+
+.. exampleinclude:: /../src/airflow/providers/common/dataquality/example_dags/example_dq_llm_generated_ruleset.py
+ :language: python
+ :start-after: [START howto_decorator_dq_check_llm_runtime_ruleset]
+ :end-before: [END howto_decorator_dq_check_llm_runtime_ruleset]
Review Comment:
I understand that this may come across as a bit pedantic but I think the documentation here veers too far into marketing language when I believe they should be purely instructional. Statements like 'Writing a RuleSet by hand for every table doesn't scale' is of course very defensible but it is a bit too opinionated for provider documentation.
I think it would not hurt to use an LLM here with specific instructions to keep it factual. I also found some of the phrasing a little awkward from a native English perspective. Again, a well-prompted LLM should be very useful for this sort of thing.
This feedback applies to the rest of the docs too. I glanced at them and they had similar issues.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Airflow Data Quality Provider part 1 [airflow]
Posted by "gopidesupavan (via GitHub)" <gi...@apache.org>.
gopidesupavan commented on code in PR #69575:
URL: https://github.com/apache/airflow/pull/69575#discussion_r3602299483
##########
providers/common/dataquality/src/airflow/providers/common/dataquality/backends/object_storage.py:
##########
@@ -0,0 +1,256 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+"""
+Object-storage results backend.
+
+Each DQ check writes a keyed JSON document plus read indexes optimized for the UI:
+
+ runs/by_task/dag_id=<dag>/task_id=<task>/date=<2026-07-04>/<started_at>__<run_uid>.json
+ Canonical run record: ``{"run": ..., "results": [...], "summary": ...}``.
+
+ runs/by_task_instance/dag_id=<dag>/task_id=<task>/<safe_run_id>__<map_index>.json
+ Latest result for a task-instance page. Last write wins across retries.
+
+ rules/by_task_rule/dag_id=<dag>/task_id=<task>/rule_uid=<uid>/<started_at>__<run_uid>.json
+ One rule result plus run context: ``{"run": ..., "result": ...}``.
+
+The duplicate files are intentional read indexes: DQ tasks write once, while the UI reads
+many times. Keeping these indexes avoids scanning all task runs for common UI views.
+"""
+
+from __future__ import annotations
+
+import json
+import logging
+from datetime import datetime, timezone
+from typing import Any
+
+from airflow.providers.common.dataquality.results import DQRun, RuleResult, build_summary
+from airflow.sdk import ObjectStoragePath
+
+log = logging.getLogger(__name__)
+
+
+class ObjectStorageResultsBackend:
+ """Persist DQ results as JSON files via ``ObjectStoragePath``."""
+
+ def __init__(self, results_path: str, conn_id: str | None = None) -> None:
+ self.root = ObjectStoragePath(results_path, conn_id=conn_id)
+
+ def write_run(self, run: DQRun, results: list[RuleResult]) -> None:
+ timestamp = run.started_at or datetime.now(tz=timezone.utc).isoformat()
+ compact_ts = self._get_safe_key(timestamp)
+ payload = self._build_run_payload(run, results)
+
+ self._write_run_file(run, timestamp[:10], compact_ts, payload)
+ self._write_task_instance_index(run, payload)
+ self._write_rule_indexes(run, results, timestamp)
+
+ def read_task_rule_history(
+ self, dag_id: str, task_id: str, rule_uid: str, limit: int = 100, before: str | None = None
Review Comment:
make sense changed
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Airflow Data Quality Provider part 1 [airflow]
Posted by "gopidesupavan (via GitHub)" <gi...@apache.org>.
gopidesupavan commented on code in PR #69575:
URL: https://github.com/apache/airflow/pull/69575#discussion_r3602306724
##########
providers/common/dataquality/src/airflow/providers/common/dataquality/assets.py:
##########
@@ -0,0 +1,190 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+"""
+Asset-level data quality declarations.
+
+Quality configuration lives inside ``Asset.extra`` under the ``airflow.dataquality`` key, so it is
+serialized with the Dag and needs no Airflow core changes. The rules travel with the asset
+definition instead of being scattered across the Dags that check it.
+"""
+
+from __future__ import annotations
+
+import logging
+from typing import TYPE_CHECKING, Any
+
+from airflow.providers.common.dataquality.exceptions import DQRuleValidationError
+from airflow.providers.common.dataquality.rules import RuleSet
+from airflow.sdk import task
+
+if TYPE_CHECKING:
+ from collections.abc import Mapping, Sequence
+
+ from airflow.sdk import Asset
+ from airflow.sdk.execution_time.comms import AssetEventDagRunReferenceResult
+
+log = logging.getLogger(__name__)
+
+DQ_EXTRA_KEY = "airflow.dataquality"
+DQ_RESULT_EXTRA_KEY = "airflow.dataquality.result"
+
+
+def asset_quality(
+ asset: Asset,
Review Comment:
atleast for the first its Asset, have not thought about AssetRef, AssetAlias. but there no harm in supporting that but will have to check.. will add a item for future track
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Airflow Data Quality Provider part 1 [airflow]
Posted by "gopidesupavan (via GitHub)" <gi...@apache.org>.
gopidesupavan commented on code in PR #69575:
URL: https://github.com/apache/airflow/pull/69575#discussion_r3575173562
##########
providers/common/dataquality/src/airflow/providers/common/dataquality/assets.py:
##########
@@ -0,0 +1,172 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+"""
+Asset-level data quality declarations.
+
+Quality configuration lives inside ``Asset.extra`` under the ``airflow.dataquality`` key, so it is
+serialized with the Dag and needs no Airflow core changes. The rules travel with the asset
+definition instead of being scattered across the Dags that check it.
+"""
+
+from __future__ import annotations
+
+import logging
+from typing import TYPE_CHECKING, Any
+
+from airflow.providers.common.dataquality.exceptions import DQRuleValidationError
+from airflow.providers.common.dataquality.rules import RuleSet
+from airflow.sdk import task
+
+if TYPE_CHECKING:
+ from collections.abc import Mapping, Sequence
+
+ from airflow.sdk import Asset
+ from airflow.sdk.execution_time.comms import AssetEventDagRunReferenceResult
+
+log = logging.getLogger(__name__)
+
+DQ_EXTRA_KEY = "airflow.dataquality"
+DQ_RESULT_EXTRA_KEY = "airflow.dataquality.result"
+
+
+def asset_quality(
+ asset: Asset,
+ *,
+ ruleset: RuleSet | dict[str, Any] | str,
+ conn_id: str | None = None,
+ table: str | None = None,
+) -> Asset:
+ """
+ Attach data quality configuration to an asset, returning the same asset.
+
+ :param asset: The asset the rules describe.
+ :param ruleset: A :class:`~airflow.providers.common.dataquality.rules.RuleSet`, its dict form, or a path
+ to a YAML ruleset file (resolved eagerly, at Dag-parse time).
+ :param conn_id: Default connection a check operator should use for this asset.
+ :param table: Default table to check; falls back to the asset name when unset.
+
+ Usage::
+
+ orders = asset_quality(
+ Asset("orders", uri="postgres://warehouse/analytics/orders"),
+ ruleset=rules,
+ conn_id="warehouse",
+ table="analytics.orders",
+ )
+ check = DQCheckOperator(task_id="dq", asset=orders)
+ """
+ if isinstance(ruleset, str):
+ ruleset = RuleSet.from_file(ruleset)
+ elif isinstance(ruleset, dict):
+ ruleset = RuleSet.from_dict(ruleset)
+ config: dict[str, Any] = {"ruleset": ruleset.to_dict()}
+ if conn_id:
+ config["conn_id"] = conn_id
+ if table:
+ config["table"] = table
+ asset.extra[DQ_EXTRA_KEY] = config
Review Comment:
ah good point , The intended contract is one owner per asset: whichever Dag calls asset_quality() on a given (name, uri) should be the only one attaching config to it. this mirrors the existing last-writer-wins behavior on extra in general, just newly visible through quality config.
I will make that explicit in the docstring, I think a real merge strategy across multiple owners would be a bigger design change ? may be that can be future..
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Airflow Data Quality Provider part 1 [airflow]
Posted by "Lee-W (via GitHub)" <gi...@apache.org>.
Lee-W commented on code in PR #69575:
URL: https://github.com/apache/airflow/pull/69575#discussion_r3594529824
##########
providers/common/dataquality/docs/rules.rst:
##########
@@ -0,0 +1,239 @@
+ .. Licensed to the Apache Software Foundation (ASF) under one
+ or more contributor license agreements. See the NOTICE file
+ distributed with this work for additional information
+ regarding copyright ownership. The ASF licenses this file
+ to you under the Apache License, Version 2.0 (the
+ "License"); you may not use this file except in compliance
+ with the License. You may obtain a copy of the License at
+
+ .. http://www.apache.org/licenses/LICENSE-2.0
+
+ .. Unless required by applicable law or agreed to in writing,
+ software distributed under the License is distributed on an
+ "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ KIND, either express or implied. See the License for the
+ specific language governing permissions and limitations
+ under the License.
+
+.. _dq:rules:
+
+Rules, rule sets, and checks
+============================
+
+Rules are data, not code. A :class:`~airflow.providers.common.dataquality.rules.RuleSet` is a named tuple of
+:class:`~airflow.providers.common.dataquality.rules.DQRule` -- plain, serializable objects with no behavior of
+their own. They describe *what* to check; :class:`~airflow.providers.common.dataquality.operators.dq_check.DQCheckOperator`
+(see :doc:`operators`) is what actually runs them. Rules are usually written by hand, but they
+can also be proposed by an LLM -- see :doc:`agents`.
+
+.. code-block:: python
+
+ from airflow.providers.common.dataquality.rules import DQRule, RuleSet
Review Comment:
should we make it part of the example Dag instead?
##########
providers/common/dataquality/src/airflow/providers/common/dataquality/backends/object_storage.py:
##########
@@ -0,0 +1,256 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+"""
+Object-storage results backend.
+
+Each DQ check writes a keyed JSON document plus read indexes optimized for the UI:
+
+ runs/by_task/dag_id=<dag>/task_id=<task>/date=<2026-07-04>/<started_at>__<run_uid>.json
+ Canonical run record: ``{"run": ..., "results": [...], "summary": ...}``.
+
+ runs/by_task_instance/dag_id=<dag>/task_id=<task>/<safe_run_id>__<map_index>.json
+ Latest result for a task-instance page. Last write wins across retries.
+
+ rules/by_task_rule/dag_id=<dag>/task_id=<task>/rule_uid=<uid>/<started_at>__<run_uid>.json
+ One rule result plus run context: ``{"run": ..., "result": ...}``.
+
+The duplicate files are intentional read indexes: DQ tasks write once, while the UI reads
+many times. Keeping these indexes avoids scanning all task runs for common UI views.
+"""
+
+from __future__ import annotations
+
+import json
+import logging
+from datetime import datetime, timezone
+from typing import Any
+
+from airflow.providers.common.dataquality.results import DQRun, RuleResult, build_summary
+from airflow.sdk import ObjectStoragePath
+
+log = logging.getLogger(__name__)
+
+
+class ObjectStorageResultsBackend:
+ """Persist DQ results as JSON files via ``ObjectStoragePath``."""
+
+ def __init__(self, results_path: str, conn_id: str | None = None) -> None:
Review Comment:
```suggestion
def __init__(self, *, results_path: str, conn_id: str | None = None) -> None:
```
let's make it keyword only, so that it will be easier for us to add new args in the future
##########
providers/common/dataquality/src/airflow/providers/common/dataquality/backends/object_storage.py:
##########
@@ -0,0 +1,256 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+"""
+Object-storage results backend.
+
+Each DQ check writes a keyed JSON document plus read indexes optimized for the UI:
+
+ runs/by_task/dag_id=<dag>/task_id=<task>/date=<2026-07-04>/<started_at>__<run_uid>.json
+ Canonical run record: ``{"run": ..., "results": [...], "summary": ...}``.
+
+ runs/by_task_instance/dag_id=<dag>/task_id=<task>/<safe_run_id>__<map_index>.json
+ Latest result for a task-instance page. Last write wins across retries.
+
+ rules/by_task_rule/dag_id=<dag>/task_id=<task>/rule_uid=<uid>/<started_at>__<run_uid>.json
+ One rule result plus run context: ``{"run": ..., "result": ...}``.
+
+The duplicate files are intentional read indexes: DQ tasks write once, while the UI reads
+many times. Keeping these indexes avoids scanning all task runs for common UI views.
+"""
+
+from __future__ import annotations
+
+import json
+import logging
+from datetime import datetime, timezone
+from typing import Any
+
+from airflow.providers.common.dataquality.results import DQRun, RuleResult, build_summary
+from airflow.sdk import ObjectStoragePath
+
+log = logging.getLogger(__name__)
+
+
+class ObjectStorageResultsBackend:
+ """Persist DQ results as JSON files via ``ObjectStoragePath``."""
+
+ def __init__(self, results_path: str, conn_id: str | None = None) -> None:
+ self.root = ObjectStoragePath(results_path, conn_id=conn_id)
+
+ def write_run(self, run: DQRun, results: list[RuleResult]) -> None:
+ timestamp = run.started_at or datetime.now(tz=timezone.utc).isoformat()
+ compact_ts = self._get_safe_key(timestamp)
+ payload = self._build_run_payload(run, results)
+
+ self._write_run_file(run, timestamp[:10], compact_ts, payload)
+ self._write_task_instance_index(run, payload)
+ self._write_rule_indexes(run, results, timestamp)
+
+ def read_task_rule_history(
+ self, dag_id: str, task_id: str, rule_uid: str, limit: int = 100, before: str | None = None
Review Comment:
The passing order kinda has its meaning. Accepting positional args is probably ok. But i would make it keyword only for explicitness
##########
providers/common/dataquality/src/airflow/providers/common/dataquality/assets.py:
##########
@@ -0,0 +1,172 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+"""
+Asset-level data quality declarations.
+
+Quality configuration lives inside ``Asset.extra`` under the ``airflow.dataquality`` key, so it is
+serialized with the Dag and needs no Airflow core changes. The rules travel with the asset
+definition instead of being scattered across the Dags that check it.
+"""
+
+from __future__ import annotations
+
+import logging
+from typing import TYPE_CHECKING, Any
+
+from airflow.providers.common.dataquality.exceptions import DQRuleValidationError
+from airflow.providers.common.dataquality.rules import RuleSet
+from airflow.sdk import task
+
+if TYPE_CHECKING:
+ from collections.abc import Mapping, Sequence
+
+ from airflow.sdk import Asset
+ from airflow.sdk.execution_time.comms import AssetEventDagRunReferenceResult
+
+log = logging.getLogger(__name__)
+
+DQ_EXTRA_KEY = "airflow.dataquality"
+DQ_RESULT_EXTRA_KEY = "airflow.dataquality.result"
+
+
+def asset_quality(
+ asset: Asset,
+ *,
+ ruleset: RuleSet | dict[str, Any] | str,
+ conn_id: str | None = None,
+ table: str | None = None,
+) -> Asset:
+ """
+ Attach data quality configuration to an asset, returning the same asset.
+
+ :param asset: The asset the rules describe.
+ :param ruleset: A :class:`~airflow.providers.common.dataquality.rules.RuleSet`, its dict form, or a path
+ to a YAML ruleset file (resolved eagerly, at Dag-parse time).
+ :param conn_id: Default connection a check operator should use for this asset.
+ :param table: Default table to check; falls back to the asset name when unset.
+
+ Usage::
+
+ orders = asset_quality(
+ Asset("orders", uri="postgres://warehouse/analytics/orders"),
+ ruleset=rules,
+ conn_id="warehouse",
+ table="analytics.orders",
+ )
+ check = DQCheckOperator(task_id="dq", asset=orders)
+ """
+ if isinstance(ruleset, str):
+ ruleset = RuleSet.from_file(ruleset)
+ elif isinstance(ruleset, dict):
+ ruleset = RuleSet.from_dict(ruleset)
+ config: dict[str, Any] = {"ruleset": ruleset.to_dict()}
+ if conn_id:
+ config["conn_id"] = conn_id
+ if table:
+ config["table"] = table
+ asset.extra[DQ_EXTRA_KEY] = config
Review Comment:
Please don't use `asset.extra`. This is going to be removed https://github.com/apache/airflow/issues/55200.
##########
providers/common/dataquality/src/airflow/providers/common/dataquality/backends/object_storage.py:
##########
@@ -0,0 +1,256 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+"""
+Object-storage results backend.
+
+Each DQ check writes a keyed JSON document plus read indexes optimized for the UI:
+
+ runs/by_task/dag_id=<dag>/task_id=<task>/date=<2026-07-04>/<started_at>__<run_uid>.json
+ Canonical run record: ``{"run": ..., "results": [...], "summary": ...}``.
+
+ runs/by_task_instance/dag_id=<dag>/task_id=<task>/<safe_run_id>__<map_index>.json
+ Latest result for a task-instance page. Last write wins across retries.
+
+ rules/by_task_rule/dag_id=<dag>/task_id=<task>/rule_uid=<uid>/<started_at>__<run_uid>.json
+ One rule result plus run context: ``{"run": ..., "result": ...}``.
+
+The duplicate files are intentional read indexes: DQ tasks write once, while the UI reads
+many times. Keeping these indexes avoids scanning all task runs for common UI views.
+"""
+
+from __future__ import annotations
+
+import json
+import logging
+from datetime import datetime, timezone
+from typing import Any
+
+from airflow.providers.common.dataquality.results import DQRun, RuleResult, build_summary
+from airflow.sdk import ObjectStoragePath
+
+log = logging.getLogger(__name__)
+
+
+class ObjectStorageResultsBackend:
+ """Persist DQ results as JSON files via ``ObjectStoragePath``."""
+
+ def __init__(self, results_path: str, conn_id: str | None = None) -> None:
+ self.root = ObjectStoragePath(results_path, conn_id=conn_id)
+
+ def write_run(self, run: DQRun, results: list[RuleResult]) -> None:
+ timestamp = run.started_at or datetime.now(tz=timezone.utc).isoformat()
+ compact_ts = self._get_safe_key(timestamp)
+ payload = self._build_run_payload(run, results)
+
+ self._write_run_file(run, timestamp[:10], compact_ts, payload)
+ self._write_task_instance_index(run, payload)
+ self._write_rule_indexes(run, results, timestamp)
+
+ def read_task_rule_history(
+ self, dag_id: str, task_id: str, rule_uid: str, limit: int = 100, before: str | None = None
+ ) -> dict[str, Any]:
+ """Return recent results for one rule produced by one task, newest first."""
+ rule_dir = (
+ self.root
+ / "rules"
+ / "by_task_rule"
+ / f"dag_id={dag_id}"
+ / f"task_id={task_id}"
+ / f"rule_uid={rule_uid}"
+ )
+ return self._read_rule_history_dir(rule_dir, limit, before)
+
+ def read_task_runs(
+ self, dag_id: str, task_id: str, limit: int = 50, before: str | None = None
+ ) -> dict[str, Any]:
+ """
+ Return recent data quality runs for one task, newest first.
+
+ Returns ``{"items": [...], "next_cursor": ...}``. ``before`` is the opaque
+ ``next_cursor`` from the previous page, so "load more" only reads runs it hasn't
+ shown yet. ``next_cursor`` is set by reading one extra entry past ``limit``: cheap
+ on every page except the last one, where there's no way to confirm history is
+ exhausted without walking to the end.
+
+ ``date=`` partition names sort correctly as plain strings, so directories are walked
+ newest-first. Filenames are ``{started_at}__{run_uid}.json``, so sorting each partition's
+ filenames before scanning also walks newest-first within the partition -- required for
+ the early exit below to stop on the actual newest runs rather than an arbitrary subset.
+ Scanning stops as soon as ``limit + 1`` matching runs have been collected — a task with
+ years of history doesn't pay for a full scan on every page.
+ """
+ task_dir = self.root / "runs" / "by_task" / f"dag_id={dag_id}" / f"task_id={task_id}"
+ if not task_dir.exists():
+ return {"items": [], "next_cursor": None}
+
+ date_dirs = sorted((path for path in task_dir.iterdir() if path.is_dir()), reverse=True)
+ runs = []
+ for date_dir in date_dirs:
+ for path in sorted(date_dir.iterdir(), key=lambda p: p.name, reverse=True):
+ if not path.name.endswith(".json"):
+ continue
+ payload = self._read_json(path)
+ if payload is None:
+ continue
+ cursor = self._get_run_payload_cursor(payload)
+ if before is not None and cursor >= before:
+ continue
Review Comment:
```suggestion
if (payload := self._read_json(path)) is None:
continue
if before is not None and (cursor := self._get_run_payload_cursor(payload)) >= before:
continue
```
nit
##########
providers/common/dataquality/src/airflow/providers/common/dataquality/assets.py:
##########
@@ -0,0 +1,190 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+"""
+Asset-level data quality declarations.
+
+Quality configuration lives inside ``Asset.extra`` under the ``airflow.dataquality`` key, so it is
+serialized with the Dag and needs no Airflow core changes. The rules travel with the asset
+definition instead of being scattered across the Dags that check it.
+"""
+
+from __future__ import annotations
+
+import logging
+from typing import TYPE_CHECKING, Any
+
+from airflow.providers.common.dataquality.exceptions import DQRuleValidationError
+from airflow.providers.common.dataquality.rules import RuleSet
+from airflow.sdk import task
+
+if TYPE_CHECKING:
+ from collections.abc import Mapping, Sequence
+
+ from airflow.sdk import Asset
+ from airflow.sdk.execution_time.comms import AssetEventDagRunReferenceResult
+
+log = logging.getLogger(__name__)
+
+DQ_EXTRA_KEY = "airflow.dataquality"
+DQ_RESULT_EXTRA_KEY = "airflow.dataquality.result"
+
+
+def asset_quality(
+ asset: Asset,
Review Comment:
Are we going to support `AssetRef`, `AssetAlias`?
##########
providers/common/dataquality/src/airflow/providers/common/dataquality/backends/object_storage.py:
##########
@@ -0,0 +1,256 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+"""
+Object-storage results backend.
+
+Each DQ check writes a keyed JSON document plus read indexes optimized for the UI:
+
+ runs/by_task/dag_id=<dag>/task_id=<task>/date=<2026-07-04>/<started_at>__<run_uid>.json
+ Canonical run record: ``{"run": ..., "results": [...], "summary": ...}``.
+
+ runs/by_task_instance/dag_id=<dag>/task_id=<task>/<safe_run_id>__<map_index>.json
+ Latest result for a task-instance page. Last write wins across retries.
+
+ rules/by_task_rule/dag_id=<dag>/task_id=<task>/rule_uid=<uid>/<started_at>__<run_uid>.json
+ One rule result plus run context: ``{"run": ..., "result": ...}``.
+
+The duplicate files are intentional read indexes: DQ tasks write once, while the UI reads
+many times. Keeping these indexes avoids scanning all task runs for common UI views.
+"""
+
+from __future__ import annotations
+
+import json
+import logging
+from datetime import datetime, timezone
+from typing import Any
+
+from airflow.providers.common.dataquality.results import DQRun, RuleResult, build_summary
+from airflow.sdk import ObjectStoragePath
+
+log = logging.getLogger(__name__)
+
+
+class ObjectStorageResultsBackend:
+ """Persist DQ results as JSON files via ``ObjectStoragePath``."""
+
+ def __init__(self, results_path: str, conn_id: str | None = None) -> None:
+ self.root = ObjectStoragePath(results_path, conn_id=conn_id)
+
+ def write_run(self, run: DQRun, results: list[RuleResult]) -> None:
+ timestamp = run.started_at or datetime.now(tz=timezone.utc).isoformat()
+ compact_ts = self._get_safe_key(timestamp)
+ payload = self._build_run_payload(run, results)
+
+ self._write_run_file(run, timestamp[:10], compact_ts, payload)
+ self._write_task_instance_index(run, payload)
+ self._write_rule_indexes(run, results, timestamp)
+
+ def read_task_rule_history(
+ self, dag_id: str, task_id: str, rule_uid: str, limit: int = 100, before: str | None = None
+ ) -> dict[str, Any]:
+ """Return recent results for one rule produced by one task, newest first."""
+ rule_dir = (
+ self.root
+ / "rules"
+ / "by_task_rule"
+ / f"dag_id={dag_id}"
+ / f"task_id={task_id}"
+ / f"rule_uid={rule_uid}"
+ )
+ return self._read_rule_history_dir(rule_dir, limit, before)
+
+ def read_task_runs(
+ self, dag_id: str, task_id: str, limit: int = 50, before: str | None = None
+ ) -> dict[str, Any]:
+ """
+ Return recent data quality runs for one task, newest first.
+
+ Returns ``{"items": [...], "next_cursor": ...}``. ``before`` is the opaque
+ ``next_cursor`` from the previous page, so "load more" only reads runs it hasn't
+ shown yet. ``next_cursor`` is set by reading one extra entry past ``limit``: cheap
+ on every page except the last one, where there's no way to confirm history is
+ exhausted without walking to the end.
+
+ ``date=`` partition names sort correctly as plain strings, so directories are walked
+ newest-first. Filenames are ``{started_at}__{run_uid}.json``, so sorting each partition's
+ filenames before scanning also walks newest-first within the partition -- required for
+ the early exit below to stop on the actual newest runs rather than an arbitrary subset.
+ Scanning stops as soon as ``limit + 1`` matching runs have been collected — a task with
+ years of history doesn't pay for a full scan on every page.
+ """
+ task_dir = self.root / "runs" / "by_task" / f"dag_id={dag_id}" / f"task_id={task_id}"
+ if not task_dir.exists():
+ return {"items": [], "next_cursor": None}
+
+ date_dirs = sorted((path for path in task_dir.iterdir() if path.is_dir()), reverse=True)
+ runs = []
+ for date_dir in date_dirs:
+ for path in sorted(date_dir.iterdir(), key=lambda p: p.name, reverse=True):
+ if not path.name.endswith(".json"):
+ continue
+ payload = self._read_json(path)
+ if payload is None:
+ continue
+ cursor = self._get_run_payload_cursor(payload)
+ if before is not None and cursor >= before:
+ continue
+ runs.append(payload)
+ if len(runs) > limit:
+ break
+ if len(runs) > limit:
+ break
+
+ ordered = sorted(runs, key=self._get_run_payload_cursor, reverse=True)
+ page = ordered[:limit]
+ next_cursor = self._get_run_payload_cursor(page[-1]) if len(ordered) > limit and page else None
+ return {"items": page, "next_cursor": next_cursor}
Review Comment:
does the output format always look like this? if so, should we make it a TypedDict
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Airflow Data Quality Provider part 1 [airflow]
Posted by "gopidesupavan (via GitHub)" <gi...@apache.org>.
gopidesupavan commented on code in PR #69575:
URL: https://github.com/apache/airflow/pull/69575#discussion_r3575174829
##########
providers/common/dataquality/src/airflow/providers/common/dataquality/backends/object_storage.py:
##########
@@ -0,0 +1,252 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+"""
+Object-storage results backend.
+
+Each DQ check writes a keyed JSON document plus read indexes optimized for the UI:
+
+ runs/by_task/dag_id=<dag>/task_id=<task>/date=<2026-07-04>/<run_uid>.json
+ Canonical run record: ``{"run": ..., "results": [...], "summary": ...}``.
+
+ runs/by_task_instance/dag_id=<dag>/task_id=<task>/<safe_run_id>__<map_index>.json
+ Latest result for a task-instance page. Last write wins across retries.
+
+ rules/by_task_rule/dag_id=<dag>/task_id=<task>/rule_uid=<uid>/<started_at>__<run_uid>.json
+ One rule result plus run context: ``{"run": ..., "result": ...}``.
+
+The duplicate files are intentional read indexes: DQ tasks write once, while the UI reads
+many times. Keeping these indexes avoids scanning all task runs for common UI views.
+"""
+
+from __future__ import annotations
+
+import json
+import logging
+from datetime import datetime, timezone
+from typing import Any
+
+from airflow.providers.common.dataquality.results import DQRun, RuleResult, build_summary
+from airflow.sdk import ObjectStoragePath
+
+log = logging.getLogger(__name__)
+
+
+class ObjectStorageResultsBackend:
+ """Persist DQ results as JSON files via ``ObjectStoragePath``."""
+
+ def __init__(self, results_path: str, conn_id: str | None = None) -> None:
+ self.root = ObjectStoragePath(results_path, conn_id=conn_id)
+
+ def write_run(self, run: DQRun, results: list[RuleResult]) -> None:
+ timestamp = run.started_at or datetime.now(tz=timezone.utc).isoformat()
+ payload = self._build_run_payload(run, results)
+
+ self._write_run_file(run, timestamp[:10], payload)
+ self._write_task_instance_index(run, payload)
+ self._write_rule_indexes(run, results, timestamp)
+
+ def read_task_rule_history(
+ self, dag_id: str, task_id: str, rule_uid: str, limit: int = 100, before: str | None = None
+ ) -> dict[str, Any]:
+ """Return recent results for one rule produced by one task, newest first."""
+ rule_dir = (
+ self.root
+ / "rules"
+ / "by_task_rule"
+ / f"dag_id={dag_id}"
+ / f"task_id={task_id}"
+ / f"rule_uid={rule_uid}"
+ )
+ return self._read_rule_history_dir(rule_dir, limit, before)
+
+ def read_task_runs(
+ self, dag_id: str, task_id: str, limit: int = 50, before: str | None = None
+ ) -> dict[str, Any]:
+ """
+ Return recent data quality runs for one task, newest first.
+
+ Returns ``{"items": [...], "next_cursor": ...}``. ``before`` is the opaque
+ ``next_cursor`` from the previous page, so "load more" only reads runs it hasn't
+ shown yet. ``next_cursor`` is set by reading one extra entry past ``limit``: cheap
+ on every page except the last one, where there's no way to confirm history is
+ exhausted without walking to the end.
+
+ ``date=`` partition names sort correctly as plain strings, so directories are walked
+ newest-first and scanning stops as soon as ``limit + 1`` matching runs have been
+ collected — a task with years of history doesn't pay for a full scan on every page.
+ """
+ task_dir = self.root / "runs" / "by_task" / f"dag_id={dag_id}" / f"task_id={task_id}"
+ if not task_dir.exists():
+ return {"items": [], "next_cursor": None}
+
+ date_dirs = sorted((path for path in task_dir.iterdir() if path.is_dir()), reverse=True)
+ runs = []
+ for date_dir in date_dirs:
+ for path in date_dir.iterdir():
Review Comment:
good catch updated :)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Airflow Data Quality Provider part 1 [airflow]
Posted by "gopidesupavan (via GitHub)" <gi...@apache.org>.
gopidesupavan commented on code in PR #69575:
URL: https://github.com/apache/airflow/pull/69575#discussion_r3575176390
##########
providers/common/dataquality/src/airflow/providers/common/dataquality/assets.py:
##########
@@ -0,0 +1,172 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+"""
+Asset-level data quality declarations.
+
+Quality configuration lives inside ``Asset.extra`` under the ``airflow.dataquality`` key, so it is
+serialized with the Dag and needs no Airflow core changes. The rules travel with the asset
+definition instead of being scattered across the Dags that check it.
+"""
+
+from __future__ import annotations
+
+import logging
+from typing import TYPE_CHECKING, Any
+
+from airflow.providers.common.dataquality.exceptions import DQRuleValidationError
+from airflow.providers.common.dataquality.rules import RuleSet
+from airflow.sdk import task
+
+if TYPE_CHECKING:
+ from collections.abc import Mapping, Sequence
+
+ from airflow.sdk import Asset
+ from airflow.sdk.execution_time.comms import AssetEventDagRunReferenceResult
+
+log = logging.getLogger(__name__)
+
+DQ_EXTRA_KEY = "airflow.dataquality"
+DQ_RESULT_EXTRA_KEY = "airflow.dataquality.result"
+
+
+def asset_quality(
+ asset: Asset,
+ *,
+ ruleset: RuleSet | dict[str, Any] | str,
+ conn_id: str | None = None,
+ table: str | None = None,
+) -> Asset:
+ """
+ Attach data quality configuration to an asset, returning the same asset.
+
+ :param asset: The asset the rules describe.
+ :param ruleset: A :class:`~airflow.providers.common.dataquality.rules.RuleSet`, its dict form, or a path
+ to a YAML ruleset file (resolved eagerly, at Dag-parse time).
+ :param conn_id: Default connection a check operator should use for this asset.
+ :param table: Default table to check; falls back to the asset name when unset.
+
+ Usage::
+
+ orders = asset_quality(
+ Asset("orders", uri="postgres://warehouse/analytics/orders"),
+ ruleset=rules,
+ conn_id="warehouse",
+ table="analytics.orders",
+ )
+ check = DQCheckOperator(task_id="dq", asset=orders)
+ """
+ if isinstance(ruleset, str):
+ ruleset = RuleSet.from_file(ruleset)
+ elif isinstance(ruleset, dict):
+ ruleset = RuleSet.from_dict(ruleset)
+ config: dict[str, Any] = {"ruleset": ruleset.to_dict()}
+ if conn_id:
+ config["conn_id"] = conn_id
+ if table:
+ config["table"] = table
+ asset.extra[DQ_EXTRA_KEY] = config
+ return asset
+
+
+def get_asset_quality_config(asset: Asset) -> dict[str, Any] | None:
+ """Return the raw ``airflow.dataquality`` config attached to an asset, if any."""
+ config = asset.extra.get(DQ_EXTRA_KEY)
+ if not isinstance(config, dict):
+ return None
+ return config
+
+
+def get_asset_ruleset(asset: Asset) -> RuleSet:
+ """Return the ruleset attached to an asset, raising when none is attached."""
+ config = get_asset_quality_config(asset)
+ if not config or "ruleset" not in config:
+ raise DQRuleValidationError(
+ f"Asset {asset.name!r} has no data quality config; attach one with asset_quality()"
+ )
+ return RuleSet.from_dict(config["ruleset"])
+
+
+def _quality_score_passes(
+ asset: Asset,
+ min_score: float,
+ triggering_asset_events: Mapping[Asset, Sequence[AssetEventDagRunReferenceResult]],
+) -> bool:
+ """Pure decision logic behind :func:`require_quality`, kept separate so it is testable without a Dag."""
+ events = triggering_asset_events.get(asset, [])
+ if not events:
+ log.warning(
+ "require_quality(%s): no triggering event for this run; skipping downstream tasks",
+ asset.name,
+ )
+ return False
+
+ summary = events[-1].extra.get(DQ_RESULT_EXTRA_KEY)
Review Comment:
nice suggestion updated to default all events..
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org