fix(ingest/redshift): Making Redshift source more verbose (#8109)

Author: treff7es

metadata-ingestion/src/datahub/ingestion/source/redshift/redshift.py

@@ -432,7 +432,7 @@ def process_schemas(self, connection, database):
         for schema in RedshiftDataDictionary.get_schemas(
             conn=connection, database=database
         ):
-            logger.info(f"Schema: {database}.{schema.name}")
+            logger.info(f"Processing schema: {database}.{schema.name}")
             if not self.config.schema_pattern.allowed(schema.name):
                 self.report.report_dropped(f"{database}.{schema.name}")
                 continue
@@ -478,11 +478,11 @@ def process_schema(
             )
 
             if self.config.include_tables:
-                logger.info("process tables")
-                if not self.db_tables[schema.database]:
-                    return
-
-                if schema.name in self.db_tables[schema.database]:
+                logger.info(f"Process tables in schema {database}.{schema.name}")
+                if (
+                    self.db_tables[schema.database]
+                    and schema.name in self.db_tables[schema.database]
+                ):
                     for table in self.db_tables[schema.database][schema.name]:
                         table.columns = schema_columns[schema.name].get(table.name, [])
                         yield from self._process_table(table, database=database)
@@ -492,10 +492,22 @@ def process_schema(
                             )
                             + 1
                         )
+                        logger.debug(
+                            f"Table processed: {schema.database}.{schema.name}.{table.name}"
+                        )
+                else:
+                    logger.info(
+                        f"No tables in cache for {schema.database}.{schema.name}, skipping"
+                    )
+            else:
+                logger.info("Table processing disabled, skipping")
 
             if self.config.include_views:
-                logger.info("process views")
-                if schema.name in self.db_views[schema.database]:
+                logger.info(f"Process views in schema {schema.database}.{schema.name}")
+                if (
+                    self.db_views[schema.database]
+                    and schema.name in self.db_views[schema.database]
+                ):
                     for view in self.db_views[schema.database][schema.name]:
                         view.columns = schema_columns[schema.name].get(view.name, [])
                         yield from self._process_view(
@@ -508,6 +520,15 @@ def process_schema(
                             )
                             + 1
                         )
+                        logger.debug(
+                            f"Table processed: {schema.database}.{schema.name}.{view.name}"
+                        )
+                else:
+                    logger.info(
+                        f"No views in cache for {schema.database}.{schema.name}, skipping"
+                    )
+            else:
+                logger.info("View processing disabled, skipping")
 
             self.report.metadata_extraction_sec[report_key] = round(
                 timer.elapsed_seconds(), 2
@@ -752,20 +773,47 @@ def cache_tables_and_views(self, connection, database):
         tables, views = RedshiftDataDictionary.get_tables_and_views(conn=connection)
         for schema in tables:
             if self.config.schema_pattern.allowed(f"{database}.{schema}"):
+                logging.debug(
+                    f"Not caching tables for schema {database}.{schema} which is not allowed by schema_pattern"
+                )
                 self.db_tables[database][schema] = []
                 for table in tables[schema]:
                     if self.config.table_pattern.allowed(
                         f"{database}.{schema}.{table.name}"
                     ):
                         self.db_tables[database][schema].append(table)
+                        self.report.table_cached[f"{database}.{schema}"] = (
+                            self.report.table_cached.get(f"{database}.{schema}", 0) + 1
+                        )
+                    else:
+                        logging.debug(
+                            f"Table {database}.{schema}.{table.name} is filtered by table_pattern"
+                        )
+                        self.report.table_filtered[f"{database}.{schema}"] = (
+                            self.report.table_filtered.get(f"{database}.{schema}", 0)
+                            + 1
+                        )
         for schema in views:
+            logging.debug(
+                f"Not caching views for schema {database}.{schema} which is not allowed by schema_pattern"
+            )
             if self.config.schema_pattern.allowed(f"{database}.{schema}"):
                 self.db_views[database][schema] = []
                 for view in views[schema]:
                     if self.config.view_pattern.allowed(
                         f"{database}.{schema}.{view.name}"
                     ):
                         self.db_views[database][schema].append(view)
+                        self.report.view_cached[f"{database}.{schema}"] = (
+                            self.report.view_cached.get(f"{database}.{schema}", 0) + 1
+                        )
+                    else:
+                        logging.debug(
+                            f"View {database}.{schema}.{table.name} is filtered by view_pattern"
+                        )
+                        self.report.view_filtered[f"{database}.{schema}"] = (
+                            self.report.view_filtered.get(f"{database}.{schema}", 0) + 1
+                        )
 
     def get_all_tables(
         self,

metadata-ingestion/src/datahub/ingestion/source/redshift/redshift_schema.py

@@ -170,7 +170,7 @@ def get_tables_and_views(
         cur = RedshiftDataDictionary.get_query_result(conn, RedshiftQuery.list_tables)
         field_names = [i[0] for i in cur.description]
         db_tables = cur.fetchall()
-
+        logger.info(f"Fetched {len(db_tables)} tables/views from Redshift")
         for table in db_tables:
             schema = table[field_names.index("schema")]
             if table[field_names.index("tabletype")] not in [
@@ -241,6 +241,10 @@ def get_tables_and_views(
                         comment=table[field_names.index("table_description")],
                     )
                 )
+        for schema_key, schema_tables in tables.items():
+            logger.info(f"In schema: {schema_key} discovered {len(tables)} tables")
+        for schema_key, schema_views in views.items():
+            logger.info(f"In schema: {schema_key} discovered {len(views)} views")
 
         return tables, views
 

metadata-ingestion/src/datahub/ingestion/source/redshift/report.py

@@ -14,7 +14,11 @@ class RedshiftReport(ProfilingSqlReport):
     usage_extraction_sec: Dict[str, float] = field(default_factory=TopKDict)
     lineage_extraction_sec: Dict[str, float] = field(default_factory=TopKDict)
     table_processed: TopKDict[str, int] = field(default_factory=TopKDict)
+    table_filtered: TopKDict[str, int] = field(default_factory=TopKDict)
+    view_filtered: TopKDict[str, int] = field(default_factory=TopKDict)
     view_processed: TopKDict[str, int] = field(default_factory=TopKDict)
+    table_cached: TopKDict[str, int] = field(default_factory=TopKDict)
+    view_cached: TopKDict[str, int] = field(default_factory=TopKDict)
     metadata_extraction_sec: TopKDict[str, float] = field(default_factory=TopKDict)
     operational_metadata_extraction_sec: TopKDict[str, float] = field(
         default_factory=TopKDict