Airflow Data Quality Provider Plugin

.pre-commit-config.yaml

@@ -783,6 +783,7 @@ repos:
           ^airflow-core/src/airflow/ui/openapi-gen/|
           ^airflow-core/src/airflow/ui/pnpm-lock\.yaml$|
           ^providers/common/ai/src/airflow/providers/common/ai/plugins/www/pnpm-lock\.yaml$|
+          ^providers/common/dataquality/src/airflow/providers/common/dataquality/plugins/www/pnpm-lock\.yaml$|
           ^airflow-core/src/airflow/ui/public/i18n/locales/de/README\.md$|
           ^airflow-core/src/airflow/ui/src/i18n/config\.ts$|
           ^\.agents/skills/airflow-translations/|

airflow-core/tests/unit/api_fastapi/core_api/routes/public/test_plugins.py

@@ -39,13 +39,14 @@ class TestGetPlugins:
             # Filters
             (
                 {},
-                19,
+                20,
                 [
                     "InformaticaProviderPlugin",
                     "MetadataCollectionPlugin",
                     "OpenLineageProviderPlugin",
                     "business_day_window_plugin",
                     "databricks_workflow",
+                    "dataquality",
                     "decreasing_priority_weight_strategy_plugin",
                     "edge_executor",
                     "hitl_review",
@@ -64,14 +65,14 @@ class TestGetPlugins:
             ),
             (
                 {"limit": 3, "offset": 3},
-                19,
+                20,
                 [
                     "business_day_window_plugin",
                     "databricks_workflow",
-                    "decreasing_priority_weight_strategy_plugin",
+                    "dataquality",
                 ],
             ),
-            ({"limit": 1}, 19, ["InformaticaProviderPlugin"]),
+            ({"limit": 1}, 20, ["InformaticaProviderPlugin"]),
         ],
     )
     def test_should_respond_200(
@@ -174,7 +175,7 @@ def test_invalid_external_view_destination_should_log_warning_and_continue(self,
         assert len(plugins_page) == 8
         assert "test_plugin_invalid" not in [p["name"] for p in plugins_page]
 
-        assert body["total_entries"] == 19
+        assert body["total_entries"] == 20
 
 
 @skip_if_force_lowest_dependencies_marker

airflow-core/tests/unit/plugins/test_plugins_manager.py

@@ -408,7 +408,7 @@ def test_does_not_double_import_entrypoint_provider_plugins(self):
         # Mock/skip loading from plugin dir
         with mock.patch("airflow.plugins_manager._load_plugins_from_plugin_directory", return_value=([], [])):
             plugins = plugins_manager._get_plugins()[0]
-        assert len(plugins) == 7
+        assert len(plugins) == 8
 
 
 class TestWindowPluginRegistration:

dev/breeze/src/airflow_breeze/commands/developer_commands.py

@@ -127,6 +127,7 @@
 from airflow_breeze.utils.path_utils import (
     AIRFLOW_ROOT_PATH,
     COMMON_AI_PLUGIN_PREK_HOOK,
+    COMMON_DATAQUALITY_PLUGIN_PREK_HOOK,
     EDGE_PLUGIN_PREK_HOOK,
     FAB_AUTH_MANAGER_WWW_PREK_HOOK,
     cleanup_python_generated_files,
@@ -663,7 +664,7 @@ def start_airflow(
     if use_airflow_version is None and not skip_assets_compilation:
         assert_prek_installed()
         # Compile provider assets if needed
-        additional_assets = [COMMON_AI_PLUGIN_PREK_HOOK]
+        additional_assets = [COMMON_AI_PLUGIN_PREK_HOOK, COMMON_DATAQUALITY_PLUGIN_PREK_HOOK]
         if executor and EDGE_EXECUTOR in executor:
             additional_assets.append(EDGE_PLUGIN_PREK_HOOK)
         if auth_manager == FAB_AUTH_MANAGER:

dev/breeze/src/airflow_breeze/utils/path_utils.py

@@ -477,6 +477,22 @@ def find_airflow_root_path_to_operate_on() -> Path:
 COMMON_AI_UI_PLUGIN_DIST_PATH = COMMON_AI_PLUGIN_PATH / "dist"
 COMMON_AI_PLUGIN_PREK_HOOK = "compile-common-ai-provider-assets"
 
+COMMON_DATAQUALITY_PLUGIN_PATH = (
+    AIRFLOW_PROVIDERS_ROOT_PATH
+    / "common"
+    / "dataquality"
+    / "src"
+    / "airflow"
+    / "providers"
+    / "common"
+    / "dataquality"
+    / "plugins"
+    / "www"
+)
+COMMON_DATAQUALITY_UI_PLUGIN_NODE_MODULES_PATH = COMMON_DATAQUALITY_PLUGIN_PATH / "node_modules"
+COMMON_DATAQUALITY_UI_PLUGIN_DIST_PATH = COMMON_DATAQUALITY_PLUGIN_PATH / "dist"
+COMMON_DATAQUALITY_PLUGIN_PREK_HOOK = "compile-common-dataquality-provider-assets"
+
 
 DAGS_PATH = AIRFLOW_ROOT_PATH / "dags"
 FILES_PATH = AIRFLOW_ROOT_PATH / "files"

dev/breeze/src/airflow_breeze/utils/run_utils.py

@@ -43,6 +43,9 @@
     COMMON_AI_PLUGIN_PREK_HOOK,
     COMMON_AI_UI_PLUGIN_DIST_PATH,
     COMMON_AI_UI_PLUGIN_NODE_MODULES_PATH,
+    COMMON_DATAQUALITY_PLUGIN_PREK_HOOK,
+    COMMON_DATAQUALITY_UI_PLUGIN_DIST_PATH,
+    COMMON_DATAQUALITY_UI_PLUGIN_NODE_MODULES_PATH,
     EDGE_PLUGIN_PREK_HOOK,
     EDGE_PLUGIN_UI_DIST_PATH,
     EDGE_PLUGIN_UI_NODE_MODULES_PATH,
@@ -528,6 +531,9 @@ def _clean_ui_assets(additional_ui_hooks: list[str]):
     if COMMON_AI_PLUGIN_PREK_HOOK in additional_ui_hooks:
         shutil.rmtree(COMMON_AI_UI_PLUGIN_NODE_MODULES_PATH, ignore_errors=True)
         shutil.rmtree(COMMON_AI_UI_PLUGIN_DIST_PATH, ignore_errors=True)
+    if COMMON_DATAQUALITY_PLUGIN_PREK_HOOK in additional_ui_hooks:
+        shutil.rmtree(COMMON_DATAQUALITY_UI_PLUGIN_NODE_MODULES_PATH, ignore_errors=True)
+        shutil.rmtree(COMMON_DATAQUALITY_UI_PLUGIN_DIST_PATH, ignore_errors=True)
 
     console_print("[success]Cleaned ui assets[/]")
 

dev/breeze/tests/test_selective_checks.py

@@ -2884,7 +2884,7 @@ def test_upgrade_to_newer_dependencies(
             ("providers/common/sql/src/airflow/providers/common/sql/common_sql_python.py",),
             {
                 "docs-list-as-string": "amazon apache.drill apache.druid apache.hive apache.iceberg "
-                "apache.impala apache.pinot clickhousedb common.ai common.compat common.sql databricks elasticsearch "
+                "apache.impala apache.pinot clickhousedb common.ai common.compat common.dataquality common.sql databricks elasticsearch "
                 "exasol google informatica jdbc microsoft.mssql mysql odbc openlineage "
                 "oracle pgvector postgres presto slack snowflake sqlite teradata trino vertica ydb",
             },

docs/spelling_wordlist.txt

@@ -294,6 +294,7 @@ conda
 conf
 Config
 config
+ConfigDict
 configfile
 configMap
 configmap

providers/common/dataquality/.gitignore

@@ -1 +1,2 @@
+www-hash.txt
 *.iml

providers/common/dataquality/.pre-commit-config.yaml

@@ -0,0 +1,59 @@
+# 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.
+---
+default_stages: [pre-commit, pre-push]
+minimum_prek_version: '0.3.4'
+default_language_version:
+  python: python3
+  node: 22.19.0
+  golang: 1.24.0
+repos:
+  - repo: local
+    hooks:
+      - id: ts-compile-lint-common-dataquality-ui
+        name: Compile Data Quality UI
+        description: Type-check changed TypeScript files in the Data Quality plugin UI
+        language: node
+        files: |
+          (?x)
+          ^src/airflow/providers/common/dataquality/plugins/www/.*\.(js|ts|tsx|yaml|css|json)$
+        exclude: |
+          (?x)
+          ^src/airflow/providers/common/dataquality/plugins/www/node-modules/.*|
+          ^src/airflow/providers/common/dataquality/plugins/www/.pnpm-store
+        entry: ../../../scripts/ci/prek/ts_compile_lint_common_dataquality.py
+        additional_dependencies: ['pnpm@10.25.0']
+        pass_filenames: true
+        require_serial: true
+      - id: compile-common-dataquality-provider-assets
+        name: Compile Common Data Quality provider assets
+        language: node
+        stages: ['pre-commit', 'manual']
+        files: ^src/airflow/providers/common/dataquality/plugins/www/
+        entry: ../../../scripts/ci/prek/compile_provider_assets.py dataquality
+        pass_filenames: false
+        additional_dependencies: ['pnpm@10.25.0']
+      - id: generate-common-dataquality-ruleset-schema
+        name: Generate Data Quality RuleSet schema
+        language: python
+        entry: ../../../scripts/ci/prek/generate_common_dataquality_ruleset_schema.py
+        pass_filenames: false
+        always_run: true
+        files: >
+          (?x)
+          ^src/airflow/providers/common/dataquality/rules/.*\.py$|
+          ^src/airflow/providers/common/dataquality/skills/dataquality-rule-authoring/references/ruleset\.schema\.json$

providers/common/dataquality/README.rst

@@ -19,7 +19,11 @@ Package ``apache-airflow-providers-common-dataquality``
 
 Release: ``0.1.0``
 
-Common Data Quality Provider
+``Data Quality Provider``
+
+Declarative data quality rules with durable, per-rule execution history. Checks run through
+``common.sql`` DB-API hooks; results are persisted to a configurable results store (object
+storage or local files) so task, run, and rule-level quality can be inspected over time.
 
 Provider package
 ----------------
@@ -37,8 +41,12 @@ see ``Requirements`` below.
 Requirements
 ------------
 
-==================  ==================
-PIP package         Version required
-==================  ==================
-``apache-airflow``  ``>=3.0.0``
-==================  ==================
+==========================================  ==================
+PIP package                                 Version required
+==========================================  ==================
+``apache-airflow``                          ``>=3.0.0``
+``apache-airflow-providers-common-compat``  ``>=1.15.0``
+``apache-airflow-providers-common-sql``     ``>=2.0.0``
+``pydantic``                                ``>=2.11.0``
+``pyyaml``                                  ``>=6.0.2``
+==========================================  ==================

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]

providers/common/dataquality/docs/assets.rst

@@ -0,0 +1,70 @@
+ .. 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:assets:
+
+Assets and quality gating
+============================
+
+Quality rules can travel with the :class:`~airflow.sdk.Asset` they describe instead of being
+scattered across every Dag that checks it, and a downstream consumer Dag can refuse to run when
+the data it was triggered by did not meet a minimum quality bar.
+
+Attaching a ruleset to an asset
+----------------------------------
+
+:func:`~airflow.providers.common.dataquality.assets.asset_quality` stores ruleset, connection, and table
+configuration inside ``Asset.extra`` under the ``airflow.dataquality`` key, so it is serialized with the
+Dag and needs no Airflow core changes:
+
+.. exampleinclude:: /../src/airflow/providers/common/dataquality/example_dags/example_dq_require_quality.py
+    :language: python
+    :start-after: [START howto_asset_quality]
+    :end-before: [END howto_asset_quality]
+
+Pass the resulting asset to :class:`~airflow.providers.common.dataquality.operators.dq_check.DQCheckOperator`
+via ``asset=`` (see :doc:`operators`) instead of ``table``/``ruleset``/``conn_id``: the operator
+resolves all three from the asset's config, adds the asset to its own outlets, and attaches its
+summary -- including the quality ``score`` used below -- to the asset event.
+
+Only one Dag should call ``asset_quality()`` for a given asset ``name``/``uri``. Airflow keeps one
+shared record per asset across all Dags, so if more than one Dag attaches (or omits) config for the
+same asset, whichever Dag parsed most recently determines what's stored.
+
+Gating a consumer Dag on quality
+------------------------------------
+
+:func:`~airflow.providers.common.dataquality.assets.require_quality` builds a ``@task.short_circuit`` task that
+reads the ``score`` off the asset event that triggered the current run, and skips every
+downstream task when that event has no quality summary at all, or its score is below
+``min_score``:
+
+.. exampleinclude:: /../src/airflow/providers/common/dataquality/example_dags/example_dq_require_quality.py
+    :language: python
+    :start-after: [START howto_require_quality]
+    :end-before: [END howto_require_quality]
+
+Put the gate first in a Dag scheduled by the asset, and chain everything else after it:
+
+.. code-block:: python
+
+    with DAG("orders_consumer", schedule=orders_asset) as consumer:
+        gate = require_quality(orders_asset, min_score=0.95)
+        gate >> process_orders()
+
+``min_score`` must be between ``0`` and ``1``; the check considers only the *most recent*
+triggering event for the asset when a run was triggered by several.

providers/common/dataquality/docs/configurations-ref.rst

@@ -0,0 +1,19 @@
+ .. 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.
+
+.. include:: /../../../../devel-common/src/sphinx_exts/includes/providers-configurations-ref.rst
+.. include:: /../../../../devel-common/src/sphinx_exts/includes/sections-and-options.rst