dev: bring ty to parity with mypy and suppress what it reports today

.github/workflows/python-ci.yml

@@ -61,6 +61,11 @@ jobs:
     - name: Run linters
       if: matrix.python == '3.11'
       run: make lint
+    - name: Type-check with ty
+      # Not the gate yet, so a failure reports without failing the job.
+      if: matrix.python == '3.11'
+      continue-on-error: true
+      run: uv run ty check
     - name: Run unit tests with coverage
       run: COVERAGE=1 make test
     - name: Generate coverage report (85%) # Coverage threshold should only increase over time — never decrease it!

.pre-commit-config.yaml

@@ -40,13 +40,43 @@ repos:
     rev: v1.18.2
     hooks:
       - id: mypy
+        # Scoped to the package while ty and mypy are compared on the same files.
+        files: ^pyiceberg/
         args:
           [--config=pyproject.toml]
         additional_dependencies:
           - types-cachetools
           - types-pytest-lazy-fixture
           - types-pytz
           - types-requests
+  # Experimental: ty is not the gate yet, so this hook is manual-stage only.
+  # Run it with: uv run prek run ty --stage manual -a
+  # Local hook so ty runs in this hook's own environment, which holds only ty and the stub
+  # packages, the same bare environment the mypy hook gets. The entry points at it through
+  # VIRTUAL_ENV, which the hook runner exports. Keep these pins in step with the types group.
+  - repo: local
+    hooks:
+      - id: ty
+        name: ty
+        entry: >-
+          ty check
+          -c 'rules.unresolved-import = "ignore"'
+          -c 'rules.unused-ignore-comment = "ignore"'
+          -c 'rules.unused-type-ignore-comment = "ignore"'
+          -c 'rules.invalid-assignment = "ignore"'
+          -c 'rules.invalid-argument-type = "ignore"'
+          -c 'rules.call-non-callable = "ignore"'
+        language: python
+        additional_dependencies:
+          - ty==0.0.84
+          - types-cachetools==7.0.0.20260713
+          - types-pytest-lazy-fixture==0.6.3.20260518
+          - types-pytz==2026.5.0.20261006
+          - types-requests==2.33.0.20261006
+        always_run: true
+        pass_filenames: false
+        require_serial: true
+        stages: [manual]
   - repo: https://github.com/igorshubovych/markdownlint-cli
     rev: v0.45.0
     hooks:

Makefile

@@ -18,6 +18,7 @@
         test test-integration test-integration-setup test-integration-exec test-integration-cleanup test-integration-rebuild \
         test-s3 test-adls test-gcs test-coverage coverage-report test test-notebook\
         docs-serve docs-build notebook notebook-infra \
+        typecheck-ty \
         clean
 
 .DEFAULT_GOAL := help
@@ -97,6 +98,9 @@ check-license: ## Check license headers
 lint: ## Run code linters via prek (pre-commit hooks)
 	uv run $(PYTHON_ARG) prek run -a
 
+typecheck-ty: ## Type-check with ty in its isolated stub-only hook environment
+	uv run $(PYTHON_ARG) prek run ty --stage manual -a
+
 # ===============
 # Testing Section
 # ===============

pyiceberg/avro/codecs/bzip2.py

@@ -33,7 +33,7 @@ def decompress(data: bytes) -> bytes:
 
 except ImportError:
 
-    class BZip2Codec(Codec):  # type: ignore
+    class BZip2Codec(Codec):  # type: ignore[no-redef]
         @staticmethod
         def compress(data: bytes) -> tuple[bytes, int]:
             raise ImportError("Python bzip2 support not installed, please install the extension")

pyiceberg/avro/codecs/snappy_codec.py

@@ -54,12 +54,12 @@ def decompress(data: bytes) -> bytes:
             checksum = data[-4:]  # store checksum before truncating data
             data = data[0:-4]  # remove checksum from the data
             uncompressed = snappy.decompress(data)
-            SnappyCodec._check_crc32(uncompressed, checksum)
+            SnappyCodec._check_crc32(uncompressed, checksum)  # ty: ignore[unresolved-attribute]
             return uncompressed
 
 except ImportError:
 
-    class SnappyCodec(Codec):  # type: ignore
+    class SnappyCodec(Codec):  # type: ignore[no-redef]
         @staticmethod
         def compress(data: bytes) -> tuple[bytes, int]:
             raise ImportError("Snappy support not installed, please install using `pip install pyiceberg[snappy]`")

pyiceberg/avro/codecs/zstandard_codec.py

@@ -43,7 +43,7 @@ def decompress(data: bytes) -> bytes:
 
 except ImportError:
 
