feat: Add tell() to OutputStream writers

Author: geruh

pyiceberg/avro/file.py

@@ -317,3 +317,6 @@ def write_block(self, objects: list[D]) -> None:
             self.encoder.write(block_content)
 
         self.encoder.write(self.sync_bytes)
+
+    def tell(self) -> int:
+        return self.output_stream.tell()

pyiceberg/io/__init__.py

@@ -140,6 +140,9 @@ class OutputStream(Protocol):  # pragma: no cover
     @abstractmethod
     def write(self, b: bytes) -> int: ...
 
+    @abstractmethod
+    def tell(self) -> int: ...
+
     @abstractmethod
     def close(self) -> None: ...
 

pyiceberg/manifest.py

@@ -1059,6 +1059,9 @@ def __exit__(
         self.closed = True
         self._writer.__exit__(exc_type, exc_value, traceback)
 
+    def tell(self) -> int:
+        return self._writer.tell()
+
     @abstractmethod
     def content(self) -> ManifestContent: ...
 

tests/utils/test_manifest.py

@@ -897,3 +897,38 @@ def test_manifest_cache_efficiency_with_many_overlapping_lists() -> None:
             if len(references) > 1:
                 for ref in references[1:]:
                     assert ref is references[0], f"All references to manifest {i} should be the same object instance"
+
+
+@pytest.mark.parametrize("format_version", [1, 2])
+def test_manifest_writer_tell(format_version: TableVersion) -> None:
+    io = load_file_io()
+    test_schema = Schema(NestedField(1, "foo", IntegerType(), False))
+
+    with TemporaryDirectory() as tmpdir:
+        output_file = io.new_output(f"{tmpdir}/test-manifest.avro")
+        with write_manifest(
+            format_version=format_version,
+            spec=UNPARTITIONED_PARTITION_SPEC,
+            schema=test_schema,
+            output_file=output_file,
+            snapshot_id=1,
+            avro_compression="null",
+        ) as writer:
+            initial_bytes = writer.tell()
+            data_file = DataFile.from_args(
+                content=DataFileContent.DATA,
+                file_path=f"{tmpdir}/data.parquet",
+                file_format=FileFormat.PARQUET,
+                partition=Record(),
+                record_count=100,
+                file_size_in_bytes=1000,
+            )
+            entry = ManifestEntry.from_args(
+                status=ManifestEntryStatus.ADDED,
+                snapshot_id=1,
+                data_file=data_file,
+            )
+            writer.add_entry(entry)
+            after_entry_bytes = writer.tell()
+
+            assert after_entry_bytes > initial_bytes, "Bytes should increase after adding entry"