feat: Add tell() to OutputStream writers (#2998)

# Rationale for this change

Currently, PyIceberg writes one manifest per snapshot operation
regardless of manifest size. In order to eventually support this we need
to be able to track written bytes without closing the file, so that we
can roll to a new file once we hit target size.

We had some of this work done in #650, but we can keep this simple and
add writers as a follow up. The nice thing is that the underlying
streams we support already have a tell() method and we just need to
expose it.

With this change in the follow up we can do:

```
with write_manifest(...) as writer:
    writer.add_entry(entry)
    if writer.tell() >= target_file_size:
        # roll to new file
```

## Are these changes tested?

Yes, added a test :)

## Are there any user-facing changes?

No

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"