Delegate manifest and manifest-list reading to iceberg-rust via pyiceberg-core #45

Activity

1. kevinjqliu commented on Sep 4, 2026

   

   kevinjqliu

   on Sep 4, 2026

   OwnerAuthor

   More actions

   Benchmark: Cython Avro reader vs pyiceberg-core manifest reader

   Setup: macOS on Apple Silicon, Python 3.13.12, pyiceberg 0.11.1 (Cython decoder_fast), pyiceberg-core 0.10.1 built from iceberg-rust main (commit 6fa8e0383), apache-avro 0.21.0. Manifests were written with PyIceberg's own write_manifest (V2, deflate, 12-column schema, identity partition, full column stats: sizes, value/null counts, lower and upper bounds on every column). Both paths read the same files in the same process. Numbers are best-of-N wall time, single thread. Script at the bottom.

   Paths measured:

   • cython: ManifestFile.fetch_manifest_entry(io), today's code path.
   • rust_parse: pyiceberg_core.manifest.read_manifest_entries(bs) only.
   • rust_convert: rust_parse + .entries() + rebuilding PyIceberg ManifestEntry / DataFile via from_args + inheritance. This is what the earlier PyIceberg PR did and is the only apples-to-apples number.

   1. End to end, as shipped (opt-level = "z")

   entries	file	cython (ms)	rust_parse (ms)	rust_convert (ms)	rust_convert / cython
   1,000	0.4 MB	11.2	57.1	70.5	6.3x
   10,000	3.6 MB	110.2	571.2	708.3	6.4x
   50,000	18 MB	656.6	2859.0	3563.5	5.4x
   200,000	73 MB	3094.0	11723.2	15515.4	5.0x

   Manifest lists:

   manifests	cython (ms)	rust_parse (ms)	rust_convert (ms)	rust_convert / cython
   100	0.57	0.36	0.70	1.2x
   1,000	3.04	2.88	6.50	2.1x
   10,000	29.09	28.66	63.85	2.2x

   2. Same, rebuilt with opt-level = 3

   The published wheel is built with opt-level = "z" (size). Changing only that flag:

   entries	cython (ms)	rust_parse (ms)	rust_convert (ms)	rust_convert / cython
   1,000	10.7	32.0	41.9	3.9x
   10,000	107.9	311.4	443.0	4.1x
   50,000	721.2	1733.4	2216.8	3.1x

   manifests	cython (ms)	rust_parse (ms)	rust_convert (ms)	rust_convert / cython
   100	0.61	0.20	0.54	0.9x
   1,000	2.89	1.45	4.58	1.6x
   10,000	26.28	13.87	44.85	1.7x

   opt-level = "z" costs about 1.6x to 1.8x on this workload.

   3. Where the time goes inside iceberg-rust (no Python), 10,000 entries, opt-level = 3

   Manifest::parse_avro split into its stages, measured in-crate:

   stage	ms	us / entry
   Avro container to generic apache_avro::Value tree (writer schema only)	129	12.9
   same, via Reader::with_schema(reader_schema) (adds Value::resolve)	296	29.6
   Value to _serde::ManifestEntryV2 (serde from_value)	52	5.2
   ManifestEntryV2 to ManifestEntry (Datum::try_from_bytes on bounds)	16	1.6
   Manifest::parse_avro total	363	36.3

   For comparison, the whole Cython path including building PyIceberg objects is about 11 us / entry.

   Observations

   • About 82% of the Rust parse is inside apache-avro before any iceberg-rust code runs. Decoding into the generic Value tree alone (12.9 us / entry) is already slower than the full Cython path.
   • Reader::with_schema more than doubles decode cost. apache-avro decodes with the writer schema and then, if the reader schema is not == the writer schema, walks the whole tree again with Value::resolve. iceberg-rust's generated reader schema never equals a PyIceberg (or Java) writer schema, so resolution always runs. The diff for this file: the reader adds the V3 fields (first_row_id, referenced_data_file, content_offset, content_size_in_bytes), the reader has equality_ids: array<int> where the writer has array<long>, and the partition field is ["null","int"] in the reader but required int in the writer.
   • Python-side conversion into PyIceberg objects (rust_convert minus rust_parse) costs about 10 us / entry at opt-level = 3, as much as the entire Cython path. A faster Rust parser alone cannot reach parity while every entry is still rebuilt as a DataFile in Python.
   • PyO3 getter overhead is small: reading every getter on every entry adds about 1 us / entry.

   Conclusions

   1. The regression is real and larger than the 2x originally reported: 5x to 6x slower end to end as shipped, 3x to 4x with opt-level = 3.
   2. It is not a PyO3 problem.
   3. Two independent fixes are needed, and both are required for parity:
      • iceberg-rust: stop going through apache_avro::Value + resolve + serde for manifest entries. Either decode directly from the writer schema into ManifestEntry, or at minimum skip with_schema resolution when the writer schema is compatible and apply projection defaults in the serde layer. This is the 30 us / entry bucket.
      • binding / PyIceberg: stop rebuilding one Python object per entry. Return Arrow (a RecordBatch in the manifest-entry schema) or construct the DataFile objects directly from Rust in one pass. This is the 10 us / entry bucket.
   4. Cheap win regardless: build the wheel with opt-level = 3. Worth re-checking the binary-size trade-off that motivated "z".

   Benchmark script (run inside bindings/python after make install: uv run --no-sync python bench_manifest.py 1000 10000 50000 200000)

   """Benchmark: PyIceberg Cython Avro reader vs pyiceberg_core (iceberg-rust) manifest reader.
   
   Generates V2 manifests / manifest lists with PyIceberg's own writer, then times:
     cython        : ManifestFile.fetch_manifest_entry(io)                (today's code path)
     rust_parse    : pyiceberg_core.manifest.read_manifest_entries(bs)     (Rust parse + PyManifest)
     rust_entries  : + .entries()                                           (materialize PyManifestEntry list)
     rust_touch    : + read every getter on every entry                     (pyo3 getter cost, no pyiceberg objects)
     rust_convert  : + build pyiceberg ManifestEntry/DataFile via from_args (apples-to-apples with cython)
   """
   from __future__ import annotations
   
   import gc
   import os
   import struct
   import sys
   import time
   from typing import Any, Callable
   
   from pyiceberg.io.pyarrow import PyArrowFileIO
   from pyiceberg.manifest import (
       DataFile,
       DataFileContent,
       FileFormat,
       ManifestContent,
       ManifestEntry,
       ManifestEntryStatus,
       ManifestFile,
       PartitionFieldSummary,
       _inherit_from_manifest,
       read_manifest_list,
       write_manifest,
       write_manifest_list,
   )
   from pyiceberg.partitioning import PartitionField, PartitionSpec
   from pyiceberg.schema import Schema
   from pyiceberg.transforms import IdentityTransform
   from pyiceberg.typedef import Record
   from pyiceberg.types import (
       BooleanType,
       DateType,
       DoubleType,
       FloatType,
       IntegerType,
       LongType,
       NestedField,
       StringType,
       TimestampType,
   )
   
   SCHEMA = Schema(
       NestedField(1, "id", IntegerType(), required=True),
       NestedField(2, "seq", LongType(), required=False),
       NestedField(3, "name", StringType(), required=False),
       NestedField(4, "price", DoubleType(), required=False),
       NestedField(5, "ratio", FloatType(), required=False),
       NestedField(6, "flag", BooleanType(), required=False),
       NestedField(7, "day", DateType(), required=False),
       NestedField(8, "ts", TimestampType(), required=False),
       NestedField(9, "c9", StringType(), required=False),
       NestedField(10, "c10", LongType(), required=False),
       NestedField(11, "c11", IntegerType(), required=False),
       NestedField(12, "c12", DoubleType(), required=False),
       schema_id=0,
       identifier_field_ids=[1],
   )
   SPEC = PartitionSpec(PartitionField(source_id=1, field_id=1000, transform=IdentityTransform(), name="id"), spec_id=0)
   SNAPSHOT_ID = 8744736658442914487
   COLS = list(range(1, 13))
   # spec-correct single-value serialization per column type (iceberg-rust decodes bounds into typed Datums)
   BOUND = {
       1: lambda v: struct.pack("<i", v),
       2: lambda v: struct.pack("<q", v),
       3: lambda v: f"name-{v:08d}".encode(),
       4: lambda v: struct.pack("<d", float(v)),
       5: lambda v: struct.pack("<f", float(v)),
       6: lambda v: b"\x01",
       7: lambda v: struct.pack("<i", v % 20000),
       8: lambda v: struct.pack("<q", 1_600_000_000_000_000 + v),
       9: lambda v: f"c9-{v:08d}".encode(),
       10: lambda v: struct.pack("<q", v),
       11: lambda v: struct.pack("<i", v),
       12: lambda v: struct.pack("<d", float(v)),
   }
   
   
   def make_entry(i: int) -> ManifestEntry:
       lower = {c: BOUND[c](i) for c in COLS}
       upper = {c: BOUND[c](i + 1000) for c in COLS}
       df = DataFile.from_args(
           content=DataFileContent.DATA,
           file_path=f"s3://bucket/warehouse/db/tbl/data/id={i % 100}/00000-{i}-d8a4223e-dc97-45a1-86e1-adaba6e8abd7-00001.parquet",
           file_format=FileFormat.PARQUET,
           partition=Record(i % 100),
           record_count=19513 + i,
           file_size_in_bytes=388872 + i,
           column_sizes={c: 100 + c for c in COLS},
           value_counts={c: 19513 for c in COLS},
           null_value_counts={c: 0 for c in COLS},
           nan_value_counts={},
           lower_bounds=lower,
           upper_bounds=upper,
           key_metadata=None,
           split_offsets=[4],
           equality_ids=None,
           sort_order_id=0,
       )
       return ManifestEntry.from_args(
           status=ManifestEntryStatus.ADDED,
           snapshot_id=SNAPSHOT_ID,
           sequence_number=None,
           file_sequence_number=None,
           data_file=df,
       )
   
   
   def gen_manifest(io: PyArrowFileIO, path: str, n: int) -> ManifestFile:
       if os.path.exists(path):
           return ManifestFile.from_args(
               manifest_path=path, manifest_length=os.path.getsize(path), partition_spec_id=0,
               content=ManifestContent.DATA, sequence_number=1, min_sequence_number=1,
               added_snapshot_id=SNAPSHOT_ID, added_files_count=n, existing_files_count=0, deleted_files_count=0,
               added_rows_count=0, existing_rows_count=0, deleted_rows_count=0, partitions=[], key_metadata=None,
           )
       t = time.perf_counter()
       with write_manifest(2, SPEC, SCHEMA, io.new_output(path), SNAPSHOT_ID, "deflate") as w:
           for i in range(n):
               w.add_entry(make_entry(i))
           mf = w.to_manifest_file()
       print(f"  generated {path} ({n} entries, {os.path.getsize(path)/1e6:.1f} MB) in {time.perf_counter()-t:.1f}s", file=sys.stderr)
       return mf
   
   
   def gen_manifest_list(io: PyArrowFileIO, path: str, n: int, template: ManifestFile) -> None:
       if os.path.exists(path):
           return
       files = []
       for i in range(n):
           files.append(ManifestFile.from_args(
               manifest_path=f"s3://bucket/warehouse/db/tbl/metadata/{i}-m0.avro", manifest_length=6000 + i,
               partition_spec_id=0, content=ManifestContent.DATA, sequence_number=i + 1, min_sequence_number=1,
               added_snapshot_id=SNAPSHOT_ID, added_files_count=10, existing_files_count=5, deleted_files_count=0,
               added_rows_count=1000, existing_rows_count=500, deleted_rows_count=0,
               partitions=[PartitionFieldSummary.from_args(contains_null=False, contains_nan=False,
                                                           lower_bound=struct.pack("<i", 0), upper_bound=struct.pack("<i", 99))],
               key_metadata=None,
           ))
       with write_manifest_list(2, io.new_output(path), SNAPSHOT_ID, None, 1, "deflate") as w:
           w.add_manifests(files)
   
   
   # ---------------------------------------------------------------- rust-side conversions
   def rust_convert_entries(manifest_file: ManifestFile, bs: bytes, discard_deleted: bool = True) -> list[ManifestEntry]:
       from pyiceberg_core import manifest as rm
   
       m = rm.read_manifest_entries(bs)
       out = []
       for e in m.entries():
           if discard_deleted and e.status == ManifestEntryStatus.DELETED:
               continue
           d = e.data_file
           df = DataFile.from_args(
               content=DataFileContent(d.content),
               file_path=d.file_path,
               file_format=FileFormat(d.file_format),
               partition=Record(*(p.value() if p is not None else None for p in d.partition)),
               record_count=d.record_count,
               file_size_in_bytes=d.file_size_in_bytes,
               column_sizes=d.column_sizes,
               value_counts=d.value_counts,
               null_value_counts=d.null_value_counts,
               nan_value_counts=d.nan_value_counts,
               lower_bounds=d.lower_bounds,
               upper_bounds=d.upper_bounds,
               key_metadata=d.key_metadata,
               split_offsets=d.split_offsets,
               equality_ids=d.equality_ids,
               sort_order_id=d.sort_order_id,
           )
           me = ManifestEntry.from_args(
               status=ManifestEntryStatus(e.status),
               snapshot_id=e.snapshot_id,
               sequence_number=e.sequence_number,
               file_sequence_number=e.file_sequence_number,
               data_file=df,
           )
           out.append(_inherit_from_manifest(me, manifest_file))
       return out
   
   
   def rust_touch_entries(bs: bytes) -> int:
       from pyiceberg_core import manifest as rm
   
       n = 0
       for e in rm.read_manifest_entries(bs).entries():
           d = e.data_file
           _ = (e.status, e.snapshot_id, e.sequence_number, e.file_sequence_number,
                d.content, d.file_path, d.file_format, [p.value() if p is not None else None for p in d.partition],
                d.record_count, d.file_size_in_bytes, d.column_sizes, d.value_counts, d.null_value_counts,
                d.nan_value_counts, d.lower_bounds, d.upper_bounds, d.key_metadata, d.split_offsets,
                d.equality_ids, d.sort_order_id)
           n += 1
       return n
   
   
   def rust_convert_list(bs: bytes) -> list[ManifestFile]:
       from pyiceberg_core import manifest as rm
   
       return [
           ManifestFile.from_args(
               manifest_path=m.manifest_path, manifest_length=m.manifest_length, partition_spec_id=m.partition_spec_id,
               content=ManifestContent(m.content), sequence_number=m.sequence_number, min_sequence_number=m.min_sequence_number,
               added_snapshot_id=m.added_snapshot_id, added_files_count=m.added_files_count,
               existing_files_count=m.existing_files_count, deleted_files_count=m.deleted_files_count,
               added_rows_count=m.added_rows_count, existing_rows_count=m.existing_rows_count,
               deleted_rows_count=m.deleted_rows_count,
               partitions=[PartitionFieldSummary.from_args(contains_null=p.contains_null, contains_nan=p.contains_nan,
                                                           lower_bound=p.lower_bound, upper_bound=p.upper_bound)
                           for p in m.partitions],
               key_metadata=m.key_metadata,
           )
           for m in rm.read_manifest_list(bs).entries()
       ]
   
   
   # ---------------------------------------------------------------- timing
   def timeit(fn: Callable[[], Any], repeats: int) -> tuple[float, float]:
       times = []
       for _ in range(repeats):
           gc.collect()
           t = time.perf_counter()
           fn()
           times.append(time.perf_counter() - t)
       times.sort()
       return times[0], times[len(times) // 2]
   
   
   def main() -> None:
       from pyiceberg_core import manifest as rm
   
       sizes = [int(s) for s in (sys.argv[1:] or ["1000", "10000", "50000", "200000"])]
       outdir = os.environ.get("BENCH_DIR", os.path.join(os.path.dirname(os.path.abspath(__file__)), "bench_data"))
       os.makedirs(outdir, exist_ok=True)
       io = PyArrowFileIO()
   
       print(f"python {sys.version.split()[0]}")
       import pyiceberg
       print(f"pyiceberg {pyiceberg.__version__}")
       from pyiceberg.avro import decoder
       print(f"decoder: {decoder.new_decoder(b'').__class__.__name__}")
       print()
   
       rows = []
       for n in sizes:
           path = os.path.join(outdir, f"manifest_v2_{n}.avro")
           mf = gen_manifest(io, path, n)
           bs = io.new_input(path).open().read()
           repeats = 7 if n <= 50_000 else 3
   
           # sanity: same number of entries and same first data file
           cy = mf.fetch_manifest_entry(io)
           ru = rust_convert_entries(mf, bs)
           assert len(cy) == len(ru) == n, (len(cy), len(ru), n)
           assert cy[0].data_file.file_path == ru[0].data_file.file_path
           assert cy[0].data_file.lower_bounds == ru[0].data_file.lower_bounds
           assert cy[0].sequence_number == ru[0].sequence_number, (cy[0].sequence_number, ru[0].sequence_number)
           del cy, ru
   
           res = {
               "read_bytes": timeit(lambda: io.new_input(path).open().read(), repeats),
               "cython": timeit(lambda: mf.fetch_manifest_entry(io), repeats),
               "rust_parse": timeit(lambda: rm.read_manifest_entries(bs), repeats),
               "rust_entries": timeit(lambda: rm.read_manifest_entries(bs).entries(), repeats),
               "rust_touch": timeit(lambda: rust_touch_entries(bs), repeats),
               "rust_convert": timeit(lambda: rust_convert_entries(mf, bs), repeats),
           }
           rows.append((n, os.path.getsize(path), res))
           print(f"manifest entries={n:>7}  file={os.path.getsize(path)/1e6:6.1f} MB   (min / median of {repeats}, ms)")
           for k, (mn, md) in res.items():
               print(f"    {k:<14} {mn*1e3:9.1f} / {md*1e3:9.1f}")
           print(f"    rust_convert / cython = {res['rust_convert'][0]/res['cython'][0]:.2f}x   rust_parse / cython = {res['rust_parse'][0]/res['cython'][0]:.2f}x")
           print()
   
       # manifest lists
       template = mf
       for n in [100, 1_000, 10_000]:
           path = os.path.join(outdir, f"manifest_list_v2_{n}.avro")
           gen_manifest_list(io, path, n, template)
           bs = io.new_input(path).open().read()
           cy = list(read_manifest_list(io.new_input(path)))
           ru = rust_convert_list(bs)
           assert len(cy) == len(ru) == n
           assert cy[-1].manifest_path == ru[-1].manifest_path
           res = {
               "cython": timeit(lambda: list(read_manifest_list(io.new_input(path))), 7),
               "rust_parse": timeit(lambda: rm.read_manifest_list(bs), 7),
               "rust_convert": timeit(lambda: rust_convert_list(bs), 7),
           }
           print(f"manifest-list entries={n:>6}  file={os.path.getsize(path)/1e3:6.1f} KB   (min / median of 7, ms)")
           for k, (mn, md) in res.items():
               print(f"    {k:<14} {mn*1e3:9.2f} / {md*1e3:9.2f}")
           print(f"    rust_convert / cython = {res['rust_convert'][0]/res['cython'][0]:.2f}x")
           print()
   
   
   if __name__ == "__main__":
       main()
2. kevinjqliu commented on Sep 4, 2026

   

   kevinjqliu

   on Sep 4, 2026

   Last edited by claude

   OwnerAuthor

   More actions

   Side finding: PyIceberg writes one Avro block per manifest entry

   While generating benchmark files it turned out that every manifest written by PyIceberg has exactly one record per Avro block. ManifestWriter.add_entry calls AvroOutputFile.write_block([entry]) for each entry (pyiceberg/manifest.py), and write_block encodes, compresses, and writes a sync marker for the list it is given. So a 10,000-entry manifest is 10,000 blocks, each with its own deflate stream and 16-byte sync marker. Java's DataFileWriter flushes blocks at 64 KB. The manifest-list writer is not affected: ManifestListWriter.add_manifests passes the whole list to a single write_block call.

   This hurts both size and read speed, for every reader. Same 10,000-entry manifest, re-encoded with fastavro at a 64 KB sync interval and otherwise identical (same writer schema and metadata):

   	1 record / block (PyIceberg today)	64 KB blocks
   blocks	10,000	73
   file size, deflate	3.6 MB	0.7 MB
   file size, uncompressed	5.0 MB	4.8 MB
   PyIceberg Cython read, deflate	11.1 us / entry	7.4 us / entry
   PyIceberg Cython read, uncompressed	8.3 us / entry	7.3 us / entry
   iceberg-rust apache-avro Value decode, deflate	12.1 us / entry	9.0 us / entry
   iceberg-rust apache-avro Value decode, uncompressed	7.9 us / entry	7.8 us / entry

   Deflate on one record at a time barely compresses (3.6 MB vs 5.0 MB raw) because each stream starts cold, and every reader pays per-block setup and inflate overhead per entry. With 64 KB blocks the same data is 5x smaller and about 1.5x faster to read for both decoders.

   Suggested fix in PyIceberg, independent of this issue: buffer entries in ManifestWriter and flush a block when the encoded size reaches a threshold (64 KB to match Java) or on close, instead of one write_block per add_entry. Readers are unaffected since the container format does not change. This also makes the "Rust vs Cython" comparison more representative of Java-written tables, which is why the benchmark comment above includes both block layouts.

   ---

   Generated by Claude Code
3. kevinjqliu commented on Sep 5, 2026

   

   kevinjqliu

   on Sep 5, 2026

   Last edited by claude

   OwnerAuthor

   More actions

   Deferred from the first iceberg-rust PR (reader parity only)

   The first PR upgrades apache-avro to 0.22 and replaces the Value-based manifest reader with the byte-level serde reader, keeping behaviour identical to the old reader. These improvements were built and measured during that work but are held back so the PR stays a pure replacement:

   • Match partition fields by field-id before name. Java sanitizes Avro field names that are not valid identifiers and records the original in a field-id attribute; the old reader (and the new one, for parity) matches by name and silently drops such values. The writer schema is available from the reader, so the mapping is one HashMap per file.
   • Accept equality_ids written as array<long>. PyIceberg still writes long (checked on main, 2026-09-07); the spec and other writers use int. The byte-level deserializer does not narrow integers, so this needs a small lenient element deserializer.
   • Error on an unknown manifest-list format-version. ManifestList::parse_avro currently reads 1 and 3 and falls back to V2 for anything else, including unrecognized values.
   • Spec-sized partition structs on the JSON path. The Avro reader produces one slot per partition field (Avro resolution always did); the JSON DataFile path still produces a struct sized to the record. Unifying them changes one round-trip test expectation.
   • Uuid as fixed[16]. On released iceberg-rust (apache-avro 0.21) uuid partition values cannot be written at all (#2913); with the 0.22 upgrade in this PR they are written as length-prefixed bytes (UuidSchema::Bytes, what 0.21 did for fixed + uuid); the spec asks for fixed[16]. Format-affecting, so it needs its own change and compatibility check.
   • Regenerate DEPENDENCIES.rust.tsv. The lists are stale repo-wide since the 0.10.0 release; regenerate in a dedicated change with the CI-matching cargo-deny 0.19.9.

   ---

   Generated by Claude Code
4. claude

   deleted a comment from kevinjqliu on Sep 7, 2026
5. claude

   deleted a comment from kevinjqliu on Sep 7, 2026