-    class ZStandardCodec(Codec):  # type: ignore
+    class ZStandardCodec(Codec):  # type: ignore[no-redef]
         @staticmethod
         def compress(data: bytes) -> tuple[bytes, int]:
             raise ImportError("Zstandard support not installed, please install using `pip install pyiceberg[zstandard]`")

pyiceberg/avro/decoder.py

@@ -137,7 +137,7 @@ def __init__(self, input_stream: bytes | InputStream) -> None:
         """Reader is a Python object on which we can call read, seek, and tell."""
         if isinstance(input_stream, bytes):
             # In the case of bytes, we wrap it into a BytesIO to make it a stream
-            self._input_stream = io.BytesIO(input_stream)
+            self._input_stream = io.BytesIO(input_stream)  # ty: ignore[invalid-assignment]
         else:
             self._input_stream = input_stream
 

pyiceberg/avro/decoder_fast.pyi

@@ -33,10 +33,10 @@ class CythonBinaryDecoder(BinaryDecoder):
     def read_int(self) -> int:
         pass
 
-    def read_ints(self, count: int) -> tuple[int, ...]:
+    def read_ints(self, count: int) -> tuple[int, ...]:  # ty: ignore[invalid-method-override]
         pass
 
-    def read_int_bytes_dict(self, count: int, dest: dict[int, bytes]) -> None:
+    def read_int_bytes_dict(self, count: int, dest: dict[int, bytes]) -> None:  # ty: ignore[invalid-method-override]
         pass
 
     def read_bytes(self) -> bytes:

pyiceberg/avro/file.py

@@ -96,7 +96,7 @@ def compression_codec(self) -> type[Codec] | None:
         if codec_name not in KNOWN_CODECS:
             raise ValueError(f"Unsupported codec: {codec_name}")
 
-        return KNOWN_CODECS[codec_name]  # type: ignore
+        return KNOWN_CODECS[codec_name]  # type: ignore[index]
 
     def get_schema(self) -> Schema:
         if _SCHEMA_KEY in self.meta:
