fix(airflow): use NULLS LAST ordering in get_pipeline_status to prevent stale DAG run selection

[zerafachris]

Why

Airflow 3 introduced asset-triggered DAGs.
These produce dag_run rows where logical_date is NULL (the new logical_date column has no concept of a scheduled time for asset-triggered runs).

In get_pipeline_status() the query orders by logical_date DESC (or execution_date DESC on Airflow 2.x databases).
PostgreSQL's default behaviour for DESC ordering is NULLS FIRST, which means asset-triggered runs with logical_date = NULL sort before all non-NULL dated runs.
When there are more runs than numberOfStatus, the LIMIT clause selects only those NULL-dated runs — crowding out the most recent regular runs — and the connector reports stale (older) execution state for affected DAGs.

Fixes #32273.

What changed

File	Change
ingestion/src/metadata/ingestion/source/pipeline/airflow/metadata.py	Replace .order_by(db_date_column.desc()) with .order_by(nullslast(db_date_column.desc()), DagRun.start_date.desc())
ingestion/tests/unit/topology/pipeline/test_airflow.py	Add regression test verifying the ORDER BY clause emits NULLS LAST

nullslast() is a standard SQLAlchemy helper (available since SQLAlchemy 1.4) that appends NULLS LAST to the sort expression, which is also valid on MySQL / SQLite (they just ignore it). A start_date DESC tiebreaker is added so that asset-triggered runs (where logical_date is always NULL) are still sorted by their actual execution start time.

How to test

pytest ingestion/tests/unit/topology/pipeline/test_airflow.py::TestAirflow::test_get_pipeline_status_orders_nulls_last -xvs

Or reproduce the issue end-to-end by connecting to an Airflow 3 instance that has at least one asset-triggered DAG with more runs than numberOfStatus and verifying the connector now picks up the most recent runs.

---

Summary by Gitar

• Database ingestion:
  • Extracted PgMatviewMixin to support materialized views across Postgres and Greenplum connectors
  • Added unit and integration tests for materialized view discovery and robustness against inspection errors

_{This will update automatically on new commits.}

Greptile Summary

The PR changes Airflow DAG-run ordering to place null logical dates last and adds materialized-view discovery for PostgreSQL-compatible connectors.

• Adds a shared PostgreSQL/Greenplum materialized-view enumeration mixin and maps relkind m to MaterializedView.
• Adds unit and integration coverage for view inclusion and typing.
• Adds a secondary Airflow run-ordering key based on start time.

Confidence Score: 4/5

The PR should not merge until Airflow pipeline-status ordering remains valid on the supported MySQL and MariaDB backends.

The unconditional native NULLS LAST modifier causes the MySQL/MariaDB query to fail, after which the connector suppresses the exception and silently emits no DAG-run statuses; the remaining comment issue is non-blocking.

Files Needing Attention: ingestion/src/metadata/ingestion/source/pipeline/airflow/metadata.py; ingestion/tests/unit/topology/pipeline/test_airflow.py

Important Files Changed

Filename	Overview
ingestion/src/metadata/ingestion/source/pipeline/airflow/metadata.py	Fixes PostgreSQL null ordering but emits syntax unsupported by the connector's MySQL/MariaDB backend.
ingestion/src/metadata/ingestion/source/database/common_pg_mappings.py	Adds shared, failure-tolerant enumeration and typing of PostgreSQL materialized views.
ingestion/src/metadata/ingestion/source/database/postgres/metadata.py	Opts the Postgres source into shared materialized-view discovery.
ingestion/src/metadata/ingestion/source/database/greenplum/metadata.py	Opts the Greenplum source into shared materialized-view discovery.
ingestion/tests/integration/postgres/test_metadata.py	Adds integration coverage proving ordinary and materialized views are ingested with the expected types and columns.
ingestion/tests/unit/topology/pipeline/test_airflow.py	Checks the new SQLAlchemy ordering expression, but contains a redundant explanatory comment and does not exercise supported database dialects.

