From 880c7430e06066721d25dac61f3945ddcb645f75 Mon Sep 17 00:00:00 2001 From: "neha.marne" Date: Thu, 2 Jul 2026 16:24:45 +0530 Subject: [PATCH 1/3] feat(sql): add debug logging for dropped tables and views in SQLAlchemy and Teradata sources --- .../ingestion/source/sql/sql_common.py | 4 + .../datahub/ingestion/source/sql/teradata.py | 8 ++ .../tests/unit/test_teradata_source.py | 134 ++++++++++++++++++ 3 files changed, 146 insertions(+) diff --git a/metadata-ingestion/src/datahub/ingestion/source/sql/sql_common.py b/metadata-ingestion/src/datahub/ingestion/source/sql/sql_common.py index e3418b725bb0..0c9e8d6e52b8 100644 --- a/metadata-ingestion/src/datahub/ingestion/source/sql/sql_common.py +++ b/metadata-ingestion/src/datahub/ingestion/source/sql/sql_common.py @@ -739,7 +739,9 @@ def loop_tables( continue self.report.report_entity_scanned(dataset_name, ent_type="table") + logger.debug(f"Scanning table: {dataset_name}") if not sql_config.table_pattern.allowed(dataset_name): + logger.debug(f"Dropped table by table_pattern: {dataset_name}") self.report.report_dropped(dataset_name) continue @@ -1142,8 +1144,10 @@ def loop_views( schema=schema, entity=view, inspector=inspector ) self.report.report_entity_scanned(dataset_name, ent_type="view") + logger.debug(f"Scanning view: {dataset_name}") if not sql_config.view_pattern.allowed(dataset_name): + logger.debug(f"Dropped view by view_pattern: {dataset_name}") self.report.report_dropped(dataset_name) continue diff --git a/metadata-ingestion/src/datahub/ingestion/source/sql/teradata.py b/metadata-ingestion/src/datahub/ingestion/source/sql/teradata.py index 682fa9cfa681..ab9d78695971 100644 --- a/metadata-ingestion/src/datahub/ingestion/source/sql/teradata.py +++ b/metadata-ingestion/src/datahub/ingestion/source/sql/teradata.py @@ -2362,8 +2362,12 @@ def process_single_view( self.report.report_entity_scanned( dataset_name, ent_type="view" ) + logger.debug(f"Scanning view: {dataset_name}") if not sql_config.view_pattern.allowed(dataset_name): + logger.debug( + f"Dropped view by view_pattern: {dataset_name}" + ) with self.report.atomic(): self.report.report_dropped(dataset_name) return results @@ -2612,8 +2616,12 @@ def _process_views_single_threaded( ) self.report.report_entity_scanned(dataset_name, ent_type="view") + logger.debug(f"Scanning view: {dataset_name}") if not sql_config.view_pattern.allowed(dataset_name): + logger.debug( + f"Dropped view by view_pattern: {dataset_name}" + ) self.report.report_dropped(dataset_name) continue diff --git a/metadata-ingestion/tests/unit/test_teradata_source.py b/metadata-ingestion/tests/unit/test_teradata_source.py index 768bbd406edd..5c987a6145e6 100644 --- a/metadata-ingestion/tests/unit/test_teradata_source.py +++ b/metadata-ingestion/tests/unit/test_teradata_source.py @@ -4859,6 +4859,140 @@ def test_warn_view_error_emits_correct_category_message( assert expected_fragment in source.report.warnings[0].message +class TestFilterDropDebugLogging: + """Pattern-filtered tables and views are dropped (not processed) and recorded. + + A fully-filtered Teradata run emits only database containers with everything else + in ``report.filtered``. These tests assert the behavior at each drop point: the + denied object is recorded in ``report.filtered`` and its ``_process_*`` handler is + never invoked. Covers both view processing paths (single- and multi-threaded) and + the base ``loop_tables`` path that Teradata tables use. + """ + + def test_single_threaded_view_drop_reports_and_skips_processing(self): + """A view denied by view_pattern is recorded in report.filtered and never + handed to _process_view (single-threaded path).""" + source = _create_source_patched( + {"max_workers": 1, "view_pattern": {"deny": [".*"]}} + ) + process_view = MagicMock() + + with ( + patch.object(source, "get_metadata_engine", return_value=MagicMock()), + patch.object( + source, + "_retry_connect", + side_effect=_mock_retry_connect(_make_mock_conn()), + ), + patch.object(source, "_retry_execute"), + patch( + "datahub.ingestion.source.sql.teradata.inspect", + return_value=_make_mock_inspector("testdb"), + ), + patch.object(source, "_process_view", process_view), + ): + wus = list( + source._process_views_single_threaded( + ["denied_view"], "testdb", source.config + ) + ) + + assert wus == [] + process_view.assert_not_called() + assert "testdb.denied_view" in list(source.report.filtered) + + def test_multi_threaded_view_drop_reports_and_skips_processing(self): + """A view denied by view_pattern is recorded in report.filtered and never + handed to _process_view (multi-threaded path).""" + source = _create_source_patched( + {"max_workers": 2, "view_pattern": {"deny": [".*"]}} + ) + process_view = MagicMock() + + with ( + patch.object( + source, "_get_or_create_pooled_engine", return_value=MagicMock() + ), + patch.object( + source, + "_retry_connect", + side_effect=_mock_retry_connect(_make_mock_conn()), + ), + patch.object(source, "_retry_execute"), + patch( + "datahub.ingestion.source.sql.teradata.inspect", + return_value=_make_mock_inspector("testdb"), + ), + patch.object(source, "_process_view", process_view), + ): + wus = list( + source._loop_views_with_connection_pool( + ["denied_view"], "testdb", source.config + ) + ) + + assert wus == [] + process_view.assert_not_called() + assert "testdb.denied_view" in list(source.report.filtered) + + def test_allowed_view_is_processed_and_not_dropped(self): + """A view that passes view_pattern is handed to _process_view and is not + recorded in report.filtered.""" + source = _create_source_patched( + {"max_workers": 1, "view_pattern": {"allow": ["testdb.allowed_view"]}} + ) + process_view = MagicMock(return_value=iter([])) + + with ( + patch.object(source, "get_metadata_engine", return_value=MagicMock()), + patch.object( + source, + "_retry_connect", + side_effect=_mock_retry_connect(_make_mock_conn()), + ), + patch.object(source, "_retry_execute"), + patch( + "datahub.ingestion.source.sql.teradata.inspect", + return_value=_make_mock_inspector("testdb"), + ), + patch.object(source, "_process_view", process_view), + ): + list( + source._process_views_single_threaded( + ["allowed_view"], "testdb", source.config + ) + ) + + process_view.assert_called_once() + assert "testdb.allowed_view" not in list(source.report.filtered) + + def test_table_drop_reports_and_skips_processing(self): + """A table denied by table_pattern is recorded in report.filtered and never + handed to _process_table. Teradata tables flow through the base loop_tables + via cached_loop_tables, which sources table names from _tables_cache.""" + source = _create_source_patched({"table_pattern": {"deny": [".*"]}}) + source._tables_cache["testdb"] = [ + TeradataTable( + database="testdb", + name="denied_table", + description=None, + object_type="Table", + create_timestamp=datetime(2024, 1, 1), + last_alter_name=None, + last_alter_timestamp=None, + request_text=None, + ) + ] + process_table = MagicMock(return_value=iter([])) + + with patch.object(source, "_process_table", process_table): + wus = list(source.cached_loop_tables(MagicMock(), "testdb", source.config)) + + assert wus == [] + process_table.assert_not_called() + assert "testdb.denied_table" in list(source.report.filtered) + + # --------------------------------------------------------------------------- # Helpers for _fetch_lineage_entries_chunked tests # --------------------------------------------------------------------------- From 7b03ece16d6f6e86509ecdcc4d6e06cd97457c1a Mon Sep 17 00:00:00 2001 From: "neha.marne" Date: Thu, 2 Jul 2026 16:46:34 +0530 Subject: [PATCH 2/3] refactor(tests): enhance logging and assertions for view processing in Teradata source tests --- .../tests/unit/test_teradata_source.py | 34 +++++++++++++++---- 1 file changed, 27 insertions(+), 7 deletions(-) diff --git a/metadata-ingestion/tests/unit/test_teradata_source.py b/metadata-ingestion/tests/unit/test_teradata_source.py index 5c987a6145e6..437b9510c8f5 100644 --- a/metadata-ingestion/tests/unit/test_teradata_source.py +++ b/metadata-ingestion/tests/unit/test_teradata_source.py @@ -4869,9 +4869,17 @@ class TestFilterDropDebugLogging: the base ``loop_tables`` path that Teradata tables use. """ - def test_single_threaded_view_drop_reports_and_skips_processing(self): + def test_single_threaded_view_drop_reports_and_skips_processing( + self, caplog: pytest.LogCaptureFixture + ): """A view denied by view_pattern is recorded in report.filtered and never - handed to _process_view (single-threaded path).""" + handed to _process_view (single-threaded path). + + This is also the one site where we pin the PR's debug logging: assert the + drop is emitted at DEBUG level. The other drop-site tests only assert the + drop *behavior* (report.filtered + skipped processing) to avoid brittle, + duplicated log-string assertions across every path. + """ source = _create_source_patched( {"max_workers": 1, "view_pattern": {"deny": [".*"]}} ) @@ -4890,6 +4898,9 @@ def test_single_threaded_view_drop_reports_and_skips_processing(self): return_value=_make_mock_inspector("testdb"), ), patch.object(source, "_process_view", process_view), + caplog.at_level( + logging.DEBUG, logger="datahub.ingestion.source.sql.teradata" + ), ): wus = list( source._process_views_single_threaded( @@ -4900,6 +4911,7 @@ def test_single_threaded_view_drop_reports_and_skips_processing(self): assert wus == [] process_view.assert_not_called() assert "testdb.denied_view" in list(source.report.filtered) + assert "Dropped view by view_pattern: testdb.denied_view" in caplog.text def test_multi_threaded_view_drop_reports_and_skips_processing(self): """A view denied by view_pattern is recorded in report.filtered and never @@ -4935,11 +4947,16 @@ def test_multi_threaded_view_drop_reports_and_skips_processing(self): process_view.assert_not_called() assert "testdb.denied_view" in list(source.report.filtered) - def test_allowed_view_is_processed_and_not_dropped(self): - """A view that passes view_pattern is handed to _process_view and is not - recorded in report.filtered.""" + def test_mixed_allow_deny_drops_denied_and_processes_allowed(self): + """With a denied and an allowed view in the same call, the denied one is + dropped and the allowed one is still processed. + + Passing both in a single call guards against an early-return regression: + the drop must ``continue`` to the next view, not abort the whole loop. This + also serves as the positive control (an allowed view reaches _process_view + and is not recorded in report.filtered).""" source = _create_source_patched( - {"max_workers": 1, "view_pattern": {"allow": ["testdb.allowed_view"]}} + {"max_workers": 1, "view_pattern": {"deny": ["testdb.denied_view"]}} ) process_view = MagicMock(return_value=iter([])) @@ -4959,11 +4976,14 @@ def test_allowed_view_is_processed_and_not_dropped(self): ): list( source._process_views_single_threaded( - ["allowed_view"], "testdb", source.config + ["denied_view", "allowed_view"], "testdb", source.config ) ) + # Denied view is dropped; the loop continues and processes the allowed view. process_view.assert_called_once() + assert process_view.call_args.kwargs["view"] == "allowed_view" + assert "testdb.denied_view" in list(source.report.filtered) assert "testdb.allowed_view" not in list(source.report.filtered) def test_table_drop_reports_and_skips_processing(self): From 686ce373c171e04510bfc541e1b791d85b2ab503 Mon Sep 17 00:00:00 2001 From: "neha.marne" Date: Thu, 2 Jul 2026 17:04:36 +0530 Subject: [PATCH 3/3] refactor(tests): add type hints to test methods in Teradata source tests --- metadata-ingestion/tests/unit/test_teradata_source.py | 8 ++++---- 1 file changed, 4 insertions(+), 4 deletions(-) diff --git a/metadata-ingestion/tests/unit/test_teradata_source.py b/metadata-ingestion/tests/unit/test_teradata_source.py index 437b9510c8f5..440bb392c747 100644 --- a/metadata-ingestion/tests/unit/test_teradata_source.py +++ b/metadata-ingestion/tests/unit/test_teradata_source.py @@ -4871,7 +4871,7 @@ class TestFilterDropDebugLogging: def test_single_threaded_view_drop_reports_and_skips_processing( self, caplog: pytest.LogCaptureFixture - ): + ) -> None: """A view denied by view_pattern is recorded in report.filtered and never handed to _process_view (single-threaded path). @@ -4913,7 +4913,7 @@ def test_single_threaded_view_drop_reports_and_skips_processing( assert "testdb.denied_view" in list(source.report.filtered) assert "Dropped view by view_pattern: testdb.denied_view" in caplog.text - def test_multi_threaded_view_drop_reports_and_skips_processing(self): + def test_multi_threaded_view_drop_reports_and_skips_processing(self) -> None: """A view denied by view_pattern is recorded in report.filtered and never handed to _process_view (multi-threaded path).""" source = _create_source_patched( @@ -4947,7 +4947,7 @@ def test_multi_threaded_view_drop_reports_and_skips_processing(self): process_view.assert_not_called() assert "testdb.denied_view" in list(source.report.filtered) - def test_mixed_allow_deny_drops_denied_and_processes_allowed(self): + def test_mixed_allow_deny_drops_denied_and_processes_allowed(self) -> None: """With a denied and an allowed view in the same call, the denied one is dropped and the allowed one is still processed. @@ -4986,7 +4986,7 @@ def test_mixed_allow_deny_drops_denied_and_processes_allowed(self): assert "testdb.denied_view" in list(source.report.filtered) assert "testdb.allowed_view" not in list(source.report.filtered) - def test_table_drop_reports_and_skips_processing(self): + def test_table_drop_reports_and_skips_processing(self) -> None: """A table denied by table_pattern is recorded in report.filtered and never handed to _process_table. Teradata tables flow through the base loop_tables via cached_loop_tables, which sources table names from _tables_cache."""