feat(elt-common): Add merge support to pyiceberg destination#167
Conversation
No support is provided for the other merge strategies yet
|
Important Review skippedAuto incremental reviews are disabled on this repository. Please check the settings in the CodeRabbit UI or the You can disable this status message by setting the WalkthroughThis PR refactors PyIceberg destination handling by consolidating type mapping and schema translation logic into focused helper modules. Changes include removing Changes
Poem
Pre-merge checks❌ Failed checks (1 warning)
✅ Passed checks (4 passed)
Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out. Comment |
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 4
🧹 Nitpick comments (9)
elt-common/tests/unit_tests/dlt_destinations/pyiceberg/test_helpers.py (2)
35-35: Test data uses integers where strings are expected.The
create_catalogfunction signature specifies**properties: str, but the test passes integer values{"a": 1, "c": 3}. Consider using string values to match the expected type signature.- name, properties = "unit_test_catalog", {"a": 1, "c": 3} + name, properties = "unit_test_catalog", {"a": "1", "c": "3"}
102-102: Typo in comment."arugments" should be "arguments".
- # decimal needs extra arugments + # decimal needs extra argumentselt-common/src/elt_common/dlt_destinations/pyiceberg/type_mapping.py (1)
25-26: Missing type annotation foriceberg_fieldparameter.The
iceberg_fieldparameter lacks a type annotation. Based on the helper function signature, this should beNestedField.- def from_destination_type(self, iceberg_field) -> TColumnType: + def from_destination_type(self, iceberg_field: "NestedField") -> TColumnType:You'll also need to import
NestedFieldfrom pyiceberg or add a forward reference.elt-common/src/elt_common/dlt_destinations/pyiceberg/pyiceberg.py (2)
148-153: Consider replacingassertwith explicit exception for replace strategy validation.The
assertstatement on line 152 may be skipped if Python runs with optimisations enabled (-Oflag). For production code paths, an explicit exception provides more reliable validation.- assert replace_strategy, f"Must be able to get replace strategy for {table_name}" + if not replace_strategy: + raise DestinationTerminalException( + f"Unable to resolve replace strategy for table '{table_name}'." + )
417-426: Redundant call toprepare_load_tableand typo.
Redundant call:
prepare_load_tableis called again here to retrievex-merge-strategy, but this method was already called earlier in the load flow. Consider passing the strategy directly or caching the prepared table to avoid redundant computation.Typo: Line 420 has "indentifier" which should be "identifier".
- strategy = self.prepare_load_table(table_name)["x-merge-strategy"] # type:ignore + # Consider accepting strategy as parameter or caching prepared table if strategy == "upsert": - # requires the indentifier fields to have been defined + # requires the identifier fields to have been defined table.upsert(elt-common/src/elt_common/dlt_destinations/pyiceberg/helpers.py (4)
50-58:create_catalogis hardcoded toRestCatalog.This factory currently only creates
RestCataloginstances. If other catalog types (e.g.,GlueCatalog,HiveCatalog) are needed in future, this would require modification. The current approach is acceptable if REST is the only supported backend.
87-94: Inconsistent handling ofprecisionandscalefor decimal type.
precisionis accessed withcolumn["precision"](raisesKeyErrorif missing), whilstscaleusescolumn.get("scale")(returnsNoneif missing). Ifscaleis missing, theKeyErrorexception won't be raised, yetNonemay not be a valid value forDecimalType.Consider using consistent access patterns:
elif dlt_type == "decimal": - try: - return DecimalType(column["precision"], column.get("scale")) # type: ignore - except KeyError: - missing = [key for key in ("precision", "scale") if key not in column] - raise TypeError( - f"Column with decimal dlt type cannot be created, missing fields: {missing}" - ) + missing = [key for key in ("precision", "scale") if key not in column] + if missing: + raise TypeError( + f"Column with decimal dlt type cannot be created, missing fields: {missing}" + ) + return DecimalType(column["precision"], column["scale"]) # type: ignore
174-182: Consider using@dataclassforPartitionTransformation.The class is a simple data container. Using
@dataclasswould reduce boilerplate and provide automatic__repr__,__eq__, etc.+from dataclasses import dataclass + +@dataclass class PartitionTransformation: - transform: str """The transform as a string representation understood by pyicberg.transforms.parse_transform., e.g. `bucket[16]`""" - column_name: str + transform: str """Column name to apply the transformation to""" - - def __init__(self, transform: str, column_name: str) -> None: - self.transform = transform - self.column_name = column_name + column_name: str
286-288: Type annotation style inconsistency.Line 286 uses
Dict[str, str] | None(Python 3.10+ union syntax), whilst the imports at line 1 useUnionfromtyping. Consider using consistent annotation style throughout.- partition_hint: Dict[str, str] | None = dlt_schema.get(PARTITION_HINT) + partition_hint: Union[Dict[str, str], None] = dlt_schema.get(PARTITION_HINT)Alternatively, if Python 3.10+ is the minimum supported version, update the imports to use
from __future__ import annotationsand use the newer syntax consistently.
📜 Review details
Configuration used: Path: .coderabbit.yaml
Review profile: CHILL
Plan: Pro
Cache: Disabled due to data retention organization setting
Knowledge base: Disabled due to data retention organization setting
📒 Files selected for processing (13)
elt-common/src/elt_common/dlt_destinations/pyiceberg/catalog.py(0 hunks)elt-common/src/elt_common/dlt_destinations/pyiceberg/factory.py(2 hunks)elt-common/src/elt_common/dlt_destinations/pyiceberg/helpers.py(1 hunks)elt-common/src/elt_common/dlt_destinations/pyiceberg/pyiceberg.py(10 hunks)elt-common/src/elt_common/dlt_destinations/pyiceberg/pyiceberg_adapter.py(3 hunks)elt-common/src/elt_common/dlt_destinations/pyiceberg/schema.py(0 hunks)elt-common/src/elt_common/dlt_destinations/pyiceberg/type_mapping.py(1 hunks)elt-common/tests/e2e_tests/conftest.py(3 hunks)elt-common/tests/e2e_tests/elt_common/dlt_destinations/pyiceberg/test_pyiceberg_pipeline.py(12 hunks)elt-common/tests/e2e_tests/elt_common/dlt_destinations/pyiceberg/utils.py(6 hunks)elt-common/tests/unit_tests/dlt_destinations/pyiceberg/test_helpers.py(1 hunks)elt-common/tests/unit_tests/dlt_destinations/pyiceberg/test_pyiceberg_adapter.py(1 hunks)warehouses/accelerator/extract_load/electricity_sharepoint/extract_and_load.py(2 hunks)
💤 Files with no reviewable changes (2)
- elt-common/src/elt_common/dlt_destinations/pyiceberg/catalog.py
- elt-common/src/elt_common/dlt_destinations/pyiceberg/schema.py
🧰 Additional context used
🧬 Code graph analysis (10)
elt-common/src/elt_common/dlt_destinations/pyiceberg/type_mapping.py (2)
elt-common/src/elt_common/dlt_destinations/pyiceberg/factory.py (1)
pyiceberg(15-58)elt-common/src/elt_common/dlt_destinations/pyiceberg/helpers.py (2)
dlt_type_to_iceberg(74-114)iceberg_to_dlt_type(117-150)
elt-common/src/elt_common/dlt_destinations/pyiceberg/factory.py (1)
elt-common/src/elt_common/dlt_destinations/pyiceberg/type_mapping.py (1)
PyIcebergTypeMapper(19-26)
elt-common/src/elt_common/dlt_destinations/pyiceberg/helpers.py (2)
elt-common/src/elt_common/dlt_destinations/pyiceberg/factory.py (1)
pyiceberg(15-58)elt-common/tests/unit_tests/dlt_destinations/pyiceberg/test_helpers.py (1)
dlt_schema(23-30)
elt-common/src/elt_common/dlt_destinations/pyiceberg/pyiceberg.py (3)
elt-common/src/elt_common/dlt_destinations/pyiceberg/factory.py (1)
pyiceberg(15-58)elt-common/src/elt_common/dlt_destinations/pyiceberg/helpers.py (4)
create_catalog(50-58)create_iceberg_schema(153-171)create_partition_spec(278-300)create_sort_order(303-320)elt-common/src/elt_common/dlt_destinations/pyiceberg/type_mapping.py (1)
from_destination_type(25-26)
elt-common/tests/e2e_tests/conftest.py (1)
elt-common/src/elt_common/dlt_destinations/pyiceberg/helpers.py (1)
create_catalog(50-58)
elt-common/tests/unit_tests/dlt_destinations/pyiceberg/test_helpers.py (1)
elt-common/src/elt_common/dlt_destinations/pyiceberg/helpers.py (8)
PartitionTrBuilder(185-224)create_catalog(50-58)create_iceberg_schema(153-171)create_partition_spec(278-300)dlt_type_to_iceberg(74-114)iceberg_to_dlt_type(117-150)namespace_exists(61-66)year(194-196)
elt-common/tests/e2e_tests/elt_common/dlt_destinations/pyiceberg/test_pyiceberg_pipeline.py (4)
elt-common/src/elt_common/dlt_destinations/pyiceberg/factory.py (1)
pyiceberg(15-58)elt-common/src/elt_common/dlt_destinations/pyiceberg/pyiceberg.py (1)
run(65-70)elt-common/tests/e2e_tests/elt_common/dlt_destinations/pyiceberg/conftest.py (2)
pipelines_dir(12-14)destination_config(18-23)elt-common/tests/e2e_tests/elt_common/dlt_destinations/pyiceberg/utils.py (3)
PyIcebergDestinationTestConfiguration(38-128)setup_pipeline(74-92)assert_table_has_data(289-314)
elt-common/tests/unit_tests/dlt_destinations/pyiceberg/test_pyiceberg_adapter.py (2)
elt-common/src/elt_common/dlt_destinations/pyiceberg/factory.py (1)
pyiceberg(15-58)elt-common/src/elt_common/dlt_destinations/pyiceberg/helpers.py (5)
PartitionTrBuilder(185-224)SortOrderBuilder(238-275)year(194-196)asc(257-259)build(265-266)
elt-common/tests/e2e_tests/elt_common/dlt_destinations/pyiceberg/utils.py (1)
elt-common/src/elt_common/dlt_destinations/pyiceberg/helpers.py (6)
PartitionTrBuilder(185-224)SortOrderBuilder(238-275)year(194-196)desc(261-263)build(265-266)asc(257-259)
elt-common/src/elt_common/dlt_destinations/pyiceberg/pyiceberg_adapter.py (1)
elt-common/src/elt_common/dlt_destinations/pyiceberg/helpers.py (5)
PartitionTransformation(174-182)PartitionTrBuilder(185-224)SortOrderSpecification(227-235)SortOrderBuilder(238-275)identity(189-191)
⏰ Context from checks skipped due to timeout of 90000ms. You can increase the timeout in your CodeRabbit configuration to a maximum of 15 minutes (900000ms). (1)
- GitHub Check: elt-common end-to-end tests
🔇 Additional comments (38)
elt-common/src/elt_common/dlt_destinations/pyiceberg/factory.py (2)
12-12: LGTM!The import path update correctly reflects the refactoring that moved
PyIcebergTypeMapperto the dedicatedtype_mappingmodule.
32-32: Merge strategy change aligns with PR objectives.Changing the supported merge strategy from
"delete-insert"to"upsert"enables the merge write disposition feature as described in the linked issue #165.warehouses/accelerator/extract_load/electricity_sharepoint/extract_and_load.py (2)
28-31: LGTM!The import update correctly adopts the new
PartitionTrBuilderAPI from the refactored module.
163-165: LGTM!The partition configuration correctly uses the new builder-based API with
PartitionTrBuilder.year("date_time").elt-common/tests/unit_tests/dlt_destinations/pyiceberg/test_pyiceberg_adapter.py (4)
1-9: LGTM!Imports are correctly structured, pulling the adapter, builders, and hint constants from the appropriate modules.
12-14: LGTM!Good test for validating the required parameter constraint—ensures the adapter raises
ValueErrorwhen neitherpartitionnorsort_orderis provided.
17-28: LGTM!The test correctly verifies that partition hints are populated in the computed table schema when using
PartitionTrBuilder.year().
30-39: LGTM!The test correctly verifies that sort order hints are populated in the computed table schema when using
SortOrderBuilder.elt-common/tests/e2e_tests/conftest.py (3)
19-19: LGTM!The import path update correctly reflects the refactoring that moved
create_catalogto thehelpersmodule.
358-360: Consistent with above changes.The
remove_bucketcall follows the same keyword argument pattern as the other Minio calls.
341-349: Minio client keyword arguments are properly supported — no issues found.The Minio Python client methods (
bucket_exists(),make_bucket()) acceptbucket_nameas their first positional parameter, which can be called using keyword argument syntax as done in the code (e.g.,bucket_name=bucket_name). This is valid Python and works correctly. The code follows proper conventions.elt-common/src/elt_common/dlt_destinations/pyiceberg/pyiceberg_adapter.py (5)
1-14: LGTM!The imports are well-structured. The
# noqa: F401onSortOrderBuilderis appropriate as it's being re-exported for external use.
16-19: LGTM!The type aliases
TPartitionTransformationandTSortOrderConfigurationimprove API clarity and support flexible input types.
51-65: LGTM!The partition handling logic correctly normalises input to a sequence and translates both string and
PartitionTransformationinputs into the appropriate hint format.
67-76: LGTM!The sort order handling follows the same pattern as partition handling, correctly building the hint dictionary.
78-82: LGTM!The validation ensures at least one of
partitionorsort_orderis specified, matching the test expectation intest_pyiceberg_adapter_requires_one_of_partition_or_sort_order.elt-common/tests/unit_tests/dlt_destinations/pyiceberg/test_helpers.py (4)
41-52: LGTM!Good coverage of both success and failure paths for
namespace_exists, with proper mocking of the catalog.
55-77: LGTM!Comprehensive coverage of type conversion including edge cases for unsupported types and decimal validation.
119-131: LGTM!Thorough validation of Iceberg schema creation including field count, identifier fields, and type mapping.
134-144: LGTM!Good validation of partition spec creation with correct source_id mapping and transform type verification.
elt-common/tests/e2e_tests/elt_common/dlt_destinations/pyiceberg/utils.py (4)
27-32: LGTM!Import updates correctly reflect the migration to builder-based partition and sort order configuration.
176-176: LGTM!Dynamic builder method invocation correctly adapts to the new
PartitionTrBuilderAPI.
237-251: LGTM!Correct usage of the
SortOrderBuilderfluent API with proper chaining and.build()calls.
301-303: LGTM!Improved assertion message will aid debugging when tests fail.
elt-common/src/elt_common/dlt_destinations/pyiceberg/pyiceberg.py (4)
35-47: LGTM!Import reorganisation aligns well with the new helper module structure.
65-70: LGTM!Correctly propagates
write_dispositionfrom the load table configuration, enabling per-table disposition support.
115-116: LGTM!Simple and effective check for DLT internal tables. Note that
_dlt_tables_prefixis a private attribute; verify this is stable across DLT versions.
370-375: LGTM!Clean integration with the new helper functions for schema and spec creation.
elt-common/tests/e2e_tests/elt_common/dlt_destinations/pyiceberg/test_pyiceberg_pipeline.py (5)
13-15: LGTM!Import updated to reflect the module reorganisation, with a clear alias.
38-49: LGTM!Well-designed factory pattern that provides flexibility for creating resources with or without primary keys, reducing test boilerplate.
173-219: LGTM!Thorough test of merge/upsert behaviour with a good volume of records to validate the operation at scale. The expected data calculation correctly accounts for updates and inserts.
222-245: LGTM!Good negative test ensuring unsupported merge strategies are properly rejected with informative error messages.
63-63: LGTM!Consistent migration to the
resource_factorypattern across all tests.elt-common/src/elt_common/dlt_destinations/pyiceberg/helpers.py (5)
1-42: Well-structured imports and constants.The use of
Finalfor hint constants and the bidirectional precision mapping dictionaries are clean and appropriate for the module's purpose.
117-150: Reverse type mapping is correctly implemented.The function properly handles nullable semantics and the hardcoded precision of 6 for timestamps aligns with Iceberg v1/v2 limitations documented elsewhere in the codebase.
153-171: Schema creation correctly tracks primary keys as identifier fields.This implementation properly supports the merge disposition by marking primary key columns as
identifier_field_ids, which is essential for the upsert merge strategy.
238-266: Builder pattern with fluent API is well-implemented.The
SortOrderBuilderproperly validates that a direction is specified before building, and the fluent API (asc()/desc()returningself) enables clean chaining.
303-320: Sort order creation correctly defaults to unsorted.The implementation properly falls back to
UNSORTED_SORT_ORDERwhen no hints are provided, and uses identity transform which aligns with the currentSortOrderBuildercapabilities.
|
Holding off on merging this so it can be tested with the catalog containing an existing table now that the identifier fields on the schema are set correctly. |
The required fields cannot be changed on the destination tables without allowing incompatible changes and we don't want to do this on the standard code paths through the dlt pipeline. Now that the |
1 participant
Summary
This correctly supports the dlt merge disposition for the custom pyiceberg destination. It also fixes a bug where the per-table write_disposition was ignored.
Fixes #165
Summary by CodeRabbit
Release Notes
New Features
Improvements
Refactoring
Tests
✏️ Tip: You can customize this high-level summary in your review settings.