Block a load when the schema drifts¶

A nightly load copies customers from Postgres into Snowflake. One morning a column was renamed upstream and the load either failed halfway or, worse, succeeded with nulls. This Dag compares the two schemas before the load and asks the model which differences would break it. Airflow branches on the answer: compatible schemas run the load, anything else notifies the team. The model only reports; no migration runs.

What this demonstrates¶

Detect schema drift: LLMSchemaCompareOperator – @task.llm_schema_compare reads both schemas through Airflow connections and returns a structured comparison with a compatible flag.
Branching with @task.branch.

Run it¶

Install the provider with the SQL extra and the providers for your databases:

pip install "apache-airflow-providers-common-ai[openai,sql]" \
    apache-airflow-providers-postgres apache-airflow-providers-snowflake

Create database connections postgres_source and snowflake_target that both contain a customers table.

Trigger the Dag:

airflow dags test example_llm_schema_compare_conditional

The check_before_etl XCom holds the full comparison. Exactly one of run_etl and notify_team runs.

The Dag¶

airflow/providers/common/ai/example_dags/example_llm_schema_compare.py[source]

@dag(tags=["example"])
def example_llm_schema_compare_conditional():
    @task.llm_schema_compare(
        llm_conn_id="pydanticai_default",
        db_conn_ids=["postgres_source", "snowflake_target"],
        table_names=["customers"],
        context_strategy="full",
    )
    def check_before_etl():
        return (
            "Compare schemas and flag any mismatches that would break data loading. "
            "No migrations allowed — report only."
        )

    @task.branch
    def decide(comparison_result):
        if comparison_result["compatible"]:
            return "run_etl"
        return "notify_team"

    comparison = check_before_etl()
    decision = decide(comparison)

    @task(task_id="run_etl")
    def run_etl():
        return "ETL completed"

    @task(task_id="notify_team")
    def notify_team():
        return "Schema drift detected — team notified"

    decision >> [run_etl(), notify_team()]

Adapting it¶

Point db_conn_ids and table_names at your own source and target. The operator also compares against files on object storage, for example a Parquet landing zone; see Detect schema drift: LLMSchemaCompareOperator.
Replace the run_etl placeholder with your load task and notify_team with a Slack or email notifier.
Add an Approval gates for LLM operators step before run_etl when a compatible-but-changed schema should still get a human’s confirmation.
Set schedule to match the load and give the Dag the same start_date so the two stay aligned.