github-actions[bot] commented on code in PR #41234: URL: https://github.com/apache/beam/pull/41234#discussion_r4546868368
########## sdks/python/apache_beam/io/lakehouse/translate.py: ########## @@ -0,0 +1,233 @@ +# +# Licensed to the Apache Software Foundation (ASF) under one or more +# contributor license agreements. See the NOTICE file distributed with +# this work for additional information regarding copyright ownership. +# The ASF licenses this file to You under the Apache License, Version 2.0 +# (the "License"); you may not use this file except in compliance with +# the License. You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. +# + +"""Zero-copy metadata translation from Delta Lake to Apache Iceberg. + +Both Delta Lake and Apache Iceberg use Parquet for data storage, so Iceberg +manifest files can point directly at Delta's Parquet files without copying +the underlying data. This module parses Delta transaction logs and +generates Iceberg v2 metadata (manifests, snapshot, partition spec). +""" + +import json +import os +import uuid +from datetime import datetime, timezone +from typing import Any, Dict, List, Optional Review Comment: **🟢 Low**: `Optional` is imported but unused, which will cause Beam's lint checks to fail. ```suggestion from typing import Any, Dict, List ``` ########## sdks/python/apache_beam/io/lakehouse/translate.py: ########## @@ -0,0 +1,233 @@ +# +# Licensed to the Apache Software Foundation (ASF) under one or more +# contributor license agreements. See the NOTICE file distributed with +# this work for additional information regarding copyright ownership. +# The ASF licenses this file to You under the Apache License, Version 2.0 +# (the "License"); you may not use this file except in compliance with +# the License. You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. +# + +"""Zero-copy metadata translation from Delta Lake to Apache Iceberg. + +Both Delta Lake and Apache Iceberg use Parquet for data storage, so Iceberg +manifest files can point directly at Delta's Parquet files without copying +the underlying data. This module parses Delta transaction logs and +generates Iceberg v2 metadata (manifests, snapshot, partition spec). +""" + +import json +import os +import uuid +from datetime import datetime, timezone +from typing import Any, Dict, List, Optional + +from apache_beam.io.filesystems import FileSystems + + +def _parse_delta_partition_spec(delta_log_dir: str) -> List[Dict[str, Any]]: + """Parse Delta partition specifications from _delta_log/*.json. + + Reads the latest Delta log checkpoint and commit entries to extract + the partition column definitions. + """ + partition_cols: List[str] = [] + try: + log_files = FileSystems.match([os.path.join(delta_log_dir, "*")])[0].metadata_list + for meta in sorted(log_files, key=lambda m: m.path): + if not meta.path.endswith(".json"): + continue + content = FileSystems.open(meta.path).read().decode("utf-8") + entry = json.loads(content) + if "partitionColumns" in entry: + partition_cols = entry["partitionColumns"] + except Exception: + pass + return [{"name": c, "type": "string"} for c in partition_cols] + + +def _list_delta_parquet_files(delta_data_dir: str) -> List[str]: + """List all Parquet files under the Delta data directory.""" + parquet_files: List[str] = [] + try: + result = FileSystems.match([delta_data_dir]) Review Comment: **🔴 Critical**: Listing Parquet files via `FileSystems.match([delta_data_dir])` has two critical issues: 1. Passing a directory URI without a recursive glob pattern does not match `.parquet` files inside partitioned subdirectories. 2. Scanning the directory directly instead of replaying `add` and `remove` actions from `_delta_log` will include tombstoned/compacted Parquet files (prior to `VACUUM`) as well as `_delta_log/*.checkpoint.parquet` files, causing data duplication or corruption. Active Parquet files (along with their exact `size`, `partitionValues`, and `stats` such as `numRecords`) should be extracted by reconciling `add` and `remove` actions from the Delta log. ########## sdks/python/apache_beam/io/lakehouse/translate.py: ########## @@ -0,0 +1,233 @@ +# +# Licensed to the Apache Software Foundation (ASF) under one or more +# contributor license agreements. See the NOTICE file distributed with +# this work for additional information regarding copyright ownership. +# The ASF licenses this file to You under the Apache License, Version 2.0 +# (the "License"); you may not use this file except in compliance with +# the License. You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. +# + +"""Zero-copy metadata translation from Delta Lake to Apache Iceberg. + +Both Delta Lake and Apache Iceberg use Parquet for data storage, so Iceberg +manifest files can point directly at Delta's Parquet files without copying +the underlying data. This module parses Delta transaction logs and +generates Iceberg v2 metadata (manifests, snapshot, partition spec). +""" + +import json +import os +import uuid +from datetime import datetime, timezone +from typing import Any, Dict, List, Optional + +from apache_beam.io.filesystems import FileSystems + + +def _parse_delta_partition_spec(delta_log_dir: str) -> List[Dict[str, Any]]: + """Parse Delta partition specifications from _delta_log/*.json. + + Reads the latest Delta log checkpoint and commit entries to extract + the partition column definitions. + """ + partition_cols: List[str] = [] + try: + log_files = FileSystems.match([os.path.join(delta_log_dir, "*")])[0].metadata_list + for meta in sorted(log_files, key=lambda m: m.path): + if not meta.path.endswith(".json"): + continue + content = FileSystems.open(meta.path).read().decode("utf-8") + entry = json.loads(content) Review 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. ########## sdks/python/apache_beam/io/lakehouse/translate.py: ########## @@ -0,0 +1,233 @@ +# +# Licensed to the Apache Software Foundation (ASF) under one or more +# contributor license agreements. See the NOTICE file distributed with +# this work for additional information regarding copyright ownership. +# The ASF licenses this file to You under the Apache License, Version 2.0 +# (the "License"); you may not use this file except in compliance with +# the License. You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. +# + +"""Zero-copy metadata translation from Delta Lake to Apache Iceberg. + +Both Delta Lake and Apache Iceberg use Parquet for data storage, so Iceberg +manifest files can point directly at Delta's Parquet files without copying +the underlying data. This module parses Delta transaction logs and +generates Iceberg v2 metadata (manifests, snapshot, partition spec). +""" + +import json +import os +import uuid +from datetime import datetime, timezone +from typing import Any, Dict, List, Optional + +from apache_beam.io.filesystems import FileSystems + + +def _parse_delta_partition_spec(delta_log_dir: str) -> List[Dict[str, Any]]: + """Parse Delta partition specifications from _delta_log/*.json. + + Reads the latest Delta log checkpoint and commit entries to extract + the partition column definitions. + """ + partition_cols: List[str] = [] + try: + log_files = FileSystems.match([os.path.join(delta_log_dir, "*")])[0].metadata_list + for meta in sorted(log_files, key=lambda m: m.path): + if not meta.path.endswith(".json"): + continue + content = FileSystems.open(meta.path).read().decode("utf-8") + entry = json.loads(content) + if "partitionColumns" in entry: + partition_cols = entry["partitionColumns"] + except Exception: + pass + return [{"name": c, "type": "string"} for c in partition_cols] + + +def _list_delta_parquet_files(delta_data_dir: str) -> List[str]: + """List all Parquet files under the Delta data directory.""" + parquet_files: List[str] = [] + try: + result = FileSystems.match([delta_data_dir]) + for metadata in result[0].metadata_list: + if metadata.path.endswith(".parquet"): + parquet_files.append(metadata.path) + except Exception: + pass + return parquet_files + + +def _generate_manifest_entry( + parquet_path: str, + partition_values: Dict[str, str], + file_size: int, + record_count: int, +) -> Dict[str, Any]: + """Generate a single Iceberg manifest entry for a Parquet file.""" + return { + "status": 0, # existing + "primary_key": None, + "file_path": parquet_path, + "file_format": "PARQUET", + "record_count": record_count, + "file_size_in_bytes": file_size, + "column_sizes": {}, + "value_counts": {}, + "null_value_counts": {}, + "nan_value_counts": {}, + "lower_bounds": {}, + "upper_bounds": {}, + "key_metadata": None, + "split_offsets": [], + "equality_deletes": None, + "position_deletes": None, + "partition": partition_values, + } + + +def translate_delta_to_iceberg( + delta_table_uri: str, + iceberg_warehouse: str, + table_identifier: str, +) -> str: + """Synthesize Iceberg v2 metadata from Delta Lake transaction logs. + + Parses Delta ``_delta_log/*.json`` transaction logs to extract the + partition specification and Parquet file references, then generates + Iceberg manifest files, a snapshot, and table metadata pointing at the + existing Parquet data. No data files are copied. + + Args: + delta_table_uri: URI of the Delta table directory (e.g. + ``gs://my-bucket/delta/orders/``). + iceberg_warehouse: URI of the Iceberg warehouse directory where + synthesized metadata will be written (e.g. + ``gs://my-bucket/iceberg/wh/``). + table_identifier: Fully qualified table identifier (e.g. + ``analytics.orders``). + + Returns: + The URI of the generated Iceberg metadata directory. + + Raises: + ValueError: If the Delta table URI is invalid or no transaction + logs are found. + """ + if not delta_table_uri: + raise ValueError("delta_table_uri must be specified") + if not iceberg_warehouse: + raise ValueError("iceberg_warehouse must be specified") + if not table_identifier: + raise ValueError("table_identifier must be specified") + + delta_log_dir = os.path.join(delta_table_uri, "_delta_log") Review 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://`. ```suggestion delta_log_dir = FileSystems.join(delta_table_uri, "_delta_log") ``` ########## sdks/python/apache_beam/io/lakehouse/translate.py: ########## @@ -0,0 +1,233 @@ +# +# Licensed to the Apache Software Foundation (ASF) under one or more +# contributor license agreements. See the NOTICE file distributed with +# this work for additional information regarding copyright ownership. +# The ASF licenses this file to You under the Apache License, Version 2.0 +# (the "License"); you may not use this file except in compliance with +# the License. You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. +# + +"""Zero-copy metadata translation from Delta Lake to Apache Iceberg. + +Both Delta Lake and Apache Iceberg use Parquet for data storage, so Iceberg +manifest files can point directly at Delta's Parquet files without copying +the underlying data. This module parses Delta transaction logs and +generates Iceberg v2 metadata (manifests, snapshot, partition spec). +""" + +import json +import os +import uuid +from datetime import datetime, timezone +from typing import Any, Dict, List, Optional + +from apache_beam.io.filesystems import FileSystems + + +def _parse_delta_partition_spec(delta_log_dir: str) -> List[Dict[str, Any]]: + """Parse Delta partition specifications from _delta_log/*.json. + + Reads the latest Delta log checkpoint and commit entries to extract + the partition column definitions. + """ + partition_cols: List[str] = [] + try: + log_files = FileSystems.match([os.path.join(delta_log_dir, "*")])[0].metadata_list + for meta in sorted(log_files, key=lambda m: m.path): + if not meta.path.endswith(".json"): + continue + content = FileSystems.open(meta.path).read().decode("utf-8") + entry = json.loads(content) + if "partitionColumns" in entry: + partition_cols = entry["partitionColumns"] + except Exception: + pass + return [{"name": c, "type": "string"} for c in partition_cols] + + +def _list_delta_parquet_files(delta_data_dir: str) -> List[str]: + """List all Parquet files under the Delta data directory.""" + parquet_files: List[str] = [] + try: + result = FileSystems.match([delta_data_dir]) + for metadata in result[0].metadata_list: + if metadata.path.endswith(".parquet"): + parquet_files.append(metadata.path) + except Exception: + pass + return parquet_files + + +def _generate_manifest_entry( + parquet_path: str, + partition_values: Dict[str, str], + file_size: int, + record_count: int, +) -> Dict[str, Any]: + """Generate a single Iceberg manifest entry for a Parquet file.""" + return { + "status": 0, # existing + "primary_key": None, + "file_path": parquet_path, + "file_format": "PARQUET", + "record_count": record_count, + "file_size_in_bytes": file_size, + "column_sizes": {}, + "value_counts": {}, + "null_value_counts": {}, + "nan_value_counts": {}, + "lower_bounds": {}, + "upper_bounds": {}, + "key_metadata": None, + "split_offsets": [], + "equality_deletes": None, + "position_deletes": None, + "partition": partition_values, + } + + +def translate_delta_to_iceberg( + delta_table_uri: str, + iceberg_warehouse: str, + table_identifier: str, +) -> str: + """Synthesize Iceberg v2 metadata from Delta Lake transaction logs. + + Parses Delta ``_delta_log/*.json`` transaction logs to extract the + partition specification and Parquet file references, then generates + Iceberg manifest files, a snapshot, and table metadata pointing at the + existing Parquet data. No data files are copied. + + Args: + delta_table_uri: URI of the Delta table directory (e.g. + ``gs://my-bucket/delta/orders/``). + iceberg_warehouse: URI of the Iceberg warehouse directory where + synthesized metadata will be written (e.g. + ``gs://my-bucket/iceberg/wh/``). + table_identifier: Fully qualified table identifier (e.g. + ``analytics.orders``). + + Returns: + The URI of the generated Iceberg metadata directory. + + Raises: + ValueError: If the Delta table URI is invalid or no transaction + logs are found. + """ + if not delta_table_uri: + raise ValueError("delta_table_uri must be specified") + if not iceberg_warehouse: + raise ValueError("iceberg_warehouse must be specified") + if not table_identifier: + raise ValueError("table_identifier must be specified") + + delta_log_dir = os.path.join(delta_table_uri, "_delta_log") + delta_data_dir = delta_table_uri + + # Parse partition specification from Delta logs + partition_cols = _parse_delta_partition_spec(delta_log_dir) + + # List existing Parquet data files + parquet_files = _list_delta_parquet_files(delta_data_dir) + + # Generate Iceberg metadata + table_uuid = str(uuid.uuid4()) + snapshot_id = int(datetime.now(timezone.utc).timestamp() * 1_000_000) + metadata_dir = os.path.join(iceberg_warehouse, table_identifier.replace(".", "/"), "metadata") + FileSystems.mkdirs(metadata_dir) + + # Build partition spec + partition_spec = { + "partition-spec-id": 0, + "spec-id": 0, + "fields": [ + { + "source-id": i + 1, + "field-id": i + 1000, + "transform": "identity", + "name": col["name"], + } + for i, col in enumerate(partition_cols) + ], + } + + # Build manifest entries + manifest_entries = [] + for pf in parquet_files: + # Extract partition values from path (delta=value/...) + partition_values = {} + for col in partition_cols: + col_name = col["name"] + # Look for col_name=value in path + for part in pf.split("/"): + if part.startswith(col_name + "="): + partition_values[col_name] = part.split("=", 1)[1] + break + manifest_entries.append( + _generate_manifest_entry( + parquet_path=pf, + partition_values=partition_values, + file_size=0, # size unknown without reading Parquet footer + record_count=0, # count unknown without reading Parquet footer + ) + ) + + # Write manifest file + manifest_file = os.path.join(metadata_dir, f"{uuid.uuid4()}.avro") + manifest_content = { + "format": 2, + "partition-spec-id": 0, + "added-snapshot-id": snapshot_id, + "added-data-files": manifest_entries, + "deleted-data-files": [], + } + FileSystems.create(manifest_file).write(json.dumps(manifest_content).encode("utf-8")) + + # Write snapshot + snapshot_file = os.path.join(metadata_dir, f"{snapshot_id}.avro") + snapshot_content = { + "snapshot-id": snapshot_id, + "parent-snapshot-id": None, + "timestamp-ms": int(datetime.now(timezone.utc).timestamp() * 1000), + "manifest-list": manifest_file, + "summary": { + "added-data-files": str(len(manifest_entries)), + "added-records": "0", + }, + "schema-id": 0, + } + FileSystems.create(snapshot_file).write(json.dumps(snapshot_content).encode("utf-8")) + + # Write table metadata + table_metadata = { + "format-version": 2, + "table-uuid": table_uuid, + "location": delta_table_uri, + "last-updated-ms": int(datetime.now(timezone.utc).timestamp() * 1000), + "last-column-id": 1000 + len(partition_cols), + "schema": { Review Comment: **🔴 Critical**: The generated `table_metadata` is incomplete and does not conform to the Iceberg v2 table metadata specification: 1. The synthesized schema only includes `partition_cols` (all hardcoded to `"string"`), omitting all non-partition data columns defined in Delta's `metaData.schemaString`. 2. 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"`. ########## sdks/python/apache_beam/io/lakehouse/translate.py: ########## @@ -0,0 +1,233 @@ +# +# Licensed to the Apache Software Foundation (ASF) under one or more +# contributor license agreements. See the NOTICE file distributed with +# this work for additional information regarding copyright ownership. +# The ASF licenses this file to You under the Apache License, Version 2.0 +# (the "License"); you may not use this file except in compliance with +# the License. You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. +# + +"""Zero-copy metadata translation from Delta Lake to Apache Iceberg. + +Both Delta Lake and Apache Iceberg use Parquet for data storage, so Iceberg +manifest files can point directly at Delta's Parquet files without copying +the underlying data. This module parses Delta transaction logs and +generates Iceberg v2 metadata (manifests, snapshot, partition spec). +""" + +import json +import os +import uuid +from datetime import datetime, timezone +from typing import Any, Dict, List, Optional + +from apache_beam.io.filesystems import FileSystems + + +def _parse_delta_partition_spec(delta_log_dir: str) -> List[Dict[str, Any]]: + """Parse Delta partition specifications from _delta_log/*.json. + + Reads the latest Delta log checkpoint and commit entries to extract + the partition column definitions. + """ + partition_cols: List[str] = [] + try: + log_files = FileSystems.match([os.path.join(delta_log_dir, "*")])[0].metadata_list + for meta in sorted(log_files, key=lambda m: m.path): + if not meta.path.endswith(".json"): + continue + content = FileSystems.open(meta.path).read().decode("utf-8") + entry = json.loads(content) + if "partitionColumns" in entry: + partition_cols = entry["partitionColumns"] + except Exception: + pass + return [{"name": c, "type": "string"} for c in partition_cols] + + +def _list_delta_parquet_files(delta_data_dir: str) -> List[str]: + """List all Parquet files under the Delta data directory.""" + parquet_files: List[str] = [] + try: + result = FileSystems.match([delta_data_dir]) + for metadata in result[0].metadata_list: + if metadata.path.endswith(".parquet"): + parquet_files.append(metadata.path) + except Exception: + pass + return parquet_files + + +def _generate_manifest_entry( + parquet_path: str, + partition_values: Dict[str, str], + file_size: int, + record_count: int, +) -> Dict[str, Any]: + """Generate a single Iceberg manifest entry for a Parquet file.""" + return { + "status": 0, # existing + "primary_key": None, + "file_path": parquet_path, + "file_format": "PARQUET", + "record_count": record_count, + "file_size_in_bytes": file_size, + "column_sizes": {}, + "value_counts": {}, + "null_value_counts": {}, + "nan_value_counts": {}, + "lower_bounds": {}, + "upper_bounds": {}, + "key_metadata": None, + "split_offsets": [], + "equality_deletes": None, + "position_deletes": None, + "partition": partition_values, + } + + +def translate_delta_to_iceberg( + delta_table_uri: str, + iceberg_warehouse: str, + table_identifier: str, +) -> str: + """Synthesize Iceberg v2 metadata from Delta Lake transaction logs. + + Parses Delta ``_delta_log/*.json`` transaction logs to extract the + partition specification and Parquet file references, then generates + Iceberg manifest files, a snapshot, and table metadata pointing at the + existing Parquet data. No data files are copied. + + Args: + delta_table_uri: URI of the Delta table directory (e.g. + ``gs://my-bucket/delta/orders/``). + iceberg_warehouse: URI of the Iceberg warehouse directory where + synthesized metadata will be written (e.g. + ``gs://my-bucket/iceberg/wh/``). + table_identifier: Fully qualified table identifier (e.g. + ``analytics.orders``). + + Returns: + The URI of the generated Iceberg metadata directory. + + Raises: + ValueError: If the Delta table URI is invalid or no transaction + logs are found. + """ + if not delta_table_uri: + raise ValueError("delta_table_uri must be specified") + if not iceberg_warehouse: + raise ValueError("iceberg_warehouse must be specified") + if not table_identifier: + raise ValueError("table_identifier must be specified") + + delta_log_dir = os.path.join(delta_table_uri, "_delta_log") + delta_data_dir = delta_table_uri + + # Parse partition specification from Delta logs + partition_cols = _parse_delta_partition_spec(delta_log_dir) + + # List existing Parquet data files + parquet_files = _list_delta_parquet_files(delta_data_dir) + + # Generate Iceberg metadata + table_uuid = str(uuid.uuid4()) + snapshot_id = int(datetime.now(timezone.utc).timestamp() * 1_000_000) + metadata_dir = os.path.join(iceberg_warehouse, table_identifier.replace(".", "/"), "metadata") + FileSystems.mkdirs(metadata_dir) + + # Build partition spec + partition_spec = { + "partition-spec-id": 0, + "spec-id": 0, + "fields": [ + { + "source-id": i + 1, + "field-id": i + 1000, + "transform": "identity", + "name": col["name"], + } + for i, col in enumerate(partition_cols) + ], + } + + # Build manifest entries + manifest_entries = [] + for pf in parquet_files: + # Extract partition values from path (delta=value/...) + partition_values = {} + for col in partition_cols: + col_name = col["name"] + # Look for col_name=value in path + for part in pf.split("/"): + if part.startswith(col_name + "="): + partition_values[col_name] = part.split("=", 1)[1] + break + manifest_entries.append( + _generate_manifest_entry( + parquet_path=pf, + partition_values=partition_values, + file_size=0, # size unknown without reading Parquet footer + record_count=0, # count unknown without reading Parquet footer + ) + ) + + # Write manifest file + manifest_file = os.path.join(metadata_dir, f"{uuid.uuid4()}.avro") + manifest_content = { + "format": 2, + "partition-spec-id": 0, + "added-snapshot-id": snapshot_id, + "added-data-files": manifest_entries, + "deleted-data-files": [], + } + FileSystems.create(manifest_file).write(json.dumps(manifest_content).encode("utf-8")) Review 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. ########## sdks/python/apache_beam/io/lakehouse/translate.py: ########## @@ -0,0 +1,233 @@ +# +# Licensed to the Apache Software Foundation (ASF) under one or more +# contributor license agreements. See the NOTICE file distributed with +# this work for additional information regarding copyright ownership. +# The ASF licenses this file to You under the Apache License, Version 2.0 +# (the "License"); you may not use this file except in compliance with +# the License. You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. +# + +"""Zero-copy metadata translation from Delta Lake to Apache Iceberg. + +Both Delta Lake and Apache Iceberg use Parquet for data storage, so Iceberg +manifest files can point directly at Delta's Parquet files without copying +the underlying data. This module parses Delta transaction logs and +generates Iceberg v2 metadata (manifests, snapshot, partition spec). +""" + +import json +import os +import uuid +from datetime import datetime, timezone +from typing import Any, Dict, List, Optional + +from apache_beam.io.filesystems import FileSystems + + +def _parse_delta_partition_spec(delta_log_dir: str) -> List[Dict[str, Any]]: + """Parse Delta partition specifications from _delta_log/*.json. + + Reads the latest Delta log checkpoint and commit entries to extract + the partition column definitions. + """ + partition_cols: List[str] = [] + try: + log_files = FileSystems.match([os.path.join(delta_log_dir, "*")])[0].metadata_list + for meta in sorted(log_files, key=lambda m: m.path): + if not meta.path.endswith(".json"): + continue + content = FileSystems.open(meta.path).read().decode("utf-8") + entry = json.loads(content) + if "partitionColumns" in entry: + partition_cols = entry["partitionColumns"] + except Exception: + pass + return [{"name": c, "type": "string"} for c in partition_cols] + + +def _list_delta_parquet_files(delta_data_dir: str) -> List[str]: + """List all Parquet files under the Delta data directory.""" + parquet_files: List[str] = [] + try: + result = FileSystems.match([delta_data_dir]) + for metadata in result[0].metadata_list: + if metadata.path.endswith(".parquet"): + parquet_files.append(metadata.path) + except Exception: + pass + return parquet_files + + +def _generate_manifest_entry( + parquet_path: str, + partition_values: Dict[str, str], + file_size: int, + record_count: int, +) -> Dict[str, Any]: + """Generate a single Iceberg manifest entry for a Parquet file.""" + return { + "status": 0, # existing + "primary_key": None, + "file_path": parquet_path, + "file_format": "PARQUET", + "record_count": record_count, + "file_size_in_bytes": file_size, + "column_sizes": {}, + "value_counts": {}, + "null_value_counts": {}, + "nan_value_counts": {}, + "lower_bounds": {}, + "upper_bounds": {}, + "key_metadata": None, + "split_offsets": [], + "equality_deletes": None, + "position_deletes": None, + "partition": partition_values, + } + + +def translate_delta_to_iceberg( + delta_table_uri: str, + iceberg_warehouse: str, + table_identifier: str, +) -> str: + """Synthesize Iceberg v2 metadata from Delta Lake transaction logs. + + Parses Delta ``_delta_log/*.json`` transaction logs to extract the + partition specification and Parquet file references, then generates + Iceberg manifest files, a snapshot, and table metadata pointing at the + existing Parquet data. No data files are copied. + + Args: + delta_table_uri: URI of the Delta table directory (e.g. + ``gs://my-bucket/delta/orders/``). + iceberg_warehouse: URI of the Iceberg warehouse directory where + synthesized metadata will be written (e.g. + ``gs://my-bucket/iceberg/wh/``). + table_identifier: Fully qualified table identifier (e.g. + ``analytics.orders``). + + Returns: + The URI of the generated Iceberg metadata directory. + + Raises: + ValueError: If the Delta table URI is invalid or no transaction + logs are found. + """ + if not delta_table_uri: + raise ValueError("delta_table_uri must be specified") + if not iceberg_warehouse: + raise ValueError("iceberg_warehouse must be specified") + if not table_identifier: + raise ValueError("table_identifier must be specified") + + delta_log_dir = os.path.join(delta_table_uri, "_delta_log") + delta_data_dir = delta_table_uri + + # Parse partition specification from Delta logs + partition_cols = _parse_delta_partition_spec(delta_log_dir) + + # List existing Parquet data files + parquet_files = _list_delta_parquet_files(delta_data_dir) + + # Generate Iceberg metadata + table_uuid = str(uuid.uuid4()) + snapshot_id = int(datetime.now(timezone.utc).timestamp() * 1_000_000) + metadata_dir = os.path.join(iceberg_warehouse, table_identifier.replace(".", "/"), "metadata") + FileSystems.mkdirs(metadata_dir) + + # Build partition spec + partition_spec = { + "partition-spec-id": 0, + "spec-id": 0, + "fields": [ + { + "source-id": i + 1, + "field-id": i + 1000, + "transform": "identity", + "name": col["name"], + } + for i, col in enumerate(partition_cols) + ], + } + + # Build manifest entries + manifest_entries = [] + for pf in parquet_files: + # Extract partition values from path (delta=value/...) + partition_values = {} + for col in partition_cols: + col_name = col["name"] + # Look for col_name=value in path + for part in pf.split("/"): + if part.startswith(col_name + "="): + partition_values[col_name] = part.split("=", 1)[1] + break + manifest_entries.append( + _generate_manifest_entry( + parquet_path=pf, + partition_values=partition_values, + file_size=0, # size unknown without reading Parquet footer + record_count=0, # count unknown without reading Parquet footer + ) + ) + + # Write manifest file + manifest_file = os.path.join(metadata_dir, f"{uuid.uuid4()}.avro") Review 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. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