@@ -257,7 +257,7 @@ def __init__(
             raise ValueError(f"Sync interval must be positive: {sync_interval}")
         self.sync_interval = sync_interval
         self._block = io.BytesIO()
-        self._block_encoder = BinaryEncoder(output_stream=self._block)
+        self._block_encoder = BinaryEncoder(output_stream=self._block)  # ty: ignore[invalid-argument-type]
         self._block_records = 0
 
     def __enter__(self) -> AvroOutputFile[D]:
@@ -308,7 +308,7 @@ def compression_codec(self) -> type[Codec] | None:
         if codec_name not in KNOWN_CODECS:
             raise ValueError(f"Unsupported codec: {codec_name}")
 
-        return KNOWN_CODECS[codec_name]  # type: ignore
+        return KNOWN_CODECS[codec_name]  # type: ignore[index]
 
     def write_block(self, objects: list[D]) -> None:
         for obj in objects:

pyiceberg/avro/reader.py

@@ -93,7 +93,7 @@ def __repr__(self) -> str:
 
 
 class NoneReader(Reader):
-    def read(self, _: BinaryDecoder) -> None:
+    def read(self, _: BinaryDecoder) -> None:  # ty: ignore[invalid-method-override]
         return None
 
     def skip(self, decoder: BinaryDecoder) -> None:
@@ -107,7 +107,7 @@ class DefaultReader(Reader):
     def __init__(self, default_value: Any) -> None:
         self.default_value = default_value
 
-    def read(self, _: BinaryDecoder) -> Any:
+    def read(self, _: BinaryDecoder) -> Any:  # ty: ignore[invalid-method-override]
         return self.default_value
 
     def skip(self, decoder: BinaryDecoder) -> None:

pyiceberg/avro/resolver.py

@@ -185,7 +185,7 @@ def visit_time(self, time_type: TimeType) -> Writer:
     def visit_timestamp(self, timestamp_type: TimestampType) -> Writer:
         return TimestampWriter()
 
-    def visit_timestamp_ns(self, timestamp_ns_type: TimestampNanoType) -> Writer:
+    def visit_timestamp_ns(self, timestamp_ns_type: TimestampNanoType) -> Writer:  # ty: ignore[invalid-method-override]
         return TimestampNanoWriter()
 
     def visit_timestamptz(self, timestamptz_type: TimestamptzType) -> Writer:
@@ -233,7 +233,7 @@ def resolve_writer(
     """
     if record_schema == file_schema:
         return construct_writer(file_schema)
-    return visit_with_partner(file_schema, record_schema, WriteSchemaResolver(), SchemaPartnerAccessor())  # type: ignore
+    return visit_with_partner(file_schema, record_schema, WriteSchemaResolver(), SchemaPartnerAccessor())  # type: ignore[misc]
 
 
 def resolve_reader(
@@ -253,7 +253,7 @@ def resolve_reader(
     Raises:
         NotImplementedError: If attempting to resolve an unrecognized object type.
     """
-    return visit_with_partner(file_schema, read_schema, ReadSchemaResolver(read_types, read_enums), SchemaPartnerAccessor())  # type: ignore
+    return visit_with_partner(file_schema, read_schema, ReadSchemaResolver(read_types, read_enums), SchemaPartnerAccessor())  # type: ignore[misc]
 
 
 class EnumReader(Reader):
@@ -276,10 +276,10 @@ def skip(self, decoder: BinaryDecoder) -> None:
 
 
 class WriteSchemaResolver(PrimitiveWithPartnerVisitor[IcebergType, Writer]):
-    def schema(self, file_schema: Schema, record_schema: IcebergType | None, result: Writer) -> Writer:
+    def schema(self, file_schema: Schema, record_schema: IcebergType | None, result: Writer) -> Writer:  # ty: ignore[invalid-method-override]
         return result
 
-    def struct(self, file_schema: StructType, record_struct: IcebergType | None, file_writers: list[Writer]) -> Writer:
+    def struct(self, file_schema: StructType, record_struct: IcebergType | None, file_writers: list[Writer]) -> Writer:  # ty: ignore[invalid-method-override]
         if not isinstance(record_struct, StructType):
             raise ResolveError(f"File/write schema are not aligned for struct, got {record_struct}")
 
@@ -301,16 +301,16 @@ def struct(self, file_schema: StructType, record_struct: IcebergType | None, fil
 
         return StructWriter(field_writers=tuple(results))
 
-    def field(self, file_field: NestedField, record_type: IcebergType | None, field_writer: Writer) -> Writer:
+    def field(self, file_field: NestedField, record_type: IcebergType | None, field_writer: Writer) -> Writer:  # ty: ignore[invalid-method-override]
         return field_writer if file_field.required else OptionWriter(field_writer)
 
-    def list(self, file_list_type: ListType, file_list: IcebergType | None, element_writer: Writer) -> Writer:
+    def list(self, file_list_type: ListType, file_list: IcebergType | None, element_writer: Writer) -> Writer:  # ty: ignore[invalid-method-override]
         return ListWriter(element_writer if file_list_type.element_required else OptionWriter(element_writer))
 
-    def map(self, file_map_type: MapType, file_primitive: IcebergType | None, key_writer: Writer, value_writer: Writer) -> Writer:
+    def map(self, file_map_type: MapType, file_primitive: IcebergType | None, key_writer: Writer, value_writer: Writer) -> Writer:  # ty: ignore[invalid-method-override]
         return MapWriter(key_writer, value_writer if file_map_type.value_required else OptionWriter(value_writer))
 
-    def primitive(self, file_primitive: PrimitiveType, record_primitive: IcebergType | None) -> Writer:
+    def primitive(self, file_primitive: PrimitiveType, record_primitive: IcebergType | None) -> Writer:  # ty: ignore[invalid-method-override]
         if record_primitive is not None:
             # ensure that the type can be projected to the expected
             if file_primitive != record_primitive:
@@ -393,16 +393,16 @@ def __init__(
         self.read_enums = read_enums
         self.context = []
 
-    def schema(self, schema: Schema, expected_schema: IcebergType | None, result: Reader) -> Reader:
+    def schema(self, schema: Schema, expected_schema: IcebergType | None, result: Reader) -> Reader:  # ty: ignore[invalid-method-override]
         return result
 
-    def before_field(self, field: NestedField, field_partner: NestedField | None) -> None:
+    def before_field(self, field: NestedField, field_partner: NestedField | None) -> None:  # ty: ignore[invalid-method-override]
         self.context.append(field.field_id)
 
-    def after_field(self, field: NestedField, field_partner: NestedField | None) -> None:
+    def after_field(self, field: NestedField, field_partner: NestedField | None) -> None:  # ty: ignore[invalid-method-override]
         self.context.pop()
 
-    def struct(self, struct: StructType, expected_struct: IcebergType | None, field_readers: list[Reader]) -> Reader:
+    def struct(self, struct: StructType, expected_struct: IcebergType | None, field_readers: list[Reader]) -> Reader:  # ty: ignore[invalid-method-override]
         read_struct_id = self.context[STRUCT_ROOT] if len(self.context) > 0 else STRUCT_ROOT
         struct_callable = self.read_types.get(read_struct_id, Record)
 
@@ -439,22 +439,22 @@ def struct(self, struct: StructType, expected_struct: IcebergType | None, field_
 
         return StructReader(tuple(results), struct_callable, expected_struct)
 
-    def field(self, field: NestedField, expected_field: IcebergType | None, field_reader: Reader) -> Reader:
+    def field(self, field: NestedField, expected_field: IcebergType | None, field_reader: Reader) -> Reader:  # ty: ignore[invalid-method-override]
         return field_reader if field.required else OptionReader(field_reader)
 
-    def list(self, list_type: ListType, expected_list: IcebergType | None, element_reader: Reader) -> Reader:
+    def list(self, list_type: ListType, expected_list: IcebergType | None, element_reader: Reader) -> Reader:  # ty: ignore[invalid-method-override]
         if expected_list and not isinstance(expected_list, ListType):
             raise ResolveError(f"File/read schema are not aligned for list, got {expected_list}")
 
         return ListReader(element_reader if list_type.element_required else OptionReader(element_reader))
 
-    def map(self, map_type: MapType, expected_map: IcebergType | None, key_reader: Reader, value_reader: Reader) -> Reader:
+    def map(self, map_type: MapType, expected_map: IcebergType | None, key_reader: Reader, value_reader: Reader) -> Reader:  # ty: ignore[invalid-method-override]
         if expected_map and not isinstance(expected_map, MapType):
             raise ResolveError(f"File/read schema are not aligned for map, got {expected_map}")
 
         return MapReader(key_reader, value_reader if map_type.value_required else OptionReader(value_reader))
 
-    def primitive(self, primitive: PrimitiveType, expected_primitive: IcebergType | None) -> Reader:
+    def primitive(self, primitive: PrimitiveType, expected_primitive: IcebergType | None) -> Reader:  # ty: ignore[invalid-method-override]
         if expected_primitive is not None:
             if not isinstance(expected_primitive, PrimitiveType):
                 raise ResolveError(f"File/read schema are not aligned for {primitive}, got {expected_primitive}")
@@ -547,7 +547,7 @@ def schema_partner(self, partner: IcebergType | None) -> IcebergType | None:
 
         raise ResolveError(f"File/read schema are not aligned for schema, got {partner}")
 
-    def field_partner(self, partner: IcebergType | None, field_id: int, field_name: str) -> IcebergType | None:
+    def field_partner(self, partner: IcebergType | None, field_id: int, field_name: str) -> IcebergType | None:  # ty: ignore[invalid-method-override]
         if isinstance(partner, StructType):
             field = partner.field(field_id)
         else:

pyiceberg/avro/writer.py

@@ -234,5 +234,5 @@ class DefaultWriter(Writer):
     writer: Writer
     value: Any
 
-    def write(self, encoder: BinaryEncoder, _: Any) -> None:
+    def write(self, encoder: BinaryEncoder, _: Any) -> None:  # ty: ignore[invalid-method-override]
         self.writer.write(encoder, self.value)

pyiceberg/catalog/__init__.py

@@ -919,7 +919,7 @@ def _convert_schema_if_needed(
 
             downcast_ns_timestamp_to_us = Config().get_bool(DOWNCAST_NS_TIMESTAMP_TO_US_ON_WRITE) or False
             if isinstance(schema, pa.Schema):
-                schema: Schema = visit_pyarrow(  # type: ignore
+                schema: Schema = visit_pyarrow(  # type: ignore[no-redef]
                     schema,
                     _ConvertToIcebergWithoutIDs(
                         downcast_ns_timestamp_to_us=downcast_ns_timestamp_to_us, format_version=format_version
@@ -1077,7 +1077,7 @@ def _create_staged_table(
         Returns:
             StagedTable: the created staged table instance.
         """
-        schema: Schema = self._convert_schema_if_needed(  # type: ignore
+        schema: Schema = self._convert_schema_if_needed(  # type: ignore[no-redef]
             schema,
             int(properties.get(TableProperties.FORMAT_VERSION, TableProperties.DEFAULT_FORMAT_VERSION)),  # type: ignore
         )

pyiceberg/catalog/bigquery_metastore.py

@@ -131,7 +131,7 @@ def create_table(
             ValueError: If the identifier is invalid, or no path is given to store metadata.
 
         """
-        schema: Schema = self._convert_schema_if_needed(schema)  # type: ignore
+        schema: Schema = self._convert_schema_if_needed(schema)  # type: ignore[no-redef]
 
         dataset_name, table_name = self.identifier_to_database_and_table(identifier)