fix: view_names issue when incremental enabled (#23858)

ingestion/src/metadata/ingestion/source/database/bigquery/metadata.py

@@ -418,14 +418,6 @@ class BigquerySource(LifeCycleQueryMixin, CommonDbSourceService, MultiDBSource):
         logic on how to handle table types, e.g., material views,...
         """
 
-        if self.incremental.enabled:
-            view_names = [
-                view_name
-                for view_name in view_names
-                if view_name
-                in self.incremental_table_processor.get_not_deleted(schema_name)
-            ]
-
         table_name_and_types = []
         for table_type, query in {
             TableType.View: BIGQUERY_GET_VIEW_NAMES,

ingestion/tests/unit/topology/database/test_bigquery.py

@@ -542,6 +542,80 @@ class BigqueryUnitTest(TestCase):
                 for either in self.bq_source.yield_table((table[0], table[1]))
             ]
 
+    def test_query_view_names_and_types_with_incremental(self):
+        """
+        Test query_view_names_and_types with incremental extraction enabled
+        """
+        self.bq_source.context.get().__dict__["database"] = MOCK_DB_NAME
+        self.bq_source.context.get().__dict__[
+            "database_schema"
+        ] = MOCK_DATABASE_SCHEMA.name.root
+
+        mock_view_results = [("view1",), ("view2",), ("view3",)]
+        mock_materialized_view_results = [("mat_view1",), ("mat_view2",)]
+
+        def mock_execute(query):
+            if "INFORMATION_SCHEMA.VIEWS" in query:
+                return mock_view_results
+            elif "INFORMATION_SCHEMA.MATERIALIZED_VIEWS" in query:
+                return mock_materialized_view_results
+            return []
+
+        self.bq_source.engine.execute = mock_execute
+
+        mock_processor = Mock()
+        mock_processor.get_not_deleted.return_value = ["view1", "mat_view1", "view3"]
+        self.bq_source.incremental_table_processor = mock_processor
+        self.bq_source.incremental.enabled = True
+
+        result = list(
+            self.bq_source.query_view_names_and_types(MOCK_DATABASE_SCHEMA.name.root)
+        )
+
+        assert len(result) == 3
+        assert result[0].name == "view1"
+        assert result[0].type_ == TableType.View
+        assert result[1].name == "view3"
+        assert result[1].type_ == TableType.View
+        assert result[2].name == "mat_view1"
+        assert result[2].type_ == TableType.MaterializedView
+
+        mock_processor.get_not_deleted.assert_called()
+
+    def test_query_view_names_and_types_without_incremental(self):
+        """
+        Test query_view_names_and_types without incremental extraction
+        """
+        self.bq_source.context.get().__dict__["database"] = MOCK_DB_NAME
+        self.bq_source.context.get().__dict__[
+            "database_schema"
+        ] = MOCK_DATABASE_SCHEMA.name.root
+
+        mock_view_results = [("view1",), ("view2",)]
+        mock_materialized_view_results = [("mat_view1",)]
+
+        def mock_execute(query):
+            if "INFORMATION_SCHEMA.VIEWS" in query:
+                return mock_view_results
+            elif "INFORMATION_SCHEMA.MATERIALIZED_VIEWS" in query:
+                return mock_materialized_view_results
+            return []
+
+        self.bq_source.engine.execute = mock_execute
+        self.bq_source.incremental.enabled = False
+
+        result = list(
+            self.bq_source.query_view_names_and_types(MOCK_DATABASE_SCHEMA.name.root)
+        )
+
+        assert len(result) == 3
+        assert result[0].name == "view1"
+        assert result[0].type_ == TableType.View
+        assert result[1].name == "view2"
+        assert result[1].type_ == TableType.View
+        assert result[2].name == "mat_view1"
+        assert result[2].type_ == TableType.MaterializedView
+
 
 class BigqueryLineageSourceTest(TestCase):
     """