fix(warehouses/accelerator): Electricity ingest: Skip files that cannot be processed.

[martyngigg]

Summary

Catch errors loading the electricity consumption data. Users would rather some occasional gaps (that can be backfilled) rather than data loading that just stops.

Refs #166

Summary by CodeRabbit

Release Notes

• Bug Fixes

  • Enhanced error handling for merge operations with improved logging of failed data loads.
  • Improved exception handling during data ingestion to prevent silent failures.

• New Features

  • Extended merge configuration capabilities for incremental data synchronisation.

• Tests

  • Expanded test coverage for merge key configurations and error scenarios.

_{✏️ Tip: You can customize this high-level summary in your review settings.}

[coderabbitai]

Important

Review skipped

Auto incremental reviews are disabled on this repository.

Please check the settings in the CodeRabbit UI or the .coderabbit.yaml file in this repository. To trigger a single review, invoke the @coderabbitai review command.

You can disable this status message by setting the reviews.review_status to false in the CodeRabbit configuration file.

Walkthrough

This PR updates merge handling for Iceberg tables with improved error handling, introduces merge_key configuration support in tests with expanded coverage, and modifies exception handling and merge strategy in the electricity SharePoint data pipeline.

Changes

Cohort / File(s)	Summary
PyIceberg merge write-disposition elt-common/src/elt_common/dlt_destinations/pyiceberg/pyiceberg.py	Enhanced merge handling to compute join_columns from merge_key marked columns when identifier_field_ids are absent. Reads merge strategy from x-merge-strategy in schema. Updated upsert path with improved error handling: logs table data before re-raising ValueError exceptions.
PyIceberg test coverage elt-common/tests/e2e_tests/elt_common/dlt_destinations/pyiceberg/test_pyiceberg_pipeline.py	Introduced optional merge_key parameter to resource_factory. Adjusted decorator construction to conditionally pass primary_key and merge_key. Added test coverage for merge_key configurations and validation error when neither primary_key nor merge_key provided.
Electricity SharePoint pipeline warehouses/accelerator/extract_load/electricity_sharepoint/extract_and_load.py	Changed exception handling in read_as_dataframe to catch all exceptions, log with file metadata, and return None instead of propagating errors. Updated rdm_data resource decorator: changed merge_key from "file_name" to "DateTime" with explicit non-nullable column specification.

Poem

🐰 Hops and whiskers, data flows,
Merge keys dance where logic goes,
Iceberg schemas, clean and bright,
DateTime marks the path aright!
Share the love, no errors show—
Watch our pipeline gracefully grow!

Pre-merge checks

❌ Failed checks (1 warning)

Check name	Status	Explanation	Resolution
Docstring Coverage	⚠️ Warning	Docstring coverage is 20.00% which is insufficient. The required threshold is 80.00%.	You can run @coderabbitai generate docstrings to improve docstring coverage.

✅ Passed checks (2 passed)

Check name	Status	Explanation
Description Check	✅ Passed	Check skipped - CodeRabbit’s high-level summary is enabled.
Title check	✅ Passed	The title directly addresses the main change: error handling to skip erroneous files during electricity data ingest, which aligns with the primary modification in the pull request.

---

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.

❤️ Share

• X
• Mastodon
• Reddit
• LinkedIn

_{Comment @coderabbitai help to get the list of available commands and usage tips.}

[coderabbitai]

Actionable comments posted: 1

🧹 Nitpick comments (4)

warehouses/accelerator/extract_load/electricity_sharepoint/extract_and_load.py (1)

120-124: Consider using logging.exception for better diagnostics.

The broad exception catch aligns with the PR objective to skip problematic files. However, using LOGGER.exception() instead of LOGGER.error() would automatically include the stack trace, which is valuable for diagnosing the root cause of failures later.

         except Exception as exc:
-            LOGGER.error(
-                f"'Error loading {file_name} ({len(file_bytes)} bytes)'. File skipped.\nDetails: {str(exc)}"
-            )
+            LOGGER.exception(
+                f"Error loading {file_name} ({len(file_bytes)} bytes). File skipped."
+            )
             return None

elt-common/src/elt_common/dlt_destinations/pyiceberg/pyiceberg.py (1)

434-437: Consider the impact of logging full table data on error.

Logging the entire existing table data (table.scan().to_arrow()) could be problematic for large tables — this may cause performance issues, excessive log sizes, or even memory exhaustion. Consider limiting the output or only logging this at DEBUG level.

                 except ValueError as exc:
-                    logger.info(f"----- Existing data ------\n{table.scan().to_arrow()}")
-                    logger.info(f"----- New data ------\n{table_data}")
-                    raise exc
+                    logger.debug(f"----- Existing data (first 10 rows) ------\n{table.scan().to_arrow().slice(0, 10)}")
+                    logger.debug(f"----- New data (first 10 rows) ------\n{table_data.slice(0, 10)}")
+                    raise

elt-common/tests/e2e_tests/elt_common/dlt_destinations/pyiceberg/test_pyiceberg_pipeline.py (2)

36-38: Add type hint for merge_key parameter.

For consistency with primary_key, add a type hint to merge_key.

 def resource_factory(
-    data: List[Dict[str, Any]] | None = None, primary_key: str | None = "id", merge_key=None
+    data: List[Dict[str, Any]] | None = None, primary_key: str | None = "id", merge_key: str | None = None
 ):

---

254-255: Consider matching a more specific exception type.

Using pytest.raises(Exception) is quite broad. If the destination raises a specific exception (e.g., DestinationTerminalException or ValueError), matching it would make the test more precise and guard against passing due to unrelated errors.

📜 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

📥 Commits

Reviewing files that changed from the base of the PR and between 6ce2f1b and 1f39ed4.

📒 Files selected for processing (3)

• elt-common/src/elt_common/dlt_destinations/pyiceberg/pyiceberg.py (1 hunks)
• elt-common/tests/e2e_tests/elt_common/dlt_destinations/pyiceberg/test_pyiceberg_pipeline.py (5 hunks)
• warehouses/accelerator/extract_load/electricity_sharepoint/extract_and_load.py (2 hunks)

🧰 Additional context used

🧬 Code graph analysis (1)

elt-common/src/elt_common/dlt_destinations/pyiceberg/pyiceberg.py (1)

elt-common/tests/unit_tests/dlt_destinations/pyiceberg/test_helpers.py (1)

• dlt_schema (23-30)

🪛 Ruff (0.14.8)

warehouses/accelerator/extract_load/electricity_sharepoint/extract_and_load.py

120-120: Do not catch blind exception: Exception

(BLE001)

---

121-123: Use logging.exception instead of logging.error

Replace with exception

(TRY400)

---

122-122: Use explicit conversion flag

Replace with conversion flag

(RUF010)

🔇 Additional comments (3)

elt-common/src/elt_common/dlt_destinations/pyiceberg/pyiceberg.py (1)

414-421: LGTM! Join columns derived from merge_key when no primary key exists.

The logic correctly falls back to extracting columns marked with merge_key=True when the table schema has no identifier fields. This enables merge operations using either primary_key or merge_key configuration.

elt-common/tests/e2e_tests/elt_common/dlt_destinations/pyiceberg/test_pyiceberg_pipeline.py (2)

179-185: Good test coverage for both primary_key and merge_key scenarios.

The parameterisation ensures merge operations work correctly whether the identifier is specified via primary_key or merge_key.

---

236-256: Good addition of negative test case.

This test ensures that merge operations fail gracefully when neither primary_key nor merge_key is provided, which validates the error handling in the destination layer.