refactor(metadata-ingestion): add cachetool to operation_config

Author: cccs-cat001

metadata-ingestion/setup.py

@@ -251,6 +251,7 @@
     # Iceberg Python SDK
     # Kept at 0.4.0 due to higher versions requiring pydantic>2, as soon as we are fine with it, bump this dependency
     "pyiceberg>=0.4.0",
+    *cachetools_lib,
 }
 
 mssql_common = {
@@ -404,13 +405,14 @@
     # UnsupportedProductError
     # https://www.elastic.co/guide/en/elasticsearch/client/python-api/current/release-notes.html#rn-7-14-0
     # https://github.com/elastic/elasticsearch-py/issues/1639#issuecomment-883587433
-    "elasticsearch": {"elasticsearch==7.13.4"},
+    "elasticsearch": {"elasticsearch==7.13.4", *cachetools_lib},
     "cassandra": {
         "cassandra-driver>=3.28.0",
         # We were seeing an error like this `numpy.dtype size changed, may indicate binary incompatibility. Expected 96 from C header, got 88 from PyObject`
         # with numpy 2.0. This likely indicates a mismatch between scikit-learn and numpy versions.
         # https://stackoverflow.com/questions/40845304/runtimewarning-numpy-dtype-size-changed-may-indicate-binary-incompatibility
         "numpy<2",
+        cachetools_lib,
     },
     "feast": {
         "feast>=0.34.0,<1",
@@ -422,7 +424,7 @@
         "numpy<2",
     },
     "grafana": {"requests"},
-    "glue": aws_common,
+    "glue": aws_common | cachetools_lib,
     # hdbcli is supported officially by SAP, sqlalchemy-hana is built on top but not officially supported
     "hana": sql_common
     | {
@@ -479,11 +481,11 @@
     | classification_lib
     | {"db-dtypes"}  # Pandas extension data types
     | cachetools_lib,
-    "s3": {*s3_base, *data_lake_profiling},
+    "s3": {*s3_base, *data_lake_profiling, *cachetools_lib},
     "gcs": {*s3_base, *data_lake_profiling},
-    "abs": {*abs_base, *data_lake_profiling},
+    "abs": {*abs_base, *data_lake_profiling, *cachetools_lib},
     "sagemaker": aws_common,
-    "salesforce": {"simple-salesforce"},
+    "salesforce": {"simple-salesforce", *cachetools_lib},
     "snowflake": snowflake_common | usage_common | sqlglot_lib,
     "snowflake-summary": snowflake_common | usage_common | sqlglot_lib,
     "snowflake-queries": snowflake_common | usage_common | sqlglot_lib,

metadata-ingestion/src/datahub/ingestion/source/iceberg/iceberg_common.py

@@ -6,7 +6,6 @@
 from pydantic import Field, validator
 from pyiceberg.catalog import Catalog, load_catalog
 from sortedcontainers import SortedList
-from functools import lru_cache
 from datahub.configuration.common import AllowDenyPattern, ConfigModel
 from datahub.configuration.source_common import DatasetSourceConfigMixin
 from datahub.ingestion.source.state.stale_entity_removal_handler import (
@@ -125,7 +124,6 @@ def validate_catalog_size(cls, value):
 
         return value
 
-    @lru_cache()
     def is_profiling_enabled(self) -> bool:
         return self.profiling.enabled and is_profiling_enabled(
             self.profiling.operation_config

metadata-ingestion/src/datahub/ingestion/source/sql/sql_config.py

@@ -2,7 +2,7 @@
 from abc import abstractmethod
 from typing import Any, Dict, Optional
 
-import cachetools
+
 import cachetools.keys
 import pydantic
 from pydantic import Field
@@ -29,7 +29,7 @@
     StatefulIngestionConfigBase,
 )
 from datahub.ingestion.source_config.operation_config import is_profiling_enabled
-from datahub.utilities.cachetools_keys import self_methodkey
+
 
 logger: logging.Logger = logging.getLogger(__name__)
 
@@ -117,14 +117,7 @@ class SQLCommonConfig(
     profiling: GEProfilingConfig = GEProfilingConfig()
     # Custom Stateful Ingestion settings
     stateful_ingestion: Optional[StatefulStaleMetadataRemovalConfig] = None
-
-    # TRICKY: The operation_config is time-dependent. Because we don't want to change
-    # whether or not we're running profiling mid-ingestion, we cache the result of this method.
-    # TODO: This decorator should be moved to the is_profiling_enabled(operation_config) method.
-    @cachetools.cached(
-        cache=cachetools.LRUCache(maxsize=1),
-        key=self_methodkey,
-    )
+    
     def is_profiling_enabled(self) -> bool:
         return self.profiling.enabled and is_profiling_enabled(
             self.profiling.operation_config

metadata-ingestion/src/datahub/ingestion/source_config/operation_config.py

@@ -1,10 +1,10 @@
 import datetime
 import logging
 from typing import Any, Dict, Optional
-
+import cachetools
 import pydantic
 from pydantic.fields import Field
-
+from datahub.utilities.cachetools_keys import self_methodkey
 from datahub.configuration.common import ConfigModel
 
 logger = logging.getLogger(__name__)
@@ -61,25 +61,28 @@ def validate_profile_date_of_month(cls, v: Optional[int]) -> Optional[int]:
             )
         return profile_date_of_month
 
-
+@cachetools.cached(
+        cache=cachetools.LRUCache(maxsize=1),
+        key=self_methodkey,
+    )
 def is_profiling_enabled(operation_config: OperationConfig) -> bool:
     if operation_config.lower_freq_profile_enabled is False:
         return True
-    logger.debug("Lower freq profiling setting is enabled.")
+    logger.info("Lower freq profiling setting is enabled.")
     today = datetime.date.today()
     if (
         operation_config.profile_day_of_week is not None
         and operation_config.profile_day_of_week != today.weekday()
     ):
-        logger.debug(
+        logger.info(
             "Profiling won't be done because weekday does not match config profile_day_of_week.",
         )
         return False
     if (
         operation_config.profile_date_of_month is not None
         and operation_config.profile_date_of_month != today.day
     ):
-        logger.debug(
+        logger.info(
             "Profiling won't be done because date of month does not match config profile_date_of_month.",
         )
         return False