fix(datalake): restore cold storage skip logging and remove duplicate tests per review

Author: mohitjeswani01

ingestion/src/metadata/ingestion/source/database/datalake/clients/gcs.py

@@ -124,6 +124,9 @@ def _discover_iceberg_dirs(
 
         for blob in bucket.list_blobs(prefix=prefix):
             if skip_cold_storage and self._should_skip_gcs_cold_storage(blob):
+                logger.debug(
+                    f"Skipping cold storage object: {blob.name} (storage_class: {getattr(blob, 'storage_class', None)})"
+                )
                 match = self._ICEBERG_METADATA_RE.match(blob.name)
                 if match:
                     cold_iceberg_dirs.add(match.group(1))

ingestion/src/metadata/ingestion/source/database/datalake/clients/s3.py

@@ -83,6 +83,11 @@ def _discover_iceberg_dirs(
             key_name = key["Key"]
             size = key.get("Size")
             if skip_cold_storage and self._should_skip_s3_cold_storage(key):
+                logger.debug(
+                    f"Skipping cold storage object: {key_name} "
+                    f"(StorageClass: {key.get('StorageClass', 'STANDARD')}, "
+                    f"ArchiveStatus: {key.get('ArchiveStatus', '')})"
+                )
                 match = self._ICEBERG_METADATA_RE.match(key_name)
                 if match:
                     cold_iceberg_dirs.add(match.group(1))

ingestion/tests/unit/source/database/test_iceberg_discovery.py

@@ -76,22 +76,6 @@ def test_gcs_iceberg_table_detected(self):
         assert name == "warehouse/orders/metadata/v2.metadata.json"
         assert size == 600
 
-    def test_gcs_iceberg_yields_one_table_per_directory(self):
-        blobs = [
-            _make_blob("warehouse/orders/metadata/v1.metadata.json", size=500),
-            _make_blob("warehouse/orders/metadata/v2.metadata.json", size=600),
-            _make_blob("warehouse/orders/data/00000-0-abc.parquet", size=8192),
-            _make_blob("warehouse/orders/data/00001-0-def.parquet", size=9216),
-        ]
-        client = _make_gcs_client(blobs)
-
-        results = list(client.get_table_names("my-bucket", prefix="warehouse"))
-
-        assert len(results) == 1
-        name, size = results[0]
-        assert name == "warehouse/orders/metadata/v2.metadata.json"
-        assert size == 600
-
     def test_gcs_multiple_iceberg_tables(self):
         blobs = [
             _make_blob("warehouse/orders/metadata/v1.metadata.json", size=400),
@@ -195,28 +179,6 @@ def test_s3_iceberg_table_detected(self):
         assert name == "warehouse/orders/metadata/v2.metadata.json"
         assert size == 600
 
-    def test_s3_iceberg_yields_one_table_per_directory(self):
-        keys = [
-            "warehouse/orders/metadata/v1.metadata.json",
-            "warehouse/orders/metadata/v2.metadata.json",
-            "warehouse/orders/data/00000-0-abc.parquet",
-        ]
-        client = self._make_s3_client(
-            keys,
-            sizes={"warehouse/orders/metadata/v2.metadata.json": 600},
-        )
-
-        with patch(
-            "metadata.ingestion.source.database.datalake.clients.s3.list_s3_objects",
-            return_value=self._s3_objects,
-        ):
-            results = list(client.get_table_names("my-bucket", prefix="warehouse"))
-
-        assert len(results) == 1
-        name, size = results[0]
-        assert name == "warehouse/orders/metadata/v2.metadata.json"
-        assert size == 600
-
     def test_s3_fallback_for_non_iceberg(self):
         keys = [
             "data/orders.csv",