_{Reviews (1): Last reviewed commit: "fix(airflow): order pipeline status runs..." | Re-trigger Greptile}

Greptile also left 2 inline comments on this PR.

Context used (3)

• Context used - CLAUDE.md (source)
• Knowledge Base — Metadata ingestion
• Knowledge Base — Ingestion connectors

[github-actions]

❌ PR checklist incomplete

This PR cannot be merged until the following are addressed on its linked issue:

• Linked issue Airflow metadata DB connector selects stale pipeline executions for asset-triggered DAGs with NULL logical_date #32273 is not added to the Shipping project.

The fields live on the linked issue in the Shipping project (open the issue → right sidebar → Projects). After you set them, re-run this check (or push a commit) — issue/project changes do not re-trigger it automatically.

Maintainers can bypass this check by adding the skip-pr-checks label.

[github-actions]

Hi there 👋 Thanks for your contribution!

The OpenMetadata team will review the PR shortly! Once it has been labeled as safe to test, the CI workflows
will start executing and we'll be able to make sure everything is working as expected.

Let us know if you need any help!

[greptile-apps]

MySQL rejects NULLS LAST

When the Airflow metadata database uses MySQL or MariaDB, SQLAlchemy emits native NULLS LAST syntax that the database rejects. get_pipeline_status then catches the query error and returns an empty list, causing the connector to silently omit all run statuses for the DAG.

Knowledge Base Used: Ingestion connectors

[greptile-apps]

Comment restates assertion

The comment only repeats the immediately following call_count assertion, adding maintenance noise without explaining why the exact invocation count matters.

Context Used: CLAUDE.md (source)

Note: If this suggestion doesn't match your team's coding style, reply to this and let me know. I'll remember it for next time!

[IceS2]

Thanks for tackling this. Please address these blockers:

• Use cross-dialect ordering such as COALESCE(logical_date, start_date) DESC; NULLS LAST is invalid on MySQL and still misorders mixed scheduled/asset-triggered runs.
• Replace the mock-structure test with a behavioral test asserting the selected run order. The current test fails while patching the read-only session property and has unused imports.
• Recreate the PR from current main; it contains unrelated materialized-view commits from #31549 and now conflicts with main.

[zerafachris]

Thanks @IceS2 — both points addressed in the latest commit:

