Surface atomic commit failures in dlt logs

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:

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