Filter schemas by current catalog in Glue (#26908)

Author: ayush-shah

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

@@ -185,9 +185,12 @@ def get_database_schema_names(self) -> Iterable[str]:
         """
         return schema names
         """
+        database_name = self.context.get().database
         for page in self._get_glue_database_and_schemas() or []:
             for schema in page.DatabaseList:
                 try:
+                    if schema.CatalogId != database_name:
+                        continue
                     schema_fqn = fqn.build(
                         self.metadata,
                         entity_type=DatabaseSchema,

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

@@ -31,8 +31,13 @@
     OpenMetadataWorkflowConfig,
 )
 from metadata.generated.schema.type.entityReference import EntityReference
+from metadata.generated.schema.type.filterPattern import FilterPattern
 from metadata.ingestion.source.database.glue.metadata import GlueSource
-from metadata.ingestion.source.database.glue.models import DatabasePage, TablePage
+from metadata.ingestion.source.database.glue.models import (
+    DatabasePage,
+    GlueSchema,
+    TablePage,
+)
 
 mock_file_path = (
     Path(__file__).parent.parent.parent / "resources/datasets/glue_db_dataset.json"
@@ -188,6 +193,29 @@ def test_database_schema_names(self):
             self.glue_source.get_database_schema_names()
         )
 
+    def test_database_schema_names_filters_other_catalogs_before_schema_filter(self):
+        self.glue_source.source_config.schemaFilterPattern = FilterPattern(
+            includes=["default"]
+        )
+        self.glue_source._get_glue_database_and_schemas = lambda: [
+            DatabasePage(
+                DatabaseList=[
+                    GlueSchema(
+                        CatalogId=MOCK_DATABASE.name.root,
+                        Name="default",
+                        Description="current catalog schema",
+                    ),
+                    GlueSchema(
+                        CatalogId="different-catalog",
+                        Name="default",
+                        Description="other catalog schema",
+                    ),
+                ]
+            )
+        ]
+
+        assert ["default"] == list(self.glue_source.get_database_schema_names())
+
     @patch("metadata.ingestion.source.database.glue.metadata.fqn")
     def test_table_names(self, fqn):
         fqn.build = mock_fqn_build