laskoviymishka commented on code in PR #2952:
URL: https://github.com/apache/iceberg-rust/pull/2952#discussion_r3712156266
##########
crates/iceberg/src/scan/task.rs:
##########
@@ -67,6 +67,21 @@ pub struct FileScanTask {
#[builder(default)]
pub record_count: Option<u64>,
+ /// The first row id assigned to the data file.
+ ///
+ /// Used to derive the `_row_id` metadata column: for a row without an
+ /// explicit `_row_id`, it is this value plus the row's ordinal position.
+ #[serde(skip_serializing_if = "Option::is_none")]
+ #[builder(default)]
+ pub first_row_id: Option<i64>,
+
+ /// The data sequence number of the data file.
Review Comment:
This reads as though the value comes off `DataFile`, but it's
`ManifestEntry::sequence_number()` — field 3 on the entry envelope, nothing
inside the data file. Worth saying so, and worth naming it as the *data*
sequence number as opposed to `file_sequence_number`, since the spec carries
both and they're easy to swap.
A word on `None` would help too: `inherit_data()` only fills this in when
the snapshot's sequence number is the initial one, so an `Existing` entry in a
malformed v2 manifest can reach the task with `None`. Better pinned here than
assumed `Some` by the `_last_updated_sequence_number` work.
##########
crates/iceberg/src/scan/mod.rs:
##########
@@ -996,6 +996,84 @@ pub mod tests {
manifest_list_write.close().await.unwrap();
}
+ /// Writes a v3 data manifest with a manifest-level `first_row_id` of
0,
+ /// so live entries inherit a per-file `first_row_id` on read.
Upgrades the
+ /// table to v3 first, so the manifest list is read as v3.
+ pub async fn setup_v3_manifest_files(&mut self) {
+ let metadata = TableMetadataBuilder::new_from_metadata(
+ self.table.metadata().clone(),
+ self.table.metadata_location().map(str::to_string),
+ )
+ .upgrade_format_version(FormatVersion::V3)
+ .unwrap()
+ .build()
+ .unwrap()
+ .metadata;
+ self.table = Table::builder()
+ .metadata(metadata)
+ .identifier(self.table.identifier().clone())
+ .file_io(self.table.file_io().clone())
+
.metadata_location(self.table.metadata_location().unwrap().to_string())
+ .runtime(test_runtime())
+ .build()
+ .unwrap();
+
+ let current_snapshot =
self.table.metadata().current_snapshot().unwrap();
+ let current_schema =
current_snapshot.schema(self.table.metadata()).unwrap();
+ let current_partition_spec =
self.table.metadata().default_partition_spec();
+
+ let parquet_file_size = self.write_parquet_data_files();
+
+ let mut writer = ManifestWriterBuilder::new(
+ self.next_manifest_file(),
+ Some(current_snapshot.snapshot_id()),
+ current_schema.clone(),
+ current_partition_spec.as_ref().clone(),
+ )
+ .build_v3_data();
+ writer
+ .add_entry(
+ ManifestEntry::builder()
+ .status(ManifestStatus::Added)
+ .data_file(
+ DataFileBuilder::default()
+ .partition_spec_id(0)
+ .content(DataContentType::Data)
+ .file_path(format!("{}/1.parquet",
&self.table_location))
+ .file_format(DataFileFormat::Parquet)
+ .file_size_in_bytes(parquet_file_size)
+ .record_count(1)
+
.partition(Struct::from_iter([Some(Literal::long(100))]))
+ .key_metadata(None)
+ .build()
+ .unwrap(),
+ )
+ .build(),
+ )
+ .unwrap();
+ let data_file_manifest =
writer.write_manifest_file().await.unwrap();
+
+ let manifest_list_writer = self
+ .table
+ .file_io()
+ .new_output(current_snapshot.manifest_list())
+ .unwrap()
+ .writer()
+ .await
+ .unwrap();
+ let mut manifest_list_write = ManifestListWriter::v3(
+ manifest_list_writer,
+ current_snapshot.snapshot_id(),
+ current_snapshot.parent_snapshot_id(),
+ current_snapshot.sequence_number(),
+ Some(0),
Review Comment:
I'd make this non-zero — `Some(42)`, anything. With a manifest-level
`first_row_id` of 0 the inherited per-file value comes out `Some(0)` as well,
so an implementation that hands back `Some(0)` unconditionally, or one that
confuses `None` with a zero default, passes this test exactly like the correct
one does.
The inheritance arithmetic is the interesting part here, and a non-zero base
is what makes it observable. The assertion below then becomes `Some(42)`.
##########
crates/iceberg/src/scan/mod.rs:
##########
@@ -1862,6 +1940,60 @@ pub mod tests {
);
}
+ #[tokio::test]
+ async fn test_plan_files_carries_row_lineage_into_file_scan_task() {
+ let mut fixture = TableTestFixture::new();
+ fixture.setup_manifest_files().await;
+
+ let mut tasks: Vec<_> = fixture
+ .table
+ .scan()
+ .build()
+ .unwrap()
+ .plan_files()
+ .await
+ .unwrap()
+ .try_collect()
+ .await
+ .unwrap();
+
+ tasks.sort_by_key(|task| task.data_file_path.to_string());
+ assert_eq!(tasks.len(), 2);
+
+ // The added file inherits the current snapshot's data sequence number,
+ // the existing file keeps the one it was written with.
+ assert_eq!(tasks[0].data_sequence_number, Some(1));
Review Comment:
The index-to-file mapping is implicit — it holds only because `1.parquet`
sorts ahead of `3.parquet`, and the comment describes that rather than
asserting it. Rename those fixture files and the two assertions silently swap
and still pass. I'd pin the path next to each sequence number:
```rust
assert_eq!(tasks[0].data_file_path, format!("{}/1.parquet",
&fixture.table_location));
assert_eq!(tasks[0].data_sequence_number, Some(1));
assert_eq!(tasks[1].data_file_path, format!("{}/3.parquet",
&fixture.table_location));
assert_eq!(tasks[1].data_sequence_number, Some(0));
```
While we're here, the `tasks.len()` assert wants to be above the sort — as
written, a scan that returns fewer tasks panics on `tasks[0]` instead of
failing with the length message.
##########
crates/iceberg/src/scan/mod.rs:
##########
@@ -1862,6 +1940,60 @@ pub mod tests {
);
}
+ #[tokio::test]
+ async fn test_plan_files_carries_row_lineage_into_file_scan_task() {
+ let mut fixture = TableTestFixture::new();
+ fixture.setup_manifest_files().await;
+
+ let mut tasks: Vec<_> = fixture
+ .table
+ .scan()
+ .build()
+ .unwrap()
+ .plan_files()
+ .await
+ .unwrap()
+ .try_collect()
+ .await
+ .unwrap();
+
+ tasks.sort_by_key(|task| task.data_file_path.to_string());
+ assert_eq!(tasks.len(), 2);
+
+ // The added file inherits the current snapshot's data sequence number,
+ // the existing file keeps the one it was written with.
+ assert_eq!(tasks[0].data_sequence_number, Some(1));
+ assert_eq!(tasks[1].data_sequence_number, Some(0));
+
+ // first_row_id is a v3 concept; a v2 manifest carries none.
+ assert!(tasks.iter().all(|task| task.first_row_id.is_none()));
+ }
+
+ #[tokio::test]
+ async fn test_plan_files_carries_inherited_first_row_id() {
+ let mut fixture = TableTestFixture::new();
+ fixture.setup_v3_manifest_files().await;
+
+ let task = fixture
+ .table
+ .scan()
+ .build()
+ .unwrap()
+ .plan_files()
+ .await
+ .unwrap()
+ .try_collect::<Vec<_>>()
+ .await
+ .unwrap()
+ .into_iter()
+ .next()
+ .expect("expected one FileScanTask");
+
+ // The manifest-level first_row_id (0) is inherited onto the entry on
+ // read, then carried onto the task.
+ assert_eq!(task.first_row_id, Some(0));
Review Comment:
Worth asserting `data_sequence_number` here too. Both fields come from the
same two-line addition in `context.rs`, so covering only one of them leaves the
v3 path half tested — a regression that broke sequence-number threading
specifically under `build_v3_data()` wouldn't be caught anywhere.
The current snapshot's sequence number is 1, so
`assert_eq!(task.data_sequence_number, Some(1));` right after this should do it.
--
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]
