DAT-4672: improve logging for commits

Author: MoIMC

pyiceberg/catalog/hive.py

@@ -15,7 +15,6 @@
 #  specific language governing permissions and limitations
 #  under the License.
 import getpass
-import logging
 import socket
 import time
 from types import TracebackType
@@ -77,6 +76,7 @@
     TableAlreadyExistsError,
     WaitingForLockException,
 )
+from pyiceberg.logger import get_logger
 from pyiceberg.partitioning import UNPARTITIONED_PARTITION_SPEC, PartitionSpec
 from pyiceberg.schema import Schema, SchemaVisitor, visit
 from pyiceberg.serializers import FromInputFile
@@ -136,7 +136,8 @@
 DEFAULT_LOCK_CHECK_MAX_WAIT_TIME = 60  # 1 min
 DEFAULT_LOCK_CHECK_RETRIES = 4
 
-logger = logging.getLogger(__name__)
+
+logger = get_logger(__name__)
 
 
 class _HiveClient:
@@ -464,27 +465,28 @@ def _create_lock_request(self, database_name: str, table_name: str) -> LockReque
 
     def _wait_for_lock(self, database_name: str, table_name: str, lockid: int, open_client: Client) -> LockResponse:
         def _on_wait_for_lock_fail(state: RetryCallState) -> None:
-            raise WaitingForLockException(
-                f"Failed after {state.attempt_number} attempts to wait on lock for {database_name}.{table_name}"
-            )
+            msg = f"Failed after {state.attempt_number} attempts to wait on lock for `{database_name}.{table_name}`"
+            logger.debug(msg)
+            raise WaitingForLockException(msg)
 
         @retry(
             retry=retry_if_exception_type(WaitingForLockException),
             wait=wait_exponential(multiplier=2, min=self._lock_check_min_wait_time, max=self._lock_check_max_wait_time),
             stop=stop_after_attempt(self._lock_check_retries),
-            before=lambda state: logger.warning(f"({state.attempt_number}) Waiting on lock for {database_name}.{table_name}..."),
+            before=lambda state: logger.debug(f"({state.attempt_number}) Waiting on lock for `{database_name}.{table_name}`..."),
             retry_error_callback=_on_wait_for_lock_fail,
         )
         def _do_wait_for_lock() -> LockResponse:
             response: LockResponse = open_client.check_lock(CheckLockRequest(lockid=lockid))
             if response.state == LockState.ACQUIRED:
+                logger.debug("Acquired lock after waiting.")
                 return response
             elif response.state == LockState.WAITING:
-                msg = f"Waiting on lock for {database_name}.{table_name}..."
-                logger.warning(msg)
-                raise WaitingForLockException(msg)
+                raise WaitingForLockException()
             else:
-                raise CommitFailedException(f"Failed to check lock for {database_name}.{table_name}, state: {response.state}")
+                raise CommitFailedException(
+                    f"Failed to check lock for {database_name}.{table_name}, lock state: {response.state}"
+                )
 
         return _do_wait_for_lock()
 
@@ -511,13 +513,14 @@ def commit_table(
         # https://github.com/apache/hive/blob/master/standalone-metastore/metastore-common/src/main/thrift/hive_metastore.thrift#L1232
         with self._client as open_client:
             lock: LockResponse = open_client.lock(self._create_lock_request(database_name, table_name))
-
             try:
                 if lock.state != LockState.ACQUIRED:
                     if lock.state == LockState.WAITING:
                         self._wait_for_lock(database_name, table_name, lock.lockid, open_client)
                     else:
-                        raise CommitFailedException(f"Failed to acquire lock for {table_identifier}, state: {lock.state}")
+                        raise CommitFailedException(f"Failed to acquire lock for {table_identifier}, lock state: {lock.state}")
+                else:
+                    logger.debug("Acquired lock on initial attempt.")
 
                 hive_table: Optional[HiveTable]
                 current_table: Optional[Table]
@@ -558,7 +561,7 @@ def commit_table(
                     )
                     self._create_hive_table(open_client, hive_table)
             except WaitingForLockException as e:
-                raise CommitFailedException(f"Failed to acquire lock for {table_identifier}, state: {lock.state}") from e
+                raise CommitFailedException(f"Failed to acquire lock for {table_identifier}, lock state: {lock.state}") from e
             finally:
                 open_client.unlock(UnlockRequest(lockid=lock.lockid))
 

pyiceberg/logger.py

@@ -0,0 +1,13 @@
+import logging
+import sys
+
+
+def get_logger(name: str) -> logging.Logger:
+    logger = logging.getLogger(name)
+    if not logger.hasHandlers():
+        handler = logging.StreamHandler(sys.stdout)
+        handler.setLevel(logging.DEBUG)
+        formatter = logging.Formatter("%(asctime)s | %(levelname)s | %(name)s | %(message)s")
+        handler.setFormatter(formatter)
+        logger.addHandler(handler)
+    return logger

pyiceberg/table/__init__.py

@@ -66,6 +66,7 @@
     rewrite_not,
 )
 from pyiceberg.io import FileIO, load_file_io
+from pyiceberg.logger import get_logger
 from pyiceberg.manifest import (
     POSITIONAL_DELETE_SCHEMA,
     DataFile,
@@ -253,7 +254,7 @@ class TableProperties:
     MIN_SNAPSHOTS_TO_KEEP_DEFAULT = 1
 
 
-logger = getLogger(__name__)
+logger = get_logger(__name__)
 
 
 class Transaction:
@@ -802,24 +803,22 @@ def commit_transaction(self) -> Table:
         def _before_attempt(state: RetryCallState):
             if state.attempt_number > 1:
                 namespace, table = self._table.name()
-                logger.debug(f"Refreshing metadata and operations for {namespace}.{table} and retrying transaction commit...")
+                logger.debug(
+                    f"({state.attempt_number - 1}) Refreshing metadata and operations for `{namespace}.{table}` and retrying transaction commit..."
+                )
                 self._table.refresh()
                 self._updates, self._requirements = (), ()
                 for op in self._snapshot_operations:
                     op._cleanup_commit_failure()
                     self._apply(*op._commit())
             logger.debug(f"Committing transaction...")
 
-        def _after_error(state: RetryCallState):
-            logger.debug(f'Encountered CommitFailedException: "{state.outcome.exception()}"...')
-
         @wraps(self.commit_transaction)
         @retry(
             wait=wait_random_exponential(min=min_wait_ms / 1000, max=max_wait_ms / 1000),
             stop=stop_after_attempt(num_retries),
             retry=retry_if_exception_type(CommitFailedException),
             before=_before_attempt,
-            after=_after_error,
             reraise=True,
         )
         def _commit_transaction():
@@ -1502,7 +1501,7 @@ def __init__(
         snapshot_id: Optional[int] = None,
         options: Properties = EMPTY_DICT,
         limit: Optional[int] = None,
-        bound_filter: Optional[BooleanExpression] = None
+        bound_filter: Optional[BooleanExpression] = None,
     ):
         self.table_metadata = table_metadata
         self.io = io