From e7c556980d764764952f7c7e762852158de3d225 Mon Sep 17 00:00:00 2001 From: TeddyCr Date: Thu, 6 Aug 2026 13:04:15 -0700 Subject: [PATCH 1/2] Fixes #29824: ingest dbt test results that have a null message PR #26812 added a guard in add_dbt_test_result() that skipped any run_results entry whose `message` was null, to drop the compiled-only test nodes that `dbt run` writes with status="success". The guard never looked at `status`, so it also dropped genuine executed data tests: dbt only populates `message` on failure/warn, so a passing test arrives as status="pass", message=null. That is why dbt test results stopped appearing after the 1.13.0 upgrade. Move the discrimination into `is_compiled_only_result()` in dbt_utils and key it on both `status` and `message`, so only status="success" entries with no message are treated as compiled-only. `status` is safe to dereference: it is a required, non-nullable Enum on every run-results model the shared parser can produce (v1-v6 plus the two cloud variants), and all connector variants parse through DbtServiceSource.get_dbt_objects. `failures` is not usable as the discriminator because it is stripped by REQUIRED_RESULTS_KEYS before parsing. Supersedes the same narrowing proposed in PR #29828 by ayush-shah. Co-Authored-By: Claude Opus 5 (1M context) --- .../source/database/dbt/dbt_utils.py | 15 ++ .../ingestion/source/database/dbt/metadata.py | 8 +- ingestion/tests/unit/test_dbt.py | 157 ++++++++++++++++++ 3 files changed, 175 insertions(+), 5 deletions(-) diff --git a/ingestion/src/metadata/ingestion/source/database/dbt/dbt_utils.py b/ingestion/src/metadata/ingestion/source/database/dbt/dbt_utils.py index 9e8e09f8b42d..c48c3c001402 100644 --- a/ingestion/src/metadata/ingestion/source/database/dbt/dbt_utils.py +++ b/ingestion/src/metadata/ingestion/source/database/dbt/dbt_utils.py @@ -29,6 +29,7 @@ NONE_KEYWORDS_LIST, CompiledQueriesEnum, DbtCommonEnum, + DbtTestSuccessEnum, RawQueriesEnum, ) from metadata.ingestion.source.database.dbt.models import SnapshotNodeLocation, UpstreamNode @@ -839,6 +840,20 @@ def get_dbt_test_primary_table_fqn(dbt_test) -> Optional[str]: # noqa: UP045 return primary_table_fqn +def is_compiled_only_result(dbt_test_result) -> bool: + """ + Tell a compiled-only run_results entry apart from an executed test result. + + ``dbt run`` and ``dbt docs generate`` list test nodes in run_results.json with + the *node* status ``success`` and ``message=null`` even though no test SQL ran. + An executed test instead carries a *test* status (``pass``/``fail``/``warn``/ + ``error``), and dbt leaves ``message`` null for passing tests, so ``message`` + alone cannot be used as the discriminator (issue #29824). ``failures`` would be + the other signal but it is dropped by ``REQUIRED_RESULTS_KEYS`` before parsing. + """ + return not dbt_test_result.message and dbt_test_result.status.value == DbtTestSuccessEnum.SUCCESS.value + + def generate_entity_link(dbt_test): """ Method returns entity link for dbt test cases. diff --git a/ingestion/src/metadata/ingestion/source/database/dbt/metadata.py b/ingestion/src/metadata/ingestion/source/database/dbt/metadata.py index 3ae74e055baf..8345a312f407 100644 --- a/ingestion/src/metadata/ingestion/source/database/dbt/metadata.py +++ b/ingestion/src/metadata/ingestion/source/database/dbt/metadata.py @@ -126,6 +126,7 @@ get_dbt_test_primary_table_fqn, get_manifest_column_name, get_snapshot_effective_schema_and_database, + is_compiled_only_result, map_dbt_metric_type, order_metrics_by_dependency, validate_custom_property_value, @@ -1980,12 +1981,9 @@ def add_dbt_test_result(self, dbt_test: dict): # noqa: C901 logger.debug(f"DBT Test Case Results not found for node: {manifest_node.name}") return - # Skip compiled-only entries: `dbt run` includes test nodes in - # run_results.json with status="success" but message=null since - # no test SQL was executed. Real results always have a message. - if not dbt_test_result.message: + if is_compiled_only_result(dbt_test_result): logger.debug( - "Skipping compiled-only test result for '%s' (message is null).", + "Skipping compiled-only test result for '%s' (status is success and message is null).", manifest_node.name, ) return diff --git a/ingestion/tests/unit/test_dbt.py b/ingestion/tests/unit/test_dbt.py index e170b711058d..e05c1b5080dc 100644 --- a/ingestion/tests/unit/test_dbt.py +++ b/ingestion/tests/unit/test_dbt.py @@ -5,6 +5,7 @@ import json import uuid from copy import deepcopy +from datetime import datetime from pathlib import Path from types import SimpleNamespace from unittest import TestCase @@ -25,6 +26,7 @@ from metadata.generated.schema.metadataIngestion.workflow import ( OpenMetadataWorkflowConfig, ) +from metadata.generated.schema.tests.basic import TestCaseStatus from metadata.generated.schema.type import entityReference from metadata.generated.schema.type.entityReference import EntityReference from metadata.generated.schema.type.entityReferenceList import EntityReferenceList @@ -66,9 +68,13 @@ from metadata.ingestion.source.database.dbt.models import DbtFiles, DbtObjects, UpstreamNode from metadata.utils.logger import ingestion_logger, set_loggers_level from metadata.utils.tag_utils import get_tag_labels +from metadata.utils.time_utils import datetime_to_timestamp logger = ingestion_logger() +DBT_TEST_UNIQUE_ID = "test.jaffle_shop.not_null_orders_order_id.cf6c17daed" +DBT_TEST_TABLE_FQN = "snowflake.jaffle_shop.public.orders" + mock_dbt_config = { "source": { "type": "dbt", @@ -3731,6 +3737,157 @@ def test_nanosecond_timestamp_is_parsed(self): source.metadata.add_test_case_results.assert_called_once() +def _run_result_payload(status, message, completed_at, unique_id=DBT_TEST_UNIQUE_ID, failures=None): + """ + Build a run_results.json payload shaped like a real dbt artifact, including + the ``failures`` key that dbt emits but OpenMetadata strips before parsing. + """ + return { + "metadata": { + "dbt_schema_version": "https://schemas.getdbt.com/dbt/run-results/v4.json", + "dbt_version": "1.11.0", + "generated_at": completed_at, + "invocation_id": str(uuid.uuid4()), + "env": {}, + }, + "results": [ + { + "status": status, + "timing": [ + { + "name": "compile", + "started_at": completed_at, + "completed_at": completed_at, + }, + { + "name": "execute", + "started_at": completed_at, + "completed_at": completed_at, + }, + ], + "thread_id": "Thread-1", + "execution_time": 0.42, + "adapter_response": {}, + "message": message, + "failures": failures, + "unique_id": unique_id, + } + ], + "elapsed_time": 1.5, + "args": {"which": "test"}, + } + + +def _parse_run_results_like_production(payload): + """ + Run a raw run_results.json payload through the exact pre-processing the + connector applies (``remove_run_result_non_required_keys``) before parsing, + so tests see the same attributes production code sees. + """ + from metadata.ingestion.source.database.dbt.dbt_service import DbtServiceSource + + DbtServiceSource.remove_run_result_non_required_keys(MagicMock(spec=DbtServiceSource), run_results=[payload]) + return parse_run_results(payload) + + +class TestAddDbtTestResultNullMessage: + """ + Regression coverage for issue #29824. + + dbt only fills ``message`` on failure/warn for many adapters, so a genuine + executed data test is reported as ``status="pass", message=null``. The + compiled-only guard added by #26812 keyed off ``message`` alone and so + dropped those real results, which is why dbt test results stopped showing + up after the 1.13.0 upgrade. + """ + + @staticmethod + def _make_source(): + source = MagicMock(spec=DbtSource) + source.add_dbt_test_result = DbtSource.add_dbt_test_result.__get__(source, DbtSource) + source.metadata = MagicMock() + source.status = MagicMock() + source.context = MagicMock() + source.context.get.return_value = SimpleNamespace(run_results_generate_time=None) + return source + + @staticmethod + def _sent_call(source): + """ + add_dbt_test_result swallows every exception into status.failed(), so an + unhandled error would otherwise look identical to a deliberate skip. + """ + assert source.status.failed.call_args_list == [], source.status.failed.call_args_list + calls = source.metadata.add_test_case_results.call_args_list + assert len(calls) == 1, "expected exactly one test case result sent to OpenMetadata" + return calls[0].kwargs + + @staticmethod + def _make_dbt_test(run_result): + return { + DbtCommonEnum.MANIFEST_NODE.value: SimpleNamespace( + name="not_null_orders_order_id", + column_name="order_id", + test_metadata=SimpleNamespace( + name="not_null", + kwargs={"column_name": "order_id", "model": "ref('orders')"}, + ), + ), + DbtCommonEnum.RESULTS.value: run_result, + DbtCommonEnum.UPSTREAM.value: [DBT_TEST_TABLE_FQN], + DbtCommonEnum.UPSTREAM_BY_NAME.value: {"orders": DBT_TEST_TABLE_FQN}, + } + + def _ingest(self, status, message): + payload = _run_result_payload(status=status, message=message, completed_at="2026-07-24T09:00:00.000000Z") + run_result = _parse_run_results_like_production(payload).results[0] + source = self._make_source() + source.add_dbt_test_result(self._make_dbt_test(run_result)) + return source + + def test_failures_key_is_stripped_before_parsing(self): + """ + `failures` is not in REQUIRED_RESULTS_KEYS, so it cannot be used to tell + an executed test from a compiled-only stub: `status` is the only signal left. + """ + payload = _run_result_payload( + status="pass", message=None, completed_at="2026-07-24T09:00:00.000000Z", failures=0 + ) + run_result = _parse_run_results_like_production(payload).results[0] + + assert getattr(run_result, "failures", None) is None + assert run_result.status.value == "pass" + + def test_passing_test_with_null_message_is_ingested(self): + kwargs = self._sent_call(self._ingest(status="pass", message=None)) + + test_case_result = kwargs["test_results"] + assert kwargs["test_case_fqn"] == f"{DBT_TEST_TABLE_FQN}.order_id.not_null_orders_order_id" + assert test_case_result.testCaseStatus == TestCaseStatus.Success + assert test_case_result.result is None + assert [value.value for value in test_case_result.testResultValue] == ["1"] + assert test_case_result.timestamp.root == datetime_to_timestamp( + datetime(2026, 7, 24, 9, 0, 0), milliseconds=True + ) + + def test_compiled_only_success_with_null_message_is_still_skipped(self): + source = self._ingest(status="success", message=None) + + source.metadata.add_test_case_results.assert_not_called() + assert source.status.failed.call_args_list == [] + + def test_failing_test_with_null_message_is_ingested(self): + test_case_result = self._sent_call(self._ingest(status="fail", message=None))["test_results"] + + assert test_case_result.testCaseStatus == TestCaseStatus.Failed + assert [value.value for value in test_case_result.testResultValue] == ["0"] + + def test_warning_test_with_null_message_is_ingested(self): + test_case_result = self._sent_call(self._ingest(status="warn", message=None))["test_results"] + + assert test_case_result.testCaseStatus == TestCaseStatus.Aborted + + class TestRemoveManifestNonRequiredKeys(TestCase): """ Tests for DbtServiceSource.remove_manifest_non_required_keys. From 1764c9fef55207b3aa17c8c02f53f1296e3016d9 Mon Sep 17 00:00:00 2001 From: TeddyCr Date: Thu, 6 Aug 2026 13:07:47 -0700 Subject: [PATCH 2/2] Fixes #29824: do not let compile-only dbt stubs shadow executed results _get_latest_result() de-duplicated a unique_id across multiple run_results.json files purely by `execute.completed_at`. A project that keeps both a `dbt test` artifact and a later `dbt docs generate` artifact therefore selected the docs run's compile-only stub (status="success", message=null), and add_dbt_test_result() then dropped it as compiled-only - so the real pass/fail result never reached OpenMetadata even with the message guard fixed. Reported on 1.13.1 / dbt 1.11 by ziggekatten. Filter compile-only entries out of the candidate set before ranking by timestamp, reusing the same is_compiled_only_result() predicate as the ingestion guard so the two can never disagree. Sharing the predicate is load-bearing rather than incidental: if the selector's notion of "compile-only" diverged from the ingester's, the selector could hand over a result the ingester then drops, which is exactly this bug. When every match is a stub the previous behaviour is kept, since there is nothing better to pick. The no-parseable-timestamp fallback now returns the first *executed* candidate rather than the first candidate overall; that behaviour change is pinned by test_executed_result_wins_when_no_timestamp_is_usable. Co-Authored-By: Claude Opus 5 (1M context) --- .../ingestion/source/database/dbt/metadata.py | 6 + ingestion/tests/unit/test_dbt.py | 217 +++++++++++++----- 2 files changed, 170 insertions(+), 53 deletions(-) diff --git a/ingestion/src/metadata/ingestion/source/database/dbt/metadata.py b/ingestion/src/metadata/ingestion/source/database/dbt/metadata.py index 8345a312f407..1968f2342b1b 100644 --- a/ingestion/src/metadata/ingestion/source/database/dbt/metadata.py +++ b/ingestion/src/metadata/ingestion/source/database/dbt/metadata.py @@ -631,12 +631,18 @@ def _get_latest_result(dbt_objects: DbtObjects, key: str): the same unique_id may appear in more than one file. Return the result with the most recent ``execute`` completed_at timestamp so that OpenMetadata always reflects the latest test state. + + Compile-only entries are only considered when nothing else matched: a + ``dbt docs generate`` artifact produced after a ``dbt test`` one carries + the newer timestamp, and preferring it would discard the real result + before add_dbt_test_result() could ingest it (issue #29824). """ matches = [ item for run_result in dbt_objects.dbt_run_results for item in run_result.results if item.unique_id == key ] if not matches: return None + matches = [item for item in matches if not is_compiled_only_result(item)] or matches if len(matches) == 1: return matches[0] diff --git a/ingestion/tests/unit/test_dbt.py b/ingestion/tests/unit/test_dbt.py index e05c1b5080dc..f9e498981c87 100644 --- a/ingestion/tests/unit/test_dbt.py +++ b/ingestion/tests/unit/test_dbt.py @@ -75,6 +75,68 @@ DBT_TEST_UNIQUE_ID = "test.jaffle_shop.not_null_orders_order_id.cf6c17daed" DBT_TEST_TABLE_FQN = "snowflake.jaffle_shop.public.orders" + +def _run_result_payload(status, message, completed_at, unique_id=DBT_TEST_UNIQUE_ID, failures=None): + """ + Build a run_results.json payload shaped like a real dbt artifact, including + the ``failures`` key that dbt emits but OpenMetadata strips before parsing. + + ``completed_at=None`` drops the timing block, which is how a result with no + usable ``execute`` timestamp reaches the timestamp fallback path. + """ + timing = ( + [ + { + "name": "compile", + "started_at": completed_at, + "completed_at": completed_at, + }, + { + "name": "execute", + "started_at": completed_at, + "completed_at": completed_at, + }, + ] + if completed_at + else [] + ) + return { + "metadata": { + "dbt_schema_version": "https://schemas.getdbt.com/dbt/run-results/v4.json", + "dbt_version": "1.11.0", + "generated_at": completed_at or "2026-07-24T09:00:00.000000Z", + "invocation_id": str(uuid.uuid4()), + "env": {}, + }, + "results": [ + { + "status": status, + "timing": timing, + "thread_id": "Thread-1", + "execution_time": 0.42, + "adapter_response": {}, + "message": message, + "failures": failures, + "unique_id": unique_id, + } + ], + "elapsed_time": 1.5, + "args": {"which": "test"}, + } + + +def _parse_run_results_like_production(payload): + """ + Run a raw run_results.json payload through the exact pre-processing the + connector applies (``remove_run_result_non_required_keys``) before parsing, + so tests see the same attributes production code sees. + """ + from metadata.ingestion.source.database.dbt.dbt_service import DbtServiceSource + + DbtServiceSource.remove_run_result_non_required_keys(MagicMock(spec=DbtServiceSource), run_results=[payload]) + return parse_run_results(payload) + + mock_dbt_config = { "source": { "type": "dbt", @@ -3130,6 +3192,108 @@ def test_nanosecond_timestamp_parsed_across_files(self): self.assertIs(got, new_result) +class TestGetLatestResultPrefersExecutedResults: + """ + Regression coverage for the second half of issue #29824. + + A project that keeps both a ``dbt test`` artifact and a later + ``dbt docs generate`` artifact has the same test unique_id in both files. + Picking purely by ``execute.completed_at`` hands back the compile-only stub + from the docs run, and the real pass/fail result is discarded before + add_dbt_test_result() ever sees it. + """ + + @staticmethod + def _dbt_objects(*results_specs): + run_results = [ + _parse_run_results_like_production( + _run_result_payload(status=status, message=message, completed_at=completed_at) + ) + for status, message, completed_at in results_specs + ] + return DbtObjects(dbt_manifest=None, dbt_run_results=run_results) + + def test_executed_result_wins_over_later_compile_only_stub(self): + dbt_objects = self._dbt_objects( + ("pass", None, "2026-07-24T07:00:00.000000Z"), + ("success", None, "2026-07-24T09:00:00.000000Z"), + ) + + selected = DbtSource._get_latest_result(dbt_objects, DBT_TEST_UNIQUE_ID) + + assert selected.status.value == "pass" + + def test_executed_result_wins_when_stub_is_listed_first(self): + dbt_objects = self._dbt_objects( + ("success", None, "2026-07-24T09:00:00.000000Z"), + ("fail", "Got 3 results, configured to fail if != 0", "2026-07-24T07:00:00.000000Z"), + ) + + selected = DbtSource._get_latest_result(dbt_objects, DBT_TEST_UNIQUE_ID) + + assert selected.status.value == "fail" + + def test_latest_still_wins_among_executed_results(self): + dbt_objects = self._dbt_objects( + ("pass", None, "2026-07-24T07:00:00.000000Z"), + ("fail", "Got 3 results, configured to fail if != 0", "2026-07-24T09:00:00.000000Z"), + ) + + selected = DbtSource._get_latest_result(dbt_objects, DBT_TEST_UNIQUE_ID) + + assert selected.status.value == "fail" + + def test_latest_stub_is_still_returned_when_nothing_was_executed(self): + dbt_objects = self._dbt_objects( + ("success", None, "2026-07-24T07:00:00.000000Z"), + ("success", None, "2026-07-24T09:00:00.000000Z"), + ) + + selected = DbtSource._get_latest_result(dbt_objects, DBT_TEST_UNIQUE_ID) + + assert selected.timing[1].completed_at == datetime.fromisoformat("2026-07-24T09:00:00+00:00") + + def test_no_match_returns_none(self): + dbt_objects = self._dbt_objects(("pass", None, "2026-07-24T07:00:00.000000Z")) + + assert DbtSource._get_latest_result(dbt_objects, "test.jaffle_shop.does_not_exist") is None + + def test_executed_result_wins_when_no_timestamp_is_usable(self): + """ + With no ``execute`` timing to rank by, selection falls back to the first + candidate. That fallback must run over executed results only, otherwise a + stub listed first still wins. + """ + dbt_objects = self._dbt_objects( + ("success", None, None), + ("fail", "Got 3 results, configured to fail if != 0", None), + ) + + selected = DbtSource._get_latest_result(dbt_objects, DBT_TEST_UNIQUE_ID) + + assert selected.status.value == "fail" + + def test_result_survives_end_to_end_through_add_dbt_test_result(self): + """ + The user-visible symptom: with a later docs-generate artifact present, + no test case result reaches OpenMetadata at all. + """ + dbt_objects = self._dbt_objects( + ("pass", None, "2026-07-24T07:00:00.000000Z"), + ("success", None, "2026-07-24T09:00:00.000000Z"), + ) + selected = DbtSource._get_latest_result(dbt_objects, DBT_TEST_UNIQUE_ID) + + source = TestAddDbtTestResultNullMessage._make_source() + source.add_dbt_test_result(TestAddDbtTestResultNullMessage._make_dbt_test(selected)) + + kwargs = TestAddDbtTestResultNullMessage._sent_call(source) + assert kwargs["test_results"].testCaseStatus == TestCaseStatus.Success + assert kwargs["test_results"].timestamp.root == datetime_to_timestamp( + datetime(2026, 7, 24, 7, 0, 0), milliseconds=True + ) + + class TestGetBlobsGroupedByDir(TestCase): """ Test cases for get_blobs_grouped_by_dir to verify streaming support, @@ -3737,59 +3901,6 @@ def test_nanosecond_timestamp_is_parsed(self): source.metadata.add_test_case_results.assert_called_once() -def _run_result_payload(status, message, completed_at, unique_id=DBT_TEST_UNIQUE_ID, failures=None): - """ - Build a run_results.json payload shaped like a real dbt artifact, including - the ``failures`` key that dbt emits but OpenMetadata strips before parsing. - """ - return { - "metadata": { - "dbt_schema_version": "https://schemas.getdbt.com/dbt/run-results/v4.json", - "dbt_version": "1.11.0", - "generated_at": completed_at, - "invocation_id": str(uuid.uuid4()), - "env": {}, - }, - "results": [ - { - "status": status, - "timing": [ - { - "name": "compile", - "started_at": completed_at, - "completed_at": completed_at, - }, - { - "name": "execute", - "started_at": completed_at, - "completed_at": completed_at, - }, - ], - "thread_id": "Thread-1", - "execution_time": 0.42, - "adapter_response": {}, - "message": message, - "failures": failures, - "unique_id": unique_id, - } - ], - "elapsed_time": 1.5, - "args": {"which": "test"}, - } - - -def _parse_run_results_like_production(payload): - """ - Run a raw run_results.json payload through the exact pre-processing the - connector applies (``remove_run_result_non_required_keys``) before parsing, - so tests see the same attributes production code sees. - """ - from metadata.ingestion.source.database.dbt.dbt_service import DbtServiceSource - - DbtServiceSource.remove_run_result_non_required_keys(MagicMock(spec=DbtServiceSource), run_results=[payload]) - return parse_run_results(payload) - - class TestAddDbtTestResultNullMessage: """ Regression coverage for issue #29824.