diff --git a/bindings/python/src/manifest.rs b/bindings/python/src/manifest.rs index 15b80c3e3e..e22c7c6933 100644 --- a/bindings/python/src/manifest.rs +++ b/bindings/python/src/manifest.rs @@ -17,10 +17,17 @@ use std::sync::Arc; +use arrow::array::RecordBatch; +use arrow::datatypes::SchemaRef; +use arrow::error::ArrowError; +use arrow::pyarrow::IntoPyArrow; +use arrow::record_batch::{RecordBatchIterator, RecordBatchReader}; use iceberg::spec::{ - FieldSummary, FormatVersion, Manifest, ManifestEntry, ManifestFile, ManifestList, + FieldSummary, FormatVersion, Manifest, ManifestContentType, ManifestEntry, ManifestFile, + ManifestList, }; use pyo3::prelude::*; +use pyo3::pybacked::PyBackedBytes; use crate::data_file::PyDataFile; use crate::error::to_py_err; @@ -226,11 +233,98 @@ pub fn read_manifest_list(bs: &[u8]) -> PyResult { }) } +/// Reads the entries of a manifest as a `pyarrow.RecordBatchReader`. +/// +/// The other arguments are the manifest's entry in the manifest list, which +/// the entries inherit snapshot IDs, sequence numbers, and first row IDs from. +/// Raises `ValueError` for an invalid manifest header. An error while reading +/// raises `pyarrow.ArrowInvalid`, which subclasses `ValueError`. The reader +/// holds a reference to `bs` until it's closed or collected, and the next call +/// into `pyiceberg_core` releases it. +#[pyfunction] +#[pyo3(signature = ( + bs, + *, + manifest_path, + manifest_length, + partition_spec_id, + content, + sequence_number, + min_sequence_number, + added_snapshot_id, + first_row_id = None, +))] +#[allow(clippy::too_many_arguments)] +pub fn read_manifest_entries_as_arrow( + py: Python<'_>, + bs: PyBackedBytes, + manifest_path: String, + manifest_length: i64, + partition_spec_id: i32, + content: i32, + sequence_number: i64, + min_sequence_number: i64, + added_snapshot_id: i64, + first_row_id: Option, +) -> PyResult> { + // Inheritance reads none of the optional fields left as None. + let manifest_file = ManifestFile { + manifest_path, + manifest_length, + partition_spec_id, + content: ManifestContentType::try_from(content).map_err(to_py_err)?, + sequence_number, + min_sequence_number, + added_snapshot_id, + added_files_count: None, + existing_files_count: None, + deleted_files_count: None, + added_rows_count: None, + existing_rows_count: None, + deleted_rows_count: None, + partitions: None, + key_metadata: None, + first_row_id, + }; + let reader = iceberg::arrow::read_manifest_entries(bs, &manifest_file).map_err(to_py_err)?; + into_pyarrow_reader(py, reader.schema(), reader) +} + +/// Reads the entries of a manifest list as a `pyarrow.RecordBatchReader`. +/// +/// Raises `ValueError` for an invalid manifest list header. An error while +/// reading raises `pyarrow.ArrowInvalid`, which subclasses `ValueError`. The +/// reader holds a reference to `bs` until it's closed or collected, and the +/// next call into `pyiceberg_core` releases it. +#[pyfunction] +pub fn read_manifest_list_as_arrow( + py: Python<'_>, + bs: PyBackedBytes, +) -> PyResult> { + let reader = iceberg::arrow::read_manifest_list(bs).map_err(to_py_err)?; + into_pyarrow_reader(py, reader.schema(), reader) +} + +/// Exports `batches` through the Arrow C stream interface. +fn into_pyarrow_reader<'py>( + py: Python<'py>, + schema: SchemaRef, + batches: impl Iterator> + Send + 'static, +) -> PyResult> { + let batches = + batches.map(|batch| batch.map_err(|err| ArrowError::ExternalError(Box::new(err)))); + let reader: Box = + Box::new(RecordBatchIterator::new(batches, schema)); + reader.into_pyarrow(py) +} + pub fn register_module(py: Python<'_>, m: &Bound<'_, PyModule>) -> PyResult<()> { let this = PyModule::new(py, "manifest")?; this.add_function(wrap_pyfunction!(read_manifest_entries, &this)?)?; this.add_function(wrap_pyfunction!(read_manifest_list, &this)?)?; + this.add_function(wrap_pyfunction!(read_manifest_entries_as_arrow, &this)?)?; + this.add_function(wrap_pyfunction!(read_manifest_list_as_arrow, &this)?)?; m.add_submodule(&this)?; py.import("sys")? diff --git a/bindings/python/tests/test_manifest.py b/bindings/python/tests/test_manifest.py index 147b81fccc..14c2bc5e45 100644 --- a/bindings/python/tests/test_manifest.py +++ b/bindings/python/tests/test_manifest.py @@ -15,9 +15,12 @@ # specific language governing permissions and limitations # under the License. +import struct +import sys from pathlib import Path from typing import Any +import pyarrow as pa import pytest from pyiceberg.io import FileIO from pyiceberg.io.pyarrow import PyArrowFileIO @@ -25,12 +28,20 @@ DataFile, DataFileContent, FileFormat, + ManifestContent, ManifestEntry, ManifestEntryStatus, ManifestFile, _manifests, + 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 LongType, NestedField, StringType @pytest.fixture(autouse=True) @@ -235,3 +246,244 @@ def test_manifest_file_without_partition_summaries(tmp_path: Path) -> None: manifest_file = manifest.read_manifest_list(bs).entries()[0] assert manifest_file.partitions is None + + +SNAPSHOT_ID = 25 + + +def _write_manifest(tmp_path: Path, num_files: int) -> tuple[bytes, ManifestFile]: + """Writes a v2 manifest whose entries inherit their sequence numbers.""" + schema = Schema( + NestedField(1, "id", LongType(), required=False), + NestedField(2, "name", StringType(), required=False), + ) + spec = PartitionSpec( + PartitionField( + source_id=1, field_id=1000, transform=IdentityTransform(), name="id" + ), + spec_id=0, + ) + path = str(tmp_path / "manifest.avro") + io = PyArrowFileIO() + with write_manifest( + format_version=2, + spec=spec, + schema=schema, + output_file=io.new_output(path), + snapshot_id=SNAPSHOT_ID, + avro_compression="deflate", + ) as writer: + for i in range(num_files): + writer.add( + ManifestEntry.from_args( + status=ManifestEntryStatus.ADDED, + snapshot_id=SNAPSHOT_ID, + data_file=DataFile.from_args( + content=DataFileContent.DATA, + file_path=f"s3://bucket/data/{i}.parquet", + file_format=FileFormat.PARQUET, + partition=Record(i), + record_count=10 + i, + file_size_in_bytes=100, + column_sizes={1: 8, 2: 16}, + value_counts={1: 10 + i, 2: 10 + i}, + null_value_counts={1: 0, 2: i}, + nan_value_counts={}, + lower_bounds={1: struct.pack(" pa.RecordBatchReader: + from pyiceberg_core import manifest + + return manifest.read_manifest_entries_as_arrow( + bs, + manifest_path=manifest_file.manifest_path, + manifest_length=manifest_file.manifest_length, + partition_spec_id=manifest_file.partition_spec_id, + content=manifest_file.content.value, + sequence_number=manifest_file.sequence_number, + min_sequence_number=manifest_file.min_sequence_number, + added_snapshot_id=manifest_file.added_snapshot_id, + first_row_id=first_row_id, + ) + + +def test_read_manifest_entries_as_arrow_matches_pyiceberg(tmp_path: Path) -> None: + bs, manifest_file = _write_manifest(tmp_path, 3) + reader = _read_entries(bs, manifest_file) + assert isinstance(reader, pa.RecordBatchReader) + + rows = reader.read_all().to_pylist() + expected = manifest_file.fetch_manifest_entry( + PyArrowFileIO(), discard_deleted=False + ) + assert [ + { + "status": row["status"], + "snapshot_id": row["snapshot_id"], + "sequence_number": row["sequence_number"], + "file_sequence_number": row["file_sequence_number"], + "file_path": row["data_file"]["file_path"], + "partition": row["data_file"]["partition"]["id"], + "record_count": row["data_file"]["record_count"], + "null_value_counts": dict(row["data_file"]["null_value_counts"]), + "lower_bounds": dict(row["data_file"]["lower_bounds"]), + } + for row in rows + ] == [ + { + "status": entry.status.value, + "snapshot_id": entry.snapshot_id, + "sequence_number": entry.sequence_number, + "file_sequence_number": entry.file_sequence_number, + "file_path": entry.data_file.file_path, + "partition": entry.data_file.partition[0], + "record_count": entry.data_file.record_count, + "null_value_counts": entry.data_file.null_value_counts, + "lower_bounds": entry.data_file.lower_bounds, + } + for entry in expected + ] + + +def test_read_manifest_entries_as_arrow_yields_batches_of_1024_rows( + tmp_path: Path, +) -> None: + bs, manifest_file = _write_manifest(tmp_path, 2500) + assert [batch.num_rows for batch in _read_entries(bs, manifest_file)] == [ + 1024, + 1024, + 452, + ] + + +def test_read_manifest_entries_as_arrow_releases_input_on_close( + tmp_path: Path, +) -> None: + from pyiceberg_core import transform + + bs, manifest_file = _write_manifest(tmp_path, 3) + refcount = sys.getrefcount(bs) + reader = _read_entries(bs, manifest_file) + assert sys.getrefcount(bs) == refcount + 1 + + # pyarrow releases the stream without PyO3 attached to the interpreter, so + # PyO3 defers the decref to its next call. + reader.close() + transform.identity(pa.array([1])) + assert sys.getrefcount(bs) == refcount + + +def test_read_manifest_entries_as_arrow_batches_outlive_reader( + tmp_path: Path, +) -> None: + bs, manifest_file = _write_manifest(tmp_path, 3) + reader = _read_entries(bs, manifest_file) + batch = reader.read_next_batch() + reader.close() + del reader, bs + + assert batch.column("data_file").field("file_path").to_pylist() == [ + f"s3://bucket/data/{i}.parquet" for i in range(3) + ] + + +def test_read_manifest_entries_as_arrow_raises_while_reading(tmp_path: Path) -> None: + bs, manifest_file = _write_manifest(tmp_path, 3) + # Assigning first row IDs to the second entry overflows i64. + reader = _read_entries(bs, manifest_file, first_row_id=2**63 - 15) + + with pytest.raises(pa.ArrowInvalid, match="row ID overflow"): + reader.read_next_batch() + + +@pytest.mark.parametrize( + "payload", [b"", b"not-an-avro-file", b"Obj\x01truncated-after-the-avro-magic"] +) +def test_read_manifest_entries_as_arrow_raises_on_invalid_avro( + tmp_path: Path, payload: bytes +) -> None: + _, manifest_file = _write_manifest(tmp_path, 1) + with pytest.raises(ValueError): + _read_entries(payload, manifest_file) + + +@pytest.mark.parametrize("format_version", [1, 2]) +def test_read_manifest_list_as_arrow_matches_pyiceberg( + tmp_path: Path, format_version: int +) -> None: + from pyiceberg_core import manifest + + path = str(tmp_path / "snap.avro") + io = PyArrowFileIO() + with write_manifest_list( + format_version=format_version, + output_file=io.new_output(path), + snapshot_id=SNAPSHOT_ID, + parent_snapshot_id=None, + sequence_number=7, + avro_compression="deflate", + ) as writer: + writer.add_manifests( + [ + ManifestFile.from_args( + manifest_path=f"s3://bucket/metadata/{i}.avro", + manifest_length=1024, + partition_spec_id=0, + added_snapshot_id=SNAPSHOT_ID, + sequence_number=-1, + min_sequence_number=-1, + added_files_count=i, + existing_files_count=0, + deleted_files_count=0, + added_rows_count=10 * i, + existing_rows_count=0, + deleted_rows_count=0, + ) + for i in range(2) + ] + ) + + reader = manifest.read_manifest_list_as_arrow(io.new_input(path).open().read()) + assert isinstance(reader, pa.RecordBatchReader) + rows = reader.read_all().to_pylist() + expected = list(read_manifest_list(io.new_input(path))) + fields = [ + "manifest_path", + "sequence_number", + "min_sequence_number", + "added_snapshot_id", + "added_files_count", + "added_rows_count", + ] + assert [{f: row[f] for f in fields} for row in rows] == [ + {f: getattr(manifest_file, f) for f in fields} for manifest_file in expected + ] + + +@pytest.mark.parametrize( + "payload", [b"", b"not-an-avro-file", b"Obj\x01truncated-after-the-avro-magic"] +) +def test_read_manifest_list_as_arrow_raises_on_invalid_avro(payload: bytes) -> None: + from pyiceberg_core import manifest + + with pytest.raises(ValueError): + manifest.read_manifest_list_as_arrow(payload) diff --git a/crates/iceberg/public-api.txt b/crates/iceberg/public-api.txt index c1c737e9f2..dd8df4d659 100644 --- a/crates/iceberg/public-api.txt +++ b/crates/iceberg/public-api.txt @@ -79,6 +79,18 @@ pub fn iceberg::arrow::ArrowReaderBuilder::with_range_coalesce_bytes(self, range pub fn iceberg::arrow::ArrowReaderBuilder::with_range_fetch_concurrency(self, range_fetch_concurrency: usize) -> Self pub fn iceberg::arrow::ArrowReaderBuilder::with_row_group_filtering_enabled(self, row_group_filtering_enabled: bool) -> Self pub fn iceberg::arrow::ArrowReaderBuilder::with_row_selection_enabled(self, row_selection_enabled: bool) -> Self +pub struct iceberg::arrow::ManifestEntryBatchReader> +impl> iceberg::arrow::ManifestEntryBatchReader +pub fn iceberg::arrow::ManifestEntryBatchReader::schema(&self) -> arrow_schema::schema::SchemaRef +impl> core::iter::traits::iterator::Iterator for iceberg::arrow::ManifestEntryBatchReader +pub type iceberg::arrow::ManifestEntryBatchReader::Item = core::result::Result +pub fn iceberg::arrow::ManifestEntryBatchReader::next(&mut self) -> core::option::Option +pub struct iceberg::arrow::ManifestListBatchReader> +impl> iceberg::arrow::ManifestListBatchReader +pub fn iceberg::arrow::ManifestListBatchReader::schema(&self) -> arrow_schema::schema::SchemaRef +impl> core::iter::traits::iterator::Iterator for iceberg::arrow::ManifestListBatchReader +pub type iceberg::arrow::ManifestListBatchReader::Item = core::result::Result +pub fn iceberg::arrow::ManifestListBatchReader::next(&mut self) -> core::option::Option pub struct iceberg::arrow::PartitionValueCalculator impl iceberg::arrow::PartitionValueCalculator pub fn iceberg::arrow::PartitionValueCalculator::calculate(&self, batch: &arrow_array::record_batch::RecordBatch) -> iceberg::Result @@ -131,8 +143,8 @@ pub fn iceberg::arrow::arrow_schema_to_schema_auto_assign_ids(schema: &arrow_sch pub fn iceberg::arrow::arrow_struct_to_literal(struct_array: &arrow_array::array::ArrayRef, ty: &iceberg::spec::StructType) -> iceberg::Result>> pub fn iceberg::arrow::arrow_type_to_type(ty: &arrow_schema::datatype::DataType) -> iceberg::Result pub fn iceberg::arrow::datum_to_arrow_type_with_ree(datum: &iceberg::spec::Datum) -> arrow_schema::datatype::DataType -pub fn iceberg::arrow::read_manifest_entries(bytes: &[u8], manifest_file: &iceberg::spec::ManifestFile) -> iceberg::Result -pub fn iceberg::arrow::read_manifest_list(bytes: &[u8]) -> iceberg::Result +pub fn iceberg::arrow::read_manifest_entries>(bytes: B, manifest_file: &iceberg::spec::ManifestFile) -> iceberg::Result> +pub fn iceberg::arrow::read_manifest_list>(bytes: B) -> iceberg::Result> pub fn iceberg::arrow::schema_to_arrow_schema(schema: &iceberg::spec::Schema) -> iceberg::Result pub fn iceberg::arrow::strip_metadata_from_schema(schema: &arrow_schema::schema::Schema) -> iceberg::Result pub fn iceberg::arrow::type_to_arrow_type(ty: &iceberg::spec::Type) -> iceberg::Result diff --git a/crates/iceberg/src/arrow/manifest.rs b/crates/iceberg/src/arrow/manifest.rs index 08bd9ed4cd..437397d1e7 100644 --- a/crates/iceberg/src/arrow/manifest.rs +++ b/crates/iceberg/src/arrow/manifest.rs @@ -18,6 +18,7 @@ //! Reading manifests and manifest lists into Arrow. use std::collections::HashMap; +use std::io::Cursor; use std::sync::Arc; use arrow_arith::boolean::{and, is_null}; @@ -32,7 +33,7 @@ use arrow_avro::reader::{Reader, ReaderBuilder}; use arrow_avro::schema::AvroSchema; use arrow_cast::{CastOptions, cast_with_options}; use arrow_ord::cmp::eq; -use arrow_schema::{ArrowError, DataType, Field, Fields, TimeUnit}; +use arrow_schema::{ArrowError, DataType, Field, Fields, SchemaRef, TimeUnit}; use arrow_select::zip::zip; use parquet::arrow::PARQUET_FIELD_ID_META_KEY; use serde_json::Value as JsonValue; @@ -48,14 +49,18 @@ use crate::spec::{ }; use crate::{Error, Result}; -/// Reads the entries of a manifest file into a [`RecordBatch`] with one row -/// per entry. +/// Rows per batch. A row's size grows with the table's column count through +/// the column stats maps. With stats on every column of a 100-column table of +/// longs, 1024 rows take about 9 MB. +const BATCH_SIZE: usize = 1024; + +/// Reads the entries of a manifest file as batches with one row per entry. /// /// `bytes` is the manifest file and `manifest_file` is its entry in the -/// manifest list. The schema is the spec's `manifest_entry` struct for the -/// latest format version, with field IDs in the Arrow field metadata, so -/// manifests of every format version read into the same schema apart from -/// `data_file.partition`, which follows the manifest's partition spec. +/// manifest list. The schema is the spec's v3 `manifest_entry` struct, with +/// field IDs in the Arrow field metadata, so v1 to v3 manifests read into the +/// same schema apart from `data_file.partition`, which follows the manifest's +/// partition spec. /// /// Fields are matched to the manifest's Avro schema by field ID. A v1 /// manifest reads with the spec's v2+ defaults, and null snapshot IDs, @@ -63,6 +68,12 @@ use crate::{Error, Result}; /// Lower and upper bounds keep the binary single-value serialization they were /// written with. /// +/// # Errors +/// +/// Returns an error if the manifest's header can't be read or names an +/// unsupported format version. Errors in the entries come from the reader, +/// which yields no batches after one. +/// /// # Example /// /// ``` @@ -92,82 +103,161 @@ use crate::{Error, Result}; /// first_row_id: None, /// }; /// -/// let batch = read_manifest_entries(&bytes, &manifest_file)?; -/// assert_eq!(batch.num_rows(), 2); +/// let mut rows = 0; +/// for batch in read_manifest_entries(&bytes, &manifest_file)? { +/// rows += batch?.num_rows(); +/// } +/// assert_eq!(rows, 2); /// # Ok::<(), Box>(()) /// ``` -pub fn read_manifest_entries(bytes: &[u8], manifest_file: &ManifestFile) -> Result { - // usize::MAX decodes the whole manifest into one batch. First row ID - // inheritance is a running sum over every entry in the file, and an empty - // manifest still returns its schema. arrow-avro sizes its buffers - // independently of the batch size, so this doesn't preallocate. - let builder = || ReaderBuilder::new().with_batch_size(usize::MAX); - let mut reader = builder().build(bytes).map_err(invalid_manifest)?; - let header = reader.avro_header(); - let metadata: HashMap> = header - .metadata() - .map(|(key, value)| (String::from_utf8_lossy(key).into_owned(), value.to_vec())) - .collect(); - let metadata = ManifestMetadata::parse(&metadata)?; - let mut avro_schema: JsonValue = serde_json::from_slice( - header - .get("avro.schema") - .ok_or_else(|| invalid_data!("manifest has no avro.schema"))?, - )?; - let partition_type = metadata - .partition_spec() - .partition_type(metadata.schema())?; - let schema = Arc::new(schema_to_arrow_schema( - &Schema::builder() - .with_fields(manifest_entry_fields(&partition_type)) - .build()?, - )?); - - // TODO(#3257): arrow-avro 59 can't decode a record with no fields, which - // is the partition of every unpartitioned manifest, so project it out and - // build it in `missing_field`. arrow-avro 60 fixes this - // (apache/arrow-rs#10771). - if remove_empty_partition(&mut avro_schema) { - reader = builder() - .with_reader_schema(AvroSchema::new(avro_schema.to_string())) - .build(bytes) +pub fn read_manifest_entries>( + bytes: B, + manifest_file: &ManifestFile, +) -> Result> { + ManifestEntryBatchReader::try_new(bytes, manifest_file, BATCH_SIZE) +} + +/// Batches of the entries of a manifest file, returned by +/// [`read_manifest_entries`]. +pub struct ManifestEntryBatchReader> { + reader: Reader>, + schema: SchemaRef, + avro_schema: JsonValue, + zero_defaults: &'static [i32], + inheritance: Inheritance, + finished: bool, +} + +/// The values that a manifest's entries inherit from the manifest list. +struct Inheritance { + manifest_path: String, + added_snapshot_id: i64, + sequence_number: i64, + content: ManifestContentType, + /// The first row ID of the next entry that inherits one, which advances + /// across batches. `None` for a manifest without a first row ID. + next_row_id: Option, +} + +impl> ManifestEntryBatchReader { + fn try_new(bytes: B, manifest_file: &ManifestFile, batch_size: usize) -> Result { + // TODO(#3257): read the header from the reader below once the + // workaround for empty partitions is gone. The reader can't give + // `bytes` back, so the header is read from a borrowing reader first. + let (metadata, mut avro_schema) = { + let header_reader = ReaderBuilder::new() + .build(bytes.as_ref()) + .map_err(invalid_manifest)?; + let header = header_reader.avro_header(); + let metadata: HashMap> = header + .metadata() + .map(|(key, value)| (String::from_utf8_lossy(key).into_owned(), value.to_vec())) + .collect(); + let avro_schema: JsonValue = serde_json::from_slice( + header + .get("avro.schema") + .ok_or_else(|| invalid_data!("manifest has no avro.schema"))?, + )?; + (ManifestMetadata::parse(&metadata)?, avro_schema) + }; + let partition_type = metadata + .partition_spec() + .partition_type(metadata.schema())?; + let schema = Arc::new(schema_to_arrow_schema( + &Schema::builder() + .with_fields(manifest_entry_fields(&partition_type)) + .build()?, + )?); + + // TODO(#3257): arrow-avro 59 can't decode a record with no fields, which + // is the partition of every unpartitioned manifest, so project it out and + // build it in `missing_field`. arrow-avro 60 fixes this + // (apache/arrow-rs#10771). + let mut builder = ReaderBuilder::new().with_batch_size(batch_size); + if remove_empty_partition(&mut avro_schema) { + builder = builder.with_reader_schema(AvroSchema::new(avro_schema.to_string())); + } + let reader = builder + .build(Cursor::new(bytes)) .map_err(invalid_manifest)?; + + // v1 manifests have no content or sequence number columns, which v2+ + // readers read as 0. A v2+ manifest may lack file_sequence_number (Java + // wrote none before 1.1.0), which reads as null and is inherited. + let zero_defaults: &[i32] = match metadata.format_version() { + FormatVersion::V1 => &[ + DATA_FILE_CONTENT_FIELD_ID, + SEQUENCE_NUMBER_FIELD_ID, + FILE_SEQUENCE_NUMBER_FIELD_ID, + ], + _ => &[DATA_FILE_CONTENT_FIELD_ID], + }; + Ok(Self { + reader, + schema, + avro_schema, + zero_defaults, + inheritance: Inheritance { + manifest_path: manifest_file.manifest_path.clone(), + added_snapshot_id: manifest_file.added_snapshot_id, + sequence_number: manifest_file.sequence_number, + content: manifest_file.content, + next_row_id: manifest_file.first_row_id.map(i64::try_from).transpose()?, + }, + finished: false, + }) } - let Some(batch) = only_batch(reader, invalid_manifest)? else { - return Ok(RecordBatch::new_empty(schema)); - }; - // v1 manifests have no content or sequence number columns, which v2+ - // readers read as 0. A v2+ manifest may lack file_sequence_number (Java - // wrote none before 1.1.0), which reads as null and is inherited. - let zero_defaults: &[i32] = match metadata.format_version() { - FormatVersion::V1 => &[ - DATA_FILE_CONTENT_FIELD_ID, - SEQUENCE_NUMBER_FIELD_ID, - FILE_SEQUENCE_NUMBER_FIELD_ID, - ], - _ => &[DATA_FILE_CONTENT_FIELD_ID], - }; - let entries = project_struct( - &StructArray::from(batch), - &avro_schema, - schema.fields(), - zero_defaults, - )?; - Ok(RecordBatch::from(inherit(entries, manifest_file)?)) + /// Returns the schema of every batch. + pub fn schema(&self) -> SchemaRef { + self.schema.clone() + } + + fn project(&mut self, batch: RecordBatch) -> Result { + let entries = project_struct( + &StructArray::from(batch), + &self.avro_schema, + self.schema.fields(), + self.zero_defaults, + )?; + Ok(RecordBatch::from(inherit(entries, &mut self.inheritance)?)) + } +} + +impl> Iterator for ManifestEntryBatchReader { + type Item = Result; + + fn next(&mut self) -> Option { + // First row ID inheritance can stop partway through a batch, so the + // batches after an error would get the wrong row IDs. + if self.finished { + return None; + } + let batch = self + .reader + .next()? + .map_err(invalid_manifest) + .and_then(|batch| self.project(batch)); + self.finished = batch.is_err(); + Some(batch) + } } -/// Reads the entries of a manifest list into a [`RecordBatch`] with one row -/// per manifest. +/// Reads the entries of a manifest list as batches with one row per manifest. +/// +/// The schema is the spec's v3 `manifest_file` struct, with field IDs in the +/// Arrow field metadata, so v1 to v3 manifest lists read into the same schema. +/// Fields are matched to the manifest list's Avro schema by field ID. The +/// format version comes from the `format-version` key that writers put in the +/// file's metadata, and a manifest list without one is v1. A v1 manifest list +/// reads with the spec's v2+ defaults. Partition summary bounds keep the binary +/// single-value serialization they were written with. /// -/// The schema is the spec's `manifest_file` struct for the latest format -/// version, with field IDs in the Arrow field metadata, so manifest lists of -/// every format version read into the same schema. Fields are matched to the -/// manifest list's Avro schema by field ID. The format version comes from the -/// `format-version` key that writers put in the file's metadata, and a -/// manifest list without one is v1. A v1 manifest list reads with the spec's -/// v2+ defaults. Partition summary bounds keep the binary single-value -/// serialization they were written with. +/// # Errors +/// +/// Returns an error if the manifest list's header can't be read or names an +/// unsupported format version. Errors in the entries come from the reader, +/// which yields no batches after one. /// /// # Example /// @@ -178,78 +268,103 @@ pub fn read_manifest_entries(bytes: &[u8], manifest_file: &ManifestFile) -> Resu /// env!("CARGO_MANIFEST_DIR"), /// "/testdata/manifests_lists/manifest-list-v2-2.avro" /// ))?; -/// let batch = read_manifest_list(&bytes)?; -/// assert_eq!(batch.num_rows(), 1); +/// let mut rows = 0; +/// for batch in read_manifest_list(&bytes)? { +/// rows += batch?.num_rows(); +/// } +/// assert_eq!(rows, 1); /// # Ok::<(), Box>(()) /// ``` -pub fn read_manifest_list(bytes: &[u8]) -> Result { - // usize::MAX decodes the whole manifest list into one batch, which - // `only_batch` checks. - let reader = ReaderBuilder::new() - .with_batch_size(usize::MAX) - .build(bytes) - .map_err(invalid_manifest_list)?; - let header = reader.avro_header(); - let avro_schema: JsonValue = serde_json::from_slice( - header - .get("avro.schema") - .ok_or_else(|| invalid_data!("manifest list has no avro.schema"))?, - )?; - // The spec doesn't define this key, but Java, PyIceberg, and iceberg-rust - // all write it, and rejecting an unknown version keeps a newer manifest - // list from reading as an older one. - let format_version = match header.get("format-version") { - Some(bytes) => serde_json::from_slice::(bytes).map_err(|err| { - invalid_data!("failed to parse format version in manifest list metadata") - .with_source(err) - })?, - None => FormatVersion::V1, - }; - let schema = Arc::new(schema_to_arrow_schema( - &Schema::builder() - .with_fields(manifest_file_fields()) - .build()?, - )?); - let Some(batch) = only_batch(reader, invalid_manifest_list)? else { - return Ok(RecordBatch::new_empty(schema)); - }; - // v1 manifest lists have no content or sequence number columns, which v2+ - // readers read as 0. - let zero_defaults: &[i32] = match format_version { - FormatVersion::V1 => &[ - MANIFEST_CONTENT_FIELD_ID, - MANIFEST_SEQUENCE_NUMBER_FIELD_ID, - MIN_SEQUENCE_NUMBER_FIELD_ID, - ], - _ => &[], - }; - let manifests = project_struct( - &StructArray::from(batch), - &avro_schema, - schema.fields(), - zero_defaults, - )?; - Ok(RecordBatch::from(manifests)) +pub fn read_manifest_list>(bytes: B) -> Result> { + ManifestListBatchReader::try_new(bytes, BATCH_SIZE) } -/// Returns the only batch that `reader` decodes, or `None` for a file without -/// records. -fn only_batch( - mut reader: Reader<&[u8]>, - invalid: fn(ArrowError) -> Error, -) -> Result> { - let Some(batch) = reader.next() else { - return Ok(None); - }; - let batch = batch.map_err(invalid)?; - // Taking only the first batch relies on arrow-avro filling it to the batch - // size, so fail instead of dropping records if that changes. - if reader.next().is_some() { - return Err(invalid_data!( - "arrow-avro split the file into more than one batch" - )); +/// Batches of the entries of a manifest list, returned by +/// [`read_manifest_list`]. +pub struct ManifestListBatchReader> { + reader: Reader>, + schema: SchemaRef, + avro_schema: JsonValue, + zero_defaults: &'static [i32], + finished: bool, +} + +impl> ManifestListBatchReader { + fn try_new(bytes: B, batch_size: usize) -> Result { + let reader = ReaderBuilder::new() + .with_batch_size(batch_size) + .build(Cursor::new(bytes)) + .map_err(invalid_manifest_list)?; + let header = reader.avro_header(); + let avro_schema: JsonValue = serde_json::from_slice( + header + .get("avro.schema") + .ok_or_else(|| invalid_data!("manifest list has no avro.schema"))?, + )?; + // The spec doesn't define this key, but Java, PyIceberg, and iceberg-rust + // all write it, and rejecting an unknown version keeps a newer manifest + // list from reading as an older one. + let format_version = match header.get("format-version") { + Some(bytes) => serde_json::from_slice::(bytes).map_err(|err| { + invalid_data!("failed to parse format version in manifest list metadata") + .with_source(err) + })?, + None => FormatVersion::V1, + }; + let schema = Arc::new(schema_to_arrow_schema( + &Schema::builder() + .with_fields(manifest_file_fields()) + .build()?, + )?); + // v1 manifest lists have no content or sequence number columns, which v2+ + // readers read as 0. + let zero_defaults: &[i32] = match format_version { + FormatVersion::V1 => &[ + MANIFEST_CONTENT_FIELD_ID, + MANIFEST_SEQUENCE_NUMBER_FIELD_ID, + MIN_SEQUENCE_NUMBER_FIELD_ID, + ], + _ => &[], + }; + Ok(Self { + reader, + schema, + avro_schema, + zero_defaults, + finished: false, + }) + } + + /// Returns the schema of every batch. + pub fn schema(&self) -> SchemaRef { + self.schema.clone() + } + + fn project(&self, batch: RecordBatch) -> Result { + Ok(RecordBatch::from(project_struct( + &StructArray::from(batch), + &self.avro_schema, + self.schema.fields(), + self.zero_defaults, + )?)) + } +} + +impl> Iterator for ManifestListBatchReader { + type Item = Result; + + fn next(&mut self) -> Option { + if self.finished { + return None; + } + let batch = self + .reader + .next()? + .map_err(invalid_manifest_list) + .and_then(|batch| self.project(batch)); + self.finished = batch.is_err(); + Some(batch) } - Ok(Some(batch)) } /// Removes `data_file.partition` from a manifest's Avro schema if it has no @@ -452,7 +567,7 @@ fn project( /// Fills in the values that the spec says entries inherit from the manifest /// list. -fn inherit(entries: StructArray, manifest_file: &ManifestFile) -> Result { +fn inherit(entries: StructArray, inheritance: &mut Inheritance) -> Result { let (fields, mut columns, nulls) = entries.into_parts(); let index = |name: &str| fields.find(name).unwrap().0; let (status, snapshot_id, sequence_number, file_sequence_number, data_file) = ( @@ -463,22 +578,18 @@ fn inherit(entries: StructArray, manifest_file: &ManifestFile) -> Result 0 || columns[file_sequence_number].null_count() > 0 { let added = eq( &columns[status], &Int32Array::new_scalar(ManifestStatus::Added as i32), )?; for column in [sequence_number, file_sequence_number] { - columns[column] = fill_nulls( - &columns[column], - Some(&added), - manifest_file.sequence_number, - )?; + columns[column] = + fill_nulls(&columns[column], Some(&added), inheritance.sequence_number)?; } } - if manifest_file.content == ManifestContentType::Data { + if inheritance.content == ManifestContentType::Data { let (data_file_fields, mut data_file_columns, data_file_nulls) = columns[data_file].as_struct().clone().into_parts(); let index = |name: &str| data_file_fields.find(name).unwrap().0; @@ -487,7 +598,7 @@ fn inherit(entries: StructArray, manifest_file: &ManifestFile) -> Result Result { - let Some(manifest_first_row_id) = manifest_file.first_row_id else { + let Some(next_row_id) = inheritance.next_row_id.as_mut() else { return Ok(new_null_array(&DataType::Int64, first_row_ids.len())); }; if first_row_ids.null_count() == 0 { return Ok(Arc::new(first_row_ids.clone())); } - let mut next_row_id = i64::try_from(manifest_first_row_id)?; let mut builder = Int64Builder::with_capacity(first_row_ids.len()); for ((first_row_id, &status), &record_count) in first_row_ids .iter() @@ -541,14 +651,14 @@ fn inherit_first_row_ids( if record_count < 0 { return Err(invalid_data!( "negative record_count {record_count} assigning first row IDs in {}", - manifest_file.manifest_path + inheritance.manifest_path )); } - builder.append_value(next_row_id); - next_row_id = next_row_id.checked_add(record_count).ok_or_else(|| { + builder.append_value(*next_row_id); + *next_row_id = next_row_id.checked_add(record_count).ok_or_else(|| { invalid_data!( "row ID overflow assigning first row IDs in {}", - manifest_file.manifest_path + inheritance.manifest_path ) })?; } @@ -573,6 +683,8 @@ mod tests { Time64MicrosecondArray, TimestampMicrosecondArray, TimestampNanosecondArray, }; use arrow_buffer::{NullBuffer, OffsetBuffer}; + use arrow_schema::SchemaRef; + use arrow_select::concat::concat_batches; use super::*; use crate::ErrorKind; @@ -671,6 +783,50 @@ mod tests { (bytes, manifest_file) } + fn concat( + schema: SchemaRef, + batches: impl Iterator>, + ) -> Result { + let batches = batches.collect::>>()?; + Ok(concat_batches(&schema, &batches)?) + } + + fn read_entries(bytes: &[u8], manifest_file: &ManifestFile) -> Result { + let reader = read_manifest_entries(bytes, manifest_file)?; + concat(reader.schema(), reader) + } + + fn read_list(bytes: &[u8]) -> Result { + let reader = read_manifest_list(bytes)?; + concat(reader.schema(), reader) + } + + /// Checks that reading in batches of 1 to 3 rows gives nonempty batches of + /// the reader's schema that concatenate to the file read as one batch. + fn assert_batches_match_whole>>( + read: impl Fn(usize) -> (SchemaRef, I), + ) { + let (schema, batches) = read(usize::MAX); + let whole = concat(schema.clone(), batches).unwrap(); + for batch_size in 1..=3 { + let (batch_schema, batches) = read(batch_size); + assert_eq!(batch_schema, schema); + let batches: Vec<_> = batches.map(Result::unwrap).collect(); + for batch in &batches { + assert_eq!(batch.schema(), schema, "batch size {batch_size}"); + assert!( + (1..=batch_size).contains(&batch.num_rows()), + "batch size {batch_size}" + ); + } + assert_eq!( + concat_batches(&schema, &batches).unwrap(), + whole, + "batch size {batch_size}" + ); + } + } + fn column(batch: &RecordBatch, name: &str) -> ArrayRef { batch.column_by_name(name).unwrap().clone() } @@ -843,7 +999,7 @@ mod tests { let spec = partition_spec(&[("i", Transform::Identity), ("s", Transform::Truncate(2))]); let (bytes, manifest_file) = write_manifest(spec, Some(1), |b| b.build_v2_data(), |_| {}).await; - let batch = read_manifest_entries(&bytes, &manifest_file).unwrap(); + let batch = read_entries(&bytes, &manifest_file).unwrap(); assert_eq!(batch.num_rows(), 0); let long = || Type::Primitive(PrimitiveType::Long); @@ -957,7 +1113,7 @@ mod tests { }, ) .await; - let batch = read_manifest_entries(&bytes, &manifest_file).unwrap(); + let batch = read_entries(&bytes, &manifest_file).unwrap(); assert_eq!(batch.num_rows(), 2); assert_eq!( @@ -1050,7 +1206,7 @@ mod tests { }, ) .await; - let batch = read_manifest_entries(&bytes, &manifest_file).unwrap(); + let batch = read_entries(&bytes, &manifest_file).unwrap(); let column = |name: &str| data_file_column(&batch, name); assert_eq!( @@ -1139,7 +1295,7 @@ mod tests { }, ) .await; - let batch = read_manifest_entries(&bytes, &manifest_file).unwrap(); + let batch = read_entries(&bytes, &manifest_file).unwrap(); let partition = data_file_column(&batch, "partition"); let partition = partition.as_struct(); @@ -1215,7 +1371,7 @@ mod tests { // The v1 default must win over inheritance, so give the manifest list // entry a sequence number that inheritance would use. manifest_file.sequence_number = 9; - let batch = read_manifest_entries(&bytes, &manifest_file).unwrap(); + let batch = read_entries(&bytes, &manifest_file).unwrap(); assert_eq!( data_file_column(&batch, "content").as_ref(), @@ -1272,7 +1428,7 @@ mod tests { ); manifest_file.added_snapshot_id = 42; manifest_file.sequence_number = 7; - let batch = read_manifest_entries(&bytes, &manifest_file).unwrap(); + let batch = read_entries(&bytes, &manifest_file).unwrap(); assert_eq!( column(&batch, "snapshot_id").as_ref(), @@ -1322,7 +1478,7 @@ mod tests { }, ); manifest_file.sequence_number = 7; - let batch = read_manifest_entries(&bytes, &manifest_file).unwrap(); + let batch = read_entries(&bytes, &manifest_file).unwrap(); assert_eq!( column(&batch, "file_sequence_number").as_ref(), @@ -1370,7 +1526,7 @@ mod tests { async fn test_first_row_id_inheritance() { let (bytes, mut manifest_file) = first_row_id_manifest().await; manifest_file.first_row_id = Some(100); - let batch = read_manifest_entries(&bytes, &manifest_file).unwrap(); + let batch = read_entries(&bytes, &manifest_file).unwrap(); // Deleted entries don't take row IDs, and entries with a first_row_id keep it. assert_eq!( data_file_column(&batch, "first_row_id").as_ref(), @@ -1378,16 +1534,25 @@ mod tests { ); manifest_file.first_row_id = None; - let batch = read_manifest_entries(&bytes, &manifest_file).unwrap(); + let batch = read_entries(&bytes, &manifest_file).unwrap(); assert_eq!(data_file_column(&batch, "first_row_id").null_count(), 5); manifest_file.first_row_id = Some(i64::MAX as u64 - 5); - let err = read_manifest_entries(&bytes, &manifest_file).unwrap_err(); + let err = read_entries(&bytes, &manifest_file).unwrap_err(); assert_eq!(err.kind(), ErrorKind::DataInvalid); // A negative record_count would assign overlapping row IDs. - let bytes = rewrite_manifest( - &bytes, + let bytes = negate_record_count(&bytes, 10); + manifest_file.first_row_id = Some(100); + let err = read_entries(&bytes, &manifest_file).unwrap_err(); + assert_eq!(err.kind(), ErrorKind::DataInvalid); + } + + /// Negates the `record_count` of the data files whose count is + /// `record_count`. + fn negate_record_count(bytes: &[u8], record_count: i64) -> Vec { + rewrite_manifest( + bytes, Codec::Null, |_| {}, |record| { @@ -1400,21 +1565,92 @@ mod tests { panic!("manifest entry has no data_file record"); }; for (name, value) in data_file { - if name == "record_count" && *value == AvroValue::Long(10) { - *value = AvroValue::Long(-10); + if name == "record_count" && *value == AvroValue::Long(record_count) { + *value = AvroValue::Long(-record_count); } } }, - ); - manifest_file.first_row_id = Some(100); - let err = read_manifest_entries(&bytes, &manifest_file).unwrap_err(); + ) + } + + #[tokio::test] + async fn test_manifest_batches_match_whole_manifest() { + let (first_row_id_bytes, mut first_row_id_manifest_file) = first_row_id_manifest().await; + first_row_id_manifest_file.first_row_id = Some(100); + let v1_files = (0..3).map(|i| { + data_file(DataContentType::Data, &format!("s3://b/data/{i}.parquet")) + .build() + .unwrap() + }); + let v1 = write_manifest( + partition_spec(&[]), + Some(1), + |b| b.build_v1(), + |writer| v1_files.for_each(|file| writer.add_file(file, -1).unwrap()), + ) + .await; + let empty = write_manifest( + partition_spec(&[("i", Transform::Identity)]), + Some(1), + |b| b.build_v2_data(), + |_| {}, + ) + .await; + let pyiceberg_bytes = std::fs::read(format!( + "{}/testdata/manifests/pyiceberg-v2-data.avro", + env!("CARGO_MANIFEST_DIR") + )) + .unwrap(); + let pyiceberg_manifest_file = ManifestFile { + sequence_number: 1, + ..listed_manifest("s3://b/metadata/0.avro", ManifestContentType::Data) + }; + + for (bytes, manifest_file) in [ + (first_row_id_bytes, first_row_id_manifest_file), + v1, + empty, + (pyiceberg_bytes, pyiceberg_manifest_file), + ] { + assert_batches_match_whole(|batch_size| { + let reader = + ManifestEntryBatchReader::try_new(&bytes, &manifest_file, batch_size).unwrap(); + (reader.schema(), reader) + }); + } + } + + #[tokio::test] + async fn test_unknown_manifest_format_version_is_an_error() { + let (bytes, manifest_file) = first_row_id_manifest().await; + let bytes = set_avro_metadata(&bytes, "format-version", Some("4")); + let Err(err) = read_manifest_entries(&bytes, &manifest_file) else { + panic!("a v4 manifest read as an older version"); + }; assert_eq!(err.kind(), ErrorKind::DataInvalid); } + #[tokio::test] + async fn test_manifest_error_ends_the_batches() { + let (bytes, mut manifest_file) = first_row_id_manifest().await; + manifest_file.first_row_id = Some(100); + // The last of the five entries fails first row ID inheritance, so the + // third batch of two is an error. + let bytes = negate_record_count(&bytes, 2); + let mut reader = ManifestEntryBatchReader::try_new(&bytes, &manifest_file, 2).unwrap(); + assert_eq!(reader.next().unwrap().unwrap().num_rows(), 2); + assert_eq!(reader.next().unwrap().unwrap().num_rows(), 2); + assert_eq!( + reader.next().unwrap().unwrap_err().kind(), + ErrorKind::DataInvalid + ); + assert!(reader.next().is_none()); + } + #[tokio::test] async fn test_compressed_manifests() { let (bytes, manifest_file) = first_row_id_manifest().await; - let expected = read_manifest_entries(&bytes, &manifest_file).unwrap(); + let expected = read_entries(&bytes, &manifest_file).unwrap(); for codec in [ Codec::Deflate(DeflateSettings::default()), Codec::Snappy, @@ -1422,7 +1658,7 @@ mod tests { ] { let bytes = rewrite_manifest(&bytes, codec, |_| {}, |_| {}); assert_eq!( - read_manifest_entries(&bytes, &manifest_file).unwrap(), + read_entries(&bytes, &manifest_file).unwrap(), expected, "{codec:?}" ); @@ -1448,7 +1684,7 @@ mod tests { ("file_path", "path"), ("i_identity", "renamed_partition"), ]); - let batch = read_manifest_entries(&bytes, &manifest_file).unwrap(); + let batch = read_entries(&bytes, &manifest_file).unwrap(); assert_eq!( data_file_column(&batch, "file_path").as_ref(), @@ -1486,7 +1722,7 @@ mod tests { }, |_| {}, ); - let err = read_manifest_entries(&bytes, &manifest_file).unwrap_err(); + let err = read_entries(&bytes, &manifest_file).unwrap_err(); assert_eq!(err.kind(), ErrorKind::DataInvalid); } @@ -1516,7 +1752,7 @@ mod tests { key_metadata: None, first_row_id: None, }; - let batch = read_manifest_entries(&bytes, &manifest_file).unwrap(); + let batch = read_entries(&bytes, &manifest_file).unwrap(); assert_eq!(batch.num_rows(), manifest.entries().len()); let paths = data_file_column(&batch, "file_path"); @@ -1587,7 +1823,7 @@ mod tests { #[tokio::test] async fn test_schema_is_spec_manifest_file() { let bytes = write_manifest_list(FormatVersion::V3, vec![]).await; - let batch = read_manifest_list(&bytes).unwrap(); + let batch = read_list(&bytes).unwrap(); assert_eq!(batch.num_rows(), 0); let long = || Type::Primitive(PrimitiveType::Long); @@ -1656,7 +1892,7 @@ mod tests { ..listed_manifest("s3://b/metadata/deletes.avro", ManifestContentType::Deletes) }; let bytes = write_manifest_list(FormatVersion::V3, vec![data, existing_deletes]).await; - let batch = read_manifest_list(&bytes).unwrap(); + let batch = read_list(&bytes).unwrap(); let schema = batch.schema(); let DataType::List(summary) = schema.field_with_name("partitions").unwrap().data_type() @@ -1725,7 +1961,7 @@ mod tests { // A manifest list without a format version reads as v1. let without_format_version = set_avro_metadata(&bytes, "format-version", None); for bytes in [bytes, without_format_version] { - let batch = read_manifest_list(&bytes).unwrap(); + let batch = read_list(&bytes).unwrap(); assert_eq!( column(&batch, "content").as_ref(), @@ -1767,7 +2003,7 @@ mod tests { ("added_files_count", "added_data_files_count"), ("contains_null", "has_null"), ]); - let batch = read_manifest_list(&bytes).unwrap(); + let batch = read_list(&bytes).unwrap(); assert_eq!( column(&batch, "manifest_path").as_ref(), @@ -1791,7 +2027,7 @@ mod tests { #[tokio::test] async fn test_invalid_manifest_list_is_an_error() { - let err = read_manifest_list(&[]).unwrap_err(); + let err = read_list(&[]).unwrap_err(); assert_eq!(err.kind(), ErrorKind::DataInvalid); let bytes = write_manifest_list(FormatVersion::V2, vec![listed_manifest( @@ -1836,7 +2072,7 @@ mod tests { without_sequence_number, unknown_format_version, ] { - let err = read_manifest_list(&bytes).unwrap_err(); + let err = read_list(&bytes).unwrap_err(); assert_eq!(err.kind(), ErrorKind::DataInvalid); } } @@ -1852,7 +2088,7 @@ mod tests { .unwrap(); let manifest_list = ManifestList::parse_with_version(&bytes, FormatVersion::V2).unwrap(); - let batch = read_manifest_list(&bytes).unwrap(); + let batch = read_list(&bytes).unwrap(); assert_eq!(batch.num_rows(), manifest_list.entries().len(), "{name}"); let paths = column(&batch, "manifest_path"); @@ -1894,4 +2130,25 @@ mod tests { } } } + + #[tokio::test] + async fn test_manifest_list_batches_match_whole_list() { + for version in [FormatVersion::V1, FormatVersion::V2, FormatVersion::V3] { + for len in [0, 3] { + let manifests = (0..len) + .map(|i| { + listed_manifest( + &format!("s3://b/metadata/{i}.avro"), + ManifestContentType::Data, + ) + }) + .collect(); + let bytes = write_manifest_list(version, manifests).await; + assert_batches_match_whole(|batch_size| { + let reader = ManifestListBatchReader::try_new(&bytes, batch_size).unwrap(); + (reader.schema(), reader) + }); + } + } + } } diff --git a/crates/iceberg/src/arrow/mod.rs b/crates/iceberg/src/arrow/mod.rs index d8343ef1ef..25a84a0c64 100644 --- a/crates/iceberg/src/arrow/mod.rs +++ b/crates/iceberg/src/arrow/mod.rs @@ -29,7 +29,9 @@ pub(crate) mod delete_filter; mod int96; mod manifest; -pub use manifest::{read_manifest_entries, read_manifest_list}; +pub use manifest::{ + ManifestEntryBatchReader, ManifestListBatchReader, read_manifest_entries, read_manifest_list, +}; mod reader; /// RecordBatch projection utilities pub mod record_batch_projector;