fix(datalake): cold-storage Iceberg dir detection and tighten JSON classification

Author: mohitjeswani01

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

@@ -127,16 +127,21 @@ def get_table_names(
         """
         bucket = self._client.get_bucket(bucket_name)
         iceberg_tables: Dict[str, Tuple[int, str, int | None]] = {}  # noqa: UP006
+        cold_iceberg_dirs: Set[str] = set()  # noqa: UP006
 
         for blob in bucket.list_blobs(prefix=prefix):
-            if skip_cold_storage and self._should_skip_gcs_cold_storage(blob):
+            is_cold = skip_cold_storage and self._should_skip_gcs_cold_storage(blob)
+            if is_cold:
                 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))
                 continue
             self._update_iceberg_entry(iceberg_tables, blob.name, blob.size)
 
-        iceberg_dirs = set(iceberg_tables.keys())
+        iceberg_dirs = set(iceberg_tables.keys()) | cold_iceberg_dirs
         for _, metadata_blob_path, size in iceberg_tables.values():
             yield metadata_blob_path, size
 

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

@@ -88,20 +88,25 @@ def get_table_names(
             kwargs["Prefix"] = prefix if prefix.endswith("/") else f"{prefix}/"
 
         iceberg_tables: Dict[str, Tuple[int, str, int | None]] = {}  # noqa: UP006
+        cold_iceberg_dirs: Set[str] = set()  # noqa: UP006
 
         for key in list_s3_objects(self._client, **kwargs):
             key_name = key["Key"]
             size = key.get("Size")
-            if skip_cold_storage and self._should_skip_s3_cold_storage(key):
+            is_cold = skip_cold_storage and self._should_skip_s3_cold_storage(key)
+            if is_cold:
                 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))
                 continue
             self._update_iceberg_entry(iceberg_tables, key_name, size)
 
-        iceberg_dirs = set(iceberg_tables.keys())
+        iceberg_dirs = set(iceberg_tables.keys()) | cold_iceberg_dirs
         for _, metadata_key, size in iceberg_tables.values():
             yield metadata_key, size
 

ingestion/src/metadata/readers/dataframe/json.py

@@ -126,9 +126,6 @@ def _read_json_object(
             and (
                 data.get("$schema") is not None  # JSON Schema files
                 or data.get("format-version") is not None  # Apache Iceberg table metadata
-                or (  # Delta Lake / Iceberg schema structure
-                    isinstance(data.get("schema"), dict) and isinstance(data.get("schema", {}).get("fields"), list)
-                )
             )
             else None
         )
@@ -158,9 +155,7 @@ def _is_json_lines(file_obj) -> bool:
                 return False
             if obj.get("$schema") is not None:
                 return False
-            if obj.get("format-version") is not None:
-                return False
-            return not (isinstance(obj.get("schema"), dict) and isinstance(obj.get("schema", {}).get("fields"), list))
+            return obj.get("format-version") is None
 
     def _read_json_smart(
         self,