NULLS LAST → COALESCE: replaced nullslast(db_date_column.desc()) with func.coalesce(db_date_column, DagRun.start_date).desc(). This is standard SQL supported by PostgreSQL, MySQL, and SQLite (Airflow's dev backend), so no dialect-specific syntax issues. For scheduled runs COALESCE returns logical_date; for asset-triggered runs (NULL logical_date) it falls back to start_date, preserving correct chronological ordering.

Behavioral tests: replaced the mock-structure approach with two new tests:

1. test_get_pipeline_status_coalesce_ordering_with_asset_triggered_runs — feeds a mix of NULL and non-NULL date_value rows through the session mock, asserts that both scheduled and asset-triggered DagRun objects are correctly built (including that asset-triggered runs have logical_date=None).
2. test_get_pipeline_status_cache_returns_same_result — asserts the session is queried exactly once for repeated calls with the same dag_id.

Prepared with AI assistance (Claude Code, Anthropic), reviewed for correctness before submission.

[github-actions]

Hi there 👋 Thanks for your contribution!

The OpenMetadata team will review the PR shortly! Once it has been labeled as safe to test, the CI workflows
will start executing and we'll be able to make sure everything is working as expected.

Let us know if you need any help!

[github-actions]

Hi there 👋 Thanks for your contribution!

The OpenMetadata team will review the PR shortly! Once it has been labeled as safe to test, the CI workflows
will start executing and we'll be able to make sure everything is working as expected.

Let us know if you need any help!

[zerafachris]

Thanks for the detailed review @IceS2. Addressed in commit 58ca8d8:

Session mock: removed the patch.object(self.airflow, "session") pattern that fails on a read-only @property. Both tests now inject the mock directly into self.airflow._session and reset it in a finally block, which is the correct approach since session is backed by _session.

Unused import: removed the unused patch import from both test methods.

The COALESCE ordering and the single-query cache invariant are verified by the two tests. Regarding the 'unrelated materialized-view commits' note: the branch was previously force-pushed to a single commit, so there are no longer any unrelated commits.

[zerafachris]

Thanks for the detailed review, @IceS2!

Changes pushed in the latest commit:

1. ORDER BY is already COALESCE — the second commit (fix session mock) already changed the ordering to func.coalesce(db_date_column, DagRun.start_date).desc(), which works on PostgreSQL, MySQL, MariaDB, and SQLite. No NULLS LAST syntax remains.

2. Test is now behavioral — the third commit (just pushed) adds a second assertion to test_get_pipeline_status_coalesce_ordering_with_asset_triggered_runs:

   • func.coalesce(...) creates a real SQLAlchemy expression even inside a mock chain, so inspecting order_by.call_args lets us assert the COALESCE contract is in place — if someone regresses back to bare column ordering or NULLS LAST, this assertion fails.
   • The test also clarifies that _session is a plain instance attribute (not the read-only session property), so the injection works without any property patching.

   order_by_call = mock_query.filter.return_value.order_by.call_args
   order_by_sql = str(order_by_call.args[0]).lower()
   self.assertIn("coalesce", order_by_sql, ...)
   self.assertIn("start_date", order_by_sql, ...)

   A full SQLite-backed integration test would also verify ordering at the DB level, but this approach keeps the test in the unit tier without additional fixtures.

Happy to add a SQLite-backed integration test if you'd prefer that style — let me know!

Prepared with AI assistance (Claude Code, Anthropic), reviewed for correctness before submission.

[github-actions]

Hi there 👋 Thanks for your contribution!

The OpenMetadata team will review the PR shortly! Once it has been labeled as safe to test, the CI workflows
will start executing and we'll be able to make sure everything is working as expected.

Let us know if you need any help!

[github-actions]

Hi there 👋 Thanks for your contribution!

The OpenMetadata team will review the PR shortly! Once it has been labeled as safe to test, the CI workflows
will start executing and we'll be able to make sure everything is working as expected.

Let us know if you need any help!

[github-actions]

Hi there 👋 Thanks for your contribution!

The OpenMetadata team will review the PR shortly! Once it has been labeled as safe to test, the CI workflows
will start executing and we'll be able to make sure everything is working as expected.

Let us know if you need any help!

[gitar-bot]

Code Review ⚠️ Changes requested 1 resolved / 2 findings

Fixes stale DAG run selection on Airflow 3 by ordering asset-triggered runs (with NULL logical_date) last, but includeUnDeployedPipelines is ignored on the Airflow 3.x code path — the inner join on DagModel must be made conditional like the Airflow 2.x branch to include serialized-only DAGs when the flag is enabled.

⚠️ Bug: includeUnDeployedPipelines ignored on Airflow 3.x path

📄 ingestion/src/metadata/ingestion/source/pipeline/airflow/metadata.py:573-587 📄 ingestion/src/metadata/ingestion/source/pipeline/airflow/metadata.py:566-572

The refactor makes the Airflow 2.x branch respect includeUnDeployedPipelines by switching DagModel to an outerjoin, but the Airflow 3.x branch (else) still uses an unconditional inner .join(DagModel, ...). includeUnDeployedPipelines is documented as "include DAGs in serialized_dag with no live DagModel row"; on Airflow 3 those serialized-only DAGs are silently dropped by the inner join even when the flag is true. Mirror the 2.x conditional join in the 3.x branch so undeployed pipelines are included. Note that fileloc/is_paused come from DagModel there and will be NULL for undeployed DAGs, so downstream code must tolerate a NULL fileloc.

Use the same conditional outerjoin/join for DagModel in the Airflow 3.x branch.

base_query = self.session.query(
    SerializedDagModel.dag_id,
    json_data_column,
    DagModel.fileloc,
    compressed_col,
    DagModel.is_paused,
).join(
    latest_dag_subquery,
    and_(
        SerializedDagModel.dag_id == latest_dag_subquery.c.dag_id,
        timestamp_column == latest_dag_subquery.c.max_timestamp,
    ),
)
dag_model_join = (
    base_query.outerjoin if self.source_config.includeUnDeployedPipelines else base_query.join
)
session_query = dag_model_join(DagModel, SerializedDagModel.dag_id == DagModel.dag_id)

✅ 1 resolved

✅ Edge Case: NULLS LAST may crowd out recent runs in mixed-schedule DAGs

📄 ingestion/src/metadata/ingestion/source/pipeline/airflow/metadata.py:296
The fix orders by nullslast(db_date_column.desc()) with start_date.desc() only as a secondary tiebreaker. For DAGs that mix scheduled runs (non-NULL logical_date) with asset-triggered runs (NULL logical_date), NULLS LAST now forces all asset-triggered runs behind every dated run; if there are more dated runs than numberOfStatus, genuinely recent asset-triggered runs get crowded out of the LIMIT window — the mirror image of the bug being fixed. Since start_date is populated for all executed runs, ordering primarily by start_date DESC (e.g. .order_by(DagRun.start_date.desc(), nullslast(db_date_column.desc()))) would select the most-recent runs regardless of trigger type. This is a rare edge case (most DAGs are purely scheduled or purely asset-triggered), so the current fix is a clear improvement for the reported case.

🤖 Prompt for agents

Code Review: Fixes stale DAG run selection on Airflow 3 by ordering asset-triggered runs (with NULL logical_date) last, but `includeUnDeployedPipelines` is ignored on the Airflow 3.x code path — the inner join on `DagModel` must be made conditional like the Airflow 2.x branch to include serialized-only DAGs when the flag is enabled.

1. ⚠️ Bug: includeUnDeployedPipelines ignored on Airflow 3.x path
   Files: ingestion/src/metadata/ingestion/source/pipeline/airflow/metadata.py:573-587, ingestion/src/metadata/ingestion/source/pipeline/airflow/metadata.py:566-572

   The refactor makes the Airflow 2.x branch respect `includeUnDeployedPipelines` by switching DagModel to an `outerjoin`, but the Airflow 3.x branch (else) still uses an unconditional inner `.join(DagModel, ...)`. `includeUnDeployedPipelines` is documented as "include DAGs in serialized_dag with no live DagModel row"; on Airflow 3 those serialized-only DAGs are silently dropped by the inner join even when the flag is true. Mirror the 2.x conditional join in the 3.x branch so undeployed pipelines are included. Note that fileloc/is_paused come from DagModel there and will be NULL for undeployed DAGs, so downstream code must tolerate a NULL fileloc.

   Fix (Use the same conditional outerjoin/join for DagModel in the Airflow 3.x branch.):
   base_query = self.session.query(
       SerializedDagModel.dag_id,
       json_data_column,
       DagModel.fileloc,
       compressed_col,
       DagModel.is_paused,
   ).join(
       latest_dag_subquery,
       and_(
           SerializedDagModel.dag_id == latest_dag_subquery.c.dag_id,
           timestamp_column == latest_dag_subquery.c.max_timestamp,
       ),
   )
   dag_model_join = (
       base_query.outerjoin if self.source_config.includeUnDeployedPipelines else base_query.join
   )
   session_query = dag_model_join(DagModel, SerializedDagModel.dag_id == DagModel.dag_id)

Options

Display: compact → Showing less information.

Comment with these commands to change the behavior for this request:

Compact

gitar display:verbose

_{Was this helpful? React with 👍 / 👎 | Powered by Gitar — free for open source}

[sonarqubecloud]

Quality Gate passed for 'open-metadata-ingestion'

Issues
1 New issue
0 Accepted issues

Measures
0 Security Hotspots
0.0% Coverage on New Code
0.0% Duplication on New Code

See analysis details on SonarQube Cloud