Repository navigation
beam: add zero-copy Delta Lake to Iceberg metadata translation - #41234
user-Param wants to merge 1 commit into
Conversation
Add apache_beam.io.lakehouse package with translate_delta_to_iceberg() function that synthesizes Iceberg v2 metadata from Delta Lake transaction logs without copying underlying Parquet data files. Features: - Parses Delta _delta_log/*.json partition definitions into Iceberg PartitionSpec - Generates Iceberg ManifestFile records pointing to existing Parquet files - Writes Iceberg snapshot and table metadata JSON - Zero-copy: data files are referenced in-place, not duplicated Fixes apache#41227 Signed-off-by: melon tusk <paramjeetsingh.dev.mail@gmail.com>
![github-actions[bot]](https://avatars.githubusercontent.com/in/15368?s=60&v=4){kind=small}
There was a problem hiding this comment.
📋 Review Summary
This PR introduces apache_beam.io.lakehouse with a translate_delta_to_iceberg utility intended to synthesize Iceberg v2 metadata from Delta Lake logs. However, the current implementation has critical correctness issues in both Delta Lake log parsing and Iceberg v2 metadata generation that prevent it from working on real Delta tables or producing valid Iceberg tables.
🔍 General Feedback
- Delta Log Parsing & Active File Resolution: Delta transaction logs (
_delta_log/*.json) use newline-delimited JSON (NDJSON) with nestedmetaData,add, andremoveactions. Parsing the entire file viajson.loads()fails on multi-line logs, and scanning the filesystem directory directly instead of reconcilingadd/removeactions risks including deleted/tombstoned data files and_delta_log/*.checkpoint.parquetfiles. - Iceberg v2 Spec Compliance: Iceberg manifest files and manifest lists must be written in binary Apache Avro format (or generated via
pyiceberg), not JSON strings saved with an.avroextension. Additionally, the table metadata JSON is missing full data column schemas and required Iceberg v2 fields. - Missing Unit Tests: Please add unit tests (e.g.,
sdks/python/apache_beam/io/lakehouse/translate_test.py) covering partitioned and unpartitioned Delta tables, active vs. removed files, and error handling when_delta_logis missing.
💡 Tip: Comment /gemini review on this pull request to trigger a new Gemini code review.
| import os | ||
| import uuid | ||
| from datetime import datetime, timezone | ||
| from typing import Any, Dict, List, Optional |
There was a problem hiding this comment.
🟢 Low: Optional is imported but unused, which will cause Beam's lint checks to fail.
| from typing import Any, Dict, List, Optional | |
| from typing import Any, Dict, List |
| if not meta.path.endswith(".json"): | ||
| continue | ||
| content = FileSystems.open(meta.path).read().decode("utf-8") | ||
| entry = json.loads(content) |
There was a problem hiding this comment.
🔴 Critical: Delta Lake commit logs (_delta_log/*.json) are newline-delimited JSON (NDJSON) where each line is a separate action object, and partition columns are nested under entry["metaData"]["partitionColumns"] rather than at the top level. Calling json.loads(content) on the entire file raises JSONDecodeError on any commit file with multiple lines, and the broad except Exception: pass on lines 51–52 silently swallows the error and returns [].
Additionally, the docstring for translate_delta_to_iceberg states that ValueError is raised if no transaction logs are found, so missing/empty logs and malformed entries should not be silently ignored.
| """List all Parquet files under the Delta data directory.""" | ||
| parquet_files: List[str] = [] | ||
| try: | ||
| result = FileSystems.match([delta_data_dir]) |
There was a problem hiding this comment.
🔴 Critical: Listing Parquet files via FileSystems.match([delta_data_dir]) has two critical issues:
- Passing a directory URI without a recursive glob pattern does not match
.parquetfiles inside partitioned subdirectories. - Scanning the directory directly instead of replaying
addandremoveactions from_delta_logwill include tombstoned/compacted Parquet files (prior toVACUUM) as well as_delta_log/*.checkpoint.parquetfiles, causing data duplication or corruption. Active Parquet files (along with their exactsize,partitionValues, andstatssuch asnumRecords) should be extracted by reconcilingaddandremoveactions from the Delta log.
| if not table_identifier: | ||
| raise ValueError("table_identifier must be specified") | ||
|
|
||
| delta_log_dir = os.path.join(delta_table_uri, "_delta_log") |
There was a problem hiding this comment.
🟡 Medium: Use FileSystems.join instead of os.path.join when constructing filesystem paths and cloud URIs (here and on lines 43, 144, 184, 195, and 230). os.path.join uses OS-specific path separators (\ on Windows), which breaks cloud URIs like gs:// or s3://.
| delta_log_dir = os.path.join(delta_table_uri, "_delta_log") | |
| delta_log_dir = FileSystems.join(delta_table_uri, "_delta_log") |
| ) | ||
|
|
||
| # Write manifest file | ||
| manifest_file = os.path.join(metadata_dir, f"{uuid.uuid4()}.avro") |
There was a problem hiding this comment.
🔴 Critical: Lines 184–207 write JSON strings (json.dumps(...)) into files named {uuid}.avro and {snapshot_id}.avro. Per the Apache Iceberg specification, both manifest files and the snapshot manifest-list file must be valid binary Apache Avro files written with the Iceberg Avro schemas (and manifest-list must contain manifest_file entries pointing to the manifest Avro files, rather than embedding the snapshot metadata object itself). Any Iceberg reader (PyIceberg, Spark, Trino, BigLake) will reject these .avro files.
| "added-data-files": manifest_entries, | ||
| "deleted-data-files": [], | ||
| } | ||
| FileSystems.create(manifest_file).write(json.dumps(manifest_content).encode("utf-8")) |
There was a problem hiding this comment.
🟠 High: FileSystems.create(...) (here and on lines 207 and 231) and FileSystems.open(...) (line 47) return file handles that are never closed. For remote filesystems in Beam (such as GcsIO and S3IO), FileSystems.create returns a buffered upload channel that must be closed via a with context manager (with FileSystems.create(...) as f:) to flush buffers and finalize the object upload.
| "location": delta_table_uri, | ||
| "last-updated-ms": int(datetime.now(timezone.utc).timestamp() * 1000), | ||
| "last-column-id": 1000 + len(partition_cols), | ||
| "schema": { |
There was a problem hiding this comment.
🔴 Critical: The generated table_metadata is incomplete and does not conform to the Iceberg v2 table metadata specification:
- The synthesized schema only includes
partition_cols(all hardcoded to"string"), omitting all non-partition data columns defined in Delta'smetaData.schemaString. - Iceberg v2 requires
"schemas"(a list of schema objects with"type": "struct"),"current-schema-id","last-partition-id","sort-orders", and"default-sort-order-id"instead of"schema"and"sort-ids".
Fixes #41227
Add apache_beam.io.lakehouse package with translate_delta_to_iceberg()
function that synthesizes Iceberg v2 metadata from Delta Lake transaction
logs without copying underlying Parquet data files.
Signed-off-by: user-Param user-Param@users.noreply.github.com