laskoviymishka commented on code in PR #2922:
URL: https://github.com/apache/iceberg-rust/pull/2922#discussion_r3675094464
##########
crates/iceberg/src/spec/manifest_list/manifest_file.rs:
##########
@@ -195,8 +195,61 @@ impl ManifestFile {
entry.inherit_data(self);
}
+ self.assign_first_row_ids(&mut entries)?;
+
Ok(Manifest::new(metadata, entries))
}
+
+ /// Assigns `first_row_id` to data-file entries that do not already have
one,
+ /// starting from the manifest-level `first_row_id` and advancing by each
+ /// entry's record count. See the row-lineage inheritance rules in
+ ///
<https://github.com/apache/iceberg/blob/main/format/spec.md#first-row-id-inheritance>.
+ fn assign_first_row_ids(&self, entries: &mut [ManifestEntry]) ->
Result<()> {
+ let Some(manifest_first_row_id) = self.first_row_id else {
+ return Ok(());
+ };
+
+ // A `first_row_id` is only valid on data manifests; delete files
always
+ // have a null `first_row_id`.
+ if self.content != ManifestContentType::Data {
Review Comment:
I'd make this a no-op rather than a hard error. The spec's "`first_row_id`
is always null for delete manifests" is a write-side constraint; Java's
`readDeleteManifest()` just omits the `firstRowId` argument to the reader and
silently ignores the field. Rejecting here means a delete manifest written by
any other client with a stray non-null `first_row_id` (buggy writer, partial
migration) is readable everywhere except iceberg-rust, where it fails the
entire scan — a one-sided interop break.
I'd return `Ok(())` here, maybe with a `tracing::warn!` to surface the
anomaly. If we do decide to keep it strict, capitalise the message to match the
others in the file. wdyt?
##########
crates/iceberg/src/spec/manifest_list/manifest_file.rs:
##########
@@ -195,8 +195,61 @@ impl ManifestFile {
entry.inherit_data(self);
}
+ self.assign_first_row_ids(&mut entries)?;
+
Ok(Manifest::new(metadata, entries))
}
+
+ /// Assigns `first_row_id` to data-file entries that do not already have
one,
+ /// starting from the manifest-level `first_row_id` and advancing by each
+ /// entry's record count. See the row-lineage inheritance rules in
+ ///
<https://github.com/apache/iceberg/blob/main/format/spec.md#first-row-id-inheritance>.
+ fn assign_first_row_ids(&self, entries: &mut [ManifestEntry]) ->
Result<()> {
+ let Some(manifest_first_row_id) = self.first_row_id else {
Review Comment:
I think we're missing Java's third `idAssigner` branch here. When the
manifest-level `first_row_id` is null on a committed read, Java doesn't just
return — it walks the entries and sets each file's `first_row_id` back to null
(`ManifestReader.idAssigner`, the `BaseFile.setFirstRowId(null)` branch).
That's the upgrade-path guard: a v3 data manifest with no manifest-level
offset but EXISTING entries that carry a previously-inherited `first_row_id`.
Java exposes null there; this early-return leaves the stale per-file values
intact, so anything computing `_row_id = first_row_id + _pos` on the Rust side
gets wrong ids. I'd add a data-manifest path that clears
`entry.data_file.first_row_id` when `self.first_row_id` is `None`. wdyt?
##########
crates/iceberg/src/spec/manifest_list/manifest_file.rs:
##########
@@ -195,8 +195,61 @@ impl ManifestFile {
entry.inherit_data(self);
}
+ self.assign_first_row_ids(&mut entries)?;
+
Ok(Manifest::new(metadata, entries))
}
+
+ /// Assigns `first_row_id` to data-file entries that do not already have
one,
+ /// starting from the manifest-level `first_row_id` and advancing by each
Review Comment:
The doc says the counter advances "by each entry's record count", but the
code only advances for entries that actually receive an assignment —
pre-assigned and DELETED entries are skipped and don't move it (which the
interleaved test verifies). I'd tighten it to say the counter advances by each
newly-assigned entry's record count, and note the skips.
##########
crates/iceberg/src/spec/manifest_list/manifest_file.rs:
##########
@@ -389,4 +436,157 @@ mod test {
.expect_err("load_manifest must fail when decrypting with the
wrong AAD prefix");
assert_eq!(err.kind(), ErrorKind::Unexpected);
}
+
+ /// Builds a data-file manifest entry with the given status, record count,
+ /// and pre-existing `first_row_id`.
+ fn data_entry(
+ status: ManifestStatus,
+ record_count: u64,
+ first_row_id: Option<i64>,
+ ) -> ManifestEntry {
+ let mut builder = DataFileBuilder::default();
+ builder
+ .content(DataContentType::Data)
+ .file_path("s3://bucket/table/data/00000.parquet".to_string())
+ .file_format(DataFileFormat::Parquet)
+ .file_size_in_bytes(4096)
+ .record_count(record_count);
+ if let Some(id) = first_row_id {
+ builder.first_row_id(Some(id));
+ }
+
+ ManifestEntry::builder()
+ .status(status)
+ .data_file(builder.build().unwrap())
+ .build()
+ }
+
+ /// Builds a manifest file with the given content type and manifest-level
+ /// `first_row_id`. Other fields are irrelevant to row-id assignment.
+ fn manifest_file(content: ManifestContentType, first_row_id: Option<u64>)
-> ManifestFile {
+ ManifestFile {
+ manifest_path: "memory:///m.avro".to_string(),
+ manifest_length: 0,
+ partition_spec_id: 0,
+ content,
+ sequence_number: 0,
+ min_sequence_number: 0,
+ added_snapshot_id: 0,
+ 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,
+ }
+ }
+
+ #[test]
+ fn test_assign_first_row_ids_interleaved() {
+ let manifest = manifest_file(ManifestContentType::Data, Some(10));
+ let mut entries = vec![
+ data_entry(ManifestStatus::Added, 3, None),
+ // A pre-assigned entry between two assigned ones: it keeps its id
and
+ // must not advance the running counter.
+ data_entry(ManifestStatus::Added, 5, Some(100)),
+ // A deleted entry likewise neither receives nor consumes a row id.
+ data_entry(ManifestStatus::Deleted, 7, None),
Review Comment:
Nice interleaved test. One case I'd add while we're here: a DELETED entry
arriving with a pre-set `first_row_id` (`data_entry(Deleted, 7, Some(999))`)
asserting it stays `Some(999)`. The skip-deleted path leaves pre-set values
untouched today, and a test makes that invariant explicit rather than
incidental.
##########
crates/iceberg/src/spec/manifest_list/manifest_file.rs:
##########
@@ -389,4 +436,157 @@ mod test {
.expect_err("load_manifest must fail when decrypting with the
wrong AAD prefix");
assert_eq!(err.kind(), ErrorKind::Unexpected);
}
+
+ /// Builds a data-file manifest entry with the given status, record count,
+ /// and pre-existing `first_row_id`.
+ fn data_entry(
+ status: ManifestStatus,
+ record_count: u64,
+ first_row_id: Option<i64>,
+ ) -> ManifestEntry {
+ let mut builder = DataFileBuilder::default();
+ builder
+ .content(DataContentType::Data)
+ .file_path("s3://bucket/table/data/00000.parquet".to_string())
+ .file_format(DataFileFormat::Parquet)
+ .file_size_in_bytes(4096)
+ .record_count(record_count);
+ if let Some(id) = first_row_id {
Review Comment:
Minor: `DataFile`'s `first_row_id` is `#[builder(default)]` without
`strip_option`, so the setter takes `Option<i64>` directly and defaults to
`None`. This whole block collapses to `builder.first_row_id(first_row_id);`.
##########
crates/iceberg/src/spec/manifest_list/manifest_file.rs:
##########
@@ -195,8 +195,61 @@ impl ManifestFile {
entry.inherit_data(self);
}
+ self.assign_first_row_ids(&mut entries)?;
Review Comment:
I think this introduces a stale-cache bug, and it's the thing I'd most want
to settle before merge.
Until now `load_manifest` was a pure function of the Avro bytes, so the
`ObjectCache` keying manifests by `manifest_path` alone (`object_cache.rs:108`)
was safe. This call makes the result depend on `self.first_row_id`, the
manifest-list-level offset — and the spec lets the same physical manifest carry
different offsets across snapshots: the v3 upgrade path (`None` in S1,
`Some(X)` in S2 after the first post-upgrade commit) and branches (two branches
can hand the same manifest disjoint offsets). A time-travel or branched read in
one `Table` lifetime hits the cache on the second request and gets the first
request's `first_row_id` values, silently wrong.
I'd fold `first_row_id` into the cache key
(`CachedObjectKey::Manifest((String, Option<u64>))`), or split the Avro parse
(cached) from the id assignment (applied post-retrieval). There's also no test
loading the same `manifest_path` twice with different offsets — that's exactly
the case that catches this, so I'd add it alongside the fix. wdyt?
##########
crates/iceberg/src/spec/manifest_list/manifest_file.rs:
##########
@@ -195,8 +195,61 @@ impl ManifestFile {
entry.inherit_data(self);
}
+ self.assign_first_row_ids(&mut entries)?;
+
Ok(Manifest::new(metadata, entries))
}
+
+ /// Assigns `first_row_id` to data-file entries that do not already have
one,
+ /// starting from the manifest-level `first_row_id` and advancing by each
+ /// entry's record count. See the row-lineage inheritance rules in
+ ///
<https://github.com/apache/iceberg/blob/main/format/spec.md#first-row-id-inheritance>.
+ fn assign_first_row_ids(&self, entries: &mut [ManifestEntry]) ->
Result<()> {
+ let Some(manifest_first_row_id) = self.first_row_id else {
+ return Ok(());
+ };
+
+ // A `first_row_id` is only valid on data manifests; delete files
always
+ // have a null `first_row_id`.
+ if self.content != ManifestContentType::Data {
+ return Err(Error::new(
+ ErrorKind::DataInvalid,
+ format!(
+ "first_row_id is not valid for delete manifests (found
{manifest_first_row_id} on {})",
+ self.manifest_path
+ ),
+ ));
+ }
+
+ let mut next_row_id = i64::try_from(manifest_first_row_id).map_err(|_|
{
Review Comment:
This `try_from` is only necessary because `ManifestFile.first_row_id` (and
`ManifestFileV3.first_row_id` in `_serde.rs`) is `Option<u64>`, while spec
field 520 and Java's `ManifestFile.firstRowId()` are signed `long` — and
`DataFile.first_row_id` is already `Option<i64>`. Non-negative values
round-trip fine over Avro, so it's low practical risk, but the unsigned type is
what forces the conversion and makes it slightly fragile.
Not blocking, but I'd consider moving both fields to `Option<i64>` to match
the spec and dropping the `try_from` (the `checked_add_unsigned` would need a
revisit). Probably its own cleanup PR.
##########
crates/iceberg/src/spec/manifest_list/manifest_file.rs:
##########
@@ -195,8 +195,61 @@ impl ManifestFile {
entry.inherit_data(self);
}
+ self.assign_first_row_ids(&mut entries)?;
+
Ok(Manifest::new(metadata, entries))
}
+
+ /// Assigns `first_row_id` to data-file entries that do not already have
one,
+ /// starting from the manifest-level `first_row_id` and advancing by each
+ /// entry's record count. See the row-lineage inheritance rules in
+ ///
<https://github.com/apache/iceberg/blob/main/format/spec.md#first-row-id-inheritance>.
+ fn assign_first_row_ids(&self, entries: &mut [ManifestEntry]) ->
Result<()> {
+ let Some(manifest_first_row_id) = self.first_row_id else {
+ return Ok(());
+ };
+
+ // A `first_row_id` is only valid on data manifests; delete files
always
+ // have a null `first_row_id`.
+ if self.content != ManifestContentType::Data {
+ return Err(Error::new(
+ ErrorKind::DataInvalid,
+ format!(
+ "first_row_id is not valid for delete manifests (found
{manifest_first_row_id} on {})",
+ self.manifest_path
+ ),
+ ));
+ }
+
+ let mut next_row_id = i64::try_from(manifest_first_row_id).map_err(|_|
{
+ Error::new(
+ ErrorKind::DataInvalid,
+ format!("Invalid first_row_id: {manifest_first_row_id}
(exceeds i64::MAX)"),
+ )
+ })?;
+
+ for entry in entries {
+ if !entry.is_alive() {
+ continue;
+ }
+
+ if entry.data_file.first_row_id.is_none() {
+ entry.data_file.first_row_id = Some(next_row_id);
+ let record_count = entry.data_file.record_count;
+ next_row_id =
next_row_id.checked_add_unsigned(record_count).ok_or_else(|| {
+ Error::new(
+ ErrorKind::DataInvalid,
+ format!(
+ "Row ID overflow assigning first_row_id in {}.
Next Row ID: {next_row_id}, Record Count: {record_count}",
Review Comment:
Small wording thing — the value is right but the label reads wrong. At this
point `next_row_id` still holds the id we just stamped onto the current entry
(the add hasn't landed), so "Next Row ID" makes an operator read it as the next
value to assign when it's actually the triggering file's own `first_row_id`.
I'd reword to something like `"file first_row_id={next_row_id} +
record_count={record_count} exceeds i64::MAX"`.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
