Extend end to end methods for applying checks

.github/workflows/fork-sync.yml

@@ -0,0 +1,73 @@
+# Sync fork PRs to a branch in this repo and open a test PR.
+# Acceptance, anomaly, and performance workflows are skipped for PRs from forks.
+# This workflow pushes the fork's code to a branch here so those tests can run on the new PR.
+#
+# Note: Uses pull_request_target for base-repo GITHUB_TOKEN write access.
+# Fork PRs from first-time contributors may need maintainer approval (Actions tab) before running.
+
+name: fork-sync
+
+on:
+  pull_request_target:
+    types: [opened, synchronize, ready_for_review]
+    branches: [main]
+  merge_group:
+    types: [checks_requested]
+  push:
+    branches: [main]
+
+permissions:
+  contents: write
+  pull-requests: write
+
+concurrency:
+  group: fork-sync-${{ github.event.pull_request.number }}
+  cancel-in-progress: false
+
+jobs:
+  sync-fork-to-branch:
+    if: github.event.pull_request.head.repo.fork && !github.event.pull_request.draft
+    runs-on: ubuntu-latest
+    steps:
+      - name: Checkout base repository
+        uses: actions/checkout@v4
+        with:
+          fetch-depth: 0
+
+      - name: Install GitHub CLI
+        uses: cli/cli-action@v2
+        with:
+          version: latest
+
+      - name: Fetch PR branch from fork
+        env:
+          GH_TOKEN: ${{ secrets.GITHUB_TOKEN }}
+        run: gh pr checkout ${{ github.event.pull_request.number }}
+
+      - name: Configure git
+        run: |
+          git config user.name "${{ github.actor }}"
+          git config user.email "${{ github.actor }}@users.noreply.github.com"
+
+      - name: Push to sync branch and create test PR
+        env:
+          GH_TOKEN: ${{ secrets.GITHUB_TOKEN }}
+          SYNC_BRANCH: "fork-test/pr-${{ github.event.pull_request.number }}"
+        run: |
+          set -e
+          git checkout -B "$SYNC_BRANCH"
+          git push --force "https://x-access-token:${GH_TOKEN}@github.com/${GITHUB_REPOSITORY}.git" "HEAD:$SYNC_BRANCH"
+
+          # Create PR if it does not exist
+          EXISTING=$(gh pr list --head "$SYNC_BRANCH" --state open --json number -q '.[0].number' 2>/dev/null || true)
+          if [ -z "$EXISTING" ]; then
+            gh pr create \
+              --base "${{ github.event.pull_request.base.ref }}" \
+              --head "$SYNC_BRANCH" \
+              --title "Fork test: PR #${{ github.event.pull_request.number }}" \
+              --body "Automated sync from fork PR for CI testing.
+
+          Original PR: ${{ github.event.pull_request.html_url }}
+
+          All tests, including acceptance, anomaly, and performance tests, run on this PR (they are skipped for fork PRs)."
+          fi

demos/dqx_demo_asset_bundle/dqx_demo_notebook.py

@@ -592,8 +592,10 @@
 maintenance_quality_checks = dq_engine.load_checks(config=WorkspaceFileChecksStorageConfig(location=sensor_rules_file_path))
 
 # Apply the checks and write the output data
+# By default, apply_checks_and_save_in_table method apply checks to the entire input table.
+# Incremental processing is supported using streaming with the AvailableNow trigger for batch-style execution, along with checkpointing to ensure consistency across runs.
 dq_engine.apply_checks_by_metadata_and_save_in_table(
-  checks=maintenance_quality_checks,
+  checks=maintenance_quality_checks,  # or provide checks_location and run_config_name to auto-load from checks storage
   input_config=InputConfig(sensor_table),
   output_config=OutputConfig(f"{sensor_table}_valid", mode="overwrite"),
   quarantine_config=OutputConfig(f"{sensor_table}_quarantine", mode="overwrite")
@@ -612,8 +614,10 @@
 maintenance_quality_checks = dq_engine.load_checks(config=WorkspaceFileChecksStorageConfig(location=maintenance_rules_file_path))
 
 # Apply the checks and write the output data
+# By default, apply_checks_and_save_in_table method apply checks to the entire input table.
+# Incremental processing is supported using streaming with the AvailableNow trigger for batch-style execution, along with checkpointing to ensure consistency across runs.
 dq_engine.apply_checks_by_metadata_and_save_in_table(
-  checks=maintenance_quality_checks,
+  checks=maintenance_quality_checks,  # or provide checks_location and run_config_name to auto-load from checks storage
   input_config=InputConfig(maintenance_table),
   output_config=OutputConfig(f"{maintenance_table}_valid", mode="overwrite"),
   quarantine_config=OutputConfig(f"{maintenance_table}_quarantine", mode="overwrite")

demos/dqx_demo_library.py

@@ -474,10 +474,12 @@
 
 # COMMAND ----------
 
-# end-to-end quality checking flow
+# End-to-end quality checking flow
+# By default, apply_checks_and_save_in_table method apply checks to the entire input table.
+# Incremental processing is supported using streaming with the AvailableNow trigger for batch-style execution, along with checkpointing to ensure consistency across runs.
 dq_engine.apply_checks_by_metadata_and_save_in_table(
     input_config=InputConfig("/databricks-datasets/delta-sharing/samples/nyctaxi_2019"),
-    checks=checks,
+    checks=checks,  # or provide checks_location and run_config_name to auto-load from checks storage
     output_config=OutputConfig(f"{demo_catalog_name}.{demo_schema_name}.dqx_e2e_output", mode="overwrite"),
     quarantine_config=OutputConfig(f"{demo_catalog_name}.{demo_schema_name}.dqx_e2e_quarantine", mode="overwrite")
 )

demos/dqx_demo_summary_metrics.py

@@ -264,8 +264,10 @@
 metrics_config = OutputConfig(location=metrics_table_name, mode="overwrite")
 
 # Read the input data, apply checks, generate metrics, and save results
+# By default, apply_checks_and_save_in_table method apply checks to the entire input table.
+# Incremental processing is supported using streaming with the AvailableNow trigger for batch-style execution, along with checkpointing to ensure consistency across runs.
 dq_engine.apply_checks_by_metadata_and_save_in_table(
-  checks=checks_from_yaml,
+  checks=checks_from_yaml,  # or provide checks_location and run_config_name to auto-load from checks storage
   input_config=input_config,
   output_config=output_config,
   metrics_config=metrics_config

demos/dqx_demo_tool.py

@@ -372,9 +372,10 @@
 
 from databricks.labs.dqx.config import InputConfig, OutputConfig
 
-
+# By default, apply_checks_and_save_in_table method apply checks to the entire input table.
+# Incremental processing is supported using streaming with the AvailableNow trigger for batch-style execution, along with checkpointing to ensure consistency across runs.
 dq_engine.apply_checks_by_metadata_and_save_in_table(
-    checks=checks,
+    checks=checks,  # or provide checks_location and run_config_name to auto-load from checks storage
     input_config=run_config.input_config,
     output_config=run_config.output_config,
     quarantine_config=run_config.quarantine_config,

demos/dqx_intermediate_demo_library.py

@@ -425,9 +425,28 @@
 quarantine_config = OutputConfig(quarantine_table)  # optional
 metrics_config = OutputConfig(metrics_table)  # optional
 
-# Use End to End method: read the data, apply the checks, write data to valid, quarantine and metrics tables
+# End to End method with preloaded checks: Apply checks to the entire input table -> write data to valid, quarantine and metrics tables
+# By default, apply_checks_and_save_in_table method apply checks to the entire input table.
+# Incremental processing is supported using streaming with the AvailableNow trigger for batch-style execution, along with checkpointing to ensure consistency across runs.
 engine.apply_checks_by_metadata_and_save_in_table(
-    checks=loaded_rules,
+    checks=loaded_rules,  # # or provide checks_location and run_config_name to auto-load from checks storage
+    input_config=input_config,
+    output_config=output_config,
+    quarantine_config=quarantine_config,
+    metrics_config=metrics_config,
+    checks_location=rules_table # for reporting purpose only
+)
+
+# COMMAND ----------
+
+display(spark.table(metrics_table))
+
+# COMMAND ----------
+
+# End to End method:
+# -> Load checks from a storage -> apply the checks to the entire input table -> write data to valid, quarantine and metrics tables
+engine.apply_checks_by_metadata_and_save_in_table(
+    checks_location=rules_table,
     input_config=input_config,
     output_config=output_config,
     quarantine_config=quarantine_config,

demos/dqx_streaming_demo_native.py

@@ -225,7 +225,7 @@
 dq_engine = DQEngine(WorkspaceClient())
 
 dq_engine.apply_checks_by_metadata_and_save_in_table(
-    checks=checks,
+    checks=checks,  # or provide checks_location and run_config_name to auto-load from checks storage
     input_config=input_config,
     output_config=output_config,
     quarantine_config=quarantine_config

docs/dqx/docs/guide/best_practices.mdx

@@ -104,6 +104,12 @@ Bootstrap rule candidates:
 * Use the profiler to generate initial quality rules for new datasets and periodically re-profile existing datasets to detect changes or shifts in data distributions.
 * Leverage AI-assisted quality rule generation to translate technical or business requirements expressed in natural language into DQX-specific rules, accelerating rule creation.
 
+<Admonition type="tip" title="Bootstrap in Pre-Production">
+Perform all rule bootstrapping (profiling, AI-assisted rule generation) in lower, pre-production environments using representative sample data.
+Promote the resulting validated rules to production rather than running the profiler or LLM-based generation directly in production.
+This avoids the need for special-purpose compute (profiler clusters) and external AI/LLM dependencies in your production environment, reducing cost, complexity, and operational risk.
+</Admonition>
+
 Continuously refine and tune rules:
 * Monitor data quality metrics (pass/fail rates, trends, and anomalies) to identify areas for improvement and reduce false positives while maintaining strong coverage.
 * Add or tune checks (e.g. update thresholds) immediately when production issues are detected, preventing recurring data failures.

docs/dqx/docs/guide/quality_checks_apply.mdx

@@ -23,12 +23,12 @@ Checks can be applied to the input data by one of the following methods of the `
 * For checks defined with DQX classes:
   * `apply_checks`: apply quality checks to a DataFrame and report issues as additional columns.
   * `apply_checks_and_split`: apply quality checks and split the input DataFrame into valid and invalid (quarantined) DataFrames.
-  * `apply_checks_and_save_in_table`: end-to-end approach to apply quality checks to a table and save results to the output table(s) via single method call.
+  * `apply_checks_and_save_in_table`: end-to-end approach to apply quality checks to a table and save results to the output table(s) via single method call. Checks can be passed directly or loaded automatically from a storage location.
 
 * For checks defined as metadata (list of dictionaries, or [loaded from a storage](/docs/guide/quality_checks_storage)):
   * `apply_checks_by_metadata`: apply quality checks to a DataFrame and report issues as additional columns.
   * `apply_checks_by_metadata_and_split`: apply quality checks and split the input DataFrame into valid and invalid (quarantined) DataFrames.
-  * `apply_checks_by_metadata_and_save_in_table`: end-to-end approach to apply quality checks to a table and save results to output table(s) via single method call.
+  * `apply_checks_by_metadata_and_save_in_table`: end-to-end approach to apply quality checks to a table and save results to output table(s) via single method call. Checks can be passed directly as metadata/dict or loaded automatically from a storage location.
 
 * There are also method for applying checks to multiple input locations at once:
   * `apply_checks_and_save_in_tables`: apply quality checks to input locations defined by provided list of `RunConfig`.
@@ -111,6 +111,8 @@ You can customize the reporting columns as described in the [additional configur
     )
 
     # Option 3 End-to-End approach: apply quality checks to a table and save results to valid and invalid (quarantined) tables
+    # By default, apply_checks_and_save_in_table method apply checks to the entire input table.
+    # Incremental processing is supported using streaming with the AvailableNow trigger for batch-style execution, along with checkpointing to ensure consistency across runs.
     dq_engine.apply_checks_and_save_in_table(
         checks=checks,
         input_config=InputConfig(location="catalog.schema.input"),
@@ -124,6 +126,15 @@ You can customize the reporting columns as described in the [additional configur
         input_config=InputConfig(location="catalog.schema.input"),
         output_config=OutputConfig(location="catalog.schema.output"),
     )
+
+    # Option 5 End-to-End approach: load checks from a storage location and apply them to the entire input table
+    dq_engine.apply_checks_and_save_in_table(
+        input_config=InputConfig(location="catalog.schema.input"),
+        output_config=OutputConfig(location="catalog.schema.valid"),
+        quarantine_config=OutputConfig(location="catalog.schema.quarantine"),
+        checks_location="catalog.schema.dq_checks",  # delta table, volume or workspace file path
+        run_config_name="default",  # required for checks stored in a table
+    )
     ```
   </TabItem>
 </Tabs>
@@ -200,6 +211,17 @@ In the example below, checks are defined in YAML syntax for convenience and then
         input_config=InputConfig(location="catalog.schema.input"),
         output_config=OutputConfig(location="catalog.schema.output"),
     )
+
+    # Option 5 End-to-End approach: load checks from a storage location and apply them to the entire input table
+    # By default, apply_checks_and_save_in_table method apply checks to the entire input table.
+    # Incremental processing is supported using streaming with the AvailableNow trigger for batch-style execution, along with checkpointing to ensure consistency across runs.
+    dq_engine.apply_checks_by_metadata_and_save_in_table(
+        input_config=InputConfig(location="catalog.schema.input"),
+        output_config=OutputConfig(location="catalog.schema.valid"),
+        quarantine_config=OutputConfig(location="catalog.schema.quarantine"),
+        checks_location="catalog.schema.dq_checks",  # delta table, volume or workspace file path
+        run_config_name="default",  # required for checks stored in a table
+    )
     ```
   </TabItem>
 </Tabs>
@@ -555,7 +577,7 @@ dq_engine.save_results_in_table(
 
 # End-to-end method: apply checks and save results in tables using DQX method
 dq_engine.apply_checks_and_save_in_table(
-    checks=checks,
+    checks=checks,  # or provide checks_location and run_config_name to auto-load from checks storage
     input_config=InputConfig(
       location="catalog.schema.input"
     ),
@@ -574,7 +596,7 @@ dq_engine.apply_checks_and_save_in_table(
 
 # End-to-end method: apply checks and save results in tables using streaming using DQX method
 dq_engine.apply_checks_and_save_in_table(
-    checks=checks,
+    checks=checks,  # or provide checks_location and run_config_name to auto-load from checks storage
     input_config=InputConfig(
       location="catalog.schema.input",
       is_streaming=True,
@@ -927,8 +949,10 @@ To enable summary metrics programmatically, create and pass a `DQMetricsObserver
     engine.save_summary_metrics(observed_metrics=metrics_observation.get, metrics_config=OutputConfig(location="catalog.schema.metrics"))
 
     # Option 3 End-to-End approach: apply quality checks to a table, save results to valid and invalid (quarantined) tables, and save metrics to a metrics table
+    # By default, apply_checks_and_save_in_table method apply checks to the entire input table.
+    # Incremental processing is supported using streaming with the AvailableNow trigger for batch-style execution, along with checkpointing to ensure consistency across runs.
     dq_engine.apply_checks_and_save_in_table(
-        checks=checks,
+        checks=checks,  # or provide checks_location and run_config_name to auto-load from checks storage
         input_config=InputConfig(location="catalog.schema.input"),
         output_config=OutputConfig(location="catalog.schema.valid"),
         quarantine_config=OutputConfig(location="catalog.schema.quarantine"),
@@ -937,7 +961,7 @@ To enable summary metrics programmatically, create and pass a `DQMetricsObserver
 
     # Option 4 End-to-End approach: apply quality checks to a table, save results to an output table, and save metrics to a metrics table
     dq_engine.apply_checks_and_save_in_table(
-        checks=checks,
+        checks=checks,  # or provide checks_location and run_config_name to auto-load from checks storage
         input_config=InputConfig(location="catalog.schema.input"),
         output_config=OutputConfig(location="catalog.schema.output"),
         metrics_config=OutputConfig(location="catalog.schema.metrics"),

docs/dqx/docs/guide/summary_metrics.mdx

@@ -88,7 +88,7 @@ Metrics are not directly accessible from the returned Spark Observation when dat
 
 ### Writing Metrics to a Table
 
-Engine methods for applying checks (e.g. `apply_checks_and_save_in_table`, `apply_checks_by_metadata_and_save_in_table`, `save_results_in_table`, `apply_checks_and_save_in_tables`, `apply_checks_and_save_in_tables_for_patterns`) can write summary metrics into a table automatically.
+End-to-end engine methods for applying checks (e.g. `apply_checks_and_save_in_table`, `apply_checks_by_metadata_and_save_in_table`, `save_results_in_table`, `apply_checks_and_save_in_tables`, `apply_checks_and_save_in_tables_for_patterns`) can write summary metrics into a table automatically using configuration.
 Metrics can be written to a table in batch or streaming. You can write metrics for different datasets or workloads into a common metrics table to track data quality over time centrally.
 
 The `name` specified in the `DQMetricsObserver` is recorded as the `run_name` column in the metrics table. It is recommended to assign a unique `name` to the observer for each job or table to enable efficient filtering when centralizing metrics in a single table.
@@ -159,8 +159,9 @@ Summary metrics can be written to a table when calling `DQEngine` methods to app
     )
 
     # Option 2: Use End to End method: read the data, apply the checks, write data to valid and quarantine tables, and write metrics to the metrics table
+    # By default, checks are applied to the entire input table. See next section for incremental support with Streaming.
     engine.apply_checks_and_save_in_table(
-        checks=checks,
+        checks=checks,  # or provide checks_location and run_config_name to auto-load from checks storage
         input_config=input_config,
         output_config=output_config,
         quarantine_config=quarantine_config,
@@ -251,7 +252,7 @@ You must use streaming metrics listener or end-to-end methods that persist the o
     # Option 2: Use End-to-End method: read the data, apply the checks, write data to valid and quarantine tables, and write metrics to the metrics table
     # Output and quarantine data will be written in streaming and summary metrics will be written for each micro-batch
     engine.apply_checks_and_save_in_table(
-        checks=checks,
+        checks=checks,  # or provide checks_location and run_config_name to auto-load from checks storage
         input_config=input_config,
         output_config=output_config,
         quarantine_config=quarantine_config,