fix lint

Author: KazuhitoT

pyiceberg/table/update/schema.py

@@ -91,13 +91,13 @@ class UpdateSchema(UpdateTableMetadata["UpdateSchema"]):
     _case_sensitive: bool
 
     # Store user intent for retry support
-    _column_additions: list[tuple[str | tuple[str, ...], IcebergType, str | None, bool, L | None]]
+    _column_additions: list[tuple[str | tuple[str, ...], IcebergType, str | None, bool, Any]]
     _column_updates: list[tuple[str | tuple[str, ...], IcebergType | None, bool | None, str | None]]
     _column_deletions: list[str | tuple[str, ...]]
     _column_renames: list[tuple[str | tuple[str, ...], str]]
     _move_operations: list[tuple[str, str | tuple[str, ...], str | tuple[str, ...] | None]]
     _optional_columns: list[str | tuple[str, ...]]
-    _default_value_updates: list[tuple[str | tuple[str, ...], L | None]]
+    _default_value_updates: list[tuple[str | tuple[str, ...], Any]]
 
     def __init__(
         self,
@@ -167,8 +167,8 @@ def _reset_state(self) -> None:
         for path_from, new_name in self._column_renames:
             self._do_rename_column(path_from, new_name)
 
-        for path, field_type, required, doc in self._column_updates:
-            self._do_update_column(path, field_type, required, doc)
+        for upd_path, upd_field_type, upd_required, upd_doc in self._column_updates:
+            self._do_update_column(upd_path, upd_field_type, upd_required, upd_doc)
 
         for path in self._optional_columns:
             self._set_column_requirement(path, required=False)
@@ -255,7 +255,7 @@ def _do_add_column(
         required: bool,
         default_value: L | None,
     ) -> None:
-        """Internal method to add a column. Used by add_column and _reset_state."""
+        """Add a column to the schema. Used by add_column and _reset_state."""
         if isinstance(path, str):
             if "." in path:
                 raise ValueError(f"Cannot add column with ambiguous name: {path}, provide a tuple instead")
@@ -340,7 +340,7 @@ def delete_column(self, path: str | tuple[str, ...]) -> UpdateSchema:
         return self
 
     def _do_delete_column(self, path: str | tuple[str, ...]) -> None:
-        """Internal method to delete a column. Used by delete_column and _reset_state."""
+        """Delete a column from the schema. Used by delete_column and _reset_state."""
         name = (path,) if isinstance(path, str) else path
         full_name = ".".join(name)
 
@@ -381,7 +381,7 @@ def rename_column(self, path_from: str | tuple[str, ...], new_name: str) -> Upda
         return self
 
     def _do_rename_column(self, path_from: str | tuple[str, ...], new_name: str) -> None:
-        """Internal method to rename a column. Used by rename_column and _reset_state."""
+        """Rename a column in the schema. Used by rename_column and _reset_state."""
         path_from = ".".join(path_from) if isinstance(path_from, tuple) else path_from
         field_from = self._schema.find_field(path_from, self._case_sensitive)
 
@@ -547,7 +547,7 @@ def _do_update_column(
         required: bool | None,
         doc: str | None,
     ) -> None:
-        """Internal method to update a column. Used by update_column and _reset_state."""
+        """Update a column in the schema. Used by update_column and _reset_state."""
         path = (path,) if isinstance(path, str) else path
         full_name = ".".join(path)
 
@@ -638,7 +638,7 @@ def move_first(self, path: str | tuple[str, ...]) -> UpdateSchema:
         return self
 
     def _do_move_first(self, path: str | tuple[str, ...]) -> None:
-        """Internal method to move a field to first position. Used by move_first and _reset_state."""
+        """Move a field to first position. Used by move_first and _reset_state."""
         full_name = ".".join(path) if isinstance(path, tuple) else path
 
         field_id = self._find_for_move(full_name)
@@ -662,7 +662,7 @@ def move_before(self, path: str | tuple[str, ...], before_path: str | tuple[str,
         return self
 
     def _do_move_before(self, path: str | tuple[str, ...], before_path: str | tuple[str, ...]) -> None:
-        """Internal method to move a field before another. Used by move_before and _reset_state."""
+        """Move a field before another. Used by move_before and _reset_state."""
         full_name = ".".join(path) if isinstance(path, tuple) else path
         field_id = self._find_for_move(full_name)
 
@@ -700,7 +700,7 @@ def move_after(self, path: str | tuple[str, ...], after_name: str | tuple[str, .
         return self
 
     def _do_move_after(self, path: str | tuple[str, ...], after_name: str | tuple[str, ...]) -> None:
-        """Internal method to move a field after another. Used by move_after and _reset_state."""
+        """Move a field after another. Used by move_after and _reset_state."""
         full_name = ".".join(path) if isinstance(path, tuple) else path
 
         field_id = self._find_for_move(full_name)

pyiceberg/table/update/spec.py

@@ -86,7 +86,8 @@ def _init_state_from_metadata(self) -> None:
     def _reset_state(self) -> None:
         """Reset state for retry, rebuilding from refreshed metadata.
 
-        This is called on transaction retry to reapply the spec changes on top of the refreshed table metadata."""
+        This is called on transaction retry to reapply the spec changes on top of the refreshed table metadata.
+        """
         self._init_state_from_metadata()
         for source_column_name, transform, partition_field_name in self._field_additions:
             self._do_add_field(source_column_name, transform, partition_field_name)

tests/catalog/test_sql.py

@@ -39,7 +39,6 @@
     SqlCatalogBaseTable,
 )
 from pyiceberg.exceptions import (
-    CommitFailedException,
     NamespaceAlreadyExistsError,
     NamespaceNotEmptyError,
     NoSuchNamespaceError,

tests/integration/test_catalog.py

@@ -18,9 +18,9 @@
 import os
 from collections.abc import Generator
 from pathlib import Path, PosixPath
+from unittest.mock import patch
 
 import pytest
-from unittest.mock import patch
 
 from pyiceberg.catalog import Catalog, MetastoreCatalog, load_catalog
 from pyiceberg.catalog.hive import HiveCatalog
@@ -306,11 +306,7 @@ def test_update_schema_conflict(test_catalog: Catalog, test_schema: Schema, tabl
     identifier = (database_name, table_name)
 
     test_catalog.create_namespace(database_name)
-    table = test_catalog.create_table(
-        identifier,
-        test_schema,
-        properties={TableProperties.COMMIT_NUM_RETRIES: "1"}
-    )
+    table = test_catalog.create_table(identifier, test_schema, properties={TableProperties.COMMIT_NUM_RETRIES: "1"})
     assert test_catalog.table_exists(identifier)
 
     original_update = table.update_schema().add_column("new_col", LongType())
@@ -330,18 +326,16 @@ def test_update_schema_conflict(test_catalog: Catalog, test_schema: Schema, tabl
 
 @pytest.mark.integration
 @pytest.mark.parametrize("test_catalog", CATALOGS)
-def test_update_schema_conflict_with_retry(test_catalog: Catalog, test_schema: Schema, table_name: str, database_name: str) -> None:
+def test_update_schema_conflict_with_retry(
+    test_catalog: Catalog, test_schema: Schema, table_name: str, database_name: str
+) -> None:
     if isinstance(test_catalog, HiveCatalog):
         pytest.skip("HiveCatalog fails in this test, need to investigate")
 
     identifier = (database_name, table_name)
 
     test_catalog.create_namespace(database_name)
-    table = test_catalog.create_table(
-        identifier,
-        test_schema,
-        properties={TableProperties.COMMIT_NUM_RETRIES: "2"}
-    )
+    table = test_catalog.create_table(identifier, test_schema, properties={TableProperties.COMMIT_NUM_RETRIES: "2"})
     assert test_catalog.table_exists(identifier)
 
     original_update = table.update_schema().add_column("new_col", LongType())
@@ -361,7 +355,7 @@ def mock_commit(
         concurrent_update.commit()
         original_update.commit()
 
-    assert commit_count == 3 # concurrent_update, original_update(fail), retry original_update(success)
+    assert commit_count == 3  # concurrent_update, original_update(fail), retry original_update(success)
 
     table = test_catalog.load_table(identifier)
     expected_schema = Schema(
@@ -578,10 +572,7 @@ def test_update_table_spec_conflict(test_catalog: Catalog, test_schema: Schema,
     test_catalog.create_namespace(database_name)
     spec = PartitionSpec(PartitionField(source_id=1, field_id=1000, transform=BucketTransform(16), name="id_bucket"))
     table = test_catalog.create_table(
-        identifier,
-        test_schema,
-        partition_spec=spec,
-        properties={TableProperties.COMMIT_NUM_RETRIES: "1"}
+        identifier, test_schema, partition_spec=spec, properties={TableProperties.COMMIT_NUM_RETRIES: "1"}
     )
 
     update = table.update_spec()
@@ -603,15 +594,14 @@ def test_update_table_spec_conflict(test_catalog: Catalog, test_schema: Schema,
 
 @pytest.mark.integration
 @pytest.mark.parametrize("test_catalog", CATALOGS)
-def test_update_table_spec_conflict_with_retry(test_catalog: Catalog, test_schema: Schema, table_name: str, database_name: str) -> None:
+def test_update_table_spec_conflict_with_retry(
+    test_catalog: Catalog, test_schema: Schema, table_name: str, database_name: str
+) -> None:
     identifier = (database_name, table_name)
     test_catalog.create_namespace(database_name)
     spec = PartitionSpec(PartitionField(source_id=1, field_id=1000, transform=BucketTransform(16), name="id_bucket"))
     table = test_catalog.create_table(
-        identifier,
-        test_schema,
-        partition_spec=spec,
-        properties={TableProperties.COMMIT_NUM_RETRIES: "2"}
+        identifier, test_schema, partition_spec=spec, properties={TableProperties.COMMIT_NUM_RETRIES: "2"}
     )
     update = table.update_spec()
     update.add_field(source_column_name="tpep_pickup_datetime", transform=BucketTransform(16), partition_field_name="shard")

tests/integration/test_writes/test_optimistic_concurrency.py

@@ -15,10 +15,11 @@
 # specific language governing permissions and limitations
 # under the License.
 
+from unittest.mock import patch
+
 import pyarrow as pa
 import pytest
 from pyspark.sql import SparkSession
-from unittest.mock import patch
 
 from pyiceberg.catalog import Catalog
 from pyiceberg.exceptions import CommitFailedException
@@ -34,7 +35,12 @@ def test_conflict_delete_delete(
 ) -> None:
     """This test should start passing once optimistic concurrency control has been implemented."""
     identifier = "default.test_conflict"
-    tbl1 = _create_table(session_catalog, identifier, {"format-version": format_version, TableProperties.COMMIT_NUM_RETRIES: "1"}, [arrow_table_with_null])
+    tbl1 = _create_table(
+        session_catalog,
+        identifier,
+        {"format-version": format_version, TableProperties.COMMIT_NUM_RETRIES: "1"},
+        [arrow_table_with_null],
+    )
     tbl2 = session_catalog.load_table(identifier)
 
     tbl1.delete("string == 'z'")
@@ -51,7 +57,12 @@ def test_conflict_delete_delete_with_retry(
 ) -> None:
     """This test should start passing once optimistic concurrency control has been implemented."""
     identifier = "default.test_conflict"
-    tbl1 = _create_table(session_catalog, identifier, {"format-version": format_version, TableProperties.COMMIT_NUM_RETRIES: "2"}, [arrow_table_with_null])
+    tbl1 = _create_table(
+        session_catalog,
+        identifier,
+        {"format-version": format_version, TableProperties.COMMIT_NUM_RETRIES: "2"},
+        [arrow_table_with_null],
+    )
     tbl2 = session_catalog.load_table(identifier)
 
     tbl1.delete("string == 'z'")
@@ -73,14 +84,20 @@ def mock_commit(
 
     assert commit_count == 2
 
+
 @pytest.mark.integration
 @pytest.mark.parametrize("format_version", [1, 2])
 def test_conflict_delete_append(
     spark: SparkSession, session_catalog: Catalog, arrow_table_with_null: pa.Table, format_version: int
 ) -> None:
     """This test should start passing once optimistic concurrency control has been implemented."""
     identifier = "default.test_conflict"
-    tbl1 = _create_table(session_catalog, identifier, {"format-version": format_version, TableProperties.COMMIT_NUM_RETRIES: "1"}, [arrow_table_with_null])
+    tbl1 = _create_table(
+        session_catalog,
+        identifier,
+        {"format-version": format_version, TableProperties.COMMIT_NUM_RETRIES: "1"},
+        [arrow_table_with_null],
+    )
     tbl2 = session_catalog.load_table(identifier)
 
     # This is allowed
@@ -98,7 +115,12 @@ def test_conflict_delete_append_with_retry(
 ) -> None:
     """This test should start passing once optimistic concurrency control has been implemented."""
     identifier = "default.test_conflict"
-    tbl1 = _create_table(session_catalog, identifier, {"format-version": format_version, TableProperties.COMMIT_NUM_RETRIES: "2"}, [arrow_table_with_null])
+    tbl1 = _create_table(
+        session_catalog,
+        identifier,
+        {"format-version": format_version, TableProperties.COMMIT_NUM_RETRIES: "2"},
+        [arrow_table_with_null],
+    )
     tbl2 = session_catalog.load_table(identifier)
 
     # This is allowed
@@ -129,7 +151,12 @@ def test_conflict_append_delete(
 ) -> None:
     """This test should start passing once optimistic concurrency control has been implemented."""
     identifier = "default.test_conflict"
-    tbl1 = _create_table(session_catalog, identifier, {"format-version": format_version, TableProperties.COMMIT_NUM_RETRIES: "1"}, [arrow_table_with_null])
+    tbl1 = _create_table(
+        session_catalog,
+        identifier,
+        {"format-version": format_version, TableProperties.COMMIT_NUM_RETRIES: "1"},
+        [arrow_table_with_null],
+    )
     tbl2 = session_catalog.load_table(identifier)
 
     tbl1.append(arrow_table_with_null)
@@ -146,7 +173,12 @@ def test_conflict_append_delete_with_retry(
 ) -> None:
     """This test should start passing once optimistic concurrency control has been implemented."""
     identifier = "default.test_conflict"
-    tbl1 = _create_table(session_catalog, identifier, {"format-version": format_version, TableProperties.COMMIT_NUM_RETRIES: "2"}, [arrow_table_with_null])
+    tbl1 = _create_table(
+        session_catalog,
+        identifier,
+        {"format-version": format_version, TableProperties.COMMIT_NUM_RETRIES: "2"},
+        [arrow_table_with_null],
+    )
     tbl2 = session_catalog.load_table(identifier)
 
     tbl1.append(arrow_table_with_null)
@@ -176,7 +208,12 @@ def test_conflict_append_append(
 ) -> None:
     """This test should start passing once optimistic concurrency control has been implemented."""
     identifier = "default.test_conflict"
-    tbl1 = _create_table(session_catalog, identifier, {"format-version": format_version, TableProperties.COMMIT_NUM_RETRIES: "1"}, [arrow_table_with_null])
+    tbl1 = _create_table(
+        session_catalog,
+        identifier,
+        {"format-version": format_version, TableProperties.COMMIT_NUM_RETRIES: "1"},
+        [arrow_table_with_null],
+    )
     tbl2 = session_catalog.load_table(identifier)
 
     tbl1.append(arrow_table_with_null)
@@ -193,7 +230,12 @@ def test_conflict_append_append_with_retry(
 ) -> None:
     """This test should start passing once optimistic concurrency control has been implemented."""
     identifier = "default.test_conflict"
-    tbl1 = _create_table(session_catalog, identifier, {"format-version": format_version, TableProperties.COMMIT_NUM_RETRIES: "2"}, [arrow_table_with_null])
+    tbl1 = _create_table(
+        session_catalog,
+        identifier,
+        {"format-version": format_version, TableProperties.COMMIT_NUM_RETRIES: "2"},
+        [arrow_table_with_null],
+    )
     tbl2 = session_catalog.load_table(identifier)
 
     tbl1.append(arrow_table_with_null)

tests/table/test_commit_retry.py

@@ -1024,9 +1024,7 @@ def mock_commit(
 
         with patch.object(catalog, "commit_table", side_effect=mock_commit):
             with table.update_spec() as update_spec:
-                update_spec.add_field(
-                    source_column_name="id", transform=BucketTransform(16), partition_field_name="id_bucket"
-                )
+                update_spec.add_field(source_column_name="id", transform=BucketTransform(16), partition_field_name="id_bucket")
 
         assert commit_count == 2