diff --git a/src/dlt_iceberg/destination_client.py b/src/dlt_iceberg/destination_client.py index 8e112ac..dde202c 100644 --- a/src/dlt_iceberg/destination_client.py +++ b/src/dlt_iceberg/destination_client.py @@ -5,7 +5,6 @@ hooks, enabling atomic commits of multiple files per table. """ -import logging import time import threading from collections import defaultdict @@ -15,7 +14,7 @@ import pyarrow as pa import pyarrow.parquet as pq -from dlt.common import pendulum +from dlt.common import logger, pendulum from dlt.common.configuration import configspec from dlt.common.destination import DestinationCapabilitiesContext, Destination from dlt.common.destination.client import ( @@ -55,9 +54,6 @@ from .merge_utils import build_primary_key_delete_filter from pyiceberg.io.pyarrow import schema_to_pyarrow -logger = logging.getLogger(__name__) - - # MODULE-LEVEL STATE: Accumulate files across client instances # Key: load_id, Value: {table_name: [(table_schema, file_path, arrow_table), ...]} # This works because dlt creates multiple client instances but they're all in the same process @@ -1037,10 +1033,11 @@ def complete_load(self, load_id: str) -> None: table_name=table_name, file_data=file_data, ) - except Exception as e: - logger.error( - f"Failed to commit files for table {identifier}: {e}", - exc_info=True, + except Exception: + # Use dlt's configured logger so failures raised from the + # complete_load hook are visible in normal pipeline logs. + logger.exception( + f"Failed to commit files for table {identifier}" ) raise else: diff --git a/tests/test_error_handling.py b/tests/test_error_handling.py index d9be02d..f739857 100644 --- a/tests/test_error_handling.py +++ b/tests/test_error_handling.py @@ -217,6 +217,39 @@ def create_test_parquet(self, temp_dir: str) -> str: pq.write_table(data, str(parquet_path)) return str(parquet_path) + def test_complete_load_logs_commit_exception_with_dlt_logger(self): + """Failures in dlt's completion hook must be visible in pipeline logs.""" + from dlt_iceberg import destination_client + from dlt_iceberg.destination_client import IcebergRestClient + + load_id = "failed-load" + destination_client._PENDING_FILES[load_id] = { + "events": [ + ( + {"name": "events", "write_disposition": "merge"}, + "events.parquet", + pa.table({"id": [1]}), + ) + ] + } + client = MagicMock() + client.config.namespace = "analytics" + client._get_catalog.return_value = MagicMock() + client._table_identifier.return_value = "analytics.events" + failure = RuntimeError("upsert exploded") + client._commit_table_files.side_effect = failure + + try: + with patch.object(destination_client.logger, "exception") as log_exception: + with pytest.raises(RuntimeError, match="upsert exploded"): + IcebergRestClient.complete_load(client, load_id) + + log_exception.assert_called_once_with( + "Failed to commit files for table analytics.events" + ) + finally: + destination_client._PENDING_FILES.pop(load_id, None) + def test_non_retryable_error_fails_immediately(self): """Test that non-retryable errors fail without retrying.""" from dlt_iceberg.destination import _iceberg_rest